Planner rules: optimizer rewrites, AQE and DPP

vecruntime plans most of its work in the columnar rule that runs after Spark's planner. A few speedups need a different logical plan, not a different physical operator for the same plan. They are optimizer rules injected through SparkSessionExtensions.injectOptimizerRule (or, when Spark's own push-down would undo them, added to its last optimizer batch). This page lists every such rule and every change we make to Spark's adaptive execution (AQE) or dynamic partition pruning (DPP): what each one rewrites, when it applies and when it does not, why the result is the same, how to switch it off, and what it was measured to do.

Each rule here follows the same contract:

• It is general SQL, not tuned to particular queries. TPC-DS queries are the evidence, not the pattern the rule matches.
• It applies only where the rewrite is exact, and declines everything else. The tests cover both sides.
• It has its own spark.vecruntime.optimizer.* switch, and it is also off when the plugin is off (spark.vecruntime.enabled=false). Tests compare against Spark that way.
• It runs the gate, the SQL golden suite, and TPC-DS SF1 checksums against Spark. A performance claim carries a 1 TB A/B with control queries (AGENTS.md section 10).

The plugin changes no AQE setting. DppThroughAggregate and TransitiveDpp add dynamic partition pruning filters where Spark's DPP does not reach; Spark's own DPP settings are unchanged.

# SelfJoinToAggregate: existence-only self-joins as a min/max aggregate

spark/src/main/scala/org/apache/spark/sql/vecruntime/SelfJoinToAggregate.scala, PR #630.

Shape. An inner self-join whose condition compares one or more key columns for equality and exactly one more column for inequality:

SELECT t1.k FROM t t1, t t2 WHERE t1.k = t2.k AND t1.v <> t2.v

The rule replaces it with:

SELECT k FROM t WHERE k IS NOT NULL GROUP BY k HAVING min(v) <> max(v)

It keeps the other side's keys as aliases of the first side's, so nothing above needs renaming.

Why it is exact. Two rows of one key with different non-null v exist exactly when that key's smallest and largest non-null v differ. A null v or a null key never satisfies the join, and it never moves a min or a max. The rewrite returns each key once, where the join returned it once per matching pair. So the rule applies only where duplicates cannot be observed:

• on the build side of a left semi or anti join, or inside an IN / NOT IN / EXISTS / NOT EXISTS subquery;
• reached from there through projections, filters and inner joins only, with no aggregate in between;
• and nothing above reads a column of the self-join other than its keys.

Declined:

• sides that are not the same plan (sameResult);
• any other conjunct in the condition;
• two or more <> columns;
• a v whose = does not agree with min / max ordering: floating point, collated strings, and complex types (supported: integers, decimals, dates, timestamps, binary-collated strings);
• a self-join whose rows are counted, or whose other columns are read above.

Measured.

• TPC-DS q95's ws_wh CTE is this join, used twice under IN, and both uses are rewritten.
• SF1 checksums match Spark.
• At 1 TB on 2026-10-06, in one leg against main 13ea9ec, q95 went from 27.8 s to 13.4 s.

# MergeFilteredAggregates: global aggregates with different filters in one pass

spark/src/main/scala/org/apache/spark/sql/vecruntime/MergeFilteredAggregates.scala, PR #631.

Shape. Two or more global aggregates (no GROUP BY) over the same input plan that differ only in their Filters. They can appear in two ways:

• as a condition-less inner join of single-row aggregates (SELECT * FROM (SELECT count(*) ... WHERE a) x, (SELECT count(*) ... WHERE b) y, folded over any number of them);
• as uncorrelated scalar subqueries of one operator.

Spark's MergeScalarSubqueries merges only identical subqueries, and it does not cover the join form.

Rewrite. The rule walks both inputs together through projections, filters and inner joins with equal conditions; the leaves must be the same plan.

• Where the filters differ, the merged plan keeps the rows of either side, Filter(c1 OR c2). The predicates' columns are carried up through the projections.
• Each aggregate keeps only its own side's rows through FILTER (WHERE p).
• A DISTINCT aggregate gets the predicate in its argument instead, IF(p, x, NULL). That works for count, sum, avg, min and max, which ignore the null the same way.
• When a third or later input is folded in, an aggregate whose filter already implies the new predicate keeps its filter unchanged. Without this, q9's predicates grew to tens of KB each.

The two forms are then finished differently:

• Cross join: it becomes one aggregate. Each side yields exactly one row, so the join is the concatenation of the two rows.
• Scalar subqueries: each mergeable group is replaced by projections of one merged aggregate, which MergeScalarSubqueries then computes once.

