This is an automated email from the ASF dual-hosted git repository.
englefly pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3b8142fe487 [refactor](bucket-hash-aggregate) refactor fe part:
implement BucketHashAggregate in physical plan translate phase (#65024)
3b8142fe487 is described below
commit 3b8142fe48740bcd2d54c3c9bb558fe54d5da3f7
Author: minghong <[email protected]>
AuthorDate: Mon Aug 17 15:00:11 2026 +0800
[refactor](bucket-hash-aggregate) refactor fe part: implement
BucketHashAggregate in physical plan translate phase (#65024)
### What problem does this PR solve?
Refactor the FE part of PR #61495: remove PhysicalBucketedHashAggregate
and instead fuse shuffle and aggregation at the translator stage.
The BE-side operators (BucketedAggSinkOperatorX /
BucketedAggSourceOperatorX) and the BucketedAggregationNode remain
unchanged.
Benefits
1. Minimally invasive: Eliminates ~290 lines of dedicated plan node code
and ~100 lines of visitor/cost/property methods across 12 files. Net
reduction of ~200+ lines in the FE codebase.
2. No new optimizer types: Removes PHYSICAL_BUCKETED_HASH_AGGREGATE from
the PlanType enum, PlanVisitor, and all auto-generated pattern
descriptors. The optimizer operates on standard PhysicalHashAggregate +
PhysicalDistribute patterns that it already understands.
3. Cleaner separation of concerns: The "what to compute" decision stays
in the optimizer (cost model discount makes the one-phase path preferred
on single-BE). The "how to execute" decision — fusing aggregation and
distribution into a single BE operator — lives in
the translator, exactly where physical plan → executable plan lowering
belongs.
4. Reuses existing infrastructure: No new property derivation rules
needed. The one-phase aggregate already requests HASH distribution by
group keys. The cost model already compares one-phase vs two-phase
paths. Only a translator-side pattern match and a cost
discount are added.
5. Lower maintenance burden: When new post-processors or visitor methods
are added to the Nereids framework, PhysicalBucketedHashAggregate no
longer needs to be updated in parallel with PhysicalHashAggregate. The
two previously duplicated code paths (translator, post-processors,
property derivers) are now unified.
Issue Number: close #xxx
Related PR: #61495
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../org/apache/doris/nereids/cost/CostModel.java | 33 +-
.../glue/translator/PhysicalPlanTranslator.java | 439 ++++++++++++++-------
.../glue/translator/PlanTranslatorContext.java | 24 ++
.../post/ProjectAggregateExpressionsForCse.java | 12 +-
.../processor/post/RuntimeFilterPruner.java | 9 +-
.../post/materialize/LazySlotPruning.java | 7 -
.../properties/ChildOutputPropertyDeriver.java | 37 +-
.../properties/ChildrenPropertiesRegulator.java | 56 +++
.../nereids/properties/RequestPropertyDeriver.java | 9 -
.../implementation/SplitAggWithoutDistinct.java | 326 ---------------
.../doris/nereids/stats/StatsCalculator.java | 7 -
.../apache/doris/nereids/trees/plans/PlanType.java | 1 -
.../physical/PhysicalBucketedHashAggregate.java | 293 --------------
.../nereids/trees/plans/visitor/PlanVisitor.java | 6 -
.../apache/doris/nereids/util/AggregateUtils.java | 45 +++
.../properties/RequestPropertyDeriverTest.java | 20 -
.../implementation/BucketedAggregateTest.java | 69 +++-
.../statistics/query/QueryStatsRecorderTest.java | 28 --
.../agg_strategy/bucketed_hash_agg.out | 40 ++
.../agg_strategy/bucketed_hash_agg.groovy | 162 ++++++++
20 files changed, 735 insertions(+), 888 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
index e21832b00bc..92aafb32006 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
@@ -38,13 +38,13 @@ import
org.apache.doris.nereids.trees.expressions.ComparisonPredicate;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.plans.AggMode;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanNodeAndHash;
import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalCTEProducer;
import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
@@ -65,6 +65,7 @@ import
org.apache.doris.nereids.trees.plans.physical.PhysicalStorageLayerAggrega
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalUnion;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
+import org.apache.doris.nereids.util.AggregateUtils;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.statistics.ColumnStatistic;
@@ -89,6 +90,8 @@ class CostModel extends PlanVisitor<Cost, PlanContext> {
// The cost of using external tables should be somewhat higher than using
internal tables,
// so when encountering a scan of an external table, a coefficient should
be applied.
static final double EXTERNAL_TABLE_SCAN_FACTOR = 5;
+ static final double BUCKETED_AGG_COST_DISCOUNT = 0.5;
+
private static final Logger LOG = LogManager.getLogger(CostModel.class);
private final int beNumber;
private final int parallelInstance;
@@ -96,7 +99,7 @@ class CostModel extends PlanVisitor<Cost, PlanContext> {
public CostModel(ConnectContext connectContext) {
SessionVariable sessionVariable = connectContext.getSessionVariable();
- if (sessionVariable.getBeNumberForTest() != -1) {
+ if (sessionVariable.getBeNumberForTest() > 0) {
// shape test, fix the BE number and instance number
beNumber = sessionVariable.getBeNumberForTest();
parallelInstance = 8;
@@ -363,25 +366,21 @@ class CostModel extends PlanVisitor<Cost, PlanContext> {
boolean isPartitioned =
!aggregate.getGroupByExpressions().isEmpty()
|| aggregate.getPartitionExpressions().filter(expressions
-> !expressions.isEmpty()).isPresent();
int factor = isPartitioned ? beNumber : 1;
- // global
+ double rowCost = inputStatistics.getRowCount() / factor;
+ // Bucketed fusion discount: when the one-phase GLOBAL
INPUT_TO_RESULT
+ // aggregate is eligible for translator fusion (correctness +
data-volume
+ // gates are enforced by ChildrenPropertiesRegulator), apply a
discount
+ // to prefer this path over two-phase aggregation.
+ if (aggregate.getAggMode() == AggMode.INPUT_TO_RESULT
+ && AggregateUtils.isBucketedHashAggEnabled(
+ aggregate.getGroupByExpressions().size())) {
+ rowCost *= BUCKETED_AGG_COST_DISCOUNT;
+ }
return Cost.of(context.getSessionVariable(),
- exprCost / 100 + inputStatistics.getRowCount() / factor,
- inputStatistics.getRowCount() / factor, 0);
+ exprCost / 100 + rowCost, rowCost, 0);
}
}
- @Override
- public Cost visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> aggregate,
PlanContext context) {
- // Bucketed agg is similar to one-phase agg: all computation on a
single BE,
- // but avoids exchange overhead. Cost is comparable to one-phase agg.
- Statistics inputStatistics = context.getChildStatistics(0);
- double exprCost = expressionTreeCost(aggregate.getExpressions());
- return Cost.of(context.getSessionVariable(),
- exprCost / 100 + inputStatistics.getRowCount(),
- inputStatistics.getRowCount(), 0);
- }
-
@Override
public Cost visitPhysicalHashJoin(
PhysicalHashJoin<? extends Plan, ? extends Plan> physicalHashJoin,
PlanContext context) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index f466f3c2884..ee2b277d1ca 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -92,6 +92,7 @@ import org.apache.doris.nereids.trees.expressions.WindowFrame;
import org.apache.doris.nereids.trees.expressions.functions.Udf;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregatePhase;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
import
org.apache.doris.nereids.trees.expressions.functions.scalar.UniqueFunction;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
@@ -109,7 +110,6 @@ import
org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalSort;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalBaseExternalTableSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalBlackholeSink;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
@@ -162,6 +162,7 @@ import org.apache.doris.nereids.types.DataType;
import org.apache.doris.nereids.types.JsonType;
import org.apache.doris.nereids.types.MapType;
import org.apache.doris.nereids.types.StructType;
+import org.apache.doris.nereids.util.AggregateUtils;
import org.apache.doris.nereids.util.ExpressionUtils;
import org.apache.doris.nereids.util.JoinUtils;
import org.apache.doris.nereids.util.RowStoreFetchChecker;
@@ -1168,6 +1169,12 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
PhysicalHashAggregate<? extends Plan> aggregate,
PlanTranslatorContext context) {
+ // Bucketed fusion path: fuse one-phase GLOBAL aggregate + distribute
+ // into BucketedAggregationNode when applicable (single-BE, no
exchange needed).
+ if (shouldUseBucketedFusion(aggregate, context)) {
+ return visitBucketedFusion(aggregate, context);
+ }
+
PlanFragment inputPlanFragment = aggregate.child(0).accept(this,
context);
List<List<Expr>> distributeExprLists =
getDistributeExprs(aggregate.child(0));
@@ -1178,50 +1185,15 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
List<SlotReference> groupSlots =
collectGroupBySlots(groupByExpressions, outputExpressions);
ArrayList<Expr> execGroupingExpressions =
translateGroupByExprs(groupByExpressions, context);
// 2. collect agg expressions and generate agg function to slot
reference map
- List<Slot> aggFunctionOutput = Lists.newArrayList();
- ArrayList<FunctionCallExpr> execAggregateFunctions =
Lists.newArrayListWithCapacity(outputExpressions.size());
- AtomicBoolean hasPartialInAggFunc = new AtomicBoolean(false);
- Set<AggregateExpression> processedAggregateExpressions =
Sets.newIdentityHashSet();
- for (NamedExpression o : outputExpressions) {
- if (o.containsType(AggregateExpression.class)) {
- aggFunctionOutput.add(o.toSlot());
-
- o.foreach(c -> {
- if (c instanceof SessionVarGuardExpr) {
- SessionVarGuardExpr guardExpr = (SessionVarGuardExpr)
c;
- if (guardExpr.child() instanceof AggregateExpression) {
- AggregateExpression aggregateExpression =
(AggregateExpression) guardExpr.child();
- if
(processedAggregateExpressions.add(aggregateExpression)) {
- execAggregateFunctions.add(
- (FunctionCallExpr)
ExpressionTranslator.translate(guardExpr, context)
- );
- hasPartialInAggFunc.set(
-
aggregateExpression.getAggregateParam().aggMode.productAggregateBuffer);
- }
- }
- // Continue through transparent guards unless this
guard directly wraps
- // the aggregate expression already processed above.
- return guardExpr.child() instanceof
AggregateExpression;
- }
- if (c instanceof AggregateExpression) {
- AggregateExpression aggregateExpression =
(AggregateExpression) c;
- if
(processedAggregateExpressions.add(aggregateExpression)) {
- execAggregateFunctions.add(
- (FunctionCallExpr)
ExpressionTranslator.translate(aggregateExpression, context)
- );
- hasPartialInAggFunc.set(
-
aggregateExpression.getAggregateParam().aggMode.productAggregateBuffer);
- }
- return true;
- }
- return false;
- });
- }
- }
+ boolean[] hasPartialInAggFunc = new boolean[1];
+ Pair<List<Slot>, ArrayList<FunctionCallExpr>> aggResult =
+ collectAggFunctions(outputExpressions, hasPartialInAggFunc,
context);
+ List<Slot> aggFunctionOutput = aggResult.first;
+ ArrayList<FunctionCallExpr> execAggregateFunctions = aggResult.second;
// An agg may have different functions, some product buffer, some
product result.
// The criterion for passing it to the be stage is: as long as there
is a product buffer function in agg,
// it must be isPartial
- boolean isPartial = hasPartialInAggFunc.get();
+ boolean isPartial = hasPartialInAggFunc[0];
// 3. generate output tuple
Pair<TupleDescriptor, List<Integer>> tupleAndIds =
@@ -1305,92 +1277,6 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
return inputPlanFragment;
}
- @Override
- public PlanFragment visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> aggregate,
- PlanTranslatorContext context) {
-
- PlanFragment inputPlanFragment = aggregate.child(0).accept(this,
context);
-
- List<Expression> groupByExpressions =
aggregate.getGroupByExpressions();
- List<NamedExpression> outputExpressions =
aggregate.getOutputExpressions();
-
- // 1. generate slot reference for each group expression
- List<SlotReference> groupSlots =
collectGroupBySlots(groupByExpressions, outputExpressions);
- ArrayList<Expr> execGroupingExpressions =
translateGroupByExprs(groupByExpressions, context);
-
- // 2. collect agg expressions and generate agg function to slot
reference map
- // Mirror SessionVarGuardExpr handling from
visitPhysicalHashAggregate: if an aggregate
- // output is wrapped by SessionVarGuardExpr, translate the guard
(which preserves
- // session-sensitive type behavior) rather than the inner
AggregateExpression directly.
- List<Slot> aggFunctionOutput = Lists.newArrayList();
- ArrayList<FunctionCallExpr> execAggregateFunctions =
Lists.newArrayListWithCapacity(outputExpressions.size());
- Set<AggregateExpression> processedAggregateExpressions =
Sets.newIdentityHashSet();
- for (NamedExpression o : outputExpressions) {
- if (o.containsType(AggregateExpression.class)) {
- aggFunctionOutput.add(o.toSlot());
-
- o.foreach(c -> {
- if (c instanceof SessionVarGuardExpr) {
- SessionVarGuardExpr guardExpr = (SessionVarGuardExpr)
c;
- if (guardExpr.child() instanceof AggregateExpression) {
- AggregateExpression aggregateExpression =
(AggregateExpression) guardExpr.child();
- if
(processedAggregateExpressions.add(aggregateExpression)) {
- execAggregateFunctions.add(
- (FunctionCallExpr)
ExpressionTranslator.translate(guardExpr, context)
- );
- }
- }
- return true;
- }
- if (c instanceof AggregateExpression) {
- AggregateExpression aggregateExpression =
(AggregateExpression) c;
- if
(processedAggregateExpressions.add(aggregateExpression)) {
- execAggregateFunctions.add(
- (FunctionCallExpr)
ExpressionTranslator.translate(aggregateExpression, context)
- );
- }
- return true;
- }
- return false;
- });
- }
- }
-
- // 3. generate output tuple
- Pair<TupleDescriptor, List<Integer>> tupleAndIds =
- buildAggOutputTuple(groupSlots, aggFunctionOutput, context);
- TupleDescriptor outputTupleDesc = tupleAndIds.first;
- List<Integer> aggFunOutputIds = tupleAndIds.second;
-
- // Bucketed agg uses AggPhase.FIRST (update semantics): raw input ->
final result.
- // Not partial — always needsFinalize.
- AggregateInfo aggInfo = AggregateInfo.create(execGroupingExpressions,
execAggregateFunctions,
- aggFunOutputIds, false /* isPartial */, outputTupleDesc,
- AggregateInfo.AggPhase.FIRST);
-
- BucketedAggregationNode bucketedAggNode = new BucketedAggregationNode(
- context.nextPlanNodeId(), inputPlanFragment.getPlanRoot(),
aggInfo, true /* needsFinalize */);
-
- bucketedAggNode.setNereidsId(aggregate.getId());
- context.getNereidsIdToPlanNodeIdMap().put(aggregate.getId(),
bucketedAggNode.getId());
-
- // Bucketed agg runs entirely within a single fragment. No exchange
needed.
- // Do NOT set hasColocatePlanNode — bucketed agg does not require
colocate
- // semantics (one-instance-per-bucket). Bucket assignment is done at
the BE
- // level via hash partitioning. Leaving this unset allows the fragment
to
- // route through UnassignedScanSingleOlapTableJob, which respects
- // parallel_pipeline_task_num as an upper bound on parallelism.
-
- setPlanRoot(inputPlanFragment, bucketedAggNode, aggregate);
- if (aggregate.getStats() != null) {
- bucketedAggNode.setCardinality((long)
aggregate.getStats().getRowCount());
- }
- updateLegacyPlanIdToPhysicalPlan(inputPlanFragment.getPlanRoot(),
aggregate);
-
- return inputPlanFragment;
- }
-
@Override
public PlanFragment visitPhysicalStorageLayerAggregate(
PhysicalStorageLayerAggregate storageLayerAggregate,
PlanTranslatorContext context) {
@@ -1675,8 +1561,15 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
PhysicalHashJoin<PhysicalPlan, PhysicalPlan> physicalHashJoin
= (PhysicalHashJoin<PhysicalPlan, PhysicalPlan>) hashJoin;
// NOTICE: We must visit from right to left, to ensure the last
fragment is root fragment
- PlanFragment rightFragment = hashJoin.child(1).accept(this, context);
- PlanFragment leftFragment = hashJoin.child(0).accept(this, context);
+ context.enterFragmentMergeChild();
+ PlanFragment rightFragment;
+ PlanFragment leftFragment;
+ try {
+ rightFragment = hashJoin.child(1).accept(this, context);
+ leftFragment = hashJoin.child(0).accept(this, context);
+ } finally {
+ context.exitFragmentMergeChild();
+ }
List<List<Expr>> distributeExprLists
= getDistributeExprs(physicalHashJoin.left(),
physicalHashJoin.right());
@@ -1947,8 +1840,15 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
// TODO: we should add a helper method to wrap this logic.
// Maybe something like private List<PlanFragment>
postOrderVisitChildren(
// PhysicalPlan plan, PlanVisitor visitor, Context context).
- PlanFragment rightFragment = nestedLoopJoin.child(1).accept(this,
context);
- PlanFragment leftFragment = nestedLoopJoin.child(0).accept(this,
context);
+ context.enterFragmentMergeChild();
+ PlanFragment rightFragment;
+ PlanFragment leftFragment;
+ try {
+ rightFragment = nestedLoopJoin.child(1).accept(this, context);
+ leftFragment = nestedLoopJoin.child(0).accept(this, context);
+ } finally {
+ context.exitFragmentMergeChild();
+ }
List<List<Expr>> distributeExprLists
= getDistributeExprs(nestedLoopJoin.child(0),
nestedLoopJoin.child(1));
PlanNode leftFragmentPlanRoot = leftFragment.getPlanRoot();
@@ -2443,8 +2343,13 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
public PlanFragment visitPhysicalSetOperation(
PhysicalSetOperation setOperation, PlanTranslatorContext context) {
List<PlanFragment> childrenFragments = new ArrayList<>();
- for (Plan plan : setOperation.children()) {
- childrenFragments.add(plan.accept(this, context));
+ context.enterFragmentMergeChild();
+ try {
+ for (Plan plan : setOperation.children()) {
+ childrenFragments.add(plan.accept(this, context));
+ }
+ } finally {
+ context.exitFragmentMergeChild();
}
List<List<Expr>> distributeExprLists =
getDistributeExprs(setOperation.children().toArray(new Plan[0]));
TupleDescriptor setTuple = generateTupleDesc(setOperation.getOutput(),
null, context);
@@ -3217,9 +3122,271 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
return leftFragment;
}
+ /**
+ * Check whether the one-phase GLOBAL hash aggregate can be fused with its
+ * distribute child into a BucketedAggregationNode. This eliminates
exchange
+ * overhead on single-BE deployments by using in-memory per-bucket merging.
+ */
+ private boolean shouldUseBucketedFusion(PhysicalHashAggregate<? extends
Plan> aggregate,
+ PlanTranslatorContext context) {
+ // Shared eligibility: session var, single-BE, GROUP BY, smooth upgrade
+ if
(!AggregateUtils.isBucketedHashAggEnabled(aggregate.getGroupByExpressions().size()))
{
+ return false;
+ }
+ // Must be one-phase: GLOBAL + INPUT_TO_RESULT
+ if (aggregate.getAggPhase() != AggPhase.GLOBAL
+ || aggregate.getAggMode() != AggMode.INPUT_TO_RESULT) {
+ return false;
+ }
+ // Exclude one-phase-only aggregates (e.g. GROUP_CONCAT with ORDER BY).
+ // BucketedAggregationNode has no sort-info field, so fusing would drop
+ // the aggregate ORDER BY contract. Only aggregates supporting
two-phase
+ // execution can be safely fused.
+ if (!supportsTwoPhaseAgg(aggregate)) {
+ return false;
+ }
+ // BucketedAggregationNode does not support sortByGroupKey
(PushTopnToAgg
+ // optimization). Regular AggregationNode fills sort info; fusing
would drop it.
+ if (aggregate.getTopnPushInfo() != null) {
+ return false;
+ }
+ // Child must be PhysicalDistribute with hash distribution matching
group keys
+ Plan child = aggregate.child(0);
+ if (!(child instanceof PhysicalDistribute)) {
+ return false;
+ }
+ // Bucketed fusion bypasses the distribute/exchange and builds
directly on the
+ // child fragment. When the child subtree contains a CTE consumer
(materialized
+ // multicast CTE), the child fragment is the MultiCastPlanFragment; a
parent
+ // distribute would then treat the aggregate output slots as consumer
slots and
+ // fail with "Required producer slot ... doesn't exist". Fall back to
the
+ // regular one-phase path (which keeps the exchange) for such plans.
+ if (containsCTEConsumer(child)) {
+ return false;
+ }
+ // The distribute's child subtree must be a unary pipeline over
exactly one
+ // olap scan. Fusing an aggregate whose input contains a join / set-op
/ CTE
+ // subtree would leave multiple olap scans in a single fragment
(rejected by
+ // UnassignedJobBuilder: "Not supported multiple scan multiple
OlapTable but
+ // not contains colocate join or bucket shuffle join"), and fusing
over a
+ // nested aggregate would break the bucket alignment between stages.
+ if (!isSingleOlapScanPipeline(aggregate.child(0).child(0))) {
+ return false;
+ }
+ // The parent is a fragment-merging node (join / set-op) that consumes
this
+ // fragment without an exchange boundary: fusing removes the exchange
that
+ // keeps the scan in its own fragment, so multiple scans would end up
in the
+ // same fragment and the scan-assignment would fail. Only fuse when the
+ // parent chain keeps an exchange boundary (e.g. a top-level
aggregate).
+ if (context.isInFragmentMergeChild()) {
+ return false;
+ }
+ DistributionSpec distSpec = ((PhysicalDistribute<?>)
child).getDistributionSpec();
+ if (!(distSpec instanceof DistributionSpecHash)) {
+ return false;
+ }
+ List<ExprId> distKeys = ((DistributionSpecHash)
distSpec).getOrderedShuffledColumns();
+ List<ExprId> groupByKeys = aggregate.getGroupByExpressions().stream()
+ .filter(SlotReference.class::isInstance)
+ .map(SlotReference.class::cast)
+ .map(SlotReference::getExprId)
+ .collect(Collectors.toList());
+ return distKeys.equals(groupByKeys);
+ }
+
+ /** Returns true if the plan subtree contains a physical CTE consumer. */
+ private boolean containsCTEConsumer(Plan plan) {
+ if (plan instanceof PhysicalCTEConsumer) {
+ return true;
+ }
+ for (Plan child : plan.children()) {
+ if (containsCTEConsumer(child)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns true if the plan subtree is a unary pipeline over exactly one
olap
+ * scan, i.e. it translates into a single-scan fragment that bucketed
fusion
+ * can safely build upon. Subtrees containing fragment-merging or
+ * distribution-changing nodes (join / set-op / CTE / nested aggregate /
+ * storage-layer aggregate) are rejected.
+ */
+ private boolean isSingleOlapScanPipeline(Plan plan) {
+ if (plan instanceof PhysicalOlapScan) {
+ return true;
+ }
+ if (plan instanceof PhysicalHashJoin
+ || plan instanceof PhysicalNestedLoopJoin
+ || plan instanceof PhysicalSetOperation
+ || plan instanceof PhysicalCTEConsumer
+ || plan instanceof PhysicalCTEAnchor
+ || plan instanceof PhysicalHashAggregate
+ || plan instanceof PhysicalStorageLayerAggregate) {
+ return false;
+ }
+ if (plan.children().size() == 1) {
+ return isSingleOlapScanPipeline(plan.child(0));
+ }
+ return false;
+ }
+
+ /**
+ * Check whether all aggregate functions in this physical hash aggregate
+ * support two-phase execution. One-phase-only aggregates (e.g.
GROUP_CONCAT
+ * with ORDER BY) cannot be bucketed because BucketedAggregationNode does
not
+ * carry sort-info metadata (aggSortInfos); fusing them would drop the
+ * aggregate ORDER BY contract and produce unordered results.
+ */
+ private boolean supportsTwoPhaseAgg(PhysicalHashAggregate<? extends Plan>
aggregate) {
+ for (NamedExpression o : aggregate.getOutputExpressions()) {
+ AtomicBoolean foundOnePhaseOnly = new AtomicBoolean(false);
+ o.foreach(c -> {
+ if (c instanceof OrderExpression) {
+ // Any aggregate function with an internal ORDER BY
+ // (e.g. GROUP_CONCAT(... ORDER BY ...)) needs sort-info
+ // metadata, which BucketedAggregationNode does not carry.
+ foundOnePhaseOnly.set(true);
+ return false;
+ }
+ if (c instanceof AggregateExpression) {
+ AggregateFunction func = ((AggregateExpression)
c).getFunction();
+ if (!func.supportAggregatePhase(AggregatePhase.TWO)) {
+ foundOnePhaseOnly.set(true);
+ }
+ return true;
+ }
+ return false;
+ });
+ if (foundOnePhaseOnly.get()) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Fuse a one-phase GLOBAL hash aggregate and its PhysicalDistribute child
+ * into a BucketedAggregationNode, skipping the exchange node entirely.
+ * Visits the distribute's child directly to keep everything in one
fragment.
+ */
+ private PlanFragment visitBucketedFusion(
+ PhysicalHashAggregate<? extends Plan> aggregate,
+ PlanTranslatorContext context) {
+ // Visit the distribute's direct child, bypassing the distribute
entirely.
+ // This avoids creating an ExchangeNode that bucketed agg does not
need.
+ Plan distributeChild = aggregate.child(0).child(0);
+ PlanFragment inputPlanFragment = distributeChild.accept(this, context);
+
+ List<Expression> groupByExpressions =
aggregate.getGroupByExpressions();
+ List<NamedExpression> outputExpressions =
aggregate.getOutputExpressions();
+
+ // 1. generate slot reference for each group expression
+ List<SlotReference> groupSlots =
collectGroupBySlots(groupByExpressions, outputExpressions);
+ ArrayList<Expr> execGroupingExpressions =
translateGroupByExprs(groupByExpressions, context);
+
+ // 2. collect agg expressions and generate agg function to slot
reference map.
+ // Reuse the shared helper from visitPhysicalHashAggregate; the
bucketed
+ // path passes null for hasPartialOut (never partial, always
needsFinalize).
+ Pair<List<Slot>, ArrayList<FunctionCallExpr>> aggResult =
+ collectAggFunctions(outputExpressions, null, context);
+ List<Slot> aggFunctionOutput = aggResult.first;
+ ArrayList<FunctionCallExpr> execAggregateFunctions = aggResult.second;
+
+ // 3. generate output tuple
+ Pair<TupleDescriptor, List<Integer>> tupleAndIds =
+ buildAggOutputTuple(groupSlots, aggFunctionOutput, context);
+ TupleDescriptor outputTupleDesc = tupleAndIds.first;
+ List<Integer> aggFunOutputIds = tupleAndIds.second;
+
+ // Bucketed agg uses AggPhase.FIRST (update semantics): raw input ->
final result.
+ // Not partial — always needsFinalize.
+ AggregateInfo aggInfo = AggregateInfo.create(execGroupingExpressions,
execAggregateFunctions,
+ aggFunOutputIds, false /* isPartial */, outputTupleDesc,
+ AggregateInfo.AggPhase.FIRST);
+
+ BucketedAggregationNode bucketedAggNode = new BucketedAggregationNode(
+ context.nextPlanNodeId(), inputPlanFragment.getPlanRoot(),
aggInfo, true);
+
+ bucketedAggNode.setNereidsId(aggregate.getId());
+ context.getNereidsIdToPlanNodeIdMap().put(aggregate.getId(),
bucketedAggNode.getId());
+
+ // Do NOT set hasColocatePlanNode — bucketed agg uses its own
hash-based
+ // bucket assignment, not colocate semantics. This allows the fragment
to
+ // route through UnassignedScanSingleOlapTableJob, which respects
+ // parallel_pipeline_task_num as an upper bound on parallelism.
+ setPlanRoot(inputPlanFragment, bucketedAggNode, aggregate);
+ if (aggregate.getStats() != null) {
+ bucketedAggNode.setCardinality((long)
aggregate.getStats().getRowCount());
+ }
+ updateLegacyPlanIdToPhysicalPlan(inputPlanFragment.getPlanRoot(),
aggregate);
+ return inputPlanFragment;
+ }
+
+ /**
+ * Collect aggregate function outputs and translate them to legacy
FunctionCallExpr.
+ * Shared by visitPhysicalHashAggregate and visitBucketedFusion.
+ *
+ * @param hasPartialOut if non-null and length >= 1, hasPartialOut[0] is
set to
+ * true when any aggregate function produces a buffer (i.e. is
partial).
+ * The bucketed path passes null.
+ */
+ private Pair<List<Slot>, ArrayList<FunctionCallExpr>> collectAggFunctions(
+ List<NamedExpression> outputExpressions,
+ boolean[] hasPartialOut,
+ PlanTranslatorContext context) {
+ List<Slot> aggFunctionOutput = Lists.newArrayList();
+ ArrayList<FunctionCallExpr> execAggregateFunctions =
+ Lists.newArrayListWithCapacity(outputExpressions.size());
+ Set<AggregateExpression> processed = Sets.newIdentityHashSet();
+ for (NamedExpression o : outputExpressions) {
+ if (o.containsType(AggregateExpression.class)) {
+ aggFunctionOutput.add(o.toSlot());
+ collectAggInTree(o, processed, execAggregateFunctions,
hasPartialOut, context);
+ }
+ }
+ return Pair.of(aggFunctionOutput, execAggregateFunctions);
+ }
+
+ /** Walk the expression tree to find and translate AggregateExpression
nodes. */
+ private void collectAggInTree(Expression expr,
+ Set<AggregateExpression> processed,
+ ArrayList<FunctionCallExpr> out,
+ boolean[] hasPartialOut,
+ PlanTranslatorContext context) {
+ if (expr instanceof SessionVarGuardExpr) {
+ SessionVarGuardExpr guard = (SessionVarGuardExpr) expr;
+ if (guard.child() instanceof AggregateExpression) {
+ AggregateExpression ae = (AggregateExpression) guard.child();
+ if (processed.add(ae)) {
+ out.add((FunctionCallExpr)
ExpressionTranslator.translate(guard, context));
+ if (hasPartialOut != null) {
+ hasPartialOut[0] |=
ae.getAggregateParam().aggMode.productAggregateBuffer;
+ }
+ }
+ }
+ return;
+ }
+ if (expr instanceof AggregateExpression) {
+ AggregateExpression ae = (AggregateExpression) expr;
+ if (processed.add(ae)) {
+ out.add((FunctionCallExpr) ExpressionTranslator.translate(ae,
context));
+ if (hasPartialOut != null) {
+ hasPartialOut[0] |=
ae.getAggregateParam().aggMode.productAggregateBuffer;
+ }
+ }
+ return;
+ }
+ for (Expression child : expr.children()) {
+ collectAggInTree(child, processed, out, hasPartialOut, context);
+ }
+ }
+
/**
* Translate group-by expressions from Nereids Expression to legacy Expr.
- * Shared by visitPhysicalHashAggregate and
visitPhysicalBucketedHashAggregate.
+ * Shared by visitPhysicalHashAggregate.
*/
private ArrayList<Expr> translateGroupByExprs(List<Expression>
groupByExpressions,
PlanTranslatorContext context) {
@@ -3237,7 +3404,7 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
/**
* Build output tuple descriptor and aggregate function output slot IDs.
* Returns Pair(outputTupleDesc, aggFunOutputIds).
- * Shared by visitPhysicalHashAggregate and
visitPhysicalBucketedHashAggregate.
+ * Shared by visitPhysicalHashAggregate.
*/
private Pair<TupleDescriptor, List<Integer>> buildAggOutputTuple(
List<SlotReference> groupSlots, List<Slot> aggFunctionOutput,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
index 936f4cbe1a9..0a09cd0670b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
@@ -111,6 +111,18 @@ public class PlanTranslatorContext {
private final Map<ScanNode, Set<SlotId>> statsUnknownColumnsMap =
Maps.newHashMap();
+ /**
+ * Depth of fragment-merging binary nodes (hash join / nested loop join /
+ * set operation) whose children are being visited right now. Bucketed
fusion
+ * removes the exchange node that would otherwise keep an olap scan in its
own
+ * fragment; when the fused fragment is consumed by such a node the scan
gets
+ * merged into a fragment that already contains other scans, which the
+ * scan-assignment jobs reject ("Not supported multiple scan multiple
+ * OlapTable but not contains colocate join or bucket shuffle join"). The
+ * translator therefore skips bucketed fusion while inside a merge child.
+ */
+ private int fragmentMergeChildDepth = 0;
+
// Per-node "is there a serial operator between me and the pipeline's
sink" flag.
// Mirrors BE's any_of(operators[idx..end], is_serial_operator) check used
by
// _add_local_exchange / need_to_local_exchange to skip LE insertion when
an ancestor
@@ -325,6 +337,18 @@ public class PlanTranslatorContext {
exprIdToColumnRef.put(exprId, columnRefExpr);
}
+ public void enterFragmentMergeChild() {
+ fragmentMergeChildDepth++;
+ }
+
+ public void exitFragmentMergeChild() {
+ fragmentMergeChildDepth--;
+ }
+
+ public boolean isInFragmentMergeChild() {
+ return fragmentMergeChildDepth > 0;
+ }
+
/**
* merge source fragment info into target fragment.
* include runtime filter info and fragment attribute.
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..a77640a55dc 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
@@ -32,7 +32,6 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunctio
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
@@ -61,16 +60,9 @@ public class ProjectAggregateExpressionsForCse extends
PlanPostProcessor {
(PhysicalHashAggregate<? extends Plan>) super.visit(aggregate,
ctx));
}
- @Override
- public Plan visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> aggregate,
CascadesContext ctx) {
- return projectAggregateCse(
- (PhysicalBucketedHashAggregate<? extends Plan>)
super.visit(aggregate, ctx));
- }
-
/**
- * Shared CSE projection logic for both PhysicalHashAggregate and
- * PhysicalBucketedHashAggregate. Extracts common sub-expressions from
+ * Shared CSE projection logic for PhysicalHashAggregate.
+ * Extracts common sub-expressions from
* aggregate function arguments into a project node beneath the aggregate.
*/
private <T extends AbstractPhysicalPlan & Aggregate<? extends Plan>>
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
index 42df8b9e7c8..ed0a1d17a94 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/RuntimeFilterPruner.java
@@ -26,7 +26,6 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.AbstractPlan;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalFilter;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
@@ -242,15 +241,9 @@ public class RuntimeFilterPruner extends PlanPostProcessor
{
return propagateEffectiveSrc(aggregate, context);
}
- @Override
- public PhysicalBucketedHashAggregate visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> aggregate,
CascadesContext context) {
- return propagateEffectiveSrc(aggregate, context);
- }
-
/**
* Visit child and propagate effective source type if applicable.
- * Shared by visitPhysicalHashAggregate and
visitPhysicalBucketedHashAggregate.
+ * Shared by visitPhysicalHashAggregate.
*
* Note: agg is not regarded as an effective source itself. For example:
* q1: A join (select x, sum(y) as z from B group by x) T on A.a = T.x
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazySlotPruning.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazySlotPruning.java
index 3273a7d82d3..fce964758f1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazySlotPruning.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/materialize/LazySlotPruning.java
@@ -24,7 +24,6 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalPlan;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalFileScan;
@@ -213,12 +212,6 @@ public class LazySlotPruning extends
DefaultPlanRewriter<LazySlotPruning.Context
return aggregate;
}
- @Override
- public Plan visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> aggregate, Context
context) {
- return aggregate;
- }
-
@Override
public Plan visitPhysicalCTEConsumer(PhysicalCTEConsumer cteConsumer,
Context context) {
return cteConsumer;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
index 823231aa317..2df7723a7ab 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildOutputPropertyDeriver.java
@@ -29,10 +29,10 @@ import
org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import
org.apache.doris.nereids.trees.expressions.functions.table.TableValuedFunction;
+import org.apache.doris.nereids.trees.plans.AggMode;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalSort;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
@@ -61,6 +61,7 @@ import
org.apache.doris.nereids.trees.plans.physical.PhysicalWindow;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalWorkTableReference;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.util.AggregateUtils;
import org.apache.doris.nereids.util.JoinUtils;
import org.apache.doris.nereids.util.Utils;
import org.apache.doris.qe.ConnectContext;
@@ -191,22 +192,36 @@ public class ChildOutputPropertyDeriver extends
PlanVisitor<PhysicalProperties,
case GLOBAL:
case DISTINCT_LOCAL:
case DISTINCT_GLOBAL:
+ // Bucketed hash agg fusion: when the one-phase GLOBAL
aggregate
+ // will be fused with its distribute child into
BucketedAggregationNode,
+ // the output is NOT hash-distributed (256-bucket internal
hash is
+ // not shuffle-compatible). Advertise ANY to prevent parent
operators
+ // from incorrectly skipping exchanges.
+ if (agg.getAggPhase().isGlobal()
+ && agg.getAggMode() == AggMode.INPUT_TO_RESULT
+ && AggregateUtils.isBucketedHashAggEnabled(
+ agg.getGroupByExpressions().size())
+ &&
isShuffleCompatible(childOutputProperty.getDistributionSpec())) {
+ return PhysicalProperties.ANY;
+ }
return new
PhysicalProperties(childOutputProperty.getDistributionSpec());
default:
throw new RuntimeException("Could not derive output properties
for agg phase: " + agg.getAggPhase());
}
}
- @Override
- public PhysicalProperties visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> agg, PlanContext
context) {
- Preconditions.checkState(childrenOutputProperties.size() == 1);
- // Although bucketed agg internally re-distributes data into 256
buckets by
- // group-by keys (so the output is "complete" per group), we cannot
claim
- // EXECUTION_BUCKETED distribution because the 256-bucket hash
function differs
- // from the shuffle hash function. Downstream operators expecting
shuffle-compatible
- // distribution would be incorrect. Preserve distribution ANY.
- return PhysicalProperties.ANY;
+ /**
+ * Returns true if the child's distribution is a shuffle-compatible hash
that the
+ * bucketed fusion pattern produces (ShuffleType.REQUIRE).
EXECUTION_BUCKETED is
+ * used by CTE dedup and colocate-join patterns — those should NOT be
treated as
+ * bucketed-fusion output because their hash functions ARE
shuffle-compatible.
+ */
+ private static boolean isShuffleCompatible(DistributionSpec distSpec) {
+ if (!(distSpec instanceof DistributionSpecHash)) {
+ return false;
+ }
+ DistributionSpecHash hashSpec = (DistributionSpecHash) distSpec;
+ return hashSpec.getShuffleType() == ShuffleType.REQUIRE;
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
index 6aa9d01c7cc..e94e4f7dfae 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java
@@ -159,6 +159,13 @@ public class ChildrenPropertiesRegulator extends
PlanVisitor<List<List<PhysicalP
// group by key is skew
return skewOnShuffleExpr(aggregate);
} else {
+ // Bucketed hash agg exception: allow one-phase GLOBAL + distribute
+ // pattern so the translator can fuse it into
BucketedAggregationNode.
+ // Gate with data-volume checks using group-level statistics to
avoid
+ // generating this pattern when bucketed agg is unsuitable.
+ if
(AggregateUtils.isBucketedHashAggEnabled(aggregate.getGroupByExpressions().size()))
{
+ return !bucketedDataVolumeGatesPass(aggregate);
+ }
return true;
}
}
@@ -218,6 +225,55 @@ public class ChildrenPropertiesRegulator extends
PlanVisitor<List<List<PhysicalP
&& children.get(0).getPlan() instanceof PhysicalDistribute;
}
+ /**
+ * Check data-volume gates for bucketed hash aggregation using group-level
+ * statistics available during property regulation. Returns true if the
+ * pattern should be allowed (stats pass or unavailable), false if it
should
+ * be banned due to unfavorable data characteristics.
+ * Mirrors the checks from the old implementBucketedPhase.
+ */
+ private boolean bucketedDataVolumeGatesPass(PhysicalHashAggregate<?
extends Plan> aggregate) {
+ Statistics inputStats =
aggregate.getGroupExpression().get().childStatistics(0);
+ if (inputStats == null) {
+ return true; // no stats → allow (other gates handle eligibility)
+ }
+ Statistics outputStats = aggregate.getGroupExpression().get()
+ .getOwnerGroup().getStatistics();
+ SessionVariable sv = ConnectContext.get().getSessionVariable();
+ double rows = inputStats.getRowCount();
+
+ // Gate 1: minimum input rows
+ if (sv.bucketedAggMinInputRows > 0 && rows <
sv.bucketedAggMinInputRows) {
+ return false;
+ }
+
+ // Gate 2: high-cardinality GROUP BY columns
+ double highCardThreshold = sv.bucketedAggHighCardThreshold;
+ if (highCardThreshold > 0) {
+ for (Expression groupByKey : aggregate.getGroupByExpressions()) {
+ ColumnStatistic colStat =
inputStats.findColumnStatistics(groupByKey);
+ if (colStat != null && !colStat.isUnKnown
+ && colStat.ndv > rows * highCardThreshold) {
+ return false;
+ }
+ }
+ }
+
+ // Gate 3: max group keys (merge phase cost dominates)
+ if (sv.bucketedAggMaxGroupKeys > 0 && outputStats != null
+ && outputStats.getRowCount() > sv.bucketedAggMaxGroupKeys) {
+ return false;
+ }
+
+ // Gate 4: aggregation output cardinality ratio
+ if (highCardThreshold > 0 && outputStats != null
+ && outputStats.getRowCount() > rows * highCardThreshold) {
+ return false;
+ }
+
+ return true;
+ }
+
private boolean childIsCTEConsumer() {
List<GroupExpression> groupExpressions =
children.get(0).children().get(0).getPhysicalExpressions();
if (groupExpressions != null && !groupExpressions.isEmpty()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
index 8f6f236dd1b..2a94a6c971b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/RequestPropertyDeriver.java
@@ -39,7 +39,6 @@ import
org.apache.doris.nereids.trees.plans.algebra.SetOperation;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalSort;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
import org.apache.doris.nereids.trees.plans.physical.PhysicalBlackholeSink;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalConnectorTableSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDictionarySink;
@@ -505,14 +504,6 @@ public class RequestPropertyDeriver extends
PlanVisitor<Void, PlanContext> {
return null;
}
- @Override
- public Void visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> agg, PlanContext
context) {
- // Bucketed agg runs entirely on a single BE — no exchange needed.
- addRequestPropertyToChildren(PhysicalProperties.ANY);
- return null;
- }
-
private boolean shouldUseParent(List<ExprId> parentHashExprIds,
PhysicalHashAggregate<? extends Plan> agg,
PlanContext context) {
if
(!context.getConnectContext().getSessionVariable().aggShuffleUseParentKey) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/SplitAggWithoutDistinct.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/SplitAggWithoutDistinct.java
index fcadf6b3c70..324e1833f79 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/SplitAggWithoutDistinct.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/SplitAggWithoutDistinct.java
@@ -17,53 +17,32 @@
package org.apache.doris.nereids.rules.implementation;
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.DistributionInfo;
-import org.apache.doris.catalog.HashDistributionInfo;
-import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.nereids.memo.Group;
-import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.rules.Rule;
import org.apache.doris.nereids.rules.RuleType;
import org.apache.doris.nereids.trees.expressions.AggregateExpression;
import org.apache.doris.nereids.trees.expressions.Alias;
-import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.SessionVarGuardExpr;
-import org.apache.doris.nereids.trees.expressions.Slot;
-import org.apache.doris.nereids.trees.expressions.SlotReference;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregatePhase;
-import
org.apache.doris.nereids.trees.expressions.functions.agg.MultiDistinction;
import
org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionVisitor;
import org.apache.doris.nereids.trees.plans.AggMode;
import org.apache.doris.nereids.trees.plans.AggPhase;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
-import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
-import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
-import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
-import org.apache.doris.nereids.trees.plans.logical.LogicalTopN;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
import org.apache.doris.nereids.util.AggregateUtils;
import org.apache.doris.nereids.util.ExpressionUtils;
import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.statistics.ColumnStatistic;
-import org.apache.doris.statistics.Statistics;
-import org.apache.doris.system.Backend;
-import org.apache.doris.system.SystemInfoService;
import com.google.common.collect.ImmutableList;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.Set;
/**SplitAgg
* only process agg without distinct function, split Agg into 2 phase: local
agg and global agg
@@ -94,11 +73,6 @@ public class SplitAggWithoutDistinct extends
OneImplementationRuleFactory {
boolean singleExecutionInstance =
AggregateUtils.isSingleExecutionInstance(ctx);
if (!singleExecutionInstance || onePhaseCandidates.isEmpty()) {
candidates.addAll(splitTwoPhase(aggregate));
- // Only add bucketed agg candidate in auto mode (aggPhase
== 0).
- // When the user forces a specific phase, respect that
choice.
- if (!singleExecutionInstance) {
- candidates.addAll(implementBucketedPhase(aggregate,
ctx));
- }
}
break;
}
@@ -193,306 +167,6 @@ public class SplitAggWithoutDistinct extends
OneImplementationRuleFactory {
aggregate.getLogicalProperties(),
aggregate.getSourceRepeat().isPresent(), localAgg));
}
- /**
- * Implements bucketed hash aggregation for single-BE deployments.
- * Fuses two-phase aggregation into a single PhysicalBucketedHashAggregate
operator,
- * eliminating exchange overhead and serialization/deserialization costs.
- *
- * Only generated when:
- * 1. enable_bucketed_hash_agg session variable is true
- * 2. Cluster has exactly one alive BE
- * 3. Aggregate has GROUP BY keys (no without-key aggregation)
- * 4. Aggregate functions support two-phase execution
- * 5. Data volume checks pass (min input rows, max group keys)
- */
- private List<Plan> implementBucketedPhase(LogicalAggregate<? extends Plan>
aggregate, ConnectContext ctx) {
- if (!ctx.getSessionVariable().enableBucketedHashAgg) {
- return ImmutableList.of();
- }
- // Only for single-BE deployments
- int beNumber = ctx.getEnv().getClusterInfo().getBackendsNumber(true);
- if (beNumber != 1) {
- return ImmutableList.of();
- }
- // Skip during smooth upgrade: old BE processes do not recognize the
- // BUCKETED_AGGREGATION_NODE plan node type, so sending such plans
would
- // cause execution failures. Check all alive BEs for the upgrade flag.
- SystemInfoService clusterInfo = ctx.getEnv().getClusterInfo();
- for (Long beId : clusterInfo.getAllBackendByCurrentCluster(true)) {
- Backend be = clusterInfo.getBackend(beId);
- if (be != null && be.isSmoothUpgradeSrc()) {
- return ImmutableList.of();
- }
- }
- // Without-key aggregation not supported in initial version
- if (aggregate.getGroupByExpressions().isEmpty()) {
- return ImmutableList.of();
- }
- // Must support two-phase execution (same check as splitTwoPhase)
- if (!aggregate.supportAggregatePhase(AggregatePhase.TWO)) {
- return ImmutableList.of();
- }
- // Skip aggregates with no aggregate functions (pure GROUP BY dedup).
- // These are produced by DistinctAggregateRewriter as the bottom dedup
phase.
- if (aggregate.getAggregateFunctions().isEmpty()) {
- return ImmutableList.of();
- }
- // Skip aggregates containing multi-distinct functions (e.g.,
multi_distinct_count,
- // multi_distinct_sum). These are semantically distinct aggregations
rewritten by
- // DistinctAggregateRewriter — they embed deduplication in the
BE-level function.
- // The bucketed agg cost model does not account for deduplication
overhead, which
- // causes the base-table bucketed path to appear artificially cheap
compared to
- // materialized views using pre-aggregated bitmap_union/hll_union.
- for (AggregateFunction func : aggregate.getAggregateFunctions()) {
- if (func instanceof MultiDistinction) {
- return ImmutableList.of();
- }
- }
- // Skip aggregates whose child group contains a LogicalAggregate,
indicating
- // a multi-phase decomposition (e.g., COUNT(DISTINCT a) GROUP BY b is
rewritten
- // to COUNT(a) GROUP BY b on top of GROUP BY a,b dedup). These stacked
- // aggregates require cross-phase coordination that bucketed agg does
not support.
- if (childGroupContainsAggregate(aggregate)) {
- return ImmutableList.of();
- }
- // Skip when sortByGroupKey optimization applies. This is detected by
- // checking if the aggregate's owner group has a LogicalTopN parent
- // whose order key expressions equal the group-by keys (produced by
- // LimitAggToTopNAgg rewrite). PhysicalBucketedHashAggregate does not
- // support sortByGroupKey, so we yield to the regular hash-agg plan.
- if (hasSortByGroupKeyTopN(aggregate)) {
- return ImmutableList.of();
- }
- // Skip when data is already distributed by the GROUP BY keys
- // (e.g., table bucketed by UserID, query GROUP BY UserID).
- // In this case the two-phase plan needs no exchange and is strictly
- // better than bucketed agg (no 256-bucket overhead, no merge phase).
- if (groupByKeysSatisfyDistribution(aggregate)) {
- return ImmutableList.of();
- }
- // Data-volume-based checks: control bucketed agg eligibility based on
- // estimated data scale, similar to ClickHouse's
group_by_two_level_threshold
- // and group_by_two_level_threshold_bytes. This reduces reliance on
- // column-level statistics which may be inaccurate or missing.
- //
- // When statistics are unavailable (groupExpression absent or
childStats null),
- // conservatively skip bucketed agg — without data volume information
we cannot
- // make an informed decision, and the risk of choosing bucketed agg in
a
- // high-cardinality scenario outweighs the potential benefit.
- if (!aggregate.getGroupExpression().isPresent()) {
- return ImmutableList.of();
- }
- GroupExpression ge = aggregate.getGroupExpression().get();
- Statistics childStats = ge.childStatistics(0);
- if (childStats == null) {
- return ImmutableList.of();
- }
- double rows = childStats.getRowCount();
- long minInputRows = ctx.getSessionVariable().bucketedAggMinInputRows;
- long maxGroupKeys = ctx.getSessionVariable().bucketedAggMaxGroupKeys;
-
- // Gate: minimum input rows.
- // When input data is too small, the overhead of initializing 256
- // per-bucket hash tables and the pipelined merge phase outweighs
- // the benefit of eliminating exchange. Skip bucketed agg.
- if (minInputRows > 0 && rows < minInputRows) {
- return ImmutableList.of();
- }
-
- // Gate: maximum estimated group keys (similar to ClickHouse's
- // group_by_two_level_threshold). When the number of distinct groups
- // is too large, the source-side merge must combine too many keys
- // across instances, and the merge cost dominates. Skip bucketed agg.
- Statistics aggStats = ge.getOwnerGroup().getStatistics();
- if (maxGroupKeys > 0 && aggStats != null && aggStats.getRowCount() >
maxGroupKeys) {
- return ImmutableList.of();
- }
-
- // High-cardinality ratio checks (existing logic).
- // These complement the absolute thresholds above with relative checks:
- // 1. Single-column NDV check: if ANY GROUP BY key's NDV > rows *
threshold,
- // the combined NDV is at least that high.
- // 2. Aggregation ratio check: if estimated output rows > rows *
threshold,
- // merge cost dominates.
- double highCardThreshold =
ctx.getSessionVariable().bucketedAggHighCardThreshold;
- for (Expression groupByKey : aggregate.getGroupByExpressions()) {
- ColumnStatistic colStat =
childStats.findColumnStatistics(groupByKey);
- if (colStat != null && !colStat.isUnKnown() && colStat.ndv > rows
* highCardThreshold) {
- return ImmutableList.of();
- }
- }
- if (aggStats != null && aggStats.getRowCount() > rows *
highCardThreshold) {
- return ImmutableList.of();
- }
- // Build output expressions: rewrite AggregateFunction ->
AggregateExpression with GLOBAL_RESULT param
- // (same as one-phase aggregation — raw input directly produces final
result).
- List<NamedExpression> aggOutput =
ExpressionUtils.rewriteDownShortCircuit(
- aggregate.getOutputExpressions(), expr -> {
- if (!(expr instanceof AggregateFunction)) {
- return expr;
- }
- return new AggregateExpression((AggregateFunction) expr,
AggregateParam.GLOBAL_RESULT);
- }
- );
- return ImmutableList.of(new PhysicalBucketedHashAggregate<>(
- aggregate.getGroupByExpressions(), aggOutput,
- aggregate.getLogicalProperties(), aggregate.child()));
- }
-
- /**
- * Check if the child group of this aggregate contains a LogicalAggregate.
- * This is used to detect aggregates produced by DISTINCT decomposition
rewrites
- * (e.g., DistinctAggregateRewriter, SplitMultiDistinctStrategy), where
the original
- * DISTINCT aggregate is split into a top non-distinct aggregate over a
bottom dedup aggregate.
- */
- private boolean childGroupContainsAggregate(LogicalAggregate<? extends
Plan> aggregate) {
- if (!aggregate.getGroupExpression().isPresent()) {
- return false;
- }
- GroupExpression groupExpr = aggregate.getGroupExpression().get();
- if (groupExpr.arity() == 0) {
- return false;
- }
- Group childGroup = groupExpr.child(0);
- for (GroupExpression childGroupExpr :
childGroup.getLogicalExpressions()) {
- if (childGroupExpr.getPlan() instanceof LogicalAggregate) {
- return true;
- }
- }
- return false;
- }
-
- /**
- * Check if a LogicalTopN parent exists whose order keys are identical to
- * the aggregate's group-by keys. This means PushTopnToAgg will later set
- * sortByGroupKey on PhysicalHashAggregate; bucketed agg doesn't support
- * that optimization so we skip it.
- *
- * Handles both TopN->Agg and TopN->Project->Agg patterns.
- */
- private boolean hasSortByGroupKeyTopN(LogicalAggregate<? extends Plan>
aggregate) {
- if (!aggregate.getGroupExpression().isPresent()) {
- return false;
- }
- List<Expression> groupByKeys = aggregate.getGroupByExpressions();
- Group ownerGroup =
aggregate.getGroupExpression().get().getOwnerGroup();
- for (GroupExpression parentGE :
ownerGroup.getParentGroupExpressions()) {
- Plan parentPlan = parentGE.getPlan();
- if (parentPlan instanceof LogicalTopN
- && AggregateUtils.isOrderKeysMatchGroupKeys(
- ((LogicalTopN<?>) parentPlan).getOrderKeys(),
groupByKeys)) {
- return true;
- }
- if (parentPlan instanceof LogicalProject &&
parentGE.getOwnerGroup() != null) {
- for (GroupExpression gpGE :
parentGE.getOwnerGroup().getParentGroupExpressions()) {
- if (gpGE.getPlan() instanceof LogicalTopN
- && AggregateUtils.isOrderKeysMatchGroupKeys(
- ((LogicalTopN<?>)
gpGE.getPlan()).getOrderKeys(), groupByKeys)) {
- return true;
- }
- }
- }
- }
- return false;
- }
-
- /**
- * Check if the GROUP BY keys of this aggregate are a superset of (or
equal to)
- * the underlying OlapTable's hash distribution columns. When this is true,
- * the data is already correctly partitioned for the aggregation, so the
- * two-phase plan (local + global) requires no exchange and is strictly
better
- * than bucketed agg (no 256-bucket overhead, no merge phase).
- *
- * Traverses the child group in the Memo to find a LogicalOlapScan,
- * then uses ExprId-based matching (consistent with
- * {@link LogicalOlapScanToPhysicalOlapScan#convertDistribution}) to check
- * whether the hash distribution columns are a subset of the GROUP BY keys.
- * For non-OlapTable children, returns false (skip this gating, allow
bucketed agg).
- */
- private boolean groupByKeysSatisfyDistribution(LogicalAggregate<? extends
Plan> aggregate) {
- if (!aggregate.getGroupExpression().isPresent()) {
- return false;
- }
- GroupExpression groupExpr = aggregate.getGroupExpression().get();
- if (groupExpr.arity() == 0) {
- return false;
- }
- LogicalOlapScan olapScan =
findLogicalOlapScanInGroup(groupExpr.child(0), 5);
- if (olapScan == null) {
- return false;
- }
- OlapTable table = olapScan.getTable();
- DistributionInfo distributionInfo = table.getDefaultDistributionInfo();
- if (!(distributionInfo instanceof HashDistributionInfo)) {
- return false;
- }
- List<Column> distributionColumns = ((HashDistributionInfo)
distributionInfo).getDistributionColumns();
- if (distributionColumns.isEmpty()) {
- return false;
- }
- // Map distribution columns to ExprIds via the scan's output slots
- List<Slot> output = olapScan.getOutput();
- Set<ExprId> distributionExprIds = new HashSet<>();
- for (Column column : distributionColumns) {
- boolean found = false;
- for (Slot slot : output) {
- if (slot instanceof SlotReference
- && ((SlotReference)
slot).getOriginalColumn().isPresent()
- && ((SlotReference)
slot).getOriginalColumn().get().getName()
- .equalsIgnoreCase(column.getName())) {
- distributionExprIds.add(slot.getExprId());
- found = true;
- break;
- }
- }
- if (!found) {
- return false;
- }
- }
- // Collect GROUP BY ExprIds
- Set<ExprId> groupByExprIds = new HashSet<>();
- for (Expression expr : aggregate.getGroupByExpressions()) {
- if (expr instanceof SlotReference) {
- groupByExprIds.add(((SlotReference) expr).getExprId());
- }
- }
- return groupByExprIds.containsAll(distributionExprIds);
- }
-
- /**
- * Recursively search through a Memo Group to find a LogicalOlapScan,
- * walking through LogicalProject and LogicalFilter nodes.
- * Returns the LogicalOlapScan if found, null otherwise.
- * maxDepth prevents infinite recursion.
- */
- private LogicalOlapScan findLogicalOlapScanInGroup(Group group, int
maxDepth) {
- if (maxDepth <= 0) {
- return null;
- }
- for (GroupExpression ge : group.getLogicalExpressions()) {
- Plan plan = ge.getPlan();
- if (plan instanceof LogicalOlapScan) {
- return (LogicalOlapScan) plan;
- }
- if ((plan instanceof LogicalProject || plan instanceof
LogicalFilter) && ge.arity() > 0) {
- LogicalOlapScan result =
findLogicalOlapScanInGroup(ge.child(0), maxDepth - 1);
- if (result != null) {
- return result;
- }
- }
- }
- return null;
- }
-
- private boolean shouldUseLocalAgg(LogicalAggregate<? extends Plan>
aggregate) {
- Statistics aggStats =
aggregate.getGroupExpression().get().getOwnerGroup().getStatistics();
- Statistics aggChildStats =
aggregate.getGroupExpression().get().childStatistics(0);
- // if gbyNdv is high, should not use local agg
- double rows = aggChildStats.getRowCount();
- double gbyNdv = aggStats.getRowCount();
- return gbyNdv * 10 < rows;
- }
-
private boolean skipRegulator(LogicalAggregate<? extends Plan> aggregate) {
for (AggregateFunction aggregateFunction :
aggregate.getAggregateFunctions()) {
if (aggregateFunction.forceSkipRegulator(AggregatePhase.ONE)) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
index ed7bd33fbd6..933da2f32d6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
@@ -100,7 +100,6 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.trees.plans.logical.LogicalWorkTableReference;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEConsumer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
@@ -999,12 +998,6 @@ public class StatsCalculator extends
DefaultPlanVisitor<Statistics, Void> {
return computeAggregate(agg, groupExpression.childStatistics(0));
}
- @Override
- public Statistics visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> agg, Void context) {
- return computeAggregate(agg, groupExpression.childStatistics(0));
- }
-
@Override
public Statistics visitPhysicalRepeat(PhysicalRepeat<? extends Plan>
repeat, Void context) {
return computeRepeat(repeat, groupExpression.childStatistics(0));
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java
index c73396d1271..27137ca88cd 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/PlanType.java
@@ -132,7 +132,6 @@ public enum PlanType {
// physical others
PHYSICAL_HASH_AGGREGATE,
- PHYSICAL_BUCKETED_HASH_AGGREGATE,
PHYSICAL_ASSERT_NUM_ROWS,
PHYSICAL_CTE_PRODUCER,
PHYSICAL_CTE_ANCHOR,
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalBucketedHashAggregate.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalBucketedHashAggregate.java
deleted file mode 100644
index 09565bdc820..00000000000
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalBucketedHashAggregate.java
+++ /dev/null
@@ -1,293 +0,0 @@
-// 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.trees.plans.physical;
-
-import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.properties.DataTrait;
-import org.apache.doris.nereids.properties.LogicalProperties;
-import org.apache.doris.nereids.properties.PhysicalProperties;
-import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.NamedExpression;
-import org.apache.doris.nereids.trees.expressions.Slot;
-import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
-import org.apache.doris.nereids.trees.expressions.functions.agg.Ndv;
-import org.apache.doris.nereids.trees.plans.Plan;
-import org.apache.doris.nereids.trees.plans.PlanType;
-import org.apache.doris.nereids.trees.plans.algebra.Aggregate;
-import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
-import org.apache.doris.nereids.util.ExpressionUtils;
-import org.apache.doris.nereids.util.Utils;
-import org.apache.doris.statistics.Statistics;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
-
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-
-/**
- * Physical bucketed hash aggregation plan node.
- *
- * Fuses two-phase aggregation (local + global) into a single operator for
single-BE deployments.
- * The sink side builds per-instance hash tables from raw input (first-phase
agg).
- * The source side merges across instances per-bucket using direct in-memory
merge
- * (no serialization/deserialization) and outputs the final result.
- *
- * This node replaces the pattern: GlobalAgg -> PhysicalDistribute -> LocalAgg
- * with a single fused operator, eliminating exchange overhead entirely.
- */
-public class PhysicalBucketedHashAggregate<CHILD_TYPE extends Plan> extends
PhysicalUnary<CHILD_TYPE>
- implements Aggregate<CHILD_TYPE> {
-
- private final List<Expression> groupByExpressions;
- private final List<NamedExpression> outputExpressions;
-
- public PhysicalBucketedHashAggregate(List<Expression> groupByExpressions,
- List<NamedExpression> outputExpressions,
- LogicalProperties logicalProperties, CHILD_TYPE child) {
- this(groupByExpressions, outputExpressions, Optional.empty(),
logicalProperties, child);
- }
-
- public PhysicalBucketedHashAggregate(List<Expression> groupByExpressions,
- List<NamedExpression> outputExpressions,
- Optional<GroupExpression> groupExpression,
- LogicalProperties logicalProperties, CHILD_TYPE child) {
- super(PlanType.PHYSICAL_BUCKETED_HASH_AGGREGATE, groupExpression,
logicalProperties, child);
- this.groupByExpressions = ImmutableList.copyOf(
- Objects.requireNonNull(groupByExpressions, "groupByExpressions
cannot be null"));
- this.outputExpressions = ImmutableList.copyOf(
- Objects.requireNonNull(outputExpressions, "outputExpressions
cannot be null"));
- }
-
- /**
- * Constructor with group expression, logical properties, physical
properties, and statistics.
- */
- public PhysicalBucketedHashAggregate(List<Expression> groupByExpressions,
- List<NamedExpression> outputExpressions,
- Optional<GroupExpression> groupExpression,
- LogicalProperties logicalProperties,
- PhysicalProperties physicalProperties, Statistics statistics,
CHILD_TYPE child) {
- super(PlanType.PHYSICAL_BUCKETED_HASH_AGGREGATE, groupExpression,
logicalProperties,
- physicalProperties, statistics, child);
- this.groupByExpressions = ImmutableList.copyOf(
- Objects.requireNonNull(groupByExpressions, "groupByExpressions
cannot be null"));
- this.outputExpressions = ImmutableList.copyOf(
- Objects.requireNonNull(outputExpressions, "outputExpressions
cannot be null"));
- }
-
- @Override
- public List<Expression> getGroupByExpressions() {
- return groupByExpressions;
- }
-
- @Override
- public List<NamedExpression> getOutputExpressions() {
- return outputExpressions;
- }
-
- @Override
- public List<NamedExpression> getOutputs() {
- return outputExpressions;
- }
-
- @Override
- public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
- return visitor.visitPhysicalBucketedHashAggregate(this, context);
- }
-
- @Override
- public List<? extends Expression> getExpressions() {
- return new ImmutableList.Builder<Expression>()
- .addAll(groupByExpressions)
- .addAll(outputExpressions)
- .build();
- }
-
- @Override
- public String toString() {
- return Utils.toSqlString("PhysicalBucketedHashAggregate[" + id.asInt()
+ "]" + getGroupIdWithPrefix(),
- "stats", statistics,
- "groupByExpr", groupByExpressions,
- "outputExpr", outputExpressions
- );
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- PhysicalBucketedHashAggregate<?> that =
(PhysicalBucketedHashAggregate<?>) o;
- return Objects.equals(groupByExpressions, that.groupByExpressions)
- && Objects.equals(outputExpressions, that.outputExpressions);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(groupByExpressions, outputExpressions);
- }
-
- @Override
- public PhysicalBucketedHashAggregate<Plan> withChildren(List<Plan>
children) {
- Preconditions.checkArgument(children.size() == 1);
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
outputExpressions,
- groupExpression, getLogicalProperties(),
- physicalProperties, statistics, children.get(0));
- }
-
- @Override
- public PhysicalBucketedHashAggregate<CHILD_TYPE> withGroupExpression(
- Optional<GroupExpression> groupExpression) {
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
outputExpressions,
- groupExpression, getLogicalProperties(), child());
- }
-
- @Override
- public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression>
groupExpression,
- Optional<LogicalProperties> logicalProperties, List<Plan>
children) {
- Preconditions.checkArgument(children.size() == 1);
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
outputExpressions,
- groupExpression, logicalProperties.get(), children.get(0));
- }
-
- @Override
- public PhysicalBucketedHashAggregate<CHILD_TYPE>
withPhysicalPropertiesAndStats(
- PhysicalProperties physicalProperties, Statistics statistics) {
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
outputExpressions,
- groupExpression, getLogicalProperties(),
- physicalProperties, statistics, child());
- }
-
- @Override
- public PhysicalBucketedHashAggregate<CHILD_TYPE>
withAggOutput(List<NamedExpression> newOutput) {
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
newOutput,
- Optional.empty(), getLogicalProperties(),
- physicalProperties, statistics, child());
- }
-
- @Override
- public String shapeInfo() {
- return "bucketedHashAgg";
- }
-
- @Override
- public List<Slot> computeOutput() {
- return outputExpressions.stream()
- .map(NamedExpression::toSlot)
- .collect(ImmutableList.toImmutableList());
- }
-
- @Override
- public PhysicalBucketedHashAggregate<CHILD_TYPE> resetLogicalProperties() {
- return new PhysicalBucketedHashAggregate<>(groupByExpressions,
outputExpressions,
- groupExpression, null,
- physicalProperties, statistics, child());
- }
-
- @Override
- public void computeUnique(DataTrait.Builder builder) {
- DataTrait childFd = child(0).getLogicalProperties().getTrait();
-
- if
(groupByExpressions.stream().anyMatch(Expression::containsVolatileExpression)) {
- return;
- }
-
- ImmutableSet.Builder<Slot> groupByKeysBuilder = ImmutableSet.builder();
- for (Expression expr : groupByExpressions) {
- groupByKeysBuilder.addAll(expr.getInputSlots());
- }
- ImmutableSet<Slot> groupByKeys = groupByKeysBuilder.build();
-
- if (groupByExpressions.isEmpty() ||
childFd.isUniformAndNotNull(groupByKeys)) {
- getOutput().forEach(builder::addUniqueSlot);
- return;
- }
-
- builder.addUniqueSlot(childFd);
- builder.addUniqueSlot(groupByKeys);
-
- if (childFd.isUniqueAndNotNull(groupByKeys)) {
- for (NamedExpression namedExpression : getOutputExpressions()) {
- if (isUniqueGroupByUnique(namedExpression)) {
- builder.addUniqueSlot(namedExpression.toSlot());
- }
- }
- }
- }
-
- @Override
- public void computeUniform(DataTrait.Builder builder) {
- DataTrait childFd = child(0).getLogicalProperties().getTrait();
- builder.addUniformSlot(childFd);
-
- if
(groupByExpressions.stream().anyMatch(Expression::containsVolatileExpression)) {
- return;
- }
-
- ImmutableSet.Builder<Slot> groupByKeysBuilder = ImmutableSet.builder();
- for (Expression expr : groupByExpressions) {
- groupByKeysBuilder.addAll(expr.getInputSlots());
- }
- ImmutableSet<Slot> groupByKeys = groupByKeysBuilder.build();
-
- if (groupByExpressions.isEmpty() ||
childFd.isUniformAndNotNull(groupByKeys)) {
- getOutput().forEach(builder::addUniformSlot);
- return;
- }
-
- if (childFd.isUniqueAndNotNull(groupByKeys)) {
- for (NamedExpression namedExpression : getOutputExpressions()) {
- if (isUniformGroupByUnique(namedExpression)) {
- builder.addUniformSlot(namedExpression.toSlot());
- }
- }
- }
- }
-
- private boolean isUniqueGroupByUnique(NamedExpression namedExpression) {
- if (namedExpression.children().size() != 1) {
- return false;
- }
- Expression agg = namedExpression.child(0);
- return ExpressionUtils.isInjectiveAgg(agg)
- &&
child().getLogicalProperties().getTrait().isUniqueAndNotNull(agg.getInputSlots());
- }
-
- private boolean isUniformGroupByUnique(NamedExpression namedExpression) {
- if (namedExpression.children().size() != 1) {
- return false;
- }
- Expression agg = namedExpression.child(0);
- return agg instanceof Count || agg instanceof Ndv;
- }
-
- @Override
- public void computeEqualSet(DataTrait.Builder builder) {
- builder.addEqualSet(child().getLogicalProperties().getTrait());
- }
-
- @Override
- public void computeFd(DataTrait.Builder builder) {
- builder.addFuncDepsDG(child().getLogicalProperties().getTrait());
- }
-}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/PlanVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/PlanVisitor.java
index a68393a7f74..55b677923c5 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/PlanVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/visitor/PlanVisitor.java
@@ -62,7 +62,6 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalJoin;
import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalSort;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEAnchor;
import org.apache.doris.nereids.trees.plans.physical.PhysicalCTEProducer;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
@@ -327,11 +326,6 @@ public abstract class PlanVisitor<R, C> implements
CommandVisitor<R, C>, Relatio
return visit(agg, context);
}
- public R visitPhysicalBucketedHashAggregate(
- PhysicalBucketedHashAggregate<? extends Plan> agg, C context) {
- return visit(agg, context);
- }
-
public R visitPhysicalStorageLayerAggregate(PhysicalStorageLayerAggregate
storageLayerAggregate, C context) {
return storageLayerAggregate.getRelation().accept(this, context);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
index d26818864b9..d68529fcfeb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
@@ -35,6 +35,8 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.statistics.ColumnStatistic;
import org.apache.doris.statistics.Statistics;
+import org.apache.doris.system.Backend;
+import org.apache.doris.system.SystemInfoService;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
@@ -222,4 +224,47 @@ public class AggregateUtils {
}
return true;
}
+
+ /**
+ * Check the basic environmental conditions for bucketed hash aggregation.
+ * This is the shared eligibility gate used by ChildrenPropertiesRegulator
+ * (to allow the one-phase-GLOBAL+distribute pattern), CostModel (for cost
+ * discount), and PhysicalPlanTranslator (for fusion into
BucketedAggregationNode).
+ *
+ * @return true if the session variable is enabled, there is exactly one
alive BE,
+ * no smooth upgrade is in progress, and the aggregate has GROUP
BY keys.
+ */
+ public static boolean isBucketedHashAggEnabled(int groupByExprCount) {
+ ConnectContext ctx = ConnectContext.get();
+ if (ctx == null) {
+ return false;
+ }
+ if (!ctx.getSessionVariable().enableBucketedHashAgg) {
+ return false;
+ }
+ // Must have GROUP BY keys (without-key aggregation not supported)
+ if (groupByExprCount == 0) {
+ return false;
+ }
+ // Correctness gate: single-BE only (cross-BE in-memory merge is
impossible).
+ // Use be_number_for_test first (set by regression tests), fall back
to real cluster count.
+ // Note: do not clamp to 1 — with zero backends bucketed agg must not
be enabled.
+ int beNumber = ctx.getSessionVariable().getBeNumberForTest();
+ if (beNumber <= 0) {
+ beNumber = ctx.getEnv().getClusterInfo().getBackendsNumber(true);
+ }
+ if (beNumber != 1) {
+ return false;
+ }
+ // Smooth upgrade safety net: old BE processes do not recognize
+ // BUCKETED_AGGREGATION_NODE plan node type
+ SystemInfoService clusterInfo = ctx.getEnv().getClusterInfo();
+ for (Long beId : clusterInfo.getAllBackendByCurrentCluster(true)) {
+ Backend be = clusterInfo.getBackend(beId);
+ if (be != null && be.isSmoothUpgradeSrc()) {
+ return false;
+ }
+ }
+ return true;
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
index 36322ed9b99..bffe00a0496 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/properties/RequestPropertyDeriverTest.java
@@ -49,7 +49,6 @@ import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.RelationId;
import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalAssertNumRows;
-import
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate;
import
org.apache.doris.nereids.trees.plans.physical.PhysicalExternalRowLevelMergeSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
@@ -493,23 +492,4 @@ class RequestPropertyDeriverTest {
Lists.newArrayList(key1.getExprId(), key2.getExprId()),
ShuffleType.REQUIRE);
Assertions.assertTrue(actual.contains(ImmutableList.of(aggProp)) &&
actual.contains(ImmutableList.of(parentProp)));
}
-
- @Test
- void testBucketedHashAggregate() {
- SlotReference key = new SlotReference("col1", IntegerType.INSTANCE);
- PhysicalBucketedHashAggregate<GroupPlan> aggregate = new
PhysicalBucketedHashAggregate<>(
- Lists.newArrayList(key),
- Lists.newArrayList(key),
- logicalProperties,
- groupPlan
- );
- GroupExpression groupExpression = new GroupExpression(aggregate);
- new Group(null, groupExpression, null);
- RequestPropertyDeriver requestPropertyDeriver = new
RequestPropertyDeriver(null, jobContext);
- List<List<PhysicalProperties>> actual
- =
requestPropertyDeriver.getRequestChildrenPropertyList(groupExpression);
- List<List<PhysicalProperties>> expected = Lists.newArrayList();
- expected.add(Lists.newArrayList(PhysicalProperties.ANY));
- Assertions.assertEquals(expected, actual);
- }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/BucketedAggregateTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/BucketedAggregateTest.java
index 91b0fc4688e..ccef8b42abf 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/BucketedAggregateTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/implementation/BucketedAggregateTest.java
@@ -25,6 +25,7 @@ import
org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.plans.AggMode;
import org.apache.doris.nereids.trees.plans.AggPhase;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
@@ -44,6 +45,13 @@ import org.junit.jupiter.api.TestInstance;
import java.util.List;
import java.util.Optional;
+/**
+ * Tests for bucketed hash aggregation. Since the bucketed fusion now happens
+ * at the translator level (PhysicalPlanTranslator fuses one-phase GLOBAL
+ * PhysicalHashAggregate + PhysicalDistribute into BucketedAggregationNode),
+ * these tests verify that the optimizer generates the correct one-phase
+ * candidates that the translator will later fuse.
+ */
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class BucketedAggregateTest implements MemoPatternMatchSupported {
private Plan rStudent;
@@ -84,42 +92,69 @@ public class BucketedAggregateTest implements
MemoPatternMatchSupported {
@Test
public void testBucketedAggDisabled() {
+ // When bucketed is disabled, the one-phase GLOBAL(INPUT_TO_RESULT)
candidate is
+ // still generated by the rule (the translator just won't fuse it).
+ // Note: PhysicalDistribute is inserted by the enforcer during full
optimization,
+ // not by the implementation rule itself. This test verifies
rule-level output.
Plan root = buildAggregateWithGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = false;
ctx.getSessionVariable().setBeNumberForTest(1);
PlanChecker.from(ctx, root)
+ .deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
- .nonMatch(physicalBucketedHashAggregate());
+ // one-phase GLOBAL INPUT_TO_RESULT candidate exists
+ .matches(physicalHashAggregate()
+ .when(agg -> agg.getAggPhase().equals(AggPhase.GLOBAL)
+ &&
agg.getAggMode().equals(AggMode.INPUT_TO_RESULT)));
}
@Test
public void testBucketedAggMultiBE() {
+ // On multi-BE, both one-phase and two-phase are generated at rule
level.
+ // The cost model will prefer two-phase (no bucketed discount for
beNumber>1).
Plan root = buildAggregateWithGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = true;
ctx.getSessionVariable().setBeNumberForTest(3);
PlanChecker.from(ctx, root)
+ .deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
- .nonMatch(physicalBucketedHashAggregate());
+ // one-phase GLOBAL INPUT_TO_RESULT candidate exists
+ .matches(physicalHashAggregate()
+ .when(agg -> agg.getAggPhase().equals(AggPhase.GLOBAL)
+ &&
agg.getAggMode().equals(AggMode.INPUT_TO_RESULT)))
+ // two-phase also exists
+ .matches(physicalHashAggregate(physicalHashAggregate()
+ .when(child ->
child.getAggPhase().equals(AggPhase.LOCAL))));
}
@Test
public void testBucketedAggNoGroupBy() {
+ // Without GROUP BY, scalar aggregation is generated (GATHER, no
distribute).
+ // Bucketed fusion only applies to grouped aggregation.
Plan root = buildAggregateWithoutGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = true;
ctx.getSessionVariable().setBeNumberForTest(1);
PlanChecker.from(ctx, root)
+ .deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
- .nonMatch(physicalBucketedHashAggregate());
+ // scalar agg is GLOBAL GATHER, no distribute child
+ .matches(physicalHashAggregate()
+ .when(agg -> agg.getAggPhase().equals(AggPhase.GLOBAL)
+ && agg.getGroupByExpressions().isEmpty()));
}
@Test
public void testBucketedAggEnabled() {
+ // With all bucketed conditions met, verify the one-phase
GLOBAL(INPUT_TO_RESULT)
+ // candidate is generated. The enforcer will later insert
PhysicalDistribute,
+ // and the translator will fuse them into BucketedAggregationNode.
+ // Key: distinguish from two-phase merge which is
GLOBAL(BUFFER_TO_RESULT).
Plan root = buildAggregateWithGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = true;
@@ -132,11 +167,19 @@ public class BucketedAggregateTest implements
MemoPatternMatchSupported {
PlanChecker.from(ctx, root)
.deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
- .matches(physicalBucketedHashAggregate());
+ // one-phase GLOBAL(INPUT_TO_RESULT) exists
+ .matches(physicalHashAggregate()
+ .when(agg -> agg.getAggPhase().equals(AggPhase.GLOBAL)
+ &&
agg.getAggMode().equals(AggMode.INPUT_TO_RESULT)))
+ // two-phase (LOCAL -> GLOBAL) also exists
+ .matches(physicalHashAggregate(physicalHashAggregate()
+ .when(child ->
child.getAggPhase().equals(AggPhase.LOCAL))));
}
@Test
public void testSingleExecutionInstanceOnlyGeneratesOnePhaseAgg() {
+ // When parallelPipelineTaskNum=1, only one-phase is generated
+ // (no two-phase because there's nowhere to distribute to).
Plan root = buildAggregateWithGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = true;
@@ -149,16 +192,18 @@ public class BucketedAggregateTest implements
MemoPatternMatchSupported {
PlanChecker.from(ctx, root)
.deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
+ // one-phase GLOBAL exists
.matches(physicalHashAggregate()
.when(agg ->
agg.getAggPhase().equals(AggPhase.GLOBAL)))
- .nonMatch(physicalHashAggregate(physicalHashAggregate()))
- .nonMatch(physicalBucketedHashAggregate());
+ // two-phase (nested agg) must NOT exist
+ .nonMatch(physicalHashAggregate(physicalHashAggregate()));
}
@Test
public void testBucketedAggForcedAggPhase() {
- // When user forces agg_phase = 2, bucketed agg should NOT be generated
- // because it is only added in auto mode (aggPhase == 0).
+ // When user forces agg_phase = 2, only two-phase is generated.
+ // The one-phase GLOBAL(INPUT_TO_RESULT) candidate is not generated,
+ // so there's no pattern for the translator to fuse.
Plan root = buildAggregateWithGroupBy();
ConnectContext ctx = MemoTestUtils.createConnectContext();
ctx.getSessionVariable().enableBucketedHashAgg = true;
@@ -171,6 +216,12 @@ public class BucketedAggregateTest implements
MemoPatternMatchSupported {
PlanChecker.from(ctx, root)
.deriveStats()
.applyImplementation(splitAggWithoutDistinctRule())
- .nonMatch(physicalBucketedHashAggregate());
+ // two-phase pattern must exist
+ .matches(physicalHashAggregate(physicalHashAggregate()
+ .when(child ->
child.getAggPhase().equals(AggPhase.LOCAL))))
+ // one-phase GLOBAL INPUT_TO_RESULT must NOT exist
+ .nonMatch(physicalHashAggregate()
+ .when(agg -> agg.getAggPhase().equals(AggPhase.GLOBAL)
+ &&
agg.getAggMode().equals(AggMode.INPUT_TO_RESULT)));
}
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/statistics/query/QueryStatsRecorderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/statistics/query/QueryStatsRecorderTest.java
index 2dbf6da6dd2..81b2aef710c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/statistics/query/QueryStatsRecorderTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/statistics/query/QueryStatsRecorderTest.java
@@ -820,34 +820,6 @@ public class QueryStatsRecorderTest {
Assertions.assertTrue(delta.getColumnStats().get("k1").queryHit, "k1:
window ORDER BY key");
}
- /**
- * PhysicalBucketedHashAggregate implements Aggregate but does not extend
PhysicalHashAggregate.
- * The Aggregate interface check must cover it.
- */
- @Test
- @SuppressWarnings("unchecked")
- public void testBucketedAggregateGroupByQueryHit() {
- ExprId id1 = new ExprId(1);
- SlotReference k1Slot = mockSlot(id1, "k1");
- PhysicalOlapScan scan = mockScan(1L, 1L, 1L, 1L,
ImmutableList.of(k1Slot));
-
- Expression groupExpr = Mockito.mock(Expression.class);
-
Mockito.when(groupExpr.getInputSlots()).thenReturn(ImmutableSet.of(k1Slot));
-
-
org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate<?>
agg =
-
Mockito.mock(org.apache.doris.nereids.trees.plans.physical.PhysicalBucketedHashAggregate.class);
- Mockito.when(agg.children()).thenReturn(ImmutableList.of(scan));
-
Mockito.when(agg.getGroupByExpressions()).thenReturn(ImmutableList.of(groupExpr));
-
Mockito.when(agg.getOutputExpressions()).thenReturn(ImmutableList.of());
- Mockito.when(agg.getOutput()).thenReturn(ImmutableList.of(k1Slot));
-
- Map<String, StatsDelta> deltas =
QueryStatsRecorder.collectDeltas((PhysicalPlan) agg);
-
- StatsDelta delta = deltas.get("1_1_1_1");
- Assertions.assertNotNull(delta, "bucketed aggregate must be recorded
via Aggregate interface");
- Assertions.assertTrue(delta.getColumnStats().get("k1").queryHit, "k1:
GROUP BY in bucketed agg");
- }
-
/**
* JOIN ON t1.k1 = t2.k2: both hash-join and other-join conjuncts →
filterHit.
*/
diff --git
a/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg.out
b/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg.out
new file mode 100644
index 00000000000..b2507b7b088
--- /dev/null
+++ b/regression-test/data/nereids_rules_p0/agg_strategy/bucketed_hash_agg.out
@@ -0,0 +1,40 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !bucketed_shape --
+PhysicalResultSink
+--PhysicalDistribute[DistributionSpecGather]
+----hashAgg[GLOBAL]
+------PhysicalDistribute[DistributionSpecHash]
+--------PhysicalProject
+----------PhysicalOlapScan[bucketed_agg_reg_test]
+
+-- !bucketed_result --
+a 90
+b 240
+c 220
+
+-- !multi_be_shape --
+PhysicalResultSink
+--PhysicalDistribute[DistributionSpecGather]
+----hashAgg[GLOBAL]
+------PhysicalDistribute[DistributionSpecHash]
+--------hashAgg[LOCAL]
+----------PhysicalProject
+------------PhysicalOlapScan[bucketed_agg_reg_test]
+
+-- !multi_be_result --
+a 90
+b 240
+c 220
+
+-- !disabled_result --
+a 90
+b 240
+c 220
+
+-- !no_group_by_result --
+550
+
+-- !count_distinct_result --
+a 4 200
+b 5 370
+c 4 340
diff --git
a/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg.groovy
b/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg.groovy
new file mode 100644
index 00000000000..fda888d84d5
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/agg_strategy/bucketed_hash_agg.groovy
@@ -0,0 +1,162 @@
+// 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("bucketed_hash_agg") {
+ // ============================================================
+ // Test: Bucketed Hash Aggregation regression
+ //
+ // Verifies that on single-BE deployments with
enable_bucketed_hash_agg=true,
+ // the translator fuses one-phase GLOBAL hash aggregate + distribute into
+ // a single BUCKETED AGGREGATE operator, eliminating exchange overhead.
+ // On multi-BE deployments, bucketed agg must NOT be used.
+ // ============================================================
+
+ // --- session settings ---
+ sql "set enable_nereids_planner=true"
+ sql "set enable_parallel_result_sink=false"
+ sql "set runtime_filter_mode=OFF"
+ sql "set parallel_pipeline_task_num=2"
+ sql "set bucketed_agg_min_input_rows=0"
+ sql "set bucketed_agg_max_group_keys=0"
+ // The table below is never analyzed, so group-by column stats are unknown
and
+ // StatsCalculator falls back to rows * DEFAULT_AGGREGATE_RATIO (1/3.0)
for the
+ // aggregate output cardinality. With the default
bucketed_agg_high_card_threshold
+ // (0.3), bucketedDataVolumeGatesPass rejects the pattern (rows/3 >
rows*0.3),
+ // so raise the threshold to make the positive fusion test deterministic.
+ sql "set bucketed_agg_high_card_threshold=1.0"
+
+ // --- create test table ---
+ sql """ DROP TABLE IF EXISTS bucketed_agg_reg_test; """
+ sql """
+ CREATE TABLE bucketed_agg_reg_test (
+ id int,
+ grp varchar(20),
+ val bigint
+ ) DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 3
+ PROPERTIES('replication_num' = '1');
+ """
+ sql """ INSERT INTO bucketed_agg_reg_test VALUES
+ (1, 'a', 10),
+ (2, 'b', 20),
+ (3, 'a', 30),
+ (4, 'b', 40),
+ (5, 'a', 50),
+ (1, 'c', 60),
+ (2, 'c', 70),
+ (3, 'b', 80),
+ (4, 'c', 90),
+ (5, 'b', 100);
+ """
+
+ // ============================================================
+ // Test 1: Positive — single-BE, bucketed enabled
+ // EXPLAIN should contain BUCKETED AGGREGATE
+ // ============================================================
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg = true;"
+
+ String query = "SELECT grp, SUM(val) FROM bucketed_agg_reg_test GROUP BY
grp;"
+ explain {
+ sql("${query}")
+ contains("BUCKETED AGGREGATE")
+ }
+
+ // Shape plan should show one-phase: hashAgg[GLOBAL] → shuffle → scan (no
LOCAL)
+ qt_bucketed_shape """explain shape plan
+ ${query}
+ """
+
+ // Verify correct results
+ order_qt_bucketed_result """
+ SELECT grp, SUM(val) FROM bucketed_agg_reg_test GROUP BY grp ORDER BY grp;
+ """
+
+ // ============================================================
+ // Test 2: Negative — be_number=3 (multi-BE), bucketed enabled
+ // Must NOT use bucketed agg, must fall back to two-phase
+ // ============================================================
+ sql "set be_number_for_test=3"
+ sql "set enable_bucketed_hash_agg = true;"
+
+ explain {
+ sql("${query}")
+ notContains("BUCKETED AGGREGATE")
+ }
+
+ // Shape plan should show two-phase: hashAgg[GLOBAL] → shuffle →
hashAgg[LOCAL] → scan
+ qt_multi_be_shape """explain shape plan
+ ${query}
+ """
+
+ // Results must match the single-BE bucketed result
+ order_qt_multi_be_result """
+ SELECT grp, SUM(val) FROM bucketed_agg_reg_test GROUP BY grp ORDER BY grp;
+ """
+
+ // ============================================================
+ // Test 3: Negative — bucketed disabled
+ // Must fall back to two-phase
+ // ============================================================
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg = false;"
+
+ explain {
+ sql("${query}")
+ notContains("BUCKETED AGGREGATE")
+ }
+
+ order_qt_disabled_result """
+ SELECT grp, SUM(val) FROM bucketed_agg_reg_test GROUP BY grp ORDER BY grp;
+ """
+
+ // ============================================================
+ // Test 4: Negative — scalar aggregation (no GROUP BY)
+ // Bucketed agg does not apply
+ // ============================================================
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg = true;"
+
+ String scalarQuery = "SELECT SUM(val) FROM bucketed_agg_reg_test;"
+ explain {
+ sql("${scalarQuery}")
+ notContains("BUCKETED AGGREGATE")
+ }
+
+ order_qt_no_group_by_result """
+ SELECT SUM(val) FROM bucketed_agg_reg_test;
+ """
+
+ // ============================================================
+ // Test 5: COUNT(DISTINCT) + GROUP BY — results must be correct
+ // ============================================================
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg = true;"
+ sql """
+ INSERT INTO bucketed_agg_reg_test VALUES
+ (6, 'a', 110),
+ (7, 'c', 120),
+ (8, 'b', 130);
+ """
+
+ order_qt_count_distinct_result """
+ SELECT grp, COUNT(DISTINCT id), SUM(val)
+ FROM bucketed_agg_reg_test
+ GROUP BY grp
+ ORDER BY grp;
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
