chore(query): add iceberg table metadata cache (#17780)

* feat(query): add iceberg table metadata cache

* feat(query): add iceberg table metadata cache

* feat(query): add iceberg table metadata cache

* feat(query): add iceberg table metadata cache

Author: sundy-li

Cargo.lock

@@ -129,7 +129,7 @@ pub struct IcebergGlueCatalogOption {
 /// Same as `CatalogNameIdent`, but with `serde` support,
 /// and can be used a s part of a value.
 // #[derive(Clone, Debug, PartialEq, Eq)]
-#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Hash, Eq)]
 pub struct CatalogName {
     pub tenant: String,
     pub catalog_name: String,

src/meta/app/src/schema/catalog.rs

@@ -16,7 +16,6 @@ use std::any::Any;
 use std::fmt::Debug;
 use std::sync::Arc;
 
-use databend_common_config::InnerConfig;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_meta_app::schema::database_name_ident::DatabaseNameIdent;
@@ -106,7 +105,6 @@ use databend_common_meta_app::schema::UpsertTableOptionReply;
 use databend_common_meta_app::schema::UpsertTableOptionReq;
 use databend_common_meta_app::schema::VirtualColumnMeta;
 use databend_common_meta_app::tenant::Tenant;
-use databend_common_meta_store::MetaStore;
 use databend_common_meta_types::MetaId;
 use databend_common_meta_types::SeqV;
 use databend_storages_common_session::SessionState;
@@ -127,12 +125,7 @@ pub struct StorageDescription {
 }
 
 pub trait CatalogCreator: Send + Sync + Debug {
-    fn try_create(
-        &self,
-        info: Arc<CatalogInfo>,
-        conf: InnerConfig,
-        meta: &MetaStore,
-    ) -> Result<Arc<dyn Catalog>>;
+    fn try_create(&self, info: Arc<CatalogInfo>) -> Result<Arc<dyn Catalog>>;
 }
 
 #[async_trait::async_trait]

src/query/catalog/src/catalog/interface.rs

@@ -54,8 +54,6 @@ pub struct CatalogManager {
 
     /// catalog_creators is the catalog creators that registered.
     pub catalog_creators: HashMap<CatalogType, Arc<dyn CatalogCreator>>,
-
-    conf: InnerConfig,
 }
 
 impl CatalogManager {
@@ -114,7 +112,7 @@ impl CatalogManager {
                     created_on: Utc::now(),
                 },
             };
-            let ctl = creator.try_create(Arc::new(ctl_info), conf.to_owned(), &meta)?;
+            let ctl = creator.try_create(Arc::new(ctl_info))?;
             external_catalogs.insert(name.clone(), ctl);
         }
 
@@ -123,7 +121,6 @@ impl CatalogManager {
             default_catalog,
             external_catalogs,
             catalog_creators,
-            conf: conf.to_owned(),
         };
 
         Ok(Arc::new(catalog_manager))
@@ -153,8 +150,7 @@ impl CatalogManager {
             .catalog_creators
             .get(&typ)
             .ok_or_else(|| ErrorCode::BadArguments(format!("unknown catalog type: {:?}", typ)))?;
-
-        creator.try_create(info, self.conf.clone(), &self.meta)
+        creator.try_create(info)
     }
 
     /// Get a catalog from manager.
@@ -184,7 +180,6 @@ impl CatalogManager {
 
         // Get catalog from metasrv.
         let info = self.meta.get_catalog(&ident).await?;
-
         self.build_catalog(info, session_state)
     }
 

src/query/catalog/src/catalog/manager.rs

@@ -108,6 +108,11 @@ pub trait Database: DynClone + Sync + Send {
         Ok(vec![])
     }
 
+    #[async_backtrace::framed]
+    async fn trigger_use(&self) -> Result<()> {
+        Ok(())
+    }
+
     #[async_backtrace::framed]
     async fn list_tables_names(&self) -> Result<Vec<String>> {
         Ok(vec![])

src/query/catalog/src/database.rs

@@ -3070,6 +3070,13 @@ pub struct CacheConfig {
     )]
     pub table_data_deserialized_memory_ratio: u64,
 
+    #[clap(
+        long = "cache-iceberg-table-meta-count",
+        value_name = "VALUE",
+        default_value = "1024"
+    )]
+    pub iceberg_table_meta_count: u64,
+
     // ----- the following options/args are all deprecated               ----
     /// Max number of cached table segment
     #[clap(long = "cache-table-meta-segment-count", value_name = "VALUE")]
@@ -3297,6 +3304,7 @@ mod cache_config_converters {
                 data_cache_key_reload_policy: value.data_cache_key_reload_policy.try_into()?,
                 table_data_deserialized_data_bytes: value.table_data_deserialized_data_bytes,
                 table_data_deserialized_memory_ratio: value.table_data_deserialized_memory_ratio,
+                iceberg_table_meta_count: value.iceberg_table_meta_count,
                 disk_cache_table_bloom_index_meta_size: value
                     .disk_cache_table_bloom_index_meta_size,
             })
