feat: impl `handle_child_pushdown_result` for UnionExec #20145

haohuaijin · 2026-02-04T07:16:35Z

this is main purpose for this pr

This one really seems like it should be reproducible from an SLT test

let me try, maybe one memory and one parquet file can reproduce

added in https://github.com/apache/datafusion/pull/20145/changes#diff-cc932d43efc8c2d251d018ecaa22d7e898b4da64d5f216c08a089e13cecfb67f

haohuaijin · 2026-02-04T07:10:17Z

i think even for this case, pushdown filter do not have some bad effect.

adriangb · 2026-02-04T14:09:22Z

Why does the Pre / Post phase need different behavior?

i'm thinking the purpose for this pr can only happen in the pre phase, post phase is for dynamic filter, seem like not related, so i keep the default behavior

Agreed. The intuition here is to let the creator of the filter decide what to do with it.
I don't like that it makes assumptions about the implementation / what the creators of the filter want to do, but I don't see a better way to handle this.
I don't think forcing creation of the FilterExec would be good at least as things currently stand.

But we should add a comment explaining this.

adriangb · 2026-02-04T14:16:50Z

In principle we could make UnionExec transparent similar to CoalesceBatchesExec.

Help me understand why we are adding the FilterExec here.
My guess is that you are trying to address the case of Child2 supporting pushdown but Child1 not supporting it.
Without this specialized implementation we would get:

FilterExec UnionExec Child1 Child2

i.e. no changes to the plan, not incorrect but we are applying filters to the output of Child2 that are unnecessary (it already applied these filters)

With this logic we get:

UnionExec FilterExec Child1 Child2

Which skips re-applying filters to the output of Child2.

Is this interpretation correct?

Correct, thank you for such good explain👍

Can you add something along these lines as a comment justifying the added complexity?

add in ded7d9b

-Original file line number
+Diff line change
@@ Expand Up / @@ -1808,6 +1808,67 @@ fn test_filter_pushdown_through_union() { @@
         );
     }
+    #[test]
+    fn test_filter_pushdown_through_union_mixed_support() {
+        // Test case where one child supports filter pushdown and one doesn't
+        let scan1 = TestScanBuilder::new(schema()).with_support(true).build();
+        let scan2 = TestScanBuilder::new(schema()).with_support(false).build();
+        let union = UnionExec::try_new(vec![scan1, scan2]).unwrap();
+        let predicate = col_lit_predicate("a", "foo", &schema());
+        let plan = Arc::new(FilterExec::try_new(predicate, union).unwrap());
+        insta::assert_snapshot!(
+            OptimizationTest::new(plan, FilterPushdown::new(), true),
+            @r"
+        OptimizationTest:
+          input:
+            - FilterExec: a@0 = foo
+            -   UnionExec
+            -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
+            -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+          output:
+            Ok:
+              - UnionExec
+              -   DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true, predicate=a@0 = foo
+              -   FilterExec: a@0 = foo
+              -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+        "
+        );
+    }
+    #[test]
+    fn test_filter_pushdown_through_union_does_not_support() {
+        // Test case where one child supports filter pushdown and one doesn't
+        let scan1 = TestScanBuilder::new(schema()).with_support(false).build();
+        let scan2 = TestScanBuilder::new(schema()).with_support(false).build();
+        let union = UnionExec::try_new(vec![scan1, scan2]).unwrap();
+        let predicate = col_lit_predicate("a", "foo", &schema());
+        let plan = Arc::new(FilterExec::try_new(predicate, union).unwrap());
+        insta::assert_snapshot!(
+            OptimizationTest::new(plan, FilterPushdown::new(), true),
+            @"
+        OptimizationTest:
+          input:
+            - FilterExec: a@0 = foo
+            -   UnionExec
+            -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+            -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+          output:
+            Ok:
+              - UnionExec
+              -   FilterExec: a@0 = foo
+              -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+              -   FilterExec: a@0 = foo
+              -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=false
+        "
+        );
+    }
     /// Schema:
     /// a: String
     /// b: String
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -36,7 +36,11 @@ use crate::execution_plan::{ @@
         InvariantLevel, boundedness_from_children, check_default_invariants,
         emission_type_from_children,
     };
