Coverage Report

Created: 2026-09-12 19:24

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format/table/iceberg_reader.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/table/iceberg_reader.h"
19
20
#include <gen_cpp/Descriptors_types.h>
21
#include <gen_cpp/Metrics_types.h>
22
#include <gen_cpp/PlanNodes_types.h>
23
#include <gen_cpp/parquet_types.h>
24
#include <glog/logging.h>
25
#include <parallel_hashmap/phmap.h>
26
#include <rapidjson/document.h>
27
28
#include <algorithm>
29
#include <cstring>
30
#include <functional>
31
#include <memory>
32
33
#include "common/compiler_util.h" // IWYU pragma: keep
34
#include "common/consts.h"
35
#include "common/status.h"
36
#include "core/assert_cast.h"
37
#include "core/block/block.h"
38
#include "core/block/column_with_type_and_name.h"
39
#include "core/column/column.h"
40
#include "core/column/column_nullable.h"
41
#include "core/column/column_string.h"
42
#include "core/column/column_vector.h"
43
#include "core/data_type/data_type_factory.hpp"
44
#include "core/data_type/data_type_nullable.h"
45
#include "core/data_type/define_primitive_type.h"
46
#include "core/data_type/primitive_type.h"
47
#include "core/string_ref.h"
48
#include "exprs/aggregate/aggregate_function.h"
49
#include "format/format_common.h"
50
#include "format/generic_reader.h"
51
#include "format/orc/vorc_reader.h"
52
#include "format/parquet/schema_desc.h"
53
#include "format/parquet/vparquet_column_chunk_reader.h"
54
#include "format/table/deletion_vector_reader.h"
55
#include "format/table/iceberg/iceberg_orc_nested_column_utils.h"
56
#include "format/table/iceberg/iceberg_parquet_nested_column_utils.h"
57
#include "format/table/iceberg_scan_semantics.h"
58
#include "format/table/nested_column_access_helper.h"
59
#include "format/table/table_schema_change_helper.h"
60
#include "runtime/runtime_state.h"
61
#include "util/coding.h"
62
#include "util/string_util.h"
63
64
namespace cctz {
65
class time_zone;
66
} // namespace cctz
67
namespace doris {
68
class RowDescriptor;
69
class SlotDescriptor;
70
class TupleDescriptor;
71
72
namespace io {
73
struct IOContext;
74
} // namespace io
75
class VExprContext;
76
} // namespace doris
77
78
namespace doris {
79
namespace {
80
81
constexpr auto kIcebergOrcAttribute = "iceberg.id";
82
83
42
bool orc_subtree_has_iceberg_id(const orc::Type* type, const std::string& attribute) {
84
42
    if (type->hasAttributeKey(attribute)) {
85
9
        return true;
86
9
    }
87
51
    for (uint64_t idx = 0; idx < type->getSubtypeCount(); ++idx) {
88
27
        if (orc_subtree_has_iceberg_id(type->getSubtype(idx), attribute)) {
89
9
            return true;
90
9
        }
91
27
    }
92
24
    return false;
93
33
}
94
95
49
bool parquet_subtree_has_iceberg_id(const FieldSchema& field) {
96
49
    if (field.field_id >= 0) {
97
26
        return true;
98
26
    }
99
23
    return std::ranges::any_of(field.children, parquet_subtree_has_iceberg_id);
100
49
}
101
102
struct ParquetEqualityFieldPath {
103
    std::vector<const FieldSchema*> fields;
104
    std::vector<size_t> child_indexes;
105
};
106
107
bool find_parquet_equality_field_path_by_id(const FieldDescriptor* descriptor, int32_t field_id,
108
8
                                            ParquetEqualityFieldPath* result) {
109
8
    DORIS_CHECK(descriptor != nullptr);
110
8
    DORIS_CHECK(result != nullptr);
111
8
    const auto find = [field_id](const auto& self, const FieldSchema* field,
112
24
                                 ParquetEqualityFieldPath* path) -> bool {
113
24
        DORIS_CHECK(field != nullptr);
114
24
        path->fields.push_back(field);
115
24
        if (field->field_id == field_id) {
116
4
            return true;
117
4
        }
118
21
        for (size_t index = 0; index < field->children.size(); ++index) {
119
4
            path->child_indexes.push_back(index);
120
4
            if (self(self, &field->children[index], path)) {
121
3
                return true;
122
3
            }
123
1
            path->child_indexes.pop_back();
124
1
        }
125
17
        path->fields.pop_back();
126
17
        return false;
127
20
    };
128
24
    for (int index = 0; index < descriptor->size(); ++index) {
129
20
        if (find(find, descriptor->get_column(index), result)) {
130
4
            return true;
131
4
        }
132
20
    }
133
4
    return false;
134
8
}
135
136
bool find_parquet_equality_field_prefix_by_id_path(
137
        const FieldDescriptor* descriptor,
138
        const std::vector<const schema::external::TField*>& table_path,
139
4
        ParquetEqualityFieldPath* result) {
140
4
    DORIS_CHECK(descriptor != nullptr);
141
4
    DORIS_CHECK(result != nullptr);
142
4
    DORIS_CHECK(!table_path.empty());
143
4
    const std::vector<FieldSchema>* candidates = nullptr;
144
5
    for (size_t path_index = 0; path_index < table_path.size(); ++path_index) {
145
5
        const auto* table_field = table_path[path_index];
146
5
        DORIS_CHECK(table_field != nullptr);
147
5
        DORIS_CHECK(table_field->__isset.id);
148
5
        const FieldSchema* match = nullptr;
149
5
        size_t match_index = 0;
150
5
        const size_t candidate_count =
151
5
                candidates == nullptr ? cast_set<size_t>(descriptor->size()) : candidates->size();
152
17
        for (size_t candidate_index = 0; candidate_index < candidate_count; ++candidate_index) {
153
12
            const auto* candidate = candidates == nullptr
154
12
                                            ? descriptor->get_column(cast_set<int>(candidate_index))
155
12
                                            : &(*candidates)[candidate_index];
156
12
            if (candidate != nullptr && candidate->field_id == table_field->id) {
157
0
                match = candidate;
158
0
                match_index = candidate_index;
159
0
                break;
160
0
            }
161
12
        }
162
5
        if (match == nullptr) {
163
5
            const auto wrapper =
164
5
                    candidates == nullptr
165
5
                            ? TableSchemaChangeHelper::BuildTableInfoUtil::
166
4
                                      find_unique_idless_parquet_wrapper_index(
167
4
                                              *table_field, descriptor->get_fields_schema())
168
5
                            : TableSchemaChangeHelper::BuildTableInfoUtil::
169
1
                                      find_unique_idless_parquet_wrapper_index(*table_field,
170
1
                                                                               *candidates);
171
5
            if (wrapper.has_value()) {
172
1
                match_index = *wrapper;
173
1
                match = candidates == nullptr ? descriptor->get_column(cast_set<int>(match_index))
174
1
                                              : &(*candidates)[match_index];
175
1
            }
176
5
        }
177
5
        if (match == nullptr) {
178
4
            return false;
179
4
        }
180
1
        if (!result->fields.empty()) {
181
0
            result->child_indexes.push_back(match_index);
182
0
        }
183
1
        result->fields.push_back(match);
184
1
        candidates = &match->children;
185
1
    }
186
0
    return true;
187
4
}
188
189
std::vector<std::string> equality_field_name_candidates(const schema::external::TField& table_field,
190
21
                                                        const std::string* leaf_fallback) {
191
21
    std::vector<std::string> candidates;
192
21
    if (table_field.__isset.name_mapping) {
193
11
        candidates.insert(candidates.end(), table_field.name_mapping.begin(),
194
11
                          table_field.name_mapping.end());
195
11
        if (table_field.__isset.name_mapping_is_authoritative &&
196
11
            table_field.name_mapping_is_authoritative) {
197
10
            return candidates;
198
10
        }
199
11
    }
200
11
    if (table_field.__isset.name) {
201
11
        candidates.push_back(table_field.name);
202
11
    }
203
11
    if (leaf_fallback != nullptr) {
204
11
        candidates.push_back(*leaf_fallback);
205
11
    }
206
11
    return candidates;
207
21
}
208
209
bool find_parquet_equality_field_prefix_by_name_path(
210
        const FieldDescriptor* descriptor,
211
        const std::vector<const schema::external::TField*>& table_path,
212
7
        const std::string& leaf_fallback, ParquetEqualityFieldPath* result) {
213
7
    DORIS_CHECK(descriptor != nullptr);
214
7
    DORIS_CHECK(result != nullptr);
215
7
    DORIS_CHECK(!table_path.empty());
216
7
    const std::vector<FieldSchema>* children = nullptr;
217
17
    for (size_t path_index = 0; path_index < table_path.size(); ++path_index) {
218
11
        const auto* table_field = table_path[path_index];
219
11
        DORIS_CHECK(table_field != nullptr);
220
11
        const auto names = equality_field_name_candidates(
221
11
                *table_field, path_index + 1 == table_path.size() ? &leaf_fallback : nullptr);
222
11
        const FieldSchema* match = nullptr;
223
11
        size_t match_index = 0;
224
11
        const size_t child_count =
225
11
                children == nullptr ? cast_set<size_t>(descriptor->size()) : children->size();
226
15
        for (const auto& name : names) {
227
29
            for (size_t child_index = 0; child_index < child_count; ++child_index) {
228
24
                const auto* child = children == nullptr
229
24
                                            ? descriptor->get_column(cast_set<int>(child_index))
230
24
                                            : &(*children)[child_index];
231
24
                if (child != nullptr && iequal(child->name, name)) {
232
10
                    match = child;
233
10
                    match_index = child_index;
234
10
                    break;
235
10
                }
236
24
            }
237
15
            if (match != nullptr) {
238
10
                break;
239
10
            }
240
15
        }
241
11
        if (match == nullptr) {
242
1
            return false;
243
1
        }
244
10
        if (!result->fields.empty()) {
245
4
            result->child_indexes.push_back(match_index);
246
4
        }
247
10
        result->fields.push_back(match);
248
10
        children = &match->children;
249
10
    }
250
6
    return true;
251
7
}
252
253
struct OrcEqualityFieldPath {
254
    std::vector<const orc::Type*> fields;
255
    std::vector<std::string> names;
256
    std::vector<size_t> child_indexes;
257
};
258
259
bool find_orc_equality_field_path_by_id(const orc::Type* root, int32_t field_id,
260
8
                                        OrcEqualityFieldPath* result) {
261
8
    DORIS_CHECK(root != nullptr);
262
8
    DORIS_CHECK(result != nullptr);
263
8
    const auto find = [field_id](const auto& self, const orc::Type* field,
264
8
                                 const std::string& field_name,
265
22
                                 OrcEqualityFieldPath* path) -> bool {
266
22
        DORIS_CHECK(field != nullptr);
267
22
        path->fields.push_back(field);
268
22
        path->names.push_back(field_name);
269
22
        if (field->hasAttributeKey(kIcebergOrcAttribute) &&
270
22
            std::stoi(field->getAttributeValue(kIcebergOrcAttribute)) == field_id) {
271
3
            return true;
272
3
        }
273
20
        for (size_t index = 0; index < field->getSubtypeCount(); ++index) {
274
3
            path->child_indexes.push_back(index);
275
3
            if (self(self, field->getSubtype(index), field->getFieldName(index), path)) {
276
2
                return true;
277
2
            }
278
1
            path->child_indexes.pop_back();
279
1
        }
280
17
        path->fields.pop_back();
281
17
        path->names.pop_back();
282
17
        return false;
283
19
    };
284
24
    for (size_t index = 0; index < root->getSubtypeCount(); ++index) {
285
19
        if (find(find, root->getSubtype(index), root->getFieldName(index), result)) {
286
3
            return true;
287
3
        }
288
19
    }
289
5
    return false;
290
8
}
291
292
bool find_orc_equality_field_prefix_by_id_path(
293
        const orc::Type* root, const std::vector<const schema::external::TField*>& table_path,
294
4
        OrcEqualityFieldPath* result) {
295
4
    DORIS_CHECK(root != nullptr);
296
4
    DORIS_CHECK(result != nullptr);
297
4
    DORIS_CHECK(!table_path.empty());
298
4
    const orc::Type* parent = root;
299
5
    for (const auto* table_field : table_path) {
300
5
        DORIS_CHECK(table_field != nullptr);
301
5
        DORIS_CHECK(table_field->__isset.id);
302
5
        const orc::Type* match = nullptr;
303
5
        size_t match_index = 0;
304
17
        for (size_t candidate_index = 0; candidate_index < parent->getSubtypeCount();
305
12
             ++candidate_index) {
306
12
            const auto* candidate = parent->getSubtype(candidate_index);
307
12
            if (candidate->hasAttributeKey(kIcebergOrcAttribute) &&
308
12
                std::stoi(candidate->getAttributeValue(kIcebergOrcAttribute)) == table_field->id) {
309
0
                match = candidate;
310
0
                match_index = candidate_index;
311
0
                break;
312
0
            }
313
12
        }
314
5
        if (match == nullptr) {
315
5
            const auto wrapper = TableSchemaChangeHelper::BuildTableInfoUtil::
316
5
                    find_unique_idless_orc_wrapper_index(*table_field, parent,
317
5
                                                         kIcebergOrcAttribute);
318
5
            if (wrapper.has_value()) {
319
1
                match_index = *wrapper;
320
1
                match = parent->getSubtype(match_index);
321
1
            }
322
5
        }
323
5
        if (match == nullptr) {
324
4
            return false;
325
4
        }
326
1
        if (!result->fields.empty()) {
327
0
            result->child_indexes.push_back(match_index);
328
0
        }
329
1
        result->fields.push_back(match);
330
1
        result->names.push_back(parent->getFieldName(match_index));
331
1
        parent = match;
332
1
    }
333
0
    return true;
334
4
}
335
336
bool find_orc_equality_field_prefix_by_name_path(
337
        const orc::Type* root, const std::vector<const schema::external::TField*>& table_path,
338
6
        const std::string& leaf_fallback, OrcEqualityFieldPath* result) {
339
6
    DORIS_CHECK(root != nullptr);
340
6
    DORIS_CHECK(result != nullptr);
341
6
    DORIS_CHECK(!table_path.empty());
342
6
    const orc::Type* parent = root;
343
15
    for (size_t path_index = 0; path_index < table_path.size(); ++path_index) {
344
10
        const auto* table_field = table_path[path_index];
345
10
        DORIS_CHECK(table_field != nullptr);
346
10
        const auto names = equality_field_name_candidates(
347
10
                *table_field, path_index + 1 == table_path.size() ? &leaf_fallback : nullptr);
348
10
        const orc::Type* match = nullptr;
349
10
        size_t match_index = 0;
350
14
        for (const auto& name : names) {
351
26
            for (size_t child_index = 0; child_index < parent->getSubtypeCount(); ++child_index) {
352
21
                if (iequal(parent->getFieldName(child_index), name)) {
353
9
                    match = parent->getSubtype(child_index);
354
9
                    match_index = child_index;
355
9
                    break;
356
9
                }
357
21
            }
358
14
            if (match != nullptr) {
359
9
                break;
360
9
            }
361
14
        }
362
10
        if (match == nullptr) {
363
1
            return false;
364
1
        }
365
9
        if (!result->fields.empty()) {
366
4
            result->child_indexes.push_back(match_index);
367
4
        }
368
9
        result->fields.push_back(match);
369
9
        result->names.push_back(parent->getFieldName(match_index));
370
9
        parent = match;
371
9
    }
372
5
    return true;
373
6
}
374
375
} // namespace
376
377
const std::string IcebergOrcReader::ICEBERG_ORC_ATTRIBUTE = kIcebergOrcAttribute;
378
379
bool IcebergTableReader::_is_fully_dictionary_encoded(
380
8
        const tparquet::ColumnMetaData& column_metadata) {
381
14
    const auto is_dictionary_encoding = [](tparquet::Encoding::type encoding) {
382
14
        return encoding == tparquet::Encoding::PLAIN_DICTIONARY ||
383
14
               encoding == tparquet::Encoding::RLE_DICTIONARY;
384
14
    };
385
12
    const auto is_data_page = [](tparquet::PageType::type page_type) {
386
12
        return page_type == tparquet::PageType::DATA_PAGE ||
387
12
               page_type == tparquet::PageType::DATA_PAGE_V2;
388
12
    };
389
8
    const auto is_level_encoding = [](tparquet::Encoding::type encoding) {
390
2
        return encoding == tparquet::Encoding::RLE || encoding == tparquet::Encoding::BIT_PACKED;
391
2
    };
392
393
    // A column chunk may have a dictionary page but still contain plain-encoded data pages.
394
    // Only treat it as dictionary-coded when all data pages are dictionary encoded.
395
8
    if (column_metadata.__isset.encoding_stats) {
396
7
        bool has_data_page_stats = false;
397
12
        for (const tparquet::PageEncodingStats& enc_stat : column_metadata.encoding_stats) {
398
12
            if (is_data_page(enc_stat.page_type) && enc_stat.count > 0) {
399
8
                has_data_page_stats = true;
400
8
                if (!is_dictionary_encoding(enc_stat.encoding)) {
401
2
                    return false;
402
2
                }
403
8
            }
404
12
        }
405
5
        if (has_data_page_stats) {
406
4
            return true;
407
4
        }
408
5
    }
409
410
2
    bool has_dict_encoding = false;
411
2
    bool has_nondict_encoding = false;
412
3
    for (const tparquet::Encoding::type& encoding : column_metadata.encodings) {
413
3
        if (is_dictionary_encoding(encoding)) {
414
1
            has_dict_encoding = true;
415
1
        }
416
417
3
        if (!is_dictionary_encoding(encoding) && !is_level_encoding(encoding)) {
418
2
            has_nondict_encoding = true;
419
2
            break;
420
2
        }
421
3
    }
422
2
    if (!has_dict_encoding || has_nondict_encoding) {
423
2
        return false;
424
2
    }
425
426
0
    return true;
427
2
}
428
429
// ============================================================================
430
// IcebergParquetReader: on_before_init_reader (Parquet-specific schema matching)
431
// ============================================================================
432
// This format-specific setup mirrors the existing reader initialization sequence.
433
// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
434
18
Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
435
18
    _column_descs = ctx->column_descs;
436
18
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
437
18
    _file_format = Fileformat::PARQUET;
438
18
    set_preserve_binary_uuid(true);
439
440
    // Get file metadata schema first (available because _open_file() already ran)
441
18
    const FieldDescriptor* field_desc = nullptr;
442
18
    RETURN_IF_ERROR(this->get_file_metadata_schema(&field_desc));
443
18
    DCHECK(field_desc != nullptr);
444
445
    // Build table_info_node by field_id or name matching.
446
    // This must happen BEFORE column classification so we can use children_column_exists
447
    // to check if a column exists in the file (by field ID, not name).
448
18
    if (!get_scan_params().__isset.history_schema_info ||
449
18
        get_scan_params().history_schema_info.empty()) [[unlikely]] {
450
1
        RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_name(ctx->tuple_descriptor, *field_desc,
451
1
                                                            ctx->table_info_node));
452
17
    } else {
453
17
        RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id_with_name_mapping(
454
17
                get_scan_params().history_schema_info.front().root_field, *field_desc,
455
17
                ctx->table_info_node, supports_iceberg_scan_semantics_v1(&get_scan_params())));
