be/src/exprs/function/function_agg_state_finalize.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 <cstring> |
21 | | |
22 | | #include "core/arena.h" |
23 | | #include "core/block/block.h" |
24 | | #include "core/column/column_nullable.h" |
25 | | #include "exprs/aggregate/aggregate_function.h" |
26 | | #include "exprs/function/function.h" |
27 | | #include "util/defer_op.h" |
28 | | |
29 | | namespace doris { |
30 | | |
31 | | class FunctionAggStateFinalize : public IFunction { |
32 | | public: |
33 | | FunctionAggStateFinalize(DataTypePtr return_type, AggregateFunctionPtr agg_function) |
34 | 24 | : _return_type(std::move(return_type)), _agg_function(std::move(agg_function)) {} |
35 | | |
36 | | static FunctionBasePtr create(const DataTypes& argument_types, const DataTypePtr& return_type, |
37 | 21 | const AggregateFunctionPtr& agg_function) { |
38 | 21 | return std::make_shared<DefaultFunction>( |
39 | 21 | std::make_shared<FunctionAggStateFinalize>(return_type, agg_function), |
40 | 21 | argument_types, return_type); |
41 | 21 | } |
42 | | |
43 | 0 | String get_name() const override { return _agg_function->get_name() + "_finalize"; } |
44 | | |
45 | 0 | size_t get_number_of_arguments() const override { return 1; } |
46 | | |
47 | | // An outer NULL can have an empty, invalid serialized payload. Do not deserialize it. |
48 | 20 | bool use_default_implementation_for_nulls() const override { return false; } |
49 | | |
50 | 0 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
51 | 0 | return _return_type; |
52 | 0 | } |
53 | | |
54 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
55 | 23 | uint32_t result, size_t input_rows_count) const override { |
56 | 23 | const auto input = |
57 | 23 | block.get_by_position(arguments[0]).column->convert_to_full_column_if_const(); |
58 | 23 | const auto* nullable = check_and_get_column<ColumnNullable>(*input); |
59 | 23 | const auto& states = nullable ? nullable->get_nested_column() : *input; |
60 | 23 | auto output = _agg_function->get_return_type()->create_column(); |
61 | 23 | _agg_function->check_result_column_type(*output); |
62 | 23 | output->reserve(input_rows_count); |
63 | | // Keep the reusable state storage separate from its per-row variable-length data. |
64 | 23 | Arena state_arena; |
65 | 23 | const auto state_size = _agg_function->size_of_data(); |
66 | 23 | auto* place = state_arena.aligned_alloc(state_size, _agg_function->align_of_data()); |
67 | 23 | Arena arena; |
68 | 205 | auto finalize_row = [&](size_t row) { |
69 | | // Serialized states can use native columns as well as strings. |
70 | 205 | _agg_function->deserialize_from_column_row(place, states, row, arena); |
71 | 205 | _agg_function->insert_result_into(place, *output); |
72 | 205 | }; |
73 | 23 | auto finalize_rows = [&]<bool is_trivial>() { |
74 | 248 | for (size_t row = 0; row < input_rows_count; ++row) { |
75 | 225 | if (nullable && nullable->is_null_at(row)) { |
76 | 20 | output->insert_default(); |
77 | 20 | continue; |
78 | 20 | } |
79 | 205 | if constexpr (is_trivial) { |
80 | | // Trivial states support zero-init and need no destruction. |
81 | 20 | std::memset(place, 0, state_size); |
82 | 20 | finalize_row(row); |
83 | 185 | } else { |
84 | 185 | _agg_function->create(place); |
85 | 185 | DEFER(_agg_function->destroy(place)); |
86 | 185 | finalize_row(row); |
87 | 185 | } |
88 | | // Destroy states before reclaiming any variable-length data they own. |
89 | 205 | arena.clear(); |
90 | 205 | } |
91 | 23 | }; _ZZNK5doris24FunctionAggStateFinalize12execute_implEPNS_15FunctionContextERNS_5BlockERKSt6vectorIjSaIjEEjmENKUlTnbvE_clILb1EEEDav Line | Count | Source | 73 | 8 | auto finalize_rows = [&]<bool is_trivial>() { | 74 | 30 | for (size_t row = 0; row < input_rows_count; ++row) { | 75 | 22 | if (nullable && nullable->is_null_at(row)) { | 76 | 2 | output->insert_default(); | 77 | 2 | continue; | 78 | 2 | } | 79 | 20 | if constexpr (is_trivial) { | 80 | | // Trivial states support zero-init and need no destruction. | 81 | 20 | std::memset(place, 0, state_size); | 82 | 20 | finalize_row(row); | 83 | | } else { | 84 | | _agg_function->create(place); | 85 | | DEFER(_agg_function->destroy(place)); | 86 | | finalize_row(row); | 87 | | } | 88 | | // Destroy states before reclaiming any variable-length data they own. | 89 | 20 | arena.clear(); | 90 | 20 | } | 91 | 8 | }; |
_ZZNK5doris24FunctionAggStateFinalize12execute_implEPNS_15FunctionContextERNS_5BlockERKSt6vectorIjSaIjEEjmENKUlTnbvE_clILb0EEEDav Line | Count | Source | 73 | 15 | auto finalize_rows = [&]<bool is_trivial>() { | 74 | 218 | for (size_t row = 0; row < input_rows_count; ++row) { | 75 | 203 | if (nullable && nullable->is_null_at(row)) { | 76 | 18 | output->insert_default(); | 77 | 18 | continue; | 78 | 18 | } | 79 | | if constexpr (is_trivial) { | 80 | | // Trivial states support zero-init and need no destruction. | 81 | | std::memset(place, 0, state_size); | 82 | | finalize_row(row); | 83 | 185 | } else { | 84 | 185 | _agg_function->create(place); | 85 | 185 | DEFER(_agg_function->destroy(place)); | 86 | 185 | finalize_row(row); | 87 | 185 | } | 88 | | // Destroy states before reclaiming any variable-length data they own. | 89 | 185 | arena.clear(); | 90 | 185 | } | 91 | 15 | }; |
|
92 | 23 | if (_agg_function->is_trivial()) { |
93 | 8 | finalize_rows.template operator()<true>(); |
94 | 15 | } else { |
95 | 15 | finalize_rows.template operator()<false>(); |
96 | 15 | } |
97 | 23 | ColumnPtr result_column = std::move(output); |
98 | 23 | if (_return_type->is_nullable()) { |
99 | 10 | result_column = wrap_in_nullable(result_column, block, arguments, input_rows_count); |
100 | 10 | } |
101 | 23 | block.replace_by_position(result, std::move(result_column)); |
102 | 23 | return Status::OK(); |
103 | 23 | } |
104 | | |
105 | | private: |
106 | | DataTypePtr _return_type; |
107 | | AggregateFunctionPtr _agg_function; |
108 | | }; |
109 | | |
110 | | } // namespace doris |