Coverage Report

Created: 2026-03-31 15:05

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
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 <string>
24
#include <string_view>
25
#include <utility>
26
#include <vector>
27
28
#include "common/status.h"
29
#include "core/assert_cast.h"
30
#include "core/block/block.h"
31
#include "core/column/column.h"
32
#include "core/column/column_nullable.h"
33
#include "core/column/column_string.h"
34
#include "core/column/column_variant.h"
35
#include "core/column/subcolumn_tree.h"
36
#include "core/data_type/data_type.h"
37
#include "core/data_type/data_type_nothing.h"
38
#include "core/data_type/data_type_nullable.h"
39
#include "core/data_type/data_type_string.h"
40
#include "core/data_type/data_type_variant.h"
41
#include "core/string_ref.h"
42
#include "exprs/function/function.h"
43
#include "exprs/function/function_helpers.h"
44
#include "exprs/function/simple_function_factory.h"
45
#include "exprs/json_functions.h"
46
#include "simdjson.h"
47
#include "util/defer_op.h"
48
#include "util/json/path_in_data.h"
49
50
namespace doris {
51
52
class FunctionVariantElement : public IFunction {
53
public:
54
    static constexpr auto name = "element_at";
55
141
    static FunctionPtr create() { return std::make_shared<FunctionVariantElement>(); }
56
57
    // Get function name.
58
1
    String get_name() const override { return name; }
59
60
270
    bool use_default_implementation_for_nulls() const override { return false; }
61
62
132
    size_t get_number_of_arguments() const override { return 2; }
63
64
138
    ColumnNumbers get_arguments_that_are_always_constant() const override { return {1}; }
65
66
8
    DataTypes get_variadic_argument_types_impl() const override {
67
8
        return {std::make_shared<DataTypeVariant>(), std::make_shared<DataTypeString>()};
68
8
    }
69
70
132
    DataTypePtr get_return_type_impl(const DataTypes& arguments) const override {
71
132
        DCHECK_EQ(arguments[0]->get_primitive_type(), TYPE_VARIANT)
72
0
                << "First argument for function: " << name
73
0
                << " should be DataTypeVariant but it has type " << arguments[0]->get_name() << ".";
74
132
        DCHECK(is_string_type(arguments[1]->get_primitive_type()))
75
0
                << "Second argument for function: " << name << " should be String but it has type "
76
0
                << arguments[1]->get_name() << ".";
77
132
        auto arg_variant = remove_nullable(arguments[0]);
78
132
        const auto& data_type_object = assert_cast<const DataTypeVariant&>(*arg_variant);
79
132
        return make_nullable(
80
132
                std::make_shared<DataTypeVariant>(data_type_object.variant_max_subcolumns_count(),
81
132
                                                  data_type_object.enable_doc_mode()));
82
132
    }
83
84
    // wrap variant column with nullable
85
    // 1. if variant is null root(empty or nothing as root), then nullable map is all null
86
    // 2. if variant is scalar variant, then use the root's nullable map
87
    // 3. if variant is hierarchical variant, then create a nullable map with all none null
88
138
    ColumnPtr wrap_variant_nullable(ColumnPtr col) const {
89
138
        const auto& var = assert_cast<const ColumnVariant&>(*col);
90
138
        if (var.is_null_root()) {
91
20
            return make_nullable(col, true);
92
20
        }
93
118
        if (var.is_scalar_variant() && var.get_root()->is_nullable()) {
94
44
            const auto* nullable = assert_cast<const ColumnNullable*>(var.get_root().get());
95
44
            return ColumnNullable::create(
96
44
                    col, nullable->get_null_map_column_ptr()->clone_resized(col->size()));
97
44
        }
98
74
        return make_nullable(col);
99
118
    }
100
101
    Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
102
138
                        uint32_t result, size_t input_rows_count) const override {
103
138
        const auto* variant_col = check_and_get_column<ColumnVariant>(
104
138
                remove_nullable(block.get_by_position(arguments[0]).column).get());
105
138
        if (!variant_col) {
106
0
            return Status::RuntimeError(
107
0
                    fmt::format("unsupported types for function {}({}, {})", get_name(),
108
0
                                block.get_by_position(arguments[0]).type->get_name(),
109
0
                                block.get_by_position(arguments[1]).type->get_name()));
110
0
        }
111
138
        if (block.empty()) {
112
0
            block.replace_by_position(result, block.get_by_position(result).type->create_column());
113
0
            return Status::OK();
114
0
        }
115
116
138
        auto index_column = block.get_by_position(arguments[1]).column;
117
138
        ColumnPtr result_column;
118
138
        RETURN_IF_ERROR(get_element_column(*variant_col, index_column, &result_column));
119
138
        if (block.get_by_position(result).type->is_nullable()) {
120
138
            result_column = wrap_variant_nullable(result_column);
121
138
        }
122
138
        block.replace_by_position(result, result_column);
123
138
        return Status::OK();
124
138
    }
125
126
private:
127
    // Return sub-path by specified prefix.
128
    // For example, for prefix a.b:
129
    // a.b.c.d -> c.d, a.b.c -> c
130
    static std::optional<std::string_view> get_sub_path(const std::string_view& path,
131
14
                                                        const std::string_view& prefix) {
132
14
        if (path.size() <= prefix.size() || path[prefix.size()] != '.') {
133
11
            return std::nullopt;
134
11
        }
135
3
        return path.substr(prefix.size() + 1);
136
14
    }
137
138
    // Extract and populate sparse column data with given path prefix
139
    // Copies data from source sparse column, extracting only the sub-paths that match the prefix
140
    static void _extract_sparse_column_from_source(ColumnVariant* src_ptr, const PathInData& path,
141
9
                                                   ColumnVariant::MutablePtr& target_ptr) {
142
9
        ColumnVariant::Subcolumn root {0, true, true};
143
        // no root, no sparse column
144
9
        const auto& sparse_data_map = assert_cast<const ColumnMap&>(*src_ptr->get_sparse_column());
145
9
        const auto& src_sparse_data_offsets = sparse_data_map.get_offsets();
146
9
        const auto& src_sparse_data_paths =
147
9
                assert_cast<const ColumnString&>(sparse_data_map.get_keys());
148
9
        const auto& src_sparse_data_values =
149
9
                assert_cast<const ColumnString&>(sparse_data_map.get_values());
150
9
        auto& sparse_data_offsets =
151
9
                assert_cast<ColumnMap&>(*target_ptr->get_sparse_column()->assume_mutable())
152
9
                        .get_offsets();
153
9
        auto [sparse_data_paths, sparse_data_values] =
154
9
                target_ptr->get_sparse_data_paths_and_values();
155
9
        StringRef prefix_ref(path.get_path());
156
9
        std::string_view path_prefix(prefix_ref.data, prefix_ref.size);
157
122
        for (size_t i = 0; i != src_sparse_data_offsets.size(); ++i) {
158
113
            size_t start = src_sparse_data_offsets[ssize_t(i) - 1];
159
113
            size_t end = src_sparse_data_offsets[ssize_t(i)];
160
113
            size_t lower_bound_index = ColumnVariant::find_path_lower_bound_in_sparse_data(
161
113
                    prefix_ref, src_sparse_data_paths, start, end);
162
164
            for (; lower_bound_index != end; ++lower_bound_index) {
163
64
                auto path_ref = src_sparse_data_paths.get_data_at(lower_bound_index);
164
64
                std::string_view nested_path(path_ref.data, path_ref.size);
165
64
                if (!nested_path.starts_with(path_prefix)) {
166
13
                    break;
167
13
                }
168
                // Don't include path that is equal to the prefix.
169
51
                if (nested_path.size() != path_prefix.size()) {
170
13
                    auto sub_path_optional = get_sub_path(nested_path, path_prefix);
171
13
                    if (!sub_path_optional.has_value()) {
172
11
                        continue;
173
11
                    }
174
2
                    std::string_view sub_path = *sub_path_optional;
175
2
                    sparse_data_paths->insert_data(sub_path.data(), sub_path.size());
176
2
                    sparse_data_values->insert_from(src_sparse_data_values, lower_bound_index);
177
38
                } else {
178
                    // insert into root column, example:  access v['b'] and b is in sparse column
179
                    // data example:
180
                    // {"b" : 123}
181
                    // {"b" : {"c" : 456}}
182
                    // b maybe in sparse column, and b.c is in subolumn, put `b` into root column to distinguish
183
                    // from "" which is empty path and root
184
38
                    root.deserialize_from_binary_column(&src_sparse_data_values, lower_bound_index);
185
38
                }
186
51
            }
187
113
            if (root.size() == sparse_data_offsets.size()) {
188
75
                root.insert_default();
189
75
            }
190
113
            sparse_data_offsets.push_back(sparse_data_paths->size());
191
113
        }
192
9
        target_ptr->get_subcolumns().create_root(root);
193
9
        target_ptr->get_doc_value_column()->assume_mutable()->resize(src_ptr->size());
194
9
        target_ptr->set_num_rows(src_ptr->size());
195
9
    }
196
197
    // Extract and populate sparse column data from doc_value column with given path prefix
198
    // Copies data from source doc_value column, extracting only the sub-paths that match the prefix
199
    static void _extract_doc_value_column_from_source(ColumnVariant* src_ptr,
200
                                                      const PathInData& path,
201
51
                                                      ColumnVariant::MutablePtr& target_ptr) {
202
51
        ColumnVariant::Subcolumn root {0, true, true};
203
51
        const auto& doc_value_data_map =
204
51
                assert_cast<const ColumnMap&>(*src_ptr->get_doc_value_column());
205
51
        const auto& src_doc_value_data_offsets = doc_value_data_map.get_offsets();
206
51
        const auto& src_doc_value_data_paths =
207
51
                assert_cast<const ColumnString&>(doc_value_data_map.get_keys());
208
51
        const auto& src_doc_value_data_values =
209
51
                assert_cast<const ColumnString&>(doc_value_data_map.get_values());
210
        // Write extracted data into target's doc_value column (not sparse) to preserve
211
        // doc mode invariant: doc_mode columns must not have sparse data.
212
51
        auto& doc_value_offsets =
213
51
                assert_cast<ColumnMap&>(*target_ptr->get_doc_value_column()->assume_mutable())
214
51
                        .get_offsets();
215
51
        auto [doc_value_paths, doc_value_values] =
216
51
                target_ptr->get_doc_value_data_paths_and_values();
217
51
        StringRef prefix_ref(path.get_path());
218
51
        std::string_view path_prefix(prefix_ref.data, prefix_ref.size);
219
102
        for (size_t i = 0; i != src_doc_value_data_offsets.size(); ++i) {
220
51
            size_t start = src_doc_value_data_offsets[ssize_t(i) - 1];
221
51
            size_t end = src_doc_value_data_offsets[ssize_t(i)];
222
51
            size_t lower_bound_index = ColumnVariant::find_path_lower_bound_in_sparse_data(
223
51
                    prefix_ref, src_doc_value_data_paths, start, end);
224
83
            for (; lower_bound_index != end; ++lower_bound_index) {
225
63
                auto path_ref = src_doc_value_data_paths.get_data_at(lower_bound_index);
226
63
                std::string_view nested_path(path_ref.data, path_ref.size);
227
63
                if (!nested_path.starts_with(path_prefix)) {
228
31
                    break;
229
31
                }
230
32
                if (nested_path.size() != path_prefix.size()) {
231
1
                    auto sub_path_optional = get_sub_path(nested_path, path_prefix);
232
1
                    if (!sub_path_optional.has_value()) {
233
0
                        continue;
234
0
                    }
235
1
                    std::string_view sub_path = *sub_path_optional;
236
1
                    doc_value_paths->insert_data(sub_path.data(), sub_path.size());
237
1
                    doc_value_values->insert_from(src_doc_value_data_values, lower_bound_index);
238
31
                } else {
239
31
                    root.deserialize_from_binary_column(&src_doc_value_data_values,
240
31
                                                        lower_bound_index);
241
31
                }
242
32
            }
243
51
            if (root.size() == doc_value_offsets.size()) {
244
20
                root.insert_default();
245
20
            }
246
51
            doc_value_offsets.push_back(doc_value_paths->size());
247
51
        }
248
51
        target_ptr->get_subcolumns().create_root(root);
249
51
        target_ptr->get_sparse_column()->assume_mutable()->resize(src_ptr->size());
250
51
        target_ptr->set_num_rows(src_ptr->size());
251
51
    }
252
253
    static Status get_element_column(const ColumnVariant& src, const ColumnPtr& index_column,
254
139
                                     ColumnPtr* result) {
255
139
        std::string field_name = index_column->get_data_at(0).to_string();
256
139
        if (src.empty()) {
257
0
            *result = ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode());
258
            // src subcolumns empty but src row count may not be 0
259
0
            (*result)->assume_mutable()->insert_many_defaults(src.size());
260
            // ColumnVariant should be finalized before parsing, finalize maybe modify original column structure
261
0
            (*result)->assume_mutable()->finalize();
262
0
            return Status::OK();
263
0
        }
264
139
        if (src.is_scalar_variant() && is_string_type(src.get_root_type()->get_primitive_type())) {
265
            // use parser to extract from root
266
74
            auto type = std::make_shared<DataTypeString>();
267
74
            MutableColumnPtr result_column = type->create_column();
268
74
            const ColumnString& docs =
269
74
                    *check_and_get_column<ColumnString>(remove_nullable(src.get_root()).get());
270
74
            simdjson::ondemand::parser parser;
271
74
            std::vector<JsonPath> parsed_paths;
272
74
            if (field_name.empty() || field_name[0] != '$') {
273
74
                field_name = "$." + field_name;
274
74
            }
275
74
            JsonFunctions::parse_json_paths(field_name, &parsed_paths);
276
74
            ColumnString* col_str = assert_cast<ColumnString*>(result_column.get());
277
149
            for (size_t i = 0; i < docs.size(); ++i) {
278
75
                if (!extract_from_document(parser, docs.get_data_at(i), parsed_paths, col_str)) {
279
15
                    VLOG_DEBUG << "failed to parse " << docs.get_data_at(i) << ", field "
280
0
                               << field_name;
281
15
                    result_column->insert_default();
282
15
                }
283
75
            }
284
74
            *result = ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(), type,
285
74
                                            std::move(result_column));
286
74
            (*result)->assume_mutable()->finalize();
287
74
            return Status::OK();
288
74
        } else {
289
65
            auto mutable_src = src.clone_finalized();
290
65
            auto* mutable_ptr = assert_cast<ColumnVariant*>(mutable_src.get());
291
65
            PathInData path(field_name);
292
65
            ColumnVariant::Subcolumns subcolumns = mutable_ptr->get_subcolumns();
293
65
            const auto* node = subcolumns.find_exact(path);
294
65
            MutableColumnPtr result_col =
295
65
                    ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode());