-    use crate::filter_pushdown::{FilterDescription, FilterPushdownPhase};
+    use crate::filter::FilterExec;
+    use crate::filter_pushdown::{
+        ChildPushdownResult, FilterDescription, FilterPushdownPhase,
+        FilterPushdownPropagation, PushedDown,
+    };
     use crate::metrics::BaselineMetrics;
     use crate::projection::{ProjectionExec, make_with_child};
     use crate::stream::ObservedStream;
@@ Expand All / @@ -49,7 +53,9 @@ use datafusion_common::{ @@
         Result, assert_or_internal_err, exec_err, internal_datafusion_err,
     };
     use datafusion_execution::TaskContext;
-    use datafusion_physical_expr::{EquivalenceProperties, PhysicalExpr, calculate_union};
+    use datafusion_physical_expr::{
+        EquivalenceProperties, PhysicalExpr, calculate_union, conjunction,
+    };
     use futures::Stream;
     use itertools::Itertools;
@@ Expand Down Expand Up / @@ -370,6 +376,80 @@ impl ExecutionPlan for UnionExec { @@
         ) -> Result<FilterDescription> {
             FilterDescription::from_children(parent_filters, &self.children())
         }
+        fn handle_child_pushdown_result(
+            &self,
+            phase: FilterPushdownPhase,
+            child_pushdown_result: ChildPushdownResult,
+            _config: &ConfigOptions,
+        ) -> Result<FilterPushdownPropagation<Arc<dyn ExecutionPlan>>> {
+            // For non-Pre phase, use default behavior
+            if !matches!(phase, FilterPushdownPhase::Pre) {
+                return Ok(FilterPushdownPropagation::if_all(child_pushdown_result));
+            }
+            // UnionExec needs specialized filter pushdown handling when children have
+            // heterogeneous pushdown support. Without this, when some children support
+            // pushdown and others don't, the default behavior would leave FilterExec
+            // above UnionExec, re-applying filters to outputs of all children—including
+            // those that already applied the filters via pushdown. This specialized
+            // implementation adds FilterExec only to children that don't support
+            // pushdown, avoiding redundant filtering and improving performance.
+            //
+            // Example: Given Child1 (no pushdown support) and Child2 (has pushdown support)
+            //   Default behavior:          This implementation:
+            //   FilterExec                 UnionExec
+            //     UnionExec                  FilterExec
+            //       Child1                     Child1
+            //       Child2(filter)           Child2(filter)
+            // Collect unsupported filters for each child
+            let mut unsupported_filters_per_child = vec![Vec::new(); self.inputs.len()];
+            for parent_filter_result in child_pushdown_result.parent_filters.iter() {
+                for (child_idx, &child_result) in
+                    parent_filter_result.child_results.iter().enumerate()
+                {
+                    if matches!(child_result, PushedDown::No) {
+                        unsupported_filters_per_child[child_idx]
+                            .push(Arc::clone(&parent_filter_result.filter));
+                    }
+                }
+            }
+            // Wrap children that have unsupported filters with FilterExec
+            let mut new_children = self.inputs.clone();
+            for (child_idx, unsupported_filters) in
+                unsupported_filters_per_child.iter().enumerate()
+            {
+                if !unsupported_filters.is_empty() {
+                    let combined_filter = conjunction(unsupported_filters.clone());
+                    new_children[child_idx] = Arc::new(FilterExec::try_new(
+                        combined_filter,
+                        Arc::clone(&self.inputs[child_idx]),
+                    )?);
+                }
+            }
+            // Check if any children were modified
+            let children_modified = new_children
+                .iter()
+                .zip(self.inputs.iter())
+                .any(|(new, old)| !Arc::ptr_eq(new, old));
+            let all_filters_pushed =
+                vec![PushedDown::Yes; 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)
+                    .with_updated_node(updated_node)
+            } else {
+                FilterPushdownPropagation::with_parent_pushdown_result(all_filters_pushed)
+            };
+            // Report all parent filters as supported since we've ensured they're applied
+            // on all children (either pushed down or via FilterExec)
+            Ok(propagation)
+        }
     }
     /// Combines multiple input streams by interleaving them.
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -674,3 +674,66 @@ logical_plan @@
     physical_plan
 )SortExec: expr=[id@0 ASC NULLS LAST], preserve_partitioning=[false]
 )--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/array_data/data.parquet]]}, projection=[id, tags], file_type=parquet, predicate=id@0 > 1 AND array_has(tags@1, rust), pruning_predicate=id_null_count@1 != row_count@2 AND id_max@0 > 1, required_guarantees=[]
