lincoln-lil commented on a change in pull request #17939:
URL: https://github.com/apache/flink/pull/17939#discussion_r764518374
##########
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);
+ break;
}
- if (primaryKeys.length == 0) {
- throw new TableException(
- String.format(
- "The sink for table '%s' has a configured
parallelism of %s, while the input parallelism is %s. "
- + "Since the configured parallelism is
different from the input's parallelism and "
- + "the changelog mode is not insert-only,
a primary key is required but could not "
- + "be found.",
-
tableSinkSpec.getObjectIdentifier().asSummaryString(),
- sinkParallelism,
- inputParallelism));
+ if (!sinkKeyBy && !needMaterialize) {
+ return inputTransform;
}
final RowDataKeySelector selector =
Review comment:
Good catch! Should not add a empty key selector here (cause all the
data go into one parallel task) though the result is correct. I'll add a case
for this.
--
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
