nodedb 0.3.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

// SPDX-License-Identifier: BUSL-1.1

//! Strict document encoding: Value/bytes → Binary Tuple.

use nodedb_types::columnar::StrictSchema;
use nodedb_types::value::Value;

use super::coerce::coerce_value;

/// Encode a `nodedb_types::Value::Object` as a Binary Tuple according to the schema.
///
/// Accepts zerompk bytes (from planner) and decodes them internally.
/// Missing nullable columns become NULL; missing non-nullable columns error.
pub fn bytes_to_binary_tuple(bytes: &[u8], schema: &StrictSchema) -> crate::Result<Vec<u8>> {
    let value =
        nodedb_types::value_from_msgpack(bytes).map_err(|e| crate::Error::Serialization {
            format: "msgpack".to_string(),
            detail: format!("zerompk decode: {e}"),
        })?;
    value_to_binary_tuple(&value, schema)
}

/// Encode a `nodedb_types::Value` as a Binary Tuple according to the schema.
pub fn value_to_binary_tuple(value: &Value, schema: &StrictSchema) -> crate::Result<Vec<u8>> {
    let map = match value {
        Value::Object(m) => m,
        _ => {
            return Err(crate::Error::BadRequest {
                detail: "strict value must be an Object".to_string(),
            });
        }
    };

    let schema_columns: std::collections::HashSet<&str> =
        schema.columns.iter().map(|c| c.name.as_str()).collect();
    if let Some(unknown) = map.keys().find(|k| !schema_columns.contains(k.as_str())) {
        return Err(crate::Error::BadRequest {
            detail: format!("unknown field '{unknown}' not present in strict schema"),
        });
    }

    let encoder = nodedb_strict::TupleEncoder::new(schema);
    let mut values = Vec::with_capacity(schema.columns.len());

    for col in &schema.columns {
        let field_val = map.get(&col.name);
        let typed = match field_val {
            None | Some(Value::Null) => {
                if !col.nullable {
                    return Err(crate::Error::BadRequest {
                        detail: format!("column '{}' is NOT NULL but no value provided", col.name),
                    });
                }
                Value::Null
            }
            Some(v) => coerce_value(v, &col.column_type, &col.name)?,
        };
        values.push(typed);
    }

    encoder
        .encode(&values)
        .map_err(|e| crate::Error::BadRequest {
            detail: format!("Binary Tuple encode: {e}"),
        })
}

/// Bitemporal variant: decode msgpack to `Value`, then encode as a Binary
/// Tuple with reserved slots 0/1/2 populated from the supplied timestamps.
pub fn bytes_to_binary_tuple_bitemporal(
    bytes: &[u8],
    schema: &StrictSchema,
    system_from_ms: i64,
    valid_from_ms: i64,
    valid_until_ms: i64,
) -> crate::Result<Vec<u8>> {
    let value =
        nodedb_types::value_from_msgpack(bytes).map_err(|e| crate::Error::Serialization {
            format: "msgpack".to_string(),
            detail: format!("zerompk decode: {e}"),
        })?;
    value_to_binary_tuple_bitemporal(
        &value,
        schema,
        system_from_ms,
        valid_from_ms,
        valid_until_ms,
    )
}

/// Bitemporal variant: encode a user-supplied `Value::Object` together
/// with the three reserved bitemporal timestamps.
pub fn value_to_binary_tuple_bitemporal(
    value: &Value,
    schema: &StrictSchema,
    system_from_ms: i64,
    valid_from_ms: i64,
    valid_until_ms: i64,
) -> crate::Result<Vec<u8>> {
    if !schema.bitemporal {
        return Err(crate::Error::BadRequest {
            detail: "schema is not bitemporal".to_string(),
        });
    }
    let map = match value {
        Value::Object(m) => m,
        _ => {
            return Err(crate::Error::BadRequest {
                detail: "strict value must be an Object".to_string(),
            });
        }
    };

    let user_columns = &schema.columns[3..];
    let user_names: std::collections::HashSet<&str> =
        user_columns.iter().map(|c| c.name.as_str()).collect();
    if let Some(unknown) = map.keys().find(|k| {
        !user_names.contains(k.as_str())
            && !nodedb_types::columnar::BITEMPORAL_RESERVED_COLUMNS.contains(&k.as_str())
    }) {
        return Err(crate::Error::BadRequest {
            detail: format!("unknown field '{unknown}' not present in strict schema"),
        });
    }

    let mut user_values = Vec::with_capacity(user_columns.len());
    for col in user_columns {
        let field_val = map.get(&col.name);
        let typed = match field_val {
            None | Some(Value::Null) => {
                if !col.nullable {
                    return Err(crate::Error::BadRequest {
                        detail: format!("column '{}' is NOT NULL but no value provided", col.name),
                    });
                }
                Value::Null
            }
            Some(v) => coerce_value(v, &col.column_type, &col.name)?,
        };
        user_values.push(typed);
    }

    let encoder = nodedb_strict::TupleEncoder::new(schema);
    encoder
        .encode_bitemporal(system_from_ms, valid_from_ms, valid_until_ms, &user_values)
        .map_err(|e| crate::Error::BadRequest {
            detail: format!("Binary Tuple encode: {e}"),
        })
}