uqa_sql/
maintenance.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Interpret VACUUM options and bind target relation metadata before storage maintenance.
8use crate::{
9    ast::{VacuumOption, VacuumOptionValue, VacuumStmt},
10    SQLError,
11};
12use std::collections::BTreeSet;
13pub struct VacuumOptions {
14    flags: VacuumFlags,
15}
16pub struct ResolvedVacuumTarget {
17    pub table: String,
18    pub include_descendants: bool,
19    pub columns: Vec<String>,
20}
21pub trait VacuumRelation {
22    fn column_names(&self) -> BTreeSet<String>;
23}
24pub trait VacuumCatalog {
25    fn resolve_relation_kind(&self, name: &str)
26        -> Result<Option<(String, &'static str)>, SQLError>;
27    fn require_table(&self, name: &str) -> Result<Box<dyn VacuumRelation + '_>, SQLError>;
28}
29pub trait VacuumPrivileges {
30    fn ensure_maintain(&self, table: &str) -> Result<(), SQLError>;
31}
32#[derive(Clone, Copy, Default)]
33struct VacuumFlags(u8);
34
35impl VacuumFlags {
36    const ANALYZE: u8 = 1 << 0;
37    const FULL: u8 = 1 << 1;
38    const DISABLE_PAGE_SKIPPING: u8 = 1 << 2;
39    const PROCESS_TOAST: u8 = 1 << 3;
40    const ONLY_DATABASE_STATS: u8 = 1 << 4;
41    const ONLY_DATABASE_STATS_CONFLICT: u8 = 1 << 5;
42
43    fn insert(&mut self, flag: u8, enabled: bool) {
44        if enabled {
45            self.0 |= flag;
46        }
47    }
48
49    pub const fn contains(self, flag: u8) -> bool {
50        self.0 & flag != 0
51    }
52}
53
54impl VacuumOptions {
55    pub const fn analyze(&self) -> bool {
56        self.flags.contains(VacuumFlags::ANALYZE)
57    }
58
59    pub const fn full(&self) -> bool {
60        self.flags.contains(VacuumFlags::FULL)
61    }
62
63    pub const fn disable_page_skipping(&self) -> bool {
64        self.flags.contains(VacuumFlags::DISABLE_PAGE_SKIPPING)
65    }
66
67    pub const fn process_toast(&self) -> bool {
68        self.flags.contains(VacuumFlags::PROCESS_TOAST)
69    }
70
71    pub const fn only_database_stats(&self) -> bool {
72        self.flags.contains(VacuumFlags::ONLY_DATABASE_STATS)
73    }
74
75    pub const fn has_only_database_stats_conflict(&self) -> bool {
76        self.flags
77            .contains(VacuumFlags::ONLY_DATABASE_STATS_CONFLICT)
78    }
79}
80
81fn vacuum_syntax_error(message: impl Into<String>) -> SQLError {
82    SQLError::Routine {
83        sqlstate: "42601".into(),
84        message: message.into(),
85    }
86}
87
88fn invalid_buffer_usage_limit() -> SQLError {
89    SQLError::Routine {
90        sqlstate: "22023".into(),
91        message: "BUFFER_USAGE_LIMIT option must be 0 or between 128 kB and 16777216 kB".into(),
92    }
93}
94
95fn vacuum_feature_error(message: impl Into<String>) -> SQLError {
96    SQLError::Routine {
97        sqlstate: "0A000".into(),
98        message: message.into(),
99    }
100}
101
102fn buffer_usage_limit_kib(value: &VacuumOptionValue) -> Result<u64, SQLError> {
103    let (amount, multiplier) = match value {
104        VacuumOptionValue::Integer(value) => {
105            let amount = u64::try_from(*value).map_err(|_| invalid_buffer_usage_limit())?;
106            return Ok(amount);
107        }
108        VacuumOptionValue::String(value) => {
109            let value = value.trim();
110            let split = value
111                .find(|character: char| !(character.is_ascii_digit() || character == '.'))
112                .unwrap_or(value.len());
113            let amount = value[..split]
114                .parse::<f64>()
115                .map_err(|_| invalid_buffer_usage_limit())?;
116            let unit = value[split..].trim().to_ascii_lowercase();
117            let multiplier = match unit.as_str() {
118                "" | "kb" => 1_f64,
119                "mb" => 1024_f64,
120                "gb" => 1024_f64 * 1024_f64,
121                "tb" => 1024_f64 * 1024_f64 * 1024_f64,
122                _ => return Err(invalid_buffer_usage_limit()),
123            };
124            (amount, multiplier)
125        }
126        VacuumOptionValue::Boolean(_) => return Err(invalid_buffer_usage_limit()),
127    };
128    let kib = amount * multiplier;
129    if !kib.is_finite() || kib < 0.0 || kib.round() > u64::MAX as f64 {
130        return Err(invalid_buffer_usage_limit());
131    }
132    Ok(kib.round() as u64)
133}
134
135fn boolean_option(option: &VacuumOption) -> Result<bool, SQLError> {
136    let Some(value) = option.value.as_ref() else {
137        return Ok(true);
138    };
139    match value {
140        VacuumOptionValue::Boolean(value) => Ok(*value),
141        VacuumOptionValue::String(value) => match value.to_ascii_lowercase().as_str() {
142            "true" | "on" => Ok(true),
143            "false" | "off" => Ok(false),
144            _ => Err(vacuum_syntax_error(format!(
145                "{} requires a Boolean value",
146                option.name
147            ))),
148        },
149        VacuumOptionValue::Integer(0) => Ok(false),
150        VacuumOptionValue::Integer(1) => Ok(true),
151        VacuumOptionValue::Integer(_) => Err(vacuum_syntax_error(format!(
152            "{} requires a Boolean value",
153            option.name
154        ))),
155    }
156}
157
158#[expect(
159    clippy::too_many_lines,
160    reason = "preserves VACUUM option validation order"
161)]
162fn validate_options(options: &[VacuumOption]) -> Result<VacuumOptions, SQLError> {
163    let mut analyze = false;
164    let mut full = false;
165    let mut parallel = 0_i64;
166    let mut buffer_usage_limit_specified = false;
167    let mut disable_page_skipping = false;
168    let mut process_toast = true;
169    let mut only_database_stats = false;
170    let mut effective_options = BTreeSet::new();
171    for option in options {
172        match option.name.as_str() {
173            "analyze" => {
174                analyze = boolean_option(option)?;
175                if analyze {
176                    effective_options.insert(option.name.as_str());
177                } else {
178                    effective_options.remove(option.name.as_str());
179                }
180            }
181            "full" => {
182                full = boolean_option(option)?;
183                if full {
184                    effective_options.insert(option.name.as_str());
185                } else {
186                    effective_options.remove(option.name.as_str());
187                }
188            }
189            "only_database_stats" => only_database_stats = boolean_option(option)?,
190            "freeze" | "skip_locked" | "skip_database_stats" => {
191                if boolean_option(option)? {
192                    effective_options.insert(option.name.as_str());
193                } else {
194                    effective_options.remove(option.name.as_str());
195                }
196            }
197            "disable_page_skipping" => {
198                disable_page_skipping = boolean_option(option)?;
199                if disable_page_skipping {
200                    effective_options.insert(option.name.as_str());
201                } else {
202                    effective_options.remove(option.name.as_str());
203                }
204            }
205            "process_toast" => process_toast = boolean_option(option)?,
206            // PostgreSQL 18 permits these with ONLY_DATABASE_STATS; they are not part of the VACOPT conflict mask used by that check.
207            "verbose" | "process_main" | "truncate" => {
208                boolean_option(option)?;
209            }
210            "index_cleanup" => {
211                if !matches!(
212                    option.value.as_ref(),
213                    Some(VacuumOptionValue::String(value)) if value.eq_ignore_ascii_case("auto")
214                ) {
215                    boolean_option(option)?;
216                }
217            }
218            "parallel" => {
219                let Some(VacuumOptionValue::Integer(workers)) = option.value.as_ref() else {
220                    return Err(vacuum_syntax_error(
221                        "parallel requires a non-negative integer value",
222                    ));
223                };
224                if !(0..=1024).contains(workers) {
225                    return Err(vacuum_syntax_error(
226                        "parallel workers for vacuum must be between 0 and 1024",
227                    ));
228                }
229                parallel = i64::from(*workers);
230            }
231            "buffer_usage_limit" => {
232                buffer_usage_limit_specified = true;
233                let kib = option
234                    .value
235                    .as_ref()
236                    .ok_or_else(invalid_buffer_usage_limit)
237                    .and_then(buffer_usage_limit_kib)?;
238                if kib != 0 && !(128..=16_777_216).contains(&kib) {
239                    return Err(invalid_buffer_usage_limit());
240                }
241            }
242            name => {
243                return Err(vacuum_syntax_error(format!(
244                    "unrecognized VACUUM option \"{name}\""
245                )));
246            }
247        }
248    }
249    if full && parallel > 0 {
250        return Err(vacuum_feature_error(
251            "VACUUM FULL cannot be performed in parallel",
252        ));
253    }
254    if buffer_usage_limit_specified && full && !analyze {
255        return Err(vacuum_feature_error(
256            "BUFFER_USAGE_LIMIT cannot be specified for VACUUM FULL",
257        ));
258    }
259    let mut flags = VacuumFlags::default();
260    flags.insert(VacuumFlags::ANALYZE, analyze);
261    flags.insert(VacuumFlags::FULL, full);
262    flags.insert(VacuumFlags::DISABLE_PAGE_SKIPPING, disable_page_skipping);
263    flags.insert(VacuumFlags::PROCESS_TOAST, process_toast);
264    flags.insert(VacuumFlags::ONLY_DATABASE_STATS, only_database_stats);
265    flags.insert(
266        VacuumFlags::ONLY_DATABASE_STATS_CONFLICT,
267        !effective_options.is_empty(),
268    );
269    Ok(VacuumOptions { flags })
270}
271
272pub fn analyze_vacuum(statement: &VacuumStmt) -> Result<VacuumOptions, SQLError> {
273    let execution = validate_options(&statement.options)?;
274    if !execution.analyze()
275        && statement
276            .targets
277            .iter()
278            .any(|target| !target.columns.is_empty())
279    {
280        return Err(SQLError::Routine {
281            sqlstate: "0A000".into(),
282            message: "ANALYZE option must be specified when a column list is provided".into(),
283        });
284    }
285    if execution.full() && execution.disable_page_skipping() {
286        return Err(vacuum_feature_error(
287            "VACUUM option DISABLE_PAGE_SKIPPING cannot be used with FULL",
288        ));
289    }
290    if execution.full() && !execution.process_toast() {
291        return Err(vacuum_feature_error(
292            "PROCESS_TOAST required with VACUUM FULL",
293        ));
294    }
295    if execution.only_database_stats() && !statement.targets.is_empty() {
296        return Err(vacuum_feature_error(
297            "ONLY_DATABASE_STATS cannot be specified with a list of tables",
298        ));
299    }
300    if execution.only_database_stats() && execution.has_only_database_stats_conflict() {
301        return Err(vacuum_feature_error(
302            "ONLY_DATABASE_STATS cannot be specified with other VACUUM options",
303        ));
304    }
305
306    Ok(execution)
307}
308pub fn bind_vacuum_targets(
309    catalog: &dyn VacuumCatalog,
310    privileges: &dyn VacuumPrivileges,
311    statement: &VacuumStmt,
312) -> Result<Vec<ResolvedVacuumTarget>, SQLError> {
313    let mut resolved_targets = Vec::with_capacity(statement.targets.len());
314    for target in &statement.targets {
315        if target
316            .catalog
317            .as_deref()
318            .is_some_and(|catalog| catalog != "uqa")
319        {
320            let qualified = format!(
321                "{}.{}",
322                target.catalog.as_deref().expect("checked catalog"),
323                target.table
324            );
325            return Err(SQLError::Routine {
326                sqlstate: "0A000".into(),
327                message: format!("cross-database references are not implemented: \"{qualified}\""),
328            });
329        }
330        let canonical = match catalog.resolve_relation_kind(&target.table)? {
331            Some((canonical, "table")) => canonical,
332            Some(_) | None => return Err(SQLError::UnknownTable(target.table.clone())),
333        };
334        let table = catalog.require_table(&canonical)?;
335        if !target.columns.is_empty() {
336            let available = table.column_names();
337            if let Some(column) = target
338                .columns
339                .iter()
340                .find(|column| !available.contains(column.as_str()))
341            {
342                return Err(SQLError::Routine {
343                    sqlstate: "42703".into(),
344                    message: format!(
345                        "column \"{column}\" of relation \"{}\" does not exist",
346                        target.table
347                    ),
348                });
349            }
350        }
351        privileges.ensure_maintain(&canonical)?;
352        resolved_targets.push(ResolvedVacuumTarget {
353            table: canonical,
354            include_descendants: target.include_descendants,
355            columns: target.columns.clone(),
356        });
357    }
358
359    Ok(resolved_targets)
360}
uqa_sql/maintenance.rs

uqa_sql/
maintenance.rs
