use std::sync::Arc;
use arrow::array::{ArrayRef, MapArray, StructArray};
use arrow::buffer::{NullBuffer, OffsetBuffer};
use arrow::datatypes::{Field, Fields};
use snafu::ResultExt;
use crate::array_decoder::derive_present_vec;
use crate::column::Column;
use crate::encoding::integer::get_unsigned_int_decoder;
use crate::encoding::PrimitiveValueDecoder;
use crate::error::{ArrowSnafu, Result};
use crate::proto::stream::Kind;
use crate::stripe::Stripe;
use super::{array_decoder_factory, ArrayBatchDecoder, PresentDecoder};
pub struct MapArrayDecoder {
keys: Box<dyn ArrayBatchDecoder>,
values: Box<dyn ArrayBatchDecoder>,
present: Option<PresentDecoder>,
lengths: Box<dyn PrimitiveValueDecoder<i64> + Send>,
fields: Fields,
}
impl MapArrayDecoder {
pub fn new(
column: &Column,
keys_field: Arc<Field>,
values_field: Arc<Field>,
stripe: &Stripe,
) -> Result<Self> {
let present = PresentDecoder::from_stripe(stripe, column);
let keys_column = &column.children()[0];
let keys = array_decoder_factory(keys_column, keys_field.data_type(), stripe)?;
let values_column = &column.children()[1];
let values = array_decoder_factory(values_column, values_field.data_type(), stripe)?;
let reader = stripe.stream_map().get(column, Kind::Length);
let lengths = get_unsigned_int_decoder(reader, column.rle_version());
let fields = Fields::from(vec![keys_field, values_field]);
Ok(Self {
keys,
values,
present,
lengths,
fields,
})
}
}
impl ArrayBatchDecoder for MapArrayDecoder {
fn next_batch(
&mut self,
batch_size: usize,
parent_present: Option<&NullBuffer>,
) -> Result<ArrayRef> {
let present =
derive_present_vec(&mut self.present, parent_present, batch_size).transpose()?;
let mut lengths = vec![0; batch_size];
if let Some(present) = &present {
self.lengths.decode_spaced(&mut lengths, present)?;
} else {
self.lengths.decode(&mut lengths)?;
}
let total_length: i64 = lengths.iter().sum();
let keys_array = self.keys.next_batch(total_length as usize, None)?;
let values_array = self.values.next_batch(total_length as usize, None)?;
let entries =
StructArray::try_new(self.fields.clone(), vec![keys_array, values_array], None)
.context(ArrowSnafu)?;
let offsets = OffsetBuffer::from_lengths(lengths.into_iter().map(|l| l as usize));
let field = Arc::new(Field::new_struct("entries", self.fields.clone(), false));
let array =
MapArray::try_new(field, offsets, entries, present, false).context(ArrowSnafu)?;
let array = Arc::new(array);
Ok(array)
}
fn skip_values(&mut self, n: usize, parent_present: Option<&NullBuffer>) -> Result<()> {
use super::skip_present_and_get_non_null_count;
let non_null_count =
skip_present_and_get_non_null_count(&mut self.present, parent_present, n)?;
let mut lengths = vec![0; non_null_count];
self.lengths.decode(&mut lengths)?;
let total_length: i64 = lengths.iter().sum();
self.keys.skip_values(total_length as usize, None)?;
self.values.skip_values(total_length as usize, None)?;
Ok(())
}
}