parquet/arrow/array_reader/

byte_view_array.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

use crate::arrow::array_reader::{ArrayReader, read_records, skip_records};
use crate::arrow::buffer::view_buffer::ViewBuffer;
use crate::arrow::decoder::{DeltaByteArrayDecoder, DictIndexDecoder};
use crate::arrow::record_reader::GenericRecordReader;
use crate::arrow::schema::parquet_to_arrow_field;
use crate::basic::{ConvertedType, Encoding};
use crate::column::page::PageIterator;
use crate::column::reader::decoder::ColumnValueDecoder;
use crate::data_type::Int32Type;
use crate::encodings::decoding::{Decoder, DeltaBitPackDecoder};
use crate::errors::{ParquetError, Result};
use crate::schema::types::ColumnDescPtr;
use crate::util::utf8::check_valid_utf8;
use arrow_array::{ArrayRef, builder::make_view};
use arrow_buffer::Buffer;
use arrow_data::ByteView;
use arrow_schema::DataType as ArrowType;
use bytes::Bytes;
use std::any::Any;

/// Returns an [`ArrayReader`] that decodes the provided byte array column to view types.
///
/// `batch_size` is a hint used to pre-allocate internal buffers and avoid
/// reallocations when reading the first batch. It is not a hard capacity
/// contract: with selective null padding (`padding_threshold`), value buffers
/// are sized from the number of values actually decoded.
pub fn make_byte_view_array_reader(
    pages: Box<dyn PageIterator>,
    column_desc: ColumnDescPtr,
    arrow_type: Option<ArrowType>,
    batch_size: usize,
    padding_threshold: Option<i16>,
) -> Result<Box<dyn ArrayReader>> {
    // Check if Arrow type is specified, else create it from Parquet type
    let data_type = match arrow_type {
        Some(t) => t,
        None => match parquet_to_arrow_field(column_desc.as_ref())?.data_type() {
            ArrowType::Utf8 | ArrowType::Utf8View => ArrowType::Utf8View,
            _ => ArrowType::BinaryView,
        },
    };

    match data_type {
        ArrowType::BinaryView | ArrowType::Utf8View => {
            let mut reader = GenericRecordReader::new(column_desc, batch_size);
            if let Some(threshold) = padding_threshold {
                reader.set_padding_threshold(threshold);
            }
            Ok(Box::new(ByteViewArrayReader::new(pages, data_type, reader)))
        }

        _ => Err(general_err!(
            "invalid data type for byte array reader read to view type - {}",
            data_type
        )),
    }
}

/// An [`ArrayReader`] for variable length byte arrays
struct ByteViewArrayReader {
    data_type: ArrowType,
    pages: Box<dyn PageIterator>,
    def_levels_buffer: Option<Vec<i16>>,
    rep_levels_buffer: Option<Vec<i16>>,
    record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
}

impl ByteViewArrayReader {
    fn new(
        pages: Box<dyn PageIterator>,
        data_type: ArrowType,
        record_reader: GenericRecordReader<ViewBuffer, ByteViewArrayColumnValueDecoder>,
    ) -> Self {
        Self {
            data_type,
            pages,
            def_levels_buffer: None,
            rep_levels_buffer: None,
            record_reader,
        }
    }
}

impl ArrayReader for ByteViewArrayReader {
    fn as_any(&self) -> &dyn Any {
        self
    }

    fn get_data_type(&self) -> &ArrowType {
        &self.data_type
    }

    fn read_records(&mut self, batch_size: usize) -> Result<usize> {
        read_records(&mut self.record_reader, self.pages.as_mut(), batch_size)
    }

    fn consume_batch(&mut self) -> Result<ArrayRef> {
        let buffer = self.record_reader.consume_record_data();
        let null_buffer = self.record_reader.consume_compact_bitmap();
        self.def_levels_buffer = self.record_reader.consume_def_levels();
        self.rep_levels_buffer = self.record_reader.consume_rep_levels();
        self.record_reader.reset();

        let array = buffer.into_array(null_buffer, &self.data_type);

        Ok(array)
    }

