Coverage Report

Created: 2026-07-07 05:17

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/parquet/schema_desc.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 "format/parquet/schema_desc.h"
19
20
#include <ctype.h>
21
22
#include <algorithm>
23
#include <ostream>
24
#include <utility>
25
26
#include "common/cast_set.h"
27
#include "common/logging.h"
28
#include "core/data_type/data_type_array.h"
29
#include "core/data_type/data_type_factory.hpp"
30
#include "core/data_type/data_type_map.h"
31
#include "core/data_type/data_type_struct.h"
32
#include "core/data_type/define_primitive_type.h"
33
#include "format/generic_reader.h"
34
#include "format/table/table_schema_change_helper.h"
35
#include "util/slice.h"
36
#include "util/string_util.h"
37
38
namespace doris {
39
40
29.2k
static bool is_group_node(const tparquet::SchemaElement& schema) {
41
29.2k
    return schema.num_children > 0;
42
29.2k
}
43
44
2.82k
static bool is_list_node(const tparquet::SchemaElement& schema) {
45
2.82k
    return schema.__isset.converted_type && schema.converted_type == tparquet::ConvertedType::LIST;
46
2.82k
}
47
48
3.67k
static bool is_map_node(const tparquet::SchemaElement& schema) {
49
3.67k
    return schema.__isset.converted_type &&
50
3.67k
           (schema.converted_type == tparquet::ConvertedType::MAP ||
51
2.22k
            schema.converted_type == tparquet::ConvertedType::MAP_KEY_VALUE);
52
3.67k
}
53
54
25.0k
static bool is_repeated_node(const tparquet::SchemaElement& schema) {
55
25.0k
    return schema.__isset.repetition_type &&
56
25.0k
           schema.repetition_type == tparquet::FieldRepetitionType::REPEATED;
57
25.0k
}
58
59
857
static bool is_required_node(const tparquet::SchemaElement& schema) {
60
857
    return schema.__isset.repetition_type &&
61
857
           schema.repetition_type == tparquet::FieldRepetitionType::REQUIRED;
62
857
}
63
64
25.5k
static bool is_optional_node(const tparquet::SchemaElement& schema) {
65
25.5k
    return schema.__isset.repetition_type &&
66
25.5k
           schema.repetition_type == tparquet::FieldRepetitionType::OPTIONAL;
67
25.5k
}
68
69
1.36k
static int num_children_node(const tparquet::SchemaElement& schema) {
70
1.36k
    return schema.__isset.num_children ? schema.num_children : 0;
71
1.36k
}
72
73
/**
74
 * `repeated_parent_def_level` is the definition level of the first ancestor node whose repetition_type equals REPEATED.
75
 * Empty array/map values are not stored in doris columns, so have to use `repeated_parent_def_level` to skip the
76
 * empty or null values in ancestor node.
77
 *
78
 * For instance, considering an array of strings with 3 rows like the following:
79
 * null, [], [a, b, c]
80
 * We can store four elements in data column: null, a, b, c
81
 * and the offsets column is: 1, 1, 4
82
 * and the null map is: 1, 0, 0
83
 * For the i-th row in array column: range from `offsets[i - 1]` until `offsets[i]` represents the elements in this row,
84
 * so we can't store empty array/map values in doris data column.
85
 * As a comparison, spark does not require `repeated_parent_def_level`,
86
 * because the spark column stores empty array/map values , and use anther length column to indicate empty values.
87
 * Please reference: https://github.com/apache/spark/blob/master/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetColumnVector.java
88
 *
89
 * Furthermore, we can also avoid store null array/map values in doris data column.
90
 * The same three rows as above, We can only store three elements in data column: a, b, c
91
 * and the offsets column is: 0, 0, 3
92
 * and the null map is: 1, 0, 0
93
 *
94
 * Inherit the repetition and definition level from parent node, if the parent node is repeated,
95
 * we should set repeated_parent_def_level = definition_level, otherwise as repeated_parent_def_level.
96
 * @param parent parent node
97
 * @param repeated_parent_def_level the first ancestor node whose repetition_type equals REPEATED
98
 */
99
3.73k
static void set_child_node_level(FieldSchema* parent, int16_t repeated_parent_def_level) {
100
8.67k
    for (auto& child : parent->children) {
101
8.67k
        child.repetition_level = parent->repetition_level;
102
8.67k
        child.definition_level = parent->definition_level;
103
8.67k
        child.repeated_parent_def_level = repeated_parent_def_level;
104
8.67k
    }
105
3.73k
}
106
107
1.33k
static bool is_struct_list_node(const tparquet::SchemaElement& schema) {
108
1.33k
    const std::string& name = schema.name;
109
1.33k
    static const Slice array_slice("array", 5);
110
1.33k
    static const Slice tuple_slice("_tuple", 6);
111
1.33k
    Slice slice(name);
112
1.33k
    return slice == array_slice || slice.ends_with(tuple_slice);
113
1.33k
}
114
115
0
std::string FieldSchema::debug_string() const {
116
0
    std::stringstream ss;
117
0
    ss << "FieldSchema(name=" << name << ", R=" << repetition_level << ", D=" << definition_level;
118
0
    if (children.size() > 0) {
119
0
        ss << ", type=" << data_type->get_name() << ", children=[";
120
0
        for (int i = 0; i < children.size(); ++i) {
121
0
            if (i != 0) {
122
0
                ss << ", ";
123
0
            }
124
0
            ss << children[i].debug_string();
125
0
        }
126
0
        ss << "]";
127
0
    } else {
128
0
        ss << ", physical_type=" << physical_type;
129
0
        ss << " , doris_type=" << data_type->get_name();
130
0
    }
131
0
    ss << ")";
132
0
    return ss.str();
133
0
}
134
135
2.84k
Status FieldDescriptor::parse_from_thrift(const std::vector<tparquet::SchemaElement>& t_schemas) {
136
2.84k
    if (t_schemas.size() == 0 || !is_group_node(t_schemas[0])) {
137
0
        return Status::InvalidArgument("Wrong parquet root schema element");
138
0
    }
139
2.84k
    const auto& root_schema = t_schemas[0];
140
2.84k
    _fields.resize(root_schema.num_children);
141
2.84k
    _next_schema_pos = 1;
142
143
19.7k
    for (int i = 0; i < root_schema.num_children; ++i) {
144
16.9k
        RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &_fields[i]));
145
16.9k
        if (_name_to_field.find(_fields[i].name) != _name_to_field.end()) {
146
0
            return Status::InvalidArgument("Duplicated field name: {}", _fields[i].name);
147
0
        }
148
16.9k
        _name_to_field.emplace(_fields[i].name, &_fields[i]);
149
16.9k
    }
