[PR] [SPARK-60001][SQL] Allow outer references bellow ASOF JOIN in correlated subqueries [spark]

["LukaZdravic (via GitHub)" <gi...@apache.org>]

LukaZdravic opened a new pull request, #59301:
URL: https://github.com/apache/spark/pull/59301

   ### What changes were proposed in this pull request?
   Subqueries can now use outer columns inside an ASOF JOIN, like a regular JOIN: in either input and in ON. LEFT ASOF JOIN still rejects them in the right input, like LEFT JOIN.
   
   The left input always carries the outer values, and ON matches them with the right input, so each left row picks a right row of its own outer row. An outer column in MATCH_CONDITION fails with `ASOF_JOIN_MATCH_CONDITION_TABLE_REFERENCE`.
   
   Follow-up: the planner checks this match row by row instead of using it as a join key, which can be slow. A separate PR can make it a key, as `ExtractEquiJoinKeys` does for regular joins. This means that it would not check the equality of tags for each row but checking only by groups and 100% skipping the checks for the tags that are not the same. On a big number of different tags this can prove quite benefitial e.g. today they are all in 1 group so 100k rows get scanned by each tag even though they do not need to, with this fix that 100k would get broken up into lets say 1000 groups ( 1000 different tags ) of 100 different rows, so different tags get at most 100 checks. 
   
   ### Why are the changes needed?
   A regular JOIN supports these queries, but ASOF JOIN rejected them:
   
   ```sql
   SELECT s.symbol,
     (SELECT max(q.bid_price)
      FROM (SELECT * FROM trades t WHERE t.symbol = s.symbol) t
      ASOF JOIN quotes q MATCH_CONDITION (t.trade_time >= q.quote_time) ON t.symbol = q.symbol)
   FROM VALUES ('AAPL'), ('MSFT') AS s(symbol);
   ```
   
   Before: `ACCESSING_OUTER_QUERY_COLUMN_IS_NOT_ALLOWED`. After: `AAPL 180.20`, `MSFT 420.50`.
   
   ### Does this PR introduce _any_ user-facing change?
   Yes. These queries now run. An outer column in MATCH_CONDITION now fails during analysis instead of in the optimizer.
   
   ### How was this patch tested?
   New queries in `join-asof-crossfeature.sql` (FVT-ASOF-9-011 to 022) cover scalar, IN, and EXISTS subqueries, LEFT ASOF JOIN, a NULL outer value, and 2 rejected cases. Each result matches the subquery run once per outer value. New tests in `DecorrelateInnerQuerySuite` and `AsOfJoinSQLSuite` check the plans and the error. Without the fix, the new tests fail.
   
   ### Was this patch authored or co-authored using generative AI tooling?
   Generated-by: Claude Code
   
   
   


-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

Re: [PR] [SPARK-60001][SQL] Allow outer references for ASOF JOIN in correlated subqueries [spark]

["HyukjinKwon (via GitHub)" <gi...@apache.org>]

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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org