Coverage Report

Created: 2026-08-07 00:14

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/table/iceberg_default_value.h
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
#pragma once
19
20
#include <gen_cpp/ExternalTableSchema_types.h>
21
#include <rapidjson/document.h>
22
#include <rapidjson/stringbuffer.h>
23
#include <rapidjson/writer.h>
24
25
#include <cstddef>
26
#include <deque>
27
#include <string>
28
#include <string_view>
29
#include <unordered_map>
30
#include <utility>
31
32
#include "common/status.h"
33
#include "core/assert_cast.h"
34
#include "core/column/column.h"
35
#include "core/data_type/data_type.h"
36
#include "core/data_type/data_type_array.h"
37
#include "core/data_type/data_type_map.h"
38
#include "core/data_type/data_type_nullable.h"
39
#include "core/data_type/data_type_struct.h"
40
#include "core/data_type/primitive_type.h"
41
#include "core/field.h"
42
#include "util/string_util.h"
43
#include "util/url_coding.h"
44
45
namespace doris::iceberg {
46
47
namespace detail {
48
49
18
inline const schema::external::TField* get_field_ptr(const schema::external::TFieldPtr& field_ptr) {
50
18
    if (!field_ptr.__isset.field_ptr || field_ptr.field_ptr == nullptr) {
51
0
        return nullptr;
52
0
    }
53
18
    return field_ptr.field_ptr.get();
54
18
}
55
56
inline const schema::external::TField* find_struct_child(
57
8
        const schema::external::TStructField& struct_field, const std::string& name) {
58
8
    if (!struct_field.__isset.fields) {
59
0
        return nullptr;
60
0
    }
61
13
    for (const auto& child_ptr : struct_field.fields) {
62
13
        const auto* child = get_field_ptr(child_ptr);
63
13
        if (child != nullptr && child->__isset.name && iequal(child->name, name)) {
64
6
            return child;
65
6
        }
66
13
    }
67
2
    for (const auto& child_ptr : struct_field.fields) {
68
2
        const auto* child = get_field_ptr(child_ptr);
69
2
        if (child == nullptr || !child->__isset.name_mapping) {
70
0
            continue;
71
0
        }
72
2
        for (const auto& alias : child->name_mapping) {
73
2
            if (iequal(alias, name)) {
74
2
                return child;
75
2
            }
76
2
        }
77
2
    }
78
0
    return nullptr;
79
2
}
80
81
76
inline int hex_value(char c) {
82
76
    if (c >= '0' && c <= '9') {
83
52
        return c - '0';
84
52
    }
85
24
    if (c >= 'a' && c <= 'f') {
86
17
        return c - 'a' + 10;
87
17
    }
88
7
    if (c >= 'A' && c <= 'F') {
89
6
        return c - 'A' + 10;
90
6
    }
91
1
    return -1;
92
7
}
93
94
5
inline Status decode_hex(std::string_view encoded, std::string* decoded) {
95
5
    DORIS_CHECK(decoded != nullptr);
96
5
    if ((encoded.size() & 1U) != 0) {
97
1
        return Status::InvalidArgument("Invalid odd-length Iceberg binary default");
98
1
    }
99
4
    decoded->resize(encoded.size() / 2);
100
41
    for (size_t index = 0; index < encoded.size(); index += 2) {
101
38
        const int high = hex_value(encoded[index]);
102
38
        const int low = hex_value(encoded[index + 1]);
103
38
        if (high < 0 || low < 0) {
104
1
            return Status::InvalidArgument("Invalid hexadecimal Iceberg binary default");
105
1
        }
106
37
        (*decoded)[index / 2] = static_cast<char>((high << 4) | low);
107
37
    }
108
3
    return Status::OK();
109
4
}
110
111
2
inline Status decode_json_binary(std::string_view encoded, std::string* decoded) {
112
2
    DORIS_CHECK(decoded != nullptr);
113
2
    const bool is_uuid = encoded.size() == 36 && encoded[8] == '-' && encoded[13] == '-' &&
114
2
                         encoded[18] == '-' && encoded[23] == '-';
115
2
    if (is_uuid) {
116
1
        std::string uuid_hex;
117
1
        uuid_hex.reserve(32);
118
37
        for (size_t index = 0; index < encoded.size(); ++index) {
119
36
            if (index != 8 && index != 13 && index != 18 && index != 23) {
120
32
                uuid_hex.push_back(encoded[index]);
121
32
            }
122
36
        }
123
1
        return decode_hex(uuid_hex, decoded);
124
1
    }
125
1
    return decode_hex(encoded, decoded);
126
2
}
127
128
5
inline std::string json_scalar_text(const rapidjson::Value& value) {
129
5
    if (value.IsString()) {
130
3
        return {value.GetString(), value.GetStringLength()};
131
3
    }
132
2
    rapidjson::StringBuffer buffer;
133
2
    rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
134
2
    value.Accept(writer);
135
2
    return {buffer.GetString(), buffer.GetSize()};
136
5
}
137
138
7
inline void normalize_timestamp_for_doris(PrimitiveType primitive_type, std::string* value) {
139
7
    if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 &&
140
7
        primitive_type != TYPE_TIMESTAMPTZ) {
141
3
        return;
142
3
    }
143
4
    if (const size_t separator = value->find('T'); separator != std::string::npos) {
144
3
        (*value)[separator] = ' ';
145
3
    }
146
4
    if (primitive_type == TYPE_TIMESTAMPTZ) {
147
1
        return;
148
1
    }
149
3
    if (value->ends_with('Z')) {
150
1
        value->pop_back();
151
1
        return;
152
1
    }
153
2
    const size_t time_start = value->find(' ');
154
2
    if (time_start == std::string::npos) {
155
1
        return;
156
1
    }
157
1
    const size_t offset = value->find_first_of("+-", time_start + 1);
158
1
    if (offset != std::string::npos) {
159
1
        value->erase(offset);
160
1
    }
161
1
}
162
163
inline Status make_null_field(const schema::external::TField& field, const DataTypePtr& data_type,
164
8
                              Field* result) {
165
8
    DORIS_CHECK(data_type != nullptr);
166
8
    DORIS_CHECK(result != nullptr);
167
8
    if (field.__isset.is_optional && !field.is_optional) {
168
1
        return Status::InvalidArgument("Required Iceberg field '{}' has a null default",
169
1
                                       field.name);
170
1
    }
171
7
    if (!data_type->is_nullable()) {
172
1
        return Status::InternalError(
173
1
                "Optional Iceberg field '{}' has a null default, but its Doris type '{}' is not "
174
1
                "nullable",
175
1
                field.name, data_type->get_name());
176
1
    }
177
6
    *result = Field();
178
6
    return Status::OK();
179
7
}
180
181
inline Status build_initial_default_field(const schema::external::TField& field,
182
                                          const DataTypePtr& data_type,
183
                                          std::deque<std::string>* binary_storage, Field* result);
184
185
inline Status build_json_default_field(const schema::external::TField& field,
186
                                       const DataTypePtr& data_type,
187
                                       const rapidjson::Value& json_value,
188
                                       std::deque<std::string>* binary_storage, Field* result);
189
190
inline Status build_json_struct_default(const schema::external::TField& field,
191
                                        const DataTypePtr& value_type,
192
                                        const rapidjson::Value& json_value,
193
5
                                        std::deque<std::string>* binary_storage, Field* result) {
194
5
    if (!json_value.IsObject() || !field.__isset.nestedField ||
195
5
        !field.nestedField.__isset.struct_field || !field.nestedField.struct_field.__isset.fields) {
196
0
        return Status::InvalidArgument("Invalid Iceberg struct default for field '{}'", field.name);
197
0
    }
198
199
5
    const auto& struct_type = assert_cast<const DataTypeStruct&>(*value_type);
200
5
    Struct struct_value;
201
5
    struct_value.reserve(struct_type.get_elements().size());
202
13
    for (size_t index = 0; index < struct_type.get_elements().size(); ++index) {
203
8
        const auto& child_name = struct_type.get_element_name(index);
204
8
        const auto* child = find_struct_child(field.nestedField.struct_field, child_name);
205
8
        if (child == nullptr || !child->__isset.id) {
206
0
            return Status::InvalidArgument(
207
0
                    "Iceberg struct default for field '{}' is missing metadata for projected "
208
0
                    "child '{}'",
209
0
                    field.name, child_name);
210
0
        }
211
212
8
        const std::string child_id = std::to_string(child->id);
213
8
        const auto member = json_value.FindMember(child_id.c_str());
214
8
        Field child_value;
215
8
        if (member == json_value.MemberEnd()) {
216
6
            RETURN_IF_ERROR(build_initial_default_field(*child, struct_type.get_element(index),
217
6
                                                        binary_storage, &child_value));
218
6
        } else {
219
2
            RETURN_IF_ERROR(build_json_default_field(*child, struct_type.get_element(index),
220
2
                                                     member->value, binary_storage, &child_value));
221
2
        }
222
8
        struct_value.push_back(std::move(child_value));
223
8
    }
224
5
    *result = Field::create_field<TYPE_STRUCT>(std::move(struct_value));
225
5
    return Status::OK();
226
5
}
227
228
// The recursive item TField describes the element schema and its field-level default metadata. It
229
// cannot represent a particular list literal's length or per-position values, so the parent
230
// initial-default keeps those values in Iceberg's single-value JSON array.
231
inline Status build_json_array_default(const schema::external::TField& field,
232
                                       const DataTypePtr& value_type,
233
                                       const rapidjson::Value& json_value,
234
1
                                       std::deque<std::string>* binary_storage, Field* result) {
235
1
    if (!json_value.IsArray() || !field.__isset.nestedField ||
236
1
        !field.nestedField.__isset.array_field ||
237
1
        !field.nestedField.array_field.__isset.item_field) {
238
0
        return Status::InvalidArgument("Invalid Iceberg list default for field '{}'", field.name);
239
0
    }
240
1
    const auto* element = get_field_ptr(field.nestedField.array_field.item_field);
241
1
    if (element == nullptr) {
242
0
        return Status::InvalidArgument(
243
0
                "Iceberg list default for field '{}' has incomplete element metadata", field.name);
244
0
    }
245
246
1
    const auto& array_type = assert_cast<const DataTypeArray&>(*value_type);
247
1
    Array array_value;
248
1
    array_value.reserve(json_value.Size());
249
2
    for (const auto& json_element : json_value.GetArray()) {
250
2
        Field element_value;
251
2
        RETURN_IF_ERROR(build_json_default_field(*element, array_type.get_nested_type(),
252
2
                                                 json_element, binary_storage, &element_value));
253
2
        array_value.push_back(std::move(element_value));
254
2
    }
255
1
    *result = Field::create_field<TYPE_ARRAY>(std::move(array_value));
256
1
    return Status::OK();
257
1
}
258
259
// The recursive key/value TFields describe entry schemas and field-level default metadata. They
260
// cannot represent the number, order, or concrete values of map entries, so the parent
261
// initial-default keeps the entries in Iceberg's single-value JSON key/value arrays.
262
inline Status build_json_map_default(const schema::external::TField& field,
263
                                     const DataTypePtr& value_type,
264
                                     const rapidjson::Value& json_value,
265
1
                                     std::deque<std::string>* binary_storage, Field* result) {
266
1
    if (!json_value.IsObject() || !json_value.HasMember("keys") || !json_value["keys"].IsArray() ||
267
1
        !json_value.HasMember("values") || !json_value["values"].IsArray() ||
268
1
        !field.__isset.nestedField || !field.nestedField.__isset.map_field ||
269
1
        !field.nestedField.map_field.__isset.key_field ||
270
1
        !field.nestedField.map_field.__isset.value_field) {
271
0
        return Status::InvalidArgument("Invalid Iceberg map default for field '{}'", field.name);
272
0
    }
273
1
    const auto& keys = json_value["keys"];
274
1
    const auto& values = json_value["values"];
275
1
    if (keys.Size() != values.Size()) {
276
0
        return Status::InvalidArgument(
277
0
                "Iceberg map default for field '{}' has {} keys but {} values", field.name,
278
0
                keys.Size(), values.Size());
279
0
    }
280
281
1
    const auto* key = get_field_ptr(field.nestedField.map_field.key_field);
282
1
    const auto* value = get_field_ptr(field.nestedField.map_field.value_field);
283
1
    if (key == nullptr || value == nullptr) {
284
0
        return Status::InvalidArgument(
285
0
                "Iceberg map default for field '{}' has incomplete key/value metadata", field.name);
286
0
    }
287
288
1
    const auto& map_type = assert_cast<const DataTypeMap&>(*value_type);
289
1
    Array key_fields;
290
1
    Array value_fields;
291
1
    key_fields.reserve(keys.Size());
292
1
    value_fields.reserve(values.Size());
293
2
    for (rapidjson::SizeType index = 0; index < keys.Size(); ++index) {
294
1
        Field key_value;
295
1
        Field mapped_value;
296
1
        RETURN_IF_ERROR(build_json_default_field(*key, map_type.get_key_type(), keys[index],
297
1
                                                 binary_storage, &key_value));
298
1
        RETURN_IF_ERROR(build_json_default_field(*value, map_type.get_value_type(), values[index],
299
1
                                                 binary_storage, &mapped_value));
300
1
        key_fields.push_back(std::move(key_value));
301
1
        value_fields.push_back(std::move(mapped_value));
302
1
    }
303
1
    Map map_value;
304
1
    map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(key_fields)));
