hhr293 commented on code in PR #12756: URL: https://github.com/apache/gluten/pull/12756#discussion_r3931185612
########## 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) Review Comment: That analysis makes sense. I’d prefer to keep this branch as a defensive fallback rather than remove it, but I’ll update the comment to make it clear. -- 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]
