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::{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(snapshot
27                .get(&relation)
28                .into_iter()
29                .flat_map(BTreeMap::values)
30                .filter(|rule| rule.definition.event == event)
31                .cloned()
32                .collect());
33        }
34        Ok(self
35            .registry
36            .read_rules()
37            .get(&relation)
38            .into_iter()
39            .flat_map(BTreeMap::values)
40            .filter(|rule| rule.definition.event == event)
41            .cloned()
42            .collect())
43    }
44
45    pub fn rules_for(&self, table: &str, event: RuleEvent) -> Result<Vec<StoredRule>, SQLError> {
46        let relation = self.analysis.resolve_rule_relation(table)?;
47        let replica = self.state.session_replication_role_is_replica();
48        Ok(self
49            .registry
50            .read_rules()
51            .get(&relation)
52            .into_iter()
53            .flat_map(BTreeMap::values)
54            .filter(|rule| {
55                (if replica {
56                    rule.enabled.fires_in_replica()
57                } else {
58                    rule.enabled.fires_in_origin()
59                }) && rule.definition.event == event
60            })
61            .cloned()
62            .collect())
63    }
64
65    pub fn relation_has_rules(&self, table: &str) -> Result<bool, SQLError> {
66        let relation = self.analysis.resolve_rule_relation(table)?;
67        Ok(self
68            .registry
69            .read_rules()
70            .get(&relation)
71            .is_some_and(|entries| !entries.is_empty()))
72    }
73
74    pub fn triggers_for(
75        &self,
76        table: &str,
77        timing: TriggerTiming,
78        event: TriggerEvent,
79        row: bool,
80        updated_columns: &[String],
81    ) -> Result<Vec<StoredTrigger>, SQLError> {
82        let relation = self.analysis.resolve_trigger_table(table)?;
83        let replica = self.state.session_replication_role_is_replica();
84        let relations = if row {
85            self.partition_trigger_sources(&relation.qualified_name())?
86        } else {
87            vec![relation.clone()]
88        };
89        let triggers = self.registry.read_triggers();
90        let mut candidates = BTreeMap::new();
91        for source in relations {
92            for trigger in triggers.get(&source).into_iter().flat_map(BTreeMap::values) {
93                let mut trigger = trigger.clone();
94                if source != relation {
95                    trigger.definition.table = relation.qualified_name();
96                }
97                candidates
98                    .entry(trigger.definition.name.clone())
99                    .or_insert(trigger);
100            }
101        }
102        Ok(candidates
103            .into_values()
104            .filter(|trigger| {
105                (if replica {
106                    trigger.enabled.fires_in_replica()
107                } else {
108                    trigger.enabled.fires_in_origin()
109                }) && trigger.definition.timing == timing
110                    && trigger.definition.row == row
111                    && trigger.definition.events.contains(&event)
112                    && (event != TriggerEvent::Update
113                        || trigger.definition.update_columns.is_empty()
114                        || trigger
115                            .definition
116                            .update_columns
117                            .iter()
118                            .any(|column| updated_columns.contains(column)))
119            })
120            .collect())
121    }
122
123    pub fn has_trigger_definition(
124        &self,
125        table: &str,
126        timing: TriggerTiming,
127        event: TriggerEvent,
128        row: bool,
129    ) -> Result<bool, SQLError> {
130        let relation = self.analysis.resolve_trigger_table(table)?;
131        let matches = |entries: &BTreeMap<String, StoredTrigger>| {
132            entries.values().any(|trigger| {
133                trigger.definition.timing == timing
134                    && trigger.definition.row == row
135                    && trigger.definition.events.contains(&event)
136            })
137        };
138        if let Some(snapshot) = self.state.query_triggers() {
139            return Ok(snapshot.get(&relation).is_some_and(matches));
140        }
141        Ok(self
142            .registry
143            .read_triggers()
144            .get(&relation)
145            .is_some_and(matches))
146    }
147
148    pub fn has_row_triggers(&self, table: &str, event: TriggerEvent) -> Result<bool, SQLError> {
149        let relation = self.analysis.resolve_trigger_table(table)?;
150        let sources = self.partition_trigger_sources(&relation.qualified_name())?;
151        let replica = self.state.session_replication_role_is_replica();
152        let triggers = self.registry.read_triggers();
153        Ok(sources.iter().any(|source| {
154            triggers.get(source).is_some_and(|entries| {
155                entries.values().any(|trigger| {
156                    (if replica {
157                        trigger.enabled.fires_in_replica()
158                    } else {
159                        trigger.enabled.fires_in_origin()
160                    }) && trigger.definition.row
161                        && trigger.definition.events.contains(&event)
162                })
163            })
164        }))
165    }
166
167    pub fn list_triggers(&self) -> Vec<StoredTrigger> {
168        if let Some(snapshot) = self.state.query_triggers() {
169            return snapshot
170                .values()
171                .flat_map(BTreeMap::values)
172                .cloned()
173                .collect();
174        }
175        self.registry
176            .read_triggers()
177            .values()
178            .flat_map(BTreeMap::values)
179            .cloned()
180            .collect()
181    }
182}
uqa_sql/catalog/events/definition/lookup/selections.rs

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