uqa_sql/catalog/events/definition/lookup/
selections.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Rule and trigger selection over pinned query catalogs and live execution registries.
8
9use std::collections::BTreeMap;
10
11use crate::ast::{RuleEvent, TriggerEvent, TriggerTiming};
12use crate::SQLError;
13
14use super::EventLookupContext;
15
16use crate::catalog::events::{selection, StoredRule, StoredTrigger};
17
18impl EventLookupContext<'_> {
19    pub fn rule_definitions_for(
20        &self,
21        table: &str,
22        event: RuleEvent,
23    ) -> Result<Vec<StoredRule>, SQLError> {
24        let relation = self.analysis.resolve_rule_relation(table)?;
25        if let Some(snapshot) = self.state.query_rules() {
26            return Ok(selection::rule_definitions(snapshot, &relation, event));
27        }
28        Ok(selection::rule_definitions(
29            &self.registry.read_rules(),
30            &relation,
31            event,
32        ))
33    }
34
35    pub fn rules_for(&self, table: &str, event: RuleEvent) -> Result<Vec<StoredRule>, SQLError> {
36        let relation = self.analysis.resolve_rule_relation(table)?;
37        let replica = self.state.session_replication_role_is_replica();
38        Ok(selection::active_rules(
39            &self.registry.read_rules(),
40            &relation,
41            event,
42            replica,
43        ))
44    }
45
46    pub fn relation_has_rules(&self, table: &str) -> Result<bool, SQLError> {
47        let relation = self.analysis.resolve_rule_relation(table)?;
48        Ok(self
49            .registry
50            .read_rules()
51            .get(&relation)
52            .is_some_and(|entries| !entries.is_empty()))
53    }
54
55    pub fn triggers_for(
56        &self,
57        table: &str,
58        timing: TriggerTiming,
59        event: TriggerEvent,
60        row: bool,
61        updated_columns: &[String],
62    ) -> Result<Vec<StoredTrigger>, SQLError> {
63        let relation = self.analysis.resolve_trigger_table(table)?;
64        let replica = self.state.session_replication_role_is_replica();
65        Ok(self
66            .matching_triggers(&relation, timing, event, row, updated_columns)?
67            .into_iter()
68            .filter(|trigger| {
69                if replica {
70                    trigger.enabled.fires_in_replica()
71                } else {
72                    trigger.enabled.fires_in_origin()
73                }
74            })
75            .collect())
76    }
77
78    /// The triggers of `table` for a timing, a level and an event, whatever replication role fires them. `PostgreSQL` copies a relation's trigger descriptor when a statement begins to write the relation and reads the replication role at each firing, so a statement takes its triggers from here once and applies the role itself.
79    pub fn trigger_definitions_for(
80        &self,
81        table: &str,
82        timing: TriggerTiming,
83        event: TriggerEvent,
84        row: bool,
85        updated_columns: &[String],
86    ) -> Result<Vec<StoredTrigger>, SQLError> {
87        let relation = self.analysis.resolve_trigger_table(table)?;
88        self.matching_triggers(&relation, timing, event, row, updated_columns)
89    }
90
91    fn matching_triggers(
92        &self,
93        relation: &uqa_core::RelationIdentity,
94        timing: TriggerTiming,
95        event: TriggerEvent,
96        row: bool,
97        updated_columns: &[String],
98    ) -> Result<Vec<StoredTrigger>, SQLError> {
99        let relations = if row {
100            self.partition_trigger_sources(&relation.qualified_name())?
101        } else {
102            vec![relation.clone()]
103        };
104        let triggers = self.registry.read_triggers();
105        let mut candidates = BTreeMap::new();
106        for source in relations {
107            for trigger in triggers.get(&source).into_iter().flat_map(BTreeMap::values) {
108                let mut trigger = trigger.clone();
109                if source != *relation {
110                    trigger.definition.table = relation.qualified_name();
111                }
112                candidates
113                    .entry(trigger.definition.name.clone())
114                    .or_insert(trigger);
115            }
116        }
117        Ok(candidates
118            .into_values()
119            .filter(|trigger| {
120                trigger.definition.timing == timing
121                    && trigger.definition.row == row
122                    && trigger.definition.events.contains(&event)
123                    && (event != TriggerEvent::Update
124                        || trigger.definition.update_columns.is_empty()
125                        || trigger
126                            .definition
127                            .update_columns
128                            .iter()
129                            .any(|column| updated_columns.contains(column)))
130            })
131            .collect())
132    }
133
134    pub fn has_trigger_definition(
135        &self,
136        table: &str,
137        timing: TriggerTiming,
138        event: TriggerEvent,
139        row: bool,
140    ) -> Result<bool, SQLError> {
141        let relation = self.analysis.resolve_trigger_table(table)?;
142        if let Some(snapshot) = self.state.query_triggers() {
143            return Ok(selection::has_trigger_definition(
144                snapshot, &relation, timing, event, row,
145            ));
146        }
147        Ok(selection::has_trigger_definition(
148            &self.registry.read_triggers(),
149            &relation,
150            timing,
151            event,
152            row,
153        ))
154    }
155
156    pub fn has_row_triggers(&self, table: &str, event: TriggerEvent) -> Result<bool, SQLError> {
157        let relation = self.analysis.resolve_trigger_table(table)?;
158        let sources = self.partition_trigger_sources(&relation.qualified_name())?;
159        let replica = self.state.session_replication_role_is_replica();
160        let triggers = self.registry.read_triggers();
161        Ok(sources.iter().any(|source| {
162            triggers.get(source).is_some_and(|entries| {
163                entries.values().any(|trigger| {
164                    (if replica {
165                        trigger.enabled.fires_in_replica()
166                    } else {
167                        trigger.enabled.fires_in_origin()
168                    }) && trigger.definition.row
169                        && trigger.definition.events.contains(&event)
170                })
171            })
172        }))
173    }
174
175    pub fn list_triggers(&self) -> Vec<StoredTrigger> {
176        if let Some(snapshot) = self.state.query_triggers() {
177            return snapshot
178                .values()
179                .flat_map(BTreeMap::values)
180                .cloned()
181                .collect();
182        }
183        self.registry
184            .read_triggers()
185            .values()
186            .flat_map(BTreeMap::values)
187            .cloned()
188            .collect()
189    }
190}
uqa_sql/catalog/events/definition/lookup/selections.rs

uqa_sql/catalog/events/definition/lookup/
selections.rs
