use std::collections::{BTreeMap, HashMap, HashSet};
use std::fmt;
use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::time::Duration;
use serde_json::Value;
use crate::atomic::AtomicTemporaryFileError;
use crate::catalog::{
check_integrity, declare_relation as declare_relation_in, empty_catalog, expand_indexes, index_identity,
parse_catalog, read_catalog_text, reference_of, relations_touching, write_catalog, write_catalog_sealed, Catalog,
CatalogCompaction, CatalogLayout, ExpandedIndex, IntegrityReport, ReadCatalogError, Relation, RelationAction,
TableEntry, TableOptions, TriggerEntry,
};
use crate::checkpoint::{read_snapshot, RestoreOutcome};
use std::sync::Arc;
use crate::crypto::{CipherOptions, FieldCipher, PasswordHashParams, Sealer};
use crate::errors::{HaysonCipherError, HaysonRelationRestrictError, HaysonRelationViolationError, HaysonSchemaError};
use crate::lock::{DirLock, DirLockError, LockOptions};
use crate::query::builder::QueryBuilder;
use crate::schema::{is_partitioned, parse_schema, FieldSpec, HaysonRecord, TableSchema};
use crate::storage::{partition_names_in, LOG_SUFFIX};
use crate::table::{describe_table_document, CachePolicy, FollowPolicy, Table, TableError};
use crate::txn::{apply_prepared, recover_incomplete_sealed, PreparedUnit, TxnError, UnitOfWork, UnitOfWorkResult};
use super::layout::{resolve_kind, stored_kind_with, Layout, LayoutKind, LayoutOptions, DEFAULT_METADATA_FILE};
const DEFAULT_SNAPSHOT_EVERY: u64 = 1000;
const DEFAULT_COMPACTION: CatalogCompaction = CatalogCompaction::Enabled {
dead_bytes_ratio: 0.4,
min_dead_bytes: 64 * 1024,
};
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
#[derive(Debug)]
pub enum DatabaseError {
Lock(DirLockError),
Schema(HaysonSchemaError),
ReadOnly {
dir: PathBuf,
operation: String,
},
FieldInUse {
table: String,
field: String,
records: usize,
},
Catalog(String),
Io(io::Error),
Log(PathBuf, io::Error),
Table(String, TableError),
Txn(TxnError),
Cipher(Box<HaysonCipherError>),
}
impl fmt::Display for DatabaseError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
DatabaseError::Lock(error) => error.fmt(f),
DatabaseError::Schema(error) => error.fmt(f),
DatabaseError::ReadOnly { dir, operation } => {
write!(f, "database \"{}\" is readonly — {operation} is not available", dir.display())
}
DatabaseError::FieldInUse { table, field, records } => write!(
f,
"field \"{field}\" of table \"{table}\" still holds a value in {records} live record(s) — pass force to drop it anyway"
),
DatabaseError::Catalog(message) => write!(f, "{message}"),
DatabaseError::Io(error) => error.fmt(f),
DatabaseError::Log(path, error) => write!(f, "{error}, open \"{}\"", path.display()),
DatabaseError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
DatabaseError::Txn(error) => error.fmt(f),
DatabaseError::Cipher(error) => error.fmt(f),
}
}
}
impl std::error::Error for DatabaseError {}
impl From<io::Error> for DatabaseError {
fn from(value: io::Error) -> Self {
DatabaseError::Io(value)
}
}
impl From<DirLockError> for DatabaseError {
fn from(value: DirLockError) -> Self {
DatabaseError::Lock(value)
}
}
impl From<HaysonSchemaError> for DatabaseError {
fn from(value: HaysonSchemaError) -> Self {
DatabaseError::Schema(value)
}
}
impl From<TxnError> for DatabaseError {
fn from(value: TxnError) -> Self {
DatabaseError::Txn(value)
}
}
impl From<HaysonCipherError> for DatabaseError {
fn from(value: HaysonCipherError) -> Self {
DatabaseError::Cipher(Box::new(value))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum FollowOption {
#[default]
Default,
Off,
Every(Duration),
}
pub const DEFAULT_FOLLOW_INTERVAL: Duration = Duration::ZERO;
impl FollowOption {
fn resolve(self, readonly: bool) -> Option<FollowPolicy> {
match self {
FollowOption::Off => None,
FollowOption::Every(interval) => Some(FollowPolicy { interval }),
FollowOption::Default if readonly => Some(FollowPolicy {
interval: DEFAULT_FOLLOW_INTERVAL,
}),
FollowOption::Default => None,
}
}
}
#[derive(Debug, Clone, Copy)]
pub enum LockMode {
Enabled(LockOptions),
Skip,
}
impl Default for LockMode {
fn default() -> Self {
LockMode::Enabled(LockOptions::default())
}
}
impl LockMode {
pub fn options(&self) -> Option<&LockOptions> {
match self {
LockMode::Enabled(options) => Some(options),
LockMode::Skip => None,
}
}
pub fn options_mut(&mut self) -> Option<&mut LockOptions> {
match self {
LockMode::Enabled(options) => Some(options),
LockMode::Skip => None,
}
}
}
#[derive(Debug, Clone)]
pub struct OpenOptions {
pub readonly: bool,
pub lock: LockMode,
pub snapshot_every: u64,
pub compaction: Option<CatalogCompaction>,
pub cypher: Option<CipherOptions>,
pub follow: FollowOption,
pub cache: CachePolicy,
pub max_image_bytes: Option<u64>,
pub password_hash: PasswordHashParams,
pub encryption: Option<CipherOptions>,
pub layout: LayoutOptions,
}
impl Default for OpenOptions {
fn default() -> Self {
OpenOptions {
readonly: false,
lock: LockMode::default(),
snapshot_every: DEFAULT_SNAPSHOT_EVERY,
compaction: None,
cypher: None,
follow: FollowOption::default(),
cache: CachePolicy::default(),
max_image_bytes: None,
password_hash: PasswordHashParams::default(),
encryption: None,
layout: LayoutOptions::default(),
}
}
}
fn now_ms() -> u64 {
chrono::Utc::now().timestamp_millis().max(0) as u64
}
fn scalar_as_text(value: &Value) -> String {
match value {
Value::String(s) => s.clone(),
other => other.to_string(),
}
}
fn fnv1a(bytes: &[u8]) -> String {
let mut hash = FNV_OFFSET;
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(FNV_PRIME);
}
format!("{hash:016x}")
}
fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
let schema = format!("{schema:?}").replace(", password_hash: false", "");
if declared.is_empty() {
return fnv1a(schema.as_bytes());
}
let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
described.sort();
fnv1a(format!("{schema}|{}", described.join("|")).as_bytes())
}
fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
let fields: BTreeMap<String, FieldSpec> = entry
.fields
.iter()
.map(|(field, spec)| (field.clone(), spec.spec.clone()))
.collect();
TableSchema {
table: name.to_string(),
revision: entry.revision as u32,
primary_key: entry.primary_key.clone(),
partition_by: entry.partition_by.clone(),
fields,
}
}
fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
let Some(declared) = catalog.triggers.as_ref() else {
return Vec::new();
};
declared
.iter()
.filter(|trigger| trigger.table == name)
.cloned()
.collect()
}
fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
for (name, table) in tables.iter_mut() {
table.set_triggers(triggers_of(catalog, name));
}
}
fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
entry
.options
.as_ref()
.and_then(&pick)
.or_else(|| catalog.defaults.as_ref().and_then(&pick))
}
fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
table.set_image_dir(dir.join(image_dir));
}
}
fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
table.set_sync(sync);
}
}
fn apply_validation(entry: &TableEntry, table: &mut Table) {
if let Some(validation) = entry.validation {
table.set_validation(validation);
}
}
fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
let marked: BTreeMap<String, String> = entry
.fields
.iter()
.filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
.collect();
if !marked.is_empty() {
table.set_deprecated_fields(marked);
}
}
fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
table.set_link_parsing(link_parsing);
}
}
fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
named
.or_else(|| {
entry
.and_then(|entry| entry.options.as_ref())
.and_then(|options| options.compaction)
})
.or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
.unwrap_or(DEFAULT_COMPACTION)
}
fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
table.set_compaction(compaction);
}
fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
table.set_cipher(cipher.cloned());
}
fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
let all = open_table
.query(&QueryBuilder::new().spec())
.map_err(|error| DatabaseError::Table(name.to_string(), error))?;
let mut residual = 0usize;
for key in &all.keys {
let record = open_table
.get(key)
.map_err(|error| DatabaseError::Table(name.to_string(), error))?;
if record.is_some_and(|record| record.contains_key(field)) {
residual += 1;
}
}
if residual > 0 {
return Err(DatabaseError::FieldInUse {
table: name.to_string(),
field: field.to_string(),
records: residual,
});
}
Ok(())
}
fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
match error.downcast::<io::Error>() {
Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
Ok(io) => DatabaseError::Catalog(io.to_string()),
Err(other) => DatabaseError::Catalog(other.to_string()),
}
}
fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
layout.tables().join(table)
}
fn on_disk(path: &Path) -> bool {
fs::symlink_metadata(path).is_ok()
}
fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
let flat = layout.log_path(&schema.table);
if !is_partitioned(schema) {
return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
}
let dir = partition_dir_of(layout, &schema.table);
let present = partition_names_in(&dir);
if present.is_empty() && on_disk(&flat) {
return Ok(Vec::new());
}
let claimed = read_snapshot(index_path)?
.and_then(|snapshot| snapshot.header.parts)
.unwrap_or_default();
Ok(claimed
.into_iter()
.filter(|part| !present.contains(&part.part))
.map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
.collect())
}
fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
let flat = layout.log_path(&schema.table);
if !is_partitioned(schema) {
return if on_disk(&flat) { vec![flat] } else { Vec::new() };
}
let dir = partition_dir_of(layout, &schema.table);
let present = partition_names_in(&dir);
if present.is_empty() {
return if on_disk(&flat) { vec![flat] } else { Vec::new() };
}
present
.iter()
.map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
.collect()
}
enum TableDrift {
Absent,
Schema { revision: u64 },
Indexes { unserved: Vec<String>, stale: Vec<String> },
}
struct DriftedTable {
name: String,
drift: TableDrift,
}
fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
let identities: HashSet<String> = present.iter().map(index_identity).collect();
let mut names: Vec<String> = expected
.iter()
.filter(|index| !identities.contains(&index_identity(index)))
.map(|index| index.name.clone())
.collect();
names.sort();
names
}
fn shape_of(schema: &TableSchema) -> TableSchema {
let mut bare = schema.clone();
for spec in bare.fields.values_mut() {
spec.index = None;
spec.unique = false;
spec.fulltext = false;
}
bare
}
fn quoted(names: &[String]) -> String {
names
.iter()
.map(|name| format!("\"{name}\""))
.collect::<Vec<_>>()
.join(", ")
}
fn stored_side_of(drift: &TableDrift) -> String {
match drift {
TableDrift::Absent => "declares no table at all".to_string(),
TableDrift::Schema { revision } => format!("declares revision {revision}"),
TableDrift::Indexes { unserved, stale } => {
let mut halves = Vec::new();
if !unserved.is_empty() {
halves.push(format!(
"declares indexes this handle does not serve: {}",
quoted(unserved)
));
}
if !stale.is_empty() {
halves.push(format!(
"no longer declares indexes this handle still serves: {}",
quoted(stale)
));
}
halves.join(", and ")
}
}
}
fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
let listed = missing
.iter()
.map(|path| format!("\"{}\"", path.display()))
.collect::<Vec<_>>()
.join(", ");
let one = missing.len() == 1;
let (its, is, it, file, log) = if one {
("its log".to_string(), "is", "it", "file", "log")
} else {
(format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
};
HaysonSchemaError(format!(
"the catalog of \"{}\" declares table \"{name}\", but {its} {listed} {is} not on disk — this handle is readonly, so nothing here will write {it}; restore the {file}, or open the database for writing to have the {log} created",
dir.display()
))
}
fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
let schema = table_schema_of(name, entry);
let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
if !missing.is_empty() {
return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
}
for path in present_logs_of(layout, &schema) {
fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
}
Ok(())
}
#[allow(clippy::too_many_arguments)]
fn open_catalog_table(
layout: &Layout,
name: &str,
catalog: &Catalog,
entry: &TableEntry,
snapshot_every: u64,
cipher: Option<&FieldCipher>,
follow: Option<FollowPolicy>,
readonly: bool,
compaction: CatalogCompaction,
cache: CachePolicy,
max_image_bytes: Option<u64>,
password_hash: PasswordHashParams,
sealer: Option<Arc<Sealer>>,
) -> Result<(Table, RestoreOutcome), DatabaseError> {
let schema = table_schema_of(name, entry);
let declared = expand_indexes(name, entry);
let hash = schema_fingerprint(&schema, &declared);
let (mut table, outcome) = Table::open_sealed(
schema,
layout.log_path(name),
layout.index_path(name),
&hash,
snapshot_every,
readonly,
&declared,
sealer,
)
.map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
apply_image_dir(layout.root(), catalog, entry, &mut table);
apply_link_parsing(catalog, entry, &mut table);
apply_sync(catalog, entry, &mut table);
apply_validation(entry, &mut table);
apply_deprecated(entry, &mut table);
apply_compaction(compaction, &mut table);
table.set_cache(cache);
table.set_max_image_bytes(max_image_bytes);
table.set_password_hash_params(password_hash);
apply_cipher(cipher, &mut table);
table.set_follow(follow);
table.set_triggers(triggers_of(catalog, name));
Ok((table, outcome))
}
fn first_table_log(layout: &Layout) -> Option<PathBuf> {
let has_log_suffix = |path: &Path| {
path.extension()
.is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
};
let entries_of = |dir: &Path| -> Vec<PathBuf> {
let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
found.sort();
found
};
let top = entries_of(layout.tables());
if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
return Some(log.clone());
}
top.iter().filter(|path| path.is_dir()).find_map(|dir| {
entries_of(dir)
.into_iter()
.find(|path| path.is_file() && has_log_suffix(path))
})
}
fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
let Some(log) = first_table_log(layout) else {
return Ok(());
};
Err(HaysonSchemaError(format!(
"\"{}\" holds table data but no \"{}\" — refusing to open it as an empty database, which would write a fresh catalog next to \"{}\" and append records of a foreign schema onto it; restore \"{}\" from a backup, or move the logs aside to start over",
layout.root().display(),
layout.metadata_file(),
log.display(),
layout.metadata_file(),
)))
}
fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
read_catalog_with(path, None)
}
fn read_catalog_with(path: &Path, sealer: Option<&Sealer>) -> Result<Option<Catalog>, DatabaseError> {
let raw = match read_catalog_text(path, sealer) {
Ok(Some(raw)) => raw,
Ok(None) => return Ok(None),
Err(ReadCatalogError::Io(error)) => return Err(error.into()),
Err(
ReadCatalogError::Cipher(error)
| ReadCatalogError::Encrypted(error)
| ReadCatalogError::NotEncrypted(error),
) => return Err(DatabaseError::Cipher(Box::new(error))),
};
let value: Value = serde_json::from_str(&raw)
.map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
Ok(Some(parse_catalog(&value)?))
}
pub(super) struct HeldLock(pub(super) Option<DirLock>);
impl HeldLock {
fn into_held(mut self) -> Option<DirLock> {
self.0.take()
}
}
impl Drop for HeldLock {
fn drop(&mut self) {
if let Some(lock) = self.0.as_mut() {
lock.release();
}
}
}
pub struct Database {
layout: Layout,
lock: Option<DirLock>,
catalog: Catalog,
tables: HashMap<String, Table>,
restore_outcomes: HashMap<String, RestoreOutcome>,
readonly: bool,
snapshot_every: u64,
compaction: Option<CatalogCompaction>,
cipher: Option<FieldCipher>,
follow: Option<FollowPolicy>,
cache: CachePolicy,
max_image_bytes: Option<u64>,
password_hash: PasswordHashParams,
sealer: Option<Arc<Sealer>>,
}
impl Database {
pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
Database::open_with(dir, OpenOptions::default())
}
pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
let sealer = options.encryption.as_ref().map(Sealer::new).transpose()?.map(Arc::new);
let (layout, lock) = match options.lock {
LockMode::Enabled(lock) if !options.readonly => {
Database::hold_for_open(dir, &options.layout, lock, sealer.as_deref())?
}
LockMode::Enabled(_) | LockMode::Skip => (
Database::resolve_layout_sealed(dir, &options.layout, sealer.as_deref())?,
HeldLock(None),
),
};
if !options.readonly {
for directory in layout.directories() {
fs::create_dir_all(directory)?;
}
}
Database::open_locked(layout, options, lock, sealer)
}
fn hold_for_open(
dir: &Path,
options: &LayoutOptions,
lock: LockOptions,
sealer: Option<&Sealer>,
) -> Result<(Layout, HeldLock), DatabaseError> {
const SETTLE_ATTEMPTS: usize = 3;
let mut probed = Database::resolve_layout_sealed(dir, options, sealer)?;
for _ in 0..SETTLE_ATTEMPTS {
fs::create_dir_all(probed.runtime())?;
let held = HeldLock(Some(DirLock::acquire(probed.lock_path(), lock)?));
let settled = Database::resolve_layout_sealed(dir, options, sealer)?;
if settled.kind() == probed.kind() {
return Ok((settled, held));
}
drop(held);
if probed.kind() == LayoutKind::Nested {
let _ = fs::remove_dir(probed.runtime());
}
probed = settled;
}
Err(HaysonSchemaError(format!(
"\"{}\" changed arrangement {SETTLE_ATTEMPTS} times while its writer lock was being taken — something keeps moving it without the lock; open it again once the moves are over",
dir.display()
))
.into())
}
pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
Database::resolve_layout_sealed(dir, options, None)
}
pub fn resolve_layout_sealed(
dir: &Path,
options: &LayoutOptions,
sealer: Option<&Sealer>,
) -> Result<Layout, DatabaseError> {
let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
let stored = stored_kind_with(&dir.join(metadata_file), sealer);
let kind = resolve_kind(dir, stored, options.kind)?;
Ok(Layout::new(dir, kind, metadata_file))
}
fn open_locked(
layout: Layout,
options: OpenOptions,
lock: HeldLock,
sealer: Option<Arc<Sealer>>,
) -> Result<Database, DatabaseError> {
let found = read_catalog_with(layout.metadata_path(), sealer.as_deref())?;
let catalog = match found {
Some(catalog) => catalog,
None => {
refuse_orphan_logs(&layout)?;
let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
if !options.readonly {
write_catalog_sealed(layout.metadata_path(), &catalog, sealer.as_deref())
.map_err(|error| DatabaseError::Catalog(error.to_string()))?;
}
catalog
}
};
let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
options.password_hash.validate()?;
let follow = options.follow.resolve(options.readonly);
let mut tables = HashMap::new();
let mut restore_outcomes = HashMap::new();
for (name, entry) in &catalog.tables {
if options.readonly {
refuse_unreadable_logs(&layout, name, entry)?;
}
let (table, outcome) = open_catalog_table(
&layout,
name,
&catalog,
entry,
options.snapshot_every,
cipher.as_ref(),
follow,
options.readonly,
compaction_of(options.compaction, &catalog, Some(entry)),
options.cache,
options.max_image_bytes,
options.password_hash,
sealer.clone(),
)?;
tables.insert(name.clone(), table);
restore_outcomes.insert(name.clone(), outcome);
}
if !options.readonly {
recover_incomplete_sealed(&layout.journal_path(), &mut tables, sealer.as_deref())?;
}
sync_table_triggers(&catalog, &mut tables);
Ok(Database {
layout,
lock: lock.into_held(),
catalog,
tables,
restore_outcomes,
readonly: options.readonly,
snapshot_every: options.snapshot_every,
compaction: options.compaction,
cipher,
follow,
cache: options.cache,
max_image_bytes: options.max_image_bytes,
password_hash: options.password_hash,
sealer,
})
}
pub fn dir(&self) -> &Path {
self.layout.root()
}
pub fn layout(&self) -> &Layout {
&self.layout
}
pub fn follow(&self) -> Option<FollowPolicy> {
self.follow
}
pub fn table_names(&self) -> Vec<&str> {
let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
names.sort_unstable();
names
}
pub fn table(&self, name: &str) -> Option<&Table> {
self.tables.get(name)
}
pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
let _ = self.renew_lock();
self.tables.get_mut(name)
}
fn assert_relations_hold(&self, table: &str, values: &Value) -> Result<(), DatabaseError> {
for relation in relations_touching(&self.catalog, Some(table)) {
if relation.from.table != table || relation.enforce != Some(true) {
continue;
}
let Some(value) = reference_of(values.get(&relation.from.field)) else {
continue;
};
let Some(to_table) = self.tables.get(&relation.to.table) else {
continue;
};
if !to_table.contains_value(&relation.to.field, &value) {
return Err(DatabaseError::Table(
table.to_string(),
TableError::Relation(HaysonRelationViolationError {
relation: relation.name.clone(),
table: table.to_string(),
field: relation.from.field.clone(),
value: scalar_as_text(&value),
}),
));
}
}
Ok(())
}
pub fn put_checked(&mut self, table: &str, document: &Value, force: bool) -> Result<HaysonRecord, DatabaseError> {
self.assert_relations_hold(table, document)?;
let root = self.layout.root().display().to_string();
let handle = self
.table_mut(table)
.ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
handle
.put(document, force)
.map_err(|error| DatabaseError::Table(table.to_string(), error))
}
pub fn patch_checked(
&mut self,
table: &str,
key: &str,
changes: &Value,
force: bool,
) -> Result<HaysonRecord, DatabaseError> {
self.assert_relations_hold(table, changes)?;
let root = self.layout.root().display().to_string();
let handle = self
.table_mut(table)
.ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
handle
.patch(key, changes, force)
.map_err(|error| DatabaseError::Table(table.to_string(), error))
}
pub fn stage_put_checked(
&mut self,
table: &str,
document: &Value,
force: bool,
) -> Result<HaysonRecord, DatabaseError> {
self.assert_relations_hold(table, document)?;
let root = self.layout.root().display().to_string();
let handle = self
.table_mut(table)
.ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
handle
.stage_put(document, force)
.map_err(|error| DatabaseError::Table(table.to_string(), error))
}
pub fn stage_patch_checked(
&mut self,
table: &str,
key: &str,
changes: &Value,
force: bool,
) -> Result<HaysonRecord, DatabaseError> {
self.assert_relations_hold(table, changes)?;
let root = self.layout.root().display().to_string();
let handle = self
.table_mut(table)
.ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
handle
.stage_patch(key, changes, force)
.map_err(|error| DatabaseError::Table(table.to_string(), error))
}
pub fn delete_checked(&mut self, table: &str, key: &str, force: bool) -> Result<bool, DatabaseError> {
if self.readonly {
return Err(DatabaseError::ReadOnly {
dir: self.layout.root().to_path_buf(),
operation: "delete".to_string(),
});
}
let root = self.layout.root().display().to_string();
if self.tables.get(table).is_none() {
return Err(DatabaseError::Catalog(format!(
"table \"{table}\" does not exist in \"{root}\""
)));
}
let incoming = relations_touching(&self.catalog, Some(table))
.into_iter()
.any(|relation| relation.to.table == table);
if !incoming {
self.renew_lock()?;
let handle = self.tables.get_mut(table).expect("checked above");
return handle
.delete(key, force)
.map_err(|error| DatabaseError::Table(table.to_string(), error));
}
let mut unit = UnitOfWork::new();
let mut visited: HashSet<(String, String)> = HashSet::new();
let existed = self.plan_delete_cascade(table, key, &mut visited, &mut unit)?;
if !existed {
return Ok(false);
}
self.commit(&mut unit)?;
Ok(true)
}
fn plan_delete_cascade(
&self,
table: &str,
key: &str,
visited: &mut HashSet<(String, String)>,
unit: &mut UnitOfWork,
) -> Result<bool, DatabaseError> {
if !visited.insert((table.to_string(), key.to_string())) {
return Ok(true);
}
let handle = self
.tables
.get(table)
.ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist")))?;
let Some(record) = handle
.get(key)
.map_err(|error| DatabaseError::Table(table.to_string(), error))?
else {
return Ok(false);
};
for relation in relations_touching(&self.catalog, Some(table)) {
if relation.to.table != table {
continue;
}
let action = relation.on_delete.unwrap_or(RelationAction::NoAction);
if matches!(action, RelationAction::NoAction) {
continue;
}
let Some(value) = reference_of(record.get(&relation.to.field)) else {
continue;
};
let Some(from_table) = self.tables.get(&relation.from.table) else {
continue;
};
let referencing = from_table
.keys_referencing(&relation.from.field, &value)
.map_err(|error| DatabaseError::Table(relation.from.table.clone(), error))?;
if referencing.is_empty() {
continue;
}
match action {
RelationAction::Restrict => {
return Err(DatabaseError::Table(
table.to_string(),
TableError::RelationRestrict(HaysonRelationRestrictError {
relation: relation.name.clone(),
table: table.to_string(),
key: key.to_string(),
referencing_table: relation.from.table.clone(),
referencing_field: relation.from.field.clone(),
referencing: referencing.len(),
}),
));
}
RelationAction::Cascade => {
for referencing_key in &referencing {
self.plan_delete_cascade(&relation.from.table, referencing_key, visited, unit)?;
}
}
RelationAction::SetNull => {
for referencing_key in &referencing {
unit.patch(
relation.from.table.as_str(),
referencing_key.clone(),
Value::Object(serde_json::Map::from_iter([(relation.from.field.clone(), Value::Null)])),
);
}
}
RelationAction::NoAction => unreachable!("filtered above"),
}
}
unit.delete(table, key.to_string());
Ok(true)
}
pub fn tick(&self) -> Result<(), DatabaseError> {
self.renew_lock()
}
fn renew_lock(&self) -> Result<(), DatabaseError> {
match &self.lock {
Some(lock) => Ok(lock.tick()?),
None => Ok(()),
}
}
pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
self.restore_outcomes.get(name).copied()
}
pub fn catalog_unchecked(&self) -> &Catalog {
&self.catalog
}
pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
let drifted = self.drifted_tables();
if drifted.is_empty() {
return Ok(&self.catalog);
}
Err(DatabaseError::Schema(self.drift_error(&drifted)))
}
pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
if let Some(drift) = self.drift_of(name) {
let drifted = [DriftedTable {
name: name.to_string(),
drift,
}];
return Err(DatabaseError::Schema(self.drift_error(&drifted)));
}
let table = self.tables.get(name).ok_or_else(|| {
DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
})?;
Ok(describe_table_document(table))
}
fn drift_of(&self, name: &str) -> Option<TableDrift> {
let table = self.tables.get(name)?;
let Some(entry) = self.catalog.tables.get(name) else {
return Some(TableDrift::Absent);
};
if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
return Some(TableDrift::Schema {
revision: entry.revision,
});
}
let declared = expand_indexes(name, entry);
let served = table.declared_indexes();
let unserved = index_names_missing_from(&declared, served);
let stale = index_names_missing_from(served, &declared);
if unserved.is_empty() && stale.is_empty() {
return None;
}
Some(TableDrift::Indexes { unserved, stale })
}
fn drifted_tables(&self) -> Vec<DriftedTable> {
let mut names: Vec<&String> = self.tables.keys().collect();
names.sort();
names
.into_iter()
.filter_map(|name| {
self.drift_of(name).map(|drift| DriftedTable {
name: name.clone(),
drift,
})
})
.collect()
}
fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
let detail = drifted
.iter()
.map(|DriftedTable { name, drift }| {
let revision = self.tables.get(name).map(|table| table.schema().revision);
let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
format!(
"\"{name}\" is open here at {open} while the catalog {}",
stored_side_of(drift)
)
})
.collect::<Vec<_>>()
.join("; ");
HaysonSchemaError(format!(
"the catalog of \"{}\" moved on under this handle: {detail} — an open table keeps the schema and the index set it was opened with, so describing it from the catalog would describe a table this handle does not serve; reopen the database to work with the catalog on disk",
self.layout.root().display()
))
}
pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
relations_touching(&self.catalog, table)
}
pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
check_integrity(&self.catalog, &refs)
}
pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
let mut changed = false;
if let Some(catalog) = read_catalog_with(self.layout.metadata_path(), self.sealer.as_deref())? {
if catalog != self.catalog {
self.catalog = catalog;
changed = true;
}
}
let missing: Vec<String> = self
.catalog
.tables
.keys()
.filter(|name| !self.tables.contains_key(*name))
.cloned()
.collect();
for name in missing {
let Some(entry) = self.catalog.tables.get(&name) else {
continue;
};
let (table, outcome) = open_catalog_table(
&self.layout,
&name,
&self.catalog,
entry,
self.snapshot_every,
self.cipher.as_ref(),
self.follow,
self.readonly,
compaction_of(self.compaction, &self.catalog, Some(entry)),
self.cache,
self.max_image_bytes,
self.password_hash,
self.sealer.clone(),
)?;
self.tables.insert(name.clone(), table);
self.restore_outcomes.insert(name, outcome);
changed = true;
}
for (name, table) in self.tables.iter_mut() {
let outcome = table
.refresh()
.map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
changed |= outcome.changed();
}
sync_table_triggers(&self.catalog, &mut self.tables);
Ok(changed)
}
pub fn unit_of_work(&mut self) -> UnitOfWork {
UnitOfWork::new()
}
pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
let prepared = self.prepare_commit(unit)?;
unit.journal_sealed(&prepared, &self.layout.journal_path(), self.sealer.as_deref())?;
self.apply_commit(prepared)
}
pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
if self.readonly {
return Err(DatabaseError::ReadOnly {
dir: self.layout.root().to_path_buf(),
operation: "commit".to_string(),
});
}
self.renew_lock()?;
Ok(unit.prepare(&mut self.tables)?)
}
pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
let journal_path = self.layout.journal_path();
Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
}
fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
if self.readonly {
return Err(DatabaseError::ReadOnly {
dir: self.layout.root().to_path_buf(),
operation: operation.to_string(),
});
}
self.renew_lock()
}
fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
write_catalog_sealed(self.layout.metadata_path(), catalog, self.sealer.as_deref()).map_err(catalog_write_error)
}
pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
self.assert_writable("createTable")?;
let schema = parse_schema(&schema_spec)?;
if schema.table != name {
return Err(DatabaseError::Schema(HaysonSchemaError(format!(
"schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
schema.table
))));
}
let now = now_ms();
let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
self.write_catalog_to_disk(&next_catalog)?;
let next_entry = &next_catalog.tables[name];
let (table, outcome) = open_catalog_table(
&self.layout,
name,
&next_catalog,
next_entry,
self.snapshot_every,
self.cipher.as_ref(),
self.follow,
self.readonly,
compaction_of(self.compaction, &next_catalog, Some(next_entry)),
self.cache,
self.max_image_bytes,
self.password_hash,
self.sealer.clone(),
)?;
self.catalog = next_catalog;
self.tables.insert(name.to_string(), table);
self.restore_outcomes.insert(name.to_string(), outcome);
Ok(())
}
pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
self.assert_writable("dropField")?;
self.flush_table(table)?;
let existing = self.tables.get(table).ok_or_else(|| {
DatabaseError::Catalog(format!(
"table \"{table}\" does not exist in \"{}\"",
self.layout.root().display()
))
})?;
let now = now_ms();
let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
if !force {
assert_no_residual_values(table, existing, field)?;
}
self.write_catalog_to_disk(&next_catalog)?;
let (mut table_handle, _) = open_catalog_table(
&self.layout,
table,
&next_catalog,
&next_entry,
self.snapshot_every,
self.cipher.as_ref(),
self.follow,
self.readonly,
compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
self.cache,
self.max_image_bytes,
self.password_hash,
self.sealer.clone(),
)?;
table_handle
.snapshot()
.map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
self.catalog = next_catalog;
self.tables.insert(table.to_string(), table_handle);
Ok(())
}
pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
self.assert_writable("declareIndex")?;
self.flush_table(table)?;
let existing = self.tables.get(table).ok_or_else(|| {
DatabaseError::Catalog(format!(
"table \"{table}\" does not exist in \"{}\"",
self.layout.root().display()
))
})?;
let now = now_ms();
let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
self.write_catalog_to_disk(&next_catalog)?;
let next_entry = &next_catalog.tables[table];
let (mut table_handle, _) = open_catalog_table(
&self.layout,
table,
&next_catalog,
next_entry,
self.snapshot_every,
self.cipher.as_ref(),
self.follow,
self.readonly,
compaction_of(self.compaction, &next_catalog, Some(next_entry)),
self.cache,
self.max_image_bytes,
self.password_hash,
self.sealer.clone(),
)?;
table_handle
.snapshot()
.map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
self.catalog = next_catalog;
self.tables.insert(table.to_string(), table_handle);
Ok(())
}
pub fn declare_relation(&mut self, relation_spec: Value, force: bool) -> Result<(), DatabaseError> {
self.assert_writable("declareRelation")?;
let from_table = relation_spec
.get("from")
.and_then(|end| end.get("table"))
.and_then(Value::as_str)
.ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"from.table\"".to_string()))?
.to_string();
let to_table = relation_spec
.get("to")
.and_then(|end| end.get("table"))
.and_then(Value::as_str)
.ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"to.table\"".to_string()))?
.to_string();
self.flush_table(&from_table)?;
self.flush_table(&to_table)?;
let from_handle = self.tables.get(&from_table).ok_or_else(|| {
DatabaseError::Catalog(format!(
"table \"{from_table}\" does not exist in \"{}\"",
self.layout.root().display()
))
})?;
let to_handle = self.tables.get(&to_table).ok_or_else(|| {
DatabaseError::Catalog(format!(
"table \"{to_table}\" does not exist in \"{}\"",
self.layout.root().display()
))
})?;
let now = now_ms();
let next_catalog = declare_relation_in(&self.catalog, &relation_spec, from_handle, to_handle, now, force)?;
self.write_catalog_to_disk(&next_catalog)?;
self.catalog = next_catalog;
Ok(())
}
pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
self.assert_writable("declareTrigger")?;
let now = now_ms();
let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
self.write_catalog_to_disk(&next_catalog)?;
self.catalog = next_catalog;
sync_table_triggers(&self.catalog, &mut self.tables);
Ok(())
}
pub fn flush(&mut self) -> Result<(), DatabaseError> {
for (name, table) in &mut self.tables {
table
.flush()
.map_err(|error| DatabaseError::Table(name.clone(), error))?;
}
Ok(())
}
fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
let Some(handle) = self.tables.get_mut(table) else {
return Ok(());
};
handle
.flush()
.map_err(|error| DatabaseError::Table(table.to_string(), error))
}
pub fn close(mut self) -> io::Result<()> {
let flushed = if self.readonly {
Ok(())
} else {
self.flush_and_snapshot_all()
};
if let Some(mut lock) = self.lock.take() {
lock.release();
}
flushed
}
fn flush_and_snapshot_all(&mut self) -> io::Result<()> {
let mut first_refusal = None;
for table in self.tables.values_mut() {
let outcome = table.flush().map_err(io::Error::from).and_then(|()| {
if table.needs_snapshot() {
table.snapshot()
} else {
Ok(())
}
});
if let Err(error) = outcome {
first_refusal.get_or_insert(error);
}
}
match first_refusal {
Some(error) => Err(error),
None => Ok(()),
}
}
}
#[cfg(test)]
mod tests {
use serde_json::json;
use tempfile::tempdir;
use super::*;
use crate::catalog::create_table;
use crate::catalog::types::{CatalogIndexKind, IndexEntry};
use crate::checkpoint::write_snapshot;
use crate::registry::table_registry::TableRegistry;
use crate::schema::parse_schema;
use crate::storage::{serialize_entry, LogMeta, LogOp};
use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
use crate::txn::JOURNAL_FILE;
const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
fn flat(dir: &Path) -> Layout {
Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
}
fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
flat(dir).log_path(table)
}
#[test]
fn a_database_and_its_tables_can_be_shared_across_threads() {
fn shareable<T: Send + Sync>() {}
shareable::<Database>();
shareable::<Table>();
shareable::<crate::registry::fulltext_index::FulltextIndex>();
}
fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
flat(dir).index_path(table)
}
fn users_schema() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
},
}))
.unwrap()
}
fn orders_schema() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string", "required": true },
},
}))
.unwrap()
}
fn catalog_of(schemas: &[TableSchema]) -> Catalog {
let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
for schema in schemas {
catalog = create_table(&catalog, &schema.table, schema, 1_000);
}
catalog
}
fn write_catalog_to(dir: &Path, catalog: &Catalog) {
write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
}
#[test]
fn open_builds_a_database_with_every_declared_table_available() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
assert_eq!(db.table_names(), vec!["orders", "users"]);
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
db.close().unwrap();
}
#[test]
fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
let dir = tempdir().unwrap();
let catalog = catalog_of(&[users_schema()]);
write_catalog_to(dir.path(), &catalog);
let schema = table_schema_of("users", &catalog.tables["users"]);
let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
let log_path = log_path(dir.path(), "users");
let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
log.append(&[serialize_entry(
&LogMeta {
key: "u1".into(),
op: LogOp::Put,
seq: 1,
ts: 1,
},
Some(&{
let mut record = crate::schema::HaysonRecord::new();
record.insert("id".into(), json!("u1"));
record.insert("name".into(), json!("Ania"));
record
}),
)])
.unwrap();
let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
write_snapshot(&index_path(dir.path(), "users"), ®istry, "users", &hash, 0, &log).unwrap();
let db = Database::open(dir.path()).unwrap();
assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
db.close().unwrap();
}
#[test]
fn open_finishes_an_unfinished_journal_before_returning() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
{
let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
let user_entries = users
.prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
.unwrap();
let order_entries = orders
.prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
.unwrap();
let mut entries = Vec::new();
for entry in &user_entries {
entries.push(JournalEntry {
t: "users".into(),
s: entry.seq(),
l: entry.line().to_string(),
});
}
for entry in &order_entries {
entries.push(JournalEntry {
t: "orders".into(),
s: entry.seq(),
l: entry.line().to_string(),
});
}
write_journal(
&dir.path().join(JOURNAL_FILE),
&Journal {
header: JournalHeader {
journal_version: JOURNAL_VERSION,
kind: "unit".into(),
id: "crash".into(),
started_at: 0,
tables: vec!["users".into(), "orders".into()],
},
entries,
},
)
.unwrap();
}
assert!(dir.path().join(JOURNAL_FILE).exists());
let db = Database::open(dir.path()).unwrap();
assert!(!dir.path().join(JOURNAL_FILE).exists());
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
assert_eq!(
db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let _first = Database::open(dir.path()).unwrap();
let second = Database::open_with(
dir.path(),
OpenOptions {
readonly: false,
lock: LockMode::Enabled(LockOptions {
timeout_ms: Some(0),
..Default::default()
}),
..Default::default()
},
);
assert!(matches!(second, Err(DatabaseError::Lock(_))));
}
fn lock_path_of(dir: &Path) -> std::path::PathBuf {
dir.join(crate::lock::LOCK_FILE)
}
fn open_refusing(dir: &Path) -> DatabaseError {
Database::open_with(
dir,
OpenOptions {
lock: LockMode::Enabled(LockOptions {
timeout_ms: Some(0),
..Default::default()
}),
..Default::default()
},
)
.map(|_| ())
.expect_err("this directory must not open")
}
#[test]
fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
},
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
let log = fs::read(dir.path().join("users.jsonl")).unwrap();
fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
let refusal = open_refusing(dir.path()).to_string();
assert!(
refusal.contains("users.jsonl"),
"the refusal has to name the log it found: {refusal}"
);
assert!(
refusal.contains(MANIFEST_FILE),
"the refusal has to name the catalog: {refusal}"
);
assert!(
!dir.path().join(MANIFEST_FILE).exists(),
"a refused open must not leave a catalog"
);
assert_eq!(
fs::read(dir.path().join("users.jsonl")).unwrap(),
log,
"the log must be untouched"
);
}
#[test]
fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
let empty = tempdir().unwrap();
assert!(
Database::open(empty.path()).is_ok(),
"an empty directory is still a new database"
);
let littered = tempdir().unwrap();
fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
fs::create_dir(littered.path().join("users-backup")).unwrap();
assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
}
#[test]
fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
let dir = tempdir().unwrap();
fs::create_dir_all(dir.path().join("notes")).unwrap();
fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
let refusal = open_refusing(dir.path()).to_string();
assert!(
refusal.contains("note.jsonl"),
"a partitioned log still counts as data: {refusal}"
);
}
#[test]
fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
for (case, document) in [
("not JSON at all", "{ this is not json".to_string()),
(
"no formatVersion",
json!({ "catalogVersion": 2, "tables": {} }).to_string(),
),
(
"the format-1 marker",
json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
),
(
"a formatVersion from the future",
json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
),
] {
let dir = tempdir().unwrap();
std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
open_refusing(dir.path());
assert!(
!lock_path_of(dir.path()).exists(),
"{case}: the refused open left its lock file behind"
);
}
}
#[test]
fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
let dir = tempdir().unwrap();
std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
open_refusing(dir.path());
std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
}
#[test]
fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let error = Database::open_with(
dir.path(),
OpenOptions {
cypher: Some(CipherOptions {
keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
key_id: None,
on_missing_key: None,
}),
..OpenOptions::default()
},
)
.map(|_| ())
.unwrap_err();
assert!(matches!(error, DatabaseError::Cipher(_)));
assert!(!lock_path_of(dir.path()).exists());
Database::open(dir.path()).unwrap().close().unwrap();
}
#[test]
fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let held = Database::open(dir.path()).unwrap();
assert!(lock_path_of(dir.path()).exists());
assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
held.close().unwrap();
assert!(!lock_path_of(dir.path()).exists());
}
fn open_skipping_the_lock(dir: &Path) -> Database {
Database::open_with(
dir,
OpenOptions {
lock: LockMode::Skip,
..Default::default()
},
)
.unwrap()
}
fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
let catalog = read_catalog(&dir.join(MANIFEST_FILE))
.unwrap()
.expect("a catalog on disk");
let mut names: Vec<String> = catalog.tables[table]
.indexes
.iter()
.flatten()
.map(|index| index.name.clone())
.collect();
names.sort();
names
}
fn two_tables_with_records(dir: &Path) {
write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
db.close().unwrap();
}
#[test]
fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
let dir = tempdir().unwrap();
two_tables_with_records(dir.path());
let mut locked = Database::open(dir.path()).unwrap();
let mut unlocked = open_skipping_the_lock(dir.path());
locked
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u3", "name": "Kasia" }), false)
.unwrap();
unlocked
.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
let third = Database::open_with(
dir.path(),
OpenOptions {
lock: LockMode::Enabled(LockOptions {
timeout_ms: Some(0),
..Default::default()
}),
..Default::default()
},
);
assert!(matches!(third, Err(DatabaseError::Lock(_))));
unlocked.close().unwrap();
locked.close().unwrap();
}
#[test]
fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
let dir = tempdir().unwrap();
two_tables_with_records(dir.path());
let mut locked = Database::open(dir.path()).unwrap();
let mut unlocked = open_skipping_the_lock(dir.path());
locked
.declare_index(
"users",
json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
)
.unwrap();
unlocked
.declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
.unwrap();
unlocked.close().unwrap();
locked.close().unwrap();
assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
}
#[test]
fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
let dir = tempdir().unwrap();
two_tables_with_records(dir.path());
let mut locked = Database::open(dir.path()).unwrap();
let mut unlocked = open_skipping_the_lock(dir.path());
locked
.declare_index(
"users",
json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
)
.unwrap();
assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
unlocked
.declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
.unwrap();
unlocked.close().unwrap();
locked.close().unwrap();
assert_eq!(
declared_on_disk(dir.path(), "users"),
vec!["by_id".to_string(), "by_name".to_string()]
);
}
fn open_follower(dir: &Path) -> Database {
Database::open_with(
dir,
OpenOptions {
readonly: true,
lock: LockMode::default(),
..Default::default()
},
)
.unwrap()
}
#[test]
fn refresh_with_nothing_new_on_disk_reports_no_change() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut writer = Database::open(dir.path()).unwrap();
writer
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let mut follower = open_follower(dir.path());
follower.refresh().unwrap();
assert!(!follower.refresh().unwrap());
writer.close().unwrap();
}
#[test]
fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut writer = Database::open(dir.path()).unwrap();
let mut follower = open_follower(dir.path());
assert_eq!(follower.table_names(), vec!["users"]);
writer
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
writer
.create_table(
"orders",
json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
}),
)
.unwrap();
writer
.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
assert!(follower.refresh().unwrap());
assert_eq!(follower.table_names(), vec!["orders", "users"]);
assert_eq!(
follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
assert_eq!(
follower
.table("orders")
.unwrap()
.get("o1")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
writer.close().unwrap();
}
#[test]
fn refresh_survives_a_compaction_run_by_the_writer_process() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut writer = Database::open(dir.path()).unwrap();
let users = writer.table_mut("users").unwrap();
users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
let mut follower = open_follower(dir.path());
assert_eq!(follower.table("users").unwrap().len(), 2);
writer.table_mut("users").unwrap().compact().unwrap();
assert!(follower.refresh().unwrap());
assert_eq!(follower.table("users").unwrap().len(), 2);
assert_eq!(
follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Anna"))
);
assert_eq!(follower.table("users").unwrap().stats().gen, 1);
writer.close().unwrap();
}
#[test]
fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania" }));
unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
let result = db.commit(&mut unit).unwrap();
assert_eq!(result.operations, 2);
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
assert_eq!(
db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.create_table(
"orders",
json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string", "required": true },
},
}),
)
.unwrap();
assert_eq!(db.table_names(), vec!["orders", "users"]);
assert!(db.catalog.tables.contains_key("orders"));
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
assert_eq!(
db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
Some(&json!("u1"))
);
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert!(reread.tables.contains_key("orders"));
db.close().unwrap();
}
#[test]
fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
let error = db
.create_table(
"orders",
json!({
"schemaVersion": 1,
"table": "invoices",
"primaryKey": "id",
"fields": { "id": { "type": "string" } },
}),
)
.expect_err("a schema naming another table is refused");
assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
let message = error.to_string();
assert!(
message.contains("invoices"),
"the refusal names the schema's table: {message}"
);
assert!(
message.contains("orders"),
"the refusal names the key it was created under: {message}"
);
assert_eq!(db.table_names(), vec!["users"]);
assert!(!db.catalog.tables.contains_key("orders"));
assert!(!db.catalog.tables.contains_key("invoices"));
assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
db.close().unwrap();
}
#[test]
fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.create_table(
"orders",
json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": { "id": { "type": "string" } },
}),
)
.unwrap();
assert_eq!(db.table("orders").unwrap().schema().table, "orders");
db.close().unwrap();
}
#[test]
fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
Database::open(dir.path()).unwrap().close().unwrap();
let mut db = Database::open_with(
dir.path(),
OpenOptions {
readonly: true,
lock: LockMode::default(),
..Default::default()
},
)
.unwrap();
let error = db
.create_table(
"orders",
json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
)
.unwrap_err();
assert!(
matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
"expected a typed readonly refusal, got {error:?}"
);
assert_eq!(db.table_names(), vec!["users"]);
}
#[test]
fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
fs::create_dir(&temporary).unwrap();
let error = db
.create_table(
"orders",
json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
)
.unwrap_err();
let refusal = match &error {
DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
.expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
};
assert_eq!(refusal.path, temporary);
assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
assert_eq!(db.table_names(), vec!["users"]);
assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
}
#[test]
fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
assert!(matches!(
catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
DatabaseError::Catalog(_)
));
}
#[test]
fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users").unwrap().delete("u1", false).unwrap();
db.drop_field("users", "name", false).unwrap();
assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
assert!(!db.catalog.tables["users"].fields.contains_key("name"));
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert!(!reread.tables["users"].fields.contains_key("name"));
db.close().unwrap();
}
#[test]
fn drop_field_refuses_a_live_residual_field_without_force() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let error = db.drop_field("users", "name", false);
assert!(error.is_err());
assert!(db.catalog.tables["users"].fields.contains_key("name"));
assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
db.drop_field("users", "name", true).unwrap();
assert!(!db.catalog.tables["users"].fields.contains_key("name"));
db.close().unwrap();
}
fn users_catalog_indexing_name() -> Catalog {
let mut catalog = catalog_of(&[users_schema()]);
catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
name: "by_name".to_string(),
fields: vec!["name".to_string()],
kind: CatalogIndexKind::Exact,
unique: false,
where_: None,
expire_after_seconds: None,
}]);
catalog
}
#[test]
fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &users_catalog_indexing_name());
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let error = db.drop_field("users", "name", false).unwrap_err();
assert!(
matches!(error, DatabaseError::Schema(_)),
"expected a schema refusal, got {error:?}"
);
let message = error.to_string();
assert!(message.contains("by_name"), "{message}");
assert!(!message.contains("live record(s)"), "{message}");
assert!(db.catalog.tables["users"].fields.contains_key("name"));
db.close().unwrap();
}
#[test]
fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &users_catalog_indexing_name());
let mut db = Database::open(dir.path()).unwrap();
let error = db.drop_field("users", "name", false).unwrap_err();
assert!(
matches!(error, DatabaseError::Schema(_)),
"expected a schema refusal, got {error:?}"
);
assert!(error.to_string().contains("by_name"));
assert!(db.catalog.tables["users"].fields.contains_key("name"));
db.close().unwrap();
}
#[test]
fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let error = db.drop_field("users", "name", false).unwrap_err();
assert!(
matches!(
&error,
DatabaseError::FieldInUse { table, field, records }
if table == "users" && field == "name" && *records == 1
),
"expected a typed field-in-use refusal, got {error:?}"
);
assert!(db.catalog.tables["users"].fields.contains_key("name"));
db.close().unwrap();
}
#[test]
fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &users_catalog_indexing_name());
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let error = db.drop_field("users", "name", true).unwrap_err();
assert!(
matches!(error, DatabaseError::Schema(_)),
"expected a schema refusal, got {error:?}"
);
assert!(error.to_string().contains("by_name"));
db.close().unwrap();
}
#[test]
fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"email": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "email": "same@example.com" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "email": "same@example.com" }), false)
.unwrap();
let error = db.declare_index(
"users",
json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
);
assert!(error.is_err());
assert!(db.catalog.tables["users"].indexes.is_none());
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert!(reread.tables["users"].indexes.is_none());
db.close().unwrap();
}
fn orders_user_relation(enforce: bool) -> Value {
json!({
"name": "orders_user",
"from": { "table": "orders", "field": "userId" },
"to": { "table": "users", "field": "id" },
"cardinality": "many-to-one",
"enforce": enforce,
})
}
#[test]
fn declare_relation_adds_a_relation_to_the_catalog_and_persists_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(false), false).unwrap();
assert_eq!(db.catalog.relations.as_ref().unwrap()[0].name, "orders_user");
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert_eq!(reread.relations.unwrap()[0].name, "orders_user");
db.close().unwrap();
}
#[test]
fn declare_relation_with_enforce_true_refuses_over_a_live_orphan_and_force_overrides() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "ghost" }), false)
.unwrap();
let error = db.declare_relation(orders_user_relation(true), false).unwrap_err();
assert!(error.to_string().contains("orders_user"));
assert!(db.catalog.relations.is_none());
db.declare_relation(orders_user_relation(true), true).unwrap();
assert_eq!(db.catalog.relations.as_ref().unwrap()[0].enforce, Some(true));
db.close().unwrap();
}
#[test]
fn put_checked_refuses_a_value_with_no_counterpart_on_an_enforced_relation() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(true), false).unwrap();
let error = db
.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
.unwrap_err();
assert!(error.to_string().contains("orders_user"));
assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
db.close().unwrap();
}
#[test]
fn put_checked_allows_a_value_with_a_live_counterpart() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(true), false).unwrap();
db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
db.close().unwrap();
}
#[test]
fn put_checked_ignores_a_relation_that_does_not_declare_enforce() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(false), false).unwrap();
db.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
.unwrap();
assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
db.close().unwrap();
}
#[test]
fn patch_checked_refuses_changing_the_field_to_a_value_with_no_counterpart() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(true), false).unwrap();
db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
let error = db
.patch_checked("orders", "o1", &json!({ "userId": "ghost" }), false)
.unwrap_err();
assert!(error.to_string().contains("orders_user"));
assert_eq!(
db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn patch_checked_leaves_the_field_alone_when_changes_does_not_touch_it() {
let dir = tempdir().unwrap();
let orders_named = parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string", "required": true },
"note": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_named]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation(true), false).unwrap();
db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
db.patch_checked("orders", "o1", &json!({ "note": "gift-wrapped" }), false)
.unwrap();
assert_eq!(
db.table("orders").unwrap().get("o1").unwrap().unwrap().get("note"),
Some(&json!("gift-wrapped"))
);
db.close().unwrap();
}
#[test]
fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"email": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
.unwrap();
db.declare_index(
"users",
json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
)
.unwrap();
assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
let spec = QueryBuilder::new()
.r#where(
"email",
crate::query::predicates::ComparisonOp::Eq,
json!("piotr@example.com"),
)
.spec();
let found = db.table("users").unwrap().query(&spec).unwrap();
assert_eq!(found.keys, vec!["u2".to_string()]);
let conflicting = db
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
assert!(conflicting.is_err());
db.close().unwrap();
}
#[test]
fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"email": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
.unwrap();
db.declare_index(
"users",
json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
)
.unwrap();
assert!(
index_path(dir.path(), "users").exists(),
"declare_index owes the table a snapshot carrying the new index's projection"
);
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
.unwrap();
assert!(
db.table("users").unwrap().needs_snapshot(),
"the handle declare_index left behind must still be able to snapshot itself"
);
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
let spec = QueryBuilder::new()
.r#where(
"email",
crate::query::predicates::ComparisonOp::Eq,
json!("piotr@example.com"),
)
.spec();
assert_eq!(
reopened.table("users").unwrap().query(&spec).unwrap().keys,
vec!["u2".to_string()]
);
assert_eq!(
declared_on_disk(dir.path(), "users"),
vec!["users_email_exact".to_string()]
);
reopened.close().unwrap();
}
#[test]
fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
catalog.defaults = Some(crate::catalog::TableOptions {
sync: Some(crate::catalog::CatalogSyncMode::Never),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("users").unwrap().sync_mode(),
crate::catalog::CatalogSyncMode::Never
);
db.close().unwrap();
}
#[test]
fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
catalog.defaults = Some(crate::catalog::TableOptions {
sync: Some(crate::catalog::CatalogSyncMode::Never),
..Default::default()
});
catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
sync: Some(crate::catalog::CatalogSyncMode::Commit),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("users").unwrap().sync_mode(),
crate::catalog::CatalogSyncMode::Commit
);
assert_eq!(
db.table("orders").unwrap().sync_mode(),
crate::catalog::CatalogSyncMode::Never
);
db.close().unwrap();
}
#[test]
fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
write_catalog_to(dir.path(), &catalog);
let mut db = Database::open(dir.path()).unwrap();
let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
let table = db.table_mut("users").unwrap();
let record = table.put(&document, false).unwrap();
assert_eq!(record.get("nickname"), None);
assert_eq!(
table.dropped_field_warnings(&document),
vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
);
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
assert_eq!(stored.get("name"), Some(&json!("Ania")));
assert_eq!(stored.get("nickname"), None);
reopened.close().unwrap();
}
#[test]
fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("users").unwrap().validation(),
crate::catalog::CatalogValidation::Strict
);
let error = db
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
.unwrap_err();
assert!(error.to_string().contains("is not declared in the schema"), "{error}");
db.close().unwrap();
}
fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
let mut catalog = catalog_of(&[users_with_age_schema()]);
let entry = catalog.tables.get_mut("users").unwrap();
entry.validation = validation;
entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
write_catalog_to(dir, &catalog);
}
fn users_with_age_schema() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
"age": { "type": "int" },
},
}))
.unwrap()
}
#[test]
fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
let dir = tempdir().unwrap();
deprecating_age(dir.path(), None, "use \"bornAt\" instead");
let mut db = Database::open(dir.path()).unwrap();
let table = db.table_mut("users").unwrap();
let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
let record = table.put(&touching, false).unwrap();
assert_eq!(record.get("age"), Some(&json!(30)));
assert_eq!(
table.deprecated_field_warnings(&touching),
vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
);
assert!(table
.deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
.is_empty());
db.close().unwrap();
}
#[test]
fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
let dir = tempdir().unwrap();
deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
let mut db = Database::open(dir.path()).unwrap();
let table = db.table_mut("users").unwrap();
table.put(&document, false).unwrap();
assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
table.put(&document, false).unwrap();
assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(
reopened.table("users").unwrap().deprecated_field_warnings(&document),
expected
);
reopened.close().unwrap();
}
}
#[test]
fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
let db = Database::open(dir.path()).unwrap();
let table = db.table("users").unwrap();
assert!(table.deprecated_fields().is_empty());
assert!(table
.deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
.is_empty());
db.close().unwrap();
}
#[test]
fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("users").unwrap().sync_mode(),
crate::catalog::CatalogSyncMode::Commit
);
db.close().unwrap();
}
#[test]
fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
sync: Some(crate::catalog::CatalogSyncMode::Never),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
{
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
}
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("users").unwrap().sync_mode(),
crate::catalog::CatalogSyncMode::Never
);
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
db.close().unwrap();
}
#[test]
fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
let dir = tempdir().unwrap();
let pages_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "pages",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"page": { "type": "markdown" },
},
}))
.unwrap();
let mut catalog = catalog_of(&[pages_schema]);
catalog.defaults = Some(crate::catalog::TableOptions {
link_parsing: Some(crate::catalog::LinkParsing::Lazy),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("pages").unwrap().link_parsing(),
crate::catalog::LinkParsing::Lazy
);
db.close().unwrap();
}
#[test]
fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
let dir = tempdir().unwrap();
let pages_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "pages",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"page": { "type": "markdown" },
},
}))
.unwrap();
let mut catalog = catalog_of(&[pages_schema]);
catalog.defaults = Some(crate::catalog::TableOptions {
link_parsing: Some(crate::catalog::LinkParsing::Lazy),
..Default::default()
});
catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
link_parsing: Some(crate::catalog::LinkParsing::Eager),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("pages").unwrap().link_parsing(),
crate::catalog::LinkParsing::Eager
);
db.close().unwrap();
}
#[test]
fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
let dir = tempdir().unwrap();
let docs_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "docs",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"cover": { "type": "image" },
},
}))
.unwrap();
let mut catalog = catalog_of(&[docs_schema]);
catalog.defaults = Some(crate::catalog::TableOptions {
image_dir: Some("shared-images".to_string()),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let mut db = Database::open(dir.path()).unwrap();
let record = db
.table_mut("docs")
.unwrap()
.put(
&json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
false,
)
.unwrap();
let physical_path = record
.get("cover")
.unwrap()
.get("physicalPath")
.unwrap()
.as_str()
.unwrap();
assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
assert!(dir.path().join("shared-images").join(physical_path).exists());
db.close().unwrap();
}
#[test]
fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string" },
"status": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
.unwrap();
db.declare_index(
"orders",
json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
)
.unwrap();
let spec = QueryBuilder::new()
.r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
.r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
.spec();
let found = db.table("orders").unwrap().query(&spec).unwrap();
assert_eq!(found.keys, vec!["o1".to_string()]);
assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
assert_eq!(found.plan.candidates, 1);
db.close().unwrap();
}
#[test]
fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"role": { "type": "string" },
"status": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
.unwrap();
db.declare_index(
"users",
json!({
"name": "users_role_active",
"fields": ["role"],
"kind": "exact",
"where": [{ "field": "status", "op": "=", "value": "active" }],
}),
)
.unwrap();
let eq = crate::query::predicates::ComparisonOp::Eq;
let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
let found = db.table("users").unwrap().query(&omitted).unwrap();
assert_eq!(
found.keys,
vec!["u1".to_string(), "u2".to_string()],
"the index holds u1 alone, but the query never asked for active records only"
);
assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
let repeated = QueryBuilder::new()
.r#where("role", eq, json!("admin"))
.r#where("status", eq, json!("active"))
.spec();
let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
assert_eq!(narrowed.keys, vec!["u1".to_string()]);
assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
assert_eq!(narrowed.plan.candidates, 1);
db.close().unwrap();
}
#[test]
fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string" },
"status": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
.unwrap();
db.declare_index(
"orders",
json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
)
.unwrap();
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
let spec = QueryBuilder::new()
.r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
.r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
.spec();
let found = reopened.table("orders").unwrap().query(&spec).unwrap();
assert_eq!(found.keys, vec!["o2".to_string()]);
assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
reopened.close().unwrap();
}
fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
let entry = catalog.tables.get_mut(table).unwrap();
entry.indexes = Some(vec![index]);
write_catalog_to(dir, &catalog);
}
fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
IndexEntry {
name: name.to_string(),
fields: fields.iter().map(|field| field.to_string()).collect(),
kind: CatalogIndexKind::Sorted,
unique: false,
where_: None,
expire_after_seconds: None,
}
}
#[test]
fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string" },
"total": { "type": "int" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
let orders = db.table_mut("orders").unwrap();
orders
.put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
.unwrap();
orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
orders
.put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
.unwrap();
db.close().unwrap();
assert!(
index_path(dir.path(), "orders").exists(),
"the fixture needs a snapshot to restore from"
);
add_index_to_catalog_on_disk(
dir.path(),
"orders",
sorted_index_entry("orders_user_total", &["userId", "total"]),
);
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
let spec = QueryBuilder::new()
.r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
.spec();
let found = reopened.table("orders").unwrap().query(&spec).unwrap();
assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
let mut keys = found.keys.clone();
keys.sort();
assert_eq!(
keys,
vec!["a".to_string(), "b".to_string()],
"b has no total, but it still has a userId"
);
reopened.close().unwrap();
}
#[test]
fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "tickets",
"primaryKey": "id",
"partitionBy": "kind",
"fields": {
"id": { "type": "string" },
"kind": { "type": "string", "required": true },
"userId": { "type": "string", "required": true },
"orderId": { "type": "int" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = Database::open(dir.path()).unwrap();
let tickets = db.table_mut("tickets").unwrap();
tickets
.put(
&json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
false,
)
.unwrap();
tickets
.put(
&json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
false,
)
.unwrap();
tickets
.put(
&json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
false,
)
.unwrap();
db.close().unwrap();
add_index_to_catalog_on_disk(
dir.path(),
"tickets",
IndexEntry {
name: "tickets_user_order".to_string(),
fields: vec!["userId".to_string(), "orderId".to_string()],
kind: CatalogIndexKind::Exact,
unique: true,
where_: None,
expire_after_seconds: None,
},
);
let mut reopened = Database::open(dir.path()).unwrap();
let tickets = reopened.table_mut("tickets").unwrap();
let clash = tickets.put(
&json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
false,
);
match clash {
Err(TableError::Unique(violation)) => {
assert_eq!(violation.field, "userId, orderId");
assert_eq!(violation.holder, "t2");
}
other => panic!("a repeated composite key must be refused, got {other:?}"),
}
tickets
.put(
&json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
false,
)
.unwrap();
reopened.close().unwrap();
}
#[test]
fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
assert!(!index_path(dir.path(), "users").exists());
db.close().unwrap();
assert!(index_path(dir.path(), "users").exists());
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
assert_eq!(
reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
assert_eq!(
reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
Some(&json!("Piotr"))
);
reopened.close().unwrap();
}
#[test]
fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
assert_eq!(
db.table("users").unwrap().len(),
0,
"nothing is committed before the close"
);
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(
reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
reopened.close().unwrap();
}
#[cfg(unix)]
fn swap_file_in_place(path: &Path) {
let swapped = path.with_extension("swap");
std::fs::copy(path, &swapped).unwrap();
std::fs::rename(&swapped, path).unwrap();
}
fn log_replaced_of(error: &io::Error) -> Option<&crate::errors::HaysonLogReplacedError> {
error.get_ref()?.downcast_ref::<crate::errors::HaysonLogReplacedError>()
}
#[cfg(unix)]
#[test]
fn a_close_refused_by_a_swapped_log_still_releases_the_directory_lock() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.stage_put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
swap_file_in_place(&log_path(dir.path(), "users"));
let error = db.close().unwrap_err();
assert!(log_replaced_of(&error).is_some(), "{error}");
assert!(!lock_path_of(dir.path()).exists(), "the lock file outlived the handle");
let started = std::time::Instant::now();
let reopened = Database::open(dir.path()).unwrap();
assert!(
started.elapsed() < Duration::from_secs(1),
"the reopen waited {:?} on a lock nobody holds",
started.elapsed()
);
assert_eq!(
reopened.table("users").unwrap().get("u2").unwrap(),
None,
"the staged write never reached the log"
);
reopened.close().unwrap();
}
#[cfg(unix)]
fn swap_older_version_in(path: &Path) {
let current = std::fs::read_to_string(path).unwrap();
let older = format!("{}\n", current.lines().next().unwrap());
let swapped = path.with_extension("swap");
std::fs::write(&swapped, older).unwrap();
std::fs::rename(&swapped, path).unwrap();
}
#[cfg(unix)]
#[test]
fn a_close_after_a_swapped_log_leaves_no_snapshot_so_the_reopen_reads_the_new_file() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Anna" }), false)
.unwrap();
swap_older_version_in(&log_path(dir.path(), "users"));
let error = db.close().unwrap_err();
assert!(log_replaced_of(&error).is_some(), "{error}");
assert!(
!index_path(dir.path(), "users").exists(),
"a snapshot describing the orphaned inode was written"
);
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(
reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania")),
"the reopen reads what the checkout put at the path"
);
reopened.close().unwrap();
}
#[cfg(unix)]
#[test]
fn refresh_after_a_swapped_log_takes_the_new_file_and_writes_land_in_it_again() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Anna" }), false)
.unwrap();
let log = log_path(dir.path(), "users");
swap_older_version_in(&log);
assert!(db
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.is_err());
assert!(db.refresh().unwrap(), "the swap counts as a change");
let users = db.table_mut("users").unwrap();
assert_eq!(users.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
assert_eq!(std::fs::read_to_string(&log).unwrap().lines().count(), 2);
db.close().unwrap();
let reopened = Database::open(dir.path()).unwrap();
assert_eq!(
reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
Some(&json!("Piotr"))
);
reopened.close().unwrap();
}
#[test]
fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open_with(
dir.path(),
OpenOptions {
snapshot_every: 2,
..Default::default()
},
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
assert!(!index_path(dir.path(), "users").exists());
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
assert!(index_path(dir.path(), "users").exists());
db.close().unwrap();
}
#[test]
fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open_with(
dir.path(),
OpenOptions {
snapshot_every: 2,
..Default::default()
},
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u3", "name": "Kasia" }), false)
.unwrap();
assert_eq!(
db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
Some(&json!("Kasia"))
);
db.close().unwrap();
}
fn users_insert_audits_into_orders() -> Value {
json!({
"name": "users_insert_audit",
"table": "users",
"event": "insert",
"effect": {
"table": "orders",
"fields": [
{ "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
{ "target": "userId", "source": { "kind": "new", "field": "id" } },
],
},
})
}
#[test]
fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
let dir = tempdir().unwrap();
let docs_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "docs",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"cover": { "type": "image" },
},
}))
.unwrap();
let mut catalog = catalog_of(&[docs_schema]);
catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
image_dir: Some("shared-images".to_string()),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let mut db = Database::open(dir.path()).unwrap();
let record = db
.table_mut("docs")
.unwrap()
.put(
&json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
false,
)
.unwrap();
let physical_path = record
.get("cover")
.unwrap()
.get("physicalPath")
.unwrap()
.as_str()
.unwrap();
assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
assert!(dir.path().join("shared-images").join(physical_path).exists());
db.close().unwrap();
}
#[test]
fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
let dir = tempdir().unwrap();
let pages_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "pages",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"page": { "type": "markdown" },
},
}))
.unwrap();
let mut catalog = catalog_of(&[pages_schema]);
catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
link_parsing: Some(crate::catalog::LinkParsing::Lazy),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let mut db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("pages").unwrap().link_parsing(),
crate::catalog::LinkParsing::Lazy
);
let outcome = db
.table_mut("pages")
.unwrap()
.put_with_warnings(
&json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
false,
)
.unwrap();
assert!(outcome.warnings.is_empty());
db.close().unwrap();
}
#[test]
fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
let dir = tempdir().unwrap();
let pages_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "pages",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"page": { "type": "markdown" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
let db = Database::open(dir.path()).unwrap();
assert_eq!(
db.table("pages").unwrap().link_parsing(),
crate::catalog::LinkParsing::Eager
);
}
#[test]
fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
assert_eq!(db.table("users").unwrap().triggers().len(), 1);
assert!(db.table("orders").unwrap().triggers().is_empty());
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
db.close().unwrap();
}
#[test]
fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania" }));
db.commit(&mut unit).unwrap();
assert_eq!(
db.table("orders")
.unwrap()
.get("o-audit")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
db.close().unwrap();
let mut reopened = Database::open(dir.path()).unwrap();
let mut unit = reopened.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania" }));
reopened.commit(&mut unit).unwrap();
assert_eq!(
reopened
.table("orders")
.unwrap()
.get("o-audit")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
reopened.close().unwrap();
}
#[test]
fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
db.declare_index(
"users",
json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
)
.unwrap();
assert_eq!(db.table("users").unwrap().triggers().len(), 1);
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania" }));
db.commit(&mut unit).unwrap();
assert_eq!(
db.table("orders")
.unwrap()
.get("o-audit")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
db.drop_field("users", "name", false).unwrap();
assert_eq!(db.table("users").unwrap().triggers().len(), 1);
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1" }));
db.commit(&mut unit).unwrap();
assert_eq!(
db.table("orders")
.unwrap()
.get("o-audit")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_trigger(users_insert_audits_into_orders()).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"revision": 2,
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
"city": { "type": "string" },
},
}),
)
.unwrap();
assert_eq!(db.table("users").unwrap().triggers().len(), 1);
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
db.commit(&mut unit).unwrap();
assert_eq!(
db.table("orders")
.unwrap()
.get("o-audit")
.unwrap()
.unwrap()
.get("userId"),
Some(&json!("u1"))
);
db.close().unwrap();
}
#[test]
fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
let spec = json!({
"name": "users_insert_audit",
"table": "users",
"event": "insert",
"effect": {
"table": "ghost",
"fields": [
{ "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
],
},
});
let error = db.declare_trigger(spec);
assert!(error.is_err());
assert!(db.catalog.triggers.is_none());
let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
assert!(reread.triggers.is_none());
db.close().unwrap();
}
mod automatic_compaction {
use super::*;
use crate::catalog::TableOptions;
fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
CatalogCompaction::Enabled {
dead_bytes_ratio,
min_dead_bytes,
}
}
fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
Database::open_with(
dir,
OpenOptions {
compaction,
..Default::default()
},
)
.unwrap()
}
fn lines_of(path: &Path) -> Vec<String> {
fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
}
fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
let users = db.table_mut(table).unwrap();
users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
for round in 0..overwrites {
users
.patch(key, &json!({ "name": format!("Ania {round}") }), false)
.unwrap();
}
}
#[test]
fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
churn(&mut db, "users", "u1", 12);
let stats = db.table("users").unwrap().stats();
assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
db.close().unwrap();
}
#[test]
fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
churn(&mut db, "users", "u1", 4);
let stats = db.table("users").unwrap().stats();
assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
assert!(stats.dead_bytes > 1);
assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
db.close().unwrap();
}
#[test]
fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
churn(&mut db, "users", "u1", 12);
let stats = db.table("users").unwrap().stats();
assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
assert_eq!(stats.dead_bytes, 0);
assert_eq!(stats.records, 1);
assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania 11"))
);
db.close().unwrap();
}
#[test]
fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
churn(&mut db, "users", "u1", 12);
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u2", "name": "Piotr" }), false)
.unwrap();
let before = db.table("users").unwrap().stats();
db.close().unwrap();
let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
let after = reopened.table("users").unwrap().stats();
assert_eq!(
after.gen, before.gen,
"the generation is carried by the snapshot, not reset"
);
assert_eq!(after.records, 2);
assert_eq!(after.dead_bytes, 0);
let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
keys.sort();
assert_eq!(keys, vec!["u1", "u2"]);
assert_eq!(
reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania 11"))
);
reopened.close().unwrap();
}
#[test]
fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
let dir = tempdir().unwrap();
let schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "tickets",
"primaryKey": "id",
"partitionBy": "kind",
"fields": {
"id": { "type": "string" },
"kind": { "type": "string", "required": true },
"note": { "type": "string" },
},
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[schema]));
let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
let tickets = db.table_mut("tickets").unwrap();
tickets
.put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
.unwrap();
let refund = dir.path().join("tickets").join("refund.jsonl");
let quiet_size = fs::metadata(&refund).unwrap().len();
tickets
.put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
.unwrap();
for round in 0..12 {
tickets
.patch("s1", &json!({ "note": format!("n{round}") }), false)
.unwrap();
}
let sale = dir.path().join("tickets").join("sale.jsonl");
assert!(
lines_of(&sale).len() < 13,
"the partition over its thresholds was rewritten, so not all 13 lines are still there"
);
assert_eq!(
fs::metadata(&refund).unwrap().len(),
quiet_size,
"the partition holding no garbage was left byte for byte"
);
assert_eq!(lines_of(&refund).len(), 1);
let stats = db.table("tickets").unwrap().stats();
assert_eq!(
stats.gen, 1,
"one pass moves the generation once, not once per partition"
);
assert_eq!(stats.records, 2);
db.close().unwrap();
}
#[test]
fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
catalog.defaults = Some(TableOptions {
compaction: Some(CatalogCompaction::Disabled),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let mut db = opened(dir.path(), None);
churn(&mut db, "users", "u1", 12);
assert_eq!(db.table("users").unwrap().stats().gen, 0);
assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
db.close().unwrap();
}
#[test]
fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
catalog.defaults = Some(TableOptions {
compaction: Some(CatalogCompaction::Disabled),
..Default::default()
});
catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
compaction: Some(enabled(0.5, 200)),
..Default::default()
});
write_catalog_to(dir.path(), &catalog);
let db = opened(dir.path(), None);
assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
db.close().unwrap();
let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
db.close().unwrap();
}
#[test]
fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
let dir = tempdir().unwrap();
let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
}),
)
.unwrap();
churn(&mut db, "users", "u1", 12);
assert!(db.table("users").unwrap().stats().gen > 0);
assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
db.close().unwrap();
}
#[test]
fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let db = opened(dir.path(), None);
assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
db.close().unwrap();
}
}
mod read_cache {
use super::*;
fn schema() -> Value {
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
})
}
fn opened(dir: &Path, cache: CachePolicy) -> Database {
Database::open_with(
dir,
OpenOptions {
cache,
..Default::default()
},
)
.unwrap()
}
#[test]
fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table("users", schema()).unwrap();
assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
db.close().unwrap();
let db = Database::open(dir.path()).unwrap();
assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
db.close().unwrap();
}
#[test]
fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
let dir = tempdir().unwrap();
let mut db = opened(dir.path(), CachePolicy::with_entries(64));
db.create_table("users", schema()).unwrap();
assert_eq!(
db.table("users").unwrap().cache_policy(),
CachePolicy::with_entries(64),
"a table created under the handle is armed on the same terms"
);
db.close().unwrap();
let db = opened(dir.path(), CachePolicy::with_entries(64));
assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
db.close().unwrap();
}
#[test]
fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
let dir = tempdir().unwrap();
let mut db = opened(dir.path(), CachePolicy::with_entries(8));
db.create_table("users", schema()).unwrap();
let users = db.table_mut("users").unwrap();
users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
let first = users.get("u1").unwrap();
let second = users.get("u1").unwrap();
assert_eq!(first, second);
assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
db.close().unwrap();
}
}
mod cypher {
use super::*;
use crate::crypto::{CipherOptions, MissingKeyPolicy};
use std::collections::HashMap;
fn key(byte: u8) -> Vec<u8> {
vec![byte; 32]
}
fn secrets_schema_json() -> Value {
json!({
"schemaVersion": 1,
"table": "secrets",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"token": { "type": "string", "cypher": true },
},
})
}
fn secrets_schema_json_with_note() -> Value {
json!({
"schemaVersion": 1,
"table": "secrets",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"token": { "type": "string", "cypher": true },
"note": { "type": "string" },
},
})
}
fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
Database::open_with(
dir,
OpenOptions {
cypher: Some(CipherOptions {
keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
key_id: key_id.map(str::to_string),
on_missing_key: None,
}),
..OpenOptions::default()
},
)
.unwrap()
}
#[test]
fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
let dir = tempdir().unwrap();
let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
db.create_table("secrets", secrets_schema_json()).unwrap();
db.table_mut("secrets")
.unwrap()
.put(&json!({ "id": "s1", "token": "hunter2" }), false)
.unwrap();
db.close().unwrap();
let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
assert_eq!(
reopened
.table("secrets")
.unwrap()
.get("s1")
.unwrap()
.unwrap()
.get("token"),
Some(&json!("hunter2"))
);
reopened.close().unwrap();
}
#[test]
fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
let dir = tempdir().unwrap();
let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
db.create_table("secrets", secrets_schema_json()).unwrap();
db.table_mut("secrets")
.unwrap()
.put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
.unwrap();
db.close().unwrap();
let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
assert_eq!(
rotated
.table("secrets")
.unwrap()
.get("s1")
.unwrap()
.unwrap()
.get("token"),
Some(&json!("written-with-k1"))
);
rotated
.table_mut("secrets")
.unwrap()
.put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
.unwrap();
rotated.close().unwrap();
let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
assert!(log.contains("$cypher$v1$k1$"));
assert!(log.contains("$cypher$v1$k2$"));
}
#[test]
fn keeps_the_keyring_across_drop_field_reopening_the_table() {
let dir = tempdir().unwrap();
let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
db.table_mut("secrets")
.unwrap()
.put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
.unwrap();
db.drop_field("secrets", "note", true).unwrap();
assert_eq!(
db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
Some(&json!("hunter2"))
);
db.close().unwrap();
}
#[test]
fn open_refuses_a_key_that_is_not_32_bytes() {
let dir = tempdir().unwrap();
let error = Database::open_with(
dir.path(),
OpenOptions {
cypher: Some(CipherOptions {
keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
key_id: None,
on_missing_key: None,
}),
..OpenOptions::default()
},
)
.map(|_| ())
.unwrap_err();
assert!(matches!(error, DatabaseError::Cipher(_)));
}
#[test]
fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
let dir = tempdir().unwrap();
let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
db.create_table("secrets", secrets_schema_json()).unwrap();
db.table_mut("secrets")
.unwrap()
.put(&json!({ "id": "s1", "token": "hunter2" }), false)
.unwrap();
db.close().unwrap();
let raw = Database::open_with(
dir.path(),
OpenOptions {
cypher: Some(CipherOptions {
keys: HashMap::from([("other".to_string(), key(2))]),
key_id: None,
on_missing_key: Some(MissingKeyPolicy::Raw),
}),
..OpenOptions::default()
},
)
.unwrap();
let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
raw.close().unwrap();
}
}
mod partitions {
use super::*;
fn notes_schema_json(partitioned: bool) -> Value {
let mut spec = json!({
"schemaVersion": 1,
"table": "notes",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"kind": { "type": "string", "required": true },
"title": { "type": "string", "required": true },
"body": { "type": "string" },
},
});
if partitioned {
spec.as_object_mut()
.unwrap()
.insert("partitionBy".into(), json!("kind"));
}
spec
}
fn lines_in(dir: &Path, part: &str) -> usize {
fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
.unwrap()
.trim()
.lines()
.filter(|line| !line.is_empty())
.count()
}
#[test]
fn a_partitioned_table_survives_the_catalog_round_trip() {
let dir = tempdir().unwrap();
{
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_schema_json(true)).unwrap();
let notes = db.table_mut("notes").unwrap();
notes
.put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
.unwrap();
notes
.put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
.unwrap();
db.close().unwrap();
}
let db = Database::open(dir.path()).unwrap();
let notes = db.table("notes").unwrap();
assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
assert_eq!(lines_in(dir.path(), "note"), 1);
assert_eq!(lines_in(dir.path(), "adr"), 1);
assert!(!dir.path().join("notes.jsonl").exists());
db.close().unwrap();
}
#[test]
fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
let dir = tempdir().unwrap();
{
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_schema_json(false)).unwrap();
let notes = db.table_mut("notes").unwrap();
notes
.put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
.unwrap();
notes
.put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
.unwrap();
notes.delete("n1", false).unwrap();
db.close().unwrap();
}
let mut db = Database::open(dir.path()).unwrap();
let mut partitioned = notes_schema_json(true);
partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
db.create_table("notes", partitioned).unwrap();
let notes = db.table("notes").unwrap();
assert_eq!(notes.get("n1").unwrap(), None);
assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
assert!(!dir.path().join("notes.jsonl").exists());
assert_eq!(lines_in(dir.path(), "adr"), 1);
db.close().unwrap();
}
#[test]
fn dropping_the_field_the_table_partitions_by_is_refused() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_schema_json(true)).unwrap();
db.table_mut("notes")
.unwrap()
.put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
.unwrap();
let error = db.drop_field("notes", "kind", true).unwrap_err();
assert!(error.to_string().contains("partitioned by"));
assert_eq!(
db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
Some(&json!("note"))
);
db.close().unwrap();
}
#[test]
fn dropping_an_unrelated_field_keeps_every_partition_readable() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_schema_json(true)).unwrap();
{
let notes = db.table_mut("notes").unwrap();
notes
.put(
&json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
false,
)
.unwrap();
notes
.put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
.unwrap();
}
db.drop_field("notes", "body", true).unwrap();
let notes = db.table("notes").unwrap();
assert!(!notes.schema().fields.contains_key("body"));
assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
assert_eq!(lines_in(dir.path(), "note"), 1);
assert_eq!(lines_in(dir.path(), "adr"), 1);
db.close().unwrap();
}
#[test]
fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_schema_json(true)).unwrap();
let mut unit = db.unit_of_work();
unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
let result = db.commit(&mut unit).unwrap();
assert_eq!(result.operations, 2);
let notes = db.table("notes").unwrap();
assert_eq!(
notes.get("n1").unwrap().unwrap().get("title"),
Some(&json!("from unit"))
);
assert_eq!(
notes.get("a1").unwrap().unwrap().get("title"),
Some(&json!("from unit too"))
);
assert_eq!(lines_in(dir.path(), "note"), 1);
assert_eq!(lines_in(dir.path(), "adr"), 1);
db.close().unwrap();
}
}
mod following_the_log {
use super::*;
fn readonly(follow: FollowOption) -> OpenOptions {
OpenOptions {
readonly: true,
follow,
..Default::default()
}
}
fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
let mut writer = Database::open(dir).unwrap();
writer.create_table("users", users_schema_spec()).unwrap();
writer
.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let reader = Database::open_with(dir, readonly(follow)).unwrap();
(writer, reader)
}
fn users_schema_spec() -> Value {
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
},
})
}
fn put(writer: &mut Database, key: &str, name: &str) {
writer
.table_mut("users")
.unwrap()
.put(&json!({ "id": key, "name": name }), false)
.unwrap();
}
#[test]
fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
let dir = tempdir().unwrap();
let mut writer = Database::open(dir.path()).unwrap();
writer.create_table("users", users_schema_spec()).unwrap();
assert_eq!(writer.follow(), None);
assert_eq!(writer.table("users").unwrap().follow(), None);
let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
assert_eq!(
reader.follow(),
Some(FollowPolicy {
interval: DEFAULT_FOLLOW_INTERVAL
})
);
assert_eq!(
reader.table("users").unwrap().follow(),
Some(FollowPolicy {
interval: DEFAULT_FOLLOW_INTERVAL
})
);
}
#[test]
fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
for index in 2..=5 {
put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
let found = reader
.table_mut("users")
.unwrap()
.get_following(&format!("u{index}"))
.unwrap();
assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
}
}
#[test]
fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
let spec = QueryBuilder::new().spec();
assert_eq!(
reader
.table_mut("users")
.unwrap()
.query_following(&spec)
.unwrap()
.matched,
1
);
put(&mut writer, "u2", "Osoba 2");
assert_eq!(
reader
.table_mut("users")
.unwrap()
.query_following(&spec)
.unwrap()
.matched,
2
);
}
#[test]
fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
put(&mut writer, "u2", "Osoba 2");
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u2")
.unwrap()
.is_none());
assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u2")
.unwrap()
.is_some());
}
#[test]
fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) =
writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u1")
.unwrap()
.is_some());
put(&mut writer, "u2", "Osoba 2");
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u2")
.unwrap()
.is_none());
std::thread::sleep(Duration::from_millis(200));
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u2")
.unwrap()
.is_some());
}
#[test]
fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
put(&mut writer, "u2", "Osoba 2");
let keys = vec!["u1".to_string(), "u2".to_string()];
let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
assert!(rows.iter().all(Option::is_some), "{rows:?}");
put(&mut writer, "u3", "Osoba 3");
let spec = QueryBuilder::new().spec();
let aggregate = crate::query::aggregate::AggregateSpec {
count: true,
..Default::default()
};
let counted = reader
.table_mut("users")
.unwrap()
.aggregate_following(&spec, &aggregate)
.unwrap();
assert_eq!(counted.count, 3);
}
#[test]
fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
let dir = tempdir().unwrap();
let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
for index in 2..=6 {
put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
}
writer.table_mut("users").unwrap().delete("u5", false).unwrap();
assert!(reader
.table_mut("users")
.unwrap()
.get_following("u6")
.unwrap()
.is_some());
writer.table_mut("users").unwrap().compact().unwrap();
let users = reader.table_mut("users").unwrap();
assert_eq!(
users.get_following("u3").unwrap().unwrap().get("name"),
Some(&json!("Osoba 3"))
);
assert!(users.get_following("u5").unwrap().is_none());
assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
}
}
mod where_the_files_land {
use super::*;
use crate::database::LayoutKind;
fn notes_schema_spec() -> Value {
json!({
"schemaVersion": 1,
"table": "notes",
"primaryKey": "id",
"partitionBy": "kind",
"fields": {
"id": { "type": "string" },
"kind": { "type": "string", "required": true, "index": "exact" },
"title": { "type": "string", "required": true },
},
})
}
fn nested(metadata_file: Option<&str>) -> OpenOptions {
OpenOptions {
layout: LayoutOptions {
kind: Some(LayoutKind::Nested),
metadata_file: metadata_file.map(str::to_string),
..Default::default()
},
..Default::default()
}
}
#[test]
fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
assert!(dir.path().join("hayson.json").is_file());
assert!(dir.path().join("users.jsonl").is_file());
assert!(!dir.path().join("tables").exists());
assert!(!dir.path().join("runtime").exists());
let db = Database::open(dir.path()).unwrap();
assert_eq!(db.layout().kind(), LayoutKind::Flat);
assert_eq!(db.layout().metadata_file(), "hayson.json");
assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
}
#[test]
fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
let dir = tempdir().unwrap();
let options = OpenOptions {
layout: LayoutOptions {
metadata_file: Some("cerebro.json".into()),
..Default::default()
},
..Default::default()
};
let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
assert!(dir.path().join("cerebro.json").is_file());
assert!(!dir.path().join("hayson.json").exists());
let db = Database::open_with(dir.path(), options).unwrap();
assert_eq!(db.layout().metadata_file(), "cerebro.json");
assert_eq!(db.table_names(), vec!["users"]);
assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
}
#[test]
fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
let dir = tempdir().unwrap();
let legacy = OpenOptions {
layout: LayoutOptions {
manifest: Some("cerebro.json".into()),
..Default::default()
},
..Default::default()
};
let mut db = Database::open_with(dir.path(), legacy).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" } },
}),
)
.unwrap();
db.close().unwrap();
let current = OpenOptions {
layout: LayoutOptions {
metadata_file: Some("cerebro.json".into()),
..Default::default()
},
..Default::default()
};
let db = Database::open_with(dir.path(), current).unwrap();
assert_eq!(db.table_names(), vec!["users"]);
assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
}
#[test]
fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
let dir = tempdir().unwrap();
let options = OpenOptions {
layout: LayoutOptions {
metadata_file: Some("cerebro.json".into()),
manifest: Some("hayson.json".into()),
..Default::default()
},
..Default::default()
};
let Err(error) = Database::open_with(dir.path(), options) else {
panic!("two disagreeing names for the metadata document were accepted")
};
assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
let message = error.to_string();
assert!(
message.contains("metadataFile") && message.contains("manifest"),
"{message}"
);
assert!(!dir.path().join("hayson.json").exists());
assert!(!dir.path().join("cerebro.json").exists());
assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
}
#[test]
fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
let dir = tempdir().unwrap();
let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
db.close().unwrap();
assert!(dir.path().join("hayson.json").is_file());
assert!(dir.path().join("tables").join("users.jsonl").is_file());
assert!(dir.path().join("index").join("users.idx").is_file());
assert!(!dir.path().join("users.jsonl").exists());
}
#[test]
fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
let dir = tempdir().unwrap();
let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
db.create_table("notes", notes_schema_spec()).unwrap();
db.table_mut("notes")
.unwrap()
.put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
.unwrap();
db.close().unwrap();
assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
}
#[test]
fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
let dir = tempdir().unwrap();
let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
let named_only = OpenOptions {
layout: LayoutOptions {
metadata_file: Some("cerebro.json".into()),
..Default::default()
},
..Default::default()
};
let db = Database::open_with(dir.path(), named_only).unwrap();
assert_eq!(db.layout().kind(), LayoutKind::Nested);
assert_eq!(
db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
Some(&json!("Ania"))
);
db.close().unwrap();
let as_flat = OpenOptions {
layout: LayoutOptions {
kind: Some(LayoutKind::Flat),
metadata_file: Some("cerebro.json".into()),
..Default::default()
},
..Default::default()
};
let Err(error) = Database::open_with(dir.path(), as_flat) else {
panic!("a nested database was opened as flat")
};
assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
assert!(error.to_string().contains("nested"), "{error}");
}
#[test]
fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
let dir = tempdir().unwrap();
let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
let mut unit = db.unit_of_work();
unit.put("users", json!({ "id": "u1", "name": "Ania" }));
unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
db.commit(&mut unit).unwrap();
db.close().unwrap();
assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
let db = Database::open_with(dir.path(), nested(None)).unwrap();
assert_eq!(db.table("users").unwrap().len(), 2);
}
}
mod a_declared_table_whose_log_is_not_on_disk {
use super::*;
fn notes_spec() -> Value {
json!({
"schemaVersion": 1,
"table": "notes",
"primaryKey": "id",
"partitionBy": "kind",
"fields": {
"id": { "type": "string" },
"kind": { "type": "string", "required": true },
"title": { "type": "string", "required": true },
},
})
}
fn readonly() -> OpenOptions {
OpenOptions {
readonly: true,
..Default::default()
}
}
fn users_with_a_record(dir: &Path) {
let mut db = Database::open(dir).unwrap();
db.create_table(
"users",
json!({
"schemaVersion": 1,
"table": "users",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "name": { "type": "string" } },
}),
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
}
fn notes_with_a_record(dir: &Path) {
let mut db = Database::open(dir).unwrap();
db.create_table("notes", notes_spec()).unwrap();
db.table_mut("notes")
.unwrap()
.put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
.unwrap();
db.close().unwrap();
}
fn refusal(dir: &Path) -> DatabaseError {
match Database::open_with(dir, readonly()) {
Err(error) => error,
Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
}
}
fn schema_refusal(error: DatabaseError) -> String {
match error {
DatabaseError::Schema(HaysonSchemaError(message)) => message,
other => panic!("expected a schema refusal, got {other:?}"),
}
}
#[test]
fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
let dir = tempdir().unwrap();
users_with_a_record(dir.path());
let log = log_path(dir.path(), "users");
fs::remove_file(&log).unwrap();
let message = schema_refusal(refusal(dir.path()));
assert!(message.contains("\"users\""), "{message}");
assert!(message.contains(&log.display().to_string()), "{message}");
assert!(message.contains("readonly"), "{message}");
assert!(!message.contains("createTable"), "{message}");
assert!(!log.exists(), "the refused open must not have created the log");
}
#[test]
fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
let dir = tempdir().unwrap();
users_with_a_record(dir.path());
fs::remove_file(log_path(dir.path(), "users")).unwrap();
let mut db = Database::open(dir.path()).unwrap();
assert_eq!(db.table_names(), vec!["users"]);
assert!(log_path(dir.path(), "users").is_file());
assert_eq!(db.table("users").unwrap().len(), 0);
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
assert_eq!(db.table("users").unwrap().len(), 1);
db.close().unwrap();
}
#[test]
fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
let dir = tempdir().unwrap();
users_with_a_record(dir.path());
fs::write(log_path(dir.path(), "users"), "").unwrap();
let db = Database::open_with(dir.path(), readonly()).unwrap();
assert_eq!(db.table_names(), vec!["users"]);
assert_eq!(db.table("users").unwrap().len(), 0);
}
#[test]
fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
let dir = tempdir().unwrap();
notes_with_a_record(dir.path());
let partition = dir.path().join("notes").join("note.jsonl");
assert!(
index_path(dir.path(), "notes").is_file(),
"the refusal reads the snapshot"
);
fs::remove_dir_all(dir.path().join("notes")).unwrap();
let message = schema_refusal(refusal(dir.path()));
assert!(message.contains("\"notes\""), "{message}");
assert!(message.contains(&partition.display().to_string()), "{message}");
assert!(message.contains("readonly"), "{message}");
assert!(!message.contains("createTable"), "{message}");
}
#[test]
fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
let dir = tempdir().unwrap();
let mut db = Database::open(dir.path()).unwrap();
db.create_table("notes", notes_spec()).unwrap();
db.close().unwrap();
let db = Database::open_with(dir.path(), readonly()).unwrap();
assert_eq!(db.table_names(), vec!["notes"]);
assert_eq!(db.table("notes").unwrap().len(), 0);
}
#[cfg(unix)]
#[test]
fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
let dir = tempdir().unwrap();
notes_with_a_record(dir.path());
let partition = dir.path().join("notes").join("note.jsonl");
fs::remove_file(&partition).unwrap();
std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
let error = refusal(dir.path());
match error {
DatabaseError::Log(path, inner) => {
assert_eq!(path, partition);
assert_eq!(inner.kind(), io::ErrorKind::NotFound);
}
other => panic!("expected a log refusal naming the partition, got {other:?}"),
}
assert!(!dir.path().join("notes").join("gone.jsonl").exists());
}
}
mod cascading_delete {
use super::*;
fn orders_schema_optional_user() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "orders",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"userId": { "type": "string" },
},
}))
.unwrap()
}
fn order_items_schema() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "order_items",
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"orderId": { "type": "string", "required": true },
},
}))
.unwrap()
}
fn orders_user_relation_on_delete(action: &str) -> Value {
json!({
"name": "orders_user",
"from": { "table": "orders", "field": "userId" },
"to": { "table": "users", "field": "id" },
"cardinality": "many-to-one",
"onDelete": action,
})
}
fn order_items_order_relation_cascade() -> Value {
json!({
"name": "order_items_order",
"from": { "table": "order_items", "field": "orderId" },
"to": { "table": "orders", "field": "id" },
"cardinality": "many-to-one",
"onDelete": "cascade",
})
}
#[test]
fn a_table_no_relation_names_as_to_goes_straight_through_plain_delete() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
db.close().unwrap();
}
#[test]
fn a_missing_key_answers_false_without_staging_anything() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("cascade"), false)
.unwrap();
let deleted = db.delete_checked("users", "ghost", false).unwrap();
assert!(!deleted);
db.close().unwrap();
}
#[test]
fn restrict_refuses_when_a_referencing_record_exists_and_leaves_everything_untouched() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("restrict"), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
let error = db.delete_checked("users", "u1", false).unwrap_err();
match error {
DatabaseError::Table(table, TableError::RelationRestrict(inner)) => {
assert_eq!(table, "users");
assert_eq!(inner.relation, "orders_user");
assert_eq!(inner.referencing_table, "orders");
assert_eq!(inner.referencing, 1);
}
other => panic!("expected a RelationRestrict refusal, got {other:?}"),
}
assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
db.close().unwrap();
}
#[test]
fn restrict_allows_the_delete_once_nothing_references_it() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("restrict"), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
db.close().unwrap();
}
#[test]
fn cascade_deletes_every_referencing_record() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("cascade"), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o2", "userId": "u1" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o3", "userId": "other" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
assert!(db.table("orders").unwrap().get("o2").unwrap().is_none());
assert!(db.table("orders").unwrap().get("o3").unwrap().is_some());
db.close().unwrap();
}
#[test]
fn cascade_recurses_through_a_chain_of_relations() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("cascade"), false)
.unwrap();
db.declare_relation(order_items_order_relation_cascade(), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
db.table_mut("order_items")
.unwrap()
.put(&json!({ "id": "i1", "orderId": "o1" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
assert!(db.table("order_items").unwrap().get("i1").unwrap().is_none());
db.close().unwrap();
}
#[test]
fn set_null_clears_the_referencing_field_instead_of_deleting_the_row() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("set-null"), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
assert_eq!(order.get("userId"), Some(&Value::Null));
db.close().unwrap();
}
#[test]
fn no_action_and_undeclared_leave_referencing_records_untouched() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("no-action"), false)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
let deleted = db.delete_checked("users", "u1", false).unwrap();
assert!(deleted);
let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
assert_eq!(order.get("userId"), Some(&Value::String("u1".to_string())));
db.close().unwrap();
}
#[test]
fn a_restrict_deep_in_the_cascade_aborts_the_whole_delete_atomically() {
let dir = tempdir().unwrap();
write_catalog_to(
dir.path(),
&catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
);
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(orders_user_relation_on_delete("cascade"), false)
.unwrap();
db.declare_relation(
json!({
"name": "order_items_order",
"from": { "table": "order_items", "field": "orderId" },
"to": { "table": "orders", "field": "id" },
"cardinality": "many-to-one",
"onDelete": "restrict",
}),
false,
)
.unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.table_mut("orders")
.unwrap()
.put(&json!({ "id": "o1", "userId": "u1" }), false)
.unwrap();
db.table_mut("order_items")
.unwrap()
.put(&json!({ "id": "i1", "orderId": "o1" }), false)
.unwrap();
let error = db.delete_checked("users", "u1", false).unwrap_err();
assert!(matches!(
error,
DatabaseError::Table(_, TableError::RelationRestrict(_))
));
assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
assert!(db.table("order_items").unwrap().get("i1").unwrap().is_some());
db.close().unwrap();
}
#[test]
fn a_cycle_of_cascading_relations_does_not_recurse_forever() {
let dir = tempdir().unwrap();
let a_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "a",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "bId": { "type": "string" } },
}))
.unwrap();
let b_schema = parse_schema(&json!({
"schemaVersion": 1,
"table": "b",
"primaryKey": "id",
"fields": { "id": { "type": "string" }, "aId": { "type": "string" } },
}))
.unwrap();
write_catalog_to(dir.path(), &catalog_of(&[a_schema, b_schema]));
let mut db = Database::open(dir.path()).unwrap();
db.declare_relation(
json!({
"name": "a_b",
"from": { "table": "a", "field": "bId" },
"to": { "table": "b", "field": "id" },
"cardinality": "many-to-one",
"onDelete": "cascade",
}),
false,
)
.unwrap();
db.declare_relation(
json!({
"name": "b_a",
"from": { "table": "b", "field": "aId" },
"to": { "table": "a", "field": "id" },
"cardinality": "many-to-one",
"onDelete": "cascade",
}),
false,
)
.unwrap();
db.table_mut("a")
.unwrap()
.put(&json!({ "id": "a1", "bId": "b1" }), false)
.unwrap();
db.table_mut("b")
.unwrap()
.put(&json!({ "id": "b1", "aId": "a1" }), false)
.unwrap();
let deleted = db.delete_checked("a", "a1", false).unwrap();
assert!(deleted);
assert!(db.table("a").unwrap().get("a1").unwrap().is_none());
assert!(db.table("b").unwrap().get("b1").unwrap().is_none());
db.close().unwrap();
}
#[test]
fn delete_checked_refuses_on_a_readonly_handle() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
{
let mut db = Database::open(dir.path()).unwrap();
db.table_mut("users")
.unwrap()
.put(&json!({ "id": "u1", "name": "Ania" }), false)
.unwrap();
db.close().unwrap();
}
let mut db = Database::open_with(
dir.path(),
OpenOptions {
readonly: true,
..OpenOptions::default()
},
)
.unwrap();
let error = db.delete_checked("users", "u1", false).unwrap_err();
assert!(matches!(error, DatabaseError::ReadOnly { .. }));
}
}
mod describing_one_table {
use super::*;
fn users_with_city() -> TableSchema {
parse_schema(&json!({
"schemaVersion": 1,
"table": "users",
"revision": 2,
"primaryKey": "id",
"fields": {
"id": { "type": "string" },
"name": { "type": "string", "required": true },
"city": { "type": "string", "index": "exact" },
},
}))
.unwrap()
}
fn users_drifted_under(dir: &Path) -> Database {
write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
let mut db = Database::open(dir).unwrap();
write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
assert!(db.refresh().unwrap());
db
}
fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
match answered {
Err(DatabaseError::Schema(error)) => error.to_string(),
other => panic!("expected a schema refusal, got {other:?}"),
}
}
#[test]
fn describes_a_table_of_its_own_while_another_one_has_drifted() {
let dir = tempdir().unwrap();
let db = users_drifted_under(dir.path());
let document = db.describe_table("orders").unwrap();
assert_eq!(document["table"], json!("orders"));
assert_eq!(document["revision"], json!(1));
assert!(db.catalog().is_err());
}
#[test]
fn refuses_the_table_that_drifted_itself() {
let dir = tempdir().unwrap();
let db = users_drifted_under(dir.path());
let message = refusal_of(db.describe_table("users"));
assert!(message.contains("\"users\""), "{message}");
assert!(message.contains("revision 2"), "{message}");
assert_eq!(db.table("users").unwrap().schema().revision, 1);
}
#[test]
fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
let dir = tempdir().unwrap();
let mut catalog = catalog_of(&[users_schema()]);
let entry = catalog.tables.get_mut("users").unwrap();
entry.indexes = Some(vec![IndexEntry {
name: "by_name".to_string(),
fields: vec!["name".to_string()],
kind: crate::catalog::CatalogIndexKind::Exact,
unique: false,
where_: None,
expire_after_seconds: None,
}]);
write_catalog_to(dir.path(), &catalog);
let db = Database::open(dir.path()).unwrap();
let document = db.describe_table("users").unwrap();
assert_eq!(
document["fields"]["name"],
json!({ "type": "string", "required": true })
);
assert!(document.as_object().unwrap().get("indexes").is_none());
assert!(db.catalog().is_ok());
}
#[test]
fn refuses_a_table_this_handle_holds_nowhere() {
let dir = tempdir().unwrap();
write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
let db = Database::open(dir.path()).unwrap();
let message = refusal_of(db.describe_table("ghosts"));
assert!(message.contains("declares no table \"ghosts\""), "{message}");
}
}
}