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