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

Database schema optimization - part 5

Browse Source
This commit is contained in:
mdecimus
2025-11-04 17:45:37 +01:00
parent 14bce0a346
commit a7815ffd6b
33 changed files with 1891 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/common/src/config/jmap/settings.rs
View File

@@ -82,7 +82,6 @@ pub struct JmapConfig {
pub encrypt_append: bool,
pub index_batch_size: usize,
pub index_all_headers: bool,
pub index_fields: AHashMap<SearchIndex, AHashSet<SearchField>>,
pub capabilities: BaseCapabilities,
@@ -358,9 +357,6 @@ impl JmapConfig {
.property("jmap.calendar.parse.max-items")
.unwrap_or(10),
index_batch_size: config.property("jmap.index.batch-size").unwrap_or(100),
index_all_headers: config
.property_or_default("jmap.index.email.all-headers", "false")
.unwrap_or(false),
index_fields: AHashMap::new(),
default_folders,
shared_folder,

41
crates/common/src/telemetry/tracers/store.rs
View File

@@ -71,7 +71,7 @@ pub(crate) fn spawn_store_tracer(builder: SubscriberBuilder, settings: StoreTrac
.set(
ValueClass::TaskQueue(TaskQueueClass::UpdateIndex {
due: now,
index: SearchIndex::TracingSpan,
index: SearchIndex::Tracing,
is_insert: true,
}),
vec![],
@@ -149,7 +149,7 @@ impl TracingStore for Store {
if let Some(search_store) = search_store {
search_store
.unindex(
SearchQuery::new(SearchIndex::TracingSpan)
SearchQuery::new(SearchIndex::Tracing)
.with_filter(SearchFilter::lt(SearchField::Id, until_span_id)),
)
.await
@@ -246,7 +246,7 @@ pub fn build_span_document(
events: Vec<Event<EventDetails>>,
index_fields: &AHashSet<SearchField>,
) -> IndexDocument {
let mut document = IndexDocument::with_default_language(Language::None);
let mut document = IndexDocument::new(SearchIndex::Tracing);
document.index_unsigned(SearchField::Id, span_id);
@@ -264,25 +264,30 @@ pub fn build_span_document(
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::QueueId.into())
{
document.insert_keyword(TracingSearchField::QueueId, queue_id);
document.index_unsigned(TracingSearchField::QueueId, queue_id);
}
}
(Key::From | Key::To | Key::Domain | Key::Hostname, Value::String(address)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::Address.into())
|| index_fields.contains(&TracingSearchField::Keywords.into())
{
document.insert_keyword(TracingSearchField::Address, address.into_string());
document.index_text(
TracingSearchField::Keywords,
&address,
Language::Unknown,
);
}
}
(Key::To, Value::Array(value)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::Address.into())
|| index_fields.contains(&TracingSearchField::Keywords.into())
{
for value in value {
if let Value::String(address) = value {
document.insert_keyword(
TracingSearchField::Address,
address.into_string(),
document.index_text(
TracingSearchField::Keywords,
&address,
Language::Unknown,
);
}
}
@@ -290,16 +295,24 @@ pub fn build_span_document(
}
(Key::RemoteIp, Value::Ipv4(ip)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::RemoteIp.into())
|| index_fields.contains(&TracingSearchField::Keywords.into())
{
document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string());
document.index_text(
TracingSearchField::Keywords,
&ip.to_string(),
Language::Unknown,
);
}
}
(Key::RemoteIp, Value::Ipv6(ip)) => {
if index_fields.is_empty()
|| index_fields.contains(&TracingSearchField::RemoteIp.into())
|| index_fields.contains(&TracingSearchField::Keywords.into())
{
document.insert_keyword(TracingSearchField::RemoteIp, ip.to_string());
document.index_text(
TracingSearchField::Keywords,
&ip.to_string(),
Language::Unknown,
);
}
}

69
crates/email/src/message/index/search.rs
View File

@@ -9,14 +9,17 @@ use crate::message::{
metadata::{ArchivedMessageMetadata, ArchivedMetadataPartType, DecodedPartContent},
};
use mail_parser::{
ArchivedHeaderName, ArchivedHeaderValue, DateTime, HeaderName, core::rkyv::ArchivedGetHeader,
decoders::html::html_to_text,
ArchivedHeaderName, ArchivedHeaderValue, DateTime, core::rkyv::ArchivedGetHeader,
decoders::html::html_to_text, parsers::fields::thread::thread_name,
};
use nlp::language::{
Language,
detect::{LanguageDetector, MIN_LANGUAGE_SCORE},
};
use nlp::language::Language;
use std::borrow::Cow;
use store::{
ahash::AHashSet,
search::{EmailSearchField, IndexDocument, SearchField},
write::SearchIndex,
};
impl ArchivedMessageMetadata {
@@ -24,11 +27,11 @@ impl ArchivedMessageMetadata {
&self,
raw_message: &[u8],
index_fields: &AHashSet<SearchField>,
index_all_headers: bool,
) -> IndexDocument {
let mut detector = LanguageDetector::new();
let mut language = Language::Unknown;
let message_contents = &self.contents[0];
let mut document = IndexDocument::with_default_language(language);
let mut document = IndexDocument::new(SearchIndex::Email);
if index_fields.is_empty()
|| index_fields.contains(&SearchField::Email(EmailSearchField::ReceivedAt))
@@ -68,7 +71,7 @@ impl ArchivedMessageMetadata {
document.index_text(
SearchField::Email(EmailSearchField::From),
value,
Language::Unknown,
Language::None,
);
});
}
@@ -81,7 +84,7 @@ impl ArchivedMessageMetadata {
document.index_text(
SearchField::Email(EmailSearchField::To),
value,
Language::Unknown,
Language::None,
);
});
}
@@ -94,7 +97,7 @@ impl ArchivedMessageMetadata {
document.index_text(
SearchField::Email(EmailSearchField::Cc),
value,
Language::Unknown,
Language::None,
);
});
}
@@ -107,7 +110,7 @@ impl ArchivedMessageMetadata {
document.index_text(
SearchField::Email(EmailSearchField::Bcc),
value,
Language::Unknown,
Language::None,
);
});
}
@@ -118,6 +121,12 @@ impl ArchivedMessageMetadata {
.contains(&SearchField::Email(EmailSearchField::Subject)))
&& let Some(subject) = header.value.as_text()
{
let subject = thread_name(subject);
if part_language.is_unknown() {
detector.detect(subject, MIN_LANGUAGE_SCORE);
}
document.index_text(
SearchField::Email(EmailSearchField::Subject),
subject,
@@ -138,17 +147,18 @@ impl ArchivedMessageMetadata {
}
}
_ => {
let field = SearchField::Email(EmailSearchField::Header(
match HeaderName::from(&header.name) {
HeaderName::Other(name) => Cow::Owned(name.into_owned()),
header_name => Cow::Borrowed(header_name.as_static_str()),
},
));
if index_all_headers || index_fields.contains(&field) {
if index_fields.contains(&SearchField::Email(EmailSearchField::Headers))
{
let mut value = String::new();
header.value.visit_text(|text| {
document.index_text(field.clone(), text, Language::Unknown);
value.push_str(text);
});
document.insert_key_value(
EmailSearchField::Headers,
header.name.as_str().to_string(),
value,
);
}
}
}
@@ -172,6 +182,10 @@ impl ArchivedMessageMetadata {
if index_fields.is_empty()
|| index_fields.contains(&SearchField::Email(EmailSearchField::Body))
{
if part_language.is_unknown() {
detector.detect(text.as_ref(), MIN_LANGUAGE_SCORE);
}
document.index_text(
SearchField::Email(EmailSearchField::Body),
text.as_ref(),
@@ -181,6 +195,10 @@ impl ArchivedMessageMetadata {
} else if index_fields.is_empty()
|| index_fields.contains(&SearchField::Email(EmailSearchField::Attachment))
{
if part_language.is_unknown() {
detector.detect(text.as_ref(), MIN_LANGUAGE_SCORE);
}
document.index_text(
SearchField::Email(EmailSearchField::Attachment),
text.as_ref(),
@@ -203,6 +221,10 @@ impl ArchivedMessageMetadata {
.headers
.header_value(&ArchivedHeaderName::Subject)
{
if nested_message_language.is_unknown() {
detector.detect(subject.as_ref(), MIN_LANGUAGE_SCORE);
}
document.index_text(
SearchField::Email(EmailSearchField::Attachment),
subject.as_ref(),
@@ -226,6 +248,11 @@ impl ArchivedMessageMetadata {
) => html_to_text(html.as_ref()).into(),
_ => unreachable!(),
};
if language.is_unknown() {
detector.detect(text.as_ref(), MIN_LANGUAGE_SCORE);
}
document.index_text(
SearchField::Email(EmailSearchField::Attachment),
text.as_ref(),
@@ -240,6 +267,10 @@ impl ArchivedMessageMetadata {
}
}
if let Some(detected_language) = detector.most_frequent_language() {
document.set_unknown_language(detected_language);
}
let has_attachment =
document.has_field(&(SearchField::Email(EmailSearchField::Attachment)));

42
crates/groupware/src/calendar/index.rs
View File

@@ -18,7 +18,10 @@ use calcard::icalendar::{
ICalendarParameterValue, ICalendarProperty, ICalendarValue,
};
use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject};
use nlp::language::Language;
use nlp::language::{
Language,
detect::{LanguageDetector, MIN_LANGUAGE_SCORE},
};
use store::{
search::{CalendarSearchField, IndexDocument, SearchField},
write::{IndexPropertyClass, SearchIndex, ValueClass},
@@ -338,7 +341,7 @@ impl ArchivedCalendarEvent {
impl ArchivedCalendarEvent {
pub fn index_document(&self, index_fields: &AHashSet<SearchField>) -> IndexDocument {
let mut document = IndexDocument::with_default_language(Language::Unknown);
let mut document = IndexDocument::new(SearchIndex::Calendar);
if index_fields.is_empty()
|| index_fields.contains(&SearchField::Calendar(CalendarSearchField::Start))
@@ -346,6 +349,7 @@ impl ArchivedCalendarEvent {
document.index_integer(CalendarSearchField::Start, self.data.event_range_start());
}
let mut detector = LanguageDetector::new();
for component in self
.data
.event
@@ -354,13 +358,15 @@ impl ArchivedCalendarEvent {
.filter(|e| e.component_type.is_scheduling_object())
{
for entry in component.entries.iter() {
let field = SearchField::Calendar(match entry.name {
ArchivedICalendarProperty::Summary => CalendarSearchField::Title,
ArchivedICalendarProperty::Description => CalendarSearchField::Description,
ArchivedICalendarProperty::Location => CalendarSearchField::Location,
ArchivedICalendarProperty::Organizer => CalendarSearchField::Owner,
ArchivedICalendarProperty::Attendee => CalendarSearchField::Attendee,
ArchivedICalendarProperty::Uid => CalendarSearchField::Uid,
let (is_lang, field) = SearchField::Calendar(match entry.name {
ArchivedICalendarProperty::Summary => (true, CalendarSearchField::Title),
ArchivedICalendarProperty::Description => {
(true, CalendarSearchField::Description)
}
ArchivedICalendarProperty::Location => (false, CalendarSearchField::Location),
ArchivedICalendarProperty::Organizer => (false, CalendarSearchField::Owner),
ArchivedICalendarProperty::Attendee => (false, CalendarSearchField::Attendee),
ArchivedICalendarProperty::Uid => (false, CalendarSearchField::Uid),
_ => continue,
});
@@ -379,16 +385,24 @@ impl ArchivedCalendarEvent {
_ => None,
}))
{
document.index_text(
field.clone(),
value.strip_prefix("mailto:").unwrap_or(value),
Language::Unknown,
);
let value = value.strip_prefix("mailto:").unwrap_or(value);
let lang = if is_lang {
detector.detect(value, MIN_LANGUAGE_SCORE);
Language::Unknown
} else {
Language::None
};
document.index_text(field.clone(), value, lang);
}
}
}
}
if let Some(detected_language) = detector.most_frequent_language() {
document.set_unknown_language(detected_language);
}
document
}
}

79
crates/groupware/src/contact/index.rs
View File

@@ -14,10 +14,13 @@ use calcard::{
},
};
use common::storage::index::{IndexValue, IndexableAndSerializableObject, IndexableObject};
use nlp::language::Language;
use nlp::language::{
Language,
detect::{LanguageDetector, MIN_LANGUAGE_SCORE},
};
use store::{
search::{ContactSearchField, IndexDocument, SearchField},
write::SearchIndex,
write::{IndexPropertyClass, SearchIndex, ValueClass},
xxhash_rust::xxh3,
};
use types::{acl::AclGrant, collection::SyncCollection, field::ContactField};
@@ -97,6 +100,13 @@ impl IndexableObject for ContactCard {
field: ContactField::Email.into(),
value: self.emails().next().into(),
},
IndexValue::Property {
field: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: self.created as u64,
}),
value: self.modified.into(),
},
IndexValue::SearchIndex {
index: SearchIndex::Contacts,
hash: self.hashes().fold(0, |acc, hash| acc ^ hash),
@@ -127,6 +137,13 @@ impl IndexableObject for &ArchivedContactCard {
field: ContactField::Email.into(),
value: self.emails().next().into(),
},
IndexValue::Property {
field: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: self.created.to_native() as u64,
}),
value: (self.modified.to_native() as u64).into(),
},
IndexValue::SearchIndex {
index: SearchIndex::Contacts,
hash: self.hashes().fold(0, |acc, hash| acc ^ hash),
@@ -244,29 +261,24 @@ impl ArchivedContactCard {
impl ArchivedContactCard {
pub fn index_document(&self, index_fields: &AHashSet<SearchField>) -> IndexDocument {
let mut document = IndexDocument::with_default_language(Language::Unknown);
if index_fields.is_empty()
|| index_fields.contains(&SearchField::Contact(ContactSearchField::Created))
{
document.index_integer(ContactSearchField::Created, self.created.to_native());
}
let mut document = IndexDocument::new(SearchIndex::Contacts);
let mut detector = LanguageDetector::new();
for entry in self.card.entries.iter() {
let field = SearchField::Contact(match entry.name {
ArchivedVCardProperty::N => ContactSearchField::Name,
ArchivedVCardProperty::Nickname => ContactSearchField::Nickname,
ArchivedVCardProperty::Org => ContactSearchField::Organization,
ArchivedVCardProperty::Email => ContactSearchField::Email,
ArchivedVCardProperty::Tel => ContactSearchField::Phone,
let (is_text, field) = SearchField::Contact(match entry.name {
ArchivedVCardProperty::N => (false, ContactSearchField::Name),
ArchivedVCardProperty::Nickname => (false, ContactSearchField::Nickname),
ArchivedVCardProperty::Org => (false, ContactSearchField::Organization),
ArchivedVCardProperty::Email => (false, ContactSearchField::Email),
ArchivedVCardProperty::Tel => (false, ContactSearchField::Phone),
ArchivedVCardProperty::Impp | ArchivedVCardProperty::Socialprofile => {
ContactSearchField::OnlineService
(false, ContactSearchField::OnlineService)
}
ArchivedVCardProperty::Adr => ContactSearchField::Address,
ArchivedVCardProperty::Note => ContactSearchField::Note,
ArchivedVCardProperty::Kind => ContactSearchField::Kind,
ArchivedVCardProperty::Uid => ContactSearchField::Uid,
ArchivedVCardProperty::Member => ContactSearchField::Member,
ArchivedVCardProperty::Adr => (false, ContactSearchField::Address),
ArchivedVCardProperty::Note => (true, ContactSearchField::Note),
ArchivedVCardProperty::Kind => (false, ContactSearchField::Kind),
ArchivedVCardProperty::Uid => (false, ContactSearchField::Uid),
ArchivedVCardProperty::Member => (false, ContactSearchField::Member),
_ => continue,
});
@@ -274,14 +286,21 @@ impl ArchivedContactCard {
for value in entry.values.iter() {
match value {
ArchivedVCardValue::Text(v) => {
document.index_text(field.clone(), v, Language::Unknown);
let lang = if is_text {
detector.detect(v.as_str(), MIN_LANGUAGE_SCORE);
Language::Unknown
} else {
Language::None
};
document.index_text(field.clone(), v, lang);
}
ArchivedVCardValue::Kind(v) => {
document.index_text(field.clone(), v.as_str(), Language::Unknown);
document.index_text(field.clone(), v.as_str(), Language::None);
}
ArchivedVCardValue::Component(v) => {
for item in v.iter() {
document.index_text(field.clone(), item, Language::Unknown);
document.index_text(field.clone(), item, Language::None);
}
}
_ => (),
@@ -290,12 +309,22 @@ impl ArchivedContactCard {
for param in entry.params.iter() {
if let ArchivedVCardParameterValue::Text(value) = &param.value {
document.index_text(field.clone(), value, Language::Unknown);
let lang = if is_text {
detector.detect(v.as_str(), MIN_LANGUAGE_SCORE);
Language::Unknown
} else {
Language::None
};
document.index_text(field.clone(), value, lang);
}
}
}
}
if let Some(detected_language) = detector.most_frequent_language() {
document.set_unknown_language(detected_language);
}
document
}
}

21
crates/http/src/management/enterprise/telemetry.rs
View File

@@ -91,16 +91,20 @@ impl TelemetryApi for Server {
if in_quote {
buf.push(' ');
} else if !buf.is_empty() {
tracing_query
.push(SearchFilter::eq(TracingSearchField::Address, buf));
tracing_query.push(SearchFilter::has_unknown_text(
TracingSearchField::Keywords,
buf,
));
buf = String::new();
}
} else if ch == '"' {
buf.push(ch);
if in_quote {
if !buf.is_empty() {
tracing_query
.push(SearchFilter::eq(TracingSearchField::Address, buf));
tracing_query.push(SearchFilter::has_unknown_text(
TracingSearchField::Keywords,
buf,
));
buf = String::new();
}
in_quote = false;
@@ -112,7 +116,10 @@ impl TelemetryApi for Server {
}
}
if !buf.is_empty() {
tracing_query.push(SearchFilter::eq(TracingSearchField::Address, buf));
tracing_query.push(SearchFilter::has_unknown_text(
TracingSearchField::Keywords,
buf,
));
}
}
let before = params
@@ -131,6 +138,8 @@ impl TelemetryApi for Server {
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
@@ -141,7 +150,7 @@ impl TelemetryApi for Server {
let span_ids = self
.search_store()
.query(SearchQuery::new(SearchIndex::TracingSpan).with_filters(tracing_query))
.query(SearchQuery::new(SearchIndex::Tracing).with_filters(tracing_query))
.await?;
let (total, span_ids) = if limit > 0 {

36
crates/imap/src/op/search.rs
View File

@@ -22,16 +22,19 @@ use imap_proto::{
};
use mail_parser::HeaderName;
use nlp::language::Language;
use std::{borrow::Cow, str::FromStr, sync::Arc, time::Instant};
use std::{str::FromStr, sync::Arc, time::Instant};
use store::{
query::log::Query,
roaring::RoaringBitmap,
search::{EmailSearchField, SearchComparator, SearchFilter, SearchQuery},
search::{
EmailSearchField, SearchComparator, SearchFilter, SearchOperator, SearchQuery, SearchValue,
},
write::{SearchIndex, now},
};
use tokio::sync::watch;
use trc::AddContext;
use types::{collection::SyncCollection, id::Id, keyword::Keyword};
use utils::map::vec_map::VecMap;
impl<T: SessionStream> Session<T> {
pub async fn handle_search(
@@ -463,29 +466,26 @@ impl<T: SessionStream> SessionData<T> {
}
Filter::Header(header, value) => {
if let Some(header) = HeaderName::parse(header) {
let is_id = matches!(
let op = if matches!(
header,
HeaderName::MessageId
| HeaderName::InReplyTo
| HeaderName::References
| HeaderName::ResentMessageId
);
let header = match header {
HeaderName::Other(value) => {
EmailSearchField::Header(Cow::Owned(value.to_ascii_lowercase()))
}
_ => EmailSearchField::Header(Cow::Borrowed(header.as_static_str())),
) || value.is_empty()
{
SearchOperator::Equal
} else {
SearchOperator::Contains
};
if !value.is_empty() {
if is_id {
filters.push(SearchFilter::eq(header, value));
} else {
filters.push(SearchFilter::has_text(header, value, Language::None));
}
} else {
filters.push(SearchFilter::exists(header));
}
filters.push(SearchFilter::cond(
EmailSearchField::Headers,
op,
SearchValue::KeyValues(
VecMap::with_capacity(1).with_append(header.into_string(), value),
),
));
}
}
Filter::Subject(text) => {

3
crates/jmap/src/calendar_event_notification/query.rs
View File

@@ -170,7 +170,8 @@ impl CalendarEventNotificationQuery for Server {
let results = SearchQuery::new(SearchIndex::InMemory)
.with_filters(filters)
.with_mask(document_ids)
.execute();
.filter()
.into_bitmap();
let mut response = QueryResponseBuilder::new(
results.len() as usize,

145
crates/jmap/src/contact/query.rs
View File

@@ -13,11 +13,17 @@ use jmap_proto::{
request::MaybeInvalid,
};
use store::{
IterateParams, U32_LEN, U64_LEN, ValueKey,
roaring::RoaringBitmap,
search::{ContactSearchField, SearchComparator, SearchFilter, SearchQuery},
write::SearchIndex,
write::{IndexPropertyClass, SearchIndex, ValueClass, key::DeserializeBigEndian},
};
use trc::AddContext;
use types::{
acl::Acl,
collection::{Collection, SyncCollection},
field::ContactField,
};
use types::{acl::Acl, collection::SyncCollection};
use utils::sanitize_email;
pub trait ContactCardQuery: Sync + Send {
@@ -28,6 +34,13 @@ pub trait ContactCardQuery: Sync + Send {
) -> impl Future<Output = trc::Result<QueryResponse>> + Send;
}
#[derive(Clone)]
struct CreatedUpdated {
document_id: u32,
created: u64,
updated: u64,
}
impl ContactCardQuery for Server {
async fn contact_card_query(
&self,
@@ -39,6 +52,62 @@ impl ContactCardQuery for Server {
let cache = self
.fetch_dav_resources(access_token, account_id, SyncCollection::AddressBook)
.await?;
let mut created_to_updated = Vec::new();
if request.filter.iter().any(|cond| {
matches!(
cond,
Filter::Property(
ContactCardFilter::CreatedBefore(_)
| ContactCardFilter::CreatedAfter(_)
| ContactCardFilter::UpdatedBefore(_)
| ContactCardFilter::UpdatedAfter(_)
)
)
}) || request.sort.as_ref().is_some_and(|v| {
v.iter().any(|sort| {
matches!(
sort.property,
ContactCardComparator::Created | ContactCardComparator::Updated
)
})
}) {
self.store()
.iterate(
IterateParams::new(
ValueKey {
account_id,
collection: Collection::ContactCard.into(),
document_id: 0,
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: 0,
}),
},
ValueKey {
account_id,
collection: Collection::ContactCard.into(),
document_id: 0,
class: ValueClass::IndexProperty(IndexPropertyClass::Integer {
property: ContactField::CreatedToUpdated.into(),
value: u64::MAX,
}),
},
)
.ascending(),
|key, value| {
created_to_updated.push(CreatedUpdated {
document_id: key.deserialize_be_u32(key.len() - U32_LEN)?,
created: key.deserialize_be_u64(key.len() - U32_LEN - U64_LEN)?,
updated: value.deserialize_be_u64(0)?,
});
Ok(true)
},
)
.await
.caused_by(trc::location!())?;
}
for cond in std::mem::take(&mut request.filter) {
match cond {
@@ -149,22 +218,38 @@ impl ContactCardQuery for Server {
));
filters.push(SearchFilter::End);
}
ContactCardFilter::CreatedBefore(before) => filters.push(SearchFilter::lt(
ContactSearchField::Created,
before.timestamp(),
)),
ContactCardFilter::CreatedAfter(after) => filters.push(SearchFilter::gt(
ContactSearchField::Created,
after.timestamp(),
)),
/*ContactCardFilter::UpdatedBefore(before) => filters.push(SearchFilter::lt(
ContactSearchField::Updated,
before.timestamp(),
)),
ContactCardFilter::UpdatedAfter(after) => filters.push(SearchFilter::gt(
ContactSearchField::Updated,
after.timestamp(),
)),*/
ContactCardFilter::CreatedBefore(before) => {
let before = before.timestamp() as u64;
filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter(
created_to_updated
.iter()
.filter_map(|cu| (cu.created < before).then_some(cu.document_id)),
)));
}
ContactCardFilter::CreatedAfter(after) => {
let after = after.timestamp() as u64;
filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter(
created_to_updated
.iter()
.filter_map(|cu| (cu.created > after).then_some(cu.document_id)),
)));
}
ContactCardFilter::UpdatedBefore(before) => {
let before = before.timestamp() as u64;
filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter(
created_to_updated
.iter()
.filter_map(|cu| (cu.updated < before).then_some(cu.document_id)),
)));
}
ContactCardFilter::UpdatedAfter(after) => {
let after = after.timestamp() as u64;
filters.push(SearchFilter::is_in_set(RoaringBitmap::from_iter(
created_to_updated
.iter()
.filter_map(|cu| (cu.updated > after).then_some(cu.document_id)),
)));
}
unsupported => {
return Err(trc::JmapEvent::UnsupportedFilter
.into_err()
@@ -192,14 +277,26 @@ impl ContactCardQuery for Server {
.unwrap_or_default()
.into_iter()
.map(|comparator| match comparator.property {
ContactCardComparator::Created => Ok(SearchComparator::field(
ContactSearchField::Created,
ContactCardComparator::Created => Ok(SearchComparator::sorted_set(
created_to_updated
.iter()
.enumerate()
.map(|(idx, u)| (u.document_id, idx as u32))
.collect(),
comparator.is_ascending,
)),
/*ContactCardComparator::Updated => Ok(SearchComparator::field(
ContactSearchField::Updated,
comparator.is_ascending,
)),*/
ContactCardComparator::Updated => {
let mut updated = created_to_updated.clone();
updated.sort_by(|a, b| a.updated.cmp(&b.updated));
Ok(SearchComparator::sorted_set(
updated
.iter()
.enumerate()
.map(|(idx, u)| (u.document_id, idx as u32))
.collect(),
comparator.is_ascending,
))
}
other => Err(trc::JmapEvent::UnsupportedSort
.into_err()
.details(other.into_string())),

44
crates/jmap/src/email/query.rs
View File

@@ -13,15 +13,18 @@ use jmap_proto::{
};
use mail_parser::HeaderName;
use nlp::language::Language;
use std::{borrow::Cow, future::Future};
use std::future::Future;
use store::{
ahash::{AHashMap, AHashSet},
roaring::RoaringBitmap,
search::{EmailSearchField, SearchComparator, SearchFilter, SearchQuery},
search::{
EmailSearchField, SearchComparator, SearchFilter, SearchOperator, SearchQuery, SearchValue,
},
write::SearchIndex,
};
use trc::AddContext;
use types::{acl::Acl, keyword::Keyword};
use utils::map::vec_map::VecMap;
pub trait EmailQuery: Sync + Send {
fn email_query(
@@ -125,35 +128,28 @@ impl EmailQuery for Server {
})?;
if let Some(header_name) = HeaderName::parse(header_name) {
let is_id = matches!(
let value = header.next();
let op = if matches!(
header_name,
HeaderName::MessageId
| HeaderName::InReplyTo
| HeaderName::References
| HeaderName::ResentMessageId
);
let header_name = match header_name {
HeaderName::Other(value) => {
EmailSearchField::Header(Cow::Owned(value.to_ascii_lowercase()))
}
_ => EmailSearchField::Header(Cow::Borrowed(
header_name.as_static_str(),
)),
) || value.is_none()
{
SearchOperator::Equal
} else {
SearchOperator::Contains
};
if let Some(value) = header.next() {
if is_id {
filters.push(SearchFilter::eq(header_name, value));
} else {
filters.push(SearchFilter::has_text(
header_name,
value,
Language::None,
));
}
} else {
filters.push(SearchFilter::exists(header_name));
}
filters.push(SearchFilter::cond(
EmailSearchField::Headers,
op,
SearchValue::KeyValues(VecMap::with_capacity(1).with_append(
header_name.into_string(),
value.unwrap_or_default(),
)),
));
}
}
EmailFilter::InMailbox(mailbox) => {

3
crates/jmap/src/file/query.rs
View File

@@ -149,7 +149,8 @@ impl FileNodeQuery for Server {
} else {
cache.document_ids(false).collect()
})
.execute();
.filter()
.into_bitmap();
let mut response = QueryResponseBuilder::new(
results.len() as usize,

111
crates/jmap/src/mailbox/query.rs
View File

@@ -16,6 +16,7 @@ use std::{
future::Future,
};
use store::{
ahash::AHashMap,
roaring::RoaringBitmap,
search::{SearchComparator, SearchFilter, SearchQuery},
write::SearchIndex,
@@ -155,14 +156,18 @@ impl MailboxQuery for Server {
// Sort as tree
if sort_as_tree {
let sorted_list = mailboxes
let sorted_set = mailboxes
.mailboxes
.items
.iter()
.map(|mailbox| (mailbox.path.as_str(), mailbox.document_id))
.collect::<BTreeMap<_, _>>();
comparators.push(SearchComparator::sorted_list(
sorted_list.into_values().collect(),
comparators.push(SearchComparator::sorted_set(
sorted_set
.into_iter()
.enumerate()
.map(|(i, (_, v))| (i as u32, v))
.collect(),
true,
));
}
@@ -176,48 +181,46 @@ impl MailboxQuery for Server {
{
comparators.push(match comparator.property {
MailboxComparator::Name => {
let sorted_list = mailboxes
let sorted_set = mailboxes
.mailboxes
.items
.iter()
.map(|mailbox| (mailbox.name.as_str(), mailbox.document_id))
.collect::<BTreeSet<_>>();
SearchComparator::sorted_list(
sorted_list.into_iter().map(|v| v.1).collect(),
SearchComparator::sorted_set(
sorted_set
.into_iter()
.enumerate()
.map(|(i, (_, v))| (i as u32, v))
.collect(),
comparator.is_ascending,
)
}
MailboxComparator::SortOrder => {
let sorted_list = mailboxes
let sorted_set = mailboxes
.mailboxes
.items
.iter()
.map(|mailbox| (mailbox.sort_order, mailbox.document_id))
.collect::<BTreeSet<_>>();
.map(|mailbox| (mailbox.document_id, mailbox.sort_order))
.collect::<AHashMap<_, _>>();
SearchComparator::sorted_list(
sorted_list.into_iter().map(|v| v.1).collect(),
comparator.is_ascending,
)
SearchComparator::sorted_set(sorted_set, comparator.is_ascending)
}
MailboxComparator::ParentId => {
let sorted_list = mailboxes
let sorted_set = mailboxes
.mailboxes
.items
.iter()
.map(|mailbox| {
(
mailbox.parent_id().map(|id| id + 1).unwrap_or_default(),
mailbox.document_id,
mailbox.parent_id().map(|id| id + 1).unwrap_or_default(),
)
})
.collect::<BTreeSet<_>>();
.collect::<AHashMap<_, _>>();
SearchComparator::sorted_list(
sorted_list.into_iter().map(|v| v.1).collect(),
comparator.is_ascending,
)
SearchComparator::sorted_set(sorted_set, comparator.is_ascending)
}
MailboxComparator::_T(other) => {
@@ -226,7 +229,7 @@ impl MailboxQuery for Server {
});
}
let results = SearchQuery::new(SearchIndex::InMemory)
let mut results = SearchQuery::new(SearchIndex::InMemory)
.with_filters(filters)
.with_comparators(comparators)
.with_mask(if access_token.is_shared(account_id) {
@@ -239,50 +242,42 @@ impl MailboxQuery for Server {
.map(|m| m.document_id)
.collect()
})
.execute();
.filter();
// Filter as tree
if filter_as_tree {
let mut new_results = RoaringBitmap::new();
for document_id in results.results() {
let mut check_id = document_id;
for _ in 0..self.core.jmap.mailbox_max_depth {
if let Some(mailbox) = mailboxes.mailbox_by_id(&check_id) {
if let Some(parent_id) = mailbox.parent_id() {
if results.results().contains(parent_id) {
check_id = parent_id;
} else {
break;
}
} else {
new_results.insert(document_id);
}
}
}
}
results.update_results(new_results);
}
let mut response = QueryResponseBuilder::new(
results.len() as usize,
results.results().len() as usize,
self.core.jmap.query_max_results,
mailboxes.get_state(true),
&request,
);
if !results.is_empty() {
// Filter as tree
if filter_as_tree {
let mut total_filtered = 0;
let mut is_page_full = false;
for document_id in &results {
let mut check_id = document_id;
for _ in 0..self.core.jmap.mailbox_max_depth {
if let Some(mailbox) = mailboxes.mailbox_by_id(&check_id) {
if let Some(parent_id) = mailbox.parent_id() {
if results.contains(parent_id) {
check_id = parent_id;
} else {
break;
}
} else {
total_filtered += 1;
if !is_page_full && !response.add(0, document_id) {
is_page_full = true;
}
}
}
}
}
if total_filtered != results.len() {
response.response.total = Some(total_filtered as usize);
}
} else {
for document_id in results {
if !response.add(0, document_id) {
break;
}
}
for document_id in results.into_sorted() {
if !response.add(0, document_id) {
break;
}
}

3
crates/jmap/src/principal/query.rs
View File

@@ -176,7 +176,8 @@ impl PrincipalQuery for Server {
let results = SearchQuery::new(SearchIndex::InMemory)
.with_filters(filters)
.with_mask(principal_ids)
.execute();
.filter()
.into_bitmap();
let mut response = QueryResponseBuilder::new(
results.len() as usize,

3
crates/jmap/src/sieve/query.rs
View File

@@ -158,7 +158,8 @@ impl SieveScriptQuery for Server {
let mut results = SearchQuery::new(SearchIndex::InMemory)
.with_filters(filters)
.with_mask(document_ids)
.execute();
.filter()
.into_bitmap();
let mut response = QueryResponseBuilder::new(
results.len() as usize,

3
crates/jmap/src/submission/query.rs
View File

@@ -196,7 +196,8 @@ impl EmailSubmissionQuery for Server {
let results = SearchQuery::new(SearchIndex::InMemory)
.with_filters(filters)
.with_mask(document_ids)
.execute();
.filter()
.into_bitmap();
let mut response = QueryResponseBuilder::new(
results.len() as usize,

12
crates/nlp/src/language/mod.rs
View File

@@ -11,6 +11,8 @@ pub mod stopwords;
use std::borrow::Cow;
use utils::config::utils::ParseValue;
use crate::tokenizers::{
Token, chinese::ChineseTokenizer, japanese::JapaneseTokenizer, word::WordTokenizer,
};
@@ -118,6 +120,10 @@ pub enum Language {
}
impl Language {
pub fn is_unknown(&self) -> bool {
matches!(self, Language::Unknown)
}
pub fn from_iso_639(code: &str) -> Option<Self> {
hashify::map!(
code.split_once('-').map(|c| c.0).unwrap_or(code).as_bytes(),
@@ -193,3 +199,9 @@ impl Language {
}
}
}
impl ParseValue for Language {
fn parse_value(value: &str) -> utils::config::Result<Self> {
Language::from_iso_639(value).ok_or_else(|| format!("Invalid language code: {}", value))
}
}

16
crates/services/src/task_manager/index.rs
View File

@@ -96,7 +96,7 @@ impl SearchIndexTask for Server {
// File indexing not implemented yet
continue;
}
SearchIndex::TracingSpan => (
SearchIndex::Tracing => (
4,
build_tracing_span_document(self, task.account_id, task.document_id).await,
),
@@ -164,7 +164,7 @@ impl SearchIndexTask for Server {
SearchIndex::Calendar => 1,
SearchIndex::Contacts => 2,
SearchIndex::File => 3,
SearchIndex::TracingSpan | SearchIndex::InMemory => unreachable!(),
SearchIndex::Tracing | SearchIndex::InMemory => unreachable!(),
};
document_deletions[idx]
@@ -205,7 +205,7 @@ impl SearchIndexTask for Server {
SearchIndex::Calendar,
SearchIndex::Contacts,
SearchIndex::File,
SearchIndex::TracingSpan,
SearchIndex::Tracing,
]) {
if !documents.is_empty()
&& let Err(err) = self.search_store().index(index, documents).await
@@ -431,7 +431,7 @@ impl ReindexIndexTask for Server {
}
}
}
SearchIndex::File | SearchIndex::TracingSpan | SearchIndex::InMemory => (),
SearchIndex::File | SearchIndex::Tracing | SearchIndex::InMemory => (),
}
// Request indexing
@@ -475,11 +475,7 @@ async fn build_email_document(
.details("Blob not found")
})?;
Ok(Some(metadata.index_document(
&raw_message,
index_fields,
server.core.jmap.index_all_headers,
)))
Ok(Some(metadata.index_document(&raw_message, index_fields)))
}
None => Ok(None),
}
@@ -536,7 +532,7 @@ async fn build_tracing_span_document(
account_id: u32,
document_id: u32,
) -> trc::Result<Option<IndexDocument>> {
let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::TracingSpan) else {
let Some(index_fields) = server.core.jmap.index_fields.get(&SearchIndex::Tracing) else {
return Ok(None);
};

26
crates/store/src/backend/composite/read_replica.rs
View File

@@ -30,7 +30,8 @@ impl SQLReadReplica {
config: &mut Config,
prefix: impl AsKey,
stores: &Stores,
create_tables: bool,
create_store_tables: bool,
create_search_tables: bool,
) -> Option<Self> {
let prefix = prefix.as_key();
let primary_id = config.value_require((&prefix, "primary"))?.to_string();
@@ -77,12 +78,12 @@ impl SQLReadReplica {
}
}
if !replicas.is_empty() {
if create_tables {
if create_store_tables {
let result = match &primary {
#[cfg(feature = "postgres")]
Store::PostgreSQL(store) => store.create_tables().await,
Store::PostgreSQL(store) => store.create_storage_tables().await,
#[cfg(feature = "mysql")]
Store::MySQL(store) => store.create_tables().await,
Store::MySQL(store) => store.create_storage_tables().await,
_ => panic!("Invalid store type"),
};
@@ -94,6 +95,23 @@ impl SQLReadReplica {
}
}
if create_search_tables {
let result = match &primary {
#[cfg(feature = "postgres")]
Store::PostgreSQL(store) => store.create_search_tables().await,
#[cfg(feature = "mysql")]
Store::MySQL(store) => store.create_search_tables().await,
_ => panic!("Invalid store type"),
};
if let Err(err) = result {
config.new_build_warning(
(&prefix, "primary"),
format!("Failed to create search tables: {err}"),
);
}
}
Some(Self {
primary,
replicas,

2
crates/store/src/backend/mysql/lookup.rs
View File

@@ -11,7 +11,7 @@ use crate::{IntoRows, QueryResult, QueryType, Value};
use super::{MysqlStore, into_error};
impl MysqlStore {
pub(crate) async fn query<T: QueryResult>(
pub(crate) async fn sql_query<T: QueryResult>(
&self,
query: &str,
params: &[Value<'_>],

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

@@ -6,10 +6,19 @@
use std::time::Duration;
use mysql_async::{OptsBuilder, Pool, PoolConstraints, PoolOpts, SslOpts, prelude::Queryable};
use mysql_async::{
Conn, OptsBuilder, Pool, PoolConstraints, PoolOpts, SslOpts, prelude::Queryable,
};
use utils::config::{Config, utils::AsKey};
use crate::*;
use crate::{
backend::mysql::MysqlSearchField,
search::{
CalendarSearchField, ContactSearchField, EmailSearchField, SearchableField,
TracingSearchField,
},
*,
};
use super::{MysqlStore, into_error};
@@ -17,7 +26,8 @@ impl MysqlStore {
pub async fn open(
config: &mut Config,
prefix: impl AsKey,
create_tables: bool,
create_store_tables: bool,
create_search_tables: bool,
) -> Option<Self> {
let prefix = prefix.as_key();
let mut opts = OptsBuilder::default()
@@ -78,14 +88,21 @@ impl MysqlStore {
conn_pool: Pool::new(opts),
};
if create_tables && let Err(err) = db.create_tables().await {
if create_store_tables && let Err(err) = db.create_storage_tables().await {
config.new_build_error(prefix.as_str(), format!("Failed to create tables: {err}"));
}
if create_search_tables && let Err(err) = db.create_search_tables().await {
config.new_build_warning(
prefix.as_str(),
format!("Failed to create search tables: {err}"),
);
}
Some(db)
}
pub(crate) async fn create_tables(&self) -> trc::Result<()> {
pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> {
let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
for table in [
@@ -155,4 +172,78 @@ impl MysqlStore {
Ok(())
}
pub(crate) async fn create_search_tables(&self) -> trc::Result<()> {
let mut conn = self.conn_pool.get_conn().await.map_err(into_error)?;
create_search_tables::<EmailSearchField>(&mut conn).await?;
create_search_tables::<CalendarSearchField>(&mut conn).await?;
create_search_tables::<ContactSearchField>(&mut conn).await?;
//create_search_tables::<FileSearchField>(&mut conn).await?;
create_search_tables::<TracingSearchField>(&mut conn).await?;
Ok(())
}
}
async fn create_search_tables<T: SearchableField + MysqlSearchField + 'static>(
conn: &mut Conn,
) -> trc::Result<()> {
let table_name = T::index().mysql_table();
let mut query = format!("CREATE TABLE IF NOT EXISTS {} (", table_name);
// Add primary key columns
let pkeys = T::primary_keys();
for pkey in pkeys {
query.push_str(&format!("{} {}, ", pkey.column(), pkey.column_type()));
}
// Add other columns
for field in T::all_fields() {
query.push_str(&format!("{} {}, ", field.column(), field.column_type()));
}
// Add primary key constraint
query.push_str("PRIMARY KEY ");
if pkeys.len() > 1 {
query.push('(');
}
for (i, pkey) in pkeys.iter().enumerate() {
if i > 0 {
query.push_str(", ");
}
query.push_str(pkey.column());
}
if pkeys.len() > 1 {
query.push(')');
}
query.push_str(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci");
conn.query_drop(&query).await.map_err(into_error)?;
// Create indexes
for field in T::all_fields() {
if field.is_text() {
let column_name = field.column();
let create_index_query = format!(
"CREATE FULLTEXT INDEX IF NOT EXISTS fts_{table_name}_{column_name} ON {table_name}({column_name})",
);
conn.query_drop(&create_index_query)
.await
.map_err(into_error)?;
}
if field.is_indexed() {
let column_name = field.column();
let create_index_query = format!(
"CREATE INDEX IF NOT EXISTS idx_{table_name}_{column_name} ON {table_name}({column_name})",
);
conn.query_drop(&create_index_query)
.await
.map_err(into_error)?;
}
}
Ok(())
}

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

@@ -4,14 +4,21 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::fmt::Display;
use crate::{
search::{
CalendarSearchField, ContactSearchField, EmailSearchField, FileSearchField, SearchField,
TracingSearchField,
},
write::SearchIndex,
};
use mysql_async::Pool;
use std::fmt::Display;
pub mod blob;
pub mod lookup;
pub mod main;
pub mod read;
pub mod search;
pub mod write;
pub struct MysqlStore {
@@ -22,3 +29,154 @@ pub struct MysqlStore {
fn into_error(err: impl Display) -> trc::Error {
trc::StoreEvent::MysqlError.reason(err)
}
impl SearchIndex {
pub(super) fn mysql_table(&self) -> &'static str {
match self {
SearchIndex::Email => "s_email",
SearchIndex::Calendar => "s_cal",
SearchIndex::Contacts => "s_card",
SearchIndex::File => "s_file",
SearchIndex::Tracing => "s_trace",
SearchIndex::InMemory => "",
}
}
}
pub(super) trait MysqlSearchField {
fn column(&self) -> &'static str;
fn column_type(&self) -> &'static str;
}
impl MysqlSearchField for EmailSearchField {
fn column(&self) -> &'static str {
match self {
EmailSearchField::From => "fadr",
EmailSearchField::To => "tadr",
EmailSearchField::Cc => "cc",
EmailSearchField::Bcc => "bcc",
EmailSearchField::Subject => "subj",
EmailSearchField::Body => "body",
EmailSearchField::Attachment => "atta",
EmailSearchField::ReceivedAt => "rcvd",
EmailSearchField::SentAt => "sent",
EmailSearchField::Size => "size",
EmailSearchField::HasAttachment => "hatt",
EmailSearchField::Headers => "hdrs",
}
}
fn column_type(&self) -> &'static str {
match self {
EmailSearchField::ReceivedAt | EmailSearchField::SentAt => "BIGINT NOT NULL",
EmailSearchField::Size => "INT NOT NULL",
EmailSearchField::HasAttachment => "BOOLEAN NOT NULL",
EmailSearchField::Headers => "JSON",
_ => "TEXT",
}
}
}
impl MysqlSearchField for CalendarSearchField {
fn column(&self) -> &'static str {
match self {
CalendarSearchField::Title => "titl",
CalendarSearchField::Description => "dscd",
CalendarSearchField::Location => "locn",
CalendarSearchField::Owner => "ownr",
CalendarSearchField::Attendee => "atnd",
CalendarSearchField::Start => "strt",
CalendarSearchField::Uid => "uid",
}
}
fn column_type(&self) -> &'static str {
match self {
CalendarSearchField::Start => "BIGINT NOT NULL",
_ => "TEXT",
}
}
}
impl MysqlSearchField for ContactSearchField {
fn column(&self) -> &'static str {
match self {
ContactSearchField::Member => "mmbr",
ContactSearchField::Name => "name",
ContactSearchField::Nickname => "nick",
ContactSearchField::Organization => "orgn",
ContactSearchField::Email => "eml",
ContactSearchField::Phone => "phon",
ContactSearchField::OnlineService => "olsv",
ContactSearchField::Address => "addr",
ContactSearchField::Note => "note",
ContactSearchField::Kind => "kind",
ContactSearchField::Uid => "uid",
}
}
fn column_type(&self) -> &'static str {
match self {
ContactSearchField::Kind | ContactSearchField::Uid => "TEXT",
_ => "TEXT",
}
}
}
impl MysqlSearchField for FileSearchField {
fn column(&self) -> &'static str {
match self {
FileSearchField::Name => "name",
FileSearchField::Content => "body",
}
}
fn column_type(&self) -> &'static str {
"TEXT"
}
}
impl MysqlSearchField for TracingSearchField {
fn column(&self) -> &'static str {
match self {
TracingSearchField::QueueId => "qid",
TracingSearchField::EventType => "etyp",
TracingSearchField::Keywords => "kwds",
}
}
fn column_type(&self) -> &'static str {
match self {
TracingSearchField::EventType => "BIGINT NOT NULL",
TracingSearchField::QueueId => "BIGINT",
TracingSearchField::Keywords => "TEXT",
}
}
}
impl MysqlSearchField for SearchField {
fn column(&self) -> &'static str {
match self {
SearchField::AccountId => "accid",
SearchField::DocumentId => "docid",
SearchField::Id => "id",
SearchField::Email(field) => field.column(),
SearchField::Calendar(field) => field.column(),
SearchField::Contact(field) => field.column(),
SearchField::File(field) => field.column(),
SearchField::Tracing(field) => field.column(),
}
}
fn column_type(&self) -> &'static str {
match self {
SearchField::AccountId => "INT NOT NULL",
SearchField::DocumentId => "INT NOT NULL",
SearchField::Id => "BIGINT NOT NULL",
SearchField::Email(field) => field.column_type(),
SearchField::Calendar(field) => field.column_type(),
SearchField::Contact(field) => field.column_type(),
SearchField::File(field) => field.column_type(),
SearchField::Tracing(field) => field.column_type(),
}
}
}

29
crates/store/src/backend/mysql/search.rs Normal file
View File

@@ -0,0 +1,29 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
backend::mysql::MysqlStore,
search::{IndexDocument, SearchDocumentId, SearchQuery},
write::SearchIndex,
};
impl MysqlStore {
pub async fn query<R: SearchDocumentId>(&self, query: SearchQuery) -> trc::Result<Vec<R>> {
todo!()
}
pub async fn index(
&self,
index: SearchIndex,
documents: Vec<IndexDocument>,
) -> trc::Result<()> {
todo!()
}
pub async fn unindex(&self, query: SearchQuery) -> trc::Result<()> {
todo!()
}
}

2
crates/store/src/backend/postgres/lookup.rs
View File

@@ -15,7 +15,7 @@ use crate::IntoRows;
use super::{PostgresStore, into_error};
impl PostgresStore {
pub(crate) async fn query<T: QueryResult>(
pub(crate) async fn sql_query<T: QueryResult>(
&self,
query: &str,
params_: &[crate::Value<'_>],

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

@@ -4,13 +4,19 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::time::Duration;
use crate::{backend::postgres::tls::MakeRustlsConnect, *};
use super::{PostgresStore, into_error};
use deadpool_postgres::{Config, ManagerConfig, PoolConfig, RecyclingMethod, Runtime};
use crate::{
backend::postgres::{PsqlSearchField, tls::MakeRustlsConnect},
search::{
CalendarSearchField, ContactSearchField, EmailSearchField, SearchableField,
TracingSearchField,
},
*,
};
use deadpool::managed::Object;
use deadpool_postgres::{Config, Manager, ManagerConfig, PoolConfig, RecyclingMethod, Runtime};
use nlp::language::Language;
use std::time::Duration;
use tokio_postgres::NoTls;
use utils::{config::utils::AsKey, rustls_client_config};
@@ -18,7 +24,8 @@ impl PostgresStore {
pub async fn open(
config: &mut utils::config::Config,
prefix: impl AsKey,
create_tables: bool,
create_store_tables: bool,
create_search_tables: bool,
) -> Option<Self> {
let prefix = prefix.as_key();
let mut cfg = Config::new();
@@ -40,7 +47,7 @@ impl PostgresStore {
if let Some(max_conn) = config.property::<usize>((&prefix, "pool.max-connections")) {
cfg.pool = PoolConfig::new(max_conn).into();
}
let db = Self {
let mut db = Self {
conn_pool: if config
.property_or_default::<bool>((&prefix, "tls.enable"), "false")
.unwrap_or_default()
@@ -63,16 +70,32 @@ impl PostgresStore {
)
})
.ok()?,
languages: config
.properties::<Language>((&prefix, "languages"))
.into_iter()
.map(|(_, v)| v)
.collect(),
};
if create_tables && let Err(err) = db.create_tables().await {
if db.languages.is_empty() {
db.languages.insert(Language::English);
}
if create_store_tables && let Err(err) = db.create_storage_tables().await {
config.new_build_error(prefix.as_str(), format!("Failed to create tables: {err}"));
}
if create_search_tables && let Err(err) = db.create_search_tables().await {
config.new_build_warning(
prefix.as_str(),
format!("Failed to create search tables: {err}"),
);
}
Some(db)
}
pub(crate) async fn create_tables(&self) -> trc::Result<()> {
pub(crate) async fn create_storage_tables(&self) -> trc::Result<()> {
let conn = self.conn_pool.get().await.map_err(into_error)?;
for table in [
@@ -138,4 +161,81 @@ impl PostgresStore {
Ok(())
}
pub(crate) async fn create_search_tables(&self) -> trc::Result<()> {
let conn = self.conn_pool.get().await.map_err(into_error)?;
create_search_tables::<EmailSearchField>(&conn).await?;
create_search_tables::<CalendarSearchField>(&conn).await?;
create_search_tables::<ContactSearchField>(&conn).await?;
//create_search_tables::<FileSearchField>(&conn).await?;
create_search_tables::<TracingSearchField>(&conn).await?;
Ok(())
}
}
async fn create_search_tables<T: SearchableField + PsqlSearchField + 'static>(
conn: &Object<Manager>,
) -> trc::Result<()> {
let table_name = T::index().psql_table();
let mut query = format!("CREATE TABLE IF NOT EXISTS {} (", table_name);
// Add primary key columns
let pkeys = T::primary_keys();
for pkey in pkeys {
query.push_str(&format!("{} {}, ", pkey.column(), pkey.column_type()));
}
// Add other columns
for field in T::all_fields() {
query.push_str(&format!("{} {}", field.column(), field.column_type()));
if let Some(sort_type) = field.sort_column_type() {
query.push_str(&format!(", {} {}", field.sort_column().unwrap(), sort_type));
}
query.push_str(", ");
}
// Add primary key constraint
query.push_str("PRIMARY KEY ");
if pkeys.len() > 1 {
query.push('(');
}
for (i, pkey) in pkeys.iter().enumerate() {
if i > 0 {
query.push_str(", ");
}
query.push_str(pkey.column());
}
if pkeys.len() > 1 {
query.push(')');
}
query.push(')');
conn.execute(&query, &[]).await.map_err(into_error)?;
// Create indexes
for field in T::all_fields() {
if field.is_text() {
let column_name = field.column();
let create_index_query = format!(
"CREATE INDEX IF NOT EXISTS fts_{table_name}_{column_name} ON {table_name} USING GIN({column_name})",
);
conn.execute(&create_index_query, &[])
.await
.map_err(into_error)?;
}
if field.is_indexed() {
let column_name = field.sort_column().unwrap_or(field.column());
let create_index_query = format!(
"CREATE INDEX IF NOT EXISTS idx_{table_name}_{column_name} ON {table_name}({column_name})",
);
conn.execute(&create_index_query, &[])
.await
.map_err(into_error)?;
}
}
Ok(())
}

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

@@ -4,22 +4,258 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::fmt::Display;
use crate::{
search::{
CalendarSearchField, ContactSearchField, EmailSearchField, FileSearchField, SearchField,
TracingSearchField,
},
write::SearchIndex,
};
use ahash::AHashSet;
use deadpool_postgres::Pool;
use nlp::language::Language;
use std::fmt::Display;
pub mod blob;
pub mod lookup;
pub mod main;
pub mod read;
pub mod search;
pub mod tls;
pub mod write;
pub struct PostgresStore {
pub(crate) conn_pool: Pool,
pub(crate) languages: AHashSet<Language>,
}
#[inline(always)]
fn into_error(err: impl Display) -> trc::Error {
trc::StoreEvent::PostgresqlError.reason(err)
}
impl SearchIndex {
pub(super) fn psql_table(&self) -> &'static str {
match self {
SearchIndex::Email => "s_email",
SearchIndex::Calendar => "s_cal",
SearchIndex::Contacts => "s_card",
SearchIndex::File => "s_file",
SearchIndex::Tracing => "s_trace",
SearchIndex::InMemory => "",
}
}
}
pub(super) trait PsqlSearchField {
fn column(&self) -> &'static str;
fn column_type(&self) -> &'static str;
fn sort_column_type(&self) -> Option<&'static str>;
fn sort_column(&self) -> Option<&'static str>;
}
impl PsqlSearchField for EmailSearchField {
fn column(&self) -> &'static str {
match self {
EmailSearchField::From => "fadr",
EmailSearchField::To => "tadr",
EmailSearchField::Cc => "cc",
EmailSearchField::Bcc => "bcc",
EmailSearchField::Subject => "subj",
EmailSearchField::Body => "body",
EmailSearchField::Attachment => "atta",
EmailSearchField::ReceivedAt => "rcvd",
EmailSearchField::SentAt => "sent",
EmailSearchField::Size => "size",
EmailSearchField::HasAttachment => "hatt",
EmailSearchField::Headers => "hdrs",
}
}
fn column_type(&self) -> &'static str {
match self {
EmailSearchField::ReceivedAt | EmailSearchField::SentAt => "BIGINT",
EmailSearchField::Size => "INTEGER",
EmailSearchField::HasAttachment => "BOOLEAN",
EmailSearchField::Headers => "JSONB",
_ => "TSVECTOR",
}
}
fn sort_column_type(&self) -> Option<&'static str> {
match self {
EmailSearchField::From | EmailSearchField::To | EmailSearchField::Subject => {
Some("TEXT")
}
_ => None,
}
}
fn sort_column(&self) -> Option<&'static str> {
match self {
EmailSearchField::From => Some("s_fr"),
EmailSearchField::To => Some("s_to"),
EmailSearchField::Subject => Some("s_sj"),
_ => None,
}
}
}
impl PsqlSearchField for CalendarSearchField {
fn column(&self) -> &'static str {
match self {
CalendarSearchField::Title => "titl",
CalendarSearchField::Description => "dscd",
CalendarSearchField::Location => "locn",
CalendarSearchField::Owner => "ownr",
CalendarSearchField::Attendee => "atnd",
CalendarSearchField::Start => "strt",
CalendarSearchField::Uid => "uid",
}
}
fn column_type(&self) -> &'static str {
match self {
CalendarSearchField::Start => "BIGINT",
CalendarSearchField::Uid => "TEXT",
_ => "TSVECTOR",
}
}
fn sort_column_type(&self) -> Option<&'static str> {
None
}
fn sort_column(&self) -> Option<&'static str> {
None
}
}
impl PsqlSearchField for ContactSearchField {
fn column(&self) -> &'static str {
match self {
ContactSearchField::Member => "mmbr",
ContactSearchField::Name => "name",
ContactSearchField::Nickname => "nick",
ContactSearchField::Organization => "orgn",
ContactSearchField::Email => "eml",
ContactSearchField::Phone => "phon",
ContactSearchField::OnlineService => "olsv",
ContactSearchField::Address => "addr",
ContactSearchField::Note => "note",
ContactSearchField::Kind => "kind",
ContactSearchField::Uid => "uid",
}
}
fn column_type(&self) -> &'static str {
match self {
ContactSearchField::Kind | ContactSearchField::Uid => "TEXT",
_ => "TSVECTOR",
}
}
fn sort_column_type(&self) -> Option<&'static str> {
None
}
fn sort_column(&self) -> Option<&'static str> {
None
}
}
impl PsqlSearchField for FileSearchField {
fn column(&self) -> &'static str {
match self {
FileSearchField::Name => "name",
FileSearchField::Content => "body",
}
}
fn column_type(&self) -> &'static str {
"TSVECTOR"
}
fn sort_column_type(&self) -> Option<&'static str> {
None
}
fn sort_column(&self) -> Option<&'static str> {
None
}
}
impl PsqlSearchField for TracingSearchField {
fn column(&self) -> &'static str {
match self {
TracingSearchField::QueueId => "qid",
TracingSearchField::EventType => "etyp",
TracingSearchField::Keywords => "kwds",
}
}
fn column_type(&self) -> &'static str {
match self {
TracingSearchField::EventType => "BIGINT",
TracingSearchField::QueueId => "BIGINT",
TracingSearchField::Keywords => "TSVECTOR",
}
}
fn sort_column_type(&self) -> Option<&'static str> {
None
}
fn sort_column(&self) -> Option<&'static str> {
None
}
}
impl PsqlSearchField for SearchField {
fn column(&self) -> &'static str {
match self {
SearchField::AccountId => "accid",
SearchField::DocumentId => "docid",
SearchField::Id => "id",
SearchField::Email(field) => field.column(),
SearchField::Calendar(field) => field.column(),
SearchField::Contact(field) => field.column(),
SearchField::File(field) => field.column(),
SearchField::Tracing(field) => field.column(),
}
}
fn column_type(&self) -> &'static str {
match self {
SearchField::AccountId => "INTEGER NOT NULL",
SearchField::DocumentId => "INTEGER NOT NULL",
SearchField::Id => "BIGINT NOT NULL",
SearchField::Email(field) => field.column_type(),
SearchField::Calendar(field) => field.column_type(),
SearchField::Contact(field) => field.column_type(),
SearchField::File(field) => field.column_type(),
SearchField::Tracing(field) => field.column_type(),
}
}
fn sort_column_type(&self) -> Option<&'static str> {
match self {
SearchField::Email(field) => field.sort_column_type(),
SearchField::Calendar(field) => field.sort_column_type(),
SearchField::Contact(field) => field.sort_column_type(),
SearchField::File(field) => field.sort_column_type(),
SearchField::Tracing(field) => field.sort_column_type(),
SearchField::AccountId | SearchField::DocumentId | SearchField::Id => None,
}
}
fn sort_column(&self) -> Option<&'static str> {
match self {
SearchField::Email(field) => field.sort_column(),
SearchField::Calendar(field) => field.sort_column(),
SearchField::Contact(field) => field.sort_column(),
SearchField::File(field) => field.sort_column(),
SearchField::Tracing(field) => field.sort_column(),
SearchField::AccountId | SearchField::DocumentId | SearchField::Id => None,
}
}
}

351
crates/store/src/backend/postgres/search.rs Normal file
View File

@@ -0,0 +1,351 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use crate::{
backend::postgres::{PostgresStore, PsqlSearchField, into_error},
search::{
IndexDocument, SearchComparator, SearchDocumentId, SearchFilter, SearchOperator,
SearchQuery, SearchValue,
},
write::SearchIndex,
};
use nlp::language::Language;
use tokio_postgres::{
IsolationLevel,
types::{ToSql, Type},
};
impl PostgresStore {
pub async fn index(&self, documents: Vec<IndexDocument>) -> trc::Result<()> {
let mut conn = self.conn_pool.get().await.map_err(into_error)?;
let trx = conn
.build_transaction()
.isolation_level(IsolationLevel::ReadCommitted)
.start()
.await
.map_err(into_error)?;
for document in documents {
let index = document.index;
let primary_keys = index.primary_keys();
let all_fields = index.all_fields();
let fields = document.fields;
let mut values = Vec::with_capacity(fields.len() + 2);
let mut query = format!("INSERT INTO {} (", index.psql_table());
for (i, field) in primary_keys.iter().chain(all_fields).enumerate() {
if i > 0 {
query.push(',');
}
query.push_str(field.column());
if let Some(sort_column) = field.sort_column() {
query.push(',');
query.push_str(sort_column);
}
}
query.push_str(") VALUES (");
for (i, field) in primary_keys.iter().chain(all_fields).enumerate() {
if i > 0 {
query.push(',');
}
if let Some(value) = fields.get(field) {
let value_ref = format!("${}", values.len() + 1);
if field.is_text() {
let language = match &value {
SearchValue::Text { language, .. }
if self.languages.contains(language) =>
{
pg_lang(language).unwrap_or("simple")
}
_ => "simple",
};
query.push_str(&format!("to_tsvector('{language}',{value_ref})"));
} else {
query.push_str(&value_ref);
}
if field.sort_column().is_some() {
query.push(',');
query.push_str(&value_ref);
}
values.push(value as &(dyn ToSql + Sync));
} else {
query.push_str("NULL");
if field.sort_column().is_some() {
query.push_str(",NULL");
}
}
}
query.push_str(") ON CONFLICT (");
for (i, pkey) in primary_keys.iter().enumerate() {
if i > 0 {
query.push(',');
}
query.push_str(pkey.column());
}
query.push_str(") DO UPDATE SET ");
for (i, field) in all_fields.iter().enumerate() {
if i > 0 {
query.push(',');
}
let column = field.column();
query.push_str(&format!("{column} = EXCLUDED.{column}"));
}
trx.execute(&query, &values).await.map_err(into_error)?;
}
trx.commit().await.map_err(into_error)
}
pub async fn query<R: SearchDocumentId>(
&self,
index: SearchIndex,
filters: &[SearchFilter],
sort: &[SearchComparator],
) -> trc::Result<Vec<R>> {
let mut query = format!(
"SELECT {} FROM {} ",
R::field().column(),
index.psql_table()
);
todo!()
}
pub async fn unindex(&self, query: SearchQuery) -> trc::Result<()> {
todo!()
}
fn build_filter<'x>(
&self,
query: &mut String,
filters: &'x [SearchFilter],
) -> Vec<&'x (dyn ToSql + Sync)> {
query.push_str("WHERE ");
let mut operator_stack = Vec::new();
let mut operator = &SearchFilter::And;
let mut is_first = true;
let values = Vec::new();
for filter in filters {
match filter {
SearchFilter::Operator { field, op, value } => {
if !is_first {
match operator {
SearchFilter::And => query.push_str(" AND "),
SearchFilter::Or => query.push_str(" OR "),
_ => (),
}
} else {
is_first = false;
}
query.push_str(field.column());
query.push(' ');
let value_ref = format!("${}", values.len() + 1);
if field.is_text() {
let language = match &value {
SearchValue::Text { language, .. }
if self.languages.contains(language) =>
{
pg_lang(language).unwrap_or("simple")
}
_ => "simple",
};
let method = match op {
SearchOperator::Equal => "phraseto_tsquery",
_ => "plainto_tsquery",
};
query.push_str(&format!("@@ {method}('{language}', {value_ref})"));
} else {
let todo = "jsonb query";
match op {
SearchOperator::LowerThan => {
query.push_str(" < ");
}
SearchOperator::LowerEqualThan => {
query.push_str(" <= ");
}
SearchOperator::GreaterThan => {
query.push_str(" > ");
}
SearchOperator::GreaterEqualThan => {
query.push_str(" >= ");
}
SearchOperator::Equal => {
query.push_str(" = ");
}
SearchOperator::Contains => {
query.push_str(" LIKE ");
}
}
query.push_str(&value_ref);
}
}
SearchFilter::And | SearchFilter::Or => {
operator_stack.push((operator, is_first));
operator = filter;
query.push('(');
}
SearchFilter::Not => {
operator_stack.push((operator, is_first));
operator = &SearchFilter::And;
query.push_str("NOT (");
}
SearchFilter::End => {
let p = operator_stack.pop().unwrap_or((&SearchFilter::And, true));
operator = p.0;
is_first = p.1;
query.push(')');
}
SearchFilter::DocumentSet(_) => (),
}
}
values
}
}
impl ToSql for SearchValue {
fn to_sql(
&self,
ty: &tokio_postgres::types::Type,
out: &mut bytes::BytesMut,
) -> Result<tokio_postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>>
where
Self: Sized,
{
match self {
SearchValue::Text { value, .. } => value.to_sql(ty, out),
SearchValue::Int(v) => match *ty {
Type::INT4 => (*v as i32).to_sql(ty, out),
_ => v.to_sql(ty, out),
},
SearchValue::Uint(v) => match *ty {
Type::INT4 => (*v as i32).to_sql(ty, out),
_ => (*v as i64).to_sql(ty, out),
},
SearchValue::Boolean(v) => v.to_sql(ty, out),
SearchValue::KeyValues(kv) => serde_json::to_string(kv)
.unwrap_or_default()
.to_sql(ty, out),
}
}
fn accepts(_: &tokio_postgres::types::Type) -> bool
where
Self: Sized,
{
true
}
fn to_sql_checked(
&self,
ty: &tokio_postgres::types::Type,
out: &mut bytes::BytesMut,
) -> Result<tokio_postgres::types::IsNull, Box<dyn std::error::Error + Sync + Send>> {
match self {
SearchValue::Text { value, .. } => value.to_sql_checked(ty, out),
SearchValue::Int(v) => match *ty {
Type::INT4 => (*v as i32).to_sql_checked(ty, out),
_ => v.to_sql_checked(ty, out),
},
SearchValue::Uint(v) => match *ty {
Type::INT4 => (*v as i32).to_sql_checked(ty, out),
_ => (*v as i64).to_sql_checked(ty, out),
},
SearchValue::Boolean(v) => v.to_sql_checked(ty, out),
SearchValue::KeyValues(kv) => serde_json::to_string(kv)
.unwrap_or_default()
.to_sql_checked(ty, out),
}
}
}
#[inline(always)]
fn pg_lang(lang: &Language) -> Option<&'static str> {
match lang {
Language::Esperanto => None,
Language::English => Some("english"),
Language::Russian => Some("russian"),
Language::Mandarin => None,
Language::Spanish => Some("spanish"),
Language::Portuguese => Some("portuguese"),
Language::Italian => Some("italian"),
Language::Bengali => None,
Language::French => Some("french"),
Language::German => Some("german"),
Language::Ukrainian => None,
Language::Georgian => None,
Language::Arabic => Some("arabic"),
Language::Hindi => Some("hindi"),
Language::Japanese => None,
Language::Hebrew => None,
Language::Yiddish => Some("yiddish"),
Language::Polish => Some("polish"),
Language::Amharic => None,
Language::Javanese => None,
Language::Korean => None,
Language::Bokmal => Some("norwegian"), // Norwegian covers Bokmål
Language::Danish => Some("danish"),
Language::Swedish => Some("swedish"),
Language::Finnish => Some("finnish"),
Language::Turkish => Some("turkish"),
Language::Dutch => Some("dutch"),
Language::Hungarian => Some("hungarian"),
Language::Czech => Some("czech"),
Language::Greek => Some("greek"),
Language::Bulgarian => None,
Language::Belarusian => None,
Language::Marathi => None,
Language::Kannada => None,
Language::Romanian => Some("romanian"),
Language::Slovene => None,
Language::Croatian => None,
Language::Serbian => Some("serbian"),
Language::Macedonian => None,
Language::Lithuanian => Some("lithuanian"),
Language::Latvian => None,
Language::Estonian => None,
Language::Tamil => Some("tamil"),
Language::Vietnamese => None,
Language::Urdu => None,
Language::Thai => None,
Language::Gujarati => None,
Language::Uzbek => None,
Language::Punjabi => None,
Language::Azerbaijani => None,
Language::Indonesian => Some("indonesian"),
Language::Telugu => None,
Language::Persian => None,
Language::Malayalam => None,
Language::Oriya => None,
Language::Burmese => None,
Language::Nepali => Some("nepali"),
Language::Sinhalese => None,
Language::Khmer => None,
Language::Turkmen => None,
Language::Akan => None,
Language::Zulu => None,
Language::Shona => None,
Language::Afrikaans => None,
Language::Latin => None,
Language::Slovak => None,
Language::Catalan => Some("catalan"),
Language::Tagalog => None,
Language::Armenian => Some("armenian"),
Language::Unknown | Language::None => None,
}
}

2
crates/store/src/backend/sqlite/lookup.rs
View File

@@ -11,7 +11,7 @@ use crate::{IntoRows, QueryResult, QueryType, Value};
use super::{SqliteStore, into_error};
impl SqliteStore {
pub(crate) async fn query<T: QueryResult>(
pub(crate) async fn sql_query<T: QueryResult>(
&self,
query: &str,
params_: &[Value<'_>],

25
crates/store/src/config.rs
View File

@@ -85,7 +85,8 @@ impl Stores {
.map(Store::from)
{
self.stores.insert(store_id.clone(), db.clone());
self.search_stores.insert(store_id.clone(), db.clone().into());
self.search_stores
.insert(store_id.clone(), db.clone().into());
self.blob_stores.insert(
store_id.clone(),
BlobStore::from(db.clone()).with_compression(compression_algo),
@@ -110,7 +111,8 @@ impl Stores {
.map(Store::from)
{
self.stores.insert(store_id.clone(), db.clone());
self.search_stores.insert(store_id.clone(), db.clone().into());
self.search_stores
.insert(store_id.clone(), db.clone().into());
self.blob_stores.insert(
store_id.clone(),
BlobStore::from(db.clone()).with_compression(compression_algo),
@@ -124,12 +126,14 @@ impl Stores {
config,
prefix,
config.is_active_store(id),
config.is_active_search_store(id),
)
.await
.map(Store::from)
{
self.stores.insert(store_id.clone(), db.clone());
self.search_stores.insert(store_id.clone(), db.clone().into());
self.search_stores
.insert(store_id.clone(), db.clone().into());
self.blob_stores.insert(
store_id.clone(),
BlobStore::from(db.clone()).with_compression(compression_algo),
@@ -143,12 +147,14 @@ impl Stores {
config,
prefix,
config.is_active_store(id),
config.is_active_search_store(id),
)
.await
.map(Store::from)
{
self.stores.insert(store_id.clone(), db.clone());
self.search_stores.insert(store_id.clone(), db.clone().into());
self.search_stores
.insert(store_id.clone(), db.clone().into());
self.blob_stores.insert(
store_id.clone(),
BlobStore::from(db.clone()).with_compression(compression_algo),
@@ -172,7 +178,8 @@ impl Stores {
crate::backend::sqlite::SqliteStore::open(config, prefix).map(Store::from)
{
self.stores.insert(store_id.clone(), db.clone());
self.search_stores.insert(store_id.clone(), db.clone().into());
self.search_stores
.insert(store_id.clone(), db.clone().into());
self.blob_stores.insert(
store_id.clone(),
BlobStore::from(db.clone()).with_compression(compression_algo),
@@ -298,6 +305,7 @@ impl Stores {
prefix,
self,
config.is_active_store(&id),
config.is_active_search_store(&id),
)
.await
{
@@ -419,6 +427,7 @@ impl Stores {
trait IsActiveStore {
fn is_active_store(&self, id: &str) -> bool;
fn is_active_in_memory_store(&self, id: &str) -> bool;
fn is_active_search_store(&self, id: &str) -> bool;
}
impl IsActiveStore for Config {
@@ -427,7 +436,6 @@ impl IsActiveStore for Config {
"storage.data",
"storage.blob",
"storage.lookup",
"storage.fts",
"tracing.history.store",
"metrics.history.store",
] {
@@ -441,6 +449,11 @@ impl IsActiveStore for Config {
false
}
fn is_active_search_store(&self, id: &str) -> bool {
self.value("storage.fts")
.is_some_and(|store_id| store_id == id)
}
fn is_active_in_memory_store(&self, id: &str) -> bool {
self.value("storage.lookup")
.is_some_and(|store_id| store_id == id)

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

@@ -127,11 +127,11 @@ impl Store {
) -> trc::Result<T> {
let result = match self {
#[cfg(feature = "sqlite")]
Self::SQLite(store) => store.query(query, &params).await,
Self::SQLite(store) => store.sql_query(query, &params).await,
#[cfg(feature = "postgres")]
Self::PostgreSQL(store) => store.query(query, &params).await,
Self::PostgreSQL(store) => store.sql_query(query, &params).await,
#[cfg(feature = "mysql")]
Self::MySQL(store) => store.query(query, &params).await,
Self::MySQL(store) => store.sql_query(query, &params).await,
_ => Err(trc::StoreEvent::NotSupported.into_err()),
};

477
crates/store/src/search/mod.rs
View File

@@ -11,7 +11,10 @@ pub mod query;
use ahash::AHashMap;
use nlp::language::Language;
use roaring::RoaringBitmap;
use std::{borrow::Cow, collections::hash_map::Entry};
use std::cmp::Ordering;
use std::collections::hash_map::Entry;
use std::ops::{BitAndAssign, BitOrAssign, BitXorAssign};
use utils::map::vec_map::VecMap;
use crate::write::SearchIndex;
@@ -23,7 +26,6 @@ pub enum SearchOperator {
GreaterEqualThan,
Equal,
Contains,
Exists,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -51,7 +53,7 @@ pub enum EmailSearchField {
SentAt,
Size,
HasAttachment,
Header(Cow<'static, str>),
Headers,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -67,7 +69,6 @@ pub enum CalendarSearchField {
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum ContactSearchField {
Created,
Member,
Kind,
Name,
@@ -91,31 +92,30 @@ pub enum FileSearchField {
pub enum TracingSearchField {
EventType,
QueueId,
Address,
RemoteIp,
Keywords,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SearchValue {
Text { value: String, language: Language },
KeyValues(VecMap<String, String>),
Int(i64),
Uint(u64),
Boolean(bool),
Keywords(Vec<SearchValue>),
}
pub trait SearchDocumentId: Sized {
fn from_u32(id: u32) -> Self;
fn from_u64(id: u64) -> Self;
fn field(&self) -> SearchField;
fn field() -> SearchField;
}
#[derive(Debug)]
pub struct SearchQuery {
index: SearchIndex,
filters: Vec<SearchFilter>,
comparators: Vec<SearchComparator>,
mask: RoaringBitmap,
pub(crate) index: SearchIndex,
pub(crate) filters: Vec<SearchFilter>,
pub(crate) comparators: Vec<SearchComparator>,
pub(crate) mask: RoaringBitmap,
}
#[derive(Debug)]
@@ -134,15 +134,24 @@ pub enum SearchFilter {
#[derive(Debug)]
pub enum SearchComparator {
Field { field: SearchField, ascending: bool },
DocumentSet { set: RoaringBitmap, ascending: bool },
SortedList { list: Vec<u32>, ascending: bool },
Field {
field: SearchField,
ascending: bool,
},
DocumentSet {
set: RoaringBitmap,
ascending: bool,
},
SortedSet {
set: AHashMap<u32, u32>,
ascending: bool,
},
}
#[derive(Debug)]
pub struct IndexDocument {
pub(crate) index: SearchIndex,
pub(crate) fields: AHashMap<SearchField, SearchValue>,
pub(crate) default_language: Language,
}
impl SearchFilter {
@@ -158,14 +167,6 @@ impl SearchFilter {
}
}
pub fn exists(field: impl Into<SearchField>) -> Self {
SearchFilter::Operator {
field: field.into(),
op: SearchOperator::Exists,
value: SearchValue::Boolean(true),
}
}
pub fn eq(field: impl Into<SearchField>, value: impl Into<SearchValue>) -> Self {
SearchFilter::Operator {
field: field.into(),
@@ -279,8 +280,8 @@ impl SearchComparator {
Self::DocumentSet { set, ascending }
}
pub fn sorted_list(list: Vec<u32>, ascending: bool) -> Self {
Self::SortedList { list, ascending }
pub fn sorted_set(set: AHashMap<u32, u32>, ascending: bool) -> Self {
Self::SortedSet { set, ascending }
}
pub fn ascending(field: impl Into<SearchField>) -> Self {
@@ -299,10 +300,10 @@ impl SearchComparator {
}
impl IndexDocument {
pub fn with_default_language(default_language: Language) -> Self {
pub fn new(index: SearchIndex) -> Self {
Self {
fields: Default::default(),
default_language,
index,
}
}
@@ -361,21 +362,24 @@ impl IndexDocument {
.insert(field.into(), SearchValue::Uint(value.into()));
}
pub fn insert_keyword(
pub fn insert_key_value(
&mut self,
field: impl Into<SearchField>,
keyword: impl Into<SearchValue>,
key: impl Into<String>,
value: impl Into<String>,
) {
let search_field = field.into();
match self.fields.entry(search_field) {
Entry::Occupied(mut entry) => {
if let SearchValue::Keywords(existing_keywords) = entry.get_mut() {
existing_keywords.push(keyword.into());
if let SearchValue::KeyValues(existing_key_values) = entry.get_mut() {
existing_key_values.append(key.into(), value.into());
}
}
Entry::Vacant(entry) => {
entry.insert(SearchValue::Keywords(vec![keyword.into()]));
let mut new_key_values = VecMap::new();
new_key_values.append(key.into(), value.into());
entry.insert(SearchValue::KeyValues(new_key_values));
}
}
}
@@ -387,6 +391,21 @@ impl IndexDocument {
pub fn has_field(&self, field: &SearchField) -> bool {
self.fields.contains_key(field)
}
pub fn set_unknown_language(&mut self, lang: Language) {
for value in self.fields.values_mut() {
if let SearchValue::Text { language, .. } = value
&& language.is_unknown()
{
*language = lang;
}
}
}
}
struct State {
pub op: SearchFilter,
pub bm: Option<RoaringBitmap>,
}
impl SearchQuery {
@@ -446,9 +465,148 @@ impl SearchQuery {
self
}
pub fn execute(&self) -> RoaringBitmap {
let todo = "implement search execution logic";
todo!()
pub fn filter(self) -> QueryResults {
if self.filters.is_empty() {
return QueryResults {
results: self.mask,
comparators: self.comparators,
};
}
let mut state: State = State {
op: SearchFilter::And,
bm: None,
};
let mut stack = Vec::new();
let mut filters = self.filters.into_iter().peekable();
let not_mask = self.mask;
while let Some(filter) = filters.next() {
let mut result = match filter {
SearchFilter::DocumentSet(set) => Some(set),
op @ (SearchFilter::And | SearchFilter::Or | SearchFilter::Not) => {
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;
}
}
SearchFilter::Operator { .. } => {
continue;
}
};
// 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);
}
}
SearchFilter::Not => {
if let Some(mut result) = result {
result.bitxor_assign(&not_mask);
dest.bitand_assign(result);
}
}
_ => unreachable!(),
}
} else if let Some(ref mut result_) = result {
if let SearchFilter::Not = state.op {
result_.bitxor_assign(&not_mask);
}
state.bm = result;
} else if let SearchFilter::Not = state.op {
state.bm = Some(not_mask.clone());
} else {
state.bm = Some(RoaringBitmap::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();
}
}
}
}
QueryResults {
results: state.bm.unwrap_or_default(),
comparators: self.comparators,
}
}
}
pub struct QueryResults {
results: RoaringBitmap,
comparators: Vec<SearchComparator>,
}
impl QueryResults {
pub fn results(&self) -> &RoaringBitmap {
&self.results
}
pub fn update_results(&mut self, results: RoaringBitmap) {
self.results = results;
}
pub fn into_bitmap(self) -> RoaringBitmap {
self.results
}
pub fn into_sorted(self) -> Vec<u32> {
let comparators = self.comparators;
let mut results = self.results.into_iter().collect::<Vec<u32>>();
if !results.is_empty() && !comparators.is_empty() {
results.sort_by(|a, b| {
for comparator in &comparators {
let (a, b, is_ascending) = match comparator {
SearchComparator::DocumentSet { set, ascending } => {
(set.contains(*a) as u32, set.contains(*b) as u32, *ascending)
}
SearchComparator::SortedSet { set, ascending } => (
*set.get(a).unwrap_or(&u32::MAX),
*set.get(b).unwrap_or(&u32::MAX),
*ascending,
),
SearchComparator::Field { .. } => continue,
};
let ordering = if is_ascending {
a.cmp(&b).reverse()
} else {
a.cmp(&b)
};
if ordering != Ordering::Equal {
return ordering;
}
}
Ordering::Equal
});
}
results
}
}
@@ -536,7 +694,7 @@ impl SearchDocumentId for u32 {
id as u32
}
fn field(&self) -> SearchField {
fn field() -> SearchField {
SearchField::DocumentId
}
}
@@ -550,7 +708,250 @@ impl SearchDocumentId for u64 {
id
}
fn field(&self) -> SearchField {
fn field() -> SearchField {
SearchField::Id
}
}
impl SearchIndex {
pub fn all_fields(&self) -> &[SearchField] {
match self {
SearchIndex::Email => EmailSearchField::all_fields(),
SearchIndex::Calendar => CalendarSearchField::all_fields(),
SearchIndex::Contacts => ContactSearchField::all_fields(),
SearchIndex::File => FileSearchField::all_fields(),
SearchIndex::Tracing => TracingSearchField::all_fields(),
SearchIndex::InMemory => unreachable!(),
}
}
pub fn primary_keys(&self) -> &'static [SearchField] {
match self {
SearchIndex::Email => EmailSearchField::primary_keys(),
SearchIndex::Calendar => CalendarSearchField::primary_keys(),
SearchIndex::Contacts => ContactSearchField::primary_keys(),
SearchIndex::File => FileSearchField::primary_keys(),
SearchIndex::Tracing => TracingSearchField::primary_keys(),
SearchIndex::InMemory => unreachable!(),
}
}
}
pub trait SearchableField: Sized {
fn index() -> SearchIndex;
fn primary_keys() -> &'static [SearchField];
fn all_fields() -> &'static [SearchField];
fn is_indexed(&self) -> bool;
fn is_text(&self) -> bool;
}
impl SearchableField for EmailSearchField {
fn index() -> SearchIndex {
SearchIndex::Email
}
fn primary_keys() -> &'static [SearchField] {
&[SearchField::AccountId, SearchField::DocumentId]
}
fn all_fields() -> &'static [SearchField] {
&[
SearchField::Email(EmailSearchField::From),
SearchField::Email(EmailSearchField::To),
SearchField::Email(EmailSearchField::Cc),
SearchField::Email(EmailSearchField::Bcc),
SearchField::Email(EmailSearchField::Subject),
SearchField::Email(EmailSearchField::Body),
SearchField::Email(EmailSearchField::Attachment),
SearchField::Email(EmailSearchField::ReceivedAt),
SearchField::Email(EmailSearchField::SentAt),
SearchField::Email(EmailSearchField::Size),
SearchField::Email(EmailSearchField::HasAttachment),
SearchField::Email(EmailSearchField::Headers),
]
}
fn is_indexed(&self) -> bool {
matches!(
self,
EmailSearchField::From
| EmailSearchField::To
| EmailSearchField::Subject
| EmailSearchField::ReceivedAt
| EmailSearchField::Size
| EmailSearchField::HasAttachment,
)
}
fn is_text(&self) -> bool {
matches!(
self,
EmailSearchField::From
| EmailSearchField::To
| EmailSearchField::Cc
| EmailSearchField::Bcc
| EmailSearchField::Subject
| EmailSearchField::Body
| EmailSearchField::Attachment,
)
}
}
impl SearchableField for CalendarSearchField {
fn index() -> SearchIndex {
SearchIndex::Calendar
}
fn primary_keys() -> &'static [SearchField] {
&[SearchField::AccountId, SearchField::DocumentId]
}
fn all_fields() -> &'static [SearchField] {
&[
SearchField::Calendar(CalendarSearchField::Title),
SearchField::Calendar(CalendarSearchField::Description),
SearchField::Calendar(CalendarSearchField::Location),
SearchField::Calendar(CalendarSearchField::Owner),
SearchField::Calendar(CalendarSearchField::Attendee),
SearchField::Calendar(CalendarSearchField::Start),
SearchField::Calendar(CalendarSearchField::Uid),
]
}
fn is_indexed(&self) -> bool {
matches!(self, CalendarSearchField::Start | CalendarSearchField::Uid)
}
fn is_text(&self) -> bool {
matches!(
self,
CalendarSearchField::Title
| CalendarSearchField::Description
| CalendarSearchField::Location
| CalendarSearchField::Owner
| CalendarSearchField::Attendee
)
}
}
impl SearchableField for ContactSearchField {
fn index() -> SearchIndex {
SearchIndex::Contacts
}
fn primary_keys() -> &'static [SearchField] {
&[SearchField::AccountId, SearchField::DocumentId]
}
fn all_fields() -> &'static [SearchField] {
&[
SearchField::Contact(ContactSearchField::Member),
SearchField::Contact(ContactSearchField::Kind),
SearchField::Contact(ContactSearchField::Name),
SearchField::Contact(ContactSearchField::Nickname),
SearchField::Contact(ContactSearchField::Organization),
SearchField::Contact(ContactSearchField::Email),
SearchField::Contact(ContactSearchField::Phone),
SearchField::Contact(ContactSearchField::OnlineService),
SearchField::Contact(ContactSearchField::Address),
SearchField::Contact(ContactSearchField::Note),
SearchField::Contact(ContactSearchField::Uid),
]
}
fn is_indexed(&self) -> bool {
matches!(self, ContactSearchField::Uid | ContactSearchField::Kind)
}
fn is_text(&self) -> bool {
matches!(
self,
ContactSearchField::Name
| ContactSearchField::Nickname
| ContactSearchField::Organization
| ContactSearchField::Email
| ContactSearchField::Phone
| ContactSearchField::OnlineService
| ContactSearchField::Address
| ContactSearchField::Note
)
}
}
impl SearchableField for FileSearchField {
fn index() -> SearchIndex {
SearchIndex::File
}
fn primary_keys() -> &'static [SearchField] {
&[SearchField::AccountId, SearchField::DocumentId]
}
fn all_fields() -> &'static [SearchField] {
&[
SearchField::File(FileSearchField::Name),
SearchField::File(FileSearchField::Content),
]
}
fn is_indexed(&self) -> bool {
false
}
fn is_text(&self) -> bool {
true
}
}
impl SearchableField for TracingSearchField {
fn index() -> SearchIndex {
SearchIndex::Tracing
}
fn primary_keys() -> &'static [SearchField] {
&[SearchField::Id]
}
fn all_fields() -> &'static [SearchField] {
&[
SearchField::Tracing(TracingSearchField::EventType),
SearchField::Tracing(TracingSearchField::QueueId),
SearchField::Tracing(TracingSearchField::Keywords),
]
}
fn is_indexed(&self) -> bool {
matches!(
self,
TracingSearchField::QueueId | TracingSearchField::EventType
)
}
fn is_text(&self) -> bool {
matches!(self, TracingSearchField::Keywords)
}
}
impl SearchField {
pub(crate) fn is_indexed(&self) -> bool {
match self {
SearchField::Email(field) => field.is_indexed(),
SearchField::Calendar(field) => field.is_indexed(),
SearchField::Contact(field) => field.is_indexed(),
SearchField::File(field) => field.is_indexed(),
SearchField::Tracing(field) => field.is_indexed(),
SearchField::AccountId | SearchField::DocumentId | SearchField::Id => false,
}
}
pub(crate) fn is_text(&self) -> bool {
match self {
SearchField::Email(field) => field.is_text(),
SearchField::Calendar(field) => field.is_text(),
SearchField::Contact(field) => field.is_text(),
SearchField::File(field) => field.is_text(),
SearchField::Tracing(field) => field.is_text(),
SearchField::AccountId | SearchField::DocumentId | SearchField::Id => false,
}
}
}

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

@@ -667,7 +667,7 @@ impl SearchIndex {
SearchIndex::Calendar => 1,
SearchIndex::Contacts => 2,
SearchIndex::File => 3,
SearchIndex::TracingSpan => 4,
SearchIndex::Tracing => 4,
SearchIndex::InMemory => unreachable!(),
}
}
@@ -678,7 +678,7 @@ impl SearchIndex {
1 => Some(SearchIndex::Calendar),
2 => Some(SearchIndex::Contacts),
3 => Some(SearchIndex::File),
4 => Some(SearchIndex::TracingSpan),
4 => Some(SearchIndex::Tracing),
_ => None,
}
}
@@ -689,7 +689,7 @@ impl SearchIndex {
SearchIndex::Calendar => "calendar",
SearchIndex::Contacts => "contacts",
SearchIndex::File => "file",
SearchIndex::TracingSpan => "tracing",
SearchIndex::Tracing => "tracing",
SearchIndex::InMemory => "in_memory",
}
}
@@ -700,7 +700,7 @@ impl SearchIndex {
"calendar" => Some(SearchIndex::Calendar),
"contacts" => Some(SearchIndex::Contacts),
"file" => Some(SearchIndex::File),
"tracing" => Some(SearchIndex::TracingSpan),
"tracing" => Some(SearchIndex::Tracing),
_ => None,
}
}

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

@@ -220,7 +220,7 @@ pub enum SearchIndex {
Calendar,
Contacts,
File,
TracingSpan,
Tracing,
InMemory,
}

2
crates/types/src/field.rs
View File

@@ -19,6 +19,7 @@ pub enum ContactField {
Uid,
Email,
Archive,
CreatedToUpdated,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -83,6 +84,7 @@ impl From<ContactField> for u8 {
match value {
ContactField::Uid => 0,
ContactField::Email => 1,
ContactField::CreatedToUpdated => 2,
ContactField::Archive => ARCHIVE_FIELD,
}
}