use std::io;
use std::ops::ControlFlow;
use super::internal_key::{INTERNAL_KEY_SUFFIX_LEN, compare_internal_keys};
pub(crate) const RESTART_INTERVAL: usize = 16;
#[derive(Debug, Clone, Copy)]
pub(crate) struct BlockHandle {
pub(crate) offset: u64,
pub(crate) size: u64,
}
pub(crate) struct Block {
data: Vec<u8>,
restarts: Vec<u32>,
entries_end: usize,
}
impl Block {
#[cfg(test)]
pub(crate) fn decode(data: Vec<u8>) -> io::Result<Self> {
Self::decode_inner(data, false)
}
pub(crate) fn decode_data_block(data: Vec<u8>) -> io::Result<Self> {
Self::decode_inner(data, true)
}
fn decode_inner(data: Vec<u8>, require_internal_keys: bool) -> io::Result<Self> {
if data.len() < 4 {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"block too small",
));
}
let num_restarts = u32::from_le_bytes(data[data.len() - 4..].try_into().unwrap()) as usize;
if num_restarts == 0 {
return Err(invalid_data("block has no restart points"));
}
let restart_bytes = num_restarts
.checked_mul(4)
.ok_or_else(|| invalid_data("block restart array size overflows"))?;
let entries_end = data
.len()
.checked_sub(4)
.and_then(|len| len.checked_sub(restart_bytes))
.ok_or_else(|| invalid_data("block restart array exceeds block length"))?;
let mut restarts = Vec::with_capacity(num_restarts);
for i in 0..num_restarts {
let offset = entries_end + i * 4;
restarts.push(u32::from_le_bytes(
data[offset..offset + 4].try_into().unwrap(),
));
}
validate_restarts_and_entries(&data[..entries_end], &restarts, require_internal_keys)?;
Ok(Self {
data,
restarts,
entries_end,
})
}
pub(crate) fn charge(&self) -> usize {
std::mem::size_of::<Self>()
+ self.data.capacity()
+ self.restarts.capacity() * std::mem::size_of::<u32>()
}
pub(crate) fn entry_data(&self) -> &[u8] {
&self.data[..self.entries_end]
}
pub(crate) fn restart_count(&self) -> usize {
self.restarts.len()
}
pub(crate) fn restart_offset(&self, idx: usize) -> usize {
self.restarts[idx] as usize
}
pub(crate) fn entry_bytes(&self, offset: usize, len: usize) -> Option<&[u8]> {
let end = offset.checked_add(len)?;
if end > self.entries_end {
return None;
}
self.data.get(offset..end)
}
pub(crate) fn scan_from<F, R>(
&self,
target: &[u8],
key_buf: &mut Vec<u8>,
mut f: F,
) -> Option<R>
where
F: FnMut(&[u8], usize, usize) -> ControlFlow<R>,
{
let data = self.entry_data();
let mut pos = self.restart_start_for(target);
key_buf.clear();
let mut below_target = true;
while pos < data.len() {
let (consumed, value_offset, value_len) = decode_entry_at(data, pos, key_buf);
pos += consumed;
if below_target {
if compare_internal_keys(key_buf, target).is_lt() {
continue;
}
below_target = false;
}
if let ControlFlow::Break(result) = f(key_buf, value_offset, value_len) {
return Some(result);
}
}
None
}
#[cfg(test)]
pub(crate) fn iter(&self) -> BlockIterator<'_> {
BlockIterator {
data: self.entry_data(),
pos: 0,
current_key: Vec::new(),
lower_bound: None,
}
}
#[cfg(test)]
pub(crate) fn iter_from<'a>(&'a self, target: &'a [u8]) -> BlockIterator<'a> {
BlockIterator {
data: self.entry_data(),
pos: self.restart_start_for(target),
current_key: Vec::new(),
lower_bound: Some(target),
}
}
fn restart_start_for(&self, target: &[u8]) -> usize {
let data = self.entry_data();
if data.is_empty() {
return 0;
}
let mut left = 0;
let mut right = self.restarts.len();
while left < right {
let mid = left + (right - left) / 2;
let restart_pos = self.restarts[mid] as usize;
let header =
decode_entry_header(data, restart_pos).expect("block entry validated at decode");
debug_assert_eq!(header.shared, 0);
let key = &data[header.key_offset..header.key_offset + header.unshared];
if compare_internal_keys(key, target).is_lt() {
left = mid + 1;
} else {
right = mid;
}
}
if left > 0 {
self.restarts[left - 1] as usize
} else {
0
}
}
}
#[cfg(test)]
pub(crate) struct BlockIterator<'a> {
data: &'a [u8],
pos: usize,
current_key: Vec<u8>,
lower_bound: Option<&'a [u8]>,
}
#[cfg(test)]
impl<'a> Iterator for BlockIterator<'a> {
type Item = (Vec<u8>, Vec<u8>);
fn next(&mut self) -> Option<Self::Item> {
loop {
if self.pos >= self.data.len() {
return None;
}
let (consumed, val_off, val_len) =
decode_entry_at(self.data, self.pos, &mut self.current_key);
self.pos += consumed;
if let Some(target) = self.lower_bound {
if compare_internal_keys(&self.current_key, target).is_lt() {
continue;
}
self.lower_bound = None;
}
let value = self.data[val_off..val_off + val_len].to_vec();
return Some((self.current_key.clone(), value));
}
}
}
pub(crate) struct BlockBuilder {
buffer: Vec<u8>,
restarts: Vec<u32>,
entry_count: usize,
last_key: Vec<u8>,
restart_interval: usize,
}
impl BlockBuilder {
pub(crate) fn new(restart_interval: usize) -> Self {
Self {
buffer: Vec::new(),
restarts: vec![0], entry_count: 0,
last_key: Vec::new(),
restart_interval,
}
}
pub(crate) fn add(&mut self, key: &[u8], value: &[u8]) {
let shared =
if self.entry_count.is_multiple_of(self.restart_interval) && self.entry_count > 0 {
self.restarts.push(self.buffer.len() as u32);
0 } else {
self.last_key
.iter()
.zip(key.iter())
.take_while(|(a, b)| a == b)
.count()
};
let unshared = key.len() - shared;
encode_varint(&mut self.buffer, shared as u64);
encode_varint(&mut self.buffer, unshared as u64);
encode_varint(&mut self.buffer, value.len() as u64);
self.buffer.extend_from_slice(&key[shared..]);
self.buffer.extend_from_slice(value);
self.last_key = key.to_vec();
self.entry_count += 1;
}
pub(crate) fn finish(mut self) -> Vec<u8> {
for restart in &self.restarts {
self.buffer.extend_from_slice(&restart.to_le_bytes());
}
self.buffer
.extend_from_slice(&(self.restarts.len() as u32).to_le_bytes());
self.buffer
}
pub(crate) fn estimated_size(&self) -> usize {
self.buffer.len() + self.restarts.len() * 4 + 4
}
pub(crate) fn is_empty(&self) -> bool {
self.entry_count == 0
}
}
pub(crate) fn decode_entry_at(
data: &[u8],
pos: usize,
prev_key: &mut Vec<u8>,
) -> (usize, usize, usize) {
let header = decode_entry_header(data, pos).expect("block entry validated at decode");
assert!(
header.shared <= prev_key.len(),
"block entry validated at decode"
);
let key_end = header.key_offset + header.unshared;
prev_key.truncate(header.shared);
prev_key.extend_from_slice(&data[header.key_offset..key_end]);
(header.consumed, key_end, header.value_len)
}
pub(crate) fn encoded_entry_size(data: &[u8], pos: usize) -> usize {
decode_entry_header(data, pos)
.expect("block entry validated at decode")
.consumed
}
fn encode_varint(buf: &mut Vec<u8>, mut value: u64) {
while value >= 0x80 {
buf.push((value as u8) | 0x80);
value >>= 7;
}
buf.push(value as u8);
}
#[cfg(test)]
pub(crate) fn decode_varint(data: &[u8]) -> (u64, usize) {
let mut result: u64 = 0;
for (i, &byte) in data.iter().take(10).enumerate() {
let shift = i * 7;
result |= ((byte & 0x7F) as u64) << shift;
if byte & 0x80 == 0 {
return (result, i + 1);
}
}
(result, data.len().min(10))
}
fn invalid_data(message: &'static str) -> io::Error {
io::Error::new(io::ErrorKind::InvalidData, message)
}
struct DecodedEntry {
key: Vec<u8>,
consumed: usize,
}
struct EntryHeader {
shared: usize,
unshared: usize,
value_len: usize,
key_offset: usize,
consumed: usize,
}
fn validate_restarts_and_entries(
data: &[u8],
restarts: &[u32],
require_internal_keys: bool,
) -> io::Result<()> {
let mut entry_offsets = Vec::new();
let mut pos = 0usize;
let mut prev_key = Vec::new();
while pos < data.len() {
entry_offsets.push(pos);
let entry = decode_block_entry_checked(data, pos, &prev_key)?;
if require_internal_keys && entry.key.len() < INTERNAL_KEY_SUFFIX_LEN {
return Err(invalid_data(
"block entry key is shorter than the internal-key trailer",
));
}
pos += entry.consumed;
prev_key = entry.key;
}
if restarts.first().copied() != Some(0) {
return Err(invalid_data("block first restart offset must be zero"));
}
let mut previous_restart = 0usize;
for (idx, &restart) in restarts.iter().enumerate() {
let restart = restart as usize;
if idx > 0 && restart <= previous_restart {
return Err(invalid_data(
"block restart offsets are not strictly increasing",
));
}
previous_restart = restart;
if data.is_empty() {
if restart != 0 {
return Err(invalid_data("empty block restart offset must be zero"));
}
continue;
}
if entry_offsets.binary_search(&restart).is_err() {
return Err(invalid_data(
"block restart offset is not an entry boundary",
));
}
if decode_entry_header(data, restart)?.shared != 0 {
return Err(invalid_data(
"block restart entry must not share a key prefix",
));
}
}
Ok(())
}
fn decode_block_entry_checked(
data: &[u8],
pos: usize,
prev_key: &[u8],
) -> io::Result<DecodedEntry> {
let header = decode_entry_header(data, pos)?;
if header.shared > prev_key.len() {
return Err(invalid_data(
"block entry shared prefix exceeds previous key",
));
}
let key_len = header
.shared
.checked_add(header.unshared)
.ok_or_else(|| invalid_data("block entry key length overflows"))?;
let mut key = Vec::with_capacity(key_len);
key.extend_from_slice(&prev_key[..header.shared]);
key.extend_from_slice(&data[header.key_offset..header.key_offset + header.unshared]);
Ok(DecodedEntry {
key,
consumed: header.consumed,
})
}
fn decode_entry_header(data: &[u8], pos: usize) -> io::Result<EntryHeader> {
if pos >= data.len() {
return Err(invalid_data("block entry offset is outside entry data"));
}
let mut offset = pos;
let shared = read_varint(data, &mut offset)?;
let unshared = read_varint(data, &mut offset)?;
let value_len = read_varint(data, &mut offset)?;
let key_end = offset
.checked_add(unshared)
.ok_or_else(|| invalid_data("block entry key length overflows"))?;
if key_end > data.len() {
return Err(invalid_data("block entry key extends past block"));
}
let value_end = key_end
.checked_add(value_len)
.ok_or_else(|| invalid_data("block entry value length overflows"))?;
if value_end > data.len() {
return Err(invalid_data("block entry value extends past block"));
}
Ok(EntryHeader {
shared,
unshared,
value_len,
key_offset: offset,
consumed: value_end - pos,
})
}
fn read_varint(data: &[u8], offset: &mut usize) -> io::Result<usize> {
let mut result: u64 = 0;
for i in 0..10 {
let Some(&byte) = data.get(*offset) else {
return Err(invalid_data("unterminated block entry varint"));
};
*offset += 1;
if i == 9 && byte > 1 {
return Err(invalid_data("block entry varint overflows u64"));
}
result |= ((byte & 0x7F) as u64) << (i * 7);
if byte & 0x80 == 0 {
return usize::try_from(result)
.map_err(|_| invalid_data("block entry varint overflows usize"));
}
}
Err(invalid_data("unterminated block entry varint"))
}
#[cfg(test)]
mod tests {
use super::*;
fn scan_pairs(block: &Block, target: &[u8]) -> Vec<(Vec<u8>, Vec<u8>)> {
let mut out = Vec::new();
let mut key_buf = Vec::new();
let done: Option<()> = block.scan_from(target, &mut key_buf, |key, off, len| {
out.push((
key.to_vec(),
block
.entry_bytes(off, len)
.expect("value in range")
.to_vec(),
));
ControlFlow::Continue(())
});
assert!(done.is_none(), "a Continue-only scan runs to the end");
out
}
#[test]
fn scan_from_matches_the_copying_iterator() {
let pairs: Vec<(Vec<u8>, Vec<u8>)> = (0..64)
.map(|i| {
(
format!("key_{i:04}").into_bytes(),
format!("value_{i}").into_bytes(),
)
})
.collect();
let refs: Vec<(&[u8], &[u8])> = pairs
.iter()
.map(|(k, v)| (k.as_slice(), v.as_slice()))
.collect();
let block = build_block(&refs);
for target in [
b"".as_ref(),
b"key_0000",
b"key_0001",
b"key_0016",
b"key_0017_suffix",
b"key_0063",
b"zzzz",
] {
let expected: Vec<_> = block.iter_from(target).collect();
assert_eq!(scan_pairs(&block, target), expected, "target {target:?}");
}
}
#[test]
fn scan_from_reuses_the_key_buffer_across_calls() {
let block = build_block(&[(b"aaaa", b"1"), (b"aaab", b"2"), (b"bbbb", b"3")]);
let mut key_buf = Vec::new();
for target in [b"aaaa".as_ref(), b"bbbb", b"aaab"] {
let first: Option<Vec<u8>> = block.scan_from(target, &mut key_buf, |key, _, _| {
ControlFlow::Break(key.to_vec())
});
assert_eq!(first.as_deref(), Some(target));
}
}
#[test]
fn entry_bytes_rejects_a_range_past_the_entry_region() {
let block = build_block(&[(b"k", b"v")]);
assert!(block.entry_bytes(0, block.entry_data().len()).is_some());
assert!(block.entry_bytes(0, block.entry_data().len() + 1).is_none());
assert!(block.entry_bytes(usize::MAX, 1).is_none());
}
fn build_block(pairs: &[(&[u8], &[u8])]) -> Block {
let mut builder = BlockBuilder::new(4);
for (k, v) in pairs {
builder.add(k, v);
}
Block::decode(builder.finish()).unwrap()
}
fn expect_block_decode_err(data: Vec<u8>) -> io::Error {
match Block::decode(data) {
Err(e) => e,
Ok(_) => panic!("expected invalid block"),
}
}
#[test]
fn test_varint_roundtrip() {
let test_values = [0u64, 1, 127, 128, 16383, 16384, u64::MAX];
for val in test_values {
let mut buf = Vec::new();
encode_varint(&mut buf, val);
let (decoded, _) = decode_varint(&buf);
assert_eq!(val, decoded);
}
}
#[test]
fn varint_zero_and_127_fit_in_one_byte() {
for v in [0u64, 42, 127] {
let mut buf = Vec::new();
encode_varint(&mut buf, v);
assert_eq!(buf.len(), 1, "value {v} should encode to 1 byte");
assert_eq!(decode_varint(&buf), (v, 1));
}
}
#[test]
fn varint_128_crosses_into_two_bytes() {
let mut buf = Vec::new();
encode_varint(&mut buf, 128);
assert_eq!(buf, vec![0x80, 0x01]);
assert_eq!(decode_varint(&buf), (128, 2));
}
#[test]
fn varint_u64_max_fits_in_ten_bytes() {
let mut buf = Vec::new();
encode_varint(&mut buf, u64::MAX);
assert_eq!(buf.len(), 10);
assert_eq!(decode_varint(&buf), (u64::MAX, 10));
}
#[test]
fn decode_varint_does_not_panic_on_long_unterminated_input() {
let buf = vec![0x80; 32];
assert_eq!(decode_varint(&buf).1, 10);
}
#[test]
fn empty_block_iterates_nothing() {
let builder = BlockBuilder::new(16);
let block = Block::decode(builder.finish()).unwrap();
assert_eq!(block.iter().count(), 0);
}
#[test]
fn single_entry_block_round_trips() {
let block = build_block(&[(b"only", b"one")]);
let entries: Vec<_> = block.iter().collect();
assert_eq!(entries, vec![(b"only".to_vec(), b"one".to_vec())]);
}
#[test]
fn iter_preserves_insert_order() {
let pairs: Vec<(Vec<u8>, Vec<u8>)> = (0..20)
.map(|i| {
(
format!("key_{:04}", i).into_bytes(),
format!("val_{}", i).into_bytes(),
)
})
.collect();
let mut b = BlockBuilder::new(4);
for (k, v) in &pairs {
b.add(k, v);
}
let block = Block::decode(b.finish()).unwrap();
let got: Vec<_> = block.iter().collect();
assert_eq!(got, pairs);
}
#[test]
fn iter_from_starts_at_first_entry_ge_target() {
let pairs: Vec<(Vec<u8>, Vec<u8>)> = (0..20)
.map(|i| {
(
format!("key_{:04}", i).into_bytes(),
format!("val_{}", i).into_bytes(),
)
})
.collect();
let mut b = BlockBuilder::new(4);
for (k, v) in &pairs {
b.add(k, v);
}
let block = Block::decode(b.finish()).unwrap();
let got: Vec<_> = block.iter_from(b"key_0010").collect();
assert_eq!(got, pairs[10..].to_vec());
let got: Vec<_> = block.iter_from(b"key_0010_suffix").collect();
assert_eq!(got, pairs[11..].to_vec());
assert_eq!(block.iter_from(b"zzzz").next(), None);
}
#[test]
fn prefix_compression_round_trips_shared_prefix_keys() {
let block = build_block(&[(b"zebra_a", b"1"), (b"zebra_b", b"2"), (b"zebra_c", b"3")]);
let got: Vec<_> = block.iter().collect();
assert_eq!(
got,
vec![
(b"zebra_a".to_vec(), b"1".to_vec()),
(b"zebra_b".to_vec(), b"2".to_vec()),
(b"zebra_c".to_vec(), b"3".to_vec()),
]
);
}
#[test]
fn restart_points_occur_every_interval_entries() {
let n = 32usize;
let mut b = BlockBuilder::new(8);
for i in 0..n {
let key = format!("k_{:04}", i);
b.add(key.as_bytes(), b"v");
}
let block = Block::decode(b.finish()).unwrap();
assert_eq!(block.restart_count(), 4);
assert_eq!(block.restart_offset(0), 0);
}
#[test]
fn decode_rejects_buffer_smaller_than_footer() {
assert!(Block::decode(vec![]).is_err());
assert!(Block::decode(vec![0u8; 3]).is_err());
}
#[test]
fn decode_rejects_restart_array_that_exceeds_block() {
let mut data = vec![0u8; 4];
data.extend_from_slice(&2u32.to_le_bytes());
let err = expect_block_decode_err(data);
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
#[test]
fn decode_rejects_restart_offset_outside_entry_data() {
let mut builder = BlockBuilder::new(16);
builder.add(b"k", b"v");
let mut data = builder.finish();
let restart_offset = data.len() - 8;
data[restart_offset] = 0xFF;
let err = expect_block_decode_err(data);
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
#[test]
fn decode_rejects_first_restart_offset_that_is_not_zero() {
let mut data = Vec::new();
encode_varint(&mut data, 0);
encode_varint(&mut data, 1);
encode_varint(&mut data, 0);
data.push(b'a');
let second_offset = data.len() as u32;
encode_varint(&mut data, 0);
encode_varint(&mut data, 1);
encode_varint(&mut data, 0);
data.push(b'b');
data.extend_from_slice(&second_offset.to_le_bytes());
data.extend_from_slice(&1u32.to_le_bytes());
let err = expect_block_decode_err(data);
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
#[test]
fn decode_rejects_truncated_entry_payload() {
let mut data = Vec::new();
encode_varint(&mut data, 0);
encode_varint(&mut data, 2);
encode_varint(&mut data, 1);
data.push(b'a');
data.extend_from_slice(&0u32.to_le_bytes());
data.extend_from_slice(&1u32.to_le_bytes());
let err = expect_block_decode_err(data);
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
#[test]
fn decode_rejects_shared_prefix_at_restart() {
let mut data = Vec::new();
encode_varint(&mut data, 0);
encode_varint(&mut data, 1);
encode_varint(&mut data, 0);
data.push(b'a');
let second_offset = data.len() as u32;
encode_varint(&mut data, 1);
encode_varint(&mut data, 1);
encode_varint(&mut data, 0);
data.push(b'b');
data.extend_from_slice(&0u32.to_le_bytes());
data.extend_from_slice(&second_offset.to_le_bytes());
data.extend_from_slice(&2u32.to_le_bytes());
let err = expect_block_decode_err(data);
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
}
#[test]
fn charge_scales_with_data_bytes() {
let small = build_block(&[(b"k", b"v")]);
let big_pairs: Vec<(Vec<u8>, Vec<u8>)> = (0..500)
.map(|i| (format!("key_{:04}", i).into_bytes(), vec![0u8; 128]))
.collect();
let mut b = BlockBuilder::new(16);
for (k, v) in &big_pairs {
b.add(k, v);
}
let big = Block::decode(b.finish()).unwrap();
assert!(big.charge() > small.charge() * 100);
}
#[test]
fn builder_is_empty_flag_reflects_entry_count() {
let mut b = BlockBuilder::new(16);
assert!(b.is_empty());
b.add(b"k", b"v");
assert!(!b.is_empty());
}
#[test]
fn builder_estimated_size_grows_monotonically() {
let mut b = BlockBuilder::new(16);
let s0 = b.estimated_size();
b.add(b"a", b"1");
let s1 = b.estimated_size();
b.add(b"b", b"2");
let s2 = b.estimated_size();
assert!(s1 >= s0);
assert!(s2 >= s1);
}
#[test]
fn entry_data_length_matches_iter_consumption() {
let pairs: Vec<(&[u8], &[u8])> = vec![(b"abc", b"1"), (b"abcd", b"22"), (b"xyz", b"333")];
let mut b = BlockBuilder::new(16);
for (k, v) in &pairs {
b.add(k, v);
}
let block = Block::decode(b.finish()).unwrap();
let data_len = block.entry_data().len();
let mut consumed = 0usize;
let mut pos = 0usize;
let data = block.entry_data();
while pos < data_len {
let size = encoded_entry_size(data, pos);
pos += size;
consumed += size;
}
assert_eq!(consumed, data_len);
}
}
#[cfg(test)]
mod scan_properties {
use super::*;
use proptest::prelude::*;
proptest! {
#[test]
fn scan_from_agrees_with_iter_from(
mut keys in proptest::collection::vec(
proptest::collection::vec(any::<u8>(), 1..12), 1..48),
target in proptest::collection::vec(any::<u8>(), 0..12),
) {
keys.sort();
keys.dedup();
let mut builder = BlockBuilder::new(RESTART_INTERVAL);
for (i, key) in keys.iter().enumerate() {
builder.add(key, format!("v{i}").as_bytes());
}
let block = Block::decode(builder.finish())?;
let expected: Vec<(Vec<u8>, Vec<u8>)> = block.iter_from(&target).collect();
let mut got = Vec::new();
let mut key_buf = Vec::new();
let stopped: Option<()> = block.scan_from(&target, &mut key_buf, |key, off, len| {
let value = block.entry_bytes(off, len).expect("value in range");
got.push((key.to_vec(), value.to_vec()));
ControlFlow::Continue(())
});
prop_assert!(stopped.is_none());
prop_assert_eq!(got, expected);
}
}
}