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/01/15 03:34:49 UTC
(incubator-paimon) branch master updated: [flink] Add validation for parallelism of the SortOperator. (#2691)
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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 3d899fa20 [flink] Add validation for parallelism of the SortOperator. (#2691)
3d899fa20 is described below
commit 3d899fa2080ef48c38668b02cfa3bb0e1acc5b5d
Author: Aitozi <10...@qq.com>
AuthorDate: Mon Jan 15 11:34:45 2024 +0800
[flink] Add validation for parallelism of the SortOperator. (#2691)
---
.../main/java/org/apache/paimon/flink/sorter/SortOperator.java | 10 +++++++++-
.../main/java/org/apache/paimon/flink/sorter/SortUtils.java | 8 +++++++-
.../java/org/apache/paimon/flink/sorter/SortOperatorTest.java | 8 +++++---
3 files changed, 21 insertions(+), 5 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
index a35a9f161..f64cf044c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
@@ -41,6 +41,7 @@ public class SortOperator extends TableStreamOperator<InternalRow>
private final int pageSize;
private final int arity;
private final int spillSortMaxNumFiles;
+ private final int sinkParallelism;
private transient BinaryExternalSortBuffer buffer;
private transient IOManager ioManager;
@@ -50,19 +51,26 @@ public class SortOperator extends TableStreamOperator<InternalRow>
RowType rowType,
long maxMemory,
int pageSize,
- int spillSortMaxNumFiles) {
+ int spillSortMaxNumFiles,
+ int sinkParallelism) {
this.keyType = keyType;
this.rowType = rowType;
this.maxMemory = maxMemory;
this.pageSize = pageSize;
this.arity = rowType.getFieldCount();
this.spillSortMaxNumFiles = spillSortMaxNumFiles;
+ this.sinkParallelism = sinkParallelism;
}
@Override
public void open() throws Exception {
super.open();
initBuffer();
+ if (sinkParallelism != getRuntimeContext().getNumberOfParallelSubtasks()) {
+ throw new IllegalArgumentException(
+ "Please ensure that the runtime parallelism of the sink matches the initial configuration "
+ + "to avoid potential issues with skewed range partitioning.");
+ }
}
@VisibleForTesting
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
index 1cc5cf8ca..4187700a2 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
@@ -94,6 +94,11 @@ public class SortUtils {
sinkParallelismValue == null
? inputStream.getParallelism()
: Integer.parseInt(sinkParallelismValue);
+ if (sinkParallelism == -1) {
+ throw new UnsupportedOperationException(
+ "The adaptive batch scheduler is not supported. Please set the sink parallelism using the key: "
+ + FlinkConnectorOptions.SINK_PARALLELISM.key());
+ }
final int sampleSize = sinkParallelism * 1000;
final int rangeNum = sinkParallelism * 10;
@@ -155,7 +160,8 @@ public class SortUtils {
longRowType,
options.writeBufferSize(),
options.pageSize(),
- options.localSortMaxNumFileHandles()))
+ options.localSortMaxNumFileHandles(),
+ sinkParallelism))
.setParallelism(sinkParallelism)
// remove the key column from every row
.map(
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
index 422df0f65..c94a92dd7 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
@@ -61,7 +61,8 @@ public class SortOperatorTest {
rowType,
MemorySize.parse("10 mb").getBytes(),
(int) MemorySize.parse("16 kb").getBytes(),
- 128) {};
+ 128,
+ 1) {};
OneInputStreamOperatorTestHarness harness = createTestHarness(sortOperator);
harness.open();
@@ -88,7 +89,7 @@ public class SortOperatorTest {
}
@Test
- public void testCloseSortOprator() throws Exception {
+ public void testCloseSortOperator() throws Exception {
RowType keyRowType =
new RowType(
Collections.singletonList(
@@ -107,7 +108,8 @@ public class SortOperatorTest {
rowType,
MemorySize.parse("10 mb").getBytes(),
(int) MemorySize.parse("16 kb").getBytes(),
- 128) {};
+ 128,
+ 1) {};
OneInputStreamOperatorTestHarness harness = createTestHarness(sortOperator);
harness.open();
File[] files = harness.getEnvironment().getIOManager().getSpillingDirectories();