kevy_embedded/
ops_table.rs

1//! Embedded `TABLE.*` capability — the declaration
2//! object, the compile call and the verify sweep. The grammar, the
3//! validation and [`kevy_index::compile_table`] all live in
4//! `kevy-index`: ONE implementation shared with the server, so the two
5//! engines cannot compile a table differently (the IDX.CREATE parity
6//! lesson; the dispatch oracle byte-compares the wire faces anyway).
7
8use std::sync::{Mutex, RwLock};
9
10use kevy_index::{
11    AdviseLog, IndexVerify, TableCatalog, TableEnsure, TableSpec, TableVerify, compile_table,
12};
13
14use crate::store::{Store, lock_write};
15use crate::{KevyError, KevyResult};
16
17/// Store-level table state (declarations only — a table's runtime
18/// footprint is its compiled indexes in the index registry).
19#[derive(Default)]
20pub(crate) struct TableReg {
21    pub(crate) catalog: RwLock<TableCatalog>,
22    /// The refusal log (the auto-declaration loop's observation
23    /// face) — fed by the typed query API's refusals, rendered by
24    /// [`Store::idx_advise`], cleared on every catalog mutation.
25    pub(crate) advise: Mutex<AdviseLog>,
26}
27
28/// Rows the per-shard column spot check samples (mirrors the server).
29const SPOTCHECK_ROWS: usize = 64;
30
31#[cfg(feature = "persist")]
32const SIDECAR: &str = "table-catalog.meta";
33
34/// One `TABLE.VERIFY` result: per compiled index its name + six
35/// counters (entries, bytes, coerce_failures, duplicates, drift,
36/// checked), plus the `(rows, type_mismatches)` spot-check pair.
37/// The legacy verify shape: six unnamed counters per index and two for
38/// the spot check. Kept for semver; superseded by [`TableVerify`],
39/// whose fields carry the names and time semantics these arrays never
40/// could (dogfood F10).
41#[deprecated(since = "4.1.0", note = "use `table_verify_report`, whose fields are named")]
42pub type TableVerifyReport = (Vec<(Vec<u8>, [u64; 6])>, [u64; 2]);
43
44impl Store {
45    /// `TABLE.DECLARE` equivalent: admit the declaration, then build
46    /// every compiled index synchronously. Atomic against catalog
47    /// errors: names are dry-run against a catalog clone first, so a
48    /// collision installs nothing.
49    pub fn table_declare(&self, spec: TableSpec) -> KevyResult<()> {
50        // Tiering floor refusal — the same precheck IDX.CREATE runs,
51        // moved ahead of the catalog mutation so a refused declare
52        // installs nothing.
53        #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
54        crate::ops_index_sync::tier_floor_check(&self.shards)?;
55        // compile_table validates for itself — a bad spec is a named
56        // refusal here, never a panic downstream (dogfood F9).
57        let compiled = compile_table(&spec).map_err(KevyError::InvalidInput)?;
58        {
59            let g = self
60                .tables
61                .catalog
62                .read()
63                .unwrap_or_else(std::sync::PoisonError::into_inner);
64            if g.get(&spec.name).is_some() {
65                return Err(KevyError::InvalidInput("table already exists".into()));
66            }
67        }
68        // Dry-run the compiled specs against a clone of the index
69        // catalog — the server admits into a clone and installs once;
70        // this is the same all-or-nothing shape.
71        {
72            let g = self
73                .indexes
74                .catalog
75                .read()
76                .unwrap_or_else(std::sync::PoisonError::into_inner);
77            let mut probe = g.1.clone();
78            for ispec in &compiled {
79                probe
80                    .create(ispec.clone())
81                    .map_err(|e| KevyError::InvalidInput(strip_err(e).into()))?;
82            }
83        }
84        {
85            let mut g = self
86                .tables
87                .catalog
88                .write()
89                .unwrap_or_else(std::sync::PoisonError::into_inner);
90            g.create(spec).map_err(|e| KevyError::InvalidInput(strip_err(&e).into()))?;
91        }
92        self.persist_table_sidecar();
93        for ispec in compiled {
94            self.register_spec(ispec)?;
95        }
96        self.advise_clear();
97        Ok(())
98    }
99
100    /// `TABLE.ENSURE` equivalent — the boot verb.
101    ///
102    /// The dogfood report's F8.2: declaring at boot is the steady
103    /// state, and plain `table_declare` punishes it — a re-declare is
104    /// an error, so the obvious code (declare, log the error at debug)
105    /// keeps running against whatever shape the *first* boot declared,
106    /// forever, silently. `ensure` gives the boot path its true verb:
107    /// an identical spec is a no-op success, a changed spec is a named
108    /// refusal carrying what changed — never a silent rebuild, because
109    /// a rebuild is a full backfill and must be asked for by name
110    /// ([`Self::table_replace`]).
111    pub fn table_ensure(&self, spec: TableSpec) -> KevyResult<TableEnsure> {
112        let existing = {
113            let g = self
114                .tables
115                .catalog
116                .read()
117                .unwrap_or_else(std::sync::PoisonError::into_inner);
118            g.get(&spec.name).cloned()
119        };
120        match existing {
121            None => {
122                self.table_declare(spec)?;
123                Ok(TableEnsure::Created)
124            }
125            Some(cur) if cur.sans_auto() == spec => Ok(TableEnsure::Unchanged),
126            Some(cur) => Err(KevyError::InvalidInput(
127                kevy_index::spec_diff(&cur.sans_auto(), &spec),
128            )),
129        }
130    }
131
132    /// `TABLE.REPLACE` equivalent — drop and redeclare, rebuilding
133    /// every compiled index from the rows. Explicitly named because it
134    /// is a full backfill: the cost is asked for, not implied. The new
135    /// spec is compiled (and therefore validated) *before* the old
136    /// table is dropped, so a bad replacement leaves the old one
137    /// standing.
138    pub fn table_replace(&self, spec: TableSpec) -> KevyResult<()> {
139        compile_table(&spec).map_err(KevyError::InvalidInput)?;
140        self.table_drop(&spec.name);
141        self.table_declare(spec)
142    }
143
144    /// `TABLE.DROP` equivalent — drops the table AND its compiled
145    /// indexes; `false` if absent.
146    pub fn table_drop(&self, name: &[u8]) -> bool {
147        let compiled: Vec<Vec<u8>> = {
148            let g = self
149                .tables
150                .catalog
151                .read()
152                .unwrap_or_else(std::sync::PoisonError::into_inner);
153            g.get(name)
154                .map(|s| {
155                    compile_table(s)
156                        .map(|c| c.into_iter().map(|i| i.name).collect())
157                        .unwrap_or_default() // catalog entries were admitted validated
158                })
159                .unwrap_or_default()
160        };
161        let hit = {
162            let mut g = self
163                .tables
164                .catalog
165                .write()
166                .unwrap_or_else(std::sync::PoisonError::into_inner);
167            g.drop_table(name)
168        };
169        if hit {
170            for iname in &compiled {
171                self.idx_drop(iname);
172            }
173            self.persist_table_sidecar();
174            self.advise_clear();
175        }
176        hit
177    }
178
179    /// Declared tables, declaration order.
180    pub fn table_list(&self) -> Vec<TableSpec> {
181        let g = self
182            .tables
183            .catalog
184            .read()
185            .unwrap_or_else(std::sync::PoisonError::into_inner);
186        g.iter().cloned().collect()
187    }
188
189    /// `TABLE.VERIFY` equivalent: each compiled index's full drift
190    /// recheck (the IDX.VERIFY discipline — composites re-derive their
191    /// byte encoding) plus a bounded column-type spot check, both
192    /// through the no-promote peek.
193    #[deprecated(since = "4.1.0", note = "use `table_verify_report`, whose fields are named")]
194    #[allow(deprecated)]
195    pub fn table_verify(&self, name: &[u8]) -> KevyResult<TableVerifyReport> {
196        let r = self.table_verify_report(name)?;
197        Ok((
198            r.per_index
199                .into_iter()
200                .map(|i| {
201                    (i.name, [i.entries, i.approx_bytes, i.coerce_failures, i.duplicates, i.drift, i.checked])
202                })
203                .collect(),
204            [r.spot_rows, r.spot_type_mismatches],
205        ))
206    }
207
208    /// `TABLE.VERIFY`, with every counter named and its time semantics
209    /// documented on the field (see [`IndexVerify`]).
210    pub fn table_verify_report(&self, name: &[u8]) -> KevyResult<TableVerify> {
211        let spec = {
212            let g = self
213                .tables
214                .catalog
215                .read()
216                .unwrap_or_else(std::sync::PoisonError::into_inner);
217            g.get(name).cloned()
218        }
219        .ok_or_else(|| KevyError::NotFound("no such table".into()))?;
220        let compiled = compile_table(&spec).map_err(KevyError::InvalidInput)?;
221        let mut per_index: Vec<(Vec<u8>, [u64; 10])> =
222            compiled.iter().map(|i| (i.name.clone(), [0u64; 10])).collect();
223        let mut spot = [0u64; 2];
224        for shard in self.shards.iter() {
225            let mut g = lock_write(shard);
226            let inner = &mut *g;
227            crate::ops_index_sync::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
228            for (slot, ispec) in per_index.iter_mut().zip(&compiled) {
229                shard_index_counts(inner, ispec, &mut slot.1);
230            }
231            let (r, m) = shard_spot_check(&mut inner.store, &spec);
232            spot[0] += r;
233            spot[1] += m;
234        }
235        Ok(TableVerify {
236            per_index: per_index
237                .into_iter()
238                .map(|(name, s)| IndexVerify {
239                    name,
240                    entries: s[0],
241                    approx_bytes: s[1],
242                    coerce_failures: s[2],
243                    duplicates: s[3],
244                    drift: s[4],
245                    checked: s[5],
246                    excluded: s[6],
247                    absent: s[7],
248                    rows: s[8],
249                    missing: s[9],
250                })
251                .collect(),
252            spot_rows: spot[0],
253            spot_type_mismatches: spot[1],
254        })
255    }
256
257    #[cfg(feature = "persist")]
258    pub(crate) fn table_boot(&self) {
259        let Some(dir) = &self.config.data_dir else { return };
260        if let Ok(text) = std::fs::read_to_string(dir.join(SIDECAR))
261            && let Some(cat) = TableCatalog::from_sidecar(&text)
262            && !cat.is_empty()
263        {
264            let mut g = self
265                .tables
266                .catalog
267                .write()
268                .unwrap_or_else(std::sync::PoisonError::into_inner);
269            *g = cat;
270        }
271    }
272
273    #[cfg(not(feature = "persist"))]
274    pub(crate) fn table_boot(&self) {}
275
276    #[cfg(feature = "persist")]
277    pub(crate) fn persist_table_sidecar(&self) {
278        let Some(dir) = &self.config.data_dir else { return };
279        let g = self
280            .tables
281            .catalog
282            .read()
283            .unwrap_or_else(std::sync::PoisonError::into_inner);
284        let tmp = dir.join("table-catalog.meta.tmp");
285        if std::fs::write(&tmp, g.to_sidecar()).is_ok() {
286            let _ = std::fs::rename(&tmp, dir.join(SIDECAR));
287        }
288    }
289
290    #[cfg(not(feature = "persist"))]
291    pub(crate) fn persist_table_sidecar(&self) {}
292}
293
294/// Catalog errors carry a leading `ERR ` for the wire; the typed
295/// `KevyError` face re-adds it, so strip here to avoid `ERR ERR`.
296fn strip_err(e: &str) -> &str {
297    e.strip_prefix("ERR ").unwrap_or(e)
298}
299
300/// One shard's contribution to one compiled index's verify counters —
301/// both directions.
302///
303/// index→row (`drift`): every held entry re-derives from its row.
304/// row→index: every prefix row classifies against the spec,
305/// with the cause kept — `absent` (NULL, by design), `coerce_failures`
306/// (present-but-wrong-type, fresh: the 4.0 lifetime counter of this
307/// name also swallowed absences), `excluded` (composite oversize, the
308/// silent two-row gap of dogfood F8/F9, now a named number), and
309/// `missing` (derives a value, has no entry — the class a drift walk
310/// structurally cannot see, and the one behind F13/F14's forgotten
311/// writers).
312///
313/// Layout: [entries, bytes, coerce_fresh, duplicates, drift, checked,
314///          excluded, absent, rows, missing].
315/// The row→index half of the walk, under the no-promote peek: every
316/// prefix row classified by cause (the server's `classify_prefix_rows`
317/// is the byte-parity twin).
318/// Returns `[coerce_fresh, excluded, absent, rows, missing]`.
319/// `window` carries the cold side's own count — the twin of the
320/// server's `cmd_table_verify::classify_prefix_rows`, including its
321/// reason: a row that slid is absent from the hot entries by design,
322/// but only if it actually reached a segment, so the rows below the
323/// boundary are reconciled against the live cold entries instead of
324/// being exempted for sitting there.
325fn classify_prefix_rows(
326    s: &mut kevy_store::Store,
327    spec: &kevy_index::IndexSpec,
328    row_keys: &[Vec<u8>],
329    indexed: &std::collections::HashSet<&[u8]>,
330    window: Option<kevy_index::WindowAudit>,
331) -> [u64; 5] {
332    let names = spec.scalar_read_names();
333    let w = spec.primary_width();
334    let mut f = [0u64; 5];
335    let mut below = 0u64;
336    for key in row_keys {
337        f[3] += 1;
338        let cls = match s.peek_hash_fields(key, &names[..w]) {
339            Ok(Some(vals)) => spec.classify_scalar(&vals),
340            // A non-hash or vanished row has no columns: NULL row.
341            _ => kevy_index::RowDerivation::Absent,
342        };
343        match cls {
344            kevy_index::RowDerivation::Indexed(_) => {
345                let slid = window.is_some_and(|wa| {
346                    let v = match s.peek_hash_fields(key, &names[..w]) {
347                        Ok(Some(vals)) => spec.derive_scalar(&vals),
348                        _ => None,
349                    };
350                    v.and_then(|v| kevy_index::window_value_of(&v, wa.shape))
351                        .is_some_and(|wv| wv < wa.boundary)
352                });
353                if !indexed.contains(key.as_slice()) {
354                    if slid {
355                        below += 1;
356                    } else {
357                        f[4] += 1;
358                    }
359                }
360            }
361            kevy_index::RowDerivation::CoerceFailed => f[0] += 1,
362            kevy_index::RowDerivation::Oversize => f[1] += 1,
363            kevy_index::RowDerivation::Absent => f[2] += 1,
364        }
365    }
366    // Every row that slid should have an entry waiting for it; the
367    // shortfall is the rows that fell between the tree and the segment.
368    f[4] += below.saturating_sub(window.map_or(0, |w| w.cold_live));
369    f
370}
371
372/// The window boundary a `missing` count must not fault rows against:
373/// a row below it slid into a cold segment on purpose. `None` when the
374/// path is unwindowed or has not slid yet (and always on wasm, where
375/// nothing slides).
376fn hot_floor_of(
377    inner: &crate::store_inner::Inner,
378    name: &[u8],
379    ty: kevy_index::ValType,
380) -> Option<kevy_index::WindowAudit> {
381    #[cfg(not(target_arch = "wasm32"))]
382    {
383        inner
384            .idx_segs
385            .windows
386            .iter()
387            .find(|(n, _)| n.as_slice() == name)
388            .and_then(|(_, w)| w.audit(ty))
389    }
390    #[cfg(target_arch = "wasm32")]
391    {
392        let _ = (inner, name, ty);
393        None
394    }
395}
396
397fn shard_index_counts(
398    inner: &mut crate::store_inner::Inner,
399    ispec: &kevy_index::IndexSpec,
400    sums: &mut [u64; 10],
401) {
402    let Some((spec, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name == ispec.name)
403    else {
404        return;
405    };
406    let stats = seg.stats();
407    let mut entries: Vec<(Vec<u8>, kevy_index::IndexValue)> = Vec::new();
408    seg.each_entry(|k, v| entries.push((k.to_vec(), v.clone())));
409    let indexed: std::collections::HashSet<&[u8]> =
410        entries.iter().map(|(k, _)| k.as_slice()).collect();
411    let spec = spec.clone();
412    let mut pat = spec.prefix.clone();
413    pat.push(b'*');
414    let row_keys = inner.store.collect_keys(Some(&pat), None);
415    let window = hot_floor_of(inner, &spec.name, spec.ty);
416    let store = &mut inner.store;
417    let (drift, fresh) = store.peek_scope(|s| {
418        let names = spec.scalar_read_names();
419        let w = spec.primary_width();
420        let mut drift = 0u64;
421        for (key, held) in &entries {
422            let actual = match s.peek_hash_fields(key, &names[..w]) {
423                Ok(Some(vals)) => spec.derive_scalar(&vals),
424                _ => None,
425            };
426            if actual.as_ref() != Some(held) {
427                drift += 1;
428            }
429        }
430        (drift, classify_prefix_rows(s, &spec, &row_keys, &indexed, window))
431    });
432    sums[0] += stats.entries;
433    sums[1] += stats.approx_bytes;
434    sums[2] += fresh[0];
435    sums[3] += stats.duplicates;
436    sums[4] += drift;
437    sums[5] += entries.len() as u64;
438    sums[6] += fresh[1];
439    sums[7] += fresh[2];
440    sums[8] += fresh[3];
441    sums[9] += fresh[4];
442}
443
444/// Sample up to [`SPOTCHECK_ROWS`] rows on one shard: every PRESENT
445/// declared-typed column must coerce (absent = NULL — Law 3; a
446/// non-hash row counts as a row with no columns).
447fn shard_spot_check(store: &mut kevy_store::Store, spec: &TableSpec) -> (u64, u64) {
448    let mut pat = spec.prefix.clone();
449    pat.push(b'*');
450    let keys = store.collect_keys(Some(&pat), Some(SPOTCHECK_ROWS));
451    let names: Vec<&[u8]> = spec.columns.iter().map(|(n, _)| n.as_slice()).collect();
452    store.peek_scope(|s| {
453        let (mut rows, mut mismatches) = (0u64, 0u64);
454        for key in &keys {
455            rows += 1;
456            let Ok(Some(vals)) = s.peek_hash_fields(key, &names) else { continue };
457            for ((_, ty), val) in spec.columns.iter().zip(&vals) {
458                if let Some(raw) = val
459                    && kevy_index::IndexValue::coerce(*ty, raw).is_none()
460                {
461                    mismatches += 1;
462                }
463            }
464        }
465        (rows, mismatches)
466    })
467}
kevy_embedded/ops_table.rs

kevy_embedded/
ops_table.rs
