pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4356951704
##########
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)
+ }
+ var newCondition = condition
+ def route(plan: LogicalPlan, isInput: Expression => Boolean): LogicalPlan
= {
+ carried.filter(sq => isInput(sq) &&
!rejected.contains(sq)).foldLeft(plan) {
+ case (p, sq) =>
+ val (rewritten, newInput, _) =
rewriteExistentialExprWithAttrs(Seq(sq), p)
Review Comment:
Confirmed and fixed, and your diagnosis of the cause is exact: `_:
InSubquery` matched the IN *inside* the `Not`, so `route` handed the bare IN to
the positive-IN arm and the null-aware condition was never built.
Verified against your fixture:
| | result |
|---|---|
| before, both configurations | `1, 2` |
| **now, both configurations** | **no rows** |
| DuckDB | no rows |
| merge base | `INTERNAL_ERROR` |
The collection is now done by `collectCarriedExistentialExprs`, a small
recursive walk rather than a `collect`, which takes a NOT IN as the `Not` over
the `InSubquery` and does not descend into a collected node. A plain `collect`
would have matched both the `Not` and the inner IN and classified the same
sub-query twice.
That also fixes the second consequence you noted:
`unknownSensitiveInSubqueries` reports the `Not`, and `observesUnknown` now
compares against the same node, so a carried `(col1 NOT IN (...)) <=> false` is
rejected rather than routed. `hasCorrelatedCondition` gained a `Not(in:
InSubquery)` case so the unit is classified by the IN it negates.
--
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]
