perf(aggregate): specialize fully ordered group keys without hashing - #25150

…pache#25150)

## Which issue does this PR close?

Closes apache#14574 for supported fully ordered multi-column keys.
Single-column specializations and additional key types remain on their
existing paths.

## Rationale for this change

When input is already sorted by all grouping expressions, DataFusion can
emit completed groups early, but its multi-column group-key mapping
still hashes each row and maintains a hash table. This adds CPU and
retained state even though equal keys must be consecutive.

## What changes are included in this PR?

- Add an internal columnar `GroupValuesOrdered` selected only for
`GroupOrdering::Full` and supported multi-column schemas. Arrow's
partition kernel identifies adjacent runs; existing column builders
retain representatives and compare the boundary key across batches.
Dense IDs, partial emission, preserving selections, and reset behavior
follow the existing contracts.
- Support boolean, integer, string, and binary keys, including their
large/view variants. Keep single-column, float, nested, dictionary,
partially ordered, and unordered inputs on established paths.
- Add differential correctness tests and a reproducible benchmark
covering two key schemas, batch sizes 127/8192, and run lengths
1/8/128/8192. Include both grouping with emission and complete ordered
`AggregateExec` execution.
- Include two small Windows lint compatibility fixes needed to run the
repository's full Clippy command: remove a redundant semicolon in
rounding and add a Windows-only `unused_async` expectation to the
unsupported FIFO example's uniform async entry point.

This builds on the problem and review in apache#9662, particularly its
single-column regression and requests for repeatable measurements and
cross-batch tests. It is distinct from apache#23986's consecutive-key cache
for clustered input, which retains the general hash path. This PR
optimizes existing streaming aggregation; it does not introduce
streaming aggregation or change whole-query asymptotic complexity.

## What is the testing strategy for this PR?

`ordered_keys_match_hash_grouping_across_batches_and_emits` compares
group IDs and emitted/preserved keys with `GroupValuesColumn<true>` for
15 key types, ascending/descending runs, NULLs, long strings, six batch
sizes, and four emission limits. It also exercises empty batches and
reset/reuse. A schema gate test covers excluded types.

`ordered_single_and_partial_final_match_unordered_execution` compares
actual `AggregateExec` results with unordered execution for Single and
Partial→Final with FILTER, using four batch sizes and asserted sorted
execution.

Validation after the September 26 conflict-resolution rebase onto `main`
at `6a792c6` (Windows, Rust 1.98.1): `cargo fmt --all -- --check`, `git
diff --check`, `cargo test --offline --locked -p
datafusion-physical-plan --lib` (**2,323 passed, 0 failed**), focused
physical-plan library Clippy with `-D warnings`, and `cargo check
--offline --locked -p datafusion-physical-plan --bench
ordered_group_values` passed. Full `cargo clippy --all-targets
--all-features -- -D warnings` was attempted but stopped while linking
`protobuf-src`'s `protoc` on this Windows machine (`LNK2019`/`LNK1120`);
it is not a passing full lint result. Fresh Linux CI remains to be seen
for the rebased head.

Earlier local validation on Windows, before this rebase:

- Rust 1.97.0: full `cargo clippy --all-targets --all-features -- -D
warnings` and formatting checks pass.
- Rust 1.98.1, CI profile, extended workspace features: 11,333 tests
passed across recorded runs, including all 2,050 physical-plan tests,
120 fuzz tests, and the two 5M-row SMJ low-memory runners. Eight tests
are ignored, with no pending test programs. Seven baseline-reproduced
failures remain, so this is not a full-workspace pass.
- Seven failures reproduce with the same assertion differences on
unmodified baseline `423df6f4`: two datasource path tests, four
floating-point interval/statistics tests, and the xtask CI command
snapshot. These remain recorded failures.
- Windows library backtrace capture was disabled for the extended run
because it made error-path tests exceptionally slow. The common crate's
650 tests, including backtrace behavior, pass separately with capture
enabled. A CRLF checkout of the Q15 fixture was normalized to its
original LF contents so the existing multi-statement skip works as
intended; no test assertion was changed.
- An external process interruption required resuming missing tests from
Cargo's compiled artifact list. Completed suite summaries and unique
individual test results were retained; this does not represent one
successful full-workspace Cargo invocation. Linux CI passed on the
pre-rebase head `a62b640`.

