github-actions[bot] commented on code in PR #66942: URL: https://github.com/apache/doris/pull/66942#discussion_r3840668653
########## be/src/exprs/aggregate/aggregate_function_state_combine.h: ########## @@ -0,0 +1,220 @@ +// 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 <utility> + +#include "exprs/aggregate/aggregate_function.h" + +namespace doris { +const static std::string AGG_COMBINE_SUFFIX = "_combine"; + +class AggregateStateCombine final : public IAggregateFunctionHelper<AggregateStateCombine> { +public: + AggregateStateCombine(AggregateFunctionPtr function, const DataTypes& argument_types_, + DataTypePtr return_type) + : IAggregateFunctionHelper(argument_types_), + _function(std::move(function)), + _return_type(std::move(return_type)) {} + + static AggregateFunctionPtr create(AggregateFunctionPtr function, + const DataTypes& argument_types_, + const DataTypePtr& return_type) { + if (function == nullptr) { + return nullptr; + } + return std::make_shared<AggregateStateCombine>(function, argument_types_, return_type); + } + + void set_version(const int version_) override { + IAggregateFunctionHelper::set_version(version_); + _function->set_version(version_); + } + + void create(AggregateDataPtr __restrict place) const override { _function->create(place); } + + void destroy_vec(AggregateDataPtr __restrict place, + const size_t num_rows) const noexcept override { + _function->destroy_vec(place, num_rows); + } + + String get_name() const override { return _function->get_name() + AGG_COMBINE_SUFFIX; } + + DataTypePtr get_return_type() const override { return _return_type; } + + void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num, + Arena& arena) const override { + _function->add(place, columns, row_num, arena); + } + + void add_batch(size_t batch_size, AggregateDataPtr* places, size_t place_offset, + const IColumn** columns, Arena& arena, bool agg_many) const override { + _function->add_batch(batch_size, places, place_offset, columns, arena, agg_many); + } + + void add_batch_selected(size_t batch_size, AggregateDataPtr* places, size_t place_offset, + const IColumn** columns, Arena& arena) const override { + _function->add_batch_selected(batch_size, places, place_offset, columns, arena); + } + + void add_batch_single_place(size_t batch_size, AggregateDataPtr place, const IColumn** columns, + Arena& arena) const override { + _function->add_batch_single_place(batch_size, place, columns, arena); + } + + void add_batch_range(size_t batch_begin, size_t batch_end, AggregateDataPtr place, + const IColumn** columns, Arena& arena, bool has_null) override { + _function->add_batch_range(batch_begin, batch_end, place, columns, arena, has_null); + } + + void add_range_single_place(int64_t partition_start, int64_t partition_end, int64_t frame_start, + int64_t frame_end, AggregateDataPtr place, const IColumn** columns, + Arena& arena, UInt8* use_null_result, + UInt8* could_use_previous_result) const override { + _function->add_range_single_place(partition_start, partition_end, frame_start, frame_end, + place, columns, arena, use_null_result, + could_use_previous_result); + } + + void reset(AggregateDataPtr place) const override { _function->reset(place); } + + void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, + Arena& arena) const override { + _function->merge(place, rhs, arena); + } + + void merge_vec(const AggregateDataPtr __restrict* __restrict places, size_t offset, + ConstAggregateDataPtr __restrict rhs, Arena& arena, + const size_t num_rows) const override { + _function->merge_vec(places, offset, rhs, arena, num_rows); + } + + void merge_vec_selected(const AggregateDataPtr __restrict* __restrict places, size_t offset, + ConstAggregateDataPtr __restrict rhs, Arena& arena, + const size_t num_rows) const override { + _function->merge_vec_selected(places, offset, rhs, arena, num_rows); + } + + void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override { + _function->serialize(place, buf); + } + + void serialize_vec(const std::vector<AggregateDataPtr>& places, size_t offset, + BufferWritable& buf, const size_t num_rows) const override { + _function->serialize_vec(places, offset, buf, num_rows); + } + + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, + MutableColumnPtr& dst, const size_t num_rows) const override { + _function->serialize_to_column(places, offset, dst, num_rows); + } + + void serialize_without_key_to_column(ConstAggregateDataPtr __restrict place, + IColumn& to) const override { + _function->serialize_without_key_to_column(place, to); + } + + void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, + Arena& arena) const override { + _function->deserialize(place, buf, arena); + } + + void deserialize_vec(AggregateDataPtr places, const ColumnString* column, Arena& arena, + size_t num_rows) const override { + _function->deserialize_vec(places, column, arena, num_rows); + } + + 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 { + _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, const size_t num_rows) const override { + _function->deserialize_and_merge_vec_selected(places, offset, rhs, column, arena, num_rows); + } + + void deserialize_and_merge(AggregateDataPtr __restrict place, AggregateDataPtr __restrict rhs, + BufferReadable& buf, Arena& arena) const override { + _function->deserialize_and_merge(place, rhs, buf, arena); + } + + void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict place, + const IColumn& column, size_t begin, size_t end, + Arena& arena) const override { + _function->deserialize_and_merge_from_column_range(place, column, begin, end, arena); + } + + void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override { Review Comment: [P1] Preserve merged states with terminal serialization In the required two-phase path for `orthogonal_bitmap_expr_calculate[_count]_combine`, local serialization finalizes the raw calculator into a separate `result`, and global merge accumulates only that field. This finalizer serializes the merged state again; the nested `write()` then overwrites `result` from the global state's empty raw calculator, emitting zero or an empty bitmap. The matching `_merge` therefore returns the wrong value. Please gate `_combine` on an explicit merge-stable serialization capability or preserve the merged serialized result, and add forced two-phase count and bitmap tests. ########## be/src/exprs/vectorized_agg_fn.cpp: ########## @@ -174,46 +175,81 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const RowDescriptor& desc, } else if (_fn.binary_type == TFunctionBinaryType::RPC) { _function = AggregateRpcUdaf::create(_fn, argument_types, _data_type); } else if (_fn.binary_type == TFunctionBinaryType::AGG_STATE) { - if (argument_types.size() != 1) { - return Status::InternalError("Agg state Function must input 1 argument but get {}", - argument_types.size()); - } - if (argument_types[0]->is_nullable()) { - return Status::InternalError("Agg state function input type must be not nullable"); - } - if (argument_types[0]->get_primitive_type() != PrimitiveType::TYPE_AGG_STATE) { - return Status::InternalError( - "Agg state function input type must be agg_state but get {}", - argument_types[0]->get_family_name()); - } - - std::string type_function_name = - assert_cast<const DataTypeAggState*>(argument_types[0].get())->get_function_name(); - if (type_function_name + AGG_UNION_SUFFIX == _fn.name.function_name) { + if (match_suffix(_fn.name.function_name, AGG_COMBINE_SUFFIX)) { if (_data_type->is_nullable()) { return Status::InternalError( - "Union function return type must be not nullable, real={}", + "Combine function return type must be not nullable, real={}", _data_type->get_name()); } if (_data_type->get_primitive_type() != PrimitiveType::TYPE_AGG_STATE) { return Status::InternalError( - "Union function return type must be AGG_STATE, real={}", + "Combine function return type must be AGG_STATE, real={}", _data_type->get_name()); } - _function = get_agg_state_function<AggregateStateUnion>(argument_types, _data_type); - } else if (type_function_name + AGG_MERGE_SUFFIX == _fn.name.function_name) { - auto type = assert_cast<const DataTypeAggState*>(argument_types[0].get()) - ->get_nested_function() - ->get_return_type(); - if (!type->equals(*_data_type)) { - return Status::InternalError("{}'s expect return type is {}, but input {}", - argument_types[0]->get_name(), type->get_name(), + const auto* state_type = assert_cast<const DataTypeAggState*>(_data_type.get()); + if (state_type->get_function_name() + AGG_COMBINE_SUFFIX != _fn.name.function_name) { + return Status::InternalError("{} not match return type {}", _fn.name.function_name, _data_type->get_name()); } - _function = get_agg_state_function<AggregateStateMerge>(argument_types, _data_type); + const auto& expected_argument_types = state_type->get_sub_types(); + if (argument_types.size() != expected_argument_types.size()) { + return Status::InternalError("Combine function {} expects {} arguments but gets {}", + _fn.name.function_name, expected_argument_types.size(), + argument_types.size()); + } + for (size_t i = 0; i < argument_types.size(); ++i) { + if (!argument_types[i]->equals(*expected_argument_types[i])) { + return Status::InternalError( + "Combine function {} argument {} expects {}, but gets {}", + _fn.name.function_name, i, expected_argument_types[i]->get_name(), + argument_types[i]->get_name()); + } + } + _function = AggregateStateCombine::create(state_type->get_nested_function(), Review Comment: [P1] Build a window-aware nested state for `_combine` This reuses the `DataTypeAggState` nested function, which is constructed with the default `is_window_function=false`, even when this evaluator represents an analytic window. The wrapper still forwards incremental support. For nullable Avg, a bounded-frame transition from a non-null row to an all-null frame reaches a window-only DCHECK in debug builds; in release it never tracks `null_count` and serializes a populated count-zero state, so merging that state produces a non-NULL/NaN result instead of NULL. Please construct a window-aware nested function or disable incremental mode for this case, and test non-null-to-all-null frame transitions. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/rollup/SingleCombinatorRollupHandler.java: ########## @@ -53,7 +54,8 @@ public boolean canRollup(AggregateFunction queryAggregateFunction, return false; } if (!(queryAggregateFunction instanceof Combinator) - && (viewFunction instanceof UnionCombinator || viewFunction instanceof StateCombinator)) { + && (viewFunction instanceof UnionCombinator || viewFunction instanceof StateCombinator + || viewFunction instanceof CombineCombinator)) { Review Comment: [P1] Store combine state directly in synchronous MVs Allowing `CombineCombinator` here makes it a valid view-side roll-up expression, but `CreateMaterializedViewCommand` still treats every generic aggregate as a raw aggregate and wraps it with `StateCombinator.create()`. For `avg_combine(v)`, that constructs a synthetic `avg_combine_state` and null-unboxes `aggFuncNameNullableMap.get("avg_combine")`, so `CREATE MATERIALIZED VIEW ... avg_combine(v)` fails before rewrite. Please recognize `_combine` as already state-valued (or reject this MV form deliberately) and cover MV creation plus roll-up. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/CombineCombinator.java: ########## @@ -0,0 +1,158 @@ +// 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.BuiltinAggregateFunctions; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.FunctionRegistry; +import org.apache.doris.catalog.FunctionSignature; +import org.apache.doris.common.Pair; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.OrderExpression; +import org.apache.doris.nereids.trees.expressions.functions.AggCombinerFunctionBuilder; +import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable; +import org.apache.doris.nereids.trees.expressions.functions.BoundFunction; +import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature; +import org.apache.doris.nereids.trees.expressions.functions.ExpressionTrait; +import org.apache.doris.nereids.trees.expressions.functions.Function; +import org.apache.doris.nereids.trees.expressions.functions.FunctionBuilder; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunctionParams; +import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait; +import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor; +import org.apache.doris.nereids.types.AggStateType; +import org.apache.doris.nereids.types.DataType; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; + +/** + * Aggregate inputs into the nested function's serialized state. + */ +public class CombineCombinator extends AggregateFunction + implements ExplicitlyCastableSignature, AlwaysNotNullable, Combinator, RollUpTrait { Review Comment: [P1] Preserve the non-null empty state across scalar subqueries This aggregate promises a non-null state through `AlwaysNotNullable`, but scalar-subquery nullability adjustment and correlated empty-input repair only recognize `NotNullableAggregateFunction`. As a result, `(select avg_combine(v) from r)` is exposed as nullable, and feeding that state to `_merge` or `_union` reaches BE as `Nullable(AggState)`, which `AggFnEvaluator::prepare()` rejects. Correlated unnesting can also substitute SQL NULL for an unmatched key instead of the serialized empty state. Please integrate `_combine` with the scalar-aggregate empty-input contract (or prevent these rewrites/uses) and cover uncorrelated plus unmatched-correlated cases. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java: ########## @@ -139,6 +146,12 @@ public Pair<BoundFunction, AggregateFunction> build(String name, List<?> argumen arguments = arguments.subList(1, arguments.size()); } return Pair.of(new StateCombinator((List<Expression>) arguments, nestedFunction), nestedFunction); + } else if (combinatorSuffix.equalsIgnoreCase(COMBINE)) { + AggregateFunction nestedFunction = buildState(nestedName, arguments); + if (!arguments.isEmpty() && arguments.get(0) instanceof Boolean && (Boolean) arguments.get(0)) { + throw new IllegalStateException(name + " doesn't support DISTINCT"); + } + return Pair.of(new CombineCombinator((List<Expression>) arguments, nestedFunction), nestedFunction); Review Comment: [P1] Preserve constructor-expanded arguments in the combine expression `AIAgg(text, task)` prepends the default resource and canonicalizes itself to three children, but this branch discards `nestedFunction.children()` and builds `CombineCombinator` from the original two arguments. The resulting `AggStateType` and translated TExpr contain only two columns, while the BE AI aggregate unconditionally reads `columns[2]`, causing an out-of-bounds access. Please propagate the nested function's canonical children (while handling the DISTINCT marker) or reject arity-changing builders, and cover the default-resource overload. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/combinator/CombineCombinator.java: ########## @@ -0,0 +1,158 @@ +// 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.BuiltinAggregateFunctions; +import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.FunctionRegistry; +import org.apache.doris.catalog.FunctionSignature; +import org.apache.doris.common.Pair; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.OrderExpression; +import org.apache.doris.nereids.trees.expressions.functions.AggCombinerFunctionBuilder; +import org.apache.doris.nereids.trees.expressions.functions.AlwaysNotNullable; +import org.apache.doris.nereids.trees.expressions.functions.BoundFunction; +import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature; +import org.apache.doris.nereids.trees.expressions.functions.ExpressionTrait; +import org.apache.doris.nereids.trees.expressions.functions.Function; +import org.apache.doris.nereids.trees.expressions.functions.FunctionBuilder; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction; +import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunctionParams; +import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait; +import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor; +import org.apache.doris.nereids.types.AggStateType; +import org.apache.doris.nereids.types.DataType; + +import com.google.common.collect.ImmutableList; + +import java.util.List; +import java.util.Objects; + +/** + * Aggregate inputs into the nested function's serialized state. + */ +public class CombineCombinator extends AggregateFunction + implements ExplicitlyCastableSignature, AlwaysNotNullable, Combinator, RollUpTrait { + + private final AggregateFunction nested; + private final AggStateType returnType; + + /** Constructor of CombineCombinator. */ + public CombineCombinator(List<Expression> arguments, AggregateFunction nested) { + super(nested.getName() + AggCombinerFunctionBuilder.COMBINE_SUFFIX, arguments); + checkArguments(arguments, nested); + this.nested = Objects.requireNonNull(nested, "nested can not be null"); + this.returnType = createReturnType(arguments, nested); + } + + private CombineCombinator(AggregateFunctionParams functionParams, AggregateFunction nested) { + super(functionParams); + checkArguments(functionParams.arguments, nested); + this.nested = Objects.requireNonNull(nested, "nested can not be null"); + this.returnType = createReturnType(functionParams.arguments, nested); + } + + private static void checkArguments(List<Expression> arguments, AggregateFunction nested) { + if (arguments.isEmpty()) { + throw new AnalysisException(String.format( + "%s_combine requires at least one argument", nested.getName())); + } + for (Expression argument : arguments) { + if (argument instanceof OrderExpression) { + throw new AnalysisException(String.format( + "%s_combine doesn't support order by expression", nested.getName())); + } + } + } + + private static AggStateType createReturnType(List<Expression> arguments, AggregateFunction nested) { + return new AggStateType(nested.getName(), + arguments.stream().map(ExpressionTrait::getDataType) + .collect(ImmutableList.toImmutableList()), + arguments.stream().map(ExpressionTrait::nullable) + .collect(ImmutableList.toImmutableList()), + BuiltinAggregateFunctions.INSTANCE.aggFuncNameNullableMap.get(nested.getName())); + } + + @Override + public CombineCombinator withChildren(List<Expression> children) { + return new CombineCombinator(getFunctionParams(children), nested); + } + + @Override + public AggregateFunction withDistinctAndChildren(boolean distinct, List<Expression> children) { + if (distinct) { + throw new AnalysisException(getName() + " doesn't support DISTINCT"); + } + return new CombineCombinator(getFunctionParams(false, children), nested); + } + + @Override + public List<FunctionSignature> getSignatures() { + return nested.getSignatures().stream() + .map(signature -> signature.withReturnType(returnType)) + .collect(ImmutableList.toImmutableList()); + } + + @Override + public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) { + return visitor.visitCombineCombinator(this, context); + } + + @Override + public DataType getDataType() { + return returnType; + } + + @Override + public AggregateFunction getNestedFunction() { + return nested; + } + + @Override + protected List<DataType> intermediateTypes() { Review Comment: [P1] Delegate the nested aggregate's phase support The wrapper delegates `intermediateTypes()` here but inherits the base `supportAggregatePhase()`, which returns true for every phase. Consequently `orthogonal_bitmap_expr_calculate[_count]_combine(...)` appears eligible for a forced `agg_phase=1` `INPUT_TO_RESULT` plan even though both nested aggregates explicitly support only `AggregatePhase.TWO`. Please delegate this policy to `nested` and add a phase-selection test for a two-phase-only aggregate; the supported two-phase serialization path also needs the separate state-preservation fix. ########## be/src/exprs/aggregate/aggregate_function_state_combine.h: ########## @@ -0,0 +1,220 @@ +// 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 <utility> + +#include "exprs/aggregate/aggregate_function.h" + +namespace doris { +const static std::string AGG_COMBINE_SUFFIX = "_combine"; + +class AggregateStateCombine final : public IAggregateFunctionHelper<AggregateStateCombine> { +public: + AggregateStateCombine(AggregateFunctionPtr function, const DataTypes& argument_types_, + DataTypePtr return_type) + : IAggregateFunctionHelper(argument_types_), + _function(std::move(function)), + _return_type(std::move(return_type)) {} + + static AggregateFunctionPtr create(AggregateFunctionPtr function, + const DataTypes& argument_types_, + const DataTypePtr& return_type) { + if (function == nullptr) { + return nullptr; + } + return std::make_shared<AggregateStateCombine>(function, argument_types_, return_type); + } + + void set_version(const int version_) override { + IAggregateFunctionHelper::set_version(version_); + _function->set_version(version_); + } + + void create(AggregateDataPtr __restrict place) const override { _function->create(place); } + + void destroy_vec(AggregateDataPtr __restrict place, + const size_t num_rows) const noexcept override { + _function->destroy_vec(place, num_rows); + } + + String get_name() const override { return _function->get_name() + AGG_COMBINE_SUFFIX; } + + DataTypePtr get_return_type() const override { return _return_type; } + + void add(AggregateDataPtr __restrict place, const IColumn** columns, ssize_t row_num, + Arena& arena) const override { + _function->add(place, columns, row_num, arena); + } + + void add_batch(size_t batch_size, AggregateDataPtr* places, size_t place_offset, + const IColumn** columns, Arena& arena, bool agg_many) const override { + _function->add_batch(batch_size, places, place_offset, columns, arena, agg_many); + } + + void add_batch_selected(size_t batch_size, AggregateDataPtr* places, size_t place_offset, + const IColumn** columns, Arena& arena) const override { + _function->add_batch_selected(batch_size, places, place_offset, columns, arena); + } + + void add_batch_single_place(size_t batch_size, AggregateDataPtr place, const IColumn** columns, + Arena& arena) const override { + _function->add_batch_single_place(batch_size, place, columns, arena); + } + + void add_batch_range(size_t batch_begin, size_t batch_end, AggregateDataPtr place, + const IColumn** columns, Arena& arena, bool has_null) override { + _function->add_batch_range(batch_begin, batch_end, place, columns, arena, has_null); + } + + void add_range_single_place(int64_t partition_start, int64_t partition_end, int64_t frame_start, + int64_t frame_end, AggregateDataPtr place, const IColumn** columns, + Arena& arena, UInt8* use_null_result, + UInt8* could_use_previous_result) const override { + _function->add_range_single_place(partition_start, partition_end, frame_start, frame_end, + place, columns, arena, use_null_result, + could_use_previous_result); + } + + void reset(AggregateDataPtr place) const override { _function->reset(place); } + + void merge(AggregateDataPtr __restrict place, ConstAggregateDataPtr rhs, + Arena& arena) const override { + _function->merge(place, rhs, arena); + } + + void merge_vec(const AggregateDataPtr __restrict* __restrict places, size_t offset, + ConstAggregateDataPtr __restrict rhs, Arena& arena, + const size_t num_rows) const override { + _function->merge_vec(places, offset, rhs, arena, num_rows); + } + + void merge_vec_selected(const AggregateDataPtr __restrict* __restrict places, size_t offset, + ConstAggregateDataPtr __restrict rhs, Arena& arena, + const size_t num_rows) const override { + _function->merge_vec_selected(places, offset, rhs, arena, num_rows); + } + + void serialize(ConstAggregateDataPtr __restrict place, BufferWritable& buf) const override { + _function->serialize(place, buf); + } + + void serialize_vec(const std::vector<AggregateDataPtr>& places, size_t offset, + BufferWritable& buf, const size_t num_rows) const override { + _function->serialize_vec(places, offset, buf, num_rows); + } + + void serialize_to_column(const std::vector<AggregateDataPtr>& places, size_t offset, + MutableColumnPtr& dst, const size_t num_rows) const override { + _function->serialize_to_column(places, offset, dst, num_rows); + } + + void serialize_without_key_to_column(ConstAggregateDataPtr __restrict place, + IColumn& to) const override { + _function->serialize_without_key_to_column(place, to); + } + + void deserialize(AggregateDataPtr __restrict place, BufferReadable& buf, + Arena& arena) const override { + _function->deserialize(place, buf, arena); + } + + void deserialize_vec(AggregateDataPtr places, const ColumnString* column, Arena& arena, + size_t num_rows) const override { + _function->deserialize_vec(places, column, arena, num_rows); + } + + 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 { + _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, const size_t num_rows) const override { + _function->deserialize_and_merge_vec_selected(places, offset, rhs, column, arena, num_rows); + } + + void deserialize_and_merge(AggregateDataPtr __restrict place, AggregateDataPtr __restrict rhs, + BufferReadable& buf, Arena& arena) const override { + _function->deserialize_and_merge(place, rhs, buf, arena); + } + + void deserialize_and_merge_from_column_range(AggregateDataPtr __restrict place, + const IColumn& column, size_t begin, size_t end, + Arena& arena) const override { + _function->deserialize_and_merge_from_column_range(place, column, begin, end, arena); + } + + void insert_result_into(ConstAggregateDataPtr __restrict place, IColumn& to) const override { + _function->serialize_without_key_to_column(place, to); + } + + void streaming_agg_serialize_to_column(const IColumn** columns, MutableColumnPtr& dst, + const size_t num_rows, Arena& arena) const override { + _function->streaming_agg_serialize_to_column(columns, dst, num_rows, arena); + } + + void destroy(AggregateDataPtr __restrict place) const noexcept override { + _function->destroy(place); + } + + bool is_trivial() const override { return _function->is_trivial(); } + + size_t size_of_data() const override { return _function->size_of_data(); } + + size_t align_of_data() const override { return _function->align_of_data(); } + + void check_input_columns_type(const IColumn** columns) const override { + _function->check_input_columns_type(columns); + } + + MutableColumnPtr create_serialize_column() const override { + return _function->create_serialize_column(); + } + + DataTypePtr get_serialized_type() const override { return _function->get_serialized_type(); } + + bool supported_incremental_mode() const override { + return _function->supported_incremental_mode(); + } + + void execute_function_with_incremental(int64_t partition_start, int64_t partition_end, + int64_t frame_start, int64_t frame_end, + AggregateDataPtr place, const IColumn** columns, + Arena& arena, bool previous_is_nul, bool end_is_nul, + bool has_null, UInt8* use_null_result, + UInt8* could_use_previous_result) const override { + _function->execute_function_with_incremental( + partition_start, partition_end, frame_start, frame_end, place, columns, arena, + previous_is_nul, end_is_nul, has_null, use_null_result, could_use_previous_result); + } + + void set_query_context(QueryContext* context) override { Review Comment: [P1] Propagate QueryContext to the wrapped AI aggregate `FunctionRegistry` now accepts `ai_agg_combine(...)`, but `AggFnEvaluator::prepare()` calls `set_query_context()` only when the outer name is exactly `ai_agg`. For `ai_agg_combine`, this forwarding method is never called; `AggregateFunctionAIAgg::create()` stores a null context and its first `add()` dereferences `_ctx` in `prepare()`, which can crash the BE. Please key this setup off the nested aggregate/capability (or explicitly handle the combinator) and cover a valid three-argument AI combine. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java: ########## @@ -65,7 +68,11 @@ public Class<? extends BoundFunction> functionClass() { @Override public boolean canApply(List<?> arguments) { - if (combinatorSuffix.equalsIgnoreCase(STATE) || combinatorSuffix.equalsIgnoreCase(FOREACH)) { + if (combinatorSuffix.equalsIgnoreCase(COMBINE)) { + // DataTypeAggState needs at least one subtype, so zero-argument aggregates such as + // count(*) cannot produce an AggState yet. count_combine(1) remains supported. + return !arguments.isEmpty() && nestedBuilder.canApply(arguments); Review Comment: [P2] Filter `_combine` resolution to aggregate builders This branch only checks the nested builder's signature, while dynamic suffix lookup tries every built-in registered under the base name. Thus `abs_combine(1)` retains the scalar `Abs` builder and `buildState()` later casts the built scalar to `AggregateFunction`, throwing an internal `ClassCastException` instead of a normal analysis error. Require the nested builder's class to be an aggregate before advertising this combinator, and add a negative scalar-name test. -- 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]