296
65
            ColumnVariant::Subcolumns new_subcolumns;
297
298
65
            if (node != nullptr) {
299
5
                std::vector<decltype(node)> nodes;
300
5
                PathsInData paths;
301
5
                ColumnVariant::Subcolumns::get_leaves_of_node(node, nodes, paths);
302
5
                for (const auto* n : nodes) {
303
5
                    PathInData new_path = n->path.copy_pop_front();
304
5
                    VLOG_DEBUG << "add node " << new_path.get_path()
305
0
                               << ", data size: " << n->data.size()
306
0
                               << ", finalized size: " << n->data.get_finalized_column().size()
307
0
                               << ", common type: " << n->data.get_least_common_type()->get_name();
308
                    // if new_path is empty, indicate it's the root column, but adding a root will return false when calling add
309
5
                    if (!new_subcolumns.add(new_path, n->data)) {
310
5
                        VLOG_DEBUG << "failed to add node " << new_path.get_path();
311
5
                    }
312
5
                }
313
314
                // handle the root node
315
5
                if (new_subcolumns.empty() && !nodes.empty()) {
316
5
                    CHECK_EQ(nodes.size(), 1);
317
5
                    new_subcolumns.create_root(ColumnVariant::Subcolumn {
318
5
                            nodes[0]->data.get_finalized_column_ptr()->assume_mutable(),
319
5
                            nodes[0]->data.get_least_common_type(), true, true});
320
5
                    auto container =
321
5
                            ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(),
322
5
                                                  std::move(new_subcolumns));
323
5
                    result_col->insert_range_from(*container, 0, container->size());
324
5
                } else {
325
0
                    auto container =
326
0
                            ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(),
327
0
                                                  std::move(new_subcolumns));
