feat(query-cache): add config to toggle meta-service ownership cache on query (#18197)

Author: drmingdrmer

src/bendsave/src/storage.rs

@@ -25,6 +25,7 @@ use bytes::BytesMut;
 use databend_common_base::base::GlobalInstance;
 use databend_common_base::runtime::GlobalIORuntime;
 use databend_common_catalog::session_type::SessionType;
+use databend_common_config::CacheConfig;
 use databend_common_config::Config;
 use databend_common_config::GlobalConfig;
 use databend_common_config::InnerConfig;
@@ -106,6 +107,7 @@ pub async fn verify_query_license(cfg: &InnerConfig) -> Result<()> {
     SessionManager::init(cfg)?;
     UserApiProvider::init(
         cfg.meta.to_meta_grpc_client_conf(),
+        &CacheConfig::default(),
         BuiltIn::default(),
         &cfg.query.tenant_id,
         cfg.query.tenant_quota.clone(),

src/meta/client/src/established_client.rs

@@ -155,7 +155,7 @@ impl fmt::Display for EstablishedClient {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
-            "EstablishedClient[uniq={}, {}]{{ target: {}, least_required_server: {}, endpoints: {} }}",
+            "EstablishedClient[uniq={}, {}]{{ target: {}, server_version: {}, endpoints: {} }}",
             self.uniq,
             self.created_at,
             self.target_endpoint,

src/query/config/src/config.rs

@@ -3143,6 +3143,11 @@ impl TryInto<InnerLocalConfig> for LocalConfig {
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, Args)]
 #[serde(default)]
 pub struct CacheConfig {
+    /// The data in meta-service using key `TenantOwnershipObjectIdent`
+    #[clap(long = "cache-enable-meta-service-ownership", default_value = "false")]
+    #[serde(default = "bool_false")]
+    pub meta_service_ownership_cache: bool,
+
     /// Enable table meta cache. Default is enabled. Set it to false to disable all the table meta caches
     #[clap(long = "cache-enable-table-meta-cache", default_value = "true")]
     #[serde(default = "bool_true")]
@@ -3380,6 +3385,11 @@ fn bool_true() -> bool {
     true
 }
 
+#[inline]
+fn bool_false() -> bool {
+    false
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, ValueEnum)]
 #[serde(rename_all = "lowercase")]
 pub enum CacheStorageTypeConfig {
@@ -3573,6 +3583,7 @@ mod cache_config_converters {
 
         fn try_from(value: CacheConfig) -> std::result::Result<Self, Self::Error> {
             Ok(Self {
+                meta_service_ownership_cache: value.meta_service_ownership_cache,
                 enable_table_meta_cache: value.enable_table_meta_cache,
                 table_meta_snapshot_count: value.table_meta_snapshot_count,
                 table_meta_segment_bytes: value.table_meta_segment_bytes,
@@ -3607,6 +3618,7 @@ mod cache_config_converters {
     impl From<inner::CacheConfig> for CacheConfig {
         fn from(value: inner::CacheConfig) -> Self {
             Self {
+                meta_service_ownership_cache: value.meta_service_ownership_cache,
                 enable_table_meta_cache: value.enable_table_meta_cache,
                 table_meta_snapshot_count: value.table_meta_snapshot_count,
                 table_meta_segment_bytes: value.table_meta_segment_bytes,

src/query/config/src/inner.rs

@@ -556,6 +556,9 @@ impl Default for LocalConfig {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct CacheConfig {
+    /// The data in meta-service using key `TenantOwnershipObjectIdent`
+    pub meta_service_ownership_cache: bool,
+
     /// Enable table meta cache. Default is enabled. Set it to false to disable all the table meta caches
     pub enable_table_meta_cache: bool,
 
@@ -727,6 +730,7 @@ impl Default for DiskCacheConfig {
 impl Default for CacheConfig {
     fn default() -> Self {
         Self {
+            meta_service_ownership_cache: false,
             enable_table_meta_cache: true,
             table_meta_snapshot_count: 256,
             table_meta_segment_bytes: 1073741824,

src/query/management/src/role/role_mgr.rs

@@ -66,7 +66,7 @@ static BUILTIN_ROLE_ACCOUNT_ADMIN: &str = "account_admin";
 pub struct RoleMgr {
     kv_api: Arc<dyn kvapi::KVApi<Error = MetaError> + Send + Sync>,
     tenant: Tenant,
-    ownership_cache: Arc<Mutex<Cache>>,
+    ownership_cache: Option<Arc<Mutex<Cache>>>,
     upgrade_to_pb: bool,
 }
 
@@ -75,7 +75,7 @@ impl RoleMgr {
         kv_api: Arc<dyn kvapi::KVApi<Error = MetaError> + Send + Sync>,
         tenant: &Tenant,
         upgrade_to_pb: bool,
-        ownership_cache: Arc<Mutex<Cache>>,
+        ownership_cache: Option<Arc<Mutex<Cache>>>,
     ) -> Self {
         RoleMgr {
             kv_api,
@@ -246,24 +246,28 @@ impl RoleApi for RoleMgr {
         let object_owner_prefix = self.ownership_object_prefix();
 
         let cached = {
-            let mut cache = self.ownership_cache.lock().await;
-            let kvs = cache.try_list_dir(object_owner_prefix.as_str()).await;
+            if let Some(ownership_cache) = &self.ownership_cache {
+                let mut cache = ownership_cache.lock().await;
+                let kvs = cache.try_list_dir(object_owner_prefix.as_str()).await;
 
-            match &kvs {
-                Ok(kvs) => {
-                    info!(
+                match &kvs {
+                    Ok(kvs) => {
+                        info!(
                         "RoleMgr::list_ownerships() returned from cache: {} keys; first key: {:?}; last key: {:?}",
                         kvs.len(),
                         kvs.first().map(|(k, _)| k),
                         kvs.last().map(|(k, _)| k),
                     );
+                    }
+                    Err(err) => {
+                        warn!("list ownerships from cache failed. err: {}; It is not a functional issue but may be a performance issue with more than 100_000 ownership records", err);
+                    }
                 }
-                Err(err) => {
-                    warn!("list ownerships from cache failed. err: {}; It is not a functional issue but may be a performance issue with more than 100_000 ownership records", err);
-                }
-            }
 
-            kvs.ok()
+                kvs.ok()
+            } else {
+                None
+            }
         };
 
         let values = if let Some(kvs) = cached {

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

@@ -103,7 +103,7 @@ async fn new_role_api(
         test_api.clone(),
         &tenant,
         enable_meta_data_upgrade_json_to_pb_from_v307,
-        Arc::new(Mutex::new(cache)),
+        Some(Arc::new(Mutex::new(cache))),
     );
     Ok((test_api, mgr))
 }

src/query/service/src/global_services.rs

@@ -135,6 +135,7 @@ impl GlobalServices {
             };
             UserApiProvider::init(
                 config.meta.to_meta_grpc_client_conf(),
+                &config.cache,
                 builtin,
                 &config.query.tenant_id,
                 config.query.tenant_quota.clone(),

src/query/service/src/test_kits/config.rs

@@ -155,6 +155,11 @@ ADDRESS = 'https://databend.com';"
         self
     }
 
+    pub fn enable_meta_service_ownership_cache(mut self) -> ConfigBuilder {
+        self.conf.cache.meta_service_ownership_cache = true;
+        self
+    }
+
     pub fn enable_table_meta_cache(mut self) -> ConfigBuilder {
         self.conf.cache.enable_table_meta_cache = true;
         self

src/query/service/tests/it/storages/testdata/configs_table_basic.txt

@@ -19,6 +19,7 @@ DB.Table: 'system'.'configs', Table: configs-table_id:1, ver:0, Engine: SystemCo
 | 'cache'   | 'inverted_index_filter_memory_ratio'            | '0'                                                                                                                                                                                                       | ''       |
 | 'cache'   | 'inverted_index_filter_size'                    | '2147483648'                                                                                                                                                                                              | ''       |
 | 'cache'   | 'inverted_index_meta_count'                     | '3000'                                                                                                                                                                                                    | ''       |
+| 'cache'   | 'meta_service_ownership_cache'                  | 'false'                                                                                                                                                                                                   | ''       |
 | 'cache'   | 'segment_block_metas_count'                     | '0'                                                                                                                                                                                                       | ''       |
 | 'cache'   | 'table_bloom_index_filter_count'                | '0'                                                                                                                                                                                                       | ''       |
 | 'cache'   | 'table_bloom_index_filter_size'                 | '2147483648'                                                                                                                                                                                              | ''       |

src/query/users/src/user_api.rs

@@ -18,6 +18,7 @@ use std::sync::Arc;
 
 use databend_common_base::base::tokio::sync::Mutex;
 use databend_common_base::base::GlobalInstance;
+use databend_common_config::CacheConfig;
 use databend_common_config::GlobalConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
@@ -65,7 +66,7 @@ pub struct UserApiProvider {
     /// When UserApiProvider is created, the cache will be created and initialized.
     /// The `Cache` instance internally stores the status about if databend-meta service supports cache.
     /// If not, every access returns [`Unsupported`] error.
-    ownership_cache: Arc<Mutex<Cache>>,
+    ownership_cache: Option<Arc<Mutex<Cache>>>,
 
     builtin: BuiltIn,
 }
@@ -74,11 +75,12 @@ impl UserApiProvider {
     #[async_backtrace::framed]
     pub async fn init(
         conf: RpcClientConf,
+        cache_config: &CacheConfig,
         builtin: BuiltIn,
         tenant: &Tenant,
         quota: Option<TenantQuota>,
     ) -> Result<()> {
-        GlobalInstance::set(Self::try_create(conf, builtin, tenant).await?);
+        GlobalInstance::set(Self::try_create(conf, cache_config, builtin, tenant).await?);
         let user_mgr = UserApiProvider::instance();
 
         if let Some(q) = quota {
@@ -92,6 +94,7 @@ impl UserApiProvider {
     #[async_backtrace::framed]
     pub async fn try_create(
         conf: RpcClientConf,
+        cache_config: &CacheConfig,
         builtin: BuiltIn,
         tenant: &Tenant,
     ) -> Result<Arc<UserApiProvider>> {
@@ -104,8 +107,13 @@ impl UserApiProvider {
 
         let client = meta_store.deref().clone();
 
-        let cache = RoleMgr::new_cache(client.clone()).await;
-        let cache = Arc::new(Mutex::new(cache));
+        let cache = if cache_config.meta_service_ownership_cache {
+            let cache = RoleMgr::new_cache(client.clone()).await;
+            let cache = Arc::new(Mutex::new(cache));
+            Some(cache)
+        } else {
+            None
+        };
 
         let user_mgr = UserApiProvider {
             meta: meta_store.clone(),
@@ -137,7 +145,7 @@ impl UserApiProvider {
         conf: RpcClientConf,
         tenant: &Tenant,
     ) -> Result<Arc<UserApiProvider>> {
-        Self::try_create(conf, BuiltIn::default(), tenant).await
+        Self::try_create(conf, &CacheConfig::default(), BuiltIn::default(), tenant).await
     }
 
     pub fn instance() -> Arc<UserApiProvider> {