150
151
2.84k
    if (_next_schema_pos != t_schemas.size()) {
152
0
        return Status::InvalidArgument("Remaining {} unparsed schema elements",
153
0
                                       t_schemas.size() - _next_schema_pos);
154
0
    }
155
156
2.84k
    return Status::OK();
157
2.84k
}
158
159
Status FieldDescriptor::parse_node_field(const std::vector<tparquet::SchemaElement>& t_schemas,
160
25.5k
                                         size_t curr_pos, FieldSchema* node_field) {
161
25.5k
    if (curr_pos >= t_schemas.size()) {
162
0
        return Status::InvalidArgument("Out-of-bounds index of schema elements");
163
0
    }
164
25.5k
    auto& t_schema = t_schemas[curr_pos];
165
25.5k
    if (is_group_node(t_schema)) {
166
        // nested structure or nullable list
167
3.67k
        return parse_group_field(t_schemas, curr_pos, node_field);
168
3.67k
    }
169
21.8k
    if (is_repeated_node(t_schema)) {
170
        // repeated <primitive-type> <name> (LIST)
171
        // produce required list<element>
172
18
        node_field->repetition_level++;
173
18
        node_field->definition_level++;
174
18
        node_field->children.resize(1);
175
18
        set_child_node_level(node_field, node_field->definition_level);
176
18
        auto child = &node_field->children[0];
177
18
        parse_physical_field(t_schema, false, child);
178
179
18
        node_field->name = t_schema.name;
180
18
        node_field->lower_case_name = to_lower(t_schema.name);
181
18
        node_field->data_type = std::make_shared<DataTypeArray>(make_nullable(child->data_type));
182
18
        _next_schema_pos = curr_pos + 1;
183
18
        node_field->field_id = t_schema.__isset.field_id ? t_schema.field_id : -1;
184
21.8k
    } else {
185
21.8k
        bool is_optional = is_optional_node(t_schema);
186
21.8k
        if (is_optional) {
187
18.6k
            node_field->definition_level++;
188
18.6k
        }
189
21.8k
        parse_physical_field(t_schema, is_optional, node_field);
190
21.8k
        _next_schema_pos = curr_pos + 1;
191
21.8k
    }
192
21.8k
    return Status::OK();
193
25.5k
}
194
195
void FieldDescriptor::parse_physical_field(const tparquet::SchemaElement& physical_schema,
196
21.8k
                                           bool is_nullable, FieldSchema* physical_field) {
197
21.8k
    physical_field->name = physical_schema.name;
198
21.8k
    physical_field->lower_case_name = to_lower(physical_field->name);
199
21.8k
    physical_field->parquet_schema = physical_schema;
200
21.8k
    physical_field->physical_type = physical_schema.type;
201
21.8k
    physical_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
202
21.8k
    _physical_fields.push_back(physical_field);
203
21.8k
    physical_field->physical_column_index = cast_set<int>(_physical_fields.size() - 1);
204
21.8k
    auto type = get_doris_type(physical_schema, is_nullable);
205
21.8k
    physical_field->data_type = type.first;
206
21.8k
    physical_field->is_type_compatibility = type.second;
207
21.8k
    physical_field->field_id = physical_schema.__isset.field_id ? physical_schema.field_id : -1;
208
21.8k
}
209
210
std::pair<DataTypePtr, bool> FieldDescriptor::get_doris_type(
211
21.8k
        const tparquet::SchemaElement& physical_schema, bool nullable) {
212
21.8k
    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};
