uqa_sql/catalog/events/definition/
lookup.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Trigger-name conflicts across current partition ancestry and descendants.
8use super::EventAnalysisContext;
9use crate::{
10    ast::TableHierarchy,
11    catalog::events::{
12        reads::{EventCatalogReads, EventLookupState},
13        StoredTrigger,
14    },
15    SQLError,
16};
17use std::collections::BTreeMap;
18use uqa_core::RelationIdentity;
19
20pub trait EventPartitionCatalog {
21    fn contains_loaded_table(&self, relation: &RelationIdentity) -> bool;
22    fn table_names(&self) -> Result<Vec<String>, String>;
23    fn try_table_hierarchy(&self, name: &str) -> Result<TableHierarchy, String>;
24}
25#[derive(Clone, Copy)]
26pub struct EventLookupContext<'a> {
27    pub analysis: EventAnalysisContext<'a>,
28    pub partitions: &'a dyn EventPartitionCatalog,
29    pub registry: &'a dyn EventCatalogReads,
30    pub state: &'a dyn EventLookupState,
31}
32impl EventLookupContext<'_> {
33    pub fn partition_trigger_sources(
34        &self,
35        table: &str,
36    ) -> Result<Vec<RelationIdentity>, SQLError> {
37        let mut current = self.analysis.resolve_trigger_table(table)?;
38        let mut sources = vec![current.clone()];
39        if !self.partitions.contains_loaded_table(&current) {
40            return Ok(sources);
41        }
42        loop {
43            let hierarchy = self
44                .partitions
45                .try_table_hierarchy(&current.qualified_name())
46                .map_err(|error| {
47                    SQLError::Internal(format!("read trigger partition hierarchy: {error}"))
48                })?;
49            if hierarchy.partition_bound.is_none() {
50                break;
51            }
52            let Some(parent) = hierarchy.parents.first() else {
53                return Err(SQLError::Internal(format!(
54                    "partition `{}` has no parent",
55                    current.qualified_name()
56                )));
57            };
58            current = RelationIdentity::from_legacy_name(parent).map_err(|error| {
59                SQLError::Internal(format!(
60                    "decode trigger partition parent `{parent}`: {error}"
61                ))
62            })?;
63            sources.push(current.clone());
64        }
65        Ok(sources)
66    }
67    pub fn ensure_partition_trigger_name_available(
68        &self,
69        relation: &RelationIdentity,
70        name: &str,
71        replacing_local: bool,
72    ) -> Result<(), SQLError> {
73        let ancestor_sources = self
74            .partition_trigger_sources(&relation.qualified_name())?
75            .into_iter()
76            .skip(1)
77            .collect::<Vec<_>>();
78        let mut descendant_relations = Vec::new();
79        for table in self
80            .partitions
81            .table_names()
82            .map_err(|error| SQLError::Internal(format!("read trigger partitions: {error}")))?
83        {
84            if table == relation.qualified_name() {
85                continue;
86            }
87            let sources = self.partition_trigger_sources(&table)?;
88            if sources.iter().skip(1).any(|source| source == relation) {
89                descendant_relations.push(RelationIdentity::from_legacy_name(&table).map_err(
90                    |error| {
91                        SQLError::Internal(format!("decode trigger partition `{table}`: {error}"))
92                    },
93                )?);
94            }
95        }
96        let triggers = self.registry.read_triggers();
97        for source in ancestor_sources {
98            if triggers
99                .get(&source)
100                .is_some_and(|entries| entries.contains_key(name))
101            {
102                return Err(super::duplicate_object(
103                    "trigger",
104                    name,
105                    &relation.qualified_name(),
106                ));
107            }
108        }
109        for descendant in descendant_relations {
110            if triggers
111                .get(&descendant)
112                .is_some_and(|entries| entries.contains_key(name))
113            {
114                return Err(super::duplicate_object(
115                    "trigger",
116                    name,
117                    &descendant.qualified_name(),
118                ));
119            }
120        }
121        if !replacing_local
122            && triggers
123                .get(relation)
124                .is_some_and(|entries| entries.contains_key(name))
125        {
126            return Err(super::duplicate_object(
127                "trigger",
128                name,
129                &relation.qualified_name(),
130            ));
131        }
132        Ok(())
133    }
134    pub fn rule_privilege_subject(&self, table: &str) -> Result<String, SQLError> {
135        let relation = self.analysis.resolve_rule_relation(table)?;
136        self.analysis
137            .catalog
138            .event_relation_owner(&relation)
139            .map(|(owner, _)| owner)
140    }
141    pub fn constraint_trigger_by_constraint_name(
142        &self,
143        table: &str,
144        name: &str,
145    ) -> Result<Option<StoredTrigger>, SQLError> {
146        let relation = self.analysis.resolve_trigger_table(table)?;
147        Ok(self
148            .registry
149            .read_triggers()
150            .get(&relation)
151            .into_iter()
152            .flat_map(BTreeMap::values)
153            .find(|trigger| {
154                trigger.definition.constraint
155                    && trigger
156                        .constraint_name
157                        .as_deref()
158                        .unwrap_or(&trigger.definition.name)
159                        == name
160            })
161            .cloned())
162    }
163}
164
165mod selections;
uqa_sql/catalog/events/definition/lookup.rs

uqa_sql/catalog/events/definition/
lookup.rs
