Function write_packed

nodedb/src/engine/timeseries/packed_partition.rs:196–258

Write a packed partition file from individual column files. When `kek` is `Some`, the columns + sparse index payload region is wrapped in a `SEGT` AES-256-GCM envelope. The NDPK footer remains in plaintext to allow footer-only range requests without decryption. The `column_ranges` in the footer record offsets within the *decrypted payload blob, not the on-disk ciphertext, so `read_column` can pa

(
    output_path: &Path,
    columns: &[(String, Vec<u8>)], // (column_name, compressed_bytes)
    sparse_index: Option<&[u8]>,
    schema: &[PackedColumnSchema],
    meta: &PartitionMeta,
    kek: O

/// payload blob, not the on-disk ciphertext, so `read_column` can pass them
/// through after decryption.
pub fn write_packed(
    output_path: &Path,
    columns: &[(String, Vec<u8>)], // (column_name, compressed_bytes)
    sparse_index: Option<&[u8]>,
    schema: &[PackedColumnSchema],
    meta: &PartitionMeta,
    kek: Option<&WalEncryptionKey>,
) -> Result<u64, PackedError> {
    let mut payload = Vec::new();
    let mut column_ranges = HashMap::new();

    // Build the payload (columns + sparse index).
    for (name, data) in columns {
        let offset = payload.len() as u64;
        payload.extend_from_slice(data);
        column_ranges.insert(name.clone(), (offset, data.len() as u64));
    }

    let sparse_index_range = sparse_index.map(|idx_data| {
        let offset = payload.len() as u64;
        payload.extend_from_slice(idx_data);
        (offset, idx_data.len() as u64)
    });

    // Optionally encrypt the payload.
    let payload_encrypted = kek.is_some();
    let payload_on_disk: Vec<u8> = if let Some(key) = kek {
        encrypt_file(key, &payload).map_err(PackedError::from)?
    } else {
        payload
    };

    // Build footer (plaintext).
    let footer = PackedFooter {
        column_ranges,
        sparse_index_range,
        schema: schema.to_vec(),
        meta: meta.clone(),
        payload_encrypted,
    };
    let footer_body = zerompk::to_msgpack_vec(&footer)
        .map_err(|e| PackedError::Io(format!("footer encode: {e}")))?;

    // Compose the final file.
    let mut buf = Vec::with_capacity(payload_on_disk.len() + footer_body.len() + PACKED_TAIL_SIZE);
    buf.extend_from_slice(&payload_on_disk);
    buf.extend_from_slice(&footer_body);

    let footer_len = footer_body.len() as u32;
    let crc = crc32c::crc32c(&footer_body);

    // Tail: footer_len (4B) | crc (4B) | version (2B) | magic (4B)
    buf.extend_from_slice(&footer_len.to_le_bytes());
    buf.extend_from_slice(&crc.to_le_bytes());
    buf.extend_from_slice(&PACKED_VERSION.to_le_bytes());
    buf.extend_from_slice(&PACKED_MAGIC);

    let total_size = buf.len() as u64;