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]
