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                let fields = if format_version >= STORE_VERSION_V12 {
442                    let field_count = Self::read_varu32_safe(buf, pos)?;
443                    let mut fields = HashMap::with_capacity(field_count);
444                    for _ in 0..field_count {
445                        let key_len = Self::read_varu32_safe(buf, pos)?;
446                        let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
447                        *pos += key_len;
448                        let value = Self::read_value_binary(buf, pos)?;
449                        fields.insert(key, value);
450                    }
451                    fields
452                } else {
453                    HashMap::new()
454                };
455                EntityData::TimeSeries(crate::storage::unified::entity::TimeSeriesData {
456                    metric,
457                    series_id,
458                    timestamp_ns,
459                    value: f64::from_le_bytes(value_bytes),
460                    tags,
461                    fields,
462                })
463            }
464            5 => {
465                // QueueMessage
466                let payload = Self::read_value_binary(buf, pos)?;
467                let enqueued_at_ns = Self::read_varu64_safe(buf, pos)?;
468                let attempts = Self::read_varu32_safe(buf, pos)? as u32;
469                EntityData::QueueMessage(crate::storage::unified::entity::QueueMessageData {
470                    payload,
471                    priority: None,
472                    enqueued_at_ns,
473                    attempts,
474                    max_attempts: 3,
475                    acked: false,
476                })
477            }
478            6 => {
479                // Row with named HashMap
480                let field_count = Self::read_varu32_safe(buf, pos)?;
481                let mut named = HashMap::with_capacity(field_count);
482                for _ in 0..field_count {
483                    let key_len = Self::read_varu32_safe(buf, pos)?;
484                    let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
485                    *pos += key_len;
486                    let value = Self::read_value_binary(buf, pos)?;
487                    named.insert(key, value);
488                }
489                EntityData::Row(RowData {
490                    columns: Vec::new(),
491                    named: Some(named),
492                    schema: None,
493                })
494            }
495            _ => return Err(format!("Unknown EntityData type: {}", data_type).into()),
496        };
497
498        // Timestamps
499        let created_at = Self::read_varu64_safe(buf, pos)?;
500        let updated_at = Self::read_varu64_safe(buf, pos)?;
501
502        // Embeddings count
503        let embedding_count = Self::read_varu32_safe(buf, pos)?;
504        let mut embeddings = Vec::with_capacity(embedding_count);
505        for _ in 0..embedding_count {
506            let name_len = Self::read_varu32_safe(buf, pos)?;
507            let name = String::from_utf8(buf[*pos..*pos + name_len].to_vec())?;
508            *pos += name_len;
509
510            let dim = Self::read_varu32_safe(buf, pos)?;
511            let mut vector = Vec::with_capacity(dim);
512            for _ in 0..dim {
513                let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
514                *pos += 4;
515                vector.push(f32::from_le_bytes(bytes));
516            }
517
518            let model_len = Self::read_varu32_safe(buf, pos)?;
519            let model = String::from_utf8(buf[*pos..*pos + model_len].to_vec())?;
520            *pos += model_len;
521
522            embeddings.push(EmbeddingSlot::new(name, vector, model));
523        }
524
525        // Cross-refs count
526        let cross_ref_count = Self::read_varu32_safe(buf, pos)?;
527        let mut cross_refs = Vec::with_capacity(cross_ref_count);
528        for _ in 0..cross_ref_count {
529            let source = Self::read_varu64_safe(buf, pos)?;
530            let target = Self::read_varu64_safe(buf, pos)?;
531            let ref_type_byte = buf[*pos];
532            *pos += 1;
533            let (target_collection, weight, created_at) = if format_version >= STORE_VERSION_V2 {
534                let coll_len = Self::read_varu32_safe(buf, pos)?;
535                let collection = String::from_utf8(buf[*pos..*pos + coll_len].to_vec())?;
536                *pos += coll_len;
537                let weight_bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
538                *pos += 4;
539                let weight = f32::from_le_bytes(weight_bytes);
540                let created_at = Self::read_varu64_safe(buf, pos)?;
541                (collection, weight, created_at)
542            } else {
543                (String::new(), 1.0, 0)
544            };
545
546            let mut cross_ref = CrossRef::new(
547                EntityId::new(source),
548                EntityId::new(target),
549                target_collection,
550                RefType::from_byte(ref_type_byte),
551            );
552            cross_ref.weight = weight;
553            cross_ref.created_at = created_at;
554            cross_refs.push(cross_ref);
555        }
556
557        // Sequence ID
558        let sequence_id = Self::read_varu64_safe(buf, pos)?;
559
560        if format_version >= STORE_VERSION_V4 {
561            if let EntityData::QueueMessage(message) = &mut data {
562                if *pos < buf.len() {
563                    let priority_present = buf[*pos] != 0;
564                    *pos += 1;
565                    message.priority = if priority_present {
566                        if *pos + 4 > buf.len() {
567                            return Err("truncated queue priority".into());
568                        }
569                        let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
570                        *pos += 4;
571                        Some(i32::from_le_bytes(bytes))
572                    } else {
573                        None
574                    };
575                    message.max_attempts = Self::read_varu32_safe(buf, pos)? as u32;
576                    if *pos >= buf.len() {
577                        return Err("truncated queue ack flag".into());
578                    }
579                    message.acked = buf[*pos] != 0;
580                    *pos += 1;
581                }
582            }
583        }
584
585        let mut entity = UnifiedEntity::new(EntityId::new(id), kind, data);
586        entity.created_at = created_at;
587        entity.updated_at = updated_at;
588        entity.sequence_id = sequence_id;
589        if format_version >= STORE_VERSION_V8 && *pos < buf.len() {
590            let has_logical_id = buf[*pos] != 0;
591            *pos += 1;
592            if has_logical_id {
593                let logical_id = Self::read_varu64_safe(buf, pos)?;
594                entity.set_logical_id(EntityId::new(logical_id));
595            }
596        }
597        if format_version >= STORE_VERSION_V9 && *pos < buf.len() {
598            let xmin = Self::read_varu64_safe(buf, pos)?;
599            let xmax = Self::read_varu64_safe(buf, pos)?;
600            entity.set_xmin(xmin);
601            entity.set_xmax(xmax);
602        }
603        if !embeddings.is_empty() || !cross_refs.is_empty() {
604            entity.embeddings_mut().extend(embeddings);
605            entity.cross_refs_mut().extend(cross_refs);
606        }
607
608        Ok(entity)
609    }
610
611    /// Safe varu32 reader that converts DecodeError to Box<dyn Error>
612    fn read_varu32_safe(buf: &[u8], pos: &mut usize) -> Result<usize, Box<dyn std::error::Error>> {
613        read_varu32(buf, pos)
614            .map(|v| v as usize)
615            .map_err(|e| format!("Decode error: {:?}", e).into())
616    }
617
618    /// Safe varu64 reader that converts DecodeError to Box<dyn Error>
619    fn read_varu64_safe(buf: &[u8], pos: &mut usize) -> Result<u64, Box<dyn std::error::Error>> {
620        read_varu64(buf, pos).map_err(|e| format!("Decode error: {:?}", e).into())
621    }
622
623    /// Write entity to binary buffer
624    pub(crate) fn write_entity_binary(
625        buf: &mut Vec<u8>,
626        entity: &UnifiedEntity,
627        format_version: u32,
628    ) {
629        // Entity ID
630        write_varu64(buf, entity.id.raw());
631
632        // EntityKind
633        match &entity.kind {
634            EntityKind::TableRow { table, row_id } => {
635                buf.push(0);
636                write_varu32(buf, table.len() as u32);
637                buf.extend_from_slice(table.as_bytes());
638                write_varu64(buf, *row_id);
639            }
640            EntityKind::GraphNode(ref node) => {
641                buf.push(1);
642                write_varu32(buf, node.label.len() as u32);
643                buf.extend_from_slice(node.label.as_bytes());
644                write_varu32(buf, node.node_type.len() as u32);
645                buf.extend_from_slice(node.node_type.as_bytes());
646            }
647            EntityKind::GraphEdge(ref edge) => {
648                buf.push(2);
649                write_varu32(buf, edge.label.len() as u32);
650                buf.extend_from_slice(edge.label.as_bytes());
651                write_varu32(buf, edge.from_node.len() as u32);
652                buf.extend_from_slice(edge.from_node.as_bytes());
653                write_varu32(buf, edge.to_node.len() as u32);
654                buf.extend_from_slice(edge.to_node.as_bytes());
655                buf.extend_from_slice(&edge.weight.to_le_bytes());
656            }
657            EntityKind::Vector { collection } => {
658                buf.push(3);
659                write_varu32(buf, collection.len() as u32);
660                buf.extend_from_slice(collection.as_bytes());
661            }
662            EntityKind::TimeSeriesPoint(ref ts) => {
663                buf.push(4);
664                write_varu32(buf, ts.series.len() as u32);
665                buf.extend_from_slice(ts.series.as_bytes());
666                write_varu32(buf, ts.metric.len() as u32);
667                buf.extend_from_slice(ts.metric.as_bytes());
668            }
669            EntityKind::QueueMessage { queue, position } => {
670                buf.push(5);
671                write_varu32(buf, queue.len() as u32);
672                buf.extend_from_slice(queue.as_bytes());
673                write_varu64(buf, *position);
674            }
675        }
676
677        // EntityData
678        match &entity.data {
679            EntityData::Row(row) => {
680                if let Some(ref named) = row.named {
681                    // Named row: type 6 = Row with named HashMap.
682                    // Sort by key so the on-disk byte sequence is
683                    // deterministic — replication relies on hashing the
684                    // serialized record to detect divergence, and a
685                    // HashMap-iteration-order race produces spurious
686                    // mismatches.
687                    buf.push(6);
688                    write_varu32(buf, named.len() as u32);
689                    let mut entries: Vec<_> = named.iter().collect();
690                    entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
691                    for (key, value) in entries {
692                        write_varu32(buf, key.len() as u32);
693                        buf.extend_from_slice(key.as_bytes());
694                        Self::write_value_binary(buf, value);
695                    }
696                } else {
697                    buf.push(0);
698                    write_varu32(buf, row.columns.len() as u32);
699                    for col in &row.columns {
700                        Self::write_value_binary(buf, col);
701                    }
702                }
703            }
704            EntityData::Node(node) => {
705                buf.push(1);
706                write_varu32(buf, node.properties.len() as u32);
707                let mut entries: Vec<_> = node.properties.iter().collect();
708                entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
709                for (key, value) in entries {
710                    write_varu32(buf, key.len() as u32);
711                    buf.extend_from_slice(key.as_bytes());
712                    Self::write_value_binary(buf, value);
713                }
714            }
715            EntityData::Edge(edge) => {
716                buf.push(2);
717                buf.extend_from_slice(&edge.weight.to_le_bytes());
718                write_varu32(buf, edge.properties.len() as u32);
719                let mut entries: Vec<_> = edge.properties.iter().collect();
720                entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
721                for (key, value) in entries {
722                    write_varu32(buf, key.len() as u32);
723                    buf.extend_from_slice(key.as_bytes());
724                    Self::write_value_binary(buf, value);
725                }
726            }
727            EntityData::Vector(vec) => {
728                buf.push(3);
729                write_varu32(buf, vec.dense.len() as u32);
730                for f in &vec.dense {
731                    buf.extend_from_slice(&f.to_le_bytes());
732                }
733                if format_version >= STORE_VERSION_V6 {
734                    buf.push(u8::from(vec.content.is_some()));
735                    if let Some(content) = &vec.content {
736                        write_varu32(buf, content.len() as u32);
737                        buf.extend_from_slice(content.as_bytes());
738                    }
739                }
740            }
741            EntityData::TimeSeries(ts) => {
742                buf.push(4);
743                write_varu32(buf, ts.metric.len() as u32);
744                buf.extend_from_slice(ts.metric.as_bytes());
745                write_varu64(buf, ts.timestamp_ns);
746                buf.extend_from_slice(&ts.value.to_le_bytes());
747                if format_version >= STORE_VERSION_V11 {
748                    // Discriminator: interned series id (1) vs inline tag map (0).
749                    // Interned points (dictionary path) persist just the id;
750                    // un-interned points (metrics remote-write, rollups) keep
751                    // their tags inline so the metrics read paths, which build
752                    // labels directly from `tags`, survive a reopen.
753                    match ts.series_id {
754                        Some(series_id) => {
755                            write_varu32(buf, 1);
756                            write_varu64(buf, series_id);
757                        }
758                        None => {
759                            write_varu32(buf, 0);
760                            write_varu32(buf, ts.tags.len() as u32);
761                            let mut tag_entries: Vec<_> = ts.tags.iter().collect();
762                            tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
763                            for (key, value) in tag_entries {
764                                write_varu32(buf, key.len() as u32);
765                                buf.extend_from_slice(key.as_bytes());
766                                write_varu32(buf, value.len() as u32);
767                                buf.extend_from_slice(value.as_bytes());
768                            }
769                        }
770                    }
771                } else if format_version >= STORE_VERSION_V5 {
772                    write_varu32(buf, ts.tags.len() as u32);
773                    let mut tag_entries: Vec<_> = ts.tags.iter().collect();
774                    tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
775                    for (key, value) in tag_entries {
776                        write_varu32(buf, key.len() as u32);
777                        buf.extend_from_slice(key.as_bytes());
778                        write_varu32(buf, value.len() as u32);
779                        buf.extend_from_slice(value.as_bytes());
780                    }
781                }
782                if format_version >= STORE_VERSION_V12 {
783                    write_varu32(buf, ts.fields.len() as u32);
784                    let mut field_entries: Vec<_> = ts.fields.iter().collect();
785                    field_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
786                    for (key, value) in field_entries {
787                        write_varu32(buf, key.len() as u32);
788                        buf.extend_from_slice(key.as_bytes());
789                        Self::write_value_binary(buf, value);
790                    }
791                }
792            }
793            EntityData::QueueMessage(msg) => {
794                buf.push(5);
795                Self::write_value_binary(buf, &msg.payload);
796                write_varu64(buf, msg.enqueued_at_ns);
797                write_varu32(buf, msg.attempts);
798            }
799        }
800
801        // Timestamps
802        write_varu64(buf, entity.created_at);
803        write_varu64(buf, entity.updated_at);
804
805        // Embeddings
806        write_varu32(buf, entity.embeddings().len() as u32);
807        for emb in entity.embeddings() {
808            write_varu32(buf, emb.name.len() as u32);
809            buf.extend_from_slice(emb.name.as_bytes());
810            write_varu32(buf, emb.vector.len() as u32);
811            for f in &emb.vector {
812                buf.extend_from_slice(&f.to_le_bytes());
813            }
814            write_varu32(buf, emb.model.len() as u32);
815            buf.extend_from_slice(emb.model.as_bytes());
816        }
817
818        // Cross-refs
819        write_varu32(buf, entity.cross_refs().len() as u32);
820        for cross_ref in entity.cross_refs() {
821            write_varu64(buf, cross_ref.source.raw());
822            write_varu64(buf, cross_ref.target.raw());
823            buf.push(cross_ref.ref_type.to_byte());
824            if format_version >= STORE_VERSION_V2 {
825                write_varu32(buf, cross_ref.target_collection.len() as u32);
826                buf.extend_from_slice(cross_ref.target_collection.as_bytes());
827                buf.extend_from_slice(&cross_ref.weight.to_le_bytes());
828                write_varu64(buf, cross_ref.created_at);
829            }
830        }
831
832        // Sequence ID
833        write_varu64(buf, entity.sequence_id);
834
835        if format_version >= STORE_VERSION_V4 {
836            if let EntityData::QueueMessage(message) = &entity.data {
837                buf.push(u8::from(message.priority.is_some()));
838                if let Some(priority) = message.priority {
839                    buf.extend_from_slice(&priority.to_le_bytes());
840                }
841                write_varu32(buf, message.max_attempts);
842                buf.push(u8::from(message.acked));
843            }
844        }
845
846        if format_version >= STORE_VERSION_V8 {
847            buf.push(u8::from(entity.has_explicit_logical_id()));
848            if entity.has_explicit_logical_id() {
849                write_varu64(buf, entity.logical_id().raw());
850            }
851        }
852        if format_version >= STORE_VERSION_V9 {
853            write_varu64(buf, entity.xmin);
854            write_varu64(buf, entity.xmax);
855        }
856    }
857
858    /// Read a Value from binary buffer
859    /// Type bytes: 0=Null, 1=Boolean, 2=Integer, 3=UnsignedInteger, 4=Float,
860    /// 5=Text, 6=Blob, 7=Timestamp, 8=Duration, 9=IpAddr, 10=MacAddr,
861    /// 11=Vector, 12=Json, 13=Uuid, 14=NodeRef, 15=EdgeRef, 16=VectorRef, 17=RowRef
862    fn read_value_binary(buf: &[u8], pos: &mut usize) -> Result<Value, Box<dyn std::error::Error>> {
863        use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
864
865        let type_byte = buf[*pos];
866        *pos += 1;
867
868        Ok(match type_byte {
869            0 => Value::Null,
870            1 => {
871                let b = buf[*pos] != 0;
872                *pos += 1;
873                Value::Boolean(b)
874            }
875            2 => {
876                let val = i64::from_le_bytes([
877                    buf[*pos],
878                    buf[*pos + 1],
879                    buf[*pos + 2],
880                    buf[*pos + 3],
881                    buf[*pos + 4],
882                    buf[*pos + 5],
883                    buf[*pos + 6],
884                    buf[*pos + 7],
885                ]);
886                *pos += 8;
887                Value::Integer(val)
888            }
889            3 => {
890                let val = u64::from_le_bytes([
891                    buf[*pos],
892                    buf[*pos + 1],
893                    buf[*pos + 2],
894                    buf[*pos + 3],
895                    buf[*pos + 4],
896                    buf[*pos + 5],
897                    buf[*pos + 6],
898                    buf[*pos + 7],
899                ]);
900                *pos += 8;
901                Value::UnsignedInteger(val)
902            }
903            4 => {
904                let val = f64::from_le_bytes([
905                    buf[*pos],
906                    buf[*pos + 1],
907                    buf[*pos + 2],
908                    buf[*pos + 3],
909                    buf[*pos + 4],
910                    buf[*pos + 5],
911                    buf[*pos + 6],
912                    buf[*pos + 7],
913                ]);
914                *pos += 8;
915                Value::Float(val)
916            }
917            5 => {
918                let len = Self::read_varu32_safe(buf, pos)?;
919                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
920                *pos += len;
921                Value::text(s)
922            }
923            6 => {
924                let len = Self::read_varu32_safe(buf, pos)?;
925                let bytes = buf[*pos..*pos + len].to_vec();
926                *pos += len;
927                Value::Blob(bytes)
928            }
929            7 => {
930                let val = i64::from_le_bytes([
931                    buf[*pos],
932                    buf[*pos + 1],
933                    buf[*pos + 2],
934                    buf[*pos + 3],
935                    buf[*pos + 4],
936                    buf[*pos + 5],
937                    buf[*pos + 6],
938                    buf[*pos + 7],
939                ]);
940                *pos += 8;
941                Value::Timestamp(val)
942            }
943            8 => {
944                let val = i64::from_le_bytes([
945                    buf[*pos],
946                    buf[*pos + 1],
947                    buf[*pos + 2],
948                    buf[*pos + 3],
949                    buf[*pos + 4],
950                    buf[*pos + 5],
951                    buf[*pos + 6],
952                    buf[*pos + 7],
953                ]);
954                *pos += 8;
955                Value::Duration(val)
956            }
957            9 => {
958                // IpAddr: first byte = version (4 or 6)
959                let version = buf[*pos];
960                *pos += 1;
961                if version == 4 {
962                    let octets = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
963                    *pos += 4;
964                    Value::IpAddr(IpAddr::V4(Ipv4Addr::from(octets)))
965                } else {
966                    let mut octets = [0u8; 16];
967                    octets.copy_from_slice(&buf[*pos..*pos + 16]);
968                    *pos += 16;
969                    Value::IpAddr(IpAddr::V6(Ipv6Addr::from(octets)))
970                }
971            }
972            10 => {
973                let mut mac = [0u8; 6];
974                mac.copy_from_slice(&buf[*pos..*pos + 6]);
975                *pos += 6;
976                Value::MacAddr(mac)
977            }
978            11 => {
979                let len = Self::read_varu32_safe(buf, pos)?;
980                let mut vector = Vec::with_capacity(len);
981                for _ in 0..len {
982                    let bytes = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
983                    *pos += 4;
984                    vector.push(f32::from_le_bytes(bytes));
985                }
986                Value::Vector(vector)
987            }
988            12 => {
989                let len = Self::read_varu32_safe(buf, pos)?;
990                let bytes = buf[*pos..*pos + len].to_vec();
991                *pos += len;
992                Value::Json(bytes)
993            }
994            13 => {
995                let mut uuid = [0u8; 16];
996                uuid.copy_from_slice(&buf[*pos..*pos + 16]);
997                *pos += 16;
998                Value::Uuid(uuid)
999            }
1000            14 => {
1001                let len = Self::read_varu32_safe(buf, pos)?;
1002                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1003                *pos += len;
1004                Value::NodeRef(s)
1005            }
1006            15 => {
1007                let len = Self::read_varu32_safe(buf, pos)?;
1008                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1009                *pos += len;
1010                Value::EdgeRef(s)
1011            }
1012            16 => {
1013                let len = Self::read_varu32_safe(buf, pos)?;
1014                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1015                *pos += len;
1016                let id = u64::from_le_bytes([
1017                    buf[*pos],
1018                    buf[*pos + 1],
1019                    buf[*pos + 2],
1020                    buf[*pos + 3],
1021                    buf[*pos + 4],
1022                    buf[*pos + 5],
1023                    buf[*pos + 6],
1024                    buf[*pos + 7],
1025                ]);
1026                *pos += 8;
1027                Value::VectorRef(s, id)
1028            }
1029            17 => {
1030                let len = Self::read_varu32_safe(buf, pos)?;
1031                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1032                *pos += len;
1033                let id = u64::from_le_bytes([
1034                    buf[*pos],
1035                    buf[*pos + 1],
1036                    buf[*pos + 2],
1037                    buf[*pos + 3],
1038                    buf[*pos + 4],
1039                    buf[*pos + 5],
1040                    buf[*pos + 6],
1041                    buf[*pos + 7],
1042                ]);
1043                *pos += 8;
1044                Value::RowRef(s, id)
1045            }
1046            18 => {
1047                let rgb = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1048                *pos += 3;
1049                Value::Color(rgb)
1050            }
1051            19 => {
1052                let len = Self::read_varu32_safe(buf, pos)?;
1053                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1054                *pos += len;
1055                Value::Email(s)
1056            }
1057            20 => {
1058                let len = Self::read_varu32_safe(buf, pos)?;
1059                let s = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1060                *pos += len;
1061                Value::Url(s)
1062            }
1063            21 => {
1064                let val = u64::from_le_bytes([
1065                    buf[*pos],
1066                    buf[*pos + 1],
1067                    buf[*pos + 2],
1068                    buf[*pos + 3],
1069                    buf[*pos + 4],
1070                    buf[*pos + 5],
1071                    buf[*pos + 6],
1072                    buf[*pos + 7],
1073                ]);
1074                *pos += 8;
1075                Value::Phone(val)
1076            }
1077            22 => {
1078                let val =
1079                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1080                *pos += 4;
1081                Value::Semver(val)
1082            }
1083            23 => {
1084                let ip =
1085                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1086                *pos += 4;
1087                let prefix = buf[*pos];
1088                *pos += 1;
1089                Value::Cidr(ip, prefix)
1090            }
1091            24 => {
1092                let val =
1093                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1094                *pos += 4;
1095                Value::Date(val)
1096            }
1097            25 => {
1098                let val =
1099                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1100                *pos += 4;
1101                Value::Time(val)
1102            }
1103            26 => {
1104                let val = i64::from_le_bytes([
1105                    buf[*pos],
1106                    buf[*pos + 1],
1107                    buf[*pos + 2],
1108                    buf[*pos + 3],
1109                    buf[*pos + 4],
1110                    buf[*pos + 5],
1111                    buf[*pos + 6],
1112                    buf[*pos + 7],
1113                ]);
1114                *pos += 8;
1115                Value::Decimal(val)
1116            }
1117            27 => {
1118                let val = buf[*pos];
1119                *pos += 1;
1120                Value::EnumValue(val)
1121            }
1122            28 => {
1123                let len = Self::read_varu32_safe(buf, pos)?;
1124                let mut elems = Vec::with_capacity(len);
1125                for _ in 0..len {
1126                    elems.push(Self::read_value_binary(buf, pos)?);
1127                }
1128                Value::Array(elems)
1129            }
1130            29 => {
1131                let val = i64::from_le_bytes([
1132                    buf[*pos],
1133                    buf[*pos + 1],
1134                    buf[*pos + 2],
1135                    buf[*pos + 3],
1136                    buf[*pos + 4],
1137                    buf[*pos + 5],
1138                    buf[*pos + 6],
1139                    buf[*pos + 7],
1140                ]);
1141                *pos += 8;
1142                Value::TimestampMs(val)
1143            }
1144            30 => {
1145                let val =
1146                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1147                *pos += 4;
1148                Value::Ipv4(val)
1149            }
1150            31 => {
1151                let mut bytes = [0u8; 16];
1152                bytes.copy_from_slice(&buf[*pos..*pos + 16]);
1153                *pos += 16;
1154                Value::Ipv6(bytes)
1155            }
1156            32 => {
1157                let ip =
1158                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1159                *pos += 4;
1160                let mask =
1161                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1162                *pos += 4;
1163                Value::Subnet(ip, mask)
1164            }
1165            33 => {
1166                let val = u16::from_le_bytes([buf[*pos], buf[*pos + 1]]);
1167                *pos += 2;
1168                Value::Port(val)
1169            }
1170            34 => {
1171                let val =
1172                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1173                *pos += 4;
1174                Value::Latitude(val)
1175            }
1176            35 => {
1177                let val =
1178                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1179                *pos += 4;
1180                Value::Longitude(val)
1181            }
1182            36 => {
1183                let lat =
1184                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1185                *pos += 4;
1186                let lon =
1187                    i32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1188                *pos += 4;
1189                Value::GeoPoint(lat, lon)
1190            }
1191            37 => {
1192                let c = [buf[*pos], buf[*pos + 1]];
1193                *pos += 2;
1194                Value::Country2(c)
1195            }
1196            38 => {
1197                let c = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1198                *pos += 3;
1199                Value::Country3(c)
1200            }
1201            39 => {
1202                let c = [buf[*pos], buf[*pos + 1]];
1203                *pos += 2;
1204                Value::Lang2(c)
1205            }
1206            40 => {
1207                let c = [
1208                    buf[*pos],
1209                    buf[*pos + 1],
1210                    buf[*pos + 2],
1211                    buf[*pos + 3],
1212                    buf[*pos + 4],
1213                ];
1214                *pos += 5;
1215                Value::Lang5(c)
1216            }
1217            41 => {
1218                let c = [buf[*pos], buf[*pos + 1], buf[*pos + 2]];
1219                *pos += 3;
1220                Value::Currency(c)
1221            }
1222            42 => {
1223                let rgba = [buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]];
1224                *pos += 4;
1225                Value::ColorAlpha(rgba)
1226            }
1227            43 => {
1228                let val = i64::from_le_bytes([
1229                    buf[*pos],
1230                    buf[*pos + 1],
1231                    buf[*pos + 2],
1232                    buf[*pos + 3],
1233                    buf[*pos + 4],
1234                    buf[*pos + 5],
1235                    buf[*pos + 6],
1236                    buf[*pos + 7],
1237                ]);
1238                *pos += 8;
1239                Value::BigInt(val)
1240            }
1241            44 => {
1242                let col_len = Self::read_varu32_safe(buf, pos)?;
1243                let col = String::from_utf8(buf[*pos..*pos + col_len].to_vec())?;
1244                *pos += col_len;
1245                let key_len = Self::read_varu32_safe(buf, pos)?;
1246                let key = String::from_utf8(buf[*pos..*pos + key_len].to_vec())?;
1247                *pos += key_len;
1248                Value::KeyRef(col, key)
1249            }
1250            45 => {
1251                let col_len = Self::read_varu32_safe(buf, pos)?;
1252                let col = String::from_utf8(buf[*pos..*pos + col_len].to_vec())?;
1253                *pos += col_len;
1254                let id = u64::from_le_bytes([
1255                    buf[*pos],
1256                    buf[*pos + 1],
1257                    buf[*pos + 2],
1258                    buf[*pos + 3],
1259                    buf[*pos + 4],
1260                    buf[*pos + 5],
1261                    buf[*pos + 6],
1262                    buf[*pos + 7],
1263                ]);
1264                *pos += 8;
1265                Value::DocRef(col, id)
1266            }
1267            46 => {
1268                let len = Self::read_varu32_safe(buf, pos)?;
1269                let name = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1270                *pos += len;
1271                Value::TableRef(name)
1272            }
1273            47 => {
1274                let val =
1275                    u32::from_le_bytes([buf[*pos], buf[*pos + 1], buf[*pos + 2], buf[*pos + 3]]);
1276                *pos += 4;
1277                Value::PageRef(val)
1278            }
1279            48 => {
1280                let len = Self::read_varu32_safe(buf, pos)?;
1281                let bytes = buf[*pos..*pos + len].to_vec();
1282                *pos += len;
1283                Value::Secret(bytes)
1284            }
1285            49 => {
1286                let len = Self::read_varu32_safe(buf, pos)?;
1287                let hash = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1288                *pos += len;
1289                Value::Password(hash)
1290            }
1291            50 => {
1292                let len = Self::read_varu32_safe(buf, pos)?;
1293                let code = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1294                *pos += len;
1295                Value::AssetCode(code)
1296            }
1297            51 => {
1298                let len = Self::read_varu32_safe(buf, pos)?;
1299                let asset_code = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1300                *pos += len;
1301                let scale = buf[*pos];
1302                *pos += 1;
1303                let minor_units = i64::from_le_bytes([
1304                    buf[*pos],
1305                    buf[*pos + 1],
1306                    buf[*pos + 2],
1307                    buf[*pos + 3],
1308                    buf[*pos + 4],
1309                    buf[*pos + 5],
1310                    buf[*pos + 6],
1311                    buf[*pos + 7],
1312                ]);
1313                *pos += 8;
1314                Value::Money {
1315                    asset_code,
1316                    minor_units,
1317                    scale,
1318                }
1319            }
1320            52 => {
1321                let len = Self::read_varu32_safe(buf, pos)?;
1322                let value = String::from_utf8(buf[*pos..*pos + len].to_vec())?;
1323                *pos += len;
1324                Value::DecimalText(value)
1325            }
1326            // C3 TOAST: compressed Text (0x85) and Blob (0x86).
1327            0x85 => {
1328                let orig_len = Self::read_varu32_safe(buf, pos)?;
1329                let comp_len = Self::read_varu32_safe(buf, pos)?;
1330                let compressed = &buf[*pos..*pos + comp_len];
1331                *pos += comp_len;
1332                let mut out = vec![0u8; orig_len];
1333                zstd::bulk::decompress_to_buffer(compressed, &mut out)
1334                    .map_err(|e| format!("C3 Text decompress: {e}"))?;
1335                Value::text(String::from_utf8(out).map_err(|e| format!("C3 Text UTF-8: {e}"))?)
1336            }
1337            0x86 => {
1338                let orig_len = Self::read_varu32_safe(buf, pos)?;
1339                let comp_len = Self::read_varu32_safe(buf, pos)?;
1340                let compressed = &buf[*pos..*pos + comp_len];
1341                *pos += comp_len;
1342                let mut out = vec![0u8; orig_len];
1343                zstd::bulk::decompress_to_buffer(compressed, &mut out)
1344                    .map_err(|e| format!("C3 Blob decompress: {e}"))?;
1345                Value::Blob(out)
1346            }
1347            _ => return Err(format!("Unknown Value type: {}", type_byte).into()),
1348        })
1349    }
1350
1351    /// Write a Value to binary buffer
1352    /// Type bytes: 0=Null, 1=Boolean, 2=Integer, 3=UnsignedInteger, 4=Float,
1353    /// 5=Text, 6=Blob, 7=Timestamp, 8=Duration, 9=IpAddr, 10=MacAddr,
1354    /// 11=Vector, 12=Json, 13=Uuid, 14=NodeRef, 15=EdgeRef, 16=VectorRef, 17=RowRef
1355    fn write_value_binary(buf: &mut Vec<u8>, value: &Value) {
1356        use std::net::IpAddr;
1357
1358        match value {
1359            Value::Null => buf.push(0),
1360            Value::Boolean(b) => {
1361                buf.push(1);
1362                buf.push(if *b { 1 } else { 0 });
1363            }
1364            Value::Integer(i) => {
1365                buf.push(2);
1366                buf.extend_from_slice(&i.to_le_bytes());
1367            }
1368            Value::UnsignedInteger(u) => {
1369                buf.push(3);
1370                buf.extend_from_slice(&u.to_le_bytes());
1371            }
1372            Value::Float(f) => {
1373                buf.push(4);
1374                buf.extend_from_slice(&f.to_le_bytes());
1375            }
1376            Value::Text(s) => {
1377                // C3 TOAST: compress Text values > 2 KiB with zstd.
1378                // Type 0x85 = compressed Text; type 5 = uncompressed (existing).
1379                // Layout for 0x85: [0x85][orig_len: varuint][comp_len: varuint][compressed bytes]
1380                const TEXT_COMPRESS_THRESHOLD: usize = 2048;
1381                if s.len() >= TEXT_COMPRESS_THRESHOLD {
1382                    if let Ok(compressed) = zstd::bulk::compress(s.as_bytes(), 3) {
1383                        if compressed.len() < s.len() {
1384                            buf.push(0x85); // compressed Text marker
1385                            write_varu32(buf, s.len() as u32); // orig_len (decompression hint)
1386                            write_varu32(buf, compressed.len() as u32);
1387                            buf.extend_from_slice(&compressed);
1388                            return; // written — skip uncompressed path
1389                        }
1390                    }
1391                }
1392                buf.push(5);
1393                write_varu32(buf, s.len() as u32);
1394                buf.extend_from_slice(s.as_bytes());
1395            }
1396            Value::Blob(bytes) => {
1397                // C3 TOAST: compress Blob values > 2 KiB with zstd.
1398                // Type 0x86 = compressed Blob; type 6 = uncompressed (existing).
1399                const BLOB_COMPRESS_THRESHOLD: usize = 2048;
1400                if bytes.len() >= BLOB_COMPRESS_THRESHOLD {
1401                    if let Ok(compressed) = zstd::bulk::compress(bytes.as_slice(), 3) {
1402                        if compressed.len() < bytes.len() {
1403                            buf.push(0x86); // compressed Blob marker
1404                            write_varu32(buf, bytes.len() as u32);
1405                            write_varu32(buf, compressed.len() as u32);
1406                            buf.extend_from_slice(&compressed);
1407                            return;
1408                        }
1409                    }
1410                }
1411                buf.push(6);
1412                write_varu32(buf, bytes.len() as u32);
1413                buf.extend_from_slice(bytes);
1414            }
1415            Value::Timestamp(t) => {
1416                buf.push(7);
1417                buf.extend_from_slice(&t.to_le_bytes());
1418            }
1419            Value::Duration(d) => {
1420                buf.push(8);
1421                buf.extend_from_slice(&d.to_le_bytes());
1422            }
1423            Value::IpAddr(ip) => {
1424                buf.push(9);
1425                match ip {
1426                    IpAddr::V4(v4) => {
1427                        buf.push(4);
1428                        buf.extend_from_slice(&v4.octets());
1429                    }
1430                    IpAddr::V6(v6) => {
1431                        buf.push(6);
1432                        buf.extend_from_slice(&v6.octets());
1433                    }
1434                }
1435            }
1436            Value::MacAddr(mac) => {
1437                buf.push(10);
1438                buf.extend_from_slice(mac);
1439            }
1440            Value::Vector(vec) => {
1441                buf.push(11);
1442                write_varu32(buf, vec.len() as u32);
1443                for f in vec {
1444                    buf.extend_from_slice(&f.to_le_bytes());
1445                }
1446            }
1447            Value::Json(bytes) => {
1448                buf.push(12);
1449                write_varu32(buf, bytes.len() as u32);
1450                buf.extend_from_slice(bytes);
1451            }
1452            Value::Uuid(uuid) => {
1453                buf.push(13);
1454                buf.extend_from_slice(uuid);
1455            }
1456            Value::NodeRef(s) => {
1457                buf.push(14);
1458                write_varu32(buf, s.len() as u32);
1459                buf.extend_from_slice(s.as_bytes());
1460            }
1461            Value::EdgeRef(s) => {
1462                buf.push(15);
1463                write_varu32(buf, s.len() as u32);
1464                buf.extend_from_slice(s.as_bytes());
1465            }
1466            Value::VectorRef(s, id) => {
1467                buf.push(16);
1468                write_varu32(buf, s.len() as u32);
1469                buf.extend_from_slice(s.as_bytes());
1470                buf.extend_from_slice(&id.to_le_bytes());
1471            }
1472            Value::RowRef(s, id) => {
1473                buf.push(17);
1474                write_varu32(buf, s.len() as u32);
1475                buf.extend_from_slice(s.as_bytes());
1476                buf.extend_from_slice(&id.to_le_bytes());
1477            }
1478            Value::Color(rgb) => {
1479                buf.push(18);
1480                buf.extend_from_slice(rgb);
1481            }
1482            Value::Email(s) => {
1483                buf.push(19);
1484                write_varu32(buf, s.len() as u32);
1485                buf.extend_from_slice(s.as_bytes());
1486            }
1487            Value::Url(s) => {
1488                buf.push(20);
1489                write_varu32(buf, s.len() as u32);
1490                buf.extend_from_slice(s.as_bytes());
1491            }
1492            Value::Phone(n) => {
1493                buf.push(21);
1494                buf.extend_from_slice(&n.to_le_bytes());
1495            }
1496            Value::Semver(v) => {
1497                buf.push(22);
1498                buf.extend_from_slice(&v.to_le_bytes());
1499            }
1500            Value::Cidr(ip, prefix) => {
1501                buf.push(23);
1502                buf.extend_from_slice(&ip.to_le_bytes());
1503                buf.push(*prefix);
1504            }
1505            Value::Date(d) => {
1506                buf.push(24);
1507                buf.extend_from_slice(&d.to_le_bytes());
1508            }
1509            Value::Time(t) => {
1510                buf.push(25);
1511                buf.extend_from_slice(&t.to_le_bytes());
1512            }
1513            Value::Decimal(v) => {
1514                buf.push(26);
1515                buf.extend_from_slice(&v.to_le_bytes());
1516            }
1517            Value::EnumValue(i) => {
1518                buf.push(27);
1519                buf.push(*i);
1520            }
1521            Value::Array(elems) => {
1522                buf.push(28);
1523                write_varu32(buf, elems.len() as u32);
1524                for elem in elems {
1525                    Self::write_value_binary(buf, elem);
1526                }
1527            }
1528            Value::TimestampMs(v) => {
1529                buf.push(29);
1530                buf.extend_from_slice(&v.to_le_bytes());
1531            }
1532            Value::Ipv4(v) => {
1533                buf.push(30);
1534                buf.extend_from_slice(&v.to_le_bytes());
1535            }
1536            Value::Ipv6(bytes) => {
1537                buf.push(31);
1538                buf.extend_from_slice(bytes);
1539            }
1540            Value::Subnet(ip, mask) => {
1541                buf.push(32);
1542                buf.extend_from_slice(&ip.to_le_bytes());
1543                buf.extend_from_slice(&mask.to_le_bytes());
1544            }
1545            Value::Port(v) => {
1546                buf.push(33);
1547                buf.extend_from_slice(&v.to_le_bytes());
1548            }
1549            Value::Latitude(v) => {
1550                buf.push(34);
1551                buf.extend_from_slice(&v.to_le_bytes());
1552            }
1553            Value::Longitude(v) => {
1554                buf.push(35);
1555                buf.extend_from_slice(&v.to_le_bytes());
1556            }
1557            Value::GeoPoint(lat, lon) => {
1558                buf.push(36);
1559                buf.extend_from_slice(&lat.to_le_bytes());
1560                buf.extend_from_slice(&lon.to_le_bytes());
1561            }
1562            Value::Country2(c) => {
1563                buf.push(37);
1564                buf.extend_from_slice(c);
1565            }
1566            Value::Country3(c) => {
1567                buf.push(38);
1568                buf.extend_from_slice(c);
1569            }
1570            Value::Lang2(c) => {
1571                buf.push(39);
1572                buf.extend_from_slice(c);
1573            }
1574            Value::Lang5(c) => {
1575                buf.push(40);
1576                buf.extend_from_slice(c);
1577            }
1578            Value::Currency(c) => {
1579                buf.push(41);
1580                buf.extend_from_slice(c);
1581            }
1582            Value::AssetCode(code) => {
1583                buf.push(50);
1584                write_varu32(buf, code.len() as u32);
1585                buf.extend_from_slice(code.as_bytes());
1586            }
1587            Value::Money {
1588                asset_code,
1589                minor_units,
1590                scale,
1591            } => {
1592                buf.push(51);
1593                write_varu32(buf, asset_code.len() as u32);
1594                buf.extend_from_slice(asset_code.as_bytes());
1595                buf.push(*scale);
1596                buf.extend_from_slice(&minor_units.to_le_bytes());
1597            }
1598            Value::ColorAlpha(rgba) => {
1599                buf.push(42);
1600                buf.extend_from_slice(rgba);
1601            }
1602            Value::BigInt(v) => {
1603                buf.push(43);
1604                buf.extend_from_slice(&v.to_le_bytes());
1605            }
1606            Value::KeyRef(col, key) => {
1607                buf.push(44);
1608                write_varu32(buf, col.len() as u32);
1609                buf.extend_from_slice(col.as_bytes());
1610                write_varu32(buf, key.len() as u32);
1611                buf.extend_from_slice(key.as_bytes());
1612            }
1613            Value::DocRef(col, id) => {
1614                buf.push(45);
1615                write_varu32(buf, col.len() as u32);
1616                buf.extend_from_slice(col.as_bytes());
1617                buf.extend_from_slice(&id.to_le_bytes());
1618            }
1619            Value::TableRef(name) => {
1620                buf.push(46);
1621                write_varu32(buf, name.len() as u32);
1622                buf.extend_from_slice(name.as_bytes());
1623            }
1624            Value::PageRef(page_id) => {
1625                buf.push(47);
1626                buf.extend_from_slice(&page_id.to_le_bytes());
1627            }
1628            Value::Secret(bytes) => {
1629                buf.push(48);
1630                write_varu32(buf, bytes.len() as u32);
1631                buf.extend_from_slice(bytes);
1632            }
1633            Value::Password(hash) => {
1634                buf.push(49);
1635                write_varu32(buf, hash.len() as u32);
1636                buf.extend_from_slice(hash.as_bytes());
1637            }
1638            Value::DecimalText(value) => {
1639                buf.push(52);
1640                write_varu32(buf, value.len() as u32);
1641                buf.extend_from_slice(value.as_bytes());
1642            }
1643        }
1644    }
1645}
1646
1647#[cfg(test)]
1648mod aux_metadata_dump_tests {
1649    use super::*;
1650    use std::collections::HashMap;
1651
1652    #[test]
1653    fn aux_metadata_round_trips_through_binary_dump() {
1654        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
1655        store.create_collection("k").expect("create collection");
1656        store.set_aux_metadata(b"contract-blob".to_vec());
1657
1658        let bytes = store.to_binary_dump_bytes();
1659        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
1660
1661        assert_eq!(reloaded.aux_metadata(), b"contract-blob".to_vec());
1662    }
1663
1664    #[test]
1665    fn empty_aux_metadata_round_trips_as_empty() {
1666        let store = UnifiedStore::with_config(UnifiedStoreConfig::default());
1667        let bytes = store.to_binary_dump_bytes();
1668        let reloaded = UnifiedStore::load_from_bytes(&bytes).expect("reload dump");
1669
1670        assert!(reloaded.aux_metadata().is_empty());
1671    }
1672
1673    #[test]
1674    fn interned_timeseries_series_reduces_repeated_tag_storage() {
1675        let tags = HashMap::from([
1676            (
1677                "host".to_string(),
1678                "srv-very-long-hostname-0001".to_string(),
1679            ),
1680            ("region".to_string(), "us-east-1".to_string()),
1681            ("service".to_string(), "checkout-api".to_string()),
1682        ]);
1683        let inline_bytes: usize = (0..100)
1684            .map(|timestamp_ns| {
1685                let entity = UnifiedEntity::new(
1686                    EntityId::new(0),
1687                    EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
1688                        series: "cpu_metrics".to_string(),
1689                        metric: "cpu.idle".to_string(),
1690                    })),
1691                    EntityData::TimeSeries(crate::storage::TimeSeriesData {
1692                        metric: "cpu.idle".to_string(),
1693                        series_id: None,
1694                        timestamp_ns,
1695                        value: 94.8,
1696                        tags: tags.clone(),
1697                        fields: HashMap::new(),
1698                    }),
1699                );
1700                UnifiedStore::serialize_entity_record(&entity, None, reddb_file::STORE_VERSION_V10)
1701                    .len()
1702            })
1703            .sum();
1704
1705        let mut dictionary_fields = HashMap::new();
1706        dictionary_fields.insert("collection".to_string(), Value::text("cpu_metrics"));
1707        dictionary_fields.insert("series_id".to_string(), Value::UnsignedInteger(0));
1708        dictionary_fields.insert("metric".to_string(), Value::text("cpu.idle"));
1709        dictionary_fields.insert(
1710            "canonical_tags".to_string(),
1711            Value::text(
1712                r#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#,
1713            ),
1714        );
1715        dictionary_fields.insert(
1716            "tags".to_string(),
1717            Value::Json(
1718                br#"{"host":"srv-very-long-hostname-0001","region":"us-east-1","service":"checkout-api"}"#
1719                    .to_vec(),
1720            ),
1721        );
1722        let dictionary = UnifiedEntity::new(
1723            EntityId::new(0),
1724            EntityKind::TableRow {
1725                table: Arc::from("red_timeseries_series"),
1726                row_id: 0,
1727            },
1728            EntityData::Row(RowData {
1729                columns: Vec::new(),
1730                named: Some(dictionary_fields),
1731                schema: None,
1732            }),
1733        );
1734        let dictionary_bytes =
1735            UnifiedStore::serialize_entity_record(&dictionary, None, STORE_VERSION_V11).len();
1736        let interned_bytes = dictionary_bytes
1737            + (0..100)
1738                .map(|timestamp_ns| {
1739                    let entity = UnifiedEntity::new(
1740                        EntityId::new(0),
1741                        EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
1742                            series: "cpu_metrics".to_string(),
1743                            metric: "cpu.idle".to_string(),
1744                        })),
1745                        EntityData::TimeSeries(crate::storage::TimeSeriesData {
1746                            metric: "cpu.idle".to_string(),
1747                            series_id: Some(0),
1748                            timestamp_ns,
1749                            value: 94.8,
1750                            tags: HashMap::new(),
1751                            fields: HashMap::new(),
1752                        }),
1753                    );
1754                    UnifiedStore::serialize_entity_record(&entity, None, STORE_VERSION_V11).len()
1755                })
1756                .sum::<usize>();
1757
1758        assert!(
1759            interned_bytes < inline_bytes,
1760            "interned series storage should shrink repeated tags: inline={inline_bytes}, interned={interned_bytes}"
1761        );
1762    }
1763}
reddb_server/storage/unified/store/impl_file.rs

reddb_server/storage/unified/store/
impl_file.rs
