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<'a>(schemas: impl IntoIterator<Item = &'a str>) -> BTreeSet<String> {
61    let mut queued = BTreeSet::new();
62    let mut created = BTreeSet::new();
63    for sql in schemas {
64        queued.extend(queued_projections(sql));
65        created.extend(created_tables(sql));
66    }
67    queued.intersection(&created).cloned().collect()
68}
69
70// Why: the queueing calls sit inside PL/pgSQL bodies, which pg_query keeps
71// as opaque strings, so the projection names are read from the text.
72fn queued_projections(sql: &str) -> Vec<String> {
73    const CALL: &str = "queue_projection_backlog(";
74    sql.match_indices(CALL)
75        .filter_map(|(at, _)| {
76            let rest = sql[at + CALL.len()..].trim_start().strip_prefix('\'')?;
77            let end = rest.find('\'')?;
78            Some(rest[..end].to_owned())
79        })
80        .collect()
81}
82
83fn created_tables(sql: &str) -> Vec<String> {
84    let Ok(parsed) = pg_query::parse(sql) else {
85        return Vec::new();
86    };
87    parsed
88        .protobuf
89        .stmts
90        .iter()
91        .filter_map(|raw| match raw.stmt.as_ref()?.node.as_ref()? {
92            NodeEnum::CreateStmt(stmt) => Some(stmt.relation.as_ref()?.relname.clone()),
93            _ => None,
94        })
95        .collect()
96}
97
98#[must_use]
99pub fn backfill_statements(sql: &str) -> Vec<BackfillStatement> {
100    let Ok(parsed) = pg_query::parse(sql) else {
101        return Vec::new();
102    };
103    parsed
104        .protobuf
105        .stmts
106        .iter()
107        .filter_map(|raw| raw.stmt.as_ref().and_then(|s| s.node.as_ref()))
108        .enumerate()
109        .filter_map(|(index, node)| {
110            classify(node).map(|(target, form)| BackfillStatement {
111                position: index + 1,
112                target,
113                form,
114            })
115        })
116        .collect()
117}
118
119fn classify(node: &NodeEnum) -> Option<(String, &'static str)> {
120    match node {
121        NodeEnum::UpdateStmt(stmt) => Some((stmt.relation.as_ref()?.relname.clone(), "UPDATE")),
122        NodeEnum::DeleteStmt(stmt) => Some((stmt.relation.as_ref()?.relname.clone(), "DELETE")),
123        NodeEnum::InsertStmt(stmt) => {
124            let select = select_of(stmt.select_stmt.as_deref())?;
125            let table = stmt.relation.as_ref()?.relname.clone();
126            if table == PROJECTION_BACKLOG && is_keys_only(select) {
127                return None;
128            }
129            Some((table, "INSERT … SELECT"))
130        },
131        NodeEnum::SelectStmt(select) => {
132            refresh_call(select).map(|name| (name, "SELECT refresh_*()"))
133        },
134        _ => None,
135    }
136}
137
138// Why: pg_query models `INSERT … VALUES` as a SelectStmt carrying
139// `values_lists`; only a select without them reads other rows.
140fn select_of(node: Option<&Node>) -> Option<&protobuf::SelectStmt> {
141    match node?.node.as_ref()? {
142        NodeEnum::SelectStmt(select) if select.values_lists.is_empty() => Some(select),
143        _ => None,
144    }
145}
146
147// Why: a keys-only queue insert projects two columns from one relation with
148// no join, aggregate or grouping — it costs one index scan, which is the
149// whole point of queuing instead of rebuilding. DISTINCT is allowed: it
150// collapses the keys before the insert instead of leaving one conflict check
151// per ledger row to `ON CONFLICT DO NOTHING`.
152fn is_keys_only(select: &protobuf::SelectStmt) -> bool {
153    let single_relation = select.from_clause.len() == 1
154        && matches!(
155            select.from_clause[0].node.as_ref(),
156            Some(NodeEnum::RangeVar(_))
157        );
158    single_relation
159        && select.target_list.len() == 2
160        && select.group_clause.is_empty()
161        && select.having_clause.is_none()
162        && select.with_clause.is_none()
163}
164
165fn refresh_call(select: &protobuf::SelectStmt) -> Option<String> {
166    select.target_list.iter().find_map(|target| {
167        let NodeEnum::ResTarget(res) = target.node.as_ref()? else {
168            return None;
169        };
170        let NodeEnum::FuncCall(call) = res.val.as_ref()?.node.as_ref()? else {
171            return None;
172        };
173        let name = call.funcname.last().and_then(|n| match n.node.as_ref()? {
174            NodeEnum::String(s) => Some(s.sval.clone()),
175            _ => None,
176        })?;
177        name.starts_with("refresh_").then_some(name)
178    })
179}
systemprompt_database/lifecycle/installation/backfill.rs

systemprompt_database/lifecycle/installation/
backfill.rs
