use std::borrow::Cow;
use std::cell::RefCell;
use std::cmp::Ordering;
use std::sync::Arc;
use rudb_common::{Cause, Error, Field, LogicalType, Result, Value, slow};
use crate::buffer::Buffer;
use crate::fsst::SymbolTable;
use crate::string::{StringColumn, StringView};
use crate::validity::Validity;
pub const VECTOR_SIZE: usize = 8192;
fn extent(at: &[u32]) -> Option<(u32, u32)> {
const FLIP: u32 = 1 << 31;
#[expect(clippy::cast_possible_wrap, reason = "the flip makes the wrap keep the order")]
let signed = |row: u32| (row ^ FLIP) as i32;
#[expect(clippy::cast_sign_loss, reason = "undoing the flip above")]
let unsigned = |row: i32| (row as u32) ^ FLIP;
if at.is_empty() {
return None;
}
let low = at.iter().fold(i32::MAX, |low, &row| low.min(signed(row)));
let high = at.iter().fold(i32::MIN, |high, &row| high.max(signed(row)));
Some((unsigned(low), unsigned(high)))
}
#[must_use]
pub fn below(codes: &[u32], len: usize) -> bool {
let Ok(len) = u32::try_from(len) else { return true };
if codes.is_empty() || codes.iter().fold(0, |bits, &code| bits | code) < len {
return true;
}
codes.iter().copied().fold(0, u32::max) < len
}
pub const MAP_KEY: &str = "key";
pub const MAP_VALUE: &str = "value";
pub type MapParts<'a> = (&'a [(u32, u32)], &'a Vector, &'a Vector);
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum Form {
Flat,
Constant,
Sequence,
Dictionary,
BitPacked,
StringView,
Fsst,
Rle,
List,
Struct,
Gathered,
}
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum Data {
Empty,
Bool(Buffer<bool>),
Int8(Buffer<i8>),
Int16(Buffer<i16>),
Int32(Buffer<i32>),
Int64(Buffer<i64>),
Int128(Buffer<i128>),
UInt8(Buffer<u8>),
UInt16(Buffer<u16>),
UInt32(Buffer<u32>),
UInt64(Buffer<u64>),
UInt128(Buffer<u128>),
Float32(Buffer<f32>),
Float64(Buffer<f64>),
Interval(Buffer<(i32, i32, i64)>),
Varlen(StringColumn),
}
impl Data {
#[must_use]
pub fn len(&self) -> usize {
macro_rules! lengths {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match self {
Self::Empty => 0,
$(Self::$variant(values) => values.len(),)+
}
};
}
crate::for_each_layout!(all, lengths)
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.len() == 0
}
#[must_use]
pub fn footprint(&self) -> usize {
macro_rules! sizes {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match self {
Self::Empty => 0,
$(Self::$variant(values) => values.footprint(),)+
}
};
}
crate::for_each_layout!(all, sizes)
}
#[must_use]
pub fn into_pages(self) -> Self {
macro_rules! paged {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match self {
Self::Empty => Self::Empty,
$(Self::$variant(values) => Self::$variant(values.into_page()),)+
}
};
}
crate::for_each_layout!(all, paged)
}
#[must_use]
pub fn signed_at(&self, index: usize) -> Option<i128> {
macro_rules! widened {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match self {
$(Self::$variant(v) => v.get(index).map(|&x| i128::from(x)),)+
_ => None,
}
};
}
crate::for_each_layout!(signed, widened)
}
#[must_use]
pub fn signed_block(&self, len: usize, out: &mut Vec<i64>) -> bool {
match self {
Self::Int8(v) => widen(v.as_slice(), len, out),
Self::Int16(v) => widen(v.as_slice(), len, out),
Self::Int32(v) => widen(v.as_slice(), len, out),
Self::Int64(v) => match v.as_slice().get(..len) {
Some(run) => {
out.extend_from_slice(run);
true
}
None => false,
},
_ => false,
}
}
#[must_use]
pub fn signed_gather(&self, len: usize, at: &[u32], out: &mut Vec<i64>) -> bool {
match self {
Self::Int8(v) => gather_widened(v.as_slice(), len, at, out),
Self::Int16(v) => gather_widened(v.as_slice(), len, at, out),
Self::Int32(v) => gather_widened(v.as_slice(), len, at, out),
Self::Int64(v) => gather_widened(v.as_slice(), len, at, out),
_ => false,
}
}
#[must_use]
pub fn signed_runs(
&self,
len: usize,
(from, to): (usize, usize),
every: usize,
out: &mut Vec<(i64, usize)>,
) -> bool {
match self {
Self::Int8(v) => runs_widened(v.as_slice(), len, (from, to), every, out),
Self::Int16(v) => runs_widened(v.as_slice(), len, (from, to), every, out),
Self::Int32(v) => runs_widened(v.as_slice(), len, (from, to), every, out),
Self::Int64(v) => runs_widened(v.as_slice(), len, (from, to), every, out),
_ => false,
}
}
#[must_use]
pub fn unsigned_at(&self, index: usize) -> Option<u128> {
macro_rules! widened {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match self {
$(Self::$variant(v) => v.get(index).map(|&x| u128::from(x)),)+
_ => None,
}
};
}
crate::for_each_layout!(unsigned, widened)
}
#[must_use]
pub fn str_at(&self, index: usize) -> Option<&str> {
match self {
Self::Varlen(column) => column.get(index),
_ => None,
}
}
#[must_use]
pub fn bytes_at(&self, index: usize) -> Option<&[u8]> {
match self {
Self::Varlen(column) => column.bytes(index),
_ => None,
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct Vector {
ty: LogicalType,
len: usize,
validity: Validity,
body: Body,
}
#[derive(Debug, Clone, PartialEq)]
enum Body {
Flat(Data),
Constant(Box<Value>),
Sequence {
start: i64,
step: i64,
},
Dictionary {
codes: Buffer<u32>,
values: Arc<Vector>,
stable: bool,
},
Packed {
words: Arc<Vec<u64>>,
width: u32,
base: i128,
offset: usize,
},
Views {
views: Vec<StringView>,
arena: Arc<Buffer<u8>>,
},
ExternalText {
source: Arc<dyn TextSource>,
},
Coded {
codes: Arc<Vec<u8>>,
spans: Vec<(u32, u32)>,
table: Arc<SymbolTable>,
},
Runs {
ends: Vec<u32>,
values: Arc<Vector>,
},
Nested {
entries: Vec<(u32, u32)>,
child: Arc<Vector>,
},
Fields {
children: Vec<Arc<Vector>>,
},
Gathered {
source: Arc<Vector>,
rids: Arc<Vec<u32>>,
offset: usize,
},
}
pub trait TextSource: std::fmt::Debug + Send + Sync {
fn len(&self) -> usize;
fn is_empty(&self) -> bool {
self.len() == 0
}
fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>>;
fn bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
Ok(self.bytes_at(index)?.map(<[u8]>::len))
}
fn bytes_lens_at(&self, indices: &[u32], into: &mut Vec<i64>) -> Result<()> {
into.reserve(indices.len());
for &index in indices {
let len = self.bytes_len_at(index as usize)?.unwrap_or_default();
into.push(i64::try_from(len).unwrap_or(i64::MAX));
}
Ok(())
}
fn chars_lens_at(&self, indices: &[u32], into: &mut Vec<i64>) -> Result<()> {
into.reserve(indices.len());
for &index in indices {
let bytes = self.bytes_at(index as usize)?.unwrap_or_default();
let characters = bytes.iter().filter(|byte| (**byte as i8) >= -0x40).count();
into.push(i64::try_from(characters).unwrap_or(i64::MAX));
}
Ok(())
}
fn sweep(
&self,
first: usize,
limit: usize,
body: &mut dyn FnMut(usize, &[u8]) -> Result<()>,
) -> Result<usize> {
if first >= limit.min(self.len()) {
return Ok(first);
}
body(first, self.bytes_at(first)?.unwrap_or_default())?;
Ok(first + 1)
}
fn visit_at(
&self,
indices: &[u32],
body: &mut dyn FnMut(usize, &[u8]) -> Result<()>,
) -> Result<()> {
for (at, &index) in indices.iter().enumerate() {
body(at, self.bytes_at(index as usize)?.unwrap_or_default())?;
}
Ok(())
}
fn might_contain(&self, first: usize, literal: &[u8]) -> Result<bool> {
let _ = (first, literal);
Ok(true)
}
fn visit(
&self,
indices: &[usize],
body: &mut dyn FnMut(usize, &[u8]) -> Result<()>,
) -> Result<()> {
for (at, &index) in indices.iter().enumerate() {
body(at, self.bytes_at(index)?.unwrap_or_default())?;
}
Ok(())
}
fn footprint(&self) -> usize;
fn ranks(&self) -> Option<usize> {
None
}
fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
let _ = (rank, wanted);
Err(Error::internal("a text source without a sorted order was asked to compare a rank"))
}
fn below(&self, ranks: usize, wanted: &[u8]) -> Result<(usize, bool)> {
search_below(self, ranks, wanted)
}
fn code_at_rank(&self, rank: usize) -> Result<u32> {
let _ = rank;
Err(Error::internal("a text source without a sorted order was asked for a rank"))
}
fn code_ranks(&self) -> Option<&[u32]> {
None
}
fn equal(&self, other: &dyn TextSource) -> bool {
self.len() == other.len()
&& (0..self.len()).all(|index| {
matches!(
(self.bytes_at(index), other.bytes_at(index)),
(Ok(left), Ok(right)) if left == right
)
})
}
}
impl PartialEq for dyn TextSource {
fn eq(&self, other: &Self) -> bool {
self.equal(other)
}
}
pub fn search_below<S>(source: &S, ranks: usize, wanted: &[u8]) -> Result<(usize, bool)>
where
S: TextSource + ?Sized,
{
let mut low = 0;
let mut high = ranks;
let mut equal = false;
while low < high {
let middle = low + (high - low) / 2;
match source.compare_rank(middle, wanted)? {
Ordering::Less => low = middle + 1,
Ordering::Greater => high = middle,
Ordering::Equal => {
equal = true;
high = middle;
}
}
}
Ok((low, equal))
}
impl Vector {
pub fn flat(ty: LogicalType, data: Data) -> Result<Self> {
let len = data.len();
if !matches!(data, Data::Empty) && layout_of(&data) != ty.physical() {
return Err(Error::internal(format!(
"a {ty} vector cannot hold {:?} data",
layout_of(&data)
)));
}
Ok(Self { ty, len, validity: Validity::AllValid, body: Body::Flat(data) })
}
pub fn from_values(ty: LogicalType, values: &[Value]) -> Result<Self> {
match &ty {
LogicalType::List(element) => {
return Self::list_from_values(element.as_ref().clone(), values);
}
LogicalType::Struct(fields) => return Self::struct_from_values(fields, values),
LogicalType::Map(key, value) => {
return Self::map_from_values(key.as_ref().clone(), value.as_ref().clone(), values);
}
_ => {}
}
let mut data = empty_data_for(&ty)?;
for value in values {
let value = stored(&ty, value)?;
push_value(&mut data, &value)?;
}
let validity = Validity::from_iter(values.len(), |index| !values[index].is_null());
Ok(Self { ty, len: values.len(), validity, body: Body::Flat(data) })
}
fn list_from_values(element: LogicalType, values: &[Value]) -> Result<Self> {
let mut flat = Vec::new();
let mut entries = Vec::with_capacity(values.len());
for value in values {
let start = u32::try_from(flat.len())
.map_err(|_| Error::internal("a list column with more than u32 elements in it"))?;
match value {
Value::Null => entries.push((start, 0)),
Value::List { values: held, .. } => {
let len = u32::try_from(held.len())
.map_err(|_| Error::internal("a list longer than u32"))?;
flat.extend_from_slice(held);
entries.push((start, len));
}
other => {
return Err(Error::internal(format!(
"{other:?} does not belong in a list vector"
)));
}
}
}
let child = Self::from_values(element, &flat)?;
let validity = Validity::from_iter(values.len(), |index| !values[index].is_null());
Ok(Self {
ty: LogicalType::list(child.ty.clone()),
len: values.len(),
validity,
body: Body::Nested { entries, child: Arc::new(child) },
})
}
pub fn list(entries: Vec<(u32, u32)>, child: Vector) -> Result<Self> {
let reach = child.len();
for &(start, len) in &entries {
if start as usize + len as usize > reach {
return Err(Error::internal(format!(
"a list entry of {len} at {start} in a child of {reach}"
)));
}
}
Ok(Self {
ty: LogicalType::list(child.ty.clone()),
len: entries.len(),
validity: Validity::AllValid,
body: Body::Nested { entries, child: Arc::new(child) },
})
}
fn struct_from_values(fields: &[Field], values: &[Value]) -> Result<Self> {
let mut children = Vec::with_capacity(fields.len());
let unnamed = Field::unnamed(fields);
for (at, field) in fields.iter().enumerate() {
let mut column = Vec::with_capacity(values.len());
for value in values {
column.push(match value {
Value::Null => Value::Null,
Value::Struct(held) if unnamed => held
.get(at)
.map(|(_, held)| held.clone())
.ok_or_else(|| Error::internal("a tuple row shorter than its type"))?,
Value::Struct(held) => held
.iter()
.find(|(name, _)| *name == field.name)
.map(|(_, held)| held.clone())
.ok_or_else(|| {
Error::internal(format!(
"a struct row with no {} field in it",
field.name
))
})?,
other => {
return Err(Error::internal(format!(
"{other:?} does not belong in a struct vector"
)));
}
});
}
children.push(Arc::new(Self::from_values(field.ty.clone(), &column)?));
}
let validity = Validity::from_iter(values.len(), |index| !values[index].is_null());
Ok(Self {
ty: LogicalType::Struct(fields.to_vec()),
len: values.len(),
validity,
body: Body::Fields { children },
})
}
pub fn structure(children: Vec<(String, Vector)>) -> Result<Self> {
let Some((_, first)) = children.first() else {
return Err(Error::internal("a struct vector of no fields, which has no length"));
};
let len = first.len();
for (name, child) in &children {
if child.len() != len {
return Err(Error::internal(format!(
"a {} field of {} rows beside a struct of {len}",
name,
child.len()
)));
}
}
let fields = children
.iter()
.map(|(name, child)| Field::new(name.clone(), child.ty.clone()))
.collect();
let children = children.into_iter().map(|(_, child)| Arc::new(child)).collect();
Ok(Self {
ty: LogicalType::Struct(fields),
len,
validity: Validity::AllValid,
body: Body::Fields { children },
})
}
#[must_use]
pub fn struct_parts(&self) -> Option<&[Arc<Self>]> {
match &self.body {
Body::Fields { children } => Some(children),
_ => None,
}
}
fn map_from_values(key: LogicalType, value: LogicalType, values: &[Value]) -> Result<Self> {
let mut keys = Vec::new();
let mut held = Vec::new();
let mut entries = Vec::with_capacity(values.len());
for row in values {
let start = u32::try_from(keys.len())
.map_err(|_| Error::internal("a map column with more than u32 entries in it"))?;
match row {
Value::Null => entries.push((start, 0)),
Value::Map { entries: pairs, .. } => {
let len = u32::try_from(pairs.len())
.map_err(|_| Error::internal("a map with more than u32 entries"))?;
for (one, other) in pairs {
keys.push(one.clone());
held.push(other.clone());
}
entries.push((start, len));
}
other => {
return Err(Error::internal(format!(
"{other:?} does not belong in a map vector"
)));
}
}
}
let child = Self::structure(vec![
(MAP_KEY.to_string(), Self::from_values(key, &keys)?),
(MAP_VALUE.to_string(), Self::from_values(value, &held)?),
])?;
let ty = LogicalType::map(
fields_of(&child.ty)[0].ty.clone(),
fields_of(&child.ty)[1].ty.clone(),
);
let validity = Validity::from_iter(values.len(), |index| !values[index].is_null());
Ok(Self {
ty,
len: values.len(),
validity,
body: Body::Nested { entries, child: Arc::new(child) },
})
}
pub fn map(entries: Vec<(u32, u32)>, keys: Vector, values: Vector) -> Result<Self> {
let key = keys.ty.clone();
let value = values.ty.clone();
let child =
Self::structure(vec![(MAP_KEY.to_string(), keys), (MAP_VALUE.to_string(), values)])?;
let mut vector = Self::list(entries, child)?;
vector.ty = LogicalType::map(key, value);
Ok(vector)
}
#[must_use]
pub fn map_parts(&self) -> Option<MapParts<'_>> {
if !matches!(self.ty, LogicalType::Map(_, _)) {
return None;
}
let (entries, child) = self.list_parts()?;
let [keys, values] = child.struct_parts()? else { return None };
Some((entries, keys, values))
}
#[must_use]
pub fn list_parts(&self) -> Option<(&[(u32, u32)], &Self)> {
match &self.body {
Body::Nested { entries, child } => Some((entries, child)),
_ => None,
}
}
#[must_use]
pub fn constant(ty: LogicalType, value: Value, len: usize) -> Self {
let validity = if value.is_null() { Validity::AllInvalid } else { Validity::AllValid };
Self { ty, len, validity, body: Body::Constant(Box::new(value)) }
}
#[must_use]
pub fn sequence(start: i64, step: i64, len: usize) -> Self {
Self {
ty: LogicalType::BigInt,
len,
validity: Validity::AllValid,
body: Body::Sequence { start, step },
}
}
pub fn dictionary(codes: Vec<u32>, values: Vector) -> Result<Self> {
Self::dictionary_over(codes, Arc::new(values))
}
pub fn dictionary_over(codes: Vec<u32>, values: Arc<Vector>) -> Result<Self> {
if !below(&codes, values.len()) {
let highest = codes.iter().copied().fold(0, u32::max);
return Err(Error::internal(format!(
"dictionary code {highest} is past the end of a {} value dictionary",
values.len()
)));
}
let (codes, values) = compose(codes, values);
Ok(Self {
ty: values.ty.clone(),
len: codes.len(),
validity: Validity::AllValid,
body: Body::Dictionary { codes: Buffer::from_vec(codes), values, stable: false },
})
}
pub fn stable_dictionary(codes: Vec<u32>, values: Arc<Vector>) -> Result<Self> {
let mut vector = Self::dictionary_over(codes, values)?;
if let Body::Dictionary { stable, .. } = &mut vector.body {
*stable = true;
}
Ok(vector)
}
pub fn stable_dictionary_validated(
codes: Vec<u32>,
values: Arc<Vector>,
highest: Option<u32>,
) -> Result<Self> {
if highest.is_some_and(|code| code as usize >= values.len()) {
return Err(Error::internal("a stable dictionary code is past its value dictionary"));
}
Ok(Self {
ty: values.ty.clone(),
len: codes.len(),
validity: Validity::AllValid,
body: Body::Dictionary { codes: Buffer::from_vec(codes), values, stable: true },
})
}
pub fn gathered(source: Arc<Vector>, rids: Arc<Vec<u32>>) -> Result<Self> {
let len = rids.len();
Self::gathered_from(source, rids, 0, len)
}
pub fn gathered_from(
source: Arc<Vector>,
rids: Arc<Vec<u32>>,
offset: usize,
len: usize,
) -> Result<Self> {
let end = offset.checked_add(len).ok_or_else(|| Error::internal("a gather that wraps"))?;
let Some(taken) = rids.get(offset..end) else {
return Err(Error::internal(format!(
"rows {offset} to {end} of a gather over {} ids",
rids.len()
)));
};
let rows = source.len();
if taken.iter().any(|&rid| rid != NO_ROW && rid as usize >= rows) {
return Err(Error::internal(format!(
"a gathered row id is past the {rows} rows of its source"
)));
}
Ok(Self {
ty: source.ty.clone(),
len,
validity: Validity::AllValid,
body: Body::Gathered { source, rids, offset },
})
}
#[must_use]
pub fn gathered_parts(&self) -> Option<(&Arc<Self>, &[u32])> {
match &self.body {
Body::Gathered { source, rids, offset } => {
Some((source, rids.get(*offset..offset + self.len)?))
}
_ => None,
}
}
#[must_use]
pub fn fold_over_source(&self) -> bool {
match &self.body {
Body::Gathered { source, .. } => source.len() < self.len,
_ => false,
}
}
pub fn runs(ends: Vec<u32>, values: Vector) -> Result<Self> {
if matches!(values.body, Body::Runs { .. }) {
return Err(Error::internal("runs of runs, which is two searches to read one row"));
}
if ends.len() != values.len() {
return Err(Error::internal(format!(
"{} runs and {} values to put in them",
ends.len(),
values.len()
)));
}
if ends.windows(2).any(|pair| pair[0] >= pair[1]) || ends.first() == Some(&0) {
return Err(Error::internal("run ends that do not increase"));
}
let len = ends.last().copied().unwrap_or(0) as usize;
Ok(Self {
ty: values.ty.clone(),
len,
validity: Validity::AllValid,
body: Body::Runs { ends, values: Arc::new(values) },
})
}
pub fn run_encoded(&self) -> Result<Self> {
let Body::Flat(data) = &self.body else {
return Ok(self.clone());
};
let ends = boundaries(data, &self.validity, self.len);
if ends.len().saturating_mul(RUNS_PAY_AT) >= self.len {
return Ok(self.clone());
}
let starts: Vec<u32> =
std::iter::once(0).chain(ends.iter().copied()).take(ends.len()).collect();
Self::runs(ends, self.gather(&starts)?)
}
pub fn packed(
ty: LogicalType,
words: Vec<u64>,
width: u32,
base: i128,
len: usize,
) -> Result<Self> {
let Some((low, high)) = layout_range(&ty) else {
return Err(Error::internal(format!("a {ty} vector has no integer layout to pack")));
};
if width == 0 || width > PACKED_WIDTH_MAX {
return Err(Error::internal(format!(
"a packed width of {width}, which is outside 1 to {PACKED_WIDTH_MAX}"
)));
}
let needed = words_for(len, width);
if words.len() < needed {
return Err(Error::internal(format!(
"{} words for {len} values of {width} bits, which needs {needed}",
words.len()
)));
}
let top = base + i128::from(u64::MAX >> (64 - width));
if base < low || top > high {
return Err(Error::internal(format!(
"packed values from {base} to {top}, which a {ty} cannot hold"
)));
}
Ok(Self {
ty,
len,
validity: Validity::AllValid,
body: Body::Packed { words: Arc::new(words), width, base, offset: 0 },
})
}
pub fn bit_packed(&self) -> Result<Self> {
let Body::Flat(data) = &self.body else {
return Ok(self.clone());
};
let Some((low, high)) = span_of(data, self.len) else {
return Ok(self.clone());
};
let Some(range) = high.checked_sub(low).and_then(|range| u64::try_from(range).ok()) else {
return Ok(self.clone());
};
let width = u64::BITS - range.leading_zeros();
if width == 0 || width > PACKED_WIDTH_MAX {
return Ok(self.clone());
}
if words_for(self.len, width) * size_of::<u64>() * PACKING_PAYS_AT
> flat_bytes(data, self.len)
{
return Ok(self.clone());
}
let Some(base) = packing_base(&self.ty, low, high, width) else {
return Ok(self.clone());
};
let words = pack(data, self.len, base, width);
let packed = Self::packed(self.ty.clone(), words, width, base, self.len)?;
Ok(packed.with_validity(self.validity.clone()))
}
pub fn string_views(
ty: LogicalType,
views: Vec<StringView>,
arena: Arc<Buffer<u8>>,
) -> Result<Self> {
if ty.physical() != rudb_common::PhysicalType::Varlen {
return Err(Error::internal(format!("a {ty} vector cannot hold string views")));
}
if views.iter().any(|view| view.bytes_in(&arena).is_none()) {
return Err(Error::internal("a string view points past the end of its arena"));
}
let len = views.len();
Ok(Self { ty, len, validity: Validity::AllValid, body: Body::Views { views, arena } })
}
pub fn external_text(ty: LogicalType, source: Arc<dyn TextSource>) -> Result<Self> {
if ty.physical() != rudb_common::PhysicalType::Varlen {
return Err(Error::internal(format!(
"a {ty} vector cannot use an external text source"
)));
}
let len = source.len();
Ok(Self { ty, len, validity: Validity::AllValid, body: Body::ExternalText { source } })
}
pub fn shared_text(self) -> Result<Self> {
let Body::Flat(Data::Varlen(column)) = self.body else {
return Ok(self);
};
let (views, arena) = column.into_parts();
let shared = Self::string_views(self.ty, views, Arc::new(arena))?;
Ok(shared.with_validity(self.validity))
}
pub fn coded(
ty: LogicalType,
codes: Arc<Vec<u8>>,
spans: Vec<(u32, u32)>,
table: Arc<SymbolTable>,
) -> Result<Self> {
if ty.physical() != rudb_common::PhysicalType::Varlen {
return Err(Error::internal(format!("a {ty} vector cannot hold FSST codes")));
}
let end = u32::try_from(codes.len()).unwrap_or(u32::MAX);
if spans.iter().any(|&(from, to)| from > to || to > end) {
return Err(Error::internal("an FSST span runs past the end of the codes"));
}
let len = spans.len();
Ok(Self {
ty,
len,
validity: Validity::AllValid,
body: Body::Coded { codes, spans, table },
})
}
pub fn compressed(self) -> Result<Self> {
let Body::Flat(Data::Varlen(column)) = &self.body else {
return Ok(self);
};
let rows: Vec<&[u8]> = (0..self.len).filter_map(|row| column.bytes(row)).collect();
if rows.len() != self.len {
return Ok(self);
}
let plain: usize = rows.iter().map(|row| row.len()).sum();
let table = SymbolTable::train(&rows);
let mut codes = Vec::with_capacity(plain);
let mut spans = Vec::with_capacity(self.len);
for row in &rows {
let from = u32::try_from(codes.len()).unwrap_or(u32::MAX);
table.compress(row, &mut codes);
spans.push((from, u32::try_from(codes.len()).unwrap_or(u32::MAX)));
}
if codes.len() * FSST_PAYS_AT > plain {
return Ok(self);
}
let coded = Self::coded(self.ty.clone(), Arc::new(codes), spans, Arc::new(table))?;
Ok(coded.with_validity(self.validity.clone()))
}
#[must_use]
pub fn as_wider_decimal(&self, target: &LogicalType) -> Option<Self> {
let (
LogicalType::Decimal { width: from, scale: held },
LogicalType::Decimal { width: into, scale },
) = (&self.ty, target)
else {
return None;
};
if held != scale || from > into || self.ty.decimal_storage() != target.decimal_storage() {
return None;
}
if !matches!(self.body, Body::Flat(_)) {
return None;
}
Some(Self {
ty: target.clone(),
len: self.len,
validity: self.validity.clone(),
body: self.body.clone(),
})
}
#[must_use]
pub fn with_validity(mut self, validity: Validity) -> Self {
self.validity = validity;
self
}
#[must_use]
pub fn logical_type(&self) -> &LogicalType {
&self.ty
}
#[must_use]
pub fn len(&self) -> usize {
self.len
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.len == 0
}
#[must_use]
pub fn footprint(&self) -> usize {
let body = match &self.body {
Body::Flat(data) => data.footprint(),
Body::Constant(value) => value.footprint(),
Body::Sequence { .. } => 0,
Body::Dictionary { codes, values, .. } => {
codes.footprint() + share(values.footprint(), values)
}
Body::Packed { words, .. } => share(words.capacity() * size_of::<u64>(), words),
Body::Views { views, arena } => {
views.capacity() * size_of::<StringView>() + share(arena.footprint(), arena)
}
Body::ExternalText { source } => share(source.footprint(), source),
Body::Coded { codes, spans, table } => {
share(codes.capacity(), codes)
+ spans.capacity() * size_of::<(u32, u32)>()
+ share(table.footprint(), table)
}
Body::Runs { ends, values } => {
ends.capacity() * size_of::<u32>() + share(values.footprint(), values)
}
Body::Gathered { source, rids, .. } => {
share(rids.capacity() * size_of::<u32>(), rids) + share(source.footprint(), source)
}
Body::Nested { entries, child } => {
entries.capacity() * size_of::<(u32, u32)>() + share(child.footprint(), child)
}
Body::Fields { children } => {
children.capacity() * size_of::<Arc<Self>>()
+ children.iter().map(|child| share(child.footprint(), child)).sum::<usize>()
}
};
size_of::<Self>() + self.validity.footprint() + body
}
#[must_use]
pub fn validity(&self) -> &Validity {
&self.validity
}
#[must_use]
pub fn is_null_at(&self, index: usize) -> bool {
if index >= self.len || !self.validity.is_valid(index) {
return true;
}
match &self.body {
Body::Dictionary { codes, values, .. } => match codes.get(index) {
Some(&code) => values.is_null_at(code as usize),
None => true,
},
Body::Runs { ends, values } => match run_holding(ends, index) {
Some(run) => values.is_null_at(run),
None => true,
},
Body::Gathered { source, rids, offset } => match rids.get(offset + index) {
Some(&NO_ROW) | None => true,
Some(&rid) => source.is_null_at(rid as usize),
},
_ => false,
}
}
#[must_use]
pub fn never_null(&self) -> bool {
if self.validity.has_nulls(self.len) {
return false;
}
match &self.body {
Body::Dictionary { values, .. } | Body::Runs { values, .. } => values.never_null(),
Body::Gathered { source, rids, offset } => {
source.never_null()
&& !rids[*offset..].iter().take(self.len).any(|&rid| rid == NO_ROW)
}
_ => true,
}
}
#[must_use]
pub fn form(&self) -> Form {
match self.body {
Body::Flat(_) => Form::Flat,
Body::Constant(_) => Form::Constant,
Body::Sequence { .. } => Form::Sequence,
Body::Dictionary { .. } => Form::Dictionary,
Body::Packed { .. } => Form::BitPacked,
Body::Views { .. } => Form::StringView,
Body::ExternalText { .. } => Form::StringView,
Body::Coded { .. } => Form::Fsst,
Body::Runs { .. } => Form::Rle,
Body::Nested { .. } => Form::List,
Body::Fields { .. } => Form::Struct,
Body::Gathered { .. } => Form::Gathered,
}
}
#[must_use]
pub fn data(&self) -> Option<&Data> {
match &self.body {
Body::Flat(data) => Some(data),
_ => None,
}
}
#[must_use]
pub fn constant_value(&self) -> Option<&Value> {
match &self.body {
Body::Constant(value) => Some(value.as_ref()),
_ => None,
}
}
#[must_use]
pub fn dictionary_parts(&self) -> Option<(&[u32], &Self)> {
match &self.body {
Body::Dictionary { codes, values, .. } => Some((codes, values.as_ref())),
_ => None,
}
}
#[must_use]
pub fn shared_dictionary_parts(&self) -> Option<(&[u32], &Arc<Self>)> {
match &self.body {
Body::Dictionary { codes, values, .. } => Some((codes, values)),
_ => None,
}
}
#[must_use]
pub fn stable_dictionary_parts(&self) -> Option<(&[u32], &Arc<Self>)> {
match &self.body {
Body::Dictionary { codes, values, stable: true } => Some((codes, values)),
_ => None,
}
}
#[must_use]
pub fn run_parts(&self) -> Option<(&[u32], &Self)> {
match &self.body {
Body::Runs { ends, values } => Some((ends, values.as_ref())),
_ => None,
}
}
#[must_use]
pub fn positions(&self) -> Option<(Cow<'_, [u32]>, &Self)> {
match &self.body {
Body::Dictionary { codes, values, .. } => Some((Cow::Borrowed(codes), values.as_ref())),
Body::Runs { ends, values } => {
let mut at = Vec::with_capacity(self.len);
for (run, &stop) in ends.iter().enumerate() {
let run = u32::try_from(run).unwrap_or(u32::MAX);
at.resize(stop as usize, run);
}
Some((Cow::Owned(at), values.as_ref()))
}
_ => None,
}
}
#[must_use]
pub fn packed_parts(&self) -> Option<Packed<'_>> {
match &self.body {
Body::Packed { words, width, base, offset } => {
Some(Packed { words, width: *width, base: *base, offset: *offset })
}
_ => None,
}
}
#[must_use]
pub fn text_parts(&self) -> Option<(&[StringView], &[u8])> {
match &self.body {
Body::Flat(Data::Varlen(column)) => Some((column.views(), column.arena())),
Body::Views { views, arena } => Some((views, arena)),
_ => None,
}
}
#[must_use]
pub fn shared_views(&self) -> Option<(&[StringView], &Arc<Buffer<u8>>)> {
match &self.body {
Body::Views { views, arena } => Some((views, arena)),
_ => None,
}
}
#[must_use]
pub fn coded_parts(&self) -> Option<Coded<'_>> {
match &self.body {
Body::Coded { codes, spans, table } => Some(Coded { codes, spans, table }),
_ => None,
}
}
#[must_use]
pub fn sequence_parts(&self) -> Option<(i64, i64)> {
match self.body {
Body::Sequence { start, step } => Some((start, step)),
_ => None,
}
}
pub fn enum_codes(&self) -> Result<Self> {
if self.ty.labels().is_none() {
return Err(Error::internal(format!("enum_code over a {} vector", self.ty)));
}
let ty = enum_code_type(&self.ty);
if let Body::Constant(value) = &self.body {
return Ok(Self::constant(ty, enum_position(&self.ty, value)?, self.len));
}
let flat = self.flatten()?;
Ok(Self { ty, ..flat })
}
#[must_use]
pub fn value_at(&self, index: usize) -> Value {
if index >= self.len || !self.validity.is_valid(index) {
return Value::Null;
}
match &self.body {
Body::Constant(value) => value.as_ref().clone(),
Body::Sequence { start, step } => Value::BigInt(start + step * index as i64),
Body::Dictionary { codes, values, .. } => match codes.get(index) {
Some(&code) => values.value_at(code as usize),
None => Value::Null,
},
Body::Runs { ends, values } => match run_holding(ends, index) {
Some(run) => values.value_at(run),
None => Value::Null,
},
Body::Gathered { source, rids, offset } => match rids.get(offset + index) {
Some(&NO_ROW) | None => Value::Null,
Some(&rid) => source.value_at(rid as usize),
},
Body::Packed { words, width, base, offset } => {
unpack(&self.ty, words, *offset, *width, *base, &[index])
.map_or(Value::Null, |data| value_from(&self.ty, &data, 0))
}
Body::Views { views, arena } => {
match views.get(index).and_then(|v| v.bytes_in(arena)) {
Some(bytes) => bytes_as(&self.ty, bytes),
None => Value::Null,
}
}
Body::ExternalText { source } => source
.bytes_at(index)
.ok()
.flatten()
.map_or(Value::Null, |bytes| bytes_as(&self.ty, bytes)),
Body::Coded { codes, spans, table } => {
match spans.get(index).and_then(|&(from, to)| {
let mut out = Vec::new();
table.decompress(codes.get(from as usize..to as usize)?, &mut out).ok()?;
Some(out)
}) {
Some(bytes) => bytes_as(&self.ty, &bytes),
None => Value::Null,
}
}
Body::Nested { entries, child } => match (entries.get(index), &self.ty) {
(Some(&(start, len)), LogicalType::Map(key, value)) => {
let pairs = child.struct_parts().unwrap_or_default();
Value::map(
key.as_ref().clone(),
value.as_ref().clone(),
(start..start + len)
.filter_map(|at| {
let [keys, values] = pairs else { return None };
Some((keys.value_at(at as usize), values.value_at(at as usize)))
})
.collect(),
)
}
(Some(&(start, len)), _) => Value::List {
element: child.ty.clone(),
values: (start..start + len).map(|at| child.value_at(at as usize)).collect(),
},
(None, _) => Value::Null,
},
Body::Fields { children } => Value::Struct(
fields_of(&self.ty)
.iter()
.zip(children)
.map(|(field, child)| (field.name.clone(), child.value_at(index)))
.collect(),
),
Body::Flat(data) => value_from(&self.ty, data, index),
}
}
pub fn value_of(&self, bytes: &[u8]) -> Value {
bytes_as(&self.ty, bytes)
}
pub fn try_value_at(&self, index: usize) -> Result<Value> {
if index >= self.len || !self.validity.is_valid(index) {
return Ok(Value::Null);
}
match &self.body {
Body::ExternalText { source } => {
Ok(source.bytes_at(index)?.map_or(Value::Null, |bytes| bytes_as(&self.ty, bytes)))
}
Body::Dictionary { codes, values, .. } => match codes.get(index) {
Some(&code) => values.try_value_at(code as usize),
None => Ok(Value::Null),
},
Body::Runs { ends, values } => match run_holding(ends, index) {
Some(run) => values.try_value_at(run),
None => Ok(Value::Null),
},
Body::Nested { entries, child } => match (entries.get(index), &self.ty) {
(Some(&(start, len)), LogicalType::Map(key, value)) => {
let pairs = child.struct_parts().unwrap_or_default();
let [keys, values] = pairs else { return Ok(Value::Null) };
let mut entries = Vec::with_capacity(len as usize);
for at in start..start + len {
entries.push((
keys.try_value_at(at as usize)?,
values.try_value_at(at as usize)?,
));
}
Ok(Value::map(key.as_ref().clone(), value.as_ref().clone(), entries))
}
(Some(&(start, len)), _) => {
let mut values = Vec::with_capacity(len as usize);
for at in start..start + len {
values.push(child.try_value_at(at as usize)?);
}
Ok(Value::List { element: child.ty.clone(), values })
}
(None, _) => Ok(Value::Null),
},
Body::Fields { children } => {
let mut values = Vec::with_capacity(children.len());
for (field, child) in fields_of(&self.ty).iter().zip(children) {
values.push((field.name.clone(), child.try_value_at(index)?));
}
Ok(Value::Struct(values))
}
_ => Ok(self.value_at(index)),
}
}
#[must_use]
pub fn text_at(&self, index: usize) -> Option<&str> {
if self.ty != LogicalType::Varchar || index >= self.len || !self.validity.is_valid(index) {
return None;
}
match &self.body {
Body::Flat(data) => data.str_at(index),
Body::Dictionary { codes, values, .. } => {
values.text_at(usize::try_from(*codes.get(index)?).ok()?)
}
Body::Runs { ends, values } => values.text_at(run_holding(ends, index)?),
Body::Gathered { source, rids, offset } => {
source.text_at(row_of(rids, *offset, index)?)
}
Body::Views { views, arena } => {
std::str::from_utf8(views.get(index)?.bytes_in(arena)?).ok()
}
Body::ExternalText { source } => {
std::str::from_utf8(source.bytes_at(index).ok().flatten()?).ok()
}
_ => None,
}
}
#[must_use]
pub fn bytes_at(&self, index: usize) -> Option<&[u8]> {
if index >= self.len || !self.validity.is_valid(index) {
return None;
}
match &self.body {
Body::Constant(value) => match value.as_ref() {
Value::Varchar(text) => Some(text.as_bytes()),
Value::Blob(bytes) => Some(bytes),
_ => None,
},
Body::Dictionary { codes, values, .. } => {
values.bytes_at(usize::try_from(*codes.get(index)?).ok()?)
}
Body::Runs { ends, values } => values.bytes_at(run_holding(ends, index)?),
Body::Gathered { source, rids, offset } => {
source.bytes_at(row_of(rids, *offset, index)?)
}
Body::Views { views, arena } => views.get(index)?.bytes_in(arena),
Body::ExternalText { source } => source.bytes_at(index).ok().flatten(),
Body::Flat(data) => data.bytes_at(index),
Body::Coded { .. }
| Body::Sequence { .. }
| Body::Packed { .. }
| Body::Nested { .. }
| Body::Fields { .. } => None,
}
}
pub fn try_bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
if index >= self.len || !self.validity.is_valid(index) {
return Ok(None);
}
match &self.body {
Body::Constant(value) => Ok(match value.as_ref() {
Value::Varchar(text) => Some(text.as_bytes()),
Value::Blob(bytes) => Some(bytes.as_slice()),
_ => None,
}),
Body::Dictionary { codes, values, .. } => match codes.get(index) {
Some(&code) => values.try_bytes_at(code as usize),
None => Ok(None),
},
Body::Runs { ends, values } => match run_holding(ends, index) {
Some(run) => values.try_bytes_at(run),
None => Ok(None),
},
Body::Gathered { source, rids, offset } => match row_of(rids, *offset, index) {
Some(row) => source.try_bytes_at(row),
None => Ok(None),
},
Body::Views { views, arena } => {
Ok(views.get(index).and_then(|view| view.bytes_in(arena)))
}
Body::ExternalText { source } => source.bytes_at(index),
Body::Flat(data) => Ok(data.bytes_at(index)),
Body::Coded { .. }
| Body::Sequence { .. }
| Body::Packed { .. }
| Body::Nested { .. }
| Body::Fields { .. } => Ok(None),
}
}
pub fn sweep_text(
&self,
first: usize,
limit: usize,
body: &mut dyn FnMut(usize, &[u8]) -> Result<()>,
) -> Result<usize> {
let limit = limit.min(self.len);
if first >= limit {
return Ok(first);
}
if let Body::ExternalText { source } = &self.body {
if matches!(self.validity, Validity::AllValid) {
return source.sweep(first, limit, body);
}
}
body(first, self.try_bytes_at(first)?.unwrap_or_default())?;
Ok(first + 1)
}
pub fn text_block_might_contain(&self, first: usize, literal: &[u8]) -> Result<bool> {
match &self.body {
Body::ExternalText { source } if matches!(self.validity, Validity::AllValid) => {
source.might_contain(first, literal)
}
_ => Ok(true),
}
}
pub fn try_values_visited(&self, indices: &[usize]) -> Result<Vec<Value>> {
if let Body::ExternalText { source } = &self.body {
if matches!(self.validity, Validity::AllValid) {
let mut out = vec![Value::Null; indices.len()];
let mut own = |at: usize, bytes: &[u8]| {
if indices[at] < self.len {
out[at] = bytes_as(&self.ty, bytes);
}
Ok(())
};
source.visit(indices, &mut own)?;
return Ok(out);
}
}
indices.iter().map(|&index| self.try_value_at(index)).collect()
}
pub fn try_bytes_len_at(&self, index: usize) -> Result<Option<usize>> {
if index >= self.len || !self.validity.is_valid(index) {
return Ok(None);
}
match &self.body {
Body::Dictionary { codes, values, .. } => match codes.get(index) {
Some(&code) => values.try_bytes_len_at(code as usize),
None => Ok(None),
},
Body::Runs { ends, values } => match run_holding(ends, index) {
Some(run) => values.try_bytes_len_at(run),
None => Ok(None),
},
Body::ExternalText { source } => source.bytes_len_at(index),
_ => Ok(self.bytes_at(index).map(<[u8]>::len)),
}
}
pub fn try_bytes_lens(&self, into: &mut Vec<i64>) -> Result<bool> {
self.lens_through(into, false, |source, indices, into| source.bytes_lens_at(indices, into))
}
pub fn try_chars_lens(&self, into: &mut Vec<i64>) -> Result<bool> {
self.lens_through(into, true, |source, indices, into| source.chars_lens_at(indices, into))
}
fn lens_through(
&self,
into: &mut Vec<i64>,
nulls: bool,
ask: impl Fn(&dyn TextSource, &[u32], &mut Vec<i64>) -> Result<()>,
) -> Result<bool> {
if !nulls && !matches!(self.validity, Validity::AllValid) {
return Ok(false);
}
into.clear();
match &self.body {
Body::ExternalText { source } => {
let Ok(rows) = u32::try_from(self.len) else { return Ok(false) };
let indices = (0..rows).collect::<Vec<_>>();
ask(source.as_ref(), &indices, into)?;
Ok(true)
}
Body::Dictionary { codes, values, .. } => match &values.body {
Body::ExternalText { source } if matches!(values.validity, Validity::AllValid) => {
let Some(codes) = codes.get(..self.len) else { return Ok(false) };
ask(source.as_ref(), codes, into)?;
Ok(true)
}
_ => Ok(false),
},
_ => Ok(false),
}
}
pub fn try_visit_text(&self, body: &mut dyn FnMut(usize, &[u8]) -> Result<()>) -> Result<bool> {
let (source, codes) = match &self.body {
Body::ExternalText { source } => (source, None),
Body::Dictionary { codes, values, .. } => match &values.body {
Body::ExternalText { source } if matches!(values.validity, Validity::AllValid) => {
let Some(codes) = codes.get(..self.len) else { return Ok(false) };
(source, Some(codes))
}
_ => return Ok(false),
},
_ => return Ok(false),
};
let Ok(len) = u32::try_from(self.len) else { return Ok(false) };
let rows: Option<Vec<u32>> = match &self.validity {
Validity::AllValid => None,
Validity::AllInvalid => return Ok(true),
Validity::Mask(mask) => Some((0..len).filter(|&row| mask.get(row as usize)).collect()),
};
let indices = match (codes, &rows) {
(Some(codes), None) => Cow::Borrowed(codes),
(Some(codes), Some(rows)) => rows.iter().map(|&row| codes[row as usize]).collect(),
(None, None) => (0..len).collect(),
(None, Some(rows)) => Cow::Borrowed(rows.as_slice()),
};
source.visit_at(&indices, &mut |at, bytes| {
let row = rows.as_ref().map_or(at, |rows| rows[at] as usize);
body(row, bytes)
})?;
Ok(true)
}
#[must_use]
pub fn ranks(&self) -> Option<usize> {
match &self.body {
Body::ExternalText { source } => source.ranks(),
_ => None,
}
}
pub fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
match &self.body {
Body::ExternalText { source } => source.compare_rank(rank, wanted),
_ => {
Err(Error::internal("a vector without a sorted order was asked to compare a rank"))
}
}
}
pub fn below(&self, ranks: usize, wanted: &[u8]) -> Result<(usize, bool)> {
match &self.body {
Body::ExternalText { source } => source.below(ranks, wanted),
_ => Err(Error::internal("a vector without a sorted order was asked for a boundary")),
}
}
pub fn code_at_rank(&self, rank: usize) -> Result<u32> {
match &self.body {
Body::ExternalText { source } => source.code_at_rank(rank),
_ => Err(Error::internal("a vector without a sorted order was asked for a rank")),
}
}
#[must_use]
pub fn code_ranks(&self) -> Option<&[u32]> {
match &self.body {
Body::ExternalText { source } => source.code_ranks(),
_ => None,
}
}
pub fn try_text_at(&self, index: usize) -> Result<Option<&str>> {
if self.ty != LogicalType::Varchar {
return Ok(None);
}
self.try_bytes_at(index)?
.map(|bytes| {
std::str::from_utf8(bytes).map_err(|error| {
Error::conversion(format!("invalid UTF-8 in VARCHAR: {error}"))
})
})
.transpose()
}
pub fn validate_external(&self) -> Result<()> {
match &self.body {
Body::ExternalText { source } => {
for index in 0..source.len() {
source.bytes_at(index)?;
}
}
Body::Dictionary { codes, values, .. } => {
if values.reaches_storage() {
for &code in codes.iter() {
values.try_bytes_at(code as usize)?;
}
}
}
Body::Runs { values, .. } | Body::Gathered { source: values, .. } => {
values.validate_external()?;
}
Body::Nested { child, .. } => child.validate_external()?,
Body::Fields { children } => {
for child in children {
child.validate_external()?;
}
}
_ => {}
}
Ok(())
}
fn reaches_storage(&self) -> bool {
match &self.body {
Body::ExternalText { .. } => true,
Body::Dictionary { values, .. }
| Body::Runs { values, .. }
| Body::Gathered { source: values, .. } => values.reaches_storage(),
Body::Nested { child, .. } => child.reaches_storage(),
Body::Fields { children } => children.iter().any(|child| child.reaches_storage()),
_ => false,
}
}
#[must_use]
pub fn signed_at(&self, index: usize) -> Option<i128> {
if index >= self.len || !self.validity.is_valid(index) {
return None;
}
match &self.body {
Body::Flat(data) => data.signed_at(index),
Body::Constant(value) => match value.as_ref() {
Value::TinyInt(x) => Some(i128::from(*x)),
Value::SmallInt(x) => Some(i128::from(*x)),
Value::Integer(x) | Value::Date(x) => Some(i128::from(*x)),
Value::BigInt(x) | Value::Time(x) | Value::Timestamp(x) => Some(i128::from(*x)),
Value::HugeInt(x) | Value::Decimal { unscaled: x, .. } => Some(*x),
_ => None,
},
Body::Sequence { start, step } => {
Some(i128::from(start.wrapping_add(step.wrapping_mul(index as i64))))
}
Body::Dictionary { codes, values, .. } => {
values.signed_at(usize::try_from(*codes.get(index)?).ok()?)
}
Body::Runs { ends, values } => values.signed_at(run_holding(ends, index)?),
Body::Gathered { source, rids, offset } => {
source.signed_at(row_of(rids, *offset, index)?)
}
Body::Packed { words, width, base, offset } => Some(
*base + i128::from(code_at(words, (*offset + index) * *width as usize, *width)),
),
Body::Coded { .. }
| Body::Views { .. }
| Body::ExternalText { .. }
| Body::Nested { .. }
| Body::Fields { .. } => None,
}
}
#[must_use]
pub fn signed_gather(&self, at: &[u32], out: &mut Vec<i64>) -> bool {
out.clear();
match &self.body {
Body::Flat(data) => data.signed_gather(self.len, at, out),
_ => false,
}
}
#[must_use]
pub fn signed_runs(
&self,
(from, to): (usize, usize),
every: usize,
out: &mut Vec<(i64, usize)>,
) -> bool {
out.clear();
match &self.body {
Body::Flat(data) => data.signed_runs(self.len, (from, to), every, out),
_ => false,
}
}
#[must_use]
pub fn signed_block(&self, out: &mut Vec<i64>) -> bool {
out.clear();
match &self.body {
Body::Flat(data) => data.signed_block(self.len, out),
Body::Constant(value) => {
let held = match value.as_ref() {
Value::TinyInt(x) => i64::from(*x),
Value::SmallInt(x) => i64::from(*x),
Value::Integer(x) | Value::Date(x) => i64::from(*x),
Value::BigInt(x) | Value::Time(x) | Value::Timestamp(x) => *x,
_ => return false,
};
out.resize(self.len, held);
true
}
Body::Sequence { start, step } => {
out.extend(
(0..self.len).map(|index| start.wrapping_add(step.wrapping_mul(index as i64))),
);
true
}
Body::Packed { words, width, base, offset } => match i64::try_from(*base) {
Ok(base) => {
let packed =
Packed { words, width: *width, base: i128::from(base), offset: *offset };
let mut block = [0u64; 64];
let mut from = 0;
out.reserve(self.len);
while from < self.len {
let rows = (64 - (*offset + from) % 64).min(self.len - from);
let codes = &mut block[..rows];
packed.unpack(from, codes);
out.extend(codes.iter().map(|&code| base.wrapping_add(code as i64)));
from += rows;
}
true
}
Err(_) => false,
},
Body::Dictionary { codes, values, .. } => {
if let Some((start, step)) = values.sequence_parts() {
let Some(codes) = codes.get(..self.len) else {
return false;
};
if codes.iter().any(|&code| code as usize >= values.len()) {
return false;
}
out.extend(
codes
.iter()
.map(|&code| start.wrapping_add(step.wrapping_mul(i64::from(code)))),
);
return true;
}
let mut entries = Vec::new();
if !values.none_null() || !values.signed_block(&mut entries) {
return false;
}
let Some(codes) = codes.get(..self.len) else {
return false;
};
out.reserve(codes.len());
for &code in codes {
match entries.get(code as usize) {
Some(&entry) => out.push(entry),
None => {
out.clear();
return false;
}
}
}
true
}
Body::Runs { .. }
| Body::Gathered { .. }
| Body::Coded { .. }
| Body::Views { .. }
| Body::ExternalText { .. }
| Body::Nested { .. }
| Body::Fields { .. } => false,
}
}
#[must_use]
pub fn none_null(&self) -> bool {
if self.validity.has_nulls(self.len) {
return false;
}
match &self.body {
Body::Dictionary { values, .. } | Body::Runs { values, .. } => values.none_null(),
Body::Gathered { source, rids, offset } => {
source.none_null()
&& !rids[*offset..].iter().take(self.len).any(|&rid| rid == NO_ROW)
}
_ => true,
}
}
pub fn iter(&self) -> impl Iterator<Item = Value> + '_ {
(0..self.len).map(|index| self.value_at(index))
}
#[must_use]
pub fn into_pages(self) -> Self {
let body = match self.body {
Body::Flat(data) => Body::Flat(data.into_pages()),
Body::Dictionary { codes, values, stable } => {
Body::Dictionary { codes: codes.into_page(), values, stable }
}
Body::Views { views, arena } => Body::Views { views, arena: paged(arena) },
other => other,
};
Self { body, ..self }
}
pub fn slice(&self, at: usize, len: usize) -> Result<Self> {
let end = at.checked_add(len).ok_or_else(|| Error::internal("a slice that wraps"))?;
if end > self.len {
return Err(Error::internal(format!("rows {at} to {end} of a vector of {}", self.len)));
}
if at == 0 && len == self.len {
return Ok(self.clone());
}
let validity = self.validity.slice(at, len);
let body = match &self.body {
Body::Constant(value) => Body::Constant(value.clone()),
Body::Sequence { start, step } => {
Body::Sequence { start: start + step * at as i64, step: *step }
}
Body::Dictionary { codes, values, stable } => Body::Dictionary {
codes: codes.slice(at, len),
values: Arc::clone(values),
stable: *stable,
},
Body::Gathered { source, rids, offset } => Body::Gathered {
source: Arc::clone(source),
rids: Arc::clone(rids),
offset: offset + at,
},
Body::Packed { words, width, base, offset } => Body::Packed {
words: Arc::clone(words),
width: *width,
base: *base,
offset: offset + at,
},
Body::Views { views, arena } => {
Body::Views { views: views[at..end].to_vec(), arena: Arc::clone(arena) }
}
Body::Coded { codes, spans, table } => Body::Coded {
codes: Arc::clone(codes),
spans: spans[at..end].to_vec(),
table: Arc::clone(table),
},
Body::Runs { ends, values } if len > 0 => {
let first = run_holding(ends, at).unwrap_or(0);
let last = run_holding(ends, end - 1).unwrap_or(first);
let cut: Vec<u32> = ends[first..=last]
.iter()
.map(|&stop| stop.min(end as u32) - at as u32)
.collect();
let values = values.slice(first, last - first + 1)?;
Body::Runs { ends: cut, values: Arc::new(values) }
}
Body::Runs { .. } => return self.gather(&[]),
Body::Nested { entries, child } => {
Body::Nested { entries: entries[at..end].to_vec(), child: Arc::clone(child) }
}
Body::Fields { children } => Body::Fields {
children: children
.iter()
.map(|child| child.slice(at, len).map(Arc::new))
.collect::<Result<Vec<_>>>()?,
},
Body::ExternalText { source } => {
let mut out = StringColumn::with_capacity(len);
for index in at..end {
out.push_bytes(source.bytes_at(index)?.unwrap_or_default());
}
Body::Flat(Data::Varlen(out))
}
Body::Flat(data) => Body::Flat(run_of(data, at, end)),
};
Ok(Self { ty: self.ty.clone(), len, validity, body })
}
pub fn flatten(&self) -> Result<Self> {
if let Body::Flat(_) = self.body {
return Ok(self.clone());
}
slow::took(Cause::Flatten);
if let Some(flat) = self.decoded_codes() {
return Ok(flat);
}
self.copied((0..self.len).collect(), false)
}
fn decoded_codes(&self) -> Option<Self> {
let Body::Dictionary { codes, values, .. } = &self.body else {
return None;
};
if !matches!(self.validity, Validity::AllValid)
|| !matches!(values.validity, Validity::AllValid)
{
return None;
}
let Body::Flat(data) = &values.body else {
return None;
};
if matches!(data, Data::Empty) {
return None;
}
let codes = codes.as_slice().get(..self.len)?;
if !below(codes, values.len) {
return None;
}
let at = codes.iter().map(|&code| code as usize).collect::<Vec<_>>();
Some(Self {
ty: self.ty.clone(),
len: self.len,
validity: Validity::AllValid,
body: Body::Flat(copy_of(data, &at)),
})
}
pub fn opened(&self) -> Result<Self> {
if let Body::Flat(_) = self.body {
return Ok(self.clone());
}
if let Some(flat) = self.decoded_codes() {
return Ok(flat);
}
self.copied((0..self.len).collect(), false)
}
pub fn into_flat(self) -> Result<Self> {
if let Body::Flat(_) = self.body {
return Ok(self);
}
self.flatten()
}
pub fn gather(&self, indices: &[u32]) -> Result<Self> {
if let Body::Dictionary { codes, values, stable: true } = &self.body {
let inside = below(indices, codes.len());
return self.stable_gathered(codes, values, indices, inside, |index| index as usize);
}
if let Body::Constant(value) = &self.body {
let null = value.is_null() && matches!(self.validity, Validity::AllInvalid);
let valid = matches!(self.validity, Validity::AllValid) && !value.is_null();
if null || (valid && below(indices, self.len)) {
return Ok(Self::constant(self.ty.clone(), value.as_ref().clone(), indices.len()));
}
}
if let Some(gathered) = self.unpacked_at(indices) {
return Ok(gathered);
}
if let Some(gathered) = self.flat_at(indices) {
return Ok(gathered);
}
self.copied(indices.iter().map(|&index| index as usize).collect(), true)
}
fn flat_at(&self, indices: &[u32]) -> Option<Self> {
let Body::Flat(data) = &self.body else { return None };
if self.validity.has_nulls(self.len) {
return None;
}
if !below(indices, self.len) {
return None;
}
macro_rules! gathered {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
$(Data::$variant(values) => {
let values = values.as_slice();
let out: Vec<$native> =
indices.iter().map(|&index| values[index as usize]).collect();
Data::$variant(Buffer::from_vec(out))
})+
Data::Empty | Data::Varlen(_) => return None,
}
};
}
let data = crate::for_each_layout!(fixed, gathered);
Some(Self {
ty: self.ty.clone(),
len: indices.len(),
validity: Validity::AllValid,
body: Body::Flat(data),
})
}
fn stable_gathered<T: Copy>(
&self,
codes: &Buffer<u32>,
values: &Arc<Vector>,
at: &[T],
inside: bool,
index: impl Fn(T) -> usize,
) -> Result<Self> {
let rows = at.len();
if inside && self.never_null() {
return Ok(Self {
ty: values.ty.clone(),
len: rows,
validity: Validity::AllValid,
body: Body::Dictionary {
codes: at.iter().map(|&at| codes[index(at)]).collect(),
values: Arc::clone(values),
stable: true,
},
});
}
let validity = if self.never_null() && at.iter().all(|&at| index(at) < self.len) {
Validity::AllValid
} else {
Validity::from_iter(rows, |row| {
at.get(row)
.map(|&at| index(at))
.is_some_and(|index| index < self.len && !self.is_null_at(index))
})
};
let gathered: Vec<u32> =
at.iter().map(|&at| codes.get(index(at)).copied().unwrap_or(0)).collect();
if matches!(values.body, Body::Dictionary { .. }) {
return Ok(
Self::stable_dictionary(gathered, Arc::clone(values))?.with_validity(validity)
);
}
let highest = (values.is_empty() && !gathered.is_empty()).then_some(0);
Ok(Self::stable_dictionary_validated(gathered, Arc::clone(values), highest)?
.with_validity(validity))
}
fn unpacked_at(&self, indices: &[u32]) -> Option<Self> {
let Body::Packed { words, width, base, offset } = &self.body else {
return None;
};
if self.validity.has_nulls(self.len) {
return None;
}
if !below(indices, self.len) {
return None;
}
let packed = Packed { words, width: *width, base: *base, offset: *offset };
let low = i64::try_from(packed.base()).ok()?;
i64::try_from(packed.ceiling()).ok()?;
#[expect(clippy::cast_possible_wrap, reason = "a code is below the span, which fits")]
let value = |code: u64| low.wrapping_add(code as i64);
#[expect(clippy::cast_possible_truncation, reason = "the layout holds every value")]
let data = match self.ty.physical() {
rudb_common::PhysicalType::Int64 => {
Data::Int64(Buffer::from_vec(packed.values_at(indices, value)))
}
rudb_common::PhysicalType::Int32 => {
Data::Int32(Buffer::from_vec(packed.values_at(indices, |code| value(code) as i32)))
}
rudb_common::PhysicalType::Int16 => {
Data::Int16(Buffer::from_vec(packed.values_at(indices, |code| value(code) as i16)))
}
_ => return None,
};
Some(Self {
ty: self.ty.clone(),
len: indices.len(),
validity: Validity::AllValid,
body: Body::Flat(data),
})
}
fn copied(&self, at: Vec<usize>, forms_stay: bool) -> Result<Self> {
let rows = at.len();
if forms_stay {
if let Body::Dictionary { codes, values, stable: true } = &self.body {
let inside = at.iter().max().is_none_or(|&top| top < codes.len());
return self.stable_gathered(codes, values, &at, inside, |index| index);
}
}
let (at, leaf) = self.resolve(at);
let live: Vec<bool> = at.iter().map(|&index| index != NOWHERE).collect();
let validity = Validity::from_run(&live);
let body = match &leaf.body {
Body::Constant(value)
if matches!(
self.ty,
LogicalType::List(_) | LogicalType::Struct(_) | LogicalType::Map(_, _)
) =>
{
if forms_stay && matches!(validity, Validity::AllValid) {
return Ok(Self::constant(self.ty.clone(), value.as_ref().clone(), rows));
}
let rows: Vec<Value> = at
.iter()
.map(
|&index| {
if index == NOWHERE { Value::Null } else { value.as_ref().clone() }
},
)
.collect();
return Self::from_values(self.ty.clone(), &rows);
}
Body::Constant(value) => {
if forms_stay && matches!(validity, Validity::AllValid) {
return Ok(Self::constant(self.ty.clone(), value.as_ref().clone(), rows));
}
let mut data = empty_data_for(&self.ty)?;
let value = stored(&self.ty, value)?;
for &index in &at {
push_value(&mut data, if index == NOWHERE { &Value::Null } else { &value })?;
}
Body::Flat(data)
}
Body::Sequence { start, step } => Body::Flat(Data::Int64(
at.iter()
.map(|&index| if index == NOWHERE { 0 } else { start + step * index as i64 })
.collect(),
)),
Body::Flat(Data::Empty) => {
return Ok(Self::constant(self.ty.clone(), Value::Null, rows));
}
Body::Flat(data) => Body::Flat(copy_of(data, &at)),
Body::Packed { words, width, base, offset } => {
Body::Flat(unpack(&self.ty, words, *offset, *width, *base, &at)?)
}
Body::Views { views, arena } if forms_stay => Body::Views {
views: at
.iter()
.map(|&index| views.get(index).copied().unwrap_or_else(StringView::empty))
.collect(),
arena: Arc::clone(arena),
},
Body::Views { views, arena } if arena.is_shared() => {
Body::Flat(Data::Varlen(StringColumn::from_parts(
at.iter()
.map(|&index| views.get(index).copied().unwrap_or_else(StringView::empty))
.collect(),
(**arena).clone(),
)))
}
Body::Views { views, arena } => {
let mut out = StringColumn::with_capacity(at.len());
out.reserve_bytes(
at.iter()
.filter_map(|&index| views.get(index))
.filter(|view| !view.is_inline())
.map(StringView::len)
.sum(),
);
for &index in &at {
let bytes = views.get(index).and_then(|view| view.bytes_in(arena));
out.push_bytes(bytes.unwrap_or_default());
}
Body::Flat(Data::Varlen(out))
}
Body::ExternalText { source } => {
let mut out = StringColumn::with_capacity(at.len());
for &index in &at {
out.push_bytes(source.bytes_at(index)?.unwrap_or_default());
}
Body::Flat(Data::Varlen(out))
}
Body::Coded { codes, spans, table } if forms_stay => Body::Coded {
codes: Arc::clone(codes),
spans: at
.iter()
.map(|&index| spans.get(index).copied().unwrap_or((0, 0)))
.collect(),
table: Arc::clone(table),
},
Body::Coded { codes, spans, table } => {
let mut out = StringColumn::with_capacity(at.len());
let mut scratch = Vec::new();
for &index in &at {
scratch.clear();
let span = spans
.get(index)
.and_then(|&(from, to)| codes.get(from as usize..to as usize));
if let Some(span) = span {
table.decompress(span, &mut scratch)?;
}
out.push_bytes(&scratch);
}
Body::Flat(Data::Varlen(out))
}
Body::Nested { entries, child } => Body::Nested {
entries: at
.iter()
.map(|&index| entries.get(index).copied().unwrap_or((0, 0)))
.collect(),
child: Arc::clone(child),
},
Body::Fields { children } => Body::Fields {
children: children
.iter()
.map(|child| child.copied(at.clone(), forms_stay).map(Arc::new))
.collect::<Result<Vec<_>>>()?,
},
Body::Dictionary { .. } | Body::Runs { .. } | Body::Gathered { .. } => {
return Err(Error::internal(
"a form that points somewhere survived being resolved",
));
}
};
Ok(Self { ty: self.ty.clone(), len: rows, validity, body })
}
fn resolve(&self, mut at: Vec<usize>) -> (Vec<usize>, &Self) {
let mut source = self;
loop {
for slot in &mut at {
if *slot >= source.len || !source.validity.is_valid(*slot) {
*slot = NOWHERE;
}
}
source = match &source.body {
Body::Dictionary { codes, values, .. } => {
for slot in &mut at {
*slot = match codes.get(*slot) {
Some(&code) => code as usize,
None => NOWHERE,
};
}
values.as_ref()
}
Body::Runs { ends, values } => {
for slot in &mut at {
*slot = run_holding(ends, *slot).unwrap_or(NOWHERE);
}
values.as_ref()
}
Body::Gathered { source: below, rids, offset } => {
for slot in &mut at {
*slot = if *slot == NOWHERE {
NOWHERE
} else {
row_of(rids, *offset, *slot).unwrap_or(NOWHERE)
};
}
below.as_ref()
}
_ => return (at, source),
};
}
}
}
impl AsRef<Vector> for Vector {
fn as_ref(&self) -> &Vector {
self
}
}
#[derive(Debug, Clone, Copy)]
pub struct Packed<'a> {
words: &'a [u64],
width: u32,
base: i128,
offset: usize,
}
impl Packed<'_> {
#[must_use]
pub fn words(&self) -> &[u64] {
self.words
}
#[must_use]
pub fn offset(&self) -> usize {
self.offset
}
#[must_use]
pub fn width(&self) -> u32 {
self.width
}
#[must_use]
pub fn base(&self) -> i128 {
self.base
}
#[must_use]
pub fn ceiling(&self) -> i128 {
self.base + i128::from(u64::MAX >> (u64::BITS - self.width))
}
#[must_use]
#[inline]
pub fn code(&self, row: usize) -> u64 {
code_at(self.words, (self.offset + row) * self.width as usize, self.width)
}
#[must_use]
pub fn code_of(&self, value: i128) -> Option<u64> {
u64::try_from(value.checked_sub(self.base)?).ok().filter(|&code| code <= self.mask())
}
fn mask(&self) -> u64 {
u64::MAX >> (u64::BITS - self.width)
}
pub fn unpack(&self, from: usize, out: &mut [u64]) {
let width = self.width as usize;
let start = self.offset + from;
let end = start + out.len();
let first = start.next_multiple_of(64).min(end);
let mut at = 0;
for row in start..first {
out[at] = code_at(self.words, row * width, self.width);
at += 1;
}
let mut row = first;
while row + 64 <= end {
let word = row / 64 * width;
let Some(words) = self.words.get(word..word + width) else { break };
let Some(Ok(block)) = out.get_mut(at..at + 64).map(<&mut [u64; 64]>::try_from) else {
break;
};
unpack_block(words, self.width, block);
row += 64;
at += 64;
}
for row in row..end {
out[at] = code_at(self.words, row * width, self.width);
at += 1;
}
}
pub fn unpack_mapped<U: Copy>(
&self,
from: usize,
rows: usize,
out: &mut Vec<U>,
value: impl Fn(u64) -> U,
) {
let width = self.width as usize;
let start = self.offset + from;
let end = start + rows;
let first = start.next_multiple_of(64).min(end);
out.reserve(rows);
for row in start..first {
out.push(value(code_at(self.words, row * width, self.width)));
}
let mut row = first;
let mut block = [0_u64; 64];
while row + 64 <= end {
let word = row / 64 * width;
let Some(words) = self.words.get(word..word + width) else { break };
unpack_block(words, self.width, &mut block);
out.extend(block.iter().map(|&code| value(code)));
row += 64;
}
for row in row..end {
out.push(value(code_at(self.words, row * width, self.width)));
}
}
pub fn codes_at<M: Fn(usize) -> usize>(&self, at: M, rows: usize) -> Vec<u64> {
let mut codes = vec![0; rows];
self.codes_into(at, rows, &mut codes);
codes
}
pub fn codes_into<M: Fn(usize) -> usize>(&self, at: M, rows: usize, out: &mut Vec<u64>) {
thread_local! {
static SPAN: RefCell<Vec<u64>> = const { RefCell::new(Vec::new()) };
}
if out.len() < rows {
out.resize(rows, 0);
}
if rows == 0 {
return;
}
let first = at(0);
let (mut low, mut high) = (first, first);
let mut ascends = true;
for index in 1..rows {
let row = at(index);
low = low.min(row);
high = high.max(row);
ascends &= row == first + index;
}
if ascends {
self.unpack(first, &mut out[..rows]);
return;
}
if high - low >= rows.saturating_mul(4) {
for (index, code) in out[..rows].iter_mut().enumerate() {
*code = self.code(at(index));
}
return;
}
let span = high - low + 1;
let mut run = SPAN.with_borrow_mut(std::mem::take);
if run.len() < span {
run.resize(span, 0);
}
self.unpack(low, &mut run[..span]);
for (index, code) in out[..rows].iter_mut().enumerate() {
*code = run[at(index) - low];
}
SPAN.with_borrow_mut(|held| *held = run);
}
pub fn values_at<T>(&self, at: &[u32], value: impl Fn(u64) -> T) -> Vec<T> {
thread_local! {
static SPAN: RefCell<Vec<u64>> = const { RefCell::new(Vec::new()) };
}
let Some((low, high)) = extent(at) else { return Vec::new() };
let (low, high) = (low as usize, high as usize);
if high - low >= at.len().saturating_mul(4) {
return at.iter().map(|&row| value(self.code(row as usize))).collect();
}
let span = high - low + 1;
let gathered = |run: &mut Vec<u64>| {
if run.len() < span {
run.resize(span, 0);
}
let run = &mut run[..span];
self.unpack(low, run);
at.iter().map(|&row| value(run[row as usize - low])).collect()
};
SPAN.with(|held| match held.try_borrow_mut() {
Ok(mut held) => gathered(&mut held),
Err(_) => gathered(&mut Vec::new()),
})
}
}
fn unpack_block(words: &[u64], width: u32, out: &mut [u64; 64]) {
macro_rules! widths {
($($width:literal)*) => {
match width {
$($width => unpack_width::<$width>(words, out),)*
_ => {
for (at, code) in out.iter_mut().enumerate() {
*code = code_at(words, at * width as usize, width);
}
}
}
};
}
widths!(1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32
33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63);
}
#[inline(always)]
fn unpack_width<const WIDTH: usize>(words: &[u64], out: &mut [u64; 64]) {
let Ok(words) = <&[u64; WIDTH]>::try_from(&words[..WIDTH]) else { return };
macro_rules! steps {
($($at:literal)*) => {
$(unpack_step::<WIDTH, $at>(words, out);)*
};
}
steps!(0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32
33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63);
}
#[inline(always)]
fn unpack_step<const WIDTH: usize, const AT: usize>(words: &[u64; WIDTH], out: &mut [u64; 64]) {
let bit = AT * WIDTH;
let word = bit / 64;
let shift = bit % 64;
let mut value = words[word] >> shift;
if shift + WIDTH > 64 {
value |= words[word + 1] << (64 - shift);
}
out[AT] = value & (u64::MAX >> (64 - WIDTH));
}
pub const PACKED_WIDTH_MAX: u32 = 63;
pub const PACKING_PAYS_AT: usize = 2;
pub const FSST_PAYS_AT: usize = 2;
#[derive(Debug, Clone, Copy)]
pub struct Coded<'a> {
codes: &'a [u8],
spans: &'a [(u32, u32)],
table: &'a SymbolTable,
}
impl Coded<'_> {
#[must_use]
pub fn table(&self) -> &SymbolTable {
self.table
}
#[must_use]
pub fn row(&self, row: usize) -> Option<&[u8]> {
let &(from, to) = self.spans.get(row)?;
self.codes.get(from as usize..to as usize)
}
#[must_use]
pub fn encode(&self, bytes: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(bytes.len());
self.table.compress(bytes, &mut out);
out
}
}
fn widen<T: Copy + Into<i64>>(run: &[T], len: usize, out: &mut Vec<i64>) -> bool {
match run.get(..len) {
Some(run) => {
out.extend(run.iter().map(|&x| x.into()));
true
}
None => false,
}
}
fn gather_widened<T: Copy + Into<i64>>(
run: &[T],
len: usize,
at: &[u32],
out: &mut Vec<i64>,
) -> bool {
let Some(run) = run.get(..len) else {
return false;
};
if !below(at, run.len()) {
return false;
}
out.extend(at.iter().map(|&row| run[row as usize].into()));
true
}
fn runs_widened<T: Copy + Eq + Into<i64>>(
run: &[T],
len: usize,
(from, to): (usize, usize),
every: usize,
out: &mut Vec<(i64, usize)>,
) -> bool {
out.clear();
let Some(values) = run.get(..len).and_then(|run| run.get(from..to)) else {
return false;
};
let Some(&first) = values.first() else {
return true;
};
let mut current = first;
for (block, stretch) in values.chunks(16).enumerate() {
if !stretch.iter().fold(false, |differ, &value| differ | (value != current)) {
continue;
}
let start = from + block * 16;
for (row, &value) in stretch.iter().enumerate() {
if value != current {
out.push((current.into(), start + row));
current = value;
}
}
if out.len() > (block * 16) / every.max(1) + 64 {
out.clear();
return false;
}
}
out.push((current.into(), to));
true
}
fn share<T: ?Sized>(bytes: usize, held: &Arc<T>) -> usize {
bytes / Arc::strong_count(held).max(1)
}
fn words_for(len: usize, width: u32) -> usize {
(len * width as usize).div_ceil(u64::BITS as usize)
}
fn layout_range(ty: &LogicalType) -> Option<(i128, i128)> {
use rudb_common::PhysicalType as P;
macro_rules! ranges {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match ty.physical() {
$(P::$variant => Some((i128::from(<$native>::MIN), i128::from(<$native>::MAX))),)+
_ => None,
}
};
}
crate::for_each_layout!(exact, ranges)
}
fn flat_bytes(data: &Data, len: usize) -> usize {
macro_rules! widths {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
Data::Empty => 0,
$(Data::$variant(_) => len * size_of::<$native>(),)+
}
};
}
crate::for_each_layout!(all, widths)
}
fn packing_base(ty: &LogicalType, low: i128, high: i128, width: u32) -> Option<i128> {
let (floor, ceiling) = layout_range(ty)?;
let span = i128::from(u64::MAX >> (64 - width));
let base = low.min(ceiling - span);
(base >= floor && base >= high - span).then_some(base)
}
fn span_of(data: &Data, len: usize) -> Option<(i128, i128)> {
macro_rules! spans {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
$(Data::$variant(values) => {
let values = values.as_slice();
let values = &values[..len.min(values.len())];
let low = values.iter().copied().min()?;
let high = values.iter().copied().max()?;
Some((i128::from(low), i128::from(high)))
})+
_ => None,
}
};
}
crate::for_each_layout!(exact, spans)
}
fn pack(data: &Data, len: usize, base: i128, width: u32) -> Vec<u64> {
let mut words = vec![0u64; words_for(len, width)];
macro_rules! packing {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
$(Data::$variant(values) => {
for (row, &value) in values.as_slice().iter().take(len).enumerate() {
let code = u64::try_from(i128::from(value) - base).unwrap_or(0);
write_code(&mut words, row * width as usize, width, code);
}
})+
_ => {}
}
};
}
crate::for_each_layout!(exact, packing);
words
}
fn unpack(
ty: &LogicalType,
words: &[u64],
offset: usize,
width: u32,
base: i128,
at: &[usize],
) -> Result<Data> {
let mut out = empty_data_for(ty)?;
let value_of = |row: usize| {
if row == NOWHERE {
return None;
}
Some(base + i128::from(code_at(words, (offset + row) * width as usize, width)))
};
macro_rules! unpacking {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match &mut out {
$(Data::$variant(values) => {
values.reserve(at.len());
for &row in at {
let value = value_of(row)
.and_then(|value| <$native>::try_from(value).ok())
.unwrap_or($zero);
values.push(value);
}
})+
_ => {
return Err(Error::internal(format!(
"a {ty} vector was packed, which no integer layout allows"
)));
}
}
};
}
crate::for_each_layout!(exact, unpacking);
Ok(out)
}
#[inline]
fn code_at(words: &[u64], bit: usize, width: u32) -> u64 {
let word = bit / u64::BITS as usize;
let shift = (bit % u64::BITS as usize) as u32;
let mask = u64::MAX >> (u64::BITS - width);
let low = words.get(word).copied().unwrap_or(0) >> shift;
let taken = u64::BITS - shift;
if taken >= width {
return low & mask;
}
let high = words.get(word + 1).copied().unwrap_or(0) << taken;
(low | high) & mask
}
fn write_code(words: &mut [u64], bit: usize, width: u32, code: u64) {
let word = bit / u64::BITS as usize;
let shift = (bit % u64::BITS as usize) as u32;
words[word] |= code << shift;
let taken = u64::BITS - shift;
if taken < width {
words[word + 1] |= code >> taken;
}
}
fn compose(codes: Vec<u32>, values: Arc<Vector>) -> (Vec<u32>, Arc<Vector>) {
if !matches!(values.validity, Validity::AllValid) {
return (codes, values);
}
let Body::Dictionary { codes: inner, values: leaf, .. } = &values.body else {
return (codes, values);
};
debug_assert!(
!matches!(leaf.body, Body::Dictionary { .. })
|| !matches!(leaf.validity, Validity::AllValid),
"a dictionary was stacked on a dictionary without going through the constructor"
);
let composed = codes.iter().map(|&code| inner[code as usize]).collect();
(composed, Arc::clone(leaf))
}
const RUNS_PAY_AT: usize = 2;
fn paged(arena: Arc<Buffer<u8>>) -> Arc<Buffer<u8>> {
if arena.is_shared() {
return arena;
}
match Arc::try_unwrap(arena) {
Ok(owned) => Arc::new(owned.into_page()),
Err(held) => held,
}
}
fn run_holding(ends: &[u32], row: usize) -> Option<usize> {
let row = u32::try_from(row).ok()?;
let run = match ends.binary_search(&row) {
Ok(at) => at + 1,
Err(at) => at,
};
(run < ends.len()).then_some(run)
}
fn boundaries(data: &Data, validity: &Validity, len: usize) -> Vec<u32> {
if len == 0 {
return Vec::new();
}
let breaks = |ends: &mut Vec<u32>, mut differs: Box<dyn FnMut(usize, usize) -> bool + '_>| {
for row in 1..len {
let same = match (validity.is_valid(row), validity.is_valid(row - 1)) {
(false, false) => true,
(true, true) => !differs(row, row - 1),
_ => false,
};
if !same {
ends.push(u32::try_from(row).unwrap_or(u32::MAX));
}
}
ends.push(u32::try_from(len).unwrap_or(u32::MAX));
};
let mut ends = Vec::new();
macro_rules! walked {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
Data::Empty => ends.push(u32::try_from(len).unwrap_or(u32::MAX)),
$(Data::$variant(values) => {
breaks(&mut ends, Box::new(|a, b| values.get(a) != values.get(b)));
})+
Data::Varlen(values) => {
breaks(&mut ends, Box::new(|a, b| values.bytes(a) != values.bytes(b)));
}
}
};
}
crate::for_each_layout!(fixed, walked);
ends
}
pub(crate) const NOWHERE: usize = usize::MAX;
pub const NO_ROW: u32 = u32::MAX;
fn row_of(rids: &[u32], offset: usize, index: usize) -> Option<usize> {
match rids.get(offset + index) {
Some(&NO_ROW) | None => None,
Some(&rid) => Some(rid as usize),
}
}
fn run_of(data: &Data, at: usize, end: usize) -> Data {
macro_rules! run {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
Data::Empty => Data::Empty,
$(Data::$variant(values) => {
let held = values.len();
let from = at.min(held);
let to = end.max(from).min(held);
if to == end {
Data::$variant(values.slice(from, end - from))
} else {
let values = values.as_slice();
let mut out = Buffer::with_capacity(end - at);
out.extend_from_slice(&values[from..to]);
for _ in to..end {
out.push($zero);
}
Data::$variant(out)
}
})+
Data::Varlen(values) => {
if let Some(shared) =
values.window(at, end).or_else(|| values.viewing(at..end))
{
return Data::Varlen(shared);
}
let views = values.views();
let mut out = StringColumn::with_capacity(end - at);
out.reserve_bytes(
views
.get(at.min(views.len())..end.min(views.len()))
.unwrap_or(&[])
.iter()
.filter(|view| !view.is_inline())
.map(StringView::len)
.sum(),
);
for index in at..end {
out.push_from(values, index);
}
Data::Varlen(out)
}
}
};
}
crate::for_each_layout!(fixed, run)
}
pub(crate) fn placed_of(data: &Data, inverse: &[u32]) -> Data {
macro_rules! placed {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
$(Data::$variant(values) => {
let mut out: Vec<$native> = vec![$zero; inverse.len()];
for (value, &to) in values.as_slice().iter().zip(inverse) {
if let Some(slot) = out.get_mut(to as usize) {
*slot = *value;
}
}
Data::$variant(Buffer::from_vec(out))
})+
Data::Empty => Data::Empty,
Data::Varlen(_) => {
let mut at = vec![NOWHERE; inverse.len()];
for (row, &to) in inverse.iter().enumerate() {
if let Some(slot) = at.get_mut(to as usize) {
*slot = row;
}
}
copy_of(data, &at)
}
}
};
}
crate::for_each_layout!(fixed, placed)
}
pub(crate) fn copy_of(data: &Data, at: &[usize]) -> Data {
macro_rules! copied {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
Data::Empty => Data::Empty,
$(Data::$variant(values) => {
let values = values.as_slice();
let mut out: Vec<$native> = Vec::with_capacity(at.len());
out.extend(at.iter().map(|&index| values.get(index).copied().unwrap_or($zero)));
Data::$variant(Buffer::from_vec(out))
})+
Data::Varlen(values) => {
if let Some(shared) = values.viewing(at.iter().copied()) {
return Data::Varlen(shared);
}
let mut out = StringColumn::with_capacity(at.len());
let views = values.views();
out.reserve_bytes(
at.iter()
.filter_map(|&index| views.get(index))
.filter(|view| !view.is_inline())
.map(StringView::len)
.sum(),
);
for &index in at {
out.push_from(values, index);
}
Data::Varlen(out)
}
}
};
}
crate::for_each_layout!(fixed, copied)
}
pub(crate) fn layout_of(data: &Data) -> rudb_common::PhysicalType {
use rudb_common::PhysicalType as P;
macro_rules! layouts {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match data {
Data::Empty => P::Empty,
$(Data::$variant(_) => P::$variant,)+
}
};
}
crate::for_each_layout!(all, layouts)
}
fn value_from(ty: &LogicalType, data: &Data, index: usize) -> Value {
let signed = || data.signed_at(index);
let unsigned = || data.unsigned_at(index);
let value = match ty {
LogicalType::Boolean => match data {
Data::Bool(v) => v.get(index).map(|&x| Value::Boolean(x)),
_ => None,
},
LogicalType::TinyInt => signed().and_then(|x| i8::try_from(x).ok()).map(Value::TinyInt),
LogicalType::SmallInt => signed().and_then(|x| i16::try_from(x).ok()).map(Value::SmallInt),
LogicalType::Integer => signed().and_then(|x| i32::try_from(x).ok()).map(Value::Integer),
LogicalType::BigInt => signed().and_then(|x| i64::try_from(x).ok()).map(Value::BigInt),
LogicalType::HugeInt => signed().map(Value::HugeInt),
LogicalType::UTinyInt => unsigned().and_then(|x| u8::try_from(x).ok()).map(Value::UTinyInt),
LogicalType::USmallInt => {
unsigned().and_then(|x| u16::try_from(x).ok()).map(Value::USmallInt)
}
LogicalType::UInteger => {
unsigned().and_then(|x| u32::try_from(x).ok()).map(Value::UInteger)
}
LogicalType::UBigInt => unsigned().and_then(|x| u64::try_from(x).ok()).map(Value::UBigInt),
LogicalType::UHugeInt => unsigned().map(Value::UHugeInt),
LogicalType::Float => match data {
Data::Float32(v) => v.get(index).map(|&x| Value::Float(x)),
_ => None,
},
LogicalType::Double => match data {
Data::Float64(v) => v.get(index).map(|&x| Value::Double(x)),
_ => None,
},
LogicalType::Decimal { width, scale } => {
signed().map(|unscaled| Value::Decimal { unscaled, width: *width, scale: *scale })
}
LogicalType::Varchar | LogicalType::Blob | LogicalType::Bit => {
data.bytes_at(index).map(|bytes| bytes_as(ty, bytes))
}
LogicalType::Enum(labels) => unsigned()
.and_then(|code| labels.get(usize::try_from(code).ok()?))
.map(|label| Value::Varchar(label.clone())),
LogicalType::Date => signed().and_then(|x| i32::try_from(x).ok()).map(Value::Date),
LogicalType::Time => signed().and_then(|x| i64::try_from(x).ok()).map(Value::Time),
LogicalType::TimeTz => signed().and_then(|x| i64::try_from(x).ok()).map(Value::TimeTz),
LogicalType::Timestamp
| LogicalType::TimestampS
| LogicalType::TimestampMs
| LogicalType::TimestampNs => {
signed().and_then(|x| i64::try_from(x).ok()).map(Value::Timestamp)
}
LogicalType::TimestampTz => {
signed().and_then(|x| i64::try_from(x).ok()).map(Value::TimestampTz)
}
LogicalType::Interval => match data {
Data::Interval(v) => {
v.get(index).map(|&(months, days, micros)| Value::Interval { months, days, micros })
}
_ => None,
},
_ => None,
};
value.unwrap_or(Value::Null)
}
fn fields_of(ty: &LogicalType) -> &[Field] {
match ty {
LogicalType::Struct(fields) => fields,
_ => &[],
}
}
fn bytes_as(ty: &LogicalType, bytes: &[u8]) -> Value {
match ty {
LogicalType::Varchar => {
std::str::from_utf8(bytes).map_or(Value::Null, |text| Value::Varchar(text.to_owned()))
}
LogicalType::Blob | LogicalType::Bit => Value::Blob(bytes.to_vec()),
_ => Value::Null,
}
}
pub(crate) fn empty_data_for(ty: &LogicalType) -> Result<Data> {
use rudb_common::PhysicalType as P;
macro_rules! empties {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match ty.physical() {
P::Empty => Data::Empty,
$(P::$variant => Data::$variant(Buffer::new()),)+
P::Varlen => Data::Varlen(StringColumn::new()),
other => {
return Err(Error::not_implemented(format!(
"a flat vector of {other:?} data, which arrives with the storage layer"
)));
}
}
};
}
Ok(crate::for_each_layout!(fixed, empties))
}
pub(crate) fn data_for(ty: &LogicalType, rows: usize) -> Result<Data> {
let mut data = empty_data_for(ty)?;
macro_rules! reserved {
($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
match &mut data {
Data::Empty => {}
$(Data::$variant(values) => values.reserve(rows),)+
Data::Varlen(values) => values.reserve_views(rows),
}
};
}
crate::for_each_layout!(fixed, reserved);
Ok(data)
}
fn stored<'v>(ty: &LogicalType, value: &'v Value) -> Result<Cow<'v, Value>> {
match (ty, value) {
(LogicalType::Enum(_), Value::Varchar(_)) => enum_position(ty, value).map(Cow::Owned),
_ => Ok(Cow::Borrowed(value)),
}
}
pub fn enum_position(ty: &LogicalType, value: &Value) -> Result<Value> {
let label = match (ty, value) {
(_, Value::Null) => return Ok(Value::Null),
(LogicalType::Enum(_), Value::Varchar(label)) => label,
_ => return Err(Error::internal(format!("{value:?} is not a value of {ty}"))),
};
let code = ty
.labels()
.and_then(|labels| labels.iter().position(|one| one == label))
.and_then(|code| u32::try_from(code).ok())
.ok_or_else(|| Error::internal(format!("{label:?} is not a value of {ty}")))?;
Ok(match enum_code_type(ty) {
LogicalType::UTinyInt => Value::UTinyInt(code as u8),
LogicalType::USmallInt => Value::USmallInt(code as u16),
_ => Value::UInteger(code),
})
}
#[must_use]
pub fn enum_code_type(ty: &LogicalType) -> LogicalType {
match ty.physical() {
rudb_common::PhysicalType::UInt8 => LogicalType::UTinyInt,
rudb_common::PhysicalType::UInt16 => LogicalType::USmallInt,
_ => LogicalType::UInteger,
}
}
fn push_value(data: &mut Data, value: &Value) -> Result<()> {
macro_rules! push {
($vec:expr, $variant:path, $zero:expr) => {
match value {
Value::Null => $vec.push($zero),
$variant(x) => $vec.push(*x),
other => {
return Err(Error::internal(format!(
"{other:?} does not belong in this vector"
)));
}
}
};
}
macro_rules! decimal {
($vec:expr, $ty:ty, $unscaled:expr) => {
match <$ty>::try_from(*$unscaled) {
Ok(x) => $vec.push(x),
Err(_) => {
return Err(Error::internal(format!(
"an unscaled decimal of {} does not fit the run its precision chose",
$unscaled
)));
}
}
};
}
match data {
Data::Empty => {}
Data::Bool(v) => push!(v, Value::Boolean, false),
Data::Int8(v) => push!(v, Value::TinyInt, 0),
Data::Int16(v) => match value {
Value::Null => v.push(0),
Value::SmallInt(x) => v.push(*x),
Value::Decimal { unscaled, .. } => decimal!(v, i16, unscaled),
other => return Err(Error::internal(format!("{other:?} is not a 16 bit value"))),
},
Data::Int32(v) => match value {
Value::Null => v.push(0),
Value::Integer(x) | Value::Date(x) => v.push(*x),
Value::Decimal { unscaled, .. } => decimal!(v, i32, unscaled),
other => return Err(Error::internal(format!("{other:?} is not a 32 bit value"))),
},
Data::Int64(v) => match value {
Value::Null => v.push(0),
Value::BigInt(x)
| Value::Time(x)
| Value::TimeTz(x)
| Value::Timestamp(x)
| Value::TimestampTz(x) => v.push(*x),
Value::Decimal { unscaled, .. } => decimal!(v, i64, unscaled),
other => return Err(Error::internal(format!("{other:?} is not a 64 bit value"))),
},
Data::Int128(v) => match value {
Value::Null => v.push(0),
Value::HugeInt(x) => v.push(*x),
Value::Decimal { unscaled, .. } => v.push(*unscaled),
other => return Err(Error::internal(format!("{other:?} is not a 128 bit value"))),
},
Data::UInt8(v) => push!(v, Value::UTinyInt, 0),
Data::UInt16(v) => push!(v, Value::USmallInt, 0),
Data::UInt32(v) => push!(v, Value::UInteger, 0),
Data::UInt64(v) => push!(v, Value::UBigInt, 0),
Data::UInt128(v) => push!(v, Value::UHugeInt, 0),
Data::Float32(v) => push!(v, Value::Float, 0.0),
Data::Float64(v) => push!(v, Value::Double, 0.0),
Data::Interval(v) => match value {
Value::Null => v.push((0, 0, 0)),
Value::Interval { months, days, micros } => v.push((*months, *days, *micros)),
other => return Err(Error::internal(format!("{other:?} is not an interval"))),
},
Data::Varlen(column) => match value {
Value::Null => {
column.push("");
}
Value::Varchar(text) => {
column.push(text);
}
Value::Blob(bytes) => {
column.push_bytes(bytes);
}
other => return Err(Error::internal(format!("{other:?} is not a string"))),
},
}
Ok(())
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use rudb_common::{Field, LogicalType, Value};
use super::{
Body, Data, FSST_PAYS_AT, Form, MAP_KEY, MAP_VALUE, NO_ROW, VECTOR_SIZE, Vector, below,
packing_base,
};
use crate::buffer::Buffer;
use crate::fsst::SymbolTable;
use crate::string::{StringColumn, StringView};
use crate::validity::Validity;
fn integers(values: &[i32]) -> Vector {
Vector::flat(LogicalType::Integer, Data::Int32(values.to_vec().into())).unwrap()
}
#[test]
fn below_agrees_with_the_largest_code_whether_the_or_settles_it_or_not() {
let cases: [(&[u32], usize); 8] = [
(&[], 0),
(&[], 5),
(&[0, 1, 8191], 8192),
(&[0, 8192], 8192),
(&[4, 1], 5),
(&[4, 5], 5),
(&[3, 4, 2], 5),
(&[7], 7),
];
for (codes, len) in cases {
let expected = codes.iter().all(|&code| (code as usize) < len);
assert_eq!(below(codes, len), expected, "{codes:?} below {len}");
}
}
#[test]
fn flattening_a_dictionary_by_its_codes_matches_the_general_copy() {
let words = Vector::from_values(
LogicalType::Varchar,
&["alpha", "a string past the inline length", ""]
.map(|text| Value::Varchar(text.into())),
)
.unwrap();
let codes = vec![2, 0, 1, 1, 0, 2, 1];
let cases = [
Vector::dictionary(codes.clone(), integers(&[7, -3, 40])).unwrap(),
Vector::dictionary(codes.clone(), words.clone()).unwrap(),
Vector::dictionary(codes.clone(), words.clone()).unwrap().slice(2, 4).unwrap(),
Vector::dictionary(codes.clone(), words.clone())
.unwrap()
.with_validity(Validity::from_run(&[true, false, true, true, true, true, false])),
Vector::dictionary(
vec![0, 1, 1],
integers(&[1, 2]).with_validity(Validity::from_run(&[true, false])),
)
.unwrap(),
];
for (case, vector) in cases.iter().enumerate() {
let flat = vector.flatten().unwrap();
let general = vector.copied((0..vector.len()).collect(), false).unwrap();
assert!(matches!(flat.body, Body::Flat(_)), "case {case}");
assert_eq!(flat.validity, general.validity, "case {case}");
for row in 0..vector.len() {
assert_eq!(flat.value_at(row), general.value_at(row), "case {case} row {row}");
}
assert_eq!(flat, vector.opened().unwrap(), "case {case}");
}
}
#[test]
fn extent_keeps_the_unsigned_order_across_the_sign_bit() {
assert_eq!(super::extent(&[]), None);
assert_eq!(super::extent(&[7]), Some((7, 7)));
let rows = [0x8000_0000, 3, u32::MAX, 0x7fff_ffff, 9];
assert_eq!(super::extent(&rows), Some((3, u32::MAX)));
}
#[test]
fn unpacking_in_bulk_reads_what_a_code_at_a_time_reads_at_every_width() {
let mut state = 0x5eed_0b17_u64;
let mut next = || {
state ^= state << 13;
state ^= state >> 7;
state ^= state << 17;
state
};
let words: Vec<u64> = (0..700).map(|_| next()).collect();
for width in 1..=super::PACKED_WIDTH_MAX {
for offset in [0, 1, 63, 64, 65] {
let packed = super::Packed { words: &words, width, base: 0, offset };
for (from, rows) in [(0, 0), (0, 1), (0, 64), (3, 200), (61, 130), (128, 512)] {
let mut out = vec![u64::MAX; rows];
packed.unpack(from, &mut out);
let want: Vec<u64> = (from..from + rows).map(|row| packed.code(row)).collect();
assert_eq!(out, want, "width {width} offset {offset} from {from}");
let mut mapped = vec![-1_i64];
packed.unpack_mapped(from, rows, &mut mapped, |code| 7 - code as i64);
let wanted: Vec<i64> = std::iter::once(-1)
.chain(want.iter().map(|&code| 7 - code as i64))
.collect();
assert_eq!(mapped, wanted, "mapped width {width} offset {offset} from {from}");
}
let at = [5_usize, 9, 9, 70, 6, 200, 131];
let want: Vec<u64> = at.iter().map(|&row| packed.code(row)).collect();
assert_eq!(packed.codes_at(|index| at[index], at.len()), want);
let far = [0_usize, 5000];
let want: Vec<u64> = far.iter().map(|&row| packed.code(row)).collect();
assert_eq!(packed.codes_at(|index| far[index], far.len()), want);
for start in [0_usize, 1, 63, 64, 65, 130] {
for rows in [1_usize, 2, 63, 64, 65, 200] {
let run: Vec<usize> = (start..start + rows).collect();
let back: Vec<usize> = run.iter().rev().copied().collect();
let mut same = run.clone();
same[rows - 1] = start;
for shape in [&run, &back, &same] {
let want: Vec<u64> =
shape.iter().map(|&row| packed.code(row)).collect();
assert_eq!(
packed.codes_at(|index| shape[index], shape.len()),
want,
"width {width} offset {offset} start {start} rows {rows}"
);
}
}
}
let mut held = vec![u64::MAX; 260];
for start in [0_usize, 1, 64, 130] {
for rows in [1_usize, 63, 64, 200] {
let run: Vec<usize> = (start..start + rows).collect();
let back: Vec<usize> = run.iter().rev().copied().collect();
for shape in [&run, &back] {
held.iter_mut().for_each(|code| *code = u64::MAX);
packed.codes_into(|index| shape[index], shape.len(), &mut held);
let want: Vec<u64> =
shape.iter().map(|&row| packed.code(row)).collect();
assert_eq!(
&held[..rows],
&want[..],
"width {width} offset {offset} start {start} rows {rows}"
);
assert!(
held[rows..].iter().all(|&code| code == u64::MAX),
"width {width} wrote past the {rows} rows it was asked for"
);
}
}
}
for rows in [&[][..], &[5, 9, 9, 70, 6, 200, 131], &[0, 5000], &[3, 4, 5, 6]] {
let want: Vec<u64> =
rows.iter().map(|&row| packed.code(row as usize)).collect();
assert_eq!(packed.values_at(rows, |code| code), want, "width {width}");
}
}
}
}
fn list(values: &[i32]) -> Value {
Value::List {
element: LogicalType::Integer,
values: values.iter().map(|&v| Value::Integer(v)).collect(),
}
}
fn list_column(rows: &[Value]) -> Vector {
Vector::from_values(LogicalType::list(LogicalType::Integer), rows).unwrap()
}
#[test]
fn a_list_column_is_one_child_and_a_range_per_row() {
let rows = vec![list(&[1, 2, 3]), list(&[]), Value::Null, list(&[4])];
let column = list_column(&rows);
assert_eq!(column.form(), Form::List);
assert_eq!(column.len(), 4);
assert_eq!(column.logical_type(), &LogicalType::list(LogicalType::Integer));
let (entries, child) = column.list_parts().expect("a list");
assert_eq!(entries, [(0, 3), (3, 0), (3, 0), (3, 1)]);
assert_eq!(child.len(), 4);
assert_eq!(column.iter().collect::<Vec<_>>(), rows);
}
#[test]
fn an_empty_list_and_a_null_list_have_the_same_entry_and_are_different_rows() {
let column = list_column(&[list(&[]), Value::Null]);
let (entries, _) = column.list_parts().expect("a list");
assert_eq!(entries[0].1, entries[1].1, "both entries are empty");
assert!(!column.is_null_at(0), "an empty list is not null");
assert!(column.is_null_at(1), "a null list is null");
assert_eq!(column.value_at(0), list(&[]));
assert_eq!(column.value_at(1), Value::Null);
}
#[test]
fn slicing_a_list_column_shares_the_child_rather_than_copying_it() {
let rows: Vec<Value> = (0..64).map(|row| list(&[row, row + 1, row + 2])).collect();
let column = list_column(&rows);
let cut = column.slice(8, 4).unwrap();
assert_eq!(cut.form(), Form::List);
assert_eq!(cut.iter().collect::<Vec<_>>(), rows[8..12]);
let (entries, child) = cut.list_parts().expect("a list");
assert_eq!(entries[0], (24, 3));
assert_eq!(child.len(), 192);
}
#[test]
fn gathering_a_list_column_permutes_the_entries_and_leaves_the_child_alone() {
let rows = vec![list(&[1]), list(&[2, 2]), list(&[3, 3, 3])];
let column = list_column(&rows);
let picked = column.gather(&[2, 0, 2]).unwrap();
assert_eq!(
picked.iter().collect::<Vec<_>>(),
[list(&[3, 3, 3]), list(&[1]), list(&[3, 3, 3])]
);
assert_eq!(picked.list_parts().expect("a list").1.len(), 6);
}
#[test]
fn a_gather_past_the_end_of_a_list_column_is_null_rather_than_somebody_elses_elements() {
let column = list_column(&[list(&[1, 2]), list(&[3])]);
let picked = column.gather(&[1, 9]).unwrap();
assert_eq!(picked.value_at(0), list(&[3]));
assert_eq!(picked.value_at(1), Value::Null);
}
#[test]
fn a_list_of_lists_nests_as_far_as_it_is_written() {
let outer = Value::List {
element: LogicalType::list(LogicalType::Integer),
values: vec![list(&[1, 2]), list(&[3])],
};
let column = Vector::from_values(
LogicalType::list(LogicalType::list(LogicalType::Integer)),
std::slice::from_ref(&outer),
)
.unwrap();
assert_eq!(column.value_at(0), outer);
assert_eq!(column.list_parts().expect("a list").1.form(), Form::List);
}
#[test]
fn the_scalar_readers_decline_a_list_instead_of_answering_about_its_elements() {
let column = list_column(&[list(&[7])]);
assert_eq!(column.signed_at(0), None);
assert_eq!(column.bytes_at(0), None);
assert_eq!(column.data(), None);
}
fn pair(a: i32, b: &str) -> Value {
Value::Struct(vec![
("a".to_string(), Value::Integer(a)),
("b".to_string(), Value::Varchar(b.to_string())),
])
}
fn pair_type() -> LogicalType {
LogicalType::Struct(vec![
Field::new("a", LogicalType::Integer),
Field::new("b", LogicalType::Varchar),
])
}
fn pair_column(rows: &[Value]) -> Vector {
Vector::from_values(pair_type(), rows).unwrap()
}
#[test]
fn a_struct_column_is_one_child_per_field_as_long_as_the_column() {
let rows = vec![pair(1, "x"), pair(2, "y"), pair(3, "z")];
let column = pair_column(&rows);
assert_eq!(column.form(), Form::Struct);
assert_eq!(column.len(), 3);
assert_eq!(column.logical_type(), &pair_type());
let children = column.struct_parts().expect("a struct");
assert_eq!(children.len(), 2);
assert_eq!(children[0].len(), 3);
assert_eq!(children[1].len(), 3);
assert_eq!(children[0].logical_type(), &LogicalType::Integer);
assert_eq!(children[1].logical_type(), &LogicalType::Varchar);
assert_eq!(column.iter().collect::<Vec<_>>(), rows);
}
#[test]
fn one_field_of_a_struct_column_is_a_column_that_is_already_there() {
let column = pair_column(&[pair(10, "x"), pair(20, "y")]);
let field = &column.struct_parts().expect("a struct")[0];
assert_eq!(field.iter().collect::<Vec<_>>(), [Value::Integer(10), Value::Integer(20)]);
assert_eq!(field.signed_at(1), Some(20), "the field is a scalar column and reads like one");
}
#[test]
fn a_null_struct_is_the_mask_at_the_top_and_not_a_struct_full_of_nulls() {
let column = pair_column(&[pair(1, "x"), Value::Null]);
assert!(!column.is_null_at(0));
assert!(column.is_null_at(1));
assert_eq!(column.value_at(1), Value::Null);
let all_null = pair_column(&[Value::Struct(vec![
("a".to_string(), Value::Null),
("b".to_string(), Value::Null),
])]);
assert!(!all_null.is_null_at(0), "a struct of nulls is a row that is there");
assert_ne!(all_null.value_at(0), Value::Null);
}
#[test]
fn slicing_a_struct_column_cuts_every_field_at_the_same_place() {
let rows: Vec<Value> = (0..64).map(|row| pair(row, "s")).collect();
let column = pair_column(&rows);
let cut = column.slice(8, 4).unwrap();
assert_eq!(cut.form(), Form::Struct);
assert_eq!(cut.iter().collect::<Vec<_>>(), rows[8..12]);
for child in cut.struct_parts().expect("a struct") {
assert_eq!(child.len(), 4);
}
}
#[test]
fn gathering_a_struct_column_gathers_every_field_at_the_same_positions() {
let column = pair_column(&[pair(1, "x"), pair(2, "y"), pair(3, "z")]);
let picked = column.gather(&[2, 0, 2]).unwrap();
assert_eq!(picked.iter().collect::<Vec<_>>(), [pair(3, "z"), pair(1, "x"), pair(3, "z")]);
for child in picked.struct_parts().expect("a struct") {
assert_eq!(child.len(), 3, "a field is as long as the gather, not as the source");
}
}
#[test]
fn a_gather_past_the_end_of_a_struct_column_is_null_in_every_field_and_at_the_top() {
let column = pair_column(&[pair(1, "x"), pair(2, "y")]);
let picked = column.gather(&[1, 9]).unwrap();
assert_eq!(picked.value_at(0), pair(2, "y"));
assert_eq!(picked.value_at(1), Value::Null);
for child in picked.struct_parts().expect("a struct") {
assert!(child.is_null_at(1), "a row that came from nowhere has no field value either");
}
}
#[test]
fn the_fields_of_a_struct_value_go_in_by_name_rather_than_by_position() {
let swapped = Value::Struct(vec![
("b".to_string(), Value::Varchar("x".to_string())),
("a".to_string(), Value::Integer(1)),
]);
let column = pair_column(&[swapped]);
assert_eq!(column.value_at(0), pair(1, "x"));
let wrong = Value::Struct(vec![
("a".to_string(), Value::Integer(1)),
("c".to_string(), Value::Varchar("x".to_string())),
]);
let failed = Vector::from_values(pair_type(), &[wrong]);
assert!(failed.is_err(), "a row with no b field is an error rather than a null b");
}
#[test]
fn a_struct_built_from_children_takes_its_field_names_from_the_caller() {
let column = Vector::structure(vec![
("a".to_string(), integers(&[1, 2, 3])),
("b".to_string(), integers(&[4, 5, 6])),
])
.expect("two columns of three");
assert_eq!(column.len(), 3);
assert_eq!(
column.logical_type(),
&LogicalType::Struct(vec![
Field::new("a", LogicalType::Integer),
Field::new("b", LogicalType::Integer),
])
);
assert_eq!(
column.value_at(1),
Value::Struct(vec![
("a".to_string(), Value::Integer(2)),
("b".to_string(), Value::Integer(5)),
])
);
}
#[test]
fn a_struct_of_uneven_children_or_of_no_children_is_refused() {
let uneven = Vector::structure(vec![
("a".to_string(), integers(&[1, 2, 3])),
("b".to_string(), integers(&[4, 5])),
]);
assert!(uneven.is_err(), "a field shorter than the struct");
assert!(Vector::structure(vec![]).is_err(), "no field to take a length from");
}
#[test]
fn a_struct_of_lists_and_a_list_of_structs_both_nest() {
let ty =
LogicalType::Struct(vec![Field::new("a", LogicalType::list(LogicalType::Integer))]);
let row = Value::Struct(vec![("a".to_string(), list(&[1, 2]))]);
let column = Vector::from_values(ty, std::slice::from_ref(&row)).unwrap();
assert_eq!(column.value_at(0), row);
assert_eq!(column.struct_parts().expect("a struct")[0].form(), Form::List);
let outer = Value::List { element: pair_type(), values: vec![pair(1, "x"), pair(2, "y")] };
let lists =
Vector::from_values(LogicalType::list(pair_type()), std::slice::from_ref(&outer))
.unwrap();
assert_eq!(lists.value_at(0), outer);
assert_eq!(lists.list_parts().expect("a list").1.form(), Form::Struct);
}
fn tags(pairs: &[(&str, &str)]) -> Value {
Value::map(
LogicalType::Varchar,
LogicalType::Varchar,
pairs
.iter()
.map(|&(key, value)| {
(Value::Varchar(key.to_string()), Value::Varchar(value.to_string()))
})
.collect(),
)
}
fn tag_column(rows: &[Value]) -> Vector {
Vector::from_values(LogicalType::map(LogicalType::Varchar, LogicalType::Varchar), rows)
.unwrap()
}
#[test]
fn a_map_column_is_a_list_whose_child_is_a_struct_of_keys_and_values() {
let rows =
vec![tags(&[("a", "b"), ("c", "d")]), tags(&[]), Value::Null, tags(&[("e", "f")])];
let column = tag_column(&rows);
assert_eq!(column.len(), 4);
assert_eq!(
column.logical_type(),
&LogicalType::map(LogicalType::Varchar, LogicalType::Varchar)
);
assert_eq!(column.form(), Form::List);
let (entries, child) = column.list_parts().expect("the layout of a list");
assert_eq!(entries, [(0, 2), (2, 0), (2, 0), (2, 1)]);
assert_eq!(child.form(), Form::Struct);
assert_eq!(
child.logical_type(),
&LogicalType::Struct(vec![
Field::new(MAP_KEY, LogicalType::Varchar),
Field::new(MAP_VALUE, LogicalType::Varchar),
])
);
let (entries, keys, values) = column.map_parts().expect("a map");
assert_eq!(entries.len(), 4);
assert_eq!(keys.text_at(0), Some("a"));
assert_eq!(values.text_at(0), Some("b"));
assert_eq!(column.iter().collect::<Vec<_>>(), rows);
}
#[test]
fn an_empty_map_and_a_null_map_are_different_rows() {
let column = tag_column(&[tags(&[]), Value::Null]);
assert!(!column.is_null_at(0), "an empty map is a row that is there");
assert!(column.is_null_at(1));
assert_eq!(column.value_at(0), tags(&[]));
assert_eq!(column.value_at(1), Value::Null);
assert_eq!(column.value_at(0).to_string(), "{}");
assert_eq!(column.value_at(1).to_string(), "NULL");
}
#[test]
fn a_map_prints_with_equals_signs_and_a_struct_prints_with_quoted_names() {
assert_eq!(tags(&[("a", "b"), ("c", "d")]).to_string(), "{a=b, c=d}");
assert_eq!(pair(1, "x").to_string(), "{'a': 1, 'b': x}");
let numbers = Value::map(
LogicalType::Integer,
LogicalType::Integer,
vec![(Value::Integer(1), Value::Integer(3)), (Value::Integer(2), Value::Integer(4))],
);
assert_eq!(numbers.to_string(), "{1=3, 2=4}");
let null_value = Value::map(
LogicalType::Varchar,
LogicalType::Varchar,
vec![(Value::Varchar("x".to_string()), Value::Null)],
);
assert_eq!(null_value.to_string(), "{x=NULL}");
}
#[test]
fn cutting_and_gathering_a_map_keeps_it_a_map() {
let rows: Vec<Value> =
(0..16).map(|row| tags(&[("k", if row % 2 == 0 { "e" } else { "o" })])).collect();
let column = tag_column(&rows);
let cut = column.slice(4, 3).unwrap();
assert!(matches!(cut.logical_type(), LogicalType::Map(_, _)), "still a map after a cut");
assert_eq!(cut.iter().collect::<Vec<_>>(), rows[4..7]);
assert_eq!(cut.map_parts().expect("a map").1.len(), 16);
let picked = column.gather(&[3, 0, 3]).unwrap();
assert!(matches!(picked.logical_type(), LogicalType::Map(_, _)));
assert_eq!(
picked.iter().collect::<Vec<_>>(),
[rows[3].clone(), rows[0].clone(), rows[3].clone()]
);
let past = column.gather(&[0, 99]).unwrap();
assert_eq!(past.value_at(1), Value::Null);
}
#[test]
fn a_map_built_from_two_columns_pairs_them_by_position() {
let keys = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("a".to_string()), Value::Varchar("c".to_string())],
)
.unwrap();
let values = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("b".to_string()), Value::Varchar("d".to_string())],
)
.unwrap();
let column = Vector::map(vec![(0, 2), (2, 0)], keys, values).expect("two rows");
assert_eq!(column.len(), 2);
assert_eq!(
column.logical_type(),
&LogicalType::map(LogicalType::Varchar, LogicalType::Varchar)
);
assert_eq!(column.value_at(0), tags(&[("a", "b"), ("c", "d")]));
assert_eq!(column.value_at(1), tags(&[]));
let short =
Vector::from_values(LogicalType::Varchar, &[Value::Varchar("a".to_string())]).unwrap();
let other =
Vector::from_values(LogicalType::Varchar, &[Value::Varchar("b".to_string())]).unwrap();
assert!(Vector::map(vec![(0, 9)], short, other).is_err(), "an entry past the end");
}
#[test]
fn a_list_is_not_a_map_however_much_its_child_looks_like_one() {
let pairs = Value::List { element: pair_type(), values: vec![pair(1, "x")] };
let column =
Vector::from_values(LogicalType::list(pair_type()), std::slice::from_ref(&pairs))
.unwrap();
assert!(column.map_parts().is_none(), "a list of structs is a list");
assert!(column.list_parts().is_some());
let map = tag_column(&[tags(&[("a", "b")])]);
assert!(map.map_parts().is_some());
assert!(map.list_parts().is_some(), "a map has a list's layout and says so");
}
#[test]
fn the_scalar_readers_decline_a_struct_of_one_integer_field() {
let ty = LogicalType::Struct(vec![Field::new("a", LogicalType::Integer)]);
let row = Value::Struct(vec![("a".to_string(), Value::Integer(7))]);
let column = Vector::from_values(ty, &[row]).unwrap();
assert_eq!(column.signed_at(0), None);
assert_eq!(column.bytes_at(0), None);
assert_eq!(column.data(), None);
}
#[test]
fn a_clustered_column_becomes_runs_and_reads_back_the_same() {
let mut values = Vec::new();
for (value, times) in [(7, 400), (8, 300), (7, 324)] {
values.extend(std::iter::repeat_n(value, times));
}
let flat = integers(&values);
let runs = flat.run_encoded().unwrap();
assert_eq!(runs.form(), Form::Rle);
assert_eq!(runs.run_parts().expect("runs").0, [400, 700, 1024]);
assert_eq!(runs.len(), flat.len());
assert_eq!(runs.iter().collect::<Vec<_>>(), flat.iter().collect::<Vec<_>>());
assert!(
runs.footprint() * 10 < flat.footprint(),
"three runs against a thousand rows: {} against {}",
runs.footprint(),
flat.footprint()
);
}
#[test]
fn a_column_that_does_not_repeat_is_left_flat() {
let flat = integers(&(0..1024).collect::<Vec<i32>>());
assert_eq!(flat.run_encoded().unwrap().form(), Form::Flat);
assert_eq!(integers(&[1, 1, 2, 2]).run_encoded().unwrap().form(), Form::Flat);
assert_eq!(integers(&[1, 1, 1, 2, 2]).run_encoded().unwrap().form(), Form::Rle);
}
#[test]
fn two_nulls_beside_each_other_are_one_run_and_a_null_between_two_equals_is_a_break() {
let mut values = vec![Value::Integer(4), Value::Integer(4)];
values.extend([Value::Null, Value::Null, Value::Null]);
values.extend(std::iter::repeat_n(Value::Integer(4), 5));
let flat = Vector::from_values(LogicalType::Integer, &values).unwrap();
let runs = flat.run_encoded().unwrap();
assert_eq!(runs.run_parts().expect("runs").0, [2, 5, 10]);
assert_eq!(runs.iter().collect::<Vec<_>>(), values);
}
#[test]
fn slicing_runs_keeps_them_runs_and_cuts_the_first_and_last_one_back() {
let flat = integers(&[1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3]);
let runs = flat.run_encoded().unwrap();
let piece = runs.slice(3, 6).unwrap();
assert_eq!(piece.form(), Form::Rle, "the form is the whole point");
assert_eq!(piece.run_parts().expect("runs").0, [1, 5, 6]);
assert_eq!(
piece.iter().collect::<Vec<_>>(),
flat.slice(3, 6).unwrap().iter().collect::<Vec<_>>()
);
assert_eq!(runs.slice(0, 0).unwrap().len(), 0);
assert_eq!(runs.slice(0, 12).unwrap().form(), Form::Rle);
}
#[test]
fn gathering_out_of_runs_walks_to_the_values_the_way_it_walks_a_dictionary() {
let mut values = vec![Value::Varchar("red".into()); 4];
values.extend([Value::Null, Value::Null, Value::Null]);
values.extend(vec![Value::Varchar("blue".into()); 4]);
let runs =
Vector::from_values(LogicalType::Varchar, &values).unwrap().run_encoded().unwrap();
assert_eq!(runs.form(), Form::Rle);
let picked = runs.gather(&[8, 0, 5, 2]).unwrap();
assert_eq!(picked.form(), Form::Flat, "a gather copies, whatever it gathered from");
assert_eq!(
picked.iter().collect::<Vec<_>>(),
[values[8].clone(), values[0].clone(), Value::Null, values[2].clone()]
);
assert_eq!(runs.text_at(1), Some("red"));
assert_eq!(runs.text_at(5), None, "a null has no text");
assert_eq!(runs.flatten().unwrap().iter().collect::<Vec<_>>(), values);
}
#[test]
fn runs_of_runs_are_refused_and_runs_of_a_dictionary_are_not() {
let inner = integers(&[1, 1, 1, 1, 2]).run_encoded().unwrap();
assert_eq!(inner.form(), Form::Rle);
let error = Vector::runs(vec![2, 8], inner).unwrap_err();
assert!(error.to_string().contains("runs of runs"), "{error}");
let words = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
let dictionary = Vector::dictionary(vec![1, 0], words).unwrap();
let stacked = Vector::runs(vec![4, 9], dictionary).unwrap();
assert_eq!(stacked.len(), 9);
assert_eq!(stacked.value_at(3), Value::Varchar("blue".into()));
assert_eq!(stacked.value_at(4), Value::Varchar("red".into()));
}
#[test]
fn run_ends_have_to_increase_and_there_is_one_value_for_each_of_them() {
let values = integers(&[1, 2]);
assert!(Vector::runs(vec![4], values.clone()).is_err(), "two values and one run");
assert!(Vector::runs(vec![4, 4], values.clone()).is_err(), "an end that repeats");
assert!(Vector::runs(vec![4, 2], values.clone()).is_err(), "an end that goes backwards");
assert!(Vector::runs(vec![0, 2], values.clone()).is_err(), "a first run holding no rows");
assert_eq!(Vector::runs(vec![4, 9], values).unwrap().len(), 9);
}
#[test]
fn a_form_that_is_already_compact_is_left_where_it_is() {
let constant = Vector::constant(LogicalType::Integer, Value::Integer(1), 1000);
assert_eq!(constant.run_encoded().unwrap().form(), Form::Constant);
assert_eq!(Vector::sequence(0, 1, 1000).run_encoded().unwrap().form(), Form::Sequence);
}
#[test]
fn both_forms_that_point_somewhere_hand_back_a_position_per_row() {
let words = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
let runs = Vector::runs(vec![3, 5], words.clone()).unwrap();
let (at, values) = runs.positions().expect("runs point somewhere");
assert_eq!(at.as_ref(), [0, 0, 0, 1, 1]);
assert_eq!(values.value_at(at[3] as usize), runs.value_at(3));
let dictionary = Vector::dictionary(vec![1, 0, 1], words).unwrap();
let (at, values) = dictionary.positions().expect("a dictionary points somewhere");
assert_eq!(at.as_ref(), [1, 0, 1]);
assert_eq!(values.value_at(at[0] as usize), dictionary.value_at(0));
assert!(integers(&[1, 2, 3]).positions().is_none(), "a flat vector points at itself");
assert!(Vector::sequence(0, 1, 4).positions().is_none(), "a sequence stores nothing");
}
#[test]
fn slicing_a_dictionary_keeps_it_a_dictionary_where_gathering_would_not() {
let values = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
let vector = Vector::dictionary(vec![0, 1, 1, 0, 1], values).unwrap();
let piece = vector.slice(1, 3).unwrap();
assert_eq!(piece.form(), Form::Dictionary, "the form is the whole point");
assert_eq!(piece.len(), 3);
assert_eq!(
piece.iter().collect::<Vec<_>>(),
[
Value::Varchar("blue".into()),
Value::Varchar("blue".into()),
Value::Varchar("red".into())
]
);
assert_eq!(vector.gather(&[1, 2, 3]).unwrap().form(), Form::Flat, "which a gather loses");
}
#[test]
fn slicing_a_dictionary_shares_the_dictionary_rather_than_copying_it() {
let values = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
let vector = Vector::dictionary(vec![0, 1, 1, 0, 1], values).unwrap();
let Body::Dictionary { values: whole, .. } = &vector.body else {
panic!("a dictionary vector holds a dictionary");
};
let piece = vector.slice(1, 3).unwrap();
let Body::Dictionary { codes, values: cut, .. } = &piece.body else {
panic!("a slice of a dictionary is a dictionary");
};
assert!(Arc::ptr_eq(whole, cut), "the cut copied the dictionary");
assert_eq!(codes.as_slice(), &[1, 1, 0], "the codes are the part that is cut");
let again = piece.slice(1, 2).unwrap();
let Body::Dictionary { values: cut, .. } = &again.body else {
panic!("a slice of a slice of a dictionary is a dictionary");
};
assert!(Arc::ptr_eq(whole, cut), "the second cut copied the dictionary");
assert_eq!(
again.iter().collect::<Vec<_>>(),
[Value::Varchar("blue".into()), Value::Varchar("red".into())]
);
}
#[test]
fn flattening_a_gather_off_a_paged_parent_takes_the_arena_rather_than_copying_it() {
let arena = Arc::new(Buffer::from_vec(b"1-URGENT2-HIGH".to_vec()));
let views = vec![
StringView::over(b"1-URGENT", 0),
StringView::over(b"2-HIGH", 8),
StringView::over(b"1-URGENT", 0),
];
let built = Vector::string_views(LogicalType::Varchar, views, arena).unwrap();
let owned = match &built.body {
Body::Views { arena, .. } => arena.is_shared(),
_ => panic!("string views are a views body"),
};
assert!(!owned, "concat builds an arena rather than reading one, so it starts owned");
let bytes = |vector: &Vector| match &vector.body {
Body::Views { arena, .. } => arena.as_slice().as_ptr() as usize,
Body::Flat(Data::Varlen(column)) => column.arena().as_ptr() as usize,
_ => panic!("a string vector holds string bytes"),
};
let gathered = |parent: &Vector| {
Vector::gathered(Arc::new(parent.clone()), Arc::new(vec![1, 0])).unwrap()
};
let paged = Vector::string_views(
LogicalType::Varchar,
built.shared_views().unwrap().0.to_vec(),
Arc::new(Buffer::from_vec(b"1-URGENT2-HIGH".to_vec())),
)
.unwrap()
.into_pages();
assert_eq!(
bytes(&gathered(&paged).flatten().unwrap()),
bytes(&paged),
"a flatten off a page shares the arena"
);
assert_ne!(
bytes(&gathered(&built).flatten().unwrap()),
bytes(&built),
"and off an owned arena it copies, which is what this changed"
);
assert_eq!(
gathered(&paged).flatten().unwrap().iter().collect::<Vec<_>>(),
[Value::Varchar("2-HIGH".into()), Value::Varchar("1-URGENT".into())]
);
}
#[test]
fn paging_a_string_column_whose_arena_has_another_holder_leaves_it_alone() {
let arena = Arc::new(Buffer::from_vec(b"red".to_vec()));
let vector =
Vector::string_views(LogicalType::Varchar, vec![StringView::over(b"red", 0)], arena)
.unwrap();
let paged = vector.clone().into_pages();
match &paged.body {
Body::Views { arena, .. } => assert!(!arena.is_shared(), "it was not ours to move"),
_ => panic!("string views are a views body"),
}
assert_eq!(paged.iter().collect::<Vec<_>>(), [Value::Varchar("red".into())]);
}
#[test]
fn a_paged_dictionary_shares_its_codes_with_its_cuts_and_clones() {
let values = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
let vector = Vector::dictionary(vec![0, 1, 1, 0, 1], values).unwrap().into_pages();
let codes = |vector: &Vector| match &vector.body {
Body::Dictionary { codes, .. } => codes.as_slice().as_ptr() as usize,
_ => panic!("a dictionary vector holds a dictionary"),
};
assert_eq!(codes(&vector.slice(1, 3).unwrap()), codes(&vector) + 4, "the cut copied");
assert_eq!(codes(&vector.clone()), codes(&vector), "the clone copied");
assert_eq!(
vector.slice(1, 3).unwrap().iter().collect::<Vec<_>>(),
[
Value::Varchar("blue".into()),
Value::Varchar("blue".into()),
Value::Varchar("red".into())
]
);
}
#[test]
fn a_slice_carries_the_nulls_that_were_in_its_range_and_not_the_others() {
let vector =
integers(&[1, 2, 3, 4]).with_validity(Validity::from_run(&[false, true, false, true]));
let piece = vector.slice(1, 2).unwrap();
assert!(piece.validity().is_valid(0));
assert!(!piece.validity().is_valid(1));
assert_eq!(piece.value_at(1), Value::Null);
}
#[test]
fn slicing_a_sequence_moves_its_start_rather_than_writing_the_values_out() {
let vector = Vector::sequence(100, 5, 10);
let piece = vector.slice(3, 4).unwrap();
assert_eq!(piece.form(), Form::Sequence);
assert_eq!(
piece.iter().collect::<Vec<_>>(),
[Value::BigInt(115), Value::BigInt(120), Value::BigInt(125), Value::BigInt(130)]
);
}
#[test]
fn slicing_a_constant_is_a_shorter_constant() {
let vector = Vector::constant(LogicalType::Integer, Value::Integer(9), 8);
let piece = vector.slice(2, 3).unwrap();
assert_eq!(piece.form(), Form::Constant);
assert_eq!(piece.len(), 3);
assert_eq!(piece.value_at(2), Value::Integer(9));
}
#[test]
fn slicing_the_whole_vector_hands_it_back_as_it_was() {
let vector = integers(&[1, 2, 3]);
assert_eq!(
vector.slice(0, 3).unwrap().iter().collect::<Vec<_>>(),
[Value::Integer(1), Value::Integer(2), Value::Integer(3)]
);
}
#[test]
fn a_gather_off_a_flat_run_with_no_nulls_answers_what_the_general_copy_does() {
let rows: Vec<i32> = (0..50).map(|row| row * 3 - 20).collect();
let vector = integers(&rows);
let positions: Vec<u32> = [49, 0, 7, 7, 31, 2].into_iter().collect();
let gathered = vector.gather(&positions).unwrap();
assert_eq!(gathered.form(), Form::Flat);
assert_eq!(
gathered.iter().collect::<Vec<_>>(),
positions.iter().map(|&at| Value::Integer(rows[at as usize])).collect::<Vec<_>>()
);
let past = vector.gather(&[3, 50]).unwrap();
assert_eq!(past.iter().collect::<Vec<_>>(), [Value::Integer(-11), Value::Null]);
}
#[test]
fn cutting_a_flat_body_answers_what_gathering_the_same_rows_answers() {
let rows: Vec<i32> = (0..70).collect();
let valid: Vec<bool> = (0..70).map(|row| row % 7 != 0 && row % 11 != 3).collect();
let vector = integers(&rows).with_validity(Validity::from_run(&valid));
for at in 0..70usize {
for len in 0..=(70 - at) {
let cut = vector.slice(at, len).unwrap();
let positions: Vec<u32> = (at..at + len).map(|row| row as u32).collect();
let gathered = vector.gather(&positions).unwrap();
assert_eq!(cut.len(), len, "rows {at} to {}", at + len);
assert_eq!(
cut.iter().collect::<Vec<_>>(),
gathered.iter().collect::<Vec<_>>(),
"rows {at} to {}",
at + len
);
}
}
}
#[test]
fn cutting_a_flat_body_over_a_page_does_not_copy_it() {
let page = Arc::new((0i64..64).collect::<Vec<_>>());
let address = page.as_ptr() as usize;
let data = Data::Int64(Buffer::from_arc(Arc::clone(&page)));
let vector = Vector::flat(LogicalType::BigInt, data).unwrap();
let cut = vector.slice(16, 8).unwrap();
assert_eq!(cut.form(), Form::Flat);
assert_eq!(cut.len(), 8);
let Some(Data::Int64(run)) = cut.data() else {
panic!("the layout changed under the test")
};
assert!(run.is_shared(), "the cut copied the run out of the page");
assert_eq!(run.as_slice().as_ptr() as usize, address + 16 * 8);
assert_eq!(run.as_slice(), &(16i64..24).collect::<Vec<_>>()[..]);
assert_eq!(cut.value_at(0), Value::BigInt(16));
let owned = Vector::flat(LogicalType::BigInt, Data::Int64((0i64..64).collect())).unwrap();
let copied = owned.slice(16, 8).unwrap();
let Some(Data::Int64(run)) = copied.data() else {
panic!("the layout changed under the test")
};
assert!(!run.is_shared());
assert_eq!(run.as_slice(), &(16i64..24).collect::<Vec<_>>()[..]);
}
#[test]
fn a_vector_over_pages_is_copied_and_cut_without_its_values_moving() {
let vector = integers(&[1, 2, 3, 4, 5, 6, 7, 8]).into_pages();
let address = |vector: &Vector| match vector.data() {
Some(Data::Int32(values)) => values.as_slice().as_ptr() as usize,
_ => panic!("the layout changed under the test"),
};
let stored = address(&vector);
assert_eq!(address(&vector.clone()), stored, "a copy moved the values");
assert_eq!(address(&vector.slice(2, 4).unwrap()), stored + 2 * 4, "a cut moved the values");
assert_eq!(
vector.slice(2, 4).unwrap().iter().collect::<Vec<_>>(),
[Value::Integer(3), Value::Integer(4), Value::Integer(5), Value::Integer(6)]
);
assert_eq!(address(&vector.clone().into_pages()), stored);
}
#[test]
fn a_string_column_over_a_page_is_cut_and_gathered_without_its_payload_moving() {
let long = ["the first of the long strings", "the second one", "and a third long one here"];
let mut built = StringColumn::with_capacity(long.len());
for text in long {
built.push(text);
}
let vector = Vector::flat(LogicalType::Varchar, Data::Varlen(built.into_page())).unwrap();
let payload = |vector: &Vector| match vector.data() {
Some(Data::Varlen(column)) => column.arena().as_ptr() as usize,
_ => panic!("the layout changed under the test"),
};
let stored = payload(&vector);
let cut = vector.slice(1, 2).unwrap();
assert_eq!(payload(&cut), stored, "a cut moved the payload");
assert_eq!(cut.text_at(0), Some(long[1]));
assert_eq!(cut.text_at(1), Some(long[2]));
let gathered = vector.gather(&[2, 0]).unwrap();
assert_eq!(payload(&gathered), stored, "a gather moved the payload");
assert_eq!(gathered.text_at(0), Some(long[2]));
assert_eq!(gathered.text_at(1), Some(long[0]));
let mut owned = StringColumn::with_capacity(long.len());
for text in long {
owned.push(text);
}
let held = Vector::flat(LogicalType::Varchar, Data::Varlen(owned)).unwrap();
let copied = held.slice(1, 2).unwrap();
assert_ne!(payload(&copied), payload(&held), "an owned payload was shared");
assert_eq!(copied.text_at(0), Some(long[1]));
}
#[test]
fn flattening_string_views_over_a_page_keeps_the_page() {
let mut built = StringColumn::with_capacity(2);
built.push("a string too long to sit inside a view");
built.push("another string that is also too long");
let (views, arena) = built.into_page().into_parts();
let stored = arena.as_slice().as_ptr() as usize;
let vector = Vector::string_views(LogicalType::Varchar, views, Arc::new(arena)).unwrap();
assert_eq!(vector.form(), Form::StringView);
let flat = vector.flatten().unwrap();
assert_eq!(flat.form(), Form::Flat);
let Some(Data::Varlen(column)) = flat.data() else {
panic!("the layout changed under the test")
};
assert_eq!(column.arena().as_ptr() as usize, stored, "the flatten moved the payload");
assert_eq!(flat.text_at(0), Some("a string too long to sit inside a view"));
assert_eq!(flat.text_at(1), Some("another string that is also too long"));
}
#[test]
fn putting_a_vector_on_pages_does_not_change_any_other_form() {
let dictionary = Vector::dictionary(
vec![0, 1, 0, 1],
Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("a".into()), Value::Varchar("b".into())],
)
.unwrap(),
)
.unwrap();
let cases = [
Vector::constant(LogicalType::Integer, Value::Integer(9), 4),
Vector::sequence(4, 0, 1),
dictionary,
];
for vector in cases {
let form = vector.form();
let paged = vector.clone().into_pages();
assert_eq!(paged.form(), form, "{form:?} changed form");
assert_eq!(paged.iter().collect::<Vec<_>>(), vector.iter().collect::<Vec<_>>());
}
}
#[test]
fn cutting_a_flat_string_column_answers_what_gathering_it_answers() {
let rows: Vec<String> =
(0..40).map(|row| "x".repeat(row % 30) + &row.to_string()).collect();
let values: Vec<Value> = rows.iter().map(|row| Value::Varchar(row.clone())).collect();
let vector = Vector::from_values(LogicalType::Varchar, &values).unwrap().flatten().unwrap();
assert_eq!(vector.form(), Form::Flat, "the cut under test is the flat one");
for at in 0..40usize {
for len in 0..=(40 - at) {
let cut = vector.slice(at, len).unwrap();
let positions: Vec<u32> = (at..at + len).map(|row| row as u32).collect();
let gathered = vector.gather(&positions).unwrap();
assert_eq!(
cut.iter().collect::<Vec<_>>(),
gathered.iter().collect::<Vec<_>>(),
"rows {at} to {}",
at + len
);
}
}
}
#[test]
fn a_slice_past_the_end_is_an_error_rather_than_a_short_vector() {
let error = integers(&[1, 2, 3]).slice(2, 2).unwrap_err();
assert!(error.to_string().contains("of a vector of 3"), "{error}");
}
#[test]
fn the_vector_size_is_the_one_the_design_is_built_around() {
assert_eq!(VECTOR_SIZE, 8192);
assert_eq!(VECTOR_SIZE % 1024, 0);
assert_eq!(VECTOR_SIZE % 64, 0);
assert_eq!(VECTOR_SIZE / 64, 128, "the words in a validity mask");
}
#[test]
fn a_flat_vector_reads_back_what_was_put_in_it() {
let vector = integers(&[1, 2, 3]);
assert_eq!(vector.form(), Form::Flat);
assert_eq!(vector.len(), 3);
assert_eq!(vector.value_at(1), Value::Integer(2));
assert_eq!(
vector.iter().collect::<Vec<_>>(),
vec![Value::Integer(1), Value::Integer(2), Value::Integer(3)]
);
}
#[test]
fn a_vector_built_from_values_reads_the_same_values_back() {
let vector = Vector::from_values(
LogicalType::Varchar,
&[
Value::Varchar("a".to_string()),
Value::Null,
Value::Varchar("a string too long to sit inside a view".to_string()),
],
)
.expect("strings and a null");
assert_eq!(vector.len(), 3);
assert_eq!(vector.value_at(0), Value::Varchar("a".to_string()));
assert_eq!(vector.value_at(1), Value::Null);
assert_eq!(
vector.value_at(2),
Value::Varchar("a string too long to sit inside a view".to_string())
);
}
#[test]
fn a_null_in_the_middle_does_not_move_the_values_after_it() {
let vector = Vector::from_values(
LogicalType::Integer,
&[Value::Integer(1), Value::Null, Value::Integer(3)],
)
.expect("integers and a null");
assert_eq!(vector.value_at(2), Value::Integer(3));
assert!(vector.validity().has_nulls(3), "the middle one is null");
}
#[test]
fn a_value_the_type_cannot_hold_is_refused() {
let wrong = Vector::from_values(LogicalType::Integer, &[Value::Varchar("x".to_string())]);
assert!(wrong.is_err(), "a string is not an integer");
}
#[test]
fn a_type_that_does_not_match_its_layout_is_refused_at_construction() {
let wrong = Vector::flat(LogicalType::Varchar, Data::Int32(vec![1].into()));
assert!(wrong.is_err());
let right = Vector::flat(LogicalType::Date, Data::Int32(vec![1].into()));
assert!(right.is_ok(), "a date is stored in an i32 and that has to be allowed");
}
#[test]
fn a_constant_vector_costs_one_value_whatever_its_length() {
let vector = Vector::constant(LogicalType::Integer, Value::Integer(7), VECTOR_SIZE);
assert_eq!(vector.form(), Form::Constant);
assert_eq!(vector.len(), VECTOR_SIZE);
assert_eq!(vector.value_at(0), Value::Integer(7));
assert_eq!(vector.value_at(VECTOR_SIZE - 1), Value::Integer(7));
assert_eq!(vector.value_at(VECTOR_SIZE), Value::Null, "past the end is null, not a panic");
}
#[test]
fn a_constant_null_is_all_invalid_without_being_told() {
let vector = Vector::constant(LogicalType::Integer, Value::Null, 8);
assert_eq!(vector.validity(), &Validity::AllInvalid);
assert_eq!(vector.value_at(3), Value::Null);
}
#[test]
fn a_sequence_vector_is_sixteen_bytes_of_row_identifiers() {
let vector = Vector::sequence(100, 1, VECTOR_SIZE);
assert_eq!(vector.form(), Form::Sequence);
assert_eq!(vector.value_at(0), Value::BigInt(100));
assert_eq!(vector.value_at(923), Value::BigInt(1023));
let stepped = Vector::sequence(0, 5, 4);
assert_eq!(
stepped.iter().collect::<Vec<_>>(),
vec![Value::BigInt(0), Value::BigInt(5), Value::BigInt(10), Value::BigInt(15)]
);
}
#[test]
fn a_dictionary_vector_reads_through_its_codes() {
let mut column = StringColumn::new();
column.push("red");
column.push("green");
let values = Vector::flat(LogicalType::Varchar, Data::Varlen(column)).unwrap();
let vector = Vector::dictionary(vec![0, 1, 1, 0], values).unwrap();
assert_eq!(vector.form(), Form::Dictionary);
assert_eq!(vector.logical_type(), &LogicalType::Varchar);
assert_eq!(vector.value_at(2), Value::Varchar("green".into()));
assert_eq!(vector.len(), 4);
}
#[test]
fn text_is_read_where_it_already_is_for_the_forms_that_store_it() {
let mut column = StringColumn::new();
column.push("red");
column.push("green");
column.push("");
let flat = Vector::flat(LogicalType::Varchar, Data::Varlen(column)).unwrap();
for index in 0..flat.len() {
assert_eq!(flat.text_at(index).map(str::to_string), text_of(&flat.value_at(index)));
}
let dictionary = Vector::dictionary(vec![1, 0, 1, 2], flat).unwrap();
for index in 0..dictionary.len() {
assert_eq!(
dictionary.text_at(index).map(str::to_string),
text_of(&dictionary.value_at(index))
);
}
assert_eq!(dictionary.text_at(4), None, "past the end");
}
#[test]
fn text_is_refused_where_it_is_not_stored_as_itself() {
let nulls =
Vector::from_values(LogicalType::Varchar, &[Value::Varchar("red".into()), Value::Null])
.unwrap();
assert_eq!(nulls.text_at(0), Some("red"));
assert_eq!(nulls.text_at(1), None, "a null has no text");
let constant = Vector::constant(LogicalType::Varchar, Value::Varchar("red".into()), 3);
assert_eq!(constant.text_at(0), None, "a constant is not stored per position");
assert_eq!(integers(&[1, 2]).text_at(0), None, "an integer is not text");
let mut bytes = StringColumn::new();
bytes.push("red");
let blob = Vector::flat(LogicalType::Blob, Data::Varlen(bytes)).unwrap();
assert_eq!(blob.text_at(0), None, "a blob is not a varchar");
}
#[test]
fn a_signed_integer_is_read_where_it_already_is_for_the_forms_that_store_it() {
let flat = integers(&[7, -3, 0, 2]);
for index in 0..flat.len() {
assert_eq!(flat.signed_at(index), signed_of(&flat.value_at(index)), "flat {index}");
}
let dictionary = Vector::dictionary(vec![1, 0, 3, 2], flat).unwrap();
for index in 0..dictionary.len() {
assert_eq!(
dictionary.signed_at(index),
signed_of(&dictionary.value_at(index)),
"dictionary {index}"
);
}
assert_eq!(dictionary.signed_at(4), None, "past the end");
let runs = Vector::runs(vec![2, 5], integers(&[4, 9])).unwrap();
for index in 0..runs.len() {
assert_eq!(runs.signed_at(index), signed_of(&runs.value_at(index)), "run {index}");
}
let constant = Vector::constant(LogicalType::BigInt, Value::BigInt(11), 3);
assert_eq!(constant.signed_at(2), Some(11));
let sequence = Vector::sequence(100, 5, 4);
for index in 0..sequence.len() {
assert_eq!(
sequence.signed_at(index),
signed_of(&sequence.value_at(index)),
"sequence {index}"
);
}
}
#[test]
fn a_window_of_a_page_packs_the_way_the_same_rows_owned_do() {
let wide: Vec<i32> = (0..122_880)
.map(|at| if at % 2 == 0 { i32::MIN + 5 + at } else { i32::MAX - 9 - at })
.collect();
let narrow: Vec<i32> = (0..122_880).map(|at| 1_000 + at % 200).collect();
for values in [wide, narrow] {
let page = integers(&values).into_pages();
let window = page.slice(0, 8_192).unwrap();
let owned = integers(&values[..8_192]);
let packed_window = window.bit_packed().unwrap();
let packed_owned = owned.bit_packed().unwrap();
assert_eq!(
packed_window.packed_parts().is_some(),
packed_owned.packed_parts().is_some()
);
for at in [0, 1, 4_095, 8_191] {
assert_eq!(packed_window.value_at(at), owned.value_at(at));
}
}
}
#[test]
fn a_signed_integer_is_refused_where_it_is_not_stored_as_itself() {
let nulls =
Vector::from_values(LogicalType::BigInt, &[Value::BigInt(4), Value::Null]).unwrap();
assert_eq!(nulls.signed_at(0), Some(4));
assert_eq!(nulls.signed_at(1), None, "a null is not a number");
let packed = integers(&[1, 2, 3, 1]).bit_packed().unwrap();
assert_eq!(packed.signed_at(0), Some(1), "a packed integer is read in code space");
let mut bytes = StringColumn::new();
bytes.push("red");
let text = Vector::flat(LogicalType::Varchar, Data::Varlen(bytes)).unwrap();
assert_eq!(text.signed_at(0), None, "a string is not a number");
let double = Vector::flat(LogicalType::Double, Data::Float64(vec![1.5].into())).unwrap();
assert_eq!(double.signed_at(0), None, "a double is not a signed integer");
}
#[test]
fn a_block_of_signed_integers_holds_what_the_row_at_a_time_accessor_hands_back() {
let mut out = Vec::new();
let shapes = [
integers(&[7, -3, 0, 2]),
Vector::flat(LogicalType::Integer, Data::Int32(vec![5, -6, 7].into())).unwrap(),
Vector::flat(LogicalType::SmallInt, Data::Int16(vec![1, -2].into())).unwrap(),
Vector::flat(LogicalType::TinyInt, Data::Int8(vec![-128, 127].into())).unwrap(),
Vector::constant(LogicalType::BigInt, Value::BigInt(11), 3),
Vector::sequence(100, 5, 4),
integers(&[1, 2, 3, 1]).bit_packed().unwrap(),
Vector::dictionary(vec![1, 0, 1, 3], integers(&[7, -3, 0, 2])).unwrap(),
Vector::dictionary(
vec![2, 2, 0],
Vector::flat(LogicalType::SmallInt, Data::Int16(vec![9, -9, 4].into())).unwrap(),
)
.unwrap(),
];
for column in &shapes {
assert!(column.signed_block(&mut out), "{:?} hands over a block", column.form());
assert_eq!(out.len(), column.len(), "{:?} filled the whole chunk", column.form());
for (index, &held) in out.iter().enumerate() {
assert_eq!(
Some(i128::from(held)),
column.signed_at(index),
"{:?} at {index}",
column.form()
);
}
}
}
#[test]
fn a_gather_of_signed_integers_holds_what_the_row_at_a_time_accessor_hands_back() {
let mut out = Vec::new();
let at = [0, 2, 2, 3];
let shapes = [
integers(&[7, -3, 0, 2]),
Vector::flat(LogicalType::Integer, Data::Int32(vec![5, -6, 7, -8].into())).unwrap(),
Vector::flat(LogicalType::TinyInt, Data::Int8(vec![-128, 127, 1, 0].into())).unwrap(),
];
for column in &shapes {
assert!(column.signed_gather(&at, &mut out), "{:?} is gathered", column.logical_type());
let wanted: Vec<i64> = at
.iter()
.map(|&row| i64::try_from(column.signed_at(row as usize).unwrap()).unwrap())
.collect();
assert_eq!(out, wanted);
}
let short = integers(&[1, 2, 3]);
assert!(!short.signed_gather(&at, &mut out), "row 3 is past the end");
assert!(out.is_empty());
assert!(!Vector::sequence(100, 5, 4).signed_gather(&at, &mut out));
assert!(integers(&[1]).signed_gather(&[], &mut out) && out.is_empty());
}
#[test]
fn the_runs_of_a_flat_column_end_where_its_values_change() {
let mut out = Vec::new();
let column =
Vector::flat(LogicalType::SmallInt, Data::Int16(vec![4, 4, 4, -1, -1, 4, 9].into()))
.unwrap();
assert!(column.signed_runs((1, 7), 1, &mut out));
assert_eq!(out, [(4, 3), (-1, 5), (4, 6), (9, 7)]);
assert!(!column.signed_runs((1, 8), 1, &mut out), "row 7 is past the end");
assert!(out.is_empty());
let changing = integers(&(0..1000).collect::<Vec<_>>());
assert!(!changing.signed_runs((0, 1000), 8, &mut out));
assert!(out.is_empty());
assert!(!Vector::sequence(100, 5, 4).signed_runs((0, 4), 8, &mut out));
}
#[test]
fn a_block_of_picked_row_numbers_holds_the_numbers_picked() {
let mut out = Vec::new();
let picked =
Vector::dictionary(vec![0, 3, 3, 8191], Vector::sequence(100, 5, 8192)).unwrap();
assert!(picked.signed_block(&mut out));
assert_eq!(out, [100, 115, 115, 100 + 5 * 8191]);
}
#[test]
fn a_block_is_refused_for_the_shapes_it_would_have_to_gather_or_widen() {
let mut out = Vec::new();
let nulled =
Vector::from_values(LogicalType::BigInt, &[Value::BigInt(4), Value::Null]).unwrap();
assert!(
!Vector::dictionary(vec![1, 0], nulled).unwrap().signed_block(&mut out),
"a dictionary with a null entry would hand its row over as a number"
);
assert!(!Vector::runs(vec![2, 5], integers(&[4, 9])).unwrap().signed_block(&mut out));
let wide = Vector::flat(LogicalType::HugeInt, Data::Int128(vec![1, 2].into())).unwrap();
assert!(!wide.signed_block(&mut out), "a hugeint does not fit sixty four bits");
let double = Vector::flat(LogicalType::Double, Data::Float64(vec![1.5].into())).unwrap();
assert!(!double.signed_block(&mut out), "a double is not a signed integer");
assert!(out.is_empty(), "a refusal leaves the buffer empty");
let nulls =
Vector::from_values(LogicalType::BigInt, &[Value::BigInt(4), Value::Null]).unwrap();
assert!(nulls.signed_block(&mut out), "a flat column with nulls still hands over");
assert_eq!(out[0], 4);
}
#[test]
fn a_vector_says_whether_it_holds_any_null_at_all() {
let flat = integers(&[7, -3, 0, 2]);
assert!(flat.none_null());
let nulls =
Vector::from_values(LogicalType::BigInt, &[Value::BigInt(4), Value::Null]).unwrap();
assert!(!nulls.none_null());
assert!(Vector::dictionary(vec![1, 0], flat.clone()).unwrap().none_null());
let holed = Vector::dictionary(vec![0, 0], nulls.clone()).unwrap();
assert!(!holed.none_null(), "a dictionary is read through to its values");
assert!(!holed.is_null_at(0), "and no code points at the null it holds");
assert!(Vector::runs(vec![2, 5], integers(&[4, 9])).unwrap().none_null());
assert!(!Vector::runs(vec![1, 2], nulls).unwrap().none_null());
assert!(Vector::constant(LogicalType::BigInt, Value::BigInt(11), 3).none_null());
assert!(!Vector::constant(LogicalType::BigInt, Value::Null, 3).none_null());
}
fn signed_of(value: &Value) -> Option<i128> {
match value {
Value::TinyInt(x) => Some(i128::from(*x)),
Value::SmallInt(x) => Some(i128::from(*x)),
Value::Integer(x) | Value::Date(x) => Some(i128::from(*x)),
Value::BigInt(x) | Value::Time(x) | Value::Timestamp(x) => Some(i128::from(*x)),
Value::HugeInt(x) | Value::Decimal { unscaled: x, .. } => Some(*x),
_ => None,
}
}
fn text_of(value: &Value) -> Option<String> {
match value {
Value::Varchar(text) => Some(text.clone()),
_ => None,
}
}
#[test]
fn a_dictionary_code_past_the_end_is_refused() {
let values = integers(&[1, 2]);
assert!(Vector::dictionary(vec![0, 2], values).is_err());
let empty = Vector::dictionary(Vec::new(), integers(&[])).expect("no codes, no values");
assert_eq!(empty.len(), 0);
assert!(Vector::dictionary(vec![0], integers(&[])).is_err());
}
#[test]
fn every_form_flattens_to_the_same_values_it_reads_out() {
let mut column = StringColumn::new();
column.push("alpha");
column.push("beta");
let dictionary = Vector::dictionary(
vec![1, 0, 1],
Vector::flat(LogicalType::Varchar, Data::Varlen(column)).unwrap(),
)
.unwrap();
let cases = [
Vector::constant(LogicalType::Integer, Value::Integer(3), 5),
Vector::sequence(7, -2, 5),
dictionary,
];
for vector in cases {
let flat = vector.flatten().unwrap();
assert_eq!(flat.form(), Form::Flat);
assert_eq!(flat.len(), vector.len());
for index in 0..vector.len() {
assert_eq!(flat.value_at(index), vector.value_at(index), "at {index}");
}
}
}
#[test]
fn a_null_still_occupies_a_position_after_flattening() {
let vector = Vector::sequence(0, 1, 4).with_validity(Validity::from_iter(4, |i| i != 1));
let flat = vector.flatten().unwrap();
assert_eq!(flat.value_at(0), Value::BigInt(0));
assert_eq!(flat.value_at(1), Value::Null);
assert_eq!(flat.value_at(2), Value::BigInt(2));
assert_eq!(flat.value_at(3), Value::BigInt(3));
}
#[test]
fn a_null_behind_a_dictionary_survives_flattening() {
let values =
Vector::from_values(LogicalType::Integer, &[Value::Integer(3), Value::Null]).unwrap();
let dictionary = Vector::dictionary(vec![1, 0, 1], values).unwrap();
let flat = dictionary.flatten().unwrap();
assert_eq!(flat.value_at(0), Value::Null);
assert_eq!(flat.value_at(1), Value::Integer(3));
assert_eq!(flat.value_at(2), Value::Null);
}
#[test]
fn gathering_reads_what_reading_one_position_at_a_time_reads() {
let mut column = StringColumn::new();
column.push("alpha");
column.push("beta");
column.push("gamma");
let cases = [
integers(&[10, 20, 30, 40]),
integers(&[10, 20, 30, 40]).with_validity(Validity::from_iter(4, |i| i != 2)),
Vector::constant(LogicalType::Integer, Value::Integer(9), 4),
Vector::sequence(100, -7, 4),
Vector::sequence(100, -7, 4).with_validity(Validity::from_iter(4, |i| i % 2 == 0)),
Vector::dictionary(
vec![2, 0, 1, 2],
Vector::flat(LogicalType::Varchar, Data::Varlen(column)).unwrap(),
)
.unwrap(),
Vector::dictionary(
vec![1, 0, 1, 0],
Vector::from_values(LogicalType::Integer, &[Value::Integer(5), Value::Null])
.unwrap(),
)
.unwrap(),
];
let wanted = [3_u32, 0, 2, 2, 1];
for vector in cases {
let gathered = vector.gather(&wanted).unwrap();
assert_eq!(gathered.len(), wanted.len());
assert_eq!(gathered.logical_type(), vector.logical_type());
for (slot, &index) in wanted.iter().enumerate() {
assert_eq!(
gathered.value_at(slot),
vector.value_at(index as usize),
"slot {slot} of {:?}",
vector.form()
);
}
}
}
#[test]
fn gathering_a_position_that_is_not_there_is_a_null_and_not_a_wrong_value() {
let vector = integers(&[1, 2, 3]);
let gathered = vector.gather(&[2, 9]).unwrap();
assert_eq!(gathered.value_at(0), Value::Integer(3));
assert_eq!(gathered.value_at(1), Value::Null);
}
#[test]
fn gathering_from_a_vector_of_no_values_is_that_many_nulls() {
let vector = Vector::flat(LogicalType::Null, Data::Empty).unwrap();
let gathered = vector.gather(&[0, 1, 2]).unwrap();
assert_eq!(gathered.len(), 3);
assert_eq!(gathered.value_at(0), Value::Null);
assert_eq!(gathered.value_at(2), Value::Null);
}
#[test]
fn gathering_a_constant_stays_a_constant() {
let vector = Vector::constant(LogicalType::Integer, Value::Integer(4), 100);
let gathered = vector.gather(&[7, 7, 99]).unwrap();
assert_eq!(gathered.form(), Form::Constant);
assert_eq!(gathered.len(), 3);
assert_eq!(gathered.value_at(2), Value::Integer(4));
}
#[test]
fn gathering_walks_a_dictionary_over_a_dictionary_to_the_values() {
let inner = Vector::dictionary(vec![2, 1, 0], integers(&[7, 8, 9]))
.unwrap()
.with_validity(Validity::from_iter(3, |index| index != 2));
let outer = Vector::dictionary(vec![1, 2], inner).unwrap();
let gathered = outer.gather(&[0, 1]).unwrap();
assert_eq!(gathered.form(), Form::Flat);
assert_eq!(gathered.value_at(0), Value::Integer(8));
assert_eq!(gathered.value_at(1), Value::Null);
}
#[test]
fn a_dictionary_over_a_dictionary_is_composed_into_one_level() {
let inner = Vector::dictionary(vec![2, 1, 0], integers(&[7, 8, 9])).unwrap();
let outer = Vector::dictionary(vec![1, 2], inner).unwrap();
let (codes, values) = outer.dictionary_parts().unwrap();
assert_eq!(codes, [1, 0]);
assert_eq!(values.form(), Form::Flat);
assert_eq!(outer.value_at(0), Value::Integer(8));
assert_eq!(outer.value_at(1), Value::Integer(7));
}
#[test]
fn stacking_dictionaries_does_not_make_them_deeper() {
let mut vector = integers(&[10, 20, 30, 40]);
for _ in 0..4 {
vector = Vector::dictionary(vec![3, 2, 1, 0], vector).unwrap();
}
let (codes, values) = vector.dictionary_parts().unwrap();
assert_eq!(values.form(), Form::Flat);
assert_eq!(codes, [0, 1, 2, 3]);
assert_eq!(
vector.iter().collect::<Vec<_>>(),
integers(&[10, 20, 30, 40]).iter().collect::<Vec<_>>()
);
}
#[test]
fn composing_a_dictionary_keeps_the_nulls_its_values_hold() {
let values =
Vector::from_values(LogicalType::Integer, &[Value::Integer(3), Value::Null]).unwrap();
let inner = Vector::dictionary(vec![1, 0, 1], values).unwrap();
let outer = Vector::dictionary(vec![0, 1], inner).unwrap();
assert_eq!(outer.dictionary_parts().unwrap().1.form(), Form::Flat);
assert_eq!(outer.value_at(0), Value::Null);
assert_eq!(outer.value_at(1), Value::Integer(3));
}
#[test]
fn a_dictionary_holding_its_own_nulls_is_not_composed_past() {
let inner = Vector::dictionary(vec![0, 1, 2], integers(&[1, 2, 3]))
.unwrap()
.with_validity(Validity::from_iter(3, |index| index != 1));
let outer = Vector::dictionary(vec![1, 2, 0], inner).unwrap();
assert_eq!(outer.dictionary_parts().unwrap().1.form(), Form::Dictionary);
assert_eq!(outer.value_at(0), Value::Null);
assert_eq!(outer.value_at(1), Value::Integer(3));
assert_eq!(outer.value_at(2), Value::Integer(1));
}
#[test]
fn a_null_behind_a_dictionary_reads_as_null_even_though_the_mask_says_otherwise() {
let values = Vector::flat(LogicalType::Integer, Data::Int32(vec![0, 7].into()))
.unwrap()
.with_validity(Validity::from_iter(2, |index| index != 0));
let vector = Vector::dictionary(vec![0, 1, 0], values).unwrap();
assert!(vector.validity().is_valid(0), "the mask at this level says present");
assert!(vector.is_null_at(0));
assert!(!vector.is_null_at(1));
assert!(vector.is_null_at(2));
assert!(vector.is_null_at(3), "a row past the end is null");
}
#[test]
fn a_null_inside_a_run_reads_as_null_even_though_the_mask_says_otherwise() {
let values = Vector::flat(LogicalType::Integer, Data::Int32(vec![0, 7].into()))
.unwrap()
.with_validity(Validity::from_iter(2, |index| index != 0));
let vector = Vector::runs(vec![2, 3], values).unwrap();
assert!(vector.validity().is_valid(0));
assert!(vector.is_null_at(0));
assert!(vector.is_null_at(1));
assert!(!vector.is_null_at(2));
}
#[test]
fn the_forms_that_hold_their_own_nulls_answer_the_same_either_way() {
let flat = Vector::flat(LogicalType::Integer, Data::Int32(vec![0, 7].into()))
.unwrap()
.with_validity(Validity::from_iter(2, |index| index != 0));
let constant = Vector::constant(LogicalType::Integer, Value::Null, 2);
let sequence = Vector::sequence(10, 2, 2);
for vector in [flat, constant, sequence] {
for row in 0..vector.len() {
assert_eq!(vector.is_null_at(row), !vector.validity().is_valid(row));
}
}
}
#[test]
fn flattening_a_flat_vector_is_the_same_vector() {
let vector = integers(&[1, 2, 3]);
assert_eq!(vector.flatten().unwrap(), vector);
}
#[test]
fn flattening_a_vector_that_owns_its_values_moves_them_rather_than_copying_them() {
let vector = integers(&[1, 2, 3, 4]);
let address = |vector: &Vector| match vector.data() {
Some(Data::Int32(values)) => values.as_slice().as_ptr() as usize,
_ => panic!("the layout changed under the test"),
};
let stored = address(&vector);
let flat = vector.into_flat().unwrap();
assert_eq!(address(&flat), stored, "the values moved");
assert_eq!(
flat.iter().collect::<Vec<_>>(),
(1..=4).map(Value::Integer).collect::<Vec<_>>()
);
let dictionary = Vector::dictionary(vec![1, 0, 1], integers(&[7, 8])).unwrap();
let flat = dictionary.clone().into_flat().unwrap();
assert_eq!(flat.form(), Form::Flat);
assert_eq!(flat.iter().collect::<Vec<_>>(), dictionary.iter().collect::<Vec<_>>());
}
#[test]
fn a_decimal_reads_its_width_and_scale_from_the_type_and_not_the_data() {
let ty = LogicalType::decimal(9, 2).unwrap();
let vector = Vector::flat(ty, Data::Int32(vec![1234].into())).unwrap();
assert_eq!(vector.value_at(0), Value::Decimal { unscaled: 1234, width: 9, scale: 2 });
assert_eq!(vector.value_at(0).to_string(), "12.34");
}
#[test]
fn a_decimal_writes_into_whichever_of_the_four_runs_its_precision_chose() {
for (width, scale, unscaled) in
[(4u8, 1u8, 25i128), (9, 2, 1234), (18, 3, 123_456), (38, 4, 1_234_567)]
{
let ty = LogicalType::decimal(width, scale).unwrap();
let value = Value::Decimal { unscaled, width, scale };
let vector = Vector::from_values(ty, &[value.clone(), Value::Null]).unwrap();
assert_eq!(vector.value_at(0), value, "a decimal of width {width}");
assert_eq!(vector.value_at(1), Value::Null, "a null decimal of width {width}");
}
}
#[test]
fn a_blob_holds_bytes_that_are_not_text() {
let bytes = |raw: &[u8]| Value::Blob(raw.to_vec());
let values = [
bytes(b"a\xffb"),
bytes(b"\x00\x01\x02"),
Value::Null,
bytes(b"\xed\xa0\x80 and long enough to leave the view"),
bytes(b""),
];
let vector = Vector::from_values(LogicalType::Blob, &values).unwrap();
for (index, value) in values.iter().enumerate() {
assert_eq!(&vector.value_at(index), value, "row {index}");
}
}
#[test]
fn a_decimal_too_wide_for_the_run_its_type_chose_is_an_error_and_not_a_wrong_number() {
let ty = LogicalType::decimal(4, 1).unwrap();
let value = Value::Decimal { unscaled: 1_000_000, width: 4, scale: 1 };
let error = Vector::from_values(ty, &[value]).unwrap_err();
assert!(error.to_string().contains("does not fit"), "{error}");
}
#[test]
fn a_flat_vector_costs_its_values_and_a_constant_costs_one() {
let flat = integers(&[1; 1000]);
assert!(
flat.footprint() >= 4000,
"a thousand i32 are four thousand bytes: {}",
flat.footprint()
);
let constant = Vector::constant(LogicalType::Integer, Value::Integer(1), 1_000_000);
assert!(constant.footprint() < 200, "a constant is one value: {}", constant.footprint());
let sequence = Vector::sequence(0, 1, 1_000_000);
assert!(sequence.footprint() < 200, "a sequence is two numbers: {}", sequence.footprint());
}
#[test]
fn a_gather_off_a_dictionary_answers_the_same_nulls_either_way_round() {
let words = [Value::Varchar("north".into()), Value::Null, Value::Varchar("south".into())];
let plain: Vec<Value> =
["north", "east", "south"].iter().map(|word| Value::Varchar((*word).into())).collect();
let clean = Arc::new(Vector::from_values(LogicalType::Varchar, &plain).unwrap());
let dirty = Arc::new(Vector::from_values(LogicalType::Varchar, &words).unwrap());
let codes = vec![0, 1, 2, 0, 1, 2];
let sources = [
Vector::stable_dictionary(codes.clone(), Arc::clone(&clean)).unwrap(),
Vector::stable_dictionary(codes.clone(), Arc::clone(&dirty)).unwrap(),
Vector::stable_dictionary(codes, Arc::clone(&clean))
.unwrap()
.with_validity(Validity::from_run(&[true, true, false, true, true, true])),
];
for source in &sources {
let picks: Vec<u32> = vec![5, 0, 3, 2, 1, 99, 4];
let taken = source.gather(&picks).unwrap();
for (row, &pick) in picks.iter().enumerate() {
assert_eq!(
taken.is_null_at(row),
source.is_null_at(pick as usize),
"row {row} of a gather of {picks:?}"
);
}
}
}
#[test]
fn a_dictionary_read_by_many_cuts_is_counted_about_once_between_them() {
let strings: Vec<Value> = (0..2000)
.map(|at| Value::Varchar(format!("a value well past the inline limit, number {at}")))
.collect();
let values = Arc::new(Vector::from_values(LogicalType::Varchar, &strings).unwrap());
let dictionary = values.footprint();
let cuts: Vec<Vector> = (0..500)
.map(|_| Vector::stable_dictionary(vec![0; 8], Arc::clone(&values)).unwrap())
.collect();
let together: usize = cuts.iter().map(Vector::footprint).sum();
assert!(
together < dictionary * 2,
"five hundred cuts are not five hundred dictionaries: {together} against {dictionary}"
);
assert!(
together > dictionary / 2,
"the dictionary is still counted: {together} against {dictionary}"
);
}
#[test]
fn a_string_vector_costs_the_bytes_of_its_long_strings() {
let short =
Vector::from_values(LogicalType::Varchar, &[Value::Varchar("red".into())]).unwrap();
let long = "a string well past the sixteen bytes a view holds inline".to_string();
let spilled =
Vector::from_values(LogicalType::Varchar, &[Value::Varchar(long.clone())]).unwrap();
assert!(
spilled.footprint() >= short.footprint() + long.len(),
"the arena is counted: {} against {}",
spilled.footprint(),
short.footprint()
);
}
#[test]
fn a_narrow_column_packs_and_reads_back_the_same_at_every_width() {
for width in 1..=20u32 {
let span = (1i64 << width) - 1;
let values: Vec<i64> =
(0..1000).map(|row| 1_000_000 + (row * 7919) % (span + 1)).collect();
let flat =
Vector::flat(LogicalType::BigInt, Data::Int64(values.clone().into())).unwrap();
let packed = flat.bit_packed().unwrap();
assert_eq!(packed.len(), flat.len());
assert_eq!(
packed.iter().collect::<Vec<_>>(),
flat.iter().collect::<Vec<_>>(),
"width {width} read back differently"
);
}
}
#[test]
fn the_width_is_the_bits_the_range_needs_and_not_the_bits_the_type_has() {
let values: Vec<i32> = (0..1024).map(|row| 40 + (row * 2560) / 1023).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into())).unwrap();
let packed = flat.bit_packed().unwrap();
assert_eq!(packed.form(), Form::BitPacked);
let parts = packed.packed_parts().expect("packed");
assert_eq!(parts.width(), 12, "0 to 2560 is twelve bits");
assert_eq!(parts.base(), 40);
assert!(
packed.footprint() * 2 < flat.footprint(),
"twelve bits against thirty two: {} against {}",
packed.footprint(),
flat.footprint()
);
}
#[test]
fn a_column_that_uses_its_whole_type_is_left_flat() {
let values: Vec<i32> = (0..1024).map(|row| row * 2_000_000 - 1_000_000_000).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into())).unwrap();
assert_eq!(flat.bit_packed().unwrap().form(), Form::Flat);
}
#[test]
fn a_column_against_the_top_of_its_type_packs_rather_than_being_refused() {
let values: Vec<i32> = (0..4096).map(|row| i32::MAX - (row % 1000)).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.clone().into())).unwrap();
let packed = flat.bit_packed().unwrap();
assert_eq!(packed.form(), Form::BitPacked);
let parts = packed.packed_parts().expect("packed");
assert_eq!(parts.width(), 10, "a thousand values apart is ten bits");
assert_eq!(
parts.base() + i128::from(u64::MAX >> (64 - parts.width())),
i128::from(i32::MAX),
"the widest code the width allows is the largest value the type holds"
);
assert_eq!(
packed.iter().collect::<Vec<_>>(),
flat.iter().collect::<Vec<_>>(),
"the values came back different"
);
}
#[test]
fn a_column_that_reaches_both_ends_of_its_type_bases_at_the_bottom_of_it() {
let values: Vec<i32> = (0..4096)
.map(|row| if row % 2 == 0 { i32::MIN + row } else { i32::MAX - row })
.collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.clone().into())).unwrap();
assert_eq!(flat.bit_packed().unwrap().form(), Form::Flat);
assert_eq!(
packing_base(&LogicalType::Integer, i128::from(i32::MIN), i128::from(i32::MAX), 32),
Some(i128::from(i32::MIN))
);
}
#[test]
fn a_column_of_one_value_is_left_to_the_run_length_form() {
let flat = integers(&[9; 1024]);
assert_eq!(flat.bit_packed().unwrap().form(), Form::Flat);
assert_eq!(flat.run_encoded().unwrap().form(), Form::Rle);
}
#[test]
fn a_string_column_has_no_range_to_pack() {
let text = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("blue".into())],
)
.unwrap();
assert_eq!(text.bit_packed().unwrap().form(), Form::Flat);
}
#[test]
fn a_cut_of_a_packed_column_stays_packed_and_shares_its_bits() {
let values: Vec<i32> = (0..1024).map(|row| 100 + row % 300).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into())).unwrap();
let packed = flat.bit_packed().unwrap();
let cut = packed.slice(500, 24).unwrap();
assert_eq!(cut.form(), Form::BitPacked);
assert_eq!(cut.len(), 24);
assert_eq!(
cut.iter().collect::<Vec<_>>(),
flat.slice(500, 24).unwrap().iter().collect::<Vec<_>>()
);
assert!(
cut.footprint() >= packed.footprint(),
"a cut shares the words rather than copying a piece of them"
);
}
#[test]
fn a_gather_of_a_packed_column_comes_out_flat_and_keeps_the_nulls() {
let values: Vec<i32> = (0..64).map(|row| 10 + row).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into())).unwrap();
let packed =
flat.bit_packed().unwrap().with_validity(Validity::from_iter(64, |row| row % 3 != 0));
let taken = packed.gather(&[0, 1, 2, 3, 62]).unwrap();
assert_eq!(taken.form(), Form::Flat);
assert_eq!(
taken.iter().collect::<Vec<_>>(),
vec![
Value::Null,
Value::Integer(11),
Value::Integer(12),
Value::Null,
Value::Integer(72)
]
);
}
#[test]
fn a_literal_outside_the_packed_range_has_no_code() {
let values: Vec<i32> = (0..256).map(|row| 1000 + row).collect();
let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into())).unwrap();
let packed = flat.bit_packed().unwrap();
let parts = packed.packed_parts().expect("packed");
assert_eq!(parts.code_of(1000), Some(0));
assert_eq!(parts.code_of(1100), Some(100));
assert_eq!(parts.code_of(999), None);
assert!(parts.ceiling() >= 1255);
assert_eq!(parts.code_of(parts.ceiling() + 1), None);
}
#[test]
fn packed_bits_can_be_handed_in_without_a_flat_vector_to_start_from() {
let packed = Vector::packed(LogicalType::SmallInt, vec![0x0000_0000_0000_4321], 4, 7, 4)
.expect("four codes of four bits");
assert_eq!(
packed.iter().collect::<Vec<_>>(),
vec![Value::SmallInt(8), Value::SmallInt(9), Value::SmallInt(10), Value::SmallInt(11)]
);
}
#[test]
fn a_packed_block_reads_what_each_row_reads() {
let words: Vec<u64> =
(0..400_u64).map(|word| word.wrapping_mul(0x9E37_79B9_7F4A_7C15)).collect();
for width in [1, 7, 13, 32, 33, 50] {
let whole = Vector::packed(LogicalType::BigInt, words.clone(), width, -1_000, 300)
.expect("enough words for 300 codes");
for (at, len) in [(0, 300), (1, 299), (63, 130), (64, 64), (100, 5), (250, 50)] {
let cut = whole.slice(at, len).expect("a cut inside the column");
let mut block = Vec::new();
assert!(cut.signed_block(&mut block));
let want: Vec<i64> = (0..len)
.map(|row| i64::try_from(cut.signed_at(row).expect("a row")).expect("fits"))
.collect();
assert_eq!(block, want, "width {width} cut at {at} for {len}");
}
}
}
#[test]
fn packed_bits_that_could_not_hold_what_they_claim_are_refused() {
assert!(Vector::packed(LogicalType::Varchar, vec![0], 4, 0, 4).is_err(), "not an integer");
assert!(Vector::packed(LogicalType::Integer, vec![0], 0, 0, 4).is_err(), "no width");
assert!(Vector::packed(LogicalType::Integer, vec![0], 64, 0, 4).is_err(), "too wide");
assert!(Vector::packed(LogicalType::Integer, vec![0], 8, 0, 9).is_err(), "too few words");
assert!(Vector::packed(LogicalType::TinyInt, vec![0], 8, 100, 8).is_err(), "would not fit");
}
fn long_strings(count: usize) -> Vector {
let values: Vec<Value> = (0..count)
.map(|row| {
Value::Varchar(format!("a string too long to sit inside a view, number {row}"))
})
.collect();
Vector::from_values(LogicalType::Varchar, &values).unwrap()
}
#[test]
fn a_string_column_in_view_form_reads_back_the_same_strings() {
let flat = long_strings(40);
let shared = flat.clone().shared_text().unwrap();
assert_eq!(shared.form(), Form::StringView);
assert_eq!(shared.len(), 40);
for row in 0..40 {
assert_eq!(shared.value_at(row), flat.value_at(row), "row {row}");
assert_eq!(shared.text_at(row), flat.text_at(row), "row {row}");
}
}
#[test]
fn a_short_string_is_read_out_of_its_view_and_never_out_of_the_arena() {
let flat = Vector::from_values(
LogicalType::Varchar,
&[Value::Varchar("red".into()), Value::Varchar("green".into()), Value::Null],
)
.unwrap();
let shared = flat.shared_text().unwrap();
let (views, arena) = shared.text_parts().unwrap();
assert!(arena.is_empty(), "three short strings need no arena");
assert_eq!(views[0].bytes_in(arena), Some(&b"red"[..]));
assert_eq!(shared.value_at(1), Value::Varchar("green".into()));
assert_eq!(shared.value_at(2), Value::Null, "the validity came across");
}
#[test]
fn a_cut_of_a_view_column_shares_the_arena_rather_than_copying_the_bytes() {
let shared = long_strings(64).shared_text().unwrap();
let cut = shared.slice(16, 8).unwrap();
assert_eq!(cut.form(), Form::StringView, "a cut of views is views");
assert_eq!(cut.len(), 8);
assert_eq!(cut.value_at(0), shared.value_at(16));
assert_eq!(cut.value_at(7), shared.value_at(23));
let (_, whole) = shared.text_parts().unwrap();
let (_, piece) = cut.text_parts().unwrap();
assert_eq!(piece.as_ptr(), whole.as_ptr(), "the cut shares the page");
assert_eq!(piece.len(), whole.len());
}
#[test]
fn a_flat_string_column_has_to_copy_the_bytes_its_cut_keeps() {
let flat = long_strings(64);
let cut = flat.slice(16, 8).unwrap();
assert_eq!(cut.form(), Form::Flat);
let (_, whole) = flat.text_parts().unwrap();
let (_, piece) = cut.text_parts().unwrap();
assert!(piece.len() < whole.len(), "the flat cut carries only what it kept");
}
#[test]
fn a_gather_of_a_view_column_keeps_the_form_and_a_flatten_copies_out_of_it() {
let shared = long_strings(32).shared_text().unwrap();
let picked: Vec<u32> = (0..32).step_by(3).collect();
let gathered = shared.gather(&picked).unwrap();
assert_eq!(gathered.form(), Form::StringView, "selecting rows moves views, not bytes");
assert_eq!(gathered.len(), picked.len());
for (row, &from) in picked.iter().enumerate() {
assert_eq!(gathered.value_at(row), shared.value_at(from as usize), "row {row}");
}
let flattened = gathered.flatten().unwrap();
assert_eq!(flattened.form(), Form::Flat);
assert_eq!(flattened.iter().collect::<Vec<_>>(), gathered.iter().collect::<Vec<_>>());
let (_, narrowed) = flattened.text_parts().unwrap();
let (_, whole) = shared.text_parts().unwrap();
assert!(narrowed.len() < whole.len(), "flattening lets the page go");
}
#[test]
fn a_null_in_a_view_column_survives_being_gathered_and_flattened() {
let shared = long_strings(8)
.with_validity(Validity::from_iter(8, |row| row % 3 != 0))
.shared_text()
.unwrap();
let gathered = shared.gather(&[0, 1, 2, 3, 4]).unwrap();
let expected =
[Value::Null, shared.value_at(1), shared.value_at(2), Value::Null, shared.value_at(4)];
assert_eq!(gathered.iter().collect::<Vec<_>>(), expected);
assert_eq!(gathered.flatten().unwrap().iter().collect::<Vec<_>>(), expected);
}
#[test]
fn both_string_forms_hand_a_kernel_the_same_views_and_the_same_bytes() {
let flat = long_strings(6);
let shared = flat.clone().shared_text().unwrap();
let (flat_views, flat_arena) = flat.text_parts().unwrap();
let (shared_views, shared_arena) = shared.text_parts().unwrap();
assert_eq!(flat_views.len(), shared_views.len());
for row in 0..6 {
assert_eq!(
flat_views[row].bytes_in(flat_arena),
shared_views[row].bytes_in(shared_arena),
"row {row}"
);
}
assert!(Vector::sequence(0, 1, 4).text_parts().is_none());
assert!(integers(&[1, 2, 3]).text_parts().is_none());
}
#[test]
fn a_column_that_is_not_strings_cannot_be_held_as_views() {
let views = vec![StringView::inline("red")];
let arena = Arc::new(Buffer::new());
let wrong = Vector::string_views(LogicalType::Integer, views, arena);
assert!(wrong.is_err(), "an integer column has no views");
assert_eq!(integers(&[1, 2]).shared_text().unwrap().form(), Form::Flat, "left alone");
}
fn sentences(count: usize) -> Vector {
let values: Vec<Value> = (0..count)
.map(|row| {
Value::Varchar(format!(
"http://example.test/catalogue/section/{}/item/{row}",
row % 7
))
})
.collect();
Vector::from_values(LogicalType::Varchar, &values).unwrap()
}
#[test]
fn a_compressed_column_reads_back_the_strings_that_went_into_it() {
let flat = sentences(64);
let coded = flat.clone().compressed().unwrap();
assert_eq!(coded.form(), Form::Fsst, "a text column compresses");
assert_eq!(coded.len(), 64);
for row in 0..64 {
assert_eq!(coded.value_at(row), flat.value_at(row), "row {row}");
}
assert_eq!(coded.flatten().unwrap(), flat, "flattening is the column it came from");
}
#[test]
fn compressing_halves_the_bytes_or_the_column_is_left_flat() {
let flat = sentences(200);
let coded = flat.clone().compressed().unwrap();
let parts = coded.coded_parts().expect("compressed");
assert_eq!(coded.text_at(0), None, "nothing to borrow until it is flattened");
let plain: usize = (0..200).map(|row| flat.text_at(row).map_or(0, str::len)).sum();
let codes: usize = (0..200).map(|row| parts.row(row).map_or(0, <[u8]>::len)).sum();
assert!(codes * FSST_PAYS_AT <= plain, "{codes} codes against {plain} bytes");
let mut seed = 0x2545_f491_4f6c_dd1du64;
let values: Vec<Value> = (0..256)
.map(|_| {
let mut text = String::new();
while text.len() < 12 {
seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1);
text.push(char::from(b'!' + ((seed >> 33) % 90) as u8));
}
Value::Varchar(text)
})
.collect();
let noise = Vector::from_values(LogicalType::Varchar, &values).unwrap();
assert_eq!(noise.compressed().unwrap().form(), Form::Flat);
}
#[test]
fn a_cut_of_a_compressed_column_shares_the_codes_and_the_table() {
let coded = sentences(64).compressed().unwrap();
let cut = coded.slice(8, 16).unwrap();
assert_eq!(cut.form(), Form::Fsst);
assert_eq!(cut.len(), 16);
for row in 0..16 {
assert_eq!(cut.value_at(row), coded.value_at(8 + row), "row {row}");
}
let (whole, piece) = (coded.coded_parts().unwrap(), cut.coded_parts().unwrap());
assert_eq!(piece.row(0), whole.row(8), "the spans point into the same codes");
}
#[test]
fn a_gather_of_a_compressed_column_stays_compressed_and_keeps_the_nulls() {
let coded = sentences(32)
.with_validity(Validity::from_iter(32, |row| row % 5 != 2))
.compressed()
.unwrap();
let picked: Vec<u32> = (0..32).step_by(2).collect();
let gathered = coded.gather(&picked).unwrap();
assert_eq!(gathered.form(), Form::Fsst, "selecting rows moves spans, not bytes");
for (row, &from) in picked.iter().enumerate() {
assert_eq!(gathered.value_at(row), coded.value_at(from as usize), "row {row}");
}
assert_eq!(
gathered.flatten().unwrap().iter().collect::<Vec<_>>(),
gathered.iter().collect::<Vec<_>>()
);
}
#[test]
fn a_literal_lands_in_the_same_codes_the_row_holding_it_does() {
let coded = sentences(40).compressed().unwrap();
let parts = coded.coded_parts().expect("compressed");
let text = coded.value_at(11);
let Value::Varchar(text) = text else { panic!("a string column reads back strings") };
assert_eq!(parts.encode(text.as_bytes()), parts.row(11).expect("row 11"));
assert_ne!(parts.encode(b"something else entirely"), parts.row(11).unwrap());
}
#[test]
fn codes_that_run_past_what_is_there_are_refused() {
let table = Arc::new(SymbolTable::empty());
let codes = Arc::new(vec![1u8, 2, 3, 4]);
let good = vec![(0u32, 2u32), (2, 4)];
assert!(
Vector::coded(LogicalType::Varchar, Arc::clone(&codes), good, Arc::clone(&table))
.is_ok()
);
let past = vec![(0u32, 9u32)];
assert!(
Vector::coded(LogicalType::Varchar, Arc::clone(&codes), past, Arc::clone(&table))
.is_err(),
"a span past the end of the codes"
);
let backwards = vec![(3u32, 1u32)];
assert!(
Vector::coded(LogicalType::Varchar, Arc::clone(&codes), backwards, Arc::clone(&table))
.is_err(),
"a span that ends before it starts"
);
let wrong = vec![(0u32, 2u32)];
assert!(
Vector::coded(LogicalType::Integer, codes, wrong, table).is_err(),
"an integer column has no codes"
);
}
#[test]
fn a_view_pointing_past_its_arena_is_refused_at_construction() {
let long = "a string too long to sit inside a view";
let arena: Arc<Buffer<u8>> = Arc::new(long.as_bytes().to_vec().into());
let good = vec![StringView::over(long.as_bytes(), 0)];
assert!(Vector::string_views(LogicalType::Varchar, good, Arc::clone(&arena)).is_ok());
let bad = vec![StringView::over(long.as_bytes(), 4)];
assert!(
Vector::string_views(LogicalType::Varchar, bad, arena).is_err(),
"four bytes short of what the view claims"
);
}
#[test]
fn a_gathered_vector_reads_the_source_row_its_id_names() {
let source = Arc::new(integers(&[10, 20, 30, 40]));
let vector = Vector::gathered(source, Arc::new(vec![3, 0, 3, 1])).unwrap();
assert_eq!(vector.form(), Form::Gathered);
assert_eq!(vector.len(), 4);
assert_eq!(
vector.iter().collect::<Vec<_>>(),
vec![Value::Integer(40), Value::Integer(10), Value::Integer(40), Value::Integer(20)]
);
}
#[test]
fn a_gathered_row_with_no_source_row_is_null_without_a_mask() {
let source = Arc::new(integers(&[10, 20]));
let vector = Vector::gathered(source, Arc::new(vec![1, NO_ROW, 0])).unwrap();
assert!(!vector.validity().has_nulls(vector.len()), "the mask at this level says nothing");
assert!(vector.is_null_at(1));
assert!(!vector.is_null_at(0) && !vector.is_null_at(2));
assert_eq!(
vector.iter().collect::<Vec<_>>(),
vec![Value::Integer(20), Value::Null, Value::Integer(10)]
);
assert!(!vector.none_null(), "a sentinel is a null and the bulk answer has to agree");
}
#[test]
fn a_gather_of_a_null_source_row_is_null() {
let source = Arc::new(
Vector::from_values(LogicalType::Integer, &[Value::Integer(7), Value::Null]).unwrap(),
);
let vector = Vector::gathered(source, Arc::new(vec![1, 0, 1])).unwrap();
assert!(vector.is_null_at(0) && vector.is_null_at(2));
assert_eq!(vector.value_at(1), Value::Integer(7));
assert!(!vector.none_null());
}
#[test]
fn a_gathered_id_past_the_end_of_its_source_is_refused() {
let source = Arc::new(integers(&[1, 2, 3]));
assert!(Vector::gathered(Arc::clone(&source), Arc::new(vec![0, 3])).is_err());
assert!(
Vector::gathered(source, Arc::new(vec![0, NO_ROW])).is_ok(),
"the sentinel is not an id past the end, it is the absence of one"
);
}
#[test]
fn cutting_a_gather_moves_where_it_starts_and_copies_nothing() {
let source = Arc::new(integers(&[10, 20, 30, 40, 50]));
let rids = Arc::new(vec![4, 3, 2, 1, 0]);
let vector = Vector::gathered(Arc::clone(&source), Arc::clone(&rids)).unwrap();
let held = Arc::strong_count(&rids);
let cut = vector.slice(1, 3).unwrap();
assert_eq!(cut.form(), Form::Gathered);
assert_eq!(
Arc::strong_count(&rids),
held + 1,
"the cut shares the ids rather than copying"
);
assert_eq!(
cut.iter().collect::<Vec<_>>(),
vec![Value::Integer(40), Value::Integer(30), Value::Integer(20)]
);
assert_eq!(cut.gathered_parts().unwrap().1, [3, 2, 1]);
}
#[test]
fn a_gather_of_a_gather_resolves_to_one_walk_over_the_source() {
let source = Arc::new(integers(&[10, 20, 30, 40]));
let inner = Vector::gathered(source, Arc::new(vec![3, 2, 1, 0])).unwrap();
let outer = inner.gather(&[0, 3]).unwrap();
assert_eq!(outer.iter().collect::<Vec<_>>(), vec![Value::Integer(40), Value::Integer(10)]);
assert_ne!(outer.form(), Form::Gathered, "the walk stops at what the ids point into");
}
#[test]
fn gathering_through_a_sentinel_keeps_it_null() {
let source = Arc::new(integers(&[10, 20]));
let inner = Vector::gathered(source, Arc::new(vec![0, NO_ROW, 1])).unwrap();
let outer = inner.gather(&[1, 2, 1]).unwrap();
assert_eq!(
outer.iter().collect::<Vec<_>>(),
vec![Value::Null, Value::Integer(20), Value::Null]
);
}
#[test]
fn folding_over_the_source_is_worth_it_only_when_the_source_is_the_shorter_one() {
let wide = Arc::new(integers(&(0..64).collect::<Vec<i32>>()));
let narrow = Arc::new(integers(&[1, 2]));
let off_wide = Vector::gathered(wide, Arc::new(vec![0, 1, 2])).unwrap();
let off_narrow = Vector::gathered(narrow, Arc::new(vec![0, 1, 0, 1, 0])).unwrap();
assert!(!off_wide.fold_over_source(), "sixty four source rows to answer three");
assert!(off_narrow.fold_over_source(), "two source rows to answer five");
assert!(!integers(&[1, 2]).fold_over_source(), "and every other form says no");
}
#[test]
fn a_gathered_string_is_read_where_the_source_put_it() {
let mut column = StringColumn::new();
column.push("red");
column.push("a string too long to sit inside a sixteen byte view");
let source = Arc::new(Vector::flat(LogicalType::Varchar, Data::Varlen(column)).unwrap());
let vector = Vector::gathered(source, Arc::new(vec![1, 0, NO_ROW])).unwrap();
assert_eq!(vector.text_at(0), Some("a string too long to sit inside a sixteen byte view"));
assert_eq!(vector.text_at(1), Some("red"));
assert_eq!(vector.text_at(2), None);
assert_eq!(vector.bytes_at(1), Some(b"red".as_slice()));
assert_eq!(vector.value_at(1), Value::Varchar("red".into()));
}
#[test]
fn the_signed_reader_of_a_gather_agrees_with_the_value_reader() {
let source = Arc::new(integers(&[10, 20, 30]));
let vector = Vector::gathered(source, Arc::new(vec![2, NO_ROW, 0, 1])).unwrap();
for row in 0..vector.len() {
let signed = vector.signed_at(row);
match vector.value_at(row) {
Value::Null => assert_eq!(signed, None),
Value::Integer(held) => assert_eq!(signed, Some(i128::from(held))),
other => panic!("an integer column answered {other}"),
}
}
}
#[test]
fn flattening_a_gather_writes_out_the_rows_it_pointed_at() {
let source = Arc::new(integers(&[10, 20, 30]));
let vector = Vector::gathered(source, Arc::new(vec![2, NO_ROW, 0])).unwrap();
let flat = vector.flatten().unwrap();
assert_eq!(flat.form(), Form::Flat);
assert_eq!(
flat.iter().collect::<Vec<_>>(),
vec![Value::Integer(30), Value::Null, Value::Integer(10)]
);
}
#[test]
fn a_parent_gathered_by_many_columns_is_counted_about_once_between_them() {
let source = Arc::new(integers(&(0..4096).collect::<Vec<i32>>()));
let rids = Arc::new(vec![0; 64]);
let alone = Vector::gathered(Arc::clone(&source), Arc::clone(&rids)).unwrap().footprint();
let many = (0..8)
.map(|_| Vector::gathered(Arc::clone(&source), Arc::clone(&rids)).unwrap())
.collect::<Vec<_>>();
let together = many.iter().map(Vector::footprint).sum::<usize>();
assert!(
together < alone * 2,
"eight gathers off one parent reported {together} against {alone} for one"
);
}
}