add ExecutionPlan::dynamic_filters() method

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -34,7 +34,11 @@ use datafusion::{
     scalar::ScalarValue,
 };
 use datafusion_catalog::memory::DataSourceExec;
-use datafusion_common::config::ConfigOptions;
+use datafusion_common::{
+    JoinType,
+    config::ConfigOptions,
+    tree_node::{TreeNode, TreeNodeRecursion},
+};
 use datafusion_datasource::{
     PartitionedFile, file_groups::FileGroup, file_scan_config::FileScanConfigBuilder,
 };
@@ -60,6 +64,7 @@ use datafusion_physical_plan::{
     coalesce_partitions::CoalescePartitionsExec,
     collect,
     filter::{FilterExec, FilterExecBuilder},
+    joins::{HashJoinExec, PartitionMode},
     projection::ProjectionExec,
     repartition::RepartitionExec,
     sorts::sort::SortExec,
@@ -3085,6 +3090,72 @@ async fn test_filter_with_projection_pushdown() {
     assert_batches_eq!(expected, &result);
 }
 
+#[test]
+fn test_discover_dynamic_expression_producers() {
+    fn producer_count(plan: &Arc<dyn ExecutionPlan>) -> usize {
+        let mut count = 0;
+        plan.apply(|node| {
+            count += node.dynamic_expressions().len();
+            Ok(TreeNodeRecursion::Continue)
+        })
+        .expect("plan traversal should succeed");
+        count
+    }
+
+    let build_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("b", DataType::Int32, false),
+    ]));
+    let build_scan = TestScanBuilder::new(Arc::clone(&build_schema))
+        .with_support(true)
+        .with_batches(vec![
+            record_batch!(("a", Utf8, ["foo", "bar"]), ("b", Int32, [1, 2])).unwrap(),
+        ])
+        .build();
+
+    let probe_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("c", DataType::Float64, false),
+    ]));
+    let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema))
+        .with_support(true)
+        .with_batches(vec![
+            record_batch!(
+                ("a", Utf8, ["foo", "bar", "baz", "qux"]),
+                ("c", Float64, [1.0, 2.0, 3.0, 4.0])
+            )
+            .unwrap(),
+        ])
+        .build();
+
+    let plan = Arc::new(
+        HashJoinExec::try_new(
+            build_scan,
+            probe_scan,
+            vec![(
+                col("a", &build_schema).unwrap(),
+                col("a", &probe_schema).unwrap(),
+            )],
+            None,
+            &JoinType::Inner,
+            None,
+            PartitionMode::CollectLeft,
+            datafusion_common::NullEquality::NullEqualsNothing,
+            false,
+        )
+        .unwrap(),
+    ) as Arc<dyn ExecutionPlan>;
+    assert_eq!(producer_count(&plan), 0);
+
+    let mut config = ConfigOptions::default();
+    config.optimizer.enable_dynamic_filter_pushdown = true;
+    config.execution.parquet.pushdown_filters = true;
+    let optimized_plan = FilterPushdown::new_post_optimization()
+        .optimize(plan, &config)
+        .unwrap();
+    assert_eq!(producer_count(&optimized_plan), 1);
+}
+
 // ==== Filter pushdown through SortExec tests ====
 
 /// FilterExec above a plain SortExec (no fetch) should be pushed below it.

datafusion/ffi/src/execution_plan.rs

@@ -33,6 +33,7 @@ use tokio::runtime::Handle;
 
 use crate::config::FFI_ConfigOptions;
 use crate::execution::FFI_TaskContext;
+use crate::physical_expr::FFI_PhysicalExpr;
 use crate::physical_expr::metrics::FFI_MetricsSet;
 use crate::plan_properties::FFI_PlanProperties;
 use crate::record_batch_stream::FFI_RecordBatchStream;
