Actually preserve predicate execution order in PushDownFilter

Author: joroKr21

datafusion/optimizer/src/push_down_filter.rs

@@ -810,25 +810,22 @@ impl OptimizerRule for PushDownFilter {
 
         match Arc::unwrap_or_clone(filter.input) {
             LogicalPlan::Filter(child_filter) => {
-                let parents_predicates = split_conjunction_owned(filter.predicate);
-
-                // remove duplicated filters
-                let child_predicates = split_conjunction_owned(child_filter.predicate);
-                let new_predicates = parents_predicates
-                    .into_iter()
-                    .chain(child_predicates)
-                    // use IndexSet to remove dupes while preserving predicate order
-                    .collect::<IndexSet<_>>()
+                // child filters first to preserve execution order
+                let new_predicates = split_conjunction_owned(child_filter.predicate)
                     .into_iter()
-                    .collect::<Vec<_>>();
+                    .chain(split_conjunction_owned(filter.predicate))
+                    // use IndexSet to remove duplicates while preserving predicate order
+                    .collect::<IndexSet<_>>();
 
                 let Some(new_predicate) = conjunction(new_predicates) else {
                     return plan_err!("at least one expression exists");
                 };
+
                 let new_filter = LogicalPlan::Filter(Filter::try_new(
                     new_predicate,
                     child_filter.input,
                 )?);
+
                 self.rewrite(new_filter, config)
             }
             LogicalPlan::Repartition(repartition) => {
@@ -2474,7 +2471,7 @@ mod tests {
             plan,
             @r"
         Projection: test.a
-          Filter: test.a >= Int64(1) AND test.a <= Int64(1)
+          Filter: test.a <= Int64(1) AND test.a >= Int64(1)
             Limit: skip=0, fetch=1
               TableScan: test
         "
@@ -3253,6 +3250,28 @@ mod tests {
         )
     }
 
+    #[test]
+    fn multi_combined_two_filters() -> Result<()> {
+        let plan = table_scan_with_pushdown_provider_builder(
+            TableProviderFilterPushDown::Inexact,
+            vec![col("a").eq(lit(10i64)), col("b").gt(lit(11i64))],
+            Some(vec![0]),
+        )?
+        .filter(col("a").eq(lit(10i64)))?
+        .filter(col("b").gt(lit(11i64)))?
+        .project(vec![col("a"), col("b")])?
+        .build()?;
+
+        assert_optimized_plan_equal!(
+            plan,
+            @r"
+        Projection: a, b
+          Filter: a = Int64(10) AND b > Int64(11)
+            TableScan: test projection=[a], partial_filters=[a = Int64(10), b > Int64(11)]
+        "
+        )
+    }
+
     #[test]
     fn multi_combined_filter_exact() -> Result<()> {
         let plan = table_scan_with_pushdown_provider_builder(
@@ -3273,6 +3292,27 @@ mod tests {
         )
     }
 
+    #[test]
+    fn multi_combined_two_filters_exact() -> Result<()> {
+        let plan = table_scan_with_pushdown_provider_builder(
+            TableProviderFilterPushDown::Exact,
+            vec![],
+            Some(vec![0]),
+        )?
+        .filter(col("a").eq(lit(10i64)))?
+        .filter(col("b").gt(lit(11i64)))?
+        .project(vec![col("a"), col("b")])?
+        .build()?;
+
+        assert_optimized_plan_equal!(
+            plan,
+            @r"
+        Projection: a, b
+          TableScan: test projection=[a], full_filters=[a = Int64(10), b > Int64(11)]
+        "
+        )
+    }
+
     #[test]
     fn test_filter_with_alias() -> Result<()> {
         // in table scan the true col name is 'test.a',