reddb_engine/storage/
entity_codec.rs

1//! Canonical logical entity adapter for native store versions V1 through V14.
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    if format_version >= STORE_VERSION_V14 {
304        buf.push(u8::from(entity.branch_version().is_some()));
305        if let Some(branch) = entity.branch_version() {
306            write_varu64(buf, branch.origin);
307            write_varu32(buf, branch.deletions.len() as u32);
308            for &(epoch, xid) in &branch.deletions {
309                write_varu64(buf, epoch);
310                write_varu64(buf, xid);
311            }
312        }
313    }
314}
315
316/// Decode one entity, retaining legacy optional-tail compatibility. All reads
317/// are checked before indexing or allocating from persisted lengths.
318pub fn read_entity_binary(
319    bytes: &[u8],
320    position: &mut usize,
321    format_version: u32,
322) -> RdbFileResult<UnifiedEntity> {
323    read_entity_from_reader(&mut NativeReader::new(bytes, position), format_version)
324}
325
326pub fn read_entity_binary_with_budget(
327    bytes: &[u8],
328    position: &mut usize,
329    format_version: u32,
330    budget: &mut reddb_file::native_value::DecodeBudget,
331) -> RdbFileResult<UnifiedEntity> {
332    budget.elements::<UnifiedEntity>(1)?;
333    read_entity_from_reader(
334        &mut NativeReader::with_budget(bytes, position, budget),
335        format_version,
336    )
337}
338
339fn read_entity_from_reader(
340    reader: &mut NativeReader<'_, '_>,
341    format_version: u32,
342) -> RdbFileResult<UnifiedEntity> {
343    let id = EntityId::new(reader.variable_u64()?);
344    let kind = match reader.byte()? {
345        0 => {
346            let table = super::system_collection_names::canonical_owned(reader.string()?);
347            reader.reserve_bytes(table.len().saturating_add(2 * std::mem::size_of::<usize>()))?;
348            EntityKind::TableRow {
349                table: table.into(),
350                row_id: reader.variable_u64()?,
351            }
352        }
353        1 => {
354            reader.reserve_elements::<GraphNodeKind>(1)?;
355            EntityKind::GraphNode(Box::new(GraphNodeKind {
356                label: reader.string()?,
357                node_type: reader.string()?,
358            }))
359        }
360        2 => {
361            reader.reserve_elements::<GraphEdgeKind>(1)?;
362            EntityKind::GraphEdge(Box::new(GraphEdgeKind {
363                label: reader.string()?,
364                from_node: reader.string()?,
365                to_node: reader.string()?,
366                weight: u32::from_le_bytes(reader.fixed()?),
367            }))
368        }
369        3 => EntityKind::Vector {
370            collection: super::system_collection_names::canonical_owned(reader.string()?),
371        },
372        4 => {
373            reader.reserve_elements::<TimeSeriesPointKind>(1)?;
374            EntityKind::TimeSeriesPoint(Box::new(TimeSeriesPointKind {
375                series: super::system_collection_names::canonical_owned(reader.string()?),
376                metric: reader.string()?,
377            }))
378        }
379        5 => EntityKind::QueueMessage {
380            queue: super::system_collection_names::canonical_owned(reader.string()?),
381            position: reader.variable_u64()?,
382        },
383        tag => return Err(invalid(format!("Unknown EntityKind type: {tag}"))),
384    };
385    let mut data = match reader.byte()? {
386        0 => {
387            let count = reader.count(1)?;
388            reader.reserve_elements::<reddb_types::Value>(count)?;
389            let mut columns = Vec::with_capacity(count);
390            for _ in 0..count {
391                columns.push(reader.value()?);
392            }
393            EntityData::Row(RowData::new(columns))
394        }
395        1 => EntityData::Node(NodeData::with_properties(read_properties(reader)?)),
396        2 => {
397            let mut edge = EdgeData::new(f32::from_le_bytes(reader.fixed()?));
398            edge.properties = read_properties(reader)?;
399            EntityData::Edge(edge)
400        }
401        3 => {
402            let mut vector = VectorData::new(read_vector(reader)?);
403            if format_version >= STORE_VERSION_V6 && reader.byte()? != 0 {
404                vector.content = Some(reader.string()?);
405            }
406            EntityData::Vector(vector)
407        }
408        4 => {
409            let metric = reader.string()?;
410            let timestamp_ns = reader.variable_u64()?;
411            let value = f64::from_le_bytes(reader.fixed()?);
412            let mut tags = HashMap::new();
413            let series_id = if format_version >= STORE_VERSION_V11 && reader.variable_u32()? == 1 {
414                Some(reader.variable_u64()?)
415            } else {
416                if format_version >= STORE_VERSION_V5 {
417                    let count = reader.count(2)?;
418                    reader.reserve_map::<String, String>(count)?;
419                    tags = HashMap::with_capacity(count);
420                    for _ in 0..count {
421                        tags.insert(reader.string()?, reader.string()?);
422                    }
423                }
424                None
425            };
426            let fields = if format_version >= STORE_VERSION_V12 {
427                read_properties(reader)?
428            } else {
429                HashMap::new()
430            };
431            EntityData::TimeSeries(TimeSeriesData {
432                metric,
433                timestamp_ns,
434                value,
435                series_id,
436                tags,
437                fields,
438            })
439        }
440        5 => EntityData::QueueMessage(QueueMessageData {
441            payload: reader.value()?,
442            priority: None,
443            enqueued_at_ns: reader.variable_u64()?,
444            attempts: reader.variable_u32()?,
445            max_attempts: 3,
446            acked: false,
447        }),
448        6 => EntityData::Row(RowData {
449            columns: Vec::new(),
450            named: Some(read_properties(reader)?),
451            schema: None,
452        }),
453        tag => return Err(invalid(format!("Unknown EntityData type: {tag}"))),
454    };
455    let created_at = reader.variable_u64()?;
456    let updated_at = reader.variable_u64()?;
457    let count = reader.count(3)?;
458    reader.reserve_elements::<EmbeddingSlot>(count)?;
459    reader.reserve_elements::<EmbeddingSlot>(count)?;
460    let mut embeddings = Vec::with_capacity(count);
461    for _ in 0..count {
462        let name = reader.string()?;
463        let vector = read_vector(reader)?;
464        let model = reader.string()?;
465        embeddings.push(EmbeddingSlot::new(name, vector, model));
466    }
467    let count = reader.count(3)?;
468    reader.reserve_elements::<CrossRef>(count)?;
469    reader.reserve_elements::<CrossRef>(count)?;
470    let mut cross_refs = Vec::with_capacity(count);
471    for _ in 0..count {
472        let source = EntityId::new(reader.variable_u64()?);
473        let target = EntityId::new(reader.variable_u64()?);
474        let ref_type = RefType::from_byte(reader.byte()?);
475        let (collection, weight, created_at) = if format_version >= STORE_VERSION_V2 {
476            (
477                super::system_collection_names::canonical_owned(reader.string()?),
478                f32::from_le_bytes(reader.fixed()?),
479                reader.variable_u64()?,
480            )
481        } else {
482            (String::new(), 1.0, 0)
483        };
484        let mut cross_ref = CrossRef::new(source, target, collection, ref_type);
485        cross_ref.weight = weight;
486        cross_ref.created_at = created_at;
487        cross_refs.push(cross_ref);
488    }
489    let sequence_id = reader.variable_u64()?;
490    if format_version >= STORE_VERSION_V4 {
491        if let EntityData::QueueMessage(message) = &mut data {
492            if reader.remaining() != 0 {
493                message.priority = if reader.byte()? != 0 {
494                    Some(i32::from_le_bytes(reader.fixed()?))
495                } else {
496                    None
497                };
498                message.max_attempts = reader.variable_u32()?;
499                message.acked = reader.byte()? != 0;
500            }
501        }
502    }
503    let mut entity = UnifiedEntity::new(id, kind, data);
504    entity.created_at = created_at;
505    entity.updated_at = updated_at;
506    entity.sequence_id = sequence_id;
507    if format_version >= STORE_VERSION_V8 && reader.remaining() != 0 && reader.byte()? != 0 {
508        entity.set_logical_id(EntityId::new(reader.variable_u64()?));
509    }
510    if format_version >= STORE_VERSION_V9 && reader.remaining() != 0 {
511        entity.set_xmin(reader.variable_u64()?);
512        entity.set_xmax(reader.variable_u64()?);
513    }
514    if format_version >= STORE_VERSION_V14 {
515        match reader.byte()? {
516            0 => {}
517            1 => {
518                let origin = reader.variable_u64()?;
519                reader.reserve_elements::<EntityAux>(1)?;
520                entity.set_branch_origin(origin);
521                let count = reader.count(2)?;
522                reader.reserve_elements::<(u64, u64)>(count)?;
523                let mut previous = None;
524                for _ in 0..count {
525                    let epoch = reader.variable_u64()?;
526                    let xid = reader.variable_u64()?;
527                    if epoch == origin || xid == 0 || previous.is_some_and(|old| old >= epoch) {
528                        return Err(invalid("invalid branch deletion stamp"));
529                    }
530                    entity.set_branch_xmax(epoch, xid);
531                    previous = Some(epoch);
532                }
533            }
534            _ => return Err(invalid("invalid branch provenance flag")),
535        }
536    }
537    if !embeddings.is_empty() || !cross_refs.is_empty() {
538        reader.reserve_elements::<EntityAux>(1)?;
539        entity.embeddings_mut().extend(embeddings);
540        entity.cross_refs_mut().extend(cross_refs);
541    }
542    Ok(entity)
543}
544
545fn read_properties(
546    reader: &mut NativeReader<'_, '_>,
547) -> RdbFileResult<HashMap<String, reddb_types::Value>> {
548    let count = reader.count(2)?;
549    reader.reserve_map::<String, reddb_types::Value>(count)?;
550    let mut properties = HashMap::with_capacity(count);
551    for _ in 0..count {
552        properties.insert(reader.string()?, reader.value()?);
553    }
554    Ok(properties)
555}
556
557fn read_vector(reader: &mut NativeReader<'_, '_>) -> RdbFileResult<Vec<f32>> {
558    let count = reader.count(4)?;
559    reader.reserve_elements::<f32>(count)?;
560    let mut vector = Vec::with_capacity(count);
561    for _ in 0..count {
562        vector.push(f32::from_le_bytes(reader.fixed()?));
563    }
564    Ok(vector)
565}
566
567fn invalid(error: impl std::fmt::Display) -> RdbFileError {
568    RdbFileError::InvalidOperation(error.to_string())
569}
reddb_engine/storage/entity_codec.rs

reddb_engine/storage/
entity_codec.rs
