physical-expr-common: deprecate PhysicalExpr::snapshot method

Which issue does this PR close?

See #21807

Rationale for this change

`PhysicalExpr::snapshot` isn't required because it only applies to
dynamic filters. The only place that its used is in
`pruning_predicate.rs`. Since there's only one use case, we can just
downcast to `DynamicFilterPhysicalExpr` and call `current()`.

What changes are included in this PR?

This change marks `PhysicalExpr::snapshot`, `snapshot_physical_expr`, and `snapshot_physical_expr_opt` as
deprecated. Callers should downcast to `DynamicFilterPhysicalExpr` and call `current()`.

Are these changes tested?

Yes, the existing tests should cover the exact same functionality.
Filter pushdown still workers with dynamic filters; we use `current()`
instead of `snapshot()` to capture the expression.

Are there any user-facing changes?

`PhysicalExpr::snapshot`, `snapshot_physical_expr`, and `snapshot_physical_expr_opt`
are deprecated and should not be used.

Author: jayshrivastava

datafusion/ffi/src/physical_expr/mod.rs

@@ -939,6 +939,8 @@ mod tests {
         assert_eq!(left, right);
     }
 
+    /// The snapshot API is deprecated.
+    #[allow(deprecated)]
     #[test]
     fn ffi_physical_expr_snapshots() -> Result<(), DataFusionError> {
         let (original, foreign_expr) = create_test_expr();

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

@@ -355,48 +355,10 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
     /// See the [`fmt_sql`] function for an example of printing `PhysicalExpr`s as SQL.
     fn fmt_sql(&self, f: &mut Formatter<'_>) -> fmt::Result;
 
-    /// Take a snapshot of this `PhysicalExpr`, if it is dynamic.
-    ///
-    /// "Dynamic" in this case means containing references to structures that may change
-    /// during plan execution, such as hash tables.
-    ///
-    /// This method is used to capture the current state of `PhysicalExpr`s that may contain
-    /// dynamic references to other operators in order to serialize it over the wire
-    /// or treat it via downcast matching.
-    ///
-    /// You should not call this method directly as it does not handle recursion.
-    /// Instead use [`snapshot_physical_expr`] to handle recursion and capture the
-    /// full state of the `PhysicalExpr`.
-    ///
-    /// This is expected to return "simple" expressions that do not have mutable state
-    /// and are composed of DataFusion's built-in `PhysicalExpr` implementations.
-    /// Callers however should *not* assume anything about the returned expressions
-    /// since callers and implementers may not agree on what "simple" or "built-in"
-    /// means.
-    /// In other words, if you need to serialize a `PhysicalExpr` across the wire
-    /// you should call this method and then try to serialize the result,
-    /// but you should handle unknown or unexpected `PhysicalExpr` implementations gracefully
-    /// just as if you had not called this method at all.
-    ///
-    /// In particular, consider:
-    /// * A `PhysicalExpr` that references the current state of a `datafusion::physical_plan::TopK`
-    ///   that is involved in a query with `SELECT * FROM t1 ORDER BY a LIMIT 10`.
-    ///   This function may return something like `a >= 12`.
-    /// * A `PhysicalExpr` that references the current state of a `datafusion::physical_plan::joins::HashJoinExec`
-    ///   from a query such as `SELECT * FROM t1 JOIN t2 ON t1.a = t2.b`.
-    ///   This function may return something like `t2.b IN (1, 5, 7)`.
-    ///
-    /// A system or function that can only deal with a hardcoded set of `PhysicalExpr` implementations
-    /// or needs to serialize this state to bytes may not be able to handle these dynamic references.
-    /// In such cases, we should return a simplified version of the `PhysicalExpr` that does not
-    /// contain these dynamic references.
-    ///
-    /// Systems that implement remote execution of plans, e.g. serialize a portion of the query plan
-    /// and send it across the wire to a remote executor may want to call this method after
-    /// every batch on the source side and broadcast / update the current snapshot to the remote executor.
-    ///
-    /// Note for implementers: this method should *not* handle recursion.
-    /// Recursion is handled in [`snapshot_physical_expr`].
+    #[deprecated(
+        since = "54.0.0",
+        note = "downcast to `DynamicFilterPhysicalExpr` and call `current()` instead"
+    )]
     fn snapshot(&self) -> Result<Option<Arc<dyn PhysicalExpr>>> {
         // By default, we return None to indicate that this PhysicalExpr does not
         // have any dynamic references or state.
@@ -616,38 +578,22 @@ pub fn fmt_sql(expr: &dyn PhysicalExpr) -> impl Display + '_ {
     Wrapper { expr }
 }
 
