1pub const MAX_FINAL_LIMIT: usize = 10_000;
11pub const MAX_QUERY_OFFSET: usize = 100_000;
12pub const MAX_RETRIEVER_K: usize = 100_000;
13pub const MAX_RETRIEVERS: usize = 32;
14pub const MAX_SPARSE_TERMS: usize = 65_536;
15pub const MAX_SET_MEMBERS: usize = 65_536;
16pub const MAX_SET_MEMBER_BYTES: usize = 1024 * 1024;
17pub const MAX_SET_INPUT_BYTES: usize = 16 * 1024 * 1024;
18pub const MAX_PROJECTION_COLUMNS: usize = 4_096;
19pub const MAX_HARD_CONDITIONS: usize = 256;
20pub const MAX_RETRIEVER_WEIGHT: f64 = 1_000_000.0;
21pub const MAX_FUSED_CANDIDATES: usize = 250_000;
22pub const MAX_RAW_INDEX_CANDIDATES: usize = 250_000;
23pub const MAX_RETRIEVER_NAME_BYTES: usize = 128;
24pub(crate) const VECTOR_WORK_QUANTUM: usize = 64;
25pub(crate) const HAMMING_WORK_QUANTUM: usize = 64;
26pub(crate) const PARSE_WORK_QUANTUM: usize = 64;
27
28pub(crate) fn work_units(items: usize, quantum: usize) -> usize {
29 items.div_ceil(quantum).max(1)
30}
31
32#[derive(Debug, Clone)]
35pub struct AiExecutionContext {
36 deadline: Option<std::time::Instant>,
37 query_time_nanos: i64,
38 initial_work: usize,
39 max_fused_candidates: usize,
40 remaining_work: std::sync::Arc<std::sync::atomic::AtomicUsize>,
41 cancelled: std::sync::Arc<std::sync::atomic::AtomicBool>,
42}
43
44impl AiExecutionContext {
45 pub fn new(deadline: Option<std::time::Instant>, work_budget: usize) -> Self {
46 Self {
47 deadline,
48 query_time_nanos: std::time::SystemTime::now()
49 .duration_since(std::time::UNIX_EPOCH)
50 .map(|duration| duration.as_nanos().min(i64::MAX as u128) as i64)
51 .unwrap_or(0),
52 initial_work: work_budget,
53 max_fused_candidates: MAX_FUSED_CANDIDATES,
54 remaining_work: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(work_budget)),
55 cancelled: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
56 }
57 }
58
59 pub fn with_timeout(timeout: std::time::Duration, work_budget: usize) -> Self {
60 Self::new(Some(std::time::Instant::now() + timeout), work_budget)
61 }
62
63 pub fn with_limits(
64 timeout: std::time::Duration,
65 work_budget: usize,
66 max_fused_candidates: usize,
67 ) -> Self {
68 let mut context = Self::with_timeout(timeout, work_budget);
69 context.max_fused_candidates = max_fused_candidates.min(MAX_FUSED_CANDIDATES);
70 context
71 }
72
73 pub fn cancel(&self) {
74 self.cancelled
75 .store(true, std::sync::atomic::Ordering::Release);
76 }
77
78 pub fn checkpoint(&self) -> crate::Result<()> {
79 if self.cancelled.load(std::sync::atomic::Ordering::Acquire) {
80 return Err(crate::MongrelError::Cancelled);
81 }
82 if self
83 .deadline
84 .is_some_and(|deadline| std::time::Instant::now() >= deadline)
85 {
86 return Err(crate::MongrelError::DeadlineExceeded);
87 }
88 Ok(())
89 }
90
91 pub fn consume(&self, work: usize) -> crate::Result<()> {
92 self.checkpoint()?;
93 let result = self.remaining_work.fetch_update(
94 std::sync::atomic::Ordering::AcqRel,
95 std::sync::atomic::Ordering::Acquire,
96 |remaining| remaining.checked_sub(work),
97 );
98 if result.is_err() {
99 return Err(crate::MongrelError::WorkBudgetExceeded);
100 }
101 self.checkpoint()
102 }
103
104 pub fn consumed_work(&self) -> usize {
105 self.initial_work.saturating_sub(
106 self.remaining_work
107 .load(std::sync::atomic::Ordering::Acquire),
108 )
109 }
110
111 pub fn work_limit(&self) -> usize {
112 self.initial_work
113 }
114
115 pub fn remaining_duration(&self) -> Option<std::time::Duration> {
116 self.deadline
117 .map(|deadline| deadline.saturating_duration_since(std::time::Instant::now()))
118 }
119
120 pub fn query_time_nanos(&self) -> i64 {
121 self.query_time_nanos
122 }
123
124 pub fn max_fused_candidates(&self) -> usize {
125 self.max_fused_candidates
126 }
127}
128
129#[derive(Debug, Clone)]
131pub enum Condition {
132 Pk(Vec<u8>),
134 BitmapEq { column_id: u16, value: Vec<u8> },
136 BitmapIn {
141 column_id: u16,
142 values: Vec<Vec<u8>>,
143 },
144 BytesPrefix { column_id: u16, prefix: Vec<u8> },
149 Ann {
151 column_id: u16,
152 query: Vec<f32>,
153 k: usize,
154 },
155 FmContains { column_id: u16, pattern: Vec<u8> },
157 FmContainsAll {
162 column_id: u16,
163 patterns: Vec<Vec<u8>>,
164 },
165 Range { column_id: u16, lo: i64, hi: i64 },
169 RangeF64 {
172 column_id: u16,
173 lo: f64,
174 lo_inclusive: bool,
175 hi: f64,
176 hi_inclusive: bool,
177 },
178 SparseMatch {
181 column_id: u16,
182 query: Vec<(u32, f32)>,
183 k: usize,
184 },
185 MinHashSimilar {
189 column_id: u16,
190 query: Vec<u64>,
191 k: usize,
192 },
193 IsNull { column_id: u16 },
197 IsNotNull { column_id: u16 },
200}
201
202#[derive(Debug, Clone)]
205pub enum Retriever {
206 Ann {
207 column_id: u16,
208 query: Vec<f32>,
209 k: usize,
210 },
211 Sparse {
212 column_id: u16,
213 query: Vec<(u32, f32)>,
214 k: usize,
215 },
216 MinHash {
217 column_id: u16,
218 members: Vec<SetMember>,
219 k: usize,
220 },
221}
222
223impl Retriever {
224 pub fn column_id(&self) -> u16 {
225 match self {
226 Self::Ann { column_id, .. }
227 | Self::Sparse { column_id, .. }
228 | Self::MinHash { column_id, .. } => *column_id,
229 }
230 }
231}
232
233pub fn encode_sparse_vector(terms: &[(u32, f32)]) -> crate::Result<Vec<u8>> {
234 Ok(bincode::serialize(terms)?)
235}
236
237#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
238#[serde(untagged)]
239pub enum SetMember {
240 String(String),
241 Number(serde_json::Number),
242 Boolean(bool),
243}
244
245impl SetMember {
246 pub(crate) fn encoded_len(&self) -> usize {
247 match self {
248 Self::String(value) => value.len(),
249 Self::Number(value) => value.to_string().len(),
250 Self::Boolean(_) => 1,
251 }
252 }
253
254 pub fn hash_v1(&self) -> u64 {
255 let value = match self {
256 Self::String(value) => serde_json::Value::String(value.clone()),
257 Self::Number(value) => serde_json::Value::Number(value.clone()),
258 Self::Boolean(value) => serde_json::Value::Bool(*value),
259 };
260 crate::index::minhash_member_hash_v1(&value).expect("SetMember is always scalar")
261 }
262}
263
264#[derive(Debug, Clone, Copy, PartialEq)]
265pub enum RetrieverScore {
266 AnnHammingDistance(u32),
267 SparseDotProduct(f64),
268 MinHashEstimatedJaccard(f32),
269}
270
271#[derive(Debug, Clone, Copy, PartialEq, Eq)]
272pub enum VectorMetric {
273 Cosine,
274 DotProduct,
275 Euclidean,
276}
277
278#[derive(Debug, Clone)]
279pub struct AnnRerankRequest {
280 pub column_id: u16,
281 pub query: Vec<f32>,
282 pub candidate_k: usize,
283 pub limit: usize,
284 pub metric: VectorMetric,
285}
286
287#[derive(Debug, Clone, Copy, PartialEq)]
288pub struct AnnRerankHit {
289 pub row_id: crate::RowId,
290 pub hamming_distance: u32,
291 pub exact_score: f32,
292}
293
294#[derive(Debug, Clone, PartialEq)]
295pub struct RetrieverHit {
296 pub row_id: crate::RowId,
297 pub rank: usize,
299 pub score: RetrieverScore,
300}
301
302#[derive(Debug, Clone)]
303pub struct SetSimilarityRequest {
304 pub column_id: u16,
305 pub members: Vec<SetMember>,
306 pub candidate_k: usize,
307 pub min_jaccard: f32,
308 pub limit: usize,
309}
310
311#[derive(Debug, Clone, PartialEq)]
312pub struct SetSimilarityHit {
313 pub row_id: crate::RowId,
314 pub estimated_jaccard: f32,
315 pub exact_jaccard: f32,
316}
317
318#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
319pub struct SetSimilarityTrace {
320 pub candidate_generation_us: u64,
321 pub gather_us: u64,
322 pub parse_us: u64,
323 pub score_us: u64,
324 pub candidate_count: usize,
325 pub verified_count: usize,
326}
327
328#[derive(Debug, Clone)]
329pub struct SearchRequest {
330 pub must: Vec<Condition>,
331 pub retrievers: Vec<NamedRetriever>,
332 pub fusion: Fusion,
333 pub rerank: Option<Rerank>,
334 pub limit: usize,
335 pub projection: Option<Vec<u16>>,
336}
337
338#[derive(Debug, Clone, Copy)]
339pub struct SearchAfter {
340 pub final_score: f64,
341 pub row_id: crate::RowId,
342}
343
344#[derive(Debug, Clone)]
345pub enum Rerank {
346 ExactVector {
347 embedding_column: u16,
348 query: Vec<f32>,
349 metric: VectorMetric,
350 candidate_limit: usize,
351 weight: f64,
352 },
353}
354
355#[derive(Debug, Clone)]
356pub struct NamedRetriever {
357 pub name: String,
358 pub weight: f64,
359 pub retriever: Retriever,
360}
361
362#[derive(Debug, Clone, Copy)]
363pub enum Fusion {
364 ReciprocalRank { constant: u32 },
365}
366
367#[derive(Debug, Clone, PartialEq)]
368pub struct ComponentScore {
369 pub retriever_name: std::sync::Arc<str>,
370 pub rank: usize,
371 pub raw_score: RetrieverScore,
372 pub contribution: f64,
373}
374
375#[derive(Debug, Clone, PartialEq)]
376pub struct SearchHit {
377 pub row_id: crate::RowId,
378 pub cells: Vec<(u16, crate::Value)>,
379 pub components: Vec<ComponentScore>,
380 pub fused_score: f64,
381 pub exact_rerank_score: Option<f32>,
382 pub final_score: f64,
383 pub final_rank: usize,
385}
386
387#[derive(Debug, Clone, Default)]
389pub struct Query {
390 pub conditions: Vec<Condition>,
391 pub limit: Option<usize>,
392 pub offset: usize,
393}
394
395impl Query {
396 pub fn new() -> Self {
397 Self::default()
398 }
399 pub fn and(mut self, c: Condition) -> Self {
400 self.conditions.push(c);
401 self
402 }
403 pub fn with_limit(mut self, limit: usize) -> Self {
404 self.limit = Some(limit);
405 self
406 }
407 pub fn with_offset(mut self, offset: usize) -> Self {
408 self.offset = offset;
409 self
410 }
411 pub fn pk(key: Vec<u8>) -> Self {
412 Self::new().and(Condition::Pk(key))
413 }
414}
415
416pub fn canonical_query_key(
425 conditions: &[Condition],
426 projection: Option<&[u16]>,
427 epoch: u64,
428) -> u64 {
429 canonical_query_key_with_version(QUERY_CACHE_KEY_VERSION, conditions, projection, epoch)
430}
431
432const QUERY_CACHE_KEY_VERSION: u64 = 2;
433
434fn canonical_query_key_with_version(
435 version: u64,
436 conditions: &[Condition],
437 projection: Option<&[u16]>,
438 epoch: u64,
439) -> u64 {
440 let fold = |seed: u64, b: u64| -> u64 { seed.wrapping_mul(0x9E3779B97F4A7C15).wrapping_add(b) };
441 let mut acc = fold(0xA5A5_A5A5_A5A5_A5A5, version);
442 acc = fold(acc, epoch);
443 let mut digests: Vec<u64> = conditions.iter().map(hash_condition).collect();
445 digests.sort_unstable();
446 let n = digests.len() as u64;
447 acc = fold(acc, n);
448 for d in digests {
449 acc = fold(acc, d);
450 }
451 match projection {
454 Some(p) => {
455 let mut p = p.to_vec();
456 p.sort_unstable();
457 p.dedup();
458 acc = fold(acc, 0x5E);
459 acc = fold(acc, p.len() as u64);
460 for id in p {
461 acc = fold(acc, id as u64);
462 }
463 }
464 None => {
465 acc = fold(acc, 0xA5);
466 }
467 }
468 acc
469}
470
471fn hash_condition(c: &Condition) -> u64 {
475 use std::hash::{Hash, Hasher};
476 let mut h = std::collections::hash_map::DefaultHasher::new();
477 match c {
478 Condition::Pk(k) => {
479 0u8.hash(&mut h);
480 k.hash(&mut h);
481 }
482 Condition::BitmapEq { column_id, value } => {
483 1u8.hash(&mut h);
484 column_id.hash(&mut h);
485 value.hash(&mut h);
486 }
487 Condition::BitmapIn { column_id, values } => {
488 2u8.hash(&mut h);
489 column_id.hash(&mut h);
490 let mut v: Vec<&Vec<u8>> = values.iter().collect();
491 v.sort();
492 v.dedup();
493 v.len().hash(&mut h);
494 for b in v {
495 b.hash(&mut h);
496 }
497 }
498 Condition::Ann {
499 column_id,
500 query,
501 k,
502 } => {
503 3u8.hash(&mut h);
504 column_id.hash(&mut h);
505 k.hash(&mut h);
506 for f in query {
507 f.to_bits().hash(&mut h);
508 }
509 }
510 Condition::FmContains { column_id, pattern } => {
511 4u8.hash(&mut h);
512 column_id.hash(&mut h);
513 pattern.hash(&mut h);
514 }
515 Condition::FmContainsAll {
516 column_id,
517 patterns,
518 } => {
519 5u8.hash(&mut h);
520 column_id.hash(&mut h);
521 let mut sorted: Vec<&[u8]> = patterns.iter().map(|p| p.as_slice()).collect();
522 sorted.sort();
523 sorted.dedup();
524 sorted.len().hash(&mut h);
525 for p in sorted {
526 p.hash(&mut h);
527 }
528 }
529 Condition::Range { column_id, lo, hi } => {
530 6u8.hash(&mut h);
531 column_id.hash(&mut h);
532 lo.hash(&mut h);
533 hi.hash(&mut h);
534 }
535 Condition::RangeF64 {
536 column_id,
537 lo,
538 lo_inclusive,
539 hi,
540 hi_inclusive,
541 } => {
542 7u8.hash(&mut h);
543 column_id.hash(&mut h);
544 lo.to_bits().hash(&mut h);
545 lo_inclusive.hash(&mut h);
546 hi.to_bits().hash(&mut h);
547 hi_inclusive.hash(&mut h);
548 }
549 Condition::SparseMatch {
550 column_id,
551 query,
552 k,
553 } => {
554 8u8.hash(&mut h);
555 column_id.hash(&mut h);
556 k.hash(&mut h);
557 let mut q: Vec<(u32, u32)> = query.iter().map(|(t, w)| (*t, w.to_bits())).collect();
558 q.sort_by_key(|(t, _)| *t);
559 for (t, wb) in q {
560 t.hash(&mut h);
561 wb.hash(&mut h);
562 }
563 }
564 Condition::MinHashSimilar {
565 column_id,
566 query,
567 k,
568 } => {
569 9u8.hash(&mut h);
570 column_id.hash(&mut h);
571 k.hash(&mut h);
572 let mut q = query.clone();
573 q.sort_unstable();
574 q.dedup();
575 for t in q {
576 t.hash(&mut h);
577 }
578 }
579 Condition::IsNull { column_id } => {
580 10u8.hash(&mut h);
581 column_id.hash(&mut h);
582 }
583 Condition::IsNotNull { column_id } => {
584 11u8.hash(&mut h);
585 column_id.hash(&mut h);
586 }
587 Condition::BytesPrefix { column_id, prefix } => {
588 12u8.hash(&mut h);
589 column_id.hash(&mut h);
590 prefix.hash(&mut h);
591 }
592 }
593 h.finish()
594}
595
596pub fn condition_columns(conditions: &[Condition]) -> Vec<u16> {
601 let mut cols: Vec<u16> = conditions
602 .iter()
603 .filter_map(|c| match c {
604 Condition::Pk(_) => None,
605 Condition::BitmapEq { column_id, .. }
606 | Condition::BitmapIn { column_id, .. }
607 | Condition::BytesPrefix { column_id, .. }
608 | Condition::Ann { column_id, .. }
609 | Condition::FmContains { column_id, .. }
610 | Condition::FmContainsAll { column_id, .. }
611 | Condition::Range { column_id, .. }
612 | Condition::RangeF64 { column_id, .. }
613 | Condition::SparseMatch { column_id, .. }
614 | Condition::MinHashSimilar { column_id, .. }
615 | Condition::IsNull { column_id }
616 | Condition::IsNotNull { column_id } => Some(*column_id),
617 })
618 .collect();
619 cols.sort_unstable();
620 cols.dedup();
621 cols
622}
623
624#[cfg(test)]
625mod tests {
626 use super::*;
627
628 #[test]
629 fn builder_chains() {
630 let q = Query::pk(b"k".to_vec()).and(Condition::Range {
631 column_id: 1,
632 lo: 0,
633 hi: 10,
634 });
635 assert_eq!(q.conditions.len(), 2);
636 }
637
638 #[test]
642 fn canonical_key_is_order_independent() {
643 let e = 7u64;
644 let a = Query::new()
645 .and(Condition::Range {
646 column_id: 1,
647 lo: 0,
648 hi: 10,
649 })
650 .and(Condition::BitmapEq {
651 column_id: 2,
652 value: b"x".to_vec(),
653 });
654 let b = Query::new()
655 .and(Condition::BitmapEq {
656 column_id: 2,
657 value: b"x".to_vec(),
658 })
659 .and(Condition::Range {
660 column_id: 1,
661 lo: 0,
662 hi: 10,
663 });
664 assert_eq!(
665 canonical_query_key(&a.conditions, None, e),
666 canonical_query_key(&b.conditions, None, e),
667 "condition order must not affect the key"
668 );
669
670 let minhash = Condition::MinHashSimilar {
671 column_id: 4,
672 query: vec![3, 1, 1, 2],
673 k: 5,
674 };
675 let minhash_canonical = Condition::MinHashSimilar {
676 column_id: 4,
677 query: vec![1, 2, 3],
678 k: 5,
679 };
680 assert_eq!(
681 canonical_query_key(std::slice::from_ref(&minhash), None, e),
682 canonical_query_key(&[minhash_canonical], None, e),
683 );
684
685 let fm = Condition::FmContainsAll {
686 column_id: 4,
687 patterns: vec![b"b".to_vec(), b"a".to_vec(), b"a".to_vec()],
688 };
689 let fm_canonical = Condition::FmContainsAll {
690 column_id: 4,
691 patterns: vec![b"a".to_vec(), b"b".to_vec()],
692 };
693 assert_eq!(
694 canonical_query_key(std::slice::from_ref(&fm), None, e),
695 canonical_query_key(&[fm_canonical], None, e),
696 );
697 assert_ne!(
698 canonical_query_key(std::slice::from_ref(&fm), None, e),
699 canonical_query_key(std::slice::from_ref(&minhash), None, e),
700 );
701 assert_ne!(
702 canonical_query_key_with_version(1, &a.conditions, None, e),
703 canonical_query_key_with_version(2, &a.conditions, None, e),
704 );
705
706 let ordered = Condition::BitmapIn {
708 column_id: 3,
709 values: vec![b"a".to_vec(), b"b".to_vec(), b"c".to_vec()],
710 };
711 let shuffled = Condition::BitmapIn {
712 column_id: 3,
713 values: vec![b"c".to_vec(), b"a".to_vec(), b"a".to_vec(), b"b".to_vec()],
714 };
715 assert_eq!(
716 canonical_query_key(std::slice::from_ref(&ordered), None, e),
717 canonical_query_key(&[shuffled], None, e),
718 "BitmapIn values must dedup+sort"
719 );
720
721 assert_ne!(
723 canonical_query_key(&a.conditions, None, e),
724 canonical_query_key(&a.conditions, None, e + 1),
725 "epoch must fold into the key"
726 );
727
728 let proj = vec![1u16, 2];
730 assert_ne!(
731 canonical_query_key(&a.conditions, None, e),
732 canonical_query_key(&a.conditions, Some(&proj), e),
733 "None projection must differ from an explicit projection"
734 );
735 let proj_rev = vec![2u16, 1];
737 assert_eq!(
738 canonical_query_key(&a.conditions, Some(&proj), e),
739 canonical_query_key(&a.conditions, Some(&proj_rev), e),
740 );
741
742 let budget = AiExecutionContext::new(None, 2);
743 budget.consume(1).unwrap();
744 assert!(matches!(
745 budget.consume(2),
746 Err(crate::MongrelError::WorkBudgetExceeded)
747 ));
748 budget.cancel();
749 assert!(matches!(
750 budget.checkpoint(),
751 Err(crate::MongrelError::Cancelled)
752 ));
753
754 let expired = AiExecutionContext::new(
755 Some(std::time::Instant::now() - std::time::Duration::from_millis(1)),
756 1,
757 );
758 assert!(matches!(
759 expired.checkpoint(),
760 Err(crate::MongrelError::DeadlineExceeded)
761 ));
762 }
763}