snowykami/meilisearch
1
0
Fork 0
mirror of https://github.com/meilisearch/meilisearch.git synced 2025-09-05 20:26:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Provide a rename argument to the swap

Browse Source
This commit is contained in:
Tamo
2025-08-07 19:35:07 +02:00
parent ae5bd9d0e3
commit ecea247e5d
12 changed files with 358 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
crates/index-scheduler/src/error.rs
View File

@ -67,6 +67,8 @@ pub enum Error {
SwapDuplicateIndexesFound(Vec<String>),
#[error("Index `{0}` not found.")]
SwapIndexNotFound(String),
#[error("Index `{0}` found during a rename. Renaming doen't overwrite the other index name.")]
SwapIndexFoundDuringRename(String),
#[error("Meilisearch cannot receive write operations because the limit of the task database has been reached. Please delete tasks to continue performing write operations.")]
NoSpaceLeftInTaskQueue,
#[error(
@ -74,6 +76,10 @@ pub enum Error {
.0.iter().map(|s| format!("`{}`", s)).collect::<Vec<_>>().join(", ")
)]
SwapIndexesNotFound(Vec<String>),
#[error("Index {} found during a rename. Renaming doen't overwrite the other index name.",
.0.iter().map(|s| format!("`{}`", s)).collect::<Vec<_>>().join(", ")
)]
SwapIndexesFoundDuringRename(Vec<String>),
#[error("Corrupted dump.")]
CorruptedDump,
#[error(
@ -203,6 +209,8 @@ impl Error {
| Error::SwapIndexNotFound(_)
| Error::NoSpaceLeftInTaskQueue
| Error::SwapIndexesNotFound(_)
| Error::SwapIndexFoundDuringRename(_)
| Error::SwapIndexesFoundDuringRename(_)
| Error::CorruptedDump
| Error::InvalidTaskDate { .. }
| Error::InvalidTaskUid { .. }
@ -271,6 +279,8 @@ impl ErrorCode for Error {
Error::SwapDuplicateIndexFound(_) => Code::InvalidSwapDuplicateIndexFound,
Error::SwapIndexNotFound(_) => Code::IndexNotFound,
Error::SwapIndexesNotFound(_) => Code::IndexNotFound,
Error::SwapIndexFoundDuringRename(_) => Code::IndexNotFound,
Error::SwapIndexesFoundDuringRename(_) => Code::IndexNotFound,
Error::InvalidTaskDate { field, .. } => (*field).into(),
Error::InvalidTaskUid { .. } => Code::InvalidTaskUids,
Error::InvalidBatchUid { .. } => Code::InvalidBatchUids,

6
crates/index-scheduler/src/queue/batches_test.rs
View File

@ -334,11 +334,11 @@ fn query_batches_special_rules() {
let kind = index_creation_task("doggo", "sheep");
let _task = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "whalo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "whalo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();
@ -442,7 +442,7 @@ fn query_batches_canceled_by() {
let kind = index_creation_task("doggo", "sheep");
let _ = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();

6
crates/index-scheduler/src/queue/tasks_test.rs
View File

@ -304,11 +304,11 @@ fn query_tasks_special_rules() {
let kind = index_creation_task("doggo", "sheep");
let _task = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "whalo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "whalo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();
@ -399,7 +399,7 @@ fn query_tasks_canceled_by() {
let kind = index_creation_task("doggo", "sheep");
let _ = index_scheduler.register(kind, None, false).unwrap();
let kind = KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("catto".to_owned(), "doggo".to_owned()), rename: false }],
};
let _task = index_scheduler.register(kind, None, false).unwrap();

5
crates/index-scheduler/src/scheduler/autobatcher_test.rs
View File

@ -88,7 +88,10 @@ fn idx_del() -> KindWithContent {
fn idx_swap() -> KindWithContent {
KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: (String::from("doggo"), String::from("catto")) }],
swaps: vec![IndexSwap {
indexes: (String::from("doggo"), String::from("catto")),
rename: false,
}],
}
}

53
crates/index-scheduler/src/scheduler/process_batch.rs
View File

@ -360,13 +360,18 @@ impl IndexScheduler {
unreachable!()
};
let mut not_found_indexes = BTreeSet::new();
for IndexSwap { indexes: (lhs, rhs) } in swaps {
for index in [lhs, rhs] {
let index_exists = self.index_mapper.index_exists(&wtxn, index)?;
if !index_exists {
not_found_indexes.insert(index);
}
let mut found_indexes_but_should_not = BTreeSet::new();
for IndexSwap { indexes: (lhs, rhs), rename } in swaps {
let index_exists = self.index_mapper.index_exists(&wtxn, lhs)?;
if !index_exists {
not_found_indexes.insert(lhs);
}
let index_exists = self.index_mapper.index_exists(&wtxn, rhs)?;
match (index_exists, rename) {
(true, true) => found_indexes_but_should_not.insert(rhs),
(false, false) => not_found_indexes.insert(rhs),
(true, false) | (false, true) => true, // random value we don't read it anyway
};
}
if !not_found_indexes.is_empty() {
if not_found_indexes.len() == 1 {
@ -379,6 +384,17 @@ impl IndexScheduler {
));
}
}
if !found_indexes_but_should_not.is_empty() {
if found_indexes_but_should_not.len() == 1 {
return Err(Error::SwapIndexFoundDuringRename(
found_indexes_but_should_not.into_iter().next().unwrap().clone(),
));
} else {
return Err(Error::SwapIndexesFoundDuringRename(
found_indexes_but_should_not.into_iter().cloned().collect(),
));
}
}
progress.update_progress(SwappingTheIndexes::SwappingTheIndexes);
for (step, swap) in swaps.iter().enumerate() {
progress.update_progress(VariableNameStep::<SwappingTheIndexes>::new(
@ -392,6 +408,7 @@ impl IndexScheduler {
task.uid,
&swap.indexes.0,
&swap.indexes.1,
swap.rename,
)?;
}
wtxn.commit()?;
@ -481,6 +498,7 @@ impl IndexScheduler {
task_id: u32,
lhs: &str,
rhs: &str,
rename: bool,
) -> Result<()> {
progress.update_progress(InnerSwappingTwoIndexes::RetrieveTheTasks);
// 1. Verify that both lhs and rhs are existing indexes
@ -488,16 +506,23 @@ impl IndexScheduler {
if !index_lhs_exists {
return Err(Error::IndexNotFound(lhs.to_owned()));
}
let index_rhs_exists = self.index_mapper.index_exists(wtxn, rhs)?;
if !index_rhs_exists {
return Err(Error::IndexNotFound(rhs.to_owned()));
if !rename {
let index_rhs_exists = self.index_mapper.index_exists(wtxn, rhs)?;
if !index_rhs_exists {
return Err(Error::IndexNotFound(rhs.to_owned()));
}
}
// 2. Get the task set for index = name that appeared before the index swap task
let mut index_lhs_task_ids = self.queue.tasks.index_tasks(wtxn, lhs)?;
index_lhs_task_ids.remove_range(task_id..);
let mut index_rhs_task_ids = self.queue.tasks.index_tasks(wtxn, rhs)?;
index_rhs_task_ids.remove_range(task_id..);
let index_rhs_task_ids = if rename {
let mut index_rhs_task_ids = self.queue.tasks.index_tasks(wtxn, rhs)?;
index_rhs_task_ids.remove_range(task_id..);
index_rhs_task_ids
} else {
RoaringBitmap::new()
};
// 3. before_name -> new_name in the task's KindWithContent
progress.update_progress(InnerSwappingTwoIndexes::UpdateTheTasks);
@ -526,7 +551,11 @@ impl IndexScheduler {
})?;
// 6. Swap in the index mapper
self.index_mapper.swap(wtxn, lhs, rhs)?;
if rename {
self.index_mapper.rename(wtxn, lhs, rhs)?;
} else {
self.index_mapper.swap(wtxn, lhs, rhs)?;
}
Ok(())
}

16
crates/index-scheduler/src/scheduler/test.rs
View File

@ -372,8 +372,8 @@ fn swap_indexes() {
.register(
KindWithContent::IndexSwap {
swaps: vec![
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()) },
IndexSwap { indexes: ("c".to_owned(), "d".to_owned()) },
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()), rename: false },
IndexSwap { indexes: ("c".to_owned(), "d".to_owned()), rename: false },
],
},
None,
@ -384,7 +384,7 @@ fn swap_indexes() {
index_scheduler
.register(
KindWithContent::IndexSwap {
swaps: vec![IndexSwap { indexes: ("a".to_owned(), "c".to_owned()) }],
swaps: vec![IndexSwap { indexes: ("a".to_owned(), "c".to_owned()), rename: false }],
},
None,
false,
@ -428,8 +428,8 @@ fn swap_indexes_errors() {
.register(
KindWithContent::IndexSwap {
swaps: vec![
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()) },
IndexSwap { indexes: ("b".to_owned(), "a".to_owned()) },
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()), rename: false },
IndexSwap { indexes: ("b".to_owned(), "a".to_owned()), rename: false },
],
},
None,
@ -446,9 +446,9 @@ fn swap_indexes_errors() {
.register(
KindWithContent::IndexSwap {
swaps: vec![
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()) },
IndexSwap { indexes: ("c".to_owned(), "e".to_owned()) },
IndexSwap { indexes: ("d".to_owned(), "f".to_owned()) },
IndexSwap { indexes: ("a".to_owned(), "b".to_owned()), rename: false },
IndexSwap { indexes: ("c".to_owned(), "e".to_owned()), rename: false },
IndexSwap { indexes: ("d".to_owned(), "f".to_owned()), rename: false },
],
},
None,

6
crates/index-scheduler/src/utils.rs
View File

@ -271,7 +271,7 @@ pub fn swap_index_uid_in_task(task: &mut Task, swap: (&str, &str)) {
}
}
K::IndexSwap { swaps } => {
for IndexSwap { indexes: (lhs, rhs) } in swaps.iter_mut() {
for IndexSwap { indexes: (lhs, rhs), rename: _ } in swaps.iter_mut() {
if lhs == swap.0 || lhs == swap.1 {
index_uids.push(lhs);
}
@ -288,7 +288,7 @@ pub fn swap_index_uid_in_task(task: &mut Task, swap: (&str, &str)) {
| K::SnapshotCreation => (),
};
if let Some(Details::IndexSwap { swaps }) = &mut task.details {
for IndexSwap { indexes: (lhs, rhs) } in swaps.iter_mut() {
for IndexSwap { indexes: (lhs, rhs), rename: _ } in swaps.iter_mut() {
if lhs == swap.0 || lhs == swap.1 {
index_uids.push(lhs);
}
@ -330,7 +330,7 @@ pub(crate) fn check_index_swap_validity(task: &Task) -> Result<()> {
if let KindWithContent::IndexSwap { swaps } = &task.kind { swaps } else { return Ok(()) };
let mut all_indexes = HashSet::new();
let mut duplicate_indexes = BTreeSet::new();
for IndexSwap { indexes: (lhs, rhs) } in swaps {
for IndexSwap { indexes: (lhs, rhs), rename: _ } in swaps {
for name in [lhs, rhs] {
let is_new = all_indexes.insert(name);
if !is_new {