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

uqa_sql/catalog/events/definition/
lookup.rs
