nodedb_array/tile/
sparse_tile.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Sparse tile payload — coordinate list + per-attribute columns.
4//!
5//! Storage layout is column-major so per-attribute codecs
6//! (`nodedb-codec`) can be applied without a transpose. Dimension values are
7//! dictionary-encoded inline: rare in genomic / single-cell workloads
8//! that have many repeats per tile.
9
10use nodedb_types::{OPEN_UPPER, Surrogate};
11use serde::{Deserialize, Serialize};
12
13use super::mbr::{MbrBuilder, TileMBR};
14use crate::error::{ArrayError, ArrayResult};
15use crate::schema::ArraySchema;
16use crate::types::cell_value::value::CellValue;
17use crate::types::coord::value::CoordValue;
18
19/// Per-row lifecycle classification for a [`SparseTile`] row.
20///
21/// Stored as a parallel `u8` column (`row_kinds`) so the segment can
22/// carry tombstone and GDPR-erasure markers that survive flush and are
23/// visible to post-flush reads. Compaction's retention policy is the
24/// only path allowed to drop these rows.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub enum RowKind {
27    Live = 0,
28    Tombstone = 1,
29    GdprErased = 2,
30}
31
32impl RowKind {
33    /// Convert a raw byte to a `RowKind`. Fails on unknown values so
34    /// corrupt data surfaces as a typed error rather than silent `Live`.
35    pub fn from_u8(b: u8) -> ArrayResult<Self> {
36        match b {
37            0 => Ok(Self::Live),
38            1 => Ok(Self::Tombstone),
39            2 => Ok(Self::GdprErased),
40            other => Err(ArrayError::SegmentCorruption {
41                detail: format!("unknown RowKind byte: {other:#04x}"),
42            }),
43        }
44    }
45
46    pub fn as_u8(self) -> u8 {
47        self as u8
48    }
49}
50
51impl From<RowKind> for u8 {
52    fn from(k: RowKind) -> u8 {
53        k.as_u8()
54    }
55}
56
57/// Per-dim dictionary: distinct values seen, and one index per cell
58/// pointing into the dictionary. Index width is selected by callers
59/// when picking the codec.
60#[derive(
61    Debug,
62    Clone,
63    PartialEq,
64    Serialize,
65    Deserialize,
66    zerompk::ToMessagePack,
67    zerompk::FromMessagePack,
68)]
69pub struct DimDict {
70    pub values: Vec<CoordValue>,
71    pub indices: Vec<u32>,
72}
73
74impl DimDict {
75    fn new() -> Self {
76        Self {
77            values: Vec::new(),
78            indices: Vec::new(),
79        }
80    }
81
82    fn push(&mut self, v: &CoordValue) {
83        if let Some(idx) = self.values.iter().position(|x| x == v) {
84            self.indices.push(idx as u32);
85        } else {
86            self.indices.push(self.values.len() as u32);
87            self.values.push(v.clone());
88        }
89    }
90
91    pub fn cardinality(&self) -> usize {
92        self.values.len()
93    }
94
95    pub fn len(&self) -> usize {
96        self.indices.len()
97    }
98
99    pub fn is_empty(&self) -> bool {
100        self.indices.is_empty()
101    }
102}
103
104/// Sparse tile payload.
105#[derive(
106    Debug,
107    Clone,
108    PartialEq,
109    Serialize,
110    Deserialize,
111    zerompk::ToMessagePack,
112    zerompk::FromMessagePack,
113)]
114pub struct SparseTile {
115    /// One dictionary per schema dim, parallel to `schema.dims`.
116    pub dim_dicts: Vec<DimDict>,
117    /// One column per schema attr, parallel to `schema.attrs`.
118    pub attr_cols: Vec<Vec<CellValue>>,
119    /// Per-row global surrogate, parallel to the dim-dict index streams.
120    /// Cross-engine bitmap joins read this column directly without
121    /// translating coords back to user-visible primary keys.
122    ///
123    /// A stored live cell holds `Some` bound surrogate. A tombstone or
124    /// erasure row holds `None`: it names a coordinate, not a row. A
125    /// derived query result row, such as an elementwise output, holds
126    /// `None` and is never stored. `Surrogate::ZERO` never appears.
127    pub surrogates: Vec<Option<Surrogate>>,
128    /// Per-cell valid-time lower bound in milliseconds (inclusive).
129    /// Parallel to `surrogates`.
130    pub valid_from_ms: Vec<i64>,
131    /// Per-cell valid-time upper bound in milliseconds (exclusive).
132    /// [`nodedb_types::OPEN_UPPER`] (`i64::MAX`) means open-ended.
133    /// Parallel to `surrogates`.
134    pub valid_until_ms: Vec<i64>,
135    /// Per-row [`RowKind`] encoded as `u8`. Parallel to `surrogates`.
136    /// `0` = Live, `1` = Tombstone, `2` = GdprErased.
137    pub row_kinds: Vec<u8>,
138    pub mbr: TileMBR,
139}
140
141impl SparseTile {
142    /// Empty tile sized for the given schema.
143    pub fn empty(schema: &ArraySchema) -> Self {
144        Self {
145            dim_dicts: (0..schema.arity()).map(|_| DimDict::new()).collect(),
146            attr_cols: (0..schema.attrs.len()).map(|_| Vec::new()).collect(),
147            surrogates: Vec::new(),
148            valid_from_ms: Vec::new(),
149            valid_until_ms: Vec::new(),
150            row_kinds: Vec::new(),
151            mbr: TileMBR::new(schema.arity(), schema.attrs.len()),
152        }
153    }
154
155    /// Count of live (non-sentinel) rows. Used for MBR statistics and
156    /// predicate-pushdown decisions. Sentinel rows are excluded.
157    ///
158    /// **Do not use this value as a row-iteration bound.** Iteration must
159    /// cover sentinel rows so callers can observe Tombstone / GdprErased
160    /// markers when applying bitemporal visibility rules — use
161    /// [`row_count`](Self::row_count) and filter with
162    /// [`row_kind`](Self::row_kind) instead.
163    pub fn nnz(&self) -> u32 {
164        self.mbr.nnz
165    }
166
167    /// Total physical row count including sentinel (Tombstone / GdprErased)
168    /// rows. Use this as the upper bound when iterating over rows; check
169    /// [`row_kind`](Self::row_kind) inside the loop and skip non-Live rows
170    /// when reading attribute columns. The attribute column vectors are
171    /// indexed by the **live-row index** (incremented only for Live rows),
172    /// not by the iteration index — confusing the two yields incorrect or
173    /// out-of-bounds reads.
174    pub fn row_count(&self) -> usize {
175        self.surrogates.len()
176    }
177
178    /// Decode the `RowKind` for the given row index.
179    ///
180    /// Returns [`RowKind::Live`] when `row_kinds` is empty (forward-compat
181    /// with pre-v4 data still in memory during migration) or when the row
182    /// index is out of range.
183    pub fn row_kind(&self, row: usize) -> ArrayResult<RowKind> {
184        match self.row_kinds.get(row) {
185            None => Ok(RowKind::Live),
186            Some(&b) => RowKind::from_u8(b),
187        }
188    }
189
190    /// The surrogate `row` holds: `Some` for a stored live cell, `None` for a
191    /// tombstone, an erasure, or a derived result row. A row index past the
192    /// surrogate column is a malformed tile.
193    pub fn row_surrogate(&self, row: usize) -> ArrayResult<Option<Surrogate>> {
194        self.surrogates
195            .get(row)
196            .copied()
197            .ok_or_else(|| ArrayError::SegmentCorruption {
198                detail: format!(
199                    "tile row {row} is past the surrogate column ({} rows)",
200                    self.surrogates.len()
201                ),
202            })
203    }
204
205    /// The surrogate of live `row`. A live row read from storage always holds
206    /// one, so a row without it is a malformed tile.
207    pub fn live_surrogate(&self, row: usize) -> ArrayResult<Surrogate> {
208        self.row_surrogate(row)?
209            .ok_or_else(|| ArrayError::SegmentCorruption {
210                detail: format!("live tile row {row} carries no surrogate"),
211            })
212    }
213
214    /// Check the identity invariant of a stored tile: every live row holds a
215    /// bound surrogate and every tombstone or erasure row holds none. The
216    /// segment codec runs it on every tile it encodes or decodes.
217    pub fn check_stored_identities(&self) -> ArrayResult<()> {
218        if self.row_kinds.len() != self.surrogates.len() {
219            return Err(ArrayError::SegmentCorruption {
220                detail: format!(
221                    "stored tile carries {} row kinds but {} surrogates",
222                    self.row_kinds.len(),
223                    self.surrogates.len()
224                ),
225            });
226        }
227        for (row, surrogate) in self.surrogates.iter().enumerate() {
228            let kind = self.row_kind(row)?;
229            match (kind, surrogate) {
230                (RowKind::Live, Some(s)) if *s != Surrogate::ZERO => {}
231                (RowKind::Tombstone | RowKind::GdprErased, None) => {}
232                (kind, surrogate) => {
233                    return Err(ArrayError::SegmentCorruption {
234                        detail: format!(
235                            "stored tile row {row} of kind {kind:?} holds surrogate \
236                             {surrogate:?}; a live row holds a bound surrogate and a \
237                             tombstone or erasure row holds none"
238                        ),
239                    });
240                }
241            }
242        }
243        Ok(())
244    }
245}
246
247/// All row-level data passed to [`SparseTileBuilder::push_row`].
248pub struct SparseRow<'a> {
249    pub coord: &'a [CoordValue],
250    pub attrs: &'a [CellValue],
251    /// `Some` bound surrogate for a stored live cell. `None` for a tombstone,
252    /// an erasure, or a derived result row that is never stored.
253    pub surrogate: Option<Surrogate>,
254    pub valid_from_ms: i64,
255    pub valid_until_ms: i64,
256    pub kind: RowKind,
257}
258
259impl<'a> SparseRow<'a> {
260    /// Construct a live row that holds its bound surrogate.
261    pub fn live(
262        coord: &'a [CoordValue],
263        attrs: &'a [CellValue],
264        surrogate: Surrogate,
265        valid_from_ms: i64,
266        valid_until_ms: i64,
267    ) -> Self {
268        Self {
269            coord,
270            attrs,
271            surrogate: Some(surrogate),
272            valid_from_ms,
273            valid_until_ms,
274            kind: RowKind::Live,
275        }
276    }
277
278    /// Construct a tombstone or erasure row. It carries no attributes and no
279    /// identity.
280    pub fn sentinel(coord: &'a [CoordValue], kind: RowKind) -> Self {
281        Self {
282            coord,
283            attrs: &[],
284            surrogate: None,
285            valid_from_ms: 0,
286            valid_until_ms: OPEN_UPPER,
287            kind,
288        }
289    }
290}
291
292/// Streaming builder. Folds `(coord, attrs)` pairs into the tile body
293/// and the MBR in one pass.
294pub struct SparseTileBuilder<'a> {
295    schema: &'a ArraySchema,
296    dim_dicts: Vec<DimDict>,
297    attr_cols: Vec<Vec<CellValue>>,
298    surrogates: Vec<Option<Surrogate>>,
299    valid_from_ms: Vec<i64>,
300    valid_until_ms: Vec<i64>,
301    row_kinds: Vec<u8>,
302    mbr: MbrBuilder,
303}
304
305impl<'a> SparseTileBuilder<'a> {
306    pub fn new(schema: &'a ArraySchema) -> Self {
307        Self {
308            schema,
309            dim_dicts: (0..schema.arity()).map(|_| DimDict::new()).collect(),
310            attr_cols: (0..schema.attrs.len()).map(|_| Vec::new()).collect(),
311            surrogates: Vec::new(),
312            valid_from_ms: Vec::new(),
313            valid_until_ms: Vec::new(),
314            row_kinds: Vec::new(),
315            mbr: MbrBuilder::new(schema.arity(), schema.attrs.len()),
316        }
317    }
318
319    /// Push a row with full bitemporal metadata.
320    ///
321    /// Sentinel rows (`Tombstone`, `GdprErased`) may have an empty `attrs`
322    /// slice — the attribute columns receive no entry for these rows, which
323    /// is correct because sentinel rows carry no payload.
324    pub fn push_row(&mut self, row: SparseRow<'_>) -> ArrayResult<()> {
325        let SparseRow {
326            coord,
327            attrs,
328            surrogate,
329            valid_from_ms,
330            valid_until_ms,
331            kind,
332        } = row;
333        match (kind, surrogate) {
334            (_, Some(s)) if s == Surrogate::ZERO => {
335                return Err(ArrayError::InvalidOp {
336                    detail: format!(
337                        "row pushed into array '{}' carries Surrogate::ZERO, which names no row",
338                        self.schema.name
339                    ),
340                });
341            }
342            (RowKind::Tombstone | RowKind::GdprErased, Some(s)) => {
343                return Err(ArrayError::InvalidOp {
344                    detail: format!(
345                        "{kind:?} row pushed into array '{}' carries surrogate {s}; a \
346                         tombstone or erasure row carries no identity",
347                        self.schema.name
348                    ),
349                });
350            }
351            _ => {}
352        }
353        if coord.len() != self.schema.arity() {
354            return Err(ArrayError::CoordArityMismatch {
355                array: self.schema.name.clone(),
356                expected: self.schema.arity(),
357                got: coord.len(),
358            });
359        }
360        // Sentinel rows carry no payload — allow empty attrs. Live rows must
361        // match the schema attr count exactly.
362        if kind == RowKind::Live && attrs.len() != self.schema.attrs.len() {
363            return Err(ArrayError::CellTypeMismatch {
364                array: self.schema.name.clone(),
365                attr: "<all>".to_string(),
366                detail: format!(
367                    "expected {} attr columns, got {}",
368                    self.schema.attrs.len(),
369                    attrs.len()
370                ),
371            });
372        }
373        for (i, c) in coord.iter().enumerate() {
374            self.dim_dicts[i].push(c);
375        }
376        // For sentinel rows `attrs` is empty — no entries added to attr_cols.
377        // For live rows, every attr col gets one entry.
378        for (i, a) in attrs.iter().enumerate() {
379            self.attr_cols[i].push(a.clone());
380        }
381        self.surrogates.push(surrogate);
382        self.valid_from_ms.push(valid_from_ms);
383        self.valid_until_ms.push(valid_until_ms);
384        self.row_kinds.push(kind.as_u8());
385        // Only fold MBR for live rows; sentinel rows have no meaningful attrs
386        // and may have coords outside the declared domain.
387        if kind == RowKind::Live {
388            self.mbr.fold(coord, attrs);
389        }
390        Ok(())
391    }
392
393    /// Push a derived live row, such as an elementwise result. It holds no
394    /// surrogate, so the tile it lands in is a query result and never stored.
395    /// A stored cell goes through [`Self::push_row`] with its bound surrogate.
396    pub fn push(&mut self, coord: &[CoordValue], attrs: &[CellValue]) -> ArrayResult<()> {
397        self.push_row(SparseRow {
398            coord,
399            attrs,
400            surrogate: None,
401            valid_from_ms: 0,
402            valid_until_ms: OPEN_UPPER,
403            kind: RowKind::Live,
404        })
405    }
406
407    pub fn build(self) -> SparseTile {
408        SparseTile {
409            dim_dicts: self.dim_dicts,
410            attr_cols: self.attr_cols,
411            surrogates: self.surrogates,
412            valid_from_ms: self.valid_from_ms,
413            valid_until_ms: self.valid_until_ms,
414            row_kinds: self.row_kinds,
415            mbr: self.mbr.build(),
416        }
417    }
418}
419
420#[cfg(test)]
421mod tests {
422    use super::*;
423    use crate::schema::ArraySchemaBuilder;
424    use crate::schema::attr_spec::{AttrSpec, AttrType};
425    use crate::schema::dim_spec::{DimSpec, DimType};
426    use crate::types::domain::{Domain, DomainBound};
427
428    fn schema() -> ArraySchema {
429        ArraySchemaBuilder::new("g")
430            .dim(DimSpec::new(
431                "chrom",
432                DimType::Int64,
433                Domain::new(DomainBound::Int64(0), DomainBound::Int64(24)),
434            ))
435            .dim(DimSpec::new(
436                "pos",
437                DimType::Int64,
438                Domain::new(DomainBound::Int64(0), DomainBound::Int64(1_000_000)),
439            ))
440            .attr(AttrSpec::new("variant", AttrType::String, false))
441            .attr(AttrSpec::new("qual", AttrType::Float64, true))
442            .tile_extents(vec![1, 1_000_000])
443            .build()
444            .unwrap()
445    }
446
447    #[test]
448    fn sparse_tile_collects_columns_in_order() {
449        let s = schema();
450        let mut b = SparseTileBuilder::new(&s);
451        b.push(
452            &[CoordValue::Int64(1), CoordValue::Int64(100)],
453            &[CellValue::String("ALT".into()), CellValue::Float64(40.0)],
454        )
455        .unwrap();
456        b.push(
457            &[CoordValue::Int64(1), CoordValue::Int64(200)],
458            &[CellValue::String("REF".into()), CellValue::Float64(50.0)],
459        )
460        .unwrap();
461        let t = b.build();
462        assert_eq!(t.nnz(), 2);
463        assert_eq!(t.attr_cols.len(), 2);
464        assert_eq!(t.attr_cols[0].len(), 2);
465        assert_eq!(t.attr_cols[1].len(), 2);
466    }
467
468    #[test]
469    fn dim_dict_dedups_repeated_values() {
470        let s = schema();
471        let mut b = SparseTileBuilder::new(&s);
472        b.push(
473            &[CoordValue::Int64(1), CoordValue::Int64(100)],
474            &[CellValue::String("ALT".into()), CellValue::Float64(40.0)],
475        )
476        .unwrap();
477        b.push(
478            &[CoordValue::Int64(1), CoordValue::Int64(200)],
479            &[CellValue::String("REF".into()), CellValue::Float64(50.0)],
480        )
481        .unwrap();
482        let t = b.build();
483        // chrom dictionary should have one entry (1 repeated).
484        assert_eq!(t.dim_dicts[0].cardinality(), 1);
485        assert_eq!(t.dim_dicts[0].len(), 2);
486    }
487
488    #[test]
489    fn sparse_tile_rejects_bad_arity() {
490        let s = schema();
491        let mut b = SparseTileBuilder::new(&s);
492        let r = b.push(
493            &[CoordValue::Int64(1)],
494            &[CellValue::String("ALT".into()), CellValue::Float64(0.0)],
495        );
496        assert!(r.is_err());
497    }
498
499    #[test]
500    fn sparse_tile_rejects_bad_attr_count() {
501        let s = schema();
502        let mut b = SparseTileBuilder::new(&s);
503        let r = b.push(
504            &[CoordValue::Int64(1), CoordValue::Int64(0)],
505            &[CellValue::String("ALT".into())],
506        );
507        assert!(r.is_err());
508    }
509
510    #[test]
511    fn empty_sparse_tile_has_zero_nnz() {
512        let s = schema();
513        let t = SparseTile::empty(&s);
514        assert_eq!(t.nnz(), 0);
515        assert_eq!(t.dim_dicts.len(), 2);
516        assert_eq!(t.attr_cols.len(), 2);
517    }
518
519    #[test]
520    fn sparse_tile_msgpack_roundtrip_carries_valid_time_bounds() {
521        let s = schema();
522        let mut b = SparseTileBuilder::new(&s);
523        b.push_row(SparseRow {
524            coord: &[CoordValue::Int64(1), CoordValue::Int64(100)],
525            attrs: &[CellValue::String("A".into()), CellValue::Float64(1.0)],
526            surrogate: Some(Surrogate::new(1)),
527            valid_from_ms: 50,
528            valid_until_ms: 150,
529            kind: RowKind::Live,
530        })
531        .unwrap();
532        b.push_row(SparseRow {
533            coord: &[CoordValue::Int64(2), CoordValue::Int64(200)],
534            attrs: &[CellValue::String("B".into()), CellValue::Float64(2.0)],
535            surrogate: Some(Surrogate::new(2)),
536            valid_from_ms: 300,
537            valid_until_ms: 900,
538            kind: RowKind::Live,
539        })
540        .unwrap();
541        let tile = b.build();
542
543        let bytes = zerompk::to_msgpack_vec(&tile).unwrap();
544        let decoded: SparseTile = zerompk::from_msgpack(&bytes).unwrap();
545
546        assert_eq!(decoded.valid_from_ms, vec![50, 300]);
547        assert_eq!(decoded.valid_until_ms, vec![150, 900]);
548        assert_eq!(decoded.nnz(), 2);
549    }
550
551    #[test]
552    fn sparse_tile_msgpack_roundtrip_carries_row_kinds() {
553        let s = schema();
554        let mut b = SparseTileBuilder::new(&s);
555        // Live row
556        b.push_row(SparseRow {
557            coord: &[CoordValue::Int64(1), CoordValue::Int64(10)],
558            attrs: &[CellValue::String("X".into()), CellValue::Float64(1.0)],
559            surrogate: Some(Surrogate::new(1)),
560            valid_from_ms: 0,
561            valid_until_ms: OPEN_UPPER,
562            kind: RowKind::Live,
563        })
564        .unwrap();
565        // Tombstone row (no attrs)
566        b.push_row(SparseRow {
567            coord: &[CoordValue::Int64(2), CoordValue::Int64(20)],
568            attrs: &[],
569            surrogate: None,
570            valid_from_ms: 0,
571            valid_until_ms: OPEN_UPPER,
572            kind: RowKind::Tombstone,
573        })
574        .unwrap();
575        // GdprErased row (no attrs)
576        b.push_row(SparseRow {
577            coord: &[CoordValue::Int64(3), CoordValue::Int64(30)],
578            attrs: &[],
579            surrogate: None,
580            valid_from_ms: 0,
581            valid_until_ms: OPEN_UPPER,
582            kind: RowKind::GdprErased,
583        })
584        .unwrap();
585        let tile = b.build();
586
587        let bytes = zerompk::to_msgpack_vec(&tile).unwrap();
588        let decoded: SparseTile = zerompk::from_msgpack(&bytes).unwrap();
589
590        assert_eq!(decoded.row_kinds.len(), 3);
591        assert_eq!(
592            RowKind::from_u8(decoded.row_kinds[0]).unwrap(),
593            RowKind::Live
594        );
595        assert_eq!(
596            RowKind::from_u8(decoded.row_kinds[1]).unwrap(),
597            RowKind::Tombstone
598        );
599        assert_eq!(
600            RowKind::from_u8(decoded.row_kinds[2]).unwrap(),
601            RowKind::GdprErased
602        );
603        assert_eq!(
604            decoded.surrogates,
605            vec![Some(Surrogate::new(1)), None, None]
606        );
607        decoded.check_stored_identities().unwrap();
608    }
609
610    /// `Surrogate::ZERO` names no row, so the builder refuses it on any row.
611    #[test]
612    fn builder_refuses_a_zero_surrogate() {
613        let s = schema();
614        let mut b = SparseTileBuilder::new(&s);
615        let r = b.push_row(SparseRow::live(
616            &[CoordValue::Int64(1), CoordValue::Int64(10)],
617            &[CellValue::String("X".into()), CellValue::Float64(1.0)],
618            Surrogate::ZERO,
619            0,
620            OPEN_UPPER,
621        ));
622        assert!(r.is_err(), "a live row under ZERO is refused");
623        assert_eq!(b.build().row_count(), 0, "the refused row is not stored");
624    }
625
626    /// A tombstone or erasure row names a coordinate, never a row identity.
627    #[test]
628    fn builder_refuses_a_sentinel_row_with_an_identity() {
629        let s = schema();
630        let mut b = SparseTileBuilder::new(&s);
631        let coord = [CoordValue::Int64(1), CoordValue::Int64(10)];
632        let mut row = SparseRow::sentinel(&coord, RowKind::Tombstone);
633        row.surrogate = Some(Surrogate::new(4));
634        assert!(b.push_row(row).is_err());
635    }
636
637    /// A derived row holds no surrogate, so its tile fails the stored-tile
638    /// check: it can be read but never written to a segment.
639    #[test]
640    fn a_derived_row_fails_the_stored_identity_check() {
641        let s = schema();
642        let mut b = SparseTileBuilder::new(&s);
643        b.push(
644            &[CoordValue::Int64(1), CoordValue::Int64(10)],
645            &[CellValue::String("X".into()), CellValue::Float64(1.0)],
646        )
647        .unwrap();
648        let tile = b.build();
649        assert_eq!(tile.row_surrogate(0).unwrap(), None);
650        assert!(tile.live_surrogate(0).is_err());
651        assert!(tile.check_stored_identities().is_err());
652    }
653}
nodedb_array/tile/sparse_tile.rs

nodedb_array/tile/
sparse_tile.rs
