uqa_storage/
statistics_maintenance.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Durable statistics-maintenance state and evaluated concurrent counter changes.
8
9use crate::{CatalogFacade, StorageBackendResult};
10use serde::{Deserialize, Serialize};
11
12const MAX_DIRTY_AGE_MS: u64 = 60_000;
13
14#[cfg(test)]
15mod tests;
16
17/// Transactional maintenance metadata. Existing column statistics remain
18/// available while this record tracks changes awaiting a replacement.
19#[derive(Clone, Default, Debug, PartialEq, Eq, Serialize, Deserialize)]
20#[serde(default)]
21pub struct StatisticsMaintenance {
22    object_id: Option<[u8; 16]>,
23    generation: u64,
24    changes: u64,
25    dirty_since_ms: u64,
26    analyzed_rows: Option<u64>,
27    statistics_format: u32,
28}
29
30impl StatisticsMaintenance {
31    pub const KEY_PREFIX: &'static str = "uqa.statistics.maintenance.v1:";
32
33    pub fn key(table: &str) -> String {
34        format!("{}{table}", Self::KEY_PREFIX)
35    }
36
37    pub fn load(catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<Self> {
38        catalog
39            .get_metadata(&Self::key(table))?
40            .map(|json| serde_json::from_str(&json).map_err(Into::into))
41            .transpose()
42            .map(Option::unwrap_or_default)
43    }
44
45    pub fn load_for(
46        catalog: &dyn CatalogFacade,
47        table: &str,
48        object_id: [u8; 16],
49    ) -> StorageBackendResult<Self> {
50        let state = Self::load(catalog, table)?;
51        Ok(
52            if state.object_id.is_some_and(|stored| stored != object_id) {
53                Self::default()
54            } else {
55                state
56            },
57        )
58    }
59
60    pub fn save(&self, catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<()> {
61        catalog.save_statistics_maintenance(table, self)
62    }
63
64    pub fn dirty(&self) -> bool {
65        self.changes != 0
66    }
67
68    pub fn missing(&self, statistics_empty: bool) -> bool {
69        self.analyzed_rows.is_none() && statistics_empty
70    }
71
72    pub fn invalidates_existing_statistics(&self) -> bool {
73        // Initial imported statistics predate this maintenance protocol.
74        // A new write adopts that baseline before marking it stale.
75        self.dirty() && self.analyzed_rows.is_some()
76    }
77
78    pub fn due(&self, missing: bool, now: u64, statistics_format: u32) -> bool {
79        self.statistics_format != statistics_format
80            || missing
81            || (self.dirty()
82                && (self.analyzed_rows == Some(0)
83                    || self.changes >= 50 + self.analyzed_rows.unwrap_or(0) / 10
84                    || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS))
85    }
86
87    pub fn analyzed_for(
88        catalog: &dyn CatalogFacade,
89        table: &str,
90        object_id: [u8; 16],
91        rows: u64,
92        statistics_format: u32,
93    ) -> StorageBackendResult<()> {
94        let mut state = Self::load_for(catalog, table, object_id)?;
95        state.object_id = Some(object_id);
96        state.advance_generation()?;
97        state.changes = 0;
98        state.dirty_since_ms = 0;
99        state.analyzed_rows = Some(rows);
100        state.statistics_format = statistics_format;
101        state.save(catalog, table)
102    }
103
104    fn advance_generation(&mut self) -> StorageBackendResult<()> {
105        self.generation = self.generation.checked_add(1).ok_or_else(|| {
106            crate::StorageBackendError::Other("statistics generation space exhausted".into())
107        })?;
108        Ok(())
109    }
110
111    pub fn record_changes(
112        &mut self,
113        object_id: [u8; 16],
114        count: u64,
115        previous_rows: Option<u64>,
116        now: u64,
117    ) -> StorageBackendResult<()> {
118        self.object_id = Some(object_id);
119        self.advance_generation()?;
120        if self.analyzed_rows.is_none() {
121            self.analyzed_rows = previous_rows;
122        }
123        if !self.dirty() {
124            self.dirty_since_ms = now;
125        }
126        self.changes = self.changes.checked_add(count).ok_or_else(|| {
127            crate::StorageBackendError::Other("statistics change count overflow".into())
128        })?;
129        Ok(())
130    }
131
132    /// Encode the fixed-size state under the same budget and cancellation control as its record.
133    pub fn encode(
134        &self,
135        control: &crate::read_control::StorageReadControl,
136    ) -> crate::mvcc::VersionResult<uqa_core::memory::BudgetedVec<u8>> {
137        crate::key_value::record_json::encode(self, control)
138    }
139
140    /// Merge evaluated counter changes. A replacement object or overlapping reset that cannot retain the original baseline remains a conditional canonical replacement.
141    pub(crate) fn merge(
142        before: &Self,
143        after: &Self,
144        current: &Self,
145    ) -> crate::mvcc::VersionResult<Option<Self>> {
146        use crate::mvcc::VersionError;
147        if before
148            .object_id
149            .is_some_and(|id| Some(id) != after.object_id)
150            || current
151                .object_id
152                .is_some_and(|id| Some(id) != after.object_id)
153            || after.generation < before.generation
154        {
155            return Ok(None);
156        }
157        let changes =
158            i128::from(current.changes) + i128::from(after.changes) - i128::from(before.changes);
159        if changes < 0 {
160            return Ok(None);
161        }
162        let changes = u64::try_from(changes)
163            .map_err(|_| VersionError::InvalidEncoding("statistics change count overflow"))?;
164        let generation = current
165            .generation
166            .checked_add(after.generation - before.generation)
167            .ok_or(VersionError::InvalidEncoding(
168                "statistics generation space exhausted",
169            ))?;
170        let changed_estimate = after.statistics_format != 0
171            && (after.analyzed_rows != before.analyzed_rows
172                || after.statistics_format != before.statistics_format);
173        Ok(Some(Self {
174            object_id: after.object_id,
175            generation,
176            changes,
177            dirty_since_ms: if changes == 0 {
178                0
179            } else {
180                [after.dirty_since_ms, current.dirty_since_ms]
181                    .into_iter()
182                    .filter(|time| *time != 0)
183                    .min()
184                    .unwrap_or(0)
185            },
186            analyzed_rows: if changed_estimate {
187                after.analyzed_rows
188            } else {
189                current.analyzed_rows.or(after.analyzed_rows)
190            },
191            statistics_format: if changed_estimate {
192                after.statistics_format
193            } else {
194                current.statistics_format
195            },
196        }))
197    }
198}
uqa_storage/statistics_maintenance.rs

uqa_storage/
statistics_maintenance.rs
