pratham76 commented on code in PR #58656:
URL: https://github.com/apache/spark/pull/58656#discussion_r4380513465
##########
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:
Fixed using the shape you described: a third-level `c1 IN (SELECT x FROM t5
WHERE x = k)` inside a nested `c1 IN (SELECT col1 FROM t4 WHERE k = c1 OR
...)`. The compared value comes from the left input of the existence join built
for the nested IN and the hoisted `x = k` reads the right one. The test asserts
`NESTED_SUBQUERY_REFERENCING_OUTER_AND_INNER_QUERY` under both values of
`decorrelatePredicateSubqueriesInJoinPredicate.enabled`. The both-inputs
rejection still applies on every route.
--
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]
