refactor: drop the opener's is_dynamic gate and remove is_dynamic_physical_expr

Now that `FilePruner` watches dynamic filters through `DynamicFilterTracker`,
a non-dynamic predicate costs nothing beyond the one-shot prune: the tracker
reports no changes (so `should_prune` never rebuilds) and `is_watching()` is
false (so the scan is not wrapped in `EarlyStoppingStream`). The creation-time
"is this dynamic?" gate therefore no longer earns its keep.

Drop the `is_dynamic_physical_expr(p) || has_statistics()` filter and let
`FilePruner::try_new` (which already requires `statistics.is_some()`) plus the
self-regulating tracker decide. Behavior is identical for every real case:
files with column statistics already passed the gate via `has_statistics()`,
and dynamic partition-column pruning relies on a present (if all-`Absent`)
statistics struct, which `try_new` still accepts. The only difference is a rare,
harmless one-shot prune for static files whose stats struct is present but
entirely `Absent`.

With its sole caller gone, remove `is_dynamic_physical_expr`. `snapshot_generation`
itself is untouched (still used by the FFI vtable and proto roundtrip); its free
function is now unused in-workspace and can be retired in a later cleanup.

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,22 @@ 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(),
-                )
-            });
+        // Build a file pruner whenever the file carries statistics; `try_new`
+        // returns `None` when it does not. The pruner runs a one-shot prune
+        // before reading (catching dynamic filters that tightened since planning,
+        // or per-file statistics that weren't available then) and only keeps
+        // re-checking as the scan proceeds while it still watches an active
+        // dynamic filter. A static predicate's tracker reports no changes, so
+        // there is no ongoing re-pruning cost — which is why no separate
+        // "is this dynamic?" gate is needed here.
+        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 +679,20 @@ 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.
-        //
-        // Dynamic filters may prune the file after initial
-        // planning, as the dynamic filter is updated during
-        // execution.
+        // 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 whenever 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.
         //
-        // 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()?
         {

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

@@ -867,14 +867,6 @@ pub fn snapshot_generation(expr: &Arc<dyn PhysicalExpr>) -> u64 {
     generation
 }
 
-/// 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.
-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
-}
-
 /// Returns true if the expression is volatile, i.e. whether it can return different
 /// results when evaluated multiple times with the same input.
 ///

datafusion/physical-expr/src/dynamic_filter_tracker.rs

@@ -93,12 +93,9 @@ impl DynamicFilterTracking {
         }
     }
 
-    /// `true` if the predicate contains any dynamic filter (complete or not).
-    ///
-    /// This matches the question previously answered by
-    /// [`is_dynamic_physical_expr`](datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr):
-    /// whether the predicate's value may differ from what was known at planning
-    /// time, and is therefore worth re-evaluating at least once.
+    /// `true` if the predicate contains any dynamic filter (complete or not),
+    /// i.e. its value may differ from what was known at planning time and is
+    /// therefore worth re-evaluating at least once.
     pub fn contains_dynamic_filter(&self) -> bool {
         !matches!(self, DynamicFilterTracking::Static)
     }