pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4427702482
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -184,8 +188,20 @@ 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
Review Comment:
Addressed together with r4414363550 in eaeb8f53e71.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,615 @@ 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 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.
+ */
+ private def rewriteCarriedExistentialExprs(join: Join): 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)
+ }
+ val unknownResult = carried.filter(sq => referencesOneInput(sq) &&
observesUnknown(sq))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult.toSeq)
+ }
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ carried.filter(sq => referencesOneInput(sq) &&
isInput(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
rewriteExistentialExprWithAttrs(Seq(sq), p)
+ newCondition = newCondition.transform {
Review Comment:
You're right: routing NOT IN first was only equivalent while every collected
unit was routed, and the keep policy for the pre-existing routes, added in the
same commit, broke that. Fixed in eaeb8f53e71 as you suggested.
`rewriteCarriedExistentialExprs` now substitutes by position: it walks the
condition with the same matching as the collector, rewrites a routed unit where
it was collected, and returns a kept unit whole. A routed bare IN can no longer
reach the IN inside a kept `Not`. The ordering and its comments are removed.
To be upfront about the test: with the in-place policy from r4414363469 now
in, a kept `Not(IN)` and a routed bare IN of the same instance no longer occur
together in any configuration where the answer can tell them apart. With either
flag off both are kept. Under the default the join-condition arm rewrites the
kept one positively afterwards (SPARK-59862). So no SQL case fails on the
substitution alone. I added your `(NOT s.b) = s2.f` repro to the shared-IN test
with each flag off, where it returns only `a = 1`. That pins the combined
behaviour, and the positional substitution keeps it from depending on the keep
policy.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +414,615 @@ 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
Review Comment:
Addressed together with r4414363509 in eaeb8f53e71.
##########
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:
Agreed, and implemented as you recommended in eaeb8f53e71.
* **NOT IN arm:** the Filter fold now tracks each conjunct's position. When
nothing reaches the anti join later, the arm keeps every uncorrelated IN or NOT
IN that the sub-query plan owns as a residual, truth-only ones included, as at
the merge base. That is when the NOT IN is the last sub-query conjunct, or when
either `decorrelatePredicateSubqueriesInJoinPredicate.enabled` or
`optimizeUncorrelatedInSubqueriesInJoinCondition.enabled` is off. Otherwise
today's handling stays: a truth-only nested NOT IN is rewritten null-aware, so
the `case j: Join` arm doesn't rewrite its IN positively.
`keepsThreeValuedResiduals` is replaced by a three-state `ThreeValuedResiduals`.
* **Carried router:** on the routes that don't reject, an uncorrelated
carried IN or NOT IN is also left in place when either flag is off.
* Both read the flags through one helper,
`joinConditionRewriteReachesUncorrelatedIn`, next to a comment in the `case j:
Join` arm saying the two must stay aligned. Correlated units, units carrying a
correlated sub-query, EXISTS, and the sub-query-plan route are unchanged.
Tests:
* The NOT IN residual test asserts that both the truth-only nested IN and
your nested NOT IN case are kept in place: no existence join in the executed
plan, and an `InSubqueryExec`.
* A new case, `... AND EXISTS (SELECT 1 FROM t2 WHERE c1 >= a)` with a NULL
in t3n, returns the standard answer under all three flag settings.
* An uncorrelated carried NOT IN stays in its existence-join condition with
each flag off.
I checked that the tests discriminate: forcing the old residual policy,
dropping the carried keep with flags off, or keeping all residuals when a later
conjunct follows each fails a case.
--
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]
