feat: accept and store expressions (including dynamic filters) in Iceberg scan nodes

codegen/src/main.rs

@@ -73,6 +73,10 @@ fn generate_iceberg(repo_root: &Path) -> Result<(), Box<dyn std::error::Error>>
             ".iceberg.DataFusionPartitioning",
             "::datafusion_proto::protobuf::Partitioning",
         )
+        .extern_path(
+            ".iceberg.DataFusionPhysicalExprNode",
+            "::datafusion_proto::protobuf::PhysicalExprNode",
+        )
         .compile_protos(&[proto_file], &[proto_dir])?;
 
     Ok(())

datafusion/distributed-iceberg/README.md

@@ -90,6 +90,14 @@ statistics and task-size estimates. Selected-file column statistics remain a
 separate change. Task-local eager variants and decoded plans report unknown
 statistics rather than applying whole-snapshot aggregates to a subset.
 
+Both scans keep the filters DataFusion pushes into them, including the dynamic
+filters that joins, top-N sorts and aggregates compute while the query runs. Scans
+return `PushedDown::No` for these filters, so DataFusion still checks every row. A join
+or aggregate only computes a dynamic filter when a scan below it is a consumer of that
+filter. Enable `iceberg.dynamic_filtering_enabled` to make Iceberg scans consumers of the
+dynamic filters pushed into them. With it on, `EXPLAIN` shows them on the scan as
+`dynamic_filters=`. Scans don't skip data with these filters yet.
+
 ```bash
 cargo test -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools
 ```

datafusion/distributed-iceberg/src/common/scan_config.rs

@@ -1,10 +1,16 @@
 use crate::{IcebergConfig, common::df_err};
 use datafusion::arrow::datatypes::SchemaRef;
 use datafusion::common::stats::Precision;
+use datafusion::common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
 use datafusion::common::{ColumnStatistics, Statistics};
 use datafusion::error::Result;
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
-use datafusion::physical_expr::Partitioning;
+use datafusion::physical_expr::expressions::{InListExpr, lit};
+use datafusion::physical_expr::{
+    DynamicFilterTracking, Partitioning, PhysicalExpr, conjunction, split_conjunction,
+};
+use datafusion::physical_expr_common::physical_expr::snapshot_physical_expr;
+use datafusion::physical_plan::apply_expression_roots;
 use datafusion::physical_plan::limit::LimitStream;
 use datafusion::physical_plan::metrics::{
     BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory, MetricType,
@@ -30,6 +36,8 @@ pub(crate) struct ScanConfig {
     pub(crate) metrics: ExecutionPlanMetricsSet,
     pub(crate) iceberg_file_io: FileIO,
     pub(crate) iceberg_runtime: iceberg::Runtime,
+    pub(crate) filters: Option<Arc<dyn PhysicalExpr>>,
+    pub(crate) dynamic_filtering_enabled: bool,
     // Clones share the delete loader across this source's local partitions.
     reader: ArrowReader,
 }
@@ -57,10 +65,77 @@ impl ScanConfig {
             metrics: Default::default(),
             iceberg_file_io,
             iceberg_runtime,
+            filters: None,
+            dynamic_filtering_enabled: config.dynamic_filtering_enabled,
             reader,
         }
     }
 
