reddb_engine/storage/
entity_codec.rs

1//! Canonical logical entity adapter for native store versions V1 through V12.
2//! Physical record framing and value bytes remain owned by `reddb-file`.
3use super::entity::*;
4use reddb_file::native_store::{
5    write_native_varu32 as write_varu32, write_native_varu64 as write_varu64,
6};
7use reddb_file::{
8    embedded::{RdbFileError, RdbFileResult},
9    native_store::*,
10    native_value::{write_native_value, NativeReader},
11};
12use std::collections::HashMap;
13
14fn persistent_name(name: &str) -> &str {
15    super::system_collection_names::canonical_or_self(name)
16}
17
18/// Write entity to binary buffer
19pub fn write_entity_binary(buf: &mut Vec<u8>, entity: &UnifiedEntity, format_version: u32) {
20    // Entity ID
21    write_varu64(buf, entity.id.raw());
22
23    // EntityKind
24    match &entity.kind {
25        EntityKind::TableRow { table, row_id } => {
26            let table = persistent_name(table);
27            buf.push(0);
28            write_varu32(buf, table.len() as u32);
29            buf.extend_from_slice(table.as_bytes());
30            write_varu64(buf, *row_id);
31        }
32        EntityKind::GraphNode(ref node) => {
33            buf.push(1);
34            write_varu32(buf, node.label.len() as u32);
35            buf.extend_from_slice(node.label.as_bytes());
36            write_varu32(buf, node.node_type.len() as u32);
37            buf.extend_from_slice(node.node_type.as_bytes());
38        }
39        EntityKind::GraphEdge(ref edge) => {
40            buf.push(2);
41            write_varu32(buf, edge.label.len() as u32);
42            buf.extend_from_slice(edge.label.as_bytes());
43            write_varu32(buf, edge.from_node.len() as u32);
44            buf.extend_from_slice(edge.from_node.as_bytes());
45            write_varu32(buf, edge.to_node.len() as u32);
46            buf.extend_from_slice(edge.to_node.as_bytes());
47            buf.extend_from_slice(&edge.weight.to_le_bytes());
48        }
49        EntityKind::Vector { collection } => {
50            let collection = persistent_name(collection);
51            buf.push(3);
52            write_varu32(buf, collection.len() as u32);
53            buf.extend_from_slice(collection.as_bytes());
54        }
55        EntityKind::TimeSeriesPoint(ref ts) => {
56            let series = persistent_name(&ts.series);
57            buf.push(4);
58            write_varu32(buf, series.len() as u32);
59            buf.extend_from_slice(series.as_bytes());
60            write_varu32(buf, ts.metric.len() as u32);
61            buf.extend_from_slice(ts.metric.as_bytes());
62        }
63        EntityKind::QueueMessage { queue, position } => {
64            let queue = persistent_name(queue);
65            buf.push(5);
66            write_varu32(buf, queue.len() as u32);
67            buf.extend_from_slice(queue.as_bytes());
68            write_varu64(buf, *position);
69        }
70    }
71
72    // EntityData
73    match &entity.data {
74        EntityData::Row(row) => {
75            if row.named.is_some()
76                || row
77                    .schema
78                    .as_ref()
79                    .is_some_and(|schema| schema.len() == row.columns.len())
80            {
81                // Both resident representations encode the same named record.
82                // Compaction must not lose field names on the next checkpoint.
83                buf.push(6);
84                let mut entries: Vec<_> = if let Some(named) = &row.named {
85                    named
86                        .iter()
87                        .map(|(name, value)| (name.as_str(), value))
88                        .collect()
89                } else {
90                    row.schema
91                        .as_ref()
92                        .expect("validated schema")
93                        .iter()
94                        .map(String::as_str)
95                        .zip(&row.columns)
96                        .collect()
97                };
98                write_varu32(buf, entries.len() as u32);
99                entries.sort_by(|a, b| a.0.cmp(b.0));
100                for (key, value) in entries {
101                    write_varu32(buf, key.len() as u32);
102                    buf.extend_from_slice(key.as_bytes());
103                    write_native_value(buf, value);
104                }
105            } else {
106                buf.push(0);
107                write_varu32(buf, row.columns.len() as u32);
108                for col in &row.columns {
109                    write_native_value(buf, col);
110                }
111            }
112        }
113        EntityData::Node(node) => {
114            buf.push(1);
115            write_varu32(buf, node.properties.len() as u32);
116            let mut entries: Vec<_> = node.properties.iter().collect();
117            entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
118            for (key, value) in entries {
119                write_varu32(buf, key.len() as u32);
120                buf.extend_from_slice(key.as_bytes());
121                write_native_value(buf, value);
122            }
123        }
124        EntityData::Edge(edge) => {
125            buf.push(2);
126            buf.extend_from_slice(&edge.weight.to_le_bytes());
127            write_varu32(buf, edge.properties.len() as u32);
128            let mut entries: Vec<_> = edge.properties.iter().collect();
129            entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
130            for (key, value) in entries {
131                write_varu32(buf, key.len() as u32);
132                buf.extend_from_slice(key.as_bytes());
133                write_native_value(buf, value);
134            }
135        }
136        EntityData::Vector(vec) => {
137            buf.push(3);
138            write_varu32(buf, vec.dense.len() as u32);
139            for f in &vec.dense {
140                buf.extend_from_slice(&f.to_le_bytes());
141            }
142            if format_version >= STORE_VERSION_V6 {
143                buf.push(u8::from(vec.content.is_some()));
144                if let Some(content) = &vec.content {
145                    write_varu32(buf, content.len() as u32);
146                    buf.extend_from_slice(content.as_bytes());
147                }
148            }
149        }
150        EntityData::TimeSeries(ts) => {
151            buf.push(4);
152            write_varu32(buf, ts.metric.len() as u32);
153            buf.extend_from_slice(ts.metric.as_bytes());
154            write_varu64(buf, ts.timestamp_ns);
155            buf.extend_from_slice(&ts.value.to_le_bytes());
156            if format_version >= STORE_VERSION_V11 {
157                // Discriminator: interned series id (1) vs inline tag map (0).
158                // Interned points (dictionary path) persist just the id;
159                // un-interned points (metrics remote-write, rollups) keep
160                // their tags inline so the metrics read paths, which build
161                // labels directly from `tags`, survive a reopen.
162                match ts.series_id {
163                    Some(series_id) => {
164                        write_varu32(buf, 1);
165                        write_varu64(buf, series_id);
166                    }
167                    None => {
168                        write_varu32(buf, 0);
169                        write_varu32(buf, ts.tags.len() as u32);
170                        let mut tag_entries: Vec<_> = ts.tags.iter().collect();
171                        tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
172                        for (key, value) in tag_entries {
173                            write_varu32(buf, key.len() as u32);
174                            buf.extend_from_slice(key.as_bytes());
175                            write_varu32(buf, value.len() as u32);
176                            buf.extend_from_slice(value.as_bytes());
177                        }
178                    }
179                }
180            } else if format_version >= STORE_VERSION_V5 {
181                write_varu32(buf, ts.tags.len() as u32);
182                let mut tag_entries: Vec<_> = ts.tags.iter().collect();
183                tag_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
184                for (key, value) in tag_entries {
185                    write_varu32(buf, key.len() as u32);
186                    buf.extend_from_slice(key.as_bytes());
187                    write_varu32(buf, value.len() as u32);
188                    buf.extend_from_slice(value.as_bytes());
189                }
190            }
191            if format_version >= STORE_VERSION_V12 {
192                write_varu32(buf, ts.fields.len() as u32);
193                let mut field_entries: Vec<_> = ts.fields.iter().collect();
194                field_entries.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()));
195                for (key, value) in field_entries {
196                    write_varu32(buf, key.len() as u32);
197                    buf.extend_from_slice(key.as_bytes());
198                    write_native_value(buf, value);
199                }
200            }
201        }
202        EntityData::QueueMessage(msg) => {
203            buf.push(5);
204            write_native_value(buf, &msg.payload);
205            write_varu64(buf, msg.enqueued_at_ns);
206            write_varu32(buf, msg.attempts);
207        }
208    }
209
210    // Timestamps
211    write_varu64(buf, entity.created_at);
212    write_varu64(buf, entity.updated_at);
213
214    // Embeddings
215    write_varu32(buf, entity.embeddings().len() as u32);
216    for emb in entity.embeddings() {
217        write_varu32(buf, emb.name.len() as u32);
218        buf.extend_from_slice(emb.name.as_bytes());
219        write_varu32(buf, emb.vector.len() as u32);
220        for f in &emb.vector {
221            buf.extend_from_slice(&f.to_le_bytes());
222        }
223        write_varu32(buf, emb.model.len() as u32);
224        buf.extend_from_slice(emb.model.as_bytes());
225    }
226
227    // Cross-refs
228    write_varu32(buf, entity.cross_refs().len() as u32);
229    for cross_ref in entity.cross_refs() {
230        write_varu64(buf, cross_ref.source.raw());
231        write_varu64(buf, cross_ref.target.raw());
232        buf.push(cross_ref.ref_type.to_byte());
233        if format_version >= STORE_VERSION_V2 {
234            let target_collection = persistent_name(&cross_ref.target_collection);
235            write_varu32(buf, target_collection.len() as u32);
236            buf.extend_from_slice(target_collection.as_bytes());
237            buf.extend_from_slice(&cross_ref.weight.to_le_bytes());
238            write_varu64(buf, cross_ref.created_at);
239        }
240    }
241
242    // Sequence ID
243    write_varu64(buf, entity.sequence_id);
244
245    if format_version >= STORE_VERSION_V4 {
246        if let EntityData::QueueMessage(message) = &entity.data {
247            buf.push(u8::from(message.priority.is_some()));
248            if let Some(priority) = message.priority {
249                buf.extend_from_slice(&priority.to_le_bytes());
250            }
251            write_varu32(buf, message.max_attempts);
252            buf.push(u8::from(message.acked));
253        }
254    }
255
256    if format_version >= STORE_VERSION_V8 {
257        buf.push(u8::from(entity.has_explicit_logical_id()));
258        if entity.has_explicit_logical_id() {
259            write_varu64(buf, entity.logical_id().raw());
260        }
261    }
262    if format_version >= STORE_VERSION_V9 {
263        write_varu64(buf, entity.xmin);
264        write_varu64(buf, entity.xmax);
265    }
266}
267
268/// Decode one entity, retaining legacy optional-tail compatibility. All reads
269/// are checked before indexing or allocating from persisted lengths.
270pub fn read_entity_binary(
271    bytes: &[u8],
272    position: &mut usize,
273    format_version: u32,
274) -> RdbFileResult<UnifiedEntity> {
275    read_entity_from_reader(&mut NativeReader::new(bytes, position), format_version)
276}
277
278pub fn read_entity_binary_with_budget(
279    bytes: &[u8],
280    position: &mut usize,
281    format_version: u32,
282    budget: &mut reddb_file::native_value::DecodeBudget,
283) -> RdbFileResult<UnifiedEntity> {
284    budget.elements::<UnifiedEntity>(1)?;
285    read_entity_from_reader(
286        &mut NativeReader::with_budget(bytes, position, budget),
287        format_version,
288    )
289}
290
291fn read_entity_from_reader(
292    reader: &mut NativeReader<'_, '_>,
293    format_version: u32,
294) -> RdbFileResult<UnifiedEntity> {
295    let id = EntityId::new(reader.variable_u64()?);
296    let kind = match reader.byte()? {
297        0 => {
298            let table = super::system_collection_names::canonical_owned(reader.string()?);
299            reader.reserve_bytes(table.len().saturating_add(2 * std::mem::size_of::<usize>()))?;
300            EntityKind::TableRow {
301                table: table.into(),
302                row_id: reader.variable_u64()?,
303            }
304        }
305        1 => {
306            reader.reserve_elements::<GraphNodeKind>(1)?;
307            EntityKind::GraphNode(Box::new(GraphNodeKind {
308                label: reader.string()?,
309                node_type: reader.string()?,
310            }))
311        }
312        2 => {
313            reader.reserve_elements::<GraphEdgeKind>(1)?;
314            EntityKind::GraphEdge(Box::new(GraphEdgeKind {
315                label: reader.string()?,
316                from_node: reader.string()?,
317                to_node: reader.string()?,
318                weight: u32::from_le_bytes(reader.fixed()?),
319            }))
320        }
321        3 => EntityKind::Vector {
322            collection: super::system_collection_names::canonical_owned(reader.string()?),
323        },
324        4 => {
325            reader.reserve_elements::<TimeSeriesPointKind>(1)?;
326            EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
327                series: super::system_collection_names::canonical_owned(reader.string()?),
328                metric: reader.string()?,
329            }))
330        }
331        5 => EntityKind::QueueMessage {
332            queue: super::system_collection_names::canonical_owned(reader.string()?),
333            position: reader.variable_u64()?,
334        },
335        tag => return Err(invalid(format!("Unknown EntityKind type: {tag}"))),
336    };
337    let mut data = match reader.byte()? {
338        0 => {
339            let count = reader.count(1)?;
340            reader.reserve_elements::<reddb_types::Value>(count)?;
341            let mut columns = Vec::with_capacity(count);
342            for _ in 0..count {
343                columns.push(reader.value()?);
344            }
345            EntityData::Row(RowData::new(columns))
346        }
347        1 => EntityData::Node(NodeData::with_properties(read_properties(reader)?)),
348        2 => {
349            let mut edge = EdgeData::new(f32::from_le_bytes(reader.fixed()?));
350            edge.properties = read_properties(reader)?;
351            EntityData::Edge(edge)
352        }
353        3 => {
354            let mut vector = VectorData::new(read_vector(reader)?);
355            if format_version >= STORE_VERSION_V6 && reader.byte()? != 0 {
356                vector.content = Some(reader.string()?);
357            }
358            EntityData::Vector(vector)
359        }
360        4 => {
361            let metric = reader.string()?;
362            let timestamp_ns = reader.variable_u64()?;
363            let value = f64::from_le_bytes(reader.fixed()?);
364            let mut tags = HashMap::new();
365            let series_id = if format_version >= STORE_VERSION_V11 && reader.variable_u32()? == 1 {
366                Some(reader.variable_u64()?)
367            } else {
368                if format_version >= STORE_VERSION_V5 {
369                    let count = reader.count(2)?;
370                    reader.reserve_map::<String, String>(count)?;
371                    tags = HashMap::with_capacity(count);
372                    for _ in 0..count {
373                        tags.insert(reader.string()?, reader.string()?);
374                    }
375                }
376                None
377            };
378            let fields = if format_version >= STORE_VERSION_V12 {
379                read_properties(reader)?
380            } else {
381                HashMap::new()
382            };
383            EntityData::TimeSeries(TimeSeriesData {
384                metric,
385                timestamp_ns,
386                value,
387                series_id,
388                tags,
389                fields,
390            })
391        }
392        5 => EntityData::QueueMessage(QueueMessageData {
393            payload: reader.value()?,
394            priority: None,
395            enqueued_at_ns: reader.variable_u64()?,
396            attempts: reader.variable_u32()?,
397            max_attempts: 3,
398            acked: false,
399        }),
400        6 => EntityData::Row(RowData {
401            columns: Vec::new(),
402            named: Some(read_properties(reader)?),
403            schema: None,
404        }),
405        tag => return Err(invalid(format!("Unknown EntityData type: {tag}"))),
406    };
407    let created_at = reader.variable_u64()?;
408    let updated_at = reader.variable_u64()?;
409    let count = reader.count(3)?;
410    reader.reserve_elements::<EmbeddingSlot>(count)?;
411    reader.reserve_elements::<EmbeddingSlot>(count)?;
412    let mut embeddings = Vec::with_capacity(count);
413    for _ in 0..count {
414        let name = reader.string()?;
415        let vector = read_vector(reader)?;
416        let model = reader.string()?;
417        embeddings.push(EmbeddingSlot::new(name, vector, model));
418    }
419    let count = reader.count(3)?;
420    reader.reserve_elements::<CrossRef>(count)?;
421    reader.reserve_elements::<CrossRef>(count)?;
422    let mut cross_refs = Vec::with_capacity(count);
423    for _ in 0..count {
424        let source = EntityId::new(reader.variable_u64()?);
425        let target = EntityId::new(reader.variable_u64()?);
426        let ref_type = RefType::from_byte(reader.byte()?);
427        let (collection, weight, created_at) = if format_version >= STORE_VERSION_V2 {
428            (
429                super::system_collection_names::canonical_owned(reader.string()?),
430                f32::from_le_bytes(reader.fixed()?),
431                reader.variable_u64()?,
432            )
433        } else {
434            (String::new(), 1.0, 0)
435        };
436        let mut cross_ref = CrossRef::new(source, target, collection, ref_type);
437        cross_ref.weight = weight;
438        cross_ref.created_at = created_at;
439        cross_refs.push(cross_ref);
440    }
441    let sequence_id = reader.variable_u64()?;
442    if format_version >= STORE_VERSION_V4 {
443        if let EntityData::QueueMessage(message) = &mut data {
444            if reader.remaining() != 0 {
445                message.priority = if reader.byte()? != 0 {
446                    Some(i32::from_le_bytes(reader.fixed()?))
447                } else {
448                    None
449                };
450                message.max_attempts = reader.variable_u32()?;
451                message.acked = reader.byte()? != 0;
452            }
453        }
454    }
455    let mut entity = UnifiedEntity::new(id, kind, data);
456    entity.created_at = created_at;
457    entity.updated_at = updated_at;
458    entity.sequence_id = sequence_id;
459    if format_version >= STORE_VERSION_V8 && reader.remaining() != 0 && reader.byte()? != 0 {
460        entity.set_logical_id(EntityId::new(reader.variable_u64()?));
461    }
462    if format_version >= STORE_VERSION_V9 && reader.remaining() != 0 {
463        entity.set_xmin(reader.variable_u64()?);
464        entity.set_xmax(reader.variable_u64()?);
465    }
466    if !embeddings.is_empty() || !cross_refs.is_empty() {
467        reader.reserve_elements::<EntityAux>(1)?;
468        entity.embeddings_mut().extend(embeddings);
469        entity.cross_refs_mut().extend(cross_refs);
470    }
471    Ok(entity)
472}
473
474fn read_properties(
475    reader: &mut NativeReader<'_, '_>,
476) -> RdbFileResult<HashMap<String, reddb_types::Value>> {
477    let count = reader.count(2)?;
478    reader.reserve_map::<String, reddb_types::Value>(count)?;
479    let mut properties = HashMap::with_capacity(count);
480    for _ in 0..count {
481        properties.insert(reader.string()?, reader.value()?);
482    }
483    Ok(properties)
484}
485
486fn read_vector(reader: &mut NativeReader<'_, '_>) -> RdbFileResult<Vec<f32>> {
487    let count = reader.count(4)?;
488    reader.reserve_elements::<f32>(count)?;
489    let mut vector = Vec::with_capacity(count);
490    for _ in 0..count {
491        vector.push(f32::from_le_bytes(reader.fixed()?));
492    }
493    Ok(vector)
494}
495
496fn invalid(error: impl std::fmt::Display) -> RdbFileError {
497    RdbFileError::InvalidOperation(error.to_string())
498}
reddb_engine/storage/entity_codec.rs

reddb_engine/storage/
entity_codec.rs
