Line data Source code
1 : //! Catalog changes applied to descriptor entries.
2 : //!
3 : //! A full catalog refresh can be expensive, so the catalog follows the DDL statements that the user runs: their
4 : //! changes (`dashql_core_session::catalog_changes`) are applied to the tables of a descriptor entry.
5 :
6 : use super::{Catalog, CatalogEntryId, CatalogLayer};
7 : use crate::error::{Error, Result};
8 : use dashql_buffers::analyzer::{QualifiedTableNameT, TableColumnT, TableT};
9 : use dashql_buffers::catalog::{CatalogChangeT, CatalogChangeType, CatalogDescriptor, CatalogDescriptorT};
10 :
11 : /// What a statement did to a relation
12 : #[derive(Clone, Copy, Debug, PartialEq, Eq)]
13 : pub enum CatalogChangeKind {
14 : /// The statement created or replaced the relation
15 : CreateRelation,
16 : /// The statement dropped the relation
17 : DropRelation,
18 : }
19 :
20 : /// A change of a relation of the catalog, implied by a successful statement
21 : #[derive(Clone, Debug, PartialEq, Eq)]
22 : pub struct CatalogChange {
23 : /// What the statement did
24 : pub kind: CatalogChangeKind,
25 : /// The database name, empty if neither the statement nor the defaults name one
26 : pub database_name: String,
27 : /// The schema name, empty if neither the statement nor the defaults name one
28 : pub schema_name: String,
29 : /// The relation name
30 : pub relation_name: String,
31 : /// Is the relation only created if it does not exist yet?
32 : pub if_not_exists: bool,
33 : /// The column names of a created relation, in declaration order
34 : pub column_names: Vec<String>,
35 : }
36 :
37 : impl CatalogChange {
38 : /// The lookup key: (database, schema, relation)
39 55 : pub fn key(&self) -> (&str, &str, &str) {
40 55 : (&self.database_name, &self.schema_name, &self.relation_name)
41 55 : }
42 :
43 : /// Read a change of the boundary, None for a change without effect
44 2 : pub fn unpack(change: &CatalogChangeT) -> Option<Self> {
45 : Some(Self {
46 2 : kind: match change.change_type {
47 1 : CatalogChangeType::CREATE_RELATION => CatalogChangeKind::CreateRelation,
48 1 : CatalogChangeType::DROP_RELATION => CatalogChangeKind::DropRelation,
49 0 : _ => return None,
50 : },
51 2 : database_name: change.database_name.clone().unwrap_or_default(),
52 2 : schema_name: change.schema_name.clone().unwrap_or_default(),
53 2 : relation_name: change.relation_name.clone().unwrap_or_default(),
54 2 : if_not_exists: change.if_not_exists,
55 2 : column_names: change.column_names.clone().unwrap_or_default(),
56 : })
57 2 : }
58 :
59 : /// Write the change for the boundary
60 189 : pub fn pack(&self) -> CatalogChangeT {
61 189 : let mut change = CatalogChangeT::default();
62 189 : change.change_type = match self.kind {
63 143 : CatalogChangeKind::CreateRelation => CatalogChangeType::CREATE_RELATION,
64 46 : CatalogChangeKind::DropRelation => CatalogChangeType::DROP_RELATION,
65 : };
66 189 : change.database_name = Some(self.database_name.clone());
67 189 : change.schema_name = Some(self.schema_name.clone());
68 189 : change.relation_name = Some(self.relation_name.clone());
69 189 : change.if_not_exists = self.if_not_exists;
70 189 : if self.kind == CatalogChangeKind::CreateRelation {
71 143 : change.column_names = Some(self.column_names.clone());
72 143 : }
73 189 : change
74 189 : }
75 : }
76 :
77 : /// Apply changes to the tables of a descriptor
78 15 : fn apply_to_descriptor(descriptor: &mut CatalogDescriptorT, changes: &[CatalogChange]) {
79 15 : let tables = descriptor.tables.get_or_insert_with(Vec::new);
80 19 : for change in changes {
81 19 : let (database, schema, relation) = change.key();
82 22 : let position = tables.iter().position(|t| {
83 22 : let name = t.table_name.as_deref();
84 52 : let part = |f: fn(&QualifiedTableNameT) -> &Option<String>| name.and_then(|n| f(n).as_deref()).unwrap_or("");
85 22 : part(|n| &n.database_name) == database && part(|n| &n.schema_name) == schema && part(|n| &n.table_name) == relation
86 22 : });
87 19 : match change.kind {
88 : CatalogChangeKind::CreateRelation => {
89 14 : if change.if_not_exists && position.is_some() {
90 1 : continue;
91 13 : }
92 13 : let mut name = QualifiedTableNameT::default();
93 13 : name.database_name = Some(database.to_string());
94 13 : name.schema_name = Some(schema.to_string());
95 13 : name.table_name = Some(relation.to_string());
96 13 : let mut table = TableT::default();
97 13 : table.table_name = Some(Box::new(name));
98 : // A descriptor declares every column name once
99 13 : let mut seen = std::collections::HashSet::new();
100 13 : table.table_columns = Some(
101 13 : change
102 13 : .column_names
103 13 : .iter()
104 20 : .filter(|c| !c.is_empty() && seen.insert(c.as_str()))
105 19 : .map(|c| {
106 19 : let mut column = TableColumnT::default();
107 19 : column.column_name = Some(c.clone());
108 19 : column
109 19 : })
110 13 : .collect(),
111 : );
112 13 : match position {
113 2 : Some(position) => tables[position] = table,
114 11 : None => tables.push(table),
115 : }
116 : }
117 : CatalogChangeKind::DropRelation => {
118 5 : if let Some(position) = position {
119 3 : tables.remove(position);
120 4 : }
121 : }
122 : }
123 : }
124 15 : }
125 :
126 : impl Catalog {
127 : /// Apply catalog changes to a descriptor entry.
128 : /// An unknown entry is created with the given layer, a known entry keeps its layer.
129 15 : pub fn apply_catalog_changes(&mut self, id: CatalogEntryId, layer: CatalogLayer, changes: &[CatalogChange]) -> Result<()> {
130 15 : let (mut descriptor, layer) = match self.descriptor_entries.get(&id) {
131 10 : Some(entry) => {
132 10 : let descriptor = flatbuffers::root::<CatalogDescriptor>(&entry.bytes)
133 10 : .map_err(|_| Error::invalid_argument("invalid CatalogDescriptor FlatBuffer"))?
134 10 : .unpack();
135 10 : (descriptor, entry.layer)
136 : }
137 5 : None => (CatalogDescriptorT::default(), layer),
138 : };
139 15 : apply_to_descriptor(&mut descriptor, changes);
140 15 : let mut builder = flatbuffers::FlatBufferBuilder::new();
141 15 : let packed = descriptor.pack(&mut builder);
142 15 : builder.finish(packed, None);
143 15 : self.replace_descriptor(id, layer, builder.finished_data())
144 15 : }
145 : }
|