456
17
    }
457
458
18
    std::unordered_set<std::string> partition_col_names;
459
18
    if (ctx->range->__isset.columns_from_path_keys) {
460
0
        partition_col_names.insert(ctx->range->columns_from_path_keys.begin(),
461
0
                                   ctx->range->columns_from_path_keys.end());
462
0
    }
463
464
    // Single pass: classify columns, detect $row_id, handle partition fallback.
465
18
    bool has_partition_from_path = false;
466
20
    for (const auto& desc : *ctx->column_descs) {
467
20
        if (desc.category == ColumnCategory::SYNTHESIZED) {
468
0
            if (desc.name == BeConsts::ICEBERG_ROWID_COL) {
469
0
                this->register_synthesized_column_handler(
470
0
                        BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status {
471
0
                            return _fill_iceberg_row_id(block, rows);
472
0
                        });
473
0
                continue;
474
0
            } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
475
0
                auto topn_row_id_column_iter = _create_topn_row_id_column_iterator();
476
0
                this->register_synthesized_column_handler(
477
0
                        desc.name,
478
0
                        [iter = std::move(topn_row_id_column_iter), this, &desc](
479
0
                                Block* block, size_t rows) -> Status {
480
0
                            return fill_topn_row_id(iter, desc.name, block, rows);
481
0
                        });
482
0
                continue;
483
0
            }
484
20
        } else if (desc.category == ColumnCategory::PARTITION_KEY) {
485
0
            bool has_partition_value = partition_col_names.contains(desc.name);
486
0
            bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name);
