mrhhsg opened a new pull request, #68651:
URL: https://github.com/apache/doris/pull/68651
### What problem does this PR solve?
Issue Number: None
Related PR: #61495 #65024 #67021 #67210 #67788 #68564 #68565 #66815 #61113
#61215 #61260 #68509
Problem Summary:
Backport bucketed hash aggregation (#61495) and all of its follow-up
refactors and fixes to branch-4.1. On a single-BE deployment a one-phase GROUP
BY aggregate is fused with its exchange into one BE operator: every pipeline
instance builds 256 per-bucket hash tables and the source side merges the
instances bucket by bucket, which avoids the exchange and the serialization of
intermediate states. It is controlled by `enable_bucketed_hash_agg` (default
`true`, the same as master) plus the `bucketed_agg_*` data-volume gates.
Picked commits, in order:
- Prerequisites that the bucketed operators depend on and branch-4.1 did not
have:
- #61113 the agg hash map iterator does not yield the NULL key
- #61215 fix the agg hash map iterator (`for_each`, `get_first`)
- #61260 GROUP BY count(*) simple-count optimization (`use_simple_count`)
- #61495 bucketed agg operator (BE sink/source operators,
`TBucketedAggregationNode` as `TPlanNode` field 53, same id as master)
- #65024 FE refactor: fuse the one-phase aggregate and its distribute in the
plan translator
- Fixes: #67021, #67210, #67788, #68564 (per-source-instance merge arenas),
#68565 (keep Java/Python UDAFs out), #66815 (aggregate-argument CSE below
distribute)
Conflict resolutions and branch-4.1 adaptations:
- #61260: `aggregation_source_operator.cpp` / `dependency.cpp` keep
branch-4.1's `IColumn::mutate(std::move(col))` and the `Status` returned by
`_destroy_agg_status`; the file is re-formatted with clang-format 16.
- #65024: `CostModel` keeps branch-4.1's `factor` computation;
`PlanTranslatorContext` only gets `fragmentMergeChildDepth` (the other fields
in the conflict belong to #63366); `SplitAggWithoutDistinct` stays as on
branch-4.1 because the master version carries #63732, which is not picked; the
#63732-only case is dropped from `BucketedAggregateTest`.
- #68564: `init_instances` is inline in `dependency.h` on branch-4.1, the
new arena initialization is added there.
- #68565: add the `algebra.Aggregate` import; the Java UDAF case lives under
`nereids_p0/javaudf` (branch-4.1 has no `query_p0/javaudf`).
- FE adaptation commit: `@VariableMgr.VarAttr`, `Expr#treeToThrift` /
`Expr.treesToThrift`, `PlanNode` constructor with `StatisticalType.AGG_NODE`.
- BE adaptation commit: the bucketed-operator parts of #64139 (`sink_impl` /
`get_block_impl`) and #63001 (`IColumn::mutate`) that branch-4.1 already has,
and the three-argument `filter_block`. In addition, the bucketed sink now uses
the default `required_data_distribution` (PASSTHROUGH for a serial child)
instead of always NOOP: branch-4.1 still plans local exchanges in BE (master
moved this to FE in #63366, which inserts the passthrough there), so with a
serial child the whole sink pipeline ran with one task. A BE UT covers it.
- `percentile_bucketed_agg_merge` (removed from branch-4.1 by #68509 because
the operator did not exist) is restored without the
`PERCENTILE_MERGE(PERCENTILE_UNION(...))` queries, which fail on branch-4.1 FE
with or without bucketed aggregation ("percentile requires second parameter
must be a constant"); the remaining results are identical to master.
Not picked on purpose: #63366 (FE local exchange planning), #63732, #65031,
#66903, #66672.
Known limitations shared with master and tracked for follow-up (the code is
identical to master): the bucketed path does not spill, does not mark blockable
aggregate functions as blockable, is not a query cache candidate, the optimizer
may choose the one-phase shape in places where the translator refuses to fuse
it, the CSE pass of #66815 can stack two projects above a materialized CTE
consumer, a failing merge on the source side can leak the moved state, and the
#61260 simple-count path skips evaluating the argument of `count(<non-nullable
expr>)`.
### Release note
Support bucketed hash aggregation on single-BE deployments (session variable
`enable_bucketed_hash_agg`, enabled by default).
### Check List (For Author)
- Test:
- Unit Test: FE BucketedAggregateTest, BucketedAggregateTranslatorTest,
RecursiveUnionFragmentMergeContextTest, RequestPropertyDeriverTest,
ChildrenPropertiesRegulatorTest, ChildOutputPropertyDeriverTest,
CostModelV1Test; BE HashTableMethodTest, AggOperator*, StreamingAgg*,
DistinctStreaming*, PartitionedAgg*, set operator tests,
BucketedAggSharedStateTest, AggOperatorRequiredDistributionTest
- Regression test (single BE, ASAN): nereids_rules_p0/agg_strategy (all,
including bucketed_hash_agg and cse_agg_distribute),
mv_p0/ut/testBucketedAggSyncMV, query_p0/aggregate
(collect_set_bucketed_agg_merge, percentile_bucketed_agg_merge and the rest of
the directory), nereids_p0/javaudf/test_javaudaf_bucketed_agg
- Behavior changed: Yes. Single-BE GROUP BY aggregates that pass the
data-volume gates use the new bucketed operator by default, as on master.
- Does this need documentation: No
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
