Ordered ARRAY_AGG accumulator retains one Arrow array per update_batch call, inflating per-row memory for grouped aggregation #25199

Is your feature request related to a problem or challenge?

Follow-up from #24392, which rewrote OrderSensitiveArrayAggAccumulator to retain payloads
as Arrow arrays instead of ScalarValues. That is a large win when update_batch is called
with reasonably sized batches, but the accumulator pushes one ArrayRef per call, so the
fixed per-array cost (ArrayData + a 64-byte-minimum buffer + the Arc) is paid per call
rather than per row:

https://github.com/apache/datafusion/blob/main/datafusion/functions-aggregate/src/array_agg.rs#L1488

Ordered ARRAY_AGG cannot use a GroupsAccumulator (groups_accumulator_supported requires
order_bys.is_empty()), so grouped queries go through GroupsAccumulatorAdapter, which calls
update_batch once per group per input batch. For a high-cardinality GROUP BY that is
typically 1–2 rows per call, which is the worst case for this layout.

Measured with the accumulator's own size(), 2048 rows, Int64 payload + Int64 ordering key:

So GROUP BY high_cardinality_col with array_agg(x ORDER BY y) — roughly the shape of #20788 —
uses about 1.4x more memory per row than it did before, while everything else got 3–4x better.

Describe the solution you'd like

Coalesce small inputs instead of retaining each as its own batch: below a row threshold,
concat into the tail batch and rewrite the affected entries rather than pushing a new
ArrayRef. That keeps the large-batch win and removes the per-call fixed cost.

Suggested fix: coalesce small inputs rather than retaining each as its own batch, e.g.

const COALESCE_ROWS: usize = 64;

// in store_batch, after compaction:
let row_count = values.len();
// ...
let start = self.entries.len();
let batch_idx = match self.batches.last() {
    Some(last) if last.len() + row_count <= COALESCE_ROWS => {
        let merged = arrow::compute::concat(&[last.as_ref(), values.as_ref()])?;
        let idx = self.batches.len() - 1;
        // rows already recorded for this batch keep their row_idx; the new rows
        // start at the old length
        let offset = self.batches[idx].len();
        self.batches[idx] = merged;
        self.entries.extend(
            (0..row_count).map(|row_idx| OrderedArrayAggEntry {
                batch_idx: idx,
                row_idx: offset + row_idx,
            }),
        );
        return Ok(Some(start..self.entries.len()));
    }
    _ => {
        self.batches.push(values);
        self.batches.len() - 1
    }
};

Describe alternatives you've considered

• A real GroupsAccumulator for ordered ARRAY_AGG, which would side-step
  GroupsAccumulatorAdapter's per-group slicing entirely. Bigger change, probably the
  right long-term answer.
• Leaving it as is — the regression is confined to very small update_batch calls.

Additional context

benches/array_agg.rs currently only exercises the unordered ArrayAggAccumulator::merge_batch,
so neither the win nor this regression is visible in CI benchmarks. A bench over the ordered path
covering both regimes (few rows per call vs. full batches, pre-ordered vs. random input) would be
worth adding alongside any fix.