Coverage Report

Created: 2026-09-14 14:38

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
be/src/format_v2/file_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
//   http://www.apache.org/licenses/LICENSE-2.0
9
// Unless required by applicable law or agreed to in writing,
10
// software distributed under the License is distributed on an
11
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
12
// KIND, either express or implied.  See the License for the
13
// specific language governing permissions and limitations
14
// under the License.
15
16
#pragma once
17
18
#include <algorithm>
19
#include <cstddef>
20
#include <cstdint>
21
#include <functional>
22
#include <limits>
23
#include <map>
24
#include <memory>
25
#include <string>
26
#include <utility>
27
#include <vector>
28
29
#include "common/cast_set.h"
30
#include "common/status.h"
31
#include "core/data_type/data_type.h"
32
#include "core/field.h"
33
#include "exprs/vexpr_fwd.h"
34
#include "format_v2/column_data.h"
35
#include "gen_cpp/PlanNodes_types.h"
36
#include "io/file_factory.h"
37
#include "io/fs/file_reader_writer_fwd.h"
38
#include "io/io_common.h"
39
40
namespace doris {
41
class Block;
42
struct ConditionCacheContext;
43
44
namespace io {
45
struct IOContext;
46
} // namespace io
47
} // namespace doris
48
49
namespace doris::format {
50
51
class TableColumnMapper;
52
struct TableColumnMapperOptions;
53
54
enum class FileFormat {
55
    PARQUET,
56
    ORC,
57
    CSV,
58
    JSON,
59
    TEXT,
60
    JNI,
61
    NATIVE,
62
    ARROW,
63
    WAL,
64
};
65
66
struct FileScanRequest {
67
835
    virtual ~FileScanRequest() = default;
68
69
    std::string debug_string() const;
70
71
    // Columns that must be read before row-level filtering. They are materialized eagerly because
72
    // conjuncts/delete_conjuncts need them to decide the selected rows.
73
    std::vector<LocalColumnIndex> predicate_columns;
74
    // Columns read after row-level filtering. A complex root may intentionally also appear in
75
    // predicate_columns when its eager predicate subtree is smaller than its final output subtree.
76
    std::vector<LocalColumnIndex> non_predicate_columns;
77
    // Predicate columns introduced only to evaluate hidden filter slots. Their values are dead
78
    // after all file-local predicates run, although the shared file block still needs row-shaped
79
    // placeholders until TableReader finalizes projected columns.
80
    std::vector<LocalColumnId> predicate_only_columns;
81
    // file-local column id -> file-local output block position.
82
    std::map<LocalColumnId, LocalIndex> local_positions;
83
    // Optional output position for a root that has independent eager-predicate and deferred-output
84
    // projections. local_positions continues to identify the position referenced by localized
85
    // predicate expressions.
86
    std::map<LocalColumnId, LocalIndex> non_predicate_positions;
87
    // Row-level filters converted to file-local expressions from table-level predicates.
88
    VExprContextSPtrs conjuncts;
89
    // Only this leading subset may participate in footer/page metadata pruning. The boundary is
90
    // inherited from table-conjunct order so an omitted slotless unsafe expression remains a fence.
91
    size_t metadata_pruning_safe_conjunct_count = std::numeric_limits<size_t>::max();
92
    // Constant split pruning may use only this table-filter prefix after mapping. A rejected
93
    // file-local rewrite is a materialization barrier even when a later filter is constant.
94
    size_t constant_pruning_safe_table_filter_count = std::numeric_limits<size_t>::max();
95
    // Delete predicates converted to file-local expressions. A TRUE result means that the row is
96
    // deleted, so readers must invert each result when building their keep filter.
97
    VExprContextSPtrs delete_conjuncts;
98
    // File-local ids retained only because Nereids keeps a minimum-width output tuple for an
99
    // explicit COUNT(*). These columns have no semantic value: for example, after pruning a scan
100
    // may retain an unsupported TIME_MILLIS leaf even though COUNT(*) only needs one row per
101
    // surviving input row. A reader may synthesize defaults instead of reading a marked column
102
    // while it remains non-predicate. If filters or equality deletes promote the same id to
103
    // predicate_columns, the value is semantically required and must still be validated and read.
104
    std::vector<LocalColumnId> count_star_placeholder_columns;
105
106
    // Absolute zero-based file row positions selected by a row-id fetch. A present but empty
107
    // vector means that no rows should be read; nullopt keeps the normal sequential scan path.
108
    // Readers require strictly increasing positions so they can seek forward without duplicating
109
    // output rows.
110
    std::optional<std::vector<int64_t>> row_ids = std::nullopt;
111
112
    // Table formats may assign semantics that legacy physical files do not encode. Each path here
113
    // identifies an unannotated Parquet group that the physical reader must validate and decode as
114
    // Variant. Keeping this explicit prevents generic Parquet scans from guessing based on names.
115
    std::vector<LocalColumnIndex> variant_schema_overrides;
116
117
2.44k
    bool is_count_star_placeholder(LocalColumnId column_id) const {
118
2.44k
        return std::ranges::find(count_star_placeholder_columns, column_id) !=
119
2.44k
               count_star_placeholder_columns.end();
120
2.44k
    }
121
122
214
    bool is_predicate_only(LocalColumnId column_id) const {
123
214
        return std::ranges::find(predicate_only_columns, column_id) != predicate_only_columns.end();
124
214
    }
125
126
1.15k
    LocalIndex non_predicate_position(LocalColumnId column_id) const {
127
1.15k
        const auto it = non_predicate_positions.find(column_id);
128
1.15k
        return it == non_predicate_positions.end() ? local_positions.at(column_id) : it->second;
129
1.15k
    }
130
131
1.03k
    bool has_deferred_non_predicate_column(LocalColumnId column_id) const {
132
1.03k
        return non_predicate_positions.contains(column_id);
133
1.03k
    }
134
135
166
    size_t block_column_count() const {
136
166
        size_t count = 0;
137
245
        for (const auto& [_, position] : local_positions) {
138
245
            count = std::max(count, position.value() + 1);
139
245
        }
140
166
        for (const auto& [_, position] : non_predicate_positions) {
141
0
            count = std::max(count, position.value() + 1);
142
0
        }
143
166
        return count;
144
166
    }
145
};
146
147
// Helper for constructing the scan-column layout in FileScanRequest.
148
// FileScanRequest keeps predicate and non-predicate columns separate because columnar readers such
149
// as Parquet can read predicate columns first, filter rows, and then lazily read the remaining
150
// projected columns. The two lists still share one file-local output block, whose positions are
151
// stored in local_positions. This builder centralizes the mechanical rules for that shared layout:
152
// - each root file column gets one stable predicate block position;
153
// - predicate columns dominate non-predicate columns because they are already returned in the file
154
//   block and can be reused for final materialization;
155
// - a smaller complex predicate subtree may get a second deferred output position;
156
// - repeated nested projections for the same root are merged instead of duplicated.
157
// TableColumnMapper should still own table-to-file semantic resolution. This helper only owns the
158
// FileScanRequest layout contract after a file-local projection has been produced.
159
class FileScanRequestBuilder {
160
public:
161
555
    explicit FileScanRequestBuilder(FileScanRequest* request) : _request(request) {
162
555
        DORIS_CHECK(_request != nullptr);
163
555
    }
164
165
264
    Status add_predicate_column(LocalColumnIndex projection) {
166
264
        return _add_column(std::move(projection), &_request->predicate_columns,
167
264
                           /*is_predicate_column=*/true);
168
264
    }
169
170
505
    Status add_non_predicate_column(LocalColumnIndex projection) {
171
505
        return _add_column(std::move(projection), &_request->non_predicate_columns,
172
505
                           /*is_predicate_column=*/false);
173
505
    }
174
175
11
    Status add_deferred_non_predicate_column(LocalColumnIndex projection) {
176
11
        const auto file_column_id = projection.column_id();
177
11
        DORIS_CHECK(file_column_id != LocalColumnId::invalid());
178
11
        DORIS_CHECK(_request->local_positions.contains(file_column_id));
179
11
        DORIS_CHECK(std::ranges::any_of(_request->predicate_columns,
180
11
                                        [&](const LocalColumnIndex& predicate) {
181
11
                                            return predicate.column_id() == file_column_id;
182
11
                                        }));
183
184
11
        if (!_request->non_predicate_positions.contains(file_column_id)) {
185
9
            _request->non_predicate_positions.emplace(file_column_id,
186
9
                                                      _next_block_position(*_request));
187
9
        }
188
11
        _sort_projection_children_by_file_id(&projection);
189
11
        auto existing = std::ranges::find_if(_request->non_predicate_columns,
190
11
                                             [&](const LocalColumnIndex& output) {
191
2
                                                 return output.column_id() == file_column_id;
192
2
                                             });
193
11
        if (existing == _request->non_predicate_columns.end()) {
194
11
            _request->non_predicate_columns.push_back(std::move(projection));
195
11
        } else {
196
0
            RETURN_IF_ERROR(merge_local_column_index(&*existing, projection));
197
0
            _sort_projection_children_by_file_id(&*existing);
198
0
        }
199
11
        if (!_request->is_predicate_only(file_column_id)) {
200
            // The eager complex value has a different physical shape from the final value and
201
            // must never leak into table materialization after its predicates have run.
202
11
            _request->predicate_only_columns.push_back(file_column_id);
203
11
        }
204
11
        return Status::OK();
205
11
    }
206
207
160
    Status add_predicate_column(LocalColumnId column_id) {
208
160
        return add_predicate_column(LocalColumnIndex::top_level(column_id));
209
160
    }
210
211
295
    Status add_non_predicate_column(LocalColumnId column_id) {
212
295
        return add_non_predicate_column(LocalColumnIndex::top_level(column_id));
213
295
    }
214
215
private:
216
715
    static LocalIndex _next_block_position(const FileScanRequest& request) {
217
715
        size_t next_position = 0;
218
8.36k
        for (const auto& [_, block_position] : request.local_positions) {
219
8.36k
            next_position = std::max(next_position, block_position.value() + 1);
220
8.36k
        }
221
715
        for (const auto& [_, block_position] : request.non_predicate_positions) {
222
3
            next_position = std::max(next_position, block_position.value() + 1);
223
3
        }
224
715
        return LocalIndex(next_position);
225
715
    }
226
227
891
    static void _sort_projection_children_by_file_id(LocalColumnIndex* projection) {
228
891
        DORIS_CHECK(projection != nullptr);
229
891
        if (projection->project_all_children) {
230
801
            return;
231
801
        }
232
107
        for (auto& child : projection->children) {
233
107
            _sort_projection_children_by_file_id(&child);
234
107
        }
235
90
        std::ranges::sort(projection->children,
236
90
                          [](const LocalColumnIndex& lhs, const LocalColumnIndex& rhs) {
237
32
                              return lhs.local_id() < rhs.local_id();
238
32
                          });
239
90
    }
240
241
    Status _add_column(LocalColumnIndex projection, std::vector<LocalColumnIndex>* scan_columns,
242
769
                       bool is_predicate_column) {
243
769
        DORIS_CHECK(scan_columns != nullptr);
244
769
        const auto file_column_id = projection.column_id();
245
769
        DORIS_CHECK(file_column_id != LocalColumnId::invalid());
246
769
        if (!is_predicate_column &&
247
769
            std::ranges::find_if(_request->predicate_columns,
248
505
                                 [&](const LocalColumnIndex& p) {
249
76
                                     return p.column_id() == file_column_id;
250
76
                                 }) != _request->predicate_columns.end() &&
251
769
            !_request->has_deferred_non_predicate_column(file_column_id)) {
252
2
            return Status::OK();
253
2
        }
254
767
        if (!_request->local_positions.contains(file_column_id)) {
255
706
            _request->local_positions.emplace(file_column_id, _next_block_position(*_request));
256
706
        }
257
258
767
        _sort_projection_children_by_file_id(&projection);
259
767
        auto existing_projection_it = std::ranges::find_if(
260
767
                *scan_columns,
261
8.10k
                [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
262
767
        if (existing_projection_it == scan_columns->end()) {
263
761
            scan_columns->push_back(std::move(projection));
264
761
        } else {
265
6
            RETURN_IF_ERROR(merge_local_column_index(&*existing_projection_it, projection));
266
6
            _sort_projection_children_by_file_id(&*existing_projection_it);
267
6
        }
268
269
767
        if (is_predicate_column && !_request->has_deferred_non_predicate_column(file_column_id)) {
270
261
            auto it = std::ranges::find_if(
271
261
                    _request->non_predicate_columns,
272
261
                    [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
273
261
            if (it != _request->non_predicate_columns.end()) {
274
13
                _request->non_predicate_columns.erase(it);
275
13
            }
276
261
        }
277
767
        return Status::OK();
278
767
    }
279
280
    FileScanRequest* _request = nullptr;
281
};
282
283
struct FileAggregateRequest {
284
    struct Column {
285
        // File-local projection for the aggregate column. For nested MIN/MAX, this points to the
286
        // single primitive leaf that can be represented by file statistics. For COUNT(col), this
287
        // points to the top-level column whose NULL-ness should be counted.
288
        LocalColumnIndex projection;
289
    };
290
291
    TPushAggOp::type agg_type = TPushAggOp::type::NONE;
292
    // Empty for COUNT(*)/row-count pushdown. Non-empty for COUNT(col), where the file reader must
293
    // return the number of non-NULL rows for the requested column instead of total rows.
294
    std::vector<Column> columns;
295
};
296
297
struct FileAggregateResult {
298
    struct Column {
299
        // Mirrors FileAggregateRequest::Column::projection so TableReader can put the returned
300
        // aggregate value back into the matching projected nested shape.
301
        LocalColumnIndex projection;
302
        bool has_min = false;
303
        bool has_max = false;
304
        Field min_value;
305
        Field max_value;
306
    };
307
308
    int64_t count = 0;
309
    std::vector<Column> columns;
310
};
311
312
/**
313
 *                                +-----> get_schema() -----------------+
314
 * FileReader() -----> init() ----|                                      -----> close()
315
 *                                +-----> open() -----> get_block() ----+
316
 */
317
class FileReader {
318
public:
319
    struct ReaderStatistics {
320
        int32_t filtered_row_groups = 0;
321
        int32_t filtered_row_groups_by_min_max = 0;
322
        int32_t filtered_row_groups_by_bloom_filter = 0;
323
        int32_t read_row_groups = 0;
324
        int64_t filtered_group_rows = 0;
325
        int64_t filtered_page_rows = 0;
326
        int64_t lazy_read_filtered_rows = 0;
327
        int64_t read_rows = 0;
328
        int64_t filtered_bytes = 0;
329
        int64_t column_read_time = 0;
330
        int64_t parse_meta_time = 0;
331
        int64_t parse_footer_time = 0;
332
        int64_t file_footer_read_calls = 0;
333
        int64_t file_footer_hit_cache = 0;
334
        int64_t file_reader_create_time = 0;
335
        int64_t open_file_num = 0;
336
        int64_t row_group_filter_time = 0;
337
        int64_t page_index_filter_time = 0;
338
        int64_t read_page_index_time = 0;
339
        int64_t parse_page_index_time = 0;
340
        int64_t predicate_filter_time = 0;
341
        int64_t dict_filter_rewrite_time = 0;
342
        int64_t bloom_filter_read_time = 0;
343
    };
344
345
    FileReader(std::shared_ptr<io::FileSystemProperties>& system_properties,
346
               std::unique_ptr<io::FileDescription>& file_description,
347
               std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile)
348
727
            : _system_properties(system_properties),
349
727
              _file_description(std::move(file_description)),
350
727
              _io_ctx(io_ctx),
351
727
              _profile(profile) {}
352
727
    virtual ~FileReader() = default;
353
354
    // Initialize file reader and parse file metadata.
355
    virtual Status init(RuntimeState* state);
356
357
    // Set the maximum row count for the next physical read batch. Readers that do not batch by
358
    // rows may ignore it.
359
0
    virtual void set_batch_size(size_t batch_size) { (void)batch_size; }
360
361
    // Get semantic file-local schema from file metadata. The file schema is determined by file
362
    // format and file content, and does not contain table/global schema semantics. A file reader may
363
    // expose raw file identifiers, such as Parquet field_id, through ColumnDefinition::identifier,
364
    // but it must not interpret table-format semantics such as Iceberg name mapping,
365
    // default/generated columns, or partition columns. File-format physical wrappers should be
366
    // normalized away before exposing this schema; for example, Parquet MAP is exposed as key/value
367
    // children rather than key_value/entry.
368
    // Doris plans external-table scan types as nullable, including all nested children of complex
369
    // types. This protects Doris from illegal or inconsistent values produced by external systems.
370
    // Therefore every ColumnDefinition::type returned here must be nullable. Complex types must
371
    // also expose nullable child types recursively, even if the physical file marks those fields as
372
    // required.
373
    // This method can only be called after init() successfully, but does not require open() to be
374
    // called.
375
    virtual Status get_schema(std::vector<ColumnDefinition>* file_schema) const = 0;
376
377
    // Create the mapper that matches this reader's scan-request capabilities. TableReader still
378
    // owns table-format semantics such as BY_NAME/BY_FIELD_ID/BY_INDEX, partition values and
379
    // default expressions; the FileReader only chooses whether file-local requests support columnar
380
    // lazy materialization/pruning or must materialize one flat list of required columns.
381
    virtual std::unique_ptr<TableColumnMapper> create_column_mapper(
382
            TableColumnMapperOptions options) const;
383
384
0
    virtual bool supports_rowid_fetch() const { return false; }
385
386
    // Open the file reader with file-local scan request. The file reader should initialize its internal state according to the request, but does not need to interpret table/global schema semantics. For example, all schema change, filter localization, default/generated/partition columns should be handled in table reader layer. This method can only be called after init() successfully.
387
667
    virtual Status open(std::shared_ptr<FileScanRequest> request) {
388
667
        if (request->row_ids.has_value()) {
389
2
            if (!supports_rowid_fetch()) {
390
0
                return Status::NotSupported("File reader does not support row-id fetch");
391
0
            }
392
2
            const auto& row_ids = *request->row_ids;
393
6
            if (std::ranges::any_of(row_ids, [](int64_t row_id) { return row_id < 0; }) ||
394
2
                std::ranges::adjacent_find(row_ids, std::greater_equal<>()) != row_ids.end()) {
395
0
                return Status::InvalidArgument(
396
0
                        "Row-id fetch requires non-negative, strictly increasing file row ids");
397
0
            }
398
2
        }
399
667
        _request = std::move(request);
400
667
        return Status::OK();
401
667
    }
402
403
    // Readers opt in only when they can keep an immutable request for the active physical
404
    // granule and switch a newer snapshot at a well-defined boundary.
405
0
    virtual bool supports_scan_request_refresh() const { return false; }
406
407
0
    virtual Status queue_scan_request(std::shared_ptr<FileScanRequest> request) {
408
0
        (void)request;
409
0
        return Status::NotSupported("FileReader does not support scan request refresh");
410
0
    }
411
412
0
    virtual Status get_block(Block* file_block, size_t* rows, bool* eof) {
413
0
        if (rows != nullptr) {
414
0
            *rows = 0;
415
0
        }
416
0
        if (eof != nullptr) {
417
0
            *eof = true;
418
0
        }
419
0
        _eof = true;
420
0
        return Status::OK();
421
0
    }
422
423
    virtual Status get_aggregate_result(const FileAggregateRequest& request,
424
0
                                        FileAggregateResult* result) {
425
0
        return Status::NotSupported("FileReader does not support aggregate pushdown");
426
0
    }
427
428
    // Condition cache is managed by TableReader and consumed by physical file readers.
429
    // On cache HIT, readers may skip granules whose cached bit is false before doing column IO.
430
    // On cache MISS, readers mark a granule true when row-level predicates keep at least one row
431
    // in that granule. Readers that cannot map batch rows to stable file-global row ids should
432
    // keep the default no-op implementation.
433
0
    virtual void set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) {}
434
435
    // Total rows covered by this physical reader. TableReader uses it to pre-size the miss bitmap.
436
    // Readers should return 0 if the metadata is unavailable or the row coordinate is unstable.
437
0
    virtual int64_t get_total_rows() const { return 0; }
438
439
218
    virtual Status close() {
440
218
        _file_reader.reset();
441
218
        _tracing_file_reader.reset();
442
218
        _io_ctx.reset();
443
218
        _eof = true;
444
218
        return Status::OK();
445
218
    }
446
447
protected:
448
0
    virtual void _init_profile() {}
449
608
    void _record_scan_rows(int64_t rows) {
450
608
        DORIS_CHECK(rows >= 0);
451
608
        _reader_statistics.read_rows += rows;
452
608
        if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
453
188
            _io_ctx->file_reader_stats->read_rows += cast_set<size_t>(rows);
454
188
        }
455
608
    }
456
457
    io::FileReaderSPtr _file_reader;
458
    // _tracing_file_reader wraps _file_reader.
459
    // _file_reader is original file reader.
460
    // _tracing_file_reader is tracing file reader with io context.
461
    // If io_ctx is null, _tracing_file_reader will be the same as file_reader.
462
    io::FileReaderSPtr _tracing_file_reader = nullptr;
463
    std::shared_ptr<FileScanRequest> _request;
464
    bool _eof = true;
465
    ReaderStatistics _reader_statistics;
466
    std::shared_ptr<io::FileSystemProperties> _system_properties;
467
    std::unique_ptr<io::FileDescription> _file_description;
468
    std::shared_ptr<io::IOContext> _io_ctx;
469
    RuntimeProfile* _profile = nullptr;
470
};
471
472
} // namespace doris::format