fix(cubesql): Use pushdown-pullup scheme for FilterSimplifyReplacer (#9278)

This should avoid unexpected unifications of simplified expression in unrelated context. For example, same expression can be present in filter and projection, but due to over-unification in filters it can receive different representation in projection, and break aliases later, during extraction

Supporting changes:

* Remove unnecessary Vec+Box for rules

* Add NOW function template
A couple of unit tests with NOW and wrapper were relying on over-unification, and broke without a function template.

* Bump default rewrite iteration limit to 500
With new rules `select_many_filters` benchmark was hitting iteration limit

Author: mcheshkov

rust/cubesql/cubesql/src/compile/mod.rs

@@ -12878,7 +12878,7 @@ ORDER BY "source"."str0" ASC
         }
         init_testing_logger();
 
-        let logical_plan = convert_select_to_query_plan(
+        let query_plan = convert_select_to_query_plan(
             r#"
             SELECT
                 CAST("public"."KibanaSampleDataEcommerce"."order_date" AS DATE) AS "order_date",
@@ -12899,11 +12899,16 @@ ORDER BY "source"."str0" ASC
             .to_string(),
             DatabaseProtocol::PostgreSQL,
         )
-        .await
-        .as_logical_plan();
+        .await;
+
+        let physical_plan = query_plan.as_physical_plan().await.unwrap();
+        println!(
+            "Physical plan: {}",
+            displayable(physical_plan.as_ref()).indent()
+        );
+
+        let logical_plan = query_plan.as_logical_plan();
 
-        let end_date = chrono::Utc::now().date_naive();
-        let start_date = end_date - chrono::Duration::days(30);
         assert_eq!(
             logical_plan.find_cube_scan_wrapped_sql().request,
             V1LoadRequestQuery {
@@ -12930,7 +12935,7 @@ ORDER BY "source"."str0" ASC
                         "cube_name": "KibanaSampleDataEcommerce",
                         "alias": "kibanasampledata",
                         "cube_params": ["KibanaSampleDataEcommerce"],
-                        "expr": format!("(((${{KibanaSampleDataEcommerce.order_date}} >= DATE('{start_date}')) AND (${{KibanaSampleDataEcommerce.order_date}} < DATE('{end_date}'))) AND (((${{KibanaSampleDataEcommerce.notes}} = $0$) OR (${{KibanaSampleDataEcommerce.notes}} = $1$)) OR (${{KibanaSampleDataEcommerce.notes}} = $2$)))"),
+                        "expr": format!("(((${{KibanaSampleDataEcommerce.order_date}} >= CAST((NOW() + INTERVAL '-30 DAY') AS DATE)) AND (${{KibanaSampleDataEcommerce.order_date}} < CAST(NOW() AS DATE))) AND (((${{KibanaSampleDataEcommerce.notes}} = $0$) OR (${{KibanaSampleDataEcommerce.notes}} = $1$)) OR (${{KibanaSampleDataEcommerce.notes}} = $2$)))"),
                         "grouping_set": null,
                     }).to_string(),
                 ]),

rust/cubesql/cubesql/src/compile/rewrite/cost.rs

@@ -104,6 +104,8 @@ impl BestCubePlan {
             LogicalPlanLanguage::OrderReplacer(_) => 1,
             LogicalPlanLanguage::MemberReplacer(_) => 1,
             LogicalPlanLanguage::FilterReplacer(_) => 1,
+            LogicalPlanLanguage::FilterSimplifyPushDownReplacer(_) => 1,
+            LogicalPlanLanguage::FilterSimplifyPullUpReplacer(_) => 1,
             LogicalPlanLanguage::TimeDimensionDateRangeReplacer(_) => 1,
             LogicalPlanLanguage::InnerAggregateSplitReplacer(_) => 1,
             LogicalPlanLanguage::OuterProjectionSplitReplacer(_) => 1,

rust/cubesql/cubesql/src/compile/rewrite/mod.rs

@@ -405,7 +405,10 @@ crate::plan_to_language! {
             members: Vec<LogicalPlan>,
             aliases: Vec<(String, String)>,
         },
-        FilterSimplifyReplacer {
+        FilterSimplifyPushDownReplacer {
+            filters: Vec<LogicalPlan>,
+        },
+        FilterSimplifyPullUpReplacer {
             filters: Vec<LogicalPlan>,
         },
         OrderReplacer {
@@ -1901,8 +1904,12 @@ fn filter_replacer(
     )
 }
 
-fn filter_simplify_replacer(members: impl Display) -> String {
-    format!("(FilterSimplifyReplacer {})", members)
+fn filter_simplify_push_down_replacer(members: impl Display) -> String {
+    format!("(FilterSimplifyPushDownReplacer {})", members)
+}
+
+fn filter_simplify_pull_up_replacer(members: impl Display) -> String {
+    format!("(FilterSimplifyPullUpReplacer {})", members)
 }
 
 fn inner_aggregate_split_replacer(members: impl Display, alias_to_cube: impl Display) -> String {

rust/cubesql/cubesql/src/compile/rewrite/rewriter.rs

@@ -208,7 +208,7 @@ impl Rewriter {
         .with_iter_limit(
             env::var("CUBESQL_REWRITE_MAX_ITERATIONS")
                 .map(|v| v.parse::<usize>().unwrap())
-                .unwrap_or(300),
+                .unwrap_or(500),
         )
         .with_node_limit(
             env::var("CUBESQL_REWRITE_MAX_NODES")
@@ -473,20 +473,16 @@ impl Rewriter {
         eval_stable_functions: bool,
     ) -> Vec<CubeRewrite> {
         let sql_push_down = Self::sql_push_down_enabled();
-        let rules: Vec<Box<dyn RewriteRules>> = vec![
-            Box::new(MemberRules::new(
-                meta_context.clone(),
-                config_obj.clone(),
-                sql_push_down,
-            )),
-            Box::new(FilterRules::new(
+        let rules: &[&dyn RewriteRules] = &[
+            &MemberRules::new(meta_context.clone(), config_obj.clone(), sql_push_down),
+            &FilterRules::new(
                 meta_context.clone(),
                 config_obj.clone(),
                 eval_stable_functions,
-            )),
-            Box::new(DateRules::new(config_obj.clone())),
-            Box::new(OrderRules::new()),
-            Box::new(CommonRules::new(config_obj.clone())),
+            ),
+            &DateRules::new(config_obj.clone()),
+            &OrderRules::new(),
+            &CommonRules::new(config_obj.clone()),
         ];
         let mut rewrites = Vec::new();
         for r in rules {