reddb_server/storage/unified/devx/
reddb.rs

1//! RedDB - Main Entry Point
2//!
3//! Unified Database with best-in-class developer experience for Tables, Graphs, and Vectors.
4
5use std::collections::HashMap;
6use std::fs;
7use std::fs::File;
8use std::io::Read;
9use std::path::{Path, PathBuf};
10use std::sync::{Arc, RwLock};
11use std::time::{SystemTime, UNIX_EPOCH};
12use std::{
13    collections::{BTreeMap, BTreeSet},
14    fmt::Debug,
15};
16
17use super::super::{
18    EntityData, EntityId, EntityKind, StoreStats, UnifiedEntity, UnifiedStore, UnifiedStoreConfig,
19};
20use super::batch::BatchBuilder;
21use super::builders::{
22    DocumentBuilder, EdgeBuilder, KvBuilder, NodeBuilder, RowBuilder, VectorBuilder,
23};
24use super::helpers::cosine_similarity;
25use super::query::QueryBuilder;
26use super::refs::{NodeRef, TableRef, VectorRef};
27use super::types::{LinkedEntity, SimilarResult};
28use super::{IndexConfig, Preprocessor, SharedPreprocessors};
29use crate::api::{Capability, CatalogSnapshot, CollectionStats, RedDBOptions, StorageMode};
30use crate::catalog::{
31    consistency_report, snapshot_store_with_declarations, CatalogConsistencyReport,
32    CatalogDeclarations, CatalogIndexStatus, CatalogModelSnapshot, CollectionDescriptor,
33    CollectionModel,
34};
35use crate::health::{storage_file_health, HealthReport, HealthState};
36use crate::index::{IndexCatalog, IndexConfig as RuntimeIndexConfig, IndexKind};
37use crate::physical::{
38    ExportDescriptor, PhysicalAnalyticsJob, PhysicalGraphProjection, PhysicalIndexState,
39    PhysicalMetadataFile,
40};
41use crate::replication::{primary::PrimaryReplication, ReplicationRole};
42use crate::serde_json::Value as JsonValue;
43use crate::storage::engine::{HnswIndex, IvfConfig, IvfIndex, IvfStats, PhysicalFileHeader};
44use crate::storage::unified::store::{
45    NativeCatalogCollectionSummary, NativeCatalogSummary, NativeExportSummary,
46    NativeManifestSummary, NativeMetadataStateSummary, NativePhysicalState, NativeRecoverySummary,
47    NativeRegistryIndexSummary, NativeRegistryJobSummary, NativeRegistryProjectionSummary,
48    NativeRegistrySummary, NativeSnapshotSummary, NativeVectorArtifactPageSummary,
49    NativeVectorArtifactSummary,
50};
51use reddb_types::Value;
52
53/// RedDB - Unified Database with Best-in-Class DevX
54///
55/// Single entry point for Tables, Graphs, and Vectors with full
56/// metadata support and cross-referencing.
57pub struct RedDB {
58    store: Arc<UnifiedStore>,
59    /// Preprocessing hooks
60    preprocessors: SharedPreprocessors,
61    /// Index configuration
62    index_config: IndexConfig,
63    /// Persistence path
64    path: Option<PathBuf>,
65    /// Construction/runtime options
66    options: RedDBOptions,
67    /// Whether the current persistence backend is page-based.
68    paged_mode: bool,
69    /// Per-collection HNSW vector index cache for fast approximate nearest neighbor search.
70    /// Lazily built on first vector similarity query per collection.
71    vector_indexes: RwLock<HashMap<String, CachedVectorIndex>>,
72    /// Default TTL policy declared at the collection level, in milliseconds.
73    collection_ttl_defaults_ms: RwLock<HashMap<String, u64>>,
74    /// In-memory cache of collection contracts keyed by collection name.
75    /// Populated lazily from `physical_metadata()` and invalidated on
76    /// `save_collection_contract` / `remove_collection_contract`.
77    /// Avoids reparsing the whole PhysicalMetadataFile JSON on every
78    /// `collection_contract(name)` lookup — which happens 3× per insert
79    /// (ensure_model, enforce_uniqueness, normalize_fields) and dominated
80    /// the insert hot path at ~30%.
81    collection_contract_cache: contract_cache::CollectionContractCache,
82    /// Optional remote storage backend for snapshot transport.
83    pub(crate) remote_backend: Option<Arc<dyn crate::storage::backend::RemoteBackend>>,
84    /// Optional CAS-capable handle for backends that implement
85    /// `AtomicRemoteBackend`. Mirrors `RedDBOptions::remote_backend_atomic`
86    /// — see that field for semantics.
87    pub(crate) remote_backend_atomic: Option<Arc<dyn crate::storage::backend::AtomicRemoteBackend>>,
88    /// Remote object key used by the remote backend.
89    pub(crate) remote_key: Option<String>,
90    /// Primary replication state (only present when role is Primary).
91    pub(crate) replication: Option<Arc<PrimaryReplication>>,
92    /// Quorum coordinator for multi-region commits (Phase 2.6 PG parity).
93    ///
94    /// Only present when role is Primary. Write path calls
95    /// `quorum.wait_for_quorum(lsn)` after appending to the primary WAL
96    /// to block until the configured quorum of replicas has acked. When
97    /// the config is `Async` (default), this returns instantly — same
98    /// semantics as pre-Phase-2.6 RedDB.
99    pub(crate) quorum: Option<Arc<crate::replication::quorum::QuorumCoordinator>>,
100    /// Eventual consistency registry (embedded mode support).
101    pub(crate) ec_registry: Arc<crate::ec::config::EcRegistry>,
102    /// Lazily-initialised ML runtime (model registry + job queue +
103    /// semantic cache). Created on first access by the SQL layer so
104    /// `ML_CLASSIFY`, `SEMANTIC_CACHE_GET/PUT`, and friends have a
105    /// shared handle without forcing every instantiation path to
106    /// know about it.
107    pub(crate) ml_runtime: std::sync::OnceLock<crate::storage::ml::MlRuntime>,
108    /// Shared semantic cache used by `SEMANTIC_CACHE_*` scalars.
109    /// Separate from `MlRuntime` because cache config is runtime-only
110    /// and doesn't need the job queue — keep it a standalone `Arc`.
111    pub(crate) semantic_cache: std::sync::OnceLock<Arc<crate::storage::ml::SemanticCache>>,
112    /// Hypertable registry — populated by `CREATE HYPERTABLE` DDL,
113    /// consumed by chunk routing, retention sweeps, and `SHOW
114    /// HYPERTABLES`. Lazy so startup stays cheap when no hypertables
115    /// exist.
116    pub(crate) hypertables:
117        std::sync::OnceLock<Arc<crate::storage::timeseries::HypertableRegistry>>,
118    /// Continuous-aggregate engine — populated by `CA_REGISTER` and
119    /// queried by `CA_REFRESH` / `CA_STATE` scalars. Same lazy shape
120    /// as the other engine handles.
121    pub(crate) continuous_aggregates: std::sync::OnceLock<
122        Arc<crate::storage::timeseries::continuous_aggregate::ContinuousAggregateEngine>,
123    >,
124    /// Per-collection `vector.turbo` runtime state (issue #693, PRD
125    /// #668). Lazily initialised: the entire map allocation is
126    /// deferred until the first turbo collection is created. Each
127    /// entry owns the in-memory `TurboQuantIndex` + optional
128    /// `TurboExtent` for the collection.
129    pub(crate) turbo_collections: std::sync::OnceLock<
130        Arc<
131            parking_lot::Mutex<
132                std::collections::HashMap<
133                    String,
134                    Arc<crate::runtime::vector_turbo_kind::TurboCollectionState>,
135                >,
136            >,
137        >,
138    >,
139    /// Join handles for `vector.turbo` background-rebuild workers
140    /// (issue #673). Each call to `turbo_state` that materialises a
141    /// fresh `TurboCollectionState` registers a handle here; the
142    /// `Drop` impl below joins every handle before releasing
143    /// `store`, so a runtime restart on the same database path is
144    /// not racy with an in-flight rebuild holding the file lock.
145    pub(crate) turbo_rebuild_workers: parking_lot::Mutex<Vec<std::thread::JoinHandle<()>>>,
146    /// Fences an embedded single-file store to this one writer process for
147    /// as long as it is open. Declared before `_ephemeral_cleanup` so it is
148    /// released before an ephemeral store's directory is removed.
149    _embedded_writer_lock: Option<reddb_file::EmbeddedRdbWriterLock>,
150    _ephemeral_cleanup: Option<EphemeralDataPathCleanup>,
151}
152
153pub(super) struct EphemeralDataPathCleanup {
154    path: PathBuf,
155}
156
157impl EphemeralDataPathCleanup {
158    pub(super) fn new(path: PathBuf) -> Self {
159        Self { path }
160    }
161}
162
163impl Drop for EphemeralDataPathCleanup {
164    fn drop(&mut self) {
165        for path in ephemeral_data_artifacts(&self.path) {
166            if path.is_dir() {
167                let _ = fs::remove_dir_all(path);
168            } else {
169                let _ = fs::remove_file(path);
170            }
171        }
172        // Each ephemeral runtime owns a dedicated `reddb-ephemeral-*`
173        // directory (see `RedDBOptions::in_memory`); remove it wholesale so
174        // temp does not accumulate empty per-instance directories.
175        if let Some(parent) = self.path.parent() {
176            if is_ephemeral_owner_dir(parent) {
177                let _ = fs::remove_dir_all(parent);
178            }
179        }
180    }
181}
182
183/// True when `dir` is a per-instance `reddb-ephemeral-*` directory created
184/// directly under the temp dir by [`crate::api::RedDBOptions::in_memory`].
185fn is_ephemeral_owner_dir(dir: &Path) -> bool {
186    let Some(dir_name) = dir.file_name().and_then(|name| name.to_str()) else {
187        return false;
188    };
189    if !dir_name.starts_with("reddb-ephemeral-") {
190        return false;
191    }
192    dir.parent()
193        .map(|grandparent| grandparent == std::env::temp_dir())
194        .unwrap_or(false)
195}
196
197pub(super) fn is_ephemeral_data_path(path: &Path) -> bool {
198    // Ephemeral runtimes place their data file inside a dedicated
199    // `reddb-ephemeral-*` directory under the temp dir, so the data file
200    // plus every sibling (audit log, WAL, snapshots) is isolated per
201    // instance and cleanup never clobbers another live runtime's artifacts.
202    path.parent().map(is_ephemeral_owner_dir).unwrap_or(false)
203}
204
205fn ephemeral_data_artifacts(data_path: &Path) -> Vec<PathBuf> {
206    let logical_wal_path = reddb_file::layout::logical_wal_path(data_path);
207    let result_cache_l2_path = reddb_file::layout::result_cache_l2_path(data_path);
208    let legacy_logical_slots_path = reddb_file::layout::legacy_logical_slots_path(data_path);
209    let mut operational_manifest_root = data_path.as_os_str().to_os_string();
210    operational_manifest_root.push(".ops");
211    let mut paths = vec![
212        data_path.to_path_buf(),
213        PathBuf::from(operational_manifest_root),
214        reddb_file::layout::unified_wal_path(data_path),
215        logical_wal_path.clone(),
216        reddb_file::layout::logical_wal_temp_path(&logical_wal_path),
217        reddb_file::layout::temp_path(data_path),
218        reddb_file::layout::atomic_temp_path(data_path),
219        result_cache_l2_path.clone(),
220        reddb_file::layout::pager_legacy_wal_path(data_path),
221        reddb_file::layout::engine_wal_path(data_path),
222        reddb_file::layout::pager_dwb_path(data_path),
223        legacy_logical_slots_path.clone(),
224        reddb_file::layout::legacy_logical_slots_temp_path(&legacy_logical_slots_path),
225        reddb_file::layout::legacy_audit_log_path(data_path),
226        reddb_file::layout::shm_path(data_path),
227        reddb_file::layout::physical_metadata_json_path(data_path),
228        reddb_file::layout::physical_metadata_binary_path(data_path),
229        reddb_file::layout::rebootstrap_staging_root(data_path),
230        reddb_file::layout::rebootstrap_pending_path(data_path),
231        reddb_file::layout::rebootstrap_ready_marker_path(data_path),
232        reddb_file::layout::rebootstrap_intent_log_path(data_path),
233        reddb_file::layout::rebootstrap_previous_path(data_path),
234        reddb_file::layout::primary_replica_root(data_path),
235        reddb_file::layout::serverless_root(data_path),
236    ];
237    paths.extend(reddb_file::layout::pager_shadow_sidecar_paths(data_path));
238    paths.extend(reddb_file::layout::pager_shadow_sidecar_paths(
239        &result_cache_l2_path,
240    ));
241    if let Some(parent) = data_path.parent() {
242        paths.push(reddb_file::layout::legacy_slow_query_log_path(parent));
243    }
244    paths.push(reddb_file::layout::support_dir_for(data_path));
245    paths
246}
247
248impl Drop for RedDB {
249    fn drop(&mut self) {
250        if self.options.storage_profile.deploy_profile == crate::storage::DeployProfile::Embedded
251            && self.options.storage_profile.packaging
252                == crate::storage::StoragePackaging::SingleFile
253            && !self.paged_mode
254            && !self.options.read_only
255        {
256            if let Some(path) = &self.path {
257                let _vault_guard = self.store.vault_publication_lock.lock();
258                let snapshot = self.store.to_binary_dump_bytes();
259                let _ = crate::storage::EmbeddedRdbArtifact::write_snapshot(path, &snapshot);
260            }
261        }
262
263        // Issue #673 — wait for every `vector.turbo` background
264        // rebuild worker to exit before our `Arc<UnifiedStore>` is
265        // released. The worker holds a strong handle to the store
266        // during its work phase, so without this join a fast
267        // `RedDBRuntime` restart on the same path observes the
268        // file lock as still held by the soon-to-exit worker.
269        let handles: Vec<_> = self.turbo_rebuild_workers.lock().drain(..).collect();
270        for h in handles {
271            let _ = h.join();
272        }
273        // Issue #674 — also join any in-flight `.tv` snapshot dump
274        // so the on-disk file is complete and renamed before a
275        // restart observes it. Without this, a fast reopen can race
276        // and see the `<path>.tv.tmp` before the atomic rename.
277        if let Some(map) = self.turbo_collections.get() {
278            let states: Vec<_> = map.lock().values().cloned().collect();
279            for state in states {
280                state.wait_snapshot();
281            }
282        }
283    }
284}
285
286/// A cached HNSW index together with the entity count at build time.
287/// When the live entity count diverges the cache is considered stale and
288/// is rebuilt on the next query.
289pub(crate) struct CachedVectorIndex {
290    pub index: Arc<RwLock<HnswIndex>>,
291    pub entity_count: usize,
292}
293
294#[derive(Debug, Clone)]
295pub struct NativeHeaderMismatch {
296    pub field: &'static str,
297    pub native: String,
298    pub expected: String,
299}
300
301#[derive(Debug, Clone)]
302pub struct NativeHeaderInspection {
303    pub native: PhysicalFileHeader,
304    pub expected: PhysicalFileHeader,
305    pub consistent: bool,
306    pub mismatches: Vec<NativeHeaderMismatch>,
307}
308
309#[derive(Debug, Clone, Copy, PartialEq, Eq)]
310pub enum NativeHeaderRepairPolicy {
311    InSync,
312    RepairNativeFromMetadata,
313    NativeAheadOfMetadata,
314}
315
316#[derive(Debug, Clone)]
317pub struct PhysicalAuthorityStatus {
318    pub preference: String,
319    pub sidecar_available: bool,
320    pub native_state_available: bool,
321    pub native_bootstrap_ready: bool,
322    pub native_registry_complete: Option<bool>,
323    pub native_recovery_complete: Option<bool>,
324    pub native_catalog_complete: Option<bool>,
325    pub sidecar_loaded_from: Option<String>,
326    pub native_header_repair_policy: Option<String>,
327    pub metadata_sequence: Option<u64>,
328    pub native_sequence: Option<u64>,
329    pub native_metadata_last_loaded_from: Option<String>,
330    pub native_metadata_generated_at_unix_ms: Option<u128>,
331}
332
333#[derive(Debug, Clone)]
334pub struct NativeVectorArtifactInspection {
335    pub collection: String,
336    pub artifact_kind: String,
337    pub root_page: u32,
338    pub page_count: u32,
339    pub byte_len: u64,
340    pub checksum: u64,
341    pub node_count: u64,
342    pub dimension: u32,
343    pub max_layer: u32,
344    pub total_connections: u64,
345    pub avg_connections: f64,
346    pub entry_point: Option<u64>,
347    pub ivf_n_lists: Option<u32>,
348    pub ivf_non_empty_lists: Option<u32>,
349    pub ivf_trained: Option<bool>,
350    pub graph_edge_count: Option<u64>,
351    pub graph_node_count: Option<u64>,
352    pub graph_label_count: Option<u32>,
353    pub text_doc_count: Option<u64>,
354    pub text_term_count: Option<u64>,
355    pub text_posting_count: Option<u64>,
356    pub document_doc_count: Option<u64>,
357    pub document_path_count: Option<u64>,
358    pub document_value_count: Option<u64>,
359    pub document_unique_value_count: Option<u64>,
360}
361
362#[derive(Debug, Clone)]
363pub struct NativeVectorArtifactBatchInspection {
364    pub inspected_count: usize,
365    pub valid_count: usize,
366    pub artifacts: Vec<NativeVectorArtifactInspection>,
367    pub failures: Vec<(String, String, String)>,
368}
369
370mod contract_cache;
371mod impl_access;
372mod impl_core_a;
373mod impl_core_b;
374mod impl_ec;
375mod impl_metadata;
376mod impl_registry;
377
378impl Default for RedDB {
379    fn default() -> Self {
380        Self::new()
381    }
382}
383
384fn infer_collection_index_kind(model: CollectionModel, index_name: &str) -> IndexKind {
385    match index_name {
386        "graph-adjacency" => IndexKind::GraphAdjacency,
387        "vector-hnsw" => IndexKind::VectorHnsw,
388        "vector-inverted" => IndexKind::VectorInverted,
389        "vector-turbo" => IndexKind::VectorTurbo,
390        "text-fulltext" => IndexKind::FullText,
391        "document-pathvalue" => IndexKind::DocumentPathValue,
392        "search-hybrid" => IndexKind::HybridSearch,
393        _ => match model {
394            CollectionModel::Graph => IndexKind::GraphAdjacency,
395            CollectionModel::Vector => IndexKind::VectorHnsw,
396            CollectionModel::Document => IndexKind::DocumentPathValue,
397            CollectionModel::Kv | CollectionModel::Config | CollectionModel::Vault => {
398                IndexKind::Hash
399            }
400            _ => IndexKind::BTree,
401        },
402    }
403}
404
405fn estimate_index_entries(collection: &CollectionDescriptor, kind: IndexKind) -> usize {
406    match kind {
407        IndexKind::BTree | IndexKind::Hash | IndexKind::Bitmap => collection.entities,
408        IndexKind::GraphAdjacency => collection.cross_refs.max(collection.entities),
409        IndexKind::VectorHnsw | IndexKind::VectorInverted | IndexKind::VectorTurbo => {
410            collection.entities
411        }
412        IndexKind::FullText => collection.entities.saturating_mul(4),
413        IndexKind::DocumentPathValue => collection.entities.saturating_mul(6),
414        IndexKind::HybridSearch => collection.entities,
415    }
416}
417
418fn estimate_index_memory(entries: usize, kind: IndexKind) -> u64 {
419    let per_entry = match kind {
420        IndexKind::BTree => 64,
421        IndexKind::Hash => 48,
422        IndexKind::Bitmap => 2, // Roaring bitmaps are very compact
423        IndexKind::GraphAdjacency => 96,
424        IndexKind::VectorHnsw => 256,
425        IndexKind::VectorInverted => 128,
426        IndexKind::VectorTurbo => 64,
427        IndexKind::FullText => 80,
428        IndexKind::DocumentPathValue => 104,
429        IndexKind::HybridSearch => 144,
430    };
431    (entries as u64).saturating_mul(per_entry)
432}
433
434fn index_backend_name(kind: IndexKind) -> &'static str {
435    match kind {
436        IndexKind::BTree => "page-btree",
437        IndexKind::Hash => "hash-map",
438        IndexKind::Bitmap => "roaring-bitmap",
439        IndexKind::GraphAdjacency => "adjacency-map",
440        IndexKind::VectorHnsw => "vector-hnsw",
441        IndexKind::VectorInverted => "vector-ivf",
442        IndexKind::VectorTurbo => "vector-turbo",
443        IndexKind::FullText => "inverted-text",
444        IndexKind::DocumentPathValue => "document-pathvalue",
445        IndexKind::HybridSearch => "hybrid-score",
446    }
447}
448
449fn fnv1a_seed() -> u64 {
450    0xcbf29ce484222325
451}
452
453fn fnv1a_hash_bytes(hash: &mut u64, bytes: &[u8]) {
454    for byte in bytes {
455        *hash ^= *byte as u64;
456        *hash = hash.wrapping_mul(0x100000001b3);
457    }
458}
459
460fn fnv1a_hash_value<T: Debug>(hash: &mut u64, value: &T) {
461    let rendered = format!("{value:?}");
462    fnv1a_hash_bytes(hash, rendered.as_bytes());
463}
reddb_server/storage/unified/devx/reddb.rs

reddb_server/storage/unified/devx/
reddb.rs
