unikdahal commented on issue #68: URL: https://github.com/apache/datafusion-iceberg/issues/68#issuecomment-6129377651
Thanks @jverhoeks for the detailed write-up, and @alexanderbianchi for the cc. This overlaps a lot with what I've been working on, so here are a few notes. Fair warning: I'm about to be away for roughly a month, so please don't wait on me for any of this. **On the iceberg-rust hook (layer 2).** apache/iceberg-rust#3351 is very close to the `DynamicPredicate` trait you sketched. A `RuntimePredicateProvider` is sampled when each `FileScanTask` starts, and its predicate is combined with the planned one. From there it feeds the existing row-group, page-index and row-filter paths. The main difference is that the provider exposes a cheap `generation()` next to `snapshot()`, so the reader can cache the bound predicate per generation instead of per task. It hasn't been reviewed yet, so feel free to take it over, rework it, or replace it with your design. I won't mind at all. **On the caching problem.** I don't think this needs a new DataFusion API: - The free function `datafusion_physical_expr_common::physical_expr::snapshot_generation(&expr)` already returns a generation for a whole expression tree, adding up the generations of any dynamic filters inside it. Static expressions contribute 0. - `DynamicFilterPhysicalExpr::update()` bumps the generation. `wait_complete()` tells you when a filter is final (and `wait_update()` when it next changes). - Since DataFusion 55, `current()` caches the remapped expression per generation, so it should return the same `Arc` until the filter changes. If your prototype predates 55, that may explain the pointer-equality misses. What worked in my Comet prototype, which attaches one join filter per Iceberg scan: 1. Read the generation, take `current()`, then read the generation again. Only use the snapshot if both reads match, because an update can land between them. 2. Convert and bind once per (generation, schema, case sensitivity) and share the result across all tasks of the scan. In #3351 the cache sits behind a `Mutex`; my later version switched to an `RwLock` so a cache hit only takes a read lock. Two more suggestions, which I haven't tested myself: 3. For scans that receive filters from several joins, key the cache on the summed generation of the whole filter expression rather than on the first non-`None` value. When a later join publishes its filter, the sum changes and the cache refreshes, so a partial predicate never gets locked in. 4. On the ~580K-key probe set: with DataFusion's defaults (`hash_join_inlist_pushdown_max_distinct_values` = 150, `hash_join_inlist_pushdown_max_size` = 128 KiB), the join doesn't push an `IN` list that large. Past the cap it publishes min/max bounds together with a `HashTableLookupExpr`. The Iceberg translation can keep the bounds and drop the hash lookup, since the join still verifies every match, and that avoids binding a huge list for each task. If you do want to keep large lists, it helps to have a hash-based membership check in the row filter, since one comparison per literal gets expensive at that size. **Soundness notes.** Treat a failure to convert, bind or evaluate a runtime predicate as "skip it", never as a query error, and never let it bypass delete handling. Be careful with float columns, where NaN and ±0 can make statistics and row evaluation disagree. Layer 1 on top of the new `IcebergDataSource` (#50) sounds like a good first step either way. I'm happy to review when I'm back, if it's still open then. -- 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]
