pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4114432613
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,171 @@ object RewritePredicateSubquery extends
Rule[LogicalPlan] with PredicateHelper {
ExistenceJoin(exists), newConditions, joinHint)
introducedAttrs += exists
exists
+ // A sub-query that `canRewrite` declined is left as it is, children
included.
+ case sq @ (_: Exists | Not(_: InSubquery) | _: InSubquery) => sq
+ case other => other.mapChildren(rewrite)
}
}
+ val newExprs = exprs.map(rewrite)
(newExprs.reduceOption(And), newPlan, introducedAttrs.toSeq)
}
+
+ /**
+ * Returns true if `e` references any of the attributes produced by `plan`.
+ */
+ private def referencesPlan(e: Expression, plan: LogicalPlan): Boolean = {
+ e.references.intersect(plan.outputSet).nonEmpty
+ }
+
+ /**
+ * Rewrites the existential sub-queries that are nested in the correlated
predicates which
+ * [[PullupCorrelatedPredicates]] hoisted out of a predicate sub-query, and
which therefore end
+ * up in the condition of the semi/anti join that replaces that sub-query.
+ *
+ * A hoisted predicate can carry a nested existential sub-query out of the
sub-query plan,
+ * because a correlated predicate is hoisted as a whole when it is a
disjunction. For example
+ *
+ * SELECT * FROM t1 WHERE EXISTS (
+ * SELECT 1 FROM t2 WHERE t1.a = t2.c1 OR t2.c1 IN (SELECT col1 FROM t3))
+ *
+ * hoists `a = c1 OR c1 IN (SELECT col1 FROM t3)` into the join condition.
Such a nested
+ * sub-query must be rewritten against the plan that produces the attributes
it references:
+ * the one above references `c1`, which is produced by the sub-query plan
and not by the outer
+ * plan, so its existence join has to be built on top of the sub-query plan.
Building it on top
+ * of the outer plan instead yields a join whose condition references an
attribute that neither
+ * of its children can produce (SPARK-59351).
+ *
+ * A nested sub-query that references both plans can be rewritten into an
existence join on
+ * neither side, so it is left in the join condition, where both plans are
in scope. Leaving it
+ * there is only correct while it is uncorrelated: the join condition of a
correlated one is
+ * dropped when it is planned as an in-subquery filter, which would silently
change the result,
+ * so a correlated one is reported as unsupported instead. Note that such a
sub-query can be
+ * correlated only to the sub-query plan, as being correlated to the outer
plan as well would
+ * require two levels of correlation, which the Analyzer rejects.
+ *
+ * A sub-query left here is further subject to the rewrite of predicate
sub-queries in join
+ * conditions, which rejects one referencing both plans under the default
configuration; it
+ * survives only with
`spark.sql.optimizer.decorrelatePredicateSubqueriesInJoinPredicate`
+ * disabled.
+ *
+ * An existence join yields only whether a row matched, so its `exists`
attribute cannot tell
+ * FALSE from unknown, while `IN` is three-valued. The two are
indistinguishable while the value
+ * only feeds a predicate, which is what a hoisted condition normally does,
but not when it
+ * reaches something else, e.g. `(c1 IN (SELECT col1 FROM t3)) <=> false`,
which is FALSE for a
+ * NULL that matches nothing and TRUE for the `exists` attribute. An IN
sub-query whose row
+ * comparison can evaluate to unknown is therefore rejected in that position
rather than
+ * rewritten. NOT IN is rewritten with a null-aware join condition of its
own, which is equally
+ * two-valued, so it is rejected there too.
+ *
+ * Returns the rewritten condition along with the updated outer and
sub-query plans.
+ */
+ private def rewriteExistentialExprInJoinCondition(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan, LogicalPlan) = {
+ val (subCond, newSubPlan) =
+ rewriteExistentialExprInSubqueryPlan(conditions, outerPlan, subPlan)
+ // The sub-queries that do not reference the sub-query plan are rewritten
against the outer
+ // plan, as they only reference attributes of the outer plan, if any.
+ val (newCond, newOuterPlan, _) = rewriteExistentialExprWithAttrs(
+ subCond.toSeq, outerPlan, e => !referencesPlan(e, subPlan))
+ (newCond, newOuterPlan, newSubPlan)
+ }
+
+ /**
+ * Rewrites the existential sub-queries in `conditions` that can only be
evaluated by the
+ * sub-query plan, that is those referencing the sub-query plan but not the
outer plan, into
+ * existence joins on top of the sub-query plan. See
+ * [[rewriteExistentialExprInJoinCondition]] for details.
+ *
+ * Returns the rewritten condition along with the updated sub-query plan.
+ */
+ private def rewriteExistentialExprInSubqueryPlan(
+ conditions: Seq[Expression],
+ outerPlan: LogicalPlan,
+ subPlan: LogicalPlan): (Option[Expression], LogicalPlan) = {
+ // A sub-query left in the join condition loses its own join condition
when it is planned
+ // there, so a correlated one would silently return a wrong result: reject
it instead. Note
+ // that this walks the whole expression, including the join condition of a
sub-query that the
+ // rewrite below declines to descend into. That is deliberate: a
correlated sub-query hidden
+ // under a declined one would equally be planned without its join
condition, or reach
+ // execution unevaluable, so it must be rejected even though nothing would
have rewritten it.
+ val referencingBothPlans = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) && referencesPlan(sq, subPlan) &&
+ referencesPlan(sq, outerPlan) => sq
+ })
+ if (referencingBothPlans.nonEmpty) {
+ throw
QueryCompilationErrors.nestedSubqueryReferencingOuterAndInnerQueryError(
+ referencingBothPlans)
+ }
+ // An IN sub-query whose result can be unknown cannot be represented by
the `exists` attribute
+ // of an existence join once that result is observable, see above.
+ val unknownResult = conditions
+ .flatMap(unknownSensitiveInSubqueries(_, inPredicate = true))
+ .filter(sq => referencesPlan(sq, subPlan) && !referencesPlan(sq,
outerPlan))
+ if (unknownResult.nonEmpty) {
+ throw
QueryCompilationErrors.nestedInSubqueryWithUnknownResultError(unknownResult)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
+ e => referencesPlan(e, subPlan) && !referencesPlan(e, outerPlan))
+ (newCond, newSubPlan)
+ }
+
+ /**
+ * Collects the IN sub-queries in `expr` whose result can be unknown and is
not consumed by a
+ * predicate, so that rewriting them into an existence join, whose `exists`
attribute is FALSE
+ * where IN is unknown, would be observable. See
[[rewriteExistentialExprInJoinCondition]].
+ *
+ * `inPredicate` states whether the value of `expr` is only ever tested for
being TRUE, which
+ * holds for the operands of AND, OR and NOT within a condition that ends up
in a Filter or a
+ * join condition. Unknown and FALSE cannot be told apart there.
+ */
+ private def unknownSensitiveInSubqueries(
+ expr: Expression,
+ inPredicate: Boolean): Seq[Expression] = expr match {
+ case And(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ case Or(left, right) =>
+ unknownSensitiveInSubqueries(left, inPredicate) ++
+ unknownSensitiveInSubqueries(right, inPredicate)
+ // NOT IN is rewritten with a null-aware join condition, which is
two-valued as well.
+ case Not(in: InSubquery) =>
Review Comment:
All three points were right, and all three are fixed.
**The error named the wrong operator.** The rejection now reports the `Not`
node rather than the
`InSubquery` underneath it, so `Not.sql` renders the user's NOT IN. While
fixing it the message
template read wrongly for the negated form, so it no longer begins "The IN
subquery": it is now
"The subquery `<subqueryExpression>` is nested in a correlated condition of
another subquery, and
its result can be unknown where it is not only tested for being true." The
tests now assert
`subqueryExpression` as well as the condition, including that it starts with
`(NOT ` exactly for the
NOT IN predicate, which is the assertion that was missing.
**The arm is no longer redundant.** It is what supplies the `Not` node to
the diagnostic, so
deleting it now changes behaviour, and it no longer duplicates a condition
that could drift: both
arms call one helper that either reports the node or descends into the
sub-query's values and list
query.
**The scaladoc contradiction is gone.** "NOT IN is excluded" was wrong — NOT
IN is rejected in a
value position, for the same reason as IN, since its null-aware rewrite is
two-valued as well. The
scaladoc on `inSubqueryMayBeUnknown` now says only what the function does,
and the contract about
NOT lives where the traversal decides it, with the explanation that a NOT
directly over an IN is
safe because its null-aware join condition is TRUE where the comparison is
unknown, so the negated
`exists` is FALSE there.
Your comment also prompted two related fixes in the same traversal, from
your other notes: it now
descends into an IN sub-query it does not collect, so a nested one in the
values or in the hoisted
condition is no longer skipped, and `<=> true`, `IF` and `CASE WHEN`
conditions are recognised as
truth-only rather than rejected.
--
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]
