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

reddb_engine/storage/
entity_codec.rs
