[DRAFT] update PushedDown to have 3 states

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<dyn ExecutionPlan> = 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::<AggregateExec>()
+            .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<dyn ExecutionPlan>, 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::<HashJoinExec>()
             .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"
+            );
+        }
     }
 }
 

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<Arc<dyn PhysicalExpr>>,
     batch_size: Option<usize>,
     batches: Vec<RecordBatch>,
@@ -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<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
     ) -> Result<TreeNodeRecursion> {
+        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<RecordBatch>,
     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<RecordBatch>) -> Self {
         self.batches = batches;
         self
     }
 
     pub fn build(self) -> Arc<dyn ExecutionPlan> {
-        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<dyn ExecutionPlan>);
                     Ok(res)
                 }
-                PushedDown::Yes => {
+                PushedDown::Exact => {
                     let res = FilterPushdownPropagation::if_all(child_pushdown_result);
                     Ok(res)
                 }

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`.
 ///

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]);
     }
 }