lincoln-lil commented on a change in pull request #17939: URL: https://github.com/apache/flink/pull/17939#discussion_r764519877
########## 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: I was thought this highlighted the single parallelism exception, but now feels like there's no difference between the two version. -- 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