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//!   "settings": {
12//!     "ursa.index_storage_limit": "32GiB"
13//!   }
14//! }
15//! ```
16//!
17//! `settings` holds the settings the database keeps (#327): today only
18//! `ursa.index_storage_limit`, as `SET` left it. It is written only when there
19//! is one, so a database that never had one is the same file as before; a
20//! version of Ursa older than 0.8 ignores it (and drops it if it rewrites the
21//! file, which then obeys the default limit again).
22//!
23//! `version` changes only when the file's shape does. A file with a newer
24//! version is refused; anything that is not this shape is reported as not a
25//! valid database. An empty file (`touch club.ursa`) is an empty catalog.
26
27use std::collections::BTreeMap;
28use std::fs::File;
29use std::io::Write;
30use std::path::Path;
31
32use datafusion::error::{DataFusionError, Result};
33use serde_json::Value;
34
35/// The `format` field of every catalog file.
36pub const FORMAT: &str = "ursa";
37/// The catalog file version this build reads and writes.
38pub const VERSION: u64 = 1;
39
40/// The contents of a catalog file.
41#[derive(Debug, Clone, Default, PartialEq, Eq)]
42pub struct CatalogFile {
43    pub statements: Vec<String>,
44    /// Settings the database keeps, by their full name
45    /// (`ursa.index_storage_limit`), as `SET` would take them.
46    pub settings: BTreeMap<String, String>,
47}
48
49impl CatalogFile {
50    /// Parse a catalog file's text. `name` is how messages name the database.
51    pub fn parse(text: &str, name: &str) -> Result<Self> {
52        if text.trim().is_empty() {
53            return Ok(Self::default());
54        }
55        let invalid = |why: String| {
56            DataFusionError::Execution(format!("{name} is not a valid Ursa database: {why}"))
57        };
58        let value: Value =
59            serde_json::from_str(text).map_err(|e| invalid(format!("it is not JSON ({e})")))?;
60        let Value::Object(fields) = value else {
61            return Err(invalid("it is not a JSON object".into()));
62        };
63        if fields.get("format").and_then(Value::as_str) != Some(FORMAT) {
64            return Err(invalid(format!("its \"format\" is not \"{FORMAT}\"")));
65        }
66        let version = fields
67            .get("version")
68            .and_then(Value::as_u64)
69            .filter(|v| *v >= 1)
70            .ok_or_else(|| invalid("its \"version\" is not a positive integer".into()))?;
71        if version > VERSION {
72            return Err(DataFusionError::Execution(format!(
73                "{name} was written by a newer version of Ursa (database format version \
74                 {version}; this version of ursa reads version {VERSION}); upgrade ursa to \
75                 open it"
76            )));
77        }
78        let statements = fields
79            .get("statements")
80            .and_then(Value::as_array)
81            .ok_or_else(|| invalid("it has no \"statements\" list".into()))?
82            .iter()
83            .map(|s| s.as_str().map(str::to_string))
84            .collect::<Option<Vec<_>>>()
85            .ok_or_else(|| invalid("an entry of \"statements\" is not a string".into()))?;
86        let settings = match fields.get("settings") {
87            None | Some(Value::Null) => BTreeMap::new(),
88            Some(Value::Object(settings)) => settings
89                .iter()
90                .map(|(k, v)| v.as_str().map(|v| (k.clone(), v.to_string())))
91                .collect::<Option<BTreeMap<_, _>>>()
92                .ok_or_else(|| invalid("a value of \"settings\" is not a string".into()))?,
93            Some(_) => return Err(invalid("its \"settings\" is not a JSON object".into())),
94        };
95        Ok(Self {
96            statements,
97            settings,
98        })
99    }
100
101    /// Read and parse the catalog file at `path`.
102    pub fn read(path: &Path, name: &str) -> Result<Self> {
103        let text = std::fs::read_to_string(path)
104            .map_err(|e| DataFusionError::Execution(format!("cannot read {name}: {e}")))?;
105        Self::parse(&text, name)
106    }
107
108    /// The file's text: `format`, `version`, then one statement per line,
109    /// then the settings if there are any, one per line, with a trailing
110    /// newline.
111    pub fn to_text(&self) -> String {
112        let quote = |s: &str| Value::String(s.to_string()).to_string();
113        let mut text = format!(
114            "{{\n  \"format\": {},\n  \"version\": {VERSION},\n  \"statements\": [",
115            quote(FORMAT)
116        );
117        for (i, statement) in self.statements.iter().enumerate() {
118            text.push_str(if i == 0 { "\n    " } else { ",\n    " });
119            text.push_str(&quote(statement));
120        }
121        if !self.statements.is_empty() {
122            text.push_str("\n  ");
123        }
124        text.push(']');
125        if !self.settings.is_empty() {
126            text.push_str(",\n  \"settings\": {");
127            for (i, (name, value)) in self.settings.iter().enumerate() {
128                text.push_str(if i == 0 { "\n    " } else { ",\n    " });
129                text.push_str(&format!("{}: {}", quote(name), quote(value)));
130            }
131            text.push_str("\n  }");
132        }
133        text.push_str("\n}\n");
134        text
135    }
136
137    /// Write the file at `path` atomically: write a temporary file in
138    /// `scratch_dir` (on the same filesystem), `fsync` it, rename it over
139    /// `path`, then `fsync` the directory.
140    pub fn write(&self, path: &Path, scratch_dir: &Path, name: &str) -> Result<()> {
141        let failed =
142            |e: std::io::Error| DataFusionError::Execution(format!("cannot save {name}: {e}"));
143        let file_name = path
144            .file_name()
145            .map(|n| n.to_string_lossy().into_owned())
146            .unwrap_or_else(|| "catalog".into());
147        let temp = scratch_dir.join(format!("{file_name}.tmp"));
148        {
149            let mut file = File::create(&temp).map_err(failed)?;
150            file.write_all(self.to_text().as_bytes()).map_err(failed)?;
151            file.sync_all().map_err(failed)?;
152        }
153        std::fs::rename(&temp, path).map_err(failed)?;
154        sync_dir(path.parent().unwrap_or(Path::new(".")));
155        Ok(())
156    }
157}
158
159/// `fsync` a directory so a rename in it is durable. Best effort: not every
160/// platform can open a directory as a file.
161fn sync_dir(dir: &Path) {
162    #[cfg(unix)]
163    if let Ok(dir) = File::open(dir) {
164        let _ = dir.sync_all();
165    }
166    #[cfg(not(unix))]
167    let _ = dir;
168}
ursa_engine/database/catalog_file.rs

ursa_engine/database/
catalog_file.rs
