use std::{
collections::{BTreeMap, BTreeSet},
time::{Duration, Instant},
};
use hyphae_core::VectorSpaceName;
use hyphae_query::{FieldPath, Record, Value};
use thiserror::Error;
use unicode_casefold::UnicodeCaseFold;
use unicode_normalization::UnicodeNormalization;
pub const MAX_LEXICAL_TOKEN_BYTES: usize = 256;
pub const MAX_LEXICAL_FIELD_WEIGHT_MICROS: u32 = 1_000_000_000;
pub const MAX_LEXICAL_FIELDS: usize = 64;
pub const MAX_LEXICAL_PATH_SEGMENTS: usize = 32;
pub const MAX_LEXICAL_PATH_SEGMENT_BYTES: usize = 1_024;
const WEIGHT_SCALE: f64 = 1_000_000.0;
const K1: f64 = 1.2;
const B: f64 = 0.75;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalField {
pub path: FieldPath,
pub weight_micros: u32,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalIndexDefinition {
pub name: VectorSpaceName,
pub fields: Vec<LexicalField>,
}
impl LexicalIndexDefinition {
pub fn new(name: VectorSpaceName, mut fields: Vec<LexicalField>) -> Result<Self, LexicalError> {
if fields.is_empty() {
return Err(LexicalError::EmptyFields);
}
if fields.len() > MAX_LEXICAL_FIELDS {
return Err(LexicalError::TooManyFields);
}
if fields.iter().any(|field| field.path.segments().is_empty()) {
return Err(LexicalError::EmptyFieldPath);
}
if fields.iter().any(|field| {
field.path.segments().len() > MAX_LEXICAL_PATH_SEGMENTS
|| field.path.segments().iter().any(|segment| {
segment.is_empty() || segment.len() > MAX_LEXICAL_PATH_SEGMENT_BYTES
})
}) {
return Err(LexicalError::InvalidFieldSegment);
}
if fields
.iter()
.any(|field| !(1..=MAX_LEXICAL_FIELD_WEIGHT_MICROS).contains(&field.weight_micros))
{
return Err(LexicalError::InvalidFieldWeight);
}
fields.sort_by(|left, right| left.path.cmp(&right.path));
if fields.windows(2).any(|pair| pair[0].path == pair[1].path) {
return Err(LexicalError::DuplicateFieldPath);
}
Ok(Self { name, fields })
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalRequest {
pub index: VectorSpaceName,
pub query: String,
pub limit: usize,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalLimits {
pub max_documents: u64,
pub max_tokens: u64,
pub max_candidates: u64,
pub max_returned: usize,
pub timeout: Duration,
}
impl Default for LexicalLimits {
fn default() -> Self {
Self {
max_documents: 1_000_000,
max_tokens: 10_000_000,
max_candidates: 100_000,
max_returned: 1_000,
timeout: Duration::from_secs(30),
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalFieldContribution {
pub path: FieldPath,
pub term_frequency: u64,
pub field_length: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalTermContribution {
pub token: String,
pub document_frequency: u64,
pub score_nanos: i64,
pub fields: Vec<LexicalFieldContribution>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalMatch {
pub key: Vec<u8>,
pub score_nanos: i64,
pub terms: Vec<LexicalTermContribution>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LexicalAbstentionReason {
NoCandidates,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalAbstention {
pub reason: LexicalAbstentionReason,
pub scanned_documents: u64,
pub query_tokens: Vec<String>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum LexicalOutcome {
Matches {
matches: Vec<LexicalMatch>,
scanned_documents: u64,
matched_documents: u64,
query_tokens: Vec<String>,
},
Abstained(LexicalAbstention),
}
#[derive(Clone, Debug, Error, Eq, PartialEq)]
pub enum LexicalError {
#[error("lexical definition requires at least one field")]
EmptyFields,
#[error("lexical definition exceeds 64 fields")]
TooManyFields,
#[error("lexical field path must be nonempty")]
EmptyFieldPath,
#[error("lexical field path contains an invalid segment")]
InvalidFieldSegment,
#[error("lexical field paths must be unique")]
DuplicateFieldPath,
#[error("lexical field weight is outside 1..=1000000000")]
InvalidFieldWeight,
#[error("lexical request index does not match the definition")]
IndexMismatch,
#[error("lexical query has no retained normalized tokens")]
EmptyQuery,
#[error("lexical result limit must be nonzero")]
ZeroLimit,
#[error("lexical result limit {requested} exceeds maximum {maximum}")]
ResultLimitExceeded {
requested: usize,
maximum: usize,
},
#[error("lexical document key must be nonempty")]
EmptyDocumentKey,
#[error("duplicate lexical document key")]
DuplicateDocumentKey,
#[error("lexical document budget exceeded: {maximum}")]
DocumentBudgetExceeded {
maximum: u64,
},
#[error("lexical token budget exceeded: {maximum}")]
TokenBudgetExceeded {
maximum: u64,
},
#[error("lexical candidate budget exceeded: {maximum}")]
CandidateBudgetExceeded {
maximum: u64,
},
#[error("materialized lexical projection is malformed")]
MalformedProjection,
#[error("lexical retrieval timed out")]
TimedOut,
#[error("lexical score arithmetic overflow or non-finite result")]
ArithmeticOverflow,
}
pub fn tokenize_v1(input: &str) -> Vec<String> {
let normalized = input.nfkc().case_fold().collect::<String>();
let mut tokens = Vec::new();
let mut token = String::new();
for character in normalized.chars() {
if character.is_alphanumeric() {
token.push(character);
} else {
push_token(&mut tokens, &mut token);
}
}
push_token(&mut tokens, &mut token);
tokens
}
fn push_token(tokens: &mut Vec<String>, token: &mut String) {
if !token.is_empty() && token.len() <= MAX_LEXICAL_TOKEN_BYTES {
tokens.push(std::mem::take(token));
} else {
token.clear();
}
}
struct AnalyzedDocument {
key: Vec<u8>,
fields: Vec<Vec<String>>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalMaterializedDocument {
pub key: Vec<u8>,
pub field_lengths: Vec<u64>,
pub term_frequencies: BTreeMap<String, Vec<u64>>,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct LexicalMaterializedCorpus {
pub document_count: u64,
pub token_count: u64,
pub total_field_lengths: Vec<u64>,
pub documents: Vec<LexicalMaterializedDocument>,
}
pub fn retrieve_lexical(
records: &[Record],
definition: &LexicalIndexDefinition,
request: &LexicalRequest,
limits: &LexicalLimits,
) -> Result<LexicalOutcome, LexicalError> {
validate_request(definition, request, limits)?;
let query_tokens = tokenize_v1(&request.query)
.into_iter()
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
if query_tokens.is_empty() {
return Err(LexicalError::EmptyQuery);
}
let started = Instant::now();
let mut token_count = u64::try_from(query_tokens.len()).unwrap_or(u64::MAX);
let mut keys = BTreeSet::new();
let mut documents = Vec::with_capacity(records.len());
let mut total_lengths = vec![0_u64; definition.fields.len()];
for record in records {
check_timeout(started, limits.timeout)?;
if u64::try_from(documents.len()).unwrap_or(u64::MAX) >= limits.max_documents {
return Err(LexicalError::DocumentBudgetExceeded {
maximum: limits.max_documents,
});
}
if record.key.is_empty() {
return Err(LexicalError::EmptyDocumentKey);
}
if !keys.insert(record.key.as_slice()) {
return Err(LexicalError::DuplicateDocumentKey);
}
let mut fields = Vec::with_capacity(definition.fields.len());
for (field_index, field) in definition.fields.iter().enumerate() {
let tokens = match field.path.resolve(&record.value) {
Some(Value::String(value)) => tokenize_v1(value),
_ => Vec::new(),
};
let length = u64::try_from(tokens.len()).unwrap_or(u64::MAX);
token_count =
token_count
.checked_add(length)
.ok_or(LexicalError::TokenBudgetExceeded {
maximum: limits.max_tokens,
})?;
if token_count > limits.max_tokens {
return Err(LexicalError::TokenBudgetExceeded {
maximum: limits.max_tokens,
});
}
total_lengths[field_index] = total_lengths[field_index]
.checked_add(length)
.ok_or(LexicalError::ArithmeticOverflow)?;
fields.push(tokens);
}
documents.push(AnalyzedDocument {
key: record.key.clone(),
fields,
});
}
score_documents(
&documents,
&total_lengths,
definition,
request,
limits,
&query_tokens,
started,
)
}
pub fn retrieve_lexical_materialized(
corpus: &LexicalMaterializedCorpus,
definition: &LexicalIndexDefinition,
request: &LexicalRequest,
limits: &LexicalLimits,
) -> Result<LexicalOutcome, LexicalError> {
validate_request(definition, request, limits)?;
let query_tokens = tokenize_v1(&request.query)
.into_iter()
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
if query_tokens.is_empty() {
return Err(LexicalError::EmptyQuery);
}
validate_materialized_corpus(corpus, definition, &query_tokens, limits)?;
let started = Instant::now();
let averages = corpus
.total_field_lengths
.iter()
.map(|length| {
if corpus.document_count == 0 {
0.0
} else {
bounded_count_as_f64(*length) / bounded_count_as_f64(corpus.document_count)
}
})
.collect::<Vec<_>>();
let frequencies = query_tokens
.iter()
.map(|token| {
let frequency = corpus
.documents
.iter()
.filter(|document| {
document
.term_frequencies
.get(token)
.is_some_and(|fields| fields.iter().any(|frequency| *frequency > 0))
})
.count();
(token.clone(), u64::try_from(frequency).unwrap_or(u64::MAX))
})
.collect::<BTreeMap<_, _>>();
let mut matches = Vec::with_capacity(corpus.documents.len().min(request.limit));
for document in &corpus.documents {
check_timeout(started, limits.timeout)?;
if let Some(matched) = score_materialized_document(
document,
corpus.document_count,
&averages,
&frequencies,
definition,
&query_tokens,
)? {
matches.push(matched);
}
}
Ok(finish_ranking(
matches,
corpus.document_count,
&query_tokens,
request.limit,
))
}
fn validate_materialized_corpus(
corpus: &LexicalMaterializedCorpus,
definition: &LexicalIndexDefinition,
query_tokens: &[String],
limits: &LexicalLimits,
) -> Result<(), LexicalError> {
if corpus.document_count > limits.max_documents {
return Err(LexicalError::DocumentBudgetExceeded {
maximum: limits.max_documents,
});
}
let total_tokens = corpus
.token_count
.checked_add(u64::try_from(query_tokens.len()).unwrap_or(u64::MAX))
.ok_or(LexicalError::TokenBudgetExceeded {
maximum: limits.max_tokens,
})?;
if total_tokens > limits.max_tokens {
return Err(LexicalError::TokenBudgetExceeded {
maximum: limits.max_tokens,
});
}
if u64::try_from(corpus.documents.len()).unwrap_or(u64::MAX) > limits.max_candidates {
return Err(LexicalError::CandidateBudgetExceeded {
maximum: limits.max_candidates,
});
}
if corpus.total_field_lengths.len() != definition.fields.len() {
return Err(LexicalError::MalformedProjection);
}
let mut keys = BTreeSet::new();
for document in &corpus.documents {
if document.key.is_empty() {
return Err(LexicalError::EmptyDocumentKey);
}
if !keys.insert(document.key.as_slice()) {
return Err(LexicalError::DuplicateDocumentKey);
}
if document.field_lengths.len() != definition.fields.len()
|| document.term_frequencies.len() != query_tokens.len()
|| query_tokens.iter().any(|token| {
document
.term_frequencies
.get(token)
.is_none_or(|frequencies| frequencies.len() != definition.fields.len())
})
{
return Err(LexicalError::MalformedProjection);
}
}
Ok(())
}
fn validate_request(
definition: &LexicalIndexDefinition,
request: &LexicalRequest,
limits: &LexicalLimits,
) -> Result<(), LexicalError> {
if request.index != definition.name {
return Err(LexicalError::IndexMismatch);
}
if request.limit == 0 {
return Err(LexicalError::ZeroLimit);
}
if request.limit > limits.max_returned {
return Err(LexicalError::ResultLimitExceeded {
requested: request.limit,
maximum: limits.max_returned,
});
}
Ok(())
}
fn score_documents(
documents: &[AnalyzedDocument],
total_lengths: &[u64],
definition: &LexicalIndexDefinition,
request: &LexicalRequest,
limits: &LexicalLimits,
query_tokens: &[String],
started: Instant,
) -> Result<LexicalOutcome, LexicalError> {
let document_count = u64::try_from(documents.len()).unwrap_or(u64::MAX);
let averages = total_lengths
.iter()
.map(|length| {
if document_count == 0 {
0.0
} else {
bounded_count_as_f64(*length) / bounded_count_as_f64(document_count)
}
})
.collect::<Vec<_>>();
let frequencies = query_tokens
.iter()
.map(|token| {
let count = documents
.iter()
.filter(|document| {
document
.fields
.iter()
.any(|field| field.iter().any(|candidate| candidate == token))
})
.count();
(token.clone(), u64::try_from(count).unwrap_or(u64::MAX))
})
.collect::<BTreeMap<_, _>>();
let mut matches = Vec::new();
for document in documents {
check_timeout(started, limits.timeout)?;
let document_match = score_document(
document,
document_count,
&averages,
&frequencies,
definition,
query_tokens,
)?;
if let Some(matched) = document_match {
if u64::try_from(matches.len()).unwrap_or(u64::MAX) >= limits.max_candidates {
return Err(LexicalError::CandidateBudgetExceeded {
maximum: limits.max_candidates,
});
}
matches.push(matched);
}
}
Ok(finish_ranking(
matches,
document_count,
query_tokens,
request.limit,
))
}
fn score_document(
document: &AnalyzedDocument,
document_count: u64,
averages: &[f64],
frequencies: &BTreeMap<String, u64>,
definition: &LexicalIndexDefinition,
query_tokens: &[String],
) -> Result<Option<LexicalMatch>, LexicalError> {
let field_lengths = document
.fields
.iter()
.map(|field| u64::try_from(field.len()).unwrap_or(u64::MAX))
.collect::<Vec<_>>();
let term_frequencies = query_tokens
.iter()
.map(|token| {
let frequencies = document
.fields
.iter()
.map(|field| {
u64::try_from(field.iter().filter(|candidate| *candidate == token).count())
.unwrap_or(u64::MAX)
})
.collect::<Vec<_>>();
(token.clone(), frequencies)
})
.collect::<BTreeMap<_, _>>();
score_statistics(
&document.key,
&field_lengths,
&term_frequencies,
document_count,
averages,
frequencies,
definition,
query_tokens,
)
}
fn score_materialized_document(
document: &LexicalMaterializedDocument,
document_count: u64,
averages: &[f64],
frequencies: &BTreeMap<String, u64>,
definition: &LexicalIndexDefinition,
query_tokens: &[String],
) -> Result<Option<LexicalMatch>, LexicalError> {
score_statistics(
&document.key,
&document.field_lengths,
&document.term_frequencies,
document_count,
averages,
frequencies,
definition,
query_tokens,
)
}
#[allow(clippy::too_many_arguments)]
fn score_statistics(
key: &[u8],
field_lengths: &[u64],
term_frequencies: &BTreeMap<String, Vec<u64>>,
document_count: u64,
averages: &[f64],
frequencies: &BTreeMap<String, u64>,
definition: &LexicalIndexDefinition,
query_tokens: &[String],
) -> Result<Option<LexicalMatch>, LexicalError> {
let mut terms = Vec::new();
let mut score_nanos = 0_i64;
for token in query_tokens {
let document_frequency = frequencies[token];
if document_frequency == 0 {
continue;
}
let mut combined_tf = 0.0_f64;
let mut fields = Vec::with_capacity(definition.fields.len());
for (index, definition_field) in definition.fields.iter().enumerate() {
let term_frequency = term_frequencies[token][index];
let field_length = field_lengths[index];
fields.push(LexicalFieldContribution {
path: definition_field.path.clone(),
term_frequency,
field_length,
});
if term_frequency > 0 && averages[index] > 0.0 {
let normalization =
1.0 - B + B * bounded_count_as_f64(field_length) / averages[index];
combined_tf += (f64::from(definition_field.weight_micros) / WEIGHT_SCALE)
* bounded_count_as_f64(term_frequency)
/ normalization;
}
}
if combined_tf == 0.0 {
continue;
}
let numerator =
bounded_count_as_f64(document_count.saturating_sub(document_frequency)) + 0.5;
let denominator = bounded_count_as_f64(document_frequency) + 0.5;
let idf = libm::log(1.0 + numerator / denominator);
let term_score = quantize_score(idf * combined_tf * (K1 + 1.0) / (combined_tf + K1))?;
score_nanos = score_nanos
.checked_add(term_score)
.ok_or(LexicalError::ArithmeticOverflow)?;
terms.push(LexicalTermContribution {
token: token.clone(),
document_frequency,
score_nanos: term_score,
fields,
});
}
Ok((score_nanos > 0).then(|| LexicalMatch {
key: key.to_vec(),
score_nanos,
terms,
}))
}
fn finish_ranking(
mut matches: Vec<LexicalMatch>,
document_count: u64,
query_tokens: &[String],
limit: usize,
) -> LexicalOutcome {
matches.sort_by(|left, right| {
right
.score_nanos
.cmp(&left.score_nanos)
.then_with(|| left.key.cmp(&right.key))
});
let matched_documents = u64::try_from(matches.len()).unwrap_or(u64::MAX);
matches.truncate(limit);
if matches.is_empty() {
LexicalOutcome::Abstained(LexicalAbstention {
reason: LexicalAbstentionReason::NoCandidates,
scanned_documents: document_count,
query_tokens: query_tokens.to_vec(),
})
} else {
LexicalOutcome::Matches {
matches,
scanned_documents: document_count,
matched_documents,
query_tokens: query_tokens.to_vec(),
}
}
}
fn quantize_score(value: f64) -> Result<i64, LexicalError> {
if !value.is_finite() || value < 0.0 {
return Err(LexicalError::ArithmeticOverflow);
}
let scaled = value * 1_000_000_000.0;
if !scaled.is_finite() {
return Err(LexicalError::ArithmeticOverflow);
}
if scaled >= maximum_i64_as_f64() {
return Ok(i64::MAX);
}
Ok(rounded_nonnegative_f64_as_i64(scaled))
}
#[allow(clippy::cast_precision_loss)]
fn bounded_count_as_f64(value: u64) -> f64 {
value as f64
}
#[allow(clippy::cast_precision_loss)]
fn maximum_i64_as_f64() -> f64 {
i64::MAX as f64
}
#[allow(clippy::cast_possible_truncation)]
fn rounded_nonnegative_f64_as_i64(value: f64) -> i64 {
libm::floor(value + 0.5) as i64
}
fn check_timeout(started: Instant, timeout: Duration) -> Result<(), LexicalError> {
if started.elapsed() >= timeout {
Err(LexicalError::TimedOut)
} else {
Ok(())
}
}
#[cfg(test)]
mod tests {
use std::collections::{BTreeMap, BTreeSet};
use super::*;
use proptest::prelude::*;
fn record(key: &[u8], title: &str, body: &str) -> Record {
Record::new(
key,
Value::Object(BTreeMap::from([
("title".into(), Value::String(title.into())),
("body".into(), Value::String(body.into())),
])),
)
}
fn definition() -> Result<LexicalIndexDefinition, LexicalError> {
LexicalIndexDefinition::new(
VectorSpaceName::new("docs").map_err(|_| LexicalError::EmptyFields)?,
vec![
LexicalField {
path: FieldPath::field("body"),
weight_micros: 1_000_000,
},
LexicalField {
path: FieldPath::field("title"),
weight_micros: 2_000_000,
},
],
)
}
fn materialize_reference_corpus(
records: &[Record],
definition: &LexicalIndexDefinition,
query: &str,
) -> LexicalMaterializedCorpus {
let query_tokens = tokenize_v1(query)
.into_iter()
.collect::<BTreeSet<_>>()
.into_iter()
.collect::<Vec<_>>();
let mut token_count = 0_u64;
let mut total_field_lengths = vec![0_u64; definition.fields.len()];
let mut documents = Vec::new();
for record in records {
let fields = definition
.fields
.iter()
.map(|field| match field.path.resolve(&record.value) {
Some(Value::String(value)) => tokenize_v1(value),
_ => Vec::new(),
})
.collect::<Vec<_>>();
let field_lengths = fields
.iter()
.map(|field| u64::try_from(field.len()).unwrap_or(u64::MAX))
.collect::<Vec<_>>();
for (total, length) in total_field_lengths.iter_mut().zip(&field_lengths) {
*total = total.saturating_add(*length);
token_count = token_count.saturating_add(*length);
}
let term_frequencies = query_tokens
.iter()
.map(|token| {
let frequencies = fields
.iter()
.map(|field| {
u64::try_from(
field.iter().filter(|candidate| *candidate == token).count(),
)
.unwrap_or(u64::MAX)
})
.collect::<Vec<_>>();
(token.clone(), frequencies)
})
.collect::<BTreeMap<_, _>>();
if term_frequencies
.values()
.any(|frequencies| frequencies.iter().any(|frequency| *frequency > 0))
{
documents.push(LexicalMaterializedDocument {
key: record.key.clone(),
field_lengths,
term_frequencies,
});
}
}
LexicalMaterializedCorpus {
document_count: u64::try_from(records.len()).unwrap_or(u64::MAX),
token_count,
total_field_lengths,
documents,
}
}
#[test]
fn tokenizer_pins_nfkc_casefold_and_alphanumeric_runs() {
assert_eq!(
tokenize_v1("Straße ABC—café"),
vec!["strasse", "abc", "café"]
);
}
#[test]
fn bm25f_is_deterministic_and_binary_key_breaks_ties() -> Result<(), LexicalError> {
let definition = definition()?;
let outcome = retrieve_lexical(
&[
record(b"b", "Rust memory", "durable engine"),
record(b"a", "Rust memory", "durable engine"),
record(b"z", "other", "nothing"),
],
&definition,
&LexicalRequest {
index: definition.name.clone(),
query: "RUST rust".into(),
limit: 10,
},
&LexicalLimits::default(),
)?;
let LexicalOutcome::Matches { matches, .. } = outcome else {
return Err(LexicalError::ArithmeticOverflow);
};
assert_eq!(matches[0].key, b"a");
assert_eq!(matches[1].key, b"b");
assert_eq!(matches[0].score_nanos, matches[1].score_nanos);
Ok(())
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(64))]
#[test]
fn materialized_scorer_matches_reference_for_random_corpora(
generated in prop::collection::vec(
("[a-z ]{0,24}", "[a-z ]{0,48}"),
0..32
),
query in "(rust|durable|engine|memory)( (rust|durable|engine|memory)){0,2}",
limit in 1_usize..16
) {
let definition = definition().map_err(|error| TestCaseError::fail(error.to_string()))?;
let records = generated
.iter()
.enumerate()
.map(|(index, (title, body))| {
record(&u64::try_from(index).unwrap_or(u64::MAX).to_be_bytes(), title, body)
})
.collect::<Vec<_>>();
let request = LexicalRequest {
index: definition.name.clone(),
query: query.clone(),
limit,
};
let limits = LexicalLimits::default();
let reference = retrieve_lexical(&records, &definition, &request, &limits)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
let corpus = materialize_reference_corpus(&records, &definition, &query);
let materialized =
retrieve_lexical_materialized(&corpus, &definition, &request, &limits)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(materialized, reference);
}
}
}