This is an automated email from the ASF dual-hosted git repository.
HappenLee 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 1b5dcfdb4c6 [feature](be) Add aggregate state finalizers and batch
merging (#68312)
1b5dcfdb4c6 is described below
commit 1b5dcfdb4c6c48b34392fdc56e6b9721942632e8
Author: HappenLee <[email protected]>
AuthorDate: Mon Oct 12 11:50:16 2026 +0800
[feature](be) Add aggregate state finalizers and batch merging (#68312)
### What problem does this PR solve?
Problem Summary:
A query that has already computed one aggregate state per key still
needs an aggregate operator to obtain its final values through
`<agg>_merge`. This repeats aggregation for the finest grouping when
states are also reused for coarser rollups.
Add the scalar `<agg>_finalize(state)` combinator. It returns one result
for each input state, using the existing aggregate implementation and
serialized-state representation. For example:
```sql
SELECT k, avg_finalize(s)
FROM (SELECT k, avg_combine(v) AS s FROM t GROUP BY k) partial;
```
The same implementation supports aggregates such as `count`, `sum`,
`min`, `max`, and `array_agg`. FE validates that the state's canonical
aggregate name matches the finalizer, derives its result type, and
treats the function as scalar. BE handles the underlying serialized
column type, skips outer NULL payloads, and releases temporary state
after each row. Constant inputs use the ordinary scalar constant path.
Existing empty-state semantics and serialized formats are unchanged.
MV rollup matching follows only direct MERGE/UNION state chains for the
same aggregate and preserves the full value expressions of
STATE/COMBINE. This prevents scalar finalizers nested in their arguments
from making different outer aggregates or different value expressions
appear equivalent, while retaining compatible state rollups.
Grouped `*_union` and `*_merge` now delegate to the existing vectorized
deserialize-and-merge interfaces, including the selected path. The
nested aggregate exposes whether its vector readers need scratch: direct
native readers receive a null scratch pointer, while readers that need
temporary states receive batch-local aligned storage. Nested
implementations manage construction/destruction, and the caller Arena
retains destination allocations. This avoids shared mutable scratch
buffers, unused scratch allocations and per-row range dispatch. The
Nullable selected fast path reads each selected row independently so
skipped destinations cannot shift the serialized input cursor.
Single-place merging and FE plans are unchanged. Q67 A/B performance has
not been measured.
Scalar finalization uses `deserialize_from_column_row()` to read
directly into its initialized empty state. Generic string-backed readers
and native serialized column readers implement the same contract, with
forwarding through nullable and state-combine wrappers. This removes the
extra temporary state allocation and merge previously performed by the
generic range interface for each finalized row; the caller retains
responsibility for state construction, destruction and exception
cleanup.
### Release note
Add `<aggregate>_finalize(AGG_STATE)` scalar functions to retrieve each
aggregate state's result without merging rows.
### Check List (For Author)
- Test
- [x] Regression test: test_agg_state_finalize on a fresh local ASAN BE
+ FE cluster; output generated by the standard runner, checked against
direct original aggregates, and verified by a normal comparison run.
- [x] Regression test: agg_state_finalize_roll_up verifies MV rejection
for differing outer aggregates/scalar arguments and selection for
matching rollups, including negative values and NULLs. Generated output
was checked independently; the normal comparison run of this suite and
test_agg_state_finalize passed.
- [x] Unit Test: 51 FE tests passed across CombinatorRollupHandlerTest,
CountTest, FinalizeCombinatorTest, StateCombinatorTest,
CombineCombinatorTest, and FunctionRegistryTest. The 9 new rollup tests
reproduced 6 failures before the fix and all pass after it. All 9
FunctionAggStateFinalizeTest cases passed under ASAN.
- [x] Manual test: compare grouped and empty AVG/COUNT/SUM/MIN/MAX,
decimal AVG and ARRAY_AGG with the original aggregates; verify rollup
AVG is 14/3 and distinguish a missing outer-join state from COUNT's
empty state.
- [ ] No need to test or manual test. Explain why:
- Behavior changed:
- [ ] No.
- [x] Yes. Add a family of scalar finalization functions; MV rollup
matching now preserves complete value expressions; existing
aggregate/state evaluation semantics are unchanged.
- Does this need documentation?
- [ ] No.
- [x] Yes. Usage and semantics are included in the function-combinator
README.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
### Additional validation
- Standard ASAN BE + FE build, FE Checkstyle, clang-format 16, build
hygiene and source whitespace checks passed.
- clang-tidy was attempted with the repository script and the
production/test translation units. New-code style and added
cognitive-complexity warnings were fixed. A fully clean run remains
blocked by existing header diagnostics and analyzer issues (including a
test-helper array-bound path that does not connect its asserted unary
arity to the input array); the local tool's crashing
`modernize-use-scoped-lock` check was disabled only for the supplemental
analysis. No repository analysis configuration was changed.
### Batch state merge validation
- `./build.sh --be -j 48` completed successfully under ASAN.
- 19 BE unit tests passed across AggregateStateUnionTest,
AggregateStateCombineTest and FunctionAggStateFinalizeTest. Six new
tests cover repeated groups, nonzero offsets, empty/all-unselected
batches, repeated batches, plain/Nullable/all-NULL SUM, Decimal SUM,
string ARRAY_AGG and nullable string MIN. The Nullable selected-row test
fails before its reader fix and passes afterward.
- `test_agg_state_union_batch` and `test_agg_state_finalize` both passed
in normal regression comparison mode. New expected output was generated
with the standard runner and independently checked against ordinary
aggregates, arithmetic sums, Decimal sums and lexicographic MIN. The new
suite also verifies that plans retain sum_merge/sum_union, and compares
ARRAY_AGG results after multiple input batches.
- clang-format 16.0.6, header hygiene and source whitespace checks
passed. clang-tidy was run via the repository script; a fully clean run
remains blocked by existing diagnostics in types.h (unmatched NOLINTEND)
and DataTypeAggState's constructor, plus existing signed/unsigned
comparisons in the Nullable wrapper. No unrelated analysis
configuration/source changes were made.
- Q67 A/B performance has not been measured.
### Scratch and direct-row reader validation
- 83 BE unit tests from 10 suites passed under ASAN, including
AggregateStateUnionTest, FunctionAggStateFinalizeTest,
AggregateFunctionExceptionTest and native aggregate reader suites.
Coverage includes scratch requirements, selected rows, nullable
wrappers, direct row reads, source-buffer lifetime and exception
cleanup.
- clang-format 16.0.6, header hygiene and source whitespace checks
passed. Repository clang-tidy checks produced no diagnostics on the
changed lines; a fully clean run remains blocked by the existing
unmatched NOLINTEND in `be/src/core/types.h`.
- No new SQL regression run or throughput benchmark was performed for
these two commits.
- The wide-state batch deserialization finding in [discussion
4217446484](https://github.com/apache/doris/pull/68312#discussion_r4217446484)
is intentionally left unchanged in this PR, following discussion. It is
not claimed as fixed by the scratch-allocation or direct-row
finalization changes.
---
be/src/exprs/aggregate/aggregate_function.h | 18 +
.../exprs/aggregate/aggregate_function_array_agg.h | 8 +
be/src/exprs/aggregate/aggregate_function_avg.h | 10 +
be/src/exprs/aggregate/aggregate_function_bitmap.h | 9 +
.../aggregate/aggregate_function_bitmap_agg.h | 9 +
be/src/exprs/aggregate/aggregate_function_count.h | 20 +
be/src/exprs/aggregate/aggregate_function_map_v2.h | 10 +
.../exprs/aggregate/aggregate_function_min_max.h | 17 +
be/src/exprs/aggregate/aggregate_function_null.h | 5 +-
.../exprs/aggregate/aggregate_function_null_v2.h | 19 +
.../aggregate/aggregate_function_state_combine.h | 10 +
.../aggregate/aggregate_function_state_union.h | 40 +-
be/src/exprs/aggregate/aggregate_function_sum.h | 10 +
.../aggregate_function_uniq_distribute_key.h | 8 +
be/src/exprs/aggregate/helpers.h | 6 +
.../exprs/function/function_agg_state_finalize.h | 110 +++++
be/src/exprs/vectorized_fn_call.cpp | 34 +-
be/test/exprs/aggregate/agg_state_union_test.cpp | 316 +++++++++++++
.../function/function_agg_state_finalize_test.cpp | 516 +++++++++++++++++++++
.../org/apache/doris/catalog/FunctionRegistry.java | 3 +-
.../glue/translator/ExpressionTranslator.java | 8 +
.../mv/rollup/AggFunctionRollUpHandler.java | 28 +-
.../mv/rollup/BothCombinatorRollupHandler.java | 4 +-
.../mv/rollup/SingleCombinatorRollupHandler.java | 2 +-
.../functions/AggCombinerFunctionBuilder.java | 14 +-
.../functions/combinator/FinalizeCombinator.java | 100 ++++
.../expressions/functions/combinator/readme.md | 22 +
.../expressions/visitor/ScalarFunctionVisitor.java | 5 +
.../mv/rollup/CombinatorRollupHandlerTest.java | 156 +++++++
.../combinator/FinalizeCombinatorTest.java | 127 +++++
.../agg_state/test_agg_state_finalize.out | 60 +++
.../agg_state/test_agg_state_union_batch.out | 37 ++
.../agg_state_finalize_roll_up.out | 36 ++
.../agg_state/test_agg_state_finalize.groovy | 104 +++++
.../agg_state/test_agg_state_union_batch.groovy | 79 ++++
.../agg_state_finalize_roll_up.groovy | 92 ++++
36 files changed, 2031 insertions(+), 21 deletions(-)
diff --git a/be/src/exprs/aggregate/aggregate_function.h
b/be/src/exprs/aggregate/aggregate_function.h
index ac780b7c025..a6c12dcbab5 100644
--- a/be/src/exprs/aggregate/aggregate_function.h
+++ b/be/src/exprs/aggregate/aggregate_function.h
@@ -166,9 +166,19 @@ public:
virtual void deserialize(AggregateDataPtr __restrict place,
BufferReadable& buf,
Arena&) const = 0;
+ /// Deserializes one row into an empty, initialized state. The caller owns
its lifecycle
+ /// and must keep the serialized column alive until the state is destroyed.
+ virtual void deserialize_from_column_row(AggregateDataPtr __restrict place,
+ const IColumn& column, size_t row,
+ Arena& arena) const = 0;
+
virtual void deserialize_vec(AggregateDataPtr places, const ColumnString*
column, Arena&,
size_t num_rows) const = 0;
+ // Returning false guarantees that both deserialize_and_merge_vec variants
accept nullptr
+ // for rhs and never access it. Generic readers require storage for
temporary states.
+ virtual bool needs_deserialize_and_merge_scratch() const { return true; }
+
virtual void deserialize_and_merge_vec(const AggregateDataPtr* places,
size_t offset,
AggregateDataPtr rhs, const
IColumn* column, Arena&,
const size_t num_rows) const = 0;
@@ -629,6 +639,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena& arena) const override {
+ DCHECK_LT(row, column.size());
+ const auto& strings = assert_cast<const ColumnString&>(column);
+ VectorBufferReader reader(strings.get_data_at(row));
+ assert_cast<const Derived*>(this)->deserialize(place, reader, arena);
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_array_agg.h
b/be/src/exprs/aggregate/aggregate_function_array_agg.h
index 5bf6d6fbee0..e4ae23d6240 100644
--- a/be/src/exprs/aggregate/aggregate_function_array_agg.h
+++ b/be/src/exprs/aggregate/aggregate_function_array_agg.h
@@ -353,6 +353,8 @@ public:
this->data(place).insert_result_into(to);
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
@@ -361,6 +363,12 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ this->data(place).deserialize_and_merge(column, row);
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_avg.h
b/be/src/exprs/aggregate/aggregate_function_avg.h
index 83eab79215e..01ac01b5196 100644
--- a/be/src/exprs/aggregate/aggregate_function_avg.h
+++ b/be/src/exprs/aggregate/aggregate_function_avg.h
@@ -273,6 +273,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
Data*>(col.get_data().data());
+ this->data(place) = data[row];
+ }
+
NO_SANITIZE_UNDEFINED void deserialize_and_merge_from_column_range(
AggregateDataPtr __restrict place, const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -287,6 +295,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_bitmap.h
b/be/src/exprs/aggregate/aggregate_function_bitmap.h
index e60ab9769d4..f65ab22b787 100644
--- a/be/src/exprs/aggregate/aggregate_function_bitmap.h
+++ b/be/src/exprs/aggregate/aggregate_function_bitmap.h
@@ -186,6 +186,13 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnBitmap&>(column);
+ this->data(place).merge(col.get_data()[row]);
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -198,6 +205,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena&,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_bitmap_agg.h
b/be/src/exprs/aggregate/aggregate_function_bitmap_agg.h
index 8d6af091066..75319df1287 100644
--- a/be/src/exprs/aggregate/aggregate_function_bitmap_agg.h
+++ b/be/src/exprs/aggregate/aggregate_function_bitmap_agg.h
@@ -172,6 +172,13 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnBitmap&>(column);
+ this->data(place).value = col.get_data()[row];
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -184,6 +191,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena&,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_count.h
b/be/src/exprs/aggregate/aggregate_function_count.h
index fb806419e9d..978343669d2 100644
--- a/be/src/exprs/aggregate/aggregate_function_count.h
+++ b/be/src/exprs/aggregate/aggregate_function_count.h
@@ -122,6 +122,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
Data*>(col.get_data().data());
+ AggregateFunctionCount::data(place) = data[row];
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -134,6 +142,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
@@ -286,6 +296,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
Data*>(col.get_data().data());
+ AggregateFunctionCountNotNullUnary::data(place) = data[row];
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -298,6 +316,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_map_v2.h
b/be/src/exprs/aggregate/aggregate_function_map_v2.h
index 47582fc6329..5e62dcdfb5c 100644
--- a/be/src/exprs/aggregate/aggregate_function_map_v2.h
+++ b/be/src/exprs/aggregate/aggregate_function_map_v2.h
@@ -243,6 +243,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnMap&>(column);
+ auto map = col[row].get<TYPE_MAP>();
+ this->data(place).add(map[0], map[1]);
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -255,6 +263,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena&,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max.h
b/be/src/exprs/aggregate/aggregate_function_min_max.h
index 416bf226196..12a87aa2497 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max.h
+++ b/be/src/exprs/aggregate/aggregate_function_min_max.h
@@ -230,6 +230,19 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena& arena) const override {
+ DCHECK_LT(row, column.size());
+ if constexpr (Data::UsesFixedLengthStateSerialization) {
+ const auto& col = assert_cast<const
ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
Data*>(col.get_data().data());
+ // An empty serialized state need not initialize its value payload.
+ this->data(place).change_if_better(data[row], arena);
+ } else {
+ Base::deserialize_from_column_row(place, column, row, arena);
+ }
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
@@ -246,6 +259,10 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override {
+ return !Data::UsesFixedLengthStateSerialization;
+ }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_null.h
b/be/src/exprs/aggregate/aggregate_function_null.h
index c4406417c21..25c97c1cd6c 100644
--- a/be/src/exprs/aggregate/aggregate_function_null.h
+++ b/be/src/exprs/aggregate/aggregate_function_null.h
@@ -244,13 +244,15 @@ public:
AggregateDataPtr rhs, const
IColumn* column,
Arena& arena, const size_t
num_rows) const override {
if (nested_function->is_trivial()) {
- BufferReadable buf({column->get_data_at(0).data, 0});
+ // Selected rows need independent readers: skipping a destination
must not leave
+ // a shared reader pointing at that row's serialized state for the
next destination.
size_t size_of_data = this->size_of_data();
if constexpr (result_is_nullable) {
for (int i = 0; i != num_rows; ++i) {
if (!places[i]) {
continue;
}
+ BufferReadable buf(column->get_data_at(i));
buf.read_binary(*(bool*)(rhs + size_of_data * i));
if (get_flag(rhs + size_of_data * i)) {
nested_function->deserialize(nested_place(rhs +
size_of_data * i), buf,
@@ -267,6 +269,7 @@ public:
} else {
for (size_t i = 0; i != num_rows; ++i) {
if (places[i]) {
+ BufferReadable buf(column->get_data_at(i));
nested_function->deserialize(rhs + size_of_data * i,
buf, arena);
}
}
diff --git a/be/src/exprs/aggregate/aggregate_function_null_v2.h
b/be/src/exprs/aggregate/aggregate_function_null_v2.h
index cb9d7dd5316..a9766e23818 100644
--- a/be/src/exprs/aggregate/aggregate_function_null_v2.h
+++ b/be/src/exprs/aggregate/aggregate_function_null_v2.h
@@ -302,6 +302,10 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override {
+ return nested_function->needs_deserialize_and_merge_scratch();
+ }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
@@ -342,6 +346,21 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena& arena) const override {
+ DCHECK_LT(row, column.size());
+ if constexpr (result_is_nullable) {
+ const auto& nullable_col = assert_cast<const
ColumnNullable&>(column);
+ if (!nullable_col.is_null_at(row)) {
+ set_flag(place);
+ nested_function->deserialize_from_column_row(
+ nested_place(place), nullable_col.get_nested_column(),
row, arena);
+ }
+ } else {
+ nested_function->deserialize_from_column_row(place, column, row,
arena);
+ }
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_state_combine.h
b/be/src/exprs/aggregate/aggregate_function_state_combine.h
index d21e0d363ac..c00524c9e58 100644
--- a/be/src/exprs/aggregate/aggregate_function_state_combine.h
+++ b/be/src/exprs/aggregate/aggregate_function_state_combine.h
@@ -139,6 +139,10 @@ public:
_function->deserialize_vec(places, column, arena, num_rows);
}
+ bool needs_deserialize_and_merge_scratch() const override {
+ return _function->needs_deserialize_and_merge_scratch();
+ }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
@@ -156,6 +160,12 @@ public:
_function->deserialize_and_merge(place, rhs, buf, arena);
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena& arena) const override {
+ DCHECK_LT(row, column.size());
+ _function->deserialize_from_column_row(place, column, row, arena);
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena& arena) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_state_union.h
b/be/src/exprs/aggregate/aggregate_function_state_union.h
index 1c8242bf53f..16b1b17592c 100644
--- a/be/src/exprs/aggregate/aggregate_function_state_union.h
+++ b/be/src/exprs/aggregate/aggregate_function_state_union.h
@@ -71,11 +71,12 @@ public:
void add_batch(size_t batch_size, AggregateDataPtr* places, size_t
place_offset,
const IColumn** columns, Arena& arena, bool agg_many) const
override {
- for (size_t i = 0; i < batch_size; ++i) {
- //the range is [i, i]
- _function->deserialize_and_merge_from_column_range(places[i] +
place_offset,
- *columns[0], i,
i, arena);
- }
+ add_batch_impl<false>(batch_size, places, place_offset, columns,
arena);
+ }
+
+ void add_batch_selected(size_t batch_size, AggregateDataPtr* places,
size_t place_offset,
+ const IColumn** columns, Arena& arena) const
override {
+ add_batch_impl<true>(batch_size, places, place_offset, columns, arena);
}
void reset(AggregateDataPtr place) const override {
_function->reset(place); }
@@ -107,6 +108,35 @@ public:
size_t align_of_data() const override { return _function->align_of_data();
}
+private:
+ template <bool selected>
+ void add_batch_impl(size_t batch_size, AggregateDataPtr* places, size_t
place_offset,
+ const IColumn** columns, Arena& arena) const {
+ // Some batch implementations read the first serialized row before
entering their loop.
+ if (batch_size == 0) {
+ return;
+ }
+ // Delegate the whole batch so native state columns (e.g. SUM) can
merge directly,
+ // instead of dispatching through the range interface for every input
row.
+ // Generic implementations need aligned scratch states and manage
their construction,
+ // destruction and exception cleanup. Release only this storage after
the batch;
+ // keep passing the caller's arena for allocations that merged states
may retain.
+ // The function can be shared by evaluators, so scratch must not be a
mutable member.
+ Arena scratch_arena;
+ AggregateDataPtr scratch = nullptr;
+ if (_function->needs_deserialize_and_merge_scratch()) {
+ scratch = scratch_arena.aligned_alloc(batch_size *
_function->size_of_data(),
+ _function->align_of_data());
+ }
+ if constexpr (selected) {
+ _function->deserialize_and_merge_vec_selected(places,
place_offset, scratch, columns[0],
+ arena, batch_size);
+ } else {
+ _function->deserialize_and_merge_vec(places, place_offset,
scratch, columns[0], arena,
+ batch_size);
+ }
+ }
+
protected:
AggregateFunctionPtr _function;
DataTypePtr _return_type;
diff --git a/be/src/exprs/aggregate/aggregate_function_sum.h
b/be/src/exprs/aggregate/aggregate_function_sum.h
index 8c00d1934ea..12b657f5d41 100644
--- a/be/src/exprs/aggregate/aggregate_function_sum.h
+++ b/be/src/exprs/aggregate/aggregate_function_sum.h
@@ -160,6 +160,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
Data*>(col.get_data().data());
+ this->data(place) = data[row];
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
@@ -172,6 +180,8 @@ public:
}
}
+ bool needs_deserialize_and_merge_scratch() const override { return false; }
+
void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
const size_t num_rows) const override {
diff --git a/be/src/exprs/aggregate/aggregate_function_uniq_distribute_key.h
b/be/src/exprs/aggregate/aggregate_function_uniq_distribute_key.h
index 7b3985ecd5a..9a3c9cffd46 100644
--- a/be/src/exprs/aggregate/aggregate_function_uniq_distribute_key.h
+++ b/be/src/exprs/aggregate/aggregate_function_uniq_distribute_key.h
@@ -211,6 +211,14 @@ public:
}
}
+ void deserialize_from_column_row(AggregateDataPtr __restrict place, const
IColumn& column,
+ size_t row, Arena&) const override {
+ DCHECK_LT(row, column.size());
+ const auto& col = assert_cast<const ColumnFixedLengthObject&>(column);
+ const auto* data = reinterpret_cast<const
UInt64*>(col.get_data().data());
+ this->data(place).count = data[row];
+ }
+
void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict
place,
const IColumn& column, size_t
begin, size_t end,
Arena&) const override {
diff --git a/be/src/exprs/aggregate/helpers.h b/be/src/exprs/aggregate/helpers.h
index eac02206e7c..48a971743de 100644
--- a/be/src/exprs/aggregate/helpers.h
+++ b/be/src/exprs/aggregate/helpers.h
@@ -36,6 +36,7 @@
* 4. deserialize_and_merge_vec_selected
* 5. serialize_without_key_to_column
* 6. deserialize_and_merge_from_column
+ * 7. deserialize_from_column_row
*/
#define CHECK_AGG_FUNCTION_SERIALIZED_TYPE(FunctionTemplate)
\
do {
\
@@ -43,6 +44,11 @@
!std::is_same_v<decltype(&FunctionTemplate::get_serialized_type),
\
decltype(&IAggregateFunction::get_serialized_type)>; \
if constexpr (_is_new_serialized_type) {
\
+ static_assert(
\
+
!std::is_same_v<decltype(&FunctionTemplate::deserialize_from_column_row), \
+ decltype(&IAggregateFunctionHelper<
\
+
FunctionTemplate>::deserialize_from_column_row)>, \
+ "need to override deserialize_from_column_row");
\
static_assert(!std::is_same_v<decltype(&FunctionTemplate::serialize_to_column),
\
decltype(&IAggregateFunctionHelper<
\
FunctionTemplate>::serialize_to_column)>, \
diff --git a/be/src/exprs/function/function_agg_state_finalize.h
b/be/src/exprs/function/function_agg_state_finalize.h
new file mode 100644
index 00000000000..aacdb65921c
--- /dev/null
+++ b/be/src/exprs/function/function_agg_state_finalize.h
@@ -0,0 +1,110 @@
+// 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.
+
+#pragma once
+
+#include <cstring>
+
+#include "core/arena.h"
+#include "core/block/block.h"
+#include "core/column/column_nullable.h"
+#include "exprs/aggregate/aggregate_function.h"
+#include "exprs/function/function.h"
+#include "util/defer_op.h"
+
+namespace doris {
+
+class FunctionAggStateFinalize : public IFunction {
+public:
+ FunctionAggStateFinalize(DataTypePtr return_type, AggregateFunctionPtr
agg_function)
+ : _return_type(std::move(return_type)),
_agg_function(std::move(agg_function)) {}
+
+ static FunctionBasePtr create(const DataTypes& argument_types, const
DataTypePtr& return_type,
+ const AggregateFunctionPtr& agg_function) {
+ return std::make_shared<DefaultFunction>(
+ std::make_shared<FunctionAggStateFinalize>(return_type,
agg_function),
+ argument_types, return_type);
+ }
+
+ String get_name() const override { return _agg_function->get_name() +
"_finalize"; }
+
+ size_t get_number_of_arguments() const override { return 1; }
+
+ // An outer NULL can have an empty, invalid serialized payload. Do not
deserialize it.
+ bool use_default_implementation_for_nulls() const override { return false;
}
+
+ DataTypePtr get_return_type_impl(const DataTypes& arguments) const
override {
+ return _return_type;
+ }
+
+ Status execute_impl(FunctionContext* context, Block& block, const
ColumnNumbers& arguments,
+ uint32_t result, size_t input_rows_count) const
override {
+ const auto input =
+
block.get_by_position(arguments[0]).column->convert_to_full_column_if_const();
+ const auto* nullable = check_and_get_column<ColumnNullable>(*input);
+ const auto& states = nullable ? nullable->get_nested_column() : *input;
+ auto output = _agg_function->get_return_type()->create_column();
+ _agg_function->check_result_column_type(*output);
+ output->reserve(input_rows_count);
+ // Keep the reusable state storage separate from its per-row
variable-length data.
+ Arena state_arena;
+ const auto state_size = _agg_function->size_of_data();
+ auto* place = state_arena.aligned_alloc(state_size,
_agg_function->align_of_data());
+ Arena arena;
+ auto finalize_row = [&](size_t row) {
+ // Serialized states can use native columns as well as strings.
+ _agg_function->deserialize_from_column_row(place, states, row,
arena);
+ _agg_function->insert_result_into(place, *output);
+ };
+ auto finalize_rows = [&]<bool is_trivial>() {
+ for (size_t row = 0; row < input_rows_count; ++row) {
+ if (nullable && nullable->is_null_at(row)) {
+ output->insert_default();
+ continue;
+ }
+ if constexpr (is_trivial) {
+ // Trivial states support zero-init and need no
destruction.
+ std::memset(place, 0, state_size);
+ finalize_row(row);
+ } else {
+ _agg_function->create(place);
+ DEFER(_agg_function->destroy(place));
+ finalize_row(row);
+ }
+ // Destroy states before reclaiming any variable-length data
they own.
+ arena.clear();
+ }
+ };
+ if (_agg_function->is_trivial()) {
+ finalize_rows.template operator()<true>();
+ } else {
+ finalize_rows.template operator()<false>();
+ }
+ ColumnPtr result_column = std::move(output);
+ if (_return_type->is_nullable()) {
+ result_column = wrap_in_nullable(result_column, block, arguments,
input_rows_count);
+ }
+ block.replace_by_position(result, std::move(result_column));
+ return Status::OK();
+ }
+
+private:
+ DataTypePtr _return_type;
+ AggregateFunctionPtr _agg_function;
+};
+
+} // namespace doris
diff --git a/be/src/exprs/vectorized_fn_call.cpp
b/be/src/exprs/vectorized_fn_call.cpp
index a908fea81d3..eeab5a03fe1 100644
--- a/be/src/exprs/vectorized_fn_call.cpp
+++ b/be/src/exprs/vectorized_fn_call.cpp
@@ -49,6 +49,7 @@
#include "exec/pipeline/pipeline_task.h"
#include "exprs/function/array/function_array_distance.h"
#include "exprs/function/function_agg_state.h"
+#include "exprs/function/function_agg_state_finalize.h"
#include "exprs/function/function_fake.h"
#include "exprs/function/function_java_udf.h"
#include "exprs/function/function_python_udf.h"
@@ -79,6 +80,7 @@ class TExprNode;
namespace doris {
const std::string AGG_STATE_SUFFIX = "_state";
+const std::string AGG_FINALIZE_SUFFIX = "_finalize";
// Now left child is a function call, we need to check if it is a distance
function
const static std::set<std::string> DISTANCE_FUNCS =
{L2DistanceApproximate::name,
@@ -216,6 +218,33 @@ size_t raw_comparison_value_size(PrimitiveType
primitive_type) {
}
}
+Status create_agg_state_finalize_function(const std::string& function_name,
+ const DataTypes& argument_types,
+ const DataTypePtr& return_type,
+ FunctionBasePtr& function) {
+ if (argument_types.size() != 1 ||
+ remove_nullable(argument_types[0])->get_primitive_type() !=
TYPE_AGG_STATE) {
+ return Status::InternalError("Finalize function requires one AGG_STATE
argument");
+ }
+ const auto state_type = remove_nullable(argument_types[0]);
+ const auto* agg_state = assert_cast<const
DataTypeAggState*>(state_type.get());
+ if (agg_state->get_function_name() + AGG_FINALIZE_SUFFIX != function_name)
{
+ return Status::InternalError("{} does not match function {}",
state_type->get_name(),
+ function_name);
+ }
+ const auto& nested = agg_state->get_nested_function();
+ auto expected_type = nested->get_return_type();
+ if (argument_types[0]->is_nullable()) {
+ expected_type = make_nullable(expected_type);
+ }
+ if (!expected_type->equals(*return_type)) {
+ return Status::InternalError("{} expects return type {}, but got {}",
function_name,
+ expected_type->get_name(),
return_type->get_name());
+ }
+ function = FunctionAggStateFinalize::create(argument_types, return_type,
nested);
+ return Status::OK();
+}
+
} // namespace
VectorizedFnCall::VectorizedFnCall(const TExprNode& node) : VExpr(node) {
@@ -294,8 +323,11 @@ Status VectorizedFnCall::prepare(RuntimeState* state,
const RowDescriptor& desc,
_function = FunctionAggState::create(
argument_types, _data_type,
assert_cast<const
DataTypeAggState*>(_data_type.get())->get_nested_function());
+ } else if (match_suffix(_fn.name.function_name, AGG_FINALIZE_SUFFIX)) {
+ RETURN_IF_ERROR(create_agg_state_finalize_function(
+ _fn.name.function_name, argument_types, _data_type,
_function));
} else {
- return Status::InternalError("Function {} is not endwith
'_state'", _fn.signature);
+ return Status::InternalError("Unsupported AggState function {}",
_fn.signature);
}
} else {
// get the function. won't prepare function.
diff --git a/be/test/exprs/aggregate/agg_state_union_test.cpp
b/be/test/exprs/aggregate/agg_state_union_test.cpp
new file mode 100644
index 00000000000..0a712b860ab
--- /dev/null
+++ b/be/test/exprs/aggregate/agg_state_union_test.cpp
@@ -0,0 +1,316 @@
+// 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.
+
+#include <gtest/gtest.h>
+
+#include <array>
+
+#include "agent/be_exec_version_manager.h"
+#include "core/column/column_complex.h"
+#include "core/data_type/data_type_bitmap.h"
+#include "core/data_type/data_type_decimal.h"
+#include "core/data_type/data_type_string.h"
+#include "exprs/aggregate/aggregate_function_state_combine.h"
+#include "exprs/aggregate/aggregate_function_state_merge.h"
+#include "exprs/aggregate/aggregate_function_state_union.h"
+#include "testutil/column_helper.h"
+
+namespace doris {
+
+// Reuse the state lifecycle forwarding while inspecting the scratch passed by
UNION/MERGE.
+class ScratchCheckingAggregateFunction final : public AggregateStateUnion {
+public:
+ using AggregateStateUnion::AggregateStateUnion;
+
+ bool needs_deserialize_and_merge_scratch() const override {
+ return _function->needs_deserialize_and_merge_scratch();
+ }
+
+ void deserialize_and_merge_vec(const AggregateDataPtr* places, size_t
offset,
+ AggregateDataPtr rhs, const IColumn*
column, Arena& arena,
+ size_t num_rows) const override {
+ check_scratch(rhs);
+ ++batch_calls;
+ _function->deserialize_and_merge_vec(places, offset, rhs, column,
arena, num_rows);
+ }
+
+ void deserialize_and_merge_vec_selected(const AggregateDataPtr* places,
size_t offset,
+ AggregateDataPtr rhs, const
IColumn* column,
+ Arena& arena, size_t num_rows)
const override {
+ check_scratch(rhs);
+ ++selected_batch_calls;
+ _function->deserialize_and_merge_vec_selected(places, offset, rhs,
column, arena, num_rows);
+ }
+
+ mutable size_t batch_calls = 0;
+ mutable size_t selected_batch_calls = 0;
+
+private:
+ void check_scratch(ConstAggregateDataPtr rhs) const {
+ EXPECT_EQ(rhs != nullptr, needs_deserialize_and_merge_scratch());
+ EXPECT_EQ(reinterpret_cast<uintptr_t>(rhs) % align_of_data(), 0);
+ }
+};
+
+class AggregateStateUnionTest : public testing::Test {
+protected:
+ static constexpr size_t GROUPS = 3;
+ using Places = std::array<AggregateDataPtr, GROUPS>;
+
+ static Places create_places(const AggregateFunctionPtr& function, Arena&
arena, size_t offset) {
+ Places places;
+ for (auto& place : places) {
+ place = arena.aligned_alloc(offset + function->size_of_data(),
+ function->align_of_data());
+ memset(place, 0x5a, offset);
+ function->create(place + offset);
+ }
+ return places;
+ }
+
+ static void compare_results(const AggregateFunctionPtr& function, const
Places& actual,
+ const Places& expected, size_t offset) {
+ auto actual_result = function->get_return_type()->create_column();
+ auto expected_result = function->get_return_type()->create_column();
+ for (size_t group = 0; group < GROUPS; ++group) {
+ function->insert_result_into(actual[group] + offset,
*actual_result);
+ function->insert_result_into(expected[group] + offset,
*expected_result);
+ EXPECT_EQ(actual_result->compare_at(group, group,
*expected_result, 1), 0)
+ << "group=" << group;
+ EXPECT_EQ(std::string(actual[group], offset), std::string(offset,
0x5a));
+ }
+ }
+
+ static void check_batches(const AggregateFunctionPtr& function,
+ const AggregateFunctionPtr& nested, const
IColumn& states,
+ bool selected) {
+ SCOPED_TRACE(selected ? "selected" : "all rows");
+ Arena arena;
+ const size_t offset = 2 * function->align_of_data();
+ auto actual = create_places(function, arena, offset);
+ auto expected = create_places(function, arena, offset);
+ DEFER({
+ for (size_t group = 0; group < GROUPS; ++group) {
+ function->destroy(actual[group] + offset);
+ function->destroy(expected[group] + offset);
+ }
+ });
+ // Repeated destinations, nonzero offsets, and a NULL-only group are
intentional.
+ const std::array<size_t, 6> groups = {0, 1, 0, 2, 0, 1};
+ std::array<AggregateDataPtr, 6> rows;
+ for (size_t row = 0; row < rows.size(); ++row) {
+ rows[row] = selected && (row == 0 || row == 2) ? nullptr :
actual[groups[row]];
+ }
+ const IColumn* columns[] = {&states};
+ // Empty batches must not inspect row zero, including the Nullable
fast path.
+ auto empty = states.clone_empty();
+ const IColumn* empty_columns[] = {empty.get()};
+ function->add_batch(0, rows.data(), offset, empty_columns, arena,
false);
+ function->add_batch_selected(0, rows.data(), offset, empty_columns,
arena);
+ for (int batch = 0; batch < 3; ++batch) {
+ if (selected) {
+ function->add_batch_selected(states.size(), rows.data(),
offset, columns, arena);
+ } else {
+ function->add_batch(states.size(), rows.data(), offset,
columns, arena, false);
+ }
+ for (size_t row = 0; row < rows.size(); ++row) {
+ if (rows[row]) {
+
nested->deserialize_and_merge_from_column_range(expected[groups[row]] + offset,
+ states,
row, row, arena);
+ }
+ }
+ // Read results after the batch scratch arena has been destroyed.
+ compare_results(nested, actual, expected, offset);
+ }
+ rows.fill(nullptr);
+ function->add_batch_selected(states.size(), rows.data(), offset,
columns, arena);
+ compare_results(nested, actual, expected, offset);
+ }
+
+ static void check(const std::string& name, const DataTypePtr& argument,
const ColumnPtr& input,
+ bool result_nullable = true) {
+ auto type = std::make_shared<DataTypeAggState>(DataTypes {argument},
result_nullable, name,
+
BeExecVersionManager::get_newest_version());
+ auto nested = type->get_nested_function();
+ check_function(nested, {input}, type);
+ }
+
+ static void check_function(const AggregateFunctionPtr& nested, const
Columns& inputs,
+ const DataTypePtr& state_type) {
+ auto states = nested->create_serialize_column();
+ std::vector<const IColumn*> columns;
+ for (const auto& input : inputs) {
+ columns.push_back(input.get());
+ }
+ Arena arena;
+ // Serialize independently built states, as produced by *_state or
*_combine.
+ for (size_t row = 0; row < inputs[0]->size(); ++row) {
+ auto* place = arena.aligned_alloc(nested->size_of_data(),
nested->align_of_data());
+ nested->create(place);
+ DEFER(nested->destroy(place));
+ nested->add(place, columns.data(), row, arena);
+ auto serialized = nested->create_serialize_column();
+ nested->serialize_without_key_to_column(place, *serialized);
+ states->insert_range_from(*serialized, 0, 1);
+ }
+ for (bool combine : {false, true}) {
+ SCOPED_TRACE(combine ? "combine reader" : "direct reader");
+ auto reader = combine ? AggregateStateCombine::create(
+ nested,
nested->get_argument_types(), state_type)
+ : nested;
+ auto checked = std::make_shared<ScratchCheckingAggregateFunction>(
+ reader, DataTypes {state_type}, state_type);
+ EXPECT_EQ(checked->needs_deserialize_and_merge_scratch(),
+ nested->needs_deserialize_and_merge_scratch());
+ for (bool merge : {false, true}) {
+ SCOPED_TRACE(merge ? "merge" : "union");
+ auto function =
+ merge ? AggregateStateMerge::create(checked,
{state_type},
+
nested->get_return_type())
+ : AggregateStateUnion::create(checked,
{state_type}, state_type);
+ check_batches(function, nested, *states, false);
+ check_batches(function, nested, *states, true);
+ }
+ EXPECT_GT(checked->batch_calls, 0);
+ EXPECT_GT(checked->selected_batch_calls, 0);
+ }
+ }
+
+ static void check_scratch_requirement(const std::string& name, const
DataTypes& arguments,
+ const Columns& inputs, bool
needs_scratch,
+ bool result_nullable = false, bool
null_v2 = false) {
+ SCOPED_TRACE(name);
+ AggregateFunctionAttr attr;
+ attr.enable_aggregate_function_null_v2 = null_v2;
+ auto nested = AggregateFunctionSimpleFactory::instance().get(
+ name, arguments, nullptr, result_nullable,
+ BeExecVersionManager::get_newest_version(), attr);
+ ASSERT_NE(nested, nullptr);
+ nested->set_version(BeExecVersionManager::get_newest_version());
+ ASSERT_EQ(nested->needs_deserialize_and_merge_scratch(),
needs_scratch);
+ check_function(nested, inputs, nested->get_serialized_type());
+ }
+};
+
+TEST_F(AggregateStateUnionTest, Sum) {
+ check("sum", std::make_shared<DataTypeInt64>(),
+ ColumnHelper::create_column<DataTypeInt64>({2, 50, 7, 0, 11, 19}));
+}
+
+TEST_F(AggregateStateUnionTest, NullableSum) {
+ check("sum", make_nullable(std::make_shared<DataTypeInt64>()),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({2, 50, 7, 0,
11, 19},
+ {0, 0, 0, 1, 0,
0}));
+}
+
+TEST_F(AggregateStateUnionTest, AllNullSum) {
+ check("sum", make_nullable(std::make_shared<DataTypeInt64>()),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({0, 0, 0, 0, 0,
0},
+ {1, 1, 1, 1, 1,
1}));
+}
+
+TEST_F(AggregateStateUnionTest, DecimalSum) {
+ auto type = std::make_shared<DataTypeDecimal128>(27, 9);
+ auto input = type->create_column();
+ auto& values = assert_cast<ColumnDecimal128V3&>(*input);
+ for (Int128 value : {2, 50, 7, 0, 11, 19}) {
+ values.insert_value(Decimal128V3(value * 1000000000));
+ }
+ check("sum", type, std::move(input));
+}
+
+TEST_F(AggregateStateUnionTest, StringArrayStateLifetime) {
+ check("array_agg", std::make_shared<DataTypeString>(),
+ ColumnHelper::create_column<DataTypeString>(
+ {std::string(4096, 'a'), "b", "c", "", std::string(8192,
'd'), "e"}),
+ false);
+}
+
+TEST_F(AggregateStateUnionTest, NullableStringMinStateLifetime) {
+ check("min", make_nullable(std::make_shared<DataTypeString>()),
+ ColumnHelper::create_nullable_column<DataTypeString>(
+ {std::string(4096, 'a'), "z", "c", "", std::string(8192,
'b'), "e"},
+ {0, 0, 0, 1, 0, 0}));
+}
+
+TEST_F(AggregateStateUnionTest, NativeNumericReadersNeedNoScratch) {
+ auto type = std::make_shared<DataTypeInt64>();
+ auto input = ColumnHelper::create_column<DataTypeInt64>({2, 50, 7, 0, 11,
19});
+ for (const auto* name : {"sum", "avg", "count", "min", "max",
"bitmap_agg"}) {
+ check_scratch_requirement(name, {type}, {input}, false);
+ }
+ check_scratch_requirement("count", {make_nullable(type)},
+
{ColumnHelper::create_nullable_column<DataTypeInt64>(
+ {2, 50, 7, 0, 11, 19}, {0, 0, 0, 1, 0,
0})},
+ false);
+}
+
+TEST_F(AggregateStateUnionTest, NativeComplexReadersNeedNoScratch) {
+ auto type = std::make_shared<DataTypeString>();
+ const std::vector<std::string> values = {std::string(4096, 'a'), "b", "c",
"",
+ std::string(8192, 'd'), "e"};
+ auto input = ColumnHelper::create_column<DataTypeString>(values);
+ // Nullable inputs select the native array reader; non-nullable inputs
select collect_list.
+ check_scratch_requirement(
+ "array_agg", {make_nullable(type)},
+ {ColumnHelper::create_nullable_column<DataTypeString>(values, {0,
0, 0, 1, 0, 0})},
+ false);
+ check_scratch_requirement("map_agg_v2", {type, type}, {input, input},
false);
+ auto bitmap_type = std::make_shared<DataTypeBitMap>();
+ auto bitmaps = ColumnBitmap::create();
+ for (UInt64 value : {2, 50, 7, 0, 11, 19}) {
+ bitmaps->insert_value(BitmapValue(value));
+ }
+ check_scratch_requirement("bitmap_union", {bitmap_type},
{std::move(bitmaps)}, false);
+}
+
+TEST_F(AggregateStateUnionTest, GenericAndStringMinMaxNeedScratch) {
+ auto type = std::make_shared<DataTypeString>();
+ auto input = ColumnHelper::create_column<DataTypeString>(
+ {std::string(4096, 'a'), "z", "c", "", std::string(8192, 'b'),
"e"});
+ for (const auto* name : {"min", "max"}) {
+ check_scratch_requirement(name, {type}, {input}, true);
+ }
+ check_scratch_requirement("array_agg", {type}, {input}, true);
+}
+
+TEST_F(AggregateStateUnionTest, NullableReadersPreserveScratchRequirement) {
+ auto type = make_nullable(std::make_shared<DataTypeInt64>());
+ for (bool all_null : {false, true}) {
+ auto input = ColumnHelper::create_nullable_column<DataTypeInt64>(
+ {2, 50, 7, 0, 11, 19}, all_null ? std::vector<UInt8> {1, 1, 1,
1, 1, 1}
+ : std::vector<UInt8> {0, 0, 0,
1, 0, 0});
+ for (bool null_v2 : {false, true}) {
+ for (bool result_nullable : {false, true}) {
+ check_scratch_requirement("sum", {type}, {input}, !null_v2,
result_nullable,
+ null_v2);
+ }
+ // The generic serialized-string reader still needs scratch inside
Nullable V2.
+ check_scratch_requirement("multi_distinct_count", {type}, {input},
true, false,
+ null_v2);
+ }
+ }
+}
+
+TEST_F(AggregateStateUnionTest, FixedLengthColumnCanStillNeedScratch) {
+ // This reader reconstructs a hash-set state from each fixed-width
serialized count.
+ check_scratch_requirement(
+ "multi_distinct_count_distribute_key",
{std::make_shared<DataTypeInt64>()},
+ {ColumnHelper::create_column<DataTypeInt64>({2, 50, 7, 0, 11,
19})}, true);
+}
+
+} // namespace doris
diff --git a/be/test/exprs/function/function_agg_state_finalize_test.cpp
b/be/test/exprs/function/function_agg_state_finalize_test.cpp
new file mode 100644
index 00000000000..77d2a0dc0fb
--- /dev/null
+++ b/be/test/exprs/function/function_agg_state_finalize_test.cpp
@@ -0,0 +1,516 @@
+// 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.
+
+#include "exprs/function/function_agg_state_finalize.h"
+
+#include <gtest/gtest.h>
+
+#include <algorithm>
+
+#include "agent/be_exec_version_manager.h"
+#include "core/column/column_array.h"
+#include "core/column/column_complex.h"
+#include "core/column/column_const.h"
+#include "core/column/column_string.h"
+#include "core/data_type/data_type_agg_state.h"
+#include "core/data_type/data_type_bitmap.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+#include "exprs/aggregate/aggregate_function_state_combine.h"
+#include "exprs/aggregate/aggregate_function_state_merge.h"
+#include "testutil/column_helper.h"
+
+namespace doris {
+
+class FunctionAggStateFinalizeTest : public testing::Test {
+protected:
+ static std::shared_ptr<DataTypeAggState> state_type(const std::string&
name,
+ const DataTypePtr&
argument_type,
+ bool result_nullable =
true) {
+ return std::make_shared<DataTypeAggState>(DataTypes {argument_type},
result_nullable, name,
+
BeExecVersionManager::get_newest_version());
+ }
+
+ static void append_state(const AggregateFunctionPtr& function, const
ColumnPtr& input,
+ IColumn& states) {
+ ASSERT_EQ(function->get_argument_types().size(), 1);
+ Arena arena;
+ auto* place = arena.aligned_alloc(function->size_of_data(),
function->align_of_data());
+ function->create(place);
+ DEFER(function->destroy(place));
+ const IColumn* columns[] = {input.get()};
+ function->check_input_columns_type(columns);
+ function->add_batch_single_place(input->size(), place, columns, arena);
+ // Some aggregates resize their no-key output instead of appending to
it.
+ auto serialized = function->create_serialize_column();
+ function->serialize_without_key_to_column(place, *serialized);
+ ASSERT_EQ(serialized->size(), 1);
+ states.insert_range_from(*serialized, 0, 1);
+ }
+
+ static ColumnPtr finalize(const DataTypePtr& type, const ColumnPtr&
states) {
+ auto nested = assert_cast<const
DataTypeAggState*>(remove_nullable(type).get())
+ ->get_nested_function();
+ auto result_type = nested->get_return_type();
+ if (type->is_nullable()) {
+ result_type = make_nullable(result_type);
+ }
+ auto function = FunctionAggStateFinalize::create({type}, result_type,
nested);
+ Block block {{states, type, "state"}, {nullptr, result_type,
"result"}};
+ EXPECT_TRUE(function->execute(nullptr, block, {0}, 1,
states->size()).ok());
+ return block.get_by_position(1).column;
+ }
+
+ static void check_row_reader(const std::string& name, const DataTypes&
arguments,
+ const Columns& inputs, bool result_nullable =
false,
+ bool null_v2 = false) {
+ SCOPED_TRACE(name);
+ SCOPED_TRACE(null_v2);
+ AggregateFunctionAttr attr;
+ attr.enable_aggregate_function_null_v2 = null_v2;
+ const auto version = BeExecVersionManager::get_newest_version();
+ auto nested = AggregateFunctionSimpleFactory::instance().get(
+ name, arguments, nullptr, result_nullable, version, attr);
+ ASSERT_NE(nested, nullptr);
+ nested->set_version(version);
+ auto actual = nested->get_return_type()->create_column();
+ auto expected = actual->clone_empty();
+ nested->check_result_column_type(*actual);
+ nested->check_result_column_type(*expected);
+ {
+ auto states = nested->create_serialize_column();
+ std::vector<const IColumn*> columns;
+ for (const auto& input : inputs) {
+ ASSERT_EQ(input->size(), 6);
+ columns.push_back(input.get());
+ }
+ Arena arena;
+ for (size_t group = 0; group < 4; ++group) {
+ AggregateFunctionGuard state(nested.get());
+ // Three independent groups followed by an empty state.
+ if (group < 3) {
+ nested->check_input_columns_type(columns.data());
+ nested->add(state.data(), columns.data(), group * 2,
arena);
+ nested->add(state.data(), columns.data(), group * 2 + 1,
arena);
+ }
+ auto serialized = nested->create_serialize_column();
+ nested->serialize_without_key_to_column(state.data(),
*serialized);
+ states->insert_range_from(*serialized, 0, 1);
+ }
+ for (bool combine : {false, true}) {
+ auto reader = combine ? AggregateStateCombine::create(nested,
arguments,
+
nested->get_serialized_type())
+ : nested;
+ for (size_t row : {2, 0, 3, 1, 0}) {
+ // Read out of order and repeat a row to catch input
mutation.
+ AggregateFunctionGuard direct(reader.get());
+ AggregateFunctionGuard merged(nested.get());
+ reader->deserialize_from_column_row(direct.data(),
*states, row, arena);
+
nested->deserialize_and_merge_from_column_range(merged.data(), *states, row,
+ row,
arena);
+ nested->insert_result_into(direct.data(), *actual);
+ nested->insert_result_into(merged.data(), *expected);
+ }
+ }
+ }
+ // Results must survive both the source column and the deserialized
states.
+ ASSERT_EQ(actual->size(), 10);
+ for (size_t row = 0; row < actual->size(); ++row) {
+ EXPECT_EQ(actual->compare_at(row, row, *expected, 1), 0) << "row="
<< row;
+ }
+ }
+};
+
+struct alignas(64) FinalizeTrackingState {
+ Int64 value = 0;
+};
+
+class FinalizeTrackingAggregateFunction final
+ : public IAggregateFunctionDataHelper<FinalizeTrackingState,
+
FinalizeTrackingAggregateFunction> {
+public:
+ FinalizeTrackingAggregateFunction()
+ : IAggregateFunctionDataHelper<FinalizeTrackingState,
+
FinalizeTrackingAggregateFunction>({}) {}
+
+ String get_name() const override { return "finalize_tracking"; }
+ DataTypePtr get_return_type() const override { return
std::make_shared<DataTypeInt64>(); }
+
+ void create(AggregateDataPtr place) const override {
+ EXPECT_EQ(reinterpret_cast<uintptr_t>(place) % align_of_data(), 0);
+ IAggregateFunctionDataHelper::create(place);
+ ++created;
+ }
+ void destroy(AggregateDataPtr place) const noexcept override {
+ IAggregateFunctionDataHelper::destroy(place);
+ ++destroyed;
+ }
+ void add(AggregateDataPtr, const IColumn**, ssize_t, Arena&) const
override {}
+ void merge(AggregateDataPtr, ConstAggregateDataPtr, Arena&) const override
{ ++merged; }
+ void serialize(ConstAggregateDataPtr place, BufferWritable& buf) const
override {
+ buf.write_binary(data(place).value);
+ }
+ void deserialize(AggregateDataPtr place, BufferReadable& buf, Arena&)
const override {
+ ++deserialized;
+ buf.read_binary(data(place).value);
+ if (data(place).value == -1) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "mock deserialize
failure");
+ }
+ }
+ void insert_result_into(ConstAggregateDataPtr place, IColumn& to) const
override {
+ if (data(place).value == -2) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "mock result insertion
failure");
+ }
+ assert_cast<ColumnInt64&>(to).insert_value(data(place).value);
+ }
+
+ void check_counts(size_t rows) const {
+ EXPECT_EQ(created, rows);
+ EXPECT_EQ(destroyed, rows);
+ EXPECT_EQ(deserialized, rows);
+ EXPECT_EQ(merged, 0);
+ }
+
+ mutable size_t created = 0;
+ mutable size_t destroyed = 0;
+ mutable size_t deserialized = 0;
+ mutable size_t merged = 0;
+};
+
+TEST_F(FunctionAggStateFinalizeTest, UsesOneAlignedStateWithoutMerging) {
+ auto aggregate = std::make_shared<FinalizeTrackingAggregateFunction>();
+ FunctionAggStateFinalize function(aggregate->get_return_type(), aggregate);
+ auto states = ColumnString::create();
+ VectorBufferWriter writer(*states);
+ constexpr Int64 rows = 128;
+ for (Int64 row = 0; row < rows; ++row) {
+ writer.write_binary(row);
+ writer.commit();
+ }
+ Block block {{std::move(states), aggregate->get_serialized_type(),
"state"},
+ {nullptr, aggregate->get_return_type(), "result"}};
+ ASSERT_TRUE(function.execute_impl(nullptr, block, {0}, 1, rows).ok());
+ aggregate->check_counts(rows);
+ for (Int64 row = 0; row < rows; ++row) {
+ EXPECT_EQ(block.get_by_position(1).column->get_int(row), row);
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest,
DestroysStateOnDeserializeAndResultFailure) {
+ for (Int64 failure : {-1, -2}) {
+ auto aggregate = std::make_shared<FinalizeTrackingAggregateFunction>();
+ FunctionAggStateFinalize function(aggregate->get_return_type(),
aggregate);
+ auto states = ColumnString::create();
+ VectorBufferWriter writer(*states);
+ writer.write_binary(Int64(7));
+ writer.commit();
+ writer.write_binary(failure);
+ writer.commit();
+ Block block {{std::move(states), aggregate->get_serialized_type(),
"state"},
+ {nullptr, aggregate->get_return_type(), "result"}};
+ EXPECT_THROW(static_cast<void>(function.execute_impl(nullptr, block,
{0}, 1, 2)),
+ Exception);
+ aggregate->check_counts(2);
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest, SingleRowNativeReadersMatchMerge) {
+ auto number = std::make_shared<DataTypeInt64>();
+ auto values = ColumnHelper::create_column<DataTypeInt64>({2, 4, 7, 7, 11,
19});
+ for (const auto* name : {"sum", "avg", "min", "max", "count", "bitmap_agg",
+ "multi_distinct_count_distribute_key"}) {
+ check_row_reader(name, {number}, {values});
+ }
+ auto string = std::make_shared<DataTypeString>();
+ auto strings = ColumnHelper::create_column<DataTypeString>(
+ {std::string(8192, 'x'), "b", "c", "c", "d", "e"});
+ check_row_reader("map_agg_v2", {string, string}, {strings, strings});
+ auto bitmap_type = std::make_shared<DataTypeBitMap>();
+ auto bitmaps = ColumnBitmap::create();
+ for (UInt64 value : {2, 4, 7, 7, 11, 19}) {
+ bitmaps->insert_value(BitmapValue(value));
+ }
+ ColumnPtr bitmap_input = std::move(bitmaps);
+ for (const auto* name : {"bitmap_union", "bitmap_intersect",
"group_bitmap_xor"}) {
+ check_row_reader(name, {bitmap_type}, {bitmap_input});
+ }
+ check_row_reader(
+ "array_agg", {make_nullable(string)},
+ {ColumnHelper::create_nullable_column<DataTypeString>(
+ {std::string(8192, 'x'), "b", "c", "", "d", "e"}, {0, 0,
0, 1, 0, 0})});
+}
+
+TEST_F(FunctionAggStateFinalizeTest, SingleRowNullableReadersMatchMerge) {
+ auto type = make_nullable(std::make_shared<DataTypeInt64>());
+ auto input = ColumnHelper::create_nullable_column<DataTypeInt64>({2, 4, 0,
0, 0, 19},
+ {0, 0, 1,
1, 1, 0});
+ for (bool null_v2 : {false, true}) {
+ for (bool result_nullable : {false, true}) {
+ for (const auto* name : {"sum", "avg", "min", "max"}) {
+ check_row_reader(name, {type}, {input}, result_nullable,
null_v2);
+ }
+ }
+ check_row_reader("count", {type}, {input}, false, null_v2);
+ // Nullable V2 can also wrap a generic string-backed state.
+ check_row_reader("multi_distinct_count", {type}, {input}, false,
null_v2);
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest, SingleRowStringReadersMatchMerge) {
+ auto type = std::make_shared<DataTypeString>();
+ auto input = ColumnHelper::create_column<DataTypeString>(
+ {std::string(8192, 'x'), "b", "c", "c", "d", "e"});
+ for (const auto* name : {"min", "max", "array_agg"}) {
+ check_row_reader(name, {type}, {input});
+ }
+ auto limit_type = std::make_shared<DataTypeInt32>();
+ auto limits = ColumnHelper::create_column<DataTypeInt32>({2, 2, 2, 2, 2,
2});
+ check_row_reader("topn", {type, limit_type}, {input, limits});
+}
+
+TEST_F(FunctionAggStateFinalizeTest, FinalizesEachRowIndependently) {
+ auto type = state_type("avg", std::make_shared<DataTypeInt64>());
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({1, 3}),
+ *states);
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({10}),
+ *states);
+ ColumnPtr serialized = std::move(states);
+ auto result = finalize(type, serialized);
+ const auto& values = assert_cast<const ColumnFloat64&>(*result).get_data();
+ ASSERT_EQ(values.size(), 2);
+ EXPECT_DOUBLE_EQ(values[0], 2);
+ EXPECT_DOUBLE_EQ(values[1], 10);
+ // Repeated execution must not consume or mutate the serialized state.
+ EXPECT_EQ(result->compare_at(0, 0, *finalize(type, serialized), 1), 0);
+}
+
+TEST_F(FunctionAggStateFinalizeTest, EmptyNonNullableAvgMatchesMerge) {
+ auto type = state_type("avg", std::make_shared<DataTypeInt64>());
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({}),
+ *states);
+ ColumnPtr serialized = std::move(states);
+ auto result = finalize(type, serialized);
+ auto nested = type->get_nested_function();
+ auto merge = AggregateStateMerge::create(nested, {type},
nested->get_return_type());
+ Arena arena;
+ auto* place = arena.aligned_alloc(merge->size_of_data(),
merge->align_of_data());
+ merge->create(place);
+ DEFER(merge->destroy(place));
+ const IColumn* columns[] = {serialized.get()};
+ merge->add(place, columns, 0, arena);
+ auto expected = nested->get_return_type()->create_column();
+ merge->check_result_column_type(*expected);
+ merge->insert_result_into(place, *expected);
+ EXPECT_EQ(result->compare_at(0, 0, *expected, 1), 0);
+}
+
+TEST_F(FunctionAggStateFinalizeTest, TrivialStatesResetAcrossOuterNulls) {
+ auto type = state_type("sum", std::make_shared<DataTypeInt64>());
+ auto function = type->get_nested_function();
+ ASSERT_TRUE(function->is_trivial());
+ auto states = type->create_column();
+ append_state(function, ColumnHelper::create_column<DataTypeInt64>({2, 4}),
*states);
+ states->insert_default();
+ append_state(function, ColumnHelper::create_column<DataTypeInt64>({}),
*states);
+ append_state(function, ColumnHelper::create_column<DataTypeInt64>({-7}),
*states);
+ append_state(function, ColumnHelper::create_column<DataTypeInt64>({10}),
*states);
+ ColumnPtr nullable_states = ColumnNullable::create(
+ std::move(states), ColumnHelper::create_column<DataTypeUInt8>({0,
1, 0, 0, 0}));
+
+ auto result = finalize(make_nullable(type), nullable_states);
+ const auto& values = assert_cast<const
ColumnNullable&>(*result).get_nested_column();
+ ASSERT_EQ(result->size(), 5);
+ EXPECT_EQ(values.get_int(0), 6);
+ EXPECT_TRUE(result->is_null_at(1));
+ EXPECT_FALSE(result->is_null_at(2));
+ EXPECT_EQ(values.get_int(2), 0);
+ EXPECT_EQ(values.get_int(3), -7);
+ EXPECT_EQ(values.get_int(4), 10);
+
+ auto repeated = finalize(make_nullable(type), nullable_states);
+ for (size_t row = 0; row < result->size(); ++row) {
+ EXPECT_EQ(result->compare_at(row, row, *repeated, 1), 0);
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest, NativeSerializedColumnsAndEmptyCount) {
+ for (const auto& name : {"count", "sum", "min", "max"}) {
+ SCOPED_TRACE(name);
+ auto type =
+ state_type(name, std::make_shared<DataTypeInt64>(), name !=
std::string("count"));
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_column<DataTypeInt64>({2, 4}),
*states);
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({}),
+ *states);
+ auto result = finalize(type, std::move(states));
+ EXPECT_EQ(result->get_int(0), name == std::string("count") ? 2
+ : name == std::string("sum") ? 6
+ : name == std::string("min") ? 2
+ : 4);
+ // Empty non-nullable states have the existing aggregate's identity
value.
+ auto expected =
type->get_nested_function()->get_return_type()->create_column();
+ Arena arena;
+ auto function = type->get_nested_function();
+ auto* place = arena.aligned_alloc(function->size_of_data(),
function->align_of_data());
+ function->create(place);
+ DEFER(function->destroy(place));
+ function->check_result_column_type(*expected);
+ function->insert_result_into(place, *expected);
+ EXPECT_EQ(result->compare_at(1, 0, *expected, 1), 0);
+ }
+}
+
+// GTest assertion macros inflate complexity in this table-driven check.
+// NOLINTNEXTLINE(readability-function-cognitive-complexity)
+TEST_F(FunctionAggStateFinalizeTest, NullableInputsAndEmptyStates) {
+ for (const auto& name : {"avg", "sum", "min", "max", "count"}) {
+ SCOPED_TRACE(name);
+ auto type = state_type(name,
make_nullable(std::make_shared<DataTypeInt64>()),
+ name != std::string("count"));
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({2,
0, 4}, {0, 1, 0}),
+ *states);
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({0},
{1}), *states);
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({},
{}), *states);
+ auto result = finalize(type, std::move(states));
+ ASSERT_EQ(result->size(), 3);
+ EXPECT_FALSE(result->is_null_at(0));
+ if (name == std::string("count")) {
+ EXPECT_EQ(result->get_int(0), 2);
+ EXPECT_EQ(result->get_int(1), 0);
+ EXPECT_EQ(result->get_int(2), 0);
+ } else {
+ EXPECT_TRUE(result->is_null_at(1));
+ EXPECT_TRUE(result->is_null_at(2));
+ }
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest, OuterNullAndConstantStates) {
+ auto type = state_type("avg",
make_nullable(std::make_shared<DataTypeInt64>()));
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_nullable_column<DataTypeInt64>({2, 4},
{0, 0}), *states);
+ ColumnPtr constant = ColumnConst::create(states->clone_resized(1), 5);
+ auto result = finalize(type, constant);
+ EXPECT_TRUE(is_column_const(*result));
+ EXPECT_EQ(result->size(), 5);
+ // NULL string payload is deliberately empty, and must not be deserialized.
+ states->insert_default();
+ ColumnPtr nullable_states = ColumnNullable::create(
+ std::move(states), ColumnHelper::create_column<DataTypeUInt8>({0,
1}));
+ result = finalize(make_nullable(type), nullable_states);
+ ASSERT_EQ(result->size(), 2);
+ EXPECT_FALSE(result->is_null_at(0));
+ EXPECT_TRUE(result->is_null_at(1));
+ auto null_state = nullable_states->clone_empty();
+ null_state->insert_from(*nullable_states, 1);
+ ColumnPtr null_constant = ColumnConst::create(std::move(null_state), 7);
+ result = finalize(make_nullable(type), null_constant);
+ EXPECT_EQ(result->size(), 7);
+ EXPECT_TRUE(result->is_null_at(0));
+}
+
+TEST_F(FunctionAggStateFinalizeTest, VariableLengthArrayResultsOwnTheirData) {
+ auto type = state_type("array_agg", std::make_shared<DataTypeString>(),
false);
+ auto states = type->create_column();
+ const std::string large(8192, 'x');
+ for (size_t group = 0; group < 16; ++group) {
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_column<DataTypeString>({large,
std::to_string(group)}),
+ *states);
+ }
+ auto result = finalize(type, std::move(states));
+ const auto& arrays = assert_cast<const ColumnArray&>(*result);
+ const auto& strings = assert_cast<const ColumnString&>(
+ assert_cast<const
ColumnNullable&>(arrays.get_data()).get_nested_column());
+ ASSERT_EQ(arrays.size(), 16);
+ for (size_t group = 0; group < 16; ++group) {
+ EXPECT_EQ(arrays.get_offsets()[group], (group + 1) * 2);
+ EXPECT_EQ(strings.get_data_at(group * 2).to_string(), large);
+ EXPECT_EQ(strings.get_data_at(group * 2 + 1).to_string(),
std::to_string(group));
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest,
CollectSetReadsMultipleValuesAndEmptyStates) {
+ auto type = state_type("collect_set", std::make_shared<DataTypeInt64>(),
false);
+ auto states = type->create_column();
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({1, 2, 2}),
+ *states);
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({}),
+ *states);
+ append_state(type->get_nested_function(),
ColumnHelper::create_column<DataTypeInt64>({9, 7, 9}),
+ *states);
+ auto result = finalize(type, std::move(states));
+ const auto& arrays = assert_cast<const ColumnArray&>(*result);
+ const auto& values = assert_cast<const
ColumnNullable&>(arrays.get_data()).get_nested_column();
+ const std::vector<std::vector<Int64>> expected {{1, 2}, {}, {7, 9}};
+ ASSERT_EQ(arrays.size(), expected.size());
+ for (size_t row = 0; row < expected.size(); ++row) {
+ std::vector<Int64> actual;
+ const auto& offsets = arrays.get_offsets();
+ for (size_t i = offsets[static_cast<ssize_t>(row) - 1]; i <
offsets[row]; ++i) {
+ actual.push_back(values.get_int(i));
+ }
+ std::ranges::sort(actual);
+ EXPECT_EQ(actual, expected[row]);
+ }
+}
+
+TEST_F(FunctionAggStateFinalizeTest, EmptyBlock) {
+ auto type = state_type("avg", std::make_shared<DataTypeInt64>());
+ auto result = finalize(type, type->create_column());
+ EXPECT_EQ(result->size(), 0);
+}
+
+TEST_F(FunctionAggStateFinalizeTest, VariableLengthStatesWithOuterNulls) {
+ auto type = state_type("collect_set", std::make_shared<DataTypeString>(),
false);
+ auto states = type->create_column();
+ auto null_map = ColumnUInt8::create();
+ const std::string large(16384, 'x');
+ for (size_t group = 0; group < 16; ++group) {
+ // Valid rows exceed the initial arena chunk; outer NULL payloads must
be skipped.
+ states->insert_default();
+ null_map->insert_value(1);
+ append_state(type->get_nested_function(),
+ ColumnHelper::create_column<DataTypeString>({large +
std::to_string(group)}),
+ *states);
+ null_map->insert_value(0);
+ }
+ auto result = finalize(make_nullable(type),
+ ColumnNullable::create(std::move(states),
std::move(null_map)));
+ const auto& arrays = assert_cast<const ColumnArray&>(
+ assert_cast<const ColumnNullable&>(*result).get_nested_column());
+ const auto& strings = assert_cast<const ColumnString&>(
+ assert_cast<const
ColumnNullable&>(arrays.get_data()).get_nested_column());
+ ASSERT_EQ(result->size(), 32);
+ for (size_t group = 0; group < 16; ++group) {
+ EXPECT_TRUE(result->is_null_at(group * 2));
+ EXPECT_FALSE(result->is_null_at(group * 2 + 1));
+ EXPECT_EQ(arrays.get_offsets()[group * 2], group);
+ EXPECT_EQ(arrays.get_offsets()[group * 2 + 1], group + 1);
+ EXPECT_EQ(strings.get_data_at(group).to_string(), large +
std::to_string(group));
+ }
+}
+
+} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionRegistry.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionRegistry.java
index 0a82f786c22..fbc3e9d84af 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionRegistry.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionRegistry.java
@@ -115,7 +115,8 @@ public class FunctionRegistry {
return containsAggregateFunction(functionBuilders);
}
if (isBuiltinAggStateCombinator(name)) {
- return !name.endsWith(AggCombinerFunctionBuilder.STATE_SUFFIX);
+ return !name.endsWith(AggCombinerFunctionBuilder.STATE_SUFFIX)
+ &&
!name.endsWith(AggCombinerFunctionBuilder.FINALIZE_SUFFIX);
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
index a77299b8c4c..948238357a4 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/ExpressionTranslator.java
@@ -92,6 +92,7 @@ import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
import
org.apache.doris.nereids.trees.expressions.functions.agg.NotNullableAggregateFunction;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.CombineCombinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.FinalizeCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.ForEachCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.MergeCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.StateCombinator;
@@ -831,6 +832,13 @@ public class ExpressionTranslator extends
DefaultExpressionVisitor<Expr, PlanTra
return isNullPredicate;
}
+ @Override
+ public Expr visitFinalizeCombinator(FinalizeCombinator combinator,
PlanTranslatorContext context) {
+ FunctionCallExpr functionCallExpr = (FunctionCallExpr)
visitScalarFunction(combinator, context);
+ functionCallExpr.getFn().setBinaryType(Function.BinaryType.AGG_STATE);
+ return functionCallExpr;
+ }
+
@Override
public Expr visitStateCombinator(StateCombinator combinator,
PlanTranslatorContext context) {
List<Expr> arguments = combinator.getArguments().stream()
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/AggFunctionRollUpHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/AggFunctionRollUpHandler.java
index 19008042451..6d02b4379a2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/AggFunctionRollUpHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/AggFunctionRollUpHandler.java
@@ -23,6 +23,11 @@ import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.functions.Function;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.Combinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.CombineCombinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.MergeCombinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.StateCombinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.UnionCombinator;
import com.google.common.collect.ImmutableList;
@@ -73,12 +78,23 @@ public abstract class AggFunctionRollUpHandler {
}
/**
- * Extract the target expression in actualFunction by targetClazz
- * Such as actualFunction def is avg_merge(avg_union(c1)), target Clazz is
Combinator
- * after extracting, the return argument is avg_union(c1)
+ * Unwrap direct MERGE/UNION state chains for the same aggregate function.
+ * STATE and COMBINE consume values, so keep their complete argument
expressions for comparison.
+ * For example, sum_combine(avg_finalize(avg_state(v))) must retain SUM
and its finalized value.
*/
- protected static <T> T extractLastExpression(Expression actualFunction,
Class<T> targetClazz) {
- List<Expression> expressions =
actualFunction.collectToList(targetClazz::isInstance);
- return targetClazz.cast(expressions.get(expressions.size() - 1));
+ protected static Combinator extractRollupCombinator(Combinator current) {
+ while (current instanceof MergeCombinator || current instanceof
UnionCombinator) {
+ Expression argument = current.getArguments().get(0);
+ if (!(argument instanceof StateCombinator || argument instanceof
CombineCombinator
+ || argument instanceof UnionCombinator)) {
+ break;
+ }
+ Combinator next = (Combinator) argument;
+ if
(!current.getNestedFunction().getName().equals(next.getNestedFunction().getName()))
{
+ break;
+ }
+ current = next;
+ }
+ return current;
}
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/BothCombinatorRollupHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/BothCombinatorRollupHandler.java
index 38c1dedcefe..c4c047ce0a7 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/BothCombinatorRollupHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/BothCombinatorRollupHandler.java
@@ -47,8 +47,8 @@ public class BothCombinatorRollupHandler extends
AggFunctionRollUpHandler {
return false;
}
if (queryAggregateFunction instanceof Combinator && viewFunction
instanceof Combinator) {
- Combinator queryCombinator =
extractLastExpression(queryAggregateFunction, Combinator.class);
- Combinator viewCombinator = extractLastExpression(viewFunction,
Combinator.class);
+ Combinator queryCombinator = extractRollupCombinator((Combinator)
queryAggregateFunction);
+ Combinator viewCombinator = extractRollupCombinator((Combinator)
viewFunction);
// construct actual aggregate function in combinator and compare
return
Objects.equals(queryCombinator.getNestedFunction().withChildren(queryCombinator.getArguments()),
viewCombinator.getNestedFunction().withChildren(viewCombinator.getArguments()));
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/SingleCombinatorRollupHandler.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/SingleCombinatorRollupHandler.java
index 9a865d7a08c..6d012a2f0c9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/SingleCombinatorRollupHandler.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/SingleCombinatorRollupHandler.java
@@ -56,7 +56,7 @@ public class SingleCombinatorRollupHandler extends
AggFunctionRollUpHandler {
if (!(queryAggregateFunction instanceof Combinator)
&& (viewFunction instanceof UnionCombinator || viewFunction
instanceof StateCombinator
|| viewFunction instanceof CombineCombinator)) {
- Combinator viewCombinator = extractLastExpression(viewFunction,
Combinator.class);
+ Combinator viewCombinator = extractRollupCombinator((Combinator)
viewFunction);
return Objects.equals(queryAggregateFunction,
viewCombinator.getNestedFunction().withChildren(viewCombinator.getArguments()));
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java
index c32162ca095..90ba14b1a2f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java
@@ -23,6 +23,7 @@ import
org.apache.doris.nereids.trees.expressions.SlotReference;
import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import
org.apache.doris.nereids.trees.expressions.functions.agg.NotSupportAggState;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.CombineCombinator;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.FinalizeCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.ForEachCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.MergeCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.StateCombinator;
@@ -43,12 +44,14 @@ public class AggCombinerFunctionBuilder extends
FunctionBuilder {
public static final String COMBINATOR_LINKER = "_";
public static final String STATE = "state";
public static final String MERGE = "merge";
+ public static final String FINALIZE = "finalize";
public static final String UNION = "union";
public static final String COMBINE = "combine";
public static final String FOREACH = "foreach";
public static final String STATE_SUFFIX = COMBINATOR_LINKER + STATE;
public static final String MERGE_SUFFIX = COMBINATOR_LINKER + MERGE;
+ public static final String FINALIZE_SUFFIX = COMBINATOR_LINKER + FINALIZE;
public static final String UNION_SUFFIX = COMBINATOR_LINKER + UNION;
public static final String COMBINE_SUFFIX = COMBINATOR_LINKER + COMBINE;
public static final String FOREACH_SUFFIX = COMBINATOR_LINKER + FOREACH;
@@ -121,7 +124,7 @@ public class AggCombinerFunctionBuilder extends
FunctionBuilder {
return (AggregateFunction) nestedBuilder.build(nestedName,
forEachargs).first;
}
- private AggregateFunction buildMergeOrUnion(String nestedName, List<?
extends Object> arguments) {
+ private AggregateFunction buildFromState(String nestedName, List<? extends
Object> arguments) {
if (arguments.size() != 1 || !(arguments.get(0) instanceof Expression)
|| !((Expression)
arguments.get(0)).getDataType().isAggStateType()) {
String argString = arguments.stream().map(arg -> {
@@ -162,10 +165,13 @@ public class AggCombinerFunctionBuilder extends
FunctionBuilder {
AggregateFunction nestedFunction = buildState(nestedName,
arguments);
return Pair.of(new CombineCombinator((List<Expression>) arguments,
nestedFunction), nestedFunction);
} else if (combinatorSuffix.equalsIgnoreCase(MERGE)) {
- AggregateFunction nestedFunction = buildMergeOrUnion(nestedName,
arguments);
+ AggregateFunction nestedFunction = buildFromState(nestedName,
arguments);
return Pair.of(new MergeCombinator((List<Expression>) arguments,
nestedFunction), nestedFunction);
+ } else if (combinatorSuffix.equalsIgnoreCase(FINALIZE)) {
+ AggregateFunction nestedFunction = buildFromState(nestedName,
arguments);
+ return Pair.of(new FinalizeCombinator((List<Expression>)
arguments, nestedFunction), nestedFunction);
} else if (combinatorSuffix.equalsIgnoreCase(UNION)) {
- AggregateFunction nestedFunction = buildMergeOrUnion(nestedName,
arguments);
+ AggregateFunction nestedFunction = buildFromState(nestedName,
arguments);
return Pair.of(new UnionCombinator((List<Expression>) arguments,
nestedFunction), nestedFunction);
} else if (combinatorSuffix.equalsIgnoreCase(FOREACH)) {
AggregateFunction nestedFunction = buildForEach(nestedName,
arguments);
@@ -182,7 +188,7 @@ public class AggCombinerFunctionBuilder extends
FunctionBuilder {
public static boolean isAggStateCombinator(String name) {
return name.toLowerCase().endsWith(STATE_SUFFIX) ||
name.toLowerCase().endsWith(MERGE_SUFFIX)
|| name.toLowerCase().endsWith(UNION_SUFFIX) ||
name.toLowerCase().endsWith(COMBINE_SUFFIX)
- || name.toLowerCase().endsWith(FOREACH_SUFFIX);
+ || name.toLowerCase().endsWith(FOREACH_SUFFIX) ||
name.toLowerCase().endsWith(FINALIZE_SUFFIX);
}
public static String getNestedName(String name) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinator.java
new file mode 100644
index 00000000000..10321d707d9
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinator.java
@@ -0,0 +1,100 @@
+// 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.trees.expressions.functions.combinator;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import
org.apache.doris.nereids.trees.expressions.functions.AggCombinerFunctionBuilder;
+import org.apache.doris.nereids.trees.expressions.functions.ComputeNullable;
+import
org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ScalarFunction;
+import
org.apache.doris.nereids.trees.expressions.functions.scalar.ScalarFunctionParams;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.AggStateType;
+
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/** Finalize each serialized aggregate state without aggregating rows. */
+public class FinalizeCombinator extends ScalarFunction
+ implements UnaryExpression, ExplicitlyCastableSignature,
ComputeNullable, Combinator {
+
+ private final AggregateFunction nested;
+
+ public FinalizeCombinator(List<Expression> arguments, AggregateFunction
nested) {
+ super(nested.getName() + AggCombinerFunctionBuilder.FINALIZE_SUFFIX,
arguments);
+ this.nested = nested;
+ checkStateFunction();
+ }
+
+ private FinalizeCombinator(ScalarFunctionParams functionParams,
AggregateFunction nested) {
+ super(functionParams);
+ this.nested = nested;
+ checkStateFunction();
+ }
+
+ private void checkStateFunction() {
+ AggStateType inputType = (AggStateType) getArgument(0).getDataType();
+ // Keep the name check because acceptsType alone accepts unrelated
aggregate states.
+ // TODO: Unify aggregate alias canonicalization across FE and BE.
Stored aliases absent
+ // from AggStateType's mapping (for example, std) remain unsupported
by finalize for now.
+ AggStateType expected = new AggStateType(nested.getName(),
inputType.getSubTypes(),
+ inputType.getSubTypeNullables(), nested.nullable());
+ if (!inputType.getFunctionName().equals(expected.getFunctionName())) {
+ throw new AnalysisException(getName() + " requires a state of " +
expected.getFunctionName()
+ + ", but got " + inputType.toSql());
+ }
+ }
+
+ @Override
+ public FinalizeCombinator withChildren(List<Expression> children) {
+ return new FinalizeCombinator(getFunctionParams(children), nested);
+ }
+
+ @Override
+ public List<FunctionSignature> getSignatures() {
+ return
ImmutableList.of(FunctionSignature.ret(nested.getDataType()).args(getArgument(0).getDataType()));
+ }
+
+ @Override
+ public boolean nullable() {
+ return getArgument(0).nullable() || nested.nullable();
+ }
+
+ @Override
+ public AggregateFunction getNestedFunction() {
+ return nested;
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitFinalizeCombinator(this, context);
+ }
+
+ @Override
+ public void checkLegalityBeforeTypeCoercion() {
+ // Stored states retain serialized parameters, not the original
constant expressions.
+ if (getArgument(0) instanceof StateCombinator || getArgument(0)
instanceof CombineCombinator) {
+ nested.checkLegalityBeforeTypeCoercion();
+ }
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/readme.md
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/readme.md
index e861156605a..28d06aad7cf 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/readme.md
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/readme.md
@@ -5,3 +5,25 @@ The function combiner is used to automatically create some
functions based on th
Currently we can generate the combinator function we want by implementing a
new `FunctionBuilder`. For example `AggStateFunctionBuilder` can generate
combiner functions created by the `_state`/`_merge` combiner, so that we can
use functions such as `sum_state` and `sum_merge` in SQL statements.
When we input `sum_state`, if `BuiltinFunctionBuilder` does not find a
function named `sum_state`, then it will try to use `AggStateFunctionBuilder`
to construct `StateCombinator`, so we get a `StateCombinator` object with
nested function sum, and use it to generate the `sum_state` function.
+
+## Finalize one aggregate state
+
+`<aggregate>_finalize(state)` is a scalar combinator that returns the result
of each
+serialized aggregate state independently. It accepts exactly one `AGG_STATE` of
+the matching aggregate (including its aliases), and uses the same result type
and
+state semantics as `<aggregate>_merge`. An outer SQL NULL state returns NULL;
+an empty serialized state follows the nested aggregate's existing behavior.
+
+For example, after pre-aggregating each key, the finest grouping can read its
AVG
+without another aggregate operator:
+
+```sql
+SET enable_agg_state = true;
+SELECT k, avg_finalize(s)
+FROM (SELECT k, avg_combine(v) AS s FROM t GROUP BY k) partial;
+```
+
+`avg_finalize` returns one result per input row. In contrast, `avg_merge`
merges
+all input states in each group before computing one result. The same scalar
+combinator applies to other aggregates supporting `AGG_STATE`, such as `sum`,
+`count`, `min`, `max`, and `array_agg`; it does not change their serialized
formats.
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
index a888b2af817..9d5e1b7485e 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java
@@ -31,6 +31,7 @@ import
org.apache.doris.nereids.trees.expressions.functions.ai.AISimilarity;
import org.apache.doris.nereids.trees.expressions.functions.ai.AISummarize;
import org.apache.doris.nereids.trees.expressions.functions.ai.AITranslate;
import org.apache.doris.nereids.trees.expressions.functions.ai.Embed;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.FinalizeCombinator;
import
org.apache.doris.nereids.trees.expressions.functions.combinator.StateCombinator;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Abs;
import org.apache.doris.nereids.trees.expressions.functions.scalar.Acos;
@@ -2882,6 +2883,10 @@ public interface ScalarFunctionVisitor<R, C> {
return visitScalarFunction(yearsDiff, context);
}
+ default R visitFinalizeCombinator(FinalizeCombinator combinator, C
context) {
+ return visitScalarFunction(combinator, context);
+ }
+
default R visitStateCombinator(StateCombinator combinator, C context) {
return visitScalarFunction(combinator, context);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/rollup/CombinatorRollupHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/rollup/CombinatorRollupHandlerTest.java
new file mode 100644
index 00000000000..4821dbef6e8
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/mv/rollup/CombinatorRollupHandlerTest.java
@@ -0,0 +1,156 @@
+// 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.exploration.mv.rollup;
+
+import org.apache.doris.catalog.FunctionRegistry;
+import org.apache.doris.common.Pair;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import
org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import
org.apache.doris.nereids.trees.expressions.functions.combinator.StateCombinator;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Abs;
+import org.apache.doris.nereids.types.DoubleType;
+import org.apache.doris.nereids.util.MemoTestUtils;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class CombinatorRollupHandlerTest {
+ private final FunctionRegistry registry = new FunctionRegistry();
+ private final SlotReference value = new SlotReference("v",
DoubleType.INSTANCE, true);
+
+ @BeforeAll
+ static void setUp() {
+ MemoTestUtils.createConnectContext();
+ }
+
+ private Expression build(String name, Expression argument) {
+ return registry.findFunctionBuilder(name, argument).build(name,
argument).first;
+ }
+
+ private Expression finalizeValue(boolean stored) {
+ Expression state = build("avg_state", value);
+ return build("avg_finalize", stored ? new SlotReference("s",
state.getDataType(), true) : state);
+ }
+
+ private boolean canRollup(AggFunctionRollUpHandler handler, Expression
query, Expression view) {
+ SlotReference mvSlot = SlotReference.of("mv_state",
view.getDataType());
+ return handler.canRollup((AggregateFunction) query, query,
Pair.of(view, mvSlot),
+ ImmutableMap.of(view, mvSlot));
+ }
+
+ private void assertRollup(AggFunctionRollUpHandler handler, Expression
query, Expression view, String name) {
+ Assertions.assertTrue(canRollup(handler, query, view));
+ SlotReference mvSlot = SlotReference.of("mv_state",
view.getDataType());
+ Function result = handler.doRollup((AggregateFunction) query, query,
Pair.of(view, mvSlot),
+ ImmutableMap.of(view, mvSlot));
+ Assertions.assertEquals(name, result.getName());
+ Assertions.assertEquals(ImmutableList.of(mvSlot),
result.getArguments());
+ }
+
+ @Test
+ void testDifferentOuterAggregatesOverFinalize() {
+ for (boolean stored : ImmutableList.of(false, true)) {
+ Expression finalized = finalizeValue(stored);
+
Assertions.assertFalse(canRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_combine", finalized), build("max_combine",
finalized)));
+ }
+ }
+
+ @Test
+ void testScalarArgumentMustMatch() {
+ Expression finalized = finalizeValue(false);
+ Assertions.assertFalse(canRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_combine", finalized), build("sum_combine", new
Abs(finalized))));
+ }
+
+ @Test
+ void testSameOuterAggregateOverFinalize() {
+ Expression finalized = finalizeValue(false);
+ assertRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_combine", finalized), build("sum_combine",
finalized), "sum_union");
+ }
+
+ @Test
+ void testMergeAndUnionStateChain() {
+ Expression state = build("sum_state", value);
+ assertRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", state), build("sum_union", state),
"sum_merge");
+ assertRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", build("sum_union", state)),
build("sum_union", state), "sum_merge");
+ }
+
+ @Test
+ void testUnionStopsAtCombineValueArguments() {
+ Expression finalized = finalizeValue(false);
+ assertRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", build("sum_combine", finalized)),
+ build("sum_union", build("sum_combine", finalized)),
"sum_merge");
+ Assertions.assertFalse(canRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", build("sum_combine", finalized)),
+ build("max_union", build("max_combine", finalized))));
+ }
+
+ @Test
+ void testStateRetainsFinalizeArgument() {
+ Expression finalized = finalizeValue(false);
+ Assertions.assertFalse(canRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", build("sum_state", finalized)),
+ build("max_union", build("max_state", finalized))));
+ }
+
+ @Test
+ void testStoredStateMustMatch() {
+ Expression state = build("sum_state", value);
+ SlotReference first = SlotReference.of("first_state",
state.getDataType());
+ SlotReference second = SlotReference.of("second_state",
state.getDataType());
+ assertRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", first), build("sum_union", first),
"sum_merge");
+ Assertions.assertFalse(canRollup(BothCombinatorRollupHandler.INSTANCE,
+ build("sum_merge", first), build("sum_union", second)));
+ }
+
+ @Test
+ void testSingleCombinatorOverFinalize() {
+ for (boolean stored : ImmutableList.of(false, true)) {
+ Expression finalized = finalizeValue(stored);
+ Sum query = new Sum(finalized);
+ assertRollup(SingleCombinatorRollupHandler.INSTANCE,
+ query, build("sum_combine", finalized), "sum_merge");
+
Assertions.assertFalse(canRollup(SingleCombinatorRollupHandler.INSTANCE,
+ query, build("max_combine", finalized)));
+
Assertions.assertFalse(canRollup(SingleCombinatorRollupHandler.INSTANCE,
+ query, build("sum_combine", new Abs(finalized))));
+ }
+ }
+
+ @Test
+ void testSingleCombinatorStateChain() {
+ Expression finalized = finalizeValue(false);
+ Sum query = new Sum(finalized);
+ assertRollup(SingleCombinatorRollupHandler.INSTANCE, query,
+ build("sum_union", StateCombinator.create(query)),
"sum_merge");
+ assertRollup(SingleCombinatorRollupHandler.INSTANCE, query,
+ build("sum_union", build("sum_combine", finalized)),
"sum_merge");
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinatorTest.java
new file mode 100644
index 00000000000..080ba6347cf
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/combinator/FinalizeCombinatorTest.java
@@ -0,0 +1,127 @@
+// 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.trees.expressions.functions.combinator;
+
+import org.apache.doris.analysis.FunctionCallExpr;
+import org.apache.doris.catalog.Function;
+import org.apache.doris.catalog.FunctionRegistry;
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.glue.translator.ExpressionTranslator;
+import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.FunctionBuilder;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Avg;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.types.AggStateType;
+import org.apache.doris.nereids.types.DecimalV3Type;
+import org.apache.doris.nereids.types.DoubleType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.util.MemoTestUtils;
+import org.apache.doris.nereids.util.PlanChecker;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+class FinalizeCombinatorTest {
+ private FinalizeCombinator build(String name, Expression argument) {
+ FunctionBuilder builder = new
FunctionRegistry().findFunctionBuilder(name, argument);
+ return (FinalizeCombinator) builder.build(name, argument).first;
+ }
+
+ @Test
+ void testScalarClassificationAndTranslation() {
+ FunctionRegistry registry = new FunctionRegistry();
+
Assertions.assertTrue(registry.isBuiltinAggStateCombinator("avg_finalize"));
+ Assertions.assertFalse(registry.isAggregateFunction(null,
"avg_finalize"));
+
Assertions.assertFalse(registry.isBuiltinAggStateCombinator("abs_finalize"));
+ SlotReference value = new SlotReference("v", IntegerType.INSTANCE,
true);
+ StateCombinator state = StateCombinator.create(new Avg(value));
+ FinalizeCombinator finalize = build("avg_finalize", state);
+ Assertions.assertEquals(DoubleType.INSTANCE, finalize.getDataType());
+ Assertions.assertTrue(finalize.nullable());
+ FunctionCallExpr translated = (FunctionCallExpr)
ExpressionTranslator.translate(
+ finalize, new PlanTranslatorContext());
+ Assertions.assertEquals("avg_finalize",
translated.getFn().getFunctionName().getFunction());
+ Assertions.assertEquals(Function.BinaryType.AGG_STATE,
translated.getFn().getBinaryType());
+ Assertions.assertEquals(Function.NullableMode.ALWAYS_NULLABLE,
translated.getFn().getNullableMode());
+ Assertions.assertEquals(state.getDataType().toCatalogDataType(),
translated.getFn().getArgs()[0]);
+ }
+
+ @Test
+ void testCountAndOuterNullability() {
+ SlotReference value = new SlotReference("v", IntegerType.INSTANCE,
true);
+ StateCombinator state = StateCombinator.create(new Count(value));
+ Assertions.assertFalse(build("count_finalize", state).nullable());
+ SlotReference nullableState = new SlotReference("s",
state.getDataType(), true);
+ Assertions.assertTrue(build("count_finalize",
nullableState).nullable());
+ }
+
+ @Test
+ void testNonNullableStateUsesNestedResultType() {
+ SlotReference value = new SlotReference("v", IntegerType.INSTANCE,
false);
+ StateCombinator state = StateCombinator.create(new Avg(value));
+ Assertions.assertFalse(build("avg_finalize", state).nullable());
+ }
+
+ @Test
+ void testDecimalResultPrecision() {
+ MemoTestUtils.createConnectContext();
+ SlotReference value = new SlotReference("v",
DecimalV3Type.createDecimalV3Type(12, 2), true);
+ FinalizeCombinator finalize = build("avg_finalize",
StateCombinator.create(new Avg(value)));
+ Assertions.assertEquals(DecimalV3Type.createDecimalV3Type(38, 4),
finalize.getDataType());
+ }
+
+ @Test
+ void testStateFunctionMustMatch() {
+ for (String sql : ImmutableList.of("select sum_finalize(avg_state(1))",
+ "select avg_finalize(sum_state(1))")) {
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class,
+ () ->
PlanChecker.from(MemoTestUtils.createConnectContext()).analyze(sql));
+ Assertions.assertTrue(exception.getMessage().contains("requires a
state of"));
+ }
+ }
+
+ @Test
+ void testAliasCanonicalization() {
+ AggStateType state = new AggStateType("var_pop",
ImmutableList.of(DoubleType.INSTANCE),
+ ImmutableList.of(true), true);
+ FinalizeCombinator finalize = build("variance_finalize", new
SlotReference("s", state, false));
+ Assertions.assertEquals("variance_finalize", finalize.getName());
+ Assertions.assertEquals(DoubleType.INSTANCE, finalize.getDataType());
+ }
+
+ @Test
+ void testInvalidArguments() {
+ for (String sql : ImmutableList.of("select avg_finalize(1)", "select
avg_finalize()",
+ "select avg_finalize(avg_state(1), avg_state(2))",
+ "select avg_finalize(distinct avg_state(1))")) {
+ Assertions.assertThrows(AnalysisException.class,
+ () ->
PlanChecker.from(MemoTestUtils.createConnectContext()).analyze(sql));
+ }
+ }
+
+ @Test
+ void testStoredParameterizedStateAndCombine() {
+ PlanChecker.from(MemoTestUtils.createConnectContext()).analyze(
+ "select topn_finalize(s) from (select topn_combine('x', 2) s)
t");
+ PlanChecker.from(MemoTestUtils.createConnectContext()).analyze(
+ "select avg_finalize(avg_combine(cast(1 as int)))");
+ }
+}
diff --git
a/regression-test/data/datatype_p0/agg_state/test_agg_state_finalize.out
b/regression-test/data/datatype_p0/agg_state/test_agg_state_finalize.out
new file mode 100644
index 00000000000..763c1420073
--- /dev/null
+++ b/regression-test/data/datatype_p0/agg_state/test_agg_state_finalize.out
@@ -0,0 +1,60 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !rows --
+1 1 1 1 1 1
+1 3 1 3 3 3
+2 10 1 10 10 10
+3 \N 0 \N \N \N
+
+-- !grouped --
+1 2 2 4 1 3
+2 10 1 10 10 10
+3 \N 0 \N \N \N
+
+-- !empty --
+\N 0 \N
+
+-- !decimal --
+1 2.5000
+2 10.0000
+3 \N
+
+-- !array --
+1 ["a", "b"]
+2 ["c"]
+3 [null]
+
+-- !parameters --
+{"a":1}
+
+-- !const --
+0 7 0
+1 7 0
+2 7 0
+
+-- !rollup --
+\N 4.666666666666667
+1 2
+2 10
+3 \N
+
+-- !stored --
+1 2 2 4
+2 10 1 10
+3 \N 0 \N
+
+-- !outer_null --
+0 \N \N
+1 2 2
+2 10 1
+3 \N 0
+4 \N \N
+
+-- !union --
+4.666666666666667
+
+-- !alias --
+\N \N
+0 0
+0 0
+0 0
+
diff --git
a/regression-test/data/datatype_p0/agg_state/test_agg_state_union_batch.out
b/regression-test/data/datatype_p0/agg_state/test_agg_state_union_batch.out
new file mode 100644
index 00000000000..a09c50a2b2e
--- /dev/null
+++ b/regression-test/data/datatype_p0/agg_state/test_agg_state_union_batch.out
@@ -0,0 +1,37 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !merge --
+0 7142142 5715717 71421.420000000 value-0
+1 7143571 5712856 71435.710000000 value-1
+2 7145000 5720000 71450.000000000 value-100
+3 7146429 5717144 71464.290000000 value-10
+4 7137858 5714283 71378.580000000 value-1005
+5 7139286 5711431 71392.860000000 value-1006
+6 7140714 \N \N \N
+
+-- !direct --
+0 7142142 5715717 71421.420000000 value-0
+1 7143571 5712856 71435.710000000 value-1
+2 7145000 5720000 71450.000000000 value-100
+3 7146429 5717144 71464.290000000 value-10
+4 7137858 5714283 71378.580000000 value-1005
+5 7139286 5711431 71392.860000000 value-1006
+6 7140714 \N \N \N
+
+-- !union --
+0 7142142 5715717 71421.420000000 value-0
+1 7143571 5712856 71435.710000000 value-1
+2 7145000 5720000 71450.000000000 value-100
+3 7146429 5717144 71464.290000000 value-10
+4 7137858 5714283 71378.580000000 value-1005
+5 7139286 5711431 71392.860000000 value-1006
+6 7140714 \N \N \N
+
+-- !array --
+0 true true
+1 true true
+2 true true
+3 true true
+4 true true
+5 true true
+6 true true
+
diff --git
a/regression-test/data/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.out
b/regression-test/data/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.out
new file mode 100644
index 00000000000..cd51bb4190f
--- /dev/null
+++
b/regression-test/data/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.out
@@ -0,0 +1,36 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !combined_before --
+1 400
+2 500
+3 \N
+
+-- !plain_before --
+1 400
+2 500
+3 \N
+
+-- !different_aggregate --
+1 400
+2 500
+3 \N
+
+-- !different_argument --
+1 400
+2 500
+3 \N
+
+-- !combined_after --
+1 400
+2 500
+3 \N
+
+-- !plain_after --
+1 400
+2 500
+3 \N
+
+-- !merge_after --
+1 400
+2 500
+3 \N
+
diff --git
a/regression-test/suites/datatype_p0/agg_state/test_agg_state_finalize.groovy
b/regression-test/suites/datatype_p0/agg_state/test_agg_state_finalize.groovy
new file mode 100644
index 00000000000..5285a97d554
--- /dev/null
+++
b/regression-test/suites/datatype_p0/agg_state/test_agg_state_finalize.groovy
@@ -0,0 +1,104 @@
+// 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("test_agg_state_finalize") {
+ sql "set enable_agg_state=true"
+ sql "drop table if exists agg_finalize_input"
+ sql """
+ create table agg_finalize_input (k int, v int, s string, d decimal(12,
2))
+ duplicate key(k) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql """
+ insert into agg_finalize_input values
+ (1, 1, 'a', 1.25), (1, 3, 'b', 3.75),
+ (2, 10, 'c', 10.00), (3, null, null, null)
+ """
+ order_qt_rows """
+ select k, avg_finalize(avg_state(v)), count_finalize(count_state(v)),
+ sum_finalize(sum_state(v)), min_finalize(min_state(v)),
max_finalize(max_state(v))
+ from agg_finalize_input
+ """
+ order_qt_grouped """
+ select k, avg_finalize(a), count_finalize(c), sum_finalize(s),
min_finalize(mi), max_finalize(ma)
+ from (select k, avg_combine(v) a, count_combine(v) c, sum_combine(v) s,
+ min_combine(v) mi, max_combine(v) ma from
agg_finalize_input group by k) t
+ """
+ order_qt_empty """
+ select avg_finalize(a), count_finalize(c), sum_finalize(s)
+ from (select avg_combine(v) a, count_combine(v) c, sum_combine(v) s
+ from agg_finalize_input where k < 0) t
+ """
+ order_qt_decimal """
+ select k, avg_finalize(a)
+ from (select k, avg_combine(d) a from agg_finalize_input group by k) t
+ """
+ order_qt_array """
+ select k, array_sort(array_agg_finalize(a))
+ from (select k, array_agg_combine(s) a from agg_finalize_input group
by k) t
+ """
+ order_qt_parameters """
+ select topn_finalize(s) from (select topn_combine('a', 2) s) t
+ """
+ order_qt_const """
+ select number, avg_finalize(avg_state(7)),
count_finalize(count_state(cast(null as int)))
+ from numbers("number"="3")
+ """
+ // The finest groups need only a scalar projection; coarser groups merge
their states.
+ order_qt_rollup """
+ with partial as (select k, avg_combine(v) a from agg_finalize_input
group by k)
+ select k, avg_finalize(a) from partial
+ union all
+ select null, avg_merge(a) from partial
+ """
+ sql "drop table if exists agg_finalize_stored"
+ sql """
+ create table agg_finalize_stored (
+ k int, a agg_state<avg(int)> generic, c agg_state<count(int)>
generic,
+ s agg_state<sum(int)> generic
+ ) aggregate key(k) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql """
+ insert into agg_finalize_stored
+ select k, avg_state(v), count_state(v), sum_state(v) from
agg_finalize_input
+ """
+ order_qt_stored """
+ select k, avg_finalize(a), count_finalize(c), sum_finalize(s) from
agg_finalize_stored
+ """
+ order_qt_outer_null """
+ select n.number, avg_finalize(t.a), count_finalize(t.c)
+ from numbers("number"="5") n left join agg_finalize_stored t on
n.number=t.k
+ """
+ order_qt_union """
+ select avg_finalize(a) from (select avg_union(a) a from
agg_finalize_stored) t
+ """
+ order_qt_alias """
+ select var_pop_finalize(variance_state(v)),
variance_finalize(var_pop_state(v))
+ from agg_finalize_input
+ """
+ for (def query : ["select sum_finalize(avg_state(1))", "select
avg_finalize(sum_state(1))"]) {
+ test {
+ sql query
+ exception "requires a state of"
+ }
+ }
+ test {
+ sql "select avg_finalize(1)"
+ exception "Can not found function"
+ }
+}
diff --git
a/regression-test/suites/datatype_p0/agg_state/test_agg_state_union_batch.groovy
b/regression-test/suites/datatype_p0/agg_state/test_agg_state_union_batch.groovy
new file mode 100644
index 00000000000..87de9d8bc1c
--- /dev/null
+++
b/regression-test/suites/datatype_p0/agg_state/test_agg_state_union_batch.groovy
@@ -0,0 +1,79 @@
+// 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("test_agg_state_union_batch") {
+ sql "set enable_agg_state=true"
+ sql "drop table if exists agg_state_union_batch_input"
+ sql """
+ create table agg_state_union_batch_input (
+ k int not null, v bigint not null, n bigint, d decimal(27, 9), s
string
+ ) duplicate key(k) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql """
+ insert into agg_state_union_batch_input
+ select number % 7, number,
+ if(number % 7 = 6 or number % 5 = 0, null, number),
+ if(number % 7 = 6, null, cast(number as decimal(27, 9)) / 100),
+ if(number % 7 = 6, null, concat('value-', cast(number as
string)))
+ from numbers("number"="10000")
+ """
+ explain {
+ sql "select k, sum_merge(sum_state(v)) from
agg_state_union_batch_input group by k"
+ contains "sum_merge"
+ }
+ explain {
+ sql "select k, sum_union(sum_state(v)) from
agg_state_union_batch_input group by k"
+ contains "sum_union"
+ }
+ // Multiple input batches, repeated groups, multiple state offsets, and an
all-NULL group.
+ order_qt_merge """
+ select k, sum_merge(sum_state(v)), sum_merge(sum_state(n)),
+ sum_merge(sum_state(d)), min_merge(min_state(s))
+ from agg_state_union_batch_input group by k
+ """
+ order_qt_direct """
+ select k, sum(v), sum(n), sum(d), min(s)
+ from agg_state_union_batch_input group by k
+ """
+ order_qt_union """
+ select k, sum_finalize(sv), sum_finalize(sn), sum_finalize(sd),
min_finalize(ms)
+ from (
+ select k, sum_union(sum_state(v)) sv, sum_union(sum_state(n)) sn,
+ sum_union(sum_state(d)) sd, min_union(min_state(s)) ms
+ from agg_state_union_batch_input group by k
+ ) t
+ """
+ // ARRAY_AGG owns nontrivial state. Read its result after all batch
scratch storage is freed.
+ order_qt_array """
+ with merged as (
+ select k, array_sort(array_agg_merge(array_agg_state(s))) a
+ from agg_state_union_batch_input group by k
+ ), united as (
+ select k, array_sort(array_agg_finalize(a)) a
+ from (
+ select k, array_agg_union(array_agg_state(s)) a
+ from agg_state_union_batch_input group by k
+ ) t
+ ), direct as (
+ select k, array_sort(array_agg(s)) a
+ from agg_state_union_batch_input group by k
+ )
+ select d.k, m.a <=> d.a, u.a <=> d.a
+ from direct d join merged m on d.k=m.k join united u on d.k=u.k
+ """
+}
diff --git
a/regression-test/suites/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.groovy
b/regression-test/suites/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.groovy
new file mode 100644
index 00000000000..cf57a756b49
--- /dev/null
+++
b/regression-test/suites/nereids_rules_p0/mv/agg_with_roll_up/agg_state_finalize_roll_up.groovy
@@ -0,0 +1,92 @@
+// 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("agg_state_finalize_roll_up") {
+ String db = context.config.getDbNameByFile(context.file)
+ sql "set enable_agg_state=true"
+ sql "set pre_materialized_view_rewrite_strategy=TRY_IN_RBO"
+ for (String mv : ["finalize_max_mv", "finalize_abs_mv", "finalize_sum_mv",
"finalize_union_mv"]) {
+ sql "drop materialized view if exists ${mv}"
+ }
+ sql "drop table if exists agg_finalize_rollup_input"
+ sql """
+ create table agg_finalize_rollup_input (k int, g int, v double)
+ duplicate key(k, g) distributed by hash(k) buckets 1
+ properties("replication_num"="1")
+ """
+ sql """
+ insert into agg_finalize_rollup_input values
+ (1, 1, -4), (1, 1, 3), (1, 2, -2), (1, 2, 7),
+ (2, 1, 5), (2, 2, null), (3, 1, null)
+ """
+ // Make the finer-group MV cheaper than scanning the base table so the
test executes its rollup.
+ sql """
+ insert into agg_finalize_rollup_input
+ select k, g, v from agg_finalize_rollup_input cross join
numbers("number"="99")
+ """
+ sql "analyze table agg_finalize_rollup_input with sync"
+
+ String combinedQuery = """
+ select k, sum_finalize(sum_combine(avg_finalize(avg_state(v))))
+ from agg_finalize_rollup_input group by k
+ """
+ String plainQuery = """
+ select k, sum(avg_finalize(avg_state(v)))
+ from agg_finalize_rollup_input group by k
+ """
+ order_qt_combined_before combinedQuery
+ order_qt_plain_before plainQuery
+
+ // Sharing the inner AVG finalizer must not make SUM compatible with MAX.
+ create_async_mv(db, "finalize_max_mv", """
+ select k, g, max_combine(avg_finalize(avg_state(v))) s
+ from agg_finalize_rollup_input group by k, g
+ """)
+ mv_rewrite_fail(combinedQuery, "finalize_max_mv")
+ mv_rewrite_fail(plainQuery, "finalize_max_mv")
+ order_qt_different_aggregate combinedQuery
+
+ // The full value expression matters even when both outer aggregates are
SUM.
+ create_async_mv(db, "finalize_abs_mv", """
+ select k, g, sum_combine(abs(avg_finalize(avg_state(v)))) s
+ from agg_finalize_rollup_input group by k, g
+ """)
+ mv_rewrite_fail(combinedQuery, "finalize_abs_mv")
+ mv_rewrite_fail(plainQuery, "finalize_abs_mv")
+ order_qt_different_argument combinedQuery
+
+ create_async_mv(db, "finalize_sum_mv", """
+ select k, g, sum_combine(avg_finalize(avg_state(v))) s
+ from agg_finalize_rollup_input group by k, g
+ """)
+ mv_rewrite_success(combinedQuery, "finalize_sum_mv")
+ mv_rewrite_success(plainQuery, "finalize_sum_mv")
+ order_qt_combined_after combinedQuery
+ order_qt_plain_after plainQuery
+
+ // Existing MERGE/UNION/STATE rollup remains available.
+ create_async_mv(db, "finalize_union_mv", """
+ select k, g, sum_union(sum_state(v)) s
+ from agg_finalize_rollup_input group by k, g
+ """)
+ String mergeQuery = """
+ select k, sum_merge(sum_state(v))
+ from agg_finalize_rollup_input group by k
+ """
+ mv_rewrite_success(mergeQuery, "finalize_union_mv")
+ order_qt_merge_after mergeQuery
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
