Coverage Report

Created: 2026-07-14 17:43

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format_v2/table/iceberg_reader.h
Line
Count
Source
1
// Licensed to the Apache Software Foundation (ASF) under one
2
// or more contributor license agreements.  See the NOTICE file
3
// distributed with this work for additional information
4
// regarding copyright ownership.  The ASF licenses this file
5
// to you under the Apache License, Version 2.0 (the
6
// "License"); you may not use this file except in compliance
7
// with the License.  You may obtain a copy of the License at
8
//
9
//   http://www.apache.org/licenses/LICENSE-2.0
10
//
11
// Unless required by applicable law or agreed to in writing,
12
// software distributed under the License is distributed on an
13
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14
// KIND, either express or implied.  See the License for the
15
// specific language governing permissions and limitations
16
// under the License.
17
18
#pragma once
19
20
#include <memory>
21
#include <optional>
22
#include <string>
23
#include <unordered_map>
24
#include <vector>
25
26
#include "common/status.h"
27
#include "core/block/block.h"
28
#include "format/table/iceberg_delete_file_reader_helper.h"
29
#include "format_v2/file_reader.h"
30
#include "format_v2/table_reader.h"
31
#include "gen_cpp/PlanNodes_types.h"
32
33
namespace doris {
34
class Block;
35
struct DeleteFileDesc;
36
namespace io {
37
struct FileDescription;
38
struct FileSystemProperties;
39
} // namespace io
40
} // namespace doris
41
42
namespace doris::format::iceberg {
43
44
// Iceberg table-level reader.
45
// It reuses TableReader for split orchestration, dynamic partition pruning and table-block
46
// finalization, while composing a FileReader for physical data-file reads instead of inheriting
47
// from a concrete file-format reader.
48
class IcebergTableReader : public format::TableReader {
49
public:
50
48
    ~IcebergTableReader() override = default;
51
42
    Status init(format::TableReadOptions&& options) override {
52
42
        RETURN_IF_ERROR(format::TableReader::init(std::move(options)));
53
42
        _mapper_options.mode = format::TableColumnMappingMode::BY_FIELD_ID;
54
42
        return Status::OK();
55
42
    }
56
57
    Status prepare_split(const format::SplitReadOptions& options) override;
58
    std::string debug_string() const override;
59
57
    format::TableColumnMappingMode mapping_mode() const override {
60
57
        return !_data_reader.file_schema.empty() && _has_field_id(_data_reader.file_schema)
61
57
                       ? format::TableColumnMappingMode::BY_FIELD_ID
62
57
                       : format::TableColumnMappingMode::BY_NAME;
63
57
    }
64
65
protected:
66
    Status materialize_virtual_columns(Block* table_block) override;
67
68
    Status customize_file_scan_request(format::FileScanRequest* file_request) override;
69
70
    bool _supports_aggregate_pushdown(TPushAggOp::type agg_type) const override;
71
72
    Status _parse_deletion_vector_file(const TTableFormatFileDesc& t_desc, DeleteFileDesc* desc,
73
                                       bool* has_delete_file) override;
74
75
    Status _init_delete_predicates(const TTableFormatFileDesc& t_desc);
76
77
private:
78
    struct EqualityDeleteFilter;
79
80
157
    bool _has_field_id(const std::vector<format::ColumnDefinition>& schema) const {
81
157
        for (const auto& field : schema) {
82
            // TopN lazy materialization asks the file reader to synthesize GLOBAL_ROWID in the
83
            // first-phase scan. That virtual column is not an Iceberg data field and therefore has
84
            // no Iceberg field id. Do not let it downgrade schema-evolution reads to BY_NAME,
85
            // otherwise old data files whose physical names predate a rename (for example,
86
            // table column `new_new_id` stored as file column `id`) are materialized as defaults.
87
108
            if (field.column_type != format::ColumnType::DATA_COLUMN) {
88
1
                continue;
89
1
            }
90
107
            if (!field.has_identifier_field_id()) {
91
7
                return false;
92
7
            }
93
100
            if (!_has_field_id(field.children)) {
94
0
                return false;
95
0
            }
96
100
        }
97
150
        return true;
98
157
    }
99
    static constexpr int MIN_SUPPORT_DELETE_FILES_VERSION = 2;
100
    static constexpr int POSITION_DELETE = 1;
101
    static constexpr int EQUALITY_DELETE = 2;
102
    static constexpr int DELETION_VECTOR = 3;
103
104
    struct RowLineageColumns {
105
        int64_t first_row_id = -1;
106
        int64_t last_updated_sequence_number = -1;
107
    };
108
109
    static constexpr const char* ICEBERG_FILE_PATH = "file_path";
110
    static constexpr const char* ICEBERG_ROW_POS = "pos";
111
    static constexpr size_t ICEBERG_FILE_PATH_BLOCK_POSITION = 0;
112
    static constexpr size_t ICEBERG_ROW_POS_BLOCK_POSITION = 1;
113
114
    class PositionDeleteRowsCollector final {
115
    public:
116
        using PositionDeleteFile = std::unordered_map<std::string, format::DeleteRows>;
117
118
        explicit PositionDeleteRowsCollector(PositionDeleteFile* rows_by_data_file);
119
120
        Status collect(const Block& block, size_t read_rows);
121
122
    private:
123
        PositionDeleteFile* _rows_by_data_file = nullptr;
124
    };
125
126
    static std::shared_ptr<io::FileSystemProperties> _delete_file_system_properties(
127
            const TFileScanRangeParams& scan_params);
128
129
    static std::unique_ptr<io::FileDescription> _delete_file_description(
130
            const TFileRangeDesc& range);
131
132
    std::string _data_file_path() const;
133
134
    // Append row position column to file scan request for position delete handling.
135
    Status _append_row_position_output_column(format::FileScanRequest* request);
136
    // Append equality delete predicates to file scan request based on the delete files in iceberg
137
    // params. DeleteVector and position delete files use the common DeleteRows path in TableReader.
138
    Status _append_equality_delete_predicates(format::FileScanRequest* request);
139
    const format::ColumnDefinition* _find_equality_delete_data_field(
140
            const EqualityDeleteFilter& filter, size_t key_idx) const;
141
    std::optional<format::ColumnDefinition> _find_equality_delete_table_field(
142
            const EqualityDeleteFilter& filter, size_t key_idx) const;
143
    void _append_equality_delete_row_count_carrier(format::FileScanRequest* request);
144
    std::string _delete_file_cache_key(const char* prefix, const std::string& path) const;
145
146
    Status _init_equality_delete_predicates(
147
            const std::vector<TIcebergDeleteFileDesc>& delete_files);
148
149
    // Read equality/position delete files.
150
    Status _create_delete_file_reader(const TIcebergDeleteFileDesc& delete_file,
151
                                      const TFileScanRangeParams& scan_params,
152
                                      IcebergDeleteFileIOContext* delete_io_ctx,
153
                                      std::unique_ptr<format::FileReader>* reader);
154
    Status _read_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,
155
                                      const TFileScanRangeParams& scan_params,
156
                                      IcebergDeleteFileIOContext* delete_io_ctx);
157
    Status _load_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,
158
                                      const TFileScanRangeParams& scan_params,
159
                                      IcebergDeleteFileIOContext* delete_io_ctx,
160
                                      EqualityDeleteFilter* result);
