[refactor](nereids) agg strategy refactor #54079
Merged
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Contributor
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
Contributor
Author
|
run buildall |
morrySnow
reviewed
Jul 30, 2025
| } | ||
| this.hboPlanStatisticsProvider = Objects.requireNonNull(Env.getCurrentEnv().getHboPlanStatisticsManager() | ||
| .getHboPlanStatisticsProvider(), "HboPlanStatisticsProvider is null"); | ||
| this.requestChildrenProperties = childrenProperties; |
| // 2. collect agg expressions and generate agg function to slot reference map | ||
| List<Slot> aggFunctionOutput = Lists.newArrayList(); | ||
| ArrayList<FunctionCallExpr> execAggregateFunctions = Lists.newArrayListWithCapacity(outputExpressions.size()); | ||
| boolean isPartial = aggregate.getAggregateParam().aggMode.productAggregateBuffer; |
Contributor
There was a problem hiding this comment.
is this a bug in base code? maybe we should change the logic that generate aggMode
Contributor
There was a problem hiding this comment.
@feiniaofeiafei why aggMode isPartial but contains partial aggregation function
| if (curChildIndex == groupExpression.arity()) { | ||
| if (!calculateEnforce(requestChildrenProperties, outputChildrenProperties)) { | ||
| return; // if error exists, return | ||
| clear(); |
Contributor
There was a problem hiding this comment.
is a bug in base code too? we should fix it in a seperate PR and add some UT and regression case for it
Comment on lines
644
to
886
| // rewriteJobs.addAll(jobs(topic("split multi distinct", | ||
| // custom(RuleType.SPLIT_MULTI_DISTINCT, () -> SplitMultiDistinct.INSTANCE)))); | ||
|
|
fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
Show resolved
Hide resolved
fe/fe-core/src/main/java/org/apache/doris/nereids/util/AggregateUtils.java
Show resolved
Hide resolved
| && children.get(0).getPlan() instanceof PhysicalDistribute) { | ||
| return ImmutableList.of(); | ||
| // 如果origin property 满足group by key, 但是不满足required, 那么禁用这个计划 | ||
| PhysicalProperties originChildProperty = originChildrenProperties.get(0); |
Comment on lines
63
to
64
| public static final List<Class<? extends AggregateFunction>> finalMultiDistinctSupportOtherFunc = | ||
| ImmutableList.of(Count.class, Sum.class, Min.class, Max.class); |
| * - Available memory resources | ||
| * - Query complexity | ||
| */ | ||
| public class DistinctAggStrategySelector extends DefaultPlanRewriter<DistinctSelectorContext> |
| /** | ||
| * Split multi distinct strategy | ||
| * */ | ||
| public class SplitMultiDistinctStrategy { |
Contributor
Author
|
run buildall |
feiniaofeiafei
force-pushed
the
agg_strategy
branch
from
cd9c2fc to
f9fee5b
Compare
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34723 ms |
TPC-DS: Total hot run time: 175102 ms |
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34390 ms |
TPC-DS: Total hot run time: 173148 ms |
feiniaofeiafei
force-pushed
the
agg_strategy
branch
from
68d260d to
a6255fe
Compare
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34130 ms |
TPC-DS: Total hot run time: 172100 ms |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34388 ms |
TPC-DS: Total hot run time: 186091 ms |
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34455 ms |
TPC-DS: Total hot run time: 186143 ms |
ClickBench: Total hot run time: 32.94 s |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34269 ms |
TPC-DS: Total hot run time: 186447 ms |
ClickBench: Total hot run time: 32.57 s |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
FE Regression Coverage ReportIncrement line coverage |
924060929
previously approved these changes
Sep 4, 2025
github-actions
bot
added
the
approved
Contributor
|
PR approved by at least one committer and no changes requested. |
Contributor
|
PR approved by anyone and no changes requested. |
feiniaofeiafei
force-pushed
the
agg_strategy
branch
from
6b77923 to
4e330cc
Compare
github-actions
bot
removed
the
approved
feiniaofeiafei
force-pushed
the
agg_strategy
branch
from
4e330cc to
df8951b
Compare
Contributor
Author
|
run buildall |
TPC-H: Total hot run time: 34763 ms |
TPC-DS: Total hot run time: 190027 ms |
ClickBench: Total hot run time: 32.95 s |
Contributor
FE UT Coverage ReportIncrement line coverage |
Contributor
FE Regression Coverage ReportIncrement line coverage |
924060929
approved these changes
Sep 5, 2025
github-actions
bot
added
the
approved
Contributor
|
PR approved by at least one committer and no changes requested. |
morrySnow
approved these changes
Sep 5, 2025
Comment on lines
+315
to
+317
| default Statistics getStats() { | ||
| throw new IllegalStateException("Not support getStats for " + getClass().getName()); | ||
| } |
wenzhenghu
pushed a commit
to wenzhenghu/doris
that referenced
this pull request
Sep 8, 2025
### What problem does this PR solve?
#### Current Issues:
1. When the children of an agg satisfy a distribution, redundant
PhysicalAggregates are generated.
2. count(distinct a, b) generates a three-stage gather: the first two
stages deduplicate a and b, the third stage gathers the data, and then
counts. (This modification avoids a direct gather by adding a local agg
before the gather.)
3. The original code did not use statistical information to determine
the different Aggregate strategies. This refactor incorporates
statistical information as a basis for selection:
3.1 Utilizes statistical information in the logic for determining
whether to use multi_distinct or CTE splitting: When the ndv of the
group by key is low and the ndv of the distinct key is high,
multi_distinct is advantageous.
3.2 In the single distinct scenario, statistical information is also
used to determine whether to use a three-stage or four-stage Aggregate.
When the ndv of the group by key is low, a four-stage Aggregate is used.
#### REWRITE PHASE
##### DistinctAggStrategySelector
Add the DistinctAggStrategySelector rule to handle scenarios with
multiple distincts. Use statistical information to determine when to use
CTE splitting or multidistinct.
After this rule is implemented, only single distinct scenarios need to
be considered.
##### DistinctAggregateRewriter
Add the DistinctAggregateSplitter rule to handle scenarios with a single
distinct. Determine whether to split or use multi-distinct. Distinguish
between scenarios with and without group by.
No group by: Not processed here. With group by: If splitting is possible
and statistical information indicates that multi-stage splitting is more
optimal, split.
So, after this is complete, cases with distinct without group by:
No processing, move to the cascades phase.
Cases with distinct with group by:
1. distinct is rewritten to multi-distinct,
2. it is split, leaving distinct.
3. if it is neither rewritten to multi-distinct nor split, then
splitting is performed directly into two, three, or four stages in the
cascades phase.
#### CASCADES STAGE
##### SplitAgg
Add a SplitAgg rule to handle scenarios without distinct.
1. Split the Agg into two layers: local and global.
This allows you to avoid splitting the Agg into two stages, for example,
in group_concat scenarios. (If not, shuffle first, then agg.)
2. This also implements a single-stage Agg.
##### SplitAggMultiPhaseWithoutGbyKey
ADD implementation rule: SplitAggMultiPhaseWithoutGbyKey
only process agg without group by key, and with one distinct function
provide there three rewrite:
```sql
select count(distinct a) from t
splitToThreePhase:
agg(count(a); distinct global)
+--gather
+--agg(count(a); distinct local)
+--agg(group by a; global)
+--hashShuffle(a)
splitToFourPhase:
agg(count(a); distinct global)
+--gather
+--agg(count(a); distinct local)
+--agg(group by a; global)
+--hashShuffle(a)
+--agg(group by a; local)
twoPhaseAggregateWithFinalMultiDistinct:
agg(sum0(c1))
+--gather
+--agg(multi_distinct(a) as c1)
+--hashShuffle(a)
```
splitToThreePhase and twoPhaseAggregateWithFinalMultiDistinct provide
plan when scan(or other plan) satisfy distributed expr
##### SplitAggMultiPhase
ADD implementation rule: SplitAggMultiPhase
only process agg with group by key, and with one distinct function
select count(distinct a) group by b (deduplicated agg hashShuffle by
group by key b)
```sql
splitToTwoPlusOnePhase:
agg(group by b, count(a); distinct global)
+--agg(group by a,b; global)
+--hashShuffle(b)
+--agg(group by a,b; local)
agg(group by b, count(a); distinct global)
+--agg(group by a,b; global)
+--hashShuffle(b)
splitToTwoPlusTwoPhase:
agg(group by b, count(a); distinct global)
+--hashShuffle(b)
+--agg(group by b, count(a); distinct local)
+--agg(group by a,b; global)
+--hashShuffle(a,b)
+--agg(group by a,b; local)
agg(group by b, count(a); distinct global)
+--hashShuffle(b)
+--agg(group by b, count(a); distinct local)
+--agg(group by a,b; global)
+--hashShuffle(a,b)
splitToOnePlusTwoPhase: (deduplicated agg hashShuffle by distinct key a)
agg(group by b, count(a); distinct global)
+--hashShuffle(b)
+--agg(group by b, count(a); distinct local)
+--agg(group by a,b; global)
+--hashShuffle(a)
```
The second plan of splitToTwoPlusOnePhase and splitToTwoPlusTwoPhase
provide plan when scan(or other plan) satisfy group by distributed expr;
splitToOnePlusTwoPhase provide plan when scan(or other plan) satisfy
distinct key distribution.
#### RequestPropertyDeriver
Implement visitPhysicalHashAggregate. The main logic is:
If the request received by the AGG is hash-distributed with the
distribution column set S1, and the request sent by the AGG is also
hash-distributed with the distribution column set S2,
and S = S2.intersect(S1), and the NDV of S1 is calculated, and if the
NDV of S1 is found to be no less than a threshold, then the AGG sends
the child a hash distribution based on S1.
Otherwise, the AGG sends the child a hash distribution based on S.
When directly generating a three-stage/four-stage AGG, the partitionby
columns are specified. If the partitionby columns are not empty, the
child is hash-distributed based on the partitionby columns.
#### ChildrenPropertiesRegulator
Disabled one-stage AGG: AGG-Distribute
Special handling for one-stage AGG with CTE:
AGG-Distribute-CTE.
If the Distribute option is Gather, the option is disabled.
If the Distribute option is HashShuffle, the option is determined based
on statistics.
If no statistics are available, the option is enabled.
If statistics are available but the shuffle column is skewed, the option
is disabled.
This was referenced Sep 11, 2025
924060929
pushed a commit
that referenced
this pull request
Sep 12, 2025
… have statistics (#55855) Related PR: #54079 Problem Summary: Using the multi-distinct strategy instead of multi-stage/CTE splitting strategy for AGG in scenarios without statistical information, consistent with the default strategy before refactor(#54079), can achieve better performance in scenarios without statistical information.
924060929
pushed a commit
that referenced
this pull request
Sep 15, 2025
…epeat (#55907) Related PR: #54079 Problem Summary: There may be some issues in scenarios where the grouping() function is present. When merging a project, ```sql -- Upper level project output: projections : "GROUPING_PREFIX_publish_date#36 originExpression=Grouping(publish_date#6) AS `dim_207`#32" --Lower level project output: projections : "GROUP PING_PREVIX_publish_date # 79 AS ` GROUP PING_PREVIX_publish_date ` # 36" ``` In theory, it should be rewritten as `GROUPPING_PREVIX_publish_date# 79 AS dim_207# 32`, but there is a problem here: The map generated by the lower level project is: key : `GROUPING_PREFIX_publish_date#36 , value: GROUPING_PREFIX_publish_date#79 ` Dealing with upper level projects `GROUPING_PREFIX_publish_date#36 originExpression=Grouping(publish_date#6) AS dim_207#32` In the upper level project, the 'virtual slot' is used, and the 'key' in the map is' slot '. Although the ExprId is the same, it cannot be hit, resulting in an error. So temporarily disable the use of CTE splitting to handle grouping+multiple counts (distinctions) scenarios.
16 tasks
morrySnow
pushed a commit
that referenced
this pull request
Sep 23, 2025
…d count distinct multi expr exists same time (#56271) ### What problem does this PR solve? Related PR: #54079 Problem Summary: 1. Added a check for multi_distinct_count(a,b) in the MultiDistinctCount constructor to prevent the use of multiple columns. Because BE doesn't report an error in this case, it only uses the first argument of the multi_distinct function, resulting in incorrect results. 2. In scenarios without a group by key, when multi_distinct_func and count(distinct a,b) appear together, the original code converts count(distinct a,b) to multi_distinct_count(a), resulting in incorrect results. The correct approach is to use multi-stage splitting when count distinct multi_expr appears. 3. Removed the mustUseMultiDistinct flag. This flag is useless.
github-actions bot
pushed a commit
that referenced
this pull request
Sep 24, 2025
…d count distinct multi expr exists same time (#56271) ### What problem does this PR solve? Related PR: #54079 Problem Summary: 1. Added a check for multi_distinct_count(a,b) in the MultiDistinctCount constructor to prevent the use of multiple columns. Because BE doesn't report an error in this case, it only uses the first argument of the multi_distinct function, resulting in incorrect results. 2. In scenarios without a group by key, when multi_distinct_func and count(distinct a,b) appear together, the original code converts count(distinct a,b) to multi_distinct_count(a), resulting in incorrect results. The correct approach is to use multi-stage splitting when count distinct multi_expr appears. 3. Removed the mustUseMultiDistinct flag. This flag is useless.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
6 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
Current Issues:
3.1 Utilizes statistical information in the logic for determining whether to use multi_distinct or CTE splitting: When the ndv of the group by key is low and the ndv of the distinct key is high, multi_distinct is advantageous.
3.2 In the single distinct scenario, statistical information is also used to determine whether to use a three-stage or four-stage Aggregate. When the ndv of the group by key is low, a four-stage Aggregate is used.
REWRITE PHASE
DistinctAggStrategySelector
Add the DistinctAggStrategySelector rule to handle scenarios with multiple distincts. Use statistical information to determine when to use CTE splitting or multidistinct.
After this rule is implemented, only single distinct scenarios need to be considered.
DistinctAggregateRewriter
Add the DistinctAggregateSplitter rule to handle scenarios with a single distinct. Determine whether to split or use multi-distinct. Distinguish between scenarios with and without group by.
No group by: Not processed here. With group by: If splitting is possible and statistical information indicates that multi-stage splitting is more optimal, split.
So, after this is complete, cases with distinct without group by:
No processing, move to the cascades phase.
Cases with distinct with group by:
CASCADES STAGE
SplitAgg
Add a SplitAgg rule to handle scenarios without distinct.
This allows you to avoid splitting the Agg into two stages, for example, in group_concat scenarios. (If not, shuffle first, then agg.)
SplitAggMultiPhaseWithoutGbyKey
ADD implementation rule: SplitAggMultiPhaseWithoutGbyKey
only process agg without group by key, and with one distinct function
provide there three rewrite:
splitToThreePhase and twoPhaseAggregateWithFinalMultiDistinct provide plan when scan(or other plan) satisfy distributed expr
SplitAggMultiPhase
ADD implementation rule: SplitAggMultiPhase
only process agg with group by key, and with one distinct function
select count(distinct a) group by b (deduplicated agg hashShuffle by group by key b)
The second plan of splitToTwoPlusOnePhase and splitToTwoPlusTwoPhase provide plan when scan(or other plan) satisfy group by distributed expr;
splitToOnePlusTwoPhase provide plan when scan(or other plan) satisfy distinct key distribution.
RequestPropertyDeriver
Implement visitPhysicalHashAggregate. The main logic is:
If the request received by the AGG is hash-distributed with the distribution column set S1, and the request sent by the AGG is also hash-distributed with the distribution column set S2,
and S = S2.intersect(S1), and the NDV of S1 is calculated, and if the NDV of S1 is found to be no less than a threshold, then the AGG sends the child a hash distribution based on S1.
Otherwise, the AGG sends the child a hash distribution based on S.
When directly generating a three-stage/four-stage AGG, the partitionby columns are specified. If the partitionby columns are not empty, the child is hash-distributed based on the partitionby columns.
ChildrenPropertiesRegulator
Disabled one-stage AGG: AGG-Distribute
Special handling for one-stage AGG with CTE:
AGG-Distribute-CTE.
If the Distribute option is Gather, the option is disabled.
If the Distribute option is HashShuffle, the option is determined based on statistics.
If no statistics are available, the option is enabled.
If statistics are available but the shuffle column is skewed, the option is disabled.
Release note
None
Check List (For Author)
Test
Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)