fix(cubesql): Disable filter pushdown over Filter(CrossJoin) (#9474)

This should help with rewrites filters on top of complex ungrouped-grouped joins as a subquery joins

Author: mcheshkov

rust/cubesql/cubesql/src/compile/engine/df/optimizers/filter_push_down.rs

@@ -111,6 +111,15 @@ fn filter_push_down(
             )
         }
         LogicalPlan::Filter(Filter { predicate, input }) => {
+            // Current DataFusion version plans complex joins as Filter(CrossJoin)
+            // So for query like `SELECT ... FROM ... JOIN ... ON complex_condition WHERE predicate`
+            // Plan can look like Filter(predicate, Filter(join_condition, CrossJoin))
+            // This optimizer can mess with filter predicates, and break join detection later in rewrites
+            // So, for now, it just completely pessimizes plans like Filter(CrossJoin)
+            if let LogicalPlan::CrossJoin(_) = input.as_ref() {
+                return issue_filter(predicates, plan.clone());
+            }
+
             // When encountering a filter, collect it to our list of predicates,
             // remove the filter from the plan and continue down the plan.
 
@@ -692,15 +701,10 @@ mod tests {
     };
     use datafusion::logical_plan::{binary_expr, col, count, lit, sum, LogicalPlanBuilder};
 
-    fn optimize(plan: &LogicalPlan) -> Result<LogicalPlan> {
+    fn optimize(plan: &LogicalPlan) -> LogicalPlan {
         let rule = FilterPushDown::new();
         rule.optimize(plan, &OptimizerConfig::new())
-    }
-
-    fn assert_optimized_plan_eq(plan: LogicalPlan, expected: &str) {
-        let optimized_plan = optimize(&plan).expect("failed to optimize plan");
-        let formatted_plan = format!("{:?}", optimized_plan);
-        assert_eq!(formatted_plan, expected);
+            .expect("failed to optimize plan")
     }
 
     #[test]
@@ -714,14 +718,7 @@ mod tests {
             .filter(col("t2.n2").gt(lit(5i32)))?
             .build()?;
 
-        let expected = "\
-              Projection: #t1.c1 AS n1, #t1.c3 AS n2, alias=t2\
-            \n  Filter: #t1.c3 > Int32(5)\
-            \n    Projection: #t1.c1, #t1.c3\
-            \n      TableScan: t1 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 
@@ -752,17 +749,7 @@ mod tests {
             .project(vec![col("c7"), col("c5"), col("c9")])?
             .build()?;
 
-        let expected = "\
-              Projection: #t3.c7, #t3.c5, #c9\
-            \n  Projection: #t3.c7, #t3.c5, #t3.c8 AS c9\
-            \n    Projection: #t2.c4 AS c7, #t2.c5, #t2.c6 AS c8, alias=t3\
-            \n      Projection: #t1.c1 AS c4, #t1.c2 AS c5, #t1.c3 AS c6, alias=t2\
-            \n        Filter: #t1.c2 > Int32(5) AND #t1.c2 <= Int32(10) AND #t1.c3 = Int32(0) AND NOT #t1.c1 < Int32(0)\
-            \n          Projection: #t1.c1, #t1.c2, #t1.c3\
-            \n            TableScan: t1 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 
@@ -782,18 +769,7 @@ mod tests {
             .project(vec![col("c1"), col("c2"), col("c3")])?
             .build()?;
 
-        let expected = "\
-              Projection: #t1.c1, #c2, #t1.c3\
-            \n  Filter: #t1.c1 > #t1.c3\
-            \n    Projection: #t1.c1, #c2, #t1.c3\
-            \n      Filter: #c2 = Int32(5)\
-            \n        Projection: #t1.c1, #t1.c2 + Int32(5) AS c2, #t1.c3\
-            \n          Filter: #t1.c3 < Int32(5)\
-            \n            Projection: #t1.c1, #t1.c2, #t1.c3\
-            \n              TableScan: t1 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 
@@ -847,16 +823,7 @@ mod tests {
             .filter(col("c3").eq(lit(0i32)))?
             .build()?;
 
-        let expected = "\
-              Projection: #t1.c1, #SUM(t1.c2) AS c2_sum, #t1.c3\
-            \n  Filter: #SUM(t1.c2) > Int32(10)\
-            \n    Aggregate: groupBy=[[#t1.c1, #t1.c3]], aggr=[[SUM(#t1.c2)]]\
-            \n      Filter: #t1.c3 = Int32(0)\
-            \n        Projection: #t1.c1, #t1.c2, #t1.c3\
-            \n          TableScan: t1 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 
@@ -897,14 +864,7 @@ mod tests {
             .filter(col("c3").eq(lit(5i32)))?
             .build()?;
 
-        let expected = "\
-              Sort: #t1.c2\
-            \n  Filter: #t1.c3 = Int32(5)\
-            \n    Projection: #t1.c1, #t1.c2, #t1.c3\
-            \n      TableScan: t1 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 
@@ -998,19 +958,7 @@ mod tests {
         .filter(col("c2").eq(lit(10i32)))?
         .build()?;
 
-        let expected = "\
-              Filter: #j2.c2 = Int32(10)\
-            \n  CrossJoin:\
-            \n    Filter: #j1.c1 = Int32(5)\
-            \n      Projection: #j1.c1\
-            \n        TableScan: j1 projection=None\
-            \n    Projection: #COUNT(UInt8(1)) AS c2, alias=j2\
-            \n      Aggregate: groupBy=[[]], aggr=[[COUNT(UInt8(1))]]\
-            \n        Projection: #j2.c2\
-            \n          TableScan: j2 projection=None\
-        ";
-
-        assert_optimized_plan_eq(plan, expected);
+        insta::assert_debug_snapshot!(optimize(&plan));
         Ok(())
     }
 

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__filter_down_cross_join_right_one_row.snap

@@ -0,0 +1,13 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Filter: #j2.c2 = Int32(10)
+  Filter: #j1.c1 = Int32(5)
+    CrossJoin:
+      Projection: #j1.c1
+        TableScan: j1 projection=None
+      Projection: #COUNT(UInt8(1)) AS c2, alias=j2
+        Aggregate: groupBy=[[]], aggr=[[COUNT(UInt8(1))]]
+          Projection: #j2.c2
+            TableScan: j2 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__filter_down_projection.snap

@@ -0,0 +1,8 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Projection: #t1.c1 AS n1, #t1.c3 AS n2, alias=t2
+  Filter: #t1.c3 > Int32(5)
+    Projection: #t1.c1, #t1.c3
+      TableScan: t1 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__filter_down_sort.snap

@@ -0,0 +1,8 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Sort: #t1.c2
+  Filter: #t1.c3 = Int32(5)
+    Projection: #t1.c1, #t1.c2, #t1.c3
+      TableScan: t1 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__filters_down_aggregate.snap

@@ -0,0 +1,10 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Projection: #t1.c1, #SUM(t1.c2) AS c2_sum, #t1.c3
+  Filter: #SUM(t1.c2) > Int32(10)
+    Aggregate: groupBy=[[#t1.c1, #t1.c3]], aggr=[[SUM(#t1.c2)]]
+      Filter: #t1.c3 = Int32(0)
+        Projection: #t1.c1, #t1.c2, #t1.c3
+          TableScan: t1 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__multiple_filters_down_projections.snap

@@ -0,0 +1,11 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Projection: #t3.c7, #t3.c5, #c9
+  Projection: #t3.c7, #t3.c5, #t3.c8 AS c9
+    Projection: #t2.c4 AS c7, #t2.c5, #t2.c6 AS c8, alias=t3
+      Projection: #t1.c1 AS c4, #t1.c2 AS c5, #t1.c3 AS c6, alias=t2
+        Filter: #t1.c2 > Int32(5) AND #t1.c2 <= Int32(10) AND #t1.c3 = Int32(0) AND NOT #t1.c1 < Int32(0)
+          Projection: #t1.c1, #t1.c2, #t1.c3
+            TableScan: t1 projection=None

rust/cubesql/cubesql/src/compile/engine/df/optimizers/snapshots/cubesql__compile__engine__df__optimizers__filter_push_down__tests__multiple_filters_down_projections_with_post_processing.snap

@@ -0,0 +1,12 @@
+---
+source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
+expression: optimize(&plan)
+---
+Projection: #t1.c1, #c2, #t1.c3
+  Filter: #t1.c1 > #t1.c3
+    Projection: #t1.c1, #c2, #t1.c3
+      Filter: #c2 = Int32(5)
+        Projection: #t1.c1, #t1.c2 + Int32(5) AS c2, #t1.c3
+          Filter: #t1.c3 < Int32(5)
+            Projection: #t1.c1, #t1.c2, #t1.c3
+              TableScan: t1 projection=None

rust/cubesql/cubesql/src/compile/test/test_cube_join_grouped.rs

@@ -899,3 +899,63 @@ LIMIT 1
         .on
         .contains(r#"${MultiTypeCube.dim_str0} IS NOT DISTINCT FROM \"t0\".\"dim_str0\""#));
 }
+
+/// Filter on top of ungrouped-grouped join with complex condition should be rewritten as well
+#[tokio::test]
+async fn test_join_ungrouped_grouped_with_filter_and_measure() {
+    if !Rewriter::sql_push_down_enabled() {
+        return;
+    }
+    init_testing_logger();
+
+    let query_plan = convert_select_to_query_plan(
+        // language=PostgreSQL
+        r#"
+SELECT "t0"."measure"
+FROM
+    MultiTypeCube
+    INNER JOIN (
+        SELECT
+            dim_str0,
+            AVG(avgPrice) AS "measure"
+        FROM
+            MultiTypeCube
+        GROUP BY 1
+    ) "t0"
+    ON (MultiTypeCube.dim_str0 IS NOT DISTINCT FROM "t0".dim_str0)
+WHERE ("t0"."measure" IS NULL)
+LIMIT 1
+;
+        "#
+        .to_string(),
+        DatabaseProtocol::PostgreSQL,
+    )
+    .await;
+
+    let physical_plan = query_plan.as_physical_plan().await.unwrap();
+    println!(
+        "Physical plan: {}",
+        displayable(physical_plan.as_ref()).indent()
+    );
+
+    let request = query_plan
+        .as_logical_plan()
+        .find_cube_scan_wrapped_sql()
+        .request;
+
+    assert_eq!(request.ungrouped, Some(true));
+
+    assert_eq!(request.subquery_joins.as_ref().unwrap().len(), 1);
+
+    let subquery = &request.subquery_joins.unwrap()[0];
+
+    assert!(!subquery.sql.contains("ungrouped"));
+    assert_eq!(subquery.join_type, "INNER");
+    assert!(subquery
+        .on
+        .contains(r#"${MultiTypeCube.dim_str0} IS NOT DISTINCT FROM \"t0\".\"dim_str0\""#));
+
+    // Outer filter
+    assert_eq!(request.segments.as_ref().unwrap().len(), 1);
+    assert!(request.segments.as_ref().unwrap()[0].contains(r#"\"t0\".\"measure\" IS NULL"#));
+}