161
    Status _resolve_equality_delete_fields(const TIcebergDeleteFileDesc& delete_file,
162
                                           const std::vector<format::ColumnDefinition>& schema,
163
                                           std::vector<format::ColumnDefinition>* delete_fields,
164
                                           EqualityDeleteFilter* result) const;
165
    Status _read_position_delete_file(const TIcebergDeleteFileDesc& delete_file,
166
                                      const TFileScanRangeParams& scan_params,
167
                                      IcebergDeleteFileIOContext* delete_io_ctx,
168
                                      PositionDeleteRowsCollector* collector);
169
170
    // Read position delete files and collect deleted row positions to update DeletePredicate.
171
    Status _init_position_delete_rows(const std::vector<TIcebergDeleteFileDesc>& delete_files);
172
173
    // Materialize row lineage virtual columns based on the position delete file.
174
    Status _materialize_iceberg_rowid(Block* table_block, size_t column_idx);
175
    Status _materialize_row_lineage_row_id(Block* table_block, size_t column_idx);
176
    Status _materialize_row_lineage_last_updated_sequence_number(Block* table_block,
177
                                                                 size_t column_idx);
178
179
    RowLineageColumns _row_lineage_columns;
180
    size_t _row_position_block_position = 0;
181
    std::optional<TIcebergFileDesc> _iceberg_params;
182
    bool _delete_predicates_initialized = false;
183
    format::DeleteRows _position_delete_rows_storage;
184
    struct EqualityDeleteFilter {
185
        std::vector<int> field_ids;
186
        // Delete-file names are retained for Iceberg tables imported from formats that did not
187
        // persist field ids. In BY_NAME mode they are the fallback binding key.
188
        std::vector<std::string> field_names;
189
        std::vector<DataTypePtr> key_types;
190
        Block delete_block;
191
    };
192
    std::vector<EqualityDeleteFilter> _equality_delete_filters;
193
    // Scanner-shared cache supplied in SplitReadOptions. Parsed delete files outlive one data-file
194
    // split and can be reused by every split referencing the same delete file.
195
    ShardedKVCache* _split_cache = nullptr;
196
197
    bool _need_row_lineage_row_id() const;
198
    bool _need_iceberg_rowid() const;
199
};
200
201
} // namespace doris::format::iceberg