LuciferYang commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4022391019
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -2678,4 +2678,133 @@ class SubquerySuite extends SharedSparkSession
assert(exposedAttribute.exprId == outerReferenceAttribute.exprId)
}
+
+ 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") {
+ Seq((1), (2), (3), (7)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ Seq(Some(3), Some(9),
None).toDF("col1").persist().createOrReplaceTempView("t3n")
+
+ // 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
+ val df1 = sql(query1)
+ // Every plan node must be able to produce the attributes it references.
+ val invalidNodes = df1.queryExecution.optimizedPlan.collect {
+ case p if p.missingInput.nonEmpty => p
+ }
+ assert(invalidNodes.isEmpty,
+ s"""Plan nodes reference non-reachable attributes:
+ |${invalidNodes.mkString("\n")}
+ |${df1.queryExecution.optimizedPlan}""".stripMargin)
+ checkAnswer(df1, Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // Same, with a nested subquery that returns no matching row.
+ val query2 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3 WHERE col1 = 3)
+ |)""".stripMargin
+ checkAnswer(sql(query2), Row(1) :: 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 t3 WHERE col1 = 3)
+ |)""".stripMargin
+ checkAnswer(sql(query3), Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // IN-subquery rewritten as a left semi join.
+ val query4 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswer(sql(query4), Row(1) :: Nil)
+
+ // NOT IN-subquery rewritten as a null-aware left anti join, with a
nested EXISTS.
+ val query5 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR EXISTS (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswer(sql(query5), 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.
+ val query6 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3n)
+ |)""".stripMargin
+ checkAnswer(sql(query6), Row(1) :: Nil)
+
+ // Without the NULL, c1 NOT IN (3, 9) holds for c1 = 1.
+ val query7 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswer(sql(query7), Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // A nested subquery that is itself correlated to the query it is nested
in.
+ val query8 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswer(sql(query8), Row(1) :: Nil)
Review Comment:
The tree-wide `optimizedPlan.collect { case p if p.missingInput.nonEmpty =>
p }` on query1 is the only assertion in the new test that can catch an invalid
plan before execution; the other seven queries only call `checkAnswer`.
`assertEmptyMissingInput` inside `checkAnswer` looks at the roots of the
analyzed/optimized/executed plans only, and `validateNoDanglingReferences` uses
`collectFirst`, which matches at the root and returns before it ever reaches
the `ExistenceJoin` below.
Your description notes that a nested subquery correlated to the query it is
nested in can still produce the right answer on an invalid plan when the nested
relation is empty at runtime, and query5 and query8 are both that shape with
only a `checkAnswer` on each. Pulling the tree-wide check into a small helper
and running it on those two would give this class of regression something to
trip on.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -454,6 +470,70 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
}
(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 = {
+ e.references.intersect(plan.outputSet).nonEmpty
+ }
+
+ /**
+ * 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, so it is left in the join condition, where both plans are in
scope. Such a
+ * sub-query is necessarily uncorrelated, as being correlated to the outer
plan as well would
+ * require two levels of correlation, which the Analyzer rejects. It is then
either planned as
+ * an in-subquery filter or reported as unsupported by the rewrite of
predicate sub-queries in
+ * join conditions.
Review Comment:
The doc says such a leftover is necessarily uncorrelated. That does not hold
for `InSubquery`: it has no `references` override, so its references are those
of its children (`values :+ query`), and the outer reference can come from
`values` while the subquery stays correlated to the inner query.
`PlanSubqueries` ignores `joinCond` when it matches the list query, so the
hoisted `col1 = c1` is gone. With
`decorrelatePredicateSubqueriesInJoinPredicate` off, `EXISTS (SELECT 1 FROM t2
WHERE a = c1 OR a IN (SELECT col1 FROM t3 WHERE col1 = c1))` returns one extra
row here, where master throws INTERNAL_ERROR_ATTRIBUTE_NOT_FOUND.
That sentence needs correcting, and it is worth settling in this PR whether
a correlated leftover should raise 0A000 as well.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -439,7 +454,8 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
ExistenceJoin(exists), Some(finalJoinCond), joinHint)
introducedAttrs += exists
Not(exists)
- case InSubquery(values, ListQuery(sub, _, _, _, conditions, subHint))
=>
+ case sq @ InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))
+ if canRewrite(sq) =>
Review Comment:
When `canRewrite` returns false the node is left as it is, and
`transformDownWithPruning` then descends into its children.
`SubqueryExpression.children` is `outerAttrs ++ joinCond`, so a subquery nested
inside the leftover's own `joinCond` does get rewritten: an `ExistenceJoin` is
grafted onto the subquery plan and an `exists` reference lands in that
`joinCond`. The 0A000 message for a doubly nested shape shows it as
`listquery(t2.c1, t2.c1, (t3.col1 = t2.c1), exists)`.
With default confs this shape ends at 0A000, so I have no repro where it
alone yields a wrong answer. But in the paths where `PlanSubqueries` drops the
whole `joinCond` (the NOT IN branch, or any branch with
`decorrelatePredicateSubqueriesInJoinPredicate` off), that `ExistenceJoin` is
computed with nothing referencing it. Worth confirming whether descending into
a declined subquery is intended here.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -454,6 +470,70 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
}
(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 = {
Review Comment:
`referencesPlan`'s `e.references.intersect(plan.outputSet).nonEmpty` already
exists in this object: the two lines in the `case j: Join` handler that test
the left and right sides are the same expression doing the same job,
classifying a subquery in a join condition by which side it references. With
two copies, a later change to the classification is easy to make in one place
only.
Reusing `referencesPlan` there would close it. Note that
`PredicateHelper.canEvaluate` looks like the same test but uses `subsetOf`, so
it is not a drop-in for either.
--
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]
