jayzhan211 commented on code in PR #26434: URL: https://github.com/apache/datafusion/pull/26434#discussion_r4305428691
########## docs/source/library-user-guide/upgrading/56.0.0.md: ########## @@ -1183,3 +1183,30 @@ A registered plan needs no `PhysicalExtensionCodec` at all, and two crates' plan `PhysicalExtensionCodec` is not deprecated: extension `PhysicalExpr`s and UDF payloads still need it, and it remains the fallback for unmigrated plans. See [issue #24625](https://github.com/apache/datafusion/issues/24625) for details. + +### Order-preservation hooks for limit pushdown are deprecated + +A file scan that advertises an output ordering now always reads its files in +order, so `ORDER BY ... LIMIT` queries no longer depend on `LimitPushdown` +telling the scan to keep its order. Before this change, pushing the fetch of a +`SortPreservingMergeExec` into an ordered Parquet scan could return the wrong +rows, because the scan then shared files between partitions or skipped row +groups based on its limit. + +The plumbing that carried this request is deprecated and no longer used: + +- `ExecutionPlan::with_preserve_order` and `DataSource::with_preserve_order` + are never called. Built-in operators no longer override them. Review Comment: Added both points: overrides are now dead code with no compiler warning, and a custom scan that advertises an output ordering must keep it by itself, also when a limit is pushed into it with `with_fetch`. ########## datafusion/datasource/src/file_scan_config/mod.rs: ########## @@ -1220,7 +1206,7 @@ impl DataSource for FileScanConfig { &self, config: &ConfigOptions, ) -> Option<Arc<dyn Any + Send + Sync>> { - if self.preserve_order + if self.preserves_file_order() Review Comment: Thanks for measuring this. I reproduced it with the same shape (16 sorted files: f00-f03 have 8M rows, f04-f15 have 0.5M; `WITH ORDER (a ASC)`; 8 partitions). The slowdown is real but smaller on my machine: `WHERE b > 0.999999 LIMIT 10` goes from 12 to 14 ms (`pushdown_filters` off) and from 4 to 6 ms (on). I went with your second option and documented the cost in the upgrade guide and the PR description. Measuring further showed why I'd rather not keep stealing for this one path. On `main`, a scan with a declared ordering already reads files in order for every other query, because the builder sets `preserve_order`. A plain `LIMIT` was the only exception, and that exception is the same path that returns wrong rows. On the same table, `main` with no `LIMIT` already pays for the uneven groups: | Query (pushdown_filters off) | main, `WITH ORDER` | main, no declared order | |---|---|---| | `SELECT sum(a + b) FROM t WHERE b > 0.5` | 233 | 93 | | `SELECT * FROM t WHERE b > 0.999999` | 32 | 13 | Work stealing does not explain this gap. With `enable_file_stream_work_stealing = false`, the no-order table still runs in about 91 ms. The real cause is file grouping. Unordered scans are cut into size-balanced byte ranges. Ordered scans keep `split_files`'s count-based chunks (`[f00, f01], [f02, f03], ...`, two 16M-row partitions), and `repartition_preserving_order` returns early once there are `target_partitions` groups. Cutting the concatenated groups into contiguous, size-balanced ranges keeps every group sorted, because each new group is a contiguous run in file order. I prototyped that on top of this PR (ms, median of 5 interleaved runs): | Query | pf | main | this PR | this PR + balanced ordered groups | |---|---|---|---|---| | `WHERE b > 0.999999 LIMIT 10` | off / on | 12 / 4 | 14 / 6 | 9 / 4 | | `sum(a + b) WHERE b > 0.5` | off / on | 233 / 310 | 232 / 310 | 93 / 119 | | `WHERE b > 0.999999` | off / on | 32 / 30 | 32 / 30 | 14 / 12 | | `ORDER BY a LIMIT 10` (+ filter variant) | off / on | 3-5 | 3-4 | 3-5 | So it fixes your case and gives a 2.3-2.5x win on full scans over ordered tables, with no work stealing and no order-sensitivity signal between rules. To stay correct, it keeps a re-cut only when `validate_orderings` keeps the same orderings for the new groups. That changes the file groups shown in EXPLAIN for ordered scans, so I opened it separately as draft #26458 (issue #26457) to keep this PR small. Measured against `main` there: 228 -> 78 ms for the full scan, 11 -> 8 ms for your `LIMIT` query, `ORDER BY` queries unchanged. ########## datafusion/physical-plan/src/limit.rs: ########## @@ -347,21 +344,16 @@ impl GlobalLimitExec { input, skip, fetch, - required_ordering, + // Deprecated, ignored. + required_ordering: _, Review Comment: Added `limit_decode_ignores_required_ordering`. It sets a `required_ordering` on the encoded `GlobalLimitExecNode` / `LocalLimitExecNode` and checks that decoding succeeds with skip/fetch kept. The sort node has no `expr`, so a decoder that still reads the field fails. I checked this: restoring the old decode makes the test fail with `PhysicalSortExpr is missing required field 'expr'`. -- 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]
