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