JunRuiLee opened a new pull request, #463:
URL: https://github.com/apache/paimon-rust/pull/463
### 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):**
```rust
// 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):**
```text
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