    fn skip_records(&mut self, num_records: usize) -> Result<usize> {
        skip_records(&mut self.record_reader, self.pages.as_mut(), num_records)
    }

    fn get_def_levels(&self) -> Option<&[i16]> {
        self.def_levels_buffer.as_deref()
    }

    fn get_rep_levels(&self) -> Option<&[i16]> {
        self.rep_levels_buffer.as_deref()
    }

    fn max_def_level(&self) -> i16 {
        self.record_reader.max_def_level()
    }
}

/// A [`ColumnValueDecoder`] for variable length byte arrays
struct ByteViewArrayColumnValueDecoder {
    dict: Option<ViewBuffer>,
    decoder: Option<ByteViewArrayDecoder>,
    validate_utf8: bool,
}

impl ColumnValueDecoder for ByteViewArrayColumnValueDecoder {
    type Buffer = ViewBuffer;

    fn new(desc: &ColumnDescPtr) -> Self {
        let validate_utf8 = desc.converted_type() == ConvertedType::UTF8;
        Self {
            dict: None,
            decoder: None,
            validate_utf8,
        }
    }

    fn set_dict(
        &mut self,
        buf: Bytes,
        num_values: u32,
        encoding: Encoding,
        _is_sorted: bool,
    ) -> Result<()> {
        if !matches!(
            encoding,
            Encoding::PLAIN | Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY
        ) {
            return Err(nyi_err!(
                "Invalid/Unsupported encoding type for dictionary: {}",
                encoding
            ));
        }

        let num_values = num_values as usize;
        let mut buffer = ViewBuffer::with_capacity(num_values);
        let mut decoder =
            ByteViewArrayDecoderPlain::new(buf, num_values, Some(num_values), self.validate_utf8);
        decoder.read(&mut buffer, usize::MAX)?;
        self.dict = Some(buffer);
        Ok(())
    }

    fn set_data(
        &mut self,
        encoding: Encoding,
        data: Bytes,
        num_levels: usize,
        num_values: Option<usize>,
    ) -> Result<()> {
        self.decoder = Some(ByteViewArrayDecoder::new(
            encoding,
            data,
            num_levels,
            num_values,
            self.validate_utf8,
        )?);
        Ok(())
    }

    fn read(&mut self, out: &mut Self::Buffer, num_values: usize) -> Result<usize> {
        let decoder = self
            .decoder
            .as_mut()
            .ok_or_else(|| general_err!("no decoder set"))?;

        decoder.read(out, num_values, self.dict.as_ref())
    }

    fn skip_values(&mut self, num_values: usize) -> Result<usize> {
        let decoder = self
            .decoder
            .as_mut()
            .ok_or_else(|| general_err!("no decoder set"))?;

        decoder.skip(num_values, self.dict.as_ref())
    }
}

/// A generic decoder from uncompressed parquet value data to [`ViewBuffer`]
pub enum ByteViewArrayDecoder {
    Plain(ByteViewArrayDecoderPlain),
    Dictionary(ByteViewArrayDecoderDictionary),
    DeltaLength(ByteViewArrayDecoderDeltaLength),
    DeltaByteArray(ByteViewArrayDecoderDelta),
}

impl ByteViewArrayDecoder {
    pub fn new(
        encoding: Encoding,
        data: Bytes,
        num_levels: usize,
        num_values: Option<usize>,
        validate_utf8: bool,
    ) -> Result<Self> {
        let decoder = match encoding {
            Encoding::PLAIN => ByteViewArrayDecoder::Plain(ByteViewArrayDecoderPlain::new(
                data,
                num_levels,
                num_values,
                validate_utf8,
            )),
            Encoding::RLE_DICTIONARY | Encoding::PLAIN_DICTIONARY => {
                ByteViewArrayDecoder::Dictionary(ByteViewArrayDecoderDictionary::new(
                    data, num_levels, num_values,
                )?)
            }
            Encoding::DELTA_LENGTH_BYTE_ARRAY => ByteViewArrayDecoder::DeltaLength(
                ByteViewArrayDecoderDeltaLength::new(data, validate_utf8)?,
            ),
            Encoding::DELTA_BYTE_ARRAY => ByteViewArrayDecoder::DeltaByteArray(
                ByteViewArrayDecoderDelta::new(data, validate_utf8)?,
            ),
            _ => {
                return Err(general_err!(
                    "unsupported encoding for byte array: {}",
                    encoding
                ));
            }
        };

        Ok(decoder)
    }

