Skip to content

Apply exact predicate filtering on the primary-key merge read path#463

Open
JunRuiLee wants to merge 7 commits into
apache:mainfrom
JunRuiLee:feat/pk-merge-residual-filter
Open

Apply exact predicate filtering on the primary-key merge read path#463
JunRuiLee wants to merge 7 commits into
apache:mainfrom
JunRuiLee:feat/pk-merge-residual-filter

Conversation

@JunRuiLee

Copy link
Copy Markdown
Contributor

Purpose

TableRead::with_filter is 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 KeyValueFileStoreScan never 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):

// PK table: id INT PRIMARY KEY, value INT
// commit 1: (1, 10), (2, 20), (3, 30)
// commit 2: (1, 11), (2, 21), (3, 31)   -- same keys, overlapping -> merge read

let rb = table.new_read_builder();
rb.with_filter(PredicateBuilder::new(&fields).equal("value", Datum::Int(21))?);
let batches = rb.new_read()?.to_arrow(rb.new_scan().plan().await?.splits())?;

// expected: [(2, 21)]
// actual (before this PR): [(1, 11), (2, 21), (3, 31)] — the value conjunct is
// dropped before the merge (correctly: pre-merge value filtering breaks dedup)
// but never re-applied after it. No error, just extra rows.

(locked by test: kv_read_applies_non_pk_filter_exactly)

2. Stats pruning resurrects a stale version (scan side, wrong data):

PK table, key id=1 written twice; files carry value stats (e.g. Flink-written):
  file A (older): (1, value=150)     value stats [100, 200]
  file B (newer): (1, value=50)      value stats [10, 60]

Query: WHERE value > 90

scan: file B's stats [10, 60] can't match value > 90  ->  file B pruned
merge: only file A left  ->  key 1 "current" value = 150
result: (1, 150) is returned

But the true current value of key 1 is 50 — the row (1, 150) is stale data that
matches the predicate, so no engine-side re-filtering can remove it. Java never
prunes individual files inside an overlapping run (KeyValueFileStoreScan) for
exactly this reason.

(locked by test: test_pk_table_stats_pruning_ignores_non_key_conjuncts)

Changes

  • Read side (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 shared arrow::residual evaluator, mirroring pypaimon's post-merge FilterRecordReader and Java's executeFilter. Predicate columns absent from the projection are widened into the internal merge read — so they receive their configured merge semantics (e.g. aggregation sum) before the filter sees them — and are projected away by the existing read_type reorder.
  • Scan side (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 MergeFileSplitRead pushes the full filter to non-overlapping sections via IntervalPartition. 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_filter docs on ReadBuilder/TableRead updated: exactness now claimed for all three read paths, with the existing per-format exception pointer retained.

@JunRuiLee JunRuiLee force-pushed the feat/pk-merge-residual-filter branch from ff9fc17 to 31f5479 Compare July 6, 2026 14:58
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant