uqa-engine 0.3.0

Engine: schema-aware table store, catalog restore, transactions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Durable domain identities and catalog publication.

use std::collections::BTreeMap;

use uqa_sql::ast::ColumnType;
use uqa_sql::SQLError;
use uqa_storage::CatalogFacade;

use crate::{Engine, StorageBackendResult};

const DOMAINS_METADATA_KEY: &str = "sql_domains_json";

pub(crate) use uqa_sql::catalog::domain::StoredDomain;

impl Engine {
    pub(crate) fn domain_default_expression(&self, ty: &ColumnType) -> Option<uqa_sql::ast::Expr> {
        uqa_sql::catalog::domain::domain_default_expression(self, ty)
    }

    pub(crate) fn try_column_insert_default_expr(
        &self,
        table: &str,
        column: &str,
    ) -> StorageBackendResult<Option<uqa_sql::ast::Expr>> {
        if let Some(default) = self.try_column_default_expr(table, column)? {
            return Ok(Some(default));
        }
        let columns = self.try_describe_table(table)?.unwrap_or_default();
        let Some(column) = columns.iter().find(|definition| definition.name == column) else {
            return Ok(None);
        };
        Ok(self.domain_default_expression(&column.ty).or_else(|| {
            matches!(column.ty, ColumnType::Domain { .. })
                .then_some(uqa_sql::ast::Expr::Literal(uqa_core::Value::Null))
        }))
    }
    pub(crate) fn domain_by_oid(&self, oid: u32) -> Option<StoredDomain> {
        self.durable
            .domains
            .read()
            .values()
            .find(|domain| domain.oid == oid)
            .cloned()
    }

    pub(crate) fn publish_domain(&self, domain: StoredDomain) -> Result<(), SQLError> {
        let mut registry = self.durable.domains.write();
        let mut next = registry.clone();
        next.insert(domain.identity.qualified_name(), domain);
        self.persist_domains(&next)?;
        *registry = next;
        drop(registry);
        self.note_catalog_registry_changed();
        Ok(())
    }

    pub(crate) fn persist_domains(
        &self,
        registry: &BTreeMap<String, StoredDomain>,
    ) -> Result<(), SQLError> {
        if let Some(catalog) = &self.storage.catalog {
            let json = serde_json::to_string(registry).map_err(|error| {
                SQLError::Internal(format!("serialize domain catalog: {error}"))
            })?;
            catalog
                .set_metadata(DOMAINS_METADATA_KEY, &json)
                .map_err(|error| SQLError::Internal(format!("persist domain catalog: {error}")))?;
        }
        Ok(())
    }

    pub(crate) fn restore_domains_from_catalog(
        &self,
        catalog: &dyn CatalogFacade,
    ) -> StorageBackendResult<()> {
        let registry = catalog
            .get_metadata(DOMAINS_METADATA_KEY)?
            .map(|json| serde_json::from_str(&json))
            .transpose()?
            .unwrap_or_default();
        *self.durable.domains.write() = registry;
        Ok(())
    }
}

impl uqa_sql::catalog::domain::DomainCatalog for Engine {
    fn domain_by_oid(&self, oid: u32) -> Option<StoredDomain> {
        Engine::domain_by_oid(self, oid)
    }
}