487
0
            if (!has_partition_value || exists_in_file) {
488
                // Keep PARTITION_KEY category stable for scan planning, but still read
489
                // from file when the column exists there.
490
0
                ctx->column_names.push_back(desc.name);
491
0
                continue;
492
0
            }
493
0
            has_partition_from_path = true;
494
20
        } else if (desc.category == ColumnCategory::REGULAR) {
495
20
            ctx->column_names.push_back(desc.name);
496
20
        } else if (desc.category == ColumnCategory::GENERATED) {
497
0
            _init_row_lineage_columns();
498
0
            if (desc.name == ROW_LINEAGE_ROW_ID) {
499
0
                ctx->column_names.push_back(desc.name);
500
0
                this->register_generated_column_handler(
501
0
                        ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status {
502
0
                            return _fill_row_lineage_row_id(block, rows);
503
0
                        });
504
0
                continue;
505
0
            } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
506
0
                ctx->column_names.push_back(desc.name);
507
0
                this->register_generated_column_handler(
508
0
                        ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER,
509
0
                        [this](Block* block, size_t rows) -> Status {
510
0
                            return _fill_row_lineage_last_updated_sequence_number(block, rows);
511
0
                        });
512
0
                continue;
513
0
            }
514
0
        }
515
20
    }
516
517
    // Set up partition value extraction if any partition columns need filling from path
