cloud-fan commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4047869684
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala:
##########
@@ -450,10 +471,100 @@ 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 cannot be rewritten into an
existence join on
+ * either side, so it is left in the join condition, where both plans are in
scope. This is only
+ * correct for an uncorrelated sub-query: 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 it 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.
+ *
+ * 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 correlated sub-query referencing both plans has to stay in the join
condition, where its
+ // own join condition is lost, so reject it rather than silently return a
wrong result.
+ val unsupported = conditions.flatMap(_.collect {
+ case sq @ (_: Exists | _: InSubquery)
+ if isCorrelatedSubquery(sq) &&
+ referencesPlan(sq, subPlan) && referencesPlan(sq, outerPlan) => sq
+ })
+ if (unsupported.nonEmpty) {
+ throw
QueryCompilationErrors.unsupportedCorrelatedSubqueryInJoinConditionError(unsupported)
+ }
+ val (newCond, newSubPlan, _) = rewriteExistentialExprWithAttrs(conditions,
subPlan,
Review Comment:
**Blocking (P1):** This call can receive a positive `InSubquery`, whose SQL
result is nullable, but the rewrite replaces it with the non-nullable `exists`
attribute. That distinction remains observable inside the parent expression:
for `((t2.c1 IN (SELECT col1 FROM t3)) <=> false)` with `t3` containing a
non-matching `NULL`, SQL produces `NULL <=> false`, which is false, while this
path produces `false <=> false` and admits the `t2` row. The new inner-plan
route therefore turns the old attribute-binding failure into silently wrong
rows. Please preserve the three-valued result of positive `IN` when
synthesizing this existence join.
**Recommended change:** Represent positive nested IN with enough synthesized
state to reconstruct TRUE, FALSE, and NULL, and add SQL subquery regressions
that place an inner-referencing positive IN under a null-sensitive parent.
**Why this works:** At the positive-IN rewrite owner, preserve both whether
any equality condition is true and whether the candidate comparisons can yield
unknown when no true match exists, then return a nullable Boolean expression
that reconstructs SQL IN semantics. Apply that owner-level representation to
both outer- and subquery-plan routes instead of treating an existence bit as
the IN value.
**Scope:** Correct the shared nested positive-IN representation and verify
three-valued results through the changed inner-side route without altering the
accepted both-side rejection policy.
**Compatibility:** The routing decision remains attribute-owner based:
inner-only nested subqueries are rewritten on the subquery plan,
outer-only/reference-free ones on the outer plan, and correlated both-side ones
are rejected.
**Risks:** Multi-column IN equality and correlated join predicates require
the unknown signal to follow SQL row-comparison semantics, not merely detect
any NULL value on the right side.
**Constraints:** Preserve existing true-match behavior, null-aware NOT IN
behavior, attribute ownership on both routing sides, and the structured
rejection for correlated both-side subqueries.
**Success:** A positive nested IN returns true for a match, false for no
match with no unknown comparison, and null for no match with an unknown
comparison, regardless of whether it is routed to the outer or subquery plan.
Null-sensitive parents observe the same positive-IN value before and after the
existential rewrite and no extra outer rows are admitted. Existing NOT IN
null-awareness and correlated both-side rejection remain unchanged.
--
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]
