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::{TableCatalog, TableSpec, 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.
31pub type TableVerifyReport = (Vec<(Vec<u8>, [u64; 6])>, [u64; 2]);
32
33impl Store {
34    /// `TABLE.DECLARE` equivalent: admit the declaration, then build
35    /// every compiled index synchronously. Atomic against catalog
36    /// errors: names are dry-run against a catalog clone first, so a
37    /// collision installs nothing.
38    pub fn table_declare(&self, spec: TableSpec) -> KevyResult<()> {
39        // Tiering floor refusal — the same precheck IDX.CREATE runs,
40        // moved ahead of the catalog mutation so a refused declare
41        // installs nothing.
42        #[cfg(all(feature = "tier", not(target_arch = "wasm32")))]
43        crate::ops_index_sync::tier_floor_check(&self.shards)?;
44        let compiled = compile_table(&spec);
45        {
46            let g = self
47                .tables
48                .catalog
49                .read()
50                .unwrap_or_else(std::sync::PoisonError::into_inner);
51            if g.get(&spec.name).is_some() {
52                return Err(KevyError::InvalidInput("table already exists".into()));
53            }
54        }
55        // Dry-run the compiled specs against a clone of the index
56        // catalog — the server admits into a clone and installs once;
57        // this is the same all-or-nothing shape.
58        {
59            let g = self
60                .indexes
61                .catalog
62                .read()
63                .unwrap_or_else(std::sync::PoisonError::into_inner);
64            let mut probe = g.1.clone();
65            for ispec in &compiled {
66                probe
67                    .create(ispec.clone())
68                    .map_err(|e| KevyError::InvalidInput(strip_err(e).into()))?;
69            }
70        }
71        {
72            let mut g = self
73                .tables
74                .catalog
75                .write()
76                .unwrap_or_else(std::sync::PoisonError::into_inner);
77            g.create(spec).map_err(|e| KevyError::InvalidInput(strip_err(&e).into()))?;
78        }
79        self.persist_table_sidecar();
80        for ispec in compiled {
81            self.register_spec(ispec)?;
82        }
83        Ok(())
84    }
85
86    /// `TABLE.DROP` equivalent — drops the table AND its compiled
87    /// indexes; `false` if absent.
88    pub fn table_drop(&self, name: &[u8]) -> bool {
89        let compiled: Vec<Vec<u8>> = {
90            let g = self
91                .tables
92                .catalog
93                .read()
94                .unwrap_or_else(std::sync::PoisonError::into_inner);
95            g.get(name)
96                .map(|s| compile_table(s).into_iter().map(|i| i.name).collect())
97                .unwrap_or_default()
98        };
99        let hit = {
100            let mut g = self
101                .tables
102                .catalog
103                .write()
104                .unwrap_or_else(std::sync::PoisonError::into_inner);
105            g.drop_table(name)
106        };
107        if hit {
108            for iname in &compiled {
109                self.idx_drop(iname);
110            }
111            self.persist_table_sidecar();
112        }
113        hit
114    }
115
116    /// Declared tables, declaration order.
117    pub fn table_list(&self) -> Vec<TableSpec> {
118        let g = self
119            .tables
120            .catalog
121            .read()
122            .unwrap_or_else(std::sync::PoisonError::into_inner);
123        g.iter().cloned().collect()
124    }
125
126    /// `TABLE.VERIFY` equivalent: each compiled index's full drift
127    /// recheck (the IDX.VERIFY discipline — composites re-derive their
128    /// byte encoding) plus a bounded column-type spot check, both
129    /// through the no-promote peek.
130    pub fn table_verify(&self, name: &[u8]) -> KevyResult<TableVerifyReport> {
131        let spec = {
132            let g = self
133                .tables
134                .catalog
135                .read()
136                .unwrap_or_else(std::sync::PoisonError::into_inner);
137            g.get(name).cloned()
138        }
139        .ok_or_else(|| KevyError::NotFound("no such table".into()))?;
140        let compiled = compile_table(&spec);
141        let mut per_index: Vec<(Vec<u8>, [u64; 6])> =
142            compiled.iter().map(|i| (i.name.clone(), [0u64; 6])).collect();
143        let mut spot = [0u64; 2];
144        for shard in self.shards.iter() {
145            let mut g = lock_write(shard);
146            let inner = &mut *g;
147            crate::ops_index_sync::sync_segs(&self.indexes, &mut inner.idx_segs, &mut inner.store);
148            for (slot, ispec) in per_index.iter_mut().zip(&compiled) {
149                shard_index_counts(inner, ispec, &mut slot.1);
150            }
151            let (r, m) = shard_spot_check(&mut inner.store, &spec);
152            spot[0] += r;
153            spot[1] += m;
154        }
155        Ok((per_index, spot))
156    }
157
158    #[cfg(feature = "persist")]
159    pub(crate) fn table_boot(&self) {
160        let Some(dir) = &self.config.data_dir else { return };
161        if let Ok(text) = std::fs::read_to_string(dir.join(SIDECAR))
162            && let Some(cat) = TableCatalog::from_sidecar(&text)
163            && !cat.is_empty()
164        {
165            let mut g = self
166                .tables
167                .catalog
168                .write()
169                .unwrap_or_else(std::sync::PoisonError::into_inner);
170            *g = cat;
171        }
172    }
173
174    #[cfg(not(feature = "persist"))]
175    pub(crate) fn table_boot(&self) {}
176
177    #[cfg(feature = "persist")]
178    fn persist_table_sidecar(&self) {
179        let Some(dir) = &self.config.data_dir else { return };
180        let g = self
181            .tables
182            .catalog
183            .read()
184            .unwrap_or_else(std::sync::PoisonError::into_inner);
185        let tmp = dir.join("table-catalog.meta.tmp");
186        if std::fs::write(&tmp, g.to_sidecar()).is_ok() {
187            let _ = std::fs::rename(&tmp, dir.join(SIDECAR));
188        }
189    }
190
191    #[cfg(not(feature = "persist"))]
192    fn persist_table_sidecar(&self) {}
193}
194
195/// Catalog errors carry a leading `ERR ` for the wire; the typed
196/// `KevyError` face re-adds it, so strip here to avoid `ERR ERR`.
197fn strip_err(e: &str) -> &str {
198    e.strip_prefix("ERR ").unwrap_or(e)
199}
200
201/// One shard's contribution to one compiled index's verify counters.
202fn shard_index_counts(
203    inner: &mut crate::store_inner::Inner,
204    ispec: &kevy_index::IndexSpec,
205    sums: &mut [u64; 6],
206) {
207    let Some((spec, seg)) = inner.idx_segs.segs.iter().find(|(s, _)| s.name == ispec.name)
208    else {
209        return;
210    };
211    let stats = seg.stats();
212    let mut entries: Vec<(Vec<u8>, kevy_index::IndexValue)> = Vec::new();
213    seg.each_entry(|k, v| entries.push((k.to_vec(), v.clone())));
214    let spec = spec.clone();
215    let store = &mut inner.store;
216    let drift = store.peek_scope(|s| {
217        let names = spec.scalar_read_names();
218        let w = spec.primary_width();
219        let mut drift = 0u64;
220        for (key, held) in &entries {
221            let actual = match s.peek_hash_fields(key, &names[..w]) {
222                Ok(Some(vals)) => spec.derive_scalar(&vals),
223                _ => None,
224            };
225            if actual.as_ref() != Some(held) {
226                drift += 1;
227            }
228        }
229        drift
230    });
231    sums[0] += stats.entries;
232    sums[1] += stats.approx_bytes;
233    sums[2] += stats.coerce_failures;
234    sums[3] += stats.duplicates;
235    sums[4] += drift;
236    sums[5] += entries.len() as u64;
237}
238
239/// Sample up to [`SPOTCHECK_ROWS`] rows on one shard: every PRESENT
240/// declared-typed column must coerce (absent = NULL — Law 3; a
241/// non-hash row counts as a row with no columns).
242fn shard_spot_check(store: &mut kevy_store::Store, spec: &TableSpec) -> (u64, u64) {
243    let mut pat = spec.prefix.clone();
244    pat.push(b'*');
245    let keys = store.collect_keys(Some(&pat), Some(SPOTCHECK_ROWS));
246    let names: Vec<&[u8]> = spec.columns.iter().map(|(n, _)| n.as_slice()).collect();
247    store.peek_scope(|s| {
248        let (mut rows, mut mismatches) = (0u64, 0u64);
249        for key in &keys {
250            rows += 1;
251            let Ok(Some(vals)) = s.peek_hash_fields(key, &names) else { continue };
252            for ((_, ty), val) in spec.columns.iter().zip(&vals) {
253                if let Some(raw) = val
254                    && kevy_index::IndexValue::coerce(*ty, raw).is_none()
255                {
256                    mismatches += 1;
257                }
258            }
259        }
260        (rows, mismatches)
261    })
262}
kevy_embedded/ops_table.rs

kevy_embedded/
ops_table.rs
