nodedb_columnar/compaction/
materialize.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Materialize live (non-deleted) rows from a flushed segment blob.
4//!
5//! Used by the durable RESTORE re-issue path: a backed-up flushed segment is
6//! decoded back into per-row `Value::Object`s so the rows can be replayed as a
7//! normal columnar `Insert` (WAL- or Raft-durable) rather than installed into
8//! in-memory-only state. Deleted rows are excluded — resurrecting them would be
9//! a correctness regression.
10
11use nodedb_types::columnar::ColumnarSchema;
12use nodedb_types::surrogate::Surrogate;
13use nodedb_types::value::Value;
14
15use crate::delete_bitmap::DeleteBitmap;
16use crate::error::ColumnarError;
17use crate::reader::OwnedSegmentReader;
18
19use super::extract::extract_row_value;
20
21/// Decode the live rows of one flushed segment blob into `Value::Object`s,
22/// paired with their per-row cross-engine surrogate.
23///
24/// - `blob` is a raw `NDBS` (plaintext) or `SEGC` (encrypted) segment.
25///   Encryption is detected from the blob magic: a `SEGC` blob is decrypted
26///   with `kek` (an error if `kek` is `None`); an `NDBS` blob is read directly.
27/// - `deletes` is the segment's delete bitmap (empty when no rows were
28///   deleted). Rows marked deleted are skipped.
29/// - `row_surrogates` is the per-row surrogate sidecar for this segment
30///   (parallel to segment row order). It may be empty (pre-surrogate snapshot)
31///   or shorter than the row count, in which case the missing entries read as
32///   `None`.
33///
34/// Returns `(Value::Object, Option<Surrogate>)` for every live row, in segment
35/// row order. The object keys are the schema column names.
36pub fn materialize_segment_live_rows(
37    blob: &[u8],
38    kek: Option<&nodedb_wal::crypto::WalEncryptionKey>,
39    schema: &ColumnarSchema,
40    deletes: &DeleteBitmap,
41    row_surrogates: &[Option<Surrogate>],
42) -> Result<Vec<(Value, Option<Surrogate>)>, ColumnarError> {
43    // `open_with_kek` is strict: it refuses a plaintext blob when a kek is
44    // supplied and refuses an encrypted blob when no kek is supplied. Pick the
45    // kek by the blob's own magic so both encrypted and plaintext segments in a
46    // single restore decode correctly.
47    let is_encrypted = blob.len() >= 4 && blob[0..4] == crate::encrypt::SEGC_MAGIC;
48    let effective_kek = if is_encrypted { kek } else { None };
49    let owned = OwnedSegmentReader::open_with_kek(blob, effective_kek)?;
50    let reader = owned.reader();
51
52    let total_rows = reader.row_count() as usize;
53    let col_count = schema.columns.len();
54    let col_indices: Vec<usize> = (0..col_count).collect();
55    let decoded_cols = reader.read_columns(&col_indices, &[])?;
56
57    let mut out = Vec::with_capacity(total_rows.saturating_sub(deletes.deleted_count() as usize));
58    for row_idx in 0..total_rows {
59        if !deletes.is_empty() && deletes.is_deleted(row_idx as u32) {
60            continue;
61        }
62        let mut map = std::collections::HashMap::with_capacity(col_count);
63        for (col_idx, decoded) in decoded_cols.iter().enumerate() {
64            let col = &schema.columns[col_idx];
65            let value = extract_row_value(decoded, row_idx, &col.column_type, &col.name)?;
66            map.insert(col.name.clone(), value);
67        }
68        let surrogate = row_surrogates.get(row_idx).copied().flatten();
69        out.push((Value::Object(map), surrogate));
70    }
71    Ok(out)
72}
nodedb_columnar/compaction/materialize.rs

nodedb_columnar/compaction/
materialize.rs
