Repository: risingwavelabs/risingwave (language: python)

Issue:
Sqlsmith: Sql feature generation
As we add more features, these should be tested via sqlsmith as well.

SQL Features:
- [ ] Generate other table functions: `generate_series`, `unnest`. See: https://github.com/singularity-data/risingwave/issues/4540
- [ ] Generate array_agg, jsonb_agg, jsonb_object_agg
- [ ] Generate select from System Tables
- [ ] Generate `NATURAL JOIN`
- [ ] Generate `JOIN` with `USING`
- [ ] Generate query w/o `GROUP BY`
- [ ] Randomly toggle session variables
- [ ] Generate `pow`/`exp`.
- [ ] Generate TopN by group: https://www.risingwave.dev/docs/current/sql-pattern-topn/
- [ ] Generate jsonb exprs, see: https://github.com/risingwavelabs/risingwave/issues/7714
- [ ] #7132
- [x] Generate order-by-clauses with `NULLS { FIRST | LAST }` (will be done in https://github.com/risingwavelabs/risingwave/pull/10864/commits/000cbc11cbc04f20abfd127b1bf4b3fd24e75c44)
- [ ] Generate `max`,`min`,`count` for `timestamptz`
- [ ] Generate `bit_and`, `bit_or` and `big_xor` in batch queries, and `bit_xor` in streaming queries.
- [ ] Generate array range access.
- [ ] https://github.com/risingwavelabs/risingwave/issues/10241
- [ ] Generate window functions other than `row_number`, `rank`.
- [ ] Generate `ALTER` DDL statements.\
- [ ] https://github.com/risingwavelabs/risingwave/issues/10516
- [ ] Generate temporal filter in sqlsmith


Discussion / hints:

Fix patch:
diff --git a/ci/workflows/pull-request.yml b/ci/workflows/pull-request.yml
index 8b852b6bb0b0a..45502ab3037c0 100644
--- a/ci/workflows/pull-request.yml
+++ b/ci/workflows/pull-request.yml
@@ -818,7 +818,7 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
-    timeout_in_minutes: 30
+    timeout_in_minutes: 32
     retry: *auto-retry
 
   # FIXME(kwannoel): Let the github PR labeller label it, if sqlsmith source files has changes.
diff --git a/proto/ddl_service.proto b/proto/ddl_service.proto
index 59d8b2f999371..8daa21a3085ad 100644
--- a/proto/ddl_service.proto
+++ b/proto/ddl_service.proto
@@ -419,6 +419,19 @@ message DropSecretResponse {
   WaitVersion version = 1;
 }
 
+message AlterSecretRequest {
+  uint32 secret_id = 1;
+  string name = 2;
+  bytes value = 3;
+  uint32 database_id = 4;
+  uint32 schema_id = 5;
+  uint32 owner_id = 6;
+}
+
+message AlterSecretResponse {
+  WaitVersion version = 1;
+}
+
 message CreateConnectionRequest {
   message PrivateLink {
     catalog.Connection.PrivateLinkService.PrivateLinkProvider provider = 1;
@@ -515,6 +528,7 @@ service DdlService {
   rpc CreateTable(CreateTableRequest) returns (CreateTableResponse);
   rpc CreateSecret(CreateSecretRequest) returns (CreateSecretResponse);
   rpc DropSecret(DropSecretRequest) returns (DropSecretResponse);
+  rpc AlterSecret(AlterSecretRequest) returns (AlterSecretResponse);
   rpc AlterName(AlterNameRequest) returns (AlterNameResponse);
   rpc AlterSource(AlterSourceRequest) returns (AlterSourceResponse);
   rpc AlterOwner(AlterOwnerRequest) returns (AlterOwnerResponse);
diff --git a/src/common/secret/src/secret_manager.rs b/src/common/secret/src/secret_manager.rs
index b6a71a4c3ebe8..723c0f5791d4d 100644
--- a/src/common/secret/src/secret_manager.rs
+++ b/src/common/secret/src/secret_manager.rs
@@ -74,7 +74,23 @@ impl LocalSecretManager {
 
     pub fn add_secret(&self, secret_id: SecretId, secret: Vec<u8>) {
         let mut secret_guard = self.secrets.write();
-        secret_guard.insert(secret_id, secret);
+        if secret_guard.insert(secret_id, secret).is_some() {
+            tracing::error!(
+                secret_id = secret_id,
+                "adding a secret but it already exists, overwriting it"
+            );
+        };
+    }
+
+    pub fn update_secret(&self, secret_id: SecretId, secret: Vec<u8>) {
+        let mut secret_guard = self.secrets.write();
+        if secret_guard.insert(secret_id, secret).is_none() {
+            tracing::error!(
+                secret_id = secret_id,
+                "updating a secret but it does not exist, adding it"
+            );
+        }
+        self.remove_secret_file_if_exist(&secret_id);
     }
 
     pub fn init_secrets(&self, secrets: Vec<PbSecret>) {
@@ -174,9 +190,7 @@ impl LocalSecretManager {
     }
 
     fn get_secret_value(pb_secret_bytes: &[u8]) -> SecretResult<Vec<u8>> {
-        let pb_secret = risingwave_pb::secret::Secret::decode(pb_secret_bytes)
-            .context("failed to decode secret")?;
-        let secret_value = match pb_secret.get_secret_backend().unwrap() {
+        let secret_value = match Self::get_pb_secret_backend(pb_secret_bytes)? {
             risingwave_pb::secret::secret::SecretBackend::Meta(backend) => backend.value.clone(),
             risingwave_pb::secret::secret::SecretBackend::HashicorpVault(_) => {
                 return Err(anyhow!("hashicorp_vault backend is not implemented yet").into())
@@ -184,4 +198,13 @@ impl LocalSecretManager {
         };
         Ok(secret_value)
     }
+
+    /// Get the secret backend from the given decrypted secret bytes.
+    pub fn get_pb_secret_backend(
+        pb_secret_bytes: &[u8],
+    ) -> SecretResult<risingwave_pb::secret::secret::SecretBackend> {
+        let pb_secret = risingwave_pb::secret::Secret::decode(pb_secret_bytes)
+            .context("failed to decode secret")?;
+        Ok(pb_secret.get_secret_backend().unwrap().clone())
+    }
 }
diff --git a/src/compute/src/observer/observer_manager.rs b/src/compute/src/observer/observer_manager.rs
index c028c1e851613..3f3a57c363930 100644
--- a/src/compute/src/observer/observer_manager.rs
+++ b/src/compute/src/observer/observer_manager.rs
@@ -38,6 +38,9 @@ impl ObserverState for ComputeObserverNode {
                     Operation::Delete => {
                         LocalSecretManager::global().remove_secret(s.id);
                     }
+                    Operation::Update => {
+                        LocalSecretManager::global().update_secret(s.id, s.value);
+                    }
                     _ => {
                         panic!("error type notification");
                     }
diff --git a/src/frontend/src/catalog/catalog_service.rs b/src/frontend/src/catalog/catalog_service.rs
index cc7cac86570c5..f7dcb919a6ad7 100644
--- a/src/frontend/src/catalog/catalog_service.rs
+++ b/src/frontend/src/catalog/catalog_service.rs
@@ -178,6 +178,16 @@ pub trait CatalogWriter: Send + Sync {
 
     async fn drop_secret(&self, secret_id: SecretId) -> Result<()>;
 
+    async fn alter_secret(
+        &self,
+        secret_id: u32,
+        secret_name: String,
+        database_id: u32,
+        schema_id: u32,
+        owner_id: u32,
+        payload: Vec<u8>,
+    ) -> Result<()>;
+
     async fn alter_name(
         &self,
         object_id: alter_name_request::Object,
@@ -511,6 +521,29 @@ impl CatalogWriter for CatalogWriterImpl {
         let version = self.meta_client.alter_swap_rename(object).await?;
         self.wait_version(version).await
     }
+
+    async fn alter_secret(
+        &self,
+        secret_id: u32,
+        secret_name: String,
+        database_id: u32,
+        schema_id: u32,
+        owner_id: u32,
+        payload: Vec<u8>,
+    ) -> Result<()> {
+        let version = self
+            .meta_client
+            .alter_secret(
+                secret_id,
+                secret_name,
+                database_id,
+                schema_id,
+                owner_id,
+                payload,
+            )
+            .await?;
+        self.wait_version(version).await
+    }
 }
 
 impl CatalogWriterImpl {
diff --git a/src/frontend/src/catalog/secret_catalog.rs b/src/frontend/src/catalog/secret_catalog.rs
index d1f9048baf0e7..db50fbc201bdb 100644
--- a/src/frontend/src/catalog/secret_catalog.rs
+++ b/src/frontend/src/catalog/secret_catalog.rs
@@ -14,7 +14,7 @@
 
 use risingwave_pb::catalog::PbSecret;
 
-use crate::catalog::{DatabaseId, OwnedByUserCatalog, SecretId};
+use crate::catalog::{DatabaseId, OwnedByUserCatalog, SchemaId, SecretId};
 use crate::user::UserId;
 
 #[derive(Clone, Debug, PartialEq, Eq, Hash)]
@@ -22,6 +22,7 @@ pub struct SecretCatalog {
     pub id: SecretId,
     pub name: String,
     pub database_id: DatabaseId,
+    pub schema_id: SchemaId,
     pub value: Vec<u8>,
     pub owner: UserId,
 }
@@ -34,6 +35,7 @@ impl From<&PbSecret> for SecretCatalog {
             owner: value.owner,
             name: value.name.clone(),
             value: value.value.clone(),
+            schema_id: value.schema_id,
         }
     }
 }
diff --git a/src/frontend/src/handler/alter_secret.rs b/src/frontend/src/handler/alter_secret.rs
new file mode 100644
index 0000000000000..3ab5ece2ee518
--- /dev/null
+++ b/src/frontend/src/handler/alter_secret.rs
@@ -0,0 +1,101 @@
+// Copyright 2024 RisingWave 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 anyhow::anyhow;
+use pgwire::pg_response::StatementType;
+use prost::Message;
+use risingwave_common::bail_not_implemented;
+use risingwave_common::license::Feature;
+use risingwave_common::secret::LocalSecretManager;
+use risingwave_pb::secret::secret;
+use risingwave_sqlparser::ast::{AlterSecretOperation, ObjectName, SqlOption};
+
+use super::create_secret::{get_secret_payload, secret_to_str};
+use super::drop_secret::fetch_secret_catalog_with_db_schema_id;
+use crate::error::Result;
+use crate::handler::{HandlerArgs, RwPgResponse};
+use crate::WithOptions;
+
+pub async fn handle_alter_secret(
+    handler_args: HandlerArgs,
+    secret_name: ObjectName,
+    sql_options: Vec<SqlOption>,
+    operation: AlterSecretOperation,
+) -> Result<RwPgResponse> {
+    Feature::SecretManagement
+        .check_available()
+        .map_err(|e| anyhow::anyhow!(e))?;
+
+    let session = handler_args.session;
+
+    if let Some((secret_catalog, _, _)) =
+        fetch_secret_catalog_with_db_schema_id(&session, &secret_name, false)?
+    {
+        let AlterSecretOperation::ChangeCredential { new_credential } = operation;
+
+        let secret_id = secret_catalog.id.secret_id();
+        let secret_payload = if sql_options.is_empty() {
+            let original_pb_secret_bytes = LocalSecretManager::global()
+                .get_secret(secret_id)
+                .ok_or(anyhow!(
+                    "Failed to get secret in secret manager, secret_id: {}",
+                    secret_id
+                ))?;
+            let original_secret_backend =
+                LocalSecretManager::get_pb_secret_backend(&original_pb_secret_bytes)?;
+            match original_secret_backend {
+                secret::SecretBackend::Meta(_) => {
+                    let new_secret_value_bytes =
+                        secret_to_str(&new_credential)?.as_bytes().to_vec();
+                    let secret_payload = risingwave_pb::secret::Secret {
+                        secret_backend: Some(risingwave_pb::secret::secret::SecretBackend::Meta(
+                            risingwave_pb::secret::SecretMetaBackend {
+                                value: new_secret_value_bytes,
+                            },
+                        )),
+                    };
+                    secret_payload.encode_to_vec()
+                }
+                secret::SecretBackend::HashicorpVault(_) => {
+                    bail_not_implemented!("hashicorp_vault backend is not implemented yet")
+                }
+            }
+        } else {
+            let with_options = WithOptions::try_from(sql_options.as_ref() as &[SqlOption])?;
+            get_secret_payload(new_credential, with_options)?
+        };
+
+        let catalog_writer = session.catalog_writer()?;
+
+        catalog_writer
+            .alter_secret(
+                secret_id,
+                secret_catalog.name.clone(),
+                secret_catalog.database_id,
+                secret_catalog.schema_id,
+                secret_catalog.owner,
+                secret_payload,
+            )
+            .await?;
+
+        Ok(RwPgResponse::empty_result(StatementType::ALTER_SECRET))
+    } else {
+        Ok(RwPgResponse::builder(StatementType::ALTER_SECRET)
+            .notice(format!(
+                "secret \"{}\" does not exist, skipping",
+                secret_name
+            ))
+            .into())
+    }
+}
diff --git a/src/frontend/src/handler/create_secret.rs b/src/frontend/src/handler/create_secret.rs
index 9810751361be3..6d5a6283e3288 100644
--- a/src/frontend/src/handler/create_secret.rs
+++ b/src/frontend/src/handler/create_secret.rs
@@ -37,60 +37,21 @@ pub async fn handle_create_secret(
 
     let session = handler_args.session.clone();
     let db_name = session.database();
-    let (schema_name, connection_name) =
+    let (schema_name, secret_name) =
         Binder::resolve_schema_qualified_name(db_name, stmt.secret_name.clone())?;
 
     if let Err(e) = session.check_secret_name_duplicated(stmt.secret_name.clone()) {
         return if stmt.if_not_exists {
             Ok(PgResponse::builder(StatementType::CREATE_SECRET)
-                .notice(format!("secret \"{}\" exists, skipping", connection_name))
+                .notice(format!("secret \"{}\" exists, skipping", secret_name))
                 .into())
         } else {
             Err(e)
         };
     }
+    let with_options = WithOptions::try_from(stmt.with_properties.0.as_ref() as &[SqlOption])?;
 
-    let secret = secret_to_str(&stmt.credential)?.as_bytes().to_vec();
-
-    // check if the secret backend is supported
-    let with_props = WithOptions::try_from(stmt.with_properties.0.as_ref() as &[SqlOption])?;
-    let secret_payload: Vec<u8> = {
-        if let Some(backend) = with_props.get(SECRET_BACKEND_KEY) {
-            match backend.to_lowercase().as_ref() {
-                SECRET_BACKEND_META => {
-                    let backend = risingwave_pb::secret::Secret {
-                        secret_backend: Some(risingwave_pb::secret::secret::SecretBackend::Meta(
-                            risingwave_pb::secret::SecretMetaBackend { value: secret },
-                        )),
-                    };
-                    backend.encode_to_vec()
-                }
-                SECRET_BACKEND_HASHICORP_VAULT => {
-                    if stmt.credential != Value::Null {
-                        return Err(ErrorCode::InvalidParameterValue(
-                            "credential must be null for hashicorp_vault backend".to_string(),
-                        )
-                        .into());
-                    }
-                    bail_not_implemented!("hashicorp_vault backend is not implemented yet")
-                }
-                _ => {
-                    return Err(ErrorCode::InvalidParameterValue(format!(
-                        "secret backend \"{}\" is not supported. Supported backends are: {}",
-                        backend,
-                        [SECRET_BACKEND_META, SECRET_BACKEND_HASHICORP_VAULT].join(",")
-                    ))
-                    .into());
-                }
-            }
-        } else {
-            return Err(ErrorCode::InvalidParameterValue(format!(
-                "secret backend is not specified in with clause. Supported backends are: {}",
-                [SECRET_BACKEND_META, SECRET_BACKEND_HASHICORP_VAULT].join(",")
-            ))
-            .into());
-        }
-    };
+    let secret_payload = get_secret_payload(stmt.credential, with_options)?;
 
     let (database_id, schema_id) = session.get_database_and_schema_id_for_create(schema_name)?;
 
@@ -108,7 +69,7 @@ pub async fn handle_create_secret(
     Ok(PgResponse::empty_result(StatementType::CREATE_SECRET))
 }
 
-fn secret_to_str(value: &Value) -> Result<String> {
+pub fn secret_to_str(value: &Value) -> Result<String> {
     match value {
         Value::DoubleQuotedString(s) | Value::SingleQuotedString(s) => Ok(s.to_string()),
         _ => Err(ErrorCode::InvalidInputSyntax(
@@ -117,3 +78,41 @@ fn secret_to_str(value: &Value) -> Result<String> {
         .into()),
     }
 }
+
+pub(crate) fn get_secret_payload(credential: Value, with_options: WithOptions) -> Result<Vec<u8>> {
+    let secret = secret_to_str(&credential)?.as_bytes().to_vec();
+
+    if let Some(backend) = with_options.get(SECRET_BACKEND_KEY) {
+        match backend.to_lowercase().as_ref() {
+            SECRET_BACKEND_META => {
+                let backend = risingwave_pb::secret::Secret {
+                    secret_backend: Some(risingwave_pb::secret::secret::SecretBackend::Meta(
+                        risingwave_pb::secret::SecretMetaBackend { value: secret },
+                    )),
+                };
+                Ok(backend.encode_to_vec())
+            }
+            SECRET_BACKEND_HASHICORP_VAULT => {
+                if credential != Value::Null {
+                    return Err(ErrorCode::InvalidParameterValue(
+                        "credential must be null for hashicorp_vault backend".to_string(),
+                    )
+                    .into());
+                }
+                bail_not_implemented!("hashicorp_vault backend is not implemented yet")
+            }
+            _ => Err(ErrorCode::InvalidParameterValue(format!(
+                "secret backend \"{}\" is not supported. Supported backends are: {}",
+                backend,
+                [SECRET_BACKEND_META, SECRET_BACKEND_HASHICORP_VAULT].join(",")
+            ))
+            .into()),
+        }
+    } else {
+        Err(ErrorCode::InvalidParameterValue(format!(
+            "secret backend is not specified in with clause. Supported backends are: {}",
+            [SECRET_BACKEND_META, SECRET_BACKEND_HASHICORP_VAULT].join(",")
+        ))
+        .into())
+    }
+}
diff --git a/src/frontend/src/handler/drop_secret.rs b/src/frontend/src/handler/drop_secret.rs
index eff4b35224b8b..4720d73bfa7e6 100644
--- a/src/frontend/src/handler/drop_secret.rs
+++ b/src/frontend/src/handler/drop_secret.rs
@@ -12,13 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use pgwire::pg_response::StatementType;
 use risingwave_common::license::Feature;
 use risingwave_sqlparser::ast::ObjectName;
 
 use crate::catalog::root_catalog::SchemaPath;
+use crate::catalog::secret_catalog::SecretCatalog;
+use crate::catalog::{DatabaseId, SchemaId};
 use crate::error::Result;
 use crate::handler::{HandlerArgs, RwPgResponse};
+use crate::session::SessionImpl;
 use crate::Binder;
 
 pub async fn handle_drop_secret(
@@ -31,37 +36,54 @@ pub async fn handle_drop_secret(
         .map_err(|e| anyhow::anyhow!(e))?;
 
     let session = handler_args.session;
+
+    if let Some((secret_catalog, _, _)) =
+        fetch_secret_catalog_with_db_schema_id(&session, &secret_name, if_exists)?
+    {
+        let catalog_writer = session.catalog_writer()?;
+        catalog_writer.drop_secret(secret_catalog.id).await?;
+
+        Ok(RwPgResponse::empty_result(StatementType::DROP_SECRET))
+    } else {
+        Ok(RwPgResponse::builder(StatementType::DROP_SECRET)
+            .notice(format!(
+                "secret \"{}\" does not exist, skipping",
+                secret_name
+            ))
+            .into())
+    }
+}
+
+/// Fetch the secret catalog and the `database/schema_id` of the source.
+pub fn fetch_secret_catalog_with_db_schema_id(
+    session: &SessionImpl,
+    secret_name: &ObjectName,
+    if_exists: bool,
+) -> Result<Option<(Arc<SecretCatalog>, DatabaseId, SchemaId)>> {
     let db_name = session.database();
-    let (schema_name, secret_name) = Binder::resolve_schema_qualified_name(db_name, secret_name)?;
+    let (schema_name, secret_name) =
+        Binder::resolve_schema_qualified_name(db_name, secret_name.clone())?;
     let search_path = session.config().search_path();
     let user_name = &session.auth_context().user_name;
-    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
 
-    let secret_id = {
-        let reader = session.env().catalog_reader().read_guard();
-        let (secret, schema_name) =
-            match reader.get_secret_by_name(db_name, schema_path, secret_name.as_str()) {
-                Ok((c, s)) => (c, s),
-                Err(e) => {
-                    return if if_exists {
-                        Ok(RwPgResponse::builder(StatementType::DROP_SECRET)
-                            .notice(format!(
-                                "secret \"{}\" does not exist, skipping",
-                                secret_name
-                            ))
-                            .into())
-                    } else {
-                        Err(e.into())
-                    };
-                }
-            };
-        session.check_privilege_for_drop_alter(schema_name, &**secret)?;
+    let schema_path = SchemaPath::new(schema_name.as_deref(), &search_path, user_name);
 
-        secret.id
-    };
+    let reader = session.env().catalog_reader().read_guard();
+    match reader.get_secret_by_name(db_name, schema_path, &secret_name) {
+        Ok((catalog, schema_name)) => {
+            session.check_privilege_for_drop_alter(schema_name, &**catalog)?;
 
-    let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_secret(secret_id).await?;
+            let db = reader.get_database_by_name(db_name)?;
+            let schema = db.get_schema_by_name(schema_name).unwrap();
 
-    Ok(RwPgResponse::empty_result(StatementType::DROP_SECRET))
+            Ok(Some((Arc::clone(catalog), db.id(), schema.id())))
+        }
+        Err(e) => {
+            if if_exists {
+                Ok(None)
+            } else {
+                Err(e.into())
+            }
+        }
+    }
 }
diff --git a/src/frontend/src/handler/mod.rs b/src/frontend/src/handler/mod.rs
index 9cf94a37c65b0..cfa6dc10277e8 100644
--- a/src/frontend/src/handler/mod.rs
+++ b/src/frontend/src/handler/mod.rs
@@ -42,6 +42,7 @@ use crate::utils::WithOptions;
 mod alter_owner;
 mod alter_parallelism;
 mod alter_rename;
+mod alter_secret;
 mod alter_set_schema;
 mod alter_source_column;
 mod alter_source_with_sr;
@@ -1076,6 +1077,11 @@ pub async fn handle(
         Statement::AlterSystem { param, value } => {
             alter_system::handle_alter_system(handler_args, param, value).await
         }
+        Statement::AlterSecret {
+            name,
+            with_options,
+            operation,
+        } => alter_secret::handle_alter_secret(handler_args, name, with_options, operation).await,
         Statement::StartTransaction { modes } => {
             transaction::handle_begin(handler_args, START_TRANSACTION, modes).await
         }
diff --git a/src/frontend/src/observer/observer_manager.rs b/src/frontend/src/observer/observer_manager.rs
index 92408c2b03885..270c467b04754 100644
--- a/src/frontend/src/observer/observer_manager.rs
+++ b/src/frontend/src/observer/observer_manager.rs
@@ -481,6 +481,9 @@ impl FrontendObserverNode {
             Operation::Delete => {
                 LocalSecretManager::global().remove_secret(secret.id);
             }
+            Operation::Update => {
+                LocalSecretManager::global().update_secret(secret.id, secret.value);
+            }
             _ => {
                 panic!("error type notification");
             }
diff --git a/src/meta/service/src/ddl_service.rs b/src/meta/service/src/ddl_service.rs
index 3556890db04cc..553aaeb986e78 100644
--- a/src/meta/service/src/ddl_service.rs
+++ b/src/meta/service/src/ddl_service.rs
@@ -181,6 +181,27 @@ impl DdlService for DdlServiceImpl {
         Ok(Response::new(DropSecretResponse { version }))
     }
 
+    async fn alter_secret(
+        &self,
+        request: Request<AlterSecretRequest>,
+    ) -> Result<Response<AlterSecretResponse>, Status> {
+        let req = request.into_inner();
+        let pb_secret = Secret {
+            id: req.get_secret_id(),
+            name: req.get_name().clone(),
+            database_id: req.get_database_id(),
+            value: req.get_value().clone(),
+            owner: req.get_owner_id(),
+            schema_id: req.get_schema_id(),
+        };
+        let version = self
+            .ddl_controller
+            .run_command(DdlCommand::AlterSecret(pb_secret))
+            .await?;
+
+        Ok(Response::new(AlterSecretResponse { version }))
+    }
+
     async fn create_schema(
         &self,
         request: Request<CreateSchemaRequest>,
diff --git a/src/meta/src/controller/catalog.rs b/src/meta/src/controller/catalog.rs
index f3dac5c39e2fb..5950f0357eee5 100644
--- a/src/meta/src/controller/catalog.rs
+++ b/src/meta/src/controller/catalog.rs
@@ -1397,6 +1397,43 @@ impl CatalogController {
         Ok(version)
     }
 
+    pub async fn alter_secret(
+        &self,
+        pb_secret: PbSecret,
+        secret_plain_payload: Vec<u8>,
+    ) -> MetaResult<NotificationVersion> {
+        let inner = self.inner.write().await;
+        let owner_id = pb_secret.owner as _;
+        let txn = inner.db.begin().await?;
+        ensure_user_id(owner_id, &txn).await?;
+        ensure_object_id(ObjectType::Database, pb_secret.database_id as _, &txn).await?;
+        ensure_object_id(ObjectType::Schema, pb_secret.schema_id as _, &txn).await?;
+
+        ensure_object_id(ObjectType::Secret, pb_secret.id as _, &txn).await?;
+        let secret: secret::ActiveModel = pb_secret.clone().into();
+        Secret::update(secret).exec(&txn).await?;
+
+        txn.commit().await?;
+
+        // Notify the compute and frontend node plain secret
+        let mut secret_plain = pb_secret;
+        secret_plain.value.clone_from(&secret_plain_payload);
+
+        LocalSecretManager::global().update_secret(secret_plain.id, secret_plain_payload);
+        self.env
+            .notification_manager()
+            .notify_compute_without_version(Operation::Update, Info::Secret(secret_plain.clone()));
+
+        let version = self
+            .notify_frontend(
+                NotificationOperation::Update,
+                NotificationInfo::Secret(secret_plain),
+            )
+            .await;
+
+        Ok(version)
+    }
+
     pub async fn get_secret_by_id(&self, secret_id: SecretId) -> MetaResult<PbSecret> {
         let inner = self.inner.read().await;
         let (secret, obj) = Secret::find_by_id(secret_id)
@@ -1417,7 +1454,7 @@ impl CatalogController {
             .ok_or_else(|| MetaError::catalog_id_not_found("secret", secret_id))?;
         ensure_object_not_refer(ObjectType::Secret, secret_id, &txn).await?;
 
-        // Find affect users with privileges on the connection.
+        // Find affect users with privileges on the secret.
         let to_update_user_ids: Vec<UserId> = UserPrivilege::find()
             .select_only()
             .distinct()
diff --git a/src/meta/src/rpc/ddl_controller.rs b/src/meta/src/rpc/ddl_controller.rs
index bb14ecefc1298..ffb46ea03a6ac 100644
--- a/src/meta/src/rpc/ddl_controller.rs
+++ b/src/meta/src/rpc/ddl_controller.rs
@@ -151,6 +151,7 @@ pub enum DdlCommand {
     CreateConnection(Connection),
     DropConnection(ConnectionId),
     CreateSecret(Secret),
+    AlterSecret(Secret),
     DropSecret(SecretId),
     CommentOn(Comment),
     CreateSubscription(Subscription),
@@ -179,6 +180,7 @@ impl DdlCommand {
             | DdlCommand::CreateConnection(_)
             | DdlCommand::CommentOn(_)
             | DdlCommand::CreateSecret(_)
+            | DdlCommand::AlterSecret(_)
             | DdlCommand::AlterSwapRename(_) => true,
             DdlCommand::CreateStreamingJob(_, _, _, _, _)
             | DdlCommand::CreateSourceWithoutStreamingJob(_)
@@ -350,6 +352,7 @@ impl DdlController {
                 }
                 DdlCommand::CreateSecret(secret) => ctrl.create_secret(secret).await,
                 DdlCommand::DropSecret(secret_id) => ctrl.drop_secret(secret_id).await,
+                DdlCommand::AlterSecret(secret) => ctrl.alter_secret(secret).await,
                 DdlCommand::AlterSourceColumn(source) => ctrl.alter_source(source).await,
                 DdlCommand::CommentOn(comment) => ctrl.comment_on(comment).await,
                 DdlCommand::CreateSubscription(subscription) => {
@@ -520,10 +523,9 @@ impl DdlController {
         .await
     }
 
-    async fn create_secret(&self, mut secret: Secret) -> MetaResult<NotificationVersion> {
-        // The 'secret' part of the request we receive from the frontend is in plaintext;
-        // here, we need to encrypt it before storing it in the catalog.
-        let secret_plain_payload = secret.value.clone();
+    // The 'secret' part of the request we receive from the frontend is in plaintext;
+    // here, we need to encrypt it before storing it in the catalog.
+    fn get_encrypted_payload(&self, secret: &Secret) -> MetaResult<Vec<u8>> {
         let secret_store_private_key = self
             .env
             .opts
@@ -531,16 +533,21 @@ impl DdlController {
             .clone()
             .ok_or_else(|| anyhow!("secret_store_private_key is not configured"))?;
 
-        let encrypted_payload = {
-            let encrypted_secret = SecretEncryption::encrypt(
-                secret_store_private_key.as_slice(),
-                secret.get_value().as_slice(),
-            )
-            .context(format!("failed to encrypt secret {}", secret.name))?;
-            encrypted_secret
-                .serialize()
-                .context(format!("failed to serialize secret {}", secret.name))?
-        };
+        let encrypted_payload = SecretEncryption::encrypt(
+            secret_store_private_key.as_slice(),
+            secret.get_value().as_slice(),
+        )
+        .context(format!("failed to encrypt secret {}", secret.name))?;
+        Ok(encrypted_payload
+            .serialize()
+            .context(format!("failed to serialize secret {}", secret.name))?)
+    }
+
+    async fn create_secret(&self, mut secret: Secret) -> MetaResult<NotificationVersion> {
+        // The 'secret' part of the request we receive from the frontend is in plaintext;
+        // here, we need to encrypt it before storing it in the catalog.
+        let secret_plain_payload = secret.value.clone();
+        let encrypted_payload = self.get_encrypted_payload(&secret)?;
         secret.value = encrypted_payload;
 
         self.metadata_manager
@@ -556,6 +563,16 @@ impl DdlController {
             .await
     }
 
+    async fn alter_secret(&self, mut secret: Secret) -> MetaResult<NotificationVersion> {
+        let secret_plain_payload = secret.value.clone();
+        let encrypted_payload = self.get_encrypted_payload(&secret)?;
+        secret.value = encrypted_payload;
+        self.metadata_manager
+            .catalog_controller
+            .alter_secret(secret, secret_plain_payload)
+            .await
+    }
+
     async fn create_subscription(
         &self,
         mut subscription: Subscription,
diff --git a/src/rpc_client/src/meta_client.rs b/src/rpc_client/src/meta_client.rs
index 8f4e6779b8e56..6efac3f414564 100644
--- a/src/rpc_client/src/meta_client.rs
+++ b/src/rpc_client/src/meta_client.rs
@@ -590,6 +590,29 @@ impl MetaClient {
             .ok_or_else(|| anyhow!("wait version not set"))?)
     }
 
+    pub async fn alter_secret(
+        &self,
+        secret_id: u32,
+        secret_name: String,
+        database_id: u32,
+        schema_id: u32,
+        owner_id: u32,
+        value: Vec<u8>,
+    ) -> Result<WaitVersion> {
+        let request = AlterSecretRequest {
+            secret_id,
+            name: secret_name,
+            database_id,
+            schema_id,
+            owner_id,
+            value,
+        };
+        let resp = self.inner.alter_secret(request).await?;
+        Ok(resp
+            .version
+            .ok_or_else(|| anyhow!("wait version not set"))?)
+    }
+
     pub async fn replace_table(
         &self,
         source: Option<PbSource>,
@@ -2095,6 +2118,7 @@ macro_rules! for_all_meta_rpc {
             ,{ ddl_client, wait, WaitRequest, WaitResponse }
             ,{ ddl_client, auto_schema_change, AutoSchemaChangeRequest, AutoSchemaChangeResponse }
             ,{ ddl_client, alter_swap_rename, AlterSwapRenameRequest, AlterSwapRenameResponse }
+            ,{ ddl_client, alter_secret, AlterSecretRequest, AlterSecretResponse }
             ,{ hummock_client, unpin_version_before, UnpinVersionBeforeRequest, UnpinVersionBeforeResponse }
             ,{ hummock_client, get_current_version, GetCurrentVersionRequest, GetCurrentVersionResponse }
             ,{ hummock_client, replay_version_delta, ReplayVersionDeltaRequest, ReplayVersionDeltaResponse }
diff --git a/src/sqlparser/src/ast/ddl.rs b/src/sqlparser/src/ast/ddl.rs
index d94cf80cb9f2d..35eac31cefa72 100644
--- a/src/sqlparser/src/ast/ddl.rs
+++ b/src/sqlparser/src/ast/ddl.rs
@@ -20,7 +20,7 @@ use core::fmt;
 #[cfg(feature = "serde")]
 use serde::{Deserialize, Serialize};
 
-use super::FormatEncodeOptions;
+use super::{FormatEncodeOptions, Value};
 use crate::ast::{
     display_comma_separated, display_separated, DataType, Expr, Ident, ObjectName, SetVariableValue,
 };
@@ -214,6 +214,12 @@ pub enum AlterConnectionOperation {
     SetSchema { new_schema_name: ObjectName },
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
+pub enum AlterSecretOperation {
+    ChangeCredential { new_credential: Value },
+}
+
 impl fmt::Display for AlterDatabaseOperation {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
@@ -481,6 +487,16 @@ impl fmt::Display for AlterConnectionOperation {
     }
 }
 
+impl fmt::Display for AlterSecretOperation {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            AlterSecretOperation::ChangeCredential { new_credential } => {
+                write!(f, "AS {new_credential}")
+            }
+        }
+    }
+}
+
 /// An `ALTER COLUMN` (`Statement::AlterTable`) operation
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 #[cfg_attr(feature = "serde", derive(Serialize, Deserialize))]
diff --git a/src/sqlparser/src/ast/mod.rs b/src/sqlparser/src/ast/mod.rs
index be9d4ae489f53..9007613f30cee 100644
--- a/src/sqlparser/src/ast/mod.rs
+++ b/src/sqlparser/src/ast/mod.rs
@@ -38,8 +38,8 @@ use winnow::PResult;
 pub use self::data_type::{DataType, StructField};
 pub use self::ddl::{
     AlterColumnOperation, AlterConnectionOperation, AlterDatabaseOperation, AlterFunctionOperation,
-    AlterSchemaOperation, AlterTableOperation, ColumnDef, ColumnOption, ColumnOptionDef,
-    ReferentialAction, SourceWatermark, TableConstraint,
+    AlterSchemaOperation, AlterSecretOperation, AlterTableOperation, ColumnDef, ColumnOption,
+    ColumnOptionDef, ReferentialAction, SourceWatermark, TableConstraint,
 };
 pub use self::legacy_source::{
     get_delimiter, AvroSchema, CompatibleFormatEncode, DebeziumAvroSchema, ProtobufSchema,
@@ -1438,6 +1438,13 @@ pub enum Statement {
         name: ObjectName,
         operation: AlterConnectionOperation,
     },
+    /// ALTER SECRET
+    AlterSecret {
+        /// Secret name
+        name: ObjectName,
+        with_options: Vec<SqlOption>,
+        operation: AlterSecretOperation,
+    },
     /// DESCRIBE TABLE OR SOURCE
     Describe {
         /// Table or Source name
@@ -1967,6 +1974,13 @@ impl fmt::Display for Statement {
             Statement::AlterConnection { name, operation } => {
                 write!(f, "ALTER CONNECTION {} {}", name, operation)
             }
+            Statement::AlterSecret { name, with_options, operation } => {
+                write!(f, "ALTER SECRET {}", name)?;
+                if !with_options.is_empty() {
+                    write!(f, " WITH ({})", display_comma_separated(with_options))?;
+                }
+                write!(f, " {}", operation)
+            }
             Statement::Discard(t) => write!(f, "DISCARD {}", t),
             Statement::Drop(stmt) => write!(f, "DROP {}", stmt),
             Statement::DropFunction {
diff --git a/src/sqlparser/src/parser.rs b/src/sqlparser/src/parser.rs
index cf7d8e4394574..81dbc54e84c80 100644
--- a/src/sqlparser/src/parser.rs
+++ b/src/sqlparser/src/parser.rs
@@ -3026,9 +3026,11 @@ impl Parser<'_> {
             self.parse_alter_system()
         } else if self.parse_keyword(Keyword::SUBSCRIPTION) {
             self.parse_alter_subscription()
+        } else if self.parse_keyword(Keyword::SECRET) {
+            self.parse_alter_secret()
         } else {
             self.expected(
-                "DATABASE, SCHEMA, TABLE, INDEX, MATERIALIZED, VIEW, SINK, SUBSCRIPTION, SOURCE, FUNCTION, USER or SYSTEM after ALTER"
+                "DATABASE, SCHEMA, TABLE, INDEX, MATERIALIZED, VIEW, SINK, SUBSCRIPTION, SOURCE, FUNCTION, USER, SECRET or SYSTEM after ALTER"
             )
         }
     }
@@ -3541,6 +3543,19 @@ impl Parser<'_> {
         Ok(Statement::AlterSystem { param, value })
     }
 
+    pub fn parse_alter_secret(&mut self) -> PResult<Statement> {
+        let secret_name = self.parse_object_name()?;
+        let with_options = self.parse_with_properties()?;
+        self.expect_keyword(Keyword::AS)?;
+        let new_credential = self.parse_value()?;
+        let operation = AlterSecretOperation::ChangeCredential { new_credential };
+        Ok(Statement::AlterSecret {
+            name: secret_name,
+            with_options,
+            operation,
+        })
+    }
+
     /// Parse a copy statement
     pub fn parse_copy(&mut self) -> PResult<Statement> {
         let table_name = self.parse_object_name()?;
diff --git a/src/utils/pgwire/src/pg_response.rs b/src/utils/pgwire/src/pg_response.rs
index 5718533f67097..33e1ec9c5c22b 100644
--- a/src/utils/pgwire/src/pg_response.rs
+++ b/src/utils/pgwire/src/pg_response.rs
@@ -89,6 +89,7 @@ pub enum StatementType {
     ALTER_FUNCTION,
     ALTER_CONNECTION,
     ALTER_SYSTEM,
+    ALTER_SECRET,
     REVOKE_PRIVILEGE,
     // Introduce ORDER_BY statement type cuz Calcite unvalidated AST has SqlKind.ORDER_BY. Note
     // that Statement Type is not designed to be one to one mapping with SqlKind.


Test patch:
diff --git a/e2e_test/ddl/secret.slt b/e2e_test/ddl/secret.slt
index 6522565986a18..cf9d16214da38 100644
--- a/e2e_test/ddl/secret.slt
+++ b/e2e_test/ddl/secret.slt
@@ -40,6 +40,26 @@ create secret secret_1 with (
   backend = 'meta'
 ) as 'demo_secret';
 
+statement ok
+alter secret secret_1 with (
+  backend = 'meta'
+) as 'demo_secret_altered';
+
+statement ok
+alter secret secret_1 as 'demo_secret_altered_again';
+
+statement error
+alter secret secret_2 with (
+  backend = 'meta'
+) as 'demo_secret_altered';
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Catalog error
+  2: secret not found: secret_2
+
+
 # wait for support for hashicorp_vault backend
 # statement ok
 # create secret secret_2 with (
diff --git a/e2e_test/sink/iceberg_sink.slt b/e2e_test/sink/iceberg_sink.slt
index b08abd8a4918c..ced999e546d14 100644
--- a/e2e_test/sink/iceberg_sink.slt
+++ b/e2e_test/sink/iceberg_sink.slt
@@ -10,6 +10,11 @@ CREATE MATERIALIZED VIEW mv6 AS SELECT * FROM t6;
 statement ok
 CREATE SECRET iceberg_s3_access_key WITH (
   backend = 'meta'
+) as 'hummockadmin_wrong';
+
+statement ok
+ALTER SECRET iceberg_s3_access_key WITH (
+  backend = 'meta'
 ) as 'hummockadmin';
 
 statement ok
diff --git a/e2e_test/source_legacy/cdc/cdc.share_stream.slt b/e2e_test/source_legacy/cdc/cdc.share_stream.slt
index 60b5f7cd52893..db301c3afc6e5 100644
--- a/e2e_test/source_legacy/cdc/cdc.share_stream.slt
+++ b/e2e_test/source_legacy/cdc/cdc.share_stream.slt
@@ -22,7 +22,31 @@ mysql --protocol=tcp -u root mytest < e2e_test/source_legacy/cdc/mysql_init_data
 statement ok
 create secret mysql_pwd with (
   backend = 'meta'
-) as '${MYSQL_PWD:}';
+) as 'incorrect_password';
+
+# create a cdc source job, with incorrct password
+statement error
+create source mysql_mytest with (
+ connector = 'mysql-cdc',
+ hostname = '${MYSQL_HOST:localhost}',
+ port = '${MYSQL_TCP_PORT:8306}',
+ username = 'rwcdc',
+ password = secret mysql_pwd,
+ database.name = 'mytest',
+ server.id = '5601'
+);
+# The detailed error message is commented out because the user IP in error message may vary in different environments.
+# ----
+# db error: ERROR: Failed to run the query
+# Caused by these errors (recent errors listed first):
+#   1: gRPC request to meta service failed: Internal error
+#   2: failed to create source worker
+#   3: failed to create SplitEnumerator
+#   4: source cannot pass validation
+#   5: Internal error: Access denied for user 'rwcdc'@'172.17.0.1' (using password: YES)
+
+statement ok
+alter secret mysql_pwd as '${MYSQL_PWD:}';
 
 # create a cdc source job, which format fixed to `FORMAT PLAIN ENCODE JSON`
 statement ok
@@ -36,7 +60,6 @@ create source mysql_mytest with (
  server.id = '5601'
 );
 
-
 statement error Should not create MATERIALIZED VIEW or SELECT directly on shared CDC source
 create materialized view mv as select * from mysql_mytest;
 
@@ -62,6 +85,22 @@ from mysql_mytest table 'mytest.products';
 # sleep to ensure (default,'Milk','Milk is a white liquid food') is consumed from Debezium message instead of backfill.
 sleep 10s
 
+### BEGIN test change secret in MySQL and ALTER SECRET in Risingwave
+onlyif can-use-recover
+statement ok
+alter secret mysql_pwd with (
+  backend = 'meta'
+) as 'new_password';
+
+onlyif can-use-recover
+system ok
+mysql --protocol=tcp -u root -e "ALTER USER 'rwcdc'@'%' IDENTIFIED BY 'new_password';"
+
+onlyif can-use-recover
+statement ok
+recover;
+### END
+
 statement error Permission denied
 drop secret mysql_pwd;
 
@@ -621,3 +660,17 @@ query II
 select * from test_pg_default_value;
 ----
 1 noris Shanghai
+
+### BEGIN reset the password to the original one
+onlyif can-use-recover
+statement ok
+alter secret mysql_pwd as '${MYSQL_PWD:}';
+
+onlyif can-use-recover
+system ok
+mysql --protocol=tcp -u root -e "ALTER USER 'rwcdc'@'%' IDENTIFIED BY '${MYSQL_PWD:}';"
+
+onlyif can-use-recover
+statement ok
+recover;
+### END
\ No newline at end of file
diff --git a/src/frontend/src/test_utils.rs b/src/frontend/src/test_utils.rs
index 810724d2bac22..77a9bc23d5a53 100644
--- a/src/frontend/src/test_utils.rs
+++ b/src/frontend/src/test_utils.rs
@@ -654,6 +654,18 @@ impl CatalogWriter for MockCatalogWriter {
     async fn alter_swap_rename(&self, _object: alter_swap_rename_request::Object) -> Result<()> {
         todo!()
     }
+
+    async fn alter_secret(
+        &self,
+        _secret_id: u32,
+        _secret_name: String,
+        _database_id: u32,
+        _schema_id: u32,
+        _owner_id: u32,
+        _payload: Vec<u8>,
+    ) -> Result<()> {
+        unreachable!()
+    }
 }
 
 impl MockCatalogWriter {