Coverage Report

Created: 2026-10-15 23:16

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/iceberg_schema_utils.h"
31
#include "format_v2/table_reader.h"
32
#include "gen_cpp/PlanNodes_types.h"
33
34
namespace doris {
35
class Block;
36
class EqualityDeleteHashIndex;
37
struct DeleteFileDesc;
38
namespace io {
39
struct FileDescription;
40
struct FileSystemProperties;
41
} // namespace io
42
} // namespace doris
43
44
namespace doris::format::iceberg {
45
46
Status prepare_iceberg_initial_default_exprs(format::ColumnDefinition* column);
47
48
// Iceberg table-level reader.
49
// It reuses TableReader for split orchestration, dynamic partition pruning and table-block
50
// finalization, while composing a FileReader for physical data-file reads instead of inheriting
51
// from a concrete file-format reader.
52
class IcebergTableReader : public format::TableReader {
53
public:
54
163
    ~IcebergTableReader() override = default;
55
    static Status validate_variant_file_mappings(
56
            FileFormat format, const std::vector<format::ColumnMapping>& mappings);
57
138
    Status init(format::TableReadOptions&& options) override {
58
138
        RETURN_IF_ERROR(format::TableReader::init(std::move(options)));
59
138
        _mapper_options.mode = format::TableColumnMappingMode::BY_FIELD_ID;
60
138
        _mapper_options.reject_missing_required_field =
61
138
                supports_iceberg_scan_semantics_v2(_scan_params);
62
138
        return Status::OK();
63
138
    }
64
65
    Status prepare_split(const format::SplitReadOptions& options) override;
66
    Status annotate_projected_column(const TFileScanSlotInfo& slot_info,
67
                                     format::ProjectedColumnBuildContext* context,
68
                                     format::ColumnDefinition* column) const override;
69
    std::string debug_string() const override;
70
321
    format::TableColumnMappingMode mapping_mode() const override {
71
321
        const bool has_field_ids = supports_iceberg_scan_semantics_v1(_scan_params)
72
321
                                           ? schema_has_any_field_id(_data_reader.file_schema)
73
321
                                           : schema_has_all_field_ids(_data_reader.file_schema);
74
321
        if (!_data_reader.file_schema.empty() && has_field_ids) {
75
275
            return format::TableColumnMappingMode::BY_FIELD_ID;
76
275
        }
77
46
        if (!_data_reader.file_schema.empty() && supports_iceberg_scan_semantics_v2(_scan_params) &&
78
46
            !_scan_has_any_authoritative_name_mapping()) {
79
            // ID-less migrated files are name-readable only while Iceberg's explicit default name
80
            // mapping exists; current names must not resurrect file fields after it is removed.
81
1
            return format::TableColumnMappingMode::BY_FIELD_ID;
82
1
        }
83
45
        return format::TableColumnMappingMode::BY_NAME;
84
46
    }
85
86
protected:
87
    // Iceberg UUID uses the same 16-byte STRING/VARBINARY carrier in data, defaults and deletes.
88
183
    bool preserve_binary_uuid() const override { return true; }
89
90
    Status validate_file_mapping(const format::TableColumnMapper& mapper) const override;
91
92
    // Iceberg's Parquet reader resolves a data file that carries no field ids at all through the
93
    // table's `schema.name-mapping.default`; when that mapping is absent it falls back to
94
    // positional ids (ParquetSchemaUtil.addFallbackIds + pruneColumnsFallback, gated by
95
    // `iceberg.netflix.unsafe-parquet-id-fallback.enabled`, default true through 1.12 and
96
    // scheduled for removal in 2.0). Mirror that decision for Parquet by stamping ordinal ids
97
    // (1..N) on the file's top-level fields, so the shared BY_FIELD_ID path resolves those files
98
    // instead of silently materializing NULL. Only top-level fields are stamped: Iceberg assigns
99
    // no ids to nested fields, so struct children of an id-less file stay unreadable without a
100
    // name mapping (or a rewrite that persists field ids).
101
    Status annotate_file_schema(std::vector<format::ColumnDefinition>* file_schema) override;
102
103
    // Iceberg's positional fallback matches an ID-less Parquet file by physical ordinals, so the
104
    // same table column can bind a different file column when the table's
105
    // `schema.name-mapping.default` is added or removed without touching the data file. The
106
    // condition cache key covers the file identity and the table-level predicate digest only, so it
107
    // cannot tell those bindings apart; keep such splits out of the cache (neither read nor
108
    // written) instead of reusing a verdict produced by the other binding.
109
    bool _should_enable_condition_cache(
110
165
            const format::FileScanRequest& file_request) const override {
111
165
        return !_positional_id_fallback &&
112
165
               format::TableReader::_should_enable_condition_cache(file_request);
113
165
    }
114
115
165
    void configure_mapper_options(format::TableColumnMapperOptions* options) const override {
116
165
        options->enable_row_lineage_virtual_columns = true;
117
165
        options->enable_iceberg_metadata_virtual_columns = true;
118
165
        options->reject_missing_required_field = supports_iceberg_scan_semantics_v2(_scan_params);
119
165
        options->allow_idless_complex_wrapper_projection =
120
165
                supports_iceberg_scan_semantics_v1(_scan_params) && _format == FileFormat::PARQUET;
121
165
    }
122
123
    Status materialize_virtual_columns(Block* table_block) override;
124
125
    Status customize_file_scan_request(format::FileScanRequest* file_request) override;
126
127
    bool _supports_aggregate_pushdown(TPushAggOp::type agg_type) const override;
128
129
    Status _parse_deletion_vector_file(const TTableFormatFileDesc& t_desc, DeleteFileDesc* desc,
130
                                       bool* has_delete_file) override;
131
132
    Status _init_delete_predicates(const TTableFormatFileDesc& t_desc);
133
134
private:
135
    struct EqualityDeleteFilter;
136
    static constexpr int MIN_SUPPORT_DELETE_FILES_VERSION = 2;
137
    static constexpr int POSITION_DELETE = 1;
138
    static constexpr int EQUALITY_DELETE = 2;
139
    static constexpr int DELETION_VECTOR = 3;
140
141
    struct RowLineageColumns {
142
        int64_t first_row_id = -1;
143
        int64_t last_updated_sequence_number = -1;
144
    };
145
146
    static constexpr const char* ICEBERG_FILE_PATH = "file_path";
147
    static constexpr const char* ICEBERG_ROW_POS = "pos";
148
    static constexpr size_t ICEBERG_FILE_PATH_BLOCK_POSITION = 0;
149
    static constexpr size_t ICEBERG_ROW_POS_BLOCK_POSITION = 1;
150
151
    bool _scan_has_any_authoritative_name_mapping() const;
152
153
    class PositionDeleteRowsCollector final {
154
    public:
155
        using PositionDeleteFile = std::unordered_map<std::string, format::DeleteRows>;
156
157
        explicit PositionDeleteRowsCollector(PositionDeleteFile* rows_by_data_file);
158
159
        Status collect(const Block& block, size_t read_rows);
160
161
    private:
162
        PositionDeleteFile* _rows_by_data_file = nullptr;
163
    };
164
165
    static std::shared_ptr<io::FileSystemProperties> _delete_file_system_properties(
166
            const TFileScanRangeParams& scan_params);
167
168
    static std::unique_ptr<io::FileDescription> _delete_file_description(
169
            const TFileRangeDesc& range);
170
171
    std::string _data_file_path() const;
172
173
    // Append row position column to file scan request for position delete handling.
174
    Status _append_row_position_output_column(format::FileScanRequest* request);
175
    // Append equality delete predicates to file scan request based on the delete files in iceberg
176
    // params. DeleteVector and position delete files use the common DeleteRows path in TableReader.
177
    using EqualityDeleteColumnPath = std::vector<const format::ColumnDefinition*>;
178
    Status _append_equality_delete_predicates(format::FileScanRequest* request);
179
    Status _build_missing_equality_delete_key_expr(const EqualityDeleteFilter& filter,
180
                                                   size_t key_idx,
181
                                                   const EqualityDeleteColumnPath& data_path,
182
                                                   format::FileScanRequest* request,
183
                                                   VExprSPtr* key_expr);
184
    Status _find_equality_delete_data_field(const EqualityDeleteFilter& filter, size_t key_idx,
185
                                            EqualityDeleteColumnPath* data_path,
186
                                            bool* complete_path) const;
187
    Status _find_equality_delete_table_field(const EqualityDeleteFilter& filter, size_t key_idx,
188
                                             format::ColumnDefinition* table_field) const;
189
    void _append_equality_delete_row_count_carrier(format::FileScanRequest* request);
190
    std::string _delete_file_cache_key(const char* prefix, const std::string& path) const;
191
192
    Status _init_equality_delete_predicates(
193
            const std::vector<TIcebergDeleteFileDesc>& delete_files);
194
195
    // Read equality/position delete files.
196
    Status _create_delete_file_reader(const TIcebergDeleteFileDesc& delete_file,
197
                                      const TFileScanRangeParams& scan_params,
198
                                      IcebergDeleteFileIOContext* delete_io_ctx,
199
                                      std::unique_ptr<format::FileReader>* reader);
200
    Status _read_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,
201
                                      const TFileScanRangeParams& scan_params,
202
                                      IcebergDeleteFileIOContext* delete_io_ctx);
203
    Status _load_equality_delete_file(const TIcebergDeleteFileDesc& delete_file,
204
                                      const TFileScanRangeParams& scan_params,
205
                                      IcebergDeleteFileIOContext* delete_io_ctx,
206
                                      EqualityDeleteFilter* result);
207
    Status _resolve_equality_delete_fields(const TIcebergDeleteFileDesc& delete_file,
208
                                           const std::vector<format::ColumnDefinition>& schema,
209
                                           std::vector<EqualityDeleteColumnPath>* delete_paths,
210
                                           EqualityDeleteFilter* result) const;
211
    Status _read_position_delete_file(const TIcebergDeleteFileDesc& delete_file,
212
                                      const TFileScanRangeParams& scan_params,
213
                                      IcebergDeleteFileIOContext* delete_io_ctx,
214
                                      PositionDeleteRowsCollector* collector);
215
216
    // Read position delete files and collect deleted row positions to update DeletePredicate.
217
    Status _init_position_delete_rows(const std::vector<TIcebergDeleteFileDesc>& delete_files);
218
219
    // Materialize row lineage virtual columns based on the position delete file.
220
    Status _materialize_iceberg_rowid(Block* table_block, size_t column_idx);
221
    Status _materialize_iceberg_file_path(Block* table_block, size_t column_idx);
222
    Status _materialize_iceberg_row_position(Block* table_block, size_t column_idx);
223
    Status _materialize_row_lineage_row_id(Block* table_block, size_t column_idx);
224
    Status _materialize_row_lineage_last_updated_sequence_number(Block* table_block,
225
                                                                 size_t column_idx);
226
227
    RowLineageColumns _row_lineage_columns;
228
    size_t _row_position_block_position = 0;
229
    std::optional<TIcebergFileDesc> _iceberg_params;
230
    bool _delete_predicates_initialized = false;
231
    // Set when this split's Parquet file carried no field ids and the table exposed no usable name
232
    // mapping, so the reader stamped Iceberg's positional fallback ids on the file's top-level
233
    // fields. Observability only; the decision itself lives in annotate_file_schema().
234
    bool _positional_id_fallback = false;
235
    format::DeleteRows _position_delete_rows_storage;
236
    struct EqualityDeleteFilter {
237
        std::vector<int> field_ids;
238
        // Delete-file names are retained for Iceberg tables imported from formats that did not
239
        // persist field ids. In BY_NAME mode they are the fallback binding key.
240
        std::vector<std::string> field_names;
241
        std::vector<DataTypePtr> key_types;
242
        Block delete_block;
243
        std::shared_ptr<const EqualityDeleteHashIndex> hash_index;
244
    };
245
    std::vector<EqualityDeleteFilter> _equality_delete_filters;
246
    // Scanner-shared cache supplied in SplitReadOptions. Parsed delete files outlive one data-file
247
    // split and can be reused by every split referencing the same delete file.
248
    ShardedKVCache* _split_cache = nullptr;
249
250
    bool _need_row_lineage_row_id() const;
251
    bool _need_iceberg_rowid() const;
252
    bool _need_iceberg_metadata() const;
253
};
254
255
} // namespace doris::format::iceberg