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, empty_catalog, expand_indexes, index_identity, parse_catalog, relations_touching, write_catalog,
13 Catalog, CatalogCompaction, CatalogLayout, ExpandedIndex, IntegrityReport, Relation, TableEntry, TableOptions,
14 TriggerEntry,
15};
16use crate::checkpoint::{read_snapshot, RestoreOutcome};
17use crate::crypto::{CipherOptions, FieldCipher};
18use crate::errors::{HaysonCipherError, HaysonSchemaError};
19use crate::lock::{DirLock, DirLockError, LockOptions};
20use crate::query::builder::QueryBuilder;
21use crate::schema::{is_partitioned, parse_schema, FieldSpec, 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, 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 fnv1a(bytes: &[u8]) -> String {
244 let mut hash = FNV_OFFSET;
245 for byte in bytes {
246 hash ^= u64::from(*byte);
247 hash = hash.wrapping_mul(FNV_PRIME);
248 }
249 format!("{hash:016x}")
250}
251
252fn schema_fingerprint(schema: &TableSchema, declared: &[ExpandedIndex]) -> String {
260 if declared.is_empty() {
261 return fnv1a(format!("{schema:?}").as_bytes());
262 }
263 let mut described: Vec<String> = declared.iter().map(|index| format!("{index:?}")).collect();
264 described.sort();
265 fnv1a(format!("{schema:?}|{}", described.join("|")).as_bytes())
266}
267
268fn table_schema_of(name: &str, entry: &TableEntry) -> TableSchema {
273 let fields: BTreeMap<String, FieldSpec> = entry
274 .fields
275 .iter()
276 .map(|(field, spec)| (field.clone(), spec.spec.clone()))
277 .collect();
278 TableSchema {
279 table: name.to_string(),
280 revision: entry.revision as u32,
281 primary_key: entry.primary_key.clone(),
282 partition_by: entry.partition_by.clone(),
283 fields,
284 }
285}
286
287fn triggers_of(catalog: &Catalog, name: &str) -> Vec<TriggerEntry> {
290 let Some(declared) = catalog.triggers.as_ref() else {
291 return Vec::new();
292 };
293 declared
294 .iter()
295 .filter(|trigger| trigger.table == name)
296 .cloned()
297 .collect()
298}
299
300fn sync_table_triggers(catalog: &Catalog, tables: &mut HashMap<String, Table>) {
306 for (name, table) in tables.iter_mut() {
307 table.set_triggers(triggers_of(catalog, name));
308 }
309}
310
311fn option_of<T>(catalog: &Catalog, entry: &TableEntry, pick: impl Fn(&TableOptions) -> Option<T>) -> Option<T> {
315 entry
316 .options
317 .as_ref()
318 .and_then(&pick)
319 .or_else(|| catalog.defaults.as_ref().and_then(&pick))
320}
321
322fn apply_image_dir(dir: &Path, catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
326 if let Some(image_dir) = option_of(catalog, entry, |options| options.image_dir.clone()) {
327 table.set_image_dir(dir.join(image_dir));
328 }
329}
330
331fn apply_sync(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
335 if let Some(sync) = option_of(catalog, entry, |options| options.sync) {
336 table.set_sync(sync);
337 }
338}
339
340fn apply_validation(entry: &TableEntry, table: &mut Table) {
348 if let Some(validation) = entry.validation {
349 table.set_validation(validation);
350 }
351}
352
353fn apply_deprecated(entry: &TableEntry, table: &mut Table) {
360 let marked: BTreeMap<String, String> = entry
361 .fields
362 .iter()
363 .filter_map(|(name, field)| field.deprecated.as_ref().map(|note| (name.clone(), note.clone())))
364 .collect();
365 if !marked.is_empty() {
366 table.set_deprecated_fields(marked);
367 }
368}
369
370fn apply_link_parsing(catalog: &Catalog, entry: &TableEntry, table: &mut Table) {
374 if let Some(link_parsing) = option_of(catalog, entry, |options| options.link_parsing) {
375 table.set_link_parsing(link_parsing);
376 }
377}
378
379fn compaction_of(named: Option<CatalogCompaction>, catalog: &Catalog, entry: Option<&TableEntry>) -> CatalogCompaction {
383 named
384 .or_else(|| {
385 entry
386 .and_then(|entry| entry.options.as_ref())
387 .and_then(|options| options.compaction)
388 })
389 .or_else(|| catalog.defaults.as_ref().and_then(|defaults| defaults.compaction))
390 .unwrap_or(DEFAULT_COMPACTION)
391}
392
393fn apply_compaction(compaction: CatalogCompaction, table: &mut Table) {
396 table.set_compaction(compaction);
397}
398
399fn apply_cipher(cipher: Option<&FieldCipher>, table: &mut Table) {
403 table.set_cipher(cipher.cloned());
404}
405
406fn assert_no_residual_values(name: &str, open_table: &Table, field: &str) -> Result<(), DatabaseError> {
410 let all = open_table
411 .query(&QueryBuilder::new().spec())
412 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
413 let mut residual = 0usize;
414 for key in &all.keys {
415 let record = open_table
416 .get(key)
417 .map_err(|error| DatabaseError::Table(name.to_string(), error))?;
418 if record.is_some_and(|record| record.contains_key(field)) {
419 residual += 1;
420 }
421 }
422 if residual > 0 {
423 return Err(DatabaseError::FieldInUse {
424 table: name.to_string(),
425 field: field.to_string(),
426 records: residual,
427 });
428 }
429 Ok(())
430}
431
432fn catalog_write_error(error: Box<dyn std::error::Error>) -> DatabaseError {
438 match error.downcast::<io::Error>() {
439 Ok(io) if AtomicTemporaryFileError::downcast(&io).is_some() => DatabaseError::Io(*io),
440 Ok(io) => DatabaseError::Catalog(io.to_string()),
441 Err(other) => DatabaseError::Catalog(other.to_string()),
442 }
443}
444
445fn partition_dir_of(layout: &Layout, table: &str) -> PathBuf {
448 layout.tables().join(table)
449}
450
451fn on_disk(path: &Path) -> bool {
454 fs::symlink_metadata(path).is_ok()
455}
456
457fn missing_logs_of(layout: &Layout, index_path: &Path, schema: &TableSchema) -> io::Result<Vec<PathBuf>> {
465 let flat = layout.log_path(&schema.table);
466 if !is_partitioned(schema) {
467 return Ok(if on_disk(&flat) { Vec::new() } else { vec![flat] });
468 }
469 let dir = partition_dir_of(layout, &schema.table);
470 let present = partition_names_in(&dir);
471 if present.is_empty() && on_disk(&flat) {
473 return Ok(Vec::new());
474 }
475 let claimed = read_snapshot(index_path)?
476 .and_then(|snapshot| snapshot.header.parts)
477 .unwrap_or_default();
478 Ok(claimed
479 .into_iter()
480 .filter(|part| !present.contains(&part.part))
481 .map(|part| dir.join(format!("{}{LOG_SUFFIX}", part.part)))
482 .collect())
483}
484
485fn present_logs_of(layout: &Layout, schema: &TableSchema) -> Vec<PathBuf> {
487 let flat = layout.log_path(&schema.table);
488 if !is_partitioned(schema) {
489 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
490 }
491 let dir = partition_dir_of(layout, &schema.table);
492 let present = partition_names_in(&dir);
493 if present.is_empty() {
494 return if on_disk(&flat) { vec![flat] } else { Vec::new() };
495 }
496 present
497 .iter()
498 .map(|part| dir.join(format!("{part}{LOG_SUFFIX}")))
499 .collect()
500}
501
502enum TableDrift {
505 Absent,
506 Schema { revision: u64 },
507 Indexes { unserved: Vec<String>, stale: Vec<String> },
508}
509
510struct DriftedTable {
511 name: String,
512 drift: TableDrift,
513}
514
515fn index_names_missing_from(expected: &[ExpandedIndex], present: &[ExpandedIndex]) -> Vec<String> {
518 let identities: HashSet<String> = present.iter().map(index_identity).collect();
519 let mut names: Vec<String> = expected
520 .iter()
521 .filter(|index| !identities.contains(&index_identity(index)))
522 .map(|index| index.name.clone())
523 .collect();
524 names.sort();
525 names
526}
527
528fn shape_of(schema: &TableSchema) -> TableSchema {
537 let mut bare = schema.clone();
538 for spec in bare.fields.values_mut() {
539 spec.index = None;
540 spec.unique = false;
541 spec.fulltext = false;
542 }
543 bare
544}
545
546fn quoted(names: &[String]) -> String {
547 names
548 .iter()
549 .map(|name| format!("\"{name}\""))
550 .collect::<Vec<_>>()
551 .join(", ")
552}
553
554fn stored_side_of(drift: &TableDrift) -> String {
555 match drift {
556 TableDrift::Absent => "declares no table at all".to_string(),
557 TableDrift::Schema { revision } => format!("declares revision {revision}"),
558 TableDrift::Indexes { unserved, stale } => {
559 let mut halves = Vec::new();
560 if !unserved.is_empty() {
561 halves.push(format!(
562 "declares indexes this handle does not serve: {}",
563 quoted(unserved)
564 ));
565 }
566 if !stale.is_empty() {
567 halves.push(format!(
568 "no longer declares indexes this handle still serves: {}",
569 quoted(stale)
570 ));
571 }
572 halves.join(", and ")
573 }
574 }
575}
576
577fn missing_logs_error(dir: &Path, name: &str, missing: &[PathBuf]) -> HaysonSchemaError {
578 let listed = missing
579 .iter()
580 .map(|path| format!("\"{}\"", path.display()))
581 .collect::<Vec<_>>()
582 .join(", ");
583 let one = missing.len() == 1;
584 let (its, is, it, file, log) = if one {
585 ("its log".to_string(), "is", "it", "file", "log")
586 } else {
587 (format!("{} of its logs", missing.len()), "are", "them", "files", "logs")
588 };
589 HaysonSchemaError(format!(
590 "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",
591 dir.display()
592 ))
593}
594
595fn refuse_unreadable_logs(layout: &Layout, name: &str, entry: &TableEntry) -> Result<(), DatabaseError> {
606 let schema = table_schema_of(name, entry);
607 let missing = missing_logs_of(layout, &layout.index_path(name), &schema)?;
608 if !missing.is_empty() {
609 return Err(DatabaseError::Schema(missing_logs_error(layout.root(), name, &missing)));
610 }
611 for path in present_logs_of(layout, &schema) {
612 fs::File::open(&path).map_err(|error| DatabaseError::Log(path, error))?;
613 }
614 Ok(())
615}
616
617#[allow(clippy::too_many_arguments)]
625fn open_catalog_table(
626 layout: &Layout,
627 name: &str,
628 catalog: &Catalog,
629 entry: &TableEntry,
630 snapshot_every: u64,
631 cipher: Option<&FieldCipher>,
632 follow: Option<FollowPolicy>,
633 readonly: bool,
634 compaction: CatalogCompaction,
635 cache: CachePolicy,
636) -> Result<(Table, RestoreOutcome), DatabaseError> {
637 let schema = table_schema_of(name, entry);
638 let declared = expand_indexes(name, entry);
639 let hash = schema_fingerprint(&schema, &declared);
640 let (mut table, outcome) = Table::open_with_snapshot_and_declared_indexes(
641 schema,
642 layout.log_path(name),
643 layout.index_path(name),
644 &hash,
645 snapshot_every,
646 readonly,
647 &declared,
648 )
649 .map_err(|error| DatabaseError::Table(name.to_string(), TableError::Io(error)))?;
650 apply_image_dir(layout.root(), catalog, entry, &mut table);
651 apply_link_parsing(catalog, entry, &mut table);
652 apply_sync(catalog, entry, &mut table);
653 apply_validation(entry, &mut table);
654 apply_deprecated(entry, &mut table);
655 apply_compaction(compaction, &mut table);
656 table.set_cache(cache);
657 apply_cipher(cipher, &mut table);
658 table.set_follow(follow);
659 table.set_triggers(triggers_of(catalog, name));
660 Ok((table, outcome))
661}
662
663fn first_table_log(layout: &Layout) -> Option<PathBuf> {
667 let has_log_suffix = |path: &Path| {
668 path.extension()
669 .is_some_and(|ext| format!(".{}", ext.to_string_lossy()) == LOG_SUFFIX)
670 };
671 let entries_of = |dir: &Path| -> Vec<PathBuf> {
672 let Ok(read) = fs::read_dir(dir) else { return Vec::new() };
673 let mut found: Vec<PathBuf> = read.filter_map(Result::ok).map(|entry| entry.path()).collect();
674 found.sort();
675 found
676 };
677
678 let top = entries_of(layout.tables());
679 if let Some(log) = top.iter().find(|path| path.is_file() && has_log_suffix(path)) {
680 return Some(log.clone());
681 }
682 top.iter().filter(|path| path.is_dir()).find_map(|dir| {
683 entries_of(dir)
684 .into_iter()
685 .find(|path| path.is_file() && has_log_suffix(path))
686 })
687}
688
689fn refuse_orphan_logs(layout: &Layout) -> Result<(), HaysonSchemaError> {
697 let Some(log) = first_table_log(layout) else {
698 return Ok(());
699 };
700 Err(HaysonSchemaError(format!(
701 "\"{}\" 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",
702 layout.root().display(),
703 layout.metadata_file(),
704 log.display(),
705 layout.metadata_file(),
706 )))
707}
708
709fn read_catalog(path: &Path) -> Result<Option<Catalog>, DatabaseError> {
710 let raw = match fs::read_to_string(path) {
711 Ok(raw) => raw,
712 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(None),
713 Err(error) => return Err(error.into()),
714 };
715 let value: Value = serde_json::from_str(&raw)
716 .map_err(|error| DatabaseError::Catalog(format!("\"{}\" is not valid JSON: {error}", path.display())))?;
717 Ok(Some(parse_catalog(&value)?))
718}
719
720pub(super) struct HeldLock(pub(super) Option<DirLock>);
731
732impl HeldLock {
733 fn into_held(mut self) -> Option<DirLock> {
734 self.0.take()
735 }
736}
737
738impl Drop for HeldLock {
739 fn drop(&mut self) {
740 if let Some(lock) = self.0.as_mut() {
741 lock.release();
742 }
743 }
744}
745
746pub struct Database {
750 layout: Layout,
751 lock: Option<DirLock>,
752 catalog: Catalog,
753 tables: HashMap<String, Table>,
754 restore_outcomes: HashMap<String, RestoreOutcome>,
755 readonly: bool,
756 snapshot_every: u64,
757 compaction: Option<CatalogCompaction>,
760 cipher: Option<FieldCipher>,
761 follow: Option<FollowPolicy>,
762 cache: CachePolicy,
765}
766
767impl Database {
768 pub fn open(dir: &Path) -> Result<Database, DatabaseError> {
769 Database::open_with(dir, OpenOptions::default())
770 }
771
772 pub fn open_with(dir: &Path, options: OpenOptions) -> Result<Database, DatabaseError> {
785 let layout = Database::resolve_layout(dir, &options.layout)?;
786
787 if !options.readonly {
788 for directory in layout.directories() {
789 fs::create_dir_all(directory)?;
790 }
791 }
792
793 let lock = HeldLock(match options.lock {
794 LockMode::Enabled(lock) if !options.readonly => Some(DirLock::acquire(layout.lock_path(), lock)?),
795 LockMode::Enabled(_) | LockMode::Skip => None,
796 });
797
798 Database::open_locked(layout, options, lock)
799 }
800
801 pub fn resolve_layout(dir: &Path, options: &LayoutOptions) -> Result<Layout, DatabaseError> {
815 let metadata_file = options.metadata_file()?.unwrap_or(DEFAULT_METADATA_FILE);
816 let stored = stored_kind_of(&dir.join(metadata_file));
817 let kind = resolve_kind(dir, stored, options.kind)?;
818 Ok(Layout::new(dir, kind, metadata_file))
819 }
820
821 fn open_locked(layout: Layout, options: OpenOptions, lock: HeldLock) -> Result<Database, DatabaseError> {
822 let found = read_catalog(layout.metadata_path())?;
823 let catalog = match found {
824 Some(catalog) => catalog,
825 None => {
826 refuse_orphan_logs(&layout)?;
827 let catalog = empty_catalog(CatalogLayout::from(layout.kind()), now_ms());
828 if !options.readonly {
829 write_catalog(layout.metadata_path(), &catalog)
830 .map_err(|error| DatabaseError::Catalog(error.to_string()))?;
831 }
832 catalog
833 }
834 };
835
836 let cipher = options.cypher.clone().map(FieldCipher::new).transpose()?;
837 let follow = options.follow.resolve(options.readonly);
838
839 let mut tables = HashMap::new();
840 let mut restore_outcomes = HashMap::new();
841 for (name, entry) in &catalog.tables {
842 if options.readonly {
843 refuse_unreadable_logs(&layout, name, entry)?;
844 }
845 let (table, outcome) = open_catalog_table(
846 &layout,
847 name,
848 &catalog,
849 entry,
850 options.snapshot_every,
851 cipher.as_ref(),
852 follow,
853 options.readonly,
854 compaction_of(options.compaction, &catalog, Some(entry)),
855 options.cache,
856 )?;
857 tables.insert(name.clone(), table);
858 restore_outcomes.insert(name.clone(), outcome);
859 }
860
861 if !options.readonly {
862 recover_incomplete(&layout.journal_path(), &mut tables)?;
863 }
864
865 sync_table_triggers(&catalog, &mut tables);
866
867 Ok(Database {
868 layout,
869 lock: lock.into_held(),
870 catalog,
871 tables,
872 restore_outcomes,
873 readonly: options.readonly,
874 snapshot_every: options.snapshot_every,
875 compaction: options.compaction,
876 cipher,
877 follow,
878 cache: options.cache,
879 })
880 }
881
882 pub fn dir(&self) -> &Path {
883 self.layout.root()
884 }
885
886 pub fn layout(&self) -> &Layout {
889 &self.layout
890 }
891
892 pub fn follow(&self) -> Option<FollowPolicy> {
895 self.follow
896 }
897
898 pub fn table_names(&self) -> Vec<&str> {
899 let mut names: Vec<&str> = self.tables.keys().map(String::as_str).collect();
900 names.sort_unstable();
901 names
902 }
903
904 pub fn table(&self, name: &str) -> Option<&Table> {
905 self.tables.get(name)
906 }
907
908 pub fn table_mut(&mut self, name: &str) -> Option<&mut Table> {
914 let _ = self.renew_lock();
915 self.tables.get_mut(name)
916 }
917
918 pub fn tick(&self) -> Result<(), DatabaseError> {
923 self.renew_lock()
924 }
925
926 fn renew_lock(&self) -> Result<(), DatabaseError> {
927 match &self.lock {
928 Some(lock) => Ok(lock.tick()?),
929 None => Ok(()),
930 }
931 }
932
933 pub fn restore_outcome(&self, name: &str) -> Option<RestoreOutcome> {
936 self.restore_outcomes.get(name).copied()
937 }
938
939 pub fn catalog_unchecked(&self) -> &Catalog {
943 &self.catalog
944 }
945
946 pub fn catalog(&self) -> Result<&Catalog, DatabaseError> {
954 let drifted = self.drifted_tables();
955 if drifted.is_empty() {
956 return Ok(&self.catalog);
957 }
958 Err(DatabaseError::Schema(self.drift_error(&drifted)))
959 }
960
961 pub fn describe_table(&self, name: &str) -> Result<Value, DatabaseError> {
973 if let Some(drift) = self.drift_of(name) {
974 let drifted = [DriftedTable {
975 name: name.to_string(),
976 drift,
977 }];
978 return Err(DatabaseError::Schema(self.drift_error(&drifted)));
979 }
980 let table = self.tables.get(name).ok_or_else(|| {
981 DatabaseError::Schema(HaysonSchemaError(format!("the catalog declares no table \"{name}\"")))
982 })?;
983 Ok(describe_table_document(table))
984 }
985
986 fn drift_of(&self, name: &str) -> Option<TableDrift> {
987 let table = self.tables.get(name)?;
988 let Some(entry) = self.catalog.tables.get(name) else {
989 return Some(TableDrift::Absent);
990 };
991 if shape_of(&table_schema_of(name, entry)) != shape_of(table.schema()) {
992 return Some(TableDrift::Schema {
993 revision: entry.revision,
994 });
995 }
996 let declared = expand_indexes(name, entry);
997 let served = table.declared_indexes();
998 let unserved = index_names_missing_from(&declared, served);
999 let stale = index_names_missing_from(served, &declared);
1000 if unserved.is_empty() && stale.is_empty() {
1001 return None;
1002 }
1003 Some(TableDrift::Indexes { unserved, stale })
1004 }
1005
1006 fn drifted_tables(&self) -> Vec<DriftedTable> {
1007 let mut names: Vec<&String> = self.tables.keys().collect();
1008 names.sort();
1009 names
1010 .into_iter()
1011 .filter_map(|name| {
1012 self.drift_of(name).map(|drift| DriftedTable {
1013 name: name.clone(),
1014 drift,
1015 })
1016 })
1017 .collect()
1018 }
1019
1020 fn drift_error(&self, drifted: &[DriftedTable]) -> HaysonSchemaError {
1021 let detail = drifted
1022 .iter()
1023 .map(|DriftedTable { name, drift }| {
1024 let revision = self.tables.get(name).map(|table| table.schema().revision);
1025 let open = revision.map_or_else(|| "not open".to_string(), |at| format!("revision {at}"));
1026 format!(
1027 "\"{name}\" is open here at {open} while the catalog {}",
1028 stored_side_of(drift)
1029 )
1030 })
1031 .collect::<Vec<_>>()
1032 .join("; ");
1033 HaysonSchemaError(format!(
1034 "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",
1035 self.layout.root().display()
1036 ))
1037 }
1038
1039 pub fn relations(&self, table: Option<&str>) -> Vec<Relation> {
1043 relations_touching(&self.catalog, table)
1044 }
1045
1046 pub fn check_integrity(&self) -> io::Result<IntegrityReport> {
1049 let refs: HashMap<String, &Table> = self.tables.iter().map(|(name, table)| (name.clone(), table)).collect();
1050 check_integrity(&self.catalog, &refs)
1051 }
1052
1053 pub fn refresh(&mut self) -> Result<bool, DatabaseError> {
1063 let mut changed = false;
1064 if let Some(catalog) = read_catalog(self.layout.metadata_path())? {
1065 if catalog != self.catalog {
1066 self.catalog = catalog;
1067 changed = true;
1068 }
1069 }
1070
1071 let missing: Vec<String> = self
1072 .catalog
1073 .tables
1074 .keys()
1075 .filter(|name| !self.tables.contains_key(*name))
1076 .cloned()
1077 .collect();
1078 for name in missing {
1079 let Some(entry) = self.catalog.tables.get(&name) else {
1080 continue;
1081 };
1082 let (table, outcome) = open_catalog_table(
1083 &self.layout,
1084 &name,
1085 &self.catalog,
1086 entry,
1087 self.snapshot_every,
1088 self.cipher.as_ref(),
1089 self.follow,
1090 self.readonly,
1091 compaction_of(self.compaction, &self.catalog, Some(entry)),
1092 self.cache,
1093 )?;
1094 self.tables.insert(name.clone(), table);
1095 self.restore_outcomes.insert(name, outcome);
1096 changed = true;
1097 }
1098
1099 for (name, table) in self.tables.iter_mut() {
1100 let outcome = table
1101 .refresh()
1102 .map_err(|error| DatabaseError::Table(name.clone(), TableError::Io(error)))?;
1103 changed |= outcome.changed();
1104 }
1105
1106 sync_table_triggers(&self.catalog, &mut self.tables);
1108 Ok(changed)
1109 }
1110
1111 pub fn unit_of_work(&mut self) -> UnitOfWork {
1114 UnitOfWork::new()
1115 }
1116
1117 pub fn commit(&mut self, unit: &mut UnitOfWork) -> Result<UnitOfWorkResult, DatabaseError> {
1124 let prepared = self.prepare_commit(unit)?;
1125 unit.journal(&prepared, &self.layout.journal_path())?;
1126 self.apply_commit(prepared)
1127 }
1128
1129 pub fn prepare_commit(&mut self, unit: &mut UnitOfWork) -> Result<PreparedUnit, DatabaseError> {
1138 if self.readonly {
1139 return Err(DatabaseError::ReadOnly {
1140 dir: self.layout.root().to_path_buf(),
1141 operation: "commit".to_string(),
1142 });
1143 }
1144 self.renew_lock()?;
1145 Ok(unit.prepare(&mut self.tables)?)
1146 }
1147
1148 pub fn apply_commit(&mut self, prepared: PreparedUnit) -> Result<UnitOfWorkResult, DatabaseError> {
1152 let journal_path = self.layout.journal_path();
1153 Ok(apply_prepared(prepared, &mut self.tables, &journal_path)?)
1154 }
1155
1156 fn assert_writable(&self, operation: &str) -> Result<(), DatabaseError> {
1160 if self.readonly {
1161 return Err(DatabaseError::ReadOnly {
1162 dir: self.layout.root().to_path_buf(),
1163 operation: operation.to_string(),
1164 });
1165 }
1166 self.renew_lock()
1167 }
1168
1169 fn write_catalog_to_disk(&self, catalog: &Catalog) -> Result<(), DatabaseError> {
1170 write_catalog(self.layout.metadata_path(), catalog).map_err(catalog_write_error)
1171 }
1172
1173 pub fn create_table(&mut self, name: &str, schema_spec: Value) -> Result<(), DatabaseError> {
1188 self.assert_writable("createTable")?;
1189 let schema = parse_schema(&schema_spec)?;
1190 if schema.table != name {
1191 return Err(DatabaseError::Schema(HaysonSchemaError(format!(
1192 "schema declares table \"{}\" but it is being created as \"{name}\" — name them the same",
1193 schema.table
1194 ))));
1195 }
1196 let now = now_ms();
1197 let next_catalog = crate::catalog::build::create_table(&self.catalog, name, &schema, now);
1198 self.write_catalog_to_disk(&next_catalog)?;
1199
1200 let next_entry = &next_catalog.tables[name];
1201 let (table, outcome) = open_catalog_table(
1202 &self.layout,
1203 name,
1204 &next_catalog,
1205 next_entry,
1206 self.snapshot_every,
1207 self.cipher.as_ref(),
1208 self.follow,
1209 self.readonly,
1210 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1211 self.cache,
1212 )?;
1213
1214 self.catalog = next_catalog;
1215 self.tables.insert(name.to_string(), table);
1216 self.restore_outcomes.insert(name.to_string(), outcome);
1217 Ok(())
1218 }
1219
1220 pub fn drop_field(&mut self, table: &str, field: &str, force: bool) -> Result<(), DatabaseError> {
1233 self.assert_writable("dropField")?;
1234 self.flush_table(table)?;
1235 let existing = self.tables.get(table).ok_or_else(|| {
1236 DatabaseError::Catalog(format!(
1237 "table \"{table}\" does not exist in \"{}\"",
1238 self.layout.root().display()
1239 ))
1240 })?;
1241
1242 let now = now_ms();
1243 let (next_catalog, next_entry) = crate::catalog::build::drop_field(&self.catalog, table, field, now)?;
1244
1245 if !force {
1246 assert_no_residual_values(table, existing, field)?;
1247 }
1248
1249 self.write_catalog_to_disk(&next_catalog)?;
1250
1251 let (mut table_handle, _) = open_catalog_table(
1252 &self.layout,
1253 table,
1254 &next_catalog,
1255 &next_entry,
1256 self.snapshot_every,
1257 self.cipher.as_ref(),
1258 self.follow,
1259 self.readonly,
1260 compaction_of(self.compaction, &next_catalog, Some(&next_entry)),
1261 self.cache,
1262 )?;
1263 table_handle
1265 .snapshot()
1266 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1267
1268 self.catalog = next_catalog;
1269 self.tables.insert(table.to_string(), table_handle);
1270 Ok(())
1271 }
1272
1273 pub fn declare_index(&mut self, table: &str, index_spec: Value) -> Result<(), DatabaseError> {
1282 self.assert_writable("declareIndex")?;
1283 self.flush_table(table)?;
1284 let existing = self.tables.get(table).ok_or_else(|| {
1285 DatabaseError::Catalog(format!(
1286 "table \"{table}\" does not exist in \"{}\"",
1287 self.layout.root().display()
1288 ))
1289 })?;
1290
1291 let now = now_ms();
1292 let next_catalog = crate::catalog::declare_index(&self.catalog, table, &index_spec, existing, now)?;
1293 self.write_catalog_to_disk(&next_catalog)?;
1294
1295 let next_entry = &next_catalog.tables[table];
1296 let (mut table_handle, _) = open_catalog_table(
1297 &self.layout,
1298 table,
1299 &next_catalog,
1300 next_entry,
1301 self.snapshot_every,
1302 self.cipher.as_ref(),
1303 self.follow,
1304 self.readonly,
1305 compaction_of(self.compaction, &next_catalog, Some(next_entry)),
1306 self.cache,
1307 )?;
1308 table_handle
1310 .snapshot()
1311 .map_err(|error| DatabaseError::Table(table.to_string(), TableError::Io(error)))?;
1312
1313 self.catalog = next_catalog;
1314 self.tables.insert(table.to_string(), table_handle);
1315 Ok(())
1316 }
1317
1318 pub fn declare_trigger(&mut self, trigger_spec: Value) -> Result<(), DatabaseError> {
1324 self.assert_writable("declareTrigger")?;
1325
1326 let now = now_ms();
1327 let next_catalog = crate::catalog::declare_trigger(&self.catalog, &trigger_spec, now)?;
1328 self.write_catalog_to_disk(&next_catalog)?;
1329
1330 self.catalog = next_catalog;
1331 sync_table_triggers(&self.catalog, &mut self.tables);
1332 Ok(())
1333 }
1334
1335 pub fn flush(&mut self) -> Result<(), DatabaseError> {
1339 for (name, table) in &mut self.tables {
1340 table
1341 .flush()
1342 .map_err(|error| DatabaseError::Table(name.clone(), error))?;
1343 }
1344 Ok(())
1345 }
1346
1347 fn flush_table(&mut self, table: &str) -> Result<(), DatabaseError> {
1351 let Some(handle) = self.tables.get_mut(table) else {
1352 return Ok(());
1353 };
1354 handle
1355 .flush()
1356 .map_err(|error| DatabaseError::Table(table.to_string(), error))
1357 }
1358
1359 pub fn close(mut self) -> io::Result<()> {
1364 if !self.readonly {
1365 for table in self.tables.values_mut() {
1366 table.flush()?;
1368 if table.needs_snapshot() {
1369 table.snapshot()?;
1370 }
1371 }
1372 }
1373 if let Some(mut lock) = self.lock.take() {
1374 lock.release();
1375 }
1376 Ok(())
1377 }
1378}
1379
1380#[cfg(test)]
1381mod tests {
1382 use serde_json::json;
1383 use tempfile::tempdir;
1384
1385 use super::*;
1386 use crate::catalog::create_table;
1387 use crate::catalog::types::{CatalogIndexKind, IndexEntry};
1388 use crate::checkpoint::write_snapshot;
1389 use crate::registry::table_registry::TableRegistry;
1390 use crate::schema::parse_schema;
1391 use crate::storage::{serialize_entry, LogMeta, LogOp};
1392 use crate::txn::journal::{write_journal, Journal, JournalEntry, JournalHeader, JOURNAL_VERSION};
1393 use crate::txn::JOURNAL_FILE;
1394
1395 const MANIFEST_FILE: &str = DEFAULT_METADATA_FILE;
1396
1397 fn flat(dir: &Path) -> Layout {
1398 Layout::new(dir, crate::database::LayoutKind::Flat, MANIFEST_FILE)
1399 }
1400
1401 fn log_path(dir: &Path, table: &str) -> std::path::PathBuf {
1402 flat(dir).log_path(table)
1403 }
1404
1405 #[test]
1409 fn a_database_and_its_tables_can_be_shared_across_threads() {
1410 fn shareable<T: Send + Sync>() {}
1411 shareable::<Database>();
1412 shareable::<Table>();
1413 shareable::<crate::registry::fulltext_index::FulltextIndex>();
1414 }
1415
1416 fn index_path(dir: &Path, table: &str) -> std::path::PathBuf {
1417 flat(dir).index_path(table)
1418 }
1419
1420 fn users_schema() -> TableSchema {
1421 parse_schema(&json!({
1422 "schemaVersion": 1,
1423 "table": "users",
1424 "primaryKey": "id",
1425 "fields": {
1426 "id": { "type": "string" },
1427 "name": { "type": "string", "required": true },
1428 },
1429 }))
1430 .unwrap()
1431 }
1432
1433 fn orders_schema() -> TableSchema {
1434 parse_schema(&json!({
1435 "schemaVersion": 1,
1436 "table": "orders",
1437 "primaryKey": "id",
1438 "fields": {
1439 "id": { "type": "string" },
1440 "userId": { "type": "string", "required": true },
1441 },
1442 }))
1443 .unwrap()
1444 }
1445
1446 fn catalog_of(schemas: &[TableSchema]) -> Catalog {
1447 let mut catalog = empty_catalog(CatalogLayout::Flat, 1_000);
1448 for schema in schemas {
1449 catalog = create_table(&catalog, &schema.table, schema, 1_000);
1450 }
1451 catalog
1452 }
1453
1454 fn write_catalog_to(dir: &Path, catalog: &Catalog) {
1455 write_catalog(&dir.join(MANIFEST_FILE), catalog).unwrap();
1456 }
1457
1458 #[test]
1459 fn open_builds_a_database_with_every_declared_table_available() {
1460 let dir = tempdir().unwrap();
1461 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1462
1463 let mut db = Database::open(dir.path()).unwrap();
1464 assert_eq!(db.table_names(), vec!["orders", "users"]);
1465
1466 db.table_mut("users")
1467 .unwrap()
1468 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1469 .unwrap();
1470 assert_eq!(
1471 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1472 Some(&json!("Ania"))
1473 );
1474 db.close().unwrap();
1475 }
1476
1477 #[test]
1478 fn open_restores_the_registry_from_a_paired_snapshot_instead_of_scanning_the_log() {
1479 let dir = tempdir().unwrap();
1480 let catalog = catalog_of(&[users_schema()]);
1481 write_catalog_to(dir.path(), &catalog);
1482
1483 let schema = table_schema_of("users", &catalog.tables["users"]);
1484 let hash = schema_fingerprint(&schema, &expand_indexes("users", &catalog.tables["users"]));
1485 let log_path = log_path(dir.path(), "users");
1486 let mut log = crate::storage::LogFile::open(&log_path, false).unwrap();
1487 log.append(&[serialize_entry(
1488 &LogMeta {
1489 key: "u1".into(),
1490 op: LogOp::Put,
1491 seq: 1,
1492 ts: 1,
1493 },
1494 Some(&{
1495 let mut record = crate::schema::HaysonRecord::new();
1496 record.insert("id".into(), json!("u1"));
1497 record.insert("name".into(), json!("Ania"));
1498 record
1499 }),
1500 )])
1501 .unwrap();
1502 let registry = TableRegistry::build_from_log(schema.clone(), &log).unwrap();
1503 write_snapshot(&index_path(dir.path(), "users"), ®istry, "users", &hash, 0, &log).unwrap();
1504
1505 let db = Database::open(dir.path()).unwrap();
1506 assert_eq!(db.restore_outcome("users"), Some(RestoreOutcome::Restored));
1507 assert_eq!(
1508 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1509 Some(&json!("Ania"))
1510 );
1511 db.close().unwrap();
1512 }
1513
1514 #[test]
1515 fn open_finishes_an_unfinished_journal_before_returning() {
1516 let dir = tempdir().unwrap();
1517 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
1518
1519 {
1520 let mut users = Table::open(users_schema(), log_path(dir.path(), "users")).unwrap();
1521 let mut orders = Table::open(orders_schema(), log_path(dir.path(), "orders")).unwrap();
1522 let user_entries = users
1523 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "u1", "name": "Ania" }))])
1524 .unwrap();
1525 let order_entries = orders
1526 .prepare_txn_ops(&[crate::table::TxnOp::Put(json!({ "id": "o1", "userId": "u1" }))])
1527 .unwrap();
1528
1529 let mut entries = Vec::new();
1530 for entry in &user_entries {
1531 entries.push(JournalEntry {
1532 t: "users".into(),
1533 s: entry.seq(),
1534 l: entry.line().to_string(),
1535 });
1536 }
1537 for entry in &order_entries {
1538 entries.push(JournalEntry {
1539 t: "orders".into(),
1540 s: entry.seq(),
1541 l: entry.line().to_string(),
1542 });
1543 }
1544 write_journal(
1545 &dir.path().join(JOURNAL_FILE),
1546 &Journal {
1547 header: JournalHeader {
1548 journal_version: JOURNAL_VERSION,
1549 kind: "unit".into(),
1550 id: "crash".into(),
1551 started_at: 0,
1552 tables: vec!["users".into(), "orders".into()],
1553 },
1554 entries,
1555 },
1556 )
1557 .unwrap();
1558 }
1560
1561 assert!(dir.path().join(JOURNAL_FILE).exists());
1562 let db = Database::open(dir.path()).unwrap();
1563 assert!(!dir.path().join(JOURNAL_FILE).exists());
1564 assert_eq!(
1565 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1566 Some(&json!("Ania"))
1567 );
1568 assert_eq!(
1569 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
1570 Some(&json!("u1"))
1571 );
1572 db.close().unwrap();
1573 }
1574
1575 #[test]
1576 fn a_second_open_of_the_same_directory_is_denied_while_the_first_still_holds_the_lock() {
1577 let dir = tempdir().unwrap();
1578 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1579
1580 let _first = Database::open(dir.path()).unwrap();
1581 let second = Database::open_with(
1582 dir.path(),
1583 OpenOptions {
1584 readonly: false,
1585 lock: LockMode::Enabled(LockOptions {
1586 timeout_ms: Some(0),
1587 ..Default::default()
1588 }),
1589 ..Default::default()
1590 },
1591 );
1592 assert!(matches!(second, Err(DatabaseError::Lock(_))));
1593 }
1594
1595 fn lock_path_of(dir: &Path) -> std::path::PathBuf {
1596 dir.join(crate::lock::LOCK_FILE)
1597 }
1598
1599 fn open_refusing(dir: &Path) -> DatabaseError {
1600 Database::open_with(
1601 dir,
1602 OpenOptions {
1603 lock: LockMode::Enabled(LockOptions {
1604 timeout_ms: Some(0),
1605 ..Default::default()
1606 }),
1607 ..Default::default()
1608 },
1609 )
1610 .map(|_| ())
1611 .expect_err("this directory must not open")
1612 }
1613
1614 #[test]
1615 fn a_directory_with_logs_but_no_catalog_is_refused_instead_of_opening_empty() {
1616 let dir = tempdir().unwrap();
1617 let mut db = Database::open(dir.path()).unwrap();
1618 db.create_table(
1619 "users",
1620 json!({
1621 "schemaVersion": 1,
1622 "table": "users",
1623 "primaryKey": "id",
1624 "fields": {
1625 "id": { "type": "string" },
1626 "name": { "type": "string", "required": true },
1627 },
1628 }),
1629 )
1630 .unwrap();
1631 db.table_mut("users")
1632 .unwrap()
1633 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1634 .unwrap();
1635 db.close().unwrap();
1636
1637 let log = fs::read(dir.path().join("users.jsonl")).unwrap();
1638 fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
1639
1640 let refusal = open_refusing(dir.path()).to_string();
1641 assert!(
1642 refusal.contains("users.jsonl"),
1643 "the refusal has to name the log it found: {refusal}"
1644 );
1645 assert!(
1646 refusal.contains(MANIFEST_FILE),
1647 "the refusal has to name the catalog: {refusal}"
1648 );
1649
1650 assert!(
1652 !dir.path().join(MANIFEST_FILE).exists(),
1653 "a refused open must not leave a catalog"
1654 );
1655 assert_eq!(
1656 fs::read(dir.path().join("users.jsonl")).unwrap(),
1657 log,
1658 "the log must be untouched"
1659 );
1660 }
1661
1662 #[test]
1663 fn an_empty_directory_and_one_holding_no_logs_still_open_as_a_new_database() {
1664 let empty = tempdir().unwrap();
1665 assert!(
1666 Database::open(empty.path()).is_ok(),
1667 "an empty directory is still a new database"
1668 );
1669
1670 let littered = tempdir().unwrap();
1671 fs::write(littered.path().join("README.md"), "# not a table\n").unwrap();
1672 fs::create_dir(littered.path().join("users-backup")).unwrap();
1673 assert!(Database::open(littered.path()).is_ok(), "nothing here is a table log");
1674 }
1675
1676 #[test]
1677 fn the_logs_of_a_partitioned_table_are_found_one_level_down_too() {
1678 let dir = tempdir().unwrap();
1679 fs::create_dir_all(dir.path().join("notes")).unwrap();
1680 fs::write(dir.path().join("notes").join("note.jsonl"), "").unwrap();
1681
1682 let refusal = open_refusing(dir.path()).to_string();
1683 assert!(
1684 refusal.contains("note.jsonl"),
1685 "a partitioned log still counts as data: {refusal}"
1686 );
1687 }
1688
1689 #[test]
1693 fn a_directory_refused_at_open_is_left_unlocked_whichever_way_the_catalog_was_broken() {
1694 for (case, document) in [
1695 ("not JSON at all", "{ this is not json".to_string()),
1696 (
1697 "no formatVersion",
1698 json!({ "catalogVersion": 2, "tables": {} }).to_string(),
1699 ),
1700 (
1701 "the format-1 marker",
1702 json!({ "hayson": 1, "formatVersion": 2, "tables": {} }).to_string(),
1703 ),
1704 (
1705 "a formatVersion from the future",
1706 json!({ "catalogVersion": 2, "formatVersion": 3, "tables": {} }).to_string(),
1707 ),
1708 ] {
1709 let dir = tempdir().unwrap();
1710 std::fs::write(dir.path().join(MANIFEST_FILE), &document).unwrap();
1711
1712 open_refusing(dir.path());
1713
1714 assert!(
1715 !lock_path_of(dir.path()).exists(),
1716 "{case}: the refused open left its lock file behind"
1717 );
1718 }
1719 }
1720
1721 #[test]
1724 fn a_directory_opens_normally_after_an_open_that_was_refused_a_broken_catalog() {
1725 let dir = tempdir().unwrap();
1726 std::fs::write(dir.path().join(MANIFEST_FILE), "{ this is not json").unwrap();
1727
1728 open_refusing(dir.path());
1729
1730 std::fs::remove_file(dir.path().join(MANIFEST_FILE)).unwrap();
1731 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1732
1733 let mut db = Database::open(dir.path()).unwrap();
1734 db.table_mut("users")
1735 .unwrap()
1736 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1737 .unwrap();
1738 db.close().unwrap();
1739 }
1740
1741 #[test]
1744 fn a_directory_refused_for_a_bad_cipher_key_is_left_unlocked_too() {
1745 let dir = tempdir().unwrap();
1746 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1747
1748 let error = Database::open_with(
1749 dir.path(),
1750 OpenOptions {
1751 cypher: Some(CipherOptions {
1752 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
1753 key_id: None,
1754 on_missing_key: None,
1755 }),
1756 ..OpenOptions::default()
1757 },
1758 )
1759 .map(|_| ())
1760 .unwrap_err();
1761
1762 assert!(matches!(error, DatabaseError::Cipher(_)));
1763 assert!(!lock_path_of(dir.path()).exists());
1764 Database::open(dir.path()).unwrap().close().unwrap();
1765 }
1766
1767 #[test]
1770 fn an_open_that_succeeded_still_holds_the_lock_until_the_handle_closes() {
1771 let dir = tempdir().unwrap();
1772 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1773
1774 let held = Database::open(dir.path()).unwrap();
1775 assert!(lock_path_of(dir.path()).exists());
1776 assert!(matches!(open_refusing(dir.path()), DatabaseError::Lock(_)));
1777
1778 held.close().unwrap();
1779 assert!(!lock_path_of(dir.path()).exists());
1780 }
1781
1782 fn open_skipping_the_lock(dir: &Path) -> Database {
1785 Database::open_with(
1786 dir,
1787 OpenOptions {
1788 lock: LockMode::Skip,
1789 ..Default::default()
1790 },
1791 )
1792 .unwrap()
1793 }
1794
1795 fn declared_on_disk(dir: &Path, table: &str) -> Vec<String> {
1796 let catalog = read_catalog(&dir.join(MANIFEST_FILE))
1797 .unwrap()
1798 .expect("a catalog on disk");
1799 let mut names: Vec<String> = catalog.tables[table]
1800 .indexes
1801 .iter()
1802 .flatten()
1803 .map(|index| index.name.clone())
1804 .collect();
1805 names.sort();
1806 names
1807 }
1808
1809 fn two_tables_with_records(dir: &Path) {
1813 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
1814 let mut db = Database::open(dir).unwrap();
1815 db.table_mut("users")
1816 .unwrap()
1817 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1818 .unwrap();
1819 db.table_mut("users")
1820 .unwrap()
1821 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
1822 .unwrap();
1823 db.close().unwrap();
1824 }
1825
1826 #[test]
1827 fn a_second_writer_that_skips_the_lock_opens_and_writes_beside_the_one_holding_it() {
1828 let dir = tempdir().unwrap();
1829 two_tables_with_records(dir.path());
1830
1831 let mut locked = Database::open(dir.path()).unwrap();
1832 let mut unlocked = open_skipping_the_lock(dir.path());
1833
1834 locked
1836 .table_mut("users")
1837 .unwrap()
1838 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
1839 .unwrap();
1840 unlocked
1841 .table_mut("orders")
1842 .unwrap()
1843 .put(&json!({ "id": "o1", "userId": "u1" }), false)
1844 .unwrap();
1845
1846 let third = Database::open_with(
1848 dir.path(),
1849 OpenOptions {
1850 lock: LockMode::Enabled(LockOptions {
1851 timeout_ms: Some(0),
1852 ..Default::default()
1853 }),
1854 ..Default::default()
1855 },
1856 );
1857 assert!(matches!(third, Err(DatabaseError::Lock(_))));
1858
1859 unlocked.close().unwrap();
1860 locked.close().unwrap();
1861 }
1862
1863 #[test]
1868 fn a_lock_skipping_handle_that_never_refreshed_loses_the_declaration_of_the_locked_one() {
1869 let dir = tempdir().unwrap();
1870 two_tables_with_records(dir.path());
1871
1872 let mut locked = Database::open(dir.path()).unwrap();
1873 let mut unlocked = open_skipping_the_lock(dir.path());
1874
1875 locked
1876 .declare_index(
1877 "users",
1878 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
1879 )
1880 .unwrap();
1881 unlocked
1882 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
1883 .unwrap();
1884
1885 unlocked.close().unwrap();
1886 locked.close().unwrap();
1887
1888 assert_eq!(declared_on_disk(dir.path(), "users"), vec!["by_id".to_string()]);
1889 }
1890
1891 #[test]
1892 fn a_lock_skipping_handle_that_refreshes_first_keeps_both_declarations() {
1893 let dir = tempdir().unwrap();
1894 two_tables_with_records(dir.path());
1895
1896 let mut locked = Database::open(dir.path()).unwrap();
1897 let mut unlocked = open_skipping_the_lock(dir.path());
1898
1899 locked
1900 .declare_index(
1901 "users",
1902 json!({ "name": "by_name", "fields": ["name"], "kind": "exact" }),
1903 )
1904 .unwrap();
1905 assert!(unlocked.refresh().unwrap(), "the catalog changed under it");
1906 unlocked
1907 .declare_index("users", json!({ "name": "by_id", "fields": ["id"], "kind": "sorted" }))
1908 .unwrap();
1909
1910 unlocked.close().unwrap();
1911 locked.close().unwrap();
1912
1913 assert_eq!(
1914 declared_on_disk(dir.path(), "users"),
1915 vec!["by_id".to_string(), "by_name".to_string()]
1916 );
1917 }
1918
1919 fn open_follower(dir: &Path) -> Database {
1922 Database::open_with(
1923 dir,
1924 OpenOptions {
1925 readonly: true,
1926 lock: LockMode::default(),
1927 ..Default::default()
1928 },
1929 )
1930 .unwrap()
1931 }
1932
1933 #[test]
1934 fn refresh_with_nothing_new_on_disk_reports_no_change() {
1935 let dir = tempdir().unwrap();
1936 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1937 let mut writer = Database::open(dir.path()).unwrap();
1938 writer
1939 .table_mut("users")
1940 .unwrap()
1941 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1942 .unwrap();
1943 let mut follower = open_follower(dir.path());
1944 follower.refresh().unwrap();
1945
1946 assert!(!follower.refresh().unwrap());
1947 writer.close().unwrap();
1948 }
1949
1950 #[test]
1951 fn refresh_picks_up_both_a_table_and_the_records_another_process_wrote() {
1952 let dir = tempdir().unwrap();
1953 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
1954 let mut writer = Database::open(dir.path()).unwrap();
1955 let mut follower = open_follower(dir.path());
1956 assert_eq!(follower.table_names(), vec!["users"]);
1957
1958 writer
1959 .table_mut("users")
1960 .unwrap()
1961 .put(&json!({ "id": "u1", "name": "Ania" }), false)
1962 .unwrap();
1963 writer
1964 .create_table(
1965 "orders",
1966 json!({
1967 "schemaVersion": 1,
1968 "table": "orders",
1969 "primaryKey": "id",
1970 "fields": { "id": { "type": "string" }, "userId": { "type": "string", "required": true } },
1971 }),
1972 )
1973 .unwrap();
1974 writer
1975 .table_mut("orders")
1976 .unwrap()
1977 .put(&json!({ "id": "o1", "userId": "u1" }), false)
1978 .unwrap();
1979
1980 assert!(follower.refresh().unwrap());
1981
1982 assert_eq!(follower.table_names(), vec!["orders", "users"]);
1983 assert_eq!(
1984 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
1985 Some(&json!("Ania"))
1986 );
1987 assert_eq!(
1988 follower
1989 .table("orders")
1990 .unwrap()
1991 .get("o1")
1992 .unwrap()
1993 .unwrap()
1994 .get("userId"),
1995 Some(&json!("u1"))
1996 );
1997 writer.close().unwrap();
1998 }
1999
2000 #[test]
2001 fn refresh_survives_a_compaction_run_by_the_writer_process() {
2002 let dir = tempdir().unwrap();
2003 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2004 let mut writer = Database::open(dir.path()).unwrap();
2005 let users = writer.table_mut("users").unwrap();
2006 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
2007 users.put(&json!({ "id": "u2", "name": "Piotr" }), false).unwrap();
2008 users.patch("u1", &json!({ "name": "Anna" }), false).unwrap();
2009 let mut follower = open_follower(dir.path());
2010 assert_eq!(follower.table("users").unwrap().len(), 2);
2011
2012 writer.table_mut("users").unwrap().compact().unwrap();
2013 assert!(follower.refresh().unwrap());
2014
2015 assert_eq!(follower.table("users").unwrap().len(), 2);
2016 assert_eq!(
2017 follower.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2018 Some(&json!("Anna"))
2019 );
2020 assert_eq!(follower.table("users").unwrap().stats().gen, 1);
2021 writer.close().unwrap();
2022 }
2023
2024 #[test]
2025 fn unit_of_work_commits_across_two_tables_and_the_database_sees_both() {
2026 let dir = tempdir().unwrap();
2027 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
2028 let mut db = Database::open(dir.path()).unwrap();
2029
2030 let mut unit = db.unit_of_work();
2031 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
2032 unit.put("orders", json!({ "id": "o1", "userId": "u1" }));
2033 let result = db.commit(&mut unit).unwrap();
2034 assert_eq!(result.operations, 2);
2035
2036 assert_eq!(
2037 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2038 Some(&json!("Ania"))
2039 );
2040 assert_eq!(
2041 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2042 Some(&json!("u1"))
2043 );
2044 db.close().unwrap();
2045 }
2046
2047 #[test]
2048 fn create_table_on_an_open_database_adds_a_new_working_table_immediately() {
2049 let dir = tempdir().unwrap();
2050 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2051 let mut db = Database::open(dir.path()).unwrap();
2052
2053 db.create_table(
2054 "orders",
2055 json!({
2056 "schemaVersion": 1,
2057 "table": "orders",
2058 "primaryKey": "id",
2059 "fields": {
2060 "id": { "type": "string" },
2061 "userId": { "type": "string", "required": true },
2062 },
2063 }),
2064 )
2065 .unwrap();
2066
2067 assert_eq!(db.table_names(), vec!["orders", "users"]);
2068 assert!(db.catalog.tables.contains_key("orders"));
2069
2070 db.table_mut("orders")
2071 .unwrap()
2072 .put(&json!({ "id": "o1", "userId": "u1" }), false)
2073 .unwrap();
2074 assert_eq!(
2075 db.table("orders").unwrap().get("o1").unwrap().unwrap().get("userId"),
2076 Some(&json!("u1"))
2077 );
2078
2079 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2080 assert!(reread.tables.contains_key("orders"));
2081 db.close().unwrap();
2082 }
2083
2084 #[test]
2085 fn create_table_refuses_a_schema_naming_a_table_other_than_the_one_being_created() {
2086 let dir = tempdir().unwrap();
2087 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2088 let mut db = Database::open(dir.path()).unwrap();
2089 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2090
2091 let error = db
2092 .create_table(
2093 "orders",
2094 json!({
2095 "schemaVersion": 1,
2096 "table": "invoices",
2097 "primaryKey": "id",
2098 "fields": { "id": { "type": "string" } },
2099 }),
2100 )
2101 .expect_err("a schema naming another table is refused");
2102
2103 assert!(matches!(error, DatabaseError::Schema(_)), "got {error:?}");
2104 let message = error.to_string();
2105 assert!(
2106 message.contains("invoices"),
2107 "the refusal names the schema's table: {message}"
2108 );
2109 assert!(
2110 message.contains("orders"),
2111 "the refusal names the key it was created under: {message}"
2112 );
2113
2114 assert_eq!(db.table_names(), vec!["users"]);
2115 assert!(!db.catalog.tables.contains_key("orders"));
2116 assert!(!db.catalog.tables.contains_key("invoices"));
2117 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2118 db.close().unwrap();
2119 }
2120
2121 #[test]
2122 fn create_table_still_takes_a_schema_naming_exactly_the_table_it_creates() {
2123 let dir = tempdir().unwrap();
2124 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2125 let mut db = Database::open(dir.path()).unwrap();
2126
2127 db.create_table(
2128 "orders",
2129 json!({
2130 "schemaVersion": 1,
2131 "table": "orders",
2132 "primaryKey": "id",
2133 "fields": { "id": { "type": "string" } },
2134 }),
2135 )
2136 .unwrap();
2137
2138 assert_eq!(db.table("orders").unwrap().schema().table, "orders");
2139 db.close().unwrap();
2140 }
2141
2142 #[test]
2143 fn create_table_refuses_on_a_readonly_database_and_changes_nothing() {
2144 let dir = tempdir().unwrap();
2145 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2146 Database::open(dir.path()).unwrap().close().unwrap();
2148 let mut db = Database::open_with(
2149 dir.path(),
2150 OpenOptions {
2151 readonly: true,
2152 lock: LockMode::default(),
2153 ..Default::default()
2154 },
2155 )
2156 .unwrap();
2157
2158 let error = db
2159 .create_table(
2160 "orders",
2161 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2162 )
2163 .unwrap_err();
2164 assert!(
2165 matches!(&error, DatabaseError::ReadOnly { dir: refused, operation } if refused == dir.path() && operation == "createTable"),
2166 "expected a typed readonly refusal, got {error:?}"
2167 );
2168 assert_eq!(db.table_names(), vec!["users"]);
2169 }
2170
2171 #[test]
2172 fn create_table_refused_by_a_directory_on_the_catalog_temporary_path_keeps_the_typed_refusal() {
2173 let dir = tempdir().unwrap();
2174 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2175 let mut db = Database::open(dir.path()).unwrap();
2176 let on_disk = fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap();
2177 let temporary = dir.path().join(format!("{MANIFEST_FILE}.tmp"));
2179 fs::create_dir(&temporary).unwrap();
2180
2181 let error = db
2182 .create_table(
2183 "orders",
2184 json!({ "schemaVersion": 1, "table": "orders", "primaryKey": "id", "fields": { "id": { "type": "string" } } }),
2185 )
2186 .unwrap_err();
2187
2188 let refusal = match &error {
2189 DatabaseError::Io(inner) => AtomicTemporaryFileError::downcast(inner)
2190 .expect("a .tmp path that could not be claimed stays a temporary-file refusal"),
2191 other => panic!("expected an io error keeping the typed refusal, got {other:?}"),
2192 };
2193 assert_eq!(refusal.path, temporary);
2194 assert!(refusal.reason.contains("could not be cleared"), "{}", refusal.reason);
2195 assert_eq!(db.table_names(), vec!["users"]);
2196 assert_eq!(fs::read_to_string(dir.path().join(MANIFEST_FILE)).unwrap(), on_disk);
2197 }
2198
2199 #[test]
2200 fn a_catalog_write_that_fails_for_any_other_reason_is_still_a_catalog_refusal() {
2201 let unreadable = catalog_write_error(Box::new(HaysonSchemaError("nope".into())));
2202
2203 assert!(matches!(unreadable, DatabaseError::Catalog(_)), "got {unreadable:?}");
2204 assert!(matches!(
2205 catalog_write_error(Box::new(io::Error::from(io::ErrorKind::PermissionDenied))),
2206 DatabaseError::Catalog(_)
2207 ));
2208 }
2209
2210 #[test]
2211 fn drop_field_without_dependents_removes_it_from_disk_and_from_memory() {
2212 let dir = tempdir().unwrap();
2213 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2214 let mut db = Database::open(dir.path()).unwrap();
2215 db.table_mut("users")
2216 .unwrap()
2217 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2218 .unwrap();
2219 db.table_mut("users").unwrap().delete("u1", false).unwrap();
2220
2221 db.drop_field("users", "name", false).unwrap();
2222
2223 assert!(!db.table("users").unwrap().schema().fields.contains_key("name"));
2224 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2225
2226 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2227 assert!(!reread.tables["users"].fields.contains_key("name"));
2228 db.close().unwrap();
2229 }
2230
2231 #[test]
2232 fn drop_field_refuses_a_live_residual_field_without_force() {
2233 let dir = tempdir().unwrap();
2234 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2235 let mut db = Database::open(dir.path()).unwrap();
2236 db.table_mut("users")
2237 .unwrap()
2238 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2239 .unwrap();
2240
2241 let error = db.drop_field("users", "name", false);
2242 assert!(error.is_err());
2243 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2244 assert!(db.table("users").unwrap().schema().fields.contains_key("name"));
2245
2246 db.drop_field("users", "name", true).unwrap();
2247 assert!(!db.catalog.tables["users"].fields.contains_key("name"));
2248 db.close().unwrap();
2249 }
2250
2251 fn users_catalog_indexing_name() -> Catalog {
2252 let mut catalog = catalog_of(&[users_schema()]);
2253 catalog.tables.get_mut("users").unwrap().indexes = Some(vec![IndexEntry {
2254 name: "by_name".to_string(),
2255 fields: vec!["name".to_string()],
2256 kind: CatalogIndexKind::Exact,
2257 unique: false,
2258 where_: None,
2259 expire_after_seconds: None,
2260 }]);
2261 catalog
2262 }
2263
2264 #[test]
2265 fn drop_field_names_the_declaration_when_residual_records_also_block_it() {
2266 let dir = tempdir().unwrap();
2267 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2268 let mut db = Database::open(dir.path()).unwrap();
2269 db.table_mut("users")
2270 .unwrap()
2271 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2272 .unwrap();
2273
2274 let error = db.drop_field("users", "name", false).unwrap_err();
2275
2276 assert!(
2277 matches!(error, DatabaseError::Schema(_)),
2278 "expected a schema refusal, got {error:?}"
2279 );
2280 let message = error.to_string();
2281 assert!(message.contains("by_name"), "{message}");
2282 assert!(!message.contains("live record(s)"), "{message}");
2283 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2284 db.close().unwrap();
2285 }
2286
2287 #[test]
2288 fn drop_field_names_the_declaration_when_it_is_the_only_thing_blocking_it() {
2289 let dir = tempdir().unwrap();
2290 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2291 let mut db = Database::open(dir.path()).unwrap();
2292
2293 let error = db.drop_field("users", "name", false).unwrap_err();
2294
2295 assert!(
2296 matches!(error, DatabaseError::Schema(_)),
2297 "expected a schema refusal, got {error:?}"
2298 );
2299 assert!(error.to_string().contains("by_name"));
2300 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2301 db.close().unwrap();
2302 }
2303
2304 #[test]
2305 fn drop_field_names_the_residual_records_when_no_declaration_blocks_it() {
2306 let dir = tempdir().unwrap();
2307 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2308 let mut db = Database::open(dir.path()).unwrap();
2309 db.table_mut("users")
2310 .unwrap()
2311 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2312 .unwrap();
2313
2314 let error = db.drop_field("users", "name", false).unwrap_err();
2315
2316 assert!(
2317 matches!(
2318 &error,
2319 DatabaseError::FieldInUse { table, field, records }
2320 if table == "users" && field == "name" && *records == 1
2321 ),
2322 "expected a typed field-in-use refusal, got {error:?}"
2323 );
2324 assert!(db.catalog.tables["users"].fields.contains_key("name"));
2325 db.close().unwrap();
2326 }
2327
2328 #[test]
2329 fn drop_field_keeps_refusing_a_declared_dependency_even_with_force() {
2330 let dir = tempdir().unwrap();
2331 write_catalog_to(dir.path(), &users_catalog_indexing_name());
2332 let mut db = Database::open(dir.path()).unwrap();
2333 db.table_mut("users")
2334 .unwrap()
2335 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2336 .unwrap();
2337
2338 let error = db.drop_field("users", "name", true).unwrap_err();
2339
2340 assert!(
2341 matches!(error, DatabaseError::Schema(_)),
2342 "expected a schema refusal, got {error:?}"
2343 );
2344 assert!(error.to_string().contains("by_name"));
2345 db.close().unwrap();
2346 }
2347
2348 #[test]
2349 fn declare_index_with_a_conflicting_unique_value_refuses_and_changes_nothing() {
2350 let dir = tempdir().unwrap();
2351 let schema = parse_schema(&json!({
2352 "schemaVersion": 1,
2353 "table": "users",
2354 "primaryKey": "id",
2355 "fields": {
2356 "id": { "type": "string" },
2357 "email": { "type": "string" },
2358 },
2359 }))
2360 .unwrap();
2361 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2362 let mut db = Database::open(dir.path()).unwrap();
2363 db.table_mut("users")
2364 .unwrap()
2365 .put(&json!({ "id": "u1", "email": "same@example.com" }), false)
2366 .unwrap();
2367 db.table_mut("users")
2368 .unwrap()
2369 .put(&json!({ "id": "u2", "email": "same@example.com" }), false)
2370 .unwrap();
2371
2372 let error = db.declare_index(
2373 "users",
2374 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2375 );
2376 assert!(error.is_err());
2377 assert!(db.catalog.tables["users"].indexes.is_none());
2378
2379 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
2380 assert!(reread.tables["users"].indexes.is_none());
2381 db.close().unwrap();
2382 }
2383
2384 #[test]
2385 fn declare_index_without_a_conflict_covers_records_written_before_the_declaration() {
2386 let dir = tempdir().unwrap();
2387 let schema = parse_schema(&json!({
2388 "schemaVersion": 1,
2389 "table": "users",
2390 "primaryKey": "id",
2391 "fields": {
2392 "id": { "type": "string" },
2393 "email": { "type": "string" },
2394 },
2395 }))
2396 .unwrap();
2397 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2398 let mut db = Database::open(dir.path()).unwrap();
2399 db.table_mut("users")
2400 .unwrap()
2401 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2402 .unwrap();
2403 db.table_mut("users")
2404 .unwrap()
2405 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2406 .unwrap();
2407 db.table_mut("users")
2408 .unwrap()
2409 .put(&json!({ "id": "u3", "email": "kasia@example.com" }), false)
2410 .unwrap();
2411
2412 db.declare_index(
2413 "users",
2414 json!({ "name": "users_email_unique", "fields": ["email"], "kind": "exact", "unique": true }),
2415 )
2416 .unwrap();
2417
2418 assert_eq!(db.catalog.tables["users"].indexes.as_ref().unwrap().len(), 1);
2419
2420 let spec = QueryBuilder::new()
2421 .r#where(
2422 "email",
2423 crate::query::predicates::ComparisonOp::Eq,
2424 json!("piotr@example.com"),
2425 )
2426 .spec();
2427 let found = db.table("users").unwrap().query(&spec).unwrap();
2428 assert_eq!(found.keys, vec!["u2".to_string()]);
2429
2430 let conflicting = db
2431 .table_mut("users")
2432 .unwrap()
2433 .put(&json!({ "id": "u4", "email": "piotr@example.com" }), false);
2434 assert!(conflicting.is_err());
2435 db.close().unwrap();
2436 }
2437
2438 #[test]
2439 fn declare_index_snapshots_the_reopened_table_and_keeps_snapshotting_it_afterwards() {
2440 let dir = tempdir().unwrap();
2441 let schema = parse_schema(&json!({
2442 "schemaVersion": 1,
2443 "table": "users",
2444 "primaryKey": "id",
2445 "fields": {
2446 "id": { "type": "string" },
2447 "email": { "type": "string" },
2448 },
2449 }))
2450 .unwrap();
2451 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2452 let mut db = Database::open(dir.path()).unwrap();
2453 db.table_mut("users")
2454 .unwrap()
2455 .put(&json!({ "id": "u1", "email": "ania@example.com" }), false)
2456 .unwrap();
2457
2458 db.declare_index(
2459 "users",
2460 json!({ "name": "users_email_exact", "fields": ["email"], "kind": "exact" }),
2461 )
2462 .unwrap();
2463
2464 assert!(
2465 index_path(dir.path(), "users").exists(),
2466 "declare_index owes the table a snapshot carrying the new index's projection"
2467 );
2468
2469 db.table_mut("users")
2470 .unwrap()
2471 .put(&json!({ "id": "u2", "email": "piotr@example.com" }), false)
2472 .unwrap();
2473 assert!(
2474 db.table("users").unwrap().needs_snapshot(),
2475 "the handle declare_index left behind must still be able to snapshot itself"
2476 );
2477 db.close().unwrap();
2478
2479 let reopened = Database::open(dir.path()).unwrap();
2480 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
2481 let spec = QueryBuilder::new()
2482 .r#where(
2483 "email",
2484 crate::query::predicates::ComparisonOp::Eq,
2485 json!("piotr@example.com"),
2486 )
2487 .spec();
2488 assert_eq!(
2489 reopened.table("users").unwrap().query(&spec).unwrap().keys,
2490 vec!["u2".to_string()]
2491 );
2492 assert_eq!(
2493 declared_on_disk(dir.path(), "users"),
2494 vec!["users_email_exact".to_string()]
2495 );
2496 reopened.close().unwrap();
2497 }
2498
2499 #[test]
2500 fn a_database_wide_sync_default_reaches_a_table_declaring_none_of_its_own() {
2501 let dir = tempdir().unwrap();
2502 let mut catalog = catalog_of(&[users_schema()]);
2503 catalog.defaults = Some(crate::catalog::TableOptions {
2504 sync: Some(crate::catalog::CatalogSyncMode::Never),
2505 ..Default::default()
2506 });
2507 write_catalog_to(dir.path(), &catalog);
2508
2509 let db = Database::open(dir.path()).unwrap();
2510 assert_eq!(
2511 db.table("users").unwrap().sync_mode(),
2512 crate::catalog::CatalogSyncMode::Never
2513 );
2514 db.close().unwrap();
2515 }
2516
2517 #[test]
2518 fn a_table_declaring_its_own_sync_mode_beats_the_database_wide_default() {
2519 let dir = tempdir().unwrap();
2520 let mut catalog = catalog_of(&[users_schema(), orders_schema()]);
2521 catalog.defaults = Some(crate::catalog::TableOptions {
2522 sync: Some(crate::catalog::CatalogSyncMode::Never),
2523 ..Default::default()
2524 });
2525 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
2526 sync: Some(crate::catalog::CatalogSyncMode::Commit),
2527 ..Default::default()
2528 });
2529 write_catalog_to(dir.path(), &catalog);
2530
2531 let db = Database::open(dir.path()).unwrap();
2532 assert_eq!(
2533 db.table("users").unwrap().sync_mode(),
2534 crate::catalog::CatalogSyncMode::Commit
2535 );
2536 assert_eq!(
2537 db.table("orders").unwrap().sync_mode(),
2538 crate::catalog::CatalogSyncMode::Never
2539 );
2540 db.close().unwrap();
2541 }
2542
2543 #[test]
2544 fn a_table_declaring_validation_warn_takes_a_write_with_a_stray_field_and_stores_it_without_that_field() {
2545 let dir = tempdir().unwrap();
2546 let mut catalog = catalog_of(&[users_schema()]);
2547 catalog.tables.get_mut("users").unwrap().validation = Some(crate::catalog::CatalogValidation::Warn);
2548 write_catalog_to(dir.path(), &catalog);
2549
2550 let mut db = Database::open(dir.path()).unwrap();
2551 let document = json!({ "id": "u1", "name": "Ania", "nickname": "Anka" });
2552 let table = db.table_mut("users").unwrap();
2553 let record = table.put(&document, false).unwrap();
2554
2555 assert_eq!(record.get("nickname"), None);
2556 assert_eq!(
2557 table.dropped_field_warnings(&document),
2558 vec!["field \"nickname\": is not declared in the schema, dropped from the write".to_string()]
2559 );
2560 db.close().unwrap();
2561
2562 let reopened = Database::open(dir.path()).unwrap();
2563 let stored = reopened.table("users").unwrap().get("u1").unwrap().unwrap();
2564 assert_eq!(stored.get("name"), Some(&json!("Ania")));
2565 assert_eq!(stored.get("nickname"), None);
2566 reopened.close().unwrap();
2567 }
2568
2569 #[test]
2570 fn a_table_declaring_no_validation_refuses_a_write_with_a_stray_field() {
2571 let dir = tempdir().unwrap();
2572 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2573
2574 let mut db = Database::open(dir.path()).unwrap();
2575 assert_eq!(
2576 db.table("users").unwrap().validation(),
2577 crate::catalog::CatalogValidation::Strict
2578 );
2579 let error = db
2580 .table_mut("users")
2581 .unwrap()
2582 .put(&json!({ "id": "u1", "name": "Ania", "nickname": "Anka" }), false)
2583 .unwrap_err();
2584
2585 assert!(error.to_string().contains("is not declared in the schema"), "{error}");
2586 db.close().unwrap();
2587 }
2588
2589 fn deprecating_age(dir: &Path, validation: Option<crate::catalog::CatalogValidation>, note: &str) {
2590 let mut catalog = catalog_of(&[users_with_age_schema()]);
2591 let entry = catalog.tables.get_mut("users").unwrap();
2592 entry.validation = validation;
2593 entry.fields.get_mut("age").unwrap().deprecated = Some(note.to_string());
2594 write_catalog_to(dir, &catalog);
2595 }
2596
2597 fn users_with_age_schema() -> TableSchema {
2598 parse_schema(&json!({
2599 "schemaVersion": 1,
2600 "table": "users",
2601 "primaryKey": "id",
2602 "fields": {
2603 "id": { "type": "string" },
2604 "name": { "type": "string", "required": true },
2605 "age": { "type": "int" },
2606 },
2607 }))
2608 .unwrap()
2609 }
2610
2611 #[test]
2612 fn a_write_touching_a_field_the_catalog_marks_deprecated_warns_with_the_field_and_the_note() {
2613 let dir = tempdir().unwrap();
2614 deprecating_age(dir.path(), None, "use \"bornAt\" instead");
2615
2616 let mut db = Database::open(dir.path()).unwrap();
2617 let table = db.table_mut("users").unwrap();
2618 let touching = json!({ "id": "u1", "name": "Ania", "age": 30 });
2619 let record = table.put(&touching, false).unwrap();
2620
2621 assert_eq!(record.get("age"), Some(&json!(30)));
2622 assert_eq!(
2623 table.deprecated_field_warnings(&touching),
2624 vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()]
2625 );
2626 assert!(table
2627 .deprecated_field_warnings(&json!({ "id": "u2", "name": "Ola" }))
2628 .is_empty());
2629 db.close().unwrap();
2630 }
2631
2632 #[test]
2633 fn the_deprecated_mark_reaches_a_table_in_either_validation_mode_and_on_every_write() {
2634 let expected = vec!["field \"age\": is deprecated — use \"bornAt\" instead".to_string()];
2635 let document = json!({ "id": "u1", "name": "Ania", "age": 30 });
2636
2637 for validation in [None, Some(crate::catalog::CatalogValidation::Warn)] {
2638 let dir = tempdir().unwrap();
2639 deprecating_age(dir.path(), validation, "use \"bornAt\" instead");
2640
2641 let mut db = Database::open(dir.path()).unwrap();
2642 let table = db.table_mut("users").unwrap();
2643 table.put(&document, false).unwrap();
2644 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
2645 table.put(&document, false).unwrap();
2646 assert_eq!(table.deprecated_field_warnings(&document), expected, "{validation:?}");
2647 db.close().unwrap();
2648
2649 let reopened = Database::open(dir.path()).unwrap();
2650 assert_eq!(
2651 reopened.table("users").unwrap().deprecated_field_warnings(&document),
2652 expected
2653 );
2654 reopened.close().unwrap();
2655 }
2656 }
2657
2658 #[test]
2659 fn a_catalog_marking_no_field_deprecated_leaves_every_table_without_the_warning() {
2660 let dir = tempdir().unwrap();
2661 write_catalog_to(dir.path(), &catalog_of(&[users_with_age_schema()]));
2662
2663 let db = Database::open(dir.path()).unwrap();
2664 let table = db.table("users").unwrap();
2665
2666 assert!(table.deprecated_fields().is_empty());
2667 assert!(table
2668 .deprecated_field_warnings(&json!({ "id": "u1", "name": "Ania", "age": 30 }))
2669 .is_empty());
2670 db.close().unwrap();
2671 }
2672
2673 #[test]
2674 fn a_catalog_naming_no_sync_anywhere_leaves_every_table_committing() {
2675 let dir = tempdir().unwrap();
2676 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
2677
2678 let db = Database::open(dir.path()).unwrap();
2679 assert_eq!(
2680 db.table("users").unwrap().sync_mode(),
2681 crate::catalog::CatalogSyncMode::Commit
2682 );
2683 db.close().unwrap();
2684 }
2685
2686 #[test]
2687 fn a_sync_never_table_survives_a_close_and_reopen_with_its_writes_and_its_mode() {
2688 let dir = tempdir().unwrap();
2689 let mut catalog = catalog_of(&[users_schema()]);
2690 catalog.tables.get_mut("users").unwrap().options = Some(crate::catalog::TableOptions {
2691 sync: Some(crate::catalog::CatalogSyncMode::Never),
2692 ..Default::default()
2693 });
2694 write_catalog_to(dir.path(), &catalog);
2695
2696 {
2697 let mut db = Database::open(dir.path()).unwrap();
2698 db.table_mut("users")
2699 .unwrap()
2700 .put(&json!({ "id": "u1", "name": "Ania" }), false)
2701 .unwrap();
2702 db.close().unwrap();
2703 }
2704
2705 let db = Database::open(dir.path()).unwrap();
2706 assert_eq!(
2707 db.table("users").unwrap().sync_mode(),
2708 crate::catalog::CatalogSyncMode::Never
2709 );
2710 assert_eq!(
2711 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
2712 Some(&json!("Ania"))
2713 );
2714 db.close().unwrap();
2715 }
2716
2717 #[test]
2718 fn a_database_wide_default_names_the_link_parsing_mode_a_table_declaring_none_opens_with() {
2719 let dir = tempdir().unwrap();
2720 let pages_schema = parse_schema(&json!({
2721 "schemaVersion": 1,
2722 "table": "pages",
2723 "primaryKey": "id",
2724 "fields": {
2725 "id": { "type": "string" },
2726 "page": { "type": "markdown" },
2727 },
2728 }))
2729 .unwrap();
2730 let mut catalog = catalog_of(&[pages_schema]);
2731 catalog.defaults = Some(crate::catalog::TableOptions {
2732 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
2733 ..Default::default()
2734 });
2735 write_catalog_to(dir.path(), &catalog);
2736
2737 let db = Database::open(dir.path()).unwrap();
2738 assert_eq!(
2739 db.table("pages").unwrap().link_parsing(),
2740 crate::catalog::LinkParsing::Lazy
2741 );
2742 db.close().unwrap();
2743 }
2744
2745 #[test]
2746 fn a_table_declaring_its_own_link_parsing_mode_beats_the_database_wide_default() {
2747 let dir = tempdir().unwrap();
2748 let pages_schema = parse_schema(&json!({
2749 "schemaVersion": 1,
2750 "table": "pages",
2751 "primaryKey": "id",
2752 "fields": {
2753 "id": { "type": "string" },
2754 "page": { "type": "markdown" },
2755 },
2756 }))
2757 .unwrap();
2758 let mut catalog = catalog_of(&[pages_schema]);
2759 catalog.defaults = Some(crate::catalog::TableOptions {
2760 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
2761 ..Default::default()
2762 });
2763 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
2764 link_parsing: Some(crate::catalog::LinkParsing::Eager),
2765 ..Default::default()
2766 });
2767 write_catalog_to(dir.path(), &catalog);
2768
2769 let db = Database::open(dir.path()).unwrap();
2770 assert_eq!(
2771 db.table("pages").unwrap().link_parsing(),
2772 crate::catalog::LinkParsing::Eager
2773 );
2774 db.close().unwrap();
2775 }
2776
2777 #[test]
2778 fn a_database_wide_default_names_the_image_dir_a_table_declaring_none_writes_under() {
2779 let dir = tempdir().unwrap();
2780 let docs_schema = parse_schema(&json!({
2781 "schemaVersion": 1,
2782 "table": "docs",
2783 "primaryKey": "id",
2784 "fields": {
2785 "id": { "type": "string" },
2786 "cover": { "type": "image" },
2787 },
2788 }))
2789 .unwrap();
2790 let mut catalog = catalog_of(&[docs_schema]);
2791 catalog.defaults = Some(crate::catalog::TableOptions {
2792 image_dir: Some("shared-images".to_string()),
2793 ..Default::default()
2794 });
2795 write_catalog_to(dir.path(), &catalog);
2796
2797 let mut db = Database::open(dir.path()).unwrap();
2798 let record = db
2799 .table_mut("docs")
2800 .unwrap()
2801 .put(
2802 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
2803 false,
2804 )
2805 .unwrap();
2806 let physical_path = record
2807 .get("cover")
2808 .unwrap()
2809 .get("physicalPath")
2810 .unwrap()
2811 .as_str()
2812 .unwrap();
2813
2814 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
2815 assert!(dir.path().join("shared-images").join(physical_path).exists());
2816 db.close().unwrap();
2817 }
2818
2819 #[test]
2820 fn declare_index_of_a_composite_field_pair_serves_a_query_matching_both_fields() {
2821 let dir = tempdir().unwrap();
2822 let schema = parse_schema(&json!({
2823 "schemaVersion": 1,
2824 "table": "orders",
2825 "primaryKey": "id",
2826 "fields": {
2827 "id": { "type": "string" },
2828 "userId": { "type": "string" },
2829 "status": { "type": "string" },
2830 },
2831 }))
2832 .unwrap();
2833 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2834 let mut db = Database::open(dir.path()).unwrap();
2835 db.table_mut("orders")
2836 .unwrap()
2837 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
2838 .unwrap();
2839 db.table_mut("orders")
2840 .unwrap()
2841 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
2842 .unwrap();
2843 db.table_mut("orders")
2844 .unwrap()
2845 .put(&json!({ "id": "o3", "userId": "u2", "status": "open" }), false)
2846 .unwrap();
2847
2848 db.declare_index(
2849 "orders",
2850 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
2851 )
2852 .unwrap();
2853
2854 let spec = QueryBuilder::new()
2855 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
2856 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("open"))
2857 .spec();
2858 let found = db.table("orders").unwrap().query(&spec).unwrap();
2859 assert_eq!(found.keys, vec!["o1".to_string()]);
2860 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
2861 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
2862 assert_eq!(found.plan.candidates, 1);
2863 db.close().unwrap();
2864 }
2865
2866 #[test]
2867 fn a_where_scoped_index_answers_only_a_query_that_repeats_its_predicate() {
2868 let dir = tempdir().unwrap();
2869 let schema = parse_schema(&json!({
2870 "schemaVersion": 1,
2871 "table": "users",
2872 "primaryKey": "id",
2873 "fields": {
2874 "id": { "type": "string" },
2875 "role": { "type": "string" },
2876 "status": { "type": "string" },
2877 },
2878 }))
2879 .unwrap();
2880 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2881 let mut db = Database::open(dir.path()).unwrap();
2882 db.table_mut("users")
2883 .unwrap()
2884 .put(&json!({ "id": "u1", "role": "admin", "status": "active" }), false)
2885 .unwrap();
2886 db.table_mut("users")
2887 .unwrap()
2888 .put(&json!({ "id": "u2", "role": "admin", "status": "inactive" }), false)
2889 .unwrap();
2890
2891 db.declare_index(
2892 "users",
2893 json!({
2894 "name": "users_role_active",
2895 "fields": ["role"],
2896 "kind": "exact",
2897 "where": [{ "field": "status", "op": "=", "value": "active" }],
2898 }),
2899 )
2900 .unwrap();
2901
2902 let eq = crate::query::predicates::ComparisonOp::Eq;
2903 let omitted = QueryBuilder::new().r#where("role", eq, json!("admin")).spec();
2904 let found = db.table("users").unwrap().query(&omitted).unwrap();
2905 assert_eq!(
2906 found.keys,
2907 vec!["u1".to_string(), "u2".to_string()],
2908 "the index holds u1 alone, but the query never asked for active records only"
2909 );
2910 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Scan);
2911
2912 let repeated = QueryBuilder::new()
2913 .r#where("role", eq, json!("admin"))
2914 .r#where("status", eq, json!("active"))
2915 .spec();
2916 let narrowed = db.table("users").unwrap().query(&repeated).unwrap();
2917 assert_eq!(narrowed.keys, vec!["u1".to_string()]);
2918 assert_eq!(narrowed.plan.strategy, crate::query::planner::QueryStrategy::Exact);
2919 assert_eq!(narrowed.plan.index, Some("users_role_active".to_string()));
2920 assert_eq!(narrowed.plan.candidates, 1);
2921 db.close().unwrap();
2922 }
2923
2924 #[test]
2925 fn a_declared_composite_index_still_serves_queries_after_closing_and_reopening_the_database() {
2926 let dir = tempdir().unwrap();
2927 let schema = parse_schema(&json!({
2928 "schemaVersion": 1,
2929 "table": "orders",
2930 "primaryKey": "id",
2931 "fields": {
2932 "id": { "type": "string" },
2933 "userId": { "type": "string" },
2934 "status": { "type": "string" },
2935 },
2936 }))
2937 .unwrap();
2938 write_catalog_to(dir.path(), &catalog_of(&[schema]));
2939 let mut db = Database::open(dir.path()).unwrap();
2940 db.table_mut("orders")
2941 .unwrap()
2942 .put(&json!({ "id": "o1", "userId": "u1", "status": "open" }), false)
2943 .unwrap();
2944 db.table_mut("orders")
2945 .unwrap()
2946 .put(&json!({ "id": "o2", "userId": "u1", "status": "closed" }), false)
2947 .unwrap();
2948 db.declare_index(
2949 "orders",
2950 json!({ "name": "orders_user_status", "fields": ["userId", "status"], "kind": "exact" }),
2951 )
2952 .unwrap();
2953 db.close().unwrap();
2954
2955 let reopened = Database::open(dir.path()).unwrap();
2956 let spec = QueryBuilder::new()
2957 .r#where("userId", crate::query::predicates::ComparisonOp::Eq, json!("u1"))
2958 .r#where("status", crate::query::predicates::ComparisonOp::Eq, json!("closed"))
2959 .spec();
2960 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
2961 assert_eq!(found.keys, vec!["o2".to_string()]);
2962 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Exact);
2963 assert_eq!(found.plan.index, Some("orders_user_status".to_string()));
2964 reopened.close().unwrap();
2965 }
2966
2967 fn add_index_to_catalog_on_disk(dir: &Path, table: &str, index: IndexEntry) {
2968 let mut catalog = read_catalog(&dir.join(MANIFEST_FILE)).unwrap().unwrap();
2969 let entry = catalog.tables.get_mut(table).unwrap();
2970 entry.indexes = Some(vec![index]);
2971 write_catalog_to(dir, &catalog);
2972 }
2973
2974 fn sorted_index_entry(name: &str, fields: &[&str]) -> IndexEntry {
2975 IndexEntry {
2976 name: name.to_string(),
2977 fields: fields.iter().map(|field| field.to_string()).collect(),
2978 kind: CatalogIndexKind::Sorted,
2979 unique: false,
2980 where_: None,
2981 expire_after_seconds: None,
2982 }
2983 }
2984
2985 #[test]
2990 fn a_snapshot_written_before_an_index_was_declared_is_rebuilt_rather_than_restored_into_it() {
2991 let dir = tempdir().unwrap();
2992 let schema = parse_schema(&json!({
2993 "schemaVersion": 1,
2994 "table": "orders",
2995 "primaryKey": "id",
2996 "fields": {
2997 "id": { "type": "string" },
2998 "userId": { "type": "string" },
2999 "total": { "type": "int" },
3000 },
3001 }))
3002 .unwrap();
3003 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3004 let mut db = Database::open(dir.path()).unwrap();
3005 let orders = db.table_mut("orders").unwrap();
3006 orders
3007 .put(&json!({ "id": "a", "userId": "u1", "total": 10 }), false)
3008 .unwrap();
3009 orders.put(&json!({ "id": "b", "userId": "u1" }), false).unwrap();
3010 orders
3011 .put(&json!({ "id": "c", "userId": "u0", "total": 5 }), false)
3012 .unwrap();
3013 db.close().unwrap();
3014 assert!(
3015 index_path(dir.path(), "orders").exists(),
3016 "the fixture needs a snapshot to restore from"
3017 );
3018
3019 add_index_to_catalog_on_disk(
3020 dir.path(),
3021 "orders",
3022 sorted_index_entry("orders_user_total", &["userId", "total"]),
3023 );
3024
3025 let reopened = Database::open(dir.path()).unwrap();
3026 assert_eq!(reopened.restore_outcome("orders"), Some(RestoreOutcome::Rebuilt));
3027
3028 let spec = QueryBuilder::new()
3029 .r#where("userId", crate::query::predicates::ComparisonOp::Gte, json!("u1"))
3030 .spec();
3031 let found = reopened.table("orders").unwrap().query(&spec).unwrap();
3032 assert_eq!(found.plan.strategy, crate::query::planner::QueryStrategy::Sorted);
3033 assert_eq!(found.plan.index, Some("orders_user_total".to_string()));
3034 let mut keys = found.keys.clone();
3035 keys.sort();
3036 assert_eq!(
3037 keys,
3038 vec!["a".to_string(), "b".to_string()],
3039 "b has no total, but it still has a userId"
3040 );
3041 reopened.close().unwrap();
3042 }
3043
3044 #[test]
3047 fn a_composite_unique_index_declared_after_a_snapshot_still_holds_across_partitions() {
3048 let dir = tempdir().unwrap();
3049 let schema = parse_schema(&json!({
3050 "schemaVersion": 1,
3051 "table": "tickets",
3052 "primaryKey": "id",
3053 "partitionBy": "kind",
3054 "fields": {
3055 "id": { "type": "string" },
3056 "kind": { "type": "string", "required": true },
3057 "userId": { "type": "string", "required": true },
3058 "orderId": { "type": "int" },
3059 },
3060 }))
3061 .unwrap();
3062 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3063 let mut db = Database::open(dir.path()).unwrap();
3064 let tickets = db.table_mut("tickets").unwrap();
3065 tickets
3066 .put(
3067 &json!({ "id": "t1", "kind": "sale", "userId": "u1", "orderId": 1 }),
3068 false,
3069 )
3070 .unwrap();
3071 tickets
3072 .put(
3073 &json!({ "id": "t2", "kind": "sale", "userId": "u1", "orderId": 2 }),
3074 false,
3075 )
3076 .unwrap();
3077 tickets
3078 .put(
3079 &json!({ "id": "t3", "kind": "refund", "userId": "u1", "orderId": 3 }),
3080 false,
3081 )
3082 .unwrap();
3083 db.close().unwrap();
3084
3085 add_index_to_catalog_on_disk(
3086 dir.path(),
3087 "tickets",
3088 IndexEntry {
3089 name: "tickets_user_order".to_string(),
3090 fields: vec!["userId".to_string(), "orderId".to_string()],
3091 kind: CatalogIndexKind::Exact,
3092 unique: true,
3093 where_: None,
3094 expire_after_seconds: None,
3095 },
3096 );
3097
3098 let mut reopened = Database::open(dir.path()).unwrap();
3099 let tickets = reopened.table_mut("tickets").unwrap();
3100 let clash = tickets.put(
3102 &json!({ "id": "t9", "kind": "refund", "userId": "u1", "orderId": 2 }),
3103 false,
3104 );
3105 match clash {
3106 Err(TableError::Unique(violation)) => {
3107 assert_eq!(violation.field, "userId, orderId");
3108 assert_eq!(violation.holder, "t2");
3109 }
3110 other => panic!("a repeated composite key must be refused, got {other:?}"),
3111 }
3112 tickets
3113 .put(
3114 &json!({ "id": "t10", "kind": "refund", "userId": "u1", "orderId": 9 }),
3115 false,
3116 )
3117 .unwrap();
3118 reopened.close().unwrap();
3119 }
3120
3121 #[test]
3122 fn close_snapshots_dirty_tables_so_a_reopen_restores_instead_of_rebuilding() {
3123 let dir = tempdir().unwrap();
3124 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3125 let mut db = Database::open(dir.path()).unwrap();
3126
3127 db.table_mut("users")
3129 .unwrap()
3130 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3131 .unwrap();
3132 db.table_mut("users")
3133 .unwrap()
3134 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3135 .unwrap();
3136 assert!(!index_path(dir.path(), "users").exists());
3137
3138 db.close().unwrap();
3139 assert!(index_path(dir.path(), "users").exists());
3140
3141 let reopened = Database::open(dir.path()).unwrap();
3142 assert_eq!(reopened.restore_outcome("users"), Some(RestoreOutcome::Restored));
3143 assert_eq!(
3144 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3145 Some(&json!("Ania"))
3146 );
3147 assert_eq!(
3148 reopened.table("users").unwrap().get("u2").unwrap().unwrap().get("name"),
3149 Some(&json!("Piotr"))
3150 );
3151 reopened.close().unwrap();
3152 }
3153
3154 #[test]
3155 fn close_commits_a_staged_write_so_a_reopen_still_finds_it() {
3156 let dir = tempdir().unwrap();
3157 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3158 let mut db = Database::open(dir.path()).unwrap();
3159 db.table_mut("users")
3160 .unwrap()
3161 .stage_put(&json!({ "id": "u1", "name": "Ania" }), false)
3162 .unwrap();
3163 assert_eq!(
3164 db.table("users").unwrap().len(),
3165 0,
3166 "nothing is committed before the close"
3167 );
3168
3169 db.close().unwrap();
3170
3171 let reopened = Database::open(dir.path()).unwrap();
3172 assert_eq!(
3173 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3174 Some(&json!("Ania"))
3175 );
3176 reopened.close().unwrap();
3177 }
3178
3179 #[test]
3180 fn a_low_snapshot_every_writes_the_index_without_waiting_for_close() {
3181 let dir = tempdir().unwrap();
3182 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3183 let mut db = Database::open_with(
3184 dir.path(),
3185 OpenOptions {
3186 snapshot_every: 2,
3187 ..Default::default()
3188 },
3189 )
3190 .unwrap();
3191
3192 db.table_mut("users")
3193 .unwrap()
3194 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3195 .unwrap();
3196 assert!(!index_path(dir.path(), "users").exists());
3197 db.table_mut("users")
3198 .unwrap()
3199 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3200 .unwrap();
3201 assert!(index_path(dir.path(), "users").exists());
3202
3203 db.close().unwrap();
3204 }
3205
3206 #[test]
3207 fn a_snapshot_taken_mid_stream_does_not_disturb_writes_that_follow_it() {
3208 let dir = tempdir().unwrap();
3209 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3210 let mut db = Database::open_with(
3211 dir.path(),
3212 OpenOptions {
3213 snapshot_every: 2,
3214 ..Default::default()
3215 },
3216 )
3217 .unwrap();
3218
3219 db.table_mut("users")
3220 .unwrap()
3221 .put(&json!({ "id": "u1", "name": "Ania" }), false)
3222 .unwrap();
3223 db.table_mut("users")
3224 .unwrap()
3225 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3226 .unwrap();
3227 db.table_mut("users")
3228 .unwrap()
3229 .put(&json!({ "id": "u3", "name": "Kasia" }), false)
3230 .unwrap();
3231 assert_eq!(
3232 db.table("users").unwrap().get("u3").unwrap().unwrap().get("name"),
3233 Some(&json!("Kasia"))
3234 );
3235 db.close().unwrap();
3236 }
3237
3238 fn users_insert_audits_into_orders() -> Value {
3239 json!({
3240 "name": "users_insert_audit",
3241 "table": "users",
3242 "event": "insert",
3243 "effect": {
3244 "table": "orders",
3245 "fields": [
3246 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3247 { "target": "userId", "source": { "kind": "new", "field": "id" } },
3248 ],
3249 },
3250 })
3251 }
3252
3253 #[test]
3254 fn a_table_declaring_an_image_dir_option_writes_bytes_under_it_instead_of_the_default() {
3255 let dir = tempdir().unwrap();
3256 let docs_schema = parse_schema(&json!({
3257 "schemaVersion": 1,
3258 "table": "docs",
3259 "primaryKey": "id",
3260 "fields": {
3261 "id": { "type": "string" },
3262 "cover": { "type": "image" },
3263 },
3264 }))
3265 .unwrap();
3266 let mut catalog = catalog_of(&[docs_schema]);
3267 catalog.tables.get_mut("docs").unwrap().options = Some(crate::catalog::TableOptions {
3268 image_dir: Some("shared-images".to_string()),
3269 ..Default::default()
3270 });
3271 write_catalog_to(dir.path(), &catalog);
3272
3273 let mut db = Database::open(dir.path()).unwrap();
3274 let record = db
3275 .table_mut("docs")
3276 .unwrap()
3277 .put(
3278 &json!({ "id": "d1", "cover": { "content": "Ynl0ZXM=", "virtualPath": "/images/a.png" } }),
3279 false,
3280 )
3281 .unwrap();
3282 let physical_path = record
3283 .get("cover")
3284 .unwrap()
3285 .get("physicalPath")
3286 .unwrap()
3287 .as_str()
3288 .unwrap();
3289
3290 assert_eq!(db.table("docs").unwrap().image_dir(), dir.path().join("shared-images"));
3291 assert!(dir.path().join("shared-images").join(physical_path).exists());
3292 db.close().unwrap();
3293 }
3294
3295 #[test]
3296 fn a_table_declaring_a_lazy_link_parsing_option_opens_with_it_instead_of_the_eager_default() {
3297 let dir = tempdir().unwrap();
3298 let pages_schema = parse_schema(&json!({
3299 "schemaVersion": 1,
3300 "table": "pages",
3301 "primaryKey": "id",
3302 "fields": {
3303 "id": { "type": "string" },
3304 "page": { "type": "markdown" },
3305 },
3306 }))
3307 .unwrap();
3308 let mut catalog = catalog_of(&[pages_schema]);
3309 catalog.tables.get_mut("pages").unwrap().options = Some(crate::catalog::TableOptions {
3310 link_parsing: Some(crate::catalog::LinkParsing::Lazy),
3311 ..Default::default()
3312 });
3313 write_catalog_to(dir.path(), &catalog);
3314
3315 let mut db = Database::open(dir.path()).unwrap();
3316 assert_eq!(
3317 db.table("pages").unwrap().link_parsing(),
3318 crate::catalog::LinkParsing::Lazy
3319 );
3320
3321 let outcome = db
3322 .table_mut("pages")
3323 .unwrap()
3324 .put_with_warnings(
3325 &json!({ "id": "a", "page": { "content": "[missing](/docs/x.md)", "virtualPath": "/docs/a.md" } }),
3326 false,
3327 )
3328 .unwrap();
3329 assert!(outcome.warnings.is_empty());
3330 db.close().unwrap();
3331 }
3332
3333 #[test]
3334 fn a_table_declaring_no_link_parsing_option_defaults_to_eager() {
3335 let dir = tempdir().unwrap();
3336 let pages_schema = parse_schema(&json!({
3337 "schemaVersion": 1,
3338 "table": "pages",
3339 "primaryKey": "id",
3340 "fields": {
3341 "id": { "type": "string" },
3342 "page": { "type": "markdown" },
3343 },
3344 }))
3345 .unwrap();
3346 write_catalog_to(dir.path(), &catalog_of(&[pages_schema]));
3347
3348 let db = Database::open(dir.path()).unwrap();
3349 assert_eq!(
3350 db.table("pages").unwrap().link_parsing(),
3351 crate::catalog::LinkParsing::Eager
3352 );
3353 }
3354
3355 #[test]
3356 fn declare_trigger_writes_the_catalog_and_refreshes_both_tables_live_trigger_set() {
3357 let dir = tempdir().unwrap();
3358 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3359 let mut db = Database::open(dir.path()).unwrap();
3360
3361 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3362
3363 assert_eq!(db.catalog.triggers.as_ref().unwrap().len(), 1);
3364 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3365 assert!(db.table("orders").unwrap().triggers().is_empty());
3366
3367 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
3368 assert_eq!(reread.triggers.as_ref().unwrap().len(), 1);
3369 assert_eq!(reread.triggers.as_ref().unwrap()[0].name, "users_insert_audit");
3370
3371 db.close().unwrap();
3372 }
3373
3374 #[test]
3375 fn a_declared_trigger_fires_its_effect_immediately_on_the_next_unit_of_work() {
3376 let dir = tempdir().unwrap();
3377 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3378 let mut db = Database::open(dir.path()).unwrap();
3379 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3380
3381 let mut unit = db.unit_of_work();
3382 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3383 db.commit(&mut unit).unwrap();
3384
3385 assert_eq!(
3386 db.table("orders")
3387 .unwrap()
3388 .get("o-audit")
3389 .unwrap()
3390 .unwrap()
3391 .get("userId"),
3392 Some(&json!("u1"))
3393 );
3394 db.close().unwrap();
3395 }
3396
3397 #[test]
3398 fn a_declared_trigger_still_fires_after_closing_and_reopening_the_database() {
3399 let dir = tempdir().unwrap();
3400 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3401 let mut db = Database::open(dir.path()).unwrap();
3402 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3403 db.close().unwrap();
3404
3405 let mut reopened = Database::open(dir.path()).unwrap();
3406 let mut unit = reopened.unit_of_work();
3407 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3408 reopened.commit(&mut unit).unwrap();
3409
3410 assert_eq!(
3411 reopened
3412 .table("orders")
3413 .unwrap()
3414 .get("o-audit")
3415 .unwrap()
3416 .unwrap()
3417 .get("userId"),
3418 Some(&json!("u1"))
3419 );
3420 reopened.close().unwrap();
3421 }
3422
3423 #[test]
3424 fn a_declared_trigger_still_fires_after_declare_index_reopened_its_source_table() {
3425 let dir = tempdir().unwrap();
3426 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3427 let mut db = Database::open(dir.path()).unwrap();
3428 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3429
3430 db.declare_index(
3431 "users",
3432 json!({ "name": "users_by_name", "fields": ["name"], "kind": "exact" }),
3433 )
3434 .unwrap();
3435
3436 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3437
3438 let mut unit = db.unit_of_work();
3439 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
3440 db.commit(&mut unit).unwrap();
3441
3442 assert_eq!(
3443 db.table("orders")
3444 .unwrap()
3445 .get("o-audit")
3446 .unwrap()
3447 .unwrap()
3448 .get("userId"),
3449 Some(&json!("u1"))
3450 );
3451 db.close().unwrap();
3452 }
3453
3454 #[test]
3455 fn a_declared_trigger_still_fires_after_drop_field_reopened_its_source_table() {
3456 let dir = tempdir().unwrap();
3457 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3458 let mut db = Database::open(dir.path()).unwrap();
3459 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3460
3461 db.drop_field("users", "name", false).unwrap();
3462
3463 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3464
3465 let mut unit = db.unit_of_work();
3466 unit.put("users", json!({ "id": "u1" }));
3467 db.commit(&mut unit).unwrap();
3468
3469 assert_eq!(
3470 db.table("orders")
3471 .unwrap()
3472 .get("o-audit")
3473 .unwrap()
3474 .unwrap()
3475 .get("userId"),
3476 Some(&json!("u1"))
3477 );
3478 db.close().unwrap();
3479 }
3480
3481 #[test]
3482 fn a_declared_trigger_still_fires_after_create_table_migrated_its_source_table() {
3483 let dir = tempdir().unwrap();
3484 write_catalog_to(dir.path(), &catalog_of(&[users_schema(), orders_schema()]));
3485 let mut db = Database::open(dir.path()).unwrap();
3486 db.declare_trigger(users_insert_audits_into_orders()).unwrap();
3487
3488 db.create_table(
3489 "users",
3490 json!({
3491 "schemaVersion": 1,
3492 "table": "users",
3493 "revision": 2,
3494 "primaryKey": "id",
3495 "fields": {
3496 "id": { "type": "string" },
3497 "name": { "type": "string", "required": true },
3498 "city": { "type": "string" },
3499 },
3500 }),
3501 )
3502 .unwrap();
3503
3504 assert_eq!(db.table("users").unwrap().triggers().len(), 1);
3505
3506 let mut unit = db.unit_of_work();
3507 unit.put("users", json!({ "id": "u1", "name": "Ania", "city": "Kraków" }));
3508 db.commit(&mut unit).unwrap();
3509
3510 assert_eq!(
3511 db.table("orders")
3512 .unwrap()
3513 .get("o-audit")
3514 .unwrap()
3515 .unwrap()
3516 .get("userId"),
3517 Some(&json!("u1"))
3518 );
3519 db.close().unwrap();
3520 }
3521
3522 #[test]
3523 fn declare_trigger_refuses_an_unknown_effect_table_and_changes_nothing_on_disk() {
3524 let dir = tempdir().unwrap();
3525 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3526 let mut db = Database::open(dir.path()).unwrap();
3527
3528 let spec = json!({
3529 "name": "users_insert_audit",
3530 "table": "users",
3531 "event": "insert",
3532 "effect": {
3533 "table": "ghost",
3534 "fields": [
3535 { "target": "id", "source": { "kind": "constant", "value": "o-audit" } },
3536 ],
3537 },
3538 });
3539 let error = db.declare_trigger(spec);
3540 assert!(error.is_err());
3541 assert!(db.catalog.triggers.is_none());
3542
3543 let reread = read_catalog(&dir.path().join(MANIFEST_FILE)).unwrap().unwrap();
3544 assert!(reread.triggers.is_none());
3545 db.close().unwrap();
3546 }
3547
3548 mod automatic_compaction {
3549 use super::*;
3550 use crate::catalog::TableOptions;
3551
3552 fn enabled(dead_bytes_ratio: f64, min_dead_bytes: u64) -> CatalogCompaction {
3553 CatalogCompaction::Enabled {
3554 dead_bytes_ratio,
3555 min_dead_bytes,
3556 }
3557 }
3558
3559 fn opened(dir: &Path, compaction: Option<CatalogCompaction>) -> Database {
3560 Database::open_with(
3561 dir,
3562 OpenOptions {
3563 compaction,
3564 ..Default::default()
3565 },
3566 )
3567 .unwrap()
3568 }
3569
3570 fn lines_of(path: &Path) -> Vec<String> {
3571 fs::read_to_string(path).unwrap().lines().map(str::to_string).collect()
3572 }
3573
3574 fn churn(db: &mut Database, table: &str, key: &str, overwrites: usize) {
3577 let users = db.table_mut(table).unwrap();
3578 users.put(&json!({ "id": key, "name": "Ania" }), false).unwrap();
3579 for round in 0..overwrites {
3580 users
3581 .patch(key, &json!({ "name": format!("Ania {round}") }), false)
3582 .unwrap();
3583 }
3584 }
3585
3586 #[test]
3587 fn garbage_under_the_absolute_floor_leaves_the_log_alone_however_large_its_share() {
3588 let dir = tempdir().unwrap();
3589 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3590 let mut db = opened(dir.path(), Some(enabled(0.1, 1_000_000)));
3591
3592 churn(&mut db, "users", "u1", 12);
3593
3594 let stats = db.table("users").unwrap().stats();
3595 assert_eq!(stats.gen, 0, "minDeadBytes was never reached");
3596 assert!(stats.dead_bytes > 0, "the overwrites did make garbage");
3597 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
3598 db.close().unwrap();
3599 }
3600
3601 #[test]
3602 fn garbage_below_the_ratio_leaves_the_log_alone_however_many_bytes_it_is() {
3603 let dir = tempdir().unwrap();
3604 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3605 let mut db = opened(dir.path(), Some(enabled(0.99, 1)));
3606
3607 churn(&mut db, "users", "u1", 4);
3608
3609 let stats = db.table("users").unwrap().stats();
3610 assert_eq!(stats.gen, 0, "four of five lines is garbage, but not 99% of the bytes");
3611 assert!(stats.dead_bytes > 1);
3612 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 5);
3613 db.close().unwrap();
3614 }
3615
3616 #[test]
3617 fn crossing_both_thresholds_rewrites_the_log_and_moves_the_generation_on() {
3618 let dir = tempdir().unwrap();
3619 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3620 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
3621
3622 churn(&mut db, "users", "u1", 12);
3623
3624 let stats = db.table("users").unwrap().stats();
3625 assert!(stats.gen > 0, "the write that crossed the thresholds compacted");
3626 assert_eq!(stats.dead_bytes, 0);
3627 assert_eq!(stats.records, 1);
3628 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
3629 assert_eq!(
3630 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3631 Some(&json!("Ania 11"))
3632 );
3633 db.close().unwrap();
3634 }
3635
3636 #[test]
3637 fn a_log_the_thresholds_rewrote_reopens_at_the_generation_and_the_records_it_left() {
3638 let dir = tempdir().unwrap();
3639 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3640 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
3641 churn(&mut db, "users", "u1", 12);
3642 db.table_mut("users")
3643 .unwrap()
3644 .put(&json!({ "id": "u2", "name": "Piotr" }), false)
3645 .unwrap();
3646 let before = db.table("users").unwrap().stats();
3647 db.close().unwrap();
3648
3649 let reopened = opened(dir.path(), Some(enabled(0.5, 200)));
3650
3651 let after = reopened.table("users").unwrap().stats();
3652 assert_eq!(
3653 after.gen, before.gen,
3654 "the generation is carried by the snapshot, not reset"
3655 );
3656 assert_eq!(after.records, 2);
3657 assert_eq!(after.dead_bytes, 0);
3658 let mut keys: Vec<&String> = reopened.table("users").unwrap().keys().collect();
3659 keys.sort();
3660 assert_eq!(keys, vec!["u1", "u2"]);
3661 assert_eq!(
3662 reopened.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
3663 Some(&json!("Ania 11"))
3664 );
3665 reopened.close().unwrap();
3666 }
3667
3668 #[test]
3669 fn a_partitioned_table_is_measured_and_rewritten_one_partition_at_a_time() {
3670 let dir = tempdir().unwrap();
3671 let schema = parse_schema(&json!({
3672 "schemaVersion": 1,
3673 "table": "tickets",
3674 "primaryKey": "id",
3675 "partitionBy": "kind",
3676 "fields": {
3677 "id": { "type": "string" },
3678 "kind": { "type": "string", "required": true },
3679 "note": { "type": "string" },
3680 },
3681 }))
3682 .unwrap();
3683 write_catalog_to(dir.path(), &catalog_of(&[schema]));
3684 let mut db = opened(dir.path(), Some(enabled(0.5, 700)));
3686
3687 let tickets = db.table_mut("tickets").unwrap();
3688 tickets
3689 .put(&json!({ "id": "r1", "kind": "refund", "note": "n" }), false)
3690 .unwrap();
3691 let refund = dir.path().join("tickets").join("refund.jsonl");
3692 let quiet_size = fs::metadata(&refund).unwrap().len();
3693
3694 tickets
3695 .put(&json!({ "id": "s1", "kind": "sale", "note": "n" }), false)
3696 .unwrap();
3697 for round in 0..12 {
3698 tickets
3699 .patch("s1", &json!({ "note": format!("n{round}") }), false)
3700 .unwrap();
3701 }
3702
3703 let sale = dir.path().join("tickets").join("sale.jsonl");
3704 assert!(
3705 lines_of(&sale).len() < 13,
3706 "the partition over its thresholds was rewritten, so not all 13 lines are still there"
3707 );
3708 assert_eq!(
3709 fs::metadata(&refund).unwrap().len(),
3710 quiet_size,
3711 "the partition holding no garbage was left byte for byte"
3712 );
3713 assert_eq!(lines_of(&refund).len(), 1);
3714 let stats = db.table("tickets").unwrap().stats();
3715 assert_eq!(
3716 stats.gen, 1,
3717 "one pass moves the generation once, not once per partition"
3718 );
3719 assert_eq!(stats.records, 2);
3720 db.close().unwrap();
3721 }
3722
3723 #[test]
3724 fn a_catalog_declaring_compaction_false_keeps_the_garbage_an_open_call_named_nothing_about() {
3725 let dir = tempdir().unwrap();
3726 let mut catalog = catalog_of(&[users_schema()]);
3727 catalog.defaults = Some(TableOptions {
3728 compaction: Some(CatalogCompaction::Disabled),
3729 ..Default::default()
3730 });
3731 write_catalog_to(dir.path(), &catalog);
3732 let mut db = opened(dir.path(), None);
3733
3734 churn(&mut db, "users", "u1", 12);
3735
3736 assert_eq!(db.table("users").unwrap().stats().gen, 0);
3737 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 13);
3738 db.close().unwrap();
3739 }
3740
3741 #[test]
3742 fn a_tables_own_entry_beats_the_catalog_defaults_and_an_open_call_beats_them_both() {
3743 let dir = tempdir().unwrap();
3744 let mut catalog = catalog_of(&[users_schema()]);
3745 catalog.defaults = Some(TableOptions {
3746 compaction: Some(CatalogCompaction::Disabled),
3747 ..Default::default()
3748 });
3749 catalog.tables.get_mut("users").unwrap().options = Some(TableOptions {
3750 compaction: Some(enabled(0.5, 200)),
3751 ..Default::default()
3752 });
3753 write_catalog_to(dir.path(), &catalog);
3754
3755 let db = opened(dir.path(), None);
3756 assert_eq!(db.table("users").unwrap().compaction(), enabled(0.5, 200));
3757 db.close().unwrap();
3758
3759 let db = opened(dir.path(), Some(CatalogCompaction::Disabled));
3760 assert_eq!(db.table("users").unwrap().compaction(), CatalogCompaction::Disabled);
3761 db.close().unwrap();
3762 }
3763
3764 #[test]
3765 fn a_table_created_under_a_handle_is_armed_on_the_same_terms_as_one_opened_with_it() {
3766 let dir = tempdir().unwrap();
3767 let mut db = opened(dir.path(), Some(enabled(0.5, 200)));
3768 db.create_table(
3769 "users",
3770 json!({
3771 "schemaVersion": 1,
3772 "table": "users",
3773 "primaryKey": "id",
3774 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
3775 }),
3776 )
3777 .unwrap();
3778
3779 churn(&mut db, "users", "u1", 12);
3780
3781 assert!(db.table("users").unwrap().stats().gen > 0);
3782 assert_eq!(lines_of(&log_path(dir.path(), "users")).len(), 1);
3783 db.close().unwrap();
3784 }
3785
3786 #[test]
3787 fn an_open_call_naming_nothing_falls_back_to_the_engine_default_thresholds() {
3788 let dir = tempdir().unwrap();
3789 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
3790 let db = opened(dir.path(), None);
3791
3792 assert_eq!(db.table("users").unwrap().compaction(), DEFAULT_COMPACTION);
3793 db.close().unwrap();
3794 }
3795 }
3796
3797 mod read_cache {
3800 use super::*;
3801
3802 fn schema() -> Value {
3803 json!({
3804 "schemaVersion": 1,
3805 "table": "users",
3806 "primaryKey": "id",
3807 "fields": { "id": { "type": "string" }, "name": { "type": "string", "required": true } },
3808 })
3809 }
3810
3811 fn opened(dir: &Path, cache: CachePolicy) -> Database {
3812 Database::open_with(
3813 dir,
3814 OpenOptions {
3815 cache,
3816 ..Default::default()
3817 },
3818 )
3819 .unwrap()
3820 }
3821
3822 #[test]
3823 fn an_open_that_names_no_cache_leaves_every_table_caching_nothing() {
3824 let dir = tempdir().unwrap();
3825 let mut db = Database::open(dir.path()).unwrap();
3826 db.create_table("users", schema()).unwrap();
3827 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
3828 db.close().unwrap();
3829
3830 let db = Database::open(dir.path()).unwrap();
3831 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::Disabled);
3832 db.close().unwrap();
3833 }
3834
3835 #[test]
3836 fn the_policy_an_open_call_names_arms_the_tables_it_opens_and_the_ones_it_creates() {
3837 let dir = tempdir().unwrap();
3838 let mut db = opened(dir.path(), CachePolicy::with_entries(64));
3839 db.create_table("users", schema()).unwrap();
3840 assert_eq!(
3841 db.table("users").unwrap().cache_policy(),
3842 CachePolicy::with_entries(64),
3843 "a table created under the handle is armed on the same terms"
3844 );
3845 db.close().unwrap();
3846
3847 let db = opened(dir.path(), CachePolicy::with_entries(64));
3848 assert_eq!(db.table("users").unwrap().cache_policy(), CachePolicy::with_entries(64));
3849 db.close().unwrap();
3850 }
3851
3852 #[test]
3853 fn a_cached_table_read_through_the_database_answers_the_same_document_twice() {
3854 let dir = tempdir().unwrap();
3855 let mut db = opened(dir.path(), CachePolicy::with_entries(8));
3856 db.create_table("users", schema()).unwrap();
3857 let users = db.table_mut("users").unwrap();
3858 users.put(&json!({ "id": "u1", "name": "Ania" }), false).unwrap();
3859
3860 let first = users.get("u1").unwrap();
3861 let second = users.get("u1").unwrap();
3862
3863 assert_eq!(first, second);
3864 assert_eq!(first.unwrap().get("name"), Some(&json!("Ania")));
3865 db.close().unwrap();
3866 }
3867 }
3868
3869 mod cypher {
3870 use super::*;
3871 use crate::crypto::{CipherOptions, MissingKeyPolicy};
3872 use std::collections::HashMap;
3873
3874 fn key(byte: u8) -> Vec<u8> {
3875 vec![byte; 32]
3876 }
3877
3878 fn secrets_schema_json() -> Value {
3879 json!({
3880 "schemaVersion": 1,
3881 "table": "secrets",
3882 "primaryKey": "id",
3883 "fields": {
3884 "id": { "type": "string" },
3885 "token": { "type": "string", "cypher": true },
3886 },
3887 })
3888 }
3889
3890 fn secrets_schema_json_with_note() -> Value {
3891 json!({
3892 "schemaVersion": 1,
3893 "table": "secrets",
3894 "primaryKey": "id",
3895 "fields": {
3896 "id": { "type": "string" },
3897 "token": { "type": "string", "cypher": true },
3898 "note": { "type": "string" },
3899 },
3900 })
3901 }
3902
3903 fn open_with_keys(dir: &Path, keys: &[(&str, u8)], key_id: Option<&str>) -> Database {
3904 Database::open_with(
3905 dir,
3906 OpenOptions {
3907 cypher: Some(CipherOptions {
3908 keys: keys.iter().map(|(id, byte)| (id.to_string(), key(*byte))).collect(),
3909 key_id: key_id.map(str::to_string),
3910 on_missing_key: None,
3911 }),
3912 ..OpenOptions::default()
3913 },
3914 )
3915 .unwrap()
3916 }
3917
3918 #[test]
3919 fn a_cypher_field_round_trips_through_open_created_table_and_reopen() {
3920 let dir = tempdir().unwrap();
3921 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
3922 db.create_table("secrets", secrets_schema_json()).unwrap();
3923 db.table_mut("secrets")
3924 .unwrap()
3925 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
3926 .unwrap();
3927 db.close().unwrap();
3928
3929 let reopened = open_with_keys(dir.path(), &[("k1", 1)], None);
3930 assert_eq!(
3931 reopened
3932 .table("secrets")
3933 .unwrap()
3934 .get("s1")
3935 .unwrap()
3936 .unwrap()
3937 .get("token"),
3938 Some(&json!("hunter2"))
3939 );
3940 reopened.close().unwrap();
3941 }
3942
3943 #[test]
3944 fn reads_with_an_old_key_and_writes_with_the_new_one_after_rotation() {
3945 let dir = tempdir().unwrap();
3946 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
3947 db.create_table("secrets", secrets_schema_json()).unwrap();
3948 db.table_mut("secrets")
3949 .unwrap()
3950 .put(&json!({ "id": "s1", "token": "written-with-k1" }), false)
3951 .unwrap();
3952 db.close().unwrap();
3953
3954 let mut rotated = open_with_keys(dir.path(), &[("k1", 1), ("k2", 2)], Some("k2"));
3955 assert_eq!(
3956 rotated
3957 .table("secrets")
3958 .unwrap()
3959 .get("s1")
3960 .unwrap()
3961 .unwrap()
3962 .get("token"),
3963 Some(&json!("written-with-k1"))
3964 );
3965 rotated
3966 .table_mut("secrets")
3967 .unwrap()
3968 .put(&json!({ "id": "s2", "token": "written-with-k2" }), false)
3969 .unwrap();
3970 rotated.close().unwrap();
3971
3972 let log = std::fs::read_to_string(dir.path().join("secrets.jsonl")).unwrap();
3973 assert!(log.contains("$cypher$v1$k1$"));
3974 assert!(log.contains("$cypher$v1$k2$"));
3975 }
3976
3977 #[test]
3978 fn keeps_the_keyring_across_drop_field_reopening_the_table() {
3979 let dir = tempdir().unwrap();
3980 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
3981 db.create_table("secrets", secrets_schema_json_with_note()).unwrap();
3982 db.table_mut("secrets")
3983 .unwrap()
3984 .put(&json!({ "id": "s1", "token": "hunter2", "note": "x" }), false)
3985 .unwrap();
3986
3987 db.drop_field("secrets", "note", true).unwrap();
3988 assert_eq!(
3989 db.table("secrets").unwrap().get("s1").unwrap().unwrap().get("token"),
3990 Some(&json!("hunter2"))
3991 );
3992 db.close().unwrap();
3993 }
3994
3995 #[test]
3996 fn open_refuses_a_key_that_is_not_32_bytes() {
3997 let dir = tempdir().unwrap();
3998 let error = Database::open_with(
3999 dir.path(),
4000 OpenOptions {
4001 cypher: Some(CipherOptions {
4002 keys: HashMap::from([("k".to_string(), vec![0u8; 16])]),
4003 key_id: None,
4004 on_missing_key: None,
4005 }),
4006 ..OpenOptions::default()
4007 },
4008 )
4009 .map(|_| ())
4010 .unwrap_err();
4011 assert!(matches!(error, DatabaseError::Cipher(_)));
4012 }
4013
4014 #[test]
4015 fn missing_key_policy_raw_hands_back_the_envelope_through_the_database() {
4016 let dir = tempdir().unwrap();
4017 let mut db = open_with_keys(dir.path(), &[("k1", 1)], None);
4018 db.create_table("secrets", secrets_schema_json()).unwrap();
4019 db.table_mut("secrets")
4020 .unwrap()
4021 .put(&json!({ "id": "s1", "token": "hunter2" }), false)
4022 .unwrap();
4023 db.close().unwrap();
4024
4025 let raw = Database::open_with(
4026 dir.path(),
4027 OpenOptions {
4028 cypher: Some(CipherOptions {
4029 keys: HashMap::from([("other".to_string(), key(2))]),
4030 key_id: None,
4031 on_missing_key: Some(MissingKeyPolicy::Raw),
4032 }),
4033 ..OpenOptions::default()
4034 },
4035 )
4036 .unwrap();
4037 let found = raw.table("secrets").unwrap().get("s1").unwrap().unwrap();
4038 assert!(found.get("token").unwrap().as_str().unwrap().contains("$cypher$v1$k1$"));
4039 raw.close().unwrap();
4040 }
4041 }
4042
4043 mod partitions {
4044 use super::*;
4045
4046 fn notes_schema_json(partitioned: bool) -> Value {
4047 let mut spec = json!({
4048 "schemaVersion": 1,
4049 "table": "notes",
4050 "primaryKey": "id",
4051 "fields": {
4052 "id": { "type": "string" },
4053 "kind": { "type": "string", "required": true },
4054 "title": { "type": "string", "required": true },
4055 "body": { "type": "string" },
4056 },
4057 });
4058 if partitioned {
4059 spec.as_object_mut()
4060 .unwrap()
4061 .insert("partitionBy".into(), json!("kind"));
4062 }
4063 spec
4064 }
4065
4066 fn lines_in(dir: &Path, part: &str) -> usize {
4067 fs::read_to_string(dir.join("notes").join(format!("{part}.jsonl")))
4068 .unwrap()
4069 .trim()
4070 .lines()
4071 .filter(|line| !line.is_empty())
4072 .count()
4073 }
4074
4075 #[test]
4076 fn a_partitioned_table_survives_the_catalog_round_trip() {
4077 let dir = tempdir().unwrap();
4078 {
4079 let mut db = Database::open(dir.path()).unwrap();
4080 db.create_table("notes", notes_schema_json(true)).unwrap();
4081 let notes = db.table_mut("notes").unwrap();
4082 notes
4083 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4084 .unwrap();
4085 notes
4086 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4087 .unwrap();
4088 db.close().unwrap();
4089 }
4090
4091 let db = Database::open(dir.path()).unwrap();
4092 let notes = db.table("notes").unwrap();
4093 assert_eq!(notes.schema().partition_by.as_deref(), Some("kind"));
4094 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4095 assert_eq!(lines_in(dir.path(), "note"), 1);
4096 assert_eq!(lines_in(dir.path(), "adr"), 1);
4097 assert!(!dir.path().join("notes.jsonl").exists());
4098 db.close().unwrap();
4099 }
4100
4101 #[test]
4102 fn declaring_partitions_on_an_existing_table_moves_its_records_into_the_new_layout() {
4103 let dir = tempdir().unwrap();
4104 {
4105 let mut db = Database::open(dir.path()).unwrap();
4106 db.create_table("notes", notes_schema_json(false)).unwrap();
4107 let notes = db.table_mut("notes").unwrap();
4108 notes
4109 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4110 .unwrap();
4111 notes
4112 .put(&json!({ "id": "a1", "kind": "adr", "title": "second" }), false)
4113 .unwrap();
4114 notes.delete("n1", false).unwrap();
4115 db.close().unwrap();
4116 }
4117
4118 let mut db = Database::open(dir.path()).unwrap();
4119 let mut partitioned = notes_schema_json(true);
4120 partitioned.as_object_mut().unwrap().insert("revision".into(), json!(2));
4121 db.create_table("notes", partitioned).unwrap();
4122
4123 let notes = db.table("notes").unwrap();
4124 assert_eq!(notes.get("n1").unwrap(), None);
4125 assert_eq!(notes.get("a1").unwrap().unwrap().get("title"), Some(&json!("second")));
4126 assert!(!dir.path().join("notes.jsonl").exists());
4127 assert_eq!(lines_in(dir.path(), "adr"), 1);
4128 db.close().unwrap();
4129 }
4130
4131 #[test]
4132 fn dropping_the_field_the_table_partitions_by_is_refused() {
4133 let dir = tempdir().unwrap();
4134 let mut db = Database::open(dir.path()).unwrap();
4135 db.create_table("notes", notes_schema_json(true)).unwrap();
4136 db.table_mut("notes")
4137 .unwrap()
4138 .put(&json!({ "id": "n1", "kind": "note", "title": "first" }), false)
4139 .unwrap();
4140
4141 let error = db.drop_field("notes", "kind", true).unwrap_err();
4142 assert!(error.to_string().contains("partitioned by"));
4143 assert_eq!(
4144 db.table("notes").unwrap().get("n1").unwrap().unwrap().get("kind"),
4145 Some(&json!("note"))
4146 );
4147 db.close().unwrap();
4148 }
4149
4150 #[test]
4151 fn dropping_an_unrelated_field_keeps_every_partition_readable() {
4152 let dir = tempdir().unwrap();
4153 let mut db = Database::open(dir.path()).unwrap();
4154 db.create_table("notes", notes_schema_json(true)).unwrap();
4155 {
4156 let notes = db.table_mut("notes").unwrap();
4157 notes
4158 .put(
4159 &json!({ "id": "n1", "kind": "note", "title": "first", "body": "gone soon" }),
4160 false,
4161 )
4162 .unwrap();
4163 notes
4164 .put(&json!({ "id": "a1", "kind": "adr", "title": "kept" }), false)
4165 .unwrap();
4166 }
4167
4168 db.drop_field("notes", "body", true).unwrap();
4169
4170 let notes = db.table("notes").unwrap();
4171 assert!(!notes.schema().fields.contains_key("body"));
4172 assert_eq!(notes.get("n1").unwrap().unwrap().get("title"), Some(&json!("first")));
4173 assert_eq!(notes.get("a1").unwrap().unwrap().get("kind"), Some(&json!("adr")));
4174 assert_eq!(lines_in(dir.path(), "note"), 1);
4175 assert_eq!(lines_in(dir.path(), "adr"), 1);
4176 db.close().unwrap();
4177 }
4178
4179 #[test]
4180 fn a_unit_of_work_lands_in_two_partitions_of_one_table() {
4181 let dir = tempdir().unwrap();
4182 let mut db = Database::open(dir.path()).unwrap();
4183 db.create_table("notes", notes_schema_json(true)).unwrap();
4184
4185 let mut unit = db.unit_of_work();
4186 unit.put("notes", json!({ "id": "n1", "kind": "note", "title": "from unit" }));
4187 unit.put("notes", json!({ "id": "a1", "kind": "adr", "title": "from unit too" }));
4188 let result = db.commit(&mut unit).unwrap();
4189 assert_eq!(result.operations, 2);
4190
4191 let notes = db.table("notes").unwrap();
4192 assert_eq!(
4193 notes.get("n1").unwrap().unwrap().get("title"),
4194 Some(&json!("from unit"))
4195 );
4196 assert_eq!(
4197 notes.get("a1").unwrap().unwrap().get("title"),
4198 Some(&json!("from unit too"))
4199 );
4200 assert_eq!(lines_in(dir.path(), "note"), 1);
4201 assert_eq!(lines_in(dir.path(), "adr"), 1);
4202 db.close().unwrap();
4203 }
4204 }
4205 mod following_the_log {
4206 use super::*;
4207
4208 fn readonly(follow: FollowOption) -> OpenOptions {
4209 OpenOptions {
4210 readonly: true,
4211 follow,
4212 ..Default::default()
4213 }
4214 }
4215
4216 fn writer_and_reader(dir: &Path, follow: FollowOption) -> (Database, Database) {
4220 let mut writer = Database::open(dir).unwrap();
4221 writer.create_table("users", users_schema_spec()).unwrap();
4222 writer
4223 .table_mut("users")
4224 .unwrap()
4225 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4226 .unwrap();
4227 let reader = Database::open_with(dir, readonly(follow)).unwrap();
4228 (writer, reader)
4229 }
4230
4231 fn users_schema_spec() -> Value {
4232 json!({
4233 "schemaVersion": 1,
4234 "table": "users",
4235 "primaryKey": "id",
4236 "fields": {
4237 "id": { "type": "string" },
4238 "name": { "type": "string", "required": true },
4239 },
4240 })
4241 }
4242
4243 fn put(writer: &mut Database, key: &str, name: &str) {
4244 writer
4245 .table_mut("users")
4246 .unwrap()
4247 .put(&json!({ "id": key, "name": name }), false)
4248 .unwrap();
4249 }
4250
4251 #[test]
4252 fn a_readonly_open_follows_by_default_and_a_writable_one_does_not() {
4253 let dir = tempdir().unwrap();
4254 let mut writer = Database::open(dir.path()).unwrap();
4255 writer.create_table("users", users_schema_spec()).unwrap();
4256
4257 assert_eq!(writer.follow(), None);
4258 assert_eq!(writer.table("users").unwrap().follow(), None);
4259
4260 let reader = Database::open_with(dir.path(), readonly(FollowOption::Default)).unwrap();
4261 assert_eq!(
4262 reader.follow(),
4263 Some(FollowPolicy {
4264 interval: DEFAULT_FOLLOW_INTERVAL
4265 })
4266 );
4267 assert_eq!(
4268 reader.table("users").unwrap().follow(),
4269 Some(FollowPolicy {
4270 interval: DEFAULT_FOLLOW_INTERVAL
4271 })
4272 );
4273 }
4274
4275 #[test]
4276 fn a_plain_get_on_a_following_reader_sees_what_the_writer_committed_after_it_opened() {
4277 let dir = tempdir().unwrap();
4278 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4279
4280 for index in 2..=5 {
4281 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4282 let found = reader
4283 .table_mut("users")
4284 .unwrap()
4285 .get_following(&format!("u{index}"))
4286 .unwrap();
4287 assert_eq!(found.unwrap().get("name"), Some(&json!(format!("Osoba {index}"))));
4288 }
4289 }
4290
4291 #[test]
4292 fn a_plain_query_on_a_following_reader_counts_what_the_writer_committed() {
4293 let dir = tempdir().unwrap();
4294 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4295
4296 let spec = QueryBuilder::new().spec();
4297 assert_eq!(
4298 reader
4299 .table_mut("users")
4300 .unwrap()
4301 .query_following(&spec)
4302 .unwrap()
4303 .matched,
4304 1
4305 );
4306
4307 put(&mut writer, "u2", "Osoba 2");
4308
4309 assert_eq!(
4310 reader
4311 .table_mut("users")
4312 .unwrap()
4313 .query_following(&spec)
4314 .unwrap()
4315 .matched,
4316 2
4317 );
4318 }
4319
4320 #[test]
4321 fn a_reader_with_follow_off_stands_still_until_refresh_is_called() {
4322 let dir = tempdir().unwrap();
4323 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Off);
4324 put(&mut writer, "u2", "Osoba 2");
4325
4326 assert!(reader
4327 .table_mut("users")
4328 .unwrap()
4329 .get_following("u2")
4330 .unwrap()
4331 .is_none());
4332 assert!(reader.table_mut("users").unwrap().refresh().unwrap().changed());
4333 assert!(reader
4334 .table_mut("users")
4335 .unwrap()
4336 .get_following("u2")
4337 .unwrap()
4338 .is_some());
4339 }
4340
4341 #[test]
4342 fn a_custom_interval_coalesces_the_freshness_check_and_lets_it_through_once_it_elapses() {
4343 let dir = tempdir().unwrap();
4344 let (mut writer, mut reader) =
4345 writer_and_reader(dir.path(), FollowOption::Every(Duration::from_millis(150)));
4346
4347 assert!(reader
4349 .table_mut("users")
4350 .unwrap()
4351 .get_following("u1")
4352 .unwrap()
4353 .is_some());
4354 put(&mut writer, "u2", "Osoba 2");
4355 assert!(reader
4356 .table_mut("users")
4357 .unwrap()
4358 .get_following("u2")
4359 .unwrap()
4360 .is_none());
4361
4362 std::thread::sleep(Duration::from_millis(200));
4363
4364 assert!(reader
4365 .table_mut("users")
4366 .unwrap()
4367 .get_following("u2")
4368 .unwrap()
4369 .is_some());
4370 }
4371
4372 #[test]
4373 fn get_many_and_the_aggregates_follow_the_log_the_same_way_a_single_get_does() {
4374 let dir = tempdir().unwrap();
4375 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4376 put(&mut writer, "u2", "Osoba 2");
4377
4378 let keys = vec!["u1".to_string(), "u2".to_string()];
4379 let rows = reader.table_mut("users").unwrap().get_many_following(&keys).unwrap();
4380 assert!(rows.iter().all(Option::is_some), "{rows:?}");
4381
4382 put(&mut writer, "u3", "Osoba 3");
4383 let spec = QueryBuilder::new().spec();
4384 let aggregate = crate::query::aggregate::AggregateSpec {
4385 count: true,
4386 ..Default::default()
4387 };
4388 let counted = reader
4389 .table_mut("users")
4390 .unwrap()
4391 .aggregate_following(&spec, &aggregate)
4392 .unwrap();
4393 assert_eq!(counted.count, 3);
4394 }
4395
4396 #[test]
4397 fn a_following_reader_reloads_instead_of_reading_the_offsets_a_foreign_compaction_moved() {
4398 let dir = tempdir().unwrap();
4399 let (mut writer, mut reader) = writer_and_reader(dir.path(), FollowOption::Default);
4400 for index in 2..=6 {
4401 put(&mut writer, &format!("u{index}"), &format!("Osoba {index}"));
4402 }
4403 writer.table_mut("users").unwrap().delete("u5", false).unwrap();
4404 assert!(reader
4405 .table_mut("users")
4406 .unwrap()
4407 .get_following("u6")
4408 .unwrap()
4409 .is_some());
4410
4411 writer.table_mut("users").unwrap().compact().unwrap();
4412
4413 let users = reader.table_mut("users").unwrap();
4414 assert_eq!(
4415 users.get_following("u3").unwrap().unwrap().get("name"),
4416 Some(&json!("Osoba 3"))
4417 );
4418 assert!(users.get_following("u5").unwrap().is_none());
4419 assert_eq!(users.query_following(&QueryBuilder::new().spec()).unwrap().matched, 5);
4420 }
4421 }
4422
4423 mod where_the_files_land {
4424 use super::*;
4425 use crate::database::LayoutKind;
4426
4427 fn notes_schema_spec() -> Value {
4428 json!({
4429 "schemaVersion": 1,
4430 "table": "notes",
4431 "primaryKey": "id",
4432 "partitionBy": "kind",
4433 "fields": {
4434 "id": { "type": "string" },
4435 "kind": { "type": "string", "required": true, "index": "exact" },
4436 "title": { "type": "string", "required": true },
4437 },
4438 })
4439 }
4440
4441 fn nested(metadata_file: Option<&str>) -> OpenOptions {
4442 OpenOptions {
4443 layout: LayoutOptions {
4444 kind: Some(LayoutKind::Nested),
4445 metadata_file: metadata_file.map(str::to_string),
4446 ..Default::default()
4447 },
4448 ..Default::default()
4449 }
4450 }
4451
4452 #[test]
4453 fn the_default_open_writes_exactly_what_it_wrote_before_layouts_existed() {
4454 let dir = tempdir().unwrap();
4455 let mut db = Database::open(dir.path()).unwrap();
4456 db.create_table(
4457 "users",
4458 json!({
4459 "schemaVersion": 1,
4460 "table": "users",
4461 "primaryKey": "id",
4462 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4463 }),
4464 )
4465 .unwrap();
4466 db.table_mut("users")
4467 .unwrap()
4468 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4469 .unwrap();
4470 db.close().unwrap();
4471
4472 assert!(dir.path().join("hayson.json").is_file());
4473 assert!(dir.path().join("users.jsonl").is_file());
4474 assert!(!dir.path().join("tables").exists());
4475 assert!(!dir.path().join("runtime").exists());
4476
4477 let db = Database::open(dir.path()).unwrap();
4478 assert_eq!(db.layout().kind(), LayoutKind::Flat);
4479 assert_eq!(db.layout().metadata_file(), "hayson.json");
4480 assert_eq!(db.layout().metadata_path(), dir.path().join("hayson.json"));
4481 }
4482
4483 #[test]
4484 fn a_custom_metadata_file_holds_the_catalog_and_nothing_is_written_under_the_default_name() {
4485 let dir = tempdir().unwrap();
4486 let options = OpenOptions {
4487 layout: LayoutOptions {
4488 metadata_file: Some("cerebro.json".into()),
4489 ..Default::default()
4490 },
4491 ..Default::default()
4492 };
4493
4494 let mut db = Database::open_with(dir.path(), options.clone()).unwrap();
4495 db.create_table(
4496 "users",
4497 json!({
4498 "schemaVersion": 1,
4499 "table": "users",
4500 "primaryKey": "id",
4501 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4502 }),
4503 )
4504 .unwrap();
4505 db.table_mut("users")
4506 .unwrap()
4507 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4508 .unwrap();
4509 db.close().unwrap();
4510
4511 assert!(dir.path().join("cerebro.json").is_file());
4512 assert!(!dir.path().join("hayson.json").exists());
4513
4514 let db = Database::open_with(dir.path(), options).unwrap();
4515 assert_eq!(db.layout().metadata_file(), "cerebro.json");
4516 assert_eq!(db.table_names(), vec!["users"]);
4517 assert!(db.table("users").unwrap().get("u1").unwrap().is_some());
4518 }
4519
4520 #[test]
4521 fn the_deprecated_manifest_name_opens_what_the_new_one_wrote_and_the_other_way_round() {
4522 let dir = tempdir().unwrap();
4523 let legacy = OpenOptions {
4524 layout: LayoutOptions {
4525 manifest: Some("cerebro.json".into()),
4526 ..Default::default()
4527 },
4528 ..Default::default()
4529 };
4530 let mut db = Database::open_with(dir.path(), legacy).unwrap();
4531 db.create_table(
4532 "users",
4533 json!({
4534 "schemaVersion": 1,
4535 "table": "users",
4536 "primaryKey": "id",
4537 "fields": { "id": { "type": "string" } },
4538 }),
4539 )
4540 .unwrap();
4541 db.close().unwrap();
4542
4543 let current = OpenOptions {
4544 layout: LayoutOptions {
4545 metadata_file: Some("cerebro.json".into()),
4546 ..Default::default()
4547 },
4548 ..Default::default()
4549 };
4550 let db = Database::open_with(dir.path(), current).unwrap();
4551 assert_eq!(db.table_names(), vec!["users"]);
4552 assert_eq!(db.layout().metadata_path(), dir.path().join("cerebro.json"));
4553 }
4554
4555 #[test]
4556 fn the_two_names_for_the_metadata_document_are_refused_when_they_disagree_touching_nothing() {
4557 let dir = tempdir().unwrap();
4558 let options = OpenOptions {
4559 layout: LayoutOptions {
4560 metadata_file: Some("cerebro.json".into()),
4561 manifest: Some("hayson.json".into()),
4562 ..Default::default()
4563 },
4564 ..Default::default()
4565 };
4566
4567 let Err(error) = Database::open_with(dir.path(), options) else {
4568 panic!("two disagreeing names for the metadata document were accepted")
4569 };
4570
4571 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
4572 let message = error.to_string();
4573 assert!(
4574 message.contains("metadataFile") && message.contains("manifest"),
4575 "{message}"
4576 );
4577 assert!(!dir.path().join("hayson.json").exists());
4578 assert!(!dir.path().join("cerebro.json").exists());
4579 assert!(!dir.path().join(crate::lock::LOCK_FILE).exists());
4580 }
4581
4582 #[test]
4583 fn the_nested_layout_sorts_the_logs_the_snapshots_and_the_lock_into_their_subdirectories() {
4584 let dir = tempdir().unwrap();
4585 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4586 db.create_table(
4587 "users",
4588 json!({
4589 "schemaVersion": 1,
4590 "table": "users",
4591 "primaryKey": "id",
4592 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4593 }),
4594 )
4595 .unwrap();
4596 db.table_mut("users")
4597 .unwrap()
4598 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4599 .unwrap();
4600
4601 assert!(dir.path().join("runtime").join(crate::lock::LOCK_FILE).is_file());
4602 db.close().unwrap();
4603
4604 assert!(dir.path().join("hayson.json").is_file());
4605 assert!(dir.path().join("tables").join("users.jsonl").is_file());
4606 assert!(dir.path().join("index").join("users.idx").is_file());
4607 assert!(!dir.path().join("users.jsonl").exists());
4608 }
4609
4610 #[test]
4611 fn a_partitioned_table_under_the_nested_layout_lands_under_tables_slash_its_own_name() {
4612 let dir = tempdir().unwrap();
4613 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4614 db.create_table("notes", notes_schema_spec()).unwrap();
4615 db.table_mut("notes")
4616 .unwrap()
4617 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
4618 .unwrap();
4619 db.close().unwrap();
4620
4621 assert!(dir.path().join("tables").join("notes").join("note.jsonl").is_file());
4622 }
4623
4624 #[test]
4625 fn a_nested_database_reopens_as_nested_without_being_told_and_is_refused_as_flat() {
4626 let dir = tempdir().unwrap();
4627 let mut db = Database::open_with(dir.path(), nested(Some("cerebro.json"))).unwrap();
4628 db.create_table(
4629 "users",
4630 json!({
4631 "schemaVersion": 1,
4632 "table": "users",
4633 "primaryKey": "id",
4634 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4635 }),
4636 )
4637 .unwrap();
4638 db.table_mut("users")
4639 .unwrap()
4640 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4641 .unwrap();
4642 db.close().unwrap();
4643
4644 let named_only = OpenOptions {
4645 layout: LayoutOptions {
4646 metadata_file: Some("cerebro.json".into()),
4647 ..Default::default()
4648 },
4649 ..Default::default()
4650 };
4651 let db = Database::open_with(dir.path(), named_only).unwrap();
4652 assert_eq!(db.layout().kind(), LayoutKind::Nested);
4653 assert_eq!(
4654 db.table("users").unwrap().get("u1").unwrap().unwrap().get("name"),
4655 Some(&json!("Ania"))
4656 );
4657 db.close().unwrap();
4658
4659 let as_flat = OpenOptions {
4660 layout: LayoutOptions {
4661 kind: Some(LayoutKind::Flat),
4662 metadata_file: Some("cerebro.json".into()),
4663 ..Default::default()
4664 },
4665 ..Default::default()
4666 };
4667 let Err(error) = Database::open_with(dir.path(), as_flat) else {
4668 panic!("a nested database was opened as flat")
4669 };
4670 assert!(matches!(error, DatabaseError::Schema(_)), "{error:?}");
4671 assert!(error.to_string().contains("nested"), "{error}");
4672 }
4673
4674 #[test]
4675 fn a_unit_of_work_under_the_nested_layout_journals_into_runtime_and_reopens_whole() {
4676 let dir = tempdir().unwrap();
4677 let mut db = Database::open_with(dir.path(), nested(None)).unwrap();
4678 db.create_table(
4679 "users",
4680 json!({
4681 "schemaVersion": 1,
4682 "table": "users",
4683 "primaryKey": "id",
4684 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4685 }),
4686 )
4687 .unwrap();
4688
4689 let mut unit = db.unit_of_work();
4690 unit.put("users", json!({ "id": "u1", "name": "Ania" }));
4691 unit.put("users", json!({ "id": "u2", "name": "Bartek" }));
4692 db.commit(&mut unit).unwrap();
4693 db.close().unwrap();
4694
4695 assert!(!dir.path().join("runtime").join(".hayson.txn").exists());
4696 let db = Database::open_with(dir.path(), nested(None)).unwrap();
4697 assert_eq!(db.table("users").unwrap().len(), 2);
4698 }
4699 }
4700
4701 mod a_declared_table_whose_log_is_not_on_disk {
4704 use super::*;
4705
4706 fn notes_spec() -> Value {
4707 json!({
4708 "schemaVersion": 1,
4709 "table": "notes",
4710 "primaryKey": "id",
4711 "partitionBy": "kind",
4712 "fields": {
4713 "id": { "type": "string" },
4714 "kind": { "type": "string", "required": true },
4715 "title": { "type": "string", "required": true },
4716 },
4717 })
4718 }
4719
4720 fn readonly() -> OpenOptions {
4721 OpenOptions {
4722 readonly: true,
4723 ..Default::default()
4724 }
4725 }
4726
4727 fn users_with_a_record(dir: &Path) {
4728 let mut db = Database::open(dir).unwrap();
4729 db.create_table(
4730 "users",
4731 json!({
4732 "schemaVersion": 1,
4733 "table": "users",
4734 "primaryKey": "id",
4735 "fields": { "id": { "type": "string" }, "name": { "type": "string" } },
4736 }),
4737 )
4738 .unwrap();
4739 db.table_mut("users")
4740 .unwrap()
4741 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4742 .unwrap();
4743 db.close().unwrap();
4744 }
4745
4746 fn notes_with_a_record(dir: &Path) {
4747 let mut db = Database::open(dir).unwrap();
4748 db.create_table("notes", notes_spec()).unwrap();
4749 db.table_mut("notes")
4750 .unwrap()
4751 .put(&json!({ "id": "n1", "kind": "note", "title": "Backend" }), false)
4752 .unwrap();
4753 db.close().unwrap();
4754 }
4755
4756 fn refusal(dir: &Path) -> DatabaseError {
4757 match Database::open_with(dir, readonly()) {
4758 Err(error) => error,
4759 Ok(_) => panic!("expected \"{}\" to be refused", dir.display()),
4760 }
4761 }
4762
4763 fn schema_refusal(error: DatabaseError) -> String {
4764 match error {
4765 DatabaseError::Schema(HaysonSchemaError(message)) => message,
4766 other => panic!("expected a schema refusal, got {other:?}"),
4767 }
4768 }
4769
4770 #[test]
4771 fn a_readonly_open_refuses_naming_the_table_and_the_log_without_advising_create_table() {
4772 let dir = tempdir().unwrap();
4773 users_with_a_record(dir.path());
4774 let log = log_path(dir.path(), "users");
4775 fs::remove_file(&log).unwrap();
4776
4777 let message = schema_refusal(refusal(dir.path()));
4778
4779 assert!(message.contains("\"users\""), "{message}");
4780 assert!(message.contains(&log.display().to_string()), "{message}");
4781 assert!(message.contains("readonly"), "{message}");
4782 assert!(!message.contains("createTable"), "{message}");
4783 assert!(!log.exists(), "the refused open must not have created the log");
4784 }
4785
4786 #[test]
4787 fn a_writable_open_of_the_same_directory_creates_the_log_and_serves_an_empty_table() {
4788 let dir = tempdir().unwrap();
4789 users_with_a_record(dir.path());
4790 fs::remove_file(log_path(dir.path(), "users")).unwrap();
4791
4792 let mut db = Database::open(dir.path()).unwrap();
4793
4794 assert_eq!(db.table_names(), vec!["users"]);
4795 assert!(log_path(dir.path(), "users").is_file());
4796 assert_eq!(db.table("users").unwrap().len(), 0);
4797 db.table_mut("users")
4798 .unwrap()
4799 .put(&json!({ "id": "u1", "name": "Ania" }), false)
4800 .unwrap();
4801 assert_eq!(db.table("users").unwrap().len(), 1);
4802 db.close().unwrap();
4803 }
4804
4805 #[test]
4807 fn a_readonly_open_accepts_a_log_that_is_present_and_empty() {
4808 let dir = tempdir().unwrap();
4809 users_with_a_record(dir.path());
4810 fs::write(log_path(dir.path(), "users"), "").unwrap();
4811
4812 let db = Database::open_with(dir.path(), readonly()).unwrap();
4813
4814 assert_eq!(db.table_names(), vec!["users"]);
4815 assert_eq!(db.table("users").unwrap().len(), 0);
4816 }
4817
4818 #[test]
4819 fn a_readonly_open_refuses_a_partitioned_table_whose_recorded_partition_is_gone() {
4820 let dir = tempdir().unwrap();
4821 notes_with_a_record(dir.path());
4822 let partition = dir.path().join("notes").join("note.jsonl");
4823 assert!(
4824 index_path(dir.path(), "notes").is_file(),
4825 "the refusal reads the snapshot"
4826 );
4827 fs::remove_dir_all(dir.path().join("notes")).unwrap();
4828
4829 let message = schema_refusal(refusal(dir.path()));
4830
4831 assert!(message.contains("\"notes\""), "{message}");
4832 assert!(message.contains(&partition.display().to_string()), "{message}");
4833 assert!(message.contains("readonly"), "{message}");
4834 assert!(!message.contains("createTable"), "{message}");
4835 }
4836
4837 #[test]
4838 fn a_readonly_open_accepts_a_partitioned_table_that_was_never_written_to() {
4839 let dir = tempdir().unwrap();
4840 let mut db = Database::open(dir.path()).unwrap();
4841 db.create_table("notes", notes_spec()).unwrap();
4842 db.close().unwrap();
4843
4844 let db = Database::open_with(dir.path(), readonly()).unwrap();
4845
4846 assert_eq!(db.table_names(), vec!["notes"]);
4847 assert_eq!(db.table("notes").unwrap().len(), 0);
4848 }
4849
4850 #[cfg(unix)]
4853 #[test]
4854 fn a_readonly_open_reports_a_partition_that_cannot_be_opened_against_its_own_path() {
4855 let dir = tempdir().unwrap();
4856 notes_with_a_record(dir.path());
4857 let partition = dir.path().join("notes").join("note.jsonl");
4858 fs::remove_file(&partition).unwrap();
4859 std::os::unix::fs::symlink(dir.path().join("notes").join("gone.jsonl"), &partition).unwrap();
4860
4861 let error = refusal(dir.path());
4862
4863 match error {
4864 DatabaseError::Log(path, inner) => {
4865 assert_eq!(path, partition);
4866 assert_eq!(inner.kind(), io::ErrorKind::NotFound);
4867 }
4868 other => panic!("expected a log refusal naming the partition, got {other:?}"),
4869 }
4870 assert!(!dir.path().join("notes").join("gone.jsonl").exists());
4871 }
4872 }
4873
4874 mod describing_one_table {
4875 use super::*;
4876
4877 fn users_with_city() -> TableSchema {
4878 parse_schema(&json!({
4879 "schemaVersion": 1,
4880 "table": "users",
4881 "revision": 2,
4882 "primaryKey": "id",
4883 "fields": {
4884 "id": { "type": "string" },
4885 "name": { "type": "string", "required": true },
4886 "city": { "type": "string", "index": "exact" },
4887 },
4888 }))
4889 .unwrap()
4890 }
4891
4892 fn users_drifted_under(dir: &Path) -> Database {
4895 write_catalog_to(dir, &catalog_of(&[users_schema(), orders_schema()]));
4896 let mut db = Database::open(dir).unwrap();
4897 write_catalog_to(dir, &catalog_of(&[users_with_city(), orders_schema()]));
4898 assert!(db.refresh().unwrap());
4899 db
4900 }
4901
4902 fn refusal_of(answered: Result<Value, DatabaseError>) -> String {
4903 match answered {
4904 Err(DatabaseError::Schema(error)) => error.to_string(),
4905 other => panic!("expected a schema refusal, got {other:?}"),
4906 }
4907 }
4908
4909 #[test]
4910 fn describes_a_table_of_its_own_while_another_one_has_drifted() {
4911 let dir = tempdir().unwrap();
4912 let db = users_drifted_under(dir.path());
4913
4914 let document = db.describe_table("orders").unwrap();
4915
4916 assert_eq!(document["table"], json!("orders"));
4917 assert_eq!(document["revision"], json!(1));
4918 assert!(db.catalog().is_err());
4919 }
4920
4921 #[test]
4922 fn refuses_the_table_that_drifted_itself() {
4923 let dir = tempdir().unwrap();
4924 let db = users_drifted_under(dir.path());
4925
4926 let message = refusal_of(db.describe_table("users"));
4927
4928 assert!(message.contains("\"users\""), "{message}");
4929 assert!(message.contains("revision 2"), "{message}");
4930 assert_eq!(db.table("users").unwrap().schema().revision, 1);
4932 }
4933
4934 #[test]
4935 fn describes_the_index_set_the_handle_serves_and_leaves_the_named_ones_out() {
4936 let dir = tempdir().unwrap();
4937 let mut catalog = catalog_of(&[users_schema()]);
4938 let entry = catalog.tables.get_mut("users").unwrap();
4939 entry.indexes = Some(vec![IndexEntry {
4940 name: "by_name".to_string(),
4941 fields: vec!["name".to_string()],
4942 kind: crate::catalog::CatalogIndexKind::Exact,
4943 unique: false,
4944 where_: None,
4945 expire_after_seconds: None,
4946 }]);
4947 write_catalog_to(dir.path(), &catalog);
4948 let db = Database::open(dir.path()).unwrap();
4949
4950 let document = db.describe_table("users").unwrap();
4951
4952 assert_eq!(
4953 document["fields"]["name"],
4954 json!({ "type": "string", "required": true })
4955 );
4956 assert!(document.as_object().unwrap().get("indexes").is_none());
4957 assert!(db.catalog().is_ok());
4958 }
4959
4960 #[test]
4961 fn refuses_a_table_this_handle_holds_nowhere() {
4962 let dir = tempdir().unwrap();
4963 write_catalog_to(dir.path(), &catalog_of(&[users_schema()]));
4964 let db = Database::open(dir.path()).unwrap();
4965
4966 let message = refusal_of(db.describe_table("ghosts"));
4967
4968 assert!(message.contains("declares no table \"ghosts\""), "{message}");
4969 }
4970 }
4971}