    /// Read up to `len` values to `out` with the optional dictionary
    pub fn read(
        &mut self,
        out: &mut ViewBuffer,
        len: usize,
        dict: Option<&ViewBuffer>,
    ) -> Result<usize> {
        match self {
            ByteViewArrayDecoder::Plain(d) => d.read(out, len),
            ByteViewArrayDecoder::Dictionary(d) => {
                let dict = dict
                    .ok_or_else(|| general_err!("dictionary required for dictionary encoding"))?;
                d.read(out, dict, len)
            }
            ByteViewArrayDecoder::DeltaLength(d) => d.read(out, len),
            ByteViewArrayDecoder::DeltaByteArray(d) => d.read(out, len),
        }
    }

    /// Skip `len` values
    pub fn skip(&mut self, len: usize, dict: Option<&ViewBuffer>) -> Result<usize> {
        match self {
            ByteViewArrayDecoder::Plain(d) => d.skip(len),
            ByteViewArrayDecoder::Dictionary(d) => {
                let dict = dict
                    .ok_or_else(|| general_err!("dictionary required for dictionary encoding"))?;
                d.skip(dict, len)
            }
            ByteViewArrayDecoder::DeltaLength(d) => d.skip(len),
            ByteViewArrayDecoder::DeltaByteArray(d) => d.skip(len),
        }
    }
}

/// Decoder from [`Encoding::PLAIN`] data to [`ViewBuffer`]
pub struct ByteViewArrayDecoderPlain {
    buf: Buffer,
    offset: usize,

    validate_utf8: bool,

    /// This is a maximum as the null count is not always known, e.g. value data from
    /// a v1 data page
    max_remaining_values: usize,
}

impl ByteViewArrayDecoderPlain {
    pub fn new(
        buf: Bytes,
        num_levels: usize,
        num_values: Option<usize>,
        validate_utf8: bool,
    ) -> Self {
        Self {
            buf: Buffer::from(buf),
            offset: 0,
            max_remaining_values: num_values.unwrap_or(num_levels),
            validate_utf8,
        }
    }

