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

icydb_core/db/session/
catalog.rs