305
1
    map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(value_fields)));
306
1
    *result = Field::create_field<TYPE_MAP>(std::move(map_value));
307
1
    return Status::OK();
308
1
}
309
310
inline Status build_json_scalar_default(const schema::external::TField& field,
311
                                        const DataTypePtr& value_type,
312
                                        const rapidjson::Value& json_value,
313
5
                                        std::deque<std::string>* binary_storage, Field* result) {
314
5
    const auto primitive_type = value_type->get_primitive_type();
315
5
    std::string serialized_value = json_scalar_text(json_value);
316
5
    const bool binary_like = (field.__isset.initial_default_value_is_base64 &&
317
5
                              field.initial_default_value_is_base64) ||
318
5
                             primitive_type == TYPE_VARBINARY;
319
5
    if (binary_like) {
320
1
        if (!json_value.IsString()) {
321
0
            return Status::InvalidArgument(
322
0
                    "Iceberg binary default for field '{}' is not a JSON string", field.name);
323
0
        }
324
1
        binary_storage->emplace_back();
325
1
        RETURN_IF_ERROR(decode_json_binary(serialized_value, &binary_storage->back()));
326
1
        if (primitive_type == TYPE_VARBINARY) {
327
0
            *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back()));
328
1
        } else if (is_string_type(primitive_type)) {
329
1
            *result = Field::create_field<TYPE_STRING>(binary_storage->back());
330
1
        } else {
331
0
            return Status::InvalidArgument(
332
0
                    "Iceberg binary default for field '{}' has incompatible Doris type '{}'",
333
0
                    field.name, value_type->get_name());
334
0
        }
