anuragmantri opened a new issue, #72: URL: https://github.com/apache/datafusion-iceberg/issues/72
Follow-up to https://github.com/apache/datafusion-iceberg/pull/55#discussion_r4235376704. With `iceberg.planning.preserve_data_ordering` on, `TableProvider::scan` lists the data files, and if at most `max_merge_files` share a sort order, `IcebergDataSource` reports it and `open()` k-way merges one reader per file, with every file open, on one thread. It merges even when nothing above the scan uses the order: - a `SELECT *`, a `GROUP BY` on another column, or a hash join; - `ORDER BY id DESC`, which keeps its `SortExec` above the merge; - `ORDER BY id` on a nullable key, because Iceberg sorts ascending fields NULLS FIRST by default while DataFusion's `ORDER BY` means NULLS LAST, so the sort stays and we pay for both. `TableProvider::scan` can't tell whether anything above will use the order. **What DataFusion 55 offers.** The `PushdownSort` rule (on by default through `datafusion.optimizer.enable_sort_pushdown`) offers each `SortExec`'s order to its input through `try_pushdown_sort`. `Exact` removes the sort, `Inexact` keeps it over a rebuilt input, and `Unsupported` changes nothing. `DataSourceExec` passes the request to the source, and `CooperativeExec`, `ProjectionExec` and order-keeping `RepartitionExec` pass it down, but `FilterExec` doesn't, so a `WHERE` clause blocks it. The call is synchronous, so answering it means keeping the file listing in `TableProvider::scan`. **Sort-merge joins.** As @mbutrovich pointed out, a design that merges only after an `Exact` answer can lose the join case that `test_sorted_scan_removes_sort_from_sort_merge_join` checks, because `RepartitionExec::try_pushdown_sort` returns `Unsupported` when the repartition doesn't keep its input's order. Measured at `target_partitions = 4` with #55: an unfiltered sort-merge join keeps both scans' order and needs no `SortExec`, because the hash repartition over a single-partition scan keeps its input's order. With a `WHERE` on a join side, DataFusion adds a `FilterExec` and a round-robin repartition under the hash repartition, which loses the order. That side gets a `SortExec` while its scan still merges. The results are correct, but the merge is wasted, and pushdown can't remove it. Sorted aggregation never gets a `SortExec`, so pushdown can't help it either. **Options** 1. Report no order, implement `try_pushdown_sort`, and merge only after an `Exact` answer. This fixes every case above, but loses filtered `ORDER BY`, sorted aggregation, and filtered or multi-partition join inputs. 2. Keep reporting the order, and add cheap paths: (a) when the sort above shares no leading column with the files' order, answer `Inexact` with a source that reads the files without merging. This is safe because the sort stays, and it fixes `DESC` and nullable `ORDER BY id`. (b) read files back to back instead of merging when their key ranges don't overlap (#A). Neither helps a `SELECT *` with no sort. 3. Ship a `PhysicalOptimizerRule` that turns merging off on scans whose order nothing relies on. It must run before `OutputRequirements` removal, so users would have to rebuild the rule list, which is new public API. 4. Change DataFusion: `try_pushdown_sort` on `FilterExec`, or a general "order is consumed" signal. **Proposal.** Do 2(a) now, 2(b) once iceberg-rust has per-file bounds, and choose between 1 and 3 from measured plans. Feedback on that choice is very welcome. I'm working on this. I have 2(a) implemented and will open it as a PR stacked on #55, and I'll drive the discussion for the rest. -- 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]
