chore(cubestore): Upgrade DF: Remove redundant RewriteInListLiterals analysis pass

Author: srh

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -1175,12 +1175,31 @@ async fn numeric_cast(service: Box<dyn SqlClient>) {
             "INSERT INTO foo.managers (id, department_id) VALUES ('a', 1), ('b', 3), ('c', 3), ('d', 5)"
         ).await.unwrap();
 
+    let query = "SELECT count(*) from foo.managers WHERE department_id in ('3', '5')";
     let result = service
-        .exec_query("SELECT count(*) from foo.managers WHERE department_id in ('3', '5')")
+        .exec_query(query)
         .await
         .unwrap();
 
     assert_eq!(result.get_rows()[0], Row::new(vec![TableValue::Int(3)]));
+
+    // Check that we're casting '3' to int and not department_id to Utf8, with our Cube-specific type_coercion changes in DF.
+    let plans = service.plan_query(query).await.unwrap();
+    let expected =
+        "Projection, [count(Int64(1))@0:count(*)]\
+        \n  LinearFinalAggregate\
+        \n    CoalescePartitions\
+        \n      ClusterSend, partitions: [[1]]\
+        \n        CoalescePartitions\
+        \n          LinearPartialAggregate\
+        \n            Projection, []\
+        \n              Filter, predicate: department_id@0 = 3 OR department_id@0 = 5\
+        \n                Scan, index: default:1:[1], fields: [department_id], predicate: department_id = Int64(3) OR department_id = Int64(5)\
+        \n                  Empty";
+    assert_eq!(
+        expected,
+        pp_phys_plan_ext(plans.router.as_ref(), &PPOptions{ traverse_past_clustersend: true, show_filters: true, ..PPOptions::none() }),
+    );
 }
 
 async fn cast_timestamp_to_utf8(service: Box<dyn SqlClient>) {
@@ -7562,12 +7581,34 @@ async fn filter_multiple_in_for_decimal(service: Box<dyn SqlClient>) {
         .exec_query("INSERT INTO s.t(i) VALUES (1), (2), (3)")
         .await
         .unwrap();
+    let query = "SELECT count(*) FROM s.t WHERE i in ('2', '3')";
     let r = service
-        .exec_query("SELECT count(*) FROM s.t WHERE i in ('2', '3')")
+        .exec_query(query)
         .await
         .unwrap();
 
     assert_eq!(to_rows(&r), rows(&[(2)]));
+
+    // Verify we're casting '2' and '3' to decimal type and not casting i to Utf8, with Cube-specific DF comparison coercion changes.
+    let plans = service.plan_query(query).await.unwrap();
+    let expected =
+        "Projection, [count(Int64(1))@0:count(*)]\
+        \n  LinearFinalAggregate\
+        \n    CoalescePartitions\
+        \n      ClusterSend, partitions: [[1]]\
+        \n        CoalescePartitions\
+        \n          LinearPartialAggregate\
+        \n            Projection, []\
+        \n              Filter, predicate: i@0 = Some(200000),18,5 OR i@0 = Some(300000),18,5\
+        \n                Scan, index: default:1:[1], fields: *, predicate: i = Decimal128(Some(200000),18,5) OR i = Decimal128(Some(300000),18,5)\
+        \n                  Sort\
+        \n                    Empty";
+
+    assert_eq!(
+        expected,
+        pp_phys_plan_ext(plans.router.as_ref(), &PPOptions{ traverse_past_clustersend: true, show_filters: true, ..PPOptions::none() }),
+    );
+
 }
 
 async fn planning_aggregate_index(service: Box<dyn SqlClient>) {

rust/cubestore/cubestore/src/queryplanner/mod.rs

@@ -17,7 +17,6 @@ pub mod serialized_plan;
 mod tail_limit;
 mod topk;
 pub mod trace_data_loaded;
-use rewrite_inlist_literals::RewriteInListLiterals;
 use serialized_plan::PreSerializedPlan;
 pub use topk::MIN_TOPK_STREAM_ROWS;
 use udfs::{registerable_aggregate_udfs, registerable_scalar_udfs};
@@ -26,7 +25,6 @@ pub mod info_schema;
 pub mod merge_sort;
 pub mod metadata_cache;
 pub mod providers;
-mod rewrite_inlist_literals;
 mod rolling;
 #[cfg(test)]
 mod test_utils;
@@ -283,7 +281,7 @@ impl QueryPlannerImpl {
 
 impl QueryPlannerImpl {
     pub fn make_execution_context(mut config: SessionConfig) -> SessionContext {
-        // The config parameter is from metadata_cache_factor (which we need to rename) but doesn't
+        // The config parameter is from metadata_cache_factory (which we need to rename) but doesn't
         // include all necessary configs.
         config
             .options_mut()
@@ -297,7 +295,6 @@ impl QueryPlannerImpl {
         for udf in registerable_scalar_udfs() {
             context.register_udf(udf);
         }
-        context.add_analyzer_rule(Arc::new(RewriteInListLiterals {}));
         context.add_optimizer_rule(Arc::new(RollingOptimizerRule {}));
 
         // TODO upgrade DF