115-offloadable-bloom-filter

src/blob/core.rs

@@ -349,6 +349,19 @@ impl Blob {
         }
     }
 
+    pub(crate) fn check_filters_in_memory(&self, key: &[u8]) -> bool {
+        trace!("check filters (range and bloom)");
+        if let FilterResult::NotContains = self.index.check_filters_in_memory(key) {
+            false
+        } else {
+            true
+        }
+    }
+
+    pub(crate) fn is_filter_offloaded(&self) -> bool {
+        self.index.is_filter_offloaded()
+    }
+
     pub(crate) fn index_memory(&self) -> usize {
         self.index.memory_used()
     }

src/blob/index/bloom.rs

@@ -5,6 +5,7 @@ use bitvec::order::Lsb0;
 #[derive(Debug, Clone)]
 pub(crate) struct Bloom {
     inner: Option<BitVec<Lsb0, u64>>,
+    offset_in_file: Option<u64>,
     bits_count: usize,
     hashers: Vec<AHasher>,
     config: Config,
@@ -17,6 +18,7 @@ impl Default for Bloom {
             bits_count: 0,
             hashers: vec![],
             config: Default::default(),
+            offset_in_file: None,
         }
     }
 }
@@ -109,17 +111,28 @@ impl Bloom {
             hashers: Self::hashers(config.hashers_count),
             config,
             bits_count,
+            offset_in_file: None,
         }
     }
 
     pub fn clear(&mut self) {
         self.inner = Some(bitvec![Lsb0, u64; 0; self.bits_count]);
+        self.offset_in_file = None;
+    }
+
+    pub fn is_offloaded(&self) -> bool {
+        self.inner.is_none()
     }
 
     pub fn offload_from_memory(&mut self) {
         self.inner = None;
     }
 
+    pub fn set_offset_in_file(&mut self, offset: u64) {
+        self.offset_in_file =
+            Some(offset + self.buffer_start_position().expect("Should not fail") as u64);
+    }
+
     pub fn hashers(k: usize) -> Vec<AHasher> {
         trace!("@TODO create configurable hashers???");
         (0..k)
@@ -147,6 +160,7 @@ impl Bloom {
             config: save.config,
             inner: Some(inner),
             bits_count: save.bits_count,
+            offset_in_file: None,
         }
     }
 
