Coverage Report

Created: 2026-07-29 14:27

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/exprs/function/parse/variant_string_parse.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/function/parse/variant_string_parse.h"
19
20
#include <cctz/time_zone.h>
21
#include <fmt/compile.h>
22
#include <fmt/format.h>
23
24
#include <algorithm>
25
#include <array>
26
#include <chrono>
27
#include <cstdint>
28
#include <limits>
29
#include <string_view>
30
#include <unordered_set>
31
#include <utility>
32
33
#include "common/config.h"
34
#include "common/exception.h"
35
#include "core/value/variant/variant_batch_builder.h"
36
#include "core/value/variant/variant_parquet_encoding.h"
37
#include "util/json/simd_json_parser.h"
38
#include "util/utf8_check.h"
39
40
namespace doris {
41
namespace {
42
43
using variant_json::FormattedScalar;
44
45
1.83k
void append_char(FormattedScalar* result, char value) {
46
1.83k
    result->bytes[result->size++] = value;
47
1.83k
}
48
49
456
void append_unsigned(FormattedScalar* result, uint64_t value, size_t minimum_digits = 1) {
50
456
    std::array<char, 32> reversed {};
51
456
    size_t digits = 0;
52
856
    do {
53
856
        reversed[digits++] = static_cast<char>('0' + value % 10);
54
856
        value /= 10;
55
856
    } while (value != 0);
56
895
    while (digits < minimum_digits) {
57
439
        reversed[digits++] = '0';
58
439
    }
59
1.75k
    while (digits != 0) {
60
1.29k
        append_char(result, reversed[--digits]);
61
1.29k
    }
62
456
}
63
64
80
void append_year(FormattedScalar* result, int64_t year) {
65
80
    if (year >= 0 && year <= 9999) {
66
80
        append_unsigned(result, static_cast<uint64_t>(year), 4);
67
80
        return;
68
80
    }
69
0
    if (year >= 0) {
70
0
        append_char(result, '+');
71
0
        append_unsigned(result, static_cast<uint64_t>(year), 4);
72
0
        return;
73
0
    }
74
0
    append_char(result, '-');
75
0
    append_unsigned(result, static_cast<uint64_t>(-(year + 1)) + 1, 4);
76
0
}
77
78
80
void append_date_time(FormattedScalar* result, const cctz::civil_second& civil, bool include_time) {
79
80
    append_year(result, civil.year());
80
80
    append_char(result, '-');
81
80
    append_unsigned(result, civil.month(), 2);
82
80
    append_char(result, '-');
83
80
    append_unsigned(result, civil.day(), 2);
84
80
    if (!include_time) {
85
34
        return;
86
34
    }
87
46
    append_char(result, ' ');
88
46
    append_unsigned(result, civil.hour(), 2);
89
46
    append_char(result, ':');
90
46
    append_unsigned(result, civil.minute(), 2);
91
46
    append_char(result, ':');
92
46
    append_unsigned(result, civil.second(), 2);
93
46
}
94
95
49
void append_fraction(FormattedScalar* result, uint64_t fraction, uint8_t digits) {
96
49
    append_char(result, '.');
97
49
    append_unsigned(result, fraction, digits);
98
49
}
99
100
10
void append_offset(FormattedScalar* result, int offset_seconds) {
101
10
    const bool negative = offset_seconds < 0;
102
10
    const uint64_t magnitude_seconds =
103
10
            negative ? static_cast<uint64_t>(-(static_cast<int64_t>(offset_seconds)))
104
10
                     : static_cast<uint64_t>(offset_seconds);
105
10
    append_char(result, negative ? '-' : '+');
106
10
    append_unsigned(result, magnitude_seconds / 3600, 2);
107
10
    append_char(result, ':');
108
10
    append_unsigned(result, (magnitude_seconds % 3600) / 60, 2);
109
10
    if (magnitude_seconds % 60 != 0) {
110
0
        append_char(result, ':');
111
0
        append_unsigned(result, magnitude_seconds % 60, 2);
112
0
    }
113
10
}
114
115
49
std::pair<int64_t, uint64_t> split_epoch(int64_t value, int64_t units_per_second) {
116
49
    int64_t seconds = value / units_per_second;
117
49
    int64_t fraction = value % units_per_second;
118
49
    if (fraction < 0) {
119
14
        --seconds;
120
14
        fraction += units_per_second;
121
14
    }
122
49
    return {seconds, static_cast<uint64_t>(fraction)};
123
49
}
124
125
167
StringRef to_string_ref(std::string_view value) {
126
167
    return {value.data(), value.size()};
127
167
}
128
129
170
void require_json_key_length(std::string_view key, uint32_t maximum) {
130
170
    if (key.size() > maximum) {
131
4
        throw Exception(ErrorCode::INVALID_ARGUMENT,
132
4
                        "Variant JSON object key length {} exceeds maximum {} bytes", key.size(),
133
4
                        maximum);
134
4
    }
135
170
}
136
137
class JsonTreeCollector {
138
public:
139
    JsonTreeCollector(VariantBatchBuilder::Row& builder, const JsonToVariantOptions& options)
140
192
            : _builder(builder), _options(options) {}
141
142
917
    void collect(SimdJSONParser::Element element, uint32_t depth) {
143
917
        variant_json::require_json_depth(depth);
144
917
        if (element.isNull()) {
145
28
            _builder.add_null();
146
889
        } else if (element.isBool()) {
147
12
            _builder.add_bool(element.getBool());
148
877
        } else if (element.isInt64()) {
149
172
            _builder.add_int(element.getInt64());
150
705
        } else if (element.isUInt64()) {
151
1
            _builder.add_largeint(static_cast<__int128>(element.getUInt64()));
152
704
        } else if (element.isDouble()) {
153
0
            _builder.add_double(element.getDouble());
154
704
        } else if (element.isString()) {
155
13
            _builder.add_string(to_string_ref(element.getString()));
156
691
        } else if (element.isArray()) {
157
551
            collect_array(element.getArray(), depth);
158
551
        } else if (element.isObject()) {
159
137
            collect_object(element.getObject(), depth);
160
137
        }
161
917
    }
162
163
private:
164
551
    void collect_array(const SimdJSONParser::Array& array, uint32_t depth) {
165
551
        auto scope = _builder.start_array();
166
571
        for (SimdJSONParser::Element child : array) {
167
571
            collect(child, depth + 1);
168
571
        }
169
551
        scope.finish();
170
551
    }
171
172
137
    void collect_object(const SimdJSONParser::Object& object, uint32_t depth) {
173
137
        auto scope = _builder.start_object();
174
137
        if (!_options.check_duplicate_json_path) {
175
142
            for (const auto& [key, child] : object) {
176
142
                require_json_key_length(key, _options.max_json_key_length);
177
142
                scope.add_key(to_string_ref(key));
178
142
                collect(child, depth + 1);
179
142
            }
180
123
            scope.finish();
181
123
            return;
182
123
        }
183
184
14
        std::unordered_set<std::string_view> seen_keys;
185
14
        seen_keys.reserve(object.size());
186
26
        for (const auto& [key, child] : object) {
187
26
            require_json_key_length(key, _options.max_json_key_length);
188
26
            if (seen_keys.emplace(key).second) {
189
15
                scope.add_key(to_string_ref(key));
190
15
                collect(child, depth + 1);
191
15
            } else {
192
11
                validate_ignored(child, depth + 1);
193
11
            }
194
26
        }
195
14
        scope.finish();
196
14
    }
197
198
141
    void validate_ignored(SimdJSONParser::Element element, uint32_t depth) const {
199
141
        variant_json::require_json_depth(depth);
200
141
        if (element.isArray()) {
201
129
            for (SimdJSONParser::Element child : element.getArray()) {
202
129
                validate_ignored(child, depth + 1);
203
129
            }
204
129
        } else if (element.isObject()) {
205
2
            for (const auto& [key, child] : element.getObject()) {
206
2
                require_json_key_length(key, _options.max_json_key_length);
207
2
                validate_ignored(child, depth + 1);
208
2
            }
209
2
        }
210
141
    }
211
212
    VariantBatchBuilder::Row& _builder;
213
    const JsonToVariantOptions& _options;
214
};
215
216
} // namespace
217
218
namespace variant_json {
219
220
2.13k
void require_json_depth(uint32_t depth) {
221
2.13k
    if (depth > VARIANT_MAX_NESTING_DEPTH) {
222
5
        throw Exception(ErrorCode::INVALID_ARGUMENT,
223
5
                        "Variant JSON traversal exceeds maximum depth {}",
224
5
                        VARIANT_MAX_NESTING_DEPTH);
225
5
    }
226
2.13k
}
227
228
614
void require_exact_json_value(VariantRef value) {
229
614
    const size_t encoded_size = value.value_size();
230
614
    if (encoded_size != value.value.size) {
231
2
        throw Exception(ErrorCode::CORRUPTION,
232
2
                        "Variant value has {} trailing bytes after its {} byte root",
233
2
                        value.value.size - encoded_size, encoded_size);
234
2
    }
235
614
}
236
237
288
void require_valid_json_utf8(StringRef value, const char* description) {
238
288
    if (value.size != 0 && !validate_utf8(value.data, value.size)) {
239
2
        throw Exception(ErrorCode::CORRUPTION, "Variant {} is not valid UTF-8", description);
240
2
    }
241
288
}
242
243
161
void require_json_object_key(StringRef key, StringRef previous_key, uint32_t field_index) {
244
161
    require_valid_json_utf8(key, "object key");
245
161
    if (field_index != 0 && previous_key.compare(key) >= 0) {
246
4
        throw Exception(ErrorCode::CORRUPTION,
247
4
                        "Variant object keys are not strictly byte-sorted at field {}",
248
4
                        field_index);
249
4
    }
250
161
}
251
252
0
[[noreturn]] void throw_unsupported_json_primitive(VariantPrimitiveId id) {
253
0
    throw Exception(ErrorCode::INVALID_ARGUMENT, "Unsupported Variant JSON primitive id {}",
254
0
                    static_cast<uint8_t>(id));
255
0
}
256
257
179
FormattedScalar format_json_int(int64_t value) {
258
179
    FormattedScalar result;
259
179
    char* end = fmt::format_to(result.bytes.data(), FMT_COMPILE("{}"), value);
260
179
    result.size = end - result.bytes.data();
261
179
    return result;
262
179
}
263
264
1
FormattedScalar format_json_float(float value) {
265
1
    FormattedScalar result;
266
1
    char* end = fmt::format_to(result.bytes.data(), FMT_COMPILE("{:.{}g}"), value,
267
1
                               std::numeric_limits<float>::digits10 + 1);
268
1
    result.size = end - result.bytes.data();
269
1
    return result;
270
1
}
271
272
57
FormattedScalar format_json_double(double value) {
273
57
    FormattedScalar result;
274
57
    char* end = fmt::format_to(result.bytes.data(), FMT_COMPILE("{:.{}g}"), value,
275
57
                               std::numeric_limits<double>::digits10 + 1);
276
57
    result.size = end - result.bytes.data();
277
57
    return result;
278
57
}
279
280
4
FormattedScalar format_json_decimal(VariantDecimal value) {
281
4
    FormattedScalar result;
282
4
    const bool negative = value.unscaled < 0;
283
4
    std::array<char, 39> reversed {};
284
4
    size_t digits = 0;
285
4
    unsigned __int128 remaining = variant_unsigned_magnitude(value.unscaled);
286
54
    do {
287
54
        reversed[digits++] = static_cast<char>('0' + remaining % 10);
288
54
        remaining /= 10;
289
54
    } while (remaining != 0);
290
291
4
    if (negative) {
292
1
        append_char(&result, '-');
293
1
    }
294
4
    if (value.scale == 0) {
295
41
        while (digits != 0) {
296
39
            append_char(&result, reversed[--digits]);
297
39
        }
298
2
        return result;
299
2
    }
300
2
    if (digits <= value.scale) {
301
0
        append_char(&result, '0');
302
0
        append_char(&result, '.');
303
0
        for (size_t zero = digits; zero < value.scale; ++zero) {
304
0
            append_char(&result, '0');
305
0
        }
306
0
        while (digits != 0) {
307
0
            append_char(&result, reversed[--digits]);
308
0
        }
309
0
        return result;
310
0
    }
311
312
12
    while (digits > value.scale) {
313
10
        append_char(&result, reversed[--digits]);
314
10
    }
315
2
    append_char(&result, '.');
316
7
    while (digits != 0) {
317
5
        append_char(&result, reversed[--digits]);
318
5
    }
319
2
    return result;
320
2
}
321
322
34
FormattedScalar format_json_date(int32_t days_since_epoch) {
323
34
    constexpr int64_t SECONDS_PER_DAY = 86'400;
324
34
    const int64_t seconds = static_cast<int64_t>(days_since_epoch) * SECONDS_PER_DAY;
325
34
    const auto lookup =
326
34
            cctz::utc_time_zone().lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds)));
