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
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 function.def.language != "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            composite_field: None,
201            resolution_error: None,
202        };
203        let candidates = self
204            .routines
205            .lookup_bound_sql_functions_by_binding(&binding)
206            .unwrap_or_default()
207            .into_iter()
208            .filter(|function| {
209                !function.def.is_procedure && routine_signature_types(&function.def).is_empty()
210            })
211            .collect::<Vec<_>>();
212        let function = match candidates.as_slice() {
213            [function] => function.clone(),
214            [] => {
215                return Err(SQLError::Routine {
216                    sqlstate: "42883".into(),
217                    message: format!("function {name}() does not exist"),
218                })
219            }
220            _ => {
221                return Err(SQLError::Internal(format!(
222                    "routine object identity for trigger function `{name}` is not unique"
223                )))
224            }
225        };
226        Self::validate_trigger_function(&function)?;
227        Ok(function)
228    }
229
230    fn ensure_trigger_creation_privilege(
231        &self,
232        relation: &RelationIdentity,
233        relation_kind: &str,
234    ) -> Result<(), SQLError> {
235        let canonical = relation.qualified_name();
236        match relation_kind {
237            "table" => {
238                let current_user = self.authority.current_role();
239                self.privileges.ensure_table_privilege_for(
240                    &canonical,
241                    &current_user,
242                    TableAclPrivilege::Trigger,
243                )
244            }
245            "view" => {
246                let view = self
247                    .catalog
248                    .restored_catalog_view_definition(&canonical)?
249                    .ok_or_else(|| {
250                        SQLError::Internal(format!(
251                            "resolved trigger view `{canonical}` has no catalog definition"
252                        ))
253                    })?;
254                self.privileges.ensure_view_privilege_for(
255                    &canonical,
256                    &view,
257                    &self.authority.current_role(),
258                    TableAclPrivilege::Trigger,
259                )
260            }
261            "foreign table" => self
262                .foreign_privileges
263                .ensure_foreign_table_privilege(&canonical, TableAclPrivilege::Trigger),
264            _ => Ok(()),
265        }
266    }
267
268    fn validate_trigger_relation_kind(
269        definition: &CreateTrigger,
270        relation: &RelationIdentity,
271        relation_kind: &str,
272    ) -> Result<(), SQLError> {
273        match relation_kind {
274            "table" if definition.timing == TriggerTiming::InsteadOf => Err(SQLError::Routine {
275                sqlstate: "42809".into(),
276                message: format!("\"{}\" is a table", relation.name),
277            }),
278            "view" if definition.timing == TriggerTiming::InsteadOf => {
279                if definition.events.contains(&TriggerEvent::Truncate)
280                    || !definition.transition_relations.is_empty()
281                {
282                    return Err(SQLError::Routine {
283                        sqlstate: "42809".into(),
284                        message: format!("\"{}\" is a view", relation.name),
285                    });
286                }
287                if !definition.row {
288                    return Err(SQLError::Routine {
289                        sqlstate: "0A000".into(),
290                        message: "INSTEAD OF triggers must be FOR EACH ROW".into(),
291                    });
292                }
293                if definition.when.is_some() {
294                    return Err(SQLError::Routine {
295                        sqlstate: "0A000".into(),
296                        message: "INSTEAD OF triggers cannot have WHEN conditions".into(),
297                    });
298                }
299                if !definition.update_columns.is_empty() {
300                    return Err(SQLError::Routine {
301                        sqlstate: "0A000".into(),
302                        message: "INSTEAD OF triggers cannot have column lists".into(),
303                    });
304                }
305                Ok(())
306            }
307            "view"
308                if definition.events.contains(&TriggerEvent::Truncate)
309                    || !definition.transition_relations.is_empty()
310                    || definition.row =>
311            {
312                Err(SQLError::Routine {
313                    sqlstate: "42809".into(),
314                    message: format!("\"{}\" is a view", relation.name),
315                })
316            }
317            "materialized view" => Err(SQLError::Routine {
318                sqlstate: "42809".into(),
319                message: format!("relation \"{}\" cannot have triggers", relation.name),
320            }),
321            "foreign table"
322                if definition.constraint || definition.timing == TriggerTiming::InsteadOf =>
323            {
324                Err(SQLError::Routine {
325                    sqlstate: "42809".into(),
326                    message: format!("\"{}\" is a foreign table", relation.name),
327                })
328            }
329            "view" | "foreign table" => Ok(()),
330            kind if kind != "table" => Err(SQLError::Routine {
331                sqlstate: "42809".into(),
332                message: format!("relation \"{}\" cannot have triggers", relation.name),
333            }),
334            _ => Ok(()),
335        }
336    }
337
338    pub fn validate_trigger_definition(
339        &self,
340        definition: &mut CreateTrigger,
341        lookup_mode: RelationLookupMode,
342    ) -> Result<(RelationIdentity, bool), SQLError> {
343        if definition.constraint && definition.or_replace {
344            return Err(SQLError::Routine {
345                sqlstate: "0A000".into(),
346                message: "CREATE OR REPLACE CONSTRAINT TRIGGER is not supported".into(),
347            });
348        }
349        if definition.constraint && (!definition.row || definition.timing != TriggerTiming::After) {
350            return Err(SQLError::Routine {
351                sqlstate: "0A000".into(),
352                message: "constraint triggers must be AFTER ROW triggers".into(),
353            });
354        }
355        if !definition.constraint
356            && (definition.deferrability.is_deferrable() || definition.referenced_table.is_some())
357        {
358            return Err(SQLError::Internal(
359                "ordinary trigger retained constraint-only metadata".into(),
360            ));
361        }
362        let (relation, relation_kind) =
363            self.resolve_trigger_relation_kind(&definition.table, lookup_mode)?;
364        definition.table = relation.qualified_name();
365        Self::validate_trigger_relation_kind(definition, &relation, relation_kind)?;
366        if lookup_mode == RelationLookupMode::Dynamic {
367            self.ensure_trigger_creation_privilege(&relation, relation_kind)?;
368        }
369        if relation_kind == "foreign table" && !definition.transition_relations.is_empty() {
370            return Err(SQLError::Routine {
371                sqlstate: "42809".into(),
372                message: format!("\"{}\" is a foreign table", relation.name),
373            });
374        }
375        if let Some(referenced_table) = definition.referenced_table.as_mut() {
376            let (referenced, referenced_kind) =
377                self.resolve_event_relation_kind(referenced_table, lookup_mode)?;
378            if referenced_kind != "table" {
379                return Err(SQLError::Routine {
380                    sqlstate: "42809".into(),
381                    message: format!("\"{}\" is a {referenced_kind}", referenced.name),
382                });
383            }
384            *referenced_table = referenced.qualified_name();
385        }
386        let requested_function = definition.function.clone();
387        let function = self.resolve_trigger_function_candidate(&requested_function, lookup_mode)?;
388        if lookup_mode == RelationLookupMode::Dynamic {
389            crate::routines::security::ensure_routine_execute_privilege_named(
390                self.authority,
391                &function.def,
392                &requested_function,
393            )?;
394        }
395        Self::validate_trigger_function(&function)?;
396        definition.function.clone_from(&function.def.name);
397        let columns = self.trigger_relation_columns(&relation, relation_kind)?;
398        if relation_kind == "table" {
399            self.validate_trigger_transition_relations(definition, &relation)?;
400        }
401        if definition.events.contains(&TriggerEvent::Truncate) && definition.row {
402            return Err(SQLError::Routine {
403                sqlstate: "0A000".into(),
404                message: "TRUNCATE FOR EACH ROW triggers are not supported".into(),
405            });
406        }
407        if !definition.update_columns.is_empty()
408            && !definition.events.contains(&TriggerEvent::Update)
409        {
410            return Err(SQLError::Routine {
411                sqlstate: "42601".into(),
412                message: "UPDATE OF columns may only be specified for an UPDATE trigger".into(),
413            });
414        }
415        let mut seen_update_columns = std::collections::BTreeSet::new();
416        for column in &definition.update_columns {
417            if !columns.iter().any(|definition| definition.name == *column) {
418                return Err(SQLError::UnknownColumn(format!(
419                    "{}.{column}",
420                    definition.table
421                )));
422            }
423            if !seen_update_columns.insert(column) {
424                return Err(SQLError::Routine {
425                    sqlstate: "42701".into(),
426                    message: format!("column \"{column}\" specified more than once"),
427                });
428            }
429        }
430        let mut condition_routine_bindings_changed = false;
431        if let Some(mut condition) = definition.when.take() {
432            condition_routine_bindings_changed =
433                self.validate_trigger_condition(definition, &columns, &mut condition)?;
434            definition.when = Some(condition);
435        }
436        Ok((relation, condition_routine_bindings_changed))
437    }
438
439    fn validate_trigger_transition_relations(
440        &self,
441        definition: &CreateTrigger,
442        relation: &RelationIdentity,
443    ) -> Result<(), SQLError> {
444        if definition.transition_relations.is_empty() {
445            return Ok(());
446        }
447        let hierarchy = self
448            .catalog
449            .loaded_table_hierarchy(relation)
450            .ok_or_else(|| {
451                SQLError::Internal(format!(
452                    "trigger table `{}` disappeared during validation",
453                    definition.table
454                ))
455            })?;
456        if definition.row && hierarchy.partition_spec.is_some() {
457            return Err(SQLError::Routine {
458                sqlstate: "0A000".into(),
459                message: format!("\"{}\" is a partitioned table", relation.name),
460            });
461        }
462        let mut old_table = None;
463        let mut new_table = None;
464        for transition in &definition.transition_relations {
465            validate_trigger_transition_relation(definition, &hierarchy, transition)?;
466            let duplicate = if transition.is_new {
467                new_table.replace(transition.name.as_str())
468            } else {
469                old_table.replace(transition.name.as_str())
470            };
471            if duplicate.is_some() {
472                return Err(SQLError::Routine {
473                    sqlstate: "42P17".into(),
474                    message: format!(
475                        "{} TABLE cannot be specified multiple times",
476                        if transition.is_new { "NEW" } else { "OLD" }
477                    ),
478                });
479            }
480        }
481        if old_table.is_some() && old_table == new_table {
482            return Err(SQLError::Routine {
483                sqlstate: "42P17".into(),
484                message: "OLD TABLE name and NEW TABLE name cannot be the same".into(),
485            });
486        }
487        Ok(())
488    }
489
490    fn validate_trigger_condition(
491        &self,
492        definition: &CreateTrigger,
493        columns: &[crate::ast::ColumnDef],
494        condition: &mut Expr,
495    ) -> Result<bool, SQLError> {
496        validate_trigger_condition_references(definition, columns, condition)?;
497        let bound = bind_expr(condition, &mut TriggerConditionTypeResolver { columns })?;
498        let lowered = crate::plan::ExpressionPlan::lower_with(bound, &|name: &str| {
499            self.routines.has_registered_aggregate_function(name)
500        });
501        let mut plan = lowered.clone();
502        let ty =
503            self.stored_routines
504                .bind_expression(&mut plan, &[], &crate::RowSchema::default())?;
505        // Bound identities and constants belong to the syntax as written, before a boolean coercion rewrites it.
506        let sites = crate::binding::syntax_sites::expression_syntax_sites(&lowered, &plan)?;
507        let changed = crate::catalog::stored_ast::bind_stored_expression_sites(condition, &sites)?;
508        if !ty.as_ref().is_some_and(is_boolean_type) {
509            if let Expr::Literal(value @ (Value::Str(_) | Value::FixedChar(_))) = condition {
510                *value = crate::expr::cast_value(value, "boolean")?;
511            } else if let Some(ty) = ty {
512                return Err(SQLError::TypeMismatch(format!(
513                    "argument of WHEN must be type boolean, not type {}",
514                    ty.sql_name()
515                )));
516            } else {
517                *condition = Expr::Cast {
518                    implicit: true,
519                    expr: Box::new(condition.clone()),
520                    ty: "boolean".into(),
521                };
522            }
523        }
524        crate::catalog::regrole_dependencies::reject_stored_regrole_constants_with(
525            self.regroles,
526            condition,
527            None,
528        )?;
529        Ok(changed)
530    }
531}
532
533fn trigger_column(name: String, ty: ColumnType) -> ColumnDef {
534    ColumnDef {
535        name,
536        ty,
537        object_id: None,
538        attribute_number: None,
539        missing_value: None,
540        primary_key: false,
541        not_null: false,
542        not_null_explicit: false,
543        not_null_name: None,
544        not_null_identity: None,
545        not_null_validated: true,
546        not_null_no_inherit: false,
547        not_null_is_local: true,
548        auto_increment: None,
549        unique: false,
550        default: None,
551        generated: None,
552        check: None,
553        check_name: None,
554        check_enforced: true,
555        check_validated: true,
556        check_no_inherit: false,
557        check_is_local: true,
558        check_object_id: None,
559        check_catalog_oid: None,
560        default_catalog_oid: None,
561        references: None,
562    }
563}
uqa_sql/catalog/events/definition/triggers.rs

uqa_sql/catalog/events/definition/
triggers.rs