213
21.8k
    try {
214
21.8k
        if (physical_schema.__isset.logicalType) {
215
7.12k
            ans = convert_to_doris_type(physical_schema.logicalType, nullable);
216
14.7k
        } else if (physical_schema.__isset.converted_type) {
217
4.64k
            ans = convert_to_doris_type(physical_schema, nullable);
218
4.64k
        }
219
21.8k
    } catch (...) {
220
        // now the Not supported exception are ignored
221
        // so those byte_array maybe be treated as varbinary(now) : string(before)
222
28
    }
223
21.8k
    if (ans.first->get_primitive_type() == PrimitiveType::INVALID_TYPE) {
224
10.1k
        switch (physical_schema.type) {
225
839
        case tparquet::Type::BOOLEAN:
226
839
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_BOOLEAN, nullable);
227
839
            break;
228
3.53k
        case tparquet::Type::INT32:
229
3.53k
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
230
3.53k
            break;
231
2.50k
        case tparquet::Type::INT64:
232
2.50k
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
233
2.50k
            break;
234
541
        case tparquet::Type::INT96:
235
541
            if (_enable_mapping_timestamp_tz) {
236
                // treat INT96 as TIMESTAMPTZ
237
24
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMESTAMPTZ, nullable,
238
24
                                                                         0, 6);
239
517
            } else {
240
                // in most cases, it's a nano timestamp
241
517
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable,
242
517
                                                                         0, 6);
243
517
            }
244
541
            break;
245
567
        case tparquet::Type::FLOAT:
246
567
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);
247
567
            break;
248
1.91k
        case tparquet::Type::DOUBLE:
249
1.91k
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_DOUBLE, nullable);
250
1.91k
            break;
251
237
        case tparquet::Type::BYTE_ARRAY:
252
237
            if (_enable_mapping_varbinary) {
253
                // if physical_schema not set logicalType and converted_type,
254
                // we treat BYTE_ARRAY as VARBINARY by default, so that we can read all data directly.
255
34
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable);
256
203
            } else {
257
203
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
258
203
            }
259
237
            break;
260
14
        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
261
14
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
262
14
            break;
263
0
        default:
264
0
            throw Exception(Status::InternalError("Not supported parquet logicalType{}",
265
0
                                                  physical_schema.type));
266
0
            break;
267
10.1k
        }
268
10.1k
    }
