filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Database schema optimization - part 9

Browse Source
This commit is contained in:
mdecimus
2025-11-09 19:28:19 +01:00
parent d9e6927606
commit 836bc5b7fd
27 changed files with 973 additions and 600 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -5411,6 +5411,8 @@ dependencies = [
"bytes",
"fallible-iterator 0.2.0",
"postgres-protocol",
"serde_core",
"serde_json",
]
[[package]]

10
crates/jmap/src/calendar_event/query.rs
View File

@@ -304,14 +304,12 @@ impl CalendarEventQuery for Server {
if !expanded_results.is_empty() {
expanded_results.sort_by(|a, b| {
for comparator in comparators {
let ordering = a
.get_property(&comparator.property)
.cmp(b.get_property(&comparator.property));
let ordering = if comparator.is_ascending {
ordering.reverse()
a.get_property(&comparator.property)
.cmp(b.get_property(&comparator.property))
} else {
ordering
b.get_property(&comparator.property)
.cmp(a.get_property(&comparator.property))
};
if ordering != Ordering::Equal {

2
crates/store/Cargo.toml
View File

@@ -33,7 +33,7 @@ num_cpus = { version = "1.17", optional = true }
blake3 = "1.8"
lz4_flex = { version = "0.11", default-features = false }
deadpool-postgres = { version = "0.14", optional = true }
tokio-postgres = { version = "0.7.10", optional = true }
tokio-postgres = { version = "0.7.10", features = ["with-serde_json-1"], optional = true }
tokio-rustls = { version = "0.26", optional = true, default-features = false, features = ["ring", "tls12"] }
rustls = { version = "0.23.5", optional = true, default-features = false, features = ["std", "ring", "tls12"] }
rustls-pki-types = { version = "1", optional = true }

6
crates/store/src/backend/elastic/main.rs
View File

@@ -103,6 +103,12 @@ impl ElasticSearchStore {
});
let body = serde_json::to_string(&body).unwrap_or_default();
let c = println!(
"Creating Elasticsearch index {} with body: {}",
T::index().es_index_name(),
body
);
assert_success(
self.client
.put(format!("{}/{}", self.url, T::index().es_index_name()))

22
crates/store/src/backend/elastic/search.rs
View File

@@ -90,7 +90,7 @@ impl ElasticSearchStore {
[
Some(("query".to_string(), build_query(filters))),
Some(("size".to_string(), Value::from(10_000))),
Some(("source".to_string(), Value::from(false))),
Some(("_source".to_string(), Value::from(false))),
(!sort.is_empty()).then(|| ("sort".to_string(), build_sort(sort))),
]
.into_iter()
@@ -98,6 +98,8 @@ impl ElasticSearchStore {
);
let request = serde_json::to_string(&query).unwrap_or_default();
let c = println!("Elasticsearch query: {}", request);
let response = assert_success(
self.client
.post(format!("{}/{}/_search", self.url, index.es_index_name()))
@@ -134,11 +136,29 @@ impl ElasticSearchStore {
.reason("Unindex operation requires at least one filter"));
}
#[cfg(feature = "test_mode")]
{
assert_success(
self.client
.get(format!(
"{}/{}/_refresh",
self.url,
filter.index.es_index_name()
))
.send()
.await,
)
.await?;
}
let query = json!({
"query": build_query(&filter.filters),
});
let request = serde_json::to_string(&query).unwrap_or_default();
let c = println!("Elasticsearch unindex query: {}", request);
let response = assert_success(
self.client
.post(format!(

22
crates/store/src/backend/mysql/main.rs
View File

@@ -204,20 +204,14 @@ async fn create_search_tables<T: SearchableField + MysqlSearchField + 'static>(
}
// Add primary key constraint
query.push_str("PRIMARY KEY ");
if pkeys.len() > 1 {
query.push('(');
}
query.push_str("PRIMARY KEY (");
for (i, pkey) in pkeys.iter().enumerate() {
if i > 0 {
query.push_str(", ");
}
query.push_str(pkey.column());
}
if pkeys.len() > 1 {
query.push(')');
}
query.push_str(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci");
query.push_str(")) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci");
conn.query_drop(&query).await.map_err(into_error)?;
@@ -226,22 +220,18 @@ async fn create_search_tables<T: SearchableField + MysqlSearchField + 'static>(
if field.is_text() {
let column_name = field.column();
let create_index_query = format!(
"CREATE FULLTEXT INDEX IF NOT EXISTS fts_{table_name}_{column_name} ON {table_name}({column_name})",
"CREATE FULLTEXT INDEX fts_{table_name}_{column_name} ON {table_name}({column_name})",
);
conn.query_drop(&create_index_query)
.await
.map_err(into_error)?;
let _ = conn.query_drop(&create_index_query).await;
}
if field.is_indexed() {
let column_name = field.column();
let create_index_query = format!(
"CREATE INDEX IF NOT EXISTS idx_{table_name}_{column_name} ON {table_name}({column_name})",
"CREATE INDEX idx_{table_name}_{column_name} ON {table_name}({column_name})",
);
conn.query_drop(&create_index_query)
.await
.map_err(into_error)?;
let _ = conn.query_drop(&create_index_query).await;
}
}

10
crates/store/src/backend/mysql/mod.rs
View File

@@ -31,7 +31,7 @@ fn into_error(err: impl Display) -> trc::Error {
}
impl SearchIndex {
fn mysql_table(&self) -> &'static str {
pub(crate) fn mysql_table(&self) -> &'static str {
match self {
SearchIndex::Email => "s_email",
SearchIndex::Calendar => "s_cal",
@@ -68,9 +68,9 @@ impl MysqlSearchField for EmailSearchField {
fn column_type(&self) -> &'static str {
match self {
EmailSearchField::ReceivedAt | EmailSearchField::SentAt => "BIGINT NOT NULL",
EmailSearchField::Size => "INT NOT NULL",
EmailSearchField::HasAttachment => "BOOLEAN NOT NULL",
EmailSearchField::ReceivedAt | EmailSearchField::SentAt => "BIGINT",
EmailSearchField::Size => "INT",
EmailSearchField::HasAttachment => "BOOLEAN",
EmailSearchField::Headers => "JSON",
_ => "TEXT",
}
@@ -146,7 +146,7 @@ impl MysqlSearchField for TracingSearchField {
fn column_type(&self) -> &'static str {
match self {
TracingSearchField::EventType => "BIGINT NOT NULL",
TracingSearchField::EventType => "BIGINT",
TracingSearchField::QueueId => "BIGINT",
TracingSearchField::Keywords => "TEXT",
}

76
crates/store/src/backend/mysql/search.rs
View File

@@ -47,7 +47,7 @@ impl MysqlStore {
}
if let Some(value) = fields.remove(field) {
let _ = write!(&mut query, "${}", values.len() + 1);
query.push('?');
values.push(value);
} else {
query.push_str("NULL");
@@ -86,6 +86,7 @@ impl MysqlStore {
if !sort.is_empty() {
build_sort(&mut query, sort);
}
let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
let s = conn.prep(query).await.map_err(into_error)?;
@@ -132,13 +133,10 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
is_first = false;
}
let value_pos = values.len() + 1;
if field.is_text() {
if field.is_text() && matches!(op, SearchOperator::Equal | SearchOperator::Contains)
{
let value = match (value, op) {
(SearchValue::Text { value, .. }, SearchOperator::Equal) => {
Value::Bytes(format!("{value:?}").into_bytes())
}
(SearchValue::Text { value, .. }, ..) => {
let mut text_query = String::with_capacity(value.len() + 1);
for item in value.split_whitespace() {
@@ -150,6 +148,9 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
Value::Bytes(text_query.into_bytes())
}
(SearchValue::Text { value, .. }, ..) => {
Value::Bytes(format!("{value:?}").into_bytes())
}
_ => {
debug_assert!(false, "Invalid search value for text field");
continue;
@@ -157,7 +158,7 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
};
let _ = write!(
query,
"MATCH({}) AGAINST(${value_pos} IN BOOLEAN MODE)",
"MATCH({}) AGAINST(? IN BOOLEAN MODE)",
field.column()
);
values.push(value);
@@ -168,47 +169,52 @@ fn build_filter(query: &mut String, filters: &[SearchFilter]) -> Vec<Value> {
if !value.is_empty() {
if op == &SearchOperator::Equal {
let _ = write!(
query,
"JSON_EXTRACT({}, ${}) = ${}",
field.column(),
value_pos,
values.len() + 1
);
let _ = write!(query, "JSON_EXTRACT({}, ?) = ?", field.column());
values.push(Value::Bytes(format!("{value:?}").into_bytes()));
} else {
let _ = write!(
query,
"JSON_EXTRACT({}, ${}) LIKE ${}",
field.column(),
value_pos,
values.len() + 1
);
let _ = write!(query, "JSON_EXTRACT({}, ?) LIKE ?", field.column(),);
values.push(Value::Bytes(format!("%{value}%").into_bytes()));
}
} else {
let _ = write!(
query,
"JSON_CONTAINS_PATH({}, 'one', ${})",
field.column(),
value_pos
);
let _ = write!(query, "JSON_CONTAINS_PATH({}, 'one', ?)", field.column(),);
}
} else {
query.push_str(field.column());
query.push(' ');
op.write_mysql(query, value_pos);
op.write_mysql(query);
values.push(to_mysql(value));
}
}
SearchFilter::And | SearchFilter::Or => {
if !is_first {
match operator {
SearchFilter::And => query.push_str(" AND "),
SearchFilter::Or => query.push_str(" OR "),
_ => (),
}
} else {
is_first = false;
}
operator_stack.push((operator, is_first));
operator = filter;
is_first = true;
query.push('(');
}
SearchFilter::Not => {
if !is_first {
match operator {
SearchFilter::And => query.push_str(" AND "),
SearchFilter::Or => query.push_str(" OR "),
_ => (),
}
} else {
is_first = false;
}
operator_stack.push((operator, is_first));
operator = &SearchFilter::And;
is_first = true;
query.push_str("NOT (");
}
SearchFilter::End => {
@@ -255,25 +261,25 @@ fn build_sort(query: &mut String, sort: &[SearchComparator]) {
}
impl SearchOperator {
fn write_mysql(&self, query: &mut String, value_pos: usize) {
fn write_mysql(&self, query: &mut String) {
match self {
SearchOperator::LowerThan => {
let _ = write!(query, " < ${value_pos}");
let _ = write!(query, "< ?");
}
SearchOperator::LowerEqualThan => {
let _ = write!(query, " <= ${value_pos}");
let _ = write!(query, "<= ?");
}
SearchOperator::GreaterThan => {
let _ = write!(query, " > ${value_pos}");
let _ = write!(query, "> ?");
}
SearchOperator::GreaterEqualThan => {
let _ = write!(query, " >= ${value_pos}");
let _ = write!(query, ">= ?");
}
SearchOperator::Equal => {
let _ = write!(query, " = ${value_pos}");
let _ = write!(query, "= ?");
}
SearchOperator::Contains => {
let _ = write!(query, " LIKE '%' CONCAT('%', ${value_pos}, '%')");
let _ = write!(query, "LIKE '%' CONCAT('%', ?, '%')");
}
}
}

10
crates/store/src/backend/postgres/main.rs
View File

@@ -197,20 +197,14 @@ async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
}
// Add primary key constraint
query.push_str("PRIMARY KEY ");
if pkeys.len() > 1 {
query.push('(');
}
query.push_str("PRIMARY KEY (");
for (i, pkey) in pkeys.iter().enumerate() {
if i > 0 {
query.push_str(", ");
}
query.push_str(pkey.column());
}
if pkeys.len() > 1 {
query.push(')');
}
query.push(')');
query.push_str("))");
conn.execute(&query, &[]).await.map_err(into_error)?;

8
crates/store/src/backend/postgres/mod.rs
View File

@@ -35,7 +35,7 @@ fn into_error(err: impl Display) -> trc::Error {
}
impl SearchIndex {
fn psql_table(&self) -> &'static str {
pub(crate) fn psql_table(&self) -> &'static str {
match self {
SearchIndex::Email => "s_email",
SearchIndex::Calendar => "s_cal",
@@ -87,6 +87,8 @@ impl PsqlSearchField for EmailSearchField {
EmailSearchField::From | EmailSearchField::To | EmailSearchField::Subject => {
Some("TEXT")
}
#[cfg(feature = "test_mode")]
EmailSearchField::Cc | EmailSearchField::Bcc => Some("TEXT"),
_ => None,
}
}
@@ -96,6 +98,10 @@ impl PsqlSearchField for EmailSearchField {
EmailSearchField::From => Some("s_fr"),
EmailSearchField::To => Some("s_to"),
EmailSearchField::Subject => Some("s_sj"),
#[cfg(feature = "test_mode")]
EmailSearchField::Bcc => Some("s_bc"),
#[cfg(feature = "test_mode")]
EmailSearchField::Cc => Some("s_cc"),
_ => None,
}
}

83
crates/store/src/backend/postgres/search.rs
View File

@@ -16,7 +16,7 @@ use nlp::language::Language;
use std::fmt::Write;
use tokio_postgres::{
IsolationLevel,
types::{ToSql, Type},
types::{FromSql, ToSql, Type, WrongType},
};
impl PostgresStore {
@@ -128,7 +128,7 @@ impl PostgresStore {
.await
.and_then(|rows| {
rows.into_iter()
.map(|row| row.try_get::<_, i64>(0).map(|v| R::from_u64(v as u64)))
.map(|row| row.try_get::<_, DocId>(0).map(|v| R::from_u64(v.0)))
.collect::<Result<Vec<R>, _>>()
})
.map_err(into_error)
@@ -172,11 +172,13 @@ impl PostgresStore {
} else {
is_first = false;
}
query.push_str(field.column());
query.push(' ');
let value_pos = values.len() + 1;
if field.is_text() {
if field.is_text()
&& matches!(op, SearchOperator::Equal | SearchOperator::Contains)
{
query.push_str(field.column());
query.push(' ');
let language = match &value {
SearchValue::Text { language, .. }
if self.languages.contains(language) =>
@@ -192,29 +194,57 @@ impl PostgresStore {
let _ = write!(query, "@@ {method}('{language}', ${value_pos})");
values.push(value as &(dyn ToSql + Sync));
} else if let SearchValue::KeyValues(kv) = value {
query.push_str(field.column());
query.push(' ');
let (key, value) = kv.iter().next().unwrap();
values.push(key as &(dyn ToSql + Sync));
if !value.is_empty() {
query.push_str("->>?");
let _ = write!(query, "->> ${value_pos} ");
op.write_pqsql(query, values.len() + 1);
values.push(value as &(dyn ToSql + Sync));
} else {
let _ = write!(query, " ? ${value_pos}");
}
} else {
query.push_str(field.sort_column().unwrap_or(field.column()));
query.push(' ');
op.write_pqsql(query, value_pos);
values.push(value as &(dyn ToSql + Sync));
}
}
SearchFilter::And | SearchFilter::Or => {
if !is_first {
match operator {
SearchFilter::And => query.push_str(" AND "),
SearchFilter::Or => query.push_str(" OR "),
_ => (),
}
} else {
is_first = false;
}
operator_stack.push((operator, is_first));
operator = filter;
is_first = true;
query.push('(');
}
SearchFilter::Not => {
if !is_first {
match operator {
SearchFilter::And => query.push_str(" AND "),
SearchFilter::Or => query.push_str(" OR "),
_ => (),
}
} else {
is_first = false;
}
operator_stack.push((operator, is_first));
operator = &SearchFilter::And;
is_first = true;
query.push_str("NOT (");
}
SearchFilter::End => {
@@ -281,9 +311,9 @@ impl ToSql for SearchValue {
_ => (*v as i64).to_sql(ty, out),
},
SearchValue::Boolean(v) => v.to_sql(ty, out),
SearchValue::KeyValues(kv) => serde_json::to_string(kv)
.unwrap_or_default()
.to_sql(ty, out),
SearchValue::KeyValues(kv) => {
serde_json::to_value(kv).unwrap_or_default().to_sql(ty, out)
}
}
}
@@ -310,33 +340,52 @@ impl ToSql for SearchValue {
_ => (*v as i64).to_sql_checked(ty, out),
},
SearchValue::Boolean(v) => v.to_sql_checked(ty, out),
SearchValue::KeyValues(kv) => serde_json::to_string(kv)
SearchValue::KeyValues(kv) => serde_json::to_value(kv)
.unwrap_or_default()
.to_sql_checked(ty, out),
}
}
}
struct DocId(u64);
impl FromSql<'_> for DocId {
fn from_sql(
ty: &tokio_postgres::types::Type,
raw: &'_ [u8],
) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
match ty {
&Type::INT4 => i32::from_sql(ty, raw).map(|v| DocId(v as u64)),
&Type::INT8 | &Type::OID => i64::from_sql(ty, raw).map(|v| DocId(v as u64)),
_ => Err(Box::new(WrongType::new::<DocId>(ty.clone()))),
}
}
fn accepts(typ: &Type) -> bool {
matches!(typ, &Type::INT4 | &Type::INT8 | &Type::OID)
}
}
impl SearchOperator {
fn write_pqsql(&self, query: &mut String, value_pos: usize) {
match self {
SearchOperator::LowerThan => {
let _ = write!(query, " < ${value_pos}");
let _ = write!(query, "< ${value_pos}");
}
SearchOperator::LowerEqualThan => {
let _ = write!(query, " <= ${value_pos}");
let _ = write!(query, "<= ${value_pos}");
}
SearchOperator::GreaterThan => {
let _ = write!(query, " > ${value_pos}");
let _ = write!(query, "> ${value_pos}");
}
SearchOperator::GreaterEqualThan => {
let _ = write!(query, " >= ${value_pos}");
let _ = write!(query, ">= ${value_pos}");
}
SearchOperator::Equal => {
let _ = write!(query, " = ${value_pos}");
let _ = write!(query, "= ${value_pos}");
}
SearchOperator::Contains => {
let _ = write!(query, " LIKE '%' || ${value_pos} || '%'");
let _ = write!(query, "LIKE '%' || ${value_pos} || '%'");
}
}
}

5
crates/store/src/dispatch/search.rs
View File

@@ -4,6 +4,8 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use trc::AddContext;
use crate::{
SearchStore, Store,
search::{
@@ -65,7 +67,8 @@ impl SearchStore {
.into_iter()
.filter(|id| query.mask.contains(*id))
.collect()
});
})
.caused_by(trc::location!());
}
// Decompose filters into external and local filters

36
crates/store/src/dispatch/store.rs
View File

@@ -20,16 +20,6 @@ use std::{ops::Range, time::Instant};
use trc::{AddContext, StoreEvent};
use types::collection::Collection;
#[cfg(feature = "test_mode")]
#[allow(clippy::type_complexity)]
static BITMAPS: std::sync::LazyLock<
std::sync::Arc<
parking_lot::Mutex<std::collections::HashMap<Vec<u8>, std::collections::HashSet<u32>>>,
>,
> = std::sync::LazyLock::new(|| {
std::sync::Arc::new(parking_lot::Mutex::new(std::collections::HashMap::new()))
});
impl Store {
pub async fn get_value<U>(&self, key: impl Key) -> trc::Result<Option<U>>
where
@@ -467,6 +457,21 @@ impl Store {
pub async fn destroy(&self) {
use crate::*;
if self.is_pg_or_mysql() {
use crate::write::SearchIndex;
for index in [
SearchIndex::Email,
SearchIndex::Calendar,
SearchIndex::Contacts,
SearchIndex::Tracing,
] {
self.sql_query::<usize>(&format!("TRUNCATE TABLE {}", index.psql_table()), vec![])
.await
.unwrap();
}
}
for subspace in [
SUBSPACE_ACL,
SUBSPACE_DIRECTORY,
@@ -488,7 +493,12 @@ impl Store {
SUBSPACE_REPORT_IN,
SUBSPACE_TELEMETRY_SPAN,
SUBSPACE_TELEMETRY_METRIC,
SUBSPACE_SEARCH_INDEX,
] {
if subspace == SUBSPACE_SEARCH_INDEX && self.is_pg_or_mysql() {
continue;
}
self.delete_range(
AnyKey {
subspace,
@@ -510,8 +520,6 @@ impl Store {
.await
.unwrap();
}
BITMAPS.lock().clear();
}
#[cfg(feature = "test_mode")]
@@ -668,6 +676,10 @@ impl Store {
(SUBSPACE_TELEMETRY_METRIC, true),
(SUBSPACE_SEARCH_INDEX, true),
] {
if subspace == SUBSPACE_SEARCH_INDEX && store.is_pg_or_mysql() {
continue;
}
let from_key = crate::write::AnyKey {
subspace,
key: vec![0u8],

4
crates/store/src/lib.rs
View File

@@ -691,8 +691,8 @@ impl Store {
pub fn is_pg_or_mysql(&self) -> bool {
match self {
#[cfg(feature = "sqlite")]
Store::SQLite(_) => true,
#[cfg(feature = "mysql")]
Store::MySQL(_) => true,
#[cfg(feature = "postgres")]
Store::PostgreSQL(_) => true,
_ => false,

93
crates/store/src/search/bm_u32.rs
View File

@@ -57,32 +57,51 @@ impl BitmapCache {
}
}
Entry::Vacant(entry) => {
let value = store
.get_value::<RoaringBitmap>(ValueKey::from(ValueClass::SearchIndex(
SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: Some(account_id),
hash,
field,
},
let from_key = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
typ: SearchIndexType::Term { hash, field },
}));
let to_key = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
typ: SearchIndexType::Term { hash, field },
}));
let key_len = (U32_LEN * 2) + hash.len() + 2;
let mut documents = RoaringBitmap::new();
store
.iterate(
IterateParams::new(from_key, to_key).no_values().ascending(),
|key, _| {
if key.len() == key_len {
documents.insert(key.deserialize_be_u32(key.len() - U32_LEN)?);
}
Ok(true)
},
)))
)
.await
.caused_by(trc::location!())?;
if let Some(bm) = &value {
if !documents.is_empty() {
if is_union {
result.bitor_assign(bm);
result.bitor_assign(&documents);
} else if idx == 0 {
result = bm.clone();
result = documents.clone();
} else {
result.bitand_assign(bm);
result.bitand_assign(&documents);
if result.is_empty() {
entry.insert(value);
entry.insert(Some(documents));
return Ok(None);
}
}
entry.insert(value);
entry.insert(Some(documents));
} else if !is_union {
entry.insert(None);
return Ok(None);
@@ -133,11 +152,11 @@ pub(crate) async fn range_to_bitmap(
}
let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: from_doc_id,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: from_doc_id,
},
field: SearchIndexField {
field_id: from_field,
len: len as u8,
@@ -153,11 +172,11 @@ pub(crate) async fn range_to_bitmap(
}
let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: end_doc_id,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: end_doc_id,
},
field: SearchIndexField {
field_id: end_field,
len: len as u8,
@@ -220,11 +239,11 @@ pub(crate) async fn sort_order(
) -> trc::Result<AHashMap<u32, u32>> {
let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
field: SearchIndexField {
field_id,
len: SEARCH_INDEX_MAX_FIELD_LEN as u8,
@@ -234,11 +253,11 @@ pub(crate) async fn sort_order(
}));
let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
field: SearchIndexField {
field_id,
len: SEARCH_INDEX_MAX_FIELD_LEN as u8,
@@ -247,14 +266,22 @@ pub(crate) async fn sort_order(
},
}));
let mut last_value = Vec::new();
let mut results = AHashMap::new();
let mut pos = 0;
store
.iterate(
IterateParams::new(begin, end).no_values().ascending(),
|key, _| {
let value = key
.get(U32_LEN + 2..key.len() - U32_LEN)
.ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?;
if value != last_value {
pos += 1;
last_value = value.to_vec();
}
results.insert(key.deserialize_be_u32(key.len() - U32_LEN)?, pos);
pos += 1;
Ok(true)
},
)

49
crates/store/src/search/bm_u64.rs
View File

@@ -56,32 +56,45 @@ impl TreemapCache {
}
}
Entry::Vacant(entry) => {
let value = store
.get_value::<RoaringTreemap>(ValueKey::from(ValueClass::SearchIndex(
SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: None,
hash,
field,
},
let from_key = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Global { id: 0 },
typ: SearchIndexType::Term { hash, field },
}));
let to_key = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Global { id: u64::MAX },
typ: SearchIndexType::Term { hash, field },
}));
let key_len = U64_LEN + hash.len() + 2;
let mut documents = RoaringTreemap::new();
store
.iterate(
IterateParams::new(from_key, to_key).no_values().ascending(),
|key, _| {
if key.len() == key_len {
documents.insert(key.deserialize_be_u64(key.len() - U64_LEN)?);
}
Ok(true)
},
)))
)
.await
.caused_by(trc::location!())?;
if let Some(bm) = &value {
if !documents.is_empty() {
if is_union {
result.bitor_assign(bm);
result.bitor_assign(&documents);
} else if idx == 0 {
result = bm.clone();
result = documents.clone();
} else {
result.bitand_assign(bm);
result.bitand_assign(&documents);
if result.is_empty() {
entry.insert(value);
entry.insert(Some(documents));
return Ok(None);
}
}
entry.insert(value);
entry.insert(Some(documents));
} else if !is_union {
entry.insert(None);
return Ok(None);
@@ -131,8 +144,8 @@ pub(crate) async fn range_to_treemap(
}
let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Global { id: from_id },
typ: SearchIndexType::Index {
id: SearchIndexId::Global { id: from_id },
field: SearchIndexField {
field_id: from_field,
len: len as u8,
@@ -148,8 +161,8 @@ pub(crate) async fn range_to_treemap(
}
let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Global { id: end_id },
typ: SearchIndexType::Index {
id: SearchIndexId::Global { id: end_id },
field: SearchIndexField {
field_id: end_field,
len: len as u8,

4
crates/store/src/search/document.rs
View File

@@ -99,6 +99,10 @@ impl IndexDocument {
self.fields.contains_key(field)
}
pub fn fields(&self) -> impl Iterator<Item = (&SearchField, &SearchValue)> {
self.fields.iter()
}
pub fn set_unknown_language(&mut self, lang: Language) {
for value in self.fields.values_mut() {
if let SearchValue::Text { language, .. } = value

4
crates/store/src/search/fields.rs
View File

@@ -41,6 +41,7 @@ impl SearchableField for EmailSearchField {
| EmailSearchField::To
| EmailSearchField::Subject
| EmailSearchField::ReceivedAt
| EmailSearchField::SentAt
| EmailSearchField::Size
| EmailSearchField::HasAttachment,
)
@@ -57,7 +58,8 @@ impl SearchableField for EmailSearchField {
| EmailSearchField::SentAt
| EmailSearchField::Size
| EmailSearchField::HasAttachment
| EmailSearchField::Bcc,
| EmailSearchField::Bcc
| EmailSearchField::Cc
)
}
}

79
crates/store/src/search/index.rs
View File

@@ -61,7 +61,10 @@ impl Store {
.push(id as u32);
}
(SearchField::Id, SearchValue::Uint(id)) => match op {
SearchOperator::LowerThan | SearchOperator::LowerEqualThan => {
SearchOperator::LowerThan => {
to_id = Some(id.saturating_sub(1));
}
SearchOperator::LowerEqualThan => {
to_id = Some(id);
}
SearchOperator::Equal => {
@@ -73,17 +76,17 @@ impl Store {
.reason("Unsupported operator for Id field"));
}
},
_ => {
filter => {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.reason("Unsupported filter"));
.details(format!("Unsupported unindex filter {filter:?}")));
}
},
SearchFilter::And | SearchFilter::Or | SearchFilter::End => {}
SearchFilter::Not | SearchFilter::DocumentSet(_) => {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.reason("Unsupported filter"));
.details(format!("Unsupported unindex filter {filter:?}")));
}
}
}
@@ -96,12 +99,11 @@ impl Store {
.get_value::<Archive<AlignedBytes>>(ValueKey::from(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Account {
account_id,
document_id,
},
id: SearchIndexId::Account {
account_id,
document_id,
},
typ: SearchIndexType::Document,
}),
))
.await
@@ -130,21 +132,19 @@ impl Store {
self.delete_range(
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
typ: SearchIndexType::Document,
})),
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
typ: SearchIndexType::Document,
})),
)
.await
@@ -153,11 +153,11 @@ impl Store {
self.delete_range(
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
field: SearchIndexField {
field_id: 0,
len: 1,
@@ -167,11 +167,11 @@ impl Store {
})),
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
typ: SearchIndexType::Index {
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
field: SearchIndexField {
field_id: u8::MAX,
len: 1,
@@ -186,16 +186,22 @@ impl Store {
self.delete_range(
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: 0,
},
typ: SearchIndexType::Term {
account_id: Some(account_id),
hash: CheekyHash::NULL,
field: 0,
},
})),
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
id: SearchIndexId::Account {
account_id,
document_id: u32::MAX,
},
typ: SearchIndexType::Term {
account_id: Some(account_id),
hash: CheekyHash::FULL,
field: u8::MAX,
},
@@ -212,9 +218,8 @@ impl Store {
.get_value::<Archive<AlignedBytes>>(ValueKey::from(ValueClass::SearchIndex(
SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Global { id },
},
id: SearchIndexId::Global { id },
typ: SearchIndexType::Document,
},
)))
.await
@@ -239,15 +244,13 @@ impl Store {
IterateParams::new(
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Global { id: 0 },
},
id: SearchIndexId::Global { id: 0 },
typ: SearchIndexType::Document,
})),
ValueKey::from(ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document {
id: SearchIndexId::Global { id: to_id },
},
id: SearchIndexId::Global { id: to_id },
typ: SearchIndexType::Document,
})),
),
|key, value| {

6
crates/store/src/search/local.rs
View File

@@ -210,11 +210,7 @@ impl QueryResults {
SearchComparator::Field { .. } => continue,
};
let ordering = if is_ascending {
a.cmp(&b).reverse()
} else {
a.cmp(&b)
};
let ordering = if is_ascending { a.cmp(&b) } else { b.cmp(&a) };
if ordering != Ordering::Equal {
return ordering;

78
crates/store/src/search/query.rs
View File

@@ -31,38 +31,45 @@ impl Store {
};
let mut stack = Vec::new();
let mask = query.mask;
let mut filters = query.filters.into_iter().peekable();
let mut bitmaps = BitmapCache::default();
let mut account_id = u32::MAX;
for filter in &query.filters {
if let SearchFilter::Operator {
field: SearchField::AccountId,
value: SearchValue::Uint(id),
..
} = filter
{
account_id = *id as u32;
break;
}
}
if account_id == u32::MAX {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.details("Account ID must be specified before other filters"));
}
#[cfg(feature = "test_mode")]
{
if query.filters.len() == 1 {
state.bm = Some(mask.clone());
}
}
let mut filters = query.filters.into_iter().peekable();
while let Some(filter) = filters.next() {
let mut result = match filter {
SearchFilter::Operator { field, op, value } => {
match &field {
SearchField::AccountId => {
if let SearchValue::Uint(id) = value {
account_id = id as u32;
} else {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.details("Account ID field requires uint value"));
}
}
SearchField::DocumentId | SearchField::Id => {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.details("Document ID field cannot be used in search queries"));
}
_ => {
if account_id == u32::MAX {
return Err(trc::StoreEvent::UnexpectedError
.into_err()
.details("Account ID must be specified before other filters"));
}
}
if matches!(field, SearchField::AccountId) {
continue;
}
if field.is_text() {
if field.is_text()
&& matches!(op, SearchOperator::Contains | SearchOperator::Equal)
{
let (value, language) = match value {
SearchValue::Text { value, language } => (value, language),
_ => {
@@ -88,16 +95,20 @@ impl Store {
} else {
let mut result = RoaringBitmap::new();
for token in Stemmer::new(&value, language, MAX_TOKEN_LENGTH) {
let hash = Some(CheekyHash::new(token.word.as_bytes()));
let stemmed_hash = token
.stemmed_word
.map(|word| CheekyHash::new(format!("{word}*")));
let mut tokens = Vec::with_capacity(3);
tokens.push(CheekyHash::new(token.word.as_bytes()));
tokens.push(CheekyHash::new(format!("{}*", token.word).as_bytes()));
if let Some(stemmed_word) = token.stemmed_word {
tokens.push(CheekyHash::new(
format!("{stemmed_word}*").as_bytes(),
));
}
let union = bitmaps
.merge_bitmaps(
self,
query.index,
account_id,
[hash, stemmed_hash].into_iter().flatten(),
tokens.into_iter(),
field.u8_id(),
true,
)
@@ -138,10 +149,13 @@ impl Store {
self,
query.index,
account_id,
[CheekyHash::new(format!("{key} {value}").as_bytes())]
.into_iter(),
SpaceTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH).map(
|value| {
CheekyHash::new(format!("{key} {value}").as_bytes())
},
),
field.u8_id(),
false,
true,
)
.await?
} else {

200
crates/store/src/search/term.rs
View File

@@ -5,19 +5,18 @@
*/
use crate::{
Deserialize, Serialize, U64_LEN,
Serialize, U64_LEN,
backend::MAX_TOKEN_LENGTH,
search::*,
write::{
Archiver, BatchBuilder, MergeResult, Params, SEARCH_INDEX_MAX_FIELD_LEN, SearchIndexClass,
SearchIndexField, SearchIndexId, SearchIndexType, ValueClass,
Archiver, BatchBuilder, SEARCH_INDEX_MAX_FIELD_LEN, SearchIndexClass, SearchIndexField,
SearchIndexId, SearchIndexType, ValueClass,
},
};
use nlp::{
language::stemmer::Stemmer,
tokenizers::{space::SpaceTokenizer, word::WordTokenizer},
};
use roaring::RoaringTreemap;
use utils::{
cheeky_hash::{CheekyBTreeMap, CheekyHash},
map::bitmap::BitPop,
@@ -153,7 +152,7 @@ impl TermIndexBuilder {
}
SearchValue::Int(v) => {
let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN];
data[..U64_LEN].copy_from_slice(&v.to_be_bytes());
data[..U64_LEN].copy_from_slice(&(v as u64).to_be_bytes());
SearchIndexField {
field_id: field.u8_id(),
@@ -218,82 +217,26 @@ impl TermIndex {
batch.set(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document { id },
id,
typ: SearchIndexType::Document,
}),
archive.serialize()?,
);
match id {
SearchIndexId::Account {
account_id,
document_id,
} => {
for term in archive.inner.terms {
let mut fields = term.fields;
while let Some(field) = fields.bit_pop() {
batch.merge_fnc(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: Some(account_id),
hash: term.hash,
field,
},
}),
Params::with_capacity(1).with_u64(document_id as u64),
|params, _, bytes| {
let document_id = params.u64(0) as u32;
if let Some(bytes) = bytes {
let mut bitmap = RoaringBitmap::deserialize(bytes)?;
if bitmap.insert(document_id) {
Ok(MergeResult::Update(bitmap.serialize()?))
} else {
Ok(MergeResult::Skip)
}
} else {
Ok(MergeResult::Update(
RoaringBitmap::from_iter([document_id]).serialize()?,
))
}
},
);
}
}
}
SearchIndexId::Global { id } => {
for term in archive.inner.terms {
let mut fields = term.fields;
while let Some(field) = fields.bit_pop() {
batch.merge_fnc(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: None,
hash: term.hash,
field,
},
}),
Params::with_capacity(1).with_u64(id),
|params, _, bytes| {
let id = params.u64(0);
if let Some(bytes) = bytes {
let mut bitmap = RoaringTreemap::deserialize(bytes)?;
if bitmap.insert(id) {
Ok(MergeResult::Update(bitmap.serialize()?))
} else {
Ok(MergeResult::Skip)
}
} else {
Ok(MergeResult::Update(
RoaringTreemap::from_iter([id]).serialize()?,
))
}
},
);
}
}
for term in archive.inner.terms {
let mut fields = term.fields;
while let Some(field) = fields.bit_pop() {
batch.set(
ValueClass::SearchIndex(SearchIndexClass {
index,
id,
typ: SearchIndexType::Term {
hash: term.hash,
field,
},
}),
vec![],
);
}
}
@@ -301,7 +244,8 @@ impl TermIndex {
batch.set(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Index { id, field },
id,
typ: SearchIndexType::Index { field },
}),
vec![],
);
@@ -312,104 +256,32 @@ impl TermIndex {
}
impl ArchivedTermIndex {
/*pub fn has_term(&self, hash: &CheekyHash, field: &SearchField) -> bool {
let hash = hash.as_raw_bytes();
self.terms
.binary_search_by(|term| term.hash.as_raw_bytes().cmp(hash))
.is_ok_and(|idx| {
(self.terms[idx].fields.to_native() & (1 << (field.u8_id() as u32))) != 0
})
}*/
pub fn delete_index(&self, batch: &mut BatchBuilder, index: SearchIndex, id: SearchIndexId) {
batch.clear(ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Document { id },
id,
typ: SearchIndexType::Document,
}));
match id {
SearchIndexId::Account {
account_id,
document_id,
} => {
for term in self.terms.iter() {
let mut fields = term.fields.to_native();
while let Some(field) = fields.bit_pop() {
batch.merge_fnc(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: Some(account_id),
hash: term.hash.to_native(),
field,
},
}),
Params::with_capacity(1).with_u64(document_id as u64),
|params, _, bytes| {
let document_id = params.u64(0) as u32;
if let Some(bytes) = bytes {
let mut bitmap = RoaringBitmap::deserialize(bytes)?;
if bitmap.remove(document_id) {
if !bitmap.is_empty() {
Ok(MergeResult::Update(bitmap.serialize()?))
} else {
Ok(MergeResult::Delete)
}
} else {
Ok(MergeResult::Skip)
}
} else {
Ok(MergeResult::Skip)
}
},
);
}
}
}
SearchIndexId::Global { id } => {
for term in self.terms.iter() {
let mut fields = term.fields.to_native();
while let Some(field) = fields.bit_pop() {
batch.merge_fnc(
ValueClass::SearchIndex(SearchIndexClass {
index,
typ: SearchIndexType::Term {
account_id: None,
hash: term.hash.to_native(),
field,
},
}),
Params::with_capacity(1).with_u64(id),
|params, _, bytes| {
let id = params.u64(0);
if let Some(bytes) = bytes {
let mut bitmap = RoaringTreemap::deserialize(bytes)?;
if bitmap.remove(id) {
if !bitmap.is_empty() {
Ok(MergeResult::Update(bitmap.serialize()?))
} else {
Ok(MergeResult::Delete)
}
} else {
Ok(MergeResult::Skip)
}
} else {
Ok(MergeResult::Skip)
}
},
);
}
}
for term in self.terms.iter() {
let mut fields = term.fields.to_native();
while let Some(field) = fields.bit_pop() {
batch.clear(ValueClass::SearchIndex(SearchIndexClass {
index,
id,
typ: SearchIndexType::Term {
hash: term.hash.to_native(),
field,
},
}));
}
}
for field in self.fields.iter() {
batch.clear(ValueClass::SearchIndex(SearchIndexClass {
index,
id,
typ: SearchIndexType::Index {
id,
field: SearchIndexField {
field_id: field.field_id,
len: field.len,

38
crates/store/src/write/key.rs
View File

@@ -442,30 +442,30 @@ impl ValueClass {
.write(u8::from(SyncCollection::ShareNotification))
.write(*notification_id),
ValueClass::SearchIndex(index) => match &index.typ {
SearchIndexType::Term {
account_id,
field,
hash,
} => {
SearchIndexType::Term { field, hash } => {
let class = index.index.as_u8();
if let Some(account_id) = account_id {
serializer
match &index.id {
SearchIndexId::Account {
account_id,
document_id,
} => serializer
.write(class)
.write(*account_id)
.write(hash.payload())
.write(hash.payload_len())
.write(*field)
} else {
serializer
.write(*document_id),
SearchIndexId::Global { id } => serializer
.write(class)
.write(hash.payload())
.write(hash.payload_len())
.write(*field)
.write(*id),
}
}
SearchIndexType::Index { id, field } => {
SearchIndexType::Index { field } => {
let class = index.index.as_u8() | 1 << 6;
match id {
match &index.id {
SearchIndexId::Account {
account_id,
document_id,
@@ -482,9 +482,9 @@ impl ValueClass {
.write(*id),
}
}
SearchIndexType::Document { id } => {
SearchIndexType::Document => {
let class = index.index.as_u8() | 2 << 6;
match id {
match &index.id {
SearchIndexId::Account {
account_id,
document_id,
@@ -600,17 +600,9 @@ impl ValueClass {
ValueClass::ChangeId => U32_LEN,
ValueClass::ShareNotification { .. } => U32_LEN + U64_LEN + 1,
ValueClass::SearchIndex(v) => match &v.typ {
SearchIndexType::Term {
account_id, hash, ..
} => {
if account_id.is_some() {
2 + U32_LEN + hash.len()
} else {
2 + hash.len()
}
}
SearchIndexType::Term { hash, .. } => U64_LEN + hash.len() + 2,
SearchIndexType::Index { field, .. } => 1 + field.len as usize + U64_LEN,
SearchIndexType::Document { id } => match id {
SearchIndexType::Document => match &v.id {
SearchIndexId::Account { .. } => 1 + U32_LEN * 2,
SearchIndexId::Global { .. } => 1 + U64_LEN,
},

16
crates/store/src/write/mod.rs
View File

@@ -194,23 +194,15 @@ pub enum IndexPropertyClass {
#[derive(Debug, PartialEq, Clone, Eq, Hash)]
pub struct SearchIndexClass {
pub index: SearchIndex,
pub id: SearchIndexId,
pub typ: SearchIndexType,
}
#[derive(Debug, PartialEq, Clone, Eq, Hash)]
pub enum SearchIndexType {
Term {
account_id: Option<u32>,
field: u8,
hash: CheekyHash,
},
Index {
id: SearchIndexId,
field: SearchIndexField,
},
Document {
id: SearchIndexId,
},
Term { field: u8, hash: CheekyHash },
Index { field: SearchIndexField },
Document,
}
pub(crate) const SEARCH_INDEX_MAX_FIELD_LEN: usize = 16;

165
tests/src/store/mod.rs
View File

@@ -19,63 +19,11 @@ pub struct TempDir {
pub path: std::path::PathBuf,
}
const CONFIG: &str = r#"
[store."s3"]
type = "s3"
access-key = "minioadmin"
secret-key = "minioadmin"
region = "eu-central-1"
endpoint = "http://localhost:9000"
bucket = "tmp"
[store."fs"]
type = "fs"
path = "{TMP}"
[store."rocksdb"]
type = "rocksdb"
path = "{TMP}/rocksdb"
[store."foundationdb"]
type = "foundationdb"
[store."sqlite"]
type = "sqlite"
path = "{TMP}/sqlite.db"
[store."postgresql"]
type = "postgresql"
host = "localhost"
port = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
[store."mysql"]
type = "mysql"
host = "localhost"
port = 3307
database = "stalwart"
user = "root"
password = "password"
[store."redis"]
type = "redis"
urls = "redis://127.0.0.1"
redis-type = "single"
[storage]
lookup = "mysql"
data = "postgresql"
blob = "sqlite"
"#;
#[tokio::test(flavor = "multi_thread")]
pub async fn store_tests() {
let insert = true;
let temp_dir = TempDir::new("store_tests", insert);
let mut config = Config::new(CONFIG.replace("{TMP}", &temp_dir.path.to_string_lossy()))
let mut config = Config::new(build_store_config(&temp_dir.path.to_string_lossy()))
.unwrap()
.assert_no_errors();
let stores = Stores::parse_all(&mut config, false).await;
@@ -95,7 +43,38 @@ pub async fn store_tests() {
//import_export::test(store.clone()).await;
ops::test(store.clone()).await;
query::test(SearchStore::Store(store.clone()), insert).await;
if insert {
temp_dir.delete();
}
}
#[tokio::test(flavor = "multi_thread")]
pub async fn search_tests() {
let insert = true;
let temp_dir = TempDir::new("search_store_tests", insert);
let mut config = Config::new(build_store_config(&temp_dir.path.to_string_lossy()))
.unwrap()
.assert_no_errors();
let stores = Stores::parse_all(&mut config, false).await;
let store_id = std::env::var("SEARCH_STORE")
.expect("Missing store type. Try running `SEARCH_STORE=<store_type> cargo test`");
let store = stores
.search_stores
.get(&store_id)
.expect("Store not found")
.clone();
println!("Testing store {}...", store_id);
if insert {
match &store {
SearchStore::Store(store) => store.destroy().await,
SearchStore::ElasticSearch(_) => (),
}
}
query::test(store, insert).await;
if insert {
temp_dir.delete();
@@ -130,3 +109,81 @@ impl TempDir {
std::fs::remove_dir_all(&self.path).unwrap();
}
}
pub fn build_store_config(temp_dir: &str) -> String {
let store = std::env::var("STORE")
.expect("Missing store type. Try running `STORE=<store_type> cargo test`");
let fts_store = std::env::var("SEARCH_STORE").unwrap_or_else(|_| store.clone());
let blob_store = std::env::var("BLOB_STORE").unwrap_or_else(|_| store.clone());
let lookup_store = std::env::var("LOOKUP_STORE").unwrap_or_else(|_| store.clone());
CONFIG
.replace("{STORE}", &store)
.replace("{SEARCH_STORE}", &fts_store)
.replace("{BLOB_STORE}", &blob_store)
.replace("{LOOKUP_STORE}", &lookup_store)
.replace("{TMP}", temp_dir)
}
const CONFIG: &str = r#"
[store."sqlite"]
type = "sqlite"
path = "{TMP}/sqlite.db"
[store."rocksdb"]
type = "rocksdb"
path = "{TMP}/rocks.db"
[store."foundationdb"]
type = "foundationdb"
[store."postgresql"]
type = "postgresql"
host = "localhost"
port = 5432
database = "stalwart"
user = "postgres"
password = "mysecretpassword"
[store."mysql"]
type = "mysql"
host = "localhost"
port = 3307
database = "stalwart"
user = "root"
password = "password"
[store."elastic"]
type = "elasticsearch"
url = "https://localhost:9200"
tls.allow-invalid-certs = true
[store."elastic".auth]
username = "elastic"
secret = "changeme"
[store."s3"]
type = "s3"
access-key = "minioadmin"
secret-key = "minioadmin"
region = "eu-central-1"
endpoint = "http://localhost:9000"
bucket = "tmp"
[store."fs"]
type = "fs"
path = "{TMP}"
[store."redis"]
type = "redis"
urls = "redis://127.0.0.1"
redis-type = "single"
[storage]
data = "{STORE}"
fts = "{SEARCH_STORE}"
blob = "{BLOB_STORE}"
lookup = "{LOOKUP_STORE}"
directory = "{STORE}"
"#;

167
tests/src/store/ops.rs
View File

@@ -19,39 +19,50 @@ use types::collection::{Collection, SyncCollection};
// FDB max value
const MAX_VALUE_SIZE: usize = 100000;
fn value_gen(chunks: impl IntoIterator<Item = (u8, usize)>) -> Vec<u8> {
let mut value = Vec::new();
for (byte, size) in chunks {
value.extend(std::iter::repeat_n(byte, size));
}
value
}
pub async fn test(db: Store) {
#[cfg(feature = "foundationdb")]
if matches!(db, Store::FoundationDb(_)) && std::env::var("SLOW_FDB_TRX").is_ok() {
if matches!(db, Store::FoundationDb(_)) {
use types::collection::Collection;
println!("Running slow FoundationDB transaction tests...");
// Create 900000 keys
println!("Running FoundationDB chunked iterator test...");
let kvs = [
("a", value_gen([(b'a', 1)])),
("b", value_gen([(b'b', MAX_VALUE_SIZE), (b'0', 1)])),
(
"c",
value_gen([
(b'c', MAX_VALUE_SIZE),
(b'1', MAX_VALUE_SIZE),
(b'2', MAX_VALUE_SIZE),
]),
),
(
"d",
value_gen([(b'd', MAX_VALUE_SIZE), (b'3', MAX_VALUE_SIZE)]),
),
("e", value_gen([(b'e', 1)])),
];
let mut batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
for n in 0..900000 {
batch.set(
ValueClass::Config(format!("key{n:10}").into_bytes()),
format!("value{n:10}").into_bytes(),
);
if n % 10000 == 0 {
db.write(batch.build_all()).await.unwrap();
batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
}
for (key, value) in &kvs {
batch.set(ValueClass::Config(key.as_bytes().to_vec()), value.clone());
}
db.write(batch.build_all()).await.unwrap();
println!("Created 900.000 keys...");
// Iterate over all keys
let mut n = 0;
let mut results = Vec::new();
db.iterate(
store::IterateParams::new(
ValueKey {
@@ -68,38 +79,110 @@ pub async fn test(db: Store) {
},
),
|key, value| {
assert_eq!(std::str::from_utf8(key).unwrap(), format!("key{n:10}"));
assert_eq!(std::str::from_utf8(value).unwrap(), format!("value{n:10}"));
n += 1;
if n % 10000 == 0 {
println!("Iterated over {n} keys");
std::thread::sleep(std::time::Duration::from_millis(1000));
}
results.push((String::from_utf8(key.to_vec()).unwrap(), value.to_vec()));
Ok(true)
},
)
.await
.unwrap();
// Delete 100 keys
let mut batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
for n in 0..900000 {
batch.clear(ValueClass::Config(format!("key{n:10}").into_bytes()));
assert_eq!(results.len(), kvs.len());
if n % 10000 == 0 {
db.write(batch.build_all()).await.unwrap();
batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
db.delete_range(
ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Config(b"".to_vec()),
},
ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Config(b"\xFF".to_vec()),
},
)
.await
.unwrap();
if std::env::var("SLOW_FDB_TRX").is_ok() {
println!("Running FoundationDB slow transaction tests...");
// Create 900000 keys
let mut batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
for n in 0..900000 {
batch.set(
ValueClass::Config(format!("key{n:10}").into_bytes()),
format!("value{n:10}").into_bytes(),
);
if n % 10000 == 0 {
db.write(batch.build_all()).await.unwrap();
batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
}
}
db.write(batch.build_all()).await.unwrap();
println!("Created 900.000 keys...");
// Iterate over all keys
let mut n = 0;
db.iterate(
store::IterateParams::new(
ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Config(b"".to_vec()),
},
ValueKey {
account_id: 0,
collection: 0,
document_id: 0,
class: ValueClass::Config(b"\xFF".to_vec()),
},
),
|key, value| {
assert_eq!(std::str::from_utf8(key).unwrap(), format!("key{n:10}"));
assert_eq!(std::str::from_utf8(value).unwrap(), format!("value{n:10}"));
n += 1;
if n % 10000 == 0 {
println!("Iterated over {n} keys");
std::thread::sleep(std::time::Duration::from_millis(1000));
}
Ok(true)
},
)
.await
.unwrap();
// Delete 100 keys
let mut batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
for n in 0..900000 {
batch.clear(ValueClass::Config(format!("key{n:10}").into_bytes()));
if n % 10000 == 0 {
db.write(batch.build_all()).await.unwrap();
batch = BatchBuilder::new();
batch
.with_account_id(0)
.with_collection(Collection::Email)
.with_document(0);
}
}
db.write(batch.build_all()).await.unwrap();
}
db.write(batch.build_all()).await.unwrap();
}
// Merge values 1000 times concurrently

378
tests/src/store/query.rs
View File

@@ -5,25 +5,23 @@
*/
use crate::store::deflate_test_resource;
use ahash::AHashSet;
use nlp::language::Language;
use std::{
fmt::Display,
io::Write,
sync::{Arc, Mutex},
time::Instant,
};
use store::{
SearchStore, SerializeInfallible,
SearchStore, Store,
ahash::AHashMap,
roaring::RoaringBitmap,
search::{
EmailSearchField, IndexDocument, SearchComparator, SearchField, SearchFilter,
SearchOperator, SearchQuery, SearchValue,
SearchOperator, SearchQuery, SearchValue, TracingSearchField,
},
write::{Operation, SearchIndex, ValueClass},
write::SearchIndex,
};
use store::{Store, ValueKey, write::BatchBuilder};
use types::collection::Collection;
use utils::map::vec_map::VecMap;
pub const FIELDS: [&str; 20] = [
@@ -85,6 +83,26 @@ const FIELD_MAPPINGS: [EmailSearchField; 20] = [
EmailSearchField::HasAttachment, // "url",
];
const ALL_IDS: &[&str] = &[
"p11293", "p79426", "p79427", "p79428", "p79429", "p79430", "d05503", "d00399", "d05352",
"p01764", "t05843", "n02478", "n02479", "n03568", "n03658", "n04327", "n04328", "n04721",
"n04739", "n05095", "n05096", "n05145", "n05157", "n05158", "n05159", "n05298", "n05303",
"n06070", "t01181", "t03571", "t05805", "t05806", "t12147", "t12154", "t12155", "ar00039",
"t12600", "p80203", "t13209", "t13560", "t13561", "t13655", "t13811", "p13352", "p13351",
"p13350", "p13349", "p13348", "p13347", "p13346", "p13345", "p13344", "p13342", "p13341",
"p13340", "p13339", "p13338", "p13337", "p13336", "p13335", "p13334", "p13333", "p13332",
"p13331", "p13330", "p13329", "p13328", "p13327", "p13326", "p13325", "p13324", "p13323",
"t13786", "p13322", "p13321", "p13320", "p13319", "p13318", "p13317", "p13316", "p13315",
"p13314", "t13588", "t13587", "t13586", "t13585", "t13584", "t13540", "t13444", "ar01154",
"ar01153", "t03681", "t12601", "ar00166", "t12625", "t12915", "p04182", "t06483", "ar00703",
"t07671", "ar00021", "t05557", "t07918", "p06298", "p05465", "p06640", "t12855", "t01355",
"t12800", "t12557", "t02078", "ar00052", "ar00627", "t00352", "t07275", "t12318", "t04931",
"t13683", "t13686", "t13687", "t13688", "t13689", "t13690", "t13691", "t13769", "t13773",
"t07151", "t13684", "t07523", "t12369", "t12567", "ar00627", "ar00052", "t00352", "t07275",
"t12318", "t04931", "t13683", "t13686", "t13687", "t13688", "t13689", "t13690", "t13691",
"t07766", "t07918", "t12993", "ar00044", "t13326", "t07614", "t12414",
];
#[allow(clippy::mutex_atomic)]
pub async fn test(store: SearchStore, do_insert: bool) {
println!("Running Store query tests...");
@@ -95,8 +113,34 @@ pub async fn test(store: SearchStore, do_insert: bool) {
.unwrap();
let now = Instant::now();
let documents = Arc::new(Mutex::new(Vec::new()));
let mut mask = RoaringBitmap::new();
let mut fields = AHashMap::new();
// Global ids test
println!("Running global id filtering tests...");
test_global(store.clone()).await;
if do_insert {
let filter_ids = std::env::var("QUICK_TEST").is_ok().then(|| {
let mut ids = AHashSet::new();
for &id in ALL_IDS {
ids.insert(id.to_string());
let id = id.as_bytes();
if id.last().unwrap() > &b'0' {
let mut alt_id = id.to_vec();
*alt_id.last_mut().unwrap() -= 1;
ids.insert(String::from_utf8(alt_id).unwrap());
}
if id.last().unwrap() < &b'9' {
let mut alt_id = id.to_vec();
*alt_id.last_mut().unwrap() += 1;
ids.insert(String::from_utf8(alt_id).unwrap());
}
}
ids
});
pool.scope_fifo(|s| {
for (document_id, record) in csv::ReaderBuilder::new()
.has_headers(true)
@@ -107,18 +151,25 @@ pub async fn test(store: SearchStore, do_insert: bool) {
let record = record.unwrap();
let documents = documents.clone();
if let Some(filter_ids) = &filter_ids {
let id = record.get(1).unwrap().to_lowercase();
if !filter_ids.contains(&id) {
continue;
}
}
s.spawn_fifo(move |_| {
let mut document = IndexDocument::new(SearchIndex::Email)
.with_account_id(0)
.with_document_id(document_id as u32);
for (pos, field) in record.iter().enumerate() {
let field_id = pos as u8;
match FIELD_MAPPINGS[pos] {
EmailSearchField::From
| EmailSearchField::To
| EmailSearchField::Cc => {
| EmailSearchField::Cc
| EmailSearchField::Bcc => {
document.index_text(
FIELD_MAPPINGS[pos],
FIELD_MAPPINGS[pos].clone(),
&field.to_lowercase(),
Language::None,
);
@@ -127,7 +178,7 @@ pub async fn test(store: SearchStore, do_insert: bool) {
| EmailSearchField::Body
| EmailSearchField::Attachment => {
document.index_text(
FIELD_MAPPINGS[pos],
FIELD_MAPPINGS[pos].clone(),
&field.to_lowercase(),
Language::English,
);
@@ -143,7 +194,7 @@ pub async fn test(store: SearchStore, do_insert: bool) {
| EmailSearchField::SentAt
| EmailSearchField::Size => {
document.index_unsigned(
FIELD_MAPPINGS[pos],
FIELD_MAPPINGS[pos].clone(),
field.parse::<u64>().unwrap_or(0),
);
}
@@ -166,36 +217,60 @@ pub async fn test(store: SearchStore, do_insert: bool) {
let now = Instant::now();
let batches = documents.lock().unwrap().drain(..).collect::<Vec<_>>();
let mut chunk = Vec::new();
let mut fts_chunk = Vec::new();
print!("Inserting... ",);
let mut chunks = Vec::new();
let mut chunk = Vec::new();
for document in batches {
let chunk_instance = Instant::now();
chunk.push({
let db = db.clone();
tokio::spawn(async move { db.write(batch.build_all()).await })
});
fts_chunk.push({
let fts_store = fts_store.clone();
tokio::spawn(async move { fts_store.index(fts_batch).await })
});
if chunk.len() == 1000 {
for handle in chunk {
handle.await.unwrap().unwrap();
let mut document_id = None;
let mut to_field = None;
for (key, value) in document.fields() {
if key == &SearchField::DocumentId {
if let SearchValue::Uint(id) = value {
document_id = Some(*id as u32);
}
} else if key == &SearchField::Email(EmailSearchField::To)
&& let SearchValue::Text { value, .. } = value
{
to_field = Some(value.to_string());
}
for handle in fts_chunk {
}
let document_id = document_id.unwrap();
let to_field = to_field.unwrap();
mask.insert(document_id);
fields.insert(document_id, to_field);
chunk.push(document);
if chunk.len() == 10 {
chunks.push(chunk);
chunk = Vec::new();
}
}
if !chunk.is_empty() {
chunks.push(chunk);
}
let mut tasks = Vec::new();
for chunk in chunks {
let chunk_instance = Instant::now();
tasks.push({
let db = store.clone();
tokio::spawn(async move { db.index(chunk).await })
});
if tasks.len() == 100 {
for handle in tasks {
handle.await.unwrap().unwrap();
}
print!(" [{} ms]", chunk_instance.elapsed().as_millis());
std::io::stdout().flush().unwrap();
chunk = Vec::new();
fts_chunk = Vec::new();
tasks = Vec::new();
}
}
if !chunk.is_empty() {
for handle in chunk {
if !tasks.is_empty() {
for handle in tasks {
handle.await.unwrap().unwrap();
}
}
@@ -203,25 +278,29 @@ pub async fn test(store: SearchStore, do_insert: bool) {
println!("\nInsert took {} ms.", now.elapsed().as_millis());
}
println!("Running filter tests...");
println!("Running account filter tests...");
let now = Instant::now();
test_filter(db.clone(), fts_store).await;
test_filter(store.clone(), &fields, &mask).await;
println!("Filtering took {} ms.", now.elapsed().as_millis());
println!("Running sort tests...");
println!("Running account sort tests...");
let now = Instant::now();
test_sort(db).await;
test_sort(store.clone(), &fields, &mask).await;
println!("Sorting took {} ms.", now.elapsed().as_millis());
println!("Running unindex tests...");
let now = Instant::now();
test_unindex(store.clone(), &fields).await;
println!("Unindexing took {} ms.", now.elapsed().as_millis());
}
pub async fn test_filter(
store: SearchStore,
fields: &AHashMap<u32, &'static str>,
mask: &RoaringBitmap,
) {
async fn test_filter(store: SearchStore, fields: &AHashMap<u32, String>, mask: &RoaringBitmap) {
let can_stem = !matches!(store, SearchStore::Store(Store::MySQL(_)));
let tests = [
(
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::has_english_text(EmailSearchField::Subject, "water"),
SearchFilter::eq(EmailSearchField::ReceivedAt, 1979u32),
],
@@ -229,6 +308,7 @@ pub async fn test_filter(
),
(
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::has_keyword(EmailSearchField::From, "gelatin"),
SearchFilter::gt(EmailSearchField::ReceivedAt, 2000u32),
SearchFilter::lt(EmailSearchField::Size, 180u32),
@@ -237,27 +317,32 @@ pub async fn test_filter(
vec!["p79426", "p79427", "p79428", "p79429", "p79430"],
),
(
vec![SearchFilter::has_english_text(
EmailSearchField::Subject,
"'rustic bridge'",
)],
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::has_english_text(EmailSearchField::Subject, "'rustic bridge'"),
],
vec!["d05503"],
),
(
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::has_english_text(EmailSearchField::Subject, "'rustic'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "study"),
SearchFilter::has_english_text(
EmailSearchField::Subject,
if can_stem { "study" } else { "studies" },
),
],
vec!["d00399", "d05352"],
),
(
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::cond(
EmailSearchField::Headers,
SearchOperator::Contains,
SearchValue::KeyValues(VecMap::from_iter([(
"artist".to_string(),
"kunst mauro".to_string(),
"kunst, mauro".to_string(),
)])),
),
SearchFilter::has_keyword(EmailSearchField::Cc, "artist"),
@@ -270,10 +355,14 @@ pub async fn test_filter(
),
(
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::Not,
SearchFilter::has_keyword(EmailSearchField::From, "oil"),
SearchFilter::End,
SearchFilter::has_english_text(EmailSearchField::Body, "bequeath"),
SearchFilter::has_english_text(
EmailSearchField::Body,
if can_stem { "bequeath" } else { "bequeathed" },
),
SearchFilter::Or,
SearchFilter::And,
SearchFilter::ge(EmailSearchField::ReceivedAt, 1900u32),
@@ -294,6 +383,7 @@ pub async fn test_filter(
(
vec![
SearchFilter::And,
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::cond(
EmailSearchField::Headers,
SearchOperator::Contains,
@@ -320,20 +410,45 @@ pub async fn test_filter(
vec!["ar00039", "t12600"],
),
(
vec![
SearchFilter::has_english_text(EmailSearchField::Subject, "study"),
SearchFilter::has_keyword(EmailSearchField::From, "paper"),
SearchFilter::has_english_text(EmailSearchField::Body, "'purchased'"),
SearchFilter::Not,
SearchFilter::has_english_text(EmailSearchField::Subject, "'anatomical'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'for'"),
SearchFilter::End,
SearchFilter::gt(EmailSearchField::ReceivedAt, 1900u32),
SearchFilter::gt(EmailSearchField::Bcc, "2008".to_string()),
],
vec![
"p80042", "p80043", "p80044", "p80045", "p80203", "t11937", "t12172",
],
if can_stem {
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::has_english_text(EmailSearchField::Subject, "study"),
SearchFilter::has_keyword(EmailSearchField::From, "paper"),
SearchFilter::has_english_text(EmailSearchField::Body, "'purchased'"),
SearchFilter::Not,
SearchFilter::Or,
SearchFilter::has_english_text(EmailSearchField::Subject, "'anatomical'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'discarded'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'untitled'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'girl'"),
SearchFilter::End,
SearchFilter::End,
SearchFilter::gt(EmailSearchField::ReceivedAt, 1900u32),
SearchFilter::gt(EmailSearchField::Bcc, "2008".to_string()),
]
} else {
vec![
SearchFilter::eq(SearchField::AccountId, 0u32),
SearchFilter::Or,
SearchFilter::has_english_text(EmailSearchField::Subject, "study"),
SearchFilter::has_english_text(EmailSearchField::Subject, "studies"),
SearchFilter::End,
SearchFilter::has_keyword(EmailSearchField::From, "paper"),
SearchFilter::has_english_text(EmailSearchField::Body, "'purchased'"),
SearchFilter::Not,
SearchFilter::Or,
SearchFilter::has_english_text(EmailSearchField::Subject, "'anatomical'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'discarded'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'untitled'"),
SearchFilter::has_english_text(EmailSearchField::Subject, "'girl'"),
SearchFilter::End,
SearchFilter::End,
SearchFilter::gt(EmailSearchField::ReceivedAt, 1900u32),
SearchFilter::gt(EmailSearchField::Bcc, "2008".to_string()),
]
},
vec!["p80203", "t13209", "t13560", "t13561"],
),
];
@@ -351,17 +466,16 @@ pub async fn test_filter(
let mut results = Vec::new();
for document_id in ids {
results.push(*fields.get(&document_id).unwrap());
results.push(fields.get(&document_id).unwrap());
}
assert_eq!(results, expected_results);
}
}
pub async fn test_sort(
store: SearchStore,
fields: &AHashMap<u32, &'static str>,
mask: &RoaringBitmap,
) {
async fn test_sort(store: SearchStore, fields: &AHashMap<u32, String>, mask: &RoaringBitmap) {
let is_reversed =
matches!(store, SearchStore::Store(Store::MySQL(_))) || store.internal_fts().is_some();
let tests = [
(
vec![
@@ -409,11 +523,19 @@ pub async fn test_sort(
SearchComparator::descending(EmailSearchField::Cc),
SearchComparator::ascending(EmailSearchField::To),
],
vec![
"ar00627", "ar00052", "t00352", "t07275", "t12318", "t04931", "t13683", "t13686",
"t13687", "t13688", "t13689", "t13690", "t13691", "t07766", "t07918", "t12993",
"ar00044", "t13326", "t07614", "t12414",
],
if !is_reversed {
vec![
"ar00052", "ar00627", "t00352", "t07275", "t12318", "t04931", "t13683",
"t13686", "t13687", "t13688", "t13689", "t13690", "t13691", "t13769", "t13773",
"t07151", "t13684", "t07523", "t12369", "t12567",
]
} else {
vec![
"ar00627", "ar00052", "t00352", "t07275", "t12318", "t04931", "t13683",
"t13686", "t13687", "t13688", "t13689", "t13690", "t13691", "t07766", "t07918",
"t12993", "ar00044", "t13326", "t07614", "t12414",
]
},
),
];
@@ -430,9 +552,119 @@ pub async fn test_sort(
.unwrap();
let mut results = Vec::new();
for document_id in ids {
results.push(*fields.get(&document_id).unwrap());
for document_id in ids.into_iter().take(expected_results.len()) {
results.push(fields.get(&document_id).unwrap());
}
assert_eq!(results, expected_results);
}
}
async fn test_unindex(store: SearchStore, fields: &AHashMap<u32, String>) {
let ids = store
.query_account(
SearchQuery::new(SearchIndex::Email)
.with_mask(RoaringBitmap::from_iter(fields.keys().copied()))
.with_account_id(0)
.with_filter(SearchFilter::has_keyword(EmailSearchField::From, "paper")),
)
.await
.unwrap();
assert!(!ids.is_empty());
let expected_count = ids.len().saturating_sub(10);
let mut query = SearchQuery::new(SearchIndex::Email)
.with_account_id(0)
.with_filter(SearchFilter::Or);
for id in ids.into_iter().take(10) {
query = query.with_filter(SearchFilter::eq(SearchField::DocumentId, id));
}
query = query.with_filter(SearchFilter::End);
store.unindex(query).await.unwrap();
assert_eq!(
store
.query_account(
SearchQuery::new(SearchIndex::Email)
.with_account_id(0)
.with_filter(SearchFilter::has_keyword(EmailSearchField::From, "paper"))
.with_mask(RoaringBitmap::from_iter(fields.keys().copied())),
)
.await
.unwrap()
.len(),
expected_count
);
}
async fn test_global(store: SearchStore) {
// Insert global ids
for (id, queue_id, etyp, keywords) in [
(0, 1000u64, 1u64, "init start"),
(1, 1000u64, 2u64, "init complete"),
(2, 1001u64, 1u64, "process start"),
(3, 1001u64, 2u64, "process complete"),
(4, 1002u64, 1u64, "cleanup start"),
(5, 1002u64, 2u64, "cleanup complete"),
] {
let mut document = IndexDocument::new(SearchIndex::Tracing).with_id(id);
document.index_unsigned(TracingSearchField::QueueId, queue_id);
document.index_unsigned(TracingSearchField::EventType, etyp);
document.index_text(TracingSearchField::Keywords, keywords, Language::None);
store.index(vec![document]).await.unwrap();
}
// Query all
assert_eq!(
store
.query_global(
SearchQuery::new(SearchIndex::Tracing)
.with_filter(SearchFilter::ge(SearchField::Id, 0u64))
)
.await
.unwrap()
.into_iter()
.collect::<AHashSet<_>>(),
AHashSet::from_iter([0, 1, 2, 3, 4, 5])
);
// Query with filter
assert_eq!(
store
.query_global(
SearchQuery::new(SearchIndex::Tracing)
.with_filter(SearchFilter::gt(SearchField::Id, 1u64))
.with_filter(SearchFilter::lt(SearchField::Id, 5u64))
.with_filter(SearchFilter::has_keyword(
TracingSearchField::Keywords,
"start",
)),
)
.await
.unwrap()
.into_iter()
.collect::<AHashSet<_>>(),
AHashSet::from_iter([2, 4])
);
// Delete by filter
store
.unindex(
SearchQuery::new(SearchIndex::Tracing)
.with_filter(SearchFilter::lt(SearchField::Id, 3u64)),
)
.await
.unwrap();
assert_eq!(
store
.query_global(
SearchQuery::new(SearchIndex::Tracing)
.with_filter(SearchFilter::ge(SearchField::Id, 0u64))
)
.await
.unwrap()
.into_iter()
.collect::<AHashSet<_>>(),
AHashSet::from_iter([3, 4, 5])
);
}