Change Approach to Exact/Inexact/Unsupported

Author: LiaCastaneda

datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs

@@ -220,12 +220,12 @@ impl FileSource for TestSource {
                 ..self.clone()
             });
             Ok(FilterPushdownPropagation::with_parent_pushdown_result(
-                vec![PushedDown::Yes; filters.len()],
+                vec![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()],
             ))
         }
     }
@@ -542,8 +542,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::Unsupported | PushedDown::Inexact => {
+                    // We have a filter that wasn't exactly pushed down, so we need to apply it
                     let new_child = FilterExec::try_new(
                         Arc::clone(&first_pushdown_result.predicate),
                         Arc::clone(&self.input),
@@ -555,7 +555,8 @@ impl ExecutionPlan for TestNode {
                     res.updated_node = Some(Arc::new(new_self) as Arc<dyn ExecutionPlan>);
                     Ok(res)
                 }
-                PushedDown::Yes => {
+                PushedDown::Exact => {
+                    // Filter was exactly pushed down, no need to apply it again
                     let res = FilterPushdownPropagation::if_all(child_pushdown_result);
                     Ok(res)
                 }

datafusion/datasource-parquet/src/source.rs

@@ -707,6 +707,9 @@ impl FileSource for ParquetSource {
             .into_iter()
             .map(|filter| {
                 if can_expr_be_pushed_down_with_schemas(&filter, table_schema) {
+                    // When pushdown_filters is true, filters will be used for row-level filtering
+                    // When pushdown_filters is false, filters will be used for stats pruning only
+                    // We'll mark them as supported here and adjust later based on pushdown_filters
                     PushedDownPredicate::supported(filter)
                 } else {
                     PushedDownPredicate::unsupported(filter)
@@ -715,19 +718,19 @@ 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 | PushedDown::Unsupported => None,
             })
             .collect_vec();
         let predicate = match source.predicate {
@@ -740,13 +743,24 @@ impl FileSource for ParquetSource {
         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).
+        // even though we updated the predicate to include the filters (they will be used for stats pruning only).
+        // In this case, we return Inexact for filters that can be pushed down (used for stats pruning)
+        // and Unsupported for filters that cannot be pushed down at all.
         if !pushdown_filters {
+            let result_discriminants = filters
+                .iter()
+                .map(|f| match f.discriminant {
+                    PushedDown::Exact => PushedDown::Inexact, // Downgrade to Inexact (stats pruning only)
+                    PushedDown::Unsupported => PushedDown::Unsupported, // Keep as Unsupported
+                    _ => unreachable!("Only Exact or Unsupported expected at this point"),
+                })
+                .collect();
             return Ok(FilterPushdownPropagation::with_parent_pushdown_result(
-                vec![PushedDown::No; filters.len()],
+                result_discriminants,
             )
             .with_updated_node(source));
         }
+        // If pushdown_filters is true, we return the original discriminants (Exact for supported filters)
         Ok(FilterPushdownPropagation::with_parent_pushdown_result(
             filters.iter().map(|f| f.discriminant).collect(),
         )

datafusion/datasource/src/file.rs

@@ -121,8 +121,9 @@ pub trait FileSource: Send + Sync {
         filters: Vec<Arc<dyn PhysicalExpr>>,
         _config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn FileSource>>> {
+        // Default implementation: don't support filter pushdown
         Ok(FilterPushdownPropagation::with_parent_pushdown_result(
-            vec![PushedDown::No; filters.len()],
+            vec![PushedDown::Unsupported; filters.len()],
         ))
     }
 

datafusion/datasource/src/source.rs

@@ -185,8 +185,9 @@ pub trait DataSource: Send + Sync + Debug {
         filters: Vec<Arc<dyn PhysicalExpr>>,
         _config: &ConfigOptions,
     ) -> Result<FilterPushdownPropagation<Arc<dyn DataSource>>> {
+        // Default implementation: don't support filter pushdown
         Ok(FilterPushdownPropagation::with_parent_pushdown_result(
-            vec![PushedDown::No; filters.len()],
+            vec![PushedDown::Unsupported; filters.len()],
         ))
     }
 }

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -294,16 +294,6 @@ impl DynamicFilterPhysicalExpr {
             .await;
     }
 
-    /// Check if this dynamic filter is being actively used by any consumers.
-    ///
-    /// Returns `true` if there are references beyond the producer (e.g., the HashJoinExec
-    /// that created the filter). This is useful to avoid computing expensive filter
-    /// expressions when no consumer will actually use them.
-    pub fn is_used(self: &Arc<Self>) -> bool {
-        // Strong count > 1 means at least one consumer is holding a reference beyond the producer.
-        Arc::strong_count(self) > 1
-    }
-
     fn render(
         &self,
         f: &mut std::fmt::Formatter<'_>,
@@ -617,47 +607,4 @@ mod test {
         // wait_complete should return immediately
         dynamic_filter.wait_complete().await;
     }
-
-    #[test]
-    fn test_is_used() {
-        // Create a dynamic filter
-        let filter = Arc::new(DynamicFilterPhysicalExpr::new(
-            vec![],
-            lit(true) as Arc<dyn PhysicalExpr>,
-        ));
-
-        // Initially, only one reference exists (the filter itself)
-        assert!(
-            !filter.is_used(),
-            "Filter should not be used with only one reference"
-        );
-
-        // Simulate a consumer holding a reference (e.g., ParquetExec)
-        let consumer1 = Arc::clone(&filter);
-        assert!(
-            filter.is_used(),
-            "Filter should be used with a consumer reference"
-        );
-
-        // Multiple consumers
-        let consumer2 = Arc::clone(&filter);
-        assert!(
-            filter.is_used(),
-            "Filter should still be used with multiple consumers"
-        );
-
-        // Drop one consumer
-        drop(consumer1);
-        assert!(
-            filter.is_used(),
-            "Filter should still be used with remaining consumer"
-        );
-
-        // Drop all consumers
-        drop(consumer2);
-        assert!(
-            !filter.is_used(),
-            "Filter should not be used after all consumers dropped"
-        );
-    }
 }

datafusion/physical-optimizer/src/filter_pushdown.rs

@@ -495,9 +495,10 @@ fn push_down_filters(
         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)
+        // Push down filters that are either Exact or Inexact (both mean the child will use them)
+        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
@@ -535,7 +536,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
@@ -548,7 +549,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,
@@ -557,7 +558,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

datafusion/physical-plan/src/filter.rs

@@ -521,19 +521,20 @@ impl ExecutionPlan for FilterExec {
         if !matches!(phase, FilterPushdownPhase::Pre) {
             return Ok(FilterPushdownPropagation::if_all(child_pushdown_result));
         }
-        // We absorb any parent filters that were not handled by our children
+        // We absorb any parent filters that were not handled by our children with exact filtering
         let unsupported_parent_filters =
             child_pushdown_result.parent_filters.iter().filter_map(|f| {
-                matches!(f.all(), PushedDown::No).then_some(Arc::clone(&f.filter))
+                // Keep filters that weren't handled with exact filtering (Inexact or Unsupported)
+                (!matches!(f.all(), PushedDown::Exact)).then_some(Arc::clone(&f.filter))
             });
         let unsupported_self_filters = child_pushdown_result
             .self_filters
             .first()
             .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();
 
@@ -592,8 +593,11 @@ impl ExecutionPlan for FilterExec {
             Some(Arc::new(new) as _)
         };
 
+        // FilterExec always applies filters exactly.
+        // Even if the child returned Inexact or Unsupported, we absorb those filters
+        // and apply them, so from the parent's perspective, the filter is handled exactly.
         Ok(FilterPushdownPropagation {
-            filters: vec![PushedDown::Yes; child_pushdown_result.parent_filters.len()],
+            filters: vec![PushedDown::Exact; child_pushdown_result.parent_filters.len()],
             updated_node,
         })
     }