jverhoeks opened a new issue, #68:
URL: https://github.com/apache/datafusion-iceberg/issues/68
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
`DynamicFilterPhysicalExpr`s from parent joins, and translates them to Iceberg
`Predicate`s 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:
```rust
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.
--
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]
