ORDER BY ... LIMIT over an OFFSET subquery returns too few rows when the table is sorted and has several files #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:

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:

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:

-- 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