1#[cfg(feature = "bench-internals")]
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::{
6 collections::{
7 BTreeMap, HashMap, HashSet,
8 hash_map::{DefaultHasher, Entry},
9 },
10 hash::{Hash, Hasher},
11 num::{NonZeroU32, NonZeroUsize},
12 ops::Range,
13 str,
14 sync::Arc,
15 time::Instant,
16};
17
18use thiserror::Error;
19
20use crate::{
21 domain::{
22 DomainError, ObjectId, ObjectRef, ObjectType, RelationName, Relationship, SubjectId,
23 SubjectRef, SubjectType,
24 },
25 error::ZanzibarError,
26 model::{Object, Relation, User},
27 snapshot::{
28 BinaryCursor, IndexProfile, SectionKind, SnapshotEncodingLayout, SnapshotFormatVersion,
29 SnapshotIoError, SnapshotLoadPhaseTimings, SnapshotLoadProfile, SnapshotReader,
30 SnapshotSectionWriter, SnapshotValidationMode, checked_add_usize, checked_mul_usize,
31 checked_u32_from_usize, checked_usize_from_u32, checked_usize_from_u64, insert_unique,
32 },
33};
34
35const DEFAULT_QUERY_LIMIT: usize = 1_000;
36const MAX_MUTATIONS_PER_BATCH: usize = 10_000;
37const MAX_PRECONDITIONS_PER_BATCH: usize = 100;
38const COMPACT_DEAD_ROWS: usize = 100_000;
39const STORE_VIEW_MAX_DELTA_MUTATIONS: usize = 100_000;
40const STORE_VIEW_MAX_DELTA_TOMBSTONES: usize = 100_000;
41const DISK_SYMBOL_HASH_LEN: usize = 8;
42const DISK_INDEX_DIRECTORY_LEN: usize = 20;
43const DISK_INDEX_KEY_LEN: usize = 12;
44const DISK_POSTING_RANGE_LEN: usize = 12;
45const DISK_ROW_ID_LEN: usize = 4;
46const SECTION_WIDTH_MASK: u16 = 0b11;
47const SYMBOL_TABLE_LEN_WIDTH_SHIFT: u16 = 2;
48const SYMBOL_TABLE_LEN_WIDTH_MASK: u16 = 0b1100;
49const INDEX_DIRECTORY_FLAG_V3_COMPACT: u16 = 1;
50const INDEX_DIRECTORY_KEY_WIDTH_SHIFT: u16 = 1;
51const INDEX_DIRECTORY_KEY_WIDTH_MASK: u16 = 0b110;
52const SNAPSHOT_INDEX_KIND_COUNT: usize = 7;
53const SNAPSHOT_INDEX_KIND_COUNT_U64: u64 = SNAPSHOT_INDEX_KIND_COUNT as u64;
54#[cfg(feature = "bench-internals")]
55static DELTA_SEGMENTS_INSPECTED: AtomicU64 = AtomicU64::new(0);
56#[cfg(feature = "bench-internals")]
57static TOMBSTONE_CHECKS: AtomicU64 = AtomicU64::new(0);
58#[cfg(feature = "bench-internals")]
59static STORE_VIEW_QUERY_CALLS: AtomicU64 = AtomicU64::new(0);
60
61#[cfg(feature = "bench-internals")]
63#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
64#[non_exhaustive]
65pub struct StoreViewReadCounters {
66 pub query_calls: u64,
68 pub delta_segments_inspected: u64,
70 pub tombstone_checks: u64,
72}
73
74#[cfg(feature = "bench-internals")]
76#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
77#[non_exhaustive]
78pub struct StoreViewDeltaStats {
79 pub checkpoint_rows: usize,
81 pub delta_segments: usize,
83 pub delta_inserted_rows: usize,
85 pub delta_deleted_rows: usize,
87 pub delta_mutations: usize,
89 pub tombstone_ratio_bps: u16,
91}
92
93#[cfg(feature = "bench-internals")]
95#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
96#[non_exhaustive]
97pub struct StorePostingHistogram {
98 pub keys: u64,
100 pub total_postings: u64,
102 pub max_posting_len: u64,
104 pub singleton_keys: u64,
106 pub keys_2_to_4: u64,
108 pub keys_5_to_16: u64,
110 pub keys_17_to_64: u64,
112 pub keys_65_to_256: u64,
114 pub keys_257_to_1024: u64,
116 pub keys_1025_to_4096: u64,
118 pub keys_over_4096: u64,
120 pub estimated_row_id_bytes: u64,
122}
123
124#[cfg(feature = "bench-internals")]
126#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
127#[non_exhaustive]
128pub struct StorePostingHistograms {
129 pub resource: StorePostingHistogram,
131 pub resource_object: StorePostingHistogram,
133 pub resource_type_relation: StorePostingHistogram,
135 pub resource_type: StorePostingHistogram,
137 pub subject: StorePostingHistogram,
139 pub subject_type_relation: StorePostingHistogram,
141 pub subject_type: StorePostingHistogram,
143}
144
145#[cfg(feature = "bench-internals")]
146impl StorePostingHistogram {
147 fn record_posting_len(&mut self, len: usize) {
148 let len = u64_from_usize_saturating(len);
149 self.keys = self.keys.saturating_add(1);
150 self.total_postings = self.total_postings.saturating_add(len);
151 self.max_posting_len = self.max_posting_len.max(len);
152 self.estimated_row_id_bytes = self.estimated_row_id_bytes.saturating_add(
153 len.saturating_mul(u64_from_usize_saturating(std::mem::size_of::<RowId>())),
154 );
155 match len {
156 0 => {}
157 1 => self.singleton_keys = self.singleton_keys.saturating_add(1),
158 2..=4 => self.keys_2_to_4 = self.keys_2_to_4.saturating_add(1),
159 5..=16 => self.keys_5_to_16 = self.keys_5_to_16.saturating_add(1),
160 17..=64 => self.keys_17_to_64 = self.keys_17_to_64.saturating_add(1),
161 65..=256 => self.keys_65_to_256 = self.keys_65_to_256.saturating_add(1),
162 257..=1024 => self.keys_257_to_1024 = self.keys_257_to_1024.saturating_add(1),
163 1025..=4096 => {
164 self.keys_1025_to_4096 = self.keys_1025_to_4096.saturating_add(1);
165 }
166 _ => self.keys_over_4096 = self.keys_over_4096.saturating_add(1),
167 }
168 }
169}
170
171#[cfg(feature = "bench-internals")]
172#[derive(Debug, Default)]
173struct ActivePostingHistogramBuilder {
174 resource: HashMap<ResourceIndexKey, usize>,
175 resource_object: HashMap<ResourceObjectIndexKey, usize>,
176 resource_type_relation: HashMap<ResourceTypeRelationIndexKey, usize>,
177 resource_type: HashMap<ObjectTypeId, usize>,
178 subject: HashMap<SubjectIndexKey, usize>,
179 subject_type_relation: HashMap<SubjectTypeRelationIndexKey, usize>,
180 subject_type: HashMap<SubjectTypeId, usize>,
181}
182
183#[cfg(feature = "bench-internals")]
184impl ActivePostingHistogramBuilder {
185 fn record_row(&mut self, row: &RelationshipRow) {
186 increment_posting(&mut self.resource, ResourceIndexKey::from(row));
187 increment_posting(&mut self.resource_object, ResourceObjectIndexKey::from(row));
188 increment_posting(
189 &mut self.resource_type_relation,
190 ResourceTypeRelationIndexKey::from(row),
191 );
192 increment_posting(&mut self.resource_type, row.resource_type);
193 for key in SubjectIndexKey::from_row(row) {
194 increment_posting(&mut self.subject, key);
195 }
196 if let Some(key) = SubjectTypeRelationIndexKey::from_row(row) {
197 increment_posting(&mut self.subject_type_relation, key);
198 }
199 increment_posting(&mut self.subject_type, row.subject_type);
200 }
201
202 fn finish(self) -> StorePostingHistograms {
203 StorePostingHistograms {
204 resource: histogram_from_posting_lengths(self.resource.into_values()),
205 resource_object: histogram_from_posting_lengths(self.resource_object.into_values()),
206 resource_type_relation: histogram_from_posting_lengths(
207 self.resource_type_relation.into_values(),
208 ),
209 resource_type: histogram_from_posting_lengths(self.resource_type.into_values()),
210 subject: histogram_from_posting_lengths(self.subject.into_values()),
211 subject_type_relation: histogram_from_posting_lengths(
212 self.subject_type_relation.into_values(),
213 ),
214 subject_type: histogram_from_posting_lengths(self.subject_type.into_values()),
215 }
216 }
217}
218
219#[cfg(feature = "bench-internals")]
220fn increment_posting<K>(postings: &mut HashMap<K, usize>, key: K)
221where
222 K: Eq + Hash,
223{
224 let count = postings.entry(key).or_default();
225 *count = count.saturating_add(1);
226}
227
228#[cfg(feature = "bench-internals")]
229fn histogram_from_posting_lengths(
230 lengths: impl IntoIterator<Item = usize>,
231) -> StorePostingHistogram {
232 let mut histogram = StorePostingHistogram::default();
233 for len in lengths {
234 histogram.record_posting_len(len);
235 }
236 histogram
237}
238
239#[cfg(feature = "bench-internals")]
240fn u64_from_usize_saturating(value: usize) -> u64 {
241 u64::try_from(value).unwrap_or(u64::MAX)
242}
243
244#[cfg(feature = "bench-internals")]
246pub fn reset_store_view_read_counters() {
247 STORE_VIEW_QUERY_CALLS.store(0, Ordering::Relaxed);
248 DELTA_SEGMENTS_INSPECTED.store(0, Ordering::Relaxed);
249 TOMBSTONE_CHECKS.store(0, Ordering::Relaxed);
250}
251
252#[cfg(feature = "bench-internals")]
254#[must_use]
255pub fn store_view_read_counters() -> StoreViewReadCounters {
256 StoreViewReadCounters {
257 query_calls: STORE_VIEW_QUERY_CALLS.load(Ordering::Relaxed),
258 delta_segments_inspected: DELTA_SEGMENTS_INSPECTED.load(Ordering::Relaxed),
259 tombstone_checks: TOMBSTONE_CHECKS.load(Ordering::Relaxed),
260 }
261}
262
263#[cfg(feature = "bench-internals")]
264fn record_store_view_query_call() {
265 STORE_VIEW_QUERY_CALLS.fetch_add(1, Ordering::Relaxed);
266}
267
268#[cfg(not(feature = "bench-internals"))]
269fn record_store_view_query_call() {}
270
271#[cfg(feature = "bench-internals")]
272fn record_delta_segment_inspected() {
273 DELTA_SEGMENTS_INSPECTED.fetch_add(1, Ordering::Relaxed);
274}
275
276#[cfg(not(feature = "bench-internals"))]
277fn record_delta_segment_inspected() {}
278
279#[cfg(feature = "bench-internals")]
280fn record_tombstone_check() {
281 TOMBSTONE_CHECKS.fetch_add(1, Ordering::Relaxed);
282}
283
284#[cfg(not(feature = "bench-internals"))]
285fn record_tombstone_check() {}
286
287#[derive(Debug, Clone, PartialEq, Eq, Error)]
289pub enum StoreError {
290 #[error("relationship already exists: {relationship}")]
292 RelationshipAlreadyExists {
293 relationship: Box<Relationship>,
295 },
296
297 #[error("relationship not found: {relationship}")]
299 RelationshipNotFound {
300 relationship: Box<Relationship>,
302 },
303
304 #[error("duplicate mutation for relationship: {relationship}")]
306 DuplicateMutation {
307 relationship: Box<Relationship>,
309 },
310
311 #[error("precondition failed: {precondition:?}")]
313 PreconditionFailed {
314 precondition: Box<Precondition>,
316 },
317
318 #[error("mutation batch too large: {actual} exceeds limit {limit}")]
320 MutationBatchTooLarge {
321 limit: usize,
323 actual: usize,
325 },
326
327 #[error("precondition batch too large: {actual} exceeds limit {limit}")]
329 PreconditionBatchTooLarge {
330 limit: usize,
332 actual: usize,
334 },
335
336 #[error("compact store capacity exceeded for {component}")]
338 CapacityExceeded {
339 component: &'static str,
341 },
342
343 #[error("compact store internal invariant failed: {reason}")]
345 InternalInvariant {
346 reason: &'static str,
348 },
349
350 #[error("compact store domain materialization failed: {message}")]
352 DomainMaterialization {
353 message: String,
355 },
356}
357
358impl From<DomainError> for StoreError {
359 fn from(value: DomainError) -> Self {
360 Self::DomainMaterialization {
361 message: value.to_string(),
362 }
363 }
364}
365
366#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
368pub struct StoreCheckKey {
369 object_type: NonZeroU32,
370 object_id: NonZeroU32,
371 relation: NonZeroU32,
372 subject_type: NonZeroU32,
373 subject_id: NonZeroU32,
374 subject_relation: Option<NonZeroU32>,
375}
376
377#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
379#[derive(Debug, Clone, Copy, PartialEq, Eq)]
380pub struct QueryLimit(NonZeroUsize);
381
382impl QueryLimit {
383 #[must_use]
389 pub const fn new(limit: NonZeroUsize) -> Self {
390 Self(limit)
391 }
392
393 #[must_use]
395 pub fn default_limit() -> Self {
396 match NonZeroUsize::new(DEFAULT_QUERY_LIMIT) {
397 Some(limit) => Self(limit),
398 None => Self(NonZeroUsize::MIN),
399 }
400 }
401
402 fn get(self) -> usize {
403 self.0.get()
404 }
405}
406
407impl Default for QueryLimit {
408 fn default() -> Self {
409 Self::default_limit()
410 }
411}
412
413#[cfg_attr(
415 feature = "serde",
416 derive(serde::Serialize, serde::Deserialize),
417 serde(rename_all = "camelCase", deny_unknown_fields)
418)]
419#[derive(Debug, Clone, PartialEq, Eq)]
420pub struct SubjectFilter {
421 subject_type: SubjectType,
422 optional_subject_id: Option<SubjectId>,
423 optional_relation: Option<RelationName>,
424}
425
426impl SubjectFilter {
427 #[must_use]
429 pub fn new(
430 subject_type: SubjectType,
431 optional_subject_id: Option<SubjectId>,
432 optional_relation: Option<RelationName>,
433 ) -> Self {
434 Self {
435 subject_type,
436 optional_subject_id,
437 optional_relation,
438 }
439 }
440
441 #[must_use]
443 pub fn exact(
444 subject_type: SubjectType,
445 subject_id: SubjectId,
446 relation: Option<RelationName>,
447 ) -> Self {
448 Self::new(subject_type, Some(subject_id), relation)
449 }
450
451 #[must_use]
453 pub fn subject_type(&self) -> &SubjectType {
454 &self.subject_type
455 }
456
457 #[must_use]
459 pub fn optional_subject_id(&self) -> Option<&SubjectId> {
460 self.optional_subject_id.as_ref()
461 }
462
463 #[must_use]
465 pub fn optional_relation(&self) -> Option<&RelationName> {
466 self.optional_relation.as_ref()
467 }
468}
469
470impl TryFrom<&User> for SubjectFilter {
471 type Error = ZanzibarError;
472
473 fn try_from(value: &User) -> Result<Self, Self::Error> {
474 match value {
475 User::UserId(id) => Ok(Self::exact(
476 SubjectType::try_from("user")?,
477 SubjectId::try_from(id.as_str())?,
478 None,
479 )),
480 User::Userset(object, relation) => Ok(Self::exact(
481 SubjectType::try_from(object.namespace.as_str())?,
482 SubjectId::try_from(object.id.as_str())?,
483 Some(RelationName::try_from(relation.0.as_str())?),
484 )),
485 }
486 }
487}
488
489#[cfg_attr(
491 feature = "serde",
492 derive(serde::Serialize, serde::Deserialize),
493 serde(rename_all = "camelCase", deny_unknown_fields)
494)]
495#[derive(Debug, Clone, PartialEq, Eq)]
496pub struct RelationshipFilter {
497 resource_type: ObjectType,
498 optional_resource_id: Option<ObjectId>,
499 optional_relation: Option<RelationName>,
500 optional_subject: Option<SubjectFilter>,
501 limit: QueryLimit,
502}
503
504impl RelationshipFilter {
505 #[must_use]
507 pub fn new(
508 resource_type: ObjectType,
509 optional_resource_id: Option<ObjectId>,
510 optional_relation: Option<RelationName>,
511 optional_subject: Option<SubjectFilter>,
512 limit: QueryLimit,
513 ) -> Self {
514 Self {
515 resource_type,
516 optional_resource_id,
517 optional_relation,
518 optional_subject,
519 limit,
520 }
521 }
522
523 #[must_use]
525 pub fn for_exact_subject(
526 resource: &ObjectRef,
527 relation: RelationName,
528 subject: SubjectFilter,
529 ) -> Self {
530 Self::new(
531 resource.object_type().clone(),
532 Some(resource.object_id().clone()),
533 Some(relation),
534 Some(subject),
535 QueryLimit::default(),
536 )
537 }
538
539 #[must_use]
541 pub fn resource_type(&self) -> &ObjectType {
542 &self.resource_type
543 }
544
545 #[must_use]
547 pub fn optional_resource_id(&self) -> Option<&ObjectId> {
548 self.optional_resource_id.as_ref()
549 }
550
551 #[must_use]
553 pub fn optional_relation(&self) -> Option<&RelationName> {
554 self.optional_relation.as_ref()
555 }
556
557 #[must_use]
559 pub fn optional_subject(&self) -> Option<&SubjectFilter> {
560 self.optional_subject.as_ref()
561 }
562}
563
564#[cfg_attr(
566 feature = "serde",
567 derive(serde::Serialize, serde::Deserialize),
568 serde(rename_all = "camelCase")
569)]
570#[derive(Debug, Clone, PartialEq, Eq)]
571pub enum RelationshipMutation {
572 Create(Relationship),
574 Touch(Relationship),
576 Delete(Relationship),
578}
579
580impl RelationshipMutation {
581 pub fn create(relationship: impl AsRef<str>) -> Result<Self, DomainError> {
588 Ok(Self::Create(relationship.as_ref().parse()?))
589 }
590
591 pub fn touch(relationship: impl AsRef<str>) -> Result<Self, DomainError> {
598 Ok(Self::Touch(relationship.as_ref().parse()?))
599 }
600
601 pub fn delete(relationship: impl AsRef<str>) -> Result<Self, DomainError> {
608 Ok(Self::Delete(relationship.as_ref().parse()?))
609 }
610
611 #[must_use]
613 pub fn relationship(&self) -> &Relationship {
614 match self {
615 Self::Create(relationship) | Self::Touch(relationship) | Self::Delete(relationship) => {
616 relationship
617 }
618 }
619 }
620}
621
622#[cfg_attr(
624 feature = "serde",
625 derive(serde::Serialize, serde::Deserialize),
626 serde(rename_all = "camelCase")
627)]
628#[derive(Debug, Clone, PartialEq, Eq)]
629pub enum Precondition {
630 MustMatch(RelationshipFilter),
632 MustNotMatch(RelationshipFilter),
634}
635
636pub trait RelationshipReader {
638 type Iter<'a>: Iterator<Item = &'a Relationship>
640 where
641 Self: 'a;
642
643 fn query_relationships(
649 &self,
650 filter: &RelationshipFilter,
651 ) -> Result<Self::Iter<'_>, StoreError>;
652
653 fn reverse_query_relationships(
659 &self,
660 filter: &SubjectFilter,
661 ) -> Result<Self::Iter<'_>, StoreError>;
662}
663
664#[derive(Debug)]
666pub struct IndexedRelationshipStore {
667 index_profile: IndexProfile,
668 interner: IdentifierInterner,
669 rows: Vec<RelationshipRow>,
670 live_rows: LiveRows,
671 dead_row_count: usize,
672 uniqueness: UniquenessState,
673 by_resource: PostingIndex<ResourceIndexKey>,
674 by_resource_object: PostingIndex<ResourceObjectIndexKey>,
675 by_resource_type_relation: PostingIndex<ResourceTypeRelationIndexKey>,
676 by_resource_type: PostingIndex<ObjectTypeId>,
677 by_subject: PostingIndex<SubjectIndexKey>,
678 by_subject_type_relation: PostingIndex<SubjectTypeRelationIndexKey>,
679 by_subject_type: PostingIndex<SubjectTypeId>,
680}
681
682impl Default for IndexedRelationshipStore {
683 fn default() -> Self {
684 Self {
685 index_profile: IndexProfile::Full,
686 interner: IdentifierInterner::default(),
687 rows: Vec::new(),
688 live_rows: LiveRows::default(),
689 dead_row_count: 0,
690 uniqueness: UniquenessState::Ready(RelationshipIdentityIndex::default()),
691 by_resource: PostingIndex::default(),
692 by_resource_object: PostingIndex::default(),
693 by_resource_type_relation: PostingIndex::default(),
694 by_resource_type: PostingIndex::default(),
695 by_subject: PostingIndex::default(),
696 by_subject_type_relation: PostingIndex::default(),
697 by_subject_type: PostingIndex::default(),
698 }
699 }
700}
701
702impl Clone for IndexedRelationshipStore {
703 fn clone(&self) -> Self {
704 Self {
705 index_profile: self.index_profile,
706 interner: self.interner.clone(),
707 rows: self.rows.clone(),
708 live_rows: self.live_rows.clone(),
709 dead_row_count: self.dead_row_count,
710 uniqueness: self.uniqueness.clone(),
711 by_resource: self.by_resource.clone(),
712 by_resource_object: self.by_resource_object.clone(),
713 by_resource_type_relation: self.by_resource_type_relation.clone(),
714 by_resource_type: self.by_resource_type.clone(),
715 by_subject: self.by_subject.clone(),
716 by_subject_type_relation: self.by_subject_type_relation.clone(),
717 by_subject_type: self.by_subject_type.clone(),
718 }
719 }
720}
721
722#[derive(Debug, Clone)]
728pub struct RelationshipStoreView {
729 checkpoint: Arc<IndexedRelationshipStore>,
730 delta: Option<StoreDelta>,
731}
732
733impl Default for RelationshipStoreView {
734 fn default() -> Self {
735 Self::from_checkpoint(Arc::new(IndexedRelationshipStore::default()))
736 }
737}
738
739impl RelationshipStoreView {
740 #[must_use]
742 pub fn from_checkpoint(checkpoint: Arc<IndexedRelationshipStore>) -> Self {
743 Self {
744 checkpoint,
745 delta: None,
746 }
747 }
748
749 pub(crate) fn apply_mutations(
750 &self,
751 mutations: impl IntoIterator<Item = RelationshipMutation>,
752 preconditions: impl IntoIterator<Item = Precondition>,
753 ) -> Result<Arc<Self>, StoreError> {
754 let preconditions = preconditions.into_iter().collect::<Vec<_>>();
755 if preconditions.len() > MAX_PRECONDITIONS_PER_BATCH {
756 return Err(StoreError::PreconditionBatchTooLarge {
757 limit: MAX_PRECONDITIONS_PER_BATCH,
758 actual: preconditions.len(),
759 });
760 }
761 for precondition in &preconditions {
762 self.check_precondition(precondition)?;
763 }
764
765 let mutations = mutations.into_iter().collect::<Vec<_>>();
766 if mutations.len() > MAX_MUTATIONS_PER_BATCH {
767 return Err(StoreError::MutationBatchTooLarge {
768 limit: MAX_MUTATIONS_PER_BATCH,
769 actual: mutations.len(),
770 });
771 }
772 let mut seen = HashSet::with_capacity(mutations.len());
773 for mutation in &mutations {
774 let relationship = mutation.relationship();
775 if !seen.insert(relationship.clone()) {
776 return Err(StoreError::DuplicateMutation {
777 relationship: Box::new(relationship.clone()),
778 });
779 }
780 }
781 if mutations.is_empty() {
782 return Ok(Arc::new(self.clone()));
783 }
784
785 let mut base = self.clone();
786 base.ensure_checkpoint_mutation_ready()?;
787
788 let mut inserted = base
789 .delta
790 .as_ref()
791 .map_or_else(IndexedRelationshipStore::default, |delta| {
792 (*delta.inserted).clone()
793 });
794 let mut deleted = base
795 .delta
796 .as_ref()
797 .map_or_else(HashSet::new, |delta| (*delta.deleted).clone());
798
799 for mutation in mutations.iter().cloned() {
800 base.apply_delta_mutation(&mut inserted, &mut deleted, mutation)?;
801 }
802
803 let previous_mutations = base
804 .delta
805 .as_ref()
806 .map_or(0, |delta| delta.mutation_count.get());
807 let mutation_count = previous_mutations
808 .checked_add(mutations.len())
809 .and_then(NonZeroUsize::new)
810 .ok_or(StoreError::CapacityExceeded {
811 component: "store view delta mutations",
812 })?;
813 let candidate = Self {
814 checkpoint: Arc::clone(&base.checkpoint),
815 delta: Some(StoreDelta {
816 inserted: Arc::new(inserted),
817 deleted_rows: Arc::new(base.deleted_relationship_rows(&deleted)),
818 deleted: Arc::new(deleted),
819 mutation_count,
820 }),
821 };
822 if candidate.should_checkpoint() {
823 return Ok(Arc::new(candidate.checkpointed()?));
824 }
825 Ok(Arc::new(candidate))
826 }
827
828 #[must_use]
830 pub fn any_resource_match(&self, filter: &RelationshipFilter) -> bool {
831 self.query_compact_relationships(filter).next().is_some()
832 }
833
834 #[must_use]
836 pub fn rows(&self) -> Vec<Relationship> {
837 let Some(delta) = &self.delta else {
838 return self.checkpoint.rows();
839 };
840 let mut relationships = self
841 .checkpoint
842 .rows()
843 .into_iter()
844 .filter(|relationship| !delta.deleted.contains(relationship))
845 .collect::<Vec<_>>();
846 relationships.extend(delta.inserted.rows());
847 relationships
848 }
849
850 pub(crate) fn encode_snapshot_sections(
851 &self,
852 writer: &mut SnapshotSectionWriter,
853 index_profile: IndexProfile,
854 layout: SnapshotEncodingLayout,
855 ) -> Result<(), SnapshotIoError> {
856 self.canonical_store()?
857 .encode_snapshot_sections(writer, index_profile, layout)
858 }
859
860 pub(crate) fn query_compact_relationships(
861 &self,
862 filter: &RelationshipFilter,
863 ) -> StoreViewCompactIter<'_> {
864 record_store_view_query_call();
865 let inserted = self
866 .delta
867 .as_ref()
868 .map(|delta| delta.inserted.query_compact_relationships(filter));
869 if inserted.is_some() {
870 record_delta_segment_inspected();
871 }
872 StoreViewCompactIter {
873 inserted,
874 checkpoint: self.checkpoint.query_compact_relationships(filter),
875 deleted: self.delta.as_ref().map(|delta| delta.deleted_rows.as_ref()),
876 phase: StoreViewIterPhase::Inserted,
877 }
878 }
879
880 pub(crate) fn reverse_query_compact_relationships(
881 &self,
882 filter: &SubjectFilter,
883 ) -> StoreViewCompactIter<'_> {
884 record_store_view_query_call();
885 let inserted = self
886 .delta
887 .as_ref()
888 .map(|delta| delta.inserted.reverse_query_compact_relationships(filter));
889 if inserted.is_some() {
890 record_delta_segment_inspected();
891 }
892 StoreViewCompactIter {
893 inserted,
894 checkpoint: self.checkpoint.reverse_query_compact_relationships(filter),
895 deleted: self.delta.as_ref().map(|delta| delta.deleted_rows.as_ref()),
896 phase: StoreViewIterPhase::Inserted,
897 }
898 }
899
900 pub(crate) fn has_reverse_subject_candidates(&self, filter: &SubjectFilter) -> bool {
901 self.delta
902 .as_ref()
903 .is_some_and(|delta| delta.inserted.has_subject_candidates(filter))
904 || self.checkpoint.has_subject_candidates(filter)
905 }
906
907 pub(crate) fn resource_relation(
908 &self,
909 resource: &ObjectRef,
910 relation: &RelationName,
911 limit: QueryLimit,
912 ) -> StoreViewCompactIter<'_> {
913 record_store_view_query_call();
914 let inserted = self
915 .delta
916 .as_ref()
917 .map(|delta| delta.inserted.resource_relation(resource, relation, limit));
918 if inserted.is_some() {
919 record_delta_segment_inspected();
920 }
921 StoreViewCompactIter {
922 inserted,
923 checkpoint: self.checkpoint.resource_relation(resource, relation, limit),
924 deleted: self.delta.as_ref().map(|delta| delta.deleted_rows.as_ref()),
925 phase: StoreViewIterPhase::Inserted,
926 }
927 }
928
929 pub(crate) fn any_resource_relation_subject(
930 &self,
931 resource: &ObjectRef,
932 relation: &RelationName,
933 subject: &SubjectFilter,
934 ) -> bool {
935 self.resource_relation_subject(resource, relation, subject)
936 .next()
937 .is_some()
938 }
939
940 fn resource_relation_subject(
941 &self,
942 resource: &ObjectRef,
943 relation: &RelationName,
944 subject: &SubjectFilter,
945 ) -> StoreViewCompactIter<'_> {
946 record_store_view_query_call();
947 let inserted = self.delta.as_ref().map(|delta| {
948 delta
949 .inserted
950 .resource_relation_subject(resource, relation, subject)
951 });
952 if inserted.is_some() {
953 record_delta_segment_inspected();
954 }
955 StoreViewCompactIter {
956 inserted,
957 checkpoint: self
958 .checkpoint
959 .resource_relation_subject(resource, relation, subject),
960 deleted: self.delta.as_ref().map(|delta| delta.deleted_rows.as_ref()),
961 phase: StoreViewIterPhase::Inserted,
962 }
963 }
964
965 pub(crate) fn index_profile(&self) -> IndexProfile {
966 self.checkpoint.index_profile()
967 }
968
969 #[cfg(feature = "bench-internals")]
971 #[must_use]
972 pub fn delta_stats(&self) -> StoreViewDeltaStats {
973 let checkpoint_rows = self.checkpoint.live_row_count();
974 let Some(delta) = &self.delta else {
975 return StoreViewDeltaStats {
976 checkpoint_rows,
977 ..StoreViewDeltaStats::default()
978 };
979 };
980 let delta_inserted_rows = delta.inserted.live_row_count();
981 let delta_deleted_rows = delta.deleted_rows.len();
982 let denominator = checkpoint_rows.saturating_add(delta_inserted_rows).max(1);
983 let ratio = delta_deleted_rows.saturating_mul(10_000) / denominator;
984 StoreViewDeltaStats {
985 checkpoint_rows,
986 delta_segments: 1,
987 delta_inserted_rows,
988 delta_deleted_rows,
989 delta_mutations: delta.mutation_count.get(),
990 tombstone_ratio_bps: u16::try_from(ratio).unwrap_or(u16::MAX),
991 }
992 }
993
994 #[cfg(feature = "bench-internals")]
996 #[must_use]
997 pub fn posting_histograms(&self) -> StorePostingHistograms {
998 let mut builder = ActivePostingHistogramBuilder::default();
999 let deleted = self.delta.as_ref().map(|delta| delta.deleted_rows.as_ref());
1000 self.checkpoint
1001 .record_active_postings(&mut builder, deleted);
1002 if let Some(delta) = &self.delta {
1003 delta.inserted.record_active_postings(&mut builder, None);
1004 }
1005 builder.finish()
1006 }
1007
1008 pub(crate) fn store_check_key(
1009 &self,
1010 object: &Object,
1011 relation: &Relation,
1012 user: &User,
1013 ) -> Option<StoreCheckKey> {
1014 if self.delta.is_some() {
1015 return None;
1016 }
1017 self.checkpoint.store_check_key(object, relation, user)
1018 }
1019
1020 pub(crate) fn store_check_key_for_relation_name(
1021 &self,
1022 object: &Object,
1023 relation: &RelationName,
1024 user: &User,
1025 ) -> Option<StoreCheckKey> {
1026 if self.delta.is_some() {
1027 return None;
1028 }
1029 self.checkpoint
1030 .store_check_key_for_relation_name(object, relation, user)
1031 }
1032
1033 fn check_precondition(&self, precondition: &Precondition) -> Result<(), StoreError> {
1034 match precondition {
1035 Precondition::MustMatch(filter) if !self.any_resource_match(filter) => {
1036 Err(StoreError::PreconditionFailed {
1037 precondition: Box::new(precondition.clone()),
1038 })
1039 }
1040 Precondition::MustNotMatch(filter) if self.any_resource_match(filter) => {
1041 Err(StoreError::PreconditionFailed {
1042 precondition: Box::new(precondition.clone()),
1043 })
1044 }
1045 _ => Ok(()),
1046 }
1047 }
1048
1049 fn ensure_checkpoint_mutation_ready(&mut self) -> Result<(), StoreError> {
1050 if self.checkpoint.has_ready_uniqueness() {
1051 return Ok(());
1052 }
1053 let mut checkpoint = (*self.checkpoint).clone();
1054 checkpoint.ensure_uniqueness_index()?;
1055 self.checkpoint = Arc::new(checkpoint);
1056 Ok(())
1057 }
1058
1059 fn apply_delta_mutation(
1060 &self,
1061 inserted: &mut IndexedRelationshipStore,
1062 deleted: &mut HashSet<Relationship>,
1063 mutation: RelationshipMutation,
1064 ) -> Result<(), StoreError> {
1065 match mutation {
1066 RelationshipMutation::Create(relationship) => {
1067 self.create_delta_relationship(inserted, deleted, relationship)
1068 }
1069 RelationshipMutation::Touch(relationship) => {
1070 self.touch_delta_relationship(inserted, deleted, &relationship)
1071 }
1072 RelationshipMutation::Delete(relationship) => {
1073 self.delete_delta_relationship(inserted, deleted, relationship)
1074 }
1075 }
1076 }
1077
1078 fn create_delta_relationship(
1079 &self,
1080 inserted: &mut IndexedRelationshipStore,
1081 deleted: &mut HashSet<Relationship>,
1082 relationship: Relationship,
1083 ) -> Result<(), StoreError> {
1084 let location = self.relationship_location(inserted, deleted, &relationship);
1085 match location {
1086 RelationshipLocation::Inserted | RelationshipLocation::CheckpointLive => {
1087 Err(StoreError::RelationshipAlreadyExists {
1088 relationship: Box::new(relationship),
1089 })
1090 }
1091 RelationshipLocation::CheckpointDeleted => {
1092 deleted.remove(&relationship);
1093 Ok(())
1094 }
1095 RelationshipLocation::Absent => inserted.insert(&relationship),
1096 }
1097 }
1098
1099 fn touch_delta_relationship(
1100 &self,
1101 inserted: &mut IndexedRelationshipStore,
1102 deleted: &mut HashSet<Relationship>,
1103 relationship: &Relationship,
1104 ) -> Result<(), StoreError> {
1105 match self.relationship_location(inserted, deleted, relationship) {
1106 RelationshipLocation::Inserted | RelationshipLocation::CheckpointLive => Ok(()),
1107 RelationshipLocation::CheckpointDeleted => {
1108 deleted.remove(relationship);
1109 Ok(())
1110 }
1111 RelationshipLocation::Absent => inserted.insert(relationship),
1112 }
1113 }
1114
1115 fn delete_delta_relationship(
1116 &self,
1117 inserted: &mut IndexedRelationshipStore,
1118 deleted: &mut HashSet<Relationship>,
1119 relationship: Relationship,
1120 ) -> Result<(), StoreError> {
1121 match self.relationship_location(inserted, deleted, &relationship) {
1122 RelationshipLocation::Inserted => inserted.delete(&relationship),
1123 RelationshipLocation::CheckpointLive => {
1124 deleted.insert(relationship);
1125 Ok(())
1126 }
1127 RelationshipLocation::CheckpointDeleted | RelationshipLocation::Absent => {
1128 Err(StoreError::RelationshipNotFound {
1129 relationship: Box::new(relationship),
1130 })
1131 }
1132 }
1133 }
1134
1135 fn relationship_location(
1136 &self,
1137 inserted: &IndexedRelationshipStore,
1138 deleted: &HashSet<Relationship>,
1139 relationship: &Relationship,
1140 ) -> RelationshipLocation {
1141 if inserted.contains_relationship(relationship) {
1142 return RelationshipLocation::Inserted;
1143 }
1144 if self.checkpoint.contains_relationship(relationship) {
1145 if deleted.contains(relationship) {
1146 RelationshipLocation::CheckpointDeleted
1147 } else {
1148 RelationshipLocation::CheckpointLive
1149 }
1150 } else {
1151 RelationshipLocation::Absent
1152 }
1153 }
1154
1155 fn should_checkpoint(&self) -> bool {
1156 self.delta.as_ref().is_some_and(|delta| {
1157 delta.mutation_count.get() >= STORE_VIEW_MAX_DELTA_MUTATIONS
1158 || delta.deleted.len() >= STORE_VIEW_MAX_DELTA_TOMBSTONES
1159 })
1160 }
1161
1162 fn checkpointed(&self) -> Result<Self, StoreError> {
1163 Ok(Self::from_checkpoint(Arc::new(self.canonical_store()?)))
1164 }
1165
1166 fn canonical_store(&self) -> Result<IndexedRelationshipStore, StoreError> {
1167 let mut store = IndexedRelationshipStore::default();
1168 for relationship in self.rows() {
1169 store.insert(&relationship)?;
1170 }
1171 Ok(store)
1172 }
1173
1174 fn deleted_relationship_rows(
1175 &self,
1176 deleted: &HashSet<Relationship>,
1177 ) -> HashSet<RelationshipRow> {
1178 deleted
1179 .iter()
1180 .filter_map(|relationship| {
1181 RelationshipRow::from_existing_relationship(relationship, &self.checkpoint.interner)
1182 })
1183 .collect()
1184 }
1185}
1186
1187#[derive(Debug, Clone)]
1188struct StoreDelta {
1189 inserted: Arc<IndexedRelationshipStore>,
1190 deleted_rows: Arc<HashSet<RelationshipRow>>,
1191 deleted: Arc<HashSet<Relationship>>,
1192 mutation_count: NonZeroUsize,
1193}
1194
1195#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1196enum RelationshipLocation {
1197 Inserted,
1198 CheckpointLive,
1199 CheckpointDeleted,
1200 Absent,
1201}
1202
1203#[derive(Debug)]
1204pub(crate) struct StoreViewCompactIter<'a> {
1205 inserted: Option<CompactRelationshipIter<'a>>,
1206 checkpoint: CompactRelationshipIter<'a>,
1207 deleted: Option<&'a HashSet<RelationshipRow>>,
1208 phase: StoreViewIterPhase,
1209}
1210
1211impl<'a> Iterator for StoreViewCompactIter<'a> {
1212 type Item = RelationshipRef<'a>;
1213
1214 fn next(&mut self) -> Option<Self::Item> {
1215 loop {
1216 match self.phase {
1217 StoreViewIterPhase::Inserted => {
1218 if let Some(inserted) = &mut self.inserted
1219 && let Some(relationship) = inserted.next()
1220 {
1221 return Some(relationship);
1222 }
1223 self.phase = StoreViewIterPhase::Checkpoint;
1224 }
1225 StoreViewIterPhase::Checkpoint => {
1226 let relationship = self.checkpoint.next()?;
1227 match self.deleted {
1228 Some(deleted) => {
1229 record_tombstone_check();
1230 if !relationship.is_deleted_by(deleted) {
1231 return Some(relationship);
1232 }
1233 }
1234 None => return Some(relationship),
1235 }
1236 }
1237 }
1238 }
1239 }
1240}
1241
1242#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1243enum StoreViewIterPhase {
1244 Inserted,
1245 Checkpoint,
1246}
1247
1248impl IndexedRelationshipStore {
1249 pub fn apply_mutations(
1256 &mut self,
1257 mutations: impl IntoIterator<Item = RelationshipMutation>,
1258 preconditions: impl IntoIterator<Item = Precondition>,
1259 ) -> Result<(), StoreError> {
1260 let preconditions = preconditions.into_iter().collect::<Vec<_>>();
1261 if preconditions.len() > MAX_PRECONDITIONS_PER_BATCH {
1262 return Err(StoreError::PreconditionBatchTooLarge {
1263 limit: MAX_PRECONDITIONS_PER_BATCH,
1264 actual: preconditions.len(),
1265 });
1266 }
1267 for precondition in &preconditions {
1268 self.check_precondition(precondition)?;
1269 }
1270
1271 let mut mutations = mutations.into_iter().collect::<Vec<_>>();
1272 if mutations.len() > MAX_MUTATIONS_PER_BATCH {
1273 return Err(StoreError::MutationBatchTooLarge {
1274 limit: MAX_MUTATIONS_PER_BATCH,
1275 actual: mutations.len(),
1276 });
1277 }
1278 if !mutations.is_empty() {
1279 self.ensure_uniqueness_index()?;
1280 }
1281 let mut seen = HashSet::with_capacity(mutations.len());
1282 for mutation in &mutations {
1283 let relationship = mutation.relationship();
1284 if !seen.insert(relationship.clone()) {
1285 return Err(StoreError::DuplicateMutation {
1286 relationship: Box::new(relationship.clone()),
1287 });
1288 }
1289 }
1290
1291 if mutations.len() == 1
1292 && let Some(mutation) = mutations.pop()
1293 {
1294 self.apply_single_mutation(mutation)?;
1295 self.compact_if_needed()?;
1296 return Ok(());
1297 }
1298
1299 if mutations
1300 .iter()
1301 .all(|mutation| matches!(mutation, RelationshipMutation::Touch(_)))
1302 {
1303 self.apply_touch_batch_in_place(&mutations)?;
1304 self.compact_if_needed()?;
1305 return Ok(());
1306 }
1307
1308 let mut candidate = self.clone();
1309 for mutation in mutations {
1310 match mutation {
1311 RelationshipMutation::Create(relationship) => candidate.create(relationship)?,
1312 RelationshipMutation::Touch(relationship) => {
1313 if !candidate.contains_relationship(&relationship) {
1314 candidate.insert(&relationship)?;
1315 }
1316 }
1317 RelationshipMutation::Delete(relationship) => candidate.delete(&relationship)?,
1318 }
1319 }
1320
1321 candidate.compact_if_needed()?;
1322 *self = candidate;
1323 Ok(())
1324 }
1325
1326 fn apply_touch_batch_in_place(
1327 &mut self,
1328 mutations: &[RelationshipMutation],
1329 ) -> Result<(), StoreError> {
1330 let new_relationships = mutations
1331 .iter()
1332 .filter(|mutation| !self.contains_relationship(mutation.relationship()))
1333 .collect::<Vec<_>>();
1334 let mut additional_symbols = 0_usize;
1335 let mut additional_bytes = 0_usize;
1336 for mutation in &new_relationships {
1337 relationship_identifier_values(mutation.relationship(), |value| {
1338 if self.interner.lookup(value).is_none() {
1339 additional_symbols = additional_symbols.saturating_add(1);
1340 additional_bytes = additional_bytes.saturating_add(value.len());
1341 }
1342 });
1343 }
1344 self.ensure_insert_capacity(
1345 new_relationships.len(),
1346 additional_symbols,
1347 additional_bytes,
1348 )?;
1349 for mutation in mutations {
1350 let relationship = mutation.relationship();
1351 if !self.contains_relationship(relationship) {
1352 self.insert(relationship)?;
1353 }
1354 }
1355 Ok(())
1356 }
1357
1358 fn ensure_insert_capacity(
1359 &self,
1360 additional_rows: usize,
1361 additional_symbols: usize,
1362 additional_bytes: usize,
1363 ) -> Result<(), StoreError> {
1364 if additional_rows == 0 {
1365 return Ok(());
1366 }
1367 let last_row_len = self
1368 .rows
1369 .len()
1370 .checked_add(additional_rows)
1371 .and_then(|value| value.checked_sub(1))
1372 .ok_or(StoreError::CapacityExceeded {
1373 component: "relationship rows",
1374 })?;
1375 RowId::from_len(last_row_len)?;
1376
1377 if additional_symbols > 0 {
1378 let last_symbol_index = self
1379 .interner
1380 .len()
1381 .checked_add(additional_symbols)
1382 .and_then(|value| value.checked_sub(1))
1383 .ok_or(StoreError::CapacityExceeded {
1384 component: "identifier interner",
1385 })?;
1386 SymbolId::from_index(last_symbol_index)?;
1387 }
1388 self.interner
1389 .byte_len()
1390 .checked_add(additional_bytes)
1391 .and_then(|value| u32::try_from(value).ok())
1392 .ok_or(StoreError::CapacityExceeded {
1393 component: "identifier interner bytes",
1394 })?;
1395 Ok(())
1396 }
1397
1398 fn ensure_uniqueness_index(&mut self) -> Result<(), StoreError> {
1399 let reason = match self.uniqueness {
1400 UniquenessState::Ready(_) => return Ok(()),
1401 UniquenessState::KnownUniqueButNotIndexed => {
1402 "full snapshot duplicate detector was dropped after validation"
1403 }
1404 UniquenessState::UntrustedNotIndexed => {
1405 "trusted snapshot contains duplicate relationship rows"
1406 }
1407 };
1408 let mut uniqueness = RelationshipIdentityIndex::default();
1409 for row in self
1410 .rows
1411 .iter()
1412 .filter(|row| self.live_rows.contains(row.row_id))
1413 {
1414 if uniqueness.find(&self.rows, row).is_some() {
1415 return Err(StoreError::InternalInvariant { reason });
1416 }
1417 uniqueness.insert(&self.rows, row.row_id, row);
1418 }
1419 self.uniqueness = UniquenessState::Ready(uniqueness);
1420 Ok(())
1421 }
1422
1423 #[must_use]
1425 pub fn any_resource_match(&self, filter: &RelationshipFilter) -> bool {
1426 self.query_compact_relationships(filter).next().is_some()
1427 }
1428
1429 #[must_use]
1431 pub fn rows(&self) -> Vec<Relationship> {
1432 self.rows
1433 .iter()
1434 .filter(|row| self.live_rows.contains(row.row_id))
1435 .filter_map(|row| self.relationship_from_row(row).ok())
1436 .collect()
1437 }
1438
1439 #[cfg(feature = "bench-internals")]
1440 fn live_row_count(&self) -> usize {
1441 self.rows.len().saturating_sub(self.dead_row_count)
1442 }
1443
1444 #[cfg(feature = "bench-internals")]
1445 fn record_active_postings(
1446 &self,
1447 builder: &mut ActivePostingHistogramBuilder,
1448 deleted: Option<&HashSet<RelationshipRow>>,
1449 ) {
1450 for row in self
1451 .rows
1452 .iter()
1453 .filter(|row| self.live_rows.contains(row.row_id))
1454 {
1455 if deleted.is_some_and(|deleted| deleted.contains(row)) {
1456 continue;
1457 }
1458 builder.record_row(row);
1459 }
1460 }
1461
1462 pub(crate) fn encode_snapshot_sections(
1463 &self,
1464 writer: &mut SnapshotSectionWriter,
1465 index_profile: IndexProfile,
1466 layout: SnapshotEncodingLayout,
1467 ) -> Result<(), SnapshotIoError> {
1468 let (symbol_table, symbol_table_flags) = encode_v3_symbol_table(
1469 &self.interner.entries,
1470 checked_u32_from_usize(self.interner.bytes.len())?,
1471 layout,
1472 )?;
1473 writer.add_section(
1474 SectionKind::SymbolBytes,
1475 self.interner.bytes.clone(),
1476 u64::try_from(self.interner.bytes.len()).map_err(|_| {
1477 SnapshotIoError::LimitExceeded {
1478 component: "symbol bytes",
1479 }
1480 })?,
1481 )?;
1482 writer.add_section_with_flags(
1483 SectionKind::SymbolTable,
1484 symbol_table_flags,
1485 symbol_table,
1486 u64::try_from(self.interner.entries.len()).map_err(|_| {
1487 SnapshotIoError::LimitExceeded {
1488 component: "symbol table",
1489 }
1490 })?,
1491 )?;
1492 let (symbol_hashes, symbol_lookup, symbol_lookup_flags) =
1493 self.interner.encode_symbol_acceleration(layout)?;
1494 let symbol_count = u64::try_from(self.interner.entries.len()).map_err(|_| {
1495 SnapshotIoError::LimitExceeded {
1496 component: "symbol lookup",
1497 }
1498 })?;
1499 writer.add_section(SectionKind::SymbolHashes, symbol_hashes, symbol_count)?;
1500 writer.add_section_with_flags(
1501 SectionKind::SymbolLookup,
1502 symbol_lookup_flags,
1503 symbol_lookup,
1504 symbol_count,
1505 )?;
1506
1507 let disk_rows = self.live_disk_rows();
1508 let row_symbol_width =
1509 snapshot_symbol_width(checked_u32_from_usize(self.interner.entries.len())?, layout);
1510 let mut rows = Vec::with_capacity(
1511 disk_rows
1512 .len()
1513 .checked_mul(checked_mul_usize(row_symbol_width.byte_len(), 6)?)
1514 .ok_or(SnapshotIoError::Format {
1515 reason: "relationship rows length overflowed",
1516 })?,
1517 );
1518 for row in &disk_rows {
1519 row.encode_width(row_symbol_width, &mut rows);
1520 }
1521 writer.add_section_with_flags(
1522 SectionKind::RelationshipRows,
1523 row_symbol_width.flag_bits(),
1524 rows,
1525 u64::try_from(disk_rows.len()).map_err(|_| SnapshotIoError::LimitExceeded {
1526 component: "relationship rows",
1527 })?,
1528 )?;
1529
1530 let indexes = EncodedSnapshotIndexes::from_rows(&disk_rows, index_profile)?;
1531 writer.add_section(
1532 SectionKind::IndexDirectory,
1533 indexes.directory,
1534 SNAPSHOT_INDEX_KIND_COUNT_U64,
1535 )?;
1536 writer.add_section(SectionKind::IndexKeys, indexes.keys, indexes.key_count)?;
1537 writer.add_section(
1538 SectionKind::PostingRanges,
1539 indexes.ranges,
1540 indexes.range_count,
1541 )?;
1542 writer.add_section(
1543 SectionKind::PostingRowIds,
1544 indexes.posting_row_ids,
1545 indexes.posting_row_id_count,
1546 )?;
1547 Ok(())
1548 }
1549
1550 pub(crate) fn decode_snapshot_sections(
1551 reader: &SnapshotReader<'_>,
1552 profile: SnapshotLoadProfile,
1553 validation: SnapshotValidationMode,
1554 ) -> Result<Self, SnapshotIoError> {
1555 Self::decode_snapshot_sections_inner(reader, profile, validation, None)
1556 }
1557
1558 #[cfg(feature = "bench-internals")]
1559 pub(crate) fn decode_snapshot_sections_with_timings(
1560 reader: &SnapshotReader<'_>,
1561 profile: SnapshotLoadProfile,
1562 validation: SnapshotValidationMode,
1563 timings: &mut SnapshotLoadPhaseTimings,
1564 ) -> Result<Self, SnapshotIoError> {
1565 Self::decode_snapshot_sections_inner(reader, profile, validation, Some(timings))
1566 }
1567
1568 fn decode_snapshot_sections_inner(
1569 reader: &SnapshotReader<'_>,
1570 profile: SnapshotLoadProfile,
1571 validation: SnapshotValidationMode,
1572 mut timings: Option<&mut SnapshotLoadPhaseTimings>,
1573 ) -> Result<Self, SnapshotIoError> {
1574 let phase_start = Instant::now();
1575 let interner = IdentifierInterner::decode_snapshot_sections(reader, validation)?;
1576 record_relationship_decode_phase(
1577 &mut timings,
1578 |timings, elapsed| {
1579 timings.symbols = elapsed;
1580 },
1581 phase_start,
1582 );
1583 let phase_start = Instant::now();
1584 let decoded_rows = decode_snapshot_rows(reader, &interner, validation)?;
1585 record_relationship_decode_phase(
1586 &mut timings,
1587 |timings, elapsed| {
1588 timings.rows = elapsed;
1589 },
1590 phase_start,
1591 );
1592 let phase_start = Instant::now();
1593 let decoded_indexes =
1594 DecodedSnapshotIndexes::decode(reader, &decoded_rows.rows, profile, validation)?;
1595 record_relationship_decode_phase(
1596 &mut timings,
1597 |timings, elapsed| {
1598 timings.indexes = elapsed;
1599 },
1600 phase_start,
1601 );
1602 Ok(Self {
1603 interner,
1604 index_profile: reader.header().index_profile,
1605 rows: decoded_rows.rows,
1606 live_rows: decoded_rows.live_rows,
1607 dead_row_count: 0,
1608 uniqueness: decoded_rows.uniqueness,
1609 by_resource: decoded_indexes.resource,
1610 by_resource_object: decoded_indexes.resource_object,
1611 by_resource_type_relation: decoded_indexes.resource_type_relation,
1612 by_resource_type: decoded_indexes.resource_type,
1613 by_subject: decoded_indexes.subject,
1614 by_subject_type_relation: decoded_indexes.subject_type_relation,
1615 by_subject_type: decoded_indexes.subject_type,
1616 })
1617 }
1618
1619 fn live_disk_rows(&self) -> Vec<DiskRelationshipRow> {
1620 let mut rows = Vec::with_capacity(self.rows.len().saturating_sub(self.dead_row_count));
1621 for row in self
1622 .rows
1623 .iter()
1624 .filter(|row| self.live_rows.contains(row.row_id))
1625 {
1626 rows.push(DiskRelationshipRow::from(row));
1627 }
1628 rows
1629 }
1630
1631 pub(crate) fn query_compact_relationships(
1632 &self,
1633 filter: &RelationshipFilter,
1634 ) -> CompactRelationshipIter<'_> {
1635 let matcher = self.resource_matcher(filter);
1636 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
1637 self.resource_candidate_row_ids(matcher)
1638 });
1639 CompactRelationshipIter {
1640 store: self,
1641 all_rows_live: self.live_rows.is_all_live(),
1642 candidates,
1643 matcher: matcher.map(CompactRelationshipMatcher::Resource),
1644 }
1645 }
1646
1647 pub(crate) fn reverse_query_compact_relationships(
1648 &self,
1649 filter: &SubjectFilter,
1650 ) -> CompactRelationshipIter<'_> {
1651 let matcher = self.subject_matcher(filter);
1652 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
1653 self.subject_candidate_row_ids(matcher)
1654 });
1655 CompactRelationshipIter {
1656 store: self,
1657 all_rows_live: self.live_rows.is_all_live(),
1658 candidates,
1659 matcher: matcher.map(CompactRelationshipMatcher::Subject),
1660 }
1661 }
1662
1663 fn has_subject_candidates(&self, filter: &SubjectFilter) -> bool {
1664 let Some(matcher) = self.subject_matcher(filter) else {
1665 return false;
1666 };
1667 self.subject_candidate_row_ids(&matcher).next().is_some()
1668 }
1669
1670 pub(crate) fn resource_relation(
1671 &self,
1672 resource: &ObjectRef,
1673 relation: &RelationName,
1674 limit: QueryLimit,
1675 ) -> CompactRelationshipIter<'_> {
1676 let matcher = self.resource_relation_matcher(resource, relation, limit);
1677 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
1678 self.resource_candidate_row_ids(matcher)
1679 });
1680 CompactRelationshipIter {
1681 store: self,
1682 all_rows_live: self.live_rows.is_all_live(),
1683 candidates,
1684 matcher: matcher.map(CompactRelationshipMatcher::Resource),
1685 }
1686 }
1687
1688 fn resource_relation_subject(
1689 &self,
1690 resource: &ObjectRef,
1691 relation: &RelationName,
1692 subject: &SubjectFilter,
1693 ) -> CompactRelationshipIter<'_> {
1694 let matcher = self.resource_relation_subject_matcher(resource, relation, subject);
1695 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
1696 self.resource_candidate_row_ids(matcher)
1697 });
1698 CompactRelationshipIter {
1699 store: self,
1700 all_rows_live: self.live_rows.is_all_live(),
1701 candidates,
1702 matcher: matcher.map(CompactRelationshipMatcher::Resource),
1703 }
1704 }
1705
1706 pub(crate) const fn index_profile(&self) -> IndexProfile {
1707 self.index_profile
1708 }
1709
1710 pub(crate) fn store_check_key(
1711 &self,
1712 object: &Object,
1713 relation: &Relation,
1714 user: &User,
1715 ) -> Option<StoreCheckKey> {
1716 self.store_check_key_for_relation(
1717 object,
1718 self.interner.lookup(relation.0.as_str()).map(RelationId)?,
1719 user,
1720 )
1721 }
1722
1723 pub(crate) fn store_check_key_for_relation_name(
1724 &self,
1725 object: &Object,
1726 relation: &RelationName,
1727 user: &User,
1728 ) -> Option<StoreCheckKey> {
1729 self.store_check_key_for_relation(
1730 object,
1731 self.interner.lookup(relation.as_str()).map(RelationId)?,
1732 user,
1733 )
1734 }
1735
1736 fn store_check_key_for_relation(
1737 &self,
1738 object: &Object,
1739 relation: RelationId,
1740 user: &User,
1741 ) -> Option<StoreCheckKey> {
1742 let subject = self.store_subject_key(user)?;
1743 Some(StoreCheckKey {
1744 object_type: self.interner.lookup(object.namespace.as_str())?.0,
1745 object_id: self.interner.lookup(object.id.as_str())?.0,
1746 relation: relation.0.0,
1747 subject_type: subject.subject_type.0.0,
1748 subject_id: subject.subject_id.0.0,
1749 subject_relation: subject.relation.map(|relation| relation.0.0),
1750 })
1751 }
1752
1753 fn store_subject_key(&self, user: &User) -> Option<SubjectIndexKey> {
1754 match user {
1755 User::UserId(id) => Some(SubjectIndexKey {
1756 subject_type: SubjectTypeId(self.interner.lookup("user")?),
1757 subject_id: SubjectIdId(self.interner.lookup(id.as_str())?),
1758 relation: None,
1759 }),
1760 User::Userset(object, relation) => Some(SubjectIndexKey {
1761 subject_type: SubjectTypeId(self.interner.lookup(object.namespace.as_str())?),
1762 subject_id: SubjectIdId(self.interner.lookup(object.id.as_str())?),
1763 relation: Some(RelationId(self.interner.lookup(relation.0.as_str())?)),
1764 }),
1765 }
1766 }
1767
1768 fn check_precondition(&self, precondition: &Precondition) -> Result<(), StoreError> {
1769 match precondition {
1770 Precondition::MustMatch(filter) if !self.any_resource_match(filter) => {
1771 Err(StoreError::PreconditionFailed {
1772 precondition: Box::new(precondition.clone()),
1773 })
1774 }
1775 Precondition::MustNotMatch(filter) if self.any_resource_match(filter) => {
1776 Err(StoreError::PreconditionFailed {
1777 precondition: Box::new(precondition.clone()),
1778 })
1779 }
1780 _ => Ok(()),
1781 }
1782 }
1783
1784 fn create(&mut self, relationship: Relationship) -> Result<(), StoreError> {
1785 if self.contains_relationship(&relationship) {
1786 return Err(StoreError::RelationshipAlreadyExists {
1787 relationship: Box::new(relationship),
1788 });
1789 }
1790 self.insert(&relationship)?;
1791 Ok(())
1792 }
1793
1794 fn apply_single_mutation(&mut self, mutation: RelationshipMutation) -> Result<(), StoreError> {
1795 match mutation {
1796 RelationshipMutation::Create(relationship) => self.create(relationship),
1797 RelationshipMutation::Touch(relationship) => {
1798 if !self.contains_relationship(&relationship) {
1799 self.insert(&relationship)?;
1800 }
1801 Ok(())
1802 }
1803 RelationshipMutation::Delete(relationship) => self.delete(&relationship),
1804 }
1805 }
1806
1807 fn insert(&mut self, relationship: &Relationship) -> Result<(), StoreError> {
1808 let row_id = RowId::from_len(self.rows.len())?;
1809 let row = RelationshipRow::from_relationship(row_id, relationship, &mut self.interner)?;
1810 let uniqueness = Self::ready_uniqueness_mut(&mut self.uniqueness)?;
1811 uniqueness.insert(&self.rows, row_id, &row);
1812 self.index_relationship(row_id, &row);
1813 self.rows.push(row);
1814 self.live_rows.insert(row_id);
1815 Ok(())
1816 }
1817
1818 fn delete(&mut self, relationship: &Relationship) -> Result<(), StoreError> {
1819 let row = self.lookup_relationship_row(relationship).ok_or_else(|| {
1820 StoreError::RelationshipNotFound {
1821 relationship: Box::new(relationship.clone()),
1822 }
1823 })?;
1824 let uniqueness = Self::ready_uniqueness_mut(&mut self.uniqueness)?;
1825 let row_id = uniqueness.remove(&self.rows, &row).ok_or_else(|| {
1826 StoreError::RelationshipNotFound {
1827 relationship: Box::new(relationship.clone()),
1828 }
1829 })?;
1830 self.live_rows.remove(row_id);
1831 self.dead_row_count = self.dead_row_count.saturating_add(1);
1832
1833 Ok(())
1834 }
1835
1836 pub(crate) fn contains_relationship(&self, relationship: &Relationship) -> bool {
1837 self.lookup_relationship_row(relationship)
1838 .is_some_and(|row| {
1839 self.uniqueness_ref()
1840 .is_some_and(|index| index.find(&self.rows, &row).is_some())
1841 })
1842 }
1843
1844 fn lookup_relationship_row(&self, relationship: &Relationship) -> Option<RelationshipRow> {
1845 RelationshipRow::from_existing_relationship(relationship, &self.interner)
1846 }
1847
1848 fn relationship_from_row(&self, row: &RelationshipRow) -> Result<Relationship, StoreError> {
1849 let resource = ObjectRef::new(
1850 ObjectType::try_from(self.interner.resolve(row.resource_type.0)?)?,
1851 ObjectId::try_from(self.interner.resolve(row.resource_id.0)?)?,
1852 );
1853 let relation = RelationName::try_from(self.interner.resolve(row.relation.0)?)?;
1854 let subject_object = ObjectRef::new(
1855 ObjectType::try_from(self.interner.resolve(row.subject_type.0)?)?,
1856 ObjectId::try_from(self.interner.resolve(row.subject_id.0)?)?,
1857 );
1858 let subject = match row.subject_relation {
1859 Some(relation) => SubjectRef::Userset {
1860 object: subject_object,
1861 relation: RelationName::try_from(self.interner.resolve(relation.0)?)?,
1862 },
1863 None => SubjectRef::Object(subject_object),
1864 };
1865 Ok(Relationship::new(resource, relation, subject))
1866 }
1867
1868 pub fn materialized_reader(&self) -> Result<MaterializedRelationshipReader<'_>, StoreError> {
1874 let mut rows = Vec::with_capacity(self.rows.len());
1875 for row in &self.rows {
1876 rows.push(self.relationship_from_row(row)?);
1877 }
1878 Ok(MaterializedRelationshipReader { store: self, rows })
1879 }
1880
1881 fn resource_matcher(&self, filter: &RelationshipFilter) -> Option<ResourceMatcher> {
1882 let resource_type = self.interner.lookup(filter.resource_type.as_str())?;
1883 let optional_resource_id = match &filter.optional_resource_id {
1884 Some(resource_id) => match self.interner.lookup(resource_id.as_str()) {
1885 Some(id) => Some(ObjectIdId(id)),
1886 None => return None,
1887 },
1888 None => None,
1889 };
1890 let optional_relation = match &filter.optional_relation {
1891 Some(relation) => match self.interner.lookup(relation.as_str()) {
1892 Some(id) => Some(RelationId(id)),
1893 None => return None,
1894 },
1895 None => None,
1896 };
1897 let optional_subject = match &filter.optional_subject {
1898 Some(subject) => match self.subject_matcher(subject) {
1899 Some(subject) => Some(subject),
1900 None => return None,
1901 },
1902 None => None,
1903 };
1904 Some(ResourceMatcher {
1905 resource_type: ObjectTypeId(resource_type),
1906 optional_resource_id,
1907 optional_relation,
1908 optional_subject,
1909 remaining: filter.limit.get(),
1910 })
1911 }
1912
1913 fn resource_relation_matcher(
1914 &self,
1915 resource: &ObjectRef,
1916 relation: &RelationName,
1917 limit: QueryLimit,
1918 ) -> Option<ResourceMatcher> {
1919 Some(ResourceMatcher {
1920 resource_type: ObjectTypeId(self.interner.lookup(resource.object_type().as_str())?),
1921 optional_resource_id: Some(ObjectIdId(
1922 self.interner.lookup(resource.object_id().as_str())?,
1923 )),
1924 optional_relation: Some(RelationId(self.interner.lookup(relation.as_str())?)),
1925 optional_subject: None,
1926 remaining: limit.get(),
1927 })
1928 }
1929
1930 fn resource_relation_subject_matcher(
1931 &self,
1932 resource: &ObjectRef,
1933 relation: &RelationName,
1934 subject: &SubjectFilter,
1935 ) -> Option<ResourceMatcher> {
1936 let mut matcher =
1937 self.resource_relation_matcher(resource, relation, QueryLimit::new(NonZeroUsize::MIN))?;
1938 matcher.optional_subject = Some(self.subject_matcher(subject)?);
1939 Some(matcher)
1940 }
1941
1942 fn subject_matcher(&self, filter: &SubjectFilter) -> Option<SubjectMatcher> {
1943 let subject_type = self.interner.lookup(filter.subject_type.as_str())?;
1944 let optional_subject_id = match &filter.optional_subject_id {
1945 Some(subject_id) => match self.interner.lookup(subject_id.as_str()) {
1946 Some(id) => Some(SubjectIdId(id)),
1947 None => return None,
1948 },
1949 None => None,
1950 };
1951 let optional_relation = match &filter.optional_relation {
1952 Some(relation) => match self.interner.lookup(relation.as_str()) {
1953 Some(id) => Some(RelationId(id)),
1954 None => return None,
1955 },
1956 None => None,
1957 };
1958 Some(SubjectMatcher {
1959 subject_type: SubjectTypeId(subject_type),
1960 optional_subject_id,
1961 optional_relation,
1962 })
1963 }
1964
1965 fn resource_candidate_row_ids(&self, matcher: &ResourceMatcher) -> CandidateRowIds<'_> {
1966 match (matcher.optional_resource_id, matcher.optional_relation) {
1967 (Some(resource_id), Some(relation)) => {
1968 let key = ResourceIndexKey {
1969 object_type: matcher.resource_type,
1970 object_id: resource_id,
1971 relation,
1972 };
1973 self.by_resource.candidates(&key)
1974 }
1975 (Some(resource_id), None) => {
1976 let key = ResourceObjectIndexKey {
1977 object_type: matcher.resource_type,
1978 object_id: resource_id,
1979 };
1980 self.by_resource_object.candidates(&key)
1981 }
1982 (None, Some(relation)) => {
1983 let key = ResourceTypeRelationIndexKey {
1984 object_type: matcher.resource_type,
1985 relation,
1986 };
1987 self.by_resource_type_relation.candidates(&key)
1988 }
1989 (None, None) => self.by_resource_type.candidates(&matcher.resource_type),
1990 }
1991 }
1992
1993 fn subject_candidate_row_ids(&self, matcher: &SubjectMatcher) -> CandidateRowIds<'_> {
1994 match (matcher.optional_subject_id, matcher.optional_relation) {
1995 (Some(subject_id), relation) => {
1996 let key = SubjectIndexKey {
1997 subject_type: matcher.subject_type,
1998 subject_id,
1999 relation,
2000 };
2001 self.by_subject.candidates(&key)
2002 }
2003 (None, Some(relation)) => {
2004 let key = SubjectTypeRelationIndexKey {
2005 subject_type: matcher.subject_type,
2006 relation,
2007 };
2008 self.by_subject_type_relation.candidates(&key)
2009 }
2010 (None, None) => self.by_subject_type.candidates(&matcher.subject_type),
2011 }
2012 }
2013
2014 fn index_relationship(&mut self, row_id: RowId, row: &RelationshipRow) {
2015 self.by_resource.insert(ResourceIndexKey::from(row), row_id);
2016 self.by_resource_object
2017 .insert(ResourceObjectIndexKey::from(row), row_id);
2018 self.by_resource_type_relation
2019 .insert(ResourceTypeRelationIndexKey::from(row), row_id);
2020 self.by_resource_type.insert(row.resource_type, row_id);
2021 for key in SubjectIndexKey::from_row(row) {
2022 self.by_subject.insert(key, row_id);
2023 }
2024 if let Some(key) = SubjectTypeRelationIndexKey::from_row(row) {
2025 self.by_subject_type_relation.insert(key, row_id);
2026 }
2027 self.by_subject_type.insert(row.subject_type, row_id);
2028 }
2029
2030 fn compact_if_needed(&mut self) -> Result<(), StoreError> {
2031 let dead_row_threshold = compaction_dead_row_threshold();
2032 if self.dead_row_count <= dead_row_threshold
2033 && self.dead_row_count.saturating_mul(4) <= self.rows.len()
2034 {
2035 return Ok(());
2036 }
2037
2038 let mut compacted = Self::default();
2039 for row in self
2040 .rows
2041 .iter()
2042 .filter(|row| self.live_rows.contains(row.row_id))
2043 {
2044 compacted.insert_compact_row_from(self, row)?;
2045 }
2046 *self = compacted;
2047 Ok(())
2048 }
2049
2050 fn insert_compact_row_from(
2051 &mut self,
2052 source: &Self,
2053 row: &RelationshipRow,
2054 ) -> Result<(), StoreError> {
2055 let row_id = RowId::from_len(self.rows.len())?;
2056 let compacted = RelationshipRow {
2057 row_id,
2058 resource_type: ObjectTypeId(self.interner.intern(source.resolve(row.resource_type.0))?),
2059 resource_id: ObjectIdId(self.interner.intern(source.resolve(row.resource_id.0))?),
2060 relation: RelationId(self.interner.intern(source.resolve(row.relation.0))?),
2061 subject_type: SubjectTypeId(self.interner.intern(source.resolve(row.subject_type.0))?),
2062 subject_id: SubjectIdId(self.interner.intern(source.resolve(row.subject_id.0))?),
2063 subject_relation: row
2064 .subject_relation
2065 .map(|relation| self.interner.intern(source.resolve(relation.0)))
2066 .transpose()?
2067 .map(RelationId),
2068 };
2069 let uniqueness = Self::ready_uniqueness_mut(&mut self.uniqueness)?;
2070 uniqueness.insert(&self.rows, row_id, &compacted);
2071 self.index_relationship(row_id, &compacted);
2072 self.rows.push(compacted);
2073 self.live_rows.insert(row_id);
2074 Ok(())
2075 }
2076
2077 fn resolve(&self, id: SymbolId) -> &str {
2078 self.interner.resolve(id).unwrap_or("<invalid>")
2079 }
2080
2081 fn uniqueness_ref(&self) -> Option<&RelationshipIdentityIndex> {
2082 match &self.uniqueness {
2083 UniquenessState::Ready(index) => Some(index),
2084 UniquenessState::KnownUniqueButNotIndexed | UniquenessState::UntrustedNotIndexed => {
2085 None
2086 }
2087 }
2088 }
2089
2090 fn has_ready_uniqueness(&self) -> bool {
2091 matches!(self.uniqueness, UniquenessState::Ready(_))
2092 }
2093
2094 fn ready_uniqueness_mut(
2095 uniqueness: &mut UniquenessState,
2096 ) -> Result<&mut RelationshipIdentityIndex, StoreError> {
2097 match uniqueness {
2098 UniquenessState::Ready(index) => Ok(index),
2099 UniquenessState::KnownUniqueButNotIndexed | UniquenessState::UntrustedNotIndexed => {
2100 Err(StoreError::InternalInvariant {
2101 reason: "relationship uniqueness index was not initialized before mutation",
2102 })
2103 }
2104 }
2105 }
2106}
2107
2108#[derive(Debug)]
2110pub struct MaterializedRelationshipReader<'a> {
2111 store: &'a IndexedRelationshipStore,
2112 rows: Vec<Relationship>,
2113}
2114
2115impl RelationshipReader for MaterializedRelationshipReader<'_> {
2116 type Iter<'a>
2117 = RelationshipIter<'a>
2118 where
2119 Self: 'a;
2120
2121 fn query_relationships(
2122 &self,
2123 filter: &RelationshipFilter,
2124 ) -> Result<Self::Iter<'_>, StoreError> {
2125 let matcher = self.store.resource_matcher(filter);
2126 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
2127 self.store.resource_candidate_row_ids(matcher)
2128 });
2129 Ok(RelationshipIter {
2130 rows: &self.rows,
2131 compact_rows: &self.store.rows,
2132 live_rows: &self.store.live_rows,
2133 all_rows_live: self.store.live_rows.is_all_live(),
2134 candidates,
2135 matcher: matcher.map(RelationshipMatcher::Resource),
2136 })
2137 }
2138
2139 fn reverse_query_relationships(
2140 &self,
2141 filter: &SubjectFilter,
2142 ) -> Result<Self::Iter<'_>, StoreError> {
2143 let matcher = self.store.subject_matcher(filter);
2144 let candidates = matcher.as_ref().map_or(CandidateRowIds::Empty, |matcher| {
2145 self.store.subject_candidate_row_ids(matcher)
2146 });
2147 Ok(RelationshipIter {
2148 rows: &self.rows,
2149 compact_rows: &self.store.rows,
2150 live_rows: &self.store.live_rows,
2151 all_rows_live: self.store.live_rows.is_all_live(),
2152 candidates,
2153 matcher: matcher.map(RelationshipMatcher::Subject),
2154 })
2155 }
2156}
2157
2158#[derive(Debug)]
2160pub struct RelationshipIter<'a> {
2161 rows: &'a [Relationship],
2162 compact_rows: &'a [RelationshipRow],
2163 live_rows: &'a LiveRows,
2164 all_rows_live: bool,
2165 candidates: CandidateRowIds<'a>,
2166 matcher: Option<RelationshipMatcher>,
2167}
2168
2169impl<'a> Iterator for RelationshipIter<'a> {
2170 type Item = &'a Relationship;
2171
2172 fn next(&mut self) -> Option<Self::Item> {
2173 let matcher = self.matcher.as_mut()?;
2174 loop {
2175 let row_id = self.candidates.next()?;
2176 if !self.all_rows_live && !self.live_rows.contains(row_id) {
2177 continue;
2178 }
2179 let compact_row = self.compact_rows.get(row_id.index())?;
2180 let relationship = self.rows.get(row_id.index())?;
2181 match matcher {
2182 RelationshipMatcher::Resource(resource) => {
2183 if resource.remaining == 0 {
2184 return None;
2185 }
2186 if compact_row.matches_resource(resource) {
2187 resource.remaining = resource.remaining.saturating_sub(1);
2188 return Some(relationship);
2189 }
2190 }
2191 RelationshipMatcher::Subject(subject) => {
2192 if compact_row.matches_subject(subject) {
2193 return Some(relationship);
2194 }
2195 }
2196 }
2197 }
2198 }
2199}
2200
2201#[derive(Debug)]
2202pub(crate) struct CompactRelationshipIter<'a> {
2203 store: &'a IndexedRelationshipStore,
2204 all_rows_live: bool,
2205 candidates: CandidateRowIds<'a>,
2206 matcher: Option<CompactRelationshipMatcher>,
2207}
2208
2209impl<'a> Iterator for CompactRelationshipIter<'a> {
2210 type Item = RelationshipRef<'a>;
2211
2212 fn next(&mut self) -> Option<Self::Item> {
2213 let matcher = self.matcher.as_mut()?;
2214 loop {
2215 let row_id = self.candidates.next()?;
2216 if !self.all_rows_live && !self.store.live_rows.contains(row_id) {
2217 continue;
2218 }
2219 let row = self.store.rows.get(row_id.index())?;
2220 match matcher {
2221 CompactRelationshipMatcher::Resource(resource) => {
2222 if resource.remaining == 0 {
2223 return None;
2224 }
2225 if row.matches_resource(resource) {
2226 resource.remaining = resource.remaining.saturating_sub(1);
2227 return Some(RelationshipRef {
2228 store: self.store,
2229 row,
2230 });
2231 }
2232 }
2233 CompactRelationshipMatcher::Subject(subject) => {
2234 if row.matches_subject(subject) {
2235 return Some(RelationshipRef {
2236 store: self.store,
2237 row,
2238 });
2239 }
2240 }
2241 }
2242 }
2243 }
2244}
2245
2246#[derive(Debug, Clone, Copy)]
2248pub(crate) struct RelationshipRef<'a> {
2249 store: &'a IndexedRelationshipStore,
2250 row: &'a RelationshipRow,
2251}
2252
2253impl RelationshipRef<'_> {
2254 fn is_deleted_by(&self, deleted: &HashSet<RelationshipRow>) -> bool {
2255 deleted.contains(self.row)
2256 }
2257
2258 pub(crate) fn resource_object_legacy(&self) -> crate::model::Object {
2259 crate::model::Object {
2260 namespace: self.store.resolve(self.row.resource_type.0).to_string(),
2261 id: self.store.resolve(self.row.resource_id.0).to_string(),
2262 }
2263 }
2264
2265 pub(crate) fn resource_type_eq(&self, expected: &ObjectType) -> bool {
2266 self.store.resolve(self.row.resource_type.0) == expected.as_str()
2267 }
2268
2269 pub(crate) fn relation_name_eq(&self, expected: &RelationName) -> bool {
2270 self.store.resolve(self.row.relation.0) == expected.as_str()
2271 }
2272
2273 pub(crate) fn relation_name_str(&self) -> &str {
2274 self.store.resolve(self.row.relation.0)
2275 }
2276
2277 pub(crate) fn direct_user_subject_id(&self) -> Option<&str> {
2278 if self.row.subject_relation.is_none()
2279 && self.store.resolve(self.row.subject_type.0) == "user"
2280 {
2281 Some(self.store.resolve(self.row.subject_id.0))
2282 } else {
2283 None
2284 }
2285 }
2286
2287 pub(crate) fn subject_userset_relation_name(
2288 &self,
2289 ) -> Result<Option<(crate::model::Object, RelationName)>, ZanzibarError> {
2290 self.row
2291 .subject_relation
2292 .map(|relation| {
2293 Ok((
2294 crate::model::Object {
2295 namespace: self.store.resolve(self.row.subject_type.0).to_string(),
2296 id: self.store.resolve(self.row.subject_id.0).to_string(),
2297 },
2298 RelationName::try_from(self.store.resolve(relation.0))?,
2299 ))
2300 })
2301 .transpose()
2302 }
2303
2304 pub(crate) fn expanded_subject(&self) -> Result<crate::model::ExpandedUserset, ZanzibarError> {
2305 match self.row.subject_relation {
2306 Some(relation) => Ok(crate::model::ExpandedUserset::Userset(
2307 crate::model::Object {
2308 namespace: self.store.resolve(self.row.subject_type.0).to_string(),
2309 id: self.store.resolve(self.row.subject_id.0).to_string(),
2310 },
2311 crate::model::Relation(self.store.resolve(relation.0).to_string()),
2312 )),
2313 None if self.store.resolve(self.row.subject_type.0) == "user" => {
2314 Ok(crate::model::ExpandedUserset::User(
2315 self.store.resolve(self.row.subject_id.0).to_string(),
2316 ))
2317 }
2318 None => Err(ZanzibarError::StorageError(format!(
2319 "legacy expand cannot represent direct subject type '{}'",
2320 self.store.resolve(self.row.subject_type.0),
2321 ))),
2322 }
2323 }
2324}
2325
2326#[derive(Debug)]
2327enum CandidateRowIds<'a> {
2328 Empty,
2329 One(Option<RowId>),
2330 OneThenSlice {
2331 first: Option<RowId>,
2332 rest: std::slice::Iter<'a, RowId>,
2333 },
2334 Slice(std::slice::Iter<'a, RowId>),
2335}
2336
2337impl Iterator for CandidateRowIds<'_> {
2338 type Item = RowId;
2339
2340 fn next(&mut self) -> Option<Self::Item> {
2341 match self {
2342 Self::Empty => None,
2343 Self::One(row_id) => row_id.take(),
2344 Self::OneThenSlice { first, rest } => first.take().or_else(|| rest.next().copied()),
2345 Self::Slice(indexes) => indexes.next().copied(),
2346 }
2347 }
2348}
2349
2350#[derive(Debug, Clone)]
2351enum PostingIndex<K> {
2352 Hash {
2353 primary: HashMap<K, RowId>,
2354 overflow: HashMap<K, Vec<RowId>>,
2355 },
2356 Sorted {
2357 keys: Vec<K>,
2358 ranges: Vec<RuntimePostingRange>,
2359 overflow: Vec<RowId>,
2360 },
2361}
2362
2363impl<K> Default for PostingIndex<K> {
2364 fn default() -> Self {
2365 Self::Hash {
2366 primary: HashMap::new(),
2367 overflow: HashMap::new(),
2368 }
2369 }
2370}
2371
2372impl<K> PostingIndex<K>
2373where
2374 K: Copy + Eq + Hash + Ord,
2375{
2376 fn from_sorted(keys: Vec<K>, ranges: Vec<RuntimePostingRange>, overflow: Vec<RowId>) -> Self {
2377 Self::Sorted {
2378 keys,
2379 ranges,
2380 overflow,
2381 }
2382 }
2383
2384 fn insert(&mut self, key: K, row_id: RowId) {
2385 self.ensure_hash_profile();
2386 if let Self::Hash { primary, overflow } = self {
2387 match primary.entry(key) {
2388 Entry::Vacant(entry) => {
2389 entry.insert(row_id);
2390 }
2391 Entry::Occupied(_) => {
2392 overflow.entry(key).or_default().push(row_id);
2393 }
2394 }
2395 }
2396 }
2397
2398 fn candidates(&self, key: &K) -> CandidateRowIds<'_> {
2399 match self {
2400 Self::Hash { primary, overflow } => {
2401 match (primary.get(key).copied(), overflow.get(key)) {
2402 (Some(row_id), Some(row_ids)) => CandidateRowIds::OneThenSlice {
2403 first: Some(row_id),
2404 rest: row_ids.iter(),
2405 },
2406 (Some(row_id), None) => CandidateRowIds::One(Some(row_id)),
2407 (None, Some(row_ids)) => CandidateRowIds::Slice(row_ids.iter()),
2408 (None, None) => CandidateRowIds::Empty,
2409 }
2410 }
2411 Self::Sorted {
2412 keys,
2413 ranges,
2414 overflow,
2415 } => match keys.binary_search(key) {
2416 Ok(index) => ranges
2417 .get(index)
2418 .map_or(CandidateRowIds::Empty, |range| range.candidates(overflow)),
2419 Err(_) => CandidateRowIds::Empty,
2420 },
2421 }
2422 }
2423
2424 fn ensure_hash_profile(&mut self) {
2425 let Self::Sorted {
2426 keys,
2427 ranges,
2428 overflow,
2429 } = self
2430 else {
2431 return;
2432 };
2433 let mut primary = HashMap::with_capacity(keys.len());
2434 let mut overflow_map = HashMap::new();
2435 for (key, range) in keys.iter().copied().zip(ranges.iter().copied()) {
2436 primary.insert(key, range.first_row_id);
2437 if let Some(row_ids) = range.overflow_slice(overflow) {
2438 overflow_map.insert(key, row_ids.to_vec());
2439 }
2440 }
2441 *self = Self::Hash {
2442 primary,
2443 overflow: overflow_map,
2444 };
2445 }
2446}
2447
2448#[derive(Debug, Clone, Copy)]
2449struct RuntimePostingRange {
2450 first_row_id: RowId,
2451 overflow_start: u32,
2452 overflow_len: u32,
2453}
2454
2455impl RuntimePostingRange {
2456 fn candidates<'a>(&self, overflow: &'a [RowId]) -> CandidateRowIds<'a> {
2457 match self.overflow_slice(overflow) {
2458 Some(row_ids) => CandidateRowIds::OneThenSlice {
2459 first: Some(self.first_row_id),
2460 rest: row_ids.iter(),
2461 },
2462 None => CandidateRowIds::One(Some(self.first_row_id)),
2463 }
2464 }
2465
2466 fn overflow_slice<'a>(&self, overflow: &'a [RowId]) -> Option<&'a [RowId]> {
2467 if self.overflow_len == 0 {
2468 return None;
2469 }
2470 let start = usize::try_from(self.overflow_start).ok()?;
2471 let len = usize::try_from(self.overflow_len).ok()?;
2472 let end = start.checked_add(len)?;
2473 overflow.get(start..end)
2474 }
2475}
2476
2477#[derive(Debug)]
2478enum RelationshipMatcher {
2479 Resource(ResourceMatcher),
2480 Subject(SubjectMatcher),
2481}
2482
2483#[derive(Debug)]
2484enum CompactRelationshipMatcher {
2485 Resource(ResourceMatcher),
2486 Subject(SubjectMatcher),
2487}
2488
2489#[derive(Debug, Clone, Copy)]
2490struct ResourceMatcher {
2491 resource_type: ObjectTypeId,
2492 optional_resource_id: Option<ObjectIdId>,
2493 optional_relation: Option<RelationId>,
2494 optional_subject: Option<SubjectMatcher>,
2495 remaining: usize,
2496}
2497
2498#[derive(Debug, Clone, Copy)]
2499struct SubjectMatcher {
2500 subject_type: SubjectTypeId,
2501 optional_subject_id: Option<SubjectIdId>,
2502 optional_relation: Option<RelationId>,
2503}
2504
2505#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2506struct SymbolId(NonZeroU32);
2507
2508impl SymbolId {
2509 fn from_index(index: usize) -> Result<Self, StoreError> {
2510 let value = index
2511 .checked_add(1)
2512 .and_then(|value| u32::try_from(value).ok())
2513 .and_then(NonZeroU32::new)
2514 .ok_or(StoreError::CapacityExceeded {
2515 component: "identifier interner",
2516 })?;
2517 Ok(Self(value))
2518 }
2519
2520 fn from_snapshot_raw(value: u32, symbol_count: u32) -> Result<Self, SnapshotIoError> {
2521 let value = NonZeroU32::new(value).ok_or(SnapshotIoError::Format {
2522 reason: "symbol id must be non-zero",
2523 })?;
2524 if value.get() > symbol_count {
2525 return Err(SnapshotIoError::Format {
2526 reason: "symbol id is out of bounds",
2527 });
2528 }
2529 Ok(Self(value))
2530 }
2531
2532 fn get(self) -> u32 {
2533 self.0.get()
2534 }
2535
2536 fn index(self) -> usize {
2537 usize::try_from(self.0.get().saturating_sub(1)).unwrap_or(usize::MAX)
2538 }
2539}
2540
2541#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2542struct ObjectTypeId(SymbolId);
2543
2544#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2545struct ObjectIdId(SymbolId);
2546
2547#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2548struct RelationId(SymbolId);
2549
2550#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2551struct SubjectTypeId(SymbolId);
2552
2553#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2554struct SubjectIdId(SymbolId);
2555
2556#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
2557struct RowId(NonZeroU32);
2558
2559impl RowId {
2560 fn from_len(len: usize) -> Result<Self, StoreError> {
2561 let value = len
2562 .checked_add(1)
2563 .and_then(|value| u32::try_from(value).ok())
2564 .and_then(NonZeroU32::new)
2565 .ok_or(StoreError::CapacityExceeded {
2566 component: "relationship rows",
2567 })?;
2568 Ok(Self(value))
2569 }
2570
2571 fn from_snapshot_raw(value: u32, row_count: u32) -> Result<Self, SnapshotIoError> {
2572 let value = NonZeroU32::new(value).ok_or(SnapshotIoError::Format {
2573 reason: "row id must be non-zero",
2574 })?;
2575 if value.get() > row_count {
2576 return Err(SnapshotIoError::Format {
2577 reason: "row id is out of bounds",
2578 });
2579 }
2580 Ok(Self(value))
2581 }
2582
2583 const fn raw(self) -> u32 {
2584 self.0.get()
2585 }
2586
2587 fn index(self) -> usize {
2588 usize::try_from(self.0.get().saturating_sub(1)).unwrap_or(usize::MAX)
2589 }
2590}
2591
2592#[derive(Debug, Clone, Default)]
2593struct IdentifierInterner {
2594 bytes: Vec<u8>,
2595 entries: Vec<InternedString>,
2596 ids_by_hash: HashMap<u64, SymbolId>,
2597 hash_collisions: HashMap<u64, Vec<SymbolId>>,
2598 hashes_by_id: Vec<u64>,
2599 sorted_lookup: Vec<SymbolId>,
2600}
2601
2602impl IdentifierInterner {
2603 fn encode_symbol_acceleration(
2604 &self,
2605 layout: SnapshotEncodingLayout,
2606 ) -> Result<(Vec<u8>, Vec<u8>, u16), SnapshotIoError> {
2607 let mut hashes_by_id = Vec::with_capacity(self.entries.len());
2608 for index in 0..self.entries.len() {
2609 let id = SymbolId::from_index(index)?;
2610 hashes_by_id.push(hash_value(self.resolve(id)?));
2611 }
2612 let mut hashes =
2613 Vec::with_capacity(checked_mul_usize(hashes_by_id.len(), DISK_SYMBOL_HASH_LEN)?);
2614 for hash in &hashes_by_id {
2615 hashes.extend_from_slice(&hash.to_le_bytes());
2616 }
2617
2618 let mut lookup = (0..self.entries.len())
2619 .map(SymbolId::from_index)
2620 .collect::<Result<Vec<_>, _>>()?;
2621 lookup.sort_by_key(|id| (hashes_by_id.get(id.index()).copied(), *id));
2622 let lookup_width =
2623 snapshot_symbol_width(checked_u32_from_usize(self.entries.len())?, layout);
2624 let mut lookup_bytes =
2625 Vec::with_capacity(checked_mul_usize(lookup.len(), lookup_width.byte_len())?);
2626 for id in lookup {
2627 lookup_width.encode_value(id.get(), &mut lookup_bytes);
2628 }
2629 Ok((hashes, lookup_bytes, lookup_width.flag_bits()))
2630 }
2631
2632 fn decode_snapshot_sections(
2633 reader: &SnapshotReader<'_>,
2634 validation: SnapshotValidationMode,
2635 ) -> Result<Self, SnapshotIoError> {
2636 let header = reader.header();
2637 let bytes = reader.section(SectionKind::SymbolBytes)?;
2638 let table = reader.section(SectionKind::SymbolTable)?;
2639 let symbol_count = checked_usize_from_u32(header.symbol_count)?;
2640 if table.row_count() != u64::from(header.symbol_count) {
2641 return Err(SnapshotIoError::Format {
2642 reason: "symbol table row count does not match header",
2643 });
2644 }
2645 let (start_width, len_width) = if header.format_version == SnapshotFormatVersion::V3 {
2646 symbol_table_widths(table.flags())?
2647 } else {
2648 if table.flags() != 0 {
2649 return Err(SnapshotIoError::Format {
2650 reason: "symbol table flags are unsupported",
2651 });
2652 }
2653 (SnapshotKeyWidth::U32, SnapshotKeyWidth::U32)
2654 };
2655 let expected_len = checked_mul_usize(
2656 symbol_count,
2657 checked_add_usize(start_width.byte_len(), len_width.byte_len())?,
2658 )?;
2659 if table.bytes().len() != expected_len {
2660 return Err(SnapshotIoError::Format {
2661 reason: "symbol table length does not match symbol count",
2662 });
2663 }
2664
2665 let mut cursor = BinaryCursor::new(table.bytes());
2666 let mut entries = Vec::with_capacity(symbol_count);
2667 for _ in 0..symbol_count {
2668 let start = start_width.read_value(&mut cursor)?;
2669 let len = len_width.read_value(&mut cursor)?;
2670 let start_usize = checked_usize_from_u32(start)?;
2671 let len_usize = checked_usize_from_u32(len)?;
2672 let end = checked_add_usize(start_usize, len_usize)?;
2673 let symbol_bytes =
2674 bytes
2675 .bytes()
2676 .get(start_usize..end)
2677 .ok_or(SnapshotIoError::Format {
2678 reason: "symbol byte range is out of bounds",
2679 })?;
2680 str::from_utf8(symbol_bytes).map_err(|_| SnapshotIoError::Format {
2681 reason: "symbol bytes are not valid utf-8",
2682 })?;
2683 entries.push(InternedString { start, len });
2684 }
2685 if !cursor.is_empty() {
2686 return Err(SnapshotIoError::Format {
2687 reason: "symbol table has trailing bytes",
2688 });
2689 }
2690 let hashes_by_id = decode_symbol_hashes(reader, header.symbol_count)?;
2691 let lookup = decode_symbol_lookup(reader, header.symbol_count, &hashes_by_id)?;
2692 Self::from_snapshot_parts(
2693 bytes.bytes().to_vec(),
2694 entries,
2695 hashes_by_id,
2696 lookup,
2697 validation,
2698 )
2699 }
2700
2701 fn from_snapshot_parts(
2702 bytes: Vec<u8>,
2703 entries: Vec<InternedString>,
2704 hashes_by_id: Vec<u64>,
2705 lookup: Vec<SymbolId>,
2706 validation: SnapshotValidationMode,
2707 ) -> Result<Self, SnapshotIoError> {
2708 let mut interner = Self {
2709 bytes,
2710 entries,
2711 ids_by_hash: HashMap::new(),
2712 hash_collisions: HashMap::new(),
2713 hashes_by_id: Vec::new(),
2714 sorted_lookup: Vec::new(),
2715 };
2716 match validation {
2717 SnapshotValidationMode::Full => {
2718 let (ids_by_hash, hash_collisions) =
2719 index_full_symbol_lookup(&interner.bytes, &interner.entries, &hashes_by_id)?;
2720 interner.ids_by_hash = ids_by_hash;
2721 interner.hash_collisions = hash_collisions;
2722 }
2723 SnapshotValidationMode::TrustedFastLoad => {
2724 interner.hashes_by_id = hashes_by_id;
2725 interner.sorted_lookup = lookup;
2726 }
2727 }
2728 Ok(interner)
2729 }
2730
2731 fn intern(&mut self, value: &str) -> Result<SymbolId, StoreError> {
2732 if let Some(id) = self.lookup(value) {
2733 return Ok(id);
2734 }
2735 let id = SymbolId::from_index(self.entries.len())?;
2736 let hash = hash_value(value);
2737 if self.sorted_lookup.is_empty() {
2738 if let Some(existing) = self.ids_by_hash.get(&hash).copied() {
2739 if self
2740 .resolve(existing)
2741 .ok()
2742 .is_some_and(|stored| stored == value)
2743 {
2744 return Ok(existing);
2745 }
2746 self.hash_collisions.entry(hash).or_default().push(id);
2747 } else {
2748 self.ids_by_hash.insert(hash, id);
2749 }
2750 } else {
2751 self.hashes_by_id.push(hash);
2752 let insert_at = self.sorted_lookup.partition_point(|candidate| {
2753 match self.symbol_hash(*candidate) {
2754 Some(candidate_hash) => (candidate_hash, *candidate) < (hash, id),
2755 None => false,
2756 }
2757 });
2758 self.sorted_lookup.insert(insert_at, id);
2759 }
2760 let start = u32::try_from(self.bytes.len()).map_err(|_| StoreError::CapacityExceeded {
2761 component: "identifier interner bytes",
2762 })?;
2763 let len = u32::try_from(value.len()).map_err(|_| StoreError::CapacityExceeded {
2764 component: "identifier interner string",
2765 })?;
2766 let end =
2767 self.bytes
2768 .len()
2769 .checked_add(value.len())
2770 .ok_or(StoreError::CapacityExceeded {
2771 component: "identifier interner bytes",
2772 })?;
2773 u32::try_from(end).map_err(|_| StoreError::CapacityExceeded {
2774 component: "identifier interner bytes",
2775 })?;
2776 self.bytes.extend_from_slice(value.as_bytes());
2777 self.entries.push(InternedString { start, len });
2778 Ok(id)
2779 }
2780
2781 fn lookup(&self, value: &str) -> Option<SymbolId> {
2782 let hash = hash_value(value);
2783 if !self.sorted_lookup.is_empty() {
2784 let start = self
2785 .sorted_lookup
2786 .partition_point(|id| self.symbol_hash(*id).is_some_and(|stored| stored < hash));
2787 for id in self.sorted_lookup.get(start..)?.iter().copied() {
2788 if self.symbol_hash(id) != Some(hash) {
2789 break;
2790 }
2791 if self.resolve(id).ok().is_some_and(|stored| stored == value) {
2792 return Some(id);
2793 }
2794 }
2795 return None;
2796 }
2797 if let Some(id) = self.ids_by_hash.get(&hash).copied()
2798 && self.resolve(id).ok().is_some_and(|stored| stored == value)
2799 {
2800 return Some(id);
2801 }
2802 self.hash_collisions.get(&hash).and_then(|ids| {
2803 ids.iter()
2804 .copied()
2805 .find(|id| self.resolve(*id).ok().is_some_and(|stored| stored == value))
2806 })
2807 }
2808
2809 fn resolve(&self, id: SymbolId) -> Result<&str, StoreError> {
2810 let entry = self
2811 .entries
2812 .get(id.index())
2813 .ok_or(StoreError::InternalInvariant {
2814 reason: "interned identifier id is out of bounds",
2815 })?;
2816 let start = usize::try_from(entry.start).map_err(|_| StoreError::InternalInvariant {
2817 reason: "interned identifier start offset is out of bounds",
2818 })?;
2819 let len = usize::try_from(entry.len).map_err(|_| StoreError::InternalInvariant {
2820 reason: "interned identifier length is out of bounds",
2821 })?;
2822 let end = start
2823 .checked_add(len)
2824 .ok_or(StoreError::InternalInvariant {
2825 reason: "interned identifier byte range overflowed",
2826 })?;
2827 let bytes = self
2828 .bytes
2829 .get(start..end)
2830 .ok_or(StoreError::InternalInvariant {
2831 reason: "interned identifier byte range is out of bounds",
2832 })?;
2833 str::from_utf8(bytes).map_err(|_| StoreError::InternalInvariant {
2834 reason: "interned identifier bytes are not valid utf-8",
2835 })
2836 }
2837
2838 fn len(&self) -> usize {
2839 self.entries.len()
2840 }
2841
2842 fn byte_len(&self) -> usize {
2843 self.bytes.len()
2844 }
2845
2846 fn symbol_hash(&self, id: SymbolId) -> Option<u64> {
2847 self.hashes_by_id.get(id.index()).copied()
2848 }
2849}
2850
2851#[derive(Debug, Clone, Copy)]
2852struct InternedString {
2853 start: u32,
2854 len: u32,
2855}
2856
2857fn encode_v3_symbol_table(
2858 entries: &[InternedString],
2859 symbol_bytes_len: u32,
2860 layout: SnapshotEncodingLayout,
2861) -> Result<(Vec<u8>, u16), SnapshotIoError> {
2862 let start_width = snapshot_symbol_width(symbol_bytes_len, layout);
2863 let max_len = entries.iter().map(|entry| entry.len).max().unwrap_or(0);
2864 let len_width = snapshot_symbol_width(max_len, layout);
2865 let entry_width = checked_add_usize(start_width.byte_len(), len_width.byte_len())?;
2866 let mut table = Vec::with_capacity(checked_mul_usize(entries.len(), entry_width)?);
2867 for entry in entries {
2868 start_width.encode_value(entry.start, &mut table);
2869 len_width.encode_value(entry.len, &mut table);
2870 }
2871 Ok((table, symbol_table_flags(start_width, len_width)))
2872}
2873
2874const fn snapshot_symbol_width(max_value: u32, layout: SnapshotEncodingLayout) -> SnapshotKeyWidth {
2875 match layout {
2876 SnapshotEncodingLayout::Compact => SnapshotKeyWidth::for_max(max_value),
2877 SnapshotEncodingLayout::CompressionFriendly => SnapshotKeyWidth::U32,
2878 }
2879}
2880
2881fn decode_symbol_hashes(
2882 reader: &SnapshotReader<'_>,
2883 symbol_count: u32,
2884) -> Result<Vec<u64>, SnapshotIoError> {
2885 let section = reader.section(SectionKind::SymbolHashes)?;
2886 if section.row_count() != u64::from(symbol_count) {
2887 return Err(SnapshotIoError::Format {
2888 reason: "symbol hash row count does not match header",
2889 });
2890 }
2891 let count = checked_usize_from_u32(symbol_count)?;
2892 let expected_len = checked_mul_usize(count, DISK_SYMBOL_HASH_LEN)?;
2893 if section.bytes().len() != expected_len {
2894 return Err(SnapshotIoError::Format {
2895 reason: "symbol hash length does not match symbol count",
2896 });
2897 }
2898
2899 let mut cursor = BinaryCursor::new(section.bytes());
2900 let mut hashes = Vec::with_capacity(count);
2901 for _ in 0..count {
2902 hashes.push(cursor.read_u64()?);
2903 }
2904 Ok(hashes)
2905}
2906
2907fn decode_symbol_lookup(
2908 reader: &SnapshotReader<'_>,
2909 symbol_count: u32,
2910 hashes_by_id: &[u64],
2911) -> Result<Vec<SymbolId>, SnapshotIoError> {
2912 let section = reader.section(SectionKind::SymbolLookup)?;
2913 if section.row_count() != u64::from(symbol_count) {
2914 return Err(SnapshotIoError::Format {
2915 reason: "symbol lookup row count does not match header",
2916 });
2917 }
2918 let count = checked_usize_from_u32(symbol_count)?;
2919 let symbol_width = if reader.header().format_version == SnapshotFormatVersion::V3 {
2920 section_width_from_flags(section.flags())?
2921 } else {
2922 if section.flags() != 0 {
2923 return Err(SnapshotIoError::Format {
2924 reason: "symbol lookup flags are unsupported",
2925 });
2926 }
2927 SnapshotKeyWidth::U32
2928 };
2929 let expected_len = checked_mul_usize(count, symbol_width.byte_len())?;
2930 if section.bytes().len() != expected_len {
2931 return Err(SnapshotIoError::Format {
2932 reason: "symbol lookup length does not match symbol count",
2933 });
2934 }
2935
2936 let mut cursor = BinaryCursor::new(section.bytes());
2937 let mut lookup = Vec::with_capacity(count);
2938 let mut previous = None;
2939 for _ in 0..count {
2940 let id = SymbolId::from_snapshot_raw(symbol_width.read_value(&mut cursor)?, symbol_count)?;
2941 let key = symbol_lookup_key(id, hashes_by_id)?;
2942 if previous.is_some_and(|candidate| candidate >= key) {
2943 return Err(SnapshotIoError::Format {
2944 reason: "symbol lookup is not strictly sorted",
2945 });
2946 }
2947 previous = Some(key);
2948 lookup.push(id);
2949 }
2950 if !cursor.is_empty() {
2951 return Err(SnapshotIoError::Format {
2952 reason: "symbol lookup has trailing bytes",
2953 });
2954 }
2955 Ok(lookup)
2956}
2957
2958fn symbol_lookup_key(
2959 id: SymbolId,
2960 hashes_by_id: &[u64],
2961) -> Result<(u64, SymbolId), SnapshotIoError> {
2962 let hash = hashes_by_id
2963 .get(id.index())
2964 .copied()
2965 .ok_or(SnapshotIoError::Format {
2966 reason: "symbol lookup hash id is out of bounds",
2967 })?;
2968 Ok((hash, id))
2969}
2970
2971type SnapshotSymbolIndex = (HashMap<u64, SymbolId>, HashMap<u64, Vec<SymbolId>>);
2972
2973fn index_full_symbol_lookup(
2974 bytes: &[u8],
2975 entries: &[InternedString],
2976 hashes_by_id: &[u64],
2977) -> Result<SnapshotSymbolIndex, SnapshotIoError> {
2978 let mut ids_by_hash = HashMap::new();
2979 let mut hash_collisions = HashMap::new();
2980 for (index, stored_hash) in hashes_by_id.iter().copied().enumerate() {
2981 let id = SymbolId::from_index(index)?;
2982 let value = resolve_symbol_entry(bytes, entries, id)?;
2983 if hash_value(value) != stored_hash {
2984 return Err(SnapshotIoError::Format {
2985 reason: "symbol lookup hash does not match symbol bytes",
2986 });
2987 }
2988 if symbol_value_exists(
2989 bytes,
2990 entries,
2991 &ids_by_hash,
2992 &hash_collisions,
2993 stored_hash,
2994 value,
2995 )? {
2996 return Err(SnapshotIoError::Format {
2997 reason: "duplicate symbol in snapshot",
2998 });
2999 }
3000 match ids_by_hash.entry(stored_hash) {
3001 Entry::Vacant(vacant) => {
3002 vacant.insert(id);
3003 }
3004 Entry::Occupied(_) => {
3005 hash_collisions
3006 .entry(stored_hash)
3007 .or_insert_with(Vec::new)
3008 .push(id);
3009 }
3010 }
3011 }
3012 Ok((ids_by_hash, hash_collisions))
3013}
3014
3015fn symbol_value_exists(
3016 bytes: &[u8],
3017 entries: &[InternedString],
3018 ids_by_hash: &HashMap<u64, SymbolId>,
3019 hash_collisions: &HashMap<u64, Vec<SymbolId>>,
3020 hash: u64,
3021 value: &str,
3022) -> Result<bool, SnapshotIoError> {
3023 if let Some(id) = ids_by_hash.get(&hash).copied()
3024 && resolve_symbol_entry(bytes, entries, id)? == value
3025 {
3026 return Ok(true);
3027 }
3028 if let Some(ids) = hash_collisions.get(&hash) {
3029 for id in ids {
3030 if resolve_symbol_entry(bytes, entries, *id)? == value {
3031 return Ok(true);
3032 }
3033 }
3034 }
3035 Ok(false)
3036}
3037
3038fn resolve_symbol_entry<'a>(
3039 bytes: &'a [u8],
3040 entries: &[InternedString],
3041 id: SymbolId,
3042) -> Result<&'a str, SnapshotIoError> {
3043 let entry = entries.get(id.index()).ok_or(SnapshotIoError::Format {
3044 reason: "symbol lookup id is out of bounds",
3045 })?;
3046 let start = checked_usize_from_u32(entry.start)?;
3047 let len = checked_usize_from_u32(entry.len)?;
3048 let end = checked_add_usize(start, len)?;
3049 let value = bytes.get(start..end).ok_or(SnapshotIoError::Format {
3050 reason: "symbol byte range is out of bounds",
3051 })?;
3052 str::from_utf8(value).map_err(|_| SnapshotIoError::Format {
3053 reason: "symbol bytes are not valid utf-8",
3054 })
3055}
3056
3057#[derive(Debug, Clone, Default)]
3058struct RelationshipIdentityIndex {
3059 primary: HashMap<u64, RowId>,
3060 hash_collisions: HashMap<u64, Vec<RowId>>,
3061}
3062
3063impl RelationshipIdentityIndex {
3064 fn find(&self, rows: &[RelationshipRow], row: &RelationshipRow) -> Option<RowId> {
3065 let hash = row.identity_hash();
3066 if let Some(row_id) = self.primary.get(&hash).copied()
3067 && rows
3068 .get(row_id.index())
3069 .is_some_and(|candidate| candidate == row)
3070 {
3071 return Some(row_id);
3072 }
3073 self.hash_collisions.get(&hash).and_then(|row_ids| {
3074 row_ids.iter().copied().find(|row_id| {
3075 rows.get(row_id.index())
3076 .is_some_and(|candidate| candidate == row)
3077 })
3078 })
3079 }
3080
3081 fn insert(&mut self, rows: &[RelationshipRow], row_id: RowId, row: &RelationshipRow) {
3082 let hash = row.identity_hash();
3083 match self.primary.get(&hash).copied() {
3084 Some(existing)
3085 if rows
3086 .get(existing.index())
3087 .is_some_and(|candidate| candidate != row) =>
3088 {
3089 self.hash_collisions.entry(hash).or_default().push(row_id);
3090 }
3091 Some(_) => {}
3092 None => {
3093 self.primary.insert(hash, row_id);
3094 }
3095 }
3096 }
3097
3098 fn remove(&mut self, rows: &[RelationshipRow], row: &RelationshipRow) -> Option<RowId> {
3099 let hash = row.identity_hash();
3100 if let Some(row_id) = self.primary.get(&hash).copied()
3101 && rows
3102 .get(row_id.index())
3103 .is_some_and(|candidate| candidate == row)
3104 {
3105 let removed = self.primary.remove(&hash)?;
3106 if let Some(collisions) = self.hash_collisions.get_mut(&hash)
3107 && let Some(promoted) = collisions.pop()
3108 {
3109 self.primary.insert(hash, promoted);
3110 if collisions.is_empty() {
3111 self.hash_collisions.remove(&hash);
3112 }
3113 }
3114 return Some(removed);
3115 }
3116
3117 let collisions = self.hash_collisions.get_mut(&hash)?;
3118 let index = collisions.iter().position(|row_id| {
3119 rows.get(row_id.index())
3120 .is_some_and(|candidate| candidate == row)
3121 })?;
3122 let removed = collisions.swap_remove(index);
3123 if collisions.is_empty() {
3124 self.hash_collisions.remove(&hash);
3125 }
3126 Some(removed)
3127 }
3128}
3129
3130#[derive(Debug, Clone)]
3131enum UniquenessState {
3132 Ready(RelationshipIdentityIndex),
3133 KnownUniqueButNotIndexed,
3134 UntrustedNotIndexed,
3135}
3136
3137#[derive(Debug, Clone)]
3138struct RelationshipRow {
3139 row_id: RowId,
3140 resource_type: ObjectTypeId,
3141 resource_id: ObjectIdId,
3142 relation: RelationId,
3143 subject_type: SubjectTypeId,
3144 subject_id: SubjectIdId,
3145 subject_relation: Option<RelationId>,
3146}
3147
3148impl PartialEq for RelationshipRow {
3149 fn eq(&self, other: &Self) -> bool {
3150 self.resource_type == other.resource_type
3151 && self.resource_id == other.resource_id
3152 && self.relation == other.relation
3153 && self.subject_type == other.subject_type
3154 && self.subject_id == other.subject_id
3155 && self.subject_relation == other.subject_relation
3156 }
3157}
3158
3159impl Eq for RelationshipRow {}
3160
3161impl Hash for RelationshipRow {
3162 fn hash<H: Hasher>(&self, state: &mut H) {
3163 self.resource_type.hash(state);
3164 self.resource_id.hash(state);
3165 self.relation.hash(state);
3166 self.subject_type.hash(state);
3167 self.subject_id.hash(state);
3168 self.subject_relation.hash(state);
3169 }
3170}
3171
3172impl RelationshipRow {
3173 fn identity_hash(&self) -> u64 {
3174 let mut state = DefaultHasher::new();
3175 self.hash(&mut state);
3176 state.finish()
3177 }
3178
3179 fn from_relationship(
3180 row_id: RowId,
3181 relationship: &Relationship,
3182 interner: &mut IdentifierInterner,
3183 ) -> Result<Self, StoreError> {
3184 let resource_type =
3185 ObjectTypeId(interner.intern(relationship.resource().object_type().as_str())?);
3186 let resource_id =
3187 ObjectIdId(interner.intern(relationship.resource().object_id().as_str())?);
3188 let relation = RelationId(interner.intern(relationship.relation().as_str())?);
3189 let (subject_type, subject_id, subject_relation) = match relationship.subject() {
3190 SubjectRef::Object(object) => (
3191 SubjectTypeId(interner.intern(object.object_type().as_str())?),
3192 SubjectIdId(interner.intern(object.object_id().as_str())?),
3193 None,
3194 ),
3195 SubjectRef::Userset { object, relation } => (
3196 SubjectTypeId(interner.intern(object.object_type().as_str())?),
3197 SubjectIdId(interner.intern(object.object_id().as_str())?),
3198 Some(RelationId(interner.intern(relation.as_str())?)),
3199 ),
3200 };
3201 Ok(Self {
3202 row_id,
3203 resource_type,
3204 resource_id,
3205 relation,
3206 subject_type,
3207 subject_id,
3208 subject_relation,
3209 })
3210 }
3211
3212 fn from_existing_relationship(
3213 relationship: &Relationship,
3214 interner: &IdentifierInterner,
3215 ) -> Option<Self> {
3216 let resource_type =
3217 ObjectTypeId(interner.lookup(relationship.resource().object_type().as_str())?);
3218 let resource_id =
3219 ObjectIdId(interner.lookup(relationship.resource().object_id().as_str())?);
3220 let relation = RelationId(interner.lookup(relationship.relation().as_str())?);
3221 let (subject_type, subject_id, subject_relation) = match relationship.subject() {
3222 SubjectRef::Object(object) => (
3223 SubjectTypeId(interner.lookup(object.object_type().as_str())?),
3224 SubjectIdId(interner.lookup(object.object_id().as_str())?),
3225 None,
3226 ),
3227 SubjectRef::Userset { object, relation } => (
3228 SubjectTypeId(interner.lookup(object.object_type().as_str())?),
3229 SubjectIdId(interner.lookup(object.object_id().as_str())?),
3230 Some(RelationId(interner.lookup(relation.as_str())?)),
3231 ),
3232 };
3233 Some(Self {
3234 row_id: RowId(NonZeroU32::MIN),
3235 resource_type,
3236 resource_id,
3237 relation,
3238 subject_type,
3239 subject_id,
3240 subject_relation,
3241 })
3242 }
3243
3244 fn matches_resource(&self, matcher: &ResourceMatcher) -> bool {
3245 self.resource_type == matcher.resource_type
3246 && matcher
3247 .optional_resource_id
3248 .is_none_or(|resource_id| self.resource_id == resource_id)
3249 && matcher
3250 .optional_relation
3251 .is_none_or(|relation| self.relation == relation)
3252 && matcher
3253 .optional_subject
3254 .is_none_or(|subject| self.matches_subject(&subject))
3255 }
3256
3257 fn matches_subject(&self, matcher: &SubjectMatcher) -> bool {
3258 self.subject_type == matcher.subject_type
3259 && matcher
3260 .optional_subject_id
3261 .is_none_or(|subject_id| self.subject_id == subject_id)
3262 && matcher
3263 .optional_relation
3264 .is_none_or(|relation| self.subject_relation == Some(relation))
3265 }
3266}
3267
3268#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3269struct ResourceIndexKey {
3270 object_type: ObjectTypeId,
3271 object_id: ObjectIdId,
3272 relation: RelationId,
3273}
3274
3275impl From<&RelationshipRow> for ResourceIndexKey {
3276 fn from(value: &RelationshipRow) -> Self {
3277 Self {
3278 object_type: value.resource_type,
3279 object_id: value.resource_id,
3280 relation: value.relation,
3281 }
3282 }
3283}
3284
3285#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3286struct ResourceObjectIndexKey {
3287 object_type: ObjectTypeId,
3288 object_id: ObjectIdId,
3289}
3290
3291impl From<&RelationshipRow> for ResourceObjectIndexKey {
3292 fn from(value: &RelationshipRow) -> Self {
3293 Self {
3294 object_type: value.resource_type,
3295 object_id: value.resource_id,
3296 }
3297 }
3298}
3299
3300#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3301struct ResourceTypeRelationIndexKey {
3302 object_type: ObjectTypeId,
3303 relation: RelationId,
3304}
3305
3306impl From<&RelationshipRow> for ResourceTypeRelationIndexKey {
3307 fn from(value: &RelationshipRow) -> Self {
3308 Self {
3309 object_type: value.resource_type,
3310 relation: value.relation,
3311 }
3312 }
3313}
3314
3315#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3316struct SubjectIndexKey {
3317 subject_type: SubjectTypeId,
3318 subject_id: SubjectIdId,
3319 relation: Option<RelationId>,
3320}
3321
3322impl SubjectIndexKey {
3323 fn from_row(row: &RelationshipRow) -> Vec<Self> {
3324 match row.subject_relation {
3325 Some(relation) => vec![
3326 Self {
3327 subject_type: row.subject_type,
3328 subject_id: row.subject_id,
3329 relation: Some(relation),
3330 },
3331 Self {
3332 subject_type: row.subject_type,
3333 subject_id: row.subject_id,
3334 relation: None,
3335 },
3336 ],
3337 None => vec![Self {
3338 subject_type: row.subject_type,
3339 subject_id: row.subject_id,
3340 relation: None,
3341 }],
3342 }
3343 }
3344}
3345
3346#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3347struct SubjectTypeRelationIndexKey {
3348 subject_type: SubjectTypeId,
3349 relation: RelationId,
3350}
3351
3352impl SubjectTypeRelationIndexKey {
3353 fn from_row(row: &RelationshipRow) -> Option<Self> {
3354 row.subject_relation.map(|relation| Self {
3355 subject_type: row.subject_type,
3356 relation,
3357 })
3358 }
3359}
3360
3361#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
3362struct DiskRelationshipRow {
3363 resource_type: u32,
3364 resource_id: u32,
3365 relation: u32,
3366 subject_type: u32,
3367 subject_id: u32,
3368 subject_relation: u32,
3369}
3370
3371impl DiskRelationshipRow {
3372 fn encode_width(&self, width: SnapshotKeyWidth, target: &mut Vec<u8>) {
3373 width.encode_value(self.resource_type, target);
3374 width.encode_value(self.resource_id, target);
3375 width.encode_value(self.relation, target);
3376 width.encode_value(self.subject_type, target);
3377 width.encode_value(self.subject_id, target);
3378 width.encode_value(self.subject_relation, target);
3379 }
3380}
3381
3382impl From<&RelationshipRow> for DiskRelationshipRow {
3383 fn from(value: &RelationshipRow) -> Self {
3384 Self {
3385 resource_type: value.resource_type.0.get(),
3386 resource_id: value.resource_id.0.get(),
3387 relation: value.relation.0.get(),
3388 subject_type: value.subject_type.0.get(),
3389 subject_id: value.subject_id.0.get(),
3390 subject_relation: value
3391 .subject_relation
3392 .map_or(0, |relation| relation.0.get()),
3393 }
3394 }
3395}
3396
3397#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
3398struct DiskIndexKey {
3399 first: u32,
3400 second: u32,
3401 third: u32,
3402}
3403
3404impl DiskIndexKey {
3405 fn max_field(self, field_count: usize) -> Result<u32, SnapshotIoError> {
3406 match field_count {
3407 1 => Ok(self.first),
3408 2 => Ok(self.first.max(self.second)),
3409 3 => Ok(self.first.max(self.second).max(self.third)),
3410 _ => Err(SnapshotIoError::Format {
3411 reason: "index key field count is unsupported",
3412 }),
3413 }
3414 }
3415}
3416
3417#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3418enum SnapshotKeyWidth {
3419 U8,
3420 U16,
3421 U24,
3422 U32,
3423}
3424
3425impl SnapshotKeyWidth {
3426 const fn byte_len(self) -> usize {
3427 match self {
3428 Self::U8 => 1,
3429 Self::U16 => 2,
3430 Self::U24 => 3,
3431 Self::U32 => 4,
3432 }
3433 }
3434
3435 const fn flag_bits(self) -> u16 {
3436 match self {
3437 Self::U8 => 0,
3438 Self::U16 => 1,
3439 Self::U24 => 2,
3440 Self::U32 => 3,
3441 }
3442 }
3443
3444 fn from_flag_bits(value: u16) -> Result<Self, SnapshotIoError> {
3445 match value {
3446 0 => Ok(Self::U8),
3447 1 => Ok(Self::U16),
3448 2 => Ok(Self::U24),
3449 3 => Ok(Self::U32),
3450 _ => Err(SnapshotIoError::Format {
3451 reason: "snapshot index key width is unsupported",
3452 }),
3453 }
3454 }
3455
3456 const fn for_max(value: u32) -> Self {
3457 if value <= u8::MAX as u32 {
3458 Self::U8
3459 } else if value <= u16::MAX as u32 {
3460 Self::U16
3461 } else if value <= 0x00FF_FFFF {
3462 Self::U24
3463 } else {
3464 Self::U32
3465 }
3466 }
3467
3468 fn encode_value(self, value: u32, target: &mut Vec<u8>) {
3469 match self {
3470 Self::U8 => target.extend(value.to_le_bytes().into_iter().take(1)),
3471 Self::U16 => target.extend(value.to_le_bytes().into_iter().take(2)),
3472 Self::U24 => target.extend(value.to_le_bytes().into_iter().take(3)),
3473 Self::U32 => target.extend_from_slice(&value.to_le_bytes()),
3474 }
3475 }
3476
3477 fn read_value(self, cursor: &mut BinaryCursor<'_>) -> Result<u32, SnapshotIoError> {
3478 match self {
3479 Self::U8 => {
3480 let [value] = cursor.read_array::<1>()?;
3481 Ok(u32::from(value))
3482 }
3483 Self::U16 => Ok(u32::from(u16::from_le_bytes(cursor.read_array::<2>()?))),
3484 Self::U24 => {
3485 let [first, second, third] = cursor.read_array::<3>()?;
3486 Ok(u32::from_le_bytes([first, second, third, 0]))
3487 }
3488 Self::U32 => Ok(u32::from_le_bytes(cursor.read_array::<4>()?)),
3489 }
3490 }
3491}
3492
3493fn section_width_from_flags(flags: u16) -> Result<SnapshotKeyWidth, SnapshotIoError> {
3494 if flags & !SECTION_WIDTH_MASK != 0 {
3495 return Err(SnapshotIoError::Format {
3496 reason: "section width flags are unsupported",
3497 });
3498 }
3499 SnapshotKeyWidth::from_flag_bits(flags & SECTION_WIDTH_MASK)
3500}
3501
3502fn symbol_table_flags(start_width: SnapshotKeyWidth, len_width: SnapshotKeyWidth) -> u16 {
3503 start_width.flag_bits() | (len_width.flag_bits() << SYMBOL_TABLE_LEN_WIDTH_SHIFT)
3504}
3505
3506fn symbol_table_widths(
3507 flags: u16,
3508) -> Result<(SnapshotKeyWidth, SnapshotKeyWidth), SnapshotIoError> {
3509 if flags & !(SECTION_WIDTH_MASK | SYMBOL_TABLE_LEN_WIDTH_MASK) != 0 {
3510 return Err(SnapshotIoError::Format {
3511 reason: "symbol table flags are unsupported",
3512 });
3513 }
3514 let start_width = SnapshotKeyWidth::from_flag_bits(flags & SECTION_WIDTH_MASK)?;
3515 let len_width = SnapshotKeyWidth::from_flag_bits(
3516 (flags & SYMBOL_TABLE_LEN_WIDTH_MASK) >> SYMBOL_TABLE_LEN_WIDTH_SHIFT,
3517 )?;
3518 Ok((start_width, len_width))
3519}
3520
3521#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3522enum SnapshotIndexEncoding {
3523 FixedV2,
3524 CompactV3 { key_width: SnapshotKeyWidth },
3525}
3526
3527impl SnapshotIndexEncoding {
3528 const fn flags(self) -> u16 {
3529 match self {
3530 Self::FixedV2 => 0,
3531 Self::CompactV3 { key_width } => {
3532 INDEX_DIRECTORY_FLAG_V3_COMPACT
3533 | (key_width.flag_bits() << INDEX_DIRECTORY_KEY_WIDTH_SHIFT)
3534 }
3535 }
3536 }
3537
3538 fn from_flags(flags: u16) -> Result<Self, SnapshotIoError> {
3539 if flags == 0 {
3540 return Ok(Self::FixedV2);
3541 }
3542 if flags & INDEX_DIRECTORY_FLAG_V3_COMPACT == 0 {
3543 return Err(SnapshotIoError::Format {
3544 reason: "index directory flags are unsupported",
3545 });
3546 }
3547 let known = INDEX_DIRECTORY_FLAG_V3_COMPACT | INDEX_DIRECTORY_KEY_WIDTH_MASK;
3548 if flags & !known != 0 {
3549 return Err(SnapshotIoError::Format {
3550 reason: "index directory flags are unsupported",
3551 });
3552 }
3553 let width_bits =
3554 (flags & INDEX_DIRECTORY_KEY_WIDTH_MASK) >> INDEX_DIRECTORY_KEY_WIDTH_SHIFT;
3555 Ok(Self::CompactV3 {
3556 key_width: SnapshotKeyWidth::from_flag_bits(width_bits)?,
3557 })
3558 }
3559}
3560
3561#[derive(Debug, Clone, Copy)]
3562struct DiskPostingRange {
3563 first_row_id: u32,
3564 overflow_start: u32,
3565 overflow_len: u32,
3566}
3567
3568impl DiskPostingRange {
3569 fn encode(&self, target: &mut Vec<u8>) {
3570 target.extend_from_slice(&self.first_row_id.to_le_bytes());
3571 target.extend_from_slice(&self.overflow_start.to_le_bytes());
3572 target.extend_from_slice(&self.overflow_len.to_le_bytes());
3573 }
3574}
3575
3576#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
3577enum SnapshotIndexKind {
3578 Resource = 1,
3579 ResourceObject = 2,
3580 ResourceTypeRelation = 3,
3581 ResourceType = 4,
3582 Subject = 5,
3583 SubjectTypeRelation = 6,
3584 SubjectType = 7,
3585}
3586
3587impl SnapshotIndexKind {
3588 const ALL: [Self; SNAPSHOT_INDEX_KIND_COUNT] = [
3589 Self::Resource,
3590 Self::ResourceObject,
3591 Self::ResourceTypeRelation,
3592 Self::ResourceType,
3593 Self::Subject,
3594 Self::SubjectTypeRelation,
3595 Self::SubjectType,
3596 ];
3597
3598 fn from_raw(value: u16) -> Result<Self, SnapshotIoError> {
3599 match value {
3600 1 => Ok(Self::Resource),
3601 2 => Ok(Self::ResourceObject),
3602 3 => Ok(Self::ResourceTypeRelation),
3603 4 => Ok(Self::ResourceType),
3604 5 => Ok(Self::Subject),
3605 6 => Ok(Self::SubjectTypeRelation),
3606 7 => Ok(Self::SubjectType),
3607 _ => Err(SnapshotIoError::Format {
3608 reason: "unknown snapshot index kind",
3609 }),
3610 }
3611 }
3612
3613 const fn raw(self) -> u16 {
3614 self as u16
3615 }
3616
3617 const fn key_field_count(self) -> usize {
3618 match self {
3619 Self::Resource | Self::Subject => 3,
3620 Self::ResourceObject | Self::ResourceTypeRelation | Self::SubjectTypeRelation => 2,
3621 Self::ResourceType | Self::SubjectType => 1,
3622 }
3623 }
3624}
3625
3626#[derive(Debug)]
3627struct EncodedSnapshotIndexes {
3628 directory: Vec<u8>,
3629 keys: Vec<u8>,
3630 ranges: Vec<u8>,
3631 posting_row_ids: Vec<u8>,
3632 key_count: u64,
3633 range_count: u64,
3634 posting_row_id_count: u64,
3635}
3636
3637impl EncodedSnapshotIndexes {
3638 fn from_rows(
3639 rows: &[DiskRelationshipRow],
3640 index_profile: IndexProfile,
3641 ) -> Result<Self, SnapshotIoError> {
3642 let mut groups = SnapshotIndexGroups::default();
3643 for (index, row) in rows.iter().copied().enumerate() {
3644 let row_id =
3645 checked_u32_from_usize(index.checked_add(1).ok_or(SnapshotIoError::Format {
3646 reason: "row id overflowed",
3647 })?)?;
3648 groups.insert_row(row, row_id, index_profile);
3649 }
3650
3651 let mut directory = Vec::with_capacity(
3652 SNAPSHOT_INDEX_KIND_COUNT
3653 .checked_mul(DISK_INDEX_DIRECTORY_LEN)
3654 .ok_or(SnapshotIoError::Format {
3655 reason: "index directory length overflowed",
3656 })?,
3657 );
3658 let mut keys = Vec::new();
3659 let mut ranges = Vec::new();
3660 let mut posting_row_ids = Vec::new();
3661 let mut key_count = 0_u32;
3662 let mut range_count = 0_u32;
3663 let mut posting_row_id_count = 0_u32;
3664
3665 for kind in SnapshotIndexKind::ALL {
3666 let group = groups.group(kind);
3667 let key_start = checked_u32_from_usize(keys.len())?;
3668 let range_start = range_count;
3669 let encoded_group =
3670 encode_v3_index_group(kind, group, &mut keys, &mut ranges, &mut posting_row_ids)?;
3671 key_count =
3672 key_count
3673 .checked_add(encoded_group.key_count)
3674 .ok_or(SnapshotIoError::Format {
3675 reason: "index key count overflowed",
3676 })?;
3677 range_count = range_count.checked_add(encoded_group.multi_count).ok_or(
3678 SnapshotIoError::Format {
3679 reason: "posting range count overflowed",
3680 },
3681 )?;
3682 posting_row_id_count = posting_row_id_count
3683 .checked_add(encoded_group.overflow_row_id_count)
3684 .ok_or(SnapshotIoError::Format {
3685 reason: "posting row id count overflowed",
3686 })?;
3687 directory.extend_from_slice(&kind.raw().to_le_bytes());
3688 directory.extend_from_slice(&encoded_group.encoding.flags().to_le_bytes());
3689 directory.extend_from_slice(&key_start.to_le_bytes());
3690 directory.extend_from_slice(&encoded_group.key_count.to_le_bytes());
3691 directory.extend_from_slice(&range_start.to_le_bytes());
3692 directory.extend_from_slice(&encoded_group.multi_count.to_le_bytes());
3693 }
3694
3695 Ok(Self {
3696 directory,
3697 keys,
3698 ranges,
3699 posting_row_ids,
3700 key_count: u64::from(key_count),
3701 range_count: u64::from(range_count),
3702 posting_row_id_count: u64::from(posting_row_id_count),
3703 })
3704 }
3705}
3706
3707#[derive(Debug, Default)]
3708struct SnapshotIndexGroups {
3709 resource: BTreeMap<DiskIndexKey, Vec<u32>>,
3710 resource_object: BTreeMap<DiskIndexKey, Vec<u32>>,
3711 resource_type_relation: BTreeMap<DiskIndexKey, Vec<u32>>,
3712 resource_type: BTreeMap<DiskIndexKey, Vec<u32>>,
3713 subject: BTreeMap<DiskIndexKey, Vec<u32>>,
3714 subject_type_relation: BTreeMap<DiskIndexKey, Vec<u32>>,
3715 subject_type: BTreeMap<DiskIndexKey, Vec<u32>>,
3716}
3717
3718impl SnapshotIndexGroups {
3719 fn insert_row(&mut self, row: DiskRelationshipRow, row_id: u32, index_profile: IndexProfile) {
3720 self.resource
3721 .entry(DiskIndexKey {
3722 first: row.resource_type,
3723 second: row.resource_id,
3724 third: row.relation,
3725 })
3726 .or_default()
3727 .push(row_id);
3728 if !index_profile.supports_broad_resource_indexes() {
3729 return;
3730 }
3731 self.resource_object
3732 .entry(DiskIndexKey {
3733 first: row.resource_type,
3734 second: row.resource_id,
3735 third: 0,
3736 })
3737 .or_default()
3738 .push(row_id);
3739 self.resource_type_relation
3740 .entry(DiskIndexKey {
3741 first: row.resource_type,
3742 second: row.relation,
3743 third: 0,
3744 })
3745 .or_default()
3746 .push(row_id);
3747 self.resource_type
3748 .entry(DiskIndexKey {
3749 first: row.resource_type,
3750 second: 0,
3751 third: 0,
3752 })
3753 .or_default()
3754 .push(row_id);
3755 if !index_profile.supports_subject_reverse_lookup() {
3756 return;
3757 }
3758 self.subject
3759 .entry(DiskIndexKey {
3760 first: row.subject_type,
3761 second: row.subject_id,
3762 third: row.subject_relation,
3763 })
3764 .or_default()
3765 .push(row_id);
3766 if row.subject_relation != 0 {
3767 self.subject
3768 .entry(DiskIndexKey {
3769 first: row.subject_type,
3770 second: row.subject_id,
3771 third: 0,
3772 })
3773 .or_default()
3774 .push(row_id);
3775 self.subject_type_relation
3776 .entry(DiskIndexKey {
3777 first: row.subject_type,
3778 second: row.subject_relation,
3779 third: 0,
3780 })
3781 .or_default()
3782 .push(row_id);
3783 }
3784 self.subject_type
3785 .entry(DiskIndexKey {
3786 first: row.subject_type,
3787 second: 0,
3788 third: 0,
3789 })
3790 .or_default()
3791 .push(row_id);
3792 }
3793
3794 fn group(&self, kind: SnapshotIndexKind) -> &BTreeMap<DiskIndexKey, Vec<u32>> {
3795 match kind {
3796 SnapshotIndexKind::Resource => &self.resource,
3797 SnapshotIndexKind::ResourceObject => &self.resource_object,
3798 SnapshotIndexKind::ResourceTypeRelation => &self.resource_type_relation,
3799 SnapshotIndexKind::ResourceType => &self.resource_type,
3800 SnapshotIndexKind::Subject => &self.subject,
3801 SnapshotIndexKind::SubjectTypeRelation => &self.subject_type_relation,
3802 SnapshotIndexKind::SubjectType => &self.subject_type,
3803 }
3804 }
3805}
3806
3807#[derive(Debug, Clone, Copy)]
3808struct EncodedIndexGroup {
3809 encoding: SnapshotIndexEncoding,
3810 key_count: u32,
3811 multi_count: u32,
3812 overflow_row_id_count: u32,
3813}
3814
3815fn encode_v3_index_group(
3816 kind: SnapshotIndexKind,
3817 group: &BTreeMap<DiskIndexKey, Vec<u32>>,
3818 keys: &mut Vec<u8>,
3819 ranges: &mut Vec<u8>,
3820 posting_row_ids: &mut Vec<u8>,
3821) -> Result<EncodedIndexGroup, SnapshotIoError> {
3822 let key_count = checked_u32_from_usize(group.len())?;
3823 let multi_count =
3824 checked_u32_from_usize(group.values().filter(|row_ids| row_ids.len() > 1).count())?;
3825 let key_width = snapshot_key_width(kind, group)?;
3826 let encoding = SnapshotIndexEncoding::CompactV3 { key_width };
3827 let mut overflow_row_id_count = 0_u32;
3828
3829 for (key, row_ids) in group.iter().filter(|(_, row_ids)| row_ids.len() == 1) {
3830 encode_v3_key(kind, *key, key_width, keys);
3831 let row_id = *row_ids.first().ok_or(SnapshotIoError::Format {
3832 reason: "empty posting list",
3833 })?;
3834 keys.extend_from_slice(&row_id.to_le_bytes());
3835 }
3836
3837 for (key, row_ids) in group.iter().filter(|(_, row_ids)| row_ids.len() > 1) {
3838 encode_v3_key(kind, *key, key_width, keys);
3839 let range = encode_v3_posting_range(row_ids, posting_row_ids)?;
3840 overflow_row_id_count = overflow_row_id_count
3841 .checked_add(range.overflow_row_id_count)
3842 .ok_or(SnapshotIoError::Format {
3843 reason: "posting row id count overflowed",
3844 })?;
3845 range.range.encode(ranges);
3846 }
3847
3848 Ok(EncodedIndexGroup {
3849 encoding,
3850 key_count,
3851 multi_count,
3852 overflow_row_id_count,
3853 })
3854}
3855
3856fn snapshot_key_width(
3857 kind: SnapshotIndexKind,
3858 group: &BTreeMap<DiskIndexKey, Vec<u32>>,
3859) -> Result<SnapshotKeyWidth, SnapshotIoError> {
3860 let field_count = kind.key_field_count();
3861 let mut max_field = 0_u32;
3862 for key in group.keys().copied() {
3863 max_field = max_field.max(key.max_field(field_count)?);
3864 }
3865 Ok(SnapshotKeyWidth::for_max(max_field))
3866}
3867
3868fn encode_v3_key(
3869 kind: SnapshotIndexKind,
3870 key: DiskIndexKey,
3871 width: SnapshotKeyWidth,
3872 target: &mut Vec<u8>,
3873) {
3874 width.encode_value(key.first, target);
3875 if kind.key_field_count() >= 2 {
3876 width.encode_value(key.second, target);
3877 }
3878 if kind.key_field_count() >= 3 {
3879 width.encode_value(key.third, target);
3880 }
3881}
3882
3883#[derive(Debug, Clone, Copy)]
3884struct EncodedPostingRange {
3885 range: DiskPostingRange,
3886 overflow_row_id_count: u32,
3887}
3888
3889fn encode_v3_posting_range(
3890 row_ids: &[u32],
3891 posting_row_ids: &mut Vec<u8>,
3892) -> Result<EncodedPostingRange, SnapshotIoError> {
3893 let (first, rest) = row_ids.split_first().ok_or(SnapshotIoError::Format {
3894 reason: "empty posting list",
3895 })?;
3896 let overflow_start = checked_u32_from_usize(posting_row_ids.len())?;
3897 let overflow_row_id_count = checked_u32_from_usize(rest.len())?;
3898 let mut previous = *first;
3899 for row_id in rest {
3900 if *row_id <= previous {
3901 return Err(SnapshotIoError::Format {
3902 reason: "posting row ids are not strictly increasing",
3903 });
3904 }
3905 let delta = row_id
3906 .checked_sub(previous)
3907 .ok_or(SnapshotIoError::Format {
3908 reason: "posting row id delta underflowed",
3909 })?;
3910 encode_posting_delta_varint(delta, posting_row_ids)?;
3911 previous = *row_id;
3912 }
3913 let overflow_len = checked_u32_from_usize(
3914 posting_row_ids
3915 .len()
3916 .checked_sub(checked_usize_from_u32(overflow_start)?)
3917 .ok_or(SnapshotIoError::Format {
3918 reason: "posting row id length underflowed",
3919 })?,
3920 )?;
3921 Ok(EncodedPostingRange {
3922 range: DiskPostingRange {
3923 first_row_id: *first,
3924 overflow_start,
3925 overflow_len,
3926 },
3927 overflow_row_id_count,
3928 })
3929}
3930
3931fn encode_posting_delta_varint(value: u32, target: &mut Vec<u8>) -> Result<(), SnapshotIoError> {
3932 if value == 0 {
3933 return Err(SnapshotIoError::Format {
3934 reason: "posting row id delta must be non-zero",
3935 });
3936 }
3937 let mut remaining = value;
3938 while remaining >= 0x80 {
3939 let low_bits = u8::try_from(remaining & 0x7F).map_err(|_| SnapshotIoError::Format {
3940 reason: "posting row id delta overflowed",
3941 })?;
3942 target.push(low_bits | 0x80);
3943 remaining >>= 7;
3944 }
3945 let final_byte = u8::try_from(remaining).map_err(|_| SnapshotIoError::Format {
3946 reason: "posting row id delta overflowed",
3947 })?;
3948 target.push(final_byte);
3949 Ok(())
3950}
3951
3952#[derive(Debug)]
3953struct DecodedSnapshotIndexes {
3954 resource: PostingIndex<ResourceIndexKey>,
3955 resource_object: PostingIndex<ResourceObjectIndexKey>,
3956 resource_type_relation: PostingIndex<ResourceTypeRelationIndexKey>,
3957 resource_type: PostingIndex<ObjectTypeId>,
3958 subject: PostingIndex<SubjectIndexKey>,
3959 subject_type_relation: PostingIndex<SubjectTypeRelationIndexKey>,
3960 subject_type: PostingIndex<SubjectTypeId>,
3961}
3962
3963impl DecodedSnapshotIndexes {
3964 fn decode(
3965 reader: &SnapshotReader<'_>,
3966 rows: &[RelationshipRow],
3967 profile: SnapshotLoadProfile,
3968 validation: SnapshotValidationMode,
3969 ) -> Result<Self, SnapshotIoError> {
3970 let directory = decode_index_directory(reader)?;
3971 let keys = decode_index_keys(reader)?;
3972 let ranges = decode_posting_ranges(reader)?;
3973 let posting_row_ids = decode_posting_row_ids(reader)?;
3974 let row_count = reader.header().relationship_count;
3975 let symbol_count = reader.header().symbol_count;
3976 validate_index_section_layout(&directory, &keys, &ranges, &posting_row_ids, row_count)?;
3977 let input = SnapshotIndexDecodeInput {
3978 directory: &directory,
3979 keys: &keys,
3980 ranges: &ranges,
3981 posting_row_ids: &posting_row_ids,
3982 rows,
3983 row_count,
3984 symbol_count,
3985 index_profile: reader.header().index_profile,
3986 profile,
3987 validation,
3988 };
3989
3990 let mut decoded_posting_row_id_count = 0_u64;
3991 let resource = decode_resource_index(&input, &mut decoded_posting_row_id_count)?;
3992 let resource_object =
3993 decode_resource_object_index(&input, &mut decoded_posting_row_id_count)?;
3994 let resource_type_relation =
3995 decode_resource_type_relation_index(&input, &mut decoded_posting_row_id_count)?;
3996 let resource_type = decode_resource_type_index(&input, &mut decoded_posting_row_id_count)?;
3997 let subject = decode_subject_index(&input, &mut decoded_posting_row_id_count)?;
3998 let subject_type_relation =
3999 decode_subject_type_relation_index(&input, &mut decoded_posting_row_id_count)?;
4000 let subject_type = decode_subject_type_index(&input, &mut decoded_posting_row_id_count)?;
4001 validate_decoded_posting_row_id_count(&posting_row_ids, decoded_posting_row_id_count)?;
4002
4003 Ok(Self {
4004 resource,
4005 resource_object,
4006 resource_type_relation,
4007 resource_type,
4008 subject,
4009 subject_type_relation,
4010 subject_type,
4011 })
4012 }
4013}
4014
4015fn decode_resource_index(
4016 input: &SnapshotIndexDecodeInput<'_>,
4017 decoded_posting_row_id_count: &mut u64,
4018) -> Result<PostingIndex<ResourceIndexKey>, SnapshotIoError> {
4019 decode_index(
4020 input,
4021 &SnapshotIndexDecoder {
4022 kind: SnapshotIndexKind::Resource,
4023 key_from_disk: resource_key_from_disk,
4024 row_matches_key: row_matches_resource_key,
4025 coverage_bit: simple_index_coverage_bit,
4026 expected_mask: |_| 1,
4027 required_by_profile: |_| true,
4028 },
4029 decoded_posting_row_id_count,
4030 )
4031}
4032
4033fn decode_resource_object_index(
4034 input: &SnapshotIndexDecodeInput<'_>,
4035 decoded_posting_row_id_count: &mut u64,
4036) -> Result<PostingIndex<ResourceObjectIndexKey>, SnapshotIoError> {
4037 decode_index(
4038 input,
4039 &SnapshotIndexDecoder {
4040 kind: SnapshotIndexKind::ResourceObject,
4041 key_from_disk: resource_object_key_from_disk,
4042 row_matches_key: row_matches_resource_object_key,
4043 coverage_bit: simple_index_coverage_bit,
4044 expected_mask: |_| 1,
4045 required_by_profile: IndexProfile::supports_broad_resource_indexes,
4046 },
4047 decoded_posting_row_id_count,
4048 )
4049}
4050
4051fn decode_resource_type_relation_index(
4052 input: &SnapshotIndexDecodeInput<'_>,
4053 decoded_posting_row_id_count: &mut u64,
4054) -> Result<PostingIndex<ResourceTypeRelationIndexKey>, SnapshotIoError> {
4055 decode_index(
4056 input,
4057 &SnapshotIndexDecoder {
4058 kind: SnapshotIndexKind::ResourceTypeRelation,
4059 key_from_disk: resource_type_relation_key_from_disk,
4060 row_matches_key: row_matches_resource_type_relation_key,
4061 coverage_bit: simple_index_coverage_bit,
4062 expected_mask: |_| 1,
4063 required_by_profile: IndexProfile::supports_broad_resource_indexes,
4064 },
4065 decoded_posting_row_id_count,
4066 )
4067}
4068
4069fn decode_resource_type_index(
4070 input: &SnapshotIndexDecodeInput<'_>,
4071 decoded_posting_row_id_count: &mut u64,
4072) -> Result<PostingIndex<ObjectTypeId>, SnapshotIoError> {
4073 decode_index(
4074 input,
4075 &SnapshotIndexDecoder {
4076 kind: SnapshotIndexKind::ResourceType,
4077 key_from_disk: resource_type_key_from_disk,
4078 row_matches_key: row_matches_resource_type_key,
4079 coverage_bit: simple_index_coverage_bit,
4080 expected_mask: |_| 1,
4081 required_by_profile: IndexProfile::supports_broad_resource_indexes,
4082 },
4083 decoded_posting_row_id_count,
4084 )
4085}
4086
4087fn decode_subject_index(
4088 input: &SnapshotIndexDecodeInput<'_>,
4089 decoded_posting_row_id_count: &mut u64,
4090) -> Result<PostingIndex<SubjectIndexKey>, SnapshotIoError> {
4091 decode_index(
4092 input,
4093 &SnapshotIndexDecoder {
4094 kind: SnapshotIndexKind::Subject,
4095 key_from_disk: subject_key_from_disk,
4096 row_matches_key: row_matches_subject_key,
4097 coverage_bit: subject_index_coverage_bit,
4098 expected_mask: |row| if row.subject_relation.is_some() { 3 } else { 1 },
4099 required_by_profile: IndexProfile::supports_subject_reverse_lookup,
4100 },
4101 decoded_posting_row_id_count,
4102 )
4103}
4104
4105fn decode_subject_type_relation_index(
4106 input: &SnapshotIndexDecodeInput<'_>,
4107 decoded_posting_row_id_count: &mut u64,
4108) -> Result<PostingIndex<SubjectTypeRelationIndexKey>, SnapshotIoError> {
4109 decode_index(
4110 input,
4111 &SnapshotIndexDecoder {
4112 kind: SnapshotIndexKind::SubjectTypeRelation,
4113 key_from_disk: subject_type_relation_key_from_disk,
4114 row_matches_key: row_matches_subject_type_relation_key,
4115 coverage_bit: simple_index_coverage_bit,
4116 expected_mask: |row| u8::from(row.subject_relation.is_some()),
4117 required_by_profile: IndexProfile::supports_subject_reverse_lookup,
4118 },
4119 decoded_posting_row_id_count,
4120 )
4121}
4122
4123fn decode_subject_type_index(
4124 input: &SnapshotIndexDecodeInput<'_>,
4125 decoded_posting_row_id_count: &mut u64,
4126) -> Result<PostingIndex<SubjectTypeId>, SnapshotIoError> {
4127 decode_index(
4128 input,
4129 &SnapshotIndexDecoder {
4130 kind: SnapshotIndexKind::SubjectType,
4131 key_from_disk: subject_type_key_from_disk,
4132 row_matches_key: row_matches_subject_type_key,
4133 coverage_bit: simple_index_coverage_bit,
4134 expected_mask: |_| 1,
4135 required_by_profile: IndexProfile::supports_subject_reverse_lookup,
4136 },
4137 decoded_posting_row_id_count,
4138 )
4139}
4140
4141#[derive(Debug, Clone, Copy)]
4142struct DiskIndexDirectoryEntry {
4143 kind: SnapshotIndexKind,
4144 encoding: SnapshotIndexEncoding,
4145 key_start: u32,
4146 key_count: u32,
4147 posting_range_start: u32,
4148 posting_range_count: u32,
4149}
4150
4151#[derive(Debug, Clone, Copy)]
4152struct SnapshotIndexDecodeInput<'a> {
4153 directory: &'a [DiskIndexDirectoryEntry],
4154 keys: &'a DecodedIndexKeys,
4155 ranges: &'a [DiskPostingRange],
4156 posting_row_ids: &'a DecodedPostingRowIds,
4157 rows: &'a [RelationshipRow],
4158 row_count: u32,
4159 symbol_count: u32,
4160 index_profile: IndexProfile,
4161 profile: SnapshotLoadProfile,
4162 validation: SnapshotValidationMode,
4163}
4164
4165#[derive(Debug)]
4166enum SnapshotIndexEntries<'a> {
4167 Borrowed {
4168 keys: &'a [DiskIndexKey],
4169 ranges: &'a [DiskPostingRange],
4170 },
4171 Owned(Vec<SnapshotIndexEntry>),
4172}
4173
4174impl SnapshotIndexEntries<'_> {
4175 fn len(&self) -> usize {
4176 match self {
4177 Self::Borrowed { keys, .. } => keys.len(),
4178 Self::Owned(entries) => entries.len(),
4179 }
4180 }
4181}
4182
4183#[derive(Debug, Clone, Copy)]
4184struct SnapshotIndexEntry {
4185 key: DiskIndexKey,
4186 range: DiskPostingRange,
4187}
4188
4189#[derive(Debug)]
4190enum DecodedIndexKeys {
4191 Fixed(Vec<DiskIndexKey>),
4192 Compact { bytes: Vec<u8>, row_count: u64 },
4193}
4194
4195#[derive(Debug)]
4196enum DecodedPostingRowIds {
4197 Fixed(Vec<RowId>),
4198 DeltaVarint { bytes: Vec<u8>, row_count: u64 },
4199}
4200
4201struct SnapshotIndexDecoder<K> {
4202 kind: SnapshotIndexKind,
4203 key_from_disk: fn(DiskIndexKey, u32) -> Result<K, SnapshotIoError>,
4204 row_matches_key: fn(&RelationshipRow, DiskIndexKey) -> bool,
4205 coverage_bit: fn(&RelationshipRow, DiskIndexKey) -> u8,
4206 expected_mask: fn(&RelationshipRow) -> u8,
4207 required_by_profile: fn(IndexProfile) -> bool,
4208}
4209
4210fn decode_index_directory(
4211 reader: &SnapshotReader<'_>,
4212) -> Result<Vec<DiskIndexDirectoryEntry>, SnapshotIoError> {
4213 let section = reader.section(SectionKind::IndexDirectory)?;
4214 if section.row_count() != SNAPSHOT_INDEX_KIND_COUNT_U64 {
4215 return Err(SnapshotIoError::Format {
4216 reason: "index directory row count is invalid",
4217 });
4218 }
4219 let expected_len = checked_mul_usize(SNAPSHOT_INDEX_KIND_COUNT, DISK_INDEX_DIRECTORY_LEN)?;
4220 if section.bytes().len() != expected_len {
4221 return Err(SnapshotIoError::Format {
4222 reason: "index directory length is invalid",
4223 });
4224 }
4225 let mut cursor = BinaryCursor::new(section.bytes());
4226 let mut entries = Vec::with_capacity(SNAPSHOT_INDEX_KIND_COUNT);
4227 let mut seen = HashSet::with_capacity(SNAPSHOT_INDEX_KIND_COUNT);
4228 for _ in 0..SNAPSHOT_INDEX_KIND_COUNT {
4229 let kind = SnapshotIndexKind::from_raw(cursor.read_u16()?)?;
4230 let flags = cursor.read_u16()?;
4231 let encoding = SnapshotIndexEncoding::from_flags(flags)?;
4232 insert_unique(&mut seen, kind, "duplicate snapshot index kind")?;
4233 entries.push(DiskIndexDirectoryEntry {
4234 kind,
4235 encoding,
4236 key_start: cursor.read_u32()?,
4237 key_count: cursor.read_u32()?,
4238 posting_range_start: cursor.read_u32()?,
4239 posting_range_count: cursor.read_u32()?,
4240 });
4241 }
4242 Ok(entries)
4243}
4244
4245fn decode_index_keys(reader: &SnapshotReader<'_>) -> Result<DecodedIndexKeys, SnapshotIoError> {
4246 let section = reader.section(SectionKind::IndexKeys)?;
4247 if reader.header().format_version == SnapshotFormatVersion::V3 {
4248 if section.bytes().is_empty() && section.row_count() != 0 {
4249 return Err(SnapshotIoError::Format {
4250 reason: "index key length does not match row count",
4251 });
4252 }
4253 return Ok(DecodedIndexKeys::Compact {
4254 bytes: section.bytes().to_vec(),
4255 row_count: section.row_count(),
4256 });
4257 }
4258 let row_count = checked_usize_from_u64(section.row_count())?;
4259 let expected_len = checked_mul_usize(row_count, DISK_INDEX_KEY_LEN)?;
4260 if section.bytes().len() != expected_len {
4261 return Err(SnapshotIoError::Format {
4262 reason: "index key length does not match row count",
4263 });
4264 }
4265 let mut cursor = BinaryCursor::new(section.bytes());
4266 let mut keys = Vec::with_capacity(row_count);
4267 for _ in 0..row_count {
4268 keys.push(DiskIndexKey {
4269 first: cursor.read_u32()?,
4270 second: cursor.read_u32()?,
4271 third: cursor.read_u32()?,
4272 });
4273 }
4274 Ok(DecodedIndexKeys::Fixed(keys))
4275}
4276
4277fn decode_posting_ranges(
4278 reader: &SnapshotReader<'_>,
4279) -> Result<Vec<DiskPostingRange>, SnapshotIoError> {
4280 let section = reader.section(SectionKind::PostingRanges)?;
4281 let row_count = checked_usize_from_u64(section.row_count())?;
4282 let expected_len = checked_mul_usize(row_count, DISK_POSTING_RANGE_LEN)?;
4283 if section.bytes().len() != expected_len {
4284 return Err(SnapshotIoError::Format {
4285 reason: "posting range length does not match row count",
4286 });
4287 }
4288 let mut cursor = BinaryCursor::new(section.bytes());
4289 let mut ranges = Vec::with_capacity(row_count);
4290 for _ in 0..row_count {
4291 ranges.push(DiskPostingRange {
4292 first_row_id: cursor.read_u32()?,
4293 overflow_start: cursor.read_u32()?,
4294 overflow_len: cursor.read_u32()?,
4295 });
4296 }
4297 Ok(ranges)
4298}
4299
4300fn decode_posting_row_ids(
4301 reader: &SnapshotReader<'_>,
4302) -> Result<DecodedPostingRowIds, SnapshotIoError> {
4303 let section = reader.section(SectionKind::PostingRowIds)?;
4304 if reader.header().format_version == SnapshotFormatVersion::V3 {
4305 return Ok(DecodedPostingRowIds::DeltaVarint {
4306 bytes: section.bytes().to_vec(),
4307 row_count: section.row_count(),
4308 });
4309 }
4310 let row_count = checked_usize_from_u64(section.row_count())?;
4311 let expected_len = checked_mul_usize(row_count, DISK_ROW_ID_LEN)?;
4312 if section.bytes().len() != expected_len {
4313 return Err(SnapshotIoError::Format {
4314 reason: "posting row id length does not match row count",
4315 });
4316 }
4317 let total_rows = reader.header().relationship_count;
4318 let mut cursor = BinaryCursor::new(section.bytes());
4319 let mut row_ids = Vec::with_capacity(row_count);
4320 for _ in 0..row_count {
4321 row_ids.push(RowId::from_snapshot_raw(cursor.read_u32()?, total_rows)?);
4322 }
4323 Ok(DecodedPostingRowIds::Fixed(row_ids))
4324}
4325
4326fn validate_index_section_layout(
4327 directory: &[DiskIndexDirectoryEntry],
4328 keys: &DecodedIndexKeys,
4329 ranges: &[DiskPostingRange],
4330 posting_row_ids: &DecodedPostingRowIds,
4331 row_count: u32,
4332) -> Result<(), SnapshotIoError> {
4333 match (keys, posting_row_ids) {
4334 (DecodedIndexKeys::Fixed(_), DecodedPostingRowIds::Fixed(_)) => Ok(()),
4335 (
4336 DecodedIndexKeys::Compact {
4337 bytes: key_bytes,
4338 row_count: key_row_count,
4339 },
4340 DecodedPostingRowIds::DeltaVarint {
4341 bytes: posting_bytes,
4342 row_count: _,
4343 },
4344 ) => validate_v3_compact_index_sections(
4345 directory,
4346 key_bytes,
4347 *key_row_count,
4348 ranges,
4349 posting_bytes,
4350 row_count,
4351 ),
4352 _ => Err(SnapshotIoError::Format {
4353 reason: "index section encodings do not match snapshot format",
4354 }),
4355 }
4356}
4357
4358fn validate_v3_compact_index_sections(
4359 directory: &[DiskIndexDirectoryEntry],
4360 key_bytes: &[u8],
4361 key_row_count: u64,
4362 ranges: &[DiskPostingRange],
4363 posting_bytes: &[u8],
4364 row_count: u32,
4365) -> Result<(), SnapshotIoError> {
4366 let mut total_keys = 0_u64;
4367 let mut total_ranges = 0_u64;
4368 let mut key_spans = Vec::with_capacity(directory.len());
4369 let mut range_spans = Vec::with_capacity(directory.len());
4370
4371 for entry in directory {
4372 let SnapshotIndexEncoding::CompactV3 { key_width } = entry.encoding else {
4373 return Err(SnapshotIoError::Format {
4374 reason: "v3 index directory contains non-compact encoding",
4375 });
4376 };
4377 total_keys =
4378 total_keys
4379 .checked_add(u64::from(entry.key_count))
4380 .ok_or(SnapshotIoError::Format {
4381 reason: "index key count overflowed",
4382 })?;
4383 total_ranges = total_ranges
4384 .checked_add(u64::from(entry.posting_range_count))
4385 .ok_or(SnapshotIoError::Format {
4386 reason: "posting range count overflowed",
4387 })?;
4388
4389 let key_span = compact_index_key_span(*entry, key_width)?;
4390 if key_span.end > key_bytes.len() {
4391 return Err(SnapshotIoError::Format {
4392 reason: "compact index key range is out of bounds",
4393 });
4394 }
4395 key_spans.push(key_span);
4396
4397 let range_span = index_range_span(*entry)?;
4398 if range_span.end > ranges.len() {
4399 return Err(SnapshotIoError::Format {
4400 reason: "posting range span is out of bounds",
4401 });
4402 }
4403 range_spans.push(range_span);
4404 }
4405
4406 if total_keys != key_row_count {
4407 return Err(SnapshotIoError::Format {
4408 reason: "index key row count does not match directory",
4409 });
4410 }
4411 if total_ranges
4412 != u64::try_from(ranges.len()).map_err(|_| SnapshotIoError::LimitExceeded {
4413 component: "posting ranges",
4414 })?
4415 {
4416 return Err(SnapshotIoError::Format {
4417 reason: "posting range row count does not match directory",
4418 });
4419 }
4420
4421 validate_spans_cover(
4422 key_spans,
4423 key_bytes.len(),
4424 "compact index key spans do not cover section",
4425 )?;
4426 validate_spans_cover(
4427 range_spans,
4428 ranges.len(),
4429 "posting range spans do not cover section",
4430 )?;
4431
4432 validate_v3_posting_row_id_spans(ranges, posting_bytes, row_count)
4433}
4434
4435fn compact_index_key_span(
4436 entry: DiskIndexDirectoryEntry,
4437 key_width: SnapshotKeyWidth,
4438) -> Result<Range<usize>, SnapshotIoError> {
4439 if entry.posting_range_count > entry.key_count {
4440 return Err(SnapshotIoError::Format {
4441 reason: "posting range count exceeds compact key count",
4442 });
4443 }
4444 let key_start = checked_usize_from_u32(entry.key_start)?;
4445 let key_count = checked_usize_from_u32(entry.key_count)?;
4446 let multi_count = checked_usize_from_u32(entry.posting_range_count)?;
4447 let singleton_count = key_count
4448 .checked_sub(multi_count)
4449 .ok_or(SnapshotIoError::Format {
4450 reason: "compact singleton count underflowed",
4451 })?;
4452 let key_len = checked_mul_usize(entry.kind.key_field_count(), key_width.byte_len())?;
4453 let singleton_entry_len = checked_add_usize(key_len, DISK_ROW_ID_LEN)?;
4454 let singleton_bytes_len = checked_mul_usize(singleton_count, singleton_entry_len)?;
4455 let multi_bytes_len = checked_mul_usize(multi_count, key_len)?;
4456 let group_bytes_len = checked_add_usize(singleton_bytes_len, multi_bytes_len)?;
4457 let key_end = checked_add_usize(key_start, group_bytes_len)?;
4458 Ok(key_start..key_end)
4459}
4460
4461fn index_range_span(entry: DiskIndexDirectoryEntry) -> Result<Range<usize>, SnapshotIoError> {
4462 let range_start = checked_usize_from_u32(entry.posting_range_start)?;
4463 let range_count = checked_usize_from_u32(entry.posting_range_count)?;
4464 let range_end = checked_add_usize(range_start, range_count)?;
4465 Ok(range_start..range_end)
4466}
4467
4468fn validate_spans_cover(
4469 mut spans: Vec<Range<usize>>,
4470 total_len: usize,
4471 reason: &'static str,
4472) -> Result<(), SnapshotIoError> {
4473 spans.sort_by(|left, right| left.start.cmp(&right.start).then(left.end.cmp(&right.end)));
4474 let mut cursor = 0_usize;
4475 for span in spans {
4476 if span.start > span.end || span.start != cursor {
4477 return Err(SnapshotIoError::Format { reason });
4478 }
4479 cursor = span.end;
4480 }
4481 if cursor != total_len {
4482 return Err(SnapshotIoError::Format { reason });
4483 }
4484 Ok(())
4485}
4486
4487fn validate_v3_posting_row_id_spans(
4488 ranges: &[DiskPostingRange],
4489 bytes: &[u8],
4490 row_count: u32,
4491) -> Result<(), SnapshotIoError> {
4492 let mut spans = Vec::new();
4493 for range in ranges {
4494 RowId::from_snapshot_raw(range.first_row_id, row_count)?;
4495 let start = checked_usize_from_u32(range.overflow_start)?;
4496 let len = checked_usize_from_u32(range.overflow_len)?;
4497 let end = checked_add_usize(start, len)?;
4498 if end > bytes.len() {
4499 return Err(SnapshotIoError::Format {
4500 reason: "posting range points outside posting row ids",
4501 });
4502 }
4503 if len == 0 {
4504 continue;
4505 }
4506 spans.push(start..end);
4507 }
4508 validate_spans_cover(
4509 spans,
4510 bytes.len(),
4511 "posting row id spans do not cover section",
4512 )?;
4513 Ok(())
4514}
4515
4516fn validate_decoded_posting_row_id_count(
4517 posting_row_ids: &DecodedPostingRowIds,
4518 decoded_count: u64,
4519) -> Result<(), SnapshotIoError> {
4520 let expected = match posting_row_ids {
4521 DecodedPostingRowIds::Fixed(row_ids) => {
4522 u64::try_from(row_ids.len()).map_err(|_| SnapshotIoError::LimitExceeded {
4523 component: "posting row ids",
4524 })?
4525 }
4526 DecodedPostingRowIds::DeltaVarint { row_count, .. } => *row_count,
4527 };
4528 if decoded_count != expected {
4529 return Err(SnapshotIoError::Format {
4530 reason: "posting row id row count does not match decoded varints",
4531 });
4532 }
4533 Ok(())
4534}
4535
4536fn increment_decoded_posting_row_id_count(counter: &mut u64) -> Result<(), SnapshotIoError> {
4537 *counter = counter.checked_add(1).ok_or(SnapshotIoError::Format {
4538 reason: "posting row id count overflowed",
4539 })?;
4540 Ok(())
4541}
4542
4543fn decode_index<K>(
4544 input: &SnapshotIndexDecodeInput<'_>,
4545 decoder: &SnapshotIndexDecoder<K>,
4546 decoded_posting_row_id_count: &mut u64,
4547) -> Result<PostingIndex<K>, SnapshotIoError>
4548where
4549 K: Copy + Eq + Hash + Ord,
4550{
4551 let entries = snapshot_index_entries(input, decoder.kind)?;
4552 if input.validation == SnapshotValidationMode::TrustedFastLoad
4553 && input.profile == SnapshotLoadProfile::FastLoad
4554 {
4555 return decode_trusted_fast_index(input, decoder, &entries, decoded_posting_row_id_count);
4556 }
4557
4558 let mut coverage = vec![0_u8; input.rows.len()];
4559 let mut sorted_keys = Vec::with_capacity(entries.len());
4560 let mut sorted_ranges = Vec::with_capacity(entries.len());
4561 let mut sorted_overflow = Vec::new();
4562 let mut latency_index = PostingIndex::default();
4563
4564 for (disk_key, range) in snapshot_index_entries_iter(&entries) {
4565 let typed_key = (decoder.key_from_disk)(disk_key, input.symbol_count)?;
4566 let row_ids = posting_row_id_iter(range, input.posting_row_ids, input.row_count)?;
4567 let overflow_start = checked_u32_from_usize(sorted_overflow.len())?;
4568 let mut overflow_len = 0_u32;
4569 let mut first = None;
4570 for row_id in row_ids {
4571 let row_id = row_id?;
4572 let row = input
4573 .rows
4574 .get(row_id.index())
4575 .ok_or(SnapshotIoError::Format {
4576 reason: "posting row id points outside relationship rows",
4577 })?;
4578 if !(decoder.row_matches_key)(row, disk_key) {
4579 return Err(SnapshotIoError::Format {
4580 reason: "posting row does not match index key",
4581 });
4582 }
4583 let bit = (decoder.coverage_bit)(row, disk_key);
4584 if bit == 0 {
4585 return Err(SnapshotIoError::Format {
4586 reason: "index coverage bit is invalid",
4587 });
4588 }
4589 let mask = coverage
4590 .get_mut(row_id.index())
4591 .ok_or(SnapshotIoError::Format {
4592 reason: "index coverage row id is out of bounds",
4593 })?;
4594 if *mask & bit != 0 {
4595 return Err(SnapshotIoError::Format {
4596 reason: "duplicate posting row id in index",
4597 });
4598 }
4599 *mask |= bit;
4600 if first.is_none() {
4601 first = Some(row_id);
4602 } else {
4603 sorted_overflow.push(row_id);
4604 overflow_len = overflow_len.checked_add(1).ok_or(SnapshotIoError::Format {
4605 reason: "posting overflow length overflowed",
4606 })?;
4607 increment_decoded_posting_row_id_count(decoded_posting_row_id_count)?;
4608 }
4609 if matches!(input.profile, SnapshotLoadProfile::Latency) {
4610 latency_index.insert(typed_key, row_id);
4611 }
4612 }
4613 let first_row_id = first.ok_or(SnapshotIoError::Format {
4614 reason: "empty posting range",
4615 })?;
4616 sorted_keys.push(typed_key);
4617 sorted_ranges.push(RuntimePostingRange {
4618 first_row_id,
4619 overflow_start,
4620 overflow_len,
4621 });
4622 }
4623
4624 let required_by_profile = (decoder.required_by_profile)(input.index_profile);
4625 for (row, actual) in input.rows.iter().zip(coverage.iter().copied()) {
4626 let expected = if required_by_profile {
4627 (decoder.expected_mask)(row)
4628 } else {
4629 0
4630 };
4631 if actual != expected {
4632 return Err(SnapshotIoError::Format {
4633 reason: "index does not cover every required row",
4634 });
4635 }
4636 }
4637
4638 match input.profile {
4639 SnapshotLoadProfile::FastLoad => Ok(PostingIndex::from_sorted(
4640 sorted_keys,
4641 sorted_ranges,
4642 sorted_overflow,
4643 )),
4644 SnapshotLoadProfile::Latency => Ok(latency_index),
4645 }
4646}
4647
4648fn decode_trusted_fast_index<K>(
4649 input: &SnapshotIndexDecodeInput<'_>,
4650 decoder: &SnapshotIndexDecoder<K>,
4651 entries: &SnapshotIndexEntries<'_>,
4652 decoded_posting_row_id_count: &mut u64,
4653) -> Result<PostingIndex<K>, SnapshotIoError>
4654where
4655 K: Copy + Eq + Hash + Ord,
4656{
4657 let mut sorted_keys = Vec::with_capacity(entries.len());
4658 let mut sorted_ranges = Vec::with_capacity(entries.len());
4659 let mut sorted_overflow = Vec::new();
4660 for (disk_key, range) in snapshot_index_entries_iter(entries) {
4661 let typed_key = (decoder.key_from_disk)(disk_key, input.symbol_count)?;
4662 let first_row_id = RowId::from_snapshot_raw(range.first_row_id, input.row_count)?;
4663 let overflow_start = checked_u32_from_usize(sorted_overflow.len())?;
4664 let mut overflow_len = 0_u32;
4665 let mut row_ids = posting_row_id_iter(range, input.posting_row_ids, input.row_count)?;
4666 let first = row_ids.next().ok_or(SnapshotIoError::Format {
4667 reason: "empty posting range",
4668 })??;
4669 if first != first_row_id {
4670 return Err(SnapshotIoError::Format {
4671 reason: "posting range first row id mismatch",
4672 });
4673 }
4674 for row_id in row_ids {
4675 sorted_overflow.push(row_id?);
4676 overflow_len = overflow_len.checked_add(1).ok_or(SnapshotIoError::Format {
4677 reason: "posting overflow length overflowed",
4678 })?;
4679 increment_decoded_posting_row_id_count(decoded_posting_row_id_count)?;
4680 }
4681 sorted_keys.push(typed_key);
4682 sorted_ranges.push(RuntimePostingRange {
4683 first_row_id,
4684 overflow_start,
4685 overflow_len,
4686 });
4687 }
4688 Ok(PostingIndex::from_sorted(
4689 sorted_keys,
4690 sorted_ranges,
4691 sorted_overflow,
4692 ))
4693}
4694
4695fn snapshot_index_entries<'a>(
4696 input: &'a SnapshotIndexDecodeInput<'_>,
4697 kind: SnapshotIndexKind,
4698) -> Result<SnapshotIndexEntries<'a>, SnapshotIoError> {
4699 let entry = input
4700 .directory
4701 .iter()
4702 .find(|entry| entry.kind == kind)
4703 .copied()
4704 .ok_or(SnapshotIoError::Format {
4705 reason: "missing snapshot index kind",
4706 })?;
4707 match (input.keys, entry.encoding) {
4708 (DecodedIndexKeys::Fixed(keys), SnapshotIndexEncoding::FixedV2) => {
4709 fixed_snapshot_index_entries(keys, input.ranges, entry)
4710 }
4711 (
4712 DecodedIndexKeys::Compact { bytes: keys, .. },
4713 SnapshotIndexEncoding::CompactV3 { key_width },
4714 ) => compact_snapshot_index_entries(keys, input.ranges, entry, key_width),
4715 _ => Err(SnapshotIoError::Format {
4716 reason: "index encoding does not match snapshot format",
4717 }),
4718 }
4719}
4720
4721fn fixed_snapshot_index_entries<'a>(
4722 keys: &'a [DiskIndexKey],
4723 ranges: &'a [DiskPostingRange],
4724 entry: DiskIndexDirectoryEntry,
4725) -> Result<SnapshotIndexEntries<'a>, SnapshotIoError> {
4726 if entry.key_count != entry.posting_range_count {
4727 return Err(SnapshotIoError::Format {
4728 reason: "index key count does not match posting range count",
4729 });
4730 }
4731 let key_start = checked_usize_from_u32(entry.key_start)?;
4732 let key_count = checked_usize_from_u32(entry.key_count)?;
4733 let key_end = checked_add_usize(key_start, key_count)?;
4734 let range_start = checked_usize_from_u32(entry.posting_range_start)?;
4735 let range_count = checked_usize_from_u32(entry.posting_range_count)?;
4736 let range_end = checked_add_usize(range_start, range_count)?;
4737 let keys = keys
4738 .get(key_start..key_end)
4739 .ok_or(SnapshotIoError::Format {
4740 reason: "index key range is out of bounds",
4741 })?;
4742 let ranges = ranges
4743 .get(range_start..range_end)
4744 .ok_or(SnapshotIoError::Format {
4745 reason: "posting range span is out of bounds",
4746 })?;
4747 validate_sorted_keys(keys)?;
4748 Ok(SnapshotIndexEntries::Borrowed { keys, ranges })
4749}
4750
4751fn compact_snapshot_index_entries<'a>(
4752 keys: &'a [u8],
4753 ranges: &'a [DiskPostingRange],
4754 entry: DiskIndexDirectoryEntry,
4755 key_width: SnapshotKeyWidth,
4756) -> Result<SnapshotIndexEntries<'a>, SnapshotIoError> {
4757 if entry.posting_range_count > entry.key_count {
4758 return Err(SnapshotIoError::Format {
4759 reason: "posting range count exceeds compact key count",
4760 });
4761 }
4762 let key_start = checked_usize_from_u32(entry.key_start)?;
4763 let key_count = checked_usize_from_u32(entry.key_count)?;
4764 let multi_count = checked_usize_from_u32(entry.posting_range_count)?;
4765 let singleton_count = key_count
4766 .checked_sub(multi_count)
4767 .ok_or(SnapshotIoError::Format {
4768 reason: "compact singleton count underflowed",
4769 })?;
4770 let key_len = checked_mul_usize(entry.kind.key_field_count(), key_width.byte_len())?;
4771 let singleton_entry_len = checked_add_usize(key_len, DISK_ROW_ID_LEN)?;
4772 let singleton_bytes_len = checked_mul_usize(singleton_count, singleton_entry_len)?;
4773 let multi_bytes_len = checked_mul_usize(multi_count, key_len)?;
4774 let group_bytes_len = checked_add_usize(singleton_bytes_len, multi_bytes_len)?;
4775 let key_end = checked_add_usize(key_start, group_bytes_len)?;
4776 let group_bytes = keys
4777 .get(key_start..key_end)
4778 .ok_or(SnapshotIoError::Format {
4779 reason: "compact index key range is out of bounds",
4780 })?;
4781 let range_start = checked_usize_from_u32(entry.posting_range_start)?;
4782 let range_end = checked_add_usize(range_start, multi_count)?;
4783 let multi_ranges = ranges
4784 .get(range_start..range_end)
4785 .ok_or(SnapshotIoError::Format {
4786 reason: "posting range span is out of bounds",
4787 })?;
4788 let singleton_bytes =
4789 group_bytes
4790 .get(..singleton_bytes_len)
4791 .ok_or(SnapshotIoError::Format {
4792 reason: "compact singleton keys are out of bounds",
4793 })?;
4794 let multi_key_bytes =
4795 group_bytes
4796 .get(singleton_bytes_len..)
4797 .ok_or(SnapshotIoError::Format {
4798 reason: "compact multi keys are out of bounds",
4799 })?;
4800 let singletons =
4801 decode_compact_singletons(entry.kind, key_width, singleton_bytes, singleton_count)?;
4802 let multis =
4803 decode_compact_multi_entries(entry.kind, key_width, multi_key_bytes, multi_ranges)?;
4804 Ok(SnapshotIndexEntries::Owned(merge_compact_entries(
4805 singletons, multis,
4806 )?))
4807}
4808
4809fn snapshot_index_entries_iter<'a>(
4810 entries: &'a SnapshotIndexEntries<'a>,
4811) -> Box<dyn Iterator<Item = (DiskIndexKey, DiskPostingRange)> + 'a> {
4812 match entries {
4813 SnapshotIndexEntries::Borrowed { keys, ranges } => {
4814 Box::new(keys.iter().copied().zip(ranges.iter().copied()))
4815 }
4816 SnapshotIndexEntries::Owned(entries) => {
4817 Box::new(entries.iter().map(|entry| (entry.key, entry.range)))
4818 }
4819 }
4820}
4821
4822fn decode_compact_singletons(
4823 kind: SnapshotIndexKind,
4824 key_width: SnapshotKeyWidth,
4825 bytes: &[u8],
4826 count: usize,
4827) -> Result<Vec<SnapshotIndexEntry>, SnapshotIoError> {
4828 let mut cursor = BinaryCursor::new(bytes);
4829 let mut entries = Vec::with_capacity(count);
4830 for _ in 0..count {
4831 let key = decode_compact_key(kind, key_width, &mut cursor)?;
4832 let row_id = cursor.read_u32()?;
4833 entries.push(SnapshotIndexEntry {
4834 key,
4835 range: DiskPostingRange {
4836 first_row_id: row_id,
4837 overflow_start: 0,
4838 overflow_len: 0,
4839 },
4840 });
4841 }
4842 validate_sorted_entries(&entries)?;
4843 Ok(entries)
4844}
4845
4846fn decode_compact_multi_entries(
4847 kind: SnapshotIndexKind,
4848 key_width: SnapshotKeyWidth,
4849 bytes: &[u8],
4850 ranges: &[DiskPostingRange],
4851) -> Result<Vec<SnapshotIndexEntry>, SnapshotIoError> {
4852 let mut cursor = BinaryCursor::new(bytes);
4853 let mut entries = Vec::with_capacity(ranges.len());
4854 for range in ranges.iter().copied() {
4855 entries.push(SnapshotIndexEntry {
4856 key: decode_compact_key(kind, key_width, &mut cursor)?,
4857 range,
4858 });
4859 }
4860 validate_sorted_entries(&entries)?;
4861 Ok(entries)
4862}
4863
4864fn decode_compact_key(
4865 kind: SnapshotIndexKind,
4866 key_width: SnapshotKeyWidth,
4867 cursor: &mut BinaryCursor<'_>,
4868) -> Result<DiskIndexKey, SnapshotIoError> {
4869 let first = key_width.read_value(cursor)?;
4870 let second = if kind.key_field_count() >= 2 {
4871 key_width.read_value(cursor)?
4872 } else {
4873 0
4874 };
4875 let third = if kind.key_field_count() >= 3 {
4876 key_width.read_value(cursor)?
4877 } else {
4878 0
4879 };
4880 Ok(DiskIndexKey {
4881 first,
4882 second,
4883 third,
4884 })
4885}
4886
4887fn merge_compact_entries(
4888 singletons: Vec<SnapshotIndexEntry>,
4889 multis: Vec<SnapshotIndexEntry>,
4890) -> Result<Vec<SnapshotIndexEntry>, SnapshotIoError> {
4891 let mut merged = Vec::with_capacity(singletons.len().checked_add(multis.len()).ok_or(
4892 SnapshotIoError::Format {
4893 reason: "compact index entry count overflowed",
4894 },
4895 )?);
4896 let mut singleton_iter = singletons.into_iter().peekable();
4897 let mut multi_iter = multis.into_iter().peekable();
4898 while singleton_iter.peek().is_some() || multi_iter.peek().is_some() {
4899 match (singleton_iter.peek().copied(), multi_iter.peek().copied()) {
4900 (Some(singleton), Some(multi)) if singleton.key < multi.key => {
4901 merged.push(singleton_iter.next().ok_or(SnapshotIoError::Format {
4902 reason: "compact singleton iterator ended unexpectedly",
4903 })?);
4904 }
4905 (Some(singleton), Some(multi)) if singleton.key > multi.key => {
4906 merged.push(multi_iter.next().ok_or(SnapshotIoError::Format {
4907 reason: "compact multi iterator ended unexpectedly",
4908 })?);
4909 }
4910 (Some(_), Some(_)) => {
4911 return Err(SnapshotIoError::Format {
4912 reason: "duplicate compact index key",
4913 });
4914 }
4915 (Some(_), None) => {
4916 merged.push(singleton_iter.next().ok_or(SnapshotIoError::Format {
4917 reason: "compact singleton iterator ended unexpectedly",
4918 })?);
4919 }
4920 (None, Some(_)) => {
4921 merged.push(multi_iter.next().ok_or(SnapshotIoError::Format {
4922 reason: "compact multi iterator ended unexpectedly",
4923 })?);
4924 }
4925 (None, None) => {}
4926 }
4927 }
4928 validate_sorted_entries(&merged)?;
4929 Ok(merged)
4930}
4931
4932fn validate_sorted_entries(entries: &[SnapshotIndexEntry]) -> Result<(), SnapshotIoError> {
4933 if entries.windows(2).any(|window| {
4934 window
4935 .first()
4936 .zip(window.get(1))
4937 .is_some_and(|(left, right)| left.key >= right.key)
4938 }) {
4939 return Err(SnapshotIoError::Format {
4940 reason: "index keys are not strictly sorted",
4941 });
4942 }
4943 Ok(())
4944}
4945
4946fn validate_sorted_keys(keys: &[DiskIndexKey]) -> Result<(), SnapshotIoError> {
4947 if keys.windows(2).any(|window| {
4948 window
4949 .first()
4950 .zip(window.get(1))
4951 .is_some_and(|(left, right)| left >= right)
4952 }) {
4953 return Err(SnapshotIoError::Format {
4954 reason: "index keys are not strictly sorted",
4955 });
4956 }
4957 Ok(())
4958}
4959
4960#[derive(Debug)]
4961enum SnapshotPostingRowIds<'a> {
4962 One {
4963 first: Option<RowId>,
4964 },
4965 FixedMany {
4966 first: Option<RowId>,
4967 rest: std::slice::Iter<'a, RowId>,
4968 previous: RowId,
4969 },
4970 DeltaVarintMany {
4971 first: Option<RowId>,
4972 rest: DeltaVarintPostingIter<'a>,
4973 },
4974}
4975
4976impl Iterator for SnapshotPostingRowIds<'_> {
4977 type Item = Result<RowId, SnapshotIoError>;
4978
4979 fn next(&mut self) -> Option<Self::Item> {
4980 match self {
4981 Self::One { first } => first.take().map(Ok),
4982 Self::FixedMany {
4983 first,
4984 rest,
4985 previous,
4986 } => {
4987 if let Some(row_id) = first.take() {
4988 return Some(Ok(row_id));
4989 }
4990 rest.next().copied().map(|row_id| {
4991 if row_id <= *previous {
4992 return Err(SnapshotIoError::Format {
4993 reason: "posting row ids are not strictly increasing",
4994 });
4995 }
4996 *previous = row_id;
4997 Ok(row_id)
4998 })
4999 }
5000 Self::DeltaVarintMany { first, rest } => first.take().map(Ok).or_else(|| rest.next()),
5001 }
5002 }
5003}
5004
5005#[derive(Debug)]
5006struct DeltaVarintPostingIter<'a> {
5007 bytes: &'a [u8],
5008 offset: usize,
5009 previous: RowId,
5010 row_count: u32,
5011}
5012
5013impl Iterator for DeltaVarintPostingIter<'_> {
5014 type Item = Result<RowId, SnapshotIoError>;
5015
5016 fn next(&mut self) -> Option<Self::Item> {
5017 if self.offset == self.bytes.len() {
5018 return None;
5019 }
5020 match decode_posting_delta_varint(self.bytes, &mut self.offset)
5021 .and_then(|delta| next_delta_row_id(self.previous, delta, self.row_count))
5022 {
5023 Ok(row_id) => {
5024 self.previous = row_id;
5025 Some(Ok(row_id))
5026 }
5027 Err(error) => {
5028 self.offset = self.bytes.len();
5029 Some(Err(error))
5030 }
5031 }
5032 }
5033}
5034
5035fn posting_row_id_iter(
5036 range: DiskPostingRange,
5037 posting_row_ids: &DecodedPostingRowIds,
5038 row_count: u32,
5039) -> Result<SnapshotPostingRowIds<'_>, SnapshotIoError> {
5040 let first = RowId::from_snapshot_raw(range.first_row_id, row_count)?;
5041 if range.overflow_len == 0 {
5042 return Ok(SnapshotPostingRowIds::One { first: Some(first) });
5043 }
5044 match posting_row_ids {
5045 DecodedPostingRowIds::Fixed(row_ids) => {
5046 let start = checked_usize_from_u32(range.overflow_start)?;
5047 let len = checked_usize_from_u32(range.overflow_len)?;
5048 let end = checked_add_usize(start, len)?;
5049 let overflow = row_ids.get(start..end).ok_or(SnapshotIoError::Format {
5050 reason: "posting range points outside posting row ids",
5051 })?;
5052 Ok(SnapshotPostingRowIds::FixedMany {
5053 first: Some(first),
5054 rest: overflow.iter(),
5055 previous: first,
5056 })
5057 }
5058 DecodedPostingRowIds::DeltaVarint { bytes, .. } => {
5059 let start = checked_usize_from_u32(range.overflow_start)?;
5060 let len = checked_usize_from_u32(range.overflow_len)?;
5061 let end = checked_add_usize(start, len)?;
5062 let overflow = bytes.get(start..end).ok_or(SnapshotIoError::Format {
5063 reason: "posting range points outside posting row ids",
5064 })?;
5065 Ok(SnapshotPostingRowIds::DeltaVarintMany {
5066 first: Some(first),
5067 rest: DeltaVarintPostingIter {
5068 bytes: overflow,
5069 offset: 0,
5070 previous: first,
5071 row_count,
5072 },
5073 })
5074 }
5075 }
5076}
5077
5078fn decode_posting_delta_varint(bytes: &[u8], offset: &mut usize) -> Result<u32, SnapshotIoError> {
5079 let mut value = 0_u32;
5080 let mut shift = 0_u32;
5081 loop {
5082 let byte = bytes.get(*offset).copied().ok_or(SnapshotIoError::Format {
5083 reason: "posting row id delta varint is truncated",
5084 })?;
5085 *offset = (*offset).checked_add(1).ok_or(SnapshotIoError::Format {
5086 reason: "posting row id delta offset overflowed",
5087 })?;
5088 let low_bits = u32::from(byte & 0x7F);
5089 if shift == 28 && low_bits > 0x0F {
5090 return Err(SnapshotIoError::Format {
5091 reason: "posting row id delta varint overflows u32",
5092 });
5093 }
5094 value |= low_bits.checked_shl(shift).ok_or(SnapshotIoError::Format {
5095 reason: "posting row id delta varint overflows u32",
5096 })?;
5097 if byte & 0x80 == 0 {
5098 if value == 0 {
5099 return Err(SnapshotIoError::Format {
5100 reason: "posting row id delta must be non-zero",
5101 });
5102 }
5103 return Ok(value);
5104 }
5105 shift = shift.checked_add(7).ok_or(SnapshotIoError::Format {
5106 reason: "posting row id delta varint overflows u32",
5107 })?;
5108 if shift > 28 {
5109 return Err(SnapshotIoError::Format {
5110 reason: "posting row id delta varint is too long",
5111 });
5112 }
5113 }
5114}
5115
5116fn next_delta_row_id(
5117 previous: RowId,
5118 delta: u32,
5119 row_count: u32,
5120) -> Result<RowId, SnapshotIoError> {
5121 let row_id = previous
5122 .raw()
5123 .checked_add(delta)
5124 .ok_or(SnapshotIoError::Format {
5125 reason: "posting row id delta overflowed",
5126 })?;
5127 RowId::from_snapshot_raw(row_id, row_count)
5128}
5129
5130fn resource_key_from_disk(
5131 key: DiskIndexKey,
5132 symbol_count: u32,
5133) -> Result<ResourceIndexKey, SnapshotIoError> {
5134 Ok(ResourceIndexKey {
5135 object_type: ObjectTypeId(SymbolId::from_snapshot_raw(key.first, symbol_count)?),
5136 object_id: ObjectIdId(SymbolId::from_snapshot_raw(key.second, symbol_count)?),
5137 relation: RelationId(SymbolId::from_snapshot_raw(key.third, symbol_count)?),
5138 })
5139}
5140
5141fn resource_object_key_from_disk(
5142 key: DiskIndexKey,
5143 symbol_count: u32,
5144) -> Result<ResourceObjectIndexKey, SnapshotIoError> {
5145 ensure_zero(
5146 key.third,
5147 "resource object index third key field must be zero",
5148 )?;
5149 Ok(ResourceObjectIndexKey {
5150 object_type: ObjectTypeId(SymbolId::from_snapshot_raw(key.first, symbol_count)?),
5151 object_id: ObjectIdId(SymbolId::from_snapshot_raw(key.second, symbol_count)?),
5152 })
5153}
5154
5155fn resource_type_relation_key_from_disk(
5156 key: DiskIndexKey,
5157 symbol_count: u32,
5158) -> Result<ResourceTypeRelationIndexKey, SnapshotIoError> {
5159 ensure_zero(
5160 key.third,
5161 "resource type relation index third key field must be zero",
5162 )?;
5163 Ok(ResourceTypeRelationIndexKey {
5164 object_type: ObjectTypeId(SymbolId::from_snapshot_raw(key.first, symbol_count)?),
5165 relation: RelationId(SymbolId::from_snapshot_raw(key.second, symbol_count)?),
5166 })
5167}
5168
5169fn resource_type_key_from_disk(
5170 key: DiskIndexKey,
5171 symbol_count: u32,
5172) -> Result<ObjectTypeId, SnapshotIoError> {
5173 ensure_zero(
5174 key.second,
5175 "resource type index second key field must be zero",
5176 )?;
5177 ensure_zero(
5178 key.third,
5179 "resource type index third key field must be zero",
5180 )?;
5181 Ok(ObjectTypeId(SymbolId::from_snapshot_raw(
5182 key.first,
5183 symbol_count,
5184 )?))
5185}
5186
5187fn subject_key_from_disk(
5188 key: DiskIndexKey,
5189 symbol_count: u32,
5190) -> Result<SubjectIndexKey, SnapshotIoError> {
5191 let relation = if key.third == 0 {
5192 None
5193 } else {
5194 Some(RelationId(SymbolId::from_snapshot_raw(
5195 key.third,
5196 symbol_count,
5197 )?))
5198 };
5199 Ok(SubjectIndexKey {
5200 subject_type: SubjectTypeId(SymbolId::from_snapshot_raw(key.first, symbol_count)?),
5201 subject_id: SubjectIdId(SymbolId::from_snapshot_raw(key.second, symbol_count)?),
5202 relation,
5203 })
5204}
5205
5206fn subject_type_relation_key_from_disk(
5207 key: DiskIndexKey,
5208 symbol_count: u32,
5209) -> Result<SubjectTypeRelationIndexKey, SnapshotIoError> {
5210 ensure_zero(
5211 key.third,
5212 "subject type relation index third key field must be zero",
5213 )?;
5214 Ok(SubjectTypeRelationIndexKey {
5215 subject_type: SubjectTypeId(SymbolId::from_snapshot_raw(key.first, symbol_count)?),
5216 relation: RelationId(SymbolId::from_snapshot_raw(key.second, symbol_count)?),
5217 })
5218}
5219
5220fn subject_type_key_from_disk(
5221 key: DiskIndexKey,
5222 symbol_count: u32,
5223) -> Result<SubjectTypeId, SnapshotIoError> {
5224 ensure_zero(
5225 key.second,
5226 "subject type index second key field must be zero",
5227 )?;
5228 ensure_zero(key.third, "subject type index third key field must be zero")?;
5229 Ok(SubjectTypeId(SymbolId::from_snapshot_raw(
5230 key.first,
5231 symbol_count,
5232 )?))
5233}
5234
5235fn ensure_zero(value: u32, reason: &'static str) -> Result<(), SnapshotIoError> {
5236 if value == 0 {
5237 Ok(())
5238 } else {
5239 Err(SnapshotIoError::Format { reason })
5240 }
5241}
5242
5243fn row_matches_resource_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5244 row.resource_type.0.get() == key.first
5245 && row.resource_id.0.get() == key.second
5246 && row.relation.0.get() == key.third
5247}
5248
5249fn row_matches_resource_object_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5250 row.resource_type.0.get() == key.first
5251 && row.resource_id.0.get() == key.second
5252 && key.third == 0
5253}
5254
5255fn row_matches_resource_type_relation_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5256 row.resource_type.0.get() == key.first && row.relation.0.get() == key.second && key.third == 0
5257}
5258
5259fn row_matches_resource_type_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5260 row.resource_type.0.get() == key.first && key.second == 0 && key.third == 0
5261}
5262
5263fn row_matches_subject_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5264 row.subject_type.0.get() == key.first
5265 && row.subject_id.0.get() == key.second
5266 && row.subject_relation.map_or(key.third == 0, |relation| {
5267 key.third == 0 || relation.0.get() == key.third
5268 })
5269}
5270
5271fn row_matches_subject_type_relation_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5272 row.subject_type.0.get() == key.first
5273 && row
5274 .subject_relation
5275 .is_some_and(|relation| relation.0.get() == key.second)
5276 && key.third == 0
5277}
5278
5279fn row_matches_subject_type_key(row: &RelationshipRow, key: DiskIndexKey) -> bool {
5280 row.subject_type.0.get() == key.first && key.second == 0 && key.third == 0
5281}
5282
5283fn simple_index_coverage_bit(_row: &RelationshipRow, _key: DiskIndexKey) -> u8 {
5284 1
5285}
5286
5287fn subject_index_coverage_bit(row: &RelationshipRow, key: DiskIndexKey) -> u8 {
5288 match (row.subject_relation, key.third) {
5289 (_, 0) => 1,
5290 (Some(relation), value) if relation.0.get() == value => 2,
5291 _ => 0,
5292 }
5293}
5294
5295#[derive(Debug)]
5296struct DecodedSnapshotRows {
5297 rows: Vec<RelationshipRow>,
5298 live_rows: LiveRows,
5299 uniqueness: UniquenessState,
5300}
5301
5302fn record_relationship_decode_phase(
5303 timings: &mut Option<&mut SnapshotLoadPhaseTimings>,
5304 record: impl FnOnce(&mut SnapshotLoadPhaseTimings, std::time::Duration),
5305 phase_start: Instant,
5306) {
5307 if let Some(timings) = timings.as_deref_mut() {
5308 record(timings, phase_start.elapsed());
5309 }
5310}
5311
5312fn decode_snapshot_rows(
5313 reader: &SnapshotReader<'_>,
5314 interner: &IdentifierInterner,
5315 validation: SnapshotValidationMode,
5316) -> Result<DecodedSnapshotRows, SnapshotIoError> {
5317 let header = reader.header();
5318 let section = reader.section(SectionKind::RelationshipRows)?;
5319 let row_count = checked_usize_from_u32(header.relationship_count)?;
5320 if section.row_count() != u64::from(header.relationship_count) {
5321 return Err(SnapshotIoError::Format {
5322 reason: "relationship row count does not match header",
5323 });
5324 }
5325 let symbol_width = if header.format_version == SnapshotFormatVersion::V3 {
5326 section_width_from_flags(section.flags())?
5327 } else {
5328 if section.flags() != 0 {
5329 return Err(SnapshotIoError::Format {
5330 reason: "relationship row flags are unsupported",
5331 });
5332 }
5333 SnapshotKeyWidth::U32
5334 };
5335 let expected_len =
5336 checked_mul_usize(row_count, checked_mul_usize(symbol_width.byte_len(), 6)?)?;
5337 if section.bytes().len() != expected_len {
5338 return Err(SnapshotIoError::Format {
5339 reason: "relationship row length does not match row count",
5340 });
5341 }
5342 let row_byte_len = checked_mul_usize(symbol_width.byte_len(), 6)?;
5343 let mut rows = Vec::with_capacity(row_count);
5344 let mut duplicate_detector = RelationshipIdentityIndex::default();
5345 let validate_semantics = validation == SnapshotValidationMode::Full;
5346 for (index, row_bytes) in section.bytes().chunks_exact(row_byte_len).enumerate() {
5347 let row_id = RowId::from_len(index)?;
5348 let fields = decode_snapshot_row_fields(row_bytes, symbol_width)?;
5349 let [
5350 resource_type,
5351 resource_id,
5352 relation,
5353 subject_type,
5354 subject_id,
5355 subject_relation,
5356 ] = fields;
5357 let row = RelationshipRow {
5358 row_id,
5359 resource_type: ObjectTypeId(SymbolId::from_snapshot_raw(
5360 resource_type,
5361 header.symbol_count,
5362 )?),
5363 resource_id: ObjectIdId(SymbolId::from_snapshot_raw(
5364 resource_id,
5365 header.symbol_count,
5366 )?),
5367 relation: RelationId(SymbolId::from_snapshot_raw(relation, header.symbol_count)?),
5368 subject_type: SubjectTypeId(SymbolId::from_snapshot_raw(
5369 subject_type,
5370 header.symbol_count,
5371 )?),
5372 subject_id: SubjectIdId(SymbolId::from_snapshot_raw(
5373 subject_id,
5374 header.symbol_count,
5375 )?),
5376 subject_relation: match subject_relation {
5377 0 => None,
5378 value => Some(RelationId(SymbolId::from_snapshot_raw(
5379 value,
5380 header.symbol_count,
5381 )?)),
5382 },
5383 };
5384 if validate_semantics {
5385 validate_row_domains(interner, &row)?;
5386 if duplicate_detector.find(&rows, &row).is_some() {
5387 return Err(SnapshotIoError::Format {
5388 reason: "duplicate relationship row in snapshot",
5389 });
5390 }
5391 duplicate_detector.insert(&rows, row_id, &row);
5392 }
5393 rows.push(row);
5394 }
5395 let uniqueness = if validate_semantics {
5396 UniquenessState::KnownUniqueButNotIndexed
5397 } else {
5398 UniquenessState::UntrustedNotIndexed
5399 };
5400 Ok(DecodedSnapshotRows {
5401 rows,
5402 live_rows: LiveRows::full(row_count),
5403 uniqueness,
5404 })
5405}
5406
5407fn decode_snapshot_row_fields(
5408 row_bytes: &[u8],
5409 symbol_width: SnapshotKeyWidth,
5410) -> Result<[u32; 6], SnapshotIoError> {
5411 match symbol_width {
5412 SnapshotKeyWidth::U8 => decode_snapshot_row_fields_u8(row_bytes),
5413 SnapshotKeyWidth::U16 => decode_snapshot_row_fields_u16(row_bytes),
5414 SnapshotKeyWidth::U24 => decode_snapshot_row_fields_u24(row_bytes),
5415 SnapshotKeyWidth::U32 => decode_snapshot_row_fields_u32(row_bytes),
5416 }
5417}
5418
5419fn decode_snapshot_row_fields_u8(row_bytes: &[u8]) -> Result<[u32; 6], SnapshotIoError> {
5420 let [first, second, third, fourth, fifth, sixth] = row_bytes_to_array(row_bytes)?;
5421 Ok([
5422 u32::from(first),
5423 u32::from(second),
5424 u32::from(third),
5425 u32::from(fourth),
5426 u32::from(fifth),
5427 u32::from(sixth),
5428 ])
5429}
5430
5431fn decode_snapshot_row_fields_u16(row_bytes: &[u8]) -> Result<[u32; 6], SnapshotIoError> {
5432 let [
5433 first_a,
5434 first_b,
5435 second_a,
5436 second_b,
5437 third_a,
5438 third_b,
5439 fourth_a,
5440 fourth_b,
5441 fifth_a,
5442 fifth_b,
5443 sixth_a,
5444 sixth_b,
5445 ] = row_bytes_to_array(row_bytes)?;
5446 Ok([
5447 u32::from(u16::from_le_bytes([first_a, first_b])),
5448 u32::from(u16::from_le_bytes([second_a, second_b])),
5449 u32::from(u16::from_le_bytes([third_a, third_b])),
5450 u32::from(u16::from_le_bytes([fourth_a, fourth_b])),
5451 u32::from(u16::from_le_bytes([fifth_a, fifth_b])),
5452 u32::from(u16::from_le_bytes([sixth_a, sixth_b])),
5453 ])
5454}
5455
5456fn decode_snapshot_row_fields_u24(row_bytes: &[u8]) -> Result<[u32; 6], SnapshotIoError> {
5457 let [
5458 first_a,
5459 first_b,
5460 first_c,
5461 second_a,
5462 second_b,
5463 second_c,
5464 third_a,
5465 third_b,
5466 third_c,
5467 fourth_a,
5468 fourth_b,
5469 fourth_c,
5470 fifth_a,
5471 fifth_b,
5472 fifth_c,
5473 sixth_a,
5474 sixth_b,
5475 sixth_c,
5476 ] = row_bytes_to_array(row_bytes)?;
5477 Ok([
5478 u32::from_le_bytes([first_a, first_b, first_c, 0]),
5479 u32::from_le_bytes([second_a, second_b, second_c, 0]),
5480 u32::from_le_bytes([third_a, third_b, third_c, 0]),
5481 u32::from_le_bytes([fourth_a, fourth_b, fourth_c, 0]),
5482 u32::from_le_bytes([fifth_a, fifth_b, fifth_c, 0]),
5483 u32::from_le_bytes([sixth_a, sixth_b, sixth_c, 0]),
5484 ])
5485}
5486
5487fn decode_snapshot_row_fields_u32(row_bytes: &[u8]) -> Result<[u32; 6], SnapshotIoError> {
5488 let [
5489 first_a,
5490 first_b,
5491 first_c,
5492 first_d,
5493 second_a,
5494 second_b,
5495 second_c,
5496 second_d,
5497 third_a,
5498 third_b,
5499 third_c,
5500 third_d,
5501 fourth_a,
5502 fourth_b,
5503 fourth_c,
5504 fourth_d,
5505 fifth_a,
5506 fifth_b,
5507 fifth_c,
5508 fifth_d,
5509 sixth_a,
5510 sixth_b,
5511 sixth_c,
5512 sixth_d,
5513 ] = row_bytes_to_array(row_bytes)?;
5514 Ok([
5515 u32::from_le_bytes([first_a, first_b, first_c, first_d]),
5516 u32::from_le_bytes([second_a, second_b, second_c, second_d]),
5517 u32::from_le_bytes([third_a, third_b, third_c, third_d]),
5518 u32::from_le_bytes([fourth_a, fourth_b, fourth_c, fourth_d]),
5519 u32::from_le_bytes([fifth_a, fifth_b, fifth_c, fifth_d]),
5520 u32::from_le_bytes([sixth_a, sixth_b, sixth_c, sixth_d]),
5521 ])
5522}
5523
5524fn row_bytes_to_array<const LEN: usize>(row_bytes: &[u8]) -> Result<[u8; LEN], SnapshotIoError> {
5525 row_bytes.try_into().map_err(|_| SnapshotIoError::Format {
5526 reason: "relationship row width is invalid",
5527 })
5528}
5529
5530fn validate_row_domains(
5531 interner: &IdentifierInterner,
5532 row: &RelationshipRow,
5533) -> Result<(), SnapshotIoError> {
5534 ObjectType::try_from(interner.resolve(row.resource_type.0)?)?;
5535 ObjectId::try_from(interner.resolve(row.resource_id.0)?)?;
5536 RelationName::try_from(interner.resolve(row.relation.0)?)?;
5537 SubjectType::try_from(interner.resolve(row.subject_type.0)?)?;
5538 SubjectId::try_from(interner.resolve(row.subject_id.0)?)?;
5539 if let Some(relation) = row.subject_relation {
5540 RelationName::try_from(interner.resolve(relation.0)?)?;
5541 }
5542 Ok(())
5543}
5544
5545#[derive(Debug, Clone)]
5546struct LiveRows {
5547 all_live_len: Option<usize>,
5548 words: Vec<u64>,
5549}
5550
5551impl Default for LiveRows {
5552 fn default() -> Self {
5553 Self {
5554 all_live_len: Some(0),
5555 words: Vec::new(),
5556 }
5557 }
5558}
5559
5560impl LiveRows {
5561 fn full(len: usize) -> Self {
5562 Self {
5563 all_live_len: Some(len),
5564 words: Vec::new(),
5565 }
5566 }
5567
5568 fn sparse_full(len: usize) -> Self {
5569 let word_count = len.div_ceil(u64::BITS as usize);
5570 let mut words = vec![u64::MAX; word_count];
5571 let remainder = len % u64::BITS as usize;
5572 if remainder != 0
5573 && let Some(last) = words.last_mut()
5574 {
5575 *last = (1_u64 << remainder) - 1;
5576 }
5577 Self {
5578 all_live_len: None,
5579 words,
5580 }
5581 }
5582
5583 fn insert(&mut self, row_id: RowId) {
5584 let index = row_id.index();
5585 if let Some(len) = self.all_live_len {
5586 if index == len {
5587 self.all_live_len = len.checked_add(1);
5588 return;
5589 }
5590 if index < len {
5591 return;
5592 }
5593 *self = Self::sparse_full(len);
5594 }
5595 let word = index / u64::BITS as usize;
5596 if word >= self.words.len() {
5597 self.words.resize(word.saturating_add(1), 0);
5598 }
5599 if let Some(value) = self.words.get_mut(word) {
5600 *value |= 1_u64 << (index % u64::BITS as usize);
5601 }
5602 }
5603
5604 fn remove(&mut self, row_id: RowId) {
5605 let index = row_id.index();
5606 if let Some(len) = self.all_live_len {
5607 *self = Self::sparse_full(len);
5608 }
5609 let word = index / u64::BITS as usize;
5610 if let Some(value) = self.words.get_mut(word) {
5611 *value &= !(1_u64 << (index % u64::BITS as usize));
5612 }
5613 }
5614
5615 fn contains(&self, row_id: RowId) -> bool {
5616 let index = row_id.index();
5617 if let Some(len) = self.all_live_len {
5618 return index < len;
5619 }
5620 let word = index / u64::BITS as usize;
5621 self.words
5622 .get(word)
5623 .is_some_and(|value| value & (1_u64 << (index % u64::BITS as usize)) != 0)
5624 }
5625
5626 fn is_all_live(&self) -> bool {
5627 self.all_live_len.is_some()
5628 }
5629}
5630
5631fn compaction_dead_row_threshold() -> usize {
5632 if cfg!(test) { 16 } else { COMPACT_DEAD_ROWS }
5633}
5634
5635fn relationship_identifier_values(relationship: &Relationship, mut visit: impl FnMut(&str)) {
5636 visit(relationship.resource().object_type().as_str());
5637 visit(relationship.resource().object_id().as_str());
5638 visit(relationship.relation().as_str());
5639 match relationship.subject() {
5640 SubjectRef::Object(object) => {
5641 visit(object.object_type().as_str());
5642 visit(object.object_id().as_str());
5643 }
5644 SubjectRef::Userset { object, relation } => {
5645 visit(object.object_type().as_str());
5646 visit(object.object_id().as_str());
5647 visit(relation.as_str());
5648 }
5649 }
5650}
5651
5652fn hash_value<T: Hash + ?Sized>(value: &T) -> u64 {
5653 let mut state = DefaultHasher::new();
5654 value.hash(&mut state);
5655 state.finish()
5656}