Coverage Report

Created: 2026-07-23 18:24

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 <map>
22
#include <memory>
23
#include <string>
24
#include <utility>
25
#include <vector>
26
27
#include "common/cast_set.h"
28
#include "common/status.h"
29
#include "core/data_type/data_type.h"
30
#include "core/field.h"
31
#include "exprs/vexpr_fwd.h"
32
#include "format_v2/column_data.h"
33
#include "gen_cpp/PlanNodes_types.h"
34
#include "io/file_factory.h"
35
#include "io/fs/file_reader_writer_fwd.h"
36
37
namespace doris {
38
class Block;
39
struct ConditionCacheContext;
40
41
namespace io {
42
struct IOContext;
43
} // namespace io
44
} // namespace doris
45
46
namespace doris::format {
47
48
class TableColumnMapper;
49
struct TableColumnMapperOptions;
50
51
enum class FileFormat {
52
    PARQUET,
53
    ORC,
54
    CSV,
55
    JSON,
56
    TEXT,
57
    JNI,
58
    NATIVE,
59
    ARROW,
60
};
61
62
struct FileScanRequest {
63
65.9k
    virtual ~FileScanRequest() = default;
64
65
    std::string debug_string() const;
66
67
    // Columns that must be read before row-level filtering. They are materialized eagerly because
68
    // conjuncts/delete_conjuncts need them to decide the selected rows.
69
    std::vector<LocalColumnIndex> predicate_columns;
70
    // Columns read after row-level filtering. Predicate columns are also available for output and
71
    // should not be duplicated here.
72
    std::vector<LocalColumnIndex> non_predicate_columns;
73
    // Predicate columns introduced only to evaluate hidden filter slots. Their values are dead
74
    // after all file-local predicates run, although the shared file block still needs row-shaped
75
    // placeholders until TableReader finalizes projected columns.
76
    std::vector<LocalColumnId> predicate_only_columns;
77
    // file-local column id -> file-local output block position.
78
    std::map<LocalColumnId, LocalIndex> local_positions;
79
    // Row-level filters converted to file-local expressions from table-level predicates. Readers
80
    // must enforce these exactly on returned rows; metadata pruning alone does not transfer
81
    // predicate ownership away from TableReader.
82
    VExprContextSPtrs conjuncts;
83
    // Delete predicates converted to file-local expressions. A TRUE result means that the row is
84
    // deleted, so readers must invert each result when building their keep filter.
85
    VExprContextSPtrs delete_conjuncts;
86
    // File-local ids retained only because Nereids keeps a minimum-width output tuple for an
87
    // explicit COUNT(*). These columns have no semantic value: for example, after pruning a scan
88
    // may retain an unsupported TIME_MILLIS leaf even though COUNT(*) only needs one row per
89
    // surviving input row. A reader may synthesize defaults instead of reading a marked column
90
    // while it remains non-predicate. If filters or equality deletes promote the same id to
91
    // predicate_columns, the value is semantically required and must still be validated and read.
92
    std::vector<LocalColumnId> count_star_placeholder_columns;
93
94
770k
    bool is_count_star_placeholder(LocalColumnId column_id) const {
95
770k
        return std::ranges::find(count_star_placeholder_columns, column_id) !=
96
770k
               count_star_placeholder_columns.end();
97
770k
    }
98
99
47.6k
    bool is_predicate_only(LocalColumnId column_id) const {
100
47.6k
        return std::ranges::find(predicate_only_columns, column_id) != predicate_only_columns.end();
101
47.6k
    }
102
};
103
104
// Helper for constructing the scan-column layout in FileScanRequest.
105
// FileScanRequest keeps predicate and non-predicate columns separate because columnar readers such
106
// as Parquet can read predicate columns first, filter rows, and then lazily read the remaining
107
// projected columns. The two lists still share one file-local output block, whose positions are
108
// stored in local_positions. This builder centralizes the mechanical rules for that shared layout:
109
// - each root file column gets one stable block position;
110
// - predicate columns dominate non-predicate columns because they are already returned in the file
111
//   block and can be reused for final materialization;
112
// - repeated nested projections for the same root are merged instead of duplicated.
113
// TableColumnMapper should still own table-to-file semantic resolution. This helper only owns the
114
// FileScanRequest layout contract after a file-local projection has been produced.
115
class FileScanRequestBuilder {
116
public:
117
394k
    explicit FileScanRequestBuilder(FileScanRequest* request) : _request(request) {
118
394k
        DORIS_CHECK(_request != nullptr);
119
394k
    }
120
121
46.8k
    Status add_predicate_column(LocalColumnIndex projection) {
122
46.8k
        return _add_column(std::move(projection), &_request->predicate_columns,
123
46.8k
                           /*is_predicate_column=*/true);
124
46.8k
    }
125
126
347k
    Status add_non_predicate_column(LocalColumnIndex projection) {
127
347k
        return _add_column(std::move(projection), &_request->non_predicate_columns,
128
347k
                           /*is_predicate_column=*/false);
129
347k
    }
130
131
10.7k
    Status add_predicate_column(LocalColumnId column_id) {
132
10.7k
        return add_predicate_column(LocalColumnIndex::top_level(column_id));
133
10.7k
    }
134
135
2.12k
    Status add_non_predicate_column(LocalColumnId column_id) {
136
2.12k
        return add_non_predicate_column(LocalColumnIndex::top_level(column_id));
137
2.12k
    }
138
139
private:
140
381k
    static LocalIndex _next_block_position(const FileScanRequest& request) {
141
381k
        size_t next_position = 0;
142
7.33M
        for (const auto& [_, block_position] : request.local_positions) {
143
7.33M
            next_position = std::max(next_position, block_position.value() + 1);
144
7.33M
        }
145
381k
        return LocalIndex(next_position);
146
381k
    }
147
148
414k
    static void _sort_projection_children_by_file_id(LocalColumnIndex* projection) {
149
414k
        DORIS_CHECK(projection != nullptr);
150
414k
        if (projection->project_all_children) {
151
407k
            return;
152
407k
        }
153
10.1k
        for (auto& child : projection->children) {
154
10.1k
            _sort_projection_children_by_file_id(&child);
155
10.1k
        }
156
6.98k
        std::ranges::sort(projection->children,
157
6.98k
                          [](const LocalColumnIndex& lhs, const LocalColumnIndex& rhs) {
158
6.18k
                              return lhs.local_id() < rhs.local_id();
159
6.18k
                          });
160
6.98k
    }
161
162
    Status _add_column(LocalColumnIndex projection, std::vector<LocalColumnIndex>* scan_columns,
163
393k
                       bool is_predicate_column) {
164
393k
        DORIS_CHECK(scan_columns != nullptr);
165
393k
        const auto file_column_id = projection.column_id();
166
393k
        DORIS_CHECK(file_column_id != LocalColumnId::invalid());
167
393k
        if (!is_predicate_column &&
168
393k
            std::ranges::find_if(_request->predicate_columns, [&](const LocalColumnIndex& p) {
169
939
                return p.column_id() == file_column_id;
170
939
            }) != _request->predicate_columns.end()) {
171
480
            return Status::OK();
172
480
        }
173
393k
        if (!_request->local_positions.contains(file_column_id)) {
174
381k
            _request->local_positions.emplace(file_column_id, _next_block_position(*_request));
175
381k
        }
176
177
393k
        _sort_projection_children_by_file_id(&projection);
178
393k
        auto existing_projection_it = std::ranges::find_if(
179
393k
                *scan_columns,
180
7.27M
                [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
181
393k
        if (existing_projection_it == scan_columns->end()) {
182
382k
            scan_columns->push_back(std::move(projection));
183
382k
        } else {
184
10.5k
            RETURN_IF_ERROR(merge_local_column_index(&*existing_projection_it, projection));
185
10.5k
            _sort_projection_children_by_file_id(&*existing_projection_it);
186
10.5k
        }
187
188
393k
        if (is_predicate_column) {
189
47.0k
            auto it = std::ranges::find_if(
190
47.0k
                    _request->non_predicate_columns,
191
65.2k
                    [&](const LocalColumnIndex& p) { return p.column_id() == file_column_id; });
192
47.0k
            if (it != _request->non_predicate_columns.end()) {
193
1.33k
                _request->non_predicate_columns.erase(it);
194
1.33k
            }
195
47.0k
        }
196
393k
        return Status::OK();
197
393k
    }
198
199
    FileScanRequest* _request = nullptr;
200
};
201
202
struct FileAggregateRequest {
203
    struct Column {
204
        // File-local projection for the aggregate column. For nested MIN/MAX, this points to the
205
        // single primitive leaf that can be represented by file statistics. For COUNT(col), this
206
        // points to the top-level column whose NULL-ness should be counted.
207
        LocalColumnIndex projection;
208
    };
209
210
    TPushAggOp::type agg_type = TPushAggOp::type::NONE;
211
    // Empty for COUNT(*)/row-count pushdown. Non-empty for COUNT(col), where the file reader must
212
    // return the number of non-NULL rows for the requested column instead of total rows.
213
    std::vector<Column> columns;
214
};
215
216
struct FileAggregateResult {
217
    struct Column {
218
        // Mirrors FileAggregateRequest::Column::projection so TableReader can put the returned
219
        // aggregate value back into the matching projected nested shape.
220
        LocalColumnIndex projection;
221
        bool has_min = false;
222
        bool has_max = false;
223
        Field min_value;
224
        Field max_value;
225
    };
226
227
    int64_t count = 0;
228
    std::vector<Column> columns;
229
};
230
231
/**
232
 *                                +-----> get_schema() -----------------+
233
 * FileReader() -----> init() ----|                                      -----> close()
234
 *                                +-----> open() -----> get_block() ----+
235
 */
236
class FileReader {
237
public:
238
    struct ReaderStatistics {
239
        int32_t filtered_row_groups = 0;
240
        int32_t filtered_row_groups_by_min_max = 0;
241
        int32_t filtered_row_groups_by_bloom_filter = 0;
242
        int32_t read_row_groups = 0;
243
        int64_t filtered_group_rows = 0;
244
        int64_t filtered_page_rows = 0;
245
        int64_t lazy_read_filtered_rows = 0;
246
        int64_t read_rows = 0;
247
        int64_t filtered_bytes = 0;
248
        int64_t column_read_time = 0;
249
        int64_t parse_meta_time = 0;
250
        int64_t parse_footer_time = 0;
251
        int64_t file_footer_read_calls = 0;
252
        int64_t file_footer_hit_cache = 0;
253
        int64_t file_reader_create_time = 0;
254
        int64_t open_file_num = 0;
255
        int64_t row_group_filter_time = 0;
256
        int64_t page_index_filter_time = 0;
257
        int64_t read_page_index_time = 0;
258
        int64_t parse_page_index_time = 0;
259
        int64_t predicate_filter_time = 0;
260
        int64_t dict_filter_rewrite_time = 0;
261
        int64_t bloom_filter_read_time = 0;
262
    };
263
264
    FileReader(std::shared_ptr<io::FileSystemProperties>& system_properties,
265
               std::unique_ptr<io::FileDescription>& file_description,
266
               std::shared_ptr<io::IOContext> io_ctx, RuntimeProfile* profile)
267
65.8k
            : _system_properties(system_properties),
268
65.8k
              _file_description(std::move(file_description)),
269
65.8k
              _io_ctx(io_ctx),
270
65.8k
              _profile(profile) {}
271
65.8k
    virtual ~FileReader() = default;
272
273
    // Initialize file reader and parse file metadata.
274
    virtual Status init(RuntimeState* state);
275
276
    // Set the maximum row count for the next physical read batch. Readers that do not batch by
277
    // rows may ignore it.
278
83.1k
    virtual void set_batch_size(size_t batch_size) { (void)batch_size; }
279
280
    // Get semantic file-local schema from file metadata. The file schema is determined by file
281
    // format and file content, and does not contain table/global schema semantics. A file reader may
282
    // expose raw file identifiers, such as Parquet field_id, through ColumnDefinition::identifier,
283
    // but it must not interpret table-format semantics such as Iceberg name mapping,
284
    // default/generated columns, or partition columns. File-format physical wrappers should be
285
    // normalized away before exposing this schema; for example, Parquet MAP is exposed as key/value
286
    // children rather than key_value/entry.
287
    // Doris plans external-table scan types as nullable, including all nested children of complex
288
    // types. This protects Doris from illegal or inconsistent values produced by external systems.
289
    // Therefore every ColumnDefinition::type returned here must be nullable. Complex types must
290
    // also expose nullable child types recursively, even if the physical file marks those fields as
291
    // required.
292
    // This method can only be called after init() successfully, but does not require open() to be
293
    // called.
294
    virtual Status get_schema(std::vector<ColumnDefinition>* file_schema) const = 0;
295
296
    // Create the mapper that matches this reader's scan-request capabilities. TableReader still
297
    // owns table-format semantics such as BY_NAME/BY_FIELD_ID/BY_INDEX, partition values and
298
    // default expressions; the FileReader only chooses whether file-local requests support columnar
299
    // lazy materialization/pruning or must materialize one flat list of required columns.
300
    virtual std::unique_ptr<TableColumnMapper> create_column_mapper(
301
            TableColumnMapperOptions options) const;
302
303
    // 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.
304
65.1k
    virtual Status open(std::shared_ptr<FileScanRequest> request) {
305
65.1k
        _request = std::move(request);
306
65.1k
        return Status::OK();
307
65.1k
    }
308
309
0
    virtual Status get_block(Block* file_block, size_t* rows, bool* eof) {
310
0
        if (rows != nullptr) {
311
0
            *rows = 0;
312
0
        }
313
0
        if (eof != nullptr) {
314
0
            *eof = true;
315
0
        }
316
0
        _eof = true;
317
0
        return Status::OK();
318
0
    }
319
320
    virtual Status get_aggregate_result(const FileAggregateRequest& request,
321
0
                                        FileAggregateResult* result) {
322
0
        return Status::NotSupported("FileReader does not support aggregate pushdown");
323
0
    }
324
325
    // Condition cache is managed by TableReader and consumed by physical file readers.
326
    // On cache HIT, readers may skip granules whose cached bit is false before doing column IO.
327
    // On cache MISS, readers mark a granule true when row-level predicates keep at least one row
328
    // in that granule. Readers that cannot map batch rows to stable file-global row ids should
329
    // keep the default no-op implementation.
330
0
    virtual void set_condition_cache_context(std::shared_ptr<ConditionCacheContext> ctx) {}
331
332
    // Total rows covered by this physical reader. TableReader uses it to pre-size the miss bitmap.
333
    // Readers should return 0 if the metadata is unavailable or the row coordinate is unstable.
334
115
    virtual int64_t get_total_rows() const { return 0; }
335
336
59.5k
    virtual Status close() {
337
59.5k
        _file_reader.reset();
338
59.5k
        _tracing_file_reader.reset();
339
59.5k
        _io_ctx.reset();
340
59.5k
        _eof = true;
341
59.5k
        return Status::OK();
342
59.5k
    }
343
344
protected:
345
0
    virtual void _init_profile() {}
346
130k
    void _record_scan_rows(int64_t rows) {
347
130k
        DORIS_CHECK(rows >= 0);
348
130k
        _reader_statistics.read_rows += rows;
349
130k
        if (_io_ctx != nullptr && _io_ctx->file_reader_stats != nullptr) {
350
130k
            _io_ctx->file_reader_stats->read_rows += cast_set<size_t>(rows);
351
130k
        }
352
130k
    }
353
354
    io::FileReaderSPtr _file_reader;
355
    // _tracing_file_reader wraps _file_reader.
356
    // _file_reader is original file reader.
357
    // _tracing_file_reader is tracing file reader with io context.
358
    // If io_ctx is null, _tracing_file_reader will be the same as file_reader.
359
    io::FileReaderSPtr _tracing_file_reader = nullptr;
360
    std::shared_ptr<FileScanRequest> _request;
361
    bool _eof = true;
362
    ReaderStatistics _reader_statistics;
363
    std::shared_ptr<io::FileSystemProperties> _system_properties;
364
    std::unique_ptr<io::FileDescription> _file_description;
365
    std::shared_ptr<io::IOContext> _io_ctx;
366
    RuntimeProfile* _profile = nullptr;
367
};
368
369
} // namespace doris::format