You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "leaves12138 (via GitHub)" <gi...@apache.org> on 2025/01/21 09:14:47 UTC
[PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
leaves12138 opened a new pull request, #4975:
URL: https://github.com/apache/paimon/pull/4975
<!-- Please specify the module before the PR name: [core] ... or [flink] ... -->
### Purpose
<!-- Linking this pull request to the issue -->
default.parallelism is not working because scan.infer-parallelism is default true
<!-- What is the purpose of the change -->
### Tests
<!-- List UT and IT cases to verify this change -->
### API and Format
<!-- Does this change affect API or storage format -->
### Documentation
<!-- Does this change introduce a new feature -->
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #4975:
URL: https://github.com/apache/paimon/pull/4975#discussion_r1948689224
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -149,9 +149,18 @@ protected Integer inferSourceParallelism(StreamExecutionEnvironment env) {
Boolean.parseBoolean(envConfig.toMap().get(FLINK_INFER_SCAN_PARALLELISM)));
}
Integer parallelism = options.get(FlinkConnectorOptions.SCAN_PARALLELISM);
- if (parallelism == null && options.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM)) {
+ if (parallelism == null
Review Comment:
Yes, as far as I am concerned, if we set default.parallelism in flink sql environment, we should not infer the parallelism, we should respect what the user writes. If we set INFER_SCAN_PARALLELISM default to true, than we must be careful, otherwise, users may be confused, cause they set the parallelism but seems not work.
We should respect environment and user as far as we can. This function is just auxiliary, we can't depend on it.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #4975:
URL: https://github.com/apache/paimon/pull/4975#discussion_r1945871181
##########
paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java:
##########
@@ -81,6 +81,8 @@ void testInferScanParallelism() throws Exception {
null);
PaimonDataStreamScanProvider runtimeProvider = runtimeProvider(tableSource);
StreamExecutionEnvironment sEnv1 = StreamExecutionEnvironment.createLocalEnvironment();
+ sEnv1.setParallelism(-1);
+ System.out.println(sEnv1.getParallelism());
Review Comment:
This print seems unnecessary.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -149,9 +149,18 @@ protected Integer inferSourceParallelism(StreamExecutionEnvironment env) {
Boolean.parseBoolean(envConfig.toMap().get(FLINK_INFER_SCAN_PARALLELISM)));
}
Integer parallelism = options.get(FlinkConnectorOptions.SCAN_PARALLELISM);
- if (parallelism == null && options.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM)) {
+ if (parallelism == null
Review Comment:
I'm trying to understand the parallelism inferring logic this PR wants to achieve, please see if this is correct.
1. If Paimon option SCAN_PARALLELISM is configured, then use this value.
2. Else if Flink configuration `parallelism.default` is configured, then use this value.
3. Else if Paimon option INFER_SCAN_PARALLELISM is set to true, and the parallelism of source can be inferred by Paimon (like when fixed bucket + unbounded stream), then Paimon would provide an inference result.
4. Else Paimon would not set the parallelism of the source operator, instead it would be Flink infra that decide the parallelism of this operator
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #4975:
URL: https://github.com/apache/paimon/pull/4975#discussion_r1948689224
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java:
##########
@@ -149,9 +149,18 @@ protected Integer inferSourceParallelism(StreamExecutionEnvironment env) {
Boolean.parseBoolean(envConfig.toMap().get(FLINK_INFER_SCAN_PARALLELISM)));
}
Integer parallelism = options.get(FlinkConnectorOptions.SCAN_PARALLELISM);
- if (parallelism == null && options.get(FlinkConnectorOptions.INFER_SCAN_PARALLELISM)) {
+ if (parallelism == null
Review Comment:
Yes, as far as I am concerned, if we set default.parallelism in flink sql environment, we should not infer the parallelism, we should respect what the user writes. If we set INFER_SCAN_PARALLELISM default to true, than we must be careful, otherwise, users may be confused, cause they set the parallelism but seems not works.
We should respect environment and user as far as we can. This function is just auxiliary, we can't depend on it.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on PR #4975:
URL: https://github.com/apache/paimon/pull/4975#issuecomment-2653197408
@leaves12138 Thanks for the update. According to our offline discussion, I learned some background information of this PR as follows.
In a Flink job where Paimon source is directly followed by a Paimon sink, the Paimon writer operator's parallelism cannot be dynamically inferred by Flink infra. The reason is because the writer operator sets its parallelism according to the upstream(source) operator, and the source operator's parallelism has explicitly decided.
So in order to allow Flink infra to change the parallelism of the writer operator, this PR made some change to the source operator's implementation. This way sounds indirect and incomplete, as Flink infra can still not set the parallelism in other situations. A better solution would be change
```java
writerDataStream.setParalleism(inputDataStream.getParallelism());
```
to
```java
writerDataStream.getTransformation().setParalleism(inputDataStream.getParallelism(), false);
```
Without regard to the background motivation, this PR itself looks good to me so I'm +1 on it. We may open a next PR in future to use the APIs mentioned above to better facilitate writer operators for Flink infra.
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [flink] Infer parallelism only in situation of parallelism is not set. [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #4975:
URL: https://github.com/apache/paimon/pull/4975
--
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-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org