Merge branch 'main' into cellulite

crates/milli/Cargo.toml

@@ -115,6 +115,7 @@ utoipa = { version = "5.4.0", features = [
 lru = "0.14.0"
 geo-types = "0.7.16"
 zerometry = "0.1.0"
+twox-hash = { version = "2.1.1", default-features = false, features = ["std", "xxhash3_64", "xxhash64"] }
 
 [dev-dependencies]
 mimalloc = { version = "0.1.47", default-features = false }

crates/milli/src/index.rs

@@ -1471,7 +1471,7 @@ impl Index {
             .0)
     }
 
-    pub(crate) fn set_updated_at(
+    pub fn set_updated_at(
         &self,
         wtxn: &mut RwTxn<'_>,
         time: &time::OffsetDateTime,

crates/milli/src/search/new/tests/integration.rs

@@ -76,6 +76,7 @@ pub fn setup_search_index_with_criteria(criteria: &[Criterion]) -> Index {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();
 

crates/milli/src/test_index.rs

@@ -84,6 +84,7 @@ impl TempIndex {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )?;
 
         if let Some(error) = operation_stats.into_iter().find_map(|stat| stat.error) {
@@ -167,6 +168,7 @@ impl TempIndex {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )?;
 
         if let Some(error) = operation_stats.into_iter().find_map(|stat| stat.error) {
@@ -242,6 +244,7 @@ fn aborting_indexation() {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();
 

crates/milli/src/update/index_documents/mod.rs

@@ -1979,6 +1979,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2031,6 +2032,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2119,6 +2121,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2308,6 +2311,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2371,6 +2375,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2425,6 +2430,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2478,6 +2484,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2533,6 +2540,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2593,6 +2601,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2646,6 +2655,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2699,6 +2709,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2910,6 +2921,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -2970,6 +2982,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 
@@ -3027,6 +3040,7 @@ mod tests {
                 &mut new_fields_ids_map,
                 &|| false,
                 Progress::default(),
+                None,
             )
             .unwrap();
 

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

@@ -17,6 +17,7 @@ use super::guess_primary_key::retrieve_or_guess_primary_key;
 use crate::documents::PrimaryKey;
 use crate::progress::{AtomicPayloadStep, Progress};
 use crate::update::new::document::{DocumentContext, Versions};
+use crate::update::new::indexer::enterprise_edition::sharding::Shards;
 use crate::update::new::steps::IndexingStep;
 use crate::update::new::thread_local::MostlySend;
 use crate::update::new::{DocumentIdentifiers, Insertion, Update};
@@ -71,6 +72,7 @@ impl<'pl> DocumentOperation<'pl> {
         new_fields_ids_map: &mut FieldsIdsMap,
         must_stop_processing: &MSP,
         progress: Progress,
+        shards: Option<&Shards>,
     ) -> Result<(DocumentOperationChanges<'pl>, Vec<PayloadStats>, Option<PrimaryKey<'pl>>)>
     where
         MSP: Fn() -> bool,
@@ -107,6 +109,7 @@ impl<'pl> DocumentOperation<'pl> {
                     &mut bytes,
                     &docids_version_offsets,
                     IndexDocumentsMethod::ReplaceDocuments,
+                    shards,
                     payload,
                 ),
                 Payload::Update(payload) => extract_addition_payload_changes(
@@ -120,6 +123,7 @@ impl<'pl> DocumentOperation<'pl> {
                     &mut bytes,
                     &docids_version_offsets,
                     IndexDocumentsMethod::UpdateDocuments,
+                    shards,
                     payload,
                 ),
                 Payload::Deletion(to_delete) => extract_deletion_payload_changes(
@@ -127,6 +131,7 @@ impl<'pl> DocumentOperation<'pl> {
                     rtxn,
                     &mut available_docids,
                     &docids_version_offsets,
+                    shards,
                     to_delete,
                 ),
             };
@@ -173,6 +178,7 @@ fn extract_addition_payload_changes<'r, 'pl: 'r>(
     bytes: &mut u64,
     main_docids_version_offsets: &hashbrown::HashMap<&'pl str, PayloadOperations<'pl>>,
     method: IndexDocumentsMethod,
+    shards: Option<&Shards>,
     payload: &'pl [u8],
 ) -> Result<hashbrown::HashMap<&'pl str, PayloadOperations<'pl>>> {
     use IndexDocumentsMethod::{ReplaceDocuments, UpdateDocuments};
@@ -210,12 +216,20 @@ fn extract_addition_payload_changes<'r, 'pl: 'r>(
             primary_key.as_ref().unwrap()
         };
 
+        let current_offset = iter.byte_offset();
+        let content = &payload[previous_offset..current_offset];
+        previous_offset = current_offset;
+
         let external_id =
             retrieved_primary_key.extract_fields_and_docid(doc, new_fields_ids_map, indexer)?;
 
         let external_id = external_id.to_de();
-        let current_offset = iter.byte_offset();
-        let document_offset = DocumentOffset { content: &payload[previous_offset..current_offset] };
+
+        if shards.is_some_and(|shards| !shards.must_process(external_id)) {
+            continue;
+        }
+
+        let document_offset = DocumentOffset { content };
 
         match main_docids_version_offsets.get(external_id) {
             None => {
@@ -299,8 +313,6 @@ fn extract_addition_payload_changes<'r, 'pl: 'r>(
                 },
             },
         }
-
-        previous_offset = iter.byte_offset();
     }
 
     if payload.is_empty() {
@@ -329,11 +341,16 @@ fn extract_deletion_payload_changes<'s, 'pl: 's>(
     rtxn: &RoTxn,
     available_docids: &mut AvailableIds,
     main_docids_version_offsets: &hashbrown::HashMap<&'s str, PayloadOperations<'pl>>,
+    shards: Option<&Shards>,
     to_delete: &'pl [&'pl str],
 ) -> Result<hashbrown::HashMap<&'s str, PayloadOperations<'pl>>> {
     let mut new_docids_version_offsets = hashbrown::HashMap::<&str, PayloadOperations<'pl>>::new();
 
     for external_id in to_delete {
+        if shards.is_some_and(|shards| !shards.must_process(external_id)) {
+            continue;
+        }
+
         match main_docids_version_offsets.get(external_id) {
             None => {
                 match index.external_documents_ids().get(rtxn, external_id) {

crates/milli/src/update/new/indexer/enterprise_edition/mod.rs

@@ -0,0 +1,6 @@
+// Copyright © 2025 Meilisearch Some Rights Reserved
+// This file is part of Meilisearch Enterprise Edition (EE).
+// Use of this source code is governed by the Business Source License 1.1,
+// as found in the LICENSE-EE file or at <https://mariadb.com/bsl11>
+
+pub mod sharding;

crates/milli/src/update/new/indexer/enterprise_edition/sharding.rs

@@ -0,0 +1,22 @@
+// Copyright © 2025 Meilisearch Some Rights Reserved
+// This file is part of Meilisearch Enterprise Edition (EE).
+// Use of this source code is governed by the Business Source License 1.1,
+// as found in the LICENSE-EE file or at <https://mariadb.com/bsl11>
+
+use std::hash::{BuildHasher as _, BuildHasherDefault};
+
+pub struct Shards {
+    pub own: Vec<String>,
+    pub others: Vec<String>,
+}
+
+impl Shards {
+    pub fn must_process(&self, docid: &str) -> bool {
+        let hasher = BuildHasherDefault::<twox_hash::XxHash3_64>::new();
+        let to_hash = |shard: &String| hasher.hash_one((shard, docid));
+
+        let max_hash = self.others.iter().map(to_hash).max().unwrap_or_default();
+
+        self.own.iter().map(to_hash).any(|hash| hash > max_hash)
+    }
+}

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

@@ -31,6 +31,7 @@ pub(crate) mod de;
 pub mod document_changes;
 mod document_deletion;
 mod document_operation;
+pub mod enterprise_edition;
 mod extract;
 mod guess_primary_key;
 mod partial_dump;

crates/milli/src/update/upgrade/mod.rs

@@ -8,7 +8,6 @@ use v1_12::{V1_12_3_To_V1_13_0, V1_12_To_V1_12_3};
 use v1_13::{V1_13_0_To_V1_13_1, V1_13_1_To_Latest_V1_13};
 use v1_14::Latest_V1_13_To_Latest_V1_14;
 use v1_15::Latest_V1_14_To_Latest_V1_15;
-use v1_16::Latest_V1_16_To_V1_17_0;
 
 use crate::constants::{VERSION_MAJOR, VERSION_MINOR, VERSION_PATCH};
 use crate::progress::{Progress, VariableNameStep};
@@ -35,7 +34,6 @@ const UPGRADE_FUNCTIONS: &[&dyn UpgradeIndex] = &[
     &Latest_V1_13_To_Latest_V1_14 {},
     &Latest_V1_14_To_Latest_V1_15 {},
     &Latest_V1_15_To_V1_16_0 {},
-    &Latest_V1_16_To_V1_17_0 {},
     // This is the last upgrade function, it will be called when the index is up to date.
     // any other upgrade function should be added before this one.
     &ToCurrentNoOp {},
@@ -64,7 +62,8 @@ const fn start(from: (u32, u32, u32)) -> Option<usize> {
         // We must handle the current version in the match because in case of a failure some index may have been upgraded but not other.
         (1, 15, _) => function_index!(6),
         (1, 16, _) => function_index!(7),
-        (1, 17, _) => function_index!(8),
+        (1, 17, _) => function_index!(7),
+        (1, 18, _) => function_index!(7),
         // We deliberately don't add a placeholder with (VERSION_MAJOR, VERSION_MINOR, VERSION_PATCH) here to force manually
         // considering dumpless upgrade.
         (_major, _minor, _patch) => return None,

crates/milli/src/update/upgrade/v1_16.rs

@@ -46,22 +46,3 @@ impl UpgradeIndex for Latest_V1_15_To_V1_16_0 {
         (1, 16, 0)
     }
 }
-
-#[allow(non_camel_case_types)]
-pub(super) struct Latest_V1_16_To_V1_17_0();
-
-impl UpgradeIndex for Latest_V1_16_To_V1_17_0 {
-    fn upgrade(
-        &self,
-        _wtxn: &mut RwTxn,
-        _index: &Index,
-        _original: (u32, u32, u32),
-        _progress: Progress,
-    ) -> Result<bool> {
-        Ok(false)
-    }
-
-    fn target_version(&self) -> (u32, u32, u32) {
-        (1, 17, 0)
-    }
-}

crates/milli/tests/search/facet_distribution.rs

@@ -59,6 +59,7 @@ fn test_facet_distribution_with_no_facet_values() {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();
 

crates/milli/tests/search/mod.rs

@@ -97,6 +97,7 @@ pub fn setup_search_index_with_criteria(criteria: &[Criterion]) -> Index {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();
 

crates/milli/tests/search/query_criteria.rs

@@ -329,6 +329,7 @@ fn criteria_ascdesc() {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();
 

crates/milli/tests/search/typo_tolerance.rs

@@ -138,6 +138,7 @@ fn test_typo_disabled_on_word() {
             &mut new_fields_ids_map,
             &|| false,
             Progress::default(),
+            None,
         )
         .unwrap();