diff --git a/crates/common/src/config/jmap/settings.rs b/crates/common/src/config/jmap/settings.rs index 161dbf7b..0f8c2f3d 100644 --- a/crates/common/src/config/jmap/settings.rs +++ b/crates/common/src/config/jmap/settings.rs @@ -367,28 +367,29 @@ impl JmapConfig { SearchIndex::Email, SearchIndex::Contacts, SearchIndex::Calendar, + SearchIndex::Tracing, ] { let mut fields = AHashSet::new(); - let todo = "implement"; - /*for field_str in config.values(&format!( - "jmap.index.{}.fields", - index.as_config_case() - )) { - match SearchField::try_from(field_str.1.as_str()) { - Ok(field) => { - fields.insert(field); - } - Err(_) => { - config.new_parse_error( - &format!( - "jmap.index.{}.fields", - index.as_config_case() - ), - format!("Invalid search field: {}", field_str.1), - ); - } - } - }*/ + let index_name = match index { + SearchIndex::Email => "email", + SearchIndex::Contacts => "contacts", + SearchIndex::Calendar => "calendar", + SearchIndex::Tracing => "tracing", + _ => unreachable!(), + }; + + if !config + .property_or_default::(&format!("jmap.index.{index_name}.enabled"), "true") + .unwrap_or(true) + { + continue; + } + + for (_, field) in + config.properties::(&format!("jmap.index.{index_name}.fields")) + { + fields.insert(field); + } jmap.index_fields.insert(index, fields); } diff --git a/crates/common/src/config/telemetry.rs b/crates/common/src/config/telemetry.rs index 4ea5be00..f891163f 100644 --- a/crates/common/src/config/telemetry.rs +++ b/crates/common/src/config/telemetry.rs @@ -98,7 +98,6 @@ pub struct WebhookTracer { #[cfg(feature = "enterprise")] pub struct StoreTracer { pub store: store::Store, - pub indexed: bool, } // SPDX-SnippetEnd @@ -538,9 +537,6 @@ impl Tracers { lossy: false, typ: TelemetrySubscriberType::StoreTracer(StoreTracer { store: store.clone(), - indexed: config - .property_or_default("tracing.history.indexed", "false") - .unwrap_or(false), }), }; diff --git a/crates/common/src/telemetry/tracers/store.rs b/crates/common/src/telemetry/tracers/store.rs index bfd24a18..17504c7e 100644 --- a/crates/common/src/telemetry/tracers/store.rs +++ b/crates/common/src/telemetry/tracers/store.rs @@ -53,30 +53,27 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac .any(|(k, v)| matches!((k, v), (Key::QueueId, Value::UInt(_)))) { // Serialize events - batch.set( - ValueClass::Telemetry(TelemetryClass::Span { span_id }), - serialize_events( - [span.as_ref()] - .into_iter() - .chain(events.iter().map(|event| event.as_ref())) - .chain([event.as_ref()].into_iter()), - events.len() + 2, - ), - ); - - if settings.indexed { - batch - .with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve - .with_document(span_id as u32) - .set( - ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { - due: now, - index: SearchIndex::Tracing, - is_insert: true, - }), - vec![], - ); - } + batch + .set( + ValueClass::Telemetry(TelemetryClass::Span { span_id }), + serialize_events( + [span.as_ref()] + .into_iter() + .chain(events.iter().map(|event| event.as_ref())) + .chain([event.as_ref()].into_iter()), + events.len() + 2, + ), + ) + .with_account_id((span_id >> 32) as u32) // TODO: This is hacky, improve + .with_document(span_id as u32) + .set( + ValueClass::TaskQueue(TaskQueueClass::UpdateIndex { + due: now, + index: SearchIndex::Tracing, + is_insert: true, + }), + vec![], + ); } } } diff --git a/crates/http/src/management/enterprise/telemetry.rs b/crates/http/src/management/enterprise/telemetry.rs index 318e8c0d..e9e12b1e 100644 --- a/crates/http/src/management/enterprise/telemetry.rs +++ b/crates/http/src/management/enterprise/telemetry.rs @@ -34,7 +34,7 @@ use std::future::Future; use store::{ ahash::{AHashMap, AHashSet}, search::{SearchField, SearchFilter, SearchQuery, TracingSearchField}, - write::SearchIndex, + write::{SearchIndex, now}, }; use trc::{ Collector, DeliveryEvent, EventType, Key, MetricType, QueueEvent, Value, @@ -122,24 +122,40 @@ impl TelemetryApi for Server { )); } } - let before = params + let values = params.get("values").is_some(); + if let Some(before) = params .parse::("before") .map(|t| t.into_inner()) .and_then(SnowflakeIdGenerator::from_timestamp) - .unwrap_or(0); - let after = params + { + tracing_query.push(SearchFilter::lt(SearchField::Id, before)); + } + if let Some(after) = params .parse::("after") .map(|t| t.into_inner()) .and_then(SnowflakeIdGenerator::from_timestamp) - .unwrap_or(0); - let values = params.get("values").is_some(); + { + tracing_query.push(SearchFilter::gt(SearchField::Id, after)); + } + if !tracing_query.iter().any(|f| { + matches!( + f, + SearchFilter::Operator { + field: SearchField::Tracing( + TracingSearchField::Keywords | TracingSearchField::QueueId + ) | SearchField::Id, + .. + } + ) + }) { + tracing_query.push(SearchFilter::gt( + SearchField::Id, + SnowflakeIdGenerator::from_timestamp(now() - 86400).unwrap_or_default(), + )); + } - tracing_query.push(SearchFilter::lt(SearchField::Id, after)); - tracing_query.push(SearchFilter::gt(SearchField::Id, before)); tracing_query.push(SearchFilter::End); - let todo = "if there is no search index, do full scan"; - let store = &self .core .enterprise @@ -150,7 +166,9 @@ impl TelemetryApi for Server { let span_ids = self .search_store() - .query(SearchQuery::new(SearchIndex::Tracing).with_filters(tracing_query)) + .query_global( + SearchQuery::new(SearchIndex::Tracing).with_filters(tracing_query), + ) .await?; let (total, span_ids) = if limit > 0 { diff --git a/crates/imap/src/op/search.rs b/crates/imap/src/op/search.rs index d4e785b8..f3e9ee90 100644 --- a/crates/imap/src/op/search.rs +++ b/crates/imap/src/op/search.rs @@ -587,7 +587,7 @@ impl SessionData { // Run query self.server .search_store() - .query( + .query_account( SearchQuery::new(SearchIndex::Email) .with_filters(filters) .with_comparators(comparators) diff --git a/crates/jmap/src/calendar_event/query.rs b/crates/jmap/src/calendar_event/query.rs index 7857bfd7..cd3067b6 100644 --- a/crates/jmap/src/calendar_event/query.rs +++ b/crates/jmap/src/calendar_event/query.rs @@ -212,7 +212,7 @@ impl CalendarEventQuery for Server { let results = self .search_store() - .query( + .query_account( SearchQuery::new(SearchIndex::Calendar) .with_filters(filters) .with_comparators(comparators) diff --git a/crates/jmap/src/contact/query.rs b/crates/jmap/src/contact/query.rs index b7512887..08c26e3f 100644 --- a/crates/jmap/src/contact/query.rs +++ b/crates/jmap/src/contact/query.rs @@ -305,7 +305,7 @@ impl ContactCardQuery for Server { let results = self .search_store() - .query( + .query_account( SearchQuery::new(SearchIndex::Contacts) .with_filters(filters) .with_comparators(comparators) diff --git a/crates/jmap/src/email/query.rs b/crates/jmap/src/email/query.rs index 2506fb02..832a179a 100644 --- a/crates/jmap/src/email/query.rs +++ b/crates/jmap/src/email/query.rs @@ -328,7 +328,7 @@ impl EmailQuery for Server { let results = self .search_store() - .query( + .query_account( SearchQuery::new(SearchIndex::Email) .with_filters(filters) .with_comparators(comparators) diff --git a/crates/nlp/src/language/mod.rs b/crates/nlp/src/language/mod.rs index 15d4fa5e..077472c2 100644 --- a/crates/nlp/src/language/mod.rs +++ b/crates/nlp/src/language/mod.rs @@ -9,15 +9,13 @@ pub mod search_snippet; pub mod stemmer; pub mod stopwords; -use std::borrow::Cow; - -use utils::config::utils::ParseValue; - -use crate::tokenizers::{ - Token, chinese::ChineseTokenizer, japanese::JapaneseTokenizer, word::WordTokenizer, -}; - use self::detect::LanguageDetector; +use crate::tokenizers::{ + Token, chinese::ChineseTokenizer, japanese::JapaneseTokenizer, space::SpaceTokenizer, + word::WordTokenizer, +}; +use std::borrow::Cow; +use utils::config::utils::ParseValue; pub type LanguageTokenizer<'x> = Box>> + 'x + Sync + Send>; @@ -36,6 +34,15 @@ impl Language { ChineseTokenizer::new(WordTokenizer::new(text, usize::MAX)) .filter(move |t| t.word.len() <= max_token_length), ), + Language::None => { + Box::new( + SpaceTokenizer::new(text, max_token_length).map(|word| Token { + word: word.into(), + from: 0, + to: 0, + }), + ) + } _ => Box::new(WordTokenizer::new(text, max_token_length)), } } diff --git a/crates/store/src/backend/composite/read_replica.rs b/crates/store/src/backend/composite/read_replica.rs index d9b40151..26cdef75 100644 --- a/crates/store/src/backend/composite/read_replica.rs +++ b/crates/store/src/backend/composite/read_replica.rs @@ -10,7 +10,8 @@ use crate::{ Deserialize, IterateParams, Key, Store, Stores, ValueKey, - write::{AssignedIds, Batch, ValueClass}, + search::{IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchQuery}, + write::{AssignedIds, Batch, SearchIndex, ValueClass}, }; use std::{ future::Future, @@ -285,4 +286,39 @@ impl SQLReadReplica { _ => panic!("Invalid store type"), } } + + pub async fn index(&self, documents: Vec) -> trc::Result<()> { + match &self.primary { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.index(documents).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.index(documents).await, + _ => panic!("Invalid store type"), + } + } + + pub async fn unindex(&self, query: SearchQuery) -> trc::Result { + match &self.primary { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.unindex(query).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.unindex(query).await, + _ => panic!("Invalid store type"), + } + } + + pub async fn query( + &self, + index: SearchIndex, + filters: &[SearchFilter], + sort: &[SearchComparator], + ) -> trc::Result> { + match &self.primary { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.query(index, filters, sort).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.query(index, filters, sort).await, + _ => panic!("Invalid store type"), + } + } } diff --git a/crates/store/src/backend/foundationdb/read.rs b/crates/store/src/backend/foundationdb/read.rs index fd6961d3..5b1863db 100644 --- a/crates/store/src/backend/foundationdb/read.rs +++ b/crates/store/src/backend/foundationdb/read.rs @@ -24,6 +24,11 @@ pub(crate) enum ChunkedValue { None, } +struct ChunkedValueCollector { + key: Vec, + bytes: Vec, +} + impl FdbStore { pub(crate) async fn get_value(&self, key: impl Key) -> trc::Result> where @@ -47,10 +52,9 @@ impl FdbStore { let begin = params.begin.serialize(WITH_SUBSPACE); let end = params.end.serialize(WITH_SUBSPACE); - let todo = "fix fdb range scan to support chunked reads"; - if !params.first { let mut last_key = vec![]; + let mut chunked_key: Option = None; 'outer: loop { let begin_selector = if last_key.is_empty() { @@ -78,8 +82,39 @@ impl FdbStore { let mut key = &[] as &[u8]; for value in values.iter() { key = value.key(); - if !cb(key.get(1..).unwrap_or_default(), value.value())? { - return Ok(()); + + // Check whether we are collecting a chunked value + let cb_key = key.get(1..).unwrap_or_default(); + let cb_value = value.value(); + + if let Some(chunk) = &mut chunked_key { + if chunk.key.len() + 1 == cb_key.len() + && cb_key[..chunk.key.len()] == chunk.key[..] + { + // This is a chunk of the current value + chunk.bytes.extend_from_slice(cb_value); + continue; + } else { + // Return collected chunked value + if !cb(&chunk.key, &chunk.bytes)? { + return Ok(()); + } + + // Reset collector + chunked_key = None; + } + } + + if cb_value.len() < MAX_VALUE_SIZE { + if !cb(cb_key, cb_value)? { + return Ok(()); + } + } else { + // Start collecting chunked value + chunked_key = Some(ChunkedValueCollector { + key: cb_key.to_vec(), + bytes: cb_value.to_vec(), + }); } } if values.more() { @@ -87,6 +122,11 @@ impl FdbStore { } } Ok(None) => { + // Return any chunked value collected + if let Some(chunked_key) = chunked_key.take() { + cb(&chunked_key.key, &chunked_key.bytes)?; + } + break 'outer; } Err(e) => { diff --git a/crates/store/src/backend/mysql/write.rs b/crates/store/src/backend/mysql/write.rs index 59a8ddb0..066611df 100644 --- a/crates/store/src/backend/mysql/write.rs +++ b/crates/store/src/backend/mysql/write.rs @@ -21,7 +21,7 @@ use std::time::{Duration, Instant}; enum CommitError { Mysql(mysql_async::Error), Internal(trc::Error), - Retry, + //Retry, } impl MysqlStore { @@ -45,13 +45,13 @@ impl MysqlStore { if [1062, 1213].contains(&err.code) && retry_count < MAX_COMMIT_ATTEMPTS && start.elapsed() < MAX_COMMIT_TIME => {} - CommitError::Retry => { + /*CommitError::Retry => { if retry_count > MAX_COMMIT_ATTEMPTS || start.elapsed() > MAX_COMMIT_TIME { return Err(trc::StoreEvent::AssertValueFailed .into_err() .caused_by(trc::location!())); } - } + }*/ CommitError::Mysql(err) => { return Err(into_error(err)); } diff --git a/crates/store/src/backend/postgres/write.rs b/crates/store/src/backend/postgres/write.rs index dbeaadaa..59612b64 100644 --- a/crates/store/src/backend/postgres/write.rs +++ b/crates/store/src/backend/postgres/write.rs @@ -22,7 +22,7 @@ use tokio_postgres::{IsolationLevel, error::SqlState}; enum CommitError { Postgres(tokio_postgres::Error), Internal(trc::Error), - Retry, + //Retry, } impl PostgresStore { @@ -52,7 +52,7 @@ impl PostgresStore { _ => return Err(into_error(err)), }, CommitError::Internal(err) => return Err(err), - CommitError::Retry => { + /*CommitError::Retry => { if retry_count > MAX_COMMIT_ATTEMPTS || start.elapsed() > MAX_COMMIT_TIME { @@ -60,7 +60,7 @@ impl PostgresStore { .into_err() .caused_by(trc::location!())); } - } + }*/ } let backoff = rand::rng().random_range(50..=300); diff --git a/crates/store/src/backend/rocksdb/main.rs b/crates/store/src/backend/rocksdb/main.rs index ba376d42..fe4c4c2f 100644 --- a/crates/store/src/backend/rocksdb/main.rs +++ b/crates/store/src/backend/rocksdb/main.rs @@ -74,6 +74,7 @@ impl RocksDbStore { SUBSPACE_BLOBS, SUBSPACE_TELEMETRY_SPAN, SUBSPACE_TELEMETRY_METRIC, + SUBSPACE_SEARCH_INDEX, ] { let cf_opts = Options::default(); cfs.push(ColumnFamilyDescriptor::new( diff --git a/crates/store/src/backend/sqlite/main.rs b/crates/store/src/backend/sqlite/main.rs index 60239283..2365c6ea 100644 --- a/crates/store/src/backend/sqlite/main.rs +++ b/crates/store/src/backend/sqlite/main.rs @@ -105,6 +105,7 @@ impl SqliteStore { SUBSPACE_BLOBS, SUBSPACE_TELEMETRY_SPAN, SUBSPACE_TELEMETRY_METRIC, + SUBSPACE_SEARCH_INDEX, ] { let table = char::from(table); conn.execute( diff --git a/crates/store/src/dispatch/search.rs b/crates/store/src/dispatch/search.rs index 82e82b1d..ede1037b 100644 --- a/crates/store/src/dispatch/search.rs +++ b/crates/store/src/dispatch/search.rs @@ -4,59 +4,357 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ -use super::DocumentSet; use crate::{ - SearchStore, - search::{IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchQuery}, + SearchStore, Store, + search::{ + IndexDocument, SearchComparator, SearchField, SearchFilter, SearchOperator, SearchQuery, + SearchValue, + }, write::SearchIndex, }; -use trc::AddContext; -use types::collection::Collection; +use std::cmp::Ordering; impl SearchStore { - pub async fn query(&self, query: SearchQuery) -> trc::Result> { - todo!() - /*match self { - SearchStore::Store(store) => { - store - .index_query(account_id, collection, filters, comparators) - .await + pub async fn query_account(&self, query: SearchQuery) -> trc::Result> { + // Pre-filter by mask + match query.mask.len().cmp(&1) { + Ordering::Equal => { + return Ok(vec![query.mask.min().unwrap()]); } - + Ordering::Less => { + return Ok(vec![]); + } + Ordering::Greater => {} + } + + // If the store does not support FTS, use the internal FTS store + if let Some(store) = self.internal_fts() { + return store.query_account(query).await; + } + + // If all filters and comparators are external, delegate to the underlying store + let mut account_id = u32::MAX; + let mut has_local_filters = false; + for filter in &query.filters { + match filter { + SearchFilter::Operator { + field: SearchField::AccountId, + op: SearchOperator::Equal, + value: SearchValue::Uint(id), + } => { + account_id = *id as u32; + } + SearchFilter::Operator { .. } => {} + _ => { + has_local_filters = true; + } + } + } + if account_id == u32::MAX { + return Err(trc::StoreEvent::UnexpectedError + .reason("Account ID filter is required for account queries") + .caused_by(trc::location!())); + } + + if !has_local_filters && query.comparators.iter().all(|c| c.is_external()) { + return self + .sub_query(query.index, &query.filters, &query.comparators) + .await + .map(|results| { + results + .into_iter() + .filter(|id| query.mask.contains(*id)) + .collect() + }); + } + + // Decompose filters into external and local filters + let mut filters = Vec::with_capacity(query.filters.len()); + let mut iter = query.filters.into_iter(); + let mut logical_op = None; + + while let Some(item) = iter.next() { + match &item { + SearchFilter::Operator { + field: SearchField::AccountId, + .. + } => {} + SearchFilter::Operator { .. } => { + let mut internal_item = None; + let mut depth = 0; + let mut external = Vec::with_capacity(5); + + // Add the logical operator if there is one + let in_logical_op = if let Some(op) = logical_op.take() { + external.push(op); + true + } else { + false + }; + external.push(item); + + for item in iter.by_ref() { + match item { + SearchFilter::And | SearchFilter::Or | SearchFilter::Not => { + depth += 1; + external.push(item); + } + SearchFilter::End if depth > 0 => { + depth -= 1; + external.push(item); + } + SearchFilter::Operator { .. } => { + external.push(item); + } + _ => { + internal_item = Some(item); + break; + } + } + } + + if in_logical_op { + external.push(SearchFilter::End); + } + + let mut internal_items = Vec::with_capacity(depth * 2); + if depth > 0 { + while depth > 0 { + let item = external.pop().unwrap(); + if matches!( + item, + SearchFilter::And | SearchFilter::Or | SearchFilter::Not + ) { + depth -= 1; + } + internal_items.push(item); + } + } + + // Add account id + if external.len() == 1 { + external.push(SearchFilter::Operator { + field: SearchField::AccountId, + op: SearchOperator::Equal, + value: SearchValue::Uint(account_id as u64), + }); + } else { + external.insert(0, SearchFilter::And); + external.push(SearchFilter::Operator { + field: SearchField::AccountId, + op: SearchOperator::Equal, + value: SearchValue::Uint(account_id as u64), + }); + external.push(SearchFilter::End); + } + + // Execute sub-query + filters.push(SearchFilter::DocumentSet( + self.sub_query(query.index, &external, &[]) + .await? + .into_iter() + .collect(), + )); + filters.extend(internal_items); + + if let Some(item) = internal_item { + filters.push(item); + } + } + _ => { + match &item { + SearchFilter::Or => { + logical_op = Some(SearchFilter::Or); + } + SearchFilter::And | SearchFilter::Not => { + logical_op = Some(SearchFilter::And); + } + _ => {} + } + filters.push(item); + } + } + } + + // Merge results locally + let results = SearchQuery::new(query.index) + .with_filters(filters) + .with_mask(query.mask) + .filter(); + + match results.results().len().cmp(&1) { + Ordering::Equal => Ok(vec![results.results().min().unwrap()]), + Ordering::Less => Ok(vec![]), + Ordering::Greater => { + if !query.comparators.is_empty() { + if query.comparators[0].is_external() { + let results = results.results(); + let filters = vec![ + SearchFilter::Operator { + field: SearchField::AccountId, + op: SearchOperator::Equal, + value: SearchValue::Uint(account_id as u64), + }, + SearchFilter::Operator { + field: SearchField::DocumentId, + op: SearchOperator::GreaterEqualThan, + value: SearchValue::Uint(results.min().unwrap() as u64), + }, + SearchFilter::Operator { + field: SearchField::DocumentId, + op: SearchOperator::LowerEqualThan, + value: SearchValue::Uint(results.max().unwrap() as u64), + }, + ]; + let comparators = query + .comparators + .into_iter() + .filter(|c| c.is_external()) + .collect::>(); + + self.sub_query(query.index, &filters, &comparators) + .await + .map(|items| { + items + .into_iter() + .filter(|id| results.contains(*id)) + .collect() + }) + } else { + Ok(results.with_comparators(query.comparators).into_sorted()) + } + } else { + Ok(results.results().iter().collect()) + } + } + } + } + + async fn sub_query( + &self, + index: SearchIndex, + filters: &[SearchFilter], + sort: &[SearchComparator], + ) -> trc::Result> { + match self { + SearchStore::Store(store) => match store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.query(index, filters, sort).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.query(index, filters, sort).await, + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + Store::SQLReadReplica(store) => store.query(index, filters, sort).await, + // SPDX-SnippetEnd + _ => unreachable!(), + }, + SearchStore::ElasticSearch(store) => store.query(index, filters, sort).await, + } + } + + pub async fn query_global(&self, query: SearchQuery) -> trc::Result> { + match self { + SearchStore::Store(store) => match store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => { + store + .query(query.index, &query.filters, &query.comparators) + .await + } + #[cfg(feature = "mysql")] + Store::MySQL(store) => { + store + .query(query.index, &query.filters, &query.comparators) + .await + } + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + Store::SQLReadReplica(store) => { + store + .query(query.index, &query.filters, &query.comparators) + .await + } + // SPDX-SnippetEnd + store => store.query_global(query).await, + }, SearchStore::ElasticSearch(store) => { store - .index_query(account_id, collection, filters, comparators) + .query(query.index, &query.filters, &query.comparators) .await } } - .caused_by(trc::location!())*/ } pub async fn index(&self, documents: Vec) -> trc::Result<()> { - todo!() - /*match self { - SearchStore::Store(store) => store.index_insert(document).await, - - SearchStore::ElasticSearch(store) => store.index_insert(document).await, + match self { + SearchStore::Store(store) => match store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.index(documents).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.index(documents).await, + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + Store::SQLReadReplica(store) => store.index(documents).await, + // SPDX-SnippetEnd + store => store.index(documents).await, + }, + SearchStore::ElasticSearch(store) => store.index(documents).await, } - .caused_by(trc::location!())*/ } - pub async fn unindex(&self, query: SearchQuery) -> trc::Result<()> { - todo!() - /*match self { - SearchStore::Store(store) => { - store - .index_remove(account_id, collection, document_ids) - .await - } - - SearchStore::ElasticSearch(store) => { - store - .index_remove(account_id, collection, document_ids) - .await - } + pub async fn unindex(&self, query: SearchQuery) -> trc::Result { + match self { + SearchStore::Store(store) => match store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(store) => store.unindex(query).await, + #[cfg(feature = "mysql")] + Store::MySQL(store) => store.unindex(query).await, + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + Store::SQLReadReplica(store) => store.unindex(query).await, + // SPDX-SnippetEnd + store => store.unindex(query).await.map(|_| 0), + }, + SearchStore::ElasticSearch(store) => store.unindex(query).await, + } + } + + pub fn internal_fts(&self) -> Option<&Store> { + match self { + SearchStore::Store(store) => match store { + #[cfg(feature = "postgres")] + Store::PostgreSQL(_) => None, + #[cfg(feature = "mysql")] + Store::MySQL(_) => None, + // SPDX-SnippetBegin + // SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + // SPDX-License-Identifier: LicenseRef-SEL + #[cfg(all(feature = "enterprise", any(feature = "postgres", feature = "mysql")))] + Store::SQLReadReplica(_) => None, + // SPDX-SnippetEnd + store => Some(store), + }, + _ => None, } - .caused_by(trc::location!())*/ + } +} + +impl SearchFilter { + pub fn is_external(&self) -> bool { + matches!(self, SearchFilter::Operator { .. }) + } +} + +impl SearchComparator { + pub fn is_external(&self) -> bool { + matches!(self, SearchComparator::Field { .. }) } } diff --git a/crates/store/src/dispatch/store.rs b/crates/store/src/dispatch/store.rs index 0506bbfd..10213f97 100644 --- a/crates/store/src/dispatch/store.rs +++ b/crates/store/src/dispatch/store.rs @@ -666,6 +666,7 @@ impl Store { (SUBSPACE_INDEXES, false), (SUBSPACE_TELEMETRY_SPAN, true), (SUBSPACE_TELEMETRY_METRIC, true), + (SUBSPACE_SEARCH_INDEX, true), ] { let from_key = crate::write::AnyKey { subspace, diff --git a/crates/store/src/search/bm_u32.rs b/crates/store/src/search/bm_u32.rs new file mode 100644 index 00000000..21ea56f9 --- /dev/null +++ b/crates/store/src/search/bm_u32.rs @@ -0,0 +1,265 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{ + IterateParams, Store, U32_LEN, ValueKey, + search::*, + write::{ + SEARCH_INDEX_MAX_FIELD_LEN, SearchIndex, SearchIndexClass, SearchIndexField, SearchIndexId, + SearchIndexType, ValueClass, + key::{DeserializeBigEndian, KeySerializer}, + }, +}; +use ahash::AHashMap; +use roaring::RoaringBitmap; +use std::{ + collections::hash_map::Entry, + ops::{BitAndAssign, BitOrAssign}, +}; +use trc::AddContext; +use utils::cheeky_hash::CheekyHash; + +#[derive(Default)] +pub(super) struct BitmapCache { + cache: AHashMap<(CheekyHash, u8), Option>, +} + +impl BitmapCache { + pub async fn merge_bitmaps( + &mut self, + store: &Store, + index: SearchIndex, + account_id: u32, + hashes: impl Iterator, + field: u8, + is_union: bool, + ) -> trc::Result> { + let mut result = RoaringBitmap::new(); + for (idx, hash) in hashes.enumerate() { + match self.cache.entry((hash, field)) { + Entry::Occupied(entry) => { + if let Some(bm) = entry.get() { + if is_union { + result.bitor_assign(bm); + } else if idx == 0 { + result = bm.clone(); + } else { + result.bitand_assign(bm); + if result.is_empty() { + return Ok(None); + } + } + } else if !is_union { + return Ok(None); + } + } + Entry::Vacant(entry) => { + let value = store + .get_value::(ValueKey::from(ValueClass::SearchIndex( + SearchIndexClass { + index, + typ: SearchIndexType::Term { + account_id: Some(account_id), + hash, + field, + }, + }, + ))) + .await + .caused_by(trc::location!())?; + if let Some(bm) = &value { + if is_union { + result.bitor_assign(bm); + } else if idx == 0 { + result = bm.clone(); + } else { + result.bitand_assign(bm); + if result.is_empty() { + entry.insert(value); + return Ok(None); + } + } + entry.insert(value); + } else if !is_union { + entry.insert(None); + return Ok(None); + } + } + } + } + + if !result.is_empty() { + Ok(Some(result)) + } else { + Ok(None) + } + } +} + +pub(crate) async fn range_to_bitmap( + store: &Store, + index: SearchIndex, + account_id: u32, + field_id: u8, + match_value: &[u8], + op: SearchOperator, +) -> trc::Result> { + let ((from_value, from_doc_id, from_field), (end_value, end_doc_id, end_field)) = match op { + SearchOperator::LowerThan => ((&[][..], 0, field_id), (match_value, 0, field_id)), + SearchOperator::LowerEqualThan => { + ((&[][..], 0, field_id), (match_value, u32::MAX, field_id)) + } + SearchOperator::GreaterThan => ( + (match_value, u32::MAX, field_id), + (&[][..], u32::MAX, field_id + 1), + ), + SearchOperator::GreaterEqualThan => ( + (match_value, 0, field_id), + (&[][..], u32::MAX, field_id + 1), + ), + SearchOperator::Equal | SearchOperator::Contains => ( + (match_value, 0, field_id), + (match_value, u32::MAX, field_id), + ), + }; + + let len = from_value.len().min(SEARCH_INDEX_MAX_FIELD_LEN); + let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN]; + if len > 0 { + data[..len].copy_from_slice(&from_value[..len]); + } + let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Account { + account_id, + document_id: from_doc_id, + }, + field: SearchIndexField { + field_id: from_field, + len: len as u8, + data, + }, + }, + })); + + let len = end_value.len().min(SEARCH_INDEX_MAX_FIELD_LEN); + let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN]; + if len > 0 { + data[..len].copy_from_slice(&end_value[..len]); + } + let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Account { + account_id, + document_id: end_doc_id, + }, + field: SearchIndexField { + field_id: end_field, + len: len as u8, + data, + }, + }, + })); + + let mut bm = RoaringBitmap::new(); + let prefix = KeySerializer::new(U32_LEN + 2) + .write(index.as_u8() | 1 << 6) + .write(account_id) + .write(field_id) + .finalize(); + let prefix_len = prefix.len(); + + store + .iterate( + IterateParams::new(begin, end).no_values().ascending(), + |key, _| { + if !key.starts_with(&prefix) { + return Ok(false); + } + + let id_pos = key.len() - U32_LEN; + let value = key + .get(prefix_len..id_pos) + .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?; + + let matches = match op { + SearchOperator::LowerThan => value < match_value, + SearchOperator::LowerEqualThan => value <= match_value, + SearchOperator::GreaterThan => value > match_value, + SearchOperator::GreaterEqualThan => value >= match_value, + SearchOperator::Equal | SearchOperator::Contains => value == match_value, + }; + + if matches { + bm.insert(key.deserialize_be_u32(id_pos)?); + } + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + if !bm.is_empty() { + Ok(Some(bm)) + } else { + Ok(None) + } +} + +pub(crate) async fn sort_order( + store: &Store, + index: SearchIndex, + account_id: u32, + field_id: u8, +) -> trc::Result> { + let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Account { + account_id, + document_id: 0, + }, + field: SearchIndexField { + field_id, + len: SEARCH_INDEX_MAX_FIELD_LEN as u8, + data: [0u8; SEARCH_INDEX_MAX_FIELD_LEN], + }, + }, + })); + let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Account { + account_id, + document_id: u32::MAX, + }, + field: SearchIndexField { + field_id, + len: SEARCH_INDEX_MAX_FIELD_LEN as u8, + data: [u8::MAX; SEARCH_INDEX_MAX_FIELD_LEN], + }, + }, + })); + + let mut results = AHashMap::new(); + let mut pos = 0; + store + .iterate( + IterateParams::new(begin, end).no_values().ascending(), + |key, _| { + results.insert(key.deserialize_be_u32(key.len() - U32_LEN)?, pos); + pos += 1; + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + Ok(results) +} diff --git a/crates/store/src/search/bm_u64.rs b/crates/store/src/search/bm_u64.rs new file mode 100644 index 00000000..7489423d --- /dev/null +++ b/crates/store/src/search/bm_u64.rs @@ -0,0 +1,204 @@ +/* + * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + */ + +use crate::{ + IterateParams, Store, U64_LEN, ValueKey, + search::*, + write::{ + SEARCH_INDEX_MAX_FIELD_LEN, SearchIndex, SearchIndexClass, SearchIndexField, SearchIndexId, + SearchIndexType, ValueClass, + key::{DeserializeBigEndian, KeySerializer}, + }, +}; +use ahash::AHashMap; +use roaring::RoaringTreemap; +use std::{ + collections::hash_map::Entry, + ops::{BitAndAssign, BitOrAssign}, +}; +use trc::AddContext; +use utils::cheeky_hash::CheekyHash; + +#[derive(Default)] +pub(super) struct TreemapCache { + cache: AHashMap<(CheekyHash, u8), Option>, +} + +impl TreemapCache { + pub async fn merge_treemaps( + &mut self, + store: &Store, + index: SearchIndex, + hashes: impl Iterator, + field: u8, + is_union: bool, + ) -> trc::Result> { + let mut result = RoaringTreemap::new(); + for (idx, hash) in hashes.enumerate() { + match self.cache.entry((hash, field)) { + Entry::Occupied(entry) => { + if let Some(bm) = entry.get() { + if is_union { + result.bitor_assign(bm); + } else if idx == 0 { + result = bm.clone(); + } else { + result.bitand_assign(bm); + if result.is_empty() { + return Ok(None); + } + } + } else if !is_union { + return Ok(None); + } + } + Entry::Vacant(entry) => { + let value = store + .get_value::(ValueKey::from(ValueClass::SearchIndex( + SearchIndexClass { + index, + typ: SearchIndexType::Term { + account_id: None, + hash, + field, + }, + }, + ))) + .await + .caused_by(trc::location!())?; + if let Some(bm) = &value { + if is_union { + result.bitor_assign(bm); + } else if idx == 0 { + result = bm.clone(); + } else { + result.bitand_assign(bm); + if result.is_empty() { + entry.insert(value); + return Ok(None); + } + } + entry.insert(value); + } else if !is_union { + entry.insert(None); + return Ok(None); + } + } + } + } + + if !result.is_empty() { + Ok(Some(result)) + } else { + Ok(None) + } + } +} + +pub(crate) async fn range_to_treemap( + store: &Store, + index: SearchIndex, + field_id: u8, + match_value: &[u8], + op: SearchOperator, +) -> trc::Result> { + let ((from_value, from_id, from_field), (end_value, end_id, end_field)) = match op { + SearchOperator::LowerThan => ((&[][..], 0, field_id), (match_value, 0, field_id)), + SearchOperator::LowerEqualThan => { + ((&[][..], 0, field_id), (match_value, u64::MAX, field_id)) + } + SearchOperator::GreaterThan => ( + (match_value, u64::MAX, field_id), + (&[][..], u64::MAX, field_id + 1), + ), + SearchOperator::GreaterEqualThan => ( + (match_value, 0, field_id), + (&[][..], u64::MAX, field_id + 1), + ), + SearchOperator::Equal | SearchOperator::Contains => ( + (match_value, 0, field_id), + (match_value, u64::MAX, field_id), + ), + }; + + let len = from_value.len().min(SEARCH_INDEX_MAX_FIELD_LEN); + let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN]; + if len > 0 { + data[..len].copy_from_slice(&from_value[..len]); + } + let begin = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Global { id: from_id }, + field: SearchIndexField { + field_id: from_field, + len: len as u8, + data, + }, + }, + })); + + let len = end_value.len().min(SEARCH_INDEX_MAX_FIELD_LEN); + let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN]; + if len > 0 { + data[..len].copy_from_slice(&end_value[..len]); + } + let end = ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { + index, + typ: SearchIndexType::Index { + id: SearchIndexId::Global { id: end_id }, + field: SearchIndexField { + field_id: end_field, + len: len as u8, + data, + }, + }, + })); + + let mut bm = RoaringTreemap::new(); + let prefix = KeySerializer::new(U64_LEN + 2) + .write(index.as_u8() | 1 << 6) + .write(field_id) + .finalize(); + let prefix_len = prefix.len(); + + store + .iterate( + IterateParams::new(begin, end).no_values().ascending(), + |key, _| { + if !key.starts_with(&prefix) { + return Ok(false); + } + + let id_pos = key.len() - U64_LEN; + let value = key + .get(prefix_len..id_pos) + .ok_or_else(|| trc::Error::corrupted_key(key, None, trc::location!()))?; + + let matches = match op { + SearchOperator::LowerThan => value < match_value, + SearchOperator::LowerEqualThan => value <= match_value, + SearchOperator::GreaterThan => value > match_value, + SearchOperator::GreaterEqualThan => value >= match_value, + SearchOperator::Equal | SearchOperator::Contains => value == match_value, + }; + + if matches { + bm.insert(key.deserialize_be_u64(id_pos)?); + } + + Ok(true) + }, + ) + .await + .caused_by(trc::location!())?; + + if !bm.is_empty() { + Ok(Some(bm)) + } else { + Ok(None) + } +} diff --git a/crates/store/src/search/document.rs b/crates/store/src/search/document.rs index ad24121f..31be6931 100644 --- a/crates/store/src/search/document.rs +++ b/crates/store/src/search/document.rs @@ -219,6 +219,11 @@ impl SearchFilter { Self::has_text(field, text, Language::Unknown) } + #[inline(always)] + pub fn has_keyword(field: impl Into, text: impl Into) -> Self { + Self::has_text(field, text, Language::None) + } + pub fn is_in_set(set: RoaringBitmap) -> Self { SearchFilter::DocumentSet(set) } diff --git a/crates/store/src/search/fields.rs b/crates/store/src/search/fields.rs index f238e35d..36a97f5f 100644 --- a/crates/store/src/search/fields.rs +++ b/crates/store/src/search/fields.rs @@ -33,15 +33,33 @@ impl SearchableField for EmailSearchField { } fn is_indexed(&self) -> bool { - matches!( - self, - EmailSearchField::From - | EmailSearchField::To - | EmailSearchField::Subject - | EmailSearchField::ReceivedAt - | EmailSearchField::Size - | EmailSearchField::HasAttachment, - ) + #[cfg(not(feature = "test_mode"))] + { + matches!( + self, + EmailSearchField::From + | EmailSearchField::To + | EmailSearchField::Subject + | EmailSearchField::ReceivedAt + | EmailSearchField::Size + | EmailSearchField::HasAttachment, + ) + } + + #[cfg(feature = "test_mode")] + { + matches!( + self, + EmailSearchField::From + | EmailSearchField::To + | EmailSearchField::Subject + | EmailSearchField::ReceivedAt + | EmailSearchField::SentAt + | EmailSearchField::Size + | EmailSearchField::HasAttachment + | EmailSearchField::Bcc, + ) + } } fn is_text(&self) -> bool { diff --git a/crates/store/src/search/index.rs b/crates/store/src/search/index.rs index 5ec19f59..01dacabe 100644 --- a/crates/store/src/search/index.rs +++ b/crates/store/src/search/index.rs @@ -189,6 +189,7 @@ impl Store { typ: SearchIndexType::Term { account_id: Some(account_id), hash: CheekyHash::NULL, + field: 0, }, })), ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { @@ -196,6 +197,7 @@ impl Store { typ: SearchIndexType::Term { account_id: Some(account_id), hash: CheekyHash::FULL, + field: u8::MAX, }, })), ) diff --git a/crates/store/src/search/local.rs b/crates/store/src/search/local.rs index f4d4a0d4..745885b1 100644 --- a/crates/store/src/search/local.rs +++ b/crates/store/src/search/local.rs @@ -163,6 +163,22 @@ impl SearchQuery { } impl QueryResults { + pub fn new(results: RoaringBitmap, comparators: Vec) -> Self { + Self { + results, + comparators, + } + } + + pub fn with_comparators(mut self, comparators: Vec) -> Self { + if self.comparators.is_empty() { + self.comparators = comparators; + } else { + self.comparators.extend(comparators); + } + self + } + pub fn results(&self) -> &RoaringBitmap { &self.results } diff --git a/crates/store/src/search/mod.rs b/crates/store/src/search/mod.rs index 7aeb6acc..b4ec2a44 100644 --- a/crates/store/src/search/mod.rs +++ b/crates/store/src/search/mod.rs @@ -4,6 +4,8 @@ * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL */ +pub mod bm_u32; +pub mod bm_u64; pub mod document; pub mod fields; pub mod index; @@ -18,6 +20,7 @@ use roaring::RoaringBitmap; use std::cmp::Ordering; use std::collections::hash_map::Entry; use std::ops::{BitAndAssign, BitOrAssign, BitXorAssign}; +use utils::config::utils::ParseValue; use utils::map::vec_map::VecMap; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -262,3 +265,56 @@ pub trait SearchableField: Sized { fn is_indexed(&self) -> bool; fn is_text(&self) -> bool; } + +impl ParseValue for SearchField { + fn parse_value(value: &str) -> utils::config::Result { + Ok(match value { + // Email + "email-from" => Self::Email(EmailSearchField::From), + "email-to" => Self::Email(EmailSearchField::To), + "email-cc" => Self::Email(EmailSearchField::Cc), + "email-bcc" => Self::Email(EmailSearchField::Bcc), + "email-subject" => Self::Email(EmailSearchField::Subject), + "email-body" => Self::Email(EmailSearchField::Body), + "email-attachment" => Self::Email(EmailSearchField::Attachment), + "email-received-at" => Self::Email(EmailSearchField::ReceivedAt), + "email-sent-at" => Self::Email(EmailSearchField::SentAt), + "email-size" => Self::Email(EmailSearchField::Size), + "email-has-attachment" => Self::Email(EmailSearchField::HasAttachment), + "email-headers" => Self::Email(EmailSearchField::Headers), + + // Calendar + "cal-title" => Self::Calendar(CalendarSearchField::Title), + "cal-desc" => Self::Calendar(CalendarSearchField::Description), + "cal-location" => Self::Calendar(CalendarSearchField::Location), + "cal-owner" => Self::Calendar(CalendarSearchField::Owner), + "cal-attendee" => Self::Calendar(CalendarSearchField::Attendee), + "cal-start" => Self::Calendar(CalendarSearchField::Start), + "cal-uid" => Self::Calendar(CalendarSearchField::Uid), + + // Contact + "contact-member" => Self::Contact(ContactSearchField::Member), + "contact-kind" => Self::Contact(ContactSearchField::Kind), + "contact-name" => Self::Contact(ContactSearchField::Name), + "contact-nickname" => Self::Contact(ContactSearchField::Nickname), + "contact-org" => Self::Contact(ContactSearchField::Organization), + "contact-email" => Self::Contact(ContactSearchField::Email), + "contact-phone" => Self::Contact(ContactSearchField::Phone), + "contact-online-service" => Self::Contact(ContactSearchField::OnlineService), + "contact-address" => Self::Contact(ContactSearchField::Address), + "contact-note" => Self::Contact(ContactSearchField::Note), + "contact-uid" => Self::Contact(ContactSearchField::Uid), + + // File + "file-name" => Self::File(FileSearchField::Name), + "file-content" => Self::File(FileSearchField::Content), + + // Tracing + "trace-event-type" => Self::Tracing(TracingSearchField::EventType), + "trace-queue-id" => Self::Tracing(TracingSearchField::QueueId), + "trace-keywords" => Self::Tracing(TracingSearchField::Keywords), + + _ => return Err(format!("Unknown search field: {value}")), + }) + } +} diff --git a/crates/store/src/search/query.rs b/crates/store/src/search/query.rs index ea856eb9..ebe88127 100644 --- a/crates/store/src/search/query.rs +++ b/crates/store/src/search/query.rs @@ -5,13 +5,19 @@ */ use crate::{ - Store, ValueKey, backend::MAX_TOKEN_LENGTH, search::{SearchFilter, SearchOperator, SearchQuery, SearchValue}, write::{SearchIndexClass, SearchIndexType, ValueClass} + Store, + backend::MAX_TOKEN_LENGTH, + search::{ + QueryResults, SearchComparator, SearchField, SearchFilter, SearchOperator, SearchQuery, + SearchValue, + bm_u32::{BitmapCache, range_to_bitmap, sort_order}, + bm_u64::{TreemapCache, range_to_treemap}, + }, }; -use nlp::language; -use roaring::RoaringBitmap; -use trc::AddContext; -use utils::cheeky_hash::{CheekyHash, CheekyHashMap}; -use std::{collections::hash_map::Entry, ops::{BitAndAssign, BitOrAssign, BitXorAssign}, sync::Arc}; +use nlp::{language::stemmer::Stemmer, tokenizers::space::SpaceTokenizer}; +use roaring::{RoaringBitmap, RoaringTreemap}; +use std::ops::{BitAndAssign, BitOrAssign, BitXorAssign}; +use utils::cheeky_hash::CheekyHash; impl Store { pub(crate) async fn query_account(&self, query: SearchQuery) -> trc::Result> { @@ -26,53 +32,152 @@ impl Store { let mut stack = Vec::new(); let mask = query.mask; let mut filters = query.filters.into_iter().peekable(); - let mut token_cache : CheekyHashMap> = CheekyHashMap::default(); - let account_id = None; + let mut bitmaps = BitmapCache::default(); + let mut account_id = u32::MAX; 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 field.is_text() { let (value, language) = match value { SearchValue::Text { value, language } => (value, language), - _ => return Err(trc::Error::InvalidInput("Expected text value for text field".into())), + _ => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Expected text value for text field")); + } }; - if op == &SearchOperator::Equal { - for token in language.tokenize_text(&value, MAX_TOKEN_LENGTH) { - let hash = CheekyHash::new(token.word.as_bytes()); - match token_cache.entry(hash) { - Entry::Occupied(entry) => { - entry.get().clone() - }, - Entry::Vacant(entry) => { - let value = self.get_value::(ValueKey::from(ValueClass::SearchIndex(SearchIndexClass { - index: query.index, - typ: SearchIndexType::Term { account_id, hash }, - }))).await.caused_by(trc::location!())?.map(Arc::new); - entry.insert(value.clone()); - value - - }, - } - - } else { - todo!() - } - + if op == SearchOperator::Equal { + bitmaps + .merge_bitmaps( + self, + query.index, + account_id, + language + .tokenize_text(&value, MAX_TOKEN_LENGTH) + .map(|token| CheekyHash::new(token.word.as_bytes())), + field.u8_id(), + false, + ) + .await? } else { - todo!() - + 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 union = bitmaps + .merge_bitmaps( + self, + query.index, + account_id, + [hash, stemmed_hash].into_iter().flatten(), + field.u8_id(), + true, + ) + .await?; + if let Some(union) = union { + if result.is_empty() { + result = union; + } else { + result.bitand_assign(&union); + if result.is_empty() { + break; + } + } + } else { + result.clear(); + break; + } + } + if !result.is_empty() { + Some(result) + } else { + None + } } + } else if field.is_json() { + let (key, value) = match value { + SearchValue::KeyValues(kv) => kv.into_iter().next().unwrap(), + _ => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Expected text value for text field")); + } + }; + if !value.is_empty() { + bitmaps + .merge_bitmaps( + self, + query.index, + account_id, + [CheekyHash::new(format!("{key} {value}").as_bytes())] + .into_iter(), + field.u8_id(), + false, + ) + .await? + } else { + bitmaps + .merge_bitmaps( + self, + query.index, + account_id, + [CheekyHash::new(key.as_bytes())].into_iter(), + field.u8_id(), + false, + ) + .await? + } + } else if field.is_indexed() { + let value = match value { + SearchValue::Text { value, .. } => value.into_bytes(), + SearchValue::Int(v) => (v as u64).to_be_bytes().to_vec(), + SearchValue::Uint(v) => v.to_be_bytes().to_vec(), + SearchValue::Boolean(v) => vec![v as u8], + SearchValue::KeyValues(_) => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Expected non key-value for non-text field")); + } + }; + range_to_bitmap(self, query.index, account_id, field.u8_id(), &value, op) + .await? } else { - todo!() - + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details(format!("Field {field:?} is not indexed"))); } - } - SearchFilter::DocumentSet(bitmap) => Some(Arc::new(bitmap)), + SearchFilter::DocumentSet(bitmap) => Some(bitmap), op @ (SearchFilter::And | SearchFilter::Or | SearchFilter::Not) => { stack.push(state); state = State { op, bm: None }; @@ -112,7 +217,7 @@ impl Store { } _ => unreachable!(), } - } else if let Some(ref mut result_) = result { + } else if let Some(result_) = &mut result { if let SearchFilter::Not = state.op { result_.bitxor_assign(&mask); } @@ -135,6 +240,140 @@ impl Store { } } - todo!() + let mut results = state.bm.unwrap_or_default(); + results.bitand_assign(&mask); + if results.len() > 1 && !query.comparators.is_empty() { + let mut comparators = Vec::with_capacity(query.comparators.len()); + for comparator in query.comparators { + let comparator = match comparator { + SearchComparator::Field { field, ascending } => SearchComparator::SortedSet { + set: sort_order(self, query.index, account_id, field.u8_id()).await?, + ascending, + }, + _ => comparator, + }; + + comparators.push(comparator); + } + + Ok(QueryResults::new(results, comparators).into_sorted()) + } else { + Ok(results.into_iter().collect::>()) + } + } + + pub(crate) async fn query_global(&self, query: SearchQuery) -> trc::Result> { + struct State { + pub op: SearchFilter, + pub bm: Option, + } + let mut state: State = State { + op: SearchFilter::And, + bm: None, + }; + let mut stack = Vec::new(); + let mut filters = query.filters.into_iter().peekable(); + let mut bitmaps = TreemapCache::default(); + + while let Some(filter) = filters.next() { + let result = match filter { + SearchFilter::Operator { field, op, value } => { + if field.is_text() { + let value = match value { + SearchValue::Text { value, .. } => value, + _ => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Expected text value for text field")); + } + }; + + bitmaps + .merge_treemaps( + self, + query.index, + SpaceTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH) + .map(|word| CheekyHash::new(word.as_bytes())), + field.u8_id(), + false, + ) + .await? + } else if field.is_indexed() || matches!(field, SearchField::Id) { + let value = match value { + SearchValue::Text { value, .. } => value.into_bytes(), + SearchValue::Int(v) => (v as u64).to_be_bytes().to_vec(), + SearchValue::Uint(v) => v.to_be_bytes().to_vec(), + SearchValue::Boolean(v) => vec![v as u8], + SearchValue::KeyValues(_) => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Expected non key-value for non-text field")); + } + }; + + range_to_treemap(self, query.index, field.u8_id(), &value, op).await? + } else { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details(format!("Field {field:?} is not indexed"))); + } + } + SearchFilter::DocumentSet(_) | SearchFilter::Not => { + return Err(trc::StoreEvent::UnexpectedError + .into_err() + .details("Unsupported filter in global search")); + } + op @ (SearchFilter::And | SearchFilter::Or) => { + stack.push(state); + state = State { op, bm: None }; + continue; + } + SearchFilter::End => { + if let Some(prev_state) = stack.pop() { + let bm = state.bm; + state = prev_state; + bm + } else { + break; + } + } + }; + + // Apply logical operation + if let Some(dest) = &mut state.bm { + match state.op { + SearchFilter::And => { + if let Some(result) = result { + dest.bitand_assign(result); + } else { + dest.clear(); + } + } + SearchFilter::Or => { + if let Some(result) = result { + dest.bitor_assign(result); + } + } + _ => unreachable!(), + } + } else if result.is_some() { + state.bm = result; + } else { + state.bm = Some(RoaringTreemap::new()); + } + + // And short circuit + if matches!(state.op, SearchFilter::And) && state.bm.as_ref().unwrap().is_empty() { + while let Some(filter) = filters.peek() { + if matches!(filter, SearchFilter::End) { + break; + } else { + filters.next(); + } + } + } + } + + Ok(state.bm.unwrap_or_default().into_iter().collect::>()) } } diff --git a/crates/store/src/search/term.rs b/crates/store/src/search/term.rs index 83fba133..983f5f3c 100644 --- a/crates/store/src/search/term.rs +++ b/crates/store/src/search/term.rs @@ -13,9 +13,15 @@ use crate::{ SearchIndexField, SearchIndexId, SearchIndexType, ValueClass, }, }; -use nlp::{language::stemmer::Stemmer, tokenizers::word::WordTokenizer}; +use nlp::{ + language::stemmer::Stemmer, + tokenizers::{space::SpaceTokenizer, word::WordTokenizer}, +}; use roaring::RoaringTreemap; -use utils::cheeky_hash::{CheekyBTreeMap, CheekyHash}; +use utils::{ + cheeky_hash::{CheekyBTreeMap, CheekyHash}, + map::bitmap::BitPop, +}; #[derive(Debug, PartialEq, Eq, rkyv::Serialize, rkyv::Deserialize, rkyv::Archive)] pub(crate) struct TermIndex { @@ -46,8 +52,16 @@ impl TermIndexBuilder { match field { SearchField::Id => { if let SearchValue::Uint(v) = value { + let mut data = [0u8; SEARCH_INDEX_MAX_FIELD_LEN]; + data[..U64_LEN].copy_from_slice(&v.to_be_bytes()); + fields.push(SearchIndexField { + field_id: field.u8_id(), + len: U64_LEN as u8, + data, + }); id = Some(v); } + continue; } SearchField::AccountId => { @@ -65,30 +79,42 @@ impl TermIndexBuilder { _ => {} } - let field_id = 1 << (field.u8_id() as u32); - let field = match value { SearchValue::Text { value, language } => { if field.is_text() { - if !matches!(language, Language::Unknown | Language::None) { - for token in Stemmer::new(&value, language, MAX_TOKEN_LENGTH) { - *terms - .entry(CheekyHash::new(token.word.as_bytes())) - .or_default() |= field_id; - - if let Some(stemmed_word) = token.stemmed_word { - *terms - .entry(CheekyHash::new( - format!("{}*", stemmed_word).as_bytes(), - )) - .or_default() |= field_id; + match language { + Language::Unknown => { + for token in WordTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH) { + terms + .entry(CheekyHash::new(token.word.as_bytes())) + .or_default() + .bit_push(field.u8_id()); } } - } else { - for token in WordTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH) { - *terms - .entry(CheekyHash::new(token.word.as_bytes())) - .or_default() |= field_id; + Language::None => { + for token in SpaceTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH) { + terms + .entry(CheekyHash::new(token.as_bytes())) + .or_default() + .bit_push(field.u8_id()); + } + } + _ => { + for token in Stemmer::new(&value, language, MAX_TOKEN_LENGTH) { + terms + .entry(CheekyHash::new(token.word.as_bytes())) + .or_default() + .bit_push(field.u8_id()); + + if let Some(stemmed_word) = token.stemmed_word { + terms + .entry(CheekyHash::new( + format!("{}*", stemmed_word).as_bytes(), + )) + .or_default() + .bit_push(field.u8_id()); + } + } } } } @@ -111,11 +137,15 @@ impl TermIndexBuilder { } SearchValue::KeyValues(map) => { for (key, value) in map { - *terms.entry(CheekyHash::new(key.as_bytes())).or_default() |= field_id; - for token in value.split_ascii_whitespace() { - *terms + terms + .entry(CheekyHash::new(key.as_bytes())) + .or_default() + .bit_push(field.u8_id()); + for token in SpaceTokenizer::new(value.as_str(), MAX_TOKEN_LENGTH) { + terms .entry(CheekyHash::new(format!("{key} {token}").as_bytes())) - .or_default() |= field_id; + .or_default() + .bit_push(field.u8_id()); } } @@ -141,12 +171,11 @@ impl TermIndexBuilder { data, } } - SearchValue::Boolean(v) if v => SearchIndexField { + SearchValue::Boolean(v) => SearchIndexField { field_id: field.u8_id(), len: 1, - data: [1u8; SEARCH_INDEX_MAX_FIELD_LEN], + data: [v as u8; SEARCH_INDEX_MAX_FIELD_LEN], }, - _ => continue, }; fields.push(field); @@ -200,62 +229,70 @@ impl TermIndex { document_id, } => { for term in archive.inner.terms { - batch.merge_fnc( - ValueClass::SearchIndex(SearchIndexClass { - index, - typ: SearchIndexType::Term { - account_id: Some(account_id), - hash: term.hash, - }, - }), - Params::with_capacity(1).with_u64(document_id as u64), - |params, _, bytes| { - let document_id = params.u64(0) as u32; + 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()?)) + 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::Skip) + Ok(MergeResult::Update( + RoaringBitmap::from_iter([document_id]).serialize()?, + )) } - } else { - Ok(MergeResult::Update( - RoaringBitmap::from_iter([document_id]).serialize()?, - )) - } - }, - ); + }, + ); + } } } SearchIndexId::Global { id } => { for term in archive.inner.terms { - batch.merge_fnc( - ValueClass::SearchIndex(SearchIndexClass { - index, - typ: SearchIndexType::Term { - account_id: None, - hash: term.hash, - }, - }), - Params::with_capacity(1).with_u64(id), - |params, _, bytes| { - let id = params.u64(0); + 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()?)) + 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::Skip) + Ok(MergeResult::Update( + RoaringTreemap::from_iter([id]).serialize()?, + )) } - } else { - Ok(MergeResult::Update( - RoaringTreemap::from_iter([id]).serialize()?, - )) - } - }, - ); + }, + ); + } } } } @@ -275,14 +312,14 @@ impl TermIndex { } impl ArchivedTermIndex { - pub fn has_term(&self, hash: &CheekyHash, field: &SearchField) -> bool { + /*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 { @@ -296,66 +333,74 @@ impl ArchivedTermIndex { document_id, } => { for term in self.terms.iter() { - batch.merge_fnc( - ValueClass::SearchIndex(SearchIndexClass { - index, - typ: SearchIndexType::Term { - account_id: Some(account_id), - hash: term.hash.to_native(), - }, - }), - Params::with_capacity(1).with_u64(document_id as u64), - |params, _, bytes| { - let document_id = params.u64(0) as u32; + 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()?)) + 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::Delete) + Ok(MergeResult::Skip) } } else { Ok(MergeResult::Skip) } - } else { - Ok(MergeResult::Skip) - } - }, - ); + }, + ); + } } } SearchIndexId::Global { id } => { for term in self.terms.iter() { - batch.merge_fnc( - ValueClass::SearchIndex(SearchIndexClass { - index, - typ: SearchIndexType::Term { - account_id: None, - hash: term.hash.to_native(), - }, - }), - Params::with_capacity(1).with_u64(id), - |params, _, bytes| { - let id = params.u64(0); + 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()?)) + 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::Delete) + Ok(MergeResult::Skip) } } else { Ok(MergeResult::Skip) } - } else { - Ok(MergeResult::Skip) - } - }, - ); + }, + ); + } } } } diff --git a/crates/store/src/write/key.rs b/crates/store/src/write/key.rs index 4414ff8d..8304a981 100644 --- a/crates/store/src/write/key.rs +++ b/crates/store/src/write/key.rs @@ -442,15 +442,25 @@ impl ValueClass { .write(u8::from(SyncCollection::ShareNotification)) .write(*notification_id), ValueClass::SearchIndex(index) => match &index.typ { - SearchIndexType::Term { account_id, hash } => { + SearchIndexType::Term { + account_id, + field, + hash, + } => { let class = index.index.as_u8(); if let Some(account_id) = account_id { serializer .write(class) .write(*account_id) - .write(hash.as_bytes()) + .write(hash.payload()) + .write(hash.payload_len()) + .write(*field) } else { - serializer.write(class).write(hash.as_bytes()) + serializer + .write(class) + .write(hash.payload()) + .write(hash.payload_len()) + .write(*field) } } SearchIndexType::Index { id, field } => { @@ -590,11 +600,13 @@ impl ValueClass { ValueClass::ChangeId => U32_LEN, ValueClass::ShareNotification { .. } => U32_LEN + U64_LEN + 1, ValueClass::SearchIndex(v) => match &v.typ { - SearchIndexType::Term { account_id, hash } => { + SearchIndexType::Term { + account_id, hash, .. + } => { if account_id.is_some() { - 1 + U32_LEN + hash.len() + 2 + U32_LEN + hash.len() } else { - 1 + hash.len() + 2 + hash.len() } } SearchIndexType::Index { field, .. } => 1 + field.len as usize + U64_LEN, diff --git a/crates/store/src/write/mod.rs b/crates/store/src/write/mod.rs index d0ca25a9..91c9a394 100644 --- a/crates/store/src/write/mod.rs +++ b/crates/store/src/write/mod.rs @@ -201,6 +201,7 @@ pub struct SearchIndexClass { pub enum SearchIndexType { Term { account_id: Option, + field: u8, hash: CheekyHash, }, Index { diff --git a/crates/utils/src/cheeky_hash.rs b/crates/utils/src/cheeky_hash.rs index f2a67447..cc21b787 100644 --- a/crates/utils/src/cheeky_hash.rs +++ b/crates/utils/src/cheeky_hash.rs @@ -84,6 +84,19 @@ impl CheekyHash { pub fn into_inner(self) -> [u8; HASH_SIZE] { self.0 } + + pub fn payload(&self) -> &[u8] { + let len = self.0[0] as usize; + if len <= HASH_PAYLOAD { + &self.0[1..1 + len] + } else { + &self.0[1..] + } + } + + pub fn payload_len(&self) -> u8 { + self.0[0] + } } impl AsRef<[u8]> for CheekyHash { diff --git a/crates/utils/src/config/mod.rs b/crates/utils/src/config/mod.rs index 70d39467..f4e997e8 100644 --- a/crates/utils/src/config/mod.rs +++ b/crates/utils/src/config/mod.rs @@ -10,11 +10,10 @@ pub mod ipmask; pub mod parser; pub mod utils; -use std::{collections::BTreeMap, time::Duration}; - use ahash::AHashMap; use compact_str::CompactString; use serde::Serialize; +use std::{collections::BTreeMap, time::Duration}; #[derive(Debug, Default, Serialize)] pub struct Config { diff --git a/crates/utils/src/map/bitmap.rs b/crates/utils/src/map/bitmap.rs index dc21a82f..51d86ae3 100644 --- a/crates/utils/src/map/bitmap.rs +++ b/crates/utils/src/map/bitmap.rs @@ -35,6 +35,11 @@ pub trait BitmapItem: From + Into + Sized + Copy { fn is_valid(&self) -> bool; } +pub trait BitPop { + fn bit_push(&mut self, item: u8); + fn bit_pop(&mut self) -> Option; +} + impl Bitmap { pub fn new() -> Self { Self::default() @@ -147,6 +152,38 @@ impl Bitmap { } } +impl BitPop for u32 { + fn bit_push(&mut self, item: u8) { + *self |= 1 << item; + } + + fn bit_pop(&mut self) -> Option { + if *self != 0 { + let item = 31 - self.leading_zeros(); + *self ^= 1 << item; + Some(item as u8) + } else { + None + } + } +} + +impl BitPop for u64 { + fn bit_push(&mut self, item: u8) { + *self |= 1 << item; + } + + fn bit_pop(&mut self) -> Option { + if *self != 0 { + let item = 63 - self.leading_zeros(); + *self ^= 1 << item; + Some(item as u8) + } else { + None + } + } +} + impl From> for Bitmap { fn from(value: ArchivedBitmap) -> Self { Self { diff --git a/tests/src/cluster/stress.rs b/tests/src/cluster/stress.rs index 494050a7..3dde1dea 100644 --- a/tests/src/cluster/stress.rs +++ b/tests/src/cluster/stress.rs @@ -200,28 +200,25 @@ async fn email_tests(server: Server, client: Arc) { join_all(futures).await; - let email_ids = server - .get_document_ids(TEST_USER_ID, Collection::Email) - .await - .unwrap() - .unwrap_or_default(); - let mailbox_ids = server - .get_document_ids(TEST_USER_ID, Collection::Mailbox) - .await - .unwrap() - .unwrap_or_default(); + let cache = server.get_cached_messages(TEST_USER_ID).await.unwrap(); + let email_ids = cache + .emails + .items + .iter() + .map(|e| e.document_id) + .collect::(); + let mailbox_ids = cache + .mailboxes + .items + .iter() + .map(|m| m.document_id) + .collect::(); assert_eq!(mailbox_ids.len(), 8); for mailbox in mailboxes.iter() { let mailbox_id = Id::from_str(mailbox).unwrap().document_id(); - let email_ids_in_mailbox = RoaringBitmap::from_iter( - server - .get_cached_messages(TEST_USER_ID) - .await - .unwrap() - .in_mailbox(mailbox_id) - .map(|m| m.document_id), - ); + let email_ids_in_mailbox = + RoaringBitmap::from_iter(cache.in_mailbox(mailbox_id).map(|m| m.document_id)); let mut email_ids_check = email_ids_in_mailbox.clone(); email_ids_check &= &email_ids; assert_eq!(email_ids_in_mailbox, email_ids_check); diff --git a/tests/src/directory/internal.rs b/tests/src/directory/internal.rs index 73b5d0c2..b82a1ab6 100644 --- a/tests/src/directory/internal.rs +++ b/tests/src/directory/internal.rs @@ -19,9 +19,8 @@ use directory::{ }; use mail_send::Credentials; use store::{ - BitmapKey, Store, ValueKey, - roaring::RoaringBitmap, - write::{BatchBuilder, BitmapClass, ValueClass}, + IndexKeyPrefix, IterateParams, Store, ValueKey, + write::{BatchBuilder, ValueClass}, }; use types::collection::Collection; @@ -655,7 +654,7 @@ async fn internal_directory() { BatchBuilder::new() .with_account_id(account_id) .with_collection(Collection::Email) - .create_document(document_id) + .with_document(document_id) .set(ValueClass::Property(0), "hello".as_bytes()) .build_all(), ) @@ -707,18 +706,7 @@ async fn internal_directory() { .map(|s| s.into()) .collect::>() ); - assert_eq!( - store - .get_bitmap(BitmapKey { - account_id: john_id, - collection: Collection::Email.into(), - class: BitmapClass::DocumentIds, - document_id: 0 - }) - .await - .unwrap(), - None - ); + assert!(!account_has_emails(&store, john_id).await); assert_eq!( store .get_value::(ValueKey { @@ -742,18 +730,7 @@ async fn internal_directory() { store.rcpt("jane@example.org").await.unwrap(), RcptType::Mailbox ); - assert_eq!( - store - .get_bitmap(BitmapKey { - account_id: jane_id, - collection: Collection::Email.into(), - class: BitmapClass::DocumentIds, - document_id: 0 - }) - .await - .unwrap(), - Some(RoaringBitmap::from_sorted_iter([document_id]).unwrap()) - ); + assert!(account_has_emails(&store, jane_id).await); assert_eq!( store .get_value::(ValueKey { @@ -1026,6 +1003,33 @@ impl TestInternalDirectory for Store { } } +async fn account_has_emails(store: &Store, account_id: u32) -> bool { + let mut has_emails = false; + store + .iterate( + IterateParams::new( + IndexKeyPrefix { + account_id, + collection: Collection::Email.into(), + field: 0, + }, + IndexKeyPrefix { + account_id, + collection: Collection::Email.into(), + field: u8::MAX, + }, + ) + .no_values(), + |_, _| { + has_emails = true; + Ok(false) + }, + ) + .await + .unwrap(); + has_emails +} + async fn assert_list_members( store: &Store, list_addr: &str, diff --git a/tests/src/jmap/auth/quota.rs b/tests/src/jmap/auth/quota.rs index 875ff399..35a732a3 100644 --- a/tests/src/jmap/auth/quota.rs +++ b/tests/src/jmap/auth/quota.rs @@ -6,11 +6,11 @@ use crate::{ directory::internal::TestInternalDirectory, - jmap::{JMAPTest, emails_purge_tombstoned, mail::delivery::SmtpConnection}, + jmap::{JMAPTest, mail::delivery::SmtpConnection, wait_for_index}, smtp::queue::QueuedEvents, }; use common::config::smtp::queue::QueueName; -use email::mailbox::INBOX_ID; +use email::{cache::MessageCacheFetch, mailbox::INBOX_ID}; use jmap::blob::upload::DISABLE_UPLOAD_QUOTA; use jmap_client::{ core::set::{SetErrorType, SetObject}, @@ -18,7 +18,7 @@ use jmap_client::{ }; use serde_json::json; use smtp::queue::spool::SmtpSpool; -use types::{collection::Collection, id::Id}; +use types::id::Id; pub async fn test(params: &mut JMAPTest) { println!("Running quota tests..."); @@ -158,7 +158,9 @@ pub async fn test(params: &mut JMAPTest) { for message_id in message_ids { client.email_destroy(&message_id).await.unwrap(); } - emails_purge_tombstoned(&server).await; + + // Wait for pending index tasks + wait_for_index(&server).await; assert_eq!( server .get_used_quota(account.id().document_id()) @@ -223,7 +225,8 @@ pub async fn test(params: &mut JMAPTest) { for message_id in message_ids { client.email_destroy(&message_id).await.unwrap(); } - emails_purge_tombstoned(&server).await; + // Wait for pending index tasks + wait_for_index(&server).await; assert_eq!( server .get_used_quota(account.id().document_id()) @@ -286,7 +289,8 @@ pub async fn test(params: &mut JMAPTest) { for message_id in message_ids { client.email_destroy(&message_id).await.unwrap(); } - emails_purge_tombstoned(&server).await; + // Wait for pending index tasks + wait_for_index(&server).await; assert_eq!( server .get_used_quota(account.id().document_id()) @@ -318,10 +322,11 @@ pub async fn test(params: &mut JMAPTest) { assert!(quota > 0 && quota <= 1024, "Quota is {}", quota); assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), 1, ); diff --git a/tests/src/jmap/mail/delivery.rs b/tests/src/jmap/mail/delivery.rs index 2dbc5296..82845a06 100644 --- a/tests/src/jmap/mail/delivery.rs +++ b/tests/src/jmap/mail/delivery.rs @@ -5,9 +5,7 @@ */ use crate::{ - directory::internal::TestInternalDirectory, - jmap::{JMAPTest}, - webdav::DummyWebDavClient, + directory::internal::TestInternalDirectory, jmap::JMAPTest, webdav::DummyWebDavClient, }; use email::{ cache::{MessageCacheFetch, email::MessageCacheAccess}, @@ -19,7 +17,6 @@ use tokio::{ io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines, ReadHalf, WriteHalf}, net::TcpStream, }; -use types::collection::Collection; pub async fn test(params: &mut JMAPTest) { println!("Running message delivery tests..."); @@ -68,15 +65,7 @@ pub async fn test(params: &mut JMAPTest) { .await .unwrap(); - assert_eq!( - server - .get_document_ids(john.id().document_id(), Collection::Email) - .await - .unwrap() - .unwrap() - .len(), - 1 - ); + assert_eq!(john_cache.emails.items.len(), 1); assert_eq!(john_cache.in_mailbox(INBOX_ID).count(), 1); assert_eq!(john_cache.in_mailbox(JUNK_ID).count(), 0); @@ -101,15 +90,7 @@ pub async fn test(params: &mut JMAPTest) { .await .unwrap(); - assert_eq!( - server - .get_document_ids(john.id().document_id(), Collection::Email) - .await - .unwrap() - .unwrap() - .len(), - 2 - ); + assert_eq!(john_cache.emails.items.len(), 2); assert_eq!(john_cache.in_mailbox(INBOX_ID).count(), 1); assert_eq!(john_cache.in_mailbox(JUNK_ID).count(), 1); @@ -152,15 +133,7 @@ END:VCARD .await .unwrap(); - assert_eq!( - server - .get_document_ids(john.id().document_id(), Collection::Email) - .await - .unwrap() - .unwrap() - .len(), - 3 - ); + assert_eq!(john_cache.emails.items.len(), 3); assert_eq!(john_cache.in_mailbox(INBOX_ID).count(), 2); assert_eq!(john_cache.in_mailbox(JUNK_ID).count(), 1); dav_client.delete_default_containers().await; @@ -195,10 +168,11 @@ END:VCARD for (account, num_messages) in [(john, 4), (jane, 1), (bill, 1)] { assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), num_messages, "for {}", @@ -232,10 +206,11 @@ END:VCARD for (account, num_messages) in [(john, 4), (jane, 2), (bill, 2)] { assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), num_messages, "for {}", @@ -267,10 +242,11 @@ END:VCARD for (account, num_messages) in [(john, 5), (jane, 3), (bill, 3)] { assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), num_messages, "for {}", diff --git a/tests/src/jmap/mail/query.rs b/tests/src/jmap/mail/query.rs index 95090f6e..d6428f6c 100644 --- a/tests/src/jmap/mail/query.rs +++ b/tests/src/jmap/mail/query.rs @@ -45,7 +45,7 @@ pub async fn test(params: &mut JMAPTest, insert: bool) { .with_collection(Collection::Mailbox); for mailbox_id in 1545..3010 { batch - .create_document(mailbox_id) + .with_document(mailbox_id) .custom(ObjectIndexBuilder::<(), _>::new().with_changes(Mailbox { name: format!("Mailbox {mailbox_id}"), role: SpecialUse::None, diff --git a/tests/src/jmap/mail/thread_merge.rs b/tests/src/jmap/mail/thread_merge.rs index e16932bd..b9b5fad7 100644 --- a/tests/src/jmap/mail/thread_merge.rs +++ b/tests/src/jmap/mail/thread_merge.rs @@ -9,6 +9,7 @@ use crate::{ store::deflate_test_resource, }; use ::email::{ + cache::MessageCacheFetch, mailbox::INBOX_ID, message::ingest::{EmailIngest, IngestEmail, IngestSource}, }; @@ -20,7 +21,7 @@ use store::{ ahash::{AHashMap, AHashSet}, rand::{self, Rng}, }; -use types::{collection::Collection, id::Id}; +use types::id::Id; pub async fn test(params: &mut JMAPTest) { test_single_thread(params).await; @@ -263,14 +264,15 @@ async fn test_multi_thread(params: &mut JMAPTest) { handle.await.expect("Task panicked"); } assert_eq!( - messages as u64, + messages, params .server - .get_document_ids(account_id, Collection::Email) + .get_cached_messages(account_id) .await .unwrap() - .unwrap() - .len() + .emails + .items + .len(), ); println!("Deleting all messages..."); params.destroy_all_mailboxes(account).await; diff --git a/tests/src/jmap/mod.rs b/tests/src/jmap/mod.rs index d28e0ab8..1772afeb 100644 --- a/tests/src/jmap/mod.rs +++ b/tests/src/jmap/mod.rs @@ -20,7 +20,6 @@ use base64::{ }; use common::{ Caches, Core, Data, Inner, KV_BAYES_MODEL_GLOBAL, Server, - auth::AccessToken, config::{ server::{Listeners, ServerProtocol}, telemetry::Telemetry, @@ -31,7 +30,6 @@ use common::{ config::{ConfigManager, Patterns}, }, }; -use email::message::delete::EmailDeletion; use http::HttpSessionManager; use hyper::{Method, header::AUTHORIZATION}; use imap::core::ImapSessionManager; @@ -50,13 +48,9 @@ use std::{ sync::Arc, time::Duration, }; -use store::{ - IterateParams, SUBSPACE_PROPERTY, Stores, ValueKey, - roaring::RoaringBitmap, - write::{AnyKey, TaskQueueClass, ValueClass, key::DeserializeBigEndian}, -}; +use store::{IterateParams, SUBSPACE_TASK_QUEUE, Stores, write::AnyKey}; use tokio::sync::watch; -use types::{blob_hash::BlobHash, id::Id}; +use types::id::Id; use utils::config::Config; pub mod auth; @@ -220,17 +214,13 @@ pub async fn wait_for_index(server: &Server) { .data .iterate( IterateParams::new( - ValueKey:: { - account_id: 0, - collection: 0, - document_id: 0, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: 0 }), + AnyKey { + subspace: SUBSPACE_TASK_QUEUE, + key: vec![0u8], }, - ValueKey:: { - account_id: u32::MAX, - collection: u8::MAX, - document_id: u32::MAX, - class: ValueClass::TaskQueue(TaskQueueClass::IndexEmail { due: u64::MAX }), + AnyKey { + subspace: SUBSPACE_TASK_QUEUE, + key: vec![u8::MAX, u8::MAX, u8::MAX, u8::MAX], }, ) .ascending(), @@ -252,7 +242,7 @@ pub async fn wait_for_index(server: &Server) { } pub async fn assert_is_empty(server: &Server) { - // Wait for pending FTS index tasks + // Wait for pending index tasks wait_for_index(server).await; // Delete bayes model @@ -262,9 +252,6 @@ pub async fn assert_is_empty(server: &Server) { .await .unwrap(); - // Purge accounts - emails_purge_tombstoned(server).await; - // Assert is empty server .store() @@ -283,51 +270,6 @@ pub async fn assert_is_empty(server: &Server) { server.inner.cache.messages.clear(); } -pub async fn emails_purge_tombstoned(server: &Server) { - let todo = "remove"; - let mut account_ids = RoaringBitmap::new(); - server - .core - .storage - .data - .iterate( - IterateParams::new( - AnyKey { - subspace: SUBSPACE_PROPERTY, - key: vec![0u8], - }, - AnyKey { - subspace: SUBSPACE_PROPERTY, - key: vec![u8::MAX, u8::MAX, u8::MAX, u8::MAX], - }, - ) - .no_values(), - |key, _| { - account_ids.insert(key.deserialize_be_u32(0).unwrap()); - - Ok(true) - }, - ) - .await - .unwrap(); - - for account_id in account_ids { - let do_add = server.inner.cache.access_tokens.get(&account_id).is_none(); - - if do_add { - server - .inner - .cache - .access_tokens - .insert(account_id, Arc::new(AccessToken::from_id(account_id))); - } - //server.emails_purge_tombstoned(account_id).await.unwrap(); - if do_add { - server.inner.cache.access_tokens.remove(&account_id); - } - } -} - async fn init_jmap_tests(store_id: &str, delete_if_exists: bool) -> JMAPTest { // Load and parse config let temp_dir = TempDir::new("jmap_tests", delete_if_exists); diff --git a/tests/src/jmap/server/enterprise.rs b/tests/src/jmap/server/enterprise.rs index 7b930125..5baa68d3 100644 --- a/tests/src/jmap/server/enterprise.rs +++ b/tests/src/jmap/server/enterprise.rs @@ -28,15 +28,19 @@ use common::{ }, telemetry::{ metrics::store::{Metric, MetricsStore, SharedMetricHistory}, - tracers::store::{TracingQuery, TracingStore}, + tracers::store::TracingStore, }, }; use http::management::enterprise::undelete::{UndeleteRequest, UndeleteResponse}; use imap_proto::ResponseType; +use nlp::language::Language; use std::{sync::Arc, time::Duration}; use store::{ rand::{self, Rng}, - write::now, + search::{ + SearchField, SearchFilter, SearchOperator, SearchQuery, SearchValue, TracingSearchField, + }, + write::{SearchIndex, now}, }; use trc::{ ipc::{bitset::Bitset, subscriber::SubscriberBuilder}, @@ -234,6 +238,7 @@ async fn alerts(server: &Server) { async fn tracing(params: &mut JMAPTest) { // Enable tracing let store = params.server.core.storage.data.clone(); + let query = params.server.core.storage.fts.clone(); TelemetrySubscriberType::StoreTracer(StoreTracer { store: store.clone(), }) @@ -243,16 +248,21 @@ async fn tracing(params: &mut JMAPTest) { ); // Make sure there are no span entries in the db - store.purge_spans(Duration::from_secs(0)).await.unwrap(); + store + .purge_spans(Duration::from_secs(0), Some(&query)) + .await + .unwrap(); assert_eq!( - store - .query_spans( - &[TracingQuery::EventType(EventType::Smtp( - SmtpEvent::ConnectionStart - ))], - 0, - 0 - ) + query + .query_global(SearchQuery::new(SearchIndex::Tracing).with_filters(vec![ + SearchFilter::Operator { + field: SearchField::Tracing(TracingSearchField::EventType), + op: SearchOperator::Equal, + value: SearchValue::Uint( + EventType::Smtp(SmtpEvent::ConnectionStart).id() as u64 + ) + } + ])) .await .unwrap(), Vec::::new() @@ -278,15 +288,24 @@ async fn tracing(params: &mut JMAPTest) { tokio::time::sleep(Duration::from_millis(200)).await; // Purge should not delete anything at this point - store.purge_spans(Duration::from_secs(1)).await.unwrap(); + store + .purge_spans(Duration::from_secs(1), Some(&query)) + .await + .unwrap(); // There should be a span entry in the db for span_type in [ EventType::Delivery(DeliveryEvent::AttemptStart), EventType::Smtp(SmtpEvent::ConnectionStart), ] { - let spans = store - .query_spans(&[TracingQuery::EventType(span_type)], 0, 0) + let spans = query + .query_global(SearchQuery::new(SearchIndex::Tracing).with_filters(vec![ + SearchFilter::Operator { + field: SearchField::Tracing(TracingSearchField::EventType), + op: SearchOperator::Equal, + value: SearchValue::Uint(span_type.id() as u64), + }, + ])) .await .unwrap(); assert_eq!(spans.len(), 1, "{span_type:?}"); @@ -298,30 +317,44 @@ async fn tracing(params: &mut JMAPTest) { // Try searching for keyword in ["bill@example.com", "jdoe@example.com", "example.com"] { - let spans = store - .query_spans(&[TracingQuery::Keywords(keyword.to_string())], 0, 0) + let spans = query + .query_global(SearchQuery::new(SearchIndex::Tracing).with_filters(vec![ + SearchFilter::Operator { + field: SearchField::Tracing(TracingSearchField::Keywords), + op: SearchOperator::Equal, + value: SearchValue::Text { + value: keyword.to_string(), + language: Language::None, + }, + }, + ])) .await .unwrap(); + assert_eq!(spans.len(), 2, "keyword: {keyword}"); assert!(spans[0] > spans[1], "keyword: {keyword}"); } // Purge should delete the span entries tokio::time::sleep(Duration::from_millis(800)).await; - store.purge_spans(Duration::from_secs(1)).await.unwrap(); + store + .purge_spans(Duration::from_secs(1), Some(&query)) + .await + .unwrap(); - for query in [ - TracingQuery::EventType(EventType::Smtp(SmtpEvent::ConnectionStart)), - TracingQuery::EventType(EventType::Delivery(DeliveryEvent::AttemptStart)), - TracingQuery::Keywords("bill@example.com".to_string()), - TracingQuery::Keywords("jdoe@example.com".to_string()), - TracingQuery::Keywords("example.com".to_string()), - ] { - assert_eq!( - store.query_spans(&[query], 0, 0).await.unwrap(), - Vec::::new() - ); - } + assert_eq!( + query + .query_global(SearchQuery::new(SearchIndex::Tracing).with_filters(vec![ + SearchFilter::Operator { + field: SearchField::Id, + op: SearchOperator::GreaterThan, + value: SearchValue::Uint(0), + }, + ])) + .await + .unwrap(), + Vec::::new() + ); } async fn metrics(params: &mut JMAPTest) { diff --git a/tests/src/jmap/server/purge.rs b/tests/src/jmap/server/purge.rs index 566fb820..2223cfdb 100644 --- a/tests/src/jmap/server/purge.rs +++ b/tests/src/jmap/server/purge.rs @@ -6,7 +6,7 @@ use crate::{ imap::{AssertResult, ImapConnection, Type}, - jmap::{JMAPTest}, + jmap::JMAPTest, }; use ahash::AHashSet; use common::Server; @@ -18,7 +18,7 @@ use email::{ }; use imap_proto::ResponseType; use store::{IterateParams, LogKey, U32_LEN, U64_LEN, write::key::DeserializeBigEndian}; -use types::{collection::Collection, id::Id}; +use types::id::Id; pub async fn test(params: &mut JMAPTest) { println!("Running purge tests..."); @@ -93,10 +93,11 @@ pub async fn test(params: &mut JMAPTest) { // Make sure both messages and changes are present assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), 6 ); @@ -111,10 +112,11 @@ pub async fn test(params: &mut JMAPTest) { // Only 4 messages should remain assert_eq!( server - .get_document_ids(account.id().document_id(), Collection::Email) + .get_cached_messages(account.id().document_id()) .await .unwrap() - .unwrap() + .emails + .items .len(), 4 ); diff --git a/tests/src/store/import_export.rs b/tests/src/store/import_export.rs index 9e75a3a6..3f0ab3df 100644 --- a/tests/src/store/import_export.rs +++ b/tests/src/store/import_export.rs @@ -10,8 +10,8 @@ use common::{Core, manager::backup::BackupParams}; use store::{ rand, write::{ - AnyKey, BatchBuilder, BitmapClass, BitmapHash, BlobOp, DirectoryClass, InMemoryClass, - Operation, QueueClass, QueueEvent, TagValue, ValueClass, + AnyKey, BatchBuilder, BlobOp, DirectoryClass, InMemoryClass, Operation, QueueClass, + QueueEvent, ValueClass, }, *, }; @@ -64,7 +64,7 @@ pub async fn test(db: Store) { batch.with_collection(collection); for document_id in [0, 10, 20, 30, 40] { - batch.create_document(document_id); + batch.with_document(document_id); if collection == Collection::Mailbox { batch @@ -196,7 +196,7 @@ pub async fn test(db: Store) { for account_id in [1, 2, 3, 4, 5] { batch - .create_document(account_id) + .with_document(account_id) .add( ValueClass::Directory(DirectoryClass::UsedQuota(account_id)), rand::random(), @@ -283,9 +283,6 @@ impl Snapshot { for (subspace, with_values) in [ (SUBSPACE_ACL, true), - (SUBSPACE_BITMAP_ID, false), - (SUBSPACE_BITMAP_TAG, false), - (SUBSPACE_BITMAP_TEXT, false), (SUBSPACE_DIRECTORY, true), (SUBSPACE_TASK_QUEUE, true), (SUBSPACE_INDEXES, false), @@ -303,7 +300,6 @@ impl Snapshot { (SUBSPACE_QUOTA, !is_sql), (SUBSPACE_REPORT_OUT, true), (SUBSPACE_REPORT_IN, true), - (SUBSPACE_FTS_INDEX, true), ] { let from_key = AnyKey { subspace, diff --git a/tests/src/store/mod.rs b/tests/src/store/mod.rs index c46f9d20..eccbde5c 100644 --- a/tests/src/store/mod.rs +++ b/tests/src/store/mod.rs @@ -5,17 +5,15 @@ */ pub mod blob; -pub mod import_export; +//pub mod import_export; pub mod lookup; pub mod ops; pub mod query; -use std::io::Read; - -use store::{FtsStore, Stores}; -use utils::config::Config; - use crate::AssertConfig; +use std::io::Read; +use store::{SearchStore, Stores}; +use utils::config::Config; pub struct TempDir { pub path: std::path::PathBuf, @@ -97,7 +95,7 @@ pub async fn store_tests() { //import_export::test(store.clone()).await; ops::test(store.clone()).await; - query::test(store.clone(), FtsStore::Store(store.clone()), insert).await; + query::test(SearchStore::Store(store.clone()), insert).await; if insert { temp_dir.delete(); diff --git a/tests/src/store/ops.rs b/tests/src/store/ops.rs index 3992e709..937a58ed 100644 --- a/tests/src/store/ops.rs +++ b/tests/src/store/ops.rs @@ -9,7 +9,10 @@ use std::collections::HashSet; use store::{ Store, ValueKey, rand::{self, Rng}, - write::{AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, ValueClass}, + write::{ + AlignedBytes, Archive, Archiver, BatchBuilder, DirectoryClass, MergeResult, Params, + ValueClass, + }, }; use types::collection::{Collection, SyncCollection}; @@ -111,15 +114,21 @@ pub async fn test(db: Store) { .with_account_id(0) .with_collection(Collection::Email) .with_document(0) - .merge(ValueClass::Property(3), |bytes| { - if let Some(bytes) = bytes { - Ok((u64::from_be_bytes(bytes.try_into().unwrap()) + 1) - .to_be_bytes() - .to_vec()) - } else { - Ok(0u64.to_be_bytes().to_vec()) - } - }); + .merge_fnc( + ValueClass::Property(3), + Params::with_capacity(0), + |_, _, bytes| { + if let Some(bytes) = bytes { + Ok(MergeResult::Update( + (u64::from_be_bytes(bytes.try_into().unwrap()) + 1) + .to_be_bytes() + .to_vec(), + )) + } else { + Ok(MergeResult::Update(0u64.to_be_bytes().to_vec())) + } + }, + ); db.write(builder.build_all()).await.unwrap() }) }); @@ -209,7 +218,23 @@ pub async fn test(db: Store) { .with_account_id(0) .with_collection(Collection::Email) .with_document(document_id) - .set_versioned(ValueClass::Property(5), archived_value, offset) + .set_fnc( + ValueClass::Property(5), + Params::with_capacity(2) + .with_bytes(archived_value) + .with_u64(offset), + |params, ids| { + let change_id = ids.current_change_id()?; + let archive = params.bytes(0); + let offset = params.u64(1); + + let mut bytes = Vec::with_capacity(archive.len()); + bytes.extend_from_slice(&archive[..offset as usize]); + bytes.extend_from_slice(&change_id.to_be_bytes()[..]); + bytes.push(archive.last().copied().unwrap()); // Marker + Ok(bytes) + }, + ) .log_container_insert(SyncCollection::Email); db.write(builder.build_all()) .await diff --git a/tests/src/store/query.rs b/tests/src/store/query.rs index 39dc6486..6256a3b2 100644 --- a/tests/src/store/query.rs +++ b/tests/src/store/query.rs @@ -13,18 +13,18 @@ use std::{ time::Instant, }; use store::{ - FtsStore, SerializeInfallible, + SearchStore, SerializeInfallible, ahash::AHashMap, - fts::{Field, FtsFilter, index::FtsDocument}, - query::sort::Pagination, - write::{BitmapClass, Operation, TagValue, ValueClass}, -}; -use store::{ - Store, ValueKey, - query::{Comparator, Filter}, - write::BatchBuilder, + roaring::RoaringBitmap, + search::{ + EmailSearchField, IndexDocument, SearchComparator, SearchField, SearchFilter, + SearchOperator, SearchQuery, SearchValue, + }, + write::{Operation, SearchIndex, ValueClass}, }; +use store::{Store, ValueKey, write::BatchBuilder}; use types::collection::Collection; +use utils::map::vec_map::VecMap; pub const FIELDS: [&str; 20] = [ "id", @@ -49,64 +49,44 @@ pub const FIELDS: [&str; 20] = [ "url", ]; -const COLLECTION_ID: Collection = Collection::Email; +/* + "title", // Subject + "year". // ReceivedAt + "width", // Size + "height", // SentAt + "artist" // Headers + "artistRole" // Cc + "medium", // From + "creditLine" // Body + "acquisitionYear" // Bcc + "accession_number" // To +*/ -enum FieldType { - Keyword, - Text, - FullText, - Integer, -} - -const FIELDS_OPTIONS: [FieldType; 20] = [ - FieldType::Integer, // "id", - FieldType::Keyword, // "accession_number", - FieldType::Text, // "artist", - FieldType::Keyword, // "artistRole", - FieldType::Integer, // "artistId", - FieldType::FullText, // "title", - FieldType::FullText, // "dateText", - FieldType::FullText, // "medium", - FieldType::FullText, // "creditLine", - FieldType::Integer, // "year", - FieldType::Integer, // "acquisitionYear", - FieldType::FullText, // "dimensions", - FieldType::Integer, // "width", - FieldType::Integer, // "height", - FieldType::Integer, // "depth", - FieldType::Text, // "units", - FieldType::FullText, // "inscription", - FieldType::Text, // "thumbnailCopyright", - FieldType::Text, // "thumbnailUrl", - FieldType::Text, // "url", +const FIELD_MAPPINGS: [EmailSearchField; 20] = [ + EmailSearchField::HasAttachment, // "id", + EmailSearchField::To, // "accession_number", + EmailSearchField::Headers, // "artist", + EmailSearchField::Cc, // "artistRole", + EmailSearchField::HasAttachment, // "artistId", + EmailSearchField::Subject, // "title", + EmailSearchField::HasAttachment, // "dateText", + EmailSearchField::From, // "medium", + EmailSearchField::Body, // "creditLine", + EmailSearchField::ReceivedAt, // "year", + EmailSearchField::Bcc, // "acquisitionYear", + EmailSearchField::HasAttachment, // "dimensions", + EmailSearchField::Size, // "width", + EmailSearchField::SentAt, // "height", + EmailSearchField::HasAttachment, // "depth", + EmailSearchField::HasAttachment, // "units", + EmailSearchField::HasAttachment, // "inscription", + EmailSearchField::HasAttachment, // "thumbnailCopyright", + EmailSearchField::HasAttachment, // "thumbnailUrl", + EmailSearchField::HasAttachment, // "url", ]; -#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug)] -pub struct FieldId(u8); - -impl From for u8 { - fn from(field_id: FieldId) -> Self { - field_id.0 - } -} -impl Display for FieldId { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{}", FIELDS[self.0 as usize]) - } -} - -impl FieldId { - pub fn new(field_id: u8) -> Field { - Field::Header(Self(field_id)) - } - - pub fn inner(&self) -> u8 { - self.0 - } -} - #[allow(clippy::mutex_atomic)] -pub async fn test(db: Store, fts_store: FtsStore, do_insert: bool) { +pub async fn test(store: SearchStore, do_insert: bool) { println!("Running Store query tests..."); let pool = rayon::ThreadPoolBuilder::new() @@ -128,85 +108,52 @@ pub async fn test(db: Store, fts_store: FtsStore, do_insert: bool) { let documents = documents.clone(); s.spawn_fifo(move |_| { - let mut fts_builder = FtsDocument::with_default_language(Language::English) + let mut document = IndexDocument::new(SearchIndex::Email) .with_account_id(0) - .with_collection(COLLECTION_ID) .with_document_id(document_id as u32); - let mut builder = BatchBuilder::new(); - builder - .with_account_id(0) - .with_collection(COLLECTION_ID) - .create_document(document_id as u32); for (pos, field) in record.iter().enumerate() { let field_id = pos as u8; - match FIELDS_OPTIONS[pos] { - FieldType::Text => { - if !field.is_empty() { - builder - .any_op(Operation::Index { - field: field_id, - key: field.to_lowercase().into_bytes(), - set: true, - }) - .set( - ValueClass::Property(field_id), - field.to_lowercase().into_bytes(), - ); - } + match FIELD_MAPPINGS[pos] { + EmailSearchField::From + | EmailSearchField::To + | EmailSearchField::Cc => { + document.index_text( + FIELD_MAPPINGS[pos], + &field.to_lowercase(), + Language::None, + ); } - FieldType::FullText => { - if !field.is_empty() { - fts_builder.index( - FieldId::new(field_id), - field.to_lowercase(), - Language::English, - ); - if field_id == 7 { - builder.any_op(Operation::Index { - field: field_id, - key: field.to_lowercase().into_bytes(), - set: true, - }); - } - } + EmailSearchField::Subject + | EmailSearchField::Body + | EmailSearchField::Attachment => { + document.index_text( + FIELD_MAPPINGS[pos], + &field.to_lowercase(), + Language::English, + ); } - FieldType::Integer => { - let field = field.parse::().unwrap_or(0); - builder - .any_op(Operation::Index { - field: field_id, - key: field.serialize(), - set: true, - }) - .set(ValueClass::Property(field_id), field.serialize()); + EmailSearchField::Headers => { + document.insert_key_value( + EmailSearchField::Headers, + "artist", + field.to_lowercase(), + ); } - FieldType::Keyword => { - if !field.is_empty() { - builder - .set( - ValueClass::Property(field_id), - field.to_lowercase().into_bytes(), - ) - .any_op(Operation::Bitmap { - class: BitmapClass::Tag { - field: field_id, - value: TagValue::Text( - field.to_lowercase().into_bytes(), - ), - }, - set: true, - }) - .any_op(Operation::Index { - field: field_id, - key: field.to_lowercase().into_bytes(), - set: true, - }); - } + EmailSearchField::ReceivedAt + | EmailSearchField::SentAt + | EmailSearchField::Size => { + document.index_unsigned( + FIELD_MAPPINGS[pos], + field.parse::().unwrap_or(0), + ); } - } + _ => { + continue; + } + }; } - documents.lock().unwrap().push((builder, fts_builder)); + documents.lock().unwrap().push(document); }); } }); @@ -223,7 +170,7 @@ pub async fn test(db: Store, fts_store: FtsStore, do_insert: bool) { let mut fts_chunk = Vec::new(); print!("Inserting... ",); - for (mut batch, fts_batch) in batches { + for document in batches { let chunk_instance = Instant::now(); chunk.push({ let db = db.clone(); @@ -267,120 +214,76 @@ pub async fn test(db: Store, fts_store: FtsStore, do_insert: bool) { println!("Sorting took {} ms.", now.elapsed().as_millis()); } -pub async fn test_filter(db: Store, fts: FtsStore) { - let mut fields = AHashMap::default(); - let mut fields_u8 = AHashMap::default(); - for (field_num, field) in FIELDS.iter().enumerate() { - fields.insert(field.to_string(), FieldId::new(field_num as u8)); - fields_u8.insert(field.to_string(), field_num as u8); - } - +pub async fn test_filter( + store: SearchStore, + fields: &AHashMap, + mask: &RoaringBitmap, +) { let tests = [ ( vec![ - Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![FtsFilter::has_english_text( - fields["title"].clone(), - "water", - )], - ) - .await - .unwrap(), - ), - Filter::eq(fields_u8["year"], 1979u32.serialize()), + SearchFilter::has_english_text(EmailSearchField::Subject, "water"), + SearchFilter::eq(EmailSearchField::ReceivedAt, 1979u32), ], vec!["p11293"], ), ( vec![ - Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![FtsFilter::has_english_text( - fields["medium"].clone(), - "gelatin", - )], - ) - .await - .unwrap(), - ), - Filter::gt(fields_u8["year"], 2000u32.serialize()), - Filter::lt(fields_u8["width"], 180u32.serialize()), - Filter::gt(fields_u8["width"], 0u32.serialize()), + SearchFilter::has_keyword(EmailSearchField::From, "gelatin"), + SearchFilter::gt(EmailSearchField::ReceivedAt, 2000u32), + SearchFilter::lt(EmailSearchField::Size, 180u32), + SearchFilter::gt(EmailSearchField::Size, 0u32), ], vec!["p79426", "p79427", "p79428", "p79429", "p79430"], ), ( - vec![Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![FtsFilter::has_english_text( - fields["title"].clone(), - "'rustic bridge'", - )], - ) - .await - .unwrap(), + vec![SearchFilter::has_english_text( + EmailSearchField::Subject, + "'rustic bridge'", )], vec!["d05503"], ), ( - vec![Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![ - FtsFilter::has_english_text(fields["title"].clone(), "'rustic'"), - FtsFilter::has_english_text(fields["title"].clone(), "study"), - ], - ) - .await - .unwrap(), - )], + vec![ + SearchFilter::has_english_text(EmailSearchField::Subject, "'rustic'"), + SearchFilter::has_english_text(EmailSearchField::Subject, "study"), + ], vec!["d00399", "d05352"], ), ( vec![ - Filter::contains(fields_u8["artist"], "kunst, mauro"), - Filter::is_in_bitmap(fields_u8["artistRole"], TagValue::Text("artist".into())), - Filter::Or, - Filter::eq(fields_u8["year"], 1969u32.serialize()), - Filter::eq(fields_u8["year"], 1971u32.serialize()), - Filter::End, + SearchFilter::cond( + EmailSearchField::Headers, + SearchOperator::Contains, + SearchValue::KeyValues(VecMap::from_iter([( + "artist".to_string(), + "kunst mauro".to_string(), + )])), + ), + SearchFilter::has_keyword(EmailSearchField::Cc, "artist"), + SearchFilter::Or, + SearchFilter::eq(EmailSearchField::ReceivedAt, 1969u32), + SearchFilter::eq(EmailSearchField::ReceivedAt, 1971u32), + SearchFilter::End, ], vec!["p01764", "t05843"], ), ( vec![ - Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![ - FtsFilter::Not, - FtsFilter::has_english_text(fields["medium"].clone(), "oil"), - FtsFilter::End, - FtsFilter::has_english_text(fields["creditLine"].clone(), "bequeath"), - ], - ) - .await - .unwrap(), - ), - Filter::Or, - Filter::And, - Filter::ge(fields_u8["year"], 1900u32.serialize()), - Filter::lt(fields_u8["year"], 1910u32.serialize()), - Filter::End, - Filter::And, - Filter::ge(fields_u8["year"], 2000u32.serialize()), - Filter::lt(fields_u8["year"], 2010u32.serialize()), - Filter::End, - Filter::End, + SearchFilter::Not, + SearchFilter::has_keyword(EmailSearchField::From, "oil"), + SearchFilter::End, + SearchFilter::has_english_text(EmailSearchField::Body, "bequeath"), + SearchFilter::Or, + SearchFilter::And, + SearchFilter::ge(EmailSearchField::ReceivedAt, 1900u32), + SearchFilter::lt(EmailSearchField::ReceivedAt, 1910u32), + SearchFilter::End, + SearchFilter::And, + SearchFilter::ge(EmailSearchField::ReceivedAt, 2000u32), + SearchFilter::lt(EmailSearchField::ReceivedAt, 2010u32), + SearchFilter::End, + SearchFilter::End, ], vec![ "n02478", "n02479", "n03568", "n03658", "n04327", "n04328", "n04721", "n04739", @@ -390,60 +293,43 @@ pub async fn test_filter(db: Store, fts: FtsStore) { ), ( vec![ - Filter::And, - Filter::contains(fields_u8["artist"], "warhol"), - Filter::Not, - Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![FtsFilter::has_english_text( - fields["title"].clone(), - "'campbell'", - )], - ) - .await - .unwrap(), + SearchFilter::And, + SearchFilter::cond( + EmailSearchField::Headers, + SearchOperator::Contains, + SearchValue::KeyValues(VecMap::from_iter([( + "artist".to_string(), + "warhol".to_string(), + )])), ), - Filter::End, - Filter::Not, - Filter::Or, - Filter::gt(fields_u8["year"], 1980u32.serialize()), - Filter::And, - Filter::gt(fields_u8["width"], 500u32.serialize()), - Filter::gt(fields_u8["height"], 500u32.serialize()), - Filter::End, - Filter::End, - Filter::End, - Filter::eq(fields_u8["acquisitionYear"], 2008u32.serialize()), - Filter::End, + SearchFilter::Not, + SearchFilter::has_english_text(EmailSearchField::Subject, "'campbell'"), + SearchFilter::End, + SearchFilter::Not, + SearchFilter::Or, + SearchFilter::gt(EmailSearchField::ReceivedAt, 1980u32), + SearchFilter::And, + SearchFilter::gt(EmailSearchField::Size, 500u32), + SearchFilter::gt(EmailSearchField::SentAt, 500u32), + SearchFilter::End, + SearchFilter::End, + SearchFilter::End, + SearchFilter::eq(EmailSearchField::Bcc, "2008".to_string()), + SearchFilter::End, ], vec!["ar00039", "t12600"], ), ( vec![ - Filter::is_in_set( - fts.query( - 0, - COLLECTION_ID, - vec![ - FtsFilter::has_english_text(fields["title"].clone(), "study"), - FtsFilter::has_english_text(fields["medium"].clone(), "paper"), - FtsFilter::has_english_text( - fields["creditLine"].clone(), - "'purchased'", - ), - FtsFilter::Not, - FtsFilter::has_english_text(fields["title"].clone(), "'anatomical'"), - FtsFilter::has_english_text(fields["title"].clone(), "'for'"), - FtsFilter::End, - ], - ) - .await - .unwrap(), - ), - Filter::gt(fields_u8["year"], 1900u32.serialize()), - Filter::gt(fields_u8["acquisitionYear"], 2000u32.serialize()), + 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", @@ -451,54 +337,44 @@ pub async fn test_filter(db: Store, fts: FtsStore) { ), ]; - for (filter, expected_results) in tests { + for (filters, expected_results) in tests { //println!("Running test: {:?}", filter); - let docset = db.filter(0, COLLECTION_ID, filter).await.unwrap(); - let sorted_docset = db - .sort( - docset, - vec![Comparator::ascending(fields_u8["accession_number"])], - Pagination::new(0, 0, None, 0), + let ids = store + .query_account( + SearchQuery::new(SearchIndex::Email) + .with_filters(filters) + .with_comparator(SearchComparator::ascending(EmailSearchField::To)) + .with_mask(mask.clone()), ) .await .unwrap(); let mut results = Vec::new(); - for document_id in sorted_docset.ids { - results.push( - db.get_value::(ValueKey { - account_id: 0, - collection: COLLECTION_ID.into(), - document_id: document_id.document_id(), - class: ValueClass::Property(fields_u8["accession_number"]), - }) - .await - .unwrap() - .unwrap(), - ); + for document_id in ids { + results.push(*fields.get(&document_id).unwrap()); } assert_eq!(results, expected_results); } } -pub async fn test_sort(db: Store) { - let mut fields = AHashMap::default(); - for (field_num, field) in FIELDS.iter().enumerate() { - fields.insert(field.to_string(), field_num as u8); - } - +pub async fn test_sort( + store: SearchStore, + fields: &AHashMap, + mask: &RoaringBitmap, +) { let tests = [ ( vec![ - Filter::gt(fields["year"], 0u32.serialize()), - Filter::gt(fields["acquisitionYear"], 0u32.serialize()), - Filter::gt(fields["width"], 0u32.serialize()), + SearchFilter::eq(SearchField::AccountId, 0u32), + SearchFilter::gt(EmailSearchField::ReceivedAt, 0u32), + SearchFilter::gt(EmailSearchField::Bcc, "0000".to_string()), + SearchFilter::gt(EmailSearchField::Size, 0u32), ], vec![ - Comparator::descending(fields["year"]), - Comparator::ascending(fields["acquisitionYear"]), - Comparator::ascending(fields["width"]), - Comparator::descending(fields["accession_number"]), + SearchComparator::descending(EmailSearchField::ReceivedAt), + SearchComparator::ascending(EmailSearchField::Bcc), + SearchComparator::ascending(EmailSearchField::Size), + SearchComparator::descending(EmailSearchField::To), ], vec![ "t13655", "t13811", "p13352", "p13351", "p13350", "p13349", "p13348", "p13347", @@ -512,12 +388,13 @@ pub async fn test_sort(db: Store) { ), ( vec![ - Filter::gt(fields["width"], 0u32.serialize()), - Filter::gt(fields["height"], 0u32.serialize()), + SearchFilter::eq(SearchField::AccountId, 0u32), + SearchFilter::gt(EmailSearchField::Size, 0u32), + SearchFilter::gt(EmailSearchField::SentAt, 0u32), ], vec![ - Comparator::descending(fields["width"]), - Comparator::ascending(fields["height"]), + SearchComparator::descending(EmailSearchField::Size), + SearchComparator::ascending(EmailSearchField::SentAt), ], vec![ "t03681", "t12601", "ar00166", "t12625", "t12915", "p04182", "t06483", "ar00703", @@ -526,11 +403,11 @@ pub async fn test_sort(db: Store) { ], ), ( - vec![], + vec![SearchFilter::eq(SearchField::AccountId, 0u32)], vec![ - Comparator::descending(fields["medium"]), - Comparator::descending(fields["artistRole"]), - Comparator::ascending(fields["accession_number"]), + SearchComparator::descending(EmailSearchField::From), + SearchComparator::descending(EmailSearchField::Cc), + SearchComparator::ascending(EmailSearchField::To), ], vec![ "ar00627", "ar00052", "t00352", "t07275", "t12318", "t04931", "t13683", "t13686", @@ -540,32 +417,21 @@ pub async fn test_sort(db: Store) { ), ]; - for (filter, sort, expected_results) in tests { + for (filters, comparators, expected_results) in tests { //println!("Running test: {:?}", sort); - let docset = db.filter(0, COLLECTION_ID, filter).await.unwrap(); - - let sorted_docset = db - .sort( - docset, - sort, - Pagination::new(expected_results.len(), 0, None, 0), + let ids = store + .query_account( + SearchQuery::new(SearchIndex::Email) + .with_filters(filters) + .with_comparators(comparators) + .with_mask(mask.clone()), ) .await .unwrap(); let mut results = Vec::new(); - for document_id in sorted_docset.ids { - results.push( - db.get_value::(ValueKey { - account_id: 0, - collection: COLLECTION_ID.into(), - document_id: document_id.document_id(), - class: ValueClass::Property(fields["accession_number"]), - }) - .await - .unwrap() - .unwrap(), - ); + for document_id in ids { + results.push(*fields.get(&document_id).unwrap()); } assert_eq!(results, expected_results); }