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
12pub(crate) use super::opts::{ScalarQueryOpts, ValueFilter};
13use super::sync_segs;
14use crate::store::{Store, lock_write};
15use crate::{KevyError, KevyResult};
16
17/// One `FILTER` predicate resolved against the spec: the stored-value
18/// position it reads, and the test built with that field's DECLARED type.
19pub(crate) fn value_test(spec: &IndexSpec, f: &ValueFilter<'_>) -> KevyResult<(usize, ValueTest)> {
20 let stored: Vec<&[u8]> = spec.values().iter().map(|v| v.name.as_slice()).collect();
21 let pos = spec
22 .values()
23 .iter()
24 .position(|v| v.name == f.field())
25 .ok_or_else(|| unknown_field("FILTER", f.field(), "store", &stored))?;
26 let ty = spec.values()[pos].ty;
27 // FILTER bounds speak the `@` time expressions on i64 fields —
28 // one grammar across both faces (and, through this shared
29 // resolver, the embedded Rust API too).
30 let now = (kevy_store::now_unix_ms() / 1000) as i64;
31 let (test, raw) = match f {
32 ValueFilter::Range { min, max, .. } => (ValueTest::range_at(ty, min, max, now), *min),
33 ValueFilter::Eq { value, .. } => (ValueTest::eq_at(ty, value, now), *value),
34 };
35 let test = test.ok_or_else(|| {
36 KevyError::InvalidInput(format!(
37 "FILTER bound '{}' is not a valid {}, which is how this index declares '{}'",
38 String::from_utf8_lossy(raw),
39 ty.tag(),
40 String::from_utf8_lossy(f.field()),
41 ))
42 })?;
43 Ok((pos, test))
44}
45
46/// The first clause field the spec does not store — the advise-log
47/// derivation for a refused [`resolve`]. Checked in the clause order
48/// [`resolve`] resolves in, so it names the same field the error does.
49fn unstored_field(spec: &IndexSpec, opts: &ScalarQueryOpts<'_>) -> Option<Vec<u8>> {
50 let stored = |f: &[u8]| spec.values().iter().any(|v| v.name == f);
51 for f in opts.filters {
52 if !stored(f.field()) {
53 return Some(f.field().to_vec());
54 }
55 }
56 if let Some((f, _)) = opts.sort
57 && !stored(f)
58 {
59 return Some(f.to_vec());
60 }
61 if let Some(f) = opts.distinct
62 && !stored(f)
63 {
64 return Some(f.to_vec());
65 }
66 opts.facets.iter().find(|f| !stored(f)).cloned()
67}
68
69/// A clause naming a field the index does not offer, saying what it does.
70pub(crate) fn unknown_field(clause: &str, bad: &[u8], verb: &str, offered: &[&[u8]]) -> KevyError {
71 let names: Vec<String> =
72 offered.iter().map(|n| String::from_utf8_lossy(n).into_owned()).collect();
73 KevyError::InvalidInput(format!(
74 "{clause} names field '{}', which this index does not {verb} — it {}: {}",
75 String::from_utf8_lossy(bad),
76 // Third person singular: a sibilant takes -es ("indexes"), anything
77 // else takes -s ("stores"). `{verb}es` for both printed "storees".
78 if verb.ends_with(['s', 'x', 'z']) { format!("{verb}es") } else { format!("{verb}s") },
79 names.join(", ")
80 ))
81}
82
83/// A clause-carrying query's answer: the page, per requested `FACET`
84/// field its `(value, count)` buckets, and — on the FILTER-with-cursor
85/// path — the cursor to resume from.
86///
87/// ```
88/// use kevy_embedded::{Config, IndexValue, ScalarQueryOpts, Store};
89///
90/// let s = Store::open(Config::default())?;
91/// s.idx_create(b"by_age", b"u:", b"age", kevy_embedded::IndexValType::I64, kevy_embedded::IndexKind::Range)?;
92/// s.hset(b"u:1", &[(b"age", b"30")])?;
93/// let page = s.idx_query_claused(b"by_age", &IndexValue::I64(0), &IndexValue::I64(99), None, 10, ScalarQueryOpts::default())?;
94/// assert_eq!(page.rows.len(), 1);
95/// # Ok::<(), kevy_embedded::KevyError>(())
96/// ```
97#[derive(Debug, Clone, PartialEq)]
98#[non_exhaustive]
99pub struct ScalarPage {
100 /// The selected rows, in the page's order.
101 ///
102 /// ```
103 /// # use kevy_embedded::*;
104 /// # let s = Store::open(Config::default())?;
105 /// # let st = [(&b"status"[..], IndexValType::Str)];
106 /// # s.idx_create_with_values(b"by_age", b"u:", b"age", IndexValType::I64, IndexKind::Range, &st)?;
107 /// # for (k, age, status) in [(&b"u:1"[..], &b"30"[..], &b"paid"[..]), (b"u:2", b"40", b"paid"), (b"u:3", b"50", b"due")] {
108 /// # s.hset(k, &[(b"age", age), (b"status", status)])?;
109 /// # }
110 /// # let (lo, hi) = (IndexValue::I64(0), IndexValue::I64(99));
111 /// // u:1..u:3 aged 30, 40, 50; two paid, one due
112 /// let paid = [ValueFilter::Eq { field: b"status", value: b"paid" }];
113 /// let opts = ScalarQueryOpts::default().with_filters(&paid);
114 /// let page = s.idx_query_claused(b"by_age", &lo, &hi, None, 10, opts)?;
115 /// assert_eq!(page.rows, [(b"u:1".to_vec(), IndexValue::I64(30)), (b"u:2".to_vec(), IndexValue::I64(40))]);
116 /// # Ok::<(), kevy_embedded::KevyError>(())
117 /// ```
118 pub rows: Vec<(Vec<u8>, IndexValue)>,
119 /// One entry per requested facet field, most frequent first.
120 ///
121 /// ```
122 /// # use kevy_embedded::*;
123 /// # let s = Store::open(Config::default())?;
124 /// # let st = [(&b"status"[..], IndexValType::Str)];
125 /// # s.idx_create_with_values(b"by_age", b"u:", b"age", IndexValType::I64, IndexKind::Range, &st)?;
126 /// # for (k, age, status) in [(&b"u:1"[..], &b"30"[..], &b"paid"[..]), (b"u:2", b"40", b"paid"), (b"u:3", b"50", b"due")] {
127 /// # s.hset(k, &[(b"age", age), (b"status", status)])?;
128 /// # }
129 /// # let (lo, hi) = (IndexValue::I64(0), IndexValue::I64(99));
130 /// let fields = [b"status".to_vec()];
131 /// let opts = ScalarQueryOpts::default().with_facets(&fields);
132 /// let page = s.idx_query_claused(b"by_age", &lo, &hi, None, 1, opts)?;
133 /// // counted over every match, not just the one-row page
134 /// assert_eq!(page.facets, [vec![(b"paid".to_vec(), 2), (b"due".to_vec(), 1)]]);
135 /// # Ok::<(), kevy_embedded::KevyError>(())
136 /// ```
137 pub facets: Vec<Vec<(Vec<u8>, u64)>>,
138 /// Resume cursor (`None` under any selection clause, which refuses
139 /// cursors at the wire and pages nothing here either).
140 ///
141 /// ```
142 /// # use kevy_embedded::*;
143 /// # let s = Store::open(Config::default())?;
144 /// # let st = [(&b"status"[..], IndexValType::Str)];
145 /// # s.idx_create_with_values(b"by_age", b"u:", b"age", IndexValType::I64, IndexKind::Range, &st)?;
146 /// # for (k, age, status) in [(&b"u:1"[..], &b"30"[..], &b"paid"[..]), (b"u:2", b"40", b"paid"), (b"u:3", b"50", b"due")] {
147 /// # s.hset(k, &[(b"age", age), (b"status", status)])?;
148 /// # }
149 /// # let (lo, hi) = (IndexValue::I64(0), IndexValue::I64(99));
150 /// let plain = ScalarQueryOpts::default();
151 /// let first = s.idx_query_claused(b"by_age", &lo, &hi, None, 2, plain)?;
152 /// let rest = s.idx_query_claused(b"by_age", &lo, &hi, first.cursor.as_ref(), 2, plain)?;
153 /// assert_eq!(rest.rows[0].0, b"u:3");
154 /// let sorted = plain.with_sort(b"status", SortOrder::Asc);
155 /// assert!(s.idx_query_claused(b"by_age", &lo, &hi, None, 2, sorted)?.cursor.is_none());
156 /// # Ok::<(), kevy_embedded::KevyError>(())
157 /// ```
158 pub cursor: Option<Cursor>,
159}
160
161/// The unmerged union of the shards' pages plus the folded facet
162/// partials (the payload slot is `()` — the wire's hydration rides only
163/// the server's chunks).
164type GatheredPages = (Vec<(ScalarHit, ())>, Vec<Vec<FacetBucket>>);
165
166/// What the spec-dependent clauses resolve to.
167struct Resolved {
168 filters: Vec<(usize, ValueTest)>,
169 sort: Option<(usize, kevy_index::SortOrder, ValType)>,
170 distinct: Option<(usize, ValType)>,
171 facets: Vec<(usize, ValType)>,
172}
173
174/// A clause's named stored-value field position + declared type.
175fn value_field(spec: &IndexSpec, clause: &str, field: &[u8]) -> KevyResult<(usize, ValType)> {
176 let stored: Vec<&[u8]> = spec.values().iter().map(|v| v.name.as_slice()).collect();
177 let pos = spec
178 .values()
179 .iter()
180 .position(|v| v.name == field)
181 .ok_or_else(|| unknown_field(clause, field, "store", &stored))?;
182 Ok((pos, spec.values()[pos].ty))
183}
184
185fn resolve(spec: &IndexSpec, opts: &ScalarQueryOpts<'_>) -> KevyResult<Resolved> {
186 let filters =
187 opts.filters.iter().map(|f| value_test(spec, f)).collect::<KevyResult<Vec<_>>>()?;
188 let sort = match opts.sort {
189 Some((field, order)) => {
190 let (pos, ty) = value_field(spec, "SORT", field)?;
191 Some((pos, order, ty))
192 }
193 None => None,
194 };
195 let distinct = match opts.distinct {
196 Some(field) => Some(value_field(spec, "DISTINCT", field)?),
197 None => None,
198 };
199 let facets = opts
200 .facets
201 .iter()
202 .map(|f| value_field(spec, "FACET", f))
203 .collect::<KevyResult<Vec<_>>>()?;
204 Ok(Resolved { filters, sort, distinct, facets })
205}
206
207impl Store {
208 /// [`Store::idx_create`] with declared stored `VALUES` columns —
209 /// the scalar kinds' G1 capability (the catalog refuses the
210 /// declaration on kinds that carry no stored-value column).
211 pub fn idx_create_with_values(
212 &self,
213 name: &[u8],
214 prefix: &[u8],
215 field: &[u8],
216 ty: ValType,
217 kind: kevy_index::IndexKind,
218 values: &[(&[u8], ValType)],
219 ) -> KevyResult<()> {
220 if prefix.is_empty() {
221 return Err(KevyError::InvalidInput("empty prefix".into()));
222 }
223 let values =
224 values.iter().map(|(n, t)| kevy_index::ValueSpec::new(*n).with_type(*t)).collect();
225 let spec = IndexSpec::builder(name, prefix, kind, ty).with_field(field).with_values(values);
226 let spec = crate::ops_index::built(spec)?;
227 self.catalog_change(|| self.register_spec(spec))
228 }
229
230 /// [`Store::idx_query`] with the stored-value clauses — the scalar
231 /// twin of [`Store::idx_match_faceted`]'s clause surface. Semantics
232 /// are the wire's: each shard selects with the clauses applied, the
233 /// union merges in the page's own order, `DISTINCT` re-collapses,
234 /// `OFFSET` drains after the merge, facet counts sum by coerced
235 /// identity. Only `FILTER` (with no other clause) is cursor-paged.
236 pub fn idx_query_claused(
237 &self,
238 name: &[u8],
239 min: &IndexValue,
240 max: &IndexValue,
241 cursor: Option<&Cursor>,
242 limit: usize,
243 opts: ScalarQueryOpts<'_>,
244 ) -> KevyResult<ScalarPage> {
245 let limit = limit.clamp(1, 100_000);
246 let offset = opts.offset.min(10_000);
247 let spec = self.claused_spec(name)?;
248 let r = self.claused_resolve(name, &spec, &opts)?;
249 let mut clauses = ScalarClauses::new(limit + offset).with_filters(&r.filters);
250 clauses = clauses.with_facets(&r.facets);
251 (clauses.sort, clauses.distinct) = (r.sort, r.distinct);
252 let (all, mut facets) = self.gather_claused(name, min, max, cursor, &clauses)?;
253 let all = merge_claused(all, r.sort.map(|(_, order, _)| order), offset, limit);
254 sort_facets(&mut facets);
255 let next = (!opts.selects() && all.len() == limit)
256 .then(|| all.last().map(|(h, ())| Cursor::new(h.value.clone(), h.key.clone())))
257 .flatten();
258 self.observe_hit(name);
259 Ok(ScalarPage {
260 rows: all.into_iter().map(|(h, ())| (h.key, h.value)).collect(),
261 facets: facets
262 .into_iter()
263 .map(|f| f.into_iter().map(|(_, label, n)| (label, n)).collect())
264 .collect(),
265 cursor: next,
266 })
267 }
268
269 /// [`Store::idx_count`] with `FILTER` applied — the total a
270 /// claused query's pages would reach, materializing nothing. The
271 /// consumer shape this closes: counting a filtered axis used to
272 /// mean fetching every page and taking its length.
273 pub fn idx_count_claused(
274 &self,
275 name: &[u8],
276 min: &IndexValue,
277 max: &IndexValue,
278 filters: &[ValueFilter<'_>],
279 ) -> KevyResult<u64> {
280 let spec = self.claused_spec(name)?;
281 let opts = ScalarQueryOpts { filters, ..ScalarQueryOpts::default() };
282 let r = self.claused_resolve(name, &spec, &opts)?;
283 let mut total = 0u64;
284 #[cfg(not(target_arch = "wasm32"))]
285 let probe = self.usage_cell(name);
286 self.for_each_segment_windowed(name, |spec, seg, win| {
287 total += seg.count_claused(min, max, &r.filters);
288 #[cfg(not(target_arch = "wasm32"))]
289 if let Some(w) = win {
290 crate::ops_index::advise::probe_window(&probe, w, min);
291 }
292 // The evicted half counts from the cold payloads — same
293 // predicates, frozen values; a corrupt segment refuses.
294 #[cfg(not(target_arch = "wasm32"))]
295 if let Some(w) = win.filter(|w| w.has_cold()) {
296 total += w
297 .cold_claused_count(spec.ty(), min, max, &r.filters)
298 .map_err(|e| KevyError::Io(std::io::Error::other(e)))?;
299 }
300 #[cfg(target_arch = "wasm32")]
301 let _ = (spec, win);
302 Ok(())
303 })?;
304 self.observe_hit(name);
305 Ok(total)
306 }
307
308 /// The named index's spec, feeding the advise log (a Range
309 /// family) when the name is not declared.
310 fn claused_spec(&self, name: &[u8]) -> KevyResult<IndexSpec> {
311 let spec = {
312 let g = self.indexes.catalog.read().unwrap_or_else(std::sync::PoisonError::into_inner);
313 g.1.get(name).map(|(s, _)| s.clone())
314 };
315 spec.ok_or_else(|| {
316 self.observe_refused(name, kevy_index::AdviseShape::Range);
317 KevyError::NotFound("no such index".into())
318 })
319 }
320
321 /// [`resolve`], feeding the advise log when a clause named a
322 /// field the index does not store.
323 fn claused_resolve(
324 &self,
325 name: &[u8],
326 spec: &IndexSpec,
327 opts: &ScalarQueryOpts<'_>,
328 ) -> KevyResult<Resolved> {
329 resolve(spec, opts).inspect_err(|_| {
330 if let Some(f) = unstored_field(spec, opts) {
331 self.observe_refused(name, kevy_index::AdviseShape::Filter(f));
332 }
333 })
334 }
335
336 /// Every shard's claused page, unmerged, with the facet buckets
337 /// folded by identity as the shards report them.
338 fn gather_claused(
339 &self,
340 name: &[u8],
341 min: &IndexValue,
342 max: &IndexValue,
343 cursor: Option<&Cursor>,
344 clauses: &ScalarClauses<'_>,
345 ) -> KevyResult<GatheredPages> {
346 let mut all: Vec<(ScalarHit, ())> = Vec::new();
347 let mut facets: Vec<Vec<FacetBucket>> = vec![Vec::new(); clauses.facets.len()];
348 let mut found = false;
349 #[cfg(not(target_arch = "wasm32"))]
350 let probe = self.usage_cell(name);
351 for shard in self.shards.iter() {
352 let mut g = lock_write(shard);
353 let inner = &mut *g;
354 sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
355 if let Some((_spec, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name() == name) {
356 found = true;
357 #[cfg(not(target_arch = "wasm32"))]
358 if let Some(w) = inner.idx_segs.window_of(name) {
359 crate::ops_index::advise::probe_window(&probe, w, min);
360 }
361 let page = seg.query_claused(min, max, cursor, clauses);
362 all.extend(page.hits.into_iter().map(|h| (h, ())));
363 fold_facets(&mut facets, page.facets);
364 // The shard's evicted half joins the union — the
365 // origin merge below re-orders, re-collapses and
366 // truncates the lot, so cold hits need no shard-level
367 // pre-merge here.
368 #[cfg(not(target_arch = "wasm32"))]
369 if let Some(w) = inner.idx_segs.window_of(name).filter(|w| w.has_cold()) {
370 let (chits, cfacets) = w
371 .cold_claused(_spec.ty(), min, max, cursor, clauses)
372 .map_err(|e| KevyError::Io(std::io::Error::other(e)))?;
373 all.extend(chits.into_iter().map(|h| (h, ())));
374 fold_facets(&mut facets, cfacets);
375 }
376 }
377 }
378 if !found {
379 return Err(KevyError::NotFound("no such index".into()));
380 }
381 Ok((all, facets))
382 }
383}