be/src/exprs/function/function_quantile_state.cpp
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 | | // This file is copied from |
18 | | // https://github.com/ClickHouse/ClickHouse/blob/master/src/Functions/FunctionBitmap.h |
19 | | // and modified by Doris |
20 | | |
21 | | #include <fmt/format.h> |
22 | | #include <glog/logging.h> |
23 | | |
24 | | #include <boost/iterator/iterator_facade.hpp> |
25 | | #include <cstddef> |
26 | | #include <memory> |
27 | | #include <utility> |
28 | | |
29 | | #include "common/compiler_util.h" // IWYU pragma: keep |
30 | | #include "common/status.h" |
31 | | #include "core/assert_cast.h" |
32 | | #include "core/block/block.h" |
33 | | #include "core/block/column_numbers.h" |
34 | | #include "core/block/column_with_type_and_name.h" |
35 | | #include "core/column/column.h" |
36 | | #include "core/column/column_complex.h" |
37 | | #include "core/column/column_const.h" |
38 | | #include "core/column/column_nullable.h" |
39 | | #include "core/column/column_string.h" |
40 | | #include "core/column/column_vector.h" |
41 | | #include "core/data_type/data_type.h" |
42 | | #include "core/data_type/data_type_nullable.h" |
43 | | #include "core/data_type/data_type_number.h" |
44 | | #include "core/data_type/data_type_quantilestate.h" // IWYU pragma: keep |
45 | | #include "core/data_type/data_type_string.h" |
46 | | #include "core/types.h" |
47 | | #include "core/value/quantile_state.h" |
48 | | #include "exec/common/util.hpp" |
49 | | #include "exprs/aggregate/aggregate_function.h" |
50 | | #include "exprs/function/function.h" |
51 | | #include "exprs/function/function_const.h" |
52 | | #include "exprs/function/function_helpers.h" |
53 | | #include "exprs/function/function_totype.h" |
54 | | #include "exprs/function/simple_function_factory.h" |
55 | | #include "util/url_coding.h" |
56 | | |
57 | | namespace doris { |
58 | | class FunctionContext; |
59 | | } // namespace doris |
60 | | |
61 | | namespace doris { |
62 | | |
63 | | struct QuantileStateEmpty { |
64 | | static constexpr auto name = "quantile_state_empty"; |
65 | | using ReturnColVec = ColumnQuantileState; |
66 | 0 | static DataTypePtr get_return_type() { return std::make_shared<DataTypeQuantileState>(); } |
67 | 0 | static auto init_value() { return QuantileState {}; } |
68 | | }; |
69 | | |
70 | | class FunctionToQuantileState : public IFunction { |
71 | | public: |
72 | | static constexpr auto name = "to_quantile_state"; |
73 | 1 | String get_name() const override { return name; } |
74 | | |
75 | 2 | static FunctionPtr create() { return std::make_shared<FunctionToQuantileState>(); } |
76 | | |
77 | 0 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
78 | 0 | return std::make_shared<DataTypeQuantileState>(); |
79 | 0 | } |
80 | | |
81 | 0 | size_t get_number_of_arguments() const override { return 2; } |
82 | | |
83 | 0 | bool use_default_implementation_for_nulls() const override { return false; } |
84 | | |
85 | | template <bool is_nullable> |
86 | | Status execute_internal(const ColumnPtr& column, const DataTypePtr& data_type, |
87 | 0 | MutableColumnPtr& column_result, float compression) const { |
88 | 0 | auto type_error = [&]() { |
89 | 0 | return Status::RuntimeError("Illegal column {} of argument of function {}", |
90 | 0 | column->get_name(), get_name()); |
91 | 0 | }; |
92 | 0 | const ColumnNullable* col_nullable = nullptr; |
93 | 0 | const ColumnUInt8* col_nullmap = nullptr; |
94 | 0 | const ColumnFloat64* col = nullptr; |
95 | 0 | const NullMap* nullmap = nullptr; |
96 | 0 | if constexpr (is_nullable) { |
97 | 0 | col_nullable = check_and_get_column<ColumnNullable>(column.get()); |
98 | 0 | col_nullmap = check_and_get_column<ColumnUInt8>( |
99 | 0 | col_nullable->get_null_map_column_ptr().get()); |
100 | 0 | col = check_and_get_column<ColumnFloat64>(col_nullable->get_nested_column_ptr().get()); |
101 | 0 | if (col == nullptr || col_nullmap == nullptr) { |
102 | 0 | return type_error(); |
103 | 0 | } |
104 | | |
105 | 0 | nullmap = &col_nullmap->get_data(); |
106 | 0 | } else { |
107 | 0 | col = check_and_get_column<ColumnFloat64>(column.get()); |
108 | 0 | } |
109 | 0 | auto* res_column = reinterpret_cast<ColumnQuantileState*>(column_result.get()); |
110 | 0 | auto& res_data = res_column->get_data(); |
111 | |
|
112 | 0 | size_t size = col->size(); |
113 | 0 | for (size_t i = 0; i < size; ++i) { |
114 | 0 | if constexpr (is_nullable) { |
115 | 0 | if ((*nullmap)[i]) { |
116 | 0 | res_data[i].clear(); |
117 | 0 | continue; |
118 | 0 | } |
119 | 0 | } |
120 | 0 | auto value = (double)col->get_data()[i]; |
121 | 0 | res_data[i].set_compression(compression); |
122 | 0 | res_data[i].add_value(value); |
123 | 0 | } |
124 | 0 | return Status::OK(); |
125 | 0 | } Unexecuted instantiation: _ZNK5doris23FunctionToQuantileState16execute_internalILb1EEENS_6StatusERKNS_3COWINS_7IColumnEE13immutable_ptrIS4_EERKSt10shared_ptrIKNS_9IDataTypeEERNS5_11mutable_ptrIS4_EEf Unexecuted instantiation: _ZNK5doris23FunctionToQuantileState16execute_internalILb0EEENS_6StatusERKNS_3COWINS_7IColumnEE13immutable_ptrIS4_EERKSt10shared_ptrIKNS_9IDataTypeEERNS5_11mutable_ptrIS4_EEf |
126 | | |
127 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
128 | 0 | uint32_t result, size_t input_rows_count) const override { |
129 | 0 | const ColumnPtr& column = block.get_by_position(arguments[0]).column; |
130 | 0 | const DataTypePtr& data_type = block.get_by_position(arguments[0]).type; |
131 | 0 | const auto* compression_arg = check_and_get_column_const<ColumnFloat32>( |
132 | 0 | block.get_by_position(arguments.back()).column.get()); |
133 | 0 | float compression = 2048; |
134 | 0 | if (compression_arg) { |
135 | 0 | auto compression_arg_val = compression_arg->get_value<TYPE_FLOAT>(); |
136 | 0 | if (compression_arg_val >= QUANTILE_STATE_COMPRESSION_MIN && |
137 | 0 | compression_arg_val <= QUANTILE_STATE_COMPRESSION_MAX) { |
138 | 0 | compression = compression_arg_val; |
139 | 0 | } |
140 | 0 | } |
141 | 0 | MutableColumnPtr column_result = get_return_type_impl({})->create_column(); |
142 | 0 | column_result->resize(input_rows_count); |
143 | |
|
144 | 0 | Status status = Status::OK(); |
145 | 0 | if (data_type->is_nullable()) { |
146 | 0 | RETURN_IF_ERROR(execute_internal<true>(column, data_type, column_result, compression)); |
147 | 0 | } else { |
148 | 0 | RETURN_IF_ERROR(execute_internal<false>(column, data_type, column_result, compression)); |
149 | 0 | } |
150 | 0 | if (status.ok()) { |
151 | 0 | block.replace_by_position(result, std::move(column_result)); |
152 | 0 | } |
153 | 0 | return status; |
154 | 0 | } |
155 | | }; |
156 | | |
157 | | class FunctionQuantileStatePercent : public IFunction { |
158 | | public: |
159 | | static constexpr auto name = "quantile_percent"; |
160 | 1 | String get_name() const override { return name; } |
161 | | |
162 | 2 | static FunctionPtr create() { return std::make_shared<FunctionQuantileStatePercent>(); } |
163 | | |
164 | 0 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
165 | 0 | return std::make_shared<DataTypeFloat64>(); |
166 | 0 | } |
167 | | |
168 | 0 | size_t get_number_of_arguments() const override { return 2; } |
169 | | |
170 | 0 | bool use_default_implementation_for_nulls() const override { return false; } |
171 | | |
172 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
173 | 0 | uint32_t result, size_t input_rows_count) const override { |
174 | 0 | auto res_data_column = ColumnFloat64::create(); |
175 | 0 | auto& res = res_data_column->get_data(); |
176 | 0 | auto data_null_map = ColumnUInt8::create(input_rows_count, 0); |
177 | 0 | auto& null_map = data_null_map->get_data(); |
178 | |
|
179 | 0 | auto column = block.get_by_position(arguments[0]).column->convert_to_full_column_if_const(); |
180 | 0 | if (const auto* nullable = check_and_get_column<const ColumnNullable>(*column)) { |
181 | 0 | VectorizedUtils::update_null_map(null_map, nullable->get_null_map_data()); |
182 | 0 | column = nullable->get_nested_column_ptr(); |
183 | 0 | } |
184 | 0 | const auto* str_col = assert_cast<const ColumnQuantileState*>(column.get()); |
185 | 0 | const auto& col_data = str_col->get_data(); |
186 | 0 | const auto* percent_arg = check_and_get_column_const<ColumnFloat32>( |
187 | 0 | block.get_by_position(arguments.back()).column.get()); |
188 | |
|
189 | 0 | if (!percent_arg) { |
190 | 0 | return Status::InvalidArgument( |
191 | 0 | "Second argument to {} must be a constant float describing type", get_name()); |
192 | 0 | } |
193 | 0 | auto percent_arg_value = percent_arg->get_value<TYPE_FLOAT>(); |
194 | 0 | if (percent_arg_value < 0 || percent_arg_value > 1) { |
195 | 0 | return Status::InvalidArgument( |
196 | 0 | "the input argument of percentage: {} is not valid, must be in range [0,1] ", |
197 | 0 | percent_arg_value); |
198 | 0 | } |
199 | | |
200 | 0 | res.reserve(input_rows_count); |
201 | 0 | for (size_t i = 0; i < input_rows_count; ++i) { |
202 | 0 | if (null_map[i]) { |
203 | | // if null push_back meaningless result to make sure idxs can be matched |
204 | 0 | res.push_back(0); |
205 | 0 | continue; |
206 | 0 | } |
207 | | |
208 | 0 | res.push_back(col_data[i].get_value_by_percentile(percent_arg_value)); |
209 | 0 | } |
210 | |
|
211 | 0 | block.replace_by_position(result, std::move(res_data_column)); |
212 | 0 | return Status::OK(); |
213 | 0 | } |
214 | | }; |
215 | | |
216 | | class FunctionQuantileStateFromBase64 : public IFunction { |
217 | | public: |
218 | | static constexpr auto name = "quantile_state_from_base64"; |
219 | 1 | String get_name() const override { return name; } |
220 | | |
221 | 2 | static FunctionPtr create() { return std::make_shared<FunctionQuantileStateFromBase64>(); } |
222 | | |
223 | 0 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
224 | 0 | return std::make_shared<DataTypeNullable>(std::make_shared<DataTypeQuantileState>()); |
225 | 0 | } |
226 | | |
227 | 0 | size_t get_number_of_arguments() const override { return 1; } |
228 | | |
229 | 0 | bool use_default_implementation_for_nulls() const override { return true; } |
230 | | |
231 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
232 | 0 | uint32_t result, size_t input_rows_count) const override { |
233 | 0 | auto res_null_map = ColumnUInt8::create(input_rows_count, 0); |
234 | 0 | auto res_data_column = ColumnQuantileState::create(); |
235 | 0 | auto& null_map = res_null_map->get_data(); |
236 | 0 | auto& res = res_data_column->get_data(); |
237 | |
|
238 | 0 | auto& argument_column = block.get_by_position(arguments[0]).column; |
239 | 0 | const auto& str_column = static_cast<const ColumnString&>(*argument_column); |
240 | 0 | const ColumnString::Chars& data = str_column.get_chars(); |
241 | 0 | const ColumnString::Offsets& offsets = str_column.get_offsets(); |
242 | |
|
243 | 0 | res.reserve(input_rows_count); |
244 | |
|
245 | 0 | std::string decode_buff; |
246 | 0 | int64_t last_decode_buff_len = 0; |
247 | 0 | int64_t curr_decode_buff_len = 0; |
248 | 0 | for (size_t i = 0; i < input_rows_count; ++i) { |
249 | 0 | const char* src_str = reinterpret_cast<const char*>(&data[offsets[i - 1]]); |
250 | 0 | int64_t src_size = offsets[i] - offsets[i - 1]; |
251 | |
|
252 | 0 | if (src_size == 0 || 0 != src_size % 4) { |
253 | 0 | res.emplace_back(); |
254 | 0 | null_map[i] = 1; |
255 | 0 | continue; |
256 | 0 | } |
257 | | |
258 | 0 | curr_decode_buff_len = src_size + 3; |
259 | 0 | if (curr_decode_buff_len > last_decode_buff_len) { |
260 | 0 | decode_buff.resize(curr_decode_buff_len); |
261 | 0 | last_decode_buff_len = curr_decode_buff_len; |
262 | 0 | } |
263 | 0 | auto outlen = base64_decode(src_str, src_size, decode_buff.data()); |
264 | 0 | if (outlen < 0) { |
265 | 0 | res.emplace_back(); |
266 | 0 | null_map[i] = 1; |
267 | 0 | } else { |
268 | 0 | doris::Slice decoded_slice(decode_buff.data(), outlen); |
269 | 0 | doris::QuantileState quantile_state; |
270 | 0 | if (!quantile_state.deserialize(decoded_slice)) { |
271 | 0 | return Status::RuntimeError(fmt::format( |
272 | 0 | "quantile_state_from_base64 decode failed: base64: {}", src_str)); |
273 | 0 | } else { |
274 | 0 | res.emplace_back(std::move(quantile_state)); |
275 | 0 | } |
276 | 0 | } |
277 | 0 | } |
278 | | |
279 | 0 | block.get_by_position(result).column = |
280 | 0 | ColumnNullable::create(std::move(res_data_column), std::move(res_null_map)); |
281 | 0 | return Status::OK(); |
282 | 0 | } |
283 | | }; |
284 | | |
285 | | struct NameQuantileStateToBase64 { |
286 | | static constexpr auto name = "quantile_state_to_base64"; |
287 | | }; |
288 | | |
289 | | struct QuantileStateToBase64 { |
290 | | using ReturnType = DataTypeString; |
291 | | static constexpr auto PrimitiveTypeImpl = PrimitiveType::TYPE_QUANTILE_STATE; |
292 | | using Type = DataTypeQuantileState::FieldType; |
293 | | using ReturnColumnType = ColumnString; |
294 | | using Chars = ColumnString::Chars; |
295 | | using Offsets = ColumnString::Offsets; |
296 | | |
297 | 1 | static Status vector(const std::vector<QuantileState>& data, Chars& chars, Offsets& offsets) { |
298 | 1 | size_t size = data.size(); |
299 | 1 | offsets.resize(size); |
300 | 1 | size_t output_char_size = 0; |
301 | 6 | for (size_t i = 0; i < size; ++i) { |
302 | 5 | auto& quantile_state_val = const_cast<QuantileState&>(data[i]); |
303 | 5 | auto ser_size = quantile_state_val.get_serialized_size(); |
304 | 5 | output_char_size += (int)(4.0 * ceil((double)ser_size / 3.0)); |
305 | 5 | } |
306 | 1 | ColumnString::check_chars_length(output_char_size, size); |
307 | 1 | chars.resize(output_char_size); |
308 | 1 | auto* chars_data = chars.data(); |
309 | | |
310 | 1 | size_t cur_ser_size = 0; |
311 | 1 | size_t last_ser_size = 0; |
312 | 1 | std::string ser_buff; |
313 | 1 | size_t encoded_offset = 0; |
314 | 6 | for (size_t i = 0; i < size; ++i) { |
315 | 5 | auto& quantile_state_val = const_cast<QuantileState&>(data[i]); |
316 | | |
317 | 5 | cur_ser_size = quantile_state_val.get_serialized_size(); |
318 | 5 | if (cur_ser_size > last_ser_size) { |
319 | 5 | last_ser_size = cur_ser_size; |
320 | 5 | ser_buff.resize(cur_ser_size); |
321 | 5 | } |
322 | 5 | size_t real_size = |
323 | 5 | quantile_state_val.serialize(reinterpret_cast<uint8_t*>(ser_buff.data())); |
324 | 5 | auto outlen = base64_encode((const unsigned char*)ser_buff.data(), real_size, |
325 | 5 | chars_data + encoded_offset); |
326 | 5 | DCHECK(outlen > 0); |
327 | | |
328 | 5 | encoded_offset += outlen; |
329 | 5 | offsets[i] = cast_set<uint32_t>(encoded_offset); |
330 | 5 | } |
331 | 1 | return Status::OK(); |
332 | 1 | } |
333 | | }; |
334 | | |
335 | | using FunctionQuantileStateToBase64 = |
336 | | FunctionUnaryToType<QuantileStateToBase64, NameQuantileStateToBase64>; |
337 | | |
338 | 1 | void register_function_quantile_state(SimpleFunctionFactory& factory) { |
339 | 1 | factory.register_function<FunctionConst<QuantileStateEmpty, false>>(); |
340 | 1 | factory.register_function<FunctionQuantileStatePercent>(); |
341 | 1 | factory.register_function<FunctionToQuantileState>(); |
342 | 1 | factory.register_function<FunctionQuantileStateFromBase64>(); |
343 | 1 | factory.register_function<FunctionQuantileStateToBase64>(); |
344 | 1 | } |
345 | | |
346 | | } // namespace doris |