UnnestExec ignores datafusion.execution.batch_size

[andygrove]

Describe the bug

UnnestExec emits exactly one output batch per input batch, however many rows the unnesting produces. datafusion.execution.batch_size is never consulted.

UnnestStream::poll_next_impl calls build_batch once per input batch and returns the whole result:

let result = build_batch(
    &batch,
    &self.schema,
    &self.list_type_columns,
    &self.struct_column_indices,
    &self.options,
)?;
...
Some(Ok(result_batch))

There is no batch_size anywhere in unnest.rs, so input_batches always equals output_batches in the operator's metrics.

Two consequences:

1. Downstream operators receive arbitrarily large batches. An 8192-row batch of 100-element lists comes back as a single 819,200-row batch.
2. Peak memory scales with input batch size times list length, rather than with batch_size. The entire unnested result for an input batch is materialized at once.

This is likely a contributing factor in #20788, though the discussion there centers on ordered array_agg buffering rather than on UnnestExec itself.

To Reproduce

-- 40 rows, each holding a 1000-element list
CREATE TABLE t AS
  SELECT i AS id, range(0, 1000) AS xs
  FROM (SELECT unnest(range(1, 41)) AS i);

SET datafusion.execution.batch_size = 8192;

EXPLAIN ANALYZE SELECT unnest(xs) FROM t;

UnnestExec reports one output batch per input batch, with output batches far larger than 8192 rows.

Expected behavior

Output batches should respect datafusion.execution.batch_size, and unnesting should not materialize the full expansion of an input batch at once.

Note that an exact size cannot be guaranteed in every case: a single input row whose list is longer than batch_size cannot be split across output batches without splitting within a row. batch_size should be an upper bound.

Additional context

find_longest_length already computes how many output rows each input row expands into, with all three NullHandling modes accounted for. Prefix-summing it gives exact chunk boundaries for depth-1 unnesting, which is enough to bound both output size and peak memory by consuming each input batch in chunks.

Recursive unnest (depth > 1) is not predictable that way, since a row's expansion depends on inner list lengths that only exist after the outer levels have been unnested. That case, and the oversized-single-row case, can be handled by slicing the built batch instead. Struct-only unnesting does not change the row count, so it is already bounded by the input batch size.