[flink] Fix that compact actions haven't handled scan parallelism

yuzelin · Mar 22, 2024 · d38fc4c · d38fc4c
1 parent 6378848
commit d38fc4c
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 14 deletions.
diff --git a/...mon/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java b/...mon/src/main/java/org/apache/paimon/flink/compact/UnawareBucketCompactionTopoBuilder.java
@@ -98,8 +98,10 @@ private DataStreamSource<AppendOnlyCompactionTask> buildSource() {
                         specifiedPartitions != null
                                 ? PredicateBuilder.partitions(specifiedPartitions, table.rowType())
                                 : null);
-
-        return BucketUnawareCompactSource.buildSource(env, source, isContinuous, tableIdentifier);
+        Integer scanParallelism =
+                Options.fromMap(table.options()).get(FlinkConnectorOptions.SCAN_PARALLELISM);
+        return BucketUnawareCompactSource.buildSource(
+                env, source, isContinuous, scanParallelism, tableIdentifier);
     }
 
     private void sinkFromSource(DataStreamSource<AppendOnlyCompactionTask> input) {

diff --git a/...flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java b/...flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
@@ -117,15 +117,21 @@ public static DataStreamSource<AppendOnlyCompactionTask> buildSource(
             StreamExecutionEnvironment env,
             BucketUnawareCompactSource source,
             boolean streaming,
+            @Nullable Integer parallelism,
             String tableIdentifier) {
         final StreamSource<AppendOnlyCompactionTask, BucketUnawareCompactSource> sourceOperator =
                 new StreamSource<>(source);
-        return new DataStreamSource<>(
-                env,
-                new CompactionTaskTypeInfo(),
-                sourceOperator,
-                false,
-                COMPACTION_COORDINATOR_NAME + " : " + tableIdentifier,
-                streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED);
+        DataStreamSource<AppendOnlyCompactionTask> dataStream =
+                new DataStreamSource<>(
+                        env,
+                        new CompactionTaskTypeInfo(),
+                        sourceOperator,
+                        false,
+                        COMPACTION_COORDINATOR_NAME + " : " + tableIdentifier,
+                        streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED);
+        if (parallelism != null) {
+            dataStream.setParallelism(parallelism);
+        }
+        return dataStream;
     }
 }
diff --git a/...mon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java b/...mon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -118,11 +118,18 @@ public DataStreamSource<RowData> build() {
 
         BucketsTable bucketsTable = new BucketsTable(table, isContinuous);
         RowType produceType = bucketsTable.rowType();
-        return env.fromSource(
-                buildSource(bucketsTable),
-                WatermarkStrategy.noWatermarks(),
-                tableIdentifier + "-compact-source",
-                InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(produceType)));
+        DataStreamSource<RowData> dataStream =
+                env.fromSource(
+                        buildSource(bucketsTable),
+                        WatermarkStrategy.noWatermarks(),
+                        tableIdentifier + "-compact-source",
+                        InternalTypeInfo.of(LogicalTypeConversion.toLogicalType(produceType)));
+        Integer parallelism =
+                Options.fromMap(table.options()).get(FlinkConnectorOptions.SCAN_PARALLELISM);
+        if (parallelism != null) {
+            dataStream.setParallelism(parallelism);
+        }
+        return dataStream;
     }
 
     private Map<String, String> streamingCompactOptions() {