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 d113d293fd [core][python] Add snapshot UUID validation to catalog
commits (#8920)
d113d293fd is described below
commit d113d293fd39894b6521f51b4ed21865fd7dd1c1
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jul 30 15:54:21 2026 +0800
[core][python] Add snapshot UUID validation to catalog commits (#8920)
---
docs/docs/concepts/spec/snapshot.md | 39 ++++----
docs/static/rest-catalog-open-api.yaml | 6 ++
.../src/main/java/org/apache/paimon/Snapshot.java | 111 ++++-----------------
.../main/java/org/apache/paimon/rest/RESTApi.java | 5 +-
.../paimon/rest/requests/CommitTableRequest.java | 15 +++
.../src/main/java/org/apache/paimon/Changelog.java | 3 +
.../org/apache/paimon/catalog/AbstractCatalog.java | 1 +
.../java/org/apache/paimon/catalog/Catalog.java | 4 +-
.../paimon/catalog/CatalogSnapshotCommit.java | 8 +-
.../org/apache/paimon/catalog/DelegateCatalog.java | 4 +-
.../paimon/catalog/RenamingSnapshotCommit.java | 8 +-
.../org/apache/paimon/catalog/SnapshotCommit.java | 8 +-
.../paimon/operation/FileStoreCommitImpl.java | 23 +++--
.../java/org/apache/paimon/rest/RESTCatalog.java | 4 +-
.../src/main/java/org/apache/paimon/tag/Tag.java | 4 +
.../test/java/org/apache/paimon/SnapshotTest.java | 51 +++++++---
.../DataEvolutionRowIdReassignerTest.java | 2 +
.../paimon/catalog/CatalogSnapshotCommitTest.java | 62 ++++++++++++
.../paimon/catalog/RenamingSnapshotCommitTest.java | 4 +-
.../paimon/operation/FileStoreCommitTest.java | 19 +++-
.../org/apache/paimon/rest/RESTCatalogServer.java | 13 ++-
.../org/apache/paimon/rest/RESTCatalogTest.java | 36 +++++++
.../test/java/org/apache/paimon/tag/TagTest.java | 4 +
paimon-python/pypaimon/api/api_request.py | 4 +
paimon-python/pypaimon/api/rest_api.py | 9 +-
paimon-python/pypaimon/catalog/catalog.py | 2 +
.../pypaimon/catalog/filesystem_catalog.py | 1 +
paimon-python/pypaimon/catalog/jdbc_catalog.py | 1 +
.../pypaimon/catalog/rest/rest_catalog.py | 10 +-
paimon-python/pypaimon/changelog/changelog.py | 1 +
paimon-python/pypaimon/common/json_util.py | 13 ++-
.../pypaimon/snapshot/catalog_snapshot_commit.py | 18 +++-
.../pypaimon/snapshot/renaming_snapshot_commit.py | 10 +-
paimon-python/pypaimon/snapshot/snapshot.py | 9 ++
paimon-python/pypaimon/snapshot/snapshot_commit.py | 10 +-
paimon-python/pypaimon/tag/tag.py | 2 +
.../pypaimon/tests/file_store_commit_test.py | 10 +-
.../pypaimon/tests/overwrite_changes_cache_test.py | 12 +--
.../tests/overwrite_commit_conflict_test.py | 16 +--
paimon-python/pypaimon/tests/rest/api_test.py | 6 +-
.../rest/rest_catalog_commit_snapshot_test.py | 77 +++++++++++++-
.../pypaimon/tests/rest/rest_permission_test.py | 1 +
paimon-python/pypaimon/tests/rest/rest_server.py | 10 ++
.../pypaimon/tests/write/dynamic_bucket_test.py | 3 +-
paimon-python/pypaimon/write/file_store_commit.py | 6 +-
45 files changed, 483 insertions(+), 182 deletions(-)
diff --git a/docs/docs/concepts/spec/snapshot.md
b/docs/docs/concepts/spec/snapshot.md
index dd2bf28ed2..d09db56a51 100644
--- a/docs/docs/concepts/spec/snapshot.md
+++ b/docs/docs/concepts/spec/snapshot.md
@@ -46,22 +46,23 @@ be visible.
Snapshot File is JSON, it includes:
1. version: Snapshot file version, current is 3.
-2. id: snapshot id, same to file name.
-3. schemaId: the corresponding schema version for this commit.
-4. baseManifestList: a manifest list recording all changes from the previous
snapshots.
-5. deltaManifestList: a manifest list recording all new changes occurred in
this snapshot.
-6. changelogManifestList: a manifest list recording all changelog produced in
this snapshot, null if no changelog is produced.
-7. indexManifest: a manifest recording all index files of this table, null if
no table index file.
-8. commitUser: usually generated by UUID, it is used for recovery of streaming
writes, one stream write job with one user.
-9. commitIdentifier: transaction id corresponding to streaming write, each
transaction may result in multiple commits for different commitKinds.
-10. commitKind: type of changes in this snapshot, including append, compact,
overwrite and analyze.
-11. timeMillis: commit time millis.
-12. logOffsets: commit log offsets.
-13. totalRecordCount: record count of all changes occurred in this snapshot.
-14. deltaRecordCount: record count of all new changes occurred in this
snapshot.
-15. changelogRecordCount: record count of all changelog produced in this
snapshot.
-16. watermark: watermark for input records, from Flink watermark mechanism,
Long.MIN_VALUE if there is no watermark.
-17. statistics: stats file name for statistics of this table.
-18. properties: additional key-value properties of this snapshot.
-19. nextRowId: next row id for row tracking.
-20. operation: logical operation type, e.g. WRITE, DELETE, UPDATE, MERGE. Null
if not set.
+2. uuid: UUID identifying the complete immutable snapshot, null for snapshots
created by older versions.
+3. id: snapshot id, same to file name.
+4. schemaId: the corresponding schema version for this commit.
+5. baseManifestList: a manifest list recording all changes from the previous
snapshots.
+6. deltaManifestList: a manifest list recording all new changes occurred in
this snapshot.
+7. changelogManifestList: a manifest list recording all changelog produced in
this snapshot, null if no changelog is produced.
+8. indexManifest: a manifest recording all index files of this table, null if
no table index file.
+9. commitUser: usually generated by UUID, it is used for recovery of streaming
writes, one stream write job with one user.
+10. commitIdentifier: transaction id corresponding to streaming write, each
transaction may result in multiple commits for different commitKinds.
+11. commitKind: type of changes in this snapshot, including append, compact,
overwrite and analyze.
+12. timeMillis: commit time millis.
+13. logOffsets: commit log offsets.
+14. totalRecordCount: record count of all changes occurred in this snapshot.
+15. deltaRecordCount: record count of all new changes occurred in this
snapshot.
+16. changelogRecordCount: record count of all changelog produced in this
snapshot.
+17. watermark: watermark for input records, from Flink watermark mechanism,
Long.MIN_VALUE if there is no watermark.
+18. statistics: stats file name for statistics of this table.
+19. properties: additional key-value properties of this snapshot.
+20. nextRowId: next row id for row tracking.
+21. operation: logical operation type, e.g. WRITE, DELETE, UPDATE, MERGE. Null
if not set.
diff --git a/docs/static/rest-catalog-open-api.yaml
b/docs/static/rest-catalog-open-api.yaml
index 51def16c43..47f5acef5e 100644
--- a/docs/static/rest-catalog-open-api.yaml
+++ b/docs/static/rest-catalog-open-api.yaml
@@ -3203,6 +3203,9 @@ components:
properties:
tableId:
type: string
+ baseSnapshotUuid:
+ type: string
+ nullable: true
snapshot:
$ref: '#/components/schemas/Snapshot'
statistics:
@@ -3268,6 +3271,9 @@ components:
type: integer
format: int32
nullable: true
+ uuid:
+ type: string
+ nullable: true
id:
type: integer
format: int64
diff --git a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
index 4a200dfecc..d34b061b78 100644
--- a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
@@ -32,6 +32,7 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.util.Map;
import java.util.Objects;
+import java.util.UUID;
/**
* This file is the entrance to all data committed at some specific time point.
@@ -49,6 +50,7 @@ public class Snapshot implements Serializable {
protected static final int CURRENT_VERSION = 3;
protected static final String FIELD_VERSION = "version";
+ protected static final String FIELD_UUID = "uuid";
protected static final String FIELD_ID = "id";
protected static final String FIELD_SCHEMA_ID = "schemaId";
protected static final String FIELD_BASE_MANIFEST_LIST =
"baseManifestList";
@@ -75,6 +77,12 @@ public class Snapshot implements Serializable {
@JsonProperty(FIELD_VERSION)
protected final int version;
+ // null for snapshots created before UUID was introduced
+ @JsonProperty(FIELD_UUID)
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @Nullable
+ protected final String uuid;
+
@JsonProperty(FIELD_ID)
protected final long id;
@@ -187,51 +195,6 @@ public class Snapshot implements Serializable {
@Nullable
protected final Operation operation;
- public Snapshot(
- long id,
- long schemaId,
- String baseManifestList,
- @Nullable Long baseManifestListSize,
- String deltaManifestList,
- @Nullable Long deltaManifestListSize,
- @Nullable String changelogManifestList,
- @Nullable Long changelogManifestListSize,
- @Nullable String indexManifest,
- String commitUser,
- long commitIdentifier,
- CommitKind commitKind,
- long timeMillis,
- long totalRecordCount,
- long deltaRecordCount,
- @Nullable Long changelogRecordCount,
- @Nullable Long watermark,
- @Nullable String statistics,
- @Nullable Map<String, String> properties,
- @Nullable Long nextRowId) {
- this(
- id,
- schemaId,
- baseManifestList,
- baseManifestListSize,
- deltaManifestList,
- deltaManifestListSize,
- changelogManifestList,
- changelogManifestListSize,
- indexManifest,
- commitUser,
- commitIdentifier,
- commitKind,
- timeMillis,
- totalRecordCount,
- deltaRecordCount,
- changelogRecordCount,
- watermark,
- statistics,
- properties,
- nextRowId,
- null);
- }
-
public Snapshot(
long id,
long schemaId,
@@ -256,6 +219,7 @@ public class Snapshot implements Serializable {
@Nullable Operation operation) {
this(
CURRENT_VERSION,
+ UUID.randomUUID().toString(),
id,
schemaId,
baseManifestList,
@@ -279,56 +243,10 @@ public class Snapshot implements Serializable {
operation);
}
- public Snapshot(
- int version,
- long id,
- long schemaId,
- String baseManifestList,
- @Nullable Long baseManifestListSize,
- String deltaManifestList,
- @Nullable Long deltaManifestListSize,
- @Nullable String changelogManifestList,
- @Nullable Long changelogManifestListSize,
- @Nullable String indexManifest,
- String commitUser,
- long commitIdentifier,
- CommitKind commitKind,
- long timeMillis,
- long totalRecordCount,
- long deltaRecordCount,
- @Nullable Long changelogRecordCount,
- @Nullable Long watermark,
- @Nullable String statistics,
- @Nullable Map<String, String> properties,
- @Nullable Long nextRowId) {
- this(
- version,
- id,
- schemaId,
- baseManifestList,
- baseManifestListSize,
- deltaManifestList,
- deltaManifestListSize,
- changelogManifestList,
- changelogManifestListSize,
- indexManifest,
- commitUser,
- commitIdentifier,
- commitKind,
- timeMillis,
- totalRecordCount,
- deltaRecordCount,
- changelogRecordCount,
- watermark,
- statistics,
- properties,
- nextRowId,
- null);
- }
-
@JsonCreator
public Snapshot(
@JsonProperty(FIELD_VERSION) int version,
+ @JsonProperty(FIELD_UUID) @Nullable String uuid,
@JsonProperty(FIELD_ID) long id,
@JsonProperty(FIELD_SCHEMA_ID) long schemaId,
@JsonProperty(FIELD_BASE_MANIFEST_LIST) String baseManifestList,
@@ -352,6 +270,7 @@ public class Snapshot implements Serializable {
@JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId,
@JsonProperty(FIELD_OPERATION) @Nullable Operation operation) {
this.version = version;
+ this.uuid = uuid;
this.id = id;
this.schemaId = schemaId;
this.baseManifestList = baseManifestList;
@@ -380,6 +299,12 @@ public class Snapshot implements Serializable {
return version;
}
+ @JsonGetter(FIELD_UUID)
+ @Nullable
+ public String uuid() {
+ return uuid;
+ }
+
@JsonGetter(FIELD_ID)
public long id() {
return id;
@@ -504,6 +429,7 @@ public class Snapshot implements Serializable {
public int hashCode() {
return Objects.hash(
version,
+ uuid,
id,
schemaId,
baseManifestList,
@@ -537,6 +463,7 @@ public class Snapshot implements Serializable {
}
Snapshot that = (Snapshot) o;
return Objects.equals(version, that.version)
+ && Objects.equals(uuid, that.uuid)
&& id == that.id
&& schemaId == that.schemaId
&& Objects.equals(baseManifestList, that.baseManifestList)
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
index 8da06988c3..401921c8d1 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTApi.java
@@ -669,6 +669,7 @@ public class RESTApi {
*
* @param identifier database name and table name.
* @param tableUuid Uuid of the table to avoid wrong commit
+ * @param baseSnapshotUuid Uuid of the snapshot on which the commit is
based
* @param snapshot snapshot for committing
* @param statistics statistics for this snapshot incremental
* @return true if commit success
@@ -679,9 +680,11 @@ public class RESTApi {
public boolean commitSnapshot(
Identifier identifier,
@Nullable String tableUuid,
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
List<PartitionStatistics> statistics) {
- CommitTableRequest request = new CommitTableRequest(tableUuid,
snapshot, statistics);
+ CommitTableRequest request =
+ new CommitTableRequest(tableUuid, baseSnapshotUuid, 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
index 18bb1bf999..8048dd669a 100644
---
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
@@ -27,6 +27,8 @@ import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGet
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import javax.annotation.Nullable;
+
import java.util.List;
/** Request for committing snapshot to table. */
@@ -34,12 +36,17 @@ import java.util.List;
public class CommitTableRequest implements RESTRequest {
private static final String FIELD_TABLE_ID = "tableId";
+ private static final String FIELD_BASE_SNAPSHOT_UUID = "baseSnapshotUuid";
private static final String FIELD_SNAPSHOT = "snapshot";
private static final String FIELD_STATISTICS = "statistics";
@JsonProperty(FIELD_TABLE_ID)
private final String tableId;
+ @JsonProperty(FIELD_BASE_SNAPSHOT_UUID)
+ @Nullable
+ private final String baseSnapshotUuid;
+
@JsonProperty(FIELD_SNAPSHOT)
private final Snapshot snapshot;
@@ -49,9 +56,11 @@ public class CommitTableRequest implements RESTRequest {
@JsonCreator
public CommitTableRequest(
@JsonProperty(FIELD_TABLE_ID) String tableId,
+ @JsonProperty(FIELD_BASE_SNAPSHOT_UUID) @Nullable String
baseSnapshotUuid,
@JsonProperty(FIELD_SNAPSHOT) Snapshot snapshot,
@JsonProperty(FIELD_STATISTICS) List<PartitionStatistics>
statistics) {
this.tableId = tableId;
+ this.baseSnapshotUuid = baseSnapshotUuid;
this.snapshot = snapshot;
this.statistics = statistics;
}
@@ -61,6 +70,12 @@ public class CommitTableRequest implements RESTRequest {
return tableId;
}
+ @JsonGetter(FIELD_BASE_SNAPSHOT_UUID)
+ @Nullable
+ public String getBaseSnapshotUuid() {
+ return baseSnapshotUuid;
+ }
+
@JsonGetter(FIELD_SNAPSHOT)
public Snapshot getSnapshot() {
return snapshot;
diff --git a/paimon-core/src/main/java/org/apache/paimon/Changelog.java
b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
index 05b76d7153..aaa1f7f4d5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Changelog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
@@ -44,6 +44,7 @@ public class Changelog extends Snapshot {
public Changelog(Snapshot snapshot) {
this(
snapshot.version(),
+ snapshot.uuid(),
snapshot.id(),
snapshot.schemaId(),
snapshot.baseManifestList(),
@@ -70,6 +71,7 @@ public class Changelog extends Snapshot {
@JsonCreator
public Changelog(
@JsonProperty(FIELD_VERSION) int version,
+ @JsonProperty(FIELD_UUID) @Nullable String uuid,
@JsonProperty(FIELD_ID) long id,
@JsonProperty(FIELD_SCHEMA_ID) long schemaId,
@JsonProperty(FIELD_BASE_MANIFEST_LIST) String baseManifestList,
@@ -94,6 +96,7 @@ public class Changelog extends Snapshot {
@JsonProperty(FIELD_OPERATION) @Nullable Operation operation) {
super(
version,
+ uuid,
id,
schemaId,
baseManifestList,
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
index 67992a755a..5784d56b0e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -681,6 +681,7 @@ public abstract class AbstractCatalog implements Catalog {
public boolean commitSnapshot(
Identifier identifier,
@Nullable String tableUuid,
+ @Nullable String baseSnapshotUuid,
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
index eb650dc95e..5b437efbab 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/Catalog.java
@@ -709,7 +709,7 @@ public interface Catalog extends AutoCloseable {
* will throw an {@link UnsupportedOperationException}, affect the
following methods:
*
* <ul>
- * <li>{@link #commitSnapshot(Identifier, String, Snapshot, List)}.
+ * <li>{@link #commitSnapshot(Identifier, String, String, Snapshot,
List)}.
* <li>{@link #loadSnapshot(Identifier)}.
* <li>{@link #rollbackTo(Identifier, Instant)}.
* <li>{@link #createBranch(Identifier, String, String)}.
@@ -729,6 +729,7 @@ public interface Catalog extends AutoCloseable {
*
* @param identifier Path of the table
* @param tableUuid Uuid of the table to avoid wrong commit
+ * @param baseSnapshotUuid Uuid of the snapshot on which the commit is
based
* @param snapshot Snapshot to be committed
* @param statistics statistics information of this change
* @return Success or not
@@ -739,6 +740,7 @@ public interface Catalog extends AutoCloseable {
boolean commitSnapshot(
Identifier identifier,
@Nullable String tableUuid,
+ @Nullable String baseSnapshotUuid,
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
index b49f1a12ef..fc50e33b24 100644
---
a/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/catalog/CatalogSnapshotCommit.java
@@ -39,11 +39,15 @@ public class CatalogSnapshotCommit implements
SnapshotCommit {
}
@Override
- public boolean commit(Snapshot snapshot, String branch,
List<PartitionStatistics> statistics)
+ public boolean commit(
+ @Nullable String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
throws Exception {
Identifier newIdentifier =
new Identifier(identifier.getDatabaseName(),
identifier.getTableName(), branch);
- return catalog.commitSnapshot(newIdentifier, uuid, snapshot,
statistics);
+ return catalog.commitSnapshot(newIdentifier, uuid, baseSnapshotUuid,
snapshot, statistics);
}
@Override
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
index 10ae54ea45..13faef105c 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/DelegateCatalog.java
@@ -306,10 +306,12 @@ public abstract class DelegateCatalog implements Catalog {
public boolean commitSnapshot(
Identifier identifier,
@Nullable String tableUuid,
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
List<PartitionStatistics> statistics)
throws TableNotExistException {
- return wrapped.commitSnapshot(identifier, tableUuid, snapshot,
statistics);
+ return wrapped.commitSnapshot(
+ identifier, tableUuid, baseSnapshotUuid, snapshot, statistics);
}
@Override
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
index a79aafb298..8a655f6076 100644
---
a/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/catalog/RenamingSnapshotCommit.java
@@ -25,6 +25,8 @@ import org.apache.paimon.operation.Lock;
import org.apache.paimon.partition.PartitionStatistics;
import org.apache.paimon.utils.SnapshotManager;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.util.List;
import java.util.concurrent.Callable;
@@ -48,7 +50,11 @@ public class RenamingSnapshotCommit implements
SnapshotCommit {
}
@Override
- public boolean commit(Snapshot snapshot, String branch,
List<PartitionStatistics> statistics)
+ public boolean commit(
+ @Nullable String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
throws Exception {
Path newSnapshotPath =
snapshotManager.branch().equals(branch)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
index 43abacd7b6..ff912e2139 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/SnapshotCommit.java
@@ -21,11 +21,17 @@ package org.apache.paimon.catalog;
import org.apache.paimon.Snapshot;
import org.apache.paimon.partition.PartitionStatistics;
+import javax.annotation.Nullable;
+
import java.util.List;
/** Interface to commit snapshot atomically. */
public interface SnapshotCommit extends AutoCloseable {
- boolean commit(Snapshot snapshot, String branch, List<PartitionStatistics>
statistics)
+ boolean commit(
+ @Nullable String baseSnapshotUuid,
+ Snapshot snapshot,
+ String branch,
+ List<PartitionStatistics> statistics)
throws Exception;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 1eacfa71f7..5725dc89a7 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -1201,7 +1201,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
callback ->
callback.call(finalBaseFiles, finalDeltaFiles,
indexFiles, newSnapshot));
try {
- success = commitSnapshotImpl(newSnapshot, deltaStatistics);
+ success = commitSnapshotImpl(latestSnapshot, newSnapshot,
deltaStatistics);
} catch (Exception e) {
// commit exception, not sure about the situation and should not
clean up the files
LOG.warn("Retry commit for exception.", e);
@@ -1336,7 +1336,7 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
nextRowId,
null);
- return commitSnapshotImpl(newSnapshot, emptyList());
+ return commitSnapshotImpl(latest, newSnapshot, emptyList());
}
@Override
@@ -1418,7 +1418,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
targetSnapshot.watermark(),
targetSnapshot.statistics(),
targetSnapshot.properties(),
- nextRowId);
+ nextRowId,
+ null);
// The rollback is an overwrite from the previous latest to the
target, so the base files,
// delta files and index changes describe the transition the callbacks
need. These are
@@ -1435,7 +1436,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
callback -> callback.call(baseFiles, deltaFiles, indexChanges,
newSnapshot));
boolean success =
- commitSnapshotImpl(newSnapshot, new
ArrayList<>(PartitionEntry.merge(deltaFiles)));
+ commitSnapshotImpl(
+ latest, newSnapshot, new
ArrayList<>(PartitionEntry.merge(deltaFiles)));
if (success) {
// Notify the post-commit callbacks so external views stay in sync
with the rolled-back
// state (e.g. Iceberg compatibility metadata and chain-table
overwrite handling).
@@ -1571,16 +1573,23 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latestSnapshot.nextRowId(),
null);
- return commitSnapshotImpl(newSnapshot, emptyList());
+ return commitSnapshotImpl(latestSnapshot, newSnapshot, emptyList());
}
- private boolean commitSnapshotImpl(Snapshot newSnapshot,
List<PartitionEntry> deltaStatistics) {
+ private boolean commitSnapshotImpl(
+ @Nullable Snapshot baseSnapshot,
+ Snapshot newSnapshot,
+ List<PartitionEntry> deltaStatistics) {
try {
List<PartitionStatistics> statistics = new
ArrayList<>(deltaStatistics.size());
for (PartitionEntry entry : deltaStatistics) {
statistics.add(entry.toPartitionStatistics(partitionComputer));
}
- return snapshotCommit.commit(newSnapshot, options.branch(),
statistics);
+ return snapshotCommit.commit(
+ baseSnapshot == null ? null : baseSnapshot.uuid(),
+ newSnapshot,
+ options.branch(),
+ statistics);
} catch (Throwable e) {
// exception when performing the atomic rename,
// we cannot clean up because we can't determine the success
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
index 3db1052c4c..f1b5d94cc1 100644
--- a/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/rest/RESTCatalog.java
@@ -466,11 +466,13 @@ public class RESTCatalog implements Catalog {
public boolean commitSnapshot(
Identifier identifier,
@Nullable String tableUuid,
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
List<PartitionStatistics> statistics)
throws TableNotExistException {
try {
- return api.commitSnapshot(identifier, tableUuid, snapshot,
statistics);
+ return api.commitSnapshot(
+ identifier, tableUuid, baseSnapshotUuid, snapshot,
statistics);
} catch (NoSuchResourceException e) {
throw new TableNotExistException(identifier, e);
} catch (ForbiddenException e) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
index 22d86b6fca..1acd425cd4 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
@@ -59,6 +59,7 @@ public class Tag extends Snapshot {
@JsonCreator
public Tag(
@JsonProperty(FIELD_VERSION) int version,
+ @JsonProperty(FIELD_UUID) @Nullable String uuid,
@JsonProperty(FIELD_ID) long id,
@JsonProperty(FIELD_SCHEMA_ID) long schemaId,
@JsonProperty(FIELD_BASE_MANIFEST_LIST) String baseManifestList,
@@ -85,6 +86,7 @@ public class Tag extends Snapshot {
@JsonProperty(FIELD_TAG_TIME_RETAINED) @Nullable Duration
tagTimeRetained) {
super(
version,
+ uuid,
id,
schemaId,
baseManifestList,
@@ -124,6 +126,7 @@ public class Tag extends Snapshot {
Snapshot snapshot, Duration tagTimeRetained, LocalDateTime
tagCreateTime) {
return new Tag(
snapshot.version(),
+ snapshot.uuid(),
snapshot.id(),
snapshot.schemaId(),
snapshot.baseManifestList(),
@@ -152,6 +155,7 @@ public class Tag extends Snapshot {
public Snapshot trimToSnapshot() {
return new Snapshot(
version,
+ uuid,
id,
schemaId,
baseManifestList,
diff --git a/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
b/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
index 3a035c8a33..d67e9feb03 100644
--- a/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
@@ -25,30 +25,49 @@ import org.apache.paimon.utils.SnapshotManager;
import org.junit.jupiter.api.Test;
+import java.util.UUID;
+
import static org.apache.paimon.catalog.Identifier.DEFAULT_MAIN_BRANCH;
import static org.assertj.core.api.Assertions.assertThat;
/** Test for snapshots. */
public class SnapshotTest {
+ @Test
+ public void testSnapshotUuid() {
+ Snapshot snapshot =
+
org.apache.paimon.utils.SnapshotManagerTest.createSnapshotWithMillis(1L, 1000L);
+
+ assertThat(snapshot.uuid()).isNotNull();
+
assertThat(UUID.fromString(snapshot.uuid()).toString()).isEqualTo(snapshot.uuid());
+
assertThat(Snapshot.fromJson(snapshot.toJson()).uuid()).isEqualTo(snapshot.uuid());
+ assertThat(
+
org.apache.paimon.utils.SnapshotManagerTest.createSnapshotWithMillis(
+ 2L, 2000L)
+ .uuid())
+ .isNotEqualTo(snapshot.uuid());
+ }
+
@Test
public void testJsonIgnoreProperties() {
- Snapshot.fromJson(
- "{\n"
- + " \"version\" : 3,\n"
- + " \"id\" : 5,\n"
- + " \"schemaId\" : 0,\n"
- + " \"baseManifestList\" : null,\n"
- + " \"deltaManifestList\" : null,\n"
- + " \"changelogManifestList\" : null,\n"
- + " \"commitUser\" : null,\n"
- + " \"commitIdentifier\" : 0,\n"
- + " \"commitKind\" : \"APPEND\",\n"
- + " \"timeMillis\" : 1234,\n"
- + " \"totalRecordCount\" : null,\n"
- + " \"deltaRecordCount\" : null,\n"
- + " \"unknownKey\" : 22222\n"
- + "}");
+ Snapshot snapshot =
+ Snapshot.fromJson(
+ "{\n"
+ + " \"version\" : 3,\n"
+ + " \"id\" : 5,\n"
+ + " \"schemaId\" : 0,\n"
+ + " \"baseManifestList\" : null,\n"
+ + " \"deltaManifestList\" : null,\n"
+ + " \"changelogManifestList\" : null,\n"
+ + " \"commitUser\" : null,\n"
+ + " \"commitIdentifier\" : 0,\n"
+ + " \"commitKind\" : \"APPEND\",\n"
+ + " \"timeMillis\" : 1234,\n"
+ + " \"totalRecordCount\" : null,\n"
+ + " \"deltaRecordCount\" : null,\n"
+ + " \"unknownKey\" : 22222\n"
+ + "}");
+ assertThat(snapshot.uuid()).isNull();
}
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index 5e5b2ad923..96b2a8f230 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -82,6 +82,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
+import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -2872,6 +2873,7 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
Snapshot staleSnapshot =
new Snapshot(
latest.version(),
+ UUID.randomUUID().toString(),
latest.id(),
latest.schemaId(),
latest.baseManifestList(),
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
new file mode 100644
index 0000000000..1b5cfeb1cc
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/CatalogSnapshotCommitTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.catalog;
+
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.utils.SnapshotManagerTest;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Tests for {@link CatalogSnapshotCommit}. */
+public class CatalogSnapshotCommitTest {
+
+ @Test
+ public void testCommitForwardsBaseSnapshotUuid() throws Exception {
+ Catalog catalog = mock(Catalog.class);
+ Identifier identifier = Identifier.create("database", "table");
+ Identifier branchIdentifier = new Identifier("database", "table",
"branch");
+ Snapshot snapshot = SnapshotManagerTest.createSnapshotWithMillis(2L,
1000L);
+ when(catalog.commitSnapshot(
+ branchIdentifier,
+ "table-uuid",
+ "base-snapshot-uuid",
+ snapshot,
+ Collections.emptyList()))
+ .thenReturn(true);
+
+ CatalogSnapshotCommit commit = new CatalogSnapshotCommit(catalog,
identifier, "table-uuid");
+
+ assertThat(commit.commit("base-snapshot-uuid", snapshot, "branch",
Collections.emptyList()))
+ .isTrue();
+ verify(catalog)
+ .commitSnapshot(
+ branchIdentifier,
+ "table-uuid",
+ "base-snapshot-uuid",
+ snapshot,
+ Collections.emptyList());
+ }
+}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
index 0924b94203..1eef5c53cf 100644
---
a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
@@ -58,7 +58,7 @@ public class RenamingSnapshotCommitTest {
Snapshot snapshot = createSnapshot(1L);
- boolean committed = commit.commit(snapshot, "main",
Collections.emptyList());
+ boolean committed = commit.commit(null, snapshot, "main",
Collections.emptyList());
assertThat(committed).isTrue();
@@ -91,7 +91,7 @@ public class RenamingSnapshotCommitTest {
IOException ex =
assertThrows(
IOException.class,
- () -> commit.commit(snapshot, "main",
Collections.emptyList()));
+ () -> commit.commit(null, snapshot, "main",
Collections.emptyList()));
assertThat(ex)
.hasMessageContaining("Commit snapshot " + snapshot.id() + "
failed")
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 97f4a5b804..98606bfb10 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -84,6 +84,8 @@ import org.junit.jupiter.params.provider.ValueSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.annotation.Nullable;
+
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -1962,12 +1964,13 @@ public class FileStoreCommitTest {
@Override
public boolean commit(
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
List<org.apache.paimon.partition.PartitionStatistics>
statistics)
throws Exception {
if (commitAttempt >= conflictDeltaFilesByAttempt.size()) {
- return delegate.commit(snapshot, branch, statistics);
+ return delegate.commit(baseSnapshotUuid, snapshot, branch,
statistics);
}
List<ManifestEntry> conflictDeltaFiles =
conflictDeltaFilesByAttempt.get(commitAttempt);
@@ -2042,8 +2045,15 @@ public class FileStoreCommitTest {
previousSnapshot == null ? null :
previousSnapshot.watermark(),
previousSnapshot == null ? null :
previousSnapshot.statistics(),
previousSnapshot == null ? null :
previousSnapshot.properties(),
- previousSnapshot == null ? null :
previousSnapshot.nextRowId());
- assertThat(delegate.commit(conflictSnapshot, branch,
Collections.emptyList())).isTrue();
+ previousSnapshot == null ? null :
previousSnapshot.nextRowId(),
+ null);
+ assertThat(
+ delegate.commit(
+ baseSnapshotUuid,
+ conflictSnapshot,
+ branch,
+ Collections.emptyList()))
+ .isTrue();
return false;
}
@@ -2100,11 +2110,12 @@ public class FileStoreCommitTest {
@Override
public boolean commit(
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
String branch,
List<org.apache.paimon.partition.PartitionStatistics>
statistics)
throws Exception {
- boolean committed = delegate.commit(snapshot, branch, statistics);
+ boolean committed = delegate.commit(baseSnapshotUuid, snapshot,
branch, statistics);
if (firstCommit) {
firstCommit = false;
assertThat(committed).isTrue();
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
index b6ae35561f..c571b19c38 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
@@ -1050,6 +1050,7 @@ public class RESTCatalogServer {
return commitSnapshot(
identifier,
requestBody.getTableId(),
+ requestBody.getBaseSnapshotUuid(),
requestBody.getSnapshot(),
requestBody.getStatistics());
}
@@ -2756,9 +2757,10 @@ public class RESTCatalogServer {
public static volatile boolean commitSuccessThrowException = false;
- private MockResponse commitSnapshot(
+ private synchronized MockResponse commitSnapshot(
Identifier identifier,
String tableId,
+ @Nullable String baseSnapshotUuid,
Snapshot snapshot,
List<PartitionStatistics> statistics)
throws Catalog.TableNotExistException {
@@ -2777,6 +2779,12 @@ public class RESTCatalogServer {
if (!tableId.equals(table.catalogEnvironment().uuid())) {
throw new Catalog.TableNotExistException(identifier);
}
+ TableSnapshot currentSnapshot =
tableLatestSnapshotStore.get(identifier.getFullName());
+ String currentSnapshotUuid =
+ currentSnapshot == null ? null :
currentSnapshot.snapshot().uuid();
+ if (!Objects.equals(currentSnapshotUuid, baseSnapshotUuid)) {
+ return mockResponse(new CommitTableResponse(false), 200);
+ }
RenamingSnapshotCommit commit =
new RenamingSnapshotCommit(table.snapshotManager(),
Lock.empty());
String branchName = identifier.getBranchName();
@@ -2785,7 +2793,8 @@ public class RESTCatalogServer {
}
TableSnapshot tableSnapshot;
try {
- boolean success = commit.commit(snapshot, branchName,
Collections.emptyList());
+ boolean success =
+ commit.commit(baseSnapshotUuid, snapshot, branchName,
Collections.emptyList());
if (!success) {
return mockResponse(new CommitTableResponse(success), 200);
}
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
index 1438b49f17..8fc5e38a25 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -406,6 +406,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
restCatalog.commitSnapshot(
identifier,
"",
+ null,
createSnapshotWithMillis(1L,
System.currentTimeMillis()),
new ArrayList<PartitionStatistics>()));
}
@@ -2237,6 +2238,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
restCatalog.commitSnapshot(
hasSnapshotTableIdentifier,
"",
+ null,
createSnapshotWithMillis(1L,
System.currentTimeMillis()),
new ArrayList<>()));
@@ -2248,6 +2250,7 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
restCatalog.commitSnapshot(
hasSnapshotTableIdentifier,
"unknown_id",
+ null,
createSnapshotWithMillis(1L,
System.currentTimeMillis()),
new ArrayList<>()));
@@ -2283,6 +2286,39 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
assertThat(snapshot).isEmpty();
}
+ @Test
+ void testCommitSnapshotChecksBaseUuid() throws Exception {
+ RESTCatalogServer.commitSuccessThrowException = false;
+ Identifier identifier = Identifier.create("test_db_a",
"snapshot_uuid_commit");
+ createTable(identifier, Maps.newHashMap(), Lists.newArrayList("col1"));
+ Table table = catalog.getTable(identifier);
+
+ Snapshot first = createSnapshotWithMillis(1L,
System.currentTimeMillis());
+ assertThat(
+ restCatalog.commitSnapshot(
+ identifier, table.uuid(), null, first,
Collections.emptyList()))
+ .isTrue();
+
+ Snapshot second = createSnapshotWithMillis(2L,
System.currentTimeMillis());
+ assertThat(
+ restCatalog.commitSnapshot(
+ identifier,
+ table.uuid(),
+ "wrong-base-snapshot-uuid",
+ second,
+ Collections.emptyList()))
+ .isFalse();
+ assertThat(
+ restCatalog.commitSnapshot(
+ identifier,
+ table.uuid(),
+ first.uuid(),
+ second,
+ Collections.emptyList()))
+ .isTrue();
+
assertThat(restCatalog.loadSnapshot(identifier).get().snapshot()).isEqualTo(second);
+ }
+
@Test
public void testTableRollback() throws Exception {
Identifier identifier = Identifier.create("test_rollback",
"table_for_rollback");
diff --git a/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
b/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
index 436e86c211..459e006f17 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
@@ -32,6 +32,8 @@ public class TagTest {
private final Snapshot snapshot =
new Snapshot(
+ 3,
+ "snapshot-uuid",
0,
0L,
null,
@@ -60,6 +62,7 @@ public class TagTest {
assertEquals(
"{\n"
+ " \"version\" : 3,\n"
+ + " \"uuid\" : \"snapshot-uuid\",\n"
+ " \"id\" : 0,\n"
+ " \"schemaId\" : 0,\n"
+ " \"baseManifestList\" : null,\n"
@@ -85,6 +88,7 @@ public class TagTest {
assertEquals(
"{\n"
+ " \"version\" : 3,\n"
+ + " \"uuid\" : \"snapshot-uuid\",\n"
+ " \"id\" : 0,\n"
+ " \"schemaId\" : 0,\n"
+ " \"baseManifestList\" : null,\n"
diff --git a/paimon-python/pypaimon/api/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index 39a01da93e..e983618387 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -74,12 +74,16 @@ class CreateTableRequest(RESTRequest):
@dataclass
class CommitTableRequest(RESTRequest):
FIELD_TABLE_ID = "tableId"
+ FIELD_BASE_SNAPSHOT_UUID = "baseSnapshotUuid"
FIELD_SNAPSHOT = "snapshot"
FIELD_STATISTICS = "statistics"
table_id: Optional[str] = json_field(FIELD_TABLE_ID)
snapshot: Snapshot = json_field(FIELD_SNAPSHOT)
statistics: List[PartitionStatistics] = json_field(FIELD_STATISTICS)
+ base_snapshot_uuid: Optional[str] = json_field(
+ FIELD_BASE_SNAPSHOT_UUID, default=None
+ )
@dataclass
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index e1ee43a616..96a0321794 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -346,6 +346,7 @@ class RESTApi:
self,
identifier: Identifier,
table_uuid: Optional[str],
+ base_snapshot_uuid: Optional[str],
snapshot: Snapshot,
statistics: List[PartitionStatistics]
) -> bool:
@@ -355,6 +356,7 @@ class RESTApi:
Args:
identifier: Database name and table name
table_uuid: UUID of the table to avoid wrong commit
+ base_snapshot_uuid: UUID of the snapshot on which the commit is
based
snapshot: Snapshot for committing
statistics: Statistics for this snapshot incremental
@@ -371,7 +373,12 @@ class RESTApi:
if statistics is None:
raise ValueError("Statistics cannot be None")
- request = CommitTableRequest(table_uuid, snapshot, statistics)
+ request = CommitTableRequest(
+ table_id=table_uuid,
+ snapshot=snapshot,
+ statistics=statistics,
+ base_snapshot_uuid=base_snapshot_uuid,
+ )
response = self.client.post_with_response_type(
self.resource_paths.commit_table(
database_name, table_name),
diff --git a/paimon-python/pypaimon/catalog/catalog.py
b/paimon-python/pypaimon/catalog/catalog.py
index c52b65b4ae..d3c9c6e17f 100644
--- a/paimon-python/pypaimon/catalog/catalog.py
+++ b/paimon-python/pypaimon/catalog/catalog.py
@@ -157,6 +157,7 @@ class Catalog(ABC):
self,
identifier: Identifier,
table_uuid: Optional[str],
+ base_snapshot_uuid: Optional[str],
snapshot: Snapshot,
statistics: List[PartitionStatistics]
) -> bool:
@@ -166,6 +167,7 @@ class Catalog(ABC):
Args:
identifier: Path of the table
table_uuid: UUID of the table to avoid wrong commit
+ base_snapshot_uuid: UUID of the snapshot on which the commit is
based
snapshot: Snapshot to be committed
statistics: Statistics information of this change
diff --git a/paimon-python/pypaimon/catalog/filesystem_catalog.py
b/paimon-python/pypaimon/catalog/filesystem_catalog.py
index 76ba5e7798..be5f7e972f 100644
--- a/paimon-python/pypaimon/catalog/filesystem_catalog.py
+++ b/paimon-python/pypaimon/catalog/filesystem_catalog.py
@@ -279,6 +279,7 @@ class FileSystemCatalog(Catalog):
self,
identifier: Identifier,
table_uuid: Optional[str],
+ base_snapshot_uuid: Optional[str],
snapshot: Snapshot,
statistics: List[PartitionStatistics]
) -> bool:
diff --git a/paimon-python/pypaimon/catalog/jdbc_catalog.py
b/paimon-python/pypaimon/catalog/jdbc_catalog.py
index ccf5dbfa1a..5ab4a41702 100644
--- a/paimon-python/pypaimon/catalog/jdbc_catalog.py
+++ b/paimon-python/pypaimon/catalog/jdbc_catalog.py
@@ -545,6 +545,7 @@ class JdbcCatalog(Catalog):
self,
identifier: Identifier,
table_uuid: Optional[str],
+ base_snapshot_uuid: Optional[str],
snapshot: Snapshot,
statistics: List[PartitionStatistics]
) -> bool:
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 612db94aa3..35db60061d 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -100,6 +100,7 @@ class RESTCatalog(Catalog):
self,
identifier: Identifier,
table_uuid: Optional[str],
+ base_snapshot_uuid: Optional[str],
snapshot: Snapshot,
statistics: List[PartitionStatistics]
) -> bool:
@@ -109,6 +110,7 @@ class RESTCatalog(Catalog):
Args:
identifier: Path of the table
table_uuid: UUID of the table to avoid wrong commit
+ base_snapshot_uuid: UUID of the snapshot on which the commit is
based
snapshot: Snapshot to be committed
statistics: Statistics information of this change
@@ -120,7 +122,13 @@ class RESTCatalog(Catalog):
TableNoPermissionException: If no permission to access this table
"""
try:
- return self.rest_api.commit_snapshot(identifier, table_uuid,
snapshot, statistics)
+ return self.rest_api.commit_snapshot(
+ identifier,
+ table_uuid,
+ base_snapshot_uuid,
+ snapshot,
+ statistics,
+ )
except NoSuchResourceException as e:
raise TableNotExistException(identifier) from e
except ForbiddenException as e:
diff --git a/paimon-python/pypaimon/changelog/changelog.py
b/paimon-python/pypaimon/changelog/changelog.py
index 3c32c832c0..f562395390 100644
--- a/paimon-python/pypaimon/changelog/changelog.py
+++ b/paimon-python/pypaimon/changelog/changelog.py
@@ -36,6 +36,7 @@ class Changelog(Snapshot):
"""Create a Changelog from a Snapshot instance."""
return Changelog(
version=snapshot.version,
+ uuid=snapshot.uuid,
id=snapshot.id,
schema_id=snapshot.schema_id,
base_manifest_list=snapshot.base_manifest_list,
diff --git a/paimon-python/pypaimon/common/json_util.py
b/paimon-python/pypaimon/common/json_util.py
index de1421d7ad..6effec7d1e 100644
--- a/paimon-python/pypaimon/common/json_util.py
+++ b/paimon-python/pypaimon/common/json_util.py
@@ -24,7 +24,8 @@ T = TypeVar("T")
def json_field(json_name: str, **kwargs):
"""Create a field with custom JSON name"""
- return field(metadata={"json_name": json_name}, **kwargs)
+ metadata = kwargs.pop("metadata", {})
+ return field(metadata={"json_name": json_name, **metadata}, **kwargs)
def optional_json_field(json_name: str, json_include: str):
@@ -164,4 +165,14 @@ class JSON:
else:
kwargs[field_name] = value
+ for field_info in fields(target_class):
+ json_name = field_info.metadata.get("json_name", field_info.name)
+ if (
+ json_name not in data
+ and "json_missing_default" in field_info.metadata
+ ):
+ kwargs[field_info.name] = field_info.metadata[
+ "json_missing_default"
+ ]
+
return target_class(**kwargs)
diff --git a/paimon-python/pypaimon/snapshot/catalog_snapshot_commit.py
b/paimon-python/pypaimon/snapshot/catalog_snapshot_commit.py
index 9850707d34..f534d61ada 100755
--- a/paimon-python/pypaimon/snapshot/catalog_snapshot_commit.py
+++ b/paimon-python/pypaimon/snapshot/catalog_snapshot_commit.py
@@ -16,7 +16,7 @@
# under the License.
import logging
-from typing import List
+from typing import List, Optional
from pypaimon.catalog.catalog import Catalog
from pypaimon.common.identifier import Identifier
@@ -43,11 +43,17 @@ class CatalogSnapshotCommit(SnapshotCommit):
self.identifier = identifier
self.uuid = uuid
- def commit(self, snapshot: Snapshot, statistics:
List[PartitionStatistics]) -> bool:
+ def commit(
+ self,
+ base_snapshot_uuid: Optional[str],
+ snapshot: Snapshot,
+ statistics: List[PartitionStatistics],
+ ) -> bool:
"""
Commit the snapshot using the catalog.
Args:
+ base_snapshot_uuid: UUID of the snapshot on which the commit is
based
snapshot: The snapshot to commit
statistics: List of partition statistics
@@ -59,7 +65,13 @@ class CatalogSnapshotCommit(SnapshotCommit):
"""
# Call catalog's commit_snapshot method
if hasattr(self.catalog, 'commit_snapshot'):
- success = self.catalog.commit_snapshot(self.identifier, self.uuid,
snapshot, statistics)
+ success = self.catalog.commit_snapshot(
+ self.identifier,
+ self.uuid,
+ base_snapshot_uuid,
+ snapshot,
+ statistics,
+ )
if success:
logger.info("Catalog snapshot commit succeeded for %s,
snapshot id %d", self.identifier, snapshot.id)
return success
diff --git a/paimon-python/pypaimon/snapshot/renaming_snapshot_commit.py
b/paimon-python/pypaimon/snapshot/renaming_snapshot_commit.py
index 93c5121cc7..8b45e75ea7 100644
--- a/paimon-python/pypaimon/snapshot/renaming_snapshot_commit.py
+++ b/paimon-python/pypaimon/snapshot/renaming_snapshot_commit.py
@@ -16,7 +16,7 @@
# under the License.
import logging
-from typing import List
+from typing import List, Optional
from pypaimon.common.file_io import FileIO
@@ -47,11 +47,17 @@ class RenamingSnapshotCommit(SnapshotCommit):
self.snapshot_manager = snapshot_manager
self.file_io: FileIO = snapshot_manager.file_io
- def commit(self, snapshot: Snapshot, statistics:
List[PartitionStatistics]) -> bool:
+ def commit(
+ self,
+ base_snapshot_uuid: Optional[str],
+ snapshot: Snapshot,
+ statistics: List[PartitionStatistics],
+ ) -> bool:
"""
Commit the snapshot using file renaming.
Args:
+ base_snapshot_uuid: UUID of the base snapshot (unused for renaming
commits)
snapshot: The snapshot to commit
statistics: List of partition statistics (currently unused but
kept for interface compatibility)
diff --git a/paimon-python/pypaimon/snapshot/snapshot.py
b/paimon-python/pypaimon/snapshot/snapshot.py
index 85dcc4e928..407f27f0d7 100644
--- a/paimon-python/pypaimon/snapshot/snapshot.py
+++ b/paimon-python/pypaimon/snapshot/snapshot.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import uuid as uuid_module
from dataclasses import dataclass
from typing import Optional
@@ -48,3 +49,11 @@ class Snapshot:
statistics: Optional[str] = optional_json_field("statistics", "non_null")
next_row_id: Optional[int] = optional_json_field("nextRowId", "non_null")
properties: Optional[dict] = optional_json_field("properties", "non_null")
+ uuid: Optional[str] = json_field(
+ "uuid",
+ default_factory=lambda: str(uuid_module.uuid4()),
+ metadata={
+ "json_include": "non_null",
+ "json_missing_default": None,
+ },
+ )
diff --git a/paimon-python/pypaimon/snapshot/snapshot_commit.py
b/paimon-python/pypaimon/snapshot/snapshot_commit.py
index 5b4c996b83..dd9c90aead 100644
--- a/paimon-python/pypaimon/snapshot/snapshot_commit.py
+++ b/paimon-python/pypaimon/snapshot/snapshot_commit.py
@@ -18,7 +18,7 @@
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
-from typing import Dict, List
+from typing import Dict, List, Optional
from pypaimon.common.json_util import json_field
from pypaimon.snapshot.snapshot import Snapshot
@@ -72,11 +72,17 @@ class SnapshotCommit(ABC):
"""Interface to commit snapshot atomically."""
@abstractmethod
- def commit(self, snapshot: Snapshot, statistics:
List[PartitionStatistics]) -> bool:
+ def commit(
+ self,
+ base_snapshot_uuid: Optional[str],
+ snapshot: Snapshot,
+ statistics: List[PartitionStatistics],
+ ) -> bool:
"""
Commit the given snapshot.
Args:
+ base_snapshot_uuid: UUID of the snapshot on which the commit is
based
snapshot: The snapshot to commit
statistics: List of partition statistics
diff --git a/paimon-python/pypaimon/tag/tag.py
b/paimon-python/pypaimon/tag/tag.py
index d793839374..80b4d5c3f1 100644
--- a/paimon-python/pypaimon/tag/tag.py
+++ b/paimon-python/pypaimon/tag/tag.py
@@ -62,6 +62,7 @@ class Tag(Snapshot):
"""
return Tag(
version=snapshot.version,
+ uuid=snapshot.uuid,
id=snapshot.id,
schema_id=snapshot.schema_id,
base_manifest_list=snapshot.base_manifest_list,
@@ -90,6 +91,7 @@ class Tag(Snapshot):
"""Convert this Tag to a Snapshot"""
return Snapshot(
version=self.version,
+ uuid=self.uuid,
id=self.id,
schema_id=self.schema_id,
base_manifest_list=self.base_manifest_list,
diff --git a/paimon-python/pypaimon/tests/file_store_commit_test.py
b/paimon-python/pypaimon/tests/file_store_commit_test.py
index 98c8867fe6..c6451f0a9b 100644
--- a/paimon-python/pypaimon/tests/file_store_commit_test.py
+++ b/paimon-python/pypaimon/tests/file_store_commit_test.py
@@ -16,6 +16,7 @@
# under the License.
import unittest
+import uuid
from datetime import datetime
from unittest.mock import MagicMock, Mock, patch
@@ -425,6 +426,7 @@ class TestFileStoreCommit(unittest.TestCase):
latest_snapshot = Mock()
latest_snapshot.id = 3
+ latest_snapshot.uuid = "base-snapshot-uuid"
latest_snapshot.total_record_count = 10
latest_snapshot.index_manifest = "index-manifest-existing"
@@ -443,10 +445,16 @@ class TestFileStoreCommit(unittest.TestCase):
)
self.assertTrue(result.is_success())
+ self.assertEqual(
+ "base-snapshot-uuid",
+ snapshot_commit.commit.call_args[0][0],
+ )
+ committed_snapshot = snapshot_commit.commit.call_args[0][1]
self.assertEqual(
"index-manifest-existing",
- snapshot_commit.commit.call_args[0][0].index_manifest
+ committed_snapshot.index_manifest
)
+ self.assertEqual(str(uuid.UUID(committed_snapshot.uuid)),
committed_snapshot.uuid)
def test_null_partition_value(
self, mock_manifest_list_manager, mock_manifest_file_manager):
diff --git a/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
b/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
index 356eaaa883..861ae19eac 100644
--- a/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
+++ b/paimon-python/pypaimon/tests/overwrite_changes_cache_test.py
@@ -103,12 +103,12 @@ class OverwriteChangesCacheTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
self._append(pd.DataFrame({'f0': [99], 'f1':
[f'x{cas["fails"]}']}))
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
OverwriteChangesProvider._full_scan_manifest_entries = spy_full
@@ -174,12 +174,12 @@ class OverwriteChangesCacheTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
self._append(pd.DataFrame({'f0': [1], 'f1':
[f'y{cas["fails"]}']}))
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
OverwriteChangesProvider._full_scan_manifest_entries = spy_full
@@ -234,12 +234,12 @@ class OverwriteChangesCacheTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
concurrent_fn(cas['fails'])
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
diff --git a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
index 738d455b85..e18439ea84 100644
--- a/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
+++ b/paimon-python/pypaimon/tests/overwrite_commit_conflict_test.py
@@ -114,14 +114,14 @@ class OverwriteCommitConflictTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
# Each conflict appends to an unrelated partition (f0=99),
advancing
# latest, then fails our CAS.
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
self._append(pd.DataFrame({'f0': [99], 'f1':
[f'x{cas["fails"]}']}))
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
@@ -178,12 +178,12 @@ class OverwriteCommitConflictTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
self._append(pd.DataFrame({'f0': [1], 'f1':
[f'y{cas["fails"]}']}))
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
@@ -255,12 +255,12 @@ class OverwriteCommitConflictTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
self._append(pd.DataFrame({'f0': [99], 'f1': ['x']}))
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
@@ -302,12 +302,12 @@ class OverwriteCommitConflictTest(unittest.TestCase):
orig_cas = fsc.snapshot_commit.commit
cas = {'fails': 0}
- def patched_cas(snapshot, statistics):
+ def patched_cas(base_snapshot_uuid, snapshot, statistics):
if snapshot.commit_kind == "OVERWRITE" and cas['fails'] < K:
cas['fails'] += 1
concurrent_fn(f'z{cas["fails"]}')
return False
- return orig_cas(snapshot, statistics)
+ return orig_cas(base_snapshot_uuid, snapshot, statistics)
fsc.snapshot_commit.commit = patched_cas
diff --git a/paimon-python/pypaimon/tests/rest/api_test.py
b/paimon-python/pypaimon/tests/rest/api_test.py
index a951461ec7..da0220e7e0 100644
--- a/paimon-python/pypaimon/tests/rest/api_test.py
+++ b/paimon-python/pypaimon/tests/rest/api_test.py
@@ -330,17 +330,17 @@ class ApiTest(unittest.TestCase):
# Test commit_snapshot with None identifier
with self.assertRaises(ValueError) as context:
- rest_api.commit_snapshot(None, "uuid", Mock(), [])
+ rest_api.commit_snapshot(None, "uuid", None, Mock(), [])
self.assertIn("Identifier cannot be None", str(context.exception))
# Test commit_snapshot with None snapshot
with self.assertRaises(ValueError) as context:
- rest_api.commit_snapshot(Mock(), "uuid", None, [])
+ rest_api.commit_snapshot(Mock(), "uuid", None, None, [])
self.assertIn("Snapshot cannot be None", str(context.exception))
# Test commit_snapshot with None statistics
with self.assertRaises(ValueError) as context:
- rest_api.commit_snapshot(Mock(), "uuid", Mock(), None)
+ rest_api.commit_snapshot(Mock(), "uuid", None, Mock(), None)
self.assertIn("Statistics cannot be None", str(context.exception))
def test_list_tables_paged_with_table_type_param(self):
diff --git
a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
index 975c6f643b..4b7ac5d8b0 100644
--- a/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_catalog_commit_snapshot_test.py
@@ -16,8 +16,11 @@
# specific language governing permissions and limitations
# under the License.
+import json
import time
import unittest
+import uuid
+from dataclasses import replace
from unittest.mock import Mock, patch
import pyarrow as pa
@@ -30,6 +33,7 @@ from pypaimon.catalog.catalog_context import CatalogContext
from pypaimon.catalog.catalog_exception import TableNotExistException
from pypaimon.catalog.rest.rest_catalog import RESTCatalog
from pypaimon.common.identifier import Identifier
+from pypaimon.common.json_util import JSON
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
from pypaimon.tests.rest.rest_base_test import RESTBaseTest
@@ -102,6 +106,7 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
result = catalog.commit_snapshot(
self.identifier,
"test-uuid",
+ "base-snapshot-uuid",
self.test_snapshot,
self.test_statistics
)
@@ -113,6 +118,7 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
mock_api_instance.commit_snapshot.assert_called_once_with(
self.identifier,
"test-uuid",
+ "base-snapshot-uuid",
self.test_snapshot,
self.test_statistics
)
@@ -134,6 +140,7 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
catalog.commit_snapshot(
self.identifier,
"test-uuid",
+ "base-snapshot-uuid",
self.test_snapshot,
self.test_statistics
)
@@ -155,6 +162,7 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
catalog.commit_snapshot(
self.identifier,
"test-uuid",
+ "base-snapshot-uuid",
self.test_snapshot,
self.test_statistics
)
@@ -168,15 +176,39 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
request = CommitTableRequest(
table_id="test-uuid",
+ base_snapshot_uuid="base-snapshot-uuid",
snapshot=self.test_snapshot,
statistics=self.test_statistics
)
# Verify request fields
self.assertEqual(request.table_id, "test-uuid")
+ self.assertEqual(request.base_snapshot_uuid, "base-snapshot-uuid")
self.assertEqual(request.snapshot, self.test_snapshot)
self.assertEqual(request.statistics, self.test_statistics)
+ legacy_request = json.loads(JSON.to_json(request))
+ legacy_request.pop("baseSnapshotUuid")
+ self.assertIsNone(
+ JSON.from_json(
+ json.dumps(legacy_request), CommitTableRequest
+ ).base_snapshot_uuid
+ )
+
+ def test_snapshot_uuid_round_trip_and_legacy_compatibility(self):
+ snapshot_uuid = self.test_snapshot.uuid
+ self.assertEqual(str(uuid.UUID(snapshot_uuid)), snapshot_uuid)
+ self.assertEqual(
+ snapshot_uuid,
+ JSON.from_json(JSON.to_json(self.test_snapshot), Snapshot).uuid,
+ )
+
+ legacy_snapshot = json.loads(JSON.to_json(self.test_snapshot))
+ legacy_snapshot.pop("uuid")
+ self.assertIsNone(
+ JSON.from_json(json.dumps(legacy_snapshot), Snapshot).uuid
+ )
+
def test_commit_table_response_creation(self):
"""Test CommitTableResponse creation."""
from pypaimon.api.api_response import CommitTableResponse
@@ -209,6 +241,7 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
result = api.commit_snapshot(
self.identifier,
"test-uuid",
+ "base-snapshot-uuid",
self.test_snapshot,
self.test_statistics
)
@@ -218,6 +251,10 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
# Verify client was called correctly
mock_client.post_with_response_type.assert_called_once()
+ request = mock_client.post_with_response_type.call_args[0][1]
+ self.assertEqual(
+ request.base_snapshot_uuid, "base-snapshot-uuid"
+ )
def test_rest_catalog_commit_snapshot_with_lance_format(self):
"""Test snapshot commit with Lance format table."""
@@ -296,6 +333,42 @@ class TestRESTCatalogCommitSnapshot(unittest.TestCase):
class TestRESTCommit(RESTBaseTest):
+ def test_commit_snapshot_checks_base_uuid(self):
+ latest = self.table.snapshot_manager().get_latest_snapshot()
+ next_snapshot = replace(
+ latest,
+ id=latest.id + 1,
+ uuid=str(uuid.uuid4()),
+ )
+ table_uuid = self.table.catalog_environment.uuid
+
+ self.assertFalse(
+ self.rest_catalog.commit_snapshot(
+ self.table.identifier,
+ table_uuid,
+ "wrong-base-snapshot-uuid",
+ next_snapshot,
+ [],
+ )
+ )
+ self.assertEqual(
+ latest.id,
+ self.table.snapshot_manager().get_latest_snapshot().id,
+ )
+ self.assertTrue(
+ self.rest_catalog.commit_snapshot(
+ self.table.identifier,
+ table_uuid,
+ latest.uuid,
+ next_snapshot,
+ [],
+ )
+ )
+ self.assertEqual(
+ next_snapshot,
+ self.table.snapshot_manager().get_latest_snapshot(),
+ )
+
def test_multiple_row_tracking_commits_preserve_all_rows(self):
pa_schema = pa.schema([('id', pa.int32())])
schema = Schema.from_pyarrow_schema(
@@ -347,8 +420,8 @@ class TestRESTCommit(RESTBaseTest):
real_commit = tc.file_store_commit.snapshot_commit.commit
- def commit_then_raise(sn, st):
- real_commit(sn, st)
+ def commit_then_raise(base_snapshot_uuid, sn, st):
+ real_commit(base_snapshot_uuid, sn, st)
raise RuntimeError("simulated")
with patch.object(tc.file_store_commit.snapshot_commit, 'commit',
side_effect=commit_then_raise):
diff --git a/paimon-python/pypaimon/tests/rest/rest_permission_test.py
b/paimon-python/pypaimon/tests/rest/rest_permission_test.py
index def1f4060d..9f4f7d5864 100644
--- a/paimon-python/pypaimon/tests/rest/rest_permission_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_permission_test.py
@@ -156,6 +156,7 @@ class RESTPermissionTest(RESTBaseTest):
self.rest_catalog.commit_snapshot(
identifier,
"test-uuid",
+ None,
test_snapshot,
test_statistics
)
diff --git a/paimon-python/pypaimon/tests/rest/rest_server.py
b/paimon-python/pypaimon/tests/rest/rest_server.py
index d1fbd528b1..69d4361bb5 100755
--- a/paimon-python/pypaimon/tests/rest/rest_server.py
+++ b/paimon-python/pypaimon/tests/rest/rest_server.py
@@ -1082,6 +1082,16 @@ class RESTCatalogServer:
ErrorResponse("SNAPSHOT", None, "Snapshot is required for
commit operation", 400), 400
)
+ table = self._get_file_table(identifier)
+ current_snapshot = table.snapshot_manager().get_latest_snapshot()
+ current_snapshot_uuid = (
+ current_snapshot.uuid if current_snapshot else None
+ )
+ if current_snapshot_uuid != commit_request.base_snapshot_uuid:
+ return self._mock_response(
+ CommitTableResponse(success=False), 200
+ )
+
# Write snapshot to file system
self._write_snapshot_files(identifier, commit_request.snapshot,
commit_request.statistics)
diff --git a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
index 30d2dcaa85..61fece4934 100644
--- a/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
+++ b/paimon-python/pypaimon/tests/write/dynamic_bucket_test.py
@@ -621,7 +621,8 @@ class DynamicBucketTest(unittest.TestCase):
original_snapshot_commit = commit.file_store_commit.snapshot_commit
calls = 0
- def lose_first_compare_and_set(snapshot, statistics):
+ def lose_first_compare_and_set(
+ base_snapshot_uuid, snapshot, statistics):
nonlocal calls
calls += 1
concurrent_writer, concurrent_commit, concurrent_messages = (
diff --git a/paimon-python/pypaimon/write/file_store_commit.py
b/paimon-python/pypaimon/write/file_store_commit.py
index 801771bf14..8c7484071a 100644
--- a/paimon-python/pypaimon/write/file_store_commit.py
+++ b/paimon-python/pypaimon/write/file_store_commit.py
@@ -615,7 +615,11 @@ class FileStoreCommit:
# Use SnapshotCommit for atomic commit
try:
with self.snapshot_commit:
- success = self.snapshot_commit.commit(snapshot_data,
statistics)
+ success = self.snapshot_commit.commit(
+ latest_snapshot.uuid if latest_snapshot else None,
+ snapshot_data,
+ statistics,
+ )
if not success:
commit_time_s = (int(time.time() * 1000) - start_millis) /
1000
logger.warning(