+    /// ANDs `filters` onto the scan's filters.
+    pub(crate) fn push_filters(&mut self, filters: Vec<Arc<dyn PhysicalExpr>>) {
+        self.filters = Some(conjunction(self.filters.take().into_iter().chain(filters)));
+    }
+
+    /// Visits the scan's filters when dynamic filtering is enabled.
+    pub(crate) fn apply_expressions(
+        &self,
+        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+    ) -> Result<TreeNodeRecursion> {
+        if !self.dynamic_filtering_enabled {
+            return Ok(TreeNodeRecursion::Continue);
+        }
+        apply_expression_roots(self.filters.iter(), f)
+    }
+
+    /// Writes the scan's dynamic filters into `f` when dynamic filtering is enabled. A join filter
+    /// can list thousands of keys, so each IN list prints as its count.
+    pub(crate) fn fmt_dynamic_filters(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        if !self.dynamic_filtering_enabled {
+            return Ok(());
+        }
+        let Some(filters) = &self.filters else {
+            return Ok(());
+        };
+        let dynamic_filters: Vec<_> = split_conjunction(filters)
+            .into_iter()
+            .filter(|filter| DynamicFilterTracking::classify(filter).contains_dynamic_filter())
+            .cloned()
+            .collect();
+        if dynamic_filters.is_empty() {
+            return Ok(());
+        }
+        write!(f, ", dynamic_filters=")?;
+        for (i, filter) in dynamic_filters.into_iter().enumerate() {
+            if i > 0 {
+                write!(f, " AND ")?;
+            }
+            // A dynamic filter is at generation 1 until it gets its first values.
+            if filter.snapshot_generation() == 1 {
+                write!(f, "DynamicFilter [ empty ]")?;
+                continue;
+            }
+            let current = snapshot_physical_expr(filter)
+                .and_then(|current| {
+                    current.transform_up(|expr| match expr.downcast_ref::<InListExpr>() {
+                        Some(in_list) => {
+                            let not = if in_list.negated() { "NOT " } else { "" };
+                            let text = format!(
+                                "{} {not}IN ([{} elements])",
+                                in_list.expr(),
+                                in_list.len()
+                            );
+                            Ok(Transformed::yes(lit(text)))
+                        }
+                        None => Ok(Transformed::no(expr)),
+                    })
+                })
+                .map_err(|_| fmt::Error)?
+                .data;
+            write!(f, "DynamicFilter [ {current} ]")?;
+        }
+        Ok(())
+    }
+
     pub(crate) fn open(
         &self,
         tasks: FileScanTaskStream,
@@ -109,6 +184,8 @@ impl fmt::Debug for ScanConfig {
             .field("metrics", &self.metrics)
             .field("iceberg_file_io", &self.iceberg_file_io)
             .field("iceberg_runtime", &self.iceberg_runtime)
+            .field("filters", &self.filters)
+            .field("dynamic_filtering_enabled", &self.dynamic_filtering_enabled)
             .finish_non_exhaustive()
     }
 }

datafusion/distributed-iceberg/src/config.rs

@@ -18,6 +18,10 @@ extensions_options! {
         pub column_stats_enabled: bool, default = false
         /// Discover and retain file tasks during physical planning instead of execution.
         pub eager_file_planning: bool, default = false
+        /// (Experimental) Whether Iceberg scans are consumers of the dynamic filters pushed
+        /// into them, so joins and aggregates compute those filters. Scans don't skip data
+        /// with them yet.
+        pub dynamic_filtering_enabled: bool, default = false
     }
 }
 

datafusion/distributed-iceberg/src/eager/codec.rs

@@ -95,16 +95,17 @@ impl PhysicalExtensionCodec for EagerIcebergCodec {
                     .collect::<Result<Vec<_>>>()
             })
             .collect::<Result<Vec<_>>>()?;
