Coverage Report

Created: 2026-08-13 12:11

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
60
bool parquet_subtree_has_iceberg_id(const FieldSchema& field) {
96
60
    if (field.field_id >= 0) {
97
26
        return true;
98
26
    }
99
34
    return std::ranges::any_of(field.children, parquet_subtree_has_iceberg_id);
100
60
}
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
20
Status IcebergParquetReader::on_before_init_reader(ReaderInitContext* ctx) {
435
20
    _column_descs = ctx->column_descs;
436
20
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
437
20
    _file_format = Fileformat::PARQUET;
438
439
    // Get file metadata schema first (available because _open_file() already ran)
440
20
    const FieldDescriptor* field_desc = nullptr;
441
20
    RETURN_IF_ERROR(this->get_file_metadata_schema(&field_desc));
442
20
    DCHECK(field_desc != nullptr);
443
444
    // Build table_info_node by field_id or name matching.
445
    // This must happen BEFORE column classification so we can use children_column_exists
446
    // to check if a column exists in the file (by field ID, not name).
447
20
    if (!get_scan_params().__isset.history_schema_info ||
448
20
        get_scan_params().history_schema_info.empty()) [[unlikely]] {
449
1
        RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_name(ctx->tuple_descriptor, *field_desc,
450
1
                                                            ctx->table_info_node));
451
19
    } else {
452
19
        RETURN_IF_ERROR(BuildTableInfoUtil::by_parquet_field_id_with_name_mapping(
453
19
                get_scan_params().history_schema_info.front().root_field, *field_desc,
454
19
                ctx->table_info_node, supports_iceberg_scan_semantics_v1(&get_scan_params())));
455
19
    }
456
457
20
    std::unordered_set<std::string> partition_col_names;
458
20
    if (ctx->range->__isset.columns_from_path_keys) {
459
0
        partition_col_names.insert(ctx->range->columns_from_path_keys.begin(),
460
0
                                   ctx->range->columns_from_path_keys.end());
461
0
    }
462
463
    // Single pass: classify columns, detect $row_id, handle partition fallback.
464
20
    bool has_partition_from_path = false;
465
22
    for (const auto& desc : *ctx->column_descs) {
466
22
        if (desc.category == ColumnCategory::SYNTHESIZED) {
467
0
            if (desc.name == BeConsts::ICEBERG_ROWID_COL) {
468
0
                this->register_synthesized_column_handler(
469
0
                        BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status {
470
0
                            return _fill_iceberg_row_id(block, rows);
471
0
                        });
472
0
                continue;
473
0
            } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
474
0
                auto topn_row_id_column_iter = _create_topn_row_id_column_iterator();
475
0
                this->register_synthesized_column_handler(
476
0
                        desc.name,
477
0
                        [iter = std::move(topn_row_id_column_iter), this, &desc](
478
0
                                Block* block, size_t rows) -> Status {
479
0
                            return fill_topn_row_id(iter, desc.name, block, rows);
480
0
                        });
481
0
                continue;
482
0
            }
483
22
        } else if (desc.category == ColumnCategory::PARTITION_KEY) {
484
0
            bool has_partition_value = partition_col_names.contains(desc.name);
485
            // A projected column absent from the table-side schema tree means the schema info
486
            // from FE is inconsistent with the scan projection. Fail this query loudly instead
487
            // of aborting the whole BE process via children_column_exists's std::out_of_range
488
            // (release) or DCHECK (debug). See #61225.
489
0
            if (!ctx->table_info_node->has_children_column(desc.name)) {
490
0
                return Status::InternalError(
491
0
                        "schema mapping is missing projected column '{}'; the schema info from FE "
492
0
                        "is inconsistent with the scan projection (file: {})",
493
0
                        desc.name, ctx->range->path);
494
0
            }
495
0
            bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name);
496
0
            if (!has_partition_value || exists_in_file) {
497
                // Keep PARTITION_KEY category stable for scan planning, but still read
498
                // from file when the column exists there.
499
0
                ctx->column_names.push_back(desc.name);
500
0
                continue;
501
0
            }
502
0
            has_partition_from_path = true;
503
22
        } else if (desc.category == ColumnCategory::REGULAR) {
504
22
            if (!ctx->table_info_node->has_children_column(desc.name)) {
505
1
                return Status::InternalError(
506
1
                        "schema mapping is missing projected column '{}'; the schema info from FE "
507
1
                        "is inconsistent with the scan projection (file: {})",
508
1
                        desc.name, ctx->range->path);
509
1
            }
510
21
            ctx->column_names.push_back(desc.name);
511
21
        } else if (desc.category == ColumnCategory::GENERATED) {
512
0
            _init_row_lineage_columns();
513
0
            if (desc.name == ROW_LINEAGE_ROW_ID) {
514
0
                ctx->column_names.push_back(desc.name);
515
0
                this->register_generated_column_handler(
516
0
                        ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status {
517
0
                            return _fill_row_lineage_row_id(block, rows);
518
0
                        });
519
0
                continue;
520
0
            } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
521
0
                ctx->column_names.push_back(desc.name);
522
0
                this->register_generated_column_handler(
523
0
                        ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER,
524
0
                        [this](Block* block, size_t rows) -> Status {
525
0
                            return _fill_row_lineage_last_updated_sequence_number(block, rows);
526
0
                        });
527
0
                continue;
528
0
            }