269
21.8k
    return ans;
270
21.8k
}
271
272
std::pair<DataTypePtr, bool> FieldDescriptor::convert_to_doris_type(
273
7.12k
        tparquet::LogicalType logicalType, bool nullable) {
274
7.12k
    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};
275
7.12k
    bool& is_type_compatibility = ans.second;
276
7.12k
    if (logicalType.__isset.STRING) {
277
4.28k
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
278
4.28k
    } else if (logicalType.__isset.DECIMAL) {
279
781
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DECIMAL128I, nullable,
280
781
                                                                 logicalType.DECIMAL.precision,
281
781
                                                                 logicalType.DECIMAL.scale);
282
2.05k
    } else if (logicalType.__isset.DATE) {
283
389
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2, nullable);
284
1.66k
    } else if (logicalType.__isset.INTEGER) {
285
1.30k
        if (logicalType.INTEGER.isSigned) {
286
1.09k
            if (logicalType.INTEGER.bitWidth <= 8) {
287
316
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);
288
774
            } else if (logicalType.INTEGER.bitWidth <= 16) {
289
542
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
290
542
            } else if (logicalType.INTEGER.bitWidth <= 32) {
291
108
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
292
124
            } else {
293
124
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
294
124
            }
295
1.09k
        } else {
296
210
            is_type_compatibility = true;
297
210
            if (logicalType.INTEGER.bitWidth <= 8) {
298
52
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
299
158
            } else if (logicalType.INTEGER.bitWidth <= 16) {
300
52
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
301
106
            } else if (logicalType.INTEGER.bitWidth <= 32) {
302
54
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
303
54
            } else {
304
52
                ans.first = DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
305
52
            }
306
210
        }
307
1.30k
    } else if (logicalType.__isset.TIME) {
308
20
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2, nullable);
309
347
    } else if (logicalType.__isset.TIMESTAMP) {
310
305
        if (_enable_mapping_timestamp_tz) {
311
4
            if (logicalType.TIMESTAMP.isAdjustedToUTC) {
312
                // treat TIMESTAMP with isAdjustedToUTC as TIMESTAMPTZ
313
4
                ans.first = DataTypeFactory::instance().create_data_type(
314
4
                        TYPE_TIMESTAMPTZ, nullable, 0,
315
4
                        logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
316
4
                return ans;
317
4
            }
318
4
        }
319
301
        ans.first = DataTypeFactory::instance().create_data_type(
320
301
                TYPE_DATETIMEV2, nullable, 0, logicalType.TIMESTAMP.unit.__isset.MILLIS ? 3 : 6);
321
301
    } else if (logicalType.__isset.JSON) {
322
2
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
323
40
    } else if (logicalType.__isset.UUID) {
324
10
        if (_enable_mapping_varbinary) {
325
7
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_VARBINARY, nullable, -1,
326
7
                                                                     -1, 16);
327
7
        } else {
328
3
            ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
329
3
        }
330
30
    } else if (logicalType.__isset.FLOAT16) {
331
14
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_FLOAT, nullable);
332
16
    } else {
333
16
        throw Exception(Status::InternalError("Not supported parquet logicalType"));
334
16
    }
335
7.10k
    return ans;
336
7.12k
}
337
338
std::pair<DataTypePtr, bool> FieldDescriptor::convert_to_doris_type(
339
4.64k
        const tparquet::SchemaElement& physical_schema, bool nullable) {
340
4.64k
    std::pair<DataTypePtr, bool> ans = {std::make_shared<DataTypeNothing>(), false};
341
4.64k
    bool& is_type_compatibility = ans.second;
342
4.64k
    switch (physical_schema.converted_type) {
343
2.91k
    case tparquet::ConvertedType::type::UTF8:
344
2.91k
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
345
2.91k
        break;
346
346
    case tparquet::ConvertedType::type::DECIMAL:
347
346
        ans.first = DataTypeFactory::instance().create_data_type(
348
346
                TYPE_DECIMAL128I, nullable, physical_schema.precision, physical_schema.scale);
349
346
        break;
350
458
    case tparquet::ConvertedType::type::DATE:
351
458
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATEV2, nullable);
352
458
        break;
353
0
    case tparquet::ConvertedType::type::TIME_MILLIS:
354
0
        [[fallthrough]];
355
0
    case tparquet::ConvertedType::type::TIME_MICROS:
356
0
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TIMEV2, nullable);
357
0
        break;
