be/src/exprs/table_function/vexplode_v2.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 | | |
18 | | #include "exprs/table_function/vexplode_v2.h" |
19 | | |
20 | | #include <glog/logging.h> |
21 | | |
22 | | #include <algorithm> |
23 | | #include <cstdint> |
24 | | #include <ostream> |
25 | | #include <span> |
26 | | |
27 | | #include "common/status.h" |
28 | | #include "core/assert_cast.h" |
29 | | #include "core/block/block.h" |
30 | | #include "core/block/column_with_type_and_name.h" |
31 | | #include "core/column/column.h" |
32 | | #include "core/column/column_array.h" |
33 | | #include "core/column/column_nothing.h" |
34 | | #include "core/column/column_struct.h" |
35 | | #include "core/column/column_variant.h" |
36 | | #include "core/column/column_variant_v2.h" |
37 | | #include "core/data_type/data_type.h" |
38 | | #include "core/data_type/data_type_array.h" |
39 | | #include "core/data_type/data_type_nothing.h" |
40 | | #include "core/data_type/data_type_nullable.h" |
41 | | #include "core/data_type/data_type_variant.h" |
42 | | #include "core/data_type/primitive_type.h" |
43 | | #include "exprs/function/cast/cast_variant_v2_internal.h" |
44 | | #include "exprs/function/function_helpers.h" |
45 | | #include "exprs/vexpr.h" |
46 | | #include "exprs/vexpr_context.h" |
47 | | |
48 | | namespace doris { |
49 | | |
50 | 47 | VExplodeV2TableFunction::VExplodeV2TableFunction() { |
51 | 47 | _fn_name = "vexplode"; |
52 | 47 | } |
53 | | |
54 | | Status VExplodeV2TableFunction::_process_init_variant(Block* block, int value_column_idx, |
55 | 0 | int children_column_idx) { |
56 | | // explode variant array |
57 | 0 | auto materialized = |
58 | 0 | block->get_by_position(value_column_idx).column->convert_to_full_column_if_const(); |
59 | 0 | std::span<const uint8_t> outer_nulls; |
60 | 0 | const IColumn* nested = materialized.get(); |
61 | 0 | if (const auto* nullable = check_and_get_column<ColumnNullable>(nested)) { |
62 | 0 | outer_nulls = nullable->get_null_map_data(); |
63 | 0 | nested = &nullable->get_nested_column(); |
64 | 0 | } |
65 | 0 | if (const auto* variant_v2 = check_and_get_column<ColumnVariantV2>(nested)) { |
66 | 0 | auto variant_type = std::make_shared<DataTypeVariant>(); |
67 | 0 | auto target_type = std::make_shared<DataTypeArray>(variant_type); |
68 | 0 | ColumnPtr array_column; |
69 | 0 | RETURN_IF_ERROR(CastWrapper::variant_v2_internal::cast_variant_to_array( |
70 | 0 | nullptr, *variant_v2, target_type, variant_v2->size(), outer_nulls, &array_column)); |
71 | 0 | _array_columns[children_column_idx] = std::move(array_column); |
72 | 0 | auto& detail = _multi_detail[children_column_idx]; |
73 | 0 | detail.output_as_variant = true; |
74 | 0 | detail.nested_type = make_nullable(std::move(variant_type)); |
75 | 0 | _variant_v2_outputs[children_column_idx] = true; |
76 | 0 | _has_variant_v2_output = true; |
77 | 0 | return Status::OK(); |
78 | 0 | } |
79 | | |
80 | 0 | auto column = remove_nullable(materialized); |
81 | 0 | auto variant_column_ptr = IColumn::mutate(std::move(column)); |
82 | 0 | auto& variant_column = assert_cast<ColumnVariant&>(*variant_column_ptr); |
83 | 0 | variant_column.finalize(); |
84 | 0 | _multi_detail[children_column_idx].output_as_variant = true; |
85 | 0 | _multi_detail[children_column_idx].variant_enable_doc_mode = variant_column.enable_doc_mode(); |
86 | 0 | if (!variant_column.is_null_root()) { |
87 | 0 | _array_columns[children_column_idx] = variant_column.get_root(); |
88 | | // We need to wrap the output nested column within a variant column. |
89 | | // Otherwise the type is missmatched |
90 | 0 | const auto* array_type = check_and_get_data_type<DataTypeArray>( |
91 | 0 | remove_nullable(variant_column.get_root_type()).get()); |
92 | 0 | if (array_type == nullptr) { |
93 | 0 | return Status::NotSupported("explode not support none array type {}", |
94 | 0 | variant_column.get_root_type()->get_name()); |
95 | 0 | } |
96 | 0 | _multi_detail[children_column_idx].nested_type = array_type->get_nested_type(); |
97 | 0 | } else { |
98 | | // null root, use nothing type |
99 | 0 | auto array_column = ColumnNullable::create(ColumnArray::create(ColumnNothing::create(0)), |
100 | 0 | ColumnUInt8::create(0)); |
101 | 0 | array_column->insert_many_defaults(variant_column.size()); |
102 | 0 | _array_columns[children_column_idx] = std::move(array_column); |
103 | 0 | _multi_detail[children_column_idx].nested_type = std::make_shared<DataTypeNothing>(); |
104 | 0 | } |
105 | 0 | return Status::OK(); |
106 | 0 | } |
107 | | |
108 | 20 | Status VExplodeV2TableFunction::process_init(Block* block, RuntimeState* state) { |
109 | 20 | auto expr_size = _expr_context->root()->children().size(); |
110 | 20 | CHECK(expr_size >= 1) << "VExplodeV2TableFunction support one or more child but has " |
111 | 0 | << expr_size; |
112 | | |
113 | 20 | int value_column_idx = -1; |
114 | 20 | _multi_detail.resize(expr_size); |
115 | 20 | _array_offsets.resize(expr_size); |
116 | 20 | _array_columns.resize(expr_size); |
117 | 20 | _variant_v2_outputs.assign(expr_size, false); |
118 | | |
119 | 46 | for (int i = 0; i < expr_size; i++) { |
120 | 26 | RETURN_IF_ERROR(_expr_context->root()->children()[i]->execute(_expr_context.get(), block, |
121 | 26 | &value_column_idx)); |
122 | 26 | if (block->get_by_position(value_column_idx).type->get_primitive_type() == TYPE_VARIANT) { |
123 | 0 | RETURN_IF_ERROR(_process_init_variant(block, value_column_idx, i)); |
124 | 26 | } else { |
125 | 26 | _array_columns[i] = block->get_by_position(value_column_idx) |
126 | 26 | .column->convert_to_full_column_if_const(); |
127 | 26 | } |
128 | 26 | if (!extract_column_array_info(*_array_columns[i], _multi_detail[i])) { |
129 | 0 | return Status::NotSupported( |
130 | 0 | "column type {} not supported now", |
131 | 0 | block->get_by_position(value_column_idx).column->get_name()); |
132 | 0 | } |
133 | 26 | if (check_and_get_column<ColumnVariantV2>(_multi_detail[i].nested_col.get()) != nullptr) { |
134 | 0 | _variant_v2_outputs[i] = true; |
135 | 0 | _has_variant_v2_output = true; |
136 | 0 | } |
137 | 26 | } |
138 | | |
139 | 20 | return Status::OK(); |
140 | 20 | } |
141 | | |
142 | 17 | bool VExplodeV2TableFunction::support_block_fast_path() const { |
143 | 17 | return _multi_detail.size() == 1 && !_has_variant_v2_output; |
144 | 17 | } |
145 | | |
146 | | Status VExplodeV2TableFunction::prepare_block_fast_path(Block* /*block*/, RuntimeState* /*state*/, |
147 | 8 | BlockFastPathContext* ctx) { |
148 | 8 | DCHECK(support_block_fast_path()); |
149 | 8 | const auto& detail = _multi_detail[0]; |
150 | 8 | if (detail.offsets_ptr == nullptr || detail.nested_col.get() == nullptr) { |
151 | 0 | return Status::InternalError("vexplode block fast path not initialized"); |
152 | 0 | } |
153 | 8 | ctx->array_nullmap_data = detail.array_nullmap_data; |
154 | 8 | ctx->offsets_ptr = detail.offsets_ptr; |
155 | 8 | ctx->nested_col = detail.nested_col; |
156 | 8 | ctx->nested_nullmap_data = detail.nested_nullmap_data; |
157 | 8 | ctx->generate_row_index = _generate_row_index; |
158 | 8 | return Status::OK(); |
159 | 8 | } |
160 | | |
161 | 31 | void VExplodeV2TableFunction::process_row(size_t row_idx) { |
162 | 31 | TableFunction::process_row(row_idx); |
163 | | |
164 | 68 | for (int i = 0; i < _multi_detail.size(); i++) { |
165 | 37 | auto& detail = _multi_detail[i]; |
166 | 37 | if (!detail.array_nullmap_data || !detail.array_nullmap_data[row_idx]) { |
167 | 30 | _array_offsets[i] = (*detail.offsets_ptr)[row_idx - 1]; |
168 | | // find max size in array |
169 | 30 | auto cur_size = (*detail.offsets_ptr)[row_idx] - _array_offsets[i]; |
170 | 30 | _cur_size = std::max<unsigned long>(_cur_size, cur_size); |
171 | 30 | } |
172 | 37 | } |
173 | 31 | _row_idx = row_idx; |
174 | 31 | } |
175 | | |
176 | 9 | void VExplodeV2TableFunction::process_close() { |
177 | 9 | _multi_detail.clear(); |
178 | 9 | _array_offsets.clear(); |
179 | 9 | _array_columns.clear(); |
180 | 9 | _variant_v2_outputs.clear(); |
181 | 9 | _row_idx = 0; |
182 | 9 | _has_variant_v2_output = false; |
183 | 9 | } |
184 | | |
185 | 36 | void VExplodeV2TableFunction::_ensure_variant_v2_output(MutableColumnPtr& column) const { |
186 | 36 | if (!_has_variant_v2_output) { |
187 | 36 | return; |
188 | 36 | } |
189 | 0 | const bool struct_output = _multi_detail.size() > 1 || _generate_row_index; |
190 | 0 | if (!struct_output) { |
191 | 0 | auto* nullable = check_and_get_column<ColumnNullable>(column.get()); |
192 | 0 | DORIS_CHECK(nullable != nullptr); |
193 | 0 | if (check_and_get_column<ColumnVariantV2>(&nullable->get_nested_column()) != nullptr) { |
194 | 0 | return; |
195 | 0 | } |
196 | 0 | DORIS_CHECK_EQ(column->size(), 0); |
197 | 0 | column = ColumnNullable::create(ColumnVariantV2::create(), ColumnUInt8::create()); |
198 | 0 | return; |
199 | 0 | } |
200 | | |
201 | 0 | IColumn* nested_output = column.get(); |
202 | 0 | if (auto* nullable = check_and_get_column<ColumnNullable>(nested_output)) { |
203 | 0 | nested_output = nullable->get_nested_column_ptr().get(); |
204 | 0 | } |
205 | 0 | auto* struct_column = check_and_get_column<ColumnStruct>(nested_output); |
206 | 0 | DORIS_CHECK(struct_column != nullptr); |
207 | 0 | for (size_t i = 0; i < _variant_v2_outputs.size(); ++i) { |
208 | 0 | if (!_variant_v2_outputs[i]) { |
209 | 0 | continue; |
210 | 0 | } |
211 | 0 | const size_t field_index = i + (_generate_row_index ? 1 : 0); |
212 | 0 | ColumnPtr& field = struct_column->get_column_ptr(field_index); |
213 | 0 | const auto* nullable = check_and_get_column<ColumnNullable>(field.get()); |
214 | 0 | DORIS_CHECK(nullable != nullptr); |
215 | 0 | if (check_and_get_column<ColumnVariantV2>(&nullable->get_nested_column()) != nullptr) { |
216 | 0 | continue; |
217 | 0 | } |
218 | 0 | DORIS_CHECK(field->empty()); |
219 | 0 | field = ColumnNullable::create(ColumnVariantV2::create(), ColumnUInt8::create()); |
220 | 0 | } |
221 | 0 | } |
222 | | |
223 | 31 | void VExplodeV2TableFunction::get_same_many_values(MutableColumnPtr& column, int length) { |
224 | 31 | _ensure_variant_v2_output(column); |
225 | 31 | if (current_empty()) { |
226 | 6 | column->insert_many_defaults(length); |
227 | 6 | return; |
228 | 6 | } |
229 | 25 | ColumnStruct* struct_column = nullptr; |
230 | 25 | std::vector<IColumn*> columns; |
231 | | |
232 | 25 | const bool multi_sub_columns = _multi_detail.size() > 1 || _generate_row_index; |
233 | | |
234 | 25 | if (multi_sub_columns) { |
235 | 9 | if (_is_nullable) { |
236 | 9 | auto* nullable_column = assert_cast<ColumnNullable*>(column.get()); |
237 | 9 | struct_column = |
238 | 9 | assert_cast<ColumnStruct*>(nullable_column->get_nested_column_ptr().get()); |
239 | 9 | auto* nullmap_column = nullable_column->get_null_map_column_ptr().get(); |
240 | 9 | nullmap_column->insert_many_defaults(length); |
241 | | |
242 | 9 | } else { |
243 | 0 | struct_column = assert_cast<ColumnStruct*>(column.get()); |
244 | 0 | } |
245 | | |
246 | 27 | for (size_t i = 0; i != _multi_detail.size(); ++i) { |
247 | 18 | columns.emplace_back(&struct_column->get_column(i + (_generate_row_index ? 1 : 0))); |
248 | 18 | } |
249 | 16 | } else { |
250 | 16 | columns.push_back(column.get()); |
251 | 16 | } |
252 | | |
253 | 25 | if (_generate_row_index) { |
254 | 0 | auto& pos_column = assert_cast<ColumnInt32&>(struct_column->get_column(0)); |
255 | 0 | pos_column.insert_many_vals(static_cast<int32_t>(_cur_offset), length); |
256 | 0 | } |
257 | | |
258 | 59 | for (int i = 0; i < _multi_detail.size(); i++) { |
259 | 34 | auto& detail = _multi_detail[i]; |
260 | 34 | size_t pos = _array_offsets[i] + _cur_offset; |
261 | 34 | size_t element_size = _multi_detail[i].array_col->size_at(_row_idx); |
262 | 34 | auto& struct_field = *columns.at(i); |
263 | 34 | if ((detail.array_nullmap_data && detail.array_nullmap_data[_row_idx])) { |
264 | 3 | struct_field.insert_many_defaults(length); |
265 | 31 | } else { |
266 | 31 | auto* nullable_column = assert_cast<ColumnNullable*>(struct_field.get_ptr().get()); |
267 | 31 | auto* nullmap_column = nullable_column->get_null_map_column_ptr().get(); |
268 | | // only need to check if the value at position pos is null |
269 | 31 | if (element_size < _cur_offset || |
270 | 31 | (detail.nested_nullmap_data && detail.nested_nullmap_data[pos])) { |
271 | 3 | nullable_column->insert_many_defaults(length); |
272 | 28 | } else { |
273 | 28 | nullable_column->get_nested_column_ptr()->insert_many_from(*detail.nested_col, pos, |
274 | 28 | length); |
275 | 28 | nullmap_column->insert_many_defaults(length); |
276 | 28 | } |
277 | 31 | } |
278 | 34 | } |
279 | 25 | } |
280 | | |
281 | 5 | int VExplodeV2TableFunction::get_value(MutableColumnPtr& column, int max_step) { |
282 | 5 | _ensure_variant_v2_output(column); |
283 | 5 | max_step = std::min(max_step, (int)(_cur_size - _cur_offset)); |
284 | 5 | const bool multi_sub_columns = _multi_detail.size() > 1 || _generate_row_index; |
285 | | |
286 | 5 | ColumnStruct* struct_column = nullptr; |
287 | 5 | std::vector<IColumn*> columns; |
288 | | |
289 | 5 | if (current_empty()) { |
290 | 0 | column->insert_default(); |
291 | 0 | max_step = 1; |
292 | 5 | } else { |
293 | 5 | if (multi_sub_columns) { |
294 | 3 | if (_is_nullable) { |
295 | 3 | auto* nullable_column = assert_cast<ColumnNullable*>(column.get()); |
296 | 3 | struct_column = |
297 | 3 | assert_cast<ColumnStruct*>(nullable_column->get_nested_column_ptr().get()); |
298 | 3 | auto* nullmap_column = nullable_column->get_null_map_column_ptr().get(); |
299 | 3 | nullmap_column->insert_many_defaults(max_step); |
300 | | |
301 | 3 | } else { |
302 | 0 | struct_column = assert_cast<ColumnStruct*>(column.get()); |
303 | 0 | } |
304 | | |
305 | 9 | for (size_t i = 0; i != _multi_detail.size(); ++i) { |
306 | 6 | columns.emplace_back(&struct_column->get_column(i + (_generate_row_index ? 1 : 0))); |
307 | 6 | } |
308 | 3 | } else { |
309 | 2 | columns.emplace_back(column.get()); |
310 | 2 | } |
311 | | |
312 | 5 | if (_generate_row_index) { |
313 | 0 | auto& pos_column = assert_cast<ColumnInt32&>(struct_column->get_column(0)); |
314 | 0 | pos_column.insert_range_of_integer(static_cast<int32_t>(_cur_offset), |
315 | 0 | static_cast<int32_t>(_cur_offset + max_step)); |
316 | 0 | } |
317 | | |
318 | 13 | for (int i = 0; i < _multi_detail.size(); i++) { |
319 | 8 | auto& detail = _multi_detail[i]; |
320 | 8 | size_t pos = _array_offsets[i] + _cur_offset; |
321 | 8 | size_t element_size = _multi_detail[i].array_col->size_at(_row_idx); |
322 | 8 | auto& struct_field = *columns.at(i); |
323 | 8 | if (detail.array_nullmap_data && detail.array_nullmap_data[_row_idx]) { |
324 | 1 | struct_field.insert_many_defaults(max_step); |
325 | 7 | } else { |
326 | 7 | auto* nullable_column = assert_cast<ColumnNullable*>(struct_field.get_ptr().get()); |
327 | 7 | auto* nullmap_column = nullable_column->get_null_map_column_ptr().get(); |
328 | 7 | if (element_size >= _cur_offset + max_step) { |
329 | 7 | nullable_column->get_nested_column_ptr()->insert_range_from(*detail.nested_col, |
330 | 7 | pos, max_step); |
331 | 7 | if (detail.nested_nullmap_data) { |
332 | 7 | size_t old_size = nullmap_column->size(); |
333 | 7 | nullmap_column->resize(old_size + max_step); |
334 | 7 | memcpy(nullmap_column->get_data().data() + old_size, |
335 | 7 | detail.nested_nullmap_data + pos, max_step * sizeof(UInt8)); |
336 | 7 | } else { |
337 | 0 | nullmap_column->insert_many_defaults(max_step); |
338 | 0 | } |
339 | 7 | } else if (element_size > _cur_offset) { |
340 | 0 | auto current_insert_num = element_size - _cur_offset; |
341 | 0 | nullable_column->get_nested_column_ptr()->insert_range_from( |
342 | 0 | *detail.nested_col, pos, current_insert_num); |
343 | 0 | if (detail.nested_nullmap_data) { |
344 | 0 | size_t old_size = nullmap_column->size(); |
345 | 0 | nullmap_column->resize(old_size + current_insert_num); |
346 | 0 | memcpy(nullmap_column->get_data().data() + old_size, |
347 | 0 | detail.nested_nullmap_data + pos, |
348 | 0 | current_insert_num * sizeof(UInt8)); |
349 | 0 | } else { |
350 | 0 | nullmap_column->insert_many_defaults(current_insert_num); |
351 | 0 | } |
352 | 0 | nullable_column->insert_many_defaults(max_step - current_insert_num); |
353 | 0 | } else { |
354 | 0 | nullable_column->insert_many_defaults(max_step); |
355 | 0 | } |
356 | 7 | } |
357 | 8 | } |
358 | 5 | } |
359 | | |
360 | 5 | forward(max_step); |
361 | 5 | return max_step; |
362 | 5 | } |
363 | | |
364 | | } // namespace doris |