Merge #3670

3670: Fix addition deletion bug r=irevoire a=irevoire

The first commit of this PR is a revert of https://github.com/meilisearch/meilisearch/pull/3667. It re-enable the auto-batching of addition and deletion of tasks. No new changes have been introduced outside of `milli`. So all the changes you see on the autobatcher have actually already been reviewed.

It fixes https://github.com/meilisearch/meilisearch/issues/3440.

### What was happening?

The issue was that the `external_documents_ids` generated in the `transform` were used in a very strange way that wasn’t compatible with the deletion of documents.
Instead of doing a clear merge between the external document IDs of the DB and the one returned by the transform + writing it on disk, we were doing some weird tricks with the soft-deleted to avoid writing the fst on disk as much as possible.
The new algorithm may be a bit slower but is way more straightforward and doesn’t change depending on if the soft deletion was used or not. Here is a list of the changes introduced:
1. We now do a clear distinction between the `new_external_documents_ids` coming from the transform and only held on RAM and the `external_documents_ids` coming from the DB.
2. The `new_external_documents_ids` (coming out of the transform) are now represented as an `fst`. We don't need to struggle with the hard, soft distinction + the soft_deleted => That's easier to understand
3. When indexing documents, we merge the `external_documents_ids` coming from the DB and the `new_external_documents_ids` coming from the transform.

### Other things introduced in this  PR

Since we constantly have to write small, very specialized fuzzers for this kind of bug, we decided to push the one used to reproduce this bug.
It's not perfect, but it's easy to improve in the future.
It'll also run for as long as possible on every merge on the main branch.

Co-authored-by: Tamo <tamo@meilisearch.com>
Co-authored-by: Loïc Lecrenier <loic.lecrenier@icloud.com>

milli/src/documents/mod.rs

@@ -111,7 +111,6 @@ pub enum Error {
     Io(#[from] io::Error),
 }
 
-#[cfg(test)]
 pub fn objects_from_json_value(json: serde_json::Value) -> Vec<crate::Object> {
     let documents = match json {
         object @ serde_json::Value::Object(_) => vec![object],
@@ -141,7 +140,6 @@ macro_rules! documents {
     }};
 }
 
-#[cfg(test)]
 pub fn documents_batch_reader_from_objects(
     objects: impl IntoIterator<Item = Object>,
 ) -> DocumentsBatchReader<std::io::Cursor<Vec<u8>>> {

milli/src/external_documents_ids.rs

@@ -106,22 +106,30 @@ impl<'a> ExternalDocumentsIds<'a> {
         map
     }
 
+    /// Return an fst of the combined hard and soft deleted ID.
+    pub fn to_fst<'b>(&'b self) -> fst::Result<Cow<'b, fst::Map<Cow<'a, [u8]>>>> {
+        if self.soft.is_empty() {
+            return Ok(Cow::Borrowed(&self.hard));
+        }
+        let union_op = self.hard.op().add(&self.soft).r#union();
+
+        let mut iter = union_op.into_stream();
+        let mut new_hard_builder = fst::MapBuilder::memory();
+        while let Some((external_id, marked_docids)) = iter.next() {
+            let value = indexed_last_value(marked_docids).unwrap();
+            if value != DELETED_ID {
+                new_hard_builder.insert(external_id, value)?;
+            }
+        }
+
+        drop(iter);
+
+        Ok(Cow::Owned(new_hard_builder.into_map().map_data(Cow::Owned)?))
+    }
+
     fn merge_soft_into_hard(&mut self) -> fst::Result<()> {
         if self.soft.len() >= self.hard.len() / 2 {
-            let union_op = self.hard.op().add(&self.soft).r#union();
-
-            let mut iter = union_op.into_stream();
-            let mut new_hard_builder = fst::MapBuilder::memory();
-            while let Some((external_id, marked_docids)) = iter.next() {
-                let value = indexed_last_value(marked_docids).unwrap();
-                if value != DELETED_ID {
-                    new_hard_builder.insert(external_id, value)?;
-                }
-            }
-
-            drop(iter);
-
-            self.hard = new_hard_builder.into_map().map_data(Cow::Owned)?;
+            self.hard = self.to_fst()?.into_owned();
             self.soft = fst::Map::default().map_data(Cow::Owned)?;
         }
 

milli/src/update/delete_documents.rs

@@ -71,7 +71,6 @@ impl std::fmt::Display for DeletionStrategy {
 pub(crate) struct DetailedDocumentDeletionResult {
     pub deleted_documents: u64,
     pub remaining_documents: u64,
-    pub soft_deletion_used: bool,
 }
 
 impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
@@ -108,11 +107,8 @@ impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
         Some(docid)
     }
     pub fn execute(self) -> Result<DocumentDeletionResult> {
-        let DetailedDocumentDeletionResult {
-            deleted_documents,
-            remaining_documents,
-            soft_deletion_used: _,
-        } = self.execute_inner()?;
+        let DetailedDocumentDeletionResult { deleted_documents, remaining_documents } =
+            self.execute_inner()?;
 
         Ok(DocumentDeletionResult { deleted_documents, remaining_documents })
     }
@@ -133,7 +129,6 @@ impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
             return Ok(DetailedDocumentDeletionResult {
                 deleted_documents: 0,
                 remaining_documents: 0,
-                soft_deletion_used: false,
             });
         }
 
@@ -149,7 +144,6 @@ impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
             return Ok(DetailedDocumentDeletionResult {
                 deleted_documents: current_documents_ids_len,
                 remaining_documents,
-                soft_deletion_used: false,
             });
         }
 
