[VL] Lazy aggregate expand: aggregate at the finest grain below Expand for grouping analytics

[malinjawi]

What changes were proposed in this pull request?

Adds LazyAggregateExpandRule to the Velox backend: a post-transform rule that, for grouping analytics (ROLLUP / CUBE / GROUPING SETS), moves the partial aggregation below the Expand operator so the input is aggregated once at the finest grain, then expands only the intermediate aggregation buffers and merges them before shuffle.

Before:

ColumnarExchange
  HashAggregate(Partial)        <- hashes (input rows x grouping sets)
    Expand                      <- input rows x grouping sets
      child

After:

ColumnarExchange
  HashAggregate(PartialMerge, flushable)  <- collapses coarse-grain duplicates pre-shuffle
    Expand                                <- fine-grain groups x grouping sets (buffers only)
      HashAggregate(Partial, flushable)   <- input rows, finest grain
        child

Gated by spark.gluten.sql.columnar.backend.velox.lazyAggregateExpand.enabled (default false). No native changes; the rewrite reuses existing companion-function, extract-struct/row-construct, and flush/abandon machinery.

Why are the changes needed?

Expand multiplies every input row by the number of grouping sets (TPC-DS q67: 9x on an 8-key rollup), and the partial aggregate pays hashing and shuffle for every copy. When the finest grain reduces the row count, aggregating first is substantially cheaper, and the coarser grouping sets collapse from already-aggregated states.

Safety on high-cardinality keys (where the finest grain does not reduce): both inserted aggregates are flushable, so Velox's abandon-partial-aggregation turns the bottom aggregate into a streaming to-intermediate conversion and the merge stage into an identity pass-through - the worst case degrades to roughly the original plan plus one cheap pass. This is the failure mode that regressed q67 for the CH backend's lazy expand (#7986, fixed there by a native re-aggregation step in #7995); here the protection comes from existing Velox machinery.

Prior art: the CH backend has this optimization since #7649 (GLUTEN-7647, default-on there). #12052 already explored the same three-stage shape for the Velox backend (including the pre-shuffle PartialMerge); this PR differs in making the finest-grain aggregate flushable so both stages inherit Velox's abandon adaptivity (the worst-case protection on high-cardinality keys), binding aggregation buffers by exprId against a verified ordering contract, the eligibility guards above, and a dedicated test suite.

Eligibility (v1, conservative)

• sum / count / min / max / avg; no DISTINCT, no FILTER, no try_sum, no bloom_filter_agg
• all modes Partial; grouping and shuffle keys are attributes/literals; at least one attribute-backed grouping key of atomic type (guards the degenerate GROUPING SETS ((),()) empty-input case)
• aggregate inputs must be pass-through columns of the Expand (this also auto-rejects the RewriteDistinctAggregates Expand); deterministic pre-projections and filters only
• float sum/avg follow the same floatingPointMode policy as flushable aggregation
• every new node passes native validation, otherwise the rule leaves the plan untouched

Benchmarks

SQL-level simulation of the rewrite on vanilla Spark 4.0.1 (local[10], 12-core machine; result sets checksum-verified identical in every case). The simulation pays one extra shuffle the actual rule avoids, so it understates the rule — but it measures the same underlying algorithm.

Real TPC-DS q67, SF10, end to end (real dsdgen data, full query incl. joins and the rank() window):

	baseline	rewrite	delta
q67 @ SF10, end-to-end	157.7s	165.6s	0.95x (within noise)

Run-to-run spread was large (baseline 147-194s, rewrite 120-208s), so this is best read as "no measurable end-to-end win", not as a precise 5% regression.

Why, measured on the same data:

rows into the rollup	5,342,291
distinct 8-key groups	4,804,937
rows per group	1.11

q67's finest grain is very nearly unique, so aggregating below the Expand has almost nothing to collapse. This optimization's premise is that the finest grain reduces the row count materially; on q67 it does not.

Synthetic sweeps, showing where the premise does and does not hold (stage-level, no window/joins):

dataset	rows per group	baseline	rewrite	delta
50M rows, 4 keys, 100k groups	~500	7.5s	2.3s	3.2x faster
30M rows, 8 keys, 9 sets	3.14	134.6s	43.3s	3.1x faster
20M rows, near-unique keys	~1	25.0s	47.6s	1.9x slower (unprotected)

An earlier revision of this description presented the middle row as "q67-shaped". That was wrong: at 3.14 rows/group it has roughly 3x more finest-grain reduction than real q67 (1.11), which is the variable that decides the outcome. The table above is corrected.

Reading of the evidence. The rewrite is a large win when the finest grain reduces (many rollup/cube workloads), a no-op when it does not (q67), and a regression when it does not AND the adaptive valve is absent — which is what the flushable-aggregate requirement in this PR exists to prevent. q67 specifically needs the amplification removed without depending on fine-grain reduction, i.e. a fused rollup operator (see #12052 discussion); that is complementary to this PR, not served by it.

Gluten A/B numbers with the flag on/off at scale will follow - keeping this as a draft until then.

How was this patch tested?

New LazyAggregateExpandSuite (20 cases): plan-shape assertions (flushable aggregates on both sides of the Expand) plus result comparison against vanilla Spark, covering rollup/cube with sum/count/min/max/avg, genuine-NULL keys vs rolled-up NULL with grouping_id, decimal sum buffers, the pre-projected q67 shape, aggregates over grouping keys, empty input, degenerate grouping sets, duplicate grouping sets, single and multi count-distinct, FILTER / strict-float / non-whitelisted-function / non-atomic-key rejections, early-abandon interplay, AQE on/off, and flushable-disabled behavior.