518
18
    if (has_partition_from_path) {
519
0
        RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
520
0
                                                  _fill_partition_values,
521
0
                                                  &_fill_partition_value_is_null));
522
0
    }
523
524
18
    _all_required_col_names = ctx->column_names;
525
526
    // Create column IDs from field descriptor
527
18
    auto column_id_result =
528
18
            _create_column_ids(field_desc, ctx->tuple_descriptor, ctx->table_info_node);
529
18
    ctx->column_ids = std::move(column_id_result.column_ids);
530
18
    ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);
531
532
    // Build field_id -> block_column_name mapping for equality delete filtering.
533
    // This was previously done in init_reader() column matching (pre-CRTP refactoring).
534
20
    for (const auto* slot : ctx->tuple_descriptor->slots()) {
535
20
        _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name());
536
20
    }
537
538
    // Process delete files (must happen before _do_init_reader so expand col IDs are included)
539
18
    RETURN_IF_ERROR(_init_row_filters());
540
541
    // Add expand column IDs for equality delete and remap expand column names
542
    // to match master's behavior:
543
    // - Use field_id to find the actual file column name in Parquet schema
544
    // - Prefix with __equality_delete_column__ to avoid name conflicts
545
    // - Correctly map table_col_name → file_col_name in table_info_node
546
18
    const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
547
18
    bool all_file_columns_have_field_ids = true;
548
18
    bool any_file_column_has_field_id = false;
549
60
    for (int i = 0; i < field_desc->size(); ++i) {
550
42
        const auto* field_schema = field_desc->get_column(i);
551
42
        if (field_schema) {
552
42
            if (field_schema->field_id < 0) {
553
19
                all_file_columns_have_field_ids = false;
554
19
            }
555
42
            if (parquet_subtree_has_iceberg_id(*field_schema)) {
556
26
                any_file_column_has_field_id = true;
557
26
            }
558
42
        }
559
42
    }
560
18
    const bool use_field_ids_for_hidden_keys =
561
18
            supports_iceberg_scan_semantics_v1(&get_scan_params())
562
18
                    ? any_file_column_has_field_id
563
18
                    : all_file_columns_have_field_ids;
564
18
    const auto find_file_column_by_name = [&](const std::string& name) -> const FieldSchema* {
565
0
        for (int j = 0; j < field_desc->size(); ++j) {
566
0
            const auto* candidate = field_desc->get_column(j);
567
0
            if (candidate != nullptr && iequal(candidate->name, name)) {
568
0
                return candidate;
569
0
            }
570
0
        }
571
0
        return nullptr;
572
0
    };
573
574
    // Rebuild _expand_col_names with proper file-column-based names
575
18
    std::vector<std::string> new_expand_col_names;
576
18
    DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
577
18
    DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
578
33
    for (size_t i = 0; i < _expand_col_names.size(); ++i) {
579
15
        const auto& old_name = _expand_col_names[i];
580
15
        const int32_t field_id = _expand_col_field_ids[i];
581
582
15
        const FieldSchema* file_column = nullptr;
583
15
        ParquetEqualityFieldPath file_path;
584
15
        bool complete_file_path = false;
585
15
        if (use_field_ids_for_hidden_keys) {
586
8
            complete_file_path =
587
8
                    find_parquet_equality_field_path_by_id(field_desc, field_id, &file_path);
588
8
            if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) {
589
4
                const auto table_path = _find_schema_field_path(field_id);
590
4
                if (!table_path.empty()) {
591
4
                    complete_file_path = find_parquet_equality_field_prefix_by_id_path(
592
4
                            field_desc, table_path, &file_path);
593
4
                }
594
4
            }
595
8
            if (!file_path.fields.empty()) {
596
5
                file_column = file_path.fields.front();
597
5
            }
598
8
        } else {
599
7
            const auto table_path = _find_schema_field_path(field_id);
600
7
            if (!table_path.empty()) {
601
7
                complete_file_path = find_parquet_equality_field_prefix_by_name_path(
602
7
                        field_desc, table_path, old_name, &file_path);
603
7
                if (!file_path.fields.empty()) {
604
6
                    file_column = file_path.fields.front();
605
6
                }
606
7
            } else {
607
0
                file_column = find_file_column_by_name(old_name);
608
0
                complete_file_path = file_column != nullptr;
609
0
            }
610
7
        }
611
612
15
        std::string leaf_name = old_name;
613
15
        if (!file_path.fields.empty()) {
614
11
            leaf_name = file_path.fields.back()->name;
615
11
        } else if (file_column != nullptr) {
616
0
            leaf_name = file_column->name;
617
0
        }
