Skip to main content

reddb_server/storage/unified/
segment.rs

1//! Unified Segment System
2//!
3//! Implements the Growing → Sealed segment lifecycle pattern. Segments are the
4//! fundamental unit of storage that handle entities of all types.
5//!
6//! # Write Path
7//!
8//! 1. WAL write (durability)
9//! 2. Growing-segment entity storage in RAM (HashMap mode or flat-vector mode
10//!    with epoch-published lock-free reads)
11//! 3. `seal()` freezes the growing segment into a Sealed segment and builds
12//!    pruning summaries over the immutable data
13//!
14//! # Read Path
15//!
16//! 1. Growing segment entity storage (most recent writes)
17//! 2. Sealed segments (older, immutable, zone-map guarded)
18//!
19//! # Lifecycle
20//!
21//! ```text
22//! Growing (in-memory, accepts writes)
23//!    ↓ seal() when full or manually triggered
24//! Sealed (immutable, zone maps built)
25//!    ↓ flush() for persistence
26//! Flushed (on disk, can be mmap'd)
27//!    ↓ archive() for cold storage
28//! Archived (compressed, infrequently accessed)
29//! ```
30
31use 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
43/// Unique identifier for a segment
44pub type SegmentId = u64;
45
46/// Segment state in its lifecycle
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum SegmentState {
49    /// Accepts writes, partial/no index
50    Growing,
51    /// Transitioning, building indices
52    Sealing,
53    /// Immutable, fully indexed
54    Sealed,
55    /// Persisted to disk
56    Flushed,
57    /// Compressed, cold storage
58    Archived,
59}
60
61impl SegmentState {
62    /// Check if segment accepts writes
63    pub fn is_writable(&self) -> bool {
64        matches!(self, Self::Growing)
65    }
66
67    /// Check if segment is queryable
68    pub fn is_queryable(&self) -> bool {
69        !matches!(self, Self::Sealing)
70    }
71
72    /// Check if segment is immutable
73    pub fn is_immutable(&self) -> bool {
74        matches!(self, Self::Sealed | Self::Flushed | Self::Archived)
75    }
76}
77
78/// Configuration for segments
79#[derive(Debug, Clone)]
80pub struct SegmentConfig {
81    /// Maximum entities before auto-sealing
82    pub max_entities: usize,
83    /// Maximum memory bytes before auto-sealing
84    pub max_bytes: usize,
85    /// Maximum age in seconds before auto-sealing
86    pub max_age_secs: u64,
87    /// Enable vector indexing when sealed
88    pub build_vector_index: bool,
89    /// Enable graph indexing when sealed
90    pub build_graph_index: bool,
91    /// Compression level for archived segments (0-9)
92    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, // 256 MB
100            max_age_secs: 3600,           // 1 hour
101            build_vector_index: true,
102            build_graph_index: true,
103            compression_level: 6,
104        }
105    }
106}
107
108/// Segment statistics
109#[derive(Debug, Clone, Default)]
110pub struct SegmentStats {
111    /// Number of entities
112    pub entity_count: usize,
113    /// Number of deleted entities
114    pub deleted_count: usize,
115    /// Approximate memory usage in bytes
116    pub memory_bytes: usize,
117    /// Number of vectors
118    pub vector_count: usize,
119    /// Number of graph nodes
120    pub node_count: usize,
121    /// Number of graph edges
122    pub edge_count: usize,
123    /// Number of table rows
124    pub row_count: usize,
125    /// Number of cross-references
126    pub cross_ref_count: usize,
127}
128
129/// Segment error types
130#[derive(Debug, Clone)]
131pub enum SegmentError {
132    /// Segment is not writable
133    NotWritable,
134    /// Entity not found
135    NotFound(EntityId),
136    /// Entity already exists
137    AlreadyExists(EntityId),
138    /// Segment is full
139    Full,
140    /// Invalid operation for current state
141    InvalidState(SegmentState),
142    /// Internal error
143    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
181/// Approximate cost of one entry in the `deleted` tombstone map: the entity
182/// id and deletion ordinal plus hashbrown's control byte and load-factor slack.
183///
184/// Tombstones outlive the entity they bury — a sealed segment holds its
185/// `deleted` map for as long as the segment lives. That map is the memory
186/// consolidation reclaims even in HashMap mode, where the entity body itself
187/// was already freed by `force_delete`.
188const 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
221/// A unified segment that stores all entity types
222pub trait UnifiedSegment: Send + Sync {
223    /// Get segment ID
224    fn id(&self) -> SegmentId;
225
226    /// Get current state
227    fn state(&self) -> SegmentState;
228
229    /// Get collection/namespace name
230    fn collection(&self) -> &str;
231
232    /// Get statistics
233    fn stats(&self) -> SegmentStats;
234
235    /// O(1) live entity count (entities minus tombstones)
236    fn entity_count(&self) -> usize;
237
238    /// Check if entity exists
239    fn contains(&self, id: EntityId) -> bool;
240
241    /// Get an entity by ID
242    fn get(&self, id: EntityId) -> Option<&UnifiedEntity>;
243
244    /// Get mutable reference to entity
245    fn get_mut(&mut self, id: EntityId) -> Option<&mut UnifiedEntity>;
246
247    /// Insert a new entity
248    fn insert(&mut self, entity: UnifiedEntity) -> Result<EntityId, SegmentError>;
249
250    /// Update an existing entity
251    fn update(&mut self, entity: UnifiedEntity) -> Result<(), SegmentError>;
252
253    /// HOT-update: like update but receives the set of field names that actually
254    /// changed. Allows skipping index work when indexed columns are unaffected.
255    /// Default: falls back to full update.
256    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    /// Delete an entity
266    fn delete(&mut self, id: EntityId) -> Result<bool, SegmentError>;
267
268    /// Get metadata for an entity
269    fn get_metadata(&self, id: EntityId) -> Option<Metadata>;
270
271    /// Set metadata for an entity
272    fn set_metadata(&mut self, id: EntityId, metadata: Metadata) -> Result<(), SegmentError>;
273
274    /// Seal the segment (make immutable)
275    fn seal(&mut self) -> Result<(), SegmentError>;
276
277    /// Check if should auto-seal based on config
278    fn should_seal(&self, config: &SegmentConfig) -> bool;
279
280    /// Iterate over all entities
281    fn iter(&self) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_>;
282
283    /// Iterate over entities of a specific kind
284    fn iter_kind(&self, kind_filter: &str) -> Box<dyn Iterator<Item = &UnifiedEntity> + '_>;
285
286    /// Search entities by metadata filter
287    fn filter_metadata(
288        &self,
289        filters: &[(String, super::metadata::MetadataFilter)],
290    ) -> Vec<EntityId>;
291}
292
293// ─────────────────────────────────────────────────────────────────────────────
294// Zone map: per-column min/max for segment pruning
295// ─────────────────────────────────────────────────────────────────────────────
296
297/// Tracks the min and max observed `Value` for one column in a segment.
298/// Used to skip segments that cannot satisfy a range or equality predicate.
299#[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/// Tag-only variant of `ZoneColPred` — used where the Value is stored separately.
386#[derive(Debug, Clone, Copy, PartialEq, Eq)]
387pub enum ZoneColPredKind {
388    Eq,
389    Gt,
390    Gte,
391    Lt,
392    Lte,
393}
394
395/// A predicate on a single column that can be checked against a `ColZone`.
396#[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    /// Returns `true` when the entire segment can be skipped (no row can match).
407    pub fn can_skip(&self, zone: &ColZone) -> bool {
408        match self {
409            // Equality: skip if val < min OR val > max
410            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            // col > val: skip if max <= val (all rows have col ≤ val, none > val)
419            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            // col >= val: skip if max < val
425            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            // col < val: skip if min >= val
431            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            // col <= val: skip if min > val
437            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/// The two physical kinds used by graph materialization passes.
447#[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
469/// Maximum physical positions examined under one segment read guard.
470pub(crate) const SCAN_BATCH_SIZE: usize = 256;
471
472/// Bounded ID provenance shared by segment journals and checkpoint projection.
473pub(crate) const CHECKPOINT_MUTATIONS_MAX: usize = 8192;
474
475/// Stable append-only positions, bounded when a scan opens the segment.
476pub(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
493/// One indexed key at a time, with an exclusive resume ID and a source-wide
494/// upper bound captured before any batch. MVCC is checked during live hydration.
495pub(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
507/// Growing segment implementation (in-memory, writable)
508pub struct GrowingSegment {
509    /// Segment ID
510    id: SegmentId,
511    /// Collection/namespace name
512    collection: String,
513    /// Current state
514    state: SegmentState,
515    /// Creation timestamp
516    created_at: u64,
517    /// Last write timestamp
518    last_write_at: u64,
519
520    /// Entity storage (HashMap for random access)
521    entities: HashMap<EntityId, UnifiedEntity>,
522    /// Append-only positions for HashMap-resident IDs. Deleted entries stay
523    /// until the segment is reclaimed; flat storage already has stable positions.
524    hash_scan_ids: Vec<EntityId>,
525    /// Flat entity storage for bulk inserts (no HashMap overhead, O(1) by offset)
526    /// Used when entity IDs are sequential from base_entity_id
527    flat_entities: Vec<UnifiedEntity>,
528    /// Base entity ID for flat_entities (flat_entities[0].id == base_entity_id)
529    base_entity_id: u64,
530    /// Whether flat storage is active (bulk insert mode)
531    use_flat: bool,
532    /// Deleted entity IDs (tombstones)
533    // Ordinals never change or disappear during the segment lifetime. Cursors
534    // skip old tombstones but keep IDs relocated after their capture boundary.
535    deleted: HashMap<EntityId, usize>,
536    /// Metadata storage (type-aware)
537    metadata: MetadataStorage,
538
539    /// Primary key index: (collection, pk_value) → EntityId
540    pk_index: BTreeMap<(String, String), EntityId>,
541    /// Lazy schema-aware unique-key lookups, independent of the legacy
542    /// first-column index. Protected by the owning segment's entity lock.
543    unique_key_lookup: parking_lot::RwLock<Vec<SegmentUniqueKey>>,
544    unique_key_lookup_bytes: AtomicU64,
545    key_lookup: Option<SegmentKeyLookup>,
546    /// Type index: kind → EntityIds
547    kind_index: HashMap<String, HashSet<EntityId>>,
548    /// Conservative physical presence, independent of kind_index and MVCC.
549    /// Bits only accumulate; unrestricted mutable access sets both bits.
550    graph_kind_mask: u8,
551    /// Retains every physical graph version until physical reclamation.
552    graph_read_index: parking_lot::RwLock<Option<SegmentGraphIndex>>,
553    graph_read_index_bytes: AtomicU64,
554    /// Cross-reference index: source → Vec<(target, ref_type)>
555    cross_ref_forward: HashMap<EntityId, Vec<(EntityId, RefType)>>,
556    /// Reverse cross-reference index: target → Vec<(source, ref_type)>
557    cross_ref_reverse: HashMap<EntityId, Vec<(EntityId, RefType)>>,
558
559    /// Per-column zone maps: col_name → (min, max) for segment pruning
560    col_zones: HashMap<String, ColZone>,
561    zone_payload_bytes: usize,
562    /// Sealed-only minmax-multi summaries built from canonical ordering.
563    sealed_col_zones: HashMap<String, MultiColZone>,
564
565    /// Sequence counter for ordering
566    sequence: AtomicU64,
567    /// Volatile checkpoint invalidation; never part of the storage format.
568    checkpoint_revision: u64,
569    checkpoint_mutations: std::collections::VecDeque<(u64, EntityId)>,
570    /// Approximate payload bytes of the entities physically resident in this
571    /// segment. Resident means "still occupying memory", which includes
572    /// tombstoned entities in flat mode (their slot must stay so positional
573    /// lookup by `id - base_entity_id` keeps working).
574    memory_bytes: AtomicU64,
575    /// Payload bytes of the tombstoned entities that are still resident.
576    /// `memory_bytes - dead_entity_bytes` is the live payload.
577    dead_entity_bytes: u64,
578    /// Count of tombstoned entities that are still resident.
579    dead_resident_count: usize,
580
581    /// Epoch counter for lock-free reads of `flat_entities`.
582    ///
583    /// Updated with `Release` ordering after every flat-mode insert so that
584    /// readers can safely access `flat_entities[0..published_flat_len]` by
585    /// loading with `Acquire` ordering, without holding the segment RwLock.
586    /// Only meaningful when `use_flat == true`; always 0 in HashMap mode.
587    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    /// Append-only slots survive insertions, deletions and sealing. The caller
613    /// owns the segment read guard and must not re-enter storage in `visit`.
614    /// IDs must be hydrated through the manager: retired sources can be stale.
615    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                // Unpublished consolidation eviction leaves an unused position
637                // without a tombstone. It must not become a candidate later.
638                None => self.get(id).is_some(),
639            };
640            if eligible {
641                visit(id);
642            }
643        }
644        true
645    }
646
647    /// Direct iteration without Box<dyn> trait dispatch. Returns false to stop early.
648    /// Uses concrete iterator types to avoid heap allocation per call.
649    #[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            // Sequential Vec — best cache locality
656            if self.deleted.is_empty() {
657                for entity in &self.flat_entities {
658                    if !f(entity) {
659                        return false;
660                    }
661                }
662                // Also walk the HashMap — entities added via per-row
663                // `insert()` after flat mode was initialized land there.
664                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            // HashMap values — random order, no boxing
689            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    /// Create a new growing segment
710    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    /// Get next sequence number
750    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            // Same fallback as `get()` — entities added via `insert()`
810            // after flat mode was initialized live only in the HashMap.
811            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        // Record before replacement, including forced sealed updates. Failed
848        // updates may leave a false positive, never a false negative.
849        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    /// Preserve the first removal order, including repeated flat deletes.
944    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    /// Tombstone a flat-mode entity in place. Its slot stays so that positional
957    /// lookup by `id - base_entity_id` keeps working, so its payload keeps
958    /// costing memory until consolidation merges the segment away.
959    ///
960    /// Returns `false` when the entity was already tombstoned.
961    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    /// Remove a HashMap-resident entity. The payload is freed immediately; only
982    /// the tombstone survives, and consolidation reclaims that.
983    ///
984    /// Returns `false` when the entity is not present.
985    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    /// Update memory estimate
1041    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    /// This segment's approximate resident bytes. One relaxed load — the
1054    /// arena's contribution to the shared accounting pool (ADR 0073 §2).
1055    /// Unlike `stats()` this never walks the entity map, so a stats reader
1056    /// does not pay `O(entities)` to observe memory.
1057    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    /// Release a resident entity's payload from the memory estimate. Only for
1070    /// paths that physically drop the entity (HashMap removal, eviction).
1071    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    /// Entities physically present, live or tombstoned-but-resident.
1088    pub(crate) fn resident_entity_count(&self) -> usize {
1089        self.flat_entities.len() + self.entities.len()
1090    }
1091
1092    /// Entities visible to readers.
1093    pub(crate) fn live_entity_count(&self) -> usize {
1094        self.resident_entity_count()
1095            .saturating_sub(self.dead_resident_count)
1096    }
1097
1098    /// Tombstones this segment carries — including the ones whose entity body
1099    /// was already freed. The set itself is what leaks without consolidation.
1100    pub(crate) fn tombstone_count(&self) -> usize {
1101        self.deleted.len()
1102    }
1103
1104    /// Approximate bytes this segment holds: resident entity payloads plus the
1105    /// tombstone set.
1106    pub(crate) fn resident_bytes(&self) -> u64 {
1107        self.memory_bytes() + self.deleted.len() as u64 * TOMBSTONE_ENTRY_BYTES
1108    }
1109
1110    /// Bytes a consolidation of this segment would return to the budget: the
1111    /// payloads of resident tombstones plus the tombstone set that buries them.
1112    pub(crate) fn reclaimable_bytes(&self) -> u64 {
1113        self.dead_entity_bytes + self.deleted.len() as u64 * TOMBSTONE_ENTRY_BYTES
1114    }
1115
1116    /// Snapshot the ids of every live entity, in iteration order.
1117    ///
1118    /// Consolidation copies a segment across several maintenance ticks, so it
1119    /// needs a stable cursor into the segment that survives concurrent
1120    /// tombstoning. Ids are stable; iteration position is not.
1121    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    /// Move an existing entity into a writable segment without assigning a
1131    /// new sequence identity. A pending update's preimage must still match
1132    /// its durable checkpoint after the MVCC delete stamp is normalized.
1133    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    /// Take an entity from another segment, preserving its identity.
1150    ///
1151    /// Unlike `insert`, this keeps `sequence_id` — the merged segment is the
1152    /// same rows, not new ones, and callers (`physical_collection_roots`, MVCC
1153    /// ordering) read that field. All derived structures are rebuilt: kind
1154    /// index, primary-key index, cross-reference forward/reverse indexes, and
1155    /// per-column zone maps.
1156    pub(crate) fn adopt_entity(&mut self, entity: UnifiedEntity, metadata: Option<Metadata>) {
1157        let id = entity.id;
1158
1159        // Keep the sequence counter ahead of every adopted entity, so a later
1160        // insert into this segment cannot re-issue a live sequence number.
1161        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    /// Physically remove an entity without leaving a tombstone.
1179    ///
1180    /// Only legal on a segment nobody can see yet — the half-built merged
1181    /// segment, when the swap discovers a copied row was deleted from its
1182    /// source mid-consolidation. A tombstone here would defeat the whole point.
1183    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    /// Estimate memory for an entity
1201    fn estimate_entity_size(entity: &UnifiedEntity) -> usize {
1202        super::memory_size::entity_bytes(entity)
1203    }
1204
1205    /// Update per-column zone maps from a newly inserted entity's fields.
1206    ///
1207    /// Handles both insert paths:
1208    /// - **Named** (`row.named`): individual inserts where fields are a `HashMap<String, Value>`
1209    /// - **Positional** (`row.columns` + `row.schema`): bulk-inserted entities stored as `Vec<Value>`
1210    ///   keyed by the shared schema. Previously this path was silently skipped, meaning zone maps
1211    ///   were always empty for bulk-loaded tables and segment pruning never fired.
1212    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    /// Returns `true` when this segment can be entirely skipped for the given predicates.
1335    /// A segment is skipped only if ALL predicates say so (conservative: any non-skippable
1336    /// predicate forces the scan to proceed).
1337    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; // ONE predicate suffices to skip
1351                }
1352            }
1353        }
1354        false
1355    }
1356
1357    /// Index an entity
1358    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        // Kind index
1382        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        // Primary key index (if applicable)
1389        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        // Cross-reference indices
1398        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    /// The owning segment read lock keeps candidates stable. Callers must not
1417    /// re-enter storage in `visit`; collect IDs and evaluate RLS after unlocking.
1418    #[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                // Only unrestricted structural mutation invalidates the index.
1459                // A stopped rebuild is discarded, never published partially.
1460                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    /// Collect at most SCAN_BATCH_SIZE physical index entries without payload
1503    /// copies. Resume by key seek, never by rescanning an already consumed prefix.
1504    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    /// Probe one declared UNIQUE/PRIMARY KEY against physical rows. Cache all
1559    /// current definitions together so alternating constraints never rebuilds
1560    /// the same segment. Schema changes replace the set; mutations invalidate it.
1561    /// The caller owns MVCC conflict policy and null-key admission.
1562    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    /// Only matching-key candidates reach `visible`; no result vector or row clones.
1646    /// Raw mutable rows are rechecked alongside indexed candidates.
1647    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        // Writers may temporarily take the index while updating its candidates,
1654        // but publish it again before releasing their exclusive segment guard.
1655        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            // Fingerprint collisions must never become logical-key matches.
1691            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    /// Check whether an early-exit probe key exists in this segment.
1705    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    /// Remove entity from indices
1720    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        // Kind index
1730        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        // Primary key index
1736        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        // Cross-reference indices
1745        self.cross_ref_forward.remove(&entity.id);
1746        // Note: reverse refs from this entity still need cleanup
1747    }
1748
1749    /// Selective re-index for updates.
1750    ///
1751    /// Skips index work that is not needed:
1752    /// - `kind_index`: entity kind is immutable, so remove+reinsert is always
1753    ///   a no-op; we skip it entirely and keep the existing entry.
1754    /// - `pk_index`: only updated when the primary-key column (first column of
1755    ///   a Row entity) actually changed. When `modified_columns` is provided,
1756    ///   we check membership; otherwise we compare old vs new pk value.
1757    /// - `cross_ref`: only rebuilt when the refs actually differ.
1758    fn reindex_for_update(
1759        &mut self,
1760        old: &UpdateIndexSnapshot,
1761        new: &UnifiedEntity,
1762        modified_columns: Option<&[String]>,
1763    ) {
1764        // kind_index: kind is immutable — the existing entry is already correct.
1765        // No remove + reinsert needed.
1766
1767        // pk_index: only update when pk column is touched
1768        let pk_changed = match &new.data {
1769            EntityData::Row(new_row) => {
1770                if let Some(cols) = modified_columns {
1771                    // Caller told us exactly what changed — check if first schema column modified
1772                    // pk is the first column; check by name against the schema or by position 0
1773                    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                        // No schema — fall back to value comparison
1782                        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            // Non-row types don't use pk_index
1789            _ => false,
1790        };
1791
1792        if pk_changed {
1793            // Remove old pk entry
1794            if let Some((collection, pk_str)) = &old.pk_index_key {
1795                self.pk_index.remove(&(collection.clone(), pk_str.clone()));
1796            }
1797            // Insert new pk entry
1798            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        // cross_ref: only rebuild when refs actually changed
1808        let new_refs = new.cross_refs();
1809        if old.cross_refs.as_slice() != new_refs {
1810            // Remove stale forward refs
1811            self.cross_ref_forward.remove(&new.id);
1812            // Prune stale entries from reverse index
1813            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            // Add new refs
1819            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    /// Get entities referencing the given entity
1833    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    /// Get entities referenced by the given entity
1838    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    /// Get age in seconds
1843    pub fn age_secs(&self) -> u64 {
1844        let now = current_unix_secs();
1845        now.saturating_sub(self.created_at)
1846    }
1847
1848    /// Get time since last write
1849    pub fn idle_secs(&self) -> u64 {
1850        let now = current_unix_secs();
1851        now.saturating_sub(self.last_write_at)
1852    }
1853
1854    /// Turbo bulk insert — minimal allocations per entity.
1855    ///
1856    /// Optimizations vs normal insert:
1857    /// - Skips cross-refs
1858    /// - Resolves kind_index once per consecutive run of the same storage type
1859    /// - Pre-allocates kind_index HashSet
1860    /// - Skips contains_key check (caller guarantees unique IDs)
1861    /// - Uses Relaxed ordering for sequence counter
1862    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        // A collection may receive a mixed-model batch. Index each item's actual
1901        // type while retaining batched lookup/reservation for homogeneous input.
1902        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        // Use flat storage (Vec) instead of HashMap — saves ~80 bytes/entity overhead
1934        if self.flat_entities.is_empty() && self.entities.is_empty() {
1935            // First bulk insert: initialize flat storage
1936            self.base_entity_id = entities.first().map(|e| e.id.raw()).unwrap_or(0);
1937            self.use_flat = true;
1938        }
1939
1940        // Collect zone-update values per column position, not per
1941        // (row, col). The old code did `col.to_string()` + HashMap
1942        // probe per cell — 15 cols × 25k rows = 375K String
1943        // allocations per bulk_insert call for the typed_insert bench.
1944        //
1945        // New shape: accumulate `Vec<Value>` per column index; after
1946        // the scan, walk each non-empty column exactly once, resolve
1947        // its name via the shared schema, and apply all observations
1948        // under a single HashMap entry. 375K allocs → ~ncols allocs.
1949        let mut columnar_zone_updates: Vec<Vec<Value>> = Vec::new();
1950        let mut columnar_schema: Option<std::sync::Arc<Vec<String>>> = None;
1951        // Named-row fallback (non-prevalidated path) still uses the
1952        // per-cell vec — these rows don't share a schema, so there's
1953        // no columnar shortcut.
1954        let mut named_zone_updates: Vec<(String, Value)> = Vec::new();
1955
1956        // Payload bytes for the whole batch, applied with one atomic add.
1957        // `bulk_insert` used to skip memory tracking entirely, which left the
1958        // budget blind to the rows that dominate a bulk-loaded store — and
1959        // blind to what consolidation reclaims from them.
1960        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                // Position-based `flat_entities` access in `get(id)`
1991                // assumes IDs are contiguous starting at
1992                // `base_entity_id`. When IDs ARE contiguous (the bulk
1993                // ingest hot path) this saves a HashMap probe per
1994                // lookup. But the global ID counter can jump between
1995                // bulk calls — CREATE INDEX, catalog DDL, even other
1996                // collections in the same store reserve IDs from the
1997                // same allocator. If we blindly `push` a gap entity
1998                // its actual position in the Vec no longer matches
1999                // `id - base`, and `flat_entities[id - base].id == id`
2000                // fails — the entity becomes invisible to `get(id)`.
2001                //
2002                // Route gap entities into the HashMap fallback
2003                // (`self.entities`). `get(id)` already falls through
2004                // there when the flat probe misses.
2005                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            // Fallback to HashMap for non-sequential inserts
2015            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        // Apply the accumulated zone updates.
2037        // Columnar path: one `col_zones.entry` call per column (not
2038        // per cell). Named-fallback path: unchanged.
2039        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                // Enter the HashMap once per column. `raw_entry_mut`
2049                // avoids the String::clone when the column already
2050                // exists in the map — but that's nightly-only, so we
2051                // pay one `clone()` per call-per-column (ncols total)
2052                // instead of the old ncols×nrows.
2053                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        // Publish the new flat length so lock-free readers can see the new entities.
2082        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    /// Delete from this segment regardless of its seal state.
2091    /// Used to mutate sealed segments when DELETE touches bulk-inserted entities.
2092    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    /// Update an entity in this segment regardless of its seal state.
2109    /// Used to mutate sealed segments when UPDATE touches bulk-inserted entities.
2110    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            // Fall through: once flat mode is initialized by a bulk_insert,
2177            // subsequent per-row `insert()` calls still write into the
2178            // HashMap (see the impl below) so reads must check both. Without
2179            // this fallback, a post-bulk single-row insert silently
2180            // disappears from `get()` / `query_all()`.
2181            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        // The caller can replace the kind through this reference. Conservatively
2201        // disable both graph pruning passes until a new segment is rebuilt.
2202        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            // Fall through to HashMap for entities inserted via
2217            // per-row `insert()` after flat mode was activated.
2218            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        // Assign sequence ID
2234        entity.sequence_id = self.next_sequence();
2235
2236        // Estimate and track memory
2237        let size = Self::estimate_entity_size(&entity);
2238        self.add_memory(size);
2239
2240        // Index the entity
2241        self.index_entity(&entity);
2242
2243        // Update column zone maps for range-based segment pruning
2244        self.update_col_zones_from_entity(&entity);
2245
2246        // Store
2247        let id = entity.id;
2248        self.entities.insert(id, entity);
2249
2250        // Update write timestamp
2251        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        // For flat storage, use tombstone (don't remove from Vec to keep indices valid)
2287        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        // Build indices on the sealed data:
2331        // - HNSW/IVF for vectors (future)
2332        // - B-tree for sorted access (future)
2333        // - Inverted index for text search (future)
2334        self.rebuild_sealed_col_zones();
2335
2336        self.state = SegmentState::Sealed;
2337        Ok(())
2338    }
2339
2340    fn should_seal(&self, config: &SegmentConfig) -> bool {
2341        // Check entity count
2342        if self.entities.len() >= config.max_entities {
2343            return true;
2344        }
2345
2346        // Check memory usage
2347        if self.memory_bytes.load(Ordering::Relaxed) as usize >= config.max_bytes {
2348            return true;
2349        }
2350
2351        // Check age
2352        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            // Chain flat entities (from bulk_insert) with any entities
2362            // that landed in the HashMap via per-row `insert()` after
2363            // flat mode was activated.
2364            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        // The iterator borrows the segment, so the index cannot change underneath
2377        // it. Borrow the ID set as well instead of copying O(kind count) entries.
2378        let ids = self.kind_index.get(kind_filter);
2379        // In flat mode entities live in `flat_entities`, not `entities` —
2380        // chain both so iter_kind doesn't drop bulk-inserted entities.
2381        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        // For growing segments, we iterate and filter. In flat mode the
2403        // IDs live in `flat_entities`, not `entities`, so chain both.
2404        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        // A bucket collision narrows candidates, but cannot change key equality.
2634        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        // Deletion must not erase presence; reconstruction may safely tighten it.
2799        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        // Candidate fingerprints are not equality proof: simulate a stale or
3042        // colliding entry by changing its row without the normal invalidation.
3043        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        // Public mutable access must invalidate before handing out the row.
3051        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        // Can write before sealing
3112        assert!(segment.state().is_writable());
3113
3114        // Seal the segment
3115        segment.seal().unwrap();
3116        assert_eq!(segment.state(), SegmentState::Sealed);
3117
3118        // Cannot write after sealing
3119        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}