Wrong rows when a subquery's ORDER BY ... LIMIT reads a sorted Parquet file #27194

Describe the bug

A subquery with ORDER BY ... LIMIT over a sorted Parquet file can return the wrong rows, with no error and in the default configuration, when an outer sort or join reads it. This affects tables with a known sort order (sorting_columns in the file, which COPY ... ORDER BY writes, or WITH ORDER) whose files have more than one row group.

To Reproduce

COPY (SELECT value AS s, value * 10 AS x FROM generate_series(1, 12) ORDER BY s)
TO 'repro/t.parquet' STORED AS PARQUET OPTIONS ('format.max_row_group_size' '3');
COPY (SELECT value AS k FROM generate_series(7, 8))
TO 'repro/b.parquet' STORED AS PARQUET;

CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'repro/t.parquet';
CREATE EXTERNAL TABLE b STORED AS PARQUET LOCATION 'repro/b.parquet';

-- Query 1
SELECT s FROM (SELECT * FROM t ORDER BY s LIMIT 3) ORDER BY s DESC;

-- Query 2
SELECT s FROM (SELECT * FROM t ORDER BY s LIMIT 3) ORDER BY x DESC LIMIT 1;

-- Query 3
SELECT i.s FROM b JOIN (SELECT s FROM t ORDER BY s LIMIT 3) i ON b.k = i.s;

Expected behavior

DuckDB returns the expected results on the same files. Tested on main at 4bf32cb.

Additional context

The file's sort order lets EnsureRequirements remove the subquery's sort, and LimitPushdown then puts the limit on the scan as limit=3. That limit keeps the first rows the scan reads, but two later rules change what it reads first:

• Queries 1 and 2: PushdownSort reverses the scan's row groups (pushdown_sort.rs:151), so the limit keeps the last rows. enable_sort_pushdown = false avoids this.
• Query 3: the post-optimization FilterPushdown pushes the join's dynamic filter into the scan (filter_pushdown.rs:427), so the limit applies after it. enable_dynamic_filter_pushdown = false avoids this.

Treating a fetch as a barrier in both rules fixes all three. On main that's 22 added lines, and all sqllogictests pass unchanged. Two snapshots from #21070 change, because they merge an outer filter into a filter that has a fetch, which changes the rows it keeps.

#23800 and #25065 don't touch these two rules. #27162 is a different bug: there the file isn't sorted across its row groups, and here it is.

Query 1: DataSourceExec: ..., projection=[s], limit=3, ..., sort_order_for_reorder=[s@0 DESC], reverse_row_groups=true
Query 2: DataSourceExec: ..., projection=[s, x], limit=3, ..., sort_order_for_reorder=[x@1 DESC], reverse_row_groups=true
Query 3: DataSourceExec: ..., projection=[s], limit=3, output_ordering=[s@0 ASC NULLS LAST], ..., predicate=DynamicFilter [ empty ]

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.