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 control: crate::ExecutionControl,
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}
42
43impl AiExecutionContext {
44 pub fn new(deadline: Option<std::time::Instant>, work_budget: usize) -> Self {
45 Self {
46 control: crate::ExecutionControl::new(deadline),
47 query_time_nanos: std::time::SystemTime::now()
48 .duration_since(std::time::UNIX_EPOCH)
49 .map(|duration| duration.as_nanos().min(i64::MAX as u128) as i64)
50 .unwrap_or(0),
51 initial_work: work_budget,
52 max_fused_candidates: MAX_FUSED_CANDIDATES,
53 remaining_work: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(work_budget)),
54 }
55 }
56
57 pub fn with_timeout(timeout: std::time::Duration, work_budget: usize) -> Self {
58 Self::new(Some(std::time::Instant::now() + timeout), work_budget)
59 }
60
61 pub fn with_limits(
62 timeout: std::time::Duration,
63 work_budget: usize,
64 max_fused_candidates: usize,
65 ) -> Self {
66 let mut context = Self::with_timeout(timeout, work_budget);
67 context.max_fused_candidates = max_fused_candidates.min(MAX_FUSED_CANDIDATES);
68 context
69 }
70
71 pub fn with_control(
72 control: crate::ExecutionControl,
73 work_budget: usize,
74 max_fused_candidates: usize,
75 ) -> Self {
76 Self {
77 control,
78 query_time_nanos: std::time::SystemTime::now()
79 .duration_since(std::time::UNIX_EPOCH)
80 .map(|duration| duration.as_nanos().min(i64::MAX as u128) as i64)
81 .unwrap_or(0),
82 initial_work: work_budget,
83 max_fused_candidates: max_fused_candidates.min(MAX_FUSED_CANDIDATES),
84 remaining_work: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(work_budget)),
85 }
86 }
87
88 pub fn cancel(&self) {
89 self.control
90 .cancel(crate::CancellationReason::ClientRequest);
91 }
92
93 pub fn checkpoint(&self) -> crate::Result<()> {
94 self.control.checkpoint()
95 }
96
97 pub fn consume(&self, work: usize) -> crate::Result<()> {
98 self.checkpoint()?;
99 let result = self.remaining_work.fetch_update(
100 std::sync::atomic::Ordering::AcqRel,
101 std::sync::atomic::Ordering::Acquire,
102 |remaining| remaining.checked_sub(work),
103 );
104 if result.is_err() {
105 return Err(crate::MongrelError::WorkBudgetExceeded);
106 }
107 self.checkpoint()
108 }
109
110 pub fn consumed_work(&self) -> usize {
111 self.initial_work.saturating_sub(
112 self.remaining_work
113 .load(std::sync::atomic::Ordering::Acquire),
114 )
115 }
116
117 pub fn work_limit(&self) -> usize {
118 self.initial_work
119 }
120
121 pub fn remaining_duration(&self) -> Option<std::time::Duration> {
122 self.control.remaining_duration()
123 }
124
125 pub fn execution_control(&self) -> &crate::ExecutionControl {
126 &self.control
127 }
128
129 pub fn query_time_nanos(&self) -> i64 {
130 self.query_time_nanos
131 }
132
133 pub fn with_query_time_nanos(&self, query_time_nanos: i64) -> Self {
134 let mut context = self.clone();
135 context.query_time_nanos = query_time_nanos;
136 context
137 }
138
139 pub fn max_fused_candidates(&self) -> usize {
140 self.max_fused_candidates
141 }
142}
143
144#[derive(Debug, Clone)]
146pub enum Condition {
147 Pk(Vec<u8>),
149 BitmapEq { column_id: u16, value: Vec<u8> },
151 BitmapIn {
156 column_id: u16,
157 values: Vec<Vec<u8>>,
158 },
159 BytesPrefix { column_id: u16, prefix: Vec<u8> },
164 Ann {
166 column_id: u16,
167 query: Vec<f32>,
168 k: usize,
169 },
170 FmContains { column_id: u16, pattern: Vec<u8> },
172 FmContainsAll {
177 column_id: u16,
178 patterns: Vec<Vec<u8>>,
179 },
180 Range { column_id: u16, lo: i64, hi: i64 },
184 RangeF64 {
187 column_id: u16,
188 lo: f64,
189 lo_inclusive: bool,
190 hi: f64,
191 hi_inclusive: bool,
192 },
193 SparseMatch {
196 column_id: u16,
197 query: Vec<(u32, f32)>,
198 k: usize,
199 },
200 MinHashSimilar {
204 column_id: u16,
205 query: Vec<u64>,
206 k: usize,
207 },
208 IsNull { column_id: u16 },
212 IsNotNull { column_id: u16 },
215}
216
217#[derive(Debug, Clone)]
220pub enum Retriever {
221 Ann {
222 column_id: u16,
223 query: Vec<f32>,
224 k: usize,
225 },
226 Sparse {
227 column_id: u16,
228 query: Vec<(u32, f32)>,
229 k: usize,
230 },
231 MinHash {
232 column_id: u16,
233 members: Vec<SetMember>,
234 k: usize,
235 },
236}
237
238impl Retriever {
239 pub fn column_id(&self) -> u16 {
240 match self {
241 Self::Ann { column_id, .. }
242 | Self::Sparse { column_id, .. }
243 | Self::MinHash { column_id, .. } => *column_id,
244 }
245 }
246}
247
248pub fn encode_sparse_vector(terms: &[(u32, f32)]) -> crate::Result<Vec<u8>> {
249 Ok(bincode::serialize(terms)?)
250}
251
252#[derive(Debug, Clone, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
253#[serde(untagged)]
254pub enum SetMember {
255 String(String),
256 Number(serde_json::Number),
257 Boolean(bool),
258}
259
260impl SetMember {
261 pub(crate) fn encoded_len(&self) -> usize {
262 match self {
263 Self::String(value) => value.len(),
264 Self::Number(value) => value.to_string().len(),
265 Self::Boolean(_) => 1,
266 }
267 }
268
269 pub fn hash_v1(&self) -> u64 {
270 let value = match self {
271 Self::String(value) => serde_json::Value::String(value.clone()),
272 Self::Number(value) => serde_json::Value::Number(value.clone()),
273 Self::Boolean(value) => serde_json::Value::Bool(*value),
274 };
275 crate::index::minhash_member_hash_v1(&value).expect("SetMember is always scalar")
276 }
277}
278
279#[derive(Debug, Clone, Copy, PartialEq)]
280pub enum RetrieverScore {
281 AnnHammingDistance(u32),
282 SparseDotProduct(f64),
283 MinHashEstimatedJaccard(f32),
284}
285
286#[derive(Debug, Clone, Copy, PartialEq, Eq)]
287pub enum VectorMetric {
288 Cosine,
289 DotProduct,
290 Euclidean,
291}
292
293#[derive(Debug, Clone)]
294pub struct AnnRerankRequest {
295 pub column_id: u16,
296 pub query: Vec<f32>,
297 pub candidate_k: usize,
298 pub limit: usize,
299 pub metric: VectorMetric,
300}
301
302#[derive(Debug, Clone, Copy, PartialEq)]
303pub struct AnnRerankHit {
304 pub row_id: crate::RowId,
305 pub hamming_distance: u32,
306 pub exact_score: f32,
307}
308
309#[derive(Debug, Clone, PartialEq)]
310pub struct RetrieverHit {
311 pub row_id: crate::RowId,
312 pub rank: usize,
314 pub score: RetrieverScore,
315}
316
317#[derive(Debug, Clone)]
318pub struct SetSimilarityRequest {
319 pub column_id: u16,
320 pub members: Vec<SetMember>,
321 pub candidate_k: usize,
322 pub min_jaccard: f32,
323 pub limit: usize,
324}
325
326#[derive(Debug, Clone, PartialEq)]
327pub struct SetSimilarityHit {
328 pub row_id: crate::RowId,
329 pub estimated_jaccard: f32,
330 pub exact_jaccard: f32,
331}
332
333#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
334pub struct SetSimilarityTrace {
335 pub candidate_generation_us: u64,
336 pub gather_us: u64,
337 pub parse_us: u64,
338 pub score_us: u64,
339 pub candidate_count: usize,
340 pub verified_count: usize,
341}
342
343#[derive(Debug, Clone)]
344pub struct SearchRequest {
345 pub must: Vec<Condition>,
346 pub retrievers: Vec<NamedRetriever>,
347 pub fusion: Fusion,
348 pub rerank: Option<Rerank>,
349 pub limit: usize,
350 pub projection: Option<Vec<u16>>,
351}
352
353#[derive(Debug, Clone, Copy)]
354pub struct SearchAfter {
355 pub final_score: f64,
356 pub row_id: crate::RowId,
357 pub returned_count: usize,
358}
359
360#[derive(Debug, Clone)]
361pub enum Rerank {
362 ExactVector {
363 embedding_column: u16,
364 query: Vec<f32>,
365 metric: VectorMetric,
366 candidate_limit: usize,
367 weight: f64,
368 },
369}
370
371#[derive(Debug, Clone)]
372pub struct NamedRetriever {
373 pub name: String,
374 pub weight: f64,
375 pub retriever: Retriever,
376}
377
378#[derive(Debug, Clone, Copy)]
379pub enum Fusion {
380 ReciprocalRank { constant: u32 },
381}
382
383#[derive(Debug, Clone, PartialEq)]
384pub struct ComponentScore {
385 pub retriever_name: std::sync::Arc<str>,
386 pub rank: usize,
387 pub raw_score: RetrieverScore,
388 pub contribution: f64,
389}
390
391#[derive(Debug, Clone, PartialEq)]
392pub struct SearchHit {
393 pub row_id: crate::RowId,
394 pub cells: Vec<(u16, crate::Value)>,
395 pub components: Vec<ComponentScore>,
396 pub fused_score: f64,
397 pub exact_rerank_score: Option<f32>,
398 pub final_score: f64,
399 pub final_rank: usize,
401}
402
403#[derive(Debug, Clone, Default)]
405pub struct Query {
406 pub conditions: Vec<Condition>,
407 pub limit: Option<usize>,
408 pub offset: usize,
409}
410
411impl Query {
412 pub fn new() -> Self {
413 Self::default()
414 }
415 pub fn and(mut self, c: Condition) -> Self {
416 self.conditions.push(c);
417 self
418 }
419 pub fn with_limit(mut self, limit: usize) -> Self {
420 self.limit = Some(limit);
421 self
422 }
423 pub fn with_offset(mut self, offset: usize) -> Self {
424 self.offset = offset;
425 self
426 }
427 pub fn pk(key: Vec<u8>) -> Self {
428 Self::new().and(Condition::Pk(key))
429 }
430}
431
432pub fn canonical_query_key(
441 conditions: &[Condition],
442 projection: Option<&[u16]>,
443 epoch: u64,
444) -> u64 {
445 canonical_query_key_with_version(QUERY_CACHE_KEY_VERSION, conditions, projection, epoch)
446}
447
448const QUERY_CACHE_KEY_VERSION: u64 = 2;
449
450fn canonical_query_key_with_version(
451 version: u64,
452 conditions: &[Condition],
453 projection: Option<&[u16]>,
454 epoch: u64,
455) -> u64 {
456 let fold = |seed: u64, b: u64| -> u64 { seed.wrapping_mul(0x9E3779B97F4A7C15).wrapping_add(b) };
457 let mut acc = fold(0xA5A5_A5A5_A5A5_A5A5, version);
458 acc = fold(acc, epoch);
459 let mut digests: Vec<u64> = conditions.iter().map(hash_condition).collect();
461 digests.sort_unstable();
462 let n = digests.len() as u64;
463 acc = fold(acc, n);
464 for d in digests {
465 acc = fold(acc, d);
466 }
467 match projection {
470 Some(p) => {
471 let mut p = p.to_vec();
472 p.sort_unstable();
473 p.dedup();
474 acc = fold(acc, 0x5E);
475 acc = fold(acc, p.len() as u64);
476 for id in p {
477 acc = fold(acc, id as u64);
478 }
479 }
480 None => {
481 acc = fold(acc, 0xA5);
482 }
483 }
484 acc
485}
486
487fn hash_condition(c: &Condition) -> u64 {
491 use std::hash::{Hash, Hasher};
492 let mut h = std::collections::hash_map::DefaultHasher::new();
493 match c {
494 Condition::Pk(k) => {
495 0u8.hash(&mut h);
496 k.hash(&mut h);
497 }
498 Condition::BitmapEq { column_id, value } => {
499 1u8.hash(&mut h);
500 column_id.hash(&mut h);
501 value.hash(&mut h);
502 }
503 Condition::BitmapIn { column_id, values } => {
504 2u8.hash(&mut h);
505 column_id.hash(&mut h);
506 let mut v: Vec<&Vec<u8>> = values.iter().collect();
507 v.sort();
508 v.dedup();
509 v.len().hash(&mut h);
510 for b in v {
511 b.hash(&mut h);
512 }
513 }
514 Condition::Ann {
515 column_id,
516 query,
517 k,
518 } => {
519 3u8.hash(&mut h);
520 column_id.hash(&mut h);
521 k.hash(&mut h);
522 for f in query {
523 f.to_bits().hash(&mut h);
524 }
525 }
526 Condition::FmContains { column_id, pattern } => {
527 4u8.hash(&mut h);
528 column_id.hash(&mut h);
529 pattern.hash(&mut h);
530 }
531 Condition::FmContainsAll {
532 column_id,
533 patterns,
534 } => {
535 5u8.hash(&mut h);
536 column_id.hash(&mut h);
537 let mut sorted: Vec<&[u8]> = patterns.iter().map(|p| p.as_slice()).collect();
538 sorted.sort();
539 sorted.dedup();
540 sorted.len().hash(&mut h);
541 for p in sorted {
542 p.hash(&mut h);
543 }
544 }
545 Condition::Range { column_id, lo, hi } => {
546 6u8.hash(&mut h);
547 column_id.hash(&mut h);
548 lo.hash(&mut h);
549 hi.hash(&mut h);
550 }
551 Condition::RangeF64 {
552 column_id,
553 lo,
554 lo_inclusive,
555 hi,
556 hi_inclusive,
557 } => {
558 7u8.hash(&mut h);
559 column_id.hash(&mut h);
560 lo.to_bits().hash(&mut h);
561 lo_inclusive.hash(&mut h);
562 hi.to_bits().hash(&mut h);
563 hi_inclusive.hash(&mut h);
564 }
565 Condition::SparseMatch {
566 column_id,
567 query,
568 k,
569 } => {
570 8u8.hash(&mut h);
571 column_id.hash(&mut h);
572 k.hash(&mut h);
573 let mut q: Vec<(u32, u32)> = query.iter().map(|(t, w)| (*t, w.to_bits())).collect();
574 q.sort_by_key(|(t, _)| *t);
575 for (t, wb) in q {
576 t.hash(&mut h);
577 wb.hash(&mut h);
578 }
579 }
580 Condition::MinHashSimilar {
581 column_id,
582 query,
583 k,
584 } => {
585 9u8.hash(&mut h);
586 column_id.hash(&mut h);
587 k.hash(&mut h);
588 let mut q = query.clone();
589 q.sort_unstable();
590 q.dedup();
591 for t in q {
592 t.hash(&mut h);
593 }
594 }
595 Condition::IsNull { column_id } => {
596 10u8.hash(&mut h);
597 column_id.hash(&mut h);
598 }
599 Condition::IsNotNull { column_id } => {
600 11u8.hash(&mut h);
601 column_id.hash(&mut h);
602 }
603 Condition::BytesPrefix { column_id, prefix } => {
604 12u8.hash(&mut h);
605 column_id.hash(&mut h);
606 prefix.hash(&mut h);
607 }
608 }
609 h.finish()
610}
611
612const CURSOR_REQUEST_HASH_VERSION: u64 = 1;
613
614fn cursor_u16(out: &mut Vec<u8>, value: u16) {
615 out.extend_from_slice(&value.to_le_bytes());
616}
617
618fn cursor_u32(out: &mut Vec<u8>, value: u32) {
619 out.extend_from_slice(&value.to_le_bytes());
620}
621
622fn cursor_u64(out: &mut Vec<u8>, value: u64) {
623 out.extend_from_slice(&value.to_le_bytes());
624}
625
626fn cursor_bytes(out: &mut Vec<u8>, value: &[u8]) {
627 cursor_u64(out, value.len() as u64);
628 out.extend_from_slice(value);
629}
630
631fn cursor_condition(condition: &Condition) -> Vec<u8> {
632 let mut out = Vec::new();
633 match condition {
634 Condition::Pk(value) => {
635 out.push(0);
636 cursor_bytes(&mut out, value);
637 }
638 Condition::BitmapEq { column_id, value } => {
639 out.push(1);
640 cursor_u16(&mut out, *column_id);
641 cursor_bytes(&mut out, value);
642 }
643 Condition::BitmapIn { column_id, values } => {
644 out.push(2);
645 cursor_u16(&mut out, *column_id);
646 let mut values = values.clone();
647 values.sort();
648 values.dedup();
649 cursor_u64(&mut out, values.len() as u64);
650 for value in values {
651 cursor_bytes(&mut out, &value);
652 }
653 }
654 Condition::Ann {
655 column_id,
656 query,
657 k,
658 } => {
659 out.push(3);
660 cursor_u16(&mut out, *column_id);
661 cursor_u64(&mut out, *k as u64);
662 cursor_u64(&mut out, query.len() as u64);
663 for value in query {
664 cursor_u32(&mut out, value.to_bits());
665 }
666 }
667 Condition::FmContains { column_id, pattern } => {
668 out.push(4);
669 cursor_u16(&mut out, *column_id);
670 cursor_bytes(&mut out, pattern);
671 }
672 Condition::FmContainsAll {
673 column_id,
674 patterns,
675 } => {
676 out.push(5);
677 cursor_u16(&mut out, *column_id);
678 let mut patterns = patterns.clone();
679 patterns.sort();
680 patterns.dedup();
681 cursor_u64(&mut out, patterns.len() as u64);
682 for pattern in patterns {
683 cursor_bytes(&mut out, &pattern);
684 }
685 }
686 Condition::Range { column_id, lo, hi } => {
687 out.push(6);
688 cursor_u16(&mut out, *column_id);
689 out.extend_from_slice(&lo.to_le_bytes());
690 out.extend_from_slice(&hi.to_le_bytes());
691 }
692 Condition::RangeF64 {
693 column_id,
694 lo,
695 lo_inclusive,
696 hi,
697 hi_inclusive,
698 } => {
699 out.push(7);
700 cursor_u16(&mut out, *column_id);
701 cursor_u64(&mut out, lo.to_bits());
702 out.push(u8::from(*lo_inclusive));
703 cursor_u64(&mut out, hi.to_bits());
704 out.push(u8::from(*hi_inclusive));
705 }
706 Condition::SparseMatch {
707 column_id,
708 query,
709 k,
710 } => {
711 out.push(8);
712 cursor_u16(&mut out, *column_id);
713 cursor_u64(&mut out, *k as u64);
714 let mut query: Vec<_> = query
715 .iter()
716 .map(|(token, weight)| (*token, weight.to_bits()))
717 .collect();
718 query.sort_unstable();
719 cursor_u64(&mut out, query.len() as u64);
720 for (token, weight) in query {
721 cursor_u32(&mut out, token);
722 cursor_u32(&mut out, weight);
723 }
724 }
725 Condition::MinHashSimilar {
726 column_id,
727 query,
728 k,
729 } => {
730 out.push(9);
731 cursor_u16(&mut out, *column_id);
732 cursor_u64(&mut out, *k as u64);
733 let mut query = query.clone();
734 query.sort_unstable();
735 query.dedup();
736 cursor_u64(&mut out, query.len() as u64);
737 for value in query {
738 cursor_u64(&mut out, value);
739 }
740 }
741 Condition::IsNull { column_id } => {
742 out.push(10);
743 cursor_u16(&mut out, *column_id);
744 }
745 Condition::IsNotNull { column_id } => {
746 out.push(11);
747 cursor_u16(&mut out, *column_id);
748 }
749 Condition::BytesPrefix { column_id, prefix } => {
750 out.push(12);
751 cursor_u16(&mut out, *column_id);
752 cursor_bytes(&mut out, prefix);
753 }
754 }
755 out
756}
757
758fn cursor_conditions(out: &mut Vec<u8>, conditions: &[Condition]) {
759 let mut encoded: Vec<_> = conditions.iter().map(cursor_condition).collect();
760 encoded.sort();
761 encoded.dedup();
762 cursor_u64(out, encoded.len() as u64);
763 for condition in encoded {
764 cursor_bytes(out, &condition);
765 }
766}
767
768fn cursor_projection(out: &mut Vec<u8>, projection: Option<&[u16]>) {
769 match projection {
770 Some(projection) => {
771 out.push(1);
772 let mut projection = projection.to_vec();
773 projection.sort_unstable();
774 projection.dedup();
775 cursor_u64(out, projection.len() as u64);
776 for column in projection {
777 cursor_u16(out, column);
778 }
779 }
780 None => out.push(0),
781 }
782}
783
784fn cursor_set_member(member: &SetMember) -> Vec<u8> {
785 let mut out = Vec::new();
786 match member {
787 SetMember::String(value) => {
788 out.push(0);
789 cursor_bytes(&mut out, value.as_bytes());
790 }
791 SetMember::Number(value) => {
792 out.push(1);
793 cursor_bytes(&mut out, value.to_string().as_bytes());
794 }
795 SetMember::Boolean(value) => {
796 out.push(2);
797 out.push(u8::from(*value));
798 }
799 }
800 out
801}
802
803fn cursor_retriever(retriever: &Retriever) -> Vec<u8> {
804 let mut out = Vec::new();
805 match retriever {
806 Retriever::Ann {
807 column_id,
808 query,
809 k,
810 } => {
811 out.push(0);
812 cursor_u16(&mut out, *column_id);
813 cursor_u64(&mut out, *k as u64);
814 cursor_u64(&mut out, query.len() as u64);
815 for value in query {
816 cursor_u32(&mut out, value.to_bits());
817 }
818 }
819 Retriever::Sparse {
820 column_id,
821 query,
822 k,
823 } => {
824 out.push(1);
825 cursor_u16(&mut out, *column_id);
826 cursor_u64(&mut out, *k as u64);
827 let mut query: Vec<_> = query
828 .iter()
829 .map(|(token, weight)| (*token, weight.to_bits()))
830 .collect();
831 query.sort_unstable();
832 cursor_u64(&mut out, query.len() as u64);
833 for (token, weight) in query {
834 cursor_u32(&mut out, token);
835 cursor_u32(&mut out, weight);
836 }
837 }
838 Retriever::MinHash {
839 column_id,
840 members,
841 k,
842 } => {
843 out.push(2);
844 cursor_u16(&mut out, *column_id);
845 cursor_u64(&mut out, *k as u64);
846 let mut members: Vec<_> = members.iter().map(cursor_set_member).collect();
847 members.sort();
848 members.dedup();
849 cursor_u64(&mut out, members.len() as u64);
850 for member in members {
851 cursor_bytes(&mut out, &member);
852 }
853 }
854 }
855 out
856}
857
858fn cursor_sha256(bytes: &[u8]) -> [u8; 32] {
859 use sha2::Digest;
860 sha2::Sha256::digest(bytes).into()
861}
862
863pub fn canonical_query_cursor_hash(
865 table: &str,
866 conditions: &[Condition],
867 projection: Option<&[u16]>,
868) -> [u8; 32] {
869 let mut out = Vec::new();
870 cursor_u64(&mut out, CURSOR_REQUEST_HASH_VERSION);
871 out.push(0);
872 cursor_bytes(&mut out, table.as_bytes());
873 cursor_conditions(&mut out, conditions);
874 cursor_projection(&mut out, projection);
875 cursor_sha256(&out)
876}
877
878pub fn canonical_search_cursor_hash(table: &str, request: &SearchRequest) -> [u8; 32] {
880 let mut out = Vec::new();
881 cursor_u64(&mut out, CURSOR_REQUEST_HASH_VERSION);
882 out.push(1);
883 cursor_bytes(&mut out, table.as_bytes());
884 cursor_conditions(&mut out, &request.must);
885 let mut retrievers: Vec<Vec<u8>> = request
886 .retrievers
887 .iter()
888 .map(|named| {
889 let mut encoded = Vec::new();
890 cursor_bytes(&mut encoded, named.name.as_bytes());
891 cursor_u64(&mut encoded, named.weight.to_bits());
892 cursor_bytes(&mut encoded, &cursor_retriever(&named.retriever));
893 encoded
894 })
895 .collect();
896 retrievers.sort();
897 cursor_u64(&mut out, retrievers.len() as u64);
898 for retriever in retrievers {
899 cursor_bytes(&mut out, &retriever);
900 }
901 match request.fusion {
902 Fusion::ReciprocalRank { constant } => {
903 out.push(0);
904 cursor_u32(&mut out, constant);
905 }
906 }
907 match &request.rerank {
908 Some(Rerank::ExactVector {
909 embedding_column,
910 query,
911 metric,
912 candidate_limit,
913 weight,
914 }) => {
915 out.push(1);
916 cursor_u16(&mut out, *embedding_column);
917 out.push(match metric {
918 VectorMetric::Cosine => 0,
919 VectorMetric::DotProduct => 1,
920 VectorMetric::Euclidean => 2,
921 });
922 cursor_u64(&mut out, *candidate_limit as u64);
923 cursor_u64(&mut out, weight.to_bits());
924 cursor_u64(&mut out, query.len() as u64);
925 for value in query {
926 cursor_u32(&mut out, value.to_bits());
927 }
928 }
929 None => out.push(0),
930 }
931 cursor_projection(&mut out, request.projection.as_deref());
932 cursor_sha256(&out)
933}
934
935pub fn condition_columns(conditions: &[Condition]) -> Vec<u16> {
940 let mut cols: Vec<u16> = conditions
941 .iter()
942 .filter_map(|c| match c {
943 Condition::Pk(_) => None,
944 Condition::BitmapEq { column_id, .. }
945 | Condition::BitmapIn { column_id, .. }
946 | Condition::BytesPrefix { column_id, .. }
947 | Condition::Ann { column_id, .. }
948 | Condition::FmContains { column_id, .. }
949 | Condition::FmContainsAll { column_id, .. }
950 | Condition::Range { column_id, .. }
951 | Condition::RangeF64 { column_id, .. }
952 | Condition::SparseMatch { column_id, .. }
953 | Condition::MinHashSimilar { column_id, .. }
954 | Condition::IsNull { column_id }
955 | Condition::IsNotNull { column_id } => Some(*column_id),
956 })
957 .collect();
958 cols.sort_unstable();
959 cols.dedup();
960 cols
961}
962
963#[cfg(test)]
964mod tests {
965 use super::*;
966
967 #[test]
968 fn builder_chains() {
969 let q = Query::pk(b"k".to_vec()).and(Condition::Range {
970 column_id: 1,
971 lo: 0,
972 hi: 10,
973 });
974 assert_eq!(q.conditions.len(), 2);
975 }
976
977 #[test]
981 fn canonical_key_is_order_independent() {
982 let e = 7u64;
983 let a = Query::new()
984 .and(Condition::Range {
985 column_id: 1,
986 lo: 0,
987 hi: 10,
988 })
989 .and(Condition::BitmapEq {
990 column_id: 2,
991 value: b"x".to_vec(),
992 });
993 let b = Query::new()
994 .and(Condition::BitmapEq {
995 column_id: 2,
996 value: b"x".to_vec(),
997 })
998 .and(Condition::Range {
999 column_id: 1,
1000 lo: 0,
1001 hi: 10,
1002 });
1003 assert_eq!(
1004 canonical_query_key(&a.conditions, None, e),
1005 canonical_query_key(&b.conditions, None, e),
1006 "condition order must not affect the key"
1007 );
1008
1009 let minhash = Condition::MinHashSimilar {
1010 column_id: 4,
1011 query: vec![3, 1, 1, 2],
1012 k: 5,
1013 };
1014 let minhash_canonical = Condition::MinHashSimilar {
1015 column_id: 4,
1016 query: vec![1, 2, 3],
1017 k: 5,
1018 };
1019 assert_eq!(
1020 canonical_query_key(std::slice::from_ref(&minhash), None, e),
1021 canonical_query_key(&[minhash_canonical], None, e),
1022 );
1023
1024 let fm = Condition::FmContainsAll {
1025 column_id: 4,
1026 patterns: vec![b"b".to_vec(), b"a".to_vec(), b"a".to_vec()],
1027 };
1028 let fm_canonical = Condition::FmContainsAll {
1029 column_id: 4,
1030 patterns: vec![b"a".to_vec(), b"b".to_vec()],
1031 };
1032 assert_eq!(
1033 canonical_query_key(std::slice::from_ref(&fm), None, e),
1034 canonical_query_key(&[fm_canonical], None, e),
1035 );
1036 assert_ne!(
1037 canonical_query_key(std::slice::from_ref(&fm), None, e),
1038 canonical_query_key(std::slice::from_ref(&minhash), None, e),
1039 );
1040 assert_ne!(
1041 canonical_query_key_with_version(1, &a.conditions, None, e),
1042 canonical_query_key_with_version(2, &a.conditions, None, e),
1043 );
1044
1045 let ordered = Condition::BitmapIn {
1047 column_id: 3,
1048 values: vec![b"a".to_vec(), b"b".to_vec(), b"c".to_vec()],
1049 };
1050 let shuffled = Condition::BitmapIn {
1051 column_id: 3,
1052 values: vec![b"c".to_vec(), b"a".to_vec(), b"a".to_vec(), b"b".to_vec()],
1053 };
1054 assert_eq!(
1055 canonical_query_key(std::slice::from_ref(&ordered), None, e),
1056 canonical_query_key(&[shuffled], None, e),
1057 "BitmapIn values must dedup+sort"
1058 );
1059
1060 assert_ne!(
1062 canonical_query_key(&a.conditions, None, e),
1063 canonical_query_key(&a.conditions, None, e + 1),
1064 "epoch must fold into the key"
1065 );
1066
1067 let proj = vec![1u16, 2];
1069 assert_ne!(
1070 canonical_query_key(&a.conditions, None, e),
1071 canonical_query_key(&a.conditions, Some(&proj), e),
1072 "None projection must differ from an explicit projection"
1073 );
1074 let proj_rev = vec![2u16, 1];
1076 assert_eq!(
1077 canonical_query_key(&a.conditions, Some(&proj), e),
1078 canonical_query_key(&a.conditions, Some(&proj_rev), e),
1079 );
1080
1081 let budget = AiExecutionContext::new(None, 2);
1082 budget.consume(1).unwrap();
1083 assert!(matches!(
1084 budget.consume(2),
1085 Err(crate::MongrelError::WorkBudgetExceeded)
1086 ));
1087 budget.cancel();
1088 assert!(matches!(
1089 budget.checkpoint(),
1090 Err(crate::MongrelError::Cancelled)
1091 ));
1092
1093 let expired = AiExecutionContext::new(
1094 Some(std::time::Instant::now() - std::time::Duration::from_millis(1)),
1095 1,
1096 );
1097 assert!(matches!(
1098 expired.checkpoint(),
1099 Err(crate::MongrelError::DeadlineExceeded)
1100 ));
1101
1102 let parent = crate::ExecutionControl::new(None);
1103 let child = AiExecutionContext::with_control(
1104 parent.child_with_deadline(None),
1105 10,
1106 MAX_FUSED_CANDIDATES,
1107 );
1108 parent.cancel(crate::CancellationReason::SessionClosed);
1109 assert!(matches!(
1110 child.checkpoint(),
1111 Err(crate::MongrelError::Cancelled)
1112 ));
1113 }
1114
1115 #[test]
1116 fn cursor_request_hashes_are_canonical_and_semantic() {
1117 let a = vec![
1118 Condition::BitmapIn {
1119 column_id: 2,
1120 values: vec![b"b".to_vec(), b"a".to_vec(), b"a".to_vec()],
1121 },
1122 Condition::Range {
1123 column_id: 3,
1124 lo: 1,
1125 hi: 9,
1126 },
1127 ];
1128 let b = vec![
1129 Condition::Range {
1130 column_id: 3,
1131 lo: 1,
1132 hi: 9,
1133 },
1134 Condition::BitmapIn {
1135 column_id: 2,
1136 values: vec![b"a".to_vec(), b"b".to_vec()],
1137 },
1138 ];
1139 assert_eq!(
1140 canonical_query_cursor_hash("docs", &a, Some(&[2, 1, 2])),
1141 canonical_query_cursor_hash("docs", &b, Some(&[1, 2])),
1142 );
1143 assert_ne!(
1144 canonical_query_cursor_hash("docs", &a, Some(&[1, 2])),
1145 canonical_query_cursor_hash("other", &a, Some(&[1, 2])),
1146 );
1147
1148 let search = SearchRequest {
1149 must: a,
1150 retrievers: vec![NamedRetriever {
1151 name: "dense".into(),
1152 weight: 1.0,
1153 retriever: Retriever::Ann {
1154 column_id: 4,
1155 query: vec![1.0, -1.0],
1156 k: 10,
1157 },
1158 }],
1159 fusion: Fusion::ReciprocalRank { constant: 60 },
1160 rerank: None,
1161 limit: 1,
1162 projection: Some(vec![2, 1]),
1163 };
1164 let mut next_page = search.clone();
1165 next_page.limit = 20;
1166 next_page.projection = Some(vec![1, 2]);
1167 assert_eq!(
1168 canonical_search_cursor_hash("docs", &search),
1169 canonical_search_cursor_hash("docs", &next_page),
1170 "page size and projection order must not change search identity",
1171 );
1172 next_page.retrievers[0].weight = 2.0;
1173 assert_ne!(
1174 canonical_search_cursor_hash("docs", &search),
1175 canonical_search_cursor_hash("docs", &next_page),
1176 );
1177 }
1178}