Conversation
github-actions
Bot
added
physical-expr
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25681 +/- ##
==========================================
+ Coverage 82.51% 82.57% +0.05%
==========================================
Files 1141 1145 +4
Lines 439848 442151 +2303
Branches 439848 442151 +2303
==========================================
+ Hits 362952 365087 +2135
- Misses 54947 55057 +110
- Partials 21949 22007 +58 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
This was referenced Sep 24, 2026
Draft
adriangb
force-pushed
the
optional-filter-producers
branch
from
September 24, 2026 23:35
11d48c2 to
dfb8fc7
Compare
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 25, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 25, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
force-pushed
the
optional-filter-producers
branch
from
September 25, 2026 22:52
b61ced8 to
399d2e0
Compare
github-actions
Bot
added
documentation
adriangb
force-pushed
the
optional-filter-producers
branch
from
September 26, 2026 00:57
399d2e0 to
0b05ba2
Compare
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 26, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This was referenced Sep 26, 2026
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 26, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 26, 2026
A row that a dynamic filter removes is a probe row without a match. Its saving is the work that such a row gets: the evaluation and the hashes of the join keys and the hash table lookup. The check of the candidates (`equal_rows_arr`) and the output indices are work for the matches only. While the filter is on, most probe rows that reach the join are matches, thus this work made the measured saving too large. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 26, 2026
A row that a dynamic filter removes is a probe row without a match. Its saving is the work that such a row gets: the evaluation and the hashes of the join keys and the hash table lookup. The check of the candidates (`equal_rows_arr`) and the output indices are work for the matches only. While the filter is on, most probe rows that reach the join are matches, thus this work made the measured saving too large. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…IN list In partitioned mode the hash join pushes a routed filter: CASE hash_repartition % N WHEN i THEN bounds_i AND key IN (list_i) ... END When every non-empty build partition pushes an InList and no partition is canceled, the routing is redundant. Routing is a deterministic function of the join keys, so a build row with key K is in the partition that a probe row with key K routes to. A test of K against the union of all lists therefore accepts the same rows as the routed CASE. The per-partition bounds reject no additional rows either, because every key in a list is inside the bounds of its partition. The filter is now `key IN (union)`. This removes the per-row routing hash from the probe side, and the pruning code can use an InList (up to `max_in_list_size`), which it cannot do with a CASE. The union is capped at 1 MiB. Each partition's list is limited independently by `hash_join_inlist_pushdown_max_size`, so the union grows with the partition count; past the cap the routed CASE, where a probe row checks only one list, stays in use. Partitions that push a hash table, and builds with a canceled partition, also keep the CASE. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The per-partition InList arrays hold one entry per build row, and the collapsed union concatenated them. On TPC-DS SF1 with 12 partitions, Q65 pushed 54,000 entries for 6 distinct keys, and Q18 pushed 10,848 entries for 1,178 distinct keys. The pruning code uses an IN list only up to `max_in_list_size` (20) entries, so these filters gave no pruning term, and the long lists made the statistics evaluation for each file range slow. The union is now deduplicated and sorted with the arrow row format before the IN list is built. This works for single and struct (multi-column) keys and dictionaries, keeps one NULL, and makes the list order deterministic. The 1 MiB cap now applies to the deduplicated union. The collapsed filter also gets one range per key column, `col >= min AND col <= max`, from the combined bounds of all non-empty partitions, before the IN list. Every key in the union is inside this range, so the filter stays exact, and the pruning code can use the range when the list has more than `max_in_list_size` entries. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… as separate filters
A partitioned hash join pushed one dynamic filter to the probe side:
DynamicFilter [ CASE hash_repartition % N
WHEN i THEN bounds_i AND membership_i ... END ]
The pruning code cannot use a `CASE`, thus the build-side bounds did not
prune files, row groups or pages, even when the probe side is clustered by
the join key.
A partitioned join now pushes two dynamic filters:
DynamicFilter [ bounds ] AND DynamicFilter [ membership ]
- Bounds: the union of the bounds of all partitions (new `bounds_union`
module, adapted from the prototype in apache#24235). It does not need routing,
so the pruning code can use it. With hash partitioning each column gets
one range. With range partitioning the disjoint ranges of the partitions
are kept (up to 8 for each column, OR'd), so the filter also rejects keys
in the gaps between them.
- Membership: the routed `CASE` without the per-partition bounds (they
reject no row that the membership check of the partition accepts), or
the collapsed IN list when every partition pushes an IN list.
The bounds stay in the `CASE` (as before) when the union cannot describe
the build side: a canceled partition, or no usable bounds. An empty build
sets both filters to `false`. The NULL escape of null-equal and null-aware
joins wraps each filter.
A collect-left join does not change: it pushes one filter that holds
`bounds AND membership`. It has no routing `CASE`, so the pruning code can
already use its bounds.
Each filter has its own expression id. The join keeps each filter that
reached a consumer, and both get `update()` and `mark_complete()`. The
proto gets a new `dynamic_filter_bounds` field; a plan without it restores
one filter that holds both the bounds and the membership check, as before.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Add the `hj_ordered_subset` SQL benchmark suite. The probe table (`events`, 100 days x 200k rows, 100k-row row groups) is sorted by the join key, and the build side matches 1 day, 10 days, or a scattered 1% of the key range (control). The bounds of the hash join dynamic filter can prune the probe row groups outside the matched range, and the membership check passes almost every remaining row. Subgroups `partitioned` (Q01-Q03) and `collect_left` (Q04-Q06) force the HashJoinExec mode, which `expect_plan` checks. An assert checks that every build row matches exactly one event. The load SQL writes the data inline; HJOS_DAYS, HJOS_ROWS_PER_DAY and HJOS_RG_SIZE size it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…orrectness Add `OptionalFilterPhysicalExpr`, a transparent wrapper that marks a filter as optional: a consumer can skip it without changing the query result. A consumer can skip it only when the wrapper is a direct conjunct of the root AND chain of its predicate. In all other positions the wrapper is transparent, because `evaluate()` always evaluates the inner expression. `snapshot()` returns the inner expression, so pruning sees through it. Also add: - `split_optional` and `is_optional_filter` helpers in `physical_expr::utils` for consumers - `PhysicalOptionalFilterNode` proto message (field 29 in `PhysicalExprNode`) with self-encoding `try_to_proto`/`try_from_proto` No producer uses the wrapper yet, so there is no behavior change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Add the `datafusion_physical_expr::filter_stats` module with the shared primitives that adaptive filter code uses to measure filters at runtime: - `Clock`: a monotonic clock in nanoseconds that tests can replace. `SystemClock` is the real clock. `ManualClock` moves only when a test moves it, thus decisions that use time are deterministic in tests. - `FilterCost`: the rows in, the rows out and the evaluation time of one filter, and the derived cost for each row and rows removed for each nanosecond. - `duration_nanos`: a `Duration` in nanoseconds, saturated to `u64::MAX`. No code uses the module yet, thus behavior does not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
… than they save Add a runtime gate that pauses optional filters (filters that are not needed for correctness, such as hash join and TopK dynamic filters) when they cost more than they save. The gate is a per-stream state machine (Evaluate / Paused with exponential backoff) that restarts evaluation when the filter changes. The gate finds the dynamic filters one time with `DynamicFilterTracking::classify` and then polls their subscriptions, so a check does not walk the filter tree. Gates do not share state. At the end of each window of evaluated batches the gate pauses the filter if the window removed no rows, or if its evaluation time is larger than the work that the removed rows save: `(rows_in - rows_out) * saving_ns_per_row`. The saving for each row is the configured minimum plus an optional value that the consumer measures and updates (`MeasuredRowSaving`). The cost rule has a margin (pause above 1.1x the saving, resume below 0.9x) so that a filter does not switch on and off when cost and saving are almost equal. Add the `datafusion.execution.optional_filter_min_saving_ns_per_row` option (default 20). No operator uses the gate yet, so behavior does not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A consumer can now give the gate a fixed cost for each evaluated row in addition to the evaluation time, with `MeasuredRowSaving::set_overhead_ns_per_row`. The gate adds it to the cost of each window. The Parquet scan uses it for the fixed cost of a row filter stage, which is larger than the evaluation time of a cheap predicate. PR: apache#25674 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Each gate paid for its own first window and its own probes. A scan opens its files at the same time, thus a filter that removes nothing cost one window in each file (TPC-H Q9: five join filters that remove no rows and a CASE routing filter that costs 83 ns for each row, in each of 12 files). `SharedGateVerdict` holds the last pause (or end of a pause) of the gates of one plan site in one atomic word. A gate without evidence of its own (before its first decision, after a filter change, after a pause) uses a pause that another gate published after the last verdict that it saw: a new gate starts paused, and a gate in its first window or in a probe window stops and pauses. Thus usually only one gate probes after a pause. A gate that keeps the filter does not use the pauses of other gates (skewed data). A filter change clears the shared pause, because it was measured on the old filter. PR: apache#25674 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A gate decided after `sample_batches` batches, whatever their size. After a selective row filter a batch can have 2 to 7 rows, and the fixed cost of each call then looks like 600 to 8000 ns for each row: ClickBench Q23 paused the TopK filter on `EventTime` (0.4 ns for each row on full batches) on such windows, and the shared verdict spread these pauses to the other files. All decisions (pause, keep, probe) now need a window of at least `sample_batches` batches and `MIN_OBSERVED_ROWS` rows. The constant moves to `filter_stats`, so that the gate and the Parquet filter placement use the same sample size. All published shared pauses come from such windows. PR: apache#25674 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The gate assumed that each removed row saves `min_saving_ns_per_row` (20 ns) after the filter. For hash join dynamic filters this is the probe work of the join, and it is much smaller in star joins with small dimension tables: 3.5 to 8 ns for each probe row on the TPC-DS SF1 `date_dim` joins (Q65, Q67), 2 ns on Q90, 17 ns on TPC-H Q9. Filters that cost 3 to 7 ns for each row and remove 80% of the rows thus stayed on, and cost more than the join work they saved (8-18% slower than `pruning_only` on the bot). `RemovedRowWork` (in `filter_stats`) is the work that the producer of a filter does for each row that the filter removes, as the producer measures it. Each `DynamicFilterPhysicalExpr` has one, shared by all its derived filters. The gate uses the smallest measured work of the dynamic filters in its filter as the saving of a removed row, and `min_saving_ns_per_row` only until the producer has measured `MIN_OBSERVED_ROWS` rows (a prior). PR: apache#25674 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
`HashJoinExec`, `SortExec` (TopK) and `AggregateExec` now push their dynamic filters down as `Optional(DynamicFilter)`. The operator itself still removes the rows that the filter would remove, so the filter is only a performance hint. The producer keeps its own unwrapped `DynamicFilterPhysicalExpr` for `update()` and `mark_complete()`; only the pushed copy is wrapped. A partitioned hash join wraps each of its two pushed filters (bounds and membership) separately, so both stay direct conjuncts of the scan predicate. There is no behavior change. `OptionalFilterPhysicalExpr` evaluates its inner expression and `snapshot()` removes the wrapper, so scans and pruning use the filter as before. Only the EXPLAIN text changes, from `DynamicFilter [...]` to `Optional(DynamicFilter [...])`. Hash join key transfer rewrites a parent filter below the wrapper, so a transferred optional dynamic filter stays optional. A transferred required filter stays required: for inner and semi joins it is an exact replacement of the parent filter. Direct downcasts that must see through the wrapper: - `HashJoinExec::consumed_dynamic_filter` unwraps its self filters before it looks for a consumer, so the join still produces its filters. - `NestedLoopJoinExec::gather_filters_for_pushdown` uses the new `as_dynamic_filter` helper to route parent dynamic filters. Add `as_dynamic_filter` and `debug_assert_optional_on_root_chain` to `physical_expr::utils`. `ParquetSource::try_pushdown_filters` now checks in debug builds that optional filters are direct conjuncts of the root AND chain. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Update the expected plans in sqllogictest files. Pushed-down dynamic filters now show as `Optional(DynamicFilter [...])`. No query result changes. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
A row that a dynamic filter removes is a probe row without a match. Its saving is the work that such a row gets: the evaluation and the hashes of the join keys and the hash table lookup. The check of the candidates (`equal_rows_arr`) and the output indices are work for the matches only. While the filter is on, most probe rows that reach the join are matches, thus this work made the measured saving too large. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
force-pushed
the
optional-filter-producers
branch
from
September 26, 2026 17:21
1bdf8bd to
72cbd47
Compare
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 27, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 27, 2026
A row that a dynamic filter removes is a probe row without a match. Its saving is the work that such a row gets: the evaluation and the hashes of the join keys and the hash table lookup. The check of the candidates (`equal_rows_arr`) and the output indices are work for the matches only. While the filter is on, most probe rows that reach the join are matches, thus this work made the measured saving too large. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 27, 2026
The tests of apache#25729 show the TopK dynamic filter in the scan predicate. With apache#25681 the pushed filter is `Optional(DynamicFilter [...])`. Integration of apache#25729 and apache#25681 in the final state branch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 27, 2026
The tests of apache#25729 (page index select-all) and the test of apache#25727 for the coalesced rows at row group boundaries show the TopK dynamic filter in the scan predicate. With apache#25681 the pushed filter is `Optional(DynamicFilter [...])`. Integration of apache#25729, apache#25727 and apache#25681 in the final state branch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Conflict: an import in datasource-parquet/src/source.rs. apache#25780 imports `split_conjunction` (for `exact_filter`), this PR imports `debug_assert_optional_on_root_chain`. Both are kept. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 28, 2026
…h removed row The hash join records, in the `RemovedRowWork` of each dynamic filter that it produces, the rows of each probe batch and the time of the work that it does for every probe row, match or no match: the evaluation and the hashes of the join keys and the hash table lookup. A row that the filter removes before the join does not get this work. The work for a matched row (the output) is not in it: the filter does not remove matched rows. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 28, 2026
A row that a dynamic filter removes is a probe row without a match. Its saving is the work that such a row gets: the evaluation and the hashes of the join keys and the hash table lookup. The check of the candidates (`equal_rows_arr`) and the output indices are work for the matches only. While the filter is on, most probe rows that reach the join are matches, thus this work made the measured saving too large. PR: apache#25681 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
added a commit
to pydantic/datafusion
that referenced
this pull request
Sep 28, 2026
The tests of apache#25729 (page index select-all) and the test of apache#25727 for the coalesced rows at row group boundaries show the TopK dynamic filter in the scan predicate. With apache#25681 the pushed filter is `Optional(DynamicFilter [...])`. Integration of apache#25729, apache#25727 and apache#25681 in the final state branch. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
feat: mark hash join, TopK and aggregate dynamic filters as optional
Which issue does this PR close?
OptionalFilterPhysicalExprto mark filters not needed for correctness #25673 and feat: add OptionalFilterGate to pause optional filters that cost more than they save #25674. Review only the top four commits (the feature, the snapshot update, the probe work measurement and its fix for matched rows). The stack is on currentmain. feat: add OptionalFilterGate to pause optional filters that cost more than they save #25674 is new in this stack: the hash join records its probe work inRemovedRowWork(filter_stats), which the gate of feat: add OptionalFilterGate to pause optional filters that cost more than they save #25674 reads.Rationale for this change
Dynamic filters from hash join, TopK and aggregate are not needed for correctness. This PR marks them with
Optional(...)(#25673). Then consumers (#25682, #25683) can skip them when they do not pay for themselves.What changes are included in this PR?
HashJoinExec, collect-leftOptional(DynamicFilter [...])update(),mark_complete()HashJoinExec, partitioned (#25713)Optional(DynamicFilter [bounds]) AND Optional(DynamicFilter [membership])update(),mark_complete()SortExec(TopK)Optional(DynamicFilter [...])update()AggregateExecOptional(DynamicFilter [...])update()The hash join also measures the work that its dynamic filters save for each removed row. For each probe batch it records, in the
RemovedRowWorkof each dynamic filter that it produces (DynamicFilterPhysicalExpr::removed_row_work, #25674), the rows and the time of the work that it does for every probe row, match or no match:equal_rows_arr), work for matches onlyThe filter removes only rows without a match. While the filter is on, most probe rows that reach the join are matches, thus work that only matches get would make the measured saving too large (TPC-DS SF1 Q31: 6 ns for each probe row with the candidate check, 0.1 to 1.2 ns without it).
A row that the filter removes before the join does not get this work. The gate (#25674) uses this measurement as the saving of a removed row, in place of the configured
optional_filter_min_saving_ns_per_row(TPC-DS SF1 star joins: 3.5 to 8 ns for each probe row ondate_dimjoins, 2 ns on Q90; TPC-H Q9: 17 ns). Without dynamic filter pushdown the join measures nothing.The two filters of a partitioned join are wrapped separately, so each one stays a direct conjunct of the scan predicate and a consumer can skip each one on its own. A transferred parent filter stays required: for inner and semi joins it replaces the original filter.
Direct downcasts now look through the wrapper:
HashJoinExec::consumed_dynamic_filter(#24601 consumer check)NestedLoopJoinExecfilter routing (as_dynamic_filter)New helpers in
physical_expr::utils(moved here from #25673, their first callers):as_dynamic_filterNestedLoopJoinExec, testsdebug_assert_optional_on_root_chainParquetSource::try_pushdown_filters(debug builds only)No behavior change: the wrapper evaluates its child, and no consumer uses the gate in this PR.
What is the testing strategy for this PR?
test_pushed_dynamic_filters_are_optionaltest_pushed_dynamic_filters_by_partition_modetest_nlj_routes_optional_dynamic_filtertest_dynamic_filter_measures_removed_row_workMIN_OBSERVED_ROWSprobe rows, a positive value afteras_dynamic_filter, debug checkshould_paniccases.sltfiles).sltlines equals the old line withoutOptional(...)physical-planlib tests,proto_integration,core_integrationand the full sqllogictest suite pass. Locally,ordered_aggregate_spill.sltfails on this branch (a spill metric; no dynamic filter in the plan), which looks unrelated. It passes on themain-based branches.Are there any user-facing changes?
EXPLAINshowsOptional(DynamicFilter [...]). The tree format does not change.DynamicFilterPhysicalExprmust look through the wrapper (datafusion_physical_expr::utils::as_dynamic_filter).🤖 Generated with Claude Code