327
34
    FormattedScalar result;
328
34
    append_date_time(&result, lookup.cs, false);
329
34
    return result;
330
34
}
331
332
FormattedScalar format_json_timestamp(int64_t value, uint8_t fractional_digits, bool utc_adjusted,
333
46
                                      const cctz::time_zone* timezone) {
334
46
    if (fractional_digits != 6 && fractional_digits != 9) {
335
0
        throw Exception(ErrorCode::INVALID_ARGUMENT,
336
0
                        "Variant timestamp JSON precision must be 6 or 9, got {}",
337
0
                        fractional_digits);
338
0
    }
339
46
    const int64_t units_per_second = fractional_digits == 6 ? 1'000'000 : 1'000'000'000;
340
46
    const auto [seconds, fraction] = split_epoch(value, units_per_second);
341
46
    const cctz::time_zone& zone =
342
46
            utc_adjusted && timezone != nullptr ? *timezone : cctz::utc_time_zone();
343
46
    const auto lookup = zone.lookup(cctz::time_point<cctz::seconds>(cctz::seconds(seconds)));
344
46
    FormattedScalar result;
345
46
    append_date_time(&result, lookup.cs, true);
346
46
    append_fraction(&result, fraction, fractional_digits);
347
46
    if (utc_adjusted) {
348
10
        append_offset(&result, lookup.offset);
349
10
    }
350
46
    return result;
351
46
}
352
353
4
FormattedScalar format_json_time_micros(int64_t value) {
354
4
    constexpr int64_t MICROS_PER_SECOND = 1'000'000;
355
4
    constexpr int64_t MICROS_PER_DAY = 86'400 * MICROS_PER_SECOND;
356
4
    if (value < 0 || value >= MICROS_PER_DAY) {
357
1
        throw Exception(ErrorCode::INVALID_ARGUMENT,
358
1
                        "Variant time value {} is outside [0, {}) microseconds", value,
359
1
                        MICROS_PER_DAY);
360
1
    }
361
3
    const auto [seconds, micros] = split_epoch(value, MICROS_PER_SECOND);
362
3
    FormattedScalar result;
363
3
    append_unsigned(&result, static_cast<uint64_t>(seconds / 3600), 2);
364
3
    append_char(&result, ':');
365
3
    append_unsigned(&result, static_cast<uint64_t>((seconds % 3600) / 60), 2);
366
3
    append_char(&result, ':');
367
3
    append_unsigned(&result, static_cast<uint64_t>(seconds % 60), 2);
368
3
    append_fraction(&result, micros, 6);
369
3
    return result;
370
4
}
371
372
3
FormattedScalar format_json_uuid(const std::array<uint8_t, 16>& value) {
373
3
    static constexpr char HEX[] = "0123456789abcdef";
374
3
    FormattedScalar result;
375
51
    for (size_t index = 0; index < value.size(); ++index) {
376
48
        if (index == 4 || index == 6 || index == 8 || index == 10) {
377
12
            append_char(&result, '-');
378
12
        }
379
48
        append_char(&result, HEX[value[index] >> 4]);
380
48
        append_char(&result, HEX[value[index] & 0x0F]);
381
48
    }
382
3
    return result;
383
3
}
384
385
} // namespace variant_json
386
387
32
JsonToVariantOptions JsonToVariantOptions::current_config() {
388
32
    return {.max_json_key_length = static_cast<uint32_t>(config::variant_max_json_key_length),
389
32
            .throw_on_invalid_json = config::variant_throw_exeception_on_invalid_json,
390
32
            .check_duplicate_json_path = config::variant_enable_duplicate_json_path_check};
391
32
}
392
393
struct JsonStringToVariantEncoder::Impl {
394
    enum class State : uint8_t { COLLECTING, FINISHED, FAILED };
395
396
180
    explicit Impl(JsonToVariantOptions options_) : options(options_) {
397
180
        if (options.max_json_key_length == 0) {
398
1
            throw Exception(ErrorCode::INVALID_ARGUMENT,
399
1
                            "Variant maximum JSON key length must be positive");
400
1
        }
401
180
    }
402
403
379
    void require_collecting() const {
404
379
        if (state == State::FINISHED) {
405
4
            throw Exception(ErrorCode::INVALID_ARGUMENT,
406
4
                            "Variant JSON encoder is already finished");
407
4
        }
408
375
        if (state == State::FAILED) {
409
3
            throw Exception(ErrorCode::INVALID_ARGUMENT,
410
3
                            "Variant JSON encoder is in a terminal failed state");
411
3
        }
412
375
    }
413
414
210
    void add_json_row(StringRef json) {
415
210
        auto row = builder.begin_row();
416
210
        if (json.size != 0 && json.data == nullptr) {
417
1
            throw Exception(ErrorCode::INVALID_ARGUMENT,
418
1
                            "Variant JSON input has a null data pointer");
419
1
        }
420
209
        if (json.size == 0) {
421
3
            auto object = row.start_object();
422
3
            object.finish();
423
3
            row.finish();
424
3
            return;
425
3
        }
426
427
206
        SimdJSONParser::Element root;
428
206
        if (!parser.parse(json.data, json.size, root)) {
429
14
            if (options.throw_on_invalid_json) {
430
7
                throw Exception(ErrorCode::INVALID_ARGUMENT, "Failed to parse JSON as Variant");
431
7
            }
432
7
            row.add_string(json);
433
7
            row.finish();
434
7
            return;
435
14
        }
436
192
        JsonTreeCollector(row, options).collect(root, 0);
437
192
        row.finish();
438
192
    }
439
440
    JsonToVariantOptions options;
441
    VariantBatchBuilder builder;
442
    SimdJSONParser parser;
443
    State state = State::COLLECTING;
444
};
445
446
JsonStringToVariantEncoder::JsonStringToVariantEncoder()
447
12
        : JsonStringToVariantEncoder(JsonToVariantOptions::current_config()) {}
