Merge #3811

3811: Bring back changes from `release-v1.2.0` to `main` r=Kerollmops a=curquiza



Co-authored-by: Loïc Lecrenier <loic.lecrenier@me.com>
Co-authored-by: meili-bors[bot] <89034592+meili-bors[bot]@users.noreply.github.com>
Co-authored-by: Tamo <tamo@meilisearch.com>
Co-authored-by: Filip Bachul <filipbachul@gmail.com>
Co-authored-by: Kerollmops <clement@meilisearch.com>
Co-authored-by: ManyTheFish <many@meilisearch.com>
Co-authored-by: Clément Renault <clement@meilisearch.com>

index-scheduler/src/batch.rs

@@ -24,6 +24,7 @@ use std::io::BufWriter;
 
 use dump::IndexMetadata;
 use log::{debug, error, info};
+use meilisearch_types::error::Code;
 use meilisearch_types::heed::{RoTxn, RwTxn};
 use meilisearch_types::milli::documents::{obkv_to_object, DocumentsBatchReader};
 use meilisearch_types::milli::heed::CompactionOption;
@@ -1491,7 +1492,12 @@ fn delete_document_by_filter(filter: &serde_json::Value, index: Index) -> Result
     Ok(if let Some(filter) = filter {
         let mut wtxn = index.write_txn()?;
 
-        let candidates = filter.evaluate(&wtxn, &index)?;
+        let candidates = filter.evaluate(&wtxn, &index).map_err(|err| match err {
+            milli::Error::UserError(milli::UserError::InvalidFilter(_)) => {
+                Error::from(err).with_custom_error_code(Code::InvalidDocumentFilter)
+            }
+            e => e.into(),
+        })?;
         let mut delete_operation = DeleteDocuments::new(&mut wtxn, &index)?;
         delete_operation.delete_documents(&candidates);
         let deleted_documents =

index-scheduler/src/error.rs

@@ -46,6 +46,8 @@ impl From<DateField> for Code {
 #[allow(clippy::large_enum_variant)]
 #[derive(Error, Debug)]
 pub enum Error {
+    #[error("{1}")]
+    WithCustomErrorCode(Code, Box<Self>),
     #[error("Index `{0}` not found.")]
     IndexNotFound(String),
     #[error("Index `{0}` already exists.")]
@@ -144,6 +146,7 @@ impl Error {
     pub fn is_recoverable(&self) -> bool {
         match self {
             Error::IndexNotFound(_)
+            | Error::WithCustomErrorCode(_, _)
             | Error::IndexAlreadyExists(_)
             | Error::SwapDuplicateIndexFound(_)
             | Error::SwapDuplicateIndexesFound(_)
@@ -176,11 +179,16 @@ impl Error {
             Error::PlannedFailure => false,
         }
     }
+
+    pub fn with_custom_error_code(self, code: Code) -> Self {
+        Self::WithCustomErrorCode(code, Box::new(self))
+    }
 }
 
 impl ErrorCode for Error {
     fn error_code(&self) -> Code {
         match self {
+            Error::WithCustomErrorCode(code, _) => *code,
             Error::IndexNotFound(_) => Code::IndexNotFound,
             Error::IndexAlreadyExists(_) => Code::IndexAlreadyExists,
             Error::SwapDuplicateIndexesFound(_) => Code::InvalidSwapDuplicateIndexFound,

index-scheduler/src/index_mapper/index_map.rs

@@ -5,6 +5,7 @@ use std::collections::BTreeMap;
 use std::path::Path;
 use std::time::Duration;
 
+use meilisearch_types::heed::flags::Flags;
 use meilisearch_types::heed::{EnvClosingEvent, EnvOpenOptions};
 use meilisearch_types::milli::Index;
 use time::OffsetDateTime;
@@ -53,6 +54,7 @@ pub struct IndexMap {
 pub struct ClosingIndex {
     uuid: Uuid,
     closing_event: EnvClosingEvent,
+    enable_mdb_writemap: bool,
     map_size: usize,
     generation: usize,
 }
@@ -68,6 +70,7 @@ impl ClosingIndex {
     pub fn wait_timeout(self, timeout: Duration) -> Option<ReopenableIndex> {
         self.closing_event.wait_timeout(timeout).then_some(ReopenableIndex {
             uuid: self.uuid,
+            enable_mdb_writemap: self.enable_mdb_writemap,
             map_size: self.map_size,
             generation: self.generation,
         })
@@ -76,6 +79,7 @@ impl ClosingIndex {
 
 pub struct ReopenableIndex {
     uuid: Uuid,
+    enable_mdb_writemap: bool,
     map_size: usize,
     generation: usize,
 }
@@ -103,7 +107,7 @@ impl ReopenableIndex {
                 return Ok(());
             }
             map.unavailable.remove(&self.uuid);
-            map.create(&self.uuid, path, None, self.map_size)?;
+            map.create(&self.uuid, path, None, self.enable_mdb_writemap, self.map_size)?;
         }
         Ok(())
     }
@@ -170,16 +174,17 @@ impl IndexMap {
         uuid: &Uuid,
         path: &Path,
         date: Option<(OffsetDateTime, OffsetDateTime)>,
+        enable_mdb_writemap: bool,
         map_size: usize,
     ) -> Result<Index> {
         if !matches!(self.get_unavailable(uuid), Missing) {
             panic!("Attempt to open an index that was unavailable");
         }
-        let index = create_or_open_index(path, date, map_size)?;
+        let index = create_or_open_index(path, date, enable_mdb_writemap, map_size)?;
         match self.available.insert(*uuid, index.clone()) {
             InsertionOutcome::InsertedNew => (),
             InsertionOutcome::Evicted(evicted_uuid, evicted_index) => {
-                self.close(evicted_uuid, evicted_index, 0);
+                self.close(evicted_uuid, evicted_index, enable_mdb_writemap, 0);
             }
             InsertionOutcome::Replaced(_) => {
                 panic!("Attempt to open an index that was already opened")
@@ -212,17 +217,30 @@ impl IndexMap {
     /// | Closing         | Closing       |
     /// | Available       | Closing       |
     ///
-    pub fn close_for_resize(&mut self, uuid: &Uuid, map_size_growth: usize) {
+    pub fn close_for_resize(
+        &mut self,
+        uuid: &Uuid,
+        enable_mdb_writemap: bool,
+        map_size_growth: usize,
+    ) {
         let Some(index) = self.available.remove(uuid) else { return; };
-        self.close(*uuid, index, map_size_growth);
+        self.close(*uuid, index, enable_mdb_writemap, map_size_growth);
     }
 
-    fn close(&mut self, uuid: Uuid, index: Index, map_size_growth: usize) {
+    fn close(
+        &mut self,
+        uuid: Uuid,
+        index: Index,
+        enable_mdb_writemap: bool,
+        map_size_growth: usize,
+    ) {
         let map_size = index.map_size().unwrap_or(DEFAULT_MAP_SIZE) + map_size_growth;
         let closing_event = index.prepare_for_closing();
         let generation = self.next_generation();
-        self.unavailable
-            .insert(uuid, Some(ClosingIndex { uuid, closing_event, map_size, generation }));
+        self.unavailable.insert(
+            uuid,
+            Some(ClosingIndex { uuid, closing_event, enable_mdb_writemap, map_size, generation }),
+        );
     }
 
     /// Attempts to delete and index.
@@ -282,11 +300,15 @@ impl IndexMap {
 fn create_or_open_index(
     path: &Path,
     date: Option<(OffsetDateTime, OffsetDateTime)>,
+    enable_mdb_writemap: bool,
     map_size: usize,
 ) -> Result<Index> {
     let mut options = EnvOpenOptions::new();
     options.map_size(clamp_to_page_size(map_size));
     options.max_readers(1024);
+    if enable_mdb_writemap {
+        unsafe { options.flag(Flags::MdbWriteMap) };
+    }
 
     if let Some((created, updated)) = date {
         Ok(Index::new_with_creation_dates(options, path, created, updated)?)

index-scheduler/src/index_mapper/mod.rs

@@ -66,6 +66,8 @@ pub struct IndexMapper {
     index_base_map_size: usize,
     /// The quantity by which the map size of an index is incremented upon reopening, in bytes.
     index_growth_amount: usize,
+    /// Whether we open a meilisearch index with the MDB_WRITEMAP option or not.
+    enable_mdb_writemap: bool,
     pub indexer_config: Arc<IndexerConfig>,
 }
 
@@ -132,15 +134,22 @@ impl IndexMapper {
         index_base_map_size: usize,
         index_growth_amount: usize,
         index_count: usize,
+        enable_mdb_writemap: bool,
         indexer_config: IndexerConfig,
     ) -> Result<Self> {
+        let mut wtxn = env.write_txn()?;
+        let index_mapping = env.create_database(&mut wtxn, Some(INDEX_MAPPING))?;
+        let index_stats = env.create_database(&mut wtxn, Some(INDEX_STATS))?;
+        wtxn.commit()?;
+
         Ok(Self {
             index_map: Arc::new(RwLock::new(IndexMap::new(index_count))),
-            index_mapping: env.create_database(Some(INDEX_MAPPING))?,
-            index_stats: env.create_database(Some(INDEX_STATS))?,
+            index_mapping,
+            index_stats,
             base_path,
             index_base_map_size,
             index_growth_amount,
+            enable_mdb_writemap,
             indexer_config: Arc::new(indexer_config),
         })
     }
@@ -171,6 +180,7 @@ impl IndexMapper {
                     &uuid,
                     &index_path,
                     date,
+                    self.enable_mdb_writemap,
                     self.index_base_map_size,
                 )?;
 
@@ -282,7 +292,11 @@ impl IndexMapper {
             .ok_or_else(|| Error::IndexNotFound(name.to_string()))?;
 
         // We remove the index from the in-memory index map.
-        self.index_map.write().unwrap().close_for_resize(&uuid, self.index_growth_amount);
+        self.index_map.write().unwrap().close_for_resize(
+            &uuid,
+            self.enable_mdb_writemap,
+            self.index_growth_amount,
+        );
 
         Ok(())
     }
@@ -347,6 +361,7 @@ impl IndexMapper {
                                 &uuid,
                                 &index_path,
                                 None,
+                                self.enable_mdb_writemap,
                                 self.index_base_map_size,
                             )?;
                         }

index-scheduler/src/lib.rs

@@ -233,6 +233,8 @@ pub struct IndexSchedulerOptions {
     pub task_db_size: usize,
     /// The size, in bytes, with which a meilisearch index is opened the first time of each meilisearch index.
     pub index_base_map_size: usize,
+    /// Whether we open a meilisearch index with the MDB_WRITEMAP option or not.
+    pub enable_mdb_writemap: bool,
     /// The size, in bytes, by which the map size of an index is increased when it resized due to being full.
     pub index_growth_amount: usize,
     /// The number of indexes that can be concurrently opened in memory.
@@ -374,6 +376,11 @@ impl IndexScheduler {
         std::fs::create_dir_all(&options.indexes_path)?;
         std::fs::create_dir_all(&options.dumps_path)?;
 
+        if cfg!(windows) && options.enable_mdb_writemap {
+            // programmer error if this happens: in normal use passing the option on Windows is an error in main
+            panic!("Windows doesn't support the MDB_WRITEMAP LMDB option");
+        }
+
         let task_db_size = clamp_to_page_size(options.task_db_size);
         let budget = if options.indexer_config.skip_index_budget {
             IndexBudget {
@@ -396,25 +403,37 @@ impl IndexScheduler {
             .open(options.tasks_path)?;
         let file_store = FileStore::new(&options.update_file_path)?;
 
+        let mut wtxn = env.write_txn()?;
+        let all_tasks = env.create_database(&mut wtxn, Some(db_name::ALL_TASKS))?;
+        let status = env.create_database(&mut wtxn, Some(db_name::STATUS))?;
+        let kind = env.create_database(&mut wtxn, Some(db_name::KIND))?;
+        let index_tasks = env.create_database(&mut wtxn, Some(db_name::INDEX_TASKS))?;
+        let canceled_by = env.create_database(&mut wtxn, Some(db_name::CANCELED_BY))?;
+        let enqueued_at = env.create_database(&mut wtxn, Some(db_name::ENQUEUED_AT))?;
+        let started_at = env.create_database(&mut wtxn, Some(db_name::STARTED_AT))?;
+        let finished_at = env.create_database(&mut wtxn, Some(db_name::FINISHED_AT))?;
+        wtxn.commit()?;
+
         // allow unreachable_code to get rids of the warning in the case of a test build.
         let this = Self {
             must_stop_processing: MustStopProcessing::default(),
             processing_tasks: Arc::new(RwLock::new(ProcessingTasks::new())),
             file_store,
-            all_tasks: env.create_database(Some(db_name::ALL_TASKS))?,
-            status: env.create_database(Some(db_name::STATUS))?,
-            kind: env.create_database(Some(db_name::KIND))?,
-            index_tasks: env.create_database(Some(db_name::INDEX_TASKS))?,
-            canceled_by: env.create_database(Some(db_name::CANCELED_BY))?,
-            enqueued_at: env.create_database(Some(db_name::ENQUEUED_AT))?,
-            started_at: env.create_database(Some(db_name::STARTED_AT))?,
-            finished_at: env.create_database(Some(db_name::FINISHED_AT))?,
+            all_tasks,
+            status,
+            kind,
+            index_tasks,
+            canceled_by,
+            enqueued_at,
+            started_at,
+            finished_at,
             index_mapper: IndexMapper::new(
                 &env,
                 options.indexes_path,
                 budget.map_size,
                 options.index_growth_amount,
                 budget.index_count,
+                options.enable_mdb_writemap,
                 options.indexer_config,
             )?,
             env,
@@ -1509,6 +1528,7 @@ mod tests {
                 dumps_path: tempdir.path().join("dumps"),
                 task_db_size: 1000 * 1000, // 1 MB, we don't use MiB on purpose.
                 index_base_map_size: 1000 * 1000, // 1 MB, we don't use MiB on purpose.
+                enable_mdb_writemap: false,
                 index_growth_amount: 1000 * 1000, // 1 MB
                 index_count: 5,
                 indexer_config,

index-scheduler/src/utils.rs

@@ -466,7 +466,7 @@ impl IndexScheduler {
                         }
                     }
                     Details::DocumentDeletionByFilter { deleted_documents, original_filter: _ } => {
-                        assert_eq!(kind.as_kind(), Kind::DocumentDeletionByFilter);
+                        assert_eq!(kind.as_kind(), Kind::DocumentDeletion);
                         let (index_uid, _) = if let KindWithContent::DocumentDeletionByFilter {
                             ref index_uid,
                             ref filter_expr,