Line data Source code
1 : //! The catalog publishes catalog entries (the declarations of analyzed scripts and descriptor pools)
2 : //! atomically and resolves table names across them. Entries are immutable once published and shared
3 : //! through [`CatalogEntryRef`]. A script entry holds only declarations and names, never the analysis,
4 : //! so publishing does not keep ASTs or earlier analyses alive (see `docs/design/core_analysis_catalog.md`).
5 :
6 : pub mod clfn;
7 : pub mod descriptor;
8 : pub mod entry;
9 : pub mod ids;
10 : pub mod names;
11 :
12 : pub use clfn::ClientFunctionSpec;
13 : pub use descriptor::DescriptorEntry;
14 : pub use entry::*;
15 : pub use ids::*;
16 : pub use names::{IndexedName, NameSearchIndex};
17 :
18 : use crate::error::{Error, Result};
19 : use dashql_buffers::catalog::{
20 : FlatCatalogEntryT, FlatCatalogT, IndexedFlatDatabaseEntryT, IndexedFlatSchemaEntryT, IndexedFlatTableEntryT,
21 : };
22 : use dashql_buffers::status::StatusCode;
23 : use std::cell::RefCell;
24 : use crate::hash::HashMap;
25 : use std::collections::{BTreeMap, BTreeSet};
26 : use std::sync::atomic::{AtomicU32, Ordering};
27 : use std::sync::Arc;
28 :
29 : /// A published script: its declarations and the names that completion searches.
30 : ///
31 : /// The catalog keeps only this part of an analysis. Neither field refers to other entries, so
32 : /// published scripts never keep earlier versions of each other alive.
33 : #[derive(Clone, Debug)]
34 : pub struct PublishedScript {
35 : /// The id of the catalog the script was analyzed against (`Catalog::id`)
36 : pub catalog_id: u64,
37 : /// The declarations
38 : pub entry: Arc<CatalogEntryData>,
39 : /// The names of the script
40 : pub names: Arc<NameSearchIndex>,
41 : }
42 :
43 : /// A published catalog entry
44 : #[derive(Clone, Debug)]
45 : pub enum CatalogEntryRef {
46 : Script(PublishedScript),
47 : Descriptor(Arc<DescriptorEntry>),
48 : }
49 :
50 : impl CatalogEntryRef {
51 13905 : pub fn data(&self) -> &CatalogEntryData {
52 13905 : self.shared_data()
53 13905 : }
54 :
55 : /// The declarations of the entry, shared with the analyzed scripts that depend on it
56 13906 : pub fn shared_data(&self) -> &Arc<CatalogEntryData> {
57 13906 : match self {
58 13698 : CatalogEntryRef::Script(s) => &s.entry,
59 208 : CatalogEntryRef::Descriptor(d) => &d.entry,
60 : }
61 13906 : }
62 :
63 : /// The name search index of the entry, built on first use
64 122 : pub fn name_search_index(&self) -> &NameSearchIndex {
65 122 : match self {
66 119 : CatalogEntryRef::Script(s) => &s.names,
67 3 : CatalogEntryRef::Descriptor(d) => d.name_search_index(),
68 : }
69 122 : }
70 : }
71 :
72 : /// A database declared by the catalog
73 : #[derive(Clone, Debug)]
74 : pub struct DatabaseDeclaration {
75 : /// The qualified object id of the database
76 : pub object_id: QualifiedCatalogObjectId,
77 : /// The database name
78 : pub database_name: Arc<str>,
79 : /// The database alias, empty when not aliased
80 : pub database_alias: Arc<str>,
81 : }
82 :
83 : /// A schema declared by the catalog
84 : #[derive(Clone, Debug)]
85 : pub struct SchemaDeclaration {
86 : /// The qualified object id, packs the database and schema ids
87 : pub object_id: QualifiedCatalogObjectId,
88 : /// The database name
89 : pub database_name: Arc<str>,
90 : /// The schema name
91 : pub schema_name: Arc<str>,
92 : }
93 :
94 : /// A catalog entry that populates a schema
95 : #[derive(Clone, Copy, Debug)]
96 : pub struct SchemaEntryInfo {
97 : /// The catalog entry
98 : pub catalog_entry_id: CatalogEntryId,
99 : /// The qualified object id of the schema
100 : pub catalog_schema_id: QualifiedCatalogObjectId,
101 : }
102 :
103 : /// Canonical ids, reserved during analysis and never reused
104 : #[derive(Debug)]
105 : struct IdReservations {
106 : /// The reserved database ids by database name
107 : database_ids_by_name: HashMap<String, CatalogDatabaseId>,
108 : /// The reserved schema ids by (database, schema)
109 : schema_ids_by_name: HashMap<(String, String), QualifiedCatalogObjectId>,
110 : /// The next database id to reserve
111 : next_database_id: CatalogDatabaseId,
112 : /// The next schema id to reserve
113 : next_schema_id: CatalogSchemaId,
114 : /// The reserved table and function ids of the entries, by (entry, kind, qualified name, occurrence)
115 : object_ids: HashMap<(CatalogEntryId, ObjectKind, QualifiedKey, u32), u32>,
116 : /// The next table and function id to reserve in an entry
117 : next_object_ids: HashMap<(CatalogEntryId, ObjectKind), u32>,
118 : /// The reserved column ids of the tables, by (table, column name, occurrence)
119 : column_ids: HashMap<(CatalogTableId, Arc<str>, u32), u32>,
120 : /// The next column id to reserve in a table
121 : next_column_ids: HashMap<CatalogTableId, u32>,
122 : }
123 :
124 : /// The kind of an object that an entry declares
125 : #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
126 : enum ObjectKind {
127 : Table,
128 : Function,
129 : }
130 :
131 : impl IdReservations {
132 : /// Reserve the id of an object of an entry
133 2050 : fn object_id(&mut self, entry: CatalogEntryId, kind: ObjectKind, name: &QualifiedKey, occurrence: u32) -> u32 {
134 2050 : let key = (entry, kind, name.clone(), occurrence);
135 2050 : if let Some(&id) = self.object_ids.get(&key) {
136 410 : return id;
137 1640 : }
138 1640 : let next = self.next_object_ids.entry((entry, kind)).or_insert(0);
139 1640 : let id = *next;
140 1640 : *next += 1;
141 1640 : self.object_ids.insert(key, id);
142 1640 : id
143 2050 : }
144 :
145 : /// Release the object and column ids of an entry
146 68 : fn release_entry(&mut self, entry: CatalogEntryId) {
147 73 : self.object_ids.retain(|(e, ..), _| *e != entry);
148 70 : self.next_object_ids.retain(|(e, _), _| *e != entry);
149 72 : self.column_ids.retain(|(table, ..), _| table.origin != entry);
150 71 : self.next_column_ids.retain(|table, _| table.origin != entry);
151 68 : }
152 : }
153 :
154 : /// The source of catalog entry ids.
155 : ///
156 : /// Catalogs that share a source never hand out the same entry id, so entry ids identify scripts and
157 : /// descriptor pools across all of these catalogs. A catalog created with [`Catalog::new`] has a
158 : /// source of its own.
159 : #[derive(Clone, Debug)]
160 : pub struct EntryIdSource {
161 : /// The next entry id to allocate
162 : next: Arc<AtomicU32>,
163 : }
164 :
165 : impl Default for EntryIdSource {
166 0 : fn default() -> Self {
167 0 : Self::new()
168 0 : }
169 : }
170 :
171 : impl EntryIdSource {
172 : /// A source that starts at the initial entry id
173 1771 : pub fn new() -> Self {
174 1771 : Self { next: Arc::new(AtomicU32::new(INITIAL_ENTRY_ID)) }
175 1771 : }
176 :
177 : /// Allocate an entry id
178 13958 : pub fn allocate(&self) -> CatalogEntryId {
179 13958 : let id = self.next.fetch_add(1, Ordering::Relaxed);
180 13958 : assert!(id != PROTO_NULL_U32, "catalog entry ID namespace exhausted");
181 13958 : id
182 13958 : }
183 :
184 : /// Was the entry id allocated by this source?
185 65 : pub fn allocated(&self, id: CatalogEntryId) -> bool {
186 65 : id >= INITIAL_ENTRY_ID && id < self.next.load(Ordering::Relaxed)
187 65 : }
188 : }
189 :
190 : /// Fail for a value that is not a layer, the FlatBuffers enum admits every byte
191 1095 : fn check_layer(layer: CatalogLayer) -> Result<()> {
192 1095 : match layer.variant_name() {
193 1094 : Some(_) => Ok(()),
194 1 : None => Err(Error::invalid_argument("invalid catalog layer")),
195 : }
196 1095 : }
197 :
198 : /// The source of unique catalog ids
199 : static NEXT_CATALOG_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
200 :
201 : /// The published state of a catalog at one point in time.
202 : ///
203 : /// Immutable once published: analyses keep the snapshot they resolved against, so their catalog
204 : /// locations stay valid however the catalog changes later. A snapshot holds declarations and names,
205 : /// never analyses, so keeping one alive keeps no earlier analysis alive.
206 : #[derive(Clone, Debug)]
207 : pub struct CatalogSnapshot {
208 : /// The process-unique id of the catalog
209 : id: u64,
210 : /// The version, bumped by every modification
211 : version: CatalogVersion,
212 : /// The published entries
213 : entries: HashMap<CatalogEntryId, CatalogEntryRef>,
214 : /// The published scripts and their layers
215 : script_entries: HashMap<CatalogEntryId, (PublishedScript, CatalogLayer)>,
216 : /// The published descriptor pools
217 : descriptor_entries: HashMap<CatalogEntryId, Arc<DescriptorEntry>>,
218 : /// The entries ordered by (layer, id)
219 : entries_ranked: BTreeSet<(CatalogLayer, CatalogEntryId)>,
220 : /// The entries by (database, schema, layer, id)
221 : entries_by_qualified_schema: BTreeMap<(Arc<str>, Arc<str>, CatalogLayer, CatalogEntryId), SchemaEntryInfo>,
222 : /// The entries by (schema, layer, id)
223 : entries_by_schema: BTreeMap<(Arc<str>, CatalogLayer, CatalogEntryId), SchemaEntryInfo>,
224 : /// The databases referenced by entries
225 : databases: BTreeMap<Arc<str>, Arc<DatabaseDeclaration>>,
226 : /// The schemas referenced by entries
227 : schemas: BTreeMap<(Arc<str>, Arc<str>), Arc<SchemaDeclaration>>,
228 : /// The client function specs by namespace, a spec shadows its namespace for `CALL`
229 : client_function_specs: BTreeMap<Arc<str>, Arc<ClientFunctionSpec>>,
230 : /// The relations that the connection reported as gone, their declarations in scripts are hidden until the
231 : /// connection has them again. An empty part of a name matches any database or schema.
232 : absent_relations: Vec<QualifiedKey>,
233 : }
234 :
235 : /// The catalog: the current snapshot and the ids that outlive snapshots.
236 : ///
237 : /// Reads go to the current snapshot (`Deref`). A modification copies the snapshot if an analysis
238 : /// still holds it (`DerefMut` is `Arc::make_mut`), so published snapshots never change.
239 : #[derive(Debug)]
240 : pub struct Catalog {
241 : /// The current snapshot
242 : current: Arc<CatalogSnapshot>,
243 : /// The canonical ids, reserved during analysis and never reused
244 : ids: RefCell<IdReservations>,
245 : /// The source of the entry ids
246 : entry_ids: EntryIdSource,
247 : }
248 :
249 : impl std::ops::Deref for Catalog {
250 : type Target = CatalogSnapshot;
251 :
252 23766 : fn deref(&self) -> &CatalogSnapshot {
253 23766 : &self.current
254 23766 : }
255 : }
256 :
257 : impl std::ops::DerefMut for Catalog {
258 8584 : fn deref_mut(&mut self) -> &mut CatalogSnapshot {
259 8584 : Arc::make_mut(&mut self.current)
260 8584 : }
261 : }
262 :
263 : impl Default for Catalog {
264 18 : fn default() -> Self {
265 18 : Self::new()
266 18 : }
267 : }
268 :
269 : impl Catalog {
270 1770 : pub fn new() -> Self {
271 1770 : Self::with_entry_ids(EntryIdSource::new())
272 1770 : }
273 :
274 : /// A catalog that allocates its entry ids from a shared source
275 1772 : pub fn with_entry_ids(entry_ids: EntryIdSource) -> Self {
276 1772 : Self {
277 1772 : current: Arc::new(CatalogSnapshot {
278 1772 : id: NEXT_CATALOG_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
279 1772 : version: 1,
280 1772 : entries: HashMap::default(),
281 1772 : script_entries: HashMap::default(),
282 1772 : descriptor_entries: HashMap::default(),
283 1772 : entries_ranked: BTreeSet::new(),
284 1772 : entries_by_qualified_schema: BTreeMap::new(),
285 1772 : entries_by_schema: BTreeMap::new(),
286 1772 : databases: BTreeMap::new(),
287 1772 : schemas: BTreeMap::new(),
288 1772 : client_function_specs: BTreeMap::new(),
289 1772 : absent_relations: Vec::new(),
290 1772 : }),
291 1772 : ids: RefCell::new(IdReservations {
292 1772 : database_ids_by_name: HashMap::default(),
293 1772 : schema_ids_by_name: HashMap::default(),
294 1772 : next_database_id: INITIAL_DATABASE_ID,
295 1772 : next_schema_id: INITIAL_SCHEMA_ID,
296 1772 : object_ids: HashMap::default(),
297 1772 : next_object_ids: HashMap::default(),
298 1772 : column_ids: HashMap::default(),
299 1772 : next_column_ids: HashMap::default(),
300 1772 : }),
301 1772 : entry_ids,
302 1772 : }
303 1772 : }
304 :
305 : /// The current snapshot
306 4003 : pub fn snapshot(&self) -> &Arc<CatalogSnapshot> {
307 4003 : &self.current
308 4003 : }
309 :
310 : /// Is a snapshot of this catalog current? A snapshot stays current while the version stays,
311 : /// i.e. while the declarations that analyses resolve against stay the same.
312 1129 : pub fn is_current(&self, snapshot: &CatalogSnapshot) -> bool {
313 1129 : snapshot.id == self.current.id && snapshot.version == self.current.version
314 1129 : }
315 : }
316 :
317 : impl CatalogSnapshot {
318 : /// The process-unique id of the catalog
319 3874 : pub fn id(&self) -> u64 {
320 3874 : self.id
321 3874 : }
322 :
323 5585 : pub fn version(&self) -> CatalogVersion {
324 5585 : self.version
325 5585 : }
326 :
327 30 : pub fn contains(&self, id: CatalogEntryId) -> bool {
328 30 : self.entries.contains_key(&id)
329 30 : }
330 :
331 1136 : pub fn entry(&self, id: CatalogEntryId) -> Option<&CatalogEntryRef> {
332 1136 : self.entries.get(&id)
333 1136 : }
334 :
335 : /// The layer of an entry
336 3668 : pub fn layer(&self, id: CatalogEntryId) -> Option<CatalogLayer> {
337 3668 : match self.script_entries.get(&id) {
338 3624 : Some((_, layer)) => Some(*layer),
339 44 : None => self.descriptor_entries.get(&id).map(|d| d.layer),
340 : }
341 3668 : }
342 :
343 : /// The entries in resolution order: by layer, then by entry id
344 1232 : pub fn entries_ranked(&self) -> impl Iterator<Item = (CatalogLayer, &CatalogEntryRef)> {
345 1241 : self.entries_ranked.iter().map(|(layer, id)| (*layer, &self.entries[id]))
346 1232 : }
347 :
348 8 : pub fn databases(&self) -> &BTreeMap<Arc<str>, Arc<DatabaseDeclaration>> {
349 8 : &self.databases
350 8 : }
351 :
352 44 : pub fn schemas(&self) -> &BTreeMap<(Arc<str>, Arc<str>), Arc<SchemaDeclaration>> {
353 44 : &self.schemas
354 44 : }
355 :
356 : /// The entries by (database, schema, layer, id)
357 18 : pub fn entries_by_qualified_schema(&self) -> &BTreeMap<(Arc<str>, Arc<str>, CatalogLayer, CatalogEntryId), SchemaEntryInfo> {
358 18 : &self.entries_by_qualified_schema
359 18 : }
360 :
361 : /// The entries by (schema, layer, id)
362 37 : pub fn entries_by_schema(&self) -> &BTreeMap<(Arc<str>, CatalogLayer, CatalogEntryId), SchemaEntryInfo> {
363 37 : &self.entries_by_schema
364 37 : }
365 : }
366 :
367 : impl Catalog {
368 : /// Allocate an entry id
369 13958 : pub fn allocate_entry_id(&self) -> CatalogEntryId {
370 13958 : self.entry_ids.allocate()
371 13958 : }
372 :
373 : /// Was the entry id allocated by the id source of the catalog?
374 65 : pub fn entry_id_allocated(&self, id: CatalogEntryId) -> bool {
375 65 : self.entry_ids.allocated(id)
376 65 : }
377 :
378 : /// Reserve the canonical id of a database
379 1359 : pub fn reserve_database_id(&self, database: &str) -> QualifiedCatalogObjectId {
380 1359 : let mut ids = self.ids.borrow_mut();
381 1359 : if let Some(&id) = ids.database_ids_by_name.get(database) {
382 480 : return QualifiedCatalogObjectId::database(id);
383 879 : }
384 879 : let id = ids.next_database_id;
385 879 : ids.next_database_id += 1;
386 879 : ids.database_ids_by_name.insert(database.to_string(), id);
387 879 : QualifiedCatalogObjectId::database(id)
388 1359 : }
389 :
390 : /// Reserve the id of a table of an entry.
391 : ///
392 : /// The id is keyed by the qualified name and the occurrence of that name among the tables of the
393 : /// entry, so a table keeps its id across versions of the entry however its declarations move. A
394 : /// fresh entry gets the ids in declaration order. Dropping the entry releases its ids.
395 2011 : pub fn reserve_table_id(&self, entry: CatalogEntryId, name: &QualifiedKey, occurrence: u32) -> CatalogTableId {
396 2011 : CatalogTableId::new(entry, self.ids.borrow_mut().object_id(entry, ObjectKind::Table, name, occurrence))
397 2011 : }
398 :
399 : /// Reserve the id of a function of an entry, like [`Catalog::reserve_table_id`]
400 39 : pub fn reserve_function_id(&self, entry: CatalogEntryId, name: &QualifiedKey, occurrence: u32) -> CatalogFunctionId {
401 39 : CatalogFunctionId::new(entry, self.ids.borrow_mut().object_id(entry, ObjectKind::Function, name, occurrence))
402 39 : }
403 :
404 : /// Reserve the id of a column of a table, keyed by the column name and its occurrence among the
405 : /// columns of the table
406 4585 : pub fn reserve_column_id(&self, table: CatalogTableId, name: &Arc<str>, occurrence: u32) -> u32 {
407 4585 : let mut ids = self.ids.borrow_mut();
408 4585 : let key = (table, name.clone(), occurrence);
409 4585 : if let Some(&id) = ids.column_ids.get(&key) {
410 378 : return id;
411 4207 : }
412 4207 : let next = ids.next_column_ids.entry(table).or_insert(0);
413 4207 : let id = *next;
414 4207 : *next += 1;
415 4207 : ids.column_ids.insert(key, id);
416 4207 : id
417 4585 : }
418 :
419 : /// Reserve the canonical id of a schema
420 1382 : pub fn reserve_schema_id(&self, database: &str, schema: &str, db_id: QualifiedCatalogObjectId) -> QualifiedCatalogObjectId {
421 1382 : let mut ids = self.ids.borrow_mut();
422 1382 : let key = (database.to_string(), schema.to_string());
423 1382 : if let Some(&id) = ids.schema_ids_by_name.get(&key) {
424 476 : return id;
425 906 : }
426 906 : let schema_id = ids.next_schema_id;
427 906 : ids.next_schema_id += 1;
428 906 : let id = QualifiedCatalogObjectId::schema(db_id.unpack_database_id(), schema_id);
429 906 : ids.schema_ids_by_name.insert(key, id);
430 906 : id
431 1382 : }
432 :
433 1035 : fn check_canonical_ids(&self, entry: &CatalogEntryData) -> Result<()> {
434 1035 : let ids = self.ids.borrow();
435 1052 : for (name, &i) in &entry.databases_by_name {
436 1049 : let object_id = entry.database_references[i].object_id;
437 1049 : match ids.database_ids_by_name.get(&**name) {
438 1049 : Some(&id) if object_id == QualifiedCatalogObjectId::database(id) => {}
439 0 : _ => return Err(Error::status(StatusCode::CATALOG_ID_OUT_OF_SYNC)),
440 : }
441 : }
442 1063 : for ((db, schema), &i) in &entry.schemas_by_qualified_name {
443 1060 : let object_id = entry.schema_references[i].object_id;
444 1060 : match ids.schema_ids_by_name.get(&(db.to_string(), schema.to_string())) {
445 1060 : Some(&id) if id == object_id => {}
446 0 : _ => return Err(Error::status(StatusCode::CATALOG_ID_OUT_OF_SYNC)),
447 : }
448 : }
449 1035 : Ok(())
450 1035 : }
451 :
452 : /// Load an analyzed script
453 882 : pub fn load_script(&mut self, script: impl Into<PublishedScript>, layer: CatalogLayer) -> Result<()> {
454 882 : self.load_scripts(&[(script.into(), layer)])
455 882 : }
456 :
457 : /// Atomically add or replace scripts
458 1455 : pub fn load_scripts<S: Clone + Into<PublishedScript>>(&mut self, scripts: &[(S, CatalogLayer)]) -> Result<()> {
459 1455 : if scripts.is_empty() {
460 450 : return Ok(());
461 1005 : }
462 1005 : let scripts: Vec<(PublishedScript, CatalogLayer)> =
463 1037 : scripts.iter().map(|(script, layer)| (script.clone().into(), *layer)).collect();
464 1005 : let mut batch_ids = BTreeSet::new();
465 1037 : for (_, layer) in &scripts {
466 1037 : check_layer(*layer)?;
467 : }
468 1037 : for (script, _) in &scripts {
469 1037 : if script.catalog_id != self.id {
470 1 : return Err(Error::status(StatusCode::CATALOG_MISMATCH));
471 1036 : }
472 1036 : let id = script.entry.catalog_entry_id;
473 1036 : if !batch_ids.insert(id) {
474 1 : return Err(Error::status(StatusCode::EXTERNAL_ID_COLLISION));
475 1035 : }
476 1035 : if self.entries.contains_key(&id) && !self.script_entries.contains_key(&id) {
477 0 : return Err(Error::status(StatusCode::EXTERNAL_ID_COLLISION));
478 1035 : }
479 1035 : self.check_canonical_ids(&script.entry)?;
480 : }
481 : // Scripts that declare what their published versions declare replace them in place: the
482 : // indexes stay valid and the catalog version stays, so the analyses of other scripts remain
483 : // current
484 1003 : let unchanged = scripts.iter().all(|(script, layer)| {
485 1003 : self.script_entries.get(&script.entry.catalog_entry_id).is_some_and(|(published, published_layer)| {
486 216 : published_layer == layer && published.entry.same_declarations(&script.entry)
487 216 : })
488 1003 : });
489 1003 : if unchanged {
490 204 : for (script, layer) in &scripts {
491 204 : let id = script.entry.catalog_entry_id;
492 204 : self.script_entries.insert(id, (script.clone(), *layer));
493 204 : self.entries.insert(id, CatalogEntryRef::Script(script.clone()));
494 204 : }
495 204 : return Ok(());
496 799 : }
497 799 : let mut staged = self.script_entries.clone();
498 829 : for (script, layer) in &scripts {
499 829 : staged.insert(script.entry.catalog_entry_id, (script.clone(), *layer));
500 829 : }
501 799 : let descriptors = self.descriptor_entries.clone();
502 799 : self.publish_entries(staged, descriptors)
503 1455 : }
504 :
505 : /// Build all indexes before committing
506 852 : fn publish_entries(
507 852 : &mut self,
508 852 : script_entries: HashMap<CatalogEntryId, (PublishedScript, CatalogLayer)>,
509 852 : descriptor_entries: HashMap<CatalogEntryId, Arc<DescriptorEntry>>,
510 852 : ) -> Result<()> {
511 852 : let mut entries = HashMap::default();
512 852 : let mut entries_ranked = BTreeSet::new();
513 852 : let mut entries_by_qualified_schema = BTreeMap::new();
514 852 : let mut entries_by_schema = BTreeMap::new();
515 852 : let mut databases: BTreeMap<Arc<str>, Arc<DatabaseDeclaration>> = BTreeMap::new();
516 852 : let mut schemas: BTreeMap<(Arc<str>, Arc<str>), Arc<SchemaDeclaration>> = BTreeMap::new();
517 :
518 949 : let mut stage = |entry: CatalogEntryRef, layer: CatalogLayer| -> Result<()> {
519 949 : let data = entry.data();
520 949 : let entry_id = data.catalog_entry_id;
521 949 : entries_ranked.insert((layer, entry_id));
522 971 : for (name, &i) in &data.databases_by_name {
523 968 : let db = &data.database_references[i];
524 968 : if let Some(existing) = databases.get(name) {
525 79 : if existing.object_id != db.object_id {
526 0 : return Err(Error::status(StatusCode::CATALOG_ID_OUT_OF_SYNC));
527 79 : }
528 79 : continue;
529 889 : }
530 889 : databases.insert(
531 889 : db.database_name.clone(),
532 889 : Arc::new(DatabaseDeclaration {
533 889 : object_id: db.object_id,
534 889 : database_name: db.database_name.clone(),
535 889 : database_alias: db.database_alias.clone(),
536 889 : }),
537 : );
538 : }
539 989 : for ((db_name, schema_name), &i) in &data.schemas_by_qualified_name {
540 986 : let schema = &data.schema_references[i];
541 986 : let key = (db_name.clone(), schema_name.clone());
542 986 : if let Some(existing) = schemas.get(&key) {
543 75 : if existing.object_id != schema.object_id {
544 0 : return Err(Error::status(StatusCode::CATALOG_ID_OUT_OF_SYNC));
545 75 : }
546 75 : continue;
547 911 : }
548 911 : if !databases.contains_key(db_name) {
549 0 : return Err(Error::status(StatusCode::CATALOG_ID_OUT_OF_SYNC));
550 911 : }
551 911 : schemas.insert(
552 911 : key,
553 911 : Arc::new(SchemaDeclaration {
554 911 : object_id: schema.object_id,
555 911 : database_name: db_name.clone(),
556 911 : schema_name: schema.schema_name.clone(),
557 911 : }),
558 : );
559 : }
560 989 : for ((db_name, schema_name), &i) in &data.schemas_by_qualified_name {
561 986 : let info = SchemaEntryInfo {
562 986 : catalog_entry_id: entry_id,
563 986 : catalog_schema_id: data.schema_references[i].object_id,
564 986 : };
565 986 : entries_by_qualified_schema.insert((db_name.clone(), schema_name.clone(), layer, entry_id), info);
566 986 : entries_by_schema.insert((schema_name.clone(), layer, entry_id), info);
567 986 : }
568 949 : if entries.insert(entry_id, entry.clone()).is_some() {
569 0 : return Err(Error::status(StatusCode::EXTERNAL_ID_COLLISION));
570 949 : }
571 949 : Ok(())
572 949 : };
573 933 : for (script, layer) in script_entries.values() {
574 885 : stage(CatalogEntryRef::Script(script.clone()), *layer)?;
575 : }
576 854 : for descriptor in descriptor_entries.values() {
577 64 : stage(CatalogEntryRef::Descriptor(descriptor.clone()), descriptor.layer)?;
578 : }
579 852 : self.script_entries = script_entries;
580 852 : self.descriptor_entries = descriptor_entries;
581 852 : self.entries = entries;
582 852 : self.entries_ranked = entries_ranked;
583 852 : self.entries_by_qualified_schema = entries_by_qualified_schema;
584 852 : self.entries_by_schema = entries_by_schema;
585 852 : self.databases = databases;
586 852 : self.schemas = schemas;
587 852 : self.version += 1;
588 852 : Ok(())
589 852 : }
590 :
591 : /// Replace a descriptor pool atomically
592 62 : pub fn replace_descriptor(&mut self, id: CatalogEntryId, layer: CatalogLayer, descriptor: &[u8]) -> Result<()> {
593 62 : if !self.entry_id_allocated(id)
594 59 : || (self.entries.contains_key(&id) && !self.descriptor_entries.contains_key(&id))
595 : {
596 4 : return Err(Error::invalid_argument("invalid or colliding receiver catalog entry ID"));
597 58 : }
598 58 : check_layer(layer)?;
599 : // Identical bytes declare the same, the catalog version stays and the analyses stay current
600 57 : if self.descriptor_entries.get(&id).is_some_and(|e| e.layer == layer && e.bytes == descriptor) {
601 1 : return Ok(());
602 56 : }
603 56 : let entry = DescriptorEntry::new(self, id, layer, descriptor)?;
604 68 : let reported: Vec<QualifiedKey> = entry.entry.table_declarations.iter().map(|t| t.table_name.key()).collect();
605 52 : let mut staged = self.descriptor_entries.clone();
606 52 : staged.insert(id, Arc::new(entry));
607 52 : let scripts = self.script_entries.clone();
608 52 : self.publish_entries(scripts, staged)?;
609 : // The connection has these relations again
610 52 : self.absent_relations.retain(|absent| !reported.iter().any(|table| same_relation(absent, table)));
611 52 : Ok(())
612 62 : }
613 :
614 : /// Update the tables of a descriptor pool with a packed `CatalogTableUpdate`, an unknown pool is created in the
615 : /// layer. Tables that the update leaves as they are keep their position, an update that changes nothing keeps
616 : /// the catalog version. A relation that has no table anymore is gone: scripts that declare it no longer provide it.
617 27 : pub fn replace_tables(&mut self, id: CatalogEntryId, layer: CatalogLayer, update: &[u8]) -> Result<()> {
618 : use dashql_buffers::analyzer::{QualifiedTableNameT, TableT};
619 : use dashql_buffers::catalog::{CatalogDescriptor, CatalogDescriptorT, CatalogTableUpdate};
620 27 : let update = flatbuffers::root::<CatalogTableUpdate>(update)
621 27 : .map_err(|_| Error::invalid_argument("invalid CatalogTableUpdate FlatBuffer"))?
622 26 : .unpack();
623 26 : let (mut descriptor, layer) = match self.descriptor_entries.get(&id) {
624 15 : Some(entry) => {
625 15 : let descriptor = flatbuffers::root::<CatalogDescriptor>(&entry.bytes)
626 15 : .map_err(|_| Error::invalid_argument("invalid CatalogDescriptor FlatBuffer"))?
627 15 : .unpack();
628 15 : (descriptor, entry.layer)
629 : }
630 11 : None => (CatalogDescriptorT::default(), layer),
631 : };
632 62 : let key_of = |name: &QualifiedTableNameT| -> QualifiedKey {
633 62 : (
634 62 : Arc::from(name.database_name.as_deref().unwrap_or("")),
635 62 : Arc::from(name.schema_name.as_deref().unwrap_or("")),
636 62 : Arc::from(name.table_name.as_deref().unwrap_or("")),
637 62 : )
638 62 : };
639 48 : let key_of_table = |table: &TableT| key_of(table.table_name.as_deref().unwrap_or(&QualifiedTableNameT::default()));
640 26 : let before = descriptor.tables.take().unwrap_or_default();
641 26 : let new_tables = update.tables.unwrap_or_default();
642 26 : let new_keys: Vec<QualifiedKey> = new_tables.iter().map(key_of_table).collect();
643 26 : let relations: Vec<QualifiedKey> = update.relations.unwrap_or_default().iter().map(key_of).collect();
644 26 : let databases = update.databases.unwrap_or_default();
645 26 : let mut tables: Vec<TableT> = Vec::with_capacity(before.len() + new_tables.len());
646 26 : let mut before_keys: Vec<QualifiedKey> = Vec::with_capacity(before.len());
647 26 : for table in &before {
648 24 : let key = key_of_table(table);
649 24 : if let Some(next) = new_keys.iter().position(|k| *k == key) {
650 3 : tables.push(new_tables[next].clone());
651 21 : } else if !relations.iter().any(|r| same_relation(r, &key)) && !databases.iter().any(|d| d.as_str() == &*key.0) {
652 14 : tables.push(table.clone());
653 14 : }
654 24 : before_keys.push(key);
655 : }
656 26 : for (table, key) in new_tables.iter().zip(&new_keys) {
657 24 : if !before_keys.contains(key) {
658 21 : tables.push(table.clone());
659 21 : }
660 : }
661 : // The relations that the connection has no table of are gone, the others are not
662 26 : let mut absent = self.absent_relations.clone();
663 26 : absent.retain(|a| !new_keys.iter().any(|k| same_relation(a, k)));
664 26 : for relation in &relations {
665 14 : if !new_keys.iter().any(|k| same_relation(relation, k)) && !absent.contains(relation) {
666 7 : absent.push(relation.clone());
667 7 : }
668 : }
669 26 : let absent_changed = absent != self.absent_relations;
670 26 : if tables == before && self.descriptor_entries.contains_key(&id) {
671 5 : if absent_changed {
672 3 : self.absent_relations = absent;
673 3 : self.version += 1;
674 3 : }
675 5 : return Ok(());
676 21 : }
677 21 : descriptor.tables = Some(tables);
678 21 : let mut builder = flatbuffers::FlatBufferBuilder::new();
679 21 : let packed = descriptor.pack(&mut builder);
680 21 : builder.finish(packed, None);
681 21 : self.replace_descriptor(id, layer, builder.finished_data())?;
682 20 : self.absent_relations = absent;
683 20 : Ok(())
684 27 : }
685 :
686 : /// The relations that the connection reported as gone
687 0 : pub fn absent_relations(&self) -> &[QualifiedKey] {
688 0 : &self.absent_relations
689 0 : }
690 :
691 : /// The bytes of a descriptor pool
692 0 : pub fn descriptor_bytes(&self, id: CatalogEntryId) -> Option<&[u8]> {
693 0 : self.descriptor_entries.get(&id).map(|entry| entry.bytes.as_slice())
694 0 : }
695 :
696 : /// Drop a descriptor pool
697 3 : pub fn drop_descriptor(&mut self, id: CatalogEntryId) -> Result<()> {
698 3 : if !self.descriptor_entries.contains_key(&id) {
699 2 : return Ok(());
700 1 : }
701 1 : self.ids.borrow_mut().release_entry(id);
702 1 : let mut staged = self.descriptor_entries.clone();
703 1 : staged.remove(&id);
704 1 : let scripts = self.script_entries.clone();
705 1 : self.publish_entries(scripts, staged)
706 3 : }
707 :
708 : /// Drop a script
709 68 : pub fn drop_script(&mut self, id: CatalogEntryId) {
710 : // Also for scripts that were analyzed but never published. A descriptor pool published under the id of the
711 : // script owns the ids, they stay until the pool is dropped.
712 68 : if !self.descriptor_entries.contains_key(&id) {
713 67 : self.ids.borrow_mut().release_entry(id);
714 68 : }
715 68 : let Some((analyzed, layer)) = self.script_entries.remove(&id) else {
716 54 : return;
717 : };
718 15 : for (db_name, schema_name) in analyzed.entry.schemas_by_qualified_name.keys() {
719 13 : self.entries_by_qualified_schema.remove(&(db_name.clone(), schema_name.clone(), layer, id));
720 13 : self.entries_by_schema.remove(&(schema_name.clone(), layer, id));
721 13 : }
722 14 : self.entries_ranked.remove(&(layer, id));
723 14 : self.entries.remove(&id);
724 15 : for (db_name, schema_name) in analyzed.entry.schemas_by_qualified_name.keys() {
725 13 : let lower = (db_name.clone(), schema_name.clone(), MIN_LAYER, 0);
726 13 : let remaining = self.entries_by_qualified_schema.range(lower..).next();
727 1 : match remaining {
728 1 : Some(((d, s, _, _), _)) if d == db_name && s == schema_name => {}
729 12 : _ => {
730 12 : self.schemas.remove(&(db_name.clone(), schema_name.clone()));
731 12 : }
732 : }
733 : }
734 15 : for name in analyzed.entry.databases_by_name.keys() {
735 13 : let referenced = self.entries.values().any(|e| e.data().databases_by_name.contains_key(name));
736 13 : if !referenced {
737 12 : self.databases.remove(name);
738 12 : }
739 : }
740 14 : self.version += 1;
741 68 : }
742 :
743 : /// Register a client function spec. Its namespace must be free, drop the spec first to replace it.
744 125 : pub fn register_client_functions(&mut self, json: &str) -> Result<Arc<str>> {
745 125 : let spec = ClientFunctionSpec::parse(json)?;
746 125 : if self.client_function_specs.contains_key(spec.namespace.as_str()) {
747 0 : return Err(Error::invalid_argument(format!("the namespace {} is already registered", spec.namespace)));
748 125 : }
749 125 : let namespace: Arc<str> = Arc::from(spec.namespace.as_str());
750 125 : self.client_function_specs.insert(namespace.clone(), Arc::new(spec));
751 125 : self.version += 1;
752 125 : Ok(namespace)
753 125 : }
754 :
755 : /// Drop a client function spec
756 0 : pub fn drop_client_functions(&mut self, namespace: &str) {
757 0 : if self.client_function_specs.remove(namespace).is_some() {
758 0 : self.version += 1;
759 0 : }
760 0 : }
761 :
762 : /// Clear the catalog. The client function specs stay, they come with the connector and not from the backend.
763 2 : pub fn clear(&mut self) {
764 2 : self.entries_by_qualified_schema.clear();
765 2 : self.entries_by_schema.clear();
766 2 : self.entries_ranked.clear();
767 2 : self.entries.clear();
768 2 : self.script_entries.clear();
769 2 : self.descriptor_entries.clear();
770 2 : self.schemas.clear();
771 2 : self.databases.clear();
772 2 : self.absent_relations.clear();
773 2 : self.version += 1;
774 2 : }
775 :
776 : }
777 :
778 : impl CatalogSnapshot {
779 : /// The client function spec of a namespace
780 126 : pub fn client_function_spec(&self, namespace: &str) -> Option<&Arc<ClientFunctionSpec>> {
781 126 : self.client_function_specs.get(namespace)
782 126 : }
783 :
784 : /// The client function specs by namespace
785 5034 : pub fn client_function_specs(&self) -> &BTreeMap<Arc<str>, Arc<ClientFunctionSpec>> {
786 5034 : &self.client_function_specs
787 5034 : }
788 :
789 : /// Flatten the catalog.
790 : ///
791 : /// Databases, schemas and tables are merged across entries and ordered by name. A table name that
792 : /// occurs in several entries is taken from the first entry in resolution order. Names go through a
793 : /// dictionary, built in resolution order, so that JS only decodes the names it renders.
794 10 : pub fn flatten(&self) -> FlatCatalogT {
795 : /// A table of the flattened catalog
796 : struct TableNode {
797 : /// The table id
798 : table_id: CatalogTableId,
799 : /// The index of the table name in the name dictionary
800 : name_id: usize,
801 : /// The index of the first column in the column nodes
802 : columns_begin: usize,
803 : /// The number of columns
804 : column_count: usize,
805 : }
806 : /// A schema of the flattened catalog
807 : struct SchemaNode {
808 : /// The schema id
809 : schema_id: CatalogSchemaId,
810 : /// The index of the schema name in the name dictionary
811 : name_id: usize,
812 : /// The table node indexes by table name
813 : children: BTreeMap<Arc<str>, usize>,
814 : }
815 : /// A database of the flattened catalog
816 : struct DatabaseNode {
817 : /// The database id
818 : database_id: CatalogDatabaseId,
819 : /// The index of the database name in the name dictionary
820 : name_id: usize,
821 : /// The schema node indexes by schema name
822 : children: BTreeMap<Arc<str>, usize>,
823 : }
824 :
825 10 : let mut name_dictionary: Vec<String> = Vec::new();
826 10 : let mut name_dictionary_index: HashMap<Arc<str>, usize> = HashMap::default();
827 111 : let mut add_name = |name: &Arc<str>| -> usize {
828 111 : *name_dictionary_index.entry(name.clone()).or_insert_with(|| {
829 103 : name_dictionary.push(name.to_string());
830 103 : name_dictionary.len() - 1
831 103 : })
832 111 : };
833 :
834 10 : let mut database_nodes: Vec<DatabaseNode> = Vec::new();
835 10 : let mut schema_nodes: Vec<SchemaNode> = Vec::new();
836 10 : let mut table_nodes: Vec<TableNode> = Vec::new();
837 : // (column id, name id)
838 10 : let mut column_nodes: Vec<(u32, usize)> = Vec::new();
839 10 : let mut root: BTreeMap<Arc<str>, usize> = BTreeMap::new();
840 10 : let mut database_node_map: HashMap<QualifiedCatalogObjectId, usize> = HashMap::default();
841 10 : let mut schema_node_map: HashMap<QualifiedCatalogObjectId, usize> = HashMap::default();
842 :
843 : // Visit entries in resolution order, the name dictionary must not depend on the hash implementation
844 13 : for (_, entry) in self.entries_ranked() {
845 11 : let data = entry.data();
846 13 : for db_ref in &data.database_references {
847 13 : if database_node_map.contains_key(&db_ref.object_id) {
848 3 : continue;
849 10 : }
850 10 : let name_id = add_name(&db_ref.database_name);
851 10 : database_nodes.push(DatabaseNode {
852 10 : database_id: db_ref.database_id(),
853 10 : name_id,
854 10 : children: BTreeMap::new(),
855 10 : });
856 10 : database_node_map.insert(db_ref.object_id, database_nodes.len() - 1);
857 10 : root.entry(db_ref.database_name.clone()).or_insert(database_nodes.len() - 1);
858 : }
859 13 : for &i in data.schemas_by_qualified_name.values() {
860 13 : let schema_ref = &data.schema_references[i];
861 13 : if schema_node_map.contains_key(&schema_ref.object_id) {
862 3 : continue;
863 10 : }
864 10 : let name_id = add_name(&schema_ref.schema_name);
865 10 : schema_nodes.push(SchemaNode {
866 10 : schema_id: schema_ref.schema_id(),
867 10 : name_id,
868 10 : children: BTreeMap::new(),
869 10 : });
870 10 : let schema_node = schema_nodes.len() - 1;
871 10 : schema_node_map.insert(schema_ref.object_id, schema_node);
872 10 : let db_node = database_node_map[&QualifiedCatalogObjectId::database(schema_ref.database_id())];
873 10 : database_nodes[db_node].children.entry(schema_ref.schema_name.clone()).or_insert(schema_node);
874 : }
875 : }
876 :
877 : // Tables are not deduplicated among entries, the first entry in resolution order wins
878 13 : for (_, entry) in self.entries_ranked() {
879 20 : for (index, table) in entry.data().table_declarations.iter().enumerate() {
880 20 : if self.is_absent(entry.data().catalog_entry_id, table) || entry.data().tables_by_qualified_name.get(&table.table_name.key()) != Some(&index) {
881 1 : continue;
882 19 : }
883 19 : let schema_node = schema_node_map[&table.catalog_schema_id];
884 19 : let table_name = &table.table_name.table_name.text;
885 19 : if schema_nodes[schema_node].children.contains_key(table_name) {
886 0 : continue;
887 19 : }
888 19 : let columns_begin = column_nodes.len();
889 72 : for column in &table.table_columns {
890 72 : let name_id = add_name(&column.column_name.text);
891 72 : column_nodes.push((column.column_id(), name_id));
892 72 : }
893 19 : let name_id = add_name(table_name);
894 19 : table_nodes.push(TableNode {
895 19 : table_id: table.table_id(),
896 19 : name_id,
897 19 : columns_begin,
898 19 : column_count: table.table_columns.len(),
899 19 : });
900 19 : schema_nodes[schema_node].children.insert(table_name.clone(), table_nodes.len() - 1);
901 : }
902 : }
903 :
904 : // (entry index, parent index, object id, name id, child begin, child count)
905 111 : let entry = |idx: usize, parent: usize, object_id: u64, name_id: usize, begin: usize, count: usize| {
906 111 : FlatCatalogEntryT {
907 111 : flat_entry_idx: idx as u32,
908 111 : flat_parent_idx: parent as u32,
909 111 : catalog_object_id: object_id,
910 111 : name_id: name_id as u32,
911 111 : child_begin: begin as u32,
912 111 : child_count: count as u32,
913 111 : }
914 111 : };
915 10 : let mut databases = Vec::new();
916 10 : let mut schemas = Vec::new();
917 10 : let mut tables = Vec::new();
918 10 : let mut columns = Vec::new();
919 10 : let mut databases_by_id = Vec::new();
920 10 : let mut schemas_by_id = Vec::new();
921 10 : let mut tables_by_id = Vec::new();
922 12 : for &db_node in root.values() {
923 10 : let db = &database_nodes[db_node];
924 10 : let db_idx = databases.len();
925 10 : databases.push(entry(db_idx, 0, db.database_id as u64, db.name_id, schemas.len(), db.children.len()));
926 10 : databases_by_id.push(IndexedFlatDatabaseEntryT {
927 10 : database_id: db.database_id,
928 10 : flat_entry_idx: db_idx as u32,
929 10 : });
930 10 : for &schema_node in db.children.values() {
931 10 : let schema = &schema_nodes[schema_node];
932 10 : let schema_idx = schemas.len();
933 10 : let schema_id = schema.schema_id as u64;
934 10 : schemas.push(entry(schema_idx, db_idx, schema_id, schema.name_id, tables.len(), schema.children.len()));
935 10 : schemas_by_id.push(IndexedFlatSchemaEntryT {
936 10 : schema_id: schema.schema_id,
937 10 : flat_entry_idx: schema_idx as u32,
938 10 : });
939 20 : for &table_node in schema.children.values() {
940 19 : let table = &table_nodes[table_node];
941 19 : let table_idx = tables.len();
942 19 : let table_id = table.table_id.pack();
943 19 : let column_count = table.column_count;
944 19 : tables.push(entry(table_idx, schema_idx, table_id, table.name_id, columns.len(), column_count));
945 19 : tables_by_id.push(IndexedFlatTableEntryT {
946 19 : table_id,
947 19 : flat_entry_idx: table_idx as u32,
948 19 : });
949 19 : let table_columns = table.columns_begin..table.columns_begin + table.column_count;
950 72 : for &(column_id, name_id) in &column_nodes[table_columns] {
951 72 : columns.push(entry(columns.len(), table_idx, column_id as u64, name_id, 0, 0));
952 72 : }
953 : }
954 : }
955 : }
956 10 : databases_by_id.sort_by_key(|e| e.database_id);
957 10 : schemas_by_id.sort_by_key(|e| e.schema_id);
958 10 : tables_by_id.sort_by_key(|e| e.table_id);
959 :
960 10 : let mut out = FlatCatalogT::default();
961 10 : out.catalog_version = self.version as u64;
962 10 : out.name_dictionary = Some(name_dictionary);
963 10 : out.databases = Some(databases);
964 10 : out.schemas = Some(schemas);
965 10 : out.tables = Some(tables);
966 10 : out.columns = Some(columns);
967 10 : out.databases_by_id = Some(databases_by_id);
968 10 : out.schemas_by_id = Some(schemas_by_id);
969 10 : out.tables_by_id = Some(tables_by_id);
970 10 : out
971 10 : }
972 :
973 : /// A table by id
974 194 : pub fn table_by_id(&self, table_id: CatalogTableId) -> Option<&TableDeclaration> {
975 194 : self.entries.get(&table_id.origin)?.data().table_by_id(table_id)
976 194 : }
977 :
978 : /// A table by location
979 7780 : pub fn table(&self, loc: TableLoc) -> Option<&TableDeclaration> {
980 7780 : self.entries.get(&loc.entry)?.data().table_declarations.get(loc.index as usize)
981 7780 : }
982 :
983 : /// Did the connection report a relation that this script declaration names as gone?
984 28 : fn is_absent(&self, entry: CatalogEntryId, table: &TableDeclaration) -> bool {
985 28 : !self.absent_relations.is_empty()
986 6 : && self.script_entries.contains_key(&entry)
987 5 : && self.absent_relations.iter().any(|absent| same_relation(absent, &table.table_name.key()))
988 28 : }
989 :
990 : /// Is a table the declaration of a relation that is gone?
991 2233 : pub fn is_absent_table(&self, loc: TableLoc) -> bool {
992 2233 : !self.absent_relations.is_empty() && self.table(loc).is_some_and(|table| self.is_absent(loc.entry, table))
993 2233 : }
994 :
995 : /// Is an object of a name index part of a relation that is gone?
996 428 : pub fn is_absent_object(&self, object: CatalogObjectRef) -> bool {
997 428 : match object {
998 132 : CatalogObjectRef::Table(table) => self.is_absent_table(table),
999 229 : CatalogObjectRef::Column(column) => self.is_absent_table(column.table),
1000 67 : _ => false,
1001 : }
1002 428 : }
1003 :
1004 : /// Resolve a table across all entries except `ignore_entry`. The matches of the first layer that has one are
1005 : /// the candidates: what the connection reports doesn't compete with what a script declares, and a script doesn't
1006 : /// provide a relation that the connection reported as gone.
1007 2224 : pub fn resolve_table(&self, name: &QualifiedTableName, ignore_entry: CatalogEntryId, out: &mut Vec<TableLoc>, limit: usize) {
1008 2224 : self.collect_table_matches(name, ignore_entry, out, limit);
1009 2227 : out.retain(|loc| !self.is_absent_table(*loc));
1010 2224 : if let Some(first) = out.first().and_then(|loc| self.layer(loc.entry)) {
1011 1843 : out.retain(|loc| self.layer(loc.entry) == Some(first));
1012 401 : }
1013 2224 : }
1014 :
1015 : /// Collect the matches of a table in resolution order
1016 2224 : fn collect_table_matches(&self, name: &QualifiedTableName, ignore_entry: CatalogEntryId, out: &mut Vec<TableLoc>, limit: usize) {
1017 2224 : let db = name.database_name.text.clone();
1018 2224 : let schema = name.schema_name.text.clone();
1019 : // Direct lookups of the qualified name
1020 2224 : let lower = (db.clone(), schema.clone(), MIN_LAYER, 0);
1021 2224 : let upper = (db.clone(), schema.clone(), MAX_LAYER, CatalogEntryId::MAX);
1022 2224 : let key = name.key();
1023 : // A name with a database is exact. Any other name reads the same in every layer: `t` of a script
1024 : // has no qualification and would hit itself directly, ahead of the `db.schema.t` of the connection.
1025 2224 : if !name.database_name.text.is_empty() {
1026 103 : for ((_, _, _, candidate), _) in self.entries_by_qualified_schema.range(lower..=upper) {
1027 71 : if *candidate == ignore_entry {
1028 0 : continue;
1029 71 : }
1030 71 : let entry = self.entries[candidate].data();
1031 71 : if let Some(&i) = entry.tables_by_qualified_name.get(&key) {
1032 57 : out.push(entry.table_loc(i));
1033 57 : if out.len() >= limit {
1034 0 : break;
1035 57 : }
1036 14 : }
1037 : }
1038 103 : return;
1039 2121 : }
1040 2121 : if !name.schema_name.text.is_empty() {
1041 : // Table and schema name: search the entries that populate the schema in any database
1042 42 : let lower = (schema.clone(), MIN_LAYER, 0);
1043 42 : let upper = (schema.clone(), MAX_LAYER, CatalogEntryId::MAX);
1044 42 : for ((schema_name, _, candidate), _) in self.entries_by_schema.range(lower..=upper) {
1045 6 : if *candidate == ignore_entry {
1046 0 : continue;
1047 6 : }
1048 6 : self.entries[candidate].data().resolve_table_in_schema(schema_name, &name.table_name.text, out, limit);
1049 6 : if out.len() >= limit {
1050 0 : break;
1051 6 : }
1052 : }
1053 : } else {
1054 : // Only the table name: search all entries in resolution order
1055 2374 : for (_, entry_id) in &self.entries_ranked {
1056 2257 : if *entry_id == ignore_entry {
1057 251 : continue;
1058 2006 : }
1059 2006 : self.entries[entry_id].data().resolve_table_everywhere(&name.table_name.text, out, limit);
1060 2006 : if out.len() >= limit {
1061 0 : break;
1062 2006 : }
1063 : }
1064 : }
1065 2224 : }
1066 : }
1067 :
1068 : /// Do two relation names name the same relation? An empty database or schema matches any. An empty table names a whole
1069 : /// schema, which only a relation with that schema is in.
1070 33 : fn same_relation(a: &QualifiedKey, b: &QualifiedKey) -> bool {
1071 36 : let part = |x: &Arc<str>, y: &Arc<str>| x.is_empty() || y.is_empty() || x == y;
1072 33 : if a.2.is_empty() || b.2.is_empty() {
1073 5 : return a.1 == b.1 && part(&a.0, &b.0);
1074 28 : }
1075 28 : a.2 == b.2 && part(&a.0, &b.0) && part(&a.1, &b.1)
1076 33 : }
1077 :
1078 : /// Describe a catalog entry of a layer
1079 4 : fn describe_entry(snapshot: &CatalogSnapshot, layer: CatalogLayer, entry: &CatalogEntryRef) -> dashql_buffers::catalog::CatalogEntryT {
1080 : use dashql_buffers::catalog::{CatalogEntryT, CatalogEntryType, SchemaDescriptorT, SchemaTableColumnT, SchemaTableT};
1081 4 : let data = entry.data();
1082 4 : let mut out = CatalogEntryT::default();
1083 4 : out.catalog_entry_id = data.catalog_entry_id;
1084 4 : out.catalog_entry_type = CatalogEntryType::DESCRIPTOR_POOL;
1085 4 : out.layer = layer;
1086 4 : match entry {
1087 : // Scripts are described as one anonymous schema with all tables
1088 : CatalogEntryRef::Script(_) => {
1089 3 : let tables = data
1090 3 : .table_declarations
1091 3 : .iter()
1092 3 : .enumerate()
1093 4 : .filter(|(_, table)| !snapshot.is_absent(data.catalog_entry_id, table))
1094 4 : .map(|(i, table)| {
1095 4 : let mut t = SchemaTableT::default();
1096 4 : t.table_id = i as u32;
1097 4 : t.table_name = Some(table.table_name.table_name.text.to_string());
1098 4 : t.columns = Some(
1099 4 : table
1100 4 : .table_columns
1101 4 : .iter()
1102 5 : .map(|c| {
1103 5 : let mut column = SchemaTableColumnT::default();
1104 5 : column.column_name = Some(c.column_name.text.to_string());
1105 5 : column
1106 5 : })
1107 4 : .collect(),
1108 : );
1109 4 : t
1110 4 : })
1111 3 : .collect();
1112 3 : let mut schema = SchemaDescriptorT::default();
1113 3 : schema.tables = Some(tables);
1114 3 : out.schemas = Some(vec![schema]);
1115 : }
1116 : CatalogEntryRef::Descriptor(_) => {
1117 1 : let schemas = data
1118 1 : .schemas_by_qualified_name
1119 1 : .iter()
1120 2 : .map(|((db, name), &i)| {
1121 2 : let schema_id = data.schema_references[i].object_id;
1122 2 : let tables = data
1123 2 : .table_declarations
1124 2 : .iter()
1125 4 : .filter(|t| t.catalog_schema_id == schema_id)
1126 2 : .map(|table| {
1127 2 : let mut t = SchemaTableT::default();
1128 2 : t.table_id = table.table_id().object;
1129 2 : t.table_name = Some(table.table_name.table_name.text.to_string());
1130 2 : t.columns = Some(
1131 2 : table
1132 2 : .table_columns
1133 2 : .iter()
1134 2 : .enumerate()
1135 3 : .map(|(i, c)| {
1136 3 : let mut column = SchemaTableColumnT::default();
1137 3 : column.column_name = Some(c.column_name.text.to_string());
1138 3 : column.ordinal_position = i as u32;
1139 3 : column
1140 3 : })
1141 2 : .collect(),
1142 : );
1143 2 : t
1144 2 : })
1145 2 : .collect();
1146 2 : let mut schema = SchemaDescriptorT::default();
1147 2 : schema.database_name = Some(db.to_string());
1148 2 : schema.schema_name = Some(name.to_string());
1149 2 : schema.tables = Some(tables);
1150 2 : schema
1151 2 : })
1152 1 : .collect();
1153 1 : out.schemas = Some(schemas);
1154 : }
1155 : }
1156 4 : out
1157 4 : }
1158 :
1159 : impl CatalogSnapshot {
1160 : /// Describe all entries in resolution order
1161 2 : pub fn describe_entries(&self) -> dashql_buffers::catalog::CatalogEntriesT {
1162 2 : let mut out = dashql_buffers::catalog::CatalogEntriesT::default();
1163 2 : out.entries = Some(self.entries_ranked().map(|(layer, e)| describe_entry(self, layer, e)).collect());
1164 2 : out
1165 2 : }
1166 :
1167 : /// Describe one entry, None if the entry is unknown
1168 3 : pub fn describe_entries_of(&self, id: CatalogEntryId) -> Option<dashql_buffers::catalog::CatalogEntriesT> {
1169 3 : let entry = self.entries.get(&id)?;
1170 2 : let layer = self.layer(id)?;
1171 2 : let mut out = dashql_buffers::catalog::CatalogEntriesT::default();
1172 2 : out.entries = Some(vec![describe_entry(self, layer, entry)]);
1173 2 : Some(out)
1174 3 : }
1175 :
1176 : /// The statistics of the catalog content
1177 3 : pub fn statistics(&self) -> dashql_buffers::catalog::CatalogStatisticsT {
1178 3 : let mut content = dashql_buffers::catalog::CatalogContentStatisticsT::default();
1179 3 : content.database_count = self.databases.len() as u32;
1180 3 : content.schema_count = self.schemas.len() as u32;
1181 3 : content.table_count = self.entries.values().map(|e| e.data().table_declarations.len() as u32).sum();
1182 3 : content.table_column_count = self
1183 3 : .entries
1184 3 : .values()
1185 3 : .flat_map(|e| e.data().table_declarations.iter())
1186 5 : .map(|t| t.table_columns.len() as u32)
1187 3 : .sum();
1188 3 : let mut out = dashql_buffers::catalog::CatalogStatisticsT::default();
1189 3 : out.content = Some(content);
1190 3 : out
1191 3 : }
1192 : }
|
