dongjoon-hyun commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4065967215
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -414,7 +433,8 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
existenceJoin, newCondition, subHint)
introducedAttrs += exists
exists
- case Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
+ case sq @ Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint)))
+ if canRewrite(sq) =>
val exists = AttributeReference("exists", BooleanType, nullable =
false)()
// Deduplicate conflicting attributes if any.
val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
Review Comment:
The deduplication anchor moves from the outer plan to the sub-query plan on
the new route, and that has two consequences.
Before this PR `newPlan` here was always the outer plan, and
`PullupCorrelatedPredicates` had already deduplicated the nested sub-query's
plan against exactly that node (`decorrelate(sub, plan, ...)` at L837/L845;
`DecorrelateInnerQuery.scala:468` derives `outputPlanInputAttrs` from
`outerPlan.inputSet`). In phase 1 `newPlan` is the sub-query plan, against
which nothing has deduplicated.
**(a) ids shared with the outer plan are no longer aliased, and `buildJoin`
cannot compensate.** `Join.computeOutput` for `ExistenceJoin` is `leftOutput :+
j.exists` (`basicLogicalOperators.scala:729-730`), so `duplicates =
outerRefs.intersect(subplan.outputSet)` at L103 never sees the nested relation
below the new join. For
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (SELECT a FROM t1))
```
the result is `Join(t1, Join(t2, t1', ExistenceJoin), LeftSemi)` where `t1'`
can reuse the outer `t1`'s exprIds across sibling subtrees — a position no
`duplicateResolved` check inspects, since that predicate compares only one
join's two children (L820). `LogicalPlanIntegrity.checkIfSameExprIdNotReused`
would fire under plan-change validation (default `Utils.isTesting`,
`SQLConf.scala:706`).
**(b) The `Exists` arm can now throw instead of aliasing.** It reaches dedup
through `buildJoin` (L432), which passes the condition (L65), and
`dedupSubqueryOnSelfJoin` throws `conflictingAttributesInJoinConditionError`
when the condition references a duplicate (L105-111). A nested correlated
`EXISTS`'s `newCondition` always references sub-plan attributes, so an id
collision with the sub-query plan surfaces as a user-facing
`CONFLICTING_ATTRIBUTES_IN_JOIN_CONDITION`. Note this line and L466 call the
helper with no `condition`, so the IN arms only alias and never throw — an
asymmetry inside the same helper.
`DeduplicateRelations` does descend into `subquery.plan`
(`DeduplicateRelations.scala:249-253`) and resolves many of these in the
analyzer, but the SPARK-21835 tests at `SubquerySuite.scala:1128-1190` exist
because conflicts still reach the optimizer. There is no test here where the
nested relation overlaps either plan.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
Review Comment:
Both items from the previous review round look like they were answered but
not pushed.
The last commit on this branch is `3e656616ced`, committed
`2026-09-21T16:49:46Z`. The two replies are `16:57:33Z` (on `:565`) and
`16:58:23Z` (here), i.e. eight minutes *after* the commit, and neither change
is in the tree:
1. The reply on `:565` says `isCorrelated` was replaced by
`hasCorrelatedCondition` reading the hoisted join condition, and that routing
now uses the surviving state. `grep -n
"hasCorrelatedCondition\|effective\|surviving"` on this file returns nothing;
`isCorrelatedSubquery` (L634-638) still reads `listQuery.isCorrelated`,
`referencesPlan` (L486-488) still reads `e.references`, and there is no `false
AND col1 = c1` regression test. The 0A000 regression against 4.0.1 reproduced
in that reply is therefore still live.
2. This line still reads
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`, without
the `.enabled` suffix that reply says was added. That is not the registered key.
The PR description also still contains "Both the routing and the rejections
read the hoisted condition and the compared values rather than `references` and
`isCorrelated` for this reason", which describes the unpushed version rather
than this code.
CI being green does not contradict any of this: nothing covers either item.
Could you check whether a commit is missing?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ case Not(child) => unknownSensitiveInSubqueries(child, inPredicate)
+ case in: InSubquery =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ // The join condition of a sub-query expression is a predicate, evaluated
by the join that
+ // the sub-query is rewritten into.
+ case sq: SubqueryExpression =>
+ sq.children.flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ case other =>
Review Comment:
This fallback treats every parent other than `And`/`Or`/`Not` as a value
position, which over-rejects shapes where the rewrite is exact.
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1 OR ((c1 IN (SELECT col1 FROM t3n)) <=> true))
```
with a NULL in `t3n` is rejected, but it should not be: `unknown <=> TRUE`
and `FALSE <=> TRUE` are both FALSE, and the existence join's condition is the
NULL-rejecting `values.zip(newSub.output).map(EqualTo.tupled)` (L467), so
`exists` is TRUE exactly when IN is TRUE. `ReplaceNullWithFalseInPredicate`
already lists this position among those where NULL and FALSE are
interchangeable (L79-82, with the comment "whether the other side is null or
false has no difference"), together with `If`'s predicate and `CaseWhen` branch
conditions — none of which this helper recognises.
The shape survives folding for precisely the queries this PR targets:
`SimplifyBinaryComparison`'s `case a EqualNullSafe TrueLiteral if !a.nullable
=> a` (`expressions.scala:620`) is guarded by `!a.nullable`, which fails
exactly when the IN is nullable; and `SimplifyConditionals`
(`expressions.scala:644, 655`) normalises `IF(c1 IN (...), true, false)` and
`CASE WHEN c1 IN (...) THEN true ELSE false END` into that same rejected
`EqualNullSafe`. So ordinary SQL lands here, and `<=> true` is rejected while
the new test rejects `<=> false` legitimately — an inconsistent pair.
Reusing the case list from `ReplaceNullWithFalseInPredicate` would close it.
(Its lambda-body cases are not reachable from this rule, so they are not part
of this.)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
Review Comment:
Two things here.
**The error names the wrong operator.** This returns `in`, the un-negated
`InSubquery`, so `nestedInSubqueryWithUnknownResultError` calls `.sql` on it
and `InSubquery.sql` is `s"(${value.sql} IN (${query.sql}))"`
(`predicates.scala:424`) — the user's `NOT` disappears and the message reads
"The IN subquery (c1 IN (...))" for a `NOT IN`. The new test feeds exactly this
shape, `(c1 NOT IN (SELECT col1 FROM t3n)) <=> false`, but asserts only on
`getCondition`, so nothing catches it. Returning the `Not` node renders
correctly via `Not.sql`.
**The arm is behaviourally redundant.** Deleting it changes nothing: `case
Not(child)` on the next line falls through to `case in: InSubquery` and returns
the identical value. It exists only to carry the comment, and the duplicated
condition can drift.
Separately, the scaladoc on `inSubqueryMayBeUnknown` (L620-621) says "NOT IN
is excluded", which contradicts both this call site and the comment directly
above it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
Review Comment:
Also minor: this adds two full, unpruned traversals of the hoisted
conditions for every correlated predicate sub-query, including the common case
where there is no nested sub-query at all.
Only the third step, `rewriteExistentialExprWithAttrs`, prunes on
tree-pattern bits (L423). This `collect` uses `TreeNode.collect`, which
`foreach`es every node, and `unknownSensitiveInSubqueries` (L574) recurses to
the leaves through `case other => other.children.flatMap(...)` with no pattern
check anywhere, allocating a result collection at each interior node even when
everything comes back empty.
For an ordinary correlated EXISTS or IN with no nested sub-query — TPC-H
Q21, TPC-DS q10/q35, most real workloads — the pre-PR code did one `O(1)` bit
test per hoisted condition; now the whole condition tree is walked twice, per
predicate sub-query, per compile. The IN arm passes `inConditions ++
conditions` (L177), so a K-column IN's synthesised equalities are walked twice
as well.
A guard at the top of this method is semantically exact, since both
collectors match only `Exists` and `InSubquery`:
```scala
if (!conditions.exists(_.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY))) {
return (conditions.reduceOption(And), subPlan)
}
```
The same check at the head of `unknownSensitiveInSubqueries` additionally
prunes sub-query-free branches of an `OR`.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -158,20 +158,23 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
// Filter the plan by applying left semi and left anti joins.
withSubquery.foldLeft(newFilter) {
case (p, Exists(sub, _, _, conditions, subHint)) =>
- val (joinCond, outerPlan) = rewriteExistentialExpr(conditions, p)
- val join = buildJoin(outerPlan,
rewriteDomainJoinsIfPresent(outerPlan, sub, joinCond),
+ val (joinCond, outerPlan, newSub) =
+ rewriteExistentialExprInJoinCondition(conditions, p, sub)
Review Comment:
The four arms hand the router differently-prepared sub-plans.
This arm and the `Not(Exists)` arm below pass the raw `sub`; the
`InSubquery` and `Not(InSubquery)` arms first compute `val dedupSub =
dedupSubqueryOnSelfJoin(p, sub, Some(values))` (L174, L190) and route against
that, which by construction no longer overlaps `p.outputSet`. For EXISTS the
dedup happens only later, inside `buildJoin` (L65).
So whenever `sub.outputSet ∩ p.outputSet` is non-empty — exactly what L103
tests for, and the shape of the SPARK-21835 self-join tests at
`SubquerySuite.scala:1128` — a single referenced attribute satisfies *both*
`referencesPlan(sq, subPlan)` and `referencesPlan(sq, outerPlan)` at L565-566.
A correlated nested sub-query that genuinely references only one side is then
rejected with `NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY`, whose
wording does not describe the query; an uncorrelated one is declined by both
passes and stranded in the join condition, where the default-configuration
handler rejects it at L248-251.
Deduplicating `sub` before this call, as the IN arm already does, would
remove the asymmetry.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !referencesPlan(e, subPlan))
Review Comment:
Phase 1 accepts on a positive test, `referencesPlan(e, subPlan) &&
!referencesPlan(e, outerPlan)` (L581), but phase 2 accepts on the *negation*,
`!referencesPlan(e, subPlan)`. The two are not complements, and the gap is
filled in the unsafe direction: whatever `references` cannot see is not left
alone, it is affirmatively grafted onto the outer plan.
`SubqueryExpression.references` is
`AttributeSet.fromAttributeSets(nonOuterScopeAttrs.map(_.references))`
(`expressions/subquery.scala:88-89`) — it drops every outer attribute
containing an `OuterScopeReference` (partitioned at `:85-86`) and never
includes the sub-query's own `joinCond`. A nested sub-query whose link to the
sub plan is invisible to `references` therefore fails phase 1's positive test,
passes this negation, and gets an `ExistenceJoin` on the outer plan whose
condition references an attribute neither child produces — the exact
invalid-plan shape SPARK-59351 fixes.
I could not establish reachability (nested correlation is rejected earlier
today), so this is a robustness point rather than a live bug. But the routing's
correctness currently rests on an unstated invariant — that `references` covers
every attribute the rewrite will put in a join condition — and a positive test
here (`references the outer plan, or references neither`) would fail safe
instead.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ case Not(child) => unknownSensitiveInSubqueries(child, inPredicate)
+ case in: InSubquery =>
Review Comment:
This arm and the `Not(InSubquery)` arm above return without recursing into
`in.children` (`values :+ query`), while the `SubqueryExpression` arm three
lines below explicitly does descend. `InSubquery` is a `Predicate`, not a
`SubqueryExpression`, so it can never reach that arm.
```sql
SELECT * FROM t1 WHERE EXISTS (
SELECT 1 FROM t2 WHERE a = c1
OR c1 IN (SELECT col1 FROM t4 WHERE k = c1 AND ((c1 IN (SELECT col1 FROM
t3n)) <=> false)))
```
The inner `<=> false` is a level-2 correlated predicate and is hoisted into
the level-2 `ListQuery`'s `joinCond`. The walk stops at the level-2
`InSubquery` with `inPredicate = true` and never examines the `t3n` IN. Phase 1
then rewrites the level-2 sub-query onto the sub plan, L468 puts the
un-inspected `EqualNullSafe` into the new `ExistenceJoin` condition, and the
`case j: Join` handler — which has no unknown guard either — turns it into a
second existence join whose `exists` is FALSE where the IN is unknown. Before
this PR that query produced the missing-input plan this PR fixes, so the path
moves from a broken plan to a silent wrong answer. The `values` side has the
same hole.
This is also inconsistent with the `referencingBothPlans` collector at
L563-567, whose comment states that it "walks the whole expression, including
the join condition of a sub-query that the rewrite below declines to descend
into".
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,14 +405,26 @@ 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(
Review Comment:
Replacing `transformDownWithPruning` with a hand-rolled recursion drops the
two things that traversal did besides walking the tree.
`TreeNode.transformDownWithPruning` applies the rule inside
`CurrentOrigin.withOrigin(origin)` (`TreeNode.scala:510-512`) and calls
`afterRule.copyTagsFrom(this)` when a node is replaced (`TreeNode.scala:525`).
The new `rewrite` does neither, so every node the three arms build — the
`AttributeReference("exists")` at L428/438/464, the
`EqualTo`/`Or`/`IsNull`/`And` join conditions, and the `Join`s at L432/457/470
— inherits the enclosing `Filter`'s ambient origin instead of the sub-query
expression's, and `TreeNodeTag`s on the replaced `Exists`/`InSubquery` are
lost. The visible effect is the `SQLQueryContext` on runtime errors raised from
the rewritten join condition: a `DIVIDE_BY_ZERO` inside a hoisted predicate now
quotes the whole `Filter` rather than the sub-query fragment.
Worth noting that this cost lands on code paths the PR does not otherwise
touch: `handleUnaryNode` (L347) routes every `UnaryNode` through here, and with
the default `canRewrite = _ => true` the new do-not-descend arm at L474-475 is
unreachable for those callers. Wrapping the arm bodies in
`CurrentOrigin.withOrigin(expr.origin)` and adding `copyTagsFrom` restores it.
Pruning and gc-churn avoidance are preserved correctly otherwise.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ case Not(child) => unknownSensitiveInSubqueries(child, inPredicate)
+ case in: InSubquery =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ // The join condition of a sub-query expression is a predicate, evaluated
by the join that
+ // the sub-query is rewritten into.
+ case sq: SubqueryExpression =>
Review Comment:
`sq.children` is `outerAttrs ++ joinCond` (`expressions/subquery.scala:91`),
but the comment above justifies `inPredicate = true` only for the join
condition. `outerAttrs` are value expressions, not predicates.
So an IN sub-query reachable through an `outerAttrs` entry — e.g. a compound
correlated expression that `PullupCorrelatedPredicates` lifts into
`outerAttrs`, which `getOuterReferences` explicitly supports via
`stripOuterReference(a)` for aggregate expressions
(`expressions/subquery.scala:236-237`) — is marked as a predicate position and
skipped by the unknown-result check although its value is not tested for TRUE.
Same failure class as the other predicate-context gaps: a three-valued IN
becomes a non-nullable `exists` bit and the answer changes.
Passing `inPredicate = true` only for the `joinCond` slice would match what
the comment already says.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 can be rewritten into an
existence join on
+ * neither side, so it is left 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 only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * 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.
+ 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 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. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ case Not(child) => unknownSensitiveInSubqueries(child, inPredicate)
+ case in: InSubquery =>
+ if (!inPredicate && inSubqueryMayBeUnknown(in)) Seq(in) else Nil
+ // The join condition of a sub-query expression is a predicate, evaluated
by the join that
+ // the sub-query is rewritten into.
+ case sq: SubqueryExpression =>
+ sq.children.flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ case other =>
+ other.children.flatMap(unknownSensitiveInSubqueries(_, inPredicate =
false))
+ }
+
+ /**
+ * Returns true if `e` is a positive IN sub-query whose row comparison can
evaluate to unknown,
+ * that is one that can return NULL rather than only TRUE or FALSE. An
existence join cannot
+ * represent that third value, see
[[rewriteExistentialExprInJoinCondition]]. NOT IN is excluded,
+ * as [[rewriteExistentialExprWithAttrs]] gives it a null-aware join
condition of its own.
+ */
+ private def inSubqueryMayBeUnknown(e: Expression): Boolean = e match {
Review Comment:
This re-derives `InSubquery.nullable` (`predicates.scala:414-423`), which is
already `values.exists(_.nullable) || query.childOutputs.exists(_.nullable)`,
but without its `LEGACY_IN_SUBQUERY_NULLABILITY` branch (`SQLConf.scala:7740`,
SPARK-43413).
I think the config-independence is the *right* call here — the three-valued
runtime behaviour being protected does not depend on a flag, and reusing
`in.nullable` would make the check unsound under it. But nothing says so and no
test pins it, and the consequence is easy to miss: under
`spark.sql.legacy.inSubqueryNullability=true` the new guard quietly stops
covering two of its own three test predicates.
- `(c1 IN (...)) IS NULL` is folded to false by `NullPropagation`'s `case
IsNull(c) if !c.nullable` (`expressions.scala:940`), so the IN leaves the plan
entirely.
- `(c1 IN (...)) <=> false` becomes `Not(InSubquery)` via
`SimplifyBinaryComparison`'s `case a EqualNullSafe FalseLiteral if !a.nullable
=> Not(a)` (`expressions.scala:622`), which this helper then classifies as
`inPredicate` and does not reject.
A comment stating the divergence is deliberate, plus a test under that flag,
would keep it from being read as an oversight. `listQuery.childOutputs` would
also be the intended accessor rather than `plan.output` (the `zip` truncation
makes them equal today, but `ListQuery` carries `numCols` precisely because
`plan.output` grows after decorrelation).
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ 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 = {
Review Comment:
Minor, but this is on a hot path and the fix is free: the operands are the
wrong way round.
`AttributeSet.intersect` is `new
AttributeSet(other.baseSet.filter(baseSet.contains))`
(`expressions/AttributeSet.scala:144-145`) — it iterates and materialises from
the *argument*. So `e.references.intersect(plan.outputSet)` walks all of
`plan.outputSet` and builds a `LinkedHashSet`, making each call
`O(|plan.output|)` rather than `O(|e.references|)`, for a boolean that
`.nonEmpty` immediately discards.
`referencesPlan` is called two to four times per nested sub-query (L565-566,
L576, L581, L541) plus twice per sub-query in the `Join` branch at L239-240.
Over a 300-column relation that is roughly 1,200 hash probes and four set
allocations per nested sub-query where a handful would do.
`plan.outputSet.intersect(e.references).nonEmpty` iterates the small side.
Since this helper now also backs L239-240, one edit improves both sites.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3227,262 @@ 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", "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")
+
+ // 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) ::
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") {
+ 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.
+ //
+ // 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.
With EXISTS the rewrite
+ // returns a Project over the join, which that handling then rejects on
its own under the
+ // default configuration, and the assertion would hold either way.
+ 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")
+ }
+ }
+
+ // 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") {
+ val df = sql(uncorrelated)
+ val plan = df.queryExecution.optimizedPlan
+ val invalidNodes = plan.collect { case p if p.missingInput.nonEmpty =>
p }
Review Comment:
This assertion cannot see what this test exists to exercise.
`QueryPlan.references` is `AttributeSet(expressions) -- producedAttributes`
(`plans/QueryPlan.scala:139-141`), and `AttributeSet(exprs)` folds
`_.references`, which for a `SubqueryExpression` is only its `outerAttrs` —
never its `joinCond`. Separately `LogicalPlan.collect` walks `children`, which
never include a sub-query expression's plan (hence the separate
`QueryPlan.subqueriesAll`). So for any node still carrying a predicate
sub-query in its condition, `missingInput` sees neither that sub-query's join
condition nor anything inside its plan.
That is exactly the `uncorrelated` case here, whose whole point is that the
`InSubquery` *survives* in the join condition. It passes today only because
that particular sub-query is uncorrelated and its `joinCond` is empty.
Two smaller points in the same area:
- The block is a local `def checkAnswerAndPlan` at L3253-3262, copied
verbatim here, and absent from the third test (L3442-3487), whose two positive
cases call only `checkAnswer`. Those are the shapes where — as the comment at
L3366-3374 explains — an invalid `ExistenceJoin` still returns the right answer
when the nested relation is empty at runtime, so a regression there trips
nothing. Lifting the helper to a suite-level `private def` and using it in all
three would fix that.
- None of the three tests asserts plan *shape*, so a fix that placed the
`ExistenceJoin` on the other side would pass identically. Asserting the join
sits on the sub-query side would pin the actual contract.
--
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]
