perf: stream a converted SortAggregate whose ordering a consumer needs

[msafonov]

Fixes the gold regression from #12/#14 (gold_device_profiles 131 vs 59 task-h, gold_user_profiles 40 vs 26 on night 2026-10-11).

Root cause

• First-level aggregate of a COUNT(DISTINCT) expansion (keys include the distinct values + gid) feeds Spark's next SortAggregate, which needs the ordering. feat: native MIN/MAX over strings and SortAggregate as a native hash aggregate #12 dropped the sort, ran a native hash aggregate with almost no reduction (16.7B → 14.2B rows), which spilled 523 GiB, and then restored a full sort above it (452 GiB spill).
• MaxMinByGroupsAccumulator::size() walked all groups after every batch: quadratic for hash aggregates with millions of groups (map stage 61.5 h vs 20 h on 1.1-4).

Change

• If a consumer requires the SortAggregate's ordering, keep the sort and run the native aggregate over it; DataFusion detects sorted input and uses the ordered (streaming) aggregate, so the output stays ordered and no sort is restored. The converted node keeps SortAggregateExec as originalPlan (correct revert, correct outputOrdering). New proto flag HashAggregate.ordered_input: if DataFusion does not detect the order, the planner sorts the output, so the promise always holds.
• Without such a consumer the sort is dropped as before (the feat: native MIN/MAX over strings and SortAggregate as a native hash aggregate #12/feat: native MAX_BY and MIN_BY with a string value #14 wins stay).
• MaxMinBy groups accumulator keeps a running byte count; size() is O(1).

Local measurement (synthetic device_profiles shape, 16M rows, local[4], 6g off-heap; task-s, 4 runs)

	vanilla	1.1-7	branch
device_profiles	~193	~388	~185
string min/max (win case)	~18	~5	~3.8
string min/max + SMJ	~16	~4	~3.1

Tests

CometAggregateSuite (new plan-shape tests: SMJ consumer, Expand → SortAggregate, chained aggregates, cost-based revert), CometExecRuleSuite, CostBasedEngineChoiceSuite, WideRowSortFallbackSuite, RevertNativeForTransitionHeavyStagesSuite, CometSqlFileTestSuite, TPC-DS plan stability v1.4/v2.7, CometExecSuite, agg fuzz full matrix; rust spark-expr + core lib tests; fmt/clippy/spotless/scalastyle clean.

🤖 Generated with Claude Code

[msafonov]

Full gold_device_profiles on GCP (test image msaf-spark-comet:sortagg-test-9153d794f, gold-thrift executor sizing: 8 cores, 34g off-heap, max 20 executors; same inputs as prod night 2026-10-11, asOf 2026-10-11):

	vanilla	1.1-4	1.1-7	branch
main query task-h	60.2	59.2	131	49.4
map stage (scan+expand+partial agg) task-h	17.3	20.0	61.5	15.9
level-1 merge stage task-h / disk spill	19.3 / 361 GB	17.5 / 123 GB	48.5 / 1046 GB	15.1 / 0 GB

Output vs prod gold.device_profiles (v3573): row count equal (2,392,958,645), key sets equal, all columns equal except adtech_revenue_total: about 3k rows differ by exactly 0.0001 (ROUND(SUM(double), 4) summed in a different order), sums within tolerance. Fast-compare verdict PASS.

🤖 Generated with Claude Code