HyukjinKwon commented on code in PR #59301:
URL: https://github.com/apache/spark/pull/59301#discussion_r4236790250
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/DecorrelateInnerQuery.scala:
##########
@@ -937,6 +937,43 @@ object DecorrelateInnerQuery extends PredicateHelper {
decorrelate(d.child, parentOuterReferences, aggregated = true,
underSetOp)
(d.copy(child = newChild), joinCond, outerReferenceMap)
+ case j: AsOfJoin =>
+ // An ASOF JOIN picks one right row for each left row, so a right
row may only match
+ // left rows of the same outer row. Unlike a regular join:
+ // - The left input always carries the outer references, including
those of the
+ // right input, so the join can match both inputs on them.
+ // - The right input is decorrelated as if it were aggregated. Its
correlated
+ // predicates either stay below the join or come back as
matching conditions.
+ // - Correlated ON predicates limit which right rows can match, so
they stay in the
+ // join instead of moving up to the outer query.
+
assert(!j.expressions.filterNot(j.condition.contains).exists(containsOuter),
+ s"Correlated column is only allowed in the ON condition of an
ASOF join: $j")
+ val outerReferences = collectOuterReferences(j.expressions)
+ if
(!SQLConf.get.getConf(SQLConf.DECORRELATE_JOIN_PREDICATE_ENABLED)) {
+ assert(outerReferences.isEmpty, s"Correlated column is not
allowed in join: $j")
+ }
+ val (newLeft, leftJoinCond, leftOuterReferenceMap) = decorrelate(
+ j.left,
+ parentOuterReferences ++ outerReferences ++
collectOuterReferencesInPlanTree(j.right),
+ aggregated,
+ underSetOp)
+ val (newRight, rightJoinCond, rightOuterReferenceMap) =
+ if (hasOuterReferences(j.right)) {
+ decorrelate(j.right, AttributeSet.empty, aggregated = true,
underSetOp)
Review Comment:
**Blocking (P1):** `decorrelate(j.right, AttributeSet.empty, aggregated =
true, underSetOp)` removes every correlated predicate that passes
`canPullUpOverAgg` from the right input. Only a bare `attr = outer(attr)`
equality also becomes an entry in `rightOuterReferenceMap`, so only that kind
gets a matching condition in ON. Other pullable equalities come back only in
`rightJoinCond` and are applied above the ASOF join, after it has already
picked the nearest row from the unfiltered right input. Examples are
`q.bid_price = CAST(outer(o.p) AS DECIMAL(5,2))` from implicit coercion,
`CAST(q.i AS BIGINT) = outer(o.l)`, and `q.symbol = upper(outer(o.s))`.
With the `trades`/`quotes` views from `join-asof-crossfeature.sql`:
```sql
SELECT (SELECT count(*)
FROM trades t
ASOF JOIN (SELECT * FROM quotes q WHERE q.bid_price = o.p) q
MATCH_CONDITION (t.trade_time >= q.quote_time)
ON t.symbol = q.symbol)
FROM VALUES (180.1) AS o(p)
```
This should return 2, because both AAPL trades match the 10:00:00 quote, the
only one at 180.10. The decorrelated plan instead matches the 10:00:11 trade to
the 10:00:10 quote (180.20), and the pulled-up `bid_price = CAST(p)` then drops
that row, so the query returns 1. Likewise, `WHERE q.symbol = upper(o.sym)`
with `'msft'` returns NULL instead of 420.50. Before this PR these queries
failed analysis; now they silently return wrong results. This also contradicts
the comment above ("either stay below the join or come back as matching
conditions"). The new right-input tests use only `<=`, `IS NULL OR <=`, or a
bare `b3 = outer(a)`, so none of them catches it. I traced this by hand and did
not run it.
**Recommended change:** In the AsOfJoin case, also enforce the right input's
returned join conditions inside the ASOF join's ON, rewritten to the left
input's outer-reference mapping. Keep returning them upward so domain-join
rewriting and grouping are unchanged. Update the case's comment to match. Add
Catalyst decorrelation plan tests and ASOF JOIN cross-feature SQL golden tests
with right-input equality correlations that are bare, coerced by type widening,
and wrapped in a function.
**Why this works:** After `decorrelate(j.right, ...)`, map each right-side
join condition through leftOuterReferenceMap (the left carries each such outer
reference as a domain or equivalent attribute) and conjoin it into newCondition
with the existing ON and matching conditions. A pullable equality like
`bid_price = CAST(outer(p))` then becomes `bid_price = CAST(p')` inside the
join, so it restricts the candidate right rows before the nearest one is
chosen. The canPullUpOverAgg one-to-one guarantee keeps the redundant upper
copy and the grouping it adds consistent.
**Scope:** Make ASOF JOIN decorrelation enforce right-input correlated
predicates inside the join, and cover right-input equality correlations end to
end.
**Compatibility:** Left-input correlations stay pulled above the join
without a domain join, and correlated ON predicates stay in the join.
**Risks:** Plan-shape expectations for the existing right-input test change,
because the bare equality also appears as a redundant ON conjunct. Any outer
reference left unmapped in the added conjuncts would remain as OuterReference
in ON; the left-domain coverage of right-input outer references must hold.
**Constraints:** Keep returning the right input's join conditions to the
enclosing query, so rewriteDomainJoins can still rewrite DomainJoins inside the
right input. Do not change the shared Filter, Aggregate, or canPullUpOverAgg
decorrelation logic used by other operators.
**Success:** A correlated equality in the right input of an INNER ASOF JOIN,
bare or wrapped by an implicit cast or a function, returns the same result as
running the subquery once per outer value. For example, `WHERE q.bid_price =
o.p` with o.p DECIMAL(4,1) = 180.1 counts 2, and `WHERE q.symbol =
upper(o.sym)` with 'msft' returns 420.50. Left-input, ON, and non-equality
right-input correlations return the same results as before the fix. The
AsOfJoin case's comment describes where right-input correlated predicates are
enforced.
**Verification:**
- **Behavior:** End-to-end ASOF JOIN correlated-subquery queries with
right-input equality correlations (bare, coerced by type widening, and wrapped
in a function) return the per-outer-value results.
- **Regression:** A decorrelation plan check shows that a non-bare
right-input correlated equality is enforced inside the decorrelated ASOF join
condition, rewritten to the left input's outer-reference attribute.
- **Compatibility:** The existing ASOF JOIN correlated-subquery golden
results (left, right, ON, both inputs, IN, EXISTS, LEFT ASOF, NULL outer value,
aggregate and LIMIT in the right input) stay 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]
