mirror of
https://github.com/meilisearch/MeiliSearch
synced 2024-11-26 23:04:26 +01:00
Merge pull request #208 from meilisearch/improve-open-or-create-index
Create two open and create index functions
This commit is contained in:
commit
c5af5de4f0
@ -12,7 +12,7 @@ use serde::{Serialize, Deserialize};
|
|||||||
use structopt::StructOpt;
|
use structopt::StructOpt;
|
||||||
use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor};
|
use termcolor::{Color, ColorChoice, ColorSpec, StandardStream, WriteColor};
|
||||||
|
|
||||||
use meilidb_core::{Highlight, Database, UpdateResult, BoxUpdateFn};
|
use meilidb_core::{Highlight, Database, UpdateResult};
|
||||||
use meilidb_schema::SchemaAttr;
|
use meilidb_schema::SchemaAttr;
|
||||||
|
|
||||||
const INDEX_NAME: &str = "default";
|
const INDEX_NAME: &str = "default";
|
||||||
@ -86,7 +86,14 @@ fn index_command(command: IndexCommand, database: Database) -> Result<(), Box<dy
|
|||||||
|
|
||||||
let (sender, receiver) = mpsc::sync_channel(100);
|
let (sender, receiver) = mpsc::sync_channel(100);
|
||||||
let update_fn = move |update: UpdateResult| sender.send(update.update_id).unwrap();
|
let update_fn = move |update: UpdateResult| sender.send(update.update_id).unwrap();
|
||||||
let index = database.open_index(INDEX_NAME, Some(Box::new(update_fn)))?;
|
let index = match database.open_index(INDEX_NAME) {
|
||||||
|
Some(index) => index,
|
||||||
|
None => database.create_index(INDEX_NAME).unwrap()
|
||||||
|
};
|
||||||
|
|
||||||
|
let done = database.set_update_callback(INDEX_NAME, Box::new(update_fn));
|
||||||
|
assert!(done, "could not set the index update function");
|
||||||
|
|
||||||
let rkv = database.rkv.read().unwrap();
|
let rkv = database.rkv.read().unwrap();
|
||||||
|
|
||||||
let schema = {
|
let schema = {
|
||||||
@ -256,8 +263,7 @@ fn crop_text(
|
|||||||
|
|
||||||
fn search_command(command: SearchCommand, database: Database) -> Result<(), Box<dyn Error>> {
|
fn search_command(command: SearchCommand, database: Database) -> Result<(), Box<dyn Error>> {
|
||||||
let rkv = database.rkv.read().unwrap();
|
let rkv = database.rkv.read().unwrap();
|
||||||
let update_fn = None as Option::<BoxUpdateFn>;
|
let index = database.open_index(INDEX_NAME).expect("Could not find index");
|
||||||
let index = database.open_index(INDEX_NAME, update_fn)?;
|
|
||||||
let reader = rkv.read().unwrap();
|
let reader = rkv.read().unwrap();
|
||||||
|
|
||||||
let schema = index.main.schema(&reader)?;
|
let schema = index.main.schema(&reader)?;
|
||||||
|
@ -1,4 +1,4 @@
|
|||||||
use std::collections::HashMap;
|
use std::collections::hash_map::{HashMap, Entry};
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::{Arc, RwLock};
|
use std::sync::{Arc, RwLock};
|
||||||
use std::{fs, thread};
|
use std::{fs, thread};
|
||||||
@ -116,53 +116,64 @@ impl Database {
|
|||||||
Ok(Database { rkv, main_store, indexes_store, indexes: RwLock::new(indexes) })
|
Ok(Database { rkv, main_store, indexes_store, indexes: RwLock::new(indexes) })
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn open_index(
|
pub fn open_index(&self, name: impl AsRef<str>) -> Option<Index> {
|
||||||
&self,
|
|
||||||
name: impl Into<String>,
|
|
||||||
update_fn: Option<BoxUpdateFn>,
|
|
||||||
) -> MResult<Index>
|
|
||||||
{
|
|
||||||
let indexes_lock = self.indexes.read().unwrap();
|
let indexes_lock = self.indexes.read().unwrap();
|
||||||
let name = name.into();
|
match indexes_lock.get(name.as_ref()) {
|
||||||
|
Some((index, ..)) => Some(index.clone()),
|
||||||
|
None => None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
match indexes_lock.get(&name) {
|
pub fn create_index(&self, name: impl AsRef<str>) -> MResult<Index> {
|
||||||
Some((index, old_update_fn, _)) => {
|
let name = name.as_ref();
|
||||||
old_update_fn.swap(update_fn.map(Arc::new));
|
let mut indexes_lock = self.indexes.write().unwrap();
|
||||||
Ok(index.clone())
|
|
||||||
},
|
|
||||||
None => {
|
|
||||||
drop(indexes_lock);
|
|
||||||
|
|
||||||
|
match indexes_lock.entry(name.to_owned()) {
|
||||||
|
Entry::Occupied(_) => Err(crate::Error::IndexAlreadyExists),
|
||||||
|
Entry::Vacant(entry) => {
|
||||||
let rkv_lock = self.rkv.read().unwrap();
|
let rkv_lock = self.rkv.read().unwrap();
|
||||||
let (sender, receiver) = crossbeam_channel::bounded(100);
|
let (sender, receiver) = crossbeam_channel::bounded(100);
|
||||||
let index = store::create(&rkv_lock, &name, sender)?;
|
let index = store::create(&rkv_lock, name, sender)?;
|
||||||
|
|
||||||
let mut writer = rkv_lock.write()?;
|
let mut writer = rkv_lock.write()?;
|
||||||
let value = rkv::Value::Blob(&[]);
|
let value = rkv::Value::Blob(&[]);
|
||||||
self.indexes_store.put(&mut writer, &name, &value)?;
|
self.indexes_store.put(&mut writer, name, &value)?;
|
||||||
|
|
||||||
{
|
|
||||||
let mut indexes_write = self.indexes.write().unwrap();
|
|
||||||
indexes_write.entry(name).or_insert_with(|| {
|
|
||||||
let rkv_clone = self.rkv.clone();
|
let rkv_clone = self.rkv.clone();
|
||||||
let index_clone = index.clone();
|
let index_clone = index.clone();
|
||||||
|
|
||||||
let update_fn = update_fn.map(Arc::new);
|
let no_update_fn = Arc::new(ArcSwapFn::empty());
|
||||||
let update_fn = Arc::new(ArcSwapFn::new(update_fn));
|
let no_update_fn_clone = no_update_fn.clone();
|
||||||
let update_fn_clone = update_fn.clone();
|
|
||||||
|
|
||||||
let handle = thread::spawn(move || {
|
let handle = thread::spawn(move || {
|
||||||
update_awaiter(receiver, rkv_clone, update_fn_clone, index_clone)
|
update_awaiter(receiver, rkv_clone, no_update_fn_clone, index_clone)
|
||||||
});
|
});
|
||||||
|
|
||||||
(index.clone(), update_fn, handle)
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
writer.commit()?;
|
writer.commit()?;
|
||||||
|
entry.insert((index.clone(), no_update_fn, handle));
|
||||||
|
|
||||||
Ok(index)
|
Ok(index)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn set_update_callback(&self, name: impl AsRef<str>, update_fn: BoxUpdateFn) -> bool {
|
||||||
|
let indexes_lock = self.indexes.read().unwrap();
|
||||||
|
match indexes_lock.get(name.as_ref()) {
|
||||||
|
Some((_, current_update_fn, _)) => {
|
||||||
|
let update_fn = Some(Arc::new(update_fn));
|
||||||
|
current_update_fn.swap(update_fn);
|
||||||
|
true
|
||||||
},
|
},
|
||||||
|
None => false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn unset_update_callback(&self, name: impl AsRef<str>) -> bool {
|
||||||
|
let indexes_lock = self.indexes.read().unwrap();
|
||||||
|
match indexes_lock.get(name.as_ref()) {
|
||||||
|
Some((_, current_update_fn, _)) => { current_update_fn.swap(None); true },
|
||||||
|
None => false,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -6,6 +6,7 @@ pub type MResult<T> = Result<T, Error>;
|
|||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub enum Error {
|
pub enum Error {
|
||||||
Io(io::Error),
|
Io(io::Error),
|
||||||
|
IndexAlreadyExists,
|
||||||
SchemaDiffer,
|
SchemaDiffer,
|
||||||
SchemaMissing,
|
SchemaMissing,
|
||||||
WordIndexMissing,
|
WordIndexMissing,
|
||||||
@ -79,6 +80,7 @@ impl fmt::Display for Error {
|
|||||||
use self::Error::*;
|
use self::Error::*;
|
||||||
match self {
|
match self {
|
||||||
Io(e) => write!(f, "{}", e),
|
Io(e) => write!(f, "{}", e),
|
||||||
|
IndexAlreadyExists => write!(f, "index already exists"),
|
||||||
SchemaDiffer => write!(f, "schemas differ"),
|
SchemaDiffer => write!(f, "schemas differ"),
|
||||||
SchemaMissing => write!(f, "this index does not have a schema"),
|
SchemaMissing => write!(f, "this index does not have a schema"),
|
||||||
WordIndexMissing => write!(f, "this index does not have a word index"),
|
WordIndexMissing => write!(f, "this index does not have a word index"),
|
||||||
|
@ -551,7 +551,7 @@ mod tests {
|
|||||||
use tempfile::TempDir;
|
use tempfile::TempDir;
|
||||||
|
|
||||||
use crate::automaton::normalize_str;
|
use crate::automaton::normalize_str;
|
||||||
use crate::database::{Database, BoxUpdateFn};
|
use crate::database::Database;
|
||||||
use crate::DocIndex;
|
use crate::DocIndex;
|
||||||
use crate::store::Index;
|
use crate::store::Index;
|
||||||
|
|
||||||
@ -646,8 +646,7 @@ mod tests {
|
|||||||
fn from_iter<I: IntoIterator<Item=(&'a str, &'a [DocIndex])>>(iter: I) -> Self {
|
fn from_iter<I: IntoIterator<Item=(&'a str, &'a [DocIndex])>>(iter: I) -> Self {
|
||||||
let tempdir = TempDir::new().unwrap();
|
let tempdir = TempDir::new().unwrap();
|
||||||
let database = Database::open_or_create(&tempdir).unwrap();
|
let database = Database::open_or_create(&tempdir).unwrap();
|
||||||
let update_fn = None as Option::<BoxUpdateFn>;
|
let index = database.create_index("default").unwrap();
|
||||||
let index = database.open_index("default", update_fn).unwrap();
|
|
||||||
|
|
||||||
let rkv = database.rkv.read().unwrap();
|
let rkv = database.rkv.read().unwrap();
|
||||||
let mut writer = rkv.write().unwrap();
|
let mut writer = rkv.write().unwrap();
|
||||||
|
Loading…
Reference in New Issue
Block a user