use std::{fmt::Debug, marker::PhantomData, sync::Arc};
use arrow_array::{
Array, ArrayRef, ArrowPrimitiveType, FixedSizeBinaryArray, GenericByteArray, PrimitiveArray,
builder::{
ArrayBuilder, BooleanBuilder, FixedSizeBinaryBuilder, GenericByteBuilder, LargeBinaryBuilder,
LargeStringBuilder, NullBuilder, PrimitiveBuilder,
},
types::{
ByteArrayType, Date32Type, Decimal128Type, Decimal256Type, Float32Type, Float64Type, Int8Type,
Int16Type, Int32Type, Int64Type, IntervalMonthDayNanoType, Time64NanosecondType,
TimestampNanosecondType, UInt8Type, UInt16Type, UInt32Type, UInt64Type,
},
};
use arrow_buffer::{Buffer, NullBuffer, bit_util, i256};
use arrow_schema::FieldRef;
use reifydb_value::{
Result,
value::{
column_view::{ColumnView, ViewData},
constraint::{bytes::MaxBytes, precision::Precision, scale::Scale},
container::{
decimal_array::{self, DECIMAL128_MAX_PRECISION, DecimalArray, DecimalView, decimal_at},
dictionary_array::DICTIONARY_ENTRY_WIDTH,
temporal_array::datetime_timezone,
uuid_array::UUID_WIDTH,
varlen_array,
wide_int_array::WideInt,
},
decimal::{Decimal, unscaled},
dictionary::DictionaryId,
value_type::{
ValueType,
field::{FieldType, named},
},
},
};
use crate::{internal_err, internal_error, value::column::push::Push};
#[derive(Debug)]
pub enum DecimalBuilder {
Decimal128 {
builder: PrimitiveBuilder<Decimal128Type>,
precision: Precision,
scale: Scale,
},
Decimal256 {
builder: PrimitiveBuilder<Decimal256Type>,
precision: Precision,
scale: Scale,
},
}
impl DecimalBuilder {
pub fn with_capacity(precision: Precision, scale: Scale, capacity: usize) -> Self {
let data_type = decimal_array::data_type(precision, scale);
if precision.value() <= DECIMAL128_MAX_PRECISION {
DecimalBuilder::Decimal128 {
builder: PrimitiveBuilder::with_capacity(capacity).with_data_type(data_type),
precision,
scale,
}
} else {
DecimalBuilder::Decimal256 {
builder: PrimitiveBuilder::with_capacity(capacity).with_data_type(data_type),
precision,
scale,
}
}
}
pub(crate) fn from_array(array: DecimalArray) -> Self {
let (precision, scale) = (array.precision(), array.scale());
match array {
DecimalArray::Decimal128(array) => DecimalBuilder::Decimal128 {
builder: primitive_builder(array),
precision,
scale,
},
DecimalArray::Decimal256(array) => DecimalBuilder::Decimal256 {
builder: primitive_builder(array),
precision,
scale,
},
}
}
pub fn precision(&self) -> Precision {
match self {
DecimalBuilder::Decimal128 {
precision,
..
}
| DecimalBuilder::Decimal256 {
precision,
..
} => *precision,
}
}
pub fn scale(&self) -> Scale {
match self {
DecimalBuilder::Decimal128 {
scale,
..
}
| DecimalBuilder::Decimal256 {
scale,
..
} => *scale,
}
}
pub fn len(&self) -> usize {
match self {
DecimalBuilder::Decimal128 {
builder,
..
} => builder.len(),
DecimalBuilder::Decimal256 {
builder,
..
} => builder.len(),
}
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn finish(&mut self) -> DecimalArray {
match self {
DecimalBuilder::Decimal128 {
builder,
..
} => DecimalArray::Decimal128(builder.finish()),
DecimalBuilder::Decimal256 {
builder,
..
} => DecimalArray::Decimal256(builder.finish()),
}
}
pub(crate) fn append_null(&mut self) {
match self {
DecimalBuilder::Decimal128 {
builder,
..
} => builder.append_null(),
DecimalBuilder::Decimal256 {
builder,
..
} => builder.append_null(),
}
}
pub fn push(&mut self, value: &Decimal) {
if let Some(fitted) = value.fits(self.precision().value(), self.scale().value()) {
return self.append_fitting(fitted.unscaled());
}
let fitted = self.widen_for(value);
self.append_fitting(fitted.unscaled());
}
pub(crate) fn append_array(&mut self, array: DecimalView<'_>) {
match (self, array) {
(
DecimalBuilder::Decimal128 {
builder,
precision,
scale,
},
DecimalView::Decimal128(array),
) if array.precision() == precision.value() && array.scale() as u8 == scale.value() => builder.append_array(array),
(
DecimalBuilder::Decimal256 {
builder,
precision,
scale,
},
DecimalView::Decimal256(array),
) if array.precision() == precision.value() && array.scale() as u8 == scale.value() => builder.append_array(array),
(this, array) => {
let nulls = decimal_nulls(array);
for index in 0..array.len() {
if nulls.as_ref().is_some_and(|nulls| nulls.is_null(index)) {
this.append_null();
continue;
}
let value = decimal_at(array, index).expect("index is below the array length");
this.push(&value);
}
}
}
}
fn append_fitting(&mut self, unscaled: i256) {
match self {
DecimalBuilder::Decimal128 {
builder,
..
} => builder.append_value(unscaled.to_i128().expect("a value within precision 38 fits i128")),
DecimalBuilder::Decimal256 {
builder,
..
} => builder.append_value(unscaled),
}
}
fn widen_for(&mut self, value: &Decimal) -> Decimal {
let (precision, scale) = (self.precision().value(), self.scale().value());
let finished = self.finish();
let nulls = decimal_nulls(&finished);
let existing: Vec<Decimal> = finished
.unscaled_values()
.into_iter()
.map(|unscaled| {
Decimal::from_parts(unscaled, scale)
.expect("a decimal builder holds only valid decimals")
})
.collect();
let integer_digits = |decimal: &Decimal| decimal.digits().saturating_sub(decimal.scale());
let needed = existing.iter().chain([value]).map(integer_digits).max().unwrap_or(0);
let wide_scale = scale.max(value.scale()).min(unscaled::MAX_DIGITS - needed);
let round = |decimal: &Decimal| {
decimal.round_to_scale(wide_scale)
.expect("rounding never adds a whole digit past the widest value")
};
let rounded: Vec<Decimal> = existing.iter().map(round).collect();
let fitted = round(value);
let wide_precision = ((precision - scale).max(needed) + wide_scale).min(unscaled::MAX_DIGITS);
let mut widened = DecimalBuilder::with_capacity(
Precision::new(wide_precision),
Scale::new(wide_scale),
rounded.len() + 1,
);
for (index, decimal) in rounded.into_iter().enumerate() {
match nulls.as_ref().is_some_and(|nulls| nulls.is_null(index)) {
true => widened.append_null(),
false => widened.append_fitting(decimal.unscaled()),
}
}
*self = widened;
fitted
}
}
fn decimal_nulls<'a>(array: impl Into<DecimalView<'a>>) -> Option<NullBuffer> {
array.into().array().logical_nulls()
}
#[derive(Debug)]
pub struct WideBuilder<T: WideInt> {
builder: FixedSizeBinaryBuilder,
marker: PhantomData<T>,
}
impl<T: WideInt> WideBuilder<T> {
pub(crate) fn with_capacity(capacity: usize) -> Self {
Self {
builder: FixedSizeBinaryBuilder::with_capacity(capacity, T::WIDTH as i32),
marker: PhantomData,
}
}
pub(crate) fn from_array(array: &FixedSizeBinaryArray) -> Self {
Self {
builder: fixed_size_builder(array),
marker: PhantomData,
}
}
pub(crate) fn append_null(&mut self) {
let mut row = [0u8; 16];
T::default().to_ordered(&mut row);
let null_row = FixedSizeBinaryArray::new(
T::WIDTH as i32,
Buffer::from(row[..T::WIDTH].to_vec()),
Some(NullBuffer::new_null(1)),
);
append_fixed_array(&mut self.builder, &null_row)
.expect("a null row of the builder width always appends");
}
pub(crate) fn append_value(&mut self, value: T) {
let mut row = [0u8; 16];
value.to_ordered(&mut row);
append_fixed(&mut self.builder, &row);
}
pub(crate) fn append_array(&mut self, array: &FixedSizeBinaryArray) -> Result<()> {
append_fixed_array(&mut self.builder, array)
}
pub(crate) fn len(&self) -> usize {
self.builder.len()
}
pub(crate) fn finish(&mut self) -> FixedSizeBinaryArray {
self.builder.finish()
}
}
#[derive(Debug)]
pub struct ColumnBuilder {
pub(crate) inner: TypedBuilder,
pub(crate) optional: bool,
}
#[derive(Debug)]
pub enum TypedBuilder {
Bool(BooleanBuilder),
Int1(PrimitiveBuilder<Int8Type>),
Int2(PrimitiveBuilder<Int16Type>),
Int4(PrimitiveBuilder<Int32Type>),
Int8(PrimitiveBuilder<Int64Type>),
Int16(WideBuilder<i128>),
Uint1(PrimitiveBuilder<UInt8Type>),
Uint2(PrimitiveBuilder<UInt16Type>),
Uint4(PrimitiveBuilder<UInt32Type>),
Uint8(PrimitiveBuilder<UInt64Type>),
Uint16(WideBuilder<u128>),
Float4(PrimitiveBuilder<Float32Type>),
Float8(PrimitiveBuilder<Float64Type>),
Date(PrimitiveBuilder<Date32Type>),
DateTime(PrimitiveBuilder<TimestampNanosecondType>),
Time(PrimitiveBuilder<Time64NanosecondType>),
Duration(PrimitiveBuilder<IntervalMonthDayNanoType>),
IdentityId(FixedSizeBinaryBuilder),
Uuid4(FixedSizeBinaryBuilder),
Uuid7(FixedSizeBinaryBuilder),
DictionaryId {
builder: FixedSizeBinaryBuilder,
dictionary_id: Option<DictionaryId>,
},
Utf8 {
builder: LargeStringBuilder,
max_bytes: MaxBytes,
},
Blob {
builder: LargeBinaryBuilder,
max_bytes: MaxBytes,
},
Decimal(DecimalBuilder),
Any {
builder: LargeBinaryBuilder,
declared_type: Option<ValueType>,
},
Digest {
builder: LargeBinaryBuilder,
inner: ValueType,
accuracy: u32,
},
None(NullBuilder),
}
impl ColumnBuilder {
pub fn with_capacity(target: ValueType, capacity: usize) -> Self {
let inner = match target {
ValueType::Option(inner) => {
return ColumnBuilder {
optional: true,
..ColumnBuilder::with_capacity(*inner, capacity)
};
}
ValueType::Boolean => TypedBuilder::Bool(BooleanBuilder::with_capacity(capacity)),
ValueType::Int1 => TypedBuilder::Int1(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Int2 => TypedBuilder::Int2(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Int4 => TypedBuilder::Int4(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Int8 => TypedBuilder::Int8(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Int16 => TypedBuilder::Int16(WideBuilder::with_capacity(capacity)),
ValueType::Uint1 => TypedBuilder::Uint1(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Uint2 => TypedBuilder::Uint2(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Uint4 => TypedBuilder::Uint4(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Uint8 => TypedBuilder::Uint8(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Uint16 => TypedBuilder::Uint16(WideBuilder::with_capacity(capacity)),
ValueType::Float4 => TypedBuilder::Float4(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Float8 => TypedBuilder::Float8(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Date => TypedBuilder::Date(PrimitiveBuilder::with_capacity(capacity)),
ValueType::DateTime => TypedBuilder::DateTime(
PrimitiveBuilder::<TimestampNanosecondType>::with_capacity(capacity)
.with_timezone(datetime_timezone()),
),
ValueType::Time => TypedBuilder::Time(PrimitiveBuilder::with_capacity(capacity)),
ValueType::Duration => TypedBuilder::Duration(PrimitiveBuilder::with_capacity(capacity)),
ValueType::IdentityId => TypedBuilder::IdentityId(FixedSizeBinaryBuilder::with_capacity(
capacity,
UUID_WIDTH as i32,
)),
ValueType::Uuid4 => {
TypedBuilder::Uuid4(FixedSizeBinaryBuilder::with_capacity(capacity, UUID_WIDTH as i32))
}
ValueType::Uuid7 => {
TypedBuilder::Uuid7(FixedSizeBinaryBuilder::with_capacity(capacity, UUID_WIDTH as i32))
}
ValueType::DictionaryId => TypedBuilder::DictionaryId {
builder: FixedSizeBinaryBuilder::with_capacity(capacity, DICTIONARY_ENTRY_WIDTH as i32),
dictionary_id: None,
},
ValueType::Utf8 => TypedBuilder::Utf8 {
builder: LargeStringBuilder::with_capacity(capacity, capacity * 16),
max_bytes: MaxBytes::MAX,
},
ValueType::Blob => TypedBuilder::Blob {
builder: LargeBinaryBuilder::with_capacity(capacity, capacity * 32),
max_bytes: MaxBytes::MAX,
},
ValueType::Decimal {
precision,
scale,
} => TypedBuilder::Decimal(DecimalBuilder::with_capacity(precision, scale, capacity)),
ValueType::Any | ValueType::Tuple(_) => TypedBuilder::Any {
builder: LargeBinaryBuilder::with_capacity(capacity, 0),
declared_type: None,
},
declared @ (ValueType::List(_) | ValueType::Record(_)) => TypedBuilder::Any {
builder: LargeBinaryBuilder::with_capacity(capacity, 0),
declared_type: Some(declared),
},
ValueType::Digest {
inner,
accuracy,
} => TypedBuilder::Digest {
builder: LargeBinaryBuilder::with_capacity(capacity, 0),
inner: *inner,
accuracy,
},
};
ColumnBuilder {
inner,
optional: false,
}
}
pub fn like(view: &ColumnView, capacity: usize) -> Self {
if view.is_none() {
return ColumnBuilder::untyped_none();
}
let mut builder = ColumnBuilder::with_capacity(view.get_type(), capacity);
builder.copy_details(&view.data);
builder
}
pub fn from_view(view: &ColumnView) -> Self {
ColumnBuilder {
inner: TypedBuilder::from_view(&view.data),
optional: view.is_nullable() || view.none_count() > 0,
}
}
pub(crate) fn untyped_none() -> Self {
ColumnBuilder {
inner: TypedBuilder::None(NullBuilder::new()),
optional: true,
}
}
pub fn inner(&self) -> &TypedBuilder {
&self.inner
}
pub fn push<T>(&mut self, value: T)
where
ColumnBuilder: Push<T>,
T: Debug,
{
<Self as Push<T>>::push(self, value)
}
pub fn set_dictionary_id(&mut self, id: DictionaryId) {
if let TypedBuilder::DictionaryId {
dictionary_id,
..
} = &mut self.inner
{
*dictionary_id = Some(id);
}
}
pub fn dictionary_id(&self) -> Option<DictionaryId> {
match &self.inner {
TypedBuilder::DictionaryId {
dictionary_id,
..
} => *dictionary_id,
_ => None,
}
}
pub fn extend(&mut self, other: &ColumnView) -> Result<()> {
match (self.is_untyped_none(), other.is_untyped_none()) {
(true, true) => {
let len = self.len() + other.len();
*self = ColumnBuilder::untyped_none();
self.append_nones(len);
return Ok(());
}
(true, false) => {
let len = self.len();
*self = ColumnBuilder::like(other, len + other.len());
self.append_nones(len);
self.optional = true;
}
(false, true) => {
self.append_nones(other.len());
return Ok(());
}
(false, false) => {}
}
let other_optional = other.is_nullable() || other.none_count() > 0;
if self.optional && other_optional && !self.same_type(other) {
return internal_err!("column type mismatch in Option extend");
}
self.append_view(other)?;
self.optional |= other_optional;
Ok(())
}
pub fn append_values(&mut self, other: &ColumnView) -> Result<()> {
let nones = other.none_count();
if nones == other.len() {
self.append_nones(other.len());
return Ok(());
}
self.append_view(other)?;
self.optional |= nones > 0;
Ok(())
}
fn append_nones(&mut self, count: usize) {
for _ in 0..count {
self.push_none();
}
}
fn same_type(&self, other: &ColumnView) -> bool {
let same_family =
matches!((&self.inner, &other.data), (TypedBuilder::Decimal(_), ViewData::Decimal(_)));
same_family || self.inner.get_type() == other.base_type()
}
fn is_untyped_none(&self) -> bool {
match &self.inner {
TypedBuilder::None(_) => true,
TypedBuilder::Any {
builder,
declared_type: None,
} if self.optional => match builder.validity_slice() {
Some(bits) => (0..builder.len()).all(|index| !bit_util::get_bit(bits, index)),
None => builder.is_empty(),
},
_ => false,
}
}
fn copy_details(&mut self, data: &ViewData) {
match (data, &mut self.inner) {
(
ViewData::Utf8 {
max_bytes: source,
..
},
TypedBuilder::Utf8 {
max_bytes: target,
..
},
)
| (
ViewData::Blob {
max_bytes: source,
..
},
TypedBuilder::Blob {
max_bytes: target,
..
},
) => *target = *source,
(
ViewData::DictionaryId {
dictionary_id: Some(source),
..
},
TypedBuilder::DictionaryId {
dictionary_id: target,
..
},
) => *target = Some(*source),
_ => {}
}
}
fn append_view(&mut self, other: &ColumnView) -> Result<()> {
match (&mut self.inner, &other.data) {
(TypedBuilder::Bool(l), ViewData::Bool(r)) => l.append_array(r),
(TypedBuilder::Int1(l), ViewData::Int1(r)) => l.append_array(r),
(TypedBuilder::Int2(l), ViewData::Int2(r)) => l.append_array(r),
(TypedBuilder::Int4(l), ViewData::Int4(r)) => l.append_array(r),
(TypedBuilder::Int8(l), ViewData::Int8(r)) => l.append_array(r),
(TypedBuilder::Int16(l), ViewData::Int16(r)) => l.append_array(r)?,
(TypedBuilder::Uint1(l), ViewData::Uint1(r)) => l.append_array(r),
(TypedBuilder::Uint2(l), ViewData::Uint2(r)) => l.append_array(r),
(TypedBuilder::Uint4(l), ViewData::Uint4(r)) => l.append_array(r),
(TypedBuilder::Uint8(l), ViewData::Uint8(r)) => l.append_array(r),
(TypedBuilder::Uint16(l), ViewData::Uint16(r)) => l.append_array(r)?,
(TypedBuilder::Float4(l), ViewData::Float4(r)) => l.append_array(r),
(TypedBuilder::Float8(l), ViewData::Float8(r)) => l.append_array(r),
(TypedBuilder::Date(l), ViewData::Date(r)) => l.append_array(r),
(TypedBuilder::DateTime(l), ViewData::DateTime(r)) => l.append_array(r),
(TypedBuilder::Time(l), ViewData::Time(r)) => l.append_array(r),
(TypedBuilder::Duration(l), ViewData::Duration(r)) => l.append_array(r),
(TypedBuilder::IdentityId(l), ViewData::IdentityId(r)) => append_fixed_array(l, r)?,
(TypedBuilder::Uuid4(l), ViewData::Uuid4(r)) => append_fixed_array(l, r)?,
(TypedBuilder::Uuid7(l), ViewData::Uuid7(r)) => append_fixed_array(l, r)?,
(
TypedBuilder::DictionaryId {
builder,
..
},
ViewData::DictionaryId {
container,
..
},
) => append_fixed_array(builder, container)?,
(
TypedBuilder::Utf8 {
builder,
..
},
ViewData::Utf8 {
container,
..
},
) => append_varlen(builder, container)?,
(
TypedBuilder::Blob {
builder,
..
},
ViewData::Blob {
container,
..
},
) => append_varlen(builder, container)?,
(TypedBuilder::Decimal(l), ViewData::Decimal(r)) => l.append_array(*r),
(
TypedBuilder::Any {
builder,
..
},
ViewData::Any {
container,
..
},
) => append_varlen(builder, container)?,
(
TypedBuilder::Digest {
builder,
inner: l_inner,
accuracy: l_accuracy,
},
ViewData::Digest {
container,
inner: r_inner,
accuracy: r_accuracy,
},
) if l_inner == r_inner && l_accuracy == r_accuracy => append_varlen(builder, container)?,
_ => return internal_err!("column type mismatch"),
}
Ok(())
}
pub fn len(&self) -> usize {
self.inner.len()
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn get_type(&self) -> ValueType {
match self.optional {
true => ValueType::Option(Box::new(self.inner.get_type())),
false => self.inner.get_type(),
}
}
pub fn finish(self, name: &str) -> (FieldRef, ArrayRef) {
let (mut field_type, array) = self.inner.finish();
if let Some(value_type) = field_type.value_type.take() {
field_type.value_type = Some(match self.optional || array.null_count() > 0 {
true => ValueType::Option(Box::new(value_type)),
false => value_type,
});
}
named(name, field_type, array)
}
}
impl TypedBuilder {
pub(crate) fn append_null(&mut self) {
match self {
TypedBuilder::Bool(b) => b.append_null(),
TypedBuilder::Int1(b) => b.append_null(),
TypedBuilder::Int2(b) => b.append_null(),
TypedBuilder::Int4(b) => b.append_null(),
TypedBuilder::Int8(b) => b.append_null(),
TypedBuilder::Int16(b) => b.append_null(),
TypedBuilder::Uint1(b) => b.append_null(),
TypedBuilder::Uint2(b) => b.append_null(),
TypedBuilder::Uint4(b) => b.append_null(),
TypedBuilder::Uint8(b) => b.append_null(),
TypedBuilder::Uint16(b) => b.append_null(),
TypedBuilder::Float4(b) => b.append_null(),
TypedBuilder::Float8(b) => b.append_null(),
TypedBuilder::Date(b) => b.append_null(),
TypedBuilder::DateTime(b) => b.append_null(),
TypedBuilder::Time(b) => b.append_null(),
TypedBuilder::Duration(b) => b.append_null(),
TypedBuilder::IdentityId(b) => b.append_null(),
TypedBuilder::Uuid4(b) => b.append_null(),
TypedBuilder::Uuid7(b) => b.append_null(),
TypedBuilder::DictionaryId {
builder,
..
} => builder.append_null(),
TypedBuilder::Utf8 {
builder,
..
} => builder.append_null(),
TypedBuilder::Blob {
builder,
..
}
| TypedBuilder::Any {
builder,
..
}
| TypedBuilder::Digest {
builder,
..
} => builder.append_null(),
TypedBuilder::Decimal(b) => b.append_null(),
TypedBuilder::None(b) => b.append_null(),
}
}
pub fn len(&self) -> usize {
match self {
TypedBuilder::Bool(b) => b.len(),
TypedBuilder::Int1(b) => b.len(),
TypedBuilder::Int2(b) => b.len(),
TypedBuilder::Int4(b) => b.len(),
TypedBuilder::Int8(b) => b.len(),
TypedBuilder::Int16(b) => b.len(),
TypedBuilder::Uint1(b) => b.len(),
TypedBuilder::Uint2(b) => b.len(),
TypedBuilder::Uint4(b) => b.len(),
TypedBuilder::Uint8(b) => b.len(),
TypedBuilder::Uint16(b) => b.len(),
TypedBuilder::Float4(b) => b.len(),
TypedBuilder::Float8(b) => b.len(),
TypedBuilder::Date(b) => b.len(),
TypedBuilder::DateTime(b) => b.len(),
TypedBuilder::Time(b) => b.len(),
TypedBuilder::Duration(b) => b.len(),
TypedBuilder::IdentityId(b) => b.len(),
TypedBuilder::Uuid4(b) => b.len(),
TypedBuilder::Uuid7(b) => b.len(),
TypedBuilder::DictionaryId {
builder,
..
} => builder.len(),
TypedBuilder::Utf8 {
builder,
..
} => builder.len(),
TypedBuilder::Blob {
builder,
..
}
| TypedBuilder::Any {
builder,
..
}
| TypedBuilder::Digest {
builder,
..
} => builder.len(),
TypedBuilder::Decimal(b) => b.len(),
TypedBuilder::None(b) => b.len(),
}
}
pub fn is_empty(&self) -> bool {
self.len() == 0
}
pub fn get_type(&self) -> ValueType {
match self {
TypedBuilder::Bool(_) => ValueType::Boolean,
TypedBuilder::Int1(_) => ValueType::Int1,
TypedBuilder::Int2(_) => ValueType::Int2,
TypedBuilder::Int4(_) => ValueType::Int4,
TypedBuilder::Int8(_) => ValueType::Int8,
TypedBuilder::Int16(_) => ValueType::Int16,
TypedBuilder::Uint1(_) => ValueType::Uint1,
TypedBuilder::Uint2(_) => ValueType::Uint2,
TypedBuilder::Uint4(_) => ValueType::Uint4,
TypedBuilder::Uint8(_) => ValueType::Uint8,
TypedBuilder::Uint16(_) => ValueType::Uint16,
TypedBuilder::Float4(_) => ValueType::Float4,
TypedBuilder::Float8(_) => ValueType::Float8,
TypedBuilder::Date(_) => ValueType::Date,
TypedBuilder::DateTime(_) => ValueType::DateTime,
TypedBuilder::Time(_) => ValueType::Time,
TypedBuilder::Duration(_) => ValueType::Duration,
TypedBuilder::IdentityId(_) => ValueType::IdentityId,
TypedBuilder::Uuid4(_) => ValueType::Uuid4,
TypedBuilder::Uuid7(_) => ValueType::Uuid7,
TypedBuilder::DictionaryId {
..
} => ValueType::DictionaryId,
TypedBuilder::Utf8 {
..
} => ValueType::Utf8,
TypedBuilder::Blob {
..
} => ValueType::Blob,
TypedBuilder::Decimal(b) => ValueType::decimal(b.precision(), b.scale()),
TypedBuilder::Any {
declared_type,
..
} => declared_type.clone().unwrap_or(ValueType::Any),
TypedBuilder::Digest {
inner,
accuracy,
..
} => ValueType::Digest {
inner: Box::new(inner.clone()),
accuracy: *accuracy,
},
TypedBuilder::None(_) => ValueType::Any,
}
}
fn finish(self) -> (FieldType, ArrayRef) {
let value_type = self.get_type();
let bare = |array: ArrayRef| {
(
FieldType {
value_type: Some(value_type.clone()),
..FieldType::default()
},
array,
)
};
match self {
TypedBuilder::Bool(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Int1(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Int2(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Int4(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Int8(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Int16(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uint1(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uint2(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uint4(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uint8(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uint16(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Float4(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Float8(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Date(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::DateTime(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Time(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Duration(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::IdentityId(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uuid4(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::Uuid7(mut b) => bare(Arc::new(b.finish())),
TypedBuilder::DictionaryId {
mut builder,
dictionary_id,
} => (
FieldType {
value_type: Some(value_type),
dictionary_id,
..FieldType::default()
},
Arc::new(builder.finish()),
),
TypedBuilder::Utf8 {
mut builder,
max_bytes,
} => (
FieldType {
value_type: Some(value_type),
max_bytes: limit(max_bytes),
..FieldType::default()
},
Arc::new(builder.finish()),
),
TypedBuilder::Blob {
mut builder,
max_bytes,
} => (
FieldType {
value_type: Some(value_type),
max_bytes: limit(max_bytes),
..FieldType::default()
},
Arc::new(builder.finish()),
),
TypedBuilder::Decimal(mut b) => bare(b.finish().into_array()),
TypedBuilder::Any {
mut builder,
declared_type,
} => (
FieldType {
value_type: Some(value_type),
declared_type,
..FieldType::default()
},
Arc::new(builder.finish()),
),
TypedBuilder::Digest {
mut builder,
..
} => bare(Arc::new(builder.finish())),
TypedBuilder::None(mut b) => (FieldType::default(), Arc::new(b.finish())),
}
}
fn from_view(data: &ViewData) -> TypedBuilder {
match data {
ViewData::Bool(a) => {
let mut builder = BooleanBuilder::with_capacity(a.len());
builder.append_array(a);
TypedBuilder::Bool(builder)
}
ViewData::Int1(a) => TypedBuilder::Int1(primitive_builder((*a).clone())),
ViewData::Int2(a) => TypedBuilder::Int2(primitive_builder((*a).clone())),
ViewData::Int4(a) => TypedBuilder::Int4(primitive_builder((*a).clone())),
ViewData::Int8(a) => TypedBuilder::Int8(primitive_builder((*a).clone())),
ViewData::Int16(a) => TypedBuilder::Int16(WideBuilder::from_array(a)),
ViewData::Uint1(a) => TypedBuilder::Uint1(primitive_builder((*a).clone())),
ViewData::Uint2(a) => TypedBuilder::Uint2(primitive_builder((*a).clone())),
ViewData::Uint4(a) => TypedBuilder::Uint4(primitive_builder((*a).clone())),
ViewData::Uint8(a) => TypedBuilder::Uint8(primitive_builder((*a).clone())),
ViewData::Uint16(a) => TypedBuilder::Uint16(WideBuilder::from_array(a)),
ViewData::Float4(a) => TypedBuilder::Float4(primitive_builder((*a).clone())),
ViewData::Float8(a) => TypedBuilder::Float8(primitive_builder((*a).clone())),
ViewData::Date(a) => TypedBuilder::Date(primitive_builder((*a).clone())),
ViewData::DateTime(a) => TypedBuilder::DateTime(primitive_builder((*a).clone())),
ViewData::Time(a) => TypedBuilder::Time(primitive_builder((*a).clone())),
ViewData::Duration(a) => TypedBuilder::Duration(primitive_builder((*a).clone())),
ViewData::IdentityId(a) => TypedBuilder::IdentityId(fixed_size_builder(a)),
ViewData::Uuid4(a) => TypedBuilder::Uuid4(fixed_size_builder(a)),
ViewData::Uuid7(a) => TypedBuilder::Uuid7(fixed_size_builder(a)),
ViewData::DictionaryId {
container,
dictionary_id,
} => TypedBuilder::DictionaryId {
builder: fixed_size_builder(container),
dictionary_id: *dictionary_id,
},
ViewData::Utf8 {
container,
max_bytes,
} => TypedBuilder::Utf8 {
builder: varlen_builder((*container).clone()),
max_bytes: *max_bytes,
},
ViewData::Blob {
container,
max_bytes,
} => TypedBuilder::Blob {
builder: varlen_builder((*container).clone()),
max_bytes: *max_bytes,
},
ViewData::Decimal(a) => {
TypedBuilder::Decimal(DecimalBuilder::from_array(DecimalArray::from(*a)))
}
ViewData::Any {
container,
declared_type,
} => TypedBuilder::Any {
builder: varlen_builder((*container).clone()),
declared_type: declared_type.clone(),
},
ViewData::Digest {
container,
inner,
accuracy,
} => TypedBuilder::Digest {
builder: varlen_builder((*container).clone()),
inner: inner.clone(),
accuracy: *accuracy,
},
ViewData::None {
array,
} => {
let mut builder = NullBuilder::new();
builder.append_nulls(array.len());
TypedBuilder::None(builder)
}
}
}
}
fn limit(max_bytes: MaxBytes) -> Option<MaxBytes> {
(max_bytes != MaxBytes::MAX).then_some(max_bytes)
}
pub(crate) fn primitive_builder<A>(array: PrimitiveArray<A>) -> PrimitiveBuilder<A>
where
A: ArrowPrimitiveType,
{
let data_type = array.data_type().clone();
match array.into_builder() {
Ok(builder) => builder.with_data_type(data_type),
Err(array) => {
let mut builder = PrimitiveBuilder::with_capacity(array.len());
builder.append_array(&array);
builder.with_data_type(data_type)
}
}
}
pub(crate) fn append_fixed(builder: &mut FixedSizeBinaryBuilder, row: &[u8]) {
if builder.append_value(row).is_err() {
panic!(
"a fixed size builder of {} byte rows can not take a row of {} bytes",
builder.finish_cloned().value_length(),
row.len()
);
}
}
fn fixed_size_builder(array: &FixedSizeBinaryArray) -> FixedSizeBinaryBuilder {
let mut builder = FixedSizeBinaryBuilder::with_capacity(array.len(), array.value_length());
append_fixed_array(&mut builder, array)
.expect("copying a fixed size array into a builder of its own width failed");
builder
}
pub(crate) fn append_fixed_array(builder: &mut FixedSizeBinaryBuilder, array: &FixedSizeBinaryArray) -> Result<()> {
builder.append_array(array).map_err(|e| internal_error!("fixed size append failed: {}", e))
}
pub(crate) fn varlen_builder<T>(array: GenericByteArray<T>) -> GenericByteBuilder<T>
where
T: ByteArrayType<Offset = i64>,
{
let array = if array.value_offsets()[0] == 0 {
match array.into_builder() {
Ok(builder) => return builder,
Err(array) => array,
}
} else {
array
};
let (data, _) = varlen_array::compact_parts(&array);
let mut builder = GenericByteBuilder::<T>::with_capacity(array.len(), data.len());
append_varlen(&mut builder, &array)
.expect("copying a shared or offset utf8 / blob array into a new builder failed");
builder
}
pub(crate) fn append_varlen<T>(builder: &mut GenericByteBuilder<T>, array: &GenericByteArray<T>) -> Result<()>
where
T: ByteArrayType<Offset = i64>,
{
builder.append_array(array).map_err(|e| internal_error!("utf8 / blob append failed: {}", e))
}
#[cfg(test)]
mod tests {
use std::str::FromStr;
use arrow_array::ArrayRef;
use arrow_buffer::{BooleanBuffer, NullBuffer};
use arrow_schema::FieldRef;
use reifydb_value::value::{
Value,
column_view::{ColumnView, ViewData},
constraint::{precision::Precision, scale::Scale},
decimal::Decimal,
value_type::ValueType,
};
use super::*;
use crate::value::column::{factory, nulls::with_nulls};
fn concat(parts: &[(FieldRef, ArrayRef)]) -> (FieldRef, ArrayRef) {
let (first, rest) = parts.split_first().unwrap();
let mut builder = ColumnBuilder::from_view(&ColumnView::try_from(first).unwrap());
for part in rest {
builder.extend(&ColumnView::try_from(part).unwrap()).unwrap();
}
builder.finish("c")
}
#[test]
fn concat_keeps_a_nullable_column_nullable_across_chunks() {
let first = with_nulls(factory::int4("c", [1, 2]), NullBuffer::new_valid(2)).unwrap();
let second = factory::int4("c", [3, 4]);
let joined = concat(&[first, second]);
let joined = ColumnView::try_from(&joined).unwrap();
assert_eq!(joined.len(), 4);
assert!(joined.is_nullable());
assert_eq!(joined.get_type(), ValueType::Option(Box::new(ValueType::Int4)));
assert_eq!(joined.get_value(3), Value::Int4(4));
}
#[test]
fn concat_carries_the_none_rows_of_every_chunk() {
let first = factory::int4_with_bitvec("c", [1, 0], BooleanBuffer::from(vec![true, false]));
let second = factory::int4_with_bitvec("c", [0, 4], BooleanBuffer::from(vec![false, true]));
let joined = concat(&[first, second]);
let joined = ColumnView::try_from(&joined).unwrap();
assert_eq!(joined.get_value(0), Value::Int4(1));
assert_eq!(joined.get_value(1), Value::none_of(ValueType::Int4));
assert_eq!(joined.get_value(2), Value::none_of(ValueType::Int4));
assert_eq!(joined.get_value(3), Value::Int4(4));
}
#[test]
fn concat_keeps_the_decimal_precision_and_scale() {
let first =
factory::decimal("c", Precision::new(9), Scale::new(2), [Decimal::from_str("1.25").unwrap()]);
let second =
factory::decimal("c", Precision::new(9), Scale::new(2), [Decimal::from_str("2.50").unwrap()]);
let joined = concat(&[first, second]);
let joined = ColumnView::try_from(&joined).unwrap();
let ViewData::Decimal(array) = &joined.data else {
panic!("expected a decimal column");
};
assert_eq!(array.precision(), Precision::new(9));
assert_eq!(array.scale(), Scale::new(2));
assert_eq!(joined.len(), 2);
}
}