github-actions[bot] commented on code in PR #66942:
URL: https://github.com/apache/doris/pull/66942#discussion_r3843175512
##########
be/src/exprs/vectorized_agg_fn.cpp:
##########
@@ -256,7 +292,7 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const
RowDescriptor& desc,
_sort_description, state);
}
- if (_fn.name.function_name == "ai_agg") {
+ if (_fn.name.function_name.starts_with("ai_agg")) {
Review Comment:
[P1] Keep QueryContext setup scoped to the actual AI aggregate
This prefix also matches unrelated aggregate UDFs with valid names such as
`ai_agg_custom`. The JAVA_UDF/PYTHON_UDF/RPC branches construct those functions
before reaching this line, and none overrides
`IAggregateFunction::set_query_context()`, so `SELECT ai_agg_custom(k) ...` now
throws the fatal AI-only base implementation during BE prepare. The current FE
code rejects all real `ai_agg_{state,combine,merge,union}` wrappers, so please
restore the exact `ai_agg` check (or dispatch on an explicit AI capability) and
cover a non-AI UDAF with this prefix.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/AggCombinerFunctionBuilder.java:
##########
@@ -159,7 +181,8 @@ public String parameterDisplayString() {
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(FOREACH_SUFFIX);
+ || name.toLowerCase().endsWith(UNION_SUFFIX) ||
name.toLowerCase().endsWith(COMBINE_SUFFIX)
Review Comment:
[P1] Classify `_combine` as aggregate before binding its arguments
This dynamically adds `_combine` to function resolution, but
`FunctionRegistry.isAggregateFunction()` still checks only exact registered
names. HAVING/ORDER BY use that classifier before choosing the argument scope.
For example, in `SELECT id, sum(age + 1) AS age FROM t GROUP BY id HAVING
avg_combine(age) IS NOT NULL`, `avg_combine` is initially treated as scalar, so
its `age` binds to the aggregate-output alias instead of `t.age`; only
afterward is the function synthesized as an aggregate, leaving an invalid
same-level dependency and an analysis failure. Please make pre-binding
classification agree with dynamic combinator resolution (while preserving UDF
precedence), and cover HAVING/ORDER BY alias collisions.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/ConvertAggStateCast.java:
##########
@@ -59,7 +60,7 @@ public static Expression convert(Cast cast) {
while (child instanceof Alias) {
child = ((Alias) child).child();
}
- if (child instanceof StateCombinator) {
+ if (child instanceof StateCombinator || child instanceof
CombineCombinator) {
Review Comment:
[P1] Validate AggState arity before rewriting combine children
`ConvertAggStateCast` runs before `CheckCast`, and this new combine branch
indexes the target subtype/nullability lists once for every source child
without first matching their sizes. For example, casting three-argument
`topn_combine(v, 10, 100)` to `AGG_STATE<topn(VARCHAR, INT)>` reaches index 2
of a two-entry target list, throwing an unchecked list-index error instead of
the normal incompatible-cast `AnalysisException`. Please validate
function/arity metadata before this loop (or leave incompatible casts for
`CheckCast`) and add a mismatched-arity negative test.
##########
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);
Review Comment:
[P1] Preserve one serialized state per group
`AggregateStateCombine` inherits `insert_result_into_vec()`, whose loop
calls this method once for every group. That is not safe as an append API: for
valid `count_combine(nullable_col)`, the nested
`AggregateFunctionCountNotNullUnary::serialize_without_key_to_column()` does
`col.resize(1)` and rewrites element 0 on every call. A grouped query with two
or more keys therefore emits N key rows but only one state row (or only the
last state), because the production aggregation source requests `num_rows`
results here while the direct test exercises only one place. Please implement
batch/state output with the nested vector serialization contract (and preserve
existing destination rows for the single-result path), then add a multi-group
nullable `count_combine`/`count_merge` 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]
