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    /// Earliest clock value that can make this unchanged state due. A clean
102    /// state has no deadline; missing, obsolete or already due statistics must
103    /// be revisited now. New committed changes still invalidate this decision.
104    pub fn next_due_at(&self, missing: bool, now: u64, statistics_format: u32) -> Option<u64> {
105        if self.due(missing, now, statistics_format) {
106            Some(now)
107        } else if self.dirty() {
108            self.dirty_since_ms.checked_add(MAX_DIRTY_AGE_MS)
109        } else {
110            None
111        }
112    }
113
114    /// 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.
115    pub fn defers(&self, unrecorded: u64, now: u64) -> bool {
116        if !self.dirty() {
117            return false;
118        }
119        let Some(analyzed_rows) = self.analyzed_rows else {
120            // Changes without analyzed rows are those of a table without statistics, whose analysis is due whatever it counts.
121            return true;
122        };
123        let threshold = change_threshold(analyzed_rows);
124        if analyzed_rows == 0
125            || self.changes >= threshold
126            || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS
127        {
128            return true;
129        }
130        self.changes.saturating_add(unrecorded) < threshold
131            && unrecorded < (threshold / 4).max(MIN_KEPT_CHANGES)
132    }
133
134    pub fn analyzed_for(
135        catalog: &dyn CatalogFacade,
136        table: &str,
137        object_id: [u8; 16],
138        rows: u64,
139        statistics_format: u32,
140        sampled_at: Option<u64>,
141    ) -> StorageBackendResult<()> {
142        let mut state = Self::load_for(catalog, table, object_id)?;
143        state.analyzed(object_id, rows, statistics_format, sampled_at)?;
144        state.save(catalog, table)
145    }
146
147    /// 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.
148    pub fn analyzed(
149        &mut self,
150        object_id: [u8; 16],
151        rows: u64,
152        statistics_format: u32,
153        sampled_at: Option<u64>,
154    ) -> StorageBackendResult<()> {
155        self.object_id = Some(object_id);
156        self.advance_generation()?;
157        self.analysis_generation = Some(self.generation);
158        self.changes = 0;
159        self.dirty_since_ms = 0;
160        self.analyzed_rows = Some(rows);
161        self.statistics_format = statistics_format;
162        self.analyzed_at = sampled_at;
163        Ok(())
164    }
165
166    /// Whether the last analysis sampled every commit up to `through`, so that changes a session kept of commits up to there need no counting.
167    pub fn covers(&self, through: Option<u64>) -> bool {
168        matches!((self.analyzed_at, through), (Some(at), Some(through)) if through <= at)
169    }
170
171    /// Row changes retain the last statistics publication; another analysis
172    /// replaces it even when the resulting estimates happen to be identical.
173    pub fn same_analysis(&self, other: &Self) -> bool {
174        self.analysis_generation == other.analysis_generation
175            && self.analyzed_at == other.analyzed_at
176            && self.analyzed_rows == other.analyzed_rows
177            && self.statistics_format == other.statistics_format
178    }
179
180    /// Publish this sample's reset over changes recorded since `before`. The caller
181    /// must first verify that the relation and its last statistics publication
182    /// still have the captured identities. A changed data generation can also
183    /// represent writes retained privately by their sessions: keep one pending
184    /// change in that case so their absence from the counter cannot mark the
185    /// sampled statistics current. The counter is a maintenance estimate.
186    pub fn merge_sample(
187        &self,
188        before: &Self,
189        current: &Self,
190        data_changed: bool,
191        now: u64,
192    ) -> StorageBackendResult<Option<Self>> {
193        let Some(mut merged) = Self::merge(before, self, current)
194            .map_err(crate::mvcc::VersionError::into_storage_error)?
195        else {
196            return Ok(None);
197        };
198        if data_changed && !merged.dirty() {
199            merged.changes = 1;
200            merged.dirty_since_ms = now;
201        }
202        Ok(Some(merged))
203    }
204
205    fn advance_generation(&mut self) -> StorageBackendResult<()> {
206        self.generation = self.generation.checked_add(1).ok_or_else(|| {
207            crate::StorageBackendError::Other("statistics generation space exhausted".into())
208        })?;
209        Ok(())
210    }
211
212    pub fn record_changes(
213        &mut self,
214        object_id: [u8; 16],
215        count: u64,
216        previous_rows: Option<u64>,
217        now: u64,
218    ) -> StorageBackendResult<()> {
219        self.object_id = Some(object_id);
220        self.advance_generation()?;
221        if self.analyzed_rows.is_none() {
222            self.analyzed_rows = previous_rows;
223        }
224        if !self.dirty() {
225            self.dirty_since_ms = now;
226        }
227        self.changes = self.changes.checked_add(count).ok_or_else(|| {
228            crate::StorageBackendError::Other("statistics change count overflow".into())
229        })?;
230        Ok(())
231    }
232
233    /// Encode the fixed-size state under the same budget and cancellation control as its record.
234    pub fn encode(
235        &self,
236        control: &crate::read_control::StorageReadControl,
237    ) -> crate::mvcc::VersionResult<uqa_core::memory::BudgetedVec<u8>> {
238        crate::key_value::record_json::encode(self, control)
239    }
240
241    /// Merge evaluated counter changes. A replacement object or overlapping reset that cannot retain the original baseline remains a conditional canonical replacement.
242    pub(crate) fn merge(
243        before: &Self,
244        after: &Self,
245        current: &Self,
246    ) -> crate::mvcc::VersionResult<Option<Self>> {
247        use crate::mvcc::VersionError;
248        if before
249            .object_id
250            .is_some_and(|id| Some(id) != after.object_id)
251            || current
252                .object_id
253                .is_some_and(|id| Some(id) != after.object_id)
254            || after.generation < before.generation
255        {
256            return Ok(None);
257        }
258        let changes =
259            i128::from(current.changes) + i128::from(after.changes) - i128::from(before.changes);
260        if changes < 0 {
261            return Ok(None);
262        }
263        let changes = u64::try_from(changes)
264            .map_err(|_| VersionError::InvalidEncoding("statistics change count overflow"))?;
265        let generation = current
266            .generation
267            .checked_add(after.generation - before.generation)
268            .ok_or(VersionError::InvalidEncoding(
269                "statistics generation space exhausted",
270            ))?;
271        let changed_estimate = after.statistics_format != 0
272            && (after.analyzed_rows != before.analyzed_rows
273                || after.statistics_format != before.statistics_format);
274        Ok(Some(Self {
275            object_id: after.object_id,
276            generation,
277            changes,
278            dirty_since_ms: if changes == 0 {
279                0
280            } else {
281                [after.dirty_since_ms, current.dirty_since_ms]
282                    .into_iter()
283                    .filter(|time| *time != 0)
284                    .min()
285                    .unwrap_or(0)
286            },
287            analyzed_rows: if changed_estimate {
288                after.analyzed_rows
289            } else {
290                current.analyzed_rows.or(after.analyzed_rows)
291            },
292            statistics_format: if changed_estimate {
293                after.statistics_format
294            } else {
295                current.statistics_format
296            },
297            // An analysis of the merged transaction is the one whose statistics it publishes.
298            analyzed_at: if after.analyzed_at == before.analyzed_at {
299                current.analyzed_at
300            } else {
301                after.analyzed_at
302            },
303            analysis_generation: if after.analysis_generation == before.analysis_generation {
304                current.analysis_generation
305            } else {
306                after.analysis_generation
307            },
308        }))
309    }
310}
uqa_storage/statistics_maintenance.rs

uqa_storage/
statistics_maintenance.rs
