chore: simplify setting API (#17138)

Author: drmingdrmer

Cargo.lock

@@ -28,7 +28,6 @@ use futures_util::TryStreamExt;
 use log::debug;
 
 use crate::kvapi;
-use crate::kvapi::GetKVReply;
 use crate::kvapi::ListKVReply;
 use crate::kvapi::MGetKVReply;
 use crate::kvapi::UpsertKVReply;
@@ -61,7 +60,7 @@ pub trait KVApi: Send + Sync {
 
     /// Get a key-value record by key.
     // TODO: #[deprecated(note = "use get_kv_stream() instead")]
-    async fn get_kv(&self, key: &str) -> Result<GetKVReply, Self::Error> {
+    async fn get_kv(&self, key: &str) -> Result<Option<SeqV>, Self::Error> {
         let mut strm = self.get_kv_stream(&[key.to_string()]).await?;
 
         let strm_item = strm

src/meta/kvapi/src/kvapi/api.rs

@@ -44,7 +44,6 @@ pub use role::RoleMgr;
 pub use serde::check_and_upgrade_to_pb;
 pub use serde::deserialize_struct;
 pub use serde::serialize_struct;
-pub use setting::SettingApi;
 pub use setting::SettingMgr;
 pub use stage::StageApi;
 pub use stage::StageMgr;

src/query/management/src/lib.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod setting_api;
 mod setting_mgr;
 
-pub use setting_api::SettingApi;
 pub use setting_mgr::SettingMgr;

src/query/management/src/setting/mod.rs

@@ -14,7 +14,6 @@
 
 use std::sync::Arc;
 
-use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::principal::SettingIdent;
 use databend_common_meta_app::principal::UserSetting;
@@ -23,13 +22,9 @@ use databend_common_meta_kvapi::kvapi;
 use databend_common_meta_kvapi::kvapi::Key;
 use databend_common_meta_types::seq_value::SeqV;
 use databend_common_meta_types::seq_value::SeqValue;
-use databend_common_meta_types::MatchSeq;
-use databend_common_meta_types::MatchSeqExt;
 use databend_common_meta_types::MetaError;
-use databend_common_meta_types::Operation;
 use databend_common_meta_types::UpsertKV;
-
-use crate::setting::SettingApi;
+use futures::TryStreamExt;
 
 pub struct SettingMgr {
     kv_api: Arc<dyn kvapi::KVApi<Error = MetaError>>,
@@ -58,16 +53,14 @@ impl SettingMgr {
 }
 
 // TODO: do not use json for setting value
-#[async_trait::async_trait]
-impl SettingApi for SettingMgr {
+impl SettingMgr {
     #[async_backtrace::framed]
     #[fastrace::trace]
-    async fn set_setting(&self, setting: UserSetting) -> Result<u64> {
+    pub async fn set_setting(&self, setting: UserSetting) -> Result<u64> {
         // Upsert.
-        let seq = MatchSeq::GE(0);
-        let val = Operation::Update(serde_json::to_vec(&setting)?);
+        let val = serde_json::to_vec(&setting)?;
         let key = self.setting_key(&setting.name);
-        let upsert = self.kv_api.upsert_kv(UpsertKV::new(&key, seq, val, None));
+        let upsert = self.kv_api.upsert_kv(UpsertKV::update(&key, &val));
 
         let (_prev, curr) = upsert.await?.unpack();
         let res_seq = curr.seq();
@@ -76,45 +69,38 @@ impl SettingApi for SettingMgr {
 
     #[async_backtrace::framed]
     #[fastrace::trace]
-    async fn get_settings(&self) -> Result<Vec<UserSetting>> {
+    pub async fn get_settings(&self) -> Result<Vec<UserSetting>> {
         let prefix = self.setting_prefix();
-        let values = self.kv_api.prefix_list_kv(&prefix).await?;
+        let mut strm = self.kv_api.list_kv(&prefix).await?;
 
-        let mut settings = Vec::with_capacity(values.len());
-        for (_, value) in values {
-            let setting = serde_json::from_slice::<UserSetting>(&value.data)?;
+        let mut settings = Vec::new();
+        while let Some(item) = strm.try_next().await? {
+            let setting: UserSetting = serde_json::from_slice(&item.value.unwrap().data)?;
             settings.push(setting);
         }
+
         Ok(settings)
     }
 
     #[async_backtrace::framed]
     #[fastrace::trace]
-    async fn get_setting(&self, name: &str, seq: MatchSeq) -> Result<SeqV<UserSetting>> {
+    pub async fn get_setting(&self, name: &str) -> Result<Option<SeqV<UserSetting>>> {
         let key = self.setting_key(name);
         let res = self.kv_api.get_kv(&key).await?;
 
-        let seq_value = res.ok_or_else(|| {
-            ErrorCode::UnknownVariable(format!("Setting '{}' does not exist.", name))
-        })?;
+        let Some(seqv) = res else {
+            return Ok(None);
+        };
 
-        match seq.match_seq(&seq_value) {
-            Ok(_) => Ok(seq_value.try_map(|d| d.try_into())?),
-            Err(_) => Err(ErrorCode::UnknownVariable(format!(
-                "Setting '{}' does not exist.",
-                name
-            ))),
-        }
+        let seqv = seqv.try_map(|d| d.try_into())?;
+        Ok(Some(seqv))
     }
 
     #[async_backtrace::framed]
     #[fastrace::trace]
-    async fn try_drop_setting(&self, name: &str, seq: MatchSeq) -> Result<()> {
+    pub async fn try_drop_setting(&self, name: &str) -> Result<()> {
         let key = self.setting_key(name);
-        let _res = self
-            .kv_api
-            .upsert_kv(UpsertKV::new(&key, seq, Operation::Delete, None))
-            .await?;
+        let _res = self.kv_api.upsert_kv(UpsertKV::delete(&key)).await?;
 
         Ok(())
     }

src/query/management/src/setting/setting_api.rs

@@ -23,7 +23,6 @@ use databend_common_meta_app::tenant::Tenant;
 use databend_common_meta_embedded::MemMeta;
 use databend_common_meta_kvapi::kvapi::KVApi;
 use databend_common_meta_types::seq_value::SeqV;
-use databend_common_meta_types::MatchSeq;
 use fastrace::func_name;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -82,13 +81,13 @@ async fn test_set_setting() -> Result<()> {
     // Get setting.
     {
         let expect = UserSetting::create("max_threads", UserSettingValue::UInt64(1));
-        let actual = mgr.get_setting("max_threads", MatchSeq::GE(0)).await?;
-        assert_eq!(actual.data, expect);
+        let actual = mgr.get_setting("max_threads").await?;
+        assert_eq!(actual.unwrap().data, expect);
     }
 
     // Drop setting.
     {
-        mgr.try_drop_setting("max_threads", MatchSeq::GE(1)).await?;
+        mgr.try_drop_setting("max_threads").await?;
     }
 
     // Get settings.
@@ -99,15 +98,13 @@ async fn test_set_setting() -> Result<()> {
 
     // Get setting.
     {
-        let res = mgr.get_setting("max_threads", MatchSeq::GE(0)).await;
-        assert!(res.is_err());
+        let res = mgr.get_setting("max_threads").await?;
+        assert!(res.is_none());
     }
 
     // Drop setting not exists.
     {
-        let res = mgr
-            .try_drop_setting("max_threads_not", MatchSeq::GE(1))
-            .await;
+        let res = mgr.try_drop_setting("max_threads_not").await;
         assert!(res.is_ok());
     }
 

src/query/management/src/setting/setting_mgr.rs

@@ -19,7 +19,6 @@ databend-common-config = { workspace = true }
 databend-common-exception = { workspace = true }
 databend-common-io = { workspace = true }
 databend-common-meta-app = { workspace = true }
-databend-common-meta-types = { workspace = true }
 databend-common-users = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }

src/query/management/tests/it/setting.rs

@@ -19,7 +19,6 @@ use databend_common_exception::Result;
 use databend_common_meta_app::principal::UserSetting;
 use databend_common_meta_app::principal::UserSettingValue;
 use databend_common_meta_app::tenant::Tenant;
-use databend_common_meta_types::MatchSeq;
 use databend_common_users::UserApiProvider;
 use log::warn;
 
@@ -44,7 +43,7 @@ impl Settings {
 
         UserApiProvider::instance()
             .setting_api(&self.tenant)
-            .try_drop_setting(key, MatchSeq::GE(1))
+            .try_drop_setting(key)
             .await
     }
 

src/query/settings/Cargo.toml

@@ -30,7 +30,6 @@ use databend_common_management::QuotaApi;
 use databend_common_management::QuotaMgr;
 use databend_common_management::RoleApi;
 use databend_common_management::RoleMgr;
-use databend_common_management::SettingApi;
 use databend_common_management::SettingMgr;
 use databend_common_management::StageApi;
 use databend_common_management::StageMgr;
@@ -152,8 +151,8 @@ impl UserApiProvider {
         Arc::new(QuotaMgr::<WRITE_PB>::create(self.client.clone(), tenant))
     }
 
-    pub fn setting_api(&self, tenant: &Tenant) -> Arc<dyn SettingApi> {
-        Arc::new(SettingMgr::create(self.client.clone(), tenant))
+    pub fn setting_api(&self, tenant: &Tenant) -> SettingMgr {
+        SettingMgr::create(self.client.clone(), tenant)
     }
     pub fn procedure_api(&self, _tenant: &Tenant) -> ProcedureMgr {
         ProcedureMgr::create(self.client.clone())