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

reddb_engine/storage/
entity_codec.rs
