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/// What a session may keep to itself of a small table, whose analysis threshold is fifty changes.
14const MIN_KEPT_CHANGES: u64 = 16;
15
16/// The changes that make an analysis due: the analyze threshold of 50 rows and the scale factor of a tenth of the analyzed rows that `PostgreSQL` uses.
17const fn change_threshold(analyzed_rows: u64) -> u64 {
18    50 + analyzed_rows / 10
19}
20
21#[cfg(test)]
22mod tests;
23
24/// Transactional maintenance metadata. Existing column statistics remain
25/// available while this record tracks changes awaiting a replacement.
26#[derive(Clone, Default, Debug, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(default)]
28pub struct StatisticsMaintenance {
29    object_id: Option<[u8; 16]>,
30    generation: u64,
31    changes: u64,
32    dirty_since_ms: u64,
33    analyzed_rows: Option<u64>,
34    statistics_format: u32,
35    /// The commit sequence the last analysis sampled at, where its provider numbers commits: every commit up to it is in the statistics. A record without it says nothing about what its analysis saw.
36    #[serde(skip_serializing_if = "Option::is_none")]
37    analyzed_at: Option<u64>,
38    /// Identity of the last analysis, independent of later counter increments.
39    /// Legacy records acquire it on their next completed analysis.
40    #[serde(skip_serializing_if = "Option::is_none")]
41    analysis_generation: Option<u64>,
42}
43
44impl StatisticsMaintenance {
45    pub const KEY_PREFIX: &'static str = "uqa.statistics.maintenance.v1:";
46
47    pub fn key(table: &str) -> String {
48        format!("{}{table}", Self::KEY_PREFIX)
49    }
50
51    pub fn load(catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<Self> {
52        catalog
53            .get_metadata(&Self::key(table))?
54            .map(|json| serde_json::from_str(&json).map_err(Into::into))
55            .transpose()
56            .map(Option::unwrap_or_default)
57    }
58
59    pub fn load_for(
60        catalog: &dyn CatalogFacade,
61        table: &str,
62        object_id: [u8; 16],
63    ) -> StorageBackendResult<Self> {
64        let state = Self::load(catalog, table)?;
65        Ok(
66            if state.object_id.is_some_and(|stored| stored != object_id) {
67                Self::default()
68            } else {
69                state
70            },
71        )
72    }
73
74    pub fn save(&self, catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<()> {
75        catalog.save_statistics_maintenance(table, self)
76    }
77
78    pub fn dirty(&self) -> bool {
79        self.changes != 0
80    }
81
82    pub fn missing(&self, statistics_empty: bool) -> bool {
83        self.analyzed_rows.is_none() && statistics_empty
84    }
85
86    pub fn invalidates_existing_statistics(&self) -> bool {
87        // Initial imported statistics predate this maintenance protocol.
88        // A new write adopts that baseline before marking it stale.
89        self.dirty() && self.analyzed_rows.is_some()
90    }
91
92    pub fn due(&self, missing: bool, now: u64, statistics_format: u32) -> bool {
93        self.statistics_format != statistics_format
94            || missing
95            || (self.dirty()
96                && (self.analyzed_rows == Some(0)
97                    || self.changes >= change_threshold(self.analyzed_rows.unwrap_or(0))
98                    || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS))
99    }
100
101    /// Whether a session may keep `unrecorded` changes of its commits to itself instead of recording them now. A record written by every commit is a write of two records to each of them, and the count decides only when an analysis becomes due, so a session records the changes that decide something: the first one after an analysis, which marks the statistics stale and starts their age, and the ones that make an analysis due. Otherwise it keeps up to a quarter of what makes one due, or sixteen changes of a small table, which bounds what sessions hide from one another and what a session loses when it ends. While an analysis is due already, more changes decide nothing.
102    pub fn defers(&self, unrecorded: u64, now: u64) -> bool {
103        if !self.dirty() {
104            return false;
105        }
106        let Some(analyzed_rows) = self.analyzed_rows else {
107            // Changes without analyzed rows are those of a table without statistics, whose analysis is due whatever it counts.
108            return true;
109        };
110        let threshold = change_threshold(analyzed_rows);
111        if analyzed_rows == 0
112            || self.changes >= threshold
113            || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS
114        {
115            return true;
116        }
117        self.changes.saturating_add(unrecorded) < threshold
118            && unrecorded < (threshold / 4).max(MIN_KEPT_CHANGES)
119    }
120
121    pub fn analyzed_for(
122        catalog: &dyn CatalogFacade,
123        table: &str,
124        object_id: [u8; 16],
125        rows: u64,
126        statistics_format: u32,
127        sampled_at: Option<u64>,
128    ) -> StorageBackendResult<()> {
129        let mut state = Self::load_for(catalog, table, object_id)?;
130        state.analyzed(object_id, rows, statistics_format, sampled_at)?;
131        state.save(catalog, table)
132    }
133
134    /// Record a completed analysis of `rows` rows, which covers every change counted before it. `sampled_at` is the commit sequence its sample read at, or `None` when that is not known.
135    pub fn analyzed(
136        &mut self,
137        object_id: [u8; 16],
138        rows: u64,
139        statistics_format: u32,
140        sampled_at: Option<u64>,
141    ) -> StorageBackendResult<()> {
142        self.object_id = Some(object_id);
143        self.advance_generation()?;
144        self.analysis_generation = Some(self.generation);
145        self.changes = 0;
146        self.dirty_since_ms = 0;
147        self.analyzed_rows = Some(rows);
148        self.statistics_format = statistics_format;
149        self.analyzed_at = sampled_at;
150        Ok(())
151    }
152
153    /// Whether the last analysis sampled every commit up to `through`, so that changes a session kept of commits up to there need no counting.
154    pub fn covers(&self, through: Option<u64>) -> bool {
155        matches!((self.analyzed_at, through), (Some(at), Some(through)) if through <= at)
156    }
157
158    /// Row changes retain the last statistics publication; another analysis
159    /// replaces it even when the resulting estimates happen to be identical.
160    pub fn same_analysis(&self, other: &Self) -> bool {
161        self.analysis_generation == other.analysis_generation
162            && self.analyzed_at == other.analyzed_at
163            && self.analyzed_rows == other.analyzed_rows
164            && self.statistics_format == other.statistics_format
165    }
166
167    /// Publish this sample's reset over changes recorded since `before`. The caller
168    /// must first verify that the relation and its last statistics publication
169    /// still have the captured identities. A changed data generation can also
170    /// represent writes retained privately by their sessions: keep one pending
171    /// change in that case so their absence from the counter cannot mark the
172    /// sampled statistics current. The counter is a maintenance estimate.
173    pub fn merge_sample(
174        &self,
175        before: &Self,
176        current: &Self,
177        data_changed: bool,
178        now: u64,
179    ) -> StorageBackendResult<Option<Self>> {
180        let Some(mut merged) = Self::merge(before, self, current)
181            .map_err(crate::mvcc::VersionError::into_storage_error)?
182        else {
183            return Ok(None);
184        };
185        if data_changed && !merged.dirty() {
186            merged.changes = 1;
187            merged.dirty_since_ms = now;
188        }
189        Ok(Some(merged))
190    }
191
192    fn advance_generation(&mut self) -> StorageBackendResult<()> {
193        self.generation = self.generation.checked_add(1).ok_or_else(|| {
194            crate::StorageBackendError::Other("statistics generation space exhausted".into())
195        })?;
196        Ok(())
197    }
198
199    pub fn record_changes(
200        &mut self,
201        object_id: [u8; 16],
202        count: u64,
203        previous_rows: Option<u64>,
204        now: u64,
205    ) -> StorageBackendResult<()> {
206        self.object_id = Some(object_id);
207        self.advance_generation()?;
208        if self.analyzed_rows.is_none() {
209            self.analyzed_rows = previous_rows;
210        }
211        if !self.dirty() {
212            self.dirty_since_ms = now;
213        }
214        self.changes = self.changes.checked_add(count).ok_or_else(|| {
215            crate::StorageBackendError::Other("statistics change count overflow".into())
216        })?;
217        Ok(())
218    }
219
220    /// Encode the fixed-size state under the same budget and cancellation control as its record.
221    pub fn encode(
222        &self,
223        control: &crate::read_control::StorageReadControl,
224    ) -> crate::mvcc::VersionResult<uqa_core::memory::BudgetedVec<u8>> {
225        crate::key_value::record_json::encode(self, control)
226    }
227
228    /// Merge evaluated counter changes. A replacement object or overlapping reset that cannot retain the original baseline remains a conditional canonical replacement.
229    pub(crate) fn merge(
230        before: &Self,
231        after: &Self,
232        current: &Self,
233    ) -> crate::mvcc::VersionResult<Option<Self>> {
234        use crate::mvcc::VersionError;
235        if before
236            .object_id
237            .is_some_and(|id| Some(id) != after.object_id)
238            || current
239                .object_id
240                .is_some_and(|id| Some(id) != after.object_id)
241            || after.generation < before.generation
242        {
243            return Ok(None);
244        }
245        let changes =
246            i128::from(current.changes) + i128::from(after.changes) - i128::from(before.changes);
247        if changes < 0 {
248            return Ok(None);
249        }
250        let changes = u64::try_from(changes)
251            .map_err(|_| VersionError::InvalidEncoding("statistics change count overflow"))?;
252        let generation = current
253            .generation
254            .checked_add(after.generation - before.generation)
255            .ok_or(VersionError::InvalidEncoding(
256                "statistics generation space exhausted",
257            ))?;
258        let changed_estimate = after.statistics_format != 0
259            && (after.analyzed_rows != before.analyzed_rows
260                || after.statistics_format != before.statistics_format);
261        Ok(Some(Self {
262            object_id: after.object_id,
263            generation,
264            changes,
265            dirty_since_ms: if changes == 0 {
266                0
267            } else {
268                [after.dirty_since_ms, current.dirty_since_ms]
269                    .into_iter()
270                    .filter(|time| *time != 0)
271                    .min()
272                    .unwrap_or(0)
273            },
274            analyzed_rows: if changed_estimate {
275                after.analyzed_rows
276            } else {
277                current.analyzed_rows.or(after.analyzed_rows)
278            },
279            statistics_format: if changed_estimate {
280                after.statistics_format
281            } else {
282                current.statistics_format
283            },
284            // An analysis of the merged transaction is the one whose statistics it publishes.
285            analyzed_at: if after.analyzed_at == before.analyzed_at {
286                current.analyzed_at
287            } else {
288                after.analyzed_at
289            },
290            analysis_generation: if after.analysis_generation == before.analysis_generation {
291                current.analysis_generation
292            } else {
293                after.analysis_generation
294            },
295        }))
296    }
297}
uqa_storage/statistics_maintenance.rs

uqa_storage/
statistics_maintenance.rs
