chore(cubestore): Upgrade DF: Partitioned index support

Author: srh

rust/cubestore/Cargo.lock

@@ -3824,19 +3824,22 @@ async fn planning_join_with_partitioned_index(service: Box<dyn SqlClient>) {
         .unwrap();
     assert_eq!(
         pp_phys_plan(p.router.as_ref()),
-        "ClusterSend, partitions: [[1, 3]]"
+        "CoalescePartitions\
+        \n  ClusterSend, partitions: [[1, 3]]"
     );
     assert_eq!(
         pp_phys_plan(p.worker.as_ref()),
-        "Worker\
-           \n  Projection, [order_id, customer_name]\
-           \n    MergeJoin, on: [customer_id@1 = customer_id@0]\
-           \n      MergeSort\
-           \n        Scan, index: #mi0:1:[1]:sort_on[customer_id], fields: [order_id, customer_id]\
-           \n          Empty\
-           \n      MergeSort\
-           \n        Scan, index: #mi0:3:[3]:sort_on[customer_id], fields: *\
-           \n          Empty",
+        "CoalescePartitions\
+        \n  Worker\
+        \n    CoalescePartitions\
+        \n      Projection, [order_id, customer_name]\
+        \n        MergeJoin, on: [customer_id@1 = customer_id@0]\
+        \n          Scan, index: #mi0:1:[1]:sort_on[customer_id], fields: [order_id, customer_id]\
+        \n            Sort\
+        \n              Empty\
+        \n          Scan, index: #mi0:3:[3]:sort_on[customer_id], fields: *\
+        \n            Sort\
+        \n              Empty"
     );
 }
 

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

@@ -18,8 +18,7 @@ base64 = "0.13.0"
 bumpalo = "3.6.1"
 tokio = { version = "1", features = ["full", "rt"] }
 warp = { version = "0.3.6" }
-#sqlparser = { git = 'https://github.com/cube-js/sqlparser-rs.git', rev = "4388f6712dae5073c2d71d74f64cae2edd418066" }
-sqlparser = { version = "0.50.0" }
+sqlparser = { git = "https://github.com/cube-js/sqlparser-rs.git", branch = "cube-42.2.0" }
 serde_derive = "1.0.115"
 serde = "1.0.115"
 serde_repr = "0.1"

rust/cubestore/cubestore/Cargo.toml

@@ -992,38 +992,37 @@ impl SqlService for SqlServiceImpl {
                     ))
                 }
             }
-            // TODO upgrade DF
-            // CubeStoreStatement::Statement(Statement::CreatePartitionedIndex {
-            //     name,
-            //     columns,
-            //     if_not_exists,
-            // }) => {
-            //     app_metrics::DATA_QUERIES.add_with_tags(
-            //         1,
-            //         Some(&vec![metrics::format_tag(
-            //             "command",
-            //             "create_partitioned_index",
-            //         )]),
-            //     );
-            //
-            //     if name.0.len() != 2 {
-            //         return Err(CubeError::user(format!(
-            //             "Expected name for PARTITIONED INDEX in the form '<SCHEMA>.<INDEX>', found: {}",
-            //             name
-            //         )));
-            //     }
-            //     let schema = &name.0[0].value;
-            //     let index = &name.0[1].value;
-            //     let res = self
-            //         .create_partitioned_index(
-            //             schema.to_string(),
-            //             index.to_string(),
-            //             columns,
-            //             if_not_exists,
-            //         )
-            //         .await?;
-            //     Ok(Arc::new(DataFrame::from(vec![res])))
-            // }
+            CubeStoreStatement::Statement(Statement::CreatePartitionedIndex {
+                name,
+                columns,
+                if_not_exists,
+            }) => {
+                app_metrics::DATA_QUERIES.add_with_tags(
+                    1,
+                    Some(&vec![metrics::format_tag(
+                        "command",
+                        "create_partitioned_index",
+                    )]),
+                );
+
+                if name.0.len() != 2 {
+                    return Err(CubeError::user(format!(
+                        "Expected name for PARTITIONED INDEX in the form '<SCHEMA>.<INDEX>', found: {}",
+                        name
+                    )));
+                }
+                let schema = &name.0[0].value;
+                let index = &name.0[1].value;
+                let res = self
+                    .create_partitioned_index(
+                        schema.to_string(),
+                        index.to_string(),
+                        columns,
+                        if_not_exists,
+                    )
+                    .await?;
+                Ok(Arc::new(DataFrame::from(vec![res])))
+            }
             CubeStoreStatement::Statement(Statement::Drop {
                 object_type, names, ..
             }) => {
@@ -1040,13 +1039,12 @@ impl SqlService for SqlServiceImpl {
                         self.db.drop_table(table.get_id()).await?;
                         &"drop_table"
                     }
-                    // TODO upgrade DF
-                    // ObjectType::PartitionedIndex => {
-                    //     let schema = names[0].0[0].value.clone();
-                    //     let name = names[0].0[1].value.clone();
-                    //     self.db.drop_partitioned_index(schema, name).await?;
-                    //     &"drop_partitioned_index"
-                    // }
+                    ObjectType::PartitionedIndex => {
+                        let schema = names[0].0[0].value.clone();
+                        let name = names[0].0[1].value.clone();
+                        self.db.drop_partitioned_index(schema, name).await?;
+                        &"drop_partitioned_index"
+                    }
                     _ => return Err(CubeError::user("Unsupported drop operation".to_string())),
                 };