feat: vacuum aggregating indexes (databendlabs#12179)

* vacuum aggregating indexes * add dry_run impl
bluntWu · Jul 27, 2023 · 8bd1221 · 8bd1221
1 parent 484b38e
commit 8bd1221
Show file tree

Hide file tree

Showing 11 changed files with 158 additions and 1 deletion.
diff --git a/src/meta/api/src/schema_api.rs b/src/meta/api/src/schema_api.rs
@@ -54,6 +54,7 @@ use common_meta_app::schema::ListCatalogReq;
 use common_meta_app::schema::ListDatabaseReq;
 use common_meta_app::schema::ListDroppedTableReq;
 use common_meta_app::schema::ListDroppedTableResp;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListTableLockRevReq;
 use common_meta_app::schema::ListTableReq;
@@ -134,6 +135,11 @@ pub trait SchemaApi: Send + Sync {
         req: ListIndexesReq,
     ) -> Result<Vec<(u64, String, IndexMeta)>, KVAppError>;
 
+    async fn list_indexes_by_table_id(
+        &self,
+        req: ListIndexesByIdReq,
+    ) -> Result<Vec<u64>, KVAppError>;
+
     // virtual column
 
     async fn create_virtual_column(

diff --git a/src/meta/api/src/schema_api_impl.rs b/src/meta/api/src/schema_api_impl.rs
@@ -107,6 +107,7 @@ use common_meta_app::schema::ListCatalogReq;
 use common_meta_app::schema::ListDatabaseReq;
 use common_meta_app::schema::ListDroppedTableReq;
 use common_meta_app::schema::ListDroppedTableResp;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListTableLockRevReq;
 use common_meta_app::schema::ListTableReq;
@@ -1171,6 +1172,46 @@ impl<KV: kvapi::KVApi<Error = MetaError>> SchemaApi for KV {
         Ok(index_metas)
     }
 
+    #[tracing::instrument(level = "debug", ret, err, skip_all)]
+    async fn list_indexes_by_table_id(
+        &self,
+        req: ListIndexesByIdReq,
+    ) -> Result<Vec<u64>, KVAppError> {
+        debug!(req = debug(&req), "SchemaApi: {}", func_name!());
+
+        // Get index id list by `prefix_list` "<prefix>/<tenant>"
+        let prefix_key = kvapi::KeyBuilder::new_prefixed(IndexNameIdent::PREFIX)
+            .push_str(&req.tenant)
+            .done();
+
+        let id_list = self.prefix_list_kv(&prefix_key).await?;
+        let mut id_name_list = Vec::with_capacity(id_list.len());
+        for (key, seq) in id_list.iter() {
+            let name_ident = IndexNameIdent::from_str_key(key).map_err(|e| {
+                KVAppError::MetaError(MetaError::from(InvalidReply::new("list_indexes", &e)))
+            })?;
+            let index_id = deserialize_u64(&seq.data)?;
+            id_name_list.push((index_id.0, name_ident.index_name));
+        }
+
+        debug!(ident = display(&prefix_key), "list_indexes");
+
+        if id_name_list.is_empty() {
+            return Ok(vec![]);
+        }
+
+        let index_ids = {
+            let index_metas = get_index_metas_by_ids(self, id_name_list).await?;
+            index_metas
+                .into_iter()
+                .filter(|(_, _, meta)| req.table_id == meta.table_id)
+                .map(|(id, _, _)| id)
+                .collect::<Vec<_>>()
+        };
+
+        Ok(index_ids)
+    }
+
     // virtual column
 
     async fn create_virtual_column(

diff --git a/src/meta/app/src/schema/index.rs b/src/meta/app/src/schema/index.rs
@@ -203,6 +203,21 @@ impl ListIndexesReq {
     }
 }
 
+#[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
+pub struct ListIndexesByIdReq {
+    pub tenant: String,
+    pub table_id: MetaId,
+}
+
+impl ListIndexesByIdReq {
+    pub fn new(tenant: impl Into<String>, table_id: MetaId) -> Self {
+        Self {
+            tenant: tenant.into(),
+            table_id,
+        }
+    }
+}
+
 mod kvapi_key_impl {
     use common_meta_kvapi::kvapi;
 

diff --git a/src/query/catalog/src/catalog/interface.rs b/src/query/catalog/src/catalog/interface.rs
@@ -45,6 +45,7 @@ use common_meta_app::schema::GetTableCopiedFileReply;
 use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::IndexMeta;
 use common_meta_app::schema::ListDroppedTableReq;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListVirtualColumnsReq;
 use common_meta_app::schema::RenameDatabaseReply;
@@ -111,6 +112,8 @@ pub trait Catalog: DynClone + Send + Sync {
 
     async fn list_indexes(&self, req: ListIndexesReq) -> Result<Vec<(u64, String, IndexMeta)>>;
 
+    async fn list_indexes_by_table_id(&self, req: ListIndexesByIdReq) -> Result<Vec<u64>>;
+
     async fn create_virtual_column(
         &self,
         req: CreateVirtualColumnReq,

diff --git a/src/query/service/src/catalogs/default/database_catalog.rs b/src/query/service/src/catalogs/default/database_catalog.rs
@@ -50,6 +50,7 @@ use common_meta_app::schema::GetTableCopiedFileReply;
 use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::IndexMeta;
 use common_meta_app::schema::ListDroppedTableReq;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListVirtualColumnsReq;
 use common_meta_app::schema::RenameDatabaseReply;
@@ -512,6 +513,11 @@ impl Catalog for DatabaseCatalog {
         self.mutable_catalog.list_indexes(req).await
     }
 
+    #[async_backtrace::framed]
+    async fn list_indexes_by_table_id(&self, req: ListIndexesByIdReq) -> Result<Vec<u64>> {
+        self.mutable_catalog.list_indexes_by_table_id(req).await
+    }
+
     // Virtual column
 
     #[async_backtrace::framed]

diff --git a/src/query/service/src/catalogs/default/immutable_catalog.rs b/src/query/service/src/catalogs/default/immutable_catalog.rs
@@ -42,6 +42,7 @@ use common_meta_app::schema::GetIndexReq;
 use common_meta_app::schema::GetTableCopiedFileReply;
 use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::IndexMeta;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListVirtualColumnsReq;
 use common_meta_app::schema::RenameDatabaseReply;
@@ -345,6 +346,11 @@ impl Catalog for ImmutableCatalog {
         unimplemented!()
     }
 
+    #[async_backtrace::framed]
+    async fn list_indexes_by_table_id(&self, _req: ListIndexesByIdReq) -> Result<Vec<u64>> {
+        unimplemented!()
+    }
+
     // Virtual column
 
     #[async_backtrace::framed]

diff --git a/src/query/service/src/catalogs/default/mutable_catalog.rs b/src/query/service/src/catalogs/default/mutable_catalog.rs
@@ -57,6 +57,7 @@ use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::IndexMeta;
 use common_meta_app::schema::ListDatabaseReq;
 use common_meta_app::schema::ListDroppedTableReq;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListTableLockRevReq;
 use common_meta_app::schema::ListVirtualColumnsReq;
@@ -258,6 +259,11 @@ impl Catalog for MutableCatalog {
         Ok(self.ctx.meta.list_indexes(req).await?)
     }
 
+    #[async_backtrace::framed]
+    async fn list_indexes_by_table_id(&self, req: ListIndexesByIdReq) -> Result<Vec<u64>> {
+        Ok(self.ctx.meta.list_indexes_by_table_id(req).await?)
+    }
+
     // Virtual column
 
     #[async_backtrace::framed]

diff --git a/src/query/service/tests/it/storages/fuse/operations/commit.rs b/src/query/service/tests/it/storages/fuse/operations/commit.rs
@@ -63,6 +63,7 @@ use common_meta_app::schema::GetIndexReq;
 use common_meta_app::schema::GetTableCopiedFileReply;
 use common_meta_app::schema::GetTableCopiedFileReq;
 use common_meta_app::schema::IndexMeta;
+use common_meta_app::schema::ListIndexesByIdReq;
 use common_meta_app::schema::ListIndexesReq;
 use common_meta_app::schema::ListVirtualColumnsReq;
 use common_meta_app::schema::RenameDatabaseReply;
@@ -706,6 +707,11 @@ impl Catalog for FakedCatalog {
         unimplemented!()
     }
 
+    #[async_backtrace::framed]
+    async fn list_indexes_by_table_id(&self, _req: ListIndexesByIdReq) -> Result<Vec<u64>> {
+        unimplemented!()
+    }
+
     #[async_backtrace::framed]
     async fn create_virtual_column(
         &self,