-        Ok(DataSourceExec::from_data_source(
-            EagerIcebergDataSource::from_file_groups(
-                scan.schema,
-                groups,
-                scan.iceberg_file_io,
-                scan.iceberg_runtime,
-                scan.fetch,
-                ctx,
-            )?,
-        ))
+        let mut source = EagerIcebergDataSource::from_file_groups(
+            scan.schema,
+            groups,
+            scan.iceberg_file_io,
+            scan.iceberg_runtime,
+            scan.fetch,
+            ctx,
+        )?;
+        source.scan.filters = scan.filters;
+        source.scan.dynamic_filtering_enabled = scan.dynamic_filtering_enabled;
+        Ok(DataSourceExec::from_data_source(source))
     }
 
     fn try_encode(
@@ -175,14 +176,22 @@ impl EagerIcebergCodec {
             proto.custom_file_io.as_deref(),
             ctx,
         )?;
-        Ok(ScanConfig::new(
+        let filters = proto
+            .filters
+            .as_ref()
+            .map(|filters| proto_converter.proto_to_physical_expr(filters, &schema, &decode_ctx))
+            .transpose()?;
+        let mut scan = ScanConfig::new(
             schema,
             partitioning,
             fetch,
             iceberg_file_io,
             self.iceberg_runtime.clone(),
             ctx,
-        ))
+        );
+        scan.filters = filters;
+        scan.dynamic_filtering_enabled = proto.dynamic_filtering_enabled;
+        Ok(scan)
     }
 
     fn encode_config(
@@ -203,6 +212,13 @@ impl EagerIcebergCodec {
             fetch: source.fetch().map(|value| value as u64),
             storage_properties: file_io.properties,
             custom_file_io: file_io.custom,
+            filters: source
+                .scan
+                .filters
+                .as_ref()
+                .map(|filters| proto_converter.physical_expr_to_proto(filters, self))
+                .transpose()?,
+            dynamic_filtering_enabled: source.scan.dynamic_filtering_enabled,
         })
     }
 }

datafusion/distributed-iceberg/src/eager/data_source.rs

@@ -3,12 +3,14 @@ use crate::eager::planned_files::PlannedFiles;
 use datafusion::arrow::datatypes::SchemaRef;
 use datafusion::common::tree_node::TreeNodeRecursion;
 use datafusion::common::{Statistics, exec_err};
+use datafusion::config::ConfigOptions;
 use datafusion::datasource::source::DataSource;
 use datafusion::error::Result;
 use datafusion::execution::{SendableRecordBatchStream, TaskContext};
 use datafusion::physical_expr::projection::ProjectionExprs;
 use datafusion::physical_expr::{EquivalenceProperties, Partitioning, PhysicalExpr};
 use datafusion::physical_plan::DisplayFormatType;
+use datafusion::physical_plan::filter_pushdown::{FilterPushdownPropagation, PushedDown};
 use datafusion::physical_plan::metrics::ExecutionPlanMetricsSet;
 use futures::StreamExt;
 use iceberg::io::FileIO;
