diff --git a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs index 3c6f554a940d1..28bdea6075344 100644 --- a/datafusion/core/tests/physical_optimizer/filter_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/filter_pushdown.rs @@ -100,6 +100,30 @@ fn test_pushdown_into_scan() { ); } +#[test] +fn test_inexact_pushdown_retains_filter_exec() { + let scan = TestScanBuilder::new(schema()) + .with_support(true) + .with_inexact_pushdown(true) + .build(); + let predicate = col_lit_predicate("a", "foo", &schema()); + let plan = Arc::new(FilterExec::try_new(predicate, scan).unwrap()); + + insta::assert_snapshot!( + OptimizationTest::new(plan, FilterPushdown::new(), true), + @r" + OptimizationTest: + input: + - FilterExec: a@0 = foo + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true + output: + Ok: + - FilterExec: a@0 = foo + - DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo + " + ); +} + #[test] fn test_pushdown_volatile_functions_not_allowed() { // Test that we do not push down filters with volatile functions @@ -2050,6 +2074,57 @@ fn test_aggregate_dynamic_filter_not_created_for_single_mode() { ); } +#[test] +fn test_aggregate_dynamic_filter_pushdown_consumer_acknowledgement() { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); + + for (supports_pushdown, inexact, expose_expressions, expected_consumer) in [ + (false, false, true, false), + (true, false, true, true), + (true, true, true, true), + (true, false, false, true), + ] { + let scan = TestScanBuilder::new(Arc::clone(&schema)) + .with_support(supports_pushdown) + .with_inexact_pushdown(inexact) + .with_expose_expressions(expose_expressions) + .build(); + let min_expr = + AggregateExprBuilder::new(min_udaf(), vec![col("a", &schema).unwrap()]) + .schema(Arc::clone(&schema)) + .alias("min_a") + .build() + .unwrap(); + let plan: Arc = Arc::new( + AggregateExec::try_new( + AggregateMode::Partial, + PhysicalGroupBy::new_single(vec![]), + vec![min_expr.into()], + vec![None], + scan, + Arc::clone(&schema), + ) + .unwrap(), + ); + + let mut config = ConfigOptions::default(); + config.execution.parquet.pushdown_filters = true; + config.optimizer.enable_dynamic_filter_pushdown = true; + let optimized = FilterPushdown::new_post_optimization() + .optimize(plan, &config) + .unwrap(); + let aggregate = optimized + .downcast_ref::() + .expect("plan should be AggregateExec"); + + assert_eq!( + !aggregate.dynamic_expressions_produced().is_empty(), + expected_consumer, + "producer state should reflect the input's planning result" + ); + } +} + #[test] fn test_pushdown_filter_on_non_first_grouping_column() { // Test that filters on non-first grouping columns are still pushed down @@ -2964,11 +3039,10 @@ async fn test_hashjoin_hash_table_pushdown_collect_left() { ); } -// Not portable to sqllogictest: verifies whether the optimized probe-side plan -// retains the HashJoinExec's dynamic filter expression. The with_support(false) -// branch has no SQL analog because parquet supports filter pushdown. +// Not portable to sqllogictest: verifies the planning-time acknowledgement for +// a dynamic filter, including a consumer hidden behind an opaque plan boundary. #[test] -fn test_hashjoin_dynamic_filter_pushdown_is_used() { +fn test_hashjoin_dynamic_filter_pushdown_consumer_acknowledgement() { fn contains_expression_id(plan: &Arc, expression_id: u64) -> bool { let mut found = false; plan.apply(|node| { @@ -2987,7 +3061,12 @@ fn test_hashjoin_dynamic_filter_pushdown_is_used() { found } - for (probe_supports_pushdown, expected_consumer) in [(false, false), (true, true)] { + for (probe_supports_pushdown, inexact, expose_expressions, expected_consumer) in [ + (false, false, true, false), + (true, false, true, true), + (true, true, true, true), + (true, false, false, true), + ] { let build_side_schema = Arc::new(Schema::new(vec![ Field::new("a", DataType::Utf8, false), Field::new("b", DataType::Utf8, false), @@ -3006,6 +3085,8 @@ fn test_hashjoin_dynamic_filter_pushdown_is_used() { ])); let probe_scan = TestScanBuilder::new(Arc::clone(&probe_side_schema)) .with_support(probe_supports_pushdown) + .with_inexact_pushdown(inexact) + .with_expose_expressions(expose_expressions) .with_batches(vec![ record_batch!( ("a", Utf8, ["aa", "ab", "ac", "ad"]), @@ -3050,17 +3131,22 @@ fn test_hashjoin_dynamic_filter_pushdown_is_used() { .downcast_ref::() .expect("Plan should be HashJoinExec"); let dynamic_filters = hash_join.dynamic_expressions_produced(); - let expression_id = dynamic_filters - .first() - .expect("Dynamic filter should be created") - .expression_id() - .expect("Dynamic filters always have an expression ID"); - assert_eq!( - contains_expression_id(hash_join.right(), expression_id), + !dynamic_filters.is_empty(), expected_consumer, - "probe consumer should be {expected_consumer} when pushdown support is {probe_supports_pushdown}" + "producer state should reflect the probe's planning result" ); + + if let Some(dynamic_filter) = dynamic_filters.first() { + let expression_id = dynamic_filter + .expression_id() + .expect("dynamic filters always have an expression ID"); + assert_eq!( + contains_expression_id(hash_join.right(), expression_id), + expose_expressions, + "expression visibility should follow the test source setting" + ); + } } } diff --git a/datafusion/core/tests/physical_optimizer/pushdown_utils.rs b/datafusion/core/tests/physical_optimizer/pushdown_utils.rs index e11573b95fd09..36d835a90d6ee 100644 --- a/datafusion/core/tests/physical_optimizer/pushdown_utils.rs +++ b/datafusion/core/tests/physical_optimizer/pushdown_utils.rs @@ -103,6 +103,8 @@ impl FileOpener for TestOpener { #[derive(Clone)] pub struct TestSource { support: bool, + inexact: bool, + expose_expressions: bool, predicate: Option>, batch_size: Option, batches: Vec, @@ -116,6 +118,8 @@ impl TestSource { let table_schema = datafusion_datasource::TableSchema::from(schema); Self { support, + inexact: false, + expose_expressions: true, metrics: ExecutionPlanMetricsSet::new(), batches, predicate: None, @@ -198,13 +202,18 @@ impl FileSource for TestSource { ), ..self.clone() }); - Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::Yes; filters.len()], - ) + Ok(FilterPushdownPropagation::with_parent_pushdown_result(vec![ + if self.inexact { + PushedDown::Inexact + } else { + PushedDown::Exact + }; + filters.len() + ]) .with_updated_node(new_node)) } else { Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], + vec![PushedDown::Unsupported; filters.len()], )) } } @@ -241,6 +250,9 @@ impl FileSource for TestSource { &self, f: &mut dyn FnMut(&Arc) -> Result, ) -> Result { + if !self.expose_expressions { + return Ok(TreeNodeRecursion::Continue); + } datafusion_physical_plan::apply_expression_roots( self.predicate.iter().chain( self.projection @@ -256,6 +268,8 @@ impl FileSource for TestSource { #[derive(Debug, Clone)] pub struct TestScanBuilder { support: bool, + inexact: bool, + expose_expressions: bool, batches: Vec, schema: SchemaRef, } @@ -264,6 +278,8 @@ impl TestScanBuilder { pub fn new(schema: SchemaRef) -> Self { Self { support: false, + inexact: false, + expose_expressions: true, batches: vec![], schema, } @@ -274,17 +290,30 @@ impl TestScanBuilder { self } + /// Have a supporting source report retained predicates as inexact. + pub fn with_inexact_pushdown(mut self, inexact: bool) -> Self { + self.inexact = inexact; + self + } + + /// Control whether the test source exposes retained expressions through + /// `apply_expressions`, allowing tests to model an opaque remote boundary. + pub fn with_expose_expressions(mut self, expose_expressions: bool) -> Self { + self.expose_expressions = expose_expressions; + self + } + pub fn with_batches(mut self, batches: Vec) -> Self { self.batches = batches; self } pub fn build(self) -> Arc { - let source = Arc::new(TestSource::new( - Arc::clone(&self.schema), - self.support, - self.batches, - )); + let mut source = + TestSource::new(Arc::clone(&self.schema), self.support, self.batches); + source.inexact = self.inexact; + source.expose_expressions = self.expose_expressions; + let source = Arc::new(source); let base_config = FileScanConfigBuilder::new(ObjectStoreUrl::parse("test://").unwrap(), source) .with_file(PartitionedFile::new("test.parquet", 123)) @@ -554,8 +583,8 @@ impl ExecutionPlan for TestNode { let first_pushdown_result = self_pushdown_result[0].clone(); match &first_pushdown_result.discriminant { - PushedDown::No => { - // We have a filter to push down + PushedDown::Inexact | PushedDown::Unsupported => { + // The filter was not applied exactly, so we still apply it. let new_child = FilterExec::try_new( Arc::clone(&first_pushdown_result.predicate), Arc::clone(&self.input), @@ -567,7 +596,7 @@ impl ExecutionPlan for TestNode { res.updated_node = Some(Arc::new(new_self) as Arc); Ok(res) } - PushedDown::Yes => { + PushedDown::Exact => { let res = FilterPushdownPropagation::if_all(child_pushdown_result); Ok(res) } diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b3ce024d66f1f..1cd25ae094bc0 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -152,7 +152,7 @@ impl VirtualColumnsState { /// the predicate references a virtual column. The contract is that callers /// route filters through /// [`ParquetSource::try_pushdown_filters`](crate::source::ParquetSource), -/// which classifies virtual-col filters as `PushedDown::No`. Erroring here +/// which classifies virtual-col filters as `PushedDown::Unsupported`. Erroring here /// prevents silent wrong results for callers that bypass that path and set /// the predicate directly on `ParquetSource`. /// diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 097b4563af5df..1b00c0e76016d 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -853,19 +853,20 @@ impl FileSource for ParquetSource { .collect(); if filters .iter() - .all(|f| matches!(f.discriminant, PushedDown::No)) + .all(|f| matches!(f.discriminant, PushedDown::Unsupported)) { // No filters can be pushed down, so we can just return the remaining filters // and avoid replacing the source in the physical plan. return Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], + vec![PushedDown::Unsupported; filters.len()], )); } let allowed_filters = filters .iter() .filter_map(|f| match f.discriminant { - PushedDown::Yes => Some(Arc::clone(&f.predicate)), - PushedDown::No => None, + PushedDown::Exact => Some(Arc::clone(&f.predicate)), + PushedDown::Inexact => Some(Arc::clone(&f.predicate)), + PushedDown::Unsupported => None, }) .collect_vec(); let predicate = match source.predicate { @@ -877,13 +878,21 @@ impl FileSource for ParquetSource { source.predicate = Some(predicate); source = source.with_pushdown_filters(pushdown_filters); let source = Arc::new(source); - // If pushdown_filters is false we tell our parents that they still have to handle the filters, - // even if we updated the predicate to include the filters (they will only be used for stats pruning). + // If pushdown_filters is false, retained filters are used only for + // statistics pruning. Report them as inexact so an ancestor still + // evaluates them while dynamic-filter producers know there is a consumer. if !pushdown_filters { - return Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], - ) - .with_updated_node(source)); + let results = filters + .iter() + .map(|filter| match filter.discriminant { + PushedDown::Exact | PushedDown::Inexact => PushedDown::Inexact, + PushedDown::Unsupported => PushedDown::Unsupported, + }) + .collect(); + return Ok( + FilterPushdownPropagation::with_parent_pushdown_result(results) + .with_updated_node(source), + ); } Ok(FilterPushdownPropagation::with_parent_pushdown_result( filters.iter().map(|f| f.discriminant).collect(), @@ -2037,7 +2046,7 @@ mod tests { fn test_try_pushdown_filters_rejects_virtual_column_refs() { // Virtual columns are produced by the reader and cannot be referenced // inside a RowFilter. `try_pushdown_filters` must report such filters - // as `PushedDown::No` so the FilterExec above the scan stays in + // as `PushedDown::Unsupported` so the FilterExec above the scan stays in // place — otherwise the scan would silently drop the predicate and // produce wrong results. use arrow::datatypes::{DataType, Field, FieldRef, Schema}; @@ -2091,22 +2100,30 @@ mod tests { assert_eq!(prop.filters.len(), 4); assert!( - matches!(prop.filters[0], PushedDown::Yes), - "file-column filter should be pushable" + matches!(prop.filters[0], PushedDown::Exact), + "file-column filter should be applied exactly" ); assert!( - matches!(prop.filters[1], PushedDown::No), + matches!(prop.filters[1], PushedDown::Unsupported), "filter referencing only a virtual column must not be pushed down" ); assert!( - matches!(prop.filters[2], PushedDown::No), + matches!(prop.filters[2], PushedDown::Unsupported), "filter mixing a virtual column with a file column must not be \ pushed down (row filter would silently drop it)" ); assert!( - matches!(prop.filters[3], PushedDown::No), + matches!(prop.filters[3], PushedDown::Unsupported), "file_row_index() rewrites to a virtual column and must not be \ pushed down" ); + + let pruning_only_source = ParquetSource::new(source.table_schema().clone()); + let pruning_filter = + logical2physical(&col("value").eq(logical_lit(1i64)), full_schema); + let prop = pruning_only_source + .try_pushdown_filters(vec![pruning_filter], &config) + .expect("statistics-only pushdown must not error"); + assert_eq!(prop.filters, vec![PushedDown::Inexact]); } } diff --git a/datafusion/datasource/src/file.rs b/datafusion/datasource/src/file.rs index f1a94f2e12363..4a5a9ff093b5d 100644 --- a/datafusion/datasource/src/file.rs +++ b/datafusion/datasource/src/file.rs @@ -187,13 +187,15 @@ pub trait FileSource: Any + Send + Sync { /// plus partition columns), before any projection is applied. /// /// Any filters that this FileSource chooses to evaluate itself should be - /// returned as `PushedDown::Yes` in the result, along with a FileSource + /// returned as `PushedDown::Exact` in the result, along with a FileSource /// instance that incorporates those filters. Such filters are logically /// applied "during" the file scan, meaning they may refer to columns not /// included in the final output projection. /// - /// Filters that cannot be pushed down should be marked as `PushedDown::No`, - /// and will be evaluated by an execution plan after the file source. + /// Filters retained only for pruning should be marked as + /// `PushedDown::Inexact`. Filters that are not retained at all should be + /// marked as `PushedDown::Unsupported`. Both will be evaluated by an + /// execution plan after the file source. /// /// See [`ExecutionPlan::handle_child_pushdown_result`] for more details. /// @@ -204,7 +206,7 @@ pub trait FileSource: Any + Send + Sync { _config: &ConfigOptions, ) -> Result>> { Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], + vec![PushedDown::Unsupported; filters.len()], )) } diff --git a/datafusion/datasource/src/source.rs b/datafusion/datasource/src/source.rs index 741010c595197..d890be577fb1d 100644 --- a/datafusion/datasource/src/source.rs +++ b/datafusion/datasource/src/source.rs @@ -190,6 +190,11 @@ pub trait DataSource: Any + Send + Sync + Debug { /// [`Self::eq_properties`] and output of any projections pushed into the /// source), not the original table schema. /// + /// Return [`PushedDown::Exact`] for predicates applied as row filters, + /// [`PushedDown::Inexact`] for predicates retained only for uses such as + /// statistics pruning, and [`PushedDown::Unsupported`] for predicates that + /// are not retained at all. + /// /// See [`ExecutionPlan::handle_child_pushdown_result`] for more details. /// /// [`ExecutionPlan::handle_child_pushdown_result`]: datafusion_physical_plan::ExecutionPlan::handle_child_pushdown_result @@ -199,7 +204,7 @@ pub trait DataSource: Any + Send + Sync + Debug { _config: &ConfigOptions, ) -> Result>> { Ok(FilterPushdownPropagation::with_parent_pushdown_result( - vec![PushedDown::No; filters.len()], + vec![PushedDown::Unsupported; filters.len()], )) } diff --git a/datafusion/physical-optimizer/src/filter_pushdown.rs b/datafusion/physical-optimizer/src/filter_pushdown.rs index 18fe151000511..c395c2f4a6935 100644 --- a/datafusion/physical-optimizer/src/filter_pushdown.rs +++ b/datafusion/physical-optimizer/src/filter_pushdown.rs @@ -501,10 +501,11 @@ fn push_down_filters( let num_self_filters = self_filtered.len(); let mut all_predicates = self_filtered.items().to_vec(); - // Apply second filter pass: collect indices of parent filters that can be pushed down - let parent_filters_for_child = parent_filtered - .chain_filter_slice(&parent_filters, |filter| { - matches!(filter.discriminant, PushedDown::Yes) + // Apply second filter pass: collect parent filters that this node can + // retain either exactly or inexactly. + let parent_filters_for_child = + parent_filtered.chain_filter_slice(&parent_filters, |filter| { + matches!(filter.discriminant, PushedDown::Exact | PushedDown::Inexact) }); // Add the filtered parent predicates to all_predicates @@ -542,7 +543,7 @@ fn push_down_filters( .collect_vec(); // Map the results from filtered self filters back to their original positions using FilteredVec let mapped_self_results = - self_filtered.map_results_to_original(all_filters, PushedDown::No); + self_filtered.map_results_to_original(all_filters, PushedDown::Unsupported); // Wrap each result with its corresponding expression let self_filter_results: Vec<_> = mapped_self_results @@ -555,7 +556,7 @@ fn push_down_filters( // Start by marking all parent filters as unsupported for this child for parent_filter_pushdown_support in parent_filter_pushdown_supports.iter_mut() { - parent_filter_pushdown_support.push(PushedDown::No); + parent_filter_pushdown_support.push(PushedDown::Unsupported); assert_eq!( parent_filter_pushdown_support.len(), child_idx + 1, @@ -564,7 +565,7 @@ fn push_down_filters( } // Map results from pushed-down filters back to original parent filter indices let mapped_parent_results = parent_filters_for_child - .map_results_to_original(parent_filters, PushedDown::No); + .map_results_to_original(parent_filters, PushedDown::Unsupported); // Update parent_filter_pushdown_supports with the mapped results // mapped_parent_results already has the results at their original indices diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs index 0f1b718a3c8e3..eba5177c024c6 100644 --- a/datafusion/physical-plan/src/aggregates/mod.rs +++ b/datafusion/physical-plan/src/aggregates/mod.rs @@ -158,12 +158,10 @@ use crate::aggregates::{ partial_reduce_stream::PartialReduceHashAggregateStream, single_stream::SingleHashAggregateStream, }; -use crate::execution_plan::{ - CardinalityEffect, EmissionType, plan_contains_expression_id, -}; +use crate::execution_plan::{CardinalityEffect, EmissionType}; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, + FilterPushdownPropagation, PushedDown, }; use crate::metrics::{ExecutionPlanMetricsSet, MetricsSet}; use crate::statistics::{ChildStats, StatisticsArgs}; @@ -2274,17 +2272,17 @@ impl ExecutionPlan for AggregateExec { ) -> Result>> { let mut result = FilterPushdownPropagation::if_any(child_pushdown_result.clone()); - // If this node tried to pushdown some dynamic filter before, now we check - // if the child accept the filter - if phase == FilterPushdownPhase::Post - && let Some(dyn_filter) = &self.dynamic_filter - { - let child_accepts_dyn_filter = dyn_filter - .filter - .expression_id() - .map(|id| plan_contains_expression_id(&self.input, id)) - .transpose()? - .unwrap_or(false); + // Use the child's planning result rather than inspecting the rewritten + // child plan. This also works when the consumer is hidden behind an + // opaque plan boundary. + if phase == FilterPushdownPhase::Post && self.dynamic_filter.is_some() { + let child_accepts_dyn_filter = child_pushdown_result + .self_filters + .first() + .and_then(|filters| filters.first()) + .is_some_and(|filter| { + !matches!(filter.discriminant, PushedDown::Unsupported) + }); if !child_accepts_dyn_filter { // Child can't consume the self dynamic filter, so disable it by setting @@ -3202,7 +3200,6 @@ mod tests { use crate::empty::EmptyExec; use crate::execution_plan::Boundedness; use crate::expressions::col; - use crate::filter::FilterExecBuilder; use crate::metrics::MetricValue; use crate::statistics::{StatisticsArgs, StatisticsContext}; use crate::test::TestMemoryExec; @@ -3238,7 +3235,7 @@ mod tests { use datafusion_physical_expr::Partitioning; use datafusion_physical_expr::PhysicalSortExpr; use datafusion_physical_expr::aggregate::AggregateExprBuilder; - use datafusion_physical_expr::expressions::{Literal, NotExpr}; + use datafusion_physical_expr::expressions::Literal; use crate::projection::ProjectionExec; use crate::repartition::RepartitionExec; @@ -8054,38 +8051,6 @@ mod tests { Ok(()) } - #[test] - fn test_plan_contains_expression_id_recurses_plans_and_expressions() -> Result<()> { - let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int64, false)])); - let empty: Arc = Arc::new(EmptyExec::new(Arc::clone(&schema))); - let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new( - vec![col("a", &schema)?], - lit(true), - )); - let expression_id = dynamic_filter - .expression_id() - .expect("dynamic filters always have an expression ID"); - - assert!(!plan_contains_expression_id(&empty, expression_id)?); - - let dynamic_filter_expr: Arc = - Arc::::clone(&dynamic_filter); - let predicate: Arc = - Arc::new(NotExpr::new(dynamic_filter_expr)); - let filter: Arc = - Arc::new(FilterExecBuilder::new(predicate, empty).build()?); - let projection: Arc = Arc::new(ProjectionExec::try_new( - [ProjectionExpr::new_from_expression( - col("a", &schema)?, - &schema, - )?], - filter, - )?); - - assert!(plan_contains_expression_id(&projection, expression_id)?); - Ok(()) - } - /// Test that [`AggregateExec::with_dynamic_filter_expr`] errors when the aggregate does not support dynamic filtering #[test] fn test_with_dynamic_filter_error_unsupported() -> Result<()> { diff --git a/datafusion/physical-plan/src/execution_plan.rs b/datafusion/physical-plan/src/execution_plan.rs index a4d081b3d9e75..f074438027697 100644 --- a/datafusion/physical-plan/src/execution_plan.rs +++ b/datafusion/physical-plan/src/execution_plan.rs @@ -884,8 +884,8 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { /// - A `FilterExec` may absorb any filters its children could not handle, /// combining them with its own predicate. If no filters remain (i.e., the /// predicate becomes trivially true), it may remove itself from the plan - /// altogether. It typically marks parent filters as supported, indicating - /// they have been handled. + /// altogether. It typically marks parent filters as exact, indicating + /// they have been fully handled. /// - A `HashJoinExec` might ignore the pushdown result if filters need to /// be applied during the join operation. It passes the parent filters back /// up wrapped in [`FilterPushdownPropagation::if_any`], discarding @@ -907,7 +907,7 @@ pub trait ExecutionPlan: Any + Debug + DisplayAs + Send + Sync { /// remaining filters during the join, and passes unhandled filters back /// up to `FilterExec`. `FilterExec` absorbs any unhandled filters, /// updates its predicate if necessary, or removes itself if the predicate - /// becomes trivial (e.g., `lit(true)`), and marks filters as supported + /// becomes trivial (e.g., `lit(true)`), and marks filters as exact /// for its parent. /// /// The default implementation is a no-op that passes the result of pushdown @@ -1118,36 +1118,6 @@ where Ok(TreeNodeRecursion::Continue) } -/// Returns whether `plan` contains a physical expression with `expression_id`. -/// -/// This traverses both the execution plan and the children of each expression root -/// reported by [`ExecutionPlan::apply_expressions`]. -pub(crate) fn plan_contains_expression_id( - plan: &Arc, - expression_id: u64, -) -> Result { - let mut found = false; - plan.apply(|node| { - node.apply_expressions(&mut |root| { - root.apply(|expr| { - if expr.expression_id() == Some(expression_id) { - found = true; - Ok(TreeNodeRecursion::Stop) - } else { - Ok(TreeNodeRecursion::Continue) - } - }) - })?; - - Ok(if found { - TreeNodeRecursion::Stop - } else { - TreeNodeRecursion::Continue - }) - })?; - Ok(found) -} - impl dyn ExecutionPlan { /// Returns `true` if the plan is of type `T`. /// diff --git a/datafusion/physical-plan/src/filter.rs b/datafusion/physical-plan/src/filter.rs index 414e5a6d8586a..aa12fbd6a119b 100644 --- a/datafusion/physical-plan/src/filter.rs +++ b/datafusion/physical-plan/src/filter.rs @@ -713,7 +713,8 @@ impl ExecutionPlan for FilterExec { .parent_filters .iter() .filter_map(|f| { - matches!(f.all(), PushedDown::No).then_some(Arc::clone(&f.filter)) + (!matches!(f.all(), PushedDown::Exact)) + .then_some(Arc::clone(&f.filter)) }) .collect(); @@ -734,8 +735,8 @@ impl ExecutionPlan for FilterExec { .expect("we have exactly one child") .iter() .filter_map(|f| match f.discriminant { - PushedDown::Yes => None, - PushedDown::No => Some(&f.predicate), + PushedDown::Exact => None, + PushedDown::Inexact | PushedDown::Unsupported => Some(&f.predicate), }) .cloned(); @@ -810,7 +811,7 @@ impl ExecutionPlan for FilterExec { }; Ok(FilterPushdownPropagation { - filters: vec![PushedDown::Yes; child_pushdown_result.parent_filters.len()], + filters: vec![PushedDown::Exact; child_pushdown_result.parent_filters.len()], updated_node, }) } diff --git a/datafusion/physical-plan/src/filter_pushdown.rs b/datafusion/physical-plan/src/filter_pushdown.rs index 382967c7ee1ef..98b56c42602c3 100644 --- a/datafusion/physical-plan/src/filter_pushdown.rs +++ b/datafusion/physical-plan/src/filter_pushdown.rs @@ -24,7 +24,7 @@ //! 2. **Optimizer Executes Pushdown**: The optimizer recursively pushes down filters for each child, //! passing the appropriate filters (`Vec>`) for that child. //! 3. **Optimizer Gathers Results**: The optimizer collects [`FilterPushdownPropagation`] results from children, -//! containing information about which filters were successfully pushed down vs. unsupported. +//! distinguishing exact filtering, inexact use, and unsupported predicates. //! 4. **Parent Responds**: The optimizer calls [`ExecutionPlan::handle_child_pushdown_result`] on the parent, //! passing a [`ChildPushdownResult`] containing the aggregated pushdown outcomes. The parent decides //! how to handle filters that couldn't be pushed down (e.g., keep them as FilterExec nodes). @@ -99,15 +99,30 @@ pub struct PushedDownPredicate { } impl PushedDownPredicate { - /// Return the wrapped [`PhysicalExpr`], discarding whether it is supported or unsupported. + /// Return the wrapped [`PhysicalExpr`], discarding its pushdown result. pub fn into_inner(self) -> Arc { self.predicate } - /// Create a new [`PushedDownPredicate`] with supported pushdown. + /// Create a new [`PushedDownPredicate`] with exact pushdown. + pub fn exact(predicate: Arc) -> Self { + Self { + discriminant: PushedDown::Exact, + predicate, + } + } + + /// Create a new [`PushedDownPredicate`] with exact pushdown. + /// + /// This is an alias for [`Self::exact`] retained for compatibility. pub fn supported(predicate: Arc) -> Self { + Self::exact(predicate) + } + + /// Create a new [`PushedDownPredicate`] with inexact pushdown. + pub fn inexact(predicate: Arc) -> Self { Self { - discriminant: PushedDown::Yes, + discriminant: PushedDown::Inexact, predicate, } } @@ -115,35 +130,46 @@ impl PushedDownPredicate { /// Create a new [`PushedDownPredicate`] with unsupported pushdown. pub fn unsupported(predicate: Arc) -> Self { Self { - discriminant: PushedDown::No, + discriminant: PushedDown::Unsupported, predicate, } } } /// Discriminant for the result of pushing down a filter into a child node. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum PushedDown { - /// The predicate was successfully pushed down into the child node. - Yes, - /// The predicate could not be pushed down into the child node. - No, + /// The predicate is retained and applied exactly as a [`FilterExec`] would + /// apply it, so an ancestor can safely stop evaluating the predicate. + /// + /// [`FilterExec`]: crate::filter::FilterExec + Exact, + /// The predicate is retained and used, but is not applied exactly. For + /// example, a data source may use it only for statistics-based pruning. + /// An ancestor must still evaluate the predicate for correctness. + Inexact, + /// The predicate is not retained or used by the child node. + Unsupported, } impl PushedDown { - /// Logical AND operation: returns `Yes` only if both operands are `Yes`. + /// Logical AND operation, returning the least capable result. pub fn and(self, other: PushedDown) -> PushedDown { match (self, other) { - (PushedDown::Yes, PushedDown::Yes) => PushedDown::Yes, - _ => PushedDown::No, + (PushedDown::Exact, PushedDown::Exact) => PushedDown::Exact, + (PushedDown::Unsupported, _) | (_, PushedDown::Unsupported) => { + PushedDown::Unsupported + } + _ => PushedDown::Inexact, } } - /// Logical OR operation: returns `Yes` if either operand is `Yes`. + /// Logical OR operation, returning the most capable result. pub fn or(self, other: PushedDown) -> PushedDown { match (self, other) { - (PushedDown::Yes, _) | (_, PushedDown::Yes) => PushedDown::Yes, - (PushedDown::No, PushedDown::No) => PushedDown::No, + (PushedDown::Exact, _) | (_, PushedDown::Exact) => PushedDown::Exact, + (PushedDown::Inexact, _) | (_, PushedDown::Inexact) => PushedDown::Inexact, + (PushedDown::Unsupported, PushedDown::Unsupported) => PushedDown::Unsupported, } } @@ -165,30 +191,28 @@ pub struct ChildFilterPushdownResult { impl ChildFilterPushdownResult { /// Combine all child results using OR logic. - /// Returns `Yes` if **any** child supports the filter. - /// Returns `No` if **all** children reject the filter or if there are no children. + /// Returns the most capable result reported by any child, or + /// [`PushedDown::Unsupported`] if there are no children. pub fn any(&self) -> PushedDown { if self.child_results.is_empty() { - // If there are no children, filters cannot be supported - PushedDown::No + PushedDown::Unsupported } else { self.child_results .iter() - .fold(PushedDown::No, |acc, result| acc.or(*result)) + .fold(PushedDown::Unsupported, |acc, result| acc.or(*result)) } } /// Combine all child results using AND logic. - /// Returns `Yes` if **all** children support the filter. - /// Returns `No` if **any** child rejects the filter or if there are no children. + /// Returns the least capable result reported by any child, or + /// [`PushedDown::Unsupported`] if there are no children. pub fn all(&self) -> PushedDown { if self.child_results.is_empty() { - // If there are no children, filters cannot be supported - PushedDown::No + PushedDown::Unsupported } else { self.child_results .iter() - .fold(PushedDown::Yes, |acc, result| acc.and(*result)) + .fold(PushedDown::Exact, |acc, result| acc.and(*result)) } } } @@ -205,7 +229,7 @@ pub struct ChildPushdownResult { /// The parent filters that were pushed down as received by the current node when [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) was called. /// Note that this may *not* be the same as the filters that were passed to the children as the current node may have modified them /// (e.g. by reassigning column indices) when it returned them from [`ExecutionPlan::gather_filters_for_pushdown`](crate::ExecutionPlan::handle_child_pushdown_result) in a [`FilterDescription`]. - /// Attached to each filter is a [`PushedDown`] *per child* that indicates whether the filter was supported or unsupported by each child. + /// Attached to each filter is a [`PushedDown`] *per child* that indicates whether the filter was applied exactly, retained for inexact use, or unsupported by each child. /// To get combined results see [`ChildFilterPushdownResult::any`] and [`ChildFilterPushdownResult::all`]. pub parent_filters: Vec, /// The result of pushing down each filter this node provided into each of it's children. @@ -266,7 +290,7 @@ impl FilterPushdownPropagation { let filters = child_pushdown_result .parent_filters .into_iter() - .map(|_| PushedDown::No) + .map(|_| PushedDown::Unsupported) .collect(); Self { filters, @@ -295,7 +319,7 @@ impl FilterPushdownPropagation { /// Describes filter pushdown for a single child node. /// /// This structure contains two types of filters: -/// - **Parent filters**: Filters received from the parent node, marked as supported or unsupported +/// - **Parent filters**: Filters received from the parent node, marked as exact, inexact, or unsupported /// - **Self filters**: Filters generated by the current node to be pushed down to this child #[derive(Debug, Clone)] pub struct ChildFilterDescription { @@ -556,3 +580,48 @@ impl FilterDescription { .collect() } } + +#[cfg(test)] +mod tests { + use super::*; + use datafusion_physical_expr::expressions::lit; + + #[test] + fn pushed_down_and_or_truth_tables() { + use PushedDown::{Exact, Inexact, Unsupported}; + + let cases = [ + (Exact, Exact, Exact, Exact), + (Exact, Inexact, Inexact, Exact), + (Exact, Unsupported, Unsupported, Exact), + (Inexact, Exact, Inexact, Exact), + (Inexact, Inexact, Inexact, Inexact), + (Inexact, Unsupported, Unsupported, Inexact), + (Unsupported, Exact, Unsupported, Exact), + (Unsupported, Inexact, Unsupported, Inexact), + (Unsupported, Unsupported, Unsupported, Unsupported), + ]; + + for (left, right, expected_and, expected_or) in cases { + assert_eq!(left.and(right), expected_and); + assert_eq!(left.or(right), expected_or); + } + } + + #[test] + fn child_results_combine_and_handle_empty_children() { + use PushedDown::{Exact, Inexact, Unsupported}; + + let result = |child_results| ChildFilterPushdownResult { + filter: lit(true), + child_results, + }; + + assert_eq!(result(vec![]).all(), Unsupported); + assert_eq!(result(vec![]).any(), Unsupported); + assert_eq!(result(vec![Exact, Inexact]).all(), Inexact); + assert_eq!(result(vec![Exact, Inexact]).any(), Exact); + assert_eq!(result(vec![Inexact, Unsupported]).all(), Unsupported); + assert_eq!(result(vec![Inexact, Unsupported]).any(), Inexact); + } +} diff --git a/datafusion/physical-plan/src/joins/hash_join/exec.rs b/datafusion/physical-plan/src/joins/hash_join/exec.rs index aa06015be7137..91e4d589a682b 100644 --- a/datafusion/physical-plan/src/joins/hash_join/exec.rs +++ b/datafusion/physical-plan/src/joins/hash_join/exec.rs @@ -24,11 +24,11 @@ use std::vec; use crate::execution_plan::{ EmissionType, boundedness_from_children, has_same_children_properties, - plan_contains_expression_id, stub_properties, + stub_properties, }; use crate::filter_pushdown::{ ChildFilterDescription, ChildPushdownResult, FilterDescription, FilterPushdownPhase, - FilterPushdownPropagation, + FilterPushdownPropagation, PushedDown, }; use crate::joins::Map; use crate::joins::array_map::ArrayMap; @@ -1449,20 +1449,11 @@ impl ExecutionPlan for HashJoinExec { consider using CoalescePartitionsExec or the EnforceDistribution rule" ); - // Only compute a dynamic filter when the probe subtree contains a consumer. - // Searching from `self` would always find the producer expression owned by this join. - let enable_dynamic_filter_pushdown = if self + // `dynamic_filter` is installed during filter pushdown only when the + // probe side reports that it retained the expression. + let enable_dynamic_filter_pushdown = self .allow_join_dynamic_filter_pushdown(context.session_config().options()) - { - self.dynamic_filter - .as_ref() - .and_then(|df| df.filter.expression_id()) - .map(|id| plan_contains_expression_id(&self.right, id)) - .transpose()? - .unwrap_or(false) - } else { - false - }; + && self.dynamic_filter.is_some(); let join_metrics = BuildProbeJoinMetrics::new(partition, &self.metrics); @@ -1813,14 +1804,16 @@ impl ExecutionPlan for HashJoinExec { assert_eq!(child_pushdown_result.self_filters.len(), 2); // Should always be 2, we have 2 children let right_child_self_filters = &child_pushdown_result.self_filters[1]; // We only push down filters to the right child // We expect 0 or 1 self filters - if let Some(filter) = right_child_self_filters.first() { - // Note that we don't check PushdDownPredicate::discrimnant because even if nothing said - // "yes, I can fully evaluate this filter" things might still use it for statistics -> it's worth updating + if let Some(filter) = right_child_self_filters + .first() + .filter(|filter| !matches!(filter.discriminant, PushedDown::Unsupported)) + { let predicate = Arc::clone(&filter.predicate); if let Ok(dynamic_filter) = Arc::downcast::(predicate) { - // We successfully pushed down our self filter - we need to make a new node with the dynamic filter + // The probe side retained the self filter, either for exact + // filtering or inexact pruning, so keep its producer state. let new_node = self .builder() .with_dynamic_filter(Some(HashJoinExecDynamicFilter { diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index cf362cdee55d3..2e1e8c9bd04ac 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -1834,7 +1834,7 @@ mod tests { format!("{}", expected_filter) ); // Verify the predicate was actually pushed down - assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes)); + assert!(matches!(pushed_filters[0].discriminant, PushedDown::Exact)); Ok(()) } @@ -1913,8 +1913,8 @@ mod tests { format!("{}", expected_filter2) ); // Verify the predicates were actually pushed down - assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes)); - assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes)); + assert!(matches!(pushed_filters[0].discriminant, PushedDown::Exact)); + assert!(matches!(pushed_filters[1].discriminant, PushedDown::Exact)); Ok(()) } @@ -1979,8 +1979,8 @@ mod tests { assert_eq!(format!("{}", pushed_filters[0].predicate), expected_filter1); assert_eq!(format!("{}", pushed_filters[1].predicate), expected_filter2); // Verify the predicates were actually pushed down - assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes)); - assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes)); + assert!(matches!(pushed_filters[0].discriminant, PushedDown::Exact)); + assert!(matches!(pushed_filters[1].discriminant, PushedDown::Exact)); Ok(()) } @@ -2044,8 +2044,8 @@ mod tests { assert_eq!(format!("{}", pushed_filters[0].predicate), expected_filter1); assert_eq!(format!("{}", pushed_filters[1].predicate), expected_filter2); // Verify the predicates were actually pushed down - assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes)); - assert!(matches!(pushed_filters[1].discriminant, PushedDown::Yes)); + assert!(matches!(pushed_filters[0].discriminant, PushedDown::Exact)); + assert!(matches!(pushed_filters[1].discriminant, PushedDown::Exact)); Ok(()) } @@ -2089,7 +2089,7 @@ mod tests { // expand to `a + 1 > 10` let pushed_filters = &description.parent_filters()[0]; - assert!(matches!(pushed_filters[0].discriminant, PushedDown::Yes)); + assert!(matches!(pushed_filters[0].discriminant, PushedDown::Exact)); assert_eq!(format!("{}", pushed_filters[0].predicate), "a@0 + 1 > 10"); Ok(()) @@ -2130,7 +2130,10 @@ mod tests { )?; let pushed_filters = &description.parent_filters()[0]; - assert!(matches!(pushed_filters[0].discriminant, PushedDown::No)); + assert!(matches!( + pushed_filters[0].discriminant, + PushedDown::Unsupported + )); // The column shouldn't be found in the alias map, so it remains unchanged with its index assert_eq!( format!("{}", pushed_filters[0].predicate), diff --git a/datafusion/physical-plan/src/sorts/sort.rs b/datafusion/physical-plan/src/sorts/sort.rs index bd47668cf38ba..7f03e2d179335 100644 --- a/datafusion/physical-plan/src/sorts/sort.rs +++ b/datafusion/physical-plan/src/sorts/sort.rs @@ -1571,7 +1571,7 @@ impl ExecutionPlan for SortExec { let unsupported_filters: Vec> = child_pushdown_result .parent_filters .iter() - .filter(|&f| matches!(f.all(), PushedDown::No)) + .filter(|&f| !matches!(f.all(), PushedDown::Exact)) .map(|f| Arc::clone(&f.filter)) .collect(); @@ -1593,7 +1593,7 @@ impl ExecutionPlan for SortExec { ) as Arc; Ok(FilterPushdownPropagation { - filters: vec![PushedDown::Yes; child_pushdown_result.parent_filters.len()], + filters: vec![PushedDown::Exact; child_pushdown_result.parent_filters.len()], updated_node: Some(new_sort), }) } @@ -3905,7 +3905,7 @@ mod tests { // Sort with fetch (TopK) must not allow filters to be pushed below it. assert!(matches!( desc.parent_filters()[0][0].discriminant, - PushedDown::No + PushedDown::Unsupported )); Ok(()) } @@ -3921,7 +3921,7 @@ mod tests { // Plain sort (no fetch) is filter-commutative. assert!(matches!( desc.parent_filters()[0][0].discriminant, - PushedDown::Yes + PushedDown::Exact )); Ok(()) } @@ -3941,7 +3941,7 @@ mod tests { // Parent filters are still blocked in the Post phase. assert!(matches!( desc.parent_filters()[0][0].discriminant, - PushedDown::No + PushedDown::Unsupported )); // But the TopK self-filter should be pushed down. assert_eq!(desc.self_filters()[0].len(), 1); diff --git a/datafusion/physical-plan/src/union.rs b/datafusion/physical-plan/src/union.rs index 160772dc22314..fceef8329f30a 100644 --- a/datafusion/physical-plan/src/union.rs +++ b/datafusion/physical-plan/src/union.rs @@ -527,7 +527,7 @@ impl ExecutionPlan for UnionExec { for (child_idx, &child_result) in parent_filter_result.child_results.iter().enumerate() { - if matches!(child_result, PushedDown::No) { + if !matches!(child_result, PushedDown::Exact) { unsupported_filters_per_child[child_idx] .push(Arc::clone(&parent_filter_result.filter)); } @@ -555,7 +555,7 @@ impl ExecutionPlan for UnionExec { .any(|(new, old)| !Arc::ptr_eq(new, old)); let all_filters_pushed = - vec![PushedDown::Yes; child_pushdown_result.parent_filters.len()]; + vec![PushedDown::Exact; child_pushdown_result.parent_filters.len()]; let propagation = if children_modified { let updated_node = UnionExec::try_new(new_children)?; FilterPushdownPropagation::with_parent_pushdown_result(all_filters_pushed) diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index e88722bbf4a3e..a0b1c11d73e56 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -25,6 +25,24 @@ in this section pertains to features and changes that have already been merged to the main branch and are awaiting release in this version. +### `PushedDown` distinguishes exact, inexact, and unsupported filter pushdown + +`datafusion_physical_plan::filter_pushdown::PushedDown` now reports whether a +consumer applies a predicate exactly, retains it for an inexact use such as +statistics pruning, or does not retain it: + +```rust,ignore +// Before // After +PushedDown::Yes PushedDown::Exact +PushedDown::No PushedDown::Unsupported +``` + +Implementations that retain a predicate without applying it as a complete row +filter must return `PushedDown::Inexact`. This keeps an exact `FilterExec` in +the plan while informing dynamic-filter producers that the predicate has a +consumer. `PushedDownPredicate::supported` remains available as an alias for +the new `PushedDownPredicate::exact` constructor. + ### `TableProvider::scan` takes `Option<&[usize]>` for the projection `TableProvider::scan` and the related APIs listed below previously took the