    pub fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
        if self.validate_utf8 {
            self.read_impl::<true>(output, len)
        } else {
            self.read_impl::<false>(output, len)
        }
    }

    fn read_impl<const VALIDATE_UTF8: bool>(
        &mut self,
        output: &mut ViewBuffer,
        len: usize,
    ) -> Result<usize> {
        // avoid creating a new buffer if the last buffer is the same as the current buffer
        // This is especially useful when row-level filtering is applied, where we call lots of small `read` over the same buffer.
        let block_id = {
            if output.buffers.last().is_some_and(|x| x.ptr_eq(&self.buf)) {
                output.buffers.len() as u32 - 1
            } else {
                output.append_block(self.buf.clone())
            }
        };

        let to_read = len.min(self.max_remaining_values);

        let buf: &[u8] = self.buf.as_ref();
        let buf_len = buf.len();
        let mut end_offset = self.offset;
        let mut utf8_validation_begin = end_offset;

        output.views.reserve(to_read);

        // Safety: we reserved enough space in output.views
        // and we will only write up to to_read views / track how many views we wrote.
        // Ideally, we would use `Vec::extend` here, but this generates sub-optimal code.
        let views_ptr = output.views.as_mut_ptr().wrapping_add(output.views.len());
        for i in 0..to_read {
            let start_offset = end_offset + 4;

            if start_offset > buf_len {
                return Err(ParquetError::EOF("eof decoding byte array".into()));
            }

            // Safety: we have checked that start_offset <= buf_len
            let len = u32::from_le_bytes(
                unsafe { buf.get_unchecked(end_offset..start_offset) }
                    .try_into()
                    .unwrap(),
            );

            end_offset = start_offset + len as usize;

            if end_offset > buf_len {
                return Err(ParquetError::EOF("eof decoding byte array".into()));
            }

            if VALIDATE_UTF8 {
                // It seems you are trying to understand what's going on here, take a breath and be patient.
                // Utf-8 validation is a non-trivial task, here are some background facts:
                // (1) Validating one 2048-byte string is much faster than validating 128 of 16-byte string.
                //     As shown in https://github.com/apache/arrow-rs/pull/6009#issuecomment-2211174229
                //     Potentially because the SIMD operations favor longer strings.
                // (2) Practical strings are short, 99% of strings are smaller than 100 bytes, as shown in paper:
                //     https://www.vldb.org/pvldb/vol17/p148-zeng.pdf, Figure 5f.
                // (3) Parquet plain encoding makes utf-8 validation harder,
                //     because it stores the length of each string right before the string.
                //     This means naive utf-8 validation will be slow, because the validation need to skip the length bytes.
                //     I.e., the validation cannot validate the buffer in one pass, but instead, validate strings chunk by chunk.
                //
                // Given the above observations, the goal is to do batch validation as much as possible.
                // The key idea is that if the length is smaller than 128 (99% of the case), then the length bytes are valid utf-8, as reasoned below:
                // If the length is smaller than 128, its 4-byte encoding are [0, 0, 0, len].
                // Each of the byte is a valid ASCII character, so they are valid utf-8.
                // Since they are all smaller than 128, the won't break a utf-8 code point (won't mess with later bytes).
                //
                // The implementation keeps a water mark `utf8_validation_begin` to track the beginning of the buffer that is not validated.
                // If the length is smaller than 128, then we continue to next string.
                // If the length is larger than 128, then we validate the buffer before the length bytes, and move the water mark to the beginning of next string.
                if len >= 128 {
                    // unfortunately, the len bytes may not be valid utf8, we need to wrap up and validate everything before it.
                    check_valid_utf8(unsafe {
                        buf.get_unchecked(utf8_validation_begin..start_offset - 4)
                    })?;
                    // move the cursor to skip the len bytes.
                    utf8_validation_begin = start_offset;
                }
            }

            let view = make_view(
                unsafe { buf.get_unchecked(start_offset..end_offset) },
                block_id,
                start_offset as u32,
            );
            // Safety: views_ptr is valid for writes, and we have reserved enough space.
            unsafe {
                views_ptr.add(i).write(view);
            }
        }

        // Safety: we have written `to_read` views to `views_ptr`
        unsafe {
            output.views.set_len(output.views.len() + to_read);
        }
        if VALIDATE_UTF8 {
            // validate values from the previously validated location up to (but not including)
            // the length of this string
            check_valid_utf8(unsafe { buf.get_unchecked(utf8_validation_begin..end_offset) })?;
        }

        self.offset = end_offset;
        self.max_remaining_values -= to_read;

        Ok(to_read)
    }

    pub fn skip(&mut self, to_skip: usize) -> Result<usize> {
        let to_skip = to_skip.min(self.max_remaining_values);
        let mut skip = 0;
        let buf: &[u8] = self.buf.as_ref();

        while self.offset < self.buf.len() && skip != to_skip {
            if self.offset + 4 > buf.len() {
                return Err(ParquetError::EOF("eof decoding byte array".into()));
            }
            let len_bytes: [u8; 4] = buf[self.offset..self.offset + 4].try_into().unwrap();
            let len = u32::from_le_bytes(len_bytes) as usize;
            skip += 1;
            self.offset = self.offset + 4 + len;
        }
        self.max_remaining_values -= skip;
        Ok(skip)
    }
}

pub struct ByteViewArrayDecoderDictionary {
    decoder: DictIndexDecoder,
}

impl ByteViewArrayDecoderDictionary {
    fn new(data: Bytes, num_levels: usize, num_values: Option<usize>) -> Result<Self> {
        Ok(Self {
            decoder: DictIndexDecoder::new(data, num_levels, num_values)?,
        })
    }

