feat: enable bloom filter index

common/cache/src/meter/bytes_meter.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::borrow::Borrow;
+use std::sync::Arc;
 
 use super::Meter;
 pub struct BytesMeter;
@@ -24,3 +25,11 @@ impl<K> Meter<K, Vec<u8>> for BytesMeter {
         v.len()
     }
 }
+
+impl<K> Meter<K, Arc<Vec<u8>>> for BytesMeter {
+    type Measure = usize;
+    fn measure<Q: ?Sized>(&self, _: &Q, v: &Arc<Vec<u8>>) -> usize
+    where K: Borrow<Q> {
+        v.len()
+    }
+}

common/catalog/src/table.rs

@@ -211,5 +211,5 @@ pub struct TableStatistics {
     pub num_rows: Option<u64>,
     pub data_size: Option<u64>,
     pub data_size_compressed: Option<u64>,
-    pub index_length: Option<u64>,
+    pub index_size: Option<u64>,
 }

common/datablocks/src/utils.rs

@@ -22,22 +22,23 @@ use common_arrow::parquet::encoding::Encoding;
 use common_arrow::parquet::metadata::ThriftFileMetaData;
 use common_arrow::parquet::write::Version;
 use common_arrow::write_parquet_file;
-use common_datavalues::DataSchemaRef;
+use common_datavalues::DataSchema;
 use common_exception::ErrorCode;
 use common_exception::Result;
 
 use crate::DataBlock;
 
