Chained CollectLeft hash joins multiply StringView buffer references (TPC-DS Q64: 8.5 GB, ~10x slower with pushdown_filters) #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:

main (95bb0a0) 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:

   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 Arcs 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