Declined:

• different leaves;
• outer, semi or anti joins between the filter and the aggregate;
• grouped aggregates;
• non-deterministic expressions;
• a DISTINCT aggregate other than the five above.

Measured.

• SF1 checksums match Spark.

• At 1 TB on 2026-10-06, in one leg against main 13ea9ec:

  Query	main (s)	With the rule (s)
  q9	48.0	10.9
  q28	35.5	24.3
  q88	32.4	9.3
  q90	7.5	11.3

• q90: the merge applies and the single web_sales scan is identical to each of the two it replaces, but its tasks waited on S3 3-4x longer than in the main leg. That turned out to be S3 variance. A recheck alternating the two images (main, rules, main, rules; 3 iterations each) measured, in seconds:

  Query	main round 1	rules round 1	main round 2	rules round 2
  q90	8.8	5.4	9.5	5.5
  q88	121.3	6.9	121.1	6.3

  Checksums were identical. In both rounds main's q88 had two iterations near 121 s and one near 37 s.

Fix shipped with it. A grouped aggregate with FILTER failed with ArrayIndexOutOfBoundsException when a batch in which no row passed the filter brought new groups. The filtered function now sees every batch.

# SharedAggregateInputs: one input for aggregates that differ only by an inferred IS NOT NULL

spark/src/main/scala/org/apache/spark/sql/vecruntime/SharedAggregateInputs.scala, issue #632.

Shape. A CTE or view with a GROUP BY used twice is inlined twice, and Spark optimizes each copy on its own. InferFiltersFromConstraints adds IS NOT NULL(k) to the copy that is later joined on a grouping key k and pushes it down to the scan; the other copy (say, one aggregated again by a coarser key) does not get it. The two inputs no longer match, so ReuseExchange cannot share them, and the input -- typically a fact scan, a dimension join and a shuffle -- runs twice. TPC-DS q65 (store_sales) and q1 (store_returns) have this shape.

Rewrite. For two aggregates with the same grouping (by position in the input), the rule removes IS NOT NULL conjuncts on grouping keys from each input, walking projections that pass the key through, filters and inner joins. If the stripped inputs are the same plan, each aggregate gets the stripped input and the removed predicates as a Filter above it, adding the key to its output first when the parent had pruned it. Below an aggregate grouping by k, IS NOT NULL(k) removes exactly the rows of the null group, so filtering the null group out afterwards gives the same result.

Placement. The rule runs in Spark's last optimizer batch, User Provided Optimizers (spark.experimental.extraOptimizations). Every earlier batch is followed by Spark's own filter push-down (the last one is Pushdown Filters from PartitionPruning), which moves the predicates back below the aggregates. SparkSessionExtensions cannot inject into that batch, so a post-hoc analyzer rule, RegisterLateOptimizerRules, adds it to the session's extraOptimizations once, before the session's first optimization. A rewritten pair has the same input, so the rule does nothing on the batch's next pass.

Declined:

• aggregates with different groupings, or grouping by expressions rather than columns;
• inputs that still differ after the IS NOT NULLs are removed (another filter, another join);
• IS NOT NULL of anything other than a grouping column, or below an outer, semi or anti join, an aggregate, a window, or a projection that computes the key.

Measured.

• SF1 checksums match Spark on q1, q65, q30, q81, q24a and q18.

• At SF1, q65 and q1 plan 4 Parquet scans instead of 6: the fact scan and its date_dim join are planned once and reused.

