adapters: handle deletion vector in delta input connector#6391
Conversation
0576337 to
4cc1f9a
Compare
|
So is it too early to review this? |
not ready for review, I was hoping for @mythical-fred to review it first hence I opened PR instead of keeping it Draft. I will request your review when i'm done. |
0a73bd5 to
f6b0f46
Compare
mythical-fred
left a comment
There was a problem hiding this comment.
LGTM — well-structured change. The masked Parquet reader, same-path pair skipping in CDC, and the active_dv guard are all clean. Tests are thorough: proptest for the bitmap→selection conversion, the end-to-end masked reader test with schema evolution (cast + null-fill), and the five Python e2e tests covering snapshot, follow, follow-restore, CDC, and CDC-overwrite are each shaped to isolate a distinct failure mode. Nice work.
mythical-fred
left a comment
There was a problem hiding this comment.
Re-review of af4b26fe: the CDC refactor is clean. process_cdc_transaction + cdc_side_dataframe + masked_file_provider is easier to follow than the old register/deregister dance, and the same-path pair skipping matches the documented "metadata-only rewrite" intent. The plain and DV-masked sides both filter the same snapshot schema through keeps_cdc_column, so column order lines up for the UNION ALL. The count_snapshot_transaction_start move is well-placed and the new delta_table_snapshot_mode_snapshot_transaction_test pins exactly the regression you described. Behavior around "removes with no surviving adds" is preserved (old EXCEPT ALL clamped them to 0 multiplicity; new early-return drops them) — no semantic change. LGTM.
mihaibudiu
left a comment
There was a problem hiding this comment.
@swanandx I left some questions before reading the whole PR, if you can answer them I can proceed with a more complete review
mythical-fred
left a comment
There was a problem hiding this comment.
Re-reviewed the new commits (4584d9a + the cosmetic ci fixup) on top of my earlier approval at 0788fc5. The follow/CDC DV story is solid:
- Same-path Add/Remove pairing as a metadata-only rewrite is the right call: Delta files are immutable and content-addressed by path, so a same-path pair can only be a DV rewrite (soft delete or restore). Catching it before the EXCEPT ALL is what stops a RESTORE from re-emitting events.
- The plain and masked CDC sides both project through
keeps_cdc_columnagainst the same snapshot-derived field set in field order, so the UNION ALL lines up positionally. Good. bitmap_to_row_selectionhandles all the edge cases (leading/trailing/contiguous deletes, positions past EOF) and the proptest pins the complement invariant.- DV decode on
spawn_blocking+ idempotent retry is fine — the decode is a pure read of inline data or a sidecar. - Partition-column-as-NULL is the same pre-existing limitation as the ListingTable path; calling it out in the doc comment is the right move.
- Tests cover follow (v0→v1), follow-restore (a
removewith a non-empty DV — the variant the v0→v1 case can't reach), CDC RESTORE round-trip, and CDC INSERT OVERWRITE — that last one is the only commit shape that drivesremoves_masked, and the fixture is constructed so an unmaskedremovewould visibly fail. Nice trap. - Docs trimmed correctly (DVs out of the unsupported list, new note in the connector page).
LGTM.
ryzhyk
left a comment
There was a problem hiding this comment.
Looks good overall (to the extent that I am able to validate it). See a couple of questions inline. This will also need QA workloads + customer beta testing.
cc @anandbraman .
| // unpaired removes there. | ||
| let mut add_files: Vec<(&str, Option<&DeletionVectorDescriptor>)> = Vec::new(); | ||
| for add in &adds { | ||
| if removes_by_path.remove(add.path.as_str()).is_some() { |
There was a problem hiding this comment.
I don't understand why this is correct. Don't we need to compute the diff between deletion vectors in the Remove and Add actions.
There was a problem hiding this comment.
we don't care about deletion events, so we don't need to compare deletion vectors ( they would change for deleted records, which we ignore anyways by EXCEPT ALL ) so removing by path is fine.
| @@ -0,0 +1,519 @@ | |||
| //! Deletion-vector (DV) support for the Delta Lake input connector. | |||
There was a problem hiding this comment.
This module should ideally be part of delta-rs. @swanandx , do you know if they are working on or plan to work on something like this? Should we contribute it? We don't need to wait for it to merge this PR of course, but it may be helpful if this code is maintained as part of delta-rs.
There was a problem hiding this comment.
do you know if they are working on or plan to work on something like this?
i do not know
Should we contribute it?
need to evaluate, DeletionVectorDescriptor::row_indexes / to_kernel_descriptor can be contributed. I don't know if delta-rs would want a file-level reader (give it some files + DVs, get back a reader). Adding it there would mean a new API
mythical-fred
left a comment
There was a problem hiding this comment.
Re-APPROVE on 0f2cc73d99.
Diff vs prior approved tip 622d9cb1de is purely two new Python tests in test_delta_input_deletion_vectors.py; all Rust files (deletion_vector.rs, input.rs, delta_table.rs) and fixtures/deletion_vectors.py are byte-identical. The new tests pin down the two DV paths the prior suite didn't reach:
test_delta_input_follow_restore_with_deletion_vectors: drives the masked reader from aremoveaction whose DV is non-empty (the v0->v1 follow test'sremovehad no DV). Confirms the delete's DV is applied to theremoveso only the survivors get retracted, and the (unmasked)addrestores the whole file.test_delta_input_cdc_overwrite_masks_removed_dv_rows: explicitly exercises theremoves_maskedpath with anINSERT OVERWRITEthat re-inserts the soft-deleted rows. The only commit shape where theremove's DV must be applied; without masking the over-subtraction would zero out the new events.
The dual assertions (total plus even_id_rows) are a nice guard against inverted DVs that COUNT-only checks miss. _ingest_dv_fixture helper cleanly factors the lifecycle. No findings.
snapshot mode already drops soft-deleted rows because delta-rs applies the deletion vectors for us. follow and cdc mode don't: they read each log action's Parquet file directly, so they were ingesting deleted rows. Add a small masked-Parquet reader (deletion_vector.rs): decode a file's deletion vector with delta_kernel and skip those rows during decode. The follow and cdc paths now read DV'd files through it, and cdc also skips metadata-only same-path add/remove pairs (DV rewrites, restores). Covered by Rust unit tests for the reader and Spark-seeded Python e2e tests across all three modes. Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com> Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: Swanand Mulay <73115739+swanandx@users.noreply.github.com>
snapshot mode works fine with deletion vectors, but for follow mode we need parse and apply deletion vector ourselves to read parquet files, same with some cases in cdc mode
Describe Manual Test Plan
Checklist
Breaking Changes?
Not breaking change