You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/08/21 05:53:28 UTC
(paimon) branch master updated: [core] Dynamic bucket mode cannot work with AQE (#4009)
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 a3eed25da [core] Dynamic bucket mode cannot work with AQE (#4009)
a3eed25da is described below
commit a3eed25da055432fa2547a1d6b631f0db5cbd0dd
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Wed Aug 21 13:53:22 2024 +0800
[core] Dynamic bucket mode cannot work with AQE (#4009)
---
.../org/apache/paimon/flink/sink/FlinkSink.java | 42 +++++++++++++++++++---
.../apache/paimon/flink/BatchFileStoreITCase.java | 17 +++++++--
2 files changed, 52 insertions(+), 7 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
index cfb8353ab..f810c464b 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSink.java
@@ -24,6 +24,7 @@ import org.apache.paimon.CoreOptions.TagCreationMode;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.options.MemorySize;
import org.apache.paimon.options.Options;
+import org.apache.paimon.table.BucketMode;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.utils.SerializableRunnable;
@@ -47,8 +48,10 @@ import org.apache.flink.table.api.config.ExecutionConfigOptions;
import javax.annotation.Nullable;
import java.io.Serializable;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedList;
+import java.util.List;
import java.util.Queue;
import java.util.Set;
@@ -222,8 +225,11 @@ public abstract class FlinkSink<T> implements Serializable {
commitUser))
.setParallelism(parallelism == null ? input.getParallelism() : parallelism);
- if (!isStreaming && table.coreOptions().writeManifestCache().getBytes() > 0) {
- assertBatchAdaptiveParallelism(env, written.getParallelism());
+ boolean writeMCacheEnabled = table.coreOptions().writeManifestCache().getBytes() > 0;
+ boolean hashDynamicMode = table.bucketMode() == BucketMode.HASH_DYNAMIC;
+ if (!isStreaming && (writeMCacheEnabled || hashDynamicMode)) {
+ assertBatchAdaptiveParallelism(
+ env, written.getParallelism(), writeMCacheEnabled, hashDynamicMode);
}
Options options = Options.fromMap(table.options());
@@ -316,11 +322,37 @@ public abstract class FlinkSink<T> implements Serializable {
public static void assertBatchAdaptiveParallelism(
StreamExecutionEnvironment env, int sinkParallelism) {
+ String msg =
+ "Paimon Sink does not support Flink's Adaptive Parallelism mode. "
+ + "Please manually turn it off or set Paimon `sink.parallelism` manually.";
+ assertBatchAdaptiveParallelism(env, sinkParallelism, msg);
+ }
+
+ public static void assertBatchAdaptiveParallelism(
+ StreamExecutionEnvironment env,
+ int sinkParallelism,
+ boolean writeMCacheEnabled,
+ boolean hashDynamicMode) {
+ List<String> messages = new ArrayList<>();
+ if (writeMCacheEnabled) {
+ messages.add("Write Manifest Cache");
+ }
+ if (hashDynamicMode) {
+ messages.add("Dynamic Bucket Mode");
+ }
+ String msg =
+ String.format(
+ "Paimon Sink with %s does not support Flink's Adaptive Parallelism mode. "
+ + "Please manually turn it off or set Paimon `sink.parallelism` manually.",
+ messages);
+ assertBatchAdaptiveParallelism(env, sinkParallelism, msg);
+ }
+
+ public static void assertBatchAdaptiveParallelism(
+ StreamExecutionEnvironment env, int sinkParallelism, String exceptionMsg) {
try {
checkArgument(
- sinkParallelism != -1 || !AdaptiveParallelism.isEnabled(env),
- "Paimon Sink does not support Flink's Adaptive Parallelism mode. "
- + "Please manually turn it off or set Paimon `sink.parallelism` manually.");
+ sinkParallelism != -1 || !AdaptiveParallelism.isEnabled(env), exceptionMsg);
} catch (NoClassDefFoundError ignored) {
// before 1.17, there is no adaptive parallelism
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
index f747a7fb6..eea8e3a3c 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BatchFileStoreITCase.java
@@ -50,12 +50,12 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
}
@Test
- public void testAdaptiveParallelism() {
+ public void testAQEWithWriteManifest() {
batchSql("ALTER TABLE T SET ('write-manifest-cache' = '1 mb')");
batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
assertThatThrownBy(() -> batchSql("INSERT INTO T SELECT a, b, c FROM T GROUP BY a,b,c"))
.hasMessageContaining(
- "Paimon Sink does not support Flink's Adaptive Parallelism mode.");
+ "Paimon Sink with [Write Manifest Cache] does not support Flink's Adaptive Parallelism mode.");
// work fine
batchSql(
@@ -66,6 +66,19 @@ public class BatchFileStoreITCase extends CatalogITCaseBase {
batchSql("INSERT INTO T SELECT a, b, c FROM T GROUP BY a,b,c");
}
+ @Test
+ public void testAQEWithDynamicBucket() {
+ batchSql("CREATE TABLE IF NOT EXISTS D_T (a INT PRIMARY KEY NOT ENFORCED, b INT, c INT)");
+ batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");
+ assertThatThrownBy(() -> batchSql("INSERT INTO D_T SELECT a, b, c FROM T GROUP BY a,b,c"))
+ .hasMessageContaining(
+ "Paimon Sink with [Dynamic Bucket Mode] does not support Flink's Adaptive Parallelism mode.");
+
+ // work fine
+ batchSql(
+ "INSERT INTO D_T /*+ OPTIONS('sink.parallelism'='1') */ SELECT a, b, c FROM T GROUP BY a,b,c");
+ }
+
@Test
public void testOverwriteEmpty() {
batchSql("INSERT INTO T VALUES (1, 11, 111), (2, 22, 222)");