618
15
        const std::string file_col_name = file_column == nullptr ? old_name : file_column->name;
619
15
        std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name;
620
621
        // Update _id_to_block_column_name
622
15
        if (field_id >= 0) {
623
15
            _id_to_block_column_name[field_id] = table_col_name;
624
15
        }
625
626
        // Update _expand_columns name
627
15
        _expand_columns[i].name = table_col_name;
628
629
15
        if (file_column == nullptr) {
630
4
            RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
631
4
                                                                     _expand_columns[i].type));
632
            // The old data file predates this equality key. Keep it in the expand block so the
633
            // synthesized-column hook can materialize its logical initial default before reader
634
            // filtering, but do not advertise it to Parquet as a physical child.
635
4
            new_expand_col_names.push_back(table_col_name);
636
4
            continue;
637
4
        }
638
639
11
        new_expand_col_names.push_back(table_col_name);
640
641
11
        if (!complete_file_path) {
642
1
            ColumnPtr missing_value;
643
1
            RETURN_IF_ERROR(_create_missing_equality_delete_value(
644
1
                    field_id, _expand_columns[i].type, file_path.fields.size(), &missing_value));
645
1
            _nested_equality_delete_columns.push_back({
646
1
                    .field_id = field_id,
647
1
                    .block_name = table_col_name,
648
1
                    .leaf_type = _expand_columns[i].type,
649
1
                    .child_indexes = file_path.child_indexes,
650
1
                    .missing_value = std::move(missing_value),
651
1
            });
652
1
            _expand_columns[i].type = make_nullable(file_column->data_type);
653
1
            _expand_columns[i].column = _expand_columns[i].type->create_column();
654
10
        } else if (!file_path.child_indexes.empty()) {
655
7
            _nested_equality_delete_columns.push_back({
656
7
                    .field_id = field_id,
657
7
                    .block_name = table_col_name,
658
7
                    .leaf_type = _expand_columns[i].type,
659
7
                    .child_indexes = file_path.child_indexes,
660
7
                    .missing_value = nullptr,
661
7
            });
662
7
            _expand_columns[i].type = make_nullable(file_column->data_type);
663
7
            _expand_columns[i].column = _expand_columns[i].type->create_column();
664
7
        }
665
666
        // A hidden nested key is read through its containing top-level struct. V1 column IDs are
667
        // pre-order ranges, so include the complete subtree before extracting the primitive leaf.
668
11
        for (uint64_t column_id = file_column->get_column_id();
669
30
             column_id <= file_column->get_max_column_id(); ++column_id) {
670
19
            ctx->column_ids.insert(column_id);
671
19
        }
672
673
        // Register in table_info_node: table_col_name → file_col_name
674
11
        ctx->column_names.push_back(table_col_name);
675
11
        ctx->table_info_node->add_children(table_col_name, file_col_name,
676
11
                                           TableSchemaChangeHelper::ConstNode::get_instance());
677
11
    }
678
18
    _expand_col_names = std::move(new_expand_col_names);
679
680
    // Enable group filtering for Iceberg
681
18
    _filter_groups = true;
682
683
18
    return Status::OK();
684
18
}
685
686
// ============================================================================
687
// IcebergParquetReader: _create_column_ids
688
// ============================================================================
689
ColumnIdResult IcebergParquetReader::_create_column_ids(
690
        const FieldDescriptor* field_desc, const TupleDescriptor* tuple_descriptor,
691
25
        const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
692
25
    auto* mutable_field_desc = const_cast<FieldDescriptor*>(field_desc);
693
25
    mutable_field_desc->assign_ids();
694
695
25
    std::unordered_map<int, const FieldSchema*> iceberg_id_to_field_schema_map;
696
117
    for (int i = 0; i < field_desc->size(); ++i) {
697
92
        const auto* field_schema = field_desc->get_column(i);
698
92
        if (!field_schema) {
699
0
            continue;
700
0
        }
701
92
        int iceberg_id = field_schema->field_id;
702
92
        iceberg_id_to_field_schema_map[iceberg_id] = field_schema;
703
92
    }
704
705
25
    std::set<uint64_t> column_ids;
706
25
    std::set<uint64_t> filter_column_ids;
707
708
25
    auto process_access_paths = [](const FieldSchema* parquet_field,
709
25
                                   const std::vector<TColumnAccessPath>& access_paths,
710
25
                                   std::set<uint64_t>& out_ids) {
711
17
        process_nested_access_paths(
712
17
                parquet_field, access_paths, out_ids,
713
17
                [](const FieldSchema* field) { return field->get_column_id(); },
714
17
                [](const FieldSchema* field) { return field->get_max_column_id(); },
715
17
                IcebergParquetNestedColumnUtils::extract_nested_column_ids);
716
17
    };
717
718
    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
719
    // table columns. When present, resolve each column by name through it so the column-id set
720
    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
721
    // otherwise fall back to matching by Iceberg field id.
722
25
    const auto* struct_node =
723
25
            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
724
725
35
    for (const auto* slot : tuple_descriptor->slots()) {
726
35
        const FieldSchema* field_schema = nullptr;
727
35
        if (struct_node != nullptr) {
728
            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
729
            // never registered as children, so check membership before querying: calling
730
            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
731
            // throws std::out_of_range from .at() in release builds.
732
22
            if (struct_node->get_children().contains(slot->col_name()) &&
733
22
                struct_node->children_column_exists(slot->col_name())) {
734
                // Use the physical child selected by the schema-mapping pass. This keeps partial-id
735
                // files in BY_NAME mode from binding a projected column through an unrelated stale
736
                // field id.
737
20
                const auto& file_column_name =
738
20
                        struct_node->children_file_column_name(slot->col_name());
739
25
                for (int i = 0; i < field_desc->size(); ++i) {
740
25
                    const auto* candidate = field_desc->get_column(i);
741
25
                    if (candidate != nullptr && candidate->name == file_column_name) {
742
20
                        field_schema = candidate;
743
20
                        break;
744
20
                    }
745
25
                }
746
20
                DORIS_CHECK(field_schema != nullptr);
747
20
            }
748
22
        } else {
749
13
            auto it = iceberg_id_to_field_schema_map.find(slot->col_unique_id());
750
13
            if (it != iceberg_id_to_field_schema_map.end()) {
751
13
                field_schema = it->second;
752
13
            }
753
13
        }
754
35
        if (field_schema == nullptr) {
755
2
            continue;
756
2
        }
757
758
33
        if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
759
33
             slot->col_type() != TYPE_MAP)) {
760
22
            column_ids.insert(field_schema->column_id);
761
22
            if (slot->is_predicate()) {
762
0
                filter_column_ids.insert(field_schema->column_id);
763
0
            }
764
22
            continue;
765
22
        }
