LuciferYang commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4046157769
##########
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:
**MEDIUM**
query4, query5 and query8 hoist `a = c1`, the key equality itself, so their
answers cannot depend on the nested subquery: `a IN {c1 ∈ t2 : a = c1 OR
nested(c1)}` reduces to `a ∈ t2` whatever `nested` returns. query8 is worse:
`c1 IN (SELECT col1 FROM t3 WHERE col1 = c1)` is logically identical to the
uncorrelated `c1 IN (SELECT col1 FROM t3)`, so no data distinguishes correlated
nesting from uncorrelated, the property the description lists it as covering.
query4 and query8 still carry the `missingInput` half; query5 does not,
since `SubqueryExpression.references` excludes the join condition, so the
nested `Exists` there has `outerAttrs = {c1}`, produced by the right child. Its
pre-fix failure is an `Unevaluable` `Exists` reaching evaluation. So the NOT IN
branch rests on "it no longer throws" alone.
Making the answers depend on the nested subquery needs the hoisted disjunct
to stop being the key equality, plus new data: `a > c1 OR ...` reduces query4
to `a ∈ t2 AND a ∈ t3`, empty on these rows. A second column on `t3`,
correlated on the one it does not project, removes query8's degenerate
correlation, though its answer stays insensitive either way.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,100 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
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 = {
+ 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. This is only
+ * correct for an uncorrelated sub-query: 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 it 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.
+ *
+ * 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.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !referencesPlan(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) = {
+ // A correlated sub-query referencing both plans has to stay in the join
condition, where its
+ // own join condition is lost, so reject it rather than silently return a
wrong result.
+ val unsupported = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) &&
+ referencesPlan(sq, subPlan) && referencesPlan(sq, outerPlan) => sq
+ })
+ if (unsupported.nonEmpty) {
+ throw
QueryCompilationErrors.unsupportedCorrelatedSubqueryInJoinConditionError(unsupported)
Review Comment:
**LOW**
`UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION` reads "Correlated
subqueries in the join predicate cannot reference both join inputs"
(`error/error-conditions.json:9756`). That reads as written for the call site
at `subquery.scala:249`, where the user wrote a JOIN, so "join predicate" and
"join inputs" name things in their query. At this call site the join is the
semi/anti join Spark synthesizes from `EXISTS` or `IN`, and the user's SQL
contains no join at all.
Both call sites raise it off a join Spark synthesized: this one, and the
subquery that stays in the join condition under the default configuration and
is rejected at `:249`. The category is right and only the sub-condition wording
does not transfer. Could it say the subquery cannot reference both the outer
query and the subquery it is nested in, or gain a sibling sub-condition? Either
one covers both sites.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,100 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
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 = {
+ 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. This is only
+ * correct for an uncorrelated sub-query: 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 it 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.
+ *
+ * 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.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !referencesPlan(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) = {
+ // A correlated sub-query referencing both plans has to stay in the join
condition, where its
+ // own join condition is lost, so reject it rather than silently return a
wrong result.
+ val unsupported = conditions.flatMap(_.collect {
Review Comment:
**LOW**
This `collect` walks the whole expression, including the join condition of a
subquery that `canRewrite` declines, since `Exists.children` is `outerAttrs ++
joinCond`. The rewrite deliberately does not descend there, which is what the
new scaladoc is about, so the rejection covers more ground than the rewrite
does: a correlated subquery buried in a declined subquery's own join condition
would fail the query even though nothing would have rewritten it.
Reaching it needs that inner subquery to reference both of the current
plans, i.e. two levels of correlation, which the scaladoc says the Analyzer
rejects, so I could not build a query that gets there: a scope mismatch rather
than a live bug. One sentence here saying the wider scan is deliberate settles
it. Narrowing the scan to match the rewrite would be the worse direction, since
a subquery hidden under a declined one would then be planned without its join
condition or reach execution unevaluable.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,100 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
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 = {
+ 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. This is only
+ * correct for an uncorrelated sub-query: 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 it 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.
+ *
+ * Returns the rewritten condition along with the updated outer and
sub-query plans.
+ */
+ private def rewriteExistentialExprInJoinCondition(
Review Comment:
**LOW**
This paragraph says a subquery referencing both plans "stays in the join
condition, where both plans are in scope", and it does say that holds only for
the uncorrelated case. What it leaves out is that the uncorrelated case then
depends on
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`.
The PR description has the rest: under the default configuration the
pre-existing rewrite of predicate subqueries in join conditions rejects it with
the same error, and it survives and returns the right answer only with that
conf off. The comment does not carry that, and the test that pins the shape
(`SubquerySuite.scala:3412`) sets the non-default value without saying why it
has to.
The description sinks out of view once this merges, so the comment and the
test are what the next reader has. One clause in the scaladoc naming the
default-configuration rejection, and a line in the test saying the conf is
there to bypass that branch, would keep the two in step.
##########
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)
+
+ // 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
+ checkAnswerAndPlan(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
+ checkAnswerAndPlan(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
+ checkAnswerAndPlan(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
+ checkAnswerAndPlan(query8, Row(1) :: Nil)
+ }
+ }
+
+ test("SPARK-59351: nested subquery referencing both the outer and the inner
query") {
+ withTempView("t1", "t2", "t3") {
+ 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")
+
+ // 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
+ // would drop its correlated predicate col1 = c1 and return an extra
row, so it is rejected.
+ val correlated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ // The rejection must not depend on whether predicate subqueries in join
conditions are
+ // decorrelated, as that rewrite rejects the same shape on its own.
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
Review Comment:
**LOW**
Deleting the new `throw` at `subquery.scala:551` leaves the `true` iteration
of this loop green: the nested subquery stays in the join condition, and since
`RewritePredicateSubquery.apply` descends into the children of the node it just
produced and the EXISTS arm returns `Project(p.output, join)`, the pre-existing
branch at `subquery.scala:249` sees that join and raises the same condition.
The `false` iteration is the discriminating one: with `case j: Join` no
longer matching, removing the new throw drops `col1 = c1`, an extra row comes
back, and `intercept` gets no exception.
No exception assertion separates the two sites: both call the same method
with one `subqueryExpression` parameter, and for this query both pass the same
expression, so the rendered `.sql` is identical. While both sites raise the
same condition the `true` iteration cannot be made to discriminate, so either
the comment should say that it reaches `subquery.scala:249` and only the
`false` iteration covers the new throw, or the case should use NOT IN as the
outer predicate: that arm returns a bare `Join` (`subquery.scala:210`), which
`transformDown` never re-offers to the `case j: Join` branch, so the new throw
is the only one that can fire under either conf.
--
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]
