feat(reader): merge-aware pushdown gate, row-group pruning, and read-volume counters - #708
Merged
Merged
Conversation
…able type A file slice with no log files has no merge: no log record can supersede a base row and no delete block can remove one, so nothing can flip a pushed predicate's outcome. The gate on `ReaderContext` answers only the table-level question -- CoW, or MOR whose predicate is confined to immutable primary keys -- and so understates what is safe. On a MOR table with a data-column predicate, every base-only slice lost pushdown: the base read returned whole columns for a post-merge filter to discard, having read every byte of them. Add the split-level question next to it, `can_push_row_filter_for_split()`, and wire the base read to that instead. CoW is the table-wide case of the same rule. A slice that does have log files is unaffected: it still needs the table-level gate to be open. The declarative harness had been standing in for the missing rule by declaring a base-only slice COPY_ON_WRITE regardless of the fixture's real table type. It now passes MERGE_ON_READ, which is what those fixtures are, so the eight base-only filter cases exercise the split gate rather than an is_cow() branch that was never true in production for them. Those cases are exact-row asserts and poison if the filter is not applied: reverting the gate to the table-level one fails all eight. They are renamed from `*_cow` to `*_base_only` to say what they now cover. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…table type
The previous commit widened the base-read gate to `can_push_row_filter() || no
log files`. Keeping the table-type disjunct inside it hides what the rule
actually is, and leaves `is_cow()` load-bearing for a decision that has nothing
to do with the table type: a CoW slice is safe because it carries no log files,
not because of what `hoodie.table.type` says. Java agrees --
`SparkFileFormatInternalRowReaderContext.getSchemaAndFiltersForRead` branches on
`getHasLogFiles()` and never on the table type.
So the gate becomes `base_read_pushdown_is_safe() = no log files on the split ||
mor_pk_safe`, and `ReaderContext::can_push_row_filter` goes away. Same outcomes:
a CoW slice has no log files and takes the first branch on its own. The doc
comment records why the split's own file list is used rather than
`ReaderContext::has_log_files` (different fact, different source, and a safety
gate should not rest on a caller-supplied boolean), and why Java's bootstrap tier
has no counterpart here.
The parquet log-block gate now reads `mor_pk_safe` directly. A log block only
exists on a slice that has log files, so the base gate's first disjunct is false
there by construction and its condition already reduced to exactly this; naming
it says what the log path requires instead of relying on a base-read predicate to
collapse the same way. No behaviour change -- the disjunct it drops, `is_cow()`,
could only be true for a CoW table with log blocks, which does not exist.
`ReaderContext::is_cow` goes with it: the gate was its only caller, and a table
type that no longer decides anything about the read is better absent than kept
warm for a future caller to misuse.
`builder_mor_pk_safe_{true,false}_*` were asserting PK-safety on a split with no
log files, where the gate is open regardless; they now use a split that merges,
so they test what their names say. Without that fix the `false` case fails.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… told to do Whether a predicate was pushed is not the same question as what pushing bought. A parquet `RowFilter` can be installed on every file and still read every byte of it: the filter decides per row, after the predicate columns are decoded. Nothing on this path reports that, so a read that skipped nothing and a read that skipped almost everything look identical from outside. `ReadVolume` hangs off `Storage` -- one per file-group read, so its scope is exactly that read -- and carries bytes fetched, round trips, row groups scanned against row groups the file has, rows the file has, and rows the stream yielded. `Storage::read_volume()` hands a consumer a clone to read once the stream has drained. Bytes and calls are counted in a `CountingReader` wrapped around the parquet object reader, at the `AsyncFileReader` boundary. That placement is what makes them exact and cache-independent: a warm re-read reports the same bytes as a cold one, which wall-clock does not. It also has to be per-read rather than inside the object store, which is shared between readers. The file's own shape is taken from the footer the reader has already fetched, so it adds no IO. `a_row_filter_that_rejects_everything_still_reads_the_file` is the case the counters exist for: `rows_out` goes to 0, `bytes_read` stays positive, and `row_groups_read` still covers the file. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
A `RowFilter` never skips IO. It decides per row, after the predicate columns have been fetched and decoded, so a highly selective predicate over a large base file still reads the file. Nothing on this path could skip a row group. `RowGroupSelector` is that missing mechanism: a caller-supplied closure over the file's parsed footer returning the row groups to keep, `None` meaning "no opinion". It runs against metadata the reader has already fetched, so consulting it costs nothing, and it runs before the row filter is installed, so the filter only ever sees groups that survived. It reaches a read the same way the filter does -- `BaseFileReadOptions::with_row_group_selector`, or `HoodieFileGroupReaderBuilder::with_row_group_selector` onto the reader context. Both mechanisms share ONE safety gate, bound to a local so the sharing is structural rather than two call sites that happen to agree. Pruning is the one that must not be left behind if that ever drifts: a row filter still sees every row, while a pruned row group is gone before the merge could have updated one of its rows into a match. A selector the gate refuses is counted (`row_group_selector_suppressed`), because a suppressed selector and an absent one both read as zero calls, and only the counter separates them. `read_schema` is new on `BaseFileReader` for the base read's schema-evolution probe, which was opening a whole stream to take its schema and dropping it. The parquet override answers from the footer instead. Correctness for the counters, not tidiness: the probe went through the same read path it was measuring, so a three-row-group file reported six row groups scanned before a single data byte moved, and the schema-evolution probe paid for a decode setup nobody polled. Falsified by disabling `with_row_groups`: the two pruning cases then return all three rows instead of one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
yihua
reviewed
Sep 3, 2026
|
|
||
| /// Answered from the footer alone: no stream is built, and no read-volume | ||
| /// counter moves for a call that reads no data. | ||
| fn read_schema<'a>( |
Contributor
There was a problem hiding this comment.
non-blocking: get_schema converts with None key-value metadata, so a file carrying ARROW:schema footer metadata reports different types than the stream the default impl consults (the current caller only uses field names, so nothing breaks today). Deriving the answer from ArrowReaderMetadata::try_new on the already-fetched footer would make the override exactly the stream schema at the same IO cost.
| /// so the selector runs. Pairs with the case above -- same file, same | ||
| /// selector, opposite outcome from `mor_pk_safe` alone. | ||
| #[tokio::test] | ||
| async fn a_pk_safe_predicate_lets_the_selector_run_on_a_merging_slice() { |
Contributor
There was a problem hiding this comment.
non-blocking: these gate tests stop at base_file_source, and the one position-merge harness case has no selector, so the claim that RowNumber stays absolute under with_row_groups (which the pruning-under-merge safety rests on) is only guaranteed by parquet-rs internals today. An end-to-end case with a selector, log files, and use_record_position asserting exact merged rows would pin that invariant against future parquet upgrades.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #708 +/- ##
==========================================
- Coverage 79.34% 79.25% -0.10%
==========================================
Files 122 122
Lines 11988 12069 +81
==========================================
+ Hits 9512 9565 +53
- Misses 2476 2504 +28 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Three related changes to base-file predicate pushdown in
reader_v2, smallest first. Each commit stands alone.1. A base-only slice may push, whatever the table type
A file slice with no log files has no merge: no log record can supersede a base row and no delete block can remove one, so nothing can flip a pushed predicate's outcome. The gate on
ReaderContextanswered only the table-level question — CoW, or MOR whose predicate is confined to immutable primary keys — so on a MOR table with a data-column predicate, every base-only slice lost pushdown: the base read returned whole columns for a post-merge filter to discard, having read every byte of them.The declarative harness had been standing in for the missing rule by declaring a base-only slice
COPY_ON_WRITEregardless of the fixture's real table type. It now passesMERGE_ON_READ, which is what those fixtures are, so the eight base-only filter cases exercise the split-level gate rather than anis_cow()branch that was never true in production for them. Those cases assert exact rows and poison if the filter is not applied — reverting the gate fails all eight.2. The gate is about whether the read merges, not about the table type
Keeping the table-type disjunct hides what the rule is: a CoW slice is safe because it carries no log files, not because of what
hoodie.table.typesays. Java agrees —SparkFileFormatInternalRowReaderContext.getSchemaAndFiltersForReadbranches ongetHasLogFiles()and never on the table type. So the gate becomeswith the same outcomes, and
ReaderContext::can_push_row_filter/is_cowgo away. The doc comment records why the split's own file list is used rather thanReaderContext::has_log_files(different fact, different source — a safety gate should not rest on a caller-supplied boolean), and why Java's bootstrap tier has no counterpart here.The parquet log-block gate now reads
mor_pk_safedirectly. A log block only exists on a slice that has log files, so the base gate's first disjunct is false there by construction and the condition already reduced to exactly this. No behaviour change: the disjunct it drops could only be true for a CoW table with log blocks.3. Row groups can be pruned, and reads can be measured
A
RowFilternever skips IO — it decides per row, after the predicate columns have been fetched and decoded. Nothing on this path could skip a row group.RowGroupSelector: a caller-supplied closure over the file's parsed footer returning the row groups to keep,Nonemeaning "no opinion". It runs against metadata the reader has already fetched, so consulting it costs nothing, and before the row filter is installed, so the filter only ever sees groups that survived. It reaches a read throughBaseFileReadOptions::with_row_group_selectororHoodieFileGroupReaderBuilder::with_row_group_selector.ReadVolumeonStorage(one per file-group read) counts bytes fetched, round trips, row groups scanned vs. row groups the file has, rows the file has, and rows the stream yielded — plusrow_group_selector_callsandrow_group_selector_suppressed. Bytes and calls are counted in aCountingReaderat theAsyncFileReaderboundary, which makes them exact and cache-independent: a warm re-read reports the same bytes as a cold one, which wall-clock does not.BaseFileReader::read_schemais new (with a default implementation, so no other format changes) for the base read's schema-evolution probe, which was opening a whole stream to take its schema and dropping it. The parquet override answers from the footer instead. This is correctness for the counters, not tidiness: the probe went through the same read path it was measuring, so a three-row-group file reported six row groups scanned before a single data byte moved.API changes
ReaderContext::can_push_row_filter,ReaderContext::is_cowbase_read_pushdown_is_safe(); the table type no longer decides anything about a readRowGroupSelector,ReadVolume,Storage::read_volume()storageBaseFileReadOptions::with_row_group_selector,HoodieFileGroupReaderBuilder::with_row_group_selectorBaseFileReader::read_schemaNo behaviour change for
Tableor DataFusion reads:resolve_reader_contextinstalls neither a filter nor a selector, so those paths push nothing, exactly as before.How are the changes test-covered
Each change was falsified by reverting it and watching the corresponding tests go red, not only by watching them pass:
calls == 0 && suppressed == 1RowFilterrejecting every rowrows_out == 0whilebytes_read > 0, which is the case the counters exist forcargo test --no-fail-fast --all-targets --all-features --workspaceandcargo clippy --all-targets --all-features --workspace --no-deps -- -D warningsare clean.