-pub fn serialize_data_blocks(
+pub fn serialize_data_blocks_with_compression(
     blocks: Vec<DataBlock>,
-    schema: &DataSchemaRef,
+    schema: impl AsRef<DataSchema>,
     buf: &mut Vec<u8>,
+    compression: CompressionOptions,
 ) -> Result<(u64, ThriftFileMetaData)> {
-    let arrow_schema = schema.to_arrow();
+    let arrow_schema = schema.as_ref().to_arrow();
 
     let row_group_write_options = WriteOptions {
         write_statistics: false,
-        compression: CompressionOptions::Lz4Raw,
+        compression,
         version: Version::V2,
     };
     let batches = blocks
@@ -75,6 +76,14 @@ pub fn serialize_data_blocks(
     }
 }
 
+pub fn serialize_data_blocks(
+    blocks: Vec<DataBlock>,
+    schema: impl AsRef<DataSchema>,
+    buf: &mut Vec<u8>,
+) -> Result<(u64, ThriftFileMetaData)> {
+    serialize_data_blocks_with_compression(blocks, schema, buf, CompressionOptions::Lz4Raw)
+}
+
 fn col_encoding(_data_type: &ArrowDataType) -> Encoding {
     // Although encoding does work, parquet2 has not implemented decoding of DeltaLengthByteArray yet, we fallback to Plain
     // From parquet2: Decoding "DeltaLengthByteArray"-encoded required V2 pages is not yet implemented for Binary.

common/fuse-meta/Cargo.toml

@@ -9,6 +9,7 @@ edition = "2021"
 [features]
 
 [dependencies]
+common-arrow = { path = "../arrow" }
 common-base = { path = "../base" }
 common-cache = { path = "../cache" }
 common-config = { path = "../config" }

common/fuse-meta/src/caches/cache.rs

@@ -14,15 +14,26 @@
 
 use common_config::QueryConfig;
 
-use crate::caches::new_memory_cache;
-use crate::caches::MemoryCache;
+use crate::caches::memory_cache::new_bytes_cache;
+use crate::caches::memory_cache::BloomIndexCache;
+use crate::caches::memory_cache::BloomIndexMetaCache;
+use crate::caches::memory_cache::BytesCache;
+use crate::caches::new_item_cache;
+use crate::caches::ItemCache;
 use crate::caches::SegmentInfoCache;
 use crate::caches::TableSnapshotCache;
 
+// default number of index meta cached, default 3000 items
+static DEFAULT_BLOOM_INDEX_META_CACHE_ITEMS: u64 = 3000;
+// default size of cached bloom filter index (in bytes), 1G
+static DEFAULT_BLOOM_INDEX_COLUMN_CACHE_SIZE: u64 = 1024 * 1024 * 1024;
+
 /// Where all the caches reside
 pub struct CacheManager {
     table_snapshot_cache: Option<TableSnapshotCache>,
     segment_info_cache: Option<SegmentInfoCache>,
+    bloom_index_cache: Option<BloomIndexCache>,
+    bloom_index_meta_cache: Option<BloomIndexMetaCache>,
     cluster_id: String,
     tenant_id: String,
 }
@@ -36,15 +47,22 @@ impl CacheManager {
             Self {
                 table_snapshot_cache: None,
                 segment_info_cache: None,
+                bloom_index_cache: None,
+                bloom_index_meta_cache: None,
                 cluster_id: config.cluster_id.clone(),
                 tenant_id: config.tenant_id.clone(),
             }
         } else {
-            let table_snapshot_cache = Self::with_capacity(config.table_cache_snapshot_count);
-            let segment_info_cache = Self::with_capacity(config.table_cache_segment_count);
+            let table_snapshot_cache = Self::new_item_cache(config.table_cache_snapshot_count);
+            let segment_info_cache = Self::new_item_cache(config.table_cache_segment_count);
+            let bloom_index_cache = Self::new_bytes_cache(DEFAULT_BLOOM_INDEX_META_CACHE_ITEMS);
+            let bloom_index_meta_cache =
+                Self::new_item_cache(DEFAULT_BLOOM_INDEX_COLUMN_CACHE_SIZE);
             Self {
                 table_snapshot_cache,
                 segment_info_cache,
+                bloom_index_cache,
+                bloom_index_meta_cache,
                 cluster_id: config.cluster_id.clone(),
                 tenant_id: config.tenant_id.clone(),
             }
@@ -59,6 +77,14 @@ impl CacheManager {
         self.segment_info_cache.clone()
     }
 
+    pub fn get_bloom_index_cache(&self) -> Option<BloomIndexCache> {
+        self.bloom_index_cache.clone()
+    }
+
+    pub fn get_bloom_index_meta_cache(&self) -> Option<BloomIndexMetaCache> {
+        self.bloom_index_meta_cache.clone()
+    }
+
     pub fn get_tenant_id(&self) -> &str {
         self.tenant_id.as_str()
     }
@@ -67,9 +93,17 @@ impl CacheManager {
         self.cluster_id.as_str()
     }
 
-    fn with_capacity<T>(capacity: u64) -> Option<MemoryCache<T>> {
+    fn new_item_cache<T>(capacity: u64) -> Option<ItemCache<T>> {
+        if capacity > 0 {
+            Some(new_item_cache(capacity))
+        } else {
+            None
+        }
+    }
+
+    fn new_bytes_cache(capacity: u64) -> Option<BytesCache> {
         if capacity > 0 {
-            Some(new_memory_cache(capacity))
+            Some(new_bytes_cache(capacity))
         } else {
             None
         }

common/fuse-meta/src/caches/memory_cache.rs

@@ -14,19 +14,37 @@
 
 use std::sync::Arc;
 
+use common_arrow::parquet::metadata::FileMetaData;
 use common_base::base::tokio::sync::RwLock;
+use common_cache::BytesMeter;
+use common_cache::Cache;
 use common_cache::Count;
 use common_cache::DefaultHashBuilder;
 use common_cache::LruCache;
 
 use crate::meta::SegmentInfo;
 use crate::meta::TableSnapshot;
 
-pub type MemoryCache<V> = Arc<RwLock<LruCache<String, Arc<V>, DefaultHashBuilder, Count>>>;
+// cache meters by counting number of items
+pub type ItemCache<V> = Arc<RwLock<LruCache<String, Arc<V>, DefaultHashBuilder, Count>>>;
 
-pub fn new_memory_cache<V>(capacity: u64) -> MemoryCache<V> {
+// cache meters by bytes
+pub type BytesCache = Arc<RwLock<LruCache<String, Arc<Vec<u8>>, DefaultHashBuilder, BytesMeter>>>;
+
+pub fn new_item_cache<V>(capacity: u64) -> ItemCache<V> {
     Arc::new(RwLock::new(LruCache::new(capacity)))
 }
 
-pub type SegmentInfoCache = MemoryCache<SegmentInfo>;
-pub type TableSnapshotCache = MemoryCache<TableSnapshot>;
+pub fn new_bytes_cache(capacity: u64) -> BytesCache {
+    let c = LruCache::with_meter_and_hasher(capacity, BytesMeter, DefaultHashBuilder::new());
+    Arc::new(RwLock::new(c))
+}
+
+pub type SegmentInfoCache = ItemCache<SegmentInfo>;
+pub type TableSnapshotCache = ItemCache<TableSnapshot>;
+/// Cache bloom filter.
+/// For each index block, columns are cached individually.
+pub type BloomIndexCache = BytesCache;
+/// FileMetaCache of bloom filter index data.
+/// Each cache item per block
+pub type BloomIndexMetaCache = ItemCache<FileMetaData>;

common/fuse-meta/src/caches/mod.rs

@@ -17,8 +17,8 @@ mod memory_cache;
 mod metrics;
 
 pub use cache::CacheManager;
-pub use memory_cache::new_memory_cache;
-pub use memory_cache::MemoryCache;
+pub use memory_cache::new_item_cache;
+pub use memory_cache::ItemCache;
 pub use memory_cache::SegmentInfoCache;
 pub use memory_cache::TableSnapshotCache;
 

common/fuse-meta/src/meta/common.rs

@@ -58,6 +58,8 @@ pub struct Statistics {
 
     pub uncompressed_byte_size: u64,
     pub compressed_byte_size: u64,
+    #[serde(default)]
+    pub index_size: u64,
 
     pub col_stats: HashMap<ColumnId, ColumnStatistics>,
 }

common/fuse-meta/src/meta/current/mod.rs

@@ -13,6 +13,7 @@
 //  limitations under the License.
 
 pub use v0::ColumnMeta;
+pub use v1::BlockBloomFilterIndex;
 pub use v1::BlockMeta;
 pub use v1::SegmentInfo;
 pub use v1::TableSnapshot;

common/fuse-meta/src/meta/mod.rs

@@ -31,5 +31,6 @@ pub use common::Statistics;
 pub use common::StatisticsOfColumns;
 pub use common::Versioned;
 pub use current::*;
+pub use versions::BlockBloomFilterIndexVersion;
 pub use versions::SegmentInfoVersion;
 pub use versions::SnapshotVersion;

common/fuse-meta/src/meta/v1/index.rs

@@ -0,0 +1,33 @@
+//  Copyright 2022 Datafuse Labs.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+use common_datablocks::DataBlock;
+
+// index data of Block, which itself is also a DataBlock.
+//
+// depends on the query conditions, columns of index data will be loaded on demand.
+pub struct BlockBloomFilterIndex {
+    // Before index mod is extracted from databend-query, we just keep the DataBlock here
+    data: DataBlock,
+}
+
+impl BlockBloomFilterIndex {
+    pub fn new(data: DataBlock) -> Self {
+        Self { data }
+    }
+
+    pub fn into_data(self) -> DataBlock {
+        self.data
+    }
+}

common/fuse-meta/src/meta/v1/mod.rs

@@ -12,9 +12,11 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+mod index;
 mod segment;
 mod snapshot;
 
+pub use index::BlockBloomFilterIndex;
 pub use segment::BlockMeta;
 pub use segment::SegmentInfo;
 pub use snapshot::TableSnapshot;

common/fuse-meta/src/meta/v1/segment.rs

@@ -49,7 +49,13 @@ pub struct BlockMeta {
     pub col_stats: HashMap<ColumnId, ColumnStatistics>,
     pub col_metas: HashMap<ColumnId, ColumnMeta>,
     pub cluster_stats: Option<ClusterStatistics>,
+    /// location of data block
     pub location: Location,
+    /// location of bloom filter index
+    pub bloom_filter_index_location: Option<Location>,
+
+    #[serde(default)]
+    pub bloom_filter_index_size: u64,
 
     /// Compression algo used to compress the columns of blocks
     ///
@@ -61,6 +67,7 @@ pub struct BlockMeta {
 }
 
 impl BlockMeta {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         row_count: u64,
         block_size: u64,
@@ -69,6 +76,8 @@ impl BlockMeta {
         col_metas: HashMap<ColumnId, ColumnMeta>,
         cluster_stats: Option<ClusterStatistics>,
         location: Location,
+        bloom_filter_index_location: Option<Location>,
+        bloom_filter_index_size: u64,
     ) -> Self {
         Self {
             row_count,
@@ -78,6 +87,8 @@ impl BlockMeta {
             col_metas,
             cluster_stats,
             location,
+            bloom_filter_index_location,
+            bloom_filter_index_size,
             compression: Compression::Lz4Raw,
         }
     }
@@ -123,6 +134,8 @@ impl From<v0::BlockMeta> for BlockMeta {
             col_metas: s.col_metas,
             cluster_stats: None,
             location: (s.location.path, DataBlock::VERSION),
+            bloom_filter_index_location: None,
+            bloom_filter_index_size: 0,
             compression: Compression::Lz4,
         }
     }