Skip to content

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

Draft
malinjawi wants to merge 1 commit into
apache:mainfrom
malinjawi:vl-lazy-aggregate-expand
Draft

[VL] Lazy aggregate expand: aggregate at the finest grain below Expand for grouping analytics#12554
malinjawi wants to merge 1 commit into
apache:mainfrom
malinjawi:vl-lazy-aggregate-expand

Conversation

@malinjawi

Copy link
Copy Markdown
Contributor

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 explored the same direction for the Velox backend; this PR adds the pre-shuffle merge stage, exprId-based attribute binding, the adaptive worst-case protection, 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 (algorithm validation)

SQL-level simulation of the rewrite on vanilla Spark 4.0.1 (local[10], 12-core machine; result sets checksum-verified identical; the simulation pays one extra shuffle the actual rule avoids, so these numbers understate the rule):

dataset fine-grain profile baseline rewrite delta
50M rows, 4 keys, 100k fine groups strong reduction 7.5s 2.3s 3.2x faster
30M rows, 8 keys, 9 sets, 3.1 rows/group (q67-shaped) weak fine reduction, strong coarse collapse 134.6s 43.3s 3.1x faster
20M rows, near-unique keys no reduction 25.0s 47.6s unprotected worst case; motivates the flushable/abandon requirement above

Gluten A/B numbers on TPC-DS (q67, q18, q22, q36, q77, q80) with the flag on/off 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.

…d for grouping analytics

For rollup/cube/grouping-sets aggregation, Expand multiplies every input
row by the number of grouping sets before the partial aggregation, so the
partial aggregate consumes and hashes (input rows x grouping sets) rows.

This adds a Velox-backend rule that rewrites eligible plans to aggregate
at the finest grain once, expand only the intermediate aggregation
buffers, and merge the expanded states before shuffle:

  partial agg <- expand <- child
becomes
  partial-merge agg <- expand(buffers) <- flushable partial agg <- child

The pre-shuffle merge collapses duplicated coarse-grain groups so shuffle
volume does not increase, and both new aggregates inherit Velox's
flush/abandon adaptivity, bounding the worst case on high-cardinality
grouping keys (the regression class seen with the ClickHouse backend's
lazy expand in apache#7986).

Ports the ClickHouse backend's LazyAggregateExpandRule (apache#7649) approach
to the Velox backend and builds on the direction of apache#12052.

Gated by spark.gluten.sql.columnar.backend.velox.lazyAggregateExpand.enabled,
default false.
@github-actions github-actions Bot added the VELOX label Jul 18, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant