Merge sorted files in IcebergDataSource only when something above the scan uses their order

Đang mở
  • Có gì không ổn?
#72 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
5/5
Thời gian dự kiến
Hơn một tuần
Mức phù hợp với người mới
25/100
Loại issue
Tính năng
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
rust
Lĩnh vực
backend-api-design, databases, performance

Hướng nghiên cứu

Read the DataFusion PushdownSort behavior and the discussion linked from PR #55, then inspect IcebergDataSource and TableProvider::scan in the repository. Compare the proposed options against the described SELECT, filtered ORDER BY, aggregation, and sort-merge join plans; completion means agreeing on an approach and validating its behavior against those cases. The issue says the author already implemented option 2(a) and plans to submit it as a stacked PR.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

Follow-up to https://github.com/apache/datafusion-iceberg/pull/55#discussion_r4235376704.

With iceberg.execution.preserve_data_ordering on, TableProvider::scan lists the data files, and if at most iceberg.execution.max_merge_files_per_partition 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 (#71).
    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.

Ngôn ngữ chính
Rust
Star
33
Fork
34
Merge trung bình
4 ngày 14 giờ
Pull request đã merge (30 ngày)
10

Chuẩn bị môi trường

Dự án này không cung cấp dev container, Dockerfile hay hướng dẫn đóng góp, nên bạn cần tự thiết lập môi trường: hãy bắt đầu từ README và xem hướng dẫn đóng góp lần đầu của chúng tôi để biết các bước chung.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/datafusion-iceberg

Tất cả issue của apache/datafusion-iceberg

Issue tương tự

Thêm issue về Rust

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.