[GitHub] [flink] JingsongLi commented on a change in pull request #17939: [FLINK-20370][table] part2: introduce 'table.exec.sink.pk-shuffle' option to auto keyby on sink's pk if parallelism are not the same for insertOnly input

[GitBox]

JingsongLi commented on a change in pull request #17939:
URL: https://github.com/apache/flink/pull/17939#discussion_r764511517



##########
File path: 
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java
##########
@@ -232,26 +246,30 @@ private int deriveSinkParallelism(
      * messages.
      */
     private Transformation<RowData> applyKeyBy(
-            ChangelogMode changelogMode,
+            TableConfig config,
             Transformation<RowData> inputTransform,
             int[] primaryKeys,
             int sinkParallelism,
-            boolean upsertMaterialize) {
-        final int inputParallelism = inputTransform.getParallelism();
-        if ((inputParallelism == sinkParallelism || 
changelogMode.containsOnly(RowKind.INSERT))
-                && !upsertMaterialize) {
-            return inputTransform;
+            int inputParallelism,
+            boolean inputInsertOnly,
+            boolean needMaterialize) {
+        boolean sameParallelism = sinkParallelism == inputParallelism;
+        final ExecutionConfigOptions.SinkShuffleByPk sinkShuffleByPk =
+                
config.getConfiguration().get(ExecutionConfigOptions.TABLE_EXEC_SINK_SHUFFLE_BY_PK);
+        boolean sinkKeyBy = false;
+        switch (sinkShuffleByPk) {
+            case NONE:
+                break;
+            case AUTO:
+                sinkKeyBy = inputInsertOnly && !sameParallelism;
+                break;
+            case FORCE:
+                // single parallelism has no problem
+                sinkKeyBy = !(sinkParallelism == 1 && inputParallelism == 1);

Review comment:
       `sinkParallelism != 1 || inputParallelism != 1`?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org