+    ###
+    # Test filter pushdown through UNION with mixed support
+    # This tests the case where one child supports filter pushdown (parquet) and one doesn't (memory table)
+    ###
+    # enable filter pushdown
+    statement ok
+    set datafusion.execution.parquet.pushdown_filters = true;
+    statement ok
+    set datafusion.optimizer.max_passes = 0;
+    # Create memory table with matching schema (a: VARCHAR, b: BIGINT)
+    statement ok
+    CREATE TABLE t_union_mem(a VARCHAR, b BIGINT) AS VALUES ('qux', 4), ('quux', 5);
+    # Create parquet table with matching schema
+    statement ok
+    CREATE EXTERNAL TABLE t_union_parquet(a VARCHAR, b BIGINT) STORED AS PARQUET
+    LOCATION 'test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet';
+    # Query results combining memory table and Parquet with filter
+    query I rowsort
+    SELECT b FROM (
+      SELECT a, b FROM t_union_mem
+      UNION ALL
+      SELECT a, b FROM t_union_parquet
+    ) WHERE b > 2;
+    ----
+    # Explain the union query - filter should be pushed to parquet but not memory table
+    query TT
+    EXPLAIN SELECT b FROM (
+      SELECT a, b FROM t_union_mem
+      UNION ALL
+      SELECT a, b FROM t_union_parquet
+    ) WHERE b > 2;
+    ----
+    logical_plan
+)Projection: b
+)--Filter: b > Int64(2)
+)----Union
+)------Projection: t_union_mem.a, t_union_mem.b
+)--------TableScan: t_union_mem
+)------Projection: t_union_parquet.a, t_union_parquet.b
+)--------TableScan: t_union_parquet
+    physical_plan
+)UnionExec
+)--FilterExec: b@0 > 2
+)----DataSourceExec: partitions=1, partition_sizes=[1]
+)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_filter_pushdown/parquet_table/1.parquet]]}, projection=[b], file_type=parquet, predicate=b@1 > 2, pruning_predicate=b_null_count@1 != row_count@2 AND b_max@0 > 2, required_guarantees=[]
+    # Clean up union test tables
+    statement ok
+    DROP TABLE t_union_mem;
+    statement ok
+    DROP TABLE t_union_parquet;

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: impl `handle_child_pushdown_result` for UnionExec #20145

Diff view

Diff view

There are no files selected for viewing

haohuaijin Feb 4, 2026 •

edited

Loading

Uh oh!

adriangb Feb 4, 2026

Uh oh!

haohuaijin Feb 4, 2026 •

edited

Loading

Uh oh!

haohuaijin Feb 4, 2026

Uh oh!

haohuaijin Feb 4, 2026

Uh oh!

adriangb Feb 4, 2026

Uh oh!

haohuaijin Feb 4, 2026 •

edited

Loading

Uh oh!

adriangb Feb 4, 2026

Uh oh!

adriangb Feb 4, 2026

Uh oh!

haohuaijin Feb 4, 2026

Uh oh!

adriangb Feb 4, 2026

Uh oh!

haohuaijin Feb 5, 2026

Uh oh!

feat: impl handle_child_pushdown_result for UnionExec #20145

Are you sure you want to change the base?

feat: impl handle_child_pushdown_result for UnionExec #20145

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

haohuaijin Feb 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

haohuaijin Feb 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

haohuaijin Feb 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

feat: impl `handle_child_pushdown_result` for UnionExec #20145

feat: impl `handle_child_pushdown_result` for UnionExec #20145

haohuaijin Feb 4, 2026 •

edited

Loading

haohuaijin Feb 4, 2026 •

edited

Loading

haohuaijin Feb 4, 2026 •

edited

Loading