jayzhan211 opened a new issue, #26690:
URL: https://github.com/apache/datafusion/issues/26690
### Describe the bug
When a table has a declared sort order and more than one file, applying
`ORDER BY ... LIMIT n` to an `OFFSET` subquery returns fewer than `n` rows, and
sometimes none at all.
The physical plan applies the outer `LIMIT` first and the `OFFSET` after it:
```text
GlobalLimitExec: skip=5, fetch=5
SortPreservingMergeExec: [a@0 ASC NULLS LAST], fetch=10
DataSourceExec: file_groups={2 groups: [p1.parquet], [p2.parquet]}, ...,
output_ordering=[a@0 ASC NULLS LAST]
```
The merge keeps only the first 10 rows, and the limit then skips 5 of those
10, so 5 rows come back instead of 10.
### To Reproduce
In `datafusion-cli`:
```sql
COPY (SELECT value AS a FROM generate_series(1, 20) ORDER BY a) TO
'/tmp/df_offset/p1.parquet';
COPY (SELECT value AS a FROM generate_series(21, 40) ORDER BY a) TO
'/tmp/df_offset/p2.parquet';
CREATE EXTERNAL TABLE t (a BIGINT) STORED AS PARQUET LOCATION
'/tmp/df_offset/' WITH ORDER (a ASC);
SELECT count(*) FROM (SELECT * FROM (SELECT * FROM t OFFSET 5) ORDER BY a
LIMIT 10) WHERE a > 0;
-- 5, expected 10
SELECT count(*) FROM (SELECT * FROM (SELECT * FROM t LIMIT 30 OFFSET 15)
ORDER BY a LIMIT 10) WHERE a > 0;
-- 0, expected 10
EXPLAIN SELECT * FROM (SELECT * FROM t OFFSET 5) ORDER BY a LIMIT 10;
```
(The `WHERE a > 0` keeps `count(*)` from being answered from statistics.)
With `SET datafusion.execution.target_partitions = 1;` the same queries
return 10 rows, and without the `OFFSET` the result is also correct.
### Expected behavior
Skipping 5 of the 40 rows leaves 35, so `ORDER BY a LIMIT 10` should return
10 rows. A correct plan would keep `skip + fetch` rows in the merge
(`fetch=15`) before the limit skips 5, or not move the fetch below the `OFFSET`
at all.
### Additional context
`EXPLAIN VERBOSE` shows the change happens in `EnsureRequirements`:
```text
-- before EnsureRequirements
SortExec: TopK(fetch=10), expr=[a@0 ASC NULLS LAST],
preserve_partitioning=[false]
GlobalLimitExec: skip=5, fetch=None
DataSourceExec: 2 file groups, output_ordering=[a@0 ASC NULLS LAST]
-- after EnsureRequirements
GlobalLimitExec: skip=5, fetch=None
SortPreservingMergeExec: [a@0 ASC NULLS LAST], fetch=10
DataSourceExec: 2 file groups, output_ordering=[a@0 ASC NULLS LAST]
```
The fetch seems to be lost in `replace_with_order_preserving_variants`, in
`datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/replace_with_order_preserving_variants.rs`:
- `plan_with_order_preserving_variants` follows the chain of
order-maintaining operators below the sort, which includes the
`GlobalLimitExec` (around line 84), and turns the `CoalescePartitionsExec`
under it into a `SortPreservingMergeExec` with the sort's fetch (around line
150, called with `requirements.plan.fetch()` around line 271).
- The new plan satisfies the ordering, so the `SortExec` and its fetch are
dropped (around lines 274-291).
Putting a fetch below an operator that drops rows (`skip > 0`) changes which
rows survive. Other places already add `skip` to the fetch when they move it
below a limit, for example `input_fetch` in `enforce_sorting/sort_pushdown.rs`.
Related but different: #26488 (sort dropped below `OFFSET` when an aggregate
reads the subquery) and #25065 (fetched merges removed during requirement
enforcement).
### Willingness to contribute
I cannot contribute a fix for this bug at this time
--
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]