-/// Take a snapshot of the given `PhysicalExpr` if it is dynamic.
-///
-/// Take a snapshot of this `PhysicalExpr` if it is dynamic.
-/// This is used to capture the current state of `PhysicalExpr`s that may contain
-/// dynamic references to other operators in order to serialize it over the wire
-/// or treat it via downcast matching.
-///
-/// See the documentation of [`PhysicalExpr::snapshot`] for more details.
-///
-/// # Returns
-///
-/// Returns a snapshot of the `PhysicalExpr` if it is dynamic, otherwise
-/// returns itself.
+#[deprecated(
+    since = "54.0.0",
+    note = "downcast to `DynamicFilterPhysicalExpr` and call `current()` instead"
+)]
+#[allow(deprecated)]
 pub fn snapshot_physical_expr(
     expr: Arc<dyn PhysicalExpr>,
 ) -> Result<Arc<dyn PhysicalExpr>> {
     snapshot_physical_expr_opt(expr).data()
 }
 
-/// Take a snapshot of the given `PhysicalExpr` if it is dynamic.
-///
-/// Take a snapshot of this `PhysicalExpr` if it is dynamic.
-/// This is used to capture the current state of `PhysicalExpr`s that may contain
-/// dynamic references to other operators in order to serialize it over the wire
-/// or treat it via downcast matching.
-///
-/// See the documentation of [`PhysicalExpr::snapshot`] for more details.
-///
-/// # Returns
-///
-/// Returns a `[`Transformed`] indicating whether a snapshot was taken,
-/// along with the resulting `PhysicalExpr`.
+#[deprecated(
+    since = "54.0.0",
+    note = "downcast to `DynamicFilterPhysicalExpr` and call `current()` instead"
+)]
+#[allow(deprecated)]
 pub fn snapshot_physical_expr_opt(
     expr: Arc<dyn PhysicalExpr>,
 ) -> Result<Transformed<Arc<dyn PhysicalExpr>>> {

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -611,14 +611,20 @@ mod test {
             &filter_schema_1,
         )
         .unwrap();
-        let snap = dynamic_filter_1.snapshot().unwrap().unwrap();
+        let dynamic_filter_1_df = dynamic_filter_1
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .expect("Expected DynamicFilterPhysicalExpr");
+        let snap = dynamic_filter_1_df.current().unwrap();
         insta::assert_snapshot!(format!("{snap:?}"), @r#"BinaryExpr { left: Column { name: "a", index: 0 }, op: Eq, right: Literal { value: Int32(42), field: Field { name: "lit", data_type: Int32 } }, fail_on_overflow: false }"#);
         let dynamic_filter_2 = reassign_expr_columns(
             Arc::clone(&dynamic_filter) as Arc<dyn PhysicalExpr>,
             &filter_schema_2,
         )
         .unwrap();
-        let snap = dynamic_filter_2.snapshot().unwrap().unwrap();
+        let dynamic_filter_2_df = dynamic_filter_2
+            .downcast_ref::<DynamicFilterPhysicalExpr>()
+            .expect("Expected DynamicFilterPhysicalExpr");
+        let snap = dynamic_filter_2_df.current().unwrap();
         insta::assert_snapshot!(format!("{snap:?}"), @r#"BinaryExpr { left: Column { name: "a", index: 1 }, op: Eq, right: Literal { value: Int32(42), field: Field { name: "lit", data_type: Int32 } }, fail_on_overflow: false }"#);
         // Both filters allow evaluating the same expression
         let batch_1 = RecordBatch::try_new(
@@ -683,6 +689,7 @@ mod test {
         assert!(arr_1.eq(&expected));
     }
 
+    #[allow(deprecated)]
     #[test]
     fn test_snapshot() {
         let expr = lit(42) as Arc<dyn PhysicalExpr>;

datafusion/physical-optimizer/src/filter_pushdown.rs

@@ -326,9 +326,10 @@ use itertools::{Itertools, izip};
 /// building a specialized [`PhysicalExpr`] that can be evaluated at runtime
 /// and internally maintains a reference to the hash table or other state.
 ///
-/// To make working with these sorts of dynamic filters more tractable we have the method [`PhysicalExpr::snapshot`]
-/// which attempts to simplify a dynamic filter into a "basic" non-dynamic filter.
-/// For a join this could mean converting it to an `InList` filter or a min/max filter for example.
+/// To make working with these sorts of dynamic filters more tractable callers can downcast a
+/// [`PhysicalExpr`] to [`DynamicFilterPhysicalExpr`] and call `current()` to get a snapshot
+/// of the expression (ie. a a static filter expression). For a join this could mean converting
+/// it to an `InList` filter or a min/max filter for example.
 /// See `datafusion/physical-plan/src/dynamic_filters.rs` for more details.
 ///
 /// # Example: Push TopK filters into Scans
@@ -376,7 +377,7 @@ use itertools::{Itertools, izip};
 /// <https://github.com/apache/datafusion/issues/15037>
 ///
 /// [`PhysicalExpr`]: datafusion_physical_plan::PhysicalExpr
-/// [`PhysicalExpr::snapshot`]: datafusion_physical_plan::PhysicalExpr::snapshot
+/// [`DynamicFilterPhysicalExpr`]: datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr
 /// [`FilterExec`]: datafusion_physical_plan::filter::FilterExec
 /// [`ProjectionExec`]: datafusion_physical_plan::projection::ProjectionExec
 /// [`AggregateExec`]: datafusion_physical_plan::aggregates::AggregateExec

datafusion/pruning/src/pruning_predicate.rs

@@ -42,9 +42,9 @@ use datafusion_common::{
     tree_node::{Transformed, TreeNode},
 };
 use datafusion_expr_common::operator::Operator;
+use datafusion_physical_expr::expressions::DynamicFilterPhysicalExpr;
 use datafusion_physical_expr::utils::{Guarantee, LiteralGuarantee};
 use datafusion_physical_expr::{PhysicalExprRef, expressions as phys_expr};
-use datafusion_physical_expr_common::physical_expr::snapshot_physical_expr_opt;
 use datafusion_physical_plan::{ColumnarValue, PhysicalExpr};
 
 /// Used to prove that arbitrary predicates (boolean expression) can not
@@ -456,22 +456,28 @@ impl PruningPredicate {
     /// details.
     ///
     /// Note that `PruningPredicate` does not attempt to normalize or simplify
-    /// the input expression unless calling [`snapshot_physical_expr_opt`]
-    /// returns a new expression.
+    /// the input expression unless any [`DynamicFilterPhysicalExpr`] nodes
+    /// are replaced with their current value.
     /// It is recommended that you pass the expressions through [`PhysicalExprSimplifier`]
     /// before calling this method to make sure the expressions can be used for pruning.
     pub fn try_new(mut expr: Arc<dyn PhysicalExpr>, schema: SchemaRef) -> Result<Self> {
-        // Get a (simpler) snapshot of the physical expr here to use with `PruningPredicate`.
-        // In particular this unravels any `DynamicFilterPhysicalExpr`s by snapshotting them
-        // so that PruningPredicate can work with a static expression.
-        let tf = snapshot_physical_expr_opt(expr)?;
+        // Replace any `DynamicFilterPhysicalExpr` nodes with their current value so
+        // that `PruningPredicate` can work with a static expression.
+        let tf = expr.transform_up(|e| {
+            if let Some(df) = e.downcast_ref::<DynamicFilterPhysicalExpr>() {
+                Ok(Transformed::yes(df.current()?))
+            } else {
+                Ok(Transformed::no(e))
+            }
+        })?;
         if tf.transformed {
             // If we had an expression such as Dynamic(part_col < 5 and col < 10)
             // (this could come from something like `select * from t order by part_col, col, limit 10`)
-            // after snapshotting and because `DynamicFilterPhysicalExpr` applies child replacements to its
-            // children after snapshotting and previously `replace_columns_with_literals` may have been called with partition values
-            // the expression we have now is `8 < 5 and col < 10`.
-            // Thus we need as simplifier pass to get `false and col < 10` => `false` here.
+            // after replacing the dynamic filter with its current value and because
+            // `DynamicFilterPhysicalExpr` applies child replacements to its children,
+            // and previously `replace_columns_with_literals` may have been called with
+            // partition values, the expression we have now is `8 < 5 and col < 10`.
+            // Thus we need a simplifier pass to get `false and col < 10` => `false` here.
             let simplifier = PhysicalExprSimplifier::new(&schema);
             expr = simplifier.simplify(tf.data)?;
         } else {