• At 1 TB on 2026-10-06, the first version (PR #636) did not fire on q65 or q1. The fact table is partitioned there, and Spark's DPP gives each copy its own pruning subquery, which broke the comparison in two ways:

  • the subquery sits in a different place among each copy's filters, and canonicalization does not reorder an AND that holds a subquery;
  • DynamicPruningSubquery canonicalizes its build keys on their own, keeping their expression ids.

  The rule now rebuilds the remaining conjuncts in one canonical order. It compares inputs with the pruning subqueries' build keys normalized against their build plan. A partitioned-fact test covers this case.

• The 1 TB A/B of the fixed rule is pending.

# DppThroughAggregate: dynamic partition pruning through an aggregate

spark/src/main/scala/org/apache/spark/sql/vecruntime/DppThroughAggregate.scala, issue #633.

Shape. Join(Inner | LeftSemi, A, F, A.k = F.k') where:

• F has a selective filter (isLikelySelective, the test Spark's DPP uses);
• A reaches, through projections, filters and inner or semi joins, an Aggregate with k among its grouping columns;
• below the aggregate, k comes from a dimension D that is inner-joined to a fact on fact.p = D.d;
• p traces to a partition column of a file scan, through projections, filters and unions of fact branches.

TPC-DS q59 (store_sales grouped by d_week_seq, each use joined to a d_month_seq range of date_dim) and q2 (web_sales union catalog_sales, the same way) have it.

Why Spark does not prune it. PartitionPruning follows the join key down through the aggregate to D.k. That is not a partition column, so it stops. The inner fact-to-dimension join gets no pruning either: D has no selective filter of its own, and a dynamic filter does not count as one.

Rewrite. The fact scan gets Filter(DynamicPruningSubquery(p, D LEFT SEMI JOIN F ON D.k = F.k', D.d)), with onlyInBroadcast = false. Spark plans it like its own DPP: a broadcast reuse where one matches, else a one-off subquery over the small dimension, and the partition filter reaches FileSourceStrategy. For a union, every branch whose scan is partitioned on the matching column gets its own filter.

Why the result is the same. The aggregate groups by k, and above it an inner or semi join keeps only the k values F produces. A fact row whose dimension row has a k outside that set only feeds groups the join drops. Every dimension row whose k is in the set keeps its d, so no row of a surviving group is removed. Null keys never join, on either side.

Placement. Runs once in the Pre-CBO batch (injectPreCBORule): after filters were pushed down, and before Spark's PartitionPruning, which skips a join that already has a pruning filter on its side.

Declined:

• a join key that is not a grouping column of the aggregate (an aggregate output such as max(...));
• an outer join that preserves the aggregate's side;
• an F with no selective filter;
• a scan that already has a pruning filter on p;
• a key computed by an expression above the aggregate rather than passed through;
• non-deterministic filters or aggregates on the path.

Measured. DppThroughAggregateSuite covers the q59 and q2 shapes, which read fewer files than with the rule off, plus the two declined shapes and the switch. Results are compared with Spark with the plugin off.

At 1 TB on 2026-10-06, main 35e797d against the rule, with legs alternated (3 iterations each). Times are median seconds, and checksums are identical:

# TransitiveDpp: dynamic partition pruning through a second join key

spark/src/main/scala/org/apache/spark/sql/vecruntime/TransitiveDpp.scala, issue #634.

Shape. An inner Join(X, D) on X.p = D.d AND X.x = D.y, where:

• p traces to a partition column of a file scan inside X, through projections, filters, unions and inner or semi joins;
• D has no selective filter of its own;
• x traces, through projections, filters and inner joins, to a join-free subtree S of X that has one.

TPC-DS q72 is this shape. inventory joins date_dim d2 on inv_date_sk = d2.d_date_sk AND d1.d_week_seq = d2.d_week_seq, and d1 is restricted to one d_year. Without the rule, inventory is read in full.

Why Spark does not prune it. PartitionPruning finds inv_date_sk as a partition column, but the filtering side, d2, has no selective predicate (hasPartitionPruningFilter). What makes d2 selective is the second equality, which comes from d1 on the other side of the same join.

Rewrite. The scan gets Filter(DynamicPruningSubquery(p, D LEFT SEMI JOIN S ON D.y = S.x, D.d)), with onlyInBroadcast = false, planned like Spark's own DPP.

Why the result is the same. Every row the join keeps has p = D.d for a D row whose y equals an x. That x comes from a row of S that passed S's filter: only inner joins lie between S and the join. So that D row's d is in the pruning set, and no row the join keeps is removed.

Placement. Pre-CBO, once, before Spark's PartitionPruning, like DppThroughAggregate.

Declined:

• a selective D (Spark's own DPP covers it);
• a non-selective S;
• a D or S larger than spark.sql.autoBroadcastJoinThreshold, since the subquery scans both once more. This also keeps a filtered fact from ever being the source;
• a non-inner join at the top;
• a source key computed by an expression;
• a scan that already has a pruning filter on p.

Measured. TransitiveDppSuite runs over two partitioned facts and a date dimension, and compares results with Spark with the plugin off:

• the q72 shape prunes inventory and reads fewer files than with the rule off;
• an unfiltered source, an outer join, and the switch leave the plan alone.

The 1 TB numbers come with the PR.

# FactBloomFilter: a runtime bloom filter from a smaller fact onto a larger one

spark/src/main/scala/org/apache/spark/sql/vecruntime/FactBloomFilter.scala, issue #641.

Shape. A shuffle equi-join (neither side broadcastable) that can prune the application side (inner, left semi, or the null-supplying side of an outer join), where:

• the application key traces, through projections, filters and inner or semi joins, to a file scan of at least spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold (10 GB by default);
• the application side is at least spark.vecruntime.optimizer.factBloomFilter.sizeRatio (10) times larger in estimated bytes than the creation side;
• that scan has no dynamic pruning or bloom filter on the key yet.

TPC-DS q93 is this shape: store_sales (2.9G rows at 1 TB) joined to store_returns (~290M) on ticket and item. Without the rule, all of store_sales is shuffled into the join, though most of it has no match.

Why Spark does not add it. InjectRuntimeFilter.extractSelectiveFilterOverScan requires a selective Filter directly over the creation-side scan. store_returns has none of its own, so Spark builds nothing for q93, q50, q16 or q94. Raising Spark's bloom caps does not help and makes q75, q78 and q80 worse (a 20-query 1 TB run; the issue has the numbers).

Rewrite. The application scan gets Filter(might_contain(ScalarSubquery(BloomFilterAggregate( xxhash64(creationKey))), xxhash64(applicationKey))) -- the same BloomFilterAggregate and BloomFilterMightContain Spark's own rule uses, so the aggregate and the probe run vectorised. A composite join key gets one filter per key. The filter's expected items come from the creation side's row estimate, or its bytes over its row width when the row count is unknown (as at 1 TB); the bit size is capped by spark.sql.optimizer.runtime.bloomFilter.maxNumItems and maxNumBits.

Evidence of selectivity (#650). Size alone says nothing about pruning: at 1 TB the rule also built filters from whole dimension tables (customer, customer_address, item), whose keys every fact row matches, and each such filter only delayed its scan (q19 2.2 -> 4.8 s, q45 2.8 -> 5.7, q64 23.8 -> 71.9). The rule now needs evidence:

• Distinct counts, when they exist -- CatalogTable.stats.colStats from ANALYZE TABLE ... COMPUTE STATISTICS FOR COLUMNS (read directly, so spark.sql.cbo.enabled is not needed), or attributeStats of a DSv2 relation that reports column statistics (Iceberg, from Puffin theta sketches). Each key is traced through projections, filters and joins to its base relation (KeyStats). At most ndv(creation) / ndv(application) of the application rows can match; the filter is added when that is at most maxSelectivity (0.5), sized by the creation key's distinct count (capped by the creation side's row estimate).
• Otherwise, a reduced creation side -- a filter on it other than the IS NOT NULL of a column the optimizer infers from join keys. A whole, unfiltered table gets no filter.

Two-level merge (#646). bloom_filter_agg makes one partial filter per creation-side task, each the whole bit array (up to maxNumBits/8 bytes), and a single Final task would merge all of them: at 1 TB q17's filters merged 595 and 801 partials in one task each (3.8-6.6 GB read, 44-59 s). The subquery is therefore bloom_filter_merge(bloom_filter_agg(xxhash64(key)) GROUP BY pmod(spark_partition_id(), mergeBuckets)): each task's partial goes to one of mergeBuckets (32) groups merged in parallel, and bloom_filter_merge ORs the group filters in one small task. Both levels use the same bits and hash count, so the OR is the single-level filter bit for bit (FactBloomFilterSuite); a group with no key gives null and is skipped.

Right-sized, partitioned filters (#653). One bloom_filter_agg is capped by Spark at maxNumItems (4M) and maxNumBits (64M bits): for tens of millions of keys it is saturated and prunes nothing, and raising the caps makes every creation-side map task ship a full-size partial. Beyond maxNumItems / 2 keys the filter is therefore partitioned: h = xxhash64(key) is repartitioned by b = pmod(h, B), one bloom_filter_agg(h) per bucket runs after that shuffle (its distribution is already satisfied, so there is no per-map-task partial), and partitioned_bloom_filter packs the B sub-filters into one value. The probe might_contain_partitioned tests sub-filter pmod(h, B); a key the creation side has always lands in its bucket, so there is no false negative. B = ceil(keys / (maxNumItems / 2)) rounded up to a power of two, so the vectorised probe picks the sub-filter as h & (B - 1) (equal to pmod(h, B)) instead of a long division per row (#659); 8 bits a key, fewer down to 4 to stay within maxTotalBits (512M bits); a filter that would need more is declined. The probe is vectorised and its sub-filters are deserialised once per executor (BloomFilterCache).

Split-block sub-filters (#659). Each sub-filter of a partitioned filter is a split-block bloom filter, the layout of Parquet and Impala: 256-bit blocks of eight 32-bit words, a key setting one bit in each word of the one block its remixed hash picks. A probe reads one 32-byte block -- a cache line -- where Spark's BloomFilterImpl reads its k bits anywhere in a multi-megabyte array. At 1 TB the probe was the filter's cost: q93's store_sales scan probes 2.9G rows, and with Spark's layout spent 27 % more task time than without the filter, more than the 41 GB of shuffle it saved. At 8 bits a key the false-positive rate is about 3 % (Spark's: about 2 %).

Vectorised probe (#664). The vectorised filter probes a partitioned filter a batch at a time (BloomKernels.probe). The probed rows' hashes are remixed in the widest long vectors (eight per op on AVX-512), then each key's block is tested in one 256-bit op: its eight bit masks from one multiply-and-shift by the salts, one compare with the block. There is no branch per key, so consecutive keys' block loads overlap. In JMH on an AVX-512 host it is 2.7 to 4 times the per-key probe, including against a 16 MB filter. On a host with 128-bit vectors only (Graviton 4: NEON, 128-bit SVE) each block is tested as two 4-lane halves with a scalar remix (NEON has no 64-bit lane multiply): 1.5 to 2.2 times the per-key probe there, where the 256-bit species would be emulated and ten times slower. Without vectors the same loop runs as branch-free scalar code. The layout and the results are unchanged.

Why the result is the same. A bloom filter has no false negatives, and a row whose key the creation side does not hold cannot survive the join. False positives only let a non-matching row through to the join, which then drops it.

Placement. The session's last optimizer batch, after Spark's DPP and runtime-filter rules, so it does not duplicate a filter they already placed and can see an existing one. At most spark.sql.optimizer.runtimeFilter.number.threshold filters per query, counting Spark's.

Declined:

• two sides of similar size (the ratio test);
• no evidence of selectivity: distinct counts give a match share above maxSelectivity, or, without them, the creation side is a whole, unfiltered table (#650);
• a broadcastable creation side (a broadcast join needs no extra filter -- this also excludes q75, q78 and q80, which regressed when Spark injected one);
• an application side below the scan-size threshold;
• the preserved side of an outer join;
• a key already carrying a dynamic pruning or bloom filter.

Measured. FactBloomFilterSuite covers the q93 shape (the larger fact gets the filters, results match Spark) and the declined shapes. The 1 TB numbers come with the PR.

Related. BloomProbeExpr now skips rows an earlier conjunct already rejected (#635), so a probe this rule adds costs nothing on rows the scan's other filters drop.

Run-time evidence (#659). Plan-time estimates of the creation side -- a filtered dimension, a join -- are what made q18, q61 and q23b slower with the rule at 1 TB. The filter's build therefore also counts the creation side's actual rows, and yields a null filter when they are not sizeRatio times fewer than the application side's estimated rows; the probe is coalesce(might_contain(...), true), so a declined filter keeps every row. A second run-time limit compares the same count with the creation key's base table: at most runtimeMaxReduced (0.2) of its rows. q18's customer filtered to six birth months keeps half its keys; its filter was built and applied, and the catalog_sales scan stage's task time doubled. No probe goes on a relation the plan reads more than once (q23b's CTE): it would cost the copies' exchange reuse. Off with spark.vecruntime.optimizer.factBloomFilter.runtimeEvidence=false.

Transitive reduction (#659). A small, selectively filtered relation D broadcast-joined above a shuffle join, on a column of one of its inputs C, first gets a bloom filter of its key on C's scan. C is then reduced and is the creation side of a filter onto the shuffle join's larger input. In q93 that is a filter on sr_reason_sk from one reason, then bloom filters onto store_sales. The inner join with D above keeps exactly the rows the filter can keep, so results are unchanged. Spark's own runtime filter leaves it out because the join with D is a broadcast. With run-time evidence, the plan-time size check only requires the application side to be the larger: the estimate of a reduced creation side is its base table's size. Off with spark.vecruntime.optimizer.factBloomFilter.transitive=false.

Shared creation-side exchange (#659). The build subquery reads BloomCreationRef, a reference to the join input that the optimizer leaves alone. The query-stage preparation rule ShareBloomCreationExchange replaces its physical placeholder with that join input's own shuffle exchange, carrying the reference's logical link so the subquery's re-planning keeps it. AQE's stage cache then materialises the exchange once, for the join and for the filter, and the build is a read of a stage the join computes anyway. A reference no shuffle join claims turns its filter into null, which every row passes. Off with spark.vecruntime.optimizer.factBloomFilter.shareExchange=false.

# NarrowBelowJoin: a substring or length of one join side's string computed below the join

spark/src/main/scala/org/apache/spark/sql/vecruntime/NarrowBelowJoin.scala, issue #635.

Shape. A projection directly over a join computes substr(x, pos, len) (also left/right), length(x) or octet_length(x), with pos/len literal and x a string or binary column of the join's smaller side by estimated size.

TPC-DS q23a groups store_sales JOIN item by substr(i_item_desc, 1, 30). Spark's PullOutGroupingExpressions leaves the substring in a projection above the join, so at 1 TB each of ~1.66G joined rows gathered the whole description from the broadcast side and then cut 30 characters of it.

Rewrite. The smaller side becomes Project(columns still needed above or by the join condition, f(x) AS a, side) and the projection above reads a: the join carries the short result, and f runs once per row of that side.

Why the result is the same. f is deterministic and cannot fail, and a join only repeats or drops each side's rows without changing their values. On the null-supplying side of an outer join an unmatched row has x null, and f(null) is null, the value the padded row gets anyway.

Placement. The session's last optimizer batch, after column pruning and filter push-down.

Declined:

• an expression over the larger side (the join already carries that side row for row);
• an expression over both sides;
• a non-narrowing expression (upper keeps the width);
• the right side of a semi or anti join (not in the output).

Measured. NarrowBelowJoinSuite covers the q23a shape, a plain projection with a length and a substring, the null-supplying side of an outer join, the declined shapes and the switch; results are compared with Spark with the plugin off. The 1 TB numbers come with the PR.

# AggregateBelowJoin: the fact aggregated by its join keys before a star join

TPC-DS q4, q11 and q74 sum store_sales (and the catalog and web facts) per customer and year, grouped by seven customer strings and d_year, after joining customer and date_dim. Spark aggregates after the joins, so every fact row is joined and then hashed on the strings: at 1 TB each of q4's fact stages shuffled 6.1 GB (537M rows). Aggregating (ss_customer_sk, ss_sold_date_sk) directly above the scan avoids that.

Shape (logical, in the late optimizer batch):

Aggregate(G, aggs, path)   path: Project / Filter / Join(inner, or left semi with F on the left) down to F

where every aggregate function is a non-distinct, unfiltered sum, count, min or max over columns of F, and every column of F the plan above needs apart from those inputs is a join key on the path (G groups by the other sides). F -- the fact's whole subtree as it is read, its scan with its filters and projections, with no join or aggregate in it -- becomes Aggregate(P, P ++ partials, F) with P those join keys, and the top aggregate combines: sum of partial sums cast to the original type, coalesce(sum(partial counts), 0), min/max of the partial minima and maxima.

The pre-aggregate goes above the fact's filters, never between them and the scan. Spark takes a scan's partition filters, dynamic pruning included, only from the filter directly over the relation: a pre-aggregate below that filter leaves the scan unpruned, the pruning then filters the aggregate's output, and two copies of a CTE pruned to different years become one scan of every partition (#675: q4 at 1 TB read 5.05 G rows instead of 1.97 G).

The result is unchanged: an inner join keeps or repeats an F row by its join keys alone, and every row of a pre-aggregated group has the same keys, so the group meets the same rows each of its rows would; sum, count, min and max distribute over that. A null key forms one group, which the join drops as it would drop its rows. Decimal sums are exact; a floating-point sum can differ in the last digits, as any change in summation order.

Declined: a DISTINCT or filtered aggregate, any other function (avg, stddev, ...), aggregate inputs from more than one side, a column of F grouped by or used above other than as a join key, outer joins, and an F that is already an aggregate. Off with spark.vecruntime.optimizer.aggregateBelowJoin.enabled=false.

Evidence of reduction is optional. With minReduction above 0, the rule fires only when statistics put the pre-aggregate's groups -- the product of its keys' distinct counts, an upper bound -- at most 1 / minReduction of the fact's rows; with requireStatistics, a fact without statistics is declined. Both are off by default (#675). A first measurement at 1 TB made q4, q11 and q74 34-104 % slower without them, and they were added for that; the cause was the lost partition pruning described above, not the keys. With the pruning kept, at 1 TB:

with identical results, and q14a, q24a, q67 and q18 (where the rule does not fire) unchanged. The product of distinct counts ignores how the keys correlate: q4's (customer_sk, sold_date_sk) pairs are far fewer than the product, and the estimate declined exactly the queries the pre-aggregate speeds up most.

Local, and judged at run time (#693). Planned as Spark plans an aggregate, the pre-aggregate was a partial aggregate, an exchange of the fact on its keys and a final one. Where the keys barely repeat that exchange is the whole cost: grouping store_sales by (item, store, date) removes about 1 % of its rows, and q47 and q57 at 1 TB took 23.7 and 11.3 s with the rule against 6.5-7.2 and 4.0-4.1 s without it. The pre-aggregate need not be global: the top aggregate recombines its partials and the joins treat every row of a group alike, so any split of the fact's rows into groups that share their keys gives the same result. So the pre-aggregate is planned as one aggregate per task with no exchange (LocalPreAggregateStrategy, a Complete aggregate with no required distribution), and ours emits and starts over as a partial aggregate does: past its memory budget, and after its first probeRows rows (262,144). A table whose rows were fewer than spark.vecruntime.agg.passThroughRatio (1.5) times its groups shows the task does not reduce, and the rest of the task passes one batch at a time. Statistics can prove a reduction -- the groups' upper bound times that ratio at most the rows -- but never disprove one, so they never decline the rewrite; a proven pre-aggregate is judged only at its budget, so a task whose first rows happen not to repeat a key keeps aggregating. aggregateBelowJoin.local=false restores Spark's two stages.

# CountBelowJoin: a join input that only supplies grouping columns, as distinct rows and a count (#692)

TPC-DS q24a and q24b join customer_address on (upper(ca_country), ca_zip) alone and read only ca_state above the join, so each sales row meets every address of its zip, and the aggregate above folds the pairs back. At 3 TB one task of that join produced 367 M rows from 0.9 M input rows (40.7 s; the stage's median task took 0.44 s and its 125 tasks read about the same bytes each). No input-size skew handling can see it: the skew is in the join's output, not its input. The pairs differ only in which address they came from, which nothing above reads.

Shape (logical, in the late optimizer batch):

Aggregate(G, aggs, path)   path: deterministic Project / Filter / inner equi-Join

where every aggregate function is a non-distinct, unfiltered sum (of a long or a double -- a decimal of up to 8 digits counts, since Spark sums it as a long of its unscaled values), count, min or max, and an input S of a join on the path is a subtree with no join or aggregate, holds no aggregate input, and is too large to broadcast. S becomes

Aggregate(K, K ++ count(*) AS c, RepartitionByExpression(joinKeys, Project(K, S)))

with K the columns of S the plan above reads and its join keys (a key expression such as upper(ca_country) becomes a column of the projection, which the join then compares), and the top aggregate weighs each row by c: sum(x) becomes sum(x * c) cast to the original type, count(...) the sum of c over the rows it would count, min and max stay. Several collapsed inputs multiply their counts.

The result is unchanged: an inner join pairs a row of the other side with each matching row of S, and rows of S equal on K give pairs equal in every column the plan above reads, so it sees each such pair c times either way. A null key forms one group, which the join drops as it drops each of its rows. Integer sums are exact (an overflow wraps, or is an error under ANSI, as before); a floating-point sum can differ in the last digits.

No exchange is added. The join needs S partitioned by its join keys, which are columns of K, so the repartition is the exchange the join would have had and every group is whole within one task. The aggregate is a local one (LocalPreAggregate), with no exchange of its own, and gives up after its first rows when its keys do not repeat -- a side joined on its unique key. An input whose partner would be broadcast is left alone, since it is streamed with no exchange.

With statistics (ANALYZE ... FOR COLUMNS, or a DSv2 source's column statistics) the rule is cheaper. The groups are bounded by the product of the distinct counts of the columns K is computed from, at most the rows:

• a bound above rows / minReduction (default 2) leaves S alone: no rows are spent finding out at run time;
• a bound that proves the reduction turns off the judgement on the first rows, which a task whose first rows do not repeat would otherwise fail;
• a bound small enough to broadcast once collapsed (bound x Spark's row size estimate within spark.sql.autoBroadcastJoinThreshold) plans S's aggregate as usual, by its own keys, and broadcasts it to the join, so the other, larger input is joined where it is read, with no exchange.

Declined: avg and every other function, a sum of a wider decimal (weighing it needs a 38-digit product: at 3 TB q23a's sum(decimal(17,2)) weighted that way left the aggregate to Spark and took 27 % longer), a DISTINCT or filtered aggregate, non-deterministic expressions on the path, outer and semi joins.

Measured at 3 TB without statistics (TPC-DS, Graviton 4 r8gd.12xlarge x 10, two runs each, alternating): over the 102 queries every run completed, 1,407 s on main against 1,296 s with the rule (−7.9 %), with checksums identical to each other and to the published 12-node run. q93, which the rule does not change, failed in one run with the rule after executors were killed for memory (exit 137) and its shuffle fetches failed; main's 10-node runs lost executors the same way in q78, q79 and q94. The rule changes 26 queries, which took 340 s on main and 251 s with it (−26 %): q24a 95.1 -> 54.9 s, q24b 100.7 -> 53.1 s, q11 −14 %, q74 −10 %, q12 and q20 about −1 s. The other rewritten queries move by a few tenths of a second either way, and q64 is about 2.7 s (6 %) slower in both runs. Off with spark.vecruntime.optimizer.countBelowJoin.enabled=false.

# DistinctBelowSemiJoin: a left semi join's right side as its distinct join columns (#840)

spark/src/main/scala/org/apache/spark/sql/vecruntime/DistinctBelowSemiJoin.scala.

A left semi join only asks whether a left row has a match, so duplicate right rows change nothing. Spark rewrites INTERSECT as Distinct(LeftSemi(left, right)) and leaves the right side as it is. TPC-DS q14's cross_items joins item (400 K rows) to catalog_sales x item (8.6 G rows at 10 TB) on (brand, class, category), and the result to web_sales x item (4.3 G rows) on the same keys, which have a few thousand values. At 10 TB the two sort-merge joins and their sorts took 5,900 of q14a's 17,300 task seconds and spilled about 230 GiB.

Shape (logical, in the late optimizer batch): Join(left, right, LeftSemi, condition) whose right is deterministic, not already an aggregate, and too large to broadcast by its size estimate. Rewrite: right becomes Aggregate(K, K, right), with K the columns of right the condition reads -- the equi-join keys and any column a non-equi conjunct compares, so the condition sees the same values.

The aggregate is local ([[LocalPreAggregate]], as for AggregateBelowJoin): one per task, no exchange of its own. The join's exchange of the right side follows it and carries the reduced rows, and a task whose first rows do not repeat (a high-cardinality key) gives it up and passes its rows through.

Off with spark.vecruntime.optimizer.distinctBelowSemiJoin.enabled=false.

Left anti joins (#904). NOT EXISTS, EXCEPT (Distinct(LeftAnti(...))) and NOT IN get the same rewrite: an anti join keeps a left row when no right row satisfies the condition, and the condition reads only the right columns kept. NOT IN's null-aware condition, (a = b) OR isnull(a = b), is no exception: a right side with a null key keeps one null row, so NOT IN stays never-true for it, and an empty right side stays empty. Off with spark.vecruntime.optimizer.distinctBelowAntiJoin.enabled=false.

At 10 TB (#840). All 99 queries ran with both rules on and again without them, same sizing and 11 executors. The rewrite fired in the twelve queries below, each with one or more extra local aggregates in the physical plan. q38 and q87 are not rewritten: their INTERSECT / EXCEPT right side is already a Distinct. Wall time, one run each:

Over the 99 queries: 2,893 s without the rules, 2,618 s with them (-9.5%); spill 1,400 GiB without, 935 GiB with.

# RemoveRedundantGroupKeys: grouping keys a unique broadcast key determines (#635)

An adaptive (runtime) optimizer rule. It runs when AQE re-optimises after a query stage finished, so a broadcast stage below an aggregate is already built and its keys can be checked on the data it actually holds.

When a broadcast build side's join key b holds every value at most once, each joined row carries exactly one build row, so every build-side column is a function of b -- and, in an inner join, of the stream key equal to b. An aggregate that groups by b (or that stream key) and also by build-side columns forms the same groups without them: they are dropped from the grouping keys, and an output that still needs one reads it with first(). TPC-DS q23a/b's frequent_ss_items groups by substr(i_item_desc, 1, 30), i_item_sk, d_date over a join with item broadcast on i_item_sk; the rule leaves (i_item_sk, d_date), which removes the 30-character string from the partial aggregate's hash table and from its shuffle. At SF10 that is 58 % fewer shuffle bytes over q23a, q23b, q39a and q39b, and q23a/q23b 26-30 % faster, with identical results.

Nothing is assumed about the data. The broadcast relation sorts a copy of its key column once, on the driver, and reports uniqueness only for a single integral key (int, long, short, byte, date) on at most 16M rows; a key that repeats, a composite key, or a broadcast Spark planned itself leaves the plan unchanged. The dependency is carried through projections, filters and joins (the build side of an outer join is null-extended as a whole, so its columns stay functions of its key), not through aggregates, unions or windows. Off with spark.vecruntime.optimizer.removeRedundantGroupKeys.enabled=false.