refactor: shrink bloom index meta cache size

Author: dantengsky

Cargo.lock

@@ -46,6 +46,7 @@ common-arrow = { path = "../../common/arrow" }
 common-ast = { path = "../ast" }
 common-base = { path = "../../common/base" }
 common-catalog = { path = "../catalog" }
+common-cache = {path = "../../common/cache"}
 common-compress = { path = "../../common/compress" }
 common-config = { path = "../config" }
 common-exception = { path = "../../common/exception" }
@@ -166,6 +167,7 @@ toml = { version = "0.7.3", default-features = false }
 url = "2.3.1"
 walkdir = "2.3.2"
 wiremock = "0.5.14"
+sysinfo = "0.28.3"
 
 [build-dependencies]
 common-building = { path = "../../common/building" }

src/query/service/Cargo.toml

@@ -485,6 +485,8 @@ impl Interpreter for InsertInterpreter {
 
                 let ctx = self.ctx.clone();
                 let overwrite = self.plan.overwrite;
+                eprintln!(">>> main piepline {:?}", build_res.main_pipeline);
+                eprintln!(">>> source piepline {:?}", build_res.sources_pipelines);
                 build_res.main_pipeline.set_on_finished(move |may_error| {
                     // capture out variable
                     let overwrite = overwrite;

src/query/service/src/interpreters/interpreter_insert.rs

@@ -94,7 +94,9 @@ impl Fragmenter {
     }
 
     pub fn build_fragment(mut self, plan: &PhysicalPlan) -> Result<PlanFragment> {
+        eprintln!("origin plan {:?}", &plan);
         let root = self.replace(plan)?;
+        eprintln!("after replace {:?}", &root);
         let mut root_fragment = PlanFragment {
             plan: root,
             fragment_type: FragmentType::Root,

src/query/service/src/schedulers/fragments/fragmenter.rs

@@ -12,6 +12,7 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use common_arrow::parquet::metadata::ThriftFileMetaData;
 use common_exception::Result;
 use common_expression::DataBlock;
 use common_expression::FunctionContext;
@@ -56,13 +57,13 @@ impl<'a> BlockWriter<'a> {
         block: DataBlock,
         col_stats: StatisticsOfColumns,
         cluster_stats: Option<ClusterStatistics>,
-    ) -> Result<BlockMeta> {
+    ) -> Result<(BlockMeta, Option<ThriftFileMetaData>)> {
         let (location, block_id) = self.location_generator.gen_block_location();
 
         let data_accessor = &self.data_accessor;
         let row_count = block.num_rows() as u64;
         let block_size = block.memory_size() as u64;
-        let (bloom_filter_index_size, bloom_filter_index_location) = self
+        let (bloom_filter_index_size, bloom_filter_index_location, meta) = self
             .build_block_index(data_accessor, schema.clone(), &block, block_id)
             .await?;
 
@@ -88,7 +89,7 @@ impl<'a> BlockWriter<'a> {
             bloom_filter_index_size,
             Compression::Lz4Raw,
         );
-        Ok(block_meta)
+        Ok((block_meta, meta))
     }
 
     pub async fn build_block_index(
@@ -97,7 +98,7 @@ impl<'a> BlockWriter<'a> {
         schema: TableSchemaRef,
         block: &DataBlock,
         block_id: Uuid,
-    ) -> Result<(u64, Option<Location>)> {
+    ) -> Result<(u64, Option<Location>, Option<ThriftFileMetaData>)> {
         let location = self
             .location_generator
             .block_bloom_index_location(&block_id);
@@ -109,18 +110,16 @@ impl<'a> BlockWriter<'a> {
             let filter_schema = bloom_index.filter_schema;
             let mut data = Vec::with_capacity(DEFAULT_BLOCK_INDEX_BUFFER_SIZE);
             let index_block_schema = &filter_schema;
-            let (size, _) = blocks_to_parquet(
+            let (size, meta) = blocks_to_parquet(
                 index_block_schema,
                 vec![index_block],
                 &mut data,
                 TableCompression::None,
             )?;
-
             data_accessor.write(&location.0, data).await?;
-
-            Ok((size, Some(location)))
+            Ok((size, Some(location), Some(meta)))
         } else {
-            Ok((0u64, None))
+            Ok((0u64, None, None))
         }
     }
 }

src/query/service/tests/it/storages/fuse/block_writer.rs

@@ -309,7 +309,7 @@ mod utils {
         let blocks: std::vec::Vec<DataBlock> = stream.try_collect().await?;
         for block in blocks {
             let stats = gen_columns_statistics(&block, None, &schema)?;
-            let block_meta = block_writer
+            let (block_meta, _index_meta) = block_writer
                 .write(FuseStorageFormat::Parquet, &schema, block, stats, None)
                 .await?;
             block_metas.push(Arc::new(block_meta));

src/query/service/tests/it/storages/fuse/operations/gc.rs

@@ -698,7 +698,7 @@ impl CompactSegmentTestFixture {
                 let block = block?;
                 let col_stats = gen_columns_statistics(&block, None, &schema)?;
 
-                let block_meta = block_writer
+                let (block_meta, _index_meta) = block_writer
                     .write(FuseStorageFormat::Parquet, &schema, block, col_stats, None)
                     .await?;
 

src/query/service/tests/it/storages/fuse/operations/mutation/segments_compact_mutator.rs

@@ -15,7 +15,10 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use common_arrow::parquet::metadata::FileMetaData;
+use common_arrow::parquet::metadata::ThriftFileMetaData;
 use common_base::base::tokio;
+use common_cache::Cache;
 use common_expression::type_check::check;
 use common_expression::types::number::Int32Type;
 use common_expression::types::number::NumberScalar;
@@ -35,6 +38,7 @@ use common_expression::Scalar;
 use common_expression::TableDataType;
 use common_expression::TableField;
 use common_expression::TableSchema;
+use common_expression::TableSchemaRefExt;
 use common_functions::aggregates::eval_aggr;
 use common_functions::scalars::BUILTIN_FUNCTIONS;
 use common_sql::evaluator::BlockOperator;
@@ -50,11 +54,19 @@ use databend_query::storages::fuse::statistics::ClusterStatsGenerator;
 use databend_query::storages::fuse::statistics::StatisticsAccumulator;
 use opendal::Operator;
 use rand::Rng;
+use storages_common_cache::InMemoryCacheBuilder;
+use storages_common_cache::InMemoryItemCacheHolder;
+use storages_common_index::BloomIndexMetaMini;
 use storages_common_table_meta::meta::BlockMeta;
 use storages_common_table_meta::meta::ClusterStatistics;
 use storages_common_table_meta::meta::ColumnStatistics;
 use storages_common_table_meta::meta::Compression;
 use storages_common_table_meta::meta::Statistics;
+use sysinfo::get_current_pid;
+use sysinfo::ProcessExt;
+use sysinfo::System;
+use sysinfo::SystemExt;
+use uuid::Uuid;
 
 use crate::storages::fuse::block_writer::BlockWriter;
 use crate::storages::fuse::table_test_fixture::TestFixture;
@@ -236,7 +248,7 @@ async fn test_accumulator() -> common_exception::Result<()> {
         let block = item?;
         let col_stats = gen_columns_statistics(&block, None, &schema)?;
         let block_writer = BlockWriter::new(&operator, &loc_generator);
-        let block_meta = block_writer
+        let (block_meta, _index_meta) = block_writer
             .write(FuseStorageFormat::Parquet, &schema, block, col_stats, None)
             .await?;
         stats_acc.add_with_block_meta(block_meta);
@@ -553,3 +565,105 @@ fn test_reduce_block_meta() -> common_exception::Result<()> {
 
     Ok(())
 }
+
+fn populate_cache<T>(cache: &InMemoryItemCacheHolder<T>, item: T, num_cache: usize)
+where T: Clone {
+    let mut c = cache.write();
+    for _ in 0..num_cache {
+        let uuid = Uuid::new_v4();
+        (*c).put(
+            format!("{}", uuid.simple()),
+            std::sync::Arc::new(item.clone()),
+        );
+    }
+}
+
+async fn setup() -> common_exception::Result<ThriftFileMetaData> {
+    let fields = (0..23)
+        .into_iter()
+        .map(|_| TableField::new("id", TableDataType::Number(NumberDataType::Int32)))
+        .collect::<Vec<_>>();
+
+    let schema = TableSchemaRefExt::create(fields);
+
+    let mut columns = vec![];
+    for _ in 0..schema.fields().len() {
+        // values do not matter
+        let column = Int32Type::from_data(vec![1]);
+        columns.push(column)
+    }
+
+    let block = DataBlock::new_from_columns(columns);
+    let operator = Operator::new(opendal::services::Memory::default())?.finish();
+    let loc_generator = TableMetaLocationGenerator::with_prefix("/".to_owned());
+    let col_stats = gen_columns_statistics(&block, None, &schema)?;
+    let block_writer = BlockWriter::new(&operator, &loc_generator);
+    let (_block_meta, thrift_file_meta) = block_writer
+        .write(FuseStorageFormat::Parquet, &schema, block, col_stats, None)
+        .await?;
+
+    Ok(thrift_file_meta.unwrap())
+}
+
+fn show_memory_usage(case: &str, base_memory_usage: u64, num_cache_items: usize) {
+    let sys = System::new_all();
+    let pid = get_current_pid().unwrap();
+    let process = sys.process(pid).unwrap();
+    {
+        let memory_after = process.memory();
+        let delta = memory_after - base_memory_usage;
+        let delta_gb = (delta as f64) / 1024.0 / 1024.0 / 1024.0;
+        eprintln!(
+            " cache type: {}, number of cached items {}, mem usage(B):{:+}, mem usage(GB){:+}",
+            case, num_cache_items, delta, delta_gb
+        );
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn test_index_meta_cache_size_file_meta_data() -> common_exception::Result<()> {
+    let thrift_file_meta = setup().await?;
+
+    let cache_number = 300_000;
+
+    let meta: FileMetaData = FileMetaData::try_from_thrift(thrift_file_meta.clone())?;
+
+    let sys = System::new_all();
+    let pid = get_current_pid().unwrap();
+    let process = sys.process(pid).unwrap();
+    let base_memory_usage = process.memory();
+
+    let cache = InMemoryCacheBuilder::new_item_cache::<FileMetaData>(cache_number as u64);
+
+    populate_cache(&cache, meta, cache_number);
+    show_memory_usage("FileMetaData", base_memory_usage, cache_number);
+
+    drop(cache);
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn test_index_meta_cache_size_bloom_meta() -> common_exception::Result<()> {
+    let thrift_file_meta = setup().await?;
+
+    let cache_number = 300_000;
+
+    let meta: FileMetaData = FileMetaData::try_from_thrift(thrift_file_meta.clone())?;
+    let bloom_index_meta = BloomIndexMetaMini::try_from(meta.clone())?;
+
+    let sys = System::new_all();
+    let pid = get_current_pid().unwrap();
+    let process = sys.process(pid).unwrap();
+    let base_memory_usage = process.memory();
+
+    let cache = InMemoryCacheBuilder::new_item_cache::<BloomIndexMetaMini>(cache_number as u64);
+    populate_cache(&cache, bloom_index_meta, cache_number);
+    show_memory_usage("BloomIndexMetaMini", base_memory_usage, cache_number);
+
+    drop(cache);
+
+    Ok(())
+}

src/query/service/tests/it/storages/fuse/statistics.rs

@@ -24,6 +24,7 @@ use storages_common_cache::CacheAccessor;
 use storages_common_cache::InMemoryItemCacheHolder;
 use storages_common_cache::NamedCache;
 use storages_common_index::filters::Xor8Filter;
+use storages_common_index::BloomIndexMetaMini;
 use storages_common_table_meta::meta::SegmentInfo;
 use storages_common_table_meta::meta::TableSnapshot;
 use storages_common_table_meta::meta::TableSnapshotStatistics;
@@ -39,9 +40,8 @@ pub type TableSnapshotStatisticCache = NamedCache<InMemoryItemCacheHolder<TableS
 /// In memory object cache of bloom filter.
 /// For each indexed data block, the bloom xor8 filter of column is cached individually
 pub type BloomIndexFilterCache = NamedCache<InMemoryItemCacheHolder<Xor8Filter>>;
-pub struct BloomIndexMeta(pub FileMetaData);
 /// In memory object cache of parquet FileMetaData of bloom index data
-pub type BloomIndexMetaCache = NamedCache<InMemoryItemCacheHolder<BloomIndexMeta>>;
+pub type BloomIndexMetaCache = NamedCache<InMemoryItemCacheHolder<BloomIndexMetaMini>>;
 /// In memory object cache of parquet FileMetaData of external parquet files
 pub type FileMetaDataCache = NamedCache<InMemoryItemCacheHolder<FileMetaData>>;
 
@@ -87,7 +87,7 @@ impl CachedObject<TableSnapshotStatistics> for TableSnapshotStatistics {
     }
 }
 
-impl CachedObject<BloomIndexMeta> for BloomIndexMeta {
+impl CachedObject<BloomIndexMetaMini> for BloomIndexMetaMini {
     type Cache = BloomIndexMetaCache;
     fn cache() -> Option<Self::Cache> {
         CacheManager::instance().get_bloom_index_meta_cache()

src/query/storages/common/cache-manager/src/caches.rs

@@ -15,6 +15,7 @@ test = false
 ignored = ["xorfilter-rs", "match-template"]
 
 [dependencies]
+common-arrow = { path = "../../../../common/arrow" }
 common-exception = { path = "../../../../common/exception" }
 common-expression = { path = "../../../expression" }
 common-functions = { path = "../../../functions" }