335
1
        return Status::OK();
336
1
    }
337
338
4
    if (is_string_type(primitive_type)) {
339
2
        if (!json_value.IsString()) {
340
0
            return Status::InvalidArgument("Iceberg string default for field '{}' is not a string",
341
0
                                           field.name);
342
0
        }
343
2
        *result = Field::create_field<TYPE_STRING>(std::move(serialized_value));
344
2
        return Status::OK();
345
2
    }
346
2
    normalize_timestamp_for_doris(primitive_type, &serialized_value);
347
2
    RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result));
348
2
    return Status::OK();
349
2
}
350
351
inline Status build_json_default_field(const schema::external::TField& field,
352
                                       const DataTypePtr& data_type,
353
                                       const rapidjson::Value& json_value,
354
13
                                       std::deque<std::string>* binary_storage, Field* result) {
355
13
    DORIS_CHECK(data_type != nullptr);
356
13
    DORIS_CHECK(binary_storage != nullptr);
357
13
    DORIS_CHECK(result != nullptr);
358
13
    if (json_value.IsNull()) {
359
1
        return make_null_field(field, data_type, result);
360
1
    }
361
362
12
    const auto value_type = remove_nullable(data_type);
363
12
    switch (value_type->get_primitive_type()) {
364
5
    case TYPE_STRUCT:
365
5
        return build_json_struct_default(field, value_type, json_value, binary_storage, result);
366
1
    case TYPE_ARRAY:
367
1
        return build_json_array_default(field, value_type, json_value, binary_storage, result);
368
1
    case TYPE_MAP:
369
1
        return build_json_map_default(field, value_type, json_value, binary_storage, result);
370
5
    default:
371
5
        return build_json_scalar_default(field, value_type, json_value, binary_storage, result);
372
12
    }
373
12
}
374
375
inline Status build_initial_default_field(const schema::external::TField& field,
376
                                          const DataTypePtr& data_type,
377
40
                                          std::deque<std::string>* binary_storage, Field* result) {
378
40
    DORIS_CHECK(data_type != nullptr);
379
40
    DORIS_CHECK(binary_storage != nullptr);
380
40
    DORIS_CHECK(result != nullptr);
381
40
    if (!field.__isset.initial_default_value) {
382
5
        if (field.__isset.is_optional && !field.is_optional) {
383
1
            return Status::InvalidArgument(
384
1
                    "Required Iceberg field '{}' is missing from the data file and has no initial "
385
1
                    "default",
386
1
                    field.name);
387
1
        }
388
4
        return make_null_field(field, data_type, result);
389
5
    }
390
391
35
    const auto value_type = remove_nullable(data_type);
392
35
    const auto primitive_type = value_type->get_primitive_type();
393
35
    if (is_complex_type(primitive_type)) {
394
7
        rapidjson::Document document;
395
7
        document.Parse(field.initial_default_value.data(), field.initial_default_value.size());
396
7
        if (document.HasParseError()) {
397
0
            return Status::InvalidArgument("Invalid Iceberg JSON initial default for field '{}'",
398
0
                                           field.name);
399
0
        }
400
7
        return build_json_default_field(field, data_type, document, binary_storage, result);
401
7
    }
402
403
28
    const bool default_is_base64 = (field.__isset.initial_default_value_is_base64 &&
404
28
                                    field.initial_default_value_is_base64) ||
405
28
                                   primitive_type == TYPE_VARBINARY;
406
28
    if (default_is_base64) {
407
3
        binary_storage->emplace_back();
408
3
        if (!base64_decode(field.initial_default_value, &binary_storage->back())) {
409
0
            return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field '{}'",
410
0
                                           field.name);
411
0
        }
412
3
        if (primitive_type == TYPE_VARBINARY) {
413
2
            *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back()));
