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

icydb_core/db/session/
catalog.rs
