You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "snuyanzin (via GitHub)" <gi...@apache.org> on 2025/06/22 22:15:17 UTC
[PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
snuyanzin opened a new pull request, #26707:
URL: https://github.com/apache/flink/pull/26707
## What is the purpose of the change
The problem of current Multiway join (`JoinToMultiJoinRule.java`) implementation is
this line https://github.com/apache/flink/blob/24ce37f70ab2904d5c9196dd43f5abe649f1c556/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java#L712-L713
1. it leads to extra planner call
2. it leads to the fact that at some point in `RelSubset` it results to `HepRelVertex` in `RelSubset#add` call which beame forbidden starting Calcite 1.35 https://github.com/apache/calcite/commit/7fc3e1b007fd9fd921c7929f27ac6cf3e75a15fd#diff-23f79975c4d8d77b7a09299831fe47c69c4f65fe8639bc4ceb956321a33da634R358.
To reproduce the failure use rebased version of PR https://github.com/apache/flink/pull/26547
## Verifying this change
existing tests and https://github.com/apache/flink/pull/26547 which will be eventually merged separately
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): ( no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: ( no)
- The serializers: ( no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: ( no)
- The S3 file system connector: ( no)
## Documentation
- Does this pull request introduce a new feature? ( no)
- If yes, how is the feature documented? (not applicable)
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26707:
URL: https://github.com/apache/flink/pull/26707#discussion_r2160508874
##########
flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/IntervalJoinUtil.scala:
##########
@@ -37,7 +38,7 @@ import org.apache.calcite.sql.validate.SqlValidatorUtil
import java.util
-import scala.collection.JavaConversions._
Review Comment:
it is deprecated, so replaced with a recommended approach
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26707:
URL: https://github.com/apache/flink/pull/26707#discussion_r2160508587
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -345,41 +327,25 @@ private List<RelNode> combineInputs(
* The join type is also set.
*
* @param joinRel join rel
- * @param combinedInputs the combined inputs to the join
* @param left left child of the joinrel
- * @param right right child of the joinrel
* @param joinSpecs the list where the join types and conditions will be copied
*/
private void combineJoinInfo(
- Join joinRel,
- List<RelNode> combinedInputs,
- RelNode left,
- RelNode right,
- List<Pair<JoinRelType, RexNode>> joinSpecs,
- List<Boolean> inputNullGenFieldList) {
Review Comment:
no idea why do we need so many args if finally half of them are not used
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26707:
URL: https://github.com/apache/flink/pull/26707#discussion_r2160509150
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -289,53 +283,41 @@ private void buildNullGenFieldList(RelNode rel, List<Boolean> isNullGenFieldList
* @param left left input into join
* @param right right input into join
* @param projFieldsList returns a list of the new combined projection fields
- * @param joinFieldRefCountsList returns a list of the new combined join field reference counts
* @return combined left and right inputs in an array
*/
private List<RelNode> combineInputs(
Join join,
RelNode left,
RelNode right,
List<ImmutableBitSet> projFieldsList,
- List<int[]> joinFieldRefCountsList,
- List<Boolean> inputNullGenFieldList) {
+ List<int[]> joinFieldRefCountsList) {
final List<RelNode> newInputs = new ArrayList<>();
- // Leave the null generating sides of an outer join intact; don't pull up those children
- // inputs into the array we're constructing.
- JoinInfo joinInfo = join.analyzeCondition();
- ImmutableIntList leftKeys = joinInfo.leftKeys;
- ImmutableIntList rightKeys = joinInfo.rightKeys;
- if (canCombine(left, join)) {
- final MultiJoin leftMultiJoin = (MultiJoin) left;
- for (int i = 0; i < leftMultiJoin.getInputs().size(); i++) {
- newInputs.add(leftMultiJoin.getInput(i));
- projFieldsList.add(leftMultiJoin.getProjFields().get(i));
- joinFieldRefCountsList.add(
- leftMultiJoin.getJoinFieldRefCountsMap().get(i).toIntArray());
- }
+ combineIfCan(join, left, newInputs, projFieldsList, joinFieldRefCountsList);
+ combineIfCan(join, right, newInputs, projFieldsList, joinFieldRefCountsList);
- } else {
- newInputs.add(left);
- projFieldsList.add(null);
- joinFieldRefCountsList.add(new int[left.getRowType().getFieldCount()]);
- }
Review Comment:
logic is identical, so extracted into `combineIfCan`
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #26707:
URL: https://github.com/apache/flink/pull/26707#issuecomment-2994486993
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "4fdeb06ee13d43f4c0e5067bb368b8b1a112d777",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "4fdeb06ee13d43f4c0e5067bb368b8b1a112d777",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 4fdeb06ee13d43f4c0e5067bb368b8b1a112d777 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "snuyanzin (via GitHub)" <gi...@apache.org>.
snuyanzin commented on code in PR #26707:
URL: https://github.com/apache/flink/pull/26707#discussion_r2160508680
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -345,41 +327,25 @@ private List<RelNode> combineInputs(
* The join type is also set.
*
* @param joinRel join rel
- * @param combinedInputs the combined inputs to the join
* @param left left child of the joinrel
- * @param right right child of the joinrel
* @param joinSpecs the list where the join types and conditions will be copied
*/
private void combineJoinInfo(
- Join joinRel,
- List<RelNode> combinedInputs,
- RelNode left,
- RelNode right,
- List<Pair<JoinRelType, RexNode>> joinSpecs,
- List<Boolean> inputNullGenFieldList) {
+ Join joinRel, RelNode left, List<Pair<JoinRelType, RexNode>> joinSpecs) {
JoinRelType joinType = joinRel.getJoinType();
- JoinInfo joinInfo = joinRel.analyzeCondition();
- ImmutableIntList leftKeys = joinInfo.leftKeys;
final RexBuilder rexBuilder = joinRel.getCluster().getRexBuilder();
boolean leftCombined = canCombine(left, joinRel);
switch (joinType) {
case LEFT:
- if (leftCombined) {
- copyJoinInfo((MultiJoin) left, joinSpecs);
- } else {
- joinSpecs.add(Pair.of(JoinRelType.INNER, rexBuilder.makeLiteral(true)));
- }
- joinSpecs.add(Pair.of(joinType, joinRel.getCondition()));
- break;
case INNER:
if (leftCombined) {
copyJoinInfo((MultiJoin) left, joinSpecs);
} else {
joinSpecs.add(Pair.of(JoinRelType.INNER, rexBuilder.makeLiteral(true)));
}
Review Comment:
may be I missed something however they are identical
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-37980][table] Adapt Multi-way join to higher Calcite versions [flink]
Posted by "gustavodemorais (via GitHub)" <gi...@apache.org>.
gustavodemorais commented on code in PR #26707:
URL: https://github.com/apache/flink/pull/26707#discussion_r2161378566
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -345,41 +327,25 @@ private List<RelNode> combineInputs(
* The join type is also set.
*
* @param joinRel join rel
- * @param combinedInputs the combined inputs to the join
* @param left left child of the joinrel
- * @param right right child of the joinrel
* @param joinSpecs the list where the join types and conditions will be copied
*/
private void combineJoinInfo(
- Join joinRel,
- List<RelNode> combinedInputs,
- RelNode left,
- RelNode right,
- List<Pair<JoinRelType, RexNode>> joinSpecs,
- List<Boolean> inputNullGenFieldList) {
+ Join joinRel, RelNode left, List<Pair<JoinRelType, RexNode>> joinSpecs) {
JoinRelType joinType = joinRel.getJoinType();
- JoinInfo joinInfo = joinRel.analyzeCondition();
- ImmutableIntList leftKeys = joinInfo.leftKeys;
final RexBuilder rexBuilder = joinRel.getCluster().getRexBuilder();
boolean leftCombined = canCombine(left, joinRel);
switch (joinType) {
case LEFT:
- if (leftCombined) {
- copyJoinInfo((MultiJoin) left, joinSpecs);
- } else {
- joinSpecs.add(Pair.of(JoinRelType.INNER, rexBuilder.makeLiteral(true)));
- }
- joinSpecs.add(Pair.of(joinType, joinRel.getCondition()));
- break;
case INNER:
if (leftCombined) {
copyJoinInfo((MultiJoin) left, joinSpecs);
} else {
joinSpecs.add(Pair.of(JoinRelType.INNER, rexBuilder.makeLiteral(true)));
}
Review Comment:
They were
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -709,9 +673,7 @@ private boolean isIntervalJoin(Join join) {
return true;
}
- FlinkLogicalJoin flinkLogicalJoin =
- (FlinkLogicalJoin) FlinkLogicalJoin.CONVERTER().convert(join);
- return IntervalJoinUtil.satisfyIntervalJoin(flinkLogicalJoin);
+ return IntervalJoinUtil.satisfyIntervalJoin(join);
Review Comment:
👍
--
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@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org