This is an automated email from the ASF dual-hosted git repository.
morrySnow 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 9736521224a [fix](aggregate) Prevent bucketed aggregation for ordered
aggregates (#67210)
9736521224a is described below
commit 9736521224a6f23571afe478f73632cc17365370
Author: morrySnow <[email protected]>
AuthorDate: Fri Aug 28 10:46:57 2026 +0800
[fix](aggregate) Prevent bucketed aggregation for ordered aggregates
(#67210)
### What problem does this PR solve?
Related PR: #65024
Problem Summary:
BucketedAggregationNode does not carry aggregate ORDER BY sort metadata.
The bucketed fusion eligibility traversal stopped at AggregateExpression
nodes whose functions otherwise support two-phase aggregation, so
GROUP_CONCAT and MULTI_DISTINCT_GROUP_CONCAT with ORDER BY could be
fused incorrectly. Continue traversing supported aggregate expressions
so internal OrderExpression nodes reject bucketed fusion.
### Release note
Fix incorrect planning of aggregate functions with internal ORDER BY
when bucketed hash aggregation is enabled.
---
.../glue/translator/PhysicalPlanTranslator.java | 4 +-
.../BucketedAggregateTranslatorTest.java | 95 ++++++++++++++++++++++
.../agg_strategy/bucketed_hash_agg.groovy | 43 ++++++++++
3 files changed, 140 insertions(+), 2 deletions(-)
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 cf5ac2f5e18..1d6923a409f 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
@@ -3429,14 +3429,14 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
// (e.g. GROUP_CONCAT(... ORDER BY ...)) needs sort-info
// metadata, which BucketedAggregationNode does not carry.
foundOnePhaseOnly.set(true);
- return false;
+ return true;
}
if (c instanceof AggregateExpression) {
AggregateFunction func = ((AggregateExpression)
c).getFunction();
if (!func.supportAggregatePhase(AggregatePhase.TWO)) {
foundOnePhaseOnly.set(true);
+ return true;
}
- return true;
}
return false;
});
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
new file mode 100644
index 00000000000..479d55e380b
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/BucketedAggregateTranslatorTest.java
@@ -0,0 +1,95 @@
+// 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.glue.translator;
+
+import org.apache.doris.planner.AggregationNode;
+import org.apache.doris.planner.BucketedAggregationNode;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.planner.PlanNode;
+import org.apache.doris.planner.Planner;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.utframe.TestWithFeService;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class BucketedAggregateTranslatorTest extends TestWithFeService {
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+
connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
+ createDatabase("bucketed_aggregate_translator_test");
+ createTable("CREATE TABLE
bucketed_aggregate_translator_test.agg_group_concat_table ("
+ + "kint INT NOT NULL, kbint INT NOT NULL, kstr STRING NOT
NULL) "
+ + "DISTRIBUTED BY HASH(kint) BUCKETS 4 "
+ + "PROPERTIES('replication_num' = '1')");
+ }
+
+ @Test
+ public void testAggregateOrderByIsNotFusedIntoBucketedAggregation() throws
Exception {
+ SessionVariable sessionVariable = connectContext.getSessionVariable();
+ int oldAggPhase = sessionVariable.aggPhase;
+ int oldBeNumberForTest = sessionVariable.getBeNumberForTest();
+ long oldBucketedAggMinInputRows =
sessionVariable.bucketedAggMinInputRows;
+ long oldBucketedAggMaxGroupKeys =
sessionVariable.bucketedAggMaxGroupKeys;
+ double oldBucketedAggHighCardThreshold =
sessionVariable.bucketedAggHighCardThreshold;
+ boolean oldEnableBucketedHashAgg =
sessionVariable.enableBucketedHashAgg;
+ boolean oldUseOnePhaseAggForGroupConcatWithOrder =
+ sessionVariable.useOnePhaseAggForGroupConcatWithOrder;
+ try {
+ sessionVariable.aggPhase = 1;
+ sessionVariable.setBeNumberForTest(1);
+ sessionVariable.bucketedAggMinInputRows = 0;
+ sessionVariable.bucketedAggMaxGroupKeys = 0;
+ sessionVariable.bucketedAggHighCardThreshold = 1.0;
+ sessionVariable.enableBucketedHashAgg = true;
+ sessionVariable.useOnePhaseAggForGroupConcatWithOrder = false;
+
+ assertUsesRegularAggregation("group_concat(kstr ORDER BY kint)");
+ assertUsesRegularAggregation("multi_distinct_group_concat(kstr
ORDER BY kint)");
+ } finally {
+ sessionVariable.aggPhase = oldAggPhase;
+ sessionVariable.setBeNumberForTest(oldBeNumberForTest);
+ sessionVariable.bucketedAggMinInputRows =
oldBucketedAggMinInputRows;
+ sessionVariable.bucketedAggMaxGroupKeys =
oldBucketedAggMaxGroupKeys;
+ sessionVariable.bucketedAggHighCardThreshold =
oldBucketedAggHighCardThreshold;
+ sessionVariable.enableBucketedHashAgg = oldEnableBucketedHashAgg;
+ sessionVariable.useOnePhaseAggForGroupConcatWithOrder =
+ oldUseOnePhaseAggForGroupConcatWithOrder;
+ }
+ }
+
+ private void assertUsesRegularAggregation(String aggregateFunction) throws
Exception {
+ Planner planner = getSQLPlanner("SELECT " + aggregateFunction
+ + " FROM
bucketed_aggregate_translator_test.agg_group_concat_table GROUP BY kbint");
+ List<BucketedAggregationNode> bucketedAggregationNodes =
Lists.newArrayList();
+ List<AggregationNode> aggregationNodes = Lists.newArrayList();
+ for (PlanFragment fragment : planner.getFragments()) {
+ PlanNode root = fragment.getPlanRoot();
+ if (root != null) {
+ root.collect(BucketedAggregationNode.class,
bucketedAggregationNodes);
+ root.collect(AggregationNode.class, aggregationNodes);
+ }
+ }
+ Assertions.assertTrue(bucketedAggregationNodes.isEmpty());
+ Assertions.assertFalse(aggregationNodes.isEmpty());
+ }
+}
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
index 1842ad8b1b7..7efcfb152d3 100644
---
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
@@ -191,4 +191,47 @@ suite("bucketed_hash_agg") {
SELECT VAR_POP(DISTINCT val), VAR_POP(id)
FROM bucketed_agg_reg_test;
"""
+
+ // ============================================================
+ // Test 7: Aggregate functions with internal ORDER BY require
+ // agg_sort_infos, which BucketedAggregationNode cannot carry.
+ // ============================================================
+ sql "set agg_phase=1"
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg=true"
+ sql "set use_one_phase_agg_for_group_concat_with_order=false"
+ sql "set parallel_pipeline_task_num=2"
+
+ sql "DROP TABLE IF EXISTS agg_group_concat_table"
+ sql """
+ CREATE TABLE agg_group_concat_table (
+ kint INT NOT NULL,
+ kbint INT NOT NULL,
+ kstr STRING NOT NULL,
+ kstr2 STRING NOT NULL,
+ kastr ARRAY<STRING> NOT NULL
+ ) ENGINE=OLAP
+ DISTRIBUTED BY HASH(kint) BUCKETS 4
+ PROPERTIES('replication_num' = '1');
+ """
+ sql """
+ INSERT INTO agg_group_concat_table VALUES
+ (1, 1, 'string1', 'string3', ['s11', 's12', 's13']),
+ (1, 2, 'string2', 'string1', ['s21', 's22', 's23']),
+ (2, 3, 'string3', 'string2', ['s31', 's32', 's33']),
+ (1, 1, 'string1', 'string3', ['s11', 's12', 's13']),
+ (1, 2, 'string2', 'string1', ['s21', 's22', 's23']),
+ (2, 3, 'string3', 'string2', ['s31', 's32', 's33']);
+ """
+
+ String groupConcatWithOrder = """
+ SELECT multi_distinct_group_concat(kstr ORDER BY kint)
+ FROM agg_group_concat_table
+ GROUP BY kbint
+ """
+ explain {
+ sql(groupConcatWithOrder)
+ notContains("BUCKETED AGGREGATE")
+ }
+ sql(groupConcatWithOrder)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
