refactor: use secrecy SerectString to hold secrets option (#3804)

* build: centralize secrecy dependency

Signed-off-by: tison <wander4096@gmail.com>

* add secrecy to sql crate

Signed-off-by: tison <wander4096@gmail.com>

* try impl

Signed-off-by: tison <wander4096@gmail.com>

* update test

Signed-off-by: tison <wander4096@gmail.com>

* make linters happy

Signed-off-by: tison <wander4096@gmail.com>

* bundle secrecy

Signed-off-by: tison <wander4096@gmail.com>

* bundle secrecy

Signed-off-by: tison <wander4096@gmail.com>

* replace secrecy

Signed-off-by: tison <wander4096@gmail.com>

* tidy clones

Signed-off-by: tison <wander4096@gmail.com>

* fixup

Signed-off-by: tison <wander4096@gmail.com>

* fixup

Signed-off-by: tison <wander4096@gmail.com>

* updated

Signed-off-by: tison <wander4096@gmail.com>

* Apply suggestions from code review

Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>

* use BTreeMap

Signed-off-by: tison <wander4096@gmail.com>

* tidy

Signed-off-by: tison <wander4096@gmail.com>

---------

Signed-off-by: tison <wander4096@gmail.com>
Co-authored-by: LFC <990479+MichaelScofield@users.noreply.github.com>

Cargo.lock

@@ -700,13 +700,13 @@ version = "0.7.2"
 dependencies = [
  "api",
  "async-trait",
+ "common-base",
  "common-error",
  "common-macro",
  "common-telemetry",
  "common-test-util",
  "digest",
  "notify",
- "secrecy",
  "sha1",
  "snafu",
  "sql",
@@ -1682,6 +1682,7 @@ dependencies = [
  "serde",
  "snafu",
  "toml 0.8.12",
+ "zeroize",
 ]
 
 [[package]]
@@ -2877,7 +2878,6 @@ dependencies = [
  "prost 0.12.4",
  "query",
  "reqwest",
- "secrecy",
  "serde",
  "servers",
  "session",
@@ -8942,16 +8942,6 @@ dependencies = [
  "syn 2.0.60",
 ]
 
-[[package]]
-name = "secrecy"
-version = "0.8.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9bd1c54ea06cfd2f6b63219704de0b9b4f72dcc2b8fdef820be6cd799780e91e"
-dependencies = [
- "serde",
- "zeroize",
-]
-
 [[package]]
 name = "security-framework"
 version = "2.10.0"
@@ -9205,7 +9195,6 @@ dependencies = [
  "rustls-pki-types",
  "schemars",
  "script",
- "secrecy",
  "serde",
  "serde_json",
  "session",
@@ -10303,7 +10292,6 @@ dependencies = [
  "rstest",
  "rstest_reuse",
  "script",
- "secrecy",
  "serde_json",
  "servers",
  "session",

licenserc.toml

@@ -22,6 +22,7 @@ includes = [
 excludes = [
     # copied sources
     "src/common/base/src/readable_size.rs",
+    "src/common/base/src/secrets.rs",
     "src/servers/src/repeated_field.rs",
     "src/servers/src/http/test_helpers.rs",
 ]

src/auth/Cargo.toml

@@ -14,12 +14,12 @@ workspace = true
 [dependencies]
 api.workspace = true
 async-trait.workspace = true
+common-base.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 common-telemetry.workspace = true
 digest = "0.10"
 notify.workspace = true
-secrecy = { version = "0.8", features = ["serde", "alloc"] }
 sha1 = "0.10"
 snafu.workspace = true
 sql.workspace = true

src/auth/src/common.rs

@@ -14,8 +14,8 @@
 
 use std::sync::Arc;
 
+use common_base::secrets::SecretString;
 use digest::Digest;
-use secrecy::SecretString;
 use sha1::Sha1;
 use snafu::{ensure, OptionExt};
 

src/auth/src/tests.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use secrecy::ExposeSecret;
+use common_base::secrets::ExposeSecret;
 
 use crate::error::{
     AccessDeniedSnafu, Result, UnsupportedPasswordTypeSnafu, UserNotFoundSnafu,

src/auth/src/user_provider.rs

@@ -21,7 +21,7 @@ use std::io;
 use std::io::BufRead;
 use std::path::Path;
 
-use secrecy::ExposeSecret;
+use common_base::secrets::ExposeSecret;
 use snafu::{ensure, OptionExt, ResultExt};
 
 use crate::common::{Identity, Password};

src/cmd/src/standalone.rs

@@ -631,7 +631,7 @@ mod tests {
         match &dn_opts.storage.providers[1] {
             datanode::config::ObjectStoreConfig::S3(s3_config) => {
                 assert_eq!(
-                    "Secret([REDACTED alloc::string::String])".to_string(),
+                    "SecretBox<alloc::string::String>([REDACTED])".to_string(),
                     format!("{:?}", s3_config.access_key_id)
                 );
             }

src/common/base/Cargo.toml

@@ -16,6 +16,7 @@ common-macro.workspace = true
 paste = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 snafu.workspace = true
+zeroize = { version = "1.6", default-features = false, features = ["alloc"] }
 
 [dev-dependencies]
 toml.workspace = true

src/common/base/src/lib.rs

@@ -17,6 +17,7 @@ pub mod buffer;
 pub mod bytes;
 #[allow(clippy::all)]
 pub mod readable_size;
+pub mod secrets;
 
 use core::any::Any;
 use std::sync::{Arc, Mutex, MutexGuard};

src/common/base/src/secrets.rs

@@ -0,0 +1,218 @@
+// This file is copied from: https://github.com/iqlusioninc/crates/blob/f98d4ccf/secrecy/src/lib.rs.
+
+//! [`SecretBox`] wrapper type for more carefully handling secret values
+//! (e.g. passwords, cryptographic keys, access tokens or other credentials)
+//!
+//! # Goals
+//!
+//! - Make secret access explicit and easy-to-audit via the
+//!   [`ExposeSecret`] and [`ExposeSecretMut`] traits.
+//! - Prevent accidental leakage of secrets via channels like debug logging
+//! - Ensure secrets are wiped from memory on drop securely
+//!   (using the [`zeroize`] crate)
+//!
+//! Presently this crate favors a simple, `no_std`-friendly, safe i.e.
+//! `forbid(unsafe_code)`-based implementation and does not provide more advanced
+//! memory protection mechanisms e.g. ones based on `mlock(2)`/`mprotect(2)`.
+//! We may explore more advanced protection mechanisms in the future.
+//! Those who don't mind `std` and `libc` dependencies should consider using
+//! the [`secrets`](https://crates.io/crates/secrets) crate.
+//!
+//! # `serde` support
+//!
+//! When the `serde` feature of this crate is enabled, the [`SecretBox`] type will
+//! receive a [`Deserialize`] impl for all `SecretBox<T>` types where
+//! `T: DeserializeOwned`. This allows *loading* secret values from data
+//! deserialized from `serde` (be careful to clean up any intermediate secrets
+//! when doing this, e.g. the unparsed input!)
+//!
+//! To prevent exfiltration of secret values via `serde`, by default `SecretBox<T>`
+//! does *not* receive a corresponding [`Serialize`] impl. If you would like
+//! types of `SecretBox<T>` to be serializable with `serde`, you will need to impl
+//! the [`SerializableSecret`] marker trait on `T`
+
+use std::fmt::Debug;
+use std::{any, fmt};
+
+use serde::{de, ser, Deserialize, Serialize};
+use zeroize::{Zeroize, ZeroizeOnDrop};
+
+/// Wrapper type for strings that contains secrets. See also [SecretBox].
+pub type SecretString = SecretBox<String>;
+
+impl From<String> for SecretString {
+    fn from(value: String) -> Self {
+        SecretString::new(Box::new(value))
+    }
+}
+
+/// Wrapper type for values that contains secrets, which attempts to limit
+/// accidental exposure and ensure secrets are wiped from memory when dropped.
+/// (e.g. passwords, cryptographic keys, access tokens or other credentials)
+///
+/// Access to the secret inner value occurs through the [`ExposeSecret`]
+/// or [`ExposeSecretMut`] traits, which provide methods for accessing the inner secret value.
+pub struct SecretBox<S: Zeroize> {
+    inner_secret: Box<S>,
+}
+
+impl<S: Zeroize> Zeroize for SecretBox<S> {
+    fn zeroize(&mut self) {
+        self.inner_secret.as_mut().zeroize()
+    }
+}
+
+impl<S: Zeroize> Drop for SecretBox<S> {
+    fn drop(&mut self) {
+        self.zeroize()
+    }
+}
+
+impl<S: Zeroize> ZeroizeOnDrop for SecretBox<S> {}
+
+impl<S: Zeroize> From<Box<S>> for SecretBox<S> {
+    fn from(source: Box<S>) -> Self {
+        Self::new(source)
+    }
+}
+
+impl<S: Zeroize> SecretBox<S> {
+    /// Create a secret value using a pre-boxed value.
+    pub fn new(boxed_secret: Box<S>) -> Self {
+        Self {
+            inner_secret: boxed_secret,
+        }
+    }
+}
+
+impl<S: Zeroize + Default> SecretBox<S> {
+    /// Create a secret value using a function that can initialize the vale in-place.
+    pub fn new_with_mut(ctr: impl FnOnce(&mut S)) -> Self {
+        let mut secret = Self::default();
+        ctr(secret.expose_secret_mut());
+        secret
+    }
+}
+
+impl<S: Zeroize + Clone> SecretBox<S> {
+    /// Create a secret value using the provided function as a constructor.
+    ///
+    /// The implementation makes an effort to zeroize the locally constructed value
+    /// before it is copied to the heap, and constructing it inside the closure minimizes
+    /// the possibility of it being accidentally copied by other code.
+    ///
+    /// **Note:** using [`Self::new`] or [`Self::new_with_mut`] is preferable when possible,
+    /// since this method's safety relies on empyric evidence and may be violated on some targets.
+    pub fn new_with_ctr(ctr: impl FnOnce() -> S) -> Self {
+        let mut data = ctr();
+        let secret = Self {
+            inner_secret: Box::new(data.clone()),
+        };
+        data.zeroize();
+        secret
+    }
+
+    /// Same as [`Self::new_with_ctr`], but the constructor can be fallible.
+    ///
+    ///
+    /// **Note:** using [`Self::new`] or [`Self::new_with_mut`] is preferable when possible,
+    /// since this method's safety relies on empyric evidence and may be violated on some targets.
+    pub fn try_new_with_ctr<E>(ctr: impl FnOnce() -> Result<S, E>) -> Result<Self, E> {
+        let mut data = ctr()?;
+        let secret = Self {
+            inner_secret: Box::new(data.clone()),
+        };
+        data.zeroize();
+        Ok(secret)
+    }
+}
+
+impl<S: Zeroize + Default> Default for SecretBox<S> {
+    fn default() -> Self {
+        Self {
+            inner_secret: Box::<S>::default(),
+        }
+    }
+}
+
+impl<S: Zeroize> Debug for SecretBox<S> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "SecretBox<{}>([REDACTED])", any::type_name::<S>())
+    }
+}
+
+impl<S> Clone for SecretBox<S>
+where
+    S: Clone + Zeroize,
+{
+    fn clone(&self) -> Self {
+        SecretBox {
+            inner_secret: self.inner_secret.clone(),
+        }
+    }
+}
+
+impl<S: Zeroize> ExposeSecret<S> for SecretBox<S> {
+    fn expose_secret(&self) -> &S {
+        self.inner_secret.as_ref()
+    }
+}
+
+impl<S: Zeroize> ExposeSecretMut<S> for SecretBox<S> {
+    fn expose_secret_mut(&mut self) -> &mut S {
+        self.inner_secret.as_mut()
+    }
+}
+
+/// Expose a reference to an inner secret
+pub trait ExposeSecret<S> {
+    /// Expose secret: this is the only method providing access to a secret.
+    fn expose_secret(&self) -> &S;
+}
+
+/// Expose a mutable reference to an inner secret
+pub trait ExposeSecretMut<S> {
+    /// Expose secret: this is the only method providing access to a secret.
+    fn expose_secret_mut(&mut self) -> &mut S;
+}
+
+/// Marker trait for secret types which can be [`Serialize`]-d by [`serde`].
+///
+/// When the `serde` feature of this crate is enabled and types are marked with
+/// this trait, they receive a [`Serialize` impl][1] for `SecretBox<T>`.
+/// (NOTE: all types which impl `DeserializeOwned` receive a [`Deserialize`]
+/// impl)
+///
+/// This is done deliberately to prevent accidental exfiltration of secrets
+/// via `serde` serialization.
+///
+/// If you really want to have `serde` serialize those types, use the
+/// [`serialize_with`][2] attribute to specify a serializer that exposes the secret.
+///
+/// [1]: https://docs.rs/secrecy/latest/secrecy/struct.Secret.html#implementations
+/// [2]: https://serde.rs/field-attrs.html#serialize_with
+pub trait SerializableSecret: Serialize {}
+
+impl<'de, T> Deserialize<'de> for SecretBox<T>
+where
+    T: Zeroize + Clone + de::DeserializeOwned + Sized,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        Self::try_new_with_ctr(|| T::deserialize(deserializer))
+    }
+}
+
+impl<T> Serialize for SecretBox<T>
+where
+    T: Zeroize + SerializableSecret + Serialize + Sized,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: ser::Serializer,
+    {
+        self.expose_secret().serialize(serializer)
+    }
+}

src/datanode/Cargo.toml

@@ -50,7 +50,6 @@ prometheus.workspace = true
 prost.workspace = true
 query.workspace = true
 reqwest.workspace = true
-secrecy = { version = "0.8", features = ["serde", "alloc"] }
 serde.workspace = true
 servers.workspace = true
 session.workspace = true

src/datanode/src/config.rs

@@ -15,6 +15,7 @@
 //! Datanode configurations
 
 use common_base::readable_size::ReadableSize;
+use common_base::secrets::SecretString;
 use common_grpc::channel_manager::{
     DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
 };
@@ -24,7 +25,6 @@ use common_wal::config::DatanodeWalConfig;
 use file_engine::config::EngineConfig as FileEngineConfig;
 use meta_client::MetaClientOptions;
 use mito2::config::MitoConfig;
-use secrecy::SecretString;
 use serde::{Deserialize, Serialize};
 use servers::export_metrics::ExportMetricsOption;
 use servers::heartbeat_options::HeartbeatOptions;
@@ -285,7 +285,7 @@ pub enum RegionEngineConfig {
 
 #[cfg(test)]
 mod tests {
-    use secrecy::ExposeSecret;
+    use common_base::secrets::ExposeSecret;
 
     use super::*;
 
@@ -308,7 +308,7 @@ mod tests {
         match &opts.storage.store {
             ObjectStoreConfig::S3(cfg) => {
                 assert_eq!(
-                    "Secret([REDACTED alloc::string::String])".to_string(),
+                    "SecretBox<alloc::string::String>([REDACTED])".to_string(),
                     format!("{:?}", cfg.access_key_id)
                 );
                 assert_eq!("access_key_id", cfg.access_key_id.expose_secret());

src/datanode/src/store/azblob.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_base::secrets::ExposeSecret;
 use common_telemetry::logging::info;
 use object_store::services::Azblob;
 use object_store::{util, ObjectStore};
-use secrecy::ExposeSecret;
 use snafu::prelude::*;
 
 use crate::config::AzblobConfig;

src/datanode/src/store/gcs.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_base::secrets::ExposeSecret;
 use common_telemetry::logging::info;
 use object_store::services::Gcs;
 use object_store::{util, ObjectStore};
-use secrecy::ExposeSecret;
 use snafu::prelude::*;
 
 use crate::config::GcsConfig;

src/datanode/src/store/oss.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_base::secrets::ExposeSecret;
 use common_telemetry::logging::info;
 use object_store::services::Oss;
 use object_store::{util, ObjectStore};
-use secrecy::ExposeSecret;
 use snafu::prelude::*;
 
 use crate::config::OssConfig;

src/datanode/src/store/s3.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_base::secrets::ExposeSecret;
 use common_telemetry::logging::info;
 use object_store::services::S3;
 use object_store::{util, ObjectStore};
-use secrecy::ExposeSecret;
 use snafu::prelude::*;
 
 use crate::config::S3Config;

src/operator/src/expr_factory.rs

@@ -129,13 +129,13 @@ pub(crate) async fn create_external_expr(
             .map_err(BoxedError::new)
             .context(ExternalSnafu)?;
 
-    let mut table_options = create.options;
+    let mut table_options = create.options.into_map();
 
-    let (object_store, files) = prepare_file_table_files(&table_options.map)
+    let (object_store, files) = prepare_file_table_files(&table_options)
         .await
         .context(PrepareFileTableSnafu)?;
 
-    let file_column_schemas = infer_file_table_schema(&object_store, &files, &table_options.map)
+    let file_column_schemas = infer_file_table_schema(&object_store, &files, &table_options)
         .await
         .context(InferFileTableSchemaSnafu)?
         .column_schemas;
@@ -176,7 +176,7 @@ pub(crate) async fn create_external_expr(
         time_index,
         primary_keys,
         create_if_not_exists: create.if_not_exists,
-        table_options: table_options.map,
+        table_options,
         table_id: None,
         engine: create.engine.to_string(),
     };
@@ -192,7 +192,8 @@ pub fn create_to_expr(create: &CreateTable, query_ctx: QueryContextRef) -> Resul
 
     let time_index = find_time_index(&create.constraints)?;
     let table_options = HashMap::from(
-        &TableOptions::try_from(create.options.as_ref()).context(UnrecognizedTableOptionSnafu)?,
+        &TableOptions::try_from_iter(create.options.to_str_map())
+            .context(UnrecognizedTableOptionSnafu)?,
     );
 
     let primary_keys = find_primary_keys(&create.columns, &create.constraints)?;

src/operator/src/statement.rs

@@ -317,8 +317,8 @@ fn to_copy_table_request(stmt: CopyTable, query_ctx: QueryContextRef) -> Result<
         schema_name,
         table_name,
         location,
-        with: with.map,
-        connection: connection.map,
+        with: with.into_map(),
+        connection: connection.into_map(),
         pattern,
         direction,
         timestamp_range,
@@ -340,8 +340,8 @@ fn to_copy_database_request(
         catalog_name,
         schema_name: database_name,
         location: arg.location,
-        with: arg.with.map,
-        connection: arg.connection.map,
+        with: arg.with.into_map(),
+        connection: arg.connection.into_map(),
         time_range,
     })
 }

src/operator/src/statement/ddl.rs

@@ -830,7 +830,7 @@ fn create_table_info(
         })
         .collect::<Result<Vec<_>>>()?;
 
-    let table_options = TableOptions::try_from(&create_table.table_options)
+    let table_options = TableOptions::try_from_iter(&create_table.table_options)
         .context(UnrecognizedTableOptionSnafu)?;
     let table_options = merge_options(table_options, schema_opts);
 

src/query/src/sql/show_create_table.rs

@@ -14,8 +14,6 @@
 
 //! Implementation of `SHOW CREATE TABLE` statement.
 
-use std::collections::HashMap;
-
 use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, SchemaRef, COMMENT_KEY};
 use humantime::format_duration;
 use snafu::ResultExt;
@@ -33,8 +31,7 @@ use crate::error::{ConvertSqlTypeSnafu, ConvertSqlValueSnafu, Result, SqlSnafu};
 
 fn create_sql_options(table_meta: &TableMeta) -> OptionMap {
     let table_opts = &table_meta.options;
-    let mut options = HashMap::with_capacity(4 + table_opts.extra_options.len());
-
+    let mut options = OptionMap::default();
     if let Some(write_buffer_size) = table_opts.write_buffer_size {
         options.insert(
             WRITE_BUFFER_SIZE_KEY.to_string(),
@@ -44,7 +41,6 @@ fn create_sql_options(table_meta: &TableMeta) -> OptionMap {
     if let Some(ttl) = table_opts.ttl {
         options.insert(TTL_KEY.to_string(), format_duration(ttl).to_string());
     }
-
     for (k, v) in table_opts
         .extra_options
         .iter()
@@ -52,8 +48,7 @@ fn create_sql_options(table_meta: &TableMeta) -> OptionMap {
     {
         options.insert(k.to_string(), v.to_string());
     }
-
-    OptionMap { map: options }
+    options
 }
 
 #[inline]

src/servers/Cargo.toml

@@ -86,7 +86,6 @@ rustls = "0.22"
 rustls-pemfile = "2.0"
 rustls-pki-types = "1.0"
 schemars.workspace = true
-secrecy = { version = "0.8", features = ["serde", "alloc"] }
 serde.workspace = true
 serde_json.workspace = true
 session.workspace = true

src/servers/src/http/authorize.rs

@@ -21,6 +21,7 @@ use axum::middleware::Next;
 use axum::response::{IntoResponse, Response};
 use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
+use common_base::secrets::SecretString;
 use common_catalog::consts::DEFAULT_SCHEMA_NAME;
 use common_catalog::parse_catalog_and_schema_from_db_string;
 use common_error::ext::ErrorExt;
@@ -28,7 +29,6 @@ use common_telemetry::warn;
 use common_time::timezone::parse_timezone;
 use common_time::Timezone;
 use headers::Header;
-use secrecy::SecretString;
 use session::context::QueryContextBuilder;
 use snafu::{ensure, OptionExt, ResultExt};
 
@@ -320,7 +320,7 @@ fn extract_influxdb_user_from_query(query: &str) -> (Option<&str>, Option<&str>)
 mod tests {
     use std::assert_matches::assert_matches;
 
-    use secrecy::ExposeSecret;
+    use common_base::secrets::ExposeSecret;
 
     use super::*;
 

src/sql/src/parsers/copy_parser.rs

@@ -378,20 +378,17 @@ mod tests {
             stmt.database_name
         );
         assert_eq!(
-            [("format".to_string(), "parquet".to_string())]
+            [("format", "parquet")]
                 .into_iter()
                 .collect::<HashMap<_, _>>(),
-            stmt.with.map
+            stmt.with.to_str_map()
         );
 
         assert_eq!(
-            [
-                ("foo".to_string(), "Bar".to_string()),
-                ("one".to_string(), "two".to_string())
-            ]
-            .into_iter()
-            .collect::<HashMap<_, _>>(),
-            stmt.connection.map
+            [("foo", "Bar"), ("one", "two")]
+                .into_iter()
+                .collect::<HashMap<_, _>>(),
+            stmt.connection.to_str_map()
         );
     }
 
@@ -417,20 +414,17 @@ mod tests {
             stmt.database_name
         );
         assert_eq!(
-            [("format".to_string(), "parquet".to_string())]
+            [("format", "parquet")]
                 .into_iter()
                 .collect::<HashMap<_, _>>(),
-            stmt.with.map
+            stmt.with.to_str_map()
         );
 
         assert_eq!(
-            [
-                ("foo".to_string(), "Bar".to_string()),
-                ("one".to_string(), "two".to_string())
-            ]
-            .into_iter()
-            .collect::<HashMap<_, _>>(),
-            stmt.connection.map
+            [("foo", "Bar"), ("one", "two")]
+                .into_iter()
+                .collect::<HashMap<_, _>>(),
+            stmt.connection.to_str_map()
         );
     }
 }

src/sql/src/parsers/create_parser.rs

@@ -1638,10 +1638,11 @@ ENGINE=mito";
                         ..
                     }
                 );
-                let options = &c.options;
-                assert_eq!(1, options.map.len());
-                let (k, v) = options.map.iter().next().unwrap();
-                assert_eq!(("ttl", "10s"), (k.as_str(), v.as_str()));
+                assert_eq!(1, c.options.len());
+                assert_eq!(
+                    [("ttl", "10s")].into_iter().collect::<HashMap<_, _>>(),
+                    c.options.to_str_map()
+                );
             }
             _ => unreachable!(),
         }

src/sql/src/statements.rs

@@ -43,7 +43,6 @@ use datatypes::schema::constraint::{CURRENT_TIMESTAMP, CURRENT_TIMESTAMP_FN};
 use datatypes::schema::{ColumnDefaultConstraint, ColumnSchema, COMMENT_KEY};
 use datatypes::types::{cast, TimestampType};
 use datatypes::value::{OrderedF32, OrderedF64, Value};
-use itertools::Itertools;
 pub use option_map::OptionMap;
 use snafu::{ensure, OptionExt, ResultExt};
 use sqlparser::ast::{ExactNumberInfo, UnaryOperator};
@@ -59,29 +58,6 @@ use crate::error::{
     SerializeColumnDefaultConstraintSnafu, TimestampOverflowSnafu, UnsupportedDefaultValueSnafu,
 };
 
-const REDACTED_OPTIONS: [&str; 2] = ["access_key_id", "secret_access_key"];
-
-/// Convert the options into redacted and sorted key-value string. Options with key in
-/// [REDACTED_OPTIONS] will be converted into `<key> = '******'`.
-fn redact_and_sort_options(options: &OptionMap) -> Vec<String> {
-    let options = options.as_ref();
-    let mut result = Vec::with_capacity(options.len());
-    let keys = options.keys().sorted();
-    for key in keys {
-        if let Some(val) = options.get(key) {
-            let redacted = REDACTED_OPTIONS
-                .iter()
-                .any(|opt| opt.eq_ignore_ascii_case(key));
-            if redacted {
-                result.push(format!("{key} = '******'"));
-            } else {
-                result.push(format!("{key} = '{}'", val.escape_default()));
-            }
-        }
-    }
-    result
-}
-
 fn parse_string_to_value(
     column_name: &str,
     s: String,

src/sql/src/statements/copy.rs

@@ -17,7 +17,7 @@ use std::fmt::Display;
 use sqlparser::ast::ObjectName;
 use sqlparser_derive::{Visit, VisitMut};
 
-use crate::statements::{redact_and_sort_options, OptionMap};
+use crate::statements::OptionMap;
 
 #[derive(Debug, Clone, PartialEq, Eq, Visit, VisitMut)]
 pub enum Copy {
@@ -53,12 +53,12 @@ impl Display for CopyTable {
                 (&args.with, &args.connection)
             }
         };
-        if !with.map.is_empty() {
-            let options = redact_and_sort_options(with);
+        if !with.is_empty() {
+            let options = with.kv_pairs();
             write!(f, " WITH ({})", options.join(", "))?;
         }
-        if !connection.map.is_empty() {
-            let options = redact_and_sort_options(connection);
+        if !connection.is_empty() {
+            let options = connection.kv_pairs();
             write!(f, " CONNECTION ({})", options.join(", "))?;
         }
         Ok(())
@@ -84,12 +84,12 @@ impl Display for CopyDatabase {
                 (&args.with, &args.connection)
             }
         };
-        if !with.map.is_empty() {
-            let options = redact_and_sort_options(with);
+        if !with.is_empty() {
+            let options = with.kv_pairs();
             write!(f, " WITH ({})", options.join(", "))?;
         }
-        if !connection.map.is_empty() {
-            let options = redact_and_sort_options(connection);
+        if !connection.is_empty() {
+            let options = connection.kv_pairs();
             write!(f, " CONNECTION ({})", options.join(", "))?;
         }
         Ok(())

src/sql/src/statements/create.rs

@@ -20,7 +20,7 @@ use sqlparser::ast::{Expr, Query};
 use sqlparser_derive::{Visit, VisitMut};
 
 use crate::ast::{ColumnDef, Ident, ObjectName, TableConstraint, Value as SqlValue};
-use crate::statements::{redact_and_sort_options, OptionMap};
+use crate::statements::OptionMap;
 
 const LINE_SEP: &str = ",\n";
 const COMMA_SEP: &str = ", ";
@@ -155,8 +155,8 @@ impl Display for CreateTable {
             writeln!(f, "{partitions}")?;
         }
         writeln!(f, "ENGINE={}", &self.engine)?;
-        if !self.options.map.is_empty() {
-            let options = redact_and_sort_options(&self.options);
+        if !self.options.is_empty() {
+            let options = self.options.kv_pairs();
             write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
         }
         Ok(())
@@ -213,8 +213,8 @@ impl Display for CreateExternalTable {
         writeln!(f, "{}", format_table_constraint(&self.constraints))?;
         writeln!(f, ")")?;
         writeln!(f, "ENGINE={}", &self.engine)?;
-        if !self.options.map.is_empty() {
-            let options = redact_and_sort_options(&self.options);
+        if !self.options.is_empty() {
+            let options = self.options.kv_pairs();
             write!(f, "WITH(\n{}\n)", format_list_indent!(options))?;
         }
         Ok(())
@@ -410,7 +410,7 @@ ENGINE=mito
                 .unwrap();
         match &result[0] {
             Statement::CreateTable(c) => {
-                assert_eq!(2, c.options.map.len());
+                assert_eq!(2, c.options.len());
             }
             _ => unreachable!(),
         }

src/sql/src/statements/option_map.rs

@@ -12,52 +12,135 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod visit;
-mod visit_mut;
+use std::collections::{BTreeMap, HashMap};
+use std::ops::ControlFlow;
 
-use std::borrow::Borrow;
-use std::collections::HashMap;
-use std::iter::FromIterator;
+use common_base::secrets::{ExposeSecret, ExposeSecretMut, SecretString};
+use sqlparser::ast::{Visit, VisitMut, Visitor, VisitorMut};
+
+const REDACTED_OPTIONS: [&str; 2] = ["access_key_id", "secret_access_key"];
 
 /// Options hashmap.
-/// Because the trait `Visit` and `VisitMut` is not implemented for `HashMap<String, String>`, we have to wrap it and implement them by ourself.
-#[derive(Clone, Eq, PartialEq, Debug)]
+#[derive(Clone, Debug, Default)]
 pub struct OptionMap {
-    pub map: HashMap<String, String>,
+    options: BTreeMap<String, String>,
+    secrets: BTreeMap<String, SecretString>,
 }
 
 impl OptionMap {
     pub fn insert(&mut self, k: String, v: String) {
-        self.map.insert(k, v);
+        if REDACTED_OPTIONS.contains(&k.as_str()) {
+            self.secrets.insert(k, SecretString::new(Box::new(v)));
+        } else {
+            self.options.insert(k, v);
+        }
     }
 
     pub fn get(&self, k: &str) -> Option<&String> {
-        self.map.get(k)
+        if let Some(value) = self.options.get(k) {
+            Some(value)
+        } else if let Some(value) = self.secrets.get(k) {
+            Some(value.expose_secret())
+        } else {
+            None
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.options.is_empty() && self.secrets.is_empty()
+    }
+
+    pub fn len(&self) -> usize {
+        self.options.len() + self.secrets.len()
+    }
+
+    pub fn to_str_map(&self) -> HashMap<&str, &str> {
+        let mut map = HashMap::with_capacity(self.len());
+        map.extend(self.options.iter().map(|(k, v)| (k.as_str(), v.as_str())));
+        map.extend(
+            self.secrets
+                .iter()
+                .map(|(k, v)| (k.as_str(), v.expose_secret().as_str())),
+        );
+        map
+    }
+
+    pub fn into_map(self) -> HashMap<String, String> {
+        let mut map = HashMap::with_capacity(self.len());
+        map.extend(self.options);
+        map.extend(
+            self.secrets
+                .into_iter()
+                .map(|(k, v)| (k, v.expose_secret().to_string())),
+        );
+        map
+    }
+
+    pub fn kv_pairs(&self) -> Vec<String> {
+        let mut result = Vec::with_capacity(self.options.len() + self.secrets.len());
+        for (k, v) in self.options.iter() {
+            result.push(format!("{k} = '{}'", v.escape_default()));
+        }
+        for (k, _) in self.secrets.iter() {
+            result.push(format!("{k} = '******'"));
+        }
+        result
     }
 }
 
 impl From<HashMap<String, String>> for OptionMap {
-    fn from(map: HashMap<String, String>) -> Self {
-        Self { map }
-    }
-}
-
-impl AsRef<HashMap<String, String>> for OptionMap {
-    fn as_ref(&self) -> &HashMap<String, String> {
-        &self.map
-    }
-}
-
-impl Borrow<HashMap<String, String>> for OptionMap {
-    fn borrow(&self) -> &HashMap<String, String> {
-        &self.map
-    }
-}
-
-impl FromIterator<(String, String)> for OptionMap {
-    fn from_iter<I: IntoIterator<Item = (String, String)>>(iter: I) -> Self {
-        Self {
-            map: iter.into_iter().collect(),
+    fn from(value: HashMap<String, String>) -> Self {
+        let mut result = OptionMap::default();
+        for (k, v) in value.into_iter() {
+            result.insert(k, v);
         }
+        result
+    }
+}
+
+impl PartialEq for OptionMap {
+    fn eq(&self, other: &Self) -> bool {
+        if self.options.ne(&other.options) {
+            return false;
+        }
+
+        if self.secrets.len() != other.secrets.len() {
+            return false;
+        }
+
+        self.secrets.iter().all(|(key, value)| {
+            other
+                .secrets
+                .get(key)
+                .map_or(false, |v| value.expose_secret() == v.expose_secret())
+        })
+    }
+}
+
+impl Eq for OptionMap {}
+
+impl Visit for OptionMap {
+    fn visit<V: Visitor>(&self, visitor: &mut V) -> ControlFlow<V::Break> {
+        for (k, v) in &self.options {
+            k.visit(visitor)?;
+            v.visit(visitor)?;
+        }
+        for (k, v) in &self.secrets {
+            k.visit(visitor)?;
+            v.expose_secret().visit(visitor)?;
+        }
+        ControlFlow::Continue(())
+    }
+}
+
+impl VisitMut for OptionMap {
+    fn visit<V: VisitorMut>(&mut self, visitor: &mut V) -> ControlFlow<V::Break> {
+        for (_, v) in self.options.iter_mut() {
+            v.visit(visitor)?;
+        }
+        for (_, v) in self.secrets.iter_mut() {
+            v.expose_secret_mut().visit(visitor)?;
+        }
+        ControlFlow::Continue(())
     }
 }

src/sql/src/statements/option_map/visit.rs

@@ -1,29 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// 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::ops::ControlFlow;
-
-use sqlparser::ast::{Visit, Visitor};
-
-use crate::statements::OptionMap;
-
-impl Visit for OptionMap {
-    fn visit<V: Visitor>(&self, visitor: &mut V) -> ControlFlow<V::Break> {
-        for (k, v) in &self.map {
-            k.visit(visitor)?;
-            v.visit(visitor)?;
-        }
-        ControlFlow::Continue(())
-    }
-}

src/sql/src/statements/option_map/visit_mut.rs

@@ -1,28 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// 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::ops::ControlFlow;
-
-use sqlparser::ast::{VisitMut, VisitorMut};
-
-use crate::statements::OptionMap;
-
-impl VisitMut for OptionMap {
-    fn visit<V: VisitorMut>(&mut self, visitor: &mut V) -> ControlFlow<V::Break> {
-        for (_, v) in self.map.iter_mut() {
-            v.visit(visitor)?;
-        }
-        ControlFlow::Continue(())
-    }
-}

src/table/src/requests.rs

@@ -29,8 +29,7 @@ use serde::{Deserialize, Serialize};
 use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, PHYSICAL_TABLE_METADATA_KEY};
 use store_api::mito_engine_options::is_mito_engine_option_key;
 
-use crate::error;
-use crate::error::ParseTableOptionSnafu;
+use crate::error::{ParseTableOptionSnafu, Result};
 use crate::metadata::{TableId, TableVersion};
 use crate::table_reference::TableReference;
 
@@ -81,12 +80,18 @@ pub const WRITE_BUFFER_SIZE_KEY: &str = "write_buffer_size";
 pub const TTL_KEY: &str = "ttl";
 pub const STORAGE_KEY: &str = "storage";
 
-impl TryFrom<&HashMap<String, String>> for TableOptions {
-    type Error = error::Error;
-
-    fn try_from(value: &HashMap<String, String>) -> Result<Self, Self::Error> {
+impl TableOptions {
+    pub fn try_from_iter<T: ToString, U: IntoIterator<Item = (T, T)>>(
+        iter: U,
+    ) -> Result<TableOptions> {
         let mut options = TableOptions::default();
-        if let Some(write_buffer_size) = value.get(WRITE_BUFFER_SIZE_KEY) {
+
+        let kvs: HashMap<String, String> = iter
+            .into_iter()
+            .map(|(k, v)| (k.to_string(), v.to_string()))
+            .collect();
+
+        if let Some(write_buffer_size) = kvs.get(WRITE_BUFFER_SIZE_KEY) {
             let size = ReadableSize::from_str(write_buffer_size).map_err(|_| {
                 ParseTableOptionSnafu {
                     key: WRITE_BUFFER_SIZE_KEY,
@@ -97,7 +102,7 @@ impl TryFrom<&HashMap<String, String>> for TableOptions {
             options.write_buffer_size = Some(size)
         }
 
-        if let Some(ttl) = value.get(TTL_KEY) {
+        if let Some(ttl) = kvs.get(TTL_KEY) {
             let ttl_value = ttl
                 .parse::<humantime::Duration>()
                 .map_err(|_| {
@@ -110,13 +115,12 @@ impl TryFrom<&HashMap<String, String>> for TableOptions {
                 .into();
             options.ttl = Some(ttl_value);
         }
-        options.extra_options = HashMap::from_iter(value.iter().filter_map(|(k, v)| {
-            if k != WRITE_BUFFER_SIZE_KEY && k != TTL_KEY {
-                Some((k.clone(), v.clone()))
-            } else {
-                None
-            }
-        }));
+
+        options.extra_options = HashMap::from_iter(
+            kvs.into_iter()
+                .filter(|(k, _)| k != WRITE_BUFFER_SIZE_KEY && k != TTL_KEY),
+        );
+
         Ok(options)
     }
 }
@@ -304,7 +308,7 @@ mod tests {
             extra_options: HashMap::new(),
         };
         let serialized_map = HashMap::from(&options);
-        let serialized = TableOptions::try_from(&serialized_map).unwrap();
+        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
         assert_eq!(options, serialized);
 
         let options = TableOptions {
@@ -313,7 +317,7 @@ mod tests {
             extra_options: HashMap::new(),
         };
         let serialized_map = HashMap::from(&options);
-        let serialized = TableOptions::try_from(&serialized_map).unwrap();
+        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
         assert_eq!(options, serialized);
 
         let options = TableOptions {
@@ -322,7 +326,7 @@ mod tests {
             extra_options: HashMap::from([("a".to_string(), "A".to_string())]),
         };
         let serialized_map = HashMap::from(&options);
-        let serialized = TableOptions::try_from(&serialized_map).unwrap();
+        let serialized = TableOptions::try_from_iter(&serialized_map).unwrap();
         assert_eq!(options, serialized);
     }
 }

tests-integration/Cargo.toml

@@ -49,7 +49,6 @@ prost.workspace = true
 query.workspace = true
 rstest = "0.17"
 rstest_reuse = "0.5"
-secrecy = "0.8"
 serde_json.workspace = true
 servers = { workspace = true, features = ["testing"] }
 session.workspace = true

tests-integration/src/test_util.rs

@@ -21,6 +21,7 @@ use std::time::Duration;
 use auth::UserProviderRef;
 use axum::Router;
 use catalog::kvbackend::KvBackendCatalogManager;
+use common_base::secrets::ExposeSecret;
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::schema_name::SchemaNameKey;
 use common_runtime::Builder as RuntimeBuilder;
@@ -39,7 +40,6 @@ use futures::future::BoxFuture;
 use object_store::services::{Azblob, Gcs, Oss, S3};
 use object_store::test_util::TempFolder;
 use object_store::ObjectStore;
-use secrecy::ExposeSecret;
 use servers::grpc::builder::GrpcServerBuilder;
 use servers::grpc::greptime_handler::GreptimeRequestHandler;
 use servers::grpc::{GrpcServer, GrpcServerConfig};