Skip to main content

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//!
10//! Transaction time is data: every committed transaction asserts
11//! `:db/txInstant` on its transaction entity (see [`bootstrap`]), so views can
12//! also be named by wall clock ([`Db::as_of_instant`], [`Db::since_instant`]).
13
14pub mod bootstrap;
15
16use std::collections::{BTreeMap, BTreeSet};
17use std::sync::{Arc, OnceLock};
18
19use corium_core::{
20    AttrId, Cardinality, Datom, EntityId, IndexOrder, Keyword, KeywordInterner, Partition, Schema,
21    Unique, Value, encoding::Encodable,
22};
23use rpds::{RedBlackTreeMapSync, VectorSync};
24
25/// The first user-assignable sequence number. Lower ids are reserved for bootstrap data.
26pub const FIRST_USER_ID: u64 = 1_000;
27
28/// Time-view selector for a database value (see `docs/design/time-model.md`).
29#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
30pub enum DbView {
31    /// Live facts as of the basis transaction.
32    #[default]
33    Current,
34    /// Facts as they stood at basis `t` (inclusive).
35    AsOf(u64),
36    /// Only live facts added after `t` (exclusive).
37    Since(u64),
38    /// Every assertion and retraction ever recorded, except `:db/noHistory` attributes.
39    History,
40}
41
42/// Registry of `:db/ident` names for entities (attributes chiefly).
43#[derive(Clone, Debug, Default)]
44pub struct Idents {
45    by_keyword: BTreeMap<Keyword, EntityId>,
46    by_id: BTreeMap<EntityId, Keyword>,
47}
48
49impl Idents {
50    /// Registers an ident for an entity.
51    pub fn insert(&mut self, keyword: Keyword, id: EntityId) {
52        self.by_id.insert(id, keyword.clone());
53        self.by_keyword.insert(keyword, id);
54    }
55
56    /// Resolves a keyword to its entity id.
57    #[must_use]
58    pub fn entid(&self, keyword: &Keyword) -> Option<EntityId> {
59        self.by_keyword.get(keyword).copied()
60    }
61
62    /// Resolves an entity id back to its ident.
63    #[must_use]
64    pub fn ident(&self, id: EntityId) -> Option<&Keyword> {
65        self.by_id.get(&id)
66    }
67
68    /// Iterates all registered idents in keyword order.
69    pub fn iter(&self) -> impl Iterator<Item = (&Keyword, &EntityId)> {
70        self.by_keyword.iter()
71    }
72}
73
74/// Per-attribute statistics driving planner selectivity estimates.
75#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
76pub struct AttrStats {
77    /// Datoms carrying this attribute in the view.
78    pub count: usize,
79    /// Distinct values for this attribute.
80    pub distinct_values: usize,
81    /// Distinct entities carrying this attribute.
82    pub distinct_entities: usize,
83}
84
85/// Whole-view statistics for the query planner.
86#[derive(Clone, Debug, Default)]
87pub struct PlannerStats {
88    /// Statistics per attribute.
89    pub per_attr: BTreeMap<AttrId, AttrStats>,
90    /// Total datoms in the view.
91    pub total_datoms: usize,
92    /// Distinct entities in the view.
93    pub entity_count: usize,
94}
95
96impl PlannerStats {
97    /// Estimated datoms matched by a scan with the given bound components.
98    #[must_use]
99    pub fn estimate(&self, e_bound: bool, a: Option<AttrId>, v_bound: bool) -> usize {
100        let attr = a.and_then(|a| self.per_attr.get(&a));
101        match (e_bound, attr) {
102            // Bound entity: at most the entity's datoms; refine by attribute.
103            (true, Some(stats)) => (stats.count / stats.distinct_entities.max(1)).max(1),
104            (true, None) => (self.total_datoms / self.entity_count.max(1)).max(1),
105            (false, Some(stats)) if v_bound => (stats.count / stats.distinct_values.max(1)).max(1),
106            (false, Some(stats)) => stats.count.max(1),
107            // Unknown attribute constant: nothing will match.
108            (false, None) if a.is_some() => 1,
109            (false, None) => self.total_datoms.max(1),
110        }
111    }
112}
113
114/// Covering index for one order.
115///
116/// A persistent (structurally shared) ordered map: cloning is O(1) and shares
117/// every unchanged node with the parent, so applying a transaction to a
118/// materialized index derives a new map that copies only the O(log n) nodes on
119/// the touched paths instead of the whole tree. This is what keeps
120/// `with_transaction` — and the per-operation folding inside `corium-tx` —
121/// from re-cloning the entire index on every write.
122type Index = RedBlackTreeMapSync<Vec<u8>, Datom>;
123
124const ORDERS: [IndexOrder; 4] = [
125    IndexOrder::Eavt,
126    IndexOrder::Aevt,
127    IndexOrder::Avet,
128    IndexOrder::Vaet,
129];
130
131const fn slot(order: IndexOrder) -> usize {
132    match order {
133        IndexOrder::Eavt => 0,
134        IndexOrder::Aevt => 1,
135        IndexOrder::Avet => 2,
136        IndexOrder::Vaet => 3,
137    }
138}
139
140/// Builds the encoded key prefix for a partial datom in one index order.
141///
142/// Components are consumed in the index's component order and encoding stops
143/// at the first missing component, so the result is a proper range prefix.
144#[must_use]
145pub fn key_prefix(
146    order: IndexOrder,
147    e: Option<EntityId>,
148    a: Option<AttrId>,
149    v: Option<&Value>,
150) -> Vec<u8> {
151    enum C {
152        E,
153        A,
154        V,
155    }
156    let components = match order {
157        IndexOrder::Eavt => [C::E, C::A, C::V],
158        IndexOrder::Aevt => [C::A, C::E, C::V],
159        IndexOrder::Avet => [C::A, C::V, C::E],
160        IndexOrder::Vaet => [C::V, C::A, C::E],
161    };
162    let mut out = Vec::new();
163    for component in components {
164        match component {
165            C::E => match e {
166                Some(e) => e.encode_into(&mut out),
167                None => break,
168            },
169            C::A => match a {
170                Some(a) => a.encode_into(&mut out),
171                None => break,
172            },
173            C::V => match v {
174                Some(v) => v.encode_into(&mut out),
175                None => break,
176            },
177        }
178    }
179    out
180}
181
182/// The `t` ↔ `:db/txInstant` correspondence for the transactions a value has
183/// seen, in both directions.
184///
185/// Both maps are persistent, so carrying them on every database value costs
186/// one O(log n) insert per transaction and nothing per clone. They are a
187/// property of the recorded transactions rather than of a time view, so a
188/// derived view (`as-of`, `since`, `history`) keeps the whole correspondence:
189/// naming a view by instant means the same thing whatever view you start from.
190///
191/// `:db/txInstant` is monotone by construction (the transactor stamps
192/// `max(now, last + 1)`), so ordering by instant and ordering by `t` agree.
193#[derive(Clone, Debug, Default)]
194pub struct TxInstants {
195    by_t: RedBlackTreeMapSync<u64, i64>,
196    by_instant: RedBlackTreeMapSync<i64, u64>,
197}
198
199impl TxInstants {
200    fn record(&mut self, t: u64, instant: i64) {
201        self.by_t.insert_mut(t, instant);
202        // Equal instants can only come from a log written without the
203        // monotonicity rule; keeping the highest `t` makes resolution pick the
204        // last transaction that shares an instant, matching `as-of`'s
205        // inclusive semantics.
206        if self.by_instant.get(&instant).is_none_or(|held| *held < t) {
207            self.by_instant.insert_mut(instant, t);
208        }
209    }
210
211    /// The commit instant of transaction `t`.
212    #[must_use]
213    pub fn instant(&self, t: u64) -> Option<i64> {
214        self.by_t.get(&t).copied()
215    }
216
217    /// The latest `t` committed at or before `instant`, or zero when no known
218    /// transaction is that old (the basis of an empty database).
219    #[must_use]
220    pub fn t_at(&self, instant: i64) -> u64 {
221        self.by_instant
222            .range(..=instant)
223            .next_back()
224            .map_or(0, |(_, t)| *t)
225    }
226
227    /// Number of transactions with a known instant.
228    #[must_use]
229    pub fn len(&self) -> usize {
230        self.by_t.size()
231    }
232
233    /// Whether no transaction instant is known.
234    #[must_use]
235    pub fn is_empty(&self) -> bool {
236        self.by_t.is_empty()
237    }
238}
239
240/// An immutable value of a database at one basis transaction and time view.
241#[derive(Clone, Debug, Default)]
242pub struct Db {
243    basis_t: u64,
244    schema: Schema,
245    recorded: VectorSync<Datom>,
246    idents: Arc<Idents>,
247    interner: Arc<KeywordInterner>,
248    view: DbView,
249    instants: TxInstants,
250    indexes: Arc<OnceLock<[Index; 4]>>,
251    stats: Arc<OnceLock<PlannerStats>>,
252}
253
254impl Db {
255    /// Creates an empty database with the supplied schema.
256    ///
257    /// The engine's own attributes ([`bootstrap`]) are installed over
258    /// `schema`, so every database understands `:db/txInstant` whether or not
259    /// the caller's schema mentions it.
260    #[must_use]
261    pub fn new(mut schema: Schema) -> Self {
262        bootstrap::install_schema(&mut schema);
263        Self {
264            schema,
265            ..Self::default()
266        }
267    }
268
269    /// Creates a current database value from a published EAVT snapshot.
270    ///
271    /// `datoms` must be the live facts at `basis_t`. Their original
272    /// transaction ids are retained, but transactions before `basis_t` that
273    /// no longer contribute a live fact are not reconstructed. This makes
274    /// the value suitable for current queries and for applying the log tail;
275    /// complete historical views still require replaying the full log.
276    ///
277    /// `:db/txInstant` datoms are live facts (nothing retracts them), so a
278    /// snapshot carries the transaction-time correspondence for every
279    /// transaction it covers — unless it was published before the engine
280    /// recorded instants as datoms, in which case instant-named views resolve
281    /// only within the replayed tail.
282    #[must_use]
283    pub fn from_current_snapshot(
284        basis_t: u64,
285        mut schema: Schema,
286        mut idents: Idents,
287        interner: KeywordInterner,
288        datoms: Vec<Datom>,
289    ) -> Self {
290        bootstrap::install(&mut schema, &mut idents);
291        let mut instants = TxInstants::default();
292        for datom in &datoms {
293            if let Datom {
294                e,
295                a,
296                v: Value::Instant(instant),
297                tx,
298                added: true,
299                ..
300            } = datom
301                && *a == bootstrap::TX_INSTANT
302                && *e == *tx
303                && tx.partition() == Partition::Tx as u32
304            {
305                instants.record(tx.sequence(), *instant);
306            }
307        }
308        Self {
309            basis_t,
310            schema,
311            recorded: datoms.into_iter().collect(),
312            idents: Arc::new(idents),
313            interner: Arc::new(interner),
314            view: DbView::Current,
315            instants,
316            indexes: Arc::new(OnceLock::new()),
317            stats: Arc::new(OnceLock::new()),
318        }
319    }
320
321    /// Attaches ident and keyword naming registries, returning the named value.
322    #[must_use]
323    pub fn with_naming(mut self, mut idents: Idents, interner: KeywordInterner) -> Self {
324        bootstrap::install(&mut self.schema, &mut idents);
325        self.idents = Arc::new(idents);
326        self.interner = Arc::new(interner);
327        self
328    }
329
330    /// Current transaction basis.
331    #[must_use]
332    pub const fn basis_t(&self) -> u64 {
333        self.basis_t
334    }
335
336    /// Schema at this basis.
337    #[must_use]
338    pub const fn schema(&self) -> &Schema {
339        &self.schema
340    }
341
342    /// Ident registry.
343    #[must_use]
344    pub fn idents(&self) -> &Idents {
345        &self.idents
346    }
347
348    /// Keyword interner used by keyword values in this database.
349    #[must_use]
350    pub fn interner(&self) -> &KeywordInterner {
351        &self.interner
352    }
353
354    /// The time view this value presents.
355    #[must_use]
356    pub const fn view(&self) -> DbView {
357        self.view
358    }
359
360    /// Every recorded assertion and retraction, in transaction order.
361    pub fn recorded_datoms(&self) -> impl Iterator<Item = &Datom> {
362        self.recorded.iter()
363    }
364
365    /// Number of recorded assertions and retractions.
366    #[must_use]
367    pub fn recorded_len(&self) -> usize {
368        self.recorded.len()
369    }
370
371    /// The datoms transactions after `t` recorded, in the order they were
372    /// recorded.
373    ///
374    /// `recorded` is append-ordered, so the answer is a suffix: the scan
375    /// walks back from the end and stops at the first datom at or before `t`,
376    /// costing time proportional to the tail rather than to the whole
377    /// history. This is what an indexing pass folds into the segments it last
378    /// published ([`corium_index::Segment::apply`]) instead of rebuilding
379    /// them.
380    ///
381    /// A value opened from a published snapshot carries that snapshot's live
382    /// datoms as its prefix rather than a transaction-ordered history; every
383    /// one of them is at or before the snapshot's basis, so the same scan
384    /// still stops at the boundary for any `t` at or after it.
385    pub fn recorded_since(&self, t: u64) -> impl Iterator<Item = &Datom> {
386        let mut start = self.recorded.len();
387        while start > 0
388            && self
389                .recorded
390                .get(start - 1)
391                .is_some_and(|datom| datom.tx.sequence() > t)
392        {
393            start -= 1;
394        }
395        // Indexed access rather than `iter().skip(start)`: the persistent
396        // vector's iterator would have to walk the whole prefix to reach the
397        // tail, which is the cost this method exists to avoid.
398        (start..self.recorded.len()).filter_map(|index| self.recorded.get(index))
399    }
400
401    /// Returns the as-of view at basis `t`: facts as they stood then.
402    #[must_use]
403    pub fn as_of(&self, t: u64) -> Self {
404        self.with_view(DbView::AsOf(t))
405    }
406
407    /// Returns the since view: only live facts added after `t`.
408    #[must_use]
409    pub fn since(&self, t: u64) -> Self {
410        self.with_view(DbView::Since(t))
411    }
412
413    /// Returns the history view: all assertions and retractions ever.
414    #[must_use]
415    pub fn history(&self) -> Self {
416        self.with_view(DbView::History)
417    }
418
419    /// The `t` ↔ `:db/txInstant` correspondence recorded by this value.
420    #[must_use]
421    pub const fn instants(&self) -> &TxInstants {
422        &self.instants
423    }
424
425    /// The commit instant of transaction `t`, when this value has seen it.
426    #[must_use]
427    pub fn tx_instant(&self, t: u64) -> Option<i64> {
428        self.instants.instant(t)
429    }
430
431    /// The latest basis committed at or before `instant` (Unix milliseconds).
432    ///
433    /// Zero when no known transaction is that old, so `as-of` an instant
434    /// before the database existed is the empty value and `since` that instant
435    /// is everything — the same interpretation Datomic gives out-of-range
436    /// instants.
437    #[must_use]
438    pub fn t_at_instant(&self, instant: i64) -> u64 {
439        self.instants.t_at(instant)
440    }
441
442    /// Returns the as-of view at wall-clock `instant` (Unix milliseconds):
443    /// facts as they stood after the last transaction committed at or before
444    /// it.
445    #[must_use]
446    pub fn as_of_instant(&self, instant: i64) -> Self {
447        self.as_of(self.t_at_instant(instant))
448    }
449
450    /// Returns the since view at wall-clock `instant` (Unix milliseconds):
451    /// only live facts added after the last transaction committed at or before
452    /// it.
453    #[must_use]
454    pub fn since_instant(&self, instant: i64) -> Self {
455        self.since(self.t_at_instant(instant))
456    }
457
458    fn with_view(&self, view: DbView) -> Self {
459        if view == self.view {
460            return self.clone();
461        }
462        Self {
463            view,
464            indexes: Arc::new(OnceLock::new()),
465            stats: Arc::new(OnceLock::new()),
466            ..self.clone()
467        }
468    }
469
470    /// Groups recorded datoms by transaction over the half-open range `[start, end)`.
471    #[must_use]
472    pub fn tx_range(&self, start: u64, end: Option<u64>) -> Vec<(u64, Vec<Datom>)> {
473        let mut by_t: BTreeMap<u64, Vec<Datom>> = BTreeMap::new();
474        for datom in &self.recorded {
475            let t = datom.tx.sequence();
476            if t >= start && end.is_none_or(|end| t < end) {
477                by_t.entry(t).or_default().push(datom.clone());
478            }
479        }
480        by_t.into_iter().collect()
481    }
482
483    /// Returns this view's facts, deterministically ordered by EAVT.
484    #[must_use]
485    pub fn datoms(&self) -> Vec<Datom> {
486        self.datoms_at(IndexOrder::Eavt).cloned().collect()
487    }
488
489    /// Iterates this view's datoms in one index order.
490    ///
491    /// AVET covers only indexed/unique attributes and VAET only reference
492    /// values, mirroring Datomic's covering-index composition.
493    pub fn datoms_at(&self, order: IndexOrder) -> impl Iterator<Item = &Datom> {
494        self.indexes()[slot(order)].values()
495    }
496
497    /// Iterates this view's datoms for one attribute in AEVT order.
498    ///
499    /// Unlike AVET, AEVT covers every installed attribute. Callers can use
500    /// this as the fallback for attribute predicates that do not have AVET
501    /// coverage.
502    pub fn datoms_for_attribute(&self, a: AttrId) -> impl Iterator<Item = &Datom> {
503        let prefix = key_prefix(IndexOrder::Aevt, None, Some(a), None);
504        self.indexes()[slot(IndexOrder::Aevt)]
505            .range(prefix.clone()..)
506            .take_while(move |(key, _)| key.starts_with(&prefix))
507            .map(|(_, datom)| datom)
508    }
509
510    /// Iterates datoms whose key in `order` starts with `prefix`.
511    pub fn datoms_prefix<'a>(
512        &'a self,
513        order: IndexOrder,
514        prefix: &'a [u8],
515    ) -> impl Iterator<Item = &'a Datom> {
516        self.indexes()[slot(order)]
517            .range(prefix.to_vec()..)
518            .take_while(move |(key, _)| key.starts_with(prefix))
519            .map(|(_, datom)| datom)
520    }
521
522    /// Iterates datoms in `order` starting from the first key at or after `start`.
523    pub fn seek_datoms<'a>(
524        &'a self,
525        order: IndexOrder,
526        start: &[u8],
527    ) -> impl Iterator<Item = &'a Datom> {
528        self.indexes()[slot(order)]
529            .range(start.to_vec()..)
530            .map(|(_, datom)| datom)
531    }
532
533    /// Iterates the AVET index for `a` over the value range `[start, end)`.
534    ///
535    /// Only indexed/unique attributes appear in AVET.
536    pub fn index_range<'a>(
537        &'a self,
538        a: AttrId,
539        start: Option<&Value>,
540        end: Option<&'a Value>,
541    ) -> impl Iterator<Item = &'a Datom> {
542        let a_prefix = key_prefix(IndexOrder::Avet, None, Some(a), None);
543        let start_key = key_prefix(IndexOrder::Avet, None, Some(a), start);
544        self.indexes()[slot(IndexOrder::Avet)]
545            .range(start_key..)
546            .take_while(move |(key, _)| key.starts_with(&a_prefix))
547            .map(|(_, datom)| datom)
548            .take_while(move |datom| end.is_none_or(|end| datom.v < *end))
549    }
550
551    /// Current values for an entity/attribute pair.
552    #[must_use]
553    pub fn values(&self, e: EntityId, a: AttrId) -> Vec<Value> {
554        let prefix = key_prefix(IndexOrder::Eavt, Some(e), Some(a), None);
555        self.datoms_prefix(IndexOrder::Eavt, &prefix)
556            .map(|datom| datom.v.clone())
557            .collect()
558    }
559
560    /// Resolves a unique attribute/value pair to its entity.
561    #[must_use]
562    pub fn lookup(&self, a: AttrId, v: &Value) -> Option<EntityId> {
563        if avet_covered(&self.schema, a) {
564            let prefix = key_prefix(IndexOrder::Avet, None, Some(a), Some(v));
565            self.datoms_prefix(IndexOrder::Avet, &prefix)
566                .next()
567                .map(|datom| datom.e)
568        } else {
569            let prefix = key_prefix(IndexOrder::Aevt, None, Some(a), None);
570            self.datoms_prefix(IndexOrder::Aevt, &prefix)
571                .find(|datom| datom.v == *v)
572                .map(|datom| datom.e)
573        }
574    }
575
576    /// Applies a committed record, returning a new database value.
577    ///
578    /// Only meaningful for the current view; time views are read-only.
579    ///
580    /// A `:db/txInstant` assertion on the transaction entity — which every
581    /// commit path materializes — is picked up as this transaction's place on
582    /// the wall clock.
583    #[must_use]
584    pub fn with_transaction(&self, t: u64, datoms: &[Datom]) -> Self {
585        match bootstrap::asserted_instant(t, datoms) {
586            Some(instant) => self.with_transaction_at(t, instant, datoms),
587            None => self.apply_transaction(t, datoms),
588        }
589    }
590
591    /// Applies a committed record whose commit instant is known separately,
592    /// materializing the `:db/txInstant` datom when `datoms` lacks one.
593    ///
594    /// Replay paths use this so a log written before Corium recorded instants
595    /// as datoms still yields instant-named views: the record's timestamp
596    /// field becomes the datom it would carry today.
597    #[must_use]
598    pub fn with_transaction_at(&self, t: u64, instant: i64, datoms: &[Datom]) -> Self {
599        let asserted = bootstrap::asserted_instant(t, datoms);
600        let mut next = if asserted.is_some() {
601            self.apply_transaction(t, datoms)
602        } else {
603            let mut with_instant = Vec::with_capacity(datoms.len() + 1);
604            with_instant.extend_from_slice(datoms);
605            with_instant.push(bootstrap::tx_instant_datom(t, instant));
606            self.apply_transaction(t, &with_instant)
607        };
608        next.instants.record(t, asserted.unwrap_or(instant));
609        next
610    }
611
612    fn apply_transaction(&self, t: u64, datoms: &[Datom]) -> Self {
613        debug_assert!(
614            self.view == DbView::Current,
615            "with_transaction applies only to the current view"
616        );
617        let mut next = self.clone();
618        next.basis_t = t;
619        // `recorded` is a persistent vector: this clone shared its whole spine
620        // with the parent, and appending copies only the O(log n) nodes on the
621        // tail path — never the entire log, even while `db_before` keeps the
622        // parent alive.
623        for datom in datoms {
624            next.recorded.push_back_mut(datom.clone());
625        }
626        next.indexes = Arc::new(OnceLock::new());
627        next.stats = Arc::new(OnceLock::new());
628        // Derive indexes incrementally when the parent already built them, so
629        // transaction pipelines don't refold the whole history per operation.
630        if let Some(parent) = self.indexes.get() {
631            let mut derived = parent.clone();
632            apply_current(&mut derived, datoms.iter(), &self.schema);
633            let _ = next.indexes.set(derived);
634        }
635        next
636    }
637
638    /// Computes basic statistics over this view's facts.
639    #[must_use]
640    pub fn stats(&self) -> DbStats {
641        let planner = self.planner_stats();
642        DbStats {
643            datoms: planner.total_datoms,
644            entities: planner.entity_count,
645            attributes: planner.per_attr.len(),
646        }
647    }
648
649    /// Planner statistics for this view, built lazily and cached.
650    #[must_use]
651    pub fn planner_stats(&self) -> &PlannerStats {
652        self.stats.get_or_init(|| {
653            let mut stats = PlannerStats::default();
654            let mut values: BTreeMap<AttrId, BTreeSet<&Value>> = BTreeMap::new();
655            let mut attr_entities: BTreeMap<AttrId, BTreeSet<EntityId>> = BTreeMap::new();
656            let mut entities: BTreeSet<EntityId> = BTreeSet::new();
657            for datom in self.datoms_at(IndexOrder::Eavt) {
658                stats.total_datoms += 1;
659                stats.per_attr.entry(datom.a).or_default().count += 1;
660                values.entry(datom.a).or_default().insert(&datom.v);
661                attr_entities.entry(datom.a).or_default().insert(datom.e);
662                entities.insert(datom.e);
663            }
664            for (a, entry) in &mut stats.per_attr {
665                entry.distinct_values = values.get(a).map_or(0, BTreeSet::len);
666                entry.distinct_entities = attr_entities.get(a).map_or(0, BTreeSet::len);
667            }
668            stats.entity_count = entities.len();
669            stats
670        })
671    }
672
673    fn indexes(&self) -> &[Index; 4] {
674        self.indexes.get_or_init(|| {
675            let mut indexes: [Index; 4] = Default::default();
676            match self.view {
677                DbView::History => {
678                    for datom in &self.recorded {
679                        if self.schema.get(datom.a).is_some_and(|a| a.no_history) {
680                            continue;
681                        }
682                        insert_datom(&mut indexes, datom, &self.schema, true);
683                    }
684                }
685                DbView::Current | DbView::AsOf(_) | DbView::Since(_) => {
686                    let cutoff = match self.view {
687                        DbView::AsOf(t) => Some(t),
688                        _ => None,
689                    };
690                    let filtered = self
691                        .recorded
692                        .iter()
693                        .filter(|d| cutoff.is_none_or(|t| d.tx.sequence() <= t));
694                    apply_current(&mut indexes, filtered, &self.schema);
695                    if let DbView::Since(t) = self.view {
696                        for index in &mut indexes {
697                            *index = index
698                                .iter()
699                                .filter(|(_, datom)| datom.tx.sequence() > t)
700                                .map(|(key, datom)| (key.clone(), datom.clone()))
701                                .collect();
702                        }
703                    }
704                }
705            }
706            indexes
707        })
708    }
709}
710
711/// Folds assertions/retractions into current-view indexes.
712///
713/// Current views key entries by components only (no transaction suffix):
714/// at most one live datom exists per `(e, a, v)`, and retractions must
715/// erase the assertion regardless of which transactions produced them.
716fn apply_current<'a>(
717    indexes: &mut [Index; 4],
718    datoms: impl Iterator<Item = &'a Datom>,
719    schema: &Schema,
720) {
721    for datom in datoms {
722        if datom.added {
723            insert_datom(indexes, datom, schema, false);
724        } else {
725            for order in ORDERS {
726                if covered(schema, order, datom) {
727                    let key = key_prefix(order, Some(datom.e), Some(datom.a), Some(&datom.v));
728                    indexes[slot(order)].remove_mut(&key);
729                }
730            }
731        }
732    }
733}
734
735fn insert_datom(indexes: &mut [Index; 4], datom: &Datom, schema: &Schema, with_tx: bool) {
736    for order in ORDERS {
737        if covered(schema, order, datom) {
738            let key = if with_tx {
739                datom.key(order)
740            } else {
741                key_prefix(order, Some(datom.e), Some(datom.a), Some(&datom.v))
742            };
743            indexes[slot(order)].insert_mut(key, datom.clone());
744        }
745    }
746}
747
748/// Whether `datom` belongs in `order`'s covering index.
749///
750/// EAVT and AEVT hold every datom; AVET holds only indexed/unique
751/// attributes and VAET only reference values, mirroring Datomic's covering
752/// index composition. An indexing pass folding a transaction tail into
753/// published segments filters it through the same rule the in-memory
754/// indexes use, so the two cannot drift.
755#[must_use]
756pub fn covered(schema: &Schema, order: IndexOrder, datom: &Datom) -> bool {
757    match order {
758        IndexOrder::Eavt | IndexOrder::Aevt => true,
759        IndexOrder::Avet => avet_covered(schema, datom.a),
760        IndexOrder::Vaet => matches!(datom.v, Value::Ref(_)),
761    }
762}
763
764/// Whether the attribute participates in the AVET covering index.
765#[must_use]
766pub fn avet_covered(schema: &Schema, a: AttrId) -> bool {
767    schema
768        .get(a)
769        .is_some_and(|attr| attr.indexed || attr.unique.is_some())
770}
771
772/// Counts over one database view.
773#[derive(Clone, Copy, Debug, Eq, PartialEq)]
774pub struct DbStats {
775    /// Facts in the view.
776    pub datoms: usize,
777    /// Entities having at least one fact in the view.
778    pub entities: usize,
779    /// Attributes used by facts in the view.
780    pub attributes: usize,
781}
782
783/// Convenience constructor for schema attributes used during bootstrap/tests.
784#[must_use]
785pub const fn attribute(
786    id: u64,
787    value_type: corium_core::ValueType,
788    cardinality: Cardinality,
789    unique: Option<Unique>,
790) -> corium_core::Attribute {
791    corium_core::Attribute {
792        id: EntityId::new(Partition::Db as u32, id),
793        value_type,
794        cardinality,
795        unique,
796        is_component: false,
797        indexed: unique.is_some(),
798        no_history: false,
799    }
800}
801
802#[cfg(test)]
803mod tests {
804    use super::*;
805    use corium_core::ValueType;
806
807    fn schema() -> Schema {
808        let mut schema = Schema::default();
809        schema.insert(attribute(1, ValueType::Str, Cardinality::One, None));
810        schema.insert(attribute(
811            2,
812            ValueType::Long,
813            Cardinality::One,
814            Some(Unique::Identity),
815        ));
816        schema.insert(attribute(3, ValueType::Ref, Cardinality::Many, None));
817        schema
818    }
819
820    fn attr(id: u64) -> AttrId {
821        EntityId::new(Partition::Db as u32, id)
822    }
823
824    fn entity(id: u64) -> EntityId {
825        EntityId::new(Partition::User as u32, id)
826    }
827
828    fn tx_entity(t: u64) -> EntityId {
829        EntityId::new(Partition::Tx as u32, t)
830    }
831
832    fn datom(e: u64, a: u64, v: Value, t: u64, added: bool) -> Datom {
833        Datom {
834            e: entity(e),
835            a: attr(a),
836            v,
837            tx: tx_entity(t),
838            added,
839        }
840    }
841
842    fn sample() -> Db {
843        Db::new(schema())
844            .with_transaction(
845                1,
846                &[
847                    datom(1, 1, Value::Str("alice".into()), 1, true),
848                    datom(1, 2, Value::Long(7), 1, true),
849                ],
850            )
851            .with_transaction(
852                2,
853                &[
854                    datom(1, 1, Value::Str("alice".into()), 2, false),
855                    datom(1, 1, Value::Str("alicia".into()), 2, true),
856                    datom(2, 3, Value::Ref(entity(1)), 2, true),
857                ],
858            )
859    }
860
861    #[test]
862    fn current_view_folds_retractions() {
863        let db = sample();
864        assert_eq!(
865            db.values(entity(1), attr(1)),
866            vec![Value::Str("alicia".into())]
867        );
868        assert_eq!(db.stats().datoms, 3);
869    }
870
871    #[test]
872    fn recorded_since_returns_the_transaction_tail() {
873        let db = sample();
874        assert_eq!(db.recorded_since(2).count(), 0);
875        let tail: Vec<_> = db.recorded_since(1).collect();
876        assert_eq!(tail.len(), 3);
877        assert!(tail.iter().all(|datom| datom.tx.sequence() == 2));
878        assert_eq!(db.recorded_since(0).count(), db.recorded_len());
879
880        // A value opened from a published snapshot carries live datoms rather
881        // than a transaction-ordered history; the tail after it still scans.
882        let snapshot = Db::from_current_snapshot(
883            2,
884            schema(),
885            Idents::default(),
886            KeywordInterner::default(),
887            db.datoms(),
888        )
889        .with_transaction(3, &[datom(3, 1, Value::Str("carol".into()), 3, true)]);
890        let tail: Vec<_> = snapshot.recorded_since(2).collect();
891        assert_eq!(tail.len(), 1);
892        assert!(tail.iter().all(|datom| datom.tx.sequence() == 3));
893    }
894
895    #[test]
896    fn attribute_scan_uses_complete_aevt_coverage() {
897        let db = sample();
898        let datoms = db.datoms_for_attribute(attr(1)).collect::<Vec<_>>();
899        assert_eq!(datoms.len(), 1);
900        assert_eq!(datoms[0].v, Value::Str("alicia".into()));
901        assert!(datoms.iter().all(|datom| datom.a == attr(1)));
902    }
903
904    #[test]
905    fn as_of_reconstructs_past_basis() {
906        let db = sample().as_of(1);
907        assert_eq!(
908            db.values(entity(1), attr(1)),
909            vec![Value::Str("alice".into())]
910        );
911        assert_eq!(db.stats().datoms, 2);
912    }
913
914    #[test]
915    fn since_excludes_older_live_facts() {
916        let db = sample().since(1);
917        // The long asserted at t=1 is invisible; the renamed string is visible.
918        assert_eq!(db.values(entity(1), attr(2)), Vec::<Value>::new());
919        assert_eq!(
920            db.values(entity(1), attr(1)),
921            vec![Value::Str("alicia".into())]
922        );
923    }
924
925    #[test]
926    fn history_exposes_assertions_and_retractions() {
927        let db = sample().history();
928        let names: Vec<_> = db
929            .datoms_prefix(
930                IndexOrder::Eavt,
931                &key_prefix(IndexOrder::Eavt, Some(entity(1)), Some(attr(1)), None),
932            )
933            .map(|d| (d.v.clone(), d.added))
934            .collect();
935        assert_eq!(names.len(), 3);
936        assert!(names.contains(&(Value::Str("alice".into()), false)));
937    }
938
939    #[test]
940    fn avet_only_covers_indexed_attributes() {
941        let db = sample();
942        assert_eq!(db.datoms_at(IndexOrder::Avet).count(), 1);
943        assert_eq!(db.datoms_at(IndexOrder::Vaet).count(), 1);
944    }
945
946    #[test]
947    fn index_range_scans_value_bounds() {
948        let db = Db::new(schema()).with_transaction(
949            1,
950            &[
951                datom(1, 2, Value::Long(1), 1, true),
952                datom(2, 2, Value::Long(5), 1, true),
953                datom(3, 2, Value::Long(9), 1, true),
954            ],
955        );
956        let hits: Vec<_> = db
957            .index_range(attr(2), Some(&Value::Long(2)), Some(&Value::Long(9)))
958            .map(|d| d.e)
959            .collect();
960        assert_eq!(hits, vec![entity(2)]);
961    }
962
963    /// The same two transactions as [`sample`], committed at known instants
964    /// the way the transactor commits them.
965    fn timed_sample() -> Db {
966        Db::new(schema())
967            .with_transaction_at(
968                1,
969                1_000,
970                &[datom(1, 1, Value::Str("alice".into()), 1, true)],
971            )
972            .with_transaction_at(
973                2,
974                2_000,
975                &[
976                    datom(1, 1, Value::Str("alice".into()), 2, false),
977                    datom(1, 1, Value::Str("alicia".into()), 2, true),
978                ],
979            )
980    }
981
982    #[test]
983    fn commit_instants_are_recorded_in_both_directions() {
984        let db = timed_sample();
985        assert_eq!(db.tx_instant(1), Some(1_000));
986        assert_eq!(db.tx_instant(2), Some(2_000));
987        assert_eq!(db.tx_instant(3), None);
988        // Exactly on a commit, before the first, between two, and after the last.
989        assert_eq!(db.t_at_instant(1_000), 1);
990        assert_eq!(db.t_at_instant(999), 0);
991        assert_eq!(db.t_at_instant(1_999), 1);
992        assert_eq!(db.t_at_instant(9_999), 2);
993    }
994
995    #[test]
996    fn instant_named_views_match_their_basis_named_equivalents() {
997        let db = timed_sample();
998        assert_eq!(db.as_of_instant(1_500).datoms(), db.as_of(1).datoms());
999        assert_eq!(db.since_instant(1_500).datoms(), db.since(1).datoms());
1000        // An instant older than the database is the empty value; `since` that
1001        // same instant is therefore everything.
1002        assert!(db.as_of_instant(0).datoms().is_empty());
1003        assert_eq!(db.since_instant(0).datoms(), db.since(0).datoms());
1004    }
1005
1006    #[test]
1007    fn transaction_time_is_queryable_data() {
1008        let db = timed_sample();
1009        let instants: Vec<_> = db
1010            .datoms_for_attribute(bootstrap::TX_INSTANT)
1011            .map(|datom| (datom.e, datom.v.clone()))
1012            .collect();
1013        assert_eq!(
1014            instants,
1015            vec![
1016                (tx_entity(1), Value::Instant(1_000)),
1017                (tx_entity(2), Value::Instant(2_000)),
1018            ]
1019        );
1020        // Indexed, so an instant range is an AVET seek rather than a scan.
1021        assert!(avet_covered(db.schema(), bootstrap::TX_INSTANT));
1022    }
1023
1024    #[test]
1025    fn derived_views_keep_the_whole_transaction_time_correspondence() {
1026        // Resolution must not depend on which view names the instant: a
1027        // `since` view hides older datoms, but the instants they were
1028        // committed at still name the same transactions.
1029        let db = timed_sample();
1030        for view in [db.as_of(1), db.since(2), db.history()] {
1031            assert_eq!(view.tx_instant(1), Some(1_000));
1032            assert_eq!(view.t_at_instant(2_500), 2);
1033        }
1034    }
1035
1036    #[test]
1037    fn an_instant_supplied_with_the_datoms_wins_over_the_replayed_one() {
1038        // A log record written by a transactor that already materialized the
1039        // datom must not gain a second, contradictory one.
1040        let db = Db::new(schema()).with_transaction_at(
1041            1,
1042            5_000,
1043            &[
1044                datom(1, 1, Value::Str("alice".into()), 1, true),
1045                bootstrap::tx_instant_datom(1, 1_000),
1046            ],
1047        );
1048        assert_eq!(db.values(tx_entity(1), bootstrap::TX_INSTANT).len(), 1);
1049        assert_eq!(db.tx_instant(1), Some(1_000));
1050    }
1051
1052    #[test]
1053    fn snapshot_ignores_tx_instant_values_on_non_transaction_entities() {
1054        let user = entity(1);
1055        let db = Db::from_current_snapshot(
1056            1,
1057            schema(),
1058            Idents::default(),
1059            KeywordInterner::default(),
1060            vec![
1061                bootstrap::tx_instant_datom(1, 1_000),
1062                Datom {
1063                    e: user,
1064                    a: bootstrap::TX_INSTANT,
1065                    v: Value::Instant(9_000),
1066                    tx: tx_entity(1),
1067                    added: true,
1068                },
1069            ],
1070        );
1071        assert_eq!(db.tx_instant(1), Some(1_000));
1072    }
1073
1074    #[test]
1075    fn tx_range_groups_by_transaction() {
1076        let ranged = sample().tx_range(2, None);
1077        assert_eq!(ranged.len(), 1);
1078        assert_eq!(ranged[0].0, 2);
1079        assert_eq!(ranged[0].1.len(), 3);
1080    }
1081
1082    #[test]
1083    fn incremental_indexes_match_rebuilt_indexes() {
1084        let base = Db::new(schema());
1085        let tx1 = [datom(1, 1, Value::Str("a".into()), 1, true)];
1086        let tx2 = [
1087            datom(1, 1, Value::Str("a".into()), 2, false),
1088            datom(1, 1, Value::Str("b".into()), 2, true),
1089        ];
1090        // Force the parent cache so with_transaction derives incrementally.
1091        let warm = base.with_transaction(1, &tx1);
1092        let _ = warm.datoms();
1093        let incremental = warm.with_transaction(2, &tx2);
1094        let cold = base.with_transaction(1, &tx1).with_transaction(2, &tx2);
1095        assert_eq!(incremental.datoms(), cold.datoms());
1096    }
1097
1098    #[test]
1099    fn with_transaction_leaves_parent_value_unchanged() {
1100        // `recorded` is a persistent vector appended via copy-on-write; a
1101        // derived value must never mutate the log or indexes still observed
1102        // through the parent (e.g. `db_before` in a `TxReport`).
1103        let parent = Db::new(schema())
1104            .with_transaction(1, &[datom(1, 1, Value::Str("alice".into()), 1, true)]);
1105        // Materialize the parent's indexes so the child derives incrementally.
1106        let parent_datoms = parent.datoms();
1107        let parent_recorded = parent.recorded_len();
1108
1109        let child = parent.with_transaction(
1110            2,
1111            &[
1112                datom(1, 1, Value::Str("alice".into()), 2, false),
1113                datom(1, 1, Value::Str("alicia".into()), 2, true),
1114            ],
1115        );
1116
1117        // Parent is frozen at its own basis, log length, and live facts.
1118        assert_eq!(parent.basis_t(), 1);
1119        assert_eq!(parent.recorded_len(), parent_recorded);
1120        assert_eq!(parent.datoms(), parent_datoms);
1121        assert_eq!(
1122            parent.values(entity(1), attr(1)),
1123            vec![Value::Str("alice".into())]
1124        );
1125        // Child reflects the new transaction.
1126        assert_eq!(child.basis_t(), 2);
1127        assert_eq!(
1128            child.values(entity(1), attr(1)),
1129            vec![Value::Str("alicia".into())]
1130        );
1131    }
1132
1133    #[test]
1134    fn planner_stats_count_attributes() {
1135        let stats_owner = sample();
1136        let stats = stats_owner.planner_stats();
1137        assert_eq!(stats.total_datoms, 3);
1138        assert_eq!(stats.per_attr[&attr(1)].count, 1);
1139        assert!(stats.estimate(false, Some(attr(1)), false) <= stats.total_datoms);
1140    }
1141}