implements the index deletion, creation and swap

This commit is contained in:
Tamo 2022-09-07 21:27:06 +02:00 committed by Clément Renault
parent 72b2e68de4
commit 30d2b24689
No known key found for this signature in database
GPG Key ID: 92ADA4E935E71FA4
4 changed files with 70 additions and 7 deletions

View File

@ -5,9 +5,9 @@ use crate::index;
#[derive(Error, Debug)] #[derive(Error, Debug)]
pub enum Error { pub enum Error {
#[error("Index `{}` not found", .0)] #[error("Index `{0}` not found")]
IndexNotFound(String), IndexNotFound(String),
#[error("Index `{}` already exists", .0)] #[error("Index `{0}` already exists")]
IndexAlreadyExists(String), IndexAlreadyExists(String),
#[error("Corrupted task queue.")] #[error("Corrupted task queue.")]
CorruptedTaskQueue, CorruptedTaskQueue,
@ -17,6 +17,8 @@ pub enum Error {
Milli(#[from] milli::Error), Milli(#[from] milli::Error),
#[error("{0}")] #[error("{0}")]
IndexError(#[from] index::error::IndexError), IndexError(#[from] index::error::IndexError),
#[error(transparent)]
IoError(#[from] std::io::Error),
#[error(transparent)] #[error(transparent)]
Anyhow(#[from] anyhow::Error), Anyhow(#[from] anyhow::Error),

View File

@ -22,6 +22,7 @@ use super::{Checked, Settings};
pub type Document = Map<String, Value>; pub type Document = Map<String, Value>;
// @kero, what is this structure? Shouldn't it move entirely to milli?
#[derive(Debug, Serialize, Deserialize, Clone)] #[derive(Debug, Serialize, Deserialize, Clone)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
pub struct IndexMeta { pub struct IndexMeta {
@ -50,6 +51,7 @@ impl IndexMeta {
} }
} }
// @kero Maybe this should be entirely generated somewhere else since it doesn't really concern the index?
#[derive(Serialize, Debug)] #[derive(Serialize, Debug)]
#[serde(rename_all = "camelCase")] #[serde(rename_all = "camelCase")]
pub struct IndexStats { pub struct IndexStats {
@ -105,6 +107,14 @@ impl Index {
self.inner.as_ref().clone().prepare_for_closing(); self.inner.as_ref().clone().prepare_for_closing();
} }
pub fn delete(self) -> Result<()> {
let path = self.path().to_path_buf();
self.inner.as_ref().clone().prepare_for_closing().wait();
std::fs::remove_file(path)?;
Ok(())
}
pub fn stats(&self) -> Result<IndexStats> { pub fn stats(&self) -> Result<IndexStats> {
let rtxn = self.read_txn()?; let rtxn = self.read_txn()?;

View File

@ -137,6 +137,13 @@ pub mod test {
} }
} }
pub fn delete(self) -> Result<()> {
match self {
MockIndex::Real(index) => index.delete(),
MockIndex::Mock(m) => unsafe { m.get("delete").call(()) },
}
}
pub fn perform_search(&self, query: SearchQuery) -> Result<SearchResult> { pub fn perform_search(&self, query: SearchQuery) -> Result<SearchResult> {
match self { match self {
MockIndex::Real(index) => index.perform_search(query), MockIndex::Real(index) => index.perform_search(query),

View File

@ -222,6 +222,14 @@ impl IndexScheduler {
// TODO: TAMO: do this later // TODO: TAMO: do this later
// self.handle_batch_result(res); // self.handle_batch_result(res);
match wtxn.commit() {
Ok(()) => log::info!("A batch of tasks was successfully completed."),
Err(e) => {
log::error!("{}", e);
continue;
}
}
} }
} }
@ -253,15 +261,37 @@ impl IndexScheduler {
} }
self.available_index.put(wtxn, &index_name, &true); self.available_index.put(wtxn, &index_name, &true);
// let index = // TODO: TAMO: give real info to the index
todo!("tamo: once I get index.rs to works"); let index = Index::open(
index_name.to_string(),
index_name.to_string(),
100_000_000,
Arc::default(),
)?;
if let Some(primary_key) = primary_key {
index.update_primary_key(primary_key.to_string())?;
}
self.index_map
.write()
.map_err(|_| Error::CorruptedTaskQueue)?
.insert(index_name.to_string(), index.clone());
} }
KindWithContent::DeleteIndex { index_name } => { KindWithContent::DeleteIndex { index_name } => {
self.index_map.write(); self.index_map.write();
if !self.available_index.delete(wtxn, &index_name)? { if !self.available_index.delete(wtxn, &index_name)? {
return Err(Error::IndexNotFound(index_name.to_string())); return Err(Error::IndexNotFound(index_name.to_string()));
} }
todo!("tamo: once I get index.rs to works"); if let Some(index) = self
.index_map
.write()
.map_err(|_| Error::CorruptedTaskQueue)?
.remove(index_name)
{
index.delete()?;
} else {
// TODO: TAMO: fix the path
std::fs::remove_file(index_name)?;
}
} }
KindWithContent::SwapIndex { lhs, rhs } => { KindWithContent::SwapIndex { lhs, rhs } => {
if !self.available_index.get(wtxn, &lhs)?.unwrap_or(false) { if !self.available_index.get(wtxn, &lhs)?.unwrap_or(false) {
@ -271,12 +301,26 @@ impl IndexScheduler {
return Err(Error::IndexNotFound(rhs.to_string())); return Err(Error::IndexNotFound(rhs.to_string()));
} }
let index_map = self let lhs_bitmap = self.index_tasks.get(wtxn, lhs)?;
let rhs_bitmap = self.index_tasks.get(wtxn, rhs)?;
// the bitmap are lazily created and thus may not exists.
if let Some(bitmap) = rhs_bitmap {
self.index_tasks.put(wtxn, lhs, &bitmap)?;
}
if let Some(bitmap) = lhs_bitmap {
self.index_tasks.put(wtxn, rhs, &bitmap)?;
}
let mut index_map = self
.index_map .index_map
.write() .write()
.map_err(|_| Error::CorruptedTaskQueue)?; .map_err(|_| Error::CorruptedTaskQueue)?;
// index_map.remove. let lhs_index = index_map.remove(lhs).unwrap();
let rhs_index = index_map.remove(rhs).unwrap();
index_map.insert(lhs.to_string(), rhs_index);
index_map.insert(rhs.to_string(), lhs_index);
} }
_ => unreachable!(), _ => unreachable!(),
}, },