use super::*;
impl UnifiedStore {
pub fn new() -> Self {
Self::with_config(UnifiedStoreConfig::default())
}
pub fn format_version(&self) -> u32 {
self.format_version.load(Ordering::SeqCst)
}
pub(crate) fn set_format_version(&self, version: u32) {
self.format_version.store(version, Ordering::SeqCst);
}
pub fn next_entity_id(&self) -> EntityId {
self.resident.next_entity_id()
}
pub fn reserve_entity_ids(&self, n: u64) -> std::ops::Range<u64> {
self.resident.reserve_entity_ids(n)
}
pub(crate) fn register_entity_id(&self, id: EntityId) {
self.resident.register_entity_id(id)
}
pub fn load_from_file(path: &Path) -> Result<Self, Box<dyn std::error::Error>> {
let file = File::open(path)?;
let mut reader = BufReader::new(file);
let mut buf = Vec::new();
reader.read_to_end(&mut buf)?;
Self::load_from_bytes(&buf)
}
pub(crate) fn load_from_bytes(buf: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
Self::load_from_bytes_with_config(buf, UnifiedStoreConfig::default())
}
pub(crate) fn load_from_bytes_with_config(
buf: &[u8],
config: UnifiedStoreConfig,
) -> Result<Self, Box<dyn std::error::Error>> {
let (resident, version) =
reddb_engine::storage::resident_store::ResidentStore::from_snapshot(
buf,
&config.manager_config,
config.auto_index_refs,
config.max_cross_refs,
)?;
let mut store = Self::with_config(config);
store.resident = resident;
store.set_format_version(version.max(STORE_VERSION_CURRENT));
Ok(store)
}
pub fn save_to_file(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
let buf = self.to_binary_dump_bytes();
reddb_file::write_native_store_bytes_atomically(path, &buf)?;
Ok(())
}
pub(crate) fn to_binary_dump_bytes(&self) -> Vec<u8> {
let _vault_guard = self.vault_publication_lock.lock();
let pending_vault_versions = self.pending_vault_versions.read();
let (_, excluded) = self.snapshot_manager.get().map_or_else(
|| (0, Default::default()),
|manager| manager.checkpoint_excluded_state(),
);
self.resident
.snapshot_bytes(&pending_vault_versions, &excluded, || {
self.set_format_version(STORE_VERSION_CURRENT);
})
}
pub(crate) fn read_entity_binary(
buf: &[u8],
pos: &mut usize,
format_version: u32,
) -> Result<UnifiedEntity, Box<dyn std::error::Error>> {
Ok(reddb_engine::storage::entity_codec::read_entity_binary(
buf,
pos,
format_version,
)?)
}
pub(crate) fn write_entity_binary(
buf: &mut Vec<u8>,
entity: &UnifiedEntity,
format_version: u32,
) {
reddb_engine::storage::entity_codec::write_entity_binary(buf, entity, format_version);
}
}
#[cfg(test)]
mod aux_metadata_dump_tests {
use super::*;
use std::collections::HashMap;
#[test]
fn snapshot_load_preserves_existing_wal_and_restores_future_appends() {
let directory = tempfile::tempdir().expect("directory");
let path = directory.path().join("snapshot.rdb");
let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
let old =
UnifiedEntity::table_row(EntityId::new(1), "rows", 1, vec![Value::text("snapshot")]);
store.insert_auto("rows", old).expect("snapshot row");
let snapshot = store.to_binary_dump_bytes();
crate::storage::EmbeddedRdbArtifact::create_with_snapshot(&path, &snapshot)
.expect("artifact");
let newer = UnifiedEntity::table_row(
EntityId::new(1),
"rows",
1,
vec![Value::text("durable WAL")],
);
let action = StoreWalAction::upsert_entity("rows", &newer, None, STORE_VERSION_CURRENT);
UnifiedStore::with_config(UnifiedStoreConfig::default().with_embedded_wal_path(&path))
.finish_paged_write([action])
.expect("durable update after snapshot");
let before = std::fs::read(&path).expect("original artifact bytes");
let loaded = UnifiedStore::load_from_bytes_with_config(
&snapshot,
UnifiedStoreConfig::default().with_embedded_wal_path(&path),
)
.expect("reconstruct snapshot");
assert_eq!(std::fs::read(&path).expect("artifact after load"), before);
assert_eq!(
loaded.config().embedded_wal_path.as_deref(),
Some(path.as_path())
);
let generation = crate::storage::EmbeddedRdbArtifact::open(&path)
.expect("boundary before future write")
.selected_superblock
.generation;
loaded
.insert_auto(
"rows",
UnifiedEntity::table_row(
EntityId::new(2),
"rows",
2,
vec![Value::text("future write")],
),
)
.expect("future writes still append WAL");
assert_eq!(
crate::storage::EmbeddedRdbArtifact::open(&path)
.expect("boundary after future write")
.selected_superblock
.generation,
generation + 1
);
}
#[test]
fn aux_metadata_round_trips_through_binary_dump() {
let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
store.create_collection("k").expect("create collection");
store.set_aux_metadata(b"contract-blob".to_vec());
let bytes = store.to_binary_dump_bytes();
let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
assert_eq!(reloaded.aux_metadata(), b"contract-blob".to_vec());
}
#[test]
fn empty_aux_metadata_round_trips_as_empty() {
let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
let bytes = store.to_binary_dump_bytes();
let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
assert!(reloaded.aux_metadata().is_empty());
}
#[test]
fn interned_timeseries_series_reduces_repeated_tag_storage() {
let tags = HashMap::from([
(
"host".to_string(),
"srv-very-long-hostname-0001".to_string(),
),
("region".to_string(), "us-east-1".to_string()),
("service".to_string(), "checkout-api".to_string()),
]);
let inline_bytes: usize = (0..100)
.map(|timestamp_ns| {
let entity = UnifiedEntity::new(
EntityId::new(0),
EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
series: "cpu_metrics".to_string(),
metric: "cpu.idle".to_string(),
})),
EntityData::TimeSeries(crate::storage::TimeSeriesData {
metric: "cpu.idle".to_string(),
series_id: None,
timestamp_ns,
value: 94.8,
tags: tags.clone(),
fields: HashMap::new(),
}),
);
UnifiedStore::serialize_entity_record(&entity, None, reddb_file::STORE_VERSION_V10)
.len()
})
.sum();
let mut dictionary_fields = HashMap::new();
dictionary_fields.insert("collection".to_string(), Value::text("cpu_metrics"));
dictionary_fields.insert("series_id".to_string(), Value::UnsignedInteger(0));
dictionary_fields.insert("metric".to_string(), Value::text("cpu.idle"));
dictionary_fields.insert(
"canonical_tags".to_string(),
Value::text(
r#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#,
),
);
dictionary_fields.insert(
"tags".to_string(),
Value::Json(
br#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#
.to_vec(),
),
);
let dictionary = UnifiedEntity::new(
EntityId::new(0),
EntityKind::TableRow {
table: Arc::from("red_timeseries_series"),
row_id: 0,
},
EntityData::Row(RowData {
columns: Vec::new(),
named: Some(dictionary_fields),
schema: None,
}),
);
let dictionary_bytes =
UnifiedStore::serialize_entity_record(&dictionary, None, STORE_VERSION_V11).len();
let interned_bytes = dictionary_bytes
+ (0..100)
.map(|timestamp_ns| {
let entity = UnifiedEntity::new(
EntityId::new(0),
EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
series: "cpu_metrics".to_string(),
metric: "cpu.idle".to_string(),
})),
EntityData::TimeSeries(crate::storage::TimeSeriesData {
metric: "cpu.idle".to_string(),
series_id: Some(0),
timestamp_ns,
value: 94.8,
tags: HashMap::new(),
fields: HashMap::new(),
}),
);
UnifiedStore::serialize_entity_record(&entity, None, STORE_VERSION_V11).len()
})
.sum::<usize>();
assert!(
interned_bytes < inline_bytes,
"interned series storage should shrink repeated tags: inline={inline_bytes}, interned={interned_bytes}"
);
}
}