529
0
        }
530
22
    }
531
532
    // Set up partition value extraction if any partition columns need filling from path
533
19
    if (has_partition_from_path) {
534
0
        RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
535
0
                                                  _fill_partition_values,
536
0
                                                  &_fill_partition_value_is_null));
537
0
    }
538
539
19
    _all_required_col_names = ctx->column_names;
540
541
    // Create column IDs from field descriptor
542
19
    auto column_id_result =
543
19
            _create_column_ids(field_desc, ctx->tuple_descriptor, ctx->table_info_node);
544
19
    ctx->column_ids = std::move(column_id_result.column_ids);
545
19
    ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);
546
547
    // Build field_id -> block_column_name mapping for equality delete filtering.
548
    // This was previously done in init_reader() column matching (pre-CRTP refactoring).
549
22
    for (const auto* slot : ctx->tuple_descriptor->slots()) {
550
22
        _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name());
551
22
    }
552
553
    // Process delete files (must happen before _do_init_reader so expand col IDs are included)
554
19
    RETURN_IF_ERROR(_init_row_filters());
555
556
    // Add expand column IDs for equality delete and remap expand column names
557
    // to match master's behavior:
558
    // - Use field_id to find the actual file column name in Parquet schema
559
    // - Prefix with __equality_delete_column__ to avoid name conflicts
560
    // - Correctly map table_col_name → file_col_name in table_info_node
561
19
    const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
562
19
    bool all_file_columns_have_field_ids = true;
563
19
    bool any_file_column_has_field_id = false;
564
72
    for (int i = 0; i < field_desc->size(); ++i) {
565
53
        const auto* field_schema = field_desc->get_column(i);
566
53
        if (field_schema) {
567
53
            if (field_schema->field_id < 0) {
568
30
                all_file_columns_have_field_ids = false;
569
30
            }
570
53
            if (parquet_subtree_has_iceberg_id(*field_schema)) {
571
26
                any_file_column_has_field_id = true;
572
26
            }
573
53
        }
574
53
    }
575
19
    const bool use_field_ids_for_hidden_keys =
576
19
            supports_iceberg_scan_semantics_v1(&get_scan_params())
577
19
                    ? any_file_column_has_field_id
578
19
                    : all_file_columns_have_field_ids;
579
19
    const auto find_file_column_by_name = [&](const std::string& name) -> const FieldSchema* {
580
0
        for (int j = 0; j < field_desc->size(); ++j) {
581
0
            const auto* candidate = field_desc->get_column(j);
582
0
            if (candidate != nullptr && iequal(candidate->name, name)) {
583
0
                return candidate;
584
0
            }
585
0
        }
586
0
        return nullptr;
587
0
    };
588
589
    // Rebuild _expand_col_names with proper file-column-based names
590
19
    std::vector<std::string> new_expand_col_names;
591
19
    DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
592
19
    DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
593
34
    for (size_t i = 0; i < _expand_col_names.size(); ++i) {
594
15
        const auto& old_name = _expand_col_names[i];
595
15
        const int32_t field_id = _expand_col_field_ids[i];
596
597
15
        const FieldSchema* file_column = nullptr;
598
15
        ParquetEqualityFieldPath file_path;
599
15
        bool complete_file_path = false;
600
15
        if (use_field_ids_for_hidden_keys) {
601
8
            complete_file_path =
602
8
                    find_parquet_equality_field_path_by_id(field_desc, field_id, &file_path);
603
8
            if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) {
604
4
                const auto table_path = _find_schema_field_path(field_id);
605
4
                if (!table_path.empty()) {
606
4
                    complete_file_path = find_parquet_equality_field_prefix_by_id_path(
607
4
                            field_desc, table_path, &file_path);
608
4
                }
609
4
            }
610
8
            if (!file_path.fields.empty()) {
611
5
                file_column = file_path.fields.front();
612
5
            }
613
8
        } else {
614
7
            const auto table_path = _find_schema_field_path(field_id);
615
7
            if (!table_path.empty()) {
616
7
                complete_file_path = find_parquet_equality_field_prefix_by_name_path(
617
7
                        field_desc, table_path, old_name, &file_path);
618
7
                if (!file_path.fields.empty()) {
619
6
                    file_column = file_path.fields.front();
620
6
                }
621
7
            } else {
622
0
                file_column = find_file_column_by_name(old_name);
623
0
                complete_file_path = file_column != nullptr;
624
0
            }
625
7
        }
626
627
15
        std::string leaf_name = old_name;
628
15
        if (!file_path.fields.empty()) {
629
11
            leaf_name = file_path.fields.back()->name;
630
11
        } else if (file_column != nullptr) {
631
0
            leaf_name = file_column->name;
632
0
        }
633
15
        const std::string file_col_name = file_column == nullptr ? old_name : file_column->name;
634
15
        std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name;
635
636
        // Update _id_to_block_column_name
