This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new a67e815ff2 [flink] Fix rollback_to procedure throwing
NullPointerException when neither snapshot_id nor tag is set (#9944)
a67e815ff2 is described below
commit a67e815ff21e041bf5c3ca2768ae682d24d16553
Author: Li Guo <[email protected]>
AuthorDate: Thu Sep 17 20:58:58 2026 -0700
[flink] Fix rollback_to procedure throwing NullPointerException when
neither snapshot_id nor tag is set (#9944)
---
docs/docs/flink/procedures/snapshots.md | 6 +-
.../flink/procedure/RollbackToProcedure.java | 7 +-
.../flink/procedure/RollbackProcedureITCase.java | 74 ++++++++++++++++++++++
3 files changed, 84 insertions(+), 3 deletions(-)
diff --git a/docs/docs/flink/procedures/snapshots.md
b/docs/docs/flink/procedures/snapshots.md
index cb5902fef8..3f1e91b8b8 100644
--- a/docs/docs/flink/procedures/snapshots.md
+++ b/docs/docs/flink/procedures/snapshots.md
@@ -198,9 +198,11 @@ To rollback to a specific version of target table.
Argument:
- `table`: the target table identifier. Cannot be empty.
-- snapshotId (Long): id of the snapshot that will roll back to.
+- `snapshot_id`: id of the snapshot that will roll back to.
-- `tagName`: name of the tag that will roll back to.
+- `tag`: name of the tag that will roll back to.
+
+Exactly one of `snapshot_id` and `tag` must be set.
**Syntax**
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToProcedure.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToProcedure.java
index 8320453f8e..7f57649e1b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToProcedure.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/procedure/RollbackToProcedure.java
@@ -66,8 +66,13 @@ public class RollbackToProcedure extends ProcedureBase {
Snapshot latestSnapshot = store.snapshotManager().latestSnapshot();
Preconditions.checkNotNull(latestSnapshot, "Latest snapshot is null,
can not rollback.");
+ boolean hasTag = !StringUtils.isNullOrWhitespaceOnly(tagName);
+ boolean hasSnapshot = snapshotId != null;
+ Preconditions.checkArgument(
+ hasTag != hasSnapshot, "Must specify exactly one of tag and
snapshot_id.");
+
long rollbackSnapshotId;
- if (!StringUtils.isNullOrWhitespaceOnly(tagName)) {
+ if (hasTag) {
table.rollbackTo(tagName);
rollbackSnapshotId =
store.newTagManager().getOrThrow(tagName).trimToSnapshot().id();
} else {
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackProcedureITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackProcedureITCase.java
index 884237a16f..668dfe82ae 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackProcedureITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/procedure/RollbackProcedureITCase.java
@@ -26,6 +26,7 @@ import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
/** IT Case for {@link RollbackToProcedure} and {@link
RollbackToTimestampProcedure}. */
@@ -59,6 +60,79 @@ public class RollbackProcedureITCase extends
CatalogITCaseBase {
.containsExactly(Row.of(latestSnapshotId, 2L));
}
+ @Test
+ public void testRollbackToRequiresExactlyOneOfTagAndSnapshotId() throws
Exception {
+ sql(
+ "CREATE TABLE T (id STRING, name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED)"
+ + " WITH ('bucket'='1', 'write-only'='true')");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ for (int i = 1; i <= 3; i++) {
+ sql("INSERT INTO T VALUES ('" + i + "', '" + i + "')");
+ }
+ assertEquals(3, snapshotManager.latestSnapshotId());
+ sql("CALL sys.create_tag(`table` => 'default.T', tag => 'tag-1',
snapshot_id => 1)");
+
+ // neither tag nor snapshot_id
+ assertThatThrownBy(() -> sql("CALL sys.rollback_to(`table` =>
'default.T')"))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Must specify exactly one of tag and
snapshot_id.");
+
+ // both tag and snapshot_id
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.rollback_to(`table` =>
'default.T', tag => 'tag-1', snapshot_id => 2)"))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Must specify exactly one of tag and
snapshot_id.");
+
+ // nothing was rolled back
+ assertEquals(3, snapshotManager.latestSnapshotId());
+ }
+
+ @Test
+ public void testRollbackToTreatsBlankTagAsMissing() throws Exception {
+ sql(
+ "CREATE TABLE T (id STRING, name STRING,"
+ + " PRIMARY KEY (id) NOT ENFORCED)"
+ + " WITH ('bucket'='1', 'write-only'='true')");
+
+ FileStoreTable table = paimonTable("T");
+ SnapshotManager snapshotManager = table.snapshotManager();
+
+ for (int i = 1; i <= 3; i++) {
+ sql("INSERT INTO T VALUES ('" + i + "', '" + i + "')");
+ }
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ // a whitespace-only tag alone: neither argument is usable
+ assertThatThrownBy(() -> sql("CALL sys.rollback_to(`table` =>
'default.T', tag => ' ')"))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Must specify exactly one of tag and
snapshot_id.");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ // explicit NULLs for both arguments behave like omitting them
+ assertThatThrownBy(
+ () ->
+ sql(
+ "CALL sys.rollback_to(`table` =>
'default.T', tag => CAST(NULL AS STRING), snapshot_id => CAST(NULL AS
BIGINT))"))
+ .rootCause()
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Must specify exactly one of tag and
snapshot_id.");
+ assertEquals(3, snapshotManager.latestSnapshotId());
+
+ // an empty tag next to a snapshot_id keeps the old behaviour: the
snapshot wins
+ assertThat(sql("CALL sys.rollback_to(`table` => 'default.T', tag =>
'', snapshot_id => 2)"))
+ .containsExactly(Row.of(3L, 2L));
+ assertEquals(2, snapshotManager.latestSnapshotId());
+ }
+
@Test
public void testRollbackToTimestamp() throws Exception {
sql(
