pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4056568739
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3227,198 @@ class SubquerySuite extends SharedSparkSession
}
}
}
+
+ 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")
+
+ // Checks the result, and that every node of the optimized plan can
produce the attributes
+ // it references. The latter is what this fix 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. checkAnswer alone would not catch it, as the
missing input checks it
+ // runs only look at the root of the plan.
+ def checkAnswerAndPlan(query: String, expected: Seq[Row]): 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)
+ checkAnswer(df, expected)
+ }
+
+ // 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) :: 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
+ checkAnswerAndPlan(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
+ checkAnswerAndPlan(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
+ checkAnswerAndPlan(query4, Row(1) :: Nil)
Review Comment:
Right on all three, including the reduction: with `a = c1` hoisted,
`a IN {c1 ∈ t2 : a = c1 OR nested(c1)}` is `a ∈ t2` whatever `nested`
returns. And query8's
`c1 IN (SELECT col1 FROM t3 WHERE col1 = c1)` is indeed equivalent to the
uncorrelated form, so it
pinned nothing about correlated nesting.
Rebuilt with your suggestion, `a > c1` as the hoisted disjunct, plus data
chosen so each answer
moves if the nested subquery is mishandled. `t1` is now `(1, 2, 3, 7, 9)`:
| query | shape | answer | what breaks it |
|---|---|---|---|
| query2 | EXISTS, nested IN over `t5 = (3, 7)`, disjoint from `t2` | `1, 9`
| nested IN wrongly true -> every row |
| query4 | outer IN, `a > c1`, nested IN over `t3` | `9` | nested IN dropped
-> empty |
| query5 | outer NOT IN, `a > c1`, nested EXISTS | `1, 2, 3, 7` | nested
EXISTS dropped -> `9` comes back |
| query6/7 | nested NOT IN over a relation with/without NULL | `1, 9` / all
rows | null-awareness lost |
| query8 | EXISTS, `a > c1`, nested IN over `t4(col1, k)` correlated on `k`,
which it does not project | `2, 3, 7, 9` | correlation dropped -> `1` comes
back |
`t4 = ((1, 9), (9, 1))` makes the correlated nested IN false for every `c1`
in `t2` while the same
IN without its correlated predicate is true for 1 and 9, so correlated and
uncorrelated nesting now
give different answers. Every query goes through the tree-wide
`missingInput` helper as well.
Your note that query5 never had the `missingInput` half is accurate —
`SubqueryExpression.references`
excludes the join condition, so the nested `Exists` has `outerAttrs = {c1}`
from the right child and
its pre-fix failure is an `Unevaluable` `Exists` reaching evaluation. With
the new data the NOT IN
branch no longer rests on "it does not throw": its answer changes if the
nested EXISTS is mishandled.
--
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]
