fix: concurrent rename two table to same name may cause override (#1368)

* fix: concurrent rename two table to same name may cause override

* fix: concurrently update system catalog table

* fix: correctness

src/catalog/src/local/manager.rs

@@ -419,6 +419,13 @@ impl CatalogManager for LocalCatalogManager {
                 schema: schema_name,
             })?;
 
+        let _lock = self.register_lock.lock().await;
+        ensure!(
+            !schema.table_exist(&request.new_table_name)?,
+            TableExistsSnafu {
+                table: &request.new_table_name
+            }
+        );
         let old_table = schema
             .table(&request.table_name)
             .await?
@@ -437,9 +444,11 @@ impl CatalogManager for LocalCatalogManager {
                 engine,
             )
             .await?;
-        Ok(schema
-            .rename_table(&request.table_name, request.new_table_name)
-            .is_ok())
+
+        let renamed = schema
+            .rename_table(&request.table_name, request.new_table_name.clone())
+            .is_ok();
+        Ok(renamed)
     }
 
     async fn deregister_table(&self, request: DeregisterTableRequest) -> Result<bool> {

src/catalog/src/local/memory.rs

@@ -324,16 +324,20 @@ impl SchemaProvider for MemorySchemaProvider {
 
     fn rename_table(&self, name: &str, new_name: String) -> Result<TableRef> {
         let mut tables = self.tables.write().unwrap();
-        if tables.get(name).is_some() {
-            let table = tables.remove(name).unwrap();
-            tables.insert(new_name, table.clone());
-            Ok(table)
-        } else {
-            TableNotFoundSnafu {
+        let Some(table) = tables.remove(name) else {
+            return TableNotFoundSnafu {
                 table_info: name.to_string(),
             }
-            .fail()?
-        }
+                .fail()?;
+        };
+        let e = match tables.entry(new_name) {
+            Entry::Vacant(e) => e,
+            Entry::Occupied(e) => {
+                return TableExistsSnafu { table: e.key() }.fail();
+            }
+        };
+        e.insert(table.clone());
+        Ok(table)
     }
 
     fn deregister_table(&self, name: &str) -> Result<Option<TableRef>> {
@@ -459,7 +463,7 @@ mod tests {
         assert!(schema.table_exist(table_name).unwrap());
 
         // rename table
-        let new_table_name = "numbers";
+        let new_table_name = "numbers_new";
         let rename_table_req = RenameTableRequest {
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),