snowykami/meilisearch
1
0
Fork 0
mirror of https://github.com/meilisearch/meilisearch.git synced 2024-11-29 16:45:30 +08:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

it probably works but it's also horrendous

Browse Source
This commit is contained in:
Irevoire 2022-10-19 18:27:18 +02:00 committed by Clément Renault
parent d20b5ddda0
commit d9218578e3
No known key found for this signature in database
GPG Key ID: 92ADA4E935E71FA4
3 changed files with 535 additions and 232 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

571
index-scheduler/src/autobatcher.rs
View File

@ -59,9 +59,10 @@ impl From<KindWithContent> for AutobatchKind {
KindWithContent::DocumentClear { .. } => AutobatchKind::DocumentClear, KindWithContent::DocumentClear { .. } => AutobatchKind::DocumentClear,
KindWithContent::Settings { KindWithContent::Settings {
allow_index_creation, allow_index_creation,
is_deletion,
.. ..
} => AutobatchKind::Settings { } => AutobatchKind::Settings {
allow_index_creation, allow_index_creation: allow_index_creation && !is_deletion,
}, },
KindWithContent::IndexDeletion { .. } => AutobatchKind::IndexDeletion, KindWithContent::IndexDeletion { .. } => AutobatchKind::IndexDeletion,
KindWithContent::IndexCreation { .. } => AutobatchKind::IndexCreation, KindWithContent::IndexCreation { .. } => AutobatchKind::IndexCreation,
@ -134,50 +135,74 @@ impl BatchKind {
impl BatchKind { impl BatchKind {
/// Returns a `ControlFlow::Break` if you must stop right now. /// Returns a `ControlFlow::Break` if you must stop right now.
/// The boolean tell you if an index has been created by the batched task.
/// To ease the writting of the code. `true` can be returned when you don't need to create an index
/// but false can't be returned if you needs to create an index.
// TODO use an AutoBatchKind as input // TODO use an AutoBatchKind as input
pub fn new(task_id: TaskId, kind: KindWithContent) -> ControlFlow<BatchKind, BatchKind> { pub fn new(
task_id: TaskId,
kind: KindWithContent,
) -> (ControlFlow<BatchKind, BatchKind>, bool) {
use AutobatchKind as K; use AutobatchKind as K;
match AutobatchKind::from(kind) { match AutobatchKind::from(kind) {
K::IndexCreation => Break(BatchKind::IndexCreation { id: task_id }), K::IndexCreation => (Break(BatchKind::IndexCreation { id: task_id }), true),
K::IndexDeletion => Break(BatchKind::IndexDeletion { ids: vec![task_id] }), K::IndexDeletion => (
K::IndexUpdate => Break(BatchKind::IndexUpdate { id: task_id }), Break(BatchKind::IndexDeletion { ids: vec![task_id] }),
K::IndexSwap => Break(BatchKind::IndexSwap { id: task_id }), false,
K::DocumentClear => Continue(BatchKind::DocumentClear { ids: vec![task_id] }), ),
K::IndexUpdate => (Break(BatchKind::IndexUpdate { id: task_id }), false),
K::IndexSwap => (Break(BatchKind::IndexSwap { id: task_id }), false),
K::DocumentClear => (
Continue(BatchKind::DocumentClear { ids: vec![task_id] }),
false,
),
K::DocumentImport { K::DocumentImport {
method, method,
allow_index_creation, allow_index_creation,
} => Continue(BatchKind::DocumentImport { } => (
Continue(BatchKind::DocumentImport {
method, method,
allow_index_creation, allow_index_creation,
import_ids: vec![task_id], import_ids: vec![task_id],
}), }),
K::DocumentDeletion => Continue(BatchKind::DocumentDeletion { allow_index_creation,
),
K::DocumentDeletion => (
Continue(BatchKind::DocumentDeletion {
deletion_ids: vec![task_id], deletion_ids: vec![task_id],
}), }),
false,
),
K::Settings { K::Settings {
allow_index_creation, allow_index_creation,
} => Continue(BatchKind::Settings { } => (
Continue(BatchKind::Settings {
allow_index_creation, allow_index_creation,
settings_ids: vec![task_id], settings_ids: vec![task_id],
}), }),
allow_index_creation,
),
} }
} }
/// Returns a `ControlFlow::Break` if you must stop right now. /// Returns a `ControlFlow::Break` if you must stop right now.
/// The boolean tell you if an index has been created by the batched task.
/// To ease the writting of the code. `true` can be returned when you don't need to create an index
/// but false can't be returned if you needs to create an index.
#[rustfmt::skip] #[rustfmt::skip]
fn accumulate(self, id: TaskId, kind: AutobatchKind) -> ControlFlow<BatchKind, BatchKind> { fn accumulate(self, id: TaskId, kind: AutobatchKind, index_already_exists: bool) -> (ControlFlow<BatchKind, BatchKind>, bool) {
use AutobatchKind as K; use AutobatchKind as K;
match (self, kind) { match (index_already_exists, self, kind) {
// We don't batch any of these operations // We don't batch any of these operations.
(this, K::IndexCreation | K::IndexUpdate | K::IndexSwap) => Break(this), (true, this, K::IndexCreation | K::IndexUpdate | K::IndexSwap) => (Break(this), true),
// We must not batch tasks that don't have the same index creation rights // We must not batch tasks that don't have the same index creation rights.
(this, kind) if this.allow_index_creation() == Some(false) && kind.allow_index_creation() == Some(true) => { (true, this, kind) if this.allow_index_creation() == Some(false) && kind.allow_index_creation() == Some(true) => {
Break(this) (Break(this), true)
}, },
// The index deletion can batch with everything but must stop after // The index deletion can batch with everything but must stop after
( (true,
BatchKind::DocumentClear { mut ids } BatchKind::DocumentClear { mut ids }
| BatchKind::DocumentDeletion { deletion_ids: mut ids } | BatchKind::DocumentDeletion { deletion_ids: mut ids }
| BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids } | BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids }
@ -185,125 +210,125 @@ impl BatchKind {
K::IndexDeletion, K::IndexDeletion,
) => { ) => {
ids.push(id); ids.push(id);
Break(BatchKind::IndexDeletion { ids }) (Break(BatchKind::IndexDeletion { ids }), true)
} }
( (true,
BatchKind::ClearAndSettings { settings_ids: mut ids, allow_index_creation: _, mut other } BatchKind::ClearAndSettings { settings_ids: mut ids, allow_index_creation: _, mut other }
| BatchKind::SettingsAndDocumentImport { import_ids: mut ids, method: _, allow_index_creation: _, settings_ids: mut other }, | BatchKind::SettingsAndDocumentImport { import_ids: mut ids, method: _, allow_index_creation: _, settings_ids: mut other },
K::IndexDeletion, K::IndexDeletion,
) => { ) => {
ids.push(id); ids.push(id);
ids.append(&mut other); ids.append(&mut other);
Break(BatchKind::IndexDeletion { ids }) (Break(BatchKind::IndexDeletion { ids }), true)
} }
( (true,
BatchKind::DocumentClear { mut ids }, BatchKind::DocumentClear { mut ids },
K::DocumentClear | K::DocumentDeletion, K::DocumentClear | K::DocumentDeletion,
) => { ) => {
ids.push(id); ids.push(id);
Continue(BatchKind::DocumentClear { ids }) (Continue(BatchKind::DocumentClear { ids }), true)
} }
( (true,
this @ BatchKind::DocumentClear { .. }, this @ BatchKind::DocumentClear { .. },
K::DocumentImport { .. } | K::Settings { .. }, K::DocumentImport { .. } | K::Settings { .. },
) => Break(this), ) => (Break(this), true),
( (true,
BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids }, BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids },
K::DocumentClear, K::DocumentClear,
) => { ) => {
ids.push(id); ids.push(id);
Continue(BatchKind::DocumentClear { ids }) (Continue(BatchKind::DocumentClear { ids }), true)
} }
// we can autobatch the same kind of document additions / updates // we can autobatch the same kind of document additions / updates
( ( true,
BatchKind::DocumentImport { method: ReplaceDocuments, allow_index_creation, mut import_ids }, BatchKind::DocumentImport { method: ReplaceDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: ReplaceDocuments, .. }, K::DocumentImport { method: ReplaceDocuments, .. },
) => { ) => {
import_ids.push(id); import_ids.push(id);
Continue(BatchKind::DocumentImport { (Continue(BatchKind::DocumentImport {
method: ReplaceDocuments, method: ReplaceDocuments,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}) }), true)
} }
( (true,
BatchKind::DocumentImport { method: UpdateDocuments, allow_index_creation, mut import_ids }, BatchKind::DocumentImport { method: UpdateDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: UpdateDocuments, .. }, K::DocumentImport { method: UpdateDocuments, .. },
) => { ) => {
import_ids.push(id); import_ids.push(id);
Continue(BatchKind::DocumentImport { (Continue(BatchKind::DocumentImport {
method: UpdateDocuments, method: UpdateDocuments,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}) }), true)
} }
// but we can't autobatch documents if it's not the same kind // but we can't autobatch documents if it's not the same kind
// this match branch MUST be AFTER the previous one // this match branch MUST be AFTER the previous one
( (true,
this @ BatchKind::DocumentImport { .. }, this @ BatchKind::DocumentImport { .. },
K::DocumentDeletion | K::DocumentImport { .. }, K::DocumentDeletion | K::DocumentImport { .. },
) => Break(this), ) => (Break(this), true),
( (true,
BatchKind::DocumentImport { method, allow_index_creation, import_ids }, BatchKind::DocumentImport { method, allow_index_creation, import_ids },
K::Settings { .. }, K::Settings { .. },
) => Continue(BatchKind::SettingsAndDocumentImport { ) => (Continue(BatchKind::SettingsAndDocumentImport {
settings_ids: vec![id], settings_ids: vec![id],
method, method,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}), }), true),
(BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentClear) => { (true, BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentClear) => {
deletion_ids.push(id); deletion_ids.push(id);
Continue(BatchKind::DocumentClear { ids: deletion_ids }) (Continue(BatchKind::DocumentClear { ids: deletion_ids }), true)
} }
(this @ BatchKind::DocumentDeletion { .. }, K::DocumentImport { .. }) => Break(this), (true, this @ BatchKind::DocumentDeletion { .. }, K::DocumentImport { .. }) => (Break(this), true),
(BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentDeletion) => { (true, BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentDeletion) => {
deletion_ids.push(id); deletion_ids.push(id);
Continue(BatchKind::DocumentDeletion { deletion_ids }) (Continue(BatchKind::DocumentDeletion { deletion_ids }), true)
} }
(this @ BatchKind::DocumentDeletion { .. }, K::Settings { .. }) => Break(this), (true, this @ BatchKind::DocumentDeletion { .. }, K::Settings { .. }) => (Break(this), true),
( (true,
BatchKind::Settings { settings_ids, allow_index_creation }, BatchKind::Settings { settings_ids, allow_index_creation },
K::DocumentClear, K::DocumentClear,
) => Continue(BatchKind::ClearAndSettings { ) => (Continue(BatchKind::ClearAndSettings {
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
other: vec![id], other: vec![id],
}), }), true),
( (true,
this @ BatchKind::Settings { .. }, this @ BatchKind::Settings { .. },
K::DocumentImport { .. } | K::DocumentDeletion, K::DocumentImport { .. } | K::DocumentDeletion,
) => Break(this), ) => (Break(this), true),
( (true,
BatchKind::Settings { mut settings_ids, allow_index_creation }, BatchKind::Settings { mut settings_ids, allow_index_creation },
K::Settings { .. }, K::Settings { .. },
) => { ) => {
settings_ids.push(id); settings_ids.push(id);
Continue(BatchKind::Settings { (Continue(BatchKind::Settings {
allow_index_creation, allow_index_creation,
settings_ids, settings_ids,
}) }), true)
} }
( (true,
BatchKind::ClearAndSettings { mut other, settings_ids, allow_index_creation }, BatchKind::ClearAndSettings { mut other, settings_ids, allow_index_creation },
K::DocumentClear, K::DocumentClear,
) => { ) => {
other.push(id); other.push(id);
Continue(BatchKind::ClearAndSettings { (Continue(BatchKind::ClearAndSettings {
other, other,
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
}) }), true)
} }
(this @ BatchKind::ClearAndSettings { .. }, K::DocumentImport { .. }) => Break(this), (true, this @ BatchKind::ClearAndSettings { .. }, K::DocumentImport { .. }) => (Break(this), true),
( (true,
BatchKind::ClearAndSettings { BatchKind::ClearAndSettings {
mut other, mut other,
settings_ids, settings_ids,
@ -312,78 +337,297 @@ impl BatchKind {
K::DocumentDeletion, K::DocumentDeletion,
) => { ) => {
other.push(id); other.push(id);
Continue(BatchKind::ClearAndSettings { (Continue(BatchKind::ClearAndSettings {
other, other,
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
}) }), true)
} }
( (true,
BatchKind::ClearAndSettings { mut settings_ids, other, allow_index_creation }, BatchKind::ClearAndSettings { mut settings_ids, other, allow_index_creation },
K::Settings { .. }, K::Settings { .. },
) => { ) => {
settings_ids.push(id); settings_ids.push(id);
Continue(BatchKind::ClearAndSettings { (Continue(BatchKind::ClearAndSettings {
other, other,
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
}) }), true)
} }
( (true,
BatchKind::SettingsAndDocumentImport { settings_ids, method: _, import_ids: mut other, allow_index_creation }, BatchKind::SettingsAndDocumentImport { settings_ids, method: _, import_ids: mut other, allow_index_creation },
K::DocumentClear, K::DocumentClear,
) => { ) => {
other.push(id); other.push(id);
Continue(BatchKind::ClearAndSettings { (Continue(BatchKind::ClearAndSettings {
settings_ids, settings_ids,
other, other,
allow_index_creation, allow_index_creation,
}) }), true)
} }
( (true,
BatchKind::SettingsAndDocumentImport { settings_ids, method: ReplaceDocuments, mut import_ids, allow_index_creation }, BatchKind::SettingsAndDocumentImport { settings_ids, method: ReplaceDocuments, mut import_ids, allow_index_creation },
K::DocumentImport { method: ReplaceDocuments, .. }, K::DocumentImport { method: ReplaceDocuments, .. },
) => { ) => {
import_ids.push(id); import_ids.push(id);
Continue(BatchKind::SettingsAndDocumentImport { (Continue(BatchKind::SettingsAndDocumentImport {
settings_ids, settings_ids,
method: ReplaceDocuments, method: ReplaceDocuments,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}) }), true)
} }
( (true,
BatchKind::SettingsAndDocumentImport { settings_ids, method: UpdateDocuments, allow_index_creation, mut import_ids }, BatchKind::SettingsAndDocumentImport { settings_ids, method: UpdateDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: UpdateDocuments, .. }, K::DocumentImport { method: UpdateDocuments, .. },
) => { ) => {
import_ids.push(id); import_ids.push(id);
Continue(BatchKind::SettingsAndDocumentImport { (Continue(BatchKind::SettingsAndDocumentImport {
settings_ids, settings_ids,
method: UpdateDocuments, method: UpdateDocuments,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}) }), true)
} }
// But we can't batch a settings and a doc op with another doc op // But we can't batch a settings and a doc op with another doc op
// this MUST be AFTER the two previous branch // this MUST be AFTER the two previous branch
( (true,
this @ BatchKind::SettingsAndDocumentImport { .. }, this @ BatchKind::SettingsAndDocumentImport { .. },
K::DocumentDeletion | K::DocumentImport { .. }, K::DocumentDeletion | K::DocumentImport { .. },
) => Break(this), ) => (Break(this), true),
( (true,
BatchKind::SettingsAndDocumentImport { mut settings_ids, method, allow_index_creation, import_ids }, BatchKind::SettingsAndDocumentImport { mut settings_ids, method, allow_index_creation, import_ids },
K::Settings { .. }, K::Settings { .. },
) => { ) => {
settings_ids.push(id); settings_ids.push(id);
Continue(BatchKind::SettingsAndDocumentImport { (Continue(BatchKind::SettingsAndDocumentImport {
settings_ids, settings_ids,
method, method,
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}) }), true)
} }
(
// TODO
// We don't batch any of these operations.
(false, this, K::IndexCreation) => (Break(this), true),
(false, this, K::IndexUpdate | K::IndexSwap) => (Break(this), false),
// We must not batch tasks that don't have the same index creation rights.
(false, this, kind) if this.allow_index_creation() == Some(false) && kind.allow_index_creation() == Some(true) => {
(Break(this), false)
},
// The index deletion can batch with everything but must stop after
(false,
BatchKind::DocumentClear { mut ids }
| BatchKind::DocumentDeletion { deletion_ids: mut ids }
| BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids }
| BatchKind::Settings { allow_index_creation: _, settings_ids: mut ids },
K::IndexDeletion,
) => {
ids.push(id);
(Break(BatchKind::IndexDeletion { ids }), false)
}
(false,
BatchKind::ClearAndSettings { settings_ids: mut ids, allow_index_creation: _, mut other }
| BatchKind::SettingsAndDocumentImport { import_ids: mut ids, method: _, allow_index_creation: _, settings_ids: mut other },
K::IndexDeletion,
) => {
ids.push(id);
ids.append(&mut other);
(Break(BatchKind::IndexDeletion { ids }), false)
}
(false,
BatchKind::DocumentClear { mut ids },
K::DocumentClear | K::DocumentDeletion,
) => {
ids.push(id);
(Continue(BatchKind::DocumentClear { ids }), false)
}
(false,
this @ BatchKind::DocumentClear { .. },
K::DocumentImport { .. } | K::Settings { .. },
) => (Break(this), false),
(false,
BatchKind::DocumentImport { method: _, allow_index_creation: _, import_ids: mut ids },
K::DocumentClear,
) => {
ids.push(id);
(Continue(BatchKind::DocumentClear { ids }), false)
}
// we can autobatch the same kind of document additions / updates
(false,
BatchKind::DocumentImport { method: ReplaceDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: ReplaceDocuments, .. },
) => {
import_ids.push(id);
(Continue(BatchKind::DocumentImport {
method: ReplaceDocuments,
allow_index_creation,
import_ids,
}), false)
}
(false,
BatchKind::DocumentImport { method: UpdateDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: UpdateDocuments, .. },
) => {
import_ids.push(id);
(Continue(BatchKind::DocumentImport {
method: UpdateDocuments,
allow_index_creation,
import_ids,
}), false)
}
// but we can't autobatch documents if it's not the same kind
// this match branch MUST be AFTER the previous one
(false,
this @ BatchKind::DocumentImport { .. },
K::DocumentDeletion | K::DocumentImport { .. },
) => (Break(this), false),
(false,
BatchKind::DocumentImport { method, allow_index_creation, import_ids },
K::Settings { .. },
) => (Continue(BatchKind::SettingsAndDocumentImport {
settings_ids: vec![id],
method,
allow_index_creation,
import_ids,
}), false),
(false, BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentClear) => {
deletion_ids.push(id);
(Continue(BatchKind::DocumentClear { ids: deletion_ids }), false)
}
(false, this @ BatchKind::DocumentDeletion { .. }, K::DocumentImport { .. }) => (Break(this), false),
(false, BatchKind::DocumentDeletion { mut deletion_ids }, K::DocumentDeletion) => {
deletion_ids.push(id);
(Continue(BatchKind::DocumentDeletion { deletion_ids }), false)
}
(false, this @ BatchKind::DocumentDeletion { .. }, K::Settings { .. }) => (Break(this), false),
(false,
BatchKind::Settings { settings_ids, allow_index_creation },
K::DocumentClear,
) => (Continue(BatchKind::ClearAndSettings {
settings_ids,
allow_index_creation,
other: vec![id],
}), false),
(false,
this @ BatchKind::Settings { .. },
K::DocumentImport { .. } | K::DocumentDeletion,
) => (Break(this), false),
(false,
BatchKind::Settings { mut settings_ids, allow_index_creation },
K::Settings { .. },
) => {
settings_ids.push(id);
(Continue(BatchKind::Settings {
allow_index_creation,
settings_ids,
}), false)
}
(false,
BatchKind::ClearAndSettings { mut other, settings_ids, allow_index_creation },
K::DocumentClear,
) => {
other.push(id);
(Continue(BatchKind::ClearAndSettings {
other,
settings_ids,
allow_index_creation,
}), false)
}
(false, this @ BatchKind::ClearAndSettings { .. }, K::DocumentImport { .. }) => (Break(this), false),
(false,
BatchKind::ClearAndSettings {
mut other,
settings_ids,
allow_index_creation,
},
K::DocumentDeletion,
) => {
other.push(id);
(Continue(BatchKind::ClearAndSettings {
other,
settings_ids,
allow_index_creation,
}), false)
}
(false,
BatchKind::ClearAndSettings { mut settings_ids, other, allow_index_creation },
K::Settings { .. },
) => {
settings_ids.push(id);
(Continue(BatchKind::ClearAndSettings {
other,
settings_ids,
allow_index_creation,
}), false)
}
(false,
BatchKind::SettingsAndDocumentImport { settings_ids, method: _, import_ids: mut other, allow_index_creation },
K::DocumentClear,
) => {
other.push(id);
(Continue(BatchKind::ClearAndSettings {
settings_ids,
other,
allow_index_creation,
}), false)
}
(false,
BatchKind::SettingsAndDocumentImport { settings_ids, method: ReplaceDocuments, mut import_ids, allow_index_creation },
K::DocumentImport { method: ReplaceDocuments, .. },
) => {
import_ids.push(id);
(Continue(BatchKind::SettingsAndDocumentImport {
settings_ids,
method: ReplaceDocuments,
allow_index_creation,
import_ids,
}), false)
}
(false,
BatchKind::SettingsAndDocumentImport { settings_ids, method: UpdateDocuments, allow_index_creation, mut import_ids },
K::DocumentImport { method: UpdateDocuments, .. },
) => {
import_ids.push(id);
(Continue(BatchKind::SettingsAndDocumentImport {
settings_ids,
method: UpdateDocuments,
allow_index_creation,
import_ids,
}), false)
}
// But we can't batch a settings and a doc op with another doc op
// this MUST be AFTER the two previous branch
(false,
this @ BatchKind::SettingsAndDocumentImport { .. },
K::DocumentDeletion | K::DocumentImport { .. },
) => (Break(this), false),
(false,
BatchKind::SettingsAndDocumentImport { mut settings_ids, method, allow_index_creation, import_ids },
K::Settings { .. },
) => {
settings_ids.push(id);
(Continue(BatchKind::SettingsAndDocumentImport {
settings_ids,
method,
allow_index_creation,
import_ids,
}), false)
}
(_,
BatchKind::IndexCreation { .. } BatchKind::IndexCreation { .. }
| BatchKind::IndexDeletion { .. } | BatchKind::IndexDeletion { .. }
| BatchKind::IndexUpdate { .. } | BatchKind::IndexUpdate { .. }
@ -392,6 +636,7 @@ impl BatchKind {
) => { ) => {
unreachable!() unreachable!()
} }
} }
} }
} }
@ -406,22 +651,35 @@ impl BatchKind {
/// ## Return /// ## Return
/// `None` if the list of tasks is empty. Otherwise, an [`AutoBatch`] that represents /// `None` if the list of tasks is empty. Otherwise, an [`AutoBatch`] that represents
/// a subset of the given tasks. /// a subset of the given tasks.
pub fn autobatch(enqueued: Vec<(TaskId, KindWithContent)>) -> Option<BatchKind> { pub fn autobatch(
enqueued: Vec<(TaskId, KindWithContent)>,
index_already_exists: bool,
) -> Option<(BatchKind, bool)> {
let mut enqueued = enqueued.into_iter(); let mut enqueued = enqueued.into_iter();
let (id, kind) = enqueued.next()?; let (id, kind) = enqueued.next()?;
let mut acc = match BatchKind::new(id, kind) {
Continue(acc) => acc, // index_exist will keep track of if the index should exist at this point after the tasks we batched.
Break(acc) => return Some(acc), let mut index_exist = index_already_exists;
let (mut acc, mut must_create_index) = match BatchKind::new(id, kind) {
(Continue(acc), create) => (acc, create),
(Break(acc), create) => return Some((acc, create)),
}; };
for (id, kind) in enqueued { for (id, kind) in enqueued {
acc = match acc.accumulate(id, kind.into()) { // if an index has been created in the previous step we can consider it exists.
Continue(acc) => acc, index_exist |= must_create_index;
Break(acc) => return Some(acc),
match acc.accumulate(id, kind.into(), index_exist) {
(Continue(a), create) => {
acc = a;
must_create_index |= create;
}
(Break(acc), create) => return Some((acc, must_create_index | create)),
}; };
} }
Some(acc) Some((acc, must_create_index))
} }
#[cfg(test)] #[cfg(test)]
@ -431,13 +689,14 @@ mod tests {
use super::*; use super::*;
use uuid::Uuid; use uuid::Uuid;
fn autobatch_from(input: impl IntoIterator<Item = KindWithContent>) -> Option<BatchKind> { fn autobatch_from(index_already_exists: bool, input: impl IntoIterator<Item = KindWithContent>) -> Option<(BatchKind, bool)> {
autobatch( autobatch(
input input
.into_iter() .into_iter()
.enumerate() .enumerate()
.map(|(id, kind)| (id as TaskId, kind.into())) .map(|(id, kind)| (id as TaskId, kind.into()))
.collect(), .collect(),
index_already_exists,
) )
} }
@ -503,129 +762,129 @@ mod tests {
#[test] #[test]
fn autobatch_simple_operation_together() { fn autobatch_simple_operation_together() {
// we can autobatch one or multiple DocumentAddition together // we can autobatch one or multiple DocumentAddition together
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp( ReplaceDocuments, true ), doc_imp(ReplaceDocuments, true )]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp( ReplaceDocuments, true ), doc_imp(ReplaceDocuments, true )]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1, 2] })");
// we can autobatch one or multiple DocumentUpdate together // we can autobatch one or multiple DocumentUpdate together
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0, 1, 2] })");
// we can autobatch one or multiple DocumentDeletion together // we can autobatch one or multiple DocumentDeletion together
debug_snapshot!(autobatch_from([doc_del()]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del()]), @"Some(DocumentDeletion { deletion_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), doc_del(), doc_del()]), @"Some(DocumentDeletion { deletion_ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_del(), doc_del(), doc_del()]), @"Some(DocumentDeletion { deletion_ids: [0, 1, 2] })");
// we can autobatch one or multiple Settings together // we can autobatch one or multiple Settings together
debug_snapshot!(autobatch_from([settings(true)]), @"Some(Settings { allow_index_creation: true, settings_ids: [0] })"); debug_snapshot!(autobatch_from(true, [settings(true)]), @"Some(Settings { allow_index_creation: true, settings_ids: [0] })");
debug_snapshot!(autobatch_from([settings(true), settings(true), settings(true)]), @"Some(Settings { allow_index_creation: true, settings_ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [settings(true), settings(true), settings(true)]), @"Some(Settings { allow_index_creation: true, settings_ids: [0, 1, 2] })");
} }
#[test] #[test]
fn simple_document_operation_dont_autobatch_with_other() { fn simple_document_operation_dont_autobatch_with_other() {
// addition, updates and deletion can't batch together // addition, updates and deletion can't batch together
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(UpdateDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_del()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_del()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_del()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_del()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentDeletion { deletion_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentDeletion { deletion_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), idx_create()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), idx_create()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), idx_create()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), idx_create()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), idx_create()]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del(), idx_create()]), @"Some(DocumentDeletion { deletion_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), idx_update()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), idx_update()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), idx_update()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), idx_update()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), idx_update()]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del(), idx_update()]), @"Some(DocumentDeletion { deletion_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), idx_swap()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), idx_swap()]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), idx_swap()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), idx_swap()]), @"Some(DocumentImport { method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_del(), idx_swap()]), @"Some(DocumentDeletion { deletion_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_del(), idx_swap()]), @"Some(DocumentDeletion { deletion_ids: [0] })");
} }
#[test] #[test]
fn document_addition_batch_with_settings() { fn document_addition_batch_with_settings() {
// simple case // simple case
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
// multiple settings and doc addition // multiple settings and doc addition
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), settings(true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [2, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), settings(true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [2, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), settings(true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [2, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), settings(true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [2, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })");
// addition and setting unordered // addition and setting unordered
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1, 3], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 2] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), doc_imp(UpdateDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1, 3], method: UpdateDocuments, allow_index_creation: true, import_ids: [0, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), doc_imp(UpdateDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1, 3], method: UpdateDocuments, allow_index_creation: true, import_ids: [0, 2] })");
// We ensure this kind of batch doesn't batch with forbidden operations // We ensure this kind of batch doesn't batch with forbidden operations
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), doc_imp(UpdateDocuments, true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), doc_imp(UpdateDocuments, true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), doc_imp(ReplaceDocuments, true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), doc_imp(ReplaceDocuments, true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), doc_del()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), doc_del()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), doc_del()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), doc_del()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), idx_create()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), idx_create()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), idx_create()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), idx_create()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), idx_update()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), idx_update()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), idx_update()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), idx_update()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), idx_swap()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), idx_swap()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), idx_swap()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), idx_swap()]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: UpdateDocuments, allow_index_creation: true, import_ids: [0] })");
} }
#[test] #[test]
fn clear_and_additions() { fn clear_and_additions() {
// these two doesn't need to batch // these two doesn't need to batch
debug_snapshot!(autobatch_from([doc_clr(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentClear { ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_clr(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentClear { ids: [0] })");
debug_snapshot!(autobatch_from([doc_clr(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentClear { ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_clr(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentClear { ids: [0] })");
// Basic use case // Basic use case
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2] })");
// This batch kind doesn't mix with other document addition // This batch kind doesn't mix with other document addition
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentClear { ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr(), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentClear { ids: [0, 1, 2] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentClear { ids: [0, 1, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr(), doc_imp(UpdateDocuments, true)]), @"Some(DocumentClear { ids: [0, 1, 2] })");
// But you can batch multiple clear together // But you can batch multiple clear together
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr(), doc_clr(), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2, 3, 4] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true), doc_clr(), doc_clr(), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2, 3, 4] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr(), doc_clr(), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2, 3, 4] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), doc_imp(UpdateDocuments, true), doc_clr(), doc_clr(), doc_clr()]), @"Some(DocumentClear { ids: [0, 1, 2, 3, 4] })");
} }
#[test] #[test]
fn clear_and_additions_and_settings() { fn clear_and_additions_and_settings() {
// A clear don't need to autobatch the settings that happens AFTER there is no documents // A clear don't need to autobatch the settings that happens AFTER there is no documents
debug_snapshot!(autobatch_from([doc_clr(), settings(true)]), @"Some(DocumentClear { ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_clr(), settings(true)]), @"Some(DocumentClear { ids: [0] })");
debug_snapshot!(autobatch_from([settings(true), doc_clr(), settings(true)]), @"Some(ClearAndSettings { other: [1], allow_index_creation: true, settings_ids: [0, 2] })"); debug_snapshot!(autobatch_from(true, [settings(true), doc_clr(), settings(true)]), @"Some(ClearAndSettings { other: [1], allow_index_creation: true, settings_ids: [0, 2] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), doc_clr()]), @"Some(ClearAndSettings { other: [0, 2], allow_index_creation: true, settings_ids: [1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), doc_clr()]), @"Some(ClearAndSettings { other: [0, 2], allow_index_creation: true, settings_ids: [1] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), doc_clr()]), @"Some(ClearAndSettings { other: [0, 2], allow_index_creation: true, settings_ids: [1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), doc_clr()]), @"Some(ClearAndSettings { other: [0, 2], allow_index_creation: true, settings_ids: [1] })");
} }
#[test] #[test]
fn anything_and_index_deletion() { fn anything_and_index_deletion() {
// The indexdeletion doesn't batch with anything that happens AFTER // The indexdeletion doesn't batch with anything that happens AFTER
debug_snapshot!(autobatch_from([idx_del(), doc_imp(ReplaceDocuments, true)]), @"Some(IndexDeletion { ids: [0] })"); debug_snapshot!(autobatch_from(true, [idx_del(), doc_imp(ReplaceDocuments, true)]), @"Some(IndexDeletion { ids: [0] })");
debug_snapshot!(autobatch_from([idx_del(), doc_imp(UpdateDocuments, true)]), @"Some(IndexDeletion { ids: [0] })"); debug_snapshot!(autobatch_from(true, [idx_del(), doc_imp(UpdateDocuments, true)]), @"Some(IndexDeletion { ids: [0] })");
debug_snapshot!(autobatch_from([idx_del(), doc_del()]), @"Some(IndexDeletion { ids: [0] })"); debug_snapshot!(autobatch_from(true, [idx_del(), doc_del()]), @"Some(IndexDeletion { ids: [0] })");
debug_snapshot!(autobatch_from([idx_del(), doc_clr()]), @"Some(IndexDeletion { ids: [0] })"); debug_snapshot!(autobatch_from(true, [idx_del(), doc_clr()]), @"Some(IndexDeletion { ids: [0] })");
debug_snapshot!(autobatch_from([idx_del(), settings(true)]), @"Some(IndexDeletion { ids: [0] })"); debug_snapshot!(autobatch_from(true, [idx_del(), settings(true)]), @"Some(IndexDeletion { ids: [0] })");
// The index deletion can accept almost any type of BatchKind and transform it to an idx_del() // The index deletion can accept almost any type of BatchKind and transform it to an idx_del()
// First, the basic cases // First, the basic cases
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_del(), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_del(), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })");
debug_snapshot!(autobatch_from([settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 1] })");
// Then the mixed cases // Then the mixed cases
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 2, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 2, 1] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 2, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), idx_del()]), @"Some(IndexDeletion { ids: [0, 2, 1] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true), doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [1, 3, 0, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true), doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [1, 3, 0, 2] })");
debug_snapshot!(autobatch_from([doc_imp(UpdateDocuments, true), settings(true), doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [1, 3, 0, 2] })"); debug_snapshot!(autobatch_from(true, [doc_imp(UpdateDocuments, true), settings(true), doc_clr(), idx_del()]), @"Some(IndexDeletion { ids: [1, 3, 0, 2] })");
} }
#[test] #[test]
fn allowed_and_disallowed_index_creation() { fn allowed_and_disallowed_index_creation() {
// doc_imp(indexes canbe)ixed with those disallowed to do so // doc_imp(indexes canbe)ixed with those disallowed to do so
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, false), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, false), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), doc_imp(ReplaceDocuments, true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: true, import_ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, false), doc_imp(ReplaceDocuments, false)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0, 1] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, false), doc_imp(ReplaceDocuments, false)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0, 1] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, true), settings(true)]), @"Some(SettingsAndDocumentImport { settings_ids: [1], method: ReplaceDocuments, allow_index_creation: true, import_ids: [0] })");
debug_snapshot!(autobatch_from([doc_imp(ReplaceDocuments, false), settings(true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0] })"); debug_snapshot!(autobatch_from(true, [doc_imp(ReplaceDocuments, false), settings(true)]), @"Some(DocumentImport { method: ReplaceDocuments, allow_index_creation: false, import_ids: [0] })");
} }
} }