637
15
        if (field_id >= 0) {
638
15
            _id_to_block_column_name[field_id] = table_col_name;
639
15
        }
640
641
        // Update _expand_columns name
642
15
        _expand_columns[i].name = table_col_name;
643
644
15
        if (file_column == nullptr) {
645
4
            RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
646
4
                                                                     _expand_columns[i].type));
647
            // The old data file predates this equality key. Keep it in the expand block so the
648
            // synthesized-column hook can materialize its logical initial default before reader
649
            // filtering, but do not advertise it to Parquet as a physical child.
650
4
            new_expand_col_names.push_back(table_col_name);
651
4
            continue;
652
4
        }
653
654
11
        new_expand_col_names.push_back(table_col_name);
655
656
11
        if (!complete_file_path) {
657
1
            ColumnPtr missing_value;
658
1
            RETURN_IF_ERROR(_create_missing_equality_delete_value(
659
1
                    field_id, _expand_columns[i].type, file_path.fields.size(), &missing_value));
660
1
            _nested_equality_delete_columns.push_back({
661
1
                    .field_id = field_id,
662
1
                    .block_name = table_col_name,
663
1
                    .leaf_type = _expand_columns[i].type,
664
1
                    .child_indexes = file_path.child_indexes,
665
1
                    .missing_value = std::move(missing_value),
666
1
            });
667
1
            _expand_columns[i].type = make_nullable(file_column->data_type);
668
1
            _expand_columns[i].column = _expand_columns[i].type->create_column();
669
10
        } else if (!file_path.child_indexes.empty()) {
670
7
            _nested_equality_delete_columns.push_back({
671
7
                    .field_id = field_id,
672
7
                    .block_name = table_col_name,
673
7
                    .leaf_type = _expand_columns[i].type,
674
7
                    .child_indexes = file_path.child_indexes,
675
7
                    .missing_value = nullptr,
676
7
            });
677
7
            _expand_columns[i].type = make_nullable(file_column->data_type);
678
7
            _expand_columns[i].column = _expand_columns[i].type->create_column();
679
7
        }
680
681
        // A hidden nested key is read through its containing top-level struct. V1 column IDs are
682
        // pre-order ranges, so include the complete subtree before extracting the primitive leaf.
683
11
        for (uint64_t column_id = file_column->get_column_id();
684
30
             column_id <= file_column->get_max_column_id(); ++column_id) {
685
19
            ctx->column_ids.insert(column_id);
686
19
        }
687
688
        // Register in table_info_node: table_col_name → file_col_name
689
11
        ctx->column_names.push_back(table_col_name);
690
11
        ctx->table_info_node->add_children(table_col_name, file_col_name,
691
11
                                           TableSchemaChangeHelper::ConstNode::get_instance());
692
11
    }
693
19
    _expand_col_names = std::move(new_expand_col_names);
694
695
    // Enable group filtering for Iceberg
696
19
    _filter_groups = true;
697
698
19
    return Status::OK();
699
19
}
700
701
// ============================================================================
702
// IcebergParquetReader: _create_column_ids
703
// ============================================================================
704
ColumnIdResult IcebergParquetReader::_create_column_ids(
705
        const FieldDescriptor* field_desc, const TupleDescriptor* tuple_descriptor,
706
26
        const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
707
26
    auto* mutable_field_desc = const_cast<FieldDescriptor*>(field_desc);
708
26
    mutable_field_desc->assign_ids();
709
710
26
    std::unordered_map<int, const FieldSchema*> iceberg_id_to_field_schema_map;
711
129
    for (int i = 0; i < field_desc->size(); ++i) {
712
103
        const auto* field_schema = field_desc->get_column(i);
713
103
        if (!field_schema) {
714
0
            continue;
715
0
        }
716
103
        int iceberg_id = field_schema->field_id;
717
103
        iceberg_id_to_field_schema_map[iceberg_id] = field_schema;
718
103
    }
719
720
26
    std::set<uint64_t> column_ids;
721
26
    std::set<uint64_t> filter_column_ids;
722
723
26
    auto process_access_paths = [](const FieldSchema* parquet_field,
724
26
                                   const std::vector<TColumnAccessPath>& access_paths,
725
26
                                   std::set<uint64_t>& out_ids) {
726
17
        process_nested_access_paths(
727
17
                parquet_field, access_paths, out_ids,
728
17
                [](const FieldSchema* field) { return field->get_column_id(); },
729
17
                [](const FieldSchema* field) { return field->get_max_column_id(); },
730
17
                IcebergParquetNestedColumnUtils::extract_nested_column_ids);
731
17
    };
732
733
    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
734
    // table columns. When present, resolve each column by name through it so the column-id set
735
    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
736
    // otherwise fall back to matching by Iceberg field id.
737
26
    const auto* struct_node =
738
26
            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
739
740
37
    for (const auto* slot : tuple_descriptor->slots()) {
741
37
        const FieldSchema* field_schema = nullptr;
742
37
        if (struct_node != nullptr) {
743
            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
744
            // never registered as children, so check membership before querying: calling
745
            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
746
            // throws std::out_of_range from .at() in release builds.
747
24
            if (struct_node->get_children().contains(slot->col_name()) &&
748
24
                struct_node->children_column_exists(slot->col_name())) {
749
                // Use the physical child selected by the schema-mapping pass. This keeps partial-id
750
                // files in BY_NAME mode from binding a projected column through an unrelated stale
751
                // field id.
752
21
                const auto& file_column_name =
753
21
                        struct_node->children_file_column_name(slot->col_name());
754
28
                for (int i = 0; i < field_desc->size(); ++i) {
755
28
                    const auto* candidate = field_desc->get_column(i);
756
28
                    if (candidate != nullptr && candidate->name == file_column_name) {
757
21
                        field_schema = candidate;
758
21
                        break;
759
21
                    }
760
28
                }
761
21
                DORIS_CHECK(field_schema != nullptr);
762
21
            }
763
24
        } else {
764
13
            auto it = iceberg_id_to_field_schema_map.find(slot->col_unique_id());
765
13
            if (it != iceberg_id_to_field_schema_map.end()) {
766
13
                field_schema = it->second;
767
13
            }
768
13
        }
769
37
        if (field_schema == nullptr) {
770
3
            continue;
771
3
        }
772
773
34
        if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
774
34
             slot->col_type() != TYPE_MAP)) {
775
23
            column_ids.insert(field_schema->column_id);
776
23
            if (slot->is_predicate()) {
777
0
                filter_column_ids.insert(field_schema->column_id);
778
0
            }
779
23
            continue;
780
23
        }
