feat(scanner): push dynamic filters into FilteredReadExec

[wjones127]

Makes FilteredReadExec a first-class DataFusion physical filter-pushdown sink (like DataSourceExec), so dynamic filters produced by upstream operators are threaded into the scan and evaluated there instead of only above it. Draft for #7562.

What changed

• FilteredReadExec implements handle_child_pushdown_result (Post phase). It absorbs any offered filter whose columns are all produced by the scan, stores it, and applies it as a per-batch filter on the fully-projected output. gather_filters_for_pushdown is left at its default, mirroring DataSourceExec.
• TakeExec forwards filters that reference only its input's columns down to its child (mirrors CoalesceBatchesExec), so a filter can reach the scan beneath a take. Filters on take-added columns stay above it.
• scanner installs FilterPushdown::new_post_optimization() in get_physical_optimizer(). Post phase only — static WHERE predicates are already baked into full_filter by Lance's logical filter plan, so running the Pre phase here would double-apply them.

A DynamicFilterPhysicalExpr re-reads its current predicate on every evaluate, so a TopK threshold that tightens as the sort's heap fills is honored per batch without re-planning.

Verified benefit: TopK (ORDER BY … LIMIT)

SELECT id, value FROM t ORDER BY value LIMIT 10 now plans as:

SortExec: TopK(fetch=10), expr=[value ASC]
  LanceRead: projection=[id, value], pushdown_filters=[DynamicFilter [ value IS NULL OR value < 2 ]]

The scan drops rows the top-k provably can't contain before they reach the sort. With a wide, late-materialized column the win compounds — the scan reads only the narrow sort key, and the downstream Take fetches the wide column for the surviving rows only:

Take: columns="id, sortkey, _rowid, (wide)"
  SortExec: TopK(fetch=10), expr=[sortkey ASC]
    LanceRead: projection=[id, sortkey], full_filter=id >= 0, pushdown_filters=[DynamicFilter [ … ]]

Scope / honest limitations

This is deliberately the smallest useful slice — the reusable protocol spine plus CPU pruning (rows dropped at the scan). Two larger wins are follow-ups:

1. Scan I/O pruning (ZoneMap). Today the pushed filter drops rows after decode; it does not yet skip scan I/O. A ZoneMapIndex/btree probe of the filter's IsIn/Range (both already answer SargableQuery) would give Parquet-style range skipping. Depends on Always update zone map index during write time #4522 (ubiquitous zone maps) and Change SargableQuery::IsIn to hold an Arrow array instead of Vec<ScalarValue> #7192 (large IsIn).
2. merge_insert / hash join. The protocol is wired uniformly, but in the current backfill plan the target LanceRead is the hash-join build side (CollectLeft, Inner), so DataFusion pushes the join's dynamic filter to the probe (the source), not the target scan. Filtering the target by source keys needs a join-side change, tracked separately.

Tests

• test_topk_dynamic_filter_pushdown — multi-fragment TopK: the dynamic filter reaches LanceRead and results are exactly the true top-k.
• test_handle_child_pushdown_result_absorbs_dynamic_filter — unit test: Post-phase absorbs and rewrites (Pre-phase declines); a filter tightened after pushdown takes effect at execute time.
• test_take_filter_pushdown_pass_through — TakeExec routes an input-column filter down and keeps a take-added-column filter above.
• Existing take + merge_insert suites (183 tests) stay green.

cargo fmt and cargo clippy -p lance --tests -- -D warnings pass.

[coderabbitai]

Important

Review skipped

Draft detected.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: QUIET

Plan: Pro Plus

Run ID: 4ebb502b-b5c6-4dd0-af80-a50282809cc1

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

_{Comment @coderabbitai help to get the list of available commands.}

[codecov]

Codecov Report

❌ Patch coverage is 97.78598% with 6 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
rust/lance/src/io/exec/filtered_read.rs	97.18%	2 Missing and 2 partials ⚠️
rust/lance/src/io/exec/take.rs	97.05%	0 Missing and 2 partials ⚠️

📢 Thoughts on this report? Let us know!