feat: support multi table engines in distributed mode (#1316)

* chore: bump greptime-proto to 59afacd

* feat: support multi table engines in distributed mode

src/frontend/src/catalog.rs

@@ -193,6 +193,7 @@ impl CatalogManager for FrontendCatalogManager {
                 table_options: (&request.table_options).into(),
                 table_id: None, // Should and will be assigned by Meta.
                 region_ids: vec![0],
+                engine: request.engine,
             };
 
             let table = dist_instance

src/frontend/src/expr_factory.rs

@@ -43,6 +43,7 @@ pub trait CreateExprFactory {
         schema_name: &str,
         table_name: &str,
         columns: &[Column],
+        engine: &str,
     ) -> crate::error::Result<CreateTableExpr>;
 }
 
@@ -57,6 +58,7 @@ impl CreateExprFactory for DefaultCreateExprFactory {
         schema_name: &str,
         table_name: &str,
         columns: &[Column],
+        engine: &str,
     ) -> Result<CreateTableExpr> {
         let table_id = None;
         let create_expr = common_grpc_expr::build_create_expr_from_insertion(
@@ -65,6 +67,7 @@ impl CreateExprFactory for DefaultCreateExprFactory {
             table_id,
             table_name,
             columns,
+            engine,
         )
         .context(BuildCreateExprOnInsertionSnafu)?;
 
@@ -96,6 +99,7 @@ pub(crate) fn create_to_expr(
         table_options,
         table_id: None,
         region_ids: vec![],
+        engine: create.engine.to_string(),
     };
     Ok(expr)
 }

src/frontend/src/instance.rs

@@ -32,6 +32,7 @@ use async_trait::async_trait;
 use catalog::remote::MetaKvBackend;
 use catalog::CatalogManagerRef;
 use common_base::Plugins;
+use common_catalog::consts::MITO_ENGINE;
 use common_error::ext::BoxedError;
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
 use common_query::Output;
@@ -298,7 +299,7 @@ impl Instance {
                     "Table {}.{}.{} does not exist, try create table",
                     catalog_name, schema_name, table_name,
                 );
-                self.create_table_by_columns(ctx, table_name, columns)
+                self.create_table_by_columns(ctx, table_name, columns, MITO_ENGINE)
                     .await?;
                 info!(
                     "Successfully created table on insertion: {}.{}.{}",
@@ -335,6 +336,7 @@ impl Instance {
         ctx: QueryContextRef,
         table_name: &str,
         columns: &[Column],
+        engine: &str,
     ) -> Result<Output> {
         let catalog_name = &ctx.current_catalog();
         let schema_name = &ctx.current_schema();
@@ -342,7 +344,7 @@ impl Instance {
         // Create table automatically, build schema from data.
         let create_expr = self
             .create_expr_factory
-            .create_expr_by_columns(catalog_name, schema_name, table_name, columns)
+            .create_expr_by_columns(catalog_name, schema_name, table_name, columns, engine)
             .await?;
 
         info!(

src/frontend/src/instance/grpc.rs

@@ -95,6 +95,7 @@ mod test {
         InsertRequest, QueryRequest,
     };
     use catalog::helper::{TableGlobalKey, TableGlobalValue};
+    use common_catalog::consts::MITO_ENGINE;
     use common_query::Output;
     use common_recordbatch::RecordBatches;
     use query::parser::QueryLanguageParser;
@@ -162,6 +163,7 @@ mod test {
                     },
                 ],
                 time_index: "ts".to_string(),
+                engine: MITO_ENGINE.to_string(),
                 ..Default::default()
             })),
         });