1use crate::{
7 db::{
8 codec::{
9 finalize_hash_sha256, new_hash_sha256, write_hash_len_u32, write_hash_str_u32,
10 write_hash_tag_u8, write_hash_u32, write_hash_u64,
11 },
12 commit::CommitSchemaFingerprint,
13 direction::Direction,
14 ordered_overlay::{OrderedOverlayEntry, OrderedOverlayVisit, visit_ordered_overlay},
15 schema::{
16 AcceptedFieldKind, AcceptedSchemaSnapshot, ConstraintActivationKind,
17 ConstraintActivationState, ConstraintId, ConstraintOrigin, ConstraintValidationJob,
18 PersistedIndexKeyItemSnapshot, PersistedIndexKeySnapshot, PersistedSchemaSnapshot,
19 SchemaVersion, accepted_constraint_field_paths, accepted_schema_cache_fingerprint,
20 accepted_schema_cache_fingerprint_for_persisted_snapshot,
21 accepted_schema_cache_fingerprint_method_version, decode_constraint_validation_job,
22 decode_persisted_schema_snapshot, encode_constraint_validation_job,
23 encode_persisted_schema_snapshot,
24 enum_catalog::{
25 AcceptedSchemaAuthority, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
26 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
27 AcceptedStoreCatalogScope, AcceptedValueCatalogHandle, CandidateSchemaRevision,
28 decode_verified_accepted_schema_revision_bundle,
29 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
30 },
31 schema_snapshot_integrity_detail,
32 },
33 },
34 error::InternalError,
35 types::EntityTag,
36};
37use ic_stable_structures::{
38 BTreeMap as StableBTreeMap, DefaultMemoryImpl, Storable, memory_manager::VirtualMemory,
39 storable::Bound as StorableBound,
40};
41use sha2::Digest;
42use std::borrow::Cow;
43#[cfg(test)]
44use std::cell::Cell;
45use std::cell::{OnceCell, Ref, RefCell};
46use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
47use std::convert::Infallible;
48use std::ops::Bound as RangeBound;
49use std::rc::Rc;
50
51const SCHEMA_KEY_BYTES_USIZE: usize = 16;
52const SCHEMA_KEY_BYTES: u32 = 16;
53const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
54const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
55const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
56const SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB: u8 = 3;
57const SCHEMA_STORE_FINGERPRINT_METHOD_VERSION: u8 = 1;
60const SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN: u8 = 1;
61const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 2;
62const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 3;
63const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL: u8 = 3;
64const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST: u8 = 29;
65const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET: u8 = 30;
66const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP: u8 = 31;
67const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE: u8 = 32;
68const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBCAT";
69const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
70const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
71
72#[cfg(test)]
73thread_local! {
74 static LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS: Cell<u64> = const { Cell::new(0) };
75 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
76}
77
78#[cfg(test)]
79pub(in crate::db) fn reset_latest_raw_snapshots_by_entity_call_count_for_tests() {
80 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(0));
81}
82
83#[cfg(test)]
84pub(in crate::db) fn latest_raw_snapshots_by_entity_call_count_for_tests() -> u64 {
85 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(Cell::get)
86}
87
88#[cfg(test)]
89fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
90 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
91}
92
93#[cfg(test)]
94fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
95 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
96}
97
98#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
107struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
108
109impl RawSchemaKey {
110 #[must_use]
112 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
113 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
114 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
115 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
116 out[12..].copy_from_slice(&version.get().to_be_bytes());
117
118 Self(out)
119 }
120
121 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
122 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
123 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
124 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
125 Self(out)
126 }
127
128 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
129 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
130 if slot > 1 {
131 return Err(InternalError::store_invariant());
132 }
133 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
134 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
135 out[12..].copy_from_slice(&slot.to_be_bytes());
136 Ok(Self(out))
137 }
138
139 fn from_constraint_validation_job(entity: EntityTag, constraint_id: ConstraintId) -> Self {
140 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
141 out[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
142 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
143 out[12..].copy_from_slice(&constraint_id.get().to_be_bytes());
144 Self(out)
145 }
146
147 #[must_use]
149 fn entity_tag(self) -> EntityTag {
150 let mut bytes = [0u8; size_of::<u64>()];
151 bytes.copy_from_slice(&self.0[4..12]);
152
153 EntityTag::new(u64::from_be_bytes(bytes))
154 }
155
156 #[must_use]
158 fn version(self) -> u32 {
159 let mut bytes = [0u8; size_of::<u32>()];
160 bytes.copy_from_slice(&self.0[12..]);
161
162 u32::from_be_bytes(bytes)
163 }
164
165 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
166 (
167 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::new(0))),
168 RangeBound::Included(Self::from_entity_version(
169 entity,
170 SchemaVersion::new(u32::MAX),
171 )),
172 )
173 }
174
175 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
176 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
177 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
178 (
179 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
180 RangeBound::Included(Self(end)),
181 )
182 }
183
184 const fn all_constraint_validation_job_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
185 let mut start = [0u8; SCHEMA_KEY_BYTES_USIZE];
186 start[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
187 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
188 end[0] = SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB;
189 (
190 RangeBound::Included(Self(start)),
191 RangeBound::Included(Self(end)),
192 )
193 }
194
195 #[cfg(test)]
196 const fn is_entity_snapshot(self) -> bool {
197 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
198 }
199
200 const fn is_accepted_root(self) -> bool {
201 self.0[0] == SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT
202 }
203
204 const fn is_constraint_validation_job(self) -> bool {
205 self.0[0] == SCHEMA_KEY_NAMESPACE_CONSTRAINT_VALIDATION_JOB
206 }
207
208 fn constraint_id(self) -> Option<ConstraintId> {
209 self.is_constraint_validation_job()
210 .then(|| ConstraintId::new(self.version()))
211 .flatten()
212 }
213}
214
215impl Storable for RawSchemaKey {
216 fn to_bytes(&self) -> Cow<'_, [u8]> {
217 Cow::Borrowed(&self.0)
218 }
219
220 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
221 debug_assert_eq!(
222 bytes.len(),
223 SCHEMA_KEY_BYTES_USIZE,
224 "RawSchemaKey::from_bytes received unexpected byte length",
225 );
226
227 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
228 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
229 }
230
231 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
232 out.copy_from_slice(bytes.as_ref());
233 Self(out)
234 }
235
236 fn into_bytes(self) -> Vec<u8> {
237 self.0.to_vec()
238 }
239
240 const BOUND: StorableBound = StorableBound::Bounded {
241 max_size: SCHEMA_KEY_BYTES,
242 is_fixed_size: true,
243 };
244}
245
246#[derive(Clone, Debug, Eq, PartialEq)]
257struct RawSchemaSnapshot {
258 payload: Vec<u8>,
259 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
260}
261
262impl RawSchemaSnapshot {
263 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
265 validate_typed_schema_snapshot_for_store(snapshot)?;
266
267 let accepted_schema_fingerprint =
268 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
269 let payload = encode_persisted_schema_snapshot(snapshot)?;
270
271 Ok(Self {
272 payload,
273 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
274 })
275 }
276
277 #[must_use]
279 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
280 Self {
281 payload,
282 accepted_schema_fingerprint: None,
283 }
284 }
285
286 #[cfg(test)]
289 #[must_use]
290 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
291 Self {
292 payload,
293 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
294 }
295 }
296
297 #[must_use]
299 const fn as_bytes(&self) -> &[u8] {
300 self.payload.as_slice()
301 }
302
303 #[must_use]
305 fn into_bytes(self) -> Vec<u8> {
306 self.payload
307 }
308
309 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
312 self.accepted_schema_fingerprint
313 .ok_or_else(InternalError::store_corruption)
314 }
315
316 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
318 let _fingerprint = self.accepted_schema_fingerprint()?;
321 decode_persisted_schema_snapshot(self.as_bytes())
322 }
323}
324
325#[cfg(test)]
326pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
327 bytes: Vec<u8>,
328) -> Result<(), InternalError> {
329 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
330 raw.decode_persisted_snapshot().map(drop)
331}
332
333#[derive(Clone, Copy, Debug, Eq, PartialEq)]
334pub(in crate::db) struct AcceptedCatalogIdentity {
335 entity_tag: EntityTag,
336 entity_path: &'static str,
337 store_path: &'static str,
338 accepted_schema_revision: AcceptedSchemaRevision,
339 accepted_schema_version: SchemaVersion,
340 fingerprint_method_version: u8,
341 accepted_schema_fingerprint: CommitSchemaFingerprint,
342}
343
344impl AcceptedCatalogIdentity {
345 #[must_use]
346 pub(in crate::db) const fn new(
347 entity_tag: EntityTag,
348 entity_path: &'static str,
349 store_path: &'static str,
350 accepted_schema_revision: AcceptedSchemaRevision,
351 accepted_schema_version: SchemaVersion,
352 accepted_schema_fingerprint: CommitSchemaFingerprint,
353 ) -> Self {
354 Self {
355 entity_tag,
356 entity_path,
357 store_path,
358 accepted_schema_revision,
359 accepted_schema_version,
360 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
361 accepted_schema_fingerprint,
362 }
363 }
364
365 #[must_use]
366 pub(in crate::db) const fn entity_tag(self) -> EntityTag {
367 self.entity_tag
368 }
369
370 #[must_use]
371 pub(in crate::db) const fn entity_path(self) -> &'static str {
372 self.entity_path
373 }
374
375 #[must_use]
376 pub(in crate::db) const fn store_path(self) -> &'static str {
377 self.store_path
378 }
379
380 #[must_use]
381 pub(in crate::db) const fn accepted_schema_revision(self) -> AcceptedSchemaRevision {
382 self.accepted_schema_revision
383 }
384
385 #[must_use]
386 pub(in crate::db) const fn accepted_schema_version(self) -> SchemaVersion {
387 self.accepted_schema_version
388 }
389
390 #[must_use]
391 pub(in crate::db) const fn fingerprint_method_version(self) -> u8 {
392 self.fingerprint_method_version
393 }
394
395 #[must_use]
396 pub(in crate::db) const fn accepted_schema_fingerprint(self) -> CommitSchemaFingerprint {
397 self.accepted_schema_fingerprint
398 }
399}
400
401#[derive(Clone, Debug, Eq, PartialEq)]
402pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
403 identity: AcceptedCatalogIdentity,
404 value_catalog: AcceptedValueCatalogHandle,
405 raw_snapshot: Rc<[u8]>,
406}
407
408impl AcceptedCatalogSnapshotSelection {
409 #[must_use]
410 const fn new(
411 identity: AcceptedCatalogIdentity,
412 value_catalog: AcceptedValueCatalogHandle,
413 raw_snapshot: Rc<[u8]>,
414 ) -> Self {
415 Self {
416 identity,
417 value_catalog,
418 raw_snapshot,
419 }
420 }
421
422 #[must_use]
423 pub(in crate::db) const fn identity(&self) -> AcceptedCatalogIdentity {
424 self.identity
425 }
426
427 #[must_use]
428 pub(in crate::db) const fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
429 &self.value_catalog
430 }
431
432 pub(in crate::db) fn from_accepted_snapshot(
434 identity: AcceptedCatalogIdentity,
435 value_catalog: AcceptedValueCatalogHandle,
436 snapshot: &AcceptedSchemaSnapshot,
437 ) -> Result<Self, InternalError> {
438 let raw_snapshot =
439 RawSchemaSnapshot::from_persisted_snapshot(snapshot.persisted_snapshot())?;
440 if raw_snapshot.accepted_schema_fingerprint()? != identity.accepted_schema_fingerprint() {
441 return Err(InternalError::store_invariant());
442 }
443
444 Ok(Self::new(
445 identity,
446 value_catalog,
447 Rc::from(raw_snapshot.into_bytes()),
448 ))
449 }
450
451 pub(in crate::db) fn from_candidate(
454 candidate: &CandidateSchemaRevision,
455 entity_tag: EntityTag,
456 entity_path: &'static str,
457 store_path: &'static str,
458 ) -> Result<Option<Self>, InternalError> {
459 if candidate.store_path() != store_path {
460 return Err(InternalError::store_corruption());
461 }
462 let Some(snapshot) = candidate.bundle().entity_snapshots().get(&entity_tag) else {
463 return Ok(None);
464 };
465 if snapshot.entity_path() != entity_path {
466 return Err(InternalError::store_corruption());
467 }
468
469 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
470 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
471 let identity = AcceptedCatalogIdentity::new(
472 entity_tag,
473 entity_path,
474 store_path,
475 candidate.revision(),
476 snapshot.version(),
477 fingerprint,
478 );
479
480 Ok(Some(Self::new(
481 identity,
482 AcceptedValueCatalogHandle::new(
483 candidate.bundle().enum_catalog().clone(),
484 candidate.bundle().composite_catalog().clone(),
485 AcceptedStoreCatalogScope::new(),
486 candidate.revision(),
487 candidate.root().fingerprint(),
488 ),
489 Rc::from(raw_snapshot.into_bytes()),
490 )))
491 }
492
493 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
494 let snapshot = decode_persisted_schema_snapshot(self.raw_snapshot.as_ref())?;
495 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
496 let identity = self.identity();
497
498 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
499 return Err(InternalError::store_invariant());
500 }
501 if accepted.entity_path() != identity.entity_path() {
502 return Err(InternalError::store_invariant());
503 }
504
505 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
506 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
507 return Err(InternalError::store_invariant());
508 }
509
510 Ok(accepted)
511 }
512}
513
514impl Storable for RawSchemaSnapshot {
515 fn to_bytes(&self) -> Cow<'_, [u8]> {
516 let Some(fingerprint) = self.accepted_schema_fingerprint else {
517 return Cow::Borrowed(self.as_bytes());
518 };
519
520 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
521 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
522 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
523 bytes.extend_from_slice(&fingerprint);
524 bytes.extend_from_slice(self.as_bytes());
525
526 Cow::Owned(bytes)
527 }
528
529 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
530 let bytes = bytes.into_owned();
531 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
532 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
533 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
534 {
535 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
536 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
537 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
538 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
539
540 return Self {
541 payload: bytes[fingerprint_end..].to_vec(),
542 accepted_schema_fingerprint: Some(fingerprint),
543 };
544 }
545
546 Self {
547 payload: bytes,
548 accepted_schema_fingerprint: None,
549 }
550 }
551
552 fn into_bytes(self) -> Vec<u8> {
553 let Some(fingerprint) = self.accepted_schema_fingerprint else {
554 return self.payload;
555 };
556
557 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
558 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
559 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
560 bytes.extend_from_slice(&fingerprint);
561 bytes.extend_from_slice(&self.payload);
562
563 bytes
564 }
565
566 const BOUND: StorableBound = StorableBound::Unbounded;
567}
568
569fn validate_typed_schema_snapshot_for_store(
573 snapshot: &PersistedSchemaSnapshot,
574) -> Result<(), InternalError> {
575 if schema_snapshot_integrity_detail(
576 "schema snapshot",
577 snapshot.version(),
578 snapshot.primary_key_field_ids(),
579 snapshot.row_layout(),
580 snapshot.fields(),
581 )
582 .is_some()
583 {
584 return Err(InternalError::store_invariant());
585 }
586
587 Ok(())
588}
589
590#[derive(Clone, Copy, Debug, Eq, PartialEq)]
599pub(in crate::db) struct SchemaStoreFootprint {
600 snapshots: u64,
601 encoded_bytes: u64,
602 latest_snapshot_bytes: u64,
603}
604
605#[derive(Clone, Copy, Debug, Eq, PartialEq)]
613pub(in crate::db) struct SchemaStoreCatalogMetadata {
614 schema_version: SchemaVersion,
615 schema_fingerprint_method_version: u8,
616 schema_fingerprint: CommitSchemaFingerprint,
617 entity_count: u64,
618}
619
620impl SchemaStoreCatalogMetadata {
621 #[must_use]
623 const fn new(
624 schema_version: SchemaVersion,
625 schema_fingerprint_method_version: u8,
626 schema_fingerprint: CommitSchemaFingerprint,
627 entity_count: u64,
628 ) -> Self {
629 Self {
630 schema_version,
631 schema_fingerprint_method_version,
632 schema_fingerprint,
633 entity_count,
634 }
635 }
636
637 #[must_use]
639 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
640 self.schema_version
641 }
642
643 #[must_use]
645 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
646 self.schema_fingerprint_method_version
647 }
648
649 #[must_use]
652 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
653 self.schema_fingerprint
654 }
655
656 #[must_use]
658 pub(in crate::db) const fn entity_count(self) -> u64 {
659 self.entity_count
660 }
661}
662
663#[derive(Clone, Copy, Debug, Eq, PartialEq)]
672pub(in crate::db) struct SchemaStoreAllocationMetadata {
673 data: SchemaStoreCatalogMetadata,
674 index: SchemaStoreCatalogMetadata,
675 schema: SchemaStoreCatalogMetadata,
676}
677
678impl SchemaStoreAllocationMetadata {
679 #[must_use]
682 const fn new(
683 data: SchemaStoreCatalogMetadata,
684 index: SchemaStoreCatalogMetadata,
685 schema: SchemaStoreCatalogMetadata,
686 ) -> Self {
687 Self {
688 data,
689 index,
690 schema,
691 }
692 }
693
694 #[must_use]
696 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
697 self.data
698 }
699
700 #[must_use]
702 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
703 self.index
704 }
705
706 #[must_use]
709 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
710 self.schema
711 }
712}
713
714impl SchemaStoreFootprint {
715 #[must_use]
717 const fn new(snapshots: u64, encoded_bytes: u64, latest_snapshot_bytes: u64) -> Self {
718 Self {
719 snapshots,
720 encoded_bytes,
721 latest_snapshot_bytes,
722 }
723 }
724
725 #[must_use]
727 pub(in crate::db) const fn snapshots(self) -> u64 {
728 self.snapshots
729 }
730
731 #[must_use]
733 pub(in crate::db) const fn encoded_bytes(self) -> u64 {
734 self.encoded_bytes
735 }
736
737 #[must_use]
739 pub(in crate::db) const fn latest_snapshot_bytes(self) -> u64 {
740 self.latest_snapshot_bytes
741 }
742}
743
744pub(in crate::db) struct PendingRelationActivationDeleteBarrier {
752 constraint_id: ConstraintId,
753 constraint_name: String,
754 source_entity_path: String,
755 field_paths: Vec<String>,
756}
757
758impl PendingRelationActivationDeleteBarrier {
759 #[must_use]
761 pub(in crate::db) const fn constraint_id(&self) -> ConstraintId {
762 self.constraint_id
763 }
764
765 #[must_use]
767 pub(in crate::db) const fn constraint_name(&self) -> &str {
768 self.constraint_name.as_str()
769 }
770
771 #[must_use]
773 pub(in crate::db) const fn source_entity_path(&self) -> &str {
774 self.source_entity_path.as_str()
775 }
776
777 #[must_use]
779 pub(in crate::db) const fn field_paths(&self) -> &[String] {
780 self.field_paths.as_slice()
781 }
782}
783
784pub struct SchemaStore {
793 backend: SchemaStoreBackend,
794 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
795 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
796}
797
798struct AcceptedSchemaBundleCache {
799 selection: AcceptedSchemaRootSelection,
800 bundle: AcceptedSchemaRevisionBundle,
801 value_catalog: AcceptedValueCatalogHandle,
802 entity_selections: RefCell<StdBTreeMap<EntityTag, AcceptedCatalogSnapshotSelection>>,
803}
804
805enum SchemaStoreBackend {
806 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
807 Journaled {
808 canonical:
809 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
810 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
811 tombstones: BTreeSet<RawSchemaKey>,
812 },
813}
814
815#[derive(Clone, Copy, Debug, Eq, PartialEq)]
817enum SchemaStoreVisit {
818 Continue,
819 Stop,
820}
821
822impl SchemaStoreVisit {
823 const fn should_stop(self) -> bool {
824 matches!(self, Self::Stop)
825 }
826}
827
828impl SchemaStore {
829 #[must_use]
831 pub const fn init_heap() -> Self {
832 Self {
833 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
834 accepted_bundle_cache: RefCell::new(None),
835 accepted_catalog_scope: OnceCell::new(),
836 }
837 }
838
839 #[must_use]
844 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
845 Self {
846 backend: SchemaStoreBackend::Journaled {
847 canonical: StableBTreeMap::init(memory),
848 live: StdBTreeMap::new(),
849 tombstones: BTreeSet::new(),
850 },
851 accepted_bundle_cache: RefCell::new(None),
852 accepted_catalog_scope: OnceCell::new(),
853 }
854 }
855
856 pub(in crate::db) fn insert_persisted_snapshot(
858 &mut self,
859 entity: EntityTag,
860 snapshot: &PersistedSchemaSnapshot,
861 ) -> Result<(), InternalError> {
862 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
863 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
864 let _ = self.insert_raw_snapshot(key, raw_snapshot);
865
866 Ok(())
867 }
868
869 pub(in crate::db) fn constraint_validation_job(
871 &self,
872 entity: EntityTag,
873 constraint_id: ConstraintId,
874 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
875 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
876 self.get_raw_snapshot(&key)
877 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
878 .transpose()
879 }
880
881 pub(in crate::db) fn apply_constraint_validation_job(
883 &mut self,
884 job: &ConstraintValidationJob,
885 ) -> Result<(), InternalError> {
886 let key =
887 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
888 let bytes = encode_constraint_validation_job(job)?;
889 let _ =
890 self.insert_raw_snapshot(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
891 Ok(())
892 }
893
894 #[expect(
896 clippy::unnecessary_wraps,
897 reason = "marker apply operations share one fallible callback contract"
898 )]
899 pub(in crate::db) fn apply_constraint_validation_job_removal(
900 &mut self,
901 entity: EntityTag,
902 constraint_id: ConstraintId,
903 ) -> Result<(), InternalError> {
904 let key = RawSchemaKey::from_constraint_validation_job(entity, constraint_id);
905 match &mut self.backend {
906 SchemaStoreBackend::Heap(map) => {
907 map.remove(&key);
908 }
909 SchemaStoreBackend::Journaled {
910 live, tombstones, ..
911 } => {
912 live.remove(&key);
913 tombstones.insert(key);
914 }
915 }
916 Ok(())
917 }
918
919 pub(in crate::db) fn fold_constraint_validation_job(
921 &mut self,
922 job: &ConstraintValidationJob,
923 ) -> Result<(), InternalError> {
924 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
925 return Err(InternalError::store_invariant());
926 };
927 let key =
928 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id());
929 let bytes = encode_constraint_validation_job(job)?;
930 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
931 Ok(())
932 }
933
934 pub(in crate::db) fn fold_constraint_validation_job_removal(
936 &mut self,
937 entity: EntityTag,
938 constraint_id: ConstraintId,
939 ) -> Result<(), InternalError> {
940 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
941 return Err(InternalError::store_invariant());
942 };
943 canonical.remove(&RawSchemaKey::from_constraint_validation_job(
944 entity,
945 constraint_id,
946 ));
947 Ok(())
948 }
949
950 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
953 let SchemaStoreBackend::Journaled {
954 live, tombstones, ..
955 } = &mut self.backend
956 else {
957 return Err(InternalError::store_invariant());
958 };
959
960 live.clear();
961 tombstones.clear();
962 self.accepted_bundle_cache.get_mut().take();
963
964 Ok(())
965 }
966
967 pub(in crate::db) fn fold_persisted_snapshot(
969 &mut self,
970 entity: EntityTag,
971 snapshot: &PersistedSchemaSnapshot,
972 ) -> Result<(), InternalError> {
973 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
974 return Err(InternalError::store_invariant());
975 };
976
977 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
978 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
979 canonical.insert(key, raw_snapshot);
980
981 Ok(())
982 }
983
984 pub(in crate::db) fn current_accepted_schema_root(
986 &self,
987 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
988 let first = self.accepted_root_slot_bytes(0)?;
989 let second = self.accepted_root_slot_bytes(1)?;
990 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
991 }
992
993 pub(in crate::db) fn current_accepted_schema_bundle(
995 &self,
996 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
997 let Some(selection) = self.current_accepted_schema_root()? else {
998 return Ok(None);
999 };
1000 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1001 let raw = self
1002 .get_raw_snapshot(&key)
1003 .ok_or_else(InternalError::store_corruption)?;
1004 let bundle =
1005 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
1006 self.validate_constraint_validation_job_closure(&bundle)?;
1007 Ok(Some(bundle))
1008 }
1009
1010 pub(in crate::db) fn current_accepted_schema_revision(
1012 &self,
1013 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
1014 Ok(self
1015 .current_accepted_schema_root()?
1016 .map(|selection| selection.root().revision()))
1017 }
1018
1019 pub(in crate::db) fn pending_relation_activation_for_target(
1025 &self,
1026 target_path: &str,
1027 ) -> Result<Option<PendingRelationActivationDeleteBarrier>, InternalError> {
1028 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1029 return Ok(None);
1030 };
1031 for snapshot in bundle.entity_snapshots().values() {
1032 let Some(candidate) = snapshot
1033 .candidate_relations()
1034 .iter()
1035 .find(|candidate| candidate.target_path() == target_path)
1036 else {
1037 continue;
1038 };
1039 let activation = snapshot
1040 .constraint_activations()
1041 .iter()
1042 .find(|activation| {
1043 matches!(
1044 activation.kind(),
1045 ConstraintActivationKind::Relation { relation_id }
1046 if *relation_id == candidate.id()
1047 )
1048 })
1049 .ok_or_else(InternalError::store_corruption)?;
1050 return Ok(Some(PendingRelationActivationDeleteBarrier {
1051 constraint_id: activation.id(),
1052 constraint_name: activation.name().to_string(),
1053 source_entity_path: snapshot.entity_path().to_string(),
1054 field_paths: accepted_constraint_field_paths(
1055 snapshot,
1056 candidate.local_field_ids(),
1057 )?,
1058 }));
1059 }
1060
1061 Ok(None)
1062 }
1063
1064 pub(in crate::db) fn entity_has_relation_to_target(
1066 &self,
1067 source_entity: EntityTag,
1068 target_path: &str,
1069 ) -> Result<bool, InternalError> {
1070 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1071 return Ok(false);
1072 };
1073 let Some(snapshot) = bundle.entity_snapshots().get(&source_entity) else {
1074 return Ok(false);
1075 };
1076
1077 Ok(snapshot
1078 .relations()
1079 .iter()
1080 .any(|relation| relation.target_path() == target_path))
1081 }
1082
1083 pub(in crate::db) fn validate_live_activation_transition(
1085 &self,
1086 candidate: &AcceptedSchemaRevisionBundle,
1087 ) -> Result<(), InternalError> {
1088 let Some(current) = self.current_accepted_schema_bundle()? else {
1089 return Ok(());
1090 };
1091 for (entity_tag, before) in current.entity_snapshots() {
1092 if before.constraint_activations().is_empty() {
1093 continue;
1094 }
1095 let after = candidate
1096 .entity_snapshots()
1097 .get(entity_tag)
1098 .ok_or_else(InternalError::store_invariant)?;
1099 if before == after {
1100 continue;
1101 }
1102 let expected_shape = before
1103 .clone()
1104 .with_constraint_catalog(after.constraint_catalog().clone());
1105 let catalog_only_transition = expected_shape == *after
1106 && before
1107 .constraint_catalog()
1108 .permits_live_activation_transition_to(after.constraint_catalog());
1109 let sql_row_local_abort_with_version =
1110 before.constraint_activations().iter().any(|activation| {
1111 activation.origin() == ConstraintOrigin::SqlDdl
1112 && matches!(
1113 activation.kind(),
1114 ConstraintActivationKind::Check { .. }
1115 | ConstraintActivationKind::NotNull { .. }
1116 )
1117 && before.version().get().checked_add(1) == Some(after.version().get())
1118 && before
1119 .constraint_catalog()
1120 .clone()
1121 .with_aborted_activation(activation.id())
1122 .is_ok_and(|catalog| catalog == *after.constraint_catalog())
1123 && before
1124 .clone()
1125 .with_constraint_catalog(after.constraint_catalog().clone())
1126 .with_schema_version(after.version())
1127 == *after
1128 });
1129 let sql_unique_abort_with_version =
1130 before.constraint_activations().iter().any(|activation| {
1131 activation.origin() == ConstraintOrigin::SqlDdl
1132 && matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1133 && before.version().get().checked_add(1) == Some(after.version().get())
1134 && before
1135 .with_aborted_unique_activation(activation.id(), after.version())
1136 .is_ok_and(|expected| expected == *after)
1137 });
1138 let not_null_promotion = before.constraint_activations().iter().any(|activation| {
1139 matches!(activation.kind(), ConstraintActivationKind::NotNull { .. })
1140 && before
1141 .with_promoted_not_null_activation(activation.id(), after.version())
1142 .is_ok_and(|expected| expected == *after)
1143 });
1144 let unique_promotion = before.constraint_activations().iter().any(|activation| {
1145 matches!(activation.kind(), ConstraintActivationKind::Unique { .. })
1146 && before
1147 .with_promoted_unique_activation(activation.id(), after.version())
1148 .is_ok_and(|expected| expected == *after)
1149 });
1150 let relation_promotion = before.constraint_activations().iter().any(|activation| {
1151 matches!(activation.kind(), ConstraintActivationKind::Relation { .. })
1152 && before
1153 .with_promoted_relation_activation(activation.id(), after.version())
1154 .is_ok_and(|expected| expected == *after)
1155 });
1156 if !catalog_only_transition
1157 && !sql_row_local_abort_with_version
1158 && !sql_unique_abort_with_version
1159 && !not_null_promotion
1160 && !unique_promotion
1161 && !relation_promotion
1162 {
1163 return Err(InternalError::store_invariant());
1164 }
1165 }
1166 Ok(())
1167 }
1168
1169 pub(in crate::db) fn validate_constraint_validation_job_closure(
1171 &self,
1172 bundle: &AcceptedSchemaRevisionBundle,
1173 ) -> Result<(), InternalError> {
1174 self.validate_constraint_validation_job_closure_with_change(bundle, None, None)
1175 }
1176
1177 pub(in crate::db) fn validate_constraint_validation_job_closure_with_change(
1180 &self,
1181 bundle: &AcceptedSchemaRevisionBundle,
1182 replacement: Option<&ConstraintValidationJob>,
1183 removal: Option<(EntityTag, ConstraintId)>,
1184 ) -> Result<(), InternalError> {
1185 if replacement.is_some() && removal.is_some() {
1186 return Err(InternalError::store_invariant());
1187 }
1188 let replacement_key = replacement.map(|job| {
1189 RawSchemaKey::from_constraint_validation_job(job.entity_tag(), job.constraint_id())
1190 });
1191 let removal_key = removal.map(|(entity_tag, constraint_id)| {
1192 RawSchemaKey::from_constraint_validation_job(entity_tag, constraint_id)
1193 });
1194 let mut expected = BTreeSet::new();
1195 for (entity_tag, snapshot) in bundle.entity_snapshots() {
1196 for activation in snapshot.constraint_activations() {
1197 let key =
1198 RawSchemaKey::from_constraint_validation_job(*entity_tag, activation.id());
1199 match activation.state() {
1200 ConstraintActivationState::EnforcingNewWrites => {
1201 if self
1202 .constraint_validation_job_after_change(
1203 key,
1204 replacement,
1205 replacement_key,
1206 removal_key,
1207 )?
1208 .is_some()
1209 {
1210 return Err(InternalError::store_corruption());
1211 }
1212 }
1213 ConstraintActivationState::Validating => {
1214 let job = self
1215 .constraint_validation_job_after_change(
1216 key,
1217 replacement,
1218 replacement_key,
1219 removal_key,
1220 )?
1221 .ok_or_else(InternalError::store_corruption)?;
1222 if job.entity_tag() != *entity_tag
1223 || job.entity_path() != snapshot.entity_path()
1224 {
1225 return Err(InternalError::store_corruption());
1226 }
1227 job.validate(Some(activation))?;
1228 expected.insert(key);
1229 }
1230 }
1231 }
1232 }
1233
1234 self.visit_constraint_validation_jobs(|key, raw| {
1235 if removal_key == Some(*key) || replacement_key == Some(*key) {
1236 return Ok(SchemaStoreVisit::Continue);
1237 }
1238 if !expected.contains(key) {
1239 return Err(InternalError::store_corruption());
1240 }
1241 let job = decode_constraint_validation_job(raw.as_bytes())?;
1242 if job.entity_tag() != key.entity_tag()
1243 || key.constraint_id() != Some(job.constraint_id())
1244 {
1245 return Err(InternalError::store_corruption());
1246 }
1247 Ok(SchemaStoreVisit::Continue)
1248 })?;
1249
1250 if let Some(key) = replacement_key
1251 && !expected.contains(&key)
1252 {
1253 return Err(InternalError::store_corruption());
1254 }
1255 if let Some(key) = removal_key
1256 && expected.contains(&key)
1257 {
1258 return Err(InternalError::store_corruption());
1259 }
1260
1261 Ok(())
1262 }
1263
1264 fn constraint_validation_job_after_change(
1265 &self,
1266 key: RawSchemaKey,
1267 replacement: Option<&ConstraintValidationJob>,
1268 replacement_key: Option<RawSchemaKey>,
1269 removal_key: Option<RawSchemaKey>,
1270 ) -> Result<Option<ConstraintValidationJob>, InternalError> {
1271 if removal_key == Some(key) {
1272 return Ok(None);
1273 }
1274 if replacement_key == Some(key) {
1275 return Ok(replacement.cloned());
1276 }
1277 self.get_raw_snapshot(&key)
1278 .map(|raw| decode_constraint_validation_job(raw.as_bytes()))
1279 .transpose()
1280 }
1281
1282 pub(in crate::db) fn current_accepted_schema_authority_matches(
1285 &self,
1286 expected: &AcceptedSchemaAuthority,
1287 ) -> Result<bool, InternalError> {
1288 let Some(store_scope) = self.accepted_catalog_scope.get() else {
1289 return Ok(false);
1290 };
1291
1292 if let Some(cached) = self
1295 .accepted_bundle_cache
1296 .try_borrow()
1297 .map_err(|_| InternalError::store_invariant())?
1298 .as_ref()
1299 {
1300 let root = cached.selection.root();
1301 return Ok(expected.matches_store_root(
1302 store_scope,
1303 root.revision(),
1304 root.fingerprint(),
1305 ));
1306 }
1307
1308 let Some(selection) = self.current_accepted_schema_root()? else {
1309 return Ok(false);
1310 };
1311 let root = selection.root();
1312
1313 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
1314 }
1315
1316 pub(in crate::db) fn publish_accepted_schema_candidate(
1320 &mut self,
1321 expected_revision: AcceptedSchemaRevision,
1322 candidate: &CandidateSchemaRevision,
1323 ) -> Result<(), InternalError> {
1324 if self.current_root_matches_candidate(candidate)? {
1325 let selection = self
1326 .current_accepted_schema_root()?
1327 .ok_or_else(InternalError::store_corruption)?;
1328 self.retain_durable_candidate_entries(candidate, selection.slot())?;
1329 return Ok(());
1330 }
1331 let first = self.accepted_root_slot_bytes(0)?;
1332 let second = self.accepted_root_slot_bytes(1)?;
1333 prepare_accepted_schema_root_publication(
1334 [first.as_deref(), second.as_deref()],
1335 expected_revision,
1336 candidate,
1337 )
1338 .map_err(map_schema_publication_error)?;
1339
1340 self.insert_durable_candidate_snapshots(candidate)?;
1341 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1342 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
1343 let persisted_bundle = self
1344 .get_raw_snapshot(&bundle_key)
1345 .ok_or_else(InternalError::store_corruption)?;
1346 let _verified = decode_verified_accepted_schema_revision_bundle(
1347 candidate.root(),
1348 persisted_bundle.as_bytes(),
1349 )?;
1350
1351 let first = self.accepted_root_slot_bytes(0)?;
1354 let second = self.accepted_root_slot_bytes(1)?;
1355 let publication = prepare_accepted_schema_root_publication(
1356 [first.as_deref(), second.as_deref()],
1357 expected_revision,
1358 candidate,
1359 )
1360 .map_err(map_schema_publication_error)?;
1361 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1362 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
1363
1364 let selected = self
1365 .current_accepted_schema_root()?
1366 .ok_or_else(InternalError::store_corruption)?;
1367 if selected.root() != candidate.root() {
1368 return Err(InternalError::store_corruption());
1369 }
1370 self.retain_durable_candidate_entries(candidate, selected.slot())?;
1371 Ok(())
1372 }
1373
1374 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
1376 &mut self,
1377 expected_revision: AcceptedSchemaRevision,
1378 candidate: &CandidateSchemaRevision,
1379 ) -> Result<(), InternalError> {
1380 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1381 return Err(InternalError::store_invariant());
1382 }
1383 if self.current_root_matches_candidate(candidate)? {
1384 let selection = self
1385 .current_accepted_schema_root()?
1386 .ok_or_else(InternalError::store_corruption)?;
1387 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
1388 return Ok(());
1389 }
1390
1391 let first = self.accepted_root_slot_bytes(0)?;
1392 let second = self.accepted_root_slot_bytes(1)?;
1393 prepare_accepted_schema_root_publication(
1394 [first.as_deref(), second.as_deref()],
1395 expected_revision,
1396 candidate,
1397 )
1398 .map_err(map_schema_publication_error)?;
1399
1400 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1401 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
1402 }
1403 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1404 self.insert_raw_snapshot(
1405 bundle_key,
1406 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
1407 );
1408 let persisted_bundle = self
1409 .get_raw_snapshot(&bundle_key)
1410 .ok_or_else(InternalError::store_corruption)?;
1411 let _verified = decode_verified_accepted_schema_revision_bundle(
1412 candidate.root(),
1413 persisted_bundle.as_bytes(),
1414 )?;
1415
1416 let first = self.accepted_root_slot_bytes(0)?;
1417 let second = self.accepted_root_slot_bytes(1)?;
1418 let publication = prepare_accepted_schema_root_publication(
1419 [first.as_deref(), second.as_deref()],
1420 expected_revision,
1421 candidate,
1422 )
1423 .map_err(map_schema_publication_error)?;
1424 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1425 self.insert_raw_snapshot(
1426 root_key,
1427 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
1428 );
1429
1430 if !self.current_root_matches_candidate(candidate)? {
1431 return Err(InternalError::store_corruption());
1432 }
1433 let selection = self
1434 .current_accepted_schema_root()?
1435 .ok_or_else(InternalError::store_corruption)?;
1436 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
1437 Ok(())
1438 }
1439
1440 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
1442 &mut self,
1443 expected_revision: AcceptedSchemaRevision,
1444 candidate: &CandidateSchemaRevision,
1445 ) -> Result<(), InternalError> {
1446 if self.canonical_root_matches_candidate(candidate)? {
1447 let first = self.canonical_root_slot_bytes(0)?;
1448 let second = self.canonical_root_slot_bytes(1)?;
1449 let selection =
1450 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1451 .ok_or_else(InternalError::store_corruption)?;
1452 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
1453 return Ok(());
1454 }
1455
1456 let first = self.canonical_root_slot_bytes(0)?;
1457 let second = self.canonical_root_slot_bytes(1)?;
1458 prepare_accepted_schema_root_publication(
1459 [first.as_deref(), second.as_deref()],
1460 expected_revision,
1461 candidate,
1462 )
1463 .map_err(map_schema_publication_error)?;
1464
1465 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1466 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
1467 }
1468 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1469 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
1470 let persisted_bundle = self
1471 .get_canonical_raw_value(&bundle_key)?
1472 .ok_or_else(InternalError::store_corruption)?;
1473 let _verified = decode_verified_accepted_schema_revision_bundle(
1474 candidate.root(),
1475 persisted_bundle.as_bytes(),
1476 )?;
1477
1478 let first = self.canonical_root_slot_bytes(0)?;
1479 let second = self.canonical_root_slot_bytes(1)?;
1480 let publication = prepare_accepted_schema_root_publication(
1481 [first.as_deref(), second.as_deref()],
1482 expected_revision,
1483 candidate,
1484 )
1485 .map_err(map_schema_publication_error)?;
1486 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1487 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
1488
1489 if !self.canonical_root_matches_candidate(candidate)? {
1490 return Err(InternalError::store_corruption());
1491 }
1492 let first = self.canonical_root_slot_bytes(0)?;
1493 let second = self.canonical_root_slot_bytes(1)?;
1494 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1495 .ok_or_else(InternalError::store_corruption)?;
1496 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
1497 Ok(())
1498 }
1499
1500 #[cfg(test)]
1502 pub(in crate::db) fn get_persisted_snapshot(
1503 &self,
1504 entity: EntityTag,
1505 version: SchemaVersion,
1506 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1507 let key = RawSchemaKey::from_entity_version(entity, version);
1508 self.get_raw_snapshot(&key)
1509 .map(|snapshot| snapshot.decode_persisted_snapshot())
1510 .transpose()
1511 }
1512
1513 pub(in crate::db) fn latest_staged_persisted_snapshot(
1518 &self,
1519 entity: EntityTag,
1520 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1521 self.latest_raw_snapshot(entity)
1522 .map(|snapshot| snapshot.decode_persisted_snapshot())
1523 .transpose()
1524 }
1525
1526 #[cfg(any(test, feature = "sql"))]
1529 pub(in crate::db) fn current_accepted_persisted_snapshot(
1530 &self,
1531 entity: EntityTag,
1532 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1533 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1534 return Ok(None);
1535 };
1536
1537 Ok(bundle.entity_snapshots().get(&entity).cloned())
1538 }
1539
1540 pub(in crate::db) fn current_accepted_catalog_selection(
1542 &self,
1543 entity: EntityTag,
1544 entity_path: &'static str,
1545 store_path: &'static str,
1546 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
1547 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1548 return Ok(None);
1549 };
1550 if bundle.store_path() != store_path {
1551 return Err(InternalError::store_corruption());
1552 }
1553 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
1554 return Ok(None);
1555 };
1556 if snapshot.entity_path() != entity_path {
1557 return Err(InternalError::store_corruption());
1558 }
1559
1560 let cache = self
1561 .accepted_bundle_cache
1562 .try_borrow()
1563 .map_err(|_| InternalError::store_invariant())?;
1564 let cached = cache.as_ref().ok_or_else(InternalError::store_invariant)?;
1565 if let Some(selection) = cached
1566 .entity_selections
1567 .try_borrow()
1568 .map_err(|_| InternalError::store_invariant())?
1569 .get(&entity)
1570 .cloned()
1571 {
1572 return Ok(Some(selection));
1573 }
1574
1575 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1576 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
1577 let identity = AcceptedCatalogIdentity::new(
1578 entity,
1579 entity_path,
1580 store_path,
1581 bundle.revision(),
1582 snapshot.version(),
1583 fingerprint,
1584 );
1585
1586 let selected = AcceptedCatalogSnapshotSelection::new(
1587 identity,
1588 cached.value_catalog.clone(),
1589 Rc::from(raw_snapshot.into_bytes()),
1590 );
1591 cached
1592 .entity_selections
1593 .try_borrow_mut()
1594 .map_err(|_| InternalError::store_invariant())?
1595 .insert(entity, selected.clone());
1596
1597 Ok(Some(selected))
1598 }
1599
1600 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
1604 &self,
1605 entity: EntityTag,
1606 entity_path: &'static str,
1607 store_path: &'static str,
1608 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
1609 let first = self.canonical_root_slot_bytes(0)?;
1610 let second = self.canonical_root_slot_bytes(1)?;
1611 let Some(selection) =
1612 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1613 else {
1614 return Ok(None);
1615 };
1616 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1617 let raw_bundle = self
1618 .get_canonical_raw_value(&bundle_key)?
1619 .ok_or_else(InternalError::store_corruption)?;
1620 let bundle = decode_verified_accepted_schema_revision_bundle(
1621 selection.root(),
1622 raw_bundle.as_bytes(),
1623 )?;
1624 if bundle.store_path() != store_path {
1625 return Err(InternalError::store_corruption());
1626 }
1627 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
1628 return Ok(None);
1629 };
1630 if snapshot.entity_path() != entity_path {
1631 return Err(InternalError::store_corruption());
1632 }
1633
1634 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1635 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
1636 let identity = AcceptedCatalogIdentity::new(
1637 entity,
1638 entity_path,
1639 store_path,
1640 bundle.revision(),
1641 snapshot.version(),
1642 fingerprint,
1643 );
1644
1645 Ok(Some(AcceptedCatalogSnapshotSelection::new(
1646 identity,
1647 AcceptedValueCatalogHandle::new(
1648 bundle.enum_catalog().clone(),
1649 bundle.composite_catalog().clone(),
1650 self.accepted_catalog_scope
1651 .get_or_init(AcceptedStoreCatalogScope::new)
1652 .clone(),
1653 bundle.revision(),
1654 selection.root().fingerprint(),
1655 ),
1656 Rc::from(raw_snapshot.into_bytes()),
1657 )))
1658 }
1659
1660 #[must_use]
1662 pub(in crate::db) fn entity_footprint(&self, entity: EntityTag) -> SchemaStoreFootprint {
1663 let mut snapshots = 0u64;
1664 let mut encoded_bytes = 0u64;
1665 let mut latest = None::<(SchemaVersion, u64)>;
1666
1667 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
1668 if key.entity_tag() != entity {
1669 return Ok(SchemaStoreVisit::Continue);
1670 }
1671
1672 let snapshot_bytes = u64::try_from(snapshot.as_bytes().len()).unwrap_or(u64::MAX);
1673 snapshots = snapshots.saturating_add(1);
1674 encoded_bytes = encoded_bytes.saturating_add(snapshot_bytes);
1675
1676 let version = SchemaVersion::new(key.version());
1677 if latest
1678 .as_ref()
1679 .is_none_or(|(latest_version, _)| version > *latest_version)
1680 {
1681 latest = Some((version, snapshot_bytes));
1682 }
1683 Ok(SchemaStoreVisit::Continue)
1684 });
1685
1686 SchemaStoreFootprint::new(
1687 snapshots,
1688 encoded_bytes,
1689 latest.map_or(0, |(_, snapshot_bytes)| snapshot_bytes),
1690 )
1691 }
1692
1693 #[cfg(test)]
1699 pub(in crate::db) fn catalog_metadata(
1700 &self,
1701 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
1702 Ok(self
1703 .allocation_metadata()?
1704 .map(SchemaStoreAllocationMetadata::schema))
1705 }
1706
1707 pub(in crate::db) fn allocation_metadata(
1714 &self,
1715 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
1716 let latest_by_entity = self.latest_raw_snapshots_by_entity();
1717 if latest_by_entity.is_empty() {
1718 return Ok(None);
1719 }
1720
1721 Ok(Some(SchemaStoreAllocationMetadata::new(
1722 derive_data_allocation_metadata(&latest_by_entity)?,
1723 derive_index_allocation_metadata(&latest_by_entity)?,
1724 derive_schema_catalog_metadata(&latest_by_entity)?,
1725 )))
1726 }
1727
1728 fn insert_raw_snapshot(
1730 &mut self,
1731 key: RawSchemaKey,
1732 snapshot: RawSchemaSnapshot,
1733 ) -> Option<RawSchemaSnapshot> {
1734 self.invalidate_accepted_bundle_cache_for_key(key);
1735 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1736 self.get_raw_snapshot_for_backend(&key)
1737 } else {
1738 None
1739 };
1740 match &mut self.backend {
1741 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
1742 SchemaStoreBackend::Journaled {
1743 live, tombstones, ..
1744 } => {
1745 tombstones.remove(&key);
1746 live.insert(key, snapshot);
1747 previous_journaled
1748 }
1749 }
1750 }
1751
1752 #[must_use]
1754 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
1755 match &self.backend {
1756 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
1757 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
1758 }
1759 }
1760
1761 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1762 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1763 Ok(self
1764 .get_raw_snapshot(&key)
1765 .map(RawSchemaSnapshot::into_bytes))
1766 }
1767
1768 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1769 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1770 Ok(self
1771 .get_canonical_raw_value(&key)?
1772 .map(RawSchemaSnapshot::into_bytes))
1773 }
1774
1775 fn current_root_matches_candidate(
1776 &self,
1777 candidate: &CandidateSchemaRevision,
1778 ) -> Result<bool, InternalError> {
1779 let Some(selection) = self.current_accepted_schema_root()? else {
1780 return Ok(false);
1781 };
1782 if selection.root() != candidate.root() {
1783 return Ok(false);
1784 }
1785 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1786 let bundle = self
1787 .get_raw_snapshot(&key)
1788 .ok_or_else(InternalError::store_corruption)?;
1789 let _verified =
1790 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1791 Ok(true)
1792 }
1793
1794 fn canonical_root_matches_candidate(
1795 &self,
1796 candidate: &CandidateSchemaRevision,
1797 ) -> Result<bool, InternalError> {
1798 let first = self.canonical_root_slot_bytes(0)?;
1799 let second = self.canonical_root_slot_bytes(1)?;
1800 let Some(selection) =
1801 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1802 else {
1803 return Ok(false);
1804 };
1805 if selection.root() != candidate.root() {
1806 return Ok(false);
1807 }
1808 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1809 let bundle = self
1810 .get_canonical_raw_value(&key)?
1811 .ok_or_else(InternalError::store_corruption)?;
1812 let _verified =
1813 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1814 Ok(true)
1815 }
1816
1817 fn get_canonical_raw_value(
1818 &self,
1819 key: &RawSchemaKey,
1820 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
1821 match &self.backend {
1822 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
1823 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1824 }
1825 }
1826
1827 fn insert_canonical_raw_value(
1828 &mut self,
1829 key: RawSchemaKey,
1830 bytes: Vec<u8>,
1831 ) -> Result<(), InternalError> {
1832 self.invalidate_accepted_bundle_cache_for_key(key);
1833 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1834 return Err(InternalError::store_invariant());
1835 };
1836 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1837 Ok(())
1838 }
1839
1840 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
1844 self.invalidate_accepted_bundle_cache_for_key(key);
1845 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
1846 match &mut self.backend {
1847 SchemaStoreBackend::Heap(map) => {
1848 map.insert(key, value);
1849 }
1850 SchemaStoreBackend::Journaled {
1851 canonical,
1852 live,
1853 tombstones,
1854 } => {
1855 live.remove(&key);
1856 tombstones.remove(&key);
1857 canonical.insert(key, value);
1858 }
1859 }
1860 }
1861
1862 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
1863 if key.is_accepted_root() {
1864 self.accepted_bundle_cache.get_mut().take();
1865 }
1866 }
1867
1868 fn insert_durable_candidate_snapshots(
1869 &mut self,
1870 candidate: &CandidateSchemaRevision,
1871 ) -> Result<(), InternalError> {
1872 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1873 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
1874 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1875 match &mut self.backend {
1876 SchemaStoreBackend::Heap(map) => {
1877 map.insert(key, value);
1878 }
1879 SchemaStoreBackend::Journaled {
1880 canonical,
1881 live,
1882 tombstones,
1883 } => {
1884 live.remove(&key);
1885 tombstones.remove(&key);
1886 canonical.insert(key, value);
1887 }
1888 }
1889 }
1890 Ok(())
1891 }
1892
1893 fn candidate_entry_keys(
1894 candidate: &CandidateSchemaRevision,
1895 root_slot: usize,
1896 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
1897 let mut keys = candidate
1898 .bundle()
1899 .entity_snapshots()
1900 .iter()
1901 .map(|(entity_tag, snapshot)| {
1902 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
1903 })
1904 .collect::<BTreeSet<_>>();
1905 keys.insert(RawSchemaKey::from_accepted_bundle(
1906 candidate.root().bundle_key(),
1907 ));
1908 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
1909 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1910 for activation in snapshot
1911 .constraint_activations()
1912 .iter()
1913 .filter(|activation| activation.state() == ConstraintActivationState::Validating)
1914 {
1915 keys.insert(RawSchemaKey::from_constraint_validation_job(
1916 *entity_tag,
1917 activation.id(),
1918 ));
1919 }
1920 }
1921 Ok(keys)
1922 }
1923
1924 fn retain_durable_candidate_entries(
1928 &mut self,
1929 candidate: &CandidateSchemaRevision,
1930 root_slot: usize,
1931 ) -> Result<(), InternalError> {
1932 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1933 self.accepted_bundle_cache.get_mut().take();
1934 match &mut self.backend {
1935 SchemaStoreBackend::Heap(map) => map.retain(|key, _| keep.contains(key)),
1936 SchemaStoreBackend::Journaled {
1937 canonical,
1938 live,
1939 tombstones,
1940 } => {
1941 let stale = canonical
1942 .iter()
1943 .filter_map(|entry| (!keep.contains(entry.key())).then_some(*entry.key()))
1944 .collect::<Vec<_>>();
1945 for key in stale {
1946 canonical.remove(&key);
1947 }
1948 live.retain(|key, _| keep.contains(key));
1949 tombstones.clear();
1950 }
1951 }
1952 Ok(())
1953 }
1954
1955 fn retain_materialized_candidate_entries(
1956 &mut self,
1957 candidate: &CandidateSchemaRevision,
1958 root_slot: usize,
1959 ) -> Result<(), InternalError> {
1960 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1961 self.accepted_bundle_cache.get_mut().take();
1962 let SchemaStoreBackend::Journaled {
1963 canonical,
1964 live,
1965 tombstones,
1966 } = &mut self.backend
1967 else {
1968 return Err(InternalError::store_invariant());
1969 };
1970 live.retain(|key, _| keep.contains(key));
1971 let canonical_keys = canonical
1972 .iter()
1973 .map(|entry| *entry.key())
1974 .collect::<Vec<_>>();
1975 for key in canonical_keys {
1976 if keep.contains(&key) {
1977 tombstones.remove(&key);
1978 } else {
1979 tombstones.insert(key);
1980 }
1981 }
1982 Ok(())
1983 }
1984
1985 fn retain_canonical_candidate_entries(
1986 &mut self,
1987 candidate: &CandidateSchemaRevision,
1988 root_slot: usize,
1989 ) -> Result<(), InternalError> {
1990 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1991 self.accepted_bundle_cache.get_mut().take();
1992 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1993 return Err(InternalError::store_invariant());
1994 };
1995 let stale = canonical
1996 .iter()
1997 .filter_map(|entry| (!keep.contains(entry.key())).then_some(*entry.key()))
1998 .collect::<Vec<_>>();
1999 for key in stale {
2000 canonical.remove(&key);
2001 }
2002 Ok(())
2003 }
2004
2005 #[must_use]
2007 #[cfg(test)]
2008 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
2009 match &self.backend {
2010 SchemaStoreBackend::Heap(map) => map.contains_key(key),
2011 SchemaStoreBackend::Journaled { .. } => {
2012 self.get_raw_snapshot_for_backend(key).is_some()
2013 }
2014 }
2015 }
2016
2017 #[must_use]
2019 #[cfg(test)]
2020 pub(in crate::db) fn len(&self) -> u64 {
2021 match &self.backend {
2022 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
2023 SchemaStoreBackend::Journaled { .. } => {
2024 let mut count = 0_u64;
2025 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2026 count = count.saturating_add(1);
2027 Ok(SchemaStoreVisit::Continue)
2028 });
2029 count
2030 }
2031 }
2032 }
2033
2034 #[must_use]
2036 #[cfg(test)]
2037 pub(in crate::db) fn is_empty(&self) -> bool {
2038 match &self.backend {
2039 SchemaStoreBackend::Heap(map) => map.is_empty(),
2040 SchemaStoreBackend::Journaled { .. } => {
2041 let mut empty = true;
2042 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
2043 empty = false;
2044 Ok(SchemaStoreVisit::Stop)
2045 });
2046 empty
2047 }
2048 }
2049 }
2050
2051 #[cfg(test)]
2053 pub(in crate::db) fn clear(&mut self) {
2054 self.accepted_bundle_cache.get_mut().take();
2055 match &mut self.backend {
2056 SchemaStoreBackend::Heap(map) => map.clear(),
2057 SchemaStoreBackend::Journaled {
2058 canonical,
2059 live,
2060 tombstones,
2061 } => {
2062 live.clear();
2063 tombstones.clear();
2064 let keys = canonical
2065 .iter()
2066 .map(|entry| *entry.key())
2067 .collect::<Vec<_>>();
2068 for key in keys {
2069 if key.is_entity_snapshot() {
2070 tombstones.insert(key);
2071 } else {
2072 canonical.remove(&key);
2073 }
2074 }
2075 }
2076 }
2077 }
2078
2079 fn current_accepted_schema_bundle_ref(
2080 &self,
2081 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
2082 let Some(selection) = self.current_accepted_schema_root()? else {
2083 self.accepted_bundle_cache
2084 .try_borrow_mut()
2085 .map_err(|_| InternalError::store_invariant())?
2086 .take();
2087 return Ok(None);
2088 };
2089
2090 let cache_matches = self
2091 .accepted_bundle_cache
2092 .try_borrow()
2093 .map_err(|_| InternalError::store_invariant())?
2094 .as_ref()
2095 .is_some_and(|cached| cached.selection == selection);
2096 if !cache_matches {
2097 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
2098 let raw = self
2099 .get_raw_snapshot(&key)
2100 .ok_or_else(InternalError::store_corruption)?;
2101 let bundle =
2102 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
2103 self.validate_constraint_validation_job_closure(&bundle)?;
2104 #[cfg(test)]
2105 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
2106 .with(|misses| misses.set(misses.get().saturating_add(1)));
2107 let value_catalog = AcceptedValueCatalogHandle::new(
2108 bundle.enum_catalog().clone(),
2109 bundle.composite_catalog().clone(),
2110 self.accepted_catalog_scope
2111 .get_or_init(AcceptedStoreCatalogScope::new)
2112 .clone(),
2113 bundle.revision(),
2114 selection.root().fingerprint(),
2115 );
2116 *self
2117 .accepted_bundle_cache
2118 .try_borrow_mut()
2119 .map_err(|_| InternalError::store_invariant())? = Some(AcceptedSchemaBundleCache {
2120 selection,
2121 bundle,
2122 value_catalog,
2123 entity_selections: RefCell::new(StdBTreeMap::new()),
2124 });
2125 }
2126
2127 let cache = self
2128 .accepted_bundle_cache
2129 .try_borrow()
2130 .map_err(|_| InternalError::store_invariant())?;
2131 Ref::filter_map(cache, |cache| {
2132 cache
2133 .as_ref()
2134 .filter(|cached| cached.selection == selection)
2135 .map(|cached| &cached.bundle)
2136 })
2137 .map(Some)
2138 .map_err(|_| InternalError::store_invariant())
2139 }
2140
2141 fn latest_raw_snapshots_by_entity(
2142 &self,
2143 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
2144 #[cfg(test)]
2145 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(calls.get().saturating_add(1)));
2146
2147 let mut latest_by_entity =
2148 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
2149
2150 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
2151 let version = SchemaVersion::new(key.version());
2152 match latest_by_entity.get_mut(&key.entity_tag()) {
2153 Some((latest_version, latest_snapshot)) if version > *latest_version => {
2154 *latest_version = version;
2155 *latest_snapshot = snapshot.clone();
2156 }
2157 None => {
2158 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
2159 }
2160 Some(_) => {}
2161 }
2162 Ok(SchemaStoreVisit::Continue)
2163 });
2164
2165 latest_by_entity
2166 }
2167
2168 fn visit_raw_snapshots<E>(
2171 &self,
2172 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2173 ) -> Result<(), E> {
2174 let bounds = RawSchemaKey::all_entity_range_bounds();
2175 match &self.backend {
2176 SchemaStoreBackend::Heap(map) => {
2177 let mut visitor = visitor;
2178 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2179 if visitor(key, snapshot)?.should_stop() {
2180 break;
2181 }
2182 }
2183 }
2184 SchemaStoreBackend::Journaled {
2185 canonical,
2186 live,
2187 tombstones,
2188 } => Self::visit_journaled_raw_snapshot_range(
2189 canonical,
2190 live,
2191 tombstones,
2192 bounds,
2193 Direction::Asc,
2194 visitor,
2195 )?,
2196 }
2197
2198 Ok(())
2199 }
2200
2201 fn visit_constraint_validation_jobs<E>(
2202 &self,
2203 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2204 ) -> Result<(), E> {
2205 let bounds = RawSchemaKey::all_constraint_validation_job_range_bounds();
2206 match &self.backend {
2207 SchemaStoreBackend::Heap(map) => {
2208 let mut visitor = visitor;
2209 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
2210 if visitor(key, snapshot)?.should_stop() {
2211 break;
2212 }
2213 }
2214 }
2215 SchemaStoreBackend::Journaled {
2216 canonical,
2217 live,
2218 tombstones,
2219 } => Self::visit_journaled_raw_snapshot_range(
2220 canonical,
2221 live,
2222 tombstones,
2223 bounds,
2224 Direction::Asc,
2225 visitor,
2226 )?,
2227 }
2228 Ok(())
2229 }
2230
2231 #[cfg(test)]
2232 #[must_use]
2233 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
2234 match &self.backend {
2235 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
2236 SchemaStoreBackend::Heap(_) => 0,
2237 }
2238 }
2239
2240 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
2241 let SchemaStoreBackend::Journaled {
2242 canonical,
2243 live,
2244 tombstones,
2245 } = &self.backend
2246 else {
2247 return None;
2248 };
2249
2250 if tombstones.contains(key) {
2251 return None;
2252 }
2253 live.get(key).cloned().or_else(|| canonical.get(key))
2254 }
2255
2256 fn latest_raw_snapshot(&self, entity: EntityTag) -> Option<RawSchemaSnapshot> {
2257 self.latest_raw_snapshot_entry(entity)
2258 .map(|(_, snapshot)| snapshot)
2259 }
2260
2261 fn latest_raw_snapshot_entry(
2262 &self,
2263 entity: EntityTag,
2264 ) -> Option<(SchemaVersion, RawSchemaSnapshot)> {
2265 let bounds = RawSchemaKey::entity_range_bounds(entity);
2266 match &self.backend {
2267 SchemaStoreBackend::Heap(map) => map
2268 .range((bounds.0, bounds.1))
2269 .next_back()
2270 .map(|(key, snapshot)| (SchemaVersion::new(key.version()), snapshot.clone())),
2271 SchemaStoreBackend::Journaled {
2272 canonical,
2273 live,
2274 tombstones,
2275 } => {
2276 let mut latest = None;
2277 let _: Result<(), Infallible> = Self::visit_journaled_raw_snapshot_range(
2278 canonical,
2279 live,
2280 tombstones,
2281 bounds,
2282 Direction::Desc,
2283 |key, snapshot| {
2284 latest = Some((SchemaVersion::new(key.version()), snapshot.clone()));
2285 Ok(SchemaStoreVisit::Stop)
2286 },
2287 );
2288 latest
2289 }
2290 }
2291 }
2292
2293 fn visit_journaled_raw_snapshot_range<E>(
2294 canonical: &StableBTreeMap<
2295 RawSchemaKey,
2296 RawSchemaSnapshot,
2297 VirtualMemory<DefaultMemoryImpl>,
2298 >,
2299 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
2300 tombstones: &BTreeSet<RawSchemaKey>,
2301 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
2302 direction: Direction,
2303 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
2304 ) -> Result<(), E> {
2305 match direction {
2306 Direction::Asc => visit_ordered_overlay(
2307 canonical.range((bounds.0, bounds.1)),
2308 live.range((bounds.0, bounds.1)),
2309 Direction::Asc,
2310 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2311 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2312 |live_entry| !tombstones.contains(live_entry.0),
2313 |entry| {
2314 let visit = match entry {
2315 OrderedOverlayEntry::Canonical(canonical_entry) => {
2316 visitor(canonical_entry.key(), &canonical_entry.value())?
2317 }
2318 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2319 };
2320 Ok(if visit.should_stop() {
2321 OrderedOverlayVisit::Stop
2322 } else {
2323 OrderedOverlayVisit::Continue
2324 })
2325 },
2326 ),
2327 Direction::Desc => visit_ordered_overlay(
2328 canonical.range((bounds.0, bounds.1)).rev(),
2329 live.range((bounds.0, bounds.1)).rev(),
2330 Direction::Desc,
2331 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
2332 |canonical_entry| !tombstones.contains(canonical_entry.key()),
2333 |live_entry| !tombstones.contains(live_entry.0),
2334 |entry| {
2335 let visit = match entry {
2336 OrderedOverlayEntry::Canonical(canonical_entry) => {
2337 visitor(canonical_entry.key(), &canonical_entry.value())?
2338 }
2339 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
2340 };
2341 Ok(if visit.should_stop() {
2342 OrderedOverlayVisit::Stop
2343 } else {
2344 OrderedOverlayVisit::Continue
2345 })
2346 },
2347 ),
2348 }
2349 }
2350}
2351
2352fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
2353 match error {
2354 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
2355 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
2356 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
2357 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
2358 }
2359}
2360
2361fn derive_data_allocation_metadata(
2362 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2363) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2364 let mut max_version = SchemaVersion::initial();
2365 let mut hasher = new_hash_sha256();
2366 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
2367
2368 for (entity, (_, snapshot)) in latest_by_entity {
2369 let persisted = snapshot.decode_persisted_snapshot()?;
2370 if persisted.version() > max_version {
2371 max_version = persisted.version();
2372 }
2373
2374 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
2375 persisted.version(),
2376 persisted.entity_path().to_string(),
2377 persisted.entity_name().to_string(),
2378 persisted.primary_key_field_ids().to_vec(),
2379 persisted.row_layout().clone(),
2380 persisted.fields().to_vec(),
2381 Vec::new(),
2382 );
2383 let constraint_catalog = crate::db::schema::AcceptedConstraintCatalog::initial(
2384 data_projection.fields(),
2385 data_projection.indexes(),
2386 data_projection.relations(),
2387 )
2388 .map_err(|_| InternalError::store_invariant())?;
2389 let data_projection = data_projection.with_constraint_catalog(constraint_catalog);
2390 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
2391
2392 write_hash_u64(&mut hasher, entity.value());
2393 write_hash_u32(&mut hasher, persisted.version().get());
2394 write_hash_len_u32(&mut hasher, encoded.len());
2395 hasher.update(encoded);
2396 }
2397
2398 Ok(finalize_schema_metadata(
2399 max_version,
2400 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2401 hasher,
2402 latest_by_entity.len(),
2403 ))
2404}
2405
2406fn derive_index_allocation_metadata(
2407 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2408) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2409 let mut max_version = SchemaVersion::initial();
2410 let mut hasher = new_hash_sha256();
2411 write_hash_tag_u8(
2412 &mut hasher,
2413 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
2414 );
2415
2416 for (entity, (_, snapshot)) in latest_by_entity {
2417 let persisted = snapshot.decode_persisted_snapshot()?;
2418 if persisted.version() > max_version {
2419 max_version = persisted.version();
2420 }
2421
2422 write_hash_u64(&mut hasher, entity.value());
2423 write_hash_u32(&mut hasher, persisted.version().get());
2424 write_hash_len_u32(&mut hasher, persisted.indexes().len());
2425 for index in persisted.indexes() {
2426 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
2427 write_hash_str_u32(&mut hasher, index.name());
2428 write_hash_str_u32(&mut hasher, index.store());
2429 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
2430 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
2431 match index.predicate_sql() {
2432 Some(predicate_sql) => {
2433 write_hash_tag_u8(&mut hasher, 1);
2434 write_hash_str_u32(&mut hasher, predicate_sql);
2435 }
2436 None => write_hash_tag_u8(&mut hasher, 0),
2437 }
2438 hash_persisted_index_key(&mut hasher, index.key());
2439 }
2440 }
2441
2442 Ok(finalize_schema_metadata(
2443 max_version,
2444 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2445 hasher,
2446 latest_by_entity.len(),
2447 ))
2448}
2449
2450fn derive_schema_catalog_metadata(
2451 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
2452) -> Result<SchemaStoreCatalogMetadata, InternalError> {
2453 let mut max_version = SchemaVersion::initial();
2454 let mut hasher = new_hash_sha256();
2455 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
2456
2457 for (entity, (version, snapshot)) in latest_by_entity {
2458 let persisted = snapshot.decode_persisted_snapshot()?;
2459 if persisted.version() > max_version {
2460 max_version = persisted.version();
2461 }
2462
2463 write_hash_u64(&mut hasher, entity.value());
2464 write_hash_u32(&mut hasher, version.get());
2465 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
2466 hasher.update(snapshot.as_bytes());
2467 }
2468
2469 Ok(finalize_schema_metadata(
2470 max_version,
2471 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
2472 hasher,
2473 latest_by_entity.len(),
2474 ))
2475}
2476
2477fn finalize_schema_metadata(
2478 schema_version: SchemaVersion,
2479 schema_fingerprint_method_version: u8,
2480 hasher: sha2::Sha256,
2481 entity_count: usize,
2482) -> SchemaStoreCatalogMetadata {
2483 let digest = finalize_hash_sha256(hasher);
2484 let mut schema_fingerprint = [0u8; 16];
2485 schema_fingerprint.copy_from_slice(&digest[..16]);
2486
2487 SchemaStoreCatalogMetadata::new(
2488 schema_version,
2489 schema_fingerprint_method_version,
2490 schema_fingerprint,
2491 u64::try_from(entity_count).unwrap_or(u64::MAX),
2492 )
2493}
2494
2495fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
2496 match key {
2497 PersistedIndexKeySnapshot::FieldPath(paths) => {
2498 write_hash_tag_u8(hasher, 1);
2499 write_hash_len_u32(hasher, paths.len());
2500 for path in paths {
2501 hash_persisted_index_field_path(hasher, path);
2502 }
2503 }
2504 PersistedIndexKeySnapshot::Items(items) => {
2505 write_hash_tag_u8(hasher, 2);
2506 write_hash_len_u32(hasher, items.len());
2507 for item in items {
2508 match item {
2509 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
2510 write_hash_tag_u8(hasher, 1);
2511 hash_persisted_index_field_path(hasher, path);
2512 }
2513 PersistedIndexKeyItemSnapshot::Expression(expression) => {
2514 write_hash_tag_u8(hasher, 2);
2515 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
2516 hash_persisted_index_field_path(hasher, expression.source());
2517 hash_accepted_field_kind(hasher, expression.input_kind());
2518 hash_accepted_field_kind(hasher, expression.output_kind());
2519 write_hash_str_u32(hasher, expression.canonical_text());
2520 }
2521 }
2522 }
2523 }
2524 }
2525}
2526
2527fn hash_persisted_index_field_path(
2528 hasher: &mut sha2::Sha256,
2529 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
2530) {
2531 write_hash_u32(hasher, path.field_id().get());
2532 write_hash_u32(hasher, u32::from(path.slot().get()));
2533 write_hash_len_u32(hasher, path.path().len());
2534 for segment in path.path() {
2535 write_hash_str_u32(hasher, segment);
2536 }
2537 hash_accepted_field_kind(hasher, path.kind());
2538 write_hash_tag_u8(hasher, u8::from(path.nullable()));
2539}
2540
2541fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
2542 match kind {
2543 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
2544 AcceptedFieldKind::Blob { max_len } => {
2545 write_hash_tag_u8(hasher, 2);
2546 hash_optional_u32(hasher, *max_len);
2547 }
2548 AcceptedFieldKind::Bool => {
2549 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
2550 }
2551 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
2552 AcceptedFieldKind::Decimal { scale } => {
2553 write_hash_tag_u8(hasher, 5);
2554 write_hash_u32(hasher, *scale);
2555 }
2556 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
2557 AcceptedFieldKind::Enum { type_id } => {
2558 write_hash_tag_u8(hasher, 7);
2559 write_hash_u32(hasher, type_id.get());
2560 }
2561 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
2562 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
2563 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
2564 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
2565 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
2566 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
2567 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
2568 AcceptedFieldKind::IntBig { max_bytes } => {
2569 write_hash_tag_u8(hasher, 15);
2570 write_hash_u32(hasher, *max_bytes);
2571 }
2572 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
2573 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
2574 AcceptedFieldKind::Text { max_len } => {
2575 write_hash_tag_u8(hasher, 18);
2576 hash_optional_u32(hasher, *max_len);
2577 }
2578 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
2579 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
2580 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
2581 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
2582 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
2583 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
2584 AcceptedFieldKind::NatBig { max_bytes } => {
2585 write_hash_tag_u8(hasher, 25);
2586 write_hash_u32(hasher, *max_bytes);
2587 }
2588 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
2589 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
2590 AcceptedFieldKind::Relation {
2591 target_path,
2592 target_entity_name,
2593 target_entity_tag,
2594 target_store_path,
2595 key_kind,
2596 } => {
2597 write_hash_tag_u8(hasher, 28);
2598 write_hash_str_u32(hasher, target_path);
2599 write_hash_str_u32(hasher, target_entity_name);
2600 write_hash_u64(hasher, target_entity_tag.value());
2601 write_hash_str_u32(hasher, target_store_path);
2602 hash_accepted_field_kind(hasher, key_kind);
2603 }
2604 AcceptedFieldKind::List(inner) => {
2605 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
2606 hash_accepted_field_kind(hasher, inner);
2607 }
2608 AcceptedFieldKind::Set(inner) => {
2609 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
2610 hash_accepted_field_kind(hasher, inner);
2611 }
2612 AcceptedFieldKind::Map { key, value } => {
2613 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
2614 hash_accepted_field_kind(hasher, key);
2615 hash_accepted_field_kind(hasher, value);
2616 }
2617 AcceptedFieldKind::Composite { type_id } => {
2618 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
2619 write_hash_u32(hasher, type_id.get());
2620 }
2621 }
2622}
2623
2624fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
2625 match value {
2626 Some(value) => {
2627 write_hash_tag_u8(hasher, 1);
2628 write_hash_u32(hasher, value);
2629 }
2630 None => write_hash_tag_u8(hasher, 0),
2631 }
2632}
2633
2634const fn persisted_index_origin_name(
2635 origin: crate::db::schema::PersistedIndexOrigin,
2636) -> &'static str {
2637 match origin {
2638 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
2639 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
2640 }
2641}
2642
2643const fn persisted_expression_op_name(
2644 op: crate::db::schema::PersistedIndexExpressionOp,
2645) -> &'static str {
2646 match op {
2647 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
2648 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
2649 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
2650 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
2651 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
2652 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
2653 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
2654 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
2655 }
2656}
2657
2658#[cfg(test)]
2663mod tests;