be/src/exprs/function/function_agg_state.h
Line | Count | Source |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | |
18 | | #pragma once |
19 | | |
20 | | #include <fmt/format.h> |
21 | | |
22 | | #include "common/status.h" |
23 | | #include "core/arena.h" |
24 | | #include "core/block/block.h" |
25 | | #include "core/block/columns_with_type_and_name.h" |
26 | | #include "core/column/column.h" |
27 | | #include "core/column/column_const.h" |
28 | | #include "core/column/column_nullable.h" |
29 | | #include "core/data_type/data_type.h" |
30 | | #include "core/data_type/data_type_agg_state.h" |
31 | | #include "core/types.h" |
32 | | #include "exprs/aggregate/aggregate_function.h" |
33 | | #include "exprs/function/function.h" |
34 | | |
35 | | namespace doris { |
36 | | |
37 | | class FunctionAggState : public IFunction { |
38 | | public: |
39 | | FunctionAggState(const DataTypes& argument_types, const DataTypePtr& return_type, |
40 | | AggregateFunctionPtr agg_function) |
41 | 0 | : _argument_types(argument_types), |
42 | 0 | _return_type(return_type), |
43 | 0 | _agg_function(agg_function), |
44 | 0 | _always_const_argument_idx(argument_types.size(), false) { |
45 | 0 | for (auto index : _agg_function->get_const_argument_indexes()) { |
46 | 0 | DORIS_CHECK_LT(index, _always_const_argument_idx.size()); |
47 | 0 | _always_const_argument_idx[index] = true; |
48 | 0 | } |
49 | 0 | } |
50 | | |
51 | | static FunctionBasePtr create(const DataTypes& argument_types, const DataTypePtr& return_type, |
52 | 0 | AggregateFunctionPtr agg_function) { |
53 | 0 | if (agg_function == nullptr) { |
54 | 0 | return nullptr; |
55 | 0 | } |
56 | 0 | return std::make_shared<DefaultFunction>( |
57 | 0 | std::make_shared<FunctionAggState>(argument_types, return_type, agg_function), |
58 | 0 | argument_types, return_type); |
59 | 0 | } |
60 | | |
61 | 0 | size_t get_number_of_arguments() const override { return _argument_types.size(); } |
62 | | |
63 | 0 | bool use_default_implementation_for_nulls() const override { return false; } |
64 | | |
65 | 0 | String get_name() const override { return fmt::format("{}_state", _agg_function->get_name()); } |
66 | | |
67 | 0 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
68 | 0 | return _return_type; |
69 | 0 | } |
70 | | |
71 | 0 | const std::vector<size_t>& get_const_argument_indexes() const override { |
72 | 0 | return _agg_function->get_const_argument_indexes(); |
73 | 0 | } |
74 | | |
75 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
76 | 0 | uint32_t result, size_t input_rows_count) const override { |
77 | 0 | auto col = _agg_function->create_serialize_column(); |
78 | 0 | std::vector<const IColumn*> agg_columns; |
79 | 0 | std::vector<ColumnPtr> save_columns; |
80 | |
|
81 | 0 | for (size_t i = 0; i < arguments.size(); i++) { |
82 | 0 | DataTypePtr signature = |
83 | 0 | assert_cast<const DataTypeAggState*>(_return_type.get())->get_sub_types()[i]; |
84 | 0 | ColumnPtr column; |
85 | 0 | if (_always_const_argument_idx[i]) { |
86 | 0 | column = block.get_by_position(arguments[i]).column; |
87 | 0 | if (const auto* const_column = check_and_get_column<ColumnConst>(*column)) { |
88 | 0 | column = const_column->get_data_column_ptr(); |
89 | 0 | } |
90 | 0 | column = ColumnConst::create(std::move(column), input_rows_count); |
91 | 0 | } else { |
92 | 0 | column = block.get_by_position(arguments[i]) |
93 | 0 | .column->convert_to_full_column_if_const(); |
94 | 0 | } |
95 | 0 | save_columns.push_back(column); |
96 | |
|
97 | 0 | if (!signature->is_nullable() && column->is_nullable()) { |
98 | 0 | return Status::InternalError( |
99 | 0 | "State function meet input nullable column, but signature is not nullable"); |
100 | 0 | } |
101 | 0 | if (!column->is_nullable() && signature->is_nullable()) { |
102 | 0 | column = make_nullable(column); |
103 | 0 | save_columns.push_back(column); |
104 | 0 | } |
105 | |
|
106 | 0 | agg_columns.push_back(column.get()); |
107 | 0 | } |
108 | 0 | _agg_function->check_input_columns_type(agg_columns.data()); |
109 | 0 | _agg_function->streaming_agg_serialize_to_column(agg_columns.data(), col, input_rows_count, |
110 | 0 | context->get_arena()); |
111 | 0 | block.replace_by_position(result, std::move(col)); |
112 | 0 | return Status::OK(); |
113 | 0 | } |
114 | | |
115 | | private: |
116 | | DataTypes _argument_types; |
117 | | DataTypePtr _return_type; |
118 | | AggregateFunctionPtr _agg_function; |
119 | | // Bool map of argument positions that must be constant by aggregate semantics. |
120 | | // FE checks these positions before execution. |
121 | | std::vector<bool> _always_const_argument_idx; |
122 | | }; |
123 | | |
124 | | } // namespace doris |