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 41cd2b473e9 [fix](eager-agg) Disable bilateral eager aggregation for
decimal SUM (#69093)
41cd2b473e9 is described below
commit 41cd2b473e9d9bb26be089f302cdfdfe6e107eaf
Author: feiniaofeiafei <[email protected]>
AuthorDate: Sat Nov 7 23:27:38 2026 +0800
[fix](eager-agg) Disable bilateral eager aggregation for decimal SUM
(#69093)
### What problem does this PR solve?
Related PR: #63690
Problem Summary:
Bilateral eager aggregation can change the result type and scale of a
decimal SUM because its compensation uses ordinary decimal
multiplication. This can cause an output type mismatch or silently lose
precision; for example, a valid result of `0.0000000004` can become
`0.000000`.
This PR disables bilateral eager aggregation when any aggregate is a
DecimalV3 SUM. Single-sided decimal SUM pushdown remains supported.
This causes either:
- An output type mismatch when an upper projection still expects the
original SUM type.
- Silent precision loss when the reduced-scale result reaches the query
output.
For example, a valid SUM of `0.0000000004` could become `0.000000`.
This Pr fix this problem:
- Introduce an internal decimal multiplication expression that preserves
the SUM result type.
- Cast the compensation input to the original SUM type when necessary.
- Preserve multiplication statistics estimation and prevent arithmetic
simplification or FE constant folding from reverting to ordinary decimal
multiplication semantics.
- Reuse the existing BE multiplication and overflow checks. No BE
changes are required.
---
.../rewrite/eageraggregation/EagerAggRewriter.java | 7 ++
.../eageraggregation/EagerAggRewriterTest.java | 57 +++++++++
.../eager_agg/decimal_sum_multiplicity.out | 133 +++++++++++++++++++++
.../data/query_p0/eager_agg/eager_agg.out | 24 +---
.../eager_agg/decimal_sum_multiplicity.groovy | 86 +++++++++++++
5 files changed, 289 insertions(+), 18 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
index 3081f36cdee..8296df29ebf 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java
@@ -35,6 +35,7 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunctio
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
@@ -261,6 +262,12 @@ public class EagerAggRewriter extends
DefaultPlanRewriter<PushDownAggContext> {
return adjustPushSideForNullToNonNull(join, context, toLeft,
toRight);
}
if (toLeft && toRight) {
+ // Decimal SUM compensation can lose scale with ordinary
multiplication, while preserving
+ // its type can reject intermediate products that would cancel in
the final SUM.
+ if (context.getAggFunctions().stream()
+ .anyMatch(func -> func instanceof Sum &&
func.getDataType().isDecimalV3Type())) {
+ return Pair.of(false, false);
+ }
return join.getJoinType().isInnerOrCrossJoin()
? Pair.of(true, true)
: Pair.of(false, false);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
index c22cdad81ce..2fe0de52abf 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriterTest.java
@@ -32,6 +32,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
import org.apache.doris.nereids.types.BigIntType;
+import org.apache.doris.nereids.types.DecimalV3Type;
import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.utframe.TestWithFeService;
@@ -74,6 +75,62 @@ class EagerAggRewriterTest extends TestWithFeService
implements MemoPatternMatch
connectContext.getSessionVariable().setDisableNereidsRules("PRUNE_EMPTY_PARTITION");
}
+ @Test
+ void testDecimalSumDisablesBilateralPushDown() throws Exception {
+ createTable("CREATE TABLE decimal_sum_l (k VARCHAR(20), v
DECIMAL(38,18))"
+ + " DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1"
+ + " PROPERTIES ('replication_num'='1')");
+ createTable("CREATE TABLE decimal_sum_r (k VARCHAR(20), w BIGINT)"
+ + " DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1"
+ + " PROPERTIES ('replication_num'='1')");
+ int oldMode =
connectContext.getSessionVariable().getEagerAggregationMode();
+ boolean oldReorder =
connectContext.getSessionVariable().isDisableJoinReorder();
+ boolean oldDecimal256 =
connectContext.getSessionVariable().enableDecimal256;
+ try {
+ connectContext.getSessionVariable().setEagerAggregationMode(1);
+ connectContext.getSessionVariable().setDisableJoinReorder(true);
+ for (boolean decimal256 : new boolean[] {false, true}) {
+ connectContext.getSessionVariable().enableDecimal256 =
decimal256;
+ DecimalV3Type expected =
DecimalV3Type.createDecimalV3Type(decimal256 ? 76 : 38, 18);
+ for (String join : new String[] {"JOIN decimal_sum_r r ON
l.k=r.k",
+ "CROSS JOIN decimal_sum_r r"}) {
+ for (String sums : new String[] {"SUM(l.v) AS sv, SUM(r.w)
AS sw",
+ "SUM(r.w) AS sw, SUM(l.v) AS sv"}) {
+ Plan plan = PlanChecker.from(connectContext)
+ .analyze("SELECT l.k AS g, " + sums + " FROM
decimal_sum_l l "
+ + join + " GROUP BY l.k")
+ .rewrite().getPlan();
+ int decimalOutputIndex = sums.startsWith("SUM(l.v)") ?
1 : 2;
+ Assertions.assertEquals(expected,
plan.getOutput().get(decimalOutputIndex).getDataType());
+ LogicalJoin<?, ?> logicalJoin = findFirstPlan(plan,
LogicalJoin.class);
+ Assertions.assertNotNull(logicalJoin,
plan.treeString());
+ Assertions.assertFalse(containsPlan(logicalJoin,
LogicalAggregate.class), plan.treeString());
+ }
+ }
+ assertNoAggregateUnderJoin("SELECT SUBSTRING(l.k, 1, 6),
SUM(l.v), SUM(r.w)"
+ + " FROM decimal_sum_l l JOIN decimal_sum_r r ON
l.k=r.k"
+ + " GROUP BY SUBSTRING(l.k, 1, 6)");
+ assertNoAggregateUnderJoin("SELECT r.k, SUM(l.v), COUNT(r.w)"
+ + " FROM decimal_sum_r r JOIN decimal_sum_l l ON
l.k=r.k GROUP BY r.k");
+ // A decimal SUM on just one side still supports eager
aggregation.
+ for (String from : new String[] {"decimal_sum_l l JOIN
decimal_sum_r r ON l.k=r.k",
+ "decimal_sum_r r JOIN decimal_sum_l l ON l.k=r.k"}) {
+ Plan plan = PlanChecker.from(connectContext)
+ .analyze("SELECT l.k, SUM(l.v) FROM " + from + "
GROUP BY l.k")
+ .rewrite().getPlan();
+ Assertions.assertEquals(expected,
plan.getOutput().get(1).getDataType());
+ LogicalJoin<?, ?> logicalJoin = findFirstPlan(plan,
LogicalJoin.class);
+ Assertions.assertNotNull(logicalJoin, plan.treeString());
+ Assertions.assertTrue(containsPlan(logicalJoin,
LogicalAggregate.class), plan.treeString());
+ }
+ }
+ } finally {
+
connectContext.getSessionVariable().setEagerAggregationMode(oldMode);
+
connectContext.getSessionVariable().setDisableJoinReorder(oldReorder);
+ connectContext.getSessionVariable().enableDecimal256 =
oldDecimal256;
+ }
+ }
+
@Test
void testNotPushAggCaseWhenToNullableSideOfOuterJoin() {
connectContext.getSessionVariable().setEagerAggregationMode(1);
diff --git
a/regression-test/data/query_p0/eager_agg/decimal_sum_multiplicity.out
b/regression-test/data/query_p0/eager_agg/decimal_sum_multiplicity.out
new file mode 100644
index 00000000000..265c3149899
--- /dev/null
+++ b/regression-test/data/query_p0/eager_agg/decimal_sum_multiplicity.out
@@ -0,0 +1,133 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !alias_false_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_false_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_false_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_false_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_false_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_false_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_false_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_false_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_false_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_false_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_false_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_false_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_true_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_true_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_true_false_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_true_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_true_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_true_false_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_true_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_true_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_true_true_off --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !alias_true_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !no_alias_true_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !insert_true_true_on --
+negative -2.246913578024691356 6
+null \N 6
+x 4.00000000E-10 4
+
+-- !cancellation_false_off --
+cancel 0E-18 4
+
+-- !cancellation_false_on --
+cancel 0E-18 4
+
+-- !cancellation_true_off --
+cancel 0E-18 4
+
+-- !cancellation_true_on --
+cancel 0E-18 4
+
diff --git a/regression-test/data/query_p0/eager_agg/eager_agg.out
b/regression-test/data/query_p0/eager_agg/eager_agg.out
index b044573b029..e061fb9fec5 100644
--- a/regression-test/data/query_p0/eager_agg/eager_agg.out
+++ b/regression-test/data/query_p0/eager_agg/eager_agg.out
@@ -5,12 +5,8 @@ PhysicalResultSink
----hashAgg[LOCAL]
------hashJoin[INNER_JOIN] hashCondition=((dt.d_date_sk = ss.ss_sold_date_sk))
otherCondition=()
--------hashJoin[INNER_JOIN] hashCondition=((ss.ss_item_sk = ws.ws_item_sk))
otherCondition=()
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[store_sales(ss)]
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[web_sales(ws)]
+----------PhysicalOlapScan[store_sales(ss)]
+----------PhysicalOlapScan[web_sales(ws)]
--------PhysicalOlapScan[date_dim(dt)]
Hint log:
@@ -47,12 +43,8 @@ PhysicalResultSink
----hashAgg[LOCAL]
------hashJoin[INNER_JOIN] hashCondition=((dt.d_date_sk = ss.ss_sold_date_sk))
otherCondition=()
--------hashJoin[INNER_JOIN] hashCondition=((ss.ss_item_sk = ws.ws_item_sk))
otherCondition=()
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[store_sales(ss)]
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[web_sales(ws)]
+----------PhysicalOlapScan[store_sales(ss)]
+----------PhysicalOlapScan[web_sales(ws)]
--------PhysicalOlapScan[date_dim(dt)]
Hint log:
@@ -131,12 +123,8 @@ PhysicalResultSink
----hashAgg[LOCAL]
------hashJoin[INNER_JOIN] hashCondition=((dt.d_date_sk = ss.ss_sold_date_sk))
otherCondition=()
--------hashJoin[INNER_JOIN] hashCondition=((ss.ss_item_sk = ws.ws_item_sk))
otherCondition=()
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[store_sales(ss)]
-----------hashAgg[GLOBAL]
-------------hashAgg[LOCAL]
---------------PhysicalOlapScan[web_sales(ws)]
+----------PhysicalOlapScan[store_sales(ss)]
+----------PhysicalOlapScan[web_sales(ws)]
--------PhysicalOlapScan[date_dim(dt)]
Hint log:
diff --git
a/regression-test/suites/query_p0/eager_agg/decimal_sum_multiplicity.groovy
b/regression-test/suites/query_p0/eager_agg/decimal_sum_multiplicity.groovy
new file mode 100644
index 00000000000..ccec7a74169
--- /dev/null
+++ b/regression-test/suites/query_p0/eager_agg/decimal_sum_multiplicity.groovy
@@ -0,0 +1,86 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("decimal_sum_multiplicity") {
+ sql """
+ DROP TABLE IF EXISTS decimal_sum_l;
+ DROP TABLE IF EXISTS decimal_sum_r;
+ DROP TABLE IF EXISTS decimal_sum_out;
+ CREATE TABLE decimal_sum_l (k VARCHAR(20) NOT NULL, v DECIMAL(38,18)
NULL)
+ DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num"="1");
+ CREATE TABLE decimal_sum_r (k VARCHAR(20) NOT NULL, w BIGINT NULL)
+ DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num"="1");
+ CREATE TABLE decimal_sum_out (k VARCHAR(20), sv DECIMAL(38,18), sw
BIGINT)
+ DUPLICATE KEY(k) DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num"="1");
+ INSERT INTO decimal_sum_l VALUES
+ ('x', 0.000000000100000000), ('x', 0.000000000100000000),
+ ('negative', -1.123456789012345678), ('negative', NULL),
+ ('null', NULL), ('null', NULL);
+ INSERT INTO decimal_sum_r VALUES
+ ('x', 1), ('x', 1), ('negative', 1), ('negative', 2), ('null', 3);
+ """
+ sql "SET disable_join_reorder=true"
+ sql "SET decimal_overflow_scale=6"
+ for (def decimal256 : [false, true]) {
+ sql "SET enable_decimal256=${decimal256}"
+ for (def checkOverflow : [false, true]) {
+ sql "SET check_overflow_for_decimal=${checkOverflow}"
+ for (def mode : [-1, 1]) {
+ sql "SET eager_aggregation_mode=${mode}"
+ def suffix = "${decimal256}_${checkOverflow}_${mode == -1 ?
'off' : 'on'}"
+ // The alias used to expose the stale output type; without it,
scale could be lost silently.
+ "order_qt_alias_${suffix}" """
+ SELECT l.k AS g, SUM(l.v) AS sv, SUM(r.w) AS sw
+ FROM decimal_sum_l l JOIN [shuffle] decimal_sum_r r ON
l.k=r.k GROUP BY l.k
+ """
+ "order_qt_no_alias_${suffix}" """
+ SELECT l.k, SUM(l.v), SUM(r.w)
+ FROM decimal_sum_l l JOIN [shuffle] decimal_sum_r r ON
l.k=r.k GROUP BY l.k
+ """
+ sql """
+ INSERT OVERWRITE TABLE decimal_sum_out
+ SELECT l.k AS g, CAST(SUM(l.v) AS DECIMAL(38,18)), SUM(r.w)
+ FROM decimal_sum_l l JOIN [shuffle] decimal_sum_r r ON
l.k=r.k GROUP BY l.k
+ """
+ "order_qt_insert_${suffix}" "SELECT * FROM decimal_sum_out"
+ }
+ }
+ }
+ // The compensation products exceed DECIMAL(38,18), but fit Int128 and
cancel in the final SUM.
+ sql """INSERT INTO decimal_sum_l VALUES
+ ('cancel_positive', 60000000000000000000.000000000000000000),
+ ('cancel_negative', -60000000000000000000.000000000000000000)"""
+ sql """INSERT INTO decimal_sum_r VALUES
+ ('cancel_positive', 1), ('cancel_positive', 1),
+ ('cancel_negative', 1), ('cancel_negative', 1)"""
+ for (def decimal256 : [false, true]) {
+ sql "SET enable_decimal256=${decimal256}"
+ sql "SET check_overflow_for_decimal=true"
+ for (def mode : [-1, 1]) {
+ sql "SET eager_aggregation_mode=${mode}"
+ "order_qt_cancellation_${decimal256}_${mode == -1 ? 'off' : 'on'}"
"""
+ SELECT SUBSTRING(l.k, 1, 6), SUM(l.v), SUM(r.w)
+ FROM decimal_sum_l l JOIN [shuffle] decimal_sum_r r ON l.k=r.k
+ WHERE l.k IN ('cancel_positive', 'cancel_negative')
+ GROUP BY SUBSTRING(l.k, 1, 6)
+ """
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