@@ -50,6 +51,9 @@ pub struct FFI_ExecutionPlan {
     /// Return a vector of children plans
     pub children: unsafe extern "C" fn(plan: &Self) -> SVec<FFI_ExecutionPlan>,
 
+    /// Return the dynamic expressions produced by this plan node.
+    pub dynamic_expressions: unsafe extern "C" fn(plan: &Self) -> SVec<FFI_PhysicalExpr>,
+
     pub with_new_children:
         unsafe extern "C" fn(plan: &Self, children: SVec<Self>) -> FFI_Result<Self>,
 
@@ -138,6 +142,16 @@ unsafe extern "C" fn children_fn_wrapper(
         .collect()
 }
 
+unsafe extern "C" fn dynamic_expressions_fn_wrapper(
+    plan: &FFI_ExecutionPlan,
+) -> SVec<FFI_PhysicalExpr> {
+    plan.inner()
+        .dynamic_expressions()
+        .into_iter()
+        .map(FFI_PhysicalExpr::from)
+        .collect()
+}
+
 unsafe extern "C" fn with_new_children_fn_wrapper(
     plan: &FFI_ExecutionPlan,
     children: SVec<FFI_ExecutionPlan>,
@@ -306,6 +320,7 @@ impl FFI_ExecutionPlan {
         Self {
             properties: properties_fn_wrapper,
             children: children_fn_wrapper,
+            dynamic_expressions: dynamic_expressions_fn_wrapper,
             with_new_children: with_new_children_fn_wrapper,
             name: name_fn_wrapper,
             execute: execute_fn_wrapper,
@@ -442,6 +457,15 @@ impl ExecutionPlan for ForeignExecutionPlan {
         }
     }
 
+    fn dynamic_expressions(
+        &self,
+    ) -> Vec<Arc<dyn datafusion_physical_plan::PhysicalExpr>> {
+        unsafe { (self.plan.dynamic_expressions)(&self.plan) }
+            .iter()
+            .map(<Arc<dyn datafusion_physical_plan::PhysicalExpr>>::from)
+            .collect()
+    }
+
     fn repartitioned(
         &self,
         target_partitions: usize,
@@ -474,15 +498,18 @@ impl ExecutionPlan for ForeignExecutionPlan {
 
 #[cfg(any(test, feature = "integration-tests"))]
 pub mod tests {
-    use datafusion_physical_plan::Partitioning;
+    #[cfg(test)]
+    use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
     use datafusion_physical_plan::execution_plan::{Boundedness, EmissionType};
+    use datafusion_physical_plan::{Partitioning, PhysicalExpr};
 
     use super::*;
 
     #[derive(Debug)]
     pub struct EmptyExec {
         props: Arc<PlanProperties>,
         children: Vec<Arc<dyn ExecutionPlan>>,
+        dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
         metrics: Option<MetricsSet>,
         statistics: Option<Statistics>,
     }
@@ -497,6 +524,7 @@ pub mod tests {
                     Boundedness::Bounded,
                 )),
                 children: Vec::default(),
+                dynamic_expressions: Vec::default(),
                 metrics: None,
                 statistics: None,
             }
@@ -511,6 +539,14 @@ pub mod tests {
             self.statistics = Some(statistics);
             self
         }
+
+        pub fn with_dynamic_expressions(
+            mut self,
+            dynamic_expressions: Vec<Arc<dyn PhysicalExpr>>,
+        ) -> Self {
+            self.dynamic_expressions = dynamic_expressions;
+            self
+        }
     }
 
     impl DisplayAs for EmptyExec {
@@ -543,6 +579,7 @@ pub mod tests {
             Ok(Arc::new(EmptyExec {
                 props: Arc::clone(&self.props),
                 children,
+                dynamic_expressions: self.dynamic_expressions.clone(),
                 metrics: self.metrics.clone(),
                 statistics: self.statistics.clone(),
             }))
@@ -556,6 +593,10 @@ pub mod tests {
             unimplemented!()
         }
 
+        fn dynamic_expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+            self.dynamic_expressions.iter().map(Arc::clone).collect()
+        }
+
         fn metrics(&self) -> Option<MetricsSet> {
             self.metrics.clone()
         }
@@ -600,6 +641,32 @@ pub mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ffi_execution_plan_dynamic_expressions() -> Result<()> {
+        let schema = Arc::new(arrow::datatypes::Schema::empty());
+        let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true)));
+        let expected_id = dynamic_filter
+            .expression_id()
+            .expect("dynamic filters always have an expression ID");
+        let expression: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as _;
+        let original_plan =
+            Arc::new(EmptyExec::new(schema).with_dynamic_expressions(vec![expression]));
+
+        let mut ffi_plan = FFI_ExecutionPlan::new(original_plan, None);
+        ffi_plan.library_marker_id = crate::mock_foreign_marker_id;
+        let foreign_plan: Arc<dyn ExecutionPlan> = (&ffi_plan).try_into()?;
+        foreign_plan.check_invariants(
+            datafusion_physical_plan::execution_plan::InvariantLevel::Always,
+        )?;
+
+        let produced = foreign_plan.dynamic_expressions();
+        assert_eq!(produced.len(), 1);
+        assert_eq!(produced[0].expression_id(), Some(expected_id));
+        drop(foreign_plan);
+        assert_eq!(produced[0].expression_id(), Some(expected_id));
+        Ok(())
+    }
+
     #[test]
     fn test_ffi_execution_plan_children() -> Result<()> {
         let schema = Arc::new(arrow::datatypes::Schema::new(vec![

datafusion/ffi/src/tests/mod.rs

@@ -28,6 +28,8 @@ use datafusion_common::stats::Precision;
 use datafusion_common::{ColumnStatistics, Statistics};
 use datafusion_common::{Result, ScalarValue};
 use datafusion_expr::{Expr, TableType};
+use datafusion_physical_expr::PhysicalExpr;
+use datafusion_physical_expr::expressions::{DynamicFilterPhysicalExpr, lit};
 use datafusion_physical_plan::ExecutionPlan;
 use sync_provider::create_sync_table_provider;
 use udf_udaf_udwf::{
@@ -107,6 +109,8 @@ pub struct ForeignLibraryModule {
 
     pub create_empty_exec: extern "C" fn() -> FFI_ExecutionPlan,
 
+    pub create_exec_with_dynamic_expressions: extern "C" fn() -> FFI_ExecutionPlan,
+
     pub create_exec_with_statistics: extern "C" fn() -> FFI_ExecutionPlan,
 
     pub create_table_with_statistics:
@@ -161,6 +165,15 @@ pub(crate) extern "C" fn create_empty_exec() -> FFI_ExecutionPlan {
     FFI_ExecutionPlan::new(plan, None)
 }
 
+pub(crate) extern "C" fn create_exec_with_dynamic_expressions() -> FFI_ExecutionPlan {
+    let schema = Arc::new(Schema::empty());
+    let expression: Arc<dyn PhysicalExpr> =
+        Arc::new(DynamicFilterPhysicalExpr::new(vec![], lit(true)));
+    let plan =
+        Arc::new(EmptyExec::new(schema).with_dynamic_expressions(vec![expression]));
+    FFI_ExecutionPlan::new(plan, None)
+}
+
 /// Returns canonical statistics used by both the producer and consumer sides of
 /// the integration tests so round-trips can be asserted without hard-coding
 /// the values in two places.
@@ -259,6 +272,7 @@ pub extern "C" fn datafusion_ffi_get_module() -> ForeignLibraryModule {
         create_rank_udwf: create_ffi_rank_func,
         create_extension_options: config::create_extension_options,
         create_empty_exec,
+        create_exec_with_dynamic_expressions,
         create_exec_with_statistics,
         create_table_with_statistics,
         create_physical_optimizer_rule:

datafusion/ffi/tests/ffi_execution_plan.rs

@@ -26,6 +26,7 @@ mod tests {
     use datafusion_ffi::execution_plan::{ExecutionPlanPrivateData, tests::EmptyExec};
     use datafusion_ffi::tests::utils::get_module;
     use datafusion_physical_plan::ExecutionPlan;
+    use datafusion_physical_plan::execution_plan::InvariantLevel;
     use std::sync::Arc;
 
     #[test]
@@ -63,6 +64,23 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_ffi_execution_plan_dynamic_expressions_cross_library()
+    -> Result<(), DataFusionError> {
+        let module = get_module()?;
+        let plan = (module.create_exec_with_dynamic_expressions)();
+        let plan: Arc<dyn ExecutionPlan> = (&plan).try_into()?;
+        assert!(plan.is::<ForeignExecutionPlan>());
+        plan.check_invariants(InvariantLevel::Always)?;
+
+        let produced = plan.dynamic_expressions();
+        assert_eq!(produced.len(), 1);
+        assert!(produced[0].expression_id().is_some());
+        drop(plan);
+        assert!(produced[0].expression_id().is_some());
+        Ok(())
+    }
+
     #[test]
     fn test_ffi_execution_plan_new_sets_runtimes_on_children()
     -> Result<(), DataFusionError> {

datafusion/physical-plan/src/aggregates/mod.rs

@@ -1960,6 +1960,16 @@ impl ExecutionPlan for AggregateExec {
         vec![&self.input]
     }
 
+    fn dynamic_expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
+        self.dynamic_filter
+            .iter()
+            .map(|dynamic_filter| {
+                Arc::<DynamicFilterPhysicalExpr>::clone(&dynamic_filter.filter)
+                    as Arc<dyn PhysicalExpr>
+            })
+            .collect()
+    }
+
     fn with_new_children(
         self: Arc<Self>,
         children: Vec<Arc<dyn ExecutionPlan>>,
@@ -7554,6 +7564,9 @@ mod tests {
             lit(false),
         ));
         let agg = agg.with_dynamic_filter_expr(Arc::clone(&new_df))?;
+        let produced = agg.dynamic_expressions();
+        assert_eq!(produced.len(), 1);
+        assert_eq!(produced[0].expression_id(), new_df.expression_id());
 
         // The aggregate's filter should now resolve to the new inner expression.
         let swapped = agg