use crate::table::zone_map::ColumnStats;
use crate::{Error, Result, Slice, ValueType, key::InternalKey, value::InternalValue};
use alloc::vec::Vec;
pub(crate) use super::column_type::comparable_bytes;
pub use super::column_type::{ByteOrder, Number, NumberKind, TypeTag};
mod cells;
mod expr;
pub(crate) use cells::{cell_refs, entries_to_cells_batch};
pub(crate) use expr::{Bounds, Cell, Choice, Values};
pub use expr::{Candidate, Expression, candidates};
const MAX_VIEW_AMPLIFICATION: usize = 2;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Column {
pub column_id: u16,
pub type_tag: TypeTag,
pub validity: Option<Vec<u8>>,
pub data: crate::Slice,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ColumnBatch {
pub row_count: u32,
pub columns: Vec<Column>,
}
const fn validity_len(row_count: u32) -> usize {
(row_count as usize).div_ceil(8)
}
fn check_validity(v: &[u8], row_count: u32) -> Result<()> {
if v.len() != validity_len(row_count) {
return Err(Error::InvalidHeader(
"columnar: validity bitmap length is not ceil(row_count / 8)",
));
}
let used = row_count % 8;
if used != 0 {
let last = v.last().copied().unwrap_or(0);
let valid_mask = (1u8 << used) - 1;
if last & !valid_mask != 0 {
return Err(Error::InvalidHeader(
"columnar: validity padding bits above row_count must be zero",
));
}
}
Ok(())
}
fn check_bytes_framing(data: &[u8], row_count: u32) -> Result<()> {
let Some(off_bytes) = (row_count as usize)
.checked_add(1)
.and_then(|count| count.checked_mul(4))
else {
return Err(Error::InvalidHeader(
"columnar: bytes offset table overflow",
));
};
let Some(table) = data.get(..off_bytes) else {
return Err(Error::InvalidHeader(
"columnar: bytes column shorter than its offset table",
));
};
let payload_len = data.len() - off_bytes;
let mut prev = 0usize;
for (i, chunk) in table.chunks_exact(4).enumerate() {
let Some(chunk) = chunk.first_chunk::<4>() else {
return Err(Error::InvalidHeader("columnar: short bytes offset"));
};
let off = u32::from_le_bytes(*chunk) as usize;
if i == 0 && off != 0 {
return Err(Error::InvalidHeader(
"columnar: first bytes offset must be zero",
));
}
if off < prev {
return Err(Error::InvalidHeader(
"columnar: bytes offsets must be non-decreasing",
));
}
if off > payload_len {
return Err(Error::InvalidHeader(
"columnar: bytes offset past payload end",
));
}
prev = off;
}
if prev != payload_len {
return Err(Error::InvalidHeader(
"columnar: final bytes offset must equal the payload length",
));
}
Ok(())
}
fn check_layout(type_tag: TypeTag, row_count: u32, data: &[u8]) -> Result<()> {
match type_tag.fixed_width() {
Some(0) => Err(Error::InvalidHeader("columnar: fixed column width is zero")),
Some(w) => {
let Some(expected) = (row_count as usize).checked_mul(w as usize) else {
return Err(Error::InvalidHeader(
"columnar: fixed column length overflow",
));
};
if data.len() != expected {
return Err(Error::InvalidHeader(
"columnar: fixed column byte length is not row_count * width",
));
}
Ok(())
}
None => check_bytes_framing(data, row_count),
}
}
impl Column {
pub(crate) fn is_valid(&self, row: u32) -> bool {
self.validity.as_deref().is_none_or(|bits| {
bits.get(row as usize / 8)
.is_some_and(|byte| byte >> (row % 8) & 1 == 1)
})
}
pub(crate) fn validate(&self, row_count: u32) -> Result<()> {
check_layout(self.type_tag, row_count, &self.data)?;
if let Some(v) = &self.validity {
check_validity(v, row_count)?;
}
Ok(())
}
fn encode_into(
&self,
row_count: u32,
encoding: crate::config::ColumnEncoding,
out: &mut Vec<u8>,
) -> Result<Expression> {
self.validate(row_count)?;
let Choice {
bytes, expression, ..
} = expr::choose(self.type_tag, row_count, &self.data, encoding)?;
let (type_tag, width) = self.type_tag.to_wire();
out.extend_from_slice(&self.column_id.to_le_bytes());
out.push(type_tag);
out.push(width);
out.push(u8::from(self.validity.is_some()));
crate::table::column_page::put_varint(out, bytes.len() as u64);
if let Some(v) = &self.validity {
out.extend_from_slice(v);
}
out.extend_from_slice(&bytes);
Ok(expression)
}
pub(crate) fn bytes_range(
&self,
row_count: u32,
start: u32,
end: u32,
) -> Result<NullsAndRange<'_>> {
if self.type_tag != TypeTag::Bytes {
return Err(Error::InvalidHeader(
"columnar: a zone describes a bytes column",
));
}
if start > end || end > row_count {
return Err(Error::InvalidHeader(
"columnar: row range outside the column",
));
}
let mut nulls = 0u32;
let mut range: Option<(&[u8], &[u8])> = None;
for row in start..end {
if !column_row_valid(self, row) {
nulls += 1;
continue;
}
let value = bytes_column_row(&self.data, row_count, row)?;
range = Some(match range {
None => (value, value),
Some((min, max)) => (min.min(value), max.max(value)),
});
}
Ok((nulls, range))
}
pub(crate) fn number_range(&self, start: u32, end: u32) -> Result<(u32, Option<(u128, u128)>)> {
let TypeTag::Number(number) = self.type_tag else {
return Err(Error::InvalidHeader(
"columnar: a number range describes a number column",
));
};
let width = usize::from(number.width);
let Some(cells) = self.data.get(start as usize * width..end as usize * width) else {
return Err(Error::InvalidHeader(
"columnar: row range outside the column",
));
};
let mut nulls = 0u32;
let mut range: Option<(u128, u128)> = None;
for (row, cell) in (start..end).zip(cells.chunks_exact(width)) {
if !column_row_valid(self, row) {
nulls += 1;
continue;
}
let value = number.ordinal(cell);
range = Some(match range {
None => (value, value),
Some((min, max)) => (min.min(value), max.max(value)),
});
}
Ok((nulls, range))
}
pub(crate) fn rows(&self, row_count: u32, start: u32, end: u32) -> Result<Self> {
if start > end || end > row_count {
return Err(Error::InvalidHeader(
"columnar: row range outside the column",
));
}
let data = if let Some(w) = self.type_tag.fixed_width() {
let w = usize::from(w);
let (from, to) = (start as usize * w, end as usize * w);
if to > self.data.len() {
return Err(Error::InvalidHeader(
"columnar: fixed column shorter than its rows",
));
}
self.data.slice(from..to)
} else {
let cells = || (start..end).map(|i| bytes_column_row(&self.data, row_count, i));
for cell in cells() {
cell?;
}
frame_bytes_column((end - start) as usize, || {
cells().map(Result::unwrap_or_default)
})?
};
let validity = self.validity.as_deref().map(|v| {
let rows = end - start;
let mut out = alloc::vec![0u8; validity_len(rows)];
for i in 0..rows {
if validity_bit(v, start + i)
&& let Some(byte) = out.get_mut((i / 8) as usize)
{
*byte |= 1u8 << (i % 8);
}
}
out
});
Ok(Self {
column_id: self.column_id,
type_tag: self.type_tag,
validity,
data,
})
}
pub(crate) fn encode_page(
&self,
row_count: u32,
stamp: crate::table::column_page::PageStamp,
encoding: crate::config::ColumnEncoding,
) -> Result<Vec<u8>> {
let mut out = Vec::new();
stamp.encode_into(&mut out);
self.encode_into(row_count, encoding, &mut out)?;
Ok(out)
}
pub(crate) fn parse_page(
bytes: &crate::Slice,
row_count: u32,
expected: crate::table::column_page::PageStamp,
) -> Result<PageColumn<'_>> {
use crate::table::column_page::PageStamp;
let mut cur = Cursor::new(bytes);
if PageStamp::decode(cur.read_array::<{ PageStamp::LEN }>()?) != expected {
return Err(Error::InvalidHeader(
"columnar: page belongs to another row group, column part or row page",
));
}
let raw = RawColumn::read(&mut cur, row_count)?;
if !cur.is_empty() {
return Err(Error::InvalidHeader(
"columnar: trailing bytes after the page's column",
));
}
raw.parse(row_count, bytes)
}
pub(crate) fn decode_page(
bytes: &crate::Slice,
row_count: u32,
expected: crate::table::column_page::PageStamp,
copied: &mut usize,
budget: &mut DecodeBudget,
) -> Result<Self> {
Self::parse_page(bytes, row_count, expected)?.decode(row_count, copied, budget)
}
fn decode_from(
cur: &mut Cursor<'_>,
bytes: &crate::Slice,
row_count: u32,
want: impl Fn(u16) -> bool,
copied: &mut usize,
budget: &mut DecodeBudget,
) -> Result<Option<(Self, usize)>> {
let raw = RawColumn::read(cur, row_count)?;
if !want(raw.column_id) {
return Ok(None);
}
let column = raw.parse(row_count, bytes)?;
let viewed = match column.values {
Values::Plain(data) => data.len(),
_ => 0,
};
Ok(Some((column.decode(row_count, copied, budget)?, viewed)))
}
}
struct RawColumn<'a> {
column_id: u16,
type_tag: TypeTag,
validity: Option<&'a [u8]>,
values: &'a [u8],
}
impl<'a> RawColumn<'a> {
fn read(cur: &mut Cursor<'a>, row_count: u32) -> Result<Self> {
let column_id = cur.read_u16()?;
let type_tag = cur.read_u8()?;
let width = cur.read_u8()?;
let has_validity = match cur.read_u8()? {
0 => false,
1 => true,
_ => {
return Err(Error::InvalidHeader(
"columnar: validity flag must be 0 or 1",
));
}
};
let values_len = cur.read_var_u32()? as usize;
let type_tag = TypeTag::from_wire(type_tag, width)?;
let validity = if has_validity {
Some(cur.read_bytes(validity_len(row_count))?)
} else {
None
};
Ok(Self {
column_id,
type_tag,
validity,
values: cur.read_bytes(values_len)?,
})
}
fn parse(self, row_count: u32, page: &'a crate::Slice) -> Result<PageColumn<'a>> {
if let Some(v) = self.validity {
check_validity(v, row_count)?;
}
Ok(PageColumn {
page,
column_id: self.column_id,
type_tag: self.type_tag,
validity: self.validity,
values: Values::parse(self.type_tag, row_count, self.values)?,
})
}
}
pub(crate) struct PageColumn<'a> {
page: &'a crate::Slice,
pub(crate) column_id: u16,
pub(crate) type_tag: TypeTag,
pub(crate) validity: Option<&'a [u8]>,
pub(crate) values: Values<'a>,
}
impl PageColumn<'_> {
pub(crate) fn decode(
self,
row_count: u32,
copied: &mut usize,
budget: &mut DecodeBudget,
) -> Result<Column> {
let built = !matches!(self.values, Values::Plain(_));
let data = budget.build(self.page.len(), |limit| {
self.values
.materialize(self.type_tag, row_count, self.page, limit)
})?;
if built {
*copied += data.len();
}
let validity = self.validity.map(|v| {
*copied += v.len();
v.to_vec()
});
let column = Column {
column_id: self.column_id,
type_tag: self.type_tag,
validity,
data,
};
column.validate(row_count)?;
Ok(column)
}
pub(crate) fn select(
&self,
row_count: u32,
bounds: &Bounds<'_>,
) -> Result<crate::table::columnar_predicate::Selection> {
let mut kept = self.values.select(self.type_tag, row_count, bounds)?;
if let Some(validity) = self.validity {
kept.intersect(&crate::table::columnar_predicate::Selection::from_bitmap(
validity, row_count,
));
}
Ok(kept)
}
pub(crate) fn decode_rows(
self,
row_count: u32,
keep: &crate::table::columnar_predicate::Selection,
copied: &mut usize,
budget: &mut DecodeBudget,
) -> Result<Column> {
let rows = keep.count();
let (type_tag, values) = (self.type_tag, self.values);
let data = budget.build(self.page.len(), |limit| {
values.materialize_rows(type_tag, row_count, keep, limit)
})?;
*copied += data.len();
let validity = self.validity.map(|v| {
let mut out = alloc::vec![0u8; validity_len(rows)];
for (at, row) in keep.rows().enumerate() {
if validity_bit(v, row)
&& let Some(byte) = out.get_mut(at / 8)
{
*byte |= 1u8 << (at % 8);
}
}
*copied += out.len();
out
});
let column = Column {
column_id: self.column_id,
type_tag: self.type_tag,
validity,
data,
};
column.validate(rows)?;
Ok(column)
}
}
#[derive(Debug)]
pub(crate) struct DecodeBudget {
allowance: u64,
built: u64,
held: u64,
}
impl Default for DecodeBudget {
fn default() -> Self {
Self {
allowance: 2 * u64::from(crate::config::MAX_BLOCK_SIZE),
built: 0,
held: 0,
}
}
}
impl DecodeBudget {
fn unbounded() -> Self {
Self {
allowance: u64::MAX / 2,
built: 0,
held: 0,
}
}
pub(crate) fn build(
&mut self,
page_len: usize,
build: impl FnOnce(u64) -> Result<Slice>,
) -> Result<Slice> {
debug_assert!(self.built <= self.allowance, "charged within the allowance");
let page_len = page_len as u64;
let limit = page_len + (self.allowance - self.built);
let data = build(limit)?;
let len = data.len() as u64;
debug_assert!(len <= limit, "a build refuses a column past its limit");
if len > page_len {
self.built += len - page_len;
}
Ok(data)
}
pub(crate) fn hold(&mut self, rows: u64) -> Result<()> {
let held = self.held + rows;
if held > self.allowance {
return Err(Error::DecompressedSizeTooLarge {
declared: held,
limit: self.allowance,
});
}
self.held = held;
Ok(())
}
pub(crate) fn rows<'a>(
&mut self,
values: Values<'a>,
type_tag: TypeTag,
n: u32,
) -> Result<expr::Rows<'a>> {
if !matches!(values, Values::Plain(_) | Values::Constant(_)) {
self.hold(u64::from(n))?;
}
values.rows(type_tag, n)
}
pub(crate) fn parsed<'a>(&mut self, column: PageColumn<'a>) -> Result<PageColumn<'a>> {
self.hold(column.values.held_rows())?;
Ok(column)
}
}
impl ColumnBatch {
pub(crate) fn row_page_cuts(&self, page_size: u32) -> Result<Vec<u32>> {
let mut cuts = Vec::new();
let mut rows_in_page = 0u32;
let mut bytes_in_page = 0u64;
for row in 0..self.row_count {
bytes_in_page += self.row_bytes(row)?;
rows_in_page += 1;
if bytes_in_page >= u64::from(page_size) {
cuts.push(rows_in_page);
rows_in_page = 0;
bytes_in_page = 0;
}
}
if rows_in_page > 0 {
cuts.push(rows_in_page);
}
Ok(cuts)
}
fn row_bytes(&self, row: u32) -> Result<u64> {
let mut bytes = 0u64;
for col in &self.columns {
bytes += match col.type_tag.fixed_width() {
Some(w) => u64::from(w),
None => bytes_column_row(&col.data, self.row_count, row)?.len() as u64 + 4,
};
}
Ok(bytes)
}
pub(crate) fn rows_bytes(&self) -> Result<u64> {
(0..self.row_count).try_fold(0u64, |sum, row| Ok(sum + self.row_bytes(row)?))
}
pub(crate) fn rows(&self, start: u32, end: u32) -> Result<Self> {
Ok(Self {
row_count: end.checked_sub(start).ok_or(Error::InvalidHeader(
"columnar: row range outside the batch",
))?,
columns: self
.columns
.iter()
.map(|col| col.rows(self.row_count, start, end))
.collect::<Result<_>>()?,
})
}
pub(crate) fn first_user_key(&self) -> Result<Option<&[u8]>> {
if self.row_count == 0 {
return Ok(None);
}
let Some(key_col) = self.columns.first().filter(|c| c.column_id == COL_USER_KEY) else {
return Err(Error::InvalidHeader(
"columnar: first column is not the user-key column",
));
};
if key_col.type_tag != TypeTag::Bytes || key_col.validity.is_some() {
return Err(Error::InvalidHeader(
"columnar: first column is not the non-null user-key column",
));
}
key_col.validate(self.row_count)?;
bytes_column_row(&key_col.data, self.row_count, 0).map(Some)
}
pub(crate) fn last_user_key(&self) -> Result<Option<&[u8]>> {
let Some(last) = self.row_count.checked_sub(1) else {
return Ok(None);
};
let Some(key_col) = self.columns.first().filter(|c| c.column_id == COL_USER_KEY) else {
return Err(Error::InvalidHeader(
"columnar: first column is not the user-key column",
));
};
if key_col.type_tag != TypeTag::Bytes || key_col.validity.is_some() {
return Err(Error::InvalidHeader(
"columnar: first column is not the non-null user-key column",
));
}
key_col.validate(self.row_count)?;
bytes_column_row(&key_col.data, self.row_count, last).map(Some)
}
#[must_use]
pub(crate) fn same_layout(&self, other: &Self) -> bool {
self.columns.len() == other.columns.len()
&& self
.columns
.iter()
.zip(&other.columns)
.all(|(a, b)| a.column_id == b.column_id && a.type_tag == b.type_tag)
}
#[must_use]
pub(crate) fn data_size(&self) -> usize {
self.columns
.iter()
.map(|c| c.data.len() + c.validity.as_ref().map_or(0, Vec::len))
.sum()
}
#[must_use]
pub(crate) fn zone_stats(&self) -> Vec<ColumnStats> {
let rows = self.row_count;
let mut stats = Vec::new();
for col in &self.columns {
if let TypeTag::Number(number) = col.type_tag {
let Ok((null_count, range)) = col.number_range(0, rows) else {
continue;
};
let (min, max) = match range {
Some((min, max)) => (
comparable_bytes(number, min).to_vec(),
comparable_bytes(number, max).to_vec(),
),
None => (Vec::new(), Vec::new()),
};
stats.push(ColumnStats {
column_id: u32::from(col.column_id),
type_tag: col.type_tag.to_wire().0,
codec_id: 0,
null_count,
row_count: rows,
min,
max,
});
continue;
}
let TypeTag::Bytes = col.type_tag else {
continue;
};
let mut min: Option<&[u8]> = None;
let mut max: Option<&[u8]> = None;
let mut null_count: u32 = 0;
for i in 0..rows {
if !column_row_valid(col, i) {
null_count += 1;
continue;
}
let Ok(value) = bytes_column_row(&col.data, rows, i) else {
continue;
};
if min.is_none_or(|m| value < m) {
min = Some(value);
}
if max.is_none_or(|m| value > m) {
max = Some(value);
}
}
stats.push(ColumnStats {
column_id: u32::from(col.column_id),
type_tag: TypeTag::Bytes.to_wire().0,
codec_id: 0,
null_count,
row_count: rows,
min: min.unwrap_or(&[]).to_vec(),
max: max.unwrap_or(&[]).to_vec(),
});
}
stats
}
pub(crate) fn page_zones(
&self,
row_pages: &[u32],
) -> Result<crate::table::column_page::PageZones> {
let ordered: Vec<&Column> = self
.columns
.iter()
.filter(|c| matches!(c.type_tag, TypeTag::Bytes | TypeTag::Number(_)))
.collect();
let mut zones = crate::table::column_page::PageZones::new(
ordered.iter().map(|c| c.column_id).collect(),
);
let mut start = 0u32;
for &rows in row_pages {
let Some(end) = start.checked_add(rows) else {
return Err(Error::InvalidHeader(
"columnar: row pages overrun the batch",
));
};
for col in &ordered {
if let TypeTag::Number(number) = col.type_tag {
let (nulls, range) = col.number_range(start, end)?;
match range {
Some((min, max)) => zones.push(
nulls,
Some((
&comparable_bytes(number, min),
&comparable_bytes(number, max),
)),
),
None => zones.push(nulls, None),
}
} else {
let (nulls, range) = col.bytes_range(self.row_count, start, end)?;
zones.push(nulls, range);
}
}
start = end;
}
Ok(zones)
}
pub(crate) fn group_zones(
&self,
row_pages: &[u32],
) -> Result<(
crate::table::column_page::PageZones,
crate::table::column_page::PageZones,
)> {
if row_pages.len() <= 1 {
return Ok(Default::default());
}
let zones = self.page_zones(row_pages)?;
Ok((
zones.only(|c| c == COL_USER_KEY),
zones.only(|c| c != COL_USER_KEY),
))
}
pub(crate) fn concat(pages: Vec<Self>) -> Result<Self> {
let mut pages = pages.into_iter();
let Some(first) = pages.next() else {
return Err(Error::InvalidHeader("columnar: row group has no row pages"));
};
let rest: Vec<Self> = pages.collect();
if rest.is_empty() {
return Ok(first);
}
let all = || core::iter::once(&first).chain(&rest);
let mut row_count = 0u32;
for page in all() {
if page.columns.len() != first.columns.len() {
return Err(Error::InvalidHeader(
"columnar: row pages disagree on their columns",
));
}
for (a, b) in first.columns.iter().zip(&page.columns) {
if a.column_id != b.column_id || a.type_tag != b.type_tag {
return Err(Error::InvalidHeader(
"columnar: row pages disagree on their columns",
));
}
b.validate(page.row_count)?;
}
let Some(sum) = row_count.checked_add(page.row_count) else {
return Err(Error::InvalidHeader(
"columnar: row group row count exceeds u32",
));
};
row_count = sum;
}
let mut columns = Vec::with_capacity(first.columns.len());
for (index, head) in first.columns.iter().enumerate() {
let parts = || all().filter_map(|page| page.columns.get(index).map(|c| (page, c)));
let data = match head.type_tag {
TypeTag::Fixed(_) | TypeTag::Number(_) => {
let len = parts().map(|(_, c)| c.data.len()).sum();
let mut out = Vec::with_capacity(len);
for (_, c) in parts() {
out.extend_from_slice(&c.data);
}
Slice::from(out)
}
TypeTag::Bytes => frame_bytes_column(row_count as usize, || {
parts().flat_map(|(page, c)| {
(0..page.row_count).map(move |i| {
bytes_column_row(&c.data, page.row_count, i).unwrap_or_default()
})
})
})?,
};
let validity = if parts().any(|(_, c)| c.validity.is_some()) {
let mut out = alloc::vec![0u8; validity_len(row_count)];
let mut at = 0u32;
for (page, c) in parts() {
for i in 0..page.row_count {
let present = c.validity.as_deref().is_none_or(|v| validity_bit(v, i));
if present && let Some(byte) = out.get_mut((at / 8) as usize) {
*byte |= 1u8 << (at % 8);
}
at += 1;
}
}
Some(out)
} else {
None
};
columns.push(Column {
column_id: head.column_id,
type_tag: head.type_tag,
validity,
data,
});
}
Ok(Self { row_count, columns })
}
#[expect(
clippy::cast_possible_truncation,
reason = "column count and per-column chunk length are bounded by the block size policy, far below u32::MAX"
)]
pub fn encode(&self) -> Result<Vec<u8>> {
let mut out = Vec::new();
out.extend_from_slice(&self.row_count.to_le_bytes());
out.extend_from_slice(&(self.columns.len() as u32).to_le_bytes());
for col in &self.columns {
col.encode_into(
self.row_count,
crate::config::ColumnEncoding::Auto,
&mut out,
)?;
}
Ok(out)
}
pub fn decode(bytes: &crate::Slice) -> Result<Self> {
Self::decode_inner(bytes, None)
}
pub fn decode_projected(bytes: &crate::Slice, wanted: &[u16]) -> Result<Self> {
Self::decode_inner(bytes, Some(wanted))
}
fn decode_inner(bytes: &crate::Slice, wanted: Option<&[u16]>) -> Result<Self> {
const MIN_COLUMN_BYTES: usize = 7;
let mut cur = Cursor::new(bytes);
let row_count = cur.read_u32()?;
let column_count = cur.read_u32()? as usize;
if column_count > cur.remaining() / MIN_COLUMN_BYTES {
return Err(Error::InvalidHeader(
"columnar: declared column count exceeds payload size",
));
}
let mut columns = Vec::with_capacity(column_count);
let mut viewed_bytes = 0usize;
let mut view_columns: Vec<usize> = Vec::new();
let want = |id: u16| wanted.is_none_or(|w| w.contains(&id));
let mut budget = DecodeBudget::unbounded();
for _ in 0..column_count {
let Some((column, viewed)) =
Column::decode_from(&mut cur, bytes, row_count, want, &mut 0, &mut budget)?
else {
continue;
};
if viewed > 0 {
viewed_bytes += viewed;
view_columns.push(columns.len());
}
columns.push(column);
}
if !cur.is_empty() {
return Err(Error::InvalidHeader(
"columnar: trailing bytes after the last column",
));
}
if !view_columns.is_empty() && bytes.len() > MAX_VIEW_AMPLIFICATION * viewed_bytes {
for idx in view_columns {
if let Some(col) = columns.get_mut(idx) {
col.data = crate::Slice::from(&col.data[..]);
}
}
}
Ok(Self { row_count, columns })
}
}
struct Cursor<'a> {
buf: &'a [u8],
pos: usize,
}
impl<'a> Cursor<'a> {
const fn new(buf: &'a [u8]) -> Self {
Self { buf, pos: 0 }
}
const fn remaining(&self) -> usize {
self.buf.len() - self.pos
}
const fn is_empty(&self) -> bool {
self.pos == self.buf.len()
}
fn read_bytes(&mut self, n: usize) -> Result<&'a [u8]> {
let Some(end) = self.pos.checked_add(n) else {
return Err(Error::InvalidHeader("columnar: read length overflow"));
};
let Some(slice) = self.buf.get(self.pos..end) else {
return Err(Error::InvalidHeader("columnar: truncated block payload"));
};
self.pos = end;
Ok(slice)
}
fn read_array<const N: usize>(&mut self) -> Result<[u8; N]> {
let arr: [u8; N] = self
.read_bytes(N)?
.try_into()
.map_err(|_| Error::InvalidHeader("columnar: short fixed-width read"))?;
Ok(arr)
}
fn read_u8(&mut self) -> Result<u8> {
let [b] = self.read_array::<1>()?;
Ok(b)
}
fn read_u16(&mut self) -> Result<u16> {
Ok(u16::from_le_bytes(self.read_array()?))
}
fn read_u32(&mut self) -> Result<u32> {
Ok(u32::from_le_bytes(self.read_array()?))
}
fn read_var_u32(&mut self) -> Result<u32> {
let mut rest = self.buf.get(self.pos..).unwrap_or_default();
let before = rest.len();
let Some(value) = crate::table::column_page::take_varint(&mut rest, 5)
.and_then(|v| u32::try_from(v).ok())
else {
return Err(Error::InvalidHeader("columnar: malformed length varint"));
};
self.pos += before - rest.len();
Ok(value)
}
}
pub const COL_USER_KEY: u16 = 0;
pub const COL_SEQNO: u16 = 1;
pub const COL_VALUE_TYPE: u16 = 2;
pub const COL_VALUE: u16 = 3;
fn build_bytes_column<'a>(rows: impl Iterator<Item = &'a [u8]>) -> Result<Vec<u8>> {
let mut offsets = Vec::new();
let mut payload = Vec::new();
let mut off: u32 = 0;
offsets.extend_from_slice(&off.to_le_bytes());
for r in rows {
let len = u32::try_from(r.len())
.map_err(|_| Error::InvalidHeader("columnar: column value exceeds u32"))?;
let Some(end) = off.checked_add(len) else {
return Err(Error::InvalidHeader("columnar: column payload exceeds u32"));
};
off = end;
payload.extend_from_slice(r);
offsets.extend_from_slice(&off.to_le_bytes());
}
offsets.extend_from_slice(&payload);
Ok(offsets)
}
fn column_row_valid(col: &Column, i: u32) -> bool {
match &col.validity {
None => true,
Some(bits) => {
let idx = i as usize;
bits.get(idx / 8)
.is_some_and(|byte| byte & (1u8 << (idx % 8)) != 0)
}
}
}
pub(crate) type NullsAndRange<'a> = (u32, Option<(&'a [u8], &'a [u8])>);
pub(crate) fn bytes_column_row(data: &[u8], row_count: u32, i: u32) -> Result<&[u8]> {
let span = bytes_column_span(data, row_count, i)?;
data.get(span)
.ok_or(Error::InvalidHeader("columnar: bytes row out of range"))
}
pub(crate) fn bytes_column_span(
data: &[u8],
row_count: u32,
i: u32,
) -> Result<core::ops::Range<usize>> {
let off_bytes = (row_count as usize + 1) * 4;
let read_off = |idx: u32| {
let base = idx as usize * 4;
data.get(base..)
.and_then(<[u8]>::first_chunk::<4>)
.map(|b| u32::from_le_bytes(*b) as usize)
};
let (Some(start), Some(end)) = (read_off(i), read_off(i + 1)) else {
return Err(Error::InvalidHeader("columnar: bytes offset truncated"));
};
if off_bytes > data.len() {
return Err(Error::InvalidHeader("columnar: bytes payload truncated"));
}
match (off_bytes.checked_add(start), off_bytes.checked_add(end)) {
(Some(start), Some(end)) if start <= end && end <= data.len() => Ok(start..end),
_ => Err(Error::InvalidHeader("columnar: bytes row out of range")),
}
}
pub(crate) fn frame_bytes_column<'a, I>(count: usize, cells: impl Fn() -> I) -> Result<Slice>
where
I: Iterator<Item = &'a [u8]>,
{
frame_bytes_column_within(count, u64::MAX, cells)
}
pub(crate) fn frame_bytes_column_within<'a, I>(
count: usize,
limit: u64,
cells: impl Fn() -> I,
) -> Result<Slice>
where
I: Iterator<Item = &'a [u8]>,
{
let too_large = |declared: u64| Error::DecompressedSizeTooLarge {
declared,
limit: limit.min(u64::from(u32::MAX)),
};
let mut total = 0u32;
let mut sized = 0usize;
for cell in cells() {
total = advance_bytes_offset(total, cell.len())?;
sized += 1;
}
let mismatch = || Error::InvalidHeader("columnar: bytes cells changed between passes");
if sized != count {
return Err(mismatch());
}
let table = count
.checked_add(1)
.and_then(|n| n.checked_mul(4))
.ok_or_else(|| too_large(count as u64))?;
let len = table
.checked_add(total as usize)
.ok_or_else(|| too_large(u64::from(total)))?;
if len as u64 > limit {
return Err(too_large(len as u64));
}
#[expect(unsafe_code, reason = "see safety")]
let mut out = unsafe { Slice::builder_unzeroed(len) };
let (offsets, payload) = out.split_at_mut(table);
let mut slots = offsets.chunks_exact_mut(4);
let mut written = 0usize;
let mut filled = 0usize;
if let Some(first) = slots.next() {
first.copy_from_slice(&0u32.to_le_bytes());
}
for cell in cells() {
let end = written + cell.len();
let (Some(dst), Some(slot)) = (payload.get_mut(written..end), slots.next()) else {
return Err(mismatch());
};
dst.copy_from_slice(cell);
written = end;
let offset = u32::try_from(written).map_err(|_| mismatch())?;
slot.copy_from_slice(&offset.to_le_bytes());
filled += 1;
}
if filled != count || written != total as usize {
return Err(mismatch());
}
Ok(Slice::from(out.freeze()))
}
fn advance_bytes_offset(acc: u32, value_len: usize) -> Result<u32> {
let len = u32::try_from(value_len).map_err(|_| Error::DecompressedSizeTooLarge {
declared: value_len as u64,
limit: u64::from(u32::MAX),
})?;
acc.checked_add(len)
.ok_or_else(|| Error::DecompressedSizeTooLarge {
declared: u64::from(acc) + u64::from(len),
limit: u64::from(u32::MAX),
})
}
pub(crate) fn gather_fixed_column<'a>(
width: usize,
count: usize,
mut cells: impl Iterator<Item = Option<&'a [u8]>>,
) -> Slice {
let len = count * width;
#[expect(unsafe_code, reason = "see safety")]
let mut out = unsafe { Slice::builder_unzeroed(len) };
for dst in out.chunks_exact_mut(width.max(1)) {
match cells.next() {
Some(Some(cell)) if cell.len() == dst.len() => dst.copy_from_slice(cell),
_ => dst.fill(0),
}
}
Slice::from(out.freeze())
}
fn validity_bit(bitmap: &[u8], row: u32) -> bool {
bitmap
.get((row / 8) as usize)
.is_some_and(|b| (b >> (row % 8)) & 1 == 1)
}
pub(crate) fn fixed_u64_row(data: &[u8], i: u32) -> Result<u64> {
let base = i as usize * 8;
match data.get(base..).and_then(<[u8]>::first_chunk::<8>) {
Some(b) => Ok(u64::from_le_bytes(*b)),
None => Err(Error::InvalidHeader("columnar: fixed8 row truncated")),
}
}
#[must_use]
pub(crate) const fn entry_row_bytes(key_len: usize, value_len: usize) -> usize {
key_len + 4 + 8 + 1 + value_len + 4
}
pub fn entries_to_column_batch(entries: &[InternalValue]) -> Result<ColumnBatch> {
let row_count = u32::try_from(entries.len())
.map_err(|_| Error::InvalidHeader("columnar: row count exceeds u32"))?;
let key_data = build_bytes_column(entries.iter().map(|e| e.key.user_key.as_ref()))?;
let value_data = build_bytes_column(entries.iter().map(|e| e.value.as_ref()))?;
let mut seqno_data = Vec::with_capacity(entries.len() * 8);
let mut vt_data = Vec::with_capacity(entries.len());
for e in entries {
seqno_data.extend_from_slice(&e.key.seqno.to_le_bytes());
vt_data.push(u8::from(e.key.value_type));
}
let columns = alloc::vec![
Column {
column_id: COL_USER_KEY,
type_tag: TypeTag::Bytes,
validity: None,
data: key_data.into(),
},
Column {
column_id: COL_SEQNO,
type_tag: TypeTag::Number(Number::U64_LE),
validity: None,
data: seqno_data.into(),
},
Column {
column_id: COL_VALUE_TYPE,
type_tag: TypeTag::Fixed(1),
validity: None,
data: vt_data.into(),
},
Column {
column_id: COL_VALUE,
type_tag: TypeTag::Bytes,
validity: None,
data: value_data.into(),
},
];
Ok(ColumnBatch { row_count, columns })
}
fn value_type_row(data: &[u8], i: u32) -> Result<ValueType> {
let Some(&byte) = data.get(i as usize) else {
return Err(Error::InvalidHeader("columnar: value-type row truncated"));
};
ValueType::try_from(byte).map_err(|()| Error::InvalidTag(("ValueType", byte)))
}
fn fixed_column_row(data: &[u8], width: u8, row: u32) -> Result<&[u8]> {
let w = width as usize;
let Some(end) = (row as usize)
.checked_add(1)
.and_then(|rows| rows.checked_mul(w))
else {
return Err(Error::InvalidHeader(
"columnar: fixed column offset overflow",
));
};
match data.get(end - w..end) {
Some(cell) => Ok(cell),
None => Err(Error::InvalidHeader("columnar: fixed column row truncated")),
}
}
fn column_cell(col: &Column, row_count: u32, row: u32) -> Result<&[u8]> {
match col.type_tag.fixed_width() {
Some(width) => fixed_column_row(&col.data, width, row),
None => bytes_column_row(&col.data, row_count, row),
}
}
fn reconstruct_row_value(
value_cols: &[Column],
row_count: u32,
row: u32,
value_type: ValueType,
) -> Result<Slice> {
if cells::holds_cells(value_cols.iter().map(|c| &c.column_id)) {
let mut row_cells = Vec::with_capacity(value_cols.len());
for col in value_cols {
row_cells.push((
col.column_id,
col.type_tag,
column_value_cell(col, row_count, row)?,
));
}
return cells::row_value(value_type, &row_cells);
}
if value_cols.iter().any(|c| c.validity.is_some()) {
let mut cells = Vec::with_capacity(value_cols.len());
for col in value_cols {
cells.push((col.type_tag, column_value_cell(col, row_count, row)?));
}
return Ok(Slice::from(frame_value_cells_nullable(&cells)?));
}
if let [single] = value_cols {
return Ok(Slice::from(column_cell(single, row_count, row)?));
}
let mut cells = Vec::with_capacity(value_cols.len());
for col in value_cols {
cells.push((col.type_tag, column_cell(col, row_count, row)?));
}
Ok(Slice::from(frame_value_cells(&cells)?))
}
fn validate_columnar_columns(
batch: &ColumnBatch,
) -> Result<(&Column, &Column, &Column, &[Column])> {
let [key_col, seqno_col, vt_col, value_cols @ ..] = batch.columns.as_slice() else {
return Err(Error::InvalidHeader(
"columnar: batch missing the intrinsic columns",
));
};
if value_cols.is_empty() {
return Err(Error::InvalidHeader(
"columnar: batch carries no value column",
));
}
if key_col.column_id != COL_USER_KEY
|| key_col.type_tag != TypeTag::Bytes
|| seqno_col.column_id != COL_SEQNO
|| seqno_col.type_tag != TypeTag::Number(Number::U64_LE)
|| vt_col.column_id != COL_VALUE_TYPE
|| vt_col.type_tag != TypeTag::Fixed(1)
{
return Err(Error::InvalidHeader(
"columnar: unexpected intrinsic column layout",
));
}
for col in [key_col, seqno_col, vt_col] {
if col.validity.is_some() {
return Err(Error::InvalidHeader(
"columnar: intrinsic columns must not be nullable",
));
}
col.validate(batch.row_count)?;
}
let mut seen_value_column_ids = Vec::with_capacity(value_cols.len());
for col in value_cols {
if col.column_id < COL_VALUE || seen_value_column_ids.contains(&col.column_id) {
return Err(Error::InvalidHeader(
"columnar: value sub-column ids must be unique and must not overlap intrinsic columns",
));
}
seen_value_column_ids.push(col.column_id);
col.validate(batch.row_count)?;
}
if cells::holds_cells(seen_value_column_ids.iter()) {
let shapes: Vec<(u16, TypeTag)> = value_cols
.iter()
.map(|c| (c.column_id, c.type_tag))
.collect();
cells::check_value_columns(&shapes)?;
}
Ok((key_col, seqno_col, vt_col, value_cols))
}
pub fn validate_columnar_ingest_batch(
batch: &ColumnBatch,
comparator: &crate::SharedComparator,
) -> Result<()> {
let (key_col, seqno_col, vt_col, value_cols) = validate_columnar_columns(batch)?;
check_ingested_field_ids(value_cols)?;
for &vt_byte in vt_col.data.iter() {
ValueType::try_from(vt_byte).map_err(|()| Error::InvalidTag(("ValueType", vt_byte)))?;
}
for i in 0..batch.row_count {
if fixed_u64_row(&seqno_col.data, i)? != 0 {
return Err(Error::FeatureUnsupported(
"columnar batch ingest requires every row seqno to be 0 (the ingestion assigns the sequence number)",
));
}
}
let mut prev: Option<&[u8]> = None;
for i in 0..batch.row_count {
let key = bytes_column_row(&key_col.data, batch.row_count, i)?;
if key.is_empty() || key.len() > u16::MAX as usize {
return Err(Error::InvalidHeader(
"columnar: user key is empty or longer than u16::MAX",
));
}
if let Some(p) = prev
&& comparator.compare(p, key) != core::cmp::Ordering::Less
{
return Err(Error::InvalidHeader(
"columnar batch ingest requires strictly increasing keys",
));
}
prev = Some(key);
}
Ok(())
}
pub(crate) fn transpose_like(
batch: &ColumnBatch,
entries: &[InternalValue],
) -> Result<Option<ColumnBatch>> {
if cells::holds_cells(batch.columns.iter().map(|c| &c.column_id)) {
return entries_to_cells_batch(entries).map(Some);
}
if !rows_round_trip(batch) {
return Ok(None);
}
entries_to_column_batch(entries).map(Some)
}
pub(crate) fn rows_round_trip(batch: &ColumnBatch) -> bool {
cells::holds_cells(batch.columns.iter().map(|c| &c.column_id))
|| matches!(
batch.columns.as_slice(),
[_, _, _, value]
if value.column_id == COL_VALUE
&& value.type_tag == TypeTag::Bytes
&& value.validity.is_none()
)
}
pub(crate) fn check_ingested_field_ids(value_cols: &[Column]) -> Result<()> {
if value_cols
.iter()
.any(|c| c.column_id >= crate::blob_tree::field_row::RESERVED_COLUMNS)
{
return Err(Error::InvalidHeader(
"columnar: a value sub-column id is one the engine keeps for itself",
));
}
Ok(())
}
pub fn column_batch_to_entries(batch: &ColumnBatch) -> Result<Vec<InternalValue>> {
let (key_col, seqno_col, vt_col, value_cols) = validate_columnar_columns(batch)?;
let mut out = Vec::with_capacity(batch.row_count as usize);
for i in 0..batch.row_count {
let user_key = bytes_column_row(&key_col.data, batch.row_count, i)?;
if user_key.is_empty() || user_key.len() > u16::MAX as usize {
return Err(Error::InvalidHeader(
"columnar: user key is empty or longer than u16::MAX",
));
}
let seqno = fixed_u64_row(&seqno_col.data, i)?;
let value_type = value_type_row(&vt_col.data, i)?;
let value = reconstruct_row_value(value_cols, batch.row_count, i, value_type)?;
out.push(InternalValue {
key: InternalKey {
user_key: Slice::from(user_key),
seqno,
value_type,
},
value,
});
}
Ok(out)
}
fn values_are_views(value_cols: &[Column]) -> bool {
matches!(
value_cols,
[c] if c.type_tag == TypeTag::Bytes && c.validity.is_none()
)
}
pub fn column_batch_into_entries(
batch: ColumnBatch,
rebuilt: &mut usize,
) -> Result<Vec<InternalValue>> {
validate_columnar_columns(&batch)?;
let row_count = batch.row_count;
let mut cols = batch.columns.into_iter();
let (Some(key_col), Some(seqno_col), Some(vt_col)) = (cols.next(), cols.next(), cols.next())
else {
return Err(Error::InvalidHeader(
"columnar: missing an intrinsic column",
));
};
let value_cols: Vec<Column> = cols.collect();
let key_data = key_col.data;
let value_source = if values_are_views(&value_cols) {
let Some(single) = value_cols.into_iter().next() else {
return Err(Error::InvalidHeader("columnar: value column vanished"));
};
ValueSource::SharedBytes(single.data)
} else {
ValueSource::Reconstruct(value_cols)
};
let mut out = Vec::with_capacity(row_count as usize);
for i in 0..row_count {
let user_key = bytes_row_slice(&key_data, row_count, i)?;
if user_key.is_empty() || user_key.len() > u16::MAX as usize {
return Err(Error::InvalidHeader(
"columnar: user key is empty or longer than u16::MAX",
));
}
let seqno = fixed_u64_row(&seqno_col.data, i)?;
let value_type = value_type_row(&vt_col.data, i)?;
let value = match &value_source {
ValueSource::SharedBytes(data) => bytes_row_slice(data, row_count, i)?,
ValueSource::Reconstruct(cols) => {
let value = reconstruct_row_value(cols, row_count, i, value_type)?;
*rebuilt += value.len();
value
}
};
out.push(InternalValue {
key: InternalKey {
user_key,
seqno,
value_type,
},
value,
});
}
Ok(out)
}
enum ValueSource {
SharedBytes(Slice),
Reconstruct(Vec<Column>),
}
fn bytes_row_slice(data: &Slice, row_count: u32, i: u32) -> Result<Slice> {
let bytes: &[u8] = data.as_ref();
let off_bytes = (row_count as usize + 1) * 4;
let read_off = |idx: u32| {
let base = idx as usize * 4;
bytes
.get(base..)
.and_then(<[u8]>::first_chunk::<4>)
.map(|b| u32::from_le_bytes(*b) as usize)
};
let (Some(start), Some(end)) = (read_off(i), read_off(i + 1)) else {
return Err(Error::InvalidHeader("columnar: bytes offset truncated"));
};
let (Some(payload_start), Some(payload_end)) =
(off_bytes.checked_add(start), off_bytes.checked_add(end))
else {
return Err(Error::InvalidHeader(
"columnar: bytes payload offset overflow",
));
};
if start > end || payload_end > bytes.len() {
return Err(Error::InvalidHeader("columnar: bytes row out of range"));
}
Ok(data.slice(payload_start..payload_end))
}
pub(crate) fn key_rows<'k>(
row_count: u32,
needle: &[u8],
comparator: &crate::comparator::SharedComparator,
key_at: impl Fn(u32) -> Result<Cell<'k>>,
) -> Result<core::ops::Range<u32>> {
let mut lo = 0u32;
let mut hi = row_count;
while lo < hi {
let mid = lo + (hi - lo) / 2;
if comparator.compare(&key_at(mid)?, needle) == core::cmp::Ordering::Less {
lo = mid + 1;
} else {
hi = mid;
}
}
let mut end = lo;
while end < row_count {
if comparator.compare(&key_at(end)?, needle) != core::cmp::Ordering::Equal {
break;
}
end += 1;
}
Ok(lo..end)
}
pub(crate) struct RowPageColumns<'a> {
pub(crate) ordinal: u16,
pub(crate) start: u32,
pub(crate) rows: u32,
pub(crate) columns: Vec<PageColumn<'a>>,
}
pub(crate) fn page_match_entries(
page: RowPageColumns<'_>,
run: core::ops::Range<u32>,
needle: &[u8],
deletes: Option<(&crate::table::delete_bitmap::DeleteBitmap, u32)>,
copied: &mut usize,
budget: &mut DecodeBudget,
out: &mut Vec<InternalValue>,
) -> Result<()> {
let rows = page.rows;
if rows == 0 {
return Err(Error::InvalidHeader(
"columnar: empty reconstructed data block",
));
}
let mut columns = page.columns.into_iter();
let (Some(seqno), Some(vt)) = (columns.next(), columns.next()) else {
return Err(Error::InvalidHeader(
"columnar: batch missing the intrinsic columns",
));
};
if seqno.column_id != COL_SEQNO
|| seqno.type_tag != TypeTag::Number(Number::U64_LE)
|| vt.column_id != COL_VALUE_TYPE
|| vt.type_tag != TypeTag::Fixed(1)
|| seqno.validity.is_some()
|| vt.validity.is_some()
{
return Err(Error::InvalidHeader(
"columnar: unexpected intrinsic column layout",
));
}
if run.start >= run.end || run.end > rows {
return Err(Error::InvalidHeader(
"columnar: key run outside its row page",
));
}
if needle.is_empty() || needle.len() > u16::MAX as usize {
return Err(Error::InvalidHeader(
"columnar: user key is empty or longer than u16::MAX",
));
}
let seqno_type = TypeTag::Number(Number::U64_LE);
let vt_type = TypeTag::Fixed(1);
let seqnos = budget.rows(seqno.values, seqno_type, rows)?;
let types = budget.rows(vt.values, vt_type, rows)?;
let mut values: Vec<(u16, TypeTag, Option<&[u8]>, expr::Rows<'_>)> =
Vec::with_capacity(columns.len());
let mut nullable = false;
for c in columns {
if c.column_id < COL_VALUE || values.iter().any(|&(id, ..)| id == c.column_id) {
return Err(Error::InvalidHeader(
"columnar: value sub-column ids must be unique and must not overlap intrinsic columns",
));
}
nullable |= c.validity.is_some();
values.push((
c.column_id,
c.type_tag,
c.validity,
budget.rows(c.values, c.type_tag, rows)?,
));
}
if values.is_empty() {
return Err(Error::InvalidHeader(
"columnar: batch carries no value column",
));
}
let holds_cells = cells::holds_cells(values.iter().map(|(id, ..)| id));
if holds_cells {
let shapes: Vec<(u16, TypeTag)> = values.iter().map(|&(id, tag, ..)| (id, tag)).collect();
cells::check_value_columns(&shapes)?;
}
let user_key = Slice::from(needle);
*copied += user_key.len();
for row in run {
if let Some((bitmap, start)) = deletes {
let Some(pos) = start.checked_add(row) else {
return Err(Error::InvalidHeader(
"columnar: row position exceeds u32::MAX",
));
};
if bitmap.contains(pos) {
continue;
}
}
let Some(seqno) = seqnos
.get(seqno_type, rows, row)?
.first_chunk::<8>()
.copied()
else {
return Err(Error::InvalidHeader("columnar: fixed8 row truncated"));
};
let Some(&vt_byte) = types.get(vt_type, rows, row)?.first() else {
return Err(Error::InvalidHeader("columnar: value-type row truncated"));
};
let value_type =
ValueType::try_from(vt_byte).map_err(|()| Error::InvalidTag(("ValueType", vt_byte)))?;
let value = if holds_cells {
let mut row_cells = Vec::with_capacity(values.len());
for (id, type_tag, validity, access) in &values {
let cell = if validity.is_none_or(|v| validity_bit(v, row)) {
Some(access.get(*type_tag, rows, row)?)
} else {
None
};
row_cells.push((*id, *type_tag, cell));
}
let borrowed: Vec<(u16, TypeTag, Option<&[u8]>)> = row_cells
.iter()
.map(|(id, tag, cell)| (*id, *tag, cell.as_deref()))
.collect();
cells::row_value(value_type, &borrowed)?
} else if let (false, [(_, type_tag, _, access)]) = (nullable, values.as_slice()) {
Slice::from(&*access.get(*type_tag, rows, row)?)
} else {
let cells = values
.iter()
.map(|(_, type_tag, validity, access)| {
Ok((
*type_tag,
if validity.is_none_or(|v| validity_bit(v, row)) {
Some(access.get(*type_tag, rows, row)?)
} else {
None
},
))
})
.collect::<Result<Vec<_>>>()?;
if nullable {
let framed: Vec<(TypeTag, Option<&[u8]>)> =
cells.iter().map(|(t, c)| (*t, c.as_deref())).collect();
Slice::from(frame_value_cells_nullable(&framed)?)
} else {
let framed: Vec<(TypeTag, &[u8])> = cells
.iter()
.map(|(t, c)| (*t, c.as_deref().unwrap_or_default()))
.collect();
Slice::from(frame_value_cells(&framed)?)
}
};
*copied += value.len();
out.push(InternalValue {
key: InternalKey {
user_key: user_key.clone(),
seqno: u64::from_le_bytes(seqno),
value_type,
},
value,
});
}
Ok(())
}
pub fn frame_value_cells(cells: &[(TypeTag, &[u8])]) -> Result<Vec<u8>> {
let mut out = Vec::new();
for (tag, cell) in cells {
match tag {
TypeTag::Fixed(width) | TypeTag::Number(Number { width, .. }) => {
if cell.len() != usize::from(*width) {
return Err(Error::InvalidHeader(
"columnar: fixed value sub-cell length does not match its type tag",
));
}
out.extend_from_slice(cell);
}
TypeTag::Bytes => {
let len = u32::try_from(cell.len()).map_err(|_| {
Error::InvalidHeader("columnar: framed value sub-cell exceeds u32")
})?;
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(cell);
}
}
}
Ok(out)
}
pub fn unframe_value_cells<'a>(blob: &'a [u8], type_tags: &[TypeTag]) -> Result<Vec<&'a [u8]>> {
let mut out = Vec::with_capacity(type_tags.len());
let mut pos = 0usize;
for tag in type_tags {
match tag {
TypeTag::Fixed(width) | TypeTag::Number(Number { width, .. }) => {
let end = pos
.checked_add(usize::from(*width))
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let cell = blob.get(pos..end).ok_or_else(|| {
Error::InvalidHeader("columnar: framed value truncated (fixed)")
})?;
out.push(cell);
pos = end;
}
TypeTag::Bytes => {
let len_end = pos
.checked_add(4)
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let len_bytes = blob
.get(pos..len_end)
.and_then(<[u8]>::first_chunk::<4>)
.ok_or_else(|| {
Error::InvalidHeader("columnar: framed value truncated (length)")
})?;
let len = u32::from_le_bytes(*len_bytes) as usize;
let end = len_end
.checked_add(len)
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let cell = blob.get(len_end..end).ok_or_else(|| {
Error::InvalidHeader("columnar: framed value truncated (bytes)")
})?;
out.push(cell);
pos = end;
}
}
}
if pos != blob.len() {
return Err(Error::InvalidHeader(
"columnar: framed value has trailing bytes",
));
}
Ok(out)
}
pub fn frame_value_cells_nullable(cells: &[(TypeTag, Option<&[u8]>)]) -> Result<Vec<u8>> {
let bitmap_len = cells.len().div_ceil(8);
let mut out = alloc::vec![0u8; bitmap_len];
for (i, (tag, cell)) in cells.iter().enumerate() {
let Some(c) = cell else {
continue; };
let byte = out
.get_mut(i / 8)
.ok_or_else(|| Error::InvalidHeader("columnar: presence bitmap index out of range"))?;
*byte |= 1u8 << (i % 8);
match tag {
TypeTag::Fixed(width) | TypeTag::Number(Number { width, .. }) => {
if c.len() != usize::from(*width) {
return Err(Error::InvalidHeader(
"columnar: fixed value sub-cell length does not match its type tag",
));
}
out.extend_from_slice(c);
}
TypeTag::Bytes => {
let len = u32::try_from(c.len()).map_err(|_| {
Error::InvalidHeader("columnar: framed value sub-cell exceeds u32")
})?;
out.extend_from_slice(&len.to_le_bytes());
out.extend_from_slice(c);
}
}
}
Ok(out)
}
pub fn unframe_value_cells_nullable<'a>(
blob: &'a [u8],
type_tags: &[TypeTag],
) -> Result<Vec<Option<&'a [u8]>>> {
let bitmap_len = type_tags.len().div_ceil(8);
let bitmap = blob.get(0..bitmap_len).ok_or_else(|| {
Error::InvalidHeader("columnar: nullable framed value truncated (presence bitmap)")
})?;
let mut pos = bitmap_len;
let mut out = Vec::with_capacity(type_tags.len());
for (i, tag) in type_tags.iter().enumerate() {
let present = bitmap.get(i / 8).is_some_and(|b| (b >> (i % 8)) & 1 == 1);
if !present {
out.push(None);
continue;
}
match tag {
TypeTag::Fixed(width) | TypeTag::Number(Number { width, .. }) => {
let end = pos
.checked_add(usize::from(*width))
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let cell = blob.get(pos..end).ok_or_else(|| {
Error::InvalidHeader("columnar: nullable framed value truncated (fixed)")
})?;
out.push(Some(cell));
pos = end;
}
TypeTag::Bytes => {
let len_end = pos
.checked_add(4)
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let len_bytes = blob
.get(pos..len_end)
.and_then(<[u8]>::first_chunk::<4>)
.ok_or_else(|| {
Error::InvalidHeader("columnar: nullable framed value truncated (length)")
})?;
let len = u32::from_le_bytes(*len_bytes) as usize;
let end = len_end
.checked_add(len)
.ok_or_else(|| Error::InvalidHeader("columnar: framed value overflow"))?;
let cell = blob.get(len_end..end).ok_or_else(|| {
Error::InvalidHeader("columnar: nullable framed value truncated (bytes)")
})?;
out.push(Some(cell));
pos = end;
}
}
}
if pos != blob.len() {
return Err(Error::InvalidHeader(
"columnar: nullable framed value has trailing bytes",
));
}
Ok(out)
}
pub fn unframe_value_cells_with_defaults<'a>(
blob: &'a [u8],
columns: &[(TypeTag, &'a [u8])],
) -> Result<Vec<&'a [u8]>> {
let tags: Vec<TypeTag> = columns.iter().map(|(t, _)| *t).collect();
let cells = unframe_value_cells_nullable(blob, &tags)?;
Ok(cells
.into_iter()
.zip(columns)
.map(|(cell, (_, default))| cell.unwrap_or(default))
.collect())
}
fn column_value_cell(col: &Column, row_count: u32, row: u32) -> Result<Option<&[u8]>> {
if let Some(validity) = &col.validity {
let Some(&byte) = validity.get((row / 8) as usize) else {
return Err(Error::InvalidHeader(
"columnar: validity bitmap shorter than row count",
));
};
if (byte >> (row % 8)) & 1 == 0 {
return Ok(None); }
}
Ok(Some(column_cell(col, row_count, row)?))
}
#[expect(clippy::expect_used, clippy::indexing_slicing, reason = "test code")]
#[cfg(test)]
mod tests;