Skip to main content

mongreldb_core/
query.rs

1//! Tool-call-native query surface.
2//!
3//! A [`Query`] is a conjunction of [`Condition`]s. Each condition resolves to a
4//! set of row ids in the shared [`crate::rowid::RowId`] space (PK exact, bitmap
5//! equality, ANN semantic, FM substring, or a column range). [`crate::Table`]
6//! intersects the sets and materializes the survivors — letting an agent express
7//! `semsearch ∩ fm_contains ∩ cat_in`, which no SQL FTS pipeline can.
8
9/// Hard safety ceilings shared by every public query surface.
10pub 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/// Cooperative deadline, cancellation, and work-budget state for expensive
33/// AI queries. Index loops call `checkpoint` and charge work with `consume`.
34#[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/// One predicate over the row-id space.
145#[derive(Debug, Clone)]
146pub enum Condition {
147    /// Primary-key exact match (encoded key bytes).
148    Pk(Vec<u8>),
149    /// Low-cardinality equality via the roaring bitmap index.
150    BitmapEq { column_id: u16, value: Vec<u8> },
151    /// Multi-value equality via the roaring bitmap index (Phase 13.5). Resolves
152    /// to the **union** of `bitmap[col].get(v)` for each value — the index-
153    /// accelerated equivalent of `col IN (v1, v2, …)` or a semi-join's runtime
154    /// value set.
155    BitmapIn {
156        column_id: u16,
157        values: Vec<Vec<u8>>,
158    },
159    /// Prefix match on a Bytes column with a bitmap index: all row-ids whose
160    /// indexed value starts with `prefix`. Exact (no residual needed) — the
161    /// bitmap's distinct keys are enumerated and filtered by prefix. Tighter
162    /// than `FmContains` for anchored `LIKE 'prefix%'`. (§5.6)
163    BytesPrefix { column_id: u16, prefix: Vec<u8> },
164    /// Semantic search via the binary-quantized ANN index.
165    Ann {
166        column_id: u16,
167        query: Vec<f32>,
168        k: usize,
169    },
170    /// Arbitrary substring via the FM index (no tokenization).
171    FmContains { column_id: u16, pattern: Vec<u8> },
172    /// Multi-segment FM intersection for `LIKE '%seg1%seg2%...'` (Priority 12).
173    /// Resolves to the **intersection** of FM lookups for each segment — a much
174    /// tighter superset than the single longest segment. DataFusion still
175    /// re-applies the real wildcard semantics (`Inexact` pushdown).
176    FmContainsAll {
177        column_id: u16,
178        patterns: Vec<Vec<u8>>,
179    },
180    /// Inclusive integer range (served by scanning the int column, later by the
181    /// learned PGM index / page-index pruning). Exclusive bounds (`>`,`<`) are
182    /// expressed exactly via ±1 in the translator.
183    Range { column_id: u16, lo: i64, hi: i64 },
184    /// Floating-point range with per-bound inclusivity (exact for `>`/`<`/`>=`/
185    /// `<=`/`BETWEEN`), served the same way as [`Condition::Range`].
186    RangeF64 {
187        column_id: u16,
188        lo: f64,
189        lo_inclusive: bool,
190        hi: f64,
191        hi_inclusive: bool,
192    },
193    /// SPLADE-style sparse retrieval: top-k row ids by sparse dot product over
194    /// shared tokens. `query` is a sparse vector `(token id → weight)`.
195    SparseMatch {
196        column_id: u16,
197        query: Vec<(u32, f32)>,
198        k: usize,
199    },
200    /// MinHash/LSH set-similarity: candidate row ids whose set is similar to
201    /// `query` (a set of 64-bit token hashes), ranked by estimated Jaccard,
202    /// truncated to `k`. Approximate (LSH recall) — the caller re-verifies.
203    MinHashSimilar {
204        column_id: u16,
205        query: Vec<u64>,
206        k: usize,
207    },
208    /// Rows where `column_id` is NULL. Resolved by decoding the column and
209    /// collecting null positions — a column scan, but no row materialization.
210    /// Page-stat aware: pages with `null_count == 0` are skipped.
211    IsNull { column_id: u16 },
212    /// Rows where `column_id` is NOT NULL. The complement of [`Self::IsNull`].
213    /// Page-stat aware: pages with `null_count == row_count` are skipped.
214    IsNotNull { column_id: u16 },
215}
216
217/// Ordered candidate generator. Unlike [`Condition`], retrieval preserves the
218/// index score and rank.
219#[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    /// One-based rank.
313    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    /// One-based rank after optional reranking.
400    pub final_rank: usize,
401}
402
403/// A conjunctive query. Empty ⇒ all rows.
404#[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
432/// Canonical 64-bit cache key for a conjunctive native query + optional
433/// projection at `epoch` (Phase 19.1 / 19.6). Conditions are commutative (they
434/// are ANDed), so each condition is hashed into its own 64-bit digest, the
435/// digests are sorted, then folded together — two queries with the same
436/// semantics in a different order produce the same key. Within a condition,
437/// `BitmapIn` values are deduped+sorted and the `SparseMatch` query is sorted
438/// by token id. `epoch` is folded in so a `commit()` (which bumps it) orphans
439/// every prior entry without an explicit sweep.
440pub 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    // Order-independent: per-condition digests, sorted, then folded.
460    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    // Projection: sorted column ids (None ⇒ "all columns", distinct from any
468    // explicit projection incl. one listing every column, by intent).
469    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
487/// Hash a single condition into a 64-bit digest (order-independent w.r.t. its
488/// siblings; see [`canonical_query_key`]). Floats are hashed via `to_bits` for
489/// determinism.
490fn 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
863/// Versioned, canonical request hash for native-query cursor binding.
864pub 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
878/// Versioned, canonical request hash for scored-search cursor binding.
879pub 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
935/// Extract the column IDs referenced by a slice of conditions (Phase 19.1
936/// hardening (c)). `Pk` references no user column (it's a row-id lookup) so it
937/// contributes nothing. Used for conservative column-based cache invalidation:
938/// a commit touching any of these columns may change the result.
939pub 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    /// Phase 19.6: order-independent canonicalization — the same conditions in a
978    /// different order, and a `BitmapIn` with shuffled/duplicate values, all
979    /// produce the same key.
980    #[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        // BitmapIn dedup + sort.
1046        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        // Epoch changes the key (invalidation).
1061        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        // Projection None vs explicit differs (by intent).
1068        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        // Projection order-independence.
1075        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}