hhr293 opened a new pull request, #58424: URL: https://github.com/apache/spark/pull/58424
## What changes were proposed in this pull request? Add an opt-in optimizer rule, `RewriteSelfJoinInequalityToAggregate`, that rewrites supported inequality self-joins inside uncorrelated `IN` subqueries into: ```sql GROUP BY equi_keys HAVING COUNT(DISTINCT inequality_col) > 1 ``` over a single copy of the relation, eliminating the self-join cross-product for patterns such as TPC-DS Q95. The rule targets two `InSubquery` shapes: * **Pattern A'**: the subquery's top-level `InnerJoin` is a direct self-join. * **Pattern A2**: the subquery contains an outer `InnerJoin` with a self-join child; only the self-join child is replaced while the outer join is preserved. The rule runs in `extendedOperatorOptimizationRules`, before `RewritePredicateSubquery` converts the predicate subquery into a semi/anti/existence join, so it operates on the uncorrelated `InSubquery` representation. The rewrite is deliberately fail-closed. It only fires when: * both self-join sides are proven to represent the same repeatable relation; * the relevant operators, expressions, and leaf source are considered repeatable; * the equi keys and inequality column use data types considered safe for replacing comparison equality with grouping / `DISTINCT` equality. Floating-point types, complex types, non-binary collated strings, and `CHAR` / `VARCHAR` keys are conservatively rejected. Correlated `InSubquery` expressions are left unchanged because the ExprId remapping performed by this rule does not rewrite correlated predicates. The rewrite also inserts `IsNotNull` filters on the equi keys before aggregation. This preserves the original equi-join's SQL three-valued NULL semantics: a normal `=` join does not match NULL keys, whereas `GROUP BY` would otherwise form a NULL group that could leak NULL into the subquery result and change `IN` / `NOT IN` behavior. The rule is controlled by: ```text spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled ``` It is `internal()` and disabled by default. ## Why are the changes needed? An inequality self-join inside an existence-style subquery, such as TPC-DS Q95, may generate a quadratic number of row pairs for each repeated equi key, even though the query only needs to determine whether at least two distinct inequality-column values exist. For a key with `n` matching rows, the self-join may examine up to `O(n^2)` row pairs. The equivalent condition can instead be evaluated using: ```sql GROUP BY equi_key HAVING COUNT(DISTINCT inequality_col) > 1 ``` which scans one copy of the relation and avoids materializing the self-join cross-product. The benefit is therefore largest when the input has sufficiently high per-key multiplicity and the self-join produces significant row expansion. ## Does this PR introduce any user-facing change? No. The rule is internal and disabled by default. Existing behavior is unchanged unless a user explicitly enables: ```text spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled=true ``` When enabled, the rewrite is only applied to supported shapes that satisfy the rule's correctness guards. Unsupported cases are left unchanged. ## How was this patch tested? Added `RewriteSelfJoinInequalityToAggregateSuite` with 31 tests covering: * positive rewrites for Pattern A' and Pattern A2; * Pattern A2 with the self-join on either side of the outer join; * multi-equi-key tuple `IN` and right-side key remapping; * structural equivalence with the corresponding aggregate plan; * configuration gating; * NULL / three-valued logic preservation for both equi keys and inequality columns; * `NOT IN` NULL semantics; * relation identity and repeatability guards; * correlated subqueries; * swapped aliases and output-position identity; * different relations with identical schemas; * `Aggregate`, `Window`, nondeterministic expressions, and `LogicalRDD` sources; * unsupported deterministic expressions; * `IS DISTINCT FROM`; * multiple or overlapping inequality columns; * unsupported join types; * floating-point and complex data types; * non-binary string collations; * `CHAR` / `VARCHAR` metadata handling; * ANSI runtime error-behavior parity. The targeted tests were run with: ```bash build/sbt 'sql/testOnly org.apache.spark.sql.execution.RewriteSelfJoinInequalityToAggregateSuite' build/sbt 'sql/scalastyle' ``` ### TPC-DS Q95 benchmark I also ran an end-to-end TPC-DS Q95 benchmark comparing the rule disabled (baseline self-join) and enabled, using the same Spark build for both runs and verifying that the query results were identical. One warm-up run per configuration was discarded, and the median of 3 measured runs was compared. Enabling the rule reduced Q95 runtime by roughly **46%** (about a **1.87x** speedup). Environment: * Vanilla Apache Spark JVM execution * No Gluten / Velox * `local[12]` * TPC-DS SF300 * Median of 3 measured runs ## Was this patch authored or co-authored using generative AI tooling? Generated-by: Claude (Opus 4.8) -- 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]
