pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4286964125
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -398,62 +406,481 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
(newExpr, newPlan)
}
+ /**
+ * Same as [[rewriteExistentialExpr]], but it also returns the newly
introduced attributes, and
+ * it only rewrites the existential sub-queries for which `canRewrite`
returns true. A sub-query
+ * that is not rewritten stays in the returned expression as it is, and is
not descended into:
+ * rewriting an existential sub-query nested in its join condition would
graft an existence join
+ * onto the plan for an `exists` reference that the sub-query left in place
may never evaluate.
+ */
private def rewriteExistentialExprWithAttrs(
exprs: Seq[Expression],
- plan: LogicalPlan): (Option[Expression], LogicalPlan, Seq[Attribute]) = {
+ plan: LogicalPlan,
+ canRewrite: Expression => Boolean = _ => true): (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 =
+ buildJoin(newPlan, rewriteDomainJoinsIfPresent(newPlan, sub,
newCondition),
+ existenceJoin, newCondition, subHint)
+ 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 = Join(newPlan,
+ rewriteDomainJoinsIfPresent(newPlan, newSub,
Some(finalJoinCond)),
+ ExistenceJoin(exists), Some(finalJoinCond), joinHint)
+ 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 = Join(newPlan, rewriteDomainJoinsIfPresent(newPlan,
newSub, newConditions),
+ ExistenceJoin(exists), newConditions, joinHint)
+ introducedAttrs += exists
+ exists
+ }
+ // A sub-query that `canRewrite` declined is left as it is, children
included.
+ case sq @ (_: Exists | Not(_: InSubquery) | _: InSubquery) => sq
+ case other => other.mapChildren(rewrite)
}
}
+ val newExprs = exprs.map(rewrite)
(newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
}
+
+ /**
+ * Returns true if `e` references any of the attributes produced by `plan`.
+ */
+ private def referencesPlan(e: Expression, plan: LogicalPlan): Boolean = {
+ // 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; attributing it to `outerPlan` instead would let a single
shared attribute
Review Comment:
Agreed, and I have picked the swap reading, since that is the mutation the
probe order actually admits.
Under a swap the probes stay exclusive, so a shared attribute becomes
outer-only: `subPlanOnly` and `bothPlans` are both false and the nested
subquery is **declined and left in the join condition** rather than routed to
the subquery plan. It is not "rejected as referencing both" -- that claim
needed the "counts for the outer plan as well" reading, which the code does not
implement. Both paragraphs and the fixture comment now say the same thing, and
the fixture-outcome sentence is kept, since it is true under the swap reading:
either order ends in the same error, because `dedupSubqueryOnSelfJoin`'s
conflict check reports the duplicated attribute the hoisted condition mentions.
The "neither plan" paragraph also needed updating for a different reason: it
said such a subquery "can be left in the join condition unharmed", which is the
assumption `r4257086397` disproved. It now points at
`rewriteCarriedExistentialExprs`.
--
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]
