fix minus

apache · Dec 13, 2023 · ec60f1c · ec60f1c
1 parent 2d300d7
commit ec60f1c
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 50 deletions.
diff --git a/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java b/...-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreCommitter.java
@@ -69,18 +69,29 @@ public boolean forceCreatingSnapshot() {
     public ManifestCommittable combine(
             long checkpointId, long watermark, List<Committable> committables) {
         ManifestCommittable manifestCommittable = new ManifestCommittable(checkpointId, watermark);
-        addFile(manifestCommittable, committables);
-        return manifestCommittable;
+        return combine(checkpointId, watermark, manifestCommittable, committables);
     }
 
     @Override
     public ManifestCommittable combine(
             long checkpointId,
             long watermark,
-            ManifestCommittable oldCommitable,
+            ManifestCommittable manifestCommittable,
             List<Committable> committables) {
-        addFile(oldCommitable, committables);
-        return oldCommitable;
+        for (Committable committable : committables) {
+            switch (committable.kind()) {
+                case FILE:
+                    CommitMessage file = (CommitMessage) committable.wrappedCommittable();
+                    manifestCommittable.addFileCommittable(file);
+                    break;
+                case LOG_OFFSET:
+                    LogOffsetCommittable offset =
+                            (LogOffsetCommittable) committable.wrappedCommittable();
+                    manifestCommittable.addLogOffset(offset.bucket(), offset.offset());
+                    break;
+            }
+        }
+        return manifestCommittable;
     }
 
     @Override
@@ -109,22 +120,6 @@ public void close() throws Exception {
         commit.close();
     }
 
-    private void addFile(ManifestCommittable manifestCommittable, List<Committable> committables) {
-        for (Committable committable : committables) {
-            switch (committable.kind()) {
-                case FILE:
-                    CommitMessage file = (CommitMessage) committable.wrappedCommittable();
-                    manifestCommittable.addFileCommittable(file);
-                    break;
-                case LOG_OFFSET:
-                    LogOffsetCommittable offset =
-                            (LogOffsetCommittable) committable.wrappedCommittable();
-                    manifestCommittable.addLogOffset(offset.bucket(), offset.offset());
-                    break;
-            }
-        }
-    }
-
     private void calcNumBytesAndRecordsOut(List<ManifestCommittable> committables) {
         if (committerMetrics == null) {
             return;

diff --git a/...k/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java b/...k/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/StoreMultiCommitter.java
@@ -89,8 +89,7 @@ public WrappedManifestCommittable combine(
             long checkpointId, long watermark, List<MultiTableCommittable> committables) {
         WrappedManifestCommittable wrappedManifestCommittable =
                 new WrappedManifestCommittable(checkpointId, watermark);
-        combineFile(checkpointId, watermark, wrappedManifestCommittable, committables);
-        return wrappedManifestCommittable;
+        return combine(checkpointId, watermark, wrappedManifestCommittable, committables);
     }
 
     @Override
@@ -99,7 +98,25 @@ public WrappedManifestCommittable combine(
             long watermark,
             WrappedManifestCommittable wrappedManifestCommittable,
             List<MultiTableCommittable> committables) {
-        combineFile(checkpointId, watermark, wrappedManifestCommittable, committables);
+        for (MultiTableCommittable committable : committables) {
+            ManifestCommittable manifestCommittable =
+                    wrappedManifestCommittable.computeCommittableIfAbsent(
+                            Identifier.create(committable.getDatabase(), committable.getTable()),
+                            checkpointId,
+                            watermark);
+
+            switch (committable.kind()) {
+                case FILE:
+                    CommitMessage file = (CommitMessage) committable.wrappedCommittable();
+                    manifestCommittable.addFileCommittable(file);
+                    break;
+                case LOG_OFFSET:
+                    LogOffsetCommittable offset =
+                            (LogOffsetCommittable) committable.wrappedCommittable();
+                    manifestCommittable.addLogOffset(offset.bucket(), offset.offset());
+                    break;
+            }
+        }
         return wrappedManifestCommittable;
     }
 
@@ -168,32 +185,6 @@ public Map<Long, List<MultiTableCommittable>> groupByCheckpoint(
         return grouped;
     }
 
-    private void combineFile(
-            long checkpointId,
-            long watermark,
-            WrappedManifestCommittable wrappedManifestCommittable,
-            List<MultiTableCommittable> committables) {
-        for (MultiTableCommittable committable : committables) {
-            ManifestCommittable manifestCommittable =
-                    wrappedManifestCommittable.computeCommittableIfAbsent(
-                            Identifier.create(committable.getDatabase(), committable.getTable()),
-                            checkpointId,
-                            watermark);
-
-            switch (committable.kind()) {
-                case FILE:
-                    CommitMessage file = (CommitMessage) committable.wrappedCommittable();
-                    manifestCommittable.addFileCommittable(file);
-                    break;
-                case LOG_OFFSET:
-                    LogOffsetCommittable offset =
-                            (LogOffsetCommittable) committable.wrappedCommittable();
-                    manifestCommittable.addLogOffset(offset.bucket(), offset.offset());
-                    break;
-            }
-        }
-    }
-
     private StoreCommitter getStoreCommitter(Identifier tableId) {
         StoreCommitter committer = tableCommitters.get(tableId);