feat: Filter out invalid splits and commit messages to improve flink …

apache · Apr 21, 2024 · 6c75f2c · 6c75f2c
1 parent e27ceb4
commit 6c75f2c
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 0 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,6 +27,7 @@
 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;
@@ -181,6 +182,17 @@ 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))

diff --git a/...java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java b/...java/org/apache/paimon/flink/source/operator/MultiTablesBatchCompactorSourceFunction.java
@@ -21,8 +21,10 @@
 import org.apache.paimon.catalog.Catalog;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.flink.utils.JavaTypeInfo;
+import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.table.source.InnerStreamTableScanImpl;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.StreamTableScan;
 
@@ -81,8 +83,19 @@ public void run(SourceContext<Tuple2<Split, String>> ctx) throws Exception {
                     for (Map.Entry<Identifier, StreamTableScan> entry : scansMap.entrySet()) {
                         Identifier identifier = entry.getKey();
                         StreamTableScan scan = entry.getValue();
+                        int maxLevel = ((InnerStreamTableScanImpl) scan).options().numLevels() - 1;
                         splits.addAll(
                                 scan.plan().splits().stream()
+                                        .filter(
+                                                split -> {
+                                                    DataSplit dataSplit = (DataSplit) split;
+                                                    if (dataSplit.dataFiles().isEmpty()) {
+                                                        return false;
+                                                    }
+                                                    return dataSplit.dataFiles().stream()
+                                                            .map(DataFileMeta::level)
+                                                            .anyMatch(level -> level != maxLevel);
+                                                })
                                         .map(split -> new Tuple2<>(split, identifier.getFullName()))
                                         .collect(Collectors.toList()));
                     }