feat(physical-expr): DynamicFilterTracker for cheap dynamic-filter change detection (#22460)

Extracted from a design discussion around the duplicated "does this
filter have a dynamic portion that might change?" / "has the filter
changed?" patterns (e.g. #22450, `FilePruner`).

## Rationale for this change

`DynamicFilterPhysicalExpr` has a rich *producer* API (`update()`,
`mark_complete()`, `wait_update()`, `wait_complete()`), but *consumers*
that hold a predicate which *contains* dynamic filters only had a bare,
recursive `snapshot_generation() -> u64`. Call sites hand-rolled the
same boilerplate around it: store a `last_generation`, recompute
`snapshot_generation(&predicate)` (a full tree walk) on **every** check,
diff it, and rebuild an expensive `PruningPredicate` on change.
`FilePruner` did exactly this, and none of these consumers exploited
`mark_complete()`.

This adds a small consumer-side counterpart so the pattern lives in one
place, driven by the existing `watch` channel rather than by re-walking
the tree.

This immediately eliminates some tree traversals (we were constantly
traversing the expression tree to check if any filters updated). Long
term I hope this makes changes like #22450 easier.

## What changes are included in this PR?

**New public API (`datafusion_physical_expr`):**
- `DynamicFilterTracking` (`classify` → `Static` / `AllComplete` /
`Watching`, plus `contains_dynamic_filter` / `watcher`) and
`DynamicFilterTracker` (`changed`). A tracker walks a (possibly
composite) predicate **once**, subscribes to every still-incomplete
dynamic filter, and answers `changed()` by polling only that shrinking
set — steady-state is one atomic load per filter, no tree walk, no lock
until something actually moves.
- Lives in the `expressions::dynamic_filters` module
(`dynamic_filters.rs` → `dynamic_filters/mod.rs`, tracker in
`dynamic_filters/tracker.rs`). The subscription plumbing (`subscribe`,
`DynamicFilterSubscription`, `DynamicFilterChange`, `observe`,
`is_complete`) is `pub(crate)`; test-only constructors are
`#[cfg(test)]`.

**Consumers:**
- `FilePruner` is driven by `DynamicFilterTracking` instead of
`snapshot_generation` polling, and now decides its own existence in
`try_new` (a static predicate with no usable stats builds no pruner).
- The Parquet opener skips wrapping the scan in `EarlyStoppingStream`
when nothing can change, and no longer needs an "is it dynamic?" gate.

## Are these changes tested?

Yes — unit tests for the tracker (classification, detect-update-once,
`mark_complete` is not a change, coalesced update+complete,
multi-filter), plus the existing `datafusion-pruning` /
`datafusion-datasource-parquet` suites (incl. the
static/dynamic/partition opener pruning test) pass unchanged.

## Are there any user-facing changes?

New public API as above (additive). One **deprecation** and one behavior
change, both documented in the [DataFusion 55.0.0 upgrade
guide](docs/source/library-user-guide/upgrading/55.0.0.md):

-
`datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr`
is **deprecated (since 55.0.0)** — downcast to
`DynamicFilterPhysicalExpr` or use `DynamicFilterTracking`.
(`snapshot_generation` itself is unchanged — still backing the FFI
vtable and proto roundtrip.)
- `FilePruner::try_new` now returns `None` for a purely static predicate
over a file with no usable column statistics (previously `Some` whenever
a statistics struct was present).

## Followups

I noticed a possible follow-up gate refinement, tracked in #22495.

This also opens up the possibility to deprecate / remove the `snapshot`
/ `generation` machinery from the public physical expr APIs. These new
APIs (the watchers, tracker) subsumes much of the functionality, and I
don't think we want to add `PhysicalExpr::watch`. And after several
releases the only thing using it right now is dynamic filters, i.e. no
other legitimate use case has materialized.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: adriangb

datafusion/datasource-parquet/src/opener/mod.rs

@@ -52,9 +52,7 @@ use datafusion_common::{ColumnStatistics, Result, ScalarValue, Statistics, exec_
 use datafusion_datasource::{PartitionedFile, TableSchema};
 use datafusion_physical_expr::simplifier::PhysicalExprSimplifier;
 use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
-use datafusion_physical_expr_common::physical_expr::{
-    PhysicalExpr, is_dynamic_physical_expr,
-};
+use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::metrics::{
     BaselineMetrics, Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory,
@@ -618,18 +616,19 @@ impl ParquetMorselizer {
             .with_category(MetricCategory::Rows)
             .global_counter("num_predicate_creation_errors");
 
-        // Apply literal replacements to projection and predicate
-        let file_pruner = predicate
-            .as_ref()
-            .filter(|p| is_dynamic_physical_expr(p) || partitioned_file.has_statistics())
-            .and_then(|p| {
-                FilePruner::try_new(
-                    Arc::clone(p),
-                    &logical_file_schema,
-                    &partitioned_file,
-                    predicate_creation_errors.clone(),
-                )
-            });
+        // `FilePruner::try_new` decides whether a pruner is worthwhile (it needs
+        // a statistics struct, and either real column statistics or a dynamic
+        // filter that can prune via partition-value folding) and returns `None`
+        // otherwise. For a static predicate the pruner's tracker reports no
+        // changes, so it runs once and adds no ongoing cost.
+        let file_pruner = predicate.as_ref().and_then(|p| {
+            FilePruner::try_new(
+                Arc::clone(p),
+                &logical_file_schema,
+                &partitioned_file,
+                predicate_creation_errors.clone(),
+            )
+        });
 
         Ok(PreparedParquetOpen {
             partition_index: self.partition_index,
@@ -677,30 +676,21 @@ impl PreparedParquetOpen {
     /// Returns `None` if the file can be skipped completely.
     fn prune_file(mut self) -> Result<Option<Self>> {
         // Prune this file using the file level statistics and partition values.
-        // Since dynamic filters may have been updated since planning it is possible that we are able
-        // to prune files now that we couldn't prune at planning time.
-        // It is assumed that there is no point in doing pruning here if the predicate is not dynamic,
-        // as it would have been done at planning time.
-        // We'll also check this after every record batch we read,
-        // and if at some point we are able to prove we can prune the file using just the file level statistics
-        // we can end the stream early.
-        //
-        // Make a FilePruner only if there is either
-        // 1. a dynamic expr in the predicate
-        // 2. the file has file-level statistics.
-        //
-        // File-level statistics may prune the file without loading
-        // any row groups or metadata.
+        // Since dynamic filters may have been updated since planning it is
+        // possible that we are able to prune files now that we couldn't prune at
+        // planning time. The `FilePruner` (built when the predicate is dynamic or
+        // the file carries statistics) also watches any still-active dynamic
+        // filter, so the
+        // `EarlyStoppingStream` wrapping the scan can re-check after each batch
+        // and end the stream early once a tightened filter proves the file can
+        // be skipped.
         //
-        // Dynamic filters may prune the file after initial
-        // planning, as the dynamic filter is updated during
-        // execution.
-        //
-        // The case where there is a dynamic filter but no
-        // statistics corresponds to a dynamic filter that
-        // references partition columns. While rare, this is possible
-        // e.g. `select * from table order by partition_col limit
-        // 10` could hit this condition.
+        // File-level statistics may prune the file without loading any row
+        // groups or metadata. Partition column predicates are already folded to
+        // literals (see `replace_columns_with_literals` above), so a dynamic
+        // filter that references only partition columns can prune here too even
+        // when the file has no column statistics, e.g.
+        // `select * from t order by partition_col limit 10`.
         if let Some(file_pruner) = &mut self.file_pruner
             && file_pruner.should_prune()?
         {
@@ -1247,16 +1237,21 @@ impl RowGroupsPrunedParquetOpen {
         }
         .into_stream();
 
-        // Wrap the stream so a dynamic filter can stop the file scan early.
-        if let Some(file_pruner) = prepared.file_pruner {
-            Ok(EarlyStoppingStream::new(
-                stream,
-                file_pruner,
-                files_ranges_pruned_statistics,
-            )
-            .boxed())
-        } else {
-            Ok(stream)
+        // Wrap the stream so a dynamic filter can stop the file scan early, but
+        // only when the pruner is still watching a filter that can change
+        // mid-scan. For a static (or already-complete) predicate the up-front
+        // `prune_file` check already captured everything that can be pruned, so
+        // per-batch re-checking would only add overhead.
+        match prepared.file_pruner {
+            Some(file_pruner) if file_pruner.is_watching() => {
+                Ok(EarlyStoppingStream::new(
+                    stream,
+                    file_pruner,
+                    files_ranges_pruned_statistics,
+                )
+                .boxed())
+            }
+            _ => Ok(stream),
         }
     }
 }

datafusion/physical-expr-common/src/physical_expr.rs

@@ -870,6 +870,12 @@ pub fn snapshot_generation(expr: &Arc<dyn PhysicalExpr>) -> u64 {
 /// Check if the given `PhysicalExpr` is dynamic.
 /// Internally this calls [`snapshot_generation`] to check if the generation is non-zero,
 /// any dynamic `PhysicalExpr` should have a non-zero generation.
+#[deprecated(
+    since = "55.0.0",
+    note = "Downcast to `DynamicFilterPhysicalExpr`, or use \
+    `DynamicFilterTracking::classify(expr).contains_dynamic_filter()` from \
+    `datafusion_physical_expr`"
+)]
 pub fn is_dynamic_physical_expr(expr: &Arc<dyn PhysicalExpr>) -> bool {
     // If the generation is non-zero, then this `PhysicalExpr` is dynamic.
     snapshot_generation(expr) != 0

…-expr/src/expressions/dynamic_filters.rs …r/src/expressions/dynamic_filters/mod.rsdatafusion/physical-expr/src/expressions/dynamic_filters.rs renamed to datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs datafusion/physical-expr/src/expressions/dynamic_filters.rs renamed to datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs

@@ -29,6 +29,9 @@ use datafusion_common::{
 use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr_common::physical_expr::DynHash;
 
+mod tracker;
+pub use tracker::{DynamicFilterTracker, DynamicFilterTracking};
+
 /// State of a dynamic filter, tracking both updates and completion.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum FilterState {
@@ -326,6 +329,31 @@ impl DynamicFilterPhysicalExpr {
             .await;
     }
 
+    /// Returns `true` if this filter has been marked complete via
+    /// [`Self::mark_complete`] and will therefore never change again.
+    pub(crate) fn is_complete(&self) -> bool {
+        self.inner.read().is_complete
+    }
+
+    /// Subscribe to this filter's updates for cheap, synchronous change
+    /// detection.
+    ///
+    /// The returned [`DynamicFilterSubscription`] lets a consumer poll whether
+    /// the filter's expression has advanced since it last looked, without
+    /// re-walking a predicate tree or re-deriving a generation on every check.
+    /// This is the building block used by [`DynamicFilterTracker`] to watch
+    /// every dynamic filter inside a (possibly composite) predicate.
+    pub(crate) fn subscribe(&self) -> DynamicFilterSubscription {
+        let mut receiver = self.state_watch.subscribe();
+        // Mark the current state as already-seen so the first `observe()` only
+        // reports updates that happen *after* subscription.
+        let last_generation = receiver.borrow_and_update().generation();
+        DynamicFilterSubscription {
+            receiver,
+            last_generation,
+        }
+    }
+
     /// Check if this dynamic filter is being actively used by any consumers.
     ///
     /// Returns `true` if there are references beyond the producer (e.g., the HashJoinExec
@@ -522,6 +550,77 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
     }
 }
 
+/// The result of polling a [`DynamicFilterSubscription`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct DynamicFilterChange {
+    /// The filter's expression advanced since the previous observation.
+    pub(crate) changed: bool,
+    /// The filter has been marked complete; it will never change again and the
+    /// subscription can be dropped.
+    pub(crate) complete: bool,
+}
+
+/// A cheap, synchronous handle for observing updates to a single
+/// [`DynamicFilterPhysicalExpr`].
+///
+/// Obtained via [`DynamicFilterPhysicalExpr::subscribe`]. Steady-state polling
+/// via [`Self::observe`] is a single atomic load (the underlying
+/// [`tokio::sync::watch`] version counter); the lock is only taken when the
+/// filter has actually been updated.
+#[derive(Debug)]
+pub(crate) struct DynamicFilterSubscription {
+    receiver: watch::Receiver<FilterState>,
+    /// Last generation we reported as "seen". Used to distinguish a real
+    /// expression update from a bare [`DynamicFilterPhysicalExpr::mark_complete`]
+    /// (which re-broadcasts the current generation without changing the
+    /// expression).
+    last_generation: u64,
+}
+
+impl DynamicFilterSubscription {
+    /// Observe the latest state of the filter.
+    ///
+    /// Reports whether the filter's expression advanced since the previous call
+    /// and whether it has since been marked complete. Cheap when nothing has
+    /// changed: a single atomic comparison with no lock acquisition.
+    pub(crate) fn observe(&mut self) -> DynamicFilterChange {
+        match self.receiver.has_changed() {
+            Ok(true) => {
+                let state = *self.receiver.borrow_and_update();
+                let changed = state.generation() > self.last_generation;
+                if changed {
+                    self.last_generation = state.generation();
+                }
+                DynamicFilterChange {
+                    changed,
+                    complete: matches!(state, FilterState::Complete { .. }),
+                }
+            }
+            Ok(false) => DynamicFilterChange {
+                changed: false,
+                complete: false,
+            },
+            // The watch sender lives inside the predicate's
+            // `DynamicFilterPhysicalExpr`, which the owner of this subscription
+            // keeps alive, so observing a dropped sender signals a bug rather
+            // than normal completion. Flag it loudly in debug builds; in release
+            // degrade to "complete" (no further updates are possible) instead of
+            // silently masking it.
+            Err(_) => {
+                debug_assert!(
+                    false,
+                    "DynamicFilterSubscription observed a dropped watch sender; \
+                     the owning predicate should keep it alive"
+                );
+                DynamicFilterChange {
+                    changed: false,
+                    complete: true,
+                }
+            }
+        }
+    }
+}
+
 /// An atomic counter used to generate monotonic u64 ids.
 struct ExpressionIdAtomicCounter {
     inner: AtomicU64,