use std::sync::Arc;
use anyhow::{Result, anyhow, bail};
use znippy_common::arrow::array::{
Array, FixedSizeBinaryArray, UInt8Array, UInt64Array,
};
use znippy_common::arrow::buffer::{Buffer, MutableBuffer, ScalarBuffer};
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::ipc::reader::StreamDecoder;
use znippy_common::arrow::ipc::writer::StreamWriter;
use znippy_common::arrow::record_batch::RecordBatch;
use crate::object::GitHashKind;
use crate::oid_index::{GitOidIndex, OidEntry, OidLayout, build_section_with_layout};
pub use git_storage_trait::ObjType;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct IndexEntry {
pub oid: Vec<u8>,
pub offset: u64,
pub len: u64,
pub obj_type: ObjType,
pub uncompressed_size: u64,
pub delta_base: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct IndexRow {
pub ordinal: u32,
pub offset: u64,
pub len: u64,
pub obj_type: ObjType,
pub uncompressed_size: u64,
pub delta_base: u64,
}
pub trait ObjectIndex: Send + Sync {
fn build(entries: &[IndexEntry]) -> Result<Self>
where
Self: Sized;
fn lookup(&self, oid: &[u8]) -> Option<IndexRow>;
fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>>;
fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>>;
fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>>;
fn sum_uncompressed(&self) -> u64;
fn count_type(&self, t: ObjType) -> usize;
fn name(&self) -> &'static str;
fn len(&self) -> usize;
fn is_empty(&self) -> bool {
self.len() == 0
}
fn ipc_bytes(&self) -> usize;
fn resident_bytes(&self) -> usize;
}
pub const COL_OID: &str = "oid";
pub const COL_OFFSET: &str = "offset";
pub const COL_LEN: &str = "len";
pub const COL_TYPE: &str = "object_type";
pub const COL_SIZE: &str = "uncompressed_size";
pub const COL_DELTA_BASE: &str = "delta_base";
pub struct Columns {
pub oid: FixedSizeBinaryArray,
pub offset: UInt64Array,
pub len: UInt64Array,
pub obj_type: UInt8Array,
pub size: UInt64Array,
pub delta_base: UInt64Array,
}
fn oid_order(entries: &[IndexEntry]) -> Vec<u32> {
let mut order: Vec<u32> = (0..entries.len() as u32).collect();
order.sort_unstable_by(|&a, &b| entries[a as usize].oid.cmp(&entries[b as usize].oid));
order
}
fn validate(entries: &[IndexEntry], order: &[u32]) -> Result<GitHashKind> {
let Some(first) = entries.first() else {
return Ok(GitHashKind::Sha1);
};
let hash = match first.oid.len() {
20 => GitHashKind::Sha1,
32 => GitHashKind::Sha256,
n => bail!("oid width {n} is neither sha1 (20) nor sha256 (32)"),
};
for e in entries {
if e.oid.len() != hash.oid_len() {
bail!(
"mixed oid widths: {} and {} in one index",
hash.oid_len(),
e.oid.len()
);
}
}
for w in order.windows(2) {
if entries[w[0] as usize].oid == entries[w[1] as usize].oid {
bail!(
"duplicate oid {} — one object cannot hold two ordinals",
hex::encode(&entries[w[0] as usize].oid)
);
}
}
Ok(hash)
}
pub struct OidResolver {
tree: GitOidIndex,
section_bytes: usize,
}
impl OidResolver {
fn build(entries: &[IndexEntry], order: &[u32], hash: GitHashKind) -> Result<Self> {
Self::build_with(entries, order, hash, OidLayout::default())
}
fn build_with(
entries: &[IndexEntry],
order: &[u32],
hash: GitHashKind,
layout: OidLayout,
) -> Result<Self> {
let oid_entries: Vec<OidEntry> = order
.iter()
.enumerate()
.map(|(rank, &i)| OidEntry {
oid: entries[i as usize].oid.clone(),
lookup_row: rank as u64,
ordinal: rank as u32,
})
.collect();
let section = build_section_with_layout(&oid_entries, hash, layout)?;
let section_bytes = section.len();
Ok(Self { tree: GitOidIndex::parse_as(section, layout)?, section_bytes })
}
#[inline]
fn ordinal(&self, oid: &[u8]) -> Option<u32> {
self.tree.lookup(oid).map(|h| h.ordinal)
}
#[inline]
fn ordinals(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
self.tree
.lookup_batch(oids)
.into_iter()
.map(|h| h.map(|h| h.ordinal))
.collect()
}
}
pub fn columns(entries: &[IndexEntry], order: &[u32], oid_len: usize) -> Columns {
let n = order.len();
let mut offset: Vec<u64> = Vec::with_capacity(n);
let mut len: Vec<u64> = Vec::with_capacity(n);
let mut ty: Vec<u8> = Vec::with_capacity(n);
let mut size: Vec<u64> = Vec::with_capacity(n);
let mut delta_base: Vec<u64> = Vec::with_capacity(n);
for &i in order {
let e = &entries[i as usize];
offset.push(e.offset);
len.push(e.len);
ty.push(e.obj_type.code());
size.push(e.uncompressed_size);
delta_base.push(e.delta_base);
}
Columns {
oid: oid_column(entries, order, oid_len),
offset: UInt64Array::new(ScalarBuffer::from(offset), None),
len: UInt64Array::new(ScalarBuffer::from(len), None),
obj_type: UInt8Array::new(ScalarBuffer::from(ty), None),
size: UInt64Array::new(ScalarBuffer::from(size), None),
delta_base: UInt64Array::new(ScalarBuffer::from(delta_base), None),
}
}
pub fn oid_column(entries: &[IndexEntry], order: &[u32], oid_len: usize) -> FixedSizeBinaryArray {
let mut bytes: Vec<u8> = Vec::with_capacity(order.len() * oid_len);
for &i in order {
bytes.extend_from_slice(&entries[i as usize].oid);
}
FixedSizeBinaryArray::new(oid_len as i32, Buffer::from_vec(bytes), None)
}
pub const PACKED_LEN: usize = 33;
const P_OFFSET: usize = 0;
const P_LEN: usize = 8;
const P_TYPE: usize = 16;
const P_SIZE: usize = 17;
const P_DELTA_BASE: usize = 25;
pub fn packed_column(entries: &[IndexEntry], order: &[u32]) -> FixedSizeBinaryArray {
let mut bytes: Vec<u8> = Vec::with_capacity(order.len() * PACKED_LEN);
for &i in order {
let e = &entries[i as usize];
bytes.extend_from_slice(&e.offset.to_le_bytes());
bytes.extend_from_slice(&e.len.to_le_bytes());
bytes.push(e.obj_type.code());
bytes.extend_from_slice(&e.uncompressed_size.to_le_bytes());
bytes.extend_from_slice(&e.delta_base.to_le_bytes());
}
debug_assert_eq!(bytes.len(), order.len() * PACKED_LEN);
FixedSizeBinaryArray::new(PACKED_LEN as i32, Buffer::from_vec(bytes), None)
}
#[inline]
fn le64(r: &[u8], at: usize) -> u64 {
u64::from_le_bytes(r[at..at + 8].try_into().unwrap())
}
fn oid_field(oid_len: usize) -> Field {
Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false)
}
fn to_ipc(batch: &RecordBatch) -> Result<Vec<u8>> {
let mut out = Vec::with_capacity(64 + batch.get_array_memory_size());
{
let mut w = StreamWriter::try_new(&mut out, batch.schema_ref())
.map_err(|e| anyhow!("ipc writer: {e}"))?;
w.write(batch).map_err(|e| anyhow!("ipc write: {e}"))?;
w.finish().map_err(|e| anyhow!("ipc finish: {e}"))?;
}
Ok(out)
}
fn decode_ipc(bytes: &[u8]) -> Result<(Buffer, RecordBatch)> {
let mut mb = MutableBuffer::with_capacity(bytes.len());
mb.extend_from_slice(bytes);
decode_ipc_buffer(mb.into())
}
fn decode_ipc_buffer(owner: Buffer) -> Result<(Buffer, RecordBatch)> {
let mut cursor = owner.clone();
let mut decoder = StreamDecoder::new().with_require_alignment(true);
let mut found: Option<RecordBatch> = None;
while !cursor.is_empty() {
match decoder.decode(&mut cursor).map_err(|e| anyhow!("ipc decode: {e}"))? {
Some(b) if found.is_none() => found = Some(b),
Some(_) => bail!("index section holds more than one record batch"),
None => {}
}
}
decoder.finish().map_err(|e| anyhow!("ipc unfinished: {e}"))?;
found.ok_or_else(|| anyhow!("index section holds no record batch"))
.map(|b| (owner, b))
}
fn col<T: Array + Clone + 'static>(batch: &RecordBatch, name: &str) -> Result<T> {
batch
.column_by_name(name)
.ok_or_else(|| anyhow!("no `{name}` column"))?
.as_any()
.downcast_ref::<T>()
.cloned()
.ok_or_else(|| anyhow!("`{name}` has an unexpected type"))
}
fn inside(ipc: &Buffer, values: &[u8]) -> bool {
if values.is_empty() {
return true;
}
let base = ipc.as_ptr() as usize;
let p = values.as_ptr() as usize;
p >= base && p + values.len() <= base + ipc.len()
}
struct ColumnarPayload {
oid: FixedSizeBinaryArray,
offset: UInt64Array,
len: UInt64Array,
obj_type: UInt8Array,
size: UInt64Array,
delta_base: UInt64Array,
}
impl ColumnarPayload {
#[inline]
fn row_at(&self, ordinal: u32) -> IndexRow {
let i = ordinal as usize;
IndexRow {
ordinal,
offset: self.offset.value(i),
len: self.len.value(i),
obj_type: ObjType::from_code(self.obj_type.value(i)).unwrap_or(ObjType::Blob),
uncompressed_size: self.size.value(i),
delta_base: self.delta_base.value(i),
}
}
#[inline]
fn extent_at(&self, ordinal: u32) -> (u64, u64) {
let i = ordinal as usize;
(self.offset.value(i), self.len.value(i))
}
fn sum_uncompressed(&self) -> u64 {
self.size.values().iter().copied().fold(0u64, u64::wrapping_add)
}
fn count_type(&self, t: ObjType) -> usize {
let c = t.code();
self.obj_type.values().iter().filter(|&&x| x == c).count()
}
fn inside(&self, oid_ipc: &Buffer, extent_ipc: &Buffer, type_ipc: &Buffer, size_ipc: &Buffer) -> bool {
inside(oid_ipc, self.oid.value_data())
&& inside(extent_ipc, self.offset.values().inner().as_slice())
&& inside(extent_ipc, self.len.values().inner().as_slice())
&& inside(extent_ipc, self.delta_base.values().inner().as_slice())
&& inside(type_ipc, self.obj_type.values().inner().as_slice())
&& inside(size_ipc, self.size.values().inner().as_slice())
}
}
fn extent_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new(COL_OFFSET, DataType::UInt64, false),
Field::new(COL_LEN, DataType::UInt64, false),
Field::new(COL_DELTA_BASE, DataType::UInt64, false),
]))
}
pub struct FourTables {
oids: OidResolver,
ipc: [Buffer; 4],
cols: ColumnarPayload,
rows: usize,
}
impl FourTables {
pub fn ipc_section_lens(&self) -> [usize; 4] {
[self.ipc[0].len(), self.ipc[1].len(), self.ipc[2].len(), self.ipc[3].len()]
}
pub fn oid_column(&self) -> &FixedSizeBinaryArray {
&self.cols.oid
}
pub fn column_is_inside_ipc(&self) -> bool {
self.cols.inside(&self.ipc[0], &self.ipc[1], &self.ipc[2], &self.ipc[3])
}
}
impl ObjectIndex for FourTables {
fn build(entries: &[IndexEntry]) -> Result<Self> {
let order = oid_order(entries);
let hash = validate(entries, &order)?;
let oid_len = hash.oid_len();
let c = columns(entries, &order, oid_len);
let oids = OidResolver::build(entries, &order, hash)?;
let s_oid = Arc::new(Schema::new(vec![oid_field(oid_len)]));
let s_type = Arc::new(Schema::new(vec![Field::new(COL_TYPE, DataType::UInt8, false)]));
let s_size = Arc::new(Schema::new(vec![Field::new(COL_SIZE, DataType::UInt64, false)]));
let b_oid = RecordBatch::try_new(s_oid, vec![Arc::new(c.oid)])?;
let b_extent = RecordBatch::try_new(
extent_schema(),
vec![Arc::new(c.offset), Arc::new(c.len), Arc::new(c.delta_base)],
)?;
let b_type = RecordBatch::try_new(s_type, vec![Arc::new(c.obj_type)])?;
let b_size = RecordBatch::try_new(s_size, vec![Arc::new(c.size)])?;
let (ipc_oid, r_oid) = decode_ipc(&to_ipc(&b_oid)?)?;
let (ipc_extent, r_extent) = decode_ipc(&to_ipc(&b_extent)?)?;
let (ipc_type, r_type) = decode_ipc(&to_ipc(&b_type)?)?;
let (ipc_size, r_size) = decode_ipc(&to_ipc(&b_size)?)?;
Ok(Self {
oids,
ipc: [ipc_oid, ipc_extent, ipc_type, ipc_size],
cols: ColumnarPayload {
oid: col(&r_oid, COL_OID)?,
offset: col(&r_extent, COL_OFFSET)?,
len: col(&r_extent, COL_LEN)?,
obj_type: col(&r_type, COL_TYPE)?,
size: col(&r_size, COL_SIZE)?,
delta_base: col(&r_extent, COL_DELTA_BASE)?,
},
rows: order.len(),
})
}
fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
self.oids.ordinal(oid).map(|o| self.cols.row_at(o))
}
fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.row_at(o))).collect()
}
fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
self.oids.ordinals(oids)
}
fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.extent_at(o))).collect()
}
fn sum_uncompressed(&self) -> u64 {
self.cols.sum_uncompressed()
}
fn count_type(&self, t: ObjType) -> usize {
self.cols.count_type(t)
}
fn name(&self) -> &'static str {
"FourTables"
}
fn len(&self) -> usize {
self.rows
}
fn ipc_bytes(&self) -> usize {
self.ipc.iter().map(|b| b.len()).sum()
}
fn resident_bytes(&self) -> usize {
self.ipc_bytes() + self.oids.section_bytes
}
}
pub struct OneTableFourColumns {
oids: OidResolver,
ipc: Buffer,
cols: ColumnarPayload,
rows: usize,
}
impl OneTableFourColumns {
pub fn oid_column(&self) -> &FixedSizeBinaryArray {
&self.cols.oid
}
pub fn column_is_inside_ipc(&self) -> bool {
self.cols.inside(&self.ipc, &self.ipc, &self.ipc, &self.ipc)
}
pub fn ipc_slice(&self) -> &[u8] {
self.ipc.as_slice()
}
}
impl ObjectIndex for OneTableFourColumns {
fn build(entries: &[IndexEntry]) -> Result<Self> {
let order = oid_order(entries);
let hash = validate(entries, &order)?;
let oid_len = hash.oid_len();
let c = columns(entries, &order, oid_len);
let oids = OidResolver::build(entries, &order, hash)?;
let schema: SchemaRef = Arc::new(Schema::new(vec![
oid_field(oid_len),
Field::new(COL_OFFSET, DataType::UInt64, false),
Field::new(COL_LEN, DataType::UInt64, false),
Field::new(COL_TYPE, DataType::UInt8, false),
Field::new(COL_SIZE, DataType::UInt64, false),
Field::new(COL_DELTA_BASE, DataType::UInt64, false),
]));
let batch = RecordBatch::try_new(
schema,
vec![
Arc::new(c.oid),
Arc::new(c.offset),
Arc::new(c.len),
Arc::new(c.obj_type),
Arc::new(c.size),
Arc::new(c.delta_base),
],
)?;
let (ipc, r) = decode_ipc(&to_ipc(&batch)?)?;
Ok(Self {
oids,
ipc,
cols: ColumnarPayload {
oid: col(&r, COL_OID)?,
offset: col(&r, COL_OFFSET)?,
len: col(&r, COL_LEN)?,
obj_type: col(&r, COL_TYPE)?,
size: col(&r, COL_SIZE)?,
delta_base: col(&r, COL_DELTA_BASE)?,
},
rows: order.len(),
})
}
fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
self.oids.ordinal(oid).map(|o| self.cols.row_at(o))
}
fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.row_at(o))).collect()
}
fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
self.oids.ordinals(oids)
}
fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.cols.extent_at(o))).collect()
}
fn sum_uncompressed(&self) -> u64 {
self.cols.sum_uncompressed()
}
fn count_type(&self, t: ObjType) -> usize {
self.cols.count_type(t)
}
fn name(&self) -> &'static str {
"OneTableFourColumns"
}
fn len(&self) -> usize {
self.rows
}
fn ipc_bytes(&self) -> usize {
self.ipc.len()
}
fn resident_bytes(&self) -> usize {
self.ipc_bytes() + self.oids.section_bytes
}
}
pub struct PackedPayload {
oids: OidResolver,
ipc: Buffer,
oid: FixedSizeBinaryArray,
payload: FixedSizeBinaryArray,
rows: usize,
}
pub const COL_PACKED: &str = "packed";
impl PackedPayload {
pub fn build_with_oid_layout(entries: &[IndexEntry], layout: OidLayout) -> Result<Self> {
let order = oid_order(entries);
let hash = validate(entries, &order)?;
let oid_len = hash.oid_len();
let oid_arr = oid_column(entries, &order, oid_len);
let packed = packed_column(entries, &order);
let oids = OidResolver::build_with(entries, &order, hash, layout)?;
Self::assemble(oids, oid_arr, packed, oid_len, order.len())
}
pub fn keyspace_phase(&self) -> usize {
self.oids.tree.keyspace_phase()
}
fn assemble(
oids: OidResolver,
oid_arr: FixedSizeBinaryArray,
packed: FixedSizeBinaryArray,
oid_len: usize,
rows: usize,
) -> Result<Self> {
let schema: SchemaRef = Arc::new(Schema::new(vec![
oid_field(oid_len),
Field::new(COL_PACKED, DataType::FixedSizeBinary(PACKED_LEN as i32), false),
]));
let batch = RecordBatch::try_new(schema, vec![Arc::new(oid_arr), Arc::new(packed)])?;
let (ipc, r) = decode_ipc(&to_ipc(&batch)?)?;
Ok(Self { oids, ipc, oid: col(&r, COL_OID)?, payload: col(&r, COL_PACKED)?, rows })
}
pub fn oid_column(&self) -> &FixedSizeBinaryArray {
&self.oid
}
pub fn column_is_inside_ipc(&self) -> bool {
inside(&self.ipc, self.oid.value_data()) && inside(&self.ipc, self.payload.value_data())
}
#[inline]
fn row_at(&self, ordinal: u32) -> IndexRow {
let r = self.payload.value(ordinal as usize);
IndexRow {
ordinal,
offset: le64(r, P_OFFSET),
len: le64(r, P_LEN),
obj_type: ObjType::from_code(r[P_TYPE]).unwrap_or(ObjType::Blob),
uncompressed_size: le64(r, P_SIZE),
delta_base: le64(r, P_DELTA_BASE),
}
}
#[inline]
fn extent_at(&self, ordinal: u32) -> (u64, u64) {
let r = self.payload.value(ordinal as usize);
(le64(r, P_OFFSET), le64(r, P_LEN))
}
}
impl ObjectIndex for PackedPayload {
fn build(entries: &[IndexEntry]) -> Result<Self> {
let order = oid_order(entries);
let hash = validate(entries, &order)?;
let oid_len = hash.oid_len();
let oid_arr = oid_column(entries, &order, oid_len);
let packed = packed_column(entries, &order);
let oids = OidResolver::build(entries, &order, hash)?;
Self::assemble(oids, oid_arr, packed, oid_len, order.len())
}
fn lookup(&self, oid: &[u8]) -> Option<IndexRow> {
self.oids.ordinal(oid).map(|o| self.row_at(o))
}
fn lookup_batch(&self, oids: &[&[u8]]) -> Vec<Option<IndexRow>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.row_at(o))).collect()
}
fn ordinals_batch(&self, oids: &[&[u8]]) -> Vec<Option<u32>> {
self.oids.ordinals(oids)
}
fn extents_batch(&self, oids: &[&[u8]]) -> Vec<Option<(u64, u64)>> {
self.oids.ordinals(oids).into_iter().map(|o| o.map(|o| self.extent_at(o))).collect()
}
fn sum_uncompressed(&self) -> u64 {
let d = self.payload.value_data();
let mut acc = 0u64;
let mut i = P_SIZE;
while i + 8 <= d.len() {
acc = acc.wrapping_add(le64(d, i));
i += PACKED_LEN;
}
acc
}
fn count_type(&self, t: ObjType) -> usize {
let c = t.code();
let d = self.payload.value_data();
let mut n = 0usize;
let mut i = P_TYPE;
while i < d.len() {
if d[i] == c {
n += 1;
}
i += PACKED_LEN;
}
n
}
fn name(&self) -> &'static str {
"PackedPayload"
}
fn len(&self) -> usize {
self.rows
}
fn ipc_bytes(&self) -> usize {
self.ipc.len()
}
fn resident_bytes(&self) -> usize {
self.ipc_bytes() + self.oids.section_bytes
}
}
pub struct Rng(pub u64);
impl Rng {
pub fn next_u64(&mut self) -> u64 {
self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15);
let mut z = self.0;
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
z ^ (z >> 31)
}
pub fn fill(&mut self, out: &mut [u8]) {
for c in out.chunks_mut(8) {
let w = self.next_u64().to_le_bytes();
let n = c.len();
c.copy_from_slice(&w[..n]);
}
}
}
pub fn synthetic_entries(n: usize, oid_len: usize, seed: u64) -> Vec<IndexEntry> {
let mut rng = Rng(seed);
let mut seen = std::collections::HashSet::with_capacity(n * 2);
let mut out: Vec<IndexEntry> = Vec::with_capacity(n);
let mut off = 12u64; while out.len() < n {
let mut oid = vec![0u8; oid_len];
rng.fill(&mut oid);
if !seen.insert(oid.clone()) {
continue;
}
let len = 32 + (rng.next_u64() % 4096);
let ty = ObjType::ALL[(rng.next_u64() % 6) as usize];
let delta_base = match (ty, out.last()) {
(ObjType::OfsDelta, Some(prev)) => prev.offset,
_ => 0,
};
out.push(IndexEntry {
oid,
offset: off,
len,
obj_type: ty,
uncompressed_size: len * (1 + rng.next_u64() % 5),
delta_base,
});
off += len;
}
out
}
#[cfg(test)]
mod tests {
use super::*;
fn arms(entries: &[IndexEntry]) -> (FourTables, OneTableFourColumns, PackedPayload) {
(
FourTables::build(entries).expect("A builds"),
OneTableFourColumns::build(entries).expect("B builds"),
PackedPayload::build(entries).expect("C builds"),
)
}
#[test]
fn the_three_layouts_return_identical_rows() {
for &oid_len in &[20usize, 32] {
let entries = synthetic_entries(500, oid_len, 0xA11CE);
let (a, b, c) = arms(&entries);
assert_eq!(a.len(), 500);
assert_eq!(b.len(), 500);
assert_eq!(c.len(), 500);
let with_base = entries.iter().filter(|e| e.delta_base != 0).count();
assert!(
with_base >= 50,
"only {with_base} of 500 entries carry a non-zero delta_base — this guard would \
sit on the identity value and could not see a column that was never written"
);
let mut hits = 0usize;
for e in &entries {
let ra = a.lookup(&e.oid).unwrap_or_else(|| {
panic!("{} missed {}", a.name(), hex::encode(&e.oid))
});
let rb = b.lookup(&e.oid).unwrap_or_else(|| {
panic!("{} missed {}", b.name(), hex::encode(&e.oid))
});
let rc = c.lookup(&e.oid).unwrap_or_else(|| {
panic!("{} missed {}", c.name(), hex::encode(&e.oid))
});
assert_eq!(ra, rb, "A and B disagree on {}", hex::encode(&e.oid));
assert_eq!(
ra,
rc,
"C disagrees with A on {}",
hex::encode(&e.oid)
);
assert_eq!(rc.offset, e.offset);
assert_eq!(rc.len, e.len);
assert_eq!(rc.obj_type, e.obj_type);
assert_eq!(rc.uncompressed_size, e.uncompressed_size);
assert_eq!(
rc.delta_base,
e.delta_base,
"delta_base did not survive C for {}",
hex::encode(&e.oid)
);
assert_eq!(ra.delta_base, e.delta_base, "delta_base did not survive A");
assert_eq!(rb.delta_base, e.delta_base, "delta_base did not survive B");
hits += 1;
}
assert_eq!(hits, 500, "every entry must resolve");
let absent = synthetic_entries(200, oid_len, 0xBEEF_0000);
for e in &absent {
assert_eq!(a.lookup(&e.oid), None);
assert_eq!(b.lookup(&e.oid), None);
assert_eq!(c.lookup(&e.oid), None);
}
let mut q: Vec<&[u8]> = Vec::new();
for (i, e) in entries.iter().enumerate() {
q.push(&e.oid);
if i < absent.len() {
q.push(&absent[i].oid);
}
}
let ba = a.lookup_batch(&q);
let bb = b.lookup_batch(&q);
let bc = c.lookup_batch(&q);
assert_eq!(ba.len(), q.len());
assert_eq!(ba, bb, "A/B batch paths disagree at oid_len {oid_len}");
assert_eq!(ba, bc, "A/C batch paths disagree at oid_len {oid_len}");
let n_hits = ba.iter().filter(|r| r.is_some()).count();
assert_eq!(n_hits, 500, "expected exactly the 500 present oids to hit");
for (i, r) in ba.iter().enumerate() {
assert_eq!(*r, c.lookup(q[i]), "C batch/serial disagree at {i}");
}
let ea = a.extents_batch(&q);
let ec = c.extents_batch(&q);
assert_eq!(ea, ec, "extent paths disagree");
let oa = a.ordinals_batch(&q);
let oc = c.ordinals_batch(&q);
assert_eq!(oa, oc, "ordinal paths disagree");
for i in 0..q.len() {
assert_eq!(ea[i], ba[i].map(|r| (r.offset, r.len)), "extent != row at {i}");
assert_eq!(oa[i], ba[i].map(|r| r.ordinal), "ordinal != row at {i}");
}
}
}
#[test]
fn delta_base_locates_a_real_entry_in_every_arm() {
let entries = synthetic_entries(2000, 20, 0xDE17A);
let (a, b, c) = arms(&entries);
let offsets: std::collections::HashSet<u64> =
entries.iter().filter_map(|e| a.lookup(&e.oid)).map(|r| r.offset).collect();
assert_eq!(offsets.len(), 2000, "every row must report a distinct offset");
let arms: [(&str, &dyn ObjectIndex); 3] = [("A", &a), ("B", &b), ("C", &c)];
let mut checked = 0usize;
for (name, idx) in arms {
let mut with_base = 0usize;
for e in &entries {
let row = idx.lookup(&e.oid).expect("stored oid resolves");
if row.delta_base == 0 {
assert_ne!(
row.obj_type,
ObjType::OfsDelta,
"{name}: an ofs-delta with no recorded base is a row that cannot be \
resolved at all"
);
continue;
}
with_base += 1;
assert!(
offsets.contains(&row.delta_base),
"{name}: delta_base {} of ordinal {} lands on no entry offset in this index",
row.delta_base,
row.ordinal
);
assert!(
row.delta_base < row.offset,
"{name}: delta_base {} is not earlier in the archive than the delta at {}",
row.delta_base,
row.offset
);
}
assert!(
with_base > 100,
"{name}: only {with_base} of 2000 rows carried a base — nothing was proven"
);
checked += with_base;
}
assert!(checked > 300, "three arms must each have checked real bases");
}
#[test]
fn column_scans_agree_across_arms_and_with_the_input() {
let entries = synthetic_entries(2000, 20, 0x5CA7);
let (a, b, c) = arms(&entries);
let want: u64 = entries.iter().map(|e| e.uncompressed_size).sum();
assert_eq!(a.sum_uncompressed(), want, "A sum_uncompressed");
assert_eq!(b.sum_uncompressed(), want, "B sum_uncompressed");
assert_eq!(c.sum_uncompressed(), want, "C sum_uncompressed");
assert!(want > 0, "the workload must actually carry sizes");
let mut total = 0usize;
for t in ObjType::ALL {
let want = entries.iter().filter(|e| e.obj_type == t).count();
assert_eq!(a.count_type(t), want, "A count_type({})", t.as_str());
assert_eq!(b.count_type(t), want, "B count_type({})", t.as_str());
assert_eq!(c.count_type(t), want, "C count_type({})", t.as_str());
assert!(want > 0, "{} must appear in the workload", t.as_str());
total += want;
}
assert_eq!(total, entries.len(), "the six types must partition the index");
}
#[test]
fn the_packed_record_is_the_documented_33_bytes() {
let entries = synthetic_entries(64, 20, 0x9F);
let c = PackedPayload::build(&entries).unwrap();
assert_eq!(PACKED_LEN, 33);
assert_eq!(PACKED_LEN, 8 + 8 + 1 + 8 + 8, "offset+len+type+size+delta_base");
let mut sorted: Vec<&IndexEntry> = entries.iter().collect();
sorted.sort_by(|x, y| x.oid.cmp(&y.oid));
let d = c.payload.value_data();
assert_eq!(d.len(), 64 * PACKED_LEN, "payload column is not 33 bytes per row");
assert!(
sorted.iter().filter(|e| e.delta_base != 0).count() >= 4,
"the fixture must carry real delta bases or bytes 25..33 are all zero"
);
for (i, e) in sorted.iter().enumerate() {
let r = &d[i * PACKED_LEN..(i + 1) * PACKED_LEN];
assert_eq!(
u64::from_le_bytes(r[0..8].try_into().unwrap()),
e.offset,
"row {i} bytes 0..8 must be the offset"
);
assert_eq!(
u64::from_le_bytes(r[8..16].try_into().unwrap()),
e.len,
"row {i} bytes 8..16 must be the len"
);
assert_eq!(
r[16],
e.obj_type.code(),
"row {i} byte 16 is {} but the type code is {}",
r[16],
e.obj_type.code()
);
assert_eq!(
u64::from_le_bytes(r[17..25].try_into().unwrap()),
e.uncompressed_size,
"row {i} bytes 17..25 must be the uncompressed size"
);
assert_eq!(
u64::from_le_bytes(r[25..33].try_into().unwrap()),
e.delta_base,
"row {i} bytes 25..33 must be the delta base"
);
}
}
#[test]
fn a_packed_row_is_one_stride_and_a_columnar_row_is_five() {
let entries = synthetic_entries(100_000, 20, 0x0FF5);
let (_, b, c) = arms(&entries);
let base = |s: &[u8]| s.as_ptr() as usize;
let off = base(b.cols.offset.values().inner().as_slice());
let len = base(b.cols.len.values().inner().as_slice());
let size = base(b.cols.size.values().inner().as_slice());
let dbase = base(b.cols.delta_base.values().inner().as_slice());
assert!(
len.abs_diff(off) >= 800_000,
"offset and len columns are only {} bytes apart — this arm is supposed to be \
columnar",
len.abs_diff(off)
);
assert!(size.abs_diff(off) >= 800_000);
assert!(
dbase.abs_diff(off) >= 800_000 && dbase.abs_diff(size) >= 800_000,
"delta_base is {} bytes from offset and {} from size — the fifth fact must be its \
own stride, not a field inside another column",
dbase.abs_diff(off),
dbase.abs_diff(size)
);
let d = c.payload.value_data();
assert_eq!(d.len(), 100_000 * PACKED_LEN);
let row7 = &d[7 * PACKED_LEN..8 * PACKED_LEN];
let r = c.lookup(&entries.iter().min_by_key(|e| e.oid.clone()).unwrap().oid);
assert!(r.is_some(), "the lexicographically first oid must resolve");
let from_bytes = IndexRow {
ordinal: 7,
offset: u64::from_le_bytes(row7[0..8].try_into().unwrap()),
len: u64::from_le_bytes(row7[8..16].try_into().unwrap()),
obj_type: ObjType::from_code(row7[16]).unwrap(),
uncompressed_size: u64::from_le_bytes(row7[17..25].try_into().unwrap()),
delta_base: u64::from_le_bytes(row7[25..33].try_into().unwrap()),
};
assert_eq!(
from_bytes,
c.row_at(7),
"one 33-byte slice must carry the whole row"
);
}
#[test]
fn ordinals_are_the_oid_lexicographic_rank_in_every_arm() {
let entries = synthetic_entries(300, 20, 7);
let (a, b, c) = arms(&entries);
let mut sorted: Vec<&IndexEntry> = entries.iter().collect();
sorted.sort_by(|x, y| x.oid.cmp(&y.oid));
for (rank, e) in sorted.iter().enumerate() {
let ra = a.lookup(&e.oid).unwrap();
let rb = b.lookup(&e.oid).unwrap();
let rc = c.lookup(&e.oid).unwrap();
assert_eq!(ra.ordinal as usize, rank, "A ordinal is not the rank");
assert_eq!(rb.ordinal as usize, rank, "B ordinal is not the rank");
assert_eq!(rc.ordinal as usize, rank, "C ordinal is not the rank");
assert_eq!(a.oid_column().value(rank), e.oid.as_slice());
assert_eq!(b.oid_column().value(rank), e.oid.as_slice());
assert_eq!(c.oid_column().value(rank), e.oid.as_slice());
}
}
#[test]
fn every_column_is_a_zero_copy_view_into_its_ipc_section() {
let entries = synthetic_entries(1000, 32, 11);
let (a, b, c) = arms(&entries);
assert!(a.column_is_inside_ipc(), "FourTables columns were copied out of IPC");
assert!(b.column_is_inside_ipc(), "OneTable columns were copied out of IPC");
assert!(c.column_is_inside_ipc(), "PackedPayload columns were copied out of IPC");
assert!(
b.ipc_bytes() >= 65_000,
"one-table section is {} bytes, too small to hold the payload",
b.ipc_bytes()
);
assert!(a.ipc_bytes() >= 65_000);
assert!(c.ipc_bytes() >= 65_000);
}
#[test]
fn require_alignment_is_load_bearing() {
let entries = synthetic_entries(64, 20, 77);
let order = oid_order(&entries);
let c = columns(&entries, &order, 20);
let batch = RecordBatch::try_new(
extent_schema(),
vec![Arc::new(c.offset), Arc::new(c.len), Arc::new(c.delta_base)],
)
.unwrap();
let ipc = to_ipc(&batch).unwrap();
let mut mb = MutableBuffer::with_capacity(64 + ipc.len());
mb.extend_from_slice(&[0u8; 64]);
mb.extend_from_slice(&ipc);
let padded: Buffer = mb.into();
let (owner, ok) = decode_ipc_buffer(padded.slice(64)).expect("aligned decode must work");
assert_eq!(ok.num_rows(), 64, "the aligned decode must yield all 64 rows");
assert!(inside(&owner, ok.column(0).to_data().buffers()[0].as_slice()));
let mut mb = MutableBuffer::with_capacity(4 + ipc.len());
mb.extend_from_slice(&[0u8; 4]);
mb.extend_from_slice(&ipc);
let skewed: Buffer = mb.into();
let err = match decode_ipc_buffer(skewed.slice(4)) {
Ok(_) => panic!("a misaligned buffer must be refused, not silently copied"),
Err(e) => e.to_string(),
};
assert!(
err.contains("Misaligned"),
"the refusal must name the misalignment, got: {err}"
);
}
#[test]
fn four_sections_pay_three_extra_ipc_frames() {
let entries = synthetic_entries(64, 20, 3);
let (a, b, _) = arms(&entries);
let extra = a.ipc_bytes() as i64 - b.ipc_bytes() as i64;
assert!(
(600..1200).contains(&extra),
"expected 600..1200 bytes of extra framing for three extra sections, got {extra} \
(A={} B={})",
a.ipc_bytes(),
b.ipc_bytes()
);
let lens = a.ipc_section_lens();
assert_eq!(lens.len(), 4);
assert!(lens[0] > lens[2], "oid section {} must exceed type section {}", lens[0], lens[2]);
}
#[test]
fn the_packed_arm_holds_the_same_payload_but_a_smaller_section() {
let mut savings = Vec::new();
for &n in &[100_000usize, 200_000] {
let entries = synthetic_entries(n, 20, 0xBEE5);
let (_, b, c) = arms(&entries);
let payload_bytes = n * (8 + 8 + 1 + 8 + 8);
assert_eq!(c.payload.value_data().len(), payload_bytes);
assert_eq!(c.payload.value_data().len(), n * PACKED_LEN);
let saving = b.ipc_bytes() as i64 - c.ipc_bytes() as i64;
assert!(
saving > 0,
"the two-column section ({}) must not be larger than the six-column one ({})",
c.ipc_bytes(),
b.ipc_bytes()
);
let expect = 4 * (n as i64) / 8;
assert!(
(expect..expect + 4096).contains(&saving),
"at {n} rows the saving is {saving} B, not the ~{expect} B that four fewer \
columns accounts for"
);
savings.push(saving);
}
assert!(
savings[1] > savings[0] * 3 / 2,
"saving did not scale: {} at 100000 rows, {} at 200000",
savings[0],
savings[1]
);
}
#[test]
fn a_prefix_collision_resolves_to_the_right_row_in_every_arm() {
let prefix = [0xde, 0xad, 0xbe, 0xef, 0x01, 0x02, 0x03, 0x04];
let mk = |last: u8, off: u64, ty: ObjType| {
let mut oid = vec![0u8; 20];
oid[..8].copy_from_slice(&prefix);
oid[8] = last;
IndexEntry {
oid,
offset: off,
len: 10,
obj_type: ty,
uncompressed_size: off * 2,
delta_base: off / 2,
}
};
let mut entries = vec![mk(0xaa, 1000, ObjType::Commit), mk(0xbb, 2000, ObjType::OfsDelta)];
entries.extend(synthetic_entries(200, 20, 99));
let (a, b, c) = arms(&entries);
for (last, off, ty) in [(0xaau8, 1000u64, ObjType::Commit), (0xbb, 2000, ObjType::OfsDelta)]
{
let oid = mk(last, off, ty).oid;
let ra = a.lookup(&oid).expect("colliding oid must resolve in A");
let rb = b.lookup(&oid).expect("colliding oid must resolve in B");
let rc = c.lookup(&oid).expect("colliding oid must resolve in C");
assert_eq!(ra, rb);
assert_eq!(ra, rc);
assert_eq!(ra.offset, off, "collision resolved to the wrong row");
assert_eq!(ra.obj_type, ty);
assert_eq!(ra.delta_base, off / 2, "collision resolved to the wrong delta base");
}
let never = mk(0xcc, 0, ObjType::Blob).oid;
assert_eq!(a.lookup(&never), None, "unstored oid on a colliding prefix must miss");
assert_eq!(b.lookup(&never), None);
assert_eq!(c.lookup(&never), None);
}
#[test]
fn all_six_pack_types_round_trip_including_the_deltas() {
let entries: Vec<IndexEntry> = ObjType::ALL
.iter()
.enumerate()
.map(|(i, &t)| {
let mut oid = vec![0u8; 32];
oid[0] = i as u8 * 17;
oid[31] = i as u8;
IndexEntry {
oid,
offset: 100 + i as u64,
len: 5,
obj_type: t,
uncompressed_size: 900 + i as u64,
delta_base: match t {
ObjType::OfsDelta => 100,
_ => 0,
},
}
})
.collect();
let (a, b, c) = arms(&entries);
let mut seen: Vec<ObjType> = Vec::new();
for e in &entries {
let ra = a.lookup(&e.oid).unwrap();
assert_eq!(ra, b.lookup(&e.oid).unwrap());
assert_eq!(ra, c.lookup(&e.oid).unwrap());
assert_eq!(ra.obj_type, e.obj_type, "type did not survive the column");
seen.push(ra.obj_type);
}
assert_eq!(seen, ObjType::ALL.to_vec(), "all six codes must be distinct");
assert_eq!(ObjType::from_code(0), None);
assert_eq!(ObjType::from_code(5), None, "git leaves 5 unused; it must not be mapped");
assert_eq!(ObjType::from_code(8), None);
}
#[test]
fn degenerate_sizes_are_clean_in_every_arm() {
let (a, b, c) = arms(&[]);
assert!(a.is_empty() && b.is_empty() && c.is_empty());
let probe = vec![7u8; 20];
assert_eq!(a.lookup(&probe), None);
assert_eq!(b.lookup(&probe), None);
assert_eq!(c.lookup(&probe), None);
assert_eq!(c.lookup_batch(&[&probe[..]]), vec![None]);
assert_eq!(c.extents_batch(&[&probe[..]]), vec![None]);
assert_eq!(c.sum_uncompressed(), 0);
assert_eq!(c.count_type(ObjType::Blob), 0);
let one = synthetic_entries(1, 20, 5);
let (a, b, c) = arms(&one);
assert_eq!(a.lookup(&one[0].oid).unwrap(), b.lookup(&one[0].oid).unwrap());
assert_eq!(a.lookup(&one[0].oid).unwrap(), c.lookup(&one[0].oid).unwrap());
assert_eq!(c.lookup(&one[0].oid).unwrap().offset, one[0].offset);
assert_eq!(c.sum_uncompressed(), one[0].uncompressed_size);
}
#[test]
fn a_wrong_width_oid_misses_in_every_arm() {
let entries = synthetic_entries(64, 20, 21);
let (a, b, c) = arms(&entries);
let short = &entries[0].oid[..8];
let mut long = entries[0].oid.clone();
long.extend_from_slice(&[0u8; 12]);
for q in [short, &long[..]] {
assert_eq!(a.lookup(q), None, "a {}-byte oid must miss", q.len());
assert_eq!(b.lookup(q), None);
assert_eq!(c.lookup(q), None);
}
assert_eq!(a.lookup_batch(&[short, &long]), vec![None, None]);
assert_eq!(b.lookup_batch(&[short, &long]), vec![None, None]);
assert_eq!(c.lookup_batch(&[short, &long]), vec![None, None]);
}
#[test]
fn build_refuses_duplicates_and_mixed_widths() {
fn why<T>(r: Result<T>, what: &str) -> String {
match r {
Ok(_) => panic!("{what} must be refused at build"),
Err(e) => e.to_string(),
}
}
let mut dup = synthetic_entries(4, 20, 1);
dup.push(dup[0].clone());
let err = why(FourTables::build(&dup), "a duplicate oid");
assert!(err.contains("duplicate oid"), "{err}");
assert!(OneTableFourColumns::build(&dup).is_err());
assert!(PackedPayload::build(&dup).is_err());
let mut mixed = synthetic_entries(4, 20, 2);
mixed.push(synthetic_entries(1, 32, 3).pop().unwrap());
let err = why(PackedPayload::build(&mixed), "a mixed-width index");
assert!(err.contains("mixed oid widths"), "{err}");
assert!(FourTables::build(&mixed).is_err());
assert!(OneTableFourColumns::build(&mixed).is_err());
}
#[test]
fn an_index_can_be_shared_across_threads() {
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<FourTables>();
assert_send_sync::<OneTableFourColumns>();
assert_send_sync::<PackedPayload>();
let entries = synthetic_entries(256, 20, 42);
let a: Arc<dyn ObjectIndex> = Arc::new(FourTables::build(&entries).unwrap());
let b: Arc<dyn ObjectIndex> = Arc::new(OneTableFourColumns::build(&entries).unwrap());
let c: Arc<dyn ObjectIndex> = Arc::new(PackedPayload::build(&entries).unwrap());
let oids: Vec<Vec<u8>> = entries.iter().map(|e| e.oid.clone()).collect();
let mut handles = Vec::new();
for idx in [a, b, c] {
let oids = oids.clone();
handles.push(std::thread::spawn(move || {
let refs: Vec<&[u8]> = oids.iter().map(|o| o.as_slice()).collect();
idx.lookup_batch(&refs).iter().filter(|r| r.is_some()).count()
}));
}
for h in handles {
assert_eq!(h.join().unwrap(), 256, "every oid must resolve off-thread");
}
}
}