@@ -209,7 +223,9 @@ impl Bloom {
         if self.bits_count == 0 {
             return Ok(false);
         }
-        let start_pos = self.buffer_start_position()?;
+        let start_pos = self
+            .offset_in_file
+            .ok_or_else(|| anyhow::anyhow!("Offset should be set for in-file operations"))?;
         for index in hashers.iter_mut().map(|hasher| {
             hasher.write(item.as_ref());
             hasher.finish() % self.bits_count as u64
@@ -241,7 +257,6 @@ impl Bloom {
 #[async_trait::async_trait]
 pub(crate) trait BloomDataProvider {
     async fn read_byte(&self, index: u64) -> Result<u8>;
-    async fn read_all(&self) -> Result<Vec<u8>>;
 }
 
 mod tests {

src/blob/index/core.rs

@@ -46,7 +46,6 @@ pub(crate) struct IndexStruct<FileIndex: FileIndexTrait> {
     mem: Option<MemoryAttrs>,
     range_filter: RangeFilter,
     bloom_filter: Bloom,
-    bloom_offset: usize,
     params: IndexParams,
     inner: State<FileIndex>,
     name: FileName,
@@ -81,7 +80,6 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
             params,
             bloom_filter: filter,
             range_filter: RangeFilter::new(),
-            bloom_offset: 0,
             inner: State::InMemory(BTreeMap::new()),
             mem,
             name,
@@ -114,6 +112,18 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
         }
     }
 
+    pub(crate) fn check_filters_in_memory(&self, key: &[u8]) -> FilterResult {
+        if !self.range_filter.contains(key) {
+            FilterResult::NotContains
+        } else if self.params.bloom_is_on
+            && self.bloom_filter.contains_in_memory(key) == Some(false)
+        {
+            FilterResult::NotContains
+        } else {
+            FilterResult::NeedAdditionalCheck
+        }
+    }
+
     pub async fn check_bloom_key(&self, key: &[u8]) -> Result<Option<bool>> {
         if self.params.bloom_is_on {
             if let Some(result) = self.bloom_filter.contains_in_memory(key) {
@@ -126,6 +136,10 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
         }
     }
 
+    pub fn is_filter_offloaded(&self) -> bool {
+        self.bloom_filter.is_offloaded()
+    }
+
     pub fn bloom_memory_allocated(&self) -> usize {
         self.bloom_filter.memory_allocated()
     }
@@ -142,14 +156,13 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
         let findex = FileIndex::from_file(name.clone(), ioring.clone()).await?;
         findex.validate().with_context(|| "Header is corrupt")?;
         let meta_buf = findex.read_meta().await?;
-        let (bloom_filter, range_filter, bloom_offset) = Self::deserialize_filters(&meta_buf)?;
+        let (bloom_filter, range_filter) = Self::deserialize_filters(&meta_buf)?;
         let params = IndexParams::new(config.bloom_config.is_some(), config.recreate_index_file);
         trace!("index restored successfuly");
         let index = Self {
             inner: State::OnDisk(findex),
             mem: None,
             name,
-            bloom_offset,
             bloom_filter,
             range_filter,
             params,
@@ -169,6 +182,7 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
             }
             debug!("blob index simple in memory headers {}", headers.len());
             let (meta_buf, bloom_offset) = self.serialize_filters()?;
+            self.bloom_filter.set_offset_in_file(bloom_offset as u64);
             let findex = FileIndex::from_records(
                 &self.name.to_path(),
                 self.ioring.clone(),
@@ -180,7 +194,6 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
             let size = findex.file_size() as usize;
             self.inner = State::OnDisk(findex);
             self.mem = None;
-            self.bloom_offset = bloom_offset;
             Ok(size)
         } else {
             Ok(0)
@@ -199,24 +212,24 @@ impl<FileIndex: FileIndexTrait> IndexStruct<FileIndex> {
         Ok((buf, bloom_offset))
     }
 
-    fn deserialize_filters(buf: &[u8]) -> Result<(Bloom, RangeFilter, usize)> {
+    fn deserialize_filters(buf: &[u8]) -> Result<(Bloom, RangeFilter)> {
         let (range_size_buf, rest_buf) = buf.split_at(size_of::<u64>());
         let range_size = deserialize(&range_size_buf)?;
         let (range_buf, bloom_buf) = rest_buf.split_at(range_size);
-        let bloom = Bloom::from_raw(bloom_buf)?;
+        let mut bloom = Bloom::from_raw(bloom_buf)?;
+        bloom.set_offset_in_file((range_size + size_of::<u64>()) as u64);
         let range = RangeFilter::from_raw(range_buf)?;
-        Ok((bloom, range, range_size + size_of::<u64>()))
+        Ok((bloom, range))
     }
 
     async fn load_in_memory(&mut self, findex: FileIndex) -> Result<()> {
         let (record_headers, records_count) = findex.get_records_headers().await?;
         self.mem = Some(compute_mem_attrs(&record_headers, records_count));
         self.inner = State::InMemory(record_headers);
         let meta_buf = findex.read_meta().await?;
-        let (bloom_filter, range_filter, bloom_offset) = Self::deserialize_filters(&meta_buf)?;
+        let (bloom_filter, range_filter) = Self::deserialize_filters(&meta_buf)?;
         self.bloom_filter = bloom_filter;
         self.range_filter = range_filter;
-        self.bloom_offset = bloom_offset;
         Ok(())
     }
 
@@ -357,20 +370,7 @@ pub(crate) trait FileIndexTrait: Sized + Send + Sync {
 impl<FileIndex: FileIndexTrait> BloomDataProvider for IndexStruct<FileIndex> {
     async fn read_byte(&self, index: u64) -> Result<u8> {
         match &self.inner {
-            State::OnDisk(findex) => findex.read_meta_at(self.bloom_offset as u64 + index).await,
-            _ => Err(anyhow::anyhow!("Can't read from in-memory index")),
-        }
-    }
-
-    async fn read_all(&self) -> Result<Vec<u8>> {
-        match &self.inner {
-            State::OnDisk(findex) => {
-                let meta = findex.read_meta().await?;
-                Ok(meta
-                    .get(self.bloom_offset..)
-                    .ok_or_else(|| anyhow::anyhow!("Incorrect bloom offset"))?
-                    .to_vec())
-            }
+            State::OnDisk(findex) => findex.read_meta_at(index).await,
             _ => Err(anyhow::anyhow!("Can't read from in-memory index")),
         }
     }

src/storage/core.rs

@@ -795,12 +795,24 @@ impl<K: Key> Storage<K> {
             .read()
             .await
             .iter()
-            .map(|blob| blob.check_filters(key.as_ref()))
-            .collect::<FuturesUnordered<_>>()
-            .any(|value| value)
-            .await;
+            .filter(|blob| !blob.is_filter_offloaded())
+            .any(|blob| blob.check_filters_in_memory(key.as_ref()));
 
-        Some(in_active || in_closed)
+        if !(in_active || in_closed) {
+            let in_closed_offloaded = inner
+                .blobs
+                .read()
+                .await
+                .iter()
+                .filter(|blob| blob.is_filter_offloaded())
+                .map(|blob| blob.check_filters(key.as_ref()))
+                .collect::<FuturesUnordered<_>>()
+                .any(|value| value)
+                .await;
+            Some(in_closed_offloaded)
+        } else {
+            Some(true)
+        }
     }
 
     /// Offload bloom filters for closed blobs