448
449
JsonStringToVariantEncoder::JsonStringToVariantEncoder(JsonToVariantOptions options)
450
180
        : _impl(std::make_unique<Impl>(options)) {}
451
452
179
JsonStringToVariantEncoder::~JsonStringToVariantEncoder() = default;
453
0
JsonStringToVariantEncoder::JsonStringToVariantEncoder(JsonStringToVariantEncoder&&) noexcept =
454
        default;
455
JsonStringToVariantEncoder& JsonStringToVariantEncoder::operator=(
456
0
        JsonStringToVariantEncoder&&) noexcept = default;
457
458
185
void JsonStringToVariantEncoder::add_json(StringRef json) {
459
185
    _impl->require_collecting();
460
185
    try {
461
185
        _impl->add_json_row(json);
462
185
    } catch (...) {
463
12
        _impl->state = Impl::State::FAILED;
464
12
        throw;
465
12
    }
466
185
}
467
468
28
Status JsonStringToVariantEncoder::try_add_json(StringRef json) {
469
28
    _impl->require_collecting();
470
28
    try {
471
28
        _impl->add_json_row(json);
472
28
        return Status::OK();
473
28
    } catch (const Exception& exception) {
474
11
        if (exception.code() == ErrorCode::INVALID_ARGUMENT) {
475
11
            return exception.to_status();
476
11
        }
477
0
        _impl->state = Impl::State::FAILED;
478
0
        throw;
479
11
    } catch (...) {
480
0
        _impl->state = Impl::State::FAILED;
481
0
        throw;
482
0
    }
483
28
}
484
485
166
VariantBatchBuilder JsonStringToVariantEncoder::finish_batch() {
486
166
    _impl->require_collecting();
487
166
    try {
488
166
        VariantBatchBuilder block = _impl->builder.finish_batch();
489
166
        _impl->state = Impl::State::FINISHED;
490
166
        return block;
491
166
    } catch (...) {
492
0
        _impl->state = Impl::State::FAILED;
493
0
        throw;
494
0
    }
495
166
}
496
497
} // namespace doris