@@ -3330,6 +3338,7 @@ mod cache_config_converters {
                 disk_cache_config: value.disk_cache_config.into(),
                 table_data_deserialized_data_bytes: value.table_data_deserialized_data_bytes,
                 table_data_deserialized_memory_ratio: value.table_data_deserialized_memory_ratio,
+                iceberg_table_meta_count: value.iceberg_table_meta_count,
                 table_meta_segment_count: None,
                 segment_block_metas_count: value.segment_block_metas_count,
             }

src/query/config/src/config.rs

@@ -636,6 +636,9 @@ pub struct CacheConfig {
     /// Only if query nodes have plenty of un-utilized memory, the working set can be fitted into,
     /// and the access pattern will benefit from caching, consider enabled this cache.
     pub table_data_deserialized_memory_ratio: u64,
+
+    /// Max number of cached table count of iceberg tables
+    pub iceberg_table_meta_count: u64,
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -733,6 +736,7 @@ impl Default for CacheConfig {
             data_cache_key_reload_policy: Default::default(),
             table_data_deserialized_data_bytes: 0,
             table_data_deserialized_memory_ratio: 0,
+            iceberg_table_meta_count: 1024,
         }
     }
 }

src/query/config/src/inner.rs

@@ -50,9 +50,12 @@ impl Interpreter for UseDatabaseInterpreter {
         if self.plan.database.trim().is_empty() {
             return Err(ErrorCode::UnknownDatabase("No database selected"));
         }
-        self.ctx
+        let db = self
+            .ctx
             .set_current_database(self.plan.database.clone())
             .await?;
+        db.trigger_use().await?;
+
         self.ctx.set_affect(QueryAffect::UseDB {
             name: self.plan.database.clone(),
         });

src/query/service/src/interpreters/interpreter_use_database.rs

@@ -41,6 +41,7 @@ use databend_common_base::runtime::MemStat;
 use databend_common_base::runtime::TrySpawn;
 use databend_common_base::JoinHandle;
 use databend_common_catalog::catalog::CATALOG_DEFAULT;
+use databend_common_catalog::database::Database;
 use databend_common_catalog::lock::LockTableOption;
 use databend_common_catalog::merge_into_join::MergeIntoJoin;
 use databend_common_catalog::plan::DataSourceInfo;
@@ -257,22 +258,26 @@ impl QueryContext {
     }
 
     #[async_backtrace::framed]
-    pub async fn set_current_database(&self, new_database_name: String) -> Result<()> {
+    pub async fn set_current_database(
+        &self,
+        new_database_name: String,
+    ) -> Result<Arc<dyn Database>> {
         let tenant_id = self.get_tenant();
         let catalog = self
             .get_catalog(self.get_current_catalog().as_str())
             .await?;
         match catalog.get_database(&tenant_id, &new_database_name).await {
-            Ok(_) => self.shared.set_current_database(new_database_name),
+            Ok(db) => {
+                self.shared.set_current_database(new_database_name);
+                Ok(db)
+            }
             Err(_) => {
                 return Err(ErrorCode::UnknownDatabase(format!(
                     "Cannot use database '{}': It does not exist.",
                     new_database_name
                 )));
             }
-        };
-
-        Ok(())
+        }
     }
 
     pub fn attach_table(&self, catalog: &str, database: &str, name: &str, table: Arc<dyn Table>) {

src/query/service/src/sessions/query_ctx.rs

@@ -8,6 +8,7 @@ DB.Table: 'system'.'caches', Table: caches-table_id:1, ver:0, Engine: SystemCach
 | 'test-node' | 'memory_cache_bloom_index_filter'            | 0        | 0        | 2147483648 | 'bytes'  | 0        | 0        | 0        |
 | 'test-node' | 'memory_cache_column_oriented_segment_info'  | 0        | 0        | 1073741824 | 'bytes'  | 0        | 0        | 0        |
 | 'test-node' | 'memory_cache_compact_segment_info'          | 0        | 0        | 1073741824 | 'bytes'  | 0        | 0        | 0        |
+| 'test-node' | 'memory_cache_iceberg_table'                 | 0        | 0        | 1024       | 'count'  | 0        | 0        | 0        |
 | 'test-node' | 'memory_cache_inverted_index_file'           | 0        | 0        | 2147483648 | 'bytes'  | 0        | 0        | 0        |
 | 'test-node' | 'memory_cache_inverted_index_file_meta_data' | 0        | 0        | 3000       | 'count'  | 0        | 0        | 0        |
 | 'test-node' | 'memory_cache_parquet_meta_data'             | 0        | 0        | 3000       | 'count'  | 0        | 0        | 0        |