gustavodemorais commented on code in PR #26702:
URL: https://github.com/apache/flink/pull/26702#discussion_r2156773303
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -546,12 +557,140 @@ private boolean canCombine(
} else {
return false;
}
- return true;
+ return haveCommonJoinKey(origJoin, join);
} else {
return false;
}
}
+ /**
+ * Checks if original Join and child multi-join have common join keys.
+ *
+ * @param origJoin original Join
+ * @param otherJoin child MultiJoin
+ * @return true if original Join and child multi-join have at least one
common JoinKey
+ */
+ private boolean haveCommonJoinKey(Join origJoin, MultiJoin otherJoin) {
Review Comment:
```suggestion
private boolean hasCommonJoinKey(Join origJoin, MultiJoin otherJoin) {
```
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java:
##########
@@ -117,6 +117,15 @@ void testThreeWayInnerJoinRelPlan() {
+ "INNER JOIN Payments p ON u.user_id_0 =
p.user_id_2");
}
+ @Test
+ void testThreeWayInnerJoinNoCJKRelPlan() {
Review Comment:
```suggestion
void testThreeWayInnerJoinNoCommonJoinKeyRelPlan() {
```
I know the name is long but I think people won't know what CJK is. Same for
the other test
##########
flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/MultiJoinTest.java:
##########
@@ -154,6 +163,16 @@ void testFourWayComplexJoinRelPlan() {
+ "LEFT JOIN Shipments s ON p.user_id_2 =
s.user_id_3");
}
+ @Test
+ void testFourWayJoinNoCJKRelPlan() {
+ util.verifyRelPlan(
+ "SELECT u.user_id_0, u.name, o.order_id, p.payment_id,
s.location "
+ + "FROM Users u "
+ + "LEFT JOIN Orders o ON u.user_id_0 = o.user_id_1 "
+ + "INNER JOIN Payments p ON u.user_id_0 = p.user_id_2 "
+ + "LEFT JOIN Shipments s ON p.payment_id =
s.user_id_3");
+ }
+
Review Comment:
This still doesn't work:
@Test
void testFourWayJoinTransitiveCommonJoinKeyRelPlan() {
util.verifyExecPlan(
"SELECT u.user_id_0, u.name, o.order_id, p.payment_id,
s.location "
+ "FROM Users u "
+ "LEFT JOIN Orders o ON u.user_id_0 = o.user_id_1 "
+ "LEFT JOIN Payments p ON o.user_id_1 = p.user_id_2
"
+ "LEFT JOIN Shipments s ON p.user_id_2 =
s.user_id_3");
}
Although not 100% required for the first version.
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -546,12 +557,140 @@ private boolean canCombine(
} else {
return false;
}
- return true;
+ return haveCommonJoinKey(origJoin, join);
} else {
return false;
}
}
+ /**
+ * Checks if original Join and child multi-join have common join keys.
Review Comment:
```suggestion
* Checks if original join and child multi-join have common join keys to
decide if we can merge them into a single MultiJoin with one more input.
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -546,12 +557,140 @@ private boolean canCombine(
} else {
return false;
}
- return true;
+ return haveCommonJoinKey(origJoin, join);
} else {
return false;
}
}
+ /**
+ * Checks if original Join and child multi-join have common join keys.
+ *
+ * @param origJoin original Join
+ * @param otherJoin child MultiJoin
+ * @return true if original Join and child multi-join have at least one
common JoinKey
+ */
+ private boolean haveCommonJoinKey(Join origJoin, MultiJoin otherJoin) {
+ Set<String> origJoinKeys = getJoinKeys(origJoin);
+ Set<String> otherJoinKeys = getJoinKeys(otherJoin);
+
+ origJoinKeys.retainAll(otherJoinKeys);
+
+ return !origJoinKeys.isEmpty();
+ }
+
+ /**
+ * Gives set of join keys as strings like [table_name.field_name].
+ *
+ * @param join Join or MultiJoin node
+ * @return set of all the join keys (keys from join conditions)
+ */
+ public Set<String> getJoinKeys(RelNode join) {
+ Set<String> joinKeys = new HashSet<>();
+ List<RexCall> conditions = Collections.emptyList();
+ List<RelNode> inputs = join.getInputs();
+
+ if (join instanceof Join) {
+ conditions = collectConjunctions(((Join) join).getCondition());
+ } else if (join instanceof MultiJoin) {
+ conditions =
+ ((MultiJoin) join)
+ .getOuterJoinConditions().stream()
+ .flatMap(cond ->
collectConjunctions(cond).stream())
+ .collect(Collectors.toList());
+ }
+
+ RelMetadataQuery mq = join.getCluster().getMetadataQuery();
+
+ for (RexCall condition : conditions) {
+ for (RexNode operand : condition.getOperands()) {
+ if (operand instanceof RexInputRef) {
+ addJoinKeysByOperand((RexInputRef) operand, inputs, mq,
joinKeys);
+ }
+ }
+ }
+
+ return joinKeys;
+ }
+
+ /**
+ * Retrieves conjunctions from joinCondition.
+ *
+ * @param joinCondition join condition
+ * @return List of RexCalls representing conditions
+ */
+ private List<RexCall> collectConjunctions(RexNode joinCondition) {
Review Comment:
We need to filter out non equiJoin conditions here or in addJoinKeysByOperand
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/JoinToMultiJoinRule.java:
##########
@@ -546,12 +557,140 @@ private boolean canCombine(
} else {
return false;
}
- return true;
+ return haveCommonJoinKey(origJoin, join);
} else {
return false;
}
}
+ /**
+ * Checks if original Join and child multi-join have common join keys.
+ *
+ * @param origJoin original Join
+ * @param otherJoin child MultiJoin
+ * @return true if original Join and child multi-join have at least one
common JoinKey
+ */
+ private boolean haveCommonJoinKey(Join origJoin, MultiJoin otherJoin) {
+ Set<String> origJoinKeys = getJoinKeys(origJoin);
+ Set<String> otherJoinKeys = getJoinKeys(otherJoin);
+
+ origJoinKeys.retainAll(otherJoinKeys);
+
+ return !origJoinKeys.isEmpty();
+ }
+
+ /**
+ * Gives set of join keys as strings like [table_name.field_name].
Review Comment:
Returns a set of join keys as strings following this format
[table_name.field_name].
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
