Support dynamic filter pushdown in FilteredReadExec

[wjones127]

DataFusion 53.1 (our vendored version) implements dynamic filter pushdown for hash joins: HashJoinExec builds a DynamicFilterPhysicalExpr from the build side's join keys once the build phase completes, and propagates it to the probe side via ExecutionPlan::gather_filters_for_pushdown / handle_child_pushdown_result (datafusion-physical-plan-53.1.0/src/joins/hash_join/{exec.rs,shared_bounds.rs}; see also the DataFusion blog post).

FilteredReadExec (rust/lance/src/io/exec/filtered_read.rs) does not implement either trait method, so this pushdown is currently a no-op for any plan where a Lance scan is the probe side of a join — the filter is generated and offered but has nowhere to land.

Why this matters

merge_insert (#7367) plans the target scan as a FilteredReadExec probed against a hash join built from the (usually smaller) source side. If FilteredReadExec accepted the pushed-down dynamic filter, it could skip decoding — or, if per-page/fragment key statistics are available, skip reading entirely — target rows that can't match any source key, in a single sequential scan pass. That would avoid the join → TakeExec round trip (and its random-I/O cost) that late-materialization approaches have to reason about, for at least the case where a dynamic filter usefully narrows the scan.

This is likely useful beyond merge_insert for any join where a Lance scan is the probe side and the build side is small/selective.

Scope

• Implement ExecutionPlan::gather_filters_for_pushdown / handle_child_pushdown_result on FilteredReadExec to accept a dynamically-populated PhysicalExpr.
• Apply the filter during scan execution: at minimum evaluate it per-batch/row before returning results; if feasible, use existing column statistics (if any) for page/fragment-level pruning.
• Benchmark against a join-heavy workload (e.g. merge_insert backfill) to confirm it reduces I/O without regressing plans where the filter doesn't help (non-clustered keys, filters that can't prune anything).

Surfaced while assessing follow-up work for #7367 (merge_insert late materialization).

[wjones127]

Feasibility findings

Feasible, and less work than it first looks — but the payoff is in the plan shape around the filter, not the filter plumbing.

Current state

• FilteredReadExec is on the probe side of the merge_insert hash join — the right position. ✅
• It implements neither gather_filters_for_pushdown nor handle_child_pushdown_result, so it inherits the defaults and the pushdown is a confirmed no-op. (No Lance exec implements these today.)
• FilterPushdown does run on the merge_insert path (create_plan → stock SessionContext → default optimizer set, incl. new_post_optimization()), but not on the general scanner path (get_physical_optimizer() omits it).
• DF's join dynamic filter is Inner-join only, so only the backfill config (update, no insert/delete-unmatched) qualifies; full upsert → Right/Full → no filter generated. Small build side → exact InList + bounds; large → hashmap + bounds.

Late materialization: Lance vs DataFusion/Parquet

These prune I/O in structurally different places.

flowchart TB
  subgraph DF["DataFusion / Parquet — prunes inside the scan"]
    P["DataSourceExec<br/>predicate + column stats<br/>skips row-groups/pages, filters on decode"]
  end
  subgraph LC["Lance — plan-level split (2 nodes)"]
    direction TB
    T["TakeExec — late/wide cols, by _rowid"] --> FR["FilteredReadExec — eager cols + filter"]
  end

Loading

• Parquet pushes the filter into the scan; it prunes via inline column statistics and filters on decode. One node.
• Lance splits columns at plan time (MaterializationStyle on the Scanner, byte-width heuristic) into eager vs late. The eager FilteredReadExec reads + filters in full; the TakeExec then fetches wide cols only for survivors. Savings come from the Take, not from the scan skipping ranges.
• The split is gated on a scan-time filter (has_refine()), which a merge_insert target scan lacks — so today it reads its full projection in one pass, no pre-join take.
• ⚠️ Placement matters for us: a pre-join take benefits from the filter; a post-join take doesn't (its input is already the matched set).

ZoneMap ≈ Parquet statistics: we have the same primitive via ZoneMapIndex (per-zone min/max), just as an explicit scalar index. So Parquet-style scan-level pruning is reachable when such an index exists — a larger index-integration follow-up.

Proposed direction (pre-join split; ~1–1.5 wk)

