[flink] Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink #3384

WencongLiu · 2024-05-22T15:45:26Z

Purpose

Linked issue: close #3385
Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink.

Tests

Add TableSortInfoTest for the TableSortInfo.
Add RangePartitionAndSortForUnawareBucketTableITCase.

API and Format

no

Documentation

Yes. The doc of FlinkConnectorOptions has been updated.

JingsongLi

Thanks @WencongLiu , left comments.

JingsongLi · 2024-05-23T03:57:38Z

...on-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/TableSortInfo.java

+            int sinkParallelism,
+            int localSampleSize,
+            int globalSampleSize) {
+        checkArgument(!sortColumns.isEmpty(), "Sort columns cannot be empty");


Don't need to check here, this is a private constructor.
You can check in the Builder.build.

Yes, placing all the check statements in the build is already sufficient. The changes have been made as requested.

JingsongLi · 2024-05-23T03:57:56Z

...on-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/TableSortInfo.java

+        public Builder setSinkParallelism(int sinkParallelism) {
+            checkArgument(
+                    sinkParallelism > 0,
+                    "The sink parallelism must be specified when sorting the table data. Please set it using the key: "


use checkArgument(boolean expression, String errorMessageTemplate, @CheckForNull Object... errorMessageArgs).

JingsongLi · 2024-05-23T04:01:03Z

...on-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/TableSortInfo.java

+        }
+
+        public Builder setRangeNumber(int rangeNumber) {
+            checkArgument(rangeNumber > 0, "Range number must be positive");


move all check to build.

JingsongLi · 2024-05-23T04:04:20Z

...on-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/shuffle/RangeShuffle.java

+
+        // 4. Remove the range index or both range index and key. (shuffle according range
+        // partition)
+        if (outputWithKey) {


You don't need to change here.
You can just output with key, and introduce a RemoveKeyOperator.

It should have no impact to performance if it is chaining to upstream operator.

The outputWithKey parameter is indeed unnecessary, and the changes have been made as requested.

JingsongLi · 2024-05-23T04:06:47Z

...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

+        if (boundedInput == null) {
+            boundedInput = !FlinkSink.isStreaming(input);
+        }
+        checkState(boundedInput, "The clustering should be executed under batch mode.");


Here we should discuss semantics, maybe we can just not sort in streaming mode?

+1 for not failing the job. We can enable the feature only when all the conditions are matched. Otherwise, if the by-columns are configured, we can print a warning log about it.

We probably should also mention the limitations (batch mode, table type, etc.) in the configuration description.

@JingsongLi
Good point, ignoring clustering in stream mode is a sensible design. This avoids the need for users to manually adjust the table configuration under streaming mode.

@xintongsong

I've removed the check and added releated warning log.

I've added the limitations of table type and batch mode in configuration description. I've also added a commit to introduce the clustering feature in paimon docs.

JingsongLi · 2024-05-23T04:07:58Z

...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

@@ -119,8 +133,86 @@ public FlinkSinkBuilder inputBounded(boolean bounded) {
        return this;
    }

+    /** Set the table sort info. */
+    public FlinkSinkBuilder setTableSortInfo(


clusteringIfPossible?

JingsongLi · 2024-05-23T04:08:47Z

...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

@@ -119,8 +133,86 @@ public FlinkSinkBuilder inputBounded(boolean bounded) {
        return this;
    }

+    /** Set the table sort info. */
+    public FlinkSinkBuilder setTableSortInfo(
+            String sortColumnsString,


respect option keys, sortColumnsString => clusterColumns

xintongsong · 2024-05-23T09:59:06Z

...on-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/TableSortInfo.java

+
+    private final List<String> sortColumns;
+
+    private final String sortStrategy;


I'd suggest to use OrderType for this field. This would allow us to check the field early, in Builder#setSortStrategy.

xintongsong · 2024-05-23T11:06:21Z

...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

+        if (boundedInput == null) {
+            boundedInput = !FlinkSink.isStreaming(input);
+        }
+        checkState(boundedInput, "The clustering should be executed under batch mode.");


+1 for not failing the job. We can enable the feature only when all the conditions are matched. Otherwise, if the by-columns are configured, we can print a warning log about it.

We probably should also mention the limitations (batch mode, table type, etc.) in the configuration description.

xintongsong · 2024-05-23T11:06:29Z

...n-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/FlinkSinkBuilder.java

@@ -119,8 +133,86 @@ public FlinkSinkBuilder inputBounded(boolean bounded) {
        return this;
    }

+    /** Set the table sort info. */
+    public FlinkSinkBuilder setTableSortInfo(


…d sort

xintongsong

@WencongLiu
Thanks for addressing my comments. LGTM.

JingsongLi

+1

JingsongLi · 2024-05-30T15:13:11Z

CC @leaves12138 to final check.

leaves12138

+1

WencongLiu force-pushed the dev_0522_1048 branch 2 times, most recently from a1d3c50 to 642b444 Compare May 23, 2024 02:35

JingsongLi requested changes May 23, 2024

View reviewed changes

xintongsong reviewed May 23, 2024

View reviewed changes

[flink] Introduce the TableSortInfo and refactor the TableSorter

ccc33b4

WencongLiu force-pushed the dev_0522_1048 branch 3 times, most recently from 582e2ab to 860bc0f Compare May 28, 2024 06:40

WencongLiu added 2 commits May 29, 2024 19:58

[flink] Introduce the range partition and sort for bucket unaware table

5695732

[flink] Add Clustering section in doc to introduce range partition an…

a8357d5

…d sort

WencongLiu force-pushed the dev_0522_1048 branch from 860bc0f to a8357d5 Compare May 30, 2024 05:36

xintongsong approved these changes May 30, 2024

View reviewed changes

JingsongLi approved these changes May 30, 2024

View reviewed changes

JingsongLi requested a review from leaves12138 May 30, 2024 15:13

leaves12138 approved these changes May 31, 2024

View reviewed changes

leaves12138 merged commit 885c2e0 into apache:master May 31, 2024
9 of 10 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[flink] Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink #3384

[flink] Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink #3384

WencongLiu commented May 22, 2024 •

edited

Loading

JingsongLi left a comment

JingsongLi May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

xintongsong May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

xintongsong May 23, 2024

WencongLiu May 28, 2024

JingsongLi May 23, 2024

WencongLiu May 28, 2024

xintongsong May 23, 2024

xintongsong May 23, 2024

xintongsong May 23, 2024

xintongsong left a comment

JingsongLi left a comment

JingsongLi commented May 30, 2024

leaves12138 left a comment


		private final List<String> sortColumns;

		private final String sortStrategy;

[flink] Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink #3384

[flink] Introduce Range Partition And Sort in Append Scalable Table Batch Writing for Flink #3384

Conversation

WencongLiu commented May 22, 2024 • edited Loading

Purpose

Tests

API and Format

Documentation

JingsongLi left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

xintongsong left a comment

Choose a reason for hiding this comment

JingsongLi left a comment

Choose a reason for hiding this comment

JingsongLi commented May 30, 2024

leaves12138 left a comment

Choose a reason for hiding this comment

WencongLiu commented May 22, 2024 •

edited

Loading