reddb_server/storage/unified/store/
impl_file.rs

1use super::*;
2
3impl UnifiedStore {
4    pub fn new() -> Self {
5        Self::with_config(UnifiedStoreConfig::default())
6    }
7
8    /// Get the current storage format version
9    pub fn format_version(&self) -> u32 {
10        self.format_version.load(Ordering::SeqCst)
11    }
12
13    pub(crate) fn set_format_version(&self, version: u32) {
14        self.format_version.store(version, Ordering::SeqCst);
15    }
16
17    /// Allocate a global entity ID
18    pub fn next_entity_id(&self) -> EntityId {
19        EntityId::new(self.next_entity_id.fetch_add(1, Ordering::SeqCst))
20    }
21
22    /// Reserve `n` contiguous global entity IDs with one fetch_add.
23    /// Caller assigns `id = EntityId::new(start + i)` per entity.
24    pub fn reserve_entity_ids(&self, n: u64) -> std::ops::Range<u64> {
25        let start = self.next_entity_id.fetch_add(n, Ordering::SeqCst);
26        start..start + n
27    }
28
29    pub(crate) fn register_entity_id(&self, id: EntityId) {
30        let candidate = id.raw().saturating_add(1);
31        let mut current = self.next_entity_id.load(Ordering::SeqCst);
32        while candidate > current {
33            match self.next_entity_id.compare_exchange(
34                current,
35                candidate,
36                Ordering::SeqCst,
37                Ordering::SeqCst,
38            ) {
39                Ok(_) => break,
40                Err(updated) => current = updated,
41            }
42        }
43    }
44
45    /// Load store from binary file
46    ///
47    /// Binary dump framing is defined by `reddb_file::native_store`.
48    ///
49    /// Store payload shape:
50    /// ```text
51    /// [collection_count: varu32]
52    /// [collections...]
53    /// [cross_ref_count: varu32]
54    /// [cross_refs...]
55    /// ```
56    pub fn load_from_file(path: &Path) -> Result<Self, Box<dyn std::error::Error>> {
57        let file = File::open(path)?;
58        let mut reader = BufReader::new(file);
59        let mut buf = Vec::new();
60        reader.read_to_end(&mut buf)?;
61        Self::load_from_bytes(&buf)
62    }
63
64    pub(crate) fn load_from_bytes(buf: &[u8]) -> Result<Self, Box<dyn std::error::Error>> {
65        Self::load_from_bytes_with_config(buf, UnifiedStoreConfig::default())
66    }
67
68    pub(crate) fn load_from_bytes_with_config(
69        buf: &[u8],
70        config: UnifiedStoreConfig,
71    ) -> Result<Self, Box<dyn std::error::Error>> {
72        let mut buf = buf.to_vec();
73        let version =
74            reddb_file::decode_native_store_header(&buf).map_err(|err| err.to_string())?;
75        let mut pos = 8;
76
77        // V3+ has CRC32 footer — verify integrity before parsing
78        reddb_file::verify_native_store_crc32_footer(&mut buf, version)
79            .map_err(|err| err.to_string())?;
80
81        let store = Self::with_config(config);
82        store.set_format_version(version);
83
84        // Read collection count
85        let collection_count =
86            reddb_file::decode_native_dump_count(&buf, &mut pos).map_err(|e| e.to_string())?;
87
88        // Read each collection
89        for _ in 0..collection_count {
90            let collection_header =
91                reddb_file::decode_native_dump_collection_header(&buf, &mut pos)
92                    .map_err(|e| e.to_string())?;
93
94            // Read each entity — V7+ includes metadata alongside entity data.
95            for _ in 0..collection_header.entity_count {
96                if version >= STORE_VERSION_V7 {
97                    // Length-prefixed entity+metadata record (serialize_entity_record format)
98                    let record_bytes = reddb_file::decode_native_dump_entity_record(&buf, &mut pos)
99                        .map_err(|e| e.to_string())?;
100
101                    let (entity, metadata) = Self::deserialize_entity_record(record_bytes, version)
102                        .map_err(|e| format!("Entity record deserialization error: {e}"))?;
103
104                    store.insert_auto(&collection_header.name, entity.clone())?;
105
106                    if let Some(metadata) = metadata {
107                        if let Some(manager) = store.get_collection(&collection_header.name) {
108                            let _ = manager.set_metadata(entity.id, metadata);
109                        }
110                    }
111                } else {
112                    // V1–V6: entity only, no metadata
113                    let entity = Self::read_entity_binary(&buf, &mut pos, version)?;
114                    store.insert_auto(&collection_header.name, entity)?;
115                }
116            }
117        }
118
119        if pos < buf.len() {
120            // Read cross-references section
121            let cross_ref_count =
122                reddb_file::decode_native_dump_count(&buf, &mut pos).map_err(|e| e.to_string())?;
123
124            for _ in 0..cross_ref_count {
125                let cross_ref = reddb_file::decode_native_dump_cross_ref(&buf, &mut pos)
126                    .map_err(|e| e.to_string())?;
127                let ref_type = RefType::from_byte(cross_ref.ref_type);
128
129                let source_collection = store
130                    .get_any(EntityId::new(cross_ref.source_id))
131                    .map(|(name, _)| name)
132                    .unwrap_or_else(|| cross_ref.target_collection.clone());
133                let _ = store.add_cross_ref(
134                    &source_collection,
135                    EntityId::new(cross_ref.source_id),
136                    &cross_ref.target_collection,
137                    EntityId::new(cross_ref.target_id),
138                    ref_type,
139                    1.0,
140                );
141            }
142        }
143
144        // V10+: a trailing opaque auxiliary-metadata blob follows the
145        // cross-references. RedDB stores the collection contracts here so the
146        // single-file artifact carries each collection's declared model across
147        // a restart. Older dumps end at the cross-refs and have no blob.
148        if version >= reddb_file::STORE_VERSION_V10 && pos < buf.len() {
149            let aux = reddb_file::decode_native_dump_entity_record(&buf, &mut pos)
150                .map_err(|e| e.to_string())?;
151            if !aux.is_empty() {
152                store.set_aux_metadata(aux.to_vec());
153            }
154        }
155
156        if store.format_version() < STORE_VERSION_CURRENT {
157            store.set_format_version(STORE_VERSION_CURRENT);
158        }
159
160        Ok(store)
161    }
162
163    /// Save store to binary file
164    ///
165    /// Uses compact binary encoding with varint for efficient storage.
166    /// No JSON - pure binary with pages and indices.
167    pub fn save_to_file(&self, path: &Path) -> Result<(), Box<dyn std::error::Error>> {
168        let buf = self.to_binary_dump_bytes();
169        reddb_file::write_native_store_bytes_atomically(path, &buf)?;
170        Ok(())
171    }
172
173    pub(crate) fn to_binary_dump_bytes(&self) -> Vec<u8> {
174        let mut buf = Vec::new();
175
176        // Version 9 includes explicit table-row logical identity plus MVCC
177        // xmin/xmax alongside the V7 metadata envelope.
178        buf.extend_from_slice(&reddb_file::encode_native_store_header(
179            STORE_VERSION_CURRENT,
180        ));
181
182        // Get all collections
183        let collections = self.collections.read();
184        reddb_file::encode_native_dump_count(&mut buf, collections.len() as u32);
185
186        let fv = STORE_VERSION_CURRENT;
187        for (name, manager) in collections.iter() {
188            // Get all entities from this collection
189            let entities = manager.query_all(|_| true);
190            reddb_file::encode_native_dump_collection_header(&mut buf, name, entities.len() as u32);
191
192            // V7+: serialize entity+metadata as a length-prefixed record.
193            // Each record: [u32 len][serialize_entity_record bytes]
194            for entity in entities {
195                let metadata = manager.get_metadata(entity.id);
196                let record = Self::serialize_entity_record(&entity, metadata.as_ref(), fv);
197                reddb_file::encode_native_dump_entity_record(&mut buf, &record);
198            }
199        }
200
201        // Write cross-references
202        let cross_refs = self.cross_refs.read();
203        let total_refs: usize = cross_refs.values().map(|v| v.len()).sum();
204        reddb_file::encode_native_dump_count(&mut buf, total_refs as u32);
205
206        for (source_id, refs) in cross_refs.iter() {
207            for (target_id, ref_type, collection) in refs {
208                reddb_file::encode_native_dump_cross_ref(
209                    &mut buf,
210                    source_id.raw(),
211                    target_id.raw(),
212                    ref_type.to_byte(),
213                    collection,
214                );
215            }
216        }
217
218        // V10+: trailing opaque auxiliary-metadata blob (collection contracts
219        // for the single-file artifact). Always written — empty when unused —
220        // so the read side can unconditionally expect it for V10 dumps.
221        {
222            let aux = self.aux_metadata.read();
223            reddb_file::encode_native_dump_entity_record(&mut buf, &aux);
224        }
225
226        self.set_format_version(STORE_VERSION_CURRENT);
227
228        reddb_file::append_native_store_crc32_footer(&mut buf);
229        buf
230    }
231
232    /// Read entity from binary buffer
233    pub(crate) fn read_entity_binary(
234        buf: &[u8],
235        pos: &mut usize,
236        format_version: u32,
237    ) -> Result<UnifiedEntity, Box<dyn std::error::Error>> {
238        // Entity ID
239        let id = read_varu64(buf, pos).map_err(|e| format!("Failed to read entity id: {:?}", e))?;
240
241        // EntityKind type byte
242        let kind_type = buf[*pos];
243        *pos += 1;
244
245        // EntityKind details
246        let kind = match kind_type {
247            0 => {
248                // TableRow
249                let table_len = Self::read_varu32_safe(buf, pos)?;
250                let table = String::from_utf8(buf[*pos..*pos + table_len].to_vec())?;
251                *pos += table_len;
252                let row_id = Self::read_varu64_safe(buf, pos)?;
253                EntityKind::TableRow {
254                    table: table.into(),
255                    row_id,
256                }
257            }
258            1 => {
259                // GraphNode
260                let label_len = Self::read_varu32_safe(buf, pos)?;
261                let label = String::from_utf8(buf[*pos..*pos + label_len].to_vec())?;
262                *pos += label_len;
263                let node_type_len = Self::read_varu32_safe(buf, pos)?;
264                let node_type = String::from_utf8(buf[*pos..*pos + node_type_len].to_vec())?;
265                *pos += node_type_len;
266                EntityKind::GraphNode(Box::new(GraphNodeKind { label, node_type }))
267            }
268            2 => {
269                // GraphEdge
270                let label_len = Self::read_varu32_safe(buf, pos)?;
271                let label = String::from_utf8(buf[*pos..*pos + label_len].to_vec())?;
272                *pos += label_len;
273                let from_node_len = Self::read_varu32_safe(buf, pos)?;
274                let from_node = String::from_utf8(buf[*pos..*pos + from_node_len].to_vec())?;
275                *pos += from_node_len;
276                let to_node_len = Self::read_varu32_safe(buf, pos)?;
277                let to_node = String::from_utf8(buf[*pos..*pos + to_node_len].to_vec())?;
278                *pos += to_node_len;
279                let weight =
280                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
281                *pos += 4;
282                EntityKind::GraphEdge(Box::new(GraphEdgeKind {
283                    label,
284                    from_node,
285                    to_node,
286                    weight,
287                }))
288            }
289            3 => {
290                // Vector
291                let collection_len = Self::read_varu32_safe(buf, pos)?;
292                let collection = String::from_utf8(buf[*pos..*pos + collection_len].to_vec())?;
293                *pos += collection_len;
294                EntityKind::Vector { collection }
295            }
296            4 => {
297                // TimeSeriesPoint
298                let series_len = Self::read_varu32_safe(buf, pos)?;
299                let series = String::from_utf8(buf[*pos..*pos + series_len].to_vec())?;
300                *pos += series_len;
301                let metric_len = Self::read_varu32_safe(buf, pos)?;
302                let metric = String::from_utf8(buf[*pos..*pos + metric_len].to_vec())?;
303                *pos += metric_len;
304                EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind { series, metric }))
305            }
306            5 => {
307                // QueueMessage
308                let queue_len = Self::read_varu32_safe(buf, pos)?;
309                let queue = String::from_utf8(buf[*pos..*pos + queue_len].to_vec())?;
310                *pos += queue_len;
311                let position = Self::read_varu64_safe(buf, pos)?;
312                EntityKind::QueueMessage { queue, position }
313            }
314            _ => return Err(format!("Unknown EntityKind type: {}", kind_type).into()),
315        };
316
317        // EntityData type byte
318        let data_type = buf[*pos];
319        *pos += 1;
320
321        // EntityData
322        let mut data = match data_type {
323            0 => {
324                // Row
325                let col_count = Self::read_varu32_safe(buf, pos)?;
326                let mut columns = Vec::with_capacity(col_count);
327                for _ in 0..col_count {
328                    columns.push(Self::read_value_binary(buf, pos)?);
329                }
330                EntityData::Row(RowData::new(columns))
331            }
332            1 => {
333                // Node
334                let prop_count = Self::read_varu32_safe(buf, pos)?;
335                let mut properties = HashMap::new();
336                for _ in 0..prop_count {
337                    let key_len = Self::read_varu32_safe(buf, pos)?;
338                    let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
339                    *pos += key_len;
340                    let value = Self::read_value_binary(buf, pos)?;
341                    properties.insert(key, value);
342                }
343                EntityData::Node(NodeData::with_properties(properties))
344            }
345            2 => {
346                // Edge
347                let weight_bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
348                *pos += 4;
349                let weight = f32::from_le_bytes(weight_bytes);
350                let prop_count = Self::read_varu32_safe(buf, pos)?;
351                let mut properties = HashMap::new();
352                for _ in 0..prop_count {
353                    let key_len = Self::read_varu32_safe(buf, pos)?;
354                    let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
355                    *pos += key_len;
356                    let value = Self::read_value_binary(buf, pos)?;
357                    properties.insert(key, value);
358                }
359                let mut edge = EdgeData::new(weight);
360                edge.properties = properties;
361                EntityData::Edge(edge)
362            }
363            3 => {
364                // Vector
365                let dim = Self::read_varu32_safe(buf, pos)?;
366                let mut dense = Vec::with_capacity(dim);
367                for _ in 0..dim {
368                    let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
369                    *pos += 4;
370                    dense.push(f32::from_le_bytes(bytes));
371                }
372                let mut vector = VectorData::new(dense);
373                if format_version >= STORE_VERSION_V6 {
374                    let has_content = buf[*pos] != 0;
375                    *pos += 1;
376                    if has_content {
377                        let content_len = Self::read_varu32_safe(buf, pos)?;
378                        vector.content =
379                            Some(String::from_utf8(buf[*pos..*pos + content_len].to_vec())?);
380                        *pos += content_len;
381                    }
382                }
383                EntityData::Vector(vector)
384            }
385            4 => {
386                // TimeSeries
387                let metric_len = Self::read_varu32_safe(buf, pos)?;
388                let metric = String::from_utf8(buf[*pos..*pos + metric_len].to_vec())?;
389                *pos += metric_len;
390                let timestamp_ns = Self::read_varu64_safe(buf, pos)?;
391                let value_bytes = [
392                    buf[*pos],
393                    buf[*pos + 1],
394                    buf[*pos + 2],
395                    buf[*pos + 3],
396                    buf[*pos + 4],
397                    buf[*pos + 5],
398                    buf[*pos + 6],
399                    buf[*pos + 7],
400                ];
401                *pos += 8;
402                let mut tags = HashMap::new();
403                let series_id = if format_version >= STORE_VERSION_V11 {
404                    // Discriminator: interned series id (1) vs inline tag map (0).
405                    // Points interned through the series dictionary carry a series
406                    // id and no inline tags; paths that do not intern (metrics
407                    // remote-write, rollups) keep their tags inline.
408                    let interned = Self::read_varu32_safe(buf, pos)?;
409                    if interned == 1 {
410                        Some(Self::read_varu64_safe(buf, pos)?)
411                    } else {
412                        let tag_count = Self::read_varu32_safe(buf, pos)?;
413                        tags = HashMap::with_capacity(tag_count);
414                        for _ in 0..tag_count {
415                            let key_len = Self::read_varu32_safe(buf, pos)?;
416                            let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
417                            *pos += key_len;
418                            let value_len = Self::read_varu32_safe(buf, pos)?;
419                            let value = String::from_utf8(buf[*pos..*pos + value_len].to_vec())?;
420                            *pos += value_len;
421                            tags.insert(key, value);
422                        }
423                        None
424                    }
425                } else if format_version >= STORE_VERSION_V5 {
426                    let tag_count = Self::read_varu32_safe(buf, pos)?;
427                    tags = HashMap::with_capacity(tag_count);
428                    for _ in 0..tag_count {
429                        let key_len = Self::read_varu32_safe(buf, pos)?;
430                        let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
431                        *pos += key_len;
432                        let value_len = Self::read_varu32_safe(buf, pos)?;
433                        let value = String::from_utf8(buf[*pos..*pos + value_len].to_vec())?;
434                        *pos += value_len;
435                        tags.insert(key, value);
436                    }
437                    None
438                } else {
439                    None
440                };
441                EntityData::TimeSeries(crate::storage::unified::entity::TimeSeriesData {
442                    metric,
443                    series_id,
444                    timestamp_ns,
445                    value: f64::from_le_bytes(value_bytes),
446                    tags,
447                })
448            }
449            5 => {
450                // QueueMessage
451                let payload = Self::read_value_binary(buf, pos)?;
452                let enqueued_at_ns = Self::read_varu64_safe(buf, pos)?;
453                let attempts = Self::read_varu32_safe(buf, pos)? as u32;
454                EntityData::QueueMessage(crate::storage::unified::entity::QueueMessageData {
455                    payload,
456                    priority: None,
457                    enqueued_at_ns,
458                    attempts,
459                    max_attempts: 3,
460                    acked: false,
461                })
462            }
463            6 => {
464                // Row with named HashMap
465                let field_count = Self::read_varu32_safe(buf, pos)?;
466                let mut named = HashMap::with_capacity(field_count);
467                for _ in 0..field_count {
468                    let key_len = Self::read_varu32_safe(buf, pos)?;
469                    let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
470                    *pos += key_len;
471                    let value = Self::read_value_binary(buf, pos)?;
472                    named.insert(key, value);
473                }
474                EntityData::Row(RowData {
475                    columns: Vec::new(),
476                    named: Some(named),
477                    schema: None,
478                })
479            }
480            _ => return Err(format!("Unknown EntityData type: {}", data_type).into()),
481        };
482
483        // Timestamps
484        let created_at = Self::read_varu64_safe(buf, pos)?;
485        let updated_at = Self::read_varu64_safe(buf, pos)?;
486
487        // Embeddings count
488        let embedding_count = Self::read_varu32_safe(buf, pos)?;
489        let mut embeddings = Vec::with_capacity(embedding_count);
490        for _ in 0..embedding_count {
491            let name_len = Self::read_varu32_safe(buf, pos)?;
492            let name = String::from_utf8(buf[*pos..*pos + name_len].to_vec())?;
493            *pos += name_len;
494
495            let dim = Self::read_varu32_safe(buf, pos)?;
496            let mut vector = Vec::with_capacity(dim);
497            for _ in 0..dim {
498                let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
499                *pos += 4;
500                vector.push(f32::from_le_bytes(bytes));
501            }
502
503            let model_len = Self::read_varu32_safe(buf, pos)?;
504            let model = String::from_utf8(buf[*pos..*pos + model_len].to_vec())?;
505            *pos += model_len;
506
507            embeddings.push(EmbeddingSlot::new(name, vector, model));
508        }
509
510        // Cross-refs count
511        let cross_ref_count = Self::read_varu32_safe(buf, pos)?;
512        let mut cross_refs = Vec::with_capacity(cross_ref_count);
513        for _ in 0..cross_ref_count {
514            let source = Self::read_varu64_safe(buf, pos)?;
515            let target = Self::read_varu64_safe(buf, pos)?;
516            let ref_type_byte = buf[*pos];
517            *pos += 1;
518            let (target_collection, weight, created_at) = if format_version >= STORE_VERSION_V2 {
519                let coll_len = Self::read_varu32_safe(buf, pos)?;
520                let collection = String::from_utf8(buf[*pos..*pos + coll_len].to_vec())?;
521                *pos += coll_len;
522                let weight_bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
523                *pos += 4;
524                let weight = f32::from_le_bytes(weight_bytes);
525                let created_at = Self::read_varu64_safe(buf, pos)?;
526                (collection, weight, created_at)
527            } else {
528                (String::new(), 1.0, 0)
529            };
530
531            let mut cross_ref = CrossRef::new(
532                EntityId::new(source),
533                EntityId::new(target),
534                target_collection,
535                RefType::from_byte(ref_type_byte),
536            );
537            cross_ref.weight = weight;
538            cross_ref.created_at = created_at;
539            cross_refs.push(cross_ref);
540        }
541
542        // Sequence ID
543        let sequence_id = Self::read_varu64_safe(buf, pos)?;
544
545        if format_version >= STORE_VERSION_V4 {
546            if let EntityData::QueueMessage(message) = &mut data {
547                if *pos < buf.len() {
548                    let priority_present = buf[*pos] != 0;
549                    *pos += 1;
550                    message.priority = if priority_present {
551                        if *pos + 4 > buf.len() {
552                            return Err("truncated queue priority".into());
553                        }
554                        let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
555                        *pos += 4;
556                        Some(i32::from_le_bytes(bytes))
557                    } else {
558                        None
559                    };
560                    message.max_attempts = Self::read_varu32_safe(buf, pos)? as u32;
561                    if *pos >= buf.len() {
562                        return Err("truncated queue ack flag".into());
563                    }
564                    message.acked = buf[*pos] != 0;
565                    *pos += 1;
566                }
567            }
568        }
569
570        let mut entity = UnifiedEntity::new(EntityId::new(id), kind, data);
571        entity.created_at = created_at;
572        entity.updated_at = updated_at;
573        entity.sequence_id = sequence_id;
574        if format_version >= STORE_VERSION_V8 && *pos < buf.len() {
575            let has_logical_id = buf[*pos] != 0;
576            *pos += 1;
577            if has_logical_id {
578                let logical_id = Self::read_varu64_safe(buf, pos)?;
579                entity.set_logical_id(EntityId::new(logical_id));
580            }
581        }
582        if format_version >= STORE_VERSION_V9 && *pos < buf.len() {
583            let xmin = Self::read_varu64_safe(buf, pos)?;
584            let xmax = Self::read_varu64_safe(buf, pos)?;
585            entity.set_xmin(xmin);
586            entity.set_xmax(xmax);
587        }
588        if !embeddings.is_empty() || !cross_refs.is_empty() {
589            entity.embeddings_mut().extend(embeddings);
590            entity.cross_refs_mut().extend(cross_refs);
591        }
592
593        Ok(entity)
594    }
595
596    /// Safe varu32 reader that converts DecodeError to Box<dyn Error>
597    fn read_varu32_safe(buf: &[u8], pos: &mut usize) -> Result<usize, Box<dyn std::error::Error>> {
598        read_varu32(buf, pos)
599            .map(|v| v as usize)
600            .map_err(|e| format!("Decode error: {:?}", e).into())
601    }
602
603    /// Safe varu64 reader that converts DecodeError to Box<dyn Error>
604    fn read_varu64_safe(buf: &[u8], pos: &mut usize) -> Result<u64, Box<dyn std::error::Error>> {
605        read_varu64(buf, pos).map_err(|e| format!("Decode error: {:?}", e).into())
606    }
607
608    /// Write entity to binary buffer
609    pub(crate) fn write_entity_binary(
610        buf: &mut Vec<u8>,
611        entity: &UnifiedEntity,
612        format_version: u32,
613    ) {
614        // Entity ID
615        write_varu64(buf, entity.id.raw());
616
617        // EntityKind
618        match &entity.kind {
619            EntityKind::TableRow { table, row_id } => {
620                buf.push(0);
621                write_varu32(buf, table.len() as u32);
622                buf.extend_from_slice(table.as_bytes());
623                write_varu64(buf, *row_id);
624            }
625            EntityKind::GraphNode(ref node) => {
626                buf.push(1);
627                write_varu32(buf, node.label.len() as u32);
628                buf.extend_from_slice(node.label.as_bytes());
629                write_varu32(buf, node.node_type.len() as u32);
630                buf.extend_from_slice(node.node_type.as_bytes());
631            }
632            EntityKind::GraphEdge(ref edge) => {
633                buf.push(2);
634                write_varu32(buf, edge.label.len() as u32);
635                buf.extend_from_slice(edge.label.as_bytes());
636                write_varu32(buf, edge.from_node.len() as u32);
637                buf.extend_from_slice(edge.from_node.as_bytes());
638                write_varu32(buf, edge.to_node.len() as u32);
639                buf.extend_from_slice(edge.to_node.as_bytes());
640                buf.extend_from_slice(&edge.weight.to_le_bytes());
641            }
642            EntityKind::Vector { collection } => {
643                buf.push(3);
644                write_varu32(buf, collection.len() as u32);
645                buf.extend_from_slice(collection.as_bytes());
646            }
647            EntityKind::TimeSeriesPoint(ref ts) => {
648                buf.push(4);
649                write_varu32(buf, ts.series.len() as u32);
650                buf.extend_from_slice(ts.series.as_bytes());
651                write_varu32(buf, ts.metric.len() as u32);
652                buf.extend_from_slice(ts.metric.as_bytes());
653            }
654            EntityKind::QueueMessage { queue, position } => {
655                buf.push(5);
656                write_varu32(buf, queue.len() as u32);
657                buf.extend_from_slice(queue.as_bytes());
658                write_varu64(buf, *position);
659            }
660        }
661
662        // EntityData
663        match &entity.data {
664            EntityData::Row(row) => {
665                if let Some(ref named) = row.named {
666                    // Named row: type 6 = Row with named HashMap.
667                    // Sort by key so the on-disk byte sequence is
668                    // deterministic — replication relies on hashing the
669                    // serialized record to detect divergence, and a
670                    // HashMap-iteration-order race produces spurious
671                    // mismatches.
672                    buf.push(6);
673                    write_varu32(buf, named.len() as u32);
674                    let mut entries: Vec<_> = named.iter().collect();
675                    entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
676                    for (key, value) in entries {
677                        write_varu32(buf, key.len() as u32);
678                        buf.extend_from_slice(key.as_bytes());
679                        Self::write_value_binary(buf, value);
680                    }
681                } else {
682                    buf.push(0);
683                    write_varu32(buf, row.columns.len() as u32);
684                    for col in &row.columns {
685                        Self::write_value_binary(buf, col);
686                    }
687                }
688            }
689            EntityData::Node(node) => {
690                buf.push(1);
691                write_varu32(buf, node.properties.len() as u32);
692                let mut entries: Vec<_> = node.properties.iter().collect();
693                entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
694                for (key, value) in entries {
695                    write_varu32(buf, key.len() as u32);
696                    buf.extend_from_slice(key.as_bytes());
697                    Self::write_value_binary(buf, value);
698                }
699            }
700            EntityData::Edge(edge) => {
701                buf.push(2);
702                buf.extend_from_slice(&edge.weight.to_le_bytes());
703                write_varu32(buf, edge.properties.len() as u32);
704                let mut entries: Vec<_> = edge.properties.iter().collect();
705                entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
706                for (key, value) in entries {
707                    write_varu32(buf, key.len() as u32);
708                    buf.extend_from_slice(key.as_bytes());
709                    Self::write_value_binary(buf, value);
710                }
711            }
712            EntityData::Vector(vec) => {
713                buf.push(3);
714                write_varu32(buf, vec.dense.len() as u32);
715                for f in &vec.dense {
716                    buf.extend_from_slice(&f.to_le_bytes());
717                }
718                if format_version >= STORE_VERSION_V6 {
719                    buf.push(u8::from(vec.content.is_some()));
720                    if let Some(content) = &vec.content {
721                        write_varu32(buf, content.len() as u32);
722                        buf.extend_from_slice(content.as_bytes());
723                    }
724                }
725            }
726            EntityData::TimeSeries(ts) => {
727                buf.push(4);
728                write_varu32(buf, ts.metric.len() as u32);
729                buf.extend_from_slice(ts.metric.as_bytes());
730                write_varu64(buf, ts.timestamp_ns);
731                buf.extend_from_slice(&ts.value.to_le_bytes());
732                if format_version >= STORE_VERSION_V11 {
733                    // Discriminator: interned series id (1) vs inline tag map (0).
734                    // Interned points (dictionary path) persist just the id;
735                    // un-interned points (metrics remote-write, rollups) keep
736                    // their tags inline so the metrics read paths, which build
737                    // labels directly from `tags`, survive a reopen.
738                    match ts.series_id {
739                        Some(series_id) => {
740                            write_varu32(buf, 1);
741                            write_varu64(buf, series_id);
742                        }
743                        None => {
744                            write_varu32(buf, 0);
745                            write_varu32(buf, ts.tags.len() as u32);
746                            let mut tag_entries: Vec<_> = ts.tags.iter().collect();
747                            tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
748                            for (key, value) in tag_entries {
749                                write_varu32(buf, key.len() as u32);
750                                buf.extend_from_slice(key.as_bytes());
751                                write_varu32(buf, value.len() as u32);
752                                buf.extend_from_slice(value.as_bytes());
753                            }
754                        }
755                    }
756                } else if format_version >= STORE_VERSION_V5 {
757                    write_varu32(buf, ts.tags.len() as u32);
758                    let mut tag_entries: Vec<_> = ts.tags.iter().collect();
759                    tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
760                    for (key, value) in tag_entries {
761                        write_varu32(buf, key.len() as u32);
762                        buf.extend_from_slice(key.as_bytes());
763                        write_varu32(buf, value.len() as u32);
764                        buf.extend_from_slice(value.as_bytes());
765                    }
766                }
767            }
768            EntityData::QueueMessage(msg) => {
769                buf.push(5);
770                Self::write_value_binary(buf, &msg.payload);
771                write_varu64(buf, msg.enqueued_at_ns);
772                write_varu32(buf, msg.attempts);
773            }
774        }
775
776        // Timestamps
777        write_varu64(buf, entity.created_at);
778        write_varu64(buf, entity.updated_at);
779
780        // Embeddings
781        write_varu32(buf, entity.embeddings().len() as u32);
782        for emb in entity.embeddings() {
783            write_varu32(buf, emb.name.len() as u32);
784            buf.extend_from_slice(emb.name.as_bytes());
785            write_varu32(buf, emb.vector.len() as u32);
786            for f in &emb.vector {
787                buf.extend_from_slice(&f.to_le_bytes());
788            }
789            write_varu32(buf, emb.model.len() as u32);
790            buf.extend_from_slice(emb.model.as_bytes());
791        }
792
793        // Cross-refs
794        write_varu32(buf, entity.cross_refs().len() as u32);
795        for cross_ref in entity.cross_refs() {
796            write_varu64(buf, cross_ref.source.raw());
797            write_varu64(buf, cross_ref.target.raw());
798            buf.push(cross_ref.ref_type.to_byte());
799            if format_version >= STORE_VERSION_V2 {
800                write_varu32(buf, cross_ref.target_collection.len() as u32);
801                buf.extend_from_slice(cross_ref.target_collection.as_bytes());
802                buf.extend_from_slice(&cross_ref.weight.to_le_bytes());
803                write_varu64(buf, cross_ref.created_at);
804            }
805        }
806
807        // Sequence ID
808        write_varu64(buf, entity.sequence_id);
809
810        if format_version >= STORE_VERSION_V4 {
811            if let EntityData::QueueMessage(message) = &entity.data {
812                buf.push(u8::from(message.priority.is_some()));
813                if let Some(priority) = message.priority {
814                    buf.extend_from_slice(&priority.to_le_bytes());
815                }
816                write_varu32(buf, message.max_attempts);
817                buf.push(u8::from(message.acked));
818            }
819        }
820
821        if format_version >= STORE_VERSION_V8 {
822            buf.push(u8::from(entity.has_explicit_logical_id()));
823            if entity.has_explicit_logical_id() {
824                write_varu64(buf, entity.logical_id().raw());
825            }
826        }
827        if format_version >= STORE_VERSION_V9 {
828            write_varu64(buf, entity.xmin);
829            write_varu64(buf, entity.xmax);
830        }
831    }
832
833    /// Read a Value from binary buffer
834    /// Type bytes: 0=Null, 1=Boolean, 2=Integer, 3=UnsignedInteger, 4=Float,
835    /// 5=Text, 6=Blob, 7=Timestamp, 8=Duration, 9=IpAddr, 10=MacAddr,
836    /// 11=Vector, 12=Json, 13=Uuid, 14=NodeRef, 15=EdgeRef, 16=VectorRef, 17=RowRef
837    fn read_value_binary(buf: &[u8], pos: &mut usize) -> Result<Value, Box<dyn std::error::Error>> {
838        use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
839
840        let type_byte = buf[*pos];
841        *pos += 1;
842
843        Ok(match type_byte {
844            0 => Value::Null,
845            1 => {
846                let b = buf[*pos] != 0;
847                *pos += 1;
848                Value::Boolean(b)
849            }
850            2 => {
851                let val = i64::from_le_bytes([
852                    buf[*pos],
853                    buf[*pos + 1],
854                    buf[*pos + 2],
855                    buf[*pos + 3],
856                    buf[*pos + 4],
857                    buf[*pos + 5],
858                    buf[*pos + 6],
859                    buf[*pos + 7],
860                ]);
861                *pos += 8;
862                Value::Integer(val)
863            }
864            3 => {
865                let val = u64::from_le_bytes([
866                    buf[*pos],
867                    buf[*pos + 1],
868                    buf[*pos + 2],
869                    buf[*pos + 3],
870                    buf[*pos + 4],
871                    buf[*pos + 5],
872                    buf[*pos + 6],
873                    buf[*pos + 7],
874                ]);
875                *pos += 8;
876                Value::UnsignedInteger(val)
877            }
878            4 => {
879                let val = f64::from_le_bytes([
880                    buf[*pos],
881                    buf[*pos + 1],
882                    buf[*pos + 2],
883                    buf[*pos + 3],
884                    buf[*pos + 4],
885                    buf[*pos + 5],
886                    buf[*pos + 6],
887                    buf[*pos + 7],
888                ]);
889                *pos += 8;
890                Value::Float(val)
891            }
892            5 => {
893                let len = Self::read_varu32_safe(buf, pos)?;
894                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
895                *pos += len;
896                Value::text(s)
897            }
898            6 => {
899                let len = Self::read_varu32_safe(buf, pos)?;
900                let bytes = buf[*pos..*pos + len].to_vec();
901                *pos += len;
902                Value::Blob(bytes)
903            }
904            7 => {
905                let val = i64::from_le_bytes([
906                    buf[*pos],
907                    buf[*pos + 1],
908                    buf[*pos + 2],
909                    buf[*pos + 3],
910                    buf[*pos + 4],
911                    buf[*pos + 5],
912                    buf[*pos + 6],
913                    buf[*pos + 7],
914                ]);
915                *pos += 8;
916                Value::Timestamp(val)
917            }
918            8 => {
919                let val = i64::from_le_bytes([
920                    buf[*pos],
921                    buf[*pos + 1],
922                    buf[*pos + 2],
923                    buf[*pos + 3],
924                    buf[*pos + 4],
925                    buf[*pos + 5],
926                    buf[*pos + 6],
927                    buf[*pos + 7],
928                ]);
929                *pos += 8;
930                Value::Duration(val)
931            }
932            9 => {
933                // IpAddr: first byte = version (4 or 6)
934                let version = buf[*pos];
935                *pos += 1;
936                if version == 4 {
937                    let octets = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
938                    *pos += 4;
939                    Value::IpAddr(IpAddr::V4(Ipv4Addr::from(octets)))
940                } else {
941                    let mut octets = [0u8; 16];
942                    octets.copy_from_slice(&buf[*pos..*pos + 16]);
943                    *pos += 16;
944                    Value::IpAddr(IpAddr::V6(Ipv6Addr::from(octets)))
945                }
946            }
947            10 => {
948                let mut mac = [0u8; 6];
949                mac.copy_from_slice(&buf[*pos..*pos + 6]);
950                *pos += 6;
951                Value::MacAddr(mac)
952            }
953            11 => {
954                let len = Self::read_varu32_safe(buf, pos)?;
955                let mut vector = Vec::with_capacity(len);
956                for _ in 0..len {
957                    let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
958                    *pos += 4;
959                    vector.push(f32::from_le_bytes(bytes));
960                }
961                Value::Vector(vector)
962            }
963            12 => {
964                let len = Self::read_varu32_safe(buf, pos)?;
965                let bytes = buf[*pos..*pos + len].to_vec();
966                *pos += len;
967                Value::Json(bytes)
968            }
969            13 => {
970                let mut uuid = [0u8; 16];
971                uuid.copy_from_slice(&buf[*pos..*pos + 16]);
972                *pos += 16;
973                Value::Uuid(uuid)
974            }
975            14 => {
976                let len = Self::read_varu32_safe(buf, pos)?;
977                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
978                *pos += len;
979                Value::NodeRef(s)
980            }
981            15 => {
982                let len = Self::read_varu32_safe(buf, pos)?;
983                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
984                *pos += len;
985                Value::EdgeRef(s)
986            }
987            16 => {
988                let len = Self::read_varu32_safe(buf, pos)?;
989                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
990                *pos += len;
991                let id = u64::from_le_bytes([
992                    buf[*pos],
993                    buf[*pos + 1],
994                    buf[*pos + 2],
995                    buf[*pos + 3],
996                    buf[*pos + 4],
997                    buf[*pos + 5],
998                    buf[*pos + 6],
999                    buf[*pos + 7],
1000                ]);
1001                *pos += 8;
1002                Value::VectorRef(s, id)
1003            }
1004            17 => {
1005                let len = Self::read_varu32_safe(buf, pos)?;
1006                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1007                *pos += len;
1008                let id = u64::from_le_bytes([
1009                    buf[*pos],
1010                    buf[*pos + 1],
1011                    buf[*pos + 2],
1012                    buf[*pos + 3],
1013                    buf[*pos + 4],
1014                    buf[*pos + 5],
1015                    buf[*pos + 6],
1016                    buf[*pos + 7],
1017                ]);
1018                *pos += 8;
1019                Value::RowRef(s, id)
1020            }
1021            18 => {
1022                let rgb = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1023                *pos += 3;
1024                Value::Color(rgb)
1025            }
1026            19 => {
1027                let len = Self::read_varu32_safe(buf, pos)?;
1028                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1029                *pos += len;
1030                Value::Email(s)
1031            }
1032            20 => {
1033                let len = Self::read_varu32_safe(buf, pos)?;
1034                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1035                *pos += len;
1036                Value::Url(s)
1037            }
1038            21 => {
1039                let val = u64::from_le_bytes([
1040                    buf[*pos],
1041                    buf[*pos + 1],
1042                    buf[*pos + 2],
1043                    buf[*pos + 3],
1044                    buf[*pos + 4],
1045                    buf[*pos + 5],
1046                    buf[*pos + 6],
1047                    buf[*pos + 7],
1048                ]);
1049                *pos += 8;
1050                Value::Phone(val)
1051            }
1052            22 => {
1053                let val =
1054                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1055                *pos += 4;
1056                Value::Semver(val)
1057            }
1058            23 => {
1059                let ip =
1060                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1061                *pos += 4;
1062                let prefix = buf[*pos];
1063                *pos += 1;
1064                Value::Cidr(ip, prefix)
1065            }
1066            24 => {
1067                let val =
1068                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1069                *pos += 4;
1070                Value::Date(val)
1071            }
1072            25 => {
1073                let val =
1074                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1075                *pos += 4;
1076                Value::Time(val)
1077            }
1078            26 => {
1079                let val = i64::from_le_bytes([
1080                    buf[*pos],
1081                    buf[*pos + 1],
1082                    buf[*pos + 2],
1083                    buf[*pos + 3],
1084                    buf[*pos + 4],
1085                    buf[*pos + 5],
1086                    buf[*pos + 6],
1087                    buf[*pos + 7],
1088                ]);
1089                *pos += 8;
1090                Value::Decimal(val)
1091            }
1092            27 => {
1093                let val = buf[*pos];
1094                *pos += 1;
1095                Value::EnumValue(val)
1096            }
1097            28 => {
1098                let len = Self::read_varu32_safe(buf, pos)?;
1099                let mut elems = Vec::with_capacity(len);
1100                for _ in 0..len {
1101                    elems.push(Self::read_value_binary(buf, pos)?);
1102                }
1103                Value::Array(elems)
1104            }
1105            29 => {
1106                let val = i64::from_le_bytes([
1107                    buf[*pos],
1108                    buf[*pos + 1],
1109                    buf[*pos + 2],
1110                    buf[*pos + 3],
1111                    buf[*pos + 4],
1112                    buf[*pos + 5],
1113                    buf[*pos + 6],
1114                    buf[*pos + 7],
1115                ]);
1116                *pos += 8;
1117                Value::TimestampMs(val)
1118            }
1119            30 => {
1120                let val =
1121                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1122                *pos += 4;
1123                Value::Ipv4(val)
1124            }
1125            31 => {
1126                let mut bytes = [0u8; 16];
1127                bytes.copy_from_slice(&buf[*pos..*pos + 16]);
1128                *pos += 16;
1129                Value::Ipv6(bytes)
1130            }
1131            32 => {
1132                let ip =
1133                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1134                *pos += 4;
1135                let mask =
1136                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1137                *pos += 4;
1138                Value::Subnet(ip, mask)
1139            }
1140            33 => {
1141                let val = u16::from_le_bytes([buf[*pos], buf[*pos + 1]]);
1142                *pos += 2;
1143                Value::Port(val)
1144            }
1145            34 => {
1146                let val =
1147                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1148                *pos += 4;
1149                Value::Latitude(val)
1150            }
1151            35 => {
1152                let val =
1153                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1154                *pos += 4;
1155                Value::Longitude(val)
1156            }
1157            36 => {
1158                let lat =
1159                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1160                *pos += 4;
1161                let lon =
1162                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1163                *pos += 4;
1164                Value::GeoPoint(lat, lon)
1165            }
1166            37 => {
1167                let c = [buf[*pos], buf[*pos + 1]];
1168                *pos += 2;
1169                Value::Country2(c)
1170            }
1171            38 => {
1172                let c = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1173                *pos += 3;
1174                Value::Country3(c)
1175            }
1176            39 => {
1177                let c = [buf[*pos], buf[*pos + 1]];
1178                *pos += 2;
1179                Value::Lang2(c)
1180            }
1181            40 => {
1182                let c = [
1183                    buf[*pos],
1184                    buf[*pos + 1],
1185                    buf[*pos + 2],
1186                    buf[*pos + 3],
1187                    buf[*pos + 4],
1188                ];
1189                *pos += 5;
1190                Value::Lang5(c)
1191            }
1192            41 => {
1193                let c = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1194                *pos += 3;
1195                Value::Currency(c)
1196            }
1197            42 => {
1198                let rgba = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
1199                *pos += 4;
1200                Value::ColorAlpha(rgba)
1201            }
1202            43 => {
1203                let val = i64::from_le_bytes([
1204                    buf[*pos],
1205                    buf[*pos + 1],
1206                    buf[*pos + 2],
1207                    buf[*pos + 3],
1208                    buf[*pos + 4],
1209                    buf[*pos + 5],
1210                    buf[*pos + 6],
1211                    buf[*pos + 7],
1212                ]);
1213                *pos += 8;
1214                Value::BigInt(val)
1215            }
1216            44 => {
1217                let col_len = Self::read_varu32_safe(buf, pos)?;
1218                let col = String::from_utf8(buf[*pos..*pos + col_len].to_vec())?;
1219                *pos += col_len;
1220                let key_len = Self::read_varu32_safe(buf, pos)?;
1221                let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
1222                *pos += key_len;
1223                Value::KeyRef(col, key)
1224            }
1225            45 => {
1226                let col_len = Self::read_varu32_safe(buf, pos)?;
1227                let col = String::from_utf8(buf[*pos..*pos + col_len].to_vec())?;
1228                *pos += col_len;
1229                let id = u64::from_le_bytes([
1230                    buf[*pos],
1231                    buf[*pos + 1],
1232                    buf[*pos + 2],
1233                    buf[*pos + 3],
1234                    buf[*pos + 4],
1235                    buf[*pos + 5],
1236                    buf[*pos + 6],
1237                    buf[*pos + 7],
1238                ]);
1239                *pos += 8;
1240                Value::DocRef(col, id)
1241            }
1242            46 => {
1243                let len = Self::read_varu32_safe(buf, pos)?;
1244                let name = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1245                *pos += len;
1246                Value::TableRef(name)
1247            }
1248            47 => {
1249                let val =
1250                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1251                *pos += 4;
1252                Value::PageRef(val)
1253            }
1254            48 => {
1255                let len = Self::read_varu32_safe(buf, pos)?;
1256                let bytes = buf[*pos..*pos + len].to_vec();
1257                *pos += len;
1258                Value::Secret(bytes)
1259            }
1260            49 => {
1261                let len = Self::read_varu32_safe(buf, pos)?;
1262                let hash = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1263                *pos += len;
1264                Value::Password(hash)
1265            }
1266            50 => {
1267                let len = Self::read_varu32_safe(buf, pos)?;
1268                let code = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1269                *pos += len;
1270                Value::AssetCode(code)
1271            }
1272            51 => {
1273                let len = Self::read_varu32_safe(buf, pos)?;
1274                let asset_code = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1275                *pos += len;
1276                let scale = buf[*pos];
1277                *pos += 1;
1278                let minor_units = i64::from_le_bytes([
1279                    buf[*pos],
1280                    buf[*pos + 1],
1281                    buf[*pos + 2],
1282                    buf[*pos + 3],
1283                    buf[*pos + 4],
1284                    buf[*pos + 5],
1285                    buf[*pos + 6],
1286                    buf[*pos + 7],
1287                ]);
1288                *pos += 8;
1289                Value::Money {
1290                    asset_code,
1291                    minor_units,
1292                    scale,
1293                }
1294            }
1295            // C3 TOAST: compressed Text (0x85) and Blob (0x86).
1296            0x85 => {
1297                let orig_len = Self::read_varu32_safe(buf, pos)?;
1298                let comp_len = Self::read_varu32_safe(buf, pos)?;
1299                let compressed = &buf[*pos..*pos + comp_len];
1300                *pos += comp_len;
1301                let mut out = vec![0u8; orig_len];
1302                zstd::bulk::decompress_to_buffer(compressed, &mut out)
1303                    .map_err(|e| format!("C3 Text decompress: {e}"))?;
1304                Value::text(String::from_utf8(out).map_err(|e| format!("C3 Text UTF-8: {e}"))?)
1305            }
1306            0x86 => {
1307                let orig_len = Self::read_varu32_safe(buf, pos)?;
1308                let comp_len = Self::read_varu32_safe(buf, pos)?;
1309                let compressed = &buf[*pos..*pos + comp_len];
1310                *pos += comp_len;
1311                let mut out = vec![0u8; orig_len];
1312                zstd::bulk::decompress_to_buffer(compressed, &mut out)
1313                    .map_err(|e| format!("C3 Blob decompress: {e}"))?;
1314                Value::Blob(out)
1315            }
1316            _ => return Err(format!("Unknown Value type: {}", type_byte).into()),
1317        })
1318    }
1319
1320    /// Write a Value to binary buffer
1321    /// Type bytes: 0=Null, 1=Boolean, 2=Integer, 3=UnsignedInteger, 4=Float,
1322    /// 5=Text, 6=Blob, 7=Timestamp, 8=Duration, 9=IpAddr, 10=MacAddr,
1323    /// 11=Vector, 12=Json, 13=Uuid, 14=NodeRef, 15=EdgeRef, 16=VectorRef, 17=RowRef
1324    fn write_value_binary(buf: &mut Vec<u8>, value: &Value) {
1325        use std::net::IpAddr;
1326
1327        match value {
1328            Value::Null => buf.push(0),
1329            Value::Boolean(b) => {
1330                buf.push(1);
1331                buf.push(if *b { 1 } else { 0 });
1332            }
1333            Value::Integer(i) => {
1334                buf.push(2);
1335                buf.extend_from_slice(&i.to_le_bytes());
1336            }
1337            Value::UnsignedInteger(u) => {
1338                buf.push(3);
1339                buf.extend_from_slice(&u.to_le_bytes());
1340            }
1341            Value::Float(f) => {
1342                buf.push(4);
1343                buf.extend_from_slice(&f.to_le_bytes());
1344            }
1345            Value::Text(s) => {
1346                // C3 TOAST: compress Text values > 2 KiB with zstd.
1347                // Type 0x85 = compressed Text; type 5 = uncompressed (existing).
1348                // Layout for 0x85: [0x85][orig_len: varuint][comp_len: varuint][compressed bytes]
1349                const TEXT_COMPRESS_THRESHOLD: usize = 2048;
1350                if s.len() >= TEXT_COMPRESS_THRESHOLD {
1351                    if let Ok(compressed) = zstd::bulk::compress(s.as_bytes(), 3) {
1352                        if compressed.len() < s.len() {
1353                            buf.push(0x85); // compressed Text marker
1354                            write_varu32(buf, s.len() as u32); // orig_len (decompression hint)
1355                            write_varu32(buf, compressed.len() as u32);
1356                            buf.extend_from_slice(&compressed);
1357                            return; // written — skip uncompressed path
1358                        }
1359                    }
1360                }
1361                buf.push(5);
1362                write_varu32(buf, s.len() as u32);
1363                buf.extend_from_slice(s.as_bytes());
1364            }
1365            Value::Blob(bytes) => {
1366                // C3 TOAST: compress Blob values > 2 KiB with zstd.
1367                // Type 0x86 = compressed Blob; type 6 = uncompressed (existing).
1368                const BLOB_COMPRESS_THRESHOLD: usize = 2048;
1369                if bytes.len() >= BLOB_COMPRESS_THRESHOLD {
1370                    if let Ok(compressed) = zstd::bulk::compress(bytes.as_slice(), 3) {
1371                        if compressed.len() < bytes.len() {
1372                            buf.push(0x86); // compressed Blob marker
1373                            write_varu32(buf, bytes.len() as u32);
1374                            write_varu32(buf, compressed.len() as u32);
1375                            buf.extend_from_slice(&compressed);
1376                            return;
1377                        }
1378                    }
1379                }
1380                buf.push(6);
1381                write_varu32(buf, bytes.len() as u32);
1382                buf.extend_from_slice(bytes);
1383            }
1384            Value::Timestamp(t) => {
1385                buf.push(7);
1386                buf.extend_from_slice(&t.to_le_bytes());
1387            }
1388            Value::Duration(d) => {
1389                buf.push(8);
1390                buf.extend_from_slice(&d.to_le_bytes());
1391            }
1392            Value::IpAddr(ip) => {
1393                buf.push(9);
1394                match ip {
1395                    IpAddr::V4(v4) => {
1396                        buf.push(4);
1397                        buf.extend_from_slice(&v4.octets());
1398                    }
1399                    IpAddr::V6(v6) => {
1400                        buf.push(6);
1401                        buf.extend_from_slice(&v6.octets());
1402                    }
1403                }
1404            }
1405            Value::MacAddr(mac) => {
1406                buf.push(10);
1407                buf.extend_from_slice(mac);
1408            }
1409            Value::Vector(vec) => {
1410                buf.push(11);
1411                write_varu32(buf, vec.len() as u32);
1412                for f in vec {
1413                    buf.extend_from_slice(&f.to_le_bytes());
1414                }
1415            }
1416            Value::Json(bytes) => {
1417                buf.push(12);
1418                write_varu32(buf, bytes.len() as u32);
1419                buf.extend_from_slice(bytes);
1420            }
1421            Value::Uuid(uuid) => {
1422                buf.push(13);
1423                buf.extend_from_slice(uuid);
1424            }
1425            Value::NodeRef(s) => {
1426                buf.push(14);
1427                write_varu32(buf, s.len() as u32);
1428                buf.extend_from_slice(s.as_bytes());
1429            }
1430            Value::EdgeRef(s) => {
1431                buf.push(15);
1432                write_varu32(buf, s.len() as u32);
1433                buf.extend_from_slice(s.as_bytes());
1434            }
1435            Value::VectorRef(s, id) => {
1436                buf.push(16);
1437                write_varu32(buf, s.len() as u32);
1438                buf.extend_from_slice(s.as_bytes());
1439                buf.extend_from_slice(&id.to_le_bytes());
1440            }
1441            Value::RowRef(s, id) => {
1442                buf.push(17);
1443                write_varu32(buf, s.len() as u32);
1444                buf.extend_from_slice(s.as_bytes());
1445                buf.extend_from_slice(&id.to_le_bytes());
1446            }
1447            Value::Color(rgb) => {
1448                buf.push(18);
1449                buf.extend_from_slice(rgb);
1450            }
1451            Value::Email(s) => {
1452                buf.push(19);
1453                write_varu32(buf, s.len() as u32);
1454                buf.extend_from_slice(s.as_bytes());
1455            }
1456            Value::Url(s) => {
1457                buf.push(20);
1458                write_varu32(buf, s.len() as u32);
1459                buf.extend_from_slice(s.as_bytes());
1460            }
1461            Value::Phone(n) => {
1462                buf.push(21);
1463                buf.extend_from_slice(&n.to_le_bytes());
1464            }
1465            Value::Semver(v) => {
1466                buf.push(22);
1467                buf.extend_from_slice(&v.to_le_bytes());
1468            }
1469            Value::Cidr(ip, prefix) => {
1470                buf.push(23);
1471                buf.extend_from_slice(&ip.to_le_bytes());
1472                buf.push(*prefix);
1473            }
1474            Value::Date(d) => {
1475                buf.push(24);
1476                buf.extend_from_slice(&d.to_le_bytes());
1477            }
1478            Value::Time(t) => {
1479                buf.push(25);
1480                buf.extend_from_slice(&t.to_le_bytes());
1481            }
1482            Value::Decimal(v) => {
1483                buf.push(26);
1484                buf.extend_from_slice(&v.to_le_bytes());
1485            }
1486            Value::EnumValue(i) => {
1487                buf.push(27);
1488                buf.push(*i);
1489            }
1490            Value::Array(elems) => {
1491                buf.push(28);
1492                write_varu32(buf, elems.len() as u32);
1493                for elem in elems {
1494                    Self::write_value_binary(buf, elem);
1495                }
1496            }
1497            Value::TimestampMs(v) => {
1498                buf.push(29);
1499                buf.extend_from_slice(&v.to_le_bytes());
1500            }
1501            Value::Ipv4(v) => {
1502                buf.push(30);
1503                buf.extend_from_slice(&v.to_le_bytes());
1504            }
1505            Value::Ipv6(bytes) => {
1506                buf.push(31);
1507                buf.extend_from_slice(bytes);
1508            }
1509            Value::Subnet(ip, mask) => {
1510                buf.push(32);
1511                buf.extend_from_slice(&ip.to_le_bytes());
1512                buf.extend_from_slice(&mask.to_le_bytes());
1513            }
1514            Value::Port(v) => {
1515                buf.push(33);
1516                buf.extend_from_slice(&v.to_le_bytes());
1517            }
1518            Value::Latitude(v) => {
1519                buf.push(34);
1520                buf.extend_from_slice(&v.to_le_bytes());
1521            }
1522            Value::Longitude(v) => {
1523                buf.push(35);
1524                buf.extend_from_slice(&v.to_le_bytes());
1525            }
1526            Value::GeoPoint(lat, lon) => {
1527                buf.push(36);
1528                buf.extend_from_slice(&lat.to_le_bytes());
1529                buf.extend_from_slice(&lon.to_le_bytes());
1530            }
1531            Value::Country2(c) => {
1532                buf.push(37);
1533                buf.extend_from_slice(c);
1534            }
1535            Value::Country3(c) => {
1536                buf.push(38);
1537                buf.extend_from_slice(c);
1538            }
1539            Value::Lang2(c) => {
1540                buf.push(39);
1541                buf.extend_from_slice(c);
1542            }
1543            Value::Lang5(c) => {
1544                buf.push(40);
1545                buf.extend_from_slice(c);
1546            }
1547            Value::Currency(c) => {
1548                buf.push(41);
1549                buf.extend_from_slice(c);
1550            }
1551            Value::AssetCode(code) => {
1552                buf.push(50);
1553                write_varu32(buf, code.len() as u32);
1554                buf.extend_from_slice(code.as_bytes());
1555            }
1556            Value::Money {
1557                asset_code,
1558                minor_units,
1559                scale,
1560            } => {
1561                buf.push(51);
1562                write_varu32(buf, asset_code.len() as u32);
1563                buf.extend_from_slice(asset_code.as_bytes());
1564                buf.push(*scale);
1565                buf.extend_from_slice(&minor_units.to_le_bytes());
1566            }
1567            Value::ColorAlpha(rgba) => {
1568                buf.push(42);
1569                buf.extend_from_slice(rgba);
1570            }
1571            Value::BigInt(v) => {
1572                buf.push(43);
1573                buf.extend_from_slice(&v.to_le_bytes());
1574            }
1575            Value::KeyRef(col, key) => {
1576                buf.push(44);
1577                write_varu32(buf, col.len() as u32);
1578                buf.extend_from_slice(col.as_bytes());
1579                write_varu32(buf, key.len() as u32);
1580                buf.extend_from_slice(key.as_bytes());
1581            }
1582            Value::DocRef(col, id) => {
1583                buf.push(45);
1584                write_varu32(buf, col.len() as u32);
1585                buf.extend_from_slice(col.as_bytes());
1586                buf.extend_from_slice(&id.to_le_bytes());
1587            }
1588            Value::TableRef(name) => {
1589                buf.push(46);
1590                write_varu32(buf, name.len() as u32);
1591                buf.extend_from_slice(name.as_bytes());
1592            }
1593            Value::PageRef(page_id) => {
1594                buf.push(47);
1595                buf.extend_from_slice(&page_id.to_le_bytes());
1596            }
1597            Value::Secret(bytes) => {
1598                buf.push(48);
1599                write_varu32(buf, bytes.len() as u32);
1600                buf.extend_from_slice(bytes);
1601            }
1602            Value::Password(hash) => {
1603                buf.push(49);
1604                write_varu32(buf, hash.len() as u32);
1605                buf.extend_from_slice(hash.as_bytes());
1606            }
1607        }
1608    }
1609}
1610
1611#[cfg(test)]
1612mod aux_metadata_dump_tests {
1613    use super::*;
1614    use std::collections::HashMap;
1615
1616    #[test]
1617    fn aux_metadata_round_trips_through_binary_dump() {
1618        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
1619        store.create_collection("k").expect("create collection");
1620        store.set_aux_metadata(b"contract-blob".to_vec());
1621
1622        let bytes = store.to_binary_dump_bytes();
1623        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
1624
1625        assert_eq!(reloaded.aux_metadata(), b"contract-blob".to_vec());
1626    }
1627
1628    #[test]
1629    fn empty_aux_metadata_round_trips_as_empty() {
1630        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
1631        let bytes = store.to_binary_dump_bytes();
1632        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
1633
1634        assert!(reloaded.aux_metadata().is_empty());
1635    }
1636
1637    #[test]
1638    fn interned_timeseries_series_reduces_repeated_tag_storage() {
1639        let tags = HashMap::from([
1640            (
1641                "host".to_string(),
1642                "srv-very-long-hostname-0001".to_string(),
1643            ),
1644            ("region".to_string(), "us-east-1".to_string()),
1645            ("service".to_string(), "checkout-api".to_string()),
1646        ]);
1647        let inline_bytes: usize = (0..100)
1648            .map(|timestamp_ns| {
1649                let entity = UnifiedEntity::new(
1650                    EntityId::new(0),
1651                    EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
1652                        series: "cpu_metrics".to_string(),
1653                        metric: "cpu.idle".to_string(),
1654                    })),
1655                    EntityData::TimeSeries(crate::storage::TimeSeriesData {
1656                        metric: "cpu.idle".to_string(),
1657                        series_id: None,
1658                        timestamp_ns,
1659                        value: 94.8,
1660                        tags: tags.clone(),
1661                    }),
1662                );
1663                UnifiedStore::serialize_entity_record(&entity, None, reddb_file::STORE_VERSION_V10)
1664                    .len()
1665            })
1666            .sum();
1667
1668        let mut dictionary_fields = HashMap::new();
1669        dictionary_fields.insert("collection".to_string(), Value::text("cpu_metrics"));
1670        dictionary_fields.insert("series_id".to_string(), Value::UnsignedInteger(0));
1671        dictionary_fields.insert("metric".to_string(), Value::text("cpu.idle"));
1672        dictionary_fields.insert(
1673            "canonical_tags".to_string(),
1674            Value::text(
1675                r#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#,
1676            ),
1677        );
1678        dictionary_fields.insert(
1679            "tags".to_string(),
1680            Value::Json(
1681                br#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#
1682                    .to_vec(),
1683            ),
1684        );
1685        let dictionary = UnifiedEntity::new(
1686            EntityId::new(0),
1687            EntityKind::TableRow {
1688                table: Arc::from("red_timeseries_series"),
1689                row_id: 0,
1690            },
1691            EntityData::Row(RowData {
1692                columns: Vec::new(),
1693                named: Some(dictionary_fields),
1694                schema: None,
1695            }),
1696        );
1697        let dictionary_bytes =
1698            UnifiedStore::serialize_entity_record(&dictionary, None, STORE_VERSION_V11).len();
1699        let interned_bytes = dictionary_bytes
1700            + (0..100)
1701                .map(|timestamp_ns| {
1702                    let entity = UnifiedEntity::new(
1703                        EntityId::new(0),
1704                        EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
1705                            series: "cpu_metrics".to_string(),
1706                            metric: "cpu.idle".to_string(),
1707                        })),
1708                        EntityData::TimeSeries(crate::storage::TimeSeriesData {
1709                            metric: "cpu.idle".to_string(),
1710                            series_id: Some(0),
1711                            timestamp_ns,
1712                            value: 94.8,
1713                            tags: HashMap::new(),
1714                        }),
1715                    );
1716                    UnifiedStore::serialize_entity_record(&entity, None, STORE_VERSION_V11).len()
1717                })
1718                .sum::<usize>();
1719
1720        assert!(
1721            interned_bytes < inline_bytes,
1722            "interned series storage should shrink repeated tags: inline={inline_bytes}, interned={interned_bytes}"
1723        );
1724    }
1725}
reddb_server/storage/unified/store/impl_file.rs

reddb_server/storage/unified/store/
impl_file.rs