flowchart TB
  HJ["HashJoinExec (Inner)"] -->|probe| TK["TakeExec (wide)<br/>needs pushdown pass-through"]
  TK --> FRE["FilteredReadExec (narrow + dynamic filter)"]
  HJ -->|build| SRC["source (small)"]

Loading

1. Implement the two trait methods on FilteredReadExec (accept + evaluate the pushed PhysicalExpr).
2. Trigger the split by reserving a placeholder filter slot (DF's DynamicFilterPhysicalExpr(lit(true)) pattern) — inject at the physical level; a logical WHERE true gets folded away.
3. Add pushdown pass-through to TakeExec so the filter reaches the scan below it.

[wjones127]

ZoneMap + dynamic filters: the "index evaluated too early" story

Follow-up to the comparison above, focused on how ZoneMap could plug in and on a real limitation it exposes: we commit to an index query before dynamic filters exist.

ZoneMap already speaks the dynamic filter's language

A hash-join dynamic filter is key IN (build keys) AND min ≤ key ≤ max. That maps 1:1 onto SargableQuery, which already has IsIn(..) and Range(..) — and both ZoneMap and btree answer them (zonemap.rs evaluate_zone_against_query; btree.rs pages_in/pages_between). ZoneMap stores per-zone min/max/null/nan and returns SearchResult::AtMost (a superset → recheck), i.e. zone/fragment pruning — the same primitive Parquet gets from column statistics, just as an explicit scalar index.

We even already have the runtime-keys→index-probe node: MapIndexExec / IndexedLookup (io/exec/scalar_index.rs) takes an input batch of values, builds SargableQuery::IsIn(values), probes the index, and emits a row-id set. So "turn runtime keys into an index query" is not greenfield — it's wired to an explicit input stream today, not to a pushed-down filter.

The limitation: query shape is frozen at plan time; only execution is deferred

flowchart LR
  A["Plan build<br/>create_filter_plan / scalar_indexed_scan<br/>→ index query SHAPE frozen"] --> B["Physical opt<br/>hash join generates<br/>dynamic filter"]
  B --> C["Execute (first poll, post-build)<br/>OnceCell runs index query<br/>→ dynamic filter is available here… but unused"]

Loading

Scalar-index usage is decided during planning (create_filter_plan → a fixed ScalarIndexExpr/index_input child + refine split). But the query executes lazily via the OnceCell in get_or_create_plan_impl — on first poll, which for a join probe is after the build side completes. So the data needed to run a dynamic-filter-aware index probe is already present at execution time; what's missing is that we've already committed to which index query to run. The dynamic filter — often the single most selective predicate on the join key — never gets to shape it.

So the answer to "should index evaluation be more deferred?" is: execution timing is already fine; what needs deferring is index-query construction — allow a pushed-down predicate to refine (or add) an index probe at execute time.

What combining looks like

When FilteredReadExec holds a pushed-down dynamic filter on column C and C has a ZoneMap/btree index: at plan-init (post-build) convert the filter's current IsIn/bounds → SargableQuery → probe the index (reusing the MapIndexExec machinery) → intersect the resulting applicable_fragments/zones with the scan's ranges → skip zones that can't contain any build key, and recheck with the (inexact) filter. This prunes the scan's own I/O — including the narrow eager column — i.e. the Parquet-parity path, not just a downstream TakeExec.

"Re-run if there's a big change" — a cost gate

This is the right framing. Baseline = the statically-planned scan (full, or the pre-planned index result). If a dynamic filter arrives that is much more selective than the static plan assumed (small InList, tight bounds), it's worth the extra index probe; otherwise skip it (probing isn't free, and ZoneMap only prunes if the key is clustered). That's the same early-vs-late/recheck cost decision as #4107 and the index-choice sophistication of #5224. One ordering subtlety: the runtime index result must intersect with any existing EvaluatedIndex mask, not replace it.

Dependencies / related roadmap

