nodedb-array 0.6.1

// SPDX-License-Identifier: Apache-2.0

// Structural sparse-tile decoder — symmetric to tile_encode.
//
// Validates the version byte, dispatches on CodecTag, and reconstructs
// a SparseTile from the structural payload. Legacy (v3) msgpack tiles are
// NOT handled here; the segment reader peeks the tag and dispatches to
// zerompk for those.

use crate::codec::column_codec::{
    decode_attr_col, decode_row_kinds, decode_surrogates, decode_timestamps_col,
};
use crate::codec::coord_rle::decode_coord_axis_rle;
use crate::codec::limits::{
    MAX_ATTRS_PER_TILE, MAX_AXES_PER_TILE, MAX_CELLS_PER_TILE, check_decoded_size, checked_capacity,
};
use crate::codec::tag::{CodecTag, peek_tag};
use crate::error::{ArrayError, ArrayResult};
use crate::tile::mbr::TileMBR;
use crate::tile::sparse_tile::{RowKind, SparseTile};

const SUPPORTED_PAYLOAD_VERSION: u8 = 1;

fn read_framed<'a>(data: &'a [u8], pos: &mut usize) -> ArrayResult<&'a [u8]> {
    if *pos + 4 > data.len() {
        return Err(ArrayError::SegmentCorruption {
            detail: "framed block: truncated length".into(),
        });
    }
    let len = u32::from_le_bytes(
        data[*pos..*pos + 4]
            .try_into()
            .expect("invariant: bounds-checked above (*pos + 4 <= data.len())"),
    ) as usize;
    *pos += 4;
    if *pos + len > data.len() {
        return Err(ArrayError::SegmentCorruption {
            detail: format!(
                "framed block: body truncated (need {len}, have {})",
                data.len() - *pos
            ),
        });
    }
    let slice = &data[*pos..*pos + len];
    *pos += len;
    Ok(slice)
}

/// Decode a tile payload previously written by `encode_sparse_tile`.
///
/// The `payload` slice must start at the tag byte (i.e. after BlockFraming
/// has been unwrapped by the segment reader).
pub fn decode_sparse_tile(payload: &[u8]) -> ArrayResult<SparseTile> {
    if payload.len() < 2 {
        return Err(ArrayError::SegmentCorruption {
            detail: "sparse tile payload too short".into(),
        });
    }

    let tag_result = peek_tag(payload).ok_or_else(|| {
        // peek_tag returns None for both legacy msgpack and unknown bytes.
        // The reader should never call us with a legacy payload.
        ArrayError::SegmentCorruption {
            detail: format!(
                "decode_sparse_tile called with legacy or unknown tag byte: {:#04x}",
                payload[0]
            ),
        }
    })?;

    let version = payload[1];
    if version != SUPPORTED_PAYLOAD_VERSION {
        return Err(ArrayError::SegmentCorruption {
            detail: format!("unsupported tile payload version: {version}"),
        });
    }

    match tag_result {
        CodecTag::Raw => decode_raw(&payload[2..]),
        CodecTag::Structural => decode_structural(&payload[2..]),
    }
}

fn decode_raw(body: &[u8]) -> ArrayResult<SparseTile> {
    let tile: SparseTile =
        zerompk::from_msgpack(body).map_err(|e| ArrayError::SegmentCorruption {
            detail: format!("raw tile decode: {e}"),
        })?;
    tile.check_stored_identities()?;
    Ok(tile)
}

