peter-toth commented on code in PR #58424:
URL: https://github.com/apache/spark/pull/58424#discussion_r4034692006
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -669,6 +669,23 @@ object SQLConf {
"for using switch statements in InSet must be non-negative and less
than or equal to 600")
.createWithDefault(400)
+ val REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED =
+
buildConf("spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled")
+ .internal()
+ .doc("When true, rewrites a supported existence-only inequality
self-join inside an " +
+ "uncorrelated IN subquery into a GROUP BY with HAVING MIN(v) <> MAX(v)
over a single " +
+ "copy of the relation, removing the self-join cross-product. The
rewrite fails closed " +
+ "unless the self-join sides are proven to read the same repeatable
source -- currently " +
+ "stock Parquet scans, Range, and local relations -- so it never
changes results for " +
+ "non-repeatable inputs. This is experimental and off by default: there
is no cost " +
+ "model, and the benefit depends on per-key multiplicity because the
eliminated " +
+ "self-join can generate quadratically many candidate pairs. Enable it
only for " +
+ "workloads known to contain such high-multiplicity self-joins.")
Review Comment:
**Finding 5.** Carried from round 2. The description covers this properly
now; the conf doc does not, and the conf doc is what someone reads before
turning the flag on.
`Min.aggBufferAttributes` is `AttributeReference("min", child.dataType)`
(`Min.scala:54,56`), so a STRING or BINARY inequality column gives the
aggregate a var-length buffer. `Aggregate.supportsHashAggregate` then fails
`isAggregateBufferMutable`, `Min`/`Max` are `DeclarativeAggregate` so object
hash aggregation is out too, and `AggUtils` falls through to
`SortAggregateExec` with two full sorts. Your own type probe is the
measurement: 0.6x at m=1 for STRING against 0.8x for INT.
The doc tells the reader the benefit depends on multiplicity. The inequality
column's width is the second thing that moves the break-even, and one clause
covers it, e.g. after "high-multiplicity self-joins": "A variable-width
inequality column (STRING or BINARY) raises the break-even multiplicity
further, because MIN/MAX over a var-length buffer falls back to sort
aggregation."
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregate.scala:
##########
@@ -0,0 +1,573 @@
+/*
+ * 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.spark.sql.execution
+
+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.catalyst.trees.TreePattern.IN_SUBQUERY
+import org.apache.spark.sql.catalyst.util.CharVarcharUtils
+import org.apache.spark.sql.execution.datasources.{HadoopFsRelation,
LogicalRelation, LogicalRelationWithTable}
+import org.apache.spark.sql.execution.datasources.parquet.ParquetFileFormat
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types._
+
+/**
+ * Rewrites supported uncorrelated IN-subquery inequality self-joins into
+ * `GROUP BY + HAVING MIN(neq) <> MAX(neq)`, avoiding the self-join
cross-product.
+ *
+ * Supports a direct self-join (Pattern A') and a self-join nested under an
outer inner join
+ * (Pattern A2, where only the self-join child becomes an Aggregate).
Unsupported and correlated
+ * shapes fail closed.
+ *
+ * Runs in `extendedOperatorOptimizationRules`, before
`RewritePredicateSubquery` turns the
+ * predicate subquery into a semi/anti/existence join, so it only sees the
uncorrelated
+ * `InSubquery` shape.
+ *
+ * Controlled by
`spark.sql.optimizer.rewriteSelfJoinInequalityToAggregate.enabled`
+ * (default false, opt-in).
+ */
+object RewriteSelfJoinInequalityToAggregate extends Rule[LogicalPlan] with
PredicateHelper {
+
+ private val MinNeqAliasName = "_rewrite_selfjoin_inequality_min"
+ private val MaxNeqAliasName = "_rewrite_selfjoin_inequality_max"
+
+ override def apply(plan: LogicalPlan): LogicalPlan = {
+ if
(!conf.getConf(SQLConf.REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED)) {
+ return plan
+ }
+
+ // Fail closed on correlated subqueries: `lq.children` holds the outer
references this rule
+ // does not remap.
+ plan.transformAllExpressionsWithPruning(_.containsPattern(IN_SUBQUERY)) {
+ 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
+ }
+ }
+ }
+
+ //
============================================================================
+ // Shared helpers
+ //
============================================================================
+
+ /**
+ * Build `Filter(min <> max, Aggregate(equiKeys, child))`, taking MIN and
MAX over the neq column.
+ * `MIN(neqCol) <> MAX(neqCol)` is true exactly when the group holds two or
more distinct non-null
+ * values.
+ *
+ * The `IsNotNull(equiKeys)` filter preserves the equi-join's NULL
semantics: `=` never matches a
+ * NULL key, but GROUP BY would fold all NULL keys into one group that can
leak NULL into a
+ * `NOT IN`. The neq column needs no filter -- MIN/MAX ignore NULL, and a
group with fewer than
+ * two non-null values has `min = max` (or both NULL, which makes `<>`
NULL), so `<>` is never
+ * true for it and the group is dropped.
+ *
+ * Sound only because the rule fires under `InSubquery`: IN / NOT IN
membership is insensitive to
+ * duplicate rows in the subquery result, and it binds columns by position
(`equalsStructurally`),
+ * so the A2 branch's column rename is harmless.
+ */
+ private def buildAggregateHavingMultipleDistinct(
+ equiKeys: Seq[Attribute],
+ neqCol: Attribute,
+ child: LogicalPlan): LogicalPlan = {
+ val minAlias = Alias(Min(neqCol).toAggregateExpression(),
MinNeqAliasName)()
+ val maxAlias = Alias(Max(neqCol).toAggregateExpression(),
MaxNeqAliasName)()
+ val aggExprs: Seq[NamedExpression] = equiKeys :+ minAlias :+ maxAlias
+ val nonNullChild = equiKeys
+ .map(a => IsNotNull(a): Expression)
+ .reduceOption(And)
+ .map(Filter(_, child))
+ .getOrElse(child)
+ val agg = Aggregate(equiKeys, aggExprs, nonNullChild)
+ Filter(Not(EqualTo(minAlias.toAttribute, maxAlias.toAttribute)), agg)
+ }
+
+ /**
+ * Rebuild the wrapper Project so every equi-key reference points at the
sjLeft attribute with a
+ * fresh output ExprId, returning `oldOutputExprId -> newOutputAttr` for
downstream references
+ * (outer join condition, top-level Project). Lookup is by ExprId (Catalyst
attribute identity),
+ * not name. Fails closed when an entry is neither an equi-key Attribute nor
`Alias(equi-key, _)`.
+ */
+ private def canonicalizeWrapper(
+ projectList: Seq[NamedExpression],
+ equiPairs: Seq[(Attribute, Attribute)],
+ newChild: LogicalPlan): Option[(Project, Map[ExprId, 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) =>
+ // Fresh exprId, but carry over qualifier / metadata so this branch
stays consistent with
+ // the Alias branch and a column keeps its metadata.
+ Some(
+ Alias(exprIdToLeft(a.exprId), a.name)(
+ qualifier = a.qualifier,
+ explicitMetadata = Some(a.metadata)): NamedExpression)
+ case al @ Alias(a: Attribute, _) if exprIdToLeft.contains(a.exprId) =>
+ // withNewChild preserves name/qualifier/metadata and exprId;
newInstance then re-stamps a
+ // fresh exprId, so Alias keeps ownership of its own metadata contract
instead of us
+ // re-listing its fields (which drift when Alias gains one).
+ Some(al.withNewChild(exprIdToLeft(a.exprId)).newInstance())
+ case _ => None
+ }
+ if (mapped.exists(_.isEmpty)) {
+ None
+ } else {
+ val newProjectList = mapped.flatten
+ val newWrapper = Project(newProjectList, newChild)
+ val remap: Map[ExprId, Attribute] =
+ oldOutput.zip(newWrapper.output).map { case (o, n) => o.exprId -> n
}.toMap
+ Some((newWrapper, remap))
+ }
+ }
+
+ /**
+ * Replace equi-key references inside a NamedExpression per `remap`,
preserving Attribute/Alias
+ * shape. Any other expression still referencing a replaced output returns
None (fail-closed) to
+ * avoid a dangling ExprId.
+ */
+ 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)
+ }
+ // withNewChild preserves the same exprId/qualifier/metadata the manual
copy did.
+ Some(if (newChild eq al.child) al else al.withNewChild(newChild))
+ case other if other.references.exists(a => remap.contains(a.exprId)) =>
+ None
+ case other => Some(other)
+ }
+
+ //
============================================================================
+ // Pattern A' / A2 dispatch (subquery plans of InSubquery)
+ //
============================================================================
+
+ private def rewriteSubqueryPlan(plan: LogicalPlan): Option[LogicalPlan] = {
+ // Match the candidate shape first -- a top-level Inner Join, optionally
under one wrapper
+ // Project. This structural match is cheap, so run it before the
whole-subquery
+ // `isRepeatablePlan` walk and skip that walk entirely for the many
subqueries that are not
+ // even shaped like a self-join.
+ 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
+ }
+
+ // Candidate-level guard: reject if any node in the whole subquery is
non-repeatable, catching
+ // nondeterminism hoisted above the self-join that the per-side
`isSameBaseRelation` misses.
+ if (!isRepeatablePlan(plan)) 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: it is a directive about the join
this rule deletes.
+ if (!innerJoin.hint.isEmpty) return None
+
+ val innerLeft = innerJoin.left
+ val innerCond = innerJoin.condition.get
+
+ val parsed = parseSelfJoinCondition(innerCond, innerLeft, innerJoin.right)
+ if (parsed.isEmpty) return None
+ val (equiPairs, neqPairs) = parsed.get
+
+ val innerLeftEquiAttrs: Seq[Attribute] = equiPairs.map(_._1)
+ val innerLeftNeqAttr: Attribute = neqPairs.head._1
+ val filtered =
+ buildAggregateHavingMultipleDistinct(innerLeftEquiAttrs,
innerLeftNeqAttr, innerLeft)
+
+ // Fail closed on a bare-Join subquery: with no wrapper Project, replacing
the self-join output
+ // with `Project(equiKeys, filtered)` shrinks arity and
RewritePredicateSubquery's positional
+ // `values.zip(sub.output)` would misbind semi predicates. Q95 subqueries
always have a Project.
+ projectListOpt match {
+ case None => None
+ case Some(pl) =>
+ canonicalizeWrapper(pl, equiPairs, filtered).map { case (newWrapper,
_) => 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 sjCond = selfJoin.condition.get
+ if (!isSameBaseRelation(sjLeft, selfJoin.right)) return None
+
+ val parsed = parseSelfJoinCondition(sjCond, sjLeft, selfJoin.right)
+ if (parsed.isEmpty) return None
+ 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
+ // A 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
+
+ // The outer join condition and any top-level Project may reference only
equi-key attrs from the
+ // self-join side (the neq column does not survive the rewrite).
+ val outerCondRefs = outerCond.references.filter(selfJoinOutputSet.contains)
+ if (!outerCondRefs.forall(a => allEquiExprIds.contains(a.exprId))) return
None
+ 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 = buildAggregateHavingMultipleDistinct(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 =>
+ // A bare self-join side (no wrapper Project) is not produced for a
fireable A2 by the
+ // normal optimizer pipeline: only equi keys are referenced above
the self-join, so
+ // ColumnPruning inserts a wrapper Project to drop the unused neq
column, leaving
+ // selfJoinProjectOpt = Some. Fail closed on the non-standard bare
shape.
+ return None
+ }
+
+ 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))
+ }
+
+ projectListOpt match {
+ case Some(pl) =>
+ val remapped = pl.map(ne => remapNamedExpressionAttributes(ne,
outputRemap))
+ if (remapped.exists(_.isEmpty)) return None
+ Some(Project(remapped.flatten, newOuterJoin))
+ case None => Some(newOuterJoin)
Review Comment:
**Finding 8.** This branch is live and untested. Every positive A2 case in
the suite has a top-level Project, so they all take `case Some(pl)`.
It is reachable. A `SELECT *` subquery keeps its Project only until
`RemoveNoopOperators` deletes it, so the subquery arrives at the rule as a bare
`Join`. Confirmed in a review worktree at this head with
```sql
SELECT k FROM T outer_t WHERE (k, k) IN (
SELECT * FROM D d, (SELECT s1.k FROM T s1 JOIN T s2
ON s1.k = s2.k AND s1.v <> s2.v) sj
WHERE d.k = sj.k)
```
The optimized subquery is `Join Inner, (k#12 = k#18)` with nothing above it,
the rule fires, and rewrite-on and rewrite-off agree as multisets on `{1, 3,
6}`.
This is the branch that hands the outer join straight back, so it has no
top-level Project remap to catch an arity or output-order mistake from
`canonicalizeWrapper`. One positive case in the shape above is enough.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/RewriteSelfJoinInequalityToAggregateSuite.scala:
##########
@@ -0,0 +1,1548 @@
+/*
+ * 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.spark.sql.execution
+
+import org.apache.spark.SparkThrowable
+import org.apache.spark.sql.{QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, EqualTo,
InSubquery, ListQuery, Not}
+import org.apache.spark.sql.catalyst.optimizer.ReorderJoin
+import org.apache.spark.sql.catalyst.plans.{Inner, LeftOuter}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter,
GlobalLimit, Join, LocalLimit, LogicalPlan}
+import org.apache.spark.sql.execution.datasources.{FileIndex,
HadoopFsRelation, LogicalRelationWithTable}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, 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
parsed, so such a test
+ * passes while covering nothing. Rejection paths are therefore tested as
single-variable pairs:
+ * the same fixture and query shape, one control that must fire and one
variant changing only the
+ * tested feature that must not. A firing control does not pin the rejection
to a line, but rules
+ * out an unrelated fixture mismatch as why its partner was rejected. (The
`LIMIT` case is such a
+ * pair: a Limit's expressions are all allowlisted, so the operator whitelist
alone can reject it.)
+ *
+ * Self-joined fixtures are real tables. This is not required for
`isSameBaseRelation` to hold --
+ * a VALUES temp view resolves to `LocalRelation`, also a
+ * [[org.apache.spark.sql.catalyst.analysis.MultiInstanceRelation]], and
`FinishAnalysis` removes
+ * the View wrapper before this rule's batch runs. Real tables are used
because they mirror the
+ * Parquet-backed shape this rule targets in Q95-style queries. `range()`
similarly needs no
+ * special treatment -- `Range` is a `MultiInstanceRelation` already.
+ */
+class RewriteSelfJoinInequalityToAggregateSuite extends QueryTest with
SharedSparkSession {
+
+ private val rewriteConf =
SQLConf.REWRITE_SELF_JOIN_INEQUALITY_TO_AGGREGATE_ENABLED.key
+
+ /** Signature aliases produced by the rewrite; presence of both => rule
definitely fired. */
+ private val MinNeqAlias = "_rewrite_selfjoin_inequality_min"
+ private val MaxNeqAlias = "_rewrite_selfjoin_inequality_max"
+
+ // Descends into subqueries: `QueryPlan.exists` does not, and the rewrite's
signature alias lives
+ // inside the IN-subquery when the rule runs on an analyzed
(not-yet-rewritten) plan.
+ private def hasAlias(plan: LogicalPlan, name: String): Boolean =
+ plan.collectFirstWithSubqueries {
+ case p if p.expressions.exists(_.exists {
+ case a: Alias if a.name == name => true
+ case _ => false
+ }) => ()
+ }.isDefined
+
+ /** Require BOTH aliases: a rewrite that emitted MIN but dropped MAX is
still a bug. */
+ private def ruleFired(plan: LogicalPlan): Boolean =
+ hasAlias(plan, MinNeqAlias) && hasAlias(plan, MaxNeqAlias)
+
+ /** Total count of a signature alias across the plan and its subqueries: one
per rewrite. */
+ private def countAlias(plan: LogicalPlan, name: String): Int =
+ plan.collectWithSubqueries {
+ case p => p.expressions.map(_.collect { case a: Alias if a.name == name
=> a }.size).sum
+ }.sum
+
+ private def assertRuleFired(sql: String): Unit = {
+ withSQLConf(rewriteConf -> "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(rewriteConf -> "true") {
+ val plan = spark.sql(sql).queryExecution.optimizedPlan
+ assert(!ruleFired(plan), s"self-join inequality rewrite must not
fire:\n$plan")
+ }
+ }
+
+ /**
+ * Optimize just the IN-subquery plan (rewrite left at its default-off) and
return the result, so
+ * a test can prove what shape the subquery reaches the rule as -- e.g. a
bare Join with no
+ * wrapper Project -- before asserting the rule declines it. Without this,
`assertRuleNotFired`
+ * alone can pass merely because the fixture never produced the shape the
guard means to reject.
+ */
+ private def optimizedInSubqueryPlan(sql: String): LogicalPlan = {
+ val analyzed = spark.sql(sql).queryExecution.analyzed
+ val subqueries = analyzed.subqueriesAll
+ assert(
+ subqueries.length == 1,
+ s"expected exactly one subquery in analyzed plan, got
${subqueries.length}:\n$analyzed")
+ spark.sessionState.optimizer.execute(subqueries.head)
+ }
+
+ /**
+ * A real table, mirroring the Parquet-backed shape this rule targets. See
the class comment for
+ * why a self-joined fixture need not be a real table.
+ */
+ 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")
+ }
+
+ /**
+ * Run `sql` twice, first with rewrite ON then OFF. The rewrite must not
change the outer query's
+ * row multiplicity, so assert ON and OFF agree as MULTISETS (a `.toSet`
here would hide a
+ * duplicated or dropped row) before handing callers the row sets their
fixed-value assertions
+ * compare against. `QueryTest.sameRows` is Spark's own multiset comparison
(order-insensitive,
+ * duplicate-sensitive) and formats the offending rows on mismatch.
+ */
+ private def runBoth(sql: String): (Set[Row], Set[Row]) = {
+ val on = withSQLConf(rewriteConf -> "true") {
+ spark.sql(sql).collect().toSeq
+ }
+ val off = withSQLConf(rewriteConf -> "false") {
+ spark.sql(sql).collect().toSeq
+ }
+ QueryTest.sameRows(on, off).foreach { error =>
+ fail(s"rewrite changed row multiplicity between ON and OFF:\n$error")
+ }
+ (on.toSet, off.toSet)
+ }
+
+ private def setupTable(): Unit = {
+ // k=1: distinct v={10,20} -> matches (has 2 non-null distinct)
+ // k=2: distinct v={30} -> no match (only 1)
+ // k=3: distinct v={40,50,60} -> matches
+ // k=4: v={70, NULL} -> no match (only 1 non-null)
+ // k=5: v={NULL, NULL} -> no match (0 non-null)
+ // k=6: v={80, 90, NULL} -> matches
+ // k=7: v={100,100} -> no match: duplicate-only,
min(v)==max(v)==100 so min<>max is
+ // false. A plain COUNT(v) > 1 would
wrongly match this group.
+ createTable(
+ "T",
+ "k INT, v INT",
+ """ (1, 10), (1, 10), (1, 20),
+ | (2, 30),
+ | (3, 40), (3, 50), (3, 60),
+ | (4, 70), (4, CAST(NULL AS INT)),
+ | (5, CAST(NULL AS INT)), (5, CAST(NULL AS INT)),
+ | (6, 80), (6, 90), (6, CAST(NULL AS INT)),
+ | (7, 100), (7, 100)""".stripMargin
+ )
+ }
+
+ // ==================== Positive: rewrite fires and is semantically
equivalent ===============
+
+ test("Pattern A': direct InSubquery self-join is rewritten (real table and
LocalRelation)") {
+ // LocalRelation is a documented supported leaf with no other firing test;
loop the same
+ // query over T and an equivalent VALUES-backed temp view.
+ withTable("T") {
+ withTempView("TV") {
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW TV AS SELECT * FROM VALUES
+ | (1, 10), (1, 10), (1, 20),
+ | (2, 30),
+ | (3, 40), (3, 50), (3, 60),
+ | (4, 70), (4, CAST(NULL AS INT)),
+ | (5, CAST(NULL AS INT)), (5, CAST(NULL AS INT)),
+ | (6, 80), (6, 90), (6, CAST(NULL AS INT)),
+ | (7, 100), (7, 100) AS TV(k, v)""".stripMargin)
+
+ Seq("T", "TV").foreach { rel =>
+ val sql =
+ s"""SELECT k FROM $rel outer_t WHERE k IN (
+ | SELECT s1.k FROM $rel s1 JOIN $rel s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+
+ assertRuleFired(sql)
+ assertMinMaxRewriteShape(optimizedPlanWith(sql, rewrite = true))
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"$rel rewrite ON $on != OFF $off")
+ // Duplicate-only groups and NULL neq values also pin the neq 3VL:
k=4 (v={70,NULL}) and
+ // k=5 (v={NULL,NULL}) do not satisfy `<>`, leaving {1,3,6}.
+ assert(on == Set(Row(1), Row(3), Row(6)), s"$rel: expected {1,3,6},
got $on")
+ }
+ }
+ }
+ }
+
+ test("InSubquery in a SELECT-list CASE WHEN is rewritten, not only in a
WHERE predicate") {
+ withTable("T") {
+ // `apply` rewrites via transformAllExpressionsWithPruning, so an
InSubquery anywhere in the
+ // plan is a candidate -- not just a WHERE filter. A Project is a valid
host (see
+ // ValidateSubqueryExpression), so pin that the rewrite fires with the
same self-join subquery
+ // inside a projected CASE WHEN. Moving it out of WHERE is the only
change from Pattern A'.
+ setupTable()
+ val subquery =
+ """SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v""".stripMargin
+ val sql =
+ s"""SELECT k, CASE WHEN k IN ($subquery) THEN 1 ELSE 0 END AS flag
+ |FROM T outer_t""".stripMargin
+
+ assertRuleFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"SELECT-list InSubquery rewrite ON $on != OFF $off")
+ }
+ }
+
+ test("Rewrite is idempotent: a second application on the rewritten plan is a
no-op") {
+ withTable("T") {
+ // After the rewrite the outer `k IN (...)` is still an InSubquery, now
over the aggregate, so
+ // the rule revisits it on later passes. Its output must be a no-op: the
aggregate no longer
+ // matches the self-join shape, so the second pass returns the plan
unchanged.
+ setupTable()
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ withSQLConf(rewriteConf -> "true") {
+ val analyzed = spark.sql(sql).queryExecution.analyzed
+ val once = RewriteSelfJoinInequalityToAggregate(analyzed)
+ assert(ruleFired(once), s"precondition: first pass should
fire:\n$once")
+ val twice = RewriteSelfJoinInequalityToAggregate(once)
+ assert(twice == once, s"rule is not idempotent:\n$once\n-- second pass
-->\n$twice")
+ }
+ }
+ }
+
+ test("Pattern A2: nested self-join is rewritten, incl. a sjRight top-Project
remap") {
+ withTable("T") {
+ withTempView("D") {
+ // Selecting `k2`, originally derived from sjRight, exercises sjRight
-> sjLeft
+ // rebinding in canonicalizeWrapper and the rebuilt output in the top
Project.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT sj.k2
+ | FROM D d, (SELECT s1.k AS k1, s2.k AS k2 FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v) sj
+ | WHERE d.k = sj.k1)""".stripMargin
+
+ assertRuleFired(sql)
+ assertMinMaxRewriteShape(optimizedPlanWith(sql, rewrite = true))
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Pattern A2 rewrite ON $on != OFF $off")
+ assert(on == Set(Row(1), Row(3), Row(6)))
+ }
+ }
+ }
+
+ test("Pattern A2: self-join on the LEFT of the outer join is rewritten") {
+ withTable("T") {
+ withTempView("D") {
+ // Mirror of the Pattern A2 test above. There the self-join is the
RIGHT child of the outer
+ // join (`selfJoinOnRight = true`); here it is the LEFT child
(`selfJoinOnRight = false`).
+ // The rule has an explicit branch for each side, so both are covered.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT d.k
+ | FROM (SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v) sj, D d
+ | WHERE sj.k = d.k)""".stripMargin
+
+ assertRuleFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Pattern A2 (self-join on left) rewrite ON $on !=
OFF $off")
+ assert(on == Set(Row(1), Row(3), Row(6)))
+ }
+ }
+ }
+
+ test("Q95-shaped query: both IN subqueries (Pattern A' and A2) are
rewritten") {
+ withTable("T") {
+ withTempView("D") {
+ // TPC-DS Q95 feeds a self-join-inequality CTE into two IN subqueries:
one selects the CTE
+ // directly (Pattern A') and one joins it with another relation
(Pattern A2). Like real Q95,
+ // the CTE also projects the two neq-side columns (wh1/wh2) that both
INs never consume, so
+ // the rewrite fires only if OptimizeSubqueries first drops them
(ColumnPruning) and merges
+ // the projections (CollapseProject). Asserting two MIN and two MAX
aliases pins both that
+ // upstream chain and that both subqueries rewrite.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+ val sql =
+ """WITH ws_wh AS (
+ | SELECT s1.k AS ordn, s1.v AS wh1, s2.v AS wh2 FROM T s1 JOIN T
s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)
+ |SELECT o.k FROM T o
+ |WHERE o.k IN (SELECT ordn FROM ws_wh)
+ | AND o.k IN (SELECT d.k FROM D d, ws_wh WHERE d.k =
ws_wh.ordn)""".stripMargin
+
+ val plan = optimizedPlanWith(sql, rewrite = true)
+ assert(
+ countAlias(plan, MinNeqAlias) == 2 && countAlias(plan, MaxNeqAlias)
== 2,
+ s"expected both IN subqueries rewritten (2 MIN + 2 MAX signature
aliases):\n$plan")
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Q95-shaped rewrite ON $on != OFF $off")
+ assert(on == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got $on")
+ }
+ }
+ }
+
+ test("Pattern A2: nondeterminism in the outer join condition must not be
rewritten") {
+ withTable("T") {
+ withTempView("D") {
+ // Both self-join sides are still repeatable here, so the per-side
`isSameBaseRelation`
+ // check would pass; the `rand()` conjunct lives on the outer join
ABOVE the self-join. Only
+ // the candidate-level `isRepeatablePlan` walk over the whole subquery
catches it, so the
+ // rule must fail closed. This is the case the candidate-level guard
exists for.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT d.k
+ | FROM D d, (SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v) sj
+ | WHERE d.k = sj.k AND rand() < 0.5)""".stripMargin
+
+ assertRuleNotFired(sql)
+ }
+ }
+ }
+
+ test("Clock-dependent STRING -> TIMESTAMP cast in the subquery is
fail-closed") {
+ withTable("TTs") {
+ // A time-only STRING cast to TIMESTAMP takes its missing date from the
runtime clock
+ // (LocalDate.now), so two self-join scans that straddle midnight can
disagree while the
+ // rewrite folds them into one evaluation. The neq column here is
derived by such a cast, so
+ // the rewrite must not fire. Isolate the cast as the sole cause with a
same-shape control
+ // whose derived neq column uses a benign INT -> BIGINT cast (which does
fire).
+ createTable(
+ "TTs",
+ "k INT, v INT, t STRING",
+ """ (1, 10, '01:00:00'), (1, 20, '02:00:00'),
+ | (2, 30, '03:00:00')""".stripMargin)
+
+ def wrapped(neqExpr: String): String =
+ s"""SELECT k FROM TTs outer_t WHERE k IN (
+ | SELECT s1.k FROM
+ | (SELECT k, $neqExpr AS nc FROM TTs) s1
+ | JOIN (SELECT k, $neqExpr AS nc FROM TTs) s2
+ | ON s1.k = s2.k AND s1.nc <> s2.nc)""".stripMargin
+
+ // Same wrapper shape with a repeatable cast fires, proving the shape
itself is supported.
+ assertRuleFired(wrapped("CAST(v AS BIGINT)"))
+
+ // A direct STRING -> TIMESTAMP cast is fail-closed.
+ assertRuleNotFired(wrapped("CAST(t AS TIMESTAMP)"))
+
+ // STRING -> TIMESTAMP_NTZ stays supported: it does not consult the
session time zone and a
+ // time-only string parses to NULL deterministically rather than
borrowing the runtime date,
+ // so the guard only rejects the clock-dependent LTZ conversion, not all
string-to-timestamp.
+ assertRuleFired(wrapped("CAST(t AS TIMESTAMP_NTZ)"))
+ }
+ }
+
+ test("Clock-dependent nested STRING -> TIMESTAMP casts (array/map/struct)
are fail-closed") {
+ withTable("TNest") {
+ // Cast.needsTimeZone recurses into ARRAY/MAP/STRUCT casts, so a
clock-dependent STRING ->
+ // TIMESTAMP_LTZ hidden at any nesting level must fail closed. The cast
rides in a WHERE
+ // filter as `CAST(nested AS <ts>) IS NOT NULL` (tree IsNotNull -> Cast
-> Attribute, all
+ // allowlisted) rather than the neq column, which the neq-column type
gate would reject
+ // wholesale and mask the recursion. So only the clock-dependent-cast
guard can
+ // decline it. For each shape the TIMESTAMP_NTZ control must fire before
the LTZ variant is
+ // required not to, so deleting a recursion arm turns the matching
negative red.
+ createTable(
+ "TNest",
+ "k INT, v INT, arr ARRAY<STRING>, mp MAP<STRING, STRING>, st STRUCT<x:
STRING>",
+ """ (1, 10, ARRAY('01:00:00'), MAP('a', '01:00:00'),
NAMED_STRUCT('x', '01:00:00')),
+ | (1, 20, ARRAY('02:00:00'), MAP('a', '02:00:00'),
NAMED_STRUCT('x', '02:00:00')),
+ | (2, 30, ARRAY('03:00:00'), MAP('a', '03:00:00'),
NAMED_STRUCT('x', '03:00:00'))"""
+ .stripMargin)
+
+ def wrapped(filterExpr: String): String =
+ s"""SELECT k FROM TNest outer_t WHERE k IN (
+ | SELECT s1.k FROM
+ | (SELECT k, v FROM TNest WHERE $filterExpr) s1
+ | JOIN (SELECT k, v FROM TNest WHERE $filterExpr) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+
+ Seq(
+ "CAST(arr AS ARRAY<TIMESTAMP>)" -> "CAST(arr AS ARRAY<TIMESTAMP_NTZ>)",
+ "CAST(mp AS MAP<STRING, TIMESTAMP>)" -> "CAST(mp AS MAP<STRING,
TIMESTAMP_NTZ>)",
+ "CAST(st AS STRUCT<x: TIMESTAMP>)" -> "CAST(st AS STRUCT<x:
TIMESTAMP_NTZ>)"
+ ).foreach { case (ltz, ntz) =>
+ assertRuleFired(wrapped(s"$ntz IS NOT NULL"))
+ assertRuleNotFired(wrapped(s"$ltz IS NOT NULL"))
+ }
+ }
+ }
+
+ test("Clock-dependent nested TIME -> TIMESTAMP casts are fail-closed") {
+ withTable("TTime") {
+ // TIME -> TIMESTAMP[_NTZ] depends on CURRENT_DATE. ComputeCurrentTime
rewrites direct
+ // scalar casts before this rule, but nested casts survive because its
TIME check does not
+ // recurse into complex types. Keep the cast in a filter so the neq type
gate cannot mask
+ // the Cast.needsTimeZone guard being tested here.
+ createTable(
+ "TTime",
+ "k INT, v INT, arr ARRAY<TIME>",
+ """ (1, 10, ARRAY(TIME'01:00:00')), (1, 20, ARRAY(TIME'02:00:00')),
+ | (2, 30, ARRAY(TIME'03:00:00'))""".stripMargin)
+
+ def wrapped(filterExpr: String): String =
+ s"""SELECT k FROM TTime outer_t WHERE k IN (
+ | SELECT s1.k FROM
+ | (SELECT k, v FROM TTime WHERE $filterExpr) s1
+ | JOIN (SELECT k, v FROM TTime WHERE $filterExpr) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+
+ // Same wrapper shape with a repeatable cast fires and computes the real
membership.
+ val safeSql = wrapped("CAST(v AS BIGINT) IS NOT NULL")
+ assertRuleFired(safeSql)
+ val onRows = withSQLConf(rewriteConf -> "true") {
spark.sql(safeSql).collect().toSeq }
+ val offRows = withSQLConf(rewriteConf -> "false") {
spark.sql(safeSql).collect().toSeq }
+ QueryTest.sameRows(onRows, offRows).foreach { error =>
+ fail(s"safe control diverges:\n$error")
+ }
+ QueryTest.sameRows(Seq(Row(1), Row(1)), onRows).foreach { error =>
+ fail(s"safe control expected two copies of Row(1):\n$error")
+ }
+
+ assertRuleNotFired(wrapped("CAST(arr AS ARRAY<TIMESTAMP>) IS NOT NULL"))
+ assertRuleNotFired(wrapped("CAST(arr AS ARRAY<TIMESTAMP_NTZ>) IS NOT
NULL"))
+ }
+ }
+
+ test("VARIANT casts are fail-closed regardless of target") {
+ // Cast.needsTimeZone(VariantType, _) is unconditionally true regardless
of the declared
+ // target -- Variant's runtime type is unknown statically, so Cast treats
any Variant-sourced
+ // cast as needing the session time zone. Delegating to it (see
isRepeatableExpression) is more
+ // conservative than a target-only check: TIMESTAMP_NTZ targets are
rejected too, not just
+ // LTZ-containing ones. A missed optimization, not a bug -- consistent
with this rule's
+ // fail-closed default. pushVariantIntoScan is disabled so the Cast
reaches the rule rather than
+ // being folded into a scan-level struct-field extraction.
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> "false") {
+ withTable("TVar") {
+ createTable(
+ "TVar",
+ "k INT, v INT, vt VARIANT",
+ """ (1, 10, parse_json('"01:00:00"')), (1, 20,
parse_json('"02:00:00"')),
+ | (2, 30, parse_json('"03:00:00"'))""".stripMargin)
+
+ def wrapped(filterExpr: String): String =
+ s"""SELECT k FROM TVar outer_t WHERE k IN (
+ | SELECT s1.k FROM
+ | (SELECT k, v FROM TVar WHERE $filterExpr) s1
+ | JOIN (SELECT k, v FROM TVar WHERE $filterExpr) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+
+ Seq(
+ "CAST(vt AS TIMESTAMP)", "CAST(vt AS TIMESTAMP_NTZ)",
+ "CAST(vt AS ARRAY<TIMESTAMP>)", "CAST(vt AS ARRAY<TIMESTAMP_NTZ>)",
+ "CAST(vt AS MAP<STRING, TIMESTAMP>)", "CAST(vt AS MAP<STRING,
TIMESTAMP_NTZ>)",
+ "CAST(vt AS STRUCT<x: TIMESTAMP>)", "CAST(vt AS STRUCT<x:
TIMESTAMP_NTZ>)"
+ ).foreach { cast =>
+ assertRuleNotFired(wrapped(s"$cast IS NOT NULL"))
+ }
+ }
+ }
+ }
+
+ test("Pattern A': multi-equi tuple IN with sjRight key remap is rewritten") {
+ withTable("TM") {
+ // Exercises the multi-equi-key path: two equi keys (k1, k2) drive the
GROUP BY, and the tuple
+ // IN projects `s1.k1, s2.k2` -- so the second output column comes from
the RIGHT self-join
+ // side and must be remapped to its sjLeft counterpart by
`canonicalizeWrapper`. Covers
+ // multiple equi keys, tuple IN arity, the two injected
IsNotNull(equiKey) filters, and the
+ // sjRight remap at once.
+ createTable(
+ "TM",
+ "k1 INT, k2 INT, v INT",
+ """ (1, 1, 10), (1, 1, 20),
+ | (1, 2, 30), (1, 2, 30),
+ | (2, 1, 40), (2, 1, 50),
+ | (CAST(NULL AS INT), 1, 60), (CAST(NULL AS INT), 1, 70),
+ | (3, CAST(NULL AS INT), 80), (3, CAST(NULL AS INT),
90)""".stripMargin)
+ val sql =
+ """SELECT k1, k2 FROM TM outer_t WHERE (k1, k2) IN (
+ | SELECT s1.k1, s2.k2 FROM TM s1 JOIN TM s2
+ | ON s1.k1 = s2.k1 AND s1.k2 = s2.k2 AND s1.v <>
s2.v)""".stripMargin
+
+ assertRuleFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"multi-equi tuple IN rewrite ON $on != OFF $off")
+ // (1,1): distinct v={10,20} -> matches; (1,2): v={30} -> no; (2,1):
v={40,50} -> matches;
+ // (NULL,1) and (3,NULL): NULL equi key filtered out by the injected
IsNotNull. ->
+ // {(1,1),(2,1)}
+ assert(on == Set(Row(1, 1), Row(2, 1)), s"expected {(1,1),(2,1)}, got
$on")
+ }
+ }
+
+ test("NULL equi-key is filtered before aggregation for NOT IN") {
+ withTable("TN") {
+ withTempView("OuterKeys") {
+ createTable(
+ "TN",
+ "k INT, v INT",
+ """ (CAST(NULL AS INT), 10),
+ | (CAST(NULL AS INT), 20),
+ | (1, 10), (1, 20),
+ | (2, 30)""".stripMargin
+ )
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW OuterKeys AS SELECT * FROM VALUES
+ | (1), (2), (3) AS OuterKeys(k)""".stripMargin)
+ val sql =
+ """SELECT k FROM OuterKeys o WHERE k NOT IN (
+ | SELECT s1.k FROM TN s1 JOIN TN s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+
+ assertRuleFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"NULL equi-key NOT IN semantics diverge: ON=$on
OFF=$off")
+ assert(on == Set(Row(2), Row(3)), s"expected {2,3}, got $on")
+ }
+ }
+ }
+
+ test("Swapped aliases must not be treated as the same self-join columns") {
+ withTable("AliasBase") {
+ // The guard under test is `sameOutputPosition`. Both queries alias the
same two base columns
+ // to the names `k` and `v` on both sides, so a rule that compares
attribute names would fire
+ // on both; only the output ordinal tells them apart.
+ createTable("AliasBase", "a INT, b INT", " (1, 10), (1, 20), (2, 30)")
+
+ val alignedSql =
+ """SELECT a FROM AliasBase outer_t WHERE a IN (
+ | SELECT s1.k
+ | FROM (SELECT a AS k, b AS v FROM AliasBase) s1
+ | JOIN (SELECT a AS k, b AS v FROM AliasBase) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(alignedSql)
+ val (alignedOn, alignedOff) = runBoth(alignedSql)
+ assert(
+ alignedOn == alignedOff,
+ s"aligned-alias control diverges: ON=$alignedOn OFF=$alignedOff")
+ assert(alignedOn == Set(Row(1)), s"aligned-alias control expected {1},
got $alignedOn")
+
+ // s1.k is `a` (output position 0) but s2.k is `b` (output position 1):
same name, different
+ // column. Rewriting this would count distinct `b` per `a`, which is a
different query.
+ val swappedSql =
+ """SELECT a FROM AliasBase outer_t WHERE a IN (
+ | SELECT s1.k
+ | FROM (SELECT a AS k, b AS v FROM AliasBase) s1
+ | JOIN (SELECT a AS v, b AS k FROM AliasBase) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(swappedSql)
+ val (on, off) = runBoth(swappedSql)
+ assert(on == off, s"swapped-alias semantics diverge: ON=$on OFF=$off")
+ assert(on.isEmpty, s"swapped-alias baseline should be empty, got $on")
+ }
+ }
+
+ test("Two different relations with the same schema must not be treated as a
self-join") {
+ withTable("TLeft", "TRight") {
+ // The guard under test is `isSameBaseRelation`: it must reject a join
between two DIFFERENT
+ // base tables even when they share a schema and column names. Distinct
Parquet tables
+ // canonicalize to distinct `rootPaths`, so `left.canonicalized ==
right.canonicalized` is
+ // false and the rewrite must not fire. This is a correctness boundary,
not a missed
+ // optimization: rewriting `TLeft JOIN TRight` as MIN(v) <> MAX(v) over
TLeft alone would drop
+ // TRight's rows and change the answer.
+ createTable("TLeft", "k INT, v INT", " (1, 10), (1, 10), (2, 30)")
+ createTable("TRight", "k INT, v INT", " (1, 20), (1, 20), (2, 30)")
+ val sql =
+ """SELECT k FROM TLeft outer_t WHERE k IN (
+ | SELECT s1.k FROM TLeft s1 JOIN TRight s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"different-relation join semantics diverge: ON=$on
OFF=$off")
+ // k=1: TLeft v={10} vs TRight v={20} -> 10<>20 true -> qualifies; k=2:
30<>30 false -> no.
+ assert(on == Set(Row(1)), s"expected {1}, got $on")
+ }
+ }
+
+ // ==================== Positive: rewritten plan is structurally the
aggregate ================
+ //
+ // Result parity (ON == OFF) does not prove the rewrite produced the GROUP
BY + HAVING
+ // MIN(v) <> MAX(v) shape rather than leaving the self-join and happening to
agree, so these
+ // controls assert the shape directly. A full plan comparison against
hand-written aggregate SQL
+ // would be wrong: InferFiltersFromConstraints adds
isnotnull(min)/isnotnull(max) to a
+ // hand-written HAVING but not to the rule's filter (a later batch), so it
would fail on redundant
+ // null predicates.
+
+ private def optimizedPlanWith(sql: String, rewrite: Boolean): LogicalPlan =
+ withSQLConf(rewriteConf -> rewrite.toString) {
+ spark.sql(sql).queryExecution.optimizedPlan
+ }
+
+ // The rewrite shape: an Aggregate emitting both signature aliases, with a
Filter on top whose
+ // condition includes Not(EqualTo(min, max)). Match the two operands by
ExprId, not by name, so a
+ // same-named attribute from elsewhere cannot satisfy it. `exists` on the
condition (not exact
+ // match) because InferFiltersFromConstraints may fold redundant
isnotnull(min/max) into it.
+ private def assertMinMaxRewriteShape(plan: LogicalPlan): Unit = {
+ val found = plan.collectFirstWithSubqueries {
+ case Filter(cond, agg: Aggregate)
+ if {
+ // Key by alias name so the shape requires exactly one MIN alias
AND one MAX alias: two
+ // same-named aliases collapse to a single map key and fail the
keySet check, which a
+ // bare `size == 2` on exprIds would not catch.
+ val signatureAttrs = agg.aggregateExpressions.collect {
+ case a: Alias if a.name == MinNeqAlias || a.name == MaxNeqAlias
=>
+ a.name -> a.toAttribute
+ }.toMap
+ signatureAttrs.keySet == Set(MinNeqAlias, MaxNeqAlias) &&
cond.exists {
+ case Not(EqualTo(l: Attribute, r: Attribute)) =>
+ Set(l.exprId, r.exprId) ==
signatureAttrs.values.map(_.exprId).toSet
+ case _ => false
+ }
+ } => ()
+ }
+ assert(found.isDefined, s"expected a MIN(v) <> MAX(v) aggregate rewrite
shape:\n$plan")
+ }
+
+ // ==================== Negative: rewrite must produce equivalent results
(or bail) ==========
+
+ test("Bare-Join subquery (no wrapper Project) fails closed: Pattern A' arity
guard") {
+ withTable("T") {
+ // The guard under test is the `projectListOpt match { case None => None
}` bail in
+ // `rewriteDirectSelfJoin`. With no wrapper Project the self-join output
is `left ++ right`;
+ // replacing it with `Project(equiKeys, aggregate)` would shrink the
arity that
+ // RewritePredicateSubquery later positionally zips against, misbinding
the semi predicates. A
+ // `SELECT *` over the self-join whose tuple IN references every output
column lets
+ // RemoveNoopOperators strip the identity Project, so the subquery
reaches the rule as a bare
+ // Join -- proven below before asserting the rule declines it.
+ setupTable()
+ val sql =
+ """SELECT k FROM T outer_t WHERE (k, v, k, v) IN (
+ | SELECT * FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ optimizedInSubqueryPlan(sql) match {
+ case _: Join =>
+ case other => fail(s"expected a bare Join subquery, got:\n$other")
+ }
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"bare-Join A' arity guard semantics diverge: ON=$on
OFF=$off")
+ }
+ }
+
+ test("Bare-Join nested self-join (no Project anywhere) fails closed: Pattern
A2 arity guard") {
+ withTable("T") {
+ withTempView("D") {
+ // The guard under test is the `case None => return None` bail in
`rewriteNestedSelfJoin`:
+ // with no wrapper Project over the self-join, changing its output
arity would misbind the
+ // positional semi-predicate zip, so the rule fails closed. `SELECT *`
plus a
+ // tuple IN over every column lets RemoveNoopOperators expose the bare
nested self-join.
+ // ReorderJoin is excluded (a valid config) so the bare nested
self-join deterministically
+ // reaches this rule rather than being reshaped away; the shape is
then asserted so the test
+ // fails loudly if it stops reaching the guard. The two sides use
renaming subqueries (ka/va
+ // vs kb/vb) so `SELECT *` yields distinct names -- a plain `T s1 JOIN
T s2` would emit two
+ // columns named `k` and fail analysis; the renames are children of
the self-join, so the
+ // identity `SELECT *` is still stripped.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+ val sql =
+ """SELECT k FROM T outer_t WHERE (k, v, k, k, v) IN (
+ | SELECT * FROM D d JOIN (
+ | SELECT * FROM (SELECT k AS ka, v AS va FROM T) s1
+ | JOIN (SELECT k AS kb, v AS vb FROM T) s2
+ | ON s1.ka = s2.kb AND s1.va <> s2.vb) sj
+ | ON d.k = sj.ka)""".stripMargin
+ withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key ->
ReorderJoin.ruleName) {
+ // A child that is a `sameResult` Inner Join carrying an inequality
is exactly the
+ // self-join rewriteNestedSelfJoin extracts; asserting it proves
execution reaches the
+ // arity guard.
+ def isTargetSelfJoin(p: LogicalPlan): Boolean = p match {
+ case j: Join if j.joinType == Inner =>
+ j.left.sameResult(j.right) &&
+ j.condition.exists(_.exists { case _: Not => true; case _ =>
false })
+ case _ => false
+ }
+ optimizedInSubqueryPlan(sql) match {
+ case j: Join if isTargetSelfJoin(j.left) ||
isTargetSelfJoin(j.right) =>
+ case other => fail(s"expected a bare nested self-join child,
got:\n$other")
+ }
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"bare-Join A2 arity guard semantics diverge:
ON=$on OFF=$off")
+ }
+ }
+ }
+ }
+
+ test("IS DISTINCT FROM is rejected by the self-join condition parser") {
+ withTable("T") {
+ setupTable()
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v IS DISTINCT FROM s2.v)""".stripMargin
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"IS DISTINCT FROM semantics diverge: ON=$on OFF=$off")
+ // Assert the full result, not just contains(4): unlike `<>`, `IS
DISTINCT FROM` treats NULL
+ // as a value, so k=4 (v={70,NULL}) qualifies alongside k=1, k=3 and k=6.
+ assert(on == Set(Row(1), Row(3), Row(4), Row(6)), s"expected {1,3,4,6},
got $on")
+ assertRuleNotFired(sql)
+ }
+ }
+
+ test("IsNotNull on a non-join column is rejected") {
+ withTable("T3") {
+ // The guard under test is the predicate parser: it accepts IsNotNull
only on a column the
+ // join condition already references, because such a predicate is
implied by the equi-key or
+ // the inequality and can be dropped, while IsNotNull(w) filters rows
the aggregate would
+ // otherwise count. The control is the same query without that one
conjunct.
+ createTable(
+ "T3",
+ "k INT, v INT, w INT",
+ """ (1, 10, 100), (1, 20, 200),
+ | (2, 30, CAST(NULL AS INT)), (2, 40, CAST(NULL AS
INT))""".stripMargin)
+
+ val controlSql =
+ """SELECT k FROM T3 outer_t WHERE k IN (
+ | SELECT s1.k FROM T3 s1 JOIN T3 s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"T3 control diverges: ON=$controlOn
OFF=$controlOff")
+ assert(controlOn == Set(Row(1), Row(2)), s"T3 control expected {1,2},
got $controlOn")
+
+ val sql =
+ """SELECT k FROM T3 outer_t WHERE k IN (
+ | SELECT s1.k FROM T3 s1 JOIN T3 s2
+ | ON s1.k = s2.k AND s1.v <> s2.v AND s1.w IS NOT
NULL)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"IsNotNull(non-join-col) semantics diverge: ON=$on
OFF=$off")
+ assert(on == Set(Row(1)), s"expected {1}, got $on")
+ }
+ }
+
+ test("Multiple inequality columns are rejected") {
+ withTable("T2") {
+ // The guard under test is `neqPairs.size != 1`. Two inequalities need
"at least two rows
+ // differing in v AND in w", which no MIN/MAX over a single column can
express. The control is
+ // the same query with only the first inequality.
+ createTable(
+ "T2",
+ "k INT, v INT, w INT",
+ """ (1, 10, 100), (1, 20, 200),
+ | (2, 30, 300),
+ | (3, 40, 100), (3, 50, 100)""".stripMargin)
+
+ val controlSql =
+ """SELECT k FROM T2 outer_t WHERE k IN (
+ | SELECT s1.k FROM T2 s1 JOIN T2 s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"T2 control diverges: ON=$controlOn
OFF=$controlOff")
+ assert(controlOn == Set(Row(1), Row(3)), s"T2 control expected {1,3},
got $controlOn")
+
+ val sql =
+ """SELECT k FROM T2 outer_t WHERE k IN (
+ | SELECT s1.k FROM T2 s1 JOIN T2 s2
+ | ON s1.k = s2.k AND s1.v <> s2.v AND s1.w <> s2.w)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"multi-column neq semantics diverge: ON=$on OFF=$off")
+ assert(on == Set(Row(1)), s"expected {1}, got $on")
+ }
+ }
+
+ test("LeftOuter self-join inside the IN subquery is rejected by the
join-type guard") {
+ withTable("T") {
+ // The guard under test is `joinType == Inner`. The MIN(v) <> MAX(v)
collapse is valid only
+ // for the INNER shape: a LEFT OUTER self-join preserves unmatched left
rows, so its subquery
+ // returns every left key; rewriting it to a per-key aggregate drops
keys and changes IN
+ // membership. The INNER control and LEFT OUTER variant differ only in
join type.
+ setupTable()
+ val controlSql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+
+ val leftOuterSql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 LEFT OUTER JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ // Confirm the optimizer keeps it LEFT OUTER, so the join-type guard is
what declines it and
+ // not some rule having reshaped the join away.
+ assert(
+ optimizedInSubqueryPlan(leftOuterSql).exists {
+ case j: Join if j.joinType == LeftOuter => true
+ case _ => false
+ },
+ "expected the subquery to still contain a LEFT OUTER join")
+ assertRuleNotFired(leftOuterSql)
+ val (on, off) = runBoth(leftOuterSql)
+ assert(on == off, s"LeftOuter-in-IN semantics diverge: ON=$on OFF=$off")
+ }
+ }
+
+ test("Join hint on the self-join is fail-closed (direct and nested)") {
+ withTable("T") {
+ withTempView("D") {
+ // The guard is `hint.isEmpty`, checked in two places:
rewriteDirectSelfJoin for a top-level
+ // self-join (Pattern A') and tryExtractSelfJoin for a self-join
nested under an outer join
+ // (Pattern A2). The rewrite deletes the self-join, so a hint on it is
a directive about a
+ // join that would vanish -- fail closed in both. A hint never changes
rows, so results are
+ // identical.
+ setupTable()
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW D AS SELECT * FROM VALUES
+ | (1), (3), (6) AS D(k)""".stripMargin)
+
+ // Pattern A': hint on the top-level self-join (rewriteDirectSelfJoin).
+ val directControl =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(directControl)
+ val directHinted =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT /*+ BROADCAST(s2) */ s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(directHinted)
+ val (dOn, dOff) = runBoth(directHinted)
+ assert(dOn == dOff, s"direct-hint semantics diverge: ON=$dOn
OFF=$dOff")
+ assert(dOn == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got
$dOn")
+
+ // Pattern A2: hint on the nested self-join (tryExtractSelfJoin).
+ val nestedControl =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT d.k
+ | FROM D d, (SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v) sj
+ | WHERE d.k = sj.k)""".stripMargin
+ assertRuleFired(nestedControl)
+ val nestedHinted =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT d.k
+ | FROM D d, (SELECT /*+ BROADCAST(s2) */ s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v) sj
+ | WHERE d.k = sj.k)""".stripMargin
+ // Confirm a hinted self-join actually survives into the subquery the
rule inspects, so the
+ // decline is attributable to the `hint.isEmpty` guard and not to the
hint reshaping the
+ // plan so that tryExtractSelfJoin never sees a candidate.
+ assert(
+ optimizedInSubqueryPlan(nestedHinted).exists {
+ case j: Join if j.left.sameResult(j.right) && !j.hint.isEmpty =>
true
+ case _ => false
+ },
+ "expected a hinted nested self-join to survive into the optimized
subquery")
+ assertRuleNotFired(nestedHinted)
+ val (nOn, nOff) = runBoth(nestedHinted)
+ assert(nOn == nOff, s"nested-hint semantics diverge: ON=$nOn
OFF=$nOff")
+ assert(nOn == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got
$nOn")
+ }
+ }
+ }
+
+ test("Config gate: rewrite disabled leaves a valid A' candidate untouched") {
+ withTable("T") {
+ setupTable()
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ withSQLConf(rewriteConf -> "false") {
+ val plan = spark.sql(sql).queryExecution.optimizedPlan
+ assert(!ruleFired(plan), s"config off must not fire rewrite:\n$plan")
+ val res = spark.sql(sql).collect().toSet
+ assert(res == Set(Row(1), Row(3), Row(6)), s"config off correctness
broken: $res")
+ }
+ }
+ }
+
+ // ==================== Correlated subquery: rule must fail-closed
====================
+
+ private def setupOuterT(): Unit = {
+ spark.sql(
+ """CREATE OR REPLACE TEMP VIEW OuterT AS SELECT * FROM VALUES
+ | (1), (3), (6) AS OuterT(k)""".stripMargin)
+ }
+
+ test("Correlated InSubquery is fail-closed") {
+ withTable("T") {
+ withTempView("OuterT") {
+ setupTable()
+ setupOuterT()
+ val sql =
+ """SELECT o.k FROM OuterT o WHERE o.k IN (
+ | SELECT s1.k FROM T s1 JOIN T s2
+ | ON s1.k = s2.k AND s1.v <> s2.v
+ | WHERE s2.k = o.k)""".stripMargin
+ // Precondition: the InSubquery is genuinely correlated (its ListQuery
carries outer
+ // references), so the not-fired result exercises the
`lq.children.isEmpty` guard in `apply`
+ // rather than an unrelated shape mismatch.
+ val analyzed = spark.sql(sql).queryExecution.analyzed
+ var sawCorrelated = false
+ analyzed.foreach { node =>
+ node.expressions.foreach(_.foreach {
+ case InSubquery(_, lq: ListQuery) => sawCorrelated ||=
lq.children.nonEmpty
+ case _ =>
+ })
+ }
+ assert(sawCorrelated, s"expected a correlated InSubquery in analyzed
plan:\n$analyzed")
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"correlated IN parity: ON=$on OFF=$off")
+ assert(on == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got $on")
+ assertRuleNotFired(sql)
+ }
+ }
+ }
+
+ // ==================== Repeatability whitelist: unknown operators
fail-closed ==============
+
+ test("Same-root Parquet leaves across an append are not the same FileIndex:
rule bails out") {
+ withTempDir { dir =>
+ val path = dir.getCanonicalPath
+ spark.sql("SELECT * FROM VALUES (1, 1), (1, 3) AS t(k, v)")
+ .write.mode("overwrite").parquet(path)
+ spark.read.parquet(path).createOrReplaceTempView("SnapBefore")
+ // A second, independent read of the same root path after an append:
`InMemoryFileIndex
+ // .equals` compares only `rootPaths`, so `sameResult` treats this pair
as the same relation
+ // even though they enumerate different files -- the precondition this
guard exists for.
+ spark.sql("SELECT 1 AS k, 2 AS v").write.mode("append").parquet(path)
+ spark.read.parquet(path).createOrReplaceTempView("SnapAfter")
+
+ def location(p: LogicalPlan): FileIndex = p.collectLeaves().collectFirst
{
+ case LogicalRelationWithTable(h: HadoopFsRelation, _) => h.location
+ }.get
+
+ val snapshotSql =
+ """SELECT k FROM SnapBefore outer_t WHERE k IN (
+ | SELECT s1.k FROM SnapBefore s1 JOIN SnapAfter s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ val analyzed = spark.sql(snapshotSql).queryExecution.analyzed
+ val join = analyzed.subqueriesAll.head.collectFirst { case j: Join => j
}.get
+ assert(join.left.sameResult(join.right),
+ s"expected sameResult to (wrongly) accept different snapshots of one
root path:\n" +
+ s"${join.left}\n${join.right}")
+ assert(!(location(join.left) eq location(join.right)),
+ "expected the two independently captured reads to have different
FileIndex instances")
+ assertRuleNotFired(snapshotSql)
+
+ // Positive control: referencing SnapBefore twice resolves it once and
shares the FileIndex
+ // (LogicalRelation.newInstance keeps the relation reference, swapping
only ExprIds), so an
+ // ordinary self-join over one already-resolved relation still fires and
computes correctly.
+ val ordinarySql =
+ """SELECT k FROM SnapBefore outer_t WHERE k IN (
+ | SELECT s1.k FROM SnapBefore s1 JOIN SnapBefore s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ val ordinaryJoin =
spark.sql(ordinarySql).queryExecution.analyzed.subqueriesAll.head
+ .collectFirst { case j: Join => j }.get
+ assert(location(ordinaryJoin.left) eq location(ordinaryJoin.right),
+ "expected an ordinary self-join to share one FileIndex instance")
+ assertRuleFired(ordinarySql)
+ val onRows = withSQLConf(rewriteConf -> "true") {
spark.sql(ordinarySql).collect().toSeq }
+ val offRows = withSQLConf(rewriteConf -> "false") {
spark.sql(ordinarySql).collect().toSeq }
+ QueryTest.sameRows(onRows, offRows).foreach { error =>
+ fail(s"ordinary self-join control diverges:\n$error")
+ }
+ QueryTest.sameRows(Seq(Row(1), Row(1)), onRows).foreach { error =>
+ fail(s"ordinary self-join control expected two copies of
Row(1):\n$error")
+ }
+ }
+ }
+
+ test("LIMIT inside the self-join subtrees breaks row-bag repeatability: rule
bails out") {
+ withTable("T") {
+ // The guard under test is the operator whitelist in
`isRowBagRepeatable`: a `LIMIT` without a
+ // total order returns an arbitrary row subset that two scans need not
agree on, so it is not
+ // row-bag repeatable. A `LIMIT` node carries only a `Literal`, so every
expression is
+ // allowlisted and `hasRepeatableExpressions` passes; with the
precondition below proving the
+ // two self-join inputs are still `sameResult`, the operator whitelist
(which does not list
+ // Limit) is the relevant remaining rejection point. A control without
`LIMIT` fires on the
+ // same shape.
+ setupTable()
+ val controlSql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM (SELECT k, v FROM T) s1
+ | JOIN (SELECT k, v FROM T) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"LIMIT control diverges: ON=$controlOn
OFF=$controlOff")
+ assert(controlOn == Set(Row(1), Row(3), Row(6)), s"expected {1,3,6}, got
$controlOn")
+
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k FROM (SELECT k, v FROM T LIMIT 2) s1
+ | JOIN (SELECT k, v FROM T LIMIT 2) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ // Precondition: both self-join inputs stay `sameResult`, ruling out a
structural mismatch,
+ // and both still contain a Limit. Since Limit carries only an
allowlisted Literal expression,
+ // the non-firing result pins the row-bag operator whitelist.
+ def containsLimit(plan: LogicalPlan): Boolean =
+ plan.exists {
+ case _: GlobalLimit | _: LocalLimit => true
+ case _ => false
+ }
+ val before = optimizedInSubqueryPlan(sql)
+ assert(
+ before.exists {
+ case j: Join if j.joinType == Inner && j.left.sameResult(j.right) &&
+ containsLimit(j.left) && containsLimit(j.right) => true
+ case _ => false
+ },
+ s"expected a sameResult self-join whose inputs both contain a
Limit:\n$before")
+ assertRuleNotFired(sql)
+ }
+ }
+
+ test("Nondeterministic self-join input is rejected") {
+ // The guard under test is `plan.deterministic` inside `isRepeatablePlan`.
Both sides use the
+ // same explicit seed, so the subplans share a canonical shape and the
rejection is not from
+ // `isSameBaseRelation`. The control replaces `rand(41) < 0.5` with a
deterministic filter,
+ // proving this Range/Filter/Project shape reaches the rewrite.
+ val controlSql =
+ """SELECT k FROM (SELECT CAST(id AS INT) AS k, CAST(id AS INT) AS v FROM
range(100)) t
+ |WHERE k IN (
+ | SELECT s1.k FROM (
+ | SELECT CAST(id % 10 AS INT) AS k, CAST(id AS INT) AS v
+ | FROM range(1000) WHERE id % 2 = 0
+ | ) s1
+ | JOIN (
+ | SELECT CAST(id % 10 AS INT) AS k, CAST(id AS INT) AS v
+ | FROM range(1000) WHERE id % 2 = 0
+ | ) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"range control diverges: ON=$controlOn
OFF=$controlOff")
+ assert(
+ controlOn == Set(Row(0), Row(2), Row(4), Row(6), Row(8)),
+ s"range control expected the even keys, got $controlOn")
+
+ val sql =
+ """SELECT k FROM (SELECT CAST(id AS INT) AS k, CAST(id AS INT) AS v FROM
range(100)) t
+ |WHERE k IN (
+ | SELECT s1.k FROM (
+ | SELECT CAST(id % 10 AS INT) AS k, CAST(id AS INT) AS v
+ | FROM range(1000) WHERE rand(41) < 0.5
+ | ) s1
+ | JOIN (
+ | SELECT CAST(id % 10 AS INT) AS k, CAST(id AS INT) AS v
+ | FROM range(1000) WHERE rand(41) < 0.5
+ | ) s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(sql)
+ }
+
+ test("LogicalRDD leaf is not a trusted repeatable source: rule bails out") {
+ withTable("RddCtl") {
+ withTempView("RddT") {
+ // The guard under test is the leaf allowlist in `isRowBagRepeatable`:
a Parquet
+ // `LogicalRelation` is trusted, but a `LogicalRDD` (createDataFrame
over an RDD) wraps an
+ // arbitrary RDD lineage whose runtime row bag Catalyst cannot prove
repeatable, so it must
+ // fail closed even though `plan.deterministic` is true. The Parquet
control uses the same
+ // schema, data and query shape and fires, pinning the rejection to
the leaf allowlist.
+ createTable("RddCtl", "k INT, v INT", " (1, 10), (1, 20), (2, 30)")
+ val controlSql =
+ """SELECT k FROM RddCtl outer_t WHERE k IN (
+ | SELECT s1.k FROM RddCtl s1 JOIN RddCtl s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"Parquet control diverges:
ON=$controlOn OFF=$controlOff")
+ assert(controlOn == Set(Row(1)), s"Parquet control expected {1}, got
$controlOn")
+
+ val schema = StructType(Seq(StructField("k", IntegerType),
StructField("v", IntegerType)))
+ val rows = spark.sparkContext.parallelize(Seq(Row(1, 10), Row(1, 20),
Row(2, 30)))
+ spark.createDataFrame(rows, schema).createOrReplaceTempView("RddT")
+ val sql =
+ """SELECT k FROM RddT outer_t WHERE k IN (
+ | SELECT s1.k FROM RddT s1 JOIN RddT s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"LogicalRDD semantics diverge: ON=$on OFF=$off")
+ assert(on == Set(Row(1)), s"expected {1}, got $on")
+ }
+ }
+ }
+
+ test("Non-allowlisted deterministic expression (Abs) fails closed") {
+ withTable("T") {
+ // The guard under test is the expression allowlist in
`isRepeatableExpression`: it trusts
+ // expression TYPES, not merely `deterministic`. Abs is deterministic
but not (yet)
+ // allowlisted, so a self-join side projecting abs(v) fails closed -- a
missed optimization,
+ // not a bug. The control uses `v + 1` (not `+ 0`, so the Add survives
arithmetic
+ // simplification) and fires; wrapping the same column in abs() is the
sole change and makes
+ // it not fire. v is INT, so the Add is a plain `Add(v, 1)` with no
decimal PromotePrecision /
+ // CheckOverflow wrappers.
+ setupTable()
+
+ val controlSql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k
+ | FROM (SELECT k, v + 1 AS x FROM T) s1
+ | JOIN (SELECT k, v + 1 AS x FROM T) s2
+ | ON s1.k = s2.k AND s1.x <> s2.x)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"Add control diverges: ON=$controlOn
OFF=$controlOff")
+ // v+1 is injective over the (non-null) v values, so distinctness per k
is unchanged: {1,3,6}.
+ assert(
+ controlOn == Set(Row(1), Row(3), Row(6)),
+ s"Add control expected {1,3,6}, got $controlOn")
+
+ val sql =
+ """SELECT k FROM T outer_t WHERE k IN (
+ | SELECT s1.k
+ | FROM (SELECT k, abs(v) AS x FROM T) s1
+ | JOIN (SELECT k, abs(v) AS x FROM T) s2
+ | ON s1.k = s2.k AND s1.x <> s2.x)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Abs-projected self-join semantics diverge: ON=$on
OFF=$off")
+ }
+ }
+
+ // ==================== Data-type safety: comparison vs grouping/MIN-MAX
equality =============
+ //
+ // The rewrite turns `<>` into MIN(v) <> MAX(v) and `=` into GROUP BY, so it
is only sound where
+ // comparison equality coincides with grouping/MIN-MAX ordering equality.
The two roles differ: an
+ // equi key needs grouping equality (binary equality for strings), the neq
column needs MIN/MAX
+ // ordering equality (binary ordering for strings). Both are positive
allowlists, not
+ // `RowOrdering.isOrderable`. These tests pin the boundary for the risky
types.
+
+ test("Float/Double neq column is rejected (comparison-vs-MIN-MAX contract,
defensive)") {
+ withTable("TFloat") {
+ // This pins the neq-column type gate: `vi` (Int, allowlisted) fires,
`vd` (Double) does not.
+ // Double fails closed defensively -- comparison vs grouping/MIN-MAX
agreement on signed zero
+ // and NaN rests on normalization details (NormalizeFloatingNumbers)
that need not match
+ // across Spark versions or native backends. Current Spark aligns them,
so OFF is {1} and the
+ // non-firing ON matches it: the test pins fail-closed behavior, not a
divergence.
+ createTable(
+ "TFloat",
+ "k INT, vi INT, vd DOUBLE",
+ """ (1, 10, 1.0), (1, 20, 2.0),
+ | (2, 30, 0.0), (2, 30, -0.0),
+ | (3, 40, CAST('NaN' AS DOUBLE)), (3, 50, CAST('NaN' AS
DOUBLE))""".stripMargin)
+
+ val controlSql =
+ """SELECT k FROM TFloat outer_t WHERE k IN (
+ | SELECT s1.k FROM TFloat s1 JOIN TFloat s2
+ | ON s1.k = s2.k AND s1.vi <> s2.vi)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"Int-neq control diverges:
ON=$controlOn OFF=$controlOff")
+ assert(controlOn == Set(Row(1), Row(3)), s"Int-neq control expected
{1,3}, got $controlOn")
+
+ val sql =
+ """SELECT k FROM TFloat outer_t WHERE k IN (
+ | SELECT s1.k FROM TFloat s1 JOIN TFloat s2
+ | ON s1.k = s2.k AND s1.vd <> s2.vd)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Double-neq semantics diverge: ON=$on OFF=$off")
+ // k=1 matches (1.0 <> 2.0); k=2 does not (0.0 = -0.0); k=3 does not
(Spark NaN = NaN).
+ assert(on == Set(Row(1)), s"Double-neq baseline expected {1}, got $on")
+ }
+ }
+
+ test("Float/Double equi-key is rejected (defensive fail-closed)") {
+ withTable("TFloatKey") {
+ // This pins the equi-key type gate: `ki` (Int) fires, `kd` (Double)
does not. Current Spark
+ // aligns floating-point comparison with grouping normalization here,
but the rule does not
+ // depend on that implementation contract, so it fails closed.
+ createTable(
+ "TFloatKey",
+ "kd DOUBLE, ki INT, v INT",
+ """ (1.0, 1, 10), (1.0, 1, 20),
+ | (2.0, 2, 30),
+ | (0.0, 3, 40), (-0.0, 3, 50)""".stripMargin)
+
+ val controlSql =
+ """SELECT ki FROM TFloatKey outer_t WHERE ki IN (
+ | SELECT s1.ki FROM TFloatKey s1 JOIN TFloatKey s2
+ | ON s1.ki = s2.ki AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"Int-key control diverges:
ON=$controlOn OFF=$controlOff")
+ assert(controlOn == Set(Row(1), Row(3)), s"Int-key control expected
{1,3}, got $controlOn")
+
+ val sql =
+ """SELECT kd FROM TFloatKey outer_t WHERE kd IN (
+ | SELECT s1.kd FROM TFloatKey s1 JOIN TFloatKey s2
+ | ON s1.kd = s2.kd AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"Double-key semantics diverge: ON=$on OFF=$off")
+ }
+ }
+
+ test("Complex-type neq column (array/struct) is rejected wholesale") {
+ withTable("TCplx") {
+ // The neq-column type gate rejects complex types wholesale, which also
covers any
+ // Float/Double nested inside them.
+ // Control (`v` Int) fires; the ARRAY<DOUBLE> and STRUCT<..DOUBLE>
variants -- identical query
+ // shape, only the neq column changed -- do not.
+ createTable(
+ "TCplx",
+ "k INT, v INT, a ARRAY<DOUBLE>, s STRUCT<x: INT, y: DOUBLE>",
+ """ (1, 10, ARRAY(1.0), NAMED_STRUCT('x', 1, 'y', 1.0)),
+ | (1, 20, ARRAY(2.0), NAMED_STRUCT('x', 2, 'y', 2.0)),
+ | (2, 30, ARRAY(1.0), NAMED_STRUCT('x', 1, 'y',
1.0))""".stripMargin)
+
+ val controlSql =
+ """SELECT k FROM TCplx outer_t WHERE k IN (
+ | SELECT s1.k FROM TCplx s1 JOIN TCplx s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(controlSql)
+ val (controlOn, controlOff) = runBoth(controlSql)
+ assert(controlOn == controlOff, s"Int-neq control diverges:
ON=$controlOn OFF=$controlOff")
+ assert(controlOn == Set(Row(1)), s"Int-neq control expected {1}, got
$controlOn")
+
+ val arraySql =
+ """SELECT k FROM TCplx outer_t WHERE k IN (
+ | SELECT s1.k FROM TCplx s1 JOIN TCplx s2
+ | ON s1.k = s2.k AND s1.a <> s2.a)""".stripMargin
+ assertRuleNotFired(arraySql)
+ val (arrayOn, arrayOff) = runBoth(arraySql)
+ assert(arrayOn == arrayOff, s"array-neq semantics diverge: ON=$arrayOn
OFF=$arrayOff")
+
+ val structSql =
+ """SELECT k FROM TCplx outer_t WHERE k IN (
+ | SELECT s1.k FROM TCplx s1 JOIN TCplx s2
+ | ON s1.k = s2.k AND s1.s <> s2.s)""".stripMargin
+ assertRuleNotFired(structSql)
+ val (structOn, structOff) = runBoth(structSql)
+ assert(structOn == structOff, s"struct-neq semantics diverge:
ON=$structOn OFF=$structOff")
+ }
+ }
+
+ test("Every remaining allowed type family fires as both equi-key and neq
column") {
+ // Int and String are covered elsewhere; these have no firing test at all.
Same type for k
+ // and v exercises isSafeEquiKeyType and isSafeNeqColumnType together.
+ case class TypeCase(sqlType: String, lo: String, hi: String)
+ val cases = Seq(
+ TypeCase("DECIMAL(10,2)", "100.50", "200.75"),
+ TypeCase("BOOLEAN", "true", "false"),
+ TypeCase("DATE", "DATE'2024-01-01'", "DATE'2024-06-01'"),
+ TypeCase("TIMESTAMP", "TIMESTAMP'2024-01-01 00:00:00'",
"TIMESTAMP'2024-06-01 00:00:00'"),
+ TypeCase(
+ "TIMESTAMP_NTZ",
+ "CAST(TIMESTAMP'2024-01-01 00:00:00' AS TIMESTAMP_NTZ)",
+ "CAST(TIMESTAMP'2024-06-01 00:00:00' AS TIMESTAMP_NTZ)"),
+ TypeCase("BINARY", "X'AA'", "X'BB'"))
+
+ cases.foreach { c =>
+ withTable("TTypes") {
+ createTable(
+ "TTypes",
+ s"k ${c.sqlType}, v ${c.sqlType}",
+ s""" (${c.lo}, ${c.lo}), (${c.lo}, ${c.hi}),
+ | (${c.hi}, ${c.lo})""".stripMargin)
+ // Project a constant, not `k`: `k` still exercises the equi-key gate
and `v` the
+ // neq-column gate, while keeping the raw value under test (e.g.
BINARY's Array[Byte])
+ // out of runBoth's Set[Row].
+ val sql =
+ """SELECT 1 AS matched FROM TTypes outer_t WHERE k IN (
+ | SELECT s1.k FROM TTypes s1 JOIN TTypes s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"${c.sqlType} control diverges: ON=$on OFF=$off")
+ assert(on == Set(Row(1)), s"${c.sqlType} unexpected result: $on")
+ }
+ }
+ }
+
+ test("String neq/equi key: default (UTF8_BINARY) fires, non-binary collation
fails closed") {
+ withTable("TStrBin", "TStrCiNeq", "TStrCiEqui") {
+ // On strings the two type gates diverge: an equi key is admitted only
under binary equality
+ // and a neq column only under binary ordering (both byte-wise), because
a non-binary
+ // collation (Spark 4.0+) routes comparison and grouping through
different code paths.
+ // UTF8_LCASE satisfies neither, so it is rejected in either role; the
default-collation
+ // control fires. The two negatives collate exactly ONE column each so
each pins its gate: a
+ // UTF8_LCASE NEQ column the neq-side check, a UTF8_LCASE EQUI key the
equi-side check.
+ // Collating both at once would leave it ambiguous which gate fired.
+ createTable(
+ "TStrBin",
+ "k STRING, v STRING",
+ """ ('a', 'x'), ('a', 'y'),
+ | ('b', 'z')""".stripMargin)
+ val binSql =
+ """SELECT k FROM TStrBin outer_t WHERE k IN (
+ | SELECT s1.k FROM TStrBin s1 JOIN TStrBin s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(binSql)
+ val (binOn, binOff) = runBoth(binSql)
+ assert(binOn == binOff, s"binary-string control diverges: ON=$binOn
OFF=$binOff")
+ assert(binOn == Set(Row("a")), s"binary-string control expected {a}, got
$binOn")
+
+ // Negative 1: only the NEQ column is non-binary collated -> neq-side
type gate rejects.
+ createTable(
+ "TStrCiNeq",
+ "k STRING, v STRING COLLATE UTF8_LCASE",
+ """ ('a', 'x'), ('a', 'y'),
+ | ('b', 'z')""".stripMargin)
+ val ciNeqSql =
+ """SELECT k FROM TStrCiNeq outer_t WHERE k IN (
+ | SELECT s1.k FROM TStrCiNeq s1 JOIN TStrCiNeq s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(ciNeqSql)
+ val (neqOn, neqOff) = runBoth(ciNeqSql)
+ assert(neqOn == neqOff, s"collated-neq semantics diverge: ON=$neqOn
OFF=$neqOff")
+
+ // Negative 2: only the EQUI key is non-binary collated -> equi-side
type gate rejects.
+ createTable(
+ "TStrCiEqui",
+ "k STRING COLLATE UTF8_LCASE, v STRING",
+ """ ('a', 'x'), ('a', 'y'),
+ | ('b', 'z')""".stripMargin)
+ val ciEquiSql =
+ """SELECT k FROM TStrCiEqui outer_t WHERE k IN (
+ | SELECT s1.k FROM TStrCiEqui s1 JOIN TStrCiEqui s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(ciEquiSql)
+ val (equiOn, equiOff) = runBoth(ciEquiSql)
+ assert(equiOn == equiOff, s"collated-equi semantics diverge: ON=$equiOn
OFF=$equiOff")
+ }
+ }
+
+ test("CHAR/VARCHAR join keys are rejected via declared-type metadata") {
+ withTable("TStr", "TCharVarchar") {
+ // CHAR/VARCHAR columns reach the optimizer as annotated StringType
(CharVarcharUtils records
+ // the declared type in the metadata). Recovering the declared type from
the metadata keeps
+ // them out of the StringType allowlist, where a dataType-only check
would admit them. The
+ // plain-STRING control fires while the CHAR(5) and VARCHAR(5) variants
fail closed. (Both `k`
+ // and `v` are CHAR/VARCHAR, so this exercises the metadata read, not
which gate rejects.)
+ createTable(
+ "TStr",
+ "k STRING, v STRING",
+ """ ('a', 'x'), ('a', 'y'),
+ | ('b', 'z')""".stripMargin)
+ val stringSql =
+ """SELECT k FROM TStr outer_t WHERE k IN (
+ | SELECT s1.k FROM TStr s1 JOIN TStr s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleFired(stringSql)
+ val (strOn, strOff) = runBoth(stringSql)
+ assert(strOn == strOff, s"string control diverges: ON=$strOn
OFF=$strOff")
+ assert(strOn == Set(Row("a")), s"string control expected {a}, got
$strOn")
+
+ Seq("CHAR(5)", "VARCHAR(5)").foreach { keyType =>
+ createTable(
+ "TCharVarchar",
+ s"k $keyType, v $keyType",
+ """ ('a', 'x'), ('a', 'y'),
+ | ('b', 'z')""".stripMargin)
+ val sql =
+ """SELECT k FROM TCharVarchar outer_t WHERE k IN (
+ | SELECT s1.k FROM TCharVarchar s1 JOIN TCharVarchar s2
+ | ON s1.k = s2.k AND s1.v <> s2.v)""".stripMargin
+ assertRuleNotFired(sql)
+ val (on, off) = runBoth(sql)
+ assert(on == off, s"$keyType not-fired query diverges: ON=$on
OFF=$off")
+ }
+ }
+ }
+
+ test("ANSI: rewrite preserves observable error behavior (throw-or-succeed
parity)") {
+ withTable("TAnsi") {
+ // Not a rejection but a parity property: the allowlist admits Cast,
which can throw under
+ // ANSI. It evaluates `CAST(s AS INT)` once per row inside the
Aggregate, the baseline
+ // self-join per row on each side -- the same rows -- so a malformed
value must make
+ // BOTH forms behave the same (both succeed with equal rows, or both
throw the same error
+ // class; a one-sided throw is a blocker). k=2 holds a non-numeric 's'.
+ createTable(
+ "TAnsi",
+ "k INT, s STRING",
+ """ (1, '10'), (1, '20'),
+ | (2, '30'), (2, 'xyz')""".stripMargin)
+ val sql =
+ """SELECT k FROM TAnsi outer_t WHERE k IN (
+ | SELECT s1.k
+ | FROM (SELECT k, CAST(s AS INT) AS x FROM TAnsi) s1
+ | JOIN (SELECT k, CAST(s AS INT) AS x FROM TAnsi) s2
+ | ON s1.k = s2.k AND s1.x <> s2.x)""".stripMargin
+
+ Seq("false", "true").foreach { ansi =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi) {
+ // The rule fires at plan level regardless of ANSI (the cast throws
only at runtime).
+ assertRuleFired(sql)
+ val on = runOutcome(sql, rewrite = true)
+ val off = runOutcome(sql, rewrite = false)
+ // Force the expected outcome shape by ansi instead of matching on
whichever shape both
+ // sides happen to agree on: under ansi=false a (Left, Left) match
on the same error class
+ // would otherwise pass without ever checking the real membership
{1}.
+ if (ansi == "false") {
+ (on, off) match {
+ case (Right(onRows), Right(offRows)) =>
+ QueryTest.sameRows(onRows, offRows).foreach { error =>
+ fail(s"ANSI=false both succeeded but diverged:\n$error")
+ }
+ // Real membership: both TAnsi rows with k=1, not just
"matches OFF".
+ QueryTest.sameRows(Seq(Row(1), Row(1)), onRows).foreach {
error =>
+ fail(s"ANSI=false expected two copies of Row(1):\n$error")
+ }
+ case _ =>
+ fail(s"ANSI=false expected both sides to succeed: ON=$on
OFF=$off")
+ }
+ } else {
+ (on, off) match {
+ case (Left(onErr), Left(offErr)) =>
+ assert(onErr == offErr,
+ s"ANSI=true both threw but different error: ON=$onErr
OFF=$offErr")
+ assert(onErr == "CAST_INVALID_INPUT",
+ s"ANSI=true expected CAST_INVALID_INPUT, got $onErr")
+ case _ =>
+ fail(s"ANSI=true expected both sides to throw: ON=$on
OFF=$off")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("ANSI: Remainder neq column preserves error behavior") {
+ withTable("TAnsiDiv") {
+ // Cast is not the only allowlisted expression that can throw under
ANSI. Remainder (`%`) on
+ // INT operands stays INT (allowlisted), so the rule fires, and it can
raise REMAINDER_BY_ZERO
+ // under ANSI -- exercising throw-parity directly rather than resting on
the Cast test alone.
+ // Row (2, 30, 0) forces `30 % 0`, which throws under ANSI and yields
NULL otherwise.
+ createTable(
+ "TAnsiDiv",
+ "k INT, v INT, w INT",
+ """ (1, 10, 3), (1, 20, 7),
+ | (2, 30, 0), (2, 40, 4)""".stripMargin)
+
+ // Remainder: INT result -> fires; assert throw-or-succeed parity under
ANSI off and on.
+ val remSql =
+ """SELECT k FROM TAnsiDiv outer_t WHERE k IN (
+ | SELECT s1.k
+ | FROM (SELECT k, v % w AS x FROM TAnsiDiv) s1
+ | JOIN (SELECT k, v % w AS x FROM TAnsiDiv) s2
+ | ON s1.k = s2.k AND s1.x <> s2.x)""".stripMargin
+ Seq("false", "true").foreach { ansi =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi) {
+ // The rule fires at plan level regardless of ANSI (the remainder
throws only at runtime).
+ assertRuleFired(remSql)
+ val on = runOutcome(remSql, rewrite = true)
+ val off = runOutcome(remSql, rewrite = false)
+ // Force the expected outcome shape by ansi; see the Cast test above
for why matching on
+ // whichever shape both sides agree on would leave the real
membership unchecked.
+ if (ansi == "false") {
+ (on, off) match {
+ case (Right(onRows), Right(offRows)) =>
+ QueryTest.sameRows(onRows, offRows).foreach { error =>
+ fail(s"Remainder ANSI=false both succeeded but
diverged:\n$error")
+ }
+ // k=1: 10%3=1, 20%7=6 -> {1,6} matches; k=2: 30%0=NULL,
40%4=0 -> {0} no match.
+ QueryTest.sameRows(Seq(Row(1), Row(1)), onRows).foreach {
error =>
+ fail(s"Remainder ANSI=false expected two copies of
Row(1):\n$error")
+ }
+ case _ =>
+ fail(s"Remainder ANSI=false expected both sides to succeed:
ON=$on OFF=$off")
+ }
+ } else {
+ (on, off) match {
+ case (Left(onErr), Left(offErr)) =>
+ assert(onErr == offErr,
+ s"Remainder ANSI=true both threw but different error:
ON=$onErr OFF=$offErr")
+ assert(onErr == "REMAINDER_BY_ZERO",
+ s"Remainder ANSI=true expected REMAINDER_BY_ZERO, got
$onErr")
+ case _ =>
+ fail(s"Remainder ANSI=true expected both sides to throw:
ON=$on OFF=$off")
+ }
+ }
+ }
+ }
+ }
+ }
+
+ test("ANSI: NULL equi-key row does not expose the neq cast to a different
filter placement") {
+ withTable("TAnsiNull") {
+ // Our injected `IsNotNull(k)` and the baseline's own
`InferFiltersFromConstraints` (inferred
+ // from `s1.k = s2.k`) are different mechanisms that could place the
guard differently
+ // relative to the throwing `CAST(s AS INT)`. Both end up pushing it
below the Project before
+ // the cast runs (Cast.nullIntolerant), on either side of ANSI -- pin
that so it stays true.
Review Comment:
**Finding 7.** The two mechanisms are not equivalent, so this parity is
conditional and the comment states it as unconditional.
Yours is unconditional. The baseline's comes from
`InferFiltersFromConstraints`, which returns the plan untouched when
`spark.sql.constraintPropagation.enabled` is false
(`sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:1923-1929`).
With that conf off the baseline never gets an `IsNotNull(k)` to push below the
Project, so the throwing cast runs for the NULL-key row, while the rewrite has
already dropped it.
I ran this test's own fixture and query in a review worktree at this head,
adding only the conf to the loop:
ANSI=true constraintPropagation=true ON=Right([1],[1])
OFF=Right([1],[1])
ANSI=true constraintPropagation=false ON=Right([1],[1])
OFF=Left(CAST_INVALID_INPUT)
The rewrite is not wrong here. It suppresses an error rather than returning
a row the filter rejects, and Spark promises nothing about when an ANSI error
is raised. The problem is that the test asserts more than what holds. Either
add `CONSTRAINT_PROPAGATION_ENABLED` to the loop and assert the outcome each
setting actually produces, or say in the comment that the parity rests on that
rule having run.
--
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]
