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