snuyanzin commented on code in PR #29724:
URL: https://github.com/apache/flink/pull/29724#discussion_r4474084229
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/FlinkBushyJoinReorderRule.java:
##########
@@ -225,70 +230,128 @@ private static JoinPlan addOuterJoinToTop(
JoinPlan bestPlan, LoptMultiJoin multiJoin, RelBuilder relBuilder)
{
List<Integer> remainIndexes =
getRemainIndexes(multiJoin.getNumJoinFactors(),
bestPlan.factorIds);
- RelNode leftNode = bestPlan.relNode;
- Set<Integer> set = new LinkedHashSet<>(bestPlan.factorIds);
+ if (multiJoin.getMultiJoinRel().isFullOuterJoin()) {
+ // The join filters of a full outer join multiJoin are the
condition of the full outer
+ // join of its two factors.
+ assert remainIndexes.size() == 1;
+ List<RexNode> joinFilters = multiJoin.getJoinFilters();
+ return joinFactor(
+ bestPlan,
+ remainIndexes.get(0),
+ JoinRelType.FULL,
+ joinFilters,
+ joinFilters,
+ multiJoin,
+ relBuilder);
+ }
+
+ JoinPlan plan = bestPlan;
for (int index : remainIndexes) {
- RelNode rightNode = multiJoin.getJoinFactor(index);
- Optional<List<RexCall>> joinConditions =
- getJoinConditions(
- bestPlan.factorIds, Collections.singleton(index),
multiJoin, true);
-
- if (!joinConditions.isPresent()) {
- // If the join condition doesn't exist, it means the join type
is cross join, we
- // will add cross join to top in method addCrossJoinToTop()
separately.
- continue;
- } else {
- List<RexCall> conditions = joinConditions.get();
- List<RexCall> newCondition =
- convertToNewCondition(
- new ArrayList<>(set),
- Collections.singletonList(index),
- conditions,
- multiJoin);
-
- // For full outer join, we return the full join type. However,
for left outer join
- // or right outer join, we will build a new join with left
join type. For example,
- // for case (A RJ B) IJ C, we will return (B IJ C) LJ A.
- JoinRelType joinType = JoinRelType.LEFT;
- if (multiJoin.getMultiJoinRel().isFullOuterJoin()) {
- assert remainIndexes.size() == 1;
- joinType = JoinRelType.FULL;
- }
- relBuilder.clear();
- leftNode =
- relBuilder
- .push(leftNode)
- .push(rightNode)
- .join(joinType, newCondition)
- .build();
+ // The inner join factors that the inner join reorder couldn't
join, e.g. those of a
+ // cross join, are added in method addCrossJoinToTop(), unless the
condition of an
+ // outer join references them.
+ if (multiJoin.isNullGenerating(index) &&
!plan.factorIds.contains(index)) {
+ plan = addOuterJoinFactor(plan, index, multiJoin, relBuilder);
}
- set.add(index);
}
- return new JoinPlan(set, leftNode);
+ return plan;
}
private static JoinPlan addCrossJoinToTop(
JoinPlan bestPlan, LoptMultiJoin multiJoin, RelBuilder relBuilder)
{
- RexBuilder rexBuilder =
multiJoin.getMultiJoinRel().getCluster().getRexBuilder();
List<Integer> remainIndexes =
getRemainIndexes(multiJoin.getNumJoinFactors(),
bestPlan.factorIds);
- RelNode leftNode = bestPlan.relNode;
- Set<Integer> set = new LinkedHashSet<>(bestPlan.factorIds);
+ JoinPlan plan = bestPlan;
for (int index : remainIndexes) {
- relBuilder.clear();
- set.add(index);
- RelNode rightNode = multiJoin.getJoinFactor(index);
- leftNode =
- relBuilder
- .push(leftNode)
- .push(rightNode)
- .join(
-
multiJoin.getMultiJoinRel().getJoinTypes().get(index),
- rexBuilder.makeLiteral(true))
- .build();
+ if (multiJoin.isNullGenerating(index)) {
+ // An outer join factor whose join condition is true.
+ plan = addOuterJoinFactor(plan, index, multiJoin, relBuilder);
+ } else {
+ plan = addInnerJoinFactor(plan, index, multiJoin, relBuilder);
+ }
+ }
+ return plan;
+ }
+
+ /**
+ * Joins a null generating factor to the plan with its whole outer join
condition. Joins the
+ * factors that the condition references first, e.g. the null generating
factor of an outer join
+ * on the right side of a right outer join.
+ */
+ private static JoinPlan addOuterJoinFactor(
+ JoinPlan plan, int index, LoptMultiJoin multiJoin, RelBuilder
relBuilder) {
+ for (int factor : multiJoin.getOuterJoinFactors(index)) {
+ if (!plan.factorIds.contains(factor)) {
+ plan =
+ multiJoin.isNullGenerating(factor)
+ ? addOuterJoinFactor(plan, factor, multiJoin,
relBuilder)
+ : addInnerJoinFactor(plan, factor, multiJoin,
relBuilder);
+ }
+ }
+ // For left outer join or right outer join, we will build a new join
with left join type.
+ // For example, for case (A RJ B) IJ C, we will return (B IJ C) LJ A.
+ return joinFactor(
+ plan,
+ index,
+ JoinRelType.LEFT,
+ RelOptUtil.conjunctions(multiJoin.getOuterJoinCond(index)),
+ Collections.emptyList(),
+ multiJoin,
+ relBuilder);
+ }
+
+ /**
+ * Joins an inner join factor to the plan with the join filters that
reference the factor and
+ * the factors of the plan only, e.g. a join filter that references three
factors once two of
+ * them are joined.
+ */
+ private static JoinPlan addInnerJoinFactor(
+ JoinPlan plan, int index, LoptMultiJoin multiJoin, RelBuilder
relBuilder) {
+ List<RexCall> joinFilters =
+ getJoinConditions(plan.factorIds,
Collections.singleton(index), multiJoin)
+ .orElse(Collections.emptyList());
+ return joinFactor(
+ plan, index, JoinRelType.INNER, joinFilters, joinFilters,
multiJoin, relBuilder);
+ }
+
+ /**
+ * Joins the factor to the plan.
+ *
+ * @param conditions the join conditions, referencing the fields of the
multiJoin
+ * @param joinFilters the join filters of the multiJoin among the
conditions
+ */
+ private static JoinPlan joinFactor(
+ JoinPlan plan,
+ int index,
+ JoinRelType joinType,
+ List<? extends RexNode> conditions,
+ List<? extends RexNode> joinFilters,
+ LoptMultiJoin multiJoin,
+ RelBuilder relBuilder) {
+ JoinConditionShuttle shuttle =
+ new JoinConditionShuttle(
+ multiJoin,
+ new ArrayList<>(plan.factorIds),
+ Collections.singletonList(index));
+ List<RexNode> newConditions = new ArrayList<>();
+ for (RexNode condition : conditions) {
+ newConditions.add(condition.accept(shuttle));
Review Comment:
isn't it duplicating `convertToNewCondition`?
--
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]
