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 d921bebcd90 [fix](agg) Normalize equivalent multi-column distinct
counts (#65206)
d921bebcd90 is described below
commit d921bebcd90a3f55a2446a706d350f2624445c62
Author: feiniaofeiafei <[email protected]>
AuthorDate: Thu Jul 23 11:44:59 2026 +0800
[fix](agg) Normalize equivalent multi-column distinct counts (#65206)
### What problem does this PR solve?
Related PR: #54079
Problem Summary: NormalizeAggregate treated reordered equivalent
multi-column COUNT DISTINCT expressions as different aggregate
functions, which could make the multi-distinct strategy construct a join
from only one aggregate and fail with an index-out-of-bounds exception.
It also retained duplicate arguments even though downstream COUNT
DISTINCT processing treats them as a set, and the CountIf conversion
indexed the original argument list using the deduplicated size.
Canonicalize multi-column distinct arguments as an ordered set during
normalization and make CountIf consume that deduplicated list directly.
### Release note
Fix planning and execution of equivalent multi-column COUNT DISTINCT
expressions with reordered or duplicate arguments.
---
.../nereids/rules/analysis/NormalizeAggregate.java | 26 ++++-
.../PushDownAggWithDistinctThroughJoinOneSide.java | 2 +-
.../apache/doris/nereids/util/AggregateUtils.java | 10 +-
.../NormalizeMultiColumnDistinctCountTest.java | 125 +++++++++++++++++++++
.../agg_strategy/distinct_agg_rewriter.groovy | 15 +--
5 files changed, 162 insertions(+), 16 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeAggregate.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeAggregate.java
index aebe40c24d7..93a31c61f1a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeAggregate.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeAggregate.java
@@ -40,6 +40,7 @@ import
org.apache.doris.nereids.trees.expressions.SubqueryExpr;
import org.apache.doris.nereids.trees.expressions.WindowExpression;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.AnyValue;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import org.apache.doris.nereids.trees.expressions.functions.generator.Unnest;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
@@ -163,7 +164,8 @@ public class NormalizeAggregate implements
RewriteRuleFactory, NormalizeToSlot {
Set<Expression> groupingByExprs =
Utils.fastToImmutableSet(aggregate.getGroupByExpressions());
// collect all trivial-agg
- List<NamedExpression> aggregateOutput =
aggregate.getOutputExpressions();
+ List<NamedExpression> aggregateOutput =
normalizeMultiColumnDistinctCount(
+ aggregate.getOutputExpressions());
Map<AggregateFunction, Map<String, String>> aggFuncs =
CollectNonWindowedAggFuncsWithSessionVar.collect(aggregateOutput);
@@ -448,6 +450,28 @@ public class NormalizeAggregate implements
RewriteRuleFactory, NormalizeToSlot {
having.get().withChildren(new
LogicalProject<>(bottomProjectsBuilder.build(), newAggregate)));
}
+ private List<NamedExpression>
normalizeMultiColumnDistinctCount(List<NamedExpression> aggregateOutput) {
+ // Multi-column distinct counts treat arguments as a set. Remove
duplicates and canonicalize equivalent
+ // counts to the first argument order so the structural equality used
below can share one aggregate result.
+ Map<ImmutableSet<Expression>, Count> distinctArgumentsToCount = new
HashMap<>();
+ return ExpressionUtils.rewriteDownShortCircuit(aggregateOutput,
expression -> {
+ if (!(expression instanceof Count)) {
+ return expression;
+ }
+ Count count = (Count) expression;
+ if (!count.isDistinct() || count.arity() <= 1) {
+ return count;
+ }
+ ImmutableSet<Expression> distinctArguments =
ImmutableSet.copyOf(count.getDistinctArguments());
+ Count normalizedCount =
distinctArgumentsToCount.get(distinctArguments);
+ if (normalizedCount == null) {
+ normalizedCount = count.withDistinctAndChildren(true,
ImmutableList.copyOf(distinctArguments));
+ distinctArgumentsToCount.put(distinctArguments,
normalizedCount);
+ }
+ return count.withDistinctAndChildren(true,
normalizedCount.children());
+ });
+ }
+
private List<NamedExpression> normalizeOutput(List<NamedExpression>
aggregateOutput,
NormalizeToSlotContext groupByToSlotContext,
NormalizeToSlotContext argsOfAggFuncNeedPushDownContext,
NormalizeToSlotContext normalizedAggFuncsToSlotContext) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggWithDistinctThroughJoinOneSide.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggWithDistinctThroughJoinOneSide.java
index 3f9ad609744..dd09b2b52a3 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggWithDistinctThroughJoinOneSide.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/PushDownAggWithDistinctThroughJoinOneSide.java
@@ -64,7 +64,7 @@ public class PushDownAggWithDistinctThroughJoinOneSide
implements RewriteRuleFac
} else {
return funcs.stream()
.allMatch(f -> (f instanceof Min || f
instanceof Max || f instanceof Sum
- || f instanceof Count) &&
f.isDistinct()
+ || f instanceof Count) &&
f.isDistinct() && f.arity() == 1
&& f.child(0) instanceof Slot);
}
})
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 947da096185..39ced0eddec 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
@@ -40,6 +40,7 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.Set;
@@ -62,10 +63,11 @@ public class AggregateUtils {
/**countDistinctMultiExprToCountIf*/
public static Expression countDistinctMultiExprToCountIf(Count count) {
- Set<Expression> arguments = ImmutableSet.copyOf(count.getArguments());
- Expression countExpr = count.getArgument(arguments.size() - 1);
- for (int i = arguments.size() - 2; i >= 0; --i) {
- Expression argument = count.getArgument(i);
+ Iterator<Expression> arguments =
ImmutableSet.copyOf(count.getArguments())
+ .asList().reverse().iterator();
+ Expression countExpr = arguments.next();
+ while (arguments.hasNext()) {
+ Expression argument = arguments.next();
If ifNull = new If(new IsNull(argument), NullLiteral.INSTANCE,
countExpr);
countExpr = assignNullType(ifNull);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/NormalizeMultiColumnDistinctCountTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/NormalizeMultiColumnDistinctCountTest.java
new file mode 100644
index 00000000000..70bb1699949
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/NormalizeMultiColumnDistinctCountTest.java
@@ -0,0 +1,125 @@
+// 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.rules.analysis;
+
+import org.apache.doris.nereids.trees.expressions.Alias;
+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.Count;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.util.AggregateUtils;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.nereids.util.PlanConstructor;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class NormalizeMultiColumnDistinctCountTest implements
MemoPatternMatchSupported {
+ @Test
+ public void testNormalizeDifferentArgumentOrder() {
+ Plan student = new
LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
PlanConstructor.student,
+ ImmutableList.of());
+ Slot id = student.getOutput().get(0);
+ Slot name = student.getOutput().get(2);
+ Alias countIdName = new Alias(new Count(true, id, name),
"count_id_name");
+ Alias countNameId = new Alias(new Count(true, name, id),
"count_name_id");
+ LogicalAggregate<Plan> root = new
LogicalAggregate<>(ImmutableList.of(),
+ ImmutableList.of(countIdName, countNameId), student);
+
+ PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+ .applyTopDown(new NormalizeAggregate())
+ .matchesFromRoot(
+ logicalProject(
+ logicalAggregate(
+ logicalProject(
+ logicalOlapScan()
+ )
+ ).when(aggregate ->
aggregate.getAggregateFunctions().size() == 1)
+ .when(aggregate ->
aggregate.getOutputExpressions().size() == 1)
+ ).when(project -> project.getProjects().size() == 2)
+ .when(project ->
project.getProjects().get(0).getExprId()
+ .equals(countIdName.getExprId()))
+ .when(project ->
project.getProjects().get(1).getExprId()
+ .equals(countNameId.getExprId()))
+ .when(project ->
project.getProjects().get(0).getInputSlots()
+
.equals(project.getProjects().get(1).getInputSlots()))
+ );
+ }
+
+ @Test
+ public void testKeepDifferentDistinctArguments() {
+ Plan student = new
LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
PlanConstructor.student,
+ ImmutableList.of());
+ Slot id = student.getOutput().get(0);
+ Slot gender = student.getOutput().get(1);
+ Slot name = student.getOutput().get(2);
+ LogicalAggregate<Plan> root = new
LogicalAggregate<>(ImmutableList.of(),
+ ImmutableList.of(
+ new Alias(new Count(true, id, name), "count_id_name"),
+ new Alias(new Count(true, id, gender),
"count_id_gender")),
+ student);
+
+ PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+ .applyTopDown(new NormalizeAggregate())
+ .matches(
+ logicalJoin(
+ logicalAggregate()
+ .when(aggregate ->
aggregate.getAggregateFunctions().size() == 1),
+ logicalAggregate()
+ .when(aggregate ->
aggregate.getAggregateFunctions().size() == 1)
+ )
+ );
+ }
+
+ @Test
+ public void testRemoveDuplicateDistinctArguments() {
+ Plan student = new
LogicalOlapScan(StatementScopeIdGenerator.newRelationId(),
PlanConstructor.student,
+ ImmutableList.of());
+ Slot id = student.getOutput().get(0);
+ Slot name = student.getOutput().get(2);
+ Count count = new Count(true, id, id, name);
+
+ Count countIf = (Count)
AggregateUtils.countDistinctMultiExprToCountIf(count);
+ Assertions.assertEquals(ImmutableSet.of(id, name),
countIf.getInputSlots());
+
+ LogicalAggregate<Plan> root = new
LogicalAggregate<>(ImmutableList.of(),
+ ImmutableList.of(
+ new Alias(count, "count_id_id_name"),
+ new Alias(new Count(true, name, id), "count_name_id")),
+ student);
+ PlanChecker.from(MemoTestUtils.createConnectContext(), root)
+ .applyTopDown(new NormalizeAggregate())
+ .matchesFromRoot(
+ logicalProject(
+ logicalAggregate(
+ logicalProject(
+ logicalOlapScan()
+ )
+ ).when(aggregate ->
aggregate.getAggregateFunctions().size() == 1)
+ .when(aggregate ->
aggregate.getAggregateFunctions().iterator().next()
+ .arity() == 2)
+ )
+ );
+ }
+}
diff --git
a/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_rewriter.groovy
b/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_rewriter.groovy
index 33fb4f62db3..d3400db766f 100644
---
a/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_rewriter.groovy
+++
b/regression-test/suites/nereids_rules_p0/agg_strategy/distinct_agg_rewriter.groovy
@@ -38,17 +38,12 @@ suite("distinct_agg_rewriter") {
sql "set agg_phase=4"
sql "select group_concat(distinct dst_key1 order by dst_key2) from
t_gbykey_10_dstkey_10_1000_id group by gby_key"
- test {
- sql """select number % 2, count(distinct cast(number as varchar),
cast(number as varchar)),
- group_concat(distinct cast(number as varchar) order by number + 1)
from numbers('number'='10') group by number % 2"""
- exception "Unsupported query"
- }
- test {
- sql """select count(distinct cast(number as varchar), cast(number as
varchar)),
- group_concat(distinct cast(number as varchar) order by number + 1)
from numbers('number'='10')"""
- exception "Unsupported query"
- }
+ sql """select number % 2, count(distinct cast(number as varchar),
cast(number as varchar)),
+ group_concat(distinct cast(number as varchar) order by number + 1) from
numbers('number'='10') group by number % 2"""
+
+ sql """select count(distinct cast(number as varchar), cast(number as
varchar)),
+ group_concat(distinct cast(number as varchar) order by number + 1) from
numbers('number'='10')"""
// expect to have 3 HashAgg instead of 4 HashAgg
sql "set agg_phase=3"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