358
12
    case tparquet::ConvertedType::type::TIMESTAMP_MILLIS:
359
12
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 3);
360
12
        break;
361
48
    case tparquet::ConvertedType::type::TIMESTAMP_MICROS:
362
48
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_DATETIMEV2, nullable, 0, 6);
363
48
        break;
364
164
    case tparquet::ConvertedType::type::INT_8:
365
164
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_TINYINT, nullable);
366
164
        break;
367
4
    case tparquet::ConvertedType::type::UINT_8:
368
4
        is_type_compatibility = true;
369
4
        [[fallthrough]];
370
173
    case tparquet::ConvertedType::type::INT_16:
371
173
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_SMALLINT, nullable);
372
173
        break;
373
4
    case tparquet::ConvertedType::type::UINT_16:
374
4
        is_type_compatibility = true;
375
4
        [[fallthrough]];
376
24
    case tparquet::ConvertedType::type::INT_32:
377
24
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_INT, nullable);
378
24
        break;
379
14
    case tparquet::ConvertedType::type::UINT_32:
380
14
        is_type_compatibility = true;
381
14
        [[fallthrough]];
382
340
    case tparquet::ConvertedType::type::INT_64:
383
340
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_BIGINT, nullable);
384
340
        break;
385
146
    case tparquet::ConvertedType::type::UINT_64:
386
146
        is_type_compatibility = true;
387
146
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_LARGEINT, nullable);
388
146
        break;
389
2
    case tparquet::ConvertedType::type::JSON:
390
2
        ans.first = DataTypeFactory::instance().create_data_type(TYPE_STRING, nullable);
391
2
        break;
392
12
    default:
393
12
        throw Exception(Status::InternalError("Not supported parquet ConvertedType: {}",
394
12
                                              physical_schema.converted_type));
395
4.64k
    }
396
4.62k
    return ans;
397
4.64k
}
398
399
Status FieldDescriptor::parse_group_field(const std::vector<tparquet::SchemaElement>& t_schemas,
400
3.67k
                                          size_t curr_pos, FieldSchema* group_field) {
401
3.67k
    auto& group_schema = t_schemas[curr_pos];
402
3.67k
    if (is_map_node(group_schema)) {
403
        // the map definition:
404
        // optional group <name> (MAP) {
405
        //   repeated group map (MAP_KEY_VALUE) {
406
        //     required <type> key;
407
        //     optional <type> value;
408
        //   }
409
        // }
410
859
        return parse_map_field(t_schemas, curr_pos, group_field);
411
859
    }
412
2.82k
    if (is_list_node(group_schema)) {
413
        // the list definition:
414
        // optional group <name> (LIST) {
415
        //   repeated group [bag | list] { // hive or spark
416
        //     optional <type> [array_element | element]; // hive or spark
417
        //   }
418
        // }
419
1.36k
        return parse_list_field(t_schemas, curr_pos, group_field);
420
1.36k
    }
421
422
1.45k
    if (is_repeated_node(group_schema)) {
423
18
        group_field->repetition_level++;
424
18
        group_field->definition_level++;
425
18
        group_field->children.resize(1);
426
18
        set_child_node_level(group_field, group_field->definition_level);
427
18
        auto struct_field = &group_field->children[0];
428
        // the list of struct:
429
        // repeated group <name> (LIST) {
430
        //   optional/required <type> <name>;
431
        //   ...
432
        // }
433
        // produce a non-null list<struct>
434
18
        RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, struct_field));
435
436
18
        group_field->name = group_schema.name;
437
18
        group_field->lower_case_name = to_lower(group_field->name);
438
18
        group_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
439
18
        group_field->data_type =
440
18
                std::make_shared<DataTypeArray>(make_nullable(struct_field->data_type));
441
18
        group_field->field_id = group_schema.__isset.field_id ? group_schema.field_id : -1;
442
1.43k
    } else {
443
1.43k
        RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos, group_field));
444
1.43k
    }
445
446
1.45k
    return Status::OK();
