Reorganise facets database indexing code

milli/src/update/facet/bulk.rs

@@ -0,0 +1,360 @@
+use crate::error::InternalError;
+use crate::heed_codec::facet::new::{
+    FacetGroupValue, FacetGroupValueCodec, FacetKey, FacetKeyCodec, MyByteSlice,
+};
+use crate::update::index_documents::{create_writer, write_into_lmdb_database, writer_into_reader};
+use crate::{FieldId, Index, Result};
+use grenad::CompressionType;
+use heed::types::ByteSlice;
+use heed::{BytesEncode, Error, RoTxn};
+use log::debug;
+use roaring::RoaringBitmap;
+use std::cmp;
+use std::fs::File;
+use std::num::NonZeroUsize;
+use time::OffsetDateTime;
+
+pub struct FacetsUpdateBulk<'i> {
+    index: &'i Index,
+    database: heed::Database<FacetKeyCodec<MyByteSlice>, FacetGroupValueCodec>,
+    pub(crate) chunk_compression_type: CompressionType,
+    pub(crate) chunk_compression_level: Option<u32>,
+    level_group_size: usize,
+    min_level_size: usize,
+}
+
+impl<'i> FacetsUpdateBulk<'i> {
+    pub fn new(
+        index: &'i Index,
+        database: heed::Database<FacetKeyCodec<MyByteSlice>, FacetGroupValueCodec>,
+    ) -> FacetsUpdateBulk<'i> {
+        FacetsUpdateBulk {
+            index,
+            database,
+            chunk_compression_type: CompressionType::None,
+            chunk_compression_level: None,
+            level_group_size: 4,
+            min_level_size: 5,
+        }
+    }
+
+    /// The number of elements from the level below that are represented by a single element in the level above
+    ///
+    /// This setting is always greater than or equal to 2.
+    pub fn level_group_size(&mut self, value: NonZeroUsize) -> &mut Self {
+        self.level_group_size = cmp::max(value.get(), 2);
+        self
+    }
+
+    /// The minimum number of elements that a level is allowed to have.
+    pub fn min_level_size(&mut self, value: NonZeroUsize) -> &mut Self {
+        self.min_level_size = value.get();
+        self
+    }
+
+    fn clear_levels(&self, wtxn: &mut heed::RwTxn, field_id: FieldId) -> Result<()> {
+        let left = FacetKey::<&[u8]> { field_id, level: 1, left_bound: &[] };
+        let right = FacetKey::<&[u8]> { field_id, level: u8::MAX, left_bound: &[] };
+        let range = left..=right;
+        self.database.delete_range(wtxn, &range).map(drop)?;
+        Ok(())
+    }
+
+    #[logging_timer::time("FacetsUpdateBulk::{}")]
+    pub fn execute(self, wtxn: &mut heed::RwTxn) -> Result<()> {
+        self.index.set_updated_at(wtxn, &OffsetDateTime::now_utc())?;
+        // We get the faceted fields to be able to create the facet levels.
+        let faceted_fields = self.index.faceted_fields_ids(wtxn)?.clone();
+
+        debug!("Computing and writing the facet values levels docids into LMDB on disk...");
+
+        for &field_id in faceted_fields.iter() {
+            self.clear_levels(wtxn, field_id)?;
+        }
+
+        let mut nested_wtxn = self.index.env.nested_write_txn(wtxn)?;
+
+        for &field_id in faceted_fields.iter() {
+            let (level_readers, all_docids) =
+                self.compute_levels_for_field_id(field_id, &nested_wtxn)?;
+
+            // TODO: this will need to be an argument to Facets as well
+            self.index.put_string_faceted_documents_ids(&mut nested_wtxn, field_id, &all_docids)?;
+
+            for level_reader in level_readers {
+                // TODO: append instead of write with merge
+                write_into_lmdb_database(
+                    &mut nested_wtxn,
+                    *self.database.as_polymorph(),
+                    level_reader,
+                    |_, _| {
+                        Err(InternalError::IndexingMergingKeys { process: "facet string levels" })?
+                    },
+                )?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn compute_levels_for_field_id(
+        &self,
+        field_id: FieldId,
+        txn: &RoTxn,
+    ) -> Result<(Vec<grenad::Reader<File>>, RoaringBitmap)> {
+        let algo = FacetsUpdateBulkAlgorithm {
+            rtxn: txn,
+            db: &self.database,
+            field_id,
+            level_group_size: self.level_group_size,
+            min_level_size: self.min_level_size,
+            chunk_compression_type: self.chunk_compression_type,
+            chunk_compression_level: self.chunk_compression_level,
+        };
+        // TODO: first check whether there is anything in level 0
+
+        let mut all_docids = RoaringBitmap::new();
+        let subwriters = algo.compute_higher_levels(32, &mut |bitmaps, _| {
+            for bitmap in bitmaps {
+                all_docids |= bitmap;
+            }
+            Ok(())
+        })?;
+        drop(algo);
+
+        Ok((subwriters, all_docids))
+    }
+}
+
+pub struct FacetsUpdateBulkAlgorithm<'t> {
+    rtxn: &'t heed::RoTxn<'t>,
+    db: &'t heed::Database<FacetKeyCodec<MyByteSlice>, FacetGroupValueCodec>,
+    chunk_compression_type: CompressionType,
+    chunk_compression_level: Option<u32>,
+    field_id: u16,
+    level_group_size: usize,
+    min_level_size: usize,
+}
+impl<'t> FacetsUpdateBulkAlgorithm<'t> {
+    fn read_level_0(
+        &self,
+        handle_group: &mut dyn FnMut(&[RoaringBitmap], &'t [u8]) -> Result<()>,
+    ) -> Result<()> {
+        // we read the elements one by one and
+        // 1. keep track of the left bound
+        // 2. fill the `bitmaps` vector to give it to level 1 once `level_group_size` elements were read
+        let mut bitmaps = vec![];
+
+        let mut level_0_prefix = vec![];
+        level_0_prefix.extend_from_slice(&self.field_id.to_be_bytes());
+        level_0_prefix.push(0);
+
+        let level_0_iter = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, ByteSlice, ByteSlice>(self.rtxn, level_0_prefix.as_slice())?
+            .remap_types::<FacetKeyCodec<MyByteSlice>, FacetGroupValueCodec>();
+
+        let mut left_bound: &[u8] = &[];
+        let mut first_iteration_for_new_group = true;
+        for el in level_0_iter {
+            let (key, value) = el?;
+            let bound = key.left_bound;
+            let docids = value.bitmap;
+
+            if first_iteration_for_new_group {
+                left_bound = bound;
+                first_iteration_for_new_group = false;
+            }
+            bitmaps.push(docids);
+
+            if bitmaps.len() == self.level_group_size {
+                handle_group(&bitmaps, left_bound)?;
+                first_iteration_for_new_group = true;
+                bitmaps.clear();
+            }
+        }
+        // don't forget to give the leftover bitmaps as well
+        if !bitmaps.is_empty() {
+            handle_group(&bitmaps, left_bound)?;
+            bitmaps.clear();
+        }
+        Ok(())
+    }
+
+    /// Compute the content of the database levels from its level 0 for the given field id.
+    ///
+    /// ## Returns:
+    /// 1. a vector of grenad::Reader. The reader at index `i` corresponds to the elements of level `i + 1`
+    /// that must be inserted into the database.
+    /// 2. a roaring bitmap of all the document ids present in the database
+    fn compute_higher_levels(
+        &self,
+        level: u8,
+        handle_group: &mut dyn FnMut(&[RoaringBitmap], &'t [u8]) -> Result<()>,
+    ) -> Result<Vec<grenad::Reader<File>>> {
+        if level == 0 {
+            self.read_level_0(handle_group)?;
+            // Level 0 is already in the database
+            return Ok(vec![]);
+        }
+        // level >= 1
+        // we compute each element of this level based on the elements of the level below it
+        // once we have computed `level_group_size` elements, we give the left bound
+        // of those elements, and their bitmaps, to the level above
+
+        let mut cur_writer = create_writer(
+            self.chunk_compression_type,
+            self.chunk_compression_level,
+            tempfile::tempfile()?,
+        );
+        let mut cur_writer_len = 0;
+
+        let mut group_sizes = vec![];
+        let mut left_bounds = vec![];
+        let mut bitmaps = vec![];
+
+        // compute the levels below
+        // in the callback, we fill `cur_writer` with the correct elements for this level
+        let mut sub_writers =
+            self.compute_higher_levels(level - 1, &mut |sub_bitmaps, left_bound| {
+                let mut combined_bitmap = RoaringBitmap::default();
+                for bitmap in sub_bitmaps {
+                    combined_bitmap |= bitmap;
+                }
+                group_sizes.push(sub_bitmaps.len() as u8);
+                left_bounds.push(left_bound);
+
+                bitmaps.push(combined_bitmap);
+                if bitmaps.len() != self.level_group_size {
+                    return Ok(());
+                }
+                let left_bound = left_bounds.first().unwrap();
+                handle_group(&bitmaps, left_bound)?;
+
+                for ((bitmap, left_bound), group_size) in
+                    bitmaps.drain(..).zip(left_bounds.drain(..)).zip(group_sizes.drain(..))
+                {
+                    let key = FacetKey { field_id: self.field_id, level, left_bound };
+                    let key =
+                        FacetKeyCodec::<MyByteSlice>::bytes_encode(&key).ok_or(Error::Encoding)?;
+                    let value = FacetGroupValue { size: group_size, bitmap };
+                    let value =
+                        FacetGroupValueCodec::bytes_encode(&value).ok_or(Error::Encoding)?;
+                    cur_writer.insert(key, value)?;
+                    cur_writer_len += 1;
+                }
+                Ok(())
+            })?;
+        // don't forget to insert the leftover elements into the writer as well
+        if !bitmaps.is_empty() && cur_writer_len >= self.level_group_size * self.min_level_size {
+            let left_bound = left_bounds.first().unwrap();
+            handle_group(&bitmaps, left_bound)?;
+            for ((bitmap, left_bound), group_size) in
+                bitmaps.drain(..).zip(left_bounds.drain(..)).zip(group_sizes.drain(..))
+            {
+                let key = FacetKey { field_id: self.field_id, level, left_bound };
+                let key =
+                    FacetKeyCodec::<MyByteSlice>::bytes_encode(&key).ok_or(Error::Encoding)?;
+                let value = FacetGroupValue { size: group_size, bitmap };
+                let value = FacetGroupValueCodec::bytes_encode(&value).ok_or(Error::Encoding)?;
+                cur_writer.insert(key, value)?;
+                cur_writer_len += 1;
+            }
+        }
+        if cur_writer_len > self.level_group_size * self.min_level_size {
+            sub_writers.push(writer_into_reader(cur_writer)?);
+        }
+        return Ok(sub_writers);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroUsize;
+
+    use crate::db_snap;
+    use crate::documents::documents_batch_reader_from_objects;
+    use crate::index::tests::TempIndex;
+
+    #[test]
+    fn test_facets_number() {
+        let test =
+            |name: &str, group_size: Option<NonZeroUsize>, min_level_size: Option<NonZeroUsize>| {
+                let mut index = TempIndex::new_with_map_size(4096 * 1000 * 10); // 40MB
+                index.index_documents_config.autogenerate_docids = true;
+                index.index_documents_config.facet_level_group_size = group_size;
+                index.index_documents_config.facet_min_level_size = min_level_size;
+
+                index
+                    .update_settings(|settings| {
+                        settings.set_filterable_fields(
+                            IntoIterator::into_iter(["facet".to_owned(), "facet2".to_owned()])
+                                .collect(),
+                        );
+                    })
+                    .unwrap();
+
+                let mut documents = vec![];
+                for i in 0..1_000 {
+                    documents.push(serde_json::json!({ "facet": i }).as_object().unwrap().clone());
+                }
+                for i in 0..100 {
+                    documents.push(serde_json::json!({ "facet2": i }).as_object().unwrap().clone());
+                }
+                let documents = documents_batch_reader_from_objects(documents);
+
+                index.add_documents(documents).unwrap();
+
+                db_snap!(index, facet_id_f64_docids, name);
+            };
+
+        test("default", None, None);
+        test("tiny_groups_tiny_levels", NonZeroUsize::new(1), NonZeroUsize::new(1));
+        test("small_groups_small_levels", NonZeroUsize::new(2), NonZeroUsize::new(2));
+        test("small_groups_large_levels", NonZeroUsize::new(2), NonZeroUsize::new(128));
+        test("large_groups_small_levels", NonZeroUsize::new(16), NonZeroUsize::new(2));
+        test("large_groups_large_levels", NonZeroUsize::new(16), NonZeroUsize::new(256));
+    }
+
+    #[test]
+    fn test_facets_string() {
+        let test = |name: &str,
+                    group_size: Option<NonZeroUsize>,
+                    min_level_size: Option<NonZeroUsize>| {
+            let mut index = TempIndex::new_with_map_size(4096 * 1000 * 10); // 40MB
+            index.index_documents_config.autogenerate_docids = true;
+            index.index_documents_config.facet_level_group_size = group_size;
+            index.index_documents_config.facet_min_level_size = min_level_size;
+
+            index
+                .update_settings(|settings| {
+                    settings.set_filterable_fields(
+                        IntoIterator::into_iter(["facet".to_owned(), "facet2".to_owned()])
+                            .collect(),
+                    );
+                })
+                .unwrap();
+
+            let mut documents = vec![];
+            for i in 0..100 {
+                documents.push(
+                    serde_json::json!({ "facet": format!("s{i:X}") }).as_object().unwrap().clone(),
+                );
+            }
+            for i in 0..10 {
+                documents.push(
+                    serde_json::json!({ "facet2": format!("s{i:X}") }).as_object().unwrap().clone(),
+                );
+            }
+            let documents = documents_batch_reader_from_objects(documents);
+
+            index.add_documents(documents).unwrap();
+
+            db_snap!(index, facet_id_string_docids, name);
+        };
+
+        test("default", None, None);
+        test("tiny_groups_tiny_levels", NonZeroUsize::new(1), NonZeroUsize::new(1));
+    }
+}

milli/src/update/facet/incremental.rs

@@ -0,0 +1,456 @@
+use crate::heed_codec::facet::new::{
+    FacetGroupValue, FacetGroupValueCodec, FacetKey, FacetKeyCodec, MyByteSlice,
+};
+use crate::search::facet::get_highest_level;
+use crate::Result;
+use heed::Error;
+use heed::{types::ByteSlice, BytesDecode, RoTxn, RwTxn};
+use roaring::RoaringBitmap;
+
+enum InsertionResult {
+    InPlace,
+    Insert,
+}
+enum DeletionResult {
+    InPlace,
+    Reduce { prev: Option<Vec<u8>>, next: Option<Vec<u8>> },
+    Remove { prev: Option<Vec<u8>>, next: Option<Vec<u8>> },
+}
+
+struct FacetUpdateIncremental {
+    db: heed::Database<FacetKeyCodec<MyByteSlice>, FacetGroupValueCodec>,
+    group_size: usize,
+    min_level_size: usize,
+    max_group_size: usize,
+}
+impl FacetUpdateIncremental {
+    fn find_insertion_key_value(
+        &self,
+        field_id: u16,
+        level: u8,
+        search_key: &[u8],
+        txn: &RoTxn,
+    ) -> Result<(FacetKey<Vec<u8>>, FacetGroupValue)> {
+        let mut prefix = vec![];
+        prefix.extend_from_slice(&field_id.to_be_bytes());
+        prefix.push(level);
+        prefix.extend_from_slice(search_key);
+
+        let mut prefix_iter = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, MyByteSlice, FacetGroupValueCodec>(txn, &prefix.as_slice())?;
+        if let Some(e) = prefix_iter.next() {
+            let (key_bytes, value) = e?;
+            Ok((
+                FacetKeyCodec::<MyByteSlice>::bytes_decode(&key_bytes)
+                    .ok_or(Error::Encoding)?
+                    .into_owned(),
+                value,
+            ))
+        } else {
+            let key = FacetKey { field_id, level, left_bound: search_key };
+            match self.db.get_lower_than(txn, &key)? {
+                Some((key, value)) => {
+                    if key.level != level || key.field_id != field_id {
+                        let mut prefix = vec![];
+                        prefix.extend_from_slice(&field_id.to_be_bytes());
+                        prefix.push(level);
+
+                        let mut iter = self
+                            .db
+                            .as_polymorph()
+                            .prefix_iter::<_, MyByteSlice, FacetGroupValueCodec>(
+                                txn,
+                                &prefix.as_slice(),
+                            )?;
+                        let (key_bytes, value) = iter.next().unwrap()?;
+                        Ok((
+                            FacetKeyCodec::<MyByteSlice>::bytes_decode(&key_bytes)
+                                .ok_or(Error::Encoding)?
+                                .into_owned(),
+                            value,
+                        ))
+                    } else {
+                        Ok((key.into_owned(), value))
+                    }
+                }
+                None => panic!(),
+            }
+        }
+    }
+
+    fn insert_in_level_0<'t>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        new_key: &[u8],
+        new_values: &RoaringBitmap,
+    ) -> Result<InsertionResult> {
+        let key = FacetKey { field_id, level: 0, left_bound: new_key };
+        let value = FacetGroupValue { bitmap: new_values.clone(), size: 1 };
+
+        let mut level0_prefix = vec![];
+        level0_prefix.extend_from_slice(&field_id.to_be_bytes());
+        level0_prefix.push(0);
+
+        let mut iter = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, ByteSlice, FacetGroupValueCodec>(&txn, &level0_prefix)?;
+
+        if iter.next().is_none() {
+            drop(iter);
+            self.db.put(txn, &key, &value)?;
+            return Ok(InsertionResult::Insert);
+        } else {
+            drop(iter);
+            let old_value = self.db.get(&txn, &key)?;
+            match old_value {
+                Some(mut updated_value) => {
+                    // now merge the two
+                    updated_value.bitmap |= value.bitmap;
+                    self.db.put(txn, &key, &updated_value)?;
+                    Ok(InsertionResult::InPlace)
+                }
+                None => {
+                    self.db.put(txn, &key, &value)?;
+                    Ok(InsertionResult::Insert)
+                }
+            }
+        }
+    }
+    fn insert_in_level<'t>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        level: u8,
+        new_key: &[u8],
+        new_values: &RoaringBitmap,
+    ) -> Result<InsertionResult> {
+        if level == 0 {
+            return self.insert_in_level_0(txn, field_id, new_key, new_values);
+        }
+
+        let max_group_size = self.max_group_size;
+
+        let (insertion_key, insertion_value) =
+            self.find_insertion_key_value(field_id, level, new_key, txn)?;
+
+        let result = self.insert_in_level(txn, field_id, level - 1, new_key.clone(), new_values)?;
+        // level below inserted an element
+
+        let insertion_key = {
+            let mut new_insertion_key = insertion_key.clone();
+            let mut modified = false;
+
+            if new_key < insertion_key.left_bound.as_slice() {
+                new_insertion_key.left_bound = new_key.to_vec();
+                modified = true;
+            }
+            if modified {
+                let is_deleted = self.db.delete(txn, &insertion_key.as_ref())?;
+                assert!(is_deleted);
+                self.db.put(txn, &new_insertion_key.as_ref(), &insertion_value)?;
+            }
+            new_insertion_key
+        };
+
+        match result {
+            // TODO: this could go above the block recomputing insertion key
+            // because we know that if we inserted in place, the key is not a new one
+            // thus it doesn't extend a group
+            InsertionResult::InPlace => {
+                let mut updated_value = self.db.get(&txn, &insertion_key.as_ref())?.unwrap();
+                updated_value.bitmap |= new_values;
+                self.db.put(txn, &insertion_key.as_ref(), &updated_value)?;
+
+                return Ok(InsertionResult::InPlace);
+            }
+            InsertionResult::Insert => {}
+        }
+        let mut updated_value = self.db.get(&txn, &insertion_key.as_ref())?.unwrap();
+
+        updated_value.size += 1;
+        if updated_value.size as usize == max_group_size {
+            // need to split it
+            // recompute left element and right element
+            // replace current group by left element
+            // add one more group to the right
+
+            let size_left = max_group_size / 2;
+            let size_right = max_group_size - size_left;
+
+            let level_below = level - 1;
+
+            let (start_key, _) = self
+                .db
+                .get_greater_than_or_equal_to(
+                    &txn,
+                    &FacetKey {
+                        field_id,
+                        level: level_below,
+                        left_bound: insertion_key.left_bound.as_slice(),
+                    },
+                )?
+                .unwrap();
+
+            let mut iter = self.db.range(&txn, &(start_key..))?.take(max_group_size);
+
+            let group_left = {
+                let mut values_left = RoaringBitmap::new();
+
+                let mut i = 0;
+                while let Some(next) = iter.next() {
+                    let (_key, value) = next?;
+                    i += 1;
+                    values_left |= &value.bitmap;
+                    if i == size_left {
+                        break;
+                    }
+                }
+
+                let key =
+                    FacetKey { field_id, level, left_bound: insertion_key.left_bound.clone() };
+                let value = FacetGroupValue { size: size_left as u8, bitmap: values_left };
+                (key, value)
+            };
+
+            let group_right = {
+                let mut values_right = RoaringBitmap::new();
+                let mut right_start_key = None;
+
+                while let Some(next) = iter.next() {
+                    let (key, value) = next?;
+                    if right_start_key.is_none() {
+                        right_start_key = Some(key.left_bound);
+                    }
+                    values_right |= &value.bitmap;
+                }
+
+                let key =
+                    FacetKey { field_id, level, left_bound: right_start_key.unwrap().to_vec() };
+                let value = FacetGroupValue { size: size_right as u8, bitmap: values_right };
+                (key, value)
+            };
+            drop(iter);
+
+            let _ = self.db.delete(txn, &insertion_key.as_ref())?;
+
+            self.db.put(txn, &group_left.0.as_ref(), &group_left.1)?;
+            self.db.put(txn, &group_right.0.as_ref(), &group_right.1)?;
+
+            Ok(InsertionResult::Insert)
+        } else {
+            let mut value = self.db.get(&txn, &insertion_key.as_ref())?.unwrap();
+            value.bitmap |= new_values;
+            value.size += 1;
+            self.db.put(txn, &insertion_key.as_ref(), &value).unwrap();
+
+            Ok(InsertionResult::InPlace)
+        }
+    }
+
+    pub fn insert<'a, 't>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        new_key: &[u8],
+        new_values: &RoaringBitmap,
+    ) -> Result<()> {
+        if new_values.is_empty() {
+            return Ok(());
+        }
+        let group_size = self.group_size;
+
+        let highest_level = get_highest_level(&txn, *self.db, field_id)?;
+
+        let result =
+            self.insert_in_level(txn, field_id, highest_level as u8, new_key, new_values)?;
+        match result {
+            InsertionResult::InPlace => return Ok(()),
+            InsertionResult::Insert => {}
+        }
+
+        let mut highest_level_prefix = vec![];
+        highest_level_prefix.extend_from_slice(&field_id.to_be_bytes());
+        highest_level_prefix.push(highest_level);
+
+        let size_highest_level = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, ByteSlice, ByteSlice>(&txn, &highest_level_prefix)?
+            .count();
+
+        if size_highest_level < self.min_level_size {
+            return Ok(());
+        }
+
+        let mut groups_iter = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, ByteSlice, FacetGroupValueCodec>(&txn, &highest_level_prefix)?;
+
+        let mut to_add = vec![];
+        for _ in 0..group_size {
+            let mut first_key = None;
+            let mut values = RoaringBitmap::new();
+            for _ in 0..group_size {
+                let (key_bytes, value_i) = groups_iter.next().unwrap()?;
+                let key_i = FacetKeyCodec::<MyByteSlice>::bytes_decode(&key_bytes)
+                    .ok_or(Error::Encoding)?;
+
+                if first_key.is_none() {
+                    first_key = Some(key_i);
+                }
+                values |= value_i.bitmap;
+            }
+            let key = FacetKey {
+                field_id,
+                level: highest_level + 1,
+                left_bound: first_key.unwrap().left_bound,
+            };
+            let value = FacetGroupValue { size: group_size as u8, bitmap: values };
+            to_add.push((key.into_owned(), value));
+        }
+        drop(groups_iter);
+        for (key, value) in to_add {
+            self.db.put(txn, &key.as_ref(), &value)?;
+        }
+        Ok(())
+    }
+
+    fn delete_in_level<'t>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        level: u8,
+        key: &[u8],
+        value: u32,
+    ) -> Result<DeletionResult> {
+        if level == 0 {
+            return self.delete_in_level_0(txn, field_id, key, value);
+        }
+        let (deletion_key, mut bitmap) =
+            self.find_insertion_key_value(field_id, level, key, txn)?;
+
+        let result = self.delete_in_level(txn, field_id, level - 1, key.clone(), value)?;
+
+        let mut decrease_size = false;
+        let (prev_key, next_key) = match result {
+            DeletionResult::InPlace => {
+                bitmap.bitmap.remove(value);
+                self.db.put(txn, &deletion_key.as_ref(), &bitmap)?;
+                return Ok(DeletionResult::InPlace);
+            }
+            DeletionResult::Reduce { prev, next } => (prev, next),
+            DeletionResult::Remove { prev, next } => {
+                decrease_size = true;
+                (prev, next)
+            }
+        };
+
+        let mut updated_value = bitmap;
+        if decrease_size {
+            updated_value.size -= 1;
+        }
+
+        if updated_value.size == 0 {
+            self.db.delete(txn, &deletion_key.as_ref())?;
+            Ok(DeletionResult::Remove { prev: prev_key, next: next_key })
+        } else {
+            let mut updated_deletion_key = deletion_key.clone();
+            if key == deletion_key.left_bound {
+                updated_deletion_key.left_bound = next_key.clone().unwrap();
+            }
+            updated_value.bitmap.remove(value);
+            let _ = self.db.delete(txn, &deletion_key.as_ref())?;
+            self.db.put(txn, &updated_deletion_key.as_ref(), &updated_value)?;
+
+            Ok(DeletionResult::Reduce { prev: prev_key, next: next_key })
+        }
+    }
+
+    fn delete_in_level_0<'t>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        key: &[u8],
+        value: u32,
+    ) -> Result<DeletionResult> {
+        let key = FacetKey { field_id, level: 0, left_bound: key };
+        let mut bitmap = self.db.get(&txn, &key)?.unwrap().bitmap;
+        bitmap.remove(value);
+
+        if bitmap.is_empty() {
+            let mut prev_key = None;
+            let mut next_key = None;
+
+            if let Some(prev) = self.db.get_lower_than(&txn, &key)? {
+                prev_key = Some(prev.0.left_bound.to_vec());
+            }
+            if let Some(next) = self.db.get_greater_than(&txn, &key)? {
+                if next.0.level == 0 {
+                    next_key = Some(next.0.left_bound.to_vec());
+                }
+            }
+            self.db.delete(txn, &key)?;
+            Ok(DeletionResult::Remove { prev: prev_key, next: next_key })
+        } else {
+            self.db.put(txn, &key, &FacetGroupValue { size: 1, bitmap })?;
+            Ok(DeletionResult::InPlace)
+        }
+    }
+
+    pub fn delete<'a, 't>(
+        &self,
+        txn: &'t mut RwTxn,
+        field_id: u16,
+        key: &[u8],
+        value: u32,
+    ) -> Result<()> {
+        if self.db.get(txn, &FacetKey { field_id, level: 0, left_bound: key })?.is_none() {
+            return Ok(());
+        }
+        let highest_level = get_highest_level(&txn, *self.db, field_id)?;
+
+        // let key_bytes = BoundCodec::bytes_encode(&key).unwrap();
+
+        let result = self.delete_in_level(txn, field_id, highest_level as u8, key, value)?;
+        match result {
+            DeletionResult::InPlace => return Ok(()),
+            DeletionResult::Reduce { .. } => {}
+            DeletionResult::Remove { .. } => {}
+        }
+        let mut highest_level_prefix = vec![];
+        highest_level_prefix.extend_from_slice(&field_id.to_be_bytes());
+        highest_level_prefix.push(highest_level);
+
+        if highest_level == 0
+            || self
+                .db
+                .as_polymorph()
+                .prefix_iter::<_, ByteSlice, ByteSlice>(&txn, &highest_level_prefix)?
+                .count()
+                >= self.group_size
+        {
+            return Ok(());
+        }
+        let mut to_delete = vec![];
+        let mut iter = self
+            .db
+            .as_polymorph()
+            .prefix_iter::<_, ByteSlice, ByteSlice>(txn, &highest_level_prefix)?;
+        while let Some(el) = iter.next() {
+            let (k, _) = el?;
+            to_delete.push(
+                FacetKeyCodec::<MyByteSlice>::bytes_decode(k).ok_or(Error::Encoding)?.into_owned(),
+            );
+        }
+        drop(iter);
+        for k in to_delete {
+            self.db.delete(txn, &k.as_ref())?;
+        }
+        Ok(())
+    }
+}

milli/src/update/facet/mod.rs

@@ -0,0 +1,2 @@
+pub mod bulk;
+pub mod incremental;