fn decode_structural(body: &[u8]) -> ArrayResult<SparseTile> {
    let mut pos = 0;

    if pos + 8 > body.len() {
        return Err(ArrayError::SegmentCorruption {
            detail: "structural tile: truncated counts".into(),
        });
    }
    let cell_count = u32::from_le_bytes(
        body[pos..pos + 4]
            .try_into()
            .expect("invariant: bounds-checked above (pos + 8 <= body.len())"),
    ) as usize;
    pos += 4;
    check_decoded_size(cell_count, MAX_CELLS_PER_TILE, "cell_count")?;
    let axis_count = u32::from_le_bytes(
        body[pos..pos + 4]
            .try_into()
            .expect("invariant: bounds-checked above (pos + 4 <= body.len())"),
    ) as usize;
    pos += 4;
    let axis_capacity = checked_capacity(
        axis_count,
        MAX_AXES_PER_TILE,
        body.len() - pos,
        4,
        "axis_count",
    )?;

    // Coordinate axes.
    let mut dim_dicts = Vec::with_capacity(axis_capacity);
    for _ in 0..axis_count {
        let axis_bytes = read_framed(body, &mut pos)?;
        let mut inner_pos = 0;
        let dict = decode_coord_axis_rle(axis_bytes, &mut inner_pos)?;
        dim_dicts.push(dict);
    }

    // Surrogates: live rows only, in row order.
    let surr_bytes = read_framed(body, &mut pos)?;
    let live_surrogates = decode_surrogates(surr_bytes)?;

    // Row kinds.
    let rk_bytes = read_framed(body, &mut pos)?;
    let row_kinds = decode_row_kinds(rk_bytes)?;

    // system_from_ms placeholder (not stored in SparseTile, just skip).
    let _sys_bytes = read_framed(body, &mut pos)?;

    // valid_from_ms.
    let vf_bytes = read_framed(body, &mut pos)?;
    let valid_from_ms = decode_timestamps_col(vf_bytes)?;

    // valid_until_ms.
    let vu_bytes = read_framed(body, &mut pos)?;
    let valid_until_ms = decode_timestamps_col(vu_bytes)?;

    // Attr columns.
    if pos + 4 > body.len() {
        return Err(ArrayError::SegmentCorruption {
            detail: "structural tile: truncated attr count".into(),
        });
    }
    let attr_count = u32::from_le_bytes(
        body[pos..pos + 4]
            .try_into()
            .expect("invariant: bounds-checked above (pos + 4 <= body.len())"),
    ) as usize;
    pos += 4;
    let attr_capacity = checked_capacity(
        attr_count,
        MAX_ATTRS_PER_TILE,
        body.len() - pos,
        4,
        "attr_count",
    )?;

    let mut attr_cols = Vec::with_capacity(attr_capacity);
    for _ in 0..attr_count {
        let col_bytes = read_framed(body, &mut pos)?;
        let col = decode_attr_col(col_bytes)?;
        attr_cols.push(col);
    }

    // Reconstruct MBR from decoded data — we don't persist the MBR inside the
    // structural payload (it lives in the footer TileEntry instead).
    // Use a zero-dimension MBR so the tile can be used for reads; compaction
    // re-derives the full MBR if needed.
    let mbr = TileMBR::new(axis_count, attr_count);

    // Validate sizes match cell_count.
    if row_kinds.len() != cell_count {
        return Err(ArrayError::SegmentCorruption {
            detail: format!(
                "structural tile: row kind count {kinds} != cell_count {cell_count}",
                kinds = row_kinds.len()
            ),
        });
    }
    let surrogates = spread_live_surrogates(&row_kinds, live_surrogates)?;

    let tile = SparseTile {
        dim_dicts,
        attr_cols,
        surrogates,
        valid_from_ms,
        valid_until_ms,
        row_kinds,
        mbr,
    };
    tile.check_stored_identities()?;
    Ok(tile)
}

