cloud-fan commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4346076647
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
+ }
+ if (carried.isEmpty) {
+ return join
+ }
+ // Where the value of a carried IN is observed beyond being TRUE, an
existence join cannot
+ // stand in for it, exactly as on the routes above.
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+
+ val rejected = carried.filter { sq =>
+ observesUnknown(sq) ||
+ (referencesInput(sq, join.left) && referencesInput(sq, join.right))
+ }
+ if (rejected.nonEmpty) {
+ // A sub-query that needs both inputs can be rewritten against neither,
and one whose
+ // unknown is observable cannot become an `exists` attribute at all.
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(rejected)
+ }
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ carried.filter(sq => isInput(sq) &&
!rejected.contains(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
rewriteExistentialExprWithAttrs(Seq(sq), p)
Review Comment:
**Blocking (P1):** `carried` holds the `InSubquery` under a `NOT IN`, not
the `Not`, so this call rewrites it through the positive-IN arm (`value =
output AND hoisted`) instead of the null-aware `Not(InSubquery)` arm, and the
`transform` below replaces only the IN, leaving `NOT exists` over a condition
that ignores NULL comparisons. For example, with t1(a) = 1, 2; t2(c1) = 5;
t4(col1, k) = (5, 100); t5(x, y) = (NULL, 100):
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
SELECT col1 FROM t4 WHERE k = c1 OR col1 NOT IN (SELECT x FROM t5 WHERE
y = k)))
```
`5 NOT IN (NULL)` is unknown, so standard SQL excludes the t4 row and
returns no rows. Here `5 = NULL` never matches, `exists` is FALSE, `NOT exists`
is TRUE, and both rows of t1 come back. At the merge base this shape failed
with INTERNAL_ERROR.
The same unit mismatch affects the rejection: `unknownSensitiveInSubqueries`
reports the `Not` node for a NOT IN, but `observesUnknown` compares it with the
bare `InSubquery` in `carried`, so a carried NOT IN in a value position such as
`(col1 NOT IN (...)) <=> false` is routed instead of rejected.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
+ }
+ if (carried.isEmpty) {
+ return join
+ }
+ // Where the value of a carried IN is observed beyond being TRUE, an
existence join cannot
+ // stand in for it, exactly as on the routes above.
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+
+ val rejected = carried.filter { sq =>
+ observesUnknown(sq) ||
+ (referencesInput(sq, join.left) && referencesInput(sq, join.right))
+ }
+ if (rejected.nonEmpty) {
+ // A sub-query that needs both inputs can be rewritten against neither,
and one whose
+ // unknown is observable cannot become an `exists` attribute at all.
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(rejected)
Review Comment:
**Non-blocking (P2):** `rejected` mixes the two rejection reasons, but this
always raises `NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY`, whose
message says the sub-query references both the outer query and the sub-query it
is nested in. A carried IN rejected only because its unknown result is
observable references neither, so users and tools matching `getCondition` get
the wrong sub-condition and a false explanation. For example, with a nullable
`t5n.x`:
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
SELECT col1 FROM t4 WHERE k = c1 OR ((col1 IN (SELECT x FROM t5n WHERE x
= k)) <=> false)))
```
The level-3 IN references only t4, so the first-level guard skips it; it is
carried by the existence join on t2 and rejected here. `a > 5 OR EXISTS (SELECT
1 FROM t2 WHERE a = c1 OR ((c1 IN (SELECT col1 FROM t4n WHERE k = c1)) <=>
false))` reaches the same throw through the Filter-predicate arm, while the
same disjunct under a top-level EXISTS raises
`NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT`, as SubquerySuite asserts. No test
reaches this throw.
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
Review Comment:
**Blocking (P1):** Only carried sub-queries with a non-empty hoisted
condition are collected here, so an uncorrelated one stays in the grafted
condition. The scaladoc above assumes `InSubqueryExec` then evaluates it
three-valued, but with `decorrelatePredicateSubqueriesInJoinPredicate.enabled`
and `optimizeUncorrelatedInSubqueriesInJoinCondition.enabled` at their default
`true`, the top-down pass reaches this existence join in the `case j: Join`
arm, which rewrites the IN against the input it references into a two-valued
`exists`. The first-level unknown guard did not catch it either, because the
sub-query plan does not own it. This is the second case from r4281444233, with
t1(a) = 1, 2, 3; t2(c1) = 1, 8, 9; t4(col1, k) = (9, 7); t3n(x) = 3, NULL:
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
SELECT col1 FROM t4 WHERE k = c1 OR ((col1 IN (SELECT x FROM t3n)) <=>
false)))
```
`9 IN (3, NULL)` is unknown, so `<=> false` is FALSE, and standard SQL
returns only 1. Here `exists <=> false` is TRUE for col1 = 9, the level-2 IN
holds for c1 = 9, and every row of t1 comes back. With the decorrelation conf
disabled the residual is planned three-valued and the answer is 1, so the two
configurations disagree on this shape.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#issuecomment-6332638268).
See **Shared repair plan 1** in the review body.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -184,8 +188,20 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
// Use EXISTS if performance matters to you.
// Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val dedupSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
+ val inConditions = values.zip(dedupSub.output).map(EqualTo.tupled)
+ // The hoisted correlated predicates may carry existential
sub-queries that only the
+ // sub-query plan can evaluate, see
rewriteExistentialExprInJoinCondition.
+ // This arm returns a bare Join, which this rule does not offer to
its own handling of
Review Comment:
**Nit (P3):** This only holds while NOT IN is the last sub-query conjunct.
The Filter arm folds every conjunct with `withSubquery.foldLeft`, and a later
EXISTS or IN arm wraps this bare anti join as `p` in `Project(p.output, Join(p,
...))`, so the top-down pass reaches it in the `case j: Join` arm, which under
the default configuration rewrites the residual into a two-valued `exists`.
With t1(a) = 1, 2, 3; t2(c1) = 1, 8, 9; t3n(col1) = 3, NULL:
```sql
SELECT * FROM t1
WHERE a NOT IN (SELECT c1 FROM t2 WHERE c1 > a OR (c1 IN (SELECT col1 FROM
t3n)) <=> false)
AND EXISTS (SELECT 1 FROM t2 WHERE c1 >= a)
```
drops row 1, where standard SQL returns 1, 2, 3. The merge base built the
same plan, so the behavior is not new; the comment's claim that the residual
stays three-valued is.
**Verification:**
- **Inspection:** The comment's stated safety condition matches the fold: a
NOT IN followed by another sub-query conjunct is described as reachable by the
join-condition rewrite.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3224,625 @@ class SubquerySuite extends SharedSparkSession
}
}
}
+
+ /**
+ * Checks the result of `query`, that every node of its optimized plan can
produce the attributes
+ * it references, and that the existence join a nested subquery was
rewritten into sits on the
+ * subquery side of the semi/anti join.
+ *
+ * The plan checks are what SPARK-59351 is about: an existence join whose
condition references an
+ * attribute produced by neither of its children can still return the right
answer when the
+ * nested relation is empty at runtime, because the invalid condition is
then never bound, and a
+ * rewrite that placed the join on the other side would return the right
answer here too. Note
+ * that missingInput cannot see a subquery expression that survives in a
condition, since
+ * QueryPlan.references folds only the outer attributes of one and collect
does not walk its
+ * plan, so it is not a substitute for asserting the shape.
+ */
+ private def checkAnswerAndPlan(
+ query: String,
+ expected: Seq[Row],
+ existenceJoinOnSubqueryPlan: Boolean = true): Unit = {
+ val df = sql(query)
+ val plan = df.queryExecution.optimizedPlan
+ val invalidNodes = plan.collect { case p if p.missingInput.nonEmpty => p }
+ assert(invalidNodes.isEmpty,
+ s"""Plan nodes reference non-reachable attributes:
+ |${invalidNodes.mkString("\n")}
+ |$plan""".stripMargin)
+ if (existenceJoinOnSubqueryPlan) {
+ val semiOrAnti = plan.collectFirst {
+ case j: Join if j.joinType == LeftSemi || j.joinType == LeftAnti => j
+ }
+ assert(semiOrAnti.nonEmpty, s"expected a semi or anti join in\n$plan")
+ val existenceJoins = plan.collect {
+ case j: Join if j.joinType.isInstanceOf[ExistenceJoin] => j
+ }
+ assert(existenceJoins.nonEmpty, s"expected an existence join in\n$plan")
+ // The existence join must be under the right side of the semi/anti
join, which is the
+ // subquery plan, and not under its left side, which is the outer plan.
+ assert(existenceJoins.forall(j =>
semiOrAnti.get.right.exists(_.fastEquals(j))),
+ s"the existence join must sit on the subquery side in\n$plan")
+ }
+ checkAnswer(df, expected)
+ }
+
+ test("SPARK-59351: nested subquery referencing the inner query becomes an
existence join") {
+ // SPARK-45580 covers the case where the nested subquery references the
outer query, in which
+ // case its existence join is built on top of the outer plan. Here the
nested subquery
+ // references the query it is nested in, so the existence join has to be
built on top of the
+ // subquery plan instead.
+ withTempView("t1", "t2", "t3", "t3n", "t4", "t5") {
+ Seq((1), (2), (3), (7),
(9)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ // t3 shares a value with t2, t5 does not, so the nested subquery
decides the answer below.
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ Seq((3), (7)).toDF("col1").persist().createOrReplaceTempView("t5")
+ Seq(Some(3), Some(9),
None).toDF("col1").persist().createOrReplaceTempView("t3n")
+ // A correlated nested IN over t4 is false for every c1 in t2, while the
same IN without its
+ // correlated predicate is true for 1 and 9, so the correlation decides
the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // EXISTS rewritten as a left semi join. The correlated predicate is a
disjunction, so it
+ // is pulled up as a whole and carries the nested IN-subquery, which
references c1, out of
+ // the subquery plan.
+ val query1 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query1, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // Same over t5, which shares no value with t2, so the nested IN is
false for every c1 and
+ // only the correlated predicate can hold: a mistake making it true
would return every row.
+ val query2 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query2, Row(1) :: Row(9) :: Nil)
+
+ // NOT EXISTS rewritten as a left anti join.
+ val query3 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE NOT EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query3, Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // IN-subquery rewritten as a left semi join. The hoisted predicate is a
> c1 rather than
+ // the key equality a = c1, so the answer is a IN (t2 INTERSECT t3) and
depends on the
+ // nested subquery: dropping it would leave no row at all.
+ val query4 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query4, Row(9) :: Nil)
+
+ // NOT IN-subquery rewritten as a null-aware left anti join, with a
nested EXISTS. Only the
+ // nested EXISTS keeps 9 out of the answer, as a > c1 alone does not
hold for it.
+ val query5 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR EXISTS (SELECT 1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query5, Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // A nested NOT IN-subquery keeps its null-aware semantics: c1 NOT IN
(3, 9, NULL) is never
+ // true, so only the correlated predicate can be satisfied. Compare with
query7, where the
+ // same NOT IN over a relation without NULL holds for c1 = 1 and returns
every row.
+ val query6 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3n)
+ |)""".stripMargin
+ checkAnswerAndPlan(query6, Row(1) :: Row(9) :: Nil)
+
+ val query7 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query7, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // A nested subquery correlated to the query it is nested in, on a
column it does not
+ // project, so its correlated predicate changes the answer: without it
the nested IN would
+ // hold for c1 = 1 and row 1 would be returned as well.
+ val query8 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t4 WHERE k = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query8, Row(2) :: Row(3) :: Row(7) :: Row(9) :: Nil)
+ }
+ }
+
+ test("SPARK-59351: nested subquery referencing both the outer and the inner
query") {
+ withTempView("t1", "t2", "t3", "t4") {
+ Seq((1), (2), (3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ // Correlated, this is false for every c1 in t2; without its correlated
predicate it holds
+ // for 1 and 9, so the correlation decides the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // The nested subquery references the outer query through its values and
the inner query
+ // through its own correlated predicate, so it can be rewritten into an
existence join on
+ // neither side and stays in the join condition. Planning it there as an
in-subquery filter
+ // drops its correlated predicate col1 = c1, which is why it is rejected
rather than left.
+ // These rows do not distinguish the two forms; the EXISTS form of this
shape does, and
+ // returned an extra row before the rejection was added.
+ //
+ // NOT IN is the outer predicate on purpose: its rewrite returns a bare
Join, which
+ // RewritePredicateSubquery does not offer to its own handling of
predicate subqueries in
+ // join conditions, so this rejection is the only one that can fire and
the assertion below
+ // pins it. With EXISTS the rewrite returns a Project over the join,
which that handling
+ // reaches under the default configuration and rejects with
+ // UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION instead, so the
query would still
+ // fail but on a different condition than the one asserted here.
+ val correlated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e = intercept[AnalysisException](sql(correlated).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ // The message must name the subquery that was rejected.
+ val reported = e.getMessageParameters.get("subqueryExpression")
+ assert(reported.contains("IN") && reported.contains("listquery"),
+ s"unexpected subqueryExpression: $reported")
+ }
+ }
+
+ // A correlated predicate that BooleanSimplification has eliminated:
+ // PullupCorrelatedPredicates retains c1 as an outer attribute of the
nested subquery for
+ // idempotency, but the subquery
+ // no longer reads it, so there is no condition to lose. It must still
be rewritten against
+ // the outer plan, which is the only plan it effectively references,
rather than be treated
+ // as referencing the subquery plan and rejected.
+ val simplifiedAwayCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ checkAnswer(sql(simplifiedAwayCorrelation), Row(1) :: Nil)
+ }
+ }
+
+ // A subquery that stays in the join condition stays there whole, since
the rewrite does
+ // not descend into one it declines, so a correlated subquery below it
is left as well.
+ // Its own references make it look routable, being owned by the subquery
plan, but
+ // nothing reaches it and planning a list query as an in-subquery filter
drops the
+ // predicate the pull-up hoisted into it. The enclosing subquery is
rejected for
+ // carrying it: PostgreSQL and DuckDB return 1 and 3 here, and losing
the correlation
+ // returns 2 as well.
+ val carriedCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + (CASE WHEN c1 IN (SELECT col1 FROM t4 WHERE k = c1) THEN
1 ELSE 0 END))
+ | IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e =
intercept[AnalysisException](sql(carriedCorrelation).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ }
+ }
+
+ // An uncorrelated nested subquery referencing both queries has no
correlated predicate to
+ // lose, so it stays in the join condition and is planned as an
in-subquery filter. The
+ // configuration is disabled because the rewrite of predicate subqueries
in join conditions
+ // rejects a subquery referencing both join inputs, which is what this
one becomes.
+ val uncorrelated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + c1) IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ withSQLConf(
+ SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
"false") {
+ // No existence join here: this subquery stays in the join condition,
which is the
+ // point of the case, so only the result and the reachability of the
plan are
+ // checked.
+ checkAnswerAndPlan(uncorrelated, Row(1) :: Row(2) :: Nil,
+ existenceJoinOnSubqueryPlan = false)
+ }
+ }
+ }
+
+ test("SPARK-59351: existential subquery carried by a rewritten nested
subquery") {
Review Comment:
**Nit (P3):** This is the only test that reaches
`rewriteCarriedExistentialExprs`, and it puts the outer EXISTS as a top-level
conjunct. The router now also runs on the existence joins built by the
Filter-predicate arm, `handleUnaryNode` and the join-condition arm. For
example, with `decorrelatePredicateSubqueriesInJoinPredicate.enabled=false`,
`SELECT * FROM t1 WHERE a > 100 OR EXISTS (SELECT 1 FROM t2 WHERE a = c1 OR c1
IN (SELECT col1 FROM t4 WHERE k = c1))` now keeps `k = c1` because the carried
IN is routed onto t2, and under the default configuration a carried sub-query
that references both inputs is now rejected by the router with a different
condition. If the routing stopped applying on those routes, every test would
stay green. Cases with the outer predicate under OR, in a SELECT list, and in a
user-written join condition, under both values of the conf, would pin it.
**Verification:**
- **Regression:** If the carried routing stopped applying on these routes,
the non-default configuration returns extra rows and the new cases fail.
- **Behavior:** With correlation that decides the answer, each route returns
the standard SQL rows under both configuration values.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala:
##########
@@ -161,4 +163,124 @@ class RewriteSubquerySuite extends PlanTest {
val existsExpected = LocalRelation(existsQuery.analyze.output).analyze
comparePlans(existsOptimized, existsExpected)
}
+
+ test("SPARK-59351: nested subquery of an EXISTS whose plan shares attributes
with the outer " +
+ "plan") {
+ // The EXISTS arms route the nested subquery against the raw subquery
plan, while the IN arms
+ // route it against the plan dedupSubqueryOnSelfJoin has already separated
from the outer
+ // plan. A self-join can leave an ExprId shared between the two plans for
the optimizer
+ // (SPARK-21835), and the analyzer is not involved here, so build that
state directly: `a` is
+ // the same attribute in the outer and in the subquery plan.
+ val a = $"a".int
+ val outer = LocalRelation(a, $"b".int)
+ val sub = LocalRelation(a)
+ val nested = LocalRelation($"col1".int)
+ assert(outer.outputSet.intersect(sub.outputSet).nonEmpty,
+ "the two plans must share an attribute for this test to mean anything")
+
+ // The nested subquery references `a`, which only the subquery plan is
meant to give it, and
+ // is correlated to it, so it must be rewritten against the subquery plan.
Classifying `a` as
Review Comment:
**Nit (P3):** This still describes the 'counts for the outer plan as well'
alternative, while the paragraph below and the `planReferences` scaladoc
describe the swap, under which `subPlanOnly` and `bothPlans` are both false and
the nested sub-query is declined and left in the semi join condition. Rewording
this sentence in terms of the swap, for example 'Attributing `a` to the outer
plan instead would decline the nested subquery and leave it in the semi join
condition', would make the fixture and the scaladoc describe the same
alternative.
**Verification:**
- **Inspection:** The fixture comment's alternative and outcome match the
scaladoc's swap reading and the exclusive probes in planReferences.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3224,625 @@ class SubquerySuite extends SharedSparkSession
}
}
}
+
+ /**
+ * Checks the result of `query`, that every node of its optimized plan can
produce the attributes
+ * it references, and that the existence join a nested subquery was
rewritten into sits on the
+ * subquery side of the semi/anti join.
+ *
+ * The plan checks are what SPARK-59351 is about: an existence join whose
condition references an
+ * attribute produced by neither of its children can still return the right
answer when the
+ * nested relation is empty at runtime, because the invalid condition is
then never bound, and a
+ * rewrite that placed the join on the other side would return the right
answer here too. Note
+ * that missingInput cannot see a subquery expression that survives in a
condition, since
+ * QueryPlan.references folds only the outer attributes of one and collect
does not walk its
+ * plan, so it is not a substitute for asserting the shape.
+ */
+ private def checkAnswerAndPlan(
+ query: String,
+ expected: Seq[Row],
+ existenceJoinOnSubqueryPlan: Boolean = true): Unit = {
+ val df = sql(query)
+ val plan = df.queryExecution.optimizedPlan
+ val invalidNodes = plan.collect { case p if p.missingInput.nonEmpty => p }
+ assert(invalidNodes.isEmpty,
+ s"""Plan nodes reference non-reachable attributes:
+ |${invalidNodes.mkString("\n")}
+ |$plan""".stripMargin)
+ if (existenceJoinOnSubqueryPlan) {
+ val semiOrAnti = plan.collectFirst {
+ case j: Join if j.joinType == LeftSemi || j.joinType == LeftAnti => j
+ }
+ assert(semiOrAnti.nonEmpty, s"expected a semi or anti join in\n$plan")
+ val existenceJoins = plan.collect {
+ case j: Join if j.joinType.isInstanceOf[ExistenceJoin] => j
+ }
+ assert(existenceJoins.nonEmpty, s"expected an existence join in\n$plan")
+ // The existence join must be under the right side of the semi/anti
join, which is the
+ // subquery plan, and not under its left side, which is the outer plan.
+ assert(existenceJoins.forall(j =>
semiOrAnti.get.right.exists(_.fastEquals(j))),
+ s"the existence join must sit on the subquery side in\n$plan")
+ }
+ checkAnswer(df, expected)
+ }
+
+ test("SPARK-59351: nested subquery referencing the inner query becomes an
existence join") {
+ // SPARK-45580 covers the case where the nested subquery references the
outer query, in which
+ // case its existence join is built on top of the outer plan. Here the
nested subquery
+ // references the query it is nested in, so the existence join has to be
built on top of the
+ // subquery plan instead.
+ withTempView("t1", "t2", "t3", "t3n", "t4", "t5") {
+ Seq((1), (2), (3), (7),
(9)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ // t3 shares a value with t2, t5 does not, so the nested subquery
decides the answer below.
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ Seq((3), (7)).toDF("col1").persist().createOrReplaceTempView("t5")
+ Seq(Some(3), Some(9),
None).toDF("col1").persist().createOrReplaceTempView("t3n")
+ // A correlated nested IN over t4 is false for every c1 in t2, while the
same IN without its
+ // correlated predicate is true for 1 and 9, so the correlation decides
the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // EXISTS rewritten as a left semi join. The correlated predicate is a
disjunction, so it
+ // is pulled up as a whole and carries the nested IN-subquery, which
references c1, out of
+ // the subquery plan.
+ val query1 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query1, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // Same over t5, which shares no value with t2, so the nested IN is
false for every c1 and
+ // only the correlated predicate can hold: a mistake making it true
would return every row.
+ val query2 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query2, Row(1) :: Row(9) :: Nil)
+
+ // NOT EXISTS rewritten as a left anti join.
+ val query3 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE NOT EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query3, Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // IN-subquery rewritten as a left semi join. The hoisted predicate is a
> c1 rather than
+ // the key equality a = c1, so the answer is a IN (t2 INTERSECT t3) and
depends on the
+ // nested subquery: dropping it would leave no row at all.
+ val query4 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query4, Row(9) :: Nil)
+
+ // NOT IN-subquery rewritten as a null-aware left anti join, with a
nested EXISTS. Only the
+ // nested EXISTS keeps 9 out of the answer, as a > c1 alone does not
hold for it.
+ val query5 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR EXISTS (SELECT 1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query5, Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // A nested NOT IN-subquery keeps its null-aware semantics: c1 NOT IN
(3, 9, NULL) is never
+ // true, so only the correlated predicate can be satisfied. Compare with
query7, where the
+ // same NOT IN over a relation without NULL holds for c1 = 1 and returns
every row.
+ val query6 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3n)
+ |)""".stripMargin
+ checkAnswerAndPlan(query6, Row(1) :: Row(9) :: Nil)
+
+ val query7 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query7, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // A nested subquery correlated to the query it is nested in, on a
column it does not
+ // project, so its correlated predicate changes the answer: without it
the nested IN would
+ // hold for c1 = 1 and row 1 would be returned as well.
+ val query8 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t4 WHERE k = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query8, Row(2) :: Row(3) :: Row(7) :: Row(9) :: Nil)
+ }
+ }
+
+ test("SPARK-59351: nested subquery referencing both the outer and the inner
query") {
+ withTempView("t1", "t2", "t3", "t4") {
+ Seq((1), (2), (3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ // Correlated, this is false for every c1 in t2; without its correlated
predicate it holds
+ // for 1 and 9, so the correlation decides the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // The nested subquery references the outer query through its values and
the inner query
+ // through its own correlated predicate, so it can be rewritten into an
existence join on
+ // neither side and stays in the join condition. Planning it there as an
in-subquery filter
+ // drops its correlated predicate col1 = c1, which is why it is rejected
rather than left.
+ // These rows do not distinguish the two forms; the EXISTS form of this
shape does, and
+ // returned an extra row before the rejection was added.
+ //
+ // NOT IN is the outer predicate on purpose: its rewrite returns a bare
Join, which
+ // RewritePredicateSubquery does not offer to its own handling of
predicate subqueries in
+ // join conditions, so this rejection is the only one that can fire and
the assertion below
+ // pins it. With EXISTS the rewrite returns a Project over the join,
which that handling
+ // reaches under the default configuration and rejects with
+ // UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION instead, so the
query would still
+ // fail but on a different condition than the one asserted here.
+ val correlated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e = intercept[AnalysisException](sql(correlated).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ // The message must name the subquery that was rejected.
+ val reported = e.getMessageParameters.get("subqueryExpression")
+ assert(reported.contains("IN") && reported.contains("listquery"),
+ s"unexpected subqueryExpression: $reported")
+ }
+ }
+
+ // A correlated predicate that BooleanSimplification has eliminated:
+ // PullupCorrelatedPredicates retains c1 as an outer attribute of the
nested subquery for
+ // idempotency, but the subquery
+ // no longer reads it, so there is no condition to lose. It must still
be rewritten against
+ // the outer plan, which is the only plan it effectively references,
rather than be treated
+ // as referencing the subquery plan and rejected.
+ val simplifiedAwayCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ checkAnswer(sql(simplifiedAwayCorrelation), Row(1) :: Nil)
+ }
+ }
+
+ // A subquery that stays in the join condition stays there whole, since
the rewrite does
+ // not descend into one it declines, so a correlated subquery below it
is left as well.
+ // Its own references make it look routable, being owned by the subquery
plan, but
+ // nothing reaches it and planning a list query as an in-subquery filter
drops the
+ // predicate the pull-up hoisted into it. The enclosing subquery is
rejected for
+ // carrying it: PostgreSQL and DuckDB return 1 and 3 here, and losing
the correlation
+ // returns 2 as well.
+ val carriedCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + (CASE WHEN c1 IN (SELECT col1 FROM t4 WHERE k = c1) THEN
1 ELSE 0 END))
+ | IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e =
intercept[AnalysisException](sql(carriedCorrelation).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ }
+ }
+
+ // An uncorrelated nested subquery referencing both queries has no
correlated predicate to
+ // lose, so it stays in the join condition and is planned as an
in-subquery filter. The
+ // configuration is disabled because the rewrite of predicate subqueries
in join conditions
+ // rejects a subquery referencing both join inputs, which is what this
one becomes.
+ val uncorrelated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + c1) IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ withSQLConf(
+ SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
"false") {
+ // No existence join here: this subquery stays in the join condition,
which is the
+ // point of the case, so only the result and the reachability of the
plan are
+ // checked.
+ checkAnswerAndPlan(uncorrelated, Row(1) :: Row(2) :: Nil,
+ existenceJoinOnSubqueryPlan = false)
+ }
+ }
+ }
+
+ test("SPARK-59351: existential subquery carried by a rewritten nested
subquery") {
+ // The nested IN over t4 is owned by the subquery plan, so it is rewritten
into an existence
+ // join and the condition PullupCorrelatedPredicates hoisted into it is
grafted onto that
+ // join. That condition carries a third-level IN correlated to t4, which
references neither
+ // the outer nor the subquery plan and so is routed by neither of the
guards above. It has to
+ // be rewritten against the inputs of the join just built: left in place,
PlanSubqueries plans
+ // it as an in-subquery filter and drops its own correlated predicate col1
= k, which returns
+ // every row of t1 instead of the one row where the correlation holds.
+ withTempView("t1", "t2", "t4", "t5") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ // No k equals any c1, so only the third-level IN can satisfy the
hoisted disjunction, and
+ // its correlated predicate col1 = k is false for every row of t4:
dropping it makes the
+ // level-2 IN hold for c1 = 1 and returns 1, 2 and 3.
+ Seq((1, 100), (9, 7)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1).toDF("col1").persist().createOrReplaceTempView("t5")
+ val query =
+ """SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t4
+ | WHERE k = c1 OR col1 IN (SELECT col1 FROM t5 WHERE col1
= k)))""".stripMargin
+ // The routing must not depend on the rewrite of predicate subqueries in
join conditions,
+ // which is what used to make these two configurations disagree: it
reached the grafted
+ // condition only when enabled.
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ checkAnswerAndPlan(query, Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: NOT IN keeps an uncorrelated three-valued residual") {
+ // The NOT IN rewrite returns a bare Join, which this rule does not offer
to its own handling
+ // of predicate subqueries in join conditions, so an uncorrelated nested
IN left in the
+ // anti-join condition stays there and InSubqueryExec evaluates it
three-valued. No existence
+ // join is built for it and it has no hoisted condition to lose, so it
must not be rejected:
+ // these queries returned the standard SQL answer before SPARK-59351 and
must keep doing so.
+ withTempView("t1", "t2", "t3n", "t4n") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ // The NULL matches no c1, so c1 IN (SELECT col1 FROM t3n) is unknown
for every row of t2.
+ Seq(Some(3), None).toDF("col1").persist().createOrReplaceTempView("t3n")
+ // Nullable col1 again, but correlated through k = c1.
+ Seq((Some(3), 1), (None, 8)).toDF("col1",
"k").persist().createOrReplaceTempView("t4n")
+
+ Seq(
+ "(c1 IN (SELECT col1 FROM t3n)) <=> false",
+ "(c1 IN (SELECT col1 FROM t3n)) IS NULL",
+ "COALESCE(c1 IN (SELECT col1 FROM t3n), true)",
+ "(c1 NOT IN (SELECT col1 FROM t3n)) <=> false").foreach { predicate =>
+ // Every such predicate is FALSE for every row of t2, so the subquery
reduces to a = c1
+ // and only 2 and 3 survive the NOT IN. Rejecting instead, or
rewriting the nested IN into
+ // a two-valued exists, would change this.
+ checkAnswer(
+ sql(s"SELECT * FROM t1 WHERE a NOT IN (SELECT c1 FROM t2 WHERE a =
c1 OR $predicate)"),
Review Comment:
**Nit (P3):** With the hoisted disjunct `a = c1` repeating the NOT IN key,
this query cannot depend on `$predicate`: S(1) contains 1 whatever the
predicate is, and for a = 2 or 3, S(a) holds only non-null values other than a.
So the answer is 2, 3 whether the residual is evaluated three-valued, rewritten
into a two-valued `exists`, or dropped; only a rejection is detected. The
comment above also says each predicate is FALSE for every row of t2, but `IS
NULL` and `COALESCE(..., true)` are TRUE there. The truth-only case at the end
asserts no plan, so 'rewritten, not kept' is unchecked too. A hoisted predicate
that is not the key equality, such as `c1 > a OR $predicate` as in query5 of
the first new test, over data where three-valued and two-valued evaluation
differ, would make each case discriminate.
This extends existing discussions
[1](https://github.com/apache/spark/pull/58656#issuecomment-5731841503),
[2](https://github.com/apache/spark/pull/58656#pullrequestreview-5247095515).
**Verification:**
- **Regression:** With the residual replaced by a two-valued rewrite, at
least one asserted answer in each residual case changes.
- **Behavior:** The rebuilt cases return the standard SQL answers, including
for NULLs in the nested list.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3224,625 @@ class SubquerySuite extends SharedSparkSession
}
}
}
+
+ /**
+ * Checks the result of `query`, that every node of its optimized plan can
produce the attributes
+ * it references, and that the existence join a nested subquery was
rewritten into sits on the
+ * subquery side of the semi/anti join.
+ *
+ * The plan checks are what SPARK-59351 is about: an existence join whose
condition references an
+ * attribute produced by neither of its children can still return the right
answer when the
+ * nested relation is empty at runtime, because the invalid condition is
then never bound, and a
+ * rewrite that placed the join on the other side would return the right
answer here too. Note
+ * that missingInput cannot see a subquery expression that survives in a
condition, since
+ * QueryPlan.references folds only the outer attributes of one and collect
does not walk its
+ * plan, so it is not a substitute for asserting the shape.
+ */
+ private def checkAnswerAndPlan(
+ query: String,
+ expected: Seq[Row],
+ existenceJoinOnSubqueryPlan: Boolean = true): Unit = {
+ val df = sql(query)
+ val plan = df.queryExecution.optimizedPlan
+ val invalidNodes = plan.collect { case p if p.missingInput.nonEmpty => p }
+ assert(invalidNodes.isEmpty,
+ s"""Plan nodes reference non-reachable attributes:
+ |${invalidNodes.mkString("\n")}
+ |$plan""".stripMargin)
+ if (existenceJoinOnSubqueryPlan) {
+ val semiOrAnti = plan.collectFirst {
+ case j: Join if j.joinType == LeftSemi || j.joinType == LeftAnti => j
+ }
+ assert(semiOrAnti.nonEmpty, s"expected a semi or anti join in\n$plan")
+ val existenceJoins = plan.collect {
+ case j: Join if j.joinType.isInstanceOf[ExistenceJoin] => j
+ }
+ assert(existenceJoins.nonEmpty, s"expected an existence join in\n$plan")
+ // The existence join must be under the right side of the semi/anti
join, which is the
+ // subquery plan, and not under its left side, which is the outer plan.
+ assert(existenceJoins.forall(j =>
semiOrAnti.get.right.exists(_.fastEquals(j))),
+ s"the existence join must sit on the subquery side in\n$plan")
+ }
+ checkAnswer(df, expected)
+ }
+
+ test("SPARK-59351: nested subquery referencing the inner query becomes an
existence join") {
+ // SPARK-45580 covers the case where the nested subquery references the
outer query, in which
+ // case its existence join is built on top of the outer plan. Here the
nested subquery
+ // references the query it is nested in, so the existence join has to be
built on top of the
+ // subquery plan instead.
+ withTempView("t1", "t2", "t3", "t3n", "t4", "t5") {
+ Seq((1), (2), (3), (7),
(9)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ // t3 shares a value with t2, t5 does not, so the nested subquery
decides the answer below.
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ Seq((3), (7)).toDF("col1").persist().createOrReplaceTempView("t5")
+ Seq(Some(3), Some(9),
None).toDF("col1").persist().createOrReplaceTempView("t3n")
+ // A correlated nested IN over t4 is false for every c1 in t2, while the
same IN without its
+ // correlated predicate is true for 1 and 9, so the correlation decides
the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // EXISTS rewritten as a left semi join. The correlated predicate is a
disjunction, so it
+ // is pulled up as a whole and carries the nested IN-subquery, which
references c1, out of
+ // the subquery plan.
+ val query1 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query1, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // Same over t5, which shares no value with t2, so the nested IN is
false for every c1 and
+ // only the correlated predicate can hold: a mistake making it true
would return every row.
+ val query2 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query2, Row(1) :: Row(9) :: Nil)
+
+ // NOT EXISTS rewritten as a left anti join.
+ val query3 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE NOT EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query3, Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // IN-subquery rewritten as a left semi join. The hoisted predicate is a
> c1 rather than
+ // the key equality a = c1, so the answer is a IN (t2 INTERSECT t3) and
depends on the
+ // nested subquery: dropping it would leave no row at all.
+ val query4 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query4, Row(9) :: Nil)
+
+ // NOT IN-subquery rewritten as a null-aware left anti join, with a
nested EXISTS. Only the
+ // nested EXISTS keeps 9 out of the answer, as a > c1 alone does not
hold for it.
+ val query5 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR EXISTS (SELECT 1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query5, Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // A nested NOT IN-subquery keeps its null-aware semantics: c1 NOT IN
(3, 9, NULL) is never
+ // true, so only the correlated predicate can be satisfied. Compare with
query7, where the
+ // same NOT IN over a relation without NULL holds for c1 = 1 and returns
every row.
+ val query6 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3n)
+ |)""".stripMargin
+ checkAnswerAndPlan(query6, Row(1) :: Row(9) :: Nil)
+
+ val query7 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query7, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // A nested subquery correlated to the query it is nested in, on a
column it does not
+ // project, so its correlated predicate changes the answer: without it
the nested IN would
+ // hold for c1 = 1 and row 1 would be returned as well.
+ val query8 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t4 WHERE k = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query8, Row(2) :: Row(3) :: Row(7) :: Row(9) :: Nil)
+ }
+ }
+
+ test("SPARK-59351: nested subquery referencing both the outer and the inner
query") {
+ withTempView("t1", "t2", "t3", "t4") {
+ Seq((1), (2), (3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ // Correlated, this is false for every c1 in t2; without its correlated
predicate it holds
+ // for 1 and 9, so the correlation decides the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // The nested subquery references the outer query through its values and
the inner query
+ // through its own correlated predicate, so it can be rewritten into an
existence join on
+ // neither side and stays in the join condition. Planning it there as an
in-subquery filter
+ // drops its correlated predicate col1 = c1, which is why it is rejected
rather than left.
+ // These rows do not distinguish the two forms; the EXISTS form of this
shape does, and
+ // returned an extra row before the rejection was added.
+ //
+ // NOT IN is the outer predicate on purpose: its rewrite returns a bare
Join, which
+ // RewritePredicateSubquery does not offer to its own handling of
predicate subqueries in
+ // join conditions, so this rejection is the only one that can fire and
the assertion below
+ // pins it. With EXISTS the rewrite returns a Project over the join,
which that handling
Review Comment:
**Nit (P3):** As written, this describes current behavior, but with EXISTS
as the outer predicate the EXISTS arm reaches the same rejection in
`rewriteExistentialExprInSubqueryPlan`, which raises
`NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY` before any join exists. The
join-condition handling would reject with
`UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION` only if that rejection
were removed. Saying so explicitly, for example 'were this rejection removed,
the EXISTS form would ...', or dropping the EXISTS sentence, would make the
rationale accurate.
**Verification:**
- **Inspection:** The rationale matches current behavior, where the EXISTS
form of the query fails with NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY,
and states any other outcome only as a counterfactual.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +406,476 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Same as [[rewriteExistentialExpr]], but it also returns the newly
introduced attributes, and
+ * it only rewrites the existential sub-queries for which `canRewrite`
returns true. A sub-query
+ * that is not rewritten stays in the returned expression as it is, and is
not descended into:
+ * rewriting an existential sub-query nested in its join condition would
graft an existence join
+ * onto the plan for an `exists` reference that the sub-query left in place
may never evaluate.
+ */
private def rewriteExistentialExprWithAttrs(
exprs: Seq[Expression],
- plan: LogicalPlan): (Option[Expression], LogicalPlan, Seq[Attribute]) = {
+ plan: LogicalPlan,
+ canRewrite: Expression => Boolean = _ => true): (Option[Expression],
LogicalPlan,
+ Seq[Attribute]) = {
var newPlan = plan
val introducedAttrs = ArrayBuffer.empty[Attribute]
- val newExprs = exprs.map { e =>
- e.transformDownWithPruning(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY)) {
- case Exists(sub, _, _, conditions, subHint) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- val existenceJoin = ExistenceJoin(exists)
- val newCondition = conditions.reduceLeftOption(And)
- newPlan =
- buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
- existenceJoin, newCondition, subHint)
- introducedAttrs += exists
- exists
- case Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- // To handle a null-aware predicate not-in-subquery in nested
conditions
- // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we transform
- // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
- //
- // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
- // is transformed into a plan below;
- // == Optimized Logical Plan ==
- // Project [id#78, v#79]
- // +- Filter ((v#79 > 0) OR NOT exists#83)
- // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
- // :- Relation[id#78,v#79] parquet
- // +- Relation[id#80] parquet
- val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
- val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan,
- rewriteDomainJoinsIfPresent(newPlan, newSub, Some(finalJoinCond)),
- ExistenceJoin(exists), Some(finalJoinCond), joinHint)
- introducedAttrs += exists
- Not(exists)
- case InSubquery(values, ListQuery(sub, _, _, _, conditions, subHint))
=>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
- ExistenceJoin(exists), newConditions, joinHint)
- introducedAttrs += exists
- exists
+ // Builds the replacement of a sub-query expression under that
expression's origin and with
+ // its tags, as TreeNode.transformDownWithPruning does for the nodes a
rule replaces. Without
+ // it every node built below, the `exists` attribute and the join
conditions included, would
+ // take the origin of the enclosing operator, and a runtime error raised
from the rewritten
+ // condition would quote that operator rather than the sub-query fragment.
+ def replacing(sq: Expression)(replacement: => Expression): Expression = {
+ val newExpr = CurrentOrigin.withOrigin(sq.origin)(replacement)
+ newExpr.copyTagsFrom(sq)
+ newExpr
+ }
+ def rewrite(expr: Expression): Expression = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return expr
+ }
+ expr match {
+ case sq @ Exists(sub, _, _, conditions, subHint) if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ val existenceJoin = ExistenceJoin(exists)
+ val newCondition = conditions.reduceLeftOption(And)
+ newPlan =
+ buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
+ existenceJoin, newCondition, subHint)
+ introducedAttrs += exists
+ exists
+ }
+ case sq @ Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint)))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ // To handle a null-aware predicate not-in-subquery in nested
conditions
+ // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we
transform
+ // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
+ //
+ // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
+ // is transformed into a plan below;
+ // == Optimized Logical Plan ==
+ // Project [id#78, v#79]
+ // +- Filter ((v#79 > 0) OR NOT exists#83)
+ // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
+ // :- Relation[id#78,v#79] parquet
+ // +- Relation[id#80] parquet
+ val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
+ val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = Join(newPlan,
+ rewriteDomainJoinsIfPresent(newPlan, newSub,
Some(finalJoinCond)),
+ ExistenceJoin(exists), Some(finalJoinCond), joinHint)
+ introducedAttrs += exists
+ Not(exists)
+ }
+ case sq @ InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan,
newSub, newConditions),
+ ExistenceJoin(exists), newConditions, joinHint)
+ introducedAttrs += exists
+ exists
+ }
+ // A sub-query that `canRewrite` declined is left as it is, children
included.
+ case sq @ (_: Exists | Not(_: InSubquery) | _: InSubquery) => sq
+ case other => other.mapChildren(rewrite)
}
}
+ val newExprs = exprs.map(rewrite)
(newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
}
+
+ /**
+ * Returns true if `e` references any of the attributes produced by `plan`.
+ */
+ private def referencesPlan(e: Expression, plan: LogicalPlan): Boolean = {
+ // Iterates the references and probes the output set, rather than
materialising the
+ // intersection: AttributeSet.intersect builds a set from its argument, so
the operand order
+ // there would decide whether this walks the references or the whole
output of `plan`. The
+ // membership probes still construct wrappers, so this is cheaper, not
allocation-free.
+ e.references.exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Returns true if `e` effectively references any of the attributes produced
by `plan`, see
+ * [[effectiveReferences]].
+ */
+ private def effectivelyReferencesPlan(e: Expression, plan: LogicalPlan):
Boolean = {
+ effectiveReferences(e).exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Which of an outer plan and a sub-query plan a set of references can be
evaluated by, as
+ * [[planReferences]] determines it.
+ */
+ private case class PlanReferences(subPlanOnly: Boolean, bothPlans: Boolean)
+
+ /**
+ * Classifies `references` against `outerPlan` and `subPlan` in a single
pass: whether they can
+ * only be evaluated by `subPlan`, and whether they need both plans.
+ *
+ * An attribute that both plans produce is attributed to `subPlan` rather
than to `outerPlan`,
+ * which is what the `subPlan` probe coming first does. The two plans share
no ExprId in any
+ * shape reached by the suite: `DeduplicateRelations` has given the
sub-query its own, and the
+ * EXISTS arms classify against the raw sub-query plan while the IN arms
classify against the
+ * plan `dedupSubqueryOnSelfJoin` has already separated from the outer plan.
The order is
+ * therefore defensive rather than something a query exercises, and it is
the safe one: a
+ * nested sub-query resolves in the scope of the sub-query it is nested in,
so an attribute
+ * both plans produce is one of that sub-query's own, and attributing it to
`outerPlan` instead
+ * would let a single shared attribute satisfy both sides of the
classification and have a
+ * nested sub-query referencing only the inner query rejected as referencing
both.
+ *
+ * An attribute neither plan produces leaves both answers false, which
declines the rewrite
+ * without rejecting the sub-query, as a sub-query referencing neither plan
can be left in the
+ * join condition unharmed.
+ */
+ private def planReferences(
+ references: AttributeSet,
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): PlanReferences = {
+ var subPlanReference = false
+ var outerPlanOnlyReference = false
+ val attrs = references.iterator
+ // Stops as soon as both plans are known to be referenced, which is all
the two answers need.
+ while (attrs.hasNext && !(subPlanReference && outerPlanOnlyReference)) {
+ val attr = attrs.next()
+ if (subPlan.outputSet.contains(attr)) {
+ subPlanReference = true
+ } else if (outerPlan.outputSet.contains(attr)) {
+ outerPlanOnlyReference = true
+ }
+ }
+ PlanReferences(
+ subPlanOnly = subPlanReference && !outerPlanOnlyReference,
+ bothPlans = subPlanReference && outerPlanOnlyReference)
+ }
+
+ /**
+ * The attributes that `e` references and can still evaluate. For an
existential sub-query
+ * expression these are the ones of the values it compares and of the
correlated condition that
+ * [[PullupCorrelatedPredicates]] hoisted into it, rather than the ones of
`references`, which
+ * also include the outer attributes the pull-up retains for idempotency.
Those outlive the
+ * condition that referenced them once BooleanSimplification has eliminated
it, as in
+ * `a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)`, which keeps `c1`
as an outer
+ * attribute of a sub-query that no longer reads it, and which must still be
rewritten against
+ * the outer plan rather than treated as referencing the sub-query plan.
+ *
+ * The sets are merged with `AttributeSet.fromAttributeSets` rather than
built from a flattened
+ * sequence of attributes: `AttributeSet.apply` would take one allocation
per attribute of a
+ * sequence as long as the total number of references, which a row-valued IN
over a wide row
+ * makes unnecessarily large.
+ */
+ private def effectiveReferences(e: Expression): AttributeSet = e match {
+ case Exists(_, _, _, joinCond, _) =>
+ AttributeSet.fromAttributeSets(joinCond.map(_.references))
+ case Not(in: InSubquery) => effectiveReferences(in)
+ case InSubquery(values, ListQuery(_, _, _, _, joinCond, _)) =>
+ AttributeSet.fromAttributeSets(values.map(_.references) ++
joinCond.map(_.references))
+ case _ => e.references
+ }
+
+ /**
+ * Rewrites the existential sub-queries that are nested in the correlated
predicates which
+ * [[PullupCorrelatedPredicates]] hoisted out of a predicate sub-query, and
which therefore end
+ * up in the condition of the semi/anti join that replaces that sub-query.
+ *
+ * A hoisted predicate can carry a nested existential sub-query out of the
sub-query plan,
+ * because a correlated predicate is hoisted as a whole when it is a
disjunction. For example
+ *
+ * SELECT * FROM t1 WHERE EXISTS (
+ * SELECT 1 FROM t2 WHERE t1.a = t2.c1 OR t2.c1 IN (SELECT col1 FROM t3))
+ *
+ * hoists `a = c1 OR c1 IN (SELECT col1 FROM t3)` into the join condition.
Such a nested
+ * sub-query must be rewritten against the plan that produces the attributes
it references:
+ * the one above references `c1`, which is produced by the sub-query plan
and not by the outer
+ * plan, so its existence join has to be built on top of the sub-query plan.
Building it on top
+ * of the outer plan instead yields a join whose condition references an
attribute that neither
+ * of its children can produce (SPARK-59351).
+ *
+ * A nested sub-query that references both plans cannot be rewritten into an
existence join on
+ * either side, since neither alone can evaluate it, so it has to stay in
the join condition,
+ * where both plans are in scope. Leaving it there is only correct while it
is uncorrelated: the
+ * join condition of a correlated one is dropped when it is planned as an
in-subquery filter,
+ * which would silently change the result, so a correlated one is reported
as unsupported
+ * instead. Note that such a sub-query can be
+ * correlated only to the sub-query plan, as being correlated to the outer
plan as well would
+ * require two levels of correlation, which the Analyzer rejects.
+ *
+ * A sub-query left here is further subject to the rewrite of predicate
sub-queries in join
+ * conditions, which rejects one referencing both plans under the default
configuration. It
+ * survives that rewrite when
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled` is
disabled, and
+ * also when it is an uncorrelated IN sub-query and
+ *
`spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled`
is disabled,
+ * since it is then not among the sub-queries that rewrite considers.
+ *
+ * An existence join yields only whether a row matched, so its `exists`
attribute cannot tell
+ * FALSE from unknown, while `IN` is three-valued. The two are
indistinguishable while the value
+ * only feeds a predicate, which is what a hoisted condition normally does,
but not when it
+ * reaches something else, e.g. `(c1 IN (SELECT col1 FROM t3)) <=> false`,
which is FALSE for a
+ * NULL that matches nothing and TRUE for the `exists` attribute. An IN
sub-query whose row
+ * comparison can evaluate to unknown is therefore rejected in that position
rather than
+ * rewritten. NOT IN is rewritten with a null-aware join condition of its
own, which is equally
+ * two-valued, so it is rejected there too.
+ *
+ * Returns the rewritten condition along with the updated outer and
sub-query plans.
+ */
+ private def rewriteExistentialExprInJoinCondition(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan, LogicalPlan) = {
+ val (subCond, newSubPlan) =
+ rewriteExistentialExprInSubqueryPlan(conditions, outerPlan, subPlan)
+ // The sub-queries that do not reference the sub-query plan are rewritten
against the outer
+ // plan, as they only reference attributes of the outer plan, if any. This
accepts on the
+ // negation of the test above rather than on a positive one, which rests
on the invariant that
+ // `effectiveReferences` sees every attribute the rewrite can put in a
join condition: one it
+ // cannot see would be rewritten here and end up in a condition the outer
plan cannot evaluate.
+ // A positive test is not available without also refusing the sub-queries
whose retained outer
+ // attributes no longer correspond to a condition, which must keep being
rewritten here.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !effectivelyReferencesPlan(e, subPlan))
+ (newCond, newOuterPlan, newSubPlan)
+ }
+
+ /**
+ * Rewrites the existential sub-queries in `conditions` that can only be
evaluated by the
+ * sub-query plan, that is those referencing the sub-query plan but not the
outer plan, into
+ * existence joins on top of the sub-query plan. See
+ * [[rewriteExistentialExprInJoinCondition]] for details.
+ *
+ * Returns the rewritten condition along with the updated sub-query plan.
+ */
+ private def rewriteExistentialExprInSubqueryPlan(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan) = {
+ // The rejection collector below is a plain `collect`, which visits every
node of the hoisted
+ // conditions whether or not it can hold a sub-query, so skip the
classification entirely when
+ // they hold none at all, which is the common case for a correlated EXISTS
or IN.
+ // [[collectUnknownSensitiveInSubqueries]] prunes subquery-free subtrees
on its own, so this
+ // guard only saves it the top-level call.
+ if (!conditions.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return (conditions.reduceOption(And), subPlan)
+ }
+ // A sub-query left in the join condition loses its own join condition
when it is planned
+ // there, so a correlated one would silently return a wrong result: reject
it instead. The
+ // same holds for a correlated sub-query nested below one that is left
there, since the
+ // rewrite leaves a declined sub-query whole, children included, and
nothing else will reach
+ // it: see carriesCorrelatedSubquery. Note that this walks the whole
expression, including the
+ // join condition of a sub-query the rewrite declines to descend into,
which is deliberate for
+ // the same reason.
+ //
+ // Which plan a candidate references is asked of it up to three times
below, once for the
+ // rejection of a sub-query referencing both plans, once to keep only the
unknown-sensitive
+ // ones the sub-query plan owns, and once more by the rewrite itself. Both
answers come from
+ // the same reference set in a single pass over it, and the result is
memoised per candidate,
+ // so neither the set nor the scan is repeated. The memo is keyed by
identity: the candidates
+ // are nodes of the expressions being classified, and comparing two of
them structurally would
+ // compare the sub-query plans they hold.
+ val classified = new java.util.IdentityHashMap[Expression,
PlanReferences]()
+ def classify(sq: Expression): PlanReferences = {
+ var references = classified.get(sq)
+ if (references == null) {
+ references = planReferences(effectiveReferences(sq), outerPlan,
subPlan)
+ classified.put(sq, references)
+ }
+ references
+ }
+ def ownedBySubPlan(sq: Expression): Boolean = classify(sq).subPlanOnly
+ def referencesBothPlans(sq: Expression): Boolean = classify(sq).bothPlans
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if referencesBothPlans(sq) &&
Review Comment:
Thanks, the correlated carried descendant is routed now and both
configurations agree on that query. The second case still fails on 4168a32841b:
with t4(col1, k) = (9, 7) and t3n(x) = (3, NULL),
rewriteCarriedExistentialExprs leaves the uncorrelated `(col1 IN (SELECT x FROM
t3n)) <=> false` in place, and under the default configuration the
join-condition arm rewrites it into a two-valued exists, so every row of t1 is
returned instead of 1; see the comment on rewriteCarriedExistentialExprs in
optimizer/subquery.scala.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4257086397","thread_id":"inline:4257086397","verdict_sha256":"3bb54f2a3a4b20553e9d3ca5caee9c4a8fc61921fc05f7c2363e10d0f4e5e7b0"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +406,476 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Same as [[rewriteExistentialExpr]], but it also returns the newly
introduced attributes, and
+ * it only rewrites the existential sub-queries for which `canRewrite`
returns true. A sub-query
+ * that is not rewritten stays in the returned expression as it is, and is
not descended into:
+ * rewriting an existential sub-query nested in its join condition would
graft an existence join
+ * onto the plan for an `exists` reference that the sub-query left in place
may never evaluate.
+ */
private def rewriteExistentialExprWithAttrs(
exprs: Seq[Expression],
- plan: LogicalPlan): (Option[Expression], LogicalPlan, Seq[Attribute]) = {
+ plan: LogicalPlan,
+ canRewrite: Expression => Boolean = _ => true): (Option[Expression],
LogicalPlan,
+ Seq[Attribute]) = {
var newPlan = plan
val introducedAttrs = ArrayBuffer.empty[Attribute]
- val newExprs = exprs.map { e =>
- e.transformDownWithPruning(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY)) {
- case Exists(sub, _, _, conditions, subHint) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- val existenceJoin = ExistenceJoin(exists)
- val newCondition = conditions.reduceLeftOption(And)
- newPlan =
- buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
- existenceJoin, newCondition, subHint)
- introducedAttrs += exists
- exists
- case Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- // To handle a null-aware predicate not-in-subquery in nested
conditions
- // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we transform
- // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
- //
- // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
- // is transformed into a plan below;
- // == Optimized Logical Plan ==
- // Project [id#78, v#79]
- // +- Filter ((v#79 > 0) OR NOT exists#83)
- // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
- // :- Relation[id#78,v#79] parquet
- // +- Relation[id#80] parquet
- val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
- val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan,
- rewriteDomainJoinsIfPresent(newPlan, newSub, Some(finalJoinCond)),
- ExistenceJoin(exists), Some(finalJoinCond), joinHint)
- introducedAttrs += exists
- Not(exists)
- case InSubquery(values, ListQuery(sub, _, _, _, conditions, subHint))
=>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
- ExistenceJoin(exists), newConditions, joinHint)
- introducedAttrs += exists
- exists
+ // Builds the replacement of a sub-query expression under that
expression's origin and with
+ // its tags, as TreeNode.transformDownWithPruning does for the nodes a
rule replaces. Without
+ // it every node built below, the `exists` attribute and the join
conditions included, would
+ // take the origin of the enclosing operator, and a runtime error raised
from the rewritten
+ // condition would quote that operator rather than the sub-query fragment.
+ def replacing(sq: Expression)(replacement: => Expression): Expression = {
+ val newExpr = CurrentOrigin.withOrigin(sq.origin)(replacement)
+ newExpr.copyTagsFrom(sq)
+ newExpr
+ }
+ def rewrite(expr: Expression): Expression = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return expr
+ }
+ expr match {
+ case sq @ Exists(sub, _, _, conditions, subHint) if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ val existenceJoin = ExistenceJoin(exists)
+ val newCondition = conditions.reduceLeftOption(And)
+ newPlan =
+ buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
+ existenceJoin, newCondition, subHint)
+ introducedAttrs += exists
+ exists
+ }
+ case sq @ Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint)))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ // To handle a null-aware predicate not-in-subquery in nested
conditions
+ // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we
transform
+ // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
+ //
+ // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
+ // is transformed into a plan below;
+ // == Optimized Logical Plan ==
+ // Project [id#78, v#79]
+ // +- Filter ((v#79 > 0) OR NOT exists#83)
+ // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
+ // :- Relation[id#78,v#79] parquet
+ // +- Relation[id#80] parquet
+ val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
+ val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = Join(newPlan,
+ rewriteDomainJoinsIfPresent(newPlan, newSub,
Some(finalJoinCond)),
+ ExistenceJoin(exists), Some(finalJoinCond), joinHint)
+ introducedAttrs += exists
+ Not(exists)
+ }
+ case sq @ InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan,
newSub, newConditions),
+ ExistenceJoin(exists), newConditions, joinHint)
+ introducedAttrs += exists
+ exists
+ }
+ // A sub-query that `canRewrite` declined is left as it is, children
included.
+ case sq @ (_: Exists | Not(_: InSubquery) | _: InSubquery) => sq
+ case other => other.mapChildren(rewrite)
}
}
+ val newExprs = exprs.map(rewrite)
(newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
}
+
+ /**
+ * Returns true if `e` references any of the attributes produced by `plan`.
+ */
+ private def referencesPlan(e: Expression, plan: LogicalPlan): Boolean = {
+ // Iterates the references and probes the output set, rather than
materialising the
+ // intersection: AttributeSet.intersect builds a set from its argument, so
the operand order
+ // there would decide whether this walks the references or the whole
output of `plan`. The
+ // membership probes still construct wrappers, so this is cheaper, not
allocation-free.
+ e.references.exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Returns true if `e` effectively references any of the attributes produced
by `plan`, see
+ * [[effectiveReferences]].
+ */
+ private def effectivelyReferencesPlan(e: Expression, plan: LogicalPlan):
Boolean = {
+ effectiveReferences(e).exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Which of an outer plan and a sub-query plan a set of references can be
evaluated by, as
+ * [[planReferences]] determines it.
+ */
+ private case class PlanReferences(subPlanOnly: Boolean, bothPlans: Boolean)
+
+ /**
+ * Classifies `references` against `outerPlan` and `subPlan` in a single
pass: whether they can
+ * only be evaluated by `subPlan`, and whether they need both plans.
+ *
+ * An attribute that both plans produce is attributed to `subPlan` rather
than to `outerPlan`,
+ * which is what the `subPlan` probe coming first does. The two plans share
no ExprId in any
Review Comment:
Thanks, the planReferences scaladoc now uses the swap reading in both
paragraphs. The fixture comment in RewriteSubquerySuite still says classifying
`a` as an outer reference as well would report it as referencing both plans;
see the comment there.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4257086406","thread_id":"inline:4257086406","verdict_sha256":"3bb54f2a3a4b20553e9d3ca5caee9c4a8fc61921fc05f7c2363e10d0f4e5e7b0"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -184,8 +188,12 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
// Use EXISTS if performance matters to you.
// Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val dedupSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
+ val inConditions = values.zip(dedupSub.output).map(EqualTo.tupled)
+ // The hoisted correlated predicates may carry existential
sub-queries that only the
+ // sub-query plan can evaluate, see
rewriteExistentialExprInJoinCondition.
+ val (subConditions, newSub) =
+ rewriteExistentialExprInSubqueryPlan(conditions, p, dedupSub)
Review Comment:
Confirmed: the NOT IN arm keeps the uncorrelated residual and the four
shapes return 2, 3 as at the merge base, while the correlated ones are still
rejected. Two follow-ups on the new comment and test are in this review: the
bare-Join premise does not hold when the NOT IN is not the last sub-query
conjunct, and the hoisted `a = c1` means the test's rows cannot tell a
three-valued residual from a two-valued rewrite.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4281437494","thread_id":"inline:4281437494","verdict_sha256":"3bb54f2a3a4b20553e9d3ca5caee9c4a8fc61921fc05f7c2363e10d0f4e5e7b0"}
-->
--
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]