@@ -144,9 +146,9 @@ impl EagerIcebergDataSource {
 impl DataSource for EagerIcebergDataSource {
     fn apply_expressions(
         &self,
-        _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+        f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
     ) -> Result<TreeNodeRecursion> {
-        Ok(TreeNodeRecursion::Continue)
+        self.scan.apply_expressions(f)
     }
     fn open(
         &self,
@@ -171,7 +173,7 @@ impl DataSource for EagerIcebergDataSource {
         if let Some(fetch) = self.scan.fetch {
             write!(f, ", fetch={fetch}")?;
         }
-        Ok(())
+        self.scan.fmt_dynamic_filters(f)
     }
     fn output_partitioning(&self) -> Partitioning {
         self.scan.partitioning.clone()
@@ -200,4 +202,21 @@ impl DataSource for EagerIcebergDataSource {
     fn metrics(&self) -> ExecutionPlanMetricsSet {
         self.scan.metrics.clone()
     }
+
+    fn try_pushdown_filters(
+        &self,
+        filters: Vec<Arc<dyn PhysicalExpr>>,
+        _config: &ConfigOptions,
+    ) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
+        let result = FilterPushdownPropagation::with_parent_pushdown_result(vec![
+            PushedDown::No;
+            filters.len()
+        ]);
+        if filters.is_empty() {
+            return Ok(result);
+        }
+        let mut source = self.clone();
+        source.scan.push_filters(filters);
+        Ok(result.with_updated_node(Arc::new(source)))
+    }
 }

datafusion/distributed-iceberg/src/lazy/codec.rs

@@ -103,16 +103,24 @@ impl PhysicalExtensionCodec for LazyIcebergCodec {
             proto.custom_file_io.as_deref(),
             ctx,
         )?;
+        let filters = proto
+            .filters
+            .as_ref()
+            .map(|filters| proto_converter.proto_to_physical_expr(filters, &schema, &decode_ctx))
+            .transpose()?;
+        let mut scan = ScanConfig::new(
+            schema,
+            partitioning,
+            fetch,
+            iceberg_file_io,
+            self.iceberg_runtime.clone(),
+            ctx,
+        );
+        scan.filters = filters;
+        scan.dynamic_filtering_enabled = proto.dynamic_filtering_enabled;
 
         Ok(DataSourceExec::from_data_source(LazyIcebergDataSource {
-            scan: ScanConfig::new(
-                schema,
-                partitioning,
-                fetch,
-                iceberg_file_io,
-                self.iceberg_runtime.clone(),
-                ctx,
-            ),
+            scan,
             column_stats: None,
             table_snapshot: None,
             feed,
@@ -150,6 +158,13 @@ impl PhysicalExtensionCodec for LazyIcebergCodec {
             fetch: source.scan.fetch.map(|value| value as u64),
             storage_properties: file_io.properties,
             custom_file_io: file_io.custom,
+            filters: source
+                .scan
+                .filters
+                .as_ref()
+                .map(|filters| proto_converter.physical_expr_to_proto(filters, self))
+                .transpose()?,
+            dynamic_filtering_enabled: source.scan.dynamic_filtering_enabled,
         };
         proto.encode(buf).map_err(|error| {
             proto_error(format!("failed to encode LazyIcebergDataSource: {error}"))
@@ -161,12 +176,16 @@ impl PhysicalExtensionCodec for LazyIcebergCodec {
 mod tests {
     use datafusion::common::Statistics;
     use datafusion::datasource::source::DataSource;
+    use datafusion::logical_expr::Operator;
+    use datafusion::physical_expr::expressions::{binary, col, lit};
+    use datafusion::prelude::SessionConfig;
     use datafusion_distributed_iceberg_test_tools::create_taxi_table;
     use datafusion_proto::physical_plan::DefaultPhysicalProtoConverter;
     use iceberg::io::{FileIOBuilder, LocalFsStorageFactory};
     use iceberg::table::Table;
 
     use super::*;
+    use crate::IcebergConfig;
     use crate::common::df_err;
     use crate::table_provider::scan::{IcebergDataSourceOptions, plan_scan};
 
@@ -186,6 +205,10 @@ mod tests {
             .runtime(iceberg::Runtime::current())
             .build()
             .map_err(df_err)?;
+        let config = SessionConfig::new().with_option_extension(IcebergConfig {
+            dynamic_filtering_enabled: true,
+            ..Default::default()
+        });
         let options = IcebergDataSourceOptions {
             fetch: Some(10),
             ..Default::default()
@@ -194,9 +217,21 @@ mod tests {
             table,
             taxi.schema()?,
             options,
-            Arc::new(TaskContext::default()),
+            Arc::new(TaskContext::default().with_session_config(config.clone())),
         )
         .await?;
+        let schema = taxi.schema()?;
+        let filter = binary(
+            col("fare_amount", &schema)?,
+            Operator::Gt,
+            lit(10.0),
+            &schema,
+        )?;
+        let pushed = iceberg_source(&plan)?.try_pushdown_filters(vec![filter], config.options())?;
+        let Some(pushed_source) = pushed.updated_node else {
+            return internal_err!("the scan should keep the filter it was offered");
+        };
+        let plan: Arc<dyn ExecutionPlan> = Arc::new(DataSourceExec::new(pushed_source));
         let codec =
             LazyIcebergCodec::new(Arc::new(LocalFsStorageFactory), iceberg::Runtime::current());
         let proto_converter = DefaultPhysicalProtoConverter {};
@@ -231,6 +266,11 @@ mod tests {
             decoded.partition_statistics(None)?.as_ref(),
             &Statistics::new_unknown(&decoded.scan.schema)
         );
+        let Some(filters) = &decoded.scan.filters else {
+            return internal_err!("the decoded scan should keep the filter");
+        };
+        insta::assert_snapshot!(filters, @"fare_amount@8 > 10");
+        assert!(decoded.scan.dynamic_filtering_enabled);
         Ok(())
     }