reddb_server/storage/unified/
store.rs

1//! Unified Store
2//!
3//! High-level API for the unified storage layer that combines tables, graphs,
4//! and vectors into a single coherent interface.
5//!
6//! # Features
7//!
8//! - Multi-collection management
9//! - Cross-collection queries
10//! - Unified entity access
11//! - Automatic ID generation
12//! - Cross-reference management
13//! - **Binary persistence** with pages, indices, and efficient encoding
14//! - **Page-based storage** via Pager for ACID durability
15//!
16//! # Persistence Modes
17//!
18//! 1. **File Mode** (`save_to_file`/`load_from_file`): Simple binary dump
19//! 2. **Paged Mode** (`open`/`persist`): Full page-based storage with B-tree indices
20
21use std::collections::{BTreeMap, HashMap};
22use std::fs::File;
23use std::io::{BufReader, Read};
24use std::path::{Path, PathBuf};
25use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
26use std::sync::Arc;
27
28use parking_lot::RwLock;
29
30use super::context_index::ContextIndex;
31use super::entity::{
32    CrossRef, EmbeddingSlot, EntityData, EntityId, EntityKind, RefType, RowData,
33    TimeSeriesPointKind, UnifiedEntity,
34};
35use super::entity_cache::EntityCache;
36use super::manager::{ManagerConfig, ManagerStats, SegmentManager};
37use super::metadata::{Metadata, MetadataFilter, MetadataValue};
38use super::segment::SegmentError;
39use crate::api::{DurabilityMode, GroupCommitOptions};
40use crate::physical::{ManifestEvent, ManifestEventKind};
41use crate::storage::engine::pager::PagerError;
42use crate::storage::engine::{BTree, BTreeError, Pager, PagerConfig, PhysicalFileHeader};
43use crate::storage::primitives::encoding::read_varu64;
44use reddb_types::types::Value;
45
46pub use reddb_file::{
47    is_supported_store_version, NativeCatalogCollectionSummary, NativeCatalogSummary,
48    NativeExportSummary, NativeManifestEntrySummary, NativeManifestSummary,
49    NativeMetadataStateSummary, NativeRecoverySummary, NativeRegistryIndexSummary,
50    NativeRegistryJobSummary, NativeRegistryProjectionSummary, NativeRegistrySummary,
51    NativeSnapshotSummary, NativeVectorArtifactPageSummary, NativeVectorArtifactSummary,
52    ENTITY_RECORD_MAGIC, METADATA_MAGIC, METADATA_OVERFLOW_MAGIC, NATIVE_BLOB_MAGIC,
53    NATIVE_CATALOG_MAGIC, NATIVE_COLLECTION_ROOTS_MAGIC, NATIVE_MANIFEST_MAGIC,
54    NATIVE_MANIFEST_SAMPLE_LIMIT, NATIVE_METADATA_STATE_MAGIC, NATIVE_RECOVERY_MAGIC,
55    NATIVE_REGISTRY_MAGIC, NATIVE_VECTOR_ARTIFACT_MAGIC, STORE_MAGIC, STORE_VERSION_CURRENT,
56    STORE_VERSION_V1, STORE_VERSION_V11, STORE_VERSION_V12, STORE_VERSION_V2, STORE_VERSION_V3,
57    STORE_VERSION_V4, STORE_VERSION_V5, STORE_VERSION_V6, STORE_VERSION_V7, STORE_VERSION_V8,
58    STORE_VERSION_V9,
59};
60
61#[derive(Debug, Clone, Default, PartialEq, Eq)]
62pub struct MvccVacuumStats {
63    pub scanned_versions: u64,
64    pub retained_versions: u64,
65    pub reclaimed_versions: u64,
66    pub retained_history_versions: u64,
67    pub reclaimed_history_versions: u64,
68    pub retained_tombstones: u64,
69    pub reclaimed_tombstones: u64,
70}
71
72impl MvccVacuumStats {
73    pub fn add(&mut self, other: &Self) {
74        self.scanned_versions += other.scanned_versions;
75        self.retained_versions += other.retained_versions;
76        self.reclaimed_versions += other.reclaimed_versions;
77        self.retained_history_versions += other.retained_history_versions;
78        self.reclaimed_history_versions += other.reclaimed_history_versions;
79        self.retained_tombstones += other.retained_tombstones;
80        self.reclaimed_tombstones += other.reclaimed_tombstones;
81    }
82}
83
84#[derive(Debug, Clone)]
85pub struct NativePhysicalState {
86    pub header: PhysicalFileHeader,
87    pub collection_roots: BTreeMap<String, u64>,
88    pub manifest: Option<NativeManifestSummary>,
89    pub registry: Option<NativeRegistrySummary>,
90    pub recovery: Option<NativeRecoverySummary>,
91    pub catalog: Option<NativeCatalogSummary>,
92    pub metadata_state: Option<NativeMetadataStateSummary>,
93    pub vector_artifact_pages: Option<Vec<NativeVectorArtifactPageSummary>>,
94}
95
96// ============================================================================
97// Configuration
98// ============================================================================
99
100/// Configuration for UnifiedStore
101#[derive(Debug, Clone)]
102pub struct UnifiedStoreConfig {
103    /// Configuration for segment managers
104    pub manager_config: ManagerConfig,
105    /// Automatically index cross-references on insert
106    pub auto_index_refs: bool,
107    /// Automatically build a HASH index on a user `id` column the first
108    /// time a row carrying that column is inserted into a collection.
109    /// Mirrors PostgreSQL's implicit primary-key index and Mongo's `_id`
110    /// default index — without it, `WHERE id = N` falls through to a
111    /// full segment scan because RedDB has no concept of an automatic
112    /// primary-key index on user-declared columns. See `docs/perf/
113    /// delete-sequential-2026-05-06.md` for the perf rationale.
114    /// Defaults to `true`; set to `false` to opt out per workload.
115    pub auto_index_id: bool,
116    /// Maximum cross-references per entity
117    pub max_cross_refs: usize,
118    /// Enable write-ahead logging
119    pub enable_wal: bool,
120    /// Durability profile for paged writes.
121    pub durability_mode: DurabilityMode,
122    /// Group-commit batching knobs when using grouped durability.
123    pub group_commit: GroupCommitOptions,
124    /// Data directory path
125    pub data_dir: Option<std::path::PathBuf>,
126    /// Embedded single-file artifact used for the internal WAL stream.
127    pub embedded_wal_path: Option<std::path::PathBuf>,
128    /// Preallocated page-cache slot count (ADR 0073 §2). `None` leaves the
129    /// pager on its own structural default; the promoted boot path always
130    /// supplies a count derived from the page-cache budget share, so the
131    /// default is reached only by direct library callers.
132    pub page_cache_slots: Option<usize>,
133}
134
135impl Default for UnifiedStoreConfig {
136    fn default() -> Self {
137        Self {
138            manager_config: ManagerConfig::default(),
139            auto_index_refs: true,
140            auto_index_id: true,
141            max_cross_refs: 1000,
142            enable_wal: false,
143            // Mirrors `RedDBOptions::default().durability_mode` — see
144            // `src/api.rs` for the rationale.
145            durability_mode: DurabilityMode::WalDurableGrouped,
146            group_commit: GroupCommitOptions::default(),
147            data_dir: None,
148            embedded_wal_path: None,
149            page_cache_slots: None,
150        }
151    }
152}
153
154impl UnifiedStoreConfig {
155    /// Create config with data directory
156    pub fn with_data_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
157        self.data_dir = Some(path.into());
158        self
159    }
160
161    /// Enable WAL
162    pub fn with_wal(mut self) -> Self {
163        self.enable_wal = true;
164        self
165    }
166
167    pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
168        self.durability_mode = mode;
169        self
170    }
171
172    pub fn with_group_commit(mut self, options: GroupCommitOptions) -> Self {
173        self.group_commit = options;
174        self
175    }
176
177    pub fn with_embedded_wal_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
178        self.embedded_wal_path = Some(path.into());
179        self
180    }
181
182    /// Pre-size the page cache from its budget share (ADR 0073 §2).
183    pub fn with_page_cache_slots(mut self, slots: usize) -> Self {
184        self.page_cache_slots = Some(slots);
185        self
186    }
187
188    /// Set max cross-references
189    pub fn with_max_refs(mut self, max: usize) -> Self {
190        self.max_cross_refs = max;
191        self
192    }
193}
194
195// ============================================================================
196// Error Types
197// ============================================================================
198
199/// Errors from UnifiedStore operations
200#[derive(Debug)]
201pub enum StoreError {
202    /// Collection already exists
203    CollectionExists(String),
204    /// Collection not found
205    CollectionNotFound(String),
206    /// Entity not found
207    EntityNotFound(EntityId),
208    /// Too many cross-references
209    TooManyRefs(EntityId),
210    /// Segment error
211    Segment(SegmentError),
212    /// I/O error
213    Io(std::io::Error),
214    /// Checksummed storage bytes failed validation.
215    StorageIntegrity(crate::api::StorageIntegrityError),
216    /// Serialization error
217    Serialization(String),
218    /// Internal error (lock poisoning, invariant violation)
219    Internal(String),
220}
221
222impl std::fmt::Display for StoreError {
223    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
224        match self {
225            Self::CollectionExists(name) => write!(f, "Collection already exists: {}", name),
226            Self::CollectionNotFound(name) => write!(f, "Collection not found: {}", name),
227            Self::EntityNotFound(id) => write!(f, "Entity not found: {}", id),
228            Self::TooManyRefs(id) => write!(f, "Too many cross-references for entity: {}", id),
229            Self::Segment(e) => write!(f, "Segment error: {:?}", e),
230            Self::Io(e) => write!(f, "I/O error: {}", e),
231            Self::StorageIntegrity(e) => write!(f, "{e}"),
232            Self::Serialization(msg) => write!(f, "Serialization error: {}", msg),
233            Self::Internal(msg) => write!(f, "Internal error: {}", msg),
234        }
235    }
236}
237
238impl std::error::Error for StoreError {}
239
240impl From<SegmentError> for StoreError {
241    fn from(e: SegmentError) -> Self {
242        Self::Segment(e)
243    }
244}
245
246impl From<std::io::Error> for StoreError {
247    fn from(e: std::io::Error) -> Self {
248        Self::Io(e)
249    }
250}
251
252// ============================================================================
253// Statistics
254// ============================================================================
255
256/// Statistics for UnifiedStore
257#[derive(Debug, Clone, Default)]
258pub struct StoreStats {
259    /// Number of collections
260    pub collection_count: usize,
261    /// Total entities across all collections
262    pub total_entities: usize,
263    /// Total memory usage in bytes
264    pub total_memory_bytes: usize,
265    /// Per-collection statistics
266    pub collections: HashMap<String, ManagerStats>,
267    /// Total cross-references
268    pub cross_ref_count: usize,
269}
270
271impl StoreStats {
272    /// Get average entities per collection
273    pub fn avg_entities_per_collection(&self) -> f64 {
274        if self.collection_count == 0 {
275            0.0
276        } else {
277            self.total_entities as f64 / self.collection_count as f64
278        }
279    }
280
281    /// Get memory in MB
282    pub fn memory_mb(&self) -> f64 {
283        self.total_memory_bytes as f64 / (1024.0 * 1024.0)
284    }
285}
286
287// ============================================================================
288// UnifiedStore - The Main API
289// ============================================================================
290
291struct CheckpointVersion {
292    manager: std::sync::Weak<SegmentManager>,
293    segments: Vec<(u64, u64)>,
294    visibility_revision: u64,
295    untracked_visibility_revision: u64,
296    excluded_xids: Arc<std::collections::HashSet<u64>>,
297    pending_vault: Arc<std::collections::HashSet<EntityId>>,
298    format_version: u32,
299    /// Raw rows omitted or normalized in the durable image. None retains the
300    /// full validation path when more than 1,024 such rows are outstanding.
301    /// Also carries bounded unprojected IDs across completed inline GC batches.
302    deferred_rows: Option<Arc<Vec<EntityId>>>,
303    /// Inline GC advanced the segment proof past rows still awaiting projection.
304    has_unprojected_rows: bool,
305}
306
307impl CheckpointVersion {
308    fn remember_deferred_row(&mut self, id: EntityId) {
309        let Some(rows) = self.deferred_rows.as_mut() else {
310            return;
311        };
312        if rows.len() == 1024 {
313            self.deferred_rows = None;
314        } else {
315            Arc::make_mut(rows).push(id);
316        }
317    }
318}
319
320/// Unified storage for tables, graphs, and vectors
321///
322/// UnifiedStore provides a single coherent interface for all data types:
323/// - **Tables**: Row-based data with columns
324/// - **Graphs**: Nodes and edges with labels
325/// - **Vectors**: Embeddings for similarity search
326///
327/// # Features
328///
329/// - Multi-collection management
330/// - Cross-collection queries
331/// - Cross-reference tracking between entities
332/// - Automatic ID generation
333/// - Segment-based storage with growing/sealed lifecycle
334///
335/// # Example
336///
337/// ```ignore
338/// use reddb::storage::{Entity, Store};
339///
340/// let store = Store::new();
341///
342/// // Create a collection
343/// store.create_collection("hosts")?;
344///
345/// // Insert an entity
346/// let entity = Entity::table_row(1, "hosts", 1, vec![]);
347/// let id = store.insert("hosts", entity)?;
348///
349/// // Query
350/// let found = store.get("hosts", id);
351/// ```
352pub struct UnifiedStore {
353    /// Store configuration
354    config: UnifiedStoreConfig,
355    /// File format version for serialization
356    format_version: AtomicU32,
357    /// Global entity ID counter
358    next_entity_id: AtomicU64,
359    /// Collections by name
360    collections: RwLock<HashMap<String, Arc<SegmentManager>>>,
361    /// Forward cross-references: source_id → [(target_id, ref_type, target_collection)]
362    cross_refs: RwLock<HashMap<EntityId, Vec<(EntityId, RefType, String)>>>,
363    /// Reverse cross-references: target_id → [(source_id, ref_type, source_collection)]
364    reverse_refs: RwLock<HashMap<EntityId, Vec<(EntityId, RefType, String)>>>,
365    /// Optional page-based storage via Pager
366    pager: Option<Arc<Pager>>,
367    /// Database file path (for paged mode)
368    db_path: Option<PathBuf>,
369    /// B-tree indices for O(log n) entity lookups by ID (per collection).
370    /// Stored as `Arc<BTree>` so hot-path callers can clone the handle out
371    /// under a read lock and release the map-level lock before doing the
372    /// actual insert — previously the outer RwLock was held for the whole
373    /// btree mutation, serialising every concurrent insert across every
374    /// collection into one global write lock.
375    btree_indices: RwLock<HashMap<String, Arc<BTree>>>,
376    /// Reuse only a successfully published generation with the same live rows.
377    checkpoint_versions: RwLock<HashMap<String, CheckpointVersion>>,
378    checkpoint_durable_rows: RwLock<checkpoint_delta::DurableRows>,
379    /// The immediately previous published trees and any older reader-pinned
380    /// trees. Root identities are persisted with the paged catalog.
381    retired_btrees: parking_lot::Mutex<Vec<RetiredBTree>>,
382    /// Cross-structure context index for unified search
383    context_index: ContextIndex,
384    /// Hot entity cache — sharded bounded LRU for `get_any` lookups.
385    /// See `entity_cache.rs` for the rationale; this replaced a single
386    /// `RwLock<HashMap>` that serialised every `delete_batch` invalidation.
387    entity_cache: EntityCache,
388    /// Graph node label index: (collection, label) → Vec<EntityId>.
389    /// O(1) lookup for MATCH (n:Label) graph patterns — avoids full collection scan.
390    graph_label_index: RwLock<HashMap<(String, String), Vec<EntityId>>>,
391    /// Whether the paged registry on page 1 must be rewritten before the next flush.
392    paged_registry_dirty: AtomicBool,
393    /// Logical store WAL / grouped durability coordinator for paged mode.
394    commit: Option<Arc<StoreCommitCoordinator>>,
395    /// Counts how often `unindex_cross_refs_batch` took the read-only fast
396    /// path (no inbound refs, no outbound refs for any deleted id) and so
397    /// avoided acquiring the `cross_refs` / `reverse_refs` write locks.
398    /// Used by tests to pin the early-exit; cheap relaxed counter otherwise.
399    unindex_cross_refs_fast_path: AtomicU64,
400    /// WAL-replayed `VectorInsert` records, captured at open time and
401    /// drained per-collection on first `vector.turbo` access (issue
402    /// #694). Boot-time recovery: the in-memory TurboQuant index is
403    /// rebuilt by replaying these FP32 vectors in WAL order, so the
404    /// rebuilt state is byte-deterministic against the pre-restart
405    /// state under a fixed codec seed.
406    pub(crate) replayed_turbo_inserts: parking_lot::Mutex<HashMap<String, Vec<(u64, Vec<f32>)>>>,
407    /// WAL-replayed probabilistic mutation deltas captured at open time.
408    /// Runtime recovery drains these after loading the latest full
409    /// probabilistic snapshot from store state.
410    pub(crate) replayed_probabilistic_deltas:
411        parking_lot::Mutex<Vec<(u8, u8, String, Vec<Vec<u8>>)>>,
412    /// Opaque store-level auxiliary metadata persisted inside the binary dump
413    /// (store format V10+). RedDB uses this to carry collection contracts
414    /// through the single-file artifact so a collection's `declared_model`
415    /// (e.g. `kv`) survives a restart instead of being re-inferred as a table.
416    /// The store treats the bytes as opaque; only RedDB interprets them.
417    pub(crate) aux_metadata: RwLock<Vec<u8>>,
418    /// Serialize vault publication with snapshot capture and WAL truncation.
419    /// Physical metadata and native header publication take it too (#2376).
420    /// Reentrant because embedded checkpoint calls the snapshot encoder while
421    /// holding the barrier, and storage maintenance may nest those calls.
422    pub(crate) vault_publication_lock: parking_lot::ReentrantMutex<()>,
423    /// Live-only vault rows excluded from a reentrant WAL-full compaction.
424    pub(crate) pending_vault_versions: RwLock<std::collections::HashSet<EntityId>>,
425    /// Retain transaction outcomes through embedded snapshot-on-drop. Without
426    /// this, aborted physical versions are reopened as committed rows.
427    pub(crate) snapshot_manager:
428        std::sync::OnceLock<Arc<crate::storage::transaction::snapshot::SnapshotManager>>,
429    embedded_commit_session: std::sync::OnceLock<crate::storage::embedded::EmbeddedCommitSession>,
430}
431
432struct RetiredBTree {
433    tree: Arc<BTree>,
434    published: bool,
435}
436
437mod builder;
438mod checkpoint_delta;
439mod checkpoint_wal;
440mod commit;
441mod impl_entities;
442mod impl_file;
443mod impl_native_a;
444mod impl_native_b;
445mod impl_native_c;
446mod impl_pages;
447mod impl_vault_import;
448mod keyed_write;
449mod mvcc_reclaim;
450pub(crate) use mvcc_reclaim::MvccReclaimCursor;
451mod native_helpers;
452
453pub use self::builder::EntityBuilder;
454pub(crate) use self::commit::DeferredStoreWalActions;
455use self::commit::{StoreCommitCoordinator, StoreWalAction};
456use self::native_helpers::*;
reddb_server/storage/unified/store.rs

reddb_server/storage/unified/
store.rs
