uqa_sql/catalog/events/
rule_condition_binding.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Structural OLD/NEW row bindings for persisted rewrite-rule conditions.
8
9use crate::ast::{InternalColumnRef, InternalRelationId, RuleEvent};
10use crate::ir::ScalarExpr;
11use crate::plan::ExpressionPlan;
12use serde::{Deserialize, Serialize};
13
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct RuleConditionBinding {
16    old_relation: Option<InternalRelationId>,
17    new_relation: Option<InternalRelationId>,
18    columns: Vec<String>,
19}
20
21impl RuleConditionBinding {
22    pub fn for_event(columns: &[String], event: RuleEvent) -> Self {
23        let old_relation = matches!(event, RuleEvent::Update | RuleEvent::Delete)
24            .then(InternalRelationId::allocate);
25        let new_relation = matches!(event, RuleEvent::Insert | RuleEvent::Update)
26            .then(InternalRelationId::allocate);
27        Self {
28            old_relation,
29            new_relation,
30            columns: columns.to_vec(),
31        }
32    }
33
34    pub const fn old_relation(&self) -> Option<InternalRelationId> {
35        self.old_relation
36    }
37
38    pub const fn new_relation(&self) -> Option<InternalRelationId> {
39        self.new_relation
40    }
41
42    pub fn old_column(&self, name: &str) -> Option<InternalColumnRef> {
43        self.column(self.old_relation, name)
44    }
45
46    pub fn new_column(&self, name: &str) -> Option<InternalColumnRef> {
47        self.column(self.new_relation, name)
48    }
49
50    pub fn column_name(&self, column: InternalColumnRef) -> Option<&str> {
51        if Some(column.relation()) != self.old_relation
52            && Some(column.relation()) != self.new_relation
53        {
54            return None;
55        }
56        self.columns.get(column.attribute()).map(String::as_str)
57    }
58
59    /// Give a deserialized condition plan process-local row identities before it can be combined with newly planned expressions.
60    pub fn reallocate_plan_relations(&self, plan: &mut ExpressionPlan) -> Self {
61        let rebound = Self {
62            old_relation: self.old_relation.map(|_| InternalRelationId::allocate()),
63            new_relation: self.new_relation.map(|_| InternalRelationId::allocate()),
64            columns: self.columns.clone(),
65        };
66        self.remap_plan(plan, &rebound);
67        rebound
68    }
69
70    /// A whole-row reference follows the live event row after ADD/DROP COLUMN, while existing field references retain their names.
71    pub fn refresh_columns(&self, columns: Vec<String>, plan: &mut ExpressionPlan) -> Self {
72        if self.columns == columns {
73            return self.clone();
74        }
75        let rebound = Self {
76            old_relation: self.old_relation.map(|_| InternalRelationId::allocate()),
77            new_relation: self.new_relation.map(|_| InternalRelationId::allocate()),
78            columns,
79        };
80        self.remap_plan(plan, &rebound);
81        rebound
82    }
83
84    /// Form the current event row without evaluating any of its fields.
85    pub fn whole_row_expression(&self, qualifier: &str, table: &str) -> Option<ScalarExpr> {
86        let relation = match qualifier {
87            "old" => self.old_relation?,
88            "new" => self.new_relation?,
89            _ => return None,
90        };
91        Some(ScalarExpr::Cast {
92            implicit: false,
93            expr: Box::new(ScalarExpr::Row(
94                (0..self.columns.len())
95                    .map(|position| ScalarExpr::InternalColumn(relation.column(position)))
96                    .collect(),
97            )),
98            ty: table.to_string(),
99        })
100    }
101
102    pub fn referenced_columns(
103        &self,
104        plan: &ExpressionPlan,
105    ) -> std::collections::BTreeSet<InternalColumnRef> {
106        let mut required = std::collections::BTreeSet::new();
107        let mut collect = |expression: &ScalarExpr| {
108            expression.visit(&mut |node| {
109                if let ScalarExpr::InternalColumn(column) = node {
110                    if self.column_name(*column).is_some() {
111                        required.insert(*column);
112                    }
113                }
114            });
115        };
116        collect(&plan.scalar);
117        for query in &plan.subqueries {
118            query.visit_scalar_expressions(&mut collect);
119        }
120        required
121    }
122
123    pub fn row_schema(&self, columns: &[crate::ast::ColumnDef]) -> crate::RowSchema {
124        let mut names = Vec::with_capacity(columns.len() * 2);
125        let mut identities = Vec::with_capacity(columns.len() * 2);
126        let mut types = Vec::with_capacity(columns.len() * 2);
127        let mut internal = Vec::with_capacity(columns.len() * 2);
128        for (side, relation) in [("old", self.old_relation()), ("new", self.new_relation())] {
129            let Some(relation) = relation else {
130                continue;
131            };
132            for (attribute, column) in columns.iter().enumerate() {
133                let (name, ty) = (&column.name, &column.ty);
134                let slot = names.len();
135                names.push(name.clone());
136                identities.push(crate::ColumnIdentity::qualified(side, name));
137                types.push(Some(ty.clone()));
138                internal.push((relation.column(attribute), slot, Some(ty.clone())));
139            }
140        }
141        let schema = crate::RowSchema::with_identities(names, identities, types)
142            .with_column_collations(
143                [self.old_relation(), self.new_relation()]
144                    .into_iter()
145                    .flatten()
146                    .flat_map(|_| columns.iter().map(|column| column.collation_oid)),
147            );
148        crate::RowSchema::with_physical_internal_aliases(&schema, &internal)
149    }
150
151    fn remap_plan(&self, plan: &mut ExpressionPlan, rebound: &Self) {
152        let mut rewrite = |expression: &mut ScalarExpr| {
153            let ScalarExpr::InternalColumn(column) = expression else {
154                return;
155            };
156            let relation = if Some(column.relation()) == self.old_relation {
157                rebound.old_relation
158            } else if Some(column.relation()) == self.new_relation {
159                rebound.new_relation
160            } else {
161                None
162            };
163            if let Some(relation) = relation {
164                if let Some(name) = self.column_name(*column) {
165                    if let Some(replacement) = rebound.column(Some(relation), name) {
166                        *column = replacement;
167                    }
168                }
169            }
170        };
171        crate::plan::rewrite_scalar_expression(&mut plan.scalar, &mut rewrite);
172        for subquery in &mut plan.subqueries {
173            subquery.rewrite_scalar_expressions(&mut rewrite);
174        }
175    }
176
177    fn column(
178        &self,
179        relation: Option<InternalRelationId>,
180        name: &str,
181    ) -> Option<InternalColumnRef> {
182        let relation = relation?;
183        self.columns
184            .iter()
185            .position(|column| column == name)
186            .map(|position| relation.column(position))
187    }
188}
189
190#[cfg(test)]
191mod tests {
192    use super::*;
193
194    #[test]
195    fn deserialized_plan_relations_are_reallocated_and_rewritten_together() {
196        let binding = RuleConditionBinding {
197            old_relation: Some(InternalRelationId::from_raw(u64::MAX - 1)),
198            new_relation: Some(InternalRelationId::from_raw(u64::MAX)),
199            columns: vec!["id".into()],
200        };
201        let mut plan = ExpressionPlan {
202            scalar: ScalarExpr::InternalColumn(binding.old_column("id").unwrap()),
203            subqueries: Vec::new(),
204        };
205
206        let rebound = binding.reallocate_plan_relations(&mut plan);
207        let ScalarExpr::InternalColumn(column) = plan.scalar else {
208            panic!("condition plan lost its structural OLD column")
209        };
210        assert_eq!(Some(column.relation()), rebound.old_relation());
211        assert_eq!(rebound.column_name(column), Some("id"));
212        assert_ne!(rebound.old_relation(), binding.old_relation());
213        assert_ne!(rebound.new_relation(), binding.new_relation());
214    }
215    #[test]
216    fn refreshed_columns_keep_field_names_and_expand_the_live_whole_row() {
217        let binding = RuleConditionBinding::for_event(
218            &["id".into(), "value".into(), "unused".into()],
219            RuleEvent::Update,
220        );
221        let mut plan = ExpressionPlan {
222            scalar: ScalarExpr::Row(vec![
223                ScalarExpr::InternalColumn(binding.new_column("value").unwrap()),
224                ScalarExpr::Column("old".into()),
225            ]),
226            subqueries: Vec::new(),
227        };
228        let refreshed =
229            binding.refresh_columns(vec!["value".into(), "id".into(), "added".into()], &mut plan);
230        assert_eq!(
231            refreshed.referenced_columns(&plan),
232            std::collections::BTreeSet::from([refreshed.new_column("value").unwrap()])
233        );
234        let ScalarExpr::Row(items) = &plan.scalar else {
235            panic!("outer row");
236        };
237        assert_eq!(
238            items[0],
239            ScalarExpr::InternalColumn(refreshed.new_column("value").unwrap())
240        );
241        let whole = refreshed
242            .whole_row_expression("old", "public.event")
243            .unwrap();
244        let ScalarExpr::Cast { expr, ty, .. } = &whole else {
245            panic!("typed whole row");
246        };
247        assert_eq!(ty, "public.event");
248        assert_eq!(
249            expr.as_ref(),
250            &ScalarExpr::Row(
251                ["value", "id", "added"]
252                    .map(|name| ScalarExpr::InternalColumn(refreshed.old_column(name).unwrap()))
253                    .to_vec()
254            )
255        );
256    }
257}
uqa_sql/catalog/events/rule_condition_binding.rs

uqa_sql/catalog/events/
rule_condition_binding.rs