781
782
11
        const auto& all_access_paths = slot->all_access_paths();
783
11
        process_access_paths(field_schema, all_access_paths, column_ids);
784
785
11
        const auto& predicate_access_paths = slot->predicate_access_paths();
786
11
        if (!predicate_access_paths.empty()) {
787
6
            process_access_paths(field_schema, predicate_access_paths, filter_column_ids);
788
6
        }
789
11
    }
790
26
    return {std::move(column_ids), std::move(filter_column_ids)};
791
26
}
792
793
// ============================================================================
794
// IcebergParquetReader: _read_position_delete_file
795
// ============================================================================
796
Status IcebergParquetReader::_read_position_delete_file(const TFileRangeDesc* delete_range,
797
1
                                                        DeleteFile* position_delete) {
798
1
    ParquetReader parquet_delete_reader(get_profile(), get_scan_params(), *delete_range,
799
1
                                        READ_DELETE_FILE_BATCH_SIZE, &get_state()->timezone_obj(),
800
1
                                        get_io_ctx(), get_state(), _meta_cache);
801
    // The delete file range has size=-1 (read whole file). We must disable
802
    // row group filtering before init; otherwise _do_init_reader returns EndOfFile
803
    // when _filter_groups && _range_size < 0.
804
1
    ParquetInitContext delete_ctx;
805
1
    delete_ctx.filter_groups = false;
806
1
    delete_ctx.column_names = delete_file_col_names;
807
1
    delete_ctx.col_name_to_block_idx =
808
1
            const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX);
809
1
    RETURN_IF_ERROR(parquet_delete_reader.init_reader(&delete_ctx));
810
811
0
    const tparquet::FileMetaData* meta_data = parquet_delete_reader.get_meta_data();
812
0
    bool dictionary_coded = true;
813
0
    for (const auto& row_group : meta_data->row_groups) {
814
0
        const auto& column_chunk = row_group.columns[ICEBERG_FILE_PATH_INDEX];
815
0
        if (!(column_chunk.__isset.meta_data && has_dict_page(column_chunk.meta_data))) {
816
0
            dictionary_coded = false;
817
0
            break;
818
0
        }
819
0
    }
820
0
    DataTypePtr data_type_file_path = make_nullable(std::make_shared<DataTypeString>());
821
0
    DataTypePtr data_type_pos = make_nullable(std::make_shared<DataTypeInt64>());
822
0
    bool eof = false;
823
0
    while (!eof) {
824
0
        Block block = {
825
0
                dictionary_coded
826
0
                        ? ColumnWithTypeAndName {ColumnNullable::create(ColumnDictI32::create(),
827
0
                                                                        ColumnUInt8::create()),
828
0
                                                 data_type_file_path, ICEBERG_FILE_PATH}
829
0
                        : ColumnWithTypeAndName {data_type_file_path, ICEBERG_FILE_PATH},
830
831
0
                {data_type_pos, ICEBERG_ROW_POS}};
832
0
        size_t read_rows = 0;
833
0
        RETURN_IF_ERROR(parquet_delete_reader.get_next_block(&block, &read_rows, &eof));
834
835
0
        if (read_rows <= 0) {
836
0
            break;
837
0
        }
838
0
        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows,
839
0
                                                        dictionary_coded));
840
0
    }
841
0
    return Status::OK();
