[core][rest] Add tableId to commit snapshot to avoid wrong commit

JingsongLi · JingsongLi · commit d88cb83e3497 · 2025-05-30T17:34:30.000+08:00
diff --git a/docs/static/rest-catalog-open-api.yaml b/docs/static/rest-catalog-open-api.yaml
@@ -2395,6 +2395,8 @@ components:
     CommitTableRequest:
       type: object
       properties:
+        tableId:
+          type: string
         snapshot:
           $ref: '#/components/schemas/Snapshot'
         statistics:
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -544,6 +544,7 @@ public PagedList<Snapshot> listSnapshotsPaged(
      * Commit snapshot for table.
      *
      * @param identifier database name and table name.
+     * @param tableUuid Uuid of the table to avoid wrong commit
      * @param snapshot snapshot for committing
      * @param statistics statistics for this snapshot incremental
      * @return true if commit success
@@ -552,8 +553,11 @@ public PagedList<Snapshot> listSnapshotsPaged(
      *     this table
      */
     public boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics) {
-        CommitTableRequest request = new CommitTableRequest(snapshot, statistics);
+            Identifier identifier,
+            @Nullable String tableUuid,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics) {
+        CommitTableRequest request = new CommitTableRequest(tableUuid, snapshot, statistics);
         CommitTableResponse response =
                 client.post(
                         resourcePaths.commitTable(
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/requests/CommitTableRequest.java b/paimon-api/src/main/java/org/apache/paimon/rest/requests/CommitTableRequest.java
@@ -33,9 +33,13 @@
 @JsonIgnoreProperties(ignoreUnknown = true)
 public class CommitTableRequest implements RESTRequest {
 
+    private static final String FIELD_TABLE_ID = "tableId";
     private static final String FIELD_SNAPSHOT = "snapshot";
     private static final String FIELD_STATISTICS = "statistics";
 
+    @JsonProperty(FIELD_TABLE_ID)
+    private final String tableId;
+
     @JsonProperty(FIELD_SNAPSHOT)
     private final Snapshot snapshot;
 
@@ -44,12 +48,19 @@ public class CommitTableRequest implements RESTRequest {
 
     @JsonCreator
     public CommitTableRequest(
+            @JsonProperty(FIELD_TABLE_ID) String tableId,
             @JsonProperty(FIELD_SNAPSHOT) Snapshot snapshot,
             @JsonProperty(FIELD_STATISTICS) List<PartitionStatistics> statistics) {
+        this.tableId = tableId;
         this.snapshot = snapshot;
         this.statistics = statistics;
     }
 
+    @JsonGetter(FIELD_TABLE_ID)
+    public String getTableId() {
+        return tableId;
+    }
+
     @JsonGetter(FIELD_SNAPSHOT)
     public Snapshot getSnapshot() {
         return snapshot;
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -477,7 +477,10 @@ public List<String> listBranches(Identifier identifier) throws TableNotExistExce
 
     @Override
     public boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics) {
+            Identifier identifier,
+            @Nullable String tableUuid,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics) {
         throw new UnsupportedOperationException();
     }
 
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -617,6 +617,7 @@ default boolean supportsListByPattern() {
      * Commit the {@link Snapshot} for table identified by the given {@link Identifier}.
      *
      * @param identifier Path of the table
+     * @param tableUuid Uuid of the table to avoid wrong commit
      * @param snapshot Snapshot to be committed
      * @param statistics statistics information of this change
      * @return Success or not
@@ -625,7 +626,10 @@ default boolean supportsListByPattern() {
      *     #supportsVersionManagement()}
      */
     boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics)
+            Identifier identifier,
+            @Nullable String tableUuid,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics)
             throws Catalog.TableNotExistException;
 
     /**
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
@@ -22,25 +22,29 @@
 import org.apache.paimon.partition.PartitionStatistics;
 import org.apache.paimon.utils.SnapshotManager;
 
+import javax.annotation.Nullable;
+
 import java.util.List;
 
 /** A {@link SnapshotCommit} using {@link Catalog} to commit. */
 public class CatalogSnapshotCommit implements SnapshotCommit {
 
     private final Catalog catalog;
     private final Identifier identifier;
+    @Nullable private final String uuid;
 
-    public CatalogSnapshotCommit(Catalog catalog, Identifier identifier) {
+    public CatalogSnapshotCommit(Catalog catalog, Identifier identifier, @Nullable String uuid) {
         this.catalog = catalog;
         this.identifier = identifier;
+        this.uuid = uuid;
     }
 
     @Override
     public boolean commit(Snapshot snapshot, String branch, List<PartitionStatistics> statistics)
             throws Exception {
         Identifier newIdentifier =
                 new Identifier(identifier.getDatabaseName(), identifier.getTableName(), branch);
-        return catalog.commitSnapshot(newIdentifier, snapshot, statistics);
+        return catalog.commitSnapshot(newIdentifier, uuid, snapshot, statistics);
     }
 
     @Override
@@ -54,14 +58,16 @@ public static class Factory implements SnapshotCommit.Factory {
         private static final long serialVersionUID = 1L;
 
         private final CatalogLoader catalogLoader;
+        @Nullable private final String uuid;
 
-        public Factory(CatalogLoader catalogLoader) {
+        public Factory(CatalogLoader catalogLoader, @Nullable String uuid) {
             this.catalogLoader = catalogLoader;
+            this.uuid = uuid;
         }
 
         @Override
         public SnapshotCommit create(Identifier identifier, SnapshotManager snapshotManager) {
-            return new CatalogSnapshotCommit(catalogLoader.load(), identifier);
+            return new CatalogSnapshotCommit(catalogLoader.load(), identifier, uuid);
         }
     }
 }
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -212,9 +212,12 @@ public List<String> listBranches(Identifier identifier) throws TableNotExistExce
 
     @Override
     public boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics)
+            Identifier identifier,
+            @Nullable String tableUuid,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics)
             throws TableNotExistException {
-        return wrapped.commitSnapshot(identifier, snapshot, statistics);
+        return wrapped.commitSnapshot(identifier, tableUuid, snapshot, statistics);
     }
 
     @Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -39,7 +39,6 @@
 import org.apache.paimon.rest.exceptions.NoSuchResourceException;
 import org.apache.paimon.rest.exceptions.NotImplementedException;
 import org.apache.paimon.rest.exceptions.ServiceFailureException;
-import org.apache.paimon.rest.requests.RollbackTableRequest;
 import org.apache.paimon.rest.responses.ErrorResponse;
 import org.apache.paimon.rest.responses.GetDatabaseResponse;
 import org.apache.paimon.rest.responses.GetFunctionResponse;
@@ -331,10 +330,13 @@ public boolean supportsVersionManagement() {
 
     @Override
     public boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics)
+            Identifier identifier,
+            @Nullable String tableUuid,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics)
             throws TableNotExistException {
         try {
-            return api.commitSnapshot(identifier, snapshot, statistics);
+            return api.commitSnapshot(identifier, tableUuid, snapshot, statistics);
         } catch (NoSuchResourceException e) {
             throw new TableNotExistException(identifier);
         } catch (ForbiddenException e) {
@@ -347,7 +349,6 @@ public boolean commitSnapshot(
     @Override
     public void rollbackTo(Identifier identifier, Instant instant)
             throws Catalog.TableNotExistException {
-        RollbackTableRequest request = new RollbackTableRequest(instant);
         try {
             api.rollbackTo(identifier, instant);
         } catch (NoSuchResourceException e) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -95,7 +95,7 @@ public boolean supportsVersionManagement() {
     public SnapshotCommit snapshotCommit(SnapshotManager snapshotManager) {
         SnapshotCommit.Factory factory;
         if (catalogLoader != null && supportsVersionManagement) {
-            factory = new CatalogSnapshotCommit.Factory(catalogLoader);
+            factory = new CatalogSnapshotCommit.Factory(catalogLoader, uuid);
         } else {
             factory = new RenamingSnapshotCommit.Factory(lockFactory, lockContext);
         }
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -773,10 +773,11 @@ private MockResponse commitTableHandle(Identifier identifier, String data) throw
         if (!tableMetadataStore.containsKey(identifier.getFullName())) {
             throw new Catalog.TableNotExistException(identifier);
         }
-        boolean success =
-                commitSnapshot(identifier, requestBody.getSnapshot(), requestBody.getStatistics());
-        CommitTableResponse response = new CommitTableResponse(success);
-        return mockResponse(response, 200);
+        return commitSnapshot(
+                identifier,
+                requestBody.getTableId(),
+                requestBody.getSnapshot(),
+                requestBody.getStatistics());
     }
 
     private MockResponse rollbackTableByIdHandle(Identifier identifier, long snapshotId)
@@ -1864,10 +1865,16 @@ private String geTableFullNameWithSnapshotId(Identifier identifier, long snapsho
         return String.format("%s-%d", identifier.getFullName(), snapshotId);
     }
 
-    private boolean commitSnapshot(
-            Identifier identifier, Snapshot snapshot, List<PartitionStatistics> statistics)
+    private MockResponse commitSnapshot(
+            Identifier identifier,
+            String tableId,
+            Snapshot snapshot,
+            List<PartitionStatistics> statistics)
             throws Catalog.TableNotExistException {
         FileStoreTable table = getFileTable(identifier);
+        if (!tableId.equals(table.catalogEnvironment().uuid())) {
+            throw new Catalog.TableNotExistException(identifier);
+        }
         RenamingSnapshotCommit commit =
                 new RenamingSnapshotCommit(table.snapshotManager(), Lock.empty());
         String branchName = identifier.getBranchName();
@@ -1989,7 +1996,8 @@ private boolean commitSnapshot(
                                                         && partition.recordCount() <= 0);
                                 return partitions.isEmpty();
                             });
-            return success;
+            CommitTableResponse response = new CommitTableResponse(success);
+            return mockResponse(response, 200);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
diff --git a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -308,6 +308,7 @@ void testApiWhenTableNoPermission() throws Exception {
                 () ->
                         restCatalog.commitSnapshot(
                                 identifier,
+                                "",
                                 createSnapshotWithMillis(1L, System.currentTimeMillis()),
                                 new ArrayList<PartitionStatistics>()));
     }
@@ -1263,10 +1264,21 @@ void testSnapshotFromREST() throws Exception {
                 () ->
                         restCatalog.commitSnapshot(
                                 hasSnapshotTableIdentifier,
+                                "",
                                 createSnapshotWithMillis(1L, System.currentTimeMillis()),
                                 new ArrayList<>()));
 
         createTable(hasSnapshotTableIdentifier, Maps.newHashMap(), Lists.newArrayList("col1"));
+
+        assertThrows(
+                Catalog.TableNotExistException.class,
+                () ->
+                        restCatalog.commitSnapshot(
+                                hasSnapshotTableIdentifier,
+                                "unknown_id",
+                                createSnapshotWithMillis(1L, System.currentTimeMillis()),
+                                new ArrayList<>()));
+
         long id = 10086;
         long millis = System.currentTimeMillis();
         updateSnapshotOnRestServer(

Original file line number	Diff line number	Diff line change
`@@ -95,7 +95,7 @@ public boolean supportsVersionManagement() {`
`95`	`95`	`public SnapshotCommit snapshotCommit(SnapshotManager snapshotManager) {`
`96`	`96`	`SnapshotCommit.Factory factory;`
`97`	`97`	`if (catalogLoader != null && supportsVersionManagement) {`
`98`		`- factory = new CatalogSnapshotCommit.Factory(catalogLoader);`
	`98`	`+ factory = new CatalogSnapshotCommit.Factory(catalogLoader, uuid);`
`99`	`99`	`} else {`
`100`	`100`	`factory = new RenamingSnapshotCommit.Factory(lockFactory, lockContext);`
`101`	`101`	`}`