hayson_engine/catalog/
declare_relation.rs

1use std::collections::HashMap;
2
3use serde_json::Value;
4
5use crate::errors::HaysonSchemaError;
6use crate::table::Table;
7
8use super::integrity::check_integrity;
9use super::parse::{parse_catalog, parse_relation};
10use super::serialize::catalog_document;
11use super::types::{Catalog, Relation};
12
13const LISTED_KEYS: usize = 20;
14
15fn describe_orphans(relation: &Relation, report: &super::integrity::IntegrityReport) -> Option<String> {
16    let mut keys: Vec<String> = report
17        .orphans
18        .iter()
19        .filter(|orphan| orphan.relation == relation.name)
20        .map(|orphan| orphan.primary_key.to_string())
21        .collect();
22    if keys.is_empty() {
23        return None;
24    }
25    keys.sort();
26    let listed = keys.iter().take(LISTED_KEYS).cloned().collect::<Vec<_>>().join(", ");
27    let rest = if keys.len() > LISTED_KEYS {
28        format!(" and {} more", keys.len() - LISTED_KEYS)
29    } else {
30        String::new()
31    };
32    Some(format!(
33        "refusing to declare relation \"{}\" with enforce: true: {} record(s) of \"{}\" already point at a value \"{}\" has no counterpart for: {listed}{rest} — nothing on disk was changed; pass force: true to declare it anyway (the orphans stay orphans, enforcement just starts from here on), or fix the data first",
34        relation.name,
35        keys.len(),
36        relation.from.table,
37        relation.to.table,
38    ))
39}
40
41/// `db.declareRelation()` — adds a named relation to the catalog. Mirrors `declare_index`'s shape:
42/// parses and validates against the catalog held in memory, refuses without writing anything, and
43/// leaves `write_catalog` to the caller.
44///
45/// When the new relation declares `enforce: true`, this additionally runs the same orphan check
46/// `checkIntegrity` does (`catalog::integrity::check_integrity`, scoped to just this relation) and
47/// refuses — unless `force` is `true` — if any live record already breaks it: turning enforcement
48/// on does not retroactively catch what was written before it existed, so declaring it silently
49/// over existing orphans would make a promise the data has already broken. `force: true` accepts
50/// that and declares anyway; the orphans already there stay orphans, enforcement starts from here.
51pub fn declare_relation(
52    catalog: &Catalog,
53    spec: &Value,
54    from_table_handle: &Table,
55    to_table_handle: &Table,
56    now: u64,
57    force: bool,
58) -> Result<Catalog, HaysonSchemaError> {
59    let declared = catalog.relations.clone().unwrap_or_default();
60    let path = format!("relations[{}]", declared.len());
61    let relation = parse_relation(spec, &path, &catalog.tables)?;
62
63    if declared.iter().any(|other| other.name == relation.name) {
64        return Err(HaysonSchemaError(format!(
65            "catalog already declares a relation named \"{}\" — a relation name is what the catalog identifies it by, so take the declared one off the catalog before declaring another under that name",
66            relation.name
67        )));
68    }
69
70    let mut next_declared = declared;
71    next_declared.push(relation.clone());
72    let mut next_catalog = catalog.clone();
73    next_catalog.updated_at = now;
74    next_catalog.relations = Some(next_declared);
75
76    parse_catalog(&catalog_document(&next_catalog)).map_err(|error| {
77        HaysonSchemaError(format!(
78            "declaring relation \"{}\" would leave a catalog this build of hayson cannot read back — {}; nothing was changed",
79            relation.name, error.0
80        ))
81    })?;
82
83    if relation.enforce == Some(true) && !force {
84        let mut tables: HashMap<String, &Table> = HashMap::new();
85        tables.insert(relation.from.table.clone(), from_table_handle);
86        tables.insert(relation.to.table.clone(), to_table_handle);
87        let report = check_integrity(&next_catalog, &tables).map_err(|error| {
88            HaysonSchemaError(format!(
89                "reading \"{}\"/\"{}\" to check relation \"{}\": {error}",
90                relation.from.table, relation.to.table, relation.name
91            ))
92        })?;
93        if let Some(message) = describe_orphans(&relation, &report) {
94            return Err(HaysonSchemaError(message));
95        }
96    }
97
98    Ok(next_catalog)
99}
100
101#[cfg(test)]
102mod tests {
103    use serde_json::json;
104    use tempfile::tempdir;
105
106    use super::*;
107    use crate::catalog::build::{create_table, empty_catalog};
108    use crate::catalog::types::CatalogLayout;
109    use crate::schema::parse_schema;
110
111    fn users_orders_catalog() -> Catalog {
112        let catalog = empty_catalog(CatalogLayout::Flat, 1_000);
113        let users = parse_schema(&json!({
114            "schemaVersion": 1, "table": "users", "primaryKey": "id",
115            "fields": { "id": { "type": "string" } },
116        }))
117        .unwrap();
118        let catalog = create_table(&catalog, "users", &users, 1_000);
119        let orders = parse_schema(&json!({
120            "schemaVersion": 1, "table": "orders", "primaryKey": "id",
121            "fields": { "id": { "type": "string" }, "userId": { "type": "string" } },
122        }))
123        .unwrap();
124        create_table(&catalog, "orders", &orders, 1_000)
125    }
126
127    fn open_table(dir: &tempfile::TempDir, name: &str, fields: Value) -> Table {
128        let mut document =
129            json!({ "schemaVersion": 1, "table": name, "primaryKey": "id", "fields": { "id": { "type": "string" } } });
130        for (key, value) in fields.as_object().unwrap() {
131            document["fields"][key] = value.clone();
132        }
133        let schema = parse_schema(&document).unwrap();
134        Table::open(schema, dir.path().join(format!("{name}.jsonl"))).unwrap()
135    }
136
137    fn relation_spec(enforce: bool) -> Value {
138        json!({
139            "name": "orders_user",
140            "from": { "table": "orders", "field": "userId" },
141            "to": { "table": "users", "field": "id" },
142            "cardinality": "many-to-one",
143            "enforce": enforce,
144        })
145    }
146
147    #[test]
148    fn declares_a_relation_with_no_orphans_to_check() {
149        let dir = tempdir().unwrap();
150        let users = open_table(&dir, "users", json!({}));
151        let orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
152        let catalog = users_orders_catalog();
153
154        let next = declare_relation(&catalog, &relation_spec(false), &orders, &users, 2_000, false).unwrap();
155
156        let relations = next.relations.unwrap();
157        assert_eq!(relations.len(), 1);
158        assert_eq!(relations[0].name, "orders_user");
159        assert_eq!(next.updated_at, 2_000);
160    }
161
162    #[test]
163    fn refuses_enforce_true_when_a_live_record_already_orphans_it() {
164        let dir = tempdir().unwrap();
165        let users = open_table(&dir, "users", json!({}));
166        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
167        orders.put(&json!({ "id": "o1", "userId": "ghost" }), false).unwrap();
168        let catalog = users_orders_catalog();
169
170        let error = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, false).unwrap_err();
171
172        assert!(error.0.contains("orders_user"));
173        assert!(error.0.contains("o1"));
174        assert!(error.0.contains("force"));
175        assert!(catalog.relations.is_none());
176    }
177
178    #[test]
179    fn force_declares_enforce_true_over_existing_orphans() {
180        let dir = tempdir().unwrap();
181        let users = open_table(&dir, "users", json!({}));
182        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
183        orders.put(&json!({ "id": "o1", "userId": "ghost" }), false).unwrap();
184        let catalog = users_orders_catalog();
185
186        let next = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, true).unwrap();
187
188        assert_eq!(next.relations.unwrap()[0].enforce, Some(true));
189    }
190
191    #[test]
192    fn enforce_true_declares_cleanly_when_every_reference_resolves() {
193        let dir = tempdir().unwrap();
194        let mut users = open_table(&dir, "users", json!({}));
195        let mut orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
196        users.put(&json!({ "id": "u1" }), false).unwrap();
197        orders.put(&json!({ "id": "o1", "userId": "u1" }), false).unwrap();
198        let catalog = users_orders_catalog();
199
200        let next = declare_relation(&catalog, &relation_spec(true), &orders, &users, 2_000, false).unwrap();
201
202        assert_eq!(next.relations.unwrap()[0].enforce, Some(true));
203    }
204
205    #[test]
206    fn refuses_a_duplicate_relation_name() {
207        let dir = tempdir().unwrap();
208        let users = open_table(&dir, "users", json!({}));
209        let orders = open_table(&dir, "orders", json!({ "userId": { "type": "string" } }));
210        let catalog = users_orders_catalog();
211        let catalog = declare_relation(&catalog, &relation_spec(false), &orders, &users, 2_000, false).unwrap();
212
213        let error = declare_relation(&catalog, &relation_spec(false), &orders, &users, 3_000, false).unwrap_err();
214        assert!(error.0.contains("already declares a relation named \"orders_user\""));
215    }
216}
hayson_engine/catalog/declare_relation.rs

hayson_engine/catalog/
declare_relation.rs
