cloud-fan commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4414370176
##########
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(
Review Comment:
Confirmed: the new test reaches the carried both-inputs rejection under both
values of decorrelatePredicateSubqueriesInJoinPredicate.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4369134668","thread_id":"inline:4369134668","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
##########
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:
Thanks, routing NOT IN units first fixes this repro, but it is not
equivalent to substituting at the collected position. On a pre-existing route
the router keeps an uncorrelated NOT IN whose unknown is observable, e.g. `(NOT
s.b) = s2.f`, yet still routes the bare `s.b`. The bare unit's substitution
then rewrites the IN inside the kept NOT, so with either join-condition flag
off -1 is returned again where the merge base returned only 1. Substituting
each unit only where it was collected covers this case as well.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4369134655","thread_id":"inline:4369134655","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
##########
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:
Thanks, the IN and EXISTS failure modes are now separated. The same sentence
still says nothing else reaches a carried sub-query, which this review raises
as a separate point.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4369134661","thread_id":"inline:4369134661","verdict_sha256":"596d6ef5f193363401f54af770e62e1e49b3bfd27c84efadb74d430a0c108603"}
-->
##########
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:
Thanks, the production comment now states the last-conjunct condition. Its
explanation of which later conjuncts expose this anti join is incomplete, which
this review raises.
<!-- SPARK_DEV_REVIEW_REPLY
{"feedback_id":"inline:4346076714","thread_id":"inline:4346076714","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]
