1use std::collections::{BTreeMap, HashMap, HashSet};
2use std::fmt;
3use std::fs;
4use std::io;
5use std::path::{Path, PathBuf};
6use std::time::Duration;
7
8use serde_json::Value;
9
10use crate::atomic::AtomicTemporaryFileError;
11use crate::catalog::{
12 check_integrity, declare_relation as declare_relation_in, empty_catalog, expand_indexes, index_identity,
13 parse_catalog, reference_of, relations_touching, write_catalog, Catalog, CatalogCompaction, CatalogLayout,
14 ExpandedIndex, IntegrityReport, Relation, RelationAction, TableEntry, TableOptions, TriggerEntry,
15};
16use crate::checkpoint::{read_snapshot, RestoreOutcome};
17use crate::crypto::{CipherOptions, FieldCipher};
18use crate::errors::{HaysonCipherError, HaysonRelationRestrictError, HaysonRelationViolationError, HaysonSchemaError};
19use crate::lock::{DirLock, DirLockError, LockOptions};
20use crate::query::builder::QueryBuilder;
21use crate::schema::{is_partitioned, parse_schema, FieldSpec, HaysonRecord, TableSchema};
22use crate::storage::{partition_names_in, LOG_SUFFIX};
23use crate::table::{describe_table_document, CachePolicy, FollowPolicy, Table, TableError};
24use crate::txn::{apply_prepared, recover_incomplete, PreparedUnit, TxnError, UnitOfWork, UnitOfWorkResult};
25
26use super::layout::{resolve_kind, stored_kind_of, Layout, LayoutKind, LayoutOptions, DEFAULT_METADATA_FILE};
27const DEFAULT_SNAPSHOT_EVERY: u64 = 1000;
30const DEFAULT_COMPACTION: CatalogCompaction = CatalogCompaction::Enabled {
33 dead_bytes_ratio: 0.4,
34 min_dead_bytes: 64 * 1024,
35};
36const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
37const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;
38
39#[derive(Debug)]
40pub enum DatabaseError {
41 Lock(DirLockError),
42 Schema(HaysonSchemaError),
43 ReadOnly {
46 dir: PathBuf,
47 operation: String,
48 },
49 FieldInUse {
52 table: String,
53 field: String,
54 records: usize,
55 },
56 Catalog(String),
57 Io(io::Error),
58 Log(PathBuf, io::Error),
61 Table(String, TableError),
62 Txn(TxnError),
63 Cipher(Box<HaysonCipherError>),
66}
67
68impl fmt::Display for DatabaseError {
69 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
70 match self {
71 DatabaseError::Lock(error) => error.fmt(f),
72 DatabaseError::Schema(error) => error.fmt(f),
73 DatabaseError::ReadOnly { dir, operation } => {
74 write!(f, "database \"{}\" is readonly — {operation} is not available", dir.display())
75 }
76 DatabaseError::FieldInUse { table, field, records } => write!(
77 f,
78 "field \"{field}\" of table \"{table}\" still holds a value in {records} live record(s) — pass force to drop it anyway"
79 ),
80 DatabaseError::Catalog(message) => write!(f, "{message}"),
81 DatabaseError::Io(error) => error.fmt(f),
82 DatabaseError::Log(path, error) => write!(f, "{error}, open \"{}\"", path.display()),
83 DatabaseError::Table(name, error) => write!(f, "table \"{name}\": {error}"),
84 DatabaseError::Txn(error) => error.fmt(f),
85 DatabaseError::Cipher(error) => error.fmt(f),
86 }
87 }
88}
89
90impl std::error::Error for DatabaseError {}
91
92impl From<io::Error> for DatabaseError {
93 fn from(value: io::Error) -> Self {
94 DatabaseError::Io(value)
95 }
96}
97
98impl From<DirLockError> for DatabaseError {
99 fn from(value: DirLockError) -> Self {
100 DatabaseError::Lock(value)
101 }
102}
103
104impl From<HaysonSchemaError> for DatabaseError {
105 fn from(value: HaysonSchemaError) -> Self {
106 DatabaseError::Schema(value)
107 }
108}
109
110impl From<TxnError> for DatabaseError {
111 fn from(value: TxnError) -> Self {
112 DatabaseError::Txn(value)
113 }
114}
115
116impl From<HaysonCipherError> for DatabaseError {
117 fn from(value: HaysonCipherError) -> Self {
118 DatabaseError::Cipher(Box::new(value))
119 }
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
130pub enum FollowOption {
131 #[default]
132 Default,
133 Off,
135 Every(Duration),
137}
138
139pub const DEFAULT_FOLLOW_INTERVAL: Duration = Duration::ZERO;
142
143impl FollowOption {
144 fn resolve(self, readonly: bool) -> Option<FollowPolicy> {
145 match self {
146 FollowOption::Off => None,
147 FollowOption::Every(interval) => Some(FollowPolicy { interval }),
148 FollowOption::Default if readonly => Some(FollowPolicy {
149 interval: DEFAULT_FOLLOW_INTERVAL,
150 }),
151 FollowOption::Default => None,
152 }
153 }
154}
155
156#[derive(Debug, Clone, Copy)]
163pub enum LockMode {
164 Enabled(LockOptions),
166 Skip,
168}
169
170impl Default for LockMode {
171 fn default() -> Self {
172 LockMode::Enabled(LockOptions::default())
173 }
174}
175
176impl LockMode {
177 pub fn options(&self) -> Option<&LockOptions> {
180 match self {
181 LockMode::Enabled(options) => Some(options),
182 LockMode::Skip => None,
183 }
184 }
185
186 pub fn options_mut(&mut self) -> Option<&mut LockOptions> {
188 match self {
189 LockMode::Enabled(options) => Some(options),
190 LockMode::Skip => None,
191 }
192 }
193}
194
195#[derive(Debug, Clone)]
197pub struct OpenOptions {
198 pub readonly: bool,
199 pub lock: LockMode,
200 pub snapshot_every: u64,
203 pub compaction: Option<CatalogCompaction>,
208 pub cypher: Option<CipherOptions>,
210 pub follow: FollowOption,
212 pub cache: CachePolicy,
217 pub max_image_bytes: Option<u64>,
221 pub layout: LayoutOptions,
226}
227
228impl Default for OpenOptions {
229 fn default() -> Self {
230 OpenOptions {
231 readonly: false,
232 lock: LockMode::default(),
233 snapshot_every: DEFAULT_SNAPSHOT_EVERY,
234 compaction: None,
235 cypher: None,
236 follow: FollowOption::default(),
237 cache: CachePolicy::default(),
238 max_image_bytes: None,
239 layout: LayoutOptions::default(),
240 }
241 }
242}
243
244fn now_ms() -> u64 {
245 chrono::Utc::now().timestamp_millis().max(0) as u64
246}
247
248fn scalar_as_text(value: &Value) -> String {
253 match value {
254 Value::String(s) => s.clone(),
255 other => other.to_string(),
256 }
257}
258
259fn fnv1a(bytes: &[u8]) -> String {
260 let mut hash = FNV_OFFSET;
261 for byte in bytes {
262 hash ^= u64::from(*byte);
263 hash = hash.wrapping_mul(FNV_PRIME);
264 }
265 format!("{hash:016x}")
266}
267
268fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
276 if declared.is_empty() {
277 return fnv1a(format!("{schema:?}").as_bytes());
278 }
279 let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
280 described.sort();
281 fnv1a(format!("{schema:?}|{}", described.join("|")).as_bytes())
282}
283
284fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
289 let fields: BTreeMap<String, FieldSpec> = entry
290 .fields
291 .iter()
292 .map(|(field, spec)| (field.clone(), spec.spec.clone()))
293 .collect();
294 TableSchema {
295 table: name.to_string(),
296 revision: entry.revision as u32,
297 primary_key: entry.primary_key.clone(),
298 partition_by: entry.partition_by.clone(),
299 fields,
300 }
301}
302
303fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
306 let Some(declared) = catalog.triggers.as_ref() else {
307 return Vec::new();
308 };
309 declared
310 .iter()
311 .filter(|trigger| trigger.table == name)
312 .cloned()
313 .collect()
314}
315
316fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
322 for (name, table) in tables.iter_mut() {
323 table.set_triggers(triggers_of(catalog, name));
324 }
325}
326
327fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
331 entry
332 .options
333 .as_ref()
334 .and_then(&pick)
335 .or_else(|| catalog.defaults.as_ref().and_then(&pick))
336}
337
338fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
342 if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
343 table.set_image_dir(dir.join(image_dir));
344 }
345}
346
347fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
351 if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
352 table.set_sync(sync);
353 }
354}
355
356fn apply_validation(entry: &TableEntry, table: &mut Table) {
364 if let Some(validation) = entry.validation {
365 table.set_validation(validation);
366 }
367}
368
369fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
376 let marked: BTreeMap<String, String> = entry
377 .fields
378 .iter()
379 .filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
380 .collect();
381 if !marked.is_empty() {
382 table.set_deprecated_fields(marked);
383 }
384}
385
386fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
390 if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
391 table.set_link_parsing(link_parsing);
392 }
393}
394
395fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
399 named
400 .or_else(|| {
401 entry
402 .and_then(|entry| entry.options.as_ref())
403 .and_then(|options| options.compaction)
404 })
405 .or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
406 .unwrap_or(DEFAULT_COMPACTION)
407}
408
409fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
412 table.set_compaction(compaction);
413}
414
415fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
419 table.set_cipher(cipher.cloned());
420}
421
422fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
426 let all = open_table
427 .query(&QueryBuilder::new().spec())
428 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
429 let mut residual = 0usize;
430 for key in &all.keys {
431 let record = open_table
432 .get(key)
433 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
434 if record.is_some_and(|record| record.contains_key(field)) {
435 residual += 1;
436 }
437 }
438 if residual > 0 {
439 return Err(DatabaseError::FieldInUse {
440 table: name.to_string(),
441 field: field.to_string(),
442 records: residual,
443 });
444 }
445 Ok(())
446}
447
448fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
454 match error.downcast::<io::Error>() {
455 Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
456 Ok(io) => DatabaseError::Catalog(io.to_string()),
457 Err(other) => DatabaseError::Catalog(other.to_string()),
458 }
459}
460
461fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
464 layout.tables().join(table)
465}
466
467fn on_disk(path: &Path) -> bool {
470 fs::symlink_metadata(path).is_ok()
471}
472
473fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
481 let flat = layout.log_path(&schema.table);
482 if !is_partitioned(schema) {
483 return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
484 }
485 let dir = partition_dir_of(layout, &schema.table);
486 let present = partition_names_in(&dir);
487 if present.is_empty() && on_disk(&flat) {
489 return Ok(Vec::new());
490 }
491 let claimed = read_snapshot(index_path)?
492 .and_then(|snapshot| snapshot.header.parts)
493 .unwrap_or_default();
494 Ok(claimed
495 .into_iter()
496 .filter(|part| !present.contains(&part.part))
497 .map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
498 .collect())
499}
500
501fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
503 let flat = layout.log_path(&schema.table);
504 if !is_partitioned(schema) {
505 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
506 }
507 let dir = partition_dir_of(layout, &schema.table);
508 let present = partition_names_in(&dir);
509 if present.is_empty() {
510 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
511 }
512 present
513 .iter()
514 .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
515 .collect()
516}
517
518enum TableDrift {
521 Absent,
522 Schema { revision: u64 },
523 Indexes { unserved: Vec<String>, stale: Vec<String> },
524}
525
526struct DriftedTable {
527 name: String,
528 drift: TableDrift,
529}
530
531fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
534 let identities: HashSet<String> = present.iter().map(index_identity).collect();
535 let mut names: Vec<String> = expected
536 .iter()
537 .filter(|index| !identities.contains(&index_identity(index)))
538 .map(|index| index.name.clone())
539 .collect();
540 names.sort();
541 names
542}
543
544fn shape_of(schema: &TableSchema) -> TableSchema {
553 let mut bare = schema.clone();
554 for spec in bare.fields.values_mut() {
555 spec.index = None;
556 spec.unique = false;
557 spec.fulltext = false;
558 }
559 bare
560}
561
562fn quoted(names: &[String]) -> String {
563 names
564 .iter()
565 .map(|name| format!("\"{name}\""))
566 .collect::<Vec<_>>()
567 .join(", ")
568}
569
570fn stored_side_of(drift: &TableDrift) -> String {
571 match drift {
572 TableDrift::Absent => "declares no table at all".to_string(),
573 TableDrift::Schema { revision } => format!("declares revision {revision}"),
574 TableDrift::Indexes { unserved, stale } => {
575 let mut halves = Vec::new();
576 if !unserved.is_empty() {
577 halves.push(format!(
578 "declares indexes this handle does not serve: {}",
579 quoted(unserved)
580 ));
581 }
582 if !stale.is_empty() {
583 halves.push(format!(
584 "no longer declares indexes this handle still serves: {}",
585 quoted(stale)
586 ));
587 }
588 halves.join(", and ")
589 }
590 }
591}
592
593fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
594 let listed = missing
595 .iter()
596 .map(|path| format!("\"{}\"", path.display()))
597 .collect::<Vec<_>>()
598 .join(", ");
599 let one = missing.len() == 1;
600 let (its, is, it, file, log) = if one {
601 ("its log".to_string(), "is", "it", "file", "log")
602 } else {
603 (format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
604 };
605 HaysonSchemaError(format!(
606 "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",
607 dir.display()
608 ))
609}
610
611fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
622 let schema = table_schema_of(name, entry);
623 let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
624 if !missing.is_empty() {
625 return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
626 }
627 for path in present_logs_of(layout, &schema) {
628 fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
629 }
630 Ok(())
631}
632
633#[allow(clippy::too_many_arguments)]
641fn open_catalog_table(
642 layout: &Layout,
643 name: &str,
644 catalog: &Catalog,
645 entry: &TableEntry,
646 snapshot_every: u64,
647 cipher: Option<&FieldCipher>,
648 follow: Option<FollowPolicy>,
649 readonly: bool,
650 compaction: CatalogCompaction,
651 cache: CachePolicy,
652 max_image_bytes: Option<u64>,
653) -> Result<(Table, RestoreOutcome), DatabaseError> {
654 let schema = table_schema_of(name, entry);
655 let declared = expand_indexes(name, entry);
656 let hash = schema_fingerprint(&schema, &declared);
657 let (mut table, outcome) = Table::open_with_snapshot_and_declared_indexes(
658 schema,
659 layout.log_path(name),
660 layout.index_path(name),
661 &hash,
662 snapshot_every,
663 readonly,
664 &declared,
665 )
666 .map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
667 apply_image_dir(layout.root(), catalog, entry, &mut table);
668 apply_link_parsing(catalog, entry, &mut table);
669 apply_sync(catalog, entry, &mut table);
670 apply_validation(entry, &mut table);
671 apply_deprecated(entry, &mut table);
672 apply_compaction(compaction, &mut table);
673 table.set_cache(cache);
674 table.set_max_image_bytes(max_image_bytes);
675 apply_cipher(cipher, &mut table);
676 table.set_follow(follow);
677 table.set_triggers(triggers_of(catalog, name));
678 Ok((table, outcome))
679}
680
681fn first_table_log(layout: &Layout) -> Option<PathBuf> {
685 let has_log_suffix = |path: &Path| {
686 path.extension()
687 .is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
688 };
689 let entries_of = |dir: &Path| -> Vec<PathBuf> {
690 let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
691 let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
692 found.sort();
693 found
694 };
695
696 let top = entries_of(layout.tables());
697 if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
698 return Some(log.clone());
699 }
700 top.iter().filter(|path| path.is_dir()).find_map(|dir| {
701 entries_of(dir)
702 .into_iter()
703 .find(|path| path.is_file() && has_log_suffix(path))
704 })
705}
706
707fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
715 let Some(log) = first_table_log(layout) else {
716 return Ok(());
717 };
718 Err(HaysonSchemaError(format!(
719 "\"{}\" 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",
720 layout.root().display(),
721 layout.metadata_file(),
722 log.display(),
723 layout.metadata_file(),
724 )))
725}
726
727fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
728 let raw = match fs::read_to_string(path) {
729 Ok(raw) => raw,
730 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
731 Err(error) => return Err(error.into()),
732 };
733 let value: Value = serde_json::from_str(&raw)
734 .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
735 Ok(Some(parse_catalog(&value)?))
736}
737
738pub(super) struct HeldLock(pub(super) Option<DirLock>);
749
750impl HeldLock {
751 fn into_held(mut self) -> Option<DirLock> {
752 self.0.take()
753 }
754}
755
756impl Drop for HeldLock {
757 fn drop(&mut self) {
758 if let Some(lock) = self.0.as_mut() {
759 lock.release();
760 }
761 }
762}
763
764pub struct Database {
768 layout: Layout,
769 lock: Option<DirLock>,
770 catalog: Catalog,
771 tables: HashMap<String, Table>,
772 restore_outcomes: HashMap<String, RestoreOutcome>,
773 readonly: bool,
774 snapshot_every: u64,
775 compaction: Option<CatalogCompaction>,
778 cipher: Option<FieldCipher>,
779 follow: Option<FollowPolicy>,
780 cache: CachePolicy,
783 max_image_bytes: Option<u64>,
784}
785
786impl Database {
787 pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
788 Database::open_with(dir, OpenOptions::default())
789 }
790
791 pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
805 let (layout, lock) = match options.lock {
806 LockMode::Enabled(lock) if !options.readonly => Database::hold_for_open(dir, &options.layout, lock)?,
807 LockMode::Enabled(_) | LockMode::Skip => (Database::resolve_layout(dir, &options.layout)?, HeldLock(None)),
808 };
809
810 if !options.readonly {
811 for directory in layout.directories() {
812 fs::create_dir_all(directory)?;
813 }
814 }
815
816 Database::open_locked(layout, options, lock)
817 }
818
819 fn hold_for_open(
840 dir: &Path,
841 options: &LayoutOptions,
842 lock: LockOptions,
843 ) -> Result<(Layout, HeldLock), DatabaseError> {
844 const SETTLE_ATTEMPTS: usize = 3;
845 let mut probed = Database::resolve_layout(dir, options)?;
846 for _ in 0..SETTLE_ATTEMPTS {
847 fs::create_dir_all(probed.runtime())?;
848 let held = HeldLock(Some(DirLock::acquire(probed.lock_path(), lock)?));
849 let settled = Database::resolve_layout(dir, options)?;
850 if settled.kind() == probed.kind() {
851 return Ok((settled, held));
852 }
853 drop(held);
855 if probed.kind() == LayoutKind::Nested {
856 let _ = fs::remove_dir(probed.runtime());
857 }
858 probed = settled;
859 }
860 Err(HaysonSchemaError(format!(
861 "\"{}\" 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",
862 dir.display()
863 ))
864 .into())
865 }
866
867 pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
881 let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
882 let stored = stored_kind_of(&dir.join(metadata_file));
883 let kind = resolve_kind(dir, stored, options.kind)?;
884 Ok(Layout::new(dir, kind, metadata_file))
885 }
886
887 fn open_locked(layout: Layout, options: OpenOptions, lock: HeldLock) -> Result<Database, DatabaseError> {
888 let found = read_catalog(layout.metadata_path())?;
889 let catalog = match found {
890 Some(catalog) => catalog,
891 None => {
892 refuse_orphan_logs(&layout)?;
893 let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
894 if !options.readonly {
895 write_catalog(layout.metadata_path(), &catalog)
896 .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
897 }
898 catalog
899 }
900 };
901
902 let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
903 let follow = options.follow.resolve(options.readonly);
904
905 let mut tables = HashMap::new();
906 let mut restore_outcomes = HashMap::new();
907 for (name, entry) in &catalog.tables {
908 if options.readonly {
909 refuse_unreadable_logs(&layout, name, entry)?;
910 }
911 let (table, outcome) = open_catalog_table(
912 &layout,
913 name,
914 &catalog,
915 entry,
916 options.snapshot_every,
917 cipher.as_ref(),
918 follow,
919 options.readonly,
920 compaction_of(options.compaction, &catalog, Some(entry)),
921 options.cache,
922 options.max_image_bytes,
923 )?;
924 tables.insert(name.clone(), table);
925 restore_outcomes.insert(name.clone(), outcome);
926 }
927
928 if !options.readonly {
929 recover_incomplete(&layout.journal_path(), &mut tables)?;
930 }
931
932 sync_table_triggers(&catalog, &mut tables);
933
934 Ok(Database {
935 layout,
936 lock: lock.into_held(),
937 catalog,
938 tables,
939 restore_outcomes,
940 readonly: options.readonly,
941 snapshot_every: options.snapshot_every,
942 compaction: options.compaction,
943 cipher,
944 follow,
945 cache: options.cache,
946 max_image_bytes: options.max_image_bytes,
947 })
948 }
949
950 pub fn dir(&self) -> &Path {
951 self.layout.root()
952 }
953
954 pub fn layout(&self) -> &Layout {
957 &self.layout
958 }
959
960 pub fn follow(&self) -> Option<FollowPolicy> {
963 self.follow
964 }
965
966 pub fn table_names(&self) -> Vec<&str> {
967 let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
968 names.sort_unstable();
969 names
970 }
971
972 pub fn table(&self, name: &str) -> Option<&Table> {
973 self.tables.get(name)
974 }
975
976 pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
982 let _ = self.renew_lock();
983 self.tables.get_mut(name)
984 }
985
986 fn assert_relations_hold(&self, table: &str, values: &Value) -> Result<(), DatabaseError> {
1004 for relation in relations_touching(&self.catalog, Some(table)) {
1005 if relation.from.table != table || relation.enforce != Some(true) {
1006 continue;
1007 }
1008 let Some(value) = reference_of(values.get(&relation.from.field)) else {
1009 continue;
1010 };
1011 let Some(to_table) = self.tables.get(&relation.to.table) else {
1012 continue;
1013 };
1014 if !to_table.contains_value(&relation.to.field, &value) {
1015 return Err(DatabaseError::Table(
1016 table.to_string(),
1017 TableError::Relation(HaysonRelationViolationError {
1018 relation: relation.name.clone(),
1019 table: table.to_string(),
1020 field: relation.from.field.clone(),
1021 value: scalar_as_text(&value),
1022 }),
1023 ));
1024 }
1025 }
1026 Ok(())
1027 }
1028
1029 pub fn put_checked(&mut self, table: &str, document: &Value, force: bool) -> Result<HaysonRecord, DatabaseError> {
1034 self.assert_relations_hold(table, document)?;
1035 let root = self.layout.root().display().to_string();
1036 let handle = self
1037 .table_mut(table)
1038 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1039 handle
1040 .put(document, force)
1041 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1042 }
1043
1044 pub fn patch_checked(
1047 &mut self,
1048 table: &str,
1049 key: &str,
1050 changes: &Value,
1051 force: bool,
1052 ) -> Result<HaysonRecord, DatabaseError> {
1053 self.assert_relations_hold(table, changes)?;
1054 let root = self.layout.root().display().to_string();
1055 let handle = self
1056 .table_mut(table)
1057 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1058 handle
1059 .patch(key, changes, force)
1060 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1061 }
1062
1063 pub fn stage_put_checked(
1068 &mut self,
1069 table: &str,
1070 document: &Value,
1071 force: bool,
1072 ) -> Result<HaysonRecord, DatabaseError> {
1073 self.assert_relations_hold(table, document)?;
1074 let root = self.layout.root().display().to_string();
1075 let handle = self
1076 .table_mut(table)
1077 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1078 handle
1079 .stage_put(document, force)
1080 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1081 }
1082
1083 pub fn stage_patch_checked(
1086 &mut self,
1087 table: &str,
1088 key: &str,
1089 changes: &Value,
1090 force: bool,
1091 ) -> Result<HaysonRecord, DatabaseError> {
1092 self.assert_relations_hold(table, changes)?;
1093 let root = self.layout.root().display().to_string();
1094 let handle = self
1095 .table_mut(table)
1096 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist in \"{root}\"")))?;
1097 handle
1098 .stage_patch(key, changes, force)
1099 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1100 }
1101
1102 pub fn delete_checked(&mut self, table: &str, key: &str, force: bool) -> Result<bool, DatabaseError> {
1117 if self.readonly {
1118 return Err(DatabaseError::ReadOnly {
1119 dir: self.layout.root().to_path_buf(),
1120 operation: "delete".to_string(),
1121 });
1122 }
1123 let root = self.layout.root().display().to_string();
1124 if self.tables.get(table).is_none() {
1125 return Err(DatabaseError::Catalog(format!(
1126 "table \"{table}\" does not exist in \"{root}\""
1127 )));
1128 }
1129 let incoming = relations_touching(&self.catalog, Some(table))
1130 .into_iter()
1131 .any(|relation| relation.to.table == table);
1132 if !incoming {
1133 self.renew_lock()?;
1134 let handle = self.tables.get_mut(table).expect("checked above");
1135 return handle
1136 .delete(key, force)
1137 .map_err(|error| DatabaseError::Table(table.to_string(), error));
1138 }
1139
1140 let mut unit = UnitOfWork::new();
1141 let mut visited: HashSet<(String, String)> = HashSet::new();
1142 let existed = self.plan_delete_cascade(table, key, &mut visited, &mut unit)?;
1143 if !existed {
1144 return Ok(false);
1145 }
1146 self.commit(&mut unit)?;
1147 Ok(true)
1148 }
1149
1150 fn plan_delete_cascade(
1160 &self,
1161 table: &str,
1162 key: &str,
1163 visited: &mut HashSet<(String, String)>,
1164 unit: &mut UnitOfWork,
1165 ) -> Result<bool, DatabaseError> {
1166 if !visited.insert((table.to_string(), key.to_string())) {
1167 return Ok(true);
1168 }
1169 let handle = self
1170 .tables
1171 .get(table)
1172 .ok_or_else(|| DatabaseError::Catalog(format!("table \"{table}\" does not exist")))?;
1173 let Some(record) = handle
1174 .get(key)
1175 .map_err(|error| DatabaseError::Table(table.to_string(), error))?
1176 else {
1177 return Ok(false);
1178 };
1179
1180 for relation in relations_touching(&self.catalog, Some(table)) {
1181 if relation.to.table != table {
1182 continue;
1183 }
1184 let action = relation.on_delete.unwrap_or(RelationAction::NoAction);
1185 if matches!(action, RelationAction::NoAction) {
1186 continue;
1187 }
1188 let Some(value) = reference_of(record.get(&relation.to.field)) else {
1189 continue;
1190 };
1191 let Some(from_table) = self.tables.get(&relation.from.table) else {
1192 continue;
1193 };
1194 let referencing = from_table
1195 .keys_referencing(&relation.from.field, &value)
1196 .map_err(|error| DatabaseError::Table(relation.from.table.clone(), error))?;
1197 if referencing.is_empty() {
1198 continue;
1199 }
1200
1201 match action {
1202 RelationAction::Restrict => {
1203 return Err(DatabaseError::Table(
1204 table.to_string(),
1205 TableError::RelationRestrict(HaysonRelationRestrictError {
1206 relation: relation.name.clone(),
1207 table: table.to_string(),
1208 key: key.to_string(),
1209 referencing_table: relation.from.table.clone(),
1210 referencing_field: relation.from.field.clone(),
1211 referencing: referencing.len(),
1212 }),
1213 ));
1214 }
1215 RelationAction::Cascade => {
1216 for referencing_key in &referencing {
1217 self.plan_delete_cascade(&relation.from.table, referencing_key, visited, unit)?;
1218 }
1219 }
1220 RelationAction::SetNull => {
1221 for referencing_key in &referencing {
1222 unit.patch(
1223 relation.from.table.as_str(),
1224 referencing_key.clone(),
1225 Value::Object(serde_json::Map::from_iter([(relation.from.field.clone(), Value::Null)])),
1226 );
1227 }
1228 }
1229 RelationAction::NoAction => unreachable!("filtered above"),
1230 }
1231 }
1232
1233 unit.delete(table, key.to_string());
1234 Ok(true)
1235 }
1236
1237 pub fn tick(&self) -> Result<(), DatabaseError> {
1242 self.renew_lock()
1243 }
1244
1245 fn renew_lock(&self) -> Result<(), DatabaseError> {
1246 match &self.lock {
1247 Some(lock) => Ok(lock.tick()?),
1248 None => Ok(()),
1249 }
1250 }
1251
1252 pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
1255 self.restore_outcomes.get(name).copied()
1256 }
1257
1258 pub fn catalog_unchecked(&self) -> &Catalog {
1262 &self.catalog
1263 }
1264
1265 pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
1273 let drifted = self.drifted_tables();
1274 if drifted.is_empty() {
1275 return Ok(&self.catalog);
1276 }
1277 Err(DatabaseError::Schema(self.drift_error(&drifted)))
1278 }
1279
1280 pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
1292 if let Some(drift) = self.drift_of(name) {
1293 let drifted = [DriftedTable {
1294 name: name.to_string(),
1295 drift,
1296 }];
1297 return Err(DatabaseError::Schema(self.drift_error(&drifted)));
1298 }
1299 let table = self.tables.get(name).ok_or_else(|| {
1300 DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
1301 })?;
1302 Ok(describe_table_document(table))
1303 }
1304
1305 fn drift_of(&self, name: &str) -> Option<TableDrift> {
1306 let table = self.tables.get(name)?;
1307 let Some(entry) = self.catalog.tables.get(name) else {
1308 return Some(TableDrift::Absent);
1309 };
1310 if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
1311 return Some(TableDrift::Schema {
1312 revision: entry.revision,
1313 });
1314 }
1315 let declared = expand_indexes(name, entry);
1316 let served = table.declared_indexes();
1317 let unserved = index_names_missing_from(&declared, served);
1318 let stale = index_names_missing_from(served, &declared);
1319 if unserved.is_empty() && stale.is_empty() {
1320 return None;
1321 }
1322 Some(TableDrift::Indexes { unserved, stale })
1323 }
1324
1325 fn drifted_tables(&self) -> Vec<DriftedTable> {
1326 let mut names: Vec<&String> = self.tables.keys().collect();
1327 names.sort();
1328 names
1329 .into_iter()
1330 .filter_map(|name| {
1331 self.drift_of(name).map(|drift| DriftedTable {
1332 name: name.clone(),
1333 drift,
1334 })
1335 })
1336 .collect()
1337 }
1338
1339 fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
1340 let detail = drifted
1341 .iter()
1342 .map(|DriftedTable { name, drift }| {
1343 let revision = self.tables.get(name).map(|table| table.schema().revision);
1344 let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
1345 format!(
1346 "\"{name}\" is open here at {open} while the catalog {}",
1347 stored_side_of(drift)
1348 )
1349 })
1350 .collect::<Vec<_>>()
1351 .join("; ");
1352 HaysonSchemaError(format!(
1353 "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",
1354 self.layout.root().display()
1355 ))
1356 }
1357
1358 pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
1362 relations_touching(&self.catalog, table)
1363 }
1364
1365 pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
1368 let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
1369 check_integrity(&self.catalog, &refs)
1370 }
1371
1372 pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
1382 let mut changed = false;
1383 if let Some(catalog) = read_catalog(self.layout.metadata_path())? {
1384 if catalog != self.catalog {
1385 self.catalog = catalog;
1386 changed = true;
1387 }
1388 }
1389
1390 let missing: Vec<String> = self
1391 .catalog
1392 .tables
1393 .keys()
1394 .filter(|name| !self.tables.contains_key(*name))
1395 .cloned()
1396 .collect();
1397 for name in missing {
1398 let Some(entry) = self.catalog.tables.get(&name) else {
1399 continue;
1400 };
1401 let (table, outcome) = open_catalog_table(
1402 &self.layout,
1403 &name,
1404 &self.catalog,
1405 entry,
1406 self.snapshot_every,
1407 self.cipher.as_ref(),
1408 self.follow,
1409 self.readonly,
1410 compaction_of(self.compaction, &self.catalog, Some(entry)),
1411 self.cache,
1412 self.max_image_bytes,
1413 )?;
1414 self.tables.insert(name.clone(), table);
1415 self.restore_outcomes.insert(name, outcome);
1416 changed = true;
1417 }
1418
1419 for (name, table) in self.tables.iter_mut() {
1420 let outcome = table
1421 .refresh()
1422 .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
1423 changed |= outcome.changed();
1424 }
1425
1426 sync_table_triggers(&self.catalog, &mut self.tables);
1428 Ok(changed)
1429 }
1430
1431 pub fn unit_of_work(&mut self) -> UnitOfWork {
1434 UnitOfWork::new()
1435 }
1436
1437 pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
1444 let prepared = self.prepare_commit(unit)?;
1445 unit.journal(&prepared, &self.layout.journal_path())?;
1446 self.apply_commit(prepared)
1447 }
1448
1449 pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
1458 if self.readonly {
1459 return Err(DatabaseError::ReadOnly {
1460 dir: self.layout.root().to_path_buf(),
1461 operation: "commit".to_string(),
1462 });
1463 }
1464 self.renew_lock()?;
1465 Ok(unit.prepare(&mut self.tables)?)
1466 }
1467
1468 pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
1472 let journal_path = self.layout.journal_path();
1473 Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
1474 }
1475
1476 fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
1480 if self.readonly {
1481 return Err(DatabaseError::ReadOnly {
1482 dir: self.layout.root().to_path_buf(),
1483 operation: operation.to_string(),
1484 });
1485 }
1486 self.renew_lock()
1487 }
1488
1489 fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
1490 write_catalog(self.layout.metadata_path(), catalog).map_err(catalog_write_error)
1491 }
1492
1493 pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
1508 self.assert_writable("createTable")?;
1509 let schema = parse_schema(&schema_spec)?;
1510 if schema.table != name {
1511 return Err(DatabaseError::Schema(HaysonSchemaError(format!(
1512 "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
1513 schema.table
1514 ))));
1515 }
1516 let now = now_ms();
1517 let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
1518 self.write_catalog_to_disk(&next_catalog)?;
1519
1520 let next_entry = &next_catalog.tables[name];
1521 let (table, outcome) = open_catalog_table(
1522 &self.layout,
1523 name,
1524 &next_catalog,
1525 next_entry,
1526 self.snapshot_every,
1527 self.cipher.as_ref(),
1528 self.follow,
1529 self.readonly,
1530 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1531 self.cache,
1532 self.max_image_bytes,
1533 )?;
1534
1535 self.catalog = next_catalog;
1536 self.tables.insert(name.to_string(), table);
1537 self.restore_outcomes.insert(name.to_string(), outcome);
1538 Ok(())
1539 }
1540
1541 pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
1554 self.assert_writable("dropField")?;
1555 self.flush_table(table)?;
1556 let existing = self.tables.get(table).ok_or_else(|| {
1557 DatabaseError::Catalog(format!(
1558 "table \"{table}\" does not exist in \"{}\"",
1559 self.layout.root().display()
1560 ))
1561 })?;
1562
1563 let now = now_ms();
1564 let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
1565
1566 if !force {
1567 assert_no_residual_values(table, existing, field)?;
1568 }
1569
1570 self.write_catalog_to_disk(&next_catalog)?;
1571
1572 let (mut table_handle, _) = open_catalog_table(
1573 &self.layout,
1574 table,
1575 &next_catalog,
1576 &next_entry,
1577 self.snapshot_every,
1578 self.cipher.as_ref(),
1579 self.follow,
1580 self.readonly,
1581 compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
1582 self.cache,
1583 self.max_image_bytes,
1584 )?;
1585 table_handle
1587 .snapshot()
1588 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1589
1590 self.catalog = next_catalog;
1591 self.tables.insert(table.to_string(), table_handle);
1592 Ok(())
1593 }
1594
1595 pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
1604 self.assert_writable("declareIndex")?;
1605 self.flush_table(table)?;
1606 let existing = self.tables.get(table).ok_or_else(|| {
1607 DatabaseError::Catalog(format!(
1608 "table \"{table}\" does not exist in \"{}\"",
1609 self.layout.root().display()
1610 ))
1611 })?;
1612
1613 let now = now_ms();
1614 let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
1615 self.write_catalog_to_disk(&next_catalog)?;
1616
1617 let next_entry = &next_catalog.tables[table];
1618 let (mut table_handle, _) = open_catalog_table(
1619 &self.layout,
1620 table,
1621 &next_catalog,
1622 next_entry,
1623 self.snapshot_every,
1624 self.cipher.as_ref(),
1625 self.follow,
1626 self.readonly,
1627 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1628 self.cache,
1629 self.max_image_bytes,
1630 )?;
1631 table_handle
1633 .snapshot()
1634 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1635
1636 self.catalog = next_catalog;
1637 self.tables.insert(table.to_string(), table_handle);
1638 Ok(())
1639 }
1640
1641 pub fn declare_relation(&mut self, relation_spec: Value, force: bool) -> Result<(), DatabaseError> {
1651 self.assert_writable("declareRelation")?;
1652
1653 let from_table = relation_spec
1654 .get("from")
1655 .and_then(|end| end.get("table"))
1656 .and_then(Value::as_str)
1657 .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"from.table\"".to_string()))?
1658 .to_string();
1659 let to_table = relation_spec
1660 .get("to")
1661 .and_then(|end| end.get("table"))
1662 .and_then(Value::as_str)
1663 .ok_or_else(|| DatabaseError::Catalog("relation spec needs a string \"to.table\"".to_string()))?
1664 .to_string();
1665
1666 self.flush_table(&from_table)?;
1667 self.flush_table(&to_table)?;
1668 let from_handle = self.tables.get(&from_table).ok_or_else(|| {
1669 DatabaseError::Catalog(format!(
1670 "table \"{from_table}\" does not exist in \"{}\"",
1671 self.layout.root().display()
1672 ))
1673 })?;
1674 let to_handle = self.tables.get(&to_table).ok_or_else(|| {
1675 DatabaseError::Catalog(format!(
1676 "table \"{to_table}\" does not exist in \"{}\"",
1677 self.layout.root().display()
1678 ))
1679 })?;
1680
1681 let now = now_ms();
1682 let next_catalog = declare_relation_in(&self.catalog, &relation_spec, from_handle, to_handle, now, force)?;
1683 self.write_catalog_to_disk(&next_catalog)?;
1684 self.catalog = next_catalog;
1685 Ok(())
1686 }
1687
1688 pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
1694 self.assert_writable("declareTrigger")?;
1695
1696 let now = now_ms();
1697 let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
1698 self.write_catalog_to_disk(&next_catalog)?;
1699
1700 self.catalog = next_catalog;
1701 sync_table_triggers(&self.catalog, &mut self.tables);
1702 Ok(())
1703 }
1704
1705 pub fn flush(&mut self) -> Result<(), DatabaseError> {
1709 for (name, table) in &mut self.tables {
1710 table
1711 .flush()
1712 .map_err(|error| DatabaseError::Table(name.clone(), error))?;
1713 }
1714 Ok(())
1715 }
1716
1717 fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
1721 let Some(handle) = self.tables.get_mut(table) else {
1722 return Ok(());
1723 };
1724 handle
1725 .flush()
1726 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1727 }
1728
1729 pub fn close(mut self) -> io::Result<()> {
1739 let flushed = if self.readonly {
1740 Ok(())
1741 } else {
1742 self.flush_and_snapshot_all()
1743 };
1744 if let Some(mut lock) = self.lock.take() {
1745 lock.release();
1746 }
1747 flushed
1748 }
1749
1750 fn flush_and_snapshot_all(&mut self) -> io::Result<()> {
1751 let mut first_refusal = None;
1752 for table in self.tables.values_mut() {
1753 let outcome = table.flush().map_err(io::Error::from).and_then(|()| {
1755 if table.needs_snapshot() {
1756 table.snapshot()
1757 } else {
1758 Ok(())
1759 }
1760 });
1761 if let Err(error) = outcome {
1762 first_refusal.get_or_insert(error);
1763 }
1764 }
1765 match first_refusal {
1766 Some(error) => Err(error),
1767 None => Ok(()),
1768 }
1769 }
1770}
1771
1772#[cfg(test)]
1773mod tests {
1774 use serde_json::json;
1775 use tempfile::tempdir;
1776
1777 use super::*;
1778 use crate::catalog::create_table;
1779 use crate::catalog::types::{CatalogIndexKind, IndexEntry};
1780 use crate::checkpoint::write_snapshot;
1781 use crate::registry::table_registry::TableRegistry;
1782 use crate::schema::parse_schema;
1783 use crate::storage::{serialize_entry, LogMeta, LogOp};
1784 use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
1785 use crate::txn::JOURNAL_FILE;
1786
1787 const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
1788
1789 fn flat(dir: &Path) -> Layout {
1790 Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
1791 }
1792
1793 fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
1794 flat(dir).log_path(table)
1795 }
1796
1797 #[test]
1801 fn a_database_and_its_tables_can_be_shared_across_threads() {
1802 fn shareable<T: Send + Sync>() {}
1803 shareable::<Database>();
1804 shareable::<Table>();
1805 shareable::<crate::registry::fulltext_index::FulltextIndex>();
1806 }
1807
1808 fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
1809 flat(dir).index_path(table)
1810 }
1811
1812 fn users_schema() -> TableSchema {
1813 parse_schema(&json!({
1814 "schemaVersion": 1,
1815 "table": "users",
1816 "primaryKey": "id",
1817 "fields": {
1818 "id": { "type": "string" },
1819 "name": { "type": "string", "required": true },
1820 },
1821 }))
1822 .unwrap()
1823 }
1824
1825 fn orders_schema() -> TableSchema {
1826 parse_schema(&json!({
1827 "schemaVersion": 1,
1828 "table": "orders",
1829 "primaryKey": "id",
1830 "fields": {
1831 "id": { "type": "string" },
1832 "userId": { "type": "string", "required": true },
1833 },
1834 }))
1835 .unwrap()
1836 }
1837
1838 fn catalog_of(schemas: &[TableSchema]) -> Catalog {
1839 let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
1840 for schema in schemas {
1841 catalog = create_table(&catalog, &schema.table, schema, 1_000);
1842 }
1843 catalog
1844 }
1845
1846 fn write_catalog_to(dir: &Path, catalog: &Catalog) {
1847 write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
1848 }
1849
1850 #[test]
1851 fn open_builds_a_database_with_every_declared_table_available() {
1852 let dir = tempdir().unwrap();
1853 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1854
1855 let mut db = Database::open(dir.path()).unwrap();
1856 assert_eq!(db.table_names(), vec!["orders", "users"]);
1857
1858 db.table_mut("users")
1859 .unwrap()
1860 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1861 .unwrap();
1862 assert_eq!(
1863 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1864 Some(&json!("Ania"))
1865 );
1866 db.close().unwrap();
1867 }
1868
1869 #[test]
1870 fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
1871 let dir = tempdir().unwrap();
1872 let catalog = catalog_of(&[users_schema()]);
1873 write_catalog_to(dir.path(), &catalog);
1874
1875 let schema = table_schema_of("users", &catalog.tables["users"]);
1876 let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
1877 let log_path = log_path(dir.path(), "users");
1878 let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
1879 log.append(&[serialize_entry(
1880 &LogMeta {
1881 key: "u1".into(),
1882 op: LogOp::Put,
1883 seq: 1,
1884 ts: 1,
1885 },
1886 Some(&{
1887 let mut record = crate::schema::HaysonRecord::new();
1888 record.insert("id".into(), json!("u1"));
1889 record.insert("name".into(), json!("Ania"));
1890 record
1891 }),
1892 )])
1893 .unwrap();
1894 let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
1895 write_snapshot(&index_path(dir.path(), "users"), ®istry, "users", &hash, 0, &log).unwrap();
1896
1897 let db = Database::open(dir.path()).unwrap();
1898 assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
1899 assert_eq!(
1900 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1901 Some(&json!("Ania"))
1902 );
1903 db.close().unwrap();
1904 }
1905
1906 #[test]
1907 fn open_finishes_an_unfinished_journal_before_returning() {
1908 let dir = tempdir().unwrap();
1909 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1910
1911 {
1912 let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
1913 let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
1914 let user_entries = users
1915 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
1916 .unwrap();
1917 let order_entries = orders
1918 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
1919 .unwrap();
1920
1921 let mut entries = Vec::new();
1922 for entry in &user_entries {
1923 entries.push(JournalEntry {
1924 t: "users".into(),
1925 s: entry.seq(),
1926 l: entry.line().to_string(),
1927 });
1928 }
1929 for entry in &order_entries {
1930 entries.push(JournalEntry {
1931 t: "orders".into(),
1932 s: entry.seq(),
1933 l: entry.line().to_string(),
1934 });
1935 }
1936 write_journal(
1937 &dir.path().join(JOURNAL_FILE),
1938 &Journal {
1939 header: JournalHeader {
1940 journal_version: JOURNAL_VERSION,
1941 kind: "unit".into(),
1942 id: "crash".into(),
1943 started_at: 0,
1944 tables: vec!["users".into(), "orders".into()],
1945 },
1946 entries,
1947 },
1948 )
1949 .unwrap();
1950 }
1952
1953 assert!(dir.path().join(JOURNAL_FILE).exists());
1954 let db = Database::open(dir.path()).unwrap();
1955 assert!(!dir.path().join(JOURNAL_FILE).exists());
1956 assert_eq!(
1957 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1958 Some(&json!("Ania"))
1959 );
1960 assert_eq!(
1961 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
1962 Some(&json!("u1"))
1963 );
1964 db.close().unwrap();
1965 }
1966
1967 #[test]
1968 fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
1969 let dir = tempdir().unwrap();
1970 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1971
1972 let _first = Database::open(dir.path()).unwrap();
1973 let second = Database::open_with(
1974 dir.path(),
1975 OpenOptions {
1976 readonly: false,
1977 lock: LockMode::Enabled(LockOptions {
1978 timeout_ms: Some(0),
1979 ..Default::default()
1980 }),
1981 ..Default::default()
1982 },
1983 );
1984 assert!(matches!(second, Err(DatabaseError::Lock(_))));
1985 }
1986
1987 fn lock_path_of(dir: &Path) -> std::path::PathBuf {
1988 dir.join(crate::lock::LOCK_FILE)
1989 }
1990
1991 fn open_refusing(dir: &Path) -> DatabaseError {
1992 Database::open_with(
1993 dir,
1994 OpenOptions {
1995 lock: LockMode::Enabled(LockOptions {
1996 timeout_ms: Some(0),
1997 ..Default::default()
1998 }),
1999 ..Default::default()
2000 },
2001 )
2002 .map(|_| ())
2003 .expect_err("this directory must not open")
2004 }
2005
2006 #[test]
2007 fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
2008 let dir = tempdir().unwrap();
2009 let mut db = Database::open(dir.path()).unwrap();
2010 db.create_table(
2011 "users",
2012 json!({
2013 "schemaVersion": 1,
2014 "table": "users",
2015 "primaryKey": "id",
2016 "fields": {
2017 "id": { "type": "string" },
2018 "name": { "type": "string", "required": true },
2019 },
2020 }),
2021 )
2022 .unwrap();
2023 db.table_mut("users")
2024 .unwrap()
2025 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2026 .unwrap();
2027 db.close().unwrap();
2028
2029 let log = fs::read(dir.path().join("users.jsonl")).unwrap();
2030 fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2031
2032 let refusal = open_refusing(dir.path()).to_string();
2033 assert!(
2034 refusal.contains("users.jsonl"),
2035 "the refusal has to name the log it found: {refusal}"
2036 );
2037 assert!(
2038 refusal.contains(MANIFEST_FILE),
2039 "the refusal has to name the catalog: {refusal}"
2040 );
2041
2042 assert!(
2044 !dir.path().join(MANIFEST_FILE).exists(),
2045 "a refused open must not leave a catalog"
2046 );
2047 assert_eq!(
2048 fs::read(dir.path().join("users.jsonl")).unwrap(),
2049 log,
2050 "the log must be untouched"
2051 );
2052 }
2053
2054 #[test]
2055 fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
2056 let empty = tempdir().unwrap();
2057 assert!(
2058 Database::open(empty.path()).is_ok(),
2059 "an empty directory is still a new database"
2060 );
2061
2062 let littered = tempdir().unwrap();
2063 fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
2064 fs::create_dir(littered.path().join("users-backup")).unwrap();
2065 assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
2066 }
2067
2068 #[test]
2069 fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
2070 let dir = tempdir().unwrap();
2071 fs::create_dir_all(dir.path().join("notes")).unwrap();
2072 fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
2073
2074 let refusal = open_refusing(dir.path()).to_string();
2075 assert!(
2076 refusal.contains("note.jsonl"),
2077 "a partitioned log still counts as data: {refusal}"
2078 );
2079 }
2080
2081 #[test]
2085 fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
2086 for (case, document) in [
2087 ("not JSON at all", "{ this is not json".to_string()),
2088 (
2089 "no formatVersion",
2090 json!({ "catalogVersion": 2, "tables": {} }).to_string(),
2091 ),
2092 (
2093 "the format-1 marker",
2094 json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
2095 ),
2096 (
2097 "a formatVersion from the future",
2098 json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
2099 ),
2100 ] {
2101 let dir = tempdir().unwrap();
2102 std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
2103
2104 open_refusing(dir.path());
2105
2106 assert!(
2107 !lock_path_of(dir.path()).exists(),
2108 "{case}: the refused open left its lock file behind"
2109 );
2110 }
2111 }
2112
2113 #[test]
2116 fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
2117 let dir = tempdir().unwrap();
2118 std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
2119
2120 open_refusing(dir.path());
2121
2122 std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
2123 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2124
2125 let mut db = Database::open(dir.path()).unwrap();
2126 db.table_mut("users")
2127 .unwrap()
2128 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2129 .unwrap();
2130 db.close().unwrap();
2131 }
2132
2133 #[test]
2136 fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
2137 let dir = tempdir().unwrap();
2138 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2139
2140 let error = Database::open_with(
2141 dir.path(),
2142 OpenOptions {
2143 cypher: Some(CipherOptions {
2144 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
2145 key_id: None,
2146 on_missing_key: None,
2147 }),
2148 ..OpenOptions::default()
2149 },
2150 )
2151 .map(|_| ())
2152 .unwrap_err();
2153
2154 assert!(matches!(error, DatabaseError::Cipher(_)));
2155 assert!(!lock_path_of(dir.path()).exists());
2156 Database::open(dir.path()).unwrap().close().unwrap();
2157 }
2158
2159 #[test]
2162 fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
2163 let dir = tempdir().unwrap();
2164 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2165
2166 let held = Database::open(dir.path()).unwrap();
2167 assert!(lock_path_of(dir.path()).exists());
2168 assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
2169
2170 held.close().unwrap();
2171 assert!(!lock_path_of(dir.path()).exists());
2172 }
2173
2174 fn open_skipping_the_lock(dir: &Path) -> Database {
2177 Database::open_with(
2178 dir,
2179 OpenOptions {
2180 lock: LockMode::Skip,
2181 ..Default::default()
2182 },
2183 )
2184 .unwrap()
2185 }
2186
2187 fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
2188 let catalog = read_catalog(&dir.join(MANIFEST_FILE))
2189 .unwrap()
2190 .expect("a catalog on disk");
2191 let mut names: Vec<String> = catalog.tables[table]
2192 .indexes
2193 .iter()
2194 .flatten()
2195 .map(|index| index.name.clone())
2196 .collect();
2197 names.sort();
2198 names
2199 }
2200
2201 fn two_tables_with_records(dir: &Path) {
2205 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
2206 let mut db = Database::open(dir).unwrap();
2207 db.table_mut("users")
2208 .unwrap()
2209 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2210 .unwrap();
2211 db.table_mut("users")
2212 .unwrap()
2213 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
2214 .unwrap();
2215 db.close().unwrap();
2216 }
2217
2218 #[test]
2219 fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
2220 let dir = tempdir().unwrap();
2221 two_tables_with_records(dir.path());
2222
2223 let mut locked = Database::open(dir.path()).unwrap();
2224 let mut unlocked = open_skipping_the_lock(dir.path());
2225
2226 locked
2228 .table_mut("users")
2229 .unwrap()
2230 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
2231 .unwrap();
2232 unlocked
2233 .table_mut("orders")
2234 .unwrap()
2235 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2236 .unwrap();
2237
2238 let third = Database::open_with(
2240 dir.path(),
2241 OpenOptions {
2242 lock: LockMode::Enabled(LockOptions {
2243 timeout_ms: Some(0),
2244 ..Default::default()
2245 }),
2246 ..Default::default()
2247 },
2248 );
2249 assert!(matches!(third, Err(DatabaseError::Lock(_))));
2250
2251 unlocked.close().unwrap();
2252 locked.close().unwrap();
2253 }
2254
2255 #[test]
2260 fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
2261 let dir = tempdir().unwrap();
2262 two_tables_with_records(dir.path());
2263
2264 let mut locked = Database::open(dir.path()).unwrap();
2265 let mut unlocked = open_skipping_the_lock(dir.path());
2266
2267 locked
2268 .declare_index(
2269 "users",
2270 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2271 )
2272 .unwrap();
2273 unlocked
2274 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2275 .unwrap();
2276
2277 unlocked.close().unwrap();
2278 locked.close().unwrap();
2279
2280 assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
2281 }
2282
2283 #[test]
2284 fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
2285 let dir = tempdir().unwrap();
2286 two_tables_with_records(dir.path());
2287
2288 let mut locked = Database::open(dir.path()).unwrap();
2289 let mut unlocked = open_skipping_the_lock(dir.path());
2290
2291 locked
2292 .declare_index(
2293 "users",
2294 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
2295 )
2296 .unwrap();
2297 assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
2298 unlocked
2299 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
2300 .unwrap();
2301
2302 unlocked.close().unwrap();
2303 locked.close().unwrap();
2304
2305 assert_eq!(
2306 declared_on_disk(dir.path(), "users"),
2307 vec!["by_id".to_string(), "by_name".to_string()]
2308 );
2309 }
2310
2311 fn open_follower(dir: &Path) -> Database {
2314 Database::open_with(
2315 dir,
2316 OpenOptions {
2317 readonly: true,
2318 lock: LockMode::default(),
2319 ..Default::default()
2320 },
2321 )
2322 .unwrap()
2323 }
2324
2325 #[test]
2326 fn refresh_with_nothing_new_on_disk_reports_no_change() {
2327 let dir = tempdir().unwrap();
2328 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2329 let mut writer = Database::open(dir.path()).unwrap();
2330 writer
2331 .table_mut("users")
2332 .unwrap()
2333 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2334 .unwrap();
2335 let mut follower = open_follower(dir.path());
2336 follower.refresh().unwrap();
2337
2338 assert!(!follower.refresh().unwrap());
2339 writer.close().unwrap();
2340 }
2341
2342 #[test]
2343 fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
2344 let dir = tempdir().unwrap();
2345 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2346 let mut writer = Database::open(dir.path()).unwrap();
2347 let mut follower = open_follower(dir.path());
2348 assert_eq!(follower.table_names(), vec!["users"]);
2349
2350 writer
2351 .table_mut("users")
2352 .unwrap()
2353 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2354 .unwrap();
2355 writer
2356 .create_table(
2357 "orders",
2358 json!({
2359 "schemaVersion": 1,
2360 "table": "orders",
2361 "primaryKey": "id",
2362 "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
2363 }),
2364 )
2365 .unwrap();
2366 writer
2367 .table_mut("orders")
2368 .unwrap()
2369 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2370 .unwrap();
2371
2372 assert!(follower.refresh().unwrap());
2373
2374 assert_eq!(follower.table_names(), vec!["orders", "users"]);
2375 assert_eq!(
2376 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2377 Some(&json!("Ania"))
2378 );
2379 assert_eq!(
2380 follower
2381 .table("orders")
2382 .unwrap()
2383 .get("o1")
2384 .unwrap()
2385 .unwrap()
2386 .get("userId"),
2387 Some(&json!("u1"))
2388 );
2389 writer.close().unwrap();
2390 }
2391
2392 #[test]
2393 fn refresh_survives_a_compaction_run_by_the_writer_process() {
2394 let dir = tempdir().unwrap();
2395 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2396 let mut writer = Database::open(dir.path()).unwrap();
2397 let users = writer.table_mut("users").unwrap();
2398 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
2399 users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
2400 users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
2401 let mut follower = open_follower(dir.path());
2402 assert_eq!(follower.table("users").unwrap().len(), 2);
2403
2404 writer.table_mut("users").unwrap().compact().unwrap();
2405 assert!(follower.refresh().unwrap());
2406
2407 assert_eq!(follower.table("users").unwrap().len(), 2);
2408 assert_eq!(
2409 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2410 Some(&json!("Anna"))
2411 );
2412 assert_eq!(follower.table("users").unwrap().stats().gen, 1);
2413 writer.close().unwrap();
2414 }
2415
2416 #[test]
2417 fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
2418 let dir = tempdir().unwrap();
2419 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2420 let mut db = Database::open(dir.path()).unwrap();
2421
2422 let mut unit = db.unit_of_work();
2423 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
2424 unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
2425 let result = db.commit(&mut unit).unwrap();
2426 assert_eq!(result.operations, 2);
2427
2428 assert_eq!(
2429 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2430 Some(&json!("Ania"))
2431 );
2432 assert_eq!(
2433 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2434 Some(&json!("u1"))
2435 );
2436 db.close().unwrap();
2437 }
2438
2439 #[test]
2440 fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
2441 let dir = tempdir().unwrap();
2442 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2443 let mut db = Database::open(dir.path()).unwrap();
2444
2445 db.create_table(
2446 "orders",
2447 json!({
2448 "schemaVersion": 1,
2449 "table": "orders",
2450 "primaryKey": "id",
2451 "fields": {
2452 "id": { "type": "string" },
2453 "userId": { "type": "string", "required": true },
2454 },
2455 }),
2456 )
2457 .unwrap();
2458
2459 assert_eq!(db.table_names(), vec!["orders", "users"]);
2460 assert!(db.catalog.tables.contains_key("orders"));
2461
2462 db.table_mut("orders")
2463 .unwrap()
2464 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2465 .unwrap();
2466 assert_eq!(
2467 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2468 Some(&json!("u1"))
2469 );
2470
2471 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2472 assert!(reread.tables.contains_key("orders"));
2473 db.close().unwrap();
2474 }
2475
2476 #[test]
2477 fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
2478 let dir = tempdir().unwrap();
2479 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2480 let mut db = Database::open(dir.path()).unwrap();
2481 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2482
2483 let error = db
2484 .create_table(
2485 "orders",
2486 json!({
2487 "schemaVersion": 1,
2488 "table": "invoices",
2489 "primaryKey": "id",
2490 "fields": { "id": { "type": "string" } },
2491 }),
2492 )
2493 .expect_err("a schema naming another table is refused");
2494
2495 assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
2496 let message = error.to_string();
2497 assert!(
2498 message.contains("invoices"),
2499 "the refusal names the schema's table: {message}"
2500 );
2501 assert!(
2502 message.contains("orders"),
2503 "the refusal names the key it was created under: {message}"
2504 );
2505
2506 assert_eq!(db.table_names(), vec!["users"]);
2507 assert!(!db.catalog.tables.contains_key("orders"));
2508 assert!(!db.catalog.tables.contains_key("invoices"));
2509 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2510 db.close().unwrap();
2511 }
2512
2513 #[test]
2514 fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
2515 let dir = tempdir().unwrap();
2516 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2517 let mut db = Database::open(dir.path()).unwrap();
2518
2519 db.create_table(
2520 "orders",
2521 json!({
2522 "schemaVersion": 1,
2523 "table": "orders",
2524 "primaryKey": "id",
2525 "fields": { "id": { "type": "string" } },
2526 }),
2527 )
2528 .unwrap();
2529
2530 assert_eq!(db.table("orders").unwrap().schema().table, "orders");
2531 db.close().unwrap();
2532 }
2533
2534 #[test]
2535 fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
2536 let dir = tempdir().unwrap();
2537 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2538 Database::open(dir.path()).unwrap().close().unwrap();
2540 let mut db = Database::open_with(
2541 dir.path(),
2542 OpenOptions {
2543 readonly: true,
2544 lock: LockMode::default(),
2545 ..Default::default()
2546 },
2547 )
2548 .unwrap();
2549
2550 let error = db
2551 .create_table(
2552 "orders",
2553 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2554 )
2555 .unwrap_err();
2556 assert!(
2557 matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
2558 "expected a typed readonly refusal, got {error:?}"
2559 );
2560 assert_eq!(db.table_names(), vec!["users"]);
2561 }
2562
2563 #[test]
2564 fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
2565 let dir = tempdir().unwrap();
2566 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2567 let mut db = Database::open(dir.path()).unwrap();
2568 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2569 let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
2571 fs::create_dir(&temporary).unwrap();
2572
2573 let error = db
2574 .create_table(
2575 "orders",
2576 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2577 )
2578 .unwrap_err();
2579
2580 let refusal = match &error {
2581 DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
2582 .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
2583 other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
2584 };
2585 assert_eq!(refusal.path, temporary);
2586 assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
2587 assert_eq!(db.table_names(), vec!["users"]);
2588 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2589 }
2590
2591 #[test]
2592 fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
2593 let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
2594
2595 assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
2596 assert!(matches!(
2597 catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
2598 DatabaseError::Catalog(_)
2599 ));
2600 }
2601
2602 #[test]
2603 fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
2604 let dir = tempdir().unwrap();
2605 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2606 let mut db = Database::open(dir.path()).unwrap();
2607 db.table_mut("users")
2608 .unwrap()
2609 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2610 .unwrap();
2611 db.table_mut("users").unwrap().delete("u1", false).unwrap();
2612
2613 db.drop_field("users", "name", false).unwrap();
2614
2615 assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
2616 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2617
2618 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2619 assert!(!reread.tables["users"].fields.contains_key("name"));
2620 db.close().unwrap();
2621 }
2622
2623 #[test]
2624 fn drop_field_refuses_a_live_residual_field_without_force() {
2625 let dir = tempdir().unwrap();
2626 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2627 let mut db = Database::open(dir.path()).unwrap();
2628 db.table_mut("users")
2629 .unwrap()
2630 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2631 .unwrap();
2632
2633 let error = db.drop_field("users", "name", false);
2634 assert!(error.is_err());
2635 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2636 assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
2637
2638 db.drop_field("users", "name", true).unwrap();
2639 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2640 db.close().unwrap();
2641 }
2642
2643 fn users_catalog_indexing_name() -> Catalog {
2644 let mut catalog = catalog_of(&[users_schema()]);
2645 catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
2646 name: "by_name".to_string(),
2647 fields: vec!["name".to_string()],
2648 kind: CatalogIndexKind::Exact,
2649 unique: false,
2650 where_: None,
2651 expire_after_seconds: None,
2652 }]);
2653 catalog
2654 }
2655
2656 #[test]
2657 fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
2658 let dir = tempdir().unwrap();
2659 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2660 let mut db = Database::open(dir.path()).unwrap();
2661 db.table_mut("users")
2662 .unwrap()
2663 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2664 .unwrap();
2665
2666 let error = db.drop_field("users", "name", false).unwrap_err();
2667
2668 assert!(
2669 matches!(error, DatabaseError::Schema(_)),
2670 "expected a schema refusal, got {error:?}"
2671 );
2672 let message = error.to_string();
2673 assert!(message.contains("by_name"), "{message}");
2674 assert!(!message.contains("live record(s)"), "{message}");
2675 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2676 db.close().unwrap();
2677 }
2678
2679 #[test]
2680 fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
2681 let dir = tempdir().unwrap();
2682 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2683 let mut db = Database::open(dir.path()).unwrap();
2684
2685 let error = db.drop_field("users", "name", false).unwrap_err();
2686
2687 assert!(
2688 matches!(error, DatabaseError::Schema(_)),
2689 "expected a schema refusal, got {error:?}"
2690 );
2691 assert!(error.to_string().contains("by_name"));
2692 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2693 db.close().unwrap();
2694 }
2695
2696 #[test]
2697 fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
2698 let dir = tempdir().unwrap();
2699 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2700 let mut db = Database::open(dir.path()).unwrap();
2701 db.table_mut("users")
2702 .unwrap()
2703 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2704 .unwrap();
2705
2706 let error = db.drop_field("users", "name", false).unwrap_err();
2707
2708 assert!(
2709 matches!(
2710 &error,
2711 DatabaseError::FieldInUse { table, field, records }
2712 if table == "users" && field == "name" && *records == 1
2713 ),
2714 "expected a typed field-in-use refusal, got {error:?}"
2715 );
2716 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2717 db.close().unwrap();
2718 }
2719
2720 #[test]
2721 fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
2722 let dir = tempdir().unwrap();
2723 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2724 let mut db = Database::open(dir.path()).unwrap();
2725 db.table_mut("users")
2726 .unwrap()
2727 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2728 .unwrap();
2729
2730 let error = db.drop_field("users", "name", true).unwrap_err();
2731
2732 assert!(
2733 matches!(error, DatabaseError::Schema(_)),
2734 "expected a schema refusal, got {error:?}"
2735 );
2736 assert!(error.to_string().contains("by_name"));
2737 db.close().unwrap();
2738 }
2739
2740 #[test]
2741 fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
2742 let dir = tempdir().unwrap();
2743 let schema = parse_schema(&json!({
2744 "schemaVersion": 1,
2745 "table": "users",
2746 "primaryKey": "id",
2747 "fields": {
2748 "id": { "type": "string" },
2749 "email": { "type": "string" },
2750 },
2751 }))
2752 .unwrap();
2753 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2754 let mut db = Database::open(dir.path()).unwrap();
2755 db.table_mut("users")
2756 .unwrap()
2757 .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
2758 .unwrap();
2759 db.table_mut("users")
2760 .unwrap()
2761 .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
2762 .unwrap();
2763
2764 let error = db.declare_index(
2765 "users",
2766 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2767 );
2768 assert!(error.is_err());
2769 assert!(db.catalog.tables["users"].indexes.is_none());
2770
2771 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2772 assert!(reread.tables["users"].indexes.is_none());
2773 db.close().unwrap();
2774 }
2775
2776 fn orders_user_relation(enforce: bool) -> Value {
2777 json!({
2778 "name": "orders_user",
2779 "from": { "table": "orders", "field": "userId" },
2780 "to": { "table": "users", "field": "id" },
2781 "cardinality": "many-to-one",
2782 "enforce": enforce,
2783 })
2784 }
2785
2786 #[test]
2787 fn declare_relation_adds_a_relation_to_the_catalog_and_persists_it() {
2788 let dir = tempdir().unwrap();
2789 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2790 let mut db = Database::open(dir.path()).unwrap();
2791
2792 db.declare_relation(orders_user_relation(false), false).unwrap();
2793
2794 assert_eq!(db.catalog.relations.as_ref().unwrap()[0].name, "orders_user");
2795 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2796 assert_eq!(reread.relations.unwrap()[0].name, "orders_user");
2797 db.close().unwrap();
2798 }
2799
2800 #[test]
2801 fn declare_relation_with_enforce_true_refuses_over_a_live_orphan_and_force_overrides() {
2802 let dir = tempdir().unwrap();
2803 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2804 let mut db = Database::open(dir.path()).unwrap();
2805 db.table_mut("orders")
2806 .unwrap()
2807 .put(&json!({ "id": "o1", "userId": "ghost" }), false)
2808 .unwrap();
2809
2810 let error = db.declare_relation(orders_user_relation(true), false).unwrap_err();
2811 assert!(error.to_string().contains("orders_user"));
2812 assert!(db.catalog.relations.is_none());
2813
2814 db.declare_relation(orders_user_relation(true), true).unwrap();
2815 assert_eq!(db.catalog.relations.as_ref().unwrap()[0].enforce, Some(true));
2816 db.close().unwrap();
2817 }
2818
2819 #[test]
2820 fn put_checked_refuses_a_value_with_no_counterpart_on_an_enforced_relation() {
2821 let dir = tempdir().unwrap();
2822 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2823 let mut db = Database::open(dir.path()).unwrap();
2824 db.declare_relation(orders_user_relation(true), false).unwrap();
2825
2826 let error = db
2827 .put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2828 .unwrap_err();
2829
2830 assert!(error.to_string().contains("orders_user"));
2831 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
2832 db.close().unwrap();
2833 }
2834
2835 #[test]
2836 fn put_checked_allows_a_value_with_a_live_counterpart() {
2837 let dir = tempdir().unwrap();
2838 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2839 let mut db = Database::open(dir.path()).unwrap();
2840 db.declare_relation(orders_user_relation(true), false).unwrap();
2841 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2842 .unwrap();
2843
2844 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2845 .unwrap();
2846
2847 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2848 db.close().unwrap();
2849 }
2850
2851 #[test]
2852 fn put_checked_ignores_a_relation_that_does_not_declare_enforce() {
2853 let dir = tempdir().unwrap();
2854 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2855 let mut db = Database::open(dir.path()).unwrap();
2856 db.declare_relation(orders_user_relation(false), false).unwrap();
2857
2858 db.put_checked("orders", &json!({ "id": "o1", "userId": "ghost" }), false)
2859 .unwrap();
2860
2861 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
2862 db.close().unwrap();
2863 }
2864
2865 #[test]
2866 fn patch_checked_refuses_changing_the_field_to_a_value_with_no_counterpart() {
2867 let dir = tempdir().unwrap();
2868 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2869 let mut db = Database::open(dir.path()).unwrap();
2870 db.declare_relation(orders_user_relation(true), false).unwrap();
2871 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2872 .unwrap();
2873 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2874 .unwrap();
2875
2876 let error = db
2877 .patch_checked("orders", "o1", &json!({ "userId": "ghost" }), false)
2878 .unwrap_err();
2879
2880 assert!(error.to_string().contains("orders_user"));
2881 assert_eq!(
2882 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2883 Some(&json!("u1"))
2884 );
2885 db.close().unwrap();
2886 }
2887
2888 #[test]
2889 fn patch_checked_leaves_the_field_alone_when_changes_does_not_touch_it() {
2890 let dir = tempdir().unwrap();
2891 let orders_named = parse_schema(&json!({
2892 "schemaVersion": 1,
2893 "table": "orders",
2894 "primaryKey": "id",
2895 "fields": {
2896 "id": { "type": "string" },
2897 "userId": { "type": "string", "required": true },
2898 "note": { "type": "string" },
2899 },
2900 }))
2901 .unwrap();
2902 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_named]));
2903 let mut db = Database::open(dir.path()).unwrap();
2904 db.declare_relation(orders_user_relation(true), false).unwrap();
2905 db.put_checked("users", &json!({ "id": "u1", "name": "Ania" }), false)
2906 .unwrap();
2907 db.put_checked("orders", &json!({ "id": "o1", "userId": "u1" }), false)
2908 .unwrap();
2909
2910 db.patch_checked("orders", "o1", &json!({ "note": "gift-wrapped" }), false)
2911 .unwrap();
2912
2913 assert_eq!(
2914 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("note"),
2915 Some(&json!("gift-wrapped"))
2916 );
2917 db.close().unwrap();
2918 }
2919
2920 #[test]
2921 fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
2922 let dir = tempdir().unwrap();
2923 let schema = parse_schema(&json!({
2924 "schemaVersion": 1,
2925 "table": "users",
2926 "primaryKey": "id",
2927 "fields": {
2928 "id": { "type": "string" },
2929 "email": { "type": "string" },
2930 },
2931 }))
2932 .unwrap();
2933 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2934 let mut db = Database::open(dir.path()).unwrap();
2935 db.table_mut("users")
2936 .unwrap()
2937 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2938 .unwrap();
2939 db.table_mut("users")
2940 .unwrap()
2941 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2942 .unwrap();
2943 db.table_mut("users")
2944 .unwrap()
2945 .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
2946 .unwrap();
2947
2948 db.declare_index(
2949 "users",
2950 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2951 )
2952 .unwrap();
2953
2954 assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
2955
2956 let spec = QueryBuilder::new()
2957 .r#where(
2958 "email",
2959 crate::query::predicates::ComparisonOp::Eq,
2960 json!("piotr@example.com"),
2961 )
2962 .spec();
2963 let found = db.table("users").unwrap().query(&spec).unwrap();
2964 assert_eq!(found.keys, vec!["u2".to_string()]);
2965
2966 let conflicting = db
2967 .table_mut("users")
2968 .unwrap()
2969 .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
2970 assert!(conflicting.is_err());
2971 db.close().unwrap();
2972 }
2973
2974 #[test]
2975 fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
2976 let dir = tempdir().unwrap();
2977 let schema = parse_schema(&json!({
2978 "schemaVersion": 1,
2979 "table": "users",
2980 "primaryKey": "id",
2981 "fields": {
2982 "id": { "type": "string" },
2983 "email": { "type": "string" },
2984 },
2985 }))
2986 .unwrap();
2987 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2988 let mut db = Database::open(dir.path()).unwrap();
2989 db.table_mut("users")
2990 .unwrap()
2991 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2992 .unwrap();
2993
2994 db.declare_index(
2995 "users",
2996 json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
2997 )
2998 .unwrap();
2999
3000 assert!(
3001 index_path(dir.path(), "users").exists(),
3002 "declare_index owes the table a snapshot carrying the new index's projection"
3003 );
3004
3005 db.table_mut("users")
3006 .unwrap()
3007 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
3008 .unwrap();
3009 assert!(
3010 db.table("users").unwrap().needs_snapshot(),
3011 "the handle declare_index left behind must still be able to snapshot itself"
3012 );
3013 db.close().unwrap();
3014
3015 let reopened = Database::open(dir.path()).unwrap();
3016 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3017 let spec = QueryBuilder::new()
3018 .r#where(
3019 "email",
3020 crate::query::predicates::ComparisonOp::Eq,
3021 json!("piotr@example.com"),
3022 )
3023 .spec();
3024 assert_eq!(
3025 reopened.table("users").unwrap().query(&spec).unwrap().keys,
3026 vec!["u2".to_string()]
3027 );
3028 assert_eq!(
3029 declared_on_disk(dir.path(), "users"),
3030 vec!["users_email_exact".to_string()]
3031 );
3032 reopened.close().unwrap();
3033 }
3034
3035 #[test]
3036 fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
3037 let dir = tempdir().unwrap();
3038 let mut catalog = catalog_of(&[users_schema()]);
3039 catalog.defaults = Some(crate::catalog::TableOptions {
3040 sync: Some(crate::catalog::CatalogSyncMode::Never),
3041 ..Default::default()
3042 });
3043 write_catalog_to(dir.path(), &catalog);
3044
3045 let db = Database::open(dir.path()).unwrap();
3046 assert_eq!(
3047 db.table("users").unwrap().sync_mode(),
3048 crate::catalog::CatalogSyncMode::Never
3049 );
3050 db.close().unwrap();
3051 }
3052
3053 #[test]
3054 fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
3055 let dir = tempdir().unwrap();
3056 let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
3057 catalog.defaults = Some(crate::catalog::TableOptions {
3058 sync: Some(crate::catalog::CatalogSyncMode::Never),
3059 ..Default::default()
3060 });
3061 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3062 sync: Some(crate::catalog::CatalogSyncMode::Commit),
3063 ..Default::default()
3064 });
3065 write_catalog_to(dir.path(), &catalog);
3066
3067 let db = Database::open(dir.path()).unwrap();
3068 assert_eq!(
3069 db.table("users").unwrap().sync_mode(),
3070 crate::catalog::CatalogSyncMode::Commit
3071 );
3072 assert_eq!(
3073 db.table("orders").unwrap().sync_mode(),
3074 crate::catalog::CatalogSyncMode::Never
3075 );
3076 db.close().unwrap();
3077 }
3078
3079 #[test]
3080 fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
3081 let dir = tempdir().unwrap();
3082 let mut catalog = catalog_of(&[users_schema()]);
3083 catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
3084 write_catalog_to(dir.path(), &catalog);
3085
3086 let mut db = Database::open(dir.path()).unwrap();
3087 let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
3088 let table = db.table_mut("users").unwrap();
3089 let record = table.put(&document, false).unwrap();
3090
3091 assert_eq!(record.get("nickname"), None);
3092 assert_eq!(
3093 table.dropped_field_warnings(&document),
3094 vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
3095 );
3096 db.close().unwrap();
3097
3098 let reopened = Database::open(dir.path()).unwrap();
3099 let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
3100 assert_eq!(stored.get("name"), Some(&json!("Ania")));
3101 assert_eq!(stored.get("nickname"), None);
3102 reopened.close().unwrap();
3103 }
3104
3105 #[test]
3106 fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
3107 let dir = tempdir().unwrap();
3108 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3109
3110 let mut db = Database::open(dir.path()).unwrap();
3111 assert_eq!(
3112 db.table("users").unwrap().validation(),
3113 crate::catalog::CatalogValidation::Strict
3114 );
3115 let error = db
3116 .table_mut("users")
3117 .unwrap()
3118 .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
3119 .unwrap_err();
3120
3121 assert!(error.to_string().contains("is not declared in the schema"), "{error}");
3122 db.close().unwrap();
3123 }
3124
3125 fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
3126 let mut catalog = catalog_of(&[users_with_age_schema()]);
3127 let entry = catalog.tables.get_mut("users").unwrap();
3128 entry.validation = validation;
3129 entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
3130 write_catalog_to(dir, &catalog);
3131 }
3132
3133 fn users_with_age_schema() -> TableSchema {
3134 parse_schema(&json!({
3135 "schemaVersion": 1,
3136 "table": "users",
3137 "primaryKey": "id",
3138 "fields": {
3139 "id": { "type": "string" },
3140 "name": { "type": "string", "required": true },
3141 "age": { "type": "int" },
3142 },
3143 }))
3144 .unwrap()
3145 }
3146
3147 #[test]
3148 fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
3149 let dir = tempdir().unwrap();
3150 deprecating_age(dir.path(), None, "use \"bornAt\" instead");
3151
3152 let mut db = Database::open(dir.path()).unwrap();
3153 let table = db.table_mut("users").unwrap();
3154 let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
3155 let record = table.put(&touching, false).unwrap();
3156
3157 assert_eq!(record.get("age"), Some(&json!(30)));
3158 assert_eq!(
3159 table.deprecated_field_warnings(&touching),
3160 vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
3161 );
3162 assert!(table
3163 .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
3164 .is_empty());
3165 db.close().unwrap();
3166 }
3167
3168 #[test]
3169 fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
3170 let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
3171 let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
3172
3173 for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
3174 let dir = tempdir().unwrap();
3175 deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
3176
3177 let mut db = Database::open(dir.path()).unwrap();
3178 let table = db.table_mut("users").unwrap();
3179 table.put(&document, false).unwrap();
3180 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3181 table.put(&document, false).unwrap();
3182 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
3183 db.close().unwrap();
3184
3185 let reopened = Database::open(dir.path()).unwrap();
3186 assert_eq!(
3187 reopened.table("users").unwrap().deprecated_field_warnings(&document),
3188 expected
3189 );
3190 reopened.close().unwrap();
3191 }
3192 }
3193
3194 #[test]
3195 fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
3196 let dir = tempdir().unwrap();
3197 write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
3198
3199 let db = Database::open(dir.path()).unwrap();
3200 let table = db.table("users").unwrap();
3201
3202 assert!(table.deprecated_fields().is_empty());
3203 assert!(table
3204 .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
3205 .is_empty());
3206 db.close().unwrap();
3207 }
3208
3209 #[test]
3210 fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
3211 let dir = tempdir().unwrap();
3212 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3213
3214 let db = Database::open(dir.path()).unwrap();
3215 assert_eq!(
3216 db.table("users").unwrap().sync_mode(),
3217 crate::catalog::CatalogSyncMode::Commit
3218 );
3219 db.close().unwrap();
3220 }
3221
3222 #[test]
3223 fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
3224 let dir = tempdir().unwrap();
3225 let mut catalog = catalog_of(&[users_schema()]);
3226 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
3227 sync: Some(crate::catalog::CatalogSyncMode::Never),
3228 ..Default::default()
3229 });
3230 write_catalog_to(dir.path(), &catalog);
3231
3232 {
3233 let mut db = Database::open(dir.path()).unwrap();
3234 db.table_mut("users")
3235 .unwrap()
3236 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3237 .unwrap();
3238 db.close().unwrap();
3239 }
3240
3241 let db = Database::open(dir.path()).unwrap();
3242 assert_eq!(
3243 db.table("users").unwrap().sync_mode(),
3244 crate::catalog::CatalogSyncMode::Never
3245 );
3246 assert_eq!(
3247 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3248 Some(&json!("Ania"))
3249 );
3250 db.close().unwrap();
3251 }
3252
3253 #[test]
3254 fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
3255 let dir = tempdir().unwrap();
3256 let pages_schema = parse_schema(&json!({
3257 "schemaVersion": 1,
3258 "table": "pages",
3259 "primaryKey": "id",
3260 "fields": {
3261 "id": { "type": "string" },
3262 "page": { "type": "markdown" },
3263 },
3264 }))
3265 .unwrap();
3266 let mut catalog = catalog_of(&[pages_schema]);
3267 catalog.defaults = Some(crate::catalog::TableOptions {
3268 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3269 ..Default::default()
3270 });
3271 write_catalog_to(dir.path(), &catalog);
3272
3273 let db = Database::open(dir.path()).unwrap();
3274 assert_eq!(
3275 db.table("pages").unwrap().link_parsing(),
3276 crate::catalog::LinkParsing::Lazy
3277 );
3278 db.close().unwrap();
3279 }
3280
3281 #[test]
3282 fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
3283 let dir = tempdir().unwrap();
3284 let pages_schema = parse_schema(&json!({
3285 "schemaVersion": 1,
3286 "table": "pages",
3287 "primaryKey": "id",
3288 "fields": {
3289 "id": { "type": "string" },
3290 "page": { "type": "markdown" },
3291 },
3292 }))
3293 .unwrap();
3294 let mut catalog = catalog_of(&[pages_schema]);
3295 catalog.defaults = Some(crate::catalog::TableOptions {
3296 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3297 ..Default::default()
3298 });
3299 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3300 link_parsing: Some(crate::catalog::LinkParsing::Eager),
3301 ..Default::default()
3302 });
3303 write_catalog_to(dir.path(), &catalog);
3304
3305 let db = Database::open(dir.path()).unwrap();
3306 assert_eq!(
3307 db.table("pages").unwrap().link_parsing(),
3308 crate::catalog::LinkParsing::Eager
3309 );
3310 db.close().unwrap();
3311 }
3312
3313 #[test]
3314 fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
3315 let dir = tempdir().unwrap();
3316 let docs_schema = parse_schema(&json!({
3317 "schemaVersion": 1,
3318 "table": "docs",
3319 "primaryKey": "id",
3320 "fields": {
3321 "id": { "type": "string" },
3322 "cover": { "type": "image" },
3323 },
3324 }))
3325 .unwrap();
3326 let mut catalog = catalog_of(&[docs_schema]);
3327 catalog.defaults = Some(crate::catalog::TableOptions {
3328 image_dir: Some("shared-images".to_string()),
3329 ..Default::default()
3330 });
3331 write_catalog_to(dir.path(), &catalog);
3332
3333 let mut db = Database::open(dir.path()).unwrap();
3334 let record = db
3335 .table_mut("docs")
3336 .unwrap()
3337 .put(
3338 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3339 false,
3340 )
3341 .unwrap();
3342 let physical_path = record
3343 .get("cover")
3344 .unwrap()
3345 .get("physicalPath")
3346 .unwrap()
3347 .as_str()
3348 .unwrap();
3349
3350 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3351 assert!(dir.path().join("shared-images").join(physical_path).exists());
3352 db.close().unwrap();
3353 }
3354
3355 #[test]
3356 fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
3357 let dir = tempdir().unwrap();
3358 let schema = parse_schema(&json!({
3359 "schemaVersion": 1,
3360 "table": "orders",
3361 "primaryKey": "id",
3362 "fields": {
3363 "id": { "type": "string" },
3364 "userId": { "type": "string" },
3365 "status": { "type": "string" },
3366 },
3367 }))
3368 .unwrap();
3369 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3370 let mut db = Database::open(dir.path()).unwrap();
3371 db.table_mut("orders")
3372 .unwrap()
3373 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3374 .unwrap();
3375 db.table_mut("orders")
3376 .unwrap()
3377 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3378 .unwrap();
3379 db.table_mut("orders")
3380 .unwrap()
3381 .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
3382 .unwrap();
3383
3384 db.declare_index(
3385 "orders",
3386 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3387 )
3388 .unwrap();
3389
3390 let spec = QueryBuilder::new()
3391 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3392 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
3393 .spec();
3394 let found = db.table("orders").unwrap().query(&spec).unwrap();
3395 assert_eq!(found.keys, vec!["o1".to_string()]);
3396 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3397 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3398 assert_eq!(found.plan.candidates, 1);
3399 db.close().unwrap();
3400 }
3401
3402 #[test]
3403 fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
3404 let dir = tempdir().unwrap();
3405 let schema = parse_schema(&json!({
3406 "schemaVersion": 1,
3407 "table": "users",
3408 "primaryKey": "id",
3409 "fields": {
3410 "id": { "type": "string" },
3411 "role": { "type": "string" },
3412 "status": { "type": "string" },
3413 },
3414 }))
3415 .unwrap();
3416 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3417 let mut db = Database::open(dir.path()).unwrap();
3418 db.table_mut("users")
3419 .unwrap()
3420 .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
3421 .unwrap();
3422 db.table_mut("users")
3423 .unwrap()
3424 .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
3425 .unwrap();
3426
3427 db.declare_index(
3428 "users",
3429 json!({
3430 "name": "users_role_active",
3431 "fields": ["role"],
3432 "kind": "exact",
3433 "where": [{ "field": "status", "op": "=", "value": "active" }],
3434 }),
3435 )
3436 .unwrap();
3437
3438 let eq = crate::query::predicates::ComparisonOp::Eq;
3439 let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
3440 let found = db.table("users").unwrap().query(&omitted).unwrap();
3441 assert_eq!(
3442 found.keys,
3443 vec!["u1".to_string(), "u2".to_string()],
3444 "the index holds u1 alone, but the query never asked for active records only"
3445 );
3446 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
3447
3448 let repeated = QueryBuilder::new()
3449 .r#where("role", eq, json!("admin"))
3450 .r#where("status", eq, json!("active"))
3451 .spec();
3452 let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
3453 assert_eq!(narrowed.keys, vec!["u1".to_string()]);
3454 assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3455 assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
3456 assert_eq!(narrowed.plan.candidates, 1);
3457 db.close().unwrap();
3458 }
3459
3460 #[test]
3461 fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
3462 let dir = tempdir().unwrap();
3463 let schema = parse_schema(&json!({
3464 "schemaVersion": 1,
3465 "table": "orders",
3466 "primaryKey": "id",
3467 "fields": {
3468 "id": { "type": "string" },
3469 "userId": { "type": "string" },
3470 "status": { "type": "string" },
3471 },
3472 }))
3473 .unwrap();
3474 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3475 let mut db = Database::open(dir.path()).unwrap();
3476 db.table_mut("orders")
3477 .unwrap()
3478 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
3479 .unwrap();
3480 db.table_mut("orders")
3481 .unwrap()
3482 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
3483 .unwrap();
3484 db.declare_index(
3485 "orders",
3486 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
3487 )
3488 .unwrap();
3489 db.close().unwrap();
3490
3491 let reopened = Database::open(dir.path()).unwrap();
3492 let spec = QueryBuilder::new()
3493 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
3494 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
3495 .spec();
3496 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3497 assert_eq!(found.keys, vec!["o2".to_string()]);
3498 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
3499 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
3500 reopened.close().unwrap();
3501 }
3502
3503 fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
3504 let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
3505 let entry = catalog.tables.get_mut(table).unwrap();
3506 entry.indexes = Some(vec![index]);
3507 write_catalog_to(dir, &catalog);
3508 }
3509
3510 fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
3511 IndexEntry {
3512 name: name.to_string(),
3513 fields: fields.iter().map(|field| field.to_string()).collect(),
3514 kind: CatalogIndexKind::Sorted,
3515 unique: false,
3516 where_: None,
3517 expire_after_seconds: None,
3518 }
3519 }
3520
3521 #[test]
3526 fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
3527 let dir = tempdir().unwrap();
3528 let schema = parse_schema(&json!({
3529 "schemaVersion": 1,
3530 "table": "orders",
3531 "primaryKey": "id",
3532 "fields": {
3533 "id": { "type": "string" },
3534 "userId": { "type": "string" },
3535 "total": { "type": "int" },
3536 },
3537 }))
3538 .unwrap();
3539 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3540 let mut db = Database::open(dir.path()).unwrap();
3541 let orders = db.table_mut("orders").unwrap();
3542 orders
3543 .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
3544 .unwrap();
3545 orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
3546 orders
3547 .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
3548 .unwrap();
3549 db.close().unwrap();
3550 assert!(
3551 index_path(dir.path(), "orders").exists(),
3552 "the fixture needs a snapshot to restore from"
3553 );
3554
3555 add_index_to_catalog_on_disk(
3556 dir.path(),
3557 "orders",
3558 sorted_index_entry("orders_user_total", &["userId", "total"]),
3559 );
3560
3561 let reopened = Database::open(dir.path()).unwrap();
3562 assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
3563
3564 let spec = QueryBuilder::new()
3565 .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
3566 .spec();
3567 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3568 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
3569 assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
3570 let mut keys = found.keys.clone();
3571 keys.sort();
3572 assert_eq!(
3573 keys,
3574 vec!["a".to_string(), "b".to_string()],
3575 "b has no total, but it still has a userId"
3576 );
3577 reopened.close().unwrap();
3578 }
3579
3580 #[test]
3583 fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
3584 let dir = tempdir().unwrap();
3585 let schema = parse_schema(&json!({
3586 "schemaVersion": 1,
3587 "table": "tickets",
3588 "primaryKey": "id",
3589 "partitionBy": "kind",
3590 "fields": {
3591 "id": { "type": "string" },
3592 "kind": { "type": "string", "required": true },
3593 "userId": { "type": "string", "required": true },
3594 "orderId": { "type": "int" },
3595 },
3596 }))
3597 .unwrap();
3598 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3599 let mut db = Database::open(dir.path()).unwrap();
3600 let tickets = db.table_mut("tickets").unwrap();
3601 tickets
3602 .put(
3603 &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
3604 false,
3605 )
3606 .unwrap();
3607 tickets
3608 .put(
3609 &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
3610 false,
3611 )
3612 .unwrap();
3613 tickets
3614 .put(
3615 &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
3616 false,
3617 )
3618 .unwrap();
3619 db.close().unwrap();
3620
3621 add_index_to_catalog_on_disk(
3622 dir.path(),
3623 "tickets",
3624 IndexEntry {
3625 name: "tickets_user_order".to_string(),
3626 fields: vec!["userId".to_string(), "orderId".to_string()],
3627 kind: CatalogIndexKind::Exact,
3628 unique: true,
3629 where_: None,
3630 expire_after_seconds: None,
3631 },
3632 );
3633
3634 let mut reopened = Database::open(dir.path()).unwrap();
3635 let tickets = reopened.table_mut("tickets").unwrap();
3636 let clash = tickets.put(
3638 &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
3639 false,
3640 );
3641 match clash {
3642 Err(TableError::Unique(violation)) => {
3643 assert_eq!(violation.field, "userId, orderId");
3644 assert_eq!(violation.holder, "t2");
3645 }
3646 other => panic!("a repeated composite key must be refused, got {other:?}"),
3647 }
3648 tickets
3649 .put(
3650 &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
3651 false,
3652 )
3653 .unwrap();
3654 reopened.close().unwrap();
3655 }
3656
3657 #[test]
3658 fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
3659 let dir = tempdir().unwrap();
3660 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3661 let mut db = Database::open(dir.path()).unwrap();
3662
3663 db.table_mut("users")
3665 .unwrap()
3666 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3667 .unwrap();
3668 db.table_mut("users")
3669 .unwrap()
3670 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3671 .unwrap();
3672 assert!(!index_path(dir.path(), "users").exists());
3673
3674 db.close().unwrap();
3675 assert!(index_path(dir.path(), "users").exists());
3676
3677 let reopened = Database::open(dir.path()).unwrap();
3678 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3679 assert_eq!(
3680 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3681 Some(&json!("Ania"))
3682 );
3683 assert_eq!(
3684 reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3685 Some(&json!("Piotr"))
3686 );
3687 reopened.close().unwrap();
3688 }
3689
3690 #[test]
3691 fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
3692 let dir = tempdir().unwrap();
3693 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3694 let mut db = Database::open(dir.path()).unwrap();
3695 db.table_mut("users")
3696 .unwrap()
3697 .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
3698 .unwrap();
3699 assert_eq!(
3700 db.table("users").unwrap().len(),
3701 0,
3702 "nothing is committed before the close"
3703 );
3704
3705 db.close().unwrap();
3706
3707 let reopened = Database::open(dir.path()).unwrap();
3708 assert_eq!(
3709 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3710 Some(&json!("Ania"))
3711 );
3712 reopened.close().unwrap();
3713 }
3714
3715 #[cfg(unix)]
3718 fn swap_file_in_place(path: &Path) {
3719 let swapped = path.with_extension("swap");
3720 std::fs::copy(path, &swapped).unwrap();
3721 std::fs::rename(&swapped, path).unwrap();
3722 }
3723
3724 fn log_replaced_of(error: &io::Error) -> Option<&crate::errors::HaysonLogReplacedError> {
3725 error.get_ref()?.downcast_ref::<crate::errors::HaysonLogReplacedError>()
3726 }
3727
3728 #[cfg(unix)]
3732 #[test]
3733 fn a_close_refused_by_a_swapped_log_still_releases_the_directory_lock() {
3734 let dir = tempdir().unwrap();
3735 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3736 let mut db = Database::open(dir.path()).unwrap();
3737 db.table_mut("users")
3738 .unwrap()
3739 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3740 .unwrap();
3741 db.table_mut("users")
3742 .unwrap()
3743 .stage_put(&json!({ "id": "u2", "name": "Piotr" }), false)
3744 .unwrap();
3745 swap_file_in_place(&log_path(dir.path(), "users"));
3746
3747 let error = db.close().unwrap_err();
3748
3749 assert!(log_replaced_of(&error).is_some(), "{error}");
3750 assert!(!lock_path_of(dir.path()).exists(), "the lock file outlived the handle");
3751 let started = std::time::Instant::now();
3752 let reopened = Database::open(dir.path()).unwrap();
3753 assert!(
3754 started.elapsed() < Duration::from_secs(1),
3755 "the reopen waited {:?} on a lock nobody holds",
3756 started.elapsed()
3757 );
3758 assert_eq!(
3759 reopened.table("users").unwrap().get("u2").unwrap(),
3760 None,
3761 "the staged write never reached the log"
3762 );
3763 reopened.close().unwrap();
3764 }
3765
3766 #[cfg(unix)]
3769 fn swap_older_version_in(path: &Path) {
3770 let current = std::fs::read_to_string(path).unwrap();
3771 let older = format!("{}\n", current.lines().next().unwrap());
3772 let swapped = path.with_extension("swap");
3773 std::fs::write(&swapped, older).unwrap();
3774 std::fs::rename(&swapped, path).unwrap();
3775 }
3776
3777 #[cfg(unix)]
3781 #[test]
3782 fn a_close_after_a_swapped_log_leaves_no_snapshot_so_the_reopen_reads_the_new_file() {
3783 let dir = tempdir().unwrap();
3784 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3785 let mut db = Database::open(dir.path()).unwrap();
3786 db.table_mut("users")
3787 .unwrap()
3788 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3789 .unwrap();
3790 db.table_mut("users")
3791 .unwrap()
3792 .put(&json!({ "id": "u1", "name": "Anna" }), false)
3793 .unwrap();
3794 swap_older_version_in(&log_path(dir.path(), "users"));
3795
3796 let error = db.close().unwrap_err();
3797
3798 assert!(log_replaced_of(&error).is_some(), "{error}");
3799 assert!(
3800 !index_path(dir.path(), "users").exists(),
3801 "a snapshot describing the orphaned inode was written"
3802 );
3803 let reopened = Database::open(dir.path()).unwrap();
3804 assert_eq!(
3805 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3806 Some(&json!("Ania")),
3807 "the reopen reads what the checkout put at the path"
3808 );
3809 reopened.close().unwrap();
3810 }
3811
3812 #[cfg(unix)]
3815 #[test]
3816 fn refresh_after_a_swapped_log_takes_the_new_file_and_writes_land_in_it_again() {
3817 let dir = tempdir().unwrap();
3818 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3819 let mut db = Database::open(dir.path()).unwrap();
3820 db.table_mut("users")
3821 .unwrap()
3822 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3823 .unwrap();
3824 db.table_mut("users")
3825 .unwrap()
3826 .put(&json!({ "id": "u1", "name": "Anna" }), false)
3827 .unwrap();
3828 let log = log_path(dir.path(), "users");
3829 swap_older_version_in(&log);
3830 assert!(db
3831 .table_mut("users")
3832 .unwrap()
3833 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3834 .is_err());
3835
3836 assert!(db.refresh().unwrap(), "the swap counts as a change");
3837
3838 let users = db.table_mut("users").unwrap();
3839 assert_eq!(users.get("u1").unwrap().unwrap().get("name"), Some(&json!("Ania")));
3840 users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
3841 assert_eq!(std::fs::read_to_string(&log).unwrap().lines().count(), 2);
3842 db.close().unwrap();
3843
3844 let reopened = Database::open(dir.path()).unwrap();
3845 assert_eq!(
3846 reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3847 Some(&json!("Piotr"))
3848 );
3849 reopened.close().unwrap();
3850 }
3851
3852 #[test]
3853 fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
3854 let dir = tempdir().unwrap();
3855 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3856 let mut db = Database::open_with(
3857 dir.path(),
3858 OpenOptions {
3859 snapshot_every: 2,
3860 ..Default::default()
3861 },
3862 )
3863 .unwrap();
3864
3865 db.table_mut("users")
3866 .unwrap()
3867 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3868 .unwrap();
3869 assert!(!index_path(dir.path(), "users").exists());
3870 db.table_mut("users")
3871 .unwrap()
3872 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3873 .unwrap();
3874 assert!(index_path(dir.path(), "users").exists());
3875
3876 db.close().unwrap();
3877 }
3878
3879 #[test]
3880 fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
3881 let dir = tempdir().unwrap();
3882 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3883 let mut db = Database::open_with(
3884 dir.path(),
3885 OpenOptions {
3886 snapshot_every: 2,
3887 ..Default::default()
3888 },
3889 )
3890 .unwrap();
3891
3892 db.table_mut("users")
3893 .unwrap()
3894 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3895 .unwrap();
3896 db.table_mut("users")
3897 .unwrap()
3898 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3899 .unwrap();
3900 db.table_mut("users")
3901 .unwrap()
3902 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
3903 .unwrap();
3904 assert_eq!(
3905 db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
3906 Some(&json!("Kasia"))
3907 );
3908 db.close().unwrap();
3909 }
3910
3911 fn users_insert_audits_into_orders() -> Value {
3912 json!({
3913 "name": "users_insert_audit",
3914 "table": "users",
3915 "event": "insert",
3916 "effect": {
3917 "table": "orders",
3918 "fields": [
3919 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3920 { "target": "userId", "source": { "kind": "new", "field": "id" } },
3921 ],
3922 },
3923 })
3924 }
3925
3926 #[test]
3927 fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
3928 let dir = tempdir().unwrap();
3929 let docs_schema = parse_schema(&json!({
3930 "schemaVersion": 1,
3931 "table": "docs",
3932 "primaryKey": "id",
3933 "fields": {
3934 "id": { "type": "string" },
3935 "cover": { "type": "image" },
3936 },
3937 }))
3938 .unwrap();
3939 let mut catalog = catalog_of(&[docs_schema]);
3940 catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
3941 image_dir: Some("shared-images".to_string()),
3942 ..Default::default()
3943 });
3944 write_catalog_to(dir.path(), &catalog);
3945
3946 let mut db = Database::open(dir.path()).unwrap();
3947 let record = db
3948 .table_mut("docs")
3949 .unwrap()
3950 .put(
3951 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3952 false,
3953 )
3954 .unwrap();
3955 let physical_path = record
3956 .get("cover")
3957 .unwrap()
3958 .get("physicalPath")
3959 .unwrap()
3960 .as_str()
3961 .unwrap();
3962
3963 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3964 assert!(dir.path().join("shared-images").join(physical_path).exists());
3965 db.close().unwrap();
3966 }
3967
3968 #[test]
3969 fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
3970 let dir = tempdir().unwrap();
3971 let pages_schema = parse_schema(&json!({
3972 "schemaVersion": 1,
3973 "table": "pages",
3974 "primaryKey": "id",
3975 "fields": {
3976 "id": { "type": "string" },
3977 "page": { "type": "markdown" },
3978 },
3979 }))
3980 .unwrap();
3981 let mut catalog = catalog_of(&[pages_schema]);
3982 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3983 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3984 ..Default::default()
3985 });
3986 write_catalog_to(dir.path(), &catalog);
3987
3988 let mut db = Database::open(dir.path()).unwrap();
3989 assert_eq!(
3990 db.table("pages").unwrap().link_parsing(),
3991 crate::catalog::LinkParsing::Lazy
3992 );
3993
3994 let outcome = db
3995 .table_mut("pages")
3996 .unwrap()
3997 .put_with_warnings(
3998 &json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
3999 false,
4000 )
4001 .unwrap();
4002 assert!(outcome.warnings.is_empty());
4003 db.close().unwrap();
4004 }
4005
4006 #[test]
4007 fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
4008 let dir = tempdir().unwrap();
4009 let pages_schema = parse_schema(&json!({
4010 "schemaVersion": 1,
4011 "table": "pages",
4012 "primaryKey": "id",
4013 "fields": {
4014 "id": { "type": "string" },
4015 "page": { "type": "markdown" },
4016 },
4017 }))
4018 .unwrap();
4019 write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
4020
4021 let db = Database::open(dir.path()).unwrap();
4022 assert_eq!(
4023 db.table("pages").unwrap().link_parsing(),
4024 crate::catalog::LinkParsing::Eager
4025 );
4026 }
4027
4028 #[test]
4029 fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
4030 let dir = tempdir().unwrap();
4031 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4032 let mut db = Database::open(dir.path()).unwrap();
4033
4034 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4035
4036 assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
4037 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4038 assert!(db.table("orders").unwrap().triggers().is_empty());
4039
4040 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4041 assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
4042 assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
4043
4044 db.close().unwrap();
4045 }
4046
4047 #[test]
4048 fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
4049 let dir = tempdir().unwrap();
4050 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4051 let mut db = Database::open(dir.path()).unwrap();
4052 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4053
4054 let mut unit = db.unit_of_work();
4055 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4056 db.commit(&mut unit).unwrap();
4057
4058 assert_eq!(
4059 db.table("orders")
4060 .unwrap()
4061 .get("o-audit")
4062 .unwrap()
4063 .unwrap()
4064 .get("userId"),
4065 Some(&json!("u1"))
4066 );
4067 db.close().unwrap();
4068 }
4069
4070 #[test]
4071 fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
4072 let dir = tempdir().unwrap();
4073 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4074 let mut db = Database::open(dir.path()).unwrap();
4075 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4076 db.close().unwrap();
4077
4078 let mut reopened = Database::open(dir.path()).unwrap();
4079 let mut unit = reopened.unit_of_work();
4080 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4081 reopened.commit(&mut unit).unwrap();
4082
4083 assert_eq!(
4084 reopened
4085 .table("orders")
4086 .unwrap()
4087 .get("o-audit")
4088 .unwrap()
4089 .unwrap()
4090 .get("userId"),
4091 Some(&json!("u1"))
4092 );
4093 reopened.close().unwrap();
4094 }
4095
4096 #[test]
4097 fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
4098 let dir = tempdir().unwrap();
4099 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4100 let mut db = Database::open(dir.path()).unwrap();
4101 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4102
4103 db.declare_index(
4104 "users",
4105 json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
4106 )
4107 .unwrap();
4108
4109 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4110
4111 let mut unit = db.unit_of_work();
4112 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4113 db.commit(&mut unit).unwrap();
4114
4115 assert_eq!(
4116 db.table("orders")
4117 .unwrap()
4118 .get("o-audit")
4119 .unwrap()
4120 .unwrap()
4121 .get("userId"),
4122 Some(&json!("u1"))
4123 );
4124 db.close().unwrap();
4125 }
4126
4127 #[test]
4128 fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
4129 let dir = tempdir().unwrap();
4130 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4131 let mut db = Database::open(dir.path()).unwrap();
4132 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4133
4134 db.drop_field("users", "name", false).unwrap();
4135
4136 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4137
4138 let mut unit = db.unit_of_work();
4139 unit.put("users", json!({ "id": "u1" }));
4140 db.commit(&mut unit).unwrap();
4141
4142 assert_eq!(
4143 db.table("orders")
4144 .unwrap()
4145 .get("o-audit")
4146 .unwrap()
4147 .unwrap()
4148 .get("userId"),
4149 Some(&json!("u1"))
4150 );
4151 db.close().unwrap();
4152 }
4153
4154 #[test]
4155 fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
4156 let dir = tempdir().unwrap();
4157 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
4158 let mut db = Database::open(dir.path()).unwrap();
4159 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
4160
4161 db.create_table(
4162 "users",
4163 json!({
4164 "schemaVersion": 1,
4165 "table": "users",
4166 "revision": 2,
4167 "primaryKey": "id",
4168 "fields": {
4169 "id": { "type": "string" },
4170 "name": { "type": "string", "required": true },
4171 "city": { "type": "string" },
4172 },
4173 }),
4174 )
4175 .unwrap();
4176
4177 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
4178
4179 let mut unit = db.unit_of_work();
4180 unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
4181 db.commit(&mut unit).unwrap();
4182
4183 assert_eq!(
4184 db.table("orders")
4185 .unwrap()
4186 .get("o-audit")
4187 .unwrap()
4188 .unwrap()
4189 .get("userId"),
4190 Some(&json!("u1"))
4191 );
4192 db.close().unwrap();
4193 }
4194
4195 #[test]
4196 fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
4197 let dir = tempdir().unwrap();
4198 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4199 let mut db = Database::open(dir.path()).unwrap();
4200
4201 let spec = json!({
4202 "name": "users_insert_audit",
4203 "table": "users",
4204 "event": "insert",
4205 "effect": {
4206 "table": "ghost",
4207 "fields": [
4208 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
4209 ],
4210 },
4211 });
4212 let error = db.declare_trigger(spec);
4213 assert!(error.is_err());
4214 assert!(db.catalog.triggers.is_none());
4215
4216 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
4217 assert!(reread.triggers.is_none());
4218 db.close().unwrap();
4219 }
4220
4221 mod automatic_compaction {
4222 use super::*;
4223 use crate::catalog::TableOptions;
4224
4225 fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
4226 CatalogCompaction::Enabled {
4227 dead_bytes_ratio,
4228 min_dead_bytes,
4229 }
4230 }
4231
4232 fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
4233 Database::open_with(
4234 dir,
4235 OpenOptions {
4236 compaction,
4237 ..Default::default()
4238 },
4239 )
4240 .unwrap()
4241 }
4242
4243 fn lines_of(path: &Path) -> Vec<String> {
4244 fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
4245 }
4246
4247 fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
4250 let users = db.table_mut(table).unwrap();
4251 users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
4252 for round in 0..overwrites {
4253 users
4254 .patch(key, &json!({ "name": format!("Ania {round}") }), false)
4255 .unwrap();
4256 }
4257 }
4258
4259 #[test]
4260 fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
4261 let dir = tempdir().unwrap();
4262 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4263 let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
4264
4265 churn(&mut db, "users", "u1", 12);
4266
4267 let stats = db.table("users").unwrap().stats();
4268 assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
4269 assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
4270 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4271 db.close().unwrap();
4272 }
4273
4274 #[test]
4275 fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
4276 let dir = tempdir().unwrap();
4277 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4278 let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
4279
4280 churn(&mut db, "users", "u1", 4);
4281
4282 let stats = db.table("users").unwrap().stats();
4283 assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
4284 assert!(stats.dead_bytes > 1);
4285 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
4286 db.close().unwrap();
4287 }
4288
4289 #[test]
4290 fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
4291 let dir = tempdir().unwrap();
4292 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4293 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4294
4295 churn(&mut db, "users", "u1", 12);
4296
4297 let stats = db.table("users").unwrap().stats();
4298 assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
4299 assert_eq!(stats.dead_bytes, 0);
4300 assert_eq!(stats.records, 1);
4301 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4302 assert_eq!(
4303 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4304 Some(&json!("Ania 11"))
4305 );
4306 db.close().unwrap();
4307 }
4308
4309 #[test]
4310 fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
4311 let dir = tempdir().unwrap();
4312 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4313 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4314 churn(&mut db, "users", "u1", 12);
4315 db.table_mut("users")
4316 .unwrap()
4317 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
4318 .unwrap();
4319 let before = db.table("users").unwrap().stats();
4320 db.close().unwrap();
4321
4322 let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
4323
4324 let after = reopened.table("users").unwrap().stats();
4325 assert_eq!(
4326 after.gen, before.gen,
4327 "the generation is carried by the snapshot, not reset"
4328 );
4329 assert_eq!(after.records, 2);
4330 assert_eq!(after.dead_bytes, 0);
4331 let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
4332 keys.sort();
4333 assert_eq!(keys, vec!["u1", "u2"]);
4334 assert_eq!(
4335 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4336 Some(&json!("Ania 11"))
4337 );
4338 reopened.close().unwrap();
4339 }
4340
4341 #[test]
4342 fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
4343 let dir = tempdir().unwrap();
4344 let schema = parse_schema(&json!({
4345 "schemaVersion": 1,
4346 "table": "tickets",
4347 "primaryKey": "id",
4348 "partitionBy": "kind",
4349 "fields": {
4350 "id": { "type": "string" },
4351 "kind": { "type": "string", "required": true },
4352 "note": { "type": "string" },
4353 },
4354 }))
4355 .unwrap();
4356 write_catalog_to(dir.path(), &catalog_of(&[schema]));
4357 let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
4359
4360 let tickets = db.table_mut("tickets").unwrap();
4361 tickets
4362 .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
4363 .unwrap();
4364 let refund = dir.path().join("tickets").join("refund.jsonl");
4365 let quiet_size = fs::metadata(&refund).unwrap().len();
4366
4367 tickets
4368 .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
4369 .unwrap();
4370 for round in 0..12 {
4371 tickets
4372 .patch("s1", &json!({ "note": format!("n{round}") }), false)
4373 .unwrap();
4374 }
4375
4376 let sale = dir.path().join("tickets").join("sale.jsonl");
4377 assert!(
4378 lines_of(&sale).len() < 13,
4379 "the partition over its thresholds was rewritten, so not all 13 lines are still there"
4380 );
4381 assert_eq!(
4382 fs::metadata(&refund).unwrap().len(),
4383 quiet_size,
4384 "the partition holding no garbage was left byte for byte"
4385 );
4386 assert_eq!(lines_of(&refund).len(), 1);
4387 let stats = db.table("tickets").unwrap().stats();
4388 assert_eq!(
4389 stats.gen, 1,
4390 "one pass moves the generation once, not once per partition"
4391 );
4392 assert_eq!(stats.records, 2);
4393 db.close().unwrap();
4394 }
4395
4396 #[test]
4397 fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
4398 let dir = tempdir().unwrap();
4399 let mut catalog = catalog_of(&[users_schema()]);
4400 catalog.defaults = Some(TableOptions {
4401 compaction: Some(CatalogCompaction::Disabled),
4402 ..Default::default()
4403 });
4404 write_catalog_to(dir.path(), &catalog);
4405 let mut db = opened(dir.path(), None);
4406
4407 churn(&mut db, "users", "u1", 12);
4408
4409 assert_eq!(db.table("users").unwrap().stats().gen, 0);
4410 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
4411 db.close().unwrap();
4412 }
4413
4414 #[test]
4415 fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
4416 let dir = tempdir().unwrap();
4417 let mut catalog = catalog_of(&[users_schema()]);
4418 catalog.defaults = Some(TableOptions {
4419 compaction: Some(CatalogCompaction::Disabled),
4420 ..Default::default()
4421 });
4422 catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
4423 compaction: Some(enabled(0.5, 200)),
4424 ..Default::default()
4425 });
4426 write_catalog_to(dir.path(), &catalog);
4427
4428 let db = opened(dir.path(), None);
4429 assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
4430 db.close().unwrap();
4431
4432 let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
4433 assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
4434 db.close().unwrap();
4435 }
4436
4437 #[test]
4438 fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
4439 let dir = tempdir().unwrap();
4440 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
4441 db.create_table(
4442 "users",
4443 json!({
4444 "schemaVersion": 1,
4445 "table": "users",
4446 "primaryKey": "id",
4447 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4448 }),
4449 )
4450 .unwrap();
4451
4452 churn(&mut db, "users", "u1", 12);
4453
4454 assert!(db.table("users").unwrap().stats().gen > 0);
4455 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
4456 db.close().unwrap();
4457 }
4458
4459 #[test]
4460 fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
4461 let dir = tempdir().unwrap();
4462 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4463 let db = opened(dir.path(), None);
4464
4465 assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
4466 db.close().unwrap();
4467 }
4468 }
4469
4470 mod read_cache {
4473 use super::*;
4474
4475 fn schema() -> Value {
4476 json!({
4477 "schemaVersion": 1,
4478 "table": "users",
4479 "primaryKey": "id",
4480 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
4481 })
4482 }
4483
4484 fn opened(dir: &Path, cache: CachePolicy) -> Database {
4485 Database::open_with(
4486 dir,
4487 OpenOptions {
4488 cache,
4489 ..Default::default()
4490 },
4491 )
4492 .unwrap()
4493 }
4494
4495 #[test]
4496 fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
4497 let dir = tempdir().unwrap();
4498 let mut db = Database::open(dir.path()).unwrap();
4499 db.create_table("users", schema()).unwrap();
4500 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4501 db.close().unwrap();
4502
4503 let db = Database::open(dir.path()).unwrap();
4504 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
4505 db.close().unwrap();
4506 }
4507
4508 #[test]
4509 fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
4510 let dir = tempdir().unwrap();
4511 let mut db = opened(dir.path(), CachePolicy::with_entries(64));
4512 db.create_table("users", schema()).unwrap();
4513 assert_eq!(
4514 db.table("users").unwrap().cache_policy(),
4515 CachePolicy::with_entries(64),
4516 "a table created under the handle is armed on the same terms"
4517 );
4518 db.close().unwrap();
4519
4520 let db = opened(dir.path(), CachePolicy::with_entries(64));
4521 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
4522 db.close().unwrap();
4523 }
4524
4525 #[test]
4526 fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
4527 let dir = tempdir().unwrap();
4528 let mut db = opened(dir.path(), CachePolicy::with_entries(8));
4529 db.create_table("users", schema()).unwrap();
4530 let users = db.table_mut("users").unwrap();
4531 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
4532
4533 let first = users.get("u1").unwrap();
4534 let second = users.get("u1").unwrap();
4535
4536 assert_eq!(first, second);
4537 assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
4538 db.close().unwrap();
4539 }
4540 }
4541
4542 mod cypher {
4543 use super::*;
4544 use crate::crypto::{CipherOptions, MissingKeyPolicy};
4545 use std::collections::HashMap;
4546
4547 fn key(byte: u8) -> Vec<u8> {
4548 vec![byte; 32]
4549 }
4550
4551 fn secrets_schema_json() -> Value {
4552 json!({
4553 "schemaVersion": 1,
4554 "table": "secrets",
4555 "primaryKey": "id",
4556 "fields": {
4557 "id": { "type": "string" },
4558 "token": { "type": "string", "cypher": true },
4559 },
4560 })
4561 }
4562
4563 fn secrets_schema_json_with_note() -> Value {
4564 json!({
4565 "schemaVersion": 1,
4566 "table": "secrets",
4567 "primaryKey": "id",
4568 "fields": {
4569 "id": { "type": "string" },
4570 "token": { "type": "string", "cypher": true },
4571 "note": { "type": "string" },
4572 },
4573 })
4574 }
4575
4576 fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
4577 Database::open_with(
4578 dir,
4579 OpenOptions {
4580 cypher: Some(CipherOptions {
4581 keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
4582 key_id: key_id.map(str::to_string),
4583 on_missing_key: None,
4584 }),
4585 ..OpenOptions::default()
4586 },
4587 )
4588 .unwrap()
4589 }
4590
4591 #[test]
4592 fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
4593 let dir = tempdir().unwrap();
4594 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4595 db.create_table("secrets", secrets_schema_json()).unwrap();
4596 db.table_mut("secrets")
4597 .unwrap()
4598 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4599 .unwrap();
4600 db.close().unwrap();
4601
4602 let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
4603 assert_eq!(
4604 reopened
4605 .table("secrets")
4606 .unwrap()
4607 .get("s1")
4608 .unwrap()
4609 .unwrap()
4610 .get("token"),
4611 Some(&json!("hunter2"))
4612 );
4613 reopened.close().unwrap();
4614 }
4615
4616 #[test]
4617 fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
4618 let dir = tempdir().unwrap();
4619 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4620 db.create_table("secrets", secrets_schema_json()).unwrap();
4621 db.table_mut("secrets")
4622 .unwrap()
4623 .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
4624 .unwrap();
4625 db.close().unwrap();
4626
4627 let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
4628 assert_eq!(
4629 rotated
4630 .table("secrets")
4631 .unwrap()
4632 .get("s1")
4633 .unwrap()
4634 .unwrap()
4635 .get("token"),
4636 Some(&json!("written-with-k1"))
4637 );
4638 rotated
4639 .table_mut("secrets")
4640 .unwrap()
4641 .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
4642 .unwrap();
4643 rotated.close().unwrap();
4644
4645 let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
4646 assert!(log.contains("$cypher$v1$k1$"));
4647 assert!(log.contains("$cypher$v1$k2$"));
4648 }
4649
4650 #[test]
4651 fn keeps_the_keyring_across_drop_field_reopening_the_table() {
4652 let dir = tempdir().unwrap();
4653 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4654 db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
4655 db.table_mut("secrets")
4656 .unwrap()
4657 .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
4658 .unwrap();
4659
4660 db.drop_field("secrets", "note", true).unwrap();
4661 assert_eq!(
4662 db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
4663 Some(&json!("hunter2"))
4664 );
4665 db.close().unwrap();
4666 }
4667
4668 #[test]
4669 fn open_refuses_a_key_that_is_not_32_bytes() {
4670 let dir = tempdir().unwrap();
4671 let error = Database::open_with(
4672 dir.path(),
4673 OpenOptions {
4674 cypher: Some(CipherOptions {
4675 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
4676 key_id: None,
4677 on_missing_key: None,
4678 }),
4679 ..OpenOptions::default()
4680 },
4681 )
4682 .map(|_| ())
4683 .unwrap_err();
4684 assert!(matches!(error, DatabaseError::Cipher(_)));
4685 }
4686
4687 #[test]
4688 fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
4689 let dir = tempdir().unwrap();
4690 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4691 db.create_table("secrets", secrets_schema_json()).unwrap();
4692 db.table_mut("secrets")
4693 .unwrap()
4694 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4695 .unwrap();
4696 db.close().unwrap();
4697
4698 let raw = Database::open_with(
4699 dir.path(),
4700 OpenOptions {
4701 cypher: Some(CipherOptions {
4702 keys: HashMap::from([("other".to_string(), key(2))]),
4703 key_id: None,
4704 on_missing_key: Some(MissingKeyPolicy::Raw),
4705 }),
4706 ..OpenOptions::default()
4707 },
4708 )
4709 .unwrap();
4710 let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
4711 assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
4712 raw.close().unwrap();
4713 }
4714 }
4715
4716 mod partitions {
4717 use super::*;
4718
4719 fn notes_schema_json(partitioned: bool) -> Value {
4720 let mut spec = json!({
4721 "schemaVersion": 1,
4722 "table": "notes",
4723 "primaryKey": "id",
4724 "fields": {
4725 "id": { "type": "string" },
4726 "kind": { "type": "string", "required": true },
4727 "title": { "type": "string", "required": true },
4728 "body": { "type": "string" },
4729 },
4730 });
4731 if partitioned {
4732 spec.as_object_mut()
4733 .unwrap()
4734 .insert("partitionBy".into(), json!("kind"));
4735 }
4736 spec
4737 }
4738
4739 fn lines_in(dir: &Path, part: &str) -> usize {
4740 fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
4741 .unwrap()
4742 .trim()
4743 .lines()
4744 .filter(|line| !line.is_empty())
4745 .count()
4746 }
4747
4748 #[test]
4749 fn a_partitioned_table_survives_the_catalog_round_trip() {
4750 let dir = tempdir().unwrap();
4751 {
4752 let mut db = Database::open(dir.path()).unwrap();
4753 db.create_table("notes", notes_schema_json(true)).unwrap();
4754 let notes = db.table_mut("notes").unwrap();
4755 notes
4756 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4757 .unwrap();
4758 notes
4759 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4760 .unwrap();
4761 db.close().unwrap();
4762 }
4763
4764 let db = Database::open(dir.path()).unwrap();
4765 let notes = db.table("notes").unwrap();
4766 assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
4767 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4768 assert_eq!(lines_in(dir.path(), "note"), 1);
4769 assert_eq!(lines_in(dir.path(), "adr"), 1);
4770 assert!(!dir.path().join("notes.jsonl").exists());
4771 db.close().unwrap();
4772 }
4773
4774 #[test]
4775 fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
4776 let dir = tempdir().unwrap();
4777 {
4778 let mut db = Database::open(dir.path()).unwrap();
4779 db.create_table("notes", notes_schema_json(false)).unwrap();
4780 let notes = db.table_mut("notes").unwrap();
4781 notes
4782 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4783 .unwrap();
4784 notes
4785 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4786 .unwrap();
4787 notes.delete("n1", false).unwrap();
4788 db.close().unwrap();
4789 }
4790
4791 let mut db = Database::open(dir.path()).unwrap();
4792 let mut partitioned = notes_schema_json(true);
4793 partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
4794 db.create_table("notes", partitioned).unwrap();
4795
4796 let notes = db.table("notes").unwrap();
4797 assert_eq!(notes.get("n1").unwrap(), None);
4798 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4799 assert!(!dir.path().join("notes.jsonl").exists());
4800 assert_eq!(lines_in(dir.path(), "adr"), 1);
4801 db.close().unwrap();
4802 }
4803
4804 #[test]
4805 fn dropping_the_field_the_table_partitions_by_is_refused() {
4806 let dir = tempdir().unwrap();
4807 let mut db = Database::open(dir.path()).unwrap();
4808 db.create_table("notes", notes_schema_json(true)).unwrap();
4809 db.table_mut("notes")
4810 .unwrap()
4811 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4812 .unwrap();
4813
4814 let error = db.drop_field("notes", "kind", true).unwrap_err();
4815 assert!(error.to_string().contains("partitioned by"));
4816 assert_eq!(
4817 db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
4818 Some(&json!("note"))
4819 );
4820 db.close().unwrap();
4821 }
4822
4823 #[test]
4824 fn dropping_an_unrelated_field_keeps_every_partition_readable() {
4825 let dir = tempdir().unwrap();
4826 let mut db = Database::open(dir.path()).unwrap();
4827 db.create_table("notes", notes_schema_json(true)).unwrap();
4828 {
4829 let notes = db.table_mut("notes").unwrap();
4830 notes
4831 .put(
4832 &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
4833 false,
4834 )
4835 .unwrap();
4836 notes
4837 .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
4838 .unwrap();
4839 }
4840
4841 db.drop_field("notes", "body", true).unwrap();
4842
4843 let notes = db.table("notes").unwrap();
4844 assert!(!notes.schema().fields.contains_key("body"));
4845 assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
4846 assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
4847 assert_eq!(lines_in(dir.path(), "note"), 1);
4848 assert_eq!(lines_in(dir.path(), "adr"), 1);
4849 db.close().unwrap();
4850 }
4851
4852 #[test]
4853 fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
4854 let dir = tempdir().unwrap();
4855 let mut db = Database::open(dir.path()).unwrap();
4856 db.create_table("notes", notes_schema_json(true)).unwrap();
4857
4858 let mut unit = db.unit_of_work();
4859 unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
4860 unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
4861 let result = db.commit(&mut unit).unwrap();
4862 assert_eq!(result.operations, 2);
4863
4864 let notes = db.table("notes").unwrap();
4865 assert_eq!(
4866 notes.get("n1").unwrap().unwrap().get("title"),
4867 Some(&json!("from unit"))
4868 );
4869 assert_eq!(
4870 notes.get("a1").unwrap().unwrap().get("title"),
4871 Some(&json!("from unit too"))
4872 );
4873 assert_eq!(lines_in(dir.path(), "note"), 1);
4874 assert_eq!(lines_in(dir.path(), "adr"), 1);
4875 db.close().unwrap();
4876 }
4877 }
4878 mod following_the_log {
4879 use super::*;
4880
4881 fn readonly(follow: FollowOption) -> OpenOptions {
4882 OpenOptions {
4883 readonly: true,
4884 follow,
4885 ..Default::default()
4886 }
4887 }
4888
4889 fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
4893 let mut writer = Database::open(dir).unwrap();
4894 writer.create_table("users", users_schema_spec()).unwrap();
4895 writer
4896 .table_mut("users")
4897 .unwrap()
4898 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4899 .unwrap();
4900 let reader = Database::open_with(dir, readonly(follow)).unwrap();
4901 (writer, reader)
4902 }
4903
4904 fn users_schema_spec() -> Value {
4905 json!({
4906 "schemaVersion": 1,
4907 "table": "users",
4908 "primaryKey": "id",
4909 "fields": {
4910 "id": { "type": "string" },
4911 "name": { "type": "string", "required": true },
4912 },
4913 })
4914 }
4915
4916 fn put(writer: &mut Database, key: &str, name: &str) {
4917 writer
4918 .table_mut("users")
4919 .unwrap()
4920 .put(&json!({ "id": key, "name": name }), false)
4921 .unwrap();
4922 }
4923
4924 #[test]
4925 fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
4926 let dir = tempdir().unwrap();
4927 let mut writer = Database::open(dir.path()).unwrap();
4928 writer.create_table("users", users_schema_spec()).unwrap();
4929
4930 assert_eq!(writer.follow(), None);
4931 assert_eq!(writer.table("users").unwrap().follow(), None);
4932
4933 let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
4934 assert_eq!(
4935 reader.follow(),
4936 Some(FollowPolicy {
4937 interval: DEFAULT_FOLLOW_INTERVAL
4938 })
4939 );
4940 assert_eq!(
4941 reader.table("users").unwrap().follow(),
4942 Some(FollowPolicy {
4943 interval: DEFAULT_FOLLOW_INTERVAL
4944 })
4945 );
4946 }
4947
4948 #[test]
4949 fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
4950 let dir = tempdir().unwrap();
4951 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4952
4953 for index in 2..=5 {
4954 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4955 let found = reader
4956 .table_mut("users")
4957 .unwrap()
4958 .get_following(&format!("u{index}"))
4959 .unwrap();
4960 assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
4961 }
4962 }
4963
4964 #[test]
4965 fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
4966 let dir = tempdir().unwrap();
4967 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4968
4969 let spec = QueryBuilder::new().spec();
4970 assert_eq!(
4971 reader
4972 .table_mut("users")
4973 .unwrap()
4974 .query_following(&spec)
4975 .unwrap()
4976 .matched,
4977 1
4978 );
4979
4980 put(&mut writer, "u2", "Osoba 2");
4981
4982 assert_eq!(
4983 reader
4984 .table_mut("users")
4985 .unwrap()
4986 .query_following(&spec)
4987 .unwrap()
4988 .matched,
4989 2
4990 );
4991 }
4992
4993 #[test]
4994 fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
4995 let dir = tempdir().unwrap();
4996 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
4997 put(&mut writer, "u2", "Osoba 2");
4998
4999 assert!(reader
5000 .table_mut("users")
5001 .unwrap()
5002 .get_following("u2")
5003 .unwrap()
5004 .is_none());
5005 assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
5006 assert!(reader
5007 .table_mut("users")
5008 .unwrap()
5009 .get_following("u2")
5010 .unwrap()
5011 .is_some());
5012 }
5013
5014 #[test]
5015 fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
5016 let dir = tempdir().unwrap();
5017 let (mut writer, mut reader) =
5018 writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
5019
5020 assert!(reader
5022 .table_mut("users")
5023 .unwrap()
5024 .get_following("u1")
5025 .unwrap()
5026 .is_some());
5027 put(&mut writer, "u2", "Osoba 2");
5028 assert!(reader
5029 .table_mut("users")
5030 .unwrap()
5031 .get_following("u2")
5032 .unwrap()
5033 .is_none());
5034
5035 std::thread::sleep(Duration::from_millis(200));
5036
5037 assert!(reader
5038 .table_mut("users")
5039 .unwrap()
5040 .get_following("u2")
5041 .unwrap()
5042 .is_some());
5043 }
5044
5045 #[test]
5046 fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
5047 let dir = tempdir().unwrap();
5048 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5049 put(&mut writer, "u2", "Osoba 2");
5050
5051 let keys = vec!["u1".to_string(), "u2".to_string()];
5052 let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
5053 assert!(rows.iter().all(Option::is_some), "{rows:?}");
5054
5055 put(&mut writer, "u3", "Osoba 3");
5056 let spec = QueryBuilder::new().spec();
5057 let aggregate = crate::query::aggregate::AggregateSpec {
5058 count: true,
5059 ..Default::default()
5060 };
5061 let counted = reader
5062 .table_mut("users")
5063 .unwrap()
5064 .aggregate_following(&spec, &aggregate)
5065 .unwrap();
5066 assert_eq!(counted.count, 3);
5067 }
5068
5069 #[test]
5070 fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
5071 let dir = tempdir().unwrap();
5072 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
5073 for index in 2..=6 {
5074 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
5075 }
5076 writer.table_mut("users").unwrap().delete("u5", false).unwrap();
5077 assert!(reader
5078 .table_mut("users")
5079 .unwrap()
5080 .get_following("u6")
5081 .unwrap()
5082 .is_some());
5083
5084 writer.table_mut("users").unwrap().compact().unwrap();
5085
5086 let users = reader.table_mut("users").unwrap();
5087 assert_eq!(
5088 users.get_following("u3").unwrap().unwrap().get("name"),
5089 Some(&json!("Osoba 3"))
5090 );
5091 assert!(users.get_following("u5").unwrap().is_none());
5092 assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
5093 }
5094 }
5095
5096 mod where_the_files_land {
5097 use super::*;
5098 use crate::database::LayoutKind;
5099
5100 fn notes_schema_spec() -> Value {
5101 json!({
5102 "schemaVersion": 1,
5103 "table": "notes",
5104 "primaryKey": "id",
5105 "partitionBy": "kind",
5106 "fields": {
5107 "id": { "type": "string" },
5108 "kind": { "type": "string", "required": true, "index": "exact" },
5109 "title": { "type": "string", "required": true },
5110 },
5111 })
5112 }
5113
5114 fn nested(metadata_file: Option<&str>) -> OpenOptions {
5115 OpenOptions {
5116 layout: LayoutOptions {
5117 kind: Some(LayoutKind::Nested),
5118 metadata_file: metadata_file.map(str::to_string),
5119 ..Default::default()
5120 },
5121 ..Default::default()
5122 }
5123 }
5124
5125 #[test]
5126 fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
5127 let dir = tempdir().unwrap();
5128 let mut db = Database::open(dir.path()).unwrap();
5129 db.create_table(
5130 "users",
5131 json!({
5132 "schemaVersion": 1,
5133 "table": "users",
5134 "primaryKey": "id",
5135 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5136 }),
5137 )
5138 .unwrap();
5139 db.table_mut("users")
5140 .unwrap()
5141 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5142 .unwrap();
5143 db.close().unwrap();
5144
5145 assert!(dir.path().join("hayson.json").is_file());
5146 assert!(dir.path().join("users.jsonl").is_file());
5147 assert!(!dir.path().join("tables").exists());
5148 assert!(!dir.path().join("runtime").exists());
5149
5150 let db = Database::open(dir.path()).unwrap();
5151 assert_eq!(db.layout().kind(), LayoutKind::Flat);
5152 assert_eq!(db.layout().metadata_file(), "hayson.json");
5153 assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
5154 }
5155
5156 #[test]
5157 fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
5158 let dir = tempdir().unwrap();
5159 let options = OpenOptions {
5160 layout: LayoutOptions {
5161 metadata_file: Some("cerebro.json".into()),
5162 ..Default::default()
5163 },
5164 ..Default::default()
5165 };
5166
5167 let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
5168 db.create_table(
5169 "users",
5170 json!({
5171 "schemaVersion": 1,
5172 "table": "users",
5173 "primaryKey": "id",
5174 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5175 }),
5176 )
5177 .unwrap();
5178 db.table_mut("users")
5179 .unwrap()
5180 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5181 .unwrap();
5182 db.close().unwrap();
5183
5184 assert!(dir.path().join("cerebro.json").is_file());
5185 assert!(!dir.path().join("hayson.json").exists());
5186
5187 let db = Database::open_with(dir.path(), options).unwrap();
5188 assert_eq!(db.layout().metadata_file(), "cerebro.json");
5189 assert_eq!(db.table_names(), vec!["users"]);
5190 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5191 }
5192
5193 #[test]
5194 fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
5195 let dir = tempdir().unwrap();
5196 let legacy = OpenOptions {
5197 layout: LayoutOptions {
5198 manifest: Some("cerebro.json".into()),
5199 ..Default::default()
5200 },
5201 ..Default::default()
5202 };
5203 let mut db = Database::open_with(dir.path(), legacy).unwrap();
5204 db.create_table(
5205 "users",
5206 json!({
5207 "schemaVersion": 1,
5208 "table": "users",
5209 "primaryKey": "id",
5210 "fields": { "id": { "type": "string" } },
5211 }),
5212 )
5213 .unwrap();
5214 db.close().unwrap();
5215
5216 let current = OpenOptions {
5217 layout: LayoutOptions {
5218 metadata_file: Some("cerebro.json".into()),
5219 ..Default::default()
5220 },
5221 ..Default::default()
5222 };
5223 let db = Database::open_with(dir.path(), current).unwrap();
5224 assert_eq!(db.table_names(), vec!["users"]);
5225 assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
5226 }
5227
5228 #[test]
5229 fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
5230 let dir = tempdir().unwrap();
5231 let options = OpenOptions {
5232 layout: LayoutOptions {
5233 metadata_file: Some("cerebro.json".into()),
5234 manifest: Some("hayson.json".into()),
5235 ..Default::default()
5236 },
5237 ..Default::default()
5238 };
5239
5240 let Err(error) = Database::open_with(dir.path(), options) else {
5241 panic!("two disagreeing names for the metadata document were accepted")
5242 };
5243
5244 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5245 let message = error.to_string();
5246 assert!(
5247 message.contains("metadataFile") && message.contains("manifest"),
5248 "{message}"
5249 );
5250 assert!(!dir.path().join("hayson.json").exists());
5251 assert!(!dir.path().join("cerebro.json").exists());
5252 assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
5253 }
5254
5255 #[test]
5256 fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
5257 let dir = tempdir().unwrap();
5258 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5259 db.create_table(
5260 "users",
5261 json!({
5262 "schemaVersion": 1,
5263 "table": "users",
5264 "primaryKey": "id",
5265 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5266 }),
5267 )
5268 .unwrap();
5269 db.table_mut("users")
5270 .unwrap()
5271 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5272 .unwrap();
5273
5274 assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
5275 db.close().unwrap();
5276
5277 assert!(dir.path().join("hayson.json").is_file());
5278 assert!(dir.path().join("tables").join("users.jsonl").is_file());
5279 assert!(dir.path().join("index").join("users.idx").is_file());
5280 assert!(!dir.path().join("users.jsonl").exists());
5281 }
5282
5283 #[test]
5284 fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
5285 let dir = tempdir().unwrap();
5286 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5287 db.create_table("notes", notes_schema_spec()).unwrap();
5288 db.table_mut("notes")
5289 .unwrap()
5290 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5291 .unwrap();
5292 db.close().unwrap();
5293
5294 assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
5295 }
5296
5297 #[test]
5298 fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
5299 let dir = tempdir().unwrap();
5300 let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
5301 db.create_table(
5302 "users",
5303 json!({
5304 "schemaVersion": 1,
5305 "table": "users",
5306 "primaryKey": "id",
5307 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5308 }),
5309 )
5310 .unwrap();
5311 db.table_mut("users")
5312 .unwrap()
5313 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5314 .unwrap();
5315 db.close().unwrap();
5316
5317 let named_only = OpenOptions {
5318 layout: LayoutOptions {
5319 metadata_file: Some("cerebro.json".into()),
5320 ..Default::default()
5321 },
5322 ..Default::default()
5323 };
5324 let db = Database::open_with(dir.path(), named_only).unwrap();
5325 assert_eq!(db.layout().kind(), LayoutKind::Nested);
5326 assert_eq!(
5327 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
5328 Some(&json!("Ania"))
5329 );
5330 db.close().unwrap();
5331
5332 let as_flat = OpenOptions {
5333 layout: LayoutOptions {
5334 kind: Some(LayoutKind::Flat),
5335 metadata_file: Some("cerebro.json".into()),
5336 ..Default::default()
5337 },
5338 ..Default::default()
5339 };
5340 let Err(error) = Database::open_with(dir.path(), as_flat) else {
5341 panic!("a nested database was opened as flat")
5342 };
5343 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
5344 assert!(error.to_string().contains("nested"), "{error}");
5345 }
5346
5347 #[test]
5348 fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
5349 let dir = tempdir().unwrap();
5350 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
5351 db.create_table(
5352 "users",
5353 json!({
5354 "schemaVersion": 1,
5355 "table": "users",
5356 "primaryKey": "id",
5357 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5358 }),
5359 )
5360 .unwrap();
5361
5362 let mut unit = db.unit_of_work();
5363 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
5364 unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
5365 db.commit(&mut unit).unwrap();
5366 db.close().unwrap();
5367
5368 assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
5369 let db = Database::open_with(dir.path(), nested(None)).unwrap();
5370 assert_eq!(db.table("users").unwrap().len(), 2);
5371 }
5372 }
5373
5374 mod a_declared_table_whose_log_is_not_on_disk {
5377 use super::*;
5378
5379 fn notes_spec() -> Value {
5380 json!({
5381 "schemaVersion": 1,
5382 "table": "notes",
5383 "primaryKey": "id",
5384 "partitionBy": "kind",
5385 "fields": {
5386 "id": { "type": "string" },
5387 "kind": { "type": "string", "required": true },
5388 "title": { "type": "string", "required": true },
5389 },
5390 })
5391 }
5392
5393 fn readonly() -> OpenOptions {
5394 OpenOptions {
5395 readonly: true,
5396 ..Default::default()
5397 }
5398 }
5399
5400 fn users_with_a_record(dir: &Path) {
5401 let mut db = Database::open(dir).unwrap();
5402 db.create_table(
5403 "users",
5404 json!({
5405 "schemaVersion": 1,
5406 "table": "users",
5407 "primaryKey": "id",
5408 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
5409 }),
5410 )
5411 .unwrap();
5412 db.table_mut("users")
5413 .unwrap()
5414 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5415 .unwrap();
5416 db.close().unwrap();
5417 }
5418
5419 fn notes_with_a_record(dir: &Path) {
5420 let mut db = Database::open(dir).unwrap();
5421 db.create_table("notes", notes_spec()).unwrap();
5422 db.table_mut("notes")
5423 .unwrap()
5424 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
5425 .unwrap();
5426 db.close().unwrap();
5427 }
5428
5429 fn refusal(dir: &Path) -> DatabaseError {
5430 match Database::open_with(dir, readonly()) {
5431 Err(error) => error,
5432 Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
5433 }
5434 }
5435
5436 fn schema_refusal(error: DatabaseError) -> String {
5437 match error {
5438 DatabaseError::Schema(HaysonSchemaError(message)) => message,
5439 other => panic!("expected a schema refusal, got {other:?}"),
5440 }
5441 }
5442
5443 #[test]
5444 fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
5445 let dir = tempdir().unwrap();
5446 users_with_a_record(dir.path());
5447 let log = log_path(dir.path(), "users");
5448 fs::remove_file(&log).unwrap();
5449
5450 let message = schema_refusal(refusal(dir.path()));
5451
5452 assert!(message.contains("\"users\""), "{message}");
5453 assert!(message.contains(&log.display().to_string()), "{message}");
5454 assert!(message.contains("readonly"), "{message}");
5455 assert!(!message.contains("createTable"), "{message}");
5456 assert!(!log.exists(), "the refused open must not have created the log");
5457 }
5458
5459 #[test]
5460 fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
5461 let dir = tempdir().unwrap();
5462 users_with_a_record(dir.path());
5463 fs::remove_file(log_path(dir.path(), "users")).unwrap();
5464
5465 let mut db = Database::open(dir.path()).unwrap();
5466
5467 assert_eq!(db.table_names(), vec!["users"]);
5468 assert!(log_path(dir.path(), "users").is_file());
5469 assert_eq!(db.table("users").unwrap().len(), 0);
5470 db.table_mut("users")
5471 .unwrap()
5472 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5473 .unwrap();
5474 assert_eq!(db.table("users").unwrap().len(), 1);
5475 db.close().unwrap();
5476 }
5477
5478 #[test]
5480 fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
5481 let dir = tempdir().unwrap();
5482 users_with_a_record(dir.path());
5483 fs::write(log_path(dir.path(), "users"), "").unwrap();
5484
5485 let db = Database::open_with(dir.path(), readonly()).unwrap();
5486
5487 assert_eq!(db.table_names(), vec!["users"]);
5488 assert_eq!(db.table("users").unwrap().len(), 0);
5489 }
5490
5491 #[test]
5492 fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
5493 let dir = tempdir().unwrap();
5494 notes_with_a_record(dir.path());
5495 let partition = dir.path().join("notes").join("note.jsonl");
5496 assert!(
5497 index_path(dir.path(), "notes").is_file(),
5498 "the refusal reads the snapshot"
5499 );
5500 fs::remove_dir_all(dir.path().join("notes")).unwrap();
5501
5502 let message = schema_refusal(refusal(dir.path()));
5503
5504 assert!(message.contains("\"notes\""), "{message}");
5505 assert!(message.contains(&partition.display().to_string()), "{message}");
5506 assert!(message.contains("readonly"), "{message}");
5507 assert!(!message.contains("createTable"), "{message}");
5508 }
5509
5510 #[test]
5511 fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
5512 let dir = tempdir().unwrap();
5513 let mut db = Database::open(dir.path()).unwrap();
5514 db.create_table("notes", notes_spec()).unwrap();
5515 db.close().unwrap();
5516
5517 let db = Database::open_with(dir.path(), readonly()).unwrap();
5518
5519 assert_eq!(db.table_names(), vec!["notes"]);
5520 assert_eq!(db.table("notes").unwrap().len(), 0);
5521 }
5522
5523 #[cfg(unix)]
5526 #[test]
5527 fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
5528 let dir = tempdir().unwrap();
5529 notes_with_a_record(dir.path());
5530 let partition = dir.path().join("notes").join("note.jsonl");
5531 fs::remove_file(&partition).unwrap();
5532 std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
5533
5534 let error = refusal(dir.path());
5535
5536 match error {
5537 DatabaseError::Log(path, inner) => {
5538 assert_eq!(path, partition);
5539 assert_eq!(inner.kind(), io::ErrorKind::NotFound);
5540 }
5541 other => panic!("expected a log refusal naming the partition, got {other:?}"),
5542 }
5543 assert!(!dir.path().join("notes").join("gone.jsonl").exists());
5544 }
5545 }
5546
5547 mod cascading_delete {
5548 use super::*;
5549
5550 fn orders_schema_optional_user() -> TableSchema {
5551 parse_schema(&json!({
5552 "schemaVersion": 1,
5553 "table": "orders",
5554 "primaryKey": "id",
5555 "fields": {
5556 "id": { "type": "string" },
5557 "userId": { "type": "string" },
5558 },
5559 }))
5560 .unwrap()
5561 }
5562
5563 fn order_items_schema() -> TableSchema {
5564 parse_schema(&json!({
5565 "schemaVersion": 1,
5566 "table": "order_items",
5567 "primaryKey": "id",
5568 "fields": {
5569 "id": { "type": "string" },
5570 "orderId": { "type": "string", "required": true },
5571 },
5572 }))
5573 .unwrap()
5574 }
5575
5576 fn orders_user_relation_on_delete(action: &str) -> Value {
5577 json!({
5578 "name": "orders_user",
5579 "from": { "table": "orders", "field": "userId" },
5580 "to": { "table": "users", "field": "id" },
5581 "cardinality": "many-to-one",
5582 "onDelete": action,
5583 })
5584 }
5585
5586 fn order_items_order_relation_cascade() -> Value {
5587 json!({
5588 "name": "order_items_order",
5589 "from": { "table": "order_items", "field": "orderId" },
5590 "to": { "table": "orders", "field": "id" },
5591 "cardinality": "many-to-one",
5592 "onDelete": "cascade",
5593 })
5594 }
5595
5596 #[test]
5597 fn a_table_no_relation_names_as_to_goes_straight_through_plain_delete() {
5598 let dir = tempdir().unwrap();
5599 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5600 let mut db = Database::open(dir.path()).unwrap();
5601 db.table_mut("users")
5602 .unwrap()
5603 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5604 .unwrap();
5605
5606 let deleted = db.delete_checked("users", "u1", false).unwrap();
5607
5608 assert!(deleted);
5609 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5610 db.close().unwrap();
5611 }
5612
5613 #[test]
5614 fn a_missing_key_answers_false_without_staging_anything() {
5615 let dir = tempdir().unwrap();
5616 write_catalog_to(
5617 dir.path(),
5618 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5619 );
5620 let mut db = Database::open(dir.path()).unwrap();
5621 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5622 .unwrap();
5623
5624 let deleted = db.delete_checked("users", "ghost", false).unwrap();
5625
5626 assert!(!deleted);
5627 db.close().unwrap();
5628 }
5629
5630 #[test]
5631 fn restrict_refuses_when_a_referencing_record_exists_and_leaves_everything_untouched() {
5632 let dir = tempdir().unwrap();
5633 write_catalog_to(
5634 dir.path(),
5635 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5636 );
5637 let mut db = Database::open(dir.path()).unwrap();
5638 db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5639 .unwrap();
5640 db.table_mut("users")
5641 .unwrap()
5642 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5643 .unwrap();
5644 db.table_mut("orders")
5645 .unwrap()
5646 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5647 .unwrap();
5648
5649 let error = db.delete_checked("users", "u1", false).unwrap_err();
5650
5651 match error {
5652 DatabaseError::Table(table, TableError::RelationRestrict(inner)) => {
5653 assert_eq!(table, "users");
5654 assert_eq!(inner.relation, "orders_user");
5655 assert_eq!(inner.referencing_table, "orders");
5656 assert_eq!(inner.referencing, 1);
5657 }
5658 other => panic!("expected a RelationRestrict refusal, got {other:?}"),
5659 }
5660 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5661 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5662 db.close().unwrap();
5663 }
5664
5665 #[test]
5666 fn restrict_allows_the_delete_once_nothing_references_it() {
5667 let dir = tempdir().unwrap();
5668 write_catalog_to(
5669 dir.path(),
5670 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5671 );
5672 let mut db = Database::open(dir.path()).unwrap();
5673 db.declare_relation(orders_user_relation_on_delete("restrict"), false)
5674 .unwrap();
5675 db.table_mut("users")
5676 .unwrap()
5677 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5678 .unwrap();
5679
5680 let deleted = db.delete_checked("users", "u1", false).unwrap();
5681
5682 assert!(deleted);
5683 db.close().unwrap();
5684 }
5685
5686 #[test]
5687 fn cascade_deletes_every_referencing_record() {
5688 let dir = tempdir().unwrap();
5689 write_catalog_to(
5690 dir.path(),
5691 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5692 );
5693 let mut db = Database::open(dir.path()).unwrap();
5694 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5695 .unwrap();
5696 db.table_mut("users")
5697 .unwrap()
5698 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5699 .unwrap();
5700 db.table_mut("orders")
5701 .unwrap()
5702 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5703 .unwrap();
5704 db.table_mut("orders")
5705 .unwrap()
5706 .put(&json!({ "id": "o2", "userId": "u1" }), false)
5707 .unwrap();
5708 db.table_mut("orders")
5709 .unwrap()
5710 .put(&json!({ "id": "o3", "userId": "other" }), false)
5711 .unwrap();
5712
5713 let deleted = db.delete_checked("users", "u1", false).unwrap();
5714
5715 assert!(deleted);
5716 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5717 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5718 assert!(db.table("orders").unwrap().get("o2").unwrap().is_none());
5719 assert!(db.table("orders").unwrap().get("o3").unwrap().is_some());
5720 db.close().unwrap();
5721 }
5722
5723 #[test]
5724 fn cascade_recurses_through_a_chain_of_relations() {
5725 let dir = tempdir().unwrap();
5726 write_catalog_to(
5727 dir.path(),
5728 &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5729 );
5730 let mut db = Database::open(dir.path()).unwrap();
5731 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5732 .unwrap();
5733 db.declare_relation(order_items_order_relation_cascade(), false)
5734 .unwrap();
5735 db.table_mut("users")
5736 .unwrap()
5737 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5738 .unwrap();
5739 db.table_mut("orders")
5740 .unwrap()
5741 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5742 .unwrap();
5743 db.table_mut("order_items")
5744 .unwrap()
5745 .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5746 .unwrap();
5747
5748 let deleted = db.delete_checked("users", "u1", false).unwrap();
5749
5750 assert!(deleted);
5751 assert!(db.table("orders").unwrap().get("o1").unwrap().is_none());
5752 assert!(db.table("order_items").unwrap().get("i1").unwrap().is_none());
5753 db.close().unwrap();
5754 }
5755
5756 #[test]
5757 fn set_null_clears_the_referencing_field_instead_of_deleting_the_row() {
5758 let dir = tempdir().unwrap();
5759 write_catalog_to(
5760 dir.path(),
5761 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5762 );
5763 let mut db = Database::open(dir.path()).unwrap();
5764 db.declare_relation(orders_user_relation_on_delete("set-null"), false)
5765 .unwrap();
5766 db.table_mut("users")
5767 .unwrap()
5768 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5769 .unwrap();
5770 db.table_mut("orders")
5771 .unwrap()
5772 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5773 .unwrap();
5774
5775 let deleted = db.delete_checked("users", "u1", false).unwrap();
5776
5777 assert!(deleted);
5778 assert!(db.table("users").unwrap().get("u1").unwrap().is_none());
5779 let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5780 assert_eq!(order.get("userId"), Some(&Value::Null));
5781 db.close().unwrap();
5782 }
5783
5784 #[test]
5785 fn no_action_and_undeclared_leave_referencing_records_untouched() {
5786 let dir = tempdir().unwrap();
5787 write_catalog_to(
5788 dir.path(),
5789 &catalog_of(&[users_schema(), orders_schema_optional_user()]),
5790 );
5791 let mut db = Database::open(dir.path()).unwrap();
5792 db.declare_relation(orders_user_relation_on_delete("no-action"), false)
5793 .unwrap();
5794 db.table_mut("users")
5795 .unwrap()
5796 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5797 .unwrap();
5798 db.table_mut("orders")
5799 .unwrap()
5800 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5801 .unwrap();
5802
5803 let deleted = db.delete_checked("users", "u1", false).unwrap();
5804
5805 assert!(deleted);
5806 let order = db.table("orders").unwrap().get("o1").unwrap().unwrap();
5807 assert_eq!(order.get("userId"), Some(&Value::String("u1".to_string())));
5808 db.close().unwrap();
5809 }
5810
5811 #[test]
5812 fn a_restrict_deep_in_the_cascade_aborts_the_whole_delete_atomically() {
5813 let dir = tempdir().unwrap();
5814 write_catalog_to(
5815 dir.path(),
5816 &catalog_of(&[users_schema(), orders_schema_optional_user(), order_items_schema()]),
5817 );
5818 let mut db = Database::open(dir.path()).unwrap();
5819 db.declare_relation(orders_user_relation_on_delete("cascade"), false)
5820 .unwrap();
5821 db.declare_relation(
5822 json!({
5823 "name": "order_items_order",
5824 "from": { "table": "order_items", "field": "orderId" },
5825 "to": { "table": "orders", "field": "id" },
5826 "cardinality": "many-to-one",
5827 "onDelete": "restrict",
5828 }),
5829 false,
5830 )
5831 .unwrap();
5832 db.table_mut("users")
5833 .unwrap()
5834 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5835 .unwrap();
5836 db.table_mut("orders")
5837 .unwrap()
5838 .put(&json!({ "id": "o1", "userId": "u1" }), false)
5839 .unwrap();
5840 db.table_mut("order_items")
5841 .unwrap()
5842 .put(&json!({ "id": "i1", "orderId": "o1" }), false)
5843 .unwrap();
5844
5845 let error = db.delete_checked("users", "u1", false).unwrap_err();
5846
5847 assert!(matches!(
5848 error,
5849 DatabaseError::Table(_, TableError::RelationRestrict(_))
5850 ));
5851 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
5852 assert!(db.table("orders").unwrap().get("o1").unwrap().is_some());
5853 assert!(db.table("order_items").unwrap().get("i1").unwrap().is_some());
5854 db.close().unwrap();
5855 }
5856
5857 #[test]
5858 fn a_cycle_of_cascading_relations_does_not_recurse_forever() {
5859 let dir = tempdir().unwrap();
5860 let a_schema = parse_schema(&json!({
5861 "schemaVersion": 1,
5862 "table": "a",
5863 "primaryKey": "id",
5864 "fields": { "id": { "type": "string" }, "bId": { "type": "string" } },
5865 }))
5866 .unwrap();
5867 let b_schema = parse_schema(&json!({
5868 "schemaVersion": 1,
5869 "table": "b",
5870 "primaryKey": "id",
5871 "fields": { "id": { "type": "string" }, "aId": { "type": "string" } },
5872 }))
5873 .unwrap();
5874 write_catalog_to(dir.path(), &catalog_of(&[a_schema, b_schema]));
5875 let mut db = Database::open(dir.path()).unwrap();
5876 db.declare_relation(
5877 json!({
5878 "name": "a_b",
5879 "from": { "table": "a", "field": "bId" },
5880 "to": { "table": "b", "field": "id" },
5881 "cardinality": "many-to-one",
5882 "onDelete": "cascade",
5883 }),
5884 false,
5885 )
5886 .unwrap();
5887 db.declare_relation(
5888 json!({
5889 "name": "b_a",
5890 "from": { "table": "b", "field": "aId" },
5891 "to": { "table": "a", "field": "id" },
5892 "cardinality": "many-to-one",
5893 "onDelete": "cascade",
5894 }),
5895 false,
5896 )
5897 .unwrap();
5898 db.table_mut("a")
5899 .unwrap()
5900 .put(&json!({ "id": "a1", "bId": "b1" }), false)
5901 .unwrap();
5902 db.table_mut("b")
5903 .unwrap()
5904 .put(&json!({ "id": "b1", "aId": "a1" }), false)
5905 .unwrap();
5906
5907 let deleted = db.delete_checked("a", "a1", false).unwrap();
5908
5909 assert!(deleted);
5910 assert!(db.table("a").unwrap().get("a1").unwrap().is_none());
5911 assert!(db.table("b").unwrap().get("b1").unwrap().is_none());
5912 db.close().unwrap();
5913 }
5914
5915 #[test]
5916 fn delete_checked_refuses_on_a_readonly_handle() {
5917 let dir = tempdir().unwrap();
5918 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
5919 {
5920 let mut db = Database::open(dir.path()).unwrap();
5921 db.table_mut("users")
5922 .unwrap()
5923 .put(&json!({ "id": "u1", "name": "Ania" }), false)
5924 .unwrap();
5925 db.close().unwrap();
5926 }
5927 let mut db = Database::open_with(
5928 dir.path(),
5929 OpenOptions {
5930 readonly: true,
5931 ..OpenOptions::default()
5932 },
5933 )
5934 .unwrap();
5935
5936 let error = db.delete_checked("users", "u1", false).unwrap_err();
5937
5938 assert!(matches!(error, DatabaseError::ReadOnly { .. }));
5939 }
5940 }
5941
5942 mod describing_one_table {
5943 use super::*;
5944
5945 fn users_with_city() -> TableSchema {
5946 parse_schema(&json!({
5947 "schemaVersion": 1,
5948 "table": "users",
5949 "revision": 2,
5950 "primaryKey": "id",
5951 "fields": {
5952 "id": { "type": "string" },
5953 "name": { "type": "string", "required": true },
5954 "city": { "type": "string", "index": "exact" },
5955 },
5956 }))
5957 .unwrap()
5958 }
5959
5960 fn users_drifted_under(dir: &Path) -> Database {
5963 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
5964 let mut db = Database::open(dir).unwrap();
5965 write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
5966 assert!(db.refresh().unwrap());
5967 db
5968 }
5969
5970 fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
5971 match answered {
5972 Err(DatabaseError::Schema(error)) => error.to_string(),
5973 other => panic!("expected a schema refusal, got {other:?}"),
5974 }
5975 }
5976
5977 #[test]
5978 fn describes_a_table_of_its_own_while_another_one_has_drifted() {
5979 let dir = tempdir().unwrap();
5980 let db = users_drifted_under(dir.path());
5981
5982 let document = db.describe_table("orders").unwrap();
5983
5984 assert_eq!(document["table"], json!("orders"));
5985 assert_eq!(document["revision"], json!(1));
5986 assert!(db.catalog().is_err());
5987 }
5988
5989 #[test]
5990 fn refuses_the_table_that_drifted_itself() {
5991 let dir = tempdir().unwrap();
5992 let db = users_drifted_under(dir.path());
5993
5994 let message = refusal_of(db.describe_table("users"));
5995
5996 assert!(message.contains("\"users\""), "{message}");
5997 assert!(message.contains("revision 2"), "{message}");
5998 assert_eq!(db.table("users").unwrap().schema().revision, 1);
6000 }
6001
6002 #[test]
6003 fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
6004 let dir = tempdir().unwrap();
6005 let mut catalog = catalog_of(&[users_schema()]);
6006 let entry = catalog.tables.get_mut("users").unwrap();
6007 entry.indexes = Some(vec![IndexEntry {
6008 name: "by_name".to_string(),
6009 fields: vec!["name".to_string()],
6010 kind: crate::catalog::CatalogIndexKind::Exact,
6011 unique: false,
6012 where_: None,
6013 expire_after_seconds: None,
6014 }]);
6015 write_catalog_to(dir.path(), &catalog);
6016 let db = Database::open(dir.path()).unwrap();
6017
6018 let document = db.describe_table("users").unwrap();
6019
6020 assert_eq!(
6021 document["fields"]["name"],
6022 json!({ "type": "string", "required": true })
6023 );
6024 assert!(document.as_object().unwrap().get("indexes").is_none());
6025 assert!(db.catalog().is_ok());
6026 }
6027
6028 #[test]
6029 fn refuses_a_table_this_handle_holds_nowhere() {
6030 let dir = tempdir().unwrap();
6031 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
6032 let db = Database::open(dir.path()).unwrap();
6033
6034 let message = refusal_of(db.describe_table("ghosts"));
6035
6036 assert!(message.contains("declares no table \"ghosts\""), "{message}");
6037 }
6038 }
6039}