HappenLee commented on code in PR #68022:
URL: https://github.com/apache/doris/pull/68022#discussion_r4059164725
##########
be/src/exprs/function/array/function_array_split.cpp:
##########
@@ -54,31 +57,51 @@ class FunctionArraySplit : public IFunction {
size_t get_number_of_arguments() const override { return 2; }
+ bool use_default_implementation_for_nulls() const override { return false;
}
+
DataTypePtr get_return_type_impl(const DataTypes& arguments) const
override {
- return std::make_shared<DataTypeArray>(make_nullable(arguments[0]));
+ auto result_type =
+
std::make_shared<DataTypeArray>(make_nullable(remove_nullable(arguments[0])));
+ return have_nullable(arguments) ? make_nullable(result_type) :
result_type;
};
Status execute_impl(FunctionContext* context, Block& block, const
ColumnNumbers& arguments,
uint32_t result, size_t input_rows_count) const
override {
- // <Nullable>(Array(<Nullable>(Int)))
+ ColumnUInt8::MutablePtr result_null_map;
+ ColumnUInt8::Container* result_null_map_data = nullptr;
+ if (block.get_by_position(result).type->is_nullable()) {
+ result_null_map = ColumnUInt8::create(input_rows_count, 0);
+ result_null_map_data = &result_null_map->get_data();
+ }
auto src_column =
block.get_by_position(arguments[0]).column->convert_to_full_column_if_const();
Review Comment:
[P2] Restore the all-NULL fast path before expanding the source
Disabling `use_default_implementation_for_nulls()` removes the wrapper's
`only_null()` pre-check, but this implementation does not replace it. With
short-circuit evaluation disabled, a large constant source array plus a
**non-constant nullable predicate column whose entire block is NULL** now
expands the source across all rows here. `do_loop` later skips the NULL rows
but preserves the expanded source payload, even though every visible result is
NULL. Previously, `PreparedFunctionImpl::default_implementation_for_nulls`
returned a constant NULL before invoking this code.
For 4096 rows and a constant array of 4096 INTs, this adds at least 64 MiB
of value data, plus nested null maps and offsets, and O(rows × array length)
work to an otherwise cheap NULL result. `array_reverse_split` is affected as
well.
Please check both original arguments for `only_null()` before any
materialization and return a constant NULL using the result type, as the
updated enumerate/filter paths already do. Add a peak-memory test with a
constant source and a **non-constant** all-NULL predicate for both split
directions; making both arguments constant would let the framework reduce
execution to one row and miss this regression.
##########
be/src/exprs/aggregate/aggregate_function_null.h:
##########
@@ -630,6 +630,63 @@ class AggregateFunctionNullVariadicInline final
arena);
}
+ void add_batch(size_t batch_size, AggregateDataPtr* places, size_t
place_offset,
+ const IColumn** columns, Arena& arena, bool /*agg_many*/)
const override {
+ add_batch_selected_impl(batch_size, columns, arena,
+ [&](size_t row) { return places[row] +
place_offset; });
+ }
+
+ void add_batch_selected(size_t batch_size, AggregateDataPtr* places,
size_t place_offset,
+ const IColumn** columns, Arena& arena) const
override {
+ add_batch_selected_impl(batch_size, columns, arena, [&](size_t row) {
+ return places[row] == nullptr ? nullptr : places[row] +
place_offset;
+ });
+ }
+
+ void add_batch_single_place(size_t batch_size, AggregateDataPtr place,
const IColumn** columns,
+ Arena& arena) const override {
+ add_batch_selected_impl(batch_size, columns, arena, [&](size_t) {
return place; });
+ }
+
+ void streaming_agg_serialize_to_column(const IColumn** columns,
MutableColumnPtr& dst,
+ size_t num_rows, Arena& arena)
const override {
+ if constexpr (!requires(const NestFuction& function) {
+ function.requires_batch_add_for_streaming();
+ }) {
+ AggregateFunctionNullBaseInline<
+ NestFuction, result_is_nullable,
+ AggregateFunctionNullVariadicInline<NestFuction,
result_is_nullable>>::
+ streaming_agg_serialize_to_column(columns, dst, num_rows,
arena);
+ return;
+ }
+
+ const size_t state_size = this->size_of_data();
+ std::vector<char> states(state_size * num_rows);
+ std::vector<AggregateDataPtr> places(num_rows);
+ size_t created_states = 0;
+ try {
+ for (; created_states < num_rows; ++created_states) {
+ places[created_states] = states.data() + state_size *
created_states;
+ this->create(places[created_states]);
+ }
+
+ add_batch(num_rows, places.data(), 0, columns, arena, false);
Review Comment:
[P2] Keep streaming foreach states bounded instead of retaining the whole
batch
The previous generic streaming implementation performed `create → add →
serialize → destroy` for each row. This branch creates and populates all row
states before serializing or destroying any of them. The marker selects this
path for multi-argument nullable `_foreach` even when every row is non-NULL and
all input offsets are already aligned, so this changes the ordinary streaming
path as well.
A concrete case is `SELECT id, min_by_foreach(s, k) FROM t GROUP BY id`,
with nullable ARRAY<STRING> `s` and matching key arrays, when high-cardinality
aggregation takes `StreamingAggLocalState`'s direct-serialization path.
`min_by` retains long strings in `SingleValueDataString::large_data`, which is
released when the state is destroyed. Those buffers used to be freed after each
row; now they remain live for the entire batch. For 4096 rows × 32 elements ×
4096-byte strings, these buffers alone grow from about 128 KiB for one row to
about 512 MiB for the batch. This estimate excludes the input, serialized
output, and Arena-backed state storage that both implementations retain.
Please separate batch input normalization from state lifetime: normalize the
selected inputs once, then serialize and destroy row states incrementally, or
at least retain the old streaming path for aligned inputs. Add a peak-memory
test using an aggregate with real heap-owned string state; the current PairSum
test checks batch normalization and results but cannot expose this lifetime
regression.
--
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]
