[improvement](external) Refine Parquet splits by row group in BE - #66802
[improvement](external) Refine Parquet splits by row group in BE#66802Gabriel39 wants to merge 3 commits into
Conversation
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes for nine inline findings:
- [P1] Mutable Hive/TVF files with no stable version identity can reuse an old footer from the new scan registry (
be/src/format_v2/parquet/parquet_file_context.cpp:347). - [P1] The scan-local footer registry retains every context and metadata-cache handle until local-state teardown (
be/src/format_v2/file_scan_context.h:57). - [P1] Every Iceberg row-group child rebuilds file-wide position/equality delete state, producing O(R*D) setup and concurrency-scaled memory (
be/src/format_v2/parquet/parquet_reader.cpp:569). - [P1] The new split-planning open bypasses configured missing/empty-file handling (
be/src/exec/scan/file_scanner_v2.cpp:630). - [P2] Exactly-one-row-group files discard and reopen an already initialized planning reader without gaining parallelism (
be/src/format_v2/parquet/parquet_reader.cpp:579). - [P2] Early predicate rejection skips later columns before their MergeRange/exact-cache ranges are activated (
be/src/format_v2/parquet/parquet_scan.cpp:2380). - [P2] Finished-range load progress advances when source ownership is retired, before the source or its generated children finish (
be/src/exec/scan/file_scanner_v2.cpp:491). - [P2] Shared FileReadCalls/FileReadTime profile counters are overwritten by scanner-local absolutes instead of aggregated (
be/src/exec/scan/file_scanner_v2.cpp:1163). - [P2] ReaderLocalCacheAdmissionRejects is permanently zero because the consumed result bit is never set (
be/src/io/cache/cached_remote_file_reader.cpp:1347).
I also checked split claim/publication/EOS and cancellation, row-group selection and global row positions, metadata COUNT handling, Hudi/Paimon compatibility, exact-cache fallback byte semantics, reader-local cache ownership/accounting, and the Q28 decoder threshold; no additional distinct defect was substantiated.
Review status: capped/incomplete. Three mandated rounds were completed, but the mutable-file identity issue was first accepted in the final allowed round, so this review cannot claim convergence.
Tests were not run, as required by the supplied review instructions. Maintenance note: be/benchmark/parquet/AGENTS.md still describes 228 scenarios/six selectivities while the changed benchmark and test now expect 266 including 5%.
|
|
||
| std::shared_ptr<const FileContext> resolved_context = std::move(file_context); | ||
| if (resolved_context == nullptr) { | ||
| if (file_context_registry != nullptr) { |
There was a problem hiding this comment.
[P1] Do not reuse this registry entry for a mutable file with no stable version identity. build_native_file_cache_key() intentionally returns an empty key when mtime is zero and !is_immutable, because an overwrite can preserve path and size while changing both footer semantics and page bytes; this unconditional scan-local lookup nevertheless keys that unstable tuple. Ordinary Hive/TVF files can take this path, and generated children reopen the path while carrying the planner's old footer, so pruning or decoding can use old metadata against new bytes. Gate registry reuse on has_stable_meta_cache_identity; for refined children, pin/revalidate the opened version or decline splitting when no stable identity exists.
| }; | ||
|
|
||
| std::mutex _lock; | ||
| std::unordered_map<std::string, std::shared_ptr<Entry>> _entries; |
There was a problem hiding this comment.
[P1] Bound the scan-local footer registry. This map and each Entry retain strong ownership until FileScanLocalState teardown, while ParquetSharedFileContext owns either the parsed footer or an ObjLRUCache CacheHandle. A remote split source can therefore stream many unique files and make metadata memory/cache pins grow with the total file count, even after all children for earlier files finish. Please expire weak entries or explicitly retire/bound them once their children are no longer queued or active.
| row_group_id); | ||
| } | ||
| FileScanSplit child; | ||
| child.source_range = shared_source_range; |
There was a problem hiding this comment.
[P1] Do not rebuild Iceberg delete indexes for every row-group child. Each child reuses the full source table-format descriptor, so its independent prepare_split() recopies/sorts the data file's complete position-delete vector and constructs a fresh equality predicate that rehashes every delete row; the cache shares only the raw mapping/Block. A file with R row groups and D deletes now pays O(R*D) setup and can retain one full delete vector/hash map per concurrent child, erasing the split benefit or exhausting memory. Share immutable prepared delete state across the children, or decline refinement for Iceberg splits with delete files until it can be shared.
| if (_current_split.is_source_split) { | ||
| std::vector<FileScanSplit> generated_splits; | ||
| bool was_split = false; | ||
| RETURN_IF_ERROR(_table_reader->build_physical_splits(_current_split, &generated_splits, |
There was a problem hiding this comment.
[P1] Preserve missing-file handling around split planning. prepare_split() has not opened the Parquet data file yet; build_physical_splits() now creates and initializes the temporary reader, so a missing object or footer EOF can first surface here. This direct RETURN_IF_ERROR bypasses the ignore_not_found_file_in_external_table and empty-file branches just above (and the equivalent get_block branches that handled this lazy init before the PR), causing configured ignorable files to fail scanner open. Classify this status through the same skip/abort/retire path before returning other errors.
| child.format_split_id = row_group_id; | ||
| splits->push_back(std::move(child)); | ||
| } | ||
| *was_split = true; |
There was a problem hiding this comment.
[P2] Reuse the planning reader for a single selected row group. This method has already initialized the temporary Parquet reader, but returning a one-element child list closes it, aborts the source state, and repeats physical-reader, schema, and table-format/delete setup for the child with no parallelism to gain. This is common for one-row-group files. Retain or transfer that initialized reader into the prepared TableReader when there is exactly one selection; merely returning was_split=false without retaining it would still reinitialize the file later.
| const auto& col = request.predicate_columns[idx]; | ||
| const auto fid = col.column_id(); | ||
| if (_current_merge_range_reader != nullptr) { | ||
| RETURN_IF_ERROR(activate_merge_ranges_for_columns({fid})); |
There was a problem hiding this comment.
[P2] Activate the columns that must be skipped after early rejection. This stages only the current predicate. If it rejects the whole batch, the code later calls NativeColumnReader::skip() for every unmaterialized predicate without activating their ranges; inside a selected span that skip parses headers and often loads page data. With the new empty initial MergeRange list, those reads delegate directly to remote storage and bypass both coalescing and exact-cache probing. Activate the skipped columns as one stage, or keep their cursor lag logical until they become reachable.
| RETURN_IF_ERROR( | ||
| _split_source->finish_source_split(_current_split, std::move(generated_splits))); | ||
| _current_split.is_source_split = false; | ||
| _state->update_num_finished_scan_range(1); |
There was a problem hiding this comment.
[P2] Separate reservation retirement from finished-range progress. This increment now runs when an ordinary source is merely prepared, and for a refined parent before any queued row-group child is read; generated children never increment later. FE load progress periodically publishes this value, so it can report every range finished/100% while the actual data is still scanning. Retire the connector reservation here, but count the FE range only at unsplit EOF or after the final child completes.
| COUNTER_SET(_file_read_bytes_counter, bytes_read); | ||
| // Scanner instances share the profile counter, so publishing an absolute value would erase | ||
| // bytes already reported by sibling scanners. | ||
| COUNTER_UPDATE(_file_read_bytes_counter, |
There was a problem hiding this comment.
[P2] Aggregate calls and time with deltas too. These counters are shared by sibling scanners for the same reason FileReadBytes is, and this PR makes one FE source produce row-group work that those scanners can consume, but the following COUNTER_SET calls publish each scanner's private cumulative values. With scanners at 10 and 20 calls, the final profile can show either 10 or 20 instead of 30; FileReadTime has the same loss, and the close path repeats it. Give both metrics per-scanner reported watermarks and COUNTER_UPDATE their deltas alongside bytes.
| RETURN_IF_ERROR(_reader_local_file_cache->get_or_load( | ||
| buffer_offset, buffer_size, block, buffer_offset - block->range().left, &lookup)); | ||
| stats.num_reader_local_cache_evict += cast_set<int64_t>(lookup.evicted); | ||
| stats.num_reader_local_cache_admission_reject += lookup.admission_rejected ? 1 : 0; |
There was a problem hiding this comment.
[P2] Record reader-local admission rejection before consuming this flag. LookupResult::admission_rejected defaults to false and is never set anywhere; in particular, the _reserve() failure path aborts and returns without changing it. Consequently the newly exposed ReaderLocalCacheAdmissionRejects counter is always zero under capacity/query/process memory pressure. Set the flag on policy/reservation denial and cover it in the existing rejection tests.
|
run buildall |
|
/review |
There was a problem hiding this comment.
Requesting changes for three distinct P2 findings after duplicate-fencing the nine live inline threads and the prior landed review body:
- Generated row-group children repeat a file-wide row-prefix pass, making the new fan-out O(R^2) in row-group count.
- Eager PARQUET-816-overlapping MergeRange inputs can publish an invalid gap counter because useful coverage is double-counted before unsigned subtraction.
- The default-on reader-local cache can turn a tiny first warm-FileCache read into a 256 KiB disk read/allocation with no guaranteed reuse.
Critical checkpoint conclusions:
- Split lifecycle and concurrency: source claim/publication/EOS, parent-to-child progress transfer, ignored-file handling, stop, close, and cancellation were traced across local and remote sources. No distinct lifecycle issue remains beyond the already-landed threads; the generic planning amplification is reported inline.
- File and table semantics: stable identity, weak footer ownership, child byte ranges, file-global row positions, aggregate/count behavior, Hudi/Paimon forwarding, deletion vectors, and Iceberg delete opt-out survived the final recheck.
- I/O, cache, resources, and observability: reservation, pinning, single-flight fill, eviction, destruction, exact-cache fallback, staged ranges, cursor/box lifetime, and profile propagation were checked. The remaining distinct cache-policy and MergeRange-counter defects are reported inline.
- Decoder, indexing, and coverage: the fragmented dictionary threshold, cursor consumption, OffsetIndex paths, and the changed 266-case matrix are internally consistent. The stale 228-case benchmark guide was duplicate-suppressed because review 4945449517 already records it. Tests were not run, as required by the supplied review instructions.
User focus: no additional user-provided focus was supplied.
Review status: converged. Two complete normal-review rounds plus separate risk-focused passes were completed; every final-round subagent returned NO_NEW_VALUABLE_FINDINGS, all candidates were adjudicated, and the live head/base still match the authoritative bundle.
| std::vector<int> selected_row_groups; | ||
| RETURN_IF_ERROR(detail::select_native_row_groups_by_scan_range( | ||
| _state->file_context.native_metadata->to_thrift(), scan_range, &row_group_first_rows, | ||
| &selected_row_groups)); |
There was a problem hiding this comment.
[P2] Reuse the row-prefix calculation across generated children. This parent call computes row_group_first_rows for the whole footer, but each child receives only format_split_id; when that child opens, select_native_row_groups_by_scan_range allocates the same R-entry vector and walks all R row groups again even for an exact id. Refining a file into R children therefore adds O(R^2) footer work and O(C*R) concurrent scratch for files with many row groups. This is distinct from the Iceberg delete-state thread because it applies to every refined Parquet file. Retain the immutable prefix in the shared context or carry each child's first_file_row so exact-id planning is O(1).
| } | ||
| } | ||
| _statistics.merged_useful_bytes += useful_bytes; | ||
| _statistics.merged_gap_bytes += bytes_read - useful_bytes; |
There was a problem hiding this comment.
[P2] Compute useful bytes over the union of eager ranges. PARQUET-816 padding can make adjacent column-chunk ranges overlap, while the eager path only sorts those ranges before the constructor; only staged additions are coalesced. This loop then counts the overlap once per range, so an 8-byte read in two padded ranges records 16 useful bytes and the unsigned bytes_read - useful_bytes expression wraps before it is stored, publishing an invalid gap counter. Coalesce constructor inputs with the staged path or measure union coverage, and cover eager overlapping ranges in the counter test.
|
|
||
| const size_t buffer_size = buffer_end - buffer_offset; | ||
| FileScannerV2ReaderLocalFileCache::LookupResult lookup; | ||
| RETURN_IF_ERROR(_reader_local_file_cache->get_or_load( |
There was a problem hiding this comment.
[P2] Avoid promoting a full 256 KiB region on every first local-cache hit by default. This call receives the aligned buffer_size, so even the added 8-byte-read test performs a 256 KiB FileCache disk read and allocation. Sparse footer/index/page probes can touch distinct regions, and generated children have separate per-reader maps, so neighboring bytes need never be reused while the default-on cache churns its 64 MiB per-scanner budget. This is distinct from the admission-rejection counter thread: the fill succeeds but performs excessive I/O. Admit after observed reuse, fill the requested/planned span, or keep this disabled until representative warm sparse-cache benchmarks show the default is safe.
ffa60c4 to
a3bf75b
Compare
|
run buildall |
|
/review |
There was a problem hiding this comment.
Automated review completed for a3bf75b.
Five new actionable findings are attached inline (1 P1, 4 P2).
Critical checkpoints:
- Split coordination and lifecycle: source claims, child publication, EOS/stop/error wakeups, last-child progress, and one-reader reuse were traced; no additional lifecycle defect remained.
- Reader/table semantics: empty/NOT_FOUND handling, Iceberg delete guards, Hudi/Paimon forwarding, aggregate paths, and late filters were checked.
- Metadata/cache identity: stable identity, weak registry ownership, shared footer lifetime, and small-HTTP staging were reviewed.
- Row coordinates: exact row-group IDs, file-wide prefixes, GLOBAL_ROWID/TopN fetch, condition-cache bases, and aggregate counts were traced.
- Performance/observability: scanner-cap detection, shared counter deltas, footer reuse attribution, and child I/O/setup amplification were checked.
- Tests/coverage: reviewed the new concurrency, lifetime, exact-row-group, ignored-error, progress, counter, and reuse coverage. Per runner instruction, no build or tests were executed.
All 2,314 authoritative diff lines across 27 changed files were reviewed. The twelve existing inline threads were read and duplicate-fenced; no substantially similar comment was resubmitted.
User focus: no additional user-provided review focus.
| group_start = std::min(group_start, chunk_range.offset); | ||
| group_end = std::max(group_end, chunk_range.offset + chunk_range.length); | ||
| } | ||
| if (group_end <= group_start) { |
There was a problem hiding this comment.
[P1] Skip zero-row groups before deriving a child range. Valid Parquet files can contain an empty row group with zero-length column chunks; the ordinary planner skips it, but refinement reaches this new Corruption branch first and fails the whole source. Skip row_group.num_rows == 0 here and cover empty groups before/between populated groups plus an all-empty file.
| _state->file_context.shared_file_context == nullptr) { | ||
| return Status::Uninitialized("ParquetReader is not open"); | ||
| } | ||
| if (!_state->file_context.shared_file_context->has_stable_identity) { |
There was a problem hiding this comment.
[P2] Keep staged small HTTP files on the initialized reader. Iceberg marks snapshot files immutable, so an HTTP(S) Parquet object at or below in_memory_file_size passes this refinement gate; every generated child then creates a private InMemoryFileReader and reloads/copies the entire object on first access. Share the staged bytes across children or decline refinement for this reader type, and cover a multi-row-group HTTP file.
| } | ||
| FileScanSplit child; | ||
| child.source_range = shared_source_range; | ||
| child.start_offset = cast_set<int64_t>(group_start); |
There was a problem hiding this comment.
[P2] Separate GLOBAL_ROWID's file mapping from the physical child range. This child start replaces the source start before _create_global_rowid_context(), and IdFileMap plus RowIdStorageReader group by path/start, so K row-group children create K second-phase scanner/file-reader setups (and K footer reads when metadata cache is disabled) for one FE source. Carry the source identity for GLOBAL_ROWID while retaining this child range for first-phase reads, with an explicit batching policy and a multi-child TopN test.
| return Status::OK(); | ||
| }; | ||
|
|
||
| std::shared_ptr<const FileContext> resolved_context = std::move(file_context); |
There was a problem hiding this comment.
[P2] Account for scan-local footer-context reuse. Supplied child contexts and successful registry reuse bypass load_context(), so they increment neither FileFooterReadCalls nor FileFooterHitCache; the new sibling/child test locks in that invisible work. Add registry request/hit/wait/miss-or-load/bypass counters while keeping physical reads and process-cache hits distinguishable.
| std::ranges::any_of(file_scan_range.ranges, [&](const auto& range) { | ||
| const auto format = | ||
| range.__isset.format_type ? range.format_type : params->format_type; | ||
| if (format == TFileFormatType::FORMAT_PARQUET) { |
There was a problem hiding this comment.
[P2] Keep the range-count cap when downstream disables refinement. This returns true for every Parquet range, but _can_refine_source_split() rejects an Iceberg range with nonempty delete_files. A one-range scan can therefore construct up to the configured scanner maximum; all but the source owner block in get_next_split() while it initializes delete state, although no child can ever be published. Mirror the delete guard here and test the resulting scanner count.
|
run buildall |
|
/review |
|
Codex automated review failed and did not complete. Error: Review step was failure (possibly timeout or cancelled) Please inspect the workflow logs and rerun the review after the underlying issue is resolved. |
There was a problem hiding this comment.
Automated review conclusion: request changes on head 2b0e338d4e42c354d4a80364f172bafb03177c06. I found one P1 correctness issue and two P2 profile/accounting issues; details are inline.
Critical checkpoints:
- Architecture and lifecycle: traced source claiming, child publication, waiter wakeups, stop/cancel, progress transfer, and 0/1/many/all-empty plus error paths. Coordination is sound; the first-split RF/cache race and lifecycle timing gaps are called out inline.
- Row semantics: checked exact Row Group selection, absolute ROW_POSITION/GLOBAL_ROWID coordinates, deletes/deletion vectors, Condition Cache granules, and COUNT/MINMAX behavior. Apart from the late-RF cache-key race, no additional row-semantic defect survived.
- Footer, cache, and resources: checked stable identity, mutable-file bypass, registry single-flight/lifetime, process-cache handles, small-HTTP staging, and additive I/O publication. Remaining concerns are already covered by the live review threads.
- Dispatch and compatibility: checked direct Parquet/ORC, Hudi/Paimon native and JNI shapes, Iceberg delete fences, and JNI/Lance option compatibility; no additional issue survived.
- Tests: reviewed all changed tests. Builds and tests were not run, per the review environment instructions; the inline comments identify the missing regressions.
- User focus: no additional user-provided review focus.
Convergence status: three full rounds were completed. Round 3 produced the P1 finding, which two independent risk challenges confirmed without finding another issue. Because this was the final allowed round, the review is capped/incomplete rather than converged.
| // waiting for its row-group children, so deferring publication until a later get_block() | ||
| // turn could let those waiters occupy the scan thread pool ahead of the producer. | ||
| bool eos = false; | ||
| RETURN_IF_ERROR(_prepare_next_split(&eos)); |
There was a problem hiding this comment.
[P1] Refresh the split digest after late Runtime Filters
This eagerly prepares the first split inside open(), before ScannerScheduler calls try_append_late_arrival_runtime_filter(). If an RF arrives during that work, a generated Parquet child or first ORC split keeps digest(P), but the reader-null refresh changes its conjuncts to P AND RF without updating the digest/coverage flag. A Condition Cache MISS that reaches EOF can then publish P AND RF survivors under P; a later P-only HIT will skip granules containing valid rows. Defer preparation until after the RF hook, or recompute/invalidate the split cache identity on reader-null refresh, and cover this arrival race with a cache-reuse regression.
| if (!has_children) { | ||
| RETURN_IF_ERROR(_complete_current_split()); | ||
| } | ||
| continue; |
There was a problem hiding this comment.
[P2] Keep FileNumber tied to source ranges
This continue skips FileNumber for the parent, while every generated Row Group child later falls through to the existing increment. One FE range therefore reports R files for R nonempty Row Groups (and zero for an all-empty refined source), whereas the unsplit/one-group path reports one. That makes the established file/range metric depend on a BE-local scheduling choice and renders profiles incomparable. Increment FileNumber once for the accepted source range, and expose generated children through a separately named counter if needed.
| std::unique_ptr<FileReader> reader; | ||
| RETURN_IF_ERROR(create_file_reader(&reader)); | ||
| DORIS_CHECK(reader != nullptr); | ||
| RETURN_IF_ERROR(reader->init(_runtime_state)); |
There was a problem hiding this comment.
[P2] Keep split planning inside the lifecycle profile hierarchy
prepare_split()'s TableReader timer has ended when this method runs, and the method opens neither the parent total nor CreateReaderTime; its init and the 0/many-child or build-error close paths also bypass the common FileReader lifecycle timers. Nested OpenReaderTime/Parquet ParseFooterTime can therefore contain the primary planning cost while their parents omit it, materially underreporting scans that use this feature. Use the existing TableReader/create/init/close scopes here, and cover 0/1/many plus error paths.
|
run buildall |
|
/review |
There was a problem hiding this comment.
Automated review conclusion: requesting changes on head cb739b37466897fc363a7066edbddbdbb6545284 for one P1 correctness regression and one P2 performance regression; details are inline.
Critical checkpoints:
- Split coordination and lifecycle: traced raw-source claims, child publication, wait/wakeup, stop, ignored errors, zero/one/many children, progress transfer, and reader ownership. The one-consumer lifecycle amplification is called out inline; no additional nonduplicate issue survived.
- Reader and table semantics: traced native Parquet planning, no-column execution, Hudi/Paimon hybrid dispatch, Iceberg/Paimon deletes, COUNT paths, schema history, and GLOBAL_ROWID identity. The positive-row root-only regression is called out inline.
- Metadata, cache, and row coordinates: checked stable identity, shared-footer ownership, exact row-group IDs, chunk validation, row prefixes, page/cache coordinates, Condition Cache digests, and late Runtime Filters. Remaining concerns are either handled or already covered by existing discussions.
- Performance and observability: checked scanner counts, source/file accounting, shared read deltas, registry metrics, and lifecycle timers. Existing live threads duplicate-fence the other concrete concerns.
- Tests and coverage: reviewed all changed tests and identified the two missing regression shapes inline. Builds and tests were not run per the supplied review instructions; the authoritative diff passes
git diff --check. - User focus: no additional user-provided focus was supplied.
Convergence status: converged. Three complete normal-review rounds plus separate risk-focused passes covered all 33 authoritative changed files. Every Round 3 reviewer returned NO_NEW_VALUABLE_FINDINGS; all candidates were adjudicated and deduplicated against the live discussions, and the live head/base still match the authoritative bundle.
| } | ||
| size_t group_start = std::numeric_limits<size_t>::max(); | ||
| size_t group_end = 0; | ||
| for (size_t column_id = 0; column_id < row_group.columns.size(); ++column_id) { |
There was a problem hiding this comment.
[P1] Preserve positive-row root-only Parquet files when refining splits. The native schema contract explicitly accepts a required root with zero physical fields for metadata-only COUNT(*), and a row group with num_rows > 0 and no chunks passes metadata validation. The ordinary full-file planner selects that group and synthesizes its rows without column readers, but this loop leaves the chunk-derived range empty and turns the same file into Corruption during split planning. If a selected nonempty group has no chunks, clear any tentative children, decline refinement, and keep the initialized source reader (or support an exact child without a physical byte envelope); add a positive-row root-only COUNT regression. This is distinct from the existing zero-row-group thread: those groups can be skipped, while this one contains real rows.
| if (_current_split.is_source_split && can_refine_source_split(_current_range)) { | ||
| std::vector<FileScanSplit> generated_splits; | ||
| bool was_split = false; | ||
| const auto split_status = _table_reader->build_physical_splits( |
There was a problem hiding this comment.
[P2] Skip row-group refinement when only one scanner can consume the children. In serial mode, or when max_file_scanners_concurrency=1 yields one constructed scanner, this call initializes and closes a planning reader, then makes that same scanner open the generated children one by one. That turns one reader lifecycle into one planning lifecycle plus R child lifecycles without any parallelism; the shared footer avoids reparsing metadata, but each child still repeats physical-delegate creation, schema/mapping/filter/request setup, and row-group planning. Gate refinement on the actual constructed scanner count and keep the ordinary unsplit path when it is below two; cover a multi-row-group file with one scanner.
2 participants
Proposed changes
This PR adds the BE-only part of Parquet row-group split refinement. It assumes FE schedules splits from the same file to the same BE instance.
The generated children share the source descriptor and materialize only their row-group bounds when claimed.
Verification
-Werror.arrow/extension/parquet_variant.hwhile compiling unchanged v1 Iceberg code.