perf(deletes): apply equality deletes via a hashed key set inside the parquet RowFilter - #2961
perf(deletes): apply equality deletes via a hashed key set inside the parquet RowFilter#2961brgr-s wants to merge 4 commits into
Conversation
… parquet RowFilter Upstream-candidate version, based on apache/main. Does not include the fixes from apache#2873 and apache#2630; see local/eq-deletes.md.
There was a problem hiding this comment.
Thanks for looking at this @brgr-s. I inlined some comments from my first pass, and I have a number of test gaps to suggest:
Scan predicate combined with an equality delete, end to end. Every new eqd_* test in row_filter.rs uses the eqd_read helper (row_filter.rs:1445), which never calls .with_predicate(...); the only with_predicate usage in this file is in a pre-existing, unrelated test (row_filter.rs:388). This is exactly the code path this PR rewired: final_predicate and eq_delete_sets used to be ANDed into one bound Predicate before binding; now they're two separately-pushed ArrowPredicates in row_filter_predicates (pipeline.rs:388-451). Nothing exercises both branches being taken in the same scan and checking the intersection is correct. Same gap one level up: no test combines a positional delete, an equality delete, and a scan predicate in one real read and checks output rows; the existing "both delete types together" test (caching_delete_file_loader.rs, formerly test_build_equality_delete_predicate_case_sensitive-adjacent) only asserts build_equality_delete_sets(...).is_ok(), not actual filtered output.
Two equality-delete files, same equality_ids, different on-disk parquet types, through the real load path. This is the happy-path counterpart to test_union_rejects_mismatched_layout: that test only proves union() rejects a mismatch when two EqDeleteSets are hand-built with different types. Nothing proves the realistic case, that evolve_schema/RecordBatchTransformer normalizes two physically-different-typed delete files (e.g. one written as int32, another as int64, same equality_ids) to the same recorded type before they ever reach union(), so the grouping-by-id-only key in build_equality_delete_sets (delete_filter.rs:213) stays safe. Right now that guarantee is established only by reading the code, not by a test; if evolve_schema's casting behavior ever regresses, nothing would catch it before union() starts erroring in production.
Equality-delete key column missing from the data file, with a non-null initial_default on that field. Ties to the row_filter.rs:127-130 finding above: add a test with a field added via schema evolution with .with_initial_default(...), an older data file written before that field existed, and an equality-delete file keyed on it, and assert on the intended behavior (spec.md:898). Worth having regardless of which way the finding gets resolved, since right now nothing pins the behavior down at all.
Decimal precision promotion through the probe. test_eq_delete_promotes_data_type_before_probe only covers Int32 -> Int64. Datum::to has an explicit branch that errors on decimal scale mismatches (datum.rs:1168-1182) and a separate branch for same-scale precision promotion; decimal columns are a realistic equality-delete key (e.g. monetary amounts) and Iceberg's supported decimal evolution (precision widening) goes through a different code path than the int case already tested.
Float/double equality-delete columns. The spec explicitly disallows these ("Float and double columns cannot be used as delete columns in equality delete files", spec.md:853), but nothing in this PR checks for or tests that. EqDeleteKey derives Hash/Eq on Datum, so a float delete column would hit IEEE-754 float equality/hashing (NaN, -0.0 vs 0.0) that the spec restriction exists specifically to avoid. Worth a test asserting a clear, explicit rejection rather than relying on callers never producing one.
Empty EqDeleteSet skip path. if set.is_empty() { continue; } (row_filter.rs:90-92) is untested directly: an equality-delete file that parses to zero keys (e.g. legitimately empty file) should produce no ArrowPredicate for that set and not affect row-group/row-selection filtering. Same code path also needs to handle every key column of a set being simultaneously absent from the data file (column_indices ends up empty, ProjectionMask::leaves(parquet_schema, [])); worth confirming RowFilter still reports the correct row count with a zero-column projected batch.
Lower confidence, likely pre-existing behavior this PR doesn't touch, but the new test suite had a natural opportunity to cover it: an equality-delete file whose key column was later dropped from the table. Spec.md:898, first sentence, requires the dropped column still be used for matching. evolve_schema's target schema is task.schema (the current schema), which by definition no longer declares a dropped field; whether RecordBatchTransformer still carries a dropped-but-still-referenced equality_ids column through is worth a test even though the mechanism predates this PR.
| let Some(pos) = batch_positions[i] else { | ||
| columns.push(vec![None; num_rows]); | ||
| continue; | ||
| }; |
There was a problem hiding this comment.
When an equality-delete key column is absent from a data file's parquet schema (schema evolution: the column was added after this file was written), the probe treats it as null for every row: columns.push(vec![None; num_rows]); continue;. Per spec.md:898, "the column value is read for older data files using normal projection rules" for exactly this case, and iceberg-rust already has a mechanism for "normal projection rules" for a missing column: RecordBatchTransformer resolves initial_default (see record_batch_transformer.rs, "Rule #3"), used for the data file's own output columns at pipeline.rs:494. The eq-delete probe runs earlier, inside the RowFilter, on raw decoded batches, before RecordBatchTransformer ever sees them, so it never consults initial_default and always substitutes null instead. Java's DeleteFilter avoids this by construction: it compares against record, which is built by the regular row reader against requiredSchema (DeleteFilter.java:99-101, 202-209), i.e. defaults are already resolved before the delete-set lookup runs at all. Concretely: a required column added later with a non-null initial_default, later used as an equality-delete key, will never delete matching rows in data files written before that column existed, because every row in those files probes as null instead of the default value. Low likelihood in practice, but it's a silent, spec-observable divergence, not a crash, so worth a test either way.
There was a problem hiding this comment.
I followed your trace and you are correct, this is a spec divergence.
Digging deeper: the old code was also spec-divergent, in two different ways:
- required missing columns lead to all rows beeing dropped for that data file
- optional missing columns on the other hand "do nothing"
This PR keeps all rows in both cases, but scoped to that specific layout group. All rows survive, but if other delete files apply (eq deletes with different layouts, pos deletes), they "can" delete rows... I'd argue that this a slight improvement (maybe ressurect some rows vs. hard dropping all), but the fact you pointed out remains: it is spec divergent, it needs to be made obvious, and it needs to be fixed (but maybe not on this PR). I think this would be worth filing an issue, even if this PR does not get accepted.
| }; | ||
|
|
||
| combined_predicate = combined_predicate.and(predicate); | ||
| let layout = set.fields.iter().map(|(_, id, _)| *id).collect(); |
There was a problem hiding this comment.
Grouping key is set.fields.iter().map(|(_, id, _)| *id).collect(), field id only, no type. In isolation this looks like it could let two type-mismatched equality-delete sets reach union() and hard-fail the read. It can't in practice: CachingDeleteFileLoader evolves every equality-delete file's batch stream against task.schema before parsing it into an EqDeleteSet (caching_delete_file_loader.rs:333-343, via BasicDeleteFileLoader::evolve_schema / RecordBatchTransformer, which casts present-but-differently-typed columns, record_batch_transformer.rs:810). So every set sharing the same scan's task.schema already carries the same recorded type for the same field id by construction, and grouping by id only mirrors Java's own grouping (Sets.newHashSet(delete.equalityFieldIds()), DeleteFilter.java:194), which also doesn't carry type in the key, for the same reason: Java projects every delete file into requiredSchema up front instead of comparing delete-file-recorded types against each other. Given that, the union() type check and its comment are effectively dead code / defense-in-depth for an invariant that already holds structurally, not the live bug I'd have guessed from reading this function alone. Worth a comment saying why the invariant holds (ties to evolve_schema in the other file) instead of "should a change break this," since the current comment reads as a guess rather than a traced guarantee. Not worth blocking on.
There was a problem hiding this comment.
I would improve the comment and explain why the invariant holds, as you suggested. I'd argue against dead code, though.
The invariant belongs to the caller, not union(). Also, DeleteFilter's state is Arc, CachingDeleteFileLoader holds one, and ArrowReader is Clone. FileScanTask is also "very pub" and "Serde", so a hand-built or deserialized plan, or a re-used reader, can put two tasks with different schemas through the same "path-keyed" cache. That is very unlikely, but it is reachable though the public API. I'd rather keept it.
| let mut result = Vec::with_capacity(groups.len()); | ||
| for mut sets in groups.into_values() { | ||
| if sets.len() == 1 { | ||
| result.push(sets.pop().unwrap()); | ||
| } else { | ||
| let mut combined = (*sets[0]).clone(); | ||
| for other in &sets[1..] { | ||
| // `union` checks if `other`s' layout matches `combined`, | ||
| // which is currently always the case. This fails should a change | ||
| // break this current invariant. | ||
| combined.union(other)?; | ||
| } | ||
| result.push(Arc::new(combined)); | ||
| } | ||
| } |
There was a problem hiding this comment.
build_equality_delete_sets runs once per data-file scan task, and whenever a data file has more than one delete file sharing a layout, it clones the first set's entire HashSet<EqDeleteKey> and unions in the rest, from scratch, every time. The PR's own benchmark numbers (7,140 and 124,750 "delete refs") are exactly the count of (data-file, delete-file) edges this repeats over, so the real cost is closer to O(data_rows + delete_refs) than the claimed O(data_rows + delete_keys) once delete files span many data files, which is the compaction case motivating the PR. Java doesn't have this cost at all: it builds one StructLikeSet per equality-id group once per DeleteFilter (DeleteFilter.java:191-211), not per data file. Fix: pass Vec<Arc<EqDeleteSet>> per layout to the row filter and probe against all of them (row is deleted if it matches any), instead of merging into one owned set per task. Arc::clone is a refcount bump; the HashSet clone is not.
There was a problem hiding this comment.
You are absolutely correct about the O(data_rows + delete_keys) claim, it is incorrect.
The correct bound ist O(data_rows + applicable_delete_refs).
Or, if we set m = applicable delete files sharing a layout, K = keys per delete file, D = rows in the data file
Old = D*m*K
New = D + m*K
My intuition on your suggestion Vec<Arc<EqDeleteSet>> is (with c_prope cost for lookup, c_clone cost for Clone):
merge: m*K*c_clone (build, once) + D*c_probe (one lookup per row)
N-set: 0 (Arc bumps) + D*m*c_probe (m lookups per row)
Merging wins when m*K*c_copy < D*(m-1)*c_probe, or, D/K > (m/(m-1))*(c_clone/c_probe).
When m grows, (m/(m-1)) -> 1, so the threshold gets smaller (assuming relativ constant c_clone vs c_probe, but I think thats Ok), and merging gets "relativly better" in the "fanout case" (for m=1, we never merge). N-set probing wins when D/K ~ small, but I'd argue that delete file application is not the dominant cost for reading in that case.
Regarding Java, I took a look at DeleteFilter.java:191-211, and I think you have applyEqDeletes and isInDeleteSets in mind. isInDeleteSets lives in a DeleteFilter instance, which is constructed per data file. I might be looking at the wrong thing, however?
| let predicate_func = | ||
| move |batch: RecordBatch| -> std::result::Result<BooleanArray, ArrowError> { | ||
| let num_rows = batch.num_rows(); | ||
|
|
||
| // Change each key column into `Datum`s once, promoting to the | ||
| // table type so the keys match the parsed delete keys under schema | ||
| // evolution. A column absent from this file reads as all-null. | ||
| let mut columns: Vec<Vec<Option<Datum>>> = Vec::with_capacity(num_cols); | ||
| for (i, target_type) in target_types.iter().enumerate() { | ||
| let Some(pos) = batch_positions[i] else { | ||
| columns.push(vec![None; num_rows]); | ||
| continue; | ||
| }; | ||
| let array = batch.column(pos); | ||
| let source_type = arrow_type_to_type(array.data_type()) | ||
| .map_err(|e| ArrowError::ComputeError(e.to_string()))?; | ||
| let source_primitive = source_type | ||
| .as_primitive_type() | ||
| .ok_or_else(|| { | ||
| ArrowError::ComputeError( | ||
| "equality delete key column is not a primitive type" | ||
| .to_string(), | ||
| ) | ||
| })? | ||
| .clone(); | ||
| let needs_promotion = source_type != *target_type; | ||
| let literals = arrow_primitive_to_literal(array, &source_type) | ||
| .map_err(|e| ArrowError::ComputeError(e.to_string()))?; | ||
|
|
||
| let mut column = Vec::with_capacity(num_rows); | ||
| for literal in literals { | ||
| let datum = match literal { | ||
| Some(literal) => { | ||
| let primitive = | ||
| literal.as_primitive_literal().ok_or_else(|| { | ||
| ArrowError::ComputeError( | ||
| "failed to convert to primitive literal" | ||
| .to_string(), | ||
| ) | ||
| })?; | ||
| let datum = Datum::new(source_primitive.clone(), primitive); | ||
| let datum = if needs_promotion { | ||
| datum | ||
| .to(target_type) | ||
| .map_err(|e| ArrowError::ComputeError(e.to_string()))? | ||
| } else { | ||
| datum | ||
| }; | ||
| Some(datum) | ||
| } | ||
| None => None, | ||
| }; | ||
| column.push(datum); | ||
| } | ||
| columns.push(column); | ||
| } | ||
|
|
||
| // One hash lookup per row. | ||
| let mut keep = Vec::with_capacity(num_rows); | ||
| let mut probe = EqDeleteKey(vec![None; num_cols]); | ||
| for row in 0..num_rows { | ||
| for (i, column) in columns.iter_mut().enumerate() { | ||
| // we can `take` because each cell is probed once. | ||
| probe.0[i] = std::mem::take(&mut column[row]); | ||
| } | ||
| keep.push(!set.keys.contains(&probe)); | ||
| } | ||
| Ok(BooleanArray::from(keep)) | ||
| }; |
There was a problem hiding this comment.
Per batch, per key column: arrow_primitive_to_literal (row_filter.rs:144) builds an owned Vec<Option<Literal>> for the whole column, then row_filter.rs:148-171 walks it again to build a Vec<Option<Datum>>. Two full passes and, for string/binary columns, two rounds of per-cell heap ownership, on the exact hot path this PR is trying to make cheap. Separately, row_filter.rs:158 does source_primitive.clone() inside the per-row loop even though source_primitive is fixed per column (computed once at row_filter.rs:134-142); cheap since PrimitiveType doesn't allocate, but still pointless per-row work. caching_delete_file_loader.rs:541-556 has the same double-pass/box-per-cell pattern already, unmodified by this PR, but it runs once per delete-file load; this PR adds a second copy of it on the probe side, which runs once per batch per applicable data file, i.e. far more often. arrow::row::RowConverter (crate arrow-row, already in Cargo.lock at 59.1.0, not yet a direct dep of crates/iceberg) builds comparable/hashable rows straight from Arrow arrays without this boxing, and is what DataFusion uses for multi-column join/group keys. For the common single-column case, DataFusion's InListExpr static filters (datafusion/physical-expr/src/expressions/in_list/{primitive_filter,static_filter}.rs) use a native-typed hashbrown/ahash set (datafusion_common::HashSet, see datafusion/common/src/lib.rs:119) instead of boxed values in std::collections::HashSet with the default SipHash, which is what EqDeleteSet::keys uses here. Worth noting Java's generic DeleteFilter doesn't push equality-delete filtering into the file decoder at all, it applies a row-by-row Predicate<T> after full materialization; pushing it into parquet-rs's RowFilter the way this PR does is a real advantage over that, so this finding is about the representation used inside the predicate, not the decision to use a RowFilter predicate.
There was a problem hiding this comment.
Regarding RowConverter, I am hesitant to put this in this PR. It is not a small change and I think it should propably go in a follow-up, with benchmarks that prove that it improves performance beyond what this PR does. Same reasoning for not trying to remove arrow_primitive_to_literal.
For double pass and the double String allocation, I have an idea that I will push today.
| )?; | ||
|
|
||
| let row_filter = ArrowReader::get_row_filter( | ||
| row_filter_predicates.push(ArrowReader::build_scan_predicate( |
There was a problem hiding this comment.
Also line 440.
Scan predicate is always pushed before the eq-delete predicates, unconditionally. parquet-rs applies RowFilter predicates in list order, threading a shrinking RowSelection between them (arrow-rs parquet/src/arrow/arrow_reader/filter.rs:139-149), so a cheap/selective predicate run first prunes work for everything after it. DataFusion's ParquetSource reorders predicates by estimated column byte-size before building the RowFilter for exactly this reason (datafusion/datasource-parquet/src/row_filter.rs:33-52). A hash-probe over one or two narrow key columns is often cheaper and more selective than an arbitrary scan predicate; this PR never reorders to take advantage of that.
There was a problem hiding this comment.
TBH, I'd keep it that way at the moment.
The machinery does exist in DataFusion, but it is opt-in (reorder_filters defaults to false).
The heuristic is compressed column byte size, as you state, so a cpu-heavy predicate over column A cannot be distinguished from a cheaper hash probe over the same column A. Across different columns, a CPU heavy predicate over a narrow column sorts before a one lookup probe over a wide one, regardless which one is cheaper or more selective.
It is a good proxy for decoder effort, which is frequently the right thing to optimize, but it is not a proxy for evaluation cost, which is what this PR aims to improve. Getting the order right would need a cost model for selectivity or per-predicate cost.
I think this is more a follow-up situation if a workload is discovered where the order measurably matters.
| let field_id_map = | ||
| Self::resolve_field_id_map(parquet_schema, arrow_schema, use_position_fallback)?; |
There was a problem hiding this comment.
resolve_field_id_map recomputes the parquet-schema-to-field-id map whenever eq_delete_sets is non-empty, even though pipeline.rs already computed an equivalent map for the scan predicate a few lines earlier via ArrowReader::build_field_id_set_and_map (projection.rs:49-62, which itself calls resolve_field_id_map). Same schema both times; minor duplicated per-file work, not per-row, so low severity, but easy to thread through instead.
There was a problem hiding this comment.
Threading it through is do-able, but it ties the predicate arm to the eq delete arm. If neither is present, we should skip calculating the schema-to-field-id-map. I felt like this was an odd shape, and the current form is more readable, while the impact is low.
|
@mbutrovich Thank you for your thourough review, really appreciate it. I will address your points shortly :) |
|
@mbutrovich I worked through your comments and will follow up with commits adressing where we agree. Specifically, because it is not inlined: the test gap is real, so more tests! I just have three comments on the test gaps: float/double equality-delete columns:
equality-delete key column later dropped from the table: this is pre-existing, but nothing would reach the probe: the scan would fail outright on I checked |
2 participants
Which issue does this PR close?
What changes are included in this PR?
Equality deletes are currently applied by turning every delete row into a predicate
expression and OR-ing them into a single tree, which is then evaluated against every row
of every data file the delete file applies to. The cost is
O(data_rows * delete_keys).This replaces the predicate tree with a hash probe:
EqDeleteSet(arrow/caching_delete_file_loader.rs) holds the delete keys as aHashSet<EqDeleteKey>, whereEqDeleteKey(Vec<Option<Datum>>)is one delete row's equality-field values. The set also carries the layout it was built for asVec<(name, field_id, Type)>.EqDeleteSet::union, which returnsErrwhen the layouts differ rather than trusting the caller.build_equality_delete_predicates(arrow/reader/row_filter.rs) turns each set into oneArrowPredicatepushed into parquet'sRowFilter, with aProjectionMaskrestricted to that set's equality columns. Only those columns are decoded to evaluate the filter, and rows are dropped during decode rather than after materialisation.Cost becomes
O(data_rows + applicable_delete_refs).We hit this compacting merge-on-read tables written by Flink upsert (
write.upsert.enabled, format-version 2), where each commit contributes an equality-delete file and the reader ends up applying a large fraction of them to every data file. Two test tables, timing the read+rewrite phase:This is the same core idea as #2343 by @t3hw — replace the per-delete-row predicate tree with a hash-set check — and that PR deserves the credit for identifying the problem and the fix. It was closed by the stale bot after 30 days of inactivity. @t3hw has since confirmed that their organisation moved off iceberg-rust and nobody is pushing it forward, and encouraged filing this.
This is an independent implementation written against current
mainrather than a revival of that branch, because #2343 cannot be pushed to from outside.NOTE 1
Does not include the fixes from #2873 and #2630 and therefore still carries the same bugs these PRs aim to fix. I decided against folding in those changes in this PR. I would rather rebase my PR after #2873 and #2630 have been merged.
NOTE 2
RowGroupMetricsEvaluator::not_eqreturnsROW_GROUP_MIGHT_MATCHunconditionally anyways.0 != -0and!=for differentNaNs. This only "hits" non-conforming writes. Might be worth a follow-up. See CommentRowConverter. This might improve performance even more, but is a bigger rebuild. See CommentAre these changes tested?
Yes — 11 new tests:
Delete-set construction and semantics (
caching_delete_file_loader.rs,delete_filter.rs):test_equality_delete_set_preserves_null_rowstest_equality_delete_set_matches_null_delete_valuetest_equality_delete_set_multiple_columnstest_equality_delete_set_multiple_delete_rowstest_build_equality_delete_sets_mixed_ids_not_mergedtest_build_equality_delete_sets_same_layout_unionedtest_union_rejects_mismatched_layoutEnd-to-end filtering through the reader (
reader/row_filter.rs):test_eq_delete_single_column_filters_matching_rowstest_eq_delete_multi_column_keeps_null_and_partial_matchestest_eq_delete_promotes_data_type_before_probetest_eq_delete_distinct_layouts_apply_independentlyAlso verified on the rebased branch:
cargo test -p iceberg --lib— 1500 passed, 0 failedcargo clippy -p iceberg --all-targets --all-features— cleancargo fmt --check— cleancargo public-api -p iceberg --all-features -ssdiffs empty againstcrates/iceberg/public-api.txt: no public API change (the new types arepub(crate)), sopublic-api.txtneeds no regeneration.AI Disclosure
I used AI to
iceberg-rustGitHub