uqa_sql/catalog/events/definition/
triggers.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7use super::EventAnalysisContext;
8use crate::{
9    ast::{
10        ColumnDef, ColumnType, CreateTrigger, Expr, FunctionReturns, TriggerEvent, TriggerTiming,
11    },
12    catalog::{
13        events::validation::{
14            is_boolean_type, validate_trigger_condition_references,
15            validate_trigger_transition_relation, TriggerConditionTypeResolver,
16        },
17        resolution::{RelationLookupMode, RelationResolution},
18        security::table::TableAclPrivilege,
19    },
20    plpgsql::bind_expr,
21    routines::{routine_signature_types, SQLUserFunction},
22    type_resolution::canonical_routine_type_name,
23    SQLError,
24};
25use std::sync::Arc;
26use uqa_core::{RelationIdentity, Value};
27
28#[cfg(test)]
29mod tests;
30
31impl EventAnalysisContext<'_> {
32    pub fn trigger_relation_definitions(&self, name: &str) -> Result<Vec<ColumnDef>, SQLError> {
33        let (relation, kind) =
34            self.resolve_trigger_relation_kind(name, RelationLookupMode::Bound)?;
35        self.trigger_relation_columns(&relation, kind)
36    }
37
38    pub fn resolve_trigger_table(&self, name: &str) -> Result<RelationIdentity, SQLError> {
39        let RelationResolution::Found(canonical, kind) =
40            self.relations.resolve_bound_relation_kind(name)?
41        else {
42            return Err(SQLError::UnknownTable(name.to_string()));
43        };
44        if !matches!(
45            kind,
46            "table" | "view" | "materialized view" | "foreign table"
47        ) {
48            return Err(SQLError::UnknownTable(name.to_string()));
49        }
50        RelationIdentity::from_legacy_name(&canonical).map_err(|error| {
51            SQLError::Internal(format!(
52                "decode bound trigger relation `{canonical}`: {error}"
53            ))
54        })
55    }
56
57    fn trigger_relation_columns(
58        &self,
59        relation: &RelationIdentity,
60        kind: &str,
61    ) -> Result<Vec<ColumnDef>, SQLError> {
62        if kind == "table" {
63            return self
64                .returning
65                .catalog
66                .try_describe_table_row_type(&relation.qualified_name())
67                .map_err(|error| SQLError::Internal(format!("read trigger columns: {error}")))?
68                .ok_or_else(|| SQLError::UnknownTable(relation.qualified_name()));
69        }
70        if kind == "foreign table" {
71            let columns = self
72                .catalog
73                .foreign_columns(relation)
74                .ok_or_else(|| SQLError::UnknownTable(relation.qualified_name()))?;
75            return Ok(columns);
76        }
77        let view = self
78            .catalog
79            .restored_catalog_view_definition(&relation.qualified_name())?
80            .ok_or_else(|| SQLError::UnknownTable(relation.qualified_name()))?;
81        let schema = self.catalog.stored_view_schema(&view)?;
82        Ok(schema
83            .columns()
84            .iter()
85            .enumerate()
86            .map(|(position, name)| {
87                let mut column = ColumnDef::nullable(
88                    schema.public_name(position).unwrap_or(name).to_string(),
89                    schema
90                        .column_type(position)
91                        .cloned()
92                        .unwrap_or(ColumnType::Text),
93                );
94                column.collation_oid = schema.column_collation(position);
95                column
96            })
97            .collect())
98    }
99
100    pub fn trigger_relation_from_resolution(
101        name: &str,
102        resolution: RelationResolution,
103    ) -> Result<(RelationIdentity, &'static str), SQLError> {
104        match resolution {
105            RelationResolution::Found(canonical, "foreign table") => {
106                let relation = RelationIdentity::from_legacy_name(&canonical).map_err(|error| {
107                    SQLError::Internal(format!(
108                        "decode resolved trigger relation `{canonical}`: {error}"
109                    ))
110                })?;
111                Ok((relation, "foreign table"))
112            }
113            resolution => Self::event_relation_from_resolution(name, resolution),
114        }
115    }
116
117    pub fn resolve_trigger_relation_kind(
118        &self,
119        name: &str,
120        lookup_mode: RelationLookupMode,
121    ) -> Result<(RelationIdentity, &'static str), SQLError> {
122        let resolution = match lookup_mode {
123            RelationLookupMode::Dynamic => self.relations.resolve_visible_relation_kind(name)?,
124            RelationLookupMode::Bound => self.relations.resolve_bound_relation_kind(name)?,
125        };
126        Self::trigger_relation_from_resolution(name, resolution)
127    }
128
129    fn resolve_trigger_function_candidate(
130        &self,
131        name: &str,
132        lookup_mode: RelationLookupMode,
133    ) -> Result<Arc<SQLUserFunction>, SQLError> {
134        let candidates = match lookup_mode {
135            RelationLookupMode::Dynamic => self.routines.lookup_visible_sql_functions(name)?,
136            RelationLookupMode::Bound => self.routines.lookup_bound_sql_functions(name),
137        }
138        .unwrap_or_default()
139        .into_iter()
140        .filter(|function| {
141            !function.def.is_procedure && routine_signature_types(&function.def).is_empty()
142        })
143        .collect::<Vec<_>>();
144        let function = match candidates.as_slice() {
145            [function] => function.clone(),
146            [] => {
147                return Err(SQLError::Routine {
148                    sqlstate: "42883".into(),
149                    message: format!("function {name}() does not exist"),
150                })
151            }
152            _ => {
153                return Err(SQLError::Routine {
154                    sqlstate: "42725".into(),
155                    message: format!("function name \"{name}\" is not unique"),
156                })
157            }
158        };
159        Ok(function)
160    }
161
162    fn validate_trigger_function(function: &SQLUserFunction) -> Result<(), SQLError> {
163        let returns_trigger = matches!(
164            &function.def.returns,
165            FunctionReturns::Scalar { type_name }
166                if canonical_routine_type_name(type_name) == "trigger"
167        );
168        if !returns_trigger {
169            return Err(SQLError::Routine {
170                sqlstate: "42P17".into(),
171                message: format!("function {} must return type trigger", function.def.name),
172            });
173        }
174        if function.def.language != "plpgsql" {
175            return Err(SQLError::Routine {
176                sqlstate: "0A000".into(),
177                message: "only LANGUAGE plpgsql trigger functions are executable".into(),
178            });
179        }
180        Ok(())
181    }
182
183    pub fn resolve_trigger_function(
184        &self,
185        name: &str,
186        lookup_mode: RelationLookupMode,
187    ) -> Result<Arc<SQLUserFunction>, SQLError> {
188        let function = self.resolve_trigger_function_candidate(name, lookup_mode)?;
189        Self::validate_trigger_function(&function)?;
190        Ok(function)
191    }
192
193    pub fn resolve_bound_trigger_function(
194        &self,
195        name: &str,
196        object_id: Option<[u8; 16]>,
197    ) -> Result<Arc<SQLUserFunction>, SQLError> {
198        let Some(object_id) = object_id else {
199            return self.resolve_trigger_function(name, RelationLookupMode::Bound);
200        };
201        let binding = crate::ast::FunctionBinding {
202            object_id: Some(object_id),
203            name: name.to_string(),
204            argument_types: Vec::new(),
205            builtin: false,
206            dispatch: None,
207            invocation: None,
208            composite_field: None,
209            resolution_error: None,
210        };
211        let candidates = self
212            .routines
213            .lookup_bound_sql_functions_by_binding(&binding)
214            .unwrap_or_default()
215            .into_iter()
216            .filter(|function| {
217                !function.def.is_procedure && routine_signature_types(&function.def).is_empty()
218            })
219            .collect::<Vec<_>>();
220        let function = match candidates.as_slice() {
221            [function] => function.clone(),
222            [] => {
223                return Err(SQLError::Routine {
224                    sqlstate: "42883".into(),
225                    message: format!("function {name}() does not exist"),
226                })
227            }
228            _ => {
229                return Err(SQLError::Internal(format!(
230                    "routine object identity for trigger function `{name}` is not unique"
231                )))
232            }
233        };
234        Self::validate_trigger_function(&function)?;
235        Ok(function)
236    }
237
238    fn ensure_trigger_creation_privilege(
239        &self,
240        relation: &RelationIdentity,
241        relation_kind: &str,
242    ) -> Result<(), SQLError> {
243        let canonical = relation.qualified_name();
244        match relation_kind {
245            "table" => {
246                let current_user = self.authority.current_role();
247                self.privileges.ensure_table_privilege_for(
248                    &canonical,
249                    &current_user,
250                    TableAclPrivilege::Trigger,
251                )
252            }
253            "view" => {
254                let view = self
255                    .catalog
256                    .restored_catalog_view_definition(&canonical)?
257                    .ok_or_else(|| {
258                        SQLError::Internal(format!(
259                            "resolved trigger view `{canonical}` has no catalog definition"
260                        ))
261                    })?;
262                self.privileges.ensure_view_privilege_for(
263                    &canonical,
264                    &view,
265                    &self.authority.current_role(),
266                    TableAclPrivilege::Trigger,
267                )
268            }
269            "foreign table" => self
270                .foreign_privileges
271                .ensure_foreign_table_privilege(&canonical, TableAclPrivilege::Trigger),
272            _ => Ok(()),
273        }
274    }
275
276    fn validate_trigger_relation_kind(
277        definition: &CreateTrigger,
278        relation: &RelationIdentity,
279        relation_kind: &str,
280    ) -> Result<(), SQLError> {
281        match relation_kind {
282            "table" if definition.timing == TriggerTiming::InsteadOf => Err(SQLError::Routine {
283                sqlstate: "42809".into(),
284                message: format!("\"{}\" is a table", relation.name),
285            }),
286            "view" if definition.timing == TriggerTiming::InsteadOf => {
287                if definition.events.contains(&TriggerEvent::Truncate)
288                    || !definition.transition_relations.is_empty()
289                {
290                    return Err(SQLError::Routine {
291                        sqlstate: "42809".into(),
292                        message: format!("\"{}\" is a view", relation.name),
293                    });
294                }
295                if !definition.row {
296                    return Err(SQLError::Routine {
297                        sqlstate: "0A000".into(),
298                        message: "INSTEAD OF triggers must be FOR EACH ROW".into(),
299                    });
300                }
301                if definition.when.is_some() {
302                    return Err(SQLError::Routine {
303                        sqlstate: "0A000".into(),
304                        message: "INSTEAD OF triggers cannot have WHEN conditions".into(),
305                    });
306                }
307                if !definition.update_columns.is_empty() {
308                    return Err(SQLError::Routine {
309                        sqlstate: "0A000".into(),
310                        message: "INSTEAD OF triggers cannot have column lists".into(),
311                    });
312                }
313                Ok(())
314            }
315            "view"
316                if definition.events.contains(&TriggerEvent::Truncate)
317                    || !definition.transition_relations.is_empty()
318                    || definition.row =>
319            {
320                Err(SQLError::Routine {
321                    sqlstate: "42809".into(),
322                    message: format!("\"{}\" is a view", relation.name),
323                })
324            }
325            "materialized view" => Err(SQLError::Routine {
326                sqlstate: "42809".into(),
327                message: format!("relation \"{}\" cannot have triggers", relation.name),
328            }),
329            "foreign table"
330                if definition.constraint || definition.timing == TriggerTiming::InsteadOf =>
331            {
332                Err(SQLError::Routine {
333                    sqlstate: "42809".into(),
334                    message: format!("\"{}\" is a foreign table", relation.name),
335                })
336            }
337            "view" | "foreign table" => Ok(()),
338            kind if kind != "table" => Err(SQLError::Routine {
339                sqlstate: "42809".into(),
340                message: format!("relation \"{}\" cannot have triggers", relation.name),
341            }),
342            _ => Ok(()),
343        }
344    }
345
346    pub fn validate_trigger_definition(
347        &self,
348        definition: &mut CreateTrigger,
349        lookup_mode: RelationLookupMode,
350    ) -> Result<(RelationIdentity, bool), SQLError> {
351        if definition.constraint && definition.or_replace {
352            return Err(SQLError::Routine {
353                sqlstate: "0A000".into(),
354                message: "CREATE OR REPLACE CONSTRAINT TRIGGER is not supported".into(),
355            });
356        }
357        if definition.constraint && (!definition.row || definition.timing != TriggerTiming::After) {
358            return Err(SQLError::Routine {
359                sqlstate: "0A000".into(),
360                message: "constraint triggers must be AFTER ROW triggers".into(),
361            });
362        }
363        if !definition.constraint
364            && (definition.deferrability.is_deferrable() || definition.referenced_table.is_some())
365        {
366            return Err(SQLError::Internal(
367                "ordinary trigger retained constraint-only metadata".into(),
368            ));
369        }
370        let (relation, relation_kind) =
371            self.resolve_trigger_relation_kind(&definition.table, lookup_mode)?;
372        definition.table = relation.qualified_name();
373        Self::validate_trigger_relation_kind(definition, &relation, relation_kind)?;
374        if lookup_mode == RelationLookupMode::Dynamic {
375            self.ensure_trigger_creation_privilege(&relation, relation_kind)?;
376        }
377        if relation_kind == "foreign table" && !definition.transition_relations.is_empty() {
378            return Err(SQLError::Routine {
379                sqlstate: "42809".into(),
380                message: format!("\"{}\" is a foreign table", relation.name),
381            });
382        }
383        if let Some(referenced_table) = definition.referenced_table.as_mut() {
384            let (referenced, referenced_kind) =
385                self.resolve_event_relation_kind(referenced_table, lookup_mode)?;
386            if referenced_kind != "table" {
387                return Err(SQLError::Routine {
388                    sqlstate: "42809".into(),
389                    message: format!("\"{}\" is a {referenced_kind}", referenced.name),
390                });
391            }
392            *referenced_table = referenced.qualified_name();
393        }
394        let requested_function = definition.function.clone();
395        let function = self.resolve_trigger_function_candidate(&requested_function, lookup_mode)?;
396        if lookup_mode == RelationLookupMode::Dynamic {
397            crate::routines::security::ensure_routine_execute_privilege_named(
398                self.authority,
399                &function.def,
400                &requested_function,
401            )?;
402        }
403        Self::validate_trigger_function(&function)?;
404        definition.function.clone_from(&function.def.name);
405        let columns = self.trigger_relation_columns(&relation, relation_kind)?;
406        if relation_kind == "table" {
407            self.validate_trigger_transition_relations(definition, &relation)?;
408        }
409        if definition.events.contains(&TriggerEvent::Truncate) && definition.row {
410            return Err(SQLError::Routine {
411                sqlstate: "0A000".into(),
412                message: "TRUNCATE FOR EACH ROW triggers are not supported".into(),
413            });
414        }
415        if !definition.update_columns.is_empty()
416            && !definition.events.contains(&TriggerEvent::Update)
417        {
418            return Err(SQLError::Routine {
419                sqlstate: "42601".into(),
420                message: "UPDATE OF columns may only be specified for an UPDATE trigger".into(),
421            });
422        }
423        let mut seen_update_columns = std::collections::BTreeSet::new();
424        for column in &definition.update_columns {
425            if !columns.iter().any(|definition| definition.name == *column) {
426                return Err(SQLError::UnknownColumn(format!(
427                    "{}.{column}",
428                    definition.table
429                )));
430            }
431            if !seen_update_columns.insert(column) {
432                return Err(SQLError::Routine {
433                    sqlstate: "42701".into(),
434                    message: format!("column \"{column}\" specified more than once"),
435                });
436            }
437        }
438        let mut condition_routine_bindings_changed = false;
439        if let Some(mut condition) = definition.when.take() {
440            condition_routine_bindings_changed =
441                self.validate_trigger_condition(definition, &columns, &mut condition)?;
442            definition.when = Some(condition);
443        }
444        Ok((relation, condition_routine_bindings_changed))
445    }
446
447    fn validate_trigger_transition_relations(
448        &self,
449        definition: &CreateTrigger,
450        relation: &RelationIdentity,
451    ) -> Result<(), SQLError> {
452        if definition.transition_relations.is_empty() {
453            return Ok(());
454        }
455        let hierarchy = self
456            .catalog
457            .loaded_table_hierarchy(relation)
458            .ok_or_else(|| {
459                SQLError::Internal(format!(
460                    "trigger table `{}` disappeared during validation",
461                    definition.table
462                ))
463            })?;
464        if definition.row && hierarchy.partition_spec.is_some() {
465            return Err(SQLError::Routine {
466                sqlstate: "0A000".into(),
467                message: format!("\"{}\" is a partitioned table", relation.name),
468            });
469        }
470        let mut old_table = None;
471        let mut new_table = None;
472        for transition in &definition.transition_relations {
473            validate_trigger_transition_relation(definition, &hierarchy, transition)?;
474            let duplicate = if transition.is_new {
475                new_table.replace(transition.name.as_str())
476            } else {
477                old_table.replace(transition.name.as_str())
478            };
479            if duplicate.is_some() {
480                return Err(SQLError::Routine {
481                    sqlstate: "42P17".into(),
482                    message: format!(
483                        "{} TABLE cannot be specified multiple times",
484                        if transition.is_new { "NEW" } else { "OLD" }
485                    ),
486                });
487            }
488        }
489        if old_table.is_some() && old_table == new_table {
490            return Err(SQLError::Routine {
491                sqlstate: "42P17".into(),
492                message: "OLD TABLE name and NEW TABLE name cannot be the same".into(),
493            });
494        }
495        Ok(())
496    }
497
498    fn validate_trigger_condition(
499        &self,
500        definition: &CreateTrigger,
501        columns: &[crate::ast::ColumnDef],
502        condition: &mut Expr,
503    ) -> Result<bool, SQLError> {
504        validate_trigger_condition_references(definition, columns, condition)?;
505        let bound = bind_expr(condition, &mut TriggerConditionTypeResolver { columns })?;
506        let lowered = crate::plan::ExpressionPlan::lower_with(bound, &|name: &str| {
507            self.routines.has_registered_aggregate_function(name)
508        });
509        let mut plan = lowered.clone();
510        let ty =
511            self.stored_routines
512                .bind_expression(&mut plan, &[], &crate::RowSchema::default())?;
513        // Bound identities and constants belong to the syntax as written, before a boolean coercion rewrites it.
514        let sites = crate::binding::syntax_sites::expression_syntax_sites(&lowered, &plan)?;
515        let changed = crate::catalog::stored_ast::bind_stored_expression_sites(condition, &sites)?;
516        if !ty.as_ref().is_some_and(is_boolean_type) {
517            if let Expr::Literal(value @ (Value::Str(_) | Value::FixedChar(_))) = condition {
518                *value = crate::expr::cast_value(value, "boolean")?;
519            } else if let Some(ty) = ty {
520                return Err(SQLError::TypeMismatch(format!(
521                    "argument of WHEN must be type boolean, not type {}",
522                    ty.sql_name()
523                )));
524            } else {
525                *condition = Expr::Cast {
526                    implicit: true,
527                    expr: Box::new(condition.clone()),
528                    ty: "boolean".into(),
529                };
530            }
531        }
532        crate::catalog::regrole_dependencies::reject_stored_regrole_constants_with(
533            self.regroles,
534            condition,
535            None,
536        )?;
537        Ok(changed)
538    }
539}
uqa_sql/catalog/events/definition/triggers.rs

uqa_sql/catalog/events/definition/
triggers.rs
