kevy_embedded/
ops_index_claused.rs

1//! The clause-carrying scalar query for the embedded API — `FILTER` /
2//! `SORT` / `DISTINCT` / `FACET` / `OFFSET` on Range/Unique indexes
3//! (the capacity arc's G1), plus [`ValueFilter`], the predicate shape
4//! this surface shares with the text MATCH clauses. A `#[path]` child
5//! of `ops_index.rs`, feature-independent of `text`.
6
7use kevy_index::{
8    Cursor, FacetBucket, IndexSpec, IndexValue, ScalarClauses, ScalarHit, ValType, ValueTest,
9    fold_facets, merge_claused, sort_facets,
10};
11
12use super::sync_segs;
13use crate::store::{Store, lock_write};
14use crate::{KevyError, KevyResult};
15
16/// One `FILTER` predicate: which stored value field it reads, and the
17/// test on it — the wire's `RANGE` / `EQ` shapes, in-process.
18///
19/// The bounds are raw bytes and are coerced with the type the field was
20/// DECLARED as, so a numeric range compares numerically rather than
21/// lexicographically.
22#[derive(Clone, Copy)]
23pub enum ValueFilter<'a> {
24    /// `field` between `min` and `max`, both inclusive.
25    Range {
26        /// The declared value field to read.
27        field: &'a [u8],
28        /// Lower bound, inclusive.
29        min: &'a [u8],
30        /// Upper bound, inclusive.
31        max: &'a [u8],
32    },
33    /// `field` exactly `value`.
34    Eq {
35        /// The declared value field to read.
36        field: &'a [u8],
37        /// The value to match.
38        value: &'a [u8],
39    },
40}
41
42impl ValueFilter<'_> {
43    pub(crate) fn field(&self) -> &[u8] {
44        match self {
45            ValueFilter::Range { field, .. } | ValueFilter::Eq { field, .. } => field,
46        }
47    }
48}
49
50/// One `FILTER` predicate resolved against the spec: the stored-value
51/// position it reads, and the test built with that field's DECLARED type.
52pub(crate) fn value_test(
53    spec: &IndexSpec,
54    f: &ValueFilter<'_>,
55) -> KevyResult<(usize, ValueTest)> {
56    let stored: Vec<&[u8]> = spec.values.iter().map(|v| v.name.as_slice()).collect();
57    let pos = spec
58        .values
59        .iter()
60        .position(|v| v.name == f.field())
61        .ok_or_else(|| unknown_field("FILTER", f.field(), "store", &stored))?;
62    let ty = spec.values[pos].ty;
63    let (test, raw) = match f {
64        ValueFilter::Range { min, max, .. } => (ValueTest::range(ty, min, max), *min),
65        ValueFilter::Eq { value, .. } => (ValueTest::eq(ty, value), *value),
66    };
67    let test = test.ok_or_else(|| {
68        KevyError::InvalidInput(format!(
69            "FILTER bound '{}' is not a valid {}, which is how this index declares '{}'",
70            String::from_utf8_lossy(raw),
71            ty.tag(),
72            String::from_utf8_lossy(f.field()),
73        ))
74    })?;
75    Ok((pos, test))
76}
77
78/// A clause naming a field the index does not offer, saying what it does.
79pub(crate) fn unknown_field(clause: &str, bad: &[u8], verb: &str, offered: &[&[u8]]) -> KevyError {
80    let names: Vec<String> =
81        offered.iter().map(|n| String::from_utf8_lossy(n).into_owned()).collect();
82    KevyError::InvalidInput(format!(
83        "{clause} names field '{}', which this index does not {verb} — it {verb}es: {}",
84        String::from_utf8_lossy(bad),
85        names.join(", ")
86    ))
87}
88
89/// Everything a scalar RANGE/EQ query carries beyond its bounds and
90/// limit — the embedded twin of the wire's optional clauses.
91/// [`ScalarQueryOpts::default`] is the plain query.
92#[derive(Clone, Copy, Default)]
93pub struct ScalarQueryOpts<'a> {
94    /// `FILTER …`: non-scoring predicates over stored values, ANDed. A
95    /// row without the stored value fails — absent is not a value.
96    pub filters: &'a [ValueFilter<'a>],
97    /// `SORT <field> ASC|DESC`: order the page by a stored value; a row
98    /// with no usable value sorts last in both directions.
99    pub sort: Option<(&'a [u8], bool)>,
100    /// `DISTINCT <field>`: at most one row per coerced value; a row
101    /// with no value is its own group.
102    pub distinct: Option<&'a [u8]>,
103    /// `FACET <field…>`: count each field's values over the whole match
104    /// set (FILTER reduces the counts; DISTINCT does not).
105    pub facets: &'a [Vec<u8>],
106    /// `OFFSET n`: rows to skip before `limit` takes effect.
107    pub offset: usize,
108}
109
110impl ScalarQueryOpts<'_> {
111    /// Whether any clause reshapes the selection (the cursor-refusing
112    /// set — `FILTER` alone pages fine).
113    pub(crate) fn selects(&self) -> bool {
114        self.sort.is_some()
115            || self.distinct.is_some()
116            || !self.facets.is_empty()
117            || self.offset > 0
118    }
119}
120
121/// A clause-carrying query's answer: the page, per requested `FACET`
122/// field its `(value, count)` buckets, and — on the FILTER-with-cursor
123/// path — the cursor to resume from.
124#[derive(Debug)]
125pub struct ScalarPage {
126    /// The selected rows, in the page's order.
127    pub rows: Vec<(Vec<u8>, IndexValue)>,
128    /// One entry per requested facet field, most frequent first.
129    pub facets: Vec<Vec<(Vec<u8>, u64)>>,
130    /// Resume cursor (`None` under any selection clause, which refuses
131    /// cursors at the wire and pages nothing here either).
132    pub cursor: Option<Cursor>,
133}
134
135/// The unmerged union of the shards' pages plus the folded facet
136/// partials (the payload slot is `()` — the wire's hydration rides only
137/// the server's chunks).
138type GatheredPages = (Vec<(ScalarHit, ())>, Vec<Vec<FacetBucket>>);
139
140/// What the spec-dependent clauses resolve to.
141struct Resolved {
142    filters: Vec<(usize, ValueTest)>,
143    sort: Option<(usize, bool, ValType)>,
144    distinct: Option<(usize, ValType)>,
145    facets: Vec<(usize, ValType)>,
146}
147
148/// A clause's named stored-value field position + declared type.
149fn value_field(
150    spec: &IndexSpec,
151    clause: &str,
152    field: &[u8],
153) -> KevyResult<(usize, ValType)> {
154    let stored: Vec<&[u8]> = spec.values.iter().map(|v| v.name.as_slice()).collect();
155    let pos = spec
156        .values
157        .iter()
158        .position(|v| v.name == field)
159        .ok_or_else(|| unknown_field(clause, field, "store", &stored))?;
160    Ok((pos, spec.values[pos].ty))
161}
162
163fn resolve(spec: &IndexSpec, opts: &ScalarQueryOpts<'_>) -> KevyResult<Resolved> {
164    let filters =
165        opts.filters.iter().map(|f| value_test(spec, f)).collect::<KevyResult<Vec<_>>>()?;
166    let sort = match opts.sort {
167        Some((field, desc)) => {
168            let (pos, ty) = value_field(spec, "SORT", field)?;
169            Some((pos, desc, ty))
170        }
171        None => None,
172    };
173    let distinct = match opts.distinct {
174        Some(field) => Some(value_field(spec, "DISTINCT", field)?),
175        None => None,
176    };
177    let facets = opts
178        .facets
179        .iter()
180        .map(|f| value_field(spec, "FACET", f))
181        .collect::<KevyResult<Vec<_>>>()?;
182    Ok(Resolved { filters, sort, distinct, facets })
183}
184
185impl Store {
186    /// [`Store::idx_create`] with declared stored `VALUES` columns —
187    /// the scalar kinds' G1 capability (the catalog refuses the
188    /// declaration on kinds that carry no stored-value column).
189    pub fn idx_create_with_values(
190        &self,
191        name: &[u8],
192        prefix: &[u8],
193        field: &[u8],
194        ty: ValType,
195        kind: kevy_index::IndexKind,
196        values: &[(&[u8], ValType)],
197    ) -> KevyResult<()> {
198        if prefix.is_empty() {
199            return Err(KevyError::InvalidInput("empty prefix".into()));
200        }
201        let spec = IndexSpec {
202            name: name.to_vec(),
203            prefix: prefix.to_vec(),
204            fields: vec![kevy_index::FieldSpec::new(field.to_vec())],
205            ty,
206            kind,
207            max_bytes: 0,
208            ann: None,
209            group_by: None,
210            with_positions: false,
211            values: values
212                .iter()
213                .map(|(n, t)| kevy_index::ValueSpec { name: n.to_vec(), ty: *t })
214                .collect(),
215            composite: None,
216        };
217        self.register_spec(spec)
218    }
219
220    /// [`Store::idx_query`] with the stored-value clauses — the scalar
221    /// twin of [`Store::idx_match_faceted`]'s clause surface. Semantics
222    /// are the wire's: each shard selects with the clauses applied, the
223    /// union merges in the page's own order, `DISTINCT` re-collapses,
224    /// `OFFSET` drains after the merge, facet counts sum by coerced
225    /// identity. Only `FILTER` (with no other clause) is cursor-paged.
226    pub fn idx_query_claused(
227        &self,
228        name: &[u8],
229        min: &IndexValue,
230        max: &IndexValue,
231        cursor: Option<&Cursor>,
232        limit: usize,
233        opts: ScalarQueryOpts<'_>,
234    ) -> KevyResult<ScalarPage> {
235        let limit = limit.clamp(1, 100_000);
236        let offset = opts.offset.min(10_000);
237        let spec = {
238            let g = self.indexes.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
239            g.1.get(name)
240                .map(|(s, _)| s.clone())
241                .ok_or_else(|| KevyError::NotFound("no such index".into()))?
242        };
243        let r = resolve(&spec, &opts)?;
244        let clauses = ScalarClauses {
245            filters: &r.filters,
246            sort: r.sort,
247            distinct: r.distinct,
248            facets: &r.facets,
249            fetch: limit + offset,
250        };
251        let (all, mut facets) = self.gather_claused(name, min, max, cursor, &clauses)?;
252        let sort_desc = r.sort.map(|(_, desc, _)| desc);
253        let all = merge_claused(all, sort_desc, r.distinct.is_some(), offset, limit);
254        sort_facets(&mut facets);
255        let next = (!opts.selects() && all.len() == limit)
256            .then(|| all.last().map(|(h, ())| Cursor { value: h.value.clone(), key: h.key.clone() }))
257            .flatten();
258        Ok(ScalarPage {
259            rows: all.into_iter().map(|(h, ())| (h.key, h.value)).collect(),
260            facets: facets
261                .into_iter()
262                .map(|f| f.into_iter().map(|(_, label, n)| (label, n)).collect())
263                .collect(),
264            cursor: next,
265        })
266    }
267
268    /// Every shard's claused page, unmerged, with the facet buckets
269    /// folded by identity as the shards report them.
270    fn gather_claused(
271        &self,
272        name: &[u8],
273        min: &IndexValue,
274        max: &IndexValue,
275        cursor: Option<&Cursor>,
276        clauses: &ScalarClauses<'_>,
277    ) -> KevyResult<GatheredPages> {
278        let mut all: Vec<(ScalarHit, ())> = Vec::new();
279        let mut facets: Vec<Vec<FacetBucket>> = vec![Vec::new(); clauses.facets.len()];
280        let mut found = false;
281        for shard in self.shards.iter() {
282            let mut g = lock_write(shard);
283            let inner = &mut *g;
284            sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
285            if let Some((_, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name == name) {
286                found = true;
287                let page = seg.query_claused(min, max, cursor, clauses);
288                all.extend(page.hits.into_iter().map(|h| (h, ())));
289                fold_facets(&mut facets, page.facets);
290            }
291        }
292        if !found {
293            return Err(KevyError::NotFound("no such index".into()));
294        }
295        Ok((all, facets))
296    }
297}
kevy_embedded/ops_index_claused.rs

kevy_embedded/
ops_index_claused.rs
