wuchong commented on a change in pull request #8718:
[FLINK-12824][table-planner-blink] Set parallelism for stream SQL
URL: https://github.com/apache/flink/pull/8718#discussion_r293750904
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/plan/nodes/physical/batch/BatchExecSink.scala
##########
@@ -92,17 +93,15 @@ class BatchExecSink[T](
}
val sinkTransformation = dsSink.getTransformation
- if (sinkTransformation.getMaxParallelism > 0) {
-
sinkTransformation.setParallelism(sinkTransformation.getMaxParallelism)
- } else {
- val configSinkParallelism = NodeResourceConfig.getSinkParallelism(
- tableEnv.getConfig.getConf)
- if (configSinkParallelism > 0) {
- sinkTransformation.setParallelism(configSinkParallelism)
- } else if (boundedStream.getParallelism > 0) {
- sinkTransformation.setParallelism(boundedStream.getParallelism)
- }
+ val configSinkParallelism = NodeResourceConfig.getSinkParallelism(
+ tableEnv.getConfig.getConf)
+
+ val maxSinkParallelism = sinkTransformation.getMaxParallelism
+
+ if (maxSinkParallelism > 0 && configSinkParallelism <=
maxSinkParallelism) {
Review comment:
The sink parallelism only works when max parallelism is set? And If the
`configSinkParallelism` is -1, the sink parallelism will be set to -1 ?
Maybe the logic should be changed to this:
```java
// only set user's parallelism when user defines a sink parallelism
if (configSinkParallelism > 0) {
// set the parallelism when user's parallelism is not larger than max
parallelism or max parallelism is not set
if (maxSinkParallelism < 0 || configSinkParallelism <= maxSinkParallelism)
{
sinkTransformation.setParallelism(configSinkParallelism)
}
}
```
And we should add cases to cover 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.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
