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