1use std::collections::HashMap;
13use std::fs::File;
14use std::io::{Read, Seek, SeekFrom};
15use std::path::{Path, PathBuf};
16use std::sync::Arc;
17
18use crate::common_config::StrategicConfig;
19use crate::meta::BlobMeta;
20use crate::plugin::ExtensionRow;
21use crate::{decompress_archive};
22use anyhow::Result;
23use arrow::array::{
24 Array, ArrayRef, BooleanBuilder, FixedSizeBinaryBuilder, Int8Builder, ListBuilder, StringBuilder,
25 UInt32Builder, UInt64Builder,
26};
27use arrow::datatypes::{DataType, Field, Schema};
28use arrow::record_batch::RecordBatch;
29use once_cell::sync::Lazy;
30
31pub type FileExtMeta = Option<(i8, ExtensionRow)>;
34
35pub static ZNIPPY_INDEX_SCHEMA: Lazy<Arc<Schema>> = Lazy::new(|| {
40 Arc::new(Schema::new(base_index_fields()))
41});
42
43fn base_index_fields() -> Vec<Field> {
44 vec![
45 Field::new("relative_path", DataType::Utf8, false),
46 Field::new("chunk_seq", DataType::UInt32, false),
47 Field::new("fdata_offset", DataType::UInt64, false),
48 Field::new("compressed", DataType::Boolean, false),
49 Field::new("uncompressed_size", DataType::UInt64, false),
50 Field::new("blob_offset", DataType::UInt64, false),
51 Field::new("blob_size", DataType::UInt64, false),
52 Field::new("checksum", DataType::FixedSizeBinary(32), false),
53 ]
54}
55
56pub fn znippy_index_schema() -> &'static Arc<Schema> {
57 &ZNIPPY_INDEX_SCHEMA
58}
59
60pub fn compose_index_schema(ext_fields: &[Field]) -> Arc<Schema> {
64 let mut fields = base_index_fields();
65 if !ext_fields.is_empty() {
66 fields.push(Field::new("pkg_type", DataType::Int8, true));
67 fields.extend(ext_fields.iter().cloned());
68 }
69 Arc::new(Schema::new(fields))
70}
71
72pub const ZNIPPY_FORMAT_VERSION: u32 = 3;
78
79pub const FORMAT_VERSION_KEY: &str = "znippy_format_version";
81
82pub fn check_format_version(metadata: &HashMap<String, String>) -> Result<()> {
87 if let Some(raw) = metadata.get(FORMAT_VERSION_KEY) {
88 let version: u32 = raw
89 .parse()
90 .map_err(|_| anyhow::anyhow!("invalid znippy archive format version {raw:?}"))?;
91 anyhow::ensure!(
92 version <= ZNIPPY_FORMAT_VERSION,
93 "znippy archive format v{version} is newer than this reader supports \
94 (max v{ZNIPPY_FORMAT_VERSION}) — upgrade znippy",
95 );
96 }
97 Ok(())
98}
99
100pub fn build_arrow_metadata_for_config(config: &StrategicConfig) -> HashMap<String, String> {
102 let mut m = HashMap::new();
103 m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
104 m.insert("max_core_in_flight".into(), config.max_core_in_flight.to_string());
105 m.insert("max_core_in_compress".into(), config.max_core_in_compress.to_string());
106 m.insert("max_mem_allowed".into(), config.max_mem_allowed.to_string());
107 m.insert("min_free_memory_ratio".into(), config.min_free_memory_ratio.to_string());
108 m.insert("file_split_block_size".into(), config.file_split_block_size.to_string());
109 m.insert("max_chunks".into(), config.max_chunks.to_string());
110 m.insert("compression_level".into(), config.compression_level.to_string());
111 m.insert("zstd_output_buffer_size".into(), config.zstd_output_buffer_size.to_string());
112 m
113}
114
115pub fn extract_config_from_arrow_metadata(
116 metadata: &HashMap<String, String>,
117) -> anyhow::Result<StrategicConfig> {
118 Ok(StrategicConfig {
119 max_core_allowed: 0,
120 max_core_in_flight: metadata
121 .get("max_core_in_flight")
122 .ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_flight'"))?
123 .parse()?,
124 max_core_in_compress: metadata
125 .get("max_core_in_compress")
126 .ok_or_else(|| anyhow::anyhow!("Missing 'max_core_in_compress'"))?
127 .parse()?,
128 max_mem_allowed: metadata
129 .get("max_mem_allowed")
130 .ok_or_else(|| anyhow::anyhow!("Missing 'max_mem_allowed'"))?
131 .parse()?,
132 min_free_memory_ratio: metadata
133 .get("min_free_memory_ratio")
134 .ok_or_else(|| anyhow::anyhow!("Missing 'min_free_memory_ratio'"))?
135 .parse()?,
136 file_split_block_size: metadata
137 .get("file_split_block_size")
138 .ok_or_else(|| anyhow::anyhow!("Missing 'file_split_block_size'"))?
139 .parse()?,
140 max_chunks: metadata
141 .get("max_chunks")
142 .ok_or_else(|| anyhow::anyhow!("Missing 'max_chunks'"))?
143 .parse()?,
144 compression_level: metadata
145 .get("compression_level")
146 .ok_or_else(|| anyhow::anyhow!("Missing 'compression_level'"))?
147 .parse()?,
148 zstd_output_buffer_size: metadata
149 .get("zstd_output_buffer_size")
150 .ok_or_else(|| anyhow::anyhow!("Missing 'zstd_output_buffer_size'"))?
151 .parse()?,
152 })
153}
154
155pub fn build_metadata_batch<F>(
160 blobs: &[BlobMeta],
161 path_resolver: F,
162 ext_meta: &[FileExtMeta],
163 ext_fields: &[Field],
164) -> arrow::error::Result<RecordBatch>
165where
166 F: Fn(u64) -> String,
167{
168 let len = blobs.len();
169
170 let mut path_builder = StringBuilder::with_capacity(len, len * 64);
171 let mut seq_builder = UInt32Builder::with_capacity(len);
172 let mut fdata_builder = UInt64Builder::with_capacity(len);
173 let mut compressed_builder = BooleanBuilder::with_capacity(len);
174 let mut size_builder = UInt64Builder::with_capacity(len);
175 let mut blob_offset_builder = UInt64Builder::with_capacity(len);
176 let mut blob_size_builder = UInt64Builder::with_capacity(len);
177 let mut checksum_builder = FixedSizeBinaryBuilder::with_capacity(len, 32);
178
179 for blob in blobs {
180 let m = &blob.chunk_meta;
181 path_builder.append_value(path_resolver(m.file_index));
182 seq_builder.append_value(m.chunk_seq);
183 fdata_builder.append_value(m.fdata_offset);
184 compressed_builder.append_value(m.compressed);
185 size_builder.append_value(m.uncompressed_size);
186 blob_offset_builder.append_value(blob.blob_offset);
187 blob_size_builder.append_value(blob.blob_size);
188 checksum_builder.append_value(m.checksum)?;
189 }
190
191 let mut columns: Vec<ArrayRef> = vec![
192 Arc::new(path_builder.finish()),
193 Arc::new(seq_builder.finish()),
194 Arc::new(fdata_builder.finish()),
195 Arc::new(compressed_builder.finish()),
196 Arc::new(size_builder.finish()),
197 Arc::new(blob_offset_builder.finish()),
198 Arc::new(blob_size_builder.finish()),
199 Arc::new(checksum_builder.finish()),
200 ];
201
202 if !ext_fields.is_empty() {
204 let mut pkg_type_builder = Int8Builder::with_capacity(len);
205 for blob in blobs {
206 match ext_meta.get(blob.chunk_meta.file_index as usize).and_then(|x| x.as_ref()) {
207 Some((type_id, _)) => pkg_type_builder.append_value(*type_id),
208 None => pkg_type_builder.append_null(),
209 }
210 }
211 columns.push(Arc::new(pkg_type_builder.finish()));
212
213 for field in ext_fields {
214 columns.push(build_ext_column(field, blobs, ext_meta));
215 }
216 }
217
218 RecordBatch::try_new(compose_index_schema(ext_fields), columns)
219}
220
221fn build_ext_column(field: &Field, blobs: &[BlobMeta], ext_meta: &[FileExtMeta]) -> ArrayRef {
224 use crate::plugin::ExtensionValue;
225 let len = blobs.len();
226 let value_for = |blob: &BlobMeta| -> Option<&ExtensionValue> {
227 ext_meta
228 .get(blob.chunk_meta.file_index as usize)
229 .and_then(|x| x.as_ref())
230 .and_then(|(_, row)| row.fields.get(field.name()))
231 };
232
233 match field.data_type() {
234 DataType::UInt32 => {
235 let mut b = UInt32Builder::with_capacity(len);
236 for blob in blobs {
237 match value_for(blob) {
238 Some(ExtensionValue::U32(n)) => b.append_value(*n),
239 _ => b.append_null(),
240 }
241 }
242 Arc::new(b.finish())
243 }
244 DataType::List(item) if item.data_type() == &DataType::Utf8 => {
246 let mut b = ListBuilder::new(StringBuilder::new()).with_field(Arc::clone(item));
247 for blob in blobs {
248 match value_for(blob) {
249 Some(ExtensionValue::StrList(v)) => {
250 for s in v {
251 b.values().append_value(s);
252 }
253 b.append(true);
254 }
255 _ => b.append(false),
256 }
257 }
258 Arc::new(b.finish())
259 }
260 _ => {
262 let mut b = StringBuilder::with_capacity(len, len * 16);
263 for blob in blobs {
264 match value_for(blob) {
265 Some(ExtensionValue::Str(s)) => b.append_value(s),
266 Some(ExtensionValue::OptStr(Some(s))) => b.append_value(s),
267 _ => b.append_null(),
268 }
269 }
270 Arc::new(b.finish())
271 }
272 }
273}
274
275pub const MULTI_INDEX_MAGIC: [u8; 8] = *b"ZNPYMIDX";
292
293#[derive(Debug, Clone, PartialEq)]
295pub struct ManifestEntry {
296 pub pkg_type: i8,
297 pub repo: String,
298 pub module_name: String,
299 pub index_offset: u64,
300 pub index_len: u64,
301 pub row_count: u64,
302}
303
304pub const LOOKUP_MODULE: &str = "__znippy_lookup__";
310
311pub const TRIE_MODULE: &str = "__znippy_trie__";
315
316pub const SIGN_ARTIFACTS_MODULE: &str = "__znippy_sign_artifacts__";
322
323pub const SIGN_ARCHIVE_MODULE: &str = "__znippy_sign_archive__";
326
327pub const META_MODULE: &str = "__znippy_meta__";
339
340pub const GUNNAR_OID_MODULE: &str = "__gunnar_oid__";
345
346pub const GUNNAR_GRAPH_MODULE: &str = "__gunnar_graph__";
352
353pub const GUNNAR_REACH_MODULE: &str = "__gunnar_reach__";
357
358pub const GUNNAR_REFS_MODULE: &str = "__gunnar_refs__";
368
369pub const GUNNAR_SECRETS_MODULE: &str = "__gunnar_secrets__";
374
375pub const ZNIPPY_DELTA_MODULE: &str = "__znippy_delta__";
402
403pub const ZNIPPY_FSMETA_MODULE: &str = "__znippy_fsmeta__";
441
442pub fn delta_map_schema() -> Arc<Schema> {
444 Arc::new(Schema::new(vec![
445 Field::new("relative_path", DataType::Utf8, false),
446 Field::new("chunk_seq", DataType::UInt32, false),
447 Field::new("base_path", DataType::Utf8, false),
448 ]))
449}
450
451pub const RESERVED_PKG_TYPE: i8 = i8::MIN;
453
454pub const RESERVED_MODULES: &[&str] = &[
463 LOOKUP_MODULE,
464 TRIE_MODULE,
465 SIGN_ARTIFACTS_MODULE,
466 SIGN_ARCHIVE_MODULE,
467 META_MODULE,
468 GUNNAR_OID_MODULE,
469 GUNNAR_GRAPH_MODULE,
470 GUNNAR_REACH_MODULE,
471 GUNNAR_REFS_MODULE,
472 GUNNAR_SECRETS_MODULE,
473 ZNIPPY_DELTA_MODULE,
474 ZNIPPY_FSMETA_MODULE,
475];
476
477pub fn is_reserved_module(module_name: &str) -> bool {
483 RESERVED_MODULES.contains(&module_name)
484}
485
486pub const CARRIED_RESERVED_MODULES: &[&str] =
518 &[GUNNAR_REFS_MODULE, GUNNAR_SECRETS_MODULE, ZNIPPY_FSMETA_MODULE];
519
520pub fn is_carried_reserved_module(module_name: &str) -> bool {
523 CARRIED_RESERVED_MODULES.contains(&module_name)
524}
525
526pub fn read_reserved_section_bytes(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
531 read_reserved_section(path, module_name)
532}
533
534pub fn lookup_schema() -> Arc<Schema> {
543 Arc::new(Schema::new(base_index_fields()))
544}
545
546pub fn data_subindex_schema() -> Arc<Schema> {
558 Arc::new(Schema::new_with_metadata(
559 base_index_fields(),
560 build_arrow_metadata_for_config(&crate::common_config::CONFIG),
561 ))
562}
563
564#[derive(Debug, Clone, PartialEq, Eq)]
566pub struct ChunkLoc {
567 pub chunk_seq: u32,
568 pub fdata_offset: u64,
569 pub blob_offset: u64,
570 pub blob_size: u64,
571 pub uncompressed_size: u64,
572 pub compressed: bool,
573 pub checksum: [u8; 32],
574}
575
576#[derive(Debug, Clone, PartialEq)]
578pub enum IndexFooter {
579 Single { index_offset: u64 },
583 Multi { manifest_offset: u64 },
585}
586
587pub fn interpret_footer(tail: &[u8]) -> IndexFooter {
590 let n = tail.len();
591 let offset = u64::from_le_bytes(tail[n - 8..].try_into().unwrap());
592 if n >= 16 && tail[n - 16..n - 8] == MULTI_INDEX_MAGIC {
593 IndexFooter::Multi { manifest_offset: offset }
594 } else {
595 IndexFooter::Single { index_offset: offset }
596 }
597}
598
599fn manifest_schema() -> Arc<Schema> {
600 Arc::new(Schema::new(vec![
601 Field::new("pkg_type", DataType::Int8, false),
602 Field::new("repo", DataType::Utf8, false),
603 Field::new("module_name", DataType::Utf8, false),
604 Field::new("index_offset", DataType::UInt64, false),
605 Field::new("index_len", DataType::UInt64, false),
606 Field::new("row_count", DataType::UInt64, false),
607 ]))
608}
609
610pub fn write_manifest_bytes(entries: &[ManifestEntry]) -> Result<Vec<u8>> {
612 use arrow::ipc::writer::StreamWriter;
613
614 let len = entries.len();
615 let mut pkg_type = Int8Builder::with_capacity(len);
616 let mut repo = StringBuilder::with_capacity(len, len * 16);
617 let mut module_name = StringBuilder::with_capacity(len, len * 16);
618 let mut index_offset = UInt64Builder::with_capacity(len);
619 let mut index_len = UInt64Builder::with_capacity(len);
620 let mut row_count = UInt64Builder::with_capacity(len);
621 for e in entries {
622 pkg_type.append_value(e.pkg_type);
623 repo.append_value(&e.repo);
624 module_name.append_value(&e.module_name);
625 index_offset.append_value(e.index_offset);
626 index_len.append_value(e.index_len);
627 row_count.append_value(e.row_count);
628 }
629
630 let schema = manifest_schema();
631 let batch = RecordBatch::try_new(
632 schema.clone(),
633 vec![
634 Arc::new(pkg_type.finish()),
635 Arc::new(repo.finish()),
636 Arc::new(module_name.finish()),
637 Arc::new(index_offset.finish()),
638 Arc::new(index_len.finish()),
639 Arc::new(row_count.finish()),
640 ],
641 )?;
642
643 let mut buf = Vec::new();
644 {
645 let mut w = StreamWriter::try_new(&mut buf, &schema)?;
646 w.write(&batch)?;
647 w.finish()?;
648 }
649 Ok(buf)
650}
651
652pub fn read_manifest_bytes(bytes: &[u8]) -> Result<Vec<ManifestEntry>> {
654 use arrow::array::{Int8Array, StringArray, UInt64Array};
655 use arrow::ipc::reader::StreamReader;
656
657 let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)?;
658 let mut out = Vec::new();
659 for batch in reader {
660 let batch = batch?;
661 let col = |name: &str| batch.column_by_name(name)
662 .ok_or_else(|| anyhow::anyhow!("manifest missing column {name}"));
663 let pkg_type = col("pkg_type")?.as_any().downcast_ref::<Int8Array>()
664 .ok_or_else(|| anyhow::anyhow!("pkg_type type"))?;
665 let repo = col("repo")?.as_any().downcast_ref::<StringArray>()
666 .ok_or_else(|| anyhow::anyhow!("repo type"))?;
667 let module_name = col("module_name")?.as_any().downcast_ref::<StringArray>()
668 .ok_or_else(|| anyhow::anyhow!("module_name type"))?;
669 let index_offset = col("index_offset")?.as_any().downcast_ref::<UInt64Array>()
670 .ok_or_else(|| anyhow::anyhow!("index_offset type"))?;
671 let index_len = col("index_len")?.as_any().downcast_ref::<UInt64Array>()
672 .ok_or_else(|| anyhow::anyhow!("index_len type"))?;
673 let row_count = col("row_count")?.as_any().downcast_ref::<UInt64Array>()
674 .ok_or_else(|| anyhow::anyhow!("row_count type"))?;
675 for i in 0..batch.num_rows() {
676 out.push(ManifestEntry {
677 pkg_type: pkg_type.value(i),
678 repo: repo.value(i).to_string(),
679 module_name: module_name.value(i).to_string(),
680 index_offset: index_offset.value(i),
681 index_len: index_len.value(i),
682 row_count: row_count.value(i),
683 });
684 }
685 }
686 Ok(out)
687}
688
689pub fn read_znippy_index(path: &Path) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
695 read_znippy_index_filtered(path, &IndexFilter::default())
696}
697
698#[derive(Debug, Clone, Default)]
704pub struct IndexFilter {
705 pub pkg_type: Option<i8>,
707 pub repo: Option<String>,
709}
710
711impl IndexFilter {
712 pub fn is_empty(&self) -> bool {
713 self.pkg_type.is_none() && self.repo.is_none()
714 }
715 fn matches(&self, e: &ManifestEntry) -> bool {
716 self.pkg_type.is_none_or(|t| e.pkg_type == t)
717 && self.repo.as_deref().is_none_or(|r| e.repo == r)
718 }
719}
720
721pub fn read_znippy_index_filtered(
723 path: &Path,
724 filter: &IndexFilter,
725) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
726 let mut file = File::open(path)?;
727 let file_len = file.metadata()?.len();
728 anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 znippy archive");
729
730 file.seek(SeekFrom::End(-16))?;
731 let mut tail = [0u8; 16];
732 file.read_exact(&mut tail)?;
733
734 match interpret_footer(&tail) {
735 IndexFooter::Multi { manifest_offset } => {
736 read_multi_index(&mut file, file_len, manifest_offset, filter)
737 }
738 IndexFooter::Single { .. } => {
739 anyhow::bail!("v0.6 archives are not supported; re-compress with v0.7")
740 }
741 }
742}
743
744fn read_multi_index(
746 file: &mut File,
747 file_len: u64,
748 manifest_offset: u64,
749 filter: &IndexFilter,
750) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
751 use arrow::ipc::reader::StreamReader;
752
753 let manifest_end = file_len.checked_sub(16)
755 .ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
756 anyhow::ensure!(manifest_offset <= manifest_end, "corrupt v0.7 manifest_offset");
757 let manifest_len = (manifest_end - manifest_offset) as usize;
758
759 file.seek(SeekFrom::Start(manifest_offset))?;
760 let mut manifest_bytes = vec![0u8; manifest_len];
761 file.read_exact(&mut manifest_bytes)?;
762 let entries = read_manifest_bytes(&manifest_bytes)?;
763
764 let mut all_batches: Vec<RecordBatch> = Vec::new();
765 let mut schema: Option<Arc<Schema>> = None;
766
767 for entry in &entries {
768 if is_reserved_module(&entry.module_name) {
771 continue;
772 }
773 if !filter.matches(entry) {
775 continue;
776 }
777 anyhow::ensure!(
780 entry.index_offset.checked_add(entry.index_len)
781 .is_some_and(|end| end <= file_len),
782 "sub-index for module {} out of bounds (offset={}, len={}, file_len={})",
783 entry.module_name, entry.index_offset, entry.index_len, file_len
784 );
785 file.seek(SeekFrom::Start(entry.index_offset))?;
786 let mut sub_bytes = vec![0u8; entry.index_len as usize];
787 file.read_exact(&mut sub_bytes)?;
788 let cursor = std::io::Cursor::new(sub_bytes);
789 let reader = StreamReader::try_new(cursor, None)?;
790 if schema.is_none() {
791 let sub_schema = reader.schema();
792 check_format_version(sub_schema.metadata())?;
795 schema = Some(sub_schema);
796 }
797 for batch in reader {
798 all_batches.push(batch.map_err(|e| anyhow::anyhow!("sub-index read error: {}", e))?);
799 }
800 }
801
802 let schema = schema.unwrap_or_else(|| Arc::new(Schema::new(base_index_fields())));
803
804 let merged = if all_batches.len() <= 1 {
806 all_batches
807 } else {
808 let batch = arrow_select::concat::concat_batches(&schema, all_batches.iter())
809 .map_err(|e| anyhow::anyhow!("concat sub-indexes: {}", e))?;
810 vec![batch]
811 };
812
813 Ok((schema, merged))
814}
815
816pub fn read_znippy_manifest(path: &Path) -> Result<Vec<ManifestEntry>> {
819 let mut file = File::open(path)?;
820 let file_len = file.metadata()?.len();
821 anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 archive");
822
823 file.seek(SeekFrom::End(-16))?;
824 let mut tail = [0u8; 16];
825 file.read_exact(&mut tail)?;
826
827 match interpret_footer(&tail) {
828 IndexFooter::Single { .. } => {
829 anyhow::bail!("not a v0.7 multi-index archive (no MULTI_INDEX_MAGIC)")
830 }
831 IndexFooter::Multi { manifest_offset } => {
832 let manifest_end = file_len - 16;
833 anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
834 let manifest_len = (manifest_end - manifest_offset) as usize;
835 file.seek(SeekFrom::Start(manifest_offset))?;
836 let mut manifest_bytes = vec![0u8; manifest_len];
837 file.read_exact(&mut manifest_bytes)?;
838 let mut entries = read_manifest_bytes(&manifest_bytes)?;
839 entries.retain(|e| !is_reserved_module(&e.module_name));
841 Ok(entries)
842 }
843 }
844}
845
846pub fn read_znippy_full_manifest(path: &Path) -> Result<(Vec<ManifestEntry>, u64)> {
855 let mut file = File::open(path)?;
856 let file_len = file.metadata()?.len();
857 read_full_manifest(&mut file, file_len)
858}
859
860fn read_full_manifest(file: &mut File, file_len: u64) -> Result<(Vec<ManifestEntry>, u64)> {
862 anyhow::ensure!(file_len >= 16, "file too small to be a v0.7 archive");
863 file.seek(SeekFrom::End(-16))?;
864 let mut tail = [0u8; 16];
865 file.read_exact(&mut tail)?;
866 let manifest_offset = match interpret_footer(&tail) {
867 IndexFooter::Multi { manifest_offset } => manifest_offset,
868 IndexFooter::Single { .. } => anyhow::bail!("not a v0.7 multi-index archive"),
869 };
870 let manifest_end = file_len.checked_sub(16)
871 .ok_or_else(|| anyhow::anyhow!("v0.7 archive too small"))?;
872 anyhow::ensure!(manifest_offset <= manifest_end, "corrupt manifest_offset");
873 let manifest_len = (manifest_end - manifest_offset) as usize;
874 file.seek(SeekFrom::Start(manifest_offset))?;
875 let mut manifest_bytes = vec![0u8; manifest_len];
876 file.read_exact(&mut manifest_bytes)?;
877 Ok((read_manifest_bytes(&manifest_bytes)?, manifest_offset))
878}
879
880fn read_reserved_section(path: &Path, module_name: &str) -> Result<Option<Vec<u8>>> {
882 let mut file = File::open(path)?;
883 let file_len = file.metadata()?.len();
884 let (entries, _) = read_full_manifest(&mut file, file_len)?;
885 let Some(entry) = entries.iter().find(|e| e.module_name == module_name) else {
886 return Ok(None);
887 };
888 anyhow::ensure!(
891 entry.index_offset.checked_add(entry.index_len)
892 .is_some_and(|end| end <= file_len),
893 "reserved section {} out of bounds (offset={}, len={}, file_len={})",
894 entry.module_name, entry.index_offset, entry.index_len, file_len
895 );
896 file.seek(SeekFrom::Start(entry.index_offset))?;
897 let mut bytes = vec![0u8; entry.index_len as usize];
898 file.read_exact(&mut bytes)?;
899 Ok(Some(bytes))
900}
901
902fn checksum32(col: &arrow::array::FixedSizeBinaryArray, i: usize) -> Result<[u8; 32]> {
915 if col.value_length() != 32 {
916 return Err(anyhow::anyhow!(
917 "checksum column has width {}, expected 32",
918 col.value_length()
919 ));
920 }
921 let mut ck = [0u8; 32];
922 ck.copy_from_slice(col.value(i));
923 Ok(ck)
924}
925
926fn decode_lookup(bytes: &[u8]) -> Result<LookupColumns> {
927 use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
928 use arrow::ipc::reader::StreamReader;
929
930 let reader = StreamReader::try_new(std::io::Cursor::new(bytes), None)?;
931 let mut cols = LookupColumns::default();
932 for batch in reader {
933 let batch = batch?;
934 let get = |n: &str| batch.column_by_name(n)
935 .ok_or_else(|| anyhow::anyhow!("lookup missing column {n}"));
936 let paths = get("relative_path")?.as_any().downcast_ref::<StringArray>()
937 .ok_or_else(|| anyhow::anyhow!("relative_path type"))?;
938 let chunk_seq = get("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
939 .ok_or_else(|| anyhow::anyhow!("chunk_seq type"))?;
940 let fdata = get("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
941 .ok_or_else(|| anyhow::anyhow!("fdata_offset type"))?;
942 let compressed = get("compressed")?.as_any().downcast_ref::<BooleanArray>()
943 .ok_or_else(|| anyhow::anyhow!("compressed type"))?;
944 let usize_col = get("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
945 .ok_or_else(|| anyhow::anyhow!("uncompressed_size type"))?;
946 let blob_offset = get("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
947 .ok_or_else(|| anyhow::anyhow!("blob_offset type"))?;
948 let blob_size = get("blob_size")?.as_any().downcast_ref::<UInt64Array>()
949 .ok_or_else(|| anyhow::anyhow!("blob_size type"))?;
950 let checksum = get("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
951 .ok_or_else(|| anyhow::anyhow!("checksum type"))?;
952 for i in 0..batch.num_rows() {
953 cols.paths.push(paths.value(i).to_string());
954 let ck = checksum32(checksum, i)?;
955 cols.locs.push(ChunkLoc {
956 chunk_seq: chunk_seq.value(i),
957 fdata_offset: fdata.value(i),
958 blob_offset: blob_offset.value(i),
959 blob_size: blob_size.value(i),
960 uncompressed_size: usize_col.value(i),
961 compressed: compressed.value(i),
962 checksum: ck,
963 });
964 }
965 }
966 Ok(cols)
967}
968
969#[derive(Default)]
970struct LookupColumns {
971 paths: Vec<String>,
972 locs: Vec<ChunkLoc>,
973}
974
975pub fn locate_file(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
982 if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
983 let cols = decode_lookup(&lookup_bytes)?;
984 let n = cols.paths.len();
985
986 let hit = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
988 let map = fst::Map::new(trie_bytes)
989 .map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
990 map.get(target.as_bytes()).map(|v| v as usize)
991 } else {
992 match cols.paths.binary_search_by(|p| p.as_str().cmp(target)) {
994 Ok(i) => Some(i),
995 Err(_) => None,
996 }
997 };
998
999 let Some(hit) = hit else { return Ok(Vec::new()); };
1000
1001 if hit >= n { return Ok(Vec::new()); }
1006
1007 let mut start = hit;
1009 while start > 0 && cols.paths[start - 1] == target { start -= 1; }
1010 let mut end = hit + 1;
1011 while end < n && cols.paths[end] == target { end += 1; }
1012
1013 let mut out: Vec<ChunkLoc> = cols.locs[start..end].to_vec();
1014 out.sort_by_key(|c| c.chunk_seq);
1015 return Ok(out);
1016 }
1017
1018 locate_file_via_index(path, target)
1020}
1021
1022#[derive(Debug, Clone, PartialEq)]
1025pub struct ArtifactMeta {
1026 pub relative_path: String,
1027 pub uncompressed_size: u64,
1029 pub chunk_count: u32,
1030 pub compressed: bool,
1032}
1033
1034pub fn get_all_files_meta(path: &Path) -> Result<Vec<ArtifactMeta>> {
1039 files_meta_impl(path, None)
1040}
1041
1042pub fn get_files_meta_with_prefix(path: &Path, prefix: &str) -> Result<Vec<ArtifactMeta>> {
1049 files_meta_impl(path, Some(prefix))
1050}
1051
1052fn files_meta_impl(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
1053 use fst::{IntoStreamer, Streamer};
1054
1055 if let Some(lookup_bytes) = read_reserved_section(path, LOOKUP_MODULE)? {
1056 let cols = decode_lookup(&lookup_bytes)?; let n = cols.paths.len();
1058
1059 let (lo, hi) = match prefix {
1061 None | Some("") => (0, n),
1062 Some(pre) => {
1063 let lo = if let Some(trie_bytes) = read_reserved_section(path, TRIE_MODULE)? {
1066 let map = fst::Map::new(trie_bytes)
1067 .map_err(|e| anyhow::anyhow!("trie open: {e}"))?;
1068 let mut stream = map.range().ge(pre.as_bytes()).into_stream();
1069 match stream.next() {
1070 Some((k, v)) if k.starts_with(pre.as_bytes()) => v as usize,
1071 _ => return Ok(Vec::new()),
1072 }
1073 } else {
1074 cols.paths.partition_point(|p| p.as_str() < pre)
1075 };
1076 if lo >= n || !cols.paths[lo].starts_with(pre) {
1077 return Ok(Vec::new());
1078 }
1079 let mut hi = lo;
1080 while hi < n && cols.paths[hi].starts_with(pre) {
1081 hi += 1;
1082 }
1083 (lo, hi)
1084 }
1085 };
1086
1087 let mut out = Vec::new();
1089 let mut i = lo;
1090 while i < hi {
1091 let p = &cols.paths[i];
1092 let mut total = 0u64;
1093 let mut count = 0u32;
1094 let mut compressed = false;
1095 while i < hi && &cols.paths[i] == p {
1096 total += cols.locs[i].uncompressed_size;
1097 count += 1;
1098 compressed |= cols.locs[i].compressed;
1099 i += 1;
1100 }
1101 out.push(ArtifactMeta {
1102 relative_path: p.clone(),
1103 uncompressed_size: total,
1104 chunk_count: count,
1105 compressed,
1106 });
1107 }
1108 return Ok(out);
1109 }
1110
1111 files_meta_via_index(path, prefix)
1112}
1113
1114pub struct ArchiveReader {
1133 paths: Vec<String>,
1135 locs: Vec<ChunkLoc>,
1137 trie: Option<fst::Map<Vec<u8>>>,
1141 file: File,
1145 file_len: u64,
1148}
1149
1150impl ArchiveReader {
1151 pub fn open(path: &Path) -> Result<Self> {
1154 let mut file = File::open(path)?;
1155 let file_len = file.metadata()?.len();
1156 let (entries, _) = read_full_manifest(&mut file, file_len)?;
1157
1158 let mut read_section = |module: &str| -> Result<Option<Vec<u8>>> {
1162 let Some(entry) = entries.iter().find(|e| e.module_name == module) else {
1163 return Ok(None);
1164 };
1165 anyhow::ensure!(
1166 entry.index_offset.checked_add(entry.index_len)
1167 .is_some_and(|end| end <= file_len),
1168 "reserved section {} out of bounds (offset={}, len={}, file_len={})",
1169 entry.module_name, entry.index_offset, entry.index_len, file_len
1170 );
1171 file.seek(SeekFrom::Start(entry.index_offset))?;
1172 let mut bytes = vec![0u8; entry.index_len as usize];
1173 file.read_exact(&mut bytes)?;
1174 Ok(Some(bytes))
1175 };
1176
1177 let lookup_bytes = read_section(LOOKUP_MODULE)?.ok_or_else(|| {
1178 anyhow::anyhow!("archive has no lookup sub-index (not a v0.7 sealed archive)")
1179 })?;
1180 let cols = decode_lookup(&lookup_bytes)?;
1181 let trie = match read_section(TRIE_MODULE)? {
1182 Some(tb) => Some(fst::Map::new(tb).map_err(|e| anyhow::anyhow!("trie open: {e}"))?),
1183 None => None,
1184 };
1185
1186 Ok(Self { paths: cols.paths, locs: cols.locs, trie, file, file_len })
1187 }
1188
1189 pub fn row_count(&self) -> usize {
1191 self.paths.len()
1192 }
1193
1194 pub fn read_file(&self, target: &str) -> Result<Vec<u8>> {
1204 let chunks = self.locate(target);
1205 anyhow::ensure!(!chunks.is_empty(), "file not found in archive: {target}");
1206 crate::decompress::reassemble_file(&self.file, self.file_len, target, &chunks)
1207 }
1208
1209 pub fn locate(&self, target: &str) -> Vec<ChunkLoc> {
1212 let n = self.paths.len();
1213 let hit = match &self.trie {
1214 Some(map) => map.get(target.as_bytes()).map(|v| v as usize),
1215 None => self.paths.binary_search_by(|p| p.as_str().cmp(target)).ok(),
1216 };
1217 let Some(hit) = hit else { return Vec::new(); };
1218
1219 if hit >= n { return Vec::new(); }
1223
1224 let mut start = hit;
1226 while start > 0 && self.paths[start - 1] == target { start -= 1; }
1227 let mut end = hit + 1;
1228 while end < n && self.paths[end] == target { end += 1; }
1229
1230 let mut out: Vec<ChunkLoc> = self.locs[start..end].to_vec();
1231 out.sort_by_key(|c| c.chunk_seq);
1232 out
1233 }
1234
1235 pub fn files_meta(&self) -> Vec<ArtifactMeta> {
1238 self.aggregate(0, self.paths.len())
1239 }
1240
1241 pub fn files_meta_with_prefix(&self, prefix: &str) -> Vec<ArtifactMeta> {
1244 let (lo, hi) = self.window(prefix);
1245 self.aggregate(lo, hi)
1246 }
1247
1248 fn window(&self, prefix: &str) -> (usize, usize) {
1252 use fst::{IntoStreamer, Streamer};
1253 let n = self.paths.len();
1254 if prefix.is_empty() {
1255 return (0, n);
1256 }
1257 let lo = match &self.trie {
1258 Some(map) => {
1259 let mut stream = map.range().ge(prefix.as_bytes()).into_stream();
1260 match stream.next() {
1261 Some((k, v)) if k.starts_with(prefix.as_bytes()) => v as usize,
1262 _ => return (0, 0),
1263 }
1264 }
1265 None => self.paths.partition_point(|p| p.as_str() < prefix),
1266 };
1267 if lo >= n || !self.paths[lo].starts_with(prefix) {
1268 return (0, 0);
1269 }
1270 let mut hi = lo;
1271 while hi < n && self.paths[hi].starts_with(prefix) {
1272 hi += 1;
1273 }
1274 (lo, hi)
1275 }
1276
1277 fn aggregate(&self, lo: usize, hi: usize) -> Vec<ArtifactMeta> {
1280 let mut out = Vec::new();
1281 let mut i = lo;
1282 while i < hi {
1283 let p = &self.paths[i];
1284 let mut total = 0u64;
1285 let mut count = 0u32;
1286 let mut compressed = false;
1287 while i < hi && &self.paths[i] == p {
1288 total += self.locs[i].uncompressed_size;
1289 count += 1;
1290 compressed |= self.locs[i].compressed;
1291 i += 1;
1292 }
1293 out.push(ArtifactMeta {
1294 relative_path: p.clone(),
1295 uncompressed_size: total,
1296 chunk_count: count,
1297 compressed,
1298 });
1299 }
1300 out
1301 }
1302}
1303
1304fn files_meta_via_index(path: &Path, prefix: Option<&str>) -> Result<Vec<ArtifactMeta>> {
1306 use arrow::array::{BooleanArray, StringArray, UInt64Array};
1307
1308 let (schema, batches) = read_znippy_index(path)?;
1309 let batch = match batches.len() {
1310 0 => return Ok(Vec::new()),
1311 1 => batches.into_iter().next().unwrap(),
1312 _ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
1313 };
1314 let col = |n: &str| batch.column_by_name(n)
1319 .ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
1320 let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
1321 .ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
1322 let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
1323 .ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
1324 let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
1325 .ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;
1326
1327 let mut agg: HashMap<&str, (u64, u32, bool)> = HashMap::new();
1329 for i in 0..batch.num_rows() {
1330 let p = paths.value(i);
1331 if let Some(pre) = prefix {
1332 if !p.starts_with(pre) { continue; }
1333 }
1334 let e = agg.entry(p).or_insert((0, 0, false));
1335 e.0 += usize_col.value(i);
1336 e.1 += 1;
1337 e.2 |= compressed.value(i);
1338 }
1339 let mut out: Vec<ArtifactMeta> = agg.into_iter().map(|(p, (sz, c, comp))| ArtifactMeta {
1340 relative_path: p.to_string(),
1341 uncompressed_size: sz,
1342 chunk_count: c,
1343 compressed: comp,
1344 }).collect();
1345 out.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
1346 Ok(out)
1347}
1348
1349fn locate_file_via_index(path: &Path, target: &str) -> Result<Vec<ChunkLoc>> {
1351 use arrow::array::{BooleanArray, FixedSizeBinaryArray, StringArray, UInt32Array, UInt64Array};
1352
1353 let (schema, batches) = read_znippy_index(path)?;
1354 let batch = match batches.len() {
1355 0 => return Ok(Vec::new()),
1356 1 => batches.into_iter().next().unwrap(),
1357 _ => arrow_select::concat::concat_batches(&schema, batches.iter())?,
1358 };
1359 let col = |n: &str| batch.column_by_name(n)
1362 .ok_or_else(|| anyhow::anyhow!("index missing column {n}"));
1363 let paths = col("relative_path")?.as_any().downcast_ref::<StringArray>()
1364 .ok_or_else(|| anyhow::anyhow!("index column relative_path has unexpected type"))?;
1365 let chunk_seq = col("chunk_seq")?.as_any().downcast_ref::<UInt32Array>()
1366 .ok_or_else(|| anyhow::anyhow!("index column chunk_seq has unexpected type"))?;
1367 let fdata = col("fdata_offset")?.as_any().downcast_ref::<UInt64Array>()
1368 .ok_or_else(|| anyhow::anyhow!("index column fdata_offset has unexpected type"))?;
1369 let compressed = col("compressed")?.as_any().downcast_ref::<BooleanArray>()
1370 .ok_or_else(|| anyhow::anyhow!("index column compressed has unexpected type"))?;
1371 let usize_col = col("uncompressed_size")?.as_any().downcast_ref::<UInt64Array>()
1372 .ok_or_else(|| anyhow::anyhow!("index column uncompressed_size has unexpected type"))?;
1373 let blob_offset = col("blob_offset")?.as_any().downcast_ref::<UInt64Array>()
1374 .ok_or_else(|| anyhow::anyhow!("index column blob_offset has unexpected type"))?;
1375 let blob_size = col("blob_size")?.as_any().downcast_ref::<UInt64Array>()
1376 .ok_or_else(|| anyhow::anyhow!("index column blob_size has unexpected type"))?;
1377 let checksum = col("checksum")?.as_any().downcast_ref::<FixedSizeBinaryArray>()
1378 .ok_or_else(|| anyhow::anyhow!("index column checksum has unexpected type"))?;
1379
1380 let mut out = Vec::new();
1381 for i in 0..batch.num_rows() {
1382 if paths.value(i) == target {
1383 let ck = checksum32(checksum, i)?;
1384 out.push(ChunkLoc {
1385 chunk_seq: chunk_seq.value(i),
1386 fdata_offset: fdata.value(i),
1387 blob_offset: blob_offset.value(i),
1388 blob_size: blob_size.value(i),
1389 uncompressed_size: usize_col.value(i),
1390 compressed: compressed.value(i),
1391 checksum: ck,
1392 });
1393 }
1394 }
1395 out.sort_by_key(|c| c.chunk_seq);
1396 Ok(out)
1397}
1398
1399pub fn is_probably_compressed(path: &Path) -> bool {
1400 if let Some(ext) = path.extension().and_then(|e| e.to_str()) {
1401 let ext = ext.to_ascii_lowercase();
1402 matches!(
1403 ext.as_str(),
1404 "zip" | "gz" | "bz2" | "xz" | "lz" | "lzma" | "7z" | "rar" | "cab"
1405 | "jar" | "war" | "ear" | "zst" | "sz" | "lz4" | "tgz" | "txz"
1406 | "tbz" | "apk" | "dmg" | "deb" | "rpm" | "arrow" | "mpeg" | "mpg"
1407 | "jpeg" | "jpg" | "gif" | "bmp" | "png" | "crate" | "znippy"
1408 | "zdata" | "parquet" | "webp" | "webm"
1409 | "age" | "iso" | "pdf"
1414 | "pack" | "idx" | "midx" | "bitmap"
1421 )
1422 } else {
1423 false
1424 }
1425}
1426
1427pub fn should_skip_compression(path: &Path) -> bool {
1436 is_probably_compressed(path)
1437}
1438
1439#[cfg(test)]
1440mod skip_compression_tests {
1441 use super::*;
1442
1443 #[test]
1444 fn skips_already_compressed_and_encrypted() {
1445 for p in ["secrets/secrets.age", "infra/talos/talos-1.13.iso", "s3/doc.pdf", "x.PDF"] {
1446 assert!(should_skip_compression(Path::new(p)), "should skip {p}");
1447 }
1448 }
1449
1450 #[test]
1451 fn compresses_plain_dump_and_config() {
1452 for p in ["dbdump/njord.dump", "njord.sql", "lakespec.json", "njordconf/serverspec.json"] {
1454 assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
1455 }
1456 }
1457
1458 #[test]
1461 fn skips_git_pack_directory_artefacts() {
1462 for p in [
1463 ".git/objects/pack/pack-9f2c.pack",
1464 ".git/objects/pack/pack-9f2c.idx",
1465 ".git/objects/pack/multi-pack-index.midx",
1466 ".git/objects/pack/pack-9f2c.bitmap",
1467 ] {
1468 assert!(should_skip_compression(Path::new(p)), "should skip {p}");
1469 }
1470 }
1471
1472 #[test]
1476 fn still_compresses_the_compressible_git_artefacts() {
1477 for p in [
1478 ".git/objects/pack/pack-9f2c.rev",
1479 ".git/objects/pack/pack-9f2c.promisor",
1480 ".git/COMMIT_EDITMSG",
1481 ".git/config",
1482 ] {
1483 assert!(!should_skip_compression(Path::new(p)), "should compress {p}");
1484 }
1485 }
1486}
1487
1488#[derive(Debug, Default)]
1489pub struct VerifyReport {
1490 pub total_files: usize,
1491 pub verified_files: usize,
1492 pub corrupt_files: usize,
1493 pub total_bytes: u64,
1494 pub verified_bytes: u64,
1495 pub corrupt_bytes: u64,
1496 pub chunks: u64,
1497}
1498
1499pub fn list_archive_contents(path: &Path) -> Result<()> {
1500 let (_schema, batches) = read_znippy_index(path)?;
1501 for batch in &batches {
1502 let paths = batch
1503 .column_by_name("relative_path")
1504 .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>())
1505 .ok_or_else(|| anyhow::anyhow!("missing relative_path column"))?;
1506 let sizes = batch
1507 .column_by_name("uncompressed_size")
1508 .and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt64Array>())
1509 .ok_or_else(|| anyhow::anyhow!("missing uncompressed_size column"))?;
1510 let chunk_seqs = batch
1511 .column_by_name("chunk_seq")
1512 .and_then(|c| c.as_any().downcast_ref::<arrow::array::UInt32Array>());
1513 let group_ids = batch
1514 .column_by_name("group_id")
1515 .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
1516 let artifact_ids = batch
1517 .column_by_name("artifact_id")
1518 .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
1519 let versions = batch
1520 .column_by_name("version")
1521 .and_then(|c| c.as_any().downcast_ref::<arrow::array::StringArray>());
1522 for i in 0..batch.num_rows() {
1523 if let Some(seqs) = chunk_seqs {
1525 if seqs.value(i) != 0 {
1526 continue;
1527 }
1528 }
1529 if let (Some(g), Some(a), Some(v)) = (group_ids, artifact_ids, versions) {
1530 if !g.is_null(i) {
1531 println!(
1532 "{}\t{}\t{}:{}:{}",
1533 paths.value(i),
1534 sizes.value(i),
1535 g.value(i),
1536 a.value(i),
1537 v.value(i)
1538 );
1539 continue;
1540 }
1541 }
1542 println!("{}\t{}", paths.value(i), sizes.value(i));
1543 }
1544 }
1545 Ok(())
1546}
1547
1548pub fn verify_archive_integrity(path: &Path) -> Result<VerifyReport> {
1549 let out_dir = PathBuf::from("/dev/null");
1550 decompress_archive(path, false, &out_dir)
1551}
1552
1553#[cfg(test)]
1554mod version_tests {
1555 use super::*;
1556
1557 #[test]
1558 fn current_and_older_versions_are_accepted() {
1559 let mut m = HashMap::new();
1560 check_format_version(&m).unwrap();
1562 m.insert(FORMAT_VERSION_KEY.into(), ZNIPPY_FORMAT_VERSION.to_string());
1564 check_format_version(&m).unwrap();
1565 m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION - 1).to_string());
1567 check_format_version(&m).unwrap();
1568 }
1569
1570 #[test]
1571 fn newer_version_is_rejected_clearly() {
1572 let mut m = HashMap::new();
1573 m.insert(FORMAT_VERSION_KEY.into(), (ZNIPPY_FORMAT_VERSION + 1).to_string());
1574 let err = check_format_version(&m).unwrap_err().to_string();
1575 assert!(err.contains("newer than this reader supports"), "got: {err}");
1576 assert!(err.contains("upgrade znippy"), "got: {err}");
1577 }
1578
1579 #[test]
1580 fn writer_records_the_current_version() {
1581 let meta = build_arrow_metadata_for_config(&crate::common_config::CONFIG);
1582 assert_eq!(
1583 meta.get(FORMAT_VERSION_KEY).map(String::as_str),
1584 Some(ZNIPPY_FORMAT_VERSION.to_string().as_str())
1585 );
1586 }
1587}
1588
1589#[cfg(test)]
1590mod checksum_width_tests {
1591 use super::*;
1592 use arrow::array::FixedSizeBinaryArray;
1593
1594 #[test]
1601 fn narrow_checksum_column_errors_instead_of_panicking() {
1602 let narrow = FixedSizeBinaryArray::try_from_iter([[0u8; 16]].into_iter()).unwrap();
1603 assert_eq!(narrow.value_length(), 16);
1604 let err = checksum32(&narrow, 0).expect_err("a 16-byte checksum column must be an Err");
1605 assert!(
1606 err.to_string().contains("width 16"),
1607 "error must name the bad width, got: {err}"
1608 );
1609 }
1610
1611 #[test]
1612 fn wide_checksum_column_errors_instead_of_truncating() {
1613 let wide = FixedSizeBinaryArray::try_from_iter([[7u8; 64]].into_iter()).unwrap();
1614 assert!(checksum32(&wide, 0).is_err(), "a 64-byte checksum column must be an Err");
1615 }
1616
1617 #[test]
1618 fn correct_width_checksum_is_read_verbatim() {
1619 let mut digest = [0u8; 32];
1620 for (i, b) in digest.iter_mut().enumerate() {
1621 *b = i as u8;
1622 }
1623 let col = FixedSizeBinaryArray::try_from_iter([digest].into_iter()).unwrap();
1624 assert_eq!(checksum32(&col, 0).unwrap(), digest);
1625 }
1626}
1627
1628#[cfg(test)]
1629mod reserved_module_tests {
1630 use super::*;
1631
1632 #[test]
1642 fn every_reserved_module_is_classified_reserved() {
1643 for m in RESERVED_MODULES {
1644 assert!(
1645 is_reserved_module(m),
1646 "'{m}' is in RESERVED_MODULES but is_reserved_module says it is DATA — \
1647 its rows would be merged into the file index"
1648 );
1649 }
1650 }
1651
1652 #[test]
1656 fn the_git_format_modules_are_reserved() {
1657 for m in [
1658 GUNNAR_OID_MODULE,
1659 GUNNAR_GRAPH_MODULE,
1660 GUNNAR_REACH_MODULE,
1661 GUNNAR_REFS_MODULE,
1662 GUNNAR_SECRETS_MODULE,
1663 ] {
1664 assert!(is_reserved_module(m), "git package-format module '{m}' must be reserved");
1665 assert!(
1666 RESERVED_MODULES.contains(&m),
1667 "git package-format module '{m}' must be in the catalog"
1668 );
1669 }
1670 assert_eq!(GUNNAR_OID_MODULE, "__gunnar_oid__");
1671 assert_eq!(GUNNAR_GRAPH_MODULE, "__gunnar_graph__");
1672 assert_eq!(GUNNAR_REACH_MODULE, "__gunnar_reach__");
1673 assert_eq!(GUNNAR_REFS_MODULE, "__gunnar_refs__");
1674 assert_eq!(GUNNAR_SECRETS_MODULE, "__gunnar_secrets__");
1675 }
1676
1677 #[test]
1680 fn ordinary_module_names_stay_data() {
1681 for m in ["maven", "git", "rust", "", "__gunnar__", "gunnar_oid", "__gunnar_oid", "objects"] {
1682 assert!(!is_reserved_module(m), "'{m}' must be treated as a DATA sub-index");
1683 }
1684 }
1685
1686 #[test]
1688 fn the_catalog_is_well_formed() {
1689 let mut seen = std::collections::HashSet::new();
1690 for m in RESERVED_MODULES {
1691 assert!(!m.is_empty(), "empty reserved module name");
1692 assert!(seen.insert(*m), "duplicate reserved module name '{m}'");
1693 }
1694 assert_eq!(seen.len(), RESERVED_MODULES.len());
1695 }
1696}