    /// Reads the next indexes from self.decoder
    /// the indexes are assumed to be indexes into `dict`
    /// the output values are written to output
    ///
    /// Assumptions / Optimization
    /// This function checks if dict.buffers() are the last buffers in `output`, and if so
    /// reuses the dictionary page buffers directly without copying data
    ///
    /// If the dictionary is empty, the buffer contains empty view.
    fn read(&mut self, output: &mut ViewBuffer, dict: &ViewBuffer, len: usize) -> Result<usize> {
        if dict.is_empty() || len == 0 {
            return Ok(0);
        }

        // Check if the last few buffer of `output`` are the same as the `dict` buffer
        // This is to avoid creating a new buffers if the same dictionary is used for multiple `read`
        let need_to_create_new_buffer = {
            if output.buffers.len() >= dict.buffers.len() {
                let offset = output.buffers.len() - dict.buffers.len();
                output.buffers[offset..]
                    .iter()
                    .zip(dict.buffers.iter())
                    .any(|(a, b)| !a.ptr_eq(b))
            } else {
                true
            }
        };

        if need_to_create_new_buffer {
            for b in &dict.buffers {
                output.buffers.push(b.clone());
            }
        }

        // Calculate the offset of the dictionary buffers in the output buffers
        // For example if the 2nd buffer in the dictionary is the 5th buffer in the output buffers,
        // then the base_buffer_idx is 5 - 2 = 3
        let base_buffer_idx = output.buffers.len() as u32 - dict.buffers.len() as u32;

        // Pre-reserve output capacity to avoid per-chunk reallocation in extend
        output.views.reserve(len);

        let mut error = None;
        let read = self.decoder.read(len, |keys| {
            if base_buffer_idx == 0 {
                // the dictionary buffers are the last buffers in output, we can directly use the views
                output
                    .views
                    .extend(keys.iter().map(|k| match dict.views.get(*k as usize) {
                        Some(&view) => view,
                        None => {
                            if error.is_none() {
                                error = Some(general_err!("invalid key={} for dictionary", *k));
                            }
                            0
                        }
                    }));
            } else {
                output
                    .views
                    .extend(keys.iter().map(|k| match dict.views.get(*k as usize) {
                        Some(&view) => {
                            let len = view as u32;
                            if len <= 12 {
                                view
                            } else {
                                let mut view = ByteView::from(view);
                                view.buffer_index += base_buffer_idx;
                                view.into()
                            }
                        }
                        None => {
                            if error.is_none() {
                                error = Some(general_err!("invalid key={} for dictionary", *k));
                            }
                            0
                        }
                    }));
            }
            Ok(())
        })?;
        if let Some(e) = error {
            return Err(e);
        }
        Ok(read)
    }

    fn skip(&mut self, dict: &ViewBuffer, to_skip: usize) -> Result<usize> {
        if dict.is_empty() {
            return Ok(0);
        }
        self.decoder.skip(to_skip)
    }
}

/// Decoder from [`Encoding::DELTA_LENGTH_BYTE_ARRAY`] data to [`ViewBuffer`]
pub struct ByteViewArrayDecoderDeltaLength {
    lengths: Vec<i32>,
    data: Bytes,
    length_offset: usize,
    data_offset: usize,
    validate_utf8: bool,
}

impl ByteViewArrayDecoderDeltaLength {
    fn new(data: Bytes, validate_utf8: bool) -> Result<Self> {
        let mut len_decoder = DeltaBitPackDecoder::<Int32Type>::new();
        len_decoder.set_data(data.clone(), 0)?;
        let values = len_decoder.values_left();

        let mut lengths = vec![0; values];
        len_decoder.get(&mut lengths)?;

        let mut total_bytes = 0;

        for l in &lengths {
            if *l < 0 {
                return Err(ParquetError::General(
                    "negative delta length byte array length".to_string(),
                ));
            }
            total_bytes += *l as usize;
        }

        if total_bytes + len_decoder.get_offset() > data.len() {
            return Err(ParquetError::General(
                "Insufficient delta length byte array bytes".to_string(),
            ));
        }

        Ok(Self {
            lengths,
            data,
            validate_utf8,
            length_offset: 0,
            data_offset: len_decoder.get_offset(),
        })
    }

    fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
        let to_read = len.min(self.lengths.len() - self.length_offset);
        output.views.reserve(to_read);

        let src_lengths = &self.lengths[self.length_offset..self.length_offset + to_read];

        // Zero copy convert `bytes::Bytes` into `arrow_buffer::Buffer`
        let bytes = Buffer::from(self.data.clone());
        let block_id = output.append_block(bytes);

        let mut current_offset = self.data_offset;
        let initial_offset = current_offset;

        output.views.extend(src_lengths.iter().map(|length| {
            let len = *length as u32;
            let start_offset = current_offset;
            current_offset += len as usize;
            // # Safety
            // The length and offset are guaranteed valid by the entry check in `new`
            make_view(
                &self.data[start_offset..start_offset + len as usize],
                block_id,
                start_offset as u32,
            )
        }));

        // Delta length encoding has continuous strings, we can validate utf8 in one go
        if self.validate_utf8 {
            check_valid_utf8(&self.data[initial_offset..current_offset])?;
        }

        self.data_offset = current_offset;
        self.length_offset += to_read;

        Ok(to_read)
    }

    fn skip(&mut self, to_skip: usize) -> Result<usize> {
        let remain_values = self.lengths.len() - self.length_offset;
        let to_skip = remain_values.min(to_skip);

        let src_lengths = &self.lengths[self.length_offset..self.length_offset + to_skip];
        let total_bytes: usize = src_lengths.iter().map(|x| *x as usize).sum();

        self.data_offset += total_bytes;
        self.length_offset += to_skip;
        Ok(to_skip)
    }
}

/// Decoder from [`Encoding::DELTA_BYTE_ARRAY`] to [`ViewBuffer`]
pub struct ByteViewArrayDecoderDelta {
    decoder: DeltaByteArrayDecoder,
    validate_utf8: bool,
}

impl ByteViewArrayDecoderDelta {
    fn new(data: Bytes, validate_utf8: bool) -> Result<Self> {
        Ok(Self {
            decoder: DeltaByteArrayDecoder::new(data)?,
            validate_utf8,
        })
    }

    // Unlike other encodings, we need to copy the data.
    //
    //  DeltaByteArray data is stored using shared prefixes/suffixes,
    // which results in potentially non-contiguous
    // strings, while Arrow encodings require contiguous strings
    //
    // <https://parquet.apache.org/docs/file-format/data-pages/encodings/#delta-strings-delta_byte_array--7>

    fn read(&mut self, output: &mut ViewBuffer, len: usize) -> Result<usize> {
        let to_reserve = len.min(self.decoder.remaining());
        output.views.reserve(to_reserve);

        // array buffer only have long strings
        let mut array_buffer: Vec<u8> = Vec::with_capacity(4096);

        let buffer_id = output.buffers.len() as u32;

        // Use unsafe ptr writes instead of per-element push to avoid
        // repeated length checks. Safety: we reserved enough space above.
        let views_ptr = output.views.as_mut_ptr();
        let initial_len = output.views.len();
        let mut write_count = 0;

        let read = if !self.validate_utf8 {
            self.decoder.read(len, |bytes| {
                let offset = array_buffer.len();
                let view = make_view(bytes, buffer_id, offset as u32);
                if bytes.len() > 12 {
                    // only copy the data to buffer if the string can not be inlined.
                    array_buffer.extend_from_slice(bytes);
                }

                // Safety: views_ptr is valid for writes, we reserved enough space,
                // and write_count < to_reserve.
                unsafe {
                    views_ptr.add(initial_len + write_count).write(view);
                }
                write_count += 1;
                Ok(())
            })?
        } else {
            // utf8 validation buffer has only short strings. These short
            // strings are inlined into the views but we copy them into a
            // contiguous buffer to accelerate validation.
            let mut utf8_validation_buffer = Vec::with_capacity(4096);

            let v = self.decoder.read(len, |bytes| {
                let offset = array_buffer.len();
                let view = make_view(bytes, buffer_id, offset as u32);
                if bytes.len() > 12 {
                    // only copy the data to buffer if the string can not be inlined.
                    array_buffer.extend_from_slice(bytes);
                } else {
                    utf8_validation_buffer.extend_from_slice(bytes);
                }

                // Safety: views_ptr is valid for writes, we reserved enough space,
                // and write_count < to_reserve. Utf-8 validation is done later.
                unsafe {
                    views_ptr.add(initial_len + write_count).write(view);
                }
                write_count += 1;
                Ok(())
            })?;
            check_valid_utf8(&array_buffer)?;
            check_valid_utf8(&utf8_validation_buffer)?;
            v
        };

        // Safety: we wrote exactly `read` views via ptr writes above
        unsafe {
            output.views.set_len(initial_len + read);
        }

        let actual_block_id = output.append_block(Buffer::from_vec(array_buffer));
        assert_eq!(actual_block_id, buffer_id);
        Ok(read)
    }

