adriangb opened a new issue, #25712:
URL: https://github.com/apache/datafusion/issues/25712
### Describe the bug
A chain of `CollectLeft` hash joins with `StringView` payload columns can
make the number of data buffer references in each array grow at every join.
Each level multiplies the count by the number of build batches. The data itself
does not grow: the references point to the same few allocations.
The result is high memory use and a large CPU cost in the memory accounting
code. On TPC-DS SF1 Q64 we measured:
| | `pushdown_filters = false` | `pushdown_filters = true` |
|---|---|---|
| Query time | 0.75–1.4 s | 4.0–10.6 s |
| Peak RSS | about 1.2 GB | about 8.5 GB |
| Largest buffer count in one build-side column | 3,024 | 1,587,600 |
| Distinct allocations behind those buffers | 2 | 2 |
`main` (95bb0a0dfa) has the same behavior. The problem is not specific to
filter pushdown: pushdown only changes whether something copies (compacts) the
arrays between the joins (see below).
### To Reproduce
1. Use the TPC-DS SF1 Parquet data from before datafusion-benchmarks#31
(surrogate keys as `Float64`, `store_sales` as one row group of 2.88 M rows).
With this data the planner puts the whole `store_sales` join chain on the build
side of several `CollectLeft` joins.
2. Run Q64 (`datafusion/core/tests/tpc-ds/64.sql`) with `datafusion-cli`:
```sql
SET datafusion.execution.parquet.pushdown_filters = true;
-- run 64.sql a few times, compare with pushdown_filters = false
```
3. Observe the time and RSS difference. `EXPLAIN ANALYZE` shows it on the
hash joins, for example:
- `HashJoinExec CollectLeft (ss_item_sk, i_item_sk)`: `build_time=1.04s`
for `build_input_rows=38.96 K` (917 µs without pushdown)
- `HashJoinExec CollectLeft (c_current_addr_sk, ca_address_sk)`:
`build_time=252.77ms` (0.73 ms without pushdown)
With the current benchmark data (integer keys) the plan is different and
this does not happen.
A CPU profile shows `RecordBatchMemoryCounter::count_buffer_memory_size`
(from `get_record_batch_memory_size`, called by `BaselineMetrics::record_poll`
in `HashJoinStream::poll_next_impl` and the build-side stream) as the top frame
by far. It is followed by hash set inserts on buffer ids,
`Arc<Buffer>::drop_slow` and `concat_byte_view`.
### Cause
1. The build side of `CollectLeft` concatenates its batches
(`collect_left_input` / `try_create_array_map`). For view arrays, arrow's
concat (`GenericByteViewBuilder::append_array`) appends the data buffer list of
every input array and does not deduplicate buffers that are the same
allocation. If N build batches share the same K buffers, the result has N × K
buffer references.
2. The join output uses `take` on the build side, so each output array keeps
the full buffer list.
3. When the next `CollectLeft` join concatenates those output batches, the
count multiplies again. We measured 1 → 6 → 42 → 252 → 1,512 → 10,584 → 63,504
→ 317,520 → 1,587,600 references per column, all pointing to 2 allocations.
4. The memory metrics walk every reference on each poll, and the reference
vectors themselves use the memory.
Pushdown only decides if the chain is reset. The coalescer passes batches
larger than half the target size through without a copy. Without pushdown, join
outputs are small, the coalescer copies them, and the copy compacts the view
buffers. With pushdown the probe input is already filtered, join outputs are
large, and nothing compacts them. The growth factor per level is the number of
build batches, which depends on scheduling. This is why the query time is
bimodal.
### Possible fixes
1. **DataFusion (small):** after the build-side concat, remove duplicate
data buffers by pointer and rewrite the views to the deduplicated buffer
indices. This changes only the views, so no string data is copied. A local
prototype brought Q64 with `pushdown_filters = true` from 4.6–6.6 s to
0.75–0.85 s and peak RSS from 8.5 GB to 1.2 GB.
2. **arrow-rs:** make `concat` / `append_array` for view arrays reuse the
buffer index of a pointer-equal buffer, at least when the whole buffer list is
the same as the previous array's. This fixes all callers.
3. **DataFusion (mitigation only):** make
`count_byte_view_array_memory_size` cheaper for repeated buffers, or compact
view arrays in the join output when the buffer count is much larger than the
number of distinct buffers.
### Additional context
Related: #16206 (the cost of cloning the N buffer `Arc`s in `take` from a
concatenated build side). This issue is the case where the buffer count also
compounds across a chain of joins.
Found while investigating dynamic filter pushdown performance for #22883.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