447
1.45k
}
448
449
Status FieldDescriptor::parse_list_field(const std::vector<tparquet::SchemaElement>& t_schemas,
450
1.36k
                                         size_t curr_pos, FieldSchema* list_field) {
451
    // the list definition:
452
    // spark and hive have three level schemas but with different schema name
453
    // spark: <column-name> - "list" - "element"
454
    // hive: <column-name> - "bag" - "array_element"
455
    // parse three level schemas to two level primitive like: LIST<INT>,
456
    // or nested structure like: LIST<MAP<INT, INT>>
457
1.36k
    auto& first_level = t_schemas[curr_pos];
458
1.36k
    if (first_level.num_children != 1) {
459
0
        return Status::InvalidArgument("List element should have only one child");
460
0
    }
461
462
1.36k
    if (curr_pos + 1 >= t_schemas.size()) {
463
0
        return Status::InvalidArgument("List element should have the second level schema");
464
0
    }
465
466
1.36k
    if (first_level.repetition_type == tparquet::FieldRepetitionType::REPEATED) {
467
0
        return Status::InvalidArgument("List element can't be a repeated schema");
468
0
    }
469
470
    // the repeated schema element
471
1.36k
    auto& second_level = t_schemas[curr_pos + 1];
472
1.36k
    if (second_level.repetition_type != tparquet::FieldRepetitionType::REPEATED) {
473
0
        return Status::InvalidArgument("The second level of list element should be repeated");
474
0
    }
475
476
    // This indicates if this list is nullable.
477
1.36k
    bool is_optional = is_optional_node(first_level);
478
1.36k
    if (is_optional) {
479
1.22k
        list_field->definition_level++;
480
1.22k
    }
481
1.36k
    list_field->repetition_level++;
482
1.36k
    list_field->definition_level++;
483
1.36k
    list_field->children.resize(1);
484
1.36k
    FieldSchema* list_child = &list_field->children[0];
485
486
1.36k
    size_t num_children = num_children_node(second_level);
487
1.36k
    if (num_children > 0) {
488
1.34k
        if (num_children == 1 && !is_struct_list_node(second_level)) {
489
            // optional field, and the third level element is the nested structure in list
490
            // produce nested structure like: LIST<INT>, LIST<MAP>, LIST<LIST<...>>
491
            // skip bag/list, it's a repeated element.
492
1.32k
            set_child_node_level(list_field, list_field->definition_level);
493
1.32k
            RETURN_IF_ERROR(parse_node_field(t_schemas, curr_pos + 2, list_child));
494
1.32k
        } else {
495
            // required field, produce the list of struct
496
18
            set_child_node_level(list_field, list_field->definition_level);
497
18
            RETURN_IF_ERROR(parse_struct_field(t_schemas, curr_pos + 1, list_child));
498
18
        }
499
1.34k
    } else if (num_children == 0) {
500
        // required two level list, for compatibility reason.
501
24
        set_child_node_level(list_field, list_field->definition_level);
502
24
        parse_physical_field(second_level, false, list_child);
503
24
        _next_schema_pos = curr_pos + 2;
504
24
    }
505
506
1.36k
    list_field->name = first_level.name;
507
1.36k
    list_field->lower_case_name = to_lower(first_level.name);
508
1.36k
    list_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
509
1.36k
    list_field->data_type =
510
1.36k
            std::make_shared<DataTypeArray>(make_nullable(list_field->children[0].data_type));
511
1.36k
    if (is_optional) {
512
1.21k
        list_field->data_type = make_nullable(list_field->data_type);
513
1.21k
    }
514
1.36k
    list_field->field_id = first_level.__isset.field_id ? first_level.field_id : -1;
515
516
1.36k
    return Status::OK();
517
1.36k
}
518
519
Status FieldDescriptor::parse_map_field(const std::vector<tparquet::SchemaElement>& t_schemas,
520
859
                                        size_t curr_pos, FieldSchema* map_field) {
521
    // the map definition in parquet:
522
    // optional group <name> (MAP) {
523
    //   repeated group map (MAP_KEY_VALUE) {
524
    //     required <type> key;
525
    //     optional <type> value;
526
    //   }
527
    // }
528
    // Map value can be optional, the map without values is a SET
529
859
    if (curr_pos + 2 >= t_schemas.size()) {
530
0
        return Status::InvalidArgument("Map element should have at least three levels");
531
0
    }
532
859
    auto& map_schema = t_schemas[curr_pos];
533
859
    if (map_schema.num_children != 1) {
534
2
        return Status::InvalidArgument(
535
2
                "Map element should have only one child(name='map', type='MAP_KEY_VALUE')");
536
2
    }
537
857
    if (is_repeated_node(map_schema)) {
538
0
        return Status::InvalidArgument("Map element can't be a repeated schema");
539
0
    }
540
857
    auto& map_key_value = t_schemas[curr_pos + 1];
541
857
    if (!is_group_node(map_key_value) || !is_repeated_node(map_key_value)) {
542
0
        return Status::InvalidArgument(
543
0
                "the second level in map must be a repeated group(key and value)");
544
0
    }
545
857
    auto& map_key = t_schemas[curr_pos + 2];
546
857
    if (!is_required_node(map_key)) {
547
12
        LOG(WARNING) << "Filed " << map_schema.name << " is map type, but with nullable key column";
548
12
    }
549
550
857
    if (map_key_value.num_children == 1) {
551
        // The map with three levels is a SET
552
0
        return parse_list_field(t_schemas, curr_pos, map_field);
553
0
    }
554
857
    if (map_key_value.num_children != 2) {
555
        // A standard map should have four levels
556
0
        return Status::InvalidArgument(
557
0
                "the second level in map(MAP_KEY_VALUE) should have two children");
558
0
    }
559
    // standard map
560
857
    bool is_optional = is_optional_node(map_schema);
561
857
    if (is_optional) {
562
805
        map_field->definition_level++;
563
805
    }
564
857
    map_field->repetition_level++;
565
857
    map_field->definition_level++;
566
567
    // Directly create key and value children instead of intermediate key_value node
568
857
    map_field->children.resize(2);
569
    // map is a repeated node, we should set the `repeated_parent_def_level` of its children as `definition_level`
570
857
    set_child_node_level(map_field, map_field->definition_level);
571
572
857
    auto key_field = &map_field->children[0];
573
857
    auto value_field = &map_field->children[1];
574
575
    // Parse key and value fields directly from the key_value group's children
576
857
    _next_schema_pos = curr_pos + 2; // Skip key_value group, go directly to key
577
857
    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, key_field));