The following benchmark measurements predate the September 26 rebase and
have not been rerun against `6a792c6`. They were collected on an Intel
Core Ultra 7 258V / Windows, Rust 1.98.1, `release-nonlto`, pinned to
logical CPU 0, with no concurrent task builds/tests. Baseline and
candidate were built from separate source directories with identical
lockfiles. The run order was baseline, candidate, candidate, baseline;
each case used 250 ms warmup, 10 samples, and at least 1 second of
measurement.

The complete ordered `AggregateExec` benchmark uses 8,192-row batches.
Speedup ranges below are `min(baseline) / max(candidate)` through
`max(baseline) / min(candidate)` across the two repetitions; they are
not confidence intervals for the ratio.

| Keys / rows per group | Baseline ms (two runs) | Candidate ms (two
runs) | Speedup range |
|---|---:|---:|---:|
| int / 1 | 6.086, 5.269 | 1.609, 1.857 | 2.84–3.78x |
| int / 8 | 2.434, 2.296 | 0.480, 0.609 | 3.77–5.07x |
| int / 128 | 2.112, 1.949 | 0.549, 0.598 | 3.26–3.85x |
| int / 8192 | 2.162, 2.059 | 0.545, 0.553 | 3.73–3.97x |
| string / 1 | 9.616, 9.665 | 4.793, 5.521 | 1.74–2.02x |
| string / 8 | 4.333, 4.422 | 1.697, 1.660 | 2.55–2.66x |
| string / 128 | 3.723, 3.464 | 1.747, 1.638 | 1.98–2.27x |
| string / 8192 | 3.907, 3.745 | 1.613, 1.607 | 2.32–2.43x |

All 16 selected grouping microbenchmark cases also have faster point
estimates in both repetitions; the weakest cross-repetition ratio is
approximately 1.06x for long strings in 127-row batches. Small
differences on one machine should not be generalized across platforms.
At batch size 8192 / run length 1, peak retained group state drops from
425,792 to 196,608 bytes for integer keys and from 950,080 to 720,896
bytes for integer/long-string keys.

The benchmark pre-generates 131,072 rows and excludes input generation,
sorting, SQL planning, and I/O. The grouping benchmark includes
partial/final emission; the execution benchmark includes SUM and output
collection on already ordered input. `group_state_bytes` reports peak
retained `GroupValues::size()`, excluding shared inputs and transient
kernel allocations; it is not process RSS.

## Are there any user-facing changes?

Eligible fully ordered multi-column aggregations use less grouping state
and avoid hashing. No SQL syntax, result semantics, configuration, or
public API changes.

This contribution is AI-assisted.

---------

Co-authored-by: hushen <[email protected]>
Co-authored-by: kosiew <[email protected]>

…6370)

## Which issue does this PR close?

Closes apache#26369. Follow-up to apache#25150.

## Rationale for this change

Fully ordered multi-column `GROUP BY` still hashes temporal, decimal,
and fixed-size binary keys. These types can use the existing
adjacent-run grouping path because Arrow's adjacent equality agrees with
the grouping builders.

In our production usecases, we are especially interested in apache#26369's
optimization extending to `Timestamp`s (timeseries data) and
`FixedSizeBinary(16)` (for UUIDs).

## What changes are included in this PR?

