Line data Source code
1 : //! Catalog entries backed by descriptor bytes.
2 :
3 : use super::entry::*;
4 : use super::ids::*;
5 : use super::Catalog;
6 : use crate::error::{Error, Result};
7 : use crate::utils::quote_identifier;
8 : use dashql_buffers::analyzer::NameTag;
9 : use dashql_buffers::catalog::CatalogDescriptor;
10 : use std::collections::BTreeSet;
11 : use dashql_parser::span::TextSpan;
12 : use std::sync::Arc;
13 :
14 : /// A name of a descriptor entry
15 : #[derive(Clone, Debug)]
16 : pub struct DescriptorName {
17 : /// The name text
18 : pub text: Arc<str>,
19 : /// The name tags
20 : pub tags: u8,
21 : /// How often the name is registered by the descriptor
22 : pub occurrences: u32,
23 : }
24 :
25 : /// Declarations imported from verified `CatalogDescriptor` bytes
26 : #[derive(Debug)]
27 : pub struct DescriptorEntry {
28 : /// The declarations of the entry, shared with the analyzed scripts that depend on it
29 : pub entry: Arc<CatalogEntryData>,
30 : /// The layer of the entry
31 : pub layer: CatalogLayer,
32 : /// The names in registration order
33 : pub names: Vec<DescriptorName>,
34 : /// The objects the names resolve to, in registration order
35 : pub name_objects: Vec<(NameId, CatalogObjectRef)>,
36 : /// The descriptor bytes
37 : pub bytes: Vec<u8>,
38 : /// The name search index, built on first use
39 : name_search_index: std::sync::OnceLock<super::NameSearchIndex>,
40 : }
41 :
42 19 : fn invalid(message: &str) -> Error {
43 19 : Error::invalid_argument(message)
44 19 : }
45 :
46 : /// Render a qualified name, skipping its empty leading parts
47 4 : fn qualified_name(parts: &[&str]) -> String {
48 7 : let first = parts.iter().position(|p| !p.is_empty()).unwrap_or(parts.len() - 1);
49 9 : parts[first..].iter().map(|p| quote_identifier(p)).collect::<Vec<_>>().join(".")
50 4 : }
51 :
52 : /// Validate a name or type text
53 438 : fn validate_text(text: Option<&str>, required: bool) -> Result<()> {
54 438 : let value = text.unwrap_or("");
55 438 : if (required && value.is_empty()) || value.contains('\0') {
56 9 : return Err(invalid("invalid catalog descriptor name or type"));
57 429 : }
58 429 : Ok(())
59 438 : }
60 :
61 : /// Verify descriptor bytes
62 75 : pub fn validate_descriptor(data: &[u8]) -> Result<CatalogDescriptor<'_>> {
63 75 : if data.is_empty() {
64 2 : return Err(invalid("invalid CatalogDescriptor FlatBuffer"));
65 73 : }
66 71 : let descriptor =
67 73 : flatbuffers::root::<CatalogDescriptor>(data).map_err(|_| invalid("invalid CatalogDescriptor FlatBuffer"))?;
68 71 : let mut table_names = BTreeSet::new();
69 71 : if let Some(tables) = descriptor.tables() {
70 88 : for table in tables {
71 88 : let name = table.table_name().ok_or_else(|| invalid("missing descriptor table name"))?;
72 87 : validate_text(name.database_name(), false)?;
73 86 : validate_text(name.schema_name(), false)?;
74 85 : validate_text(name.table_name(), true)?;
75 82 : let key = (
76 82 : name.database_name().unwrap_or(""),
77 82 : name.schema_name().unwrap_or(""),
78 82 : name.table_name().unwrap_or(""),
79 82 : );
80 82 : if !table_names.insert(key) {
81 1 : let table = qualified_name(&[key.0, key.1, key.2]);
82 1 : return Err(invalid(&format!("duplicate descriptor table name {table}")));
83 81 : }
84 81 : let mut columns = BTreeSet::new();
85 81 : if let Some(values) = table.table_columns() {
86 113 : for column in values {
87 113 : validate_text(column.column_name(), true)?;
88 112 : let column_name = column.column_name().unwrap_or("");
89 112 : if !columns.insert(column_name) {
90 2 : let table = qualified_name(&[key.0, key.1, key.2]);
91 2 : let column = quote_identifier(column_name);
92 2 : return Err(invalid(&format!("duplicate descriptor column name {column} in table {table}")));
93 110 : }
94 : }
95 0 : }
96 : }
97 0 : }
98 61 : if let Some(functions) = descriptor.function_declarations() {
99 46 : for function in functions {
100 12 : let function_name =
101 13 : function.function_name().ok_or_else(|| invalid("missing descriptor function name"))?;
102 12 : validate_text(function_name.database_name(), false)?;
103 12 : validate_text(function_name.schema_name(), false)?;
104 12 : validate_text(function_name.function_name(), true)?;
105 11 : validate_text(function.return_type(), true)?;
106 10 : let mut params = BTreeSet::new();
107 10 : if let Some(values) = function.params() {
108 10 : for param in values {
109 10 : validate_text(param.param_name(), false)?;
110 10 : validate_text(param.param_type(), true)?;
111 9 : let name = param.param_name().unwrap_or("");
112 9 : if !name.is_empty() && !params.insert(name) {
113 1 : let qualified = qualified_name(&[
114 1 : function_name.database_name().unwrap_or(""),
115 1 : function_name.schema_name().unwrap_or(""),
116 1 : function_name.function_name().unwrap_or(""),
117 1 : ]);
118 1 : let param = quote_identifier(name);
119 1 : return Err(invalid(&format!("duplicate descriptor parameter name {param} in function {qualified}")));
120 8 : }
121 : }
122 0 : }
123 : }
124 15 : }
125 56 : Ok(descriptor)
126 75 : }
127 :
128 : impl DescriptorEntry {
129 : /// The name search index
130 4 : pub fn name_search_index(&self) -> &super::NameSearchIndex {
131 4 : self.name_search_index.get_or_init(|| {
132 3 : let mut names: Vec<super::IndexedName> = self
133 3 : .names
134 3 : .iter()
135 3 : .map(|n| super::IndexedName {
136 10 : text: n.text.clone(),
137 10 : occurrences: n.occurrences,
138 10 : location: Default::default(),
139 10 : tags: NameTag(n.tags),
140 10 : objects: Vec::new(),
141 10 : })
142 3 : .collect();
143 10 : for &(name, object) in &self.name_objects {
144 10 : names[name as usize].objects.push(object);
145 10 : }
146 3 : super::NameSearchIndex::new(names)
147 3 : })
148 4 : }
149 :
150 : /// Import descriptor bytes
151 47 : pub fn new(catalog: &Catalog, id: CatalogEntryId, layer: CatalogLayer, bytes: &[u8]) -> Result<Self> {
152 47 : let descriptor = validate_descriptor(bytes)?;
153 44 : let mut out = DescriptorEntry {
154 44 : entry: Arc::new(CatalogEntryData::new(id, catalog.version())),
155 44 : layer,
156 44 : names: Vec::new(),
157 44 : name_objects: Vec::new(),
158 44 : bytes: bytes.to_vec(),
159 44 : name_search_index: Default::default(),
160 44 : };
161 : // The new entry is published with the next catalog version, as are its tables
162 44 : Arc::make_mut(&mut out.entry).catalog_version = catalog.version() + 1;
163 44 : let table_version = catalog.version() + 1;
164 44 : let mut names_by_text = std::collections::HashMap::<Arc<str>, NameId>::new();
165 293 : let mut register = |out: &mut DescriptorEntry, text: &str, tag: NameTag| -> Name {
166 293 : let id = match names_by_text.get(text) {
167 56 : Some(&id) => {
168 56 : let name = &mut out.names[id as usize];
169 56 : name.tags |= tag.0;
170 56 : name.occurrences += 1;
171 56 : id
172 : }
173 : None => {
174 237 : let id = out.names.len() as NameId;
175 237 : let text: Arc<str> = Arc::from(text);
176 237 : out.names.push(DescriptorName {
177 237 : text: text.clone(),
178 237 : tags: tag.0,
179 237 : occurrences: 1,
180 237 : });
181 237 : names_by_text.insert(text, id);
182 237 : id
183 : }
184 : };
185 293 : Name {
186 293 : entry: out.entry.catalog_entry_id,
187 293 : id,
188 293 : text: out.names[id as usize].text.clone(),
189 293 : }
190 293 : };
191 67 : let register_schema = |out: &mut DescriptorEntry, database: &Name, schema: &Name| -> QualifiedCatalogObjectId {
192 67 : let entry = Arc::make_mut(&mut out.entry);
193 67 : let db_id = match entry.databases_by_name.get(&database.text) {
194 22 : Some(&i) => entry.database_references[i].object_id,
195 : None => {
196 45 : let db_id = catalog.reserve_database_id(&database.text);
197 45 : entry.database_references.push(DatabaseReference {
198 45 : object_id: db_id,
199 45 : database_name: database.text.clone(),
200 45 : database_alias: Arc::from(""),
201 45 : });
202 45 : let index = entry.database_references.len() - 1;
203 45 : entry.databases_by_name.insert(database.text.clone(), index);
204 45 : out.name_objects.push((database.id, CatalogObjectRef::Database(ObjectLoc { entry: id, index: index as u32 })));
205 45 : db_id
206 : }
207 : };
208 67 : let key = (database.text.clone(), schema.text.clone());
209 67 : if let Some(&i) = entry.schemas_by_qualified_name.get(&key) {
210 13 : return entry.schema_references[i].object_id;
211 54 : }
212 54 : let schema_id = catalog.reserve_schema_id(&database.text, &schema.text, db_id);
213 54 : entry.schema_references.push(SchemaReference {
214 54 : object_id: schema_id,
215 54 : database_name: database.text.clone(),
216 54 : schema_name: schema.text.clone(),
217 54 : });
218 54 : let index = entry.schema_references.len() - 1;
219 54 : entry.schemas_by_qualified_name.insert(key, index);
220 54 : out.name_objects.push((schema.id, CatalogObjectRef::Schema(ObjectLoc { entry: id, index: index as u32 })));
221 54 : schema_id
222 67 : };
223 :
224 : // The ids are keyed by name and occurrence, so a table keeps its id when the descriptor of the
225 : // entry is replaced with one that declares it at another position
226 44 : let mut table_occurrences: std::collections::HashMap<QualifiedKey, u32> = Default::default();
227 44 : let mut function_occurrences: std::collections::HashMap<QualifiedKey, u32> = Default::default();
228 67 : let next_occurrence = |occurrences: &mut std::collections::HashMap<QualifiedKey, u32>, key: &QualifiedKey| {
229 67 : let occurrence = occurrences.entry(key.clone()).or_insert(0);
230 67 : *occurrence += 1;
231 67 : *occurrence - 1
232 67 : };
233 44 : if let Some(tables) = descriptor.tables() {
234 63 : for source in tables {
235 63 : let name = source.table_name().unwrap();
236 63 : let db = register(&mut out, name.database_name().unwrap_or(""), NameTag::DATABASE_NAME);
237 63 : let schema = register(&mut out, name.schema_name().unwrap_or(""), NameTag::SCHEMA_NAME);
238 63 : let table_name = register(&mut out, name.table_name().unwrap_or(""), NameTag::TABLE_NAME);
239 63 : let schema_id = register_schema(&mut out, &db, &schema);
240 63 : let table_index = out.entry.table_declarations.len();
241 63 : let qualified_name = QualifiedTableName {
242 63 : ast_node_id: None,
243 63 : database_name: db.clone(),
244 63 : schema_name: schema.clone(),
245 63 : table_name: table_name.clone(),
246 63 : };
247 63 : let key = qualified_name.key();
248 63 : let table_id = catalog.reserve_table_id(id, &key, next_occurrence(&mut table_occurrences, &key));
249 63 : let mut table = TableDeclaration::new(schema_id, table_id, qualified_name);
250 63 : let mut column_occurrences: std::collections::HashMap<Arc<str>, u32> = Default::default();
251 63 : table.catalog_version = table_version;
252 63 : table.location = source.location().map(|l| TextSpan::new(l.offset(), l.length()));
253 63 : let table_loc = TableLoc {
254 63 : entry: id,
255 63 : index: table_index as u32,
256 63 : };
257 63 : out.name_objects.push((table_name.id, CatalogObjectRef::Table(table_loc)));
258 63 : if let Some(columns) = source.table_columns() {
259 89 : for source_column in columns {
260 89 : let column_name =
261 89 : register(&mut out, source_column.column_name().unwrap_or(""), NameTag::COLUMN_NAME);
262 89 : let ordinal = table.table_columns.len() as u32;
263 89 : let column_loc = ColumnLoc {
264 89 : table: table_loc,
265 89 : column: ordinal,
266 89 : };
267 89 : out.name_objects.push((column_name.id, CatalogObjectRef::Column(column_loc)));
268 89 : table.table_columns_by_name.entry(column_name.text.clone()).or_insert(ordinal as usize);
269 89 : Arc::make_mut(&mut out.entry).table_columns_by_name.entry(column_name.text.clone()).or_default().push(ColumnLoc {
270 89 : table: TableLoc {
271 89 : entry: id,
272 89 : index: table_index as u32,
273 89 : },
274 89 : column: ordinal,
275 89 : });
276 89 : table.table_columns.push(TableColumn {
277 89 : object_id: QualifiedCatalogObjectId::table_column(table_id, {
278 89 : let occurrence = column_occurrences.entry(column_name.text.clone()).or_insert(0);
279 89 : *occurrence += 1;
280 89 : catalog.reserve_column_id(table_id, &column_name.text, *occurrence - 1)
281 : }),
282 89 : table: Some(table_id),
283 89 : ast_node_id: None,
284 89 : column_name,
285 89 : location: source_column.location().map(|l| TextSpan::new(l.offset(), l.length())),
286 : });
287 : }
288 0 : }
289 63 : let entry = Arc::make_mut(&mut out.entry);
290 63 : entry.tables_by_id.insert(table_id.object, table_index);
291 63 : entry.tables_by_qualified_name.entry(table.table_name.key()).or_insert(table_index);
292 63 : entry.tables_by_unqualified_name.entry(table_name.text.clone()).or_default().push(table_index);
293 63 : entry
294 63 : .tables_by_unqualified_schema
295 63 : .entry((schema.text.clone(), db.text.clone()))
296 63 : .or_default()
297 63 : .push(table_index);
298 63 : entry.table_declarations.push(table);
299 : }
300 0 : }
301 44 : if let Some(functions) = descriptor.function_declarations() {
302 29 : for source in functions {
303 4 : let name = source.function_name().unwrap();
304 4 : let db = register(&mut out, name.database_name().unwrap_or(""), NameTag::DATABASE_NAME);
305 4 : let schema = register(&mut out, name.schema_name().unwrap_or(""), NameTag::SCHEMA_NAME);
306 4 : let function_name = register(&mut out, name.function_name().unwrap_or(""), NameTag::FUNCTION_NAME);
307 4 : let schema_id = register_schema(&mut out, &db, &schema);
308 4 : let function_index = out.entry.function_declarations.len();
309 4 : let key = (db.text.clone(), schema.text.clone(), function_name.text.clone());
310 4 : let function_id = catalog.reserve_function_id(id, &key, next_occurrence(&mut function_occurrences, &key));
311 4 : let mut params = Vec::new();
312 4 : if let Some(values) = source.params() {
313 4 : for param in values {
314 3 : let param_name = register(&mut out, param.param_name().unwrap_or(""), NameTag::NONE);
315 3 : params.push(FunctionParam {
316 3 : ast_node_id: None,
317 3 : param_name,
318 3 : param_type: Arc::from(param.param_type().unwrap_or("")),
319 3 : });
320 3 : }
321 0 : }
322 4 : let function = FunctionDeclaration {
323 4 : object_id: QualifiedCatalogObjectId::function(function_id),
324 4 : catalog_schema_id: schema_id,
325 4 : is_aggregate: source.is_aggregate(),
326 4 : ast_node_id: None,
327 4 : ast_statement_id: None,
328 4 : function_name: QualifiedFunctionName {
329 4 : ast_node_id: None,
330 4 : database_name: db,
331 4 : schema_name: schema,
332 4 : function_name,
333 4 : },
334 4 : params,
335 4 : return_type: Arc::from(source.return_type().unwrap_or("")),
336 4 : location: source.location().map(|l| TextSpan::new(l.offset(), l.length())),
337 : };
338 4 : out.name_objects.push((
339 4 : function.function_name.function_name.id,
340 4 : CatalogObjectRef::Function(ObjectLoc {
341 4 : entry: id,
342 4 : index: function_index as u32,
343 4 : }),
344 4 : ));
345 4 : let entry = Arc::make_mut(&mut out.entry);
346 4 : entry.functions_by_qualified_name.entry(function.function_name.key()).or_insert(function_index);
347 4 : entry
348 4 : .functions_by_unqualified_name
349 4 : .entry(function.function_name.function_name.text.clone())
350 4 : .or_default()
351 4 : .push(function_index);
352 4 : entry.function_declarations.push(function);
353 : }
354 15 : }
355 44 : Ok(out)
356 47 : }
357 : }
|
