implement create index

This commit is contained in:
mpostma 2021-02-08 10:47:34 +01:00
parent ed44e684cc
commit ec047eefd2
No known key found for this signature in database
GPG Key ID: CBC8A7C1D7A28C3A
5 changed files with 69 additions and 22 deletions

View File

@ -125,6 +125,11 @@ impl Data {
.find(|i| i.name == name.as_ref()))
}
pub fn create_index(&self, name: impl AsRef<str>, primary_key: Option<impl AsRef<str>>) -> anyhow::Result<IndexMetadata> {
let meta = self.index_controller.create_index(name, primary_key)?;
Ok(meta)
}
#[inline]
pub fn http_payload_size_limit(&self) -> usize {
self.options.http_payload_size_limit.get_bytes() as usize

View File

@ -2,7 +2,7 @@ use std::fs::{create_dir_all, remove_dir_all};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::bail;
use anyhow::{bail, Context};
use chrono::{DateTime, Utc};
use dashmap::{DashMap, mapref::entry::Entry};
use heed::{Env, EnvOpenOptions, Database, types::{Str, SerdeJson, ByteSlice}, RoTxn, RwTxn};
@ -142,14 +142,14 @@ impl IndexStore {
Some(res) => Ok(res),
None => {
let uuid = Uuid::new_v4();
let result = self.create_index_txn(&mut txn, uuid, name, update_size, index_size)?;
let (index, updates, _) = self.create_index_txn(&mut txn, uuid, name, update_size, index_size)?;
// If we fail to commit the transaction, we must delete the database from the
// file-system.
if let Err(e) = txn.commit() {
self.clean_db(uuid);
return Err(e)?;
}
Ok(result)
Ok((index, updates))
},
}
}
@ -171,7 +171,7 @@ impl IndexStore {
name: impl AsRef<str>,
update_store_size: u64,
index_store_size: u64,
) -> anyhow::Result<(Arc<Index>, Arc<UpdateStore>)> {
) -> anyhow::Result<(Arc<Index>, Arc<UpdateStore>, IndexMeta)> {
let created_at = Utc::now();
let meta = IndexMeta { update_store_size, index_store_size, uuid: uuid.clone(), created_at };
@ -189,7 +189,7 @@ impl IndexStore {
self.uuid_to_index.insert(uuid, (index.clone(), update_store.clone()));
Ok((index, update_store))
Ok((index, update_store, meta))
}
/// Same as `get_or_create`, but returns an error if the index already exists.
@ -198,7 +198,7 @@ impl IndexStore {
name: impl AsRef<str>,
update_size: u64,
index_size: u64,
) -> anyhow::Result<(Arc<Index>, Arc<UpdateStore>)> {
) -> anyhow::Result<(Arc<Index>, Arc<UpdateStore>, IndexMeta)> {
let uuid = Uuid::new_v4();
let mut txn = self.env.write_txn()?;
@ -216,8 +216,10 @@ impl IndexStore {
Ok(result)
}
/// Returns each index associated with it's metadata;
pub fn list_indexes(&self) -> anyhow::Result<Vec<(String, IndexMeta)>> {
/// Returns each index associated with its metadata:
/// (index_name, IndexMeta, primary_key)
/// This method will force all the indexes to be loaded.
pub fn list_indexes(&self) -> anyhow::Result<Vec<(String, IndexMeta, Option<String>)>> {
let txn = self.env.read_txn()?;
let metas = self.name_to_uuid
.iter(&txn)?
@ -225,10 +227,16 @@ impl IndexStore {
.map_err(|e| { error!("error decoding entry while listing indexes: {}", e); e }).ok());
let mut indexes = Vec::new();
for (name, uuid) in metas {
// get index to retrieve primary key
let (index, _) = self.get_index_txn(&txn, name)?
.with_context(|| format!("could not load index {:?}", name))?;
let primary_key = index.primary_key(&index.read_txn()?)?
.map(String::from);
// retieve meta
let meta = self.uuid_to_index_meta
.get(&txn, &uuid)?
.unwrap_or_else(|| panic!("corrupted database, index {} should exist.", name));
indexes.push((name.to_owned(), meta));
.with_context(|| format!("could not retieve meta for index {:?}", name))?;
indexes.push((name.to_owned(), meta, primary_key));
}
Ok(indexes)
}

View File

@ -5,7 +5,7 @@ mod update_handler;
use std::path::Path;
use std::sync::Arc;
use anyhow::bail;
use anyhow::{bail, Context};
use itertools::Itertools;
use milli::Index;
@ -58,9 +58,24 @@ impl IndexController for LocalIndexController {
Ok(pending.into())
}
fn create_index<S: AsRef<str>>(&self, index_uid: S) -> anyhow::Result<()> {
self.indexes.create_index(index_uid, self.update_db_size, self.index_db_size)?;
Ok(())
fn create_index(&self, index_name: impl AsRef<str>, primary_key: Option<impl AsRef<str>>) -> anyhow::Result<IndexMetadata> {
let (index, _, meta) = self.indexes.create_index(&index_name, self.update_db_size, self.index_db_size)?;
if let Some(ref primary_key) = primary_key {
if let Err(e) = update_primary_key(index, primary_key).context("error creating index") {
// TODO: creating index could not be completed, delete everything.
Err(e)?
}
}
let meta = IndexMetadata {
name: index_name.as_ref().to_owned(),
uuid: meta.uuid.clone(),
created_at: meta.created_at,
updated_at: meta.created_at,
primary_key: primary_key.map(|n| n.as_ref().to_owned()),
};
Ok(meta)
}
fn delete_index<S: AsRef<str>>(&self, _index_uid: S) -> anyhow::Result<()> {
@ -107,7 +122,7 @@ impl IndexController for LocalIndexController {
fn list_indexes(&self) -> anyhow::Result<Vec<IndexMetadata>> {
let metas = self.indexes.list_indexes()?;
let mut output_meta = Vec::new();
for (name, meta) in metas {
for (name, meta, primary_key) in metas {
let created_at = meta.created_at;
let uuid = meta.uuid;
let updated_at = self
@ -122,7 +137,7 @@ impl IndexController for LocalIndexController {
created_at,
updated_at,
uuid,
primary_key: None,
primary_key,
};
output_meta.push(index_meta);
}
@ -130,6 +145,17 @@ impl IndexController for LocalIndexController {
}
}
fn update_primary_key(index: impl AsRef<Index>, primary_key: impl AsRef<str>) -> anyhow::Result<()> {
let index = index.as_ref();
let mut txn = index.write_txn()?;
if index.primary_key(&txn)?.is_some() {
bail!("primary key already set.")
}
index.put_primary_key(&mut txn, primary_key.as_ref())?;
txn.commit()?;
Ok(())
}
#[cfg(test)]
mod test {
use super::*;

View File

@ -127,7 +127,7 @@ pub trait IndexController {
fn update_settings<S: AsRef<str>>(&self, index_uid: S, settings: Settings) -> anyhow::Result<UpdateStatus>;
/// Create an index with the given `index_uid`.
fn create_index<S: AsRef<str>>(&self, index_uid: S) -> Result<()>;
fn create_index(&self, index_uid: impl AsRef<str>, primary_key: Option<impl AsRef<str>>) -> Result<IndexMetadata>;
/// Delete index with the given `index_uid`, attempting to close it beforehand.
fn delete_index<S: AsRef<str>>(&self, index_uid: S) -> Result<()>;

View File

@ -54,17 +54,25 @@ async fn get_index(
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct IndexCreateRequest {
name: Option<String>,
uid: Option<String>,
name: String,
primary_key: Option<String>,
}
#[post("/indexes", wrap = "Authentication::Private")]
async fn create_index(
_data: web::Data<Data>,
_body: web::Json<IndexCreateRequest>,
data: web::Data<Data>,
body: web::Json<IndexCreateRequest>,
) -> Result<HttpResponse, ResponseError> {
todo!()
match data.create_index(&body.name, body.primary_key.clone()) {
Ok(meta) => {
let json = serde_json::to_string(&meta).unwrap();
Ok(HttpResponse::Ok().body(json))
}
Err(e) => {
error!("error creating index: {}", e);
unimplemented!()
}
}
}
#[derive(Debug, Deserialize)]