Repository navigation
fix: reserve the concatenated build batch and computed join keys in HashJoinExec - #25496
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #25496 +/- ##
========================================
Coverage 82.42% 82.42%
========================================
Files 1138 1138
Lines 435501 435691 +190
Branches 435501 435691 +190
========================================
+ Hits 358955 359127 +172
- Misses 54841 54847 +6
- Partials 21705 21717 +12 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
left a comment
There was a problem hiding this comment.
Thanks for working on this. The memory accounting changes look good overall. I have two non-blocking suggestions below, but neither needs to hold up the PR.
| &mut reservation, | ||
| &metrics, | ||
| )?; | ||
| let left_values = evaluate_expressions_to_arrays(&on_left, &batch)?; |
There was a problem hiding this comment.
One thing worth noting is that evaluate_expressions_to_arrays can materialize computed build keys before their retained buffers are counted and reserved below. That leaves a temporary gap between pool accounting and actual RSS.
I don't think this needs to block this PR. This PR addresses the unaccounted retained concat and computed-key buffers, and the RecordBatchMemoryCounter accounting below handles the latter once evaluation completes. Arbitrary physical expressions and UDFs also don't have a reliable generic allocation upper bound, and similar pre-evaluation temporary allocations already exist in update_hash and the ArrayMap null check.
I think reservation-backed expression evaluation would be better handled as a broader follow-up rather than expanding the scope here.
There was a problem hiding this comment.
Agreed, thanks for spelling it out. The keys are evaluated before they are counted, so there is a window where the pool is behind RSS. Since there is no generic upper bound for what an arbitrary expression or UDF allocates, and update_hash and the ArrayMap null check have the same gap today, I'd rather keep this PR to the retained buffers and leave reservation-backed expression evaluation as a follow-up.
|
|
||
| /// Only the views of a view array are copied, its data buffers stay shared. | ||
| #[test] | ||
| fn concat_build_batches_view_data_not_reserved_twice() -> Result<()> { |
There was a problem hiding this comment.
Could be nice to add BinaryView coverage alongside Utf8View here. estimate_concat_allocation handles both variants in the same match arm, so this would give us some symmetry in the test coverage. Definitely not something I'd block the PR on.
There was a problem hiding this comment.
Good point, done in e103926: concat_test_batches now has a nullable BinaryView column as well, so the other concat tests cover it too, and concat_build_batches_view_data_not_reserved_twice runs for both the Utf8View and the BinaryView column.
…ashJoinExec (apache#25496) ## Which issue does this PR close? - Closes apache#25269. ## Rationale for this change A hash join reserves every build-side batch as it arrives, then copies all of them into a single batch without reserving the copy, while the input batches stay alive until the build finishes. Near the memory limit the process therefore holds about twice the build side while the memory pool sees it once, so instead of a `ResourcesExhausted` error the process can be killed by the OS. It also means nothing that reacts to pool pressure can rely on what a hash join reports. Measured on a 1.25 GB build side (TPC-H SF1 `lineitem`, 16 columns, as the build side): the pool was charged ~1.25 GB while peak RSS was ~2.46 GB. ## What changes are included in this PR? - `concat_build_batches`: reserves the copy before `concat_batches`, drops the input batches, then trims the reservation to what the concatenated batch retains (`get_record_batch_memory_size`), growing instead if the estimate was low. `build_mem_used` follows every grow and shrink. - Nothing extra is reserved when there is nothing to copy: concatenating a single batch is zero-copy, and for `Utf8View`/`BinaryView` columns only the views are copied while the data buffers stay shared with the inputs. - `try_create_array_map` becomes `array_map_key_range`: it only decides whether the perfect hash join applies and reserves the `ArrayMap`. Both the hash map and the `ArrayMap` paths now hand the batches to the helper by value, so the inputs are gone before the visited bitmap and the null-aware scope maps are allocated. - Join key arrays that do not share the buffers of the build batch (any on-expression that is not a plain column) are kept for the whole join and are now reserved too. This uses a new public `RecordBatchMemoryCounter::count_array`. This does not reduce memory usage: the build still peaks at about twice the build side, it is now counted. A memory-limited query that only fit because the copy was not counted can now fail with `ResourcesExhausted`. ## What is the testing strategy for this PR? - `join_build_concat_is_reserved` (CollectLeft and Partitioned, hash map and `ArrayMap`): fails with `ResourcesExhausted` at map + 1.5x inputs, passes at map + 3x inputs. It fails on `main` in all four variants. - `join_build_key_arrays_are_reserved` (hash map and `ArrayMap`): with a single build batch and an `a1 + 1` join key, the join fails at inputs + map + half the key array and passes with room for it; a plain column key passes at the lower limit. It fails when the reservation is removed. - Unit tests of the helper: output equals `concat_batches` in both orders, the copy is reserved and the reservation then matches the retained batch, a single batch is not reserved twice, view data buffers are not reserved twice. - Existing `single_partition_join_overallocation` / `partitioned_join_overallocation` are unchanged. ## Are there any user-facing changes? One public addition: `RecordBatchMemoryCounter::count_array`. Hash joins running close to their memory limit may now report `ResourcesExhausted` where they previously exceeded the limit unnoticed.
…ashJoinExec (apache#25496) ## Which issue does this PR close? - Closes apache#25269. ## Rationale for this change A hash join reserves every build-side batch as it arrives, then copies all of them into a single batch without reserving the copy, while the input batches stay alive until the build finishes. Near the memory limit the process therefore holds about twice the build side while the memory pool sees it once, so instead of a `ResourcesExhausted` error the process can be killed by the OS. It also means nothing that reacts to pool pressure can rely on what a hash join reports. Measured on a 1.25 GB build side (TPC-H SF1 `lineitem`, 16 columns, as the build side): the pool was charged ~1.25 GB while peak RSS was ~2.46 GB. ## What changes are included in this PR? - `concat_build_batches`: reserves the copy before `concat_batches`, drops the input batches, then trims the reservation to what the concatenated batch retains (`get_record_batch_memory_size`), growing instead if the estimate was low. `build_mem_used` follows every grow and shrink. - Nothing extra is reserved when there is nothing to copy: concatenating a single batch is zero-copy, and for `Utf8View`/`BinaryView` columns only the views are copied while the data buffers stay shared with the inputs. - `try_create_array_map` becomes `array_map_key_range`: it only decides whether the perfect hash join applies and reserves the `ArrayMap`. Both the hash map and the `ArrayMap` paths now hand the batches to the helper by value, so the inputs are gone before the visited bitmap and the null-aware scope maps are allocated. - Join key arrays that do not share the buffers of the build batch (any on-expression that is not a plain column) are kept for the whole join and are now reserved too. This uses a new public `RecordBatchMemoryCounter::count_array`. This does not reduce memory usage: the build still peaks at about twice the build side, it is now counted. A memory-limited query that only fit because the copy was not counted can now fail with `ResourcesExhausted`. ## What is the testing strategy for this PR? - `join_build_concat_is_reserved` (CollectLeft and Partitioned, hash map and `ArrayMap`): fails with `ResourcesExhausted` at map + 1.5x inputs, passes at map + 3x inputs. It fails on `main` in all four variants. - `join_build_key_arrays_are_reserved` (hash map and `ArrayMap`): with a single build batch and an `a1 + 1` join key, the join fails at inputs + map + half the key array and passes with room for it; a plain column key passes at the lower limit. It fails when the reservation is removed. - Unit tests of the helper: output equals `concat_batches` in both orders, the copy is reserved and the reservation then matches the retained batch, a single batch is not reserved twice, view data buffers are not reserved twice. - Existing `single_partition_join_overallocation` / `partitioned_join_overallocation` are unchanged. ## Are there any user-facing changes? One public addition: `RecordBatchMemoryCounter::count_array`. Hash joins running close to their memory limit may now report `ResourcesExhausted` where they previously exceeded the limit unnoticed.
…uest A sort that spills merges its runs, in partitions that end their input early, with as many buffers as the pool grants, and a partition still sorting then finds no memory left (apache/datafusion#25804, finding 9): TPC-DS Q67 on GPORCA failed so 10 times in 10, and the window queries of WIN1 (the three largest per group by rank(), a running sum) did not finish. apache/datafusion#26264 fixes it. rust/Cargo.toml now takes DataFusion from a commit of that pull request's branch (MisterRaindrop/datafusion 5c59c075f, main with the fix) until a release has it. Main also has the spill pool fix (apache/datafusion#24891) that rust/vendor carried for 55.2.0: the vendored crate and the [patch.crates-io] go. Main's hash join reserves more for its build side than 55.2.0's did: the copy it concatenates the batches into, while it still holds them (apache/datafusion#25496). SpillingHashJoin collected a build side under its size and 40 bytes a row, which the join had used within; past it, the join reserved as memory that cannot spill, which a pool the partitions' build sides had filled did not grant: TPC-DS Q4 failed under both planners and two SHJ joins under work_mem 2 MB. A build side read to its end now also reserves what the join will take beyond: its table as DataFusion estimates it (on one integer key, the most an array over the keys' range may take, which the join builds where the keys are dense: a LEFT JOIN of SHJ failed for it one run in a dozen) and the copy. One that does not get it spills. Two tests, each failing without its part. An aggregate now reserves for its accumulators when it starts (apache/datafusion#24692), which a pool full to its limit does not grant: the test of a join that fits in a full pool counts its rows itself. Rust tests 10 runs of 10; regression 22/22; TPC-DS SF1 205/206 matching PostgreSQL (Q71's ties), 83.9 s against 133.8 s; Q67 on GPORCA 10/10; TPC-H SF1 and SF10 44/44, 9.4 s and 55.3 s against 31.2 s and 205.0 s; SHJ 15 runs and work_mem 1 MB without a diff; the window queries finish with PostgreSQL's rows, top 3 2.1 s against 9.0 s, running sum 3.0 s against 6.4 s.
Which issue does this PR close?
Rationale for this change
A hash join reserves every build-side batch as it arrives, then copies all of them
into a single batch without reserving the copy, while the input batches stay alive
until the build finishes. Near the memory limit the process therefore holds about
twice the build side while the memory pool sees it once, so instead of a
ResourcesExhaustederror the process can be killed by the OS. It also meansnothing that reacts to pool pressure can rely on what a hash join reports.
Measured on a 1.25 GB build side (TPC-H SF1
lineitem, 16 columns, as the buildside): the pool was charged ~1.25 GB while peak RSS was ~2.46 GB.
What changes are included in this PR?
concat_build_batches: reserves the copy beforeconcat_batches, drops the inputbatches, then trims the reservation to what the concatenated batch retains
(
get_record_batch_memory_size), growing instead if the estimate was low.build_mem_usedfollows every grow and shrink.batch is zero-copy, and for
Utf8View/BinaryViewcolumns only the views arecopied while the data buffers stay shared with the inputs.
try_create_array_mapbecomesarray_map_key_range: it only decides whether theperfect hash join applies and reserves the
ArrayMap. Both the hash map and theArrayMappaths now hand the batches to the helper by value, so the inputs aregone before the visited bitmap and the null-aware scope maps are allocated.
that is not a plain column) are kept for the whole join and are now reserved too.
This uses a new public
RecordBatchMemoryCounter::count_array.This does not reduce memory usage: the build still peaks at about twice the build
side, it is now counted. A memory-limited query that only fit because the copy was
not counted can now fail with
ResourcesExhausted.What is the testing strategy for this PR?
join_build_concat_is_reserved(CollectLeft and Partitioned, hash map andArrayMap): fails withResourcesExhaustedat map + 1.5x inputs, passes atmap + 3x inputs. It fails on
mainin all four variants.join_build_key_arrays_are_reserved(hash map andArrayMap): with a single buildbatch and an
a1 + 1join key, the join fails at inputs + map + half the key arrayand passes with room for it; a plain column key passes at the lower limit. It fails
when the reservation is removed.
concat_batchesin both orders, the copyis reserved and the reservation then matches the retained batch, a single batch
is not reserved twice, view data buffers are not reserved twice.
single_partition_join_overallocation/partitioned_join_overallocationare unchanged.
Are there any user-facing changes?
One public addition:
RecordBatchMemoryCounter::count_array. Hash joins running closeto their memory limit may now report
ResourcesExhaustedwhere they previously exceeded the limit unnoticed.