766
767
11
        const auto& all_access_paths = slot->all_access_paths();
768
11
        process_access_paths(field_schema, all_access_paths, column_ids);
769
770
11
        const auto& predicate_access_paths = slot->predicate_access_paths();
771
11
        if (!predicate_access_paths.empty()) {
772
6
            process_access_paths(field_schema, predicate_access_paths, filter_column_ids);
773
6
        }
774
11
    }
775
25
    return {std::move(column_ids), std::move(filter_column_ids)};
776
25
}
777
778
// ============================================================================
779
// IcebergParquetReader: _read_position_delete_file
780
// ============================================================================
781
Status IcebergParquetReader::_read_position_delete_file(const TFileRangeDesc* delete_range,
782
2
                                                        DeleteFile* position_delete) {
783
2
    ParquetReader parquet_delete_reader(get_profile(), get_scan_params(), *delete_range,
784
2
                                        READ_DELETE_FILE_BATCH_SIZE, &get_state()->timezone_obj(),
785
2
                                        get_io_ctx(), get_state(), _meta_cache);
786
    // The delete file range has size=-1 (read whole file). We must disable
787
    // row group filtering before init; otherwise _do_init_reader returns EndOfFile
788
    // when _filter_groups && _range_size < 0.
789
2
    ParquetInitContext delete_ctx;
790
2
    delete_ctx.filter_groups = false;
791
2
    delete_ctx.column_names = delete_file_col_names;
792
2
    delete_ctx.col_name_to_block_idx =
793
2
            const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX);
794
2
    RETURN_IF_ERROR(parquet_delete_reader.init_reader(&delete_ctx));
795
796
0
    const tparquet::FileMetaData* meta_data = parquet_delete_reader.get_meta_data();
797
0
    bool dictionary_coded = true;
798
0
    for (const auto& row_group : meta_data->row_groups) {
799
0
        const auto& column_chunk = row_group.columns[ICEBERG_FILE_PATH_INDEX];
800
0
        if (!(column_chunk.__isset.meta_data && has_dict_page(column_chunk.meta_data))) {
801
0
            dictionary_coded = false;
802
0
            break;
803
0
        }
804
0
    }
805
0
    DataTypePtr data_type_file_path = make_nullable(std::make_shared<DataTypeString>());
806
0
    DataTypePtr data_type_pos = make_nullable(std::make_shared<DataTypeInt64>());
807
0
    bool eof = false;
808
0
    while (!eof) {
809
0
        Block block = {
810
0
                dictionary_coded
811
0
                        ? ColumnWithTypeAndName {ColumnNullable::create(ColumnDictI32::create(),
812
0
                                                                        ColumnUInt8::create()),
813
0
                                                 data_type_file_path, ICEBERG_FILE_PATH}
814
0
                        : ColumnWithTypeAndName {data_type_file_path, ICEBERG_FILE_PATH},
815
816
0
                {data_type_pos, ICEBERG_ROW_POS}};
817
0
        size_t read_rows = 0;
818
0
        RETURN_IF_ERROR(parquet_delete_reader.get_next_block(&block, &read_rows, &eof));
819
820
0
        if (read_rows <= 0) {
821
0
            break;
822
0
        }
823
0
        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows,
824
0
                                                        dictionary_coded));
825
0
    }
826
0
    return Status::OK();
827
0
};
828
829
// ============================================================================
830
// IcebergOrcReader: on_before_init_reader (ORC-specific schema matching)
831
// ============================================================================
832
// This format-specific setup mirrors the existing reader initialization sequence.
833
// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
834
16
Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
835
16
    _column_descs = ctx->column_descs;
836
16
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
837
16
    _file_format = Fileformat::ORC;
838
839
    // Get ORC file type first (available because _create_file_reader() already ran)
840
16
    const orc::Type* orc_type_ptr = nullptr;
841
16
    RETURN_IF_ERROR(this->get_file_type(&orc_type_ptr));
842
843
    // Build table_info_node by field_id or name matching.
844
    // This must happen BEFORE column classification so we can use children_column_exists
845
    // to check if a column exists in the file (by field ID, not name).
846
16
    if (!get_scan_params().__isset.history_schema_info ||
847
16
        get_scan_params().history_schema_info.empty()) [[unlikely]] {
848
1
        RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_name(ctx->tuple_descriptor, orc_type_ptr,
849
1
                                                        ctx->table_info_node));
850
15
    } else {
851
15
        RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_field_id_with_name_mapping(
852
15
                get_scan_params().history_schema_info.front().root_field, orc_type_ptr,
853
15
                ICEBERG_ORC_ATTRIBUTE, ctx->table_info_node,
854
15
                supports_iceberg_scan_semantics_v1(&get_scan_params())));
855
15
    }
856
857
16
    std::unordered_set<std::string> partition_col_names;
858
16
    if (ctx->range->__isset.columns_from_path_keys) {
859
0
        partition_col_names.insert(ctx->range->columns_from_path_keys.begin(),
860
0
                                   ctx->range->columns_from_path_keys.end());
861
0
    }
862
863
    // Single pass: classify columns, detect $row_id, handle partition fallback.
864
16
    bool has_partition_from_path = false;
865
19
    for (const auto& desc : *ctx->column_descs) {
866
19
        if (desc.category == ColumnCategory::SYNTHESIZED) {
867
0
            if (desc.name == BeConsts::ICEBERG_ROWID_COL) {
868
0
                this->register_synthesized_column_handler(
869
0
                        BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status {
870
0
                            return _fill_iceberg_row_id(block, rows);
871
0
                        });
872
0
                continue;
873
0
            } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
874
0
                auto topn_row_id_column_iter = _create_topn_row_id_column_iterator();
875
0
                this->register_synthesized_column_handler(
876
0
                        desc.name,
877
0
                        [iter = std::move(topn_row_id_column_iter), this, &desc](
878
0
                                Block* block, size_t rows) -> Status {
879
0
                            return fill_topn_row_id(iter, desc.name, block, rows);
880
0
                        });
881
0
                continue;
882
0
            }
883
19
        } else if (desc.category == ColumnCategory::PARTITION_KEY) {
884
0
            bool has_partition_value = partition_col_names.contains(desc.name);
885
0
            bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name);
886
0
            if (!has_partition_value || exists_in_file) {
887
0
                ctx->column_names.push_back(desc.name);
888
0
                continue;
889
0
            }
890
0
            has_partition_from_path = true;
