[VL] Lazy aggregate expand: aggregate at the finest grain below Expand for grouping analytics#12554
Draft
malinjawi wants to merge 1 commit into
Draft
[VL] Lazy aggregate expand: aggregate at the finest grain below Expand for grouping analytics#12554malinjawi wants to merge 1 commit into
malinjawi wants to merge 1 commit into
Conversation
…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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Adds
LazyAggregateExpandRuleto 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:
After:
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, notry_sum, nobloom_filter_aggGROUPING SETS ((),())empty-input case)RewriteDistinctAggregatesExpand); deterministic pre-projections and filters onlysum/avgfollow the samefloatingPointModepolicy as flushable aggregationBenchmarks (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):
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 withgrouping_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.