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

kevy_embedded/
ops_table.rs
