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}
39
40impl StatisticsMaintenance {
41    pub const KEY_PREFIX: &'static str = "uqa.statistics.maintenance.v1:";
42
43    pub fn key(table: &str) -> String {
44        format!("{}{table}", Self::KEY_PREFIX)
45    }
46
47    pub fn load(catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<Self> {
48        catalog
49            .get_metadata(&Self::key(table))?
50            .map(|json| serde_json::from_str(&json).map_err(Into::into))
51            .transpose()
52            .map(Option::unwrap_or_default)
53    }
54
55    pub fn load_for(
56        catalog: &dyn CatalogFacade,
57        table: &str,
58        object_id: [u8; 16],
59    ) -> StorageBackendResult<Self> {
60        let state = Self::load(catalog, table)?;
61        Ok(
62            if state.object_id.is_some_and(|stored| stored != object_id) {
63                Self::default()
64            } else {
65                state
66            },
67        )
68    }
69
70    pub fn save(&self, catalog: &dyn CatalogFacade, table: &str) -> StorageBackendResult<()> {
71        catalog.save_statistics_maintenance(table, self)
72    }
73
74    pub fn dirty(&self) -> bool {
75        self.changes != 0
76    }
77
78    pub fn missing(&self, statistics_empty: bool) -> bool {
79        self.analyzed_rows.is_none() && statistics_empty
80    }
81
82    pub fn invalidates_existing_statistics(&self) -> bool {
83        // Initial imported statistics predate this maintenance protocol.
84        // A new write adopts that baseline before marking it stale.
85        self.dirty() && self.analyzed_rows.is_some()
86    }
87
88    pub fn due(&self, missing: bool, now: u64, statistics_format: u32) -> bool {
89        self.statistics_format != statistics_format
90            || missing
91            || (self.dirty()
92                && (self.analyzed_rows == Some(0)
93                    || self.changes >= change_threshold(self.analyzed_rows.unwrap_or(0))
94                    || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS))
95    }
96
97    /// 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.
98    pub fn defers(&self, unrecorded: u64, now: u64) -> bool {
99        if !self.dirty() {
100            return false;
101        }
102        let Some(analyzed_rows) = self.analyzed_rows else {
103            // Changes without analyzed rows are those of a table without statistics, whose analysis is due whatever it counts.
104            return true;
105        };
106        let threshold = change_threshold(analyzed_rows);
107        if analyzed_rows == 0
108            || self.changes >= threshold
109            || now.saturating_sub(self.dirty_since_ms) >= MAX_DIRTY_AGE_MS
110        {
111            return true;
112        }
113        self.changes.saturating_add(unrecorded) < threshold
114            && unrecorded < (threshold / 4).max(MIN_KEPT_CHANGES)
115    }
116
117    pub fn analyzed_for(
118        catalog: &dyn CatalogFacade,
119        table: &str,
120        object_id: [u8; 16],
121        rows: u64,
122        statistics_format: u32,
123        sampled_at: Option<u64>,
124    ) -> StorageBackendResult<()> {
125        let mut state = Self::load_for(catalog, table, object_id)?;
126        state.analyzed(object_id, rows, statistics_format, sampled_at)?;
127        state.save(catalog, table)
128    }
129
130    /// 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.
131    pub fn analyzed(
132        &mut self,
133        object_id: [u8; 16],
134        rows: u64,
135        statistics_format: u32,
136        sampled_at: Option<u64>,
137    ) -> StorageBackendResult<()> {
138        self.object_id = Some(object_id);
139        self.advance_generation()?;
140        self.changes = 0;
141        self.dirty_since_ms = 0;
142        self.analyzed_rows = Some(rows);
143        self.statistics_format = statistics_format;
144        self.analyzed_at = sampled_at;
145        Ok(())
146    }
147
148    /// Whether the last analysis sampled every commit up to `through`, so that changes a session kept of commits up to there need no counting.
149    pub fn covers(&self, through: Option<u64>) -> bool {
150        matches!((self.analyzed_at, through), (Some(at), Some(through)) if through <= at)
151    }
152
153    fn advance_generation(&mut self) -> StorageBackendResult<()> {
154        self.generation = self.generation.checked_add(1).ok_or_else(|| {
155            crate::StorageBackendError::Other("statistics generation space exhausted".into())
156        })?;
157        Ok(())
158    }
159
160    pub fn record_changes(
161        &mut self,
162        object_id: [u8; 16],
163        count: u64,
164        previous_rows: Option<u64>,
165        now: u64,
166    ) -> StorageBackendResult<()> {
167        self.object_id = Some(object_id);
168        self.advance_generation()?;
169        if self.analyzed_rows.is_none() {
170            self.analyzed_rows = previous_rows;
171        }
172        if !self.dirty() {
173            self.dirty_since_ms = now;
174        }
175        self.changes = self.changes.checked_add(count).ok_or_else(|| {
176            crate::StorageBackendError::Other("statistics change count overflow".into())
177        })?;
178        Ok(())
179    }
180
181    /// Encode the fixed-size state under the same budget and cancellation control as its record.
182    pub fn encode(
183        &self,
184        control: &crate::read_control::StorageReadControl,
185    ) -> crate::mvcc::VersionResult<uqa_core::memory::BudgetedVec<u8>> {
186        crate::key_value::record_json::encode(self, control)
187    }
188
189    /// Merge evaluated counter changes. A replacement object or overlapping reset that cannot retain the original baseline remains a conditional canonical replacement.
190    pub(crate) fn merge(
191        before: &Self,
192        after: &Self,
193        current: &Self,
194    ) -> crate::mvcc::VersionResult<Option<Self>> {
195        use crate::mvcc::VersionError;
196        if before
197            .object_id
198            .is_some_and(|id| Some(id) != after.object_id)
199            || current
200                .object_id
201                .is_some_and(|id| Some(id) != after.object_id)
202            || after.generation < before.generation
203        {
204            return Ok(None);
205        }
206        let changes =
207            i128::from(current.changes) + i128::from(after.changes) - i128::from(before.changes);
208        if changes < 0 {
209            return Ok(None);
210        }
211        let changes = u64::try_from(changes)
212            .map_err(|_| VersionError::InvalidEncoding("statistics change count overflow"))?;
213        let generation = current
214            .generation
215            .checked_add(after.generation - before.generation)
216            .ok_or(VersionError::InvalidEncoding(
217                "statistics generation space exhausted",
218            ))?;
219        let changed_estimate = after.statistics_format != 0
220            && (after.analyzed_rows != before.analyzed_rows
221                || after.statistics_format != before.statistics_format);
222        Ok(Some(Self {
223            object_id: after.object_id,
224            generation,
225            changes,
226            dirty_since_ms: if changes == 0 {
227                0
228            } else {
229                [after.dirty_since_ms, current.dirty_since_ms]
230                    .into_iter()
231                    .filter(|time| *time != 0)
232                    .min()
233                    .unwrap_or(0)
234            },
235            analyzed_rows: if changed_estimate {
236                after.analyzed_rows
237            } else {
238                current.analyzed_rows.or(after.analyzed_rows)
239            },
240            statistics_format: if changed_estimate {
241                after.statistics_format
242            } else {
243                current.statistics_format
244            },
245            // An analysis of the merged transaction is the one whose statistics it publishes.
246            analyzed_at: if after.analyzed_at == before.analyzed_at {
247                current.analyzed_at
248            } else {
249                after.analyzed_at
250            },
251        }))
252    }
253}
uqa_storage/statistics_maintenance.rs

uqa_storage/
statistics_maintenance.rs
