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

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.
 

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

@@ -3,11 +3,11 @@ source: cubesql/src/compile/engine/df/optimizers/filter_push_down.rs
 expression: optimize(&plan)
 ---
 Filter: #j2.c2 = Int32(10)
-  CrossJoin:
-    Filter: #j1.c1 = Int32(5)
+  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
+      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/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"#));
+}