This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 69390bdd2 [flink] fix sink parallelism in case of AQE (#4397)
69390bdd2 is described below
commit 69390bdd2b7d5f4dc716f0884886fa4d5756a305
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Tue Oct 29 14:56:26 2024 +0800
[flink] fix sink parallelism in case of AQE (#4397)
---
.../apache/paimon/flink/sink/AdaptiveParallelism.java | 18 ++++++++++++++++++
.../org/apache/paimon/flink/sink/FlinkSinkBuilder.java | 9 ++++-----
2 files changed, 22 insertions(+), 5 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AdaptiveParallelism.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AdaptiveParallelism.java
index 09b14779c..0fd7cf9f3 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AdaptiveParallelism.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/AdaptiveParallelism.java
@@ -18,7 +18,9 @@
package org.apache.paimon.flink.sink;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.BatchExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/** Get adaptive config from Flink. Only work for Flink 1.17+. */
@@ -27,4 +29,20 @@ public class AdaptiveParallelism {
public static boolean isEnabled(StreamExecutionEnvironment env) {
return
env.getConfiguration().get(BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_ENABLED);
}
+
+ /**
+ * Get default max parallelism of AdaptiveBatchScheduler of Flink. See
{@link
+ *
org.apache.flink.runtime.scheduler.adaptivebatch.AdaptiveBatchSchedulerFactory#getDefaultMaxParallelism(Configuration,
+ * ExecutionConfig)}.
+ */
+ public static int getDefaultMaxParallelism(
+ ReadableConfig configuration, ExecutionConfig executionConfig) {
+ return configuration
+
.getOptional(BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_MAX_PARALLELISM)
+ .orElse(
+ executionConfig.getParallelism() ==
ExecutionConfig.PARALLELISM_DEFAULT
+ ?
BatchExecutionOptions.ADAPTIVE_AUTO_PARALLELISM_MAX_PARALLELISM
+ .defaultValue()
+ : executionConfig.getParallelism());
+ }
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
index 1ff12ac93..dcccd0a1a 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java
@@ -51,7 +51,6 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import static org.apache.flink.configuration.CoreOptions.DEFAULT_PARALLELISM;
import static
org.apache.paimon.flink.FlinkConnectorOptions.CLUSTERING_SAMPLE_FACTOR;
import static
org.apache.paimon.flink.FlinkConnectorOptions.CLUSTERING_STRATEGY;
import static
org.apache.paimon.flink.FlinkConnectorOptions.MIN_CLUSTERING_SAMPLE_FACTOR;
@@ -318,11 +317,11 @@ public class FlinkSinkBuilder {
parallelismSource = "input parallelism";
parallelism = input.getParallelism();
} else {
- parallelismSource = DEFAULT_PARALLELISM.key();
+ parallelismSource = "AdaptiveBatchScheduler's default max
parallelism";
parallelism =
- input.getExecutionEnvironment()
- .getConfiguration()
- .get(DEFAULT_PARALLELISM);
+ AdaptiveParallelism.getDefaultMaxParallelism(
+
input.getExecutionEnvironment().getConfiguration(),
+ input.getExecutionConfig());
}
String msg =
String.format(
