use crate::graph::storage::mapped::mmap_vec::{MmapOrVec, MmapPod};
use std::io;
pub(super) trait PackedElement: MmapPod {
const WIDTH: usize;
fn decode_le(bytes: &[u8]) -> Self;
fn write_le(self, writer: &mut dyn io::Write) -> io::Result<()>;
}
macro_rules! impl_packed_element {
($type:ty) => {
impl PackedElement for $type {
const WIDTH: usize = std::mem::size_of::<Self>();
fn decode_le(bytes: &[u8]) -> Self {
Self::from_le_bytes(bytes.try_into().expect("validated packed element width"))
}
fn write_le(self, writer: &mut dyn io::Write) -> io::Result<()> {
writer.write_all(&self.to_le_bytes())
}
}
};
}
impl_packed_element!(u32);
impl_packed_element!(u64);
impl_packed_element!(i32);
impl_packed_element!(i64);
impl_packed_element!(f64);
impl PackedElement for u8 {
const WIDTH: usize = 1;
fn decode_le(bytes: &[u8]) -> Self {
bytes[0]
}
fn write_le(self, writer: &mut dyn io::Write) -> io::Result<()> {
writer.write_all(&[self])
}
}
pub(super) fn write_packed_values<T: PackedElement>(
values: &MmapOrVec<T>,
writer: &mut impl io::Write,
) -> io::Result<()> {
if cfg!(target_endian = "little") {
values.write_to(writer)
} else {
for index in 0..values.len() {
values.get(index).write_le(writer)?;
}
Ok(())
}
}
pub(crate) const INT64_DELTA_TAG: &str = "int64d";
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum IntColumnEncoding {
Raw,
Auto,
}
#[inline]
fn zigzag(value: i64) -> u64 {
((value << 1) ^ (value >> 63)) as u64
}
#[inline]
fn unzigzag(value: u64) -> i64 {
((value >> 1) as i64) ^ -((value & 1) as i64)
}
fn push_varint(buf: &mut Vec<u8>, mut value: u64) {
loop {
let byte = (value & 0x7f) as u8;
value >>= 7;
if value == 0 {
buf.push(byte);
return;
}
buf.push(byte | 0x80);
}
}
fn read_varint(bytes: &[u8], cursor: &mut usize) -> io::Result<u64> {
let mut value: u64 = 0;
for shift in 0..10u32 {
let byte = *bytes.get(*cursor).ok_or_else(|| {
io::Error::new(
io::ErrorKind::InvalidData,
"delta-varint column ends mid-value",
)
})?;
*cursor += 1;
value |= u64::from(byte & 0x7f) << (shift * 7);
if byte & 0x80 == 0 {
return Ok(value);
}
}
Err(io::Error::new(
io::ErrorKind::InvalidData,
"delta-varint column has an over-long varint",
))
}
pub(crate) fn encode_int64_delta_if_smaller(
data: &MmapOrVec<i64>,
nulls: &MmapOrVec<u8>,
) -> Option<Vec<u8>> {
let rows = data.len();
if rows != nulls.len() {
return None;
}
let fixed_len = rows.checked_mul(std::mem::size_of::<i64>())?;
let mut buf: Vec<u8> = Vec::with_capacity(rows.saturating_mul(3));
let mut previous: i64 = 0;
for index in 0..rows {
let value = data.get(index);
push_varint(&mut buf, zigzag(value.wrapping_sub(previous)));
previous = value;
}
if buf.len() >= fixed_len {
return None;
}
for index in 0..rows {
buf.push(nulls.get(index));
}
Some(buf)
}
pub(crate) fn decode_int64_delta(blob: &[u8], rows: usize) -> io::Result<(Vec<u8>, &[u8])> {
if blob.len() < rows.saturating_mul(2) {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"delta-varint column has {} bytes; {rows} rows need at least {}",
blob.len(),
rows.saturating_mul(2)
),
));
}
let mut values: Vec<u8> = Vec::with_capacity(rows.saturating_mul(std::mem::size_of::<i64>()));
let mut cursor = 0usize;
let mut previous: i64 = 0;
for _ in 0..rows {
let value = previous.wrapping_add(unzigzag(read_varint(blob, &mut cursor)?));
values.extend_from_slice(&value.to_le_bytes());
previous = value;
}
let nulls = &blob[cursor..];
if nulls.len() != rows {
return Err(io::Error::new(
io::ErrorKind::InvalidData,
format!(
"delta-varint column has {} trailing null bytes; expected {rows}",
nulls.len()
),
));
}
Ok((values, nulls))
}
#[cfg(test)]
mod tests {
use super::*;
fn round_trip(values: &[i64], nulls: &[u8]) -> Option<Vec<i64>> {
let data = MmapOrVec::from_vec(values.to_vec());
let null_col = MmapOrVec::from_vec(nulls.to_vec());
let blob = encode_int64_delta_if_smaller(&data, &null_col)?;
let (bytes, decoded_nulls) = decode_int64_delta(&blob, values.len()).unwrap();
assert_eq!(decoded_nulls, nulls);
Some(
bytes
.as_chunks::<8>()
.0
.iter()
.map(|c| i64::from_le_bytes(*c))
.collect(),
)
}
#[test]
fn delta_round_trips_the_shapes_it_accepts() {
let monotonic: Vec<i64> = (0..1000).collect();
assert_eq!(
round_trip(&monotonic, &vec![0u8; 1000]).unwrap(),
monotonic,
"monotonic ids must survive the delta form"
);
let mixed_signs = vec![0, -1, i64::MAX, i64::MIN, 0, 7, -7];
let nulls = vec![0, 1, 0, 1, 0, 0, 1];
assert_eq!(
round_trip(&mixed_signs, &nulls).unwrap(),
mixed_signs,
"the extremes must survive the wrapping delta"
);
assert!(
encode_int64_delta_if_smaller(
&MmapOrVec::from_vec(Vec::<i64>::new()),
&MmapOrVec::from_vec(Vec::<u8>::new())
)
.is_none(),
"an empty column has nothing to gain and must stay fixed-width"
);
}
#[test]
fn incompressible_values_stay_on_the_fixed_width_form() {
let mut state: u64 = 0x2545_F491_4F6C_DD1D;
let hostile: Vec<i64> = (0..256)
.map(|_| {
state = state
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
state as i64
})
.collect();
assert!(
encode_int64_delta_if_smaller(
&MmapOrVec::from_vec(hostile),
&MmapOrVec::from_vec(vec![0u8; 256])
)
.is_none(),
"the delta form must not be chosen when it is larger"
);
}
#[test]
fn truncated_and_overlong_blobs_are_rejected() {
let data = MmapOrVec::from_vec((0..100i64).collect::<Vec<_>>());
let nulls = MmapOrVec::from_vec(vec![0u8; 100]);
let blob = encode_int64_delta_if_smaller(&data, &nulls).unwrap();
assert!(
decode_int64_delta(&blob[..blob.len() - 1], 100).is_err(),
"a short null run must not decode"
);
assert!(
decode_int64_delta(&blob, 101).is_err(),
"a row-count mismatch must not decode"
);
assert!(
decode_int64_delta(&[0x80u8; 12], 1).is_err(),
"an over-long varint must not decode"
);
assert!(
decode_int64_delta(&blob, usize::MAX / 4).is_err(),
"a row count the blob cannot possibly hold must be refused before \
anything is reserved for it"
);
}
}