842
0
};
843
844
// ============================================================================
845
// IcebergOrcReader: on_before_init_reader (ORC-specific schema matching)
846
// ============================================================================
847
// This format-specific setup mirrors the existing reader initialization sequence.
848
// NOLINTNEXTLINE(readability-function-cognitive-complexity,readability-function-size)
849
16
Status IcebergOrcReader::on_before_init_reader(ReaderInitContext* ctx) {
850
16
    _column_descs = ctx->column_descs;
851
16
    _fill_col_name_to_block_idx = ctx->col_name_to_block_idx;
852
16
    _file_format = Fileformat::ORC;
853
854
    // Get ORC file type first (available because _create_file_reader() already ran)
855
16
    const orc::Type* orc_type_ptr = nullptr;
856
16
    RETURN_IF_ERROR(this->get_file_type(&orc_type_ptr));
857
858
    // Build table_info_node by field_id or name matching.
859
    // This must happen BEFORE column classification so we can use children_column_exists
860
    // to check if a column exists in the file (by field ID, not name).
861
16
    if (!get_scan_params().__isset.history_schema_info ||
862
16
        get_scan_params().history_schema_info.empty()) [[unlikely]] {
863
1
        RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_name(ctx->tuple_descriptor, orc_type_ptr,
864
1
                                                        ctx->table_info_node));
865
15
    } else {
866
15
        RETURN_IF_ERROR(BuildTableInfoUtil::by_orc_field_id_with_name_mapping(
867
15
                get_scan_params().history_schema_info.front().root_field, orc_type_ptr,
868
15
                ICEBERG_ORC_ATTRIBUTE, ctx->table_info_node,
869
15
                supports_iceberg_scan_semantics_v1(&get_scan_params())));
870
15
    }
871
872
16
    std::unordered_set<std::string> partition_col_names;
873
16
    if (ctx->range->__isset.columns_from_path_keys) {
874
0
        partition_col_names.insert(ctx->range->columns_from_path_keys.begin(),
875
0
                                   ctx->range->columns_from_path_keys.end());
876
0
    }
877
878
    // Single pass: classify columns, detect $row_id, handle partition fallback.
879
16
    bool has_partition_from_path = false;
880
19
    for (const auto& desc : *ctx->column_descs) {
881
19
        if (desc.category == ColumnCategory::SYNTHESIZED) {
882
0
            if (desc.name == BeConsts::ICEBERG_ROWID_COL) {
883
0
                this->register_synthesized_column_handler(
884
0
                        BeConsts::ICEBERG_ROWID_COL, [this](Block* block, size_t rows) -> Status {
885
0
                            return _fill_iceberg_row_id(block, rows);
886
0
                        });
887
0
                continue;
888
0
            } else if (desc.name.starts_with(BeConsts::GLOBAL_ROWID_COL)) {
889
0
                auto topn_row_id_column_iter = _create_topn_row_id_column_iterator();
890
0
                this->register_synthesized_column_handler(
891
0
                        desc.name,
892
0
                        [iter = std::move(topn_row_id_column_iter), this, &desc](
893
0
                                Block* block, size_t rows) -> Status {
894
0
                            return fill_topn_row_id(iter, desc.name, block, rows);
895
0
                        });
896
0
                continue;
897
0
            }
898
19
        } else if (desc.category == ColumnCategory::PARTITION_KEY) {
899
0
            bool has_partition_value = partition_col_names.contains(desc.name);
900
            // A projected column absent from the table-side schema tree means the schema info
901
            // from FE is inconsistent with the scan projection. Fail this query loudly instead
902
            // of aborting the whole BE process via children_column_exists's std::out_of_range
903
            // (release) or DCHECK (debug). See #61225.
904
0
            if (!ctx->table_info_node->has_children_column(desc.name)) {
905
0
                return Status::InternalError(
906
0
                        "schema mapping is missing projected column '{}'; the schema info from FE "
907
0
                        "is inconsistent with the scan projection (file: {})",
908
0
                        desc.name, ctx->range->path);
909
0
            }
910
0
            bool exists_in_file = ctx->table_info_node->children_column_exists(desc.name);
911
0
            if (!has_partition_value || exists_in_file) {
912
0
                ctx->column_names.push_back(desc.name);
913
0
                continue;
914
0
            }
915
0
            has_partition_from_path = true;
916
19
        } else if (desc.category == ColumnCategory::REGULAR) {
917
19
            if (!ctx->table_info_node->has_children_column(desc.name)) {
918
0
                return Status::InternalError(
919
0
                        "schema mapping is missing projected column '{}'; the schema info from FE "
920
0
                        "is inconsistent with the scan projection (file: {})",
921
0
                        desc.name, ctx->range->path);
922
0
            }
923
19
            ctx->column_names.push_back(desc.name);
924
19
        } else if (desc.category == ColumnCategory::GENERATED) {
925
0
            _init_row_lineage_columns();
926
0
            if (desc.name == ROW_LINEAGE_ROW_ID) {
927
0
                ctx->column_names.push_back(desc.name);
928
0
                this->register_generated_column_handler(
929
0
                        ROW_LINEAGE_ROW_ID, [this](Block* block, size_t rows) -> Status {
930
0
                            return _fill_row_lineage_row_id(block, rows);
931
0
                        });
932
0
                continue;
933
0
            } else if (desc.name == ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER) {
934
0
                ctx->column_names.push_back(desc.name);
935
0
                this->register_generated_column_handler(
936
0
                        ROW_LINEAGE_LAST_UPDATED_SEQ_NUMBER,
937
0
                        [this](Block* block, size_t rows) -> Status {
938
0
                            return _fill_row_lineage_last_updated_sequence_number(block, rows);
939
0
                        });
940
0
                continue;
941
0
            }
942
0
        }
943
19
    }
