github-actions[bot] commented on code in PR #68312:
URL: https://github.com/apache/doris/pull/68312#discussion_r4217446484
##########
be/src/exprs/aggregate/aggregate_function_state_union.h:
##########
@@ -107,6 +108,32 @@ class AggregateStateUnion : public
AggregateFunctionNonFinalBase,
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;
+ auto* 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],
Review Comment:
[P1] Bound deserialization of wide state batches. In a one-phase grouped
`min_merge`/`max_merge` over nonnullable STRING states, the string min/max
vector reader deserializes every row before merging, keeping a whole batch of
long-string copies live instead of the previous one-row scratch. A 4096-row
batch of 128 KiB states can add roughly 512 MiB of peak memory. In the selected
LIMIT path (for example, `agg_phase=1`, `push_topn_to_agg=false`, `LIMIT 1`),
it also deserializes rows whose `places[i]` is null, whereas the former
selected `add` skipped them. Please skip rejected rows before decoding and
bound scratch for active rows; cover a wide all-skipped batch.
##########
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) {
Review Comment:
[P2] Reuse scratch when finalizing string-backed states. This call runs once
per input row, and the generic `deserialize_and_merge_from_column_range`
allocates `std::vector<char>(size_of_data())`, constructs a second state,
merges it, and destroys it each time. `collect_set` and `topn` use that generic
path, so finalizing one million stored states incurs one million small heap
allocations in addition to the destination state already allocated here. Reuse
aligned scratch across the block or add a direct one-state finalization path.
##########
be/src/exprs/aggregate/aggregate_function_state_union.h:
##########
@@ -107,6 +108,32 @@ class AggregateStateUnion : public
AggregateFunctionNonFinalBase,
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;
Review Comment:
[P2] Skip scratch allocation for direct native state readers.
`add_batch_impl` allocates `batch_size * size_of_data()` on every nonempty
batch, but direct nonnullable SUM, AVG, and COUNT vector readers ignore `rhs`
and merge from fixed-length columns. A 4096-row block creates and frees this
unused buffer for each such aggregate expression, adding allocator work to the
hot path this change aims to speed up. Expose whether the nested reader needs
scratch and allocate it only when it does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
