pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4427705961
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3224,822 @@ 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. The EXISTS form of this query fails on the same condition,
raised by the same
+ // rejection in rewriteExistentialExprInSubqueryPlan before any join
exists; were that
+ // rejection removed, the EXISTS form would instead reach the
join-condition handling, which
+ // rejects it with UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION
under the default
+ // configuration, so only the NOT IN form pins this rejection on its own.
+ 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: a carried subquery is classified as on the subquery-plan
route") {
+ // The carried router applies the same rules as
rewriteExistentialExprInSubqueryPlan, against
+ // the inputs of the existence join rather than the outer and the subquery
plan.
+ withTempView("t1", "t2", "t4", "t5", "t3n", "t5n") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((9, 7)).toDF("col1", "k").persist().createOrReplaceTempView("t4")
+ Seq(Some(3), None).toDF("x").persist().createOrReplaceTempView("t3n")
+ Seq((Some(3), 7)).toDF("x", "k").persist().createOrReplaceTempView("t5n")
+
+ // A carried NOT IN is collected as the Not over the IN, so the rewrite
takes its null-aware
+ // arm. Taking the positive-IN arm instead leaves NOT exists over a
condition that ignores
+ // NULL comparisons: 5 NOT IN (NULL) is unknown, so n4's (5, 100) row is
excluded and the
+ // answer is empty, where the two-valued form returns every row of n1,
that is 1 and 2.
+ withTempView("n1", "n2", "n4", "n5") {
+ Seq(1, 2).toDF("a").persist().createOrReplaceTempView("n1")
+ Seq(5).toDF("c1").persist().createOrReplaceTempView("n2")
+ Seq((5, 100)).toDF("col1", "k").persist().createOrReplaceTempView("n4")
+ Seq((Option.empty[Int], 100)).toDF("x",
"y").persist().createOrReplaceTempView("n5")
+ val carriedNotIn =
+ """SELECT * FROM n1 WHERE EXISTS (
+ | SELECT 1 FROM n2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM n4
+ | WHERE k = c1 OR col1 NOT IN (SELECT x FROM n5 WHERE y =
k)))""".stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(sql(carriedNotIn), Nil)
+ }
+ }
+ }
+
+ // An uncorrelated carried IN is routed too, not left for the
join-condition rewrite, which
+ // has no unknown guard and would make it a two-valued exists under the
default
+ // configuration. Its unknown is observable here, so it is rejected
under both settings.
+ val uncorrelatedUnknown =
+ """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)))""".stripMargin
+ // A correlated one in the same position, which must report the same
condition rather than
+ // the one about referencing both queries.
+ val correlatedUnknown =
+ """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)))""".stripMargin
+ Seq(uncorrelatedUnknown, correlatedUnknown).foreach { query =>
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ val e = intercept[AnalysisException](sql(query).collect())
+ assert(e.getCondition ==
"UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY." +
+ "NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT", query)
+ }
+ }
+ }
+
+ // The same shape where only truth is observed is still rewritten and
answers correctly.
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(
+ 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))))""".stripMargin),
+ Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: carried subqueries are routed on every route that builds
an existence join") {
+ // rewriteCarriedExistentialExprs runs on every existence join this rule
builds, not only on
+ // the ones the semi/anti join arms build for a top-level predicate
subquery. On the routes
+ // below the outer EXISTS is itself rewritten into an existence join by
+ // rewriteExistentialExprWithAttrs, so the nested IN hoisted into its
condition is carried
+ // rather than routed by the first-level classification, and only this
router reaches it.
+ withTempView("t1", "t2", "t4", "t6") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ // The correlated nested IN is false for every c1 of t2, while the same
IN without k = c1
+ // holds for c1 = 1, so losing that predicate turns every answer below
into every row.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1, 2, 3).toDF("b").persist().createOrReplaceTempView("t6")
+ val exists =
+ "EXISTS (SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (SELECT col1 FROM t4
WHERE k = c1))"
+
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ // The Filter-predicate arm: under OR the EXISTS is not a conjunct
of its own, so it
+ // goes through rewriteExistentialExpr rather than through the semi
join arm.
+ checkAnswer(sql(s"SELECT * FROM t1 WHERE a > 100 OR $exists"),
Row(1) :: Nil)
+ // handleUnaryNode: the EXISTS is a projection rather than a
predicate.
+ checkAnswer(
+ sql(s"SELECT a, $exists AS e FROM t1"),
+ Row(1, true) :: Row(2, false) :: Row(3, false) :: Nil)
+ }
+ }
+
+ // The join-condition arm reaches the router only when it runs at all,
which is what
+ // decorrelatePredicateSubqueriesInJoinPredicate gates; with it disabled
the subquery stays
+ // in the join condition, which is pre-existing behaviour this PR does
not change.
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
"true") {
+ checkAnswer(
+ sql(s"SELECT t1.a FROM t1 JOIN t6 ON t1.a = t6.b AND $exists"),
Row(1) :: Nil)
+ }
+ }
+ }
+
+ test("SPARK-59351: a carried IN shared between a bare and a negated
position") {
+ // Filter pushdown substitutes the alias, so the same IN instance is
carried bare and under
+ // a Not. Routing the bare one must not replace the one inside the Not, or
the null-aware
+ // existence join built for the Not is left unreferenced and NOT exists
over a two-valued
+ // match returns the rows where the IN is unknown. Standard SQL keeps only
a = 1.
+ withTempView("s1", "s2", "s3") {
+ Seq(Some(1), Some(-1),
None).toDF("a").persist().createOrReplaceTempView("s1")
+ Seq(1).toDF("c1").persist().createOrReplaceTempView("s2")
+ Seq(Some(3), None).toDF("col1").persist().createOrReplaceTempView("s3")
+ val query =
+ """SELECT s.a FROM (SELECT a, a IN (SELECT col1 FROM s3) AS b FROM s1)
s
+ |WHERE s.a > 0 OR EXISTS (SELECT 1 FROM s2 WHERE s.a = s2.c1 OR s.b
OR NOT s.b)"""
+ .stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(sql(query), Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: a carried subquery referencing both inputs of its
existence join") {
+ // The third-level IN takes its compared value c1 from the left input of
the existence join
+ // built for the nested IN over t4, and its hoisted x = k reads the right
one. Neither input
+ // can evaluate it, and left in place it is planned as an in-subquery
filter that drops x = k.
+ withTempView("t1", "t2", "t4", "t5") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1, 9).toDF("x").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 c1 IN (SELECT x FROM t5 WHERE x =
k)))""".stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ val e = intercept[AnalysisException](sql(query).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY", conf)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: carried unknown-result rejection is limited to the
subquery-plan route") {
+ // On the routes that predate this change the carried IN below was
answered, and it is not
+ // rejected: its handling there is left to SPARK-59862. The columns are
nullable but hold no
+ // NULL, so the two-valued and the three-valued answers agree and a
rejection would be the
+ // only difference. The first query takes its compared value from the
right input of the
+ // existence join built for the EXISTS, the second from the left one.
+ withTempView("t1", "t2", "t3") {
+ Seq(Some(1), Some(2),
Some(3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(Some(1), Some(8),
Some(9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq(Some(8)).toDF("col1").persist().createOrReplaceTempView("t3")
Review Comment:
Fixed in eaeb8f53e71. `t3` now holds a non-matching NULL. With each
join-condition flag off, both the right-input and the left-input query must
return only 1, keep the IN in the existence-join condition, and plan it through
`InSubqueryExec`. Removing the in-place keep fails the case. Under the default
the test only checks that the query isn't rejected, since the two-valued answer
there is SPARK-59862.
--
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]
