jayzhan211 opened a new issue, #26710:
URL: https://github.com/apache/datafusion/issues/26710
### Describe the bug
A correlated scalar subquery that picks one row with `ORDER BY ... LIMIT 1`
fails at physical planning with `This feature is not implemented: Physical plan
does not support logical expression ScalarSubquery(<subquery>)`. The common
case is "the latest row at or before a time", such as the exchange rate on the
day of an order.
### To Reproduce
```sql
CREATE TABLE rates (base VARCHAR, quote VARCHAR, as_of DATE, rate
DECIMAL(12, 6));
INSERT INTO rates VALUES
('EUR', 'USD', DATE '2025-01-01', 1.10),
('EUR', 'USD', DATE '2025-01-10', 1.20),
('EUR', 'GBP', DATE '2025-01-05', 0.85),
('USD', 'GBP', DATE '2025-01-01', 0.70);
CREATE TABLE orders (id INT, currency VARCHAR, closed_at TIMESTAMP);
INSERT INTO orders VALUES
(1, 'USD', TIMESTAMP '2025-01-05 10:00:00'),
(2, 'USD', TIMESTAMP '2025-01-10 09:00:00'),
(3, 'GBP', TIMESTAMP '2025-01-02 00:00:00'),
(4, 'JPY', TIMESTAMP '2025-01-05 00:00:00'),
(5, NULL, TIMESTAMP '2025-01-05 00:00:00'),
(6, 'USD', NULL);
CREATE TABLE invoices (id INT, order_id INT, issued_at TIMESTAMP, total INT);
INSERT INTO invoices VALUES
(1, 1, TIMESTAMP '2025-01-05 11:00:00', 10),
(2, 1, TIMESTAMP '2025-01-06 11:00:00', 20),
(3, 2, TIMESTAMP '2025-01-11 11:00:00', 30),
(4, 2, TIMESTAMP '2025-01-11 11:00:00', 40);
-- Query 1: the latest rate on or before the order's date
SELECT o.id,
(SELECT r.rate FROM rates r
WHERE r.base = 'EUR' AND r.quote = o.currency AND r.as_of <=
CAST(o.closed_at AS DATE)
ORDER BY r.as_of DESC LIMIT 1) AS rate
FROM orders o ORDER BY o.id;
-- Query 2: the latest invoice of each order
SELECT o.id,
(SELECT i.total FROM invoices i WHERE i.order_id = o.id
ORDER BY i.issued_at DESC, i.id DESC LIMIT 1) AS last_total
FROM orders o ORDER BY o.id;
```
### Expected behavior
| Query | DataFusion returns | Expected |
|---|---|---|
| 1 | `Physical plan does not support logical expression
ScalarSubquery(<subquery>)` | `1.10` for order 1, `1.20` for order 2, `NULL`
for orders 3-6 |
| 2 | `Physical plan does not support logical expression
ScalarSubquery(<subquery>)` | `20` for order 1, `40` for order 2 (the tie is
broken by `i.id DESC`), `NULL` for orders 3-6 |
DuckDB returns the expected results.
### Additional context
- The cause: `PullUpCorrelatedExpr` stops at a `Sort` or a `Limit` above a
correlated filter (`decorrelate.rs:266-283`; it only removes a `Limit` for
`EXISTS`). So `ScalarSubqueryToJoin` leaves the subquery in place
(`scalar_subquery_to_join.rs:353`), and the physical planner cannot run it.
- This shows up in the dbolympics benchmark (#26661): 14 of its 409 queries
need it. 11 fail on it today (q077 q174 q176 q181 q183 q194 q365 q368 q383 q386
q388), and q156, q178 and q182 fail on it once #26704 is fixed.
- The subqueries come in two shapes, and `main` can already run each one
when the query is written differently:
- **A range on the sort key** (9 queries: q174 q176 q178 q181 q182 q183
q383 q386 q388, all like query 1). The filter has equalities plus `inner.key <=
outer.value`, and the subquery orders by `inner.key DESC`. That is what the
ASOF join computes: one row per outer row, the greatest matching key, `NULL`
when nothing matches. Query 1 written this way returns the expected rows, and
q174 written this way runs:
```sql
SELECT o.id, r.rate
FROM orders o
ASOF JOIN (SELECT * FROM rates WHERE base = 'EUR') r
MATCH_CONDITION (CAST(o.closed_at AS DATE) >= r.as_of)
ON o.currency = r.quote
ORDER BY o.id;
```
- **Equalities only** (4 queries: q156 q194 q365 q368, like query 2). This
is the first row per correlation key. Query 2 written this way returns the
expected rows:
```sql
SELECT o.id, l.total
FROM orders o
LEFT JOIN (
SELECT order_id, total,
row_number() OVER (PARTITION BY order_id ORDER BY issued_at
DESC, id DESC) AS rn
FROM invoices
) l ON l.order_id = o.id AND l.rn = 1
ORDER BY o.id;
```
- q077 has both shapes, and its range lookup also orders by a second
column to break ties (`ORDER BY ws.started_at DESC, ws.id DESC`), which an ASOF
match on one column does not express.
- A fix direction: let `ScalarSubqueryToJoin` turn a `Limit 1` over a `Sort`
over a correlated filter into one of the two plans above: an ASOF join when the
filter is equalities plus one range predicate on the only sort key, in the
direction that the sort picks; otherwise a left join with the first row per key
when the filter is equalities only. `NULLS FIRST` or `NULLS LAST` on the range
key does not matter, because the range predicate drops `NULL` keys.
- Related: #16137 (a correlated scalar subquery without an aggregate or a
`LIMIT`), #14554 and #5492 (range correlation in general).
### Willingness to contribute
I'm not planning to fix this myself, so anyone is welcome to pick it up.
Feel free to ping me with questions about the cause or when you'd like a review.
--
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]
