Merge branch 'main' into feat-gist-index

Author: b41sh

src/common/base/src/runtime/perf/query_perf.rs

@@ -55,7 +55,7 @@ impl QueryPerf {
             .blocklist(&["libc", "libgcc", "pthread", "vdso"])
             .set_filter_func(filter_closure)
             .build()
-            .map_err(|_e| ErrorCode::Internal("Failed to create profiler"))?;
+            .map_err(|e| ErrorCode::Internal(format!("Failed to create profiler, {e}")))?;
         debug!("starting perf with frequency: {}", frequency);
         let mut payload = ThreadTracker::new_tracking_payload();
         payload.perf_enabled = true;

src/meta/api/src/kv_fetch_util.rs

@@ -26,7 +26,9 @@ use databend_common_meta_types::MetaNetworkError;
 use databend_common_meta_types::SeqV;
 use databend_common_meta_types::TxnGetResponse;
 use databend_common_meta_types::UpsertKV;
+use databend_common_meta_types::errors;
 use databend_common_proto_conv::FromToProto;
+use futures::TryStreamExt;
 
 use crate::deserialize_struct;
 use crate::deserialize_u64;
@@ -155,6 +157,42 @@ pub async fn list_u64_value<K: kvapi::Key>(
     Ok((structured_keys, values))
 }
 
+/// Batch get u64 values by keys.
+///
+/// Returns a vec of Option<u64> in the same order as input keys.
+/// None means the key does not exist.
+pub async fn mget_u64_values<K: kvapi::Key>(
+    kv_api: &(impl kvapi::KVApi<Error = MetaError> + ?Sized),
+    keys: &[K],
+) -> Result<Vec<Option<u64>>, MetaError> {
+    if keys.is_empty() {
+        return Ok(vec![]);
+    }
+
+    let str_keys: Vec<String> = keys.iter().map(|k| k.to_string_key()).collect();
+    let mut strm = kv_api.get_kv_stream(&str_keys).await?;
+
+    let mut results = Vec::with_capacity(keys.len());
+    while let Some(item) = strm.try_next().await? {
+        if let Some(seq_v) = item.value {
+            let id = *deserialize_u64(&seq_v.data)?;
+            results.push(Some(id));
+        } else {
+            results.push(None);
+        }
+    }
+
+    if results.len() != keys.len() {
+        return Err(
+            errors::IncompleteStream::new(keys.len() as u64, results.len() as u64)
+                .context(" while mget_u64_values")
+                .into(),
+        );
+    }
+
+    Ok(results)
+}
+
 /// Generate an id on metasrv.
 ///
 /// Ids are categorized by generators.

src/meta/api/src/table_api.rs

@@ -128,6 +128,7 @@ use crate::kv_app_error::KVAppError;
 use crate::kv_fetch_util::deserialize_id_get_response;
 use crate::kv_fetch_util::deserialize_struct_get_response;
 use crate::kv_fetch_util::mget_pb_values;
+use crate::kv_fetch_util::mget_u64_values;
 use crate::kv_pb_api::KVPbApi;
 use crate::kv_pb_crud_api::KVPbCrudApi;
 use crate::list_u64_value;
@@ -1530,6 +1531,71 @@ where
         return Ok(Arc::new(tb_info));
     }
 
+    /// Get multiple tables by db_id and table names in batch.
+    /// Returns TableInfo for tables that exist, in the same order as input.
+    #[logcall::logcall]
+    #[fastrace::trace]
+    async fn mget_tables(
+        &self,
+        db_id: u64,
+        db_name: &str,
+        table_names: &[String],
+    ) -> Result<Vec<Arc<TableInfo>>, KVAppError> {
+        debug!(db_id = db_id, table_names :? = table_names; "TableApi: {}", func_name!());
+
+        // Build DBIdTableName keys for all table names
+        let dbid_tbnames: Vec<DBIdTableName> = table_names
+            .iter()
+            .map(|name| DBIdTableName::new(db_id, name))
+            .collect();
+
+        // Batch get table ids
+        let table_ids = mget_u64_values(self, &dbid_tbnames).await?;
+
+        // Collect valid table ids with their names
+        let mut valid_tables: Vec<(String, u64)> = Vec::with_capacity(table_names.len());
+        for (dbid_tbname, table_id_opt) in dbid_tbnames.into_iter().zip(table_ids.into_iter()) {
+            if let Some(table_id) = table_id_opt {
+                valid_tables.push((dbid_tbname.table_name, table_id));
+            }
+        }
+
+        if valid_tables.is_empty() {
+            return Ok(vec![]);
+        }
+
+        // Batch get table metas
+        let table_id_keys: Vec<TableId> = valid_tables
+            .iter()
+            .map(|(_, id)| TableId { table_id: *id })
+            .collect();
+        let seq_metas = self.get_pb_values_vec(table_id_keys).await?;
+
+        // Build TableInfo for valid tables
+        let db_type = DatabaseType::NormalDB;
+        let mut results = Vec::with_capacity(valid_tables.len());
+        for ((table_name, table_id), seq_meta_opt) in
+            valid_tables.into_iter().zip(seq_metas.into_iter())
+        {
+            if let Some(seq_meta) = seq_meta_opt {
+                let tb_info = TableInfo {
+                    ident: TableIdent {
+                        table_id,
+                        seq: seq_meta.seq,
+                    },
+                    desc: format!("'{}'.'{}'", db_name, table_name),
+                    name: table_name,
+                    meta: seq_meta.data,
+                    db_type: db_type.clone(),
+                    catalog_info: Default::default(),
+                };
+                results.push(Arc::new(tb_info));
+            }
+        }
+
+        Ok(results)
+    }
+
     #[logcall::logcall]
     #[fastrace::trace]
     async fn get_table_by_id(

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

@@ -296,6 +296,21 @@ pub trait Catalog: DynClone + Send + Sync + Debug {
         table_name: &str,
     ) -> Result<Arc<dyn Table>>;
 
+    /// Get multiple tables by db and table names.
+    /// Returns tables in the same order as the input table_names.
+    /// If a table is not found, it will not be included in the result.
+    async fn mget_tables(
+        &self,
+        _tenant: &Tenant,
+        _db_name: &str,
+        _table_names: &[String],
+    ) -> Result<Vec<Arc<dyn Table>>> {
+        Err(ErrorCode::Unimplemented(format!(
+            "'mget_tables' not implemented for catalog {}",
+            self.name()
+        )))
+    }
+
     // Get one table identified as dropped by db and table name.
     async fn get_table_history(
         &self,

src/query/catalog/src/database.rs

@@ -120,11 +120,21 @@ pub trait Database: DynClone + Sync + Send {
         )))
     }
 
+    /// Get multiple tables by names.
+    /// Returns tables in the same order as input, skipping tables that are not found.
+    #[async_backtrace::framed]
+    async fn mget_tables(&self, _table_names: &[String]) -> Result<Vec<Arc<dyn Table>>> {
+        Err(ErrorCode::Unimplemented(format!(
+            "UnImplement mget_tables in {} Database",
+            self.name()
+        )))
+    }
+
     // Get one table history by db and table name.
     #[async_backtrace::framed]
     async fn get_table_history(&self, _table_name: &str) -> Result<Vec<Arc<dyn Table>>> {
         Err(ErrorCode::Unimplemented(format!(
-            "UnImplement get_table in {} Database",
+            "UnImplement get_table_history in {} Database",
             self.name()
         )))
     }