578
857
    RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, value_field));
579
580
857
    map_field->name = map_schema.name;
581
857
    map_field->lower_case_name = to_lower(map_field->name);
582
857
    map_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
583
857
    map_field->data_type = std::make_shared<DataTypeMap>(make_nullable(key_field->data_type),
584
857
                                                         make_nullable(value_field->data_type));
585
857
    if (is_optional) {
586
805
        map_field->data_type = make_nullable(map_field->data_type);
587
805
    }
588
857
    map_field->field_id = map_schema.__isset.field_id ? map_schema.field_id : -1;
589
590
857
    return Status::OK();
591
857
}
592
593
Status FieldDescriptor::parse_struct_field(const std::vector<tparquet::SchemaElement>& t_schemas,
594
1.47k
                                           size_t curr_pos, FieldSchema* struct_field) {
595
    // the nested column in parquet, parse group to struct.
596
1.47k
    auto& struct_schema = t_schemas[curr_pos];
597
1.47k
    bool is_optional = is_optional_node(struct_schema);
598
1.47k
    if (is_optional) {
599
1.28k
        struct_field->definition_level++;
600
1.28k
    }
601
1.47k
    auto num_children = struct_schema.num_children;
602
1.47k
    struct_field->children.resize(num_children);
603
1.47k
    set_child_node_level(struct_field, struct_field->repeated_parent_def_level);
604
1.47k
    _next_schema_pos = curr_pos + 1;
605
7.03k
    for (int i = 0; i < num_children; ++i) {
606
5.56k
        RETURN_IF_ERROR(parse_node_field(t_schemas, _next_schema_pos, &struct_field->children[i]));
607
5.56k
    }
608
1.47k
    struct_field->name = struct_schema.name;
609
1.47k
    struct_field->lower_case_name = to_lower(struct_field->name);
610
1.47k
    struct_field->column_id = UNASSIGNED_COLUMN_ID; // Initialize column_id
611
612
1.47k
    struct_field->field_id = struct_schema.__isset.field_id ? struct_schema.field_id : -1;
613
1.47k
    DataTypes res_data_types;
614
1.47k
    std::vector<String> names;
615
7.02k
    for (int i = 0; i < num_children; ++i) {
616
5.55k
        res_data_types.push_back(make_nullable(struct_field->children[i].data_type));
617
5.55k
        names.push_back(struct_field->children[i].name);
618
5.55k
    }
619
1.47k
    struct_field->data_type = std::make_shared<DataTypeStruct>(res_data_types, names);
620
1.47k
    if (is_optional) {
621
1.28k
        struct_field->data_type = make_nullable(struct_field->data_type);
622
1.28k
    }
623
1.47k
    return Status::OK();
624
1.47k
}
625
626
5
int FieldDescriptor::get_column_index(const std::string& column) const {
627
15
    for (int32_t i = 0; i < _fields.size(); i++) {
628
15
        if (_fields[i].name == column) {
629
5
            return i;
630
5
        }
631
15
    }
632
0
    return -1;
633
5
}
634
635
60.1k
FieldSchema* FieldDescriptor::get_column(const std::string& name) const {
636
60.1k
    auto it = _name_to_field.find(name);
637
60.1k
    if (it != _name_to_field.end()) {
638
60.1k
        return it->second;
639
60.1k
    }
640
18.4E
    throw Exception(Status::InternalError("Name {} not found in FieldDescriptor!", name));
641
0
    return nullptr;
642
60.1k
}
643
644
1.47k
void FieldDescriptor::get_column_names(std::unordered_set<std::string>* names) const {
645
1.47k
    names->clear();
646
8.88k
    for (const FieldSchema& f : _fields) {
647
8.88k
        names->emplace(f.name);
648
8.88k
    }
649
1.47k
}
650
651
0
std::string FieldDescriptor::debug_string() const {
652
0
    std::stringstream ss;
653
0
    ss << "fields=[";
654
0
    for (int i = 0; i < _fields.size(); ++i) {
655
0
        if (i != 0) {
656
0
            ss << ", ";
657
0
        }
658
0
        ss << _fields[i].debug_string();
659
0
    }
660
0
    ss << "]";
661
0
    return ss.str();
662
0
}
663
664
1.31k
void FieldDescriptor::assign_ids() {
665
1.31k
    uint64_t next_id = 1;
666
7.52k
    for (auto& field : _fields) {
667
7.52k
        field.assign_ids(next_id);
668
7.52k
    }
669
1.31k
}
670
671
0
const FieldSchema* FieldDescriptor::find_column_by_id(uint64_t column_id) const {
672
0
    for (const auto& field : _fields) {
673
0
        if (auto result = field.find_column_by_id(column_id)) {
674
0
            return result;
675
0
        }
676
0
    }
677
0
    return nullptr;
678
0
}
679
680
11.8k
void FieldSchema::assign_ids(uint64_t& next_id) {
681
11.8k
    column_id = next_id++;
682
683
11.8k
    for (auto& child : children) {
684
4.27k
        child.assign_ids(next_id);
685
4.27k
    }
686
687
11.8k
    max_column_id = next_id - 1;
688
11.8k
}
689
690
0
const FieldSchema* FieldSchema::find_column_by_id(uint64_t target_id) const {
691
0
    if (column_id == target_id) {
692
0
        return this;
693
0
    }
694
695
0
    for (const auto& child : children) {
696
0
        if (auto result = child.find_column_by_id(target_id)) {
697
0
            return result;
698
0
        }
699
0
    }
700
701
0
    return nullptr;
702
0
}
703
704
16.4k
uint64_t FieldSchema::get_column_id() const {
705
16.4k
    return column_id;
706
16.4k
}
707
708
0
void FieldSchema::set_column_id(uint64_t id) {
709
0
    column_id = id;
710
0
}
711
712
904
uint64_t FieldSchema::get_max_column_id() const {
713
904
    return max_column_id;
714
904
}
715
716
} // namespace doris