Copilot commented on code in PR #12756: URL: https://github.com/apache/gluten/pull/12756#discussion_r3901980939
########## backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,734 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.config.VeloxConfig + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, IntegerType, LongType, ShortType, TimestampType} + +/** + * Rewrites self-join with inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery 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 with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.gluten.sql.rewrite.selfJoinInequality` (default false, opt-in). + */ +case class RewriteSelfJoinInequalityToAggregate(spark: SparkSession) + extends Rule[LogicalPlan] + with PredicateHelper + with Logging { + + private val CountDistinctAliasName = "_gluten_rw_selfjoin_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!VeloxConfig.get.enableRewriteSelfJoinInequality) { + logDebug("RewriteSelfJoinInequalityToAggregate: disabled via config, skipping") + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressions { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) + case _ => None + } + if (mapped.exists(_.isEmpty)) { + None + } else { + val newProjectList = mapped.flatten + val newWrapper = Project(newProjectList, newChild) + val newOutput = newWrapper.output + val remap: Map[ExprId, Attribute] = + oldOutput.zip(newOutput).map { case (o, n) => o.exprId -> n }.toMap + Some((newWrapper, remap)) + } + } + + /** + * Replace equi-key attribute references inside a NamedExpression according to `remap`, while + * preserving the NamedExpression shape. + * + * `Expression.transformUp` returns `Expression`, not `NamedExpression`. We avoid a blanket + * `asInstanceOf[NamedExpression]` by handling the two shapes that can appear in a Project's + * `projectList` explicitly: a bare Attribute (whose top-level may itself be replaced) and an + * Alias (which stays an Alias while its child is transformed). Any other NamedExpression shape we + * do not rewrite is left as-is ONLY if it does not reference a replaced self-join output; + * otherwise it would carry a stale ExprId, so returns None to fail the whole rewrite closed. + */ + private def remapNamedExpressionAttributes( + ne: NamedExpression, + remap: Map[ExprId, Attribute]): Option[NamedExpression] = ne match { + case a: Attribute if remap.contains(a.exprId) => Some(remap(a.exprId)) + case a: Attribute => Some(a) + case al: Alias => + val newChild = al.child.transformUp { + case a: Attribute if remap.contains(a.exprId) => remap(a.exprId) + } + Some( + if (newChild eq al.child) { + al + } else { + Alias(newChild, al.name)( + al.exprId, + al.qualifier, + al.explicitMetadata, + al.nonInheritableMetadataKeys) + }) + case other if other.references.exists(a => remap.contains(a.exprId)) => + // Fail-closed: a NamedExpression we do not rewrite (neither a bare Attribute nor an Alias) + // that still references a replaced self-join output would be left with a dangling ExprId. + // Refuse the rewrite rather than emit a plan with a stale reference. + None + case other => Some(other) + } + + // ============================================================================ + // Pattern A' / A2 dispatch (subquery plans of InSubquery) + // ============================================================================ + + private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = { + // Candidate-level nondeterminism guard: reject if ANY node in the whole subquery plan + // is non-repeatable (Rand, LIMIT-without-ORDER-BY, Sample, Offset, streaming). This + // catches nondeterminism that has been hoisted above the self-join by an earlier + // optimizer rule -- the per-side `isSameBaseRelation` check alone would miss it because + // both innerLeft/innerRight can look deterministic after such a hoist. + if (!isRepeatablePlan(plan)) return None + + val (projectListOpt, innerJoin): (Option[Seq[NamedExpression]], Join) = plan match { + case Project(pl, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(pl), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + if (isSameBaseRelation(innerJoin.left, innerJoin.right)) { + rewriteDirectSelfJoin(projectListOpt, innerJoin) + } else { + rewriteNestedSelfJoin(projectListOpt, innerJoin) + } + } Review Comment: `rewriteSubqueryPlan` already rejects non-repeatable candidates via `isRepeatablePlan(plan)`, but `isSameBaseRelation` then re-walks both self-join sides with `isRepeatablePlan(left/right)` (and this happens multiple times on the nested path as well). This is a concrete extra traversal cost in the optimizer. Consider splitting `isSameBaseRelation` into (a) structural equality (`canonicalized`) and (b) repeatability, and in the subquery rewrite path only pay the repeatability walk once (or memoize `isRepeatablePlan` per subtree for the duration of `rewriteSubqueryPlan`). ########## backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,734 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.config.VeloxConfig + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, IntegerType, LongType, ShortType, TimestampType} + +/** + * Rewrites self-join with inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery 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 with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.gluten.sql.rewrite.selfJoinInequality` (default false, opt-in). + */ +case class RewriteSelfJoinInequalityToAggregate(spark: SparkSession) + extends Rule[LogicalPlan] + with PredicateHelper + with Logging { + + private val CountDistinctAliasName = "_gluten_rw_selfjoin_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!VeloxConfig.get.enableRewriteSelfJoinInequality) { + logDebug("RewriteSelfJoinInequalityToAggregate: disabled via config, skipping") + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressions { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) + case _ => None + } + if (mapped.exists(_.isEmpty)) { + None + } else { + val newProjectList = mapped.flatten + val newWrapper = Project(newProjectList, newChild) + val newOutput = newWrapper.output + val remap: Map[ExprId, Attribute] = + oldOutput.zip(newOutput).map { case (o, n) => o.exprId -> n }.toMap + Some((newWrapper, remap)) + } + } + + /** + * Replace equi-key attribute references inside a NamedExpression according to `remap`, while + * preserving the NamedExpression shape. + * + * `Expression.transformUp` returns `Expression`, not `NamedExpression`. We avoid a blanket + * `asInstanceOf[NamedExpression]` by handling the two shapes that can appear in a Project's + * `projectList` explicitly: a bare Attribute (whose top-level may itself be replaced) and an + * Alias (which stays an Alias while its child is transformed). Any other NamedExpression shape we + * do not rewrite is left as-is ONLY if it does not reference a replaced self-join output; + * otherwise it would carry a stale ExprId, so returns None to fail the whole rewrite closed. + */ + private def remapNamedExpressionAttributes( + ne: NamedExpression, + remap: Map[ExprId, Attribute]): Option[NamedExpression] = ne match { + case a: Attribute if remap.contains(a.exprId) => Some(remap(a.exprId)) + case a: Attribute => Some(a) + case al: Alias => + val newChild = al.child.transformUp { + case a: Attribute if remap.contains(a.exprId) => remap(a.exprId) + } + Some( + if (newChild eq al.child) { + al + } else { + Alias(newChild, al.name)( + al.exprId, + al.qualifier, + al.explicitMetadata, + al.nonInheritableMetadataKeys) + }) + case other if other.references.exists(a => remap.contains(a.exprId)) => + // Fail-closed: a NamedExpression we do not rewrite (neither a bare Attribute nor an Alias) + // that still references a replaced self-join output would be left with a dangling ExprId. + // Refuse the rewrite rather than emit a plan with a stale reference. + None + case other => Some(other) + } + + // ============================================================================ + // Pattern A' / A2 dispatch (subquery plans of InSubquery) + // ============================================================================ + + private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = { + // Candidate-level nondeterminism guard: reject if ANY node in the whole subquery plan + // is non-repeatable (Rand, LIMIT-without-ORDER-BY, Sample, Offset, streaming). This + // catches nondeterminism that has been hoisted above the self-join by an earlier + // optimizer rule -- the per-side `isSameBaseRelation` check alone would miss it because + // both innerLeft/innerRight can look deterministic after such a hoist. + if (!isRepeatablePlan(plan)) return None + + val (projectListOpt, innerJoin): (Option[Seq[NamedExpression]], Join) = plan match { + case Project(pl, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(pl), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + if (isSameBaseRelation(innerJoin.left, innerJoin.right)) { + rewriteDirectSelfJoin(projectListOpt, innerJoin) + } else { + rewriteNestedSelfJoin(projectListOpt, innerJoin) + } + } + + // ============================================================================ + // Pattern A' : direct self-join at subquery top level + // ============================================================================ + + private def rewriteDirectSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + innerJoin: Join): Option[LogicalPlan] = { + // Fail closed on an explicit join hint. This rule deletes the self-join outright; a user + // `/*+ BROADCAST(...) */` (or SHUFFLE_*/MERGE) is an explicit optimizer directive about THAT + // join, and an opt-in rewrite has no business silently discarding it. Not a correctness bug + // (hints do not change results), but the conservative maintainer choice. `JoinHint.NONE` exists + // across all Spark versions Gluten supports, so no shim is needed. + if (innerJoin.hint != JoinHint.NONE) return None + + val innerLeft = innerJoin.left + val innerRight = innerJoin.right + val innerCond = innerJoin.condition.get + + val parsed = parseSelfJoinCondition(innerCond, innerLeft, innerRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val innerLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val innerLeftNeqAttr: Attribute = neqPairs.head._1 + val filtered = buildAggregateHavingDistinctGt1(innerLeftEquiAttrs, innerLeftNeqAttr, innerLeft) + + // Fail-closed on bare-Join subqueries: without a wrapping Project the subquery output + // is the full self-join output (both sides' columns). Replacing that with + // `Project(equiKeys, filtered)` shrinks the output; if the enclosing InSubquery + // referenced a non-equi column by position, `values.zip(sub.output).map(EqualTo.tupled)` + // inside RewritePredicateSubquery would build an incorrect semi condition. Q95's + // subqueries all have an explicit Project wrapper, so this branch does not affect it. + projectListOpt match { + case None => + None + case Some(pl) => + canonicalizeWrapper(pl, equiPairs, filtered).map { + case (newWrapper, _) => + logDebug( + s"Pattern A' - equiKeys=[${innerLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${innerLeftNeqAttr.name}" + + s", outCols=[${newWrapper.projectList.map(_.name).mkString(",")}]") + newWrapper + } + } + } + + // ============================================================================ + // Pattern A2 : self-join nested inside another InnerJoin in the subquery + // ============================================================================ + + private def rewriteNestedSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + outerJoin: Join): Option[LogicalPlan] = { + val outerCond = outerJoin.condition.get + + val (selfJoinSide, selfJoinOnRight) = + tryExtractSelfJoin(outerJoin.right) match { + case Some(_) => (outerJoin.right, true) + case None => + tryExtractSelfJoin(outerJoin.left) match { + case Some(_) => (outerJoin.left, false) + case None => return None + } + } + + val (selfJoinProjectOpt, selfJoin) = selfJoinSide match { + case p @ Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(p), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + val sjLeft = selfJoin.left + val sjRight = selfJoin.right + val sjCond = selfJoin.condition.get + if (!isSameBaseRelation(sjLeft, sjRight)) return None + + val parsed = parseSelfJoinCondition(sjCond, sjLeft, sjRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val sjLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val sjLeftNeqAttr: Attribute = neqPairs.head._1 + + val selfJoinOutputSet = selfJoinSide.outputSet + val sjEquiExprIds: Set[ExprId] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId, r.exprId) }.toSet + // wrapper Project may reproject equi-keys under fresh alias exprIds; include those. + val wrapperEquiExprIds: Set[ExprId] = selfJoinProjectOpt.toSeq.flatMap { + p => + p.projectList.flatMap { + case a: Attribute if sjEquiExprIds.contains(a.exprId) => Some(a.exprId) + case al @ Alias(a: Attribute, _) if sjEquiExprIds.contains(a.exprId) => Some(al.exprId) + case _ => None + } + }.toSet + val allEquiExprIds = sjEquiExprIds ++ wrapperEquiExprIds + + // Outer join condition may reference only equi-key attrs from the self-join side. + val outerCondRefs = outerCond.references.filter(selfJoinOutputSet.contains) + if (!outerCondRefs.forall(a => allEquiExprIds.contains(a.exprId))) return None + + // Top-level subquery Project may reference only equi-key attrs from the self-join side. + val projectOk = projectListOpt.forall { + pl => + val refs = pl.flatMap(_.references).filter(selfJoinOutputSet.contains) + refs.forall(a => allEquiExprIds.contains(a.exprId)) + } + if (!projectOk) return None + + val filtered = buildAggregateHavingDistinctGt1(sjLeftEquiAttrs, sjLeftNeqAttr, sjLeft) + + val (newSelfJoinSide, outputRemap): (LogicalPlan, Map[ExprId, Attribute]) = + selfJoinProjectOpt match { + case Some(wp) => + canonicalizeWrapper(wp.projectList, equiPairs, filtered) match { + case Some((newWrapper, remap)) => (newWrapper, remap) + case None => return None + } + case None if projectListOpt.isEmpty => + // Fail-closed: with neither a wrapper Project around the self-join nor a top-level + // subquery Project, the outer join currently exposes every self-join column, and + // replacing the self-join with `Project(equiKeys, filtered)` would shrink the outer + // join's right-hand output arity. RewritePredicateSubquery's positional zip + // (`values.zip(sub.output).map(EqualTo.tupled)`) would then bind semi predicates to + // the wrong attributes -- silently dropping components of a tuple IN. A + // top-level Project (`projectListOpt`) is what would let the arity be preserved + // by the top-level rewrite loop; without one, refuse to rewrite. + return None + case None => + // No wrapper Project but there IS a top-level subquery Project: shrinking the outer + // join's self-join-side output is safe because the top-level Project is rewritten + // consistently via `outputRemap` below and the top-level rewrite loop ensures + // subquery output arity matches what the enclosing InSubquery expects. + // Outer references may point at sjRight equi-attributes; remap them to sjLeft + // (same output position in a valid self-join). + val newP = Project(sjLeftEquiAttrs, filtered) + val remap: Map[ExprId, Attribute] = + equiPairs.map { case (l, r) => r.exprId -> l }.toMap + (newP, remap) + } + + // Rewrite outer join condition to use new wrapper output attributes. + val newOuterCond = outerCond.transformUp { + case a: Attribute if outputRemap.contains(a.exprId) => outputRemap(a.exprId) + } + + val newOuterJoin = if (selfJoinOnRight) { + outerJoin.copy(right = newSelfJoinSide, condition = Some(newOuterCond)) + } else { + outerJoin.copy(left = newSelfJoinSide, condition = Some(newOuterCond)) + } + + // Rewrite top-level Project references. + val result = projectListOpt match { + case Some(pl) => + val remapped = pl.map(ne => remapNamedExpressionAttributes(ne, outputRemap)) + if (remapped.exists(_.isEmpty)) return None + Project(remapped.flatten, newOuterJoin) + case None => newOuterJoin + } + + logDebug( + s"Pattern A2 - equiKeys=[${sjLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${sjLeftNeqAttr.name}") + Some(result) + } + + private def tryExtractSelfJoin(plan: LogicalPlan): Option[Join] = { + val join = plan match { + case Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => j + case j: Join if j.joinType == Inner && j.condition.isDefined => j + case _ => return None + } + // A hinted self-join is not an extraction candidate: fail closed so the same node the rewrite + // would delete is never even recognized here. See `rewriteDirectSelfJoin` for the rationale. + if (join.hint != JoinHint.NONE) return None + if (!isSameBaseRelation(join.left, join.right)) return None + val parsed = parseSelfJoinCondition(join.condition.get, join.left, join.right) + if (parsed.isEmpty) return None + Some(join) + } + + // ============================================================================ + // parseSelfJoinCondition + isSameBaseRelation + // ============================================================================ + + private def outputOrdinal(plan: LogicalPlan, attr: Attribute): Int = + plan.output.indexWhere(_.exprId == attr.exprId) + + private def sameOutputPosition( + leftPlan: LogicalPlan, + rightPlan: LogicalPlan, + leftAttr: Attribute, + rightAttr: Attribute): Boolean = { + val leftPos = outputOrdinal(leftPlan, leftAttr) + val rightPos = outputOrdinal(rightPlan, rightAttr) + leftPos >= 0 && rightPos >= 0 && leftPos == rightPos + } + + /** + * Parse a join condition into equi-pairs and inequality-pairs. Accepts only: + * - `EqualTo(attr, attr)` where the two attrs come from opposite sides, + * - `Not(EqualTo(attr, attr))` -- same side rule, + * - `IsNotNull(attr)` where the attr is one of the join columns. + * Anything else in the condition disqualifies the whole rewrite (fail-closed). + */ + private def parseSelfJoinCondition( + condition: Expression, + leftPlan: LogicalPlan, + rightPlan: LogicalPlan) + : Option[(Seq[(Attribute, Attribute)], Seq[(Attribute, Attribute)])] = { + + val leftOutput = leftPlan.outputSet + val rightOutput = rightPlan.outputSet + val predicates = splitConjunctivePredicates(condition) + + val equiPairs = predicates.collect { + case EqualTo(l: Attribute, r: Attribute) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + case EqualTo(r: Attribute, l: Attribute) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + } + + val neqPairs = predicates.collect { + case Not(EqualTo(l: Attribute, r: Attribute)) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + case Not(EqualTo(r: Attribute, l: Attribute)) + if leftOutput.contains(l) && rightOutput.contains(r) => + (l, r) + } + + // Only IsNotNull predicates on join columns are safe to drop -- they're redundant with + // the join semantics or auto-added by InferFiltersFromConstraints. IsNotNull on other + // columns changes semantics if we drop it; bail out. + val joinAttrIds: Set[ExprId] = + (equiPairs ++ neqPairs).flatMap { case (l, r) => Seq(l.exprId, r.exprId) }.toSet + val isNotNullOnJoinCols = predicates.count { + case IsNotNull(a: Attribute) if joinAttrIds.contains(a.exprId) => true + case _ => false + } + + val totalMatched = equiPairs.size + neqPairs.size + isNotNullOnJoinCols + if (totalMatched != predicates.size) return None + + if (equiPairs.isEmpty || neqPairs.isEmpty) return None + + // Only rewrite the single-inequality case. Multiple inequality conjuncts cannot be represented + // by COUNT(DISTINCT) over a single column. + if (neqPairs.size != 1) return None + + // Canonicalization intentionally erases cosmetic Alias names, so name equality cannot prove + // that the two predicate ends refer to the same underlying column. Resolve each end by its own + // ExprId against its child output and require matching output ordinals instead. + val equiValid = equiPairs.forall { + case (l, r) => sameOutputPosition(leftPlan, rightPlan, l, r) + } + val neqValid = neqPairs.forall { + case (l, r) => sameOutputPosition(leftPlan, rightPlan, l, r) + } + if (!equiValid || !neqValid) return None + + // Equi-key output positions must be distinct across pairs. Keep the same positional identity + // here so swapped or duplicate aliases cannot make two different underlying columns look equal. + val leftEquiOrdinals = equiPairs.map { case (l, _) => outputOrdinal(leftPlan, l) } + if (leftEquiOrdinals.exists(_ < 0)) return None + if (leftEquiOrdinals.distinct.size != leftEquiOrdinals.size) return None + + // Defensive: reject when the neq column overlaps an equi-key column + // (e.g. `t1.k = t2.k AND t1.k <> t2.k`). + val neqLeftOrdinal = outputOrdinal(leftPlan, neqPairs.head._1) + if (neqLeftOrdinal < 0 || leftEquiOrdinals.contains(neqLeftOrdinal)) return None + + // Datatype safety is checked LAST, only after correspondence is proven: these really are the + // paired equi-keys and neq column. The rewrite replaces the join predicates `=` / `<>` with + // GROUP BY / COUNT(DISTINCT), i.e. it swaps comparison equality for grouping/distinct equality, + // so every column moved into the aggregate -- both ends of every pair, not just the sjLeft side + // -- must be a type where the two equalities provably coincide (see + // [[isSafeComparisonGroupingType]]). Checking both ends (rather than trusting + // `left.canonicalized == right.canonicalized` to imply matching type/metadata) keeps this + // robust if a future metadata-aware type gate is added. + val comparisonAttrs = (equiPairs ++ neqPairs).flatMap { case (l, r) => Seq(l, r) } + if (!comparisonAttrs.forall(a => isSafeComparisonGroupingType(a.dataType))) return None + + Some((equiPairs, neqPairs)) + } + + /** + * True iff comparison equality (`=` / `<>`) and grouping/distinct equality provably coincide for + * `dataType`, so the key can be safely moved from a join predicate into GROUP BY / + * COUNT(DISTINCT). + * + * This is a POSITIVE allowlist, not "orderable minus a blacklist". The property we must prove is + * not that a type can be ordered, but that its `=` / `<>` semantics and its grouping/distinct + * semantics are identical. `RowOrdering.isOrderable` answers the former, not the latter, so it is + * not a sufficient proof here. Rather than depend on non-trivial, version-dependent equality + * contracts, we allow only the small set of types whose two equalities coincide unconditionally + * and which the target workload (TPC-DS Q95) actually needs: + * - Float/Double: their equality around NaN and signed zero relies on normalization semantics + * (`NormalizeFloatingNumbers` and friends) that we do not want this rule -- which spans + * multiple Spark versions and native execution paths -- to depend on. + * - String / CHAR / VARCHAR: non-binary collation semantics make the equivalence non-trivial + * and version-dependent, and CHAR/VARCHAR may already appear as StringType-plus-metadata by + * the optimizer, so `dataType` alone cannot even see the declared type. String is therefore + * outside the initial allowlist. + * Everything else -- complex types (Array/Map/Struct), UDTs, and any future/unknown type -- fails + * closed for the same reason: we would rather miss the rewrite than depend on a contract we have + * not proven holds across every supported backend. + */ + private def isSafeComparisonGroupingType(dataType: DataType): Boolean = dataType match { + case ByteType | ShortType | IntegerType | LongType => true + case _: DecimalType => true + case BooleanType => true + case DateType => true + case TimestampType => true + case BinaryType => true + case _ => false + } + + /** + * True iff `plan` produces the same row bag on every evaluation. + * + * This is the primary safety guard for the rewrite, which folds two occurrences of the same + * subtree into one aggregate -- sound only when both occurrences produce identical row bags. We + * check it at TWO levels: + * - candidate level: the enclosing subquery, before descending into the self-join. Catches + * nondeterminism that has been hoisted OUT of the join by an earlier optimizer rule -- e.g. a + * `Filter(rand(...))` moved to sit above the join rather than on each side. Without this, + * `isSameBaseRelation(innerLeft, innerRight)` could pass (both sides look deterministic) + * while the enclosing plan still contains `Rand`. + * - relation level: [[isSameBaseRelation]] additionally requires the two sides to be + * structurally identical. + * + * Attribute-level `plan.deterministic` alone is NOT sufficient. Catalyst's + * `Expression.deterministic` only checks explicit `Nondeterministic` annotation; several + * operators produce a runtime-nondeterministic row bag even though every expression they contain + * is `deterministic == true`: + * - `Aggregate` with `First` / `Last` / `collect_list` / `min_by` / `max_by` (tie order), + * - `Window` with `row_number()` / `rank()` over a non-total order, + * - `Limit` / `LocalLimit` / `Sample` / `Offset` (row-bag operator-level nondeterminism), + * - streaming sources. + * + * This rule collapses two evaluations of the same subtree into one aggregate; repeatability must + * be provable, not assumed. That is why both the operator check and the expression check below + * are WHITELISTS rather than blacklists -- unknown operators and unknown expression types default + * to reject. + * + * Expression support is allowlisted, not blacklisted. `plan.deterministic` relies on each + * expression's reported `deterministic` contract; that is necessary but insufficient for unknown + * expression types whose repeatability has not been established -- a builtin that carries hidden + * state yet reports `deterministic == true` would otherwise be trusted silently. For a rewrite + * that folds two evaluations of a subtree into one aggregate we prefer to miss an optimization + * than to misapply one, so new expression types are added to [[isRepeatableExpression]] only + * after their repeatability has been established. `plan.deterministic` is kept as a cheap + * fast-reject, but the expression allowlist is what actually proves repeatability. + * + * `plan.subqueriesAll.isEmpty` additionally fail-closes on any embedded expression subquery + * (scalar / IN / EXISTS). `plan.exists` in `isRowBagRepeatable` walks only the operator tree and + * does not descend into expression subqueries, and `plan.deterministic` does not prove a nested + * subquery is row-bag repeatable (e.g. an uncorrelated `LIMIT 1` without `ORDER BY`). Rejecting + * any embedded subquery keeps the repeatability proof confined to the operator whitelist below. + */ + private def isRepeatablePlan(plan: LogicalPlan): Boolean = { + // The operator/source whitelist is checked before the expression whitelist so that a plan whose + // operator is itself unknown -- e.g. Aggregate (carries AggregateExpression) or Window (carries + // WindowExpression / SortOrder) -- is attributed to isRowBagRepeatable rather than being masked + // by the fact that those operators also carry non-allowlisted expressions. + plan.deterministic && + !plan.isStreaming && + plan.subqueriesAll.isEmpty && + isRowBagRepeatable(plan) && + hasRepeatableExpressions(plan) + } + + /** + * Operator whitelist for `isRepeatablePlan`. A plan is row-bag repeatable only when every node is + * known to produce a repeatable output row bag from repeatable children. Unknown operators and + * unknown leaf sources fail closed. + * + * Kept intentionally narrow -- the target workload (Q95-shape self-join in a subquery) only needs + * a Parquet relation scan optionally wrapped in Project / Filter / SubqueryAlias plus the + * self-join itself. Range and LocalRelation are also trusted deterministic leaves. Adding an + * operator here requires proving: + * - it does not reorder its input non-deterministically, + * - it does not depend on shuffle-merge or tie-broken orderings, + * - it produces the same output row bag on every evaluation. + * + * Arbitrary `LeafNode`s are intentionally not trusted. For example, `LogicalRDD` may wrap an + * arbitrary RDD lineage whose runtime behavior is invisible to Catalyst's `plan.deterministic`; + * `InMemoryRelation`, `DataSourceV2Relation` and custom leaves are likewise rejected until + * proven. Streaming sources reach here as `LeafNode`s but are already filtered upstream by + * `plan.isStreaming` in [[isRepeatablePlan]]. + * + * `LogicalRelation` is trusted only when its underlying relation is a `HadoopFsRelation` whose + * `fileFormat` is EXACTLY `ParquetFileFormat` (`getClass == classOf[ParquetFileFormat]`, not + * `isInstanceOf`). `HadoopFsRelation.fileFormat` can be any `FileFormat`, including custom + * formats whose scan is not provably a repeatable row bag; `ParquetFileFormat` is also non-final, + * so a third-party subclass could override its scan. The target workload only needs stock + * Parquet, so every other `FileFormat` -- subclasses of `ParquetFileFormat` included -- and every + * non-`HadoopFsRelation` fail closed. + * + * This helper checks operators and leaf sources only; expression-type repeatability is a separate + * concern handled by [[hasRepeatableExpressions]], and both are joined in [[isRepeatablePlan]]. + */ + private def isRowBagRepeatable(plan: LogicalPlan): Boolean = !plan.exists { + // TreeNode exposes `exists` but not `forall`, so invert: a whitelisted operator maps to `false` + // ("does not break repeatability") and everything else to `true`; negating the whole `exists` + // then means "every operator is whitelisted". + case _: Project => false + case _: Filter => false + case _: SubqueryAlias => false + // Join is included because our target pattern IS a Join; both children get recursed into. + case _: Join => false + // Explicitly trusted leaves. + case _: Range => false + case _: LocalRelation => false + case relation: LogicalRelation => + relation.relation match { + case h: HadoopFsRelation if h.fileFormat.getClass == classOf[ParquetFileFormat] => false + case _ => true + } + // Everything else -- Aggregate (First/Last), Window (row_number ties), Limit, Sample, + // Offset, Distinct, Union, Except, Intersect, Sort (may break ties nondeterministically), + // Expand, Generate, and opaque leaf sources -- fail-closed. + case _ => true + } + + /** + * Expression-type check for [[isRepeatablePlan]], kept separate from the operator whitelist in + * [[isRowBagRepeatable]] so the two safety layers read independently. A plan passes only when + * every expression carried by every operator node is repeatable per [[isRepeatableExpression]]; a + * whitelisted operator holding an unknown expression -- e.g. `Project(Abs(v))` -- fails closed. + */ + private def hasRepeatableExpressions(plan: LogicalPlan): Boolean = { + !plan.exists(node => node.expressions.exists(expr => !isRepeatableExpression(expr))) + } + + /** + * Expression repeatability is allowlisted, consistently with the operator whitelist in + * [[isRowBagRepeatable]]. Only expression types whose result is determined entirely by repeatable + * children are accepted; unknown expression types fail closed. This rule folds two evaluations of + * the same subtree into one, so missing an optimization is preferable to assuming repeatability + * for an expression whose runtime behavior has not been proven. + * + * The whole tree is checked: an expression is repeatable only when its root type is on the + * allowlist AND all of its children are themselves repeatable, so e.g. `Add(v, Abs(w))` is + * rejected even though `Add` is allowlisted. + * + * The initial set covers what TPC-DS Q95 and the tests need: column / literal references, Alias, + * Cast, basic arithmetic, boolean connectives, the six comparisons plus null-safe equality, and + * null checks. Expressions such as `Abs` / `Coalesce` / `CaseWhen` are intentionally absent -- a + * self-join over them is simply not rewritten (a missed optimization, not a correctness bug) + * until each is added here after its repeatability has been established. Decimal wrappers such as + * `PromotePrecision` / `CheckOverflow` are likewise absent and may fail closed; that is + * acceptable and must not be worked around by trusting them just to make an arithmetic variant + * fire. + */ + private def isRepeatableExpression(expr: Expression): Boolean = expr match { + case _: Attribute | _: Literal => + true + case _: Alias | _: Cast | _: Add | _: Subtract | _: Multiply | _: Divide | _: Remainder | + _: And | _: Or | _: Not | _: EqualTo | _: EqualNullSafe | _: LessThan | + _: LessThanOrEqual | _: GreaterThan | _: GreaterThanOrEqual | _: IsNull | _: IsNotNull => + expr.children.forall(isRepeatableExpression) + case _ => + false + } + + /** + * True iff `left` and `right` are the same plan modulo canonicalization AND each side is a + * repeatable plan. See [[isRepeatablePlan]] for the repeatability contract. + */ + private def isSameBaseRelation(left: LogicalPlan, right: LogicalPlan): Boolean = { + left.canonicalized == right.canonicalized && + isRepeatablePlan(left) && isRepeatablePlan(right) Review Comment: `rewriteSubqueryPlan` already rejects non-repeatable candidates via `isRepeatablePlan(plan)`, but `isSameBaseRelation` then re-walks both self-join sides with `isRepeatablePlan(left/right)` (and this happens multiple times on the nested path as well). This is a concrete extra traversal cost in the optimizer. Consider splitting `isSameBaseRelation` into (a) structural equality (`canonicalized`) and (b) repeatability, and in the subquery rewrite path only pay the repeatability walk once (or memoize `isRepeatablePlan` per subtree for the duration of `rewriteSubqueryPlan`). ########## backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,734 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.config.VeloxConfig + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, IntegerType, LongType, ShortType, TimestampType} + +/** + * Rewrites self-join with inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery 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 with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.gluten.sql.rewrite.selfJoinInequality` (default false, opt-in). + */ +case class RewriteSelfJoinInequalityToAggregate(spark: SparkSession) + extends Rule[LogicalPlan] + with PredicateHelper + with Logging { + + private val CountDistinctAliasName = "_gluten_rw_selfjoin_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!VeloxConfig.get.enableRewriteSelfJoinInequality) { + logDebug("RewriteSelfJoinInequalityToAggregate: disabled via config, skipping") + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressions { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) + case _ => None + } + if (mapped.exists(_.isEmpty)) { + None + } else { + val newProjectList = mapped.flatten + val newWrapper = Project(newProjectList, newChild) + val newOutput = newWrapper.output + val remap: Map[ExprId, Attribute] = + oldOutput.zip(newOutput).map { case (o, n) => o.exprId -> n }.toMap + Some((newWrapper, remap)) + } + } + + /** + * Replace equi-key attribute references inside a NamedExpression according to `remap`, while + * preserving the NamedExpression shape. + * + * `Expression.transformUp` returns `Expression`, not `NamedExpression`. We avoid a blanket + * `asInstanceOf[NamedExpression]` by handling the two shapes that can appear in a Project's + * `projectList` explicitly: a bare Attribute (whose top-level may itself be replaced) and an + * Alias (which stays an Alias while its child is transformed). Any other NamedExpression shape we + * do not rewrite is left as-is ONLY if it does not reference a replaced self-join output; + * otherwise it would carry a stale ExprId, so returns None to fail the whole rewrite closed. + */ + private def remapNamedExpressionAttributes( + ne: NamedExpression, + remap: Map[ExprId, Attribute]): Option[NamedExpression] = ne match { + case a: Attribute if remap.contains(a.exprId) => Some(remap(a.exprId)) + case a: Attribute => Some(a) + case al: Alias => + val newChild = al.child.transformUp { + case a: Attribute if remap.contains(a.exprId) => remap(a.exprId) + } + Some( + if (newChild eq al.child) { + al + } else { + Alias(newChild, al.name)( + al.exprId, + al.qualifier, + al.explicitMetadata, + al.nonInheritableMetadataKeys) + }) + case other if other.references.exists(a => remap.contains(a.exprId)) => + // Fail-closed: a NamedExpression we do not rewrite (neither a bare Attribute nor an Alias) + // that still references a replaced self-join output would be left with a dangling ExprId. + // Refuse the rewrite rather than emit a plan with a stale reference. + None + case other => Some(other) + } + + // ============================================================================ + // Pattern A' / A2 dispatch (subquery plans of InSubquery) + // ============================================================================ + + private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = { + // Candidate-level nondeterminism guard: reject if ANY node in the whole subquery plan + // is non-repeatable (Rand, LIMIT-without-ORDER-BY, Sample, Offset, streaming). This + // catches nondeterminism that has been hoisted above the self-join by an earlier + // optimizer rule -- the per-side `isSameBaseRelation` check alone would miss it because + // both innerLeft/innerRight can look deterministic after such a hoist. + if (!isRepeatablePlan(plan)) return None + + val (projectListOpt, innerJoin): (Option[Seq[NamedExpression]], Join) = plan match { + case Project(pl, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(pl), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + if (isSameBaseRelation(innerJoin.left, innerJoin.right)) { + rewriteDirectSelfJoin(projectListOpt, innerJoin) + } else { + rewriteNestedSelfJoin(projectListOpt, innerJoin) + } + } + + // ============================================================================ + // Pattern A' : direct self-join at subquery top level + // ============================================================================ + + private def rewriteDirectSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + innerJoin: Join): Option[LogicalPlan] = { + // Fail closed on an explicit join hint. This rule deletes the self-join outright; a user + // `/*+ BROADCAST(...) */` (or SHUFFLE_*/MERGE) is an explicit optimizer directive about THAT + // join, and an opt-in rewrite has no business silently discarding it. Not a correctness bug + // (hints do not change results), but the conservative maintainer choice. `JoinHint.NONE` exists + // across all Spark versions Gluten supports, so no shim is needed. + if (innerJoin.hint != JoinHint.NONE) return None + + val innerLeft = innerJoin.left + val innerRight = innerJoin.right + val innerCond = innerJoin.condition.get + + val parsed = parseSelfJoinCondition(innerCond, innerLeft, innerRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val innerLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val innerLeftNeqAttr: Attribute = neqPairs.head._1 + val filtered = buildAggregateHavingDistinctGt1(innerLeftEquiAttrs, innerLeftNeqAttr, innerLeft) + + // Fail-closed on bare-Join subqueries: without a wrapping Project the subquery output + // is the full self-join output (both sides' columns). Replacing that with + // `Project(equiKeys, filtered)` shrinks the output; if the enclosing InSubquery + // referenced a non-equi column by position, `values.zip(sub.output).map(EqualTo.tupled)` + // inside RewritePredicateSubquery would build an incorrect semi condition. Q95's + // subqueries all have an explicit Project wrapper, so this branch does not affect it. + projectListOpt match { + case None => + None + case Some(pl) => + canonicalizeWrapper(pl, equiPairs, filtered).map { + case (newWrapper, _) => + logDebug( + s"Pattern A' - equiKeys=[${innerLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${innerLeftNeqAttr.name}" + + s", outCols=[${newWrapper.projectList.map(_.name).mkString(",")}]") + newWrapper + } + } + } + + // ============================================================================ + // Pattern A2 : self-join nested inside another InnerJoin in the subquery + // ============================================================================ + + private def rewriteNestedSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + outerJoin: Join): Option[LogicalPlan] = { + val outerCond = outerJoin.condition.get + + val (selfJoinSide, selfJoinOnRight) = + tryExtractSelfJoin(outerJoin.right) match { + case Some(_) => (outerJoin.right, true) + case None => + tryExtractSelfJoin(outerJoin.left) match { + case Some(_) => (outerJoin.left, false) + case None => return None + } + } + + val (selfJoinProjectOpt, selfJoin) = selfJoinSide match { + case p @ Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(p), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + val sjLeft = selfJoin.left + val sjRight = selfJoin.right + val sjCond = selfJoin.condition.get + if (!isSameBaseRelation(sjLeft, sjRight)) return None + + val parsed = parseSelfJoinCondition(sjCond, sjLeft, sjRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + Review Comment: `rewriteNestedSelfJoin` uses `tryExtractSelfJoin` only to decide which side contains the self-join, but then re-extracts the join again (and later re-parses the self-join condition). This duplicates work and makes the control flow harder to follow. A tangible simplification is to have `tryExtractSelfJoin` return the extracted shape needed downstream (e.g., `(projectOpt, join, parsedPairs)`), and reuse that result rather than matching/parsing a second time. ########## backends-velox/src/main/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregate.scala: ########## @@ -0,0 +1,734 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.config.VeloxConfig + +import org.apache.spark.internal.Logging +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.catalyst.expressions._ +import org.apache.spark.sql.catalyst.expressions.aggregate._ +import org.apache.spark.sql.catalyst.plans._ +import org.apache.spark.sql.catalyst.plans.logical._ +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation} +import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat +import org.apache.spark.sql.types.{BinaryType, BooleanType, ByteType, DataType, DateType, DecimalType, IntegerType, LongType, ShortType, TimestampType} + +/** + * Rewrites self-join with inequality into GROUP BY + HAVING COUNT(DISTINCT) > 1. + * + * Targets the two uncorrelated InSubquery shapes exercised by TPC-DS Q95: + * + * - Pattern A': the subquery 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 with Aggregate and the outer join is preserved. + * + * Both patterns require an existence-only membership context so row-count multiplicity from the + * original self-join cross-product does not affect semantics. Correlated InSubquery expressions are + * intentionally fail-closed because the ExprId remapping performed here does not rewrite correlated + * predicates. + * + * Both patterns share: + * - [[buildAggregateHavingDistinctGt1]] to construct `Filter(cnt > 1, Aggregate)` + * - [[canonicalizeWrapper]] to rebuild a wrapping Project so every equi-key reference points to + * the sjLeft-side attribute, with **fresh exprIds** (Spark's SPARK-21835 style -- no reuse of + * original exprIds), returning an old->new attribute remap for downstream rewrite. + * + * Controlled by `spark.gluten.sql.rewrite.selfJoinInequality` (default false, opt-in). + */ +case class RewriteSelfJoinInequalityToAggregate(spark: SparkSession) + extends Rule[LogicalPlan] + with PredicateHelper + with Logging { + + private val CountDistinctAliasName = "_gluten_rw_selfjoin_cnt_distinct" + + override def apply(plan: LogicalPlan): LogicalPlan = { + if (!VeloxConfig.get.enableRewriteSelfJoinInequality) { + logDebug("RewriteSelfJoinInequalityToAggregate: disabled via config, skipping") + return plan + } + + // Pattern A' / A2: rewrite uncorrelated InSubquery plans. + // Correlated subqueries carry outer references / correlated join conditions in + // `SubqueryExpression.children`; fail closed because this rule does not remap them. + val rewritten = plan.transformAllExpressions { + case in @ InSubquery(_, lq: ListQuery) if lq.children.isEmpty => + rewriteSubqueryPlan(lq.plan) match { + case Some(newSub) => in.copy(query = lq.copy(plan = newSub)) + case None => in + } + } + if (!(rewritten eq plan)) { + logDebug( + "RewriteSelfJoinInequalityToAggregate: rewrote self-join to " + + "GROUP BY + HAVING COUNT(DISTINCT) > 1") + } + rewritten + } + // ============================================================================ + // Shared helpers + // ============================================================================ + + /** + * Build `Filter(cnt > 1, Aggregate(equiKeys, [equiKeys, cnt_alias], Filter(IsNotNull(equiKeys), + * child)))`. Returns the Filter node whose output is `equiKeys ++ [count_alias_attr]`. + * + * The extra `IsNotNull(equiKeys)` filter is essential to preserve the original equi-join's NULL + * semantics. Under SQL 3VL, `left.k = right.k` never matches when either side is NULL, so the + * original self-join drops rows with NULL equi-keys. Aggregate, in contrast, groups NULL keys + * together into a single "NULL group" -- if that group has >= 2 distinct non-null neq values, + * COUNT(DISTINCT) > 1 fires and injects NULL into the subquery output. That leaked NULL then + * turns `NOT IN` into a spurious empty result (Spark's null-aware anti-join uses + * `Or(equi, IsNull(equi))` which any NULL sub-row satisfies) and can flip IN/NOT IN outcomes. The + * neq column needs no such filter: `COUNT(DISTINCT col)` already ignores NULL. + */ + private def buildAggregateHavingDistinctGt1( + equiKeys: Seq[Attribute], + neqCol: Attribute, + child: LogicalPlan): LogicalPlan = { + val countExpr = AggregateExpression( + Count(Seq(neqCol)), + mode = Complete, + isDistinct = true, + filter = None, + NamedExpression.newExprId) + val countAlias = Alias(countExpr, CountDistinctAliasName)() + // Seq[Attribute] is a Seq[NamedExpression] via covariance; no cast needed. + val aggExprs: Seq[NamedExpression] = equiKeys :+ countAlias + val nonNullChild = equiKeys + .map(a => IsNotNull(a): Expression) + .reduceOption(And) + .map(Filter(_, child)) + .getOrElse(child) + val agg = Aggregate(equiKeys, aggExprs, nonNullChild) + Filter(GreaterThan(countAlias.toAttribute, Literal(1L, LongType)), agg) + } + + /** + * Canonicalize a Project so every equi-key reference points at the sjLeft-side attribute. + * [[parseSelfJoinCondition]] has already verified that each pair refers to the same output + * position on the two structurally identical self-join sides. Uses **fresh exprIds** (no reuse of + * original wrapper output exprIds) -- the same technique Spark's own `dedupSubqueryOnSelfJoin` + * uses when it needs to change subquery output. + * + * Returns the rebuilt Project and a map `oldWrapperOutputExprId -> newWrapperOutputAttr`, so + * downstream references (outer join condition, top-level Project) can be updated consistently. + * + * `equiPairs` provides the definitive ExprId-based lookup: `equiPair (l, r)` binds + * `l.exprId -> l` (identity) and `r.exprId -> l` (sjRight -> sjLeft). Attribute identity in + * Catalyst is ExprId, not name; two columns can share a name with distinct ExprIds. Name-based + * lookup would silently drop such entries via `.toMap`. + * + * Fails (returns None) when a projectList entry is neither an equi-key Attribute (by ExprId) nor + * `Alias(equi-key Attribute, _)`. Fail-closed. + */ + private def canonicalizeWrapper( + projectList: Seq[NamedExpression], + equiPairs: Seq[(Attribute, Attribute)], + newChild: LogicalPlan): Option[(Project, Map[ExprId, Attribute])] = { + // ExprId-based canonical map: any equi-key attribute (either side) -> sjLeft attribute. + val exprIdToLeft: Map[ExprId, Attribute] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId -> l, r.exprId -> l) }.toMap + val oldOutput: Seq[Attribute] = projectList.map(_.toAttribute) + val mapped: Seq[Option[NamedExpression]] = projectList.map { + case a: Attribute if exprIdToLeft.contains(a.exprId) => + // Wrap every rewritten output slot in a fresh Alias. + // + // When a wrapper reprojects BOTH sides of the same equi pair (e.g. + // `SELECT s1.k, s2.k FROM T s1 JOIN T s2 ON s1.k = s2.k AND s1.v <> s2.v`), + // both entries collapse to the same sjLeft Attribute after the self-join is + // rewritten. Duplicate output ExprIds are not illegal in Spark (`SELECT a, a` + // is a valid Project), but fresh Aliases give each output slot an independent + // identity, which keeps the `oldOutput -> newOutput` remap 1-to-1 and lets + // downstream references (outer join condition, top-level Project) be updated + // unambiguously via ExprId. + // + // The fresh ExprId is on the Alias ITSELF; the referenced child keeps its + // original ExprId. Spark's logical-plan integrity checks reject reusing a + // referenced ExprId as the Alias's own ExprId, not duplication across slots. + Some(Alias(exprIdToLeft(a.exprId), a.name)(): NamedExpression) + case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) => + // Fresh exprId; do NOT reuse `al.exprId`. Reusing another expression's exprId + // is the pattern that Spark 3.3 flags via structural-integrity checks. + Some(Alias(exprIdToLeft(a.exprId), al.name)(): NamedExpression) + case _ => None + } + if (mapped.exists(_.isEmpty)) { + None + } else { + val newProjectList = mapped.flatten + val newWrapper = Project(newProjectList, newChild) + val newOutput = newWrapper.output + val remap: Map[ExprId, Attribute] = + oldOutput.zip(newOutput).map { case (o, n) => o.exprId -> n }.toMap + Some((newWrapper, remap)) + } + } + + /** + * Replace equi-key attribute references inside a NamedExpression according to `remap`, while + * preserving the NamedExpression shape. + * + * `Expression.transformUp` returns `Expression`, not `NamedExpression`. We avoid a blanket + * `asInstanceOf[NamedExpression]` by handling the two shapes that can appear in a Project's + * `projectList` explicitly: a bare Attribute (whose top-level may itself be replaced) and an + * Alias (which stays an Alias while its child is transformed). Any other NamedExpression shape we + * do not rewrite is left as-is ONLY if it does not reference a replaced self-join output; + * otherwise it would carry a stale ExprId, so returns None to fail the whole rewrite closed. + */ + private def remapNamedExpressionAttributes( + ne: NamedExpression, + remap: Map[ExprId, Attribute]): Option[NamedExpression] = ne match { + case a: Attribute if remap.contains(a.exprId) => Some(remap(a.exprId)) + case a: Attribute => Some(a) + case al: Alias => + val newChild = al.child.transformUp { + case a: Attribute if remap.contains(a.exprId) => remap(a.exprId) + } + Some( + if (newChild eq al.child) { + al + } else { + Alias(newChild, al.name)( + al.exprId, + al.qualifier, + al.explicitMetadata, + al.nonInheritableMetadataKeys) + }) + case other if other.references.exists(a => remap.contains(a.exprId)) => + // Fail-closed: a NamedExpression we do not rewrite (neither a bare Attribute nor an Alias) + // that still references a replaced self-join output would be left with a dangling ExprId. + // Refuse the rewrite rather than emit a plan with a stale reference. + None + case other => Some(other) + } + + // ============================================================================ + // Pattern A' / A2 dispatch (subquery plans of InSubquery) + // ============================================================================ + + private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = { + // Candidate-level nondeterminism guard: reject if ANY node in the whole subquery plan + // is non-repeatable (Rand, LIMIT-without-ORDER-BY, Sample, Offset, streaming). This + // catches nondeterminism that has been hoisted above the self-join by an earlier + // optimizer rule -- the per-side `isSameBaseRelation` check alone would miss it because + // both innerLeft/innerRight can look deterministic after such a hoist. + if (!isRepeatablePlan(plan)) return None + + val (projectListOpt, innerJoin): (Option[Seq[NamedExpression]], Join) = plan match { + case Project(pl, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(pl), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + if (isSameBaseRelation(innerJoin.left, innerJoin.right)) { + rewriteDirectSelfJoin(projectListOpt, innerJoin) + } else { + rewriteNestedSelfJoin(projectListOpt, innerJoin) + } + } + + // ============================================================================ + // Pattern A' : direct self-join at subquery top level + // ============================================================================ + + private def rewriteDirectSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + innerJoin: Join): Option[LogicalPlan] = { + // Fail closed on an explicit join hint. This rule deletes the self-join outright; a user + // `/*+ BROADCAST(...) */` (or SHUFFLE_*/MERGE) is an explicit optimizer directive about THAT + // join, and an opt-in rewrite has no business silently discarding it. Not a correctness bug + // (hints do not change results), but the conservative maintainer choice. `JoinHint.NONE` exists + // across all Spark versions Gluten supports, so no shim is needed. + if (innerJoin.hint != JoinHint.NONE) return None + + val innerLeft = innerJoin.left + val innerRight = innerJoin.right + val innerCond = innerJoin.condition.get + + val parsed = parseSelfJoinCondition(innerCond, innerLeft, innerRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val innerLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val innerLeftNeqAttr: Attribute = neqPairs.head._1 + val filtered = buildAggregateHavingDistinctGt1(innerLeftEquiAttrs, innerLeftNeqAttr, innerLeft) + + // Fail-closed on bare-Join subqueries: without a wrapping Project the subquery output + // is the full self-join output (both sides' columns). Replacing that with + // `Project(equiKeys, filtered)` shrinks the output; if the enclosing InSubquery + // referenced a non-equi column by position, `values.zip(sub.output).map(EqualTo.tupled)` + // inside RewritePredicateSubquery would build an incorrect semi condition. Q95's + // subqueries all have an explicit Project wrapper, so this branch does not affect it. + projectListOpt match { + case None => + None + case Some(pl) => + canonicalizeWrapper(pl, equiPairs, filtered).map { + case (newWrapper, _) => + logDebug( + s"Pattern A' - equiKeys=[${innerLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${innerLeftNeqAttr.name}" + + s", outCols=[${newWrapper.projectList.map(_.name).mkString(",")}]") + newWrapper + } + } + } + + // ============================================================================ + // Pattern A2 : self-join nested inside another InnerJoin in the subquery + // ============================================================================ + + private def rewriteNestedSelfJoin( + projectListOpt: Option[Seq[NamedExpression]], + outerJoin: Join): Option[LogicalPlan] = { + val outerCond = outerJoin.condition.get + + val (selfJoinSide, selfJoinOnRight) = + tryExtractSelfJoin(outerJoin.right) match { + case Some(_) => (outerJoin.right, true) + case None => + tryExtractSelfJoin(outerJoin.left) match { + case Some(_) => (outerJoin.left, false) + case None => return None + } + } + + val (selfJoinProjectOpt, selfJoin) = selfJoinSide match { + case p @ Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => + (Some(p), j) + case j: Join if j.joinType == Inner && j.condition.isDefined => + (None, j) + case _ => return None + } + + val sjLeft = selfJoin.left + val sjRight = selfJoin.right + val sjCond = selfJoin.condition.get + if (!isSameBaseRelation(sjLeft, sjRight)) return None + + val parsed = parseSelfJoinCondition(sjCond, sjLeft, sjRight) + if (parsed.isEmpty) return None + // parseSelfJoinCondition has validated column correspondence and equi-key uniqueness. + val (equiPairs, neqPairs) = parsed.get + + val sjLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1) + val sjLeftNeqAttr: Attribute = neqPairs.head._1 + + val selfJoinOutputSet = selfJoinSide.outputSet + val sjEquiExprIds: Set[ExprId] = + equiPairs.flatMap { case (l, r) => Seq(l.exprId, r.exprId) }.toSet + // wrapper Project may reproject equi-keys under fresh alias exprIds; include those. + val wrapperEquiExprIds: Set[ExprId] = selfJoinProjectOpt.toSeq.flatMap { + p => + p.projectList.flatMap { + case a: Attribute if sjEquiExprIds.contains(a.exprId) => Some(a.exprId) + case al @ Alias(a: Attribute, _) if sjEquiExprIds.contains(a.exprId) => Some(al.exprId) + case _ => None + } + }.toSet + val allEquiExprIds = sjEquiExprIds ++ wrapperEquiExprIds + + // Outer join condition may reference only equi-key attrs from the self-join side. + val outerCondRefs = outerCond.references.filter(selfJoinOutputSet.contains) + if (!outerCondRefs.forall(a => allEquiExprIds.contains(a.exprId))) return None + + // Top-level subquery Project may reference only equi-key attrs from the self-join side. + val projectOk = projectListOpt.forall { + pl => + val refs = pl.flatMap(_.references).filter(selfJoinOutputSet.contains) + refs.forall(a => allEquiExprIds.contains(a.exprId)) + } + if (!projectOk) return None + + val filtered = buildAggregateHavingDistinctGt1(sjLeftEquiAttrs, sjLeftNeqAttr, sjLeft) + + val (newSelfJoinSide, outputRemap): (LogicalPlan, Map[ExprId, Attribute]) = + selfJoinProjectOpt match { + case Some(wp) => + canonicalizeWrapper(wp.projectList, equiPairs, filtered) match { + case Some((newWrapper, remap)) => (newWrapper, remap) + case None => return None + } + case None if projectListOpt.isEmpty => + // Fail-closed: with neither a wrapper Project around the self-join nor a top-level + // subquery Project, the outer join currently exposes every self-join column, and + // replacing the self-join with `Project(equiKeys, filtered)` would shrink the outer + // join's right-hand output arity. RewritePredicateSubquery's positional zip + // (`values.zip(sub.output).map(EqualTo.tupled)`) would then bind semi predicates to + // the wrong attributes -- silently dropping components of a tuple IN. A + // top-level Project (`projectListOpt`) is what would let the arity be preserved + // by the top-level rewrite loop; without one, refuse to rewrite. + return None + case None => + // No wrapper Project but there IS a top-level subquery Project: shrinking the outer + // join's self-join-side output is safe because the top-level Project is rewritten + // consistently via `outputRemap` below and the top-level rewrite loop ensures + // subquery output arity matches what the enclosing InSubquery expects. + // Outer references may point at sjRight equi-attributes; remap them to sjLeft + // (same output position in a valid self-join). + val newP = Project(sjLeftEquiAttrs, filtered) + val remap: Map[ExprId, Attribute] = + equiPairs.map { case (l, r) => r.exprId -> l }.toMap + (newP, remap) + } + + // Rewrite outer join condition to use new wrapper output attributes. + val newOuterCond = outerCond.transformUp { + case a: Attribute if outputRemap.contains(a.exprId) => outputRemap(a.exprId) + } + + val newOuterJoin = if (selfJoinOnRight) { + outerJoin.copy(right = newSelfJoinSide, condition = Some(newOuterCond)) + } else { + outerJoin.copy(left = newSelfJoinSide, condition = Some(newOuterCond)) + } + + // Rewrite top-level Project references. + val result = projectListOpt match { + case Some(pl) => + val remapped = pl.map(ne => remapNamedExpressionAttributes(ne, outputRemap)) + if (remapped.exists(_.isEmpty)) return None + Project(remapped.flatten, newOuterJoin) + case None => newOuterJoin + } + + logDebug( + s"Pattern A2 - equiKeys=[${sjLeftEquiAttrs.map(_.name).mkString(",")}]" + + s", neqCol=${sjLeftNeqAttr.name}") + Some(result) + } + + private def tryExtractSelfJoin(plan: LogicalPlan): Option[Join] = { + val join = plan match { + case Project(_, j: Join) if j.joinType == Inner && j.condition.isDefined => j + case j: Join if j.joinType == Inner && j.condition.isDefined => j + case _ => return None + } + // A hinted self-join is not an extraction candidate: fail closed so the same node the rewrite + // would delete is never even recognized here. See `rewriteDirectSelfJoin` for the rationale. + if (join.hint != JoinHint.NONE) return None + if (!isSameBaseRelation(join.left, join.right)) return None + val parsed = parseSelfJoinCondition(join.condition.get, join.left, join.right) + if (parsed.isEmpty) return None + Some(join) + } Review Comment: `rewriteNestedSelfJoin` uses `tryExtractSelfJoin` only to decide which side contains the self-join, but then re-extracts the join again (and later re-parses the self-join condition). This duplicates work and makes the control flow harder to follow. A tangible simplification is to have `tryExtractSelfJoin` return the extracted shape needed downstream (e.g., `(projectOpt, join, parsedPairs)`), and reuse that result rather than matching/parsing a second time. ########## backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala: ########## @@ -0,0 +1,929 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.execution.WholeStageTransformerSuite + +import org.apache.spark.SparkConf +import org.apache.spark.sql.Row +import org.apache.spark.sql.catalyst.expressions.{Alias, GreaterThan, Literal} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Count} +import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{IntegerType, LongType, StructField, StructType} + +/** + * Correctness tests for [[RewriteSelfJoinInequalityToAggregate]]. + * + * Positive A' / A2 cases assert both result equivalence and that the rewrite actually fired. + * + * `assert(!ruleFired(plan))` on its own only proves the rewrite did not happen -- not that it was + * the guard under test that stopped it. A fixture whose two self-join sides are not structurally + * identical is rejected by `isSameBaseRelation` before any predicate is even parsed, and such a + * test passes while covering nothing. So six important rejection paths -- the predicate parser, the + * single-inequality requirement, output-position identity, the nondeterminism guard, the + * leaf-source allowlist (LogicalRDD vs Parquet), and the expression-type allowlist (`abs(v)` vs + * `v + 1`) -- are tested as single-variable pairs: the same fixture and the same query shape, one + * control query that must fire and one variant that changes only the feature under test and must + * not. A firing control does not pin the rejection to a particular line, but it does rule out an + * unrelated fixture mismatch as the reason its partner was rejected. The row-bag whitelist + * (Aggregate, Window) stays a plain negative: dropping the operator would change the query shape + * rather than one feature. + * + * Self-joined fixtures are real tables, not temp views over VALUES. Spark deduplicates a self-join + * over a [[org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation]] via `newInstance()`, + * which refreshes one side's ExprIds without inserting a rename-only Project, so both sides stay + * structurally identical. A temp view over VALUES cannot, and Spark renames one side with a Project + * instead, which would make `isSameBaseRelation` false for every self-join below. `range()` needs + * no such treatment -- Range is a MultiInstanceRelation already. + */ +class RewriteSelfJoinInequalityToAggregateSuite extends WholeStageTransformerSuite { + + override protected val resourcePath: String = "/tpch-data-parquet" + override protected val fileFormat: String = "parquet" + + override protected def sparkConf: SparkConf = super.sparkConf + .set("spark.gluten.sql.rewrite.selfJoinInequality", "true") + .set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key, "-1") + + /** Signature alias produced by the rewrite; presence => rule definitely fired. */ + private val CountDistinctAlias = "_gluten_rw_selfjoin_cnt_distinct" + + private def ruleFired(plan: LogicalPlan): Boolean = + plan.exists { + p => + p.expressions.exists(_.exists { + case a: Alias if a.name == CountDistinctAlias => true + case _ => false + }) + } + + private def assertRuleFired(sql: String): Unit = { + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(ruleFired(plan), s"self-join inequality rewrite should fire:\n$plan") + } + } + + private def assertRuleNotFired(sql: String): Unit = { + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(!ruleFired(plan), s"self-join inequality rewrite must not fire:\n$plan") + } + } + + /** Optimized plan with the rewrite enabled. */ + private def onOptimizedPlan(sql: String): LogicalPlan = { + var plan: LogicalPlan = null + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + plan = spark.sql(sql).queryExecution.optimizedPlan + } + plan + } Review Comment: This helper relies on a mutable var initialized to `null`. Since `withSQLConf` returns the block result, this can be simplified to return the optimized plan directly (no mutable state / null sentinel), which reduces the chance of accidental NPEs if the method is modified later. ########## backends-velox/src/test/scala/org/apache/gluten/extension/RewriteSelfJoinInequalityToAggregateSuite.scala: ########## @@ -0,0 +1,929 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gluten.extension + +import org.apache.gluten.execution.WholeStageTransformerSuite + +import org.apache.spark.SparkConf +import org.apache.spark.sql.Row +import org.apache.spark.sql.catalyst.expressions.{Alias, GreaterThan, Literal} +import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Count} +import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, LogicalPlan} +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.types.{IntegerType, LongType, StructField, StructType} + +/** + * Correctness tests for [[RewriteSelfJoinInequalityToAggregate]]. + * + * Positive A' / A2 cases assert both result equivalence and that the rewrite actually fired. + * + * `assert(!ruleFired(plan))` on its own only proves the rewrite did not happen -- not that it was + * the guard under test that stopped it. A fixture whose two self-join sides are not structurally + * identical is rejected by `isSameBaseRelation` before any predicate is even parsed, and such a + * test passes while covering nothing. So six important rejection paths -- the predicate parser, the + * single-inequality requirement, output-position identity, the nondeterminism guard, the + * leaf-source allowlist (LogicalRDD vs Parquet), and the expression-type allowlist (`abs(v)` vs + * `v + 1`) -- are tested as single-variable pairs: the same fixture and the same query shape, one + * control query that must fire and one variant that changes only the feature under test and must + * not. A firing control does not pin the rejection to a particular line, but it does rule out an + * unrelated fixture mismatch as the reason its partner was rejected. The row-bag whitelist + * (Aggregate, Window) stays a plain negative: dropping the operator would change the query shape + * rather than one feature. + * + * Self-joined fixtures are real tables, not temp views over VALUES. Spark deduplicates a self-join + * over a [[org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation]] via `newInstance()`, + * which refreshes one side's ExprIds without inserting a rename-only Project, so both sides stay + * structurally identical. A temp view over VALUES cannot, and Spark renames one side with a Project + * instead, which would make `isSameBaseRelation` false for every self-join below. `range()` needs + * no such treatment -- Range is a MultiInstanceRelation already. + */ +class RewriteSelfJoinInequalityToAggregateSuite extends WholeStageTransformerSuite { + + override protected val resourcePath: String = "/tpch-data-parquet" + override protected val fileFormat: String = "parquet" + + override protected def sparkConf: SparkConf = super.sparkConf + .set("spark.gluten.sql.rewrite.selfJoinInequality", "true") + .set(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key, "-1") + + /** Signature alias produced by the rewrite; presence => rule definitely fired. */ + private val CountDistinctAlias = "_gluten_rw_selfjoin_cnt_distinct" + + private def ruleFired(plan: LogicalPlan): Boolean = + plan.exists { + p => + p.expressions.exists(_.exists { + case a: Alias if a.name == CountDistinctAlias => true + case _ => false + }) + } + + private def assertRuleFired(sql: String): Unit = { + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(ruleFired(plan), s"self-join inequality rewrite should fire:\n$plan") + } + } + + private def assertRuleNotFired(sql: String): Unit = { + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + val plan = spark.sql(sql).queryExecution.optimizedPlan + assert(!ruleFired(plan), s"self-join inequality rewrite must not fire:\n$plan") + } + } + + /** Optimized plan with the rewrite enabled. */ + private def onOptimizedPlan(sql: String): LogicalPlan = { + var plan: LogicalPlan = null + withSQLConf("spark.gluten.sql.rewrite.selfJoinInequality" -> "true") { + plan = spark.sql(sql).queryExecution.optimizedPlan + } + plan + } + + /** + * Assert the rewrite's structural contract without pinning the full optimized plan. Exact + * optimized-plan equality is brittle across the Spark versions Gluten supports, because + * optimizer-derived filters, alias names, and Project collapsing differ between versions. We + * verify only the invariant the rewrite owns: a `HAVING count > 1` [[Filter]] sitting directly + * over a COUNT(DISTINCT) [[Aggregate]]. We deliberately do not assert the plan has no join: the + * rewrite may legitimately leave other joins in place, including Pattern A2's preserved outer + * join, and `assertRuleFired` already proves the self-join inequality was the thing rewritten. + */ + private def assertAggregateRewriteShape(plan: LogicalPlan): Unit = + assert( + hasAggregateRewriteShape(plan), + s"expected a HAVING count > 1 Filter directly over a COUNT(DISTINCT) Aggregate:\n$plan") + + /** True iff a `HAVING count > 1` Filter sits directly over a COUNT(DISTINCT) Aggregate. */ + private def hasAggregateRewriteShape(plan: LogicalPlan): Boolean = plan.exists { + case Filter(cond, agg: Aggregate) => + val hasDistinctCount = agg.aggregateExpressions.exists(_.exists { + case ae: AggregateExpression => ae.isDistinct && ae.aggregateFunction.isInstanceOf[Count] + case _ => false + }) + val hasGt1 = cond.exists { + case GreaterThan(_, Literal(v: Long, LongType)) => v == 1L + case _ => false + } + hasDistinctCount && hasGt1 + case _ => false + } + + /** + * A real table, so that a self-join of it dedups into two structurally identical sides. See the + * class comment for why a temp view over VALUES cannot be used for a self-joined fixture. + */ + private def createTable(name: String, schema: String, values: String): Unit = { + spark.sql(s"DROP TABLE IF EXISTS $name") + spark.sql(s"CREATE TABLE $name($schema) USING parquet") + spark.sql(s"INSERT INTO $name SELECT * FROM VALUES $values") + } Review Comment: The suite creates/drops real catalog tables with fixed names (e.g., `T`, `TM`, `TLeft`, etc.) but doesnβt scope cleanup to each test (beyond `DROP TABLE IF EXISTS` at creation time). If the build runs tests in parallel or shares the same metastore/warehouse across suites, this can cause cross-suite interference or intermittent failures. Consider using per-test unique table names (e.g., suffix with a UUID/test name) and/or `withTable(...)`-style guards to ensure cleanup even on test failures. ########## docs/velox-configuration.md: ########## @@ -84,6 +84,7 @@ nav_order: 16 | spark.gluten.sql.columnar.backend.velox.valueStream.dynamicFilter.enabled | π Dynamic | false | Whether to apply dynamic filters pushed down from hash probe in the ValueStream (shuffle reader) operator to filter rows before they reach the hash join. | | spark.gluten.sql.enable.enhancedFeatures | π Dynamic | true | Enable some features including iceberg native write and other features. | | spark.gluten.sql.rewrite.castArrayToString | π Dynamic | true | When true, rewrite `cast(array as String)` to `concat('[', array_join(array, ', ', null), ']')` to allow offloading to Velox. | +| spark.gluten.sql.rewrite.selfJoinInequality | π Dynamic | false | When true, rewrite supported uncorrelated InSubquery self-joins with inequality predicate into GROUP BY + HAVING COUNT(DISTINCT) > 1. Currently targets Parquet-backed direct and nested self-join shapes exercised by TPC-DS Q95. Opt-in default (false) until the rewrite has been exercised more broadly across workloads. | Review Comment: The description includes SQL/logical concepts (`InSubquery`, `GROUP BY`, `HAVING`, `COUNT(DISTINCT) > 1`) that are easier to read (and less ambiguous) when formatted as inline code in Markdown. Updating the doc string to wrap those tokens in backticks would improve readability and consistency with nearby entries. -- 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]
