mirror of
https://github.com/meilisearch/MeiliSearch
synced 2024-12-24 13:40:31 +01:00
make retrieval non blocking
This commit is contained in:
parent
6766de437f
commit
8bb1b6146f
@ -166,80 +166,92 @@ impl Data {
|
||||
}
|
||||
}
|
||||
|
||||
pub fn retrieve_documents(
|
||||
pub async fn retrieve_documents<S>(
|
||||
&self,
|
||||
index: impl AsRef<str>,
|
||||
index: impl AsRef<str> + Send + Sync + 'static,
|
||||
offset: usize,
|
||||
limit: usize,
|
||||
attributes_to_retrieve: Option<&[&str]>,
|
||||
) -> anyhow::Result<Vec<Map<String, Value>>> {
|
||||
let index = self.index_controller
|
||||
.index(&index)?
|
||||
.with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
|
||||
attributes_to_retrieve: Option<Vec<S>>,
|
||||
) -> anyhow::Result<Vec<Map<String, Value>>>
|
||||
where
|
||||
S: AsRef<str> + Send + Sync + 'static
|
||||
{
|
||||
let index_controller = self.index_controller.clone();
|
||||
let documents: anyhow::Result<_> = tokio::task::spawn_blocking(move || {
|
||||
let index = index_controller
|
||||
.index(&index)?
|
||||
.with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
|
||||
|
||||
let txn = index.read_txn()?;
|
||||
let txn = index.read_txn()?;
|
||||
|
||||
let fields_ids_map = index.fields_ids_map(&txn)?;
|
||||
let fields_ids_map = index.fields_ids_map(&txn)?;
|
||||
|
||||
let attributes_to_retrieve_ids = match attributes_to_retrieve {
|
||||
Some(attrs) => attrs
|
||||
.as_ref()
|
||||
.iter()
|
||||
.filter_map(|f| fields_ids_map.id(f))
|
||||
.collect::<Vec<_>>(),
|
||||
None => fields_ids_map.iter().map(|(id, _)| id).collect(),
|
||||
};
|
||||
let attributes_to_retrieve_ids = match attributes_to_retrieve {
|
||||
Some(attrs) => attrs
|
||||
.iter()
|
||||
.filter_map(|f| fields_ids_map.id(f.as_ref()))
|
||||
.collect::<Vec<_>>(),
|
||||
None => fields_ids_map.iter().map(|(id, _)| id).collect(),
|
||||
};
|
||||
|
||||
let iter = index.documents.range(&txn, &(..))?
|
||||
.skip(offset)
|
||||
.take(limit);
|
||||
let iter = index.documents.range(&txn, &(..))?
|
||||
.skip(offset)
|
||||
.take(limit);
|
||||
|
||||
let mut documents = Vec::new();
|
||||
let mut documents = Vec::new();
|
||||
|
||||
for entry in iter {
|
||||
let (_id, obkv) = entry?;
|
||||
let object = obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, obkv)?;
|
||||
documents.push(object);
|
||||
}
|
||||
for entry in iter {
|
||||
let (_id, obkv) = entry?;
|
||||
let object = obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, obkv)?;
|
||||
documents.push(object);
|
||||
}
|
||||
|
||||
Ok(documents)
|
||||
Ok(documents)
|
||||
}).await?;
|
||||
documents
|
||||
}
|
||||
|
||||
pub fn retrieve_document(
|
||||
pub async fn retrieve_document<S>(
|
||||
&self,
|
||||
index: impl AsRef<str>,
|
||||
document_id: impl AsRef<str>,
|
||||
attributes_to_retrieve: Option<&[&str]>,
|
||||
) -> anyhow::Result<Map<String, Value>> {
|
||||
let index = self.index_controller
|
||||
.index(&index)?
|
||||
.with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
|
||||
let txn = index.read_txn()?;
|
||||
index: impl AsRef<str> + Sync + Send + 'static,
|
||||
document_id: impl AsRef<str> + Sync + Send + 'static,
|
||||
attributes_to_retrieve: Option<Vec<S>>,
|
||||
) -> anyhow::Result<Map<String, Value>>
|
||||
where
|
||||
S: AsRef<str> + Sync + Send + 'static,
|
||||
{
|
||||
let index_controller = self.index_controller.clone();
|
||||
let document: anyhow::Result<_> = tokio::task::spawn_blocking(move || {
|
||||
let index = index_controller
|
||||
.index(&index)?
|
||||
.with_context(|| format!("Index {:?} doesn't exist", index.as_ref()))?;
|
||||
let txn = index.read_txn()?;
|
||||
|
||||
let fields_ids_map = index.fields_ids_map(&txn)?;
|
||||
let fields_ids_map = index.fields_ids_map(&txn)?;
|
||||
|
||||
let attributes_to_retrieve_ids = match attributes_to_retrieve {
|
||||
Some(attrs) => attrs
|
||||
.as_ref()
|
||||
.iter()
|
||||
.filter_map(|f| fields_ids_map.id(f))
|
||||
.collect::<Vec<_>>(),
|
||||
None => fields_ids_map.iter().map(|(id, _)| id).collect(),
|
||||
};
|
||||
let attributes_to_retrieve_ids = match attributes_to_retrieve {
|
||||
Some(attrs) => attrs
|
||||
.iter()
|
||||
.filter_map(|f| fields_ids_map.id(f.as_ref()))
|
||||
.collect::<Vec<_>>(),
|
||||
None => fields_ids_map.iter().map(|(id, _)| id).collect(),
|
||||
};
|
||||
|
||||
let internal_id = index
|
||||
.external_documents_ids(&txn)?
|
||||
.get(document_id.as_ref().as_bytes())
|
||||
.with_context(|| format!("Document with id {} not found", document_id.as_ref()))?;
|
||||
let internal_id = index
|
||||
.external_documents_ids(&txn)?
|
||||
.get(document_id.as_ref().as_bytes())
|
||||
.with_context(|| format!("Document with id {} not found", document_id.as_ref()))?;
|
||||
|
||||
let document = index.documents(&txn, std::iter::once(internal_id))?
|
||||
.into_iter()
|
||||
.next()
|
||||
.map(|(_, d)| d);
|
||||
let document = index.documents(&txn, std::iter::once(internal_id))?
|
||||
.into_iter()
|
||||
.next()
|
||||
.map(|(_, d)| d);
|
||||
|
||||
match document {
|
||||
Some(document) => Ok(obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, document)?),
|
||||
None => bail!("Document with id {} not found", document_id.as_ref()),
|
||||
}
|
||||
match document {
|
||||
Some(document) => Ok(obkv_to_json(&attributes_to_retrieve_ids, &fields_ids_map, document)?),
|
||||
None => bail!("Document with id {} not found", document_id.as_ref()),
|
||||
}
|
||||
}).await?;
|
||||
document
|
||||
}
|
||||
}
|
||||
|
@ -55,9 +55,9 @@ async fn get_document(
|
||||
data: web::Data<Data>,
|
||||
path: web::Path<DocumentParam>,
|
||||
) -> Result<HttpResponse, ResponseError> {
|
||||
let index = &path.index_uid;
|
||||
let id = &path.document_id;
|
||||
match data.retrieve_document(index, id, None) {
|
||||
let index = path.index_uid.clone();
|
||||
let id = path.document_id.clone();
|
||||
match data.retrieve_document(index, id, None as Option<Vec<String>>).await {
|
||||
Ok(document) => {
|
||||
let json = serde_json::to_string(&document).unwrap();
|
||||
Ok(HttpResponse::Ok().body(json))
|
||||
@ -99,13 +99,14 @@ async fn get_all_documents(
|
||||
.as_ref()
|
||||
.map(|attrs| attrs
|
||||
.split(",")
|
||||
.map(String::from)
|
||||
.collect::<Vec<_>>());
|
||||
|
||||
match data.retrieve_documents(
|
||||
&path.index_uid,
|
||||
path.index_uid.clone(),
|
||||
params.offset.unwrap_or(DEFAULT_RETRIEVE_DOCUMENTS_OFFSET),
|
||||
params.limit.unwrap_or(DEFAULT_RETRIEVE_DOCUMENTS_LIMIT),
|
||||
attributes_to_retrieve.as_deref()) {
|
||||
attributes_to_retrieve).await {
|
||||
Ok(docs) => {
|
||||
let json = serde_json::to_string(&docs).unwrap();
|
||||
Ok(HttpResponse::Ok().body(json))
|
||||
|
Loading…
x
Reference in New Issue
Block a user