feat: Filter splits that do not require compaction

apache · Apr 21, 2024 · 52284e5 · 52284e5
1 parent 6c75f2c
commit 52284e5
Showing 1 changed file with 0 additions and 12 deletions.
diff --git a/...nk-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java b/...nk-common/src/main/java/org/apache/paimon/flink/sink/MultiTablesStoreCompactOperator.java
@@ -27,7 +27,6 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.ChannelComputer;
-import org.apache.paimon.table.sink.CommitMessageImpl;
 import org.apache.paimon.utils.Preconditions;
 
 import org.apache.flink.runtime.state.StateInitializationContext;
@@ -182,17 +181,6 @@ protected List<MultiTableCommittable> prepareCommit(boolean waitCompaction, long
             StoreSinkWrite write = entry.getValue();
             committables.addAll(
                     write.prepareCommit(waitCompaction, checkpointId).stream()
-                            .filter(
-                                    committable -> {
-                                        if (committable.kind() == Committable.Kind.FILE) {
-                                            CommitMessageImpl commitMessage =
-                                                    (CommitMessageImpl)
-                                                            committable.wrappedCommittable();
-                                            return !commitMessage.newFilesIncrement().isEmpty()
-                                                    && !commitMessage.compactIncrement().isEmpty();
-                                        }
-                                        return true;
-                                    })
                             .map(
                                     committable ->
                                             MultiTableCommittable.fromCommittable(key, committable))