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

uqa_sql/catalog/events/definition/
triggers.rs
