1use std::collections::{BTreeMap, HashMap, HashSet};
32use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
33use std::time::{SystemTime, UNIX_EPOCH};
34
35use super::entity::{CrossRef, EntityData, EntityId, EntityKind, RefType, UnifiedEntity};
36use super::metadata::{Metadata, MetadataStorage};
37use super::segment_graph_index::SegmentGraphIndex;
38use super::segment_key_lookup::SegmentKeyLookup;
39use super::segment_unique_key::SegmentUniqueKey;
40use crate::storage::query::value_compare::partial_compare_values;
41use reddb_types::{value_to_canonical_key, CanonicalKey, Value};
42
43pub type SegmentId = u64;
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum SegmentState {
49 Growing,
51 Sealing,
53 Sealed,
55 Flushed,
57 Archived,
59}
60
61impl SegmentState {
62 pub fn is_writable(&self) -> bool {
64 matches!(self, Self::Growing)
65 }
66
67 pub fn is_queryable(&self) -> bool {
69 !matches!(self, Self::Sealing)
70 }
71
72 pub fn is_immutable(&self) -> bool {
74 matches!(self, Self::Sealed | Self::Flushed | Self::Archived)
75 }
76}
77
78#[derive(Debug, Clone)]
80pub struct SegmentConfig {
81 pub max_entities: usize,
83 pub max_bytes: usize,
85 pub max_age_secs: u64,
87 pub build_vector_index: bool,
89 pub build_graph_index: bool,
91 pub compression_level: u8,
93}
94
95impl Default for SegmentConfig {
96 fn default() -> Self {
97 Self {
98 max_entities: 100_000,
99 max_bytes: 256 * 1024 * 1024, max_age_secs: 3600, build_vector_index: true,
102 build_graph_index: true,
103 compression_level: 6,
104 }
105 }
106}
107
108#[derive(Debug, Clone, Default)]
110pub struct SegmentStats {
111 pub entity_count: usize,
113 pub deleted_count: usize,
115 pub memory_bytes: usize,
117 pub vector_count: usize,
119 pub node_count: usize,
121 pub edge_count: usize,
123 pub row_count: usize,
125 pub cross_ref_count: usize,
127}
128
129#[derive(Debug, Clone)]
131pub enum SegmentError {
132 NotWritable,
134 NotFound(EntityId),
136 AlreadyExists(EntityId),
138 Full,
140 InvalidState(SegmentState),
142 Internal(String),
144}
145
146impl std::fmt::Display for SegmentError {
147 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
148 match self {
149 Self::NotWritable => write!(f, "segment is not writable"),
150 Self::NotFound(id) => write!(f, "entity not found: {}", id),
151 Self::AlreadyExists(id) => write!(f, "entity already exists: {}", id),
152 Self::Full => write!(f, "segment is full"),
153 Self::InvalidState(state) => write!(f, "invalid operation for state: {:?}", state),
154 Self::Internal(msg) => write!(f, "internal error: {}", msg),
155 }
156 }
157}
158
159impl std::error::Error for SegmentError {}
160
161fn current_unix_secs() -> u64 {
162 SystemTime::now()
163 .duration_since(UNIX_EPOCH)
164 .unwrap_or_default()
165 .as_secs()
166}
167
168fn entity_id_from_probe_key(key: &[u8]) -> Option<EntityId> {
169 if let Ok(text) = std::str::from_utf8(key) {
170 if let Ok(raw) = text.parse::<u64>() {
171 return Some(EntityId::new(raw));
172 }
173 }
174
175 let bytes: [u8; 8] = key.try_into().ok()?;
176 Some(EntityId::new(u64::from_le_bytes(bytes)))
177}
178
179const SEALED_MULTI_ZONE_MAX_INTERVALS: usize = 4;
180
181const TOMBSTONE_ENTRY_BYTES: u64 = 24;
189
190#[derive(Debug, Clone)]
191struct UpdateIndexSnapshot {
192 pk_column_name: Option<String>,
193 pk_value: Option<Value>,
194 pk_index_key: Option<(String, String)>,
195 cross_refs: Vec<CrossRef>,
196}
197
198impl UpdateIndexSnapshot {
199 fn from_entity(entity: &UnifiedEntity) -> Self {
200 let (pk_column_name, pk_value) = match &entity.data {
201 EntityData::Row(row) => (
202 row.schema
203 .as_deref()
204 .and_then(|schema| schema.first().cloned()),
205 row.columns.first().cloned(),
206 ),
207 _ => (None, None),
208 };
209 let pk_index_key = pk_value
210 .as_ref()
211 .map(|value| (entity.kind.collection().to_string(), format!("{:?}", value)));
212 Self {
213 pk_column_name,
214 pk_value,
215 pk_index_key,
216 cross_refs: entity.cross_refs().to_vec(),
217 }
218 }
219}
220
221pub trait UnifiedSegment: Send + Sync {
223 fn id(&self) -> SegmentId;
225
226 fn state(&self) -> SegmentState;
228
229 fn collection(&self) -> &str;
231
232 fn stats(&self) -> SegmentStats;
234
235 fn entity_count(&self) -> usize;
237
238 fn contains(&self, id: EntityId) -> bool;
240
241 fn get(&self, id: EntityId) -> Option<&UnifiedEntity>;
243
244 fn get_mut(&mut self, id: EntityId) -> Option<&mut UnifiedEntity>;
246
247 fn insert(&mut self, entity: UnifiedEntity) -> Result<EntityId, SegmentError>;
249
250 fn update(&mut self, entity: UnifiedEntity) -> Result<(), SegmentError>;
252
253 fn update_hot(
257 &mut self,
258 entity: UnifiedEntity,
259 modified_columns: &[String],
260 ) -> Result<(), SegmentError> {
261 let _ = modified_columns;
262 self.update(entity)
263 }
264
265 fn delete(&mut self, id: EntityId) -> Result<bool, SegmentError>;
267
268 fn get_metadata(&self, id: EntityId) -> Option<Metadata>;
270
271 fn set_metadata(&mut self, id: EntityId, metadata: Metadata) -> Result<(), SegmentError>;
273
274 fn seal(&mut self) -> Result<(), SegmentError>;
276
277 fn should_seal(&self, config: &SegmentConfig) -> bool;
279
280 fn iter(&self) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_>;
282
283 fn iter_kind(&self, kind_filter: &str) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_>;
285
286 fn filter_metadata(
288 &self,
289 filters: &[(String, super::metadata::MetadataFilter)],
290 ) -> Vec<EntityId>;
291}
292
293#[derive(Debug, Clone)]
300pub struct ColZone {
301 pub min: Value,
302 pub max: Value,
303 min_key: Option<CanonicalKey>,
304 max_key: Option<CanonicalKey>,
305}
306
307impl ColZone {
308 fn payload_bytes(&self) -> usize {
309 super::memory_size::value_heap_bytes(&self.min)
310 .saturating_add(super::memory_size::value_heap_bytes(&self.max))
311 .saturating_add(
312 self.min_key
313 .as_ref()
314 .map_or(0, super::memory_size::canonical_heap_bytes),
315 )
316 .saturating_add(
317 self.max_key
318 .as_ref()
319 .map_or(0, super::memory_size::canonical_heap_bytes),
320 )
321 }
322
323 fn new(v: Value) -> Self {
324 Self {
325 min_key: value_to_canonical_key(&v),
326 max_key: value_to_canonical_key(&v),
327 min: v.clone(),
328 max: v,
329 }
330 }
331
332 fn with_bounds(min: Value, max: Value) -> Self {
333 Self {
334 min_key: value_to_canonical_key(&min),
335 max_key: value_to_canonical_key(&max),
336 min,
337 max,
338 }
339 }
340
341 fn update(&mut self, v: &Value) {
342 if compare_zone_values(v, None, &self.min, self.min_key.as_ref())
343 .map(|o| o == std::cmp::Ordering::Less)
344 .unwrap_or(false)
345 {
346 self.min = v.clone();
347 self.min_key = value_to_canonical_key(v);
348 }
349 if compare_zone_values(v, None, &self.max, self.max_key.as_ref())
350 .map(|o| o == std::cmp::Ordering::Greater)
351 .unwrap_or(false)
352 {
353 self.max = v.clone();
354 self.max_key = value_to_canonical_key(v);
355 }
356 }
357}
358
359#[derive(Debug, Clone, Default)]
360pub struct MultiColZone {
361 pub intervals: Vec<ColZone>,
362}
363
364impl MultiColZone {
365 fn can_skip(&self, pred: &ZoneColPred<'_>) -> bool {
366 !self.intervals.is_empty() && self.intervals.iter().all(|zone| pred.can_skip(zone))
367 }
368}
369
370fn compare_zone_values(
371 left: &Value,
372 left_key: Option<&CanonicalKey>,
373 right: &Value,
374 right_key: Option<&CanonicalKey>,
375) -> Option<std::cmp::Ordering> {
376 partial_compare_values(left, right).or_else(|| {
377 let left_key = left_key.cloned().or_else(|| value_to_canonical_key(left))?;
378 let right_key = right_key
379 .cloned()
380 .or_else(|| value_to_canonical_key(right))?;
381 (left_key.family() == right_key.family()).then(|| left_key.cmp(&right_key))
382 })
383}
384
385#[derive(Debug, Clone, Copy, PartialEq, Eq)]
387pub enum ZoneColPredKind {
388 Eq,
389 Gt,
390 Gte,
391 Lt,
392 Lte,
393}
394
395#[derive(Debug, Clone)]
397pub enum ZoneColPred<'a> {
398 Eq(&'a Value),
399 Gt(&'a Value),
400 Gte(&'a Value),
401 Lt(&'a Value),
402 Lte(&'a Value),
403}
404
405impl<'a> ZoneColPred<'a> {
406 pub fn can_skip(&self, zone: &ColZone) -> bool {
408 match self {
409 ZoneColPred::Eq(val) => {
411 compare_zone_values(val, None, &zone.min, zone.min_key.as_ref())
412 .map(|o| o == std::cmp::Ordering::Less)
413 .unwrap_or(false)
414 || compare_zone_values(val, None, &zone.max, zone.max_key.as_ref())
415 .map(|o| o == std::cmp::Ordering::Greater)
416 .unwrap_or(false)
417 }
418 ZoneColPred::Gt(val) => {
420 compare_zone_values(&zone.max, zone.max_key.as_ref(), val, None)
421 .map(|o| o != std::cmp::Ordering::Greater)
422 .unwrap_or(false)
423 }
424 ZoneColPred::Gte(val) => {
426 compare_zone_values(&zone.max, zone.max_key.as_ref(), val, None)
427 .map(|o| o == std::cmp::Ordering::Less)
428 .unwrap_or(false)
429 }
430 ZoneColPred::Lt(val) => {
432 compare_zone_values(&zone.min, zone.min_key.as_ref(), val, None)
433 .map(|o| o != std::cmp::Ordering::Less)
434 .unwrap_or(false)
435 }
436 ZoneColPred::Lte(val) => {
438 compare_zone_values(&zone.min, zone.min_key.as_ref(), val, None)
439 .map(|o| o == std::cmp::Ordering::Greater)
440 .unwrap_or(false)
441 }
442 }
443 }
444}
445
446#[derive(Clone, Copy)]
448pub(crate) enum GraphEntityKind {
449 Node,
450 Edge,
451}
452
453impl GraphEntityKind {
454 fn mask(self) -> u8 {
455 match self {
456 Self::Node => 1,
457 Self::Edge => 2,
458 }
459 }
460
461 pub(crate) fn matches(self, kind: &EntityKind) -> bool {
462 matches!(
463 (self, kind),
464 (Self::Node, EntityKind::GraphNode(_)) | (Self::Edge, EntityKind::GraphEdge(_))
465 )
466 }
467}
468
469pub(crate) const SCAN_BATCH_SIZE: usize = 256;
471
472pub(crate) const CHECKPOINT_MUTATIONS_MAX: usize = 8192;
474
475pub(crate) struct SegmentScanCursor {
477 flat_end: usize,
478 hash_end: usize,
479 deleted_end: usize,
480 position: usize,
481}
482
483impl SegmentScanCursor {
484 pub(crate) fn remaining(&self) -> usize {
485 self.flat_end + self.hash_end - self.position
486 }
487
488 pub(crate) fn finished(&self) -> bool {
489 self.remaining() == 0
490 }
491}
492
493pub(crate) struct GraphIndexCursor {
496 key_index: usize,
497 after: Option<EntityId>,
498 upper: EntityId,
499}
500
501impl GraphIndexCursor {
502 pub(crate) fn finished(&self, keys: &[EntityId]) -> bool {
503 self.key_index == keys.len()
504 }
505}
506
507pub struct GrowingSegment {
509 id: SegmentId,
511 collection: String,
513 state: SegmentState,
515 created_at: u64,
517 last_write_at: u64,
519
520 entities: HashMap<EntityId, UnifiedEntity>,
522 hash_scan_ids: Vec<EntityId>,
525 flat_entities: Vec<UnifiedEntity>,
528 base_entity_id: u64,
530 use_flat: bool,
532 deleted: HashMap<EntityId, usize>,
536 metadata: MetadataStorage,
538
539 pk_index: BTreeMap<(String, String), EntityId>,
541 unique_key_lookup: parking_lot::RwLock<Vec<SegmentUniqueKey>>,
544 unique_key_lookup_bytes: AtomicU64,
545 key_lookup: Option<SegmentKeyLookup>,
546 kind_index: HashMap<String, HashSet<EntityId>>,
548 graph_kind_mask: u8,
551 graph_read_index: parking_lot::RwLock<Option<SegmentGraphIndex>>,
553 graph_read_index_bytes: AtomicU64,
554 cross_ref_forward: HashMap<EntityId, Vec<(EntityId, RefType)>>,
556 cross_ref_reverse: HashMap<EntityId, Vec<(EntityId, RefType)>>,
558
559 col_zones: HashMap<String, ColZone>,
561 zone_payload_bytes: usize,
562 sealed_col_zones: HashMap<String, MultiColZone>,
564
565 sequence: AtomicU64,
567 checkpoint_revision: u64,
569 checkpoint_mutations: std::collections::VecDeque<(u64, EntityId)>,
570 memory_bytes: AtomicU64,
575 dead_entity_bytes: u64,
578 dead_resident_count: usize,
580
581 pub(crate) published_flat_len: AtomicUsize,
588}
589
590impl GrowingSegment {
591 pub(crate) fn may_contain_graph_kind(&self, kind: GraphEntityKind) -> bool {
592 self.graph_kind_mask & kind.mask() != 0
593 }
594
595 fn observe_graph_kind(&mut self, kind: &EntityKind) {
596 self.graph_kind_mask |= match kind {
597 EntityKind::GraphNode(_) => GraphEntityKind::Node.mask(),
598 EntityKind::GraphEdge(_) => GraphEntityKind::Edge.mask(),
599 _ => 0,
600 };
601 }
602
603 pub(crate) fn scan_cursor(&self) -> SegmentScanCursor {
604 SegmentScanCursor {
605 flat_end: self.flat_entities.len(),
606 hash_end: self.hash_scan_ids.len(),
607 deleted_end: self.deleted.len(),
608 position: 0,
609 }
610 }
611
612 pub(crate) fn scan_batch(
616 &self,
617 cursor: &mut SegmentScanCursor,
618 before_work: &mut impl FnMut() -> bool,
619 visit: &mut impl FnMut(EntityId),
620 ) -> bool {
621 for _ in 0..SCAN_BATCH_SIZE {
622 if cursor.finished() {
623 break;
624 }
625 if !before_work() {
626 return false;
627 }
628 let id = if cursor.position < cursor.flat_end {
629 self.flat_entities[cursor.position].id
630 } else {
631 self.hash_scan_ids[cursor.position - cursor.flat_end]
632 };
633 cursor.position += 1;
634 let eligible = match self.deleted.get(&id) {
635 Some(&ordinal) => ordinal >= cursor.deleted_end,
636 None => self.get(id).is_some(),
639 };
640 if eligible {
641 visit(id);
642 }
643 }
644 true
645 }
646
647 #[inline]
650 pub fn for_each_fast<F>(&self, mut f: F) -> bool
651 where
652 F: FnMut(&UnifiedEntity) -> bool,
653 {
654 if self.use_flat {
655 if self.deleted.is_empty() {
657 for entity in &self.flat_entities {
658 if !f(entity) {
659 return false;
660 }
661 }
662 for entity in self.entities.values() {
665 if !f(entity) {
666 return false;
667 }
668 }
669 } else {
670 for entity in &self.flat_entities {
671 if self.deleted.contains_key(&entity.id) {
672 continue;
673 }
674 if !f(entity) {
675 return false;
676 }
677 }
678 for entity in self.entities.values() {
679 if self.deleted.contains_key(&entity.id) {
680 continue;
681 }
682 if !f(entity) {
683 return false;
684 }
685 }
686 }
687 } else {
688 if self.deleted.is_empty() {
690 for entity in self.entities.values() {
691 if !f(entity) {
692 return false;
693 }
694 }
695 } else {
696 for entity in self.entities.values() {
697 if self.deleted.contains_key(&entity.id) {
698 continue;
699 }
700 if !f(entity) {
701 return false;
702 }
703 }
704 }
705 }
706 true
707 }
708
709 pub fn new(id: SegmentId, collection: impl Into<String>) -> Self {
711 let now = current_unix_secs();
712
713 Self {
714 id,
715 collection: collection.into(),
716 state: SegmentState::Growing,
717 created_at: now,
718 last_write_at: now,
719 entities: HashMap::new(),
720 hash_scan_ids: Vec::new(),
721 flat_entities: Vec::new(),
722 base_entity_id: 0,
723 use_flat: false,
724 deleted: HashMap::new(),
725 metadata: MetadataStorage::new(),
726 pk_index: BTreeMap::new(),
727 unique_key_lookup: parking_lot::RwLock::new(Vec::new()),
728 unique_key_lookup_bytes: AtomicU64::new(0),
729 key_lookup: Some(SegmentKeyLookup::default()),
730 kind_index: HashMap::new(),
731 graph_read_index: parking_lot::RwLock::new(Some(SegmentGraphIndex::default())),
732 graph_read_index_bytes: AtomicU64::new(0),
733 graph_kind_mask: 0,
734 cross_ref_forward: HashMap::new(),
735 cross_ref_reverse: HashMap::new(),
736 col_zones: HashMap::new(),
737 zone_payload_bytes: 0,
738 sealed_col_zones: HashMap::new(),
739 sequence: AtomicU64::new(0),
740 checkpoint_revision: 0,
741 checkpoint_mutations: std::collections::VecDeque::new(),
742 memory_bytes: AtomicU64::new(0),
743 dead_entity_bytes: 0,
744 dead_resident_count: 0,
745 published_flat_len: AtomicUsize::new(0),
746 }
747 }
748
749 fn next_sequence(&self) -> u64 {
751 self.sequence.fetch_add(1, Ordering::SeqCst)
752 }
753
754 pub(crate) fn checkpoint_revision(&self) -> u64 {
755 self.checkpoint_revision
756 }
757
758 fn mark_checkpoint_entity_dirty(&mut self, id: EntityId) {
759 self.checkpoint_revision = self
760 .checkpoint_revision
761 .checked_add(1)
762 .expect("segment mutation counter must not overflow");
763 if self.checkpoint_mutations.len() == CHECKPOINT_MUTATIONS_MAX {
764 self.checkpoint_mutations.pop_front();
765 }
766 self.checkpoint_mutations
767 .push_back((self.checkpoint_revision, id));
768 }
769
770 pub(crate) fn checkpoint_mutations_since(&self, revision: u64) -> Option<Vec<EntityId>> {
771 if revision == self.checkpoint_revision {
772 return Some(Vec::new());
773 }
774 let mut expected = revision.checked_add(1)?;
775 let mut ids = Vec::new();
776 for &(sequence, id) in &self.checkpoint_mutations {
777 if sequence <= revision {
778 continue;
779 }
780 if sequence != expected {
781 return None;
782 }
783 ids.push(id);
784 expected = expected.checked_add(1)?;
785 }
786 (expected == self.checkpoint_revision.checked_add(1)?).then_some(ids)
787 }
788
789 pub(crate) fn checkpoint_tracking_bytes(&self) -> usize {
790 self.checkpoint_mutations.capacity() * std::mem::size_of::<(u64, EntityId)>()
791 }
792
793 fn has_live_entity(&self, id: EntityId) -> bool {
794 if self.deleted.contains_key(&id) {
795 return false;
796 }
797 if self.use_flat {
798 let raw = id.raw();
799 if raw >= self.base_entity_id {
800 let idx = (raw - self.base_entity_id) as usize;
801 if self
802 .flat_entities
803 .get(idx)
804 .is_some_and(|entity| entity.id == id)
805 {
806 return true;
807 }
808 }
809 self.entities.contains_key(&id)
812 } else {
813 self.entities.contains_key(&id)
814 }
815 }
816
817 fn update_existing_entity_in_place(
818 &mut self,
819 entity: &UnifiedEntity,
820 ) -> Result<UpdateIndexSnapshot, SegmentError> {
821 self.mark_checkpoint_entity_dirty(entity.id);
822 self.invalidate_unique_key_lookup();
823 if self
824 .get(entity.id)
825 .is_some_and(|old| SegmentKeyLookup::key(old) != SegmentKeyLookup::key(entity))
826 {
827 if let Some(mut index) = self.key_lookup.take() {
828 index.remove(
829 self.get(entity.id)
830 .expect("invariant: existing replacement row"),
831 );
832 index.insert(entity);
833 self.key_lookup = Some(index);
834 }
835 }
836 if (self.graph_kind_mask != 0
837 || matches!(
838 entity.kind,
839 EntityKind::GraphNode(_) | EntityKind::GraphEdge(_)
840 ))
841 && self
842 .get(entity.id)
843 .is_some_and(|old| !SegmentGraphIndex::same_keys(old, entity))
844 {
845 self.invalidate_graph_read_index();
846 }
847 self.observe_graph_kind(&entity.kind);
850 if self.use_flat {
851 let raw = entity.id.raw();
852 if raw >= self.base_entity_id {
853 let idx = (raw - self.base_entity_id) as usize;
854 if let Some(slot) = self
855 .flat_entities
856 .get_mut(idx)
857 .filter(|slot| slot.id == entity.id)
858 {
859 let snapshot = UpdateIndexSnapshot::from_entity(slot);
860 let old_bytes = Self::estimate_entity_size(slot);
861 slot.clone_from(entity);
862 self.replace_entity_memory(old_bytes, Self::estimate_entity_size(entity));
863 return Ok(snapshot);
864 }
865 }
866 let Some(slot) = self.entities.get_mut(&entity.id) else {
867 return Err(SegmentError::NotFound(entity.id));
868 };
869 let snapshot = UpdateIndexSnapshot::from_entity(slot);
870 let old_bytes = Self::estimate_entity_size(slot);
871 slot.clone_from(entity);
872 self.replace_entity_memory(old_bytes, Self::estimate_entity_size(entity));
873 Ok(snapshot)
874 } else {
875 let Some(slot) = self.entities.get_mut(&entity.id) else {
876 return Err(SegmentError::NotFound(entity.id));
877 };
878 let snapshot = UpdateIndexSnapshot::from_entity(slot);
879 let old_bytes = Self::estimate_entity_size(slot);
880 slot.clone_from(entity);
881 self.replace_entity_memory(old_bytes, Self::estimate_entity_size(entity));
882 Ok(snapshot)
883 }
884 }
885
886 fn apply_hot_update_with_metadata(
887 &mut self,
888 entity: &UnifiedEntity,
889 modified_columns: &[String],
890 metadata: Option<&Metadata>,
891 ) -> Result<(), SegmentError> {
892 let old = self.update_existing_entity_in_place(entity)?;
893 self.reindex_for_update(&old, entity, Some(modified_columns));
894 self.update_col_zones_from_entity(entity);
895 if let Some(metadata) = metadata {
896 self.metadata.set_all(entity.id, metadata);
897 }
898 Ok(())
899 }
900
901 fn apply_update_with_metadata(
902 &mut self,
903 entity: &UnifiedEntity,
904 metadata: Option<&Metadata>,
905 ) -> Result<(), SegmentError> {
906 let old = self.update_existing_entity_in_place(entity)?;
907 self.reindex_for_update(&old, entity, None);
908 self.update_col_zones_from_entity(entity);
909 if let Some(metadata) = metadata {
910 self.metadata.set_all(entity.id, metadata);
911 }
912 Ok(())
913 }
914
915 pub fn update_hot_batch_with_metadata<'a, I>(&mut self, items: I) -> Result<(), SegmentError>
916 where
917 I: IntoIterator<Item = (&'a UnifiedEntity, &'a [String], Option<&'a Metadata>)>,
918 {
919 if !self.state.is_writable() {
920 return Err(SegmentError::NotWritable);
921 }
922
923 let items: Vec<(&UnifiedEntity, &[String], Option<&Metadata>)> =
924 items.into_iter().collect();
925 if items.is_empty() {
926 return Ok(());
927 }
928
929 for (entity, _, _) in &items {
930 if !self.has_live_entity(entity.id) {
931 return Err(SegmentError::NotFound(entity.id));
932 }
933 }
934
935 for (entity, modified_columns, metadata) in items {
936 self.apply_hot_update_with_metadata(entity, modified_columns, metadata)?;
937 }
938
939 self.last_write_at = current_unix_secs();
940 Ok(())
941 }
942
943 fn record_tombstone(&mut self, id: EntityId) -> bool {
945 self.mark_checkpoint_entity_dirty(id);
946 let ordinal = self.deleted.len();
947 match self.deleted.entry(id) {
948 std::collections::hash_map::Entry::Vacant(entry) => {
949 entry.insert(ordinal);
950 true
951 }
952 std::collections::hash_map::Entry::Occupied(_) => false,
953 }
954 }
955
956 fn tombstone_flat_entity(&mut self, idx: usize, id: EntityId) -> bool {
962 self.invalidate_unique_key_lookup();
963 if !self.record_tombstone(id) {
964 return false;
965 }
966 if let Some(index) = self.key_lookup.as_mut() {
967 index.remove(&self.flat_entities[idx]);
968 }
969 if let Some(index) = self.graph_read_index.get_mut() {
970 index.remove(&self.flat_entities[idx]);
971 self.graph_read_index_bytes
972 .store(index.memory_bytes(), Ordering::Relaxed);
973 }
974 let size = Self::estimate_entity_size(&self.flat_entities[idx]) as u64;
975 self.dead_entity_bytes += size;
976 self.dead_resident_count += 1;
977 self.metadata.remove_all(id);
978 true
979 }
980
981 fn remove_hashmap_entity(&mut self, id: EntityId) -> bool {
986 self.invalidate_unique_key_lookup();
987 let Some(entity) = self.entities.remove(&id) else {
988 return false;
989 };
990 self.release_memory(Self::estimate_entity_size(&entity));
991 self.unindex_entity(&entity);
992 self.metadata.remove_all(id);
993 self.record_tombstone(id);
994 true
995 }
996
997 pub fn delete_batch(&mut self, ids: &[EntityId]) -> Result<Vec<EntityId>, SegmentError> {
998 if !self.state.is_writable() {
999 return Err(SegmentError::NotWritable);
1000 }
1001 if ids.is_empty() {
1002 return Ok(Vec::new());
1003 }
1004
1005 let mut deleted_ids = Vec::with_capacity(ids.len());
1006
1007 if self.use_flat {
1008 for &id in ids {
1009 let raw = id.raw();
1010 if raw < self.base_entity_id {
1011 if self.remove_hashmap_entity(id) {
1012 deleted_ids.push(id);
1013 }
1014 continue;
1015 }
1016 let idx = (raw - self.base_entity_id) as usize;
1017 if idx < self.flat_entities.len() && self.flat_entities[idx].id == id {
1018 if self.tombstone_flat_entity(idx, id) {
1019 deleted_ids.push(id);
1020 }
1021 } else if self.remove_hashmap_entity(id) {
1022 deleted_ids.push(id);
1023 }
1024 }
1025 } else {
1026 for &id in ids {
1027 if self.remove_hashmap_entity(id) {
1028 deleted_ids.push(id);
1029 }
1030 }
1031 }
1032
1033 if !deleted_ids.is_empty() {
1034 self.last_write_at = current_unix_secs();
1035 }
1036
1037 Ok(deleted_ids)
1038 }
1039
1040 fn add_memory(&self, bytes: usize) {
1042 self.memory_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
1043 }
1044
1045 fn replace_entity_memory(&self, previous: usize, current: usize) {
1046 if current >= previous {
1047 self.add_memory(current - previous);
1048 } else {
1049 self.release_memory(previous - current);
1050 }
1051 }
1052
1053 pub fn memory_bytes(&self) -> u64 {
1058 self.memory_bytes.load(Ordering::Relaxed)
1059 + self.zone_payload_bytes as u64
1060 + self.unique_key_lookup_bytes.load(Ordering::Relaxed)
1061 + self
1062 .key_lookup
1063 .as_ref()
1064 .map_or(0, SegmentKeyLookup::memory_bytes)
1065 + self.graph_read_index_bytes.load(Ordering::Relaxed)
1066 + (self.hash_scan_ids.capacity() * std::mem::size_of::<EntityId>()) as u64
1067 }
1068
1069 fn release_memory(&self, bytes: usize) {
1072 let mut current = self.memory_bytes.load(Ordering::Relaxed);
1073 loop {
1074 let next = current.saturating_sub(bytes as u64);
1075 match self.memory_bytes.compare_exchange_weak(
1076 current,
1077 next,
1078 Ordering::Relaxed,
1079 Ordering::Relaxed,
1080 ) {
1081 Ok(_) => break,
1082 Err(observed) => current = observed,
1083 }
1084 }
1085 }
1086
1087 pub(crate) fn resident_entity_count(&self) -> usize {
1089 self.flat_entities.len() + self.entities.len()
1090 }
1091
1092 pub(crate) fn live_entity_count(&self) -> usize {
1094 self.resident_entity_count()
1095 .saturating_sub(self.dead_resident_count)
1096 }
1097
1098 pub(crate) fn tombstone_count(&self) -> usize {
1101 self.deleted.len()
1102 }
1103
1104 pub(crate) fn resident_bytes(&self) -> u64 {
1107 self.memory_bytes() + self.deleted.len() as u64 * TOMBSTONE_ENTRY_BYTES
1108 }
1109
1110 pub(crate) fn reclaimable_bytes(&self) -> u64 {
1113 self.dead_entity_bytes + self.deleted.len() as u64 * TOMBSTONE_ENTRY_BYTES
1114 }
1115
1116 pub(crate) fn live_entity_ids(&self) -> Vec<EntityId> {
1122 let mut ids = Vec::with_capacity(self.live_entity_count());
1123 self.for_each_fast(|entity| {
1124 ids.push(entity.id);
1125 true
1126 });
1127 ids
1128 }
1129
1130 pub(crate) fn insert_preserving_sequence(
1134 &mut self,
1135 entity: UnifiedEntity,
1136 ) -> Result<EntityId, SegmentError> {
1137 if !self.state.is_writable() {
1138 return Err(SegmentError::NotWritable);
1139 }
1140 if self.entities.contains_key(&entity.id) {
1141 return Err(SegmentError::AlreadyExists(entity.id));
1142 }
1143 let id = entity.id;
1144 self.adopt_entity(entity, None);
1145 self.last_write_at = current_unix_secs();
1146 Ok(id)
1147 }
1148
1149 pub(crate) fn adopt_entity(&mut self, entity: UnifiedEntity, metadata: Option<Metadata>) {
1157 let id = entity.id;
1158
1159 let next = entity.sequence_id.saturating_add(1);
1162 if self.sequence.load(Ordering::Relaxed) < next {
1163 self.sequence.store(next, Ordering::Relaxed);
1164 }
1165
1166 self.add_memory(Self::estimate_entity_size(&entity));
1167 self.index_entity(&entity);
1168 self.update_col_zones_from_entity(&entity);
1169 self.entities.insert(id, entity);
1170
1171 if let Some(metadata) = metadata {
1172 if !metadata.is_empty() {
1173 self.metadata.set_all(id, &metadata);
1174 }
1175 }
1176 }
1177
1178 pub(crate) fn evict_entity(&mut self, id: EntityId) -> bool {
1184 self.mark_checkpoint_entity_dirty(id);
1185 self.invalidate_unique_key_lookup();
1186 let Some(entity) = self.entities.remove(&id) else {
1187 return false;
1188 };
1189 self.release_memory(Self::estimate_entity_size(&entity));
1190 self.unindex_entity(&entity);
1191 for cross_ref in entity.cross_refs() {
1192 if let Some(reverse) = self.cross_ref_reverse.get_mut(&cross_ref.target) {
1193 reverse.retain(|(source, _)| *source != id);
1194 }
1195 }
1196 self.metadata.remove_all(id);
1197 true
1198 }
1199
1200 fn estimate_entity_size(entity: &UnifiedEntity) -> usize {
1202 super::memory_size::entity_bytes(entity)
1203 }
1204
1205 fn update_column_zone(&mut self, column: &str, value: &Value) {
1213 if matches!(value, Value::Null) {
1214 return;
1215 }
1216 match self.col_zones.entry(column.to_string()) {
1217 std::collections::hash_map::Entry::Occupied(mut entry) => {
1218 let before = entry.get().payload_bytes();
1219 entry.get_mut().update(value);
1220 self.zone_payload_bytes = self
1221 .zone_payload_bytes
1222 .saturating_sub(before)
1223 .saturating_add(entry.get().payload_bytes());
1224 }
1225 std::collections::hash_map::Entry::Vacant(entry) => {
1226 let zone = ColZone::new(value.clone());
1227 self.zone_payload_bytes =
1228 self.zone_payload_bytes.saturating_add(zone.payload_bytes());
1229 entry.insert(zone);
1230 }
1231 }
1232 }
1233
1234 fn update_col_zones_from_entity(&mut self, entity: &UnifiedEntity) {
1235 if let EntityData::Row(row) = &entity.data {
1236 if let Some(named) = &row.named {
1237 for (column, value) in named {
1238 self.update_column_zone(column, value);
1239 }
1240 } else if let Some(schema) = &row.schema {
1241 for (column, value) in schema.iter().zip(&row.columns) {
1242 self.update_column_zone(column, value);
1243 }
1244 }
1245 }
1246 }
1247
1248 fn rebuild_sealed_col_zones(&mut self) {
1249 let mut values_by_col: HashMap<String, Vec<(CanonicalKey, Value)>> = HashMap::new();
1250 let mut family_by_col: HashMap<String, reddb_types::CanonicalKeyFamily> = HashMap::new();
1251 let mut mixed_family_cols = HashSet::new();
1252 let mut unsupported_cols = HashSet::new();
1253
1254 let mut observe_row = |row: &super::entity::RowData| {
1255 for (col, value) in row.iter_fields() {
1256 if matches!(value, Value::Null) {
1257 continue;
1258 }
1259 let Some(key) = value_to_canonical_key(value) else {
1260 unsupported_cols.insert(col.to_string());
1261 continue;
1262 };
1263 match family_by_col.get(col).copied() {
1264 Some(existing) if existing != key.family() => {
1265 mixed_family_cols.insert(col.to_string());
1266 }
1267 None => {
1268 family_by_col.insert(col.to_string(), key.family());
1269 }
1270 _ => {}
1271 }
1272 values_by_col
1273 .entry(col.to_string())
1274 .or_default()
1275 .push((key, value.clone()));
1276 }
1277 };
1278
1279 if self.use_flat {
1280 for entity in &self.flat_entities {
1281 if self.deleted.contains_key(&entity.id) {
1282 continue;
1283 }
1284 if let EntityData::Row(row) = &entity.data {
1285 observe_row(row);
1286 }
1287 }
1288 } else {
1289 for entity in self.entities.values() {
1290 if self.deleted.contains_key(&entity.id) {
1291 continue;
1292 }
1293 if let EntityData::Row(row) = &entity.data {
1294 observe_row(row);
1295 }
1296 }
1297 }
1298
1299 let mut sealed_col_zones = HashMap::new();
1300 for (col, mut entries) in values_by_col {
1301 if mixed_family_cols.contains(&col)
1302 || unsupported_cols.contains(&col)
1303 || entries.is_empty()
1304 {
1305 continue;
1306 }
1307 entries.sort_unstable_by(|left, right| left.0.cmp(&right.0));
1308 entries.dedup_by(|left, right| left.0 == right.0);
1309
1310 let intervals = build_minmax_multi_intervals(&entries, SEALED_MULTI_ZONE_MAX_INTERVALS);
1311 if intervals.len() > 1 {
1312 sealed_col_zones.insert(col, MultiColZone { intervals });
1313 }
1314 }
1315
1316 let previous = self
1317 .sealed_col_zones
1318 .values()
1319 .flat_map(|zones| &zones.intervals)
1320 .map(ColZone::payload_bytes)
1321 .fold(0, usize::saturating_add);
1322 let current = sealed_col_zones
1323 .values()
1324 .flat_map(|zones| &zones.intervals)
1325 .map(ColZone::payload_bytes)
1326 .fold(0, usize::saturating_add);
1327 self.zone_payload_bytes = self
1328 .zone_payload_bytes
1329 .saturating_sub(previous)
1330 .saturating_add(current);
1331 self.sealed_col_zones = sealed_col_zones;
1332 }
1333
1334 pub fn can_skip_zone_preds(&self, preds: &[(&str, ZoneColPred<'_>)]) -> bool {
1338 if preds.is_empty() {
1339 return false;
1340 }
1341 for (col, pred) in preds {
1342 if let Some(zone) = self.sealed_col_zones.get(*col) {
1343 if zone.can_skip(pred) {
1344 return true;
1345 }
1346 continue;
1347 }
1348 if let Some(zone) = self.col_zones.get(*col) {
1349 if pred.can_skip(zone) {
1350 return true; }
1352 }
1353 }
1354 false
1355 }
1356
1357 fn index_entity(&mut self, entity: &UnifiedEntity) {
1359 self.mark_checkpoint_entity_dirty(entity.id);
1360 if let Some(index) = self.key_lookup.as_mut() {
1361 index.insert(entity);
1362 }
1363 self.hash_scan_ids.push(entity.id);
1364 self.observe_graph_kind(&entity.kind);
1365 if let Some(index) = self.graph_read_index.get_mut() {
1366 index.insert(entity);
1367 self.graph_read_index_bytes
1368 .store(index.memory_bytes(), Ordering::Relaxed);
1369 }
1370 let indexes = self.unique_key_lookup.get_mut();
1371 for index in indexes.iter_mut() {
1372 index.insert(entity);
1373 }
1374 self.unique_key_lookup_bytes.store(
1375 indexes
1376 .iter()
1377 .map(|index| index.memory_bytes() as u64)
1378 .sum(),
1379 Ordering::Relaxed,
1380 );
1381 let kind_key = entity.kind.storage_type().to_string();
1383 self.kind_index
1384 .entry(kind_key)
1385 .or_default()
1386 .insert(entity.id);
1387
1388 if let EntityData::Row(row) = &entity.data {
1390 if let Some(first_col) = row.columns.first() {
1391 let pk_str = format!("{:?}", first_col);
1392 self.pk_index
1393 .insert((entity.kind.collection().to_string(), pk_str), entity.id);
1394 }
1395 }
1396
1397 for cross_ref in entity.cross_refs() {
1399 self.cross_ref_forward
1400 .entry(cross_ref.source)
1401 .or_default()
1402 .push((cross_ref.target, cross_ref.ref_type));
1403
1404 self.cross_ref_reverse
1405 .entry(cross_ref.target)
1406 .or_default()
1407 .push((cross_ref.source, cross_ref.ref_type));
1408 }
1409 }
1410
1411 fn invalidate_graph_read_index(&mut self) {
1412 *self.graph_read_index.get_mut() = None;
1413 self.graph_read_index_bytes.store(0, Ordering::Relaxed);
1414 }
1415
1416 #[cfg(test)]
1419 pub(crate) fn visit_graph_index_candidates(
1420 &self,
1421 kind: GraphEntityKind,
1422 keys: &[EntityId],
1423 before_work: &mut impl FnMut() -> bool,
1424 visit: &mut impl FnMut(&UnifiedEntity) -> bool,
1425 ) -> bool {
1426 let Some(cached) = self.graph_read_index_for_query(before_work) else {
1427 return false;
1428 };
1429 let index = cached.as_ref().expect("complete graph index");
1430 for key in keys {
1431 if !before_work() {
1432 return false;
1433 }
1434 for id in index.candidates(kind, *key) {
1435 if !before_work() {
1436 return false;
1437 }
1438 if let Some(entity) = self.get(id) {
1439 if !visit(entity) {
1440 return false;
1441 }
1442 }
1443 }
1444 }
1445 true
1446 }
1447
1448 fn graph_read_index_for_query(
1449 &self,
1450 before_work: &mut impl FnMut() -> bool,
1451 ) -> Option<parking_lot::RwLockReadGuard<'_, Option<SegmentGraphIndex>>> {
1452 let mut cached = self.graph_read_index.read();
1453 if cached.is_none() {
1454 drop(cached);
1455 let mut writer = self.graph_read_index.write();
1456 if writer.is_none() {
1457 let mut index = SegmentGraphIndex::default();
1458 if !self.for_each_fast(|entity| {
1461 if !before_work() {
1462 return false;
1463 }
1464 index.insert(entity);
1465 true
1466 }) {
1467 return None;
1468 }
1469 self.graph_read_index_bytes
1470 .store(index.memory_bytes(), Ordering::Relaxed);
1471 *writer = Some(index);
1472 }
1473 cached = parking_lot::RwLockWriteGuard::downgrade(writer);
1474 }
1475 Some(cached)
1476 }
1477
1478 pub(crate) fn graph_index_cursor(
1479 &self,
1480 kind: GraphEntityKind,
1481 keys: &[EntityId],
1482 before_work: &mut impl FnMut() -> bool,
1483 ) -> Option<GraphIndexCursor> {
1484 let cached = self.graph_read_index_for_query(before_work)?;
1485 let index = cached.as_ref().expect("complete graph index");
1486 let mut upper = EntityId::new(0);
1487 for key in keys {
1488 if !before_work() {
1489 return None;
1490 }
1491 if let Some(last) = index.candidates(kind, *key).next_back() {
1492 upper = upper.max(last);
1493 }
1494 }
1495 Some(GraphIndexCursor {
1496 key_index: 0,
1497 after: None,
1498 upper,
1499 })
1500 }
1501
1502 pub(crate) fn graph_index_batch(
1505 &self,
1506 kind: GraphEntityKind,
1507 keys: &[EntityId],
1508 cursor: &mut GraphIndexCursor,
1509 before_work: &mut impl FnMut() -> bool,
1510 visit: &mut impl FnMut(EntityId),
1511 ) -> bool {
1512 self.graph_index_batch_with_limit(kind, keys, cursor, before_work, visit, SCAN_BATCH_SIZE)
1513 }
1514
1515 pub(crate) fn graph_index_batch_with_limit(
1516 &self,
1517 kind: GraphEntityKind,
1518 keys: &[EntityId],
1519 cursor: &mut GraphIndexCursor,
1520 before_work: &mut impl FnMut() -> bool,
1521 visit: &mut impl FnMut(EntityId),
1522 limit: usize,
1523 ) -> bool {
1524 assert!(
1525 limit > 0 && limit <= SCAN_BATCH_SIZE,
1526 "bounded graph index batch"
1527 );
1528 let Some(cached) = self.graph_read_index_for_query(before_work) else {
1529 return false;
1530 };
1531 let index = cached.as_ref().expect("complete graph index");
1532 let mut count = 0;
1533 while !cursor.finished(keys) && count < limit {
1534 let key = keys[cursor.key_index];
1535 if !before_work() {
1536 return false;
1537 }
1538 for id in index
1539 .candidates_after(kind, key, cursor.after, cursor.upper)
1540 .take(limit - count)
1541 {
1542 if !before_work() {
1543 return false;
1544 }
1545 cursor.after = Some(id);
1546 count += 1;
1547 visit(id);
1548 }
1549 if count == limit && cursor.after != Some(cursor.upper) {
1550 break;
1551 }
1552 cursor.key_index += 1;
1553 cursor.after = None;
1554 }
1555 true
1556 }
1557
1558 pub(crate) fn find_unique_key_conflict(
1563 &self,
1564 columns: &[Vec<String>],
1565 key_index: usize,
1566 signatures: &[String],
1567 exclude_id: Option<EntityId>,
1568 reserves_key: &impl Fn(&UnifiedEntity) -> bool,
1569 ) -> Option<EntityId> {
1570 assert!(
1571 key_index < columns.len(),
1572 "constraint index belongs to current schema"
1573 );
1574 let cached = self.unique_key_lookup.read();
1575 if cached.iter().map(|index| &index.columns).eq(columns.iter()) {
1576 return self.find_unique_key_candidate(
1577 &cached[key_index],
1578 signatures,
1579 exclude_id,
1580 reserves_key,
1581 );
1582 }
1583 drop(cached);
1584 let mut cached = self.unique_key_lookup.write();
1585 if !cached.iter().map(|index| &index.columns).eq(columns.iter()) {
1586 let mut indexes: Vec<_> = columns
1587 .iter()
1588 .map(|columns| SegmentUniqueKey::new(columns))
1589 .collect();
1590 self.for_each_fast(|entity| {
1591 for index in &mut indexes {
1592 index.insert(entity);
1593 }
1594 true
1595 });
1596 self.unique_key_lookup_bytes.store(
1597 indexes
1598 .iter()
1599 .map(|index| index.memory_bytes() as u64)
1600 .sum(),
1601 Ordering::Relaxed,
1602 );
1603 *cached = indexes;
1604 }
1605 self.find_unique_key_candidate(&cached[key_index], signatures, exclude_id, reserves_key)
1606 }
1607
1608 fn find_unique_key_candidate(
1609 &self,
1610 index: &SegmentUniqueKey,
1611 signatures: &[String],
1612 exclude_id: Option<EntityId>,
1613 reserves_key: &impl Fn(&UnifiedEntity) -> bool,
1614 ) -> Option<EntityId> {
1615 index.candidates(signatures).iter().copied().find(|id| {
1616 if Some(*id) == exclude_id {
1617 return false;
1618 }
1619 let Some(
1620 entity @ UnifiedEntity {
1621 data: EntityData::Row(row),
1622 ..
1623 },
1624 ) = self.get(*id)
1625 else {
1626 return false;
1627 };
1628 index
1629 .columns
1630 .iter()
1631 .zip(signatures)
1632 .all(|(column, signature)| {
1633 row.get_field(column)
1634 .is_some_and(|value| format!("{value:?}") == *signature)
1635 })
1636 && reserves_key(entity)
1637 })
1638 }
1639
1640 fn invalidate_unique_key_lookup(&mut self) {
1641 *self.unique_key_lookup.get_mut() = Vec::new();
1642 self.unique_key_lookup_bytes.store(0, Ordering::Relaxed);
1643 }
1644
1645 pub(crate) fn find_key_entity(
1648 &self,
1649 key: &str,
1650 newest: bool,
1651 visible: &mut impl FnMut(&UnifiedEntity) -> bool,
1652 ) -> Option<&UnifiedEntity> {
1653 let index = self
1656 .key_lookup
1657 .as_ref()
1658 .expect("segment owns its key index");
1659 self.find_key_candidate(index, key, newest, visible)
1660 }
1661
1662 fn find_key_candidate(
1663 &self,
1664 index: &SegmentKeyLookup,
1665 key: &str,
1666 newest: bool,
1667 visible: &mut impl FnMut(&UnifiedEntity) -> bool,
1668 ) -> Option<&UnifiedEntity> {
1669 self.select_key_entity(
1670 index
1671 .candidates(key)
1672 .iter()
1673 .chain(index.mutated_candidates())
1674 .filter_map(|id| self.get(*id)),
1675 key,
1676 newest,
1677 visible,
1678 )
1679 }
1680
1681 fn select_key_entity<'a>(
1682 &self,
1683 entities: impl Iterator<Item = &'a UnifiedEntity>,
1684 key: &str,
1685 newest: bool,
1686 visible: &mut impl FnMut(&UnifiedEntity) -> bool,
1687 ) -> Option<&'a UnifiedEntity> {
1688 let mut best: Option<&UnifiedEntity> = None;
1689 for entity in entities {
1690 if SegmentKeyLookup::key(entity) != Some(key) || !visible(entity) {
1692 continue;
1693 }
1694 if !newest {
1695 return Some(entity);
1696 }
1697 if best.is_none_or(|current| entity.xmin >= current.xmin) {
1698 best = Some(entity);
1699 }
1700 }
1701 best
1702 }
1703
1704 pub fn may_contain_exact_key(&self, key: &[u8]) -> bool {
1706 if let Some(id) = entity_id_from_probe_key(key) {
1707 return self.has_live_entity(id);
1708 }
1709
1710 let Ok(pk_str) = std::str::from_utf8(key) else {
1711 return false;
1712 };
1713 let pk_key = (self.collection.clone(), pk_str.to_string());
1714 self.pk_index
1715 .get(&pk_key)
1716 .is_some_and(|id| self.has_live_entity(*id))
1717 }
1718
1719 fn unindex_entity(&mut self, entity: &UnifiedEntity) {
1721 if let Some(index) = self.key_lookup.as_mut() {
1722 index.remove(entity);
1723 }
1724 if let Some(index) = self.graph_read_index.get_mut() {
1725 index.remove(entity);
1726 self.graph_read_index_bytes
1727 .store(index.memory_bytes(), Ordering::Relaxed);
1728 }
1729 let kind_key = entity.kind.storage_type().to_string();
1731 if let Some(set) = self.kind_index.get_mut(&kind_key) {
1732 set.remove(&entity.id);
1733 }
1734
1735 if let EntityData::Row(row) = &entity.data {
1737 if let Some(first_col) = row.columns.first() {
1738 let pk_str = format!("{:?}", first_col);
1739 self.pk_index
1740 .remove(&(entity.kind.collection().to_string(), pk_str));
1741 }
1742 }
1743
1744 self.cross_ref_forward.remove(&entity.id);
1746 }
1748
1749 fn reindex_for_update(
1759 &mut self,
1760 old: &UpdateIndexSnapshot,
1761 new: &UnifiedEntity,
1762 modified_columns: Option<&[String]>,
1763 ) {
1764 let pk_changed = match &new.data {
1769 EntityData::Row(new_row) => {
1770 if let Some(cols) = modified_columns {
1771 let pk_col_name = old.pk_column_name.as_deref().or_else(|| {
1774 new_row
1775 .schema
1776 .as_deref()
1777 .and_then(|schema| schema.first().map(|name| name.as_str()))
1778 });
1779 match pk_col_name {
1780 Some(pk_name) => cols.iter().any(|c| c.eq_ignore_ascii_case(pk_name)),
1781 None => old.pk_value.as_ref() != new_row.columns.first(),
1783 }
1784 } else {
1785 old.pk_value.as_ref() != new_row.columns.first()
1786 }
1787 }
1788 _ => false,
1790 };
1791
1792 if pk_changed {
1793 if let Some((collection, pk_str)) = &old.pk_index_key {
1795 self.pk_index.remove(&(collection.clone(), pk_str.clone()));
1796 }
1797 if let EntityData::Row(row) = &new.data {
1799 if let Some(first_col) = row.columns.first() {
1800 let pk_str = format!("{:?}", first_col);
1801 self.pk_index
1802 .insert((new.kind.collection().to_string(), pk_str), new.id);
1803 }
1804 }
1805 }
1806
1807 let new_refs = new.cross_refs();
1809 if old.cross_refs.as_slice() != new_refs {
1810 self.cross_ref_forward.remove(&new.id);
1812 for cross_ref in &old.cross_refs {
1814 if let Some(rev) = self.cross_ref_reverse.get_mut(&cross_ref.target) {
1815 rev.retain(|(src, _)| *src != new.id);
1816 }
1817 }
1818 for cross_ref in new_refs {
1820 self.cross_ref_forward
1821 .entry(cross_ref.source)
1822 .or_default()
1823 .push((cross_ref.target, cross_ref.ref_type));
1824 self.cross_ref_reverse
1825 .entry(cross_ref.target)
1826 .or_default()
1827 .push((cross_ref.source, cross_ref.ref_type));
1828 }
1829 }
1830 }
1831
1832 pub fn get_references_to(&self, id: EntityId) -> Vec<(EntityId, RefType)> {
1834 self.cross_ref_reverse.get(&id).cloned().unwrap_or_default()
1835 }
1836
1837 pub fn get_references_from(&self, id: EntityId) -> Vec<(EntityId, RefType)> {
1839 self.cross_ref_forward.get(&id).cloned().unwrap_or_default()
1840 }
1841
1842 pub fn age_secs(&self) -> u64 {
1844 let now = current_unix_secs();
1845 now.saturating_sub(self.created_at)
1846 }
1847
1848 pub fn idle_secs(&self) -> u64 {
1850 let now = current_unix_secs();
1851 now.saturating_sub(self.last_write_at)
1852 }
1853
1854 pub fn bulk_insert(
1863 &mut self,
1864 entities: Vec<UnifiedEntity>,
1865 ) -> Result<Vec<EntityId>, SegmentError> {
1866 if !self.state.is_writable() {
1867 return Err(SegmentError::NotWritable);
1868 }
1869
1870 for entity in &entities {
1871 self.mark_checkpoint_entity_dirty(entity.id);
1872 }
1873
1874 if let Some(index) = self.key_lookup.as_mut() {
1875 for entity in &entities {
1876 index.insert(entity);
1877 }
1878 }
1879
1880 let indexes = self.unique_key_lookup.get_mut();
1881 for index in indexes.iter_mut() {
1882 for entity in &entities {
1883 index.insert(entity);
1884 }
1885 }
1886 self.unique_key_lookup_bytes.store(
1887 indexes
1888 .iter()
1889 .map(|index| index.memory_bytes() as u64)
1890 .sum(),
1891 Ordering::Relaxed,
1892 );
1893
1894 let n = entities.len();
1895
1896 if entities.is_empty() {
1897 return Ok(Vec::new());
1898 }
1899
1900 for run in
1903 entities.chunk_by(|left, right| left.kind.storage_type() == right.kind.storage_type())
1904 {
1905 let kind = run[0].kind.storage_type();
1906 self.observe_graph_kind(&run[0].kind);
1907 if matches!(
1908 run[0].kind,
1909 EntityKind::GraphNode(_) | EntityKind::GraphEdge(_)
1910 ) {
1911 if let Some(index) = self.graph_read_index.get_mut() {
1912 for entity in run {
1913 index.insert(entity);
1914 }
1915 self.graph_read_index_bytes
1916 .store(index.memory_bytes(), Ordering::Relaxed);
1917 }
1918 }
1919 let kind_set = match self.kind_index.get_mut(kind) {
1920 Some(ids) => ids,
1921 None => self.kind_index.entry(kind.to_string()).or_default(),
1922 };
1923 kind_set.reserve(run.len());
1924 kind_set.extend(run.iter().map(|entity| entity.id));
1925 }
1926
1927 let now = current_unix_secs();
1928
1929 let base_seq = self.sequence.fetch_add(n as u64, Ordering::Relaxed);
1930
1931 let mut ids = Vec::with_capacity(n);
1932
1933 if self.flat_entities.is_empty() && self.entities.is_empty() {
1935 self.base_entity_id = entities.first().map(|e| e.id.raw()).unwrap_or(0);
1937 self.use_flat = true;
1938 }
1939
1940 let mut columnar_zone_updates: Vec<Vec<Value>> = Vec::new();
1950 let mut columnar_schema: Option<std::sync::Arc<Vec<String>>> = None;
1951 let mut named_zone_updates: Vec<(String, Value)> = Vec::new();
1955
1956 let mut batch_bytes: usize = 0;
1961
1962 if self.use_flat {
1963 self.flat_entities.reserve(n);
1964 for (i, mut entity) in entities.into_iter().enumerate() {
1965 entity.sequence_id = base_seq + i as u64;
1966 let id = entity.id;
1967 ids.push(id);
1968 batch_bytes += Self::estimate_entity_size(&entity);
1969 if let EntityData::Row(row) = &entity.data {
1970 if row.schema.is_some() && !row.columns.is_empty() {
1971 if columnar_zone_updates.is_empty() {
1972 columnar_zone_updates = vec![Vec::with_capacity(n); row.columns.len()];
1973 columnar_schema = row.schema.clone();
1974 }
1975 for (ci, val) in row.columns.iter().enumerate() {
1976 if !matches!(val, Value::Null) {
1977 if let Some(bucket) = columnar_zone_updates.get_mut(ci) {
1978 bucket.push(val.clone());
1979 }
1980 }
1981 }
1982 } else {
1983 for (col, val) in row.iter_fields() {
1984 if !matches!(val, Value::Null) {
1985 named_zone_updates.push((col.to_string(), val.clone()));
1986 }
1987 }
1988 }
1989 }
1990 let expected = self.base_entity_id + self.flat_entities.len() as u64;
2006 if id.raw() == expected {
2007 self.flat_entities.push(entity);
2008 } else {
2009 self.hash_scan_ids.push(id);
2010 self.entities.insert(id, entity);
2011 }
2012 }
2013 } else {
2014 self.entities.reserve(n);
2016 self.hash_scan_ids.reserve(n);
2017 let mut pairs = Vec::with_capacity(n);
2018 for (i, mut entity) in entities.into_iter().enumerate() {
2019 entity.sequence_id = base_seq + i as u64;
2020 let id = entity.id;
2021 ids.push(id);
2022 batch_bytes += Self::estimate_entity_size(&entity);
2023 if let EntityData::Row(row) = &entity.data {
2024 for (col, val) in row.iter_fields() {
2025 if !matches!(val, Value::Null) {
2026 named_zone_updates.push((col.to_string(), val.clone()));
2027 }
2028 }
2029 }
2030 self.hash_scan_ids.push(id);
2031 pairs.push((id, entity));
2032 }
2033 self.entities.extend(pairs);
2034 }
2035
2036 if !columnar_zone_updates.is_empty() {
2040 let schema = columnar_schema.as_ref();
2041 for (ci, values) in columnar_zone_updates.into_iter().enumerate() {
2042 if values.is_empty() {
2043 continue;
2044 }
2045 let Some(col_name) = schema.and_then(|s| s.get(ci)) else {
2046 continue;
2047 };
2048 let mut iter = values.into_iter();
2054 if let Some(first) = iter.next() {
2055 let previous = self
2056 .col_zones
2057 .get(col_name)
2058 .map_or(0, ColZone::payload_bytes);
2059 let zone = self
2060 .col_zones
2061 .entry(col_name.clone())
2062 .and_modify(|z| z.update(&first))
2063 .or_insert_with(|| ColZone::new(first));
2064 for v in iter {
2065 zone.update(&v);
2066 }
2067 self.zone_payload_bytes = self
2068 .zone_payload_bytes
2069 .saturating_sub(previous)
2070 .saturating_add(zone.payload_bytes());
2071 }
2072 }
2073 }
2074 for (col, val) in named_zone_updates {
2075 self.update_column_zone(&col, &val);
2076 }
2077
2078 self.add_memory(batch_bytes);
2079 self.last_write_at = now;
2080
2081 if self.use_flat {
2083 self.published_flat_len
2084 .store(self.flat_entities.len(), Ordering::Release);
2085 }
2086
2087 Ok(ids)
2088 }
2089
2090 pub(crate) fn force_delete(&mut self, id: EntityId) -> bool {
2093 if self.use_flat {
2094 let raw = id.raw();
2095 if raw >= self.base_entity_id {
2096 let idx = (raw - self.base_entity_id) as usize;
2097 if idx < self.flat_entities.len() && self.flat_entities[idx].id == id {
2098 self.tombstone_flat_entity(idx, id);
2099 return true;
2100 }
2101 }
2102 return self.remove_hashmap_entity(id);
2103 }
2104
2105 self.remove_hashmap_entity(id)
2106 }
2107
2108 pub(crate) fn force_update_with_metadata(
2111 &mut self,
2112 entity: &UnifiedEntity,
2113 modified_columns: &[String],
2114 metadata: Option<&Metadata>,
2115 ) -> Result<(), SegmentError> {
2116 self.apply_hot_update_with_metadata(entity, modified_columns, metadata)
2117 }
2118}
2119
2120impl UnifiedSegment for GrowingSegment {
2121 fn id(&self) -> SegmentId {
2122 self.id
2123 }
2124
2125 fn state(&self) -> SegmentState {
2126 self.state
2127 }
2128
2129 fn collection(&self) -> &str {
2130 &self.collection
2131 }
2132
2133 fn stats(&self) -> SegmentStats {
2134 let mut stats = SegmentStats {
2135 entity_count: self.live_entity_count(),
2136 deleted_count: self.deleted.len(),
2137 memory_bytes: self.resident_bytes() as usize,
2138 ..Default::default()
2139 };
2140
2141 for entity in self.entities.values() {
2142 match &entity.kind {
2143 EntityKind::TableRow { .. } => stats.row_count += 1,
2144 EntityKind::GraphNode(_) => stats.node_count += 1,
2145 EntityKind::GraphEdge(_) => stats.edge_count += 1,
2146 EntityKind::Vector { .. } => stats.vector_count += 1,
2147 EntityKind::TimeSeriesPoint(_) => stats.row_count += 1,
2148 EntityKind::QueueMessage { .. } => stats.row_count += 1,
2149 }
2150 stats.cross_ref_count += entity.cross_refs().len();
2151 }
2152
2153 stats
2154 }
2155
2156 fn entity_count(&self) -> usize {
2157 self.live_entity_count()
2158 }
2159
2160 fn contains(&self, id: EntityId) -> bool {
2161 self.has_live_entity(id)
2162 }
2163
2164 fn get(&self, id: EntityId) -> Option<&UnifiedEntity> {
2165 if self.deleted.contains_key(&id) {
2166 return None;
2167 }
2168 if self.use_flat {
2169 let raw = id.raw();
2170 if raw >= self.base_entity_id {
2171 let idx = (raw - self.base_entity_id) as usize;
2172 if let Some(entity) = self.flat_entities.get(idx).filter(|e| e.id == id) {
2173 return Some(entity);
2174 }
2175 }
2176 self.entities.get(&id)
2182 } else {
2183 self.entities.get(&id)
2184 }
2185 }
2186
2187 fn get_mut(&mut self, id: EntityId) -> Option<&mut UnifiedEntity> {
2188 self.mark_checkpoint_entity_dirty(id);
2189 self.invalidate_unique_key_lookup();
2190 self.invalidate_graph_read_index();
2191 if self.deleted.contains_key(&id) || !self.state.is_writable() {
2192 return None;
2193 }
2194 if let Some(mut index) = self.key_lookup.take() {
2195 if let Some(entity) = self.get(id) {
2196 index.mark_mutated(entity);
2197 }
2198 self.key_lookup = Some(index);
2199 }
2200 self.graph_kind_mask = GraphEntityKind::Node.mask() | GraphEntityKind::Edge.mask();
2203 if self.use_flat {
2204 let raw = id.raw();
2205 if raw >= self.base_entity_id {
2206 let idx = (raw - self.base_entity_id) as usize;
2207 if self
2208 .flat_entities
2209 .get(idx)
2210 .map(|e| e.id == id)
2211 .unwrap_or(false)
2212 {
2213 return self.flat_entities.get_mut(idx);
2214 }
2215 }
2216 self.entities.get_mut(&id)
2219 } else {
2220 self.entities.get_mut(&id)
2221 }
2222 }
2223
2224 fn insert(&mut self, mut entity: UnifiedEntity) -> Result<EntityId, SegmentError> {
2225 if !self.state.is_writable() {
2226 return Err(SegmentError::NotWritable);
2227 }
2228
2229 if self.entities.contains_key(&entity.id) {
2230 return Err(SegmentError::AlreadyExists(entity.id));
2231 }
2232
2233 entity.sequence_id = self.next_sequence();
2235
2236 let size = Self::estimate_entity_size(&entity);
2238 self.add_memory(size);
2239
2240 self.index_entity(&entity);
2242
2243 self.update_col_zones_from_entity(&entity);
2245
2246 let id = entity.id;
2248 self.entities.insert(id, entity);
2249
2250 self.last_write_at = current_unix_secs();
2252
2253 Ok(id)
2254 }
2255
2256 fn update(&mut self, entity: UnifiedEntity) -> Result<(), SegmentError> {
2257 if !self.state.is_writable() {
2258 return Err(SegmentError::NotWritable);
2259 }
2260
2261 self.apply_update_with_metadata(&entity, None)?;
2262 self.last_write_at = current_unix_secs();
2263
2264 Ok(())
2265 }
2266
2267 fn update_hot(
2268 &mut self,
2269 entity: UnifiedEntity,
2270 modified_columns: &[String],
2271 ) -> Result<(), SegmentError> {
2272 if !self.state.is_writable() {
2273 return Err(SegmentError::NotWritable);
2274 }
2275
2276 self.apply_hot_update_with_metadata(&entity, modified_columns, None)?;
2277 self.last_write_at = current_unix_secs();
2278 Ok(())
2279 }
2280
2281 fn delete(&mut self, id: EntityId) -> Result<bool, SegmentError> {
2282 if !self.state.is_writable() {
2283 return Err(SegmentError::NotWritable);
2284 }
2285
2286 if self.use_flat {
2288 let raw = id.raw();
2289 if raw >= self.base_entity_id {
2290 let idx = (raw - self.base_entity_id) as usize;
2291 if idx < self.flat_entities.len() && self.flat_entities[idx].id == id {
2292 self.tombstone_flat_entity(idx, id);
2293 return Ok(true);
2294 }
2295 }
2296 return Ok(self.remove_hashmap_entity(id));
2297 }
2298
2299 Ok(self.remove_hashmap_entity(id))
2300 }
2301
2302 fn get_metadata(&self, id: EntityId) -> Option<Metadata> {
2303 if !self.has_live_entity(id) {
2304 return None;
2305 }
2306 Some(self.metadata.get_all(id))
2307 }
2308
2309 fn set_metadata(&mut self, id: EntityId, metadata: Metadata) -> Result<(), SegmentError> {
2310 self.mark_checkpoint_entity_dirty(id);
2311 if !self.state.is_writable() {
2312 return Err(SegmentError::NotWritable);
2313 }
2314
2315 if !self.has_live_entity(id) {
2316 return Err(SegmentError::NotFound(id));
2317 }
2318
2319 self.metadata.set_all(id, &metadata);
2320 Ok(())
2321 }
2322
2323 fn seal(&mut self) -> Result<(), SegmentError> {
2324 if self.state != SegmentState::Growing {
2325 return Err(SegmentError::InvalidState(self.state));
2326 }
2327
2328 self.state = SegmentState::Sealing;
2329
2330 self.rebuild_sealed_col_zones();
2335
2336 self.state = SegmentState::Sealed;
2337 Ok(())
2338 }
2339
2340 fn should_seal(&self, config: &SegmentConfig) -> bool {
2341 if self.entities.len() >= config.max_entities {
2343 return true;
2344 }
2345
2346 if self.memory_bytes.load(Ordering::Relaxed) as usize >= config.max_bytes {
2348 return true;
2349 }
2350
2351 if self.age_secs() >= config.max_age_secs {
2353 return true;
2354 }
2355
2356 false
2357 }
2358
2359 fn iter(&self) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_> {
2360 let base: Box<dyn Iterator<Item = &UnifiedEntity>> = if self.use_flat {
2361 Box::new(self.flat_entities.iter().chain(self.entities.values()))
2365 } else {
2366 Box::new(self.entities.values())
2367 };
2368 if self.deleted.is_empty() {
2369 base
2370 } else {
2371 Box::new(base.filter(|e| !self.deleted.contains_key(&e.id)))
2372 }
2373 }
2374
2375 fn iter_kind(&self, kind_filter: &str) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_> {
2376 let ids = self.kind_index.get(kind_filter);
2379 let flat: Box<dyn Iterator<Item = &UnifiedEntity>> = if self.use_flat {
2382 Box::new(self.flat_entities.iter())
2383 } else {
2384 Box::new(std::iter::empty())
2385 };
2386 Box::new(flat.chain(self.entities.values()).filter(move |e| {
2387 if self.deleted.contains_key(&e.id) {
2388 return false;
2389 }
2390 if let Some(ids) = ids {
2391 ids.contains(&e.id)
2392 } else {
2393 false
2394 }
2395 }))
2396 }
2397
2398 fn filter_metadata(
2399 &self,
2400 filters: &[(String, super::metadata::MetadataFilter)],
2401 ) -> Vec<EntityId> {
2402 let flat_ids: Box<dyn Iterator<Item = EntityId> + '_> = if self.use_flat {
2405 Box::new(self.flat_entities.iter().map(|e| e.id))
2406 } else {
2407 Box::new(std::iter::empty())
2408 };
2409 flat_ids
2410 .chain(self.entities.keys().copied())
2411 .filter(|id| {
2412 if self.deleted.contains_key(id) {
2413 return false;
2414 }
2415 let metadata = self.metadata.get_all(*id);
2416 metadata.matches_all(filters)
2417 })
2418 .collect()
2419 }
2420}
2421
2422fn build_minmax_multi_intervals(
2423 entries: &[(CanonicalKey, Value)],
2424 max_intervals: usize,
2425) -> Vec<ColZone> {
2426 if entries.is_empty() {
2427 return Vec::new();
2428 }
2429 if entries.len() == 1 || max_intervals <= 1 {
2430 return vec![ColZone::with_bounds(
2431 entries[0].1.clone(),
2432 entries[entries.len() - 1].1.clone(),
2433 )];
2434 }
2435
2436 let mut split_points = if entries.len() <= max_intervals {
2437 (1..entries.len()).collect::<Vec<_>>()
2438 } else {
2439 let target_splits = max_intervals - 1;
2440 let mut selected = select_gap_split_points(entries, target_splits);
2441 if selected.len() < target_splits {
2442 for bucket in 1..max_intervals {
2443 let idx = bucket * entries.len() / max_intervals;
2444 if idx == 0 || idx >= entries.len() || selected.contains(&idx) {
2445 continue;
2446 }
2447 selected.push(idx);
2448 if selected.len() >= target_splits {
2449 break;
2450 }
2451 }
2452 }
2453 selected.sort_unstable();
2454 selected.dedup();
2455 selected
2456 };
2457
2458 split_points.push(entries.len());
2459
2460 let mut out = Vec::with_capacity(split_points.len());
2461 let mut start = 0usize;
2462 for end in split_points {
2463 if end <= start {
2464 continue;
2465 }
2466 out.push(ColZone::with_bounds(
2467 entries[start].1.clone(),
2468 entries[end - 1].1.clone(),
2469 ));
2470 start = end;
2471 }
2472
2473 if out.is_empty() {
2474 out.push(ColZone::with_bounds(
2475 entries[0].1.clone(),
2476 entries[entries.len() - 1].1.clone(),
2477 ));
2478 }
2479
2480 out
2481}
2482
2483fn select_gap_split_points(entries: &[(CanonicalKey, Value)], max_splits: usize) -> Vec<usize> {
2484 let mut gaps = Vec::new();
2485 for idx in 1..entries.len() {
2486 if let Some(score) = canonical_gap_score(&entries[idx - 1].0, &entries[idx].0) {
2487 if score > 0.0 {
2488 gaps.push((score, idx));
2489 }
2490 }
2491 }
2492 gaps.sort_by(|left, right| {
2493 right
2494 .0
2495 .partial_cmp(&left.0)
2496 .unwrap_or(std::cmp::Ordering::Equal)
2497 .then_with(|| left.1.cmp(&right.1))
2498 });
2499 gaps.into_iter()
2500 .take(max_splits)
2501 .map(|(_, idx)| idx)
2502 .collect()
2503}
2504
2505fn canonical_gap_score(left: &CanonicalKey, right: &CanonicalKey) -> Option<f64> {
2506 if left.family() != right.family() {
2507 return None;
2508 }
2509 match (left, right) {
2510 (CanonicalKey::Signed(_, l), CanonicalKey::Signed(_, r)) => {
2511 Some(r.saturating_sub(*l) as f64)
2512 }
2513 (CanonicalKey::Unsigned(_, l), CanonicalKey::Unsigned(_, r)) => {
2514 Some(r.saturating_sub(*l) as f64)
2515 }
2516 (CanonicalKey::Float(l), CanonicalKey::Float(r)) => {
2517 Some((f64::from_bits(*r) - f64::from_bits(*l)).abs())
2518 }
2519 _ => None,
2520 }
2521}
2522
2523#[cfg(test)]
2524mod tests {
2525 use super::*;
2526 use crate::storage::unified::entity::RowData;
2527 use crate::storage::unified::MetadataValue;
2528 use reddb_types::Value;
2529
2530 fn keyed_row(id: u64, key: &str, xmin: u64, xmax: u64) -> UnifiedEntity {
2531 let mut entity = UnifiedEntity::table_row(EntityId::new(id), "keys", id, vec![]);
2532 entity.data = EntityData::Row(RowData::with_names(
2533 vec![Value::text(key), Value::Integer(id as i64)],
2534 vec!["key".into(), "value".into()],
2535 ));
2536 entity.set_xmin(xmin);
2537 entity.set_xmax(xmax);
2538 entity
2539 }
2540
2541 fn keyed_probe(segment: &GrowingSegment, key: &str) -> Option<EntityId> {
2542 segment
2543 .find_key_entity(key, true, &mut |_| true)
2544 .map(|entity| entity.id)
2545 }
2546
2547 #[test]
2548 fn bulk_checkpoint_journal_preserves_prior_changes_and_ignores_rejected_batches() {
2549 let mut segment = GrowingSegment::new(1, "keys");
2550 segment.insert(keyed_row(1, "one", 1, 0)).expect("row");
2551 segment
2552 .bulk_insert(vec![keyed_row(2, "two", 1, 0), keyed_row(3, "three", 1, 0)])
2553 .expect("batch");
2554 assert_eq!(
2555 segment.checkpoint_mutations_since(0),
2556 Some(vec![EntityId::new(1), EntityId::new(2), EntityId::new(3)])
2557 );
2558 let revision = segment.checkpoint_revision();
2559 segment.bulk_insert(Vec::new()).expect("empty batch");
2560 assert_eq!(segment.checkpoint_revision(), revision);
2561 segment.seal().expect("seal segment");
2562 assert!(matches!(
2563 segment.bulk_insert(vec![keyed_row(4, "four", 1, 0)]),
2564 Err(SegmentError::NotWritable)
2565 ));
2566 assert_eq!(segment.checkpoint_revision(), revision);
2567 assert_eq!(segment.checkpoint_mutations_since(revision), Some(vec![]));
2568 }
2569
2570 #[test]
2571 fn bulk_checkpoint_journal_exhaustion_keeps_a_bounded_complete_suffix() {
2572 let mut segment = GrowingSegment::new(1, "keys");
2573 segment
2574 .bulk_insert(
2575 (1..=CHECKPOINT_MUTATIONS_MAX as u64 + 1)
2576 .map(|id| keyed_row(id, "key", 1, 0))
2577 .collect(),
2578 )
2579 .expect("large batch");
2580 assert!(segment.checkpoint_mutations_since(0).is_none());
2581 assert_eq!(
2582 segment
2583 .checkpoint_mutations_since(1)
2584 .expect("complete suffix")
2585 .len(),
2586 CHECKPOINT_MUTATIONS_MAX
2587 );
2588 let revision = segment.checkpoint_revision();
2589 let id = EntityId::new(CHECKPOINT_MUTATIONS_MAX as u64 + 2);
2590 segment
2591 .bulk_insert(vec![keyed_row(id.raw(), "next", 1, 0)])
2592 .expect("later batch");
2593 assert_eq!(segment.checkpoint_mutations_since(revision), Some(vec![id]));
2594 assert_eq!(segment.checkpoint_mutations.len(), CHECKPOINT_MUTATIONS_MAX);
2595 }
2596
2597 #[test]
2598 fn key_lookup_visits_only_requested_key_versions() {
2599 let mut segment = GrowingSegment::new(1, "keys");
2600 segment
2601 .bulk_insert(
2602 (1..=4096)
2603 .map(|id| keyed_row(id, &format!("other-{id}"), 1, 0))
2604 .collect(),
2605 )
2606 .expect("unrelated rows");
2607 segment
2608 .insert(keyed_row(5000, "wanted", 2, 8))
2609 .expect("old version");
2610 segment
2611 .insert(keyed_row(5001, "wanted", 8, 0))
2612 .expect("new version");
2613 let mut visits = 0;
2614 let result = segment.find_key_entity("wanted", true, &mut |entity| {
2615 visits += 1;
2616 entity.xmin < 8
2617 });
2618 assert_eq!(result.map(|entity| entity.id), Some(EntityId::new(5000)));
2619 assert_eq!(
2620 visits, 2,
2621 "unrelated rows must never reach visibility evaluation"
2622 );
2623 assert_eq!(keyed_probe(&segment, "wanted"), Some(EntityId::new(5001)));
2624 let mut misses = 0;
2625 assert!(segment
2626 .find_key_entity("absent", true, &mut |_| {
2627 misses += 1;
2628 true
2629 })
2630 .is_none());
2631 assert_eq!(misses, 0);
2632
2633 let mut collision = SegmentKeyLookup::default();
2635 collision.insert(&keyed_row(1, "wanted", 1, 0));
2636 assert!(segment
2637 .find_key_candidate(&collision, "wanted", true, &mut |_| true)
2638 .is_none());
2639 }
2640
2641 #[test]
2642 fn raw_mutation_does_not_disable_unrelated_key_candidates() {
2643 let mut segment = GrowingSegment::new(1, "keys");
2644 segment
2645 .bulk_insert(
2646 (1..=4096)
2647 .map(|id| keyed_row(id, &format!("key-{id}"), 1, 0))
2648 .collect(),
2649 )
2650 .expect("rows");
2651 for _ in 0..3 {
2652 *segment.get_mut(EntityId::new(1)).expect("mutable") = keyed_row(1, "renamed", 2, 0);
2653 }
2654 let index = segment.key_lookup.as_ref().expect("index retained");
2655 assert_eq!(index.candidates("key-4096"), &[EntityId::new(4096)]);
2656 assert_eq!(index.mutated_candidates().count(), 1);
2657 assert!(index.candidates("key-1").is_empty());
2658 assert_eq!(keyed_probe(&segment, "renamed"), Some(EntityId::new(1)));
2659 assert_eq!(keyed_probe(&segment, "key-4096"), Some(EntityId::new(4096)));
2660 segment.delete(EntityId::new(1)).expect("delete raw row");
2661 assert_eq!(
2662 segment
2663 .key_lookup
2664 .as_ref()
2665 .unwrap()
2666 .mutated_candidates()
2667 .count(),
2668 0
2669 );
2670 assert_eq!(keyed_probe(&segment, "renamed"), None);
2671 }
2672
2673 #[test]
2674 fn key_lookup_tracks_mutations_sealing_and_adoption() {
2675 let mut segment = GrowingSegment::new(1, "keys");
2676 segment
2677 .bulk_insert(vec![keyed_row(1, "flat", 1, 0)])
2678 .expect("flat row");
2679 segment
2680 .insert(keyed_row(3, "hash", 1, 0))
2681 .expect("hash fallback");
2682 segment
2683 .bulk_insert(vec![keyed_row(5, "gap", 1, 0)])
2684 .expect("bulk gap");
2685 assert!(segment.key_lookup.as_ref().expect("index").memory_bytes() > 0);
2686 assert_eq!(keyed_probe(&segment, "gap"), Some(EntityId::new(5)));
2687 segment
2688 .update(keyed_row(3, "renamed", 2, 0))
2689 .expect("rename key");
2690 assert!(
2691 segment.key_lookup.is_some(),
2692 "key replacements maintain the index"
2693 );
2694 assert_eq!(keyed_probe(&segment, "hash"), None);
2695 assert_eq!(keyed_probe(&segment, "renamed"), Some(EntityId::new(3)));
2696
2697 *segment.get_mut(EntityId::new(3)).expect("mutable row") = keyed_row(3, "raw", 3, 0);
2698 assert!(
2699 segment.key_lookup.is_some(),
2700 "raw mutation preserves unaffected key candidates"
2701 );
2702 assert_eq!(keyed_probe(&segment, "renamed"), None);
2703 assert_eq!(keyed_probe(&segment, "raw"), Some(EntityId::new(3)));
2704 segment.seal().expect("seal");
2705 segment
2706 .force_update_with_metadata(&keyed_row(5, "sealed", 4, 0), &["key".into()], None)
2707 .expect("sealed rename");
2708 assert_eq!(keyed_probe(&segment, "gap"), None);
2709 assert_eq!(keyed_probe(&segment, "sealed"), Some(EntityId::new(5)));
2710 assert!(segment.force_delete(EntityId::new(1)));
2711 assert!(segment.force_delete(EntityId::new(3)));
2712 assert_eq!(keyed_probe(&segment, "flat"), None);
2713 assert_eq!(keyed_probe(&segment, "raw"), None);
2714
2715 let mut adopted = GrowingSegment::new(2, "keys");
2716 segment.for_each_fast(|entity| {
2717 adopted.adopt_entity(entity.clone(), None);
2718 true
2719 });
2720 assert_eq!(keyed_probe(&adopted, "sealed"), Some(EntityId::new(5)));
2721 assert!(adopted.evict_entity(EntityId::new(5)));
2722 assert_eq!(keyed_probe(&adopted, "sealed"), None);
2723 }
2724
2725 #[test]
2726 #[ignore = "diagnostic microbenchmark; no timing assertions"]
2727 fn key_lookup_scaling_diagnostic() {
2728 use std::{hint::black_box, time::Instant};
2729 for count in [100, 10_000, 100_000] {
2730 let mut segment = GrowingSegment::new(1, "keys");
2731 segment
2732 .bulk_insert(
2733 (1..=count)
2734 .map(|id| keyed_row(id, &format!("key-{id}"), 1, 0))
2735 .collect(),
2736 )
2737 .expect("rows");
2738 let key = format!("key-{count}");
2739 let started = Instant::now();
2740 for _ in 0..1000 {
2741 assert_eq!(
2742 keyed_probe(black_box(&segment), black_box(&key)),
2743 Some(EntityId::new(count))
2744 );
2745 }
2746 let indexed = started.elapsed();
2747 let started = Instant::now();
2748 for _ in 0..1000 {
2749 let mut found = None;
2750 segment.for_each_fast(|entity| {
2751 if SegmentKeyLookup::key(entity) == Some(black_box(key.as_str())) {
2752 found = Some(entity.id);
2753 return false;
2754 }
2755 true
2756 });
2757 assert_eq!(black_box(found), Some(EntityId::new(count)));
2758 }
2759 eprintln!(
2760 "rows={count} reads=1000 indexed={indexed:?} scan={:?}",
2761 started.elapsed()
2762 );
2763 }
2764 }
2765
2766 fn pruning_node(id: u64) -> UnifiedEntity {
2767 UnifiedEntity::graph_node(EntityId::new(id), "mixed", "person", Default::default())
2768 }
2769
2770 fn pruning_edge(id: u64) -> UnifiedEntity {
2771 UnifiedEntity::graph_edge(
2772 EntityId::new(id),
2773 "mixed",
2774 "a",
2775 "b",
2776 1.0,
2777 Default::default(),
2778 )
2779 }
2780
2781 #[test]
2782 fn graph_presence_covers_bulk_gaps_inserts_and_adoption() {
2783 let mut segment = GrowingSegment::new(1, "mixed");
2784 assert!(!segment.may_contain_graph_kind(GraphEntityKind::Node));
2785 assert!(!segment.may_contain_graph_kind(GraphEntityKind::Edge));
2786 segment
2787 .bulk_insert(vec![
2788 UnifiedEntity::vector(EntityId::new(1), "mixed", vec![]),
2789 pruning_node(2),
2790 pruning_edge(4),
2791 ])
2792 .expect("mixed bulk with gap");
2793 assert!(segment.may_contain_graph_kind(GraphEntityKind::Node));
2794 assert!(segment.may_contain_graph_kind(GraphEntityKind::Edge));
2795 segment.seal().expect("seal");
2796 assert!(segment.force_delete(EntityId::new(2)));
2797 assert!(segment.force_delete(EntityId::new(4)));
2798 assert!(segment.may_contain_graph_kind(GraphEntityKind::Node));
2800 let mut rebuilt = GrowingSegment::new(2, "mixed");
2801 for item in segment.iter() {
2802 rebuilt.adopt_entity(item.clone(), None);
2803 }
2804 assert!(!rebuilt.may_contain_graph_kind(GraphEntityKind::Node));
2805 assert!(!rebuilt.may_contain_graph_kind(GraphEntityKind::Edge));
2806 rebuilt.adopt_entity(pruning_node(5), None);
2807 assert!(rebuilt.may_contain_graph_kind(GraphEntityKind::Node));
2808 rebuilt.insert(pruning_edge(6)).expect("single edge");
2809 assert!(rebuilt.may_contain_graph_kind(GraphEntityKind::Edge));
2810 }
2811
2812 #[test]
2813 fn graph_presence_covers_replacement_and_unrestricted_mutation() {
2814 for flat in [false, true] {
2815 for operation in 0..4 {
2816 let mut segment = GrowingSegment::new(1, "mixed");
2817 let vector = UnifiedEntity::vector(EntityId::new(1), "mixed", vec![]);
2818 if flat {
2819 segment.bulk_insert(vec![vector]).expect("flat vector");
2820 } else {
2821 segment.insert(vector).expect("hashmap vector");
2822 }
2823 match operation {
2824 0 => segment.update(pruning_node(1)).expect("replace with node"),
2825 1 => segment
2826 .update_hot(pruning_node(1), &[])
2827 .expect("hot replacement"),
2828 2 => {
2829 segment.seal().expect("seal vector");
2830 segment
2831 .force_update_with_metadata(&pruning_node(1), &[], None)
2832 .expect("sealed replacement");
2833 }
2834 _ => {
2835 *segment.get_mut(EntityId::new(1)).expect("mutable vector") =
2836 pruning_node(1)
2837 }
2838 }
2839 assert!(segment.may_contain_graph_kind(GraphEntityKind::Node));
2840 assert!(matches!(
2841 segment.get(EntityId::new(1)).expect("node").kind,
2842 EntityKind::GraphNode(_)
2843 ));
2844 if operation == 3 {
2845 assert!(segment.may_contain_graph_kind(GraphEntityKind::Edge));
2846 }
2847 segment
2848 .force_update_with_metadata(&pruning_edge(1), &[], None)
2849 .expect("replace with edge");
2850 assert!(segment.may_contain_graph_kind(GraphEntityKind::Edge));
2851 assert!(segment.may_contain_graph_kind(GraphEntityKind::Node));
2852 }
2853 }
2854 }
2855
2856 fn primary_key_row(id: u64, key: Value) -> UnifiedEntity {
2857 UnifiedEntity::new(
2858 EntityId::new(id),
2859 EntityKind::TableRow {
2860 table: "keys".into(),
2861 row_id: id,
2862 },
2863 EntityData::Row(RowData::with_names(
2864 vec![Value::text("payload"), key],
2865 vec!["payload".into(), "primary".into()],
2866 )),
2867 )
2868 }
2869
2870 fn primary_key_probe(
2871 segment: &GrowingSegment,
2872 key: Value,
2873 exclude: Option<EntityId>,
2874 ) -> Option<EntityId> {
2875 segment.find_unique_key_conflict(
2876 &[vec!["primary".into()]],
2877 0,
2878 &[format!("{key:?}")],
2879 exclude,
2880 &|_| true,
2881 )
2882 }
2883
2884 #[test]
2885 fn unique_key_lookup_follows_segment_mutations_and_sealing() {
2886 let mut segment = GrowingSegment::new(1, "keys");
2887 segment
2888 .bulk_insert(vec![primary_key_row(1, Value::Integer(10))])
2889 .expect("bulk row");
2890 let payload_bytes = segment.memory_bytes();
2891 assert_eq!(
2892 primary_key_probe(&segment, Value::Integer(10), None),
2893 Some(EntityId::new(1))
2894 );
2895 assert!(segment.memory_bytes() > payload_bytes);
2896 segment
2897 .insert(primary_key_row(3, Value::Integer(30)))
2898 .expect("hashmap fallback row");
2899 segment
2900 .bulk_insert(vec![primary_key_row(4, Value::Integer(40))])
2901 .expect("bulk gap row");
2902 assert_eq!(
2903 primary_key_probe(&segment, Value::Integer(40), None),
2904 Some(EntityId::new(4))
2905 );
2906 assert_eq!(
2907 primary_key_probe(&segment, Value::Integer(30), None),
2908 Some(EntityId::new(3))
2909 );
2910 segment
2911 .update_hot(primary_key_row(1, Value::Integer(11)), &["primary".into()])
2912 .expect("replace flat key");
2913 assert_eq!(segment.unique_key_lookup_bytes.load(Ordering::Relaxed), 0);
2914 assert_eq!(primary_key_probe(&segment, Value::Integer(10), None), None);
2915 assert_eq!(
2916 primary_key_probe(&segment, Value::Integer(11), None),
2917 Some(EntityId::new(1))
2918 );
2919 segment
2920 .delete_batch(&[EntityId::new(1), EntityId::new(3)])
2921 .expect("delete flat and map");
2922 assert_eq!(primary_key_probe(&segment, Value::Integer(11), None), None);
2923 assert_eq!(primary_key_probe(&segment, Value::Integer(30), None), None);
2924 segment.seal().expect("seal");
2925 assert_eq!(
2926 primary_key_probe(&segment, Value::Integer(40), None),
2927 Some(EntityId::new(4))
2928 );
2929 segment
2930 .force_update_with_metadata(
2931 &primary_key_row(4, Value::Integer(41)),
2932 &["primary".into()],
2933 None,
2934 )
2935 .expect("sealed update");
2936 assert_eq!(primary_key_probe(&segment, Value::Integer(40), None), None);
2937 assert_eq!(
2938 primary_key_probe(&segment, Value::Integer(41), None),
2939 Some(EntityId::new(4))
2940 );
2941 assert!(segment.force_delete(EntityId::new(4)));
2942 assert_eq!(primary_key_probe(&segment, Value::Integer(41), None), None);
2943 }
2944
2945 #[test]
2946 fn unique_lookup_retains_all_current_definitions_and_rechecks_candidates() {
2947 let mut segment = GrowingSegment::new(1, "keys");
2948 segment
2949 .bulk_insert(vec![
2950 primary_key_row(1, Value::Integer(10)),
2951 primary_key_row(2, Value::Integer(20)),
2952 ])
2953 .expect("initial rows");
2954 let definitions = vec![
2955 vec!["primary".into()],
2956 vec!["payload".into()],
2957 vec!["primary".into(), "payload".into()],
2958 ];
2959 let payload = format!("{:?}", Value::text("payload"));
2960 assert_eq!(
2961 segment.find_unique_key_conflict(
2962 &definitions,
2963 1,
2964 &[payload.clone()],
2965 None,
2966 &|entity| entity.id != EntityId::new(1)
2967 ),
2968 Some(EntityId::new(2))
2969 );
2970 assert_eq!(segment.unique_key_lookup.read().len(), 3);
2971 segment
2972 .insert(primary_key_row(3, Value::Integer(30)))
2973 .expect("append");
2974 assert_eq!(
2975 segment.find_unique_key_conflict(
2976 &definitions,
2977 2,
2978 &[format!("{:?}", Value::Integer(30)), payload.clone()],
2979 None,
2980 &|_| true
2981 ),
2982 Some(EntityId::new(3))
2983 );
2984 let bytes = segment.unique_key_lookup_bytes.load(Ordering::Relaxed);
2985 assert_eq!(
2986 segment.find_unique_key_conflict(
2987 &definitions,
2988 0,
2989 &[format!("{:?}", Value::Integer(20))],
2990 None,
2991 &|_| true
2992 ),
2993 Some(EntityId::new(2))
2994 );
2995 assert_eq!(segment.unique_key_lookup.read().len(), 3);
2996 assert_eq!(
2997 segment.unique_key_lookup_bytes.load(Ordering::Relaxed),
2998 bytes
2999 );
3000 let reduced = vec![vec!["payload".into()]];
3001 assert_eq!(
3002 segment.find_unique_key_conflict(&reduced, 0, &[payload], None, &|_| true),
3003 Some(EntityId::new(1))
3004 );
3005 assert_eq!(segment.unique_key_lookup.read().len(), 1);
3006 assert!(segment.unique_key_lookup_bytes.load(Ordering::Relaxed) < bytes);
3007 }
3008
3009 #[test]
3010 fn unique_key_lookup_preserves_signature_semantics_and_exclusion() {
3011 let mut segment = GrowingSegment::new(1, "keys");
3012 let values = [
3013 Value::Float(f64::NAN),
3014 Value::Float(-0.0),
3015 Value::Float(0.0),
3016 Value::Integer(1),
3017 Value::UnsignedInteger(1),
3018 Value::text("1"),
3019 ];
3020 for (offset, value) in values.iter().enumerate() {
3021 let id = offset as u64 + 1;
3022 segment
3023 .insert(primary_key_row(id, value.clone()))
3024 .expect("distinct physical row");
3025 assert_eq!(
3026 primary_key_probe(&segment, value.clone(), None),
3027 Some(EntityId::new(id))
3028 );
3029 assert_eq!(
3030 primary_key_probe(&segment, value.clone(), Some(EntityId::new(id))),
3031 None
3032 );
3033 }
3034 segment
3035 .insert(primary_key_row(20, Value::Integer(1)))
3036 .expect("duplicate physical key");
3037 assert_eq!(
3038 primary_key_probe(&segment, Value::Integer(1), Some(EntityId::new(4))),
3039 Some(EntityId::new(20))
3040 );
3041 segment
3044 .entities
3045 .insert(EntityId::new(4), primary_key_row(4, Value::Integer(99)));
3046 assert_eq!(
3047 primary_key_probe(&segment, Value::Integer(1), Some(EntityId::new(20))),
3048 None
3049 );
3050 let entity = segment.get_mut(EntityId::new(4)).expect("mutable row");
3052 *entity = primary_key_row(4, Value::Integer(100));
3053 assert_eq!(
3054 primary_key_probe(&segment, Value::Integer(100), None),
3055 Some(EntityId::new(4))
3056 );
3057 }
3058
3059 #[test]
3060 fn test_growing_segment_basic() {
3061 let mut segment = GrowingSegment::new(1, "test");
3062
3063 let entity = UnifiedEntity::table_row(
3064 EntityId::new(1),
3065 "users",
3066 1,
3067 vec![Value::text("Alice".to_string())],
3068 );
3069
3070 let id = segment.insert(entity).unwrap();
3071 assert_eq!(id, EntityId::new(1));
3072 assert!(segment.contains(id));
3073
3074 let stats = segment.stats();
3075 assert_eq!(stats.entity_count, 1);
3076 assert_eq!(stats.row_count, 1);
3077 }
3078
3079 #[test]
3080 fn test_segment_metadata() {
3081 let mut segment = GrowingSegment::new(1, "test");
3082
3083 let entity = UnifiedEntity::table_row(
3084 EntityId::new(1),
3085 "users",
3086 1,
3087 vec![Value::text("Alice".to_string())],
3088 );
3089 segment.insert(entity).unwrap();
3090
3091 let mut meta = Metadata::new();
3092 meta.set("role", MetadataValue::String("admin".to_string()));
3093 meta.set("level", MetadataValue::Int(5));
3094
3095 segment.set_metadata(EntityId::new(1), meta).unwrap();
3096
3097 let retrieved = segment.get_metadata(EntityId::new(1)).unwrap();
3098 assert_eq!(
3099 retrieved.get("role"),
3100 Some(&MetadataValue::String("admin".to_string()))
3101 );
3102 }
3103
3104 #[test]
3105 fn test_segment_seal() {
3106 let mut segment = GrowingSegment::new(1, "test");
3107
3108 let entity = UnifiedEntity::vector(EntityId::new(1), "embeddings", vec![0.1, 0.2, 0.3]);
3109 segment.insert(entity).unwrap();
3110
3111 assert!(segment.state().is_writable());
3113
3114 segment.seal().unwrap();
3116 assert_eq!(segment.state(), SegmentState::Sealed);
3117
3118 let entity2 = UnifiedEntity::vector(EntityId::new(2), "embeddings", vec![0.4, 0.5, 0.6]);
3120 assert!(segment.insert(entity2).is_err());
3121 }
3122
3123 #[test]
3124 fn test_should_seal() {
3125 let mut segment = GrowingSegment::new(1, "test");
3126
3127 let config = SegmentConfig {
3128 max_entities: 2,
3129 ..Default::default()
3130 };
3131
3132 assert!(!segment.should_seal(&config));
3133
3134 segment
3135 .insert(UnifiedEntity::vector(EntityId::new(1), "v", vec![0.1]))
3136 .unwrap();
3137 assert!(!segment.should_seal(&config));
3138
3139 segment
3140 .insert(UnifiedEntity::vector(EntityId::new(2), "v", vec![0.2]))
3141 .unwrap();
3142 assert!(segment.should_seal(&config));
3143 }
3144
3145 #[test]
3146 fn test_cross_references() {
3147 let mut segment = GrowingSegment::new(1, "test");
3148
3149 let mut entity1 = UnifiedEntity::table_row(
3150 EntityId::new(1),
3151 "hosts",
3152 1,
3153 vec![Value::text("192.168.1.1".to_string())],
3154 );
3155 entity1.add_cross_ref(CrossRef::new(
3156 EntityId::new(1),
3157 EntityId::new(2),
3158 "nodes",
3159 RefType::RowToNode,
3160 ));
3161 segment.insert(entity1).unwrap();
3162
3163 let refs_from = segment.get_references_from(EntityId::new(1));
3164 assert_eq!(refs_from.len(), 1);
3165 assert_eq!(refs_from[0], (EntityId::new(2), RefType::RowToNode));
3166
3167 let refs_to = segment.get_references_to(EntityId::new(2));
3168 assert_eq!(refs_to.len(), 1);
3169 assert_eq!(refs_to[0], (EntityId::new(1), RefType::RowToNode));
3170 }
3171
3172 #[test]
3173 fn test_zone_predicate_uses_canonical_fallback_for_email_values() {
3174 let mut zone = ColZone::new(Value::Email("bravo@example.com".to_string()));
3175 zone.update(&Value::Email("delta@example.com".to_string()));
3176
3177 let probe = Value::Email("alpha@example.com".to_string());
3178 assert!(ZoneColPred::Eq(&probe).can_skip(&zone));
3179
3180 let in_range = Value::Email("charlie@example.com".to_string());
3181 assert!(!ZoneColPred::Eq(&in_range).can_skip(&zone));
3182 }
3183
3184 #[test]
3185 fn test_sealed_multi_zone_prunes_numeric_gap_outlier() {
3186 let mut segment = GrowingSegment::new(1, "test");
3187
3188 for (row_id, age) in [(1_u64, 1_i64), (2, 2), (3, 3), (4, 1000)] {
3189 let entity = UnifiedEntity::new(
3190 EntityId::new(row_id),
3191 EntityKind::TableRow {
3192 table: "users".into(),
3193 row_id,
3194 },
3195 EntityData::Row(RowData::with_names(
3196 vec![Value::Integer(age)],
3197 vec!["age".to_string()],
3198 )),
3199 );
3200 segment.insert(entity).unwrap();
3201 }
3202
3203 segment.seal().unwrap();
3204
3205 let miss = Value::Integer(500);
3206 assert!(segment.can_skip_zone_preds(&[("age", ZoneColPred::Eq(&miss))]));
3207
3208 let hit = Value::Integer(1000);
3209 assert!(!segment.can_skip_zone_preds(&[("age", ZoneColPred::Eq(&hit))]));
3210 }
3211}