icydb_core/db/session/
catalog.rs

1//! Module: db::session::catalog
2//! Responsibility: session-owned catalog, schema-description, and storage
3//! observability surfaces.
4//! Does not own: schema reconciliation policy, query planning, or storage
5//! mutation.
6//! Boundary: converts accepted/generated schema authority into stable
7//! introspection DTOs at the session facade.
8
9#[cfg(feature = "sql")]
10use crate::db::schema::SchemaInfo;
11#[cfg(feature = "sql")]
12use crate::db::schema::show_indexes_for_schema_info_with_runtime_state;
13#[cfg(feature = "sql-explain")]
14use crate::db::{IndexState, QueryError, query::plan::VisibleIndexes};
15use crate::{
16    db::{
17        DbSession, EntityCatalogCounts, EntityCatalogDescription, EntityIdentityDescription,
18        EntitySchemaDescription, SchemaApplicationTarget, SchemaChangeJobId, SchemaChangeProgress,
19        SchemaChangeReceipt, StorageReport, StoreCatalogDescription,
20        commit::database_incarnation_id,
21        schema::{
22            AcceptedFieldKind, ConstraintValidationJob, PersistedFieldSnapshot,
23            describe_accepted_entity_with_persisted_schema, describe_accepted_identity,
24        },
25    },
26    error::InternalError,
27    traits::CanisterKind,
28};
29use icydb_schema::{SchemaProposal, SchemaSubmissionKey, TargetDatabaseIdentity};
30
31fn relation_field_count(fields: &[PersistedFieldSnapshot]) -> usize {
32    fields
33        .iter()
34        .filter(|field| persisted_kind_is_relation_field(field.kind()))
35        .count()
36}
37
38fn persisted_kind_is_relation_field(kind: &AcceptedFieldKind) -> bool {
39    match kind {
40        AcceptedFieldKind::Relation { .. } => true,
41        AcceptedFieldKind::List(inner) | AcceptedFieldKind::Set(inner) => {
42            matches!(inner.as_ref(), AcceptedFieldKind::Relation { .. })
43        }
44        _ => false,
45    }
46}
47
48impl<C: CanisterKind> DbSession<C> {
49    /// Apply one exact source-keyed schema proposal through accepted catalog
50    /// authority and return its durable idempotent receipt.
51    pub fn apply_schema(
52        &self,
53        proposal: &SchemaProposal,
54    ) -> Result<SchemaChangeReceipt, InternalError> {
55        crate::db::schema::apply_schema(&self.db, proposal)
56    }
57
58    /// Issue the opaque database/store identities and exact accepted head used
59    /// to compose one optimistic schema proposal.
60    pub fn schema_application_target(&self) -> Result<SchemaApplicationTarget, InternalError> {
61        crate::db::schema::schema_application_target(&self.db)
62    }
63
64    /// Load one durable schema-application receipt by exact target and
65    /// submission identity.
66    pub fn schema_application_receipt(
67        &self,
68        database_identity: TargetDatabaseIdentity,
69        submission_key: &SchemaSubmissionKey,
70    ) -> Result<Option<SchemaChangeReceipt>, InternalError> {
71        crate::db::schema::schema_application_receipt(&self.db, database_identity, submission_key)
72    }
73
74    /// Advance one pending schema application by at most one bounded
75    /// activation step.
76    pub fn continue_schema_application(
77        &self,
78        job_id: SchemaChangeJobId,
79        acknowledged_receipt: Option<u64>,
80    ) -> Result<SchemaChangeProgress, InternalError> {
81        crate::db::schema::continue_schema_application(&self.db, job_id, acknowledged_receipt)
82    }
83
84    /// Abort one pending schema application after acknowledging any retained
85    /// finding page by exact sequence.
86    pub fn abort_schema_application(
87        &self,
88        job_id: SchemaChangeJobId,
89        acknowledged_receipt: Option<u64>,
90    ) -> Result<SchemaChangeProgress, InternalError> {
91        crate::db::schema::abort_schema_application(&self.db, job_id, acknowledged_receipt)
92    }
93
94    // Return one stable, human-readable index listing for one resolved
95    // store/accepted-schema pair, attaching the current runtime lifecycle state
96    // when the registry can resolve the backing store handle.
97    #[cfg(feature = "sql")]
98    pub(in crate::db) fn show_indexes_for_store_schema_info(
99        &self,
100        store_path: &str,
101        schema: &SchemaInfo,
102        snapshot: &crate::db::schema::PersistedSchemaSnapshot,
103    ) -> Vec<String> {
104        let runtime_state = self
105            .db
106            .with_store_registry(|registry| registry.try_get_store(store_path).ok())
107            .map(|store| store.index_state());
108
109        show_indexes_for_schema_info_with_runtime_state(schema, snapshot, runtime_state)
110    }
111
112    /// Return one stable list of accepted runtime entity catalog entries.
113    pub fn show_entities(&self) -> Result<Vec<EntityCatalogDescription>, InternalError> {
114        let runtime_entities = self.db.accepted_runtime_entities()?;
115        let mut entities = Vec::with_capacity(runtime_entities.len());
116
117        for runtime_entity in runtime_entities {
118            let store = self.db.recovered_store(runtime_entity.store_path())?;
119            let storage = store
120                .storage_capabilities()
121                .storage_mode()
122                .as_str()
123                .to_string();
124            let accepted = self.accepted_schema_catalog_context_for_runtime_entity(
125                runtime_entity.clone(),
126                store,
127            )?;
128            let snapshot = accepted.snapshot().persisted_snapshot();
129
130            entities.push(EntityCatalogDescription::new(
131                snapshot.entity_name().to_string(),
132                snapshot.entity_path().to_string(),
133                runtime_entity.store_path().to_string(),
134                storage,
135                EntityCatalogCounts::new(
136                    u32::try_from(snapshot.fields().len()).unwrap_or(u32::MAX),
137                    u32::try_from(snapshot.indexes().len()).unwrap_or(u32::MAX),
138                    u32::try_from(relation_field_count(snapshot.fields())).unwrap_or(u32::MAX),
139                    snapshot.version().get(),
140                ),
141            ));
142        }
143
144        Ok(entities)
145    }
146
147    /// Return one stable list of runtime-registered stores.
148    #[must_use]
149    pub fn show_stores(&self) -> Vec<StoreCatalogDescription> {
150        self.db.runtime_store_catalog()
151    }
152
153    /// Return one stable list of runtime-registered stable-memory allocations.
154    #[must_use]
155    pub fn show_memory(&self) -> Vec<crate::db::MemoryCatalogDescription> {
156        self.db.runtime_memory_catalog()
157    }
158
159    // Resolve the exact secondary-index set that is visible to planner-owned
160    // query planning for one recovered store and accepted schema pair.
161    #[cfg(feature = "sql-explain")]
162    pub(in crate::db::session) fn visible_indexes_for_store_accepted_schema(
163        &self,
164        store_path: &str,
165        schema_info: &SchemaInfo,
166    ) -> Result<VisibleIndexes, QueryError> {
167        // Phase 1: resolve the recovered store state once at the session
168        // boundary so query/executor planning does not reopen lifecycle checks.
169        let store = self
170            .db
171            .recovered_store(store_path)
172            .map_err(QueryError::execute)?;
173        let state = store.index_state();
174        if state != IndexState::Ready {
175            return Ok(VisibleIndexes::none());
176        }
177        debug_assert_eq!(state, IndexState::Ready);
178
179        // Phase 2: planner-visible indexes are accepted schema contracts once
180        // the recovered store is query-visible.
181        let visible_indexes = VisibleIndexes::accepted_schema_visible(schema_info);
182        debug_assert!(visible_indexes.accepted_field_path_contracts_are_consistent());
183        debug_assert!(visible_indexes.accepted_expression_contracts_are_consistent());
184        debug_assert_eq!(
185            visible_indexes.accepted_expression_index_count(),
186            Some(visible_indexes.accepted_expression_indexes().len()),
187        );
188
189        Ok(visible_indexes)
190    }
191
192    /// Return one schema description selected by an immutable authored source key.
193    pub fn try_describe_entity_by_source_key(
194        &self,
195        entity_source: &str,
196    ) -> Result<EntitySchemaDescription, InternalError> {
197        let catalog = self.accepted_schema_catalog_context_for_entity_source_key(entity_source)?;
198        self.describe_accepted_catalog(&catalog)
199    }
200
201    /// Return one schema description selected by its accepted display name.
202    pub fn try_describe_entity_by_name(
203        &self,
204        entity: &str,
205    ) -> Result<EntitySchemaDescription, InternalError> {
206        let catalog = self.accepted_schema_catalog_context_for_entity_name(Some(entity))?;
207        self.describe_accepted_catalog(&catalog)
208    }
209
210    fn describe_accepted_catalog(
211        &self,
212        catalog: &crate::db::session::AcceptedSchemaCatalogContext,
213    ) -> Result<EntitySchemaDescription, InternalError> {
214        let validation_jobs = self.constraint_validation_jobs_for_accepted_catalog(catalog)?;
215        let identity = self.identity_description_for_accepted_catalog(catalog)?;
216
217        describe_accepted_entity_with_persisted_schema(
218            catalog.snapshot(),
219            catalog.value_catalog_handle(),
220            validation_jobs.as_slice(),
221            identity,
222        )
223    }
224
225    pub(in crate::db::session) fn identity_description_for_accepted_catalog(
226        &self,
227        catalog: &crate::db::session::AcceptedSchemaCatalogContext,
228    ) -> Result<Option<EntityIdentityDescription>, InternalError> {
229        let Some(identity) = catalog.inspection_plan().identity_inspection() else {
230            return Ok(None);
231        };
232        let catalog_identity = catalog.identity();
233        let store = self.db.recovered_store(catalog_identity.store_path())?;
234        let incarnation = database_incarnation_id()?;
235        let high_water = store.with_schema(|schema_store| {
236            schema_store.identity_high_water_for_integrity(
237                incarnation,
238                catalog_identity.entity_tag(),
239                identity.field_id(),
240                identity.accepted_kind(),
241            )
242        })?;
243        describe_accepted_identity(identity, high_water).map(Some)
244    }
245
246    pub(in crate::db::session) fn constraint_validation_jobs_for_accepted_catalog(
247        &self,
248        catalog: &crate::db::session::AcceptedSchemaCatalogContext,
249    ) -> Result<Vec<ConstraintValidationJob>, InternalError> {
250        let identity = catalog.inspection_plan().identity();
251        let store = self.db.recovered_store(identity.store_path())?;
252        store.with_schema(|schema_store| {
253            let jobs = catalog
254                .snapshot()
255                .persisted_snapshot()
256                .constraint_activations()
257                .iter()
258                .map(|activation| {
259                    schema_store.constraint_validation_job(identity.entity_tag(), activation.id())
260                })
261                .collect::<Result<Vec<_>, InternalError>>()?;
262            jobs.into_iter()
263                .flatten()
264                .map(|job| {
265                    if job.entity_tag() != identity.entity_tag()
266                        || job.entity_path() != catalog.snapshot().entity_path()
267                    {
268                        return Err(InternalError::store_invariant());
269                    }
270                    Ok(job)
271                })
272                .collect()
273        })
274    }
275
276    /// Build one point-in-time storage report for observability endpoints.
277    pub fn storage_report(
278        &self,
279        name_to_path: &[(&'static str, &'static str)],
280    ) -> Result<StorageReport, InternalError> {
281        self.db.storage_report(name_to_path)
282    }
283
284    /// Build one point-in-time storage report using default entity-path labels.
285    pub fn storage_report_default(&self) -> Result<StorageReport, InternalError> {
286        self.db.storage_report_default()
287    }
288}
icydb_core/db/session/catalog.rs

icydb_core/db/session/
catalog.rs
