[core] Fix calculation of lastModifiedCommitIdentifier of writer

apache · Mar 13, 2024 · 760adb4 · 760adb4
1 parent ec31a23
commit 760adb4
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 4 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java
@@ -177,7 +177,7 @@ public List<CommitMessage> prepareCommit(boolean waitCompaction, long commitIden
         } else {
             latestCommittedIdentifier =
                     snapshotManager
-                            .latestSnapshotOfUser(commitUser)
+                            .latestSnapshotOfUser(commitUser, branchName)
                             .map(Snapshot::commitIdentifier)
                             .orElse(Long.MIN_VALUE);
         }

diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -120,7 +120,15 @@ public KeyValueFileStoreWrite(
             KeyValueFieldsExtractor extractor,
             String tableName,
             String branchName) {
-        super(commitUser, snapshotManager, scan, options, indexFactory, deletionVectorsMaintainerFactory, tableName, branchName);
+        super(
+                commitUser,
+                snapshotManager,
+                scan,
+                options,
+                indexFactory,
+                deletionVectorsMaintainerFactory,
+                tableName,
+                branchName);
         this.fileIO = fileIO;
         this.keyType = keyType;
         this.valueType = valueType;

diff --git a/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java b/paimon-core/src/main/java/org/apache/paimon/utils/SnapshotManager.java
@@ -299,7 +299,11 @@ private Predicate<Path> nonSnapshotFileFilter() {
     }
 
     public Optional<Snapshot> latestSnapshotOfUser(String user) {
-        Long latestId = latestSnapshotId();
+        return latestSnapshotOfUser(user, DEFAULT_MAIN_BRANCH);
+    }
+
+    public Optional<Snapshot> latestSnapshotOfUser(String user, String branchName) {
+        Long latestId = latestSnapshotId(branchName);
         if (latestId == null) {
             return Optional.empty();
         }
@@ -310,7 +314,7 @@ public Optional<Snapshot> latestSnapshotOfUser(String user) {
                         "Latest snapshot id is not null, but earliest snapshot id is null. "
                                 + "This is unexpected.");
         for (long id = latestId; id >= earliestId; id--) {
-            Snapshot snapshot = snapshot(id);
+            Snapshot snapshot = snapshot(branchName, id);
             if (user.equals(snapshot.commitUser())) {
                 return Optional.of(snapshot);
             }