refactor(pruning): drive FilePruner via DynamicFilterTracker

Replace `FilePruner`'s hand-rolled `snapshot_generation()` polling (store last
`u64`, recompute + diff on every `should_prune`) with a `DynamicFilterTracking`
classification computed once at construction. The pruner rebuilds the pruning
predicate on the first check and thereafter only when a watched dynamic filter
has actually moved.

This also lets the Parquet opener skip wrapping the scan in `EarlyStoppingStream`
when the predicate is static or its dynamic filters are already complete: the
up-front `prune_file` check already captured everything such a predicate can
prune, so per-batch re-checking was pure overhead.

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

Author: adriangb

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

@@ -1247,16 +1247,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/pruning/src/file_pruner.rs

@@ -22,7 +22,8 @@ use std::sync::Arc;
 use arrow::datatypes::{FieldRef, SchemaRef};
 use datafusion_common::{Result, internal_datafusion_err, pruning::PrunableStatistics};
 use datafusion_datasource::PartitionedFile;
-use datafusion_physical_expr_common::physical_expr::{PhysicalExpr, snapshot_generation};
+use datafusion_physical_expr::DynamicFilterTracking;
+use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use datafusion_physical_plan::metrics::Count;
 use log::debug;
 
@@ -34,8 +35,14 @@ use crate::build_pruning_predicate;
 /// which substitutes partition column references with their literal values before
 /// the predicate reaches this pruner.
 pub struct FilePruner {
-    predicate_generation: Option<u64>,
     predicate: Arc<dyn PhysicalExpr>,
+    /// Tracks the dynamic filters inside `predicate` so we only rebuild the
+    /// pruning predicate when one of them has actually moved.
+    tracking: DynamicFilterTracking,
+    /// Whether [`Self::should_prune`] has built+evaluated the pruning predicate
+    /// at least once. The first check always runs; subsequent checks only run
+    /// when a watched dynamic filter changed.
+    checked_once: bool,
     /// Schema used for pruning (the logical file schema).
     file_schema: SchemaRef,
     file_stats_pruning: PrunableStatistics,
@@ -80,31 +87,44 @@ impl FilePruner {
         let file_stats = partitioned_file.statistics.as_ref()?;
         let file_stats_pruning =
             PrunableStatistics::new(vec![file_stats.clone()], Arc::clone(file_schema));
+        let tracking = DynamicFilterTracking::classify(&predicate);
         Some(Self {
-            predicate_generation: None,
             predicate,
+            tracking,
+            checked_once: false,
             file_schema: Arc::clone(file_schema),
             file_stats_pruning,
             predicate_creation_errors,
         })
     }
 
+    /// Returns `true` if this pruner watches a dynamic filter that can still
+    /// change, meaning [`Self::should_prune`] is worth re-checking as the scan
+    /// progresses. When `false`, the predicate is effectively static for the
+    /// remainder of the scan and the caller can avoid wrapping the stream in a
+    /// per-batch re-pruning adapter.
+    pub fn is_watching(&self) -> bool {
+        matches!(self.tracking, DynamicFilterTracking::Watching(_))
+    }
+
     pub fn should_prune(&mut self) -> Result<bool> {
-        // Check if the predicate has changed since last invocation by tracking
-        // its "generation". Dynamic filter expressions can change their values
-        // during query execution, so we use generation tracking to detect when
-        // the predicate has been updated and needs to be rebuilt.
+        // Building the pruning predicate is expensive (it involves expression
+        // analysis), so we only do it on the first check and whenever a dynamic
+        // filter inside the predicate has actually moved.
         //
-        // If the generation hasn't changed, we can skip rebuilding the pruning
-        // predicate, which is an expensive operation involving expression analysis.
-        let new_generation = snapshot_generation(&self.predicate);
-        if let Some(current_generation) = self.predicate_generation.as_mut() {
-            if *current_generation == new_generation {
-                return Ok(false);
-            }
-            *current_generation = new_generation;
+        // Dynamic filter expressions can change their values during query
+        // execution; `DynamicFilterTracking` watches the still-incomplete
+        // filters and reports a change at most once per update. A purely static
+        // predicate (or one whose dynamic filters have all completed) is checked
+        // exactly once.
+        let should_build = if self.checked_once {
+            self.tracking.watcher().is_some_and(|w| w.changed())
         } else {
-            self.predicate_generation = Some(new_generation);
+            self.checked_once = true;
+            true
+        };
+        if !should_build {
+            return Ok(false);
         }
         let pruning_predicate = build_pruning_predicate(
             Arc::clone(&self.predicate),