geoffreyclaude opened a new issue, #24927: URL: https://github.com/apache/datafusion/issues/24927
<!-- Issue title: SELECT DISTINCT ... ORDER BY ... LIMIT can return too few rows with multiple partitions --> ## Describe the bug DataFusion can return too few rows when `DISTINCT`, `ORDER BY`, and `LIMIT` are used together. The example below contains three unique values: `0`, `1`, and `2`. It asks DataFusion to remove duplicates, sort the result, and return up to three rows. All three values should therefore be returned, but DataFusion returns only two: `0` and `2`. This is a wrong query result rather than a performance problem. ## To Reproduce I reproduced this with `datafusion-cli` v55.0.0 built from `main` at commit [`20d1c5676118e0bca955ccf92c0d2cf665384d0d`](https://github.com/apache/datafusion/commit/20d1c5676118e0bca955ccf92c0d2cf665384d0d). Paste the following into the DataFusion CLI: ```sql SET datafusion.execution.target_partitions = 2; WITH s(x) AS ( VALUES (0), (1), (2) ORDER BY 1 LIMIT 3 ), t AS ( SELECT * FROM s UNION ALL SELECT * FROM s UNION ALL SELECT * FROM s ) SELECT DISTINCT x FROM t ORDER BY x LIMIT 3; ``` The `s` query contains exactly three rows, so its `LIMIT 3` does not remove anything. It is present because, without it, DataFusion removes the inner `ORDER BY` and does not create the execution plan that exposes the bug. The `UNION ALL` repeats those rows three times. Before `DISTINCT` is applied, the values are therefore: ```text 0, 1, 2, 0, 1, 2, 0, 1, 2 ``` The first `SET` statement asks DataFusion to process the query using two partitions. It is needed to reproduce the problem. Actual result: ```text +---+ | x | +---+ | 0 | | 2 | +---+ ``` ## Expected behavior The query should return all three unique values: ```text +---+ | x | +---+ | 0 | | 1 | | 2 | +---+ ``` `DISTINCT` should reduce the repeated input to those three rows, and the final `LIMIT 3` should allow all three rows through. ## Additional context The query returns the expected three rows if `datafusion.execution.target_partitions` is changed to `1`. It also returns the expected rows if the final `LIMIT 3` is removed. ### Why DataFusion appears to lose a value DataFusion implements `DISTINCT` in stages when the work is split across partitions. Each partition first finds its distinct values, and DataFusion then combines those partial results to remove duplicates between partitions. In this query, DataFusion moves the final limit into the plan too early. It keeps only three rows in each partition *before* the partial results have been combined. Those three places can all be taken by copies of the same value, so another value is discarded. The later `DISTINCT` step cannot recover a value that has already been removed. In the physical plan, the important part is the `TopK` with `limit: 3` below the final aggregate: ```text SortPreservingMergeExec (limit: 3) AggregateExec (final DISTINCT) SortExec(TopK) (limit: 3) <-- LIMIT is applied too early RepartitionExec (2 partitions) AggregateExec (partial DISTINCT) ``` `TopK` is DataFusion's optimized implementation of `ORDER BY ... LIMIT`. ### Related issues I could not find an open issue for this exact case. - [#24272](https://github.com/apache/datafusion/issues/24272) also reports incorrect results from an ordered limit, but through a different path: it is about Parquet row-group pruning losing an ordering requirement. This issue is about applying a limit before the final `DISTINCT` step. - [#23822](https://github.com/apache/datafusion/issues/23822) is the closest report I found. It also involved `DISTINCT ... ORDER BY ... LIMIT`, but it was closed after [#21107](https://github.com/apache/datafusion/pull/21107). The query above still fails on the current `main` branch. - [#16638](https://github.com/apache/datafusion/issues/16638) and [#16641](https://github.com/apache/datafusion/pull/16641) fixed a similar problem where a limit was moved through an anti join. - [#12748](https://github.com/apache/datafusion/issues/12748) and [#12766](https://github.com/apache/datafusion/pull/12766) discuss the same general safety rule: a limit must not be moved across a step that can reduce the number of rows unless doing so is known to be safe. <details> <summary>Possible implementation location</summary> DataFusion internally calls the row count attached to a sort its `fetch`. The likely gap is the `satisfy_parent` branch in [`pushdown_sorts_helper`](https://github.com/apache/datafusion/blob/20d1c5676118e0bca955ccf92c0d2cf665384d0d/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs#L248-L270). That branch passes the fetch to the child when the child already provides the requested ordering. A nearby path, [`pushdown_requirement_to_children`](https://github.com/apache/datafusion/blob/20d1c5676118e0bca955ccf92c0d2cf665384d0d/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs#L324-L360), checks whether moving the fetch across an operator is safe. The `satisfy_parent` branch appears to bypass those checks. In particular, `AggregateExec` can reduce the number of rows, which makes moving the fetch below it unsafe here. </details> ### AI assistance Codex was used to help develop the reproducer and draft this issue. -- 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]
