use alloc::sync::Arc;
use alloc::vec::Vec;
use crate::table::columnar::{
COL_SEQNO, COL_USER_KEY, COL_VALUE_TYPE, Column, ColumnBatch, Number, TypeTag,
bytes_column_row, frame_bytes_column,
};
use crate::{Error, Slice};
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum Absent {
Null,
Default(Slice),
Error,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ProjectedField {
column_id: u16,
type_tag: Option<TypeTag>,
absent: Absent,
}
impl ProjectedField {
pub fn new(column_id: u16, type_tag: TypeTag, absent: Absent) -> crate::Result<Self> {
if intrinsic_type(column_id).is_some() {
return Err(Error::Projection(
"projection: an intrinsic column is projected by id, not declared",
));
}
if type_tag.fixed_width() == Some(0) {
return Err(Error::Projection(
"projection: a fixed-width field has a width of zero",
));
}
if let (Absent::Default(value), Some(width)) = (&absent, type_tag.fixed_width())
&& value.len() != usize::from(width)
{
return Err(Error::Projection(
"projection: a default does not fit its field's width",
));
}
Ok(Self {
column_id,
type_tag: Some(type_tag),
absent,
})
}
#[must_use]
pub fn by_id(column_id: u16) -> Self {
Self {
column_id,
type_tag: intrinsic_type(column_id),
absent: Absent::Null,
}
}
#[must_use]
pub fn column_id(&self) -> u16 {
self.column_id
}
#[must_use]
pub fn type_tag(&self) -> Option<TypeTag> {
self.type_tag
}
#[must_use]
pub fn absent(&self) -> &Absent {
&self.absent
}
}
fn intrinsic_type(column_id: u16) -> Option<TypeTag> {
match column_id {
COL_USER_KEY => Some(TypeTag::Bytes),
COL_SEQNO => Some(TypeTag::Number(Number::U64_LE)),
COL_VALUE_TYPE => Some(TypeTag::Fixed(1)),
_ => None,
}
}
pub trait ValueProjector: Send + Sync {
fn project(&self, key: &[u8], value: &[u8], row: &mut ProjectedRow<'_>) -> crate::Result<()>;
}
pub struct ProjectedRow<'a> {
fields: &'a [ProjectedField],
cells: &'a mut [Option<Vec<u8>>],
}
impl<'a> ProjectedRow<'a> {
pub(crate) fn new(fields: &'a [ProjectedField], cells: &'a mut [Option<Vec<u8>>]) -> Self {
Self { fields, cells }
}
#[must_use]
pub fn fields(&self) -> &[ProjectedField] {
self.fields
}
pub fn set(&mut self, index: usize, cell: &[u8]) -> crate::Result<()> {
let field = self.fields.get(index).ok_or(Error::Projection(
"projection: the projector set a field the projection does not have",
))?;
let type_tag = field.type_tag.ok_or(Error::Projection(
"projection: a row value is projected into a field with no declared type",
))?;
if let Some(width) = type_tag.fixed_width()
&& cell.len() != usize::from(width)
{
return Err(Error::Projection(
"projection: the projector wrote a cell of the wrong width",
));
}
let slot = self.cells.get_mut(index).ok_or(Error::Projection(
"projection: the projector set a field the projection does not have",
))?;
let buf = slot.get_or_insert_with(Vec::new);
buf.clear();
buf.extend_from_slice(cell);
Ok(())
}
}
#[derive(Clone, Default)]
pub struct Projection {
fields: Vec<ProjectedField>,
projector: Option<Arc<dyn ValueProjector>>,
}
impl core::fmt::Debug for Projection {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("Projection")
.field("fields", &self.fields)
.field("projector", &self.projector.is_some())
.finish()
}
}
impl Projection {
#[must_use]
pub fn new() -> Self {
Self::default()
}
#[must_use]
pub fn column(mut self, column_id: u16) -> Self {
self.fields.push(ProjectedField::by_id(column_id));
self
}
#[must_use]
pub fn field(mut self, field: ProjectedField) -> Self {
self.fields.push(field);
self
}
#[must_use]
pub fn projector(mut self, projector: Arc<dyn ValueProjector>) -> Self {
self.projector = Some(projector);
self
}
#[must_use]
pub fn fields(&self) -> &[ProjectedField] {
&self.fields
}
#[must_use]
pub fn value_projector(&self) -> Option<&Arc<dyn ValueProjector>> {
self.projector.as_ref()
}
pub(crate) fn column_ids(&self) -> Vec<u16> {
self.fields.iter().map(|f| f.column_id).collect()
}
}
impl From<&[u16]> for Projection {
fn from(ids: &[u16]) -> Self {
Self {
fields: ids.iter().copied().map(ProjectedField::by_id).collect(),
projector: None,
}
}
}
impl<const N: usize> From<&[u16; N]> for Projection {
fn from(ids: &[u16; N]) -> Self {
Self::from(ids.as_slice())
}
}
impl From<&Vec<u16>> for Projection {
fn from(ids: &Vec<u16>) -> Self {
Self::from(ids.as_slice())
}
}
impl From<&Self> for Projection {
fn from(projection: &Self) -> Self {
projection.clone()
}
}
pub fn is_declared(field: &ProjectedField) -> bool {
field.type_tag.is_some() && intrinsic_type(field.column_id).is_none()
}
pub fn project_decided(
batch: ColumnBatch,
fields: &[ProjectedField],
projector: Option<&dyn ValueProjector>,
whole_id: u16,
) -> crate::Result<ColumnBatch> {
let row_count = batch.row_count;
let rows = row_count as usize;
let find = |id: u16| batch.columns.iter().find(|c| c.column_id == id);
let (Some(values), Some(keys)) = (find(whole_id), find(COL_USER_KEY)) else {
return Err(MISSING_SCAN_COLUMN);
};
let types = find(COL_VALUE_TYPE);
let declared: Vec<ProjectedField> = fields.iter().filter(|f| is_declared(f)).cloned().collect();
let mut read: Vec<Option<Vec<Option<Vec<u8>>>>> = Vec::with_capacity(rows);
let mut cells: Vec<Option<Vec<u8>>> = alloc::vec![None; declared.len()];
for row in 0..row_count {
let is_value = match types {
Some(types) => {
let byte = *types.data.get(row as usize).ok_or(MISSING_SCAN_COLUMN)?;
returns_a_value(
crate::ValueType::try_from(byte)
.map_err(|()| Error::InvalidTag(("ValueType", byte)))?,
)
}
None => true,
};
if !(is_value && values.is_valid(row)) {
read.push(None);
continue;
}
let projector = projector.ok_or(UNREADABLE)?;
cells.fill(None);
let key = bytes_column_row(&keys.data, row_count, row)?;
let value = bytes_column_row(&values.data, row_count, row)?;
projector.project(key, value, &mut ProjectedRow::new(&declared, &mut cells))?;
read.push(Some(core::mem::take(&mut cells)));
cells = alloc::vec![None; declared.len()];
}
if read.iter().all(Option::is_none) {
return Ok(batch);
}
let ColumnBatch {
row_count,
mut columns,
} = batch;
for (index, field) in declared.iter().enumerate() {
let Some(at) = columns.iter().position(|c| c.column_id == field.column_id) else {
return Err(MISSING_SCAN_COLUMN);
};
let old = columns.get(at).ok_or(MISSING_SCAN_COLUMN)?;
let mut column: Vec<Option<Vec<u8>>> = Vec::with_capacity(rows);
for (row, projected) in (0..row_count).zip(&read) {
column.push(match projected {
Some(cells) => cells.get(index).cloned().flatten(),
None if old.is_valid(row) => Some(cell_of(old, row_count, row)?.to_vec()),
None => None,
});
}
let built = built_column(field, &column)?;
if let Some(slot) = columns.get_mut(at) {
*slot = built;
}
}
Ok(ColumnBatch { row_count, columns })
}
fn cell_of(column: &Column, rows: u32, row: u32) -> crate::Result<&[u8]> {
match column.type_tag.fixed_width() {
Some(width) => {
let width = usize::from(width);
let start = row as usize * width;
column
.data
.get(start..start + width)
.ok_or(MISSING_SCAN_COLUMN)
}
None => bytes_column_row(&column.data, rows, row),
}
}
pub(super) const UNREADABLE: Error = Error::Projection(
"projection: a row returned stores its value whole, and declared fields are read out \
of it through a projector, which is not set",
);
fn returns_a_value(value_type: crate::ValueType) -> bool {
matches!(
value_type,
crate::ValueType::Value | crate::ValueType::MergeOperand
)
}
fn built_column(field: &ProjectedField, cells: &[Option<Vec<u8>>]) -> crate::Result<Column> {
let type_tag = field.type_tag.ok_or(ABSENT_FIELD)?;
let count = cells.len();
let validity = cells.iter().any(Option::is_none).then(|| {
let mut bits = alloc::vec![0u8; count.div_ceil(8)];
for (row, cell) in cells.iter().enumerate() {
if cell.is_some()
&& let Some(byte) = bits.get_mut(row / 8)
{
*byte |= 1 << (row % 8);
}
}
bits
});
let data = match type_tag.fixed_width() {
Some(width) => {
let width = usize::from(width);
let mut data = alloc::vec![0u8; count * width];
for (slot, cell) in data.chunks_exact_mut(width).zip(cells) {
if let Some(cell) = cell {
slot.copy_from_slice(cell);
}
}
Slice::from(data)
}
None => frame_bytes_column(count, || cells.iter().map(|c| c.as_deref().unwrap_or(&[])))?,
};
Ok(Column {
column_id: field.column_id,
type_tag,
validity,
data,
})
}
const MISSING_SCAN_COLUMN: Error =
Error::InvalidHeader("columnar_scan: a whole-value batch is missing its key or value column");
pub fn absent_column(
field: &ProjectedField,
rows: u32,
is_value: &dyn Fn(u32) -> bool,
) -> crate::Result<Column> {
let type_tag = field.type_tag.ok_or(ABSENT_FIELD)?;
let count = rows as usize;
let (validity, data) = match &field.absent {
Absent::Error if (0..rows).any(is_value) => return Err(ABSENT_FIELD),
Absent::Null | Absent::Error => {
let data = match type_tag.fixed_width() {
Some(width) => Slice::from(alloc::vec![0u8; count * usize::from(width)]),
None => frame_bytes_column(count, || core::iter::repeat_n(&[][..], count))?,
};
(Some(alloc::vec![0u8; count.div_ceil(8)]), data)
}
Absent::Default(value) => {
let data = match type_tag.fixed_width() {
Some(_) => Slice::from(value.repeat(count)),
None => frame_bytes_column(count, || core::iter::repeat_n(&**value, count))?,
};
(None, data)
}
};
Ok(Column {
column_id: field.column_id,
type_tag,
validity,
data,
})
}
const ABSENT_FIELD: Error = Error::Projection(
"projection: a projected field is absent from a row and its declaration does not allow it",
);
pub fn conform(batch: ColumnBatch, fields: &[ProjectedField]) -> crate::Result<ColumnBatch> {
conform_with(batch, fields, Conform::Strict).map(|(batch, _)| batch)
}
pub fn conform_lenient(
batch: ColumnBatch,
fields: &[ProjectedField],
judged: Option<u16>,
) -> crate::Result<(ColumnBatch, Vec<u16>)> {
conform_with(batch, fields, Conform::Lenient { judged })
}
pub const MISTYPED: Error =
Error::Projection("projection: a segment stores a projected field under another type");
#[derive(Clone, Copy)]
enum Conform {
Strict,
Lenient { judged: Option<u16> },
}
fn conform_with(
batch: ColumnBatch,
fields: &[ProjectedField],
mode: Conform,
) -> crate::Result<(ColumnBatch, Vec<u16>)> {
let strict = matches!(mode, Conform::Strict);
let defers = |field: &ProjectedField| match mode {
Conform::Strict => false,
Conform::Lenient { judged } => {
judged != Some(field.column_id) && matches!(field.absent, Absent::Default(_))
}
};
let ColumnBatch {
row_count,
mut columns,
} = batch;
let types = columns
.iter()
.find(|c| c.column_id == COL_VALUE_TYPE)
.map(|c| c.data.clone());
let is_value = |row: u32| {
strict
&& types.as_ref().is_none_or(|types| {
types
.get(row as usize)
.and_then(|&byte| crate::ValueType::try_from(byte).ok())
.is_some_and(returns_a_value)
})
};
let mut mistyped = Vec::new();
let mut out = Vec::with_capacity(fields.len().max(columns.len()));
for field in fields {
let at = columns.iter().position(|c| c.column_id == field.column_id);
let column = match at {
Some(at) => {
let column = columns.remove(at);
if field.type_tag.is_some_and(|t| t != column.type_tag) {
if strict {
return Err(MISTYPED);
}
mistyped.push(field.column_id);
let null = ProjectedField {
absent: Absent::Null,
..field.clone()
};
out.push(absent_column(&null, row_count, &|_| false)?);
continue;
}
if defers(field) {
column
} else {
fill_nulls(column, field, row_count, &is_value)?
}
}
None if !strict && field.type_tag.is_none() => continue,
None if defers(field) => {
let null = ProjectedField {
absent: Absent::Null,
..field.clone()
};
absent_column(&null, row_count, &|_| false)?
}
None => absent_column(field, row_count, &is_value)?,
};
out.push(column);
}
out.extend(columns);
Ok((
ColumnBatch {
row_count,
columns: out,
},
mistyped,
))
}
fn fill_nulls(
column: Column,
field: &ProjectedField,
rows: u32,
is_value: &dyn Fn(u32) -> bool,
) -> crate::Result<Column> {
let Some(validity) = &column.validity else {
return Ok(column);
};
let is_null = |row: u32| {
validity
.get(row as usize / 8)
.is_none_or(|byte| byte >> (row % 8) & 1 == 0)
};
if !(0..rows).any(is_null) {
return Ok(column);
}
let value = match &field.absent {
Absent::Error if (0..rows).any(|row| is_null(row) && is_value(row)) => {
return Err(ABSENT_FIELD);
}
Absent::Null | Absent::Error => return Ok(column),
Absent::Default(value) => value,
};
let count = rows as usize;
let data = if let Some(width) = column.type_tag.fixed_width() {
let width = usize::from(width);
let mut data = column.data.to_vec();
for row in (0..rows).filter(|&row| is_null(row)) {
let at = row as usize * width;
data.get_mut(at..at + width)
.ok_or(Error::InvalidHeader("columnar: fixed column row truncated"))?
.copy_from_slice(value);
}
Slice::from(data)
} else {
let cells = (0..rows)
.map(|row| {
if is_null(row) {
Ok(&**value)
} else {
bytes_column_row(&column.data, rows, row)
}
})
.collect::<crate::Result<Vec<&[u8]>>>()?;
frame_bytes_column(count, || cells.iter().copied())?
};
Ok(Column {
column_id: column.column_id,
type_tag: column.type_tag,
validity: None,
data,
})
}
#[cfg(test)]
#[expect(clippy::unwrap_used, reason = "test code")]
mod tests;