uqa-engine 0.2.3

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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

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

//! Schema and column-statistics restoration.

use super::{
    BTreeMap, CatalogFacade, ColumnStatsRow, Engine, StorageBackendError, StorageBackendResult,
    Value,
};
use crate::engine_statistics::value_size;

impl Engine {
    pub(super) fn restore_schemas_from_catalog(
        &self,
        catalog: &dyn CatalogFacade,
    ) -> StorageBackendResult<()> {
        let schemas = catalog.load_schema_rows()?;
        for schema in &schemas {
            Self::validate_schema_name(&schema.name)?;
        }
        if !schemas.iter().any(|schema| schema.name == "public") {
            return Err(StorageBackendError::Other(
                "catalog is missing required schema `public`".to_string(),
            ));
        }
        *self.durable.schemas.write() = schemas
            .into_iter()
            .map(crate::engine_state::SchemaSecurity::from_row)
            .collect();
        Ok(())
    }

    pub(crate) fn load_column_stats_from_catalog(
        catalog: &dyn CatalogFacade,
        table_name: &str,
    ) -> StorageBackendResult<BTreeMap<String, uqa_planner::ColumnStats>> {
        let mut out = BTreeMap::new();
        for row in catalog.load_column_stats(table_name)? {
            out.insert(row.column_name.clone(), Self::column_stats_from_row(row)?);
        }
        Ok(out)
    }

    fn column_stats_from_row(
        row: ColumnStatsRow,
    ) -> StorageBackendResult<uqa_planner::ColumnStats> {
        // Old catalogs may contain entire documents in statistical values.
        // Bound decoding too, so opening one does not reconstruct those large
        // strings before the automatic worker replaces the legacy snapshot.
        let histogram =
            Self::decode_column_stat_list(&row.histogram_json, value_size::HISTOGRAM_VALUES)?;
        let (mcv_values, mcv_frequencies) = if row.mcv_values_json.len()
            > value_size::encoded_list_bytes(value_size::MCV_VALUES)
            || row.mcv_frequencies_json.len() > value_size::MCV_VALUES * 64
        {
            (Vec::new(), Vec::new())
        } else {
            let values: Vec<Value> = serde_json::from_str(&row.mcv_values_json)?;
            let frequencies: Vec<f64> = serde_json::from_str(&row.mcv_frequencies_json)?;
            if values.len() > value_size::MCV_VALUES {
                (Vec::new(), Vec::new())
            } else if values.len() != frequencies.len() {
                return Err(StorageBackendError::Other(format!(
                    "mismatched MCV values and frequencies for column `{}`",
                    row.column_name
                )));
            } else {
                values
                    .into_iter()
                    .zip(frequencies)
                    .filter(|(value, _)| value_size::accepts(value))
                    .unzip()
            }
        };
        Ok(uqa_planner::ColumnStats {
            distinct_count: row.distinct_count.try_into().map_err(|_| {
                StorageBackendError::Other(format!(
                    "negative distinct_count for column `{}`",
                    row.column_name
                ))
            })?,
            null_count: row.null_count.try_into().map_err(|_| {
                StorageBackendError::Other(format!(
                    "negative null_count for column `{}`",
                    row.column_name
                ))
            })?,
            min_value: Self::decode_column_stat_value(row.min_value)?,
            max_value: Self::decode_column_stat_value(row.max_value)?,
            row_count: row.row_count.try_into().map_err(|_| {
                StorageBackendError::Other(format!(
                    "negative row_count for column `{}`",
                    row.column_name
                ))
            })?,
            histogram,
            mcv_values,
            mcv_frequencies,
        })
    }

    fn decode_column_stat_list(raw: &str, limit: usize) -> StorageBackendResult<Vec<Value>> {
        if raw.len() > value_size::encoded_list_bytes(limit) {
            return Ok(Vec::new());
        }
        let values: Vec<Value> = serde_json::from_str(raw)?;
        if values.len() > limit {
            return Ok(Vec::new());
        }
        Ok(values.into_iter().filter(value_size::accepts).collect())
    }

    fn decode_column_stat_value(raw: Option<String>) -> StorageBackendResult<Option<Value>> {
        let Some(raw) = raw else {
            return Ok(None);
        };
        if raw.len() > value_size::ENCODED_VALUE_BYTES {
            return Ok(None);
        }
        match serde_json::from_str::<Value>(&raw)? {
            Value::Null => Ok(None),
            value => Ok(value_size::accepts(&value).then_some(value)),
        }
    }
}