feat: implement rename table (GreptimeTeam#802)

* feat: support renaming tables in the mito table engine

* chore: add test for table engine

* chore: fix test

src/api/greptime/v1/ddl.proto

@@ -38,6 +38,7 @@ message AlterExpr {
   oneof kind {
     AddColumns add_columns = 4;
     DropColumns drop_columns = 5;
+    RenameTable rename_table = 6;
   }
 }
 
@@ -60,6 +61,10 @@ message DropColumns {
   repeated DropColumn drop_columns = 1;
 }
 
+message RenameTable {
+  string new_table_name = 1;
+}
+
 message AddColumn {
   ColumnDef column_def = 1;
   bool is_key = 2;

src/common/grpc-expr/src/alter.rs

@@ -15,7 +15,7 @@
 use std::sync::Arc;
 
 use api::v1::alter_expr::Kind;
-use api::v1::{AlterExpr, CreateTableExpr, DropColumns};
+use api::v1::{AlterExpr, CreateTableExpr, DropColumns, RenameTable};
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use datatypes::schema::{ColumnSchema, SchemaBuilder, SchemaRef};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -87,6 +87,16 @@ pub fn alter_expr_to_request(expr: AlterExpr) -> Result<Option<AlterTableRequest
             };
             Ok(Some(request))
         }
+        Some(Kind::RenameTable(RenameTable { new_table_name })) => {
+            let alter_kind = AlterKind::RenameTable { new_table_name };
+            let request = AlterTableRequest {
+                catalog_name,
+                schema_name,
+                table_name: expr.table_name,
+                alter_kind,
+            };
+            Ok(Some(request))
+        }
         None => Ok(None),
     }
 }

src/datanode/src/sql/alter.rs

