Repository navigation
fix(arrow): load equality deletes on fields nested in structs - #3453
JeonDaehong wants to merge 2 commits into
Conversation
|
@mbutrovich could you take a look when you have time? This fixes loading equality deletes on nested fields (#3450); applying them on scan also needs #2906, which you've been reviewing. |
|
Converted to draft to stay within the new limit of 3 open PRs (#3465). The change is still ready for review; I will mark it ready again when one of my other PRs lands. |
a4adabb to
1fc7628
Compare
|
Rebased onto main (d2679a0) and marked ready again now that #3454 is in. Full workspace tests, clippy, doc and MSRV pass locally. #2906 is still open, so the end-to-end scan test will come with it as planned. @mbutrovich @laskoviymishka could you take a look when you have time? |
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Thanks, reading nested equality IDs through the top-level struct and binding by full name fixes the field not found load.
Before merge, I’d still want three things:
-
equality_delete_projectionshould returnResultand error with the bad field ID when an equality ID is missing from the scan schema, instead of silently dropping it. -
Two tests through
load_deletes, so both paths are covered:- missing equality ID in the schema -> error
- delete written with our
EqualityDeleteWriterons.f-> loaded predicate iss.f IS NULL OR s.f != <value>
The second one also answers the inline question.
-
Rebase on main.
security_auditis failing onRUSTSEC-2026-0385inctutils; #3519 already fixed that.
The required / initial-default sibling case, where an s{f} delete file does not contain the other struct fields, can be a follow-up rather than part of this PR.
@mbutrovich, could you also sanity-check the loader projection against #2906?
| // equality_ids columns, not all table columns. | ||
| // columns holding the equality_ids, not all table columns. | ||
| let equality_ids_vec = task.equality_ids().unwrap().to_vec(); | ||
| let projection = equality_delete_projection(&schema, &equality_ids_vec); |
There was a problem hiding this comment.
Does a delete file from our own EqualityDeleteWriter still load correctly through this projection? For s.f the writer writes a flat top-level f with id 3 (record_batch_projector.rs#L72-L75), while this asks the file for id 2.
Could we add a load_deletes-level test that writes the delete with EqualityDeleteWriter? The current test hardcodes &[2] and calls evolve_schema directly, so neither call site sees a nested id. Checking the bound field id directly would also hold up better than matching the Debug string.
There was a problem hiding this comment.
good catch, it didn't. EqualityDeleteWriter writes s.f as a top-level f (id 3), so asking the file for id 2 found nothing and s came back null. the predicate ended up as s.f IS NOT NULL, which deletes the rows where s.f is null and keeps s.f = 7.
43d3169 reads the file as the scan schema pruned to the equality fields, like Java's DeleteFilter, and moves a top-level column whose id is nested back under its parent struct first, so both layouts load the same way. test_nested_field_equality_delete_from_equality_delete_writer writes the file with EqualityDeleteWriter and goes through load_deletes; without the re-nesting it fails with that IS NOT NULL predicate. the tests now compare the bound predicate and its field ids instead of the Debug string.
| .as_struct() | ||
| .fields() | ||
| .iter() | ||
| .filter(|field| equality_ids.iter().any(|&id| contains(field, id))) |
There was a problem hiding this comment.
Should this return an error when an equality id doesn't resolve in the scan schema? As I read it, an unmatched id is filtered out silently, top-level ones too, e.g. an eq-delete column dropped while the delete file is still live. If every id drops, the batch has no columns and the delete becomes AlwaysTrue. If only some drop, rows match on fewer columns and we over-delete.
main fails with field not found, and Java's DeleteFilter fails with Cannot find required field for ID. Could we keep the error and add a missing-id test?
There was a problem hiding this comment.
yes, an id missing from the scan schema now fails with DataInvalid naming the id, before the file is read. test_equality_delete_with_id_missing_from_schema_fails covers it through load_deletes.
Equality ids may name a field nested in a struct, which the spec allows and EqualityDeleteWriter writes. Loading such a delete file failed because schema evolution only maps top-level field ids, so the delete file is now read by the top-level fields that hold the equality ids. The delete predicate also used the leaf name (`f`), which binds to a top-level column of the same name or fails; it now uses the full name (`s.f`).
…fields Equality ids missing from the scan schema now fail instead of being dropped. The delete file is read as the scan schema pruned to the equality fields, like Java's DeleteFilter, and a nested field that EqualityDeleteWriter wrote as a top-level column is put back under its parent struct first, so both layouts load.
1fc7628 to
43d3169
Compare
|
@laskoviymishka addressed all three and rebased on main, ready for another look. |
I’ll take a look shortly. :) |
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @JeonDaehong, and thanks @laskoviymishka for asking me to check this against #2906.
I applied #2906 on top of this head and ran end-to-end scans. s.f = 7 deletes only the matching row with both delete file layouts, and it leaves a top-level f = 7 row and a null s row alone.
One case is reachable with this PR alone. A delete row with a null s on a required leaf (s.g, with s optional) loads as s.g IS NOT NULL, which binds to AlwaysTrue (#3292), so the scan silently keeps the row. On main the same file fails with field not found, and on this head every other nested case still fails at the row filter until #2906 lands. So this PR turns that case from an error into a wrong result until #3306 lands. With #3306 also applied, my test for it passes. Could the PR description call this out and link #3292, so it's visible when this merges ahead of #3306?
| // columns holding the equality_ids, not all table columns. | ||
| let equality_ids_vec = task.equality_ids().unwrap().to_vec(); | ||
| let evolved_stream = BasicDeleteFileLoader::evolve_schema( | ||
| let evolved_stream = evolve_equality_delete_stream( |
There was a problem hiding this comment.
Should we check the equality ids against schema before try_start_eq_del_load claims the file? Right now the new DataInvalid error returns after insert_equality_delete has registered the receiver, so the sender is dropped and the spawned task panics at eq_del.await.unwrap(). Running test_equality_delete_with_id_missing_from_schema_fails on this head prints:
panicked at crates/iceberg/src/arrow/delete_filter.rs:269:39:
called `Result::unwrap()` on an `Err` value: RecvError(())
That is the path in #3274: the entry stays Loading, and a later read of the same delete file through the same DeleteFilter waits forever. #3274 covers I/O and parse failures in general, but this check needs only the scan schema, so it can run before any state changes. It would also run before parquet_to_batch_stream opens the file, which happens first today because it's an argument to evolve_equality_delete_stream. Could the test also check that a second load_deletes call on the same loader returns the error instead of hanging?
| /// A delete file in Java's layout: `s.f` under its parent struct `s`. | ||
| #[tokio::test] | ||
| async fn test_nested_field_equality_delete_in_struct_layout() { |
There was a problem hiding this comment.
What do you think about adding the end-to-end scan test in this PR, marked #[ignore = "needs #2906"]? Then #2906 only has to remove the attribute, and the test gets reviewed next to the loader code it covers. The repo already uses a reasoned #[ignore] in avro/schema.rs. I'd cover both delete file layouts, a data row with a null s, a top-level f with the same value, and a null s in the delete file on the required s.g. The last one also needs #3306, so its ignore reason can name both PRs.
| fn nest_top_level_equality_columns(batch: RecordBatch, schema: &Schema) -> Result<RecordBatch> { | ||
| let top_level_ids: HashSet<i32> = schema | ||
| .as_struct() | ||
| .fields() | ||
| .iter() | ||
| .map(|field| field.id) | ||
| .collect(); | ||
| let mut flat = HashMap::new(); | ||
| for (index, field) in batch.schema().fields().iter().enumerate() { | ||
| if let Some(id) = try_get_field_id_from_metadata(field)? | ||
| && !top_level_ids.contains(&id) | ||
| && schema.field_by_id(id).is_some() | ||
| { | ||
| flat.insert(id, index); | ||
| } |
There was a problem hiding this comment.
Could we work out the nesting once per stream instead of once per batch? top_level_ids, flat, the output Schema, and the flat.values().any(...) scan per column depend only on the batch schema and pruned, and the batch schema is the same for every batch of a file. Computing the column mapping and the output schema from the first batch (or from the Parquet file's Arrow schema), then rebuilding each batch from the column indices, would drop the per-batch HashSet, HashMap, and Schema allocations.
Which issue does this PR close?
What changes are included in this PR?
Equality ids may name a field nested in a struct (the spec allows it and
EqualityDeleteWriterwrites it), but loading such a delete file failed withfield not found:f), which fails to bind or binds to a top-level column with the same name. It now uses the full name (s.f).test_delete_file_loader_parse_equality_deletesexpected the leaf namesafor a nested field, so its expected string changes tos.sa.Applying the delete to data rows also needs predicates on nested leaves in the row filter, which #2906 adds. With #2906 merged locally, an end-to-end scan deletes the expected rows (including rows whose parent struct is null). I'll add that end-to-end test once #2906 lands.
cc @brgr-s, as discussed in #2961.
Are these changes tested?
Yes.
test_nested_field_equality_delete_predicatecovers loading and the full-name predicate (the table also has a top-level column with the same leaf name).cargo test -p iceberg --lib, clippy and fmt are clean.AI Disclosure
I used Claude Code to find this, write the change and run the tests. I reviewed it before opening the PR.