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::{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::manager::{ManagerConfig, ManagerStats, SegmentManager};
36use super::metadata::{Metadata, MetadataFilter, MetadataValue};
37use crate::api::{DurabilityMode, GroupCommitOptions};
38use crate::physical::{ManifestEvent, ManifestEventKind};
39use crate::storage::engine::pager::PagerError;
40use crate::storage::engine::{BTree, BTreeError, Pager, PagerConfig, PhysicalFileHeader};
41use crate::storage::primitives::encoding::read_varu64;
42use reddb_types::types::Value;
43
44pub(crate) fn trace_startup_phase(path: &Path, phase: &'static str, started: std::time::Instant) {
45    tracing::debug!(
46        target: "reddb::storage::startup",
47        path = %path.display(),
48        phase,
49        elapsed = ?started.elapsed(),
50        "database startup phase completed"
51    );
52}
53
54pub use reddb_file::{
55    is_supported_store_version, NativeCatalogCollectionSummary, NativeCatalogSummary,
56    NativeExportSummary, NativeManifestEntrySummary, NativeManifestSummary,
57    NativeMetadataStateSummary, NativeRecoverySummary, NativeRegistryIndexSummary,
58    NativeRegistryJobSummary, NativeRegistryProjectionSummary, NativeRegistrySummary,
59    NativeSnapshotSummary, NativeVectorArtifactPageSummary, NativeVectorArtifactSummary,
60    ENTITY_RECORD_MAGIC, METADATA_MAGIC, METADATA_OVERFLOW_MAGIC, NATIVE_BLOB_MAGIC,
61    NATIVE_CATALOG_MAGIC, NATIVE_COLLECTION_ROOTS_MAGIC, NATIVE_MANIFEST_MAGIC,
62    NATIVE_MANIFEST_SAMPLE_LIMIT, NATIVE_METADATA_STATE_MAGIC, NATIVE_RECOVERY_MAGIC,
63    NATIVE_REGISTRY_MAGIC, NATIVE_VECTOR_ARTIFACT_MAGIC, STORE_MAGIC, STORE_VERSION_CURRENT,
64    STORE_VERSION_V1, STORE_VERSION_V11, STORE_VERSION_V12, STORE_VERSION_V2, STORE_VERSION_V3,
65    STORE_VERSION_V4, STORE_VERSION_V5, STORE_VERSION_V6, STORE_VERSION_V7, STORE_VERSION_V8,
66    STORE_VERSION_V9,
67};
68
69#[derive(Debug, Clone, Default, PartialEq, Eq)]
70pub struct MvccVacuumStats {
71    pub scanned_versions: u64,
72    pub retained_versions: u64,
73    pub reclaimed_versions: u64,
74    pub retained_history_versions: u64,
75    pub reclaimed_history_versions: u64,
76    pub retained_tombstones: u64,
77    pub reclaimed_tombstones: u64,
78}
79
80impl MvccVacuumStats {
81    pub fn add(&mut self, other: &Self) {
82        self.scanned_versions += other.scanned_versions;
83        self.retained_versions += other.retained_versions;
84        self.reclaimed_versions += other.reclaimed_versions;
85        self.retained_history_versions += other.retained_history_versions;
86        self.reclaimed_history_versions += other.reclaimed_history_versions;
87        self.retained_tombstones += other.retained_tombstones;
88        self.reclaimed_tombstones += other.reclaimed_tombstones;
89    }
90}
91
92#[derive(Debug, Clone)]
93pub struct NativePhysicalState {
94    pub header: PhysicalFileHeader,
95    pub collection_roots: BTreeMap<String, u64>,
96    pub manifest: Option<NativeManifestSummary>,
97    pub registry: Option<NativeRegistrySummary>,
98    pub recovery: Option<NativeRecoverySummary>,
99    pub catalog: Option<NativeCatalogSummary>,
100    pub metadata_state: Option<NativeMetadataStateSummary>,
101    pub vector_artifact_pages: Option<Vec<NativeVectorArtifactPageSummary>>,
102}
103
104// ============================================================================
105// Configuration
106// ============================================================================
107
108/// Configuration for UnifiedStore
109#[derive(Debug, Clone)]
110pub struct UnifiedStoreConfig {
111    /// Configuration for segment managers
112    pub manager_config: ManagerConfig,
113    /// Automatically index cross-references on insert
114    pub auto_index_refs: bool,
115    /// Automatically build a HASH index on a user `id` column the first
116    /// time a row carrying that column is inserted into a collection.
117    /// Mirrors PostgreSQL's implicit primary-key index and Mongo's `_id`
118    /// default index — without it, `WHERE id = N` falls through to a
119    /// full segment scan because RedDB has no concept of an automatic
120    /// primary-key index on user-declared columns. See `docs/perf/
121    /// delete-sequential-2026-05-06.md` for the perf rationale.
122    /// Defaults to `true`; set to `false` to opt out per workload.
123    pub auto_index_id: bool,
124    /// Maximum cross-references per entity
125    pub max_cross_refs: usize,
126    /// Enable write-ahead logging
127    pub enable_wal: bool,
128    /// Durability profile for paged writes.
129    pub durability_mode: DurabilityMode,
130    /// Group-commit batching knobs when using grouped durability.
131    pub group_commit: GroupCommitOptions,
132    /// Data directory path
133    pub data_dir: Option<std::path::PathBuf>,
134    /// Embedded single-file artifact used for the internal WAL stream.
135    pub embedded_wal_path: Option<std::path::PathBuf>,
136    /// Preallocated page-cache slot count (ADR 0073 §2). `None` leaves the
137    /// pager on its own structural default; the promoted boot path always
138    /// supplies a count derived from the page-cache budget share, so the
139    /// default is reached only by direct library callers.
140    pub page_cache_slots: Option<usize>,
141    /// Shared resident arena allowance for disk-backed collections. The host
142    /// supplies the segment share of its process memory budget.
143    pub resident_memory_bytes_max: Option<u64>,
144}
145
146impl Default for UnifiedStoreConfig {
147    fn default() -> Self {
148        Self {
149            manager_config: ManagerConfig::default(),
150            auto_index_refs: true,
151            auto_index_id: true,
152            max_cross_refs: 1000,
153            enable_wal: false,
154            // Mirrors `RedDBOptions::default().durability_mode` — see
155            // `src/api.rs` for the rationale.
156            durability_mode: DurabilityMode::WalDurableGrouped,
157            group_commit: GroupCommitOptions::default(),
158            data_dir: None,
159            embedded_wal_path: None,
160            page_cache_slots: None,
161            resident_memory_bytes_max: None,
162        }
163    }
164}
165
166impl UnifiedStoreConfig {
167    /// Create config with data directory
168    pub fn with_data_dir(mut self, path: impl Into<std::path::PathBuf>) -> Self {
169        self.data_dir = Some(path.into());
170        self
171    }
172
173    /// Enable WAL
174    pub fn with_wal(mut self) -> Self {
175        self.enable_wal = true;
176        self
177    }
178
179    pub fn with_durability_mode(mut self, mode: DurabilityMode) -> Self {
180        self.durability_mode = mode;
181        self
182    }
183
184    pub fn with_group_commit(mut self, options: GroupCommitOptions) -> Self {
185        self.group_commit = options;
186        self
187    }
188
189    pub fn with_embedded_wal_path(mut self, path: impl Into<std::path::PathBuf>) -> Self {
190        self.embedded_wal_path = Some(path.into());
191        self
192    }
193
194    /// Pre-size the page cache from its budget share (ADR 0073 §2).
195    pub fn with_page_cache_slots(mut self, slots: usize) -> Self {
196        self.page_cache_slots = Some(slots);
197        self
198    }
199
200    /// Set max cross-references
201    pub fn with_max_refs(mut self, max: usize) -> Self {
202        self.max_cross_refs = max;
203        self
204    }
205}
206
207// ============================================================================
208// Error Types
209// ============================================================================
210
211pub use reddb_engine::storage::error::StoreError;
212
213// ============================================================================
214// Statistics
215// ============================================================================
216
217/// Statistics for UnifiedStore
218#[derive(Debug, Clone, Default)]
219pub struct StoreStats {
220    /// Number of collections
221    pub collection_count: usize,
222    /// Total entities across all collections
223    pub total_entities: usize,
224    /// Total memory usage in bytes
225    pub total_memory_bytes: usize,
226    /// Per-collection statistics
227    pub collections: HashMap<String, ManagerStats>,
228    /// Total cross-references
229    pub cross_ref_count: usize,
230}
231
232impl StoreStats {
233    /// Get average entities per collection
234    pub fn avg_entities_per_collection(&self) -> f64 {
235        if self.collection_count == 0 {
236            0.0
237        } else {
238            self.total_entities as f64 / self.collection_count as f64
239        }
240    }
241
242    /// Get memory in MB
243    pub fn memory_mb(&self) -> f64 {
244        self.total_memory_bytes as f64 / (1024.0 * 1024.0)
245    }
246}
247
248// ============================================================================
249// UnifiedStore - The Main API
250// ============================================================================
251
252struct CheckpointVersion {
253    manager: std::sync::Weak<SegmentManager>,
254    segments: Vec<(u64, u64)>,
255    visibility_revision: u64,
256    untracked_visibility_revision: u64,
257    excluded_xids: Arc<std::collections::HashSet<u64>>,
258    pending_vault: Arc<std::collections::HashSet<EntityId>>,
259    format_version: u32,
260    /// Raw rows omitted or normalized in the durable image. None retains the
261    /// full validation path when more than 1,024 such rows are outstanding.
262    /// Also carries bounded unprojected IDs across completed inline GC batches.
263    deferred_rows: Option<Arc<Vec<EntityId>>>,
264    /// The resident image includes rows omitted or normalized in the checkpoint.
265    has_unprojected_rows: bool,
266}
267
268impl CheckpointVersion {
269    fn remember_deferred_row(&mut self, id: EntityId) {
270        self.has_unprojected_rows = true;
271        let Some(rows) = self.deferred_rows.as_mut() else {
272            return;
273        };
274        if rows.len() == 1024 {
275            self.deferred_rows = None;
276        } else {
277            Arc::make_mut(rows).push(id);
278        }
279    }
280}
281
282/// Unified storage for tables, graphs, and vectors
283///
284/// UnifiedStore provides a single coherent interface for all data types:
285/// - **Tables**: Row-based data with columns
286/// - **Graphs**: Nodes and edges with labels
287/// - **Vectors**: Embeddings for similarity search
288///
289/// # Features
290///
291/// - Multi-collection management
292/// - Cross-collection queries
293/// - Cross-reference tracking between entities
294/// - Automatic ID generation
295/// - Segment-based storage with growing/sealed lifecycle
296///
297/// # Example
298///
299/// ```ignore
300/// use reddb::storage::{Entity, Store};
301///
302/// let store = Store::new();
303///
304/// // Create a collection
305/// store.create_collection("hosts")?;
306///
307/// // Insert an entity
308/// let entity = Entity::table_row(1, "hosts", 1, vec![]);
309/// let id = store.insert("hosts", entity)?;
310///
311/// // Query
312/// let found = store.get("hosts", id);
313/// ```
314pub struct UnifiedStore {
315    /// Canonical resident state; native fields below retain durability policy.
316    resident: reddb_engine::storage::resident_store::ResidentStore,
317    /// Store configuration
318    config: UnifiedStoreConfig,
319    /// File format version for serialization
320    format_version: AtomicU32,
321    /// Optional page-based storage via Pager
322    pager: Option<Arc<Pager>>,
323    /// Database file path (for paged mode)
324    db_path: Option<PathBuf>,
325    /// B-tree indices for O(log n) entity lookups by ID (per collection).
326    /// Stored as `Arc<BTree>` so hot-path callers can clone the handle out
327    /// under a read lock and release the map-level lock before doing the
328    /// actual insert — previously the outer RwLock was held for the whole
329    /// btree mutation, serialising every concurrent insert across every
330    /// collection into one global write lock.
331    btree_indices: RwLock<HashMap<String, Arc<BTree>>>,
332    key_indices: RwLock<HashMap<String, PublishedKeyIndex>>,
333    secondary_index_specs: RwLock<secondary_index::SecondaryIndexSpecs>,
334    recovery_xid_summary: RwLock<Option<recovery_xid::RecoveryXidSummary>>,
335    /// Reuse only a successfully published generation with the same live rows.
336    checkpoint_versions: RwLock<HashMap<String, CheckpointVersion>>,
337    checkpoint_durable_rows: RwLock<checkpoint_delta::DurableRows>,
338    /// The immediately previous published trees and any older reader-pinned
339    /// trees. Root identities are persisted with the paged catalog.
340    retired_btrees: parking_lot::Mutex<Vec<RetiredBTree>>,
341    /// Epoch in the durable catalog and the epoch being published, if any.
342    ownership_catalog_epoch: AtomicU64,
343    ownership_active_epoch: AtomicU64,
344    /// Detached artifact pages stay unavailable until native metadata is durable.
345    detached_native_blob_pages: parking_lot::Mutex<std::collections::BTreeSet<u32>>,
346    /// Logical store WAL / grouped durability coordinator for paged mode.
347    commit: Option<Arc<StoreCommitCoordinator>>,
348    /// WAL-replayed `VectorInsert` records, captured at open time and
349    /// drained per-collection on first `vector.turbo` access (issue
350    /// #694). Boot-time recovery: the in-memory TurboQuant index is
351    /// rebuilt by replaying these FP32 vectors in WAL order, so the
352    /// rebuilt state is byte-deterministic against the pre-restart
353    /// state under a fixed codec seed.
354    pub(crate) replayed_turbo_inserts: parking_lot::Mutex<HashMap<String, Vec<(u64, Vec<f32>)>>>,
355    /// WAL-replayed probabilistic mutation deltas captured at open time.
356    /// Runtime recovery drains these after loading the latest full
357    /// probabilistic snapshot from store state.
358    pub(crate) replayed_probabilistic_deltas:
359        parking_lot::Mutex<Vec<(u8, u8, String, Vec<Vec<u8>>)>>,
360    /// Serialize vault publication with snapshot capture and WAL truncation.
361    /// Physical metadata and native header publication take it too (#2376).
362    /// Reentrant because embedded checkpoint calls the snapshot encoder while
363    /// holding the barrier, and storage maintenance may nest those calls.
364    pub(crate) vault_publication_lock: parking_lot::ReentrantMutex<()>,
365    /// Live-only vault rows excluded from a reentrant WAL-full compaction.
366    pub(crate) pending_vault_versions: RwLock<std::collections::HashSet<EntityId>>,
367    /// Retain transaction outcomes through embedded snapshot-on-drop. Without
368    /// this, aborted physical versions are reopened as committed rows.
369    pub(crate) snapshot_manager:
370        std::sync::OnceLock<Arc<crate::storage::transaction::snapshot::SnapshotManager>>,
371    embedded_commit_session: std::sync::OnceLock<crate::storage::embedded::EmbeddedCommitSession>,
372    #[cfg(test)]
373    pub(crate) test_fail_deferred_wal_append: std::sync::atomic::AtomicBool,
374    #[cfg(test)]
375    pub(crate) test_require_deferred_publication_fence: std::sync::atomic::AtomicBool,
376}
377
378#[derive(Clone)]
379struct PublishedKeyIndex {
380    indexed: bool,
381    source: std::sync::Weak<BTree>,
382    tree: Arc<BTree>,
383    secondary: BTreeMap<String, secondary_index::PublishedSecondaryIndex>,
384}
385
386struct RetiredBTree {
387    tree: Arc<BTree>,
388    published: bool,
389}
390
391mod builder;
392mod checkpoint_cold_append;
393mod checkpoint_delta;
394mod checkpoint_wal;
395mod commit;
396mod impl_entities;
397mod impl_file;
398mod impl_native_a;
399mod impl_native_b;
400mod impl_native_c;
401mod impl_pages;
402mod impl_vault_import;
403mod key_index;
404mod keyed_write;
405mod mvcc_reclaim;
406pub(crate) use mvcc_reclaim::MvccReclaimCursor;
407pub(crate) mod count_index;
408mod native_helpers;
409mod recovery_xid;
410pub mod secondary_index;
411
412pub(crate) use self::commit::DeferredStoreWalActions;
413use self::commit::{StoreCommitCoordinator, StoreWalAction};
414use self::native_helpers::*;
415
416#[cfg(test)]
417mod shared_state_tests {
418    use super::*;
419
420    #[test]
421    fn native_mutations_and_shared_reads_use_one_resident_state() {
422        let store = UnifiedStore::new();
423        let resident: &reddb_engine::storage::resident_store::ResidentStore = &store.resident;
424        store.create_collection("nodes").expect("collection");
425        let native = store.get_collection("nodes").expect("native manager");
426        let shared = resident.get_collection("nodes").expect("shared manager");
427        assert!(
428            Arc::ptr_eq(&native, &shared),
429            "one canonical collection owner"
430        );
431        let id = store
432            .insert_auto(
433                "nodes",
434                UnifiedEntity::graph_node(EntityId::new(0), "Person", "person", HashMap::new()),
435            )
436            .expect("native mutation");
437        assert_eq!(id.raw(), 1);
438        assert_eq!(
439            resident.get_any(id).expect("shared read").1.kind,
440            store.get("nodes", id).expect("native read").kind
441        );
442        assert_eq!(
443            resident.lookup_graph_nodes_by_label_in("nodes", "Person"),
444            [id]
445        );
446        assert_eq!(resident.entity_cache_stats().entries, 1);
447        assert!(store.delete("nodes", id).expect("native delete"));
448        assert!(resident.get_any(id).is_none());
449        assert!(resident.lookup_graph_nodes_by_label("Person").is_empty());
450        assert_eq!(resident.entity_cache_stats().entries, 0);
451        store.set_aux_metadata(b"catalog".to_vec());
452        assert_eq!(resident.aux_metadata(), b"catalog");
453        assert_eq!(
454            store.next_entity_id().raw(),
455            2,
456            "native allocator is the shared allocator"
457        );
458    }
459}
reddb_server/storage/unified/store.rs

reddb_server/storage/unified/
store.rs