328
0
                    container->clear_sparse_column();
329
0
                    _extract_sparse_column_from_source(mutable_ptr, path, container);
330
0
                    result_col->insert_range_from(*container, 0, container->size());
331
0
                }
332
60
            } else {
333
60
                auto container =
334
60
                        ColumnVariant::create(src.max_subcolumns_count(), src.enable_doc_mode(),
335
60
                                              std::move(new_subcolumns));
336
60
                const auto& sparse_offsets = mutable_ptr->serialized_sparse_column_offsets();
337
60
                if (sparse_offsets.back() == sparse_offsets[-1]) {
338
51
                    _extract_doc_value_column_from_source(mutable_ptr, path, container);
339
51
                } else {
340
9
                    _extract_sparse_column_from_source(mutable_ptr, path, container);
341
9
                }
342
60
                result_col->insert_range_from(*container, 0, container->size());
343
60
            }
344
65
            *result = result_col->get_ptr();
345
            // ColumnVariant should be finalized before parsing, finalize maybe modify original column structure
346
65
            (*result)->assume_mutable()->finalize();
347
65
            VLOG_DEBUG << "dump new object "
348
0
                       << static_cast<const ColumnVariant*>(result_col.get())->debug_string()
349
0
                       << ", path " << path.get_path();
350
65
            return Status::OK();
