use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::os::unix::fs::FileExt;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::{AtomicU64, Ordering};
use anyhow::{Context, Result};
use znippy_common::arrow::array::{
Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, LargeBinaryArray, LargeBinaryBuilder,
StringArray, StringBuilder, UInt8Array, UInt8Builder, UInt32Array, UInt32Builder,
};
use znippy_common::arrow::buffer::Buffer;
use znippy_common::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
use znippy_common::arrow::ipc::reader::StreamDecoder;
use znippy_common::arrow::ipc::writer::{
DictionaryTracker, IpcDataGenerator, IpcWriteOptions, write_message,
};
use znippy_common::arrow::ipc::root_as_message;
use znippy_common::arrow::record_batch::RecordBatch;
use znippy_zoomies::stree::STree64Mmap;
use crate::object::GitObjectKind;
use crate::oid_index::key_for_oid;
pub const COL_OID: &str = "oid";
pub const COL_TYPE: &str = "object_type";
pub const COL_MODE: &str = "mode";
pub const COL_PATH: &str = "path";
pub const COL_PAYLOAD: &str = "payload";
pub const FLUSH_BYTES: usize = 64 << 20;
pub fn exploded_schema(oid_len: usize) -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new(COL_OID, DataType::FixedSizeBinary(oid_len as i32), false),
Field::new(COL_TYPE, DataType::UInt8, false),
Field::new(COL_MODE, DataType::UInt32, true),
Field::new(COL_PATH, DataType::Utf8, true),
Field::new(COL_PAYLOAD, DataType::LargeBinary, false),
]))
}
pub fn kind_code(k: GitObjectKind) -> u8 {
match k {
GitObjectKind::Commit => 1,
GitObjectKind::Tree => 2,
GitObjectKind::Blob => 3,
GitObjectKind::Tag => 4,
}
}
pub fn kind_of(code: u8) -> Option<GitObjectKind> {
match code {
1 => Some(GitObjectKind::Commit),
2 => Some(GitObjectKind::Tree),
3 => Some(GitObjectKind::Blob),
4 => Some(GitObjectKind::Tag),
_ => None,
}
}
pub const TOMBSTONE: u8 = 0;
#[derive(Debug, Clone)]
struct Pending {
oid: Vec<u8>,
code: u8,
mode: Option<u32>,
path: Option<String>,
payload: Vec<u8>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Batch {
at: u64,
len: u64,
rows: u32,
body: u64,
pay_data: u64,
}
const EXTENT_UNAVAILABLE: u64 = u64::MAX;
fn payload_data_offset(meta: &[u8]) -> Option<u64> {
let msg = root_as_message(meta).ok()?;
let rb = msg.header_as_record_batch()?;
if rb.compression().is_some() {
return None;
}
let bufs = rb.buffers()?;
if bufs.len() != 12 {
return None;
}
let last = bufs.get(bufs.len() - 1);
(last.offset() >= 0).then(|| last.offset() as u64)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct Located {
batch: u32,
row: u32,
kind: GitObjectKind,
pay_at: u64,
pay_len: u64,
}
struct OidTree {
oid_len: usize,
count: usize,
keys: Vec<u8>,
oids: Vec<u8>,
locs: Vec<Located>,
tree: Option<STree64Mmap>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Keep {
First,
Last,
}
struct OidTreeBuilder {
oid_len: usize,
oids: Vec<u8>,
locs: Vec<Option<Located>>,
}
impl OidTreeBuilder {
fn with_capacity(rows: usize) -> Self {
Self { oid_len: 0, oids: Vec::new(), locs: Vec::with_capacity(rows) }
}
fn push(&mut self, oid: &[u8], loc: Option<Located>) -> Result<()> {
if self.locs.is_empty() && self.oids.is_empty() {
self.oid_len = oid.len();
self.oids.reserve(self.locs.capacity() * self.oid_len);
}
anyhow::ensure!(
oid.len() == self.oid_len,
"the exploded index holds {}-byte oids and a {}-byte one arrived",
self.oid_len,
oid.len()
);
self.oids.extend_from_slice(oid);
self.locs.push(loc);
Ok(())
}
fn oid_at(&self, i: usize) -> &[u8] {
&self.oids[i * self.oid_len..(i + 1) * self.oid_len]
}
fn finish(self, keep: Keep) -> OidTree {
let n = self.locs.len();
let mut order: Vec<u32> = (0..n as u32).collect();
order.sort_by(|&a, &b| self.oid_at(a as usize).cmp(self.oid_at(b as usize)));
let mut keys: Vec<u8> = Vec::with_capacity(n * 8);
let mut oids: Vec<u8> = Vec::with_capacity(n * self.oid_len);
let mut locs: Vec<Located> = Vec::with_capacity(n);
let mut i = 0usize;
while i < n {
let oid = self.oid_at(order[i] as usize);
let mut j = i + 1;
while j < n && self.oid_at(order[j] as usize) == oid {
j += 1;
}
let pick = match keep {
Keep::First => order[i],
Keep::Last => order[j - 1],
} as usize;
if let Some(loc) = self.locs[pick] {
keys.extend_from_slice(&key_for_oid(oid).to_le_bytes());
oids.extend_from_slice(oid);
locs.push(loc);
}
i = j;
}
let count = locs.len();
let tree = (count > 0).then(|| STree64Mmap::new_with_stride(&keys, count, 8));
OidTree { oid_len: self.oid_len, count, keys, oids, locs, tree }
}
}
impl OidTree {
fn len(&self) -> usize {
self.count
}
fn oid_at(&self, i: usize) -> &[u8] {
&self.oids[i * self.oid_len..(i + 1) * self.oid_len]
}
fn key_at(&self, i: usize) -> i64 {
i64::from_le_bytes(self.keys[i * 8..i * 8 + 8].try_into().unwrap())
}
fn expand_run(&self, pos: usize, key: i64) -> std::ops::Range<usize> {
let mut lo = pos;
while lo > 0 && self.key_at(lo - 1) == key {
lo -= 1;
}
let mut hi = pos + 1;
while hi < self.count && self.key_at(hi) == key {
hi += 1;
}
lo..hi
}
fn find(&self, oid: &[u8]) -> Option<Located> {
if oid.len() != self.oid_len {
return None;
}
let tree = self.tree.as_ref()?;
let key = key_for_oid(oid);
let pos = tree.find_exact(key, &self.keys)?;
self.verify(pos, key, oid)
}
fn verify(&self, pos: usize, key: i64, oid: &[u8]) -> Option<Located> {
self.expand_run(pos, key)
.find(|&i| self.oid_at(i) == oid)
.map(|i| self.locs[i])
}
fn iter(&self) -> impl Iterator<Item = (&[u8], Located)> + '_ {
(0..self.count).map(move |i| (self.oid_at(i), self.locs[i]))
}
#[cfg(test)]
fn find_by_binary_search(&self, oid: &[u8]) -> Option<Located> {
if oid.len() != self.oid_len || self.count == 0 {
return None;
}
let mut lo = 0usize;
let mut hi = self.count;
while lo < hi {
let mid = lo + (hi - lo) / 2;
if self.oid_at(mid) < oid { lo = mid + 1 } else { hi = mid }
}
(lo < self.count && self.oid_at(lo) == oid).then(|| self.locs[lo])
}
#[cfg(test)]
fn candidate_run(&self, key: i64) -> std::ops::Range<usize> {
let Some(tree) = self.tree.as_ref() else { return 0..0 };
let Some(pos) = tree.find_exact(key, &self.keys) else { return 0..0 };
self.expand_run(pos, key)
}
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub struct Stats {
pub rows: u64,
pub written: u64,
pub served: u64,
pub rederived: u64,
pub absent: u64,
}
pub struct ExplodedArchive {
path: PathBuf,
schema_msg: Mutex<Option<Vec<u8>>>,
batches: Mutex<Vec<Batch>>,
index: Mutex<Option<OidTree>>,
cache: Mutex<Vec<(u32, RecordBatch)>>,
pending: Mutex<Vec<Pending>>,
pending_bytes: AtomicU64,
end: Mutex<u64>,
oid_len: Mutex<Option<usize>>,
policy: ExplodePolicy,
skipped: AtomicU64,
written: AtomicU64,
served: AtomicU64,
pread_served: AtomicU64,
rederived: AtomicU64,
absent: AtomicU64,
}
impl ExplodedArchive {
pub fn open(path: &Path) -> Result<Self> {
Self::open_with_policy(path, ExplodePolicy::from_env()?)
}
pub fn open_with_policy(path: &Path, policy: ExplodePolicy) -> Result<Self> {
if let Some(parent) = path.parent()
&& !parent.as_os_str().is_empty()
{
std::fs::create_dir_all(parent)
.with_context(|| format!("creating {}", parent.display()))?;
}
let (schema_msg, batches, end, oid_len) = Self::walk(path)?;
Ok(Self {
path: path.to_path_buf(),
schema_msg: Mutex::new(schema_msg),
batches: Mutex::new(batches),
index: Mutex::new(None),
cache: Mutex::new(Vec::new()),
pending: Mutex::new(Vec::new()),
pending_bytes: AtomicU64::new(0),
end: Mutex::new(end),
oid_len: Mutex::new(oid_len),
policy,
skipped: AtomicU64::new(0),
written: AtomicU64::new(0),
served: AtomicU64::new(0),
pread_served: AtomicU64::new(0),
rederived: AtomicU64::new(0),
absent: AtomicU64::new(0),
})
}
pub fn path(&self) -> &Path {
&self.path
}
#[allow(clippy::type_complexity)]
fn walk(path: &Path) -> Result<(Option<Vec<u8>>, Vec<Batch>, u64, Option<usize>)> {
let Ok(mut f) = File::open(path) else {
return Ok((None, Vec::new(), 0, None));
};
let total = f.metadata()?.len();
let mut schema_msg: Option<Vec<u8>> = None;
let mut oid_len: Option<usize> = None;
let mut batches = Vec::new();
let mut at = 0u64;
while at + 8 <= total {
f.seek(SeekFrom::Start(at))?;
let mut hdr = [0u8; 8];
if f.read_exact(&mut hdr).is_err() {
break;
}
if u32::from_le_bytes(hdr[0..4].try_into().unwrap()) != 0xFFFF_FFFF {
break;
}
let meta_len = u32::from_le_bytes(hdr[4..8].try_into().unwrap()) as u64;
if meta_len == 0 {
break; }
if at + 8 + meta_len > total {
break; }
let mut meta = vec![0u8; meta_len as usize];
if f.read_exact(&mut meta).is_err() {
break;
}
let Ok(msg) = root_as_message(&meta) else {
break;
};
let body = msg.bodyLength().max(0) as u64;
let whole = 8 + meta_len + body;
if at + whole > total {
break; }
if let Some(rb) = msg.header_as_record_batch() {
batches.push(Batch {
at,
len: whole,
rows: rb.length().max(0) as u32,
body: at + 8 + meta_len,
pay_data: payload_data_offset(&meta).unwrap_or(EXTENT_UNAVAILABLE),
});
} else if msg.header_as_schema().is_some() {
schema_msg = Some({
let mut whole_msg = Vec::with_capacity((8 + meta_len) as usize);
whole_msg.extend_from_slice(&hdr);
whole_msg.extend_from_slice(&meta);
whole_msg
});
oid_len = Self::oid_len_of_schema_msg(schema_msg.as_ref().unwrap());
}
at += whole;
}
Ok((schema_msg, batches, at, oid_len))
}
fn oid_len_of_schema_msg(msg: &[u8]) -> Option<usize> {
let mut dec = StreamDecoder::new();
let mut buf = Buffer::from_vec(msg.to_vec());
dec.decode(&mut buf).ok()?;
let schema = dec.schema()?;
match schema.field_with_name(COL_OID).ok()?.data_type() {
DataType::FixedSizeBinary(n) => Some(*n as usize),
_ => None,
}
}
pub fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
self.explode_at(oid, kind, None, None, payload)
}
pub fn explode_at(
&self,
oid: &[u8],
kind: GitObjectKind,
mode: Option<u32>,
path: Option<&str>,
payload: &[u8],
) -> Result<()> {
if !self.policy.wants(kind) {
self.skipped.fetch_add(1, Ordering::Relaxed);
return Ok(());
}
{
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
p.push(Pending {
oid: oid.to_vec(),
code: kind_code(kind),
mode,
path: path.map(str::to_owned),
payload: payload.to_vec(),
});
}
let n = self
.pending_bytes
.fetch_add(payload.len() as u64, Ordering::AcqRel)
+ payload.len() as u64;
if n >= FLUSH_BYTES as u64 {
self.flush()?;
}
Ok(())
}
fn build_batch(rows: &[Pending], oid_len: usize) -> Result<RecordBatch> {
let mut oid_b = FixedSizeBinaryBuilder::with_capacity(rows.len(), oid_len as i32);
let mut kind_b = UInt8Builder::with_capacity(rows.len());
let mut mode_b = UInt32Builder::with_capacity(rows.len());
let mut path_b = StringBuilder::new();
let mut pay_b = LargeBinaryBuilder::new();
for r in rows {
oid_b
.append_value(&r.oid)
.map_err(|e| anyhow::anyhow!("exploded oid column: {e}"))?;
kind_b.append_value(r.code);
mode_b.append_option(r.mode);
path_b.append_option(r.path.as_deref());
pay_b.append_value(&r.payload);
}
RecordBatch::try_new(
exploded_schema(oid_len),
vec![
Arc::new(oid_b.finish()),
Arc::new(kind_b.finish()),
Arc::new(mode_b.finish()),
Arc::new(path_b.finish()),
Arc::new(pay_b.finish()),
],
)
.context("building the exploded batch")
}
pub fn flush(&self) -> Result<u64> {
let rows = {
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if p.is_empty() {
return Ok(0);
}
self.pending_bytes.store(0, Ordering::Release);
std::mem::take(&mut *p)
};
let n = rows.len() as u64;
let width = rows[0].oid.len();
anyhow::ensure!(
rows.iter().all(|r| r.oid.len() == width),
"one exploded batch cannot hold two hash widths"
);
{
let mut w = self
.oid_len
.lock()
.map_err(|_| anyhow::anyhow!("exploded oid width poisoned"))?;
match *w {
Some(have) => anyhow::ensure!(
have == width,
"this table holds {have}-byte oids and a {width}-byte one arrived"
),
None => *w = Some(width),
}
}
let batch = Self::build_batch(&rows, width)?;
let opts = IpcWriteOptions::default();
let ipc = IpcDataGenerator::default();
let mut tracker = DictionaryTracker::new(false);
let mut end = self
.end
.lock()
.map_err(|_| anyhow::anyhow!("exploded end poisoned"))?;
let mut f = OpenOptions::new()
.create(true)
.read(true)
.write(true)
.truncate(false)
.open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
if f.metadata()?.len() != *end {
f.set_len(*end)
.with_context(|| format!("truncating {} to {}", self.path.display(), *end))?;
}
f.seek(SeekFrom::Start(*end))?;
let mut schema_guard = self
.schema_msg
.lock()
.map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
if schema_guard.is_none() {
let enc =
ipc.schema_to_bytes_with_dictionary_tracker(&exploded_schema(width), &mut tracker, &opts);
let mut msg = Vec::new();
let (meta, body) = write_message(&mut msg, enc, &opts)
.map_err(|e| anyhow::anyhow!("encoding the exploded schema: {e}"))?;
debug_assert_eq!(meta + body, msg.len());
f.write_all(&msg).context("writing the exploded schema")?;
*end += msg.len() as u64;
*schema_guard = Some(msg);
}
drop(schema_guard);
let (dicts, enc) = ipc
.encode(&batch, &mut tracker, &opts, &mut Default::default())
.map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
anyhow::ensure!(
dicts.is_empty(),
"the exploded schema has no dictionary columns, yet {} arrived",
dicts.len()
);
let mut msg = Vec::new();
let (meta, body) = write_message(&mut msg, enc, &opts)
.map_err(|e| anyhow::anyhow!("encoding an exploded batch: {e}"))?;
debug_assert_eq!(meta + body, msg.len());
f.write_all(&msg).context("appending an exploded batch")?;
let placed = Batch {
at: *end,
len: msg.len() as u64,
rows: rows.len() as u32,
body: *end + meta as u64,
pay_data: payload_data_offset(&msg[8..meta]).unwrap_or(EXTENT_UNAVAILABLE),
};
*end += msg.len() as u64;
let batch_no = {
let mut b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
b.push(placed);
(b.len() - 1) as u32
};
drop(end);
let mut idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if let Some(old) = idx.take() {
let mut retired: std::collections::HashSet<&[u8]> = std::collections::HashSet::new();
for r in rows.iter() {
if kind_of(r.code).is_none() {
retired.insert(r.oid.as_slice());
}
}
let mut b = OidTreeBuilder::with_capacity(old.len() + rows.len());
for (oid, loc) in old.iter() {
if !retired.contains(oid) {
b.push(oid, Some(loc))?;
}
}
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
let off = pay.value_offsets();
for (i, r) in rows.iter().enumerate() {
if let Some(kind) = kind_of(r.code)
&& !retired.contains(r.oid.as_slice())
{
let (pay_at, pay_len) = if placed.pay_data == EXTENT_UNAVAILABLE {
(EXTENT_UNAVAILABLE, 0)
} else {
(
placed.body + placed.pay_data + off[i] as u64,
(off[i + 1] - off[i]) as u64,
)
};
b.push(
&r.oid,
Some(Located { batch: batch_no, row: i as u32, kind, pay_at, pay_len }),
)?;
}
}
*idx = Some(b.finish(Keep::First));
}
self.written.fetch_add(n, Ordering::Relaxed);
Ok(n)
}
const CACHE: usize = 4;
fn read_batch(&self, no: u32) -> Result<RecordBatch> {
{
let mut c = self
.cache
.lock()
.map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
if let Some(at) = c.iter().position(|(n, _)| *n == no) {
let hit = c.remove(at);
let batch = hit.1.clone();
c.insert(0, hit);
return Ok(batch);
}
}
let placed = {
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
*b.get(no as usize)
.with_context(|| format!("exploded batch {no} is not in this table"))?
};
let schema_msg = {
let s = self
.schema_msg
.lock()
.map_err(|_| anyhow::anyhow!("exploded schema poisoned"))?;
s.clone()
.context("the exploded table has batches but no schema message")?
};
let f = File::open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
let mut raw = vec![0u8; placed.len as usize];
f.read_exact_at(&mut raw, placed.at)
.with_context(|| format!("reading exploded batch {no}"))?;
let mut dec = StreamDecoder::new();
let mut head = Buffer::from_vec(schema_msg);
dec.decode(&mut head)
.map_err(|e| anyhow::anyhow!("decoding the exploded schema: {e}"))?;
let mut body = Buffer::from_vec(raw);
let batch = dec
.decode(&mut body)
.map_err(|e| anyhow::anyhow!("decoding exploded batch {no}: {e}"))?
.with_context(|| format!("exploded batch {no} decoded to no rows"))?;
{
let mut c = self
.cache
.lock()
.map_err(|_| anyhow::anyhow!("exploded batch cache poisoned"))?;
c.retain(|(n, _)| *n != no);
c.insert(0, (no, batch.clone()));
c.truncate(Self::CACHE);
}
Ok(batch)
}
fn index_now(&self) -> Result<()> {
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if idx.is_some() {
return Ok(());
}
}
let (count, rows, placed) = {
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
(b.len() as u32, b.iter().map(|x| x.rows as usize).sum::<usize>(), b.clone())
};
let mut fold = OidTreeBuilder::with_capacity(rows);
for no in 0..count {
let batch = self.read_batch(no)?;
let oid = batch
.column_by_name(COL_OID)
.and_then(|c| c.as_any().downcast_ref::<FixedSizeBinaryArray>())
.context("the exploded table has no oid column")?;
let kinds = batch
.column_by_name(COL_TYPE)
.and_then(|c| c.as_any().downcast_ref::<UInt8Array>())
.context("the exploded table has no object_type column")?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
let off = pay.value_offsets();
let b = placed[no as usize];
for i in 0..batch.num_rows() {
fold.push(
oid.value(i),
kind_of(kinds.value(i)).map(|kind| {
let (pay_at, pay_len) = if b.pay_data == EXTENT_UNAVAILABLE {
(EXTENT_UNAVAILABLE, 0)
} else {
(
b.body + b.pay_data + off[i] as u64,
(off[i + 1] - off[i]) as u64,
)
};
Located { batch: no, row: i as u32, kind, pay_at, pay_len }
}),
)?;
}
}
let out = fold.finish(Keep::Last);
let mut idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if idx.is_none() {
*idx = Some(out);
}
Ok(())
}
fn find(&self, oid: &[u8]) -> Result<Option<Located>> {
self.index_now()?;
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
Ok(idx.as_ref().expect("index_now built it").find(oid))
}
pub fn content(&self, oid: &[u8]) -> Result<Option<(GitObjectKind, Vec<u8>)>> {
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
let Some(kind) = kind_of(r.code) else {
self.absent.fetch_add(1, Ordering::Relaxed);
return Ok(None);
};
self.served.fetch_add(1, Ordering::Relaxed);
return Ok(Some((kind, r.payload.clone())));
}
}
let Some(loc) = self.find(oid)? else {
self.absent.fetch_add(1, Ordering::Relaxed);
return Ok(None);
};
if loc.pay_at != EXTENT_UNAVAILABLE {
let f = File::open(&self.path)
.with_context(|| format!("opening {}", self.path.display()))?;
let mut buf = vec![0u8; loc.pay_len as usize];
f.read_exact_at(&mut buf, loc.pay_at)
.with_context(|| format!("preading payload at {}", loc.pay_at))?;
self.served.fetch_add(1, Ordering::Relaxed);
self.pread_served.fetch_add(1, Ordering::Relaxed);
return Ok(Some((loc.kind, buf)));
}
let batch = self.read_batch(loc.batch)?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
self.served.fetch_add(1, Ordering::Relaxed);
Ok(Some((loc.kind, pay.value(loc.row as usize).to_vec())))
}
pub fn pread_served(&self) -> u64 {
self.pread_served.load(Ordering::Relaxed)
}
pub fn name(&self, oid: &[u8]) -> Result<Option<(Option<u32>, Option<String>)>> {
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
return match kind_of(r.code) {
Some(_) => Ok(Some((r.mode, r.path.clone()))),
None => Ok(None),
};
}
}
let Some(loc) = self.find(oid)? else {
return Ok(None);
};
let batch = self.read_batch(loc.batch)?;
let mode = batch
.column_by_name(COL_MODE)
.and_then(|c| c.as_any().downcast_ref::<UInt32Array>())
.context("the exploded table has no mode column")?;
let path = batch
.column_by_name(COL_PATH)
.and_then(|c| c.as_any().downcast_ref::<StringArray>())
.context("the exploded table has no path column")?;
let i = loc.row as usize;
Ok(Some((
(!mode.is_null(i)).then(|| mode.value(i)),
(!path.is_null(i)).then(|| path.value(i).to_owned()),
)))
}
pub fn has(&self, oid: &[u8]) -> Result<bool> {
{
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
if let Some(r) = p.iter().rev().find(|r| r.oid == oid) {
return Ok(kind_of(r.code).is_some());
}
}
Ok(self.find(oid)?.is_some())
}
pub fn of_kind(&self, kind: GitObjectKind) -> Result<Vec<(Vec<u8>, Vec<u8>)>> {
self.flush()?;
self.index_now()?;
let mut by_batch: std::collections::BTreeMap<u32, Vec<(u32, Vec<u8>)>> =
std::collections::BTreeMap::new();
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
for (oid, loc) in idx.as_ref().expect("index_now built it").iter() {
if loc.kind == kind {
by_batch.entry(loc.batch).or_default().push((loc.row, oid.to_vec()));
}
}
}
let mut out = Vec::new();
for (no, mut wanted) in by_batch {
wanted.sort_unstable_by_key(|(r, _)| *r);
let batch = self.read_batch(no)?;
let pay = batch
.column_by_name(COL_PAYLOAD)
.and_then(|c| c.as_any().downcast_ref::<LargeBinaryArray>())
.context("the exploded table has no payload column")?;
for (row, oid) in wanted {
out.push((oid, pay.value(row as usize).to_vec()));
}
}
Ok(out)
}
pub fn retain(&self, live: &dyn Fn(&[u8]) -> bool) -> Result<u64> {
self.flush()?;
self.index_now()?;
let dead: Vec<Vec<u8>> = {
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
idx.as_ref()
.expect("index_now built it")
.iter()
.filter(|(oid, _)| !live(oid))
.map(|(oid, _)| oid.to_vec())
.collect()
};
if dead.is_empty() {
return Ok(0);
}
{
let mut p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
for oid in &dead {
p.push(Pending {
oid: oid.clone(),
code: TOMBSTONE,
mode: None,
path: None,
payload: Vec::new(),
});
}
}
self.flush()?;
Ok(dead.len() as u64)
}
pub fn rows(&self) -> Result<u64> {
let pending = {
let p = self
.pending
.lock()
.map_err(|_| anyhow::anyhow!("exploded pending poisoned"))?;
p.len() as u64
};
{
let idx = self
.index
.lock()
.map_err(|_| anyhow::anyhow!("exploded index poisoned"))?;
if let Some(idx) = idx.as_ref() {
return Ok(idx.len() as u64 + pending);
}
}
let b = self
.batches
.lock()
.map_err(|_| anyhow::anyhow!("exploded batches poisoned"))?;
Ok(b.iter().map(|x| x.rows as u64).sum::<u64>() + pending)
}
pub fn disk_bytes(&self) -> u64 {
std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
}
pub fn note_rederived(&self) {
self.rederived.fetch_add(1, Ordering::Relaxed);
}
pub fn policy(&self) -> ExplodePolicy {
self.policy
}
pub fn skipped(&self) -> u64 {
self.skipped.load(Ordering::Relaxed)
}
pub fn stats(&self) -> Stats {
Stats {
rows: self.rows().unwrap_or(0),
written: self.written.load(Ordering::Relaxed),
served: self.served.load(Ordering::Relaxed),
rederived: self.rederived.load(Ordering::Relaxed),
absent: self.absent.load(Ordering::Relaxed),
}
}
pub fn engine_stats(&self) -> crate::exploded::ExplodedStats {
let s = self.stats();
crate::exploded::ExplodedStats {
rows: s.rows,
written: s.written,
served: s.served,
rederived: s.rederived,
absent: s.absent,
}
}
}
impl crate::exploded::PayloadSink for ExplodedArchive {
fn explode(&self, oid: &[u8], kind: GitObjectKind, payload: &[u8]) -> Result<()> {
ExplodedArchive::explode(self, oid, kind, payload)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExplodePolicy {
Off,
Graph,
#[default]
Full,
}
impl ExplodePolicy {
pub const fn wants(self, kind: GitObjectKind) -> bool {
match self {
ExplodePolicy::Off => false,
ExplodePolicy::Graph => matches!(kind, GitObjectKind::Commit | GitObjectKind::Tree),
ExplodePolicy::Full => true,
}
}
pub const fn as_str(self) -> &'static str {
match self {
ExplodePolicy::Off => "off",
ExplodePolicy::Graph => "graph",
ExplodePolicy::Full => "full",
}
}
pub fn parse(s: &str) -> Result<Self> {
match s.trim().to_ascii_lowercase().as_str() {
"off" | "none" => Ok(ExplodePolicy::Off),
"graph" | "commits-and-trees" => Ok(ExplodePolicy::Graph),
"full" | "all" => Ok(ExplodePolicy::Full),
other => anyhow::bail!("unknown explode policy {other:?}; expected off, graph or full"),
}
}
pub fn from_env() -> Result<Self> {
match crate::arms::read_env(crate::arms::ENV_EXPLODE) {
Some(v) => Self::parse(&v),
None => Ok(Self::default()),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::exploded::PayloadSink as _;
fn tmp(name: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!("exploded-arrow-{}-{name}", std::process::id()));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d.join("objects.exploded")
}
fn oid(n: u8) -> [u8; 20] {
[n; 20]
}
#[test]
fn the_table_holds_the_payload_and_a_name_in_one_row() {
let s = exploded_schema(20);
let names: Vec<&str> = s.fields().iter().map(|f| f.name().as_str()).collect();
assert_eq!(names, vec![COL_OID, COL_TYPE, COL_MODE, COL_PATH, COL_PAYLOAD]);
assert_eq!(
s.field_with_name(COL_PAYLOAD).unwrap().data_type(),
&DataType::LargeBinary,
"Binary offsets are i32 and would cap one batch at 2 GiB of payload"
);
assert!(
s.field_with_name(COL_PATH).unwrap().is_nullable(),
"a bare pack resolve knows no path; null must be a legal answer"
);
}
#[test]
fn the_file_stays_close_to_the_payload_total() {
let t = ExplodedArchive::open_with_policy(&tmp("size"), ExplodePolicy::Full).unwrap();
let payload = vec![7u8; 4096];
let n = 2000u64;
for i in 0..n {
let mut o = [0u8; 20];
o[..8].copy_from_slice(&i.to_le_bytes());
t.explode(&o, GitObjectKind::Blob, &payload).unwrap();
}
t.flush().unwrap();
let payload_total = n * payload.len() as u64;
let on_disk = t.disk_bytes();
assert!(
on_disk < payload_total + payload_total / 4,
"{on_disk} bytes on disk against {payload_total} of payload — the medium is amplifying"
);
assert_eq!(t.rows().unwrap(), n);
}
#[test]
fn a_point_read_preads_the_exact_payload_without_decoding_the_batch() {
let path = tmp("pread");
let pay = |b: u8, i: u8| vec![b ^ i; 100 + i as usize * 7];
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
for i in 0..40u8 {
t.explode(&[i; 20], GitObjectKind::Blob, &pay(0, i)).unwrap();
}
t.flush().unwrap();
assert!(t.content(&[0u8; 20]).unwrap().is_some(), "build the index");
for i in 40..80u8 {
t.explode(&[i; 20], GitObjectKind::Blob, &pay(1, i)).unwrap();
}
t.flush().unwrap();
let before = t.pread_served();
for i in 0..80u8 {
let (kind, got) = t.content(&[i; 20]).unwrap().expect("every oid is live");
assert_eq!(kind, GitObjectKind::Blob);
let want = if i < 40 { pay(0, i) } else { pay(1, i) };
assert_eq!(got, want, "oid {i}: the pread returned some OTHER bytes");
}
assert_eq!(
t.pread_served() - before,
80,
"a read fell back to the batch decode — the extent was not computed"
);
drop(t);
let t = ExplodedArchive::open_with_policy(&path, ExplodePolicy::Full).unwrap();
for i in 0..80u8 {
let (_, got) = t.content(&[i; 20]).unwrap().expect("survives a reopen");
let want = if i < 40 { pay(0, i) } else { pay(1, i) };
assert_eq!(got, want, "oid {i} after reopen");
}
assert_eq!(t.pread_served(), 80, "the reopened table must pread too");
}
#[test]
fn a_payload_round_trips_through_the_table() {
let t = ExplodedArchive::open_with_policy(&tmp("rt"), ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"tree deadbeef\n")
.unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"hello world")
.unwrap();
t.flush().unwrap();
assert_eq!(
t.content(&oid(2)).unwrap().unwrap(),
(GitObjectKind::Blob, b"hello world".to_vec())
);
assert_eq!(
t.content(&oid(1)).unwrap().unwrap(),
(GitObjectKind::Commit, b"tree deadbeef\n".to_vec())
);
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(9)).unwrap());
}
#[test]
fn a_row_can_carry_the_name_it_was_seen_at() {
let t = ExplodedArchive::open_with_policy(&tmp("named"), ExplodePolicy::Full).unwrap();
t.explode_at(
&oid(1),
GitObjectKind::Blob,
Some(0o100755),
Some("scripts/build.sh"),
b"#!/bin/sh\n",
)
.unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"anonymous").unwrap();
t.flush().unwrap();
assert_eq!(
t.name(&oid(1)).unwrap().unwrap(),
(Some(0o100755), Some("scripts/build.sh".to_owned()))
);
assert_eq!(
t.name(&oid(2)).unwrap().unwrap(),
(None, None),
"null is 'not recorded', and must not read as a real name"
);
}
#[test]
fn the_table_survives_a_reopen_and_keeps_appending() {
let p = tmp("reopen");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(5), GitObjectKind::Tree, b"100644 f\0").unwrap();
t.flush().unwrap();
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(t.rows().unwrap(), 1, "row count comes from batch metadata");
t.explode(&oid(6), GitObjectKind::Blob, b"second session").unwrap();
t.flush().unwrap();
assert_eq!(t.rows().unwrap(), 2);
assert_eq!(
t.content(&oid(5)).unwrap().unwrap(),
(GitObjectKind::Tree, b"100644 f\0".to_vec())
);
assert_eq!(
t.content(&oid(6)).unwrap().unwrap(),
(GitObjectKind::Blob, b"second session".to_vec())
);
let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(again.rows().unwrap(), 2);
assert_eq!(again.batches.lock().unwrap().len(), 2);
}
#[test]
fn an_unflushed_row_is_still_served() {
let t = ExplodedArchive::open_with_policy(&tmp("unflushed"), ExplodePolicy::Full).unwrap();
t.explode(&oid(4), GitObjectKind::Commit, b"buffered").unwrap();
assert_eq!(
t.content(&oid(4)).unwrap().unwrap(),
(GitObjectKind::Commit, b"buffered".to_vec())
);
assert!(t.has(&oid(4)).unwrap());
}
#[test]
fn an_absent_oid_is_none_and_counted() {
let t = ExplodedArchive::open_with_policy(&tmp("absent"), ExplodePolicy::Full).unwrap();
assert!(t.content(&oid(3)).unwrap().is_none());
assert_eq!(t.stats().absent, 1);
assert_eq!(t.stats().served, 0);
}
#[test]
fn of_kind_selects_by_kind() {
let t = ExplodedArchive::open_with_policy(&tmp("kind"), ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c1").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"bb").unwrap();
t.explode(&oid(3), GitObjectKind::Commit, b"c2").unwrap();
let commits = t.of_kind(GitObjectKind::Commit).unwrap();
assert_eq!(commits.len(), 2);
assert!(commits.iter().all(|(_, p)| p[0] == b'c'));
}
#[test]
fn retain_drops_only_what_is_dead_and_the_drop_survives_a_reopen() {
let p = tmp("retain");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Blob, b"a").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
t.flush().unwrap();
assert_eq!(t.retain(&|o: &[u8]| o[0] == 1).unwrap(), 1);
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(2)).unwrap());
assert_eq!(
t.of_kind(GitObjectKind::Blob).unwrap().len(),
1,
"a scan must not resurrect what retain dropped"
);
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert!(t.has(&oid(1)).unwrap(), "the live row did not survive");
assert!(
!t.has(&oid(2)).unwrap(),
"a reopen resurrected a retired row — the tombstone is not durable"
);
assert!(t.content(&oid(2)).unwrap().is_none());
assert_eq!(t.of_kind(GitObjectKind::Blob).unwrap().len(), 1);
}
#[test]
fn a_sha256_oid_keeps_all_thirty_two_bytes() {
let p = tmp("sha256");
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
let o = [3u8; 32];
t.explode(&o, GitObjectKind::Tree, b"t").unwrap();
t.flush().unwrap();
assert!(t.has(&o).unwrap());
assert_eq!(t.content(&o).unwrap().unwrap().1, b"t");
t.explode(&oid(1), GitObjectKind::Blob, b"narrow").unwrap();
let err = t.flush().expect_err("a 20-byte oid cannot join a 32-byte table");
assert!(format!("{err}").contains("32-byte oids"), "{err}");
}
#[test]
fn a_torn_tail_is_dropped_and_the_table_keeps_working() {
let p = tmp("torn");
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.explode(&oid(1), GitObjectKind::Blob, b"complete").unwrap();
t.flush().unwrap();
}
{
let mut f = OpenOptions::new().append(true).open(&p).unwrap();
f.write_all(&[0xFF, 0xFF, 0xFF, 0xFF, 0x40, 0x00, 0x00, 0x00, 1, 2, 3])
.unwrap();
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
assert_eq!(t.rows().unwrap(), 1, "the torn message contributes nothing");
t.explode(&oid(2), GitObjectKind::Blob, b"after").unwrap();
t.flush().unwrap();
assert_eq!(t.rows().unwrap(), 2);
assert_eq!(
t.content(&oid(2)).unwrap().unwrap().1,
b"after".to_vec(),
"the append landed at the last good boundary, not after the garbage"
);
}
#[test]
fn graph_keeps_commits_and_trees_and_drops_blobs() {
let p = ExplodePolicy::Graph;
assert!(p.wants(GitObjectKind::Commit));
assert!(p.wants(GitObjectKind::Tree));
assert!(!p.wants(GitObjectKind::Blob));
assert!(!ExplodePolicy::Off.wants(GitObjectKind::Commit));
assert!(ExplodePolicy::Full.wants(GitObjectKind::Blob));
assert_eq!(
ExplodePolicy::default(),
ExplodePolicy::Full,
"the default keeps §14's eager table whole, so `adopt_journal`'s \
row-count check keeps meaning what it did"
);
}
#[test]
fn a_declined_object_is_not_stored_and_is_counted() {
let t = ExplodedArchive::open_with_policy(&tmp("policy"), ExplodePolicy::Graph).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"bbbb").unwrap();
t.flush().unwrap();
assert!(t.has(&oid(1)).unwrap());
assert!(!t.has(&oid(2)).unwrap());
assert_eq!(t.skipped(), 1);
assert_eq!(t.rows().unwrap(), 1);
}
#[test]
fn off_stores_nothing_and_leaves_no_file() {
let p = tmp("off");
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Off).unwrap();
t.explode(&oid(1), GitObjectKind::Commit, b"c").unwrap();
t.explode(&oid(2), GitObjectKind::Blob, b"b").unwrap();
assert_eq!(t.flush().unwrap(), 0);
assert_eq!(t.rows().unwrap(), 0);
assert_eq!(t.skipped(), 2);
assert!(!p.exists(), "a disabled table must not create a file");
}
fn colliding_oids(groups: usize, per: usize) -> Vec<[u8; 20]> {
let mut out = Vec::with_capacity(groups * per);
for g in 0..groups {
let mut z = (g as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
let prefix = (z ^ (z >> 31)).to_be_bytes();
for k in 0..per {
let mut o = [0u8; 20];
o[..8].copy_from_slice(&prefix);
o[8] = k as u8;
o[9..13].copy_from_slice(&(g as u32).to_be_bytes());
out.push(o);
}
}
out
}
#[test]
fn the_stree_agrees_with_the_binary_search_it_replaced_on_colliding_prefixes() {
let t = ExplodedArchive::open_with_policy(&tmp("stree-agree"), ExplodePolicy::Full).unwrap();
let oids = colliding_oids(1000, 4);
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("payload-{i}").as_bytes())
.unwrap();
}
t.flush().unwrap();
t.index_now().unwrap();
let idx = t.index.lock().unwrap();
let tree = idx.as_ref().expect("index_now built it");
assert_eq!(tree.len(), oids.len());
let mut colliding_runs = 0usize;
for o in &oids {
let run = tree.candidate_run(key_for_oid(o));
assert_eq!(
run.len(),
4,
"expected a 4-entry candidate run for {}, got {run:?} — the collision \
the rest of this test rests on is not there",
hex::encode(o)
);
colliding_runs += 1;
}
assert_eq!(colliding_runs, oids.len());
assert!(
oids.iter().any(|o| o[0] >= 0x80) && oids.iter().any(|o| o[0] < 0x80),
"premise: the prefixes must straddle 0x80"
);
for o in &oids {
assert_eq!(
tree.find(o),
tree.find_by_binary_search(o),
"the stree and the binary search disagree at {}: {:?} vs {:?}",
hex::encode(o),
tree.find(o),
tree.find_by_binary_search(o)
);
assert!(tree.find(o).is_some(), "{} vanished", hex::encode(o));
}
for o in oids.iter().step_by(4) {
let mut absent = *o;
absent[8] = 0xff;
assert_eq!(tree.find(&absent), None, "{} was never stored", hex::encode(absent));
assert_eq!(tree.find_by_binary_search(&absent), None);
}
drop(idx);
for (i, o) in oids.iter().enumerate() {
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("payload-{i}").into_bytes(),
"{} came back with another object's bytes",
hex::encode(o)
);
}
}
#[test]
fn a_tombstoned_oid_stays_absent_when_a_live_oid_shares_its_prefix() {
let p = tmp("stree-tomb");
let oids = colliding_oids(64, 4);
let dead: Vec<[u8; 20]> = oids.iter().copied().step_by(4).collect();
{
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("live-{i}").as_bytes())
.unwrap();
}
t.flush().unwrap();
let retired = t.retain(&|o: &[u8]| !dead.iter().any(|d| d == o)).unwrap();
assert_eq!(retired, dead.len() as u64);
for d in &dead {
assert!(!t.has(d).unwrap(), "{} was retired", hex::encode(d));
assert!(
t.content(d).unwrap().is_none(),
"the retired oid {} was answered with a prefix-mate's row",
hex::encode(d)
);
}
for (i, o) in oids.iter().enumerate() {
if i % 4 == 0 {
continue;
}
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("live-{i}").into_bytes(),
"{} is alive and must still answer with its OWN bytes",
hex::encode(o)
);
}
}
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
for (i, o) in oids.iter().enumerate() {
if i % 4 == 0 {
assert!(
!t.has(o).unwrap(),
"a reopen resurrected a retired oid: {}",
hex::encode(o)
);
} else {
assert_eq!(
t.content(o).unwrap().unwrap().1,
format!("live-{i}").into_bytes(),
"a reopen lost or misrouted the live oid {}",
hex::encode(o)
);
}
}
assert_eq!(
t.of_kind(GitObjectKind::Blob).unwrap().len(),
oids.len() - dead.len()
);
assert_eq!(t.rows().unwrap(), (oids.len() - dead.len()) as u64);
}
#[test]
fn the_incremental_index_and_a_rebuilt_one_agree() {
let p = tmp("stree-incr");
let oids = colliding_oids(50, 4);
let t = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
t.index_now().unwrap();
for (i, o) in oids.iter().enumerate() {
t.explode(o, GitObjectKind::Blob, format!("v-{i}").as_bytes())
.unwrap();
if i % 37 == 0 {
t.flush().unwrap();
}
}
t.flush().unwrap();
t.retain(&|o: &[u8]| o[8] != 3).unwrap();
let live: Vec<(Vec<u8>, Vec<u8>)> = {
let idx = t.index.lock().unwrap();
idx.as_ref()
.unwrap()
.iter()
.map(|(o, l)| (o.to_vec(), vec![l.batch as u8, l.row as u8]))
.collect()
};
drop(t);
let again = ExplodedArchive::open_with_policy(&p, ExplodePolicy::Full).unwrap();
again.index_now().unwrap();
let rebuilt = again.index.lock().unwrap();
let rebuilt = rebuilt.as_ref().unwrap();
assert_eq!(rebuilt.len(), live.len(), "the two folds kept a different number of oids");
for (o, _) in &live {
assert!(
rebuilt.find(o).is_some(),
"the live process and a reopen disagree about {}: Some(..) vs None",
hex::encode(o)
);
}
for o in oids.iter().filter(|o| o[8] == 3) {
assert!(rebuilt.find(o).is_none(), "{} was retired", hex::encode(o));
}
}
#[test]
#[ignore = "measurement; run under --release with --ignored"]
fn lookup_cost_before_and_after() {
use std::time::Instant;
let n = 1_000_000usize;
let oids = colliding_oids(n / 4, 4);
let mut b = OidTreeBuilder::with_capacity(oids.len());
for (i, o) in oids.iter().enumerate() {
b.push(o, Some(Located { batch: (i / 100_000) as u32, row: i as u32, kind: GitObjectKind::Blob, pay_at: EXTENT_UNAVAILABLE, pay_len: 0 }))
.unwrap();
}
let build = Instant::now();
let tree = b.finish(Keep::Last);
let build = build.elapsed();
let mut old: Vec<(Vec<u8>, Located)> =
tree.iter().map(|(o, l)| (o.to_vec(), l)).collect();
old.sort_by(|a, b| a.0.cmp(&b.0));
let mut probes: Vec<[u8; 20]> = Vec::with_capacity(200_000);
for i in (0..oids.len()).step_by(oids.len() / 100_000) {
probes.push(oids[i]);
let mut m = oids[i];
m[8] = 0xfe;
probes.push(m);
}
let mut sink = 0u64;
let t0 = Instant::now();
for p in &probes {
if let Ok(i) = old.binary_search_by(|(o, _)| o.as_slice().cmp(&p[..])) {
sink += old[i].1.row as u64;
}
}
let vec_of_vec = t0.elapsed();
let t0 = Instant::now();
for p in &probes {
if let Some(l) = tree.find_by_binary_search(p) {
sink += l.row as u64;
}
}
let flat_bsearch = t0.elapsed();
let t0 = Instant::now();
for p in &probes {
if let Some(l) = tree.find(p) {
sink += l.row as u64;
}
}
let stree = t0.elapsed();
for p in &probes {
let want = tree.find_by_binary_search(p);
assert_eq!(tree.find(p), want, "arms disagree at {}", hex::encode(p));
}
let per = |d: std::time::Duration| d.as_secs_f64() * 1e9 / probes.len() as f64;
println!(
"exploded oid lookup, {} entries ({} groups of 4 colliding prefixes), \
{} probes half of them misses:\n \
Vec<(Vec<u8>,Located)> binary_search {:>7.1} ns/probe\n \
flat array binary_search {:>7.1} ns/probe\n \
stree {:>7.1} ns/probe\n \
tree build (sort + fold + stree) {:.3} s sink={sink}",
tree.len(),
oids.len() / 4,
probes.len(),
per(vec_of_vec),
per(flat_bsearch),
per(stree),
build.as_secs_f64(),
);
}
#[test]
fn an_unknown_policy_is_refused() {
assert_eq!(ExplodePolicy::parse("full").unwrap(), ExplodePolicy::Full);
let err = ExplodePolicy::parse("ful").expect_err("a typo must be refused");
assert!(format!("{err}").contains("unknown explode policy"), "{err}");
}
}