Introduce papaya

Cargo.lock

@@ -3798,6 +3798,7 @@ dependencies = [
  "obkv",
  "once_cell",
  "ordered-float",
+ "papaya",
  "rand",
  "rayon",
  "rayon-par-bridge",
@@ -4189,6 +4190,16 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "papaya"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6827e3fc394523c21d4464d02c0bb1c19966ea4a58a9844ad6d746214179d2bc"
+dependencies = [
+ "equivalent",
+ "seize",
+]
+
 [[package]]
 name = "parking_lot"
 version = "0.12.3"
@@ -5104,6 +5115,16 @@ dependencies = [
  "time",
 ]
 
+[[package]]
+name = "seize"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e4b8d813387d566f627f3ea1b914c068aac94c40ae27ec43f5f33bde65abefe7"
+dependencies = [
+ "libc",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "semver"
 version = "1.0.18"

crates/milli/Cargo.toml

@@ -111,6 +111,7 @@ utoipa = { version = "5.3.1", features = [
     "openapi_extensions",
 ] }
 lru = "0.13.0"
+papaya = "0.2.1"
 
 [dev-dependencies]
 mimalloc = { version = "0.1.43", default-features = false }

crates/milli/src/update/new/indexer/document_operation.rs

@@ -6,6 +6,7 @@ use bumparaw_collections::RawMap;
 use hashbrown::hash_map::Entry;
 use heed::RoTxn;
 use memmap2::Mmap;
+use rayon::iter::{IndexedParallelIterator as _, IntoParallelIterator as _, ParallelIterator as _};
 use rayon::slice::ParallelSlice;
 use rustc_hash::FxBuildHasher;
 use serde_json::value::RawValue;
@@ -88,18 +89,23 @@ impl<'pl> DocumentOperation<'pl> {
         let (step, progress_step) = AtomicPayloadStep::new(payload_count as u32);
         progress.update_progress(progress_step);
 
-        for (payload_index, operation) in operations.into_iter().enumerate() {
-            if must_stop_processing() {
-                return Err(InternalError::AbortedIndexation.into());
-            }
-            step.store(payload_index as u32, Ordering::Relaxed);
+        let operations_stats = operations
+            .into_par_iter()
+            .enumerate()
+            .map(|(payload_index, operation)| {
+                // if must_stop_processing() {
+                //     return Err(InternalError::AbortedIndexation.into());
+                // }
 
                 let mut bytes = 0;
+                let rtxn = index.read_txn().unwrap();
+                step.fetch_add(1, Ordering::Relaxed);
+
                 let result = match operation {
                     Payload::Replace(payload) => extract_addition_payload_changes(
                         indexer,
                         index,
-                    rtxn,
+                        &rtxn,
                         primary_key_from_op,
                         &mut primary_key,
                         new_fields_ids_map,
@@ -112,7 +118,7 @@ impl<'pl> DocumentOperation<'pl> {
                     Payload::Update(payload) => extract_addition_payload_changes(
                         indexer,
                         index,
-                    rtxn,
+                        &rtxn,
                         primary_key_from_op,
                         &mut primary_key,
                         new_fields_ids_map,
@@ -124,7 +130,7 @@ impl<'pl> DocumentOperation<'pl> {
                     ),
                     Payload::Deletion(to_delete) => extract_deletion_payload_changes(
                         index,
-                    rtxn,
+                        &rtxn,
                         &mut available_docids,
                         &docids_version_offsets,
                         to_delete,
@@ -132,20 +138,24 @@ impl<'pl> DocumentOperation<'pl> {
                 };
 
                 let mut document_count = 0;
-            let error = match result {
+                match result {
                     Ok(new_docids_version_offsets) => {
                         document_count = new_docids_version_offsets.len() as u64;
                         // If we don't have any error then we can merge the content of this payload
                         // into to main payload. Else we just drop this payload extraction.
-                    merge_version_offsets(&mut docids_version_offsets, new_docids_version_offsets);
-                    None
+                        merge_version_offsets(
+                            &mut docids_version_offsets,
+                            new_docids_version_offsets,
+                        );
+                        Ok(PayloadStats { document_count, bytes, error: None })
                     }
-                Err(Error::UserError(user_error)) => Some(user_error),
-                Err(e) => return Err(e),
-            };
-            operations_stats.push(PayloadStats { document_count, bytes, error });
+                    Err(Error::UserError(user_error)) => {
+                        Ok(PayloadStats { document_count, bytes, error: Some(user_error) })
                     }
-        step.store(payload_count as u32, Ordering::Relaxed);
+                    Err(e) => Err(e),
+                }
+            })
+            .collect::<crate::Result<Vec<_>>>()?;
 
         // TODO We must drain the HashMap into a Vec because rayon::hash_map::IntoIter: !Clone
         let mut docids_version_offsets: bumpalo::collections::vec::Vec<_> =