351
65
        }
352
139
    }
353
354
    static Status extract_from_document(simdjson::ondemand::parser& parser, const StringRef& doc,
355
75
                                        const std::vector<JsonPath>& paths, ColumnString* column) {
356
75
        try {
357
75
            simdjson::padded_string json_str {doc.data, doc.size};
358
75
            simdjson::ondemand::document document = parser.iterate(json_str);
359
75
            simdjson::ondemand::object object = document.get_object();
360
75
            simdjson::ondemand::value value;
361
75
            RETURN_IF_ERROR(JsonFunctions::extract_from_object(object, paths, &value));
362
65
            _write_data_to_column(value, column);
363
65
        } catch (simdjson::simdjson_error& e) {
364
5
            VLOG_DEBUG << "simdjson parse exception: " << e.what();
365
5
            return Status::DataQualityError("simdjson parse exception {}", e.what());
366
5
        }
367
60
        return Status::OK();
368
75
    }
369
370
60
    static void _write_data_to_column(simdjson::ondemand::value& value, ColumnString* column) {
371
60
        switch (value.type()) {
372
0
        case simdjson::ondemand::json_type::null: {
373
0
            column->insert_default();
374
0
            break;
375
0
        }
376
0
        case simdjson::ondemand::json_type::boolean: {
377
0
            if (value.get_bool()) {
378
0
                column->insert_data("1", 1);
379
0
            } else {
380
0
                column->insert_data("0", 1);
381
0
            }
382
0
            break;
383
0
        }
384
60
        default: {
385
60
            auto value_str = simdjson::to_json_string(value).value();
386
60
            column->insert_data(value_str.data(), value_str.length());
387
60
        }
388
60
        }
389
60
    }
390
};
391
392
8
void register_function_variant_element(SimpleFunctionFactory& factory) {
393
8
    factory.register_function<FunctionVariantElement>();
394
8
}
395
396
} // namespace doris