use crate::types::bytes::ByteArrayNativeType;
use std::{any::Any, sync::Arc};
use crate::{
types::{BinaryType, ByteArrayType, LargeBinaryType, LargeUtf8Type, RunEndIndexType, Utf8Type},
ArrayRef, ArrowPrimitiveType, RunArray,
};
use super::{ArrayBuilder, GenericByteBuilder, PrimitiveBuilder};
use arrow_buffer::ArrowNativeType;
#[derive(Debug)]
pub struct GenericByteRunBuilder<R, V>
where
R: ArrowPrimitiveType,
V: ByteArrayType,
{
run_ends_builder: PrimitiveBuilder<R>,
values_builder: GenericByteBuilder<V>,
current_value: Vec<u8>,
has_current_value: bool,
current_run_end_index: usize,
prev_run_end_index: usize,
}
impl<R, V> Default for GenericByteRunBuilder<R, V>
where
R: ArrowPrimitiveType,
V: ByteArrayType,
{
fn default() -> Self {
Self::new()
}
}
impl<R, V> GenericByteRunBuilder<R, V>
where
R: ArrowPrimitiveType,
V: ByteArrayType,
{
pub fn new() -> Self {
Self {
run_ends_builder: PrimitiveBuilder::new(),
values_builder: GenericByteBuilder::<V>::new(),
current_value: Vec::new(),
has_current_value: false,
current_run_end_index: 0,
prev_run_end_index: 0,
}
}
pub fn with_capacity(capacity: usize, data_capacity: usize) -> Self {
Self {
run_ends_builder: PrimitiveBuilder::with_capacity(capacity),
values_builder: GenericByteBuilder::<V>::with_capacity(capacity, data_capacity),
current_value: Vec::new(),
has_current_value: false,
current_run_end_index: 0,
prev_run_end_index: 0,
}
}
}
impl<R, V> ArrayBuilder for GenericByteRunBuilder<R, V>
where
R: RunEndIndexType,
V: ByteArrayType,
{
fn as_any(&self) -> &dyn Any {
self
}
fn as_any_mut(&mut self) -> &mut dyn Any {
self
}
fn into_box_any(self: Box<Self>) -> Box<dyn Any> {
self
}
fn len(&self) -> usize {
self.current_run_end_index
}
fn finish(&mut self) -> ArrayRef {
Arc::new(self.finish())
}
fn finish_cloned(&self) -> ArrayRef {
Arc::new(self.finish_cloned())
}
}
impl<R, V> GenericByteRunBuilder<R, V>
where
R: RunEndIndexType,
V: ByteArrayType,
{
pub fn append_option(&mut self, input_value: Option<impl AsRef<V::Native>>) {
match input_value {
Some(value) => self.append_value(value),
None => self.append_null(),
}
}
pub fn append_value(&mut self, input_value: impl AsRef<V::Native>) {
let value: &[u8] = input_value.as_ref().as_ref();
if !self.has_current_value {
self.append_run_end();
self.current_value.extend_from_slice(value);
self.has_current_value = true;
} else if self.current_value.as_slice() != value {
self.append_run_end();
self.current_value.clear();
self.current_value.extend_from_slice(value);
}
self.current_run_end_index += 1;
}
pub fn append_null(&mut self) {
if self.has_current_value {
self.append_run_end();
self.current_value.clear();
self.has_current_value = false;
}
self.current_run_end_index += 1;
}
pub fn finish(&mut self) -> RunArray<R> {
self.append_run_end();
self.current_value.clear();
self.has_current_value = false;
self.current_run_end_index = 0;
self.prev_run_end_index = 0;
let run_ends_array = self.run_ends_builder.finish();
let values_array = self.values_builder.finish();
RunArray::<R>::try_new(&run_ends_array, &values_array).unwrap()
}
pub fn finish_cloned(&self) -> RunArray<R> {
let mut run_ends_array = self.run_ends_builder.finish_cloned();
let mut values_array = self.values_builder.finish_cloned();
if self.prev_run_end_index != self.current_run_end_index {
let mut run_end_builder = run_ends_array.into_builder().unwrap();
let mut values_builder = values_array.into_builder().unwrap();
self.append_run_end_with_builders(&mut run_end_builder, &mut values_builder);
run_ends_array = run_end_builder.finish();
values_array = values_builder.finish();
}
RunArray::<R>::try_new(&run_ends_array, &values_array).unwrap()
}
fn append_run_end(&mut self) {
if self.prev_run_end_index == self.current_run_end_index {
return;
}
let run_end_index = self.run_end_index_as_native();
self.run_ends_builder.append_value(run_end_index);
if self.has_current_value {
let slice = self.current_value.as_slice();
let native = unsafe {
V::Native::from_bytes_unchecked(slice)
};
self.values_builder.append_value(native);
} else {
self.values_builder.append_null();
}
self.prev_run_end_index = self.current_run_end_index;
}
fn append_run_end_with_builders(
&self,
run_ends_builder: &mut PrimitiveBuilder<R>,
values_builder: &mut GenericByteBuilder<V>,
) {
let run_end_index = self.run_end_index_as_native();
run_ends_builder.append_value(run_end_index);
if self.has_current_value {
let slice = self.current_value.as_slice();
let native = unsafe {
V::Native::from_bytes_unchecked(slice)
};
values_builder.append_value(native);
} else {
values_builder.append_null();
}
}
fn run_end_index_as_native(&self) -> R::Native {
R::Native::from_usize(self.current_run_end_index).unwrap_or_else(|| {
panic!(
"Cannot convert the value {} from `usize` to native form of arrow datatype {}",
self.current_run_end_index,
R::DATA_TYPE
)
})
}
}
impl<R, V, S> Extend<Option<S>> for GenericByteRunBuilder<R, V>
where
R: RunEndIndexType,
V: ByteArrayType,
S: AsRef<V::Native>,
{
fn extend<T: IntoIterator<Item = Option<S>>>(&mut self, iter: T) {
for elem in iter {
self.append_option(elem);
}
}
}
pub type StringRunBuilder<K> = GenericByteRunBuilder<K, Utf8Type>;
pub type LargeStringRunBuilder<K> = GenericByteRunBuilder<K, LargeUtf8Type>;
pub type BinaryRunBuilder<K> = GenericByteRunBuilder<K, BinaryType>;
pub type LargeBinaryRunBuilder<K> = GenericByteRunBuilder<K, LargeBinaryType>;
#[cfg(test)]
mod tests {
use super::*;
use crate::array::Array;
use crate::cast::AsArray;
use crate::types::{Int16Type, Int32Type};
use crate::GenericByteArray;
use crate::Int16RunArray;
fn test_bytes_run_builder<T>(values: Vec<&T::Native>)
where
T: ByteArrayType,
<T as ByteArrayType>::Native: PartialEq,
<T as ByteArrayType>::Native: AsRef<<T as ByteArrayType>::Native>,
{
let mut builder = GenericByteRunBuilder::<Int16Type, T>::new();
builder.append_value(values[0]);
builder.append_value(values[0]);
builder.append_value(values[0]);
builder.append_null();
builder.append_null();
builder.append_value(values[1]);
builder.append_value(values[1]);
builder.append_value(values[2]);
builder.append_value(values[2]);
builder.append_value(values[2]);
builder.append_value(values[2]);
let array = builder.finish();
assert_eq!(array.len(), 11);
assert_eq!(array.null_count(), 0);
assert_eq!(array.logical_null_count(), 2);
assert_eq!(array.run_ends().values(), &[3, 5, 7, 11]);
let av = array.values();
let ava: &GenericByteArray<T> = av.as_any().downcast_ref::<GenericByteArray<T>>().unwrap();
assert_eq!(*ava.value(0), *values[0]);
assert!(ava.is_null(1));
assert_eq!(*ava.value(2), *values[1]);
assert_eq!(*ava.value(3), *values[2]);
}
#[test]
fn test_string_run_builder() {
test_bytes_run_builder::<Utf8Type>(vec!["abc", "def", "ghi"]);
}
#[test]
fn test_string_run_builder_with_empty_strings() {
test_bytes_run_builder::<Utf8Type>(vec!["abc", "", "ghi"]);
}
#[test]
fn test_binary_run_builder() {
test_bytes_run_builder::<BinaryType>(vec![b"abc", b"def", b"ghi"]);
}
fn test_bytes_run_builder_finish_cloned<T>(values: Vec<&T::Native>)
where
T: ByteArrayType,
<T as ByteArrayType>::Native: PartialEq,
<T as ByteArrayType>::Native: AsRef<<T as ByteArrayType>::Native>,
{
let mut builder = GenericByteRunBuilder::<Int16Type, T>::new();
builder.append_value(values[0]);
builder.append_null();
builder.append_value(values[1]);
builder.append_value(values[1]);
builder.append_value(values[0]);
let mut array: Int16RunArray = builder.finish_cloned();
assert_eq!(array.len(), 5);
assert_eq!(array.null_count(), 0);
assert_eq!(array.logical_null_count(), 1);
assert_eq!(array.run_ends().values(), &[1, 2, 4, 5]);
let av = array.values();
let ava: &GenericByteArray<T> = av.as_any().downcast_ref::<GenericByteArray<T>>().unwrap();
assert_eq!(ava.value(0), values[0]);
assert!(ava.is_null(1));
assert_eq!(ava.value(2), values[1]);
assert_eq!(ava.value(3), values[0]);
builder.append_value(values[0]);
builder.append_value(values[0]);
builder.append_value(values[1]);
array = builder.finish();
assert_eq!(array.len(), 8);
assert_eq!(array.null_count(), 0);
assert_eq!(array.logical_null_count(), 1);
assert_eq!(array.run_ends().values(), &[1, 2, 4, 7, 8]);
let av2 = array.values();
let ava2: &GenericByteArray<T> =
av2.as_any().downcast_ref::<GenericByteArray<T>>().unwrap();
assert_eq!(ava2.value(0), values[0]);
assert!(ava2.is_null(1));
assert_eq!(ava2.value(2), values[1]);
assert_eq!(ava2.value(3), values[0]);
assert_eq!(ava2.value(4), values[1]);
}
#[test]
fn test_string_run_builder_finish_cloned() {
test_bytes_run_builder_finish_cloned::<Utf8Type>(vec!["abc", "def", "ghi"]);
}
#[test]
fn test_binary_run_builder_finish_cloned() {
test_bytes_run_builder_finish_cloned::<BinaryType>(vec![b"abc", b"def", b"ghi"]);
}
#[test]
fn test_extend() {
let mut builder = StringRunBuilder::<Int32Type>::new();
builder.extend(["a", "a", "a", "", "", "b", "b"].into_iter().map(Some));
builder.extend(["b", "cupcakes", "cupcakes"].into_iter().map(Some));
let array = builder.finish();
assert_eq!(array.len(), 10);
assert_eq!(array.run_ends().values(), &[3, 5, 8, 10]);
let str_array = array.values().as_string::<i32>();
assert_eq!(str_array.value(0), "a");
assert_eq!(str_array.value(1), "");
assert_eq!(str_array.value(2), "b");
assert_eq!(str_array.value(3), "cupcakes");
}
}