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, PersistedIndexKeyItemSnapshot,
17 PersistedIndexKeySnapshot, PersistedSchemaSnapshot, SchemaVersion,
18 accepted_schema_cache_fingerprint,
19 accepted_schema_cache_fingerprint_for_persisted_snapshot,
20 accepted_schema_cache_fingerprint_method_version, decode_persisted_schema_snapshot,
21 encode_persisted_schema_snapshot,
22 enum_catalog::{
23 AcceptedSchemaAuthority, AcceptedSchemaPublicationError, AcceptedSchemaRevision,
24 AcceptedSchemaRevisionBundle, AcceptedSchemaRootSelection,
25 AcceptedStoreCatalogScope, AcceptedValueCatalogHandle, CandidateSchemaRevision,
26 decode_verified_accepted_schema_revision_bundle,
27 prepare_accepted_schema_root_publication, select_current_accepted_schema_root,
28 },
29 schema_snapshot_integrity_detail,
30 },
31 },
32 error::InternalError,
33 types::EntityTag,
34};
35use ic_stable_structures::{
36 BTreeMap as StableBTreeMap, DefaultMemoryImpl, Storable, memory_manager::VirtualMemory,
37 storable::Bound as StorableBound,
38};
39use sha2::Digest;
40use std::borrow::Cow;
41#[cfg(test)]
42use std::cell::Cell;
43use std::cell::{OnceCell, Ref, RefCell};
44use std::collections::{BTreeMap as StdBTreeMap, BTreeSet};
45use std::convert::Infallible;
46use std::ops::Bound as RangeBound;
47
48const SCHEMA_KEY_BYTES_USIZE: usize = 16;
49const SCHEMA_KEY_BYTES: u32 = 16;
50const SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT: u8 = 0;
51const SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE: u8 = 1;
52const SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT: u8 = 2;
53pub(in crate::db) const MAX_SCHEMA_SNAPSHOT_BYTES: u32 = 512 * 1024;
54const SCHEMA_STORE_FINGERPRINT_METHOD_VERSION: u8 = 1;
57const SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN: u8 = 1;
58const SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 2;
59const SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN: u8 = 3;
60const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL: u8 = 3;
61const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST: u8 = 29;
62const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET: u8 = 30;
63const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP: u8 = 31;
64const ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE: u8 = 32;
65const RAW_SCHEMA_SNAPSHOT_MAGIC: &[u8; 8] = b"ICYDBCAT";
66const RAW_SCHEMA_SNAPSHOT_VALUE_VERSION: u8 = 1;
67const RAW_SCHEMA_SNAPSHOT_HEADER_BYTES: usize = 25;
68
69#[cfg(test)]
70thread_local! {
71 static LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS: Cell<u64> = const { Cell::new(0) };
72 static ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES: Cell<u64> = const { Cell::new(0) };
73}
74
75#[cfg(test)]
76pub(in crate::db) fn reset_latest_raw_snapshots_by_entity_call_count_for_tests() {
77 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(0));
78}
79
80#[cfg(test)]
81pub(in crate::db) fn latest_raw_snapshots_by_entity_call_count_for_tests() -> u64 {
82 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(Cell::get)
83}
84
85#[cfg(test)]
86fn reset_accepted_schema_bundle_cache_miss_count_for_tests() {
87 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(|misses| misses.set(0));
88}
89
90#[cfg(test)]
91fn accepted_schema_bundle_cache_miss_count_for_tests() -> u64 {
92 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES.with(Cell::get)
93}
94
95#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
104struct RawSchemaKey([u8; SCHEMA_KEY_BYTES_USIZE]);
105
106impl RawSchemaKey {
107 #[must_use]
109 fn from_entity_version(entity: EntityTag, version: SchemaVersion) -> Self {
110 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
111 out[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
112 out[4..12].copy_from_slice(&entity.value().to_be_bytes());
113 out[12..].copy_from_slice(&version.get().to_be_bytes());
114
115 Self(out)
116 }
117
118 fn from_accepted_bundle(bundle_key: super::enum_catalog::AcceptedSchemaBundleKey) -> Self {
119 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
120 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_BUNDLE;
121 out[4..12].copy_from_slice(&bundle_key.get().to_be_bytes());
122 Self(out)
123 }
124
125 fn from_accepted_root_slot(slot: usize) -> Result<Self, InternalError> {
126 let slot = u32::try_from(slot).map_err(|_| InternalError::store_invariant())?;
127 if slot > 1 {
128 return Err(InternalError::store_invariant());
129 }
130 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
131 out[0] = SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT;
132 out[12..].copy_from_slice(&slot.to_be_bytes());
133 Ok(Self(out))
134 }
135
136 #[must_use]
138 fn entity_tag(self) -> EntityTag {
139 let mut bytes = [0u8; size_of::<u64>()];
140 bytes.copy_from_slice(&self.0[4..12]);
141
142 EntityTag::new(u64::from_be_bytes(bytes))
143 }
144
145 #[must_use]
147 fn version(self) -> u32 {
148 let mut bytes = [0u8; size_of::<u32>()];
149 bytes.copy_from_slice(&self.0[12..]);
150
151 u32::from_be_bytes(bytes)
152 }
153
154 fn entity_range_bounds(entity: EntityTag) -> (RangeBound<Self>, RangeBound<Self>) {
155 (
156 RangeBound::Included(Self::from_entity_version(entity, SchemaVersion::new(0))),
157 RangeBound::Included(Self::from_entity_version(
158 entity,
159 SchemaVersion::new(u32::MAX),
160 )),
161 )
162 }
163
164 const fn all_entity_range_bounds() -> (RangeBound<Self>, RangeBound<Self>) {
165 let mut end = [u8::MAX; SCHEMA_KEY_BYTES_USIZE];
166 end[0] = SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT;
167 (
168 RangeBound::Included(Self([0; SCHEMA_KEY_BYTES_USIZE])),
169 RangeBound::Included(Self(end)),
170 )
171 }
172
173 #[cfg(test)]
174 const fn is_entity_snapshot(self) -> bool {
175 self.0[0] == SCHEMA_KEY_NAMESPACE_ENTITY_SNAPSHOT
176 }
177
178 const fn is_accepted_root(self) -> bool {
179 self.0[0] == SCHEMA_KEY_NAMESPACE_ACCEPTED_ROOT
180 }
181}
182
183impl Storable for RawSchemaKey {
184 fn to_bytes(&self) -> Cow<'_, [u8]> {
185 Cow::Borrowed(&self.0)
186 }
187
188 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
189 debug_assert_eq!(
190 bytes.len(),
191 SCHEMA_KEY_BYTES_USIZE,
192 "RawSchemaKey::from_bytes received unexpected byte length",
193 );
194
195 if bytes.len() != SCHEMA_KEY_BYTES_USIZE {
196 return Self([0u8; SCHEMA_KEY_BYTES_USIZE]);
197 }
198
199 let mut out = [0u8; SCHEMA_KEY_BYTES_USIZE];
200 out.copy_from_slice(bytes.as_ref());
201 Self(out)
202 }
203
204 fn into_bytes(self) -> Vec<u8> {
205 self.0.to_vec()
206 }
207
208 const BOUND: StorableBound = StorableBound::Bounded {
209 max_size: SCHEMA_KEY_BYTES,
210 is_fixed_size: true,
211 };
212}
213
214#[derive(Clone, Debug, Eq, PartialEq)]
225struct RawSchemaSnapshot {
226 payload: Vec<u8>,
227 accepted_schema_fingerprint: Option<CommitSchemaFingerprint>,
228}
229
230impl RawSchemaSnapshot {
231 fn from_persisted_snapshot(snapshot: &PersistedSchemaSnapshot) -> Result<Self, InternalError> {
233 validate_typed_schema_snapshot_for_store(snapshot)?;
234
235 let accepted_schema_fingerprint =
236 accepted_schema_cache_fingerprint_for_persisted_snapshot(snapshot)?;
237 let payload = encode_persisted_schema_snapshot(snapshot)?;
238
239 Ok(Self {
240 payload,
241 accepted_schema_fingerprint: Some(accepted_schema_fingerprint),
242 })
243 }
244
245 #[must_use]
247 const fn from_encoded_control_record(payload: Vec<u8>) -> Self {
248 Self {
249 payload,
250 accepted_schema_fingerprint: None,
251 }
252 }
253
254 #[cfg(test)]
257 #[must_use]
258 const fn from_unchecked_persisted_snapshot_payload(payload: Vec<u8>) -> Self {
259 Self {
260 payload,
261 accepted_schema_fingerprint: Some([0; size_of::<CommitSchemaFingerprint>()]),
262 }
263 }
264
265 #[must_use]
267 const fn as_bytes(&self) -> &[u8] {
268 self.payload.as_slice()
269 }
270
271 #[must_use]
273 fn into_bytes(self) -> Vec<u8> {
274 self.payload
275 }
276
277 fn accepted_schema_fingerprint(&self) -> Result<CommitSchemaFingerprint, InternalError> {
280 self.accepted_schema_fingerprint
281 .ok_or_else(InternalError::store_corruption)
282 }
283
284 fn decode_persisted_snapshot(&self) -> Result<PersistedSchemaSnapshot, InternalError> {
286 let _fingerprint = self.accepted_schema_fingerprint()?;
289 decode_persisted_schema_snapshot(self.as_bytes())
290 }
291}
292
293#[cfg(test)]
294pub(in crate::db::schema) fn validate_raw_schema_snapshot_bytes_for_tests(
295 bytes: Vec<u8>,
296) -> Result<(), InternalError> {
297 let raw = <RawSchemaSnapshot as Storable>::from_bytes(Cow::Owned(bytes));
298 raw.decode_persisted_snapshot().map(drop)
299}
300
301#[derive(Clone, Copy, Debug, Eq, PartialEq)]
302pub(in crate::db) struct AcceptedCatalogIdentity {
303 entity_tag: EntityTag,
304 entity_path: &'static str,
305 store_path: &'static str,
306 accepted_schema_revision: AcceptedSchemaRevision,
307 accepted_schema_version: SchemaVersion,
308 fingerprint_method_version: u8,
309 accepted_schema_fingerprint: CommitSchemaFingerprint,
310}
311
312impl AcceptedCatalogIdentity {
313 #[must_use]
314 pub(in crate::db) const fn new(
315 entity_tag: EntityTag,
316 entity_path: &'static str,
317 store_path: &'static str,
318 accepted_schema_revision: AcceptedSchemaRevision,
319 accepted_schema_version: SchemaVersion,
320 accepted_schema_fingerprint: CommitSchemaFingerprint,
321 ) -> Self {
322 Self {
323 entity_tag,
324 entity_path,
325 store_path,
326 accepted_schema_revision,
327 accepted_schema_version,
328 fingerprint_method_version: accepted_schema_cache_fingerprint_method_version(),
329 accepted_schema_fingerprint,
330 }
331 }
332
333 #[must_use]
334 pub(in crate::db) const fn entity_tag(self) -> EntityTag {
335 self.entity_tag
336 }
337
338 #[must_use]
339 pub(in crate::db) const fn entity_path(self) -> &'static str {
340 self.entity_path
341 }
342
343 #[must_use]
344 pub(in crate::db) const fn store_path(self) -> &'static str {
345 self.store_path
346 }
347
348 #[must_use]
349 pub(in crate::db) const fn accepted_schema_revision(self) -> AcceptedSchemaRevision {
350 self.accepted_schema_revision
351 }
352
353 #[must_use]
354 pub(in crate::db) const fn accepted_schema_version(self) -> SchemaVersion {
355 self.accepted_schema_version
356 }
357
358 #[must_use]
359 pub(in crate::db) const fn fingerprint_method_version(self) -> u8 {
360 self.fingerprint_method_version
361 }
362
363 #[must_use]
364 pub(in crate::db) const fn accepted_schema_fingerprint(self) -> CommitSchemaFingerprint {
365 self.accepted_schema_fingerprint
366 }
367}
368
369#[derive(Clone, Debug, Eq, PartialEq)]
370pub(in crate::db) struct AcceptedCatalogSnapshotSelection {
371 identity: AcceptedCatalogIdentity,
372 value_catalog: AcceptedValueCatalogHandle,
373 raw_snapshot: Vec<u8>,
374}
375
376impl AcceptedCatalogSnapshotSelection {
377 #[must_use]
378 const fn new(
379 identity: AcceptedCatalogIdentity,
380 value_catalog: AcceptedValueCatalogHandle,
381 raw_snapshot: Vec<u8>,
382 ) -> Self {
383 Self {
384 identity,
385 value_catalog,
386 raw_snapshot,
387 }
388 }
389
390 #[must_use]
391 pub(in crate::db) const fn identity(&self) -> AcceptedCatalogIdentity {
392 self.identity
393 }
394
395 #[must_use]
396 pub(in crate::db) const fn value_catalog_handle(&self) -> &AcceptedValueCatalogHandle {
397 &self.value_catalog
398 }
399
400 pub(in crate::db) fn from_accepted_snapshot(
402 identity: AcceptedCatalogIdentity,
403 value_catalog: AcceptedValueCatalogHandle,
404 snapshot: &AcceptedSchemaSnapshot,
405 ) -> Result<Self, InternalError> {
406 let raw_snapshot =
407 RawSchemaSnapshot::from_persisted_snapshot(snapshot.persisted_snapshot())?;
408 if raw_snapshot.accepted_schema_fingerprint()? != identity.accepted_schema_fingerprint() {
409 return Err(InternalError::store_invariant());
410 }
411
412 Ok(Self::new(
413 identity,
414 value_catalog,
415 raw_snapshot.into_bytes(),
416 ))
417 }
418
419 pub(in crate::db) fn from_candidate(
422 candidate: &CandidateSchemaRevision,
423 entity_tag: EntityTag,
424 entity_path: &'static str,
425 store_path: &'static str,
426 ) -> Result<Option<Self>, InternalError> {
427 if candidate.store_path() != store_path {
428 return Err(InternalError::store_corruption());
429 }
430 let Some(snapshot) = candidate.bundle().entity_snapshots().get(&entity_tag) else {
431 return Ok(None);
432 };
433 if snapshot.entity_path() != entity_path {
434 return Err(InternalError::store_corruption());
435 }
436
437 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
438 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
439 let identity = AcceptedCatalogIdentity::new(
440 entity_tag,
441 entity_path,
442 store_path,
443 candidate.revision(),
444 snapshot.version(),
445 fingerprint,
446 );
447
448 Ok(Some(Self::new(
449 identity,
450 AcceptedValueCatalogHandle::new(
451 candidate.bundle().enum_catalog().clone(),
452 candidate.bundle().composite_catalog().clone(),
453 AcceptedStoreCatalogScope::new(),
454 candidate.revision(),
455 candidate.root().fingerprint(),
456 ),
457 raw_snapshot.into_bytes(),
458 )))
459 }
460
461 pub(in crate::db) fn decode_verified(&self) -> Result<AcceptedSchemaSnapshot, InternalError> {
462 let snapshot = decode_persisted_schema_snapshot(&self.raw_snapshot)?;
463 let accepted = AcceptedSchemaSnapshot::try_new(snapshot)?;
464 let identity = self.identity();
465
466 if accepted.persisted_snapshot().version() != identity.accepted_schema_version() {
467 return Err(InternalError::store_invariant());
468 }
469 if accepted.entity_path() != identity.entity_path() {
470 return Err(InternalError::store_invariant());
471 }
472
473 let decoded_fingerprint = accepted_schema_cache_fingerprint(&accepted)?;
474 if decoded_fingerprint != identity.accepted_schema_fingerprint() {
475 return Err(InternalError::store_invariant());
476 }
477
478 Ok(accepted)
479 }
480}
481
482impl Storable for RawSchemaSnapshot {
483 fn to_bytes(&self) -> Cow<'_, [u8]> {
484 let Some(fingerprint) = self.accepted_schema_fingerprint else {
485 return Cow::Borrowed(self.as_bytes());
486 };
487
488 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
489 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
490 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
491 bytes.extend_from_slice(&fingerprint);
492 bytes.extend_from_slice(self.as_bytes());
493
494 Cow::Owned(bytes)
495 }
496
497 fn from_bytes(bytes: Cow<'_, [u8]>) -> Self {
498 let bytes = bytes.into_owned();
499 if bytes.len() >= RAW_SCHEMA_SNAPSHOT_HEADER_BYTES
500 && &bytes[..RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_MAGIC
501 && bytes[RAW_SCHEMA_SNAPSHOT_MAGIC.len()] == RAW_SCHEMA_SNAPSHOT_VALUE_VERSION
502 {
503 let fingerprint_start = RAW_SCHEMA_SNAPSHOT_MAGIC.len() + size_of::<u8>();
504 let fingerprint_end = fingerprint_start + size_of::<CommitSchemaFingerprint>();
505 let mut fingerprint = [0_u8; size_of::<CommitSchemaFingerprint>()];
506 fingerprint.copy_from_slice(&bytes[fingerprint_start..fingerprint_end]);
507
508 return Self {
509 payload: bytes[fingerprint_end..].to_vec(),
510 accepted_schema_fingerprint: Some(fingerprint),
511 };
512 }
513
514 Self {
515 payload: bytes,
516 accepted_schema_fingerprint: None,
517 }
518 }
519
520 fn into_bytes(self) -> Vec<u8> {
521 let Some(fingerprint) = self.accepted_schema_fingerprint else {
522 return self.payload;
523 };
524
525 let mut bytes = Vec::with_capacity(RAW_SCHEMA_SNAPSHOT_HEADER_BYTES + self.payload.len());
526 bytes.extend_from_slice(RAW_SCHEMA_SNAPSHOT_MAGIC);
527 bytes.push(RAW_SCHEMA_SNAPSHOT_VALUE_VERSION);
528 bytes.extend_from_slice(&fingerprint);
529 bytes.extend_from_slice(&self.payload);
530
531 bytes
532 }
533
534 const BOUND: StorableBound = StorableBound::Unbounded;
535}
536
537fn validate_typed_schema_snapshot_for_store(
541 snapshot: &PersistedSchemaSnapshot,
542) -> Result<(), InternalError> {
543 if schema_snapshot_integrity_detail(
544 "schema snapshot",
545 snapshot.version(),
546 snapshot.primary_key_field_ids(),
547 snapshot.row_layout(),
548 snapshot.fields(),
549 )
550 .is_some()
551 {
552 return Err(InternalError::store_invariant());
553 }
554
555 Ok(())
556}
557
558#[derive(Clone, Copy, Debug, Eq, PartialEq)]
567pub(in crate::db) struct SchemaStoreFootprint {
568 snapshots: u64,
569 encoded_bytes: u64,
570 latest_snapshot_bytes: u64,
571}
572
573#[derive(Clone, Copy, Debug, Eq, PartialEq)]
581pub(in crate::db) struct SchemaStoreCatalogMetadata {
582 schema_version: SchemaVersion,
583 schema_fingerprint_method_version: u8,
584 schema_fingerprint: CommitSchemaFingerprint,
585 entity_count: u64,
586}
587
588impl SchemaStoreCatalogMetadata {
589 #[must_use]
591 const fn new(
592 schema_version: SchemaVersion,
593 schema_fingerprint_method_version: u8,
594 schema_fingerprint: CommitSchemaFingerprint,
595 entity_count: u64,
596 ) -> Self {
597 Self {
598 schema_version,
599 schema_fingerprint_method_version,
600 schema_fingerprint,
601 entity_count,
602 }
603 }
604
605 #[must_use]
607 pub(in crate::db) const fn schema_version(self) -> SchemaVersion {
608 self.schema_version
609 }
610
611 #[must_use]
613 pub(in crate::db) const fn schema_fingerprint_method_version(self) -> u8 {
614 self.schema_fingerprint_method_version
615 }
616
617 #[must_use]
620 pub(in crate::db) const fn schema_fingerprint(self) -> CommitSchemaFingerprint {
621 self.schema_fingerprint
622 }
623
624 #[must_use]
626 pub(in crate::db) const fn entity_count(self) -> u64 {
627 self.entity_count
628 }
629}
630
631#[derive(Clone, Copy, Debug, Eq, PartialEq)]
640pub(in crate::db) struct SchemaStoreAllocationMetadata {
641 data: SchemaStoreCatalogMetadata,
642 index: SchemaStoreCatalogMetadata,
643 schema: SchemaStoreCatalogMetadata,
644}
645
646impl SchemaStoreAllocationMetadata {
647 #[must_use]
650 const fn new(
651 data: SchemaStoreCatalogMetadata,
652 index: SchemaStoreCatalogMetadata,
653 schema: SchemaStoreCatalogMetadata,
654 ) -> Self {
655 Self {
656 data,
657 index,
658 schema,
659 }
660 }
661
662 #[must_use]
664 pub(in crate::db) const fn data(self) -> SchemaStoreCatalogMetadata {
665 self.data
666 }
667
668 #[must_use]
670 pub(in crate::db) const fn index(self) -> SchemaStoreCatalogMetadata {
671 self.index
672 }
673
674 #[must_use]
677 pub(in crate::db) const fn schema(self) -> SchemaStoreCatalogMetadata {
678 self.schema
679 }
680}
681
682impl SchemaStoreFootprint {
683 #[must_use]
685 const fn new(snapshots: u64, encoded_bytes: u64, latest_snapshot_bytes: u64) -> Self {
686 Self {
687 snapshots,
688 encoded_bytes,
689 latest_snapshot_bytes,
690 }
691 }
692
693 #[must_use]
695 pub(in crate::db) const fn snapshots(self) -> u64 {
696 self.snapshots
697 }
698
699 #[must_use]
701 pub(in crate::db) const fn encoded_bytes(self) -> u64 {
702 self.encoded_bytes
703 }
704
705 #[must_use]
707 pub(in crate::db) const fn latest_snapshot_bytes(self) -> u64 {
708 self.latest_snapshot_bytes
709 }
710}
711
712pub struct SchemaStore {
721 backend: SchemaStoreBackend,
722 accepted_bundle_cache: RefCell<Option<AcceptedSchemaBundleCache>>,
723 accepted_catalog_scope: OnceCell<AcceptedStoreCatalogScope>,
724}
725
726struct AcceptedSchemaBundleCache {
727 selection: AcceptedSchemaRootSelection,
728 bundle: AcceptedSchemaRevisionBundle,
729}
730
731enum SchemaStoreBackend {
732 Heap(StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>),
733 Journaled {
734 canonical:
735 StableBTreeMap<RawSchemaKey, RawSchemaSnapshot, VirtualMemory<DefaultMemoryImpl>>,
736 live: StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
737 tombstones: BTreeSet<RawSchemaKey>,
738 },
739}
740
741#[derive(Clone, Copy, Debug, Eq, PartialEq)]
743enum SchemaStoreVisit {
744 Continue,
745 Stop,
746}
747
748impl SchemaStoreVisit {
749 const fn should_stop(self) -> bool {
750 matches!(self, Self::Stop)
751 }
752}
753
754impl SchemaStore {
755 #[must_use]
757 pub const fn init_heap() -> Self {
758 Self {
759 backend: SchemaStoreBackend::Heap(StdBTreeMap::new()),
760 accepted_bundle_cache: RefCell::new(None),
761 accepted_catalog_scope: OnceCell::new(),
762 }
763 }
764
765 #[must_use]
770 pub fn init_journaled(memory: VirtualMemory<DefaultMemoryImpl>) -> Self {
771 Self {
772 backend: SchemaStoreBackend::Journaled {
773 canonical: StableBTreeMap::init(memory),
774 live: StdBTreeMap::new(),
775 tombstones: BTreeSet::new(),
776 },
777 accepted_bundle_cache: RefCell::new(None),
778 accepted_catalog_scope: OnceCell::new(),
779 }
780 }
781
782 pub(in crate::db) fn insert_persisted_snapshot(
784 &mut self,
785 entity: EntityTag,
786 snapshot: &PersistedSchemaSnapshot,
787 ) -> Result<(), InternalError> {
788 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
789 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
790 let _ = self.insert_raw_snapshot(key, raw_snapshot);
791
792 Ok(())
793 }
794
795 pub(in crate::db) fn reset_journaled_live_projection(&mut self) -> Result<(), InternalError> {
798 let SchemaStoreBackend::Journaled {
799 live, tombstones, ..
800 } = &mut self.backend
801 else {
802 return Err(InternalError::store_invariant());
803 };
804
805 live.clear();
806 tombstones.clear();
807 self.accepted_bundle_cache.get_mut().take();
808
809 Ok(())
810 }
811
812 pub(in crate::db) fn fold_persisted_snapshot(
814 &mut self,
815 entity: EntityTag,
816 snapshot: &PersistedSchemaSnapshot,
817 ) -> Result<(), InternalError> {
818 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
819 return Err(InternalError::store_invariant());
820 };
821
822 let key = RawSchemaKey::from_entity_version(entity, snapshot.version());
823 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
824 canonical.insert(key, raw_snapshot);
825
826 Ok(())
827 }
828
829 pub(in crate::db) fn current_accepted_schema_root(
831 &self,
832 ) -> Result<Option<AcceptedSchemaRootSelection>, InternalError> {
833 let first = self.accepted_root_slot_bytes(0)?;
834 let second = self.accepted_root_slot_bytes(1)?;
835 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])
836 }
837
838 pub(in crate::db) fn current_accepted_schema_bundle(
840 &self,
841 ) -> Result<Option<AcceptedSchemaRevisionBundle>, InternalError> {
842 let Some(selection) = self.current_accepted_schema_root()? else {
843 return Ok(None);
844 };
845 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
846 let raw = self
847 .get_raw_snapshot(&key)
848 .ok_or_else(InternalError::store_corruption)?;
849 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes()).map(Some)
850 }
851
852 pub(in crate::db) fn current_accepted_schema_revision(
854 &self,
855 ) -> Result<Option<AcceptedSchemaRevision>, InternalError> {
856 Ok(self
857 .current_accepted_schema_root()?
858 .map(|selection| selection.root().revision()))
859 }
860
861 pub(in crate::db) fn current_accepted_schema_authority_matches(
864 &self,
865 expected: &AcceptedSchemaAuthority,
866 ) -> Result<bool, InternalError> {
867 let Some(store_scope) = self.accepted_catalog_scope.get() else {
868 return Ok(false);
869 };
870
871 if let Some(cached) = self
874 .accepted_bundle_cache
875 .try_borrow()
876 .map_err(|_| InternalError::store_invariant())?
877 .as_ref()
878 {
879 let root = cached.selection.root();
880 return Ok(expected.matches_store_root(
881 store_scope,
882 root.revision(),
883 root.fingerprint(),
884 ));
885 }
886
887 let Some(selection) = self.current_accepted_schema_root()? else {
888 return Ok(false);
889 };
890 let root = selection.root();
891
892 Ok(expected.matches_store_root(store_scope, root.revision(), root.fingerprint()))
893 }
894
895 pub(in crate::db) fn publish_accepted_schema_candidate(
899 &mut self,
900 expected_revision: AcceptedSchemaRevision,
901 candidate: &CandidateSchemaRevision,
902 ) -> Result<(), InternalError> {
903 if self.current_root_matches_candidate(candidate)? {
904 let selection = self
905 .current_accepted_schema_root()?
906 .ok_or_else(InternalError::store_corruption)?;
907 self.retain_durable_candidate_entries(candidate, selection.slot())?;
908 return Ok(());
909 }
910 let first = self.accepted_root_slot_bytes(0)?;
911 let second = self.accepted_root_slot_bytes(1)?;
912 prepare_accepted_schema_root_publication(
913 [first.as_deref(), second.as_deref()],
914 expected_revision,
915 candidate,
916 )
917 .map_err(map_schema_publication_error)?;
918
919 self.insert_durable_candidate_snapshots(candidate)?;
920 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
921 self.insert_durable_raw_value(bundle_key, candidate.encoded_bundle().to_vec());
922 let persisted_bundle = self
923 .get_raw_snapshot(&bundle_key)
924 .ok_or_else(InternalError::store_corruption)?;
925 let _verified = decode_verified_accepted_schema_revision_bundle(
926 candidate.root(),
927 persisted_bundle.as_bytes(),
928 )?;
929
930 let first = self.accepted_root_slot_bytes(0)?;
933 let second = self.accepted_root_slot_bytes(1)?;
934 let publication = prepare_accepted_schema_root_publication(
935 [first.as_deref(), second.as_deref()],
936 expected_revision,
937 candidate,
938 )
939 .map_err(map_schema_publication_error)?;
940 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
941 self.insert_durable_raw_value(root_key, publication.encoded_root().to_vec());
942
943 let selected = self
944 .current_accepted_schema_root()?
945 .ok_or_else(InternalError::store_corruption)?;
946 if selected.root() != candidate.root() {
947 return Err(InternalError::store_corruption());
948 }
949 self.retain_durable_candidate_entries(candidate, selected.slot())?;
950 Ok(())
951 }
952
953 pub(in crate::db) fn apply_journaled_accepted_schema_candidate(
955 &mut self,
956 expected_revision: AcceptedSchemaRevision,
957 candidate: &CandidateSchemaRevision,
958 ) -> Result<(), InternalError> {
959 if !matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
960 return Err(InternalError::store_invariant());
961 }
962 if self.current_root_matches_candidate(candidate)? {
963 let selection = self
964 .current_accepted_schema_root()?
965 .ok_or_else(InternalError::store_corruption)?;
966 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
967 return Ok(());
968 }
969
970 let first = self.accepted_root_slot_bytes(0)?;
971 let second = self.accepted_root_slot_bytes(1)?;
972 prepare_accepted_schema_root_publication(
973 [first.as_deref(), second.as_deref()],
974 expected_revision,
975 candidate,
976 )
977 .map_err(map_schema_publication_error)?;
978
979 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
980 self.insert_persisted_snapshot(*entity_tag, snapshot)?;
981 }
982 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
983 self.insert_raw_snapshot(
984 bundle_key,
985 RawSchemaSnapshot::from_encoded_control_record(candidate.encoded_bundle().to_vec()),
986 );
987 let persisted_bundle = self
988 .get_raw_snapshot(&bundle_key)
989 .ok_or_else(InternalError::store_corruption)?;
990 let _verified = decode_verified_accepted_schema_revision_bundle(
991 candidate.root(),
992 persisted_bundle.as_bytes(),
993 )?;
994
995 let first = self.accepted_root_slot_bytes(0)?;
996 let second = self.accepted_root_slot_bytes(1)?;
997 let publication = prepare_accepted_schema_root_publication(
998 [first.as_deref(), second.as_deref()],
999 expected_revision,
1000 candidate,
1001 )
1002 .map_err(map_schema_publication_error)?;
1003 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1004 self.insert_raw_snapshot(
1005 root_key,
1006 RawSchemaSnapshot::from_encoded_control_record(publication.encoded_root().to_vec()),
1007 );
1008
1009 if !self.current_root_matches_candidate(candidate)? {
1010 return Err(InternalError::store_corruption());
1011 }
1012 let selection = self
1013 .current_accepted_schema_root()?
1014 .ok_or_else(InternalError::store_corruption)?;
1015 self.retain_materialized_candidate_entries(candidate, selection.slot())?;
1016 Ok(())
1017 }
1018
1019 pub(in crate::db) fn fold_journaled_accepted_schema_candidate(
1021 &mut self,
1022 expected_revision: AcceptedSchemaRevision,
1023 candidate: &CandidateSchemaRevision,
1024 ) -> Result<(), InternalError> {
1025 if self.canonical_root_matches_candidate(candidate)? {
1026 let first = self.canonical_root_slot_bytes(0)?;
1027 let second = self.canonical_root_slot_bytes(1)?;
1028 let selection =
1029 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1030 .ok_or_else(InternalError::store_corruption)?;
1031 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
1032 return Ok(());
1033 }
1034
1035 let first = self.canonical_root_slot_bytes(0)?;
1036 let second = self.canonical_root_slot_bytes(1)?;
1037 prepare_accepted_schema_root_publication(
1038 [first.as_deref(), second.as_deref()],
1039 expected_revision,
1040 candidate,
1041 )
1042 .map_err(map_schema_publication_error)?;
1043
1044 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1045 self.fold_persisted_snapshot(*entity_tag, snapshot)?;
1046 }
1047 let bundle_key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1048 self.insert_canonical_raw_value(bundle_key, candidate.encoded_bundle().to_vec())?;
1049 let persisted_bundle = self
1050 .get_canonical_raw_value(&bundle_key)?
1051 .ok_or_else(InternalError::store_corruption)?;
1052 let _verified = decode_verified_accepted_schema_revision_bundle(
1053 candidate.root(),
1054 persisted_bundle.as_bytes(),
1055 )?;
1056
1057 let first = self.canonical_root_slot_bytes(0)?;
1058 let second = self.canonical_root_slot_bytes(1)?;
1059 let publication = prepare_accepted_schema_root_publication(
1060 [first.as_deref(), second.as_deref()],
1061 expected_revision,
1062 candidate,
1063 )
1064 .map_err(map_schema_publication_error)?;
1065 let root_key = RawSchemaKey::from_accepted_root_slot(publication.target_slot())?;
1066 self.insert_canonical_raw_value(root_key, publication.encoded_root().to_vec())?;
1067
1068 if !self.canonical_root_matches_candidate(candidate)? {
1069 return Err(InternalError::store_corruption());
1070 }
1071 let first = self.canonical_root_slot_bytes(0)?;
1072 let second = self.canonical_root_slot_bytes(1)?;
1073 let selection = select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1074 .ok_or_else(InternalError::store_corruption)?;
1075 self.retain_canonical_candidate_entries(candidate, selection.slot())?;
1076 Ok(())
1077 }
1078
1079 #[cfg(test)]
1081 pub(in crate::db) fn get_persisted_snapshot(
1082 &self,
1083 entity: EntityTag,
1084 version: SchemaVersion,
1085 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1086 let key = RawSchemaKey::from_entity_version(entity, version);
1087 self.get_raw_snapshot(&key)
1088 .map(|snapshot| snapshot.decode_persisted_snapshot())
1089 .transpose()
1090 }
1091
1092 pub(in crate::db) fn latest_staged_persisted_snapshot(
1097 &self,
1098 entity: EntityTag,
1099 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1100 self.latest_raw_snapshot(entity)
1101 .map(|snapshot| snapshot.decode_persisted_snapshot())
1102 .transpose()
1103 }
1104
1105 #[cfg(any(test, feature = "sql"))]
1108 pub(in crate::db) fn current_accepted_persisted_snapshot(
1109 &self,
1110 entity: EntityTag,
1111 ) -> Result<Option<PersistedSchemaSnapshot>, InternalError> {
1112 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1113 return Ok(None);
1114 };
1115
1116 Ok(bundle.entity_snapshots().get(&entity).cloned())
1117 }
1118
1119 pub(in crate::db) fn current_accepted_catalog_selection(
1121 &self,
1122 entity: EntityTag,
1123 entity_path: &'static str,
1124 store_path: &'static str,
1125 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
1126 let Some(bundle) = self.current_accepted_schema_bundle_ref()? else {
1127 return Ok(None);
1128 };
1129 if bundle.store_path() != store_path {
1130 return Err(InternalError::store_corruption());
1131 }
1132 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
1133 return Ok(None);
1134 };
1135 if snapshot.entity_path() != entity_path {
1136 return Err(InternalError::store_corruption());
1137 }
1138
1139 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1140 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
1141 let identity = AcceptedCatalogIdentity::new(
1142 entity,
1143 entity_path,
1144 store_path,
1145 bundle.revision(),
1146 snapshot.version(),
1147 fingerprint,
1148 );
1149
1150 Ok(Some(AcceptedCatalogSnapshotSelection::new(
1151 identity,
1152 AcceptedValueCatalogHandle::new(
1153 bundle.enum_catalog().clone(),
1154 bundle.composite_catalog().clone(),
1155 self.accepted_catalog_scope
1156 .get_or_init(AcceptedStoreCatalogScope::new)
1157 .clone(),
1158 bundle.revision(),
1159 self.current_accepted_schema_root()?
1160 .ok_or_else(InternalError::store_corruption)?
1161 .root()
1162 .fingerprint(),
1163 ),
1164 raw_snapshot.into_bytes(),
1165 )))
1166 }
1167
1168 pub(in crate::db) fn current_canonical_accepted_catalog_selection(
1172 &self,
1173 entity: EntityTag,
1174 entity_path: &'static str,
1175 store_path: &'static str,
1176 ) -> Result<Option<AcceptedCatalogSnapshotSelection>, InternalError> {
1177 let first = self.canonical_root_slot_bytes(0)?;
1178 let second = self.canonical_root_slot_bytes(1)?;
1179 let Some(selection) =
1180 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1181 else {
1182 return Ok(None);
1183 };
1184 let bundle_key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1185 let raw_bundle = self
1186 .get_canonical_raw_value(&bundle_key)?
1187 .ok_or_else(InternalError::store_corruption)?;
1188 let bundle = decode_verified_accepted_schema_revision_bundle(
1189 selection.root(),
1190 raw_bundle.as_bytes(),
1191 )?;
1192 if bundle.store_path() != store_path {
1193 return Err(InternalError::store_corruption());
1194 }
1195 let Some(snapshot) = bundle.entity_snapshots().get(&entity) else {
1196 return Ok(None);
1197 };
1198 if snapshot.entity_path() != entity_path {
1199 return Err(InternalError::store_corruption());
1200 }
1201
1202 let raw_snapshot = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1203 let fingerprint = raw_snapshot.accepted_schema_fingerprint()?;
1204 let identity = AcceptedCatalogIdentity::new(
1205 entity,
1206 entity_path,
1207 store_path,
1208 bundle.revision(),
1209 snapshot.version(),
1210 fingerprint,
1211 );
1212
1213 Ok(Some(AcceptedCatalogSnapshotSelection::new(
1214 identity,
1215 AcceptedValueCatalogHandle::new(
1216 bundle.enum_catalog().clone(),
1217 bundle.composite_catalog().clone(),
1218 self.accepted_catalog_scope
1219 .get_or_init(AcceptedStoreCatalogScope::new)
1220 .clone(),
1221 bundle.revision(),
1222 selection.root().fingerprint(),
1223 ),
1224 raw_snapshot.into_bytes(),
1225 )))
1226 }
1227
1228 #[must_use]
1230 pub(in crate::db) fn entity_footprint(&self, entity: EntityTag) -> SchemaStoreFootprint {
1231 let mut snapshots = 0u64;
1232 let mut encoded_bytes = 0u64;
1233 let mut latest = None::<(SchemaVersion, u64)>;
1234
1235 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
1236 if key.entity_tag() != entity {
1237 return Ok(SchemaStoreVisit::Continue);
1238 }
1239
1240 let snapshot_bytes = u64::try_from(snapshot.as_bytes().len()).unwrap_or(u64::MAX);
1241 snapshots = snapshots.saturating_add(1);
1242 encoded_bytes = encoded_bytes.saturating_add(snapshot_bytes);
1243
1244 let version = SchemaVersion::new(key.version());
1245 if latest
1246 .as_ref()
1247 .is_none_or(|(latest_version, _)| version > *latest_version)
1248 {
1249 latest = Some((version, snapshot_bytes));
1250 }
1251 Ok(SchemaStoreVisit::Continue)
1252 });
1253
1254 SchemaStoreFootprint::new(
1255 snapshots,
1256 encoded_bytes,
1257 latest.map_or(0, |(_, snapshot_bytes)| snapshot_bytes),
1258 )
1259 }
1260
1261 #[cfg(test)]
1267 pub(in crate::db) fn catalog_metadata(
1268 &self,
1269 ) -> Result<Option<SchemaStoreCatalogMetadata>, InternalError> {
1270 Ok(self
1271 .allocation_metadata()?
1272 .map(SchemaStoreAllocationMetadata::schema))
1273 }
1274
1275 pub(in crate::db) fn allocation_metadata(
1282 &self,
1283 ) -> Result<Option<SchemaStoreAllocationMetadata>, InternalError> {
1284 let latest_by_entity = self.latest_raw_snapshots_by_entity();
1285 if latest_by_entity.is_empty() {
1286 return Ok(None);
1287 }
1288
1289 Ok(Some(SchemaStoreAllocationMetadata::new(
1290 derive_data_allocation_metadata(&latest_by_entity)?,
1291 derive_index_allocation_metadata(&latest_by_entity)?,
1292 derive_schema_catalog_metadata(&latest_by_entity)?,
1293 )))
1294 }
1295
1296 fn insert_raw_snapshot(
1298 &mut self,
1299 key: RawSchemaKey,
1300 snapshot: RawSchemaSnapshot,
1301 ) -> Option<RawSchemaSnapshot> {
1302 self.invalidate_accepted_bundle_cache_for_key(key);
1303 let previous_journaled = if matches!(self.backend, SchemaStoreBackend::Journaled { .. }) {
1304 self.get_raw_snapshot_for_backend(&key)
1305 } else {
1306 None
1307 };
1308 match &mut self.backend {
1309 SchemaStoreBackend::Heap(map) => map.insert(key, snapshot),
1310 SchemaStoreBackend::Journaled {
1311 live, tombstones, ..
1312 } => {
1313 tombstones.remove(&key);
1314 live.insert(key, snapshot);
1315 previous_journaled
1316 }
1317 }
1318 }
1319
1320 #[must_use]
1322 fn get_raw_snapshot(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
1323 match &self.backend {
1324 SchemaStoreBackend::Heap(map) => map.get(key).cloned(),
1325 SchemaStoreBackend::Journaled { .. } => self.get_raw_snapshot_for_backend(key),
1326 }
1327 }
1328
1329 fn accepted_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1330 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1331 Ok(self
1332 .get_raw_snapshot(&key)
1333 .map(RawSchemaSnapshot::into_bytes))
1334 }
1335
1336 fn canonical_root_slot_bytes(&self, slot: usize) -> Result<Option<Vec<u8>>, InternalError> {
1337 let key = RawSchemaKey::from_accepted_root_slot(slot)?;
1338 Ok(self
1339 .get_canonical_raw_value(&key)?
1340 .map(RawSchemaSnapshot::into_bytes))
1341 }
1342
1343 fn current_root_matches_candidate(
1344 &self,
1345 candidate: &CandidateSchemaRevision,
1346 ) -> Result<bool, InternalError> {
1347 let Some(selection) = self.current_accepted_schema_root()? else {
1348 return Ok(false);
1349 };
1350 if selection.root() != candidate.root() {
1351 return Ok(false);
1352 }
1353 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1354 let bundle = self
1355 .get_raw_snapshot(&key)
1356 .ok_or_else(InternalError::store_corruption)?;
1357 let _verified =
1358 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1359 Ok(true)
1360 }
1361
1362 fn canonical_root_matches_candidate(
1363 &self,
1364 candidate: &CandidateSchemaRevision,
1365 ) -> Result<bool, InternalError> {
1366 let first = self.canonical_root_slot_bytes(0)?;
1367 let second = self.canonical_root_slot_bytes(1)?;
1368 let Some(selection) =
1369 select_current_accepted_schema_root([first.as_deref(), second.as_deref()])?
1370 else {
1371 return Ok(false);
1372 };
1373 if selection.root() != candidate.root() {
1374 return Ok(false);
1375 }
1376 let key = RawSchemaKey::from_accepted_bundle(candidate.root().bundle_key());
1377 let bundle = self
1378 .get_canonical_raw_value(&key)?
1379 .ok_or_else(InternalError::store_corruption)?;
1380 let _verified =
1381 decode_verified_accepted_schema_revision_bundle(candidate.root(), bundle.as_bytes())?;
1382 Ok(true)
1383 }
1384
1385 fn get_canonical_raw_value(
1386 &self,
1387 key: &RawSchemaKey,
1388 ) -> Result<Option<RawSchemaSnapshot>, InternalError> {
1389 match &self.backend {
1390 SchemaStoreBackend::Journaled { canonical, .. } => Ok(canonical.get(key)),
1391 SchemaStoreBackend::Heap(_) => Err(InternalError::store_invariant()),
1392 }
1393 }
1394
1395 fn insert_canonical_raw_value(
1396 &mut self,
1397 key: RawSchemaKey,
1398 bytes: Vec<u8>,
1399 ) -> Result<(), InternalError> {
1400 self.invalidate_accepted_bundle_cache_for_key(key);
1401 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1402 return Err(InternalError::store_invariant());
1403 };
1404 canonical.insert(key, RawSchemaSnapshot::from_encoded_control_record(bytes));
1405 Ok(())
1406 }
1407
1408 fn insert_durable_raw_value(&mut self, key: RawSchemaKey, bytes: Vec<u8>) {
1412 self.invalidate_accepted_bundle_cache_for_key(key);
1413 let value = RawSchemaSnapshot::from_encoded_control_record(bytes);
1414 match &mut self.backend {
1415 SchemaStoreBackend::Heap(map) => {
1416 map.insert(key, value);
1417 }
1418 SchemaStoreBackend::Journaled {
1419 canonical,
1420 live,
1421 tombstones,
1422 } => {
1423 live.remove(&key);
1424 tombstones.remove(&key);
1425 canonical.insert(key, value);
1426 }
1427 }
1428 }
1429
1430 fn invalidate_accepted_bundle_cache_for_key(&mut self, key: RawSchemaKey) {
1431 if key.is_accepted_root() {
1432 self.accepted_bundle_cache.get_mut().take();
1433 }
1434 }
1435
1436 fn insert_durable_candidate_snapshots(
1437 &mut self,
1438 candidate: &CandidateSchemaRevision,
1439 ) -> Result<(), InternalError> {
1440 for (entity_tag, snapshot) in candidate.bundle().entity_snapshots() {
1441 let key = RawSchemaKey::from_entity_version(*entity_tag, snapshot.version());
1442 let value = RawSchemaSnapshot::from_persisted_snapshot(snapshot)?;
1443 match &mut self.backend {
1444 SchemaStoreBackend::Heap(map) => {
1445 map.insert(key, value);
1446 }
1447 SchemaStoreBackend::Journaled {
1448 canonical,
1449 live,
1450 tombstones,
1451 } => {
1452 live.remove(&key);
1453 tombstones.remove(&key);
1454 canonical.insert(key, value);
1455 }
1456 }
1457 }
1458 Ok(())
1459 }
1460
1461 fn candidate_entry_keys(
1462 candidate: &CandidateSchemaRevision,
1463 root_slot: usize,
1464 ) -> Result<BTreeSet<RawSchemaKey>, InternalError> {
1465 let mut keys = candidate
1466 .bundle()
1467 .entity_snapshots()
1468 .iter()
1469 .map(|(entity_tag, snapshot)| {
1470 RawSchemaKey::from_entity_version(*entity_tag, snapshot.version())
1471 })
1472 .collect::<BTreeSet<_>>();
1473 keys.insert(RawSchemaKey::from_accepted_bundle(
1474 candidate.root().bundle_key(),
1475 ));
1476 keys.insert(RawSchemaKey::from_accepted_root_slot(root_slot)?);
1477 Ok(keys)
1478 }
1479
1480 fn retain_durable_candidate_entries(
1484 &mut self,
1485 candidate: &CandidateSchemaRevision,
1486 root_slot: usize,
1487 ) -> Result<(), InternalError> {
1488 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1489 self.accepted_bundle_cache.get_mut().take();
1490 match &mut self.backend {
1491 SchemaStoreBackend::Heap(map) => map.retain(|key, _| keep.contains(key)),
1492 SchemaStoreBackend::Journaled {
1493 canonical,
1494 live,
1495 tombstones,
1496 } => {
1497 let stale = canonical
1498 .iter()
1499 .filter_map(|entry| (!keep.contains(entry.key())).then_some(*entry.key()))
1500 .collect::<Vec<_>>();
1501 for key in stale {
1502 canonical.remove(&key);
1503 }
1504 live.retain(|key, _| keep.contains(key));
1505 tombstones.clear();
1506 }
1507 }
1508 Ok(())
1509 }
1510
1511 fn retain_materialized_candidate_entries(
1512 &mut self,
1513 candidate: &CandidateSchemaRevision,
1514 root_slot: usize,
1515 ) -> Result<(), InternalError> {
1516 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1517 self.accepted_bundle_cache.get_mut().take();
1518 let SchemaStoreBackend::Journaled {
1519 canonical,
1520 live,
1521 tombstones,
1522 } = &mut self.backend
1523 else {
1524 return Err(InternalError::store_invariant());
1525 };
1526 live.retain(|key, _| keep.contains(key));
1527 let canonical_keys = canonical
1528 .iter()
1529 .map(|entry| *entry.key())
1530 .collect::<Vec<_>>();
1531 for key in canonical_keys {
1532 if keep.contains(&key) {
1533 tombstones.remove(&key);
1534 } else {
1535 tombstones.insert(key);
1536 }
1537 }
1538 Ok(())
1539 }
1540
1541 fn retain_canonical_candidate_entries(
1542 &mut self,
1543 candidate: &CandidateSchemaRevision,
1544 root_slot: usize,
1545 ) -> Result<(), InternalError> {
1546 let keep = Self::candidate_entry_keys(candidate, root_slot)?;
1547 self.accepted_bundle_cache.get_mut().take();
1548 let SchemaStoreBackend::Journaled { canonical, .. } = &mut self.backend else {
1549 return Err(InternalError::store_invariant());
1550 };
1551 let stale = canonical
1552 .iter()
1553 .filter_map(|entry| (!keep.contains(entry.key())).then_some(*entry.key()))
1554 .collect::<Vec<_>>();
1555 for key in stale {
1556 canonical.remove(&key);
1557 }
1558 Ok(())
1559 }
1560
1561 #[must_use]
1563 #[cfg(test)]
1564 fn contains_raw_snapshot(&self, key: &RawSchemaKey) -> bool {
1565 match &self.backend {
1566 SchemaStoreBackend::Heap(map) => map.contains_key(key),
1567 SchemaStoreBackend::Journaled { .. } => {
1568 self.get_raw_snapshot_for_backend(key).is_some()
1569 }
1570 }
1571 }
1572
1573 #[must_use]
1575 #[cfg(test)]
1576 pub(in crate::db) fn len(&self) -> u64 {
1577 match &self.backend {
1578 SchemaStoreBackend::Heap(map) => u64::try_from(map.len()).unwrap_or(u64::MAX),
1579 SchemaStoreBackend::Journaled { .. } => {
1580 let mut count = 0_u64;
1581 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
1582 count = count.saturating_add(1);
1583 Ok(SchemaStoreVisit::Continue)
1584 });
1585 count
1586 }
1587 }
1588 }
1589
1590 #[must_use]
1592 #[cfg(test)]
1593 pub(in crate::db) fn is_empty(&self) -> bool {
1594 match &self.backend {
1595 SchemaStoreBackend::Heap(map) => map.is_empty(),
1596 SchemaStoreBackend::Journaled { .. } => {
1597 let mut empty = true;
1598 let _: Result<(), Infallible> = self.visit_raw_snapshots(|_key, _snapshot| {
1599 empty = false;
1600 Ok(SchemaStoreVisit::Stop)
1601 });
1602 empty
1603 }
1604 }
1605 }
1606
1607 #[cfg(test)]
1609 pub(in crate::db) fn clear(&mut self) {
1610 self.accepted_bundle_cache.get_mut().take();
1611 match &mut self.backend {
1612 SchemaStoreBackend::Heap(map) => map.clear(),
1613 SchemaStoreBackend::Journaled {
1614 canonical,
1615 live,
1616 tombstones,
1617 } => {
1618 live.clear();
1619 tombstones.clear();
1620 let keys = canonical
1621 .iter()
1622 .map(|entry| *entry.key())
1623 .collect::<Vec<_>>();
1624 for key in keys {
1625 if key.is_entity_snapshot() {
1626 tombstones.insert(key);
1627 } else {
1628 canonical.remove(&key);
1629 }
1630 }
1631 }
1632 }
1633 }
1634
1635 fn current_accepted_schema_bundle_ref(
1636 &self,
1637 ) -> Result<Option<Ref<'_, AcceptedSchemaRevisionBundle>>, InternalError> {
1638 let Some(selection) = self.current_accepted_schema_root()? else {
1639 self.accepted_bundle_cache
1640 .try_borrow_mut()
1641 .map_err(|_| InternalError::store_invariant())?
1642 .take();
1643 return Ok(None);
1644 };
1645
1646 let cache_matches = self
1647 .accepted_bundle_cache
1648 .try_borrow()
1649 .map_err(|_| InternalError::store_invariant())?
1650 .as_ref()
1651 .is_some_and(|cached| cached.selection == selection);
1652 if !cache_matches {
1653 let key = RawSchemaKey::from_accepted_bundle(selection.root().bundle_key());
1654 let raw = self
1655 .get_raw_snapshot(&key)
1656 .ok_or_else(InternalError::store_corruption)?;
1657 let bundle =
1658 decode_verified_accepted_schema_revision_bundle(selection.root(), raw.as_bytes())?;
1659 #[cfg(test)]
1660 ACCEPTED_SCHEMA_BUNDLE_CACHE_MISSES
1661 .with(|misses| misses.set(misses.get().saturating_add(1)));
1662 *self
1663 .accepted_bundle_cache
1664 .try_borrow_mut()
1665 .map_err(|_| InternalError::store_invariant())? =
1666 Some(AcceptedSchemaBundleCache { selection, bundle });
1667 }
1668
1669 let cache = self
1670 .accepted_bundle_cache
1671 .try_borrow()
1672 .map_err(|_| InternalError::store_invariant())?;
1673 Ref::filter_map(cache, |cache| {
1674 cache
1675 .as_ref()
1676 .filter(|cached| cached.selection == selection)
1677 .map(|cached| &cached.bundle)
1678 })
1679 .map(Some)
1680 .map_err(|_| InternalError::store_invariant())
1681 }
1682
1683 fn latest_raw_snapshots_by_entity(
1684 &self,
1685 ) -> StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)> {
1686 #[cfg(test)]
1687 LATEST_RAW_SNAPSHOTS_BY_ENTITY_CALLS.with(|calls| calls.set(calls.get().saturating_add(1)));
1688
1689 let mut latest_by_entity =
1690 StdBTreeMap::<EntityTag, (SchemaVersion, RawSchemaSnapshot)>::new();
1691
1692 let _: Result<(), std::convert::Infallible> = self.visit_raw_snapshots(|key, snapshot| {
1693 let version = SchemaVersion::new(key.version());
1694 match latest_by_entity.get_mut(&key.entity_tag()) {
1695 Some((latest_version, latest_snapshot)) if version > *latest_version => {
1696 *latest_version = version;
1697 *latest_snapshot = snapshot.clone();
1698 }
1699 None => {
1700 latest_by_entity.insert(key.entity_tag(), (version, snapshot.clone()));
1701 }
1702 Some(_) => {}
1703 }
1704 Ok(SchemaStoreVisit::Continue)
1705 });
1706
1707 latest_by_entity
1708 }
1709
1710 fn visit_raw_snapshots<E>(
1713 &self,
1714 visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
1715 ) -> Result<(), E> {
1716 let bounds = RawSchemaKey::all_entity_range_bounds();
1717 match &self.backend {
1718 SchemaStoreBackend::Heap(map) => {
1719 let mut visitor = visitor;
1720 for (key, snapshot) in map.range((bounds.0, bounds.1)) {
1721 if visitor(key, snapshot)?.should_stop() {
1722 break;
1723 }
1724 }
1725 }
1726 SchemaStoreBackend::Journaled {
1727 canonical,
1728 live,
1729 tombstones,
1730 } => Self::visit_journaled_raw_snapshot_range(
1731 canonical,
1732 live,
1733 tombstones,
1734 bounds,
1735 Direction::Asc,
1736 visitor,
1737 )?,
1738 }
1739
1740 Ok(())
1741 }
1742
1743 #[cfg(test)]
1744 #[must_use]
1745 pub(in crate::db) fn canonical_len_for_tests(&self) -> u64 {
1746 match &self.backend {
1747 SchemaStoreBackend::Journaled { canonical: map, .. } => map.len(),
1748 SchemaStoreBackend::Heap(_) => 0,
1749 }
1750 }
1751
1752 fn get_raw_snapshot_for_backend(&self, key: &RawSchemaKey) -> Option<RawSchemaSnapshot> {
1753 let SchemaStoreBackend::Journaled {
1754 canonical,
1755 live,
1756 tombstones,
1757 } = &self.backend
1758 else {
1759 return None;
1760 };
1761
1762 if tombstones.contains(key) {
1763 return None;
1764 }
1765 live.get(key).cloned().or_else(|| canonical.get(key))
1766 }
1767
1768 fn latest_raw_snapshot(&self, entity: EntityTag) -> Option<RawSchemaSnapshot> {
1769 self.latest_raw_snapshot_entry(entity)
1770 .map(|(_, snapshot)| snapshot)
1771 }
1772
1773 fn latest_raw_snapshot_entry(
1774 &self,
1775 entity: EntityTag,
1776 ) -> Option<(SchemaVersion, RawSchemaSnapshot)> {
1777 let bounds = RawSchemaKey::entity_range_bounds(entity);
1778 match &self.backend {
1779 SchemaStoreBackend::Heap(map) => map
1780 .range((bounds.0, bounds.1))
1781 .next_back()
1782 .map(|(key, snapshot)| (SchemaVersion::new(key.version()), snapshot.clone())),
1783 SchemaStoreBackend::Journaled {
1784 canonical,
1785 live,
1786 tombstones,
1787 } => {
1788 let mut latest = None;
1789 let _: Result<(), Infallible> = Self::visit_journaled_raw_snapshot_range(
1790 canonical,
1791 live,
1792 tombstones,
1793 bounds,
1794 Direction::Desc,
1795 |key, snapshot| {
1796 latest = Some((SchemaVersion::new(key.version()), snapshot.clone()));
1797 Ok(SchemaStoreVisit::Stop)
1798 },
1799 );
1800 latest
1801 }
1802 }
1803 }
1804
1805 fn visit_journaled_raw_snapshot_range<E>(
1806 canonical: &StableBTreeMap<
1807 RawSchemaKey,
1808 RawSchemaSnapshot,
1809 VirtualMemory<DefaultMemoryImpl>,
1810 >,
1811 live: &StdBTreeMap<RawSchemaKey, RawSchemaSnapshot>,
1812 tombstones: &BTreeSet<RawSchemaKey>,
1813 bounds: (RangeBound<RawSchemaKey>, RangeBound<RawSchemaKey>),
1814 direction: Direction,
1815 mut visitor: impl FnMut(&RawSchemaKey, &RawSchemaSnapshot) -> Result<SchemaStoreVisit, E>,
1816 ) -> Result<(), E> {
1817 match direction {
1818 Direction::Asc => visit_ordered_overlay(
1819 canonical.range((bounds.0, bounds.1)),
1820 live.range((bounds.0, bounds.1)),
1821 Direction::Asc,
1822 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1823 |canonical_entry| !tombstones.contains(canonical_entry.key()),
1824 |live_entry| !tombstones.contains(live_entry.0),
1825 |entry| {
1826 let visit = match entry {
1827 OrderedOverlayEntry::Canonical(canonical_entry) => {
1828 visitor(canonical_entry.key(), &canonical_entry.value())?
1829 }
1830 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
1831 };
1832 Ok(if visit.should_stop() {
1833 OrderedOverlayVisit::Stop
1834 } else {
1835 OrderedOverlayVisit::Continue
1836 })
1837 },
1838 ),
1839 Direction::Desc => visit_ordered_overlay(
1840 canonical.range((bounds.0, bounds.1)).rev(),
1841 live.range((bounds.0, bounds.1)).rev(),
1842 Direction::Desc,
1843 |canonical_entry, live_entry| canonical_entry.key().cmp(live_entry.0),
1844 |canonical_entry| !tombstones.contains(canonical_entry.key()),
1845 |live_entry| !tombstones.contains(live_entry.0),
1846 |entry| {
1847 let visit = match entry {
1848 OrderedOverlayEntry::Canonical(canonical_entry) => {
1849 visitor(canonical_entry.key(), &canonical_entry.value())?
1850 }
1851 OrderedOverlayEntry::Live((key, snapshot)) => visitor(key, snapshot)?,
1852 };
1853 Ok(if visit.should_stop() {
1854 OrderedOverlayVisit::Stop
1855 } else {
1856 OrderedOverlayVisit::Continue
1857 })
1858 },
1859 ),
1860 }
1861 }
1862}
1863
1864fn map_schema_publication_error(error: AcceptedSchemaPublicationError) -> InternalError {
1865 match error {
1866 AcceptedSchemaPublicationError::StaleSchemaRevision { .. }
1867 | AcceptedSchemaPublicationError::RevisionExhausted => InternalError::store_unsupported(),
1868 AcceptedSchemaPublicationError::InvalidCandidate => InternalError::store_invariant(),
1869 AcceptedSchemaPublicationError::CorruptRootSlots => InternalError::store_corruption(),
1870 }
1871}
1872
1873fn derive_data_allocation_metadata(
1874 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1875) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1876 let mut max_version = SchemaVersion::initial();
1877 let mut hasher = new_hash_sha256();
1878 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_DATA_ALLOCATION_FINGERPRINT_DOMAIN);
1879
1880 for (entity, (_, snapshot)) in latest_by_entity {
1881 let persisted = snapshot.decode_persisted_snapshot()?;
1882 if persisted.version() > max_version {
1883 max_version = persisted.version();
1884 }
1885
1886 let data_projection = PersistedSchemaSnapshot::new_with_primary_key_fields_and_indexes(
1887 persisted.version(),
1888 persisted.entity_path().to_string(),
1889 persisted.entity_name().to_string(),
1890 persisted.primary_key_field_ids().to_vec(),
1891 persisted.row_layout().clone(),
1892 persisted.fields().to_vec(),
1893 Vec::new(),
1894 );
1895 let encoded = encode_persisted_schema_snapshot(&data_projection)?;
1896
1897 write_hash_u64(&mut hasher, entity.value());
1898 write_hash_u32(&mut hasher, persisted.version().get());
1899 write_hash_len_u32(&mut hasher, encoded.len());
1900 hasher.update(encoded);
1901 }
1902
1903 Ok(finalize_schema_metadata(
1904 max_version,
1905 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
1906 hasher,
1907 latest_by_entity.len(),
1908 ))
1909}
1910
1911fn derive_index_allocation_metadata(
1912 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1913) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1914 let mut max_version = SchemaVersion::initial();
1915 let mut hasher = new_hash_sha256();
1916 write_hash_tag_u8(
1917 &mut hasher,
1918 SCHEMA_STORE_INDEX_ALLOCATION_FINGERPRINT_DOMAIN,
1919 );
1920
1921 for (entity, (_, snapshot)) in latest_by_entity {
1922 let persisted = snapshot.decode_persisted_snapshot()?;
1923 if persisted.version() > max_version {
1924 max_version = persisted.version();
1925 }
1926
1927 write_hash_u64(&mut hasher, entity.value());
1928 write_hash_u32(&mut hasher, persisted.version().get());
1929 write_hash_len_u32(&mut hasher, persisted.indexes().len());
1930 for index in persisted.indexes() {
1931 write_hash_u32(&mut hasher, u32::from(index.ordinal()));
1932 write_hash_str_u32(&mut hasher, index.name());
1933 write_hash_str_u32(&mut hasher, index.store());
1934 write_hash_tag_u8(&mut hasher, u8::from(index.unique()));
1935 write_hash_str_u32(&mut hasher, persisted_index_origin_name(index.origin()));
1936 match index.predicate_sql() {
1937 Some(predicate_sql) => {
1938 write_hash_tag_u8(&mut hasher, 1);
1939 write_hash_str_u32(&mut hasher, predicate_sql);
1940 }
1941 None => write_hash_tag_u8(&mut hasher, 0),
1942 }
1943 hash_persisted_index_key(&mut hasher, index.key());
1944 }
1945 }
1946
1947 Ok(finalize_schema_metadata(
1948 max_version,
1949 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
1950 hasher,
1951 latest_by_entity.len(),
1952 ))
1953}
1954
1955fn derive_schema_catalog_metadata(
1956 latest_by_entity: &StdBTreeMap<EntityTag, (SchemaVersion, RawSchemaSnapshot)>,
1957) -> Result<SchemaStoreCatalogMetadata, InternalError> {
1958 let mut max_version = SchemaVersion::initial();
1959 let mut hasher = new_hash_sha256();
1960 write_hash_tag_u8(&mut hasher, SCHEMA_STORE_CATALOG_FINGERPRINT_DOMAIN);
1961
1962 for (entity, (version, snapshot)) in latest_by_entity {
1963 let persisted = snapshot.decode_persisted_snapshot()?;
1964 if persisted.version() > max_version {
1965 max_version = persisted.version();
1966 }
1967
1968 write_hash_u64(&mut hasher, entity.value());
1969 write_hash_u32(&mut hasher, version.get());
1970 write_hash_len_u32(&mut hasher, snapshot.as_bytes().len());
1971 hasher.update(snapshot.as_bytes());
1972 }
1973
1974 Ok(finalize_schema_metadata(
1975 max_version,
1976 SCHEMA_STORE_FINGERPRINT_METHOD_VERSION,
1977 hasher,
1978 latest_by_entity.len(),
1979 ))
1980}
1981
1982fn finalize_schema_metadata(
1983 schema_version: SchemaVersion,
1984 schema_fingerprint_method_version: u8,
1985 hasher: sha2::Sha256,
1986 entity_count: usize,
1987) -> SchemaStoreCatalogMetadata {
1988 let digest = finalize_hash_sha256(hasher);
1989 let mut schema_fingerprint = [0u8; 16];
1990 schema_fingerprint.copy_from_slice(&digest[..16]);
1991
1992 SchemaStoreCatalogMetadata::new(
1993 schema_version,
1994 schema_fingerprint_method_version,
1995 schema_fingerprint,
1996 u64::try_from(entity_count).unwrap_or(u64::MAX),
1997 )
1998}
1999
2000fn hash_persisted_index_key(hasher: &mut sha2::Sha256, key: &PersistedIndexKeySnapshot) {
2001 match key {
2002 PersistedIndexKeySnapshot::FieldPath(paths) => {
2003 write_hash_tag_u8(hasher, 1);
2004 write_hash_len_u32(hasher, paths.len());
2005 for path in paths {
2006 hash_persisted_index_field_path(hasher, path);
2007 }
2008 }
2009 PersistedIndexKeySnapshot::Items(items) => {
2010 write_hash_tag_u8(hasher, 2);
2011 write_hash_len_u32(hasher, items.len());
2012 for item in items {
2013 match item {
2014 PersistedIndexKeyItemSnapshot::FieldPath(path) => {
2015 write_hash_tag_u8(hasher, 1);
2016 hash_persisted_index_field_path(hasher, path);
2017 }
2018 PersistedIndexKeyItemSnapshot::Expression(expression) => {
2019 write_hash_tag_u8(hasher, 2);
2020 write_hash_str_u32(hasher, persisted_expression_op_name(expression.op()));
2021 hash_persisted_index_field_path(hasher, expression.source());
2022 hash_accepted_field_kind(hasher, expression.input_kind());
2023 hash_accepted_field_kind(hasher, expression.output_kind());
2024 write_hash_str_u32(hasher, expression.canonical_text());
2025 }
2026 }
2027 }
2028 }
2029 }
2030}
2031
2032fn hash_persisted_index_field_path(
2033 hasher: &mut sha2::Sha256,
2034 path: &crate::db::schema::PersistedIndexFieldPathSnapshot,
2035) {
2036 write_hash_u32(hasher, path.field_id().get());
2037 write_hash_u32(hasher, u32::from(path.slot().get()));
2038 write_hash_len_u32(hasher, path.path().len());
2039 for segment in path.path() {
2040 write_hash_str_u32(hasher, segment);
2041 }
2042 hash_accepted_field_kind(hasher, path.kind());
2043 write_hash_tag_u8(hasher, u8::from(path.nullable()));
2044}
2045
2046fn hash_accepted_field_kind(hasher: &mut sha2::Sha256, kind: &AcceptedFieldKind) {
2047 match kind {
2048 AcceptedFieldKind::Account => write_hash_tag_u8(hasher, 1),
2049 AcceptedFieldKind::Blob { max_len } => {
2050 write_hash_tag_u8(hasher, 2);
2051 hash_optional_u32(hasher, *max_len);
2052 }
2053 AcceptedFieldKind::Bool => {
2054 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_BOOL);
2055 }
2056 AcceptedFieldKind::Date => write_hash_tag_u8(hasher, 4),
2057 AcceptedFieldKind::Decimal { scale } => {
2058 write_hash_tag_u8(hasher, 5);
2059 write_hash_u32(hasher, *scale);
2060 }
2061 AcceptedFieldKind::Duration => write_hash_tag_u8(hasher, 6),
2062 AcceptedFieldKind::Enum { type_id } => {
2063 write_hash_tag_u8(hasher, 7);
2064 write_hash_u32(hasher, type_id.get());
2065 }
2066 AcceptedFieldKind::Float32 => write_hash_tag_u8(hasher, 8),
2067 AcceptedFieldKind::Float64 => write_hash_tag_u8(hasher, 9),
2068 AcceptedFieldKind::Int8 => write_hash_tag_u8(hasher, 10),
2069 AcceptedFieldKind::Int16 => write_hash_tag_u8(hasher, 11),
2070 AcceptedFieldKind::Int32 => write_hash_tag_u8(hasher, 12),
2071 AcceptedFieldKind::Int64 => write_hash_tag_u8(hasher, 13),
2072 AcceptedFieldKind::Int128 => write_hash_tag_u8(hasher, 14),
2073 AcceptedFieldKind::IntBig { max_bytes } => {
2074 write_hash_tag_u8(hasher, 15);
2075 write_hash_u32(hasher, *max_bytes);
2076 }
2077 AcceptedFieldKind::Principal => write_hash_tag_u8(hasher, 16),
2078 AcceptedFieldKind::Subaccount => write_hash_tag_u8(hasher, 17),
2079 AcceptedFieldKind::Text { max_len } => {
2080 write_hash_tag_u8(hasher, 18);
2081 hash_optional_u32(hasher, *max_len);
2082 }
2083 AcceptedFieldKind::Timestamp => write_hash_tag_u8(hasher, 19),
2084 AcceptedFieldKind::Nat8 => write_hash_tag_u8(hasher, 20),
2085 AcceptedFieldKind::Nat16 => write_hash_tag_u8(hasher, 21),
2086 AcceptedFieldKind::Nat32 => write_hash_tag_u8(hasher, 22),
2087 AcceptedFieldKind::Nat64 => write_hash_tag_u8(hasher, 23),
2088 AcceptedFieldKind::Nat128 => write_hash_tag_u8(hasher, 24),
2089 AcceptedFieldKind::NatBig { max_bytes } => {
2090 write_hash_tag_u8(hasher, 25);
2091 write_hash_u32(hasher, *max_bytes);
2092 }
2093 AcceptedFieldKind::Ulid => write_hash_tag_u8(hasher, 26),
2094 AcceptedFieldKind::Unit => write_hash_tag_u8(hasher, 27),
2095 AcceptedFieldKind::Relation {
2096 target_path,
2097 target_entity_name,
2098 target_entity_tag,
2099 target_store_path,
2100 key_kind,
2101 } => {
2102 write_hash_tag_u8(hasher, 28);
2103 write_hash_str_u32(hasher, target_path);
2104 write_hash_str_u32(hasher, target_entity_name);
2105 write_hash_u64(hasher, target_entity_tag.value());
2106 write_hash_str_u32(hasher, target_store_path);
2107 hash_accepted_field_kind(hasher, key_kind);
2108 }
2109 AcceptedFieldKind::List(inner) => {
2110 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_LIST);
2111 hash_accepted_field_kind(hasher, inner);
2112 }
2113 AcceptedFieldKind::Set(inner) => {
2114 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_SET);
2115 hash_accepted_field_kind(hasher, inner);
2116 }
2117 AcceptedFieldKind::Map { key, value } => {
2118 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_MAP);
2119 hash_accepted_field_kind(hasher, key);
2120 hash_accepted_field_kind(hasher, value);
2121 }
2122 AcceptedFieldKind::Composite { type_id } => {
2123 write_hash_tag_u8(hasher, ACCEPTED_FIELD_KIND_FINGERPRINT_TAG_COMPOSITE);
2124 write_hash_u32(hasher, type_id.get());
2125 }
2126 }
2127}
2128
2129fn hash_optional_u32(hasher: &mut sha2::Sha256, value: Option<u32>) {
2130 match value {
2131 Some(value) => {
2132 write_hash_tag_u8(hasher, 1);
2133 write_hash_u32(hasher, value);
2134 }
2135 None => write_hash_tag_u8(hasher, 0),
2136 }
2137}
2138
2139const fn persisted_index_origin_name(
2140 origin: crate::db::schema::PersistedIndexOrigin,
2141) -> &'static str {
2142 match origin {
2143 crate::db::schema::PersistedIndexOrigin::Generated => "generated",
2144 crate::db::schema::PersistedIndexOrigin::SqlDdl => "sql_ddl",
2145 }
2146}
2147
2148const fn persisted_expression_op_name(
2149 op: crate::db::schema::PersistedIndexExpressionOp,
2150) -> &'static str {
2151 match op {
2152 crate::db::schema::PersistedIndexExpressionOp::Lower => "lower",
2153 crate::db::schema::PersistedIndexExpressionOp::Upper => "upper",
2154 crate::db::schema::PersistedIndexExpressionOp::Trim => "trim",
2155 crate::db::schema::PersistedIndexExpressionOp::LowerTrim => "lower_trim",
2156 crate::db::schema::PersistedIndexExpressionOp::Date => "date",
2157 crate::db::schema::PersistedIndexExpressionOp::Year => "year",
2158 crate::db::schema::PersistedIndexExpressionOp::Month => "month",
2159 crate::db::schema::PersistedIndexExpressionOp::Day => "day",
2160 }
2161}
2162
2163#[cfg(test)]
2168mod tests;