- **Only production code change is 12 new types** added to the `match`
under `supports_schema`: the rest are tests and benches.
- Enable timestamps with all units and timezone metadata, dates, valid
time units, durations, intervals, Decimal32/64/128/256, and every valid
fixed-size binary width.
- Preserve type metadata and the existing selection of single-column
specializations.
- Add a `fully_ordered` group to the existing SQL aggregation Criterion
benchmark, with integer/string controls, timestamp run-length variants,
and a timestamp `SUM(v)` case. The first two commits contain only
benchmarks, followed by the type expansion and regression tests,
including the UUID coverage.

## What is the testing strategy for this PR?

Differential tests compare ordered and hash grouping across nulls,
units/timezones, decimal metadata, sliced batches, and group emission.
Dedicated tests cover zero-width binary and 16-byte binary keys that
share their first 15 bytes, including cross-batch continuation and keys
that differ only in the last byte. SQL regression tests cover
single-stage and partial/final aggregation, mixed sort directions, null
ordering, and aggregate filters.

The benchmarks check input types, row and group counts, and fully
clustered aggregation without sorting or repartitioning in the measured
query. The timestamp `SUM(v)` case also checks every group's sum before
timing.

<details>
<summary>SQL benchmark results</summary>

[Latest benchmark bot
report](apache#26370 (comment)),
comparing the benchmark-only baseline `dc6a10cf3c8` with `a398f8768ca`.
Both refs contain the same 17 benchmark cases. The table shows rounded
speedups from the report. The controls are roughly unchanged; the small
differences may be benchmark noise. Values below 1× mean slower.

| Second grouping key / query | Rows per group | Approx. speedup |
| --- | ---: | ---: |
| Int64 (control) | 31 | ~0.95× |
| Utf8View (control) | 31 | ~0.99× |
| Timestamp (ns) | 31 | ~7.0× |
| Timestamp (µs, UTC) | 31 | ~6.9× |
| Date32 | 31 | ~7.5× |
| Time32 (ms) | 31 | ~7.5× |
| Time64 (ns) | 31 | ~7.0× |
| Duration (ns) | 31 | ~7.0× |
| Interval (day/time) | 31 | ~6.4× |
| Interval (month/day/ns) | 31 | ~5.3× |
| Decimal128(30, 4) | 31 | ~6.1× |
| Decimal256(60, 4) | 31 | ~5.2× |
| FixedSizeBinary(16) | 31 | ~4.6× |
| FixedSizeBinary(7) | 31 | ~4.4× |
| Timestamp (ns), singleton groups | 1 | ~2.9× |
| Timestamp (ns), long groups | 512 | ~7.7× |
| Timestamp (ns), `SUM(v)` | 31 | ~4.0× |

The 16 key-only cases use the same two-key deduplication query:

```sql
SELECT k1, k2 FROM ordered_input GROUP BY k1, k2
```

The additional timestamp case includes accumulator work:

```sql
SELECT k1, k2, SUM(v) FROM ordered_input GROUP BY k1, k2
```

Here `v` is the nonconstant source row number. The timestamp `SUM(v)`
case improved from approximately 43.9 ms to 11.0 ms (~4.0×).

- **Input:** about 2.1 million rows. `k1` is an Int32 prefix that
advances every 32 groups; nullable `k2` cycles through 32 values,
including one null, using the type listed above. Both keys vary.
- **Group sizes:** normally 31 rows per key pair, so runs cross the
8,192-row batch boundaries. The additional timestamp cases use one or
512 rows per group to vary cardinality.
- **Before timing:** generate the data, sort by both keys, and
materialize it into one in-memory partition with its ordering declared
to the planner.
- **Measured work:** SQL planning, query execution, and collection of
all results. The key-only cases focus on grouping the keys; the
timestamp `SUM(v)` case also updates an accumulator.
- **Setup checks:** verify input type, row count, and output group
count; require fully clustered aggregation and fail if the measured plan
adds sorting or repartitioning. The `SUM(v)` case verifies every group's
result, including null timestamp keys and the final short group, before
timing.

</details>

## Are there any user-facing changes?

Already-sorted multi-column aggregations can use lower-overhead key
storage for these types. Query results stay the same.

Developed with AI assistance.