github-actions[bot] commented on code in PR #67067: URL: https://github.com/apache/doris/pull/67067#discussion_r4025714427
########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/joinorder/JoinReorderRule.java: ########## @@ -0,0 +1,177 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.rules.rewrite.joinorder; + +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.logical.LogicalJoin; +import org.apache.doris.nereids.trees.plans.logical.LogicalProject; +import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter; + +import com.google.common.collect.Lists; + +import java.util.List; +import java.util.stream.Stream; + +/**JoinReorderRule*/ +public class JoinReorderRule extends DefaultPlanRewriter<Void> { + public static final JoinReorderRule INSTANCE = new JoinReorderRule(); + public static final int MAX_ATOM_NUM_FOR_GREEDY = 16; + + public Plan rewrite(Plan plan, Void context) { + return plan.accept(this, context); + } + + @Override + public Plan visitLogicalJoin( + LogicalJoin<? extends Plan, ? extends Plan> join, + Void context) { + if (!isReorderable(join)) { + // The current join is a boundary, but its children may contain independent join clusters. + return DefaultPlanRewriter.visitChildren(this, join, context); + } + + // The current join is the root of a cluster. Reorder the current cluster and recursively + // process independent clusters below its boundaries. + return reorderCluster(join, context); + } + + private Plan reorderCluster( + LogicalJoin<? extends Plan, ? extends Plan> root, + Void context) { + JoinCluster cluster = new JoinCluster(root.getOutput()); + Plan fallback = rewriteAndCollectCluster(root, cluster, context); + + // Use the fallback when the best candidate increases the number of cross joins. + Plan reordered = reorder(cluster); + return reordered == null ? fallback : reordered; + } + + private int countCrossJoinsInCluster(Plan plan) { + if (plan instanceof LogicalJoin + && isReorderable((LogicalJoin<?, ?>) plan)) { + LogicalJoin<?, ?> join = (LogicalJoin<?, ?>) plan; + int currentCrossJoinCount = join.getJoinType().isCrossJoin() ? 1 : 0; + return currentCrossJoinCount + + countCrossJoinsInCluster(join.left()) + + countCrossJoinsInCluster(join.right()); + } + if (plan instanceof LogicalProject + && isTransparentProject((LogicalProject<?>) plan)) { + return countCrossJoinsInCluster(plan.child(0)); + } + return 0; + } + + /* + * Traverses once to collect the current reorderable join cluster and rewrite independent + * clusters below its boundaries. + * Collects inputs, predicates, and the cross-join count into the cluster parameter, and + * returns the fallback plan. + */ + private Plan rewriteAndCollectCluster(Plan plan, JoinCluster cluster, Void context) { + if (plan instanceof LogicalJoin + && isReorderable((LogicalJoin<?, ?>) plan)) { + LogicalJoin<?, ?> join = (LogicalJoin<?, ?>) plan; + cluster.addPredicates(join.getHashJoinConjuncts()); + cluster.addPredicates(join.getOtherJoinConjuncts()); + if (join.getJoinType().isCrossJoin()) { + cluster.crossJoinCount++; + } + Plan left = rewriteAndCollectCluster(join.left(), cluster, context); + Plan right = rewriteAndCollectCluster(join.right(), cluster, context); + return left == join.left() && right == join.right() + ? join + : join.withChildren(left, right); + } + if (plan instanceof LogicalProject + && isTransparentProject((LogicalProject<?>) plan)) { + LogicalProject<?> project = (LogicalProject<?>) plan; + + /* + * The project contains only existing slots and does not replace any ExprId, so predicates + * from upper joins do not need to be rewritten and flattening can continue through it. + * The project at the cluster root restores column pruning and the original output order. + */ + Plan child = rewriteAndCollectCluster(project.child(), cluster, context); + return child == project.child() ? project : project.withChildren(child); + } + + // The plan is a boundary of the current cluster and may contain independent clusters. + Plan rewrittenInput = plan.accept(this, context); + cluster.addInput(rewrittenInput); + return rewrittenInput; + } + + private boolean isTransparentProject(LogicalProject<?> project) { + return !project.isDistinct() && project.isAllSlots(); + } + + private Plan reorder(JoinCluster joinCluster) { + if (joinCluster.inputs.size() > MAX_ATOM_NUM_FOR_GREEDY) { + return null; + } + JoinReorderGreedy reorderGreedy = new JoinReorderGreedy(); + if (!reorderGreedy.reorder(joinCluster.inputs, joinCluster.predicates)) { + return null; + } + List<Plan> plans = reorderGreedy.getResult(); + if (plans.isEmpty()) { + return null; + } + Plan bestPlan = plans.get(0); + Plan candidate = joinCluster.originalOutput.equals(bestPlan.getOutput()) + ? bestPlan + : new LogicalProject<>((List) joinCluster.originalOutput, bestPlan); + return countCrossJoinsInCluster(candidate) <= joinCluster.crossJoinCount + ? candidate + : null; + } + + private static class JoinCluster { + private final List<Plan> inputs = Lists.newArrayList(); + private final List<Expression> predicates = Lists.newArrayList(); + private final List<Slot> originalOutput; + private int crossJoinCount; + + JoinCluster(List<Slot> originalOutput) { + this.originalOutput = originalOutput; + } + + private void addInput(Plan input) { + inputs.add(input); + } + + private void addPredicates(List<Expression> predicates) { + this.predicates.addAll(predicates); + } + } + + private boolean isReorderable(LogicalJoin<?, ?> join) { + return join.getJoinType().isInnerOrCrossJoin() + && !join.isMarkJoin() + && !join.getJoinType().isAsofJoin() + && !join.isLeadingJoin() + && !join.hasDistributeHint() + && Stream.concat( + join.getHashJoinConjuncts().stream(), + join.getOtherJoinConjuncts().stream()) + .noneMatch(Expression::containsVolatileExpression); Review Comment: [P1] Fence non-movable predicates from join reassociation This guard only rejects volatile conjuncts, so deterministic `NoneMovableFunction`s such as `assert_true` remain reorderable. For `(A JOIN B ON A.id=B.id AND assert_true(A.v+B.v>0,'bad')) JOIN C ON B.cid=C.id`, a failing A/B row whose `cid` is absent from C throws in the original lower join. If the greedy order builds B-C first, that row is removed before the A/B predicate is attached at the top, and the required error is silently suppressed. Please treat conjuncts containing `NoneMovableFunction` as a cluster boundary (or preserve them at their original join) and add this pruning case to the tests. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggThroughJoinOnPkFk.java: ########## @@ -138,17 +140,29 @@ public List<Rule> buildRules() { // select primary_table_pk, primary_table_other from primary_table join foreign_table on pk = fk // group by pk, primary_table_other_cols; private LogicalAggregate<?> eliminatePrimaryOutput(LogicalAggregate<?> agg, Plan child, - Plan primary, Plan foreign) { + PrimaryForeignInfo primaryForeignInfo) { + Set<Slot> groupBySlots = agg.getGroupByExpressions().stream() Review Comment: [P1] Fence non-movable aggregates on the indirect-FD path This new indirect-FD admission can move an error-producing aggregate below a nullable-FK inner join. With `F.row_id` as a key, nullable `F.parent_id` referencing `P.id`, and `GROUP BY F.row_id, COUNT(assert_true(F.x>0,'bad'))`, the closure proves `row_id -> parent_id` and the rewrite injects `parent_id` into the lower aggregate. A failing row whose FK is NULL is discarded before evaluation in the original plan, but the pushed aggregate evaluates it first and throws. Please reject output aggregates containing `NoneMovableFunction` here and add a nullable-FK/unmatched-row regression; the new test's non-null FK and `count(slot)` do not cover this. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java: ########## @@ -100,14 +100,11 @@ public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> { @Override public Plan visit(Plan plan, PushDownAggContext context) { - return plan; + return genAggregate(plan, context); Review Comment: [P1] Do not push non-movable aggregate arguments below the join The generic fallback makes previously unsupported nodes such as `Limit` aggregate-placement boundaries, but the pushdown admission checks reject only volatile expressions. A deterministic `NoneMovableFunction` is still unsafe: for `GROUP BY B.g, COUNT(assert_true(A.x>0,'bad'))` over `Join(Limit(A), B)`, an unmatched failing A row is discarded before the top aggregate in the original plan; the new partial aggregate above `Limit` evaluates it before the join and throws. Please fence aggregate functions containing `NoneMovableFunction` (or keep them above the original evaluation boundary) and add an unmatched-row regression. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/ReorderJoinBeforeEagerAgg.java: ########## @@ -0,0 +1,60 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.nereids.rules.rewrite.eageraggregation; + +import org.apache.doris.nereids.jobs.JobContext; +import org.apache.doris.nereids.rules.rewrite.ColumnPruning; +import org.apache.doris.nereids.rules.rewrite.joinorder.JoinReorderRule; +import org.apache.doris.nereids.stats.StatsCalculator; +import org.apache.doris.nereids.trees.plans.Plan; +import org.apache.doris.nereids.trees.plans.algebra.CatalogRelation; +import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter; +import org.apache.doris.qe.ConnectContext; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; + +/** Reorder joins before eager aggregation. */ +public class ReorderJoinBeforeEagerAgg implements CustomRewriter { + private static final Logger LOG = LoggerFactory.getLogger(ReorderJoinBeforeEagerAgg.class); + + @Override + public Plan rewriteRoot(Plan plan, JobContext jobContext) { + List<CatalogRelation> scans = plan.collectToList(CatalogRelation.class::isInstance); + StatsCalculator.disableJoinReorderIfStatsInvalid(scans, jobContext.getCascadesContext()); + ConnectContext connectContext = jobContext.getCascadesContext().getConnectContext(); + if (connectContext.getSessionVariable().isDisableJoinReorder() + || jobContext.getCascadesContext().isLeadingDisableJoinReorder() + || !connectContext.getSessionVariable().enableJoinReorderBeforeEagerAgg) { + return plan; + } + long startNanos = System.nanoTime(); + Plan reorderedPlan = JoinReorderRule.INSTANCE.rewrite(plan, null); + if (LOG.isDebugEnabled()) { + double elapsedMs = (System.nanoTime() - startNanos) / 1_000_000.0; + LOG.debug("{} join reorder before eager aggregation [changed={}, elapsedMs={}]", + connectContext.getQueryIdentifier(), reorderedPlan != plan, elapsedMs); + } + if (reorderedPlan == plan) { + return plan; + } + return new ColumnPruning().rewriteRoot(reorderedPlan, jobContext); Review Comment: [P1] Restore the Filter(Window) shape after this pruning pass This late `ColumnPruning` inserts `Project(item_sk, rnk)` between q44's `Filter(rnk < 11)` and `Window` after the earlier `PUSH_DOWN_FILTERS` cleanup has already run. `CreatePartitionTopNFromWindow` later matches only `Filter(Window)`, so both rank branches lose their early `PhysicalPartitionTopN` and retain the full window/local+merge sorts. The refreshed goldens show the same loss in all ten q44 variants (20 operators, including SF1000/SF10T). Please run the corresponding filter/project cleanup after this pruning or teach the window rule to see through the project, and retain a shape assertion for both branches. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