944
945
16
    if (has_partition_from_path) {
946
0
        RETURN_IF_ERROR(_extract_partition_values(*ctx->range, ctx->tuple_descriptor,
947
0
                                                  _fill_partition_values,
948
0
                                                  &_fill_partition_value_is_null));
949
0
    }
950
951
16
    _all_required_col_names = ctx->column_names;
952
953
    // Create column IDs from ORC type
954
16
    auto column_id_result =
955
16
            _create_column_ids(orc_type_ptr, ctx->tuple_descriptor, ctx->table_info_node);
956
16
    ctx->column_ids = std::move(column_id_result.column_ids);
957
16
    ctx->filter_column_ids = std::move(column_id_result.filter_column_ids);
958
959
    // Build field_id -> block_column_name mapping for equality delete filtering.
960
19
    for (const auto* slot : ctx->tuple_descriptor->slots()) {
961
19
        _id_to_block_column_name.emplace(slot->col_unique_id(), slot->col_name());
962
19
    }
963
964
    // Process delete files (must happen before _do_init_reader so expand col IDs are included)
965
16
    RETURN_IF_ERROR(_init_row_filters());
966
967
    // Add expand column IDs for equality delete and remap expand column names
968
    // (matching master's behavior with __equality_delete_column__ prefix)
969
16
    const static std::string EQ_DELETE_PRE = "__equality_delete_column__";
970
16
    bool all_file_columns_have_field_ids = true;
971
52
    for (uint64_t i = 0; i < orc_type_ptr->getSubtypeCount(); ++i) {
972
36
        const orc::Type* sub_type = orc_type_ptr->getSubtype(i);
973
36
        if (!sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
974
16
            all_file_columns_have_field_ids = false;
975
16
        }
976
36
    }
977
16
    const bool use_field_ids_for_hidden_keys =
978
16
            supports_iceberg_scan_semantics_v1(&get_scan_params())
979
16
                    ? orc_subtree_has_iceberg_id(orc_type_ptr, ICEBERG_ORC_ATTRIBUTE)
980
16
                    : all_file_columns_have_field_ids;
981
16
    const auto find_file_column_by_name = [&](const std::string& name) -> const orc::Type* {
982
0
        for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
983
0
            if (iequal(orc_type_ptr->getFieldName(j), name)) {
984
0
                return orc_type_ptr->getSubtype(j);
985
0
            }
986
0
        }
987
0
        return nullptr;
988
0
    };
989
990
16
    std::vector<std::string> new_expand_col_names;
991
16
    DORIS_CHECK(_expand_col_names.size() == _expand_col_field_ids.size());
992
16
    DORIS_CHECK(_expand_col_names.size() == _expand_columns.size());
