kosiew opened a new pull request, #24364: URL: https://github.com/apache/datafusion/pull/24364
## Which issue does this PR close? * Part of #23393 ## Rationale for this change The symmetric hash join can retain an output `RecordBatch` in its batch transformer without including that batch in the stream's memory reservation. As a result, the stream can under-account memory while either `NoopBatchTransformer` or `BatchSplitter` holds a batch. The retained batch may also share Arrow buffers or nested arrays with other stream state, so simply adding individual batch memory sizes can double-count shared allocations. This PR accounts for transformer-retained batches at the symmetric hash join stream reservation boundary while deduplicating shared Arrow allocations. Corrected accounting can increase reservation pressure under bounded memory pools, without changing join results. ## What changes are included in this PR? * Extend `RecordBatchMemoryCounter` with sequence-aware accounting for both Arrow buffers and array-object overhead, including shared nested arrays. * Add `BatchTransformer::count_memory` so `NoopBatchTransformer` and `BatchSplitter` report memory retained by their current batch through the shared counter. * Include the left and right input buffers and transformer-retained batch in a single `RecordBatchMemoryCounter` when calculating `SymmetricHashJoinStream::size()`, avoiding duplicate accounting of shared allocations. * Add `SymmetricHashJoinStream::update_reservation()` and update the reservation when a transformer acquires or releases a batch. * Leave the unrelated `OneSideHashJoiner` accounting formulas unchanged. ## Are these changes tested? Yes. The patch adds the following regression tests: * `test_record_batch_memory_counter_array_overhead_shared_across_batches` * `test_record_batch_memory_counter_deduplicates_shared_nested_array_overhead` * `test_record_batch_memory_counter_deduplicates_recursive_shared_children` * `batch_transformers_count_retained_batch_memory` * `stream_accounts_for_transformer_batches_once` * `stream_deduplicates_nested_transformer_batches` * `symmetric_hash_join_reserves_transformer_batch` These cover transformer-retained batches for both `NoopBatchTransformer` and `BatchSplitter`, shared buffers and nested arrays, reservation/metric updates when transformer ownership changes, and bounded-memory behavior when the previously unaccounted retained batch pushes the join over the configured memory limit. ## Are there any user-facing changes? There are no changes to join result semantics or public APIs. Memory accounting for symmetric hash joins is more complete. Workloads running near a configured memory limit may therefore encounter memory-pressure behavior earlier because transformer-retained batches are now included in the stream reservation. ## LLM-generated code disclosure This PR includes LLM-generated code and comments. All LLM-generated content has been manually reviewed. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
