Returns the wrong grouped row when an outer `GROUP BY` consumes a subquery containing `ORDER BY ... OFFSET`.
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
The inner query is correct when executed by itself. In the outer query,
`EXPLAIN VERBOSE` shows that the physical optimizer removes the inner
`SortExec` but keeps `GlobalLimitExec: skip=1`. The result is therefore
consistent with applying `OFFSET` to physical input order instead of to the
ordered rows.
### To Reproduce
Environment:
```text
datafusion: 54.0.0
pyarrow: 24.0.0
Python: 3.12.3
Platform: Linux 6.8.0-45-generic x86_64
```
```python
from __future__ import annotations
import datafusion
import pyarrow as pa
from datafusion import SessionContext
rows = [
{"grp": "physical_second", "sort_key": 2},
{"grp": "sorted_first", "sort_key": 1},
]
schema = pa.schema(
[
pa.field("grp", pa.string(), nullable=False),
pa.field("sort_key", pa.int64(), nullable=False),
]
)
ctx = SessionContext()
ctx.register_record_batches(
"t0",
[[pa.RecordBatch.from_pylist(rows, schema=schema)]],
)
inner_query = """
SELECT grp, sort_key
FROM t0
ORDER BY sort_key ASC NULLS LAST
OFFSET 1
"""
grouped_query = f"""
SELECT grp, COUNT(*) AS n
FROM ({inner_query}) q
GROUP BY grp
ORDER BY grp
"""
inner_result = ctx.sql(inner_query).to_pydict()
grouped_result = ctx.sql(grouped_query).to_pydict()
print(f"datafusion={datafusion.__version__}")
print(f"pyarrow={pa.__version__}")
print("inner result:", inner_result)
print("grouped result:", grouped_result)
explain = ctx.sql("EXPLAIN VERBOSE " + grouped_query).to_pydict()
for plan_type, plan in zip(
explain["plan_type"],
explain["plan"],
strict=True,
):
if plan_type == "physical_plan":
print("physical plan:")
print(plan)
assert inner_result == {
"grp": ["physical_second"],
"sort_key": [2],
}
assert grouped_result == {
"grp": ["physical_second"],
"n": [1],
}
```
Observed output:
```text
datafusion=54.0.0
pyarrow=24.0.0
inner result: {'grp': ['physical_second'], 'sort_key': [2]}
grouped result: {'grp': ['sorted_first'], 'n': [1]}
physical plan:
SortExec: expr=[grp@0 ASC NULLS LAST], preserve_partitioning=[false]
ProjectionExec: expr=[grp@0 as grp, count(Int64(1))@1 as n]
AggregateExec: mode=SinglePartitioned, gby=[grp@0 as grp], aggr=[count(Int64(1))]
GlobalLimitExec: skip=1, fetch=None
DataSourceExec: partitions=1, partition_sizes=[1]
```
The logical plan still contains `Sort -> Limit -> Aggregate`, but the physical
plan has removed the sort below the offset.
### Expected behavior
### Expected behavior
Sorting by `sort_key ASC` produces:
```text
sorted_first, 1
physical_second, 2
```
`OFFSET 1` must therefore retain `physical_second`. The outer grouping should
return:
```text
grp=physical_second, n=1
```
The outer query should preserve the same row selected by the inner ordered
offset query.
### Additional context
- DuckDB `1.5.4` returns the expected `physical_second` row for the same SQL.
- The issue was reproduced deterministically in three fresh confirmations and
was originally exposed by two different generated queries; both reduce to
this two-row example.
- This may be related to #15886, but in this case the subquery sort is
semantically observable because it feeds `OFFSET` and cannot be removed.
- #22489 / #22525 involve an inner `ORDER BY/LIMIT` followed by an outer
`ORDER BY/OFFSET`; this reproducer has the opposite shape: an inner
`ORDER BY/OFFSET` is consumed by an outer `GROUP BY`.
- A lexical search did not find an issue with this exact query shape.
Contributor guide
Assessment
This issue has not been assessed yet.