Dynamic filter pushdown into IcebergTableScan

Đang mở
  • Có gì không ổn?
#68 3 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
35/100
Loại issue
Tính năng
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
rust
Lĩnh vực
backend, databases, performance

Hướng nghiên cứu

Start with crates/datafusion/src/read/scan.rs and trace how IcebergTableScan gathers and handles filter pushdown, then follow the existing expr_to_predicate path. The issue describes two layers: per-batch filtering here and a per-task dynamic predicate hook in iceberg-rust; the latter needs changes outside this repository. Done means dynamic filters reach the scan and results remain correct, with row-group pruning requiring the iceberg-rust hook.

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

Mô tả

Moved from apache/iceberg-rust#2376 at the suggestion of @alexanderbianchi, now that the DataFusion integration lives here.

Is your feature request related to a problem or challenge?

IcebergTableScan (crates/datafusion/src/read/scan.rs) does not take part in DataFusion's dynamic filter pushdown. The filters that HashJoinExec builds at execution time never reach the Iceberg scan. Probe-side scans like lineitem therefore can't prune row groups using the build side's join keys.

Example: TPC-H q14 (lineitem JOIN part ON l_partkey = p_partkey) at SF10. Even with enable_dynamic_filter_pushdown, parquet.pushdown_filters, parquet.reorder_filters and parquet.bloom_filter_on_read all on, lineitem is scanned in full. The dynamic filter is only evaluated row by row in the FilterExec above the scan, which doesn't skip any row groups. On the same data, Trino's Iceberg connector runs this query about 5x faster. Trino samples its DynamicFilter once per split and feeds it into row-group statistics and bloom filters before reading any data.

The scan takes filters: &[Expr] once and converts them into a single static Predicate. Nothing feeds it a predicate that only becomes useful mid-execution.

Describe the solution you'd like

Two layers:

1. In this repo: IcebergTableScan implements gather_filters_for_pushdown / handle_child_pushdown_result, accepts DynamicFilterPhysicalExprs from parent joins, and translates them to Iceberg Predicates through the existing expr_to_predicate path. The current soundness rule still applies: the predicate only has to be implied by the filter, and unconvertible parts are dropped.

2. In iceberg-rust: a hook that samples the dynamic predicate once per FileScanTask, along these lines:

pub trait DynamicPredicate: Send + Sync {
    /// Called right before a FileScanTask opens its Parquet file.
    /// None = not selective yet (e.g. build side not finished).
    fn current(&self) -> Option<Predicate>;
}

impl TableScanBuilder {
    pub fn with_dynamic_predicate(self, dp: Arc<dyn DynamicPredicate>) -> Self;
}

ArrowReader::process_file_scan_task would AND the result with the static predicate. From there it flows into the row filter, row-group pruning and page-index selection that the reader already supports.

Until (2) lands, (1) can apply the filters per batch after decode. That is already a win, but it doesn't reach row-group pruning.

Results from a downstream prototype
  • Per-batch only (layer 1): -21% total TPC-H time at SF1, -9.4% at SF10.
  • Per-task DynamicPredicate (both layers): -12.4% at SF10, all 22 queries returning matching results.
Open problem: caching the bound predicate across tasks

Each FileScanTask pays for conversion plus Predicate::bind. For a hash-join probe set of about 580K keys (q04 at SF10), bind takes about 61 ms and conversion about 19 ms. With about 88 lineitem files and parallelism 12, that's roughly 560 ms of wall-clock per scan.

I tried five cache designs and all of them regressed: Mutex + Arc::ptr_eq, OnceLock, IN-list caps of 4096 and 200, and per-filter Mutex<CachedBound>. Each failed in one of three ways:

  • DynamicFilterPhysicalExpr::current() calls remap_children, which returns a new Arc on the join pushdown path. Pointer-equality caches never hit.
  • In multi-join scans the filters seal at different times. A first-non-None cache locks in a partial predicate. That cost q04 +34%, q06 +74% and q16 +42%.
  • Locking on every per-task call adds contention across concurrent scan tasks.

A correct cache needs DataFusion to tell consumers when a filter's state has changed and when it is final. That could be an is_sealed() or a monotonic generation() on DynamicFilterPhysicalExpr. If that's the right direction, I can open a matching issue in apache/datafusion.

Willingness to contribute

Yes. I can open a PR for layer 1 here, and a PR in iceberg-rust for the DynamicPredicate hook.

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.