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

reddb_server/storage/unified/
store.rs
