pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4356967243
##########
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:
Fixed. `rejected` is split into two sets, each raising the condition whose
message describes it:
```scala
val referencingBothInputs = carried.filter { sq =>
referencesBothInputs(sq) &&
(hasCorrelatedCondition(sq) || carriesCorrelatedSubquery(negatedIn(sq)))
}
if (referencingBothInputs.nonEmpty) { throw
...nestedSubqueryReferencingOuterAndInnerQueryError... }
val unknownResult = carried.filter(sq => referencesOneInput(sq) &&
observesUnknown(sq))
if (unknownResult.nonEmpty) { throw
...nestedInSubqueryWithUnknownResultError... }
```
Your example now reports `NESTED_IN_SUBQUERY_WITH_UNKNOWN_RESULT` under both
configurations, and the new test asserts the condition for a carried IN
rejected for that reason, both correlated and uncorrelated -- so the throw is
reached by a test now, which it was not before.
--
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]
