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 60c5a8d357e [fix](fe) Do not fuse partial aggregate functions into
bucketed aggregation (#67021)
60c5a8d357e is described below
commit 60c5a8d357e03ec3ff4adba172b4ec72d0dde482
Author: minghong <[email protected]>
AuthorDate: Mon Aug 24 10:19:42 2026 +0800
[fix](fe) Do not fuse partial aggregate functions into bucketed aggregation
(#67021)
### What problem does this PR solve?
Issue Number: close #xxx
Problem Summary:
A DISTINCT aggregate query that also contains a non-distinct aggregate
function (e.g. "select stddev_pop(distinct a), stddev_pop(b) from t")
can fail at runtime with:
Aggregate function NullableV2(stddev) result type check failed:
Column type String is not compatible with data type DOUBLE
The 3-phase DISTINCT plan (SplitAggMultiPhaseWithoutGbyKey) builds a
dedup aggregate that is one-phase GLOBAL(INPUT_TO_RESULT) with group-by
keys, but carries the non-distinct functions in INPUT_TO_BUFFER mode.
The bucketed fusion path introduced in the translator
(shouldUseBucketedFusion / visitBucketedFusion) treated this node as a
genuine one-phase aggregate and fused it into BucketedAggregationNode,
hardcoding isPartial=false and needsFinalize=true. The output tuple slot
of a buffer-producing function is Varchar (AggregateExpression maps
productAggregateBuffer to the serialized Varchar type), while the BE
writes the function's final result (DOUBLE for stddev) into that slot,
so the BE result-type check fails.
Fix: reject bucketed fusion when the aggregate's output contains any
buffer-producing (partial) aggregate function. Such aggregates keep the
regular AggregationNode path, which serializes when isPartial. Genuine
one-phase aggregates (all functions INPUT_TO_RESULT) still fuse.
---
.../glue/translator/PhysicalPlanTranslator.java | 41 ++++++++++++++++++++++
.../agg_strategy/bucketed_hash_agg.out | 10 ++++++
.../agg_strategy/bucketed_hash_agg.groovy | 32 +++++++++++++++++
3 files changed, 83 insertions(+)
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 fe4c32e3c6a..8c1395c9923 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
@@ -3303,6 +3303,19 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
|| aggregate.getAggMode() != AggMode.INPUT_TO_RESULT) {
return false;
}
+ // BucketedAggregationNode always finalizes into the output tuple slot
+ // types (need_finalize=true, isPartial=false), so fusing an aggregate
+ // whose functions produce buffers (partial) would fail the BE
+ // result-type check: the slot type of a buffer-producing function is
+ // Varchar while the function's final return type (e.g. DOUBLE for
+ // stddev) is what insert_result_into writes. The one-phase GLOBAL
+ // dedup aggregate of a 3-phase DISTINCT plan has exactly this shape —
+ // the node itself is INPUT_TO_RESULT but its non-distinct functions
+ // run in INPUT_TO_BUFFER mode — and must stay on the regular
+ // AggregationNode path, which serializes when isPartial.
+ if (containsPartialAggFunction(aggregate)) {
+ 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
@@ -3432,6 +3445,34 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
return true;
}
+ /**
+ * Check whether the aggregate's output contains any buffer-producing
+ * (partial) aggregate function, i.e. an AggregateExpression in a mode with
+ * productAggregateBuffer=true. BucketedAggregationNode cannot carry such
+ * functions: it always finalizes into the output tuple slot types, while a
+ * buffer-producing function's slot type is the serialized Varchar type
+ * (AggregateExpression.getDataType) — writing the final result (e.g.
DOUBLE
+ * for stddev) into that String column fails the BE result-type check.
+ */
+ private boolean containsPartialAggFunction(PhysicalHashAggregate<? extends
Plan> aggregate) {
+ for (NamedExpression o : aggregate.getOutputExpressions()) {
+ AtomicBoolean foundPartial = new AtomicBoolean(false);
+ o.foreach(c -> {
+ if (c instanceof AggregateExpression) {
+ if (((AggregateExpression)
c).getAggregateParam().aggMode.productAggregateBuffer) {
+ foundPartial.set(true);
+ }
+ return true;
+ }
+ return false;
+ });
+ if (foundPartial.get()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* Fuse a one-phase GLOBAL hash aggregate and its PhysicalDistribute child
* into a BucketedAggregationNode, skipping the exchange node entirely.
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
index b2507b7b088..71c5b3d3082 100644
--- 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
@@ -38,3 +38,13 @@ c 220
a 4 200
b 5 370
c 4 340
+
+-- !distinct_stddev_pop_result --
+37.416573867739416 2.1289773081689303
+
+-- !distinct_stddev_samp_result --
+38.94440481849308 2.215909838025097
+
+-- !distinct_var_pop_result --
+1400 4.5325443786982245
+
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 fda888d84d5..1842ad8b1b7 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
@@ -159,4 +159,36 @@ suite("bucketed_hash_agg") {
GROUP BY grp
ORDER BY grp;
"""
+
+ // ============================================================
+ // Test 6: DISTINCT stddev/var mixed with a non-distinct aggregate.
+ // 3-phase DISTINCT plans build a one-phase GLOBAL(INPUT_TO_RESULT)
+ // dedup aggregate whose non-distinct functions run in
+ // INPUT_TO_BUFFER mode (Varchar output slots). Such an aggregate
+ // must NOT be fused into BucketedAggregationNode — the bucketed
+ // node always finalizes into the tuple slot types, so writing the
+ // final DOUBLE result into the Varchar slot fails the BE
+ // result-type check ("Column type String is not compatible with
+ // data type DOUBLE").
+ // parallel_pipeline_task_num=1 makes the single-execution-instance
+ // path pick the 3-phase plan deterministically.
+ // ============================================================
+ sql "set be_number_for_test=1"
+ sql "set enable_bucketed_hash_agg = true;"
+ sql "set parallel_pipeline_task_num=1"
+
+ order_qt_distinct_stddev_pop_result """
+ SELECT STDDEV_POP(DISTINCT val), STDDEV_POP(id)
+ FROM bucketed_agg_reg_test;
+ """
+
+ order_qt_distinct_stddev_samp_result """
+ SELECT STDDEV_SAMP(DISTINCT val), STDDEV_SAMP(id)
+ FROM bucketed_agg_reg_test;
+ """
+
+ order_qt_distinct_var_pop_result """
+ SELECT VAR_POP(DISTINCT val), VAR_POP(id)
+ FROM bucketed_agg_reg_test;
+ """
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
