systemprompt_database/lifecycle/installation/
backfill.rs

1//! Finds the statements in a migration that populate data rather than
2//! declare structure: a backfill.
3//!
4//! Under the data paradigm (`crate::projection`) an upgrade migration is DDL
5//! plus one keys-only `INSERT INTO projection_backlog … SELECT`. Everything a
6//! fact needs is in the base schema — the projector, the statement trigger
7//! that queues keys, the drainer job — so a fresh install (which stamps
8//! migrations and never runs them) and an upgraded one reach the same state
9//! by the same code. A migration that instead `UPDATE`s a fact column,
10//! `INSERT … SELECT`s a rollup or calls `refresh_*()` inline populates the
11//! upgraded database by a path the fresh one never takes, on the boot path,
12//! at a cost that grows with history.
13//!
14//! The detector reports `UPDATE`, `DELETE`, select-driven `INSERT` and any
15//! `SELECT` that calls a `refresh_*` routine; the single sanctioned keys-only
16//! `INSERT INTO projection_backlog SELECT <projection>, <key> FROM <one
17//! relation>` is not reported. An unparseable body yields nothing, since
18//! `check_migration_references` already refuses it with the parse error.
19//!
20//! A `DELETE` from `projection_backlog`, or from a projection's own fact
21//! table, is cleanup rather than backfill: it removes rows the base schema's
22//! projector would no longer produce, and a fresh install never had them.
23//! [`projection_tables`] derives that set from the base schema — every
24//! projection a schema queues through `queue_projection_backlog('<name>', …)`
25//! whose name is also a table the base schema creates — and
26//! [`BackfillStatement::is_projection_cleanup`] applies it.
27//!
28//! Both repositories' static gates (`no_backfill_after_floor`) use this one
29//! detector so "backfill" means the same thing on each side.
30//!
31//! Copyright (c) systemprompt.io — Business Source License 1.1.
32//! See <https://systemprompt.io> for licensing details.
33
34use std::collections::BTreeSet;
35
36use pg_query::NodeEnum;
37use pg_query::protobuf::{self, Node};
38
39pub const PROJECTION_BACKLOG: &str = "projection_backlog";
40
41/// One data-populating statement. `position` is 1-based, matching how the
42/// runner numbers statements when one fails. `target` is the table written or
43/// the function called.
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct BackfillStatement {
46    pub position: usize,
47    pub target: String,
48    pub form: &'static str,
49}
50
51impl BackfillStatement {
52    #[must_use]
53    pub fn is_projection_cleanup(&self, projection_tables: &BTreeSet<String>) -> bool {
54        self.form == "DELETE"
55            && (self.target == PROJECTION_BACKLOG || projection_tables.contains(&self.target))
56    }
57}
58
59#[must_use]
60pub fn projection_tables<S: AsRef<str>>(schemas: impl IntoIterator<Item = S>) -> BTreeSet<String> {
61    let mut queued = BTreeSet::new();
62    let mut created = BTreeSet::new();
63    for sql in schemas {
64        let sql = sql.as_ref();
65        queued.extend(queued_projections(sql));
66        created.extend(created_tables(sql));
67    }
68    queued.intersection(&created).cloned().collect()
69}
70
71// Why: the queueing calls sit inside PL/pgSQL bodies, which pg_query keeps
72// as opaque strings, so the projection names are read from the text.
73fn queued_projections(sql: &str) -> Vec<String> {
74    const CALL: &str = "queue_projection_backlog(";
75    sql.match_indices(CALL)
76        .filter_map(|(at, _)| {
77            let rest = sql[at + CALL.len()..].trim_start().strip_prefix('\'')?;
78            let end = rest.find('\'')?;
79            Some(rest[..end].to_owned())
80        })
81        .collect()
82}
83
84fn created_tables(sql: &str) -> Vec<String> {
85    let Ok(parsed) = pg_query::parse(sql) else {
86        return Vec::new();
87    };
88    parsed
89        .protobuf
90        .stmts
91        .iter()
92        .filter_map(|raw| match raw.stmt.as_ref()?.node.as_ref()? {
93            NodeEnum::CreateStmt(stmt) => Some(stmt.relation.as_ref()?.relname.clone()),
94            _ => None,
95        })
96        .collect()
97}
98
99#[must_use]
100pub fn backfill_statements(sql: &str) -> Vec<BackfillStatement> {
101    let Ok(parsed) = pg_query::parse(sql) else {
102        return Vec::new();
103    };
104    parsed
105        .protobuf
106        .stmts
107        .iter()
108        .filter_map(|raw| raw.stmt.as_ref().and_then(|s| s.node.as_ref()))
109        .enumerate()
110        .filter_map(|(index, node)| {
111            classify(node).map(|(target, form)| BackfillStatement {
112                position: index + 1,
113                target,
114                form,
115            })
116        })
117        .collect()
118}
119
120fn classify(node: &NodeEnum) -> Option<(String, &'static str)> {
121    match node {
122        NodeEnum::UpdateStmt(stmt) => Some((stmt.relation.as_ref()?.relname.clone(), "UPDATE")),
123        NodeEnum::DeleteStmt(stmt) => Some((stmt.relation.as_ref()?.relname.clone(), "DELETE")),
124        NodeEnum::InsertStmt(stmt) => {
125            let select = select_of(stmt.select_stmt.as_deref())?;
126            let table = stmt.relation.as_ref()?.relname.clone();
127            if table == PROJECTION_BACKLOG && is_keys_only(select) {
128                return None;
129            }
130            Some((table, "INSERT … SELECT"))
131        },
132        NodeEnum::SelectStmt(select) => {
133            refresh_call(select).map(|name| (name, "SELECT refresh_*()"))
134        },
135        _ => None,
136    }
137}
138
139// Why: pg_query models `INSERT … VALUES` as a SelectStmt carrying
140// `values_lists`; only a select without them reads other rows.
141fn select_of(node: Option<&Node>) -> Option<&protobuf::SelectStmt> {
142    match node?.node.as_ref()? {
143        NodeEnum::SelectStmt(select) if select.values_lists.is_empty() => Some(select),
144        _ => None,
145    }
146}
147
148// Why: a keys-only queue insert projects two columns from one relation with
149// no join, aggregate or grouping — it costs one index scan, which is the
150// whole point of queuing instead of rebuilding. DISTINCT is allowed: it
151// collapses the keys before the insert instead of leaving one conflict check
152// per ledger row to `ON CONFLICT DO NOTHING`.
153fn is_keys_only(select: &protobuf::SelectStmt) -> bool {
154    let single_relation = select.from_clause.len() == 1
155        && matches!(
156            select.from_clause[0].node.as_ref(),
157            Some(NodeEnum::RangeVar(_))
158        );
159    single_relation
160        && select.target_list.len() == 2
161        && select.group_clause.is_empty()
162        && select.having_clause.is_none()
163        && select.with_clause.is_none()
164}
165
166fn refresh_call(select: &protobuf::SelectStmt) -> Option<String> {
167    select.target_list.iter().find_map(|target| {
168        let NodeEnum::ResTarget(res) = target.node.as_ref()? else {
169            return None;
170        };
171        let NodeEnum::FuncCall(call) = res.val.as_ref()?.node.as_ref()? else {
172            return None;
173        };
174        let name = call.funcname.last().and_then(|n| match n.node.as_ref()? {
175            NodeEnum::String(s) => Some(s.sval.clone()),
176            _ => None,
177        })?;
178        name.starts_with("refresh_").then_some(name)
179    })
180}
systemprompt_database/lifecycle/installation/backfill.rs

systemprompt_database/lifecycle/installation/
backfill.rs