@@ -218,7 +212,6 @@ impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
             return Ok(DetailedDocumentDeletionResult {
                 deleted_documents: self.to_delete_docids.len(),
                 remaining_documents: documents_ids.len(),
-                soft_deletion_used: true,
             });
         }
 
@@ -441,7 +434,6 @@ impl<'t, 'u, 'i> DeleteDocuments<'t, 'u, 'i> {
         Ok(DetailedDocumentDeletionResult {
             deleted_documents: self.to_delete_docids.len(),
             remaining_documents: documents_ids.len(),
-            soft_deletion_used: false,
         })
     }
 

milli/src/update/index_documents/mod.rs

@@ -236,7 +236,7 @@ where
             primary_key,
             fields_ids_map,
             field_distribution,
-            mut external_documents_ids,
+            new_external_documents_ids,
             new_documents_ids,
             replaced_documents_ids,
             documents_count,
@@ -363,9 +363,6 @@ where
             deletion_builder.delete_documents(&replaced_documents_ids);
             let deleted_documents_result = deletion_builder.execute_inner()?;
             debug!("{} documents actually deleted", deleted_documents_result.deleted_documents);
-            if !deleted_documents_result.soft_deletion_used {
-                external_documents_ids.delete_soft_deleted_documents_ids_from_fsts()?;
-            }
         }
 
         let index_documents_ids = self.index.documents_ids(self.wtxn)?;
@@ -445,6 +442,9 @@ where
         self.index.put_primary_key(self.wtxn, &primary_key)?;
 
         // We write the external documents ids into the main database.
+        let mut external_documents_ids = self.index.external_documents_ids(self.wtxn)?;
+        external_documents_ids.insert_ids(&new_external_documents_ids)?;
+        let external_documents_ids = external_documents_ids.into_static();
         self.index.put_external_documents_ids(self.wtxn, &external_documents_ids)?;
 
         let all_documents_ids = index_documents_ids | new_documents_ids;
@@ -2514,4 +2514,170 @@ mod tests {
         db_snap!(index, word_fid_docids, 3, @"4c2e2a1832e5802796edc1638136d933");
         db_snap!(index, word_position_docids, 3, @"74f556b91d161d997a89468b4da1cb8f");
     }
