refactor: fold the FilePruner creation gate into try_new; deprecate is_dynamic_physical_expr

The Parquet opener gated `FilePruner` creation on
`is_dynamic_physical_expr(p) || has_statistics()`, where `is_dynamic_physical_expr`
answered "is it dynamic?" via `snapshot_generation(p) != 0`.

`FilePruner` already classifies its predicate once (`DynamicFilterTracking`) to
drive the change tracker, so the same classification can answer the gate. Move
the decision into `FilePruner::try_new`: it returns `None` when the file has no
statistics struct, or when the predicate is purely static and the file has no
usable column statistics (planning already did everything such a pruner could).
A dynamic predicate is still accepted (it may prune via partition-value folding
even without column statistics). The opener now just calls `try_new`.

With its last internal caller gone, `is_dynamic_physical_expr` is deprecated
(since 55.0.0) rather than removed; downstream users should downcast to
`DynamicFilterPhysicalExpr` or use `DynamicFilterTracking`. See the 55.0.0
upgrade guide. `snapshot_generation` itself is unchanged (FFI vtable + proto).

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.
-        //
-        // 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 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.
         //
-        // 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

@@ -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

datafusion/pruning/src/file_pruner.rs

@@ -76,18 +76,35 @@ impl FilePruner {
         })
     }
 
-    /// Create a new file pruner if statistics are available.
-    /// Returns None if this file does not have statistics.
+    /// Create a file pruner for this file, or `None` when pruning it cannot
+    /// help.
+    ///
+    /// Returns `None` when the file has no statistics struct to evaluate a
+    /// pruning predicate against, or when the predicate is purely static and the
+    /// file has no usable column statistics — in that case planning already did
+    /// everything such a pruner could. A predicate carrying a dynamic filter is
+    /// always accepted (given a statistics struct), since it may prune via
+    /// partition-value folding even without column statistics.
     pub fn try_new(
         predicate: Arc<dyn PhysicalExpr>,
         file_schema: &SchemaRef,
         partitioned_file: &PartitionedFile,
         predicate_creation_errors: Count,
     ) -> Option<Self> {
+        // A pruning predicate is evaluated against a statistics struct, so one
+        // must exist (its columns may all be `Absent`).
         let file_stats = partitioned_file.statistics.as_ref()?;
+        let tracking = DynamicFilterTracking::classify(&predicate);
+        // Only build a pruner when it could prune something planning didn't
+        // already: the file has real column statistics, or the predicate carries
+        // a dynamic filter (whose value, or folded partition columns, can prune
+        // even without column statistics). For a purely static predicate with no
+        // usable stats there is nothing to gain.
+        if !partitioned_file.has_statistics() && !tracking.contains_dynamic_filter() {
+            return None;
+        }
         let file_stats_pruning =
             PrunableStatistics::new(vec![file_stats.clone()], Arc::clone(file_schema));
-        let tracking = DynamicFilterTracking::classify(&predicate);
         Some(Self {
             predicate,
             tracking,

docs/source/library-user-guide/upgrading/55.0.0.md

@@ -0,0 +1,72 @@
+<!---
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing,
+  software distributed under the License is distributed on an
+  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+  KIND, either express or implied.  See the License for the
+  specific language governing permissions and limitations
+  under the License.
+-->
+
+# Upgrade Guides
+
+## DataFusion 55.0.0
+
+**Note:** DataFusion `55.0.0` has not been released yet. The information provided
+in this section pertains to features and changes that have already been merged
+to the main branch and are awaiting release in this version.
+
+### `is_dynamic_physical_expr` is deprecated
+
+`datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr` is
+deprecated. It was a thin wrapper over `snapshot_generation(expr) != 0` used to
+ask "does this predicate contain a dynamic filter?".
+
+Prefer asking the question directly against the concrete type. For a one-off
+check, downcast to `DynamicFilterPhysicalExpr`:
+
+```rust
+use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
+use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
+
+let mut is_dynamic = false;
+predicate.apply(|e| {
+    if e.downcast_ref::<DynamicFilterPhysicalExpr>().is_some() {
+        is_dynamic = true;
+        Ok(TreeNodeRecursion::Stop)
+    } else {
+        Ok(TreeNodeRecursion::Continue)
+    }
+})?;
+```
+
+If you also need to know whether the dynamic filters can still change (and to be
+notified when they do), use the new `DynamicFilterTracking` /
+`DynamicFilterTracker` API in `datafusion_physical_expr`:
+
+```rust
+use datafusion_physical_expr::DynamicFilterTracking;
+
+let tracking = DynamicFilterTracking::classify(&predicate);
+if tracking.contains_dynamic_filter() {
+    // worth re-evaluating the predicate at runtime
+}
+```
+
+### `FilePruner::try_new` no longer builds a pruner for static predicates without statistics
+
+`datafusion_pruning::FilePruner::try_new` now returns `None` when the predicate
+is purely static _and_ the file carries no usable column statistics, because
+such a pruner can never prune anything beyond what planning already did.
+Previously it returned `Some` whenever a statistics struct was present (the
+"is this worth pruning?" decision lived in the Parquet opener). Files with column
+statistics, and predicates that carry a dynamic filter, are unaffected.

docs/source/library-user-guide/upgrading/index.rst

@@ -21,6 +21,7 @@ Upgrade Guides
 .. toctree::
    :maxdepth: 1
 
+   DataFusion 55.0.0 <55.0.0>
    DataFusion 54.0.0 <54.0.0>
    DataFusion 53.0.0 <53.0.0>
    DataFusion 52.0.0 <52.0.0>