ursa_engine/database/
catalog_file.rs

1//! The catalog file: a versioned JSON list of the statements that defined
2//! each object, in the order they are replayed.
3//!
4//! ```json
5//! {
6//!   "format": "ursa",
7//!   "version": 1,
8//!   "statements": [
9//!     "CREATE VIEW ties AS SELECT * FROM 'karate_edges.csv'"
10//!   ]
11//! }
12//! ```
13//!
14//! `version` changes only when the file's shape does. A file with a newer
15//! version is refused; anything that is not this shape is reported as not a
16//! valid database. An empty file (`touch club.ursa`) is an empty catalog.
17
18use std::fs::File;
19use std::io::Write;
20use std::path::Path;
21
22use datafusion::error::{DataFusionError, Result};
23use serde_json::Value;
24
25/// The `format` field of every catalog file.
26pub const FORMAT: &str = "ursa";
27/// The catalog file version this build reads and writes.
28pub const VERSION: u64 = 1;
29
30/// The contents of a catalog file.
31#[derive(Debug, Clone, Default, PartialEq, Eq)]
32pub struct CatalogFile {
33    pub statements: Vec<String>,
34}
35
36impl CatalogFile {
37    /// Parse a catalog file's text. `name` is how messages name the database.
38    pub fn parse(text: &str, name: &str) -> Result<Self> {
39        if text.trim().is_empty() {
40            return Ok(Self::default());
41        }
42        let invalid = |why: String| {
43            DataFusionError::Execution(format!("{name} is not a valid Ursa database: {why}"))
44        };
45        let value: Value =
46            serde_json::from_str(text).map_err(|e| invalid(format!("it is not JSON ({e})")))?;
47        let Value::Object(fields) = value else {
48            return Err(invalid("it is not a JSON object".into()));
49        };
50        if fields.get("format").and_then(Value::as_str) != Some(FORMAT) {
51            return Err(invalid(format!("its \"format\" is not \"{FORMAT}\"")));
52        }
53        let version = fields
54            .get("version")
55            .and_then(Value::as_u64)
56            .filter(|v| *v >= 1)
57            .ok_or_else(|| invalid("its \"version\" is not a positive integer".into()))?;
58        if version > VERSION {
59            return Err(DataFusionError::Execution(format!(
60                "{name} was written by a newer version of Ursa (database format version \
61                 {version}; this version of ursa reads version {VERSION}); upgrade ursa to \
62                 open it"
63            )));
64        }
65        let statements = fields
66            .get("statements")
67            .and_then(Value::as_array)
68            .ok_or_else(|| invalid("it has no \"statements\" list".into()))?
69            .iter()
70            .map(|s| s.as_str().map(str::to_string))
71            .collect::<Option<Vec<_>>>()
72            .ok_or_else(|| invalid("an entry of \"statements\" is not a string".into()))?;
73        Ok(Self { statements })
74    }
75
76    /// Read and parse the catalog file at `path`.
77    pub fn read(path: &Path, name: &str) -> Result<Self> {
78        let text = std::fs::read_to_string(path)
79            .map_err(|e| DataFusionError::Execution(format!("cannot read {name}: {e}")))?;
80        Self::parse(&text, name)
81    }
82
83    /// The file's text: `format`, `version`, then one statement per line,
84    /// with a trailing newline.
85    pub fn to_text(&self) -> String {
86        let quote = |s: &str| Value::String(s.to_string()).to_string();
87        let mut text = format!(
88            "{{\n  \"format\": {},\n  \"version\": {VERSION},\n  \"statements\": [",
89            quote(FORMAT)
90        );
91        for (i, statement) in self.statements.iter().enumerate() {
92            text.push_str(if i == 0 { "\n    " } else { ",\n    " });
93            text.push_str(&quote(statement));
94        }
95        if !self.statements.is_empty() {
96            text.push_str("\n  ");
97        }
98        text.push_str("]\n}\n");
99        text
100    }
101
102    /// Write the file at `path` atomically: write a temporary file in
103    /// `scratch_dir` (on the same filesystem), `fsync` it, rename it over
104    /// `path`, then `fsync` the directory.
105    pub fn write(&self, path: &Path, scratch_dir: &Path, name: &str) -> Result<()> {
106        let failed =
107            |e: std::io::Error| DataFusionError::Execution(format!("cannot save {name}: {e}"));
108        let file_name = path
109            .file_name()
110            .map(|n| n.to_string_lossy().into_owned())
111            .unwrap_or_else(|| "catalog".into());
112        let temp = scratch_dir.join(format!("{file_name}.tmp"));
113        {
114            let mut file = File::create(&temp).map_err(failed)?;
115            file.write_all(self.to_text().as_bytes()).map_err(failed)?;
116            file.sync_all().map_err(failed)?;
117        }
118        std::fs::rename(&temp, path).map_err(failed)?;
119        sync_dir(path.parent().unwrap_or(Path::new(".")));
120        Ok(())
121    }
122}
123
124/// `fsync` a directory so a rename in it is durable. Best effort: not every
125/// platform can open a directory as a file.
126fn sync_dir(dir: &Path) {
127    #[cfg(unix)]
128    if let Ok(dir) = File::open(dir) {
129        let _ = dir.sync_all();
130    }
131    #[cfg(not(unix))]
132    let _ = dir;
133}
ursa_engine/database/catalog_file.rs

ursa_engine/database/
catalog_file.rs
