cloud-fan commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4414363509
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +417,649 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, a correlated IN or NOT IN is planned
by `PlanSubqueries`
Review Comment:
**Nit (P3):** 'Nothing else reaches it' holds only with
`decorrelatePredicateSubqueriesInJoinPredicate.enabled` off. Under the default,
the top-down pass visits the existence joins this rule has just built in the
`case j: Join` arm, which rewrites a carried correlated IN or EXISTS against
the input it references. The PR's own test comment says the same ('it reached
the grafted condition only when enabled'). The next paragraph's 'for a join the
user wrote' has the same problem. The scaladoc would match the code if it said
that the hoisted condition is dropped only with that setting off, and that
under the default the router adds config independence and the unknown-result
guard. The 'only this router reaches it' comment in the carried-routes test
needs the same change.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#discussion_r4369134661).
**Verification:**
- **Inspection:** The router's scaladoc and the carried-route test comment
both state that, under the default
decorrelatePredicateSubqueriesInJoinPredicate setting, the join-condition
handling also reaches the existence joins this rule builds.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +417,649 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, a correlated IN or NOT IN is planned
by `PlanSubqueries`
+ * as an in-subquery filter that drops its own hoisted condition, which
returns extra rows, and
+ * a correlated EXISTS reaches execution as an unevaluable `Exists` and
fails.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is routed as well, rather than left for that
handling to rewrite
+ * under its default configuration: it has no hoisted condition to lose, but
it is subject to the
+ * same unknown-result guard as one on the sub-query-plan route, and that
handling has no such
+ * guard. Consuming it here is what keeps an IN whose result is observably
unknown from becoming
+ * a two-valued `exists` attribute.
+ *
+ * The classification mirrors [[rewriteExistentialExprInSubqueryPlan]], with
this join's inputs
+ * in place of the outer and the sub-query plan: a sub-query referencing
exactly one input is
+ * routed against it, one referencing both is rejected when it is correlated
or carries a
+ * correlated sub-query, since neither input alone can evaluate it, and one
whose unknown result
+ * is observable is rejected. A sub-query referencing both inputs or neither
and carrying no
+ * correlation is left where it is, as it is at the first level.
+ *
+ * The unknown-result rejection applies only when `rejectsUnknownResult` is
set, which is the
+ * case for the existence joins built on the sub-query-plan route, where the
same shapes failed
+ * with an internal error. On the routes that predate this change an
uncorrelated IN whose
+ * unknown result is observable is left where it is, as it was, and a
correlated one is routed
+ * without the rejection. Their two-valued handling is tracked by
SPARK-59862.
+ *
+ * A NOT IN is routed before a bare IN, since the substitution of a unit
replaces every
+ * occurrence of it, including one inside a `Not` over the same instance.
+ */
+ private def rewriteCarriedExistentialExprs(
+ join: Join,
+ rejectsUnknownResult: Boolean): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ val carried = ArrayBuffer.empty[Expression]
+ collectCarriedExistentialExprs(condition, carried)
+ if (carried.isEmpty) {
+ return join
+ }
+ // A NOT IN is carried as the Not over the IN, which is the unit the
rewrite consumes and the
+ // unit the unknown-result collector reports, so both agree on the node
below.
+ def negatedIn(sq: Expression): Expression = sq match {
+ case Not(in: InSubquery) => in
+ case other => other
+ }
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+ def referencesBothInputs(sq: Expression): Boolean =
+ referencesInput(sq, join.left) && referencesInput(sq, join.right)
+ def referencesOneInput(sq: Expression): Boolean =
+ referencesInput(sq, join.left) != referencesInput(sq, join.right)
+
+ // The two rejections of the sub-query-plan route apply here for the same
reasons, against the
+ // inputs of this join rather than against the outer and the sub-query
plan, and each raises
+ // the condition whose message describes it.
+ val referencingBothInputs = carried.filter { sq =>
+ referencesBothInputs(sq) &&
+ (hasCorrelatedCondition(sq) ||
carriesCorrelatedSubquery(negatedIn(sq)))
+ }
+ if (referencingBothInputs.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothInputs.toSeq)
+ }
+ // Only the sub-query-plan route rejects. An existence join built on
another route answered
+ // such a sub-query before this routing existed, three-valued where it was
left in place, and
+ // that is left to SPARK-59862: see `rejectsUnknownResult`.
+ def uncorrelated(sq: Expression): Boolean =
+ !hasCorrelatedCondition(sq) && !carriesCorrelatedSubquery(negatedIn(sq))
+ if (rejectsUnknownResult) {
+ val unknownResult = carried.filter(sq => referencesOneInput(sq) &&
observesUnknown(sq))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult.toSeq)
+ }
+ }
+ // A NOT IN unit goes before a bare IN: the substitution of a unit
replaces every occurrence
+ // of it in the condition, so a bare IN routed first would also replace
the one inside a
+ // `Not` over the same instance, and the null-aware join built for that
`Not` would be left
+ // unreferenced. The sort is stable, so the order within each kind is the
order collected.
+ val routed = carried.sortBy {
+ case Not(_: InSubquery) => 0
+ case _ => 1
+ }.filterNot(sq => !rejectsUnknownResult && observesUnknown(sq) &&
uncorrelated(sq))
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ routed.filter(sq => referencesOneInput(sq) &&
isInput(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
+ rewriteExistentialExprWithAttrs(Seq(sq), p, rejectsUnknownResult =
rejectsUnknownResult)
+ newCondition = newCondition.transform {
+ case e if e.fastEquals(sq) => rewritten.get
+ }
+ newInput
+ }
+ }
+ val newLeft = route(join.left, referencesInput(_, join.left))
+ val newRight = route(join.right, referencesInput(_, join.right))
+ if (newLeft.fastEquals(join.left) && newRight.fastEquals(join.right)) {
+ join
+ } else {
+ // Drop the `exists` attributes the nested existence joins introduced,
keeping this join's
+ // own output, which already carries its own.
+ Project(join.output, join.copy(
+ left = newLeft, right = newRight, condition = Some(newCondition)))
+ }
+ }
+
+ /**
+ * Collects the existential sub-queries carried by the condition of an
existence join, as the
+ * units [[rewriteExistentialExprWithAttrs]] rewrites: a NOT IN is collected
as the [[Not]] over
+ * the [[InSubquery]], so the rewrite takes its null-aware arm and the
unknown-result collector,
+ * which reports the same node, can be compared against it.
+ *
+ * A collected sub-query is not descended into. One nested below it is
reached when it is
+ * rewritten, since the join built for it goes through the routing in turn,
and one below a
+ * sub-query left in place is left with it, as everywhere else in this rule.
+ */
+ private def collectCarriedExistentialExprs(
+ expr: Expression,
+ collected: ArrayBuffer[Expression]): Unit = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return
+ }
+ expr match {
+ case sq @ Not(_: InSubquery) => collected += sq
+ case sq: Exists => collected += sq
+ case sq: InSubquery => collected += sq
+ case other => other.children.foreach(collectCarriedExistentialExprs(_,
collected))
+ }
+ }
+
+ /**
+ * 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(
exprs: Seq[Expression],
- plan: LogicalPlan): (Option[Expression], LogicalPlan, Seq[Attribute]) = {
+ plan: LogicalPlan,
+ canRewrite: Expression => Boolean = _ => true,
+ rejectsUnknownResult: Boolean = false): (Option[Expression], LogicalPlan,
+ Seq[Attribute]) = {
var newPlan = plan
val introducedAttrs = ArrayBuffer.empty[Attribute]
- val newExprs = exprs.map { e =>
- e.transformDownWithPruning(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY)) {
- case Exists(sub, _, _, conditions, subHint) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- val existenceJoin = ExistenceJoin(exists)
- val newCondition = conditions.reduceLeftOption(And)
- newPlan =
- buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
- existenceJoin, newCondition, subHint)
- introducedAttrs += exists
- exists
- case Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- // To handle a null-aware predicate not-in-subquery in nested
conditions
- // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we transform
- // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
- //
- // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
- // is transformed into a plan below;
- // == Optimized Logical Plan ==
- // Project [id#78, v#79]
- // +- Filter ((v#79 > 0) OR NOT exists#83)
- // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
- // :- Relation[id#78,v#79] parquet
- // +- Relation[id#80] parquet
- val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
- val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan,
- rewriteDomainJoinsIfPresent(newPlan, newSub, Some(finalJoinCond)),
- ExistenceJoin(exists), Some(finalJoinCond), joinHint)
- introducedAttrs += exists
- Not(exists)
- case InSubquery(values, ListQuery(sub, _, _, _, conditions, subHint))
=>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
- ExistenceJoin(exists), newConditions, joinHint)
- introducedAttrs += exists
- exists
+ // Builds the replacement of a sub-query expression under that
expression's origin and with
+ // its tags, as TreeNode.transformDownWithPruning does for the nodes a
rule replaces. Without
+ // it every node built below, the `exists` attribute and the join
conditions included, would
+ // take the origin of the enclosing operator, and a runtime error raised
from the rewritten
+ // condition would quote that operator rather than the sub-query fragment.
+ def replacing(sq: Expression)(replacement: => Expression): Expression = {
+ val newExpr = CurrentOrigin.withOrigin(sq.origin)(replacement)
+ newExpr.copyTagsFrom(sq)
+ newExpr
+ }
+ def rewrite(expr: Expression): Expression = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return expr
+ }
+ expr match {
+ case sq @ Exists(sub, _, _, conditions, subHint) if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ val existenceJoin = ExistenceJoin(exists)
+ val newCondition = conditions.reduceLeftOption(And)
+ newPlan = rewriteCarriedExistentialExprs(
+ buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
+ existenceJoin, newCondition, subHint),
+ rejectsUnknownResult)
+ introducedAttrs += exists
+ exists
+ }
+ case sq @ Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint)))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ // To handle a null-aware predicate not-in-subquery in nested
conditions
+ // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we
transform
+ // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
+ //
+ // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
+ // is transformed into a plan below;
+ // == Optimized Logical Plan ==
+ // Project [id#78, v#79]
+ // +- Filter ((v#79 > 0) OR NOT exists#83)
+ // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
+ // :- Relation[id#78,v#79] parquet
+ // +- Relation[id#80] parquet
+ val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
+ val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = rewriteCarriedExistentialExprs(Join(newPlan,
+ rewriteDomainJoinsIfPresent(newPlan, newSub,
Some(finalJoinCond)),
+ ExistenceJoin(exists), Some(finalJoinCond), joinHint),
rejectsUnknownResult)
+ introducedAttrs += exists
+ Not(exists)
+ }
+ case sq @ InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = rewriteCarriedExistentialExprs(
+ Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
+ ExistenceJoin(exists), newConditions, joinHint),
rejectsUnknownResult)
+ 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 = {
+ // Iterates the references and probes the output set, rather than
materialising the
+ // intersection: AttributeSet.intersect builds a set from its argument, so
the operand order
+ // there would decide whether this walks the references or the whole
output of `plan`. The
+ // membership probes still construct wrappers, so this is cheaper, not
allocation-free.
+ e.references.exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Returns true if `e` effectively references any of the attributes produced
by `plan`, see
+ * [[effectiveReferences]].
+ */
+ private def effectivelyReferencesPlan(e: Expression, plan: LogicalPlan):
Boolean = {
+ effectiveReferences(e).exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Which of an outer plan and a sub-query plan a set of references can be
evaluated by, as
+ * [[planReferences]] determines it.
+ */
+ private case class PlanReferences(subPlanOnly: Boolean, bothPlans: Boolean)
+
+ /**
+ * Classifies `references` against `outerPlan` and `subPlan` in a single
pass: whether they can
+ * only be evaluated by `subPlan`, and whether they need both plans.
+ *
+ * An attribute that both plans produce is attributed to `subPlan` rather
than to `outerPlan`,
+ * which is what the `subPlan` probe coming first does. A nested sub-query
resolves in the scope
+ * of the sub-query it is nested in, so an attribute both plans produce is
one of that
+ * sub-query's own.
+ *
+ * The probes are exclusive, so swapping their order would attribute such an
attribute to
+ * `outerPlan` alone: `subPlanOnly` and `bothPlans` would both be false, and
a nested sub-query
+ * whose only reference is the shared attribute would be declined and left
in the join condition
+ * rather than routed to the sub-query plan that can evaluate it.
+ *
+ * No plan the analyzer produces from SQL reaches this:
`DeduplicateRelations` has given the
+ * sub-query its own ExprIds, and the EXISTS arms classify against the raw
sub-query plan while
+ * the IN arms classify against the plan `dedupSubqueryOnSelfJoin` has
already separated from the
+ * outer plan. What does reach it is the shared-attribute fixture in
`RewriteSubquerySuite`,
+ * which builds the overlap directly because the analyzer is not involved
there. That fixture
+ * pins the shape, not the outcome of this ordering: under either order it
ends in the same
+ * error, since `dedupSubqueryOnSelfJoin`'s pre-existing conflict check
reports the duplicated
+ * attribute that the hoisted condition mentions.
+ *
+ * An attribute neither plan produces leaves both answers false, which
declines the rewrite here.
+ * That does not leave the sub-query unexamined: one carried by a nested
sub-query that is
+ * rewritten is routed against the inputs of the join built for it, see
+ * [[rewriteCarriedExistentialExprs]].
+ */
+ private def planReferences(
+ references: AttributeSet,
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): PlanReferences = {
+ var subPlanReference = false
+ var outerPlanOnlyReference = false
+ val attrs = references.iterator
+ // Stops as soon as both plans are known to be referenced, which is all
the two answers need.
+ while (attrs.hasNext && !(subPlanReference && outerPlanOnlyReference)) {
+ val attr = attrs.next()
+ if (subPlan.outputSet.contains(attr)) {
+ subPlanReference = true
+ } else if (outerPlan.outputSet.contains(attr)) {
+ outerPlanOnlyReference = true
+ }
+ }
+ PlanReferences(
+ subPlanOnly = subPlanReference && !outerPlanOnlyReference,
+ bothPlans = subPlanReference && outerPlanOnlyReference)
+ }
+
+ /**
+ * The attributes that `e` references and can still evaluate. For an
existential sub-query
+ * expression these are the ones of the values it compares and of the
correlated condition that
+ * [[PullupCorrelatedPredicates]] hoisted into it, rather than the ones of
`references`, which
+ * also include the outer attributes the pull-up retains for idempotency.
Those outlive the
+ * condition that referenced them once BooleanSimplification has eliminated
it, as in
+ * `a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)`, which keeps `c1`
as an outer
+ * attribute of a sub-query that no longer reads it, and which must still be
rewritten against
+ * the outer plan rather than treated as referencing the sub-query plan.
+ *
+ * The sets are merged with `AttributeSet.fromAttributeSets` rather than
built from a flattened
+ * sequence of attributes: `AttributeSet.apply` would take one allocation
per attribute of a
+ * sequence as long as the total number of references, which a row-valued IN
over a wide row
+ * makes unnecessarily large.
+ */
+ private def effectiveReferences(e: Expression): AttributeSet = e match {
+ case Exists(_, _, _, joinCond, _) =>
+ AttributeSet.fromAttributeSets(joinCond.map(_.references))
+ case Not(in: InSubquery) => effectiveReferences(in)
+ case InSubquery(values, ListQuery(_, _, _, _, joinCond, _)) =>
+ AttributeSet.fromAttributeSets(values.map(_.references) ++
joinCond.map(_.references))
+ case _ => e.references
+ }
+
+ /**
+ * 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, since neither alone can evaluate it, so it has to stay 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 that rewrite when
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled` is
disabled, and
+ * also when it is an uncorrelated IN sub-query and
+ *
`spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled`
is disabled,
+ * since it is then not among the sub-queries that rewrite considers.
+ *
+ * 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.
+ *
+ * The rejection covers the sub-queries the sub-query plan owns. An IN
sub-query that only
+ * references the outer plan is still rewritten two-valued, as it was before
SPARK-59351, which
+ * SPARK-59862 tracks.
+ *
+ * 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. This
accepts on the
+ // negation of the test above rather than on a positive one, which rests
on the invariant that
+ // `effectiveReferences` sees every attribute the rewrite can put in a
join condition: one it
+ // cannot see would be rewritten here and end up in a condition the outer
plan cannot evaluate.
+ // A positive test is not available without also refusing the sub-queries
whose retained outer
+ // attributes no longer correspond to a condition, which must keep being
rewritten here.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !effectivelyReferencesPlan(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,
+ keepsThreeValuedResiduals: Boolean = false): (Option[Expression],
LogicalPlan) = {
+ // The rejection collector below is a plain `collect`, which visits every
node of the hoisted
+ // conditions whether or not it can hold a sub-query, so skip the
classification entirely when
+ // they hold none at all, which is the common case for a correlated EXISTS
or IN.
+ // [[collectUnknownSensitiveInSubqueries]] prunes subquery-free subtrees
on its own, so this
+ // guard only saves it the top-level call.
+ if (!conditions.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return (conditions.reduceOption(And), subPlan)
+ }
+ // 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. The
+ // same holds for a correlated sub-query nested below one that is left
there, since the
+ // rewrite leaves a declined sub-query whole, children included, and
nothing else will reach
+ // it: see carriesCorrelatedSubquery. Note that this walks the whole
expression, including the
+ // join condition of a sub-query the rewrite declines to descend into,
which is deliberate for
+ // the same reason.
+ //
+ // Which plan a candidate references is asked of it up to three times
below, once for the
+ // rejection of a sub-query referencing both plans, once to keep only the
unknown-sensitive
+ // ones the sub-query plan owns, and once more by the rewrite itself. Both
answers come from
+ // the same reference set in a single pass over it, and the result is
memoised per candidate,
+ // so neither the set nor the scan is repeated. The memo is keyed by
identity: the candidates
+ // are nodes of the expressions being classified, and comparing two of
them structurally would
+ // compare the sub-query plans they hold.
+ val classified = new java.util.IdentityHashMap[Expression,
PlanReferences]()
+ def classify(sq: Expression): PlanReferences = {
+ var references = classified.get(sq)
+ if (references == null) {
+ references = planReferences(effectiveReferences(sq), outerPlan,
subPlan)
+ classified.put(sq, references)
+ }
+ references
+ }
+ def ownedBySubPlan(sq: Expression): Boolean = classify(sq).subPlanOnly
+ def referencesBothPlans(sq: Expression): Boolean = classify(sq).bothPlans
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if referencesBothPlans(sq) &&
+ (hasCorrelatedCondition(sq) || carriesCorrelatedSubquery(sq)) => 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. Unless
the caller keeps such
+ // a sub-query as a residual, in which case no existence join is built for
it and there is
+ // nothing to reject: see `keepsThreeValuedResiduals`.
+ val unknownSensitive = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, truthOnly = true))
+ .filter(ownedBySubPlan)
+ val (residual, unknownResult) = unknownSensitive.partition(sq =>
+ keepsThreeValuedResiduals && !hasCorrelatedCondition(sq) &&
!carriesCorrelatedSubquery(sq))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ def canRewrite(sq: Expression): Boolean =
+ ownedBySubPlan(sq) && !residual.exists(_.fastEquals(sq))
+ val (newCond, newSubPlan, _) =
+ rewriteExistentialExprWithAttrs(conditions, subPlan, canRewrite,
rejectsUnknownResult = true)
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and
whose value is observed
+ * beyond being TRUE, so that rewriting them into an existence join, whose
`exists` attribute is
+ * FALSE where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `truthOnly` states whether the value of `expr` is only ever tested for
being TRUE, which is
+ * where unknown and FALSE cannot be told apart. That holds for the operands
of AND and OR within
+ * a condition that ends up in a Filter or a join condition, and for the
positions
+ * [[ReplaceNullWithFalseInPredicate]] rewrites for the same reason: the
operand of an
+ * `EqualNullSafe` against TRUE, and the predicate of an `If` or of a
`CaseWhen` branch. The
+ * result branches of a conditional are consumed wherever the conditional
itself is, so they
+ * inherit its context rather than being value positions of their own.
+ *
+ * It does not hold under NOT, where `NOT FALSE` is TRUE while `NOT unknown`
is unknown. A NOT
+ * directly over an IN sub-query is still safe, but for another reason: it
is rewritten with a
+ * null-aware join condition of its own, which is TRUE where the comparison
is unknown, so the
+ * negation of its `exists` attribute is FALSE there, as SQL requires.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ truthOnly: Boolean): Seq[Expression] = {
+ val reported = ArrayBuffer.empty[Expression]
+ collectUnknownSensitiveInSubqueries(expr, truthOnly, reported)
+ reported.toSeq
+ }
+
+ /**
+ * Appends what [[unknownSensitiveInSubqueries]] reports to `reported`, in
the order the
+ * expression is walked. The results are accumulated rather than
concatenated on the way back up,
+ * so a condition with many reportable sub-queries does not copy the prefix
at every node.
+ */
+ private def collectUnknownSensitiveInSubqueries(
+ expr: Expression,
+ truthOnly: Boolean,
+ reported: ArrayBuffer[Expression]): Unit = expr match {
+ case _ if !expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY) =>
+ case And(left, right) =>
+ collectUnknownSensitiveInSubqueries(left, truthOnly, reported)
+ collectUnknownSensitiveInSubqueries(right, truthOnly, reported)
+ case Or(left, right) =>
+ collectUnknownSensitiveInSubqueries(left, truthOnly, reported)
+ collectUnknownSensitiveInSubqueries(right, truthOnly, reported)
+ // `null <=> true` and `false <=> true` are both FALSE, so only truth is
observed here whatever
+ // consumes the comparison.
+ case EqualNullSafe(left, Literal.TrueLiteral) =>
+ collectUnknownSensitiveInSubqueries(left, truthOnly = true, reported)
+ case EqualNullSafe(Literal.TrueLiteral, right) =>
+ collectUnknownSensitiveInSubqueries(right, truthOnly = true, reported)
+ // COALESCE(x, false) maps both unknown and FALSE to FALSE, so only truth
is observed. A
+ // different default does not: COALESCE(x, true) is TRUE where x is
unknown and FALSE where x
+ // is FALSE.
+ case Coalesce(child +: rest) if
rest.headOption.contains(Literal.FalseLiteral) =>
+ collectUnknownSensitiveInSubqueries(child, truthOnly = true, reported)
+ case If(predicate, trueValue, falseValue) =>
+ collectUnknownSensitiveInSubqueries(predicate, truthOnly = true,
reported)
+ collectUnknownSensitiveInSubqueries(trueValue, truthOnly, reported)
+ collectUnknownSensitiveInSubqueries(falseValue, truthOnly, reported)
+ case CaseWhen(branches, elseValue) =>
+ branches.foreach { case (condition, value) =>
+ collectUnknownSensitiveInSubqueries(condition, truthOnly = true,
reported)
+ collectUnknownSensitiveInSubqueries(value, truthOnly, reported)
+ }
+ elseValue.foreach(collectUnknownSensitiveInSubqueries(_, truthOnly,
reported))
+ // A NOT directly over an IN sub-query is handled as that IN is, by
`collectOrDescend`: it is
+ // reported in a value position, where the unknown its null-aware rewrite
preserves would be
Review Comment:
**Nit (P3):** The null-aware rewrite does not preserve the unknown. Its
condition `c OR isnull(c)` makes `exists` TRUE where the comparison is unknown,
so `NOT exists` is FALSE there. That is why the
`rewriteExistentialExprInJoinCondition` scaladoc calls it 'equally two-valued'.
For `(c1 NOT IN (...)) <=> false` with a non-matching NULL, SQL gives FALSE but
the rewrite would give TRUE. What a value position would observe is the FALSE
the rewrite yields in place of unknown, so the comment could say 'where the
FALSE its null-aware rewrite yields in place of unknown would be observable'.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#discussion_r4281437515).
**Verification:**
- **Inspection:** The collector's NOT IN comment states that the null-aware
rewrite turns an unknown NOT IN into FALSE, and that this is why a
value-position NOT IN is reported.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +417,649 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, a correlated IN or NOT IN is planned
by `PlanSubqueries`
+ * as an in-subquery filter that drops its own hoisted condition, which
returns extra rows, and
+ * a correlated EXISTS reaches execution as an unevaluable `Exists` and
fails.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is routed as well, rather than left for that
handling to rewrite
+ * under its default configuration: it has no hoisted condition to lose, but
it is subject to the
+ * same unknown-result guard as one on the sub-query-plan route, and that
handling has no such
+ * guard. Consuming it here is what keeps an IN whose result is observably
unknown from becoming
+ * a two-valued `exists` attribute.
+ *
+ * The classification mirrors [[rewriteExistentialExprInSubqueryPlan]], with
this join's inputs
+ * in place of the outer and the sub-query plan: a sub-query referencing
exactly one input is
+ * routed against it, one referencing both is rejected when it is correlated
or carries a
+ * correlated sub-query, since neither input alone can evaluate it, and one
whose unknown result
+ * is observable is rejected. A sub-query referencing both inputs or neither
and carrying no
+ * correlation is left where it is, as it is at the first level.
+ *
+ * The unknown-result rejection applies only when `rejectsUnknownResult` is
set, which is the
+ * case for the existence joins built on the sub-query-plan route, where the
same shapes failed
+ * with an internal error. On the routes that predate this change an
uncorrelated IN whose
+ * unknown result is observable is left where it is, as it was, and a
correlated one is routed
+ * without the rejection. Their two-valued handling is tracked by
SPARK-59862.
+ *
+ * A NOT IN is routed before a bare IN, since the substitution of a unit
replaces every
+ * occurrence of it, including one inside a `Not` over the same instance.
+ */
+ private def rewriteCarriedExistentialExprs(
+ join: Join,
+ rejectsUnknownResult: Boolean): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ val carried = ArrayBuffer.empty[Expression]
+ collectCarriedExistentialExprs(condition, carried)
+ if (carried.isEmpty) {
+ return join
+ }
+ // A NOT IN is carried as the Not over the IN, which is the unit the
rewrite consumes and the
+ // unit the unknown-result collector reports, so both agree on the node
below.
+ def negatedIn(sq: Expression): Expression = sq match {
+ case Not(in: InSubquery) => in
+ case other => other
+ }
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+ def referencesBothInputs(sq: Expression): Boolean =
+ referencesInput(sq, join.left) && referencesInput(sq, join.right)
+ def referencesOneInput(sq: Expression): Boolean =
+ referencesInput(sq, join.left) != referencesInput(sq, join.right)
+
+ // The two rejections of the sub-query-plan route apply here for the same
reasons, against the
+ // inputs of this join rather than against the outer and the sub-query
plan, and each raises
+ // the condition whose message describes it.
+ val referencingBothInputs = carried.filter { sq =>
+ referencesBothInputs(sq) &&
+ (hasCorrelatedCondition(sq) ||
carriesCorrelatedSubquery(negatedIn(sq)))
+ }
+ if (referencingBothInputs.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothInputs.toSeq)
+ }
+ // Only the sub-query-plan route rejects. An existence join built on
another route answered
+ // such a sub-query before this routing existed, three-valued where it was
left in place, and
+ // that is left to SPARK-59862: see `rejectsUnknownResult`.
+ def uncorrelated(sq: Expression): Boolean =
+ !hasCorrelatedCondition(sq) && !carriesCorrelatedSubquery(negatedIn(sq))
+ if (rejectsUnknownResult) {
+ val unknownResult = carried.filter(sq => referencesOneInput(sq) &&
observesUnknown(sq))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult.toSeq)
+ }
+ }
+ // A NOT IN unit goes before a bare IN: the substitution of a unit
replaces every occurrence
+ // of it in the condition, so a bare IN routed first would also replace
the one inside a
+ // `Not` over the same instance, and the null-aware join built for that
`Not` would be left
+ // unreferenced. The sort is stable, so the order within each kind is the
order collected.
+ val routed = carried.sortBy {
+ case Not(_: InSubquery) => 0
+ case _ => 1
+ }.filterNot(sq => !rejectsUnknownResult && observesUnknown(sq) &&
uncorrelated(sq))
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ routed.filter(sq => referencesOneInput(sq) &&
isInput(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
+ rewriteExistentialExprWithAttrs(Seq(sq), p, rejectsUnknownResult =
rejectsUnknownResult)
+ newCondition = newCondition.transform {
+ case e if e.fastEquals(sq) => rewritten.get
+ }
+ newInput
+ }
+ }
+ val newLeft = route(join.left, referencesInput(_, join.left))
+ val newRight = route(join.right, referencesInput(_, join.right))
+ if (newLeft.fastEquals(join.left) && newRight.fastEquals(join.right)) {
+ join
+ } else {
+ // Drop the `exists` attributes the nested existence joins introduced,
keeping this join's
+ // own output, which already carries its own.
+ Project(join.output, join.copy(
+ left = newLeft, right = newRight, condition = Some(newCondition)))
+ }
+ }
+
+ /**
+ * Collects the existential sub-queries carried by the condition of an
existence join, as the
+ * units [[rewriteExistentialExprWithAttrs]] rewrites: a NOT IN is collected
as the [[Not]] over
+ * the [[InSubquery]], so the rewrite takes its null-aware arm and the
unknown-result collector,
+ * which reports the same node, can be compared against it.
+ *
+ * A collected sub-query is not descended into. One nested below it is
reached when it is
+ * rewritten, since the join built for it goes through the routing in turn,
and one below a
+ * sub-query left in place is left with it, as everywhere else in this rule.
+ */
+ private def collectCarriedExistentialExprs(
+ expr: Expression,
+ collected: ArrayBuffer[Expression]): Unit = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return
+ }
+ expr match {
+ case sq @ Not(_: InSubquery) => collected += sq
+ case sq: Exists => collected += sq
+ case sq: InSubquery => collected += sq
+ case other => other.children.foreach(collectCarriedExistentialExprs(_,
collected))
+ }
+ }
+
+ /**
+ * 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(
exprs: Seq[Expression],
- plan: LogicalPlan): (Option[Expression], LogicalPlan, Seq[Attribute]) = {
+ plan: LogicalPlan,
+ canRewrite: Expression => Boolean = _ => true,
+ rejectsUnknownResult: Boolean = false): (Option[Expression], LogicalPlan,
+ Seq[Attribute]) = {
var newPlan = plan
val introducedAttrs = ArrayBuffer.empty[Attribute]
- val newExprs = exprs.map { e =>
- e.transformDownWithPruning(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY)) {
- case Exists(sub, _, _, conditions, subHint) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- val existenceJoin = ExistenceJoin(exists)
- val newCondition = conditions.reduceLeftOption(And)
- newPlan =
- buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
- existenceJoin, newCondition, subHint)
- introducedAttrs += exists
- exists
- case Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))) =>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- // To handle a null-aware predicate not-in-subquery in nested
conditions
- // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we transform
- // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
- //
- // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
- // is transformed into a plan below;
- // == Optimized Logical Plan ==
- // Project [id#78, v#79]
- // +- Filter ((v#79 > 0) OR NOT exists#83)
- // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
- // :- Relation[id#78,v#79] parquet
- // +- Relation[id#80] parquet
- val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
- val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan,
- rewriteDomainJoinsIfPresent(newPlan, newSub, Some(finalJoinCond)),
- ExistenceJoin(exists), Some(finalJoinCond), joinHint)
- introducedAttrs += exists
- Not(exists)
- case InSubquery(values, ListQuery(sub, _, _, _, conditions, subHint))
=>
- val exists = AttributeReference("exists", BooleanType, nullable =
false)()
- // Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
- val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
- val joinHint = JoinHint(None, subHint)
- newPlan = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
- ExistenceJoin(exists), newConditions, joinHint)
- introducedAttrs += exists
- exists
+ // Builds the replacement of a sub-query expression under that
expression's origin and with
+ // its tags, as TreeNode.transformDownWithPruning does for the nodes a
rule replaces. Without
+ // it every node built below, the `exists` attribute and the join
conditions included, would
+ // take the origin of the enclosing operator, and a runtime error raised
from the rewritten
+ // condition would quote that operator rather than the sub-query fragment.
+ def replacing(sq: Expression)(replacement: => Expression): Expression = {
+ val newExpr = CurrentOrigin.withOrigin(sq.origin)(replacement)
+ newExpr.copyTagsFrom(sq)
+ newExpr
+ }
+ def rewrite(expr: Expression): Expression = {
+ if (!expr.containsAnyPattern(EXISTS_SUBQUERY, IN_SUBQUERY)) {
+ return expr
+ }
+ expr match {
+ case sq @ Exists(sub, _, _, conditions, subHint) if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ val existenceJoin = ExistenceJoin(exists)
+ val newCondition = conditions.reduceLeftOption(And)
+ newPlan = rewriteCarriedExistentialExprs(
+ buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
+ existenceJoin, newCondition, subHint),
+ rejectsUnknownResult)
+ introducedAttrs += exists
+ exists
+ }
+ case sq @ Not(InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint)))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ // To handle a null-aware predicate not-in-subquery in nested
conditions
+ // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we
transform
+ // `inCondition` (t1.id=t2.id) into `(inCondition) OR
ISNULL(inCondition)`.
+ //
+ // For example, `SELECT * FROM t1 WHERE v > 0 OR t1.id NOT IN
(SELECT id FROM t2)`
+ // is transformed into a plan below;
+ // == Optimized Logical Plan ==
+ // Project [id#78, v#79]
+ // +- Filter ((v#79 > 0) OR NOT exists#83)
+ // +- Join ExistenceJoin(exists#83), ((id#78 = id#80) OR
isnull((id#78 = id#80)))
+ // :- Relation[id#78,v#79] parquet
+ // +- Relation[id#80] parquet
+ val nullAwareJoinConds = inConditions.map(c => Or(c, IsNull(c)))
+ val finalJoinCond = (nullAwareJoinConds ++
conditions).reduceLeft(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = rewriteCarriedExistentialExprs(Join(newPlan,
+ rewriteDomainJoinsIfPresent(newPlan, newSub,
Some(finalJoinCond)),
+ ExistenceJoin(exists), Some(finalJoinCond), joinHint),
rejectsUnknownResult)
+ introducedAttrs += exists
+ Not(exists)
+ }
+ case sq @ InSubquery(values, ListQuery(sub, _, _, _, conditions,
subHint))
+ if canRewrite(sq) =>
+ replacing(sq) {
+ val exists = AttributeReference("exists", BooleanType, nullable =
false)()
+ // Deduplicate conflicting attributes if any.
+ val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
+ val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val newConditions = (inConditions ++
conditions).reduceLeftOption(And)
+ val joinHint = JoinHint(None, subHint)
+ newPlan = rewriteCarriedExistentialExprs(
+ Join(newPlan, rewriteDomainJoinsIfPresent(newPlan, newSub,
newConditions),
+ ExistenceJoin(exists), newConditions, joinHint),
rejectsUnknownResult)
+ 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 = {
+ // Iterates the references and probes the output set, rather than
materialising the
+ // intersection: AttributeSet.intersect builds a set from its argument, so
the operand order
+ // there would decide whether this walks the references or the whole
output of `plan`. The
+ // membership probes still construct wrappers, so this is cheaper, not
allocation-free.
+ e.references.exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Returns true if `e` effectively references any of the attributes produced
by `plan`, see
+ * [[effectiveReferences]].
+ */
+ private def effectivelyReferencesPlan(e: Expression, plan: LogicalPlan):
Boolean = {
+ effectiveReferences(e).exists(plan.outputSet.contains)
+ }
+
+ /**
+ * Which of an outer plan and a sub-query plan a set of references can be
evaluated by, as
+ * [[planReferences]] determines it.
+ */
+ private case class PlanReferences(subPlanOnly: Boolean, bothPlans: Boolean)
+
+ /**
+ * Classifies `references` against `outerPlan` and `subPlan` in a single
pass: whether they can
+ * only be evaluated by `subPlan`, and whether they need both plans.
+ *
+ * An attribute that both plans produce is attributed to `subPlan` rather
than to `outerPlan`,
+ * which is what the `subPlan` probe coming first does. A nested sub-query
resolves in the scope
+ * of the sub-query it is nested in, so an attribute both plans produce is
one of that
+ * sub-query's own.
+ *
+ * The probes are exclusive, so swapping their order would attribute such an
attribute to
+ * `outerPlan` alone: `subPlanOnly` and `bothPlans` would both be false, and
a nested sub-query
+ * whose only reference is the shared attribute would be declined and left
in the join condition
+ * rather than routed to the sub-query plan that can evaluate it.
+ *
+ * No plan the analyzer produces from SQL reaches this:
`DeduplicateRelations` has given the
+ * sub-query its own ExprIds, and the EXISTS arms classify against the raw
sub-query plan while
+ * the IN arms classify against the plan `dedupSubqueryOnSelfJoin` has
already separated from the
+ * outer plan. What does reach it is the shared-attribute fixture in
`RewriteSubquerySuite`,
+ * which builds the overlap directly because the analyzer is not involved
there. That fixture
+ * pins the shape, not the outcome of this ordering: under either order it
ends in the same
+ * error, since `dedupSubqueryOnSelfJoin`'s pre-existing conflict check
reports the duplicated
+ * attribute that the hoisted condition mentions.
+ *
+ * An attribute neither plan produces leaves both answers false, which
declines the rewrite here.
+ * That does not leave the sub-query unexamined: one carried by a nested
sub-query that is
+ * rewritten is routed against the inputs of the join built for it, see
+ * [[rewriteCarriedExistentialExprs]].
+ */
+ private def planReferences(
+ references: AttributeSet,
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): PlanReferences = {
+ var subPlanReference = false
+ var outerPlanOnlyReference = false
+ val attrs = references.iterator
+ // Stops as soon as both plans are known to be referenced, which is all
the two answers need.
+ while (attrs.hasNext && !(subPlanReference && outerPlanOnlyReference)) {
+ val attr = attrs.next()
+ if (subPlan.outputSet.contains(attr)) {
+ subPlanReference = true
+ } else if (outerPlan.outputSet.contains(attr)) {
+ outerPlanOnlyReference = true
+ }
+ }
+ PlanReferences(
+ subPlanOnly = subPlanReference && !outerPlanOnlyReference,
+ bothPlans = subPlanReference && outerPlanOnlyReference)
+ }
+
+ /**
+ * The attributes that `e` references and can still evaluate. For an
existential sub-query
+ * expression these are the ones of the values it compares and of the
correlated condition that
+ * [[PullupCorrelatedPredicates]] hoisted into it, rather than the ones of
`references`, which
+ * also include the outer attributes the pull-up retains for idempotency.
Those outlive the
+ * condition that referenced them once BooleanSimplification has eliminated
it, as in
+ * `a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)`, which keeps `c1`
as an outer
+ * attribute of a sub-query that no longer reads it, and which must still be
rewritten against
+ * the outer plan rather than treated as referencing the sub-query plan.
+ *
+ * The sets are merged with `AttributeSet.fromAttributeSets` rather than
built from a flattened
+ * sequence of attributes: `AttributeSet.apply` would take one allocation
per attribute of a
+ * sequence as long as the total number of references, which a row-valued IN
over a wide row
+ * makes unnecessarily large.
+ */
+ private def effectiveReferences(e: Expression): AttributeSet = e match {
+ case Exists(_, _, _, joinCond, _) =>
+ AttributeSet.fromAttributeSets(joinCond.map(_.references))
+ case Not(in: InSubquery) => effectiveReferences(in)
+ case InSubquery(values, ListQuery(_, _, _, _, joinCond, _)) =>
+ AttributeSet.fromAttributeSets(values.map(_.references) ++
joinCond.map(_.references))
+ case _ => e.references
+ }
+
+ /**
+ * 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, since neither alone can evaluate it, so it has to stay 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 that rewrite when
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled` is
disabled, and
+ * also when it is an uncorrelated IN sub-query and
+ *
`spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled`
is disabled,
+ * since it is then not among the sub-queries that rewrite considers.
+ *
+ * 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.
+ *
+ * The rejection covers the sub-queries the sub-query plan owns. An IN
sub-query that only
+ * references the outer plan is still rewritten two-valued, as it was before
SPARK-59351, which
+ * SPARK-59862 tracks.
+ *
+ * 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. This
accepts on the
+ // negation of the test above rather than on a positive one, which rests
on the invariant that
+ // `effectiveReferences` sees every attribute the rewrite can put in a
join condition: one it
+ // cannot see would be rewritten here and end up in a condition the outer
plan cannot evaluate.
+ // A positive test is not available without also refusing the sub-queries
whose retained outer
+ // attributes no longer correspond to a condition, which must keep being
rewritten here.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !effectivelyReferencesPlan(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,
+ keepsThreeValuedResiduals: Boolean = false): (Option[Expression],
LogicalPlan) = {
+ // The rejection collector below is a plain `collect`, which visits every
node of the hoisted
+ // conditions whether or not it can hold a sub-query, so skip the
classification entirely when
+ // they hold none at all, which is the common case for a correlated EXISTS
or IN.
+ // [[collectUnknownSensitiveInSubqueries]] prunes subquery-free subtrees
on its own, so this
+ // guard only saves it the top-level call.
+ if (!conditions.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return (conditions.reduceOption(And), subPlan)
+ }
+ // 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. The
+ // same holds for a correlated sub-query nested below one that is left
there, since the
+ // rewrite leaves a declined sub-query whole, children included, and
nothing else will reach
+ // it: see carriesCorrelatedSubquery. Note that this walks the whole
expression, including the
+ // join condition of a sub-query the rewrite declines to descend into,
which is deliberate for
+ // the same reason.
+ //
+ // Which plan a candidate references is asked of it up to three times
below, once for the
+ // rejection of a sub-query referencing both plans, once to keep only the
unknown-sensitive
+ // ones the sub-query plan owns, and once more by the rewrite itself. Both
answers come from
+ // the same reference set in a single pass over it, and the result is
memoised per candidate,
+ // so neither the set nor the scan is repeated. The memo is keyed by
identity: the candidates
+ // are nodes of the expressions being classified, and comparing two of
them structurally would
+ // compare the sub-query plans they hold.
+ val classified = new java.util.IdentityHashMap[Expression,
PlanReferences]()
+ def classify(sq: Expression): PlanReferences = {
+ var references = classified.get(sq)
+ if (references == null) {
+ references = planReferences(effectiveReferences(sq), outerPlan,
subPlan)
+ classified.put(sq, references)
+ }
+ references
+ }
+ def ownedBySubPlan(sq: Expression): Boolean = classify(sq).subPlanOnly
+ def referencesBothPlans(sq: Expression): Boolean = classify(sq).bothPlans
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if referencesBothPlans(sq) &&
+ (hasCorrelatedCondition(sq) || carriesCorrelatedSubquery(sq)) => 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. Unless
the caller keeps such
+ // a sub-query as a residual, in which case no existence join is built for
it and there is
+ // nothing to reject: see `keepsThreeValuedResiduals`.
+ val unknownSensitive = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, truthOnly = true))
+ .filter(ownedBySubPlan)
+ val (residual, unknownResult) = unknownSensitive.partition(sq =>
Review Comment:
**Non-blocking (P2):** A truth-only, uncorrelated nested NOT IN with a
nullable comparison is now rewritten here into a null-aware existence join on
the sub-query plan, where the merge base left it in the anti-join condition.
For example, in `SELECT * FROM t1 WHERE a NOT IN (SELECT c1 FROM t2 WHERE c1 >
a OR c1 NOT IN (SELECT col1 FROM t3))`, the join condition `(c1 = col1) OR
isnull(c1 = col1)` has no equi key, so it is planned as a
BroadcastNestedLoopJoin over t2 x t3. At the merge base, InSubqueryExec
answered the nested NOT IN three-valued with a hash-set probe. The answer is
unchanged, but the query now does |t2| x |t3| extra condition evaluations, a
cost that grows with the nested list and buys no correctness.
This applies whenever no later join-condition rewrite reaches the anti join:
when this NOT IN is the last sub-query conjunct of its Filter, or when
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled` or
`spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled`
is off. Under either setting off, the carried router does the same to
uncorrelated truth-only units on the pre-existing routes. Where a later
sub-query conjunct wraps the anti join under the default configuration, the
`case j: Join` arm would rewrite the IN under the NOT positively and make the
nested NOT IN two-valued, so the null-aware rewrite is needed there.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#discussion_r4281437494).
**Recommended change:** Keep uncorrelated nested IN and NOT IN sub-queries
where the merge base evaluated them correctly, and rewrite them only where the
fix needs it. In the NOT IN arm, keep as a residual in the anti-join condition
every uncorrelated, sub-query-plan-owned nested IN or NOT IN that carries no
correlated sub-query, not only the ones whose unknown is observable, whenever
no later join-condition rewrite reaches that anti join: when this NOT IN is the
last sub-query conjunct of its Filter, or when
spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled or
spark.sql.optimizer.optimizeUncorrelatedInSubqueriesInJoinCondition.enabled is
false. Otherwise keep today's handling, whose null-aware rewrite of a
truth-only nested NOT IN is what keeps the join-condition rewrite from making
it two-valued. In the carried router, on routes that do not reject an unknown
result, leave an uncorrelated carried IN or NOT IN in place whenever the
join-condition handling w
ould not rewrite it, that is when either of those settings is false. Update
the NOT IN residual test so its truth-only case asserts the residual instead of
an existence join, and add plan assertions for a nested NOT IN with a nullable
comparison in the predicate-subquery tests.
**Why this works:** InSubqueryExec already evaluates an uncorrelated
sub-query that stays in a bare anti-join or existence-join condition
three-valued, with a hash-set probe, and needs no existence join. Leaving such
sub-queries in place wherever no later join-condition rewrite reaches them
reproduces the merge-base plan exactly where the merge base was already
correct. Where that rewrite does reach them, the existing null-aware rewrite
stays. Correlated sub-queries, sub-queries carrying a correlated descendant,
and every sub-query on the sub-query-plan route still go through the existing
routing and rejections.
**Scope:** Residual and in-place policy for uncorrelated nested IN and NOT
IN in RewritePredicateSubquery, with updated NOT IN residual and carried-route
tests.
**Compatibility:** Every result this head already gets right stays the same.
The sub-queries this repair leaves in place are evaluated three-valued by
InSubqueryExec, as at the merge base, and only where no later join-condition
rewrite reaches them; every other one keeps today's rewrite. Only plans differ,
and only where the conjunct position or the join-condition settings already
decide whether uncorrelated IN sub-queries in join conditions are rewritten.
**Risks:** Plans for these shapes go back to InSubqueryExec, which collects
and broadcasts the nested list on the driver whatever its size, as at the merge
base. The change can be rolled back by reverting the residual policy alone.
Reading the join-condition settings in the carried router duplicates the gating
of the join-condition arm, so the two must stay aligned if that gating changes.
Keying the NOT IN arm's residuals on its position in the conjunct fold ties the
arm to that fold order, so the two must stay aligned if the fold changes.
**Constraints:** Correlated nested sub-queries, and those carrying a
correlated descendant, must still be routed or rejected, because PlanSubqueries
drops their hoisted condition. The EXISTS, NOT EXISTS and IN arms and the
sub-query-plan route keep their current routing and unknown-result rejection.
An Exists is never kept as a residual. Under the default configuration, a
truth-only uncorrelated nested NOT IN under an outer NOT IN that a later
sub-query conjunct wraps keeps its null-aware rewrite.
**Exclusions:**
- An unknown-observing residual under an outer NOT IN that a later sub-query
conjunct wraps is still made two-valued by the join-condition rewrite under the
default configuration, as at the merge base; that is left to SPARK-59862.
**Success:** An uncorrelated, truth-only, sub-query-plan-owned nested IN or
NOT IN under an outer NOT IN that is the last sub-query conjunct is evaluated
in the anti-join condition through InSubqueryExec, and no existence join is
built for it. Under the default configuration, a truth-only uncorrelated nested
NOT IN under an outer NOT IN followed by another sub-query conjunct is still
rewritten null-aware and returns the standard SQL answer. With either
join-condition setting disabled, an uncorrelated carried IN or NOT IN on the
Filter-predicate, projection, join-condition or outer route stays in its
existence-join condition. Every predicate-subquery test answer is unchanged.
**Verification:**
- **Regression:** For SELECT * FROM t1 WHERE a NOT IN (SELECT c1 FROM t2
WHERE c1 > a OR c1 NOT IN (SELECT col1 FROM t3)) with a nullable t3.col1, the
executed plan contains no nested-loop existence join and plans the nested
sub-query as an in-subquery filter, and the answer matches standard SQL.
- **Regression:** For a truth-only uncorrelated nested positive IN under an
outer NOT IN, no existence join is built on the sub-query side and the answer
is unchanged.
- **Behavior:** With
decorrelatePredicateSubqueriesInJoinPredicate.enabled=false, and separately
with optimizeUncorrelatedInSubqueriesInJoinCondition.enabled=false, an
uncorrelated carried NOT IN with a nullable comparison on the Filter-predicate
route stays in its existence-join condition, and the query returns the standard
SQL answer.
- **Behavior:** Under the default configuration, SELECT * FROM t1 WHERE a
NOT IN (SELECT c1 FROM t2 WHERE c1 > a OR c1 NOT IN (SELECT col1 FROM t3n)) AND
EXISTS (SELECT 1 FROM t2 WHERE c1 >= a), with a NULL in t3n that matches
nothing, returns the standard SQL answer, as it does on this head.
- **Compatibility:** Correlated nested sub-queries under an outer NOT IN are
still rewritten or rejected as today. The carried NOT IN, shared bare-and-NOT
IN and carried routing cases keep their answers under both values of each
join-condition setting.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -184,8 +188,23 @@ object RewritePredicateSubquery extends Rule[LogicalPlan]
with PredicateHelper {
// Use EXISTS if performance matters to you.
// Deduplicate conflicting attributes if any.
- val newSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
- val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
+ val dedupSub = dedupSubqueryOnSelfJoin(p, sub, Some(values))
+ val inConditions = values.zip(dedupSub.output).map(EqualTo.tupled)
+ // The hoisted correlated predicates may carry existential
sub-queries that only the
+ // sub-query plan can evaluate, see
rewriteExistentialExprInJoinCondition.
+ // This arm returns a bare Join, which this rule does not offer to
its own handling of
+ // predicate sub-queries in join conditions, so an uncorrelated
sub-query left in the
+ // anti-join condition stays there and `InSubqueryExec` evaluates it
with the
+ // three-valued semantics an existence join cannot carry. That is
what this arm did
+ // before SPARK-59351, so it is not rejected here. It only holds
while this NOT IN is
+ // the last sub-query conjunct of the Filter: the fold over the
conjuncts wraps this
+ // anti join as `p` of a later EXISTS or IN arm, in
`Project(p.output, Join(p, ...))`,
Review Comment:
**Nit (P3):** The last-conjunct condition is right, but any later sub-query
conjunct wraps this anti join, not only an EXISTS or IN. A later NOT IN arm
builds a bare `Join` with this anti join as its left input, and the generic
`case (p, predicate)` arm builds `Project(p.output, Filter(..., existence joins
over p))`. In both cases the top-down pass then reaches the anti join in the
`case j: Join` arm. For example, in `a NOT IN (...) AND b NOT IN (...)`, the
first NOT IN's residual is rewritten two-valued under the default
configuration. Saying 'a later sub-query conjunct' here, and in the comment of
the 'NOT IN keeps an uncorrelated three-valued residual' test, would cover
these cases.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#discussion_r4346076714).
**Verification:**
- **Inspection:** The comments' explanation covers a later NOT IN conjunct
and a generic predicate conjunct as well as EXISTS and IN, matching how the
fold builds each arm's result.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +417,649 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, a correlated IN or NOT IN is planned
by `PlanSubqueries`
+ * as an in-subquery filter that drops its own hoisted condition, which
returns extra rows, and
+ * a correlated EXISTS reaches execution as an unevaluable `Exists` and
fails.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is routed as well, rather than left for that
handling to rewrite
+ * under its default configuration: it has no hoisted condition to lose, but
it is subject to the
+ * same unknown-result guard as one on the sub-query-plan route, and that
handling has no such
+ * guard. Consuming it here is what keeps an IN whose result is observably
unknown from becoming
+ * a two-valued `exists` attribute.
+ *
+ * The classification mirrors [[rewriteExistentialExprInSubqueryPlan]], with
this join's inputs
+ * in place of the outer and the sub-query plan: a sub-query referencing
exactly one input is
+ * routed against it, one referencing both is rejected when it is correlated
or carries a
+ * correlated sub-query, since neither input alone can evaluate it, and one
whose unknown result
+ * is observable is rejected. A sub-query referencing both inputs or neither
and carrying no
+ * correlation is left where it is, as it is at the first level.
+ *
+ * The unknown-result rejection applies only when `rejectsUnknownResult` is
set, which is the
+ * case for the existence joins built on the sub-query-plan route, where the
same shapes failed
+ * with an internal error. On the routes that predate this change an
uncorrelated IN whose
+ * unknown result is observable is left where it is, as it was, and a
correlated one is routed
+ * without the rejection. Their two-valued handling is tracked by
SPARK-59862.
+ *
+ * A NOT IN is routed before a bare IN, since the substitution of a unit
replaces every
+ * occurrence of it, including one inside a `Not` over the same instance.
+ */
+ private def rewriteCarriedExistentialExprs(
+ join: Join,
+ rejectsUnknownResult: Boolean): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ val carried = ArrayBuffer.empty[Expression]
+ collectCarriedExistentialExprs(condition, carried)
+ if (carried.isEmpty) {
+ return join
+ }
+ // A NOT IN is carried as the Not over the IN, which is the unit the
rewrite consumes and the
+ // unit the unknown-result collector reports, so both agree on the node
below.
+ def negatedIn(sq: Expression): Expression = sq match {
+ case Not(in: InSubquery) => in
+ case other => other
+ }
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
Review Comment:
**Nit (P3):** `referencesInput` rebuilds `effectiveReferences(sq)` and
probes one input on every call. Through `referencesBothInputs`,
`referencesOneInput` and the two `route` filters, that happens about ten times
per carried unit, although `join.left` and `join.right` are fixed for the whole
call. Each rebuild is proportional to the unit's reference count, so a wide
row-valued IN pays for it repeatedly. `rewriteExistentialExprInSubqueryPlan`
already classifies each candidate once and memoizes the result. Computing each
unit's effective references once here, and deriving the left, right, one-input
and both-inputs answers from that result, would match it.
This extends an [existing
discussion](https://github.com/apache/spark/pull/58656#discussion_r4248986929).
**Verification:**
- **Inspection:** Every left or right input classification of a carried unit
reads one per-unit result rather than rebuilding the effective-reference set.
- **Compatibility:** Routing, both rejections and the in-place filter give
the same results for the carried routing, carried NOT IN, shared IN and
both-inputs cases.
##########
sql/core/src/test/scala/org/apache/spark/sql/SubquerySuite.scala:
##########
@@ -3227,4 +3224,822 @@ class SubquerySuite extends SharedSparkSession
}
}
}
+
+ /**
+ * Checks the result of `query`, that every node of its optimized plan can
produce the attributes
+ * it references, and that the existence join a nested subquery was
rewritten into sits on the
+ * subquery side of the semi/anti join.
+ *
+ * The plan checks are what SPARK-59351 is about: an existence join whose
condition references an
+ * attribute produced by neither of its children can still return the right
answer when the
+ * nested relation is empty at runtime, because the invalid condition is
then never bound, and a
+ * rewrite that placed the join on the other side would return the right
answer here too. Note
+ * that missingInput cannot see a subquery expression that survives in a
condition, since
+ * QueryPlan.references folds only the outer attributes of one and collect
does not walk its
+ * plan, so it is not a substitute for asserting the shape.
+ */
+ private def checkAnswerAndPlan(
+ query: String,
+ expected: Seq[Row],
+ existenceJoinOnSubqueryPlan: Boolean = true): Unit = {
+ val df = sql(query)
+ val plan = df.queryExecution.optimizedPlan
+ val invalidNodes = plan.collect { case p if p.missingInput.nonEmpty => p }
+ assert(invalidNodes.isEmpty,
+ s"""Plan nodes reference non-reachable attributes:
+ |${invalidNodes.mkString("\n")}
+ |$plan""".stripMargin)
+ if (existenceJoinOnSubqueryPlan) {
+ val semiOrAnti = plan.collectFirst {
+ case j: Join if j.joinType == LeftSemi || j.joinType == LeftAnti => j
+ }
+ assert(semiOrAnti.nonEmpty, s"expected a semi or anti join in\n$plan")
+ val existenceJoins = plan.collect {
+ case j: Join if j.joinType.isInstanceOf[ExistenceJoin] => j
+ }
+ assert(existenceJoins.nonEmpty, s"expected an existence join in\n$plan")
+ // The existence join must be under the right side of the semi/anti
join, which is the
+ // subquery plan, and not under its left side, which is the outer plan.
+ assert(existenceJoins.forall(j =>
semiOrAnti.get.right.exists(_.fastEquals(j))),
+ s"the existence join must sit on the subquery side in\n$plan")
+ }
+ checkAnswer(df, expected)
+ }
+
+ test("SPARK-59351: nested subquery referencing the inner query becomes an
existence join") {
+ // SPARK-45580 covers the case where the nested subquery references the
outer query, in which
+ // case its existence join is built on top of the outer plan. Here the
nested subquery
+ // references the query it is nested in, so the existence join has to be
built on top of the
+ // subquery plan instead.
+ withTempView("t1", "t2", "t3", "t3n", "t4", "t5") {
+ Seq((1), (2), (3), (7),
(9)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ // t3 shares a value with t2, t5 does not, so the nested subquery
decides the answer below.
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ Seq((3), (7)).toDF("col1").persist().createOrReplaceTempView("t5")
+ Seq(Some(3), Some(9),
None).toDF("col1").persist().createOrReplaceTempView("t3n")
+ // A correlated nested IN over t4 is false for every c1 in t2, while the
same IN without its
+ // correlated predicate is true for 1 and 9, so the correlation decides
the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // EXISTS rewritten as a left semi join. The correlated predicate is a
disjunction, so it
+ // is pulled up as a whole and carries the nested IN-subquery, which
references c1, out of
+ // the subquery plan.
+ val query1 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query1, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // Same over t5, which shares no value with t2, so the nested IN is
false for every c1 and
+ // only the correlated predicate can hold: a mistake making it true
would return every row.
+ val query2 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query2, Row(1) :: Row(9) :: Nil)
+
+ // NOT EXISTS rewritten as a left anti join.
+ val query3 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE NOT EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t5)
+ |)""".stripMargin
+ checkAnswerAndPlan(query3, Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // IN-subquery rewritten as a left semi join. The hoisted predicate is a
> c1 rather than
+ // the key equality a = c1, so the answer is a IN (t2 INTERSECT t3) and
depends on the
+ // nested subquery: dropping it would leave no row at all.
+ val query4 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query4, Row(9) :: Nil)
+
+ // NOT IN-subquery rewritten as a null-aware left anti join, with a
nested EXISTS. Only the
+ // nested EXISTS keeps 9 out of the answer, as a > c1 alone does not
hold for it.
+ val query5 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR EXISTS (SELECT 1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query5, Row(1) :: Row(2) :: Row(3) :: Row(7) :: Nil)
+
+ // A nested NOT IN-subquery keeps its null-aware semantics: c1 NOT IN
(3, 9, NULL) is never
+ // true, so only the correlated predicate can be satisfied. Compare with
query7, where the
+ // same NOT IN over a relation without NULL holds for c1 = 1 and returns
every row.
+ val query6 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3n)
+ |)""".stripMargin
+ checkAnswerAndPlan(query6, Row(1) :: Row(9) :: Nil)
+
+ val query7 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR c1 NOT IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ checkAnswerAndPlan(query7, Row(1) :: Row(2) :: Row(3) :: Row(7) ::
Row(9) :: Nil)
+
+ // A nested subquery correlated to the query it is nested in, on a
column it does not
+ // project, so its correlated predicate changes the answer: without it
the nested IN would
+ // hold for c1 = 1 and row 1 would be returned as well.
+ val query8 =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT c1
+ | FROM t2
+ | WHERE a > c1
+ | OR c1 IN (SELECT col1 FROM t4 WHERE k = c1)
+ |)""".stripMargin
+ checkAnswerAndPlan(query8, Row(2) :: Row(3) :: Row(7) :: Row(9) :: Nil)
+ }
+ }
+
+ test("SPARK-59351: nested subquery referencing both the outer and the inner
query") {
+ withTempView("t1", "t2", "t3", "t4") {
+ Seq((1), (2), (3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq((1), (8), (9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((3), (9)).toDF("col1").persist().createOrReplaceTempView("t3")
+ // Correlated, this is false for every c1 in t2; without its correlated
predicate it holds
+ // for 1 and 9, so the correlation decides the answer below.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+
+ // The nested subquery references the outer query through its values and
the inner query
+ // through its own correlated predicate, so it can be rewritten into an
existence join on
+ // neither side and stays in the join condition. Planning it there as an
in-subquery filter
+ // drops its correlated predicate col1 = c1, which is why it is rejected
rather than left.
+ // These rows do not distinguish the two forms; the EXISTS form of this
shape does, and
+ // returned an extra row before the rejection was added.
+ //
+ // NOT IN is the outer predicate on purpose: its rewrite returns a bare
Join, which
+ // RewritePredicateSubquery does not offer to its own handling of
predicate subqueries in
+ // join conditions, so this rejection is the only one that can fire and
the assertion below
+ // pins it. The EXISTS form of this query fails on the same condition,
raised by the same
+ // rejection in rewriteExistentialExprInSubqueryPlan before any join
exists; were that
+ // rejection removed, the EXISTS form would instead reach the
join-condition handling, which
+ // rejects it with UNSUPPORTED_CORRELATED_EXPRESSION_IN_JOIN_CONDITION
under the default
+ // configuration, so only the NOT IN form pins this rejection on its own.
+ val correlated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE a NOT IN (
+ | SELECT c1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e = intercept[AnalysisException](sql(correlated).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ // The message must name the subquery that was rejected.
+ val reported = e.getMessageParameters.get("subqueryExpression")
+ assert(reported.contains("IN") && reported.contains("listquery"),
+ s"unexpected subqueryExpression: $reported")
+ }
+ }
+
+ // A correlated predicate that BooleanSimplification has eliminated:
+ // PullupCorrelatedPredicates retains c1 as an outer attribute of the
nested subquery for
+ // idempotency, but the subquery
+ // no longer reads it, so there is no condition to lose. It must still
be rewritten against
+ // the outer plan, which is the only plan it effectively references,
rather than be treated
+ // as referencing the subquery plan and rejected.
+ val simplifiedAwayCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR a IN (SELECT col1 FROM t3 WHERE false AND col1 = c1)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ checkAnswer(sql(simplifiedAwayCorrelation), Row(1) :: Nil)
+ }
+ }
+
+ // A subquery that stays in the join condition stays there whole, since
the rewrite does
+ // not descend into one it declines, so a correlated subquery below it
is left as well.
+ // Its own references make it look routable, being owned by the subquery
plan, but
+ // nothing reaches it and planning a list query as an in-subquery filter
drops the
+ // predicate the pull-up hoisted into it. The enclosing subquery is
rejected for
+ // carrying it: PostgreSQL and DuckDB return 1 and 3 here, and losing
the correlation
+ // returns 2 as well.
+ val carriedCorrelation =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + (CASE WHEN c1 IN (SELECT col1 FROM t4 WHERE k = c1) THEN
1 ELSE 0 END))
+ | IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ val e =
intercept[AnalysisException](sql(carriedCorrelation).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY")
+ }
+ }
+
+ // An uncorrelated nested subquery referencing both queries has no
correlated predicate to
+ // lose, so it stays in the join condition and is planned as an
in-subquery filter. The
+ // configuration is disabled because the rewrite of predicate subqueries
in join conditions
+ // rejects a subquery referencing both join inputs, which is what this
one becomes.
+ val uncorrelated =
+ """
+ |SELECT *
+ |FROM t1
+ |WHERE EXISTS (
+ | SELECT 1
+ | FROM t2
+ | WHERE a = c1
+ | OR (a + c1) IN (SELECT col1 FROM t3)
+ |)""".stripMargin
+ withSQLConf(
+ SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
"false") {
+ // No existence join here: this subquery stays in the join condition,
which is the
+ // point of the case, so only the result and the reachability of the
plan are
+ // checked.
+ checkAnswerAndPlan(uncorrelated, Row(1) :: Row(2) :: Nil,
+ existenceJoinOnSubqueryPlan = false)
+ }
+ }
+ }
+
+ test("SPARK-59351: existential subquery carried by a rewritten nested
subquery") {
+ // The nested IN over t4 is owned by the subquery plan, so it is rewritten
into an existence
+ // join and the condition PullupCorrelatedPredicates hoisted into it is
grafted onto that
+ // join. That condition carries a third-level IN correlated to t4, which
references neither
+ // the outer nor the subquery plan and so is routed by neither of the
guards above. It has to
+ // be rewritten against the inputs of the join just built: left in place,
PlanSubqueries plans
+ // it as an in-subquery filter and drops its own correlated predicate col1
= k, which returns
+ // every row of t1 instead of the one row where the correlation holds.
+ withTempView("t1", "t2", "t4", "t5") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ // No k equals any c1, so only the third-level IN can satisfy the
hoisted disjunction, and
+ // its correlated predicate col1 = k is false for every row of t4:
dropping it makes the
+ // level-2 IN hold for c1 = 1 and returns 1, 2 and 3.
+ Seq((1, 100), (9, 7)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1).toDF("col1").persist().createOrReplaceTempView("t5")
+ val query =
+ """SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2
+ | WHERE a = c1
+ | OR c1 IN (SELECT col1 FROM t4
+ | WHERE k = c1 OR col1 IN (SELECT col1 FROM t5 WHERE col1
= k)))""".stripMargin
+ // The routing must not depend on the rewrite of predicate subqueries in
join conditions,
+ // which is what used to make these two configurations disagree: it
reached the grafted
+ // condition only when enabled.
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ checkAnswerAndPlan(query, Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: a carried subquery is classified as on the subquery-plan
route") {
+ // The carried router applies the same rules as
rewriteExistentialExprInSubqueryPlan, against
+ // the inputs of the existence join rather than the outer and the subquery
plan.
+ withTempView("t1", "t2", "t4", "t5", "t3n", "t5n") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((9, 7)).toDF("col1", "k").persist().createOrReplaceTempView("t4")
+ Seq(Some(3), None).toDF("x").persist().createOrReplaceTempView("t3n")
+ Seq((Some(3), 7)).toDF("x", "k").persist().createOrReplaceTempView("t5n")
+
+ // A carried NOT IN is collected as the Not over the IN, so the rewrite
takes its null-aware
+ // arm. Taking the positive-IN arm instead leaves NOT exists over a
condition that ignores
+ // NULL comparisons: 5 NOT IN (NULL) is unknown, so n4's (5, 100) row is
excluded and the
+ // answer is empty, where the two-valued form returns every row of n1,
that is 1 and 2.
+ withTempView("n1", "n2", "n4", "n5") {
+ Seq(1, 2).toDF("a").persist().createOrReplaceTempView("n1")
+ Seq(5).toDF("c1").persist().createOrReplaceTempView("n2")
+ Seq((5, 100)).toDF("col1", "k").persist().createOrReplaceTempView("n4")
+ Seq((Option.empty[Int], 100)).toDF("x",
"y").persist().createOrReplaceTempView("n5")
+ val carriedNotIn =
+ """SELECT * FROM n1 WHERE EXISTS (
+ | SELECT 1 FROM n2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM n4
+ | WHERE k = c1 OR col1 NOT IN (SELECT x FROM n5 WHERE y =
k)))""".stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(sql(carriedNotIn), Nil)
+ }
+ }
+ }
+
+ // An uncorrelated carried IN is routed too, not left for the
join-condition rewrite, which
+ // has no unknown guard and would make it a two-valued exists under the
default
+ // configuration. Its unknown is observable here, so it is rejected
under both settings.
+ val uncorrelatedUnknown =
+ """SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM t4
+ | WHERE k = c1 OR ((col1 IN (SELECT x FROM t3n)) <=>
false)))""".stripMargin
+ // A correlated one in the same position, which must report the same
condition rather than
+ // the one about referencing both queries.
+ val correlatedUnknown =
+ """SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM t4
+ | WHERE k = c1
+ | OR ((col1 IN (SELECT x FROM t5n WHERE x = k)) <=>
false)))""".stripMargin
+ Seq(uncorrelatedUnknown, correlatedUnknown).foreach { query =>
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ val e = intercept[AnalysisException](sql(query).collect())
+ assert(e.getCondition ==
"UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY." +
+ "NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT", query)
+ }
+ }
+ }
+
+ // The same shape where only truth is observed is still rewritten and
answers correctly.
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(
+ sql("""SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM t4
+ | WHERE k = c1 OR (col1 IN (SELECT x FROM
t3n))))""".stripMargin),
+ Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: carried subqueries are routed on every route that builds
an existence join") {
+ // rewriteCarriedExistentialExprs runs on every existence join this rule
builds, not only on
+ // the ones the semi/anti join arms build for a top-level predicate
subquery. On the routes
+ // below the outer EXISTS is itself rewritten into an existence join by
+ // rewriteExistentialExprWithAttrs, so the nested IN hoisted into its
condition is carried
+ // rather than routed by the first-level classification, and only this
router reaches it.
+ withTempView("t1", "t2", "t4", "t6") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ // The correlated nested IN is false for every c1 of t2, while the same
IN without k = c1
+ // holds for c1 = 1, so losing that predicate turns every answer below
into every row.
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1, 2, 3).toDF("b").persist().createOrReplaceTempView("t6")
+ val exists =
+ "EXISTS (SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (SELECT col1 FROM t4
WHERE k = c1))"
+
+ Seq("true", "false").foreach { decorrelateInJoinCondition =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
+ decorrelateInJoinCondition) {
+ // The Filter-predicate arm: under OR the EXISTS is not a conjunct
of its own, so it
+ // goes through rewriteExistentialExpr rather than through the semi
join arm.
+ checkAnswer(sql(s"SELECT * FROM t1 WHERE a > 100 OR $exists"),
Row(1) :: Nil)
+ // handleUnaryNode: the EXISTS is a projection rather than a
predicate.
+ checkAnswer(
+ sql(s"SELECT a, $exists AS e FROM t1"),
+ Row(1, true) :: Row(2, false) :: Row(3, false) :: Nil)
+ }
+ }
+
+ // The join-condition arm reaches the router only when it runs at all,
which is what
+ // decorrelatePredicateSubqueriesInJoinPredicate gates; with it disabled
the subquery stays
+ // in the join condition, which is pre-existing behaviour this PR does
not change.
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
"true") {
+ checkAnswer(
+ sql(s"SELECT t1.a FROM t1 JOIN t6 ON t1.a = t6.b AND $exists"),
Row(1) :: Nil)
+ }
+ }
+ }
+
+ test("SPARK-59351: a carried IN shared between a bare and a negated
position") {
+ // Filter pushdown substitutes the alias, so the same IN instance is
carried bare and under
+ // a Not. Routing the bare one must not replace the one inside the Not, or
the null-aware
+ // existence join built for the Not is left unreferenced and NOT exists
over a two-valued
+ // match returns the rows where the IN is unknown. Standard SQL keeps only
a = 1.
+ withTempView("s1", "s2", "s3") {
+ Seq(Some(1), Some(-1),
None).toDF("a").persist().createOrReplaceTempView("s1")
+ Seq(1).toDF("c1").persist().createOrReplaceTempView("s2")
+ Seq(Some(3), None).toDF("col1").persist().createOrReplaceTempView("s3")
+ val query =
+ """SELECT s.a FROM (SELECT a, a IN (SELECT col1 FROM s3) AS b FROM s1)
s
+ |WHERE s.a > 0 OR EXISTS (SELECT 1 FROM s2 WHERE s.a = s2.c1 OR s.b
OR NOT s.b)"""
+ .stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ checkAnswer(sql(query), Row(1) :: Nil)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: a carried subquery referencing both inputs of its
existence join") {
+ // The third-level IN takes its compared value c1 from the left input of
the existence join
+ // built for the nested IN over t4, and its hoisted x = k reads the right
one. Neither input
+ // can evaluate it, and left in place it is planned as an in-subquery
filter that drops x = k.
+ withTempView("t1", "t2", "t4", "t5") {
+ Seq(1, 2, 3).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(1, 8, 9).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq((1, 9), (9, 1)).toDF("col1",
"k").persist().createOrReplaceTempView("t4")
+ Seq(1, 9).toDF("x").persist().createOrReplaceTempView("t5")
+ val query =
+ """SELECT * FROM t1 WHERE EXISTS (
+ | SELECT 1 FROM t2 WHERE a = c1 OR c1 IN (
+ | SELECT col1 FROM t4
+ | WHERE k = c1 OR c1 IN (SELECT x FROM t5 WHERE x =
k)))""".stripMargin
+ Seq("true", "false").foreach { conf =>
+
withSQLConf(SQLConf.DECORRELATE_PREDICATE_SUBQUERIES_IN_JOIN_CONDITION.key ->
conf) {
+ val e = intercept[AnalysisException](sql(query).collect())
+ assert(e.getCondition == "UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY."
+
+ "NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY", conf)
+ }
+ }
+ }
+ }
+
+ test("SPARK-59351: carried unknown-result rejection is limited to the
subquery-plan route") {
+ // On the routes that predate this change the carried IN below was
answered, and it is not
+ // rejected: its handling there is left to SPARK-59862. The columns are
nullable but hold no
+ // NULL, so the two-valued and the three-valued answers agree and a
rejection would be the
+ // only difference. The first query takes its compared value from the
right input of the
+ // existence join built for the EXISTS, the second from the left one.
+ withTempView("t1", "t2", "t3") {
+ Seq(Some(1), Some(2),
Some(3)).toDF("a").persist().createOrReplaceTempView("t1")
+ Seq(Some(1), Some(8),
Some(9)).toDF("c1").persist().createOrReplaceTempView("t2")
+ Seq(Some(8)).toDF("col1").persist().createOrReplaceTempView("t3")
Review Comment:
**Nit (P3):** With `t3` holding only 8, the two-valued and three-valued
answers agree, as the test comment says, so this case shows only that nothing
is rejected. Removing the `filterNot` that leaves these INs in place would
leave every assertion passing. Adding a non-matching NULL to `t3` and asserting
under `decorrelatePredicateSubqueriesInJoinPredicate.enabled=false` would pin
it. There the rightInput query returns only 1 when the IN stays in place, and
1, 2 and 3 after a two-valued rewrite. Under the default setting both forms
return 1, 2 and 3 (SPARK-59862), so only the disabled setting can tell them
apart.
**Verification:**
- **Regression:** With a list column holding a non-matching NULL and the
join-condition setting disabled, the right-input and left-input queries on the
Filter-predicate route return the three-valued answer. The case fails if the
uncorrelated unknown-observing carried IN is routed into a two-valued existence
join.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
+ }
+ if (carried.isEmpty) {
+ return join
+ }
+ // Where the value of a carried IN is observed beyond being TRUE, an
existence join cannot
+ // stand in for it, exactly as on the routes above.
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+
+ val rejected = carried.filter { sq =>
+ observesUnknown(sq) ||
+ (referencesInput(sq, join.left) && referencesInput(sq, join.right))
+ }
+ if (rejected.nonEmpty) {
+ // A sub-query that needs both inputs can be rewritten against neither,
and one whose
+ // unknown is observable cannot become an `exists` attribute at all.
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(rejected)
+ }
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ carried.filter(sq => isInput(sq) &&
!rejected.contains(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
rewriteExistentialExprWithAttrs(Seq(sq), p)
Review Comment:
Confirmed on 2666748431d: collectCarriedExistentialExprs collects a NOT IN
as the Not over its IN, so the router takes the null-aware arm. observesUnknown
now compares against the same Not node the unknown collector reports.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4346076647","thread_id":"inline:4346076647","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
Review Comment:
Confirmed on 2666748431d: the router now considers uncorrelated carried INs,
and on the sub-query-plan route this shape is rejected with
NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT under both settings. The later narrowing
of that rejection to the sub-query-plan route leaves the older routes to
SPARK-59862, as discussed.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4346076676","thread_id":"inline:4346076676","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,568 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Rewrites the existential sub-queries carried by the condition of an
existence join that this
+ * rule has just built, against whichever of that join's own inputs can
evaluate them.
+ *
+ * [[rewriteExistentialExprWithAttrs]] grafts the condition that
[[PullupCorrelatedPredicates]]
+ * hoisted into a nested sub-query onto the existence join it builds for
that sub-query, and
+ * replaces the sub-query atomically without descending into what the
condition carries. A
+ * correlated existential sub-query carried along references the plan of the
sub-query it was
+ * hoisted out of, which is an input of the join just built and of no plan
seen before it, so
+ * nothing else reaches it: left there, `PlanSubqueries` plans it as an
in-subquery filter and
+ * drops its own hoisted condition, which returns extra rows.
+ *
+ * This is what the handling of predicate sub-queries in join conditions
does for a join the user
+ * wrote, and doing it here too makes the result independent of
+ *
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate.enabled`,
which gates that
+ * handling. The routing recurses, since the joins built here go through it
in turn.
+ *
+ * An uncorrelated sub-query is deliberately left where it is. It has no
hoisted condition to
+ * lose, and `InSubqueryExec` evaluates it with the three-valued semantics
that the `exists`
+ * attribute of an existence join cannot carry.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): LogicalPlan = {
+ if (!join.condition.exists(_.containsAnyPattern(EXISTS_SUBQUERY,
IN_SUBQUERY))) {
+ return join
+ }
+ val condition = join.condition.get
+ // Only a correlated sub-query has to be consumed here, see above.
+ val carried = condition.collect {
+ case sq @ (_: Exists | _: InSubquery) if hasCorrelatedCondition(sq) => sq
+ }
+ if (carried.isEmpty) {
+ return join
+ }
+ // Where the value of a carried IN is observed beyond being TRUE, an
existence join cannot
+ // stand in for it, exactly as on the routes above.
+ val unknownSensitive = unknownSensitiveInSubqueries(condition, truthOnly =
true)
+ def observesUnknown(sq: Expression): Boolean =
unknownSensitive.exists(_.fastEquals(sq))
+ def referencesInput(sq: Expression, input: LogicalPlan): Boolean =
+ effectiveReferences(sq).exists(input.outputSet.contains)
+
+ val rejected = carried.filter { sq =>
+ observesUnknown(sq) ||
+ (referencesInput(sq, join.left) && referencesInput(sq, join.right))
+ }
+ if (rejected.nonEmpty) {
+ // A sub-query that needs both inputs can be rewritten against neither,
and one whose
+ // unknown is observable cannot become an `exists` attribute at all.
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(rejected)
Review Comment:
Confirmed: the carried router raises
NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY only for a sub-query that
references both inputs, and NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT for the
unknown-result case. The new tests reach both.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4346076695","thread_id":"inline:4346076695","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
--
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]
