corium_db/
lib.rs

1//! Immutable database values: time views, covering-index access, naming,
2//! per-attribute statistics, and bootstrap metadata.
3//!
4//! A [`Db`] is a value: cheap to clone, never mutated in place. Time views
5//! ([`Db::as_of`], [`Db::since`], [`Db::history`]) wrap the same recorded
6//! datoms with a different fold policy — no copying of facts. The four
7//! covering indexes for a view are materialized lazily on first read and
8//! shared by every clone of that value.
9
10use std::collections::{BTreeMap, BTreeSet};
11use std::sync::{Arc, OnceLock};
12
13use corium_core::{
14    AttrId, Cardinality, Datom, EntityId, IndexOrder, Keyword, KeywordInterner, Partition, Schema,
15    Unique, Value, encoding::Encodable,
16};
17
18/// The first user-assignable sequence number. Lower ids are reserved for bootstrap data.
19pub const FIRST_USER_ID: u64 = 1_000;
20
21/// Time-view selector for a database value (see `docs/design/time-model.md`).
22#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
23pub enum DbView {
24    /// Live facts as of the basis transaction.
25    #[default]
26    Current,
27    /// Facts as they stood at basis `t` (inclusive).
28    AsOf(u64),
29    /// Only live facts added after `t` (exclusive).
30    Since(u64),
31    /// Every assertion and retraction ever recorded, except `:db/noHistory` attributes.
32    History,
33}
34
35/// Registry of `:db/ident` names for entities (attributes chiefly).
36#[derive(Clone, Debug, Default)]
37pub struct Idents {
38    by_keyword: BTreeMap<Keyword, EntityId>,
39    by_id: BTreeMap<EntityId, Keyword>,
40}
41
42impl Idents {
43    /// Registers an ident for an entity.
44    pub fn insert(&mut self, keyword: Keyword, id: EntityId) {
45        self.by_id.insert(id, keyword.clone());
46        self.by_keyword.insert(keyword, id);
47    }
48
49    /// Resolves a keyword to its entity id.
50    #[must_use]
51    pub fn entid(&self, keyword: &Keyword) -> Option<EntityId> {
52        self.by_keyword.get(keyword).copied()
53    }
54
55    /// Resolves an entity id back to its ident.
56    #[must_use]
57    pub fn ident(&self, id: EntityId) -> Option<&Keyword> {
58        self.by_id.get(&id)
59    }
60
61    /// Iterates all registered idents in keyword order.
62    pub fn iter(&self) -> impl Iterator<Item = (&Keyword, &EntityId)> {
63        self.by_keyword.iter()
64    }
65}
66
67/// Per-attribute statistics driving planner selectivity estimates.
68#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
69pub struct AttrStats {
70    /// Datoms carrying this attribute in the view.
71    pub count: usize,
72    /// Distinct values for this attribute.
73    pub distinct_values: usize,
74    /// Distinct entities carrying this attribute.
75    pub distinct_entities: usize,
76}
77
78/// Whole-view statistics for the query planner.
79#[derive(Clone, Debug, Default)]
80pub struct PlannerStats {
81    /// Statistics per attribute.
82    pub per_attr: BTreeMap<AttrId, AttrStats>,
83    /// Total datoms in the view.
84    pub total_datoms: usize,
85    /// Distinct entities in the view.
86    pub entity_count: usize,
87}
88
89impl PlannerStats {
90    /// Estimated datoms matched by a scan with the given bound components.
91    #[must_use]
92    pub fn estimate(&self, e_bound: bool, a: Option<AttrId>, v_bound: bool) -> usize {
93        let attr = a.and_then(|a| self.per_attr.get(&a));
94        match (e_bound, attr) {
95            // Bound entity: at most the entity's datoms; refine by attribute.
96            (true, Some(stats)) => (stats.count / stats.distinct_entities.max(1)).max(1),
97            (true, None) => (self.total_datoms / self.entity_count.max(1)).max(1),
98            (false, Some(stats)) if v_bound => (stats.count / stats.distinct_values.max(1)).max(1),
99            (false, Some(stats)) => stats.count.max(1),
100            // Unknown attribute constant: nothing will match.
101            (false, None) if a.is_some() => 1,
102            (false, None) => self.total_datoms.max(1),
103        }
104    }
105}
106
107type Index = BTreeMap<Vec<u8>, Datom>;
108
109const ORDERS: [IndexOrder; 4] = [
110    IndexOrder::Eavt,
111    IndexOrder::Aevt,
112    IndexOrder::Avet,
113    IndexOrder::Vaet,
114];
115
116const fn slot(order: IndexOrder) -> usize {
117    match order {
118        IndexOrder::Eavt => 0,
119        IndexOrder::Aevt => 1,
120        IndexOrder::Avet => 2,
121        IndexOrder::Vaet => 3,
122    }
123}
124
125/// Builds the encoded key prefix for a partial datom in one index order.
126///
127/// Components are consumed in the index's component order and encoding stops
128/// at the first missing component, so the result is a proper range prefix.
129#[must_use]
130pub fn key_prefix(
131    order: IndexOrder,
132    e: Option<EntityId>,
133    a: Option<AttrId>,
134    v: Option<&Value>,
135) -> Vec<u8> {
136    enum C {
137        E,
138        A,
139        V,
140    }
141    let components = match order {
142        IndexOrder::Eavt => [C::E, C::A, C::V],
143        IndexOrder::Aevt => [C::A, C::E, C::V],
144        IndexOrder::Avet => [C::A, C::V, C::E],
145        IndexOrder::Vaet => [C::V, C::A, C::E],
146    };
147    let mut out = Vec::new();
148    for component in components {
149        match component {
150            C::E => match e {
151                Some(e) => e.encode_into(&mut out),
152                None => break,
153            },
154            C::A => match a {
155                Some(a) => a.encode_into(&mut out),
156                None => break,
157            },
158            C::V => match v {
159                Some(v) => v.encode_into(&mut out),
160                None => break,
161            },
162        }
163    }
164    out
165}
166
167/// An immutable value of a database at one basis transaction and time view.
168#[derive(Clone, Debug, Default)]
169pub struct Db {
170    basis_t: u64,
171    schema: Schema,
172    recorded: Arc<Vec<Datom>>,
173    idents: Arc<Idents>,
174    interner: Arc<KeywordInterner>,
175    view: DbView,
176    indexes: Arc<OnceLock<[Index; 4]>>,
177    stats: Arc<OnceLock<PlannerStats>>,
178}
179
180impl Db {
181    /// Creates an empty database with the supplied schema.
182    #[must_use]
183    pub fn new(schema: Schema) -> Self {
184        Self {
185            schema,
186            ..Self::default()
187        }
188    }
189
190    /// Attaches ident and keyword naming registries, returning the named value.
191    #[must_use]
192    pub fn with_naming(mut self, idents: Idents, interner: KeywordInterner) -> Self {
193        self.idents = Arc::new(idents);
194        self.interner = Arc::new(interner);
195        self
196    }
197
198    /// Current transaction basis.
199    #[must_use]
200    pub const fn basis_t(&self) -> u64 {
201        self.basis_t
202    }
203
204    /// Schema at this basis.
205    #[must_use]
206    pub const fn schema(&self) -> &Schema {
207        &self.schema
208    }
209
210    /// Ident registry.
211    #[must_use]
212    pub fn idents(&self) -> &Idents {
213        &self.idents
214    }
215
216    /// Keyword interner used by keyword values in this database.
217    #[must_use]
218    pub fn interner(&self) -> &KeywordInterner {
219        &self.interner
220    }
221
222    /// The time view this value presents.
223    #[must_use]
224    pub const fn view(&self) -> DbView {
225        self.view
226    }
227
228    /// Every recorded assertion and retraction, in transaction order.
229    #[must_use]
230    pub fn recorded_datoms(&self) -> &[Datom] {
231        &self.recorded
232    }
233
234    /// Returns the as-of view at basis `t`: facts as they stood then.
235    #[must_use]
236    pub fn as_of(&self, t: u64) -> Self {
237        self.with_view(DbView::AsOf(t))
238    }
239
240    /// Returns the since view: only live facts added after `t`.
241    #[must_use]
242    pub fn since(&self, t: u64) -> Self {
243        self.with_view(DbView::Since(t))
244    }
245
246    /// Returns the history view: all assertions and retractions ever.
247    #[must_use]
248    pub fn history(&self) -> Self {
249        self.with_view(DbView::History)
250    }
251
252    fn with_view(&self, view: DbView) -> Self {
253        if view == self.view {
254            return self.clone();
255        }
256        Self {
257            view,
258            indexes: Arc::new(OnceLock::new()),
259            stats: Arc::new(OnceLock::new()),
260            ..self.clone()
261        }
262    }
263
264    /// Groups recorded datoms by transaction over the half-open range `[start, end)`.
265    #[must_use]
266    pub fn tx_range(&self, start: u64, end: Option<u64>) -> Vec<(u64, Vec<Datom>)> {
267        let mut by_t: BTreeMap<u64, Vec<Datom>> = BTreeMap::new();
268        for datom in self.recorded.iter() {
269            let t = datom.tx.sequence();
270            if t >= start && end.is_none_or(|end| t < end) {
271                by_t.entry(t).or_default().push(datom.clone());
272            }
273        }
274        by_t.into_iter().collect()
275    }
276
277    /// Returns this view's facts, deterministically ordered by EAVT.
278    #[must_use]
279    pub fn datoms(&self) -> Vec<Datom> {
280        self.datoms_at(IndexOrder::Eavt).cloned().collect()
281    }
282
283    /// Iterates this view's datoms in one index order.
284    ///
285    /// AVET covers only indexed/unique attributes and VAET only reference
286    /// values, mirroring Datomic's covering-index composition.
287    pub fn datoms_at(&self, order: IndexOrder) -> impl Iterator<Item = &Datom> {
288        self.indexes()[slot(order)].values()
289    }
290
291    /// Iterates datoms whose key in `order` starts with `prefix`.
292    pub fn datoms_prefix<'a>(
293        &'a self,
294        order: IndexOrder,
295        prefix: &'a [u8],
296    ) -> impl Iterator<Item = &'a Datom> {
297        self.indexes()[slot(order)]
298            .range(prefix.to_vec()..)
299            .take_while(move |(key, _)| key.starts_with(prefix))
300            .map(|(_, datom)| datom)
301    }
302
303    /// Iterates datoms in `order` starting from the first key at or after `start`.
304    pub fn seek_datoms<'a>(
305        &'a self,
306        order: IndexOrder,
307        start: &[u8],
308    ) -> impl Iterator<Item = &'a Datom> {
309        self.indexes()[slot(order)]
310            .range(start.to_vec()..)
311            .map(|(_, datom)| datom)
312    }
313
314    /// Iterates the AVET index for `a` over the value range `[start, end)`.
315    ///
316    /// Only indexed/unique attributes appear in AVET.
317    pub fn index_range<'a>(
318        &'a self,
319        a: AttrId,
320        start: Option<&Value>,
321        end: Option<&'a Value>,
322    ) -> impl Iterator<Item = &'a Datom> {
323        let a_prefix = key_prefix(IndexOrder::Avet, None, Some(a), None);
324        let start_key = key_prefix(IndexOrder::Avet, None, Some(a), start);
325        self.indexes()[slot(IndexOrder::Avet)]
326            .range(start_key..)
327            .take_while(move |(key, _)| key.starts_with(&a_prefix))
328            .map(|(_, datom)| datom)
329            .take_while(move |datom| end.is_none_or(|end| datom.v < *end))
330    }
331
332    /// Current values for an entity/attribute pair.
333    #[must_use]
334    pub fn values(&self, e: EntityId, a: AttrId) -> Vec<Value> {
335        let prefix = key_prefix(IndexOrder::Eavt, Some(e), Some(a), None);
336        self.datoms_prefix(IndexOrder::Eavt, &prefix)
337            .map(|datom| datom.v.clone())
338            .collect()
339    }
340
341    /// Resolves a unique attribute/value pair to its entity.
342    #[must_use]
343    pub fn lookup(&self, a: AttrId, v: &Value) -> Option<EntityId> {
344        if avet_covered(&self.schema, a) {
345            let prefix = key_prefix(IndexOrder::Avet, None, Some(a), Some(v));
346            self.datoms_prefix(IndexOrder::Avet, &prefix)
347                .next()
348                .map(|datom| datom.e)
349        } else {
350            let prefix = key_prefix(IndexOrder::Aevt, None, Some(a), None);
351            self.datoms_prefix(IndexOrder::Aevt, &prefix)
352                .find(|datom| datom.v == *v)
353                .map(|datom| datom.e)
354        }
355    }
356
357    /// Applies a committed record, returning a new database value.
358    ///
359    /// Only meaningful for the current view; time views are read-only.
360    #[must_use]
361    pub fn with_transaction(&self, t: u64, datoms: &[Datom]) -> Self {
362        debug_assert!(
363            self.view == DbView::Current,
364            "with_transaction applies only to the current view"
365        );
366        let mut next = self.clone();
367        next.basis_t = t;
368        Arc::make_mut(&mut next.recorded).extend_from_slice(datoms);
369        next.indexes = Arc::new(OnceLock::new());
370        next.stats = Arc::new(OnceLock::new());
371        // Derive indexes incrementally when the parent already built them, so
372        // transaction pipelines don't refold the whole history per operation.
373        if let Some(parent) = self.indexes.get() {
374            let mut derived = parent.clone();
375            apply_current(&mut derived, datoms.iter(), &self.schema);
376            let _ = next.indexes.set(derived);
377        }
378        next
379    }
380
381    /// Computes basic statistics over this view's facts.
382    #[must_use]
383    pub fn stats(&self) -> DbStats {
384        let planner = self.planner_stats();
385        DbStats {
386            datoms: planner.total_datoms,
387            entities: planner.entity_count,
388            attributes: planner.per_attr.len(),
389        }
390    }
391
392    /// Planner statistics for this view, built lazily and cached.
393    #[must_use]
394    pub fn planner_stats(&self) -> &PlannerStats {
395        self.stats.get_or_init(|| {
396            let mut stats = PlannerStats::default();
397            let mut values: BTreeMap<AttrId, BTreeSet<&Value>> = BTreeMap::new();
398            let mut attr_entities: BTreeMap<AttrId, BTreeSet<EntityId>> = BTreeMap::new();
399            let mut entities: BTreeSet<EntityId> = BTreeSet::new();
400            for datom in self.datoms_at(IndexOrder::Eavt) {
401                stats.total_datoms += 1;
402                stats.per_attr.entry(datom.a).or_default().count += 1;
403                values.entry(datom.a).or_default().insert(&datom.v);
404                attr_entities.entry(datom.a).or_default().insert(datom.e);
405                entities.insert(datom.e);
406            }
407            for (a, entry) in &mut stats.per_attr {
408                entry.distinct_values = values.get(a).map_or(0, BTreeSet::len);
409                entry.distinct_entities = attr_entities.get(a).map_or(0, BTreeSet::len);
410            }
411            stats.entity_count = entities.len();
412            stats
413        })
414    }
415
416    fn indexes(&self) -> &[Index; 4] {
417        self.indexes.get_or_init(|| {
418            let mut indexes: [Index; 4] = Default::default();
419            match self.view {
420                DbView::History => {
421                    for datom in self.recorded.iter() {
422                        if self.schema.get(datom.a).is_some_and(|a| a.no_history) {
423                            continue;
424                        }
425                        insert_datom(&mut indexes, datom, &self.schema, true);
426                    }
427                }
428                DbView::Current | DbView::AsOf(_) | DbView::Since(_) => {
429                    let cutoff = match self.view {
430                        DbView::AsOf(t) => Some(t),
431                        _ => None,
432                    };
433                    let filtered = self
434                        .recorded
435                        .iter()
436                        .filter(|d| cutoff.is_none_or(|t| d.tx.sequence() <= t));
437                    apply_current(&mut indexes, filtered, &self.schema);
438                    if let DbView::Since(t) = self.view {
439                        for index in &mut indexes {
440                            index.retain(|_, datom| datom.tx.sequence() > t);
441                        }
442                    }
443                }
444            }
445            indexes
446        })
447    }
448}
449
450/// Folds assertions/retractions into current-view indexes.
451///
452/// Current views key entries by components only (no transaction suffix):
453/// at most one live datom exists per `(e, a, v)`, and retractions must
454/// erase the assertion regardless of which transactions produced them.
455fn apply_current<'a>(
456    indexes: &mut [Index; 4],
457    datoms: impl Iterator<Item = &'a Datom>,
458    schema: &Schema,
459) {
460    for datom in datoms {
461        if datom.added {
462            insert_datom(indexes, datom, schema, false);
463        } else {
464            for order in ORDERS {
465                if covered(schema, order, datom) {
466                    let key = key_prefix(order, Some(datom.e), Some(datom.a), Some(&datom.v));
467                    indexes[slot(order)].remove(&key);
468                }
469            }
470        }
471    }
472}
473
474fn insert_datom(indexes: &mut [Index; 4], datom: &Datom, schema: &Schema, with_tx: bool) {
475    for order in ORDERS {
476        if covered(schema, order, datom) {
477            let key = if with_tx {
478                datom.key(order)
479            } else {
480                key_prefix(order, Some(datom.e), Some(datom.a), Some(&datom.v))
481            };
482            indexes[slot(order)].insert(key, datom.clone());
483        }
484    }
485}
486
487fn covered(schema: &Schema, order: IndexOrder, datom: &Datom) -> bool {
488    match order {
489        IndexOrder::Eavt | IndexOrder::Aevt => true,
490        IndexOrder::Avet => avet_covered(schema, datom.a),
491        IndexOrder::Vaet => matches!(datom.v, Value::Ref(_)),
492    }
493}
494
495/// Whether the attribute participates in the AVET covering index.
496#[must_use]
497pub fn avet_covered(schema: &Schema, a: AttrId) -> bool {
498    schema
499        .get(a)
500        .is_some_and(|attr| attr.indexed || attr.unique.is_some())
501}
502
503/// Counts over one database view.
504#[derive(Clone, Copy, Debug, Eq, PartialEq)]
505pub struct DbStats {
506    /// Facts in the view.
507    pub datoms: usize,
508    /// Entities having at least one fact in the view.
509    pub entities: usize,
510    /// Attributes used by facts in the view.
511    pub attributes: usize,
512}
513
514/// Convenience constructor for schema attributes used during bootstrap/tests.
515#[must_use]
516pub const fn attribute(
517    id: u64,
518    value_type: corium_core::ValueType,
519    cardinality: Cardinality,
520    unique: Option<Unique>,
521) -> corium_core::Attribute {
522    corium_core::Attribute {
523        id: EntityId::new(Partition::Db as u32, id),
524        value_type,
525        cardinality,
526        unique,
527        is_component: false,
528        indexed: unique.is_some(),
529        no_history: false,
530    }
531}
532
533#[cfg(test)]
534mod tests {
535    use super::*;
536    use corium_core::ValueType;
537
538    fn schema() -> Schema {
539        let mut schema = Schema::default();
540        schema.insert(attribute(1, ValueType::Str, Cardinality::One, None));
541        schema.insert(attribute(
542            2,
543            ValueType::Long,
544            Cardinality::One,
545            Some(Unique::Identity),
546        ));
547        schema.insert(attribute(3, ValueType::Ref, Cardinality::Many, None));
548        schema
549    }
550
551    fn attr(id: u64) -> AttrId {
552        EntityId::new(Partition::Db as u32, id)
553    }
554
555    fn entity(id: u64) -> EntityId {
556        EntityId::new(Partition::User as u32, id)
557    }
558
559    fn tx_entity(t: u64) -> EntityId {
560        EntityId::new(Partition::Tx as u32, t)
561    }
562
563    fn datom(e: u64, a: u64, v: Value, t: u64, added: bool) -> Datom {
564        Datom {
565            e: entity(e),
566            a: attr(a),
567            v,
568            tx: tx_entity(t),
569            added,
570        }
571    }
572
573    fn sample() -> Db {
574        Db::new(schema())
575            .with_transaction(
576                1,
577                &[
578                    datom(1, 1, Value::Str("alice".into()), 1, true),
579                    datom(1, 2, Value::Long(7), 1, true),
580                ],
581            )
582            .with_transaction(
583                2,
584                &[
585                    datom(1, 1, Value::Str("alice".into()), 2, false),
586                    datom(1, 1, Value::Str("alicia".into()), 2, true),
587                    datom(2, 3, Value::Ref(entity(1)), 2, true),
588                ],
589            )
590    }
591
592    #[test]
593    fn current_view_folds_retractions() {
594        let db = sample();
595        assert_eq!(
596            db.values(entity(1), attr(1)),
597            vec![Value::Str("alicia".into())]
598        );
599        assert_eq!(db.stats().datoms, 3);
600    }
601
602    #[test]
603    fn as_of_reconstructs_past_basis() {
604        let db = sample().as_of(1);
605        assert_eq!(
606            db.values(entity(1), attr(1)),
607            vec![Value::Str("alice".into())]
608        );
609        assert_eq!(db.stats().datoms, 2);
610    }
611
612    #[test]
613    fn since_excludes_older_live_facts() {
614        let db = sample().since(1);
615        // The long asserted at t=1 is invisible; the renamed string is visible.
616        assert_eq!(db.values(entity(1), attr(2)), Vec::<Value>::new());
617        assert_eq!(
618            db.values(entity(1), attr(1)),
619            vec![Value::Str("alicia".into())]
620        );
621    }
622
623    #[test]
624    fn history_exposes_assertions_and_retractions() {
625        let db = sample().history();
626        let names: Vec<_> = db
627            .datoms_prefix(
628                IndexOrder::Eavt,
629                &key_prefix(IndexOrder::Eavt, Some(entity(1)), Some(attr(1)), None),
630            )
631            .map(|d| (d.v.clone(), d.added))
632            .collect();
633        assert_eq!(names.len(), 3);
634        assert!(names.contains(&(Value::Str("alice".into()), false)));
635    }
636
637    #[test]
638    fn avet_only_covers_indexed_attributes() {
639        let db = sample();
640        assert_eq!(db.datoms_at(IndexOrder::Avet).count(), 1);
641        assert_eq!(db.datoms_at(IndexOrder::Vaet).count(), 1);
642    }
643
644    #[test]
645    fn index_range_scans_value_bounds() {
646        let db = Db::new(schema()).with_transaction(
647            1,
648            &[
649                datom(1, 2, Value::Long(1), 1, true),
650                datom(2, 2, Value::Long(5), 1, true),
651                datom(3, 2, Value::Long(9), 1, true),
652            ],
653        );
654        let hits: Vec<_> = db
655            .index_range(attr(2), Some(&Value::Long(2)), Some(&Value::Long(9)))
656            .map(|d| d.e)
657            .collect();
658        assert_eq!(hits, vec![entity(2)]);
659    }
660
661    #[test]
662    fn tx_range_groups_by_transaction() {
663        let ranged = sample().tx_range(2, None);
664        assert_eq!(ranged.len(), 1);
665        assert_eq!(ranged[0].0, 2);
666        assert_eq!(ranged[0].1.len(), 3);
667    }
668
669    #[test]
670    fn incremental_indexes_match_rebuilt_indexes() {
671        let base = Db::new(schema());
672        let tx1 = [datom(1, 1, Value::Str("a".into()), 1, true)];
673        let tx2 = [
674            datom(1, 1, Value::Str("a".into()), 2, false),
675            datom(1, 1, Value::Str("b".into()), 2, true),
676        ];
677        // Force the parent cache so with_transaction derives incrementally.
678        let warm = base.with_transaction(1, &tx1);
679        let _ = warm.datoms();
680        let incremental = warm.with_transaction(2, &tx2);
681        let cold = base.with_transaction(1, &tx1).with_transaction(2, &tx2);
682        assert_eq!(incremental.datoms(), cold.datoms());
683    }
684
685    #[test]
686    fn planner_stats_count_attributes() {
687        let stats_owner = sample();
688        let stats = stats_owner.planner_stats();
689        assert_eq!(stats.total_datoms, 3);
690        assert_eq!(stats.per_attr[&attr(1)].count, 1);
691        assert!(stats.estimate(false, Some(attr(1)), false) <= stats.total_datoms);
692    }
693}
corium_db/lib.rs

corium_db/
lib.rs
