[VL] Rewrite selfjoin inequality to aggregate

[hhr293]

What changes are proposed in this pull request?

Adds a Catalyst optimizer rule RewriteSelfJoinInequalityToAggregate for the Velox backend.

When the downstream context only cares about existence / membership semantics, the rule rewrites a class of inequality self-joins into GROUP BY + HAVING COUNT(DISTINCT ...) > 1.

Supported plan patterns

• Pattern A' — InSubquery / Exists whose subquery contains a top-level inner self-join
• Pattern A2 — nested: an outer InnerJoin with a self-join on one side; only the self-join subtree is replaced
• Pattern A — a LeftSemi / LeftAnti already present in the input plan whose right child is an inner self-join

Benchmark Performance

On TPC-DS SF=300 (3 executors x 4 cores, Intel Xeon GNR), the observed
q95 result shows there is roughly up to ~150x reduction in the affected build-side rows/data volume and a ~69% reduction in q95 wall time.

The other 107 TPC-DS queries showed no material regression in the same test.

How was this patch tested?

Tests added in gluten

19 Scala tests in RewriteSelfJoinInequalityToAggregateSuite pass, covering:

• Positive — EXISTS / IN with direct self-join (Pattern A')
• Positive — nested InSubquery with outer InnerJoin on self-join child (Pattern A2)
• Positive — correlated EXISTS / IN on a safely remappable equi-key
• NULL / 3VL — rows with only-NULL or single-non-null inequality column excluded
• Fail-closed — plain top-level InnerJoin, LeftOuter, IS DISTINCT FROM, IsNotNull on non-join column, multi-column inequality, equi/neq column overlap
• Fail-closed — correlated predicate on the inequality column (s2.v) or non-key column (s2.w)
• Fail-closed — Aggregate(first(v)), Window(ROW_NUMBER()), uncorrelated rand() filter inside subquery
• Config gate — rule disabled when spark.gluten.sql.rewrite.selfJoinInequality=false

Local verification

A differential SQL harness was also used during development, comparing rewrite ON vs OFF over 50 targeted cases (NULL / 3VL, IN / NOT IN / EXISTS, multi-key joins, aliases, ExprId remap, unsupported predicates and data types, correlated subqueries, repeatability guards). All tests passed.

Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.7)

[Copilot]

Pull request overview

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a new Catalyst optimizer rule to rewrite inequality self-joins under existence-only semantics into a GROUP BY ... HAVING COUNT(DISTINCT ...) > 1 form to reduce intermediate row explosion and improve performance (e.g., TPC-DS q95).

Changes:

• Introduces RewriteSelfJoinInequalityToAggregate optimizer rule and injects it into Velox rule API.
• Adds an opt-in config flag spark.gluten.sql.rewrite.selfJoinInequality (default false).
• Adds a dedicated Scala test suite covering positive, negative, config-gated, and NULL/3VL parity cases.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 4 comments.

File	Description
backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala	Implements the rewrite rule for multiple existence-only patterns (A, A’, A2).
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala	Adds config plumbing + doc string for gating the rewrite.
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala	Registers the new optimizer rule in the Velox injector.
backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala	Adds unit tests validating when the rule should/shouldn’t fire and semantic parity.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Suppressed comments (4)

backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala:104

• Using logInfo for optimizer-rule rewrites is likely too noisy in production (other rules in this module generally don’t log at INFO). Consider downgrading this message (and the other logInfo calls in this rule) to logDebug, or guarding with log.isDebugEnabled to avoid flooding driver logs when the rule fires frequently.

    if (!(rewritten eq plan)) {
      logInfo("RewriteSelfJoinInequalityToAggregate: rewrote self-join to " +
        "GROUP BY + HAVING COUNT(DISTINCT) > 1")
    }

backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala:322

• Same issue as above: building sjLeftEquiByName with .toMap can silently drop entries if equi-key names are duplicated (e.g., via upstream projection aliases), leading to incorrect remapping. Add a distinct-name guard (fail-closed) before constructing the map (or use ExprId-based mapping).

    val sjLeftEquiByName: Map[String, Attribute] = sjLeftEquiAttrs.map(a => a.name -> a).toMap

backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala:510

• nameToInnerLeft is keyed by column name and built via .toMap, which drops duplicates. If the inner equi-keys have duplicate names (possible when the self-join inputs are projections with repeated aliases), wrapper alias remapping can become ambiguous and incorrect. Add a distinct-name guard on innerLeftEquiAttrs before creating this map (or use ExprId-based mapping only).

    val nameToInnerLeft: Map[String, Attribute] =
      innerLeftEquiAttrs.map(a => a.name -> a).toMap

backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala:104

• The test named "Pattern A': EXISTS subquery with bare self-join" isn’t actually exercising a self-join inside the subquery (it’s a correlated EXISTS over a single scan). That leaves the Exists + self-join path untested. Consider adjusting this test to use an EXISTS whose subquery body contains the self-join, so it actually covers the intended pattern.

  test("Pattern A': EXISTS subquery with bare self-join produces equivalent results") {
    setupTable()
    val sql =
      """SELECT k FROM T ws1 WHERE EXISTS (
        |  SELECT 1 FROM T s WHERE s.k = ws1.k AND s.v <> ws1.v)""".stripMargin
    val (on, off) = runBoth(sql)
    assert(on == off, s"rewrite ON $on != OFF $off")
    // Ground truth: only k in {1,3,6} have >=2 non-null distinct v.
    assert(on == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got $on")

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

Suppressed comments (4)

backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala:95

• This test name says it covers a “bare self-join”, but the SQL only scans T once and uses a correlated inequality (s.v <> ws1.v), so it doesn’t exercise the self-join rewrite path. Either rename the test to reflect what it actually covers, or change the SQL to include a self-join if the intent is to validate Pattern A' for EXISTS.

  test("Pattern A': EXISTS subquery with bare self-join produces equivalent results") {

backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala:300

• Pattern A' currently requires a wrapping Project whose projectList entries are equi-key attributes/aliases (via canonicalizeWrapper). That means common EXISTS forms like EXISTS (SELECT 1 FROM t1 JOIN t2 ...) (Project of a literal) will be rejected here, even though the config/docs and scaladoc claim support for Exists. Consider either (a) relaxing the wrapper handling for EXISTS when the Project doesn’t reference join outputs, or (b) clarifying in docs/scaladoc which EXISTS shapes are actually supported (e.g., only after subquery rewrite into LeftSemi/LeftAnti).

    projectListOpt match {
      case None =>
        None
      case Some(pl) =>
        canonicalizeWrapper(pl, equiPairs, filtered).map {

docs/velox-configuration.md:87

• This config description states the rewrite applies to Exists, but the current implementation’s Pattern A' path requires the subquery to project equi-key attributes (so a typical EXISTS (SELECT 1 ...) shape won’t match). Suggest clarifying the docs here to avoid over-promising (or update the rule to support the common EXISTS projection shape).

| spark.gluten.sql.rewrite.selfJoinInequality                                      | 🔄 Dynamic    | false             | When true, rewrite self-join with inequality predicate into GROUP BY + HAVING COUNT(DISTINCT) > 1 under existence semantics (InSubquery, Exists, LeftSemi, LeftAnti). Opt-in default (false) until the rewrite has been exercised more broadly across workloads.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |

backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:861

• The config docstring here lists Exists as a supported context. Given Pattern A' currently requires the subquery to project equi keys (so EXISTS (SELECT 1 ...) is typically rejected), it’d be better to either clarify the supported EXISTS shapes in this docstring or drop Exists until it’s fully supported.

      .doc("When true, rewrite self-join with inequality predicate into" +
        " GROUP BY + HAVING COUNT(DISTINCT) > 1 under existence semantics" +
        " (InSubquery, Exists, LeftSemi, LeftAnti). Opt-in default (false)" +
        " until the rewrite has been exercised more broadly across workloads.")

[hhr293]

The CI job failed while downloading Maven from Maven Central with:
HTTP 429 Too Many Requests so it could not download apache-maven-3.9.16-bin.tar.gz, which looks like an external dependency download / CI infrastructure issue and not related to this patch.