993
30
    for (size_t i = 0; i < _expand_col_names.size(); ++i) {
994
14
        const auto& old_name = _expand_col_names[i];
995
14
        const int32_t field_id = _expand_col_field_ids[i];
996
997
14
        const orc::Type* file_column = nullptr;
998
14
        OrcEqualityFieldPath file_path;
999
14
        bool complete_file_path = false;
1000
14
        if (use_field_ids_for_hidden_keys) {
1001
8
            complete_file_path =
1002
8
                    find_orc_equality_field_path_by_id(orc_type_ptr, field_id, &file_path);
1003
8
            if (!complete_file_path && supports_iceberg_scan_semantics_v2(&get_scan_params())) {
1004
4
                const auto table_path = _find_schema_field_path(field_id);
1005
4
                if (!table_path.empty()) {
1006
4
                    complete_file_path = find_orc_equality_field_prefix_by_id_path(
1007
4
                            orc_type_ptr, table_path, &file_path);
1008
4
                }
1009
4
            }
1010
8
            if (!file_path.fields.empty()) {
1011
4
                file_column = file_path.fields.front();
1012
4
            }
1013
8
        } else {
1014
6
            const auto table_path = _find_schema_field_path(field_id);
1015
6
            if (!table_path.empty()) {
1016
6
                complete_file_path = find_orc_equality_field_prefix_by_name_path(
1017
6
                        orc_type_ptr, table_path, old_name, &file_path);
1018
6
                if (!file_path.fields.empty()) {
1019
5
                    file_column = file_path.fields.front();
1020
5
                }
1021
6
            } else {
1022
0
                file_column = find_file_column_by_name(old_name);
1023
0
                complete_file_path = file_column != nullptr;
1024
0
            }
1025
6
        }
1026
1027
14
        std::string file_col_name = old_name;
1028
14
        std::string leaf_name = old_name;
1029
14
        if (!file_path.fields.empty()) {
1030
9
            file_col_name = file_path.names.front();
1031
9
            leaf_name = file_path.names.back();
1032
9
        } else if (file_column != nullptr) {
1033
0
            for (uint64_t j = 0; j < orc_type_ptr->getSubtypeCount(); ++j) {
1034
0
                if (orc_type_ptr->getSubtype(j) == file_column) {
1035
0
                    file_col_name = orc_type_ptr->getFieldName(j);
1036
0
                    leaf_name = file_col_name;
1037
0
                    break;
1038
0
                }
1039
0
            }
1040
0
        }
1041
14
        std::string table_col_name = EQ_DELETE_PRE + std::to_string(field_id) + "_" + leaf_name;
1042
1043
14
        if (field_id >= 0) {
1044
14
            _id_to_block_column_name[field_id] = table_col_name;
1045
14
        }
1046
14
        _expand_columns[i].name = table_col_name;
1047
14
        if (file_column == nullptr) {
1048
5
            RETURN_IF_ERROR(_register_missing_equality_delete_column(field_id, table_col_name,
1049
5
                                                                     _expand_columns[i].type));
1050
            // The old data file predates this equality key. Keep it in the expand block so the
1051
            // synthesized-column hook can materialize its logical initial default before ORC's
1052
            // block-size checks. Adding it to column_names/table_info_node would mark it as an
1053
            // existing ORC child and make OrcReader read a column that is not present in the file.
1054
5
            new_expand_col_names.push_back(table_col_name);
1055
5
            continue;
1056
5
        }
1057
9
        new_expand_col_names.push_back(table_col_name);
1058
1059
9
        if (!complete_file_path) {
1060
1
            ColumnPtr missing_value;
1061
1
            RETURN_IF_ERROR(_create_missing_equality_delete_value(
1062
1
                    field_id, _expand_columns[i].type, file_path.fields.size(), &missing_value));
1063
1
            _nested_equality_delete_columns.push_back({
1064
1
                    .field_id = field_id,
1065
1
                    .block_name = table_col_name,
1066
1
                    .leaf_type = _expand_columns[i].type,
1067
1
                    .child_indexes = file_path.child_indexes,
1068
1
                    .missing_value = std::move(missing_value),
1069
1
            });
1070
1
            _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column));
1071
1
            _expand_columns[i].column = _expand_columns[i].type->create_column();
1072
8
        } else if (!file_path.child_indexes.empty()) {
1073
6
            _nested_equality_delete_columns.push_back({
1074
6
                    .field_id = field_id,
1075
6
                    .block_name = table_col_name,
1076
6
                    .leaf_type = _expand_columns[i].type,
1077
6
                    .child_indexes = file_path.child_indexes,
1078
6
                    .missing_value = nullptr,
1079
6
            });
1080
6
            _expand_columns[i].type = make_nullable(convert_to_doris_type(file_column));
1081
6
            _expand_columns[i].column = _expand_columns[i].type->create_column();
1082
6
        }
1083
1084
9
        for (uint64_t column_id = file_column->getColumnId();
1085
25
             column_id <= file_column->getMaximumColumnId(); ++column_id) {
1086
16
            ctx->column_ids.insert(column_id);
1087
16
        }
1088
1089
9
        ctx->column_names.push_back(table_col_name);
1090
9
        ctx->table_info_node->add_children(table_col_name, file_col_name,
1091
9
                                           TableSchemaChangeHelper::ConstNode::get_instance());
1092
9
    }
1093
16
    _expand_col_names = std::move(new_expand_col_names);
1094
1095
16
    return Status::OK();
1096
16
}
1097
1098
// ============================================================================
1099
// IcebergOrcReader: _create_column_ids
1100
// ============================================================================
1101
ColumnIdResult IcebergOrcReader::_create_column_ids(
1102
        const orc::Type* orc_type, const TupleDescriptor* tuple_descriptor,
1103
23
        const std::shared_ptr<TableSchemaChangeHelper::Node>& table_info_node) {
1104
23
    std::unordered_map<int, const orc::Type*> iceberg_id_to_orc_type_map;
1105
109
    for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
1106
86
        const auto* orc_sub_type = orc_type->getSubtype(i);
1107
86
        if (!orc_sub_type) {
1108
0
            continue;
1109
0
        }
1110
86
        if (!orc_sub_type->hasAttributeKey(ICEBERG_ORC_ATTRIBUTE)) {
1111
17
            continue;
1112
17
        }
1113
69
        int iceberg_id = std::stoi(orc_sub_type->getAttributeValue(ICEBERG_ORC_ATTRIBUTE));
1114
69
        iceberg_id_to_orc_type_map[iceberg_id] = orc_sub_type;
1115
69
    }
1116
1117
23
    std::set<uint64_t> column_ids;
1118
23
    std::set<uint64_t> filter_column_ids;
1119
1120
23
    auto process_access_paths = [](const orc::Type* orc_field,
1121
23
                                   const std::vector<TColumnAccessPath>& access_paths,
1122
23
                                   std::set<uint64_t>& out_ids) {
1123
17
        process_nested_access_paths(
1124
17
                orc_field, access_paths, out_ids,
1125
17
                [](const orc::Type* type) { return type->getColumnId(); },
1126
17
                [](const orc::Type* type) { return type->getMaximumColumnId(); },
1127
17
                IcebergOrcNestedColumnUtils::extract_nested_column_ids);
1128
17
    };