891
19
        } else if (desc.category == ColumnCategory::REGULAR) {
892
19
            ctx->column_names.push_back(desc.name);
893
19
        } else if (desc.category == ColumnCategory::GENERATED) {
894
0
            _init_row_lineage_columns();
895
0
            if (desc.name == ROW_LINEAGE_ROW_ID) {
896
0
                ctx->column_names.push_back(desc.name);
897
0
                this->register_generated_column_handler(
898
0
                        ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status {
899
0
                            return _fill_row_lineage_row_id(block, rows);
900
0
                        });
901
0
                continue;
902
0
            } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
903
0
                ctx->column_names.push_back(desc.name);
904
0
                this->register_generated_column_handler(
905
0
                        ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER,
906
0
                        [this](Block* block, size_t rows) -> Status {
907
0
                            return _fill_row_lineage_last_updated_sequence_number(block, rows);
908
0
                        });
909
0
                continue;
910
0
            }
911
0
        }
912
19
    }
913
914
16
    if (has_partition_from_path) {
915
0
        RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
916
0
                                                  _fill_partition_values,
917
0
                                                  &_fill_partition_value_is_null));
918
0
    }
919
920
16
    _all_required_col_names = ctx->column_names;
921
922
    // Create column IDs from ORC type
923
16
    auto column_id_result =
924
16
            _create_column_ids(orc_type_ptr, ctx->tuple_descriptor, ctx->table_info_node);
925
16
    ctx->column_ids = std::move(column_id_result.column_ids);
926
16
    ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);
927
928
    // Build field_id -> block_column_name mapping for equality delete filtering.
929
19
    for (const auto* slot : ctx->tuple_descriptor->slots()) {
930
19
        _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name());
931
19
    }
932
933
    // Process delete files (must happen before _do_init_reader so expand col IDs are included)
934
16
    RETURN_IF_ERROR(_init_row_filters());
935
936
    // Add expand column IDs for equality delete and remap expand column names
937
    // (matching master's behavior with __equality_delete_column__ prefix)
938
16
    const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
939
16
    bool all_file_columns_have_field_ids = true;
940
52
    for (uint64_t i = 0; i < orc_type_ptr->getSubtypeCount(); ++i) {
941
36
        const orc::Type* sub_type = orc_type_ptr->getSubtype(i);
942
36
        if (!sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
943
16
            all_file_columns_have_field_ids = false;
944
16
        }
945
36
    }
946
16
    const bool use_field_ids_for_hidden_keys =
947
16
            supports_iceberg_scan_semantics_v1(&get_scan_params())
948
16
                    ? orc_subtree_has_iceberg_id(orc_type_ptr, ICEBERG_ORC_ATTRIBUTE)
949
16
                    : all_file_columns_have_field_ids;
950
16
    const auto find_file_column_by_name = [&](const std::string& name) -> const orc::Type* {
951
0
        for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
952
0
            if (iequal(orc_type_ptr->getFieldName(j), name)) {
953
0
                return orc_type_ptr->getSubtype(j);
954
0
            }
955
0
        }
956
0
        return nullptr;
957
0
    };
958
959
16
    std::vector<std::string> new_expand_col_names;
960
16
    DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
961
16
    DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
962
30
    for (size_t i = 0; i < _expand_col_names.size(); ++i) {
963
14
        const auto& old_name = _expand_col_names[i];
964
14
        const int32_t field_id = _expand_col_field_ids[i];
965
966
14
        const orc::Type* file_column = nullptr;
967
14
        OrcEqualityFieldPath file_path;
968
14
        bool complete_file_path = false;
969
14
        if (use_field_ids_for_hidden_keys) {
970
8
            complete_file_path =
971
8
                    find_orc_equality_field_path_by_id(orc_type_ptr, field_id, &file_path);
972
8
            if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) {
973
4
                const auto table_path = _find_schema_field_path(field_id);
974
4
                if (!table_path.empty()) {
975
4
                    complete_file_path = find_orc_equality_field_prefix_by_id_path(
976
4
                            orc_type_ptr, table_path, &file_path);
977
4
                }
978
4
            }
979
8
            if (!file_path.fields.empty()) {
980
4
                file_column = file_path.fields.front();
981
4
            }
982
8
        } else {
983
6
            const auto table_path = _find_schema_field_path(field_id);
984
6
            if (!table_path.empty()) {
985
6
                complete_file_path = find_orc_equality_field_prefix_by_name_path(
986
6
                        orc_type_ptr, table_path, old_name, &file_path);
987
6
                if (!file_path.fields.empty()) {
988
5
                    file_column = file_path.fields.front();
989
5
                }
990
6
            } else {
991
0
                file_column = find_file_column_by_name(old_name);
992
0
                complete_file_path = file_column != nullptr;
993
0
            }
994
6
        }
995
996
14
        std::string file_col_name = old_name;
997
14
        std::string leaf_name = old_name;
998
14
        if (!file_path.fields.empty()) {
999
9
            file_col_name = file_path.names.front();
1000
9
            leaf_name = file_path.names.back();
1001
9
        } else if (file_column != nullptr) {
1002
0
            for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
1003
0
                if (orc_type_ptr->getSubtype(j) == file_column) {
1004
0
                    file_col_name = orc_type_ptr->getFieldName(j);
1005
0
                    leaf_name = file_col_name;
1006
0
                    break;
1007
0
                }
1008
0
            }
1009
0
        }
1010
14
        std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name;
1011
1012
14
        if (field_id >= 0) {
1013
14
            _id_to_block_column_name[field_id] = table_col_name;
1014
14
        }
1015
14
        _expand_columns[i].name = table_col_name;
1016
14
        if (file_column == nullptr) {
1017
5
            RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
1018
5
                                                                     _expand_columns[i].type));
1019
            // The old data file predates this equality key. Keep it in the expand block so the
1020
            // synthesized-column hook can materialize its logical initial default before ORC's
1021
            // block-size checks. Adding it to column_names/table_info_node would mark it as an
1022
            // existing ORC child and make OrcReader read a column that is not present in the file.
1023
5
            new_expand_col_names.push_back(table_col_name);
1024
5
            continue;
1025
5
        }
1026
9
        new_expand_col_names.push_back(table_col_name);
1027
1028
9
        if (!complete_file_path) {
1029
1
            ColumnPtr missing_value;
1030
1
            RETURN_IF_ERROR(_create_missing_equality_delete_value(
1031
1
                    field_id, _expand_columns[i].type, file_path.fields.size(), &missing_value));
1032
1
            _nested_equality_delete_columns.push_back({
1033
1
                    .field_id = field_id,
1034
1
                    .block_name = table_col_name,
1035
1
                    .leaf_type = _expand_columns[i].type,
1036
1
                    .child_indexes = file_path.child_indexes,
1037
1
                    .missing_value = std::move(missing_value),
1038
1
            });
1039
1
            _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column));
1040
1
            _expand_columns[i].column = _expand_columns[i].type->create_column();