    fn skip(&mut self, to_skip: usize) -> Result<usize> {
        self.decoder.skip(to_skip)
    }
}

#[cfg(test)]
mod tests {
    use arrow_array::StringViewArray;
    use arrow_buffer::Buffer;

    use crate::{
        arrow::{
            array_reader::test_util::{byte_array_all_encodings, encode_byte_array, utf8_column},
            buffer::view_buffer::ViewBuffer,
            record_reader::buffer::ValuesBuffer,
        },
        basic::Encoding,
        column::reader::decoder::ColumnValueDecoder,
        data_type::ByteArray,
    };

    use super::*;

    #[test]
    fn test_byte_array_string_view_decoder() {
        let (pages, encoded_dictionary) =
            byte_array_all_encodings(vec!["hello", "world", "large payload over 12 bytes", "b"]);

        let column_desc = utf8_column();
        let mut decoder = ByteViewArrayColumnValueDecoder::new(&column_desc);

        decoder
            .set_dict(encoded_dictionary, 4, Encoding::RLE_DICTIONARY, false)
            .unwrap();

        for (encoding, page) in pages {
            let mut output = ViewBuffer::with_capacity(0);
            decoder.set_data(encoding, page, 4, Some(4)).unwrap();

            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
            assert_eq!(decoder.read(&mut output, 1).unwrap(), 1);
            assert_eq!(decoder.read(&mut output, 2).unwrap(), 2);
            assert_eq!(decoder.read(&mut output, 4).unwrap(), 0);

            assert_eq!(output.views.len(), 4);

            let valid = [false, false, true, true, false, true, true, false, false];
            let valid_buffer = Buffer::from_iter(valid.iter().copied());

            output
                .pad_nulls(0, 4, valid.len(), valid_buffer.as_slice())
                .unwrap();
            let array = output.into_array(Some(valid_buffer), &ArrowType::Utf8View);
            let strings = array.as_any().downcast_ref::<StringViewArray>().unwrap();

            assert_eq!(
                strings.iter().collect::<Vec<_>>(),
                vec![
                    None,
                    None,
                    Some("hello"),
                    Some("world"),
                    None,
                    Some("large payload over 12 bytes"),
                    Some("b"),
                    None,
                    None,
                ]
            );
        }
    }

    #[test]
    fn test_byte_view_array_plain_decoder_reuse_buffer() {
        let byte_array = vec!["hello", "world", "large payload over 12 bytes", "b"];
        let byte_array: Vec<ByteArray> = byte_array.into_iter().map(|x| x.into()).collect();
        let pages = encode_byte_array(Encoding::PLAIN, &byte_array);

        let column_desc = utf8_column();
        let mut decoder = ByteViewArrayColumnValueDecoder::new(&column_desc);

        let mut view_buffer = ViewBuffer::with_capacity(0);
        decoder.set_data(Encoding::PLAIN, pages, 4, None).unwrap();
        decoder.read(&mut view_buffer, 1).unwrap();
        decoder.read(&mut view_buffer, 1).unwrap();
        assert_eq!(view_buffer.buffers.len(), 1);

        decoder.read(&mut view_buffer, 1).unwrap();
        assert_eq!(view_buffer.buffers.len(), 1);
    }
}