spg_engine/maintenance.rs
1//! Table-maintenance executors: `ANALYZE` (re-stat) and
2//! `COMPACT COLD SEGMENTS` (cold-segment merge). Split out of
3//! `lib.rs` (cut 19); verbatim move, the only edit beyond
4//! visibility is reuniting the `exec_compact_cold_segments` doc,
5//! whose first half had drifted above `set_session_param`.
6
7use alloc::string::{String, ToString};
8use alloc::vec::Vec;
9
10use spg_storage::{ColumnSchema, CompactReport, DataType, IndexKind, Row, StorageError, Value};
11
12use crate::{
13 COMPACTION_TARGET_DEFAULT_BYTES, Engine, EngineError, QueryResult, canonical_value_repr,
14 is_internal_table_name, sort_values_for_histogram, statistics,
15};
16
17impl Engine {
18 /// v6.2.0 — `ANALYZE [<table>]` runtime. Bare `ANALYZE` walks
19 /// every user table; `ANALYZE <name>` re-stats one. For each
20 /// target table, single-pass scan + per-column histogram +
21 /// `null_frac` + `n_distinct`. Replaces the table's prior
22 /// stats; resets the modified-row counter.
23 ///
24 /// v6.2.0 doesn't sample — it scans the full table. v6.2.x
25 /// can add reservoir sampling at the > 100 K-row mark; not a
26 /// scope blocker for the current commit since rows ≤ 100 K
27 /// analyse in milliseconds.
28 pub(crate) fn exec_analyze(
29 &mut self,
30 target: Option<&str>,
31 ) -> Result<QueryResult, EngineError> {
32 // v7.38 元机制 D acceptor — `SPG_TEST_STATS_FROZEN=1` turns
33 // ANALYZE into a no-op so a test's statistic-version snapshot
34 // is stable across sessions. Pairs with the plan-cache
35 // version-aware invalidation: freezing stats also freezes
36 // plan reuse, which is what regression tests want when
37 // proving "same SQL, same plan".
38 if self.env_cfg().stats_frozen {
39 return Ok(QueryResult::CommandOk {
40 affected: 0,
41 modified_catalog: false,
42 });
43 }
44 let names: Vec<String> = if let Some(name) = target {
45 // Verify the table exists; surface a clear error if not.
46 if self.catalog.get(name).is_none() {
47 return Err(EngineError::Storage(StorageError::TableNotFound {
48 name: name.to_string(),
49 }));
50 }
51 alloc::vec![name.to_string()]
52 } else {
53 self.catalog
54 .table_names()
55 .into_iter()
56 .filter(|n| !is_internal_table_name(n))
57 .collect()
58 };
59 let mut analysed = 0usize;
60 for table_name in &names {
61 self.analyze_one_table(table_name)?;
62 analysed += 1;
63 }
64 // v6.3.1 — plan cache invalidation. Bump stats version so
65 // future lookups see the new generation, and selectively
66 // evict every plan whose `source_tables` overlap with the
67 // ANALYZE target set. Bare ANALYZE (all tables) clears the
68 // whole cache.
69 if analysed > 0 {
70 self.statistics.bump_version();
71 if target.is_some() {
72 for t in &names {
73 self.plan_cache.evict_referencing(t);
74 }
75 } else {
76 self.plan_cache.clear();
77 }
78 }
79 Ok(QueryResult::CommandOk {
80 affected: analysed,
81 modified_catalog: true,
82 })
83 }
84
85 /// Walk a single table's rows once and (re-)populate per-column
86 /// stats. Drops the existing stats for `table` first so columns
87 /// that have been DROP-ed between ANALYZEs don't leave stale
88 /// rows.
89 fn analyze_one_table(&mut self, table_name: &str) -> Result<(), EngineError> {
90 let table = self.catalog.get(table_name).ok_or_else(|| {
91 EngineError::Storage(StorageError::TableNotFound {
92 name: table_name.to_string(),
93 })
94 })?;
95 let schema = table.schema().clone();
96 let row_count = table.rows().len();
97 // For each column, collect (sorted) non-NULL textual values
98 // + count NULLs; then ask `statistics::build_histogram` to
99 // produce the 101 bounds and `estimate_n_distinct` the
100 // distinct count.
101 self.statistics.clear_table(table_name);
102 for (col_pos, col_schema) in schema.columns.iter().enumerate() {
103 // v6.2.0 skip: vector columns have their own stats
104 // shape (HNSW graph topology). v6.2 deliberation #1.
105 if matches!(col_schema.ty, DataType::Vector { .. }) {
106 continue;
107 }
108 let mut non_null_values: Vec<Value<'static>> = Vec::with_capacity(row_count);
109 let mut nulls: u64 = 0;
110 for row in table.rows() {
111 match row.values.get(col_pos) {
112 Some(Value::Null) | None => nulls += 1,
113 Some(v) => non_null_values.push(v.clone()),
114 }
115 }
116 // Sort by type-aware ordering (Int as int, Text as
117 // lex, etc.) so histogram bounds reflect the column's
118 // natural order — not lexicographic on the string
119 // representation, which would put "9" after "49".
120 non_null_values.sort_by(|a, b| sort_values_for_histogram(a, b));
121 let non_null: Vec<String> = non_null_values.iter().map(canonical_value_repr).collect();
122 let null_frac = if row_count == 0 {
123 0.0
124 } else {
125 #[allow(clippy::cast_precision_loss)]
126 let f = nulls as f32 / row_count as f32;
127 f
128 };
129 let n_distinct = statistics::estimate_n_distinct(&non_null);
130 let histogram_bounds = statistics::build_histogram(&non_null);
131 self.statistics.set(
132 table_name.to_string(),
133 col_schema.name.clone(),
134 statistics::ColumnStats {
135 null_frac,
136 n_distinct,
137 histogram_bounds,
138 },
139 );
140 }
141 self.statistics.reset_modified(table_name);
142 // v6.7.0 — refresh the per-table cold_rows cache. Walk the
143 // BTree indices and count Cold locators (MAX across
144 // indices); store the result on the table. Surfaced via
145 // `spg_statistic.cold_row_count` (new column) and
146 // `spg_stat_segment.table_name` (new column).
147 let cold_count = {
148 let table = self
149 .active_catalog()
150 .get(table_name)
151 .expect("table still present");
152 table.count_cold_locators()
153 };
154 let table_mut = self
155 .active_catalog_mut()
156 .get_mut(table_name)
157 .expect("table still present");
158 table_mut.set_cold_row_count(cold_count);
159 Ok(())
160 }
161
162 /// v6.7.3 — `COMPACT COLD SEGMENTS` runtime path. Drives the
163 /// engine-layer compaction shim with the default
164 /// 4 MiB segment-size threshold. spg-server intercepts the
165 /// SQL before it reaches the engine on a server build —
166 /// it reads `SPG_COMPACTION_TARGET_SEGMENT_BYTES`, calls
167 /// `Engine::compact_cold_segments_with_target` directly with
168 /// the env value, and persists every merged segment to
169 /// `<db>.spg/segments/`. This arm only fires for engine-only
170 /// callers (spg-embedded, lib tests); in that mode merged
171 /// segments live in memory and are dropped at process exit.
172 pub(crate) fn exec_compact_cold_segments(&mut self) -> Result<QueryResult, EngineError> {
173 let target = COMPACTION_TARGET_DEFAULT_BYTES;
174 let reports = self.compact_cold_segments_with_target(target)?;
175 let columns = alloc::vec![
176 ColumnSchema::new("table_name", DataType::Text, false),
177 ColumnSchema::new("index_name", DataType::Text, false),
178 ColumnSchema::new("sources_merged", DataType::BigInt, false),
179 ColumnSchema::new("merged_segment_id", DataType::BigInt, false),
180 ColumnSchema::new("merged_rows", DataType::BigInt, false),
181 ColumnSchema::new("deleted_rows_pruned", DataType::BigInt, false),
182 ColumnSchema::new("bytes_reclaimed_estimate", DataType::BigInt, false),
183 ];
184 let rows: Vec<Row<'static>> = reports
185 .into_iter()
186 .map(|(tname, iname, report)| {
187 Row::new(alloc::vec![
188 Value::text(tname),
189 Value::text(iname),
190 Value::BigInt(i64::try_from(report.sources.len()).unwrap_or(i64::MAX)),
191 Value::BigInt(i64::from(report.merged_segment_id.unwrap_or(0))),
192 Value::BigInt(i64::try_from(report.merged_rows).unwrap_or(i64::MAX)),
193 Value::BigInt(i64::try_from(report.deleted_rows_pruned).unwrap_or(i64::MAX),),
194 Value::BigInt(
195 i64::try_from(report.bytes_reclaimed_estimate).unwrap_or(i64::MAX),
196 ),
197 ])
198 })
199 .collect();
200 Ok(QueryResult::Rows { columns, rows })
201 }
202
203 /// v6.7.3 — public shim around `Catalog::compact_cold_segments`
204 /// driving every BTree index on every user table. Returns one
205 /// `(table, index, report)` triple for each merge that
206 /// actually happened (no-op (table, index) pairs are filtered
207 /// out so callers can size persist-side work to the live
208 /// merges). Caller is responsible for persisting each
209 /// `report.merged_segment_bytes` and updating the on-disk
210 /// segment registry; engine layer is no_std and never
211 /// touches disk.
212 ///
213 /// Marks every touched table's cached `cold_row_count` stale
214 /// — compaction GC'd some shadowed rows, so the count must be
215 /// re-derived on the next ANALYZE.
216 pub fn compact_cold_segments_with_target(
217 &mut self,
218 target_segment_bytes: u64,
219 ) -> Result<Vec<(String, String, CompactReport)>, EngineError> {
220 let table_names = self.active_catalog().table_names();
221 let mut reports: Vec<(String, String, CompactReport)> = Vec::new();
222 for tname in table_names {
223 if is_internal_table_name(&tname) {
224 continue;
225 }
226 let idx_names: Vec<String> = {
227 let Some(t) = self.active_catalog().get(&tname) else {
228 continue;
229 };
230 t.indices()
231 .iter()
232 .filter(|i| matches!(i.kind, IndexKind::BTree(_)))
233 .map(|i| i.name.clone())
234 .collect()
235 };
236 for iname in idx_names {
237 let report = self
238 .active_catalog_mut()
239 .compact_cold_segments(&tname, &iname, target_segment_bytes)
240 .map_err(EngineError::Storage)?;
241 if report.merged_segment_id.is_some() {
242 if let Some(t) = self.active_catalog_mut().get_mut(&tname) {
243 t.mark_cold_row_count_stale();
244 }
245 reports.push((tname.clone(), iname, report));
246 }
247 }
248 }
249 Ok(reports)
250 }
251}