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, Some(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: Option<&[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: Option<&[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                        || updated_columns.is_none()
125                        || trigger.definition.update_columns.is_empty()
126                        || trigger.definition.update_columns.iter().any(|column| {
127                            updated_columns.is_some_and(|columns| columns.contains(column))
128                        }))
129            })
130            .collect())
131    }
132
133    pub fn has_trigger_definition(
134        &self,
135        table: &str,
136        timing: TriggerTiming,
137        event: TriggerEvent,
138        row: bool,
139    ) -> Result<bool, SQLError> {
140        let relation = self.analysis.resolve_trigger_table(table)?;
141        if let Some(snapshot) = self.state.query_triggers() {
142            return Ok(selection::has_trigger_definition(
143                snapshot, &relation, timing, event, row,
144            ));
145        }
146        Ok(selection::has_trigger_definition(
147            &self.registry.read_triggers(),
148            &relation,
149            timing,
150            event,
151            row,
152        ))
153    }
154
155    pub fn has_row_triggers(&self, table: &str, event: TriggerEvent) -> Result<bool, SQLError> {
156        let relation = self.analysis.resolve_trigger_table(table)?;
157        let sources = self.partition_trigger_sources(&relation.qualified_name())?;
158        let replica = self.state.session_replication_role_is_replica();
159        let triggers = self.registry.read_triggers();
160        Ok(sources.iter().any(|source| {
161            triggers.get(source).is_some_and(|entries| {
162                entries.values().any(|trigger| {
163                    (if replica {
164                        trigger.enabled.fires_in_replica()
165                    } else {
166                        trigger.enabled.fires_in_origin()
167                    }) && trigger.definition.row
168                        && trigger.definition.events.contains(&event)
169                })
170            })
171        }))
172    }
173
174    pub fn list_triggers(&self) -> Vec<StoredTrigger> {
175        if let Some(snapshot) = self.state.query_triggers() {
176            return snapshot
177                .values()
178                .flat_map(BTreeMap::values)
179                .cloned()
180                .collect();
181        }
182        self.registry
183            .read_triggers()
184            .values()
185            .flat_map(BTreeMap::values)
186            .cloned()
187            .collect()
188    }
189}
uqa_sql/catalog/events/definition/lookup/selections.rs

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