Prepare for processing documents edition

index-scheduler/src/autobatcher.rs

@@ -24,6 +24,7 @@ enum AutobatchKind {
         allow_index_creation: bool,
         primary_key: Option<String>,
     },
+    DocumentEdition,
     DocumentDeletion,
     DocumentDeletionByFilter,
     DocumentClear,
@@ -63,6 +64,7 @@ impl From<KindWithContent> for AutobatchKind {
                 primary_key,
                 ..
             } => AutobatchKind::DocumentImport { method, allow_index_creation, primary_key },
+            KindWithContent::DocumentEdition { .. } => AutobatchKind::DocumentEdition,
             KindWithContent::DocumentDeletion { .. } => AutobatchKind::DocumentDeletion,
             KindWithContent::DocumentClear { .. } => AutobatchKind::DocumentClear,
             KindWithContent::DocumentDeletionByFilter { .. } => {
@@ -98,6 +100,9 @@ pub enum BatchKind {
         primary_key: Option<String>,
         operation_ids: Vec<TaskId>,
     },
+    DocumentEdition {
+        id: TaskId,
+    },
     DocumentDeletion {
         deletion_ids: Vec<TaskId>,
     },
@@ -199,6 +204,7 @@ impl BatchKind {
                 }),
                 allow_index_creation,
             ),
+            K::DocumentEdition => (Break(BatchKind::DocumentEdition { id: task_id }), false),
             K::DocumentDeletion => {
                 (Continue(BatchKind::DocumentDeletion { deletion_ids: vec![task_id] }), false)
             }
