yew1eb opened a new pull request, #59988:
URL: https://github.com/apache/spark/pull/59988

   ### What changes were proposed in this pull request?
   
   This PR adds a new optimizer rule, `RemoveRedundantGenerate`, that removes a 
`Generate` below a DISTINCT (group-only) `Aggregate` when the generated columns 
are not referenced by the aggregate.
   
   For a non-outer `Generate`, repeating an input row once per generated 
element does not change the result of a group-only aggregate, so the `Generate` 
is replaced with a `size(expr) > 0` filter that keeps exactly the rows an empty 
or null collection would have dropped. For an outer `Generate`, which produces 
one row per input row, the `Generate` is removed outright. The rule also looks 
through a single deterministic `Project` between the aggregate and the 
`Generate`, which is the shape produced for `SELECT DISTINCT ... LATERAL VIEW 
explode(...)`.
   
   The rewrite is intentionally limited to the case where generated columns are 
unused. Exploding is not injective: `(1, [1, 2])` and `(1, [1, 3])` both 
produce `(1, 1)`, so pushing a DISTINCT on generated columns below the 
`Generate` would change the result. Aggregates with aggregate functions are 
excluded for the same reason. The rule is registered in 
`Optimizer.defaultBatches` and `RuleIdCollection`, so it can be excluded with 
`spark.sql.optimizer.excludedRules`.
   
   ### Why are the changes needed?
   
   `SELECT DISTINCT ... FROM t LATERAL VIEW explode(...)` explodes every input 
row into N * K rows before the aggregate even when the exploded value is not 
selected, e.g.:
   
   ```sql
   SELECT DISTINCT id FROM t LATERAL VIEW explode(items) v AS item
   ```
   
   The optimized plan keeps the `Generate` under the aggregate, so the shuffle 
and aggregation run on the exploded row count. With this rule the plan becomes 
`Aggregate(id, Filter(size(items) > 0, t))` and the exploded rows are never 
materialized.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No. Query results are unchanged; only the optimized plan changes. The rule 
can be excluded through `spark.sql.optimizer.excludedRules`.
   
   ### How was this patch tested?
   
   New tests:
   
   - `RemoveRedundantGenerateSuite` (catalyst, `PlanTest`): positive cases for 
array/map explode, outer explode, aliased and foldable grouping keys, a 
deterministic explode input, and a deterministic projection between the 
aggregate and the `Generate`; negative cases for generated columns in the 
grouping keys or the projection, aggregate functions, non-deterministic 
grouping expressions or projections, and generators that are not `ExplodeBase`.
   - `RemoveRedundantGenerateQuerySuite` (sql/core, `QueryTest`): result 
equivalence with the rule enabled and disabled, covering empty arrays, null 
arrays, duplicate rows, maps and outer explode, plus checks that the rule does 
not fire when the generated column is selected or an aggregate function is used.
   
   Commands:
   
   ```
   build/sbt "catalyst/testOnly *RemoveRedundantGenerateSuite"
   build/sbt "sql/testOnly *RemoveRedundantGenerateQuerySuite"
   build/sbt "catalyst/testOnly *GenerateOptimizationSuite *ColumnPruningSuite 
*RemoveRedundantAggregatesSuite"
   ```
   
   ### Was this patch authored or co-authored using generative AI tooling?
   
   Generated-by: DeepSeek Harness (deepseek-flash)
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to