1129
1130
    // The Iceberg schema-mapping root is a StructNode whose registered children are the real
1131
    // table columns. When present, resolve each column by name through it so the column-id set
1132
    // stays consistent with the schema-mapping decision (BY_ID or BY_NAME/name-mapping);
1133
    // otherwise fall back to matching by Iceberg field id.
1134
23
    const auto* struct_node =
1135
23
            dynamic_cast<const TableSchemaChangeHelper::StructNode*>(table_info_node.get());
1136
1137
34
    for (const auto* slot : tuple_descriptor->slots()) {
1138
34
        const orc::Type* orc_field = nullptr;
1139
34
        if (struct_node != nullptr) {
1140
            // Synthesized/metadata slots (e.g. the TopN global row-id or the $row_id column) are
1141
            // never registered as children, so check membership before querying: calling
1142
            // children_column_exists() on an unregistered name DCHECK-aborts in debug builds and
1143
            // throws std::out_of_range from .at() in release builds.
1144
21
            if (struct_node->get_children().contains(slot->col_name()) &&
1145
21
                struct_node->children_column_exists(slot->col_name())) {
1146
                // Select the physical child resolved by the shared schema-mapping pass. Hidden
1147
                // equality keys and projected columns must obey the same BY_NAME decision for
1148
                // partial-id ORC files.
1149
19
                const auto& file_column_name =
1150
19
                        struct_node->children_file_column_name(slot->col_name());
1151
23
                for (uint64_t i = 0; i < orc_type->getSubtypeCount(); ++i) {
1152
23
                    if (orc_type->getFieldName(i) == file_column_name) {
1153
19
                        orc_field = orc_type->getSubtype(i);
1154
19
                        break;
1155
19
                    }
1156
23
                }
1157
19
                DORIS_CHECK(orc_field != nullptr);
1158
19
            }
1159
21
        } else {
1160
13
            auto it = iceberg_id_to_orc_type_map.find(slot->col_unique_id());
1161
13
            if (it != iceberg_id_to_orc_type_map.end()) {
1162
13
                orc_field = it->second;
1163
13
            }
1164
13
        }
1165
34
        if (orc_field == nullptr) {
1166
2
            continue;
1167
2
        }
1168
1169
32
        if ((slot->col_type() != TYPE_STRUCT && slot->col_type() != TYPE_ARRAY &&
1170
32
             slot->col_type() != TYPE_MAP)) {
1171
21
            column_ids.insert(orc_field->getColumnId());
1172
21
            if (slot->is_predicate()) {
1173
0
                filter_column_ids.insert(orc_field->getColumnId());
1174
0
            }
1175
21
            continue;
1176
21
        }
1177
1178
11
        const auto& all_access_paths = slot->all_access_paths();
1179
11
        process_access_paths(orc_field, all_access_paths, column_ids);
1180
1181
11
        const auto& predicate_access_paths = slot->predicate_access_paths();
1182
11
        if (!predicate_access_paths.empty()) {
1183
6
            process_access_paths(orc_field, predicate_access_paths, filter_column_ids);
1184
6
        }
1185
11
    }
1186
1187
23
    return {std::move(column_ids), std::move(filter_column_ids)};
1188
23
}
1189
1190
// ============================================================================
1191
// IcebergOrcReader: _read_position_delete_file
1192
// ============================================================================
1193
Status IcebergOrcReader::_read_position_delete_file(const TFileRangeDesc* delete_range,
1194
0
                                                    DeleteFile* position_delete) {
1195
0
    OrcReader orc_delete_reader(get_profile(), get_state(), get_scan_params(), *delete_range,
1196
0
                                READ_DELETE_FILE_BATCH_SIZE, get_state()->timezone(), get_io_ctx(),
1197
0
                                _meta_cache);
1198
0
    OrcInitContext delete_ctx;
1199
0
    delete_ctx.column_names = delete_file_col_names;
1200
0
    delete_ctx.col_name_to_block_idx =
1201
0
            const_cast<std::unordered_map<std::string, uint32_t>*>(&DELETE_COL_NAME_TO_BLOCK_IDX);
1202
0
    RETURN_IF_ERROR(orc_delete_reader.init_reader(&delete_ctx));
1203
1204
0
    bool eof = false;
1205
0
    DataTypePtr data_type_file_path {new DataTypeString};
1206
0
    DataTypePtr data_type_pos {new DataTypeInt64};
1207
0
    while (!eof) {
1208
0
        Block block = {{data_type_file_path, ICEBERG_FILE_PATH}, {data_type_pos, ICEBERG_ROW_POS}};
1209
1210
0
        size_t read_rows = 0;
1211
0
        RETURN_IF_ERROR(orc_delete_reader.get_next_block(&block, &read_rows, &eof));
1212
1213
0
        RETURN_IF_ERROR(_gen_position_delete_file_range(block, position_delete, read_rows, false));
1214
0
    }
1215
0
    return Status::OK();
1216
0
}
1217
1218
} // namespace doris