refactor: alter view only need to modify table option

src/query/service/src/interpreters/interpreter_view_alter.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use databend_common_exception::ErrorCode;
@@ -22,6 +23,8 @@ use databend_common_meta_app::schema::CreateTableReq;
 use databend_common_meta_app::schema::DropTableByIdReq;
 use databend_common_meta_app::schema::TableMeta;
 use databend_common_meta_app::schema::TableNameIdent;
+use databend_common_meta_app::schema::UpsertTableOptionReq;
+use databend_common_meta_types::MatchSeq;
 use databend_common_sql::plans::AlterViewPlan;
 use databend_common_sql::Planner;
 use databend_common_storages_view::view_table::VIEW_ENGINE;
@@ -54,27 +57,12 @@ impl Interpreter for AlterViewInterpreter {
 
     #[async_backtrace::framed]
     async fn execute2(&self) -> Result<PipelineBuildResult> {
-        // drop view
         let catalog = self.ctx.get_catalog(&self.plan.catalog).await?;
         if let Ok(tbl) = catalog
             .get_table(&self.plan.tenant, &self.plan.database, &self.plan.view_name)
             .await
         {
-            let db = catalog
-                .get_database(&self.plan.tenant, &self.plan.database)
-                .await?;
-            catalog
-                .drop_table_by_id(DropTableByIdReq {
-                    if_exists: true,
-                    tenant: self.plan.tenant.clone(),
-                    table_name: self.plan.view_name.clone(),
-                    tb_id: tbl.get_id(),
-                    db_id: db.get_db_info().ident.db_id,
-                })
-                .await?;
-
-            // create new view
-            let mut options = BTreeMap::new();
+            let mut options = HashMap::new();
             let subquery = if self.plan.column_names.is_empty() {
                 self.plan.subquery.clone()
             } else {
@@ -94,22 +82,17 @@ impl Interpreter for AlterViewInterpreter {
                     self.plan.column_names.join(", ")
                 )
             };
-            options.insert("query".to_string(), subquery);
+            options.insert("query".to_string(), Some(subquery));
 
-            let plan = CreateTableReq {
-                create_option: CreateOption::CreateIfNotExists,
-                name_ident: TableNameIdent {
-                    tenant: self.plan.tenant.clone(),
-                    db_name: self.plan.database.clone(),
-                    table_name: self.plan.view_name.clone(),
-                },
-                table_meta: TableMeta {
-                    engine: VIEW_ENGINE.to_string(),
-                    options,
-                    ..Default::default()
-                },
+            let req = UpsertTableOptionReq {
+                table_id: tbl.get_id(),
+                seq: MatchSeq::Exact(tbl.get_table_info().ident.seq),
+                options,
             };
-            catalog.create_table(plan).await?;
+
+            catalog
+                .upsert_table_option(&self.plan.tenant, &self.plan.database, req)
+                .await?;
 
             Ok(PipelineBuildResult::create())
         } else {