mod cc;
pub(crate) use cc::StoreFusionScratch;
mod normalize;
mod rrf;
mod rules;
use std::collections::BTreeSet;
pub use rules::{
ALPHA_POLICY_VERSION, FusionRule, LEXICAL_RULE_ALPHA, RARE_DOCUMENT_FREQUENCY_THRESHOLD,
RuleSignals,
};
pub const DEFAULT_ALPHA: f64 = 0.7;
pub const HYBRID_NORMALIZATION_POLICY_VERSION: u16 = 1;
pub const RRF_K: u32 = 60;
pub const DEFAULT_MAX_ROUNDS: usize = 8;
pub const HYBRID_WINDOW_FLOOR: usize = 50;
pub const HYBRID_WINDOW_PER_K: usize = 5;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ScorePrecision {
Exact,
Estimated,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CandidateCoverage {
Exhaustive,
CertifiedBounded,
Approximate,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct HybridProvenance {
pub vector_precision: ScorePrecision,
pub vector_coverage: CandidateCoverage,
pub lexical_coverage: CandidateCoverage,
pub cross_scores_complete: bool,
}
#[derive(Clone, Debug, PartialEq)]
pub struct VectorCandidate<I> {
id: I,
squared_l2: f64,
precision: ScorePrecision,
}
impl<I> VectorCandidate<I> {
#[must_use]
pub const fn exact(id: I, squared_l2: f64) -> Self {
Self {
id,
squared_l2,
precision: ScorePrecision::Exact,
}
}
#[must_use]
pub const fn estimated(id: I, score: f64) -> Self {
Self {
id,
squared_l2: score,
precision: ScorePrecision::Estimated,
}
}
#[must_use]
pub const fn id(&self) -> &I {
&self.id
}
#[must_use]
pub const fn squared_l2(&self) -> f64 {
self.squared_l2
}
#[must_use]
pub const fn precision(&self) -> ScorePrecision {
self.precision
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct LexicalCandidate<I> {
id: I,
bm25: f64,
}
impl<I> LexicalCandidate<I> {
#[must_use]
pub const fn new(id: I, bm25: f64) -> Self {
Self { id, bm25 }
}
#[must_use]
pub const fn id(&self) -> &I {
&self.id
}
#[must_use]
pub const fn bm25(&self) -> f64 {
self.bm25
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct HybridQuery {
pub k: usize,
pub alpha: Option<f64>,
pub rule_signals: RuleSignals,
pub rules_enabled: bool,
pub max_rounds: usize,
pub epoch: Option<crate::epoch::EpochIdentity>,
}
impl HybridQuery {
#[must_use]
pub const fn new(k: usize) -> Self {
Self {
k,
alpha: None,
rule_signals: RuleSignals::none(),
rules_enabled: false,
max_rounds: DEFAULT_MAX_ROUNDS,
epoch: None,
}
}
#[must_use]
pub const fn with_alpha(mut self, alpha: f64) -> Self {
self.alpha = Some(alpha);
self
}
#[must_use]
pub const fn with_rule_signals(mut self, signals: RuleSignals) -> Self {
self.rule_signals = signals;
self
}
#[must_use]
pub const fn with_rules(mut self) -> Self {
self.rules_enabled = true;
self
}
#[must_use]
pub const fn without_rules(mut self) -> Self {
self.rules_enabled = false;
self
}
#[must_use]
pub const fn with_max_rounds(mut self, max_rounds: usize) -> Self {
self.max_rounds = max_rounds;
self
}
#[must_use]
pub const fn with_epoch(mut self, epoch: crate::epoch::EpochIdentity) -> Self {
self.epoch = Some(epoch);
self
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FusionMethod {
ConvexCombination,
ReciprocalRankFusion,
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub enum FusionLeg {
Vector,
Lexical,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum DegenerateKind {
Empty,
SingleHit,
AllScoresEqual,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct DegenerateLeg {
pub leg: FusionLeg,
pub kind: DegenerateKind,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FusionTermination {
StableBound,
ListsExhausted,
BudgetFullMaterialization,
WindowUnproven,
ApproximateCandidates,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct VectorBounds {
pub min_squared_l2: f64,
pub max_squared_l2: f64,
pub next_unseen_squared_l2: Option<f64>,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct LexicalBounds {
pub max_bm25: f64,
pub min_bm25: f64,
pub next_unseen_bm25: Option<f64>,
}
#[derive(Clone, Copy, Debug, Default, PartialEq)]
pub struct LegBounds {
pub vector: Option<VectorBounds>,
pub lexical: Option<LexicalBounds>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct FusionReport {
pub method: FusionMethod,
pub effective_alpha: f64,
pub applied_rules: Vec<FusionRule>,
pub degenerate_legs: Vec<DegenerateLeg>,
pub rounds: usize,
pub budget_exhausted: bool,
pub termination: FusionTermination,
pub alpha_policy_version: u16,
pub epoch: Option<crate::epoch::EpochIdentity>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct FusedHit<K> {
pub key: K,
pub vector_squared_l2: Option<f64>,
pub lexical_bm25: Option<f64>,
pub fused_score: f64,
}
#[derive(Clone, Debug, PartialEq)]
pub struct FusionOutcome<K> {
pub hits: Vec<FusedHit<K>>,
pub report: FusionReport,
}
#[derive(Clone, Debug, PartialEq)]
pub enum FusionError {
InvalidAlpha(f64),
NonFiniteScore {
leg: FusionLeg,
rank: usize,
},
NegativeScore {
leg: FusionLeg,
rank: usize,
},
EstimatedVectorScore {
rank: usize,
},
InvalidBounds {
leg: FusionLeg,
detail: &'static str,
},
UnrankedInput {
leg: FusionLeg,
rank: usize,
},
MissingDocumentIdentity {
leg: FusionLeg,
rank: usize,
},
DuplicateDocumentIdentity {
leg: FusionLeg,
rank: usize,
},
Timeout {
partial: bool,
},
Cancelled {
partial: bool,
},
ReadCancelled {
partial: bool,
},
LegThreadStart {
leg: FusionLeg,
detail: String,
},
LegPanic {
leg: FusionLeg,
detail: &'static str,
},
Leg {
leg: FusionLeg,
kind: LegFailureKind,
detail: String,
},
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum LegFailureKind {
Store(crate::lifecycle::StoreErrorKind),
Scan,
Graph,
Segment,
Lexical,
Invariant,
Caller,
}
impl std::fmt::Display for FusionError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::InvalidAlpha(alpha) => write!(formatter, "fusion alpha {alpha} is outside 0..=1"),
Self::NonFiniteScore { leg, rank } => {
write!(formatter, "{leg:?} score at rank {rank} is non-finite")
}
Self::NegativeScore { leg, rank } => {
write!(formatter, "{leg:?} score at rank {rank} is negative")
}
Self::EstimatedVectorScore { rank } => {
write!(
formatter,
"vector score at rank {rank} was not exactly rescored"
)
}
Self::InvalidBounds { leg, detail } => {
write!(formatter, "{leg:?} leg bounds are invalid: {detail}")
}
Self::UnrankedInput { leg, rank } => {
write!(
formatter,
"{leg:?} input changes score direction at rank {rank}"
)
}
Self::MissingDocumentIdentity { leg, rank } => write!(
formatter,
"{leg:?} candidate at rank {rank} lacks a common document identity"
),
Self::DuplicateDocumentIdentity { leg, rank } => write!(
formatter,
"{leg:?} candidate at rank {rank} repeats a common document identity"
),
Self::Timeout { partial } => {
write!(
formatter,
"hybrid query deadline expired (partial={partial})"
)
}
Self::Cancelled { partial } => {
write!(formatter, "hybrid query was cancelled (partial={partial})")
}
Self::ReadCancelled { partial } => write!(
formatter,
"store close cancelled hybrid query (partial={partial})"
),
Self::LegThreadStart { leg, detail } => {
write!(formatter, "{leg:?} leg thread did not start: {detail}")
}
Self::LegPanic { leg, detail } => {
write!(formatter, "{leg:?} leg panic was contained: {detail}")
}
Self::Leg { leg, kind, detail } => {
write!(formatter, "{leg:?} leg failed ({kind:?}): {detail}")
}
}
}
}
impl std::error::Error for FusionError {}
impl From<crate::lifecycle::QueryError> for FusionError {
fn from(error: crate::lifecycle::QueryError) -> Self {
match error {
crate::lifecycle::QueryError::Timeout { .. } => Self::Timeout { partial: false },
crate::lifecycle::QueryError::Cancelled { .. } => Self::Cancelled { partial: false },
crate::lifecycle::QueryError::ReadCancelled { .. } => {
Self::ReadCancelled { partial: false }
}
crate::lifecycle::QueryError::Scan(crate::scan::ScanError::Timeout { .. }) => {
Self::Timeout { partial: false }
}
crate::lifecycle::QueryError::Scan(crate::scan::ScanError::Cancelled { .. }) => {
Self::Cancelled { partial: false }
}
crate::lifecycle::QueryError::Scan(crate::scan::ScanError::ReadCancelled {
..
}) => Self::ReadCancelled { partial: false },
other => {
let kind = match &other {
crate::lifecycle::QueryError::Store(error) => {
LegFailureKind::Store(error.kind())
}
crate::lifecycle::QueryError::Scan(_) => LegFailureKind::Scan,
crate::lifecycle::QueryError::Graph(_) => LegFailureKind::Graph,
crate::lifecycle::QueryError::Timeout { .. }
| crate::lifecycle::QueryError::Cancelled { .. }
| crate::lifecycle::QueryError::ReadCancelled { .. } => LegFailureKind::Caller,
};
Self::Leg {
leg: FusionLeg::Vector,
kind,
detail: other.to_string(),
}
}
}
}
}
#[derive(Clone, Debug)]
pub(crate) struct JoinedVector<K> {
pub(crate) key: K,
pub(crate) squared_l2: f64,
}
#[derive(Clone, Debug)]
pub(crate) struct JoinedLexical<K> {
pub(crate) key: K,
pub(crate) bm25: f64,
}
pub fn execute_hybrid<K, V, L, VectorRun, LexicalRun, VectorJoin, LexicalJoin>(
query: &HybridQuery,
vector_run: VectorRun,
lexical_run: LexicalRun,
vector_join: VectorJoin,
lexical_join: LexicalJoin,
) -> Result<FusionOutcome<K>, FusionError>
where
K: Clone + Ord,
VectorRun: FnOnce() -> Result<Vec<VectorCandidate<V>>, FusionError>,
LexicalRun: FnOnce() -> Result<Vec<LexicalCandidate<L>>, FusionError>,
VectorJoin: FnMut(&V) -> Option<K>,
LexicalJoin: FnMut(&L) -> Option<K>,
{
let vector = vector_run()?;
let lexical = lexical_run()?;
fuse(query, &vector, &lexical, vector_join, lexical_join)
}
pub fn fuse<K, V, L, VectorJoin, LexicalJoin>(
query: &HybridQuery,
vector: &[VectorCandidate<V>],
lexical: &[LexicalCandidate<L>],
mut vector_join: VectorJoin,
mut lexical_join: LexicalJoin,
) -> Result<FusionOutcome<K>, FusionError>
where
K: Clone + Ord,
VectorJoin: FnMut(&V) -> Option<K>,
LexicalJoin: FnMut(&L) -> Option<K>,
{
let (effective_alpha, applied_rules) = rules::effective_alpha(query)?;
let joined_vector = join_vector(vector, &mut vector_join)?;
let joined_lexical = join_lexical(lexical, &mut lexical_join)?;
cc::fuse_joined(
query,
&joined_vector,
&joined_lexical,
effective_alpha,
applied_rules,
)
}
pub fn fuse_bounded<K, V, L, VectorJoin, LexicalJoin>(
query: &HybridQuery,
vector_window: &[VectorCandidate<V>],
lexical_window: &[LexicalCandidate<L>],
bounds: LegBounds,
mut vector_join: VectorJoin,
mut lexical_join: LexicalJoin,
) -> Result<FusionOutcome<K>, FusionError>
where
K: Clone + Ord,
VectorJoin: FnMut(&V) -> Option<K>,
LexicalJoin: FnMut(&L) -> Option<K>,
{
let (effective_alpha, applied_rules) = rules::effective_alpha(query)?;
let joined_vector = join_vector(vector_window, &mut vector_join)?;
let joined_lexical = join_lexical(lexical_window, &mut lexical_join)?;
cc::fuse_bounded_joined(
query,
&joined_vector,
&joined_lexical,
bounds,
effective_alpha,
applied_rules,
)
}
pub(crate) fn fuse_store_bounded<K, V, L, VectorJoin, LexicalJoin>(
query: &HybridQuery,
vector_window: &[VectorCandidate<V>],
lexical_window: &[LexicalCandidate<L>],
bounds: LegBounds,
mut vector_join: VectorJoin,
mut lexical_join: LexicalJoin,
scratch: &mut StoreFusionScratch<K>,
) -> Result<FusionOutcome<K>, FusionError>
where
K: Clone + Ord,
VectorJoin: FnMut(&V) -> Option<K>,
LexicalJoin: FnMut(&L) -> Option<K>,
{
let (alpha, rules) = rules::effective_alpha(query)?;
let vector = join_vector(vector_window, &mut vector_join)?;
let lexical = join_lexical(lexical_window, &mut lexical_join)?;
cc::fuse_store_bounded_joined(query, &vector, &lexical, bounds, alpha, rules, scratch)
}
fn join_vector<K, V, Join>(
candidates: &[VectorCandidate<V>],
join: &mut Join,
) -> Result<Vec<JoinedVector<K>>, FusionError>
where
K: Clone + Ord,
Join: FnMut(&V) -> Option<K>,
{
let mut joined = Vec::with_capacity(candidates.len());
let mut identities = BTreeSet::new();
let mut previous = None;
for (rank, candidate) in candidates.iter().enumerate() {
if candidate.precision != ScorePrecision::Exact {
return Err(FusionError::EstimatedVectorScore { rank });
}
if !candidate.squared_l2.is_finite() {
return Err(FusionError::NonFiniteScore {
leg: FusionLeg::Vector,
rank,
});
}
if candidate.squared_l2 < 0.0 {
return Err(FusionError::NegativeScore {
leg: FusionLeg::Vector,
rank,
});
}
if previous.is_some_and(|score| candidate.squared_l2 < score) {
return Err(FusionError::UnrankedInput {
leg: FusionLeg::Vector,
rank,
});
}
previous = Some(candidate.squared_l2);
let key = join(&candidate.id).ok_or(FusionError::MissingDocumentIdentity {
leg: FusionLeg::Vector,
rank,
})?;
if !identities.insert(key.clone()) {
return Err(FusionError::DuplicateDocumentIdentity {
leg: FusionLeg::Vector,
rank,
});
}
joined.push(JoinedVector {
key,
squared_l2: candidate.squared_l2,
});
}
joined.sort_by(|left, right| {
left.squared_l2
.total_cmp(&right.squared_l2)
.then_with(|| left.key.cmp(&right.key))
});
Ok(joined)
}
fn join_lexical<K, L, Join>(
candidates: &[LexicalCandidate<L>],
join: &mut Join,
) -> Result<Vec<JoinedLexical<K>>, FusionError>
where
K: Clone + Ord,
Join: FnMut(&L) -> Option<K>,
{
let mut joined = Vec::with_capacity(candidates.len());
let mut identities = BTreeSet::new();
let mut previous = None;
for (rank, candidate) in candidates.iter().enumerate() {
if !candidate.bm25.is_finite() {
return Err(FusionError::NonFiniteScore {
leg: FusionLeg::Lexical,
rank,
});
}
if candidate.bm25 < 0.0 {
return Err(FusionError::NegativeScore {
leg: FusionLeg::Lexical,
rank,
});
}
if previous.is_some_and(|score| candidate.bm25 > score) {
return Err(FusionError::UnrankedInput {
leg: FusionLeg::Lexical,
rank,
});
}
previous = Some(candidate.bm25);
let key = join(&candidate.id).ok_or(FusionError::MissingDocumentIdentity {
leg: FusionLeg::Lexical,
rank,
})?;
if !identities.insert(key.clone()) {
return Err(FusionError::DuplicateDocumentIdentity {
leg: FusionLeg::Lexical,
rank,
});
}
joined.push(JoinedLexical {
key,
bm25: candidate.bm25,
});
}
joined.sort_by(|left, right| {
right
.bm25
.total_cmp(&left.bm25)
.then_with(|| left.key.cmp(&right.key))
});
Ok(joined)
}