sunchao commented on PR #58424: URL: https://github.com/apache/spark/pull/58424#issuecomment-5547619531
**I think this optimization is worth having in Spark: it recognizes when an expensive join is only being used to answer a simple yes/no question, and replaces it with a much cheaper calculation.** The technical idea is sound. The main tradeoff is maintaining substantial correctness machinery for a fairly narrow set of queries. I rechecked the PR: its head is still `15aa88d1`, the commit we reviewed. Our review found no actionable correctness defect. It remains a draft, and the author has announced additional fixes and test improvements before marking it ready. [Author’s update](https://github.com/apache/spark/pull/58424#issuecomment-5543195034) Consider the question: **“Which orders contain items from more than one warehouse?”** One way to answer it is to compare the sales table with itself. For each order, find two rows whose warehouses differ: ```sql SELECT a.order_id FROM sales a JOIN sales b ON a.order_id = b.order_id AND a.warehouse_id <> b.warehouse_id ``` If one order contains 50 rows from warehouse A and 50 from warehouse B, that join has **5,000 matching ordered pairs**. But when this result is used inside `IN (...)`, all those pairs communicate the same fact: that order qualifies. The optimization instead calculates: ```sql SELECT order_id FROM sales WHERE order_id IS NOT NULL GROUP BY order_id HAVING COUNT(DISTINCT warehouse_id) > 1 ``` It processes the group’s 100 input rows and its distinct warehouse values without generating those pairs. That is a meaningful algorithmic improvement. The restriction to membership is essential. If the surrounding query actually needs the pairs, their counts, or columns from both rows, this replacement would change the answer. It also must preserve duplicate rows in the query **outside** the `IN`. The implementation handles this carefully: 1. **Find a supported membership subquery.** It runs before Spark converts `IN` into join operators, while that context remains explicit. It supports a direct self-join and a self-join nested inside another inner join. 2. **Prove that the two inputs are interchangeable.** Matching table names is insufficient. The rule checks equivalent plans, corresponding column positions, deterministic expressions, and supported sources and operators. 3. **Build the aggregate while preserving SQL semantics.** Null grouping keys are removed because ordinary equality joins never match them. `COUNT(DISTINCT)` already ignores null warehouse values, matching the behavior of `<>`. 4. **Repair column references.** Removing one side of a join changes which columns exist. The rule rebuilds projections and updates Spark’s internal column identities, including references from the surrounding join. Most of the implementation’s complexity comes from proving and preserving these conditions. The actual aggregate construction is small. [Implementation](https://github.com/apache/spark/blob/15aa88d1f7567c2219aeacf0c0e2e194f167f947/sql/core/src/main/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregate.scala) The important tradeoffs are: | Choice | Benefit | Cost or limitation | |---|---|---| | Replace pair generation with aggregation | Avoids potentially quadratic join work and removes one copy of the self-join input | Distinct aggregation can still shuffle, consume memory, and spill | | Restrict the rewrite to membership queries | Makes collapsing duplicate pairs valid | Excludes many ordinary inequality joins | | Explicit source, operator, expression, and type allowlists | Unproven cases keep their existing plans | Many potentially valid queries receive no benefit; the allowlists need maintenance | | Preserve surrounding joins and projections | Covers more useful queries, including the targeted Q95 shapes | Requires considerable column-reference bookkeeping | | Disable the rule by default | Allows controlled evaluation | Merging the PR alone gives existing users no automatic speedup | One further performance consideration: **`COUNT(DISTINCT)` calculates more information than the predicate needs.** Once two different values exist, the answer is already yes, but the implementation uses Spark’s general distinct-count machinery. A future implementation could investigate a bounded “two distinct values” aggregate, or an extrema-based formulation where ordering and equality semantics permit it. That is an improvement opportunity; the current approach benefits from reusing established execution operators. The reported performance result is encouraging: the author measured **about 46% lower runtime, or 1.87× speedup, for TPC-DS Q95**, using vanilla JVM Spark, SF300, `local[12]`, and the median of three measured runs after warm-up. Results were checked for equality. This supports substantial benefit for the target query, although I did not independently reproduce that benchmark. [Benchmark details](https://github.com/apache/spark/pull/58424) For a stronger performance case, I would want measurements across different group sizes and distinct-value counts, including small groups where aggregation’s overhead matters. Shuffle volume, spill, and intermediate row counts would help explain the gain and its limits. The current evidence supports a targeted optimization; it does not establish a general workload speedup. The correctness evidence is reasonably strong for this draft: - Our earlier review covered five independent scopes and found no actionable correctness defect. - The new suite’s **31 tests passed in CI** at the reviewed commit. [CI run](https://github.com/hhr293/spark/actions/runs/33352322392) - Those 31 tests also passed in our adapted local setup, and additional checks supported preservation of tuple membership and duplicate outer rows. - The local setup used cached Spark classes with compatibility adaptations. We did **not** complete a clean native build of the PR. There are still worthwhile refinements. Several test helpers compare sets, which can conceal duplicate-count regressions; those assertions should preserve multiplicity. Comments promising broad ANSI error parity also need care: our probing found an evaluation difference consistent with existing filter-pushdown behavior, rather than a demonstrated query-result defect. The announced Gluten follow-ups include preserving alias metadata and adding focused tests that prevent accidental changes to the number and ordering of subquery output columns. Those are sensible improvements to incorporate before reviewing the final Spark commit. [Metadata review](https://github.com/apache/gluten/pull/12756#discussion_r3930589969), [output-shape tests](https://github.com/apache/gluten/pull/12756#discussion_r3930589952) **Its immediate value for our production workloads is less established.** In the production samples we inspected, we did not confirm an additional business workload that satisfies all the current conditions. Several apparent matches needed actual row pairs or compared different inputs; some workloads already used aggregation to answer the same underlying question. That investigation was substantial, but it was not an exhaustive census. The source restriction is especially consequential for us: the implementation accepts the exact stock Parquet file-format class. Normal Delta scans use a different class and are therefore excluded—even when their SQL has the right shape. Supporting Delta would require a separate correctness assessment; simply enabling this flag would not unlock those workloads. [Source restriction](https://github.com/apache/spark/blob/15aa88d1f7567c2219aeacf0c0e2e194f167f947/sql/core/src/main/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregate.scala#L659-L685) My recommendation is to **support the upstream direction, retain the opt-in default, and review the author’s remaining changes before treating the draft as finished**. I would prioritize stronger multiplicity tests and reproducible performance evidence over expanding the rule’s scope immediately. For our fork, I would first identify a recurring eligible workload before investing in a port. **Why it deserves a place in Spark:** a query pattern can be narrow yet disproportionately expensive. This rule lets Spark recognize the underlying question, avoid unnecessary pair generation, and improve existing SQL without requiring every application author to discover and implement the rewrite themselves. The reported Q95 improvement shows that this can have meaningful value. Its strongest justification is the size of the avoidable work when it applies; broad production adoption remains to be demonstrated. -- 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]
