Line data Source code
1 : //! Catalog changes implied by successful statements.
2 : //!
3 : //! A full catalog refresh can be expensive, so the catalog follows the DDL statements that the user
4 : //! runs: a successful `CREATE TABLE`, `CREATE TABLE AS`, `CREATE VIEW`, `SELECT INTO` or `DROP`
5 : //! becomes a [`CatalogChange`] that is applied to a descriptor entry
6 : //! (`dashql_core::catalog::Catalog::apply_catalog_changes`). The changes of a connection are kept as
7 : //! SQL, a script of `CREATE TABLE` and `DROP TABLE` statements that is rendered and parsed with the
8 : //! same derivation.
9 :
10 : use dashql_buffers::parser::{AttributeKey, Node, NodeType, StatementType, TempType};
11 : use dashql_core::analyzer::AnalyzedScript;
12 : use dashql_core::catalog::changes::{CatalogChange, CatalogChangeKind};
13 : use dashql_core::catalog::Catalog;
14 : use dashql_core::error::{Error, Result};
15 : use dashql_core::pack::pack_parsed_statement;
16 : use dashql_core::script::Script;
17 : use dashql_parser::parser::ParsedScript;
18 :
19 : /// The header of rendered changes SQL
20 : const CHANGES_SQL_HEADER: &str = "-- Catalog changes of executed statements";
21 :
22 : /// How statements are turned into catalog changes
23 : #[derive(Clone, Debug, Default)]
24 : pub struct CatalogChangeOptions {
25 : /// Follow temporary relations? They only live as long as the database session.
26 : pub include_temporary: bool,
27 : }
28 :
29 : /// The names that qualify the relations of unqualified statements, they depend on the connection
30 : #[derive(Clone, Debug, Default)]
31 : pub struct CatalogNameDefaults {
32 : /// The database of relations without a database name
33 : pub database_name: String,
34 : /// The schema of relations without a schema name
35 : pub schema_name: String,
36 : }
37 :
38 : /// The child of a node with an attribute key
39 1717 : fn attribute<'a>(parsed: &'a ParsedScript, node: &Node, key: AttributeKey) -> Option<&'a Node> {
40 3420 : parsed.children(node).iter().find(|c| c.attribute_key() == key)
41 1717 : }
42 :
43 : /// Is an attribute a `true` bool node?
44 839 : fn is_true(node: Option<&Node>) -> bool {
45 839 : node.is_some_and(|n| n.node_type() == NodeType::BOOL && n.children_begin_or_value() != 0)
46 839 : }
47 :
48 : /// Does an attribute declare a temporary relation? `UNLOGGED` relations are persistent.
49 867 : fn is_temporary(node: Option<&Node>) -> bool {
50 867 : node.is_some_and(|n| {
51 19 : n.node_type() == NodeType::ENUM_SQL_TEMP_TYPE
52 19 : && matches!(TempType(n.children_begin_or_value() as u8), TempType::DEFAULT | TempType::LOCAL | TempType::GLOBAL)
53 19 : })
54 867 : }
55 :
56 : /// The change of a statement, before the relation is named
57 : struct StatementChange {
58 : /// What the statement does
59 : kind: CatalogChangeKind,
60 : /// Is the relation only created if it does not exist yet?
61 : if_not_exists: bool,
62 : }
63 :
64 : /// Classify a statement
65 2179 : fn classify_statement(parsed: &ParsedScript, statement_id: usize, include_temporary: bool) -> Option<StatementChange> {
66 2179 : let statement = &parsed.statements[statement_id];
67 2179 : let root = &parsed.nodes[statement.root as usize];
68 2179 : let create = |temp: AttributeKey, if_not_exists: Option<AttributeKey>| {
69 870 : if !include_temporary && is_temporary(attribute(parsed, root, temp)) {
70 8 : return None;
71 862 : }
72 : Some(StatementChange {
73 862 : kind: CatalogChangeKind::CreateRelation,
74 862 : if_not_exists: if_not_exists.is_some_and(|key| is_true(attribute(parsed, root, key))),
75 : })
76 870 : };
77 2179 : match statement.statement_type {
78 810 : StatementType::CREATE_TABLE => create(AttributeKey::SQL_CREATE_TABLE_TEMP, Some(AttributeKey::SQL_CREATE_TABLE_IF_NOT_EXISTS)),
79 35 : StatementType::CREATE_TABLE_AS => create(AttributeKey::SQL_CREATE_AS_TEMP, Some(AttributeKey::SQL_CREATE_AS_IF_NOT_EXISTS)),
80 25 : StatementType::CREATE_VIEW => create(AttributeKey::SQL_VIEW_TEMP, None),
81 : StatementType::SELECT_INTO => {
82 11 : let into = attribute(parsed, root, AttributeKey::SQL_SELECT_INTO)?;
83 11 : if !include_temporary && is_temporary(attribute(parsed, into, AttributeKey::SQL_TEMP_TYPE)) {
84 2 : return None;
85 9 : }
86 9 : Some(StatementChange {
87 9 : kind: CatalogChangeKind::CreateRelation,
88 9 : if_not_exists: false,
89 9 : })
90 : }
91 56 : StatementType::DROP_TABLE | StatementType::DROP_VIEW => Some(StatementChange {
92 56 : kind: CatalogChangeKind::DropRelation,
93 56 : if_not_exists: false,
94 56 : }),
95 1242 : _ => None,
96 : }
97 2179 : }
98 :
99 : /// Derive the catalog change of a statement of an analyzed script.
100 : ///
101 : /// The relation is named as written, see [`qualify_catalog_changes`]. Returns None for statements that
102 : /// leave the catalog as it is, for excluded temporary relations, and for statements whose relation the
103 : /// analysis could not name.
104 2179 : pub fn derive_catalog_change(analyzed: &AnalyzedScript, statement_id: usize, options: &CatalogChangeOptions) -> Option<CatalogChange> {
105 2179 : let parsed = &analyzed.parsed;
106 2179 : if statement_id >= parsed.statements.len() {
107 0 : return None;
108 2179 : }
109 2179 : let classified = classify_statement(parsed, statement_id, options.include_temporary)?;
110 927 : let target = pack_parsed_statement(parsed, statement_id).target?;
111 927 : let relation_name = target.relation_name.filter(|n| !n.is_empty())?;
112 927 : let mut change = CatalogChange {
113 927 : kind: classified.kind,
114 927 : database_name: target.database_name.unwrap_or_default(),
115 927 : schema_name: target.schema_name.unwrap_or_default(),
116 927 : relation_name,
117 927 : if_not_exists: classified.if_not_exists,
118 927 : column_names: Vec::new(),
119 927 : };
120 927 : if classified.kind == CatalogChangeKind::CreateRelation {
121 871 : let table = analyzed
122 871 : .entry
123 871 : .table_declarations
124 871 : .iter()
125 2059 : .find(|t| t.ast_statement_id == Some(statement_id as u32))?;
126 : // The analyzer sorts the columns of CREATE TABLE by name, the AST nodes restore the declaration order
127 871 : let mut columns: Vec<_> = table.table_columns.iter().collect();
128 9843 : columns.sort_by_key(|c| c.ast_node_id.unwrap_or(u32::MAX));
129 2682 : change.column_names = columns.iter().map(|c| c.column_name.text.to_string()).collect();
130 56 : }
131 927 : Some(change)
132 2179 : }
133 :
134 : /// Fill the missing database and schema names of changes with the defaults
135 25 : pub fn qualify_catalog_changes(changes: &mut [CatalogChange], defaults: &CatalogNameDefaults) {
136 26 : for change in changes {
137 52 : for (name, default) in [
138 26 : (&mut change.database_name, &defaults.database_name),
139 26 : (&mut change.schema_name, &defaults.schema_name),
140 26 : ] {
141 52 : if name.is_empty() {
142 40 : name.clone_from(default);
143 40 : }
144 : }
145 : }
146 25 : }
147 :
148 : /// Compact changes to the last change of every relation, in the order of the last changes.
149 : ///
150 : /// `IF NOT EXISTS` is resolved against the earlier changes of the same relation: after a creation it
151 : /// changes nothing, after a drop it creates the relation unconditionally.
152 3 : pub fn compact_catalog_changes(changes: Vec<CatalogChange>) -> Vec<CatalogChange> {
153 3 : let mut compacted: Vec<CatalogChange> = Vec::new();
154 12 : for mut change in changes {
155 14 : let previous = compacted.iter().position(|c| c.key() == change.key());
156 12 : if let Some(previous) = previous {
157 4 : if change.if_not_exists {
158 2 : if compacted[previous].kind == CatalogChangeKind::CreateRelation {
159 1 : continue;
160 1 : }
161 1 : change.if_not_exists = false;
162 2 : }
163 3 : compacted.remove(previous);
164 8 : }
165 11 : compacted.push(change);
166 : }
167 3 : compacted
168 3 : }
169 :
170 : /// Quote an identifier
171 31 : fn quote_identifier(identifier: &str, out: &mut String) {
172 31 : out.push('"');
173 31 : out.push_str(&identifier.replace('"', "\"\""));
174 31 : out.push('"');
175 31 : }
176 :
177 : /// Write the qualified name of a change
178 8 : fn write_qualified_name(change: &CatalogChange, out: &mut String) {
179 8 : let (database, schema, relation) = change.key();
180 8 : if !database.is_empty() {
181 8 : quote_identifier(database, out);
182 8 : out.push('.');
183 8 : }
184 8 : if !schema.is_empty() || !database.is_empty() {
185 8 : quote_identifier(schema, out);
186 8 : out.push('.');
187 8 : }
188 8 : quote_identifier(relation, out);
189 8 : }
190 :
191 : /// Render compacted changes as SQL
192 3 : pub fn render_catalog_changes_sql(changes: &[CatalogChange]) -> String {
193 3 : let mut out = String::from(CHANGES_SQL_HEADER);
194 8 : for change in changes {
195 8 : match change.kind {
196 : CatalogChangeKind::CreateRelation => {
197 6 : out.push_str(if change.if_not_exists { "\n\nCREATE TABLE IF NOT EXISTS " } else { "\n\nCREATE TABLE " });
198 6 : write_qualified_name(change, &mut out);
199 6 : out.push_str(" (");
200 7 : for (i, column) in change.column_names.iter().enumerate() {
201 7 : out.push_str(if i == 0 { "\n " } else { ",\n " });
202 7 : quote_identifier(column, &mut out);
203 7 : out.push_str(" VARCHAR");
204 : }
205 6 : out.push_str(if change.column_names.is_empty() { ");" } else { "\n);" });
206 : }
207 2 : CatalogChangeKind::DropRelation => {
208 2 : out.push_str("\n\nDROP TABLE ");
209 2 : write_qualified_name(change, &mut out);
210 2 : out.push(';');
211 2 : }
212 : }
213 : }
214 3 : out.push('\n');
215 3 : out
216 3 : }
217 :
218 : /// Parse changes SQL into its changes. Its statements are fully qualified.
219 5 : pub fn parse_catalog_changes_sql(text: &str) -> Result<Vec<CatalogChange>> {
220 5 : let catalog = Catalog::new();
221 5 : let mut script = Script::new(&catalog);
222 5 : script.replace_text(text);
223 5 : script.analyze(&catalog, true)?;
224 5 : let analyzed = script.analyzed.clone().ok_or_else(|| Error::invalid_argument("catalog changes SQL is not analyzed"))?;
225 5 : if !analyzed.parsed.errors.is_empty() || !analyzed.parsed.scanned.errors.is_empty() {
226 1 : return Err(Error::invalid_argument("invalid catalog changes SQL"));
227 4 : }
228 4 : let options = CatalogChangeOptions::default();
229 4 : Ok((0..analyzed.parsed.statements.len())
230 8 : .filter_map(|id| derive_catalog_change(&analyzed, id, &options))
231 4 : .collect())
232 5 : }
233 :
234 : /// Append changes to changes SQL and compact it
235 2 : pub fn append_catalog_changes_sql(changes_sql: &str, changes: &[CatalogChange]) -> Result<String> {
236 2 : let mut all = parse_catalog_changes_sql(changes_sql)?;
237 2 : all.extend(changes.iter().cloned());
238 2 : Ok(render_catalog_changes_sql(&compact_catalog_changes(all)))
239 2 : }
|
