nodedb 0.4.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

// SPDX-License-Identifier: BUSL-1.1

//! Encode `PhysicalPlan::Columnar` / `Timeseries` / `Text` / `Spatial`
//! sync-engine variants into `ReplicatedWrite`.

use super::super::types::ReplicatedWrite;
use nodedb_types::Surrogate;

pub(super) fn columnar_ingest(
    collection: &str,
    payload: &[u8],
    surrogates: &[Surrogate],
    schema_bytes: &[u8],
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::ColumnarIngest {
        collection: collection.to_owned(),
        payload: payload.to_vec(),
        schema_bytes: schema_bytes.to_vec(),
        surrogates: surrogates.iter().map(|s| s.as_u32()).collect(),
        provenance,
    }
}

pub(super) fn timeseries_ingest(
    collection: &str,
    payload: &[u8],
    format: &str,
    surrogates: &[Surrogate],
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::TimeseriesIngest {
        collection: collection.to_owned(),
        payload: payload.to_vec(),
        format: format.to_owned(),
        surrogates: surrogates.iter().map(|s| s.as_u32()).collect(),
        provenance,
    }
}

pub(super) fn fts_index(
    collection: &str,
    surrogate: u32,
    text: &str,
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::FtsIndex {
        collection: collection.to_owned(),
        surrogate,
        text: text.to_owned(),
        provenance,
    }
}

pub(super) fn fts_delete(
    collection: &str,
    surrogate: u32,
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::FtsDelete {
        collection: collection.to_owned(),
        surrogate,
        provenance,
    }
}

pub(super) fn spatial_insert(
    collection: &str,
    field: &str,
    surrogate: u32,
    geometry: &nodedb_types::geometry::Geometry,
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::SpatialInsert {
        collection: collection.to_owned(),
        field: field.to_owned(),
        surrogate,
        // Geometry is plain serializable data — encoding is infallible (same
        // contract as `ReplicatedEntry::to_bytes`). Fail loud rather than
        // replicate empty bytes that would error on follower decode.
        geometry_bytes: zerompk::to_msgpack_vec(geometry)
            .expect("Geometry serialization is infallible"),
        provenance,
    }
}

pub(super) fn spatial_delete(
    collection: &str,
    field: &str,
    surrogate: u32,
    provenance: Option<Vec<u8>>,
) -> ReplicatedWrite {
    ReplicatedWrite::SpatialDelete {
        collection: collection.to_owned(),
        field: field.to_owned(),
        surrogate,
        provenance,
    }
}

/// Columnar predicate DELETE / UPDATE replicates as a `ColumnarBulkDml`
/// entry: each replica re-scans local columnar state at the committed log
/// position and applies the predicate deterministically (Raft log order ⇒
/// identical prior state ⇒ identical matching set), exactly like the
/// Document `BulkDml` sibling.
pub(super) fn bulk_delete(collection: &str, filters: &[u8]) -> ReplicatedWrite {
    ReplicatedWrite::ColumnarBulkDml {
        collection: collection.to_owned(),
        filters: filters.to_vec(),
        is_update: false,
        updates: Vec::new(),
    }
}

pub(super) fn bulk_update(
    collection: &str,
    filters: &[u8],
    updates: &[(String, Vec<u8>)],
) -> ReplicatedWrite {
    ReplicatedWrite::ColumnarBulkDml {
        collection: collection.to_owned(),
        filters: filters.to_vec(),
        is_update: true,
        updates: updates.to_vec(),
    }
}