[flink] Disable compaction topology when streaming with bounded input…

… stream (apache#2463)
tongcheng-elong · Dec 7, 2023 · 9c2fb10 · 9c2fb10
1 parent c7e8432
commit 9c2fb10
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 3 deletions.
diff --git a/...link/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java b/...link/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -44,6 +44,7 @@ public class FlinkSinkBuilder {
     @Nullable private Map<String, String> overwritePartition;
     @Nullable private LogSinkFunction logSinkFunction;
     @Nullable private Integer parallelism;
+    private boolean boundedInput = false;
     private boolean compactSink = false;
 
     public FlinkSinkBuilder(FileStoreTable table) {
@@ -79,6 +80,11 @@ public FlinkSinkBuilder withParallelism(@Nullable Integer parallelism) {
         return this;
     }
 
+    public FlinkSinkBuilder withBoundedInputStream(boolean bounded) {
+        this.boundedInput = bounded;
+        return this;
+    }
+
     public FlinkSinkBuilder forCompact(boolean compactSink) {
         this.compactSink = compactSink;
         return this;
@@ -142,7 +148,8 @@ private DataStreamSink<?> buildUnawareBucketSink(DataStream<InternalRow> input)
                         (AppendOnlyFileStoreTable) table,
                         overwritePartition,
                         logSinkFunction,
-                        parallelism)
+                        parallelism,
+                        boundedInput)
                 .sinkFrom(input);
     }
 }
diff --git a/...nk/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java b/...nk/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkTableSinkBase.java
@@ -133,6 +133,7 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
                                 .withLogSinkFunction(logSinkFunction)
                                 .withOverwritePartition(overwrite ? staticPartitions : null)
                                 .withParallelism(conf.get(FlinkConnectorOptions.SINK_PARALLELISM))
+                                .withBoundedInputStream(context.isBounded())
                                 .build());
     }
 

diff --git a/...aimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketWriteSink.java b/...aimon-flink-common/src/main/java/org/apache/paimon/flink/sink/UnawareBucketWriteSink.java
@@ -40,16 +40,19 @@ public class UnawareBucketWriteSink extends FileStoreSink {
     private final boolean enableCompaction;
     private final AppendOnlyFileStoreTable table;
     private final Integer parallelism;
+    private final boolean boundedInput;
 
     public UnawareBucketWriteSink(
             AppendOnlyFileStoreTable table,
             Map<String, String> overwritePartitions,
             LogSinkFunction logSinkFunction,
-            Integer parallelism) {
+            Integer parallelism,
+            boolean boundedInput) {
         super(table, overwritePartitions, logSinkFunction);
         this.table = table;
         this.enableCompaction = !table.coreOptions().writeOnly();
         this.parallelism = parallelism;
+        this.boundedInput = boundedInput;
     }
 
     @Override
@@ -63,7 +66,8 @@ public DataStreamSink<?> sinkFrom(DataStream<InternalRow> input, String initialC
                                 .get(ExecutionOptions.RUNTIME_MODE)
                         == RuntimeExecutionMode.STREAMING;
         // if enable compaction, we need to add compaction topology to this job
-        if (enableCompaction && isStreamingMode) {
+        if (enableCompaction && isStreamingMode && !boundedInput) {
+            // if streaming mode with bounded input, we disable compaction topology
             UnawareBucketCompactionTopoBuilder builder =
                     new UnawareBucketCompactionTopoBuilder(
                             input.getExecutionEnvironment(), table.name(), table);