@@ -222,7 +228,7 @@ impl BatchKind {
 
         match (self, kind) {
             // We don't batch any of these operations
-            (this, K::IndexCreation | K::IndexUpdate | K::IndexSwap | K::DocumentDeletionByFilter) => Break(this),
+            (this, K::IndexCreation | K::IndexUpdate | K::IndexSwap | K::DocumentEdition | K::DocumentDeletionByFilter) => Break(this),
             // We must not batch tasks that don't have the same index creation rights if the index doesn't already exists.
             (this, kind) if !index_already_exists && this.allow_index_creation() == Some(false) && kind.allow_index_creation() == Some(true) => {
                 Break(this)
@@ -519,6 +525,7 @@ impl BatchKind {
                 | BatchKind::IndexDeletion { .. }
                 | BatchKind::IndexUpdate { .. }
                 | BatchKind::IndexSwap { .. }
+                | BatchKind::DocumentEdition { .. }
                 | BatchKind::DocumentDeletionByFilter { .. },
                 _,
             ) => {

index-scheduler/src/batch.rs

@@ -106,6 +106,10 @@ pub(crate) enum IndexOperation {
         operations: Vec<DocumentOperation>,
         tasks: Vec<Task>,
     },
+    DocumentEdition {
+        index_uid: String,
+        task: Task,
+    },
     IndexDocumentDeletionByFilter {
         index_uid: String,
         task: Task,
@@ -164,7 +168,8 @@ impl Batch {
                 | IndexOperation::DocumentClear { tasks, .. } => {
                     RoaringBitmap::from_iter(tasks.iter().map(|task| task.uid))
                 }
-                IndexOperation::IndexDocumentDeletionByFilter { task, .. } => {
+                IndexOperation::DocumentEdition { task, .. }
+                | IndexOperation::IndexDocumentDeletionByFilter { task, .. } => {
                     RoaringBitmap::from_sorted_iter(std::iter::once(task.uid)).unwrap()
                 }
                 IndexOperation::SettingsAndDocumentOperation {
@@ -228,6 +233,7 @@ impl IndexOperation {
     pub fn index_uid(&self) -> &str {
         match self {
             IndexOperation::DocumentOperation { index_uid, .. }
+            | IndexOperation::DocumentEdition { index_uid, .. }
             | IndexOperation::IndexDocumentDeletionByFilter { index_uid, .. }
             | IndexOperation::DocumentClear { index_uid, .. }
             | IndexOperation::Settings { index_uid, .. }
@@ -243,6 +249,9 @@ impl fmt::Display for IndexOperation {
             IndexOperation::DocumentOperation { .. } => {
                 f.write_str("IndexOperation::DocumentOperation")
             }
+            IndexOperation::DocumentEdition { .. } => {
+                f.write_str("IndexOperation::DocumentEdition")
+            }
             IndexOperation::IndexDocumentDeletionByFilter { .. } => {
                 f.write_str("IndexOperation::IndexDocumentDeletionByFilter")
             }
@@ -295,6 +304,21 @@ impl IndexScheduler {
                     _ => unreachable!(),
                 }
             }
+            BatchKind::DocumentEdition { id } => {
+                let task = self.get_task(rtxn, id)?.ok_or(Error::CorruptedTaskQueue)?;
+                match &task.kind {
+                    KindWithContent::DocumentEdition { index_uid, .. } => {
+                        Ok(Some(Batch::IndexOperation {
+                            op: IndexOperation::DocumentEdition {
+                                index_uid: index_uid.clone(),
+                                task,
+                            },
+                            must_create_index: false,
+                        }))
+                    }
+                    _ => unreachable!(),
+                }
+            }
             BatchKind::DocumentOperation { method, operation_ids, .. } => {
                 let tasks = self.get_existing_tasks(rtxn, operation_ids)?;
                 let primary_key = tasks
@@ -1386,6 +1410,9 @@ impl IndexScheduler {
 
                 Ok(tasks)
             }
+            IndexOperation::DocumentEdition { .. } => {
+                todo!()
+            }
             IndexOperation::IndexDocumentDeletionByFilter { mut task, index_uid: _ } => {
                 let filter =
                     if let KindWithContent::DocumentDeletionByFilter { filter_expr, .. } =

index-scheduler/src/insta_snapshot.rs

@@ -177,6 +177,12 @@ fn snapshot_details(d: &Details) -> String {
         } => {
             format!("{{ received_documents: {received_documents}, indexed_documents: {indexed_documents:?} }}")
         }
+        Details::DocumentEdition {
+            edited_documents,
+            edition_code,
+        } => {
+            format!("{{ edited_documents: {edited_documents:?}, edition_code: {edition_code:?} }}")
+        }
         Details::SettingsUpdate { settings } => {
             format!("{{ settings: {settings:?} }}")
         }

index-scheduler/src/utils.rs

@@ -238,6 +238,7 @@ pub fn swap_index_uid_in_task(task: &mut Task, swap: (&str, &str)) {
     let mut index_uids = vec![];
     match &mut task.kind {
         K::DocumentAdditionOrUpdate { index_uid, .. } => index_uids.push(index_uid),
+        K::DocumentEdition { index_uid, .. } => index_uids.push(index_uid),
         K::DocumentDeletion { index_uid, .. } => index_uids.push(index_uid),
         K::DocumentDeletionByFilter { index_uid, .. } => index_uids.push(index_uid),
         K::DocumentClear { index_uid } => index_uids.push(index_uid),
@@ -408,7 +409,26 @@ impl IndexScheduler {
                                 match status {
                                     Status::Succeeded => assert!(indexed_documents <= received_documents),
                                     Status::Failed | Status::Canceled => assert_eq!(indexed_documents, 0),
-                                    status => panic!("DocumentAddition can't have an indexed_document set if it's {}", status),
+                                    status => panic!("DocumentAddition can't have an indexed_documents set if it's {}", status),
+                                }
+                            }
+                            None => {
+                                assert!(matches!(status, Status::Enqueued | Status::Processing))
+                            }
+                        }
+                    }
+                    Details::DocumentEdition { edited_documents, .. } => {
+                        assert_eq!(kind.as_kind(), Kind::DocumentEdition);
+                        match edited_documents {
+                            Some(edited_documents) => {
+                                assert!(matches!(
+                                    status,
+                                    Status::Succeeded | Status::Failed | Status::Canceled
+                                ));
+                                match status {
+                                    Status::Succeeded => (),
+                                    Status::Failed | Status::Canceled => assert_eq!(edited_documents, 0),
+                                    status => panic!("DocumentEdition can't have an edited_documents set if it's {}", status),
                                 }
                             }
                             None => {