This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 48d2fc60ef30b0f247cd44c8da6f9e9d32404f45 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(
