use crate::compression;
use crate::directory::TileEntry;
use crate::curve::BlobOrdering;
use crate::error::{Error, Result};
use crate::metadata::Metadata;
use crate::tile::TileId;
use crate::types::Compression;
use memmap2::Mmap;
use rayon::prelude::*;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::fs::{self, File, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
pub const PACKED_FORMAT: &str = "stt-packed";
pub const PACKED_FORMAT_VERSION: u32 = 1;
pub const DEFAULT_PACK_TARGET_BYTES: u64 = 64 * 1024 * 1024;
fn crc32c_tag(bytes: &[u8]) -> u32 {
crc32c::crc32c(bytes)
}
fn blake3_128_hex(bytes: &[u8]) -> String {
let hash = blake3::hash(bytes);
hash.as_bytes()[..16]
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
pub const DIRECTORY_ENCODING_ZSTD: &str = "zstd";
pub const DIRECTORY_LAYOUT_PAGED: &str = "paged";
pub const DIRECTORY_LAYOUT_SINGLE: &str = "single";
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DirectoryRef {
pub key: String,
pub length: u64,
#[serde(rename = "directoryVersion")]
pub directory_version: u8,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub encoding: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub layout: Option<String>,
#[serde(default, rename = "rootLength", skip_serializing_if = "Option::is_none")]
pub root_length: Option<u64>,
#[serde(default, rename = "pageCount", skip_serializing_if = "Option::is_none")]
pub page_count: Option<u64>,
#[serde(default, rename = "pageEntries", skip_serializing_if = "Option::is_none")]
pub page_entries: Option<u64>,
}
impl DirectoryRef {
pub fn is_paged(&self) -> bool {
self.layout.as_deref() == Some(DIRECTORY_LAYOUT_PAGED)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PackRef {
pub key: String,
pub length: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Manifest {
pub format: String,
#[serde(rename = "formatVersion")]
pub format_version: u32,
pub compression: String,
pub directory: DirectoryRef,
pub packs: Vec<PackRef>,
pub metadata: Metadata,
}
impl Manifest {
pub fn from_json_bytes(bytes: &[u8]) -> Result<Self> {
serde_json::from_slice(bytes)
.map_err(|e| Error::InvalidArchive(format!("manifest JSON decode failed: {e}")))
}
pub fn to_json_bytes(&self) -> Result<Vec<u8>> {
serde_json::to_vec_pretty(self)
.map_err(|e| Error::Other(format!("manifest JSON encode failed: {e}")))
}
}
struct PendingTile {
z: u8,
x: u32,
y: u32,
hilbert: u64,
time_start: i64,
time_end: i64,
cover_t_min: Option<i64>,
feature_count: u32,
temporal_bucket_ms: Option<u64>,
payload: Vec<u8>,
}
pub struct PackWriter {
out_dir: PathBuf,
ordering: BlobOrdering,
pack_target_bytes: u64,
pending: Vec<PendingTile>,
page_entries: Option<usize>,
zstd_level: i32,
}
impl PackWriter {
pub fn create<P: AsRef<Path>>(
out_dir: P,
ordering: BlobOrdering,
pack_target_bytes: u64,
) -> Result<Self> {
Ok(Self {
out_dir: out_dir.as_ref().to_path_buf(),
ordering,
pack_target_bytes: pack_target_bytes.max(1),
pending: Vec::new(),
page_entries: None,
zstd_level: compression::ZSTD_LEVEL,
})
}
pub fn with_paging(mut self, page_entries: Option<usize>) -> Self {
self.page_entries = page_entries.map(|k| k.max(1));
self
}
pub fn with_zstd_level(mut self, level: i32) -> Self {
self.zstd_level = level.clamp(1, compression::ZSTD_LEVEL_MAX);
self
}
#[allow(clippy::too_many_arguments)]
pub fn add_tile_full(
&mut self,
id: &TileId,
time_start: i64,
time_end: i64,
cover_t_min: Option<i64>,
feature_count: u32,
temporal_bucket_ms: Option<u64>,
payload: &[u8],
) -> Result<()> {
self.pending.push(PendingTile {
z: id.z,
x: id.x,
y: id.y,
hilbert: id.hilbert_index(),
time_start,
time_end,
cover_t_min,
feature_count,
temporal_bucket_ms,
payload: payload.to_vec(),
});
Ok(())
}
pub fn tile_count(&self) -> usize {
self.pending.len()
}
pub fn finalize(self, metadata: &Metadata) -> Result<Manifest> {
let PackWriter {
out_dir,
ordering,
pack_target_bytes,
mut pending,
page_entries,
zstd_level,
} = self;
let base_bucket = metadata.temporal_bucket_ms.max(1) as i64;
let tb = |p: &PendingTile| {
let b = p
.temporal_bucket_ms
.map(|v| v as i64)
.unwrap_or(base_bucket)
.max(1);
p.time_start.div_euclid(b)
};
let (tb_min, tb_max) = pending.iter().fold((i64::MAX, i64::MIN), |(lo, hi), p| {
let t = tb(p);
(lo.min(t), hi.max(t))
});
let tb_span = if pending.is_empty() { 0 } else { tb_max - tb_min };
let ordering = match ordering {
BlobOrdering::Auto => {
let max_z = pending.iter().map(|p| p.z).max().unwrap_or(0) as u32;
let time_bits = crate::curve::bits_for((tb_span.max(0) + 1) as u64);
BlobOrdering::choose(max_z, time_bits)
}
other => other,
};
pending.sort_by_key(|p| {
(
crate::curve::space_time_key(
ordering,
p.z,
p.x,
p.y,
p.hilbert,
p.time_start,
tb(p),
tb_min,
tb_span,
),
p.z,
p.x,
p.y,
p.time_start,
p.temporal_bucket_ms,
)
});
struct Blob {
compressed: Vec<u8>,
uncompressed_size: u32,
crc: u32,
}
let mut blobs: Vec<Blob> = Vec::new();
let mut blob_dedup: HashMap<[u8; 32], usize> = HashMap::new();
let mut tile_blob: Vec<usize> = Vec::with_capacity(pending.len());
const COMPRESS_CHUNK: usize = 8192;
for chunk in pending.chunks(COMPRESS_CHUNK) {
let compressed_chunk: Vec<Vec<u8>> = chunk
.par_iter()
.map(|p| compression::compress_zstd_with_dict_level(&p.payload, None, zstd_level))
.collect::<Result<Vec<_>>>()?;
for (p, compressed) in chunk.iter().zip(compressed_chunk) {
let key = *blake3::hash(&compressed).as_bytes();
let idx = if let Some(&i) = blob_dedup.get(&key) {
i
} else {
let i = blobs.len();
let crc = crc32c_tag(&compressed);
let uncompressed_size = p.payload.len() as u32;
blobs.push(Blob {
compressed,
uncompressed_size,
crc,
});
blob_dedup.insert(key, i);
i
};
tile_blob.push(idx);
}
}
struct Placement {
pack_id: u32,
offset: u64,
}
let mut placements: Vec<Placement> = Vec::with_capacity(blobs.len());
let mut packs_blob_ranges: Vec<(usize, usize)> = Vec::new(); if !blobs.is_empty() {
let mut pack_start = 0usize;
let mut pack_id = 0u32;
let mut cur_offset = 0u64;
for (i, blob) in blobs.iter().enumerate() {
let blen = blob.compressed.len() as u64;
if i > pack_start && cur_offset + blen > pack_target_bytes {
packs_blob_ranges.push((pack_start, i));
pack_start = i;
pack_id += 1;
cur_offset = 0;
}
placements.push(Placement {
pack_id,
offset: cur_offset,
});
cur_offset += blen;
}
packs_blob_ranges.push((pack_start, blobs.len()));
}
let mut entries: Vec<TileEntry> = Vec::with_capacity(pending.len());
for (p, &bi) in pending.iter().zip(tile_blob.iter()) {
let blob = &blobs[bi];
let pl = &placements[bi];
entries.push(TileEntry {
zoom: p.z,
x: p.x,
y: p.y,
time_start: p.time_start,
time_end: p.time_end,
pack_id: pl.pack_id,
offset: pl.offset,
length: blob.compressed.len() as u32,
uncompressed_size: blob.uncompressed_size,
feature_count: p.feature_count,
hilbert: p.hilbert,
crc32c: blob.crc,
temporal_bucket_ms: p.temporal_bucket_ms,
cover_t_min: p.cover_t_min,
});
}
entries.sort_by_key(|e| (e.zoom, e.hilbert, e.time_start, e.temporal_bucket_ms));
let mut metadata = metadata.clone();
metadata.tile_count = entries.len() as u64;
metadata.feature_count = entries.iter().map(|e| u64::from(e.feature_count)).sum();
fs::create_dir_all(&out_dir)?;
let index_dir = out_dir.join("index");
let packs_dir = out_dir.join("packs");
fs::create_dir_all(&index_dir)?;
fs::create_dir_all(&packs_dir)?;
let mut pack_refs: Vec<PackRef> = Vec::with_capacity(packs_blob_ranges.len());
for (start, end) in &packs_blob_ranges {
let mut bytes: Vec<u8> = Vec::new();
for blob in &blobs[*start..*end] {
bytes.extend_from_slice(&blob.compressed);
}
let hex = blake3_128_hex(&bytes);
let rel = format!("packs/{hex}.sttp");
let final_path = out_dir.join(&rel);
write_atomic(&packs_dir, &final_path, &bytes)?;
pack_refs.push(PackRef {
key: rel,
length: bytes.len() as u64,
});
}
let (index_bytes, directory_ref_fields): (Vec<u8>, _) = if let Some(k) = page_entries {
let paged =
crate::directory_page::encode_paged_directory_level(&entries, k, true, zstd_level)?;
(
paged.bytes,
(
Some(DIRECTORY_LAYOUT_PAGED.to_string()),
Some(paged.root_length),
Some(paged.page_count as u64),
Some(paged.page_entries as u64),
),
)
} else {
let index_plain = crate::directory::encode_directory(&entries);
let bytes = compression::compress_zstd_with_dict_level(&index_plain, None, zstd_level)?;
(bytes, (None, None, None, None))
};
let index_hex = blake3_128_hex(&index_bytes);
let index_rel = format!("index/{index_hex}.sttd");
let index_path = out_dir.join(&index_rel);
write_atomic(&index_dir, &index_path, &index_bytes)?;
let (layout, root_length, page_count, page_entries_field) = directory_ref_fields;
let manifest = Manifest {
format: PACKED_FORMAT.to_string(),
format_version: PACKED_FORMAT_VERSION,
compression: "zstd".to_string(),
directory: DirectoryRef {
key: index_rel,
length: index_bytes.len() as u64,
directory_version: crate::directory::DIRECTORY_VERSION,
encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
layout,
root_length,
page_count,
page_entries: page_entries_field,
},
packs: pack_refs,
metadata,
};
let manifest_bytes = manifest.to_json_bytes()?;
let manifest_path = out_dir.join("manifest.json");
let mut f = File::create(&manifest_path)?;
f.write_all(&manifest_bytes)?;
f.flush()?;
Ok(manifest)
}
}
fn decode_directory_object(bytes: &[u8], encoding: Option<&str>) -> Result<Vec<u8>> {
match encoding {
None => Ok(bytes.to_vec()),
Some(DIRECTORY_ENCODING_ZSTD) => compression::decompress_zstd_with_dict(bytes, None),
Some(other) => Err(Error::InvalidArchive(format!(
"unknown directory encoding {other:?} (this reader supports absent or \"zstd\")"
))),
}
}
fn decode_directory_entries(bytes: &[u8], dref: &DirectoryRef) -> Result<Vec<TileEntry>> {
if dref.is_paged() {
let root_length = dref.root_length.ok_or_else(|| {
Error::InvalidArchive("paged directory: manifest missing rootLength".into())
})?;
let zstd = dref.encoding.as_deref() == Some(DIRECTORY_ENCODING_ZSTD);
crate::directory_page::decode_paged_directory(bytes, root_length, zstd)
} else {
let raw = decode_directory_object(bytes, dref.encoding.as_deref())?;
crate::directory::decode_directory(&raw)
}
}
fn write_atomic(dir: &Path, final_path: &Path, bytes: &[u8]) -> Result<()> {
let tmp = dir.join(format!(
".tmp-{}-{}",
std::process::id(),
blake3_128_hex(bytes)
));
{
let mut f = OpenOptions::new()
.write(true)
.create(true)
.truncate(true)
.open(&tmp)?;
f.write_all(bytes)?;
f.flush()?;
}
fs::rename(&tmp, final_path)?;
Ok(())
}
pub fn verify_packed_objects<P: AsRef<Path>>(manifest_path: P) -> Result<Vec<String>> {
let manifest_path = manifest_path.as_ref();
let root = manifest_path
.parent()
.ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?;
let manifest = Manifest::from_json_bytes(&fs::read(manifest_path)?)?;
let mut issues = Vec::new();
if manifest.format != PACKED_FORMAT {
issues.push(format!(
"manifest format is {:?}, expected {PACKED_FORMAT:?}",
manifest.format
));
}
if manifest.format_version != PACKED_FORMAT_VERSION {
issues.push(format!(
"manifest formatVersion is {}, expected {PACKED_FORMAT_VERSION}",
manifest.format_version
));
}
if manifest.directory.directory_version != crate::directory::DIRECTORY_VERSION {
issues.push(format!(
"directoryVersion is {}, expected {}",
manifest.directory.directory_version,
crate::directory::DIRECTORY_VERSION
));
}
fn check_object(
root: &Path,
key: &str,
declared_len: u64,
prefix: &str,
ext: &str,
issues: &mut Vec<String>,
) {
match fs::read(root.join(key)) {
Ok(bytes) => {
if bytes.len() as u64 != declared_len {
issues.push(format!(
"{key}: on-disk length {} != manifest-declared {declared_len}",
bytes.len()
));
}
let expected = format!("{prefix}/{}.{ext}", blake3_128_hex(&bytes));
if key != expected {
issues.push(format!(
"{key}: content-address mismatch (bytes hash to {expected})"
));
}
}
Err(e) => issues.push(format!("{key}: cannot read object ({e})")),
}
}
check_object(
root,
&manifest.directory.key,
manifest.directory.length,
"index",
"sttd",
&mut issues,
);
for p in &manifest.packs {
check_object(root, &p.key, p.length, "packs", "sttp", &mut issues);
}
match fs::read(root.join(&manifest.directory.key)) {
Ok(dir_bytes) => match decode_directory_entries(&dir_bytes, &manifest.directory) {
Ok(entries) => {
if let Some(max_pid) = entries.iter().map(|e| e.pack_id).max() {
if max_pid as usize >= manifest.packs.len() {
issues.push(format!(
"directory references pack_id {max_pid} but the manifest lists only {} pack(s)",
manifest.packs.len()
));
}
}
}
Err(e) => issues.push(format!("directory failed to decode: {e}")),
},
Err(_) => {}
}
if manifest.directory.is_paged() {
match manifest.directory.root_length {
Some(rl) => {
if let Ok(dir_bytes) = fs::read(root.join(&manifest.directory.key)) {
let zstd =
manifest.directory.encoding.as_deref() == Some(DIRECTORY_ENCODING_ZSTD);
match crate::directory_page::verify_paged_structure(&dir_bytes, rl, zstd) {
Ok(mut more) => issues.append(&mut more),
Err(e) => issues.push(format!("paged structure check failed: {e}")),
}
}
}
None => issues.push("paged directory: manifest missing rootLength".into()),
}
}
Ok(issues)
}
struct LoadedPack {
mmap: Option<Mmap>,
path: PathBuf,
length: u64,
}
pub struct PackedReader {
entries: Vec<TileEntry>,
metadata: Metadata,
compression: Compression,
packs: Vec<std::cell::RefCell<LoadedPack>>,
}
impl PackedReader {
pub fn open<P: AsRef<Path>>(manifest_path: P) -> Result<Self> {
let manifest_path = manifest_path.as_ref();
let root = manifest_path
.parent()
.ok_or_else(|| Error::InvalidArchive("manifest path has no parent dir".into()))?
.to_path_buf();
let manifest_bytes = fs::read(manifest_path)?;
let manifest = Manifest::from_json_bytes(&manifest_bytes)?;
if manifest.format != PACKED_FORMAT {
return Err(Error::InvalidArchive(format!(
"not a packed manifest: format={:?} (expected {PACKED_FORMAT:?})",
manifest.format
)));
}
let compression = match manifest.compression.as_str() {
"zstd" => Compression::Zstd,
"none" => Compression::None,
other => {
return Err(Error::InvalidArchive(format!(
"unknown packed compression {other:?}"
)))
}
};
let dir_path = root.join(&manifest.directory.key);
let dir_bytes = fs::read(&dir_path)?;
let entries = decode_directory_entries(&dir_bytes, &manifest.directory)?;
let packs = manifest
.packs
.iter()
.map(|p| {
std::cell::RefCell::new(LoadedPack {
mmap: None,
path: root.join(&p.key),
length: p.length,
})
})
.collect();
Ok(Self {
entries,
metadata: manifest.metadata,
compression,
packs,
})
}
pub fn entries(&self) -> &[TileEntry] {
&self.entries
}
pub fn metadata(&self) -> &Metadata {
&self.metadata
}
pub fn read_payload(&self, entry: &TileEntry) -> Result<Vec<u8>> {
let cell = self.packs.get(entry.pack_id as usize).ok_or_else(|| {
Error::InvalidArchive(format!(
"tile {:?} references pack {} but only {} packs exist",
entry.tile_id(),
entry.pack_id,
self.packs.len()
))
})?;
let payload = {
let mut pack = cell.borrow_mut();
if pack.mmap.is_none() {
let file = File::open(&pack.path)?;
let mmap = unsafe { Mmap::map(&file) }
.map_err(|e| Error::Other(format!("mmap failed: {e}")))?;
if mmap.len() as u64 != pack.length {
return Err(Error::InvalidArchive(format!(
"pack {} is {} bytes, manifest declared {}",
pack.path.display(),
mmap.len(),
pack.length
)));
}
pack.mmap = Some(mmap);
}
let mmap = pack.mmap.as_ref().expect("just loaded");
let start = entry.offset as usize;
let end = start + entry.length as usize;
if end > mmap.len() {
return Err(Error::InvalidArchive(format!(
"tile {:?} blob range {start}..{end} exceeds pack size {}",
entry.tile_id(),
mmap.len()
)));
}
let compressed = &mmap[start..end];
if crc32c_tag(compressed) != entry.crc32c {
return Err(Error::InvalidArchive(format!(
"tile {:?} failed integrity check (corrupt pack)",
entry.tile_id()
)));
}
if self.compression == Compression::Zstd {
compression::decompress_zstd_with_dict(compressed, None)?
} else {
compression::decompress(compressed, self.compression)?
}
};
if payload.len() != entry.uncompressed_size as usize {
return Err(Error::InvalidArchive(format!(
"tile {:?} decompressed to {} bytes, expected {}",
entry.tile_id(),
payload.len(),
entry.uncompressed_size
)));
}
Ok(payload)
}
pub fn read_layers(&self, entry: &TileEntry) -> Result<Vec<crate::arrow_tile::DecodedLayer>> {
let payload = self.read_payload(entry)?;
crate::arrow_tile::decode_tile(&payload)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::arrow_tile::{encode_tile, ColumnarLayer, GeometryColumn};
fn point_layer(name: &str, ids: Vec<u64>, t0: i64) -> ColumnarLayer {
let n = ids.len();
ColumnarLayer {
name: name.to_string(),
feature_ids: ids,
start_times: vec![t0; n],
end_times: vec![t0 + 100; n],
geometry: GeometryColumn::Point(vec![[-122.4, 37.7]; n]),
vertex_times: None,
vertex_values: None,
triangles: None,
vertex_value_matrix: None,
properties: vec![],
}
}
fn distinct_tile(seed: u64) -> Vec<u8> {
let ids: Vec<u64> = (0..6).map(|i| seed * 100 + i).collect();
encode_tile(&[point_layer("default", ids, (seed as i64) * 7)]).unwrap()
}
#[test]
fn blake3_128_hex_is_32_chars() {
let h = blake3_128_hex(b"hello");
assert_eq!(h.len(), 32);
assert!(h.chars().all(|c| c.is_ascii_hexdigit()));
}
#[test]
fn manifest_json_roundtrips() {
let m = Manifest {
format: PACKED_FORMAT.to_string(),
format_version: PACKED_FORMAT_VERSION,
compression: "zstd".to_string(),
directory: DirectoryRef {
key: "index/abc.sttd".to_string(),
length: 42,
directory_version: 5,
encoding: Some(DIRECTORY_ENCODING_ZSTD.to_string()),
layout: None,
root_length: None,
page_count: None,
page_entries: None,
},
packs: vec![
PackRef { key: "packs/a.sttp".to_string(), length: 100 },
PackRef { key: "packs/b.sttp".to_string(), length: 200 },
],
metadata: Metadata::new("manifest-test"),
};
let bytes = m.to_json_bytes().unwrap();
let s = String::from_utf8(bytes.clone()).unwrap();
assert!(s.contains("\"formatVersion\""), "{s}");
assert!(s.contains("\"directoryVersion\""), "{s}");
assert!(s.contains("\"stt-packed\""), "{s}");
let back = Manifest::from_json_bytes(&bytes).unwrap();
assert_eq!(back.format, m.format);
assert_eq!(back.format_version, m.format_version);
assert_eq!(back.packs.len(), 2);
assert_eq!(back.directory.directory_version, 5);
assert_eq!(back.directory.encoding.as_deref(), Some("zstd"));
assert_eq!(back.metadata.name, "manifest-test");
let mut legacy_json: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
legacy_json["directory"]
.as_object_mut()
.unwrap()
.remove("encoding");
let legacy_back =
Manifest::from_json_bytes(&serde_json::to_vec(&legacy_json).unwrap()).unwrap();
assert_eq!(legacy_back.directory.encoding, None);
let legacy_out =
String::from_utf8(legacy_back.to_json_bytes().unwrap()).unwrap();
assert!(!legacy_out.contains("\"encoding\""), "{legacy_out}");
}
#[test]
fn packwriter_roundtrips_through_multiple_packs() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
let static_payload = encode_tile(&[point_layer("default", vec![7, 8, 9], 0)]).unwrap();
let mut expected: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
let bucket = 3_600_000i64;
for k in 0..30u64 {
let zoom = if k % 2 == 0 { 9u8 } else { 10u8 };
let b = (k % 5) as i64; let t = b * bucket;
let payload = if k == 13 || k == 27 {
static_payload.clone() } else {
distinct_tile(k)
};
let id = TileId::new(zoom, (k % 7) as u32, (k / 7) as u32, t as u64);
w.add_tile_full(&id, t, t + bucket - 1, Some(t), 6, Some(bucket as u64), &payload)
.unwrap();
expected.push((id, t, t + bucket - 1, payload));
}
assert_eq!(w.tile_count(), 30);
let meta = Metadata::new("packed-roundtrip").with_temporal_bucket_ms(bucket as u64);
let manifest = w.finalize(&meta).unwrap();
assert!(manifest.packs.len() > 1, "expected multiple packs, got {}", manifest.packs.len());
for p in &manifest.packs {
let bytes = fs::read(out.join(&p.key)).unwrap();
assert_eq!(bytes.len() as u64, p.length);
}
for p in &manifest.packs {
let bytes = fs::read(out.join(&p.key)).unwrap();
let hex = blake3_128_hex(&bytes);
assert_eq!(p.key, format!("packs/{hex}.sttp"));
}
let dir_bytes = fs::read(out.join(&manifest.directory.key)).unwrap();
assert_eq!(
manifest.directory.key,
format!("index/{}.sttd", blake3_128_hex(&dir_bytes))
);
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
let max_pack = reader.entries().iter().map(|e| e.pack_id).max().unwrap();
assert_eq!(max_pack as usize, manifest.packs.len() - 1);
let observed: std::collections::BTreeSet<u32> =
reader.entries().iter().map(|e| e.pack_id).collect();
for pid in 0..manifest.packs.len() as u32 {
assert!(observed.contains(&pid), "pack_id {pid} unused");
}
assert_eq!(reader.metadata().name, "packed-roundtrip");
assert_eq!(reader.metadata().temporal_bucket_ms, bucket as u64);
assert_eq!(reader.entries().len(), 30);
assert_eq!(manifest.metadata.tile_count, 30);
assert_eq!(manifest.metadata.feature_count, 30 * 6);
assert_eq!(reader.metadata().tile_count, 30);
assert_eq!(reader.metadata().feature_count, 30 * 6);
for (id, ts, _te, payload) in &expected {
let e = reader
.entries()
.iter()
.find(|e| {
e.zoom == id.z && e.x == id.x && e.y == id.y && e.time_start == *ts
})
.expect("entry present");
let got = reader.read_payload(e).unwrap();
assert_eq!(&got, payload, "payload mismatch for {id:?}");
}
let static_entries: Vec<&TileEntry> = reader
.entries()
.iter()
.filter(|e| {
let g = reader.read_payload(e).unwrap();
g == static_payload
})
.collect();
assert_eq!(static_entries.len(), 2);
assert_eq!(static_entries[0].pack_id, static_entries[1].pack_id);
assert_eq!(static_entries[0].offset, static_entries[1].offset);
}
#[test]
fn paged_directory_writer_roundtrips_and_matches_single() {
let bucket = 3_600_000i64;
let mut input: Vec<(TileId, i64, i64, Vec<u8>)> = Vec::new();
for k in 0..120u64 {
let zoom = [6u8, 10, 13][(k % 3) as usize];
let b = (k % 4) as i64;
let t = b * bucket;
let id = TileId::new(zoom, (k % 11) as u32, (k / 11) as u32, t as u64);
input.push((id, t, t + bucket - 1, distinct_tile(k)));
}
let build = |out: &Path, page_entries: Option<usize>| -> Manifest {
let mut w = PackWriter::create(out, BlobOrdering::Auto, 16 * 1024)
.unwrap()
.with_paging(page_entries);
for (id, ts, te, payload) in &input {
w.add_tile_full(id, *ts, *te, Some(*ts), 6, Some(bucket as u64), payload)
.unwrap();
}
let meta = Metadata::new("paged-roundtrip").with_temporal_bucket_ms(bucket as u64);
w.finalize(&meta).unwrap()
};
let dir = tempfile::tempdir().unwrap();
let single_out = dir.path().join("single");
let paged_out = dir.path().join("paged");
let single = build(&single_out, None);
let paged = build(&paged_out, Some(16));
assert!(single.directory.layout.is_none());
assert_eq!(paged.directory.layout.as_deref(), Some(DIRECTORY_LAYOUT_PAGED));
assert!(paged.directory.root_length.unwrap() > 0);
assert!(paged.directory.page_count.unwrap() >= 2, "expected multiple leaf pages");
assert_eq!(paged.directory.page_entries, Some(16));
assert_eq!(paged.directory.directory_version, crate::directory::DIRECTORY_VERSION);
assert_eq!(paged.directory.encoding.as_deref(), Some(DIRECTORY_ENCODING_ZSTD));
let r_single = PackedReader::open(single_out.join("manifest.json")).unwrap();
let r_paged = PackedReader::open(paged_out.join("manifest.json")).unwrap();
assert_eq!(r_paged.entries().len(), 120);
assert_eq!(r_single.entries().len(), 120);
let keys = |r: &PackedReader| -> Vec<(u8, u32, u32, i64, i64, u32, Option<u64>, Option<i64>)> {
let mut v: Vec<_> = r
.entries()
.iter()
.map(|e| {
(
e.zoom, e.x, e.y, e.time_start, e.time_end, e.feature_count,
e.temporal_bucket_ms, e.cover_t_min,
)
})
.collect();
v.sort();
v
};
assert_eq!(keys(&r_single), keys(&r_paged));
for (id, ts, _te, payload) in &input {
let e = r_paged
.entries()
.iter()
.find(|x| x.zoom == id.z && x.x == id.x && x.y == id.y && x.time_start == *ts)
.expect("paged entry present");
assert_eq!(&r_paged.read_payload(e).unwrap(), payload, "payload mismatch {id:?}");
}
let issues = verify_packed_objects(paged_out.join("manifest.json")).unwrap();
assert!(issues.is_empty(), "paged verify issues: {issues:?}");
}
#[test]
fn oversized_blob_gets_its_own_pack() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let big_ids: Vec<u64> = (0..4000).collect();
let big = encode_tile(&[ColumnarLayer {
name: "default".to_string(),
feature_ids: big_ids.clone(),
start_times: vec![0; big_ids.len()],
end_times: vec![100; big_ids.len()],
geometry: GeometryColumn::Point(
(0..big_ids.len()).map(|i| [i as f64 * 0.01, i as f64 * 0.013]).collect(),
),
vertex_times: None,
vertex_values: None,
triangles: None,
vertex_value_matrix: None,
properties: vec![],
}])
.unwrap();
let big_compressed_len = compression::compress_zstd_with_dict(&big, None).unwrap().len();
let mut w = PackWriter::create(&out, BlobOrdering::SpatialMajor, 4 * 1024).unwrap();
assert!(big_compressed_len as u64 > 4 * 1024);
w.add_tile_full(&TileId::new(10, 0, 0, 0), 0, 100, None, 4000, None, &big).unwrap();
for k in 1..4u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
}
let _manifest = w.finalize(&Metadata::new("big")).unwrap();
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
let big_entry = reader.entries().iter().find(|e| e.x == 0).unwrap();
assert_eq!(reader.read_payload(big_entry).unwrap(), big);
}
#[test]
fn corrupt_pack_blob_is_detected_on_read() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 64 * 1024 * 1024).unwrap();
for k in 0..4u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p).unwrap();
}
let manifest = w.finalize(&Metadata::new("crc")).unwrap();
let entry = PackedReader::open(out.join("manifest.json")).unwrap().entries()[0].clone();
assert!(PackedReader::open(out.join("manifest.json"))
.unwrap()
.read_payload(&entry)
.is_ok());
let pack_path = out.join(&manifest.packs[entry.pack_id as usize].key);
let mut bytes = fs::read(&pack_path).unwrap();
bytes[entry.offset as usize] ^= 0xff;
fs::write(&pack_path, &bytes).unwrap();
let reader = PackedReader::open(out.join("manifest.json")).unwrap();
assert!(
reader.read_payload(&entry).is_err(),
"corrupt pack blob must fail the read-path CRC32C check"
);
}
#[test]
fn rebuilds_are_byte_reproducible() {
let bucket = 3_600_000i64;
let mut tiles: Vec<(TileId, i64, Option<u64>, Vec<u8>)> = Vec::new();
for k in 0..10u64 {
let t = (k % 4) as i64 * bucket;
tiles.push((
TileId::new(9, (k % 5) as u32, (k / 5) as u32, t as u64),
t,
None,
distinct_tile(k),
));
}
tiles.push((TileId::new(9, 1, 0, 0), 0, None, distinct_tile(100)));
tiles.push((
TileId::new(9, 1, 0, 0),
0,
Some(24 * bucket as u64),
distinct_tile(101),
));
let meta = Metadata::new("repro").with_temporal_bucket_ms(bucket as u64);
let build = |order: &[usize]| {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
for &i in order {
let (id, t, b, payload) = &tiles[i];
w.add_tile_full(id, *t, t + bucket - 1, Some(*t), 6, *b, payload)
.unwrap();
}
let manifest = w.finalize(&meta).unwrap();
(dir, manifest)
};
let forward: Vec<usize> = (0..tiles.len()).collect();
let reverse: Vec<usize> = (0..tiles.len()).rev().collect();
let (_d1, m1) = build(&forward);
let (_d2, m2) = build(&reverse);
assert_eq!(m1.directory.key, m2.directory.key, "directory hash must be stable");
assert_eq!(
m1.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
m2.packs.iter().map(|p| &p.key).collect::<Vec<_>>(),
"pack hashes must be stable across rebuilds"
);
assert_eq!(m1.to_json_bytes().unwrap(), m2.to_json_bytes().unwrap());
}
#[test]
fn directory_encoding_compressed_and_raw_both_read() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
for k in 0..8u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
.unwrap();
}
let manifest = w.finalize(&Metadata::new("dir-enc")).unwrap();
let manifest_path = out.join("manifest.json");
assert_eq!(manifest.directory.encoding.as_deref(), Some("zstd"));
let at_rest = fs::read(out.join(&manifest.directory.key)).unwrap();
assert_eq!(at_rest.len() as u64, manifest.directory.length);
let raw = compression::decompress_zstd_with_dict(&at_rest, None).unwrap();
assert!(crate::directory::decode_directory(&raw).is_ok());
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let entries_compressed = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
let raw_hex = blake3_128_hex(&raw);
let raw_rel = format!("index/{raw_hex}.sttd");
fs::write(out.join(&raw_rel), &raw).unwrap();
let mut legacy = manifest.clone();
legacy.directory = DirectoryRef {
key: raw_rel,
length: raw.len() as u64,
directory_version: crate::directory::DIRECTORY_VERSION,
encoding: None,
layout: None,
root_length: None,
page_count: None,
page_entries: None,
};
fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let entries_raw = PackedReader::open(&manifest_path).unwrap().entries().to_vec();
assert_eq!(entries_raw, entries_compressed);
legacy.directory.encoding = Some("br".to_string());
fs::write(&manifest_path, legacy.to_json_bytes().unwrap()).unwrap();
assert!(PackedReader::open(&manifest_path).is_err());
}
#[test]
fn verify_packed_objects_clean_then_detects_corruption() {
let dir = tempfile::tempdir().unwrap();
let out = dir.path().join("dataset");
let mut w = PackWriter::create(&out, BlobOrdering::Auto, 8 * 1024).unwrap();
for k in 0..12u64 {
let p = distinct_tile(k);
w.add_tile_full(&TileId::new(10, k as u32, 0, 0), 0, 100, None, 6, None, &p)
.unwrap();
}
let manifest = w.finalize(&Metadata::new("verify")).unwrap();
let manifest_path = out.join("manifest.json");
assert!(verify_packed_objects(&manifest_path).unwrap().is_empty());
let pack0 = out.join(&manifest.packs[0].key);
let mut bytes = fs::read(&pack0).unwrap();
bytes[0] ^= 0xff;
fs::write(&pack0, &bytes).unwrap();
let issues = verify_packed_objects(&manifest_path).unwrap();
assert!(
issues.iter().any(|s| s.contains("content-address mismatch")),
"expected a content-address mismatch, got {issues:?}"
);
}
}