feat: Implement Modify Table Connection

src/common/storage/src/operator.rs

@@ -615,7 +615,7 @@ pub async fn check_operator(
 
     GlobalIORuntime::instance()
         .spawn(async move {
-            let res = op.stat("/").await;
+            let res = op.stat("databend_storage_checker").await;
             match res {
                 Ok(_) => Ok(()),
                 Err(e) if e.kind() == opendal::ErrorKind::NotFound => Ok(()),
@@ -626,7 +626,7 @@ pub async fn check_operator(
         .expect("join must succeed")
         .map_err(|cause| {
             ErrorCode::StorageUnavailable(format!(
-                "current configured storage is not available: config: {:?}, cause: {cause}",
+                "current configured storage is not valid: config: {:?}, cause: {cause}",
                 params
             ))
         })

src/meta/app/src/storage/storage_params.rs

@@ -58,6 +58,28 @@ impl Default for StorageParams {
 }
 
 impl StorageParams {
+    /// Get the storage type as a string.
+    pub fn storage_type(&self) -> String {
+        match self {
+            StorageParams::Azblob(_) => "azblob".to_string(),
+            StorageParams::Fs(_) => "fs".to_string(),
+            StorageParams::Ftp(_) => "ftp".to_string(),
+            StorageParams::Gcs(_) => "gcs".to_string(),
+            StorageParams::Hdfs(_) => "hdfs".to_string(),
+            StorageParams::Http(_) => "http".to_string(),
+            StorageParams::Ipfs(_) => "ipfs".to_string(),
+            StorageParams::Memory => "memory".to_string(),
+            StorageParams::Moka(_) => "moka".to_string(),
+            StorageParams::Obs(_) => "obs".to_string(),
+            StorageParams::Oss(_) => "oss".to_string(),
+            StorageParams::S3(_) => "s3".to_string(),
+            StorageParams::Webhdfs(_) => "webhdfs".to_string(),
+            StorageParams::Cos(_) => "cos".to_string(),
+            StorageParams::Huggingface(_) => "huggingface".to_string(),
+            StorageParams::None => "none".to_string(),
+        }
+    }
+
     /// Whether this storage params is secure.
     ///
     /// Query will forbid this storage config unless `allow_insecure` has been enabled.
@@ -157,6 +179,37 @@ impl StorageParams {
 
         Ok(sp)
     }
+
+    /// Apply the update from another StorageParams.
+    ///
+    /// Only specific storage params like `credential` can be updated.
+    pub fn apply_update(self, other: Self) -> Result<Self> {
+        match (self, other) {
+            (StorageParams::Azblob(mut s1), StorageParams::Azblob(s2)) => {
+                s1.account_name = s2.account_name;
+                s1.account_key = s2.account_key;
+                s1.network_config = s2.network_config;
+                Ok(Self::Azblob(s1))
+            }
+            (StorageParams::Gcs(mut s1), StorageParams::Gcs(s2)) => {
+                s1.credential = s2.credential;
+                s1.network_config = s2.network_config;
+                Ok(Self::Gcs(s1))
+            }
+            (StorageParams::S3(mut s1), StorageParams::S3(s2)) => {
+                s1.access_key_id = s2.access_key_id;
+                s1.secret_access_key = s2.secret_access_key;
+                s1.security_token = s2.security_token;
+                s1.master_key = s2.master_key;
+                s1.network_config = s2.network_config;
+                Ok(Self::S3(s1))
+            }
+            (s1, s2) => Err(ErrorCode::StorageOther(format!(
+                "Cannot apply update from {:?} to {:?}",
+                &s1, &s2
+            ))),
+        }
+    }
 }
 
 /// StorageParams will be displayed by `{protocol}://{key1=value1},{key2=value2}`

src/query/ast/src/ast/statements/table.rs

@@ -466,6 +466,9 @@ pub enum AlterTableAction {
         targets: Vec<Identifier>,
     },
     RefreshTableCache,
+    ModifyConnection {
+        new_connection: BTreeMap<String, String>,
+    },
 }
 
 impl Display for AlterTableAction {
@@ -538,6 +541,11 @@ impl Display for AlterTableAction {
             AlterTableAction::RefreshTableCache => {
                 write!(f, "REFRESH CACHE")?;
             }
+            AlterTableAction::ModifyConnection { new_connection } => {
+                write!(f, "CONNECTION=(")?;
+                write_space_separated_string_map(f, new_connection)?;
+                write!(f, ")")?;
+            }
         };
         Ok(())
     }

src/query/ast/src/parser/statement.rs

@@ -3991,6 +3991,15 @@ pub fn alter_table_action(i: Input) -> IResult<AlterTableAction> {
         |(_, _)| AlterTableAction::RefreshTableCache,
     );
 
+    let modify_table_connection = map(
+        rule! {
+            CONNECTION ~ ^"=" ~ #connection_options
+        },
+        |(_, _, connection_options)| AlterTableAction::ModifyConnection {
+            new_connection: connection_options,
+        },
+    );
+
     rule!(
         #alter_table_cluster_key
         | #drop_table_cluster_key
@@ -4005,6 +4014,7 @@ pub fn alter_table_action(i: Input) -> IResult<AlterTableAction> {
         | #set_table_options
         | #unset_table_options
         | #refresh_cache
+        | #modify_table_connection
     )(i)
 }
 

src/query/ast/tests/it/parser.rs

@@ -925,6 +925,8 @@ SELECT * from s;"#,
         r#"CREATE SEQUENCE seq comment='test'"#,
         r#"DESCRIBE SEQUENCE seq"#,
         r#"SHOW SEQUENCES LIKE '%seq%'"#,
+        r#"ALTER TABLE p1 CONNECTION=(CONNECTION_NAME='test')"#,
+        r#"ALTER table t connection=(access_key_id ='x' secret_access_key ='y' endpoint_url='http://127.0.0.1:9900')"#,
     ];
 
     for case in cases {

src/query/ast/tests/it/testdata/stmt.txt

@@ -26341,3 +26341,81 @@ ShowSequences {
 }
 
 
+---------- Input ----------
+ALTER TABLE p1 CONNECTION=(CONNECTION_NAME='test')
+---------- Output ---------
+ALTER TABLE p1 CONNECTION=(connection_name = 'test')
+---------- AST ------------
+AlterTable(
+    AlterTableStmt {
+        if_exists: false,
+        table_reference: Table {
+            span: Some(
+                12..14,
+            ),
+            catalog: None,
+            database: None,
+            table: Identifier {
+                span: Some(
+                    12..14,
+                ),
+                name: "p1",
+                quote: None,
+                ident_type: None,
+            },
+            alias: None,
+            temporal: None,
+            with_options: None,
+            pivot: None,
+            unpivot: None,
+            sample: None,
+        },
+        action: ModifyConnection {
+            new_connection: {
+                "connection_name": "test",
+            },
+        },
+    },
+)
+
+
+---------- Input ----------
+ALTER table t connection=(access_key_id ='x' secret_access_key ='y' endpoint_url='http://127.0.0.1:9900')
+---------- Output ---------
+ALTER TABLE t CONNECTION=(access_key_id = 'x' endpoint_url = 'http://127.0.0.1:9900' secret_access_key = 'y')
+---------- AST ------------
+AlterTable(
+    AlterTableStmt {
+        if_exists: false,
+        table_reference: Table {
+            span: Some(
+                12..13,
+            ),
+            catalog: None,
+            database: None,
+            table: Identifier {
+                span: Some(
+                    12..13,
+                ),
+                name: "t",
+                quote: None,
+                ident_type: None,
+            },
+            alias: None,
+            temporal: None,
+            with_options: None,
+            pivot: None,
+            unpivot: None,
+            sample: None,
+        },
+        action: ModifyConnection {
+            new_connection: {
+                "access_key_id": "x",
+                "endpoint_url": "http://127.0.0.1:9900",
+                "secret_access_key": "y",
+            },
+        },
+    },
+)
+
+

src/query/service/src/interpreters/access/privilege_access.rs

@@ -1008,6 +1008,9 @@ impl AccessChecker for PrivilegeAccess {
             Plan::ModifyTableComment(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Alter, false, false).await?
             }
+            Plan::ModifyTableConnection(plan) => {
+                self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Alter, false, false).await?
+            }
             Plan::DropTableColumn(plan) => {
                 self.validate_table_access(&plan.catalog, &plan.database, &plan.table, UserPrivilegeType::Alter, false, false).await?
             }

src/query/service/src/interpreters/interpreter_factory.rs

@@ -34,6 +34,7 @@ use super::interpreter_mutation::MutationInterpreter;
 use super::interpreter_table_index_create::CreateTableIndexInterpreter;
 use super::interpreter_table_index_drop::DropTableIndexInterpreter;
 use super::interpreter_table_index_refresh::RefreshTableIndexInterpreter;
+use super::interpreter_table_modify_connection::ModifyTableConnectionInterpreter;
 use super::interpreter_table_set_options::SetOptionsInterpreter;
 use super::interpreter_user_stage_drop::DropUserStageInterpreter;
 use super::*;
@@ -334,6 +335,9 @@ impl InterpreterFactory {
             Plan::ModifyTableComment(new_comment) => Ok(Arc::new(
                 ModifyTableCommentInterpreter::try_create(ctx, *new_comment.clone())?,
             )),
+            Plan::ModifyTableConnection(new_connection) => Ok(Arc::new(
+                ModifyTableConnectionInterpreter::try_create(ctx, *new_connection.clone())?,
+            )),
             Plan::RenameTableColumn(rename_table_column) => Ok(Arc::new(
                 RenameTableColumnInterpreter::try_create(ctx, *rename_table_column.clone())?,
             )),

src/query/service/src/interpreters/interpreter_table_modify_connection.rs

@@ -0,0 +1,143 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use databend_common_ast::ast::UriLocation;
+use databend_common_catalog::table::TableExt;
+use databend_common_exception::ErrorCode;
+use databend_common_exception::Result;
+use databend_common_meta_app::schema::DatabaseType;
+use databend_common_sql::binder::parse_storage_params_from_uri;
+use databend_common_sql::plans::ModifyTableConnectionPlan;
+use databend_common_storage::check_operator;
+use databend_common_storage::init_operator;
+use databend_common_storages_stream::stream_table::STREAM_ENGINE;
+use databend_common_storages_view::view_table::VIEW_ENGINE;
+use log::debug;
+
+use crate::interpreters::interpreter_table_add_column::commit_table_meta;
+use crate::interpreters::Interpreter;
+use crate::pipelines::PipelineBuildResult;
+use crate::sessions::QueryContext;
+use crate::sessions::TableContext;
+
+pub struct ModifyTableConnectionInterpreter {
+    ctx: Arc<QueryContext>,
+    plan: ModifyTableConnectionPlan,
+}
+
+impl ModifyTableConnectionInterpreter {
+    pub fn try_create(ctx: Arc<QueryContext>, plan: ModifyTableConnectionPlan) -> Result<Self> {
+        Ok(ModifyTableConnectionInterpreter { ctx, plan })
+    }
+}
+
+#[async_trait::async_trait]
+impl Interpreter for ModifyTableConnectionInterpreter {
+    fn name(&self) -> &str {
+        "ModifyTableConnectionInterpreter"
+    }
+
+    fn is_ddl(&self) -> bool {
+        true
+    }
+
+    #[async_backtrace::framed]
+    async fn execute2(&self) -> Result<PipelineBuildResult> {
+        let catalog_name = self.plan.catalog.as_str();
+        let db_name = self.plan.database.as_str();
+        let tbl_name = self.plan.table.as_str();
+
+        let table = self
+            .ctx
+            .get_catalog(catalog_name)
+            .await?
+            .get_table(&self.ctx.get_tenant(), db_name, tbl_name)
+            .await?;
+
+        // check mutability
+        table.check_mutable()?;
+
+        let table_info = table.get_table_info();
+        let engine = table.engine();
+        if matches!(engine, VIEW_ENGINE | STREAM_ENGINE) {
+            return Err(ErrorCode::TableEngineNotSupported(format!(
+                "{}.{} engine is {} that doesn't support alter",
+                &self.plan.database, &self.plan.table, engine
+            )));
+        }
+        if table_info.db_type != DatabaseType::NormalDB {
+            return Err(ErrorCode::TableEngineNotSupported(format!(
+                "{}.{} doesn't support alter",
+                &self.plan.database, &self.plan.table
+            )));
+        }
+        let Some(old_sp) = table_info.meta.storage_params.clone() else {
+            return Err(ErrorCode::TableEngineNotSupported(format!(
+                "{}.{} is not an external table, cannot alter connection",
+                &self.plan.database, &self.plan.table
+            )));
+        };
+
+        debug!("old storage params before update: {old_sp:?}");
+
+        // This location is used to parse the storage parameters from the URI.
+        //
+        // We don't really this this location to replace the old one, we just parse it out and change the storage parameters on needs.
+        let mut location = UriLocation::new(
+            // The storage type is not changeable, we just use the old one.
+            old_sp.storage_type(),
+            // name is not changeable, we just use a dummy value here.
+            "test".to_string(),
+            // root is not changeable, we just use a dummy value here.
+            "/".to_string(),
+            self.plan.new_connection.clone(),
+        );
+        // NOTE: never use this storage params directly.
+        let updated_sp = parse_storage_params_from_uri(
+            &mut location,
+            Some(self.ctx.as_ref() as _),
+            "when ALTER TABLE CONNECTION",
+        )
+        .await?;
+
+        debug!("storage params used for update: {updated_sp:?}");
+        let new_sp = old_sp.apply_update(updated_sp)?;
+        debug!("new storage params been updated: {new_sp:?}");
+
+        // Check the storage params via init operator.
+        let op = init_operator(&new_sp).map_err(|err| {
+            ErrorCode::InvalidConfig(format!(
+                "Input storage config for stage is invalid: {err:?}"
+            ))
+        })?;
+        check_operator(&op, &new_sp).await?;
+
+        let catalog = self.ctx.get_catalog(self.plan.catalog.as_str()).await?;
+        let mut new_table_meta = table_info.meta.clone();
+        new_table_meta.storage_params = Some(new_sp);
+
+        commit_table_meta(
+            &self.ctx,
+            table.as_ref(),
+            table_info,
+            new_table_meta,
+            catalog,
+        )
+        .await?;
+
+        Ok(PipelineBuildResult::create())
+    }
+}