• Always update zone map index during write time #4522 (always update zone map at write) — the force multiplier. If zone maps are maintained automatically, the join key almost always has a usable one, so this pruning is broadly applicable rather than opt-in. Arguably the precondition for it to matter in practice.
• Change SargableQuery::IsIn to hold an Arrow array instead of Vec<ScalarValue> #7192 (SargableQuery::IsIn hold an Arrow array) — a dynamic-filter InList can be large; IsIn currently holds Vec<ScalarValue>. Feeding a big key set into an index probe efficiently depends on this. Direct dependency.
• Potential scanner optimization for late materialization plus recheck #4107 / Add more sophistication to index choice when multiple indexes apply #5224 — the selectivity/early-vs-late cost model and multi-index choice (prefer exact over inexact).
• Column Statistics #4163 (column statistics) — min/max exposure overlaps with what ZoneMap already computes.
• Existing ZoneMap caveats to keep in view: Zonemap index reading too many files #5130 (reads too many files), ZoneMap scalar index returns incomplete results on datasets with stable row IDs #7434 (incomplete results with stable row IDs).

Bottom line: the ingredients already exist — ZoneMap answers IsIn/Range, a runtime-keys→index-probe node exists, and index execution is already lazy/post-build. The missing piece is letting a pushed-down dynamic filter refine the index query at execute time, behind a selectivity gate. The largest practical dependencies are #4522 (ubiquitous zone maps) and #7192 (efficient large IsIn).

[wjones127]

Reframing: make the scan a general DataFusion dynamic-filter sink

Stepping back from the merge_insert/Take specifics — the real goal here is to make FilteredReadExec a first-class filter-pushdown target, like DataSourceExec (Parquet), so it benefits from current and future DataFusion optimizations uniformly rather than one integration per producer.

The pushdown protocol is source-agnostic. Both HashJoinExec and SortExec (TopK) create a DynamicFilterPhysicalExpr and offer it in the Post phase via the same gather_filters_for_pushdown path (sorts/sort.rs:972,1370 and joins/hash_join/exec.rs). From the scan's side it's identical: "here's an Arc<dyn PhysicalExpr> over your output columns." Implement the protocol once and every producer is covered.

flowchart TB
  HJ["HashJoinExec (Inner) — key IN (build keys) + bounds"] --> P
  TK["SortExec + fetch (TopK) — sort_col < threshold"] --> P
  FUT["future DF dynamic filters …"] --> P
  P["FilterPushdown (Post phase)<br/>gather / handle protocol"] --> FR["FilteredReadExec<br/>accept · evaluate · prune"]

Loading

The deliverable

1. gather_filters_for_pushdown — advertise "I accept filters over my output columns" (validate with ChildFilterDescription::from_child).
2. handle_child_pushdown_result — store the accepted expr, rebuild the node.
3. Evaluate at execute time; prune I/O where possible (see ZoneMap below), else post-filter per batch.

Everything about downstream Take placement / merge_insert plan shape is a consumer concern, not part of this — dropping it from scope.

TopK is the better lead case than joins

• Not gated. Any ORDER BY … LIMIT k over a Lance scan qualifies, vs. the Inner-join-only restriction on the hash-join filter.
• It's a Range predicate (sort_col < threshold), which ZoneMap/btree prune best — skip every zone whose min exceeds the current threshold. Common, high-value shape (top-k, recent-N).

One behavioral difference to design for

The TopK filter tightens during execution — the threshold moves as the heap fills — whereas a join filter is set once, after the build completes. So the scan must not snapshot the predicate once at plan-init; it should re-read .current() as it advances and re-prune per fragment/zone. A poll-the-current-value evaluate/prune loop serves the set-once (join), monotonically-tightening (TopK), and future cases with the same code.

ZoneMap as the I/O-pruning layer (see prior comment)

SargableQuery already has Range and IsIn, and ZoneMap/btree answer both — so a Range (TopK) or IsIn+bounds (join) filter maps straight onto an index probe returning AtMost zone pruning. That's what turns "accept + post-filter" (CPU only) into real scan-level I/O savings, the Parquet-parity path.

Phase decision

Dynamic filters are Post-phase (join, TopK). Static WHERE predicates arrive in the Pre phase and overlap Lance's existing logical TableProvider::scan(filters) path. Suggest Post-only first (avoid double-applying), and treat Pre-phase unification — eventually retiring the logical path — as a separate follow-up.

---

Net: the spine of this issue is "make FilteredReadExec a general DF dynamic-filter sink, with ZoneMap as the pruning layer." Joins and merge_insert are one consumer; TopK is another (and a cleaner first target); future DF work comes for free.