nodedb_array/codec/
tile_encode.rs

1// SPDX-License-Identifier: Apache-2.0
2
3// Structural sparse-tile encoder.
4//
5// encode_sparse_tile writes the following into `out` (BlockFraming is applied
6// by the segment writer around this payload):
7//
8//   [u8 tag]          CodecTag::Raw or CodecTag::Structural
9//   [u8 version = 1]  payload format version
10//   [u32 LE cell_count]
11//   [u32 LE axis_count]
12//   per axis: [u32 LE encoded_len][coord_rle payload]
13//   [u32 LE surrogates_len][fastlanes payload — live rows only]
14//   [u32 LE row_kinds_len][raw u8s]
15//   [u32 LE system_from_ms_len][gorilla payload — absent for Raw tag]
16//   [u32 LE valid_from_ms_len][gorilla payload]
17//   [u32 LE valid_until_ms_len][gorilla payload]
18//   [u32 LE attr_count]
19//   per attr: [u32 LE col_len][column_codec payload]
20//
21// For CodecTag::Raw, the full SparseTile is serialised with zerompk and
22// written verbatim after the two header bytes (tag + version). The decoder
23// mirrors this.
24
25use crate::codec::column_codec::{
26    encode_attr_col, encode_row_kinds, encode_surrogates, encode_timestamps_col,
27};
28use crate::codec::coord_rle::encode_coord_axis_rle;
29use crate::codec::tag::CodecTag;
30use crate::error::{ArrayError, ArrayResult};
31use crate::tile::sparse_tile::SparseTile;
32
33const PAYLOAD_VERSION: u8 = 1;
34
35/// Threshold below which we fall back to Raw (zerompk) encoding.
36const STRUCTURAL_MIN_CELLS: usize = 8;
37
38fn choose_tag(tile: &SparseTile) -> CodecTag {
39    let cell_count = tile.surrogates.len();
40    if cell_count < STRUCTURAL_MIN_CELLS {
41        return CodecTag::Raw;
42    }
43    // Sentinel-only: all row_kinds non-zero and no attr columns have data.
44    let all_sentinel = !tile.row_kinds.is_empty()
45        && tile.row_kinds.iter().all(|&k| k != 0)
46        && tile.attr_cols.iter().all(|col| col.is_empty());
47    if all_sentinel {
48        return CodecTag::Raw;
49    }
50    CodecTag::Structural
51}
52
53fn write_framed(chunk: &[u8], out: &mut Vec<u8>) {
54    out.extend_from_slice(&(chunk.len() as u32).to_le_bytes());
55    out.extend_from_slice(chunk);
56}
57
58/// Encode a `SparseTile` into `out`. The segment writer wraps this payload
59/// in BlockFraming (length + CRC).
60pub fn encode_sparse_tile(tile: &SparseTile, out: &mut Vec<u8>) -> ArrayResult<()> {
61    // A stored live cell always holds a bound surrogate. A derived result tile
62    // holds none and is refused here, before any byte is written.
63    tile.check_stored_identities()?;
64    let tag = choose_tag(tile);
65    out.push(tag.as_byte());
66    out.push(PAYLOAD_VERSION);
67
68    match tag {
69        CodecTag::Raw => encode_raw(tile, out),
70        CodecTag::Structural => encode_structural(tile, out),
71    }
72}
73
74fn encode_raw(tile: &SparseTile, out: &mut Vec<u8>) -> ArrayResult<()> {
75    let bytes = zerompk::to_msgpack_vec(tile).map_err(|e| ArrayError::SegmentCorruption {
76        detail: format!("raw tile encode: {e}"),
77    })?;
78    out.extend_from_slice(&bytes);
79    Ok(())
80}
81
82fn encode_structural(tile: &SparseTile, out: &mut Vec<u8>) -> ArrayResult<()> {
83    let cell_count = tile.surrogates.len() as u32;
84    let axis_count = tile.dim_dicts.len() as u32;
85    out.extend_from_slice(&cell_count.to_le_bytes());
86    out.extend_from_slice(&axis_count.to_le_bytes());
87
88    // Coordinate axes.
89    for dict in &tile.dim_dicts {
90        let mut axis_buf = Vec::new();
91        encode_coord_axis_rle(dict, &mut axis_buf)?;
92        write_framed(&axis_buf, out);
93    }
94
95    // Surrogates: live rows only, in row order. A tombstone or erasure row
96    // holds none, and the decoder restores the gaps from the row kinds.
97    let live_surrogates: Vec<_> = tile.surrogates.iter().copied().flatten().collect();
98    let surr_bytes = encode_surrogates(&live_surrogates)?;
99    write_framed(&surr_bytes, out);
100
101    // Row kinds.
102    let rk_bytes = encode_row_kinds(&tile.row_kinds);
103    write_framed(&rk_bytes, out);
104
105    // Timestamp columns.
106    let sys_bytes = encode_timestamps_col(&[]);
107    // system_from_ms is not stored in SparseTile (it's the tile_id's field).
108    // We emit an empty placeholder to keep the format symmetric.
109    write_framed(&sys_bytes, out);
110
111    let vf_bytes = encode_timestamps_col(&tile.valid_from_ms);
112    write_framed(&vf_bytes, out);
113
114    let vu_bytes = encode_timestamps_col(&tile.valid_until_ms);
115    write_framed(&vu_bytes, out);
116
117    // Attr columns.
118    let attr_count = tile.attr_cols.len() as u32;
119    out.extend_from_slice(&attr_count.to_le_bytes());
120    for col in &tile.attr_cols {
121        let col_bytes = encode_attr_col(col)?;
122        write_framed(&col_bytes, out);
123    }
124
125    Ok(())
126}
127
128#[cfg(test)]
129mod tests {
130    use super::*;
131    use crate::codec::tile_decode::decode_sparse_tile;
132    use crate::schema::ArraySchemaBuilder;
133    use crate::schema::attr_spec::{AttrSpec, AttrType};
134    use crate::schema::dim_spec::{DimSpec, DimType};
135    use crate::tile::sparse_tile::{RowKind, SparseRow, SparseTileBuilder};
136    use crate::types::cell_value::value::CellValue;
137    use crate::types::coord::value::CoordValue;
138    use crate::types::domain::{Domain, DomainBound};
139    use nodedb_types::{OPEN_UPPER, Surrogate};
140
141    fn schema() -> crate::schema::ArraySchema {
142        ArraySchemaBuilder::new("t")
143            .dim(DimSpec::new(
144                "x",
145                DimType::Int64,
146                Domain::new(DomainBound::Int64(0), DomainBound::Int64(1_000_000)),
147            ))
148            .dim(DimSpec::new(
149                "y",
150                DimType::Int64,
151                Domain::new(DomainBound::Int64(0), DomainBound::Int64(1_000_000)),
152            ))
153            .attr(AttrSpec::new("v", AttrType::Int64, true))
154            .tile_extents(vec![1000, 1000])
155            .build()
156            .unwrap()
157    }
158
159    fn make_tile(s: &crate::schema::ArraySchema, n: usize) -> SparseTile {
160        let mut b = SparseTileBuilder::new(s);
161        for i in 0..n {
162            b.push_row(SparseRow {
163                coord: &[CoordValue::Int64(i as i64), CoordValue::Int64(i as i64 * 2)],
164                attrs: &[CellValue::Int64(i as i64)],
165                surrogate: Some(Surrogate::new(i as u32 + 1)),
166                valid_from_ms: i as i64 * 10,
167                valid_until_ms: OPEN_UPPER,
168                kind: RowKind::Live,
169            })
170            .unwrap();
171        }
172        b.build()
173    }
174
175    #[test]
176    fn small_tile_uses_raw_tag() {
177        let s = schema();
178        let tile = make_tile(&s, 3);
179        let mut buf = Vec::new();
180        encode_sparse_tile(&tile, &mut buf).unwrap();
181        assert_eq!(buf[0], CodecTag::Raw.as_byte());
182    }
183
184    #[test]
185    fn large_tile_uses_structural_tag() {
186        let s = schema();
187        let tile = make_tile(&s, 20);
188        let mut buf = Vec::new();
189        encode_sparse_tile(&tile, &mut buf).unwrap();
190        assert_eq!(buf[0], CodecTag::Structural.as_byte());
191    }
192
193    #[test]
194    fn small_tile_roundtrip() {
195        let s = schema();
196        let tile = make_tile(&s, 5);
197        let mut buf = Vec::new();
198        encode_sparse_tile(&tile, &mut buf).unwrap();
199        let decoded = decode_sparse_tile(&buf).unwrap();
200        assert_eq!(decoded.surrogates, tile.surrogates);
201        assert_eq!(decoded.valid_from_ms, tile.valid_from_ms);
202        assert_eq!(decoded.row_kinds, tile.row_kinds);
203    }
204
205    #[test]
206    fn large_tile_roundtrip() {
207        let s = schema();
208        let tile = make_tile(&s, 100);
209        let mut buf = Vec::new();
210        encode_sparse_tile(&tile, &mut buf).unwrap();
211        let decoded = decode_sparse_tile(&buf).unwrap();
212        assert_eq!(decoded.surrogates, tile.surrogates);
213        assert_eq!(decoded.attr_cols, tile.attr_cols);
214        assert_eq!(decoded.dim_dicts.len(), tile.dim_dicts.len());
215    }
216
217    #[test]
218    fn sentinel_only_tile_uses_raw() {
219        let s = schema();
220        let mut b = SparseTileBuilder::new(&s);
221        for i in 0..20 {
222            b.push_row(SparseRow {
223                coord: &[CoordValue::Int64(i), CoordValue::Int64(i)],
224                attrs: &[],
225                surrogate: None,
226                valid_from_ms: 0,
227                valid_until_ms: OPEN_UPPER,
228                kind: RowKind::Tombstone,
229            })
230            .unwrap();
231        }
232        let tile = b.build();
233        let mut buf = Vec::new();
234        encode_sparse_tile(&tile, &mut buf).unwrap();
235        assert_eq!(buf[0], CodecTag::Raw.as_byte());
236    }
237
238    #[test]
239    fn version_byte_is_one() {
240        let s = schema();
241        let tile = make_tile(&s, 20);
242        let mut buf = Vec::new();
243        encode_sparse_tile(&tile, &mut buf).unwrap();
244        assert_eq!(buf[1], PAYLOAD_VERSION);
245    }
246
247    #[test]
248    fn empty_tile_encodes_and_decodes() {
249        let s = schema();
250        let tile = SparseTile::empty(&s);
251        let mut buf = Vec::new();
252        encode_sparse_tile(&tile, &mut buf).unwrap();
253        let decoded = decode_sparse_tile(&buf).unwrap();
254        assert_eq!(decoded.surrogates, tile.surrogates);
255    }
256
257    /// Tombstone rows interleaved with live rows keep no identity through the
258    /// structural codec, and every live row keeps its own.
259    #[test]
260    fn structural_roundtrip_keeps_sentinel_rows_without_identity() {
261        let s = schema();
262        let mut b = SparseTileBuilder::new(&s);
263        for i in 0..20i64 {
264            let coord = [CoordValue::Int64(i), CoordValue::Int64(i)];
265            if i % 3 == 0 {
266                b.push_row(SparseRow::sentinel(&coord, RowKind::Tombstone))
267                    .unwrap();
268            } else {
269                b.push_row(SparseRow::live(
270                    &coord,
271                    &[CellValue::Int64(i)],
272                    Surrogate::new(100 + i as u32),
273                    0,
274                    OPEN_UPPER,
275                ))
276                .unwrap();
277            }
278        }
279        let tile = b.build();
280        let mut buf = Vec::new();
281        encode_sparse_tile(&tile, &mut buf).unwrap();
282        assert_eq!(buf[0], CodecTag::Structural.as_byte());
283        let decoded = decode_sparse_tile(&buf).unwrap();
284        assert_eq!(decoded.surrogates, tile.surrogates);
285        assert_eq!(decoded.surrogates[0], None);
286        assert_eq!(decoded.surrogates[1], Some(Surrogate::new(101)));
287    }
288
289    /// A derived result tile holds no identity, so it is never written.
290    #[test]
291    fn a_derived_tile_is_refused_by_the_encoder() {
292        let s = schema();
293        let mut b = SparseTileBuilder::new(&s);
294        b.push(
295            &[CoordValue::Int64(1), CoordValue::Int64(1)],
296            &[CellValue::Int64(1)],
297        )
298        .unwrap();
299        let mut buf = Vec::new();
300        assert!(encode_sparse_tile(&b.build(), &mut buf).is_err());
301        assert!(buf.is_empty(), "nothing is written for a refused tile");
302    }
303}
nodedb_array/codec/tile_encode.rs

nodedb_array/codec/
tile_encode.rs
