From cc461b1331b43c22e23ebec4c8bcb7398f15d076 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cl=C3=A9ment=20Renault?= Date: Thu, 31 Oct 2019 11:13:37 +0100 Subject: [PATCH] Display enqueued along with processed updates --- meilidb-core/examples/from_file.rs | 4 +- meilidb-core/src/database.rs | 2 +- meilidb-core/src/lib.rs | 2 +- meilidb-core/src/store/mod.rs | 30 ++++++++---- meilidb-core/src/store/updates.rs | 4 +- meilidb-core/src/store/updates_results.rs | 14 ++++-- meilidb-core/src/update/mod.rs | 56 ++++++++++++++++++++--- 7 files changed, 85 insertions(+), 27 deletions(-) diff --git a/meilidb-core/examples/from_file.rs b/meilidb-core/examples/from_file.rs index 8ba7dd1d5..a6b20da0f 100644 --- a/meilidb-core/examples/from_file.rs +++ b/meilidb-core/examples/from_file.rs @@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize}; use structopt::StructOpt; use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor}; -use meilidb_core::{Database, Highlight, UpdateResult}; +use meilidb_core::{Database, Highlight, ProcessedUpdateResult}; use meilidb_schema::SchemaAttr; const INDEX_NAME: &str = "default"; @@ -97,7 +97,7 @@ fn index_command(command: IndexCommand, database: Database) -> Result<(), Box index, None => database.create_index(INDEX_NAME).unwrap(), diff --git a/meilidb-core/src/database.rs b/meilidb-core/src/database.rs index 45ddb56b7..416b806a4 100644 --- a/meilidb-core/src/database.rs +++ b/meilidb-core/src/database.rs @@ -11,7 +11,7 @@ use log::{debug, error}; use crate::{store, update, Index, MResult}; -pub type BoxUpdateFn = Box; +pub type BoxUpdateFn = Box; type ArcSwapFn = arc_swap::ArcSwapOption; pub struct Database { diff --git a/meilidb-core/src/lib.rs b/meilidb-core/src/lib.rs index 9a9c3ed80..f2cc31db4 100644 --- a/meilidb-core/src/lib.rs +++ b/meilidb-core/src/lib.rs @@ -24,7 +24,7 @@ pub use self::number::{Number, ParseNumberError}; pub use self::ranked_map::RankedMap; pub use self::raw_document::RawDocument; pub use self::store::Index; -pub use self::update::{UpdateResult, UpdateStatus, UpdateType}; +pub use self::update::{EnqueuedUpdateResult, ProcessedUpdateResult, UpdateStatus, UpdateType}; use ::serde::{Deserialize, Serialize}; use zerocopy::{AsBytes, FromBytes}; diff --git a/meilidb-core/src/store/mod.rs b/meilidb-core/src/store/mod.rs index 3198f455f..a8ea439ed 100644 --- a/meilidb-core/src/store/mod.rs +++ b/meilidb-core/src/store/mod.rs @@ -219,17 +219,29 @@ impl Index { } pub fn all_updates_status(&self, reader: &heed::RoTxn) -> MResult> { - match self.updates_results.last_update_id(reader)? { - Some((last_id, _)) => { - let mut updates = Vec::with_capacity(last_id as usize + 1); - for id in 0..=last_id { - let update = self.update_status(reader, id)?; - updates.push(update); - } - Ok(updates) + let mut updates = Vec::new(); + let mut last_update_result_id = 0; + + // retrieve all updates results + if let Some((last_id, _)) = self.updates_results.last_update_id(reader)? { + updates.reserve(last_id as usize); + + for id in 0..=last_id { + let update = self.update_status(reader, id)?; + updates.push(update); + last_update_result_id = id; } - None => Ok(Vec::new()), } + + // retrieve all enqueued updates + if let Some((last_id, _)) = self.updates.last_update_id(reader)? { + for id in last_update_result_id + 1..last_id { + let update = self.update_status(reader, id)?; + updates.push(update); + } + } + + Ok(updates) } pub fn query_builder(&self) -> QueryBuilder { diff --git a/meilidb-core/src/store/updates.rs b/meilidb-core/src/store/updates.rs index e4e634dcd..4b3cdc51c 100644 --- a/meilidb-core/src/store/updates.rs +++ b/meilidb-core/src/store/updates.rs @@ -26,9 +26,9 @@ impl Updates { } // TODO do not trigger deserialize if possible - pub fn contains(self, reader: &heed::RoTxn, update_id: u64) -> ZResult { + pub fn get(self, reader: &heed::RoTxn, update_id: u64) -> ZResult> { let update_id = BEU64::new(update_id); - self.updates.get(reader, &update_id).map(|v| v.is_some()) + self.updates.get(reader, &update_id) } pub fn put_update( diff --git a/meilidb-core/src/store/updates_results.rs b/meilidb-core/src/store/updates_results.rs index 51db65cc1..0a9d370a9 100644 --- a/meilidb-core/src/store/updates_results.rs +++ b/meilidb-core/src/store/updates_results.rs @@ -1,15 +1,19 @@ use super::BEU64; -use crate::update::UpdateResult; +use crate::update::ProcessedUpdateResult; use heed::types::{OwnedType, SerdeBincode}; use heed::Result as ZResult; #[derive(Copy, Clone)] pub struct UpdatesResults { - pub(crate) updates_results: heed::Database, SerdeBincode>, + pub(crate) updates_results: + heed::Database, SerdeBincode>, } impl UpdatesResults { - pub fn last_update_id(self, reader: &heed::RoTxn) -> ZResult> { + pub fn last_update_id( + self, + reader: &heed::RoTxn, + ) -> ZResult> { match self.updates_results.last(reader)? { Some((key, data)) => Ok(Some((key.get(), data))), None => Ok(None), @@ -20,7 +24,7 @@ impl UpdatesResults { self, writer: &mut heed::RwTxn, update_id: u64, - update_result: &UpdateResult, + update_result: &ProcessedUpdateResult, ) -> ZResult<()> { let update_id = BEU64::new(update_id); self.updates_results.put(writer, &update_id, update_result) @@ -30,7 +34,7 @@ impl UpdatesResults { self, reader: &heed::RoTxn, update_id: u64, - ) -> ZResult> { + ) -> ZResult> { let update_id = BEU64::new(update_id); self.updates_results.get(reader, &update_id) } diff --git a/meilidb-core/src/update/mod.rs b/meilidb-core/src/update/mod.rs index 755df56c0..360539fe2 100644 --- a/meilidb-core/src/update/mod.rs +++ b/meilidb-core/src/update/mod.rs @@ -42,6 +42,36 @@ pub enum Update { StopWordsDeletion(BTreeSet), } +impl Update { + pub fn update_type(&self) -> UpdateType { + match self { + Update::ClearAll => UpdateType::ClearAll, + Update::Schema(schema) => UpdateType::Schema { + schema: schema.clone(), + }, + Update::Customs(_) => UpdateType::Customs, + Update::DocumentsAddition(addition) => UpdateType::DocumentsAddition { + number: addition.len(), + }, + Update::DocumentsDeletion(deletion) => UpdateType::DocumentsDeletion { + number: deletion.len(), + }, + Update::SynonymsAddition(addition) => UpdateType::SynonymsAddition { + number: addition.len(), + }, + Update::SynonymsDeletion(deletion) => UpdateType::SynonymsDeletion { + number: deletion.len(), + }, + Update::StopWordsAddition(addition) => UpdateType::StopWordsAddition { + number: addition.len(), + }, + Update::StopWordsDeletion(deletion) => UpdateType::StopWordsDeletion { + number: deletion.len(), + }, + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub enum UpdateType { ClearAll, @@ -61,17 +91,23 @@ pub struct DetailedDuration { } #[derive(Debug, Clone, Serialize, Deserialize)] -pub struct UpdateResult { +pub struct ProcessedUpdateResult { pub update_id: u64, pub update_type: UpdateType, pub result: Result<(), String>, pub detailed_duration: DetailedDuration, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct EnqueuedUpdateResult { + pub update_id: u64, + pub update_type: UpdateType, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub enum UpdateStatus { - Enqueued, - Processed(UpdateResult), + Enqueued(EnqueuedUpdateResult), + Processed(ProcessedUpdateResult), Unknown, } @@ -84,8 +120,11 @@ pub fn update_status( match updates_results_store.update_result(reader, update_id)? { Some(result) => Ok(UpdateStatus::Processed(result)), None => { - if updates_store.contains(reader, update_id)? { - Ok(UpdateStatus::Enqueued) + if let Some(update) = updates_store.get(reader, update_id)? { + Ok(UpdateStatus::Enqueued(EnqueuedUpdateResult { + update_id, + update_type: update.update_type(), + })) } else { Ok(UpdateStatus::Unknown) } @@ -110,7 +149,10 @@ pub fn next_update_id( Ok(new_update_id) } -pub fn update_task(writer: &mut heed::RwTxn, index: store::Index) -> MResult> { +pub fn update_task( + writer: &mut heed::RwTxn, + index: store::Index, +) -> MResult> { let (update_id, update) = match index.updates.pop_front(writer)? { Some(value) => value, None => return Ok(None), @@ -259,7 +301,7 @@ pub fn update_task(writer: &mut heed::RwTxn, index: store::Index) -> MResult