Apply exact predicate filtering on the primary-key merge read path#463
Open
JunRuiLee wants to merge 7 commits into
Open
Apply exact predicate filtering on the primary-key merge read path#463JunRuiLee wants to merge 7 commits into
JunRuiLee wants to merge 7 commits into
Conversation
ff9fc17 to
31f5479
Compare
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
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.
Purpose
TableRead::with_filteris enforced exactly on append and data-evolution read paths (#448), but the primary-key merge path silently dropped non-key predicate conjuncts and never re-applied them — direct API consumers get rows that don't match the filter.Worse, scan planning pruned files of merge-read PK tables by non-key value stats. That can drop the file holding the newest version of a key and resurrect a stale version from a surviving file: wrong data that no downstream re-filtering can repair, because the stale row itself matches the predicate. Java's
KeyValueFileStoreScannever prunes individual files within an overlapping run for exactly this reason; the per-file pruning was a porting gap. It triggers on files carrying real value stats — e.g. Flink/Java-written PK tables read by Rust (Rust-written PK files currently carry empty value stats, so pruning fails open there).Reproduce
1. Non-key filter silently ignored (read side):
(locked by test:
kv_read_applies_non_pk_filter_exactly)2. Stats pruning resurrects a stale version (scan side, wrong data):
(locked by test:
test_pk_table_stats_pruning_ignores_non_key_conjuncts)Changes
kv_file_reader.rs): key conjuncts still push below the merge (unchanged — they prune safely); the full data predicate is now enforced by an exact post-merge residual filter using the sharedarrow::residualevaluator, mirroring pypaimon's post-mergeFilterRecordReaderand Java'sexecuteFilter. Predicate columns absent from the projection are widened into the internal merge read — so they receive their configured merge semantics (e.g. aggregationsum) before the filter sees them — and are projected away by the existing read_type reorder.table_scan.rs): per-file value-stats pruning for merge-read PK tables is restricted to key conjuncts (all versions of a key share the key columns, so key pruning can never separate versions). Deletion-vector tables are exempt: they read raw with per-row masks, full pruning stays safe. Limit-hint logic unchanged.Filter placement is load-bearing, not a convenience: dropping a row by a non-key value before dedup can change which version of a key survives. Tests lock this from both directions — a filter matching only a superseded version returns 0 rows, and a filter matching only an aggregation-merged value (sum absent from every input row) returns the merged row.
Future optimization (out of scope): Java's
MergeFileSplitReadpushes the full filter to non-overlapping sections viaIntervalPartition. Rust already covers the largest case through raw-convertible splits bypassing the merge reader entirely; per-section pushdown for mixed splits can follow. It is an I/O optimization only — the residual stays either way.Tests
12 new E2E tests through the public read path (write → commit → scan →
to_arrow) covering dedup / partial-update / aggregation engines, unprojected predicate columns, compound and string predicates, empty projection (COUNT(*)) row counts, superseded-value regression, AlwaysFalse, and schema-evolution null-fill semantics; plus scan-layer pruning gate tests and conjunct-projection unit tests.cargo test -p paimon --lib: 1191 passed.cargo test -p paimon-datafusion --test pk_tables: 51 passed.API and Format
No public API or storage-format change (all touched types are crate-internal).
Documentation
with_filterdocs onReadBuilder/TableReadupdated: exactness now claimed for all three read paths, with the existing per-format exception pointer retained.