perf(arrow): buffer positional-delete runs to drop per-row key allocation - #3015
perf(arrow): buffer positional-delete runs to drop per-row key allocation#3015anoopj wants to merge 2 commits into
Conversation
…tion parse_positional_deletes_record_batch_stream called result.entry(file_path.to_string()) for every deleted row, allocating and hashing a fresh path string per row even though a positional delete file repeats the same data-file path across all of its rows (the records are sorted by (file_path, pos)). Within each Arrow batch, buffer the contiguous run of positions for one path and merge it into the map in a single entry lookup, so the key is allocated and hashed once per run instead of once per row. Grouping is per batch, not across the whole stream (the run buffer is scoped to the batch), so a path spanning batches is flushed once per batch it appears in. Behavior is unchanged: positions are still inserted one-by-one (no ordering precondition), and a path that recurs merges into its existing delete vector, so results are identical regardless of ordering. Also drops an unused schema binding. Microbenchmark filling HashMap<String, RoaringTreemap> from 500k rows in 8192-row batches: one data file 32.9 -> 8.5 ns/row (3.9x); 50 data files 36.8 -> 10.0 ns/row (3.7x).
…ted cases Replace the single grouping test with two: one spec-compliant case (rows sorted by (file_path, pos), covering multi-position runs, several files in a batch, and a run spanning the batch boundary), and one deliberately spec-noncompliant unsorted case that proves the reader does not depend on the spec sort order. A path split into non-contiguous runs still merges into one delete vector rather than dropping positions. The non-compliance is called out in the test name and doc comment so the input is not mistaken for a realistic one.
3fb1647 to
721b42e
Compare
mbutrovich
left a comment
mbutrovich
left a comment
There was a problem hiding this comment.
First pass, thanks @anoopj!
The benchmark numbers in the description measure filling a bare HashMap<String, RoaringTreemap>, not parse_positional_deletes_record_batch_stream itself. The function's downcast, iteration, and validation overhead aren't part of that loop, so the ns/row numbers likely overstate the end-to-end win. Re-running the benchmark through the actual function would give a more representative number.
| let delete_vector = result.entry(file_path.to_string()).or_default(); | ||
| for &pos in positions { | ||
| delete_vector.insert(pos); | ||
| } |
There was a problem hiding this comment.
DeleteVector::insert_positions (delete_vector.rs:56) already bulk-appends a sorted slice into the underlying RoaringTreemap in one call, and it's currently unused (#[allow(dead_code)]). Since a run here is ascending in the common spec-compliant case, calling insert_positions first and falling back to this per-element loop only on Err would get the append-based speedup for the case your benchmark targets, while keeping the non-compliant-run handling your test at line 1007 covers.
| let mut run_path: Option<&str> = None; | ||
| let mut run_positions: Vec<u64> = Vec::new(); |
There was a problem hiding this comment.
run_positions is declared inside the while loop, so it drops and rebuilds from empty every batch, discarding the capacity it grew to from the previous batch's runs. run_path needs to stay batch-scoped since it borrows the batch's StringArray, but run_positions is owned and could be hoisted above the loop, with a .clear() added after the final flush at line 399 to match.
| file_path: &str, | ||
| positions: &[u64], | ||
| ) { | ||
| if positions.is_empty() { |
There was a problem hiding this comment.
This guard can't be hit: both call sites only run after run_path was set to Some, which always happens in the same iteration as a push onto run_positions. Since this is a caller-guaranteed invariant rather than a real precondition, debug_assert!(!positions.is_empty()) would document it instead of silently swallowing a case that can't occur.
| .or_default() | ||
| .insert(pos as u64); | ||
| if run_path != Some(file_path) { | ||
| if let Some(run_path) = run_path { |
There was a problem hiding this comment.
This binding shadows the outer run_path, which is reassigned two lines down at line 392, two different bindings sharing one name. Renaming this one to prev_path would make the reassignment read as updating the loop state rather than the just-matched value.
2 participants
Which issue does this PR close?
What changes are included in this PR?
parse_positional_deletes_record_batch_streamis allocating and hashing a fresh path string per row even though a positional delete file repeats the same data-file path across all of its rows (the records are sorted by (file_path, pos)).This PR fixes the inefficiency. Behavior is unchanged: positions are still inserted one-by-one, and a path that recurs merges into its existing delete vector, so results are identical regardless of ordering.
Microbenchmark filling HashMap<String, RoaringTreemap> from 500k rows in 8192-row batches: one data file 32.9 -> 8.5 ns/row (3.9x); 50 data files 36.8 -> 10.0 ns/row (3.7x).
Are these changes tested?
Added tests, existing test coverage is there also
AI Disclosure
Assisted with Claude code