1041
8
        } else if (!file_path.child_indexes.empty()) {
1042
6
            _nested_equality_delete_columns.push_back({
1043
6
                    .field_id = field_id,
1044
6
                    .block_name = table_col_name,
1045
6
                    .leaf_type = _expand_columns[i].type,
1046
6
                    .child_indexes = file_path.child_indexes,
1047
6
                    .missing_value = nullptr,
1048
6
            });
1049
6
            _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column));
1050
6
            _expand_columns[i].column = _expand_columns[i].type->create_column();
1051
6
        }
1052
1053
9
        for (uint64_t column_id = file_column->getColumnId();
1054
25
             column_id <= file_column->getMaximumColumnId(); ++column_id) {
1055
16
            ctx->column_ids.insert(column_id);
1056
16
        }
1057
1058
9
        ctx->column_names.push_back(table_col_name);
1059
9
        ctx->table_info_node->add_children(table_col_name, file_col_name,
1060
9
                                           TableSchemaChangeHelper::ConstNode::get_instance());
1061
9
    }
1062
16
    _expand_col_names = std::move(new_expand_col_names);
1063
1064
16
    return Status::OK();
1065
16
}
1066
1067
// ============================================================================
1068
// IcebergOrcReader: _create_column_ids
1069
// ============================================================================
1070
ColumnIdResult IcebergOrcReader::_create_column_ids(
1071
        const orc::Type* orc_type, const TupleDescriptor* tuple_descriptor,
1072
23
        const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
1073
23
    std::unordered_map<int, const orc::Type*> iceberg_id_to_orc_type_map;
1074
109
    for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
1075
86
        const auto* orc_sub_type = orc_type->getSubtype(i);
1076
86
        if (!orc_sub_type) {
1077
0
            continue;
1078
0
        }
1079
86
        if (!orc_sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
1080
17
            continue;
1081
17
        }
1082
69
        int iceberg_id = std::stoi(orc_sub_type->getAttributeValue(ICEBERG_ORC_ATTRIBUTE));
1083
69
        iceberg_id_to_orc_type_map[iceberg_id] = orc_sub_type;
1084
69
    }
1085
1086
23
    std::set<uint64_t> column_ids;
1087
23
    std::set<uint64_t> filter_column_ids;
1088
1089
23
    auto process_access_paths = [](const orc::Type* orc_field,
1090
23
                                   const std::vector<TColumnAccessPath>& access_paths,
1091
23
                                   std::set<uint64_t>& out_ids) {
1092
17
        process_nested_access_paths(
1093
17
                orc_field, access_paths, out_ids,
1094
17
                [](const orc::Type* type) { return type->getColumnId(); },
1095
17
                [](const orc::Type* type) { return type->getMaximumColumnId(); },
1096
17
                IcebergOrcNestedColumnUtils::extract_nested_column_ids);
1097
17
    };
1098
1099
    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
1100
    // table columns. When present, resolve each column by name through it so the column-id set
1101
    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
1102
    // otherwise fall back to matching by Iceberg field id.
1103
23
    const auto* struct_node =
1104
23
            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
1105
1106
34
    for (const auto* slot : tuple_descriptor->slots()) {
1107
34
        const orc::Type* orc_field = nullptr;
1108
34
        if (struct_node != nullptr) {
1109
            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
1110
            // never registered as children, so check membership before querying: calling
1111
            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
1112
            // throws std::out_of_range from .at() in release builds.
1113
21
            if (struct_node->get_children().contains(slot->col_name()) &&
1114
21
                struct_node->children_column_exists(slot->col_name())) {
1115
                // Select the physical child resolved by the shared schema-mapping pass. Hidden
1116
                // equality keys and projected columns must obey the same BY_NAME decision for
1117
                // partial-id ORC files.
1118
19
                const auto& file_column_name =
1119
19
                        struct_node->children_file_column_name(slot->col_name());
1120
23
                for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
1121
23
                    if (orc_type->getFieldName(i) == file_column_name) {
1122
19
                        orc_field = orc_type->getSubtype(i);
1123
19
                        break;
1124
19
                    }
1125
23
                }
1126
19
                DORIS_CHECK(orc_field != nullptr);
1127
19
            }
1128
21
        } else {
1129
13
            auto it = iceberg_id_to_orc_type_map.find(slot->col_unique_id());
1130
13
            if (it != iceberg_id_to_orc_type_map.end()) {
1131
13
                orc_field = it->second;
1132
13
            }
1133
13
        }
1134
34
        if (orc_field == nullptr) {
1135
2
            continue;
1136
2
        }
1137
1138
32
        if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
1139
32
             slot->col_type() != TYPE_MAP)) {
1140
21
            column_ids.insert(orc_field->getColumnId());
1141
21
            if (slot->is_predicate()) {
1142
0
                filter_column_ids.insert(orc_field->getColumnId());
1143
0
            }
1144
21
            continue;
1145
21
        }
1146
1147
11
        const auto& all_access_paths = slot->all_access_paths();
1148
11
        process_access_paths(orc_field, all_access_paths, column_ids);
1149
1150
11
        const auto& predicate_access_paths = slot->predicate_access_paths();
1151
11
        if (!predicate_access_paths.empty()) {
1152
6
            process_access_paths(orc_field, predicate_access_paths, filter_column_ids);
1153
6
        }
1154
11
    }
1155
1156
23
    return {std::move(column_ids), std::move(filter_column_ids)};
1157
23
}
1158
1159
// ============================================================================
1160
// IcebergOrcReader: _read_position_delete_file
1161
// ============================================================================
1162
Status IcebergOrcReader::_read_position_delete_file(const TFileRangeDesc* delete_range,
1163
0
                                                    DeleteFile* position_delete) {
1164
0
    OrcReader orc_delete_reader(get_profile(), get_state(), get_scan_params(), *delete_range,
1165
0
                                READ_DELETE_FILE_BATCH_SIZE, get_state()->timezone(), get_io_ctx(),
1166
0
                                _meta_cache);
1167
0
    OrcInitContext delete_ctx;
1168
0
    delete_ctx.column_names = delete_file_col_names;
1169
0
    delete_ctx.col_name_to_block_idx =
1170
0
            const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX);
1171
0
    RETURN_IF_ERROR(orc_delete_reader.init_reader(&delete_ctx));
1172
1173
0
    bool eof = false;
1174
0
    DataTypePtr data_type_file_path {new DataTypeString};
1175
0
    DataTypePtr data_type_pos {new DataTypeInt64};
1176
0
    while (!eof) {
1177
0
        Block block = {{data_type_file_path, ICEBERG_FILE_PATH}, {data_type_pos, ICEBERG_ROW_POS}};
1178
1179
0
        size_t read_rows = 0;
1180
0
        RETURN_IF_ERROR(orc_delete_reader.get_next_block(&block, &read_rows, &eof));
1181
1182
0
        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows, false));
1183
0
    }
1184
0
    return Status::OK();
1185
0
}
1186
1187
} // namespace doris