/// Place the live-row surrogate column back onto every row: a live row takes
/// the next surrogate, a tombstone or erasure row holds none. The column must
/// hold exactly one surrogate per live row.
fn spread_live_surrogates(
    row_kinds: &[u8],
    live_surrogates: Vec<nodedb_types::Surrogate>,
) -> ArrayResult<Vec<Option<nodedb_types::Surrogate>>> {
    let live_count = live_surrogates.len();
    let mut live = live_surrogates.into_iter();
    let mut surrogates = Vec::with_capacity(row_kinds.len());
    for (row, &kind) in row_kinds.iter().enumerate() {
        if RowKind::from_u8(kind)? == RowKind::Live {
            let surrogate = live.next().ok_or_else(|| ArrayError::SegmentCorruption {
                detail: format!(
                    "structural tile: live row {row} has no surrogate ({live_count} stored)"
                ),
            })?;
            surrogates.push(Some(surrogate));
        } else {
            surrogates.push(None);
        }
    }
    if live.next().is_some() {
        return Err(ArrayError::SegmentCorruption {
            detail: format!("structural tile: {live_count} surrogates stored for fewer live rows"),
        });
    }
    Ok(surrogates)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codec::tile_encode::encode_sparse_tile;
    use crate::schema::ArraySchemaBuilder;
    use crate::schema::attr_spec::{AttrSpec, AttrType};
    use crate::schema::dim_spec::{DimSpec, DimType};
    use crate::tile::sparse_tile::{RowKind, SparseRow, SparseTileBuilder};
    use crate::types::cell_value::value::CellValue;
    use crate::types::coord::value::CoordValue;
    use crate::types::domain::{Domain, DomainBound};
    use nodedb_types::{OPEN_UPPER, Surrogate};

    fn schema() -> crate::schema::ArraySchema {
        ArraySchemaBuilder::new("t")
            .dim(DimSpec::new(
                "x",
                DimType::Int64,
                Domain::new(DomainBound::Int64(0), DomainBound::Int64(1_000_000)),
            ))
            .attr(AttrSpec::new("v", AttrType::Int64, true))
            .tile_extents(vec![1000])
            .build()
            .unwrap()
    }

    fn make_tile(s: &crate::schema::ArraySchema, n: usize) -> SparseTile {
        let mut b = SparseTileBuilder::new(s);
        for i in 0..n {
            b.push_row(SparseRow {
                coord: &[CoordValue::Int64(i as i64)],
                attrs: &[CellValue::Int64(i as i64 * 2)],
                surrogate: Some(Surrogate::new(i as u32 + 1)),
                valid_from_ms: i as i64,
                valid_until_ms: OPEN_UPPER,
                kind: RowKind::Live,
            })
            .unwrap();
        }
        b.build()
    }

    fn roundtrip(tile: &SparseTile) -> SparseTile {
        let mut buf = Vec::new();
        encode_sparse_tile(tile, &mut buf).unwrap();
        decode_sparse_tile(&buf).unwrap()
    }

    #[test]
    fn empty_tile_roundtrip() {
        let s = schema();
        let tile = SparseTile::empty(&s);
        let out = roundtrip(&tile);
        assert_eq!(out.surrogates, tile.surrogates);
        assert_eq!(out.row_kinds, tile.row_kinds);
    }

    #[test]
    fn small_tile_roundtrip() {
        let s = schema();
        let tile = make_tile(&s, 4);
        let out = roundtrip(&tile);
        assert_eq!(out.valid_from_ms, tile.valid_from_ms);
        assert_eq!(out.attr_cols, tile.attr_cols);
    }

    #[test]
    fn structural_tile_roundtrip() {
        let s = schema();
        let tile = make_tile(&s, 50);
        let out = roundtrip(&tile);
        assert_eq!(out.surrogates.len(), tile.surrogates.len());
        assert_eq!(out.valid_from_ms, tile.valid_from_ms);
        assert_eq!(out.valid_until_ms, tile.valid_until_ms);
        assert_eq!(out.attr_cols, tile.attr_cols);
        assert_eq!(out.row_kinds, tile.row_kinds);
    }

    #[test]
    fn one_thousand_cells_roundtrip() {
        let s = schema();
        let tile = make_tile(&s, 1000);
        let out = roundtrip(&tile);
        assert_eq!(out.surrogates.len(), 1000);
        assert_eq!(out.dim_dicts[0].indices, tile.dim_dicts[0].indices);
    }

    #[test]
    fn tombstone_rows_roundtrip() {
        let s = schema();
        let mut b = SparseTileBuilder::new(&s);
        for i in 0..20 {
            b.push_row(SparseRow {
                coord: &[CoordValue::Int64(i)],
                attrs: &[CellValue::Int64(i)],
                surrogate: Some(Surrogate::new(i as u32 + 1)),
                valid_from_ms: 0,
                valid_until_ms: OPEN_UPPER,
                kind: RowKind::Live,
            })
            .unwrap();
        }
        b.push_row(SparseRow {
            coord: &[CoordValue::Int64(99)],
            attrs: &[],
            surrogate: None,
            valid_from_ms: 0,
            valid_until_ms: OPEN_UPPER,
            kind: RowKind::Tombstone,
        })
        .unwrap();
        let tile = b.build();
        let out = roundtrip(&tile);
        assert_eq!(out.row_kinds, tile.row_kinds);
        assert_eq!(out.surrogates, tile.surrogates);
        assert_eq!(out.surrogates[20], None);
    }

    /// A structural payload whose live-row surrogate column is one short is
    /// refused: a live row never decodes without its identity.
    #[test]
    fn a_live_row_without_a_stored_surrogate_is_refused() {
        let kinds = [
            RowKind::Live.as_u8(),
            RowKind::Tombstone.as_u8(),
            RowKind::Live.as_u8(),
        ];
        assert!(spread_live_surrogates(&kinds, vec![Surrogate::new(1)]).is_err());
        assert!(
            spread_live_surrogates(
                &kinds,
                vec![Surrogate::new(1), Surrogate::new(2), Surrogate::new(3)]
            )
            .is_err(),
            "a surrogate left over after the last live row is refused"
        );
        assert_eq!(
            spread_live_surrogates(&kinds, vec![Surrogate::new(1), Surrogate::new(2)]).unwrap(),
            vec![Some(Surrogate::new(1)), None, Some(Surrogate::new(2))]
        );
    }

    #[test]
    fn huge_axis_count_in_tiny_structural_payload_is_rejected_before_allocation() {
        let mut body = Vec::new();
        body.extend_from_slice(&0u32.to_le_bytes());
        body.extend_from_slice(&(MAX_AXES_PER_TILE as u32).to_le_bytes());
        let error = decode_structural(&body).unwrap_err();
        assert!(matches!(error, ArrayError::SegmentCorruption { .. }));
    }

    #[test]
    fn invalid_version_returns_error() {
        let s = schema();
        let tile = make_tile(&s, 20);
        let mut buf = Vec::new();
        encode_sparse_tile(&tile, &mut buf).unwrap();
        // Corrupt the version byte.
        buf[1] = 99;
        let err = decode_sparse_tile(&buf).unwrap_err();
        assert!(matches!(err, ArrayError::SegmentCorruption { .. }));
    }

    #[test]
    fn valid_time_variants_roundtrip() {
        let s = schema();
        let mut b = SparseTileBuilder::new(&s);
        b.push_row(SparseRow {
            coord: &[CoordValue::Int64(1)],
            attrs: &[CellValue::Int64(10)],
            surrogate: Some(Surrogate::new(1)),
            valid_from_ms: 100,
            valid_until_ms: 500,
            kind: RowKind::Live,
        })
        .unwrap();
        b.push_row(SparseRow {
            coord: &[CoordValue::Int64(2)],
            attrs: &[CellValue::Int64(20)],
            surrogate: Some(Surrogate::new(2)),
            valid_from_ms: 200,
            valid_until_ms: OPEN_UPPER,
            kind: RowKind::Live,
        })
        .unwrap();
        // need >=8 to get Structural
        for i in 3..20 {
            b.push_row(SparseRow {
                coord: &[CoordValue::Int64(i)],
                attrs: &[CellValue::Int64(i)],
                surrogate: Some(Surrogate::new(i as u32)),
                valid_from_ms: i * 10,
                valid_until_ms: OPEN_UPPER,
                kind: RowKind::Live,
            })
            .unwrap();
        }
        let tile = b.build();
        let out = roundtrip(&tile);
        assert_eq!(out.valid_from_ms[0], 100);
        assert_eq!(out.valid_until_ms[0], 500);
        assert_eq!(out.valid_until_ms[1], OPEN_UPPER);
    }
}