From 5aab3de4053a6b4cf26e83c8979efac6c47bb7b6 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:04:13 +0000 Subject: [PATCH 1/9] feat: keep pushed filters in Iceberg scans DataFusion offers each scan the filters above it, including the dynamic filters that joins, top-N sorts and aggregates fill in while the query runs. Iceberg scans didn't keep them. Keeping them is the first step toward skipping files with them. Both scans now keep every pushed filter, ANDed into one predicate. They answer that they don't apply them, so DataFusion still checks every row. EXPLAIN shows the dynamic ones on the scan as dynamic_filters=, and the codecs carry all of them to workers. With iceberg.dynamic_filtering_enabled on, scans also report the filters through apply_expressions. Joins and aggregates fill in a dynamic filter only when a scan reports it, and the coordinator sends updates to the workers holding those scans. The setting is off by default, and the plan carries it to workers. Enable it only once every worker supports it: an older worker can fail a query whose scan reports a dynamic filter. --- codegen/src/main.rs | 4 + datafusion/distributed-iceberg/README.md | 10 + .../src/common/scan_config.rs | 45 +++- datafusion/distributed-iceberg/src/config.rs | 2 + .../distributed-iceberg/src/eager/codec.rs | 42 ++-- .../src/eager/data_source.rs | 25 ++- .../distributed-iceberg/src/lazy/codec.rs | 87 +++++++- .../src/lazy/data_source.rs | 47 +++-- .../src/proto/generated/iceberg.rs | 14 ++ .../src/proto/iceberg.proto | 11 + .../tests/dynamic_filtering.rs | 192 ++++++++++++++++++ .../tests/limit_pushdown.rs | 4 +- .../tests/projection_pushdown.rs | 4 +- 13 files changed, 438 insertions(+), 49 deletions(-) create mode 100644 datafusion/distributed-iceberg/tests/dynamic_filtering.rs diff --git a/codegen/src/main.rs b/codegen/src/main.rs index dd9440dd5..f8922b032 100644 --- a/codegen/src/main.rs +++ b/codegen/src/main.rs @@ -73,6 +73,10 @@ fn generate_iceberg(repo_root: &Path) -> Result<(), Box> ".iceberg.DataFusionPartitioning", "::datafusion_proto::protobuf::Partitioning", ) + .extern_path( + ".iceberg.DataFusionPhysicalExprNode", + "::datafusion_proto::protobuf::PhysicalExprNode", + ) .compile_protos(&[proto_file], &[proto_dir])?; Ok(()) diff --git a/datafusion/distributed-iceberg/README.md b/datafusion/distributed-iceberg/README.md index 5d5c5a4c9..92cf2d66f 100644 --- a/datafusion/distributed-iceberg/README.md +++ b/datafusion/distributed-iceberg/README.md @@ -90,6 +90,16 @@ 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 fill in while the query runs. Scans +answer that they don't apply them, so DataFusion still checks every row. `EXPLAIN` +shows the dynamic ones on the scan as `dynamic_filters=`. Enable `iceberg.dynamic_filtering_enabled` to +report those filters to the joins and aggregates that produce them, which fill a +dynamic filter in only when something consumes it. Scans don't skip data with them +yet. Workers don't see `iceberg.*` settings, so the plan carries the coordinator's +value to them. Enable it only once every worker runs a version that supports it: an +older worker can fail a query whose scan reports a dynamic filter. + ```bash cargo test -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools ``` diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index 600d6e48c..7614c5416 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -1,10 +1,14 @@ use crate::{IcebergConfig, common::df_err}; use datafusion::arrow::datatypes::SchemaRef; use datafusion::common::stats::Precision; +use datafusion::common::tree_node::TreeNodeRecursion; use datafusion::common::{ColumnStatistics, Statistics}; use datafusion::error::Result; use datafusion::execution::{SendableRecordBatchStream, TaskContext}; -use datafusion::physical_expr::Partitioning; +use datafusion::physical_expr::{ + DynamicFilterTracking, Partitioning, PhysicalExpr, conjunction, split_conjunction, +}; +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 +34,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>, + pub(crate) dynamic_filtering_enabled: bool, // Clones share the delete loader across this source's local partitions. reader: ArrowReader, } @@ -57,10 +63,45 @@ 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>) { + 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) -> Result, + ) -> Result { + if !self.dynamic_filtering_enabled { + return Ok(TreeNodeRecursion::Continue); + } + apply_expression_roots(self.filters.iter(), f) + } + + /// Writes the scan's dynamic filters. Its static filters already show in the + /// `FilterExec` above it, since the scan doesn't apply them. + pub(crate) fn fmt_dynamic_filters(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + 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={}", conjunction(dynamic_filters)) + } + pub(crate) fn open( &self, tasks: FileScanTaskStream, @@ -109,6 +150,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() } } diff --git a/datafusion/distributed-iceberg/src/config.rs b/datafusion/distributed-iceberg/src/config.rs index d9c24e850..4a7f3df76 100644 --- a/datafusion/distributed-iceberg/src/config.rs +++ b/datafusion/distributed-iceberg/src/config.rs @@ -18,6 +18,8 @@ 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 + /// Whether to enable dynamic filtering. + pub dynamic_filtering_enabled: bool, default = false } } diff --git a/datafusion/distributed-iceberg/src/eager/codec.rs b/datafusion/distributed-iceberg/src/eager/codec.rs index 9aa97478d..94befaecc 100644 --- a/datafusion/distributed-iceberg/src/eager/codec.rs +++ b/datafusion/distributed-iceberg/src/eager/codec.rs @@ -95,16 +95,17 @@ impl PhysicalExtensionCodec for EagerIcebergCodec { .collect::>>() }) .collect::>>()?; - 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,24 @@ impl EagerIcebergCodec { proto.custom_file_io.as_deref(), ctx, )?; - Ok(ScanConfig::new( + // Decode with the plan's converter, so a filter keeps sharing its state with the + // operator that fills it in when both run in the same task. + 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 +214,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, }) } } diff --git a/datafusion/distributed-iceberg/src/eager/data_source.rs b/datafusion/distributed-iceberg/src/eager/data_source.rs index b4017fb3a..06df942f4 100644 --- a/datafusion/distributed-iceberg/src/eager/data_source.rs +++ b/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) -> Result, + f: &mut dyn FnMut(&Arc) -> Result, ) -> Result { - 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>, + _config: &ConfigOptions, + ) -> Result>> { + 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))) + } } diff --git a/datafusion/distributed-iceberg/src/lazy/codec.rs b/datafusion/distributed-iceberg/src/lazy/codec.rs index 779f6e716..506d8353e 100644 --- a/datafusion/distributed-iceberg/src/lazy/codec.rs +++ b/datafusion/distributed-iceberg/src/lazy/codec.rs @@ -103,16 +103,26 @@ impl PhysicalExtensionCodec for LazyIcebergCodec { proto.custom_file_io.as_deref(), ctx, )?; + // Decode with the plan's converter, so a filter keeps sharing its state with the + // operator that fills it in when both run in the same task. + 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 +160,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 +178,17 @@ impl PhysicalExtensionCodec for LazyIcebergCodec { mod tests { use datafusion::common::Statistics; use datafusion::datasource::source::DataSource; + use datafusion::logical_expr::Operator; + use datafusion::physical_expr::PhysicalExpr; + use datafusion::physical_expr::expressions::{BinaryExpr, Column, 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}; @@ -234,6 +256,55 @@ mod tests { Ok(()) } + #[tokio::test] + async fn roundtrips_pushed_filters_and_whether_the_scan_reports_them() -> Result<()> { + let taxi = create_taxi_table().await?; + let config = SessionConfig::new().with_option_extension(IcebergConfig { + dynamic_filtering_enabled: true, + ..Default::default() + }); + let plan = plan_scan( + taxi.table().clone(), + taxi.schema()?, + Default::default(), + Arc::new(TaskContext::default().with_session_config(config.clone())), + ) + .await?; + let fare_over_ten: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new_with_schema( + "fare_amount", + taxi.schema()?.as_ref(), + )?), + Operator::Gt, + lit(10.0), + )); + let pushed = + iceberg_source(&plan)?.try_pushdown_filters(vec![fare_over_ten], config.options())?; + let Some(source) = pushed.updated_node else { + return internal_err!("the scan should keep the filter it was offered"); + }; + + let codec = + LazyIcebergCodec::new(Arc::new(LocalFsStorageFactory), iceberg::Runtime::current()); + let proto_converter = DefaultPhysicalProtoConverter {}; + let mut buf = Vec::new(); + codec.try_encode( + Arc::new(DataSourceExec::new(source)), + &mut buf, + &proto_converter, + )?; + let decoded_plan = + codec.try_decode(&buf, &[], &TaskContext::default(), &proto_converter)?; + let decoded = iceberg_source(&decoded_plan)?; + + 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(()) + } + fn iceberg_source(plan: &Arc) -> Result<&LazyIcebergDataSource> { let Some(exec) = plan.downcast_ref::() else { return internal_err!("expected a DataSourceExec"); diff --git a/datafusion/distributed-iceberg/src/lazy/data_source.rs b/datafusion/distributed-iceberg/src/lazy/data_source.rs index 38686490e..97339bab3 100644 --- a/datafusion/distributed-iceberg/src/lazy/data_source.rs +++ b/datafusion/distributed-iceberg/src/lazy/data_source.rs @@ -126,9 +126,9 @@ impl LazyIcebergDataSource { impl DataSource for LazyIcebergDataSource { fn apply_expressions( &self, - _f: &mut dyn FnMut(&Arc) -> Result, + f: &mut dyn FnMut(&Arc) -> Result, ) -> Result { - Ok(TreeNodeRecursion::Continue) + self.scan.apply_expressions(f) } fn open( @@ -149,20 +149,20 @@ impl DataSource for LazyIcebergDataSource { fn fmt_as(&self, _t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result { write!(f, "format=iceberg")?; - let Some(feed) = self.feed.inner() else { - return Ok(()); - }; - write!(f, ", table={}", feed.iceberg_table.identifier())?; - if let Some(projection) = &feed.projection { - write!(f, ", projection=[{}]", projection.join(", "))?; + // Workers' copies have no feed, but still show their dynamic filters. + if let Some(feed) = self.feed.inner() { + write!(f, ", table={}", feed.iceberg_table.identifier())?; + if let Some(projection) = &feed.projection { + write!(f, ", projection=[{}]", projection.join(", "))?; + } + if let Some(predicate) = &feed.predicates { + write!(f, ", predicate={predicate}")?; + } + if let Some(fetch) = self.scan.fetch { + write!(f, ", fetch={fetch}")?; + } } - if let Some(predicate) = &feed.predicates { - write!(f, ", predicate={predicate}")?; - } - if let Some(fetch) = self.scan.fetch { - write!(f, ", fetch={fetch}")?; - } - Ok(()) + self.scan.fmt_dynamic_filters(f) } fn output_partitioning(&self) -> Partitioning { @@ -212,12 +212,17 @@ impl DataSource for LazyIcebergDataSource { filters: Vec>, _config: &ConfigOptions, ) -> Result>> { - // TODO: Allow this DataSource to be pushed down filters. Some filters might be more - // straight forward to accept, like simple predicates, but some others might require - // a bit more work, like dynamic filters. - Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], - )) + // The scan keeps the filters but doesn't apply them, so DataFusion still checks every row. + 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))) } fn try_pushdown_sort( diff --git a/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs b/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs index 1a19d3ea9..4ca5feae5 100644 --- a/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs +++ b/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs @@ -15,6 +15,13 @@ pub struct IcebergDataSource { ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>, #[prost(bytes = "vec", optional, tag = "6")] pub custom_file_io: ::core::option::Option<::prost::alloc::vec::Vec>, + /// Filters pushed into the scan, ANDed together. + #[prost(message, optional, tag = "7")] + pub filters: ::core::option::Option<::datafusion_proto::protobuf::PhysicalExprNode>, + /// Whether the scan takes part in dynamic filtering. It reports its filters to + /// the operators that fill them in. + #[prost(bool, tag = "8")] + pub dynamic_filtering_enabled: bool, } #[allow(clippy::all)] #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] @@ -36,6 +43,13 @@ pub struct ScanConfig { ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>, #[prost(bytes = "vec", optional, tag = "5")] pub custom_file_io: ::core::option::Option<::prost::alloc::vec::Vec>, + /// Filters pushed into the scan, ANDed together. + #[prost(message, optional, tag = "6")] + pub filters: ::core::option::Option<::datafusion_proto::protobuf::PhysicalExprNode>, + /// Whether the scan takes part in dynamic filtering. It reports its filters to + /// the operators that fill them in. + #[prost(bool, tag = "7")] + pub dynamic_filtering_enabled: bool, } #[allow(clippy::all)] #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/datafusion/distributed-iceberg/src/proto/iceberg.proto b/datafusion/distributed-iceberg/src/proto/iceberg.proto index b02d978ac..1a936e936 100644 --- a/datafusion/distributed-iceberg/src/proto/iceberg.proto +++ b/datafusion/distributed-iceberg/src/proto/iceberg.proto @@ -8,6 +8,11 @@ message IcebergDataSource { optional uint64 fetch = 4; map storage_properties = 5; optional bytes custom_file_io = 6; + // Filters pushed into the scan, ANDed together. + DataFusionPhysicalExprNode filters = 7; + // Whether the scan takes part in dynamic filtering. It reports its filters to + // the operators that fill them in. + bool dynamic_filtering_enabled = 8; } message WorkUnitFeed { @@ -17,6 +22,7 @@ message WorkUnitFeed { // The generated Rust fields use DataFusion's canonical protobuf types. message DataFusionSchema {} message DataFusionPartitioning {} +message DataFusionPhysicalExprNode {} message ScanConfig { DataFusionSchema schema = 1; @@ -24,6 +30,11 @@ message ScanConfig { optional uint64 fetch = 3; map storage_properties = 4; optional bytes custom_file_io = 5; + // Filters pushed into the scan, ANDed together. + DataFusionPhysicalExprNode filters = 6; + // Whether the scan takes part in dynamic filtering. It reports its filters to + // the operators that fill them in. + bool dynamic_filtering_enabled = 7; } message AssignedScan { ScanConfig config = 1; diff --git a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs new file mode 100644 index 000000000..dde8d0d25 --- /dev/null +++ b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs @@ -0,0 +1,192 @@ +mod tests { + use datafusion::arrow::util::pretty::pretty_format_batches; + use datafusion::common::Result; + use datafusion::physical_plan::collect; + use datafusion_distributed::{ + DistributedExt, display_plan_ascii, rewrite_distributed_plan_with_dynamic_filters, + }; + use datafusion_distributed_iceberg::IcebergExt; + use datafusion_distributed_iceberg_test_tools::IcebergTestHarness; + use datafusion_distributed_iceberg_test_tools::create_taxi_table; + use std::sync::Arc; + + #[tokio::test] + async fn lazy_planning_join_fills_in_the_filter_the_scan_reports() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_session_extension(|state| state.with_iceberg_integration(Default::default())) + .with_table(create_taxi_table().await?) + .with_config_option("iceberg.eager_file_planning", false) + .with_config_option("iceberg.dynamic_filtering_enabled", true) + .build() + .await?; + + let plan = harness + .physical_plan( + "SELECT COUNT(*) AS trips FROM taxi \ + JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ + WHERE taxi.trip_distance >= 2.0", + ) + .await?; + let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + + insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" + ProjectionExec: expr=[count(Int64(1))@0 as trips] + AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] + CoalescePartitionsExec + AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] + HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] + ProjectionExec: expr=[column1@0 as day] + DataSourceExec: partitions=1, partition_sizes=[1] + FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] + DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] + "); + insta::assert_snapshot!(pretty_format_batches(&batches)?, @" + +-------+ + | trips | + +-------+ + | 11714 | + +-------+ + "); + Ok(()) + } + + #[tokio::test] + async fn eager_planning_join_fills_in_the_filter_the_scan_reports() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_session_extension(|state| state.with_iceberg_integration(Default::default())) + .with_table(create_taxi_table().await?) + .with_config_option("iceberg.eager_file_planning", true) + .with_config_option("iceberg.dynamic_filtering_enabled", true) + .build() + .await?; + + let plan = harness + .physical_plan( + "SELECT COUNT(*) AS trips FROM taxi \ + JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ + WHERE taxi.trip_distance >= 2.0", + ) + .await?; + let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + + insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" + ProjectionExec: expr=[count(Int64(1))@0 as trips] + AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] + CoalescePartitionsExec + AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] + HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] + ProjectionExec: expr=[column1@0 as day] + DataSourceExec: partitions=1, partition_sizes=[1] + FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] + DataSourceExec: format=iceberg, planned_files=7, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] + "); + insta::assert_snapshot!(pretty_format_batches(&batches)?, @" + +-------+ + | trips | + +-------+ + | 11714 | + +-------+ + "); + Ok(()) + } + + #[tokio::test] + async fn join_leaves_its_filter_empty_when_the_scan_does_not_report_it() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_session_extension(|state| state.with_iceberg_integration(Default::default())) + .with_table(create_taxi_table().await?) + .with_config_option("iceberg.eager_file_planning", false) + .with_config_option("iceberg.dynamic_filtering_enabled", false) + .build() + .await?; + + let plan = harness + .physical_plan( + "SELECT COUNT(*) AS trips FROM taxi \ + JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ + WHERE taxi.trip_distance >= 2.0", + ) + .await?; + let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + + insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" + ProjectionExec: expr=[count(Int64(1))@0 as trips] + AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] + CoalescePartitionsExec + AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] + HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] + ProjectionExec: expr=[column1@0 as day] + DataSourceExec: partitions=1, partition_sizes=[1] + FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] + DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ empty ] + "); + insta::assert_snapshot!(pretty_format_batches(&batches)?, @" + +-------+ + | trips | + +-------+ + | 11714 | + +-------+ + "); + Ok(()) + } + + #[tokio::test] + async fn eager_planning_workers_fill_in_the_filter_their_scans_report() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_session_extension(|state| state.with_iceberg_integration(Default::default())) + .with_table(create_taxi_table().await?) + .with_config_option("iceberg.eager_file_planning", true) + .with_config_option("iceberg.dynamic_filtering_enabled", true) + .with_workers(2) + .configure_session(|state| { + state + .with_distributed_file_scan_config_bytes_per_partition(1)? + .with_distributed_dynamic_filter_collection(true) + })? + .build() + .await?; + + let plan = harness + .physical_plan( + "SELECT COUNT(*) AS trips FROM taxi \ + JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ + WHERE taxi.trip_distance >= 2.0", + ) + .await?; + let task_ctx = harness.state().task_ctx(); + let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)).await?; + let plan = rewrite_distributed_plan_with_dynamic_filters(plan, &task_ctx).await?; + + insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" + ┌───── DistributedExec + │ ProjectionExec: expr=[count(Int64(1))@0 as trips] + │ AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] + │ CoalescePartitionsExec + │ [Stage 2] => NetworkCoalesceExec: output_partitions=8, input_tasks=2 + └────────────────────────────────────────────────── + ┌───── Stage 2 ── tasks=2, partitions=8 + │ AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] + │ HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] + │ CoalescePartitionsExec + │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=1, stage_partitions=2, input_tasks=1 + │ FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] + │ DistributedLeafExec: + │ t0: DataSourceExec: format=iceberg, planned_files=3, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] + │ t1: DataSourceExec: format=iceberg, planned_files=4, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] + └────────────────────────────────────────────────── + ┌───── Stage 1 ── tasks=1, partitions=2 + │ BroadcastExec: input_partitions=1, consumer_tasks=2, output_partitions=2 + │ ProjectionExec: expr=[column1@0 as day] + │ DataSourceExec: partitions=1, partition_sizes=[1] + └────────────────────────────────────────────────── + "); + insta::assert_snapshot!(pretty_format_batches(&batches)?, @" + +-------+ + | trips | + +-------+ + | 11714 | + +-------+ + "); + Ok(()) + } +} diff --git a/datafusion/distributed-iceberg/tests/limit_pushdown.rs b/datafusion/distributed-iceberg/tests/limit_pushdown.rs index 0c23e0455..87ee3e17d 100644 --- a/datafusion/distributed-iceberg/tests/limit_pushdown.rs +++ b/datafusion/distributed-iceberg/tests/limit_pushdown.rs @@ -24,7 +24,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST], fetch=3 SortExec: TopK(fetch=3), expr=[pickup_at@1 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@4 = 2024-01-10 - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, passenger_count, trip_distance, pickup_date], predicate=pickup_date = 2024-01-10 + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, passenger_count, trip_distance, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+-----------------+---------------+-------------+ @@ -130,7 +130,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@2 ASC NULLS LAST], fetch=5 SortExec: TopK(fetch=5), expr=[pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@2 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@3 = 2024-01-10, projection=[vendor_id@0, pickup_at@1, pickup_location_id@2] - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10 + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+ diff --git a/datafusion/distributed-iceberg/tests/projection_pushdown.rs b/datafusion/distributed-iceberg/tests/projection_pushdown.rs index b93d68e01..b9c91ce32 100644 --- a/datafusion/distributed-iceberg/tests/projection_pushdown.rs +++ b/datafusion/distributed-iceberg/tests/projection_pushdown.rs @@ -62,7 +62,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@2 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST], fetch=3 SortExec: TopK(fetch=3), expr=[pickup_at@2 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@3 = 2024-01-10, projection=[vendor_id@0, pickup_location_id@2, pickup_at@1] - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10 + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] "); insta::assert_snapshot!(batches, @r" +-----------+--------------------+ @@ -126,7 +126,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@5 ASC NULLS LAST], fetch=1 SortExec: TopK(fetch=1), expr=[pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@5 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@12 = 2024-01-10 - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, dropoff_at, passenger_count, trip_distance, pickup_location_id, dropoff_location_id, payment_type, fare_amount, tip_amount, tolls_amount, total_amount, pickup_date], predicate=pickup_date = 2024-01-10 + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, dropoff_at, passenger_count, trip_distance, pickup_location_id, dropoff_location_id, payment_type, fare_amount, tip_amount, tolls_amount, total_amount, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+---------------------+-----------------+---------------+--------------------+---------------------+--------------+-------------+------------+--------------+--------------+-------------+ From 860d45ae7675452b4f2293116c4bd82bfb9aa423 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:33:27 +0000 Subject: [PATCH 2/9] nit: minor comment/doc cleanup --- datafusion/distributed-iceberg/README.md | 12 +++++------- .../distributed-iceberg/src/common/scan_config.rs | 3 +-- datafusion/distributed-iceberg/src/config.rs | 2 +- 3 files changed, 7 insertions(+), 10 deletions(-) diff --git a/datafusion/distributed-iceberg/README.md b/datafusion/distributed-iceberg/README.md index 92cf2d66f..81ffbeee4 100644 --- a/datafusion/distributed-iceberg/README.md +++ b/datafusion/distributed-iceberg/README.md @@ -92,13 +92,11 @@ 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 fill in while the query runs. Scans -answer that they don't apply them, so DataFusion still checks every row. `EXPLAIN` -shows the dynamic ones on the scan as `dynamic_filters=`. Enable `iceberg.dynamic_filtering_enabled` to -report those filters to the joins and aggregates that produce them, which fill a -dynamic filter in only when something consumes it. Scans don't skip data with them -yet. Workers don't see `iceberg.*` settings, so the plan carries the coordinator's -value to them. Enable it only once every worker runs a version that supports it: an -older worker can fail a query whose scan reports a dynamic filter. +return `PushedDown::No` for these filters, so DataFusion still checks every row. `EXPLAIN` +shows the dynamic ones on the scan as `dynamic_filters=`. +Joins and aggregates only fill the dynamic filters if there is a consumer for them - +enabling `iceberg.dynamic_filtering_enabled` informs the producers +that the scan will consume them. ```bash cargo test -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index 7614c5416..c77645a11 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -85,8 +85,7 @@ impl ScanConfig { apply_expression_roots(self.filters.iter(), f) } - /// Writes the scan's dynamic filters. Its static filters already show in the - /// `FilterExec` above it, since the scan doesn't apply them. + /// Writes the scan's dynamic filters into `f` pub(crate) fn fmt_dynamic_filters(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let Some(filters) = &self.filters else { return Ok(()); diff --git a/datafusion/distributed-iceberg/src/config.rs b/datafusion/distributed-iceberg/src/config.rs index 4a7f3df76..089c797e4 100644 --- a/datafusion/distributed-iceberg/src/config.rs +++ b/datafusion/distributed-iceberg/src/config.rs @@ -18,7 +18,7 @@ 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 - /// Whether to enable dynamic filtering. + /// (Experimental) Whether to enable dynamic filtering. pub dynamic_filtering_enabled: bool, default = false } } From 8e036791bed46d3d29746e589fd64905ae6dc3c6 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:59:16 +0000 Subject: [PATCH 3/9] nit: test cleanup --- .../distributed-iceberg/src/lazy/codec.rs | 64 +++++----------- .../tests/dynamic_filtering.rs | 74 ++----------------- 2 files changed, 25 insertions(+), 113 deletions(-) diff --git a/datafusion/distributed-iceberg/src/lazy/codec.rs b/datafusion/distributed-iceberg/src/lazy/codec.rs index 506d8353e..47b149911 100644 --- a/datafusion/distributed-iceberg/src/lazy/codec.rs +++ b/datafusion/distributed-iceberg/src/lazy/codec.rs @@ -208,6 +208,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() @@ -216,9 +220,23 @@ mod tests { table, taxi.schema()?, options, - Arc::new(TaskContext::default()), + Arc::new(TaskContext::default().with_session_config(config.clone())), ) .await?; + let fare_over_ten: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new_with_schema( + "fare_amount", + taxi.schema()?.as_ref(), + )?), + Operator::Gt, + lit(10.0), + )); + let pushed = + iceberg_source(&plan)?.try_pushdown_filters(vec![fare_over_ten], 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 = Arc::new(DataSourceExec::new(pushed_source)); let codec = LazyIcebergCodec::new(Arc::new(LocalFsStorageFactory), iceberg::Runtime::current()); let proto_converter = DefaultPhysicalProtoConverter {}; @@ -253,50 +271,6 @@ mod tests { decoded.partition_statistics(None)?.as_ref(), &Statistics::new_unknown(&decoded.scan.schema) ); - Ok(()) - } - - #[tokio::test] - async fn roundtrips_pushed_filters_and_whether_the_scan_reports_them() -> Result<()> { - let taxi = create_taxi_table().await?; - let config = SessionConfig::new().with_option_extension(IcebergConfig { - dynamic_filtering_enabled: true, - ..Default::default() - }); - let plan = plan_scan( - taxi.table().clone(), - taxi.schema()?, - Default::default(), - Arc::new(TaskContext::default().with_session_config(config.clone())), - ) - .await?; - let fare_over_ten: Arc = Arc::new(BinaryExpr::new( - Arc::new(Column::new_with_schema( - "fare_amount", - taxi.schema()?.as_ref(), - )?), - Operator::Gt, - lit(10.0), - )); - let pushed = - iceberg_source(&plan)?.try_pushdown_filters(vec![fare_over_ten], config.options())?; - let Some(source) = pushed.updated_node else { - return internal_err!("the scan should keep the filter it was offered"); - }; - - let codec = - LazyIcebergCodec::new(Arc::new(LocalFsStorageFactory), iceberg::Runtime::current()); - let proto_converter = DefaultPhysicalProtoConverter {}; - let mut buf = Vec::new(); - codec.try_encode( - Arc::new(DataSourceExec::new(source)), - &mut buf, - &proto_converter, - )?; - let decoded_plan = - codec.try_decode(&buf, &[], &TaskContext::default(), &proto_converter)?; - let decoded = iceberg_source(&decoded_plan)?; - let Some(filters) = &decoded.scan.filters else { return internal_err!("the decoded scan should keep the filter"); }; diff --git a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs index dde8d0d25..59a1dfd14 100644 --- a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs +++ b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs @@ -1,5 +1,4 @@ mod tests { - use datafusion::arrow::util::pretty::pretty_format_batches; use datafusion::common::Result; use datafusion::physical_plan::collect; use datafusion_distributed::{ @@ -11,7 +10,7 @@ mod tests { use std::sync::Arc; #[tokio::test] - async fn lazy_planning_join_fills_in_the_filter_the_scan_reports() -> Result<()> { + async fn lazy_planning_scan_gets_the_join_filter() -> Result<()> { let harness = IcebergTestHarness::builder() .with_session_extension(|state| state.with_iceberg_integration(Default::default())) .with_table(create_taxi_table().await?) @@ -27,7 +26,7 @@ mod tests { WHERE taxi.trip_distance >= 2.0", ) .await?; - let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + collect(Arc::clone(&plan), harness.state().task_ctx()).await?; insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" ProjectionExec: expr=[count(Int64(1))@0 as trips] @@ -40,58 +39,11 @@ mod tests { FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] "); - insta::assert_snapshot!(pretty_format_batches(&batches)?, @" - +-------+ - | trips | - +-------+ - | 11714 | - +-------+ - "); - Ok(()) - } - - #[tokio::test] - async fn eager_planning_join_fills_in_the_filter_the_scan_reports() -> Result<()> { - let harness = IcebergTestHarness::builder() - .with_session_extension(|state| state.with_iceberg_integration(Default::default())) - .with_table(create_taxi_table().await?) - .with_config_option("iceberg.eager_file_planning", true) - .with_config_option("iceberg.dynamic_filtering_enabled", true) - .build() - .await?; - - let plan = harness - .physical_plan( - "SELECT COUNT(*) AS trips FROM taxi \ - JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ - WHERE taxi.trip_distance >= 2.0", - ) - .await?; - let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; - - insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" - ProjectionExec: expr=[count(Int64(1))@0 as trips] - AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] - CoalescePartitionsExec - AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] - ProjectionExec: expr=[column1@0 as day] - DataSourceExec: partitions=1, partition_sizes=[1] - FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] - DataSourceExec: format=iceberg, planned_files=7, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] - "); - insta::assert_snapshot!(pretty_format_batches(&batches)?, @" - +-------+ - | trips | - +-------+ - | 11714 | - +-------+ - "); Ok(()) } #[tokio::test] - async fn join_leaves_its_filter_empty_when_the_scan_does_not_report_it() -> Result<()> { + async fn scan_gets_an_empty_join_filter_when_dynamic_filtering_is_off() -> Result<()> { let harness = IcebergTestHarness::builder() .with_session_extension(|state| state.with_iceberg_integration(Default::default())) .with_table(create_taxi_table().await?) @@ -107,7 +59,7 @@ mod tests { WHERE taxi.trip_distance >= 2.0", ) .await?; - let batches = collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + collect(Arc::clone(&plan), harness.state().task_ctx()).await?; insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" ProjectionExec: expr=[count(Int64(1))@0 as trips] @@ -120,18 +72,11 @@ mod tests { FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ empty ] "); - insta::assert_snapshot!(pretty_format_batches(&batches)?, @" - +-------+ - | trips | - +-------+ - | 11714 | - +-------+ - "); Ok(()) } #[tokio::test] - async fn eager_planning_workers_fill_in_the_filter_their_scans_report() -> Result<()> { + async fn eager_planning_workers_get_the_join_filter() -> Result<()> { let harness = IcebergTestHarness::builder() .with_session_extension(|state| state.with_iceberg_integration(Default::default())) .with_table(create_taxi_table().await?) @@ -154,7 +99,7 @@ mod tests { ) .await?; let task_ctx = harness.state().task_ctx(); - let batches = collect(Arc::clone(&plan), Arc::clone(&task_ctx)).await?; + collect(Arc::clone(&plan), Arc::clone(&task_ctx)).await?; let plan = rewrite_distributed_plan_with_dynamic_filters(plan, &task_ctx).await?; insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" @@ -180,13 +125,6 @@ mod tests { │ DataSourceExec: partitions=1, partition_sizes=[1] └────────────────────────────────────────────────── "); - insta::assert_snapshot!(pretty_format_batches(&batches)?, @" - +-------+ - | trips | - +-------+ - | 11714 | - +-------+ - "); Ok(()) } } From ccc2d92e212dc2a46185a2ea79379833e16ec500 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:22:03 +0000 Subject: [PATCH 4/9] nit: comment/docs cleanup --- datafusion/distributed-iceberg/README.md | 10 +++++----- .../distributed-iceberg/src/common/scan_config.rs | 2 +- datafusion/distributed-iceberg/src/config.rs | 4 +++- 3 files changed, 9 insertions(+), 7 deletions(-) diff --git a/datafusion/distributed-iceberg/README.md b/datafusion/distributed-iceberg/README.md index 81ffbeee4..dad7aa99a 100644 --- a/datafusion/distributed-iceberg/README.md +++ b/datafusion/distributed-iceberg/README.md @@ -91,12 +91,12 @@ 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 fill in while the query runs. Scans +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. `EXPLAIN` -shows the dynamic ones on the scan as `dynamic_filters=`. -Joins and aggregates only fill the dynamic filters if there is a consumer for them - -enabling `iceberg.dynamic_filtering_enabled` informs the producers -that the scan will consume them. +shows the dynamic ones on the scan as `dynamic_filters=`. 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. Scans don't skip data with these filters yet. ```bash cargo test -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index c77645a11..8f52a1310 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -85,7 +85,7 @@ impl ScanConfig { apply_expression_roots(self.filters.iter(), f) } - /// Writes the scan's dynamic filters into `f` + /// Writes the scan's dynamic filters into `f`. pub(crate) fn fmt_dynamic_filters(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { let Some(filters) = &self.filters else { return Ok(()); diff --git a/datafusion/distributed-iceberg/src/config.rs b/datafusion/distributed-iceberg/src/config.rs index 089c797e4..2f9e38c53 100644 --- a/datafusion/distributed-iceberg/src/config.rs +++ b/datafusion/distributed-iceberg/src/config.rs @@ -18,7 +18,9 @@ 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 to enable dynamic filtering. + /// (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 } } From 038a86766446a6938a5026cd9fa0e9aa4161bb41 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:25:59 +0000 Subject: [PATCH 5/9] feat: show dynamic filters in Iceberg scan plans only when dynamic filtering is on With iceberg.dynamic_filtering_enabled off, scans aren't consumers of the dynamic filters pushed into them, so those filters stay empty. Plans no longer print them in that case. --- datafusion/distributed-iceberg/README.md | 10 +++--- .../src/common/scan_config.rs | 5 ++- .../tests/dynamic_filtering.rs | 33 ------------------- .../tests/limit_pushdown.rs | 4 +-- .../tests/projection_pushdown.rs | 4 +-- 5 files changed, 13 insertions(+), 43 deletions(-) diff --git a/datafusion/distributed-iceberg/README.md b/datafusion/distributed-iceberg/README.md index dad7aa99a..9c5b6507b 100644 --- a/datafusion/distributed-iceberg/README.md +++ b/datafusion/distributed-iceberg/README.md @@ -92,11 +92,11 @@ 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. `EXPLAIN` -shows the dynamic ones on the scan as `dynamic_filters=`. 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. Scans don't skip data with these filters yet. +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 diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index 8f52a1310..60125e742 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -85,8 +85,11 @@ impl ScanConfig { apply_expression_roots(self.filters.iter(), f) } - /// Writes the scan's dynamic filters into `f`. + /// Writes the scan's dynamic filters into `f` when dynamic filtering is enabled. 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(()); }; diff --git a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs index 59a1dfd14..220931842 100644 --- a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs +++ b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs @@ -42,39 +42,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn scan_gets_an_empty_join_filter_when_dynamic_filtering_is_off() -> Result<()> { - let harness = IcebergTestHarness::builder() - .with_session_extension(|state| state.with_iceberg_integration(Default::default())) - .with_table(create_taxi_table().await?) - .with_config_option("iceberg.eager_file_planning", false) - .with_config_option("iceberg.dynamic_filtering_enabled", false) - .build() - .await?; - - let plan = harness - .physical_plan( - "SELECT COUNT(*) AS trips FROM taxi \ - JOIN (VALUES (DATE '2024-01-10')) AS days(day) ON taxi.pickup_date = days.day \ - WHERE taxi.trip_distance >= 2.0", - ) - .await?; - collect(Arc::clone(&plan), harness.state().task_ctx()).await?; - - insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" - ProjectionExec: expr=[count(Int64(1))@0 as trips] - AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] - CoalescePartitionsExec - AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] - ProjectionExec: expr=[column1@0 as day] - DataSourceExec: partitions=1, partition_sizes=[1] - FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] - DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ empty ] - "); - Ok(()) - } - #[tokio::test] async fn eager_planning_workers_get_the_join_filter() -> Result<()> { let harness = IcebergTestHarness::builder() diff --git a/datafusion/distributed-iceberg/tests/limit_pushdown.rs b/datafusion/distributed-iceberg/tests/limit_pushdown.rs index 87ee3e17d..0c23e0455 100644 --- a/datafusion/distributed-iceberg/tests/limit_pushdown.rs +++ b/datafusion/distributed-iceberg/tests/limit_pushdown.rs @@ -24,7 +24,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST], fetch=3 SortExec: TopK(fetch=3), expr=[pickup_at@1 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@4 = 2024-01-10 - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, passenger_count, trip_distance, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, passenger_count, trip_distance, pickup_date], predicate=pickup_date = 2024-01-10 "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+-----------------+---------------+-------------+ @@ -130,7 +130,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@2 ASC NULLS LAST], fetch=5 SortExec: TopK(fetch=5), expr=[pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@2 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@3 = 2024-01-10, projection=[vendor_id@0, pickup_at@1, pickup_location_id@2] - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10 "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+ diff --git a/datafusion/distributed-iceberg/tests/projection_pushdown.rs b/datafusion/distributed-iceberg/tests/projection_pushdown.rs index b9c91ce32..b93d68e01 100644 --- a/datafusion/distributed-iceberg/tests/projection_pushdown.rs +++ b/datafusion/distributed-iceberg/tests/projection_pushdown.rs @@ -62,7 +62,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@2 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST], fetch=3 SortExec: TopK(fetch=3), expr=[pickup_at@2 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@3 = 2024-01-10, projection=[vendor_id@0, pickup_location_id@2, pickup_at@1] - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, pickup_location_id, pickup_date], predicate=pickup_date = 2024-01-10 "); insta::assert_snapshot!(batches, @r" +-----------+--------------------+ @@ -126,7 +126,7 @@ mod tests { SortPreservingMergeExec: [pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@5 ASC NULLS LAST], fetch=1 SortExec: TopK(fetch=1), expr=[pickup_at@1 ASC NULLS LAST, vendor_id@0 ASC NULLS LAST, pickup_location_id@5 ASC NULLS LAST], preserve_partitioning=[true] FilterExec: pickup_date@12 = 2024-01-10 - DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, dropoff_at, passenger_count, trip_distance, pickup_location_id, dropoff_location_id, payment_type, fare_amount, tip_amount, tolls_amount, total_amount, pickup_date], predicate=pickup_date = 2024-01-10, dynamic_filters=DynamicFilter [ empty ] + DataSourceExec: format=iceberg, table=default.taxi, projection=[vendor_id, pickup_at, dropoff_at, passenger_count, trip_distance, pickup_location_id, dropoff_location_id, payment_type, fare_amount, tip_amount, tolls_amount, total_amount, pickup_date], predicate=pickup_date = 2024-01-10 "); insta::assert_snapshot!(batches, @r" +-----------+---------------------+---------------------+-----------------+---------------+--------------------+---------------------+--------------+-------------+------------+--------------+--------------+-------------+ From cab7f002d1bf071a359e4428ca7dc7a6e476a4e0 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Tue, 10 Nov 2026 10:16:05 +0000 Subject: [PATCH 6/9] nit: reword proto comments the spell check flags --- datafusion/distributed-iceberg/src/proto/generated/iceberg.rs | 4 ++-- datafusion/distributed-iceberg/src/proto/iceberg.proto | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs b/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs index 4ca5feae5..8e512a298 100644 --- a/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs +++ b/datafusion/distributed-iceberg/src/proto/generated/iceberg.rs @@ -15,7 +15,7 @@ pub struct IcebergDataSource { ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>, #[prost(bytes = "vec", optional, tag = "6")] pub custom_file_io: ::core::option::Option<::prost::alloc::vec::Vec>, - /// Filters pushed into the scan, ANDed together. + /// Filters pushed into the scan. A row must pass all of them. #[prost(message, optional, tag = "7")] pub filters: ::core::option::Option<::datafusion_proto::protobuf::PhysicalExprNode>, /// Whether the scan takes part in dynamic filtering. It reports its filters to @@ -43,7 +43,7 @@ pub struct ScanConfig { ::std::collections::HashMap<::prost::alloc::string::String, ::prost::alloc::string::String>, #[prost(bytes = "vec", optional, tag = "5")] pub custom_file_io: ::core::option::Option<::prost::alloc::vec::Vec>, - /// Filters pushed into the scan, ANDed together. + /// Filters pushed into the scan. A row must pass all of them. #[prost(message, optional, tag = "6")] pub filters: ::core::option::Option<::datafusion_proto::protobuf::PhysicalExprNode>, /// Whether the scan takes part in dynamic filtering. It reports its filters to diff --git a/datafusion/distributed-iceberg/src/proto/iceberg.proto b/datafusion/distributed-iceberg/src/proto/iceberg.proto index 1a936e936..613c4f33c 100644 --- a/datafusion/distributed-iceberg/src/proto/iceberg.proto +++ b/datafusion/distributed-iceberg/src/proto/iceberg.proto @@ -8,7 +8,7 @@ message IcebergDataSource { optional uint64 fetch = 4; map storage_properties = 5; optional bytes custom_file_io = 6; - // Filters pushed into the scan, ANDed together. + // Filters pushed into the scan. A row must pass all of them. DataFusionPhysicalExprNode filters = 7; // Whether the scan takes part in dynamic filtering. It reports its filters to // the operators that fill them in. @@ -30,7 +30,7 @@ message ScanConfig { optional uint64 fetch = 3; map storage_properties = 4; optional bytes custom_file_io = 5; - // Filters pushed into the scan, ANDed together. + // Filters pushed into the scan. A row must pass all of them. DataFusionPhysicalExprNode filters = 6; // Whether the scan takes part in dynamic filtering. It reports its filters to // the operators that fill them in. From 4dad4321ba543c5c315370a0a8186c8ed31cb677 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Tue, 10 Nov 2026 16:34:17 +0000 Subject: [PATCH 7/9] feat: print long IN lists in Iceberg scan dynamic filters as their count A join filter can list thousands of keys, which filled the plan. An IN list of more than 5 values now prints as its count, for example `pickup_date@0 IN ([7 elements])`. Only the printed plan changes. --- .../src/common/scan_config.rs | 36 +++++++++++++++++-- .../tests/dynamic_filtering.rs | 35 ++++++++++++++++++ 2 files changed, 68 insertions(+), 3 deletions(-) diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index 60125e742..da903c871 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -1,13 +1,15 @@ use crate::{IcebergConfig, common::df_err}; use datafusion::arrow::datatypes::SchemaRef; use datafusion::common::stats::Precision; -use datafusion::common::tree_node::TreeNodeRecursion; +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::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::{ @@ -85,7 +87,8 @@ impl ScanConfig { apply_expression_roots(self.filters.iter(), f) } - /// Writes the scan's dynamic filters into `f` when dynamic filtering is enabled. + /// Writes the scan's dynamic filters into `f` when dynamic filtering is enabled. A join filter + /// can list thousands of keys, so an IN list of more than 5 values prints as its count. pub(crate) fn fmt_dynamic_filters(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { if !self.dynamic_filtering_enabled { return Ok(()); @@ -101,7 +104,21 @@ impl ScanConfig { if dynamic_filters.is_empty() { return Ok(()); } - write!(f, ", dynamic_filters={}", conjunction(dynamic_filters)) + 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).map_err(|_| fmt::Error)?; + let current = count_long_in_lists(current).map_err(|_| fmt::Error)?; + write!(f, "DynamicFilter [ {current} ]")?; + } + Ok(()) } pub(crate) fn open( @@ -158,6 +175,19 @@ impl fmt::Debug for ScanConfig { } } +/// `expr` with each IN list of more than 5 values replaced by text giving its count, for printing. +fn count_long_in_lists(expr: Arc) -> Result> { + expr.transform_up(|expr| match expr.downcast_ref::() { + Some(in_list) if in_list.len() > 5 => { + 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))) + } + _ => Ok(Transformed::no(expr)), + }) + .map(|expr| expr.data) +} + /// Snapshot summary keys defined by the Iceberg table spec: /// https://iceberg.apache.org/spec/#optional-snapshot-summary-fields /// diff --git a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs index 220931842..bd9df65dc 100644 --- a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs +++ b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs @@ -42,6 +42,41 @@ mod tests { Ok(()) } + #[tokio::test] + async fn the_plan_shows_the_length_of_a_long_join_filter() -> Result<()> { + let harness = IcebergTestHarness::builder() + .with_session_extension(|state| state.with_iceberg_integration(Default::default())) + .with_table(create_taxi_table().await?) + .with_config_option("iceberg.eager_file_planning", false) + .with_config_option("iceberg.dynamic_filtering_enabled", true) + .build() + .await?; + + // The join filter lists seven days. The plan shows how many instead of listing them. + let plan = harness + .physical_plan( + "SELECT COUNT(*) AS trips FROM taxi \ + JOIN (VALUES (DATE '2024-01-01'), (DATE '2024-01-02'), (DATE '2024-01-03'), \ + (DATE '2024-01-04'), (DATE '2024-01-05'), (DATE '2024-01-06'), \ + (DATE '2024-01-07')) AS days(day) \ + ON taxi.pickup_date = days.day", + ) + .await?; + collect(Arc::clone(&plan), harness.state().task_ctx()).await?; + + insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" + ProjectionExec: expr=[count(Int64(1))@0 as trips] + AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] + CoalescePartitionsExec + AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] + HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] + ProjectionExec: expr=[column1@0 as day] + DataSourceExec: partitions=1, partition_sizes=[1] + DataSourceExec: format=iceberg, table=default.taxi, projection=[pickup_date], dynamic_filters=DynamicFilter [ pickup_date@0 >= 2024-01-01 AND pickup_date@0 <= 2024-01-07 AND pickup_date@0 IN ([7 elements]) ] + "); + Ok(()) + } + #[tokio::test] async fn eager_planning_workers_get_the_join_filter() -> Result<()> { let harness = IcebergTestHarness::builder() From 08003241590fefa929043f400a708cec669fef66 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:45:49 +0000 Subject: [PATCH 8/9] nit: print every IN list in Iceberg scan dynamic filters as its count Every IN list now prints as its count, so a plan stays short however many keys a join filter lists. The rewrite moves into fmt_dynamic_filters, and the plan test for long lists goes away. --- .../src/common/scan_config.rs | 34 +++++++-------- .../tests/dynamic_filtering.rs | 41 ++----------------- 2 files changed, 21 insertions(+), 54 deletions(-) diff --git a/datafusion/distributed-iceberg/src/common/scan_config.rs b/datafusion/distributed-iceberg/src/common/scan_config.rs index da903c871..99ecb7d99 100644 --- a/datafusion/distributed-iceberg/src/common/scan_config.rs +++ b/datafusion/distributed-iceberg/src/common/scan_config.rs @@ -88,7 +88,7 @@ impl ScanConfig { } /// Writes the scan's dynamic filters into `f` when dynamic filtering is enabled. A join filter - /// can list thousands of keys, so an IN list of more than 5 values prints as its count. + /// 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(()); @@ -114,8 +114,23 @@ impl ScanConfig { write!(f, "DynamicFilter [ empty ]")?; continue; } - let current = snapshot_physical_expr(filter).map_err(|_| fmt::Error)?; - let current = count_long_in_lists(current).map_err(|_| fmt::Error)?; + let current = snapshot_physical_expr(filter) + .and_then(|current| { + current.transform_up(|expr| match expr.downcast_ref::() { + 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(()) @@ -175,19 +190,6 @@ impl fmt::Debug for ScanConfig { } } -/// `expr` with each IN list of more than 5 values replaced by text giving its count, for printing. -fn count_long_in_lists(expr: Arc) -> Result> { - expr.transform_up(|expr| match expr.downcast_ref::() { - Some(in_list) if in_list.len() > 5 => { - 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))) - } - _ => Ok(Transformed::no(expr)), - }) - .map(|expr| expr.data) -} - /// Snapshot summary keys defined by the Iceberg table spec: /// https://iceberg.apache.org/spec/#optional-snapshot-summary-fields /// diff --git a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs index bd9df65dc..09a8d5438 100644 --- a/datafusion/distributed-iceberg/tests/dynamic_filtering.rs +++ b/datafusion/distributed-iceberg/tests/dynamic_filtering.rs @@ -37,42 +37,7 @@ mod tests { ProjectionExec: expr=[column1@0 as day] DataSourceExec: partitions=1, partition_sizes=[1] FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] - DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] - "); - Ok(()) - } - - #[tokio::test] - async fn the_plan_shows_the_length_of_a_long_join_filter() -> Result<()> { - let harness = IcebergTestHarness::builder() - .with_session_extension(|state| state.with_iceberg_integration(Default::default())) - .with_table(create_taxi_table().await?) - .with_config_option("iceberg.eager_file_planning", false) - .with_config_option("iceberg.dynamic_filtering_enabled", true) - .build() - .await?; - - // The join filter lists seven days. The plan shows how many instead of listing them. - let plan = harness - .physical_plan( - "SELECT COUNT(*) AS trips FROM taxi \ - JOIN (VALUES (DATE '2024-01-01'), (DATE '2024-01-02'), (DATE '2024-01-03'), \ - (DATE '2024-01-04'), (DATE '2024-01-05'), (DATE '2024-01-06'), \ - (DATE '2024-01-07')) AS days(day) \ - ON taxi.pickup_date = days.day", - ) - .await?; - collect(Arc::clone(&plan), harness.state().task_ctx()).await?; - - insta::assert_snapshot!(display_plan_ascii(plan.as_ref(), false), @" - ProjectionExec: expr=[count(Int64(1))@0 as trips] - AggregateExec: mode=Final, gby=[], aggr=[count(Int64(1))] - CoalescePartitionsExec - AggregateExec: mode=Partial, gby=[], aggr=[count(Int64(1))] - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(day@0, pickup_date@0)], projection=[] - ProjectionExec: expr=[column1@0 as day] - DataSourceExec: partitions=1, partition_sizes=[1] - DataSourceExec: format=iceberg, table=default.taxi, projection=[pickup_date], dynamic_filters=DynamicFilter [ pickup_date@0 >= 2024-01-01 AND pickup_date@0 <= 2024-01-07 AND pickup_date@0 IN ([7 elements]) ] + DataSourceExec: format=iceberg, table=default.taxi, projection=[trip_distance, pickup_date], predicate=trip_distance >= 2, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN ([1 elements]) ] "); Ok(()) } @@ -118,8 +83,8 @@ mod tests { │ [Stage 1] => NetworkBroadcastExec: partitions_per_consumer=1, stage_partitions=2, input_tasks=1 │ FilterExec: trip_distance@0 >= 2, projection=[pickup_date@1] │ DistributedLeafExec: - │ t0: DataSourceExec: format=iceberg, planned_files=3, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] - │ t1: DataSourceExec: format=iceberg, planned_files=4, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN (SET) ([2024-01-10]) ] + │ t0: DataSourceExec: format=iceberg, planned_files=3, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN ([1 elements]) ] + │ t1: DataSourceExec: format=iceberg, planned_files=4, dynamic_filters=DynamicFilter [ pickup_date@1 >= 2024-01-10 AND pickup_date@1 <= 2024-01-10 AND pickup_date@1 IN ([1 elements]) ] └────────────────────────────────────────────────── ┌───── Stage 1 ── tasks=1, partitions=2 │ BroadcastExec: input_partitions=1, consumer_tasks=2, output_partitions=2 From 1d7e7cc6755589b3aed84a2a3f1ba151b9f85719 Mon Sep 17 00:00:00 2001 From: Anirudh <6716458+Perseus@users.noreply.github.com> Date: Wed, 11 Nov 2026 14:22:31 +0000 Subject: [PATCH 9/9] nit: tidy the Iceberg codecs' filter decoding Removes the comment about decoding filters with the plan's converter, since that is the usual way to decode them. The lazy codec test builds its filter with DataFusion's binary and col helpers. --- .../distributed-iceberg/src/eager/codec.rs | 2 -- .../distributed-iceberg/src/lazy/codec.rs | 19 +++++++------------ 2 files changed, 7 insertions(+), 14 deletions(-) diff --git a/datafusion/distributed-iceberg/src/eager/codec.rs b/datafusion/distributed-iceberg/src/eager/codec.rs index 94befaecc..b447b4d69 100644 --- a/datafusion/distributed-iceberg/src/eager/codec.rs +++ b/datafusion/distributed-iceberg/src/eager/codec.rs @@ -176,8 +176,6 @@ impl EagerIcebergCodec { proto.custom_file_io.as_deref(), ctx, )?; - // Decode with the plan's converter, so a filter keeps sharing its state with the - // operator that fills it in when both run in the same task. let filters = proto .filters .as_ref() diff --git a/datafusion/distributed-iceberg/src/lazy/codec.rs b/datafusion/distributed-iceberg/src/lazy/codec.rs index 47b149911..44c3212b2 100644 --- a/datafusion/distributed-iceberg/src/lazy/codec.rs +++ b/datafusion/distributed-iceberg/src/lazy/codec.rs @@ -103,8 +103,6 @@ impl PhysicalExtensionCodec for LazyIcebergCodec { proto.custom_file_io.as_deref(), ctx, )?; - // Decode with the plan's converter, so a filter keeps sharing its state with the - // operator that fills it in when both run in the same task. let filters = proto .filters .as_ref() @@ -179,8 +177,7 @@ mod tests { use datafusion::common::Statistics; use datafusion::datasource::source::DataSource; use datafusion::logical_expr::Operator; - use datafusion::physical_expr::PhysicalExpr; - use datafusion::physical_expr::expressions::{BinaryExpr, Column, lit}; + 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; @@ -223,16 +220,14 @@ mod tests { Arc::new(TaskContext::default().with_session_config(config.clone())), ) .await?; - let fare_over_ten: Arc = Arc::new(BinaryExpr::new( - Arc::new(Column::new_with_schema( - "fare_amount", - taxi.schema()?.as_ref(), - )?), + let schema = taxi.schema()?; + let filter = binary( + col("fare_amount", &schema)?, Operator::Gt, lit(10.0), - )); - let pushed = - iceberg_source(&plan)?.try_pushdown_filters(vec![fare_over_ten], config.options())?; + &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"); };