MinMaxStructState::emit_to(EmitTo::First(n)) subtracts row counts from a byte counter #26891

Describe the bug

Filed as a follow up to review comments in apache/datafusion-comet#7191 (comment).

In datafusion-functions-aggregate/src/min_max/min_max_struct.rs, set_value adds get_array_memory_size() bytes to total_data_bytes for each stored struct. But emit_to(EmitTo::First(n)) subtracts s.len() for each emitted value, which is 1 for a one-row StructArray. So size() keeps counting memory for groups that were already emitted. It only resets on EmitTo::All.

Ordered aggregation emits First(n) after every batch, so the reported size grows with the number of groups seen. Reproduction: a Partial AggregateExec feeding a Final one over sorted input, with 4,096 groups of 4 KiB struct values, 64-row batches, a 2 MiB GreedyMemoryPool and the disk manager disabled. It fails with ResourcesExhausted in OrderedFinalAggregateStream. Subtracting get_array_memory_size() of the emitted values instead makes it complete.

The code looks copied from min_max_bytes, where len() is a byte count.

To Reproduce

No response

Expected behavior

No response

Additional context

No response

Willingness to contribute

None