[flink] Optimizing parallelism for fixed bucekt and non-partitioned t…

…able.
apache · Dec 4, 2024 · 632a57b · 632a57b
1 parent 812ef05
commit 632a57b
Showing 1 changed file with 9 additions and 0 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
@@ -265,6 +265,15 @@ protected DataStreamSink<?> buildDynamicBucketSink(
     }
 
     protected DataStreamSink<?> buildForFixedBucket(DataStream<InternalRow> input) {
+        int bucketNums = table.bucketSpec().getNumBuckets();
+        if (parallelism == null
+                && bucketNums < input.getParallelism()
+                && table.partitionKeys().isEmpty()) {
+            // For non-partitioned table, if the bucketNums is less than job parallelism.
+            LOG.warn(
+                    "For non-partitioned table, the writerOperator's parallelism will be set to bucketNums if the bucketNums is less than inputOperator's parallelism.");
+            parallelism = bucketNums;
+        }
         DataStream<InternalRow> partitioned =
                 partition(
                         input,