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