apache / apache/datafusion

Returns the wrong grouped row when an outer `GROUP BY` consumes a subquery containing `ORDER BY ... OFFSET`.

Open
#23,534 1 comment 0 reactions 1 assignee Claimed by @jayhan94 View on GitHub
bug
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.