@@ -76,13 +76,9 @@ impl SqlHandler {
             AlterTableOperation::DropColumn { name } => AlterKind::DropColumns {
                 names: vec![name.value.clone()],
             },
-            AlterTableOperation::RenameTable { .. } => {
-                // TODO update proto to support alter table name
-                return error::InvalidSqlSnafu {
-                    msg: "rename table not unsupported yet".to_string(),
-                }
-                .fail();
-            }
+            AlterTableOperation::RenameTable { new_table_name } => AlterKind::RenameTable {
+                new_table_name: new_table_name.clone(),
+            },
         };
         Ok(AlterTableRequest {
             catalog_name: Some(table_ref.catalog.to_string()),
@@ -145,9 +141,21 @@ mod tests {
     async fn test_alter_to_request_with_renaming_table() {
         let handler = create_mock_sql_handler().await;
         let alter_table = parse_sql("ALTER TABLE test_table RENAME table_t;");
-        let err = handler
+        let req = handler
             .alter_to_request(alter_table, TableReference::bare("test_table"))
-            .unwrap_err();
-        assert_matches!(err, crate::error::Error::InvalidSql { .. });
+            .unwrap();
+        assert_eq!(req.catalog_name, Some("greptime".to_string()));
+        assert_eq!(req.schema_name, Some("public".to_string()));
+        assert_eq!(req.table_name, "test_table");
+
+        let alter_kind = req.alter_kind;
+        assert_matches!(alter_kind, AlterKind::RenameTable { .. });
+
+        match alter_kind {
+            AlterKind::RenameTable { new_table_name } => {
+                assert_eq!(new_table_name, "table_t");
+            }
+            _ => unreachable!(),
+        }
     }
 }

src/mito/src/engine.rs

@@ -1066,6 +1066,50 @@ mod tests {
         assert_eq!(new_schema.version(), old_schema.version() + 1);
     }
 
+    #[tokio::test]
+    async fn test_alter_rename_table() {
+        let (engine, table_engine, _table, object_store, _dir) =
+            test_util::setup_mock_engine_and_table().await;
+
+        let new_table_name = "table_t";
+        // test rename table
+        let req = AlterTableRequest {
+            catalog_name: None,
+            schema_name: None,
+            table_name: TABLE_NAME.to_string(),
+            alter_kind: AlterKind::RenameTable {
+                new_table_name: new_table_name.to_string(),
+            },
+        };
+        let ctx = EngineContext::default();
+        let table = table_engine.alter_table(&ctx, req).await.unwrap();
+
+        assert_eq!(table.table_info().name, new_table_name);
+
+        let table_engine = MitoEngine::new(EngineConfig::default(), engine, object_store);
+        let open_req = OpenTableRequest {
+            catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+            schema_name: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: new_table_name.to_string(),
+            table_id: 1,
+            region_numbers: vec![0],
+        };
+
+        // test reopen table
+        let reopened = table_engine
+            .open_table(&ctx, open_req.clone())
+            .await
+            .unwrap()
+            .unwrap();
+        let reopened = reopened
+            .as_any()
+            .downcast_ref::<MitoTable<MockRegion>>()
+            .unwrap();
+        assert_eq!(reopened.table_info(), table.table_info());
+        assert_eq!(reopened.table_info().name, new_table_name);
+        assert_eq!(reopened.manifest().last_version(), 2);
+    }
+
     #[tokio::test]
     async fn test_drop_table() {
         common_telemetry::init_default_ut_logging();

src/mito/src/table.rs

@@ -167,20 +167,26 @@ impl<R: Region> Table for MitoTable<R> {
 
         let table_info = self.table_info();
         let table_name = &table_info.name;
-        let table_meta = &table_info.meta;
-        let mut new_meta = table_meta
-            .builder_with_alter_kind(table_name, &req.alter_kind)?
-            .build()
-            .context(error::BuildTableMetaSnafu { table_name })
-            .map_err(BoxedError::new)
-            .context(table_error::TableOperationSnafu)?;
-
-        let alter_op = create_alter_operation(table_name, &req.alter_kind, &mut new_meta)?;
 
         let mut new_info = TableInfo::clone(&*table_info);
+        // setup new table info
+        match &req.alter_kind {
+            AlterKind::RenameTable { new_table_name } => {
+                new_info.name = new_table_name.clone();
+            }
+            AlterKind::AddColumns { .. } | AlterKind::DropColumns { .. } => {
+                let table_meta = &table_info.meta;
+                let new_meta = table_meta
+                    .builder_with_alter_kind(table_name, &req.alter_kind)?
+                    .build()
+                    .context(error::BuildTableMetaSnafu { table_name })
+                    .map_err(BoxedError::new)
+                    .context(table_error::TableOperationSnafu)?;
+                new_info.meta = new_meta;
+            }
+        }
         // Increase version of the table.
         new_info.ident.version = table_info.ident.version + 1;
-        new_info.meta = new_meta;
 
         // Persist the alteration to the manifest.
         logging::debug!(
@@ -201,27 +207,30 @@ impl<R: Region> Table for MitoTable<R> {
             .map_err(BoxedError::new)
             .context(table_error::TableOperationSnafu)?;
 
-        // TODO(yingwen): Error handling. Maybe the region need to provide a method to
-        // validate the request first.
-        let region = self.region();
-        let region_meta = region.in_memory_metadata();
-        let alter_req = AlterRequest {
-            operation: alter_op,
-            version: region_meta.version(),
-        };
-        // Alter the region.
-        logging::debug!(
-            "start altering region {} of table {}, with request {:?}",
-            region.name(),
-            table_name,
-            alter_req,
-        );
-        region
-            .alter(alter_req)
-            .await
-            .map_err(BoxedError::new)
-            .context(table_error::TableOperationSnafu)?;
-
+        if let Some(alter_op) =
+            create_alter_operation(table_name, &req.alter_kind, &mut new_info.meta)?
+        {
+            // TODO(yingwen): Error handling. Maybe the region need to provide a method to
+            // validate the request first.
+            let region = self.region();
+            let region_meta = region.in_memory_metadata();
+            let alter_req = AlterRequest {
+                operation: alter_op,
+                version: region_meta.version(),
+            };
+            // Alter the region.
+            logging::debug!(
+                "start altering region {} of table {}, with request {:?}",
+                region.name(),
+                table_name,
+                alter_req,
+            );
+            region
+                .alter(alter_req)
+                .await
+                .map_err(BoxedError::new)
+                .context(table_error::TableOperationSnafu)?;
+        }
         // Update in memory metadata of the table.
         self.set_table_info(new_info);
 
@@ -432,22 +441,24 @@ fn create_alter_operation(
     table_name: &str,
     alter_kind: &AlterKind,
     table_meta: &mut TableMeta,
-) -> TableResult<AlterOperation> {
+) -> TableResult<Option<AlterOperation>> {
     match alter_kind {
         AlterKind::AddColumns { columns } => {
             create_add_columns_operation(table_name, columns, table_meta)
         }
-        AlterKind::DropColumns { names } => Ok(AlterOperation::DropColumns {
+        AlterKind::DropColumns { names } => Ok(Some(AlterOperation::DropColumns {
             names: names.to_vec(),
-        }),
+        })),
+        // No need to build alter operation when reaming tables.
+        AlterKind::RenameTable { .. } => Ok(None),
     }
 }
 
 fn create_add_columns_operation(
     table_name: &str,
     requests: &[AddColumnRequest],
     table_meta: &mut TableMeta,
-) -> TableResult<AlterOperation> {
+) -> TableResult<Option<AlterOperation>> {
     let columns = requests
         .iter()
         .map(|request| {
@@ -461,7 +472,7 @@ fn create_add_columns_operation(
         })
         .collect::<TableResult<Vec<_>>>()?;
 
-    Ok(AlterOperation::AddColumns { columns })
+    Ok(Some(AlterOperation::AddColumns { columns }))
 }
 
 #[cfg(test)]

src/sql/src/statements/alter.rs

@@ -80,12 +80,8 @@ impl TryFrom<AlterTable> for AlterExpr {
                     drop_columns: vec![DropColumn { name: name.value }],
                 })
             }
-            AlterTableOperation::RenameTable { .. } => {
-                // TODO update proto to support alter table name
-                return UnsupportedAlterTableStatementSnafu {
-                    msg: "rename table not supported yet",
-                }
-                .fail();
+            AlterTableOperation::RenameTable { new_table_name } => {
+                alter_expr::Kind::RenameTable(api::v1::RenameTable { new_table_name })
             }
         };
         let expr = AlterExpr {

src/table/src/metadata.rs

@@ -132,6 +132,8 @@ impl TableMeta {
         match alter_kind {
             AlterKind::AddColumns { columns } => self.add_columns(table_name, columns),
             AlterKind::DropColumns { names } => self.remove_columns(table_name, names),
+            // No need to rebuild table meta when renaming tables.
+            AlterKind::RenameTable { .. } => Ok(TableMetaBuilder::default()),
         }
     }
 

src/table/src/requests.rs

@@ -80,6 +80,7 @@ pub struct AddColumnRequest {
 pub enum AlterKind {
     AddColumns { columns: Vec<AddColumnRequest> },
     DropColumns { names: Vec<String> },
+    RenameTable { new_table_name: String },
 }
 
 /// Drop table request