136
index-scheduler/src/batch.rs
View File

@ -57,7 +57,10 @@ pub(crate) enum Batch {
TaskDeletion(Task), TaskDeletion(Task),
Snapshot(Vec<Task>), Snapshot(Vec<Task>),
Dump(Task), Dump(Task),
IndexOperation(IndexOperation), IndexOperation {
op: IndexOperation,
must_create_index: bool,
},
IndexCreation { IndexCreation {
index_uid: String, index_uid: String,
primary_key: Option<String>, primary_key: Option<String>,
@ -142,7 +145,7 @@ impl Batch {
Batch::Snapshot(tasks) | Batch::IndexDeletion { tasks, .. } => { Batch::Snapshot(tasks) | Batch::IndexDeletion { tasks, .. } => {
tasks.iter().map(|task| task.uid).collect() tasks.iter().map(|task| task.uid).collect()
} }
Batch::IndexOperation(operation) => match operation { Batch::IndexOperation { op, .. } => match op {
IndexOperation::DocumentImport { tasks, .. } IndexOperation::DocumentImport { tasks, .. }
| IndexOperation::DocumentDeletion { tasks, .. } | IndexOperation::DocumentDeletion { tasks, .. }
| IndexOperation::Settings { tasks, .. } | IndexOperation::Settings { tasks, .. }
@ -165,6 +168,19 @@ impl Batch {
} }
} }
impl IndexOperation {
pub fn index_uid(&self) -> &str {
match self {
IndexOperation::DocumentImport { index_uid, .. }
| IndexOperation::DocumentDeletion { index_uid, .. }
| IndexOperation::DocumentClear { index_uid, .. }
| IndexOperation::Settings { index_uid, .. }
| IndexOperation::DocumentClearAndSetting { index_uid, .. }
| IndexOperation::SettingsAndDocumentImport { index_uid, .. } => index_uid,
}
}
}
impl IndexScheduler { impl IndexScheduler {
/// Convert an [`BatchKind`](crate::autobatcher::BatchKind) into a [`Batch`]. /// Convert an [`BatchKind`](crate::autobatcher::BatchKind) into a [`Batch`].
/// ///
@ -177,14 +193,16 @@ impl IndexScheduler {
rtxn: &RoTxn, rtxn: &RoTxn,
index_uid: String, index_uid: String,
batch: BatchKind, batch: BatchKind,
must_create_index: bool,
) -> Result<Option<Batch>> { ) -> Result<Option<Batch>> {
match batch { match batch {
BatchKind::DocumentClear { ids } => { BatchKind::DocumentClear { ids } => Ok(Some(Batch::IndexOperation {
Ok(Some(Batch::IndexOperation(IndexOperation::DocumentClear { op: IndexOperation::DocumentClear {
tasks: self.get_existing_tasks(rtxn, ids)?, tasks: self.get_existing_tasks(rtxn, ids)?,
index_uid, index_uid,
}))) },
} must_create_index,
})),
BatchKind::DocumentImport { BatchKind::DocumentImport {
method, method,
import_ids, import_ids,
@ -212,8 +230,8 @@ impl IndexScheduler {
} }
} }
Ok(Some(Batch::IndexOperation( Ok(Some(Batch::IndexOperation {
IndexOperation::DocumentImport { op: IndexOperation::DocumentImport {
index_uid, index_uid,
primary_key, primary_key,
method, method,
@ -222,7 +240,8 @@ impl IndexScheduler {
content_files, content_files,
tasks, tasks,
}, },
))) must_create_index,
}))
} }
BatchKind::DocumentDeletion { deletion_ids } => { BatchKind::DocumentDeletion { deletion_ids } => {
let tasks = self.get_existing_tasks(rtxn, deletion_ids)?; let tasks = self.get_existing_tasks(rtxn, deletion_ids)?;
@ -237,13 +256,14 @@ impl IndexScheduler {
} }
} }
Ok(Some(Batch::IndexOperation( Ok(Some(Batch::IndexOperation {
IndexOperation::DocumentDeletion { op: IndexOperation::DocumentDeletion {
index_uid, index_uid,
documents, documents,
tasks, tasks,
}, },
))) must_create_index,
}))
} }
BatchKind::Settings { BatchKind::Settings {
settings_ids, settings_ids,
@ -263,12 +283,15 @@ impl IndexScheduler {
} }
} }
Ok(Some(Batch::IndexOperation(IndexOperation::Settings { Ok(Some(Batch::IndexOperation {
op: IndexOperation::Settings {
index_uid, index_uid,
settings, settings,
allow_index_creation, allow_index_creation,
tasks, tasks,
}))) },
must_create_index,
}))
} }
BatchKind::ClearAndSettings { BatchKind::ClearAndSettings {
other, other,
@ -283,15 +306,20 @@ impl IndexScheduler {
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
}, },
must_create_index,
)? )?
.unwrap() .unwrap()
{ {
Batch::IndexOperation(IndexOperation::Settings { Batch::IndexOperation {
op:
IndexOperation::Settings {
index_uid, index_uid,
settings, settings,
tasks, tasks,
.. ..
}) => (index_uid, settings, tasks), },
..
} => (index_uid, settings, tasks),
_ => unreachable!(), _ => unreachable!(),
}; };
let (index_uid, cleared_tasks) = match self let (index_uid, cleared_tasks) = match self
@ -299,24 +327,27 @@ impl IndexScheduler {
rtxn, rtxn,
index_uid, index_uid,
BatchKind::DocumentClear { ids: other }, BatchKind::DocumentClear { ids: other },
must_create_index,
)? )?
.unwrap() .unwrap()
{ {
Batch::IndexOperation(IndexOperation::DocumentClear { index_uid, tasks }) => { Batch::IndexOperation {
(index_uid, tasks) op: IndexOperation::DocumentClear { index_uid, tasks },
} ..
} => (index_uid, tasks),
_ => unreachable!(), _ => unreachable!(),
}; };
Ok(Some(Batch::IndexOperation( Ok(Some(Batch::IndexOperation {
IndexOperation::DocumentClearAndSetting { op: IndexOperation::DocumentClearAndSetting {
index_uid, index_uid,
cleared_tasks, cleared_tasks,
allow_index_creation, allow_index_creation,
settings, settings,
settings_tasks, settings_tasks,
}, },
))) must_create_index,
}))
} }
BatchKind::SettingsAndDocumentImport { BatchKind::SettingsAndDocumentImport {
settings_ids, settings_ids,
@ -331,6 +362,7 @@ impl IndexScheduler {
settings_ids, settings_ids,
allow_index_creation, allow_index_creation,
}, },
must_create_index,
)?; )?;
let document_import = self.create_next_batch_index( let document_import = self.create_next_batch_index(
@ -341,24 +373,33 @@ impl IndexScheduler {
allow_index_creation, allow_index_creation,
import_ids, import_ids,
}, },
must_create_index,
)?; )?;
match (document_import, settings) { match (document_import, settings) {
( (
Some(Batch::IndexOperation(IndexOperation::DocumentImport { Some(Batch::IndexOperation {
op:
IndexOperation::DocumentImport {
primary_key, primary_key,
documents_counts, documents_counts,
content_files, content_files,
tasks: document_import_tasks, tasks: document_import_tasks,
.. ..
})), },
Some(Batch::IndexOperation(IndexOperation::Settings { ..
}),
Some(Batch::IndexOperation {
op:
IndexOperation::Settings {
settings, settings,
tasks: settings_tasks, tasks: settings_tasks,
.. ..
})), },
) => Ok(Some(Batch::IndexOperation( ..
IndexOperation::SettingsAndDocumentImport { }),
) => Ok(Some(Batch::IndexOperation {
op: IndexOperation::SettingsAndDocumentImport {
index_uid, index_uid,
primary_key, primary_key,
method, method,
@ -369,7 +410,8 @@ impl IndexScheduler {
settings, settings,
settings_tasks, settings_tasks,
}, },
))), must_create_index,
})),
_ => unreachable!(), _ => unreachable!(),
} }
} }
@ -466,6 +508,7 @@ impl IndexScheduler {
// AT LEAST one index. We can use the right or left one it doesn't // AT LEAST one index. We can use the right or left one it doesn't
// matter. // matter.
let index_name = task.indexes().unwrap()[0]; let index_name = task.indexes().unwrap()[0];
let index_already_exists = self.index_mapper.exists(rtxn, &index_name)?;
let index_tasks = self.index_tasks(rtxn, index_name)? & enqueued; let index_tasks = self.index_tasks(rtxn, index_name)? & enqueued;
@ -486,8 +529,15 @@ impl IndexScheduler {
}) })
.collect::<Result<Vec<_>>>()?; .collect::<Result<Vec<_>>>()?;
if let Some(batchkind) = crate::autobatcher::autobatch(enqueued) { if let Some((batchkind, create_index)) =
return self.create_next_batch_index(rtxn, index_name.to_string(), batchkind); crate::autobatcher::autobatch(enqueued, index_already_exists)
{
return self.create_next_batch_index(
rtxn,
index_name.to_string(),
batchkind,
create_index,
);
} }
} }
@ -679,20 +729,12 @@ impl IndexScheduler {
task.status = Status::Succeeded; task.status = Status::Succeeded;
Ok(vec![task]) Ok(vec![task])
} }
Batch::IndexOperation(operation) => { Batch::IndexOperation {
#[rustfmt::skip] op,
let index = match operation { must_create_index,
IndexOperation::DocumentDeletion { ref index_uid, .. } } => {
| IndexOperation::DocumentClear { ref index_uid, .. } => { let index_uid = op.index_uid();
// only get the index, don't create it let index = if must_create_index {
let rtxn = self.env.read_txn()?;
self.index_mapper.index(&rtxn, index_uid)?
}
IndexOperation::DocumentImport { ref index_uid, allow_index_creation, .. }
| IndexOperation::Settings { ref index_uid, allow_index_creation, .. }
| IndexOperation::DocumentClearAndSetting { ref index_uid, allow_index_creation, .. }
| IndexOperation::SettingsAndDocumentImport {ref index_uid, allow_index_creation, .. } => {
if allow_index_creation {
// create the index if it doesn't already exist // create the index if it doesn't already exist
let mut wtxn = self.env.write_txn()?; let mut wtxn = self.env.write_txn()?;
let index = self.index_mapper.create_index(&mut wtxn, index_uid)?; let index = self.index_mapper.create_index(&mut wtxn, index_uid)?;
@ -701,12 +743,10 @@ impl IndexScheduler {
} else { } else {
let rtxn = self.env.read_txn()?; let rtxn = self.env.read_txn()?;
self.index_mapper.index(&rtxn, index_uid)? self.index_mapper.index(&rtxn, index_uid)?
}
}
}; };
let mut index_wtxn = index.write_txn()?; let mut index_wtxn = index.write_txn()?;
let tasks = self.apply_index_operation(&mut index_wtxn, &index, operation)?; let tasks = self.apply_index_operation(&mut index_wtxn, &index, op)?;
index_wtxn.commit()?; index_wtxn.commit()?;
Ok(tasks) Ok(tasks)

4
index-scheduler/src/index_mapper.rs
View File

@ -130,6 +130,10 @@ impl IndexMapper {
Ok(()) Ok(())
} }
pub fn exists(&self, rtxn: &RoTxn, name: &str) -> Result<bool> {
Ok(self.index_mapping.get(rtxn, name)?.is_some())
}
/// Return an index, may open it if it wasn't already opened. /// Return an index, may open it if it wasn't already opened.
pub fn index(&self, rtxn: &RoTxn, name: &str) -> Result<Index> { pub fn index(&self, rtxn: &RoTxn, name: &str) -> Result<Index> {
let uuid = self let uuid = self