414
2
        } else if (is_string_type(primitive_type)) {
415
1
            *result = Field::create_field<TYPE_STRING>(binary_storage->back());
416
1
        } else {
417
0
            return Status::InvalidArgument(
418
0
                    "Iceberg field '{}' marks its initial default as Base64, but Doris type '{}' "
419
0
                    "cannot contain binary data",
420
0
                    field.name, value_type->get_name());
421
0
        }
422
3
        return Status::OK();
423
3
    }
424
425
25
    RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(field.initial_default_value, *result));
426
25
    return Status::OK();
427
25
}
428
429
} // namespace detail
430
431
// Builds an owned one-row column for an Iceberg field that is absent from an old data file.
432
// Complex values follow Iceberg's JSON single-value encoding. Struct members omitted from the
433
// encoded value are recursively populated from the child field's own initial default.
434
inline Status create_initial_default_column(const schema::external::TField& field,
435
34
                                            const DataTypePtr& data_type, ColumnPtr* result) {
436
34
    DORIS_CHECK(data_type != nullptr);
437
34
    DORIS_CHECK(result != nullptr);
438
439
34
    auto column = data_type->create_column();
440
34
    std::deque<std::string> binary_storage;
441
34
    Field value;
442
34
    RETURN_IF_ERROR(detail::build_initial_default_field(field, data_type, &binary_storage, &value));
443
    // The column copies every String/StringView leaf before binary_storage is destroyed.
444
33
    column->insert(value);
445
446
33
    *result = std::move(column);
447
33
    return Status::OK();
448
34
}
449
450
16
inline ColumnPtr repeat_initial_default_column(const ColumnPtr& default_column, size_t rows) {
451
16
    DORIS_CHECK(default_column);
452
16
    DORIS_CHECK_EQ(default_column->size(), 1);
453
454
16
    auto repeated_column = default_column->clone_empty();
455
16
    repeated_column->insert_many_from(*default_column, 0, rows);
456
16
    return repeated_column;
457
16
}
458
459
inline Status append_initial_default(
460
        const schema::external::TField& field, const DataTypePtr& data_type, size_t rows,
461
        std::unordered_map<int32_t, std::pair<DataTypePtr, ColumnPtr>>* prepared_values,
462
9
        ColumnPtr* destination) {
463
9
    DORIS_CHECK(data_type != nullptr);
464
9
    DORIS_CHECK(prepared_values != nullptr);
465
9
    DORIS_CHECK(destination != nullptr);
466
9
    DORIS_CHECK(field.__isset.id);
467
468
9
    auto prepared_value = prepared_values->find(field.id);
469
9
    if (prepared_value == prepared_values->end()) {
470
8
        ColumnPtr default_column;
471
8
        RETURN_IF_ERROR(create_initial_default_column(field, data_type, &default_column));
472
8
        prepared_value =
473
8
                prepared_values
474
8
                        ->emplace(field.id, std::make_pair(data_type, std::move(default_column)))
475
8
                        .first;
476
8
    } else {
477
        // One Iceberg field ID resolves to one query type. Hold the first DataTypePtr so equivalent
478
        // complex types reconstructed for later Blocks reuse the same prepared value.
479
1
        DORIS_CHECK(prepared_value->second.first->equals(*data_type));
480
1
    }
481
482
9
    auto mutable_destination = IColumn::mutate(std::move(*destination));
483
9
    mutable_destination->insert_many_from(*prepared_value->second.second, 0, rows);
484
9
    *destination = std::move(mutable_destination);
485
9
    return Status::OK();
486
9
}
487
488
} // namespace doris::iceberg