+
+    #[test]
+    fn reproduce_the_bug() {
+        /*
+            [milli/examples/fuzz.rs:69] &batches = [
+            Batch(
+                [
+                    AddDoc(
+                        { "id": 1, "doggo": "bernese" }, => internal 0
+                    ),
+                ],
+            ),
+            Batch(
+                [
+                    DeleteDoc(
+                        1, => delete internal 0
+                    ),
+                    AddDoc(
+                        { "id": 0, "catto": "jorts" }, => internal 1
+                    ),
+                ],
+            ),
+            Batch(
+                [
+                    AddDoc(
+                        { "id": 1, "catto": "jorts" }, => internal 2
+                    ),
+                ],
+            ),
+        ]
+        */
+        let mut index = TempIndex::new();
+        index.index_documents_config.deletion_strategy = DeletionStrategy::AlwaysHard;
+
+        // START OF BATCH
+
+        println!("--- ENTERING BATCH 1");
+
+        let mut wtxn = index.write_txn().unwrap();
+
+        let builder = IndexDocuments::new(
+            &mut wtxn,
+            &index,
+            &index.indexer_config,
+            index.index_documents_config.clone(),
+            |_| (),
+            || false,
+        )
+        .unwrap();
+
+        // OP
+
+        let documents = documents!([
+            { "id": 1, "doggo": "bernese" },
+        ]);
+        let (builder, added) = builder.add_documents(documents).unwrap();
+        insta::assert_display_snapshot!(added.unwrap(), @"1");
+
+        // FINISHING
+        let addition = builder.execute().unwrap();
+        insta::assert_debug_snapshot!(addition, @r###"
+        DocumentAdditionResult {
+            indexed_documents: 1,
+            number_of_documents: 1,
+        }
+        "###);
+        wtxn.commit().unwrap();
+
+        db_snap!(index, documents, @r###"
+        {"id":1,"doggo":"bernese"}
+        "###);
+        db_snap!(index, external_documents_ids, @r###"
+        soft:
+        hard:
+        1                        0
+        "###);
+
+        // A first batch of documents has been inserted
+
+        // BATCH 2
+
+        println!("--- ENTERING BATCH 2");
+
+        let mut wtxn = index.write_txn().unwrap();
+
+        let builder = IndexDocuments::new(
+            &mut wtxn,
+            &index,
+            &index.indexer_config,
+            index.index_documents_config.clone(),
+            |_| (),
+            || false,
+        )
+        .unwrap();
+
+        let (builder, removed) = builder.remove_documents(vec![S("1")]).unwrap();
+        insta::assert_display_snapshot!(removed.unwrap(), @"1");
+
+        let documents = documents!([
+            { "id": 0, "catto": "jorts" },
+        ]);
+        let (builder, added) = builder.add_documents(documents).unwrap();
+        insta::assert_display_snapshot!(added.unwrap(), @"1");
+
+        let addition = builder.execute().unwrap();
+        insta::assert_debug_snapshot!(addition, @r###"
+        DocumentAdditionResult {
+            indexed_documents: 1,
+            number_of_documents: 1,
+        }
+        "###);
+        wtxn.commit().unwrap();
+
+        db_snap!(index, documents, @r###"
+        {"id":0,"catto":"jorts"}
+        "###);
+
+        db_snap!(index, external_documents_ids, @r###"
+        soft:
+        hard:
+        0                        1
+        "###);
+
+        db_snap!(index, soft_deleted_documents_ids, @"[]");
+
+        // BATCH 3
+
+        println!("--- ENTERING BATCH 3");
+
+        let mut wtxn = index.write_txn().unwrap();
+
+        let builder = IndexDocuments::new(
+            &mut wtxn,
+            &index,
+            &index.indexer_config,
+            index.index_documents_config.clone(),
+            |_| (),
+            || false,
+        )
+        .unwrap();
+
+        let documents = documents!([
+            { "id": 1, "catto": "jorts" },
+        ]);
+        let (builder, added) = builder.add_documents(documents).unwrap();
+        insta::assert_display_snapshot!(added.unwrap(), @"1");
+
+        let addition = builder.execute().unwrap();
+        insta::assert_debug_snapshot!(addition, @r###"
+        DocumentAdditionResult {
+            indexed_documents: 1,
+            number_of_documents: 2,
+        }
+        "###);
+        wtxn.commit().unwrap();
+
+        db_snap!(index, documents, @r###"
+        {"id":1,"catto":"jorts"}
+        {"id":0,"catto":"jorts"}
+        "###);
+
+        // Ensuring all the returned IDs actually exists
+        let rtxn = index.read_txn().unwrap();
+        let res = index.search(&rtxn).execute().unwrap();
+        index.documents(&rtxn, res.documents_ids).unwrap();
+    }
 }

milli/src/update/index_documents/transform.rs

@@ -21,15 +21,14 @@ use crate::error::{Error, InternalError, UserError};
 use crate::index::{db_name, main_key};
 use crate::update::{AvailableDocumentsIds, ClearDocuments, UpdateIndexingStep};
 use crate::{
-    ExternalDocumentsIds, FieldDistribution, FieldId, FieldIdMapMissingEntry, FieldsIdsMap, Index,
-    Result, BEU32,
+    FieldDistribution, FieldId, FieldIdMapMissingEntry, FieldsIdsMap, Index, Result, BEU32,
 };
 
 pub struct TransformOutput {
     pub primary_key: String,
     pub fields_ids_map: FieldsIdsMap,
     pub field_distribution: FieldDistribution,
-    pub external_documents_ids: ExternalDocumentsIds<'static>,
+    pub new_external_documents_ids: fst::Map<Cow<'static, [u8]>>,
     pub new_documents_ids: RoaringBitmap,
     pub replaced_documents_ids: RoaringBitmap,
     pub documents_count: usize,
@@ -568,8 +567,6 @@ impl<'a, 'i> Transform<'a, 'i> {
             }))?
             .to_string();
 
-        let mut external_documents_ids = self.index.external_documents_ids(wtxn)?;
-
         // We create a final writer to write the new documents in order from the sorter.
         let mut writer = create_writer(
             self.indexer_settings.chunk_compression_type,
@@ -651,13 +648,12 @@ impl<'a, 'i> Transform<'a, 'i> {
             fst_new_external_documents_ids_builder.insert(key, value)
         })?;
         let new_external_documents_ids = fst_new_external_documents_ids_builder.into_map();
-        external_documents_ids.insert_ids(&new_external_documents_ids)?;
 
         Ok(TransformOutput {
             primary_key,
             fields_ids_map: self.fields_ids_map,
             field_distribution,
-            external_documents_ids: external_documents_ids.into_static(),
+            new_external_documents_ids: new_external_documents_ids.map_data(Cow::Owned).unwrap(),
             new_documents_ids: self.new_documents_ids,
             replaced_documents_ids: self.replaced_documents_ids,
             documents_count: self.documents_count,
@@ -691,7 +687,8 @@ impl<'a, 'i> Transform<'a, 'i> {
         let new_external_documents_ids = {
             let mut external_documents_ids = self.index.external_documents_ids(wtxn)?;
             external_documents_ids.delete_soft_deleted_documents_ids_from_fsts()?;
-            external_documents_ids
+            // This call should be free and can't fail since the previous method merged both fsts.
+            external_documents_ids.into_static().to_fst()?.into_owned()
         };
 
         let documents_ids = self.index.documents_ids(wtxn)?;
@@ -776,7 +773,7 @@ impl<'a, 'i> Transform<'a, 'i> {
             primary_key,
             fields_ids_map: new_fields_ids_map,
             field_distribution,
-            external_documents_ids: new_external_documents_ids.into_static(),
+            new_external_documents_ids,
             new_documents_ids: documents_ids,
             replaced_documents_ids: RoaringBitmap::default(),
             documents_count,