shouweikun commented on a change in pull request #13789: URL: https://github.com/apache/flink/pull/13789#discussion_r511716192
########## File path: flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/common/CommonPhysicalSink.scala ########## @@ -99,11 +105,33 @@ class CommonPhysicalSink ( val operator = new SinkOperator(env.clean(sinkFunction), rowtimeFieldIndex, enforcer) + val inputParallelism = inputTransformation.getParallelism + val taskParallelism = env.getParallelism + val parallelism = if (runtimeProvider.isInstanceOf[ParallelismProvider]) runtimeProvider.asInstanceOf[ParallelismProvider].getParallelism.orElse(inputParallelism).intValue() + else inputParallelism + + if (implicitly[Ordering[Int]].lteq(parallelism, 0)) throw new RuntimeException(s"the configured sink parallelism: $parallelism should not be less than zero or equal to zero") + if (implicitly[Ordering[Int]].gt(parallelism, taskParallelism)) throw new RuntimeException(s"the configured sink parallelism: $parallelism is larger than the task max parallelism: $taskParallelism") + + val primaryKeys = TableSchemaUtils.getPrimaryKeyIndices(catalogTable.getSchema) + val containedRowKinds = changelogMode.getContainedKinds.toSet + val theFinalInputTransformation = if(inputParallelism == parallelism) inputTransformation //if the parallelism is not changed, do nothing + else (containedRowKinds, primaryKeys.toList) match { + // fixme : if rowKinds only contains delete, is there somethinng to do with? Currently do nothing. + case (_, _) if(containedRowKinds == Set(RowKind.DELETE)) => inputTransformation + case (_, _) if(containedRowKinds == Set(RowKind.INSERT)) => inputTransformation + // fixme: for retract mode (insert and delete contains only), is there somethinng to do with? Currently do nothing. + case (_, _) if(containedRowKinds == Set(RowKind.INSERT,RowKind.DELETE)) => inputTransformation + case (_, Nil) if(containedRowKinds.contains(RowKind.UPDATE_AFTER)) => throw new RuntimeException(s"ChangelogMode contains ${RowKind.UPDATE_AFTER}, but no primaryKeys were found") + case (_, _) if(containedRowKinds.contains(RowKind.UPDATE_AFTER)) => new DataStream[RowData](env,inputTransformation).keyBy(primaryKeys:_*).getTransformation + case _ => throw new RuntimeException(s"the changelogMode is: ${containedRowKinds.mkString(",")}, which is not supported") + } + Review comment: I enumerated all the changelog mode conditions by using Scala match pattern. So that we can esaily modify every condition in its own scope. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org