Avoid single-chunk grouped accumulator states - #9597

Rebases #8379 onto develop and addresses the outstanding review
comments.

The grouped accumulator now takes caller-assigned dense `GroupIds`
(`{ids: ArrayRef, num_groups}`) directly, instead of requiring values to
be pre-sorted into complete list groups. Ids may repeat, arrive out of
order, or be absent from a batch.

## Changes

- Dispatch grouped kernels on both the values encoding and the group-id
  encoding, with wildcard fallbacks in each dimension.
- Cache materialized and validated group ids across clones, so several
  aggregates over the same ids validate once.
- Let each aggregate own its dense grouped state (`GroupedState`), with
  the generic one-partial-per-group implementation as the fallback.
- Store grouped count state as one `u64` per group and grouped sum state
  in monomorphic `u64`/`i64`/`f64`/decimal vectors, choosing the decimal
  width once from the sum dtype.
- Port `SumV2` onto the same dense sums, adding the per-group empty flag
  its SQL semantics need, so both sums share one accumulation kernel.
- Merge count and sum partial arrays without `execute_scalar` per row.
- Adapt between typed scatter for unclustered ids and per-run reduction
  for clustered ids.
- Use one stable counting-sort gather per batch in the universal
  fallback instead of one `take` per group.
- Keep `GroupedArray::dense_input` as the list-group adapter for
  `list_sum`, which restores list semantics (null, empty, and all-null
  lists sum to null) on top of the dense result.

## Rebase notes

`develop` reverted SQL null-on-empty semantics for `Sum` (#9324) and
re-landed them as a separate `SumV2` (#9590), so this branch follows
both contracts: `Sum` starts a group at zero and treats a null partial
as overflow, while `SumV2` keeps `{sum, is_overflow, is_empty}` and
finalizes an untouched group to null. `SumAggregateOpts` is gone, and
both aggregates take `NumericalAggregateOpts`.

`SumV2`'s list-shaped `PrimitiveGroupedSumV2EncodingKernel` is replaced
by a dense kernel that reuses `Sum`'s accumulation and then clears the
empty flag of every group that saw a valid value, so `sum_v2` keeps a
fast grouped path rather than falling back to per-group accumulators.

The single-partial `flush` shortcut from #9597 no longer applies: dense
state produces one partial array per flush, so there is no chunk to
collapse, and its three tests went with it.

## Review comments

- Group ids are an `ArrayRef` so kernels can dispatch on their encoding
  (Constant, RLE, ...), and `GroupIds` is the "Group struct" the
  accumulator API talks in.
- The hand-rolled `for_each_valid_idx` helpers are gone: the kernels
  take the mask's bit buffer, handle the all-true and all-false cases
  directly, and let `BitBuffer::for_each_set_index` walk the rest
  instead of materializing the mask's index vector.
- The run-detection helpers shared by the count and sum kernels moved to
  `aggregate_fn::kernels` instead of being duplicated per aggregate.
- The accumulator tracks `num_groups` itself, so `flush_partials` and
  `finish` no longer take it; `ensure_groups` is exposed for callers
  that need trailing groups no batch mentioned.
- `merge_group` names its parameters `into_group` / `from_group`.
- The benchmark input builder is now `DenseGroupedInput::clustered`.

## list_sum performance

`list_sum` is the one caller that has to build dense ids for input that
is already grouped, so it pays for the new API. The adapter avoids the
two costs that dominated: it reuses the element array as the values
array whenever the groups already cover it in order (null lists keep
their own ids, and their meaningless sums are masked out afterwards),
and it shares one set of group ranges and validity between the
adaptation and the mask.

What remains is the id array itself, roughly half a nanosecond per
element. Medians from `cargo bench -p vortex-array --bench list_sum`,
develop (d0a59fe) vs this branch on the same machine:

| benchmark | develop | this branch |
|---|---:|---:|
| `fsl_sum_large` | 5.1 us | 7.3 us |
| `fsl_sum_medium` | 3.4 us | 4.8 us |
| `fsl_sum_small` | 2.4 us | 3.2 us |
| `list_sum_large` | 11.2 us | 13.7 us |
| `list_sum_medium` | 8.9 us | 9.9 us |
| `list_sum_small` | 7.6 us | 7.5 us |
| `list_sum_nullable_elements_large` | 57.6 us | 18.1 us |
| `list_sum_nullable_elements_medium` | 31.2 us | 12.3 us |
| `listview_sum_large` | 6.9 us | 9.6 us |
| `listview_sum_medium` | 4.9 us | 6.0 us |
| `listview_sum_small` | 3.8 us | 3.9 us |

Lists whose elements are all valid are 20-40% slower; lists with
nullable elements are 2.5-3x faster, since the old path sliced the
element mask per group. Closing the remaining gap wants a grouped
kernel registered for run-encoded group ids, which the new two-encoding
kernel lookup already supports - a follow-up rather than part of this
change.

## Validation

- `cargo +nightly fmt --all`
- `cargo build -p vortex-array --all-targets`
- `cargo test -p vortex-array --lib` (3,509 passed, 1 ignored)
- `cargo test --doc -p vortex-array`
- `cargo bench -p vortex-array --bench list_sum`
- `cargo bench -p vortex-array --bench aggregate_grouped -- --test`
- `cargo clippy -p vortex-array --all-targets --all-features`
- `RUSTDOCFLAGS="-D warnings" cargo doc --no-deps -p vortex-array`

A workspace-wide build ran out of disk in this environment; the change
is confined to `vortex-array`, and no other crate references the grouped
accumulator API.

Signed-off-by: Nicholas Gates <[email protected]>
Signed-off-by: Robert Kruszewski <[email protected]>
Co-authored-by: Claude <[email protected]>
Claude-Session: https://claude.ai/code/session_01B8VcUXWmDmNH2JjJY4aDQ8

Rebases #8379 onto develop and addresses the outstanding review
comments.

The grouped accumulator now takes caller-assigned dense `GroupIds`
(`{ids: ArrayRef, num_groups}`) directly, instead of requiring values to
be pre-sorted into complete list groups. Ids may repeat, arrive out of
order, or be absent from a batch.

## Changes

- Dispatch grouped kernels on both the values encoding and the group-id
  encoding, with wildcard fallbacks in each dimension.
- Cache materialized and validated group ids across clones, so several
  aggregates over the same ids validate once.
- Let each aggregate own its dense grouped state (`GroupedState`), with
  the generic one-partial-per-group implementation as the fallback.
- Store grouped count state as one `u64` per group and grouped sum state
  in monomorphic `u64`/`i64`/`f64`/decimal vectors, choosing the decimal
  width once from the sum dtype.
- Port `SumV2` onto the same dense sums, adding the per-group empty flag
  its SQL semantics need, so both sums share one accumulation kernel.
- Merge count and sum partial arrays without `execute_scalar` per row.
- Adapt between typed scatter for unclustered ids and per-run reduction
  for clustered ids.
- Use one stable counting-sort gather per batch in the universal
  fallback instead of one `take` per group.
- Keep `GroupedArray::dense_input` as the list-group adapter for
  `list_sum`, which restores list semantics (null, empty, and all-null
  lists sum to null) on top of the dense result.

## Rebase notes

`develop` reverted SQL null-on-empty semantics for `Sum` (#9324) and
re-landed them as a separate `SumV2` (#9590), so this branch follows
both contracts: `Sum` starts a group at zero and treats a null partial
as overflow, while `SumV2` keeps `{sum, is_overflow, is_empty}` and
finalizes an untouched group to null. `SumAggregateOpts` is gone, and
both aggregates take `NumericalAggregateOpts`.

`SumV2`'s list-shaped `PrimitiveGroupedSumV2EncodingKernel` is replaced
by a dense kernel that reuses `Sum`'s accumulation and then clears the
empty flag of every group that saw a valid value, so `sum_v2` keeps a
fast grouped path rather than falling back to per-group accumulators.

`list_sum` now sums with `SumV2` (#9621), so its empty and all-null list
masking is gone; it only has to restore the one case dense ids cannot
express, a null list.

The single-partial `flush` shortcut from #9597 no longer applies: dense
state produces one partial array per flush, so there is no chunk to
collapse, and its three tests went with it.

## Review comments

- Group ids are an `ArrayRef` so kernels can dispatch on their encoding
  (Constant, RLE, ...), and `GroupIds` is the "Group struct" the
  accumulator API talks in.
- The hand-rolled `for_each_valid_idx` helpers are gone: the kernels
  take the mask's bit buffer, handle the all-true and all-false cases
  directly, and let `BitBuffer::for_each_set_index` walk the rest
  instead of materializing the mask's index vector.
- The run-detection helpers shared by the count and sum kernels moved to
  `aggregate_fn::kernels` instead of being duplicated per aggregate.
- The accumulator tracks `num_groups` itself, so `flush_partials` and
  `finish` no longer take it; `ensure_groups` is exposed for callers
  that need trailing groups no batch mentioned.
- `merge_group` names its parameters `into_group` / `from_group`.
- The benchmark input builder is now `DenseGroupedInput::clustered`.

## list_sum performance

`list_sum` is the one caller that has to build dense ids for input that
is already grouped, so it pays for the new API: one `u32` per element,
read once to sum and once to clear empty flags, plus a mask that nulls
the lists dense ids cannot represent. The adapter already avoids the
larger costs - it reuses the element array as the values array whenever
the groups cover it in order, and shares one set of group ranges and
validity with the mask - but the id array itself remains.

Fastest-of-100 from `cargo bench -p vortex-array --bench list_sum`,
develop (1fe8dda) vs this branch on the same machine. Medians are noisy
on this host, so these are the fastest samples:

| benchmark | develop | this branch |
|---|---:|---:|
| `fsl_sum_large` | 7.2 us | 11.1 us |
| `fsl_sum_medium` | 5.8 us | 8.2 us |
| `fsl_sum_small` | 4.8 us | 6.6 us |
| `list_sum_large` | 11.5 us | 18.5 us |
| `list_sum_medium` | 9.9 us | 13.8 us |
| `list_sum_small` | 9.0 us | 10.8 us |
| `list_sum_nullable_elements_large` | 21.9 us | 20.9 us |
| `list_sum_nullable_elements_medium` | 14.9 us | 15.3 us |
| `listview_sum_large` | 7.3 us | 14.1 us |
| `listview_sum_medium` | 6.1 us | 9.7 us |
| `listview_sum_small` | 5.2 us | 7.2 us |

Lists that are already grouped are 20-90% slower; only the
nullable-element cases come out level. `SumV2`'s list kernel reduces one
range at a time and never touches a per-element id, which is what dense
ids cost here.

Closing this wants a grouped kernel registered for run-encoded group
ids, so a list adapter can hand over `(offsets, group ids)` without
materializing one id per element and the kernel can reduce a run at a
time. The two-encoding kernel lookup added here already supports
registering it; the kernel itself is follow-up work, not part of this
rebase.

## Validation

- `cargo +nightly fmt --all`
- `cargo build -p vortex-array --all-targets`
- `cargo test -p vortex-array --lib` (3,511 passed, 1 ignored)
- `cargo test --doc -p vortex-array`
- `cargo bench -p vortex-array --bench list_sum`
- `cargo bench -p vortex-array --bench aggregate_grouped -- --test`
- `cargo clippy -p vortex-array --all-targets --all-features`
- `RUSTDOCFLAGS="-D warnings" cargo doc --no-deps -p vortex-array`

A workspace-wide build ran out of disk in this environment; the change
is confined to `vortex-array`, and no other crate references the grouped
accumulator API.

Signed-off-by: Nicholas Gates <[email protected]>
Signed-off-by: Robert Kruszewski <[email protected]>
Co-authored-by: Claude <[email protected]>
Claude-Session: https://claude.ai/code/session_01B8VcUXWmDmNH2JjJY4aDQ8

Rebases #8379 onto develop and addresses the outstanding review
comments.

The grouped accumulator now takes caller-assigned dense `GroupIds`
(`{ids: ArrayRef, num_groups}`) directly, instead of requiring values to
be pre-sorted into complete list groups. Ids may repeat, arrive out of
order, or be absent from a batch.

## Changes

- Dispatch grouped kernels on both the values encoding and the group-id
  encoding, with wildcard fallbacks in each dimension.
- Carry run-encoded group ids as `PiecewiseSequence` runs, and register
  `Count`, `Sum`, and `SumV2` kernels for that encoding so already-grouped
  input reduces a run per group instead of reading an id per row.
- Cache materialized and validated group ids across clones, so several
  aggregates over the same ids validate once.
- Let each aggregate own its dense grouped state (`GroupedState`), with
  the generic one-partial-per-group implementation as the fallback.
- Store grouped count state as one `u64` per group and grouped sum state
  in monomorphic `u64`/`i64`/`f64`/decimal vectors, choosing the decimal
  width once from the sum dtype.
- Port `SumV2` onto the same dense sums, adding the per-group empty flag
  its SQL semantics need, so both sums share one accumulation kernel.
- Merge count and sum partial arrays without `execute_scalar` per row.
- Adapt between typed scatter for unclustered ids and per-run reduction
  for clustered ids.
- Use one stable counting-sort gather per batch in the universal
  fallback instead of one `take` per group.
- Keep `GroupedArray::dense_input` as the list-group adapter for
  `list_sum`, which restores list semantics (null, empty, and all-null
  lists sum to null) on top of the dense result.

## Rebase notes

`develop` reverted SQL null-on-empty semantics for `Sum` (#9324) and
re-landed them as a separate `SumV2` (#9590), so this branch follows
both contracts: `Sum` starts a group at zero and treats a null partial
as overflow, while `SumV2` keeps `{sum, is_overflow, is_empty}` and
finalizes an untouched group to null. `SumAggregateOpts` is gone, and
both aggregates take `NumericalAggregateOpts`.

`SumV2`'s list-shaped `PrimitiveGroupedSumV2EncodingKernel` is replaced
by a dense kernel that reuses `Sum`'s accumulation and then clears the
empty flag of every group that saw a valid value, so `sum_v2` keeps a
fast grouped path rather than falling back to per-group accumulators.

`list_sum` now sums with `SumV2` (#9621), so its empty and all-null list
masking is gone; it only has to restore the one case dense ids cannot
express, a null list.

The single-partial `flush` shortcut from #9597 no longer applies: dense
state produces one partial array per flush, so there is no chunk to
collapse, and its three tests went with it.

## Review comments

- Group ids are an `ArrayRef` so kernels can dispatch on their encoding
  (Constant, RLE, ...), and `GroupIds` is the "Group struct" the
  accumulator API talks in.
- The hand-rolled `for_each_valid_idx` helpers are gone: the kernels
  take the mask's bit buffer, handle the all-true and all-false cases
  directly, and let `BitBuffer::for_each_set_index` walk the rest
  instead of materializing the mask's index vector.
- The run-detection helpers shared by the count and sum kernels moved to
  `aggregate_fn::kernels` instead of being duplicated per aggregate.
- The accumulator tracks `num_groups` itself, so `flush_partials` and
  `finish` no longer take it; `ensure_groups` is exposed for callers
  that need trailing groups no batch mentioned.
- `merge_group` names its parameters `into_group` / `from_group`.
- The benchmark input builder is now `DenseGroupedInput::clustered`.

## Run-encoded group ids

An input that is already grouped - a list array adapted by
`GroupedArray::dense_input` - would otherwise pay a `u32` per element
just to say "these rows belong to that group". When the groups cover the
element array in order, the adapter now hands over
`GroupIds::from_runs`: the ids are `PiecewiseSequence` runs (each group
id repeated for its run length, via a zero multiplier), and `Count`,
`Sum`, and `SumV2` each register a kernel for that group-id encoding
which reduces one run at a time. `GroupIds::runs` caches the decoded
runs the same way `validated_ids` caches materialized ids, so several
aggregates over the same ids decode once.

Each run kernel falls back to per-row accumulation when the ids turn out
not to be runs, so registering for the encoding never loses the generic
path. Tests assert run ids and materialized ids agree for every aggregate
across empty runs, all-null groups, overflow, and NaN options.

## list_sum performance

`list_sum` is the caller that pays for the dense-ids API on input that is
already grouped. Best-of-three alternating runs of the two benchmark
binaries, develop (1fe8dda) vs this branch on the same machine:

| benchmark | develop | this branch |
|---|---:|---:|
| `fsl_sum_large` | 7.0 us | 10.9 us |
| `fsl_sum_medium` | 5.8 us | 8.5 us |
| `fsl_sum_small` | 4.8 us | 6.6 us |
| `list_sum_large` | 11.4 us | 15.6 us |
| `list_sum_medium` | 9.6 us | 12.7 us |
| `list_sum_small` | 8.5 us | 10.8 us |
| `list_sum_nullable_elements_large` | 21.8 us | 27.8 us |
| `list_sum_nullable_elements_medium` | 14.2 us | 19.0 us |
| `listview_sum_large` | 7.2 us | 11.4 us |
| `listview_sum_medium` | 5.8 us | 8.7 us |
| `listview_sum_small` | 5.0 us | 7.3 us |

Still 27-59% slower. The run kernels removed the per-element id work,
and what is left scales with groups rather than elements: these
benchmarks run 100-500 groups over 800-4000 elements, where the dense
pipeline's fixed costs - building the run ids, dense state, flushing it
to arrays, and the extra mask that restores a null list - are the whole
measurement. `SumV2`'s list kernel instead writes its three output
buffers in one pass over the ranges.

Whether `list_sum` should keep a list-shaped kernel for that case, or
accept the overhead in exchange for one grouped API, is a call for this
PR's reviewers. The dense path is the one that wins at engine scale: the
`aggregate_grouped` benchmarks, which run 65k rows, are what this API
exists for.

## Validation

- `cargo +nightly fmt --all`
- `cargo build -p vortex-array --all-targets`
- `cargo test -p vortex-array --lib` (3,515 passed, 1 ignored)
- `cargo test --doc -p vortex-array`
- `cargo bench -p vortex-array --bench list_sum`
- `cargo bench -p vortex-array --bench aggregate_grouped -- --test`
- `cargo clippy -p vortex-array --all-targets --all-features`
- `RUSTDOCFLAGS="-D warnings" cargo doc --no-deps -p vortex-array`

A workspace-wide build ran out of disk in this environment; the change
is confined to `vortex-array`, and no other crate references the grouped
accumulator API.

Signed-off-by: Nicholas Gates <[email protected]>
Signed-off-by: Robert Kruszewski <[email protected]>
Co-authored-by: Claude <[email protected]>
Claude-Session: https://claude.ai/code/session_01B8VcUXWmDmNH2JjJY4aDQ8

Rebases #8379 onto develop and addresses the outstanding review
comments.

The grouped accumulator now takes caller-assigned dense `GroupIds`
(`{ids: ArrayRef, num_groups}`) directly, instead of requiring values to
be pre-sorted into complete list groups. Ids may repeat, arrive out of
order, or be absent from a batch.

## Changes

- Dispatch grouped kernels on both the values encoding and the group-id
  encoding, with wildcard fallbacks in each dimension.
- Carry run-encoded group ids as `PiecewiseSequence` runs, and register
  `Count`, `Sum`, and `SumV2` kernels for that encoding so already-grouped
  input reduces a run per group instead of reading an id per row.
- Cache materialized and validated group ids across clones, so several
  aggregates over the same ids validate once.
- Let each aggregate own its dense grouped state (`GroupedState`), with
  the generic one-partial-per-group implementation as the fallback.
- Store grouped count state as one `u64` per group and grouped sum state
  in monomorphic `u64`/`i64`/`f64`/decimal vectors, choosing the decimal
  width once from the sum dtype.
- Port `SumV2` onto the same dense sums, adding the per-group empty flag
  its SQL semantics need, so both sums share one accumulation kernel.
- Merge count and sum partial arrays without `execute_scalar` per row.
- Adapt between typed scatter for unclustered ids and per-run reduction
  for clustered ids.
- Use one stable counting-sort gather per batch in the universal
  fallback instead of one `take` per group.
- Keep `GroupedArray::dense_input` as the list-group adapter for
  `list_sum`, which restores list semantics (null, empty, and all-null
  lists sum to null) on top of the dense result.

## Rebase notes

Rebased again onto develop (1a09eb4), 419 commits on. The aggregate
vtable now lends `AggregateArgs` (options plus resolved dtypes) to every
execution method, so `grouped_state` and `partials_to_array` take args
too, and `GroupedAccumulator` holds `AggregateDTypes` rather than three
dtype fields. `combine_partials` is gone: the default grouped state
parses with `partial_from_scalar` and folds with `merge_partials`.
Kernel lookup moved to the `ExecutionCtx` snapshot, so `AggregateKernels`
now carries the single `(aggregate, values encoding, group ids encoding)`
registry this PR introduces. `array_session()` is gone from tests, and
`DecimalType::smallest_decimal_value_type` is now
`DecimalDType::canonical_values_type`.

`develop` reverted SQL null-on-empty semantics for `Sum` (#9324) and
re-landed them as a separate `SumV2` (#9590), so this branch follows
both contracts: `Sum` starts a group at zero and treats a null partial
as overflow, while `SumV2` keeps `{sum, is_overflow, is_empty}` and
finalizes an untouched group to null. `SumAggregateOpts` is gone, and
both aggregates take `NumericalAggregateOpts`.

`SumV2`'s list-shaped `PrimitiveGroupedSumV2EncodingKernel` is replaced
by a dense kernel that reuses `Sum`'s accumulation and then clears the
empty flag of every group that saw a valid value, so `sum_v2` keeps a
fast grouped path rather than falling back to per-group accumulators.

`list_sum` now sums with `SumV2` (#9621), so its empty and all-null list
masking is gone; it only has to restore the one case dense ids cannot
express, a null list.

The single-partial `flush` shortcut from #9597 no longer applies: dense
state produces one partial array per flush, so there is no chunk to
collapse, and its three tests went with it.

## Review comments

- Group ids are an `ArrayRef` so kernels can dispatch on their encoding
  (Constant, RLE, ...), and `GroupIds` is the "Group struct" the
  accumulator API talks in.
- The hand-rolled `for_each_valid_idx` helpers are gone: the kernels
  take the mask's bit buffer, handle the all-true and all-false cases
  directly, and let `BitBuffer::for_each_set_index` walk the rest
  instead of materializing the mask's index vector.
- The run-detection helpers shared by the count and sum kernels moved to
  `aggregate_fn::kernels` instead of being duplicated per aggregate.
- The accumulator tracks `num_groups` itself, so `flush_partials` and
  `finish` no longer take it; `ensure_groups` is exposed for callers
  that need trailing groups no batch mentioned.
- `merge_group` names its parameters `into_group` / `from_group`.
- The benchmark input builder is now `DenseGroupedInput::clustered`.

## Run-encoded group ids

An input that is already grouped - a list array adapted by
`GroupedArray::dense_input` - would otherwise pay a `u32` per element
just to say "these rows belong to that group". When the groups cover the
element array in order, the adapter now hands over
`GroupIds::from_runs`: the ids are `PiecewiseSequence` runs (each group
id repeated for its run length, via a zero multiplier), and `Count`,
`Sum`, and `SumV2` each register a kernel for that group-id encoding
which reduces one run at a time. `GroupIds::runs` caches the decoded
runs the same way `validated_ids` caches materialized ids, so several
aggregates over the same ids decode once.

Each run kernel falls back to per-row accumulation when the ids turn out
not to be runs, so registering for the encoding never loses the generic
path. Tests assert run ids and materialized ids agree for every aggregate
across empty runs, all-null groups, overflow, and NaN options.

## list_sum performance

`list_sum` is the caller that pays for the dense-ids API on input that is
already grouped. Best-of-three alternating runs of the two benchmark
binaries, develop (1fe8dda) vs this branch on the same machine:

| benchmark | develop | this branch |
|---|---:|---:|
| `fsl_sum_large` | 7.0 us | 10.9 us |
| `fsl_sum_medium` | 5.8 us | 8.5 us |
| `fsl_sum_small` | 4.8 us | 6.6 us |
| `list_sum_large` | 11.4 us | 15.6 us |
| `list_sum_medium` | 9.6 us | 12.7 us |
| `list_sum_small` | 8.5 us | 10.8 us |
| `list_sum_nullable_elements_large` | 21.8 us | 27.8 us |
| `list_sum_nullable_elements_medium` | 14.2 us | 19.0 us |
| `listview_sum_large` | 7.2 us | 11.4 us |
| `listview_sum_medium` | 5.8 us | 8.7 us |
| `listview_sum_small` | 5.0 us | 7.3 us |

Still 27-59% slower. The run kernels removed the per-element id work,
and what is left scales with groups rather than elements: these
benchmarks run 100-500 groups over 800-4000 elements, where the dense
pipeline's fixed costs - building the run ids, dense state, flushing it
to arrays, and the extra mask that restores a null list - are the whole
measurement. `SumV2`'s list kernel instead writes its three output
buffers in one pass over the ranges.

Whether `list_sum` should keep a list-shaped kernel for that case, or
accept the overhead in exchange for one grouped API, is a call for this
PR's reviewers. The dense path is the one that wins at engine scale: the
`aggregate_grouped` benchmarks, which run 65k rows, are what this API
exists for.

## Validation

- `cargo clippy -p vortex-array --all-targets --all-features -- -D warnings`
- `cargo test -p vortex-array --lib` (5,270 passed, 1 ignored)
- `cargo test --doc -p vortex-array` (85 passed, 21 ignored)
- `cargo +nightly-2026-09-10 fmt -p vortex-array`

Benchmarks were not re-run on this rebase, so the `list_sum` figures
above are from the previous base.

Signed-off-by: Nicholas Gates <[email protected]>
Signed-off-by: Robert Kruszewski <[email protected]>
Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01B8VcUXWmDmNH2JjJY4aDQ8

Rebases #8379 onto develop and addresses the outstanding review
comments.

The grouped accumulator now takes caller-assigned dense `GroupIds`
(`{ids: ArrayRef, num_groups}`) directly, instead of requiring values to
be pre-sorted into complete list groups. Ids may repeat, arrive out of
order, or be absent from a batch.

## Changes

- Dispatch grouped kernels on both the values encoding and the group-id
  encoding, with wildcard fallbacks in each dimension.
- Carry run-encoded group ids as `PiecewiseSequence` runs, and register
  `Count`, `Sum`, and `SumV2` kernels for that encoding so already-grouped
  input reduces a run per group instead of reading an id per row.
- Cache materialized and validated group ids across clones, so several
  aggregates over the same ids validate once.
- Let each aggregate own its dense grouped state (`GroupedState`), with
  the generic one-partial-per-group implementation as the fallback.
- Store grouped count state as one `u64` per group and grouped sum state
  in monomorphic `u64`/`i64`/`f64`/decimal vectors, choosing the decimal
  width once from the sum dtype.
- Port `SumV2` onto the same dense sums, adding the per-group empty flag
  its SQL semantics need, so both sums share one accumulation kernel.
- Merge count and sum partial arrays without `execute_scalar` per row.
- Adapt between typed scatter for unclustered ids and per-run reduction
  for clustered ids.
- Use one stable counting-sort gather per batch in the universal
  fallback instead of one `take` per group.
- Keep `GroupedArray::dense_input` as the list-group adapter for
  `list_sum`, which restores list semantics (null, empty, and all-null
  lists sum to null) on top of the dense result.

## Rebase notes

Rebased again onto develop (96fc43d), 478 commits past the original base.
The aggregate vtable lends `AggregateArgs` (options plus resolved dtypes) to
every execution method, so `grouped_state` and `partials_to_array` take args
too, and `GroupedAccumulator` holds `AggregateDTypes` rather than three
dtype fields. `combine_partials` is gone: the default grouped state parses
with `partial_from_scalar` and folds with `merge_partials`. Kernel lookup
moved to the `ExecutionCtx` snapshot, so `AggregateKernels` now carries the
single `(aggregate, values encoding, group ids encoding)` registry this PR
introduces. `array_session()` is gone from tests, and
`DecimalType::smallest_decimal_value_type` is now
`DecimalDType::canonical_values_type`.

`develop` reverted SQL null-on-empty semantics for `Sum` (#9324) and
re-landed them as a separate `SumV2` (#9590), so this branch follows
both contracts: `Sum` starts a group at zero and treats a null partial
as overflow, while `SumV2` keeps `{sum, is_overflow, is_empty}` and
finalizes an untouched group to null. `SumAggregateOpts` is gone, and
both aggregates take `NumericalAggregateOpts`.

`SumV2`'s list-shaped `PrimitiveGroupedSumV2EncodingKernel` is replaced
by a dense kernel that reuses `Sum`'s accumulation and then clears the
empty flag of every group that saw a valid value, so `sum_v2` keeps a
fast grouped path rather than falling back to per-group accumulators.

`list_sum` now sums with `SumV2` (#9621), so its empty and all-null list
masking is gone; it only has to restore the one case dense ids cannot
express, a null list.

The single-partial `flush` shortcut from #9597 no longer applies: dense
state produces one partial array per flush, so there is no chunk to
collapse, and its three tests went with it.

## Review comments

- Group ids are an `ArrayRef` so kernels can dispatch on their encoding
  (Constant, RLE, ...), and `GroupIds` is the "Group struct" the
  accumulator API talks in.
- The hand-rolled `for_each_valid_idx` helpers are gone: the kernels
  take the mask's bit buffer, handle the all-true and all-false cases
  directly, and let `BitBuffer::for_each_set_index` walk the rest
  instead of materializing the mask's index vector.
- The run-detection helpers shared by the count and sum kernels moved to
  `aggregate_fn::kernels` instead of being duplicated per aggregate.
- The accumulator tracks `num_groups` itself, so `flush_partials` and
  `finish` no longer take it; `ensure_groups` is exposed for callers
  that need trailing groups no batch mentioned.
- `merge_group` names its parameters `into_group` / `from_group`.
- The benchmark input builder is now `DenseGroupedInput::clustered`.

## Run-encoded group ids

An input that is already grouped - a list array adapted by
`GroupedArray::dense_input` - would otherwise pay a `u32` per element
just to say "these rows belong to that group". When the groups cover the
element array in order, the adapter now hands over
`GroupIds::from_runs`: the ids are `PiecewiseSequence` runs (each group
id repeated for its run length, via a zero multiplier), and `Count`,
`Sum`, and `SumV2` each register a kernel for that group-id encoding
which reduces one run at a time. `GroupIds::runs` caches the decoded
runs the same way `validated_ids` caches materialized ids, so several
aggregates over the same ids decode once.

Each run kernel falls back to per-row accumulation when the ids turn out
not to be runs, so registering for the encoding never loses the generic
path. Tests assert run ids and materialized ids agree for every aggregate
across empty runs, all-null groups, overflow, and NaN options.

## list_sum performance

`list_sum` is the caller that pays for the dense-ids API on input that is
already grouped. Best-of-three alternating runs of the two benchmark
binaries, develop (1fe8dda) vs this branch on the same machine:

| benchmark | develop | this branch |
|---|---:|---:|
| `fsl_sum_large` | 7.0 us | 10.9 us |
| `fsl_sum_medium` | 5.8 us | 8.5 us |
| `fsl_sum_small` | 4.8 us | 6.6 us |
| `list_sum_large` | 11.4 us | 15.6 us |
| `list_sum_medium` | 9.6 us | 12.7 us |
| `list_sum_small` | 8.5 us | 10.8 us |
| `list_sum_nullable_elements_large` | 21.8 us | 27.8 us |
| `list_sum_nullable_elements_medium` | 14.2 us | 19.0 us |
| `listview_sum_large` | 7.2 us | 11.4 us |
| `listview_sum_medium` | 5.8 us | 8.7 us |
| `listview_sum_small` | 5.0 us | 7.3 us |

Still 27-59% slower. The run kernels removed the per-element id work,
and what is left scales with groups rather than elements: these
benchmarks run 100-500 groups over 800-4000 elements, where the dense
pipeline's fixed costs - building the run ids, dense state, flushing it
to arrays, and the extra mask that restores a null list - are the whole
measurement. `SumV2`'s list kernel instead writes its three output
buffers in one pass over the ranges.

Whether `list_sum` should keep a list-shaped kernel for that case, or
accept the overhead in exchange for one grouped API, is a call for this
PR's reviewers. The dense path is the one that wins at engine scale: the
`aggregate_grouped` benchmarks, which run 65k rows, are what this API
exists for.

## Validation

- `cargo clippy -p vortex-array --all-targets --all-features -- -D warnings`
- `cargo test -p vortex-array --lib` (5,347 passed, 1 ignored)
- `cargo test --doc -p vortex-array` (85 passed, 21 ignored)
- `cargo +nightly-2026-09-10 fmt -p vortex-array`

Benchmarks were not re-run on this rebase, so the `list_sum` figures
above are from the previous base.

Signed-off-by: Nicholas Gates <[email protected]>
Signed-off-by: Robert Kruszewski <[email protected]>
Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01B8VcUXWmDmNH2JjJY4aDQ8