ursa_engine/database/
mod.rs

1//! The `.ursa` database file: a session's catalog, kept between processes.
2//!
3//! `ursa club.ursa` opens (or creates) a [`Database`] and installs it on the
4//! session. The design is the Tempra page `/sql/ursa-file.md` in
5//! `cl_dixon/ursa` (cldixon/ursa#237). On disk:
6//!
7//! ```text
8//! club.ursa          the catalog: the statements that defined each object
9//! club.ursa.d/       created on first use
10//!   lock             held while a process has the database open for writing
11//!   indexes/         stored graph indexes: `manifest.json` and `<hash>.idx`
12//! ```
13//!
14//! This module owns the file, the sidecar directory and the lock. It stores
15//! the catalog as an ordered list of statement texts ([`catalog_file`]) and
16//! knows nothing of SQL: which statement defines which object, and replaying
17//! them, is `ursa-sql`'s `database` module.
18//!
19//! A database is reached from the session config ([`database`],
20//! [`database_of_config`]), so any code holding the session state or a task
21//! context (the graph index cache, which stores indexes in it) can find it.
22//!
23//! **Relative paths** follow DuckDB. A relative file path in SQL (in `FROM`,
24//! a graph's edge or vertex table, an external table's `LOCATION`, a
25//! `COPY … TO` target) resolves against the current directory when the
26//! statement runs, with or without a database. When a definition is stored in
27//! the catalog, each of its local file paths is rewritten relative to the
28//! database's directory ([`Database::stored_location`]), or made absolute if
29//! it is outside that directory; replaying the catalog resolves the stored
30//! paths against the database's directory ([`Database::replayed_location`]).
31//! So a later process finds the same files from any directory, and a database
32//! and its data files can move together. URLs are never rewritten.
33
34use std::fs::File;
35use std::path::{Component, Path, PathBuf};
36use std::sync::{Arc, Mutex, OnceLock};
37use std::time::{Duration, Instant};
38
39use datafusion::error::{DataFusionError, Result};
40use datafusion::execution::context::SessionState;
41use datafusion::prelude::SessionConfig;
42
43pub mod catalog_file;
44pub mod index_file;
45pub mod index_store;
46
47pub use catalog_file::{CatalogFile, FORMAT, VERSION};
48pub use index_file::IndexKey;
49pub use index_store::IndexStore;
50
51/// How long a writer waits for another process to close the database.
52pub const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(2);
53
54/// How to open a database.
55#[derive(Debug, Clone)]
56pub struct OpenOptions {
57    /// Open without the lock, and refuse statements that change the catalog.
58    /// The file must exist.
59    pub read_only: bool,
60    /// How long to wait for the lock before failing with "… is in use by
61    /// another process".
62    pub lock_timeout: Duration,
63}
64
65impl Default for OpenOptions {
66    fn default() -> Self {
67        Self {
68            read_only: false,
69            lock_timeout: DEFAULT_LOCK_TIMEOUT,
70        }
71    }
72}
73
74impl OpenOptions {
75    pub fn read_only() -> Self {
76        Self {
77            read_only: true,
78            ..Self::default()
79        }
80    }
81}
82
83/// An open `.ursa` database.
84#[derive(Debug)]
85pub struct Database {
86    /// The path as the user gave it: how messages name the database.
87    name: String,
88    /// The catalog file, absolute.
89    path: PathBuf,
90    /// The directory holding the catalog file, absolute: stored paths are
91    /// relative to it.
92    dir: PathBuf,
93    /// `club.ursa.d`, absolute.
94    sidecar: PathBuf,
95    read_only: bool,
96    /// The locked `club.ursa.d/lock`, held until the database is dropped.
97    /// `None` when read-only.
98    _lock: Option<File>,
99    /// The catalog's statements, as last read or written.
100    statements: Mutex<Vec<String>>,
101    /// The stored graph indexes, opened on first use.
102    indexes: OnceLock<Arc<IndexStore>>,
103}
104
105impl Database {
106    /// Open the database at `path`, creating it (and its sidecar directory)
107    /// unless read-only. A writer takes the lock first, waiting up to
108    /// `options.lock_timeout`, then reads the catalog.
109    pub fn open(path: impl AsRef<Path>, options: &OpenOptions) -> Result<Self> {
110        let given = path.as_ref();
111        let name = given.display().to_string();
112        let path = std::path::absolute(given).map_err(|e| {
113            DataFusionError::Execution(format!("cannot open database '{name}': {e}"))
114        })?;
115        let dir = path
116            .parent()
117            .map(Path::to_path_buf)
118            .unwrap_or_else(|| PathBuf::from("/"));
119        let file_name = path.file_name().ok_or_else(|| {
120            DataFusionError::Execution(format!("'{name}' does not name a database file"))
121        })?;
122        let mut sidecar_name = file_name.to_os_string();
123        sidecar_name.push(".d");
124        let sidecar = dir.join(sidecar_name);
125
126        if options.read_only {
127            if !path.exists() {
128                return Err(DataFusionError::Execution(format!(
129                    "cannot open '{name}' read-only: the file does not exist"
130                )));
131            }
132            let catalog = CatalogFile::read(&path, &name)?;
133            return Ok(Self {
134                name,
135                path,
136                dir,
137                sidecar,
138                read_only: true,
139                _lock: None,
140                statements: Mutex::new(catalog.statements),
141                indexes: OnceLock::new(),
142            });
143        }
144
145        if !dir.is_dir() {
146            return Err(DataFusionError::Execution(format!(
147                "cannot create database '{name}': the directory {} does not exist",
148                dir.display()
149            )));
150        }
151        let io = |what: &str, e: std::io::Error| {
152            DataFusionError::Execution(format!("cannot open database '{name}': {what}: {e}"))
153        };
154        std::fs::create_dir_all(&sidecar)
155            .map_err(|e| io(&format!("creating {}", sidecar.display()), e))?;
156        let lock = File::options()
157            .read(true)
158            .write(true)
159            .create(true)
160            .truncate(false)
161            .open(sidecar.join("lock"))
162            .map_err(|e| io("opening its lock file", e))?;
163        acquire(&lock, &name, options.lock_timeout)?;
164
165        let statements = if path.exists() {
166            CatalogFile::read(&path, &name)?.statements
167        } else {
168            let empty = CatalogFile::default();
169            empty.write(&path, &sidecar, &name)?;
170            empty.statements
171        };
172        Ok(Self {
173            name,
174            path,
175            dir,
176            sidecar,
177            read_only: false,
178            _lock: Some(lock),
179            statements: Mutex::new(statements),
180            indexes: OnceLock::new(),
181        })
182    }
183
184    /// The database as the user named it (`club.ursa`).
185    pub fn name(&self) -> &str {
186        &self.name
187    }
188
189    /// The catalog file, absolute.
190    pub fn path(&self) -> &Path {
191        &self.path
192    }
193
194    /// The directory holding the catalog file: stored relative paths are
195    /// relative to it.
196    pub fn dir(&self) -> &Path {
197        &self.dir
198    }
199
200    /// The sidecar directory, `club.ursa.d`. It exists unless read-only.
201    pub fn sidecar_dir(&self) -> &Path {
202        &self.sidecar
203    }
204
205    /// Where stored graph indexes go (`club.ursa.d/indexes`), created on
206    /// the first write.
207    pub fn index_dir(&self) -> PathBuf {
208        self.sidecar.join("indexes")
209    }
210
211    /// The database's stored graph indexes. A read-only database reads them
212    /// and never writes.
213    pub fn index_store(&self) -> Arc<IndexStore> {
214        Arc::clone(
215            self.indexes
216                .get_or_init(|| Arc::new(IndexStore::new(self.index_dir(), !self.read_only))),
217        )
218    }
219
220    /// Wait for stored-index writes still running.
221    pub fn flush(&self) {
222        if let Some(store) = self.indexes.get() {
223            store.flush();
224        }
225    }
226
227    pub fn is_read_only(&self) -> bool {
228        self.read_only
229    }
230
231    /// The catalog's statements, in replay order.
232    pub fn statements(&self) -> Vec<String> {
233        self.statements
234            .lock()
235            .unwrap_or_else(|e| e.into_inner())
236            .clone()
237    }
238
239    /// Replace the catalog with `statements`, rewriting the file atomically
240    /// (temporary file, `fsync`, rename). Refused when read-only.
241    pub fn save(&self, statements: Vec<String>) -> Result<()> {
242        if self.read_only {
243            return Err(DataFusionError::Plan(format!(
244                "{} is open read-only",
245                self.name
246            )));
247        }
248        let mut current = self.statements.lock().unwrap_or_else(|e| e.into_inner());
249        if *current == statements {
250            return Ok(());
251        }
252        let file = CatalogFile {
253            statements: statements.clone(),
254        };
255        file.write(&self.path, &self.sidecar, &self.name)?;
256        *current = statements;
257        Ok(())
258    }
259
260    /// How the catalog stores `location`, written in a statement run from
261    /// `cwd`: a local path inside the database's directory becomes relative to
262    /// it, any other local path becomes absolute. `None` when the location is
263    /// stored as written: a URL, a `~/…` path, or a path already in its
264    /// stored form.
265    pub fn stored_location(&self, location: &str, cwd: &Path) -> Option<String> {
266        if !is_local_path(location) {
267            return None;
268        }
269        let absolute = normalize(&cwd.join(location));
270        let canonical_dir = std::fs::canonicalize(&self.dir).ok();
271        let inside = [Some(&self.dir), canonical_dir.as_ref()]
272            .into_iter()
273            .flatten()
274            .find_map(|dir| absolute.strip_prefix(dir).ok());
275        let mut stored = match inside {
276            Some(relative) if relative.as_os_str().is_empty() => ".".to_string(),
277            Some(relative) => relative.to_string_lossy().into_owned(),
278            None => absolute.to_string_lossy().into_owned(),
279        };
280        keep_trailing_separator(location, &mut stored);
281        (stored != location).then_some(stored)
282    }
283
284    /// `location` from a stored statement as replay runs it: a relative local
285    /// path resolved against the database's directory; `None` for anything
286    /// else (a URL, an absolute path, `~/…`), which runs as stored.
287    pub fn replayed_location(&self, location: &str) -> Option<String> {
288        if !is_relative_path(location) {
289            return None;
290        }
291        let mut resolved = normalize(&self.dir.join(location))
292            .to_string_lossy()
293            .into_owned();
294        keep_trailing_separator(location, &mut resolved);
295        Some(resolved)
296    }
297}
298
299impl Drop for Database {
300    /// Finish writing stored indexes before the lock is released.
301    fn drop(&mut self) {
302        self.flush();
303    }
304}
305
306/// Whether `location` is a local path (not a URL, not starting with `~`).
307pub fn is_local_path(location: &str) -> bool {
308    if location.is_empty() || location.starts_with('~') {
309        return false;
310    }
311    match url::Url::parse(location) {
312        // A one-letter scheme is a Windows drive letter.
313        Ok(url) => url.scheme().len() == 1,
314        Err(_) => true,
315    }
316}
317
318/// Whether `location` is a relative local path (not a URL, not absolute, not
319/// starting with `~`).
320pub fn is_relative_path(location: &str) -> bool {
321    is_local_path(location) && Path::new(location).is_relative()
322}
323
324/// `path` with `.` and `..` components removed lexically, without touching
325/// the filesystem (the path may be a glob, or not exist yet).
326fn normalize(path: &Path) -> PathBuf {
327    let mut out = PathBuf::new();
328    for component in path.components() {
329        match component {
330            Component::CurDir => {}
331            Component::ParentDir => match out.components().next_back() {
332                Some(Component::Normal(_)) => {
333                    out.pop();
334                }
335                // `..` at the root stays at the root.
336                Some(Component::RootDir | Component::Prefix(_)) => {}
337                _ => out.push(".."),
338            },
339            other => out.push(other),
340        }
341    }
342    out
343}
344
345/// A directory written with a trailing separator (`data/`) keeps it.
346fn keep_trailing_separator(original: &str, rewritten: &mut String) {
347    let separator = std::path::MAIN_SEPARATOR;
348    if (original.ends_with('/') || original.ends_with(separator))
349        && !rewritten.ends_with(['/', separator])
350    {
351        rewritten.push(separator);
352    }
353}
354
355/// Take the exclusive lock on `lock`, retrying until `timeout` has passed.
356fn acquire(lock: &File, name: &str, timeout: Duration) -> Result<()> {
357    let start = Instant::now();
358    loop {
359        match lock.try_lock() {
360            Ok(()) => return Ok(()),
361            Err(std::fs::TryLockError::WouldBlock) => {
362                if start.elapsed() >= timeout {
363                    return Err(DataFusionError::Execution(format!(
364                        "{name} is in use by another process"
365                    )));
366                }
367                std::thread::sleep(Duration::from_millis(50));
368            }
369            Err(std::fs::TryLockError::Error(e)) => {
370                return Err(DataFusionError::Execution(format!(
371                    "cannot lock database '{name}': {e}"
372                )))
373            }
374        }
375    }
376}
377
378/// Where a session keeps its database: empty for an in-memory session, set
379/// once by opening or attaching one. Installed on every session's config by
380/// `new_session`, so `ATTACH` can fill it later.
381#[derive(Debug, Default)]
382pub struct DatabaseSlot {
383    database: OnceLock<Arc<Database>>,
384}
385
386impl DatabaseSlot {
387    pub fn get(&self) -> Option<Arc<Database>> {
388        self.database.get().cloned()
389    }
390
391    /// Put `database` in the slot; fails if the session already has one.
392    pub fn set(&self, database: Arc<Database>) -> Result<()> {
393        self.database
394            .set(database)
395            .map_err(|_| DataFusionError::Plan("this session already has a database open".into()))
396    }
397}
398
399/// Install an empty database slot on `config` (done by `new_session`).
400pub(crate) fn with_database_slot(config: SessionConfig) -> SessionConfig {
401    config.with_extension(Arc::new(DatabaseSlot::default()))
402}
403
404/// The session's database slot.
405pub fn slot(state: &SessionState) -> Arc<DatabaseSlot> {
406    slot_of_config(state.config())
407}
408
409fn slot_of_config(config: &SessionConfig) -> Arc<DatabaseSlot> {
410    config
411        .get_extension::<DatabaseSlot>()
412        .expect("an Ursa session has a database slot (built by ursa_engine::new_session)")
413}
414
415/// The session's database, if it has one.
416pub fn database(state: &SessionState) -> Option<Arc<Database>> {
417    database_of_config(state.config())
418}
419
420/// The database of the session `config` belongs to, if it has one. A task
421/// context's `session_config()` reaches it too.
422pub fn database_of_config(config: &SessionConfig) -> Option<Arc<Database>> {
423    config.get_extension::<DatabaseSlot>()?.get()
424}
425
426/// Wait for the session's stored-index writes, if it has a database. Call it
427/// before a process exits, so an index built by its last query is stored.
428pub fn flush(state: &SessionState) {
429    if let Some(db) = database(state) {
430        db.flush();
431    }
432}
433
434#[cfg(test)]
435mod tests;
ursa_engine/database/mod.rs

ursa_engine/database/
mod.rs
