This is an automated email from the ASF dual-hosted git repository. englefly pushed a commit to branch cse-agg-distribute in repository https://gitbox.apache.org/repos/asf/doris.git
commit d30c762ac1a0dc559ce7946c4cef58d3aef19c97 Author: englefly <[email protected]> AuthorDate: Mon Aug 17 00:23:25 2026 +0800 [fe](nereids) Extract aggregate-argument CSE below distribute ### What problem does this PR solve? Problem Summary: ProjectAggregateExpressionsForCse skipped one-phase aggregates whose child is a PhysicalDistribute (aggregate -> distribute -> scan). Such plans therefore never received aggregate-argument CSE: for SELECT SUM(a+b), MAX(a+b) ... GROUP BY g the two aggregate functions each re-evaluated a+b for every row. This also hit the bucketed fusion path (one-phase aggregate + distribute fused into BucketedAggregationNode), which inherited the same shape after the fusion moved to the translator. The post-processor now inserts the CSE project below the distribute instead of between the aggregate and the distribute. This keeps the aggregate -> distribute adjacency (bucketed fusion and property derivation still see the same shape), preserves every distribution-key slot (the group-by slots are part of the extracted input), and places the computed argument inside the scan fragment, so it is evaluated once per row before the exchange. The fused bucketed plan becomes BucketedAgg(sum(x), max(x)) -> Project(a+b AS x) -> scan. Verified: - explain shape plan shows hashAgg[GLOBAL] -> PhysicalDistribute -> PhysicalProject(CSE) -> scan for one-phase aggregates (join child and bucketed fusion scenarios), and the translated plan references the extracted slot from both SUM and MAX. - New regression suite cse_agg_distribute covers the bucketed fusion shape (multiContains on the shared slot, BUCKETED AGGREGATE + VSELECT present) and the plain one-phase join-child shape, plus result correctness. - BucketedAggregateTest (6) and the agg_strategy suites pass. ### Release note None ### Check List (For Author) - Test: Regression test / Unit Test --- .../post/ProjectAggregateExpressionsForCse.java | 50 ++++++++++++- .../agg_strategy/cse_agg_distribute.out | 9 +++ .../agg_strategy/cse_agg_distribute.groovy | 86 ++++++++++++++++++++++ 3 files changed, 143 insertions(+), 2 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/ProjectAggregateExpressionsForCse.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/ProjectAggregateExpressionsForCse.java index c9bb5c6a7c8..62231bc2269 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/ProjectAggregateExpressionsForCse.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/ProjectAggregateExpressionsForCse.java @@ -72,12 +72,21 @@ public class ProjectAggregateExpressionsForCse extends PlanPostProcessor { * Shared CSE projection logic for both PhysicalHashAggregate and * PhysicalBucketedHashAggregate. Extracts common sub-expressions from * aggregate function arguments into a project node beneath the aggregate. + * + * <p>For one-phase aggregates whose child is a PhysicalDistribute + * (aggregate -> distribute -> scan), the CSE project is inserted below the + * distribute so that the distribution-key slots stay intact and the exchange + * only carries the (already pruned) aggregate input. The translator's bucketed + * fusion (fusing one-phase aggregate + distribute into BucketedAggregationNode) + * builds directly on the distribute's child, so the fused plan naturally + * becomes BucketedAgg(sum(x), max(x)) -> Project(a+b AS x) -> scan and the + * common aggregate argument is evaluated once per row instead of once per + * aggregate function.</p> */ private <T extends AbstractPhysicalPlan & Aggregate<? extends Plan>> Plan projectAggregateCse(T aggregate) { // For multi-phase aggregates, only process the 1st phase. - // Bucketed agg is always single-phase, but keep the same guard for safety. - if (aggregate.child() instanceof PhysicalDistribute || aggregate.child() instanceof Aggregate) { + if (aggregate.child() instanceof Aggregate) { return aggregate; } @@ -169,6 +178,43 @@ public class ProjectAggregateExpressionsForCse extends PlanPostProcessor { project = project.withPhysicalPropertiesAndStats(projectPhysicalProperties, project.getStats()); return (Plan) aggregate.withAggOutput(aggOutputReplaced) .withChildren(project); + } else if (aggregate.child() instanceof PhysicalDistribute) { + // One-phase aggregate over a distribute (aggregate -> distribute -> scan): + // insert the CSE project between the distribute and its child, instead of + // between the aggregate and the distribute. This keeps the aggregate's + // child as a distribute (so bucketed fusion and the property machinery + // still see the same shape), and the project lands inside the scan + // fragment, so the common aggregate argument is computed once per row + // before the exchange. After bucketed fusion bypasses the distribute, + // the executed plan is BucketedAgg(sum(x), max(x)) -> Project(a+b AS x) + // -> scan. + PhysicalDistribute<?> distribute = (PhysicalDistribute<?>) aggregate.child(); + List<NamedExpression> projections = new ArrayList<>(); + projections.addAll(inputSlots); + projections.addAll(cseCandidates.values()); + List<Slot> projectOutput = new ImmutableList.Builder<Slot>() + .addAll(inputSlots) + .addAll(slotMap.values()) + .build(); + LogicalProperties projectLogicalProperties = new LogicalProperties( + () -> projectOutput, + () -> DataTrait.EMPTY_TRAIT + ); + AbstractPhysicalPlan distributeChild = ((AbstractPhysicalPlan) distribute.child()); + PhysicalProperties projectPhysicalProperties = ChildOutputPropertyDeriver.computeProjectOutputProperties( + projections, distributeChild.getPhysicalProperties()); + PhysicalProject<? extends Plan> project = new PhysicalProject<>(projections, Optional.empty(), + projectLogicalProperties, + projectPhysicalProperties, + distributeChild.getStats(), + distribute.child()); + // withChildren keeps the distribution spec and physical properties of the + // distribute unchanged; its output now comes from the CSE project, which + // still carries every distribution-key slot (the group-by slots are part + // of inputSlots above). + PhysicalDistribute<Plan> newDistribute = distribute.withChildren(ImmutableList.of(project)); + return (Plan) aggregate.withAggOutput(aggOutputReplaced) + .withChildren(newDistribute); } else { List<NamedExpression> projections = new ArrayList<>(); projections.addAll(inputSlots); diff --git a/regression-test/data/nereids_rules_p0/agg_strategy/cse_agg_distribute.out b/regression-test/data/nereids_rules_p0/agg_strategy/cse_agg_distribute.out new file mode 100644 index 00000000000..b46cfcecf0c --- /dev/null +++ b/regression-test/data/nereids_rules_p0/agg_strategy/cse_agg_distribute.out @@ -0,0 +1,9 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !bucketed_result -- +g1 33 19 +g2 22 15 + +-- !one_phase_join_result -- +g1 33 19 33 19 +g2 22 15 22 15 + diff --git a/regression-test/suites/nereids_rules_p0/agg_strategy/cse_agg_distribute.groovy b/regression-test/suites/nereids_rules_p0/agg_strategy/cse_agg_distribute.groovy new file mode 100644 index 00000000000..b2db5a15a0b --- /dev/null +++ b/regression-test/suites/nereids_rules_p0/agg_strategy/cse_agg_distribute.groovy @@ -0,0 +1,86 @@ +// 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. + +suite("cse_agg_distribute") { + sql "SET enable_nereids_planner=true" + sql "SET enable_fallback_to_original_planner=false" + sql "SET runtime_filter_mode=OFF" + + sql "DROP TABLE IF EXISTS cse_agg_distribute_tbl" + sql """ + CREATE TABLE cse_agg_distribute_tbl ( + id int, + grp varchar(20), + a int, + b int + ) DUPLICATE KEY(id) + DISTRIBUTED BY HASH(id) BUCKETS 3 + PROPERTIES('replication_num' = '1') + """ + sql """ INSERT INTO cse_agg_distribute_tbl VALUES + (1, 'g1', 1, 2), + (2, 'g2', 3, 4), + (3, 'g1', 5, 6), + (4, 'g2', 7, 8), + (5, 'g1', 9, 10) + """ + + // one-phase aggregate over a single scan: SUM(a+b) and MAX(a+b) share the + // same argument, so the aggregate-argument CSE must extract "a+b" into a + // project node and make both functions reference the extracted slot. + String query = "SELECT grp, SUM(a+b), MAX(a+b) FROM cse_agg_distribute_tbl GROUP BY grp" + + // --------------------------------------------------------------------- + // bucketed fusion path (one-phase aggregate -> distribute -> scan is + // fused into BucketedAggregationNode): the CSE project must be preserved + // below the fused node, i.e. BucketedAgg(sum(x), max(x)) -> Project(a+b AS x) + // -> scan. The aggregate output must reference the extracted slot twice + // (once for SUM, once for MAX) instead of recomputing a+b per function. + // --------------------------------------------------------------------- + sql "set enable_bucketed_hash_agg=true" + sql "set bucketed_agg_min_input_rows=0" + sql "set bucketed_agg_high_card_threshold=1.0" + explain { + sql("${query}") + contains("BUCKETED AGGREGATE") + contains("VSELECT") + multiContains("cast(a as BIGINT) + cast(b as BIGINT))[#", 2) + } + order_qt_bucketed_result """${query} ORDER BY grp""" + + // --------------------------------------------------------------------- + // plain one-phase aggregate over a distribute (aggregate is a join child, + // so the distribute is required by the join): the CSE project must be + // inserted below the distribute, keeping the distribution-key slots + // intact, and both aggregates must reference the extracted slot. + // --------------------------------------------------------------------- + sql "set agg_phase=1" + sql "set enable_bucketed_hash_agg=false" + String joinQuery = """ + SELECT t1.grp, t1.s, t1.m, t2.s2, t2.m2 FROM + (SELECT grp, SUM(a+b) s, MAX(a+b) m FROM cse_agg_distribute_tbl GROUP BY grp) t1 + JOIN (SELECT grp, SUM(a+b) s2, MAX(a+b) m2 FROM cse_agg_distribute_tbl GROUP BY grp) t2 + ON t1.grp = t2.grp + """ + explain { + sql("${joinQuery}") + contains("VEXCHANGE") + contains("VSELECT") + multiContains("cast(a as BIGINT) + cast(b as BIGINT))[#", 4) + } + order_qt_one_phase_join_result """${joinQuery} ORDER BY t1.grp""" +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
