be/src/exprs/function/function_variant_element.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 <glog/logging.h> |
19 | | #include <stddef.h> |
20 | | |
21 | | #include <memory> |
22 | | #include <ostream> |
23 | | #include <span> |
24 | | #include <string> |
25 | | #include <string_view> |
26 | | #include <utility> |
27 | | #include <vector> |
28 | | |
29 | | #include "common/status.h" |
30 | | #include "core/assert_cast.h" |
31 | | #include "core/block/block.h" |
32 | | #include "core/column/column.h" |
33 | | #include "core/column/column_nullable.h" |
34 | | #include "core/column/column_string.h" |
35 | | #include "core/column/column_variant.h" |
36 | | #include "core/column/subcolumn_tree.h" |
37 | | #include "core/column/variant_v2/column_variant_v2.h" |
38 | | #include "core/data_type/data_type.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_number.h" |
42 | | #include "core/data_type/data_type_string.h" |
43 | | #include "core/data_type/data_type_variant.h" |
44 | | #include "core/string_ref.h" |
45 | | #include "exprs/function/function.h" |
46 | | #include "exprs/function/function_helpers.h" |
47 | | #include "exprs/function/function_variant_element_v2.h" |
48 | | #include "exprs/function/simple_function_factory.h" |
49 | | #include "exprs/json_functions.h" |
50 | | #include "simdjson.h" |
51 | | #include "util/defer_op.h" |
52 | | #include "util/json/path_in_data.h" |
53 | | |
54 | | namespace doris { |
55 | | |
56 | | class FunctionVariantElement : public IFunction { |
57 | | public: |
58 | | static constexpr auto name = "element_at"; |
59 | 137 | static FunctionPtr create() { return std::make_shared<FunctionVariantElement>(); } |
60 | | |
61 | | // Get function name. |
62 | 2 | String get_name() const override { return name; } |
63 | | |
64 | 258 | bool use_default_implementation_for_nulls() const override { return false; } |
65 | | |
66 | 129 | size_t get_number_of_arguments() const override { return 2; } |
67 | | |
68 | 129 | ColumnNumbers get_arguments_that_are_always_constant() const override { return {1}; } |
69 | | |
70 | 7 | DataTypes get_variadic_argument_types_impl() const override { |
71 | 7 | return {std::make_shared<DataTypeVariant>(), std::make_shared<DataTypeString>()}; |
72 | 7 | } |
73 | | |
74 | 129 | DataTypePtr get_return_type_impl(const DataTypes& arguments) const override { |
75 | 129 | DCHECK_EQ(arguments[0]->get_primitive_type(), TYPE_VARIANT) |
76 | 0 | << "First argument for function: " << name |
77 | 0 | << " should be DataTypeVariant but it has type " << arguments[0]->get_name() << "."; |
78 | 129 | const PrimitiveType index_type = remove_nullable(arguments[1])->get_primitive_type(); |
79 | 129 | DCHECK(is_string_type(index_type) || is_int_or_bool(index_type)) |
80 | 0 | << "Second argument for function: " << name |
81 | 0 | << " should be String or Integer but it has type " << arguments[1]->get_name() |
82 | 0 | << "."; |
83 | 129 | auto arg_variant = remove_nullable(arguments[0]); |
84 | 129 | return make_nullable(std::move(arg_variant)); |
85 | 129 | } |
86 | | |
87 | | // wrap variant column with nullable |
88 | | // 1. if variant is null root(empty or nothing as root), then nullable map is all null |
89 | | // 2. if variant is scalar variant, then use the root's nullable map |
90 | | // 3. if variant is hierarchical variant, then create a nullable map with all none null |
91 | 129 | ColumnPtr wrap_variant_nullable(ColumnPtr col) const { |
92 | 129 | const auto& var = assert_cast<const ColumnVariant&>(*col); |
93 | 129 | if (var.is_null_root()) { |
94 | 20 | return make_nullable(col, true); |
95 | 20 | } |
96 | 109 | if (var.is_scalar_variant() && is_column_nullable(*var.get_root())) { |
97 | 29 | const auto* nullable = assert_cast<const ColumnNullable*>(var.get_root().get()); |
98 | 29 | return ColumnNullable::create(col, nullable->get_null_map_column_ptr()); |
99 | 29 | } |
100 | 80 | return make_nullable(col); |
101 | 109 | } |
102 | | |
103 | | // Keep legacy/V2 physical-column dispatch in one entry point so nullable handling stays shared. |
104 | | // NOLINTNEXTLINE(readability-function-size) |
105 | | Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments, |
106 | 129 | uint32_t result, size_t input_rows_count) const override { |
107 | 129 | const ColumnPtr materialized = |
108 | 129 | block.get_by_position(arguments[0]).column->convert_to_full_column_if_const(); |
109 | 129 | const IColumn* physical = materialized.get(); |
110 | 129 | std::span<const uint8_t> outer_nulls; |
111 | 129 | if (const auto* nullable = check_and_get_column<ColumnNullable>(physical)) { |
112 | 129 | outer_nulls = nullable->get_null_map_data(); |
113 | 129 | physical = &nullable->get_nested_column(); |
114 | 129 | } |
115 | 129 | if (const auto* variant_v2 = check_and_get_column<ColumnVariantV2>(physical)) { |
116 | 0 | if (block.empty()) { |
117 | 0 | block.replace_by_position(result, ColumnNullable::create(ColumnVariantV2::create(), |
118 | 0 | ColumnUInt8::create())); |
119 | 0 | return Status::OK(); |
120 | 0 | } |
121 | | |
122 | 0 | auto replace_with_all_null_result = [&]() { |
123 | 0 | auto null_values = ColumnVariantV2::create(); |
124 | 0 | null_values->insert_many_defaults(variant_v2->size()); |
125 | 0 | block.replace_by_position( |
126 | 0 | result, ColumnNullable::create(std::move(null_values), |
127 | 0 | ColumnUInt8::create(variant_v2->size(), 1))); |
128 | 0 | }; |
129 | 0 | const auto& index_argument = block.get_by_position(arguments[1]); |
130 | 0 | const ColumnPtr materialized_index = |
131 | 0 | index_argument.column->convert_to_full_column_if_const(); |
132 | 0 | const IColumn* index_column = materialized_index.get(); |
133 | 0 | if (index_column->is_null_at(0)) { |
134 | 0 | replace_with_all_null_result(); |
135 | 0 | return Status::OK(); |
136 | 0 | } |
137 | 0 | if (const auto* nullable = check_and_get_column<ColumnNullable>(*index_column)) { |
138 | 0 | index_column = &nullable->get_nested_column(); |
139 | 0 | } |
140 | |
|
141 | 0 | std::optional<VariantElementV2PathSegment> segment; |
142 | 0 | const PrimitiveType index_type = |
143 | 0 | remove_nullable(index_argument.type)->get_primitive_type(); |
144 | 0 | if (is_string_type(index_type)) { |
145 | 0 | segment = VariantElementV2PathSegment::object_key(index_column->get_data_at(0)); |
146 | 0 | } else if (is_int_or_bool(index_type)) { |
147 | 0 | const int64_t sql_index = index_column->get_int(0); |
148 | 0 | if (sql_index == 0) { |
149 | 0 | replace_with_all_null_result(); |
150 | 0 | return Status::OK(); |
151 | 0 | } |
152 | 0 | segment = VariantElementV2PathSegment::array_index(sql_index > 0 ? sql_index - 1 |
153 | 0 | : sql_index); |
154 | 0 | } else { |
155 | 0 | return Status::RuntimeError("unsupported index type {} for function {}", |
156 | 0 | index_argument.type->get_name(), get_name()); |
157 | 0 | } |
158 | 0 | std::unique_ptr<ResolvedVariantElementV2Path> path; |
159 | 0 | RETURN_IF_ERROR(resolve_variant_element_v2_path(std::span(&*segment, 1), &path)); |
160 | 0 | ColumnPtr result_column; |
161 | 0 | RETURN_IF_ERROR( |
162 | 0 | extract_variant_element_v2(*variant_v2, *path, outer_nulls, &result_column)); |
163 | 0 | block.replace_by_position(result, std::move(result_column)); |
164 | 0 | return Status::OK(); |
165 | 0 | } |
166 | | |
167 | 129 | const auto* variant_col = |
168 | 129 | check_and_get_column<ColumnVariant>(remove_nullable(materialized).get()); |
169 | 129 | if (!variant_col) { |
170 | 0 | return Status::RuntimeError( |
171 | 0 | fmt::format("unsupported types for function {}({}, {})", get_name(), |
172 | 0 | block.get_by_position(arguments[0]).type->get_name(), |
173 | 0 | block.get_by_position(arguments[1]).type->get_name())); |
174 | 0 | } |
175 | 129 | if (block.empty()) { |
176 | 0 | block.replace_by_position(result, block.get_by_position(result).type->create_column()); |
177 | 0 | return Status::OK(); |
178 | 0 | } |
179 | | |
180 | 129 | auto index_column = block.get_by_position(arguments[1]).column; |
181 | 129 | ColumnPtr result_column; |
182 | 129 | RETURN_IF_ERROR(get_element_column(*variant_col, index_column, &result_column)); |
183 | 129 | if (block.get_by_position(result).type->is_nullable()) { |
184 | 129 | result_column = wrap_variant_nullable(result_column); |
185 | 129 | } |
186 | 129 | block.replace_by_position(result, result_column); |
187 | 129 | return Status::OK(); |
188 | 129 | } |
189 | | |
190 | | private: |
191 | | // Return sub-path by specified prefix. |
192 | | // For example, for prefix a.b: |
193 | | // a.b.c.d -> c.d, a.b.c -> c |
194 | | static std::optional<std::string_view> get_sub_path(const std::string_view& path, |
195 | 4 | const std::string_view& prefix) { |
196 | 4 | if (path.size() <= prefix.size() || path[prefix.size()] != '.') { |
197 | 1 | return std::nullopt; |
198 | 1 | } |
199 | 3 | return path.substr(prefix.size() + 1); |
200 | 4 | } |
201 | | |
202 | | // Extract and populate sparse column data with given path prefix |
203 | | // Copies data from source sparse column, extracting only the sub-paths that match the prefix |
204 | | static void _extract_sparse_column_from_source(ColumnVariant* src_ptr, const PathInData& path, |
205 | 1 | ColumnVariant::MutablePtr& target_ptr) { |
206 | 1 | ColumnVariant::Subcolumn root {0, true, true}; |
207 | | // no root, no sparse column |
208 | 1 | const auto& sparse_data_map = assert_cast<const ColumnMap&>(*src_ptr->get_sparse_column()); |
209 | 1 | const auto& src_sparse_data_offsets = sparse_data_map.get_offsets(); |
210 | 1 | const auto& src_sparse_data_paths = |
211 | 1 | assert_cast<const ColumnString&>(sparse_data_map.get_keys()); |
212 | 1 | const auto& src_sparse_data_values = |
213 | 1 | assert_cast<const ColumnString&>(sparse_data_map.get_values()); |
214 | 1 | auto& sparse_data_offsets = |
215 | 1 | assert_cast<ColumnMap&>(target_ptr->get_sparse_column_mutable()).get_offsets(); |
216 | 1 | auto [sparse_data_paths, sparse_data_values] = |
217 | 1 | target_ptr->get_sparse_data_paths_and_values(); |
218 | 1 | StringRef prefix_ref(path.get_path()); |
219 | 1 | std::string_view path_prefix(prefix_ref.data, prefix_ref.size); |
220 | 2 | for (size_t i = 0; i != src_sparse_data_offsets.size(); ++i) { |
221 | 1 | size_t start = src_sparse_data_offsets[ssize_t(i) - 1]; |
222 | 1 | size_t end = src_sparse_data_offsets[ssize_t(i)]; |
223 | 1 | size_t lower_bound_index = ColumnVariant::find_path_lower_bound_in_sparse_data( |
224 | 1 | prefix_ref, src_sparse_data_paths, start, end); |
225 | 4 | for (; lower_bound_index != end; ++lower_bound_index) { |
226 | 3 | auto path_ref = src_sparse_data_paths.get_data_at(lower_bound_index); |
227 | 3 | std::string_view nested_path(path_ref.data, path_ref.size); |
228 | 3 | if (!nested_path.starts_with(path_prefix)) { |
229 | 0 | break; |
230 | 0 | } |
231 | | // Don't include path that is equal to the prefix. |
232 | 3 | if (nested_path.size() != path_prefix.size()) { |
233 | 3 | auto sub_path_optional = get_sub_path(nested_path, path_prefix); |
234 | 3 | if (!sub_path_optional.has_value()) { |
235 | 1 | continue; |
236 | 1 | } |
237 | 2 | std::string_view sub_path = *sub_path_optional; |
238 | 2 | sparse_data_paths->insert_data(sub_path.data(), sub_path.size()); |
239 | 2 | sparse_data_values->insert_from(src_sparse_data_values, lower_bound_index); |
240 | 2 | } else { |
241 | | // insert into root column, example: access v['b'] and b is in sparse column |
242 | | // data example: |
243 | | // {"b" : 123} |
244 | | // {"b" : {"c" : 456}} |
245 | | // b maybe in sparse column, and b.c is in subolumn, put `b` into root column to distinguish |
246 | | // from "" which is empty path and root |
247 | 0 | root.deserialize_from_binary_column(&src_sparse_data_values, lower_bound_index); |
248 | 0 | } |
249 | 3 | } |
250 | 1 | if (root.size() == sparse_data_offsets.size()) { |
251 | 1 | root.insert_default(); |
252 | 1 | } |
253 | 1 | sparse_data_offsets.push_back(sparse_data_paths->size()); |
254 | 1 | } |
255 | 1 | target_ptr->get_subcolumns().create_root(root); |
256 | 1 | target_ptr->get_doc_value_column_mutable().resize(src_ptr->size()); |
257 | 1 | target_ptr->set_num_rows(src_ptr->size()); |
258 | 1 | } |
259 | | |
260 | | // Extract and populate sparse column data from doc_value column with given path prefix |
261 | | // Copies data from source doc_value column, extracting only the sub-paths that match the prefix |
262 | | static void _extract_doc_value_column_from_source(ColumnVariant* src_ptr, |
263 | | const PathInData& path, |
264 | 49 | ColumnVariant::MutablePtr& target_ptr) { |
265 | 49 | ColumnVariant::Subcolumn root {0, true, true}; |
266 | 49 | const auto& doc_value_data_map = |
267 | 49 | assert_cast<const ColumnMap&>(*src_ptr->get_doc_value_column()); |
268 | 49 | const auto& src_doc_value_data_offsets = doc_value_data_map.get_offsets(); |
269 | 49 | const auto& src_doc_value_data_paths = |
270 | 49 | assert_cast<const ColumnString&>(doc_value_data_map.get_keys()); |
271 | 49 | const auto& src_doc_value_data_values = |
272 | 49 | assert_cast<const ColumnString&>(doc_value_data_map.get_values()); |
273 | 49 | const bool write_to_doc_value = target_ptr->enable_doc_mode(); |
274 | | // Ordinary Variant extraction keeps the selected prefix in sparse data, matching the |
275 | | // source branch behavior. Only doc-mode columns keep extracted data in doc_value. |
276 | 49 | auto& extracted_offsets = |
277 | 49 | assert_cast<ColumnMap&>(write_to_doc_value |
278 | 49 | ? target_ptr->get_doc_value_column_mutable() |
279 | 49 | : target_ptr->get_sparse_column_mutable()) |
280 | 49 | .get_offsets(); |
281 | 49 | auto [extracted_paths, extracted_values] = |
282 | 49 | write_to_doc_value ? target_ptr->get_doc_value_data_paths_and_values() |
283 | 49 | : target_ptr->get_sparse_data_paths_and_values(); |
284 | 49 | StringRef prefix_ref(path.get_path()); |
285 | 49 | std::string_view path_prefix(prefix_ref.data, prefix_ref.size); |
286 | 98 | for (size_t i = 0; i != src_doc_value_data_offsets.size(); ++i) { |
287 | 49 | size_t start = src_doc_value_data_offsets[ssize_t(i) - 1]; |
288 | 49 | size_t end = src_doc_value_data_offsets[ssize_t(i)]; |
289 | 49 | size_t lower_bound_index = ColumnVariant::find_path_lower_bound_in_sparse_data( |
290 | 49 | prefix_ref, src_doc_value_data_paths, start, end); |
291 | 79 | for (; lower_bound_index != end; ++lower_bound_index) { |
292 | 61 | auto path_ref = src_doc_value_data_paths.get_data_at(lower_bound_index); |
293 | 61 | std::string_view nested_path(path_ref.data, path_ref.size); |
294 | 61 | if (!nested_path.starts_with(path_prefix)) { |
295 | 31 | break; |
296 | 31 | } |
297 | 30 | if (nested_path.size() != path_prefix.size()) { |
298 | 1 | auto sub_path_optional = get_sub_path(nested_path, path_prefix); |
299 | 1 | if (!sub_path_optional.has_value()) { |
300 | 0 | continue; |
301 | 0 | } |
302 | 1 | std::string_view sub_path = *sub_path_optional; |
303 | 1 | extracted_paths->insert_data(sub_path.data(), sub_path.size()); |
304 | 1 | extracted_values->insert_from(src_doc_value_data_values, lower_bound_index); |
305 | 29 | } else { |
306 | 29 | root.deserialize_from_binary_column(&src_doc_value_data_values, |
307 | 29 | lower_bound_index); |
308 | 29 | } |
309 | 30 | } |
310 | 49 | if (root.size() == extracted_offsets.size()) { |
311 | 20 | root.insert_default(); |
312 | 20 | } |
313 | 49 | extracted_offsets.push_back(extracted_paths->size()); |
314 | 49 | } |
315 | 49 | target_ptr->get_subcolumns().create_root(root); |
316 | 49 | if (write_to_doc_value) { |
317 | 49 | target_ptr->get_sparse_column_mutable().resize(src_ptr->size()); |
318 | 49 | } else { |
319 | 0 | target_ptr->get_doc_value_column_mutable().resize(src_ptr->size()); |
320 | 0 | } |
321 | 49 | target_ptr->set_num_rows(src_ptr->size()); |
322 | 49 | } |
323 | | |
324 | | static Status get_element_column(const ColumnVariant& src, const ColumnPtr& index_column, |
325 | 133 | ColumnPtr* result) { |
326 | 133 | std::string field_name = index_column->get_data_at(0).to_string(); |
327 | 133 | if (src.empty()) { |
328 | 0 | *result = ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode()); |
329 | | // src subcolumns empty but src row count may not be 0 |
330 | 0 | (*result)->assert_mutable()->insert_many_defaults(src.size()); |
331 | | // ColumnVariant should be finalized before parsing, finalize maybe modify original column structure |
332 | 0 | (*result)->assert_mutable()->finalize(); |
333 | 0 | return Status::OK(); |
334 | 0 | } |
335 | 133 | if (src.is_scalar_variant() && is_string_type(src.get_root_type()->get_primitive_type())) { |
336 | | // use parser to extract from root |
337 | 83 | auto type = std::make_shared<DataTypeString>(); |
338 | 83 | MutableColumnPtr result_column = type->create_column(); |
339 | 83 | const ColumnString& docs = |
340 | 83 | *assert_cast<const ColumnString*>(remove_nullable(src.get_root()).get()); |
341 | 83 | simdjson::ondemand::parser parser; |
342 | 83 | std::vector<JsonPath> parsed_paths; |
343 | 83 | if (field_name.empty() || field_name[0] != '$') { |
344 | 83 | field_name = "$." + field_name; |
345 | 83 | } |
346 | 83 | JsonFunctions::parse_json_paths(field_name, &parsed_paths); |
347 | 83 | ColumnString* col_str = static_cast<ColumnString*>(result_column.get()); |
348 | 167 | for (size_t i = 0; i < docs.size(); ++i) { |
349 | 84 | if (!extract_from_document(parser, docs.get_data_at(i), parsed_paths, col_str)) { |
350 | 15 | VLOG_DEBUG << "failed to parse " << docs.get_data_at(i) << ", field " |
351 | 0 | << field_name; |
352 | 15 | result_column->insert_default(); |
353 | 15 | } |
354 | 84 | } |
355 | 83 | *result = ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(), type, |
356 | 83 | std::move(result_column)); |
357 | 83 | (*result)->assert_mutable()->finalize(); |
358 | 83 | return Status::OK(); |
359 | 83 | } else { |
360 | 50 | auto mutable_src = src.clone_finalized(); |
361 | 50 | auto* mutable_ptr = assert_cast<ColumnVariant*>(mutable_src.get()); |
362 | 50 | PathInData path(field_name); |
363 | 50 | ColumnVariant::Subcolumns subcolumns = mutable_ptr->get_subcolumns(); |
364 | 50 | const auto* node = subcolumns.find_exact(path); |
365 | 50 | MutableColumnPtr result_col = |
366 | 50 | ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode()); |
367 | 50 | ColumnVariant::Subcolumns new_subcolumns; |
368 | | |
369 | 50 | if (node != nullptr) { |
370 | 0 | std::vector<decltype(node)> nodes; |
371 | 0 | PathsInData paths; |
372 | 0 | ColumnVariant::Subcolumns::get_leaves_of_node(node, nodes, paths); |
373 | 0 | for (const auto* n : nodes) { |
374 | 0 | PathInData new_path = n->path.copy_pop_front(); |
375 | 0 | VLOG_DEBUG << "add node " << new_path.get_path() |
376 | 0 | << ", data size: " << n->data.size() |
377 | 0 | << ", finalized size: " << n->data.get_finalized_column().size() |
378 | 0 | << ", common type: " << n->data.get_least_common_type()->get_name(); |
379 | | // if new_path is empty, indicate it's the root column, but adding a root will return false when calling add |
380 | 0 | if (!new_subcolumns.add(new_path, n->data)) { |
381 | 0 | VLOG_DEBUG << "failed to add node " << new_path.get_path(); |
382 | 0 | } |
383 | 0 | } |
384 | | |
385 | | // handle the root node |
386 | 0 | if (new_subcolumns.empty() && !nodes.empty()) { |
387 | 0 | CHECK_EQ(nodes.size(), 1); |
388 | 0 | new_subcolumns.create_root(ColumnVariant::Subcolumn { |
389 | 0 | IColumn::mutate(nodes[0]->data.get_finalized_column_ptr()), |
390 | 0 | nodes[0]->data.get_least_common_type(), true, true}); |
391 | 0 | auto container = |
392 | 0 | ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(), |
393 | 0 | std::move(new_subcolumns)); |
394 | 0 | result_col->insert_range_from(*container, 0, container->size()); |
395 | 0 | } else { |
396 | 0 | auto container = |
397 | 0 | ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(), |
398 | 0 | std::move(new_subcolumns)); |
399 | 0 | container->clear_sparse_column(); |
400 | 0 | _extract_sparse_column_from_source(mutable_ptr, path, container); |
401 | 0 | result_col->insert_range_from(*container, 0, container->size()); |
402 | 0 | } |
403 | 50 | } else { |
404 | 50 | auto container = |
405 | 50 | ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(), |
406 | 50 | std::move(new_subcolumns)); |
407 | 50 | const auto& sparse_offsets = mutable_ptr->serialized_sparse_column_offsets(); |
408 | 50 | if (sparse_offsets.back() == sparse_offsets[-1]) { |
409 | 49 | _extract_doc_value_column_from_source(mutable_ptr, path, container); |
410 | 49 | } else { |
411 | 1 | _extract_sparse_column_from_source(mutable_ptr, path, container); |
412 | 1 | } |
413 | 50 | result_col->insert_range_from(*container, 0, container->size()); |
414 | 50 | } |
415 | | // ColumnVariant should be finalized before parsing, finalize maybe modify original column structure |
416 | 50 | result_col->finalize(); |
417 | 50 | VLOG_DEBUG << "dump new object " |
418 | 0 | << static_cast<const ColumnVariant*>(result_col.get())->debug_string() |
419 | 0 | << ", path " << path.get_path(); |
420 | 50 | *result = std::move(result_col); |
421 | 50 | return Status::OK(); |
422 | 50 | } |
423 | 133 | } |
424 | | |
425 | | static Status extract_from_document(simdjson::ondemand::parser& parser, const StringRef& doc, |
426 | 84 | const std::vector<JsonPath>& paths, ColumnString* column) { |
427 | 84 | try { |
428 | 84 | simdjson::padded_string json_str {doc.data, doc.size}; |
429 | 84 | simdjson::ondemand::document document = parser.iterate(json_str); |
430 | 84 | simdjson::ondemand::object object = document.get_object(); |
431 | 84 | simdjson::ondemand::value value; |
432 | 84 | RETURN_IF_ERROR(JsonFunctions::extract_from_object(object, paths, &value)); |
433 | 74 | _write_data_to_column(value, column); |
434 | 74 | } catch (simdjson::simdjson_error& e) { |
435 | 5 | VLOG_DEBUG << "simdjson parse exception: " << e.what(); |
436 | 5 | return Status::DataQualityError("simdjson parse exception {}", e.what()); |
437 | 5 | } |
438 | 69 | return Status::OK(); |
439 | 84 | } |
440 | | |
441 | 69 | static void _write_data_to_column(simdjson::ondemand::value& value, ColumnString* column) { |
442 | 69 | switch (value.type()) { |
443 | 0 | case simdjson::ondemand::json_type::null: { |
444 | 0 | column->insert_default(); |
445 | 0 | break; |
446 | 0 | } |
447 | 0 | case simdjson::ondemand::json_type::boolean: { |
448 | 0 | if (value.get_bool()) { |
449 | 0 | column->insert_data("1", 1); |
450 | 0 | } else { |
451 | 0 | column->insert_data("0", 1); |
452 | 0 | } |
453 | 0 | break; |
454 | 0 | } |
455 | 10 | case simdjson::ondemand::json_type::string: { |
456 | | // Extract the raw (unescaped) string value rather than its JSON |
457 | | // representation. simdjson::to_json_string would keep the surrounding |
458 | | // double quotes (e.g. "2026-05-20"), which leaks into the result and |
459 | | // makes scalar-string variants inconsistent with structured ones. |
460 | 10 | std::string_view value_str = value.get_string().value(); |
461 | 10 | column->insert_data(value_str.data(), value_str.length()); |
462 | 10 | break; |
463 | 0 | } |
464 | 59 | default: { |
465 | 59 | auto value_str = simdjson::to_json_string(value).value(); |
466 | 59 | column->insert_data(value_str.data(), value_str.length()); |
467 | 59 | } |
468 | 69 | } |
469 | 69 | } |
470 | | }; |
471 | | |
472 | | class FunctionVariantElementByInteger final : public FunctionVariantElement { |
473 | | public: |
474 | | static constexpr auto name = FunctionVariantElement::name; |
475 | 8 | static FunctionPtr create() { return std::make_shared<FunctionVariantElementByInteger>(); } |
476 | | |
477 | 7 | DataTypes get_variadic_argument_types_impl() const override { |
478 | 7 | return {std::make_shared<DataTypeVariant>(), std::make_shared<DataTypeInt64>()}; |
479 | 7 | } |
480 | | }; |
481 | | |
482 | 7 | void register_function_variant_element(SimpleFunctionFactory& factory) { |
483 | 7 | factory.register_function<FunctionVariantElement>(); |
484 | 7 | factory.register_function<FunctionVariantElementByInteger>(); |
485 | 7 | } |
486 | | |
487 | | } // namespace doris |