parquet/column/writer/

mod.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.

//! Contains column writer API.

use bytes::Bytes;
use half::f16;

use crate::bloom_filter::Sbbf;
use crate::file::page_index::column_index::ColumnIndexMetaData;
use crate::file::page_index::offset_index::OffsetIndexMetaData;
use std::cmp::Ordering;
use std::collections::{BTreeSet, VecDeque};
use std::str;

use crate::basic::{
    BoundaryOrder, Compression, ConvertedType, Encoding, EncodingMask, LogicalType, PageType,
    SortOrder, Type,
};
use crate::column::page::{CompressedPage, Page, PageWriteSpec, PageWriter};
use crate::column::writer::encoder::{ColumnValueEncoder, ColumnValueEncoderImpl, ColumnValues};
use crate::compression::{Codec, CodecOptionsBuilder, create_codec};
use crate::data_type::private::ParquetValueType;
use crate::data_type::*;
use crate::encodings::levels::LevelEncoder;
#[cfg(feature = "encryption")]
use crate::encryption::encrypt::get_column_crypto_metadata;
use crate::errors::{ParquetError, Result};
use crate::file::metadata::{
    ColumnChunkMetaData, ColumnChunkMetaDataBuilder, ColumnIndexBuilder, LevelHistogram,
    OffsetIndexBuilder, PageEncodingStats,
};
use crate::file::properties::{
    EnabledStatistics, ResolvedColumnProperties, WriterProperties, WriterPropertiesPtr,
    WriterVersion,
};
use crate::file::statistics::{Statistics, ValueStatistics};
use crate::schema::types::{BasicTypeInfo, ColumnDescPtr, ColumnDescriptor};

mod byte_budget_chunker;
pub(crate) mod encoder;

use byte_budget_chunker::{ByteBudgetChunker, SubBatchStrategy};

macro_rules! downcast_writer {
    ($e:expr, $i:ident, $b:expr) => {
        match $e {
            Self::BoolColumnWriter($i) => $b,
            Self::Int32ColumnWriter($i) => $b,
            Self::Int64ColumnWriter($i) => $b,
            Self::Int96ColumnWriter($i) => $b,
            Self::FloatColumnWriter($i) => $b,
            Self::DoubleColumnWriter($i) => $b,
            Self::ByteArrayColumnWriter($i) => $b,
            Self::FixedLenByteArrayColumnWriter($i) => $b,
        }
    };
}

/// Column writer for a Parquet type.
///
/// See [`get_column_writer`] to create instances of this type
pub enum ColumnWriter<'a> {
    /// Column writer for boolean type
    BoolColumnWriter(ColumnWriterImpl<'a, BoolType>),
    /// Column writer for int32 type
    Int32ColumnWriter(ColumnWriterImpl<'a, Int32Type>),
    /// Column writer for int64 type
    Int64ColumnWriter(ColumnWriterImpl<'a, Int64Type>),
    /// Column writer for int96 (timestamp) type
    Int96ColumnWriter(ColumnWriterImpl<'a, Int96Type>),
    /// Column writer for float type
    FloatColumnWriter(ColumnWriterImpl<'a, FloatType>),
    /// Column writer for double type
    DoubleColumnWriter(ColumnWriterImpl<'a, DoubleType>),
    /// Column writer for byte array type
    ByteArrayColumnWriter(ColumnWriterImpl<'a, ByteArrayType>),
    /// Column writer for fixed length byte array type
    FixedLenByteArrayColumnWriter(ColumnWriterImpl<'a, FixedLenByteArrayType>),
}

impl ColumnWriter<'_> {
    /// Returns the estimated total memory usage
    #[cfg(feature = "arrow")]
    pub(crate) fn memory_size(&self) -> usize {
        downcast_writer!(self, typed, typed.memory_size())
    }

    /// Returns the estimated total encoded bytes for this column writer
    #[cfg(feature = "arrow")]
    pub(crate) fn get_estimated_total_bytes(&self) -> u64 {
        downcast_writer!(self, typed, typed.get_estimated_total_bytes())
    }

    /// Finalize the currently buffered values as a data page.
    ///
    /// This is used by content-defined chunking to force a page boundary at
    /// content-determined positions.
    #[cfg(feature = "arrow")]
    pub(crate) fn add_data_page(&mut self) -> Result<()> {
        downcast_writer!(self, typed, typed.add_data_page())
    }

    /// Sets a pre-computed distinct count on this column writer.
    ///
    /// See [`GenericColumnWriter::set_distinct_count_override`] for details.
    #[cfg(feature = "arrow")]
    pub(crate) fn set_distinct_count_override(&mut self, count: u64) {
        downcast_writer!(self, typed, typed.set_distinct_count_override(count))
    }

    /// Close this [`ColumnWriter`], returning the metadata for the column chunk.
    pub fn close(self) -> Result<ColumnCloseResult> {
        downcast_writer!(self, typed, typed.close())
    }
}

/// Create a specific column writer corresponding to column descriptor `descr`.
pub fn get_column_writer<'a>(
    descr: ColumnDescPtr,
    props: WriterPropertiesPtr,
    page_writer: Box<dyn PageWriter + 'a>,
) -> ColumnWriter<'a> {
    match descr.physical_type() {
        Type::BOOLEAN => {
            ColumnWriter::BoolColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::INT32 => {
            ColumnWriter::Int32ColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::INT64 => {
            ColumnWriter::Int64ColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::INT96 => {
            ColumnWriter::Int96ColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::FLOAT => {
            ColumnWriter::FloatColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::DOUBLE => {
            ColumnWriter::DoubleColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::BYTE_ARRAY => {
            ColumnWriter::ByteArrayColumnWriter(ColumnWriterImpl::new(descr, props, page_writer))
        }
        Type::FIXED_LEN_BYTE_ARRAY => ColumnWriter::FixedLenByteArrayColumnWriter(
            ColumnWriterImpl::new(descr, props, page_writer),
        ),
    }
}

/// Gets a typed column writer for the specific type `T`, by "up-casting" `col_writer` of
/// non-generic type to a generic column writer type `ColumnWriterImpl`.
///
/// # Panics
///
/// Panics if actual enum value for `col_writer` does not match the type `T`.
pub fn get_typed_column_writer<T: DataType>(col_writer: ColumnWriter) -> ColumnWriterImpl<T> {
    T::get_column_writer(col_writer).unwrap_or_else(|| {
        panic!(
            "Failed to convert column writer into a typed column writer for `{}` type",
            T::get_physical_type()
        )
    })
}

/// Similar to `get_typed_column_writer` but returns a reference.
pub fn get_typed_column_writer_ref<'a, 'b: 'a, T: DataType>(
    col_writer: &'b ColumnWriter<'a>,
) -> &'b ColumnWriterImpl<'a, T> {
    T::get_column_writer_ref(col_writer).unwrap_or_else(|| {
        panic!(
            "Failed to convert column writer into a typed column writer for `{}` type",
            T::get_physical_type()
        )
    })
}

/// Similar to `get_typed_column_writer` but returns a reference.
pub fn get_typed_column_writer_mut<'a, 'b: 'a, T: DataType>(
    col_writer: &'a mut ColumnWriter<'b>,
) -> &'a mut ColumnWriterImpl<'b, T> {
    T::get_column_writer_mut(col_writer).unwrap_or_else(|| {
        panic!(
            "Failed to convert column writer into a typed column writer for `{}` type",
            T::get_physical_type()
        )
    })
}

/// Metadata for a column chunk of a Parquet file.
///
/// Note this structure is returned by [`ColumnWriter::close`].
#[derive(Debug, Clone)]
pub struct ColumnCloseResult {
    /// The total number of bytes written
    pub bytes_written: u64,
    /// The total number of rows written
    pub rows_written: u64,
    /// Metadata for this column chunk
    pub metadata: ColumnChunkMetaData,
    /// Optional bloom filter for this column
    pub bloom_filter: Option<Sbbf>,
    /// Optional column index, for filtering
    pub column_index: Option<ColumnIndexMetaData>,
    /// Optional offset index, identifying page locations
    pub offset_index: Option<OffsetIndexMetaData>,
}

impl ColumnCloseResult {
    /// Rewrite the page offsets for a dictionary-first on-disk layout.
    ///
    /// A writer that buffers the whole column chunk and splices it later (the
    /// Arrow path) may accept the data pages *before* the dictionary page so the
    /// data pages can stream straight through, then emit the dictionary page
    /// first at splice. The offsets recorded during encoding therefore assume a
    /// data-pages-first layout; call this with the serialized length of the
    /// dictionary page to move it to offset 0 and shift every data page after
    /// it. A `dictionary_len` of 0 (no dictionary page) leaves the result
    /// unchanged.
    pub fn update_dictionary_location(mut self, dictionary_len: usize) -> Result<Self> {
        if dictionary_len > 0 {
            self.metadata = self
                .metadata
                .into_builder()
                .set_dictionary_page_offset(Some(0))
                .set_data_page_offset(dictionary_len as i64)
                .build()?;
            if let Some(offset_index) = self.offset_index.as_mut() {
                let mut offset = dictionary_len as i64;
                for location in &mut offset_index.page_locations {
                    location.offset = offset;
                    offset += location.compressed_page_size as i64;
                }
            }
        }
        Ok(self)
    }
}

// Metrics per page
#[derive(Default)]
struct PageMetrics {
    num_buffered_values: u32,
    num_buffered_rows: u32,
    /// Encoded bytes that the data page byte limit does not apply to,
    /// because they belong to the page's mandatory first value and cannot be
    /// moved elsewhere. Zero unless that value alone exceeded the limit
    /// *and* the encoding compresses against the preceding value; see
    /// [`ColumnValueEncoder::compresses_against_previous_value`].
    page_size_exemption: usize,
    num_page_nulls: u64,
    num_page_nans: Option<u64>,
    repetition_level_histogram: Option<LevelHistogram>,
    definition_level_histogram: Option<LevelHistogram>,
}

impl PageMetrics {
    fn new() -> Self {
        Default::default()
    }

    /// Initialize the repetition level histogram
    fn with_repetition_level_histogram(mut self, max_level: i16) -> Self {
        self.repetition_level_histogram = LevelHistogram::try_new(max_level);
        self
    }

    /// Initialize the definition level histogram
    fn with_definition_level_histogram(mut self, max_level: i16) -> Self {
        self.definition_level_histogram = LevelHistogram::try_new(max_level);
        self
    }

    /// Resets the state of this `PageMetrics` to the initial state.
    /// If histograms have been initialized their contents will be reset to zero.
    fn new_page(&mut self) {
        self.num_buffered_values = 0;
        self.num_buffered_rows = 0;
        self.page_size_exemption = 0;
        self.num_page_nulls = 0;
        self.num_page_nans = None;
        self.repetition_level_histogram
            .as_mut()
            .map(LevelHistogram::reset);
        self.definition_level_histogram
            .as_mut()
            .map(LevelHistogram::reset);
    }
}

// Metrics per column writer
#[derive(Default)]
struct ColumnMetrics<T: Default> {
    total_bytes_written: u64,
    total_rows_written: u64,
    total_uncompressed_size: u64,
    total_compressed_size: u64,
    total_num_values: u64,
    dictionary_page_offset: Option<u64>,
    data_page_offset: Option<u64>,
    min_column_value: Option<T>,
    max_column_value: Option<T>,
    num_column_nulls: u64,
    num_column_nans: Option<u64>,
    column_distinct_count: Option<u64>,
    variable_length_bytes: Option<i64>,
    repetition_level_histogram: Option<LevelHistogram>,
    definition_level_histogram: Option<LevelHistogram>,
}

impl<T: Default> ColumnMetrics<T> {
    fn new() -> Self {
        Default::default()
    }

    /// Initialize the repetition level histogram
    fn with_repetition_level_histogram(mut self, max_level: i16) -> Self {
        self.repetition_level_histogram = LevelHistogram::try_new(max_level);
        self
    }

    /// Initialize the definition level histogram
    fn with_definition_level_histogram(mut self, max_level: i16) -> Self {
        self.definition_level_histogram = LevelHistogram::try_new(max_level);
        self
    }

    /// Sum `page_histogram` into `chunk_histogram`
    fn update_histogram(
        chunk_histogram: &mut Option<LevelHistogram>,
        page_histogram: Option<&LevelHistogram>,
    ) {
        if let (Some(page_hist), Some(chunk_hist)) = (page_histogram, chunk_histogram) {
            chunk_hist.add(page_hist);
        }
    }

    /// Sum the provided PageMetrics histograms into the chunk histograms. Does nothing if
    /// page histograms are not initialized.
    fn update_from_page_metrics(&mut self, page_metrics: &PageMetrics) {
        ColumnMetrics::<T>::update_histogram(
            &mut self.definition_level_histogram,
            page_metrics.definition_level_histogram.as_ref(),
        );
        ColumnMetrics::<T>::update_histogram(
            &mut self.repetition_level_histogram,
            page_metrics.repetition_level_histogram.as_ref(),
        );
    }

    /// Sum the provided page variable_length_bytes into the chunk variable_length_bytes
    fn update_variable_length_bytes(&mut self, variable_length_bytes: Option<i64>) {
        if let Some(var_bytes) = variable_length_bytes {
            *self.variable_length_bytes.get_or_insert(0) += var_bytes;
        }
    }
}

/// Borrowed view of level data, analogous to `&str` for `LevelData`'s `String`.
///
/// `LevelDataRef` can be constructed from `LevelData` and directly from an existing
/// `&[i16]` without allocating.
///
/// The variants are different physical representations of the same logical
/// sequence of levels.
#[derive(Debug, Clone, Copy)]
pub(crate) enum LevelDataRef<'a> {
    Absent,
    Materialized(&'a [i16]),
    Uniform { value: i16, count: usize },
}

impl<'a> From<&'a [i16]> for LevelDataRef<'a> {
    fn from(levels: &'a [i16]) -> Self {
        Self::Materialized(levels)
    }
}

impl<'a> From<Option<&'a [i16]>> for LevelDataRef<'a> {
    fn from(levels: Option<&'a [i16]>) -> Self {
        levels.map_or(Self::Absent, Self::from)
    }
}

impl LevelDataRef<'_> {
    pub(crate) fn len(self) -> usize {
        match self {
            Self::Absent => 0,
            Self::Materialized(values) => values.len(),
            Self::Uniform { count, .. } => count,
        }
    }

    pub(crate) fn first(self) -> Option<i16> {
        match self {
            Self::Absent => None,
            Self::Materialized(values) => values.first().copied(),
            Self::Uniform { value, count } => (count > 0).then_some(value),
        }
    }

    #[cfg(feature = "arrow")]
    pub(crate) fn value_at(self, idx: usize) -> Option<i16> {
        match self {
            Self::Absent => None,
            Self::Materialized(values) => values.get(idx).copied(),
            Self::Uniform { value, count } => (idx < count).then_some(value),
        }
    }

    pub(crate) fn slice(self, offset: usize, len: usize) -> Self {
        match self {
            Self::Absent => Self::Absent,
            Self::Materialized(values) => Self::Materialized(&values[offset..offset + len]),
            Self::Uniform { value, .. } => Self::Uniform { value, count: len },
        }
    }

    /// Count of positions in this slice that represent an actual value
    /// (definition level equal to `max_def`). `Absent` means the column has
    /// `max_def == 0` and every position is a value, so the implicit count
    /// is the caller-supplied `total`.
    pub(crate) fn value_count(self, total: usize, max_def: i16) -> usize {
        match self {
            Self::Absent => total,
            Self::Materialized(values) => values.iter().filter(|&&d| d == max_def).count(),
            Self::Uniform { value, count } => {
                if value == max_def {
                    count
                } else {
                    0
                }
            }
        }
    }
}

/// Typed column writer for a primitive column.
pub type ColumnWriterImpl<'a, T> = GenericColumnWriter<'a, ColumnValueEncoderImpl<T>>;

/// Generic column writer for a primitive Parquet column
pub struct GenericColumnWriter<'a, E: ColumnValueEncoder> {
    // Column writer properties
    descr: ColumnDescPtr,
    props: WriterPropertiesPtr,
    /// Per-column settings for [`Self::descr`], resolved once here so that the
    /// per-batch and per-page write paths never search the per-column override
    /// map in `props` again.
    column_props: ResolvedColumnProperties,

    page_writer: Box<dyn PageWriter + 'a>,
    codec: Compression,
    compressor: Option<Box<dyn Codec>>,
    encoder: E,

    page_metrics: PageMetrics,
    // Metrics per column writer
    column_metrics: ColumnMetrics<E::T>,

    /// Pre-computed distinct count to write into column chunk statistics.
    /// When set, takes precedence over `column_metrics.column_distinct_count`.
    distinct_count_override: Option<u64>,

    /// The order of encodings within the generated metadata does not impact its meaning,
    /// but we use a BTreeSet so that the output is deterministic
    encodings: BTreeSet<Encoding>,
    encoding_stats: Vec<PageEncodingStats>,
    // Streaming level encoders for definition/repetition levels.
    def_levels_encoder: LevelEncoder,
    rep_levels_encoder: LevelEncoder,
    data_pages: VecDeque<CompressedPage>,
    // column index and offset index
    column_index_builder: ColumnIndexBuilder,
    offset_index_builder: Option<OffsetIndexBuilder>,

    // Below fields used to incrementally check boundary order across data pages.
    // We assume they are ascending/descending until proven wrong.
    data_page_boundary_ascending: bool,
    data_page_boundary_descending: bool,
    /// (min, max)
    last_non_null_data_page_min_max: Option<(E::T, E::T)>,
}

impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a, E> {
    /// Returns a new instance of [`GenericColumnWriter`].
    pub fn new(
        descr: ColumnDescPtr,
        props: WriterPropertiesPtr,
        page_writer: Box<dyn PageWriter + 'a>,
    ) -> Self {
        let column_props = props.resolve_column_properties(descr.path());
        let codec = column_props.compression;
        let codec_options = CodecOptionsBuilder::default().build();
        let compressor = create_codec(codec, &codec_options).unwrap();
        let encoder = E::try_new(&descr, props.as_ref(), &column_props).unwrap();

        let statistics_enabled = column_props.statistics_enabled;

        let mut encodings = BTreeSet::new();
        // Used for level information
        encodings.insert(Encoding::RLE);

        let mut page_metrics = PageMetrics::new();
        let mut column_metrics = ColumnMetrics::<E::T>::new();

        // Initialize level histograms if collecting page or chunk statistics
        if statistics_enabled != EnabledStatistics::None {
            page_metrics = page_metrics
                .with_repetition_level_histogram(descr.max_rep_level())
                .with_definition_level_histogram(descr.max_def_level());
            column_metrics = column_metrics
                .with_repetition_level_histogram(descr.max_rep_level())
                .with_definition_level_histogram(descr.max_def_level())
        }

        // Disable column_index_builder if not collecting page statistics.
        let mut column_index_builder = ColumnIndexBuilder::new(descr.physical_type());
        if statistics_enabled != EnabledStatistics::Page {
            column_index_builder.to_invalid()
        }

        // Disable offset_index_builder if requested by user.
        let offset_index_builder = match props.offset_index_disabled() {
            false => Some(OffsetIndexBuilder::new()),
            _ => None,
        };

        Self {
            def_levels_encoder: Self::create_level_encoder(descr.max_def_level(), &props),
            rep_levels_encoder: Self::create_level_encoder(descr.max_rep_level(), &props),
            descr,
            props,
            column_props,
            page_writer,
            codec,
            compressor,
            encoder,
            data_pages: VecDeque::new(),
            page_metrics,
            column_metrics,
            distinct_count_override: None,
            column_index_builder,
            offset_index_builder,
            encodings,
            encoding_stats: vec![],
            data_page_boundary_ascending: true,
            data_page_boundary_descending: true,
            last_non_null_data_page_min_max: None,
        }
    }

    /// Sets a pre-computed distinct count to write into column chunk statistics.
    ///
    /// When set, this value is written as `distinct_count` in the row group statistics
    /// footer. It takes precedence over any `distinct_count` passed through
    /// [`Self::write_batch_with_statistics`].
    #[cfg(feature = "arrow")]
    pub(crate) fn set_distinct_count_override(&mut self, count: u64) {
        self.distinct_count_override = Some(count);
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) fn write_batch_internal(
        &mut self,
        values: &E::Values,
        value_indices: Option<&[usize]>,
        def_levels: LevelDataRef<'_>,
        rep_levels: LevelDataRef<'_>,
        min: Option<&E::T>,
        max: Option<&E::T>,
        distinct_count: Option<u64>,
    ) -> Result<usize> {
        // Check if number of definition levels is the same as number of repetition levels.
        if def_levels.len() != 0 && rep_levels.len() != 0 && def_levels.len() != rep_levels.len() {
            return Err(general_err!(
                "Inconsistent length of definition and repetition levels: {} != {}",
                def_levels.len(),
                rep_levels.len()
            ));
        }

        // We check for DataPage limits only after we have inserted the values. If a user
        // writes a large number of values, the DataPage size can be well above the limit.
        //
        // The purpose of this chunking is to bound this. Even if a user writes large
        // number of values, the chunking will ensure that we add data page at a
        // reasonable pagesize limit.

        // TODO: find out why we don't account for size of levels when we estimate page
        // size.
        let num_levels = def_levels.len().max(rep_levels.len());
        let num_levels = if num_levels > 0 {
            num_levels
        } else {
            value_indices.map_or_else(|| values.len(), |i| i.len())
        };

        if let Some(min) = min {
            update_min(&self.descr, min, &mut self.column_metrics.min_column_value);
        }
        if let Some(max) = max {
            update_max(&self.descr, max, &mut self.column_metrics.max_column_value);
        }

        // We can only set the distinct count if there are no other writes
        if self.encoder.num_values() == 0 {
            self.column_metrics.column_distinct_count = distinct_count;
        } else {
            self.column_metrics.column_distinct_count = None;
        }

        let mut values_offset = 0;
        let mut levels_offset = 0;
        let both_levels_compact = !matches!(def_levels, LevelDataRef::Materialized(_))
            && !matches!(rep_levels, LevelDataRef::Materialized(_));
        let has_levels = !matches!(def_levels, LevelDataRef::Absent)
            || !matches!(rep_levels, LevelDataRef::Absent);

        // When both level vectors are compact (Uniform or Absent), there is no
        // materialized slice to split and the per-mini-batch work is O(1), so we
        // can safely use a much larger batch size.
        let base_batch_size = if both_levels_compact && has_levels {
            self.props.data_page_row_count_limit()
        } else {
            self.props.write_batch_size()
        };
        debug_assert!(base_batch_size > 0);

        let chunker = ByteBudgetChunker::new(&self.descr, &self.column_props, base_batch_size);
        while levels_offset < num_levels {
            let mut end_offset = num_levels.min(levels_offset + base_batch_size);

            // Split at record boundary
            if let LevelDataRef::Materialized(levels) = rep_levels {
                while end_offset < levels.len() && levels[end_offset] != 0 {
                    end_offset += 1;
                }
            }

            let chunk_size = end_offset - levels_offset;
            let chunk_def = def_levels.slice(levels_offset, chunk_size);
            let chunk_rep = rep_levels.slice(levels_offset, chunk_size);

            // Key decision point: can we write this whole chunk as one
            // mini-batch (the common case — small or fixed-width values, no
            // further page-size accounting needed), or must we fall back to
            // byte-budget-aware sub-batching to keep a page from overshooting
            // `data_page_size_limit`? `pick_sub_batch` returns `None` for
            // the former, and otherwise how wide one mini-batch may be.
            let sub_batch = chunker.pick_sub_batch(
                &self.encoder,
                values,
                value_indices,
                chunk_def,
                values_offset,
                chunk_size,
            );

            match sub_batch {
                None => {
                    values_offset += self.write_mini_batch(
                        values,
                        values_offset,
                        value_indices,
                        chunk_size,
                        chunk_def,
                        chunk_rep,
                    )?;
                }
                Some(sub_batch) => {
                    values_offset += self.write_granular_chunk(
                        values,
                        values_offset,
                        value_indices,
                        chunk_size,
                        chunk_def,
                        chunk_rep,
                        sub_batch,
                    )?;
                }
            }
            levels_offset = end_offset;
        }

        // Return total number of values processed.
        Ok(values_offset)
    }

    /// Writes batch of values, definition levels and repetition levels.
    /// Returns number of values processed (written).
    ///
    /// If definition and repetition levels are provided, we write fully those levels and
    /// select how many values to write (this number will be returned), since number of
    /// actual written values may be smaller than provided values.
    ///
    /// If only values are provided, then all values are written and the length of
    /// of the values buffer is returned.
    ///
    /// Definition and/or repetition levels can be omitted, if values are
    /// non-nullable and/or non-repeated.
    pub fn write_batch(
        &mut self,
        values: &E::Values,
        def_levels: Option<&[i16]>,
        rep_levels: Option<&[i16]>,
    ) -> Result<usize> {
        self.write_batch_internal(
            values,
            None,
            LevelDataRef::from(def_levels),
            LevelDataRef::from(rep_levels),
            None,
            None,
            None,
        )
    }

    /// Writer may optionally provide pre-calculated statistics for use when computing
    /// chunk-level statistics
    ///
    /// NB: [`WriterProperties::statistics_enabled`] must be set to [`EnabledStatistics::Chunk`]
    /// for these statistics to take effect. If [`EnabledStatistics::None`] they will be ignored,
    /// and if [`EnabledStatistics::Page`] the chunk statistics will instead be computed from the
    /// computed page statistics
    pub fn write_batch_with_statistics(
        &mut self,
        values: &E::Values,
        def_levels: Option<&[i16]>,
        rep_levels: Option<&[i16]>,
        min: Option<&E::T>,
        max: Option<&E::T>,
        distinct_count: Option<u64>,
    ) -> Result<usize> {
        self.write_batch_internal(
            values,
            None,
            LevelDataRef::from(def_levels),
            LevelDataRef::from(rep_levels),
            min,
            max,
            distinct_count,
        )
    }

    /// Returns the estimated total memory usage.
    ///
    /// Unlike [`Self::get_estimated_total_bytes`] this is an estimate
    /// of the current memory usage and not the final anticipated encoded size.
    #[cfg(feature = "arrow")]
    pub(crate) fn memory_size(&self) -> usize {
        // In-flight encoder buffers, plus any completed pages still held on the
        // heap: the dictionary-column data pages buffered here (column-at-a-time
        // path), plus whatever the page writer keeps resident. A page writer
        // that spills completed pages off-heap reports far less than the bytes
        // it was handed, so this tracks real memory rather than bytes written.
        self.encoder.estimated_memory_size()
            + self
                .data_pages
                .iter()
                .map(|page| page.memory_usage())
                .sum::<usize>()
            + self.page_writer.buffered_memory_size()
    }

    /// Returns total number of bytes written by this column writer so far.
    /// This value is also returned when column writer is closed.
    ///
    /// Note: this value does not include any buffered data that has not
    /// yet been flushed to a page.
    pub fn get_total_bytes_written(&self) -> u64 {
        self.column_metrics.total_bytes_written
    }

    /// Returns the estimated total encoded bytes for this column writer.
    ///
    /// Unlike [`Self::get_total_bytes_written`] this includes an estimate
    /// of any data that has not yet been flushed to a page, based on it's
    /// anticipated encoded size.
    #[cfg(feature = "arrow")]
    pub(crate) fn get_estimated_total_bytes(&self) -> u64 {
        self.data_pages
            .iter()
            .map(|page| page.data().len() as u64)
            .sum::<u64>()
            + self.column_metrics.total_bytes_written
            + self.encoder.estimated_data_page_size() as u64
            + self.encoder.estimated_dict_page_size().unwrap_or_default() as u64
    }

    /// Returns total number of rows written by this column writer so far.
    /// This value is also returned when column writer is closed.
    pub fn get_total_rows_written(&self) -> u64 {
        self.column_metrics.total_rows_written
    }

    /// Returns a reference to a [`ColumnDescPtr`]
    pub fn get_descriptor(&self) -> &ColumnDescPtr {
        &self.descr
    }

    /// Finalizes writes and closes the column writer.
    /// Returns total bytes written, total rows written and column chunk metadata.
    pub fn close(mut self) -> Result<ColumnCloseResult> {
        if self.page_metrics.num_buffered_values > 0 {
            self.add_data_page()?;
        }
        if self.encoder.has_dictionary() {
            self.write_dictionary_page()?;
        }
        self.flush_data_pages()?;
        let metadata = self.build_column_metadata()?;
        self.page_writer.close()?;

        let write_bloom_filter = self.props.bloom_filter_for_dictionary_encoded_chunks()
            || self.has_non_dictionary_data_page();
        let bloom_filter = self
            .encoder
            .flush_bloom_filter()
            .filter(|_| write_bloom_filter);

        let boundary_order = match (
            self.data_page_boundary_ascending,
            self.data_page_boundary_descending,
        ) {
            // If the lists are composed of equal elements then will be marked as ascending
            // (Also the case if all pages are null pages)
            (true, _) => BoundaryOrder::ASCENDING,
            (false, true) => BoundaryOrder::DESCENDING,
            (false, false) => BoundaryOrder::UNORDERED,
        };
        self.column_index_builder.set_boundary_order(boundary_order);

        let column_index = match self.column_index_builder.valid() {
            true => Some(self.column_index_builder.build()?),
            false => None,
        };

        let offset_index = self.offset_index_builder.map(|b| b.build());

        Ok(ColumnCloseResult {
            bytes_written: self.column_metrics.total_bytes_written,
            rows_written: self.column_metrics.total_rows_written,
            bloom_filter,
            metadata,
            column_index,
            offset_index,
        })
    }

    /// Writes a chunk in sub-batches sized by `sub_batch`, checking the page
    /// byte limit after each. This keeps the page size close to
    /// `data_page_size_limit` instead of overshooting it by a whole chunk.
    ///
    /// [`SubBatchStrategy::Values`] windows are cut on value boundaries,
    /// walking definition levels to find where the value budget is used up;
    /// [`SubBatchStrategy::Levels`] windows are a fixed level count. See
    /// [`SubBatchStrategy`] for which budget gets which and why.
    ///
    /// For repeated/nested columns sub-batches then extend to the next
    /// `rep == 0` boundary so a record never spans data pages, matching the
    /// parquet format rule. A record holding several over-limit values
    /// therefore still exceeds the budget; that is inherent to the format.
    ///
    /// Returns the total number of values consumed across all sub-batches.
    ///
    /// `#[inline(never)]` keeps this slow path — only reached for
    /// variable-width columns whose values need page splitting — out of
    /// the hot `write_batch_internal` loop.
    #[expect(clippy::too_many_arguments)]
    #[inline(never)]
    fn write_granular_chunk(
        &mut self,
        values: &E::Values,
        values_offset: usize,
        value_indices: Option<&[usize]>,
        chunk_size: usize,
        chunk_def: LevelDataRef<'_>,
        chunk_rep: LevelDataRef<'_>,
        sub_batch: SubBatchStrategy,
    ) -> Result<usize> {
        // The chunker always sizes a sub-batch to at least one value or one
        // level, and a value-exact window spans at least one level, so each
        // iteration below makes progress (`sub_end > sub_start`).
        debug_assert!(
            matches!(sub_batch, SubBatchStrategy::Values(n) | SubBatchStrategy::Levels(n) if n >= 1),
            "chunker must size at least one value or level"
        );
        let max_def_level = self.descr.max_def_level();
        let mut values_consumed = 0;
        let mut sub_start = 0;
        while sub_start < chunk_size {
            let window_end = match sub_batch {
                SubBatchStrategy::Values(n) => {
                    Self::window_end_for_values(chunk_def, chunk_size, max_def_level, sub_start, n)
                }
                SubBatchStrategy::Levels(n) => (sub_start + n).min(chunk_size),
            };
            let sub_end = match chunk_rep {
                LevelDataRef::Materialized(levels) => {
                    // Extend the window to the next record boundary
                    // (rep == 0) so a record never spans data pages. Packing
                    // whole records rather than stepping one record at a time
                    // avoids calling `write_mini_batch` per record: records
                    // average only a handful of levels, so a record-at-a-time
                    // step would issue many more mini-batches than necessary.
                    let mut e = window_end;
                    while e < chunk_size && levels[e] != 0 {
                        e += 1;
                    }
                    e
                }
                _ => window_end,
            };
            let sub_len = sub_end - sub_start;
            let written = self.write_mini_batch(
                values,
                values_offset + values_consumed,
                value_indices,
                sub_len,
                chunk_def.slice(sub_start, sub_len),
                chunk_rep.slice(sub_start, sub_len),
            )?;
            values_consumed += written;
            sub_start = sub_end;
        }
        Ok(values_consumed)
    }

    fn has_non_dictionary_data_page(&self) -> bool {
        self.encoding_stats.iter().any(|stats| {
            matches!(
                stats.page_type,
                PageType::DATA_PAGE | PageType::DATA_PAGE_V2
            ) && !matches!(
                stats.encoding,
                Encoding::PLAIN_DICTIONARY | Encoding::RLE_DICTIONARY
            )
        })
    }

    /// Index one past the last level of a sub-batch window that starts at
    /// `start` and covers at most `max_values` values, clamped to
    /// `chunk_size`.
    ///
    /// Nulls trailing the last value are left to the next window, so a window
    /// ends immediately after the value that exhausts its budget. The window
    /// always spans at least one level, so callers make progress: `start` is
    /// below `chunk_size` and `max_values` is at least one.
    ///
    /// Only nullable and nested columns pay the level walk. It runs solely in
    /// the granular path, whose values are by definition large enough to
    /// overflow a page budget, so touching each of the chunk's levels once is
    /// noise next to writing those values — and that path already makes full
    /// def-level and rep-level passes.
    fn window_end_for_values(
        chunk_def: LevelDataRef<'_>,
        chunk_size: usize,
        max_def_level: i16,
        start: usize,
        max_values: usize,
    ) -> usize {
        match chunk_def {
            // `max_def_level == 0`: every level is a value.
            LevelDataRef::Absent => (start + max_values).min(chunk_size),
            LevelDataRef::Uniform { value, .. } => {
                if value == max_def_level {
                    (start + max_values).min(chunk_size)
                } else {
                    // Uniformly below max def: the chunk holds no values at
                    // all, so no window boundary can help. (The chunker
                    // returns `None` for such a chunk, so this is defensive.)
                    chunk_size
                }
            }
            LevelDataRef::Materialized(levels) => {
                let mut seen = 0;
                let mut end = start;
                while end < chunk_size && seen < max_values {
                    if levels[end] == max_def_level {
                        seen += 1;
                    }
                    end += 1;
                }
                end
            }
        }
    }

    /// Creates a new streaming level encoder appropriate for the writer version.
    fn create_level_encoder(max_level: i16, props: &WriterProperties) -> LevelEncoder {
        match props.writer_version() {
            WriterVersion::PARQUET_1_0 => LevelEncoder::v1_streaming(max_level),
            WriterVersion::PARQUET_2_0 => LevelEncoder::v2_streaming(max_level),
        }
    }

    /// Writes mini batch of values, definition and repetition levels.
    /// This allows fine-grained processing of values and maintaining a reasonable
    /// page size.
    fn write_mini_batch(
        &mut self,
        values: &E::Values,
        values_offset: usize,
        value_indices: Option<&[usize]>,
        num_levels: usize,
        def_levels: LevelDataRef<'_>,
        rep_levels: LevelDataRef<'_>,
    ) -> Result<usize> {
        // Process definition levels and determine how many values to write.
        let values_to_write = if self.descr.max_def_level() > 0 {
            let max_def = self.descr.max_def_level();
            match def_levels {
                LevelDataRef::Absent => {
                    return Err(general_err!(
                        "Definition levels are required, because max definition level = {}",
                        self.descr.max_def_level()
                    ));
                }
                LevelDataRef::Materialized(levels) => {
                    // General path for caller-provided or already-materialized
                    // level buffers.
                    let mut values_to_write = 0usize;
                    let encoder = &mut self.def_levels_encoder;
                    match self.page_metrics.definition_level_histogram.as_mut() {
                        Some(histogram) => encoder.put_with_observer(levels, |level, count| {
                            values_to_write += count * (level == max_def) as usize;
                            histogram.increment_by(level, count as i64);
                        }),
                        None => encoder.put_with_observer(levels, |level, count| {
                            values_to_write += count * (level == max_def) as usize;
                        }),
                    };
                    self.page_metrics.num_page_nulls += (levels.len() - values_to_write) as u64;
                    values_to_write
                }
                LevelDataRef::Uniform { value, count } => {
                    // Fast path for all-null, all-valid, or otherwise uniform
                    // definition levels without materializing a level buffer.
                    let encoder = &mut self.def_levels_encoder;
                    match self.page_metrics.definition_level_histogram.as_mut() {
                        Some(histogram) => {
                            encoder.put_n_with_observer(value, count, |level, run_len| {
                                histogram.increment_by(level, run_len as i64);
                            })
                        }
                        None => encoder.put_n_with_observer(value, count, |_, _| {}),
                    }
                    let values_to_write = count * (value == max_def) as usize;
                    self.page_metrics.num_page_nulls += (count - values_to_write) as u64;
                    values_to_write
                }
            }
        } else {
            num_levels
        };

        // Process repetition levels and determine how many rows we are about to process.
        if self.descr.max_rep_level() > 0 {
            // A row could contain more than one value.
            let first_level = rep_levels.first().ok_or_else(|| {
                general_err!(
                    "Repetition levels are required, because max repetition level = {}",
                    self.descr.max_rep_level()
                )
            })?;

            if first_level != 0 {
                return Err(general_err!(
                    "Write must start at a record boundary, got non-zero repetition level of {}",
                    first_level
                ));
            }

            let mut new_rows = 0u32;
            match rep_levels {
                LevelDataRef::Absent => unreachable!(),
                LevelDataRef::Materialized(levels) => {
                    let encoder = &mut self.rep_levels_encoder;
                    match self.page_metrics.repetition_level_histogram.as_mut() {
                        Some(histogram) => encoder.put_with_observer(levels, |level, count| {
                            new_rows += (count as u32) * (level == 0) as u32;
                            histogram.increment_by(level, count as i64);
                        }),
                        None => encoder.put_with_observer(levels, |level, count| {
                            new_rows += (count as u32) * (level == 0) as u32;
                        }),
                    };
                }
                LevelDataRef::Uniform { value, count } => {
                    let encoder = &mut self.rep_levels_encoder;
                    match self.page_metrics.repetition_level_histogram.as_mut() {
                        Some(histogram) => {
                            encoder.put_n_with_observer(value, count, |level, run_len| {
                                new_rows += (run_len as u32) * (level == 0) as u32;
                                histogram.increment_by(level, run_len as i64);
                            })
                        }
                        None => encoder.put_n_with_observer(value, count, |level, run_len| {
                            new_rows += (run_len as u32) * (level == 0) as u32;
                        }),
                    }
                }
            }
            self.page_metrics.num_buffered_rows += new_rows;
        } else {
            // Each value is exactly one row.
            // Equals to the number of values, we count nulls as well.
            self.page_metrics.num_buffered_rows += num_levels as u32;
        }

        match value_indices {
            Some(indices) => {
                let indices = &indices[values_offset..values_offset + values_to_write];
                self.encoder.write_gather(values, indices)?;
            }
            None => self.encoder.write(values, values_offset, values_to_write)?,
        }

        let page_was_empty = self.page_metrics.num_buffered_values == 0;
        self.page_metrics.num_buffered_values += num_levels as u32;

        if page_was_empty && values_to_write == 1 {
            self.set_page_size_exemption();
        }

        if self.should_add_data_page() {
            self.add_data_page()?;
        }

        if self.should_dict_fallback() {
            self.dict_fallback()?;
        }

        Ok(values_to_write)
    }

    /// Returns true if we need to fall back to non-dictionary encoding.
    ///
    /// We can only fall back if dictionary encoder is set and we have exceeded dictionary
    /// size.
    #[inline]
    fn should_dict_fallback(&self) -> bool {
        match self.encoder.estimated_dict_page_size() {
            Some(size) => size >= self.column_props.dictionary_page_size_limit,
            None => false,
        }
    }

    /// Exempt a page's mandatory first value from the data page byte limit,
    /// when that value alone already exceeds it.
    ///
    /// Parquet requires every data page to hold at least one value, so such a
    /// value cannot be split out no matter how the limit is set. Counting it
    /// against the limit makes the limit unsatisfiable, and
    /// [`Self::should_add_data_page`] then cuts a page after every single
    /// value.
    ///
    /// For `DELTA_BYTE_ARRAY` that costs more than the extra pages. A value is
    /// stored as a suffix of the value before it, and a page boundary resets
    /// what "the value before it" refers to, so one value per page means every
    /// value is stored in full: a column of large values sharing long prefixes
    /// writes exactly the bytes `PLAIN` would
    /// ([#10489](https://github.com/apache/arrow-rs/issues/10489)).
    ///
    /// Only encodings that compress against the preceding value opt in, so
    /// `PLAIN` and `DELTA_LENGTH_BYTE_ARRAY` keep their tighter one-value page
    /// bound.
    ///
    /// The caller's trigger keys on a page-opening mini-batch holding exactly
    /// one value. `write_granular_chunk` cuts windows after an exact value
    /// count, so an over-limit value gets a single-value mini-batch whether
    /// or not the chunk contains nulls; see
    /// `test_column_writer_delta_byte_array_nullable_shared_prefix_dedup`.
    /// The exception is a repeated column, where a record holding several
    /// over-limit values cannot be split across pages at all.
    #[cold]
    fn set_page_size_exemption(&mut self) {
        if !self.encoder.compresses_against_previous_value() {
            return;
        }
        let size = self.encoder.estimated_data_page_size();
        if size >= self.column_props.data_page_size_limit {
            self.page_metrics.page_size_exemption = size;
        }
    }

    /// Returns true if there is enough data for a data page, false otherwise.
    #[inline]
    fn should_add_data_page(&self) -> bool {
        // This is necessary in the event of a much larger dictionary size than page size
        //
        // In such a scenario the dictionary decoder may return an estimated encoded
        // size in excess of the page size limit, even when there are no buffered values
        if self.page_metrics.num_buffered_values == 0 {
            return false;
        }

        self.page_metrics.num_buffered_rows as usize >= self.props.data_page_row_count_limit()
            || self
                .encoder
                .estimated_data_page_size()
                .saturating_sub(self.page_metrics.page_size_exemption)
                >= self.column_props.data_page_size_limit
    }

    /// Performs dictionary fallback.
    /// Prepares and writes dictionary and all data pages into page writer.
    fn dict_fallback(&mut self) -> Result<()> {
        // At this point we know that we need to fall back.
        if self.page_metrics.num_buffered_values > 0 {
            self.add_data_page()?;
        }
        self.write_dictionary_page()?;
        self.flush_data_pages()?;
        Ok(())
    }

    // For float columns, always provide Some(n), even if n is 0
    // For non-float columns, always provide None
    fn get_nan_count<T: ParquetValueType>(&self) -> Option<i64> {
        let nan_count = || {
            let nan_count = self.page_metrics.num_page_nans.unwrap_or(0);
            match i64::try_from(nan_count) {
                Ok(count) => Some(count),
                _ => Some(i64::MAX),
            }
        };
        match T::PHYSICAL_TYPE {
            Type::FLOAT | Type::DOUBLE => nan_count(),
            Type::FIXED_LEN_BYTE_ARRAY
                if matches!(self.descr.logical_type_ref(), Some(LogicalType::Float16)) =>
            {
                nan_count()
            }
            _ => None,
        }
    }

    /// Update the column index and offset index when adding the data page
    fn update_column_offset_index(
        &mut self,
        page_statistics: Option<&ValueStatistics<E::T>>,
        page_variable_length_bytes: Option<i64>,
    ) {
        // update the column index
        let null_page =
            (self.page_metrics.num_buffered_rows as u64) == self.page_metrics.num_page_nulls;
        // a page contains only null values,
        // and writers have to set the corresponding entries in min_values and max_values to byte[0]
        if null_page && self.column_index_builder.valid() {
            self.column_index_builder.append(
                null_page,
                vec![],
                vec![],
                self.page_metrics.num_page_nulls as i64,
                self.get_nan_count::<E::T>(),
            );
        } else if self.column_index_builder.valid() {
            // from page statistics
            // If can't get the page statistics, ignore this column/offset index for this column chunk
            match &page_statistics {
                None => {
                    self.column_index_builder.to_invalid();
                }
                Some(stat) => {
                    // Check if min/max are still ascending/descending across pages
                    let new_min = stat.min_opt().unwrap();
                    let new_max = stat.max_opt().unwrap();
                    if let Some((last_min, last_max)) = &self.last_non_null_data_page_min_max {
                        let basic_info = self.descr.get_basic_info();
                        if self.data_page_boundary_ascending {
                            // If last min/max are greater than new min/max then not ascending anymore
                            let not_ascending = compare_greater(basic_info, last_min, new_min)
                                || compare_greater(basic_info, last_max, new_max);
                            if not_ascending {
                                self.data_page_boundary_ascending = false;
                            }
                        }

                        if self.data_page_boundary_descending {
                            // If new min/max are greater than last min/max then not descending anymore
                            let not_descending = compare_greater(basic_info, new_min, last_min)
                                || compare_greater(basic_info, new_max, last_max);
                            if not_descending {
                                self.data_page_boundary_descending = false;
                            }
                        }
                    }
                    self.last_non_null_data_page_min_max = Some((new_min.clone(), new_max.clone()));

                    if self.can_truncate_value() {
                        self.column_index_builder.append(
                            null_page,
                            self.truncate_min_value(
                                self.props.column_index_truncate_length(),
                                stat.min_bytes_opt().unwrap(),
                            )
                            .0,
                            self.truncate_max_value(
                                self.props.column_index_truncate_length(),
                                stat.max_bytes_opt().unwrap(),
                            )
                            .0,
                            self.page_metrics.num_page_nulls as i64,
                            self.get_nan_count::<E::T>(),
                        );
                    } else {
                        self.column_index_builder.append(
                            null_page,
                            stat.min_bytes_opt().unwrap().to_vec(),
                            stat.max_bytes_opt().unwrap().to_vec(),
                            self.page_metrics.num_page_nulls as i64,
                            self.get_nan_count::<E::T>(),
                        );
                    }
                }
            }
        }

        // Append page histograms to the `ColumnIndex` histograms
        self.column_index_builder.append_histograms(
            &self.page_metrics.repetition_level_histogram,
            &self.page_metrics.definition_level_histogram,
        );

        // Update the offset index
        if let Some(builder) = self.offset_index_builder.as_mut() {
            builder.append_row_count(self.page_metrics.num_buffered_rows as i64);
            builder.append_unencoded_byte_array_data_bytes(page_variable_length_bytes);
        }
    }

    /// Determine if we should allow truncating min/max values for this column's statistics
    fn can_truncate_value(&self) -> bool {
        match self.descr.physical_type() {
            // Don't truncate for Float16 and Decimal because their sort order is different
            // from that of FIXED_LEN_BYTE_ARRAY sort order.
            // So truncation of those types could lead to inaccurate min/max statistics
            Type::FIXED_LEN_BYTE_ARRAY
                if !matches!(
                    self.descr.logical_type_ref(),
                    Some(&LogicalType::Decimal { .. } | &LogicalType::Float16)
                ) =>
            {
                true
            }
            Type::BYTE_ARRAY => true,
            // Truncation only applies for fba/binary physical types
            _ => false,
        }
    }

    /// Returns `true` if this column's logical type is a UTF-8 string.
    fn is_utf8(&self) -> bool {
        self.get_descriptor().logical_type_ref() == Some(&LogicalType::String)
            || self.get_descriptor().converted_type() == ConvertedType::UTF8
    }

    /// Truncates a binary statistic to at most `truncation_length` bytes.
    ///
    /// If truncation is not possible, returns `data`.
    ///
    /// The `bool` in the returned tuple indicates whether truncation occurred or not.
    ///
    /// UTF-8 Note:
    /// If the column type indicates UTF-8, and `data` contains valid UTF-8, then the result will
    /// also remain valid UTF-8, but may be less than `truncation_length` bytes to avoid splitting
    /// on non-character boundaries.
    fn truncate_min_value(&self, truncation_length: Option<usize>, data: &[u8]) -> (Vec<u8>, bool) {
        truncation_length
            .filter(|l| data.len() > *l)
            .and_then(|l|
                // don't do extra work if this column isn't UTF-8
                if self.is_utf8() {
                    match str::from_utf8(data) {
                        Ok(str_data) => truncate_utf8(str_data, l),
                        Err(_) => Some(data[..l].to_vec()),
                    }
                } else {
                    Some(data[..l].to_vec())
                }
            )
            .map(|truncated| (truncated, true))
            .unwrap_or_else(|| (data.to_vec(), false))
    }

    /// Truncates a binary statistic to at most `truncation_length` bytes, and then increment the
    /// final byte(s) to yield a valid upper bound. This may result in a result of less than
    /// `truncation_length` bytes if the last byte(s) overflows.
    ///
    /// If truncation is not possible, returns `data`.
    ///
    /// The `bool` in the returned tuple indicates whether truncation occurred or not.
    ///
    /// UTF-8 Note:
    /// If the column type indicates UTF-8, and `data` contains valid UTF-8, then the result will
    /// also remain valid UTF-8 (but again may be less than `truncation_length` bytes). If `data`
    /// does not contain valid UTF-8, then truncation will occur as if the column is non-string
    /// binary.
    fn truncate_max_value(&self, truncation_length: Option<usize>, data: &[u8]) -> (Vec<u8>, bool) {
        truncation_length
            .filter(|l| data.len() > *l)
            .and_then(|l|
                // don't do extra work if this column isn't UTF-8
                if self.is_utf8() {
                    match str::from_utf8(data) {
                        Ok(str_data) => truncate_and_increment_utf8(str_data, l),
                        Err(_) => increment(data[..l].to_vec()),
                    }
                } else {
                    increment(data[..l].to_vec())
                }
            )
            .map(|truncated| (truncated, true))
            .unwrap_or_else(|| (data.to_vec(), false))
    }

    /// Truncate the min and max values that will be written to a data page
    /// header or column chunk Statistics
    fn truncate_statistics(&self, statistics: Statistics) -> Statistics {
        let backwards_compatible_min_max = self.descr.sort_order().is_signed();
        match statistics {
            Statistics::ByteArray(stats) if stats._internal_has_min_max_set() => {
                let (min, did_truncate_min) = self.truncate_min_value(
                    self.props.statistics_truncate_length(),
                    stats.min_bytes_opt().unwrap(),
                );
                let (max, did_truncate_max) = self.truncate_max_value(
                    self.props.statistics_truncate_length(),
                    stats.max_bytes_opt().unwrap(),
                );
                Statistics::ByteArray(
                    ValueStatistics::new(
                        Some(min.into()),
                        Some(max.into()),
                        stats.distinct_count(),
                        stats.null_count_opt(),
                        backwards_compatible_min_max,
                    )
                    .with_max_is_exact(!did_truncate_max)
                    .with_min_is_exact(!did_truncate_min),
                )
            }
            Statistics::FixedLenByteArray(stats)
                if (stats._internal_has_min_max_set() && self.can_truncate_value()) =>
            {
                let (min, did_truncate_min) = self.truncate_min_value(
                    self.props.statistics_truncate_length(),
                    stats.min_bytes_opt().unwrap(),
                );
                let (max, did_truncate_max) = self.truncate_max_value(
                    self.props.statistics_truncate_length(),
                    stats.max_bytes_opt().unwrap(),
                );
                Statistics::FixedLenByteArray(
                    ValueStatistics::new(
                        Some(min.into()),
                        Some(max.into()),
                        stats.distinct_count(),
                        stats.null_count_opt(),
                        backwards_compatible_min_max,
                    )
                    .with_max_is_exact(!did_truncate_max)
                    .with_min_is_exact(!did_truncate_min),
                )
            }
            stats => stats,
        }
    }

    /// Adds data page.
    /// Data page is either buffered in case of dictionary encoding or written directly.
    pub(crate) fn add_data_page(&mut self) -> Result<()> {
        // Extract encoded values
        let values_data = self.encoder.flush_data_page()?;

        let max_def_level = self.descr.max_def_level();
        let max_rep_level = self.descr.max_rep_level();

        self.column_metrics.num_column_nulls += self.page_metrics.num_page_nulls;

        if let Some(nan_count) = values_data.nan_count {
            *self.column_metrics.num_column_nans.get_or_insert(0) += nan_count;
            self.page_metrics.num_page_nans = Some(nan_count);
        }

        let page_statistics = match (values_data.min_value, values_data.max_value) {
            (Some(min), Some(max)) => {
                // Update chunk level statistics
                update_min(&self.descr, &min, &mut self.column_metrics.min_column_value);
                update_max(&self.descr, &max, &mut self.column_metrics.max_column_value);

                (self.column_props.statistics_enabled == EnabledStatistics::Page).then_some(
                    ValueStatistics::new(
                        Some(min),
                        Some(max),
                        None,
                        Some(self.page_metrics.num_page_nulls),
                        false,
                    )
                    .with_nan_count(values_data.nan_count),
                )
            }
            _ => None,
        };

        // update column and offset index
        self.update_column_offset_index(
            page_statistics.as_ref(),
            values_data.variable_length_bytes,
        );

        // Update histograms and variable_length_bytes in column_metrics
        self.column_metrics
            .update_from_page_metrics(&self.page_metrics);
        self.column_metrics
            .update_variable_length_bytes(values_data.variable_length_bytes);

        // From here on, we only need page statistics if they will be written to the page header.
        let page_statistics = page_statistics
            .filter(|_| self.column_props.write_page_header_statistics)
            .map(|stats| self.truncate_statistics(Statistics::from(stats)));

        let compressed_page = match self.props.writer_version() {
            WriterVersion::PARQUET_1_0 => {
                let mut buffer = vec![];

                if max_rep_level > 0 {
                    self.rep_levels_encoder
                        .flush_to(|data| buffer.extend_from_slice(data));
                }

                if max_def_level > 0 {
                    self.def_levels_encoder
                        .flush_to(|data| buffer.extend_from_slice(data));
                }

                buffer.extend_from_slice(&values_data.buf);
                let uncompressed_size = buffer.len();

                if let Some(ref mut cmpr) = self.compressor {
                    let mut compressed_buf = Vec::with_capacity(uncompressed_size);
                    cmpr.compress(&buffer[..], &mut compressed_buf)?;
                    compressed_buf.shrink_to_fit();
                    buffer = compressed_buf;
                }

                let data_page = Page::DataPage {
                    buf: buffer.into(),
                    num_values: self.page_metrics.num_buffered_values,
                    encoding: values_data.encoding,
                    def_level_encoding: Encoding::RLE,
                    rep_level_encoding: Encoding::RLE,
                    statistics: page_statistics,
                };

                CompressedPage::new(data_page, uncompressed_size)
            }
            WriterVersion::PARQUET_2_0 => {
                let mut rep_levels_byte_len = 0;
                let mut def_levels_byte_len = 0;
                let mut buffer = vec![];

                if max_rep_level > 0 {
                    self.rep_levels_encoder
                        .flush_to(|data| buffer.extend_from_slice(data));
                    rep_levels_byte_len = buffer.len();
                }

                if max_def_level > 0 {
                    self.def_levels_encoder
                        .flush_to(|data| buffer.extend_from_slice(data));
                    def_levels_byte_len = buffer.len() - rep_levels_byte_len;
                }

                let uncompressed_size =
                    rep_levels_byte_len + def_levels_byte_len + values_data.buf.len();

                // Data Page v2 compresses values only.
                let is_compressed = match self.compressor {
                    Some(ref mut cmpr) => {
                        let buffer_len = buffer.len();
                        cmpr.compress(&values_data.buf, &mut buffer)?;
                        let compressed_values_size = buffer.len() - buffer_len;
                        let threshold = self.column_props.data_page_v2_compression_ratio_threshold;
                        if (compressed_values_size as f64) >= (uncompressed_size as f64) * threshold
                        {
                            buffer.truncate(buffer_len);
                            buffer.extend_from_slice(&values_data.buf);
                            false
                        } else {
                            true
                        }
                    }
                    None => {
                        buffer.extend_from_slice(&values_data.buf);
                        false
                    }
                };

                let data_page = Page::DataPageV2 {
                    buf: buffer.into(),
                    num_values: self.page_metrics.num_buffered_values,
                    encoding: values_data.encoding,
                    num_nulls: self.page_metrics.num_page_nulls as u32,
                    num_rows: self.page_metrics.num_buffered_rows,
                    def_levels_byte_len: def_levels_byte_len as u32,
                    rep_levels_byte_len: rep_levels_byte_len as u32,
                    is_compressed,
                    statistics: page_statistics,
                };

                CompressedPage::new(data_page, uncompressed_size)
            }
        };

        // Check if we need to buffer data page or flush it to the sink directly.
        //
        // For dictionary-encoded columns the dictionary page must be written
        // first, but it is not final until all values are seen, so completed
        // data pages are normally buffered here until `close`. A page writer
        // that defers final layout (the Arrow path) instead orders pages itself
        // at flush, so we stream the data pages straight through and never let
        // them accumulate in memory.
        if self.encoder.has_dictionary() && !self.page_writer.defers_dictionary_ordering() {
            self.data_pages.push_back(compressed_page);
        } else {
            self.write_data_page(compressed_page)?;
        }

        // Update total number of rows.
        self.column_metrics.total_rows_written += self.page_metrics.num_buffered_rows as u64;
        self.page_metrics.new_page();

        Ok(())
    }

    /// Finalises any outstanding data pages and flushes buffered data pages from
    /// dictionary encoding into underlying sink.
    #[inline]
    fn flush_data_pages(&mut self) -> Result<()> {
        // Write all outstanding data to a new page.
        if self.page_metrics.num_buffered_values > 0 {
            self.add_data_page()?;
        }

        while let Some(page) = self.data_pages.pop_front() {
            self.write_data_page(page)?;
        }

        Ok(())
    }

    /// Assembles column chunk metadata.
    fn build_column_metadata(&mut self) -> Result<ColumnChunkMetaData> {
        let total_compressed_size = self.column_metrics.total_compressed_size as i64;
        let total_uncompressed_size = self.column_metrics.total_uncompressed_size as i64;
        let num_values = self.column_metrics.total_num_values as i64;
        let dict_page_offset = self.column_metrics.dictionary_page_offset.map(|v| v as i64);
        // If data page offset is not set, then no pages have been written
        let data_page_offset = self.column_metrics.data_page_offset.unwrap_or(0) as i64;

        let mut builder = ColumnChunkMetaData::builder(self.descr.clone())
            .set_compression(self.codec)
            .set_encodings_mask(EncodingMask::new_from_encodings(self.encodings.iter()))
            .set_page_encoding_stats(self.encoding_stats.clone())
            .set_total_compressed_size(total_compressed_size)
            .set_total_uncompressed_size(total_uncompressed_size)
            .set_num_values(num_values)
            .set_data_page_offset(data_page_offset)
            .set_dictionary_page_offset(dict_page_offset);

        if self.column_props.statistics_enabled != EnabledStatistics::None {
            let backwards_compatible_min_max = self.descr.sort_order().is_signed();

            let distinct_count = self
                .distinct_count_override
                .or(self.column_metrics.column_distinct_count);
            let statistics = ValueStatistics::<E::T>::new(
                self.column_metrics.min_column_value.clone(),
                self.column_metrics.max_column_value.clone(),
                distinct_count,
                Some(self.column_metrics.num_column_nulls),
                false,
            )
            .with_nan_count(self.column_metrics.num_column_nans)
            .with_backwards_compatible_min_max(backwards_compatible_min_max)
            .into();

            let statistics = self.truncate_statistics(statistics);

            builder = builder
                .set_statistics(statistics)
                .set_unencoded_byte_array_data_bytes(self.column_metrics.variable_length_bytes)
                .set_repetition_level_histogram(
                    self.column_metrics.repetition_level_histogram.take(),
                )
                .set_definition_level_histogram(
                    self.column_metrics.definition_level_histogram.take(),
                );

            if let Some(geo_stats) = self.encoder.flush_geospatial_statistics() {
                builder = builder.set_geo_statistics(geo_stats);
            }
        }

        builder = self.set_column_chunk_encryption_properties(builder);

        let metadata = builder.build()?;
        Ok(metadata)
    }

    /// Writes compressed data page into underlying sink and updates global metrics.
    #[inline]
    fn write_data_page(&mut self, page: CompressedPage) -> Result<()> {
        self.encodings.insert(page.encoding());
        match self.encoding_stats.last_mut() {
            Some(encoding_stats)
                if encoding_stats.page_type == page.page_type()
                    && encoding_stats.encoding == page.encoding() =>
            {
                encoding_stats.count += 1;
            }
            _ => {
                // data page type does not change inside a file
                // encoding can currently only change from dictionary to non-dictionary once
                self.encoding_stats.push(PageEncodingStats {
                    page_type: page.page_type(),
                    encoding: page.encoding(),
                    count: 1,
                });
            }
        }
        let page_spec = self.page_writer.write_page(page)?;
        // update offset index
        // compressed_size = header_size + compressed_data_size
        if let Some(builder) = self.offset_index_builder.as_mut() {
            builder
                .append_offset_and_size(page_spec.offset as i64, page_spec.compressed_size as i32)
        }
        self.update_metrics_for_page(page_spec);
        Ok(())
    }

    /// Writes dictionary page into underlying sink.
    #[inline]
    fn write_dictionary_page(&mut self) -> Result<()> {
        let compressed_page = {
            let mut page = self
                .encoder
                .flush_dict_page()?
                .ok_or_else(|| general_err!("Dictionary encoder is not set"))?;

            let uncompressed_size = page.buf.len();

            if let Some(ref mut cmpr) = self.compressor {
                let mut output_buf = Vec::with_capacity(uncompressed_size);
                cmpr.compress(&page.buf, &mut output_buf)?;
                page.buf = Bytes::from(output_buf);
            }

            let dict_page = Page::DictionaryPage {
                buf: page.buf,
                num_values: page.num_values as u32,
                encoding: self.props.dictionary_page_encoding(),
                is_sorted: page.is_sorted,
            };
            CompressedPage::new(dict_page, uncompressed_size)
        };

        self.encodings.insert(compressed_page.encoding());
        self.encoding_stats.push(PageEncodingStats {
            page_type: PageType::DICTIONARY_PAGE,
            encoding: compressed_page.encoding(),
            count: 1,
        });
        let page_spec = self.page_writer.write_page(compressed_page)?;
        self.update_metrics_for_page(page_spec);
        // For the directory page, don't need to update column/offset index.
        Ok(())
    }

    /// Updates column writer metrics with each page metadata.
    #[inline]
    fn update_metrics_for_page(&mut self, page_spec: PageWriteSpec) {
        self.column_metrics.total_uncompressed_size += page_spec.uncompressed_size as u64;
        self.column_metrics.total_compressed_size += page_spec.compressed_size as u64;
        self.column_metrics.total_bytes_written += page_spec.bytes_written;

        match page_spec.page_type {
            PageType::DATA_PAGE | PageType::DATA_PAGE_V2 => {
                self.column_metrics.total_num_values += page_spec.num_values as u64;
                if self.column_metrics.data_page_offset.is_none() {
                    self.column_metrics.data_page_offset = Some(page_spec.offset);
                }
            }
            PageType::DICTIONARY_PAGE => {
                assert!(
                    self.column_metrics.dictionary_page_offset.is_none(),
                    "Dictionary offset is already set"
                );
                self.column_metrics.dictionary_page_offset = Some(page_spec.offset);
            }
            PageType::INDEX_PAGE => {}
        }
    }

    #[inline]
    #[cfg(feature = "encryption")]
    fn set_column_chunk_encryption_properties(
        &self,
        builder: ColumnChunkMetaDataBuilder,
    ) -> ColumnChunkMetaDataBuilder {
        if let Some(encryption_properties) = self.props.file_encryption_properties.as_ref() {
            builder.set_column_crypto_metadata(get_column_crypto_metadata(
                encryption_properties,
                &self.descr,
            ))
        } else {
            builder
        }
    }

    #[inline]
    #[cfg(not(feature = "encryption"))]
    fn set_column_chunk_encryption_properties(
        &self,
        builder: ColumnChunkMetaDataBuilder,
    ) -> ColumnChunkMetaDataBuilder {
        builder
    }
}

fn update_min<T: ParquetValueType>(descr: &ColumnDescriptor, val: &T, min: &mut Option<T>) {
    match min {
        None => *min = Some(val.clone()),
        Some(min) => {
            let basic_type_info = descr.get_basic_info();
            let is_min_nan = is_nan(basic_type_info, min);
            let is_val_nan = is_nan(basic_type_info, val);
            match (is_min_nan, is_val_nan) {
                // current min is not NaN, but incoming is NaN: skip
                (false, true) => {}
                // current min is NaN, but incoming is not: assign val to min
                (true, false) => *min = val.clone(),
                // both NaN or non-NaN, safe to call update_stat()
                _ => {
                    update_stat::<T, _>(val, min, |cur| compare_greater(basic_type_info, cur, val))
                }
            }
        }
    }
}

fn update_max<T: ParquetValueType>(descr: &ColumnDescriptor, val: &T, max: &mut Option<T>) {
    match max {
        None => *max = Some(val.clone()),
        Some(max) => {
            let basic_type_info = descr.get_basic_info();
            let is_max_nan = is_nan(basic_type_info, max);
            let is_val_nan = is_nan(basic_type_info, val);
            match (is_max_nan, is_val_nan) {
                // current max is not NaN, but incoming is NaN: skip
                (false, true) => {}
                // current max is NaN, but incoming is not: assign val to max
                (true, false) => *max = val.clone(),
                // both NaN or non-NaN, safe to call update_stat()
                _ => {
                    update_stat::<T, _>(val, max, |cur| compare_greater(basic_type_info, val, cur))
                }
            }
        }
    }
}

#[inline]
#[expect(clippy::eq_op)]
fn is_nan<T: ParquetValueType>(basic_type_info: &BasicTypeInfo, val: &T) -> bool {
    match T::PHYSICAL_TYPE {
        Type::FLOAT | Type::DOUBLE => val != val,
        Type::FIXED_LEN_BYTE_ARRAY
            if matches!(basic_type_info.sort_order(), SortOrder::TOTAL_ORDER) =>
        {
            // taken from f16 impl, but skips creating f16. just compare the bits as u16.
            let val = val.as_bytes();
            // Float16 is stored little endian
            let uval = ((val[1] as u16) << 8) | val[0] as u16;
            uval & 0x7FFFu16 > 0x7C00u16
        }
        _ => false,
    }
}

/// Perform a conditional update of `cur`
///
/// Calls `should_update` with the value of `cur`, and updates `cur` to `Some(val)` if it
/// returns `true`. `cur` must not be `None` or this will panic.
fn update_stat<T: ParquetValueType, F>(val: &T, cur: &mut T, should_update: F)
where
    F: Fn(&T) -> bool,
{
    if should_update(cur) {
        *cur = val.clone();
    }
}

/// Evaluate `a > b` according to underlying logical type.
fn compare_greater<T: ParquetValueType>(basic_type_info: &BasicTypeInfo, a: &T, b: &T) -> bool {
    match T::PHYSICAL_TYPE {
        Type::FLOAT => {
            let a = f32::from_le_bytes(a.as_bytes().try_into().unwrap());
            let b = f32::from_le_bytes(b.as_bytes().try_into().unwrap());
            return a.total_cmp(&b) == Ordering::Greater;
        }
        Type::DOUBLE => {
            let a = f64::from_le_bytes(a.as_bytes().try_into().unwrap());
            let b = f64::from_le_bytes(b.as_bytes().try_into().unwrap());
            return a.total_cmp(&b) == Ordering::Greater;
        }
        Type::INT32 | Type::INT64
            if matches!(basic_type_info.sort_order(), SortOrder::UNSIGNED) =>
        {
            return compare_greater_unsigned_int(a, b);
        }
        Type::FIXED_LEN_BYTE_ARRAY
            if matches!(basic_type_info.sort_order(), SortOrder::TOTAL_ORDER) =>
        {
            return compare_greater_f16(a.as_bytes(), b.as_bytes());
        }
        Type::FIXED_LEN_BYTE_ARRAY | Type::BYTE_ARRAY
            if matches!(basic_type_info.converted_type(), ConvertedType::DECIMAL)
                || matches!(
                    basic_type_info.logical_type_ref(),
                    Some(LogicalType::Decimal(_))
                ) =>
        {
            return compare_greater_byte_array_decimals(a.as_bytes(), b.as_bytes());
        }

        _ => {}
    }

    // compare independent of logical / converted type
    a > b
}

// ----------------------------------------------------------------------
// Encoding support for column writer.
// This mirrors parquet-mr default encodings for writes. See:
// https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV1ValuesWriterFactory.java
// https://github.com/apache/parquet-mr/blob/master/parquet-column/src/main/java/org/apache/parquet/column/values/factory/DefaultV2ValuesWriterFactory.java

/// Returns encoding for a column when no other encoding is provided in writer properties.
fn fallback_encoding(kind: Type, props: &WriterProperties) -> Encoding {
    match (kind, props.writer_version()) {
        (Type::BOOLEAN, WriterVersion::PARQUET_2_0) => Encoding::RLE,
        (Type::INT32, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BINARY_PACKED,
        (Type::INT64, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BINARY_PACKED,
        (Type::BYTE_ARRAY, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BYTE_ARRAY,
        (Type::FIXED_LEN_BYTE_ARRAY, WriterVersion::PARQUET_2_0) => Encoding::DELTA_BYTE_ARRAY,
        _ => Encoding::PLAIN,
    }
}

/// Returns true if dictionary is supported for column writer, false otherwise.
fn has_dictionary_support(kind: Type, props: &WriterProperties) -> bool {
    match (kind, props.writer_version()) {
        // Booleans do not support dict encoding and should use a fallback encoding.
        (Type::BOOLEAN, _) => false,
        // Dictionary encoding was not enabled in PARQUET 1.0
        (Type::FIXED_LEN_BYTE_ARRAY, WriterVersion::PARQUET_1_0) => false,
        (Type::FIXED_LEN_BYTE_ARRAY, WriterVersion::PARQUET_2_0) => true,
        _ => true,
    }
}

#[inline]
fn compare_greater_unsigned_int<T: ParquetValueType>(a: &T, b: &T) -> bool {
    a.as_u64().unwrap() > b.as_u64().unwrap()
}

#[inline]
fn compare_greater_f16(a: &[u8], b: &[u8]) -> bool {
    let a = f16::from_le_bytes(a.try_into().unwrap());
    let b = f16::from_le_bytes(b.try_into().unwrap());
    a.total_cmp(&b) == Ordering::Greater
}

/// Signed comparison of bytes arrays
fn compare_greater_byte_array_decimals(a: &[u8], b: &[u8]) -> bool {
    let a_length = a.len();
    let b_length = b.len();

    if a_length == 0 || b_length == 0 {
        return a_length > 0;
    }

    let first_a: u8 = a[0];
    let first_b: u8 = b[0];

    // We can short circuit for different signed numbers or
    // for equal length bytes arrays that have different first bytes.
    // The equality requirement is necessary for sign extension cases.
    // 0xFF10 should be equal to 0x10 (due to big endian sign extension).
    if (0x80 & first_a) != (0x80 & first_b) || (a_length == b_length && first_a != first_b) {
        return (first_a as i8) > (first_b as i8);
    }

    // When the lengths are unequal and the numbers are of the same sign,
    // sign-extend the shorter value: if any of the longer value's extra
    // leading bytes differs from the sign-extension byte it has the larger
    // magnitude, and otherwise those bytes are redundant and the aligned
    // equal-length tails decide via unsigned lexicographical comparison.

    let extension: u8 = if (first_a as i8) < 0 { 0xFF } else { 0 };

    if a_length != b_length {
        let not_equal = if a_length > b_length {
            let lead_length = a_length - b_length;
            a[0..lead_length].iter().any(|&x| x != extension)
        } else {
            let lead_length = b_length - a_length;
            b[0..lead_length].iter().any(|&x| x != extension)
        };

        if not_equal {
            let negative_values: bool = (first_a as i8) < 0;
            let a_longer: bool = a_length > b_length;
            return if negative_values { !a_longer } else { a_longer };
        }
    }

    let tail_length = a_length.min(b_length);
    (a[a_length - tail_length..]) > (b[b_length - tail_length..])
}

/// Truncate a UTF-8 slice to the longest prefix that is still a valid UTF-8 string,
/// while being less than `length` bytes and non-empty. Returns `None` if truncation
/// is not possible within those constraints.
///
/// The caller guarantees that data.len() > length.
fn truncate_utf8(data: &str, length: usize) -> Option<Vec<u8>> {
    let split = (1..=length).rfind(|x| data.is_char_boundary(*x))?;
    Some(data.as_bytes()[..split].to_vec())
}

/// Truncate a UTF-8 slice and increment it's final character. The returned value is the
/// longest such slice that is still a valid UTF-8 string while being less than `length`
/// bytes and non-empty. Returns `None` if no such transformation is possible.
///
/// The caller guarantees that data.len() > length.
fn truncate_and_increment_utf8(data: &str, length: usize) -> Option<Vec<u8>> {
    // UTF-8 is max 4 bytes, so start search 3 back from desired length
    let lower_bound = length.saturating_sub(3);
    let split = (lower_bound..=length).rfind(|x| data.is_char_boundary(*x))?;
    increment_utf8(data.get(..split)?)
}

/// Increment the final character in a UTF-8 string in such a way that the returned result
/// is still a valid UTF-8 string. The returned string may be shorter than the input if the
/// last character(s) cannot be incremented (due to overflow or producing invalid code points).
/// Returns `None` if the string cannot be incremented.
///
/// Note that this implementation will not promote an N-byte code point to (N+1) bytes.
fn increment_utf8(data: &str) -> Option<Vec<u8>> {
    for (idx, original_char) in data.char_indices().rev() {
        let original_len = original_char.len_utf8();
        if let Some(next_char) = char::from_u32(original_char as u32 + 1) {
            // do not allow increasing byte width of incremented char
            if next_char.len_utf8() == original_len {
                let mut result = data.as_bytes()[..idx + original_len].to_vec();
                next_char.encode_utf8(&mut result[idx..]);
                return Some(result);
            }
        }
    }

    None
}

/// Try and increment the bytes from right to left.
///
/// Returns `None` if all bytes are set to `u8::MAX`.
fn increment(mut data: Vec<u8>) -> Option<Vec<u8>> {
    for byte in data.iter_mut().rev() {
        let (incremented, overflow) = byte.overflowing_add(1);
        *byte = incremented;

        if !overflow {
            return Some(data);
        }
    }

    None
}

#[cfg(test)]
mod tests {
    use crate::{
        file::{properties::DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH, writer::SerializedFileWriter},
        schema::parser::parse_message_type,
    };
    use core::str;
    use rand::distr::uniform::SampleUniform;
    use std::{fs::File, sync::Arc};

    use crate::column::{
        page::PageReader,
        reader::{ColumnReaderImpl, get_column_reader, get_typed_column_reader},
    };
    use crate::file::writer::TrackedWrite;
    use crate::file::{
        properties::ReaderProperties, reader::SerializedPageReader, writer::SerializedPageWriter,
    };
    use crate::schema::types::{ColumnPath, Type as SchemaType};
    use crate::util::test_common::rand_gen::random_numbers_range;

    use super::*;

    #[test]
    fn test_column_writer_inconsistent_def_rep_length() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 1, props);
        let res = writer.write_batch(&[1, 2, 3, 4], Some(&[1, 1, 1]), Some(&[0, 0]));
        assert!(res.is_err());
        if let Err(err) = res {
            assert_eq!(
                format!("{err}"),
                "Parquet error: Inconsistent length of definition and repetition levels: 3 != 2"
            );
        }
    }

    #[test]
    fn test_column_writer_invalid_def_levels() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 0, props);
        let res = writer.write_batch(&[1, 2, 3, 4], None, None);
        assert!(res.is_err());
        if let Err(err) = res {
            assert_eq!(
                format!("{err}"),
                "Parquet error: Definition levels are required, because max definition level = 1"
            );
        }
    }

    #[test]
    fn test_column_writer_invalid_rep_levels() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 1, props);
        let res = writer.write_batch(&[1, 2, 3, 4], None, None);
        assert!(res.is_err());
        if let Err(err) = res {
            assert_eq!(
                format!("{err}"),
                "Parquet error: Repetition levels are required, because max repetition level = 1"
            );
        }
    }

    #[test]
    fn test_column_writer_not_enough_values_to_write() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 0, props);
        let res = writer.write_batch(&[1, 2], Some(&[1, 1, 1, 1]), None);
        assert!(res.is_err());
        if let Err(err) = res {
            assert_eq!(
                format!("{err}"),
                "Parquet error: Expected to write 4 values, but have only 2"
            );
        }
    }

    #[test]
    fn test_column_writer_write_only_one_dictionary_page() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();
        // First page should be correctly written.
        writer.add_data_page().unwrap();
        writer.write_dictionary_page().unwrap();
        let err = writer.write_dictionary_page().unwrap_err().to_string();
        assert_eq!(err, "Parquet error: Dictionary encoder is not set");
    }

    #[test]
    fn test_column_writer_error_when_writing_disabled_dictionary() {
        let page_writer = get_test_page_writer();
        let props = Arc::new(
            WriterProperties::builder()
                .set_dictionary_enabled(false)
                .build(),
        );
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();
        let err = writer.write_dictionary_page().unwrap_err().to_string();
        assert_eq!(err, "Parquet error: Dictionary encoder is not set");
    }

    #[test]
    fn test_column_writer_boolean_type_does_not_support_dictionary() {
        let page_writer = get_test_page_writer();
        let props = Arc::new(
            WriterProperties::builder()
                .set_dictionary_enabled(true)
                .build(),
        );
        let mut writer = get_test_column_writer::<BoolType>(page_writer, 0, 0, props);
        writer
            .write_batch(&[true, false, true, false], None, None)
            .unwrap();

        let r = writer.close().unwrap();
        // PlainEncoder uses bit writer to write boolean values, which all fit into 1
        // byte.
        assert_eq!(r.bytes_written, 1);
        assert_eq!(r.rows_written, 4);

        let metadata = r.metadata;
        assert_eq!(
            metadata.encodings().collect::<Vec<_>>(),
            vec![Encoding::PLAIN, Encoding::RLE]
        );
        assert_eq!(metadata.num_values(), 4); // just values
        assert_eq!(metadata.dictionary_page_offset(), None);
    }

    #[test]
    fn test_bloom_filter_for_dictionary_encoded_chunks() {
        fn bloom_filter_written(dictionary_enabled: bool, for_dictionary_chunks: bool) -> bool {
            let props = Arc::new(
                WriterProperties::builder()
                    .set_dictionary_enabled(dictionary_enabled)
                    .set_bloom_filter_enabled(true)
                    .set_bloom_filter_for_dictionary_encoded_chunks(for_dictionary_chunks)
                    .build(),
            );
            let mut writer =
                get_test_column_writer::<Int32Type>(get_test_page_writer(), 0, 0, props);
            writer.write_batch(&[1, 2, 1, 2], None, None).unwrap();
            writer.close().unwrap().bloom_filter.is_some()
        }

        assert!(bloom_filter_written(true, true));
        assert!(bloom_filter_written(false, true));
        assert!(bloom_filter_written(false, false));
        assert!(!bloom_filter_written(true, false));
    }

    #[test]
    fn test_column_writer_default_encoding_support_bool() {
        check_encoding_write_support::<BoolType>(
            WriterVersion::PARQUET_1_0,
            true,
            &[true, false],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<BoolType>(
            WriterVersion::PARQUET_1_0,
            false,
            &[true, false],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<BoolType>(
            WriterVersion::PARQUET_2_0,
            true,
            &[true, false],
            None,
            &[Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE, 1)],
        );
        check_encoding_write_support::<BoolType>(
            WriterVersion::PARQUET_2_0,
            false,
            &[true, false],
            None,
            &[Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE, 1)],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_int32() {
        check_encoding_write_support::<Int32Type>(
            WriterVersion::PARQUET_1_0,
            true,
            &[1, 2],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int32Type>(
            WriterVersion::PARQUET_1_0,
            false,
            &[1, 2],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<Int32Type>(
            WriterVersion::PARQUET_2_0,
            true,
            &[1, 2],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int32Type>(
            WriterVersion::PARQUET_2_0,
            false,
            &[1, 2],
            None,
            &[Encoding::RLE, Encoding::DELTA_BINARY_PACKED],
            &[encoding_stats(
                PageType::DATA_PAGE_V2,
                Encoding::DELTA_BINARY_PACKED,
                1,
            )],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_int64() {
        check_encoding_write_support::<Int64Type>(
            WriterVersion::PARQUET_1_0,
            true,
            &[1, 2],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int64Type>(
            WriterVersion::PARQUET_1_0,
            false,
            &[1, 2],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<Int64Type>(
            WriterVersion::PARQUET_2_0,
            true,
            &[1, 2],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int64Type>(
            WriterVersion::PARQUET_2_0,
            false,
            &[1, 2],
            None,
            &[Encoding::RLE, Encoding::DELTA_BINARY_PACKED],
            &[encoding_stats(
                PageType::DATA_PAGE_V2,
                Encoding::DELTA_BINARY_PACKED,
                1,
            )],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_int96() {
        check_encoding_write_support::<Int96Type>(
            WriterVersion::PARQUET_1_0,
            true,
            &[Int96::from(vec![1, 2, 3])],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int96Type>(
            WriterVersion::PARQUET_1_0,
            false,
            &[Int96::from(vec![1, 2, 3])],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<Int96Type>(
            WriterVersion::PARQUET_2_0,
            true,
            &[Int96::from(vec![1, 2, 3])],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<Int96Type>(
            WriterVersion::PARQUET_2_0,
            false,
            &[Int96::from(vec![1, 2, 3])],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE_V2, Encoding::PLAIN, 1)],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_float() {
        check_encoding_write_support::<FloatType>(
            WriterVersion::PARQUET_1_0,
            true,
            &[1.0, 2.0],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<FloatType>(
            WriterVersion::PARQUET_1_0,
            false,
            &[1.0, 2.0],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<FloatType>(
            WriterVersion::PARQUET_2_0,
            true,
            &[1.0, 2.0],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<FloatType>(
            WriterVersion::PARQUET_2_0,
            false,
            &[1.0, 2.0],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE_V2, Encoding::PLAIN, 1)],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_double() {
        check_encoding_write_support::<DoubleType>(
            WriterVersion::PARQUET_1_0,
            true,
            &[1.0, 2.0],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<DoubleType>(
            WriterVersion::PARQUET_1_0,
            false,
            &[1.0, 2.0],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<DoubleType>(
            WriterVersion::PARQUET_2_0,
            true,
            &[1.0, 2.0],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<DoubleType>(
            WriterVersion::PARQUET_2_0,
            false,
            &[1.0, 2.0],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE_V2, Encoding::PLAIN, 1)],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_byte_array() {
        check_encoding_write_support::<ByteArrayType>(
            WriterVersion::PARQUET_1_0,
            true,
            &[ByteArray::from(vec![1u8])],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<ByteArrayType>(
            WriterVersion::PARQUET_1_0,
            false,
            &[ByteArray::from(vec![1u8])],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<ByteArrayType>(
            WriterVersion::PARQUET_2_0,
            true,
            &[ByteArray::from(vec![1u8])],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<ByteArrayType>(
            WriterVersion::PARQUET_2_0,
            false,
            &[ByteArray::from(vec![1u8])],
            None,
            &[Encoding::RLE, Encoding::DELTA_BYTE_ARRAY],
            &[encoding_stats(
                PageType::DATA_PAGE_V2,
                Encoding::DELTA_BYTE_ARRAY,
                1,
            )],
        );
    }

    #[test]
    fn test_column_writer_default_encoding_support_fixed_len_byte_array() {
        check_encoding_write_support::<FixedLenByteArrayType>(
            WriterVersion::PARQUET_1_0,
            true,
            &[ByteArray::from(vec![1u8]).into()],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<FixedLenByteArrayType>(
            WriterVersion::PARQUET_1_0,
            false,
            &[ByteArray::from(vec![1u8]).into()],
            None,
            &[Encoding::PLAIN, Encoding::RLE],
            &[encoding_stats(PageType::DATA_PAGE, Encoding::PLAIN, 1)],
        );
        check_encoding_write_support::<FixedLenByteArrayType>(
            WriterVersion::PARQUET_2_0,
            true,
            &[ByteArray::from(vec![1u8]).into()],
            Some(0),
            &[Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY],
            &[
                encoding_stats(PageType::DICTIONARY_PAGE, Encoding::PLAIN, 1),
                encoding_stats(PageType::DATA_PAGE_V2, Encoding::RLE_DICTIONARY, 1),
            ],
        );
        check_encoding_write_support::<FixedLenByteArrayType>(
            WriterVersion::PARQUET_2_0,
            false,
            &[ByteArray::from(vec![1u8]).into()],
            None,
            &[Encoding::RLE, Encoding::DELTA_BYTE_ARRAY],
            &[encoding_stats(
                PageType::DATA_PAGE_V2,
                Encoding::DELTA_BYTE_ARRAY,
                1,
            )],
        );
    }

    #[test]
    fn test_column_writer_check_metadata() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();

        let r = writer.close().unwrap();
        assert_eq!(r.bytes_written, 20);
        assert_eq!(r.rows_written, 4);

        let metadata = r.metadata;
        assert_eq!(
            metadata.encodings().collect::<Vec<_>>(),
            vec![Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY]
        );
        assert_eq!(metadata.num_values(), 4);
        assert_eq!(metadata.compressed_size(), 20);
        assert_eq!(metadata.uncompressed_size(), 20);
        assert_eq!(metadata.data_page_offset(), 0);
        assert_eq!(metadata.dictionary_page_offset(), Some(0));
        if let Some(stats) = metadata.statistics() {
            assert_eq!(stats.null_count_opt(), Some(0));
            assert_eq!(stats.distinct_count_opt(), None);
            if let Statistics::Int32(stats) = stats {
                assert_eq!(stats.min_opt().unwrap(), &1);
                assert_eq!(stats.max_opt().unwrap(), &4);
            } else {
                panic!("expecting Statistics::Int32");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_column_writer_check_byte_array_min_max() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_decimals_column_writer::<ByteArrayType>(page_writer, 0, 0, props);
        writer
            .write_batch(
                &[
                    ByteArray::from(vec![
                        255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 179u8, 172u8, 19u8,
                        35u8, 231u8, 90u8, 0u8, 0u8,
                    ]),
                    ByteArray::from(vec![
                        255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 228u8, 62u8, 146u8,
                        152u8, 177u8, 56u8, 0u8, 0u8,
                    ]),
                    ByteArray::from(vec![
                        0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
                        0u8,
                    ]),
                    ByteArray::from(vec![
                        0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 41u8, 162u8, 36u8, 26u8, 246u8,
                        44u8, 0u8, 0u8,
                    ]),
                ],
                None,
                None,
            )
            .unwrap();
        let metadata = writer.close().unwrap().metadata;
        if let Some(stats) = metadata.statistics() {
            if let Statistics::ByteArray(stats) = stats {
                assert_eq!(
                    stats.min_opt().unwrap(),
                    &ByteArray::from(vec![
                        255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 179u8, 172u8, 19u8,
                        35u8, 231u8, 90u8, 0u8, 0u8,
                    ])
                );
                assert_eq!(
                    stats.max_opt().unwrap(),
                    &ByteArray::from(vec![
                        0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 41u8, 162u8, 36u8, 26u8, 246u8,
                        44u8, 0u8, 0u8,
                    ])
                );
            } else {
                panic!("expecting Statistics::ByteArray");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_column_writer_byte_array_min_max_unequal_lengths() {
        // Byte-array decimal min/max with values of different encoded lengths
        // https://github.com/apache/arrow-rs/issues/10860
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_decimals_column_writer::<ByteArrayType>(page_writer, 0, 0, props);
        writer
            .write_batch(
                &[
                    ByteArray::from(vec![0u8, 255u8]),      // 255
                    ByteArray::from(vec![0u8, 128u8, 0u8]), // 32768
                    ByteArray::from(vec![255u8, 127u8]),    // -129
                    ByteArray::from(vec![128u8]),           // -128
                ],
                None,
                None,
            )
            .unwrap();
        let metadata = writer.close().unwrap().metadata;
        let stats = metadata.statistics().expect("metadata missing statistics");
        let Statistics::ByteArray(stats) = stats else {
            panic!("expecting Statistics::ByteArray");
        };
        // -129
        assert_eq!(
            stats.min_opt().unwrap(),
            &ByteArray::from(vec![255u8, 127u8])
        );
        // 32768
        assert_eq!(
            stats.max_opt().unwrap(),
            &ByteArray::from(vec![0u8, 128u8, 0u8])
        );
    }

    #[test]
    fn test_column_writer_uint32_converted_type_min_max() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_unsigned_int_given_as_converted_column_writer::<Int32Type>(
            page_writer,
            0,
            0,
            props,
        );
        writer.write_batch(&[0, 1, 2, 3, 4, 5], None, None).unwrap();
        let metadata = writer.close().unwrap().metadata;
        if let Some(stats) = metadata.statistics() {
            if let Statistics::Int32(stats) = stats {
                assert_eq!(stats.min_opt().unwrap(), &0,);
                assert_eq!(stats.max_opt().unwrap(), &5,);
            } else {
                panic!("expecting Statistics::Int32");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_column_writer_precalculated_statistics() {
        let page_writer = get_test_page_writer();
        let props = Arc::new(
            WriterProperties::builder()
                .set_statistics_enabled(EnabledStatistics::Chunk)
                .build(),
        );
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer
            .write_batch_with_statistics(
                &[1, 2, 3, 4],
                None,
                None,
                Some(&-17),
                Some(&9000),
                Some(55),
            )
            .unwrap();

        let r = writer.close().unwrap();
        assert_eq!(r.bytes_written, 20);
        assert_eq!(r.rows_written, 4);

        let metadata = r.metadata;
        assert_eq!(
            metadata.encodings().collect::<Vec<_>>(),
            vec![Encoding::PLAIN, Encoding::RLE, Encoding::RLE_DICTIONARY]
        );
        assert_eq!(metadata.num_values(), 4);
        assert_eq!(metadata.compressed_size(), 20);
        assert_eq!(metadata.uncompressed_size(), 20);
        assert_eq!(metadata.data_page_offset(), 0);
        assert_eq!(metadata.dictionary_page_offset(), Some(0));
        if let Some(stats) = metadata.statistics() {
            assert_eq!(stats.null_count_opt(), Some(0));
            assert_eq!(stats.distinct_count_opt().unwrap_or(0), 55);
            if let Statistics::Int32(stats) = stats {
                assert_eq!(stats.min_opt().unwrap(), &-17);
                assert_eq!(stats.max_opt().unwrap(), &9000);
            } else {
                panic!("expecting Statistics::Int32");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_mixed_precomputed_statistics() {
        let mut buf = Vec::with_capacity(100);
        let mut write = TrackedWrite::new(&mut buf);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));
        let props = Arc::new(
            WriterProperties::builder()
                .set_write_page_header_statistics(true)
                .build(),
        );
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);

        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();
        writer
            .write_batch_with_statistics(&[5, 6, 7], None, None, Some(&5), Some(&7), Some(3))
            .unwrap();

        let r = writer.close().unwrap();

        let stats = r.metadata.statistics().unwrap();
        assert_eq!(stats.min_bytes_opt().unwrap(), 1_i32.to_le_bytes());
        assert_eq!(stats.max_bytes_opt().unwrap(), 7_i32.to_le_bytes());
        assert_eq!(stats.null_count_opt(), Some(0));
        assert!(stats.distinct_count_opt().is_none());

        drop(write);

        let props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .set_read_page_statistics(true)
            .build();
        let reader = SerializedPageReader::new_with_properties(
            Arc::new(Bytes::from(buf)),
            &r.metadata,
            r.rows_written as usize,
            None,
            Arc::new(props),
        )
        .unwrap();

        let pages = reader.collect::<Result<Vec<_>>>().unwrap();
        assert_eq!(pages.len(), 2);

        assert_eq!(pages[0].page_type(), PageType::DICTIONARY_PAGE);
        assert_eq!(pages[1].page_type(), PageType::DATA_PAGE);

        let page_statistics = pages[1].statistics().unwrap();
        assert_eq!(
            page_statistics.min_bytes_opt().unwrap(),
            1_i32.to_le_bytes()
        );
        assert_eq!(
            page_statistics.max_bytes_opt().unwrap(),
            7_i32.to_le_bytes()
        );
        assert_eq!(page_statistics.null_count_opt(), Some(0));
        assert!(page_statistics.distinct_count_opt().is_none());
    }

    #[test]
    fn test_disabled_statistics() {
        let mut buf = Vec::with_capacity(100);
        let mut write = TrackedWrite::new(&mut buf);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));
        let props = WriterProperties::builder()
            .set_statistics_enabled(EnabledStatistics::None)
            .set_writer_version(WriterVersion::PARQUET_2_0)
            .build();
        let props = Arc::new(props);

        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 0, props);
        writer
            .write_batch(&[1, 2, 3, 4], Some(&[1, 0, 0, 1, 1, 1]), None)
            .unwrap();

        let r = writer.close().unwrap();
        assert!(r.metadata.statistics().is_none());

        drop(write);

        let props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .build();
        let reader = SerializedPageReader::new_with_properties(
            Arc::new(Bytes::from(buf)),
            &r.metadata,
            r.rows_written as usize,
            None,
            Arc::new(props),
        )
        .unwrap();

        let pages = reader.collect::<Result<Vec<_>>>().unwrap();
        assert_eq!(pages.len(), 2);

        assert_eq!(pages[0].page_type(), PageType::DICTIONARY_PAGE);
        assert_eq!(pages[1].page_type(), PageType::DATA_PAGE_V2);

        match &pages[1] {
            Page::DataPageV2 {
                num_values,
                num_nulls,
                num_rows,
                statistics,
                ..
            } => {
                assert_eq!(*num_values, 6);
                assert_eq!(*num_nulls, 2);
                assert_eq!(*num_rows, 6);
                assert!(statistics.is_none());
            }
            _ => unreachable!(),
        }
    }

    #[test]
    fn test_column_writer_empty_column_roundtrip() {
        let props = Default::default();
        column_roundtrip::<Int32Type>(props, &[], None, None);
    }

    #[test]
    fn test_column_writer_non_nullable_values_roundtrip() {
        let props = Default::default();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 0, 0);
    }

    #[test]
    fn test_column_writer_nullable_non_repeated_values_roundtrip() {
        let props = Default::default();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 0);
    }

    #[test]
    fn test_column_writer_nullable_repeated_values_roundtrip() {
        let props = Default::default();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    fn test_column_writer_dictionary_fallback_small_data_page() {
        let props = WriterProperties::builder()
            .set_dictionary_page_size_limit(32)
            .set_data_page_size_limit(32)
            .build();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    #[cfg_attr(miri, ignore)] // Takes too long
    fn test_column_writer_small_write_batch_size() {
        for i in &[1usize, 2, 5, 10, 11, 1023] {
            let props = WriterProperties::builder().set_write_batch_size(*i).build();

            column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 10);
        }
    }

    #[test]
    fn test_column_writer_dictionary_disabled_v1() {
        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .build();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    fn test_column_writer_dictionary_disabled_v2() {
        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_2_0)
            .set_dictionary_enabled(false)
            .build();
        column_roundtrip_random::<Int32Type>(props, 1024, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    fn test_column_writer_compression_v1() {
        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_compression(Compression::SNAPPY)
            .build();
        column_roundtrip_random::<Int32Type>(props, 2048, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    fn test_column_writer_compression_v2() {
        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_2_0)
            .set_compression(Compression::SNAPPY)
            .build();
        column_roundtrip_random::<Int32Type>(props, 2048, i32::MIN, i32::MAX, 10, 10);
    }

    #[test]
    fn test_column_writer_v2_compression_ratio_threshold() {
        fn write_v2_page(threshold: f64) -> bool {
            let mut buf = Vec::with_capacity(4096);
            let mut write = TrackedWrite::new(&mut buf);
            let page_writer = Box::new(SerializedPageWriter::new(&mut write));
            let props = Arc::new(
                WriterProperties::builder()
                    .set_writer_version(WriterVersion::PARQUET_2_0)
                    .set_compression(Compression::SNAPPY)
                    .set_dictionary_enabled(false)
                    .set_data_page_v2_compression_ratio_threshold(threshold)
                    .build(),
            );

            let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
            let values: Vec<i32> = vec![42; 4096];
            writer.write_batch(&values, None, None).unwrap();
            let r = writer.close().unwrap();
            drop(write);

            let reader_props = ReaderProperties::builder()
                .set_backward_compatible_lz4(false)
                .build();
            let reader = SerializedPageReader::new_with_properties(
                Arc::new(Bytes::from(buf)),
                &r.metadata,
                r.rows_written as usize,
                None,
                Arc::new(reader_props),
            )
            .unwrap();
            let pages = reader.collect::<Result<Vec<_>>>().unwrap();
            let data_page = pages
                .iter()
                .find(|p| p.page_type() == PageType::DATA_PAGE_V2)
                .expect("expected a v2 data page");
            match data_page {
                Page::DataPageV2 { is_compressed, .. } => *is_compressed,
                _ => unreachable!(),
            }
        }

        // Default threshold keeps the compressed buffer for constant data.
        assert!(write_v2_page(1.0));
        // A strict threshold (require >1000x reduction) discards it.
        assert!(!write_v2_page(0.001));
    }

    #[test]
    fn test_column_writer_add_data_pages_with_dict() {
        // ARROW-5129: Test verifies that we add data page in case of dictionary encoding
        // and no fallback occurred so far.
        let mut file = tempfile::tempfile().unwrap();
        let mut write = TrackedWrite::new(&mut file);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));
        let props = Arc::new(
            WriterProperties::builder()
                .set_data_page_size_limit(10)
                .set_write_batch_size(3) // write 3 values at a time
                .build(),
        );
        let data = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer.write_batch(data, None, None).unwrap();
        let r = writer.close().unwrap();

        drop(write);

        // Read pages and check the sequence
        let props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .build();
        let mut page_reader = Box::new(
            SerializedPageReader::new_with_properties(
                Arc::new(file),
                &r.metadata,
                r.rows_written as usize,
                None,
                Arc::new(props),
            )
            .unwrap(),
        );
        let mut res = Vec::new();
        while let Some(page) = page_reader.get_next_page().unwrap() {
            res.push((page.page_type(), page.num_values(), page.buffer().len()));
        }
        assert_eq!(
            res,
            vec![
                (PageType::DICTIONARY_PAGE, 10, 40),
                (PageType::DATA_PAGE, 9, 10),
                (PageType::DATA_PAGE, 1, 3),
            ]
        );
        assert_eq!(
            r.metadata.page_encoding_stats(),
            Some(&vec![
                PageEncodingStats {
                    page_type: PageType::DICTIONARY_PAGE,
                    encoding: Encoding::PLAIN,
                    count: 1
                },
                PageEncodingStats {
                    page_type: PageType::DATA_PAGE,
                    encoding: Encoding::RLE_DICTIONARY,
                    count: 2,
                }
            ])
        );
    }

    #[test]
    fn test_column_writer_column_data_page_size_limit() {
        let props = Arc::new(
            WriterProperties::builder()
                .set_writer_version(WriterVersion::PARQUET_1_0)
                .set_dictionary_enabled(false)
                .set_data_page_size_limit(1000)
                .set_column_data_page_size_limit(ColumnPath::from("col"), 10)
                .set_write_batch_size(3)
                .build(),
        );
        let data = &[1, 2, 3, 4, 5, 6, 7, 8, 9, 10];

        let col_values =
            write_and_collect_page_values(ColumnPath::from("col"), Arc::clone(&props), data);
        let other_values = write_and_collect_page_values(ColumnPath::from("other"), props, data);

        assert_eq!(col_values, vec![3, 3, 3, 1]);
        assert_eq!(other_values, vec![10]);
    }

    #[test]
    fn test_column_writer_caps_page_size_for_large_byte_array_values() {
        // Regression: the post-write data page byte limit check only fires
        // at mini-batch boundaries, so a 1024-row mini-batch of multi-MiB
        // BYTE_ARRAY values used to buffer multiple GiB into a single page
        // before the limit was even consulted. With the threshold-based
        // granular mode this batch should split into ~one page per value.
        let value_size = 64 * 1024; // 64 KiB per value
        let page_byte_limit = 16 * 1024; // 16 KiB page limit
        let num_rows = 64;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::PLAIN)
            .set_data_page_size_limit(page_byte_limit)
            // Default write_batch_size (1024) — without the fix this
            // buffers the entire input into a single ~4 MiB page.
            .build();

        let data: Vec<_> = (0..num_rows)
            .map(|i| ByteArray::from(vec![i as u8; value_size]))
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0, &data, None, None);

        // Every value must end up somewhere.
        let total_values: u32 = pages.data_pages.iter().map(|(_, n)| n).sum();
        assert_eq!(total_values as usize, num_rows);
        // Without the fix this assertion fired with one ~4 MiB page; the
        // threshold splits the input so that no page holds more than a
        // single oversized value's worth of bytes.
        assert!(
            pages.data_pages.len() >= num_rows / 2,
            "expected pages to be cut close to one per value, got {:?}",
            pages.data_pages,
        );
        // Each page must be bounded by roughly one value's worth of bytes;
        // parquet allows a single oversized value to occupy a page by
        // itself but never lets us pile many of them together.
        for (size, _) in &pages.data_pages {
            assert!(
                *size <= value_size + 64,
                "page size {size} exceeds one-value bound ({}B) — pages {:?}",
                value_size + 64,
                pages.data_pages,
            );
        }
    }

    #[test]
    fn test_column_writer_delta_byte_array_dedups_large_shared_prefix_values() {
        // Regression for https://github.com/apache/arrow-rs/issues/10489.
        // 16 identical 64 KiB values against a 16 KiB page limit: every value
        // is over the limit on its own, and `DELTA_BYTE_ARRAY` should still
        // dedup them down to about one value's worth of bytes in total.
        let value_size = 64 * 1024; // 64 KiB per value, > the page limit
        let page_byte_limit = 16 * 1024;
        let num_rows = 16;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::DELTA_BYTE_ARRAY)
            .set_data_page_size_limit(page_byte_limit)
            .set_statistics_enabled(EnabledStatistics::None)
            .build();

        // Identical values: one full value plus `num_rows - 1` zero-length
        // suffixes is all this column should cost.
        let data: Vec<_> = (0..num_rows)
            .map(|_| ByteArray::from(vec![b'a'; value_size]))
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0, &data, None, None);

        // Every value must still end up somewhere.
        let total_values: u32 = pages.data_pages.iter().map(|(_, n)| n).sum();
        assert_eq!(total_values as usize, num_rows);

        // Before the fix this was `num_rows * value_size` — byte for byte
        // what PLAIN produces, i.e. the encoding doing no work at all.
        let total_bytes: usize = pages.data_pages.iter().map(|(size, _)| size).sum();
        assert!(
            total_bytes < 2 * value_size,
            "expected under 2x a single value ({}B) for {num_rows} identical \
             values, got {total_bytes}B across pages {:?}",
            2 * value_size,
            pages.data_pages,
        );
    }

    #[test]
    fn test_column_writer_delta_byte_array_bounds_pages_without_shared_prefix() {
        // Companion to the test above: same shape, but the values share no
        // prefix, so there is nothing to dedup and pages must stay bounded
        // by the value size. This is why the exemption covers one value
        // rather than dropping the byte budget altogether.
        let value_size = 64 * 1024;
        let page_byte_limit = 16 * 1024;
        let num_rows = 16;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::DELTA_BYTE_ARRAY)
            .set_data_page_size_limit(page_byte_limit)
            .set_statistics_enabled(EnabledStatistics::None)
            .build();

        // No two values share a prefix: they differ at the first byte.
        let data: Vec<_> = (0..num_rows)
            .map(|i| ByteArray::from(vec![i as u8; value_size]))
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0, &data, None, None);

        let total_values: u32 = pages.data_pages.iter().map(|(_, n)| n).sum();
        assert_eq!(total_values as usize, num_rows);

        // Expect at most two values per page: the exempted first value plus
        // one more that trips the budget.
        let upper_bound = 2 * value_size + 64;
        for (size, n_values) in &pages.data_pages {
            assert!(
                *size <= upper_bound,
                "page size {size} exceeds two-value bound ({upper_bound}B); pages {:?}",
                pages.data_pages,
            );
            assert!(
                *n_values <= 2,
                "page holds {n_values} values, expected at most 2; pages {:?}",
                pages.data_pages,
            );
        }
    }

    #[test]
    fn test_column_writer_caps_page_size_with_sparse_nulls() {
        // `PLAIN` keeps ratio-scaled windows, so a sparsely-null chunk puts
        // *two* over-limit values on a page rather than one. That is the
        // deliberate choice: value-exact windows would cut this to one, but
        // `PLAIN` stores a value identically wherever it lands, so the output
        // is byte for byte the same either way while the page count doubles.
        //
        // What has to hold is that the bound stays a small constant and does
        // not scale with `write_batch_size` — the failure #9972 fixed, where
        // a page took a whole mini-batch. A window spans
        // `ceil(values * levels / values_in_chunk)` levels, which covers at
        // most two values however sparse the nulls are, so two is the whole
        // exposure. Assert it exactly: this test fails if the encoding gate
        // on value-exact windows is dropped (pages would hold one value) as
        // well as if the bound is lost (they would hold many).
        let value_size = 64 * 1024;
        let page_byte_limit = 16 * 1024;
        let num_values = 16;

        let props = WriterProperties::builder()
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::PLAIN)
            .set_data_page_size_limit(page_byte_limit)
            .set_statistics_enabled(EnabledStatistics::None)
            .build();

        let data: Vec<_> = (0..num_values)
            .map(|_| ByteArray::from(vec![b'a'; value_size]))
            .collect();
        // 17 levels: a null at index 8, values everywhere else.
        let def_levels: Vec<i16> = (0..num_values as i16 + 1)
            .map(|i| i16::from(i != 8))
            .collect();
        let pages =
            write_and_collect_pages::<ByteArrayType>(props, 1, 0, &data, Some(&def_levels), None);

        // At most two values' payload on any page, and never a whole
        // mini-batch's worth.
        let upper_bound = 2 * value_size + 64;
        for (size, n_levels) in &pages.data_pages {
            assert!(
                *size <= upper_bound,
                "page size {size} exceeds two-value bound ({upper_bound}B); pages {:?}",
                pages.data_pages,
            );
            assert!(
                *n_levels <= 3,
                "page holds {n_levels} levels, expected at most 3 (two values + a null); \
                 pages {:?}",
                pages.data_pages,
            );
        }
        // Two-level windows over 17 levels: eight pages carrying two levels
        // and a ninth holding the remainder.
        let num_levels: usize = num_values + 1;
        assert_eq!(pages.data_pages.len(), num_levels.div_ceil(2));
    }

    #[test]
    fn test_column_writer_delta_byte_array_nullable_shared_prefix_dedup() {
        // A null in the chunk must not cost dedup. The first-value exemption
        // fires when a page's opening mini-batch holds exactly one value, and
        // `write_granular_chunk` cuts windows after an exact value count, so
        // the over-limit value that opens the page gets a single-value
        // mini-batch regardless of where nulls fall.
        //
        // This pinned `[2, 2, 2, 2, 9]` before #10538: the chunker scaled the
        // one-value budget by the chunk's 17:16 level:value ratio and rounded
        // up to two-level windows, so most pages opened with two values,
        // missed the exemption, and stored their first value in full — ~5
        // full values against ~1 here (`PLAIN` stores all 16).
        let value_size = 64 * 1024;
        let page_byte_limit = 16 * 1024;
        let num_values = 16;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::DELTA_BYTE_ARRAY)
            .set_data_page_size_limit(page_byte_limit)
            .set_statistics_enabled(EnabledStatistics::None)
            .build();

        let data: Vec<_> = (0..num_values)
            .map(|_| ByteArray::from(vec![b'a'; value_size]))
            .collect();
        // 17 levels: a null at index 8, values everywhere else.
        let def_levels: Vec<i16> = (0..num_values as i16 + 1)
            .map(|i| i16::from(i != 8))
            .collect();
        let pages =
            write_and_collect_pages::<ByteArrayType>(props, 1, 0, &data, Some(&def_levels), None);

        // One page holding all 17 levels: the first value is exempt and every
        // later value dedups down to a length pair.
        let per_page_values: Vec<u32> = pages.data_pages.iter().map(|(_, n)| *n).collect();
        assert_eq!(per_page_values, vec![num_values as u32 + 1]);

        let total_bytes: usize = pages.data_pages.iter().map(|(size, _)| size).sum();
        assert!(
            total_bytes < 2 * value_size,
            "expected ~one full value's worth of bytes (full dedup), \
             got {total_bytes}B across pages {:?}",
            pages.data_pages,
        );
    }

    #[test]
    fn test_column_writer_caps_page_size_for_large_values_in_list() {
        // Coverage for the Materialized-rep branch of
        // `write_granular_chunk`. The flat-column regression test
        // exercises the per-level step; this exercises the
        // record-by-record step used when rep levels are present.
        //
        // Column is `list<required binary>` (max_def = 1, max_rep = 1)
        // with 3 records of 3 large blobs each. The page byte limit is
        // smaller than a single blob, so granular mode kicks in, and the
        // Materialized-rep arm of `write_granular_chunk` steps from one
        // `rep == 0` boundary to the next so a record never spans pages.
        let value_size = 32 * 1024;
        let page_byte_limit = 16 * 1024;
        let values_per_record = 3;
        let num_records = 3;
        let num_values = values_per_record * num_records;

        // rep levels: 0, 1, 1, 0, 1, 1, 0, 1, 1
        let mut rep_levels = Vec::with_capacity(num_values);
        for _ in 0..num_records {
            rep_levels.push(0i16);
            rep_levels.extend(std::iter::repeat_n(1i16, values_per_record - 1));
        }
        let def_levels = vec![1i16; num_values];

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::PLAIN)
            .set_data_page_size_limit(page_byte_limit)
            .build();

        let data: Vec<_> = (0..num_values)
            .map(|i| ByteArray::from(vec![i as u8; value_size]))
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(
            props,
            1,
            1,
            &data,
            Some(&def_levels),
            Some(&rep_levels),
        );
        let data_pages = pages.data_pages;

        // The Materialized-rep arm groups levels by record, and each
        // record's bytes blow the page byte limit on its own, so we get
        // exactly one page per record.
        assert_eq!(
            data_pages.len(),
            num_records,
            "expected one data page per record, got {data_pages:?}"
        );
        for (bytes, n_values) in &data_pages {
            assert_eq!(
                *n_values as usize, values_per_record,
                "each page must hold a whole record's leaves, got {data_pages:?}"
            );
            // Each page is one full record (its leaves cannot be split),
            // so allow up to `values_per_record` blobs of payload plus a
            // small fudge for level encoding overhead.
            let upper_bound = values_per_record * (value_size + 16);
            assert!(
                *bytes <= upper_bound,
                "page size {bytes} exceeds whole-record bound ({upper_bound}); pages {data_pages:?}"
            );
        }
    }

    #[test]
    fn test_column_writer_caps_page_size_with_nullable_large_values() {
        // Coverage for `LevelDataRef::value_count` on Materialized def
        // levels: a nullable column with mixed nulls and large values.
        // `value_count` must return the actual non-null count so the
        // byte estimate reflects bytes that will actually be written,
        // not the level count.
        let value_size = 32 * 1024;
        let page_byte_limit = 16 * 1024;
        let num_levels = 32;

        // Alternating null / non-null: 16 nulls and 16 values.
        let def_levels: Vec<i16> = (0..num_levels as i16).map(|i| i % 2).collect();
        let num_values = def_levels.iter().filter(|&&d| d == 1).count();

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::PLAIN)
            .set_data_page_size_limit(page_byte_limit)
            .build();

        let data: Vec<_> = (0..num_values)
            .map(|i| ByteArray::from(vec![i as u8; value_size]))
            .collect();
        let pages =
            write_and_collect_pages::<ByteArrayType>(props, 1, 0, &data, Some(&def_levels), None);
        let data_pages: Vec<_> = pages.data_pages.iter().map(|(size, _)| *size).collect();

        // With 16 actual values of 32 KiB each and a 16 KiB page limit,
        // every non-null value should get its own page (plus possibly
        // adjacent nulls). At minimum, the number of pages must be
        // roughly the value count, not 1 (which is what `main` produced).
        assert!(
            data_pages.len() >= num_values / 2,
            "expected at least {} pages for {num_values} large values, got {} pages: {data_pages:?}",
            num_values / 2,
            data_pages.len(),
        );
        // No page contains more than ~one value's worth of payload bytes.
        for size in &data_pages {
            assert!(
                *size <= value_size + 64,
                "page size {size} exceeds one-value bound; pages {data_pages:?}"
            );
        }
    }

    #[test]
    fn test_column_writer_dict_enabled_large_values_post_spill() {
        // While dictionary encoding is active, `has_dictionary()` short-
        // circuits `estimated_value_bytes` — the byte estimate is plain-
        // encoded size but dict-encoded pages only store small RLE
        // indices, so we'd otherwise shrink pages spuriously. Once the
        // dictionary spills (each value is large + unique), plain
        // encoding takes over and the byte-budget sub-batch kicks in.
        //
        // This test makes sure the writer survives that transition and
        // produces bounded pages thereafter.
        let value_size = 64 * 1024;
        let page_byte_limit = 16 * 1024;
        let num_rows = 32;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(true)
            // Force a small dict so it spills quickly even though
            // each value here is unique.
            .set_dictionary_page_size_limit(1024)
            .set_data_page_size_limit(page_byte_limit)
            // Small mini-batches so dict fallback happens part-way
            // through the input, leaving subsequent mini-batches to
            // exercise the post-spill plain-encoding path that the
            // page-size fix actually targets.
            .set_write_batch_size(4)
            .build();

        let data: Vec<_> = (0..num_rows)
            .map(|i| ByteArray::from(vec![i as u8; value_size]))
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0, &data, None, None);
        let data_pages: Vec<_> = pages.data_pages.iter().map(|(size, _)| *size).collect();

        // After spill, plain encoding writes one ~64 KiB value per page.
        // Without the fix, post-spill writes still buffered all 32
        // values into a single ~2 MiB page.
        assert!(
            data_pages.len() >= num_rows / 2,
            "expected >= {} data pages after dict spill, got {} ({data_pages:?})",
            num_rows / 2,
            data_pages.len(),
        );
        for size in &data_pages {
            assert!(
                *size <= value_size + 64,
                "page size {size} exceeds one-value bound; pages {data_pages:?}"
            );
        }
    }

    #[test]
    fn test_column_writer_caps_dictionary_page_size() {
        // A column of large *distinct* values with dictionary encoding on:
        // the dictionary page accumulates the values themselves, and its
        // spill check runs only once per mini-batch. Without bounding the
        // dictionary-encoding mini-batch, one `write_batch_size` mini-batch
        // would intern `write_batch_size * value_size` bytes into the
        // dictionary page before the check fires (~16 MiB here). The chunker
        // must sub-batch the dictionary-encoding phase too.
        let value_size = 8 * 1024;
        let dict_page_limit = 64 * 1024;
        let num_rows = 2048;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(true)
            .set_dictionary_page_size_limit(dict_page_limit)
            .build();

        let data: Vec<_> = (0..num_rows)
            .map(|i| {
                // each value distinct, so the dictionary cannot dedup them
                let mut v = vec![0u8; value_size];
                v[..8].copy_from_slice(&(i as u64).to_le_bytes());
                ByteArray::from(v)
            })
            .collect();
        let pages = write_and_collect_pages::<ByteArrayType>(props, 0, 0, &data, None, None);
        let dict_page_size = pages.dict_page_size;

        assert!(
            dict_page_size > 0,
            "expected the column to dictionary-encode"
        );
        // Bounded near the limit (~2x from the post-mini-batch check). Before
        // the fix the dictionary page reached num_rows * value_size (~16 MiB,
        // 256x the limit).
        assert!(
            dict_page_size <= 3 * dict_page_limit,
            "dictionary page {dict_page_size} exceeds 3x the {dict_page_limit} limit",
        );
    }

    #[test]
    fn test_column_writer_caps_page_size_for_fixed_len_byte_array() {
        // Coverage for `ParquetValueType::byte_size` override on
        // `FixedLenByteArray`. With `type_length = 1`, each plain-encoded
        // value is one byte, so a 4-byte page byte limit forces the
        // sub-batch sizer to write ~4 values per page rather than one
        // page for the whole batch.
        let page_byte_limit = 4;
        let num_values = 128;

        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_1_0)
            .set_dictionary_enabled(false)
            .set_encoding(Encoding::PLAIN)
            .set_data_page_size_limit(page_byte_limit)
            .build();

        let data: Vec<_> = (0..num_values)
            .map(|i| {
                let mut fla = FixedLenByteArray::default();
                fla.set_data(Bytes::from(vec![i as u8]));
                fla
            })
            .collect();
        let pages =
            write_and_collect_pages::<FixedLenByteArrayType>(props, 0, 0, &data, None, None);
        let data_pages: Vec<_> = pages.data_pages.iter().map(|(size, _)| *size).collect();

        // Without the fix this is a single 128-byte page; with the fix
        // the byte budget caps each page at ~`page_byte_limit` bytes.
        assert!(
            data_pages.len() >= num_values / 8,
            "expected pages capped by byte budget, got {data_pages:?}"
        );
        for size in &data_pages {
            assert!(
                *size <= page_byte_limit * 4,
                "page size {size} larger than expected; pages {data_pages:?}"
            );
        }
    }

    #[test]
    fn test_bool_statistics() {
        let stats = statistics_roundtrip::<BoolType>(&[true, false, false, true]);
        // Booleans have an unsigned sort order and so are not compatible
        // with the deprecated `min` and `max` statistics
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Boolean(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &false);
            assert_eq!(stats.max_opt().unwrap(), &true);
        } else {
            panic!("expecting Statistics::Boolean, got {stats:?}");
        }
    }

    #[test]
    fn test_int32_statistics() {
        let stats = statistics_roundtrip::<Int32Type>(&[-1, 3, -2, 2]);
        assert!(stats.is_min_max_backwards_compatible());
        if let Statistics::Int32(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2);
            assert_eq!(stats.max_opt().unwrap(), &3);
        } else {
            panic!("expecting Statistics::Int32, got {stats:?}");
        }
    }

    #[test]
    fn test_int64_statistics() {
        let stats = statistics_roundtrip::<Int64Type>(&[-1, 3, -2, 2]);
        assert!(stats.is_min_max_backwards_compatible());
        if let Statistics::Int64(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2);
            assert_eq!(stats.max_opt().unwrap(), &3);
        } else {
            panic!("expecting Statistics::Int64, got {stats:?}");
        }
    }

    #[test]
    fn test_int96_statistics() {
        let input = vec![
            Int96::from(vec![1, 20, 30]),
            Int96::from(vec![3, 20, 10]),
            Int96::from(vec![0, 20, 30]),
            Int96::from(vec![2, 20, 30]),
        ]
        .into_iter()
        .collect::<Vec<Int96>>();

        let stats = statistics_roundtrip::<Int96Type>(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Int96(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &Int96::from(vec![3, 20, 10]));
            assert_eq!(stats.max_opt().unwrap(), &Int96::from(vec![2, 20, 30]));
        } else {
            panic!("expecting Statistics::Int96, got {stats:?}");
        }
    }

    #[test]
    fn test_float_statistics() {
        let stats = statistics_roundtrip::<FloatType>(&[-1.0, 3.0, -2.0, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2.0);
            assert_eq!(stats.max_opt().unwrap(), &3.0);
        } else {
            panic!("expecting Statistics::Float, got {stats:?}");
        }
    }

    #[test]
    fn test_double_statistics() {
        let stats = statistics_roundtrip::<DoubleType>(&[-1.0, 3.0, -2.0, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2.0);
            assert_eq!(stats.max_opt().unwrap(), &3.0);
        } else {
            panic!("expecting Statistics::Double, got {stats:?}");
        }
    }

    #[test]
    fn test_byte_array_statistics() {
        let input = ["aawaa", "zz", "aaw", "m", "qrs"]
            .iter()
            .map(|&s| s.into())
            .collect::<Vec<_>>();

        let stats = statistics_roundtrip::<ByteArrayType>(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::ByteArray(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &ByteArray::from("aaw"));
            assert_eq!(stats.max_opt().unwrap(), &ByteArray::from("zz"));
        } else {
            panic!("expecting Statistics::ByteArray, got {stats:?}");
        }
    }

    #[test]
    fn test_fixed_len_byte_array_statistics() {
        let input = ["aawaa", "zz   ", "aaw  ", "m    ", "qrs  "]
            .iter()
            .map(|&s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = statistics_roundtrip::<FixedLenByteArrayType>(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::FixedLenByteArray(stats) = stats {
            let expected_min: FixedLenByteArray = ByteArray::from("aaw  ").into();
            assert_eq!(stats.min_opt().unwrap(), &expected_min);
            let expected_max: FixedLenByteArray = ByteArray::from("zz   ").into();
            assert_eq!(stats.max_opt().unwrap(), &expected_max);
        } else {
            panic!("expecting Statistics::FixedLenByteArray, got {stats:?}");
        }
    }

    #[test]
    fn test_ieee754_total_order_float() {
        // Test IEEE 754 total order for f32
        // Order should be: -NaN < -Inf < -1.0 < -0.0 < +0.0 < 1.0 < +Inf < +NaN
        let neg_nan = f32::from_bits(0xffc00000); // a NaN with the sign bit set
        let neg_inf = f32::NEG_INFINITY;
        let neg_one = -1.0_f32;
        let neg_zero = -0.0_f32;
        let pos_zero = 0.0_f32;
        let pos_one = 1.0_f32;
        let pos_inf = f32::INFINITY;
        let pos_nan = f32::from_bits(0x7fc00000); // a NaN with the sign bit unset

        let values = vec![
            pos_nan, neg_zero, pos_inf, neg_one, neg_nan, pos_one, neg_inf, pos_zero,
        ];

        let stats = statistics_roundtrip::<FloatType>(&values);
        if let Statistics::Float(stats) = stats {
            // With IEEE 754 total order, min should be -NaN, max should be +NaN
            // But since we filter out NaN values, min should be -Inf, max should be +Inf
            assert_eq!(stats.min_opt().unwrap(), &neg_inf);
            assert_eq!(stats.max_opt().unwrap(), &pos_inf);
            assert_eq!(stats.nan_count_opt(), Some(2)); // neg_nan and pos_nan
        } else {
            panic!("Expected float statistics");
        }
    }

    #[test]
    fn test_ieee754_total_order_float_only_nan() {
        // Test IEEE 754 total order for various NaN representations
        // They should be ordered by the significand
        let neg_nan1 = f32::from_bits(0xffc00000); // sign bit set, significand x400000
        let neg_nan2 = f32::from_bits(0xffc00001); // sign bit set, significand x400001
        let neg_nan3 = f32::from_bits(0xffc00002); // sign bit set, significand x400002
        let pos_nan1 = f32::from_bits(0x7fc00000); // sign bit unset, significand x400000
        let pos_nan2 = f32::from_bits(0x7fc00001); // sign bit unset, significand x400001
        let pos_nan3 = f32::from_bits(0x7fc00002); // sign bit unset, significand x400002

        let values = vec![neg_nan1, neg_nan2, neg_nan3, pos_nan1, pos_nan2, pos_nan3];

        let stats = statistics_roundtrip::<FloatType>(&values);
        if let Statistics::Float(stats) = stats {
            // With IEEE 754 total order, min should be `neg_nan3`, max `pos_nan3`
            assert_eq!(
                stats.min_opt().unwrap().total_cmp(&neg_nan3),
                Ordering::Equal
            );
            assert_eq!(
                stats.max_opt().unwrap().total_cmp(&pos_nan3),
                Ordering::Equal
            );
            assert_eq!(stats.nan_count_opt(), Some(6));
        } else {
            panic!("Expected float statistics");
        }
    }

    #[test]
    fn test_ieee754_total_order_double() {
        // Test IEEE 754 total order for f64
        let neg_nan = f64::from_bits(0xfff8000000000000);
        let neg_inf = f64::NEG_INFINITY;
        let neg_one = -1.0_f64;
        let neg_zero = -0.0_f64;
        let pos_zero = 0.0_f64;
        let pos_one = 1.0_f64;
        let pos_inf = f64::INFINITY;
        let pos_nan = f64::from_bits(0x7ff8000000000000);

        let values = vec![
            pos_nan, neg_zero, pos_inf, neg_one, neg_nan, pos_one, neg_inf, pos_zero,
        ];

        let stats = statistics_roundtrip::<DoubleType>(&values);
        if let Statistics::Double(stats) = stats {
            // With IEEE 754 total order, and NaN filtering
            assert_eq!(stats.min_opt().unwrap(), &neg_inf);
            assert_eq!(stats.max_opt().unwrap(), &pos_inf);
            assert_eq!(stats.nan_count_opt(), Some(2));
        } else {
            panic!("Expected double statistics");
        }
    }

    #[test]
    fn test_ieee754_total_order_double_only_nan() {
        // Test IEEE 754 total order for various NaN representations
        // They should be ordered by the significand
        let neg_nan1 = f64::from_bits(0xfff8000000000000);
        let neg_nan2 = f64::from_bits(0xfff8000000000001);
        let neg_nan3 = f64::from_bits(0xfff8000000000002);
        let pos_nan1 = f64::from_bits(0x7ff8000000000000);
        let pos_nan2 = f64::from_bits(0x7ff8000000000001);
        let pos_nan3 = f64::from_bits(0x7ff8000000000002);

        let values = vec![neg_nan1, neg_nan2, neg_nan3, pos_nan1, pos_nan2, pos_nan3];

        let stats = statistics_roundtrip::<DoubleType>(&values);
        if let Statistics::Double(stats) = stats {
            // With IEEE 754 total order, min should be `neg_nan3`, max `pos_nan3`
            assert_eq!(
                stats.min_opt().unwrap().total_cmp(&neg_nan3),
                Ordering::Equal
            );
            assert_eq!(
                stats.max_opt().unwrap().total_cmp(&pos_nan3),
                Ordering::Equal
            );
            assert_eq!(stats.nan_count_opt(), Some(6));
        } else {
            panic!("Expected float statistics");
        }
    }

    #[test]
    fn test_ieee754_total_order_zeros() {
        // Test that -0.0 and +0.0 are handled correctly
        let values = vec![-0.0_f32, 0.0_f32, -0.0_f32, 0.0_f32];

        let stats = statistics_roundtrip::<FloatType>(&values);
        if let Statistics::Float(stats) = stats {
            // With IEEE 754 total order, -0.0 < +0.0
            assert_eq!(stats.min_opt().unwrap().to_bits(), (-0.0_f32).to_bits());
            assert_eq!(stats.max_opt().unwrap().to_bits(), 0.0_f32.to_bits());
        } else {
            panic!("Expected float statistics");
        }
    }

    #[test]
    #[cfg_attr(miri, ignore)] // inline assembly is not supported
    fn test_column_writer_check_float16_min_max() {
        let input = [
            -f16::ONE,
            f16::from_f32(3.0),
            -f16::from_f32(2.0),
            f16::from_f32(2.0),
        ]
        .into_iter()
        .map(|s| ByteArray::from(s).into())
        .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(
            stats.min_opt().unwrap(),
            &ByteArray::from(-f16::from_f32(2.0))
        );
        assert_eq!(
            stats.max_opt().unwrap(),
            &ByteArray::from(f16::from_f32(3.0))
        );
    }

    #[test]
    #[cfg_attr(miri, ignore)] // inline assembly is not supported
    fn test_column_writer_check_float16_nan_middle() {
        let input = [f16::ONE, f16::NAN, f16::ONE + f16::ONE]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
        assert_eq!(
            stats.max_opt().unwrap(),
            &ByteArray::from(f16::ONE + f16::ONE)
        );
        assert_eq!(stats.nan_count_opt(), Some(1));
    }

    #[test]
    #[cfg_attr(miri, ignore)] // inline assembly is not supported
    fn test_float16_statistics_nan_middle() {
        let input = [f16::ONE, f16::NAN, f16::ONE + f16::ONE]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
        assert_eq!(
            stats.max_opt().unwrap(),
            &ByteArray::from(f16::ONE + f16::ONE)
        );
        assert_eq!(stats.nan_count_opt(), Some(1));
    }

    #[test]
    #[cfg_attr(miri, ignore)] // inline assembly is not supported
    fn test_float16_statistics_nan_start() {
        let input = [f16::NAN, f16::ONE, f16::ONE + f16::ONE]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
        assert_eq!(
            stats.max_opt().unwrap(),
            &ByteArray::from(f16::ONE + f16::ONE)
        );
        assert_eq!(stats.nan_count_opt(), Some(1));
    }

    #[test]
    fn test_float16_statistics_nan_only() {
        let input = [f16::NAN, f16::NAN]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert_eq!(
            stats.min_bytes_opt(),
            Some(ByteArray::from(f16::NAN).as_bytes())
        );
        assert_eq!(
            stats.max_bytes_opt(),
            Some(ByteArray::from(f16::NAN).as_bytes())
        );
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.nan_count_opt(), Some(2));
    }

    #[test]
    fn test_float16_statistics_zero_only() {
        let input = std::iter::once(f16::ZERO)
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ZERO));
        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
    }

    #[test]
    fn test_float16_statistics_neg_zero_only() {
        let input = std::iter::once(f16::NEG_ZERO)
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
    }

    #[test]
    fn test_float16_statistics_zero_min() {
        let input = [f16::ZERO, f16::ONE, f16::NAN, f16::PI]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ZERO));
        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::PI));
    }

    #[test]
    fn test_float16_statistics_neg_zero_max() {
        let input = [f16::NEG_ZERO, f16::NEG_ONE, f16::NAN, -f16::PI]
            .into_iter()
            .map(|s| ByteArray::from(s).into())
            .collect::<Vec<_>>();

        let stats = float16_statistics_roundtrip(&input);
        assert!(!stats.is_min_max_backwards_compatible());
        assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(-f16::PI));
        assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
    }

    #[test]
    fn test_float_statistics_nan_middle() {
        let stats = statistics_roundtrip::<FloatType>(&[1.0, f32::NAN, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &1.0);
            assert_eq!(stats.max_opt().unwrap(), &2.0);
            assert_eq!(stats.nan_count_opt(), Some(1))
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_float_statistics_nan_start() {
        let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, 1.0, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &1.0);
            assert_eq!(stats.max_opt().unwrap(), &2.0);
            assert_eq!(stats.nan_count_opt(), Some(1))
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_float_statistics_nan_only() {
        let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, f32::NAN]);
        assert_eq!(stats.min_bytes_opt(), Some(f32::NAN.as_bytes()));
        assert_eq!(stats.max_bytes_opt(), Some(f32::NAN.as_bytes()));
        assert_eq!(stats.nan_count_opt(), Some(2));
        assert!(!stats.is_min_max_backwards_compatible());
        assert!(matches!(stats, Statistics::Float(_)));
    }

    #[test]
    fn test_float_statistics_zero_only() {
        let stats = statistics_roundtrip::<FloatType>(&[0.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &0.0);
            assert!(stats.min_opt().unwrap().is_sign_positive());
            assert_eq!(stats.max_opt().unwrap(), &0.0);
            assert!(stats.max_opt().unwrap().is_sign_positive());
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_float_statistics_neg_zero_only() {
        let stats = statistics_roundtrip::<FloatType>(&[-0.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-0.0);
            assert!(stats.min_opt().unwrap().is_sign_negative());
            assert_eq!(stats.max_opt().unwrap(), &-0.0);
            assert!(stats.max_opt().unwrap().is_sign_negative());
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_float_statistics_zero_min() {
        let stats = statistics_roundtrip::<FloatType>(&[0.0, 1.0, f32::NAN, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &0.0);
            assert!(stats.min_opt().unwrap().is_sign_positive());
            assert_eq!(stats.max_opt().unwrap(), &2.0);
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_float_statistics_neg_zero_max() {
        let stats = statistics_roundtrip::<FloatType>(&[-0.0, -1.0, f32::NAN, -2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Float(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2.0);
            assert_eq!(stats.max_opt().unwrap(), &-0.0);
            assert!(stats.max_opt().unwrap().is_sign_negative());
        } else {
            panic!("expecting Statistics::Float");
        }
    }

    #[test]
    fn test_double_statistics_nan_middle() {
        let stats = statistics_roundtrip::<DoubleType>(&[1.0, f64::NAN, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &1.0);
            assert_eq!(stats.max_opt().unwrap(), &2.0);
            assert_eq!(stats.nan_count_opt(), Some(1))
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_double_statistics_nan_start() {
        let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, 1.0, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &1.0);
            assert_eq!(stats.max_opt().unwrap(), &2.0);
            assert_eq!(stats.nan_count_opt(), Some(1))
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_double_statistics_nan_only() {
        let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, f64::NAN]);
        assert_eq!(stats.min_bytes_opt(), Some(f64::NAN.as_bytes()));
        assert_eq!(stats.max_bytes_opt(), Some(f64::NAN.as_bytes()));
        assert_eq!(stats.nan_count_opt(), Some(2));
        assert!(matches!(stats, Statistics::Double(_)));
        assert!(!stats.is_min_max_backwards_compatible());
    }

    #[test]
    fn test_double_statistics_zero_only() {
        let stats = statistics_roundtrip::<DoubleType>(&[0.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &0.0);
            assert!(stats.min_opt().unwrap().is_sign_positive());
            assert_eq!(stats.max_opt().unwrap(), &0.0);
            assert!(stats.max_opt().unwrap().is_sign_positive());
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_double_statistics_neg_zero_only() {
        let stats = statistics_roundtrip::<DoubleType>(&[-0.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-0.0);
            assert!(stats.min_opt().unwrap().is_sign_negative());
            assert_eq!(stats.max_opt().unwrap(), &-0.0);
            assert!(stats.max_opt().unwrap().is_sign_negative());
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_double_statistics_zero_min() {
        let stats = statistics_roundtrip::<DoubleType>(&[0.0, 1.0, f64::NAN, 2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &0.0);
            assert!(stats.min_opt().unwrap().is_sign_positive());
            assert_eq!(stats.max_opt().unwrap(), &2.0);
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_double_statistics_neg_zero_max() {
        let stats = statistics_roundtrip::<DoubleType>(&[-0.0, -1.0, f64::NAN, -2.0]);
        assert!(!stats.is_min_max_backwards_compatible());
        if let Statistics::Double(stats) = stats {
            assert_eq!(stats.min_opt().unwrap(), &-2.0);
            assert_eq!(stats.max_opt().unwrap(), &-0.0);
            assert!(stats.max_opt().unwrap().is_sign_negative());
        } else {
            panic!("expecting Statistics::Double");
        }
    }

    #[test]
    fn test_compare_greater_byte_array_decimals() {
        assert!(!compare_greater_byte_array_decimals(&[], &[],),);
        assert!(compare_greater_byte_array_decimals(&[1u8,], &[],),);
        assert!(!compare_greater_byte_array_decimals(&[], &[1u8,],),);
        assert!(compare_greater_byte_array_decimals(&[1u8,], &[0u8,],),);
        assert!(!compare_greater_byte_array_decimals(&[1u8,], &[1u8,],),);
        assert!(compare_greater_byte_array_decimals(&[1u8, 0u8,], &[0u8,],),);
        assert!(!compare_greater_byte_array_decimals(
            &[0u8, 1u8,],
            &[1u8, 0u8,],
        ),);
        assert!(!compare_greater_byte_array_decimals(
            &[255u8, 35u8, 0u8, 0u8,],
            &[0u8,],
        ),);
        assert!(compare_greater_byte_array_decimals(
            &[0u8,],
            &[255u8, 35u8, 0u8, 0u8,],
        ),);

        // Unequal lengths where the longer value's extra leading bytes are all
        // sign extension, so the aligned tails decide.
        // https://github.com/apache/arrow-rs/issues/10860

        // 32768 > 255
        assert!(compare_greater_byte_array_decimals(
            &[0u8, 128u8, 0u8,],
            &[0u8, 255u8,],
        ),);
        assert!(!compare_greater_byte_array_decimals(
            &[0u8, 255u8,],
            &[0u8, 128u8, 0u8,],
        ),);
        // -128 > -129
        assert!(compare_greater_byte_array_decimals(
            &[128u8,],
            &[255u8, 127u8,],
        ),);
        assert!(!compare_greater_byte_array_decimals(
            &[255u8, 127u8,],
            &[128u8,],
        ),);
        // -128 > -256
        assert!(compare_greater_byte_array_decimals(
            &[128u8,],
            &[255u8, 0u8,],
        ),);
        // 10 (with a redundant leading zero) > 5
        assert!(compare_greater_byte_array_decimals(&[0u8, 10u8,], &[5u8,],),);
        assert!(compare_greater_byte_array_decimals(&[10u8,], &[0u8, 5u8,],),);
        // equal values of different lengths are not greater in either direction
        assert!(!compare_greater_byte_array_decimals(
            &[255u8, 128u8,],
            &[128u8,],
        ),);
        assert!(!compare_greater_byte_array_decimals(
            &[128u8,],
            &[255u8, 128u8,],
        ),);
    }

    #[test]
    fn test_column_index_with_null_pages() {
        // write a single page of all nulls
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 1, 0, props);
        writer.write_batch(&[], Some(&[0, 0, 0, 0]), None).unwrap();

        let r = writer.close().unwrap();
        assert!(r.column_index.is_some());
        let col_idx = r.column_index.unwrap();
        let ColumnIndexMetaData::INT32(col_idx) = col_idx else {
            panic!("wrong stats type")
        };
        // null_pages should be true for page 0
        assert!(col_idx.is_null_page(0));
        // min and max should be empty byte arrays
        assert!(col_idx.min_value(0).is_none());
        assert!(col_idx.max_value(0).is_none());
        // null_counts should be defined and be 4 for page 0
        assert!(col_idx.null_count(0).is_some());
        assert_eq!(col_idx.null_count(0), Some(4));
        // there is no repetition so rep histogram should be absent
        assert!(col_idx.repetition_level_histogram(0).is_none());
        // definition_level_histogram should be present and should be 0:4, 1:0
        assert!(col_idx.definition_level_histogram(0).is_some());
        assert_eq!(col_idx.definition_level_histogram(0).unwrap(), &[4, 0]);
    }

    #[test]
    fn test_column_offset_index_metadata() {
        // write data
        // and check the offset index and column index
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();
        // first page
        writer.flush_data_pages().unwrap();
        // second page
        writer.write_batch(&[4, 8, 2, -5], None, None).unwrap();

        let r = writer.close().unwrap();
        let column_index = r.column_index.unwrap();
        let offset_index = r.offset_index.unwrap();

        assert_eq!(8, r.rows_written);

        // column index
        let ColumnIndexMetaData::INT32(column_index) = column_index else {
            panic!("wrong stats type")
        };
        assert_eq!(2, column_index.num_pages());
        assert_eq!(2, offset_index.page_locations.len());
        assert_eq!(BoundaryOrder::UNORDERED, column_index.boundary_order);
        for idx in 0..2 {
            assert!(!column_index.is_null_page(idx));
            assert_eq!(0, column_index.null_count(0).unwrap());
        }

        if let Some(stats) = r.metadata.statistics() {
            assert_eq!(stats.null_count_opt(), Some(0));
            assert_eq!(stats.distinct_count_opt(), None);
            if let Statistics::Int32(stats) = stats {
                // first page is [1,2,3,4]
                // second page is [-5,2,4,8]
                // note that we don't increment here, as this is a non BinaryArray type.
                assert_eq!(stats.min_opt(), column_index.min_value(1));
                assert_eq!(stats.max_opt(), column_index.max_value(1));
            } else {
                panic!("expecting Statistics::Int32");
            }
        } else {
            panic!("metadata missing statistics");
        }

        // page location
        assert_eq!(0, offset_index.page_locations[0].first_row_index);
        assert_eq!(4, offset_index.page_locations[1].first_row_index);
    }

    /// Verify min/max value truncation in the column index works as expected
    #[test]
    fn test_column_offset_index_metadata_truncating() {
        // write data
        // and check the offset index and column index
        let page_writer = get_test_page_writer();
        let props = WriterProperties::builder()
            .set_statistics_truncate_length(None) // disable column index truncation
            .build()
            .into();
        let mut writer = get_test_column_writer::<FixedLenByteArrayType>(page_writer, 0, 0, props);

        let mut data = vec![FixedLenByteArray::default(); 3];
        // This is the expected min value - "aaa..."
        data[0].set_data(Bytes::from(vec![97_u8; 200]));
        // This is the expected max value - "ZZZ..."
        data[1].set_data(Bytes::from(vec![112_u8; 200]));
        data[2].set_data(Bytes::from(vec![98_u8; 200]));

        writer.write_batch(&data, None, None).unwrap();

        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();
        let column_index = r.column_index.unwrap();
        let offset_index = r.offset_index.unwrap();

        let ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(column_index) = column_index else {
            panic!("wrong stats type")
        };

        assert_eq!(3, r.rows_written);

        // column index
        assert_eq!(1, column_index.num_pages());
        assert_eq!(1, offset_index.page_locations.len());
        assert_eq!(BoundaryOrder::ASCENDING, column_index.boundary_order);
        assert!(!column_index.is_null_page(0));
        assert_eq!(Some(0), column_index.null_count(0));

        if let Some(stats) = r.metadata.statistics() {
            assert_eq!(stats.null_count_opt(), Some(0));
            assert_eq!(stats.distinct_count_opt(), None);
            if let Statistics::FixedLenByteArray(stats) = stats {
                let column_index_min_value = column_index.min_value(0).unwrap();
                let column_index_max_value = column_index.max_value(0).unwrap();

                // Column index stats are truncated, while the column chunk's aren't.
                assert_ne!(stats.min_bytes_opt().unwrap(), column_index_min_value);
                assert_ne!(stats.max_bytes_opt().unwrap(), column_index_max_value);

                assert_eq!(
                    column_index_min_value.len(),
                    DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH.unwrap()
                );
                assert_eq!(column_index_min_value, &[97_u8; 64]);
                assert_eq!(
                    column_index_max_value.len(),
                    DEFAULT_COLUMN_INDEX_TRUNCATE_LENGTH.unwrap()
                );

                // We expect the last byte to be incremented
                assert_eq!(
                    *column_index_max_value.last().unwrap(),
                    *column_index_max_value.first().unwrap() + 1
                );
            } else {
                panic!("expecting Statistics::FixedLenByteArray");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_column_offset_index_truncating_spec_example() {
        // write data
        // and check the offset index and column index
        let page_writer = get_test_page_writer();

        // Truncate values at 1 byte
        let builder = WriterProperties::builder().set_column_index_truncate_length(Some(1));
        let props = Arc::new(builder.build());
        let mut writer = get_test_column_writer::<FixedLenByteArrayType>(page_writer, 0, 0, props);

        let mut data = vec![FixedLenByteArray::default(); 1];
        // This is the expected min value
        data[0].set_data(Bytes::from(String::from("Blart Versenwald III")));

        writer.write_batch(&data, None, None).unwrap();

        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();
        let column_index = r.column_index.unwrap();
        let offset_index = r.offset_index.unwrap();

        let ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(column_index) = column_index else {
            panic!("wrong stats type")
        };

        assert_eq!(1, r.rows_written);

        // column index
        assert_eq!(1, column_index.num_pages());
        assert_eq!(1, offset_index.page_locations.len());
        assert_eq!(BoundaryOrder::ASCENDING, column_index.boundary_order);
        assert!(!column_index.is_null_page(0));
        assert_eq!(Some(0), column_index.null_count(0));

        if let Some(stats) = r.metadata.statistics() {
            assert_eq!(stats.null_count_opt(), Some(0));
            assert_eq!(stats.distinct_count_opt(), None);
            if let Statistics::FixedLenByteArray(_stats) = stats {
                let column_index_min_value = column_index.min_value(0).unwrap();
                let column_index_max_value = column_index.max_value(0).unwrap();

                assert_eq!(column_index_min_value.len(), 1);
                assert_eq!(column_index_max_value.len(), 1);

                assert_eq!(b"B", column_index_min_value);
                assert_eq!(b"C", column_index_max_value);

                assert_ne!(column_index_min_value, stats.min_bytes_opt().unwrap());
                assert_ne!(column_index_max_value, stats.max_bytes_opt().unwrap());
            } else {
                panic!("expecting Statistics::FixedLenByteArray");
            }
        } else {
            panic!("metadata missing statistics");
        }
    }

    #[test]
    fn test_float16_min_max_no_truncation() {
        // Even if we set truncation to occur at 1 byte, we should not truncate for Float16
        let builder = WriterProperties::builder().set_column_index_truncate_length(Some(1));
        let props = Arc::new(builder.build());
        let page_writer = get_test_page_writer();
        let mut writer = get_test_float16_column_writer(page_writer, props);

        let expected_value = f16::PI.to_le_bytes().to_vec();
        let data = vec![ByteArray::from(expected_value.clone()).into()];
        writer.write_batch(&data, None, None).unwrap();
        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();

        // stats should still be written
        // ensure bytes weren't truncated for column index
        let column_index = r.column_index.unwrap();
        let ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(column_index) = column_index else {
            panic!("wrong stats type")
        };
        let column_index_min_bytes = column_index.min_value(0).unwrap();
        let column_index_max_bytes = column_index.max_value(0).unwrap();
        assert_eq!(expected_value, column_index_min_bytes);
        assert_eq!(expected_value, column_index_max_bytes);

        // ensure bytes weren't truncated for statistics
        let stats = r.metadata.statistics().unwrap();
        if let Statistics::FixedLenByteArray(stats) = stats {
            let stats_min_bytes = stats.min_bytes_opt().unwrap();
            let stats_max_bytes = stats.max_bytes_opt().unwrap();
            assert_eq!(expected_value, stats_min_bytes);
            assert_eq!(expected_value, stats_max_bytes);
        } else {
            panic!("expecting Statistics::FixedLenByteArray");
        }
    }

    #[test]
    fn test_decimal_min_max_no_truncation() {
        // Even if we set truncation to occur at 1 byte, we should not truncate for Decimal
        let builder = WriterProperties::builder().set_column_index_truncate_length(Some(1));
        let props = Arc::new(builder.build());
        let page_writer = get_test_page_writer();
        let mut writer =
            get_test_decimals_column_writer::<FixedLenByteArrayType>(page_writer, 0, 0, props);

        let expected_value = vec![
            255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 255u8, 179u8, 172u8, 19u8, 35u8,
            231u8, 90u8, 0u8, 0u8,
        ];
        let data = vec![ByteArray::from(expected_value.clone()).into()];
        writer.write_batch(&data, None, None).unwrap();
        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();

        // stats should still be written
        // ensure bytes weren't truncated for column index
        let column_index = r.column_index.unwrap();
        let ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(column_index) = column_index else {
            panic!("wrong stats type")
        };
        let column_index_min_bytes = column_index.min_value(0).unwrap();
        let column_index_max_bytes = column_index.max_value(0).unwrap();
        assert_eq!(expected_value, column_index_min_bytes);
        assert_eq!(expected_value, column_index_max_bytes);

        // ensure bytes weren't truncated for statistics
        let stats = r.metadata.statistics().unwrap();
        if let Statistics::FixedLenByteArray(stats) = stats {
            let stats_min_bytes = stats.min_bytes_opt().unwrap();
            let stats_max_bytes = stats.max_bytes_opt().unwrap();
            assert_eq!(expected_value, stats_min_bytes);
            assert_eq!(expected_value, stats_max_bytes);
        } else {
            panic!("expecting Statistics::FixedLenByteArray");
        }
    }

    #[test]
    fn test_statistics_truncating_byte_array_default() {
        let page_writer = get_test_page_writer();

        // The default truncate length is 64 bytes
        let props = WriterProperties::builder().build().into();
        let mut writer = get_test_column_writer::<ByteArrayType>(page_writer, 0, 0, props);

        let mut data = vec![ByteArray::default(); 1];
        data[0].set_data(Bytes::from(String::from(
            "This string is longer than 64 bytes, so it will almost certainly be truncated.",
        )));
        writer.write_batch(&data, None, None).unwrap();
        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();

        assert_eq!(1, r.rows_written);

        let stats = r.metadata.statistics().expect("statistics");
        if let Statistics::ByteArray(stats) = stats {
            let min_value = stats.min_opt().unwrap();
            let max_value = stats.max_opt().unwrap();

            assert!(!stats.min_is_exact());
            assert!(!stats.max_is_exact());

            let expected_len = 64;
            assert_eq!(min_value.len(), expected_len);
            assert_eq!(max_value.len(), expected_len);

            let expected_min =
                "This string is longer than 64 bytes, so it will almost certainly".as_bytes();
            assert_eq!(expected_min, min_value.as_bytes());
            // note the max value is different from the min value: the last byte is incremented
            let expected_max =
                "This string is longer than 64 bytes, so it will almost certainlz".as_bytes();
            assert_eq!(expected_max, max_value.as_bytes());
        } else {
            panic!("expecting Statistics::ByteArray");
        }
    }

    #[test]
    fn test_statistics_truncating_byte_array() {
        let page_writer = get_test_page_writer();

        const TEST_TRUNCATE_LENGTH: usize = 1;

        // Truncate values at 1 byte
        let builder =
            WriterProperties::builder().set_statistics_truncate_length(Some(TEST_TRUNCATE_LENGTH));
        let props = Arc::new(builder.build());
        let mut writer = get_test_column_writer::<ByteArrayType>(page_writer, 0, 0, props);

        let mut data = vec![ByteArray::default(); 1];
        // This is the expected min value
        data[0].set_data(Bytes::from(String::from("Blart Versenwald III")));

        writer.write_batch(&data, None, None).unwrap();

        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();

        assert_eq!(1, r.rows_written);

        let stats = r.metadata.statistics().expect("statistics");
        assert_eq!(stats.null_count_opt(), Some(0));
        assert_eq!(stats.distinct_count_opt(), None);
        if let Statistics::ByteArray(stats) = stats {
            let min_value = stats.min_opt().unwrap();
            let max_value = stats.max_opt().unwrap();

            assert!(!stats.min_is_exact());
            assert!(!stats.max_is_exact());

            assert_eq!(min_value.len(), TEST_TRUNCATE_LENGTH);
            assert_eq!(max_value.len(), TEST_TRUNCATE_LENGTH);

            assert_eq!(b"B", min_value.as_bytes());
            assert_eq!(b"C", max_value.as_bytes());
        } else {
            panic!("expecting Statistics::ByteArray");
        }
    }

    #[test]
    fn test_statistics_truncating_fixed_len_byte_array() {
        let page_writer = get_test_page_writer();

        const TEST_TRUNCATE_LENGTH: usize = 1;

        // Truncate values at 1 byte
        let builder =
            WriterProperties::builder().set_statistics_truncate_length(Some(TEST_TRUNCATE_LENGTH));
        let props = Arc::new(builder.build());
        let mut writer = get_test_column_writer::<FixedLenByteArrayType>(page_writer, 0, 0, props);

        let mut data = vec![FixedLenByteArray::default(); 1];

        const PSEUDO_DECIMAL_VALUE: i128 = 6541894651216648486512564456564654;
        const PSEUDO_DECIMAL_BYTES: [u8; 16] = PSEUDO_DECIMAL_VALUE.to_be_bytes();

        const EXPECTED_MIN: [u8; TEST_TRUNCATE_LENGTH] = [PSEUDO_DECIMAL_BYTES[0]]; // parquet specifies big-endian order for decimals
        const EXPECTED_MAX: [u8; TEST_TRUNCATE_LENGTH] =
            [PSEUDO_DECIMAL_BYTES[0].overflowing_add(1).0];

        // This is the expected min value
        data[0].set_data(Bytes::from(PSEUDO_DECIMAL_BYTES.as_slice()));

        writer.write_batch(&data, None, None).unwrap();

        writer.flush_data_pages().unwrap();

        let r = writer.close().unwrap();

        assert_eq!(1, r.rows_written);

        let stats = r.metadata.statistics().expect("statistics");
        assert_eq!(stats.null_count_opt(), Some(0));
        assert_eq!(stats.distinct_count_opt(), None);
        if let Statistics::FixedLenByteArray(stats) = stats {
            let min_value = stats.min_opt().unwrap();
            let max_value = stats.max_opt().unwrap();

            assert!(!stats.min_is_exact());
            assert!(!stats.max_is_exact());

            assert_eq!(min_value.len(), TEST_TRUNCATE_LENGTH);
            assert_eq!(max_value.len(), TEST_TRUNCATE_LENGTH);

            assert_eq!(EXPECTED_MIN.as_slice(), min_value.as_bytes());
            assert_eq!(EXPECTED_MAX.as_slice(), max_value.as_bytes());

            let reconstructed_min = i128::from_be_bytes([
                min_value.as_bytes()[0],
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
            ]);

            let reconstructed_max = i128::from_be_bytes([
                max_value.as_bytes()[0],
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
                0,
            ]);

            // check that the inner value is correctly bounded by the min/max
            println!("min: {reconstructed_min} {PSEUDO_DECIMAL_VALUE}");
            assert!(reconstructed_min <= PSEUDO_DECIMAL_VALUE);
            println!("max {reconstructed_max} {PSEUDO_DECIMAL_VALUE}");
            assert!(reconstructed_max >= PSEUDO_DECIMAL_VALUE);
        } else {
            panic!("expecting Statistics::FixedLenByteArray");
        }
    }

    #[test]
    fn test_send() {
        fn test<T: Send>() {}
        test::<ColumnWriterImpl<Int32Type>>();
    }

    #[test]
    fn test_increment() {
        let v = increment(vec![0, 0, 0]).unwrap();
        assert_eq!(&v, &[0, 0, 1]);

        // Handle overflow
        let v = increment(vec![0, 255, 255]).unwrap();
        assert_eq!(&v, &[1, 0, 0]);

        // Return `None` if all bytes are u8::MAX
        let v = increment(vec![255, 255, 255]);
        assert!(v.is_none());
    }

    #[test]
    fn test_increment_utf8() {
        let test_inc = |o: &str, expected: &str| {
            if let Ok(v) = String::from_utf8(increment_utf8(o).unwrap()) {
                // Got the expected result...
                assert_eq!(v, expected);
                // and it's greater than the original string
                assert!(*v > *o);
                // Also show that BinaryArray level comparison works here
                let mut greater = ByteArray::new();
                greater.set_data(Bytes::from(v));
                let mut original = ByteArray::new();
                original.set_data(Bytes::from(o.as_bytes().to_vec()));
                assert!(greater > original);
            } else {
                panic!("Expected incremented UTF8 string to also be valid.");
            }
        };

        // Basic ASCII case
        test_inc("hello", "hellp");

        // 1-byte ending in max 1-byte
        test_inc("a\u{7f}", "b");

        // 1-byte max should not truncate as it would need 2-byte code points
        assert!(increment_utf8("\u{7f}\u{7f}").is_none());

        // UTF8 string
        test_inc("❤️🧡💛💚💙💜", "❤️🧡💛💚💙💝");

        // 2-byte without overflow
        test_inc("éééé", "éééê");

        // 2-byte that overflows lowest byte
        test_inc("\u{ff}\u{ff}", "\u{ff}\u{100}");

        // 2-byte ending in max 2-byte
        test_inc("a\u{7ff}", "b");

        // Max 2-byte should not truncate as it would need 3-byte code points
        assert!(increment_utf8("\u{7ff}\u{7ff}").is_none());

        // 3-byte without overflow [U+800, U+800] -> [U+800, U+801] (note that these
        // characters should render right to left).
        test_inc("ࠀࠀ", "ࠀࠁ");

        // 3-byte ending in max 3-byte
        test_inc("a\u{ffff}", "b");

        // Max 3-byte should not truncate as it would need 4-byte code points
        assert!(increment_utf8("\u{ffff}\u{ffff}").is_none());

        // 4-byte without overflow
        test_inc("𐀀𐀀", "𐀀𐀁");

        // 4-byte ending in max unicode
        test_inc("a\u{10ffff}", "b");

        // Max 4-byte should not truncate
        assert!(increment_utf8("\u{10ffff}\u{10ffff}").is_none());

        // Skip over surrogate pair range (0xD800..=0xDFFF)
        //test_inc("a\u{D7FF}", "a\u{e000}");
        test_inc("a\u{D7FF}", "b");
    }

    #[test]
    fn test_truncate_utf8() {
        // No-op
        let data = "❤️🧡💛💚💙💜";
        let r = truncate_utf8(data, data.len()).unwrap();
        assert_eq!(r.len(), data.len());
        assert_eq!(&r, data.as_bytes());

        // We slice it away from the UTF8 boundary
        let r = truncate_utf8(data, 13).unwrap();
        assert_eq!(r.len(), 10);
        assert_eq!(&r, "❤️🧡".as_bytes());

        // One multi-byte code point, and a length shorter than it, so we can't slice it
        let r = truncate_utf8("\u{0836}", 1);
        assert!(r.is_none());

        // Test truncate and increment for max bounds on UTF-8 statistics
        // 7-bit (i.e. ASCII)
        let r = truncate_and_increment_utf8("yyyyyyyyy", 8).unwrap();
        assert_eq!(&r, b"yyyyyyyz");

        // 2-byte without overflow
        let r = truncate_and_increment_utf8("ééééé", 7).unwrap();
        assert_eq!(&r, "ééê".as_bytes());

        // 2-byte that overflows lowest byte
        let r = truncate_and_increment_utf8("\u{ff}\u{ff}\u{ff}\u{ff}\u{ff}", 8).unwrap();
        assert_eq!(&r, "\u{ff}\u{ff}\u{ff}\u{100}".as_bytes());

        // max 2-byte should not truncate as it would need 3-byte code points
        let r = truncate_and_increment_utf8("߿߿߿߿߿", 8);
        assert!(r.is_none());

        // 3-byte without overflow [U+800, U+800, U+800] -> [U+800, U+801] (note that these
        // characters should render right to left).
        let r = truncate_and_increment_utf8("ࠀࠀࠀࠀ", 8).unwrap();
        assert_eq!(&r, "ࠀࠁ".as_bytes());

        // max 3-byte should not truncate as it would need 4-byte code points
        let r = truncate_and_increment_utf8("\u{ffff}\u{ffff}\u{ffff}", 8);
        assert!(r.is_none());

        // 4-byte without overflow
        let r = truncate_and_increment_utf8("𐀀𐀀𐀀𐀀", 9).unwrap();
        assert_eq!(&r, "𐀀𐀁".as_bytes());

        // max 4-byte should not truncate
        let r = truncate_and_increment_utf8("\u{10ffff}\u{10ffff}", 8);
        assert!(r.is_none());
    }

    #[test]
    // Check fallback truncation of statistics that should be UTF-8, but aren't
    // (see https://github.com/apache/arrow-rs/pull/6870).
    fn test_byte_array_truncate_invalid_utf8_statistics() {
        let message_type = "
            message test_schema {
                OPTIONAL BYTE_ARRAY a (UTF8);
            }
        ";
        let schema = Arc::new(parse_message_type(message_type).unwrap());

        // Create Vec<ByteArray> containing non-UTF8 bytes
        let data = vec![ByteArray::from(vec![128u8; 32]); 7];
        let def_levels = [1, 1, 1, 1, 0, 1, 0, 1, 0, 1];
        let file: File = tempfile::tempfile().unwrap();
        let props = Arc::new(
            WriterProperties::builder()
                .set_statistics_enabled(EnabledStatistics::Chunk)
                .set_statistics_truncate_length(Some(8))
                .build(),
        );

        let mut writer = SerializedFileWriter::new(&file, schema, props).unwrap();
        let mut row_group_writer = writer.next_row_group().unwrap();

        let mut col_writer = row_group_writer.next_column().unwrap().unwrap();
        col_writer
            .typed::<ByteArrayType>()
            .write_batch(&data, Some(&def_levels), None)
            .unwrap();
        col_writer.close().unwrap();
        row_group_writer.close().unwrap();
        let file_metadata = writer.close().unwrap();
        let stats = file_metadata.row_group(0).column(0).statistics().unwrap();
        assert!(!stats.max_is_exact());
        // Truncation of invalid UTF-8 should fall back to binary truncation, so last byte should
        // be incremented by 1.
        assert_eq!(
            stats.max_bytes_opt().map(|v| v.to_vec()),
            Some([128, 128, 128, 128, 128, 128, 128, 129].to_vec())
        );
    }

    #[test]
    fn test_increment_max_binary_chars() {
        let r = increment(vec![0xFF, 0xFE, 0xFD, 0xFF, 0xFF]);
        assert_eq!(&r.unwrap(), &[0xFF, 0xFE, 0xFE, 0x00, 0x00]);

        let incremented = increment(vec![0xFF, 0xFF, 0xFF]);
        assert!(incremented.is_none())
    }

    #[test]
    fn test_no_column_index_when_stats_disabled() {
        // https://github.com/apache/arrow-rs/issues/6010
        // Test that column index is not created/written for all-nulls column when page
        // statistics are disabled.
        let descr = Arc::new(get_test_column_descr::<Int32Type>(1, 0));
        let props = Arc::new(
            WriterProperties::builder()
                .set_statistics_enabled(EnabledStatistics::None)
                .build(),
        );
        let column_writer = get_column_writer(descr, props, get_test_page_writer());
        let mut writer = get_typed_column_writer::<Int32Type>(column_writer);

        let data = Vec::new();
        let def_levels = vec![0; 10];
        writer.write_batch(&data, Some(&def_levels), None).unwrap();
        writer.flush_data_pages().unwrap();

        let column_close_result = writer.close().unwrap();
        assert!(column_close_result.offset_index.is_some());
        assert!(column_close_result.column_index.is_none());
    }

    #[test]
    fn test_no_offset_index_when_disabled() {
        // Test that offset indexes can be disabled
        let descr = Arc::new(get_test_column_descr::<Int32Type>(1, 0));
        let props = Arc::new(
            WriterProperties::builder()
                .set_statistics_enabled(EnabledStatistics::None)
                .set_offset_index_disabled(true)
                .build(),
        );
        let column_writer = get_column_writer(descr, props, get_test_page_writer());
        let mut writer = get_typed_column_writer::<Int32Type>(column_writer);

        let data = Vec::new();
        let def_levels = vec![0; 10];
        writer.write_batch(&data, Some(&def_levels), None).unwrap();
        writer.flush_data_pages().unwrap();

        let column_close_result = writer.close().unwrap();
        assert!(column_close_result.offset_index.is_none());
        assert!(column_close_result.column_index.is_none());
    }

    #[test]
    fn test_offset_index_overridden() {
        // Test that offset indexes are not disabled when gathering page statistics
        let descr = Arc::new(get_test_column_descr::<Int32Type>(1, 0));
        let props = Arc::new(
            WriterProperties::builder()
                .set_statistics_enabled(EnabledStatistics::Page)
                .set_offset_index_disabled(true)
                .build(),
        );
        let column_writer = get_column_writer(descr, props, get_test_page_writer());
        let mut writer = get_typed_column_writer::<Int32Type>(column_writer);

        let data = Vec::new();
        let def_levels = vec![0; 10];
        writer.write_batch(&data, Some(&def_levels), None).unwrap();
        writer.flush_data_pages().unwrap();

        let column_close_result = writer.close().unwrap();
        assert!(column_close_result.offset_index.is_some());
        assert!(column_close_result.column_index.is_some());
    }

    #[test]
    fn test_boundary_order() -> Result<()> {
        let descr = Arc::new(get_test_column_descr::<Int32Type>(1, 0));
        // min max both ascending
        let column_close_result = write_multiple_pages::<Int32Type>(
            &descr,
            &[
                &[Some(-10), Some(10)],
                &[Some(-5), Some(11)],
                &[None],
                &[Some(-5), Some(11)],
            ],
        )?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::ASCENDING));

        // min max both descending
        let column_close_result = write_multiple_pages::<Int32Type>(
            &descr,
            &[
                &[Some(10), Some(11)],
                &[Some(5), Some(11)],
                &[None],
                &[Some(-5), Some(0)],
            ],
        )?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::DESCENDING));

        // min max both equal
        let column_close_result = write_multiple_pages::<Int32Type>(
            &descr,
            &[&[Some(10), Some(11)], &[None], &[Some(10), Some(11)]],
        )?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::ASCENDING));

        // only nulls
        let column_close_result =
            write_multiple_pages::<Int32Type>(&descr, &[&[None], &[None], &[None]])?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::ASCENDING));

        // one page
        let column_close_result =
            write_multiple_pages::<Int32Type>(&descr, &[&[Some(-10), Some(10)]])?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::ASCENDING));

        // one non-null page
        let column_close_result =
            write_multiple_pages::<Int32Type>(&descr, &[&[Some(-10), Some(10)], &[None]])?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::ASCENDING));

        // min max both unordered
        let column_close_result = write_multiple_pages::<Int32Type>(
            &descr,
            &[
                &[Some(10), Some(11)],
                &[Some(11), Some(16)],
                &[None],
                &[Some(-5), Some(0)],
            ],
        )?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::UNORDERED));

        // min max both ordered in different orders
        let column_close_result = write_multiple_pages::<Int32Type>(
            &descr,
            &[
                &[Some(1), Some(9)],
                &[Some(2), Some(8)],
                &[None],
                &[Some(3), Some(7)],
            ],
        )?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::UNORDERED));

        Ok(())
    }

    #[test]
    fn test_boundary_order_logical_type() -> Result<()> {
        // ensure that logical types account for different sort order than underlying
        // physical type representation
        let f16_descr = Arc::new(get_test_float16_column_descr(1, 0));
        let fba_descr = {
            let type_ = SchemaType::primitive_type_builder(
                "col",
                FixedLenByteArrayType::get_physical_type(),
            )
            .with_length(2)
            .build()?;
            Arc::new(ColumnDescriptor::new(
                Arc::new(type_),
                1,
                0,
                ColumnPath::from("col"),
            ))
        };

        let values: &[&[Option<FixedLenByteArray>]] = &[
            &[Some(FixedLenByteArray::from(ByteArray::from(f16::ONE)))],
            &[Some(FixedLenByteArray::from(ByteArray::from(f16::ZERO)))],
            &[Some(FixedLenByteArray::from(ByteArray::from(
                f16::NEG_ZERO,
            )))],
            &[Some(FixedLenByteArray::from(ByteArray::from(f16::NEG_ONE)))],
        ];

        // f16 descending
        let column_close_result =
            write_multiple_pages::<FixedLenByteArrayType>(&f16_descr, values)?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::DESCENDING));

        // same bytes, but fba unordered
        let column_close_result =
            write_multiple_pages::<FixedLenByteArrayType>(&fba_descr, values)?;
        let boundary_order = column_close_result
            .column_index
            .unwrap()
            .get_boundary_order();
        assert_eq!(boundary_order, Some(BoundaryOrder::UNORDERED));

        Ok(())
    }

    #[test]
    fn test_interval_stats_should_not_have_min_max() {
        let input = [
            vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
            vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1],
            vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2],
        ]
        .into_iter()
        .map(|s| ByteArray::from(s).into())
        .collect::<Vec<_>>();

        let page_writer = get_test_page_writer();
        let mut writer = get_test_interval_column_writer(page_writer);
        writer.write_batch(&input, None, None).unwrap();

        let metadata = writer.close().unwrap().metadata;
        let stats = if let Some(Statistics::FixedLenByteArray(stats)) = metadata.statistics() {
            stats.clone()
        } else {
            panic!("metadata missing statistics");
        };
        assert!(stats.min_bytes_opt().is_none());
        assert!(stats.max_bytes_opt().is_none());
    }

    #[test]
    #[cfg(feature = "arrow")]
    fn test_column_writer_get_estimated_total_bytes() {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<Int32Type>(page_writer, 0, 0, props);
        assert_eq!(writer.get_estimated_total_bytes(), 0);

        writer.write_batch(&[1, 2, 3, 4], None, None).unwrap();
        writer.add_data_page().unwrap();
        let size_with_one_page = writer.get_estimated_total_bytes();
        assert_eq!(size_with_one_page, 20);

        writer.write_batch(&[5, 6, 7, 8], None, None).unwrap();
        writer.add_data_page().unwrap();
        let size_with_two_pages = writer.get_estimated_total_bytes();
        // different pages have different compressed lengths
        assert_eq!(size_with_two_pages, 20 + 21);
    }

    fn write_multiple_pages<T: DataType>(
        column_descr: &Arc<ColumnDescriptor>,
        pages: &[&[Option<T::T>]],
    ) -> Result<ColumnCloseResult> {
        let column_writer = get_column_writer(
            column_descr.clone(),
            Default::default(),
            get_test_page_writer(),
        );
        let mut writer = get_typed_column_writer::<T>(column_writer);

        for &page in pages {
            let values = page.iter().filter_map(Clone::clone).collect::<Vec<_>>();
            let def_levels = page
                .iter()
                .map(|maybe_value| i16::from(maybe_value.is_some()))
                .collect::<Vec<_>>();
            writer.write_batch(&values, Some(&def_levels), None)?;
            writer.flush_data_pages()?;
        }

        writer.close()
    }

    /// Performs write-read roundtrip with randomly generated values and levels.
    /// `max_size` is maximum number of values or levels (if `max_def_level` > 0) to write
    /// for a column.
    fn column_roundtrip_random<T: DataType>(
        props: WriterProperties,
        max_size: usize,
        min_value: T::T,
        max_value: T::T,
        max_def_level: i16,
        max_rep_level: i16,
    ) where
        T::T: PartialOrd + SampleUniform + Copy,
    {
        let mut num_values: usize = 0;

        let mut buf: Vec<i16> = Vec::new();
        let def_levels = if max_def_level > 0 {
            random_numbers_range(max_size, 0, max_def_level + 1, &mut buf);
            for &dl in &buf[..] {
                if dl == max_def_level {
                    num_values += 1;
                }
            }
            Some(&buf[..])
        } else {
            num_values = max_size;
            None
        };

        let mut buf: Vec<i16> = Vec::new();
        let rep_levels = if max_rep_level > 0 {
            random_numbers_range(max_size, 0, max_rep_level + 1, &mut buf);
            buf[0] = 0; // Must start on record boundary
            Some(&buf[..])
        } else {
            None
        };

        let mut values: Vec<T::T> = Vec::new();
        random_numbers_range(num_values, min_value, max_value, &mut values);

        column_roundtrip::<T>(props, &values[..], def_levels, rep_levels);
    }

    /// Performs write-read roundtrip and asserts written values and levels.
    fn column_roundtrip<T: DataType>(
        props: WriterProperties,
        values: &[T::T],
        def_levels: Option<&[i16]>,
        rep_levels: Option<&[i16]>,
    ) {
        let mut file = tempfile::tempfile().unwrap();
        let mut write = TrackedWrite::new(&mut file);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));

        let max_def_level = match def_levels {
            Some(buf) => *buf.iter().max().unwrap_or(&0i16),
            None => 0i16,
        };

        let max_rep_level = match rep_levels {
            Some(buf) => *buf.iter().max().unwrap_or(&0i16),
            None => 0i16,
        };

        let mut max_batch_size = values.len();
        if let Some(levels) = def_levels {
            max_batch_size = max_batch_size.max(levels.len());
        }
        if let Some(levels) = rep_levels {
            max_batch_size = max_batch_size.max(levels.len());
        }

        let mut writer =
            get_test_column_writer::<T>(page_writer, max_def_level, max_rep_level, Arc::new(props));

        let values_written = writer.write_batch(values, def_levels, rep_levels).unwrap();
        assert_eq!(values_written, values.len());
        let result = writer.close().unwrap();

        drop(write);

        let props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .build();
        let page_reader = Box::new(
            SerializedPageReader::new_with_properties(
                Arc::new(file),
                &result.metadata,
                result.rows_written as usize,
                None,
                Arc::new(props),
            )
            .unwrap(),
        );
        let mut reader = get_test_column_reader::<T>(page_reader, max_def_level, max_rep_level);

        let mut actual_values = Vec::with_capacity(max_batch_size);
        let mut actual_def_levels = def_levels.map(|_| Vec::with_capacity(max_batch_size));
        let mut actual_rep_levels = rep_levels.map(|_| Vec::with_capacity(max_batch_size));

        let (_, values_read, levels_read) = reader
            .read_records(
                max_batch_size,
                actual_def_levels.as_mut(),
                actual_rep_levels.as_mut(),
                &mut actual_values,
            )
            .unwrap();

        // Assert values, definition and repetition levels.

        assert_eq!(&actual_values[..values_read], values);
        match actual_def_levels {
            Some(ref vec) => assert_eq!(Some(&vec[..levels_read]), def_levels),
            None => assert_eq!(None, def_levels),
        }
        match actual_rep_levels {
            Some(ref vec) => assert_eq!(Some(&vec[..levels_read]), rep_levels),
            None => assert_eq!(None, rep_levels),
        }

        // Assert written rows.

        if let Some(levels) = actual_rep_levels {
            let mut actual_rows_written = 0;
            for l in levels {
                if l == 0 {
                    actual_rows_written += 1;
                }
            }
            assert_eq!(actual_rows_written, result.rows_written);
        } else if actual_def_levels.is_some() {
            assert_eq!(levels_read as u64, result.rows_written);
        } else {
            assert_eq!(values_read as u64, result.rows_written);
        }
    }

    /// Performs write of provided values and returns column metadata of those values.
    /// Used to test encoding support for column writer.
    fn column_write_and_get_metadata<T: DataType>(
        props: WriterProperties,
        values: &[T::T],
    ) -> ColumnChunkMetaData {
        let page_writer = get_test_page_writer();
        let props = Arc::new(props);
        let mut writer = get_test_column_writer::<T>(page_writer, 0, 0, props);
        writer.write_batch(values, None, None).unwrap();
        writer.close().unwrap().metadata
    }

    // Helper function to more compactly create a PageEncodingStats struct.
    fn encoding_stats(page_type: PageType, encoding: Encoding, count: i32) -> PageEncodingStats {
        PageEncodingStats {
            page_type,
            encoding,
            count,
        }
    }

    // Function to use in tests for EncodingWriteSupport. This checks that dictionary
    // offset and encodings to make sure that column writer uses provided by trait
    // encodings.
    fn check_encoding_write_support<T: DataType>(
        version: WriterVersion,
        dict_enabled: bool,
        data: &[T::T],
        dictionary_page_offset: Option<i64>,
        encodings: &[Encoding],
        page_encoding_stats: &[PageEncodingStats],
    ) {
        let props = WriterProperties::builder()
            .set_writer_version(version)
            .set_dictionary_enabled(dict_enabled)
            .build();
        let meta = column_write_and_get_metadata::<T>(props, data);
        assert_eq!(meta.dictionary_page_offset(), dictionary_page_offset);
        assert_eq!(meta.encodings().collect::<Vec<_>>(), encodings);
        assert_eq!(meta.page_encoding_stats().unwrap(), page_encoding_stats);
    }

    /// Returns column writer.
    fn get_test_column_writer<'a, T: DataType>(
        page_writer: Box<dyn PageWriter + 'a>,
        max_def_level: i16,
        max_rep_level: i16,
        props: WriterPropertiesPtr,
    ) -> ColumnWriterImpl<'a, T> {
        let descr = Arc::new(get_test_column_descr::<T>(max_def_level, max_rep_level));
        let column_writer = get_column_writer(descr, props, page_writer);
        get_typed_column_writer::<T>(column_writer)
    }

    fn get_test_column_writer_with_path<'a, T: DataType>(
        page_writer: Box<dyn PageWriter + 'a>,
        max_def_level: i16,
        max_rep_level: i16,
        props: WriterPropertiesPtr,
        path: ColumnPath,
    ) -> ColumnWriterImpl<'a, T> {
        let descr = Arc::new(get_test_column_descr_with_path::<T>(
            max_def_level,
            max_rep_level,
            path,
        ));
        let column_writer = get_column_writer(descr, props, page_writer);
        get_typed_column_writer::<T>(column_writer)
    }

    /// Pages collected by [`write_and_collect_pages`].
    struct CollectedPages {
        /// `(compressed byte size, value count)` for every data page, in order.
        data_pages: Vec<(usize, u32)>,
        /// Largest dictionary page seen, or 0 if the column wasn't dict-encoded.
        dict_page_size: usize,
    }

    /// Writes `data` (with optional def/rep levels) through a raw
    /// `ColumnWriterImpl` configured by `props`, then re-reads the file and
    /// returns its page layout. Shared by the page-size regression tests so
    /// each only has to express its props, input, and assertions.
    fn write_and_collect_pages<T: DataType>(
        props: WriterProperties,
        max_def_level: i16,
        max_rep_level: i16,
        data: &[T::T],
        def_levels: Option<&[i16]>,
        rep_levels: Option<&[i16]>,
    ) -> CollectedPages {
        let mut file = tempfile::tempfile().unwrap();
        let mut write = TrackedWrite::new(&mut file);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));
        let mut writer =
            get_test_column_writer::<T>(page_writer, max_def_level, max_rep_level, Arc::new(props));
        writer.write_batch(data, def_levels, rep_levels).unwrap();
        let r = writer.close().unwrap();
        drop(write);

        let read_props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .build();
        let mut page_reader = Box::new(
            SerializedPageReader::new_with_properties(
                Arc::new(file),
                &r.metadata,
                r.rows_written as usize,
                None,
                Arc::new(read_props),
            )
            .unwrap(),
        );

        let mut collected = CollectedPages {
            data_pages: Vec::new(),
            dict_page_size: 0,
        };
        while let Some(page) = page_reader.get_next_page().unwrap() {
            match page.page_type() {
                PageType::DATA_PAGE | PageType::DATA_PAGE_V2 => {
                    collected
                        .data_pages
                        .push((page.buffer().len(), page.num_values()));
                }
                PageType::DICTIONARY_PAGE => {
                    collected.dict_page_size = collected.dict_page_size.max(page.buffer().len());
                }
                PageType::INDEX_PAGE => {}
            }
        }
        collected
    }

    /// Returns column reader.
    fn get_test_column_reader<T: DataType>(
        page_reader: Box<dyn PageReader>,
        max_def_level: i16,
        max_rep_level: i16,
    ) -> ColumnReaderImpl<T> {
        let descr = Arc::new(get_test_column_descr::<T>(max_def_level, max_rep_level));
        let column_reader = get_column_reader(descr, page_reader);
        get_typed_column_reader::<T>(column_reader)
    }

    /// Returns descriptor for primitive column.
    fn get_test_column_descr<T: DataType>(
        max_def_level: i16,
        max_rep_level: i16,
    ) -> ColumnDescriptor {
        let path = ColumnPath::from("col");
        let type_ = SchemaType::primitive_type_builder("col", T::get_physical_type())
            // length is set for "encoding support" tests for FIXED_LEN_BYTE_ARRAY type,
            // it should be no-op for other types
            .with_length(1)
            .build()
            .unwrap();
        ColumnDescriptor::new(Arc::new(type_), max_def_level, max_rep_level, path)
    }

    fn get_test_column_descr_with_path<T: DataType>(
        max_def_level: i16,
        max_rep_level: i16,
        path: ColumnPath,
    ) -> ColumnDescriptor {
        let name = path.string();
        let type_ = SchemaType::primitive_type_builder(&name, T::get_physical_type())
            // length is set for "encoding support" tests for FIXED_LEN_BYTE_ARRAY type,
            // it should be no-op for other types
            .with_length(1)
            .build()
            .unwrap();
        ColumnDescriptor::new(Arc::new(type_), max_def_level, max_rep_level, path)
    }

    fn write_and_collect_page_values(
        path: ColumnPath,
        props: WriterPropertiesPtr,
        data: &[i32],
    ) -> Vec<u32> {
        let mut file = tempfile::tempfile().unwrap();
        let mut write = TrackedWrite::new(&mut file);
        let page_writer = Box::new(SerializedPageWriter::new(&mut write));
        let mut writer =
            get_test_column_writer_with_path::<Int32Type>(page_writer, 0, 0, props, path);
        writer.write_batch(data, None, None).unwrap();
        let r = writer.close().unwrap();

        drop(write);

        let props = ReaderProperties::builder()
            .set_backward_compatible_lz4(false)
            .build();
        let mut page_reader = Box::new(
            SerializedPageReader::new_with_properties(
                Arc::new(file),
                &r.metadata,
                r.rows_written as usize,
                None,
                Arc::new(props),
            )
            .unwrap(),
        );

        let mut values_per_page = Vec::new();
        while let Some(page) = page_reader.get_next_page().unwrap() {
            assert_eq!(page.page_type(), PageType::DATA_PAGE);
            values_per_page.push(page.num_values());
        }

        values_per_page
    }

    /// Returns page writer that collects pages without serializing them.
    fn get_test_page_writer() -> Box<dyn PageWriter> {
        Box::new(TestPageWriter {})
    }

    struct TestPageWriter {}

    impl PageWriter for TestPageWriter {
        fn write_page(&mut self, page: CompressedPage) -> Result<PageWriteSpec> {
            let mut res = PageWriteSpec::new();
            res.page_type = page.page_type();
            res.uncompressed_size = page.uncompressed_size();
            res.compressed_size = page.compressed_size();
            res.num_values = page.num_values();
            res.offset = 0;
            res.bytes_written = page.data().len() as u64;
            Ok(res)
        }

        fn close(&mut self) -> Result<()> {
            Ok(())
        }
    }

    /// Write data into parquet using [`get_test_page_writer`] and [`get_test_column_writer`] and returns generated statistics.
    fn statistics_roundtrip<T: DataType>(values: &[<T as DataType>::T]) -> Statistics {
        let page_writer = get_test_page_writer();
        let props = Default::default();
        let mut writer = get_test_column_writer::<T>(page_writer, 0, 0, props);
        writer.write_batch(values, None, None).unwrap();

        let metadata = writer.close().unwrap().metadata;
        if let Some(stats) = metadata.statistics() {
            stats.clone()
        } else {
            panic!("metadata missing statistics");
        }
    }

    /// Returns Decimals column writer.
    fn get_test_decimals_column_writer<T: DataType>(
        page_writer: Box<dyn PageWriter>,
        max_def_level: i16,
        max_rep_level: i16,
        props: WriterPropertiesPtr,
    ) -> ColumnWriterImpl<'static, T> {
        let descr = Arc::new(get_test_decimals_column_descr::<T>(
            max_def_level,
            max_rep_level,
        ));
        let column_writer = get_column_writer(descr, props, page_writer);
        get_typed_column_writer::<T>(column_writer)
    }

    /// Returns descriptor for Decimal type with primitive column.
    fn get_test_decimals_column_descr<T: DataType>(
        max_def_level: i16,
        max_rep_level: i16,
    ) -> ColumnDescriptor {
        let path = ColumnPath::from("col");
        let type_ = SchemaType::primitive_type_builder("col", T::get_physical_type())
            .with_length(16)
            .with_logical_type(Some(LogicalType::decimal(2, 3)))
            .with_scale(2)
            .with_precision(3)
            .build()
            .unwrap();
        ColumnDescriptor::new(Arc::new(type_), max_def_level, max_rep_level, path)
    }

    fn float16_statistics_roundtrip(
        values: &[FixedLenByteArray],
    ) -> ValueStatistics<FixedLenByteArray> {
        let page_writer = get_test_page_writer();
        let mut writer = get_test_float16_column_writer(page_writer, Default::default());
        writer.write_batch(values, None, None).unwrap();

        let metadata = writer.close().unwrap().metadata;
        if let Some(Statistics::FixedLenByteArray(stats)) = metadata.statistics() {
            stats.clone()
        } else {
            panic!("metadata missing statistics");
        }
    }

    fn get_test_float16_column_writer(
        page_writer: Box<dyn PageWriter>,
        props: WriterPropertiesPtr,
    ) -> ColumnWriterImpl<'static, FixedLenByteArrayType> {
        let descr = Arc::new(get_test_float16_column_descr(0, 0));
        let column_writer = get_column_writer(descr, props, page_writer);
        get_typed_column_writer::<FixedLenByteArrayType>(column_writer)
    }

    fn get_test_float16_column_descr(max_def_level: i16, max_rep_level: i16) -> ColumnDescriptor {
        let path = ColumnPath::from("col");
        let type_ =
            SchemaType::primitive_type_builder("col", FixedLenByteArrayType::get_physical_type())
                .with_length(2)
                .with_logical_type(Some(LogicalType::Float16))
                .build()
                .unwrap();
        ColumnDescriptor::new(Arc::new(type_), max_def_level, max_rep_level, path)
    }

    fn get_test_interval_column_writer(
        page_writer: Box<dyn PageWriter>,
    ) -> ColumnWriterImpl<'static, FixedLenByteArrayType> {
        let descr = Arc::new(get_test_interval_column_descr());
        let column_writer = get_column_writer(descr, Default::default(), page_writer);
        get_typed_column_writer::<FixedLenByteArrayType>(column_writer)
    }

    fn get_test_interval_column_descr() -> ColumnDescriptor {
        let path = ColumnPath::from("col");
        let type_ =
            SchemaType::primitive_type_builder("col", FixedLenByteArrayType::get_physical_type())
                .with_length(12)
                .with_converted_type(ConvertedType::INTERVAL)
                .build()
                .unwrap();
        ColumnDescriptor::new(Arc::new(type_), 0, 0, path)
    }

    /// Returns column writer for UINT32 Column provided as ConvertedType only
    fn get_test_unsigned_int_given_as_converted_column_writer<'a, T: DataType>(
        page_writer: Box<dyn PageWriter + 'a>,
        max_def_level: i16,
        max_rep_level: i16,
        props: WriterPropertiesPtr,
    ) -> ColumnWriterImpl<'a, T> {
        let descr = Arc::new(get_test_converted_type_unsigned_integer_column_descr::<T>(
            max_def_level,
            max_rep_level,
        ));
        let column_writer = get_column_writer(descr, props, page_writer);
        get_typed_column_writer::<T>(column_writer)
    }

    /// Returns column descriptor for UINT32 Column provided as ConvertedType only
    fn get_test_converted_type_unsigned_integer_column_descr<T: DataType>(
        max_def_level: i16,
        max_rep_level: i16,
    ) -> ColumnDescriptor {
        let path = ColumnPath::from("col");
        let type_ = SchemaType::primitive_type_builder("col", T::get_physical_type())
            .with_converted_type(ConvertedType::UINT_32)
            .build()
            .unwrap();
        ColumnDescriptor::new(Arc::new(type_), max_def_level, max_rep_level, path)
    }

    #[test]
    fn test_page_v2_snappy_compression_fallback() {
        // Test that PageV2 sets is_compressed to false when Snappy compression increases data size
        let page_writer = TestPageWriter {};

        // Create WriterProperties with PageV2 and Snappy compression
        let props = WriterProperties::builder()
            .set_writer_version(WriterVersion::PARQUET_2_0)
            // Disable dictionary to ensure data is written directly
            .set_dictionary_enabled(false)
            .set_compression(Compression::SNAPPY)
            .build();

        let mut column_writer =
            get_test_column_writer::<ByteArrayType>(Box::new(page_writer), 0, 0, Arc::new(props));

        // Create small, simple data that Snappy compression will likely increase in size
        // due to compression overhead for very small data
        let values = vec![ByteArray::from("a")];

        column_writer.write_batch(&values, None, None).unwrap();

        let result = column_writer.close().unwrap();
        assert_eq!(
            result.metadata.uncompressed_size(),
            result.metadata.compressed_size()
        );
    }

    struct ColumnRoundTripUniform<'a, T: DataType> {
        props: WriterProperties,
        values: &'a [T::T],
        def_levels: LevelDataRef<'a>,
        rep_levels: LevelDataRef<'a>,
        max_def_level: i16,
        max_rep_level: i16,
        expected_values: &'a [T::T],
        expected_def_levels: Option<&'a [i16]>,
        expected_rep_levels: Option<&'a [i16]>,
    }

    impl<'a, T: DataType> ColumnRoundTripUniform<'a, T>
    where
        T::T: PartialEq + std::fmt::Debug,
    {
        fn new() -> Self {
            Self {
                props: Default::default(),
                values: &[],
                def_levels: LevelDataRef::Absent,
                rep_levels: LevelDataRef::Absent,
                max_def_level: 0,
                max_rep_level: 0,
                expected_values: &[],
                expected_def_levels: None,
                expected_rep_levels: None,
            }
        }

        fn with_props(mut self, props: WriterProperties) -> Self {
            self.props = props;
            self
        }

        fn with_values(mut self, values: &'a [T::T]) -> Self {
            self.values = values;
            self
        }

        fn with_def_levels(mut self, def_levels: LevelDataRef<'a>) -> Self {
            self.def_levels = def_levels;
            self
        }

        fn with_rep_levels(mut self, rep_levels: LevelDataRef<'a>) -> Self {
            self.rep_levels = rep_levels;
            self
        }

        fn with_max_def_level(mut self, max_def_level: i16) -> Self {
            self.max_def_level = max_def_level;
            self
        }

        fn with_max_rep_level(mut self, max_rep_level: i16) -> Self {
            self.max_rep_level = max_rep_level;
            self
        }

        fn with_expected_values(mut self, expected_values: &'a [T::T]) -> Self {
            self.expected_values = expected_values;
            self
        }

        fn with_expected_def_levels(mut self, expected_def_levels: &'a [i16]) -> Self {
            self.expected_def_levels = Some(expected_def_levels);
            self
        }

        fn with_expected_rep_levels(mut self, expected_rep_levels: &'a [i16]) -> Self {
            self.expected_rep_levels = Some(expected_rep_levels);
            self
        }

        /// Write-then-read roundtrip using `write_batch_internal` with the given
        /// [`LevelDataRef`] variants, and assert the read-back matches `expected_*`.
        fn run(self) {
            let mut file = tempfile::tempfile().unwrap();
            let mut write = TrackedWrite::new(&mut file);
            let page_writer = Box::new(SerializedPageWriter::new(&mut write));
            let mut writer = get_test_column_writer::<T>(
                page_writer,
                self.max_def_level,
                self.max_rep_level,
                Arc::new(self.props),
            );

            writer
                .write_batch_internal(
                    self.values,
                    None,
                    self.def_levels,
                    self.rep_levels,
                    None,
                    None,
                    None,
                )
                .unwrap();
            let result = writer.close().unwrap();
            drop(write);

            let props = ReaderProperties::builder()
                .set_backward_compatible_lz4(false)
                .build();
            let page_reader = Box::new(
                SerializedPageReader::new_with_properties(
                    Arc::new(file),
                    &result.metadata,
                    result.rows_written as usize,
                    None,
                    Arc::new(props),
                )
                .unwrap(),
            );
            let mut reader =
                get_test_column_reader::<T>(page_reader, self.max_def_level, self.max_rep_level);

            let batch_size = self
                .expected_def_levels
                .map_or(self.expected_values.len(), |l| l.len());
            let mut actual_values = Vec::with_capacity(batch_size);
            let mut actual_def = self
                .expected_def_levels
                .map(|_| Vec::with_capacity(batch_size));
            let mut actual_rep = self
                .expected_rep_levels
                .map(|_| Vec::with_capacity(batch_size));

            let (_, values_read, levels_read) = reader
                .read_records(
                    batch_size,
                    actual_def.as_mut(),
                    actual_rep.as_mut(),
                    &mut actual_values,
                )
                .unwrap();

            assert_eq!(&actual_values[..values_read], self.expected_values);
            if let Some(ref v) = actual_def {
                assert_eq!(&v[..levels_read], self.expected_def_levels.unwrap());
            }
            if let Some(ref v) = actual_rep {
                assert_eq!(&v[..levels_read], self.expected_rep_levels.unwrap());
            }
        }
    }

    #[test]
    fn test_level_data_ref_value_count() {
        // `value_count` is what the byte-budget chunker uses to convert a
        // chunk's level span into a leaf-value count. It must work for any
        // column shape — flat, nullable, or nested — because the leaf
        // values array is decoupled from the rep/def level stream.
        let max_def = 2;
        // Non-nullable / unrepeated: no def levels materialized — every
        // level is a value.
        assert_eq!(LevelDataRef::Absent.value_count(64, max_def), 64);
        // Uniform run of present values, and of nulls.
        assert_eq!(
            LevelDataRef::Uniform {
                value: max_def,
                count: 40
            }
            .value_count(40, max_def),
            40
        );
        assert_eq!(
            LevelDataRef::Uniform {
                value: max_def - 1,
                count: 40
            }
            .value_count(40, max_def),
            0
        );
        // Materialized def levels (nullable / nested): only levels equal to
        // `max_def` are values; empty-list / null levels are not.
        let levels = [2i16, 0, 2, 1, 2, 2, 0];
        assert_eq!(
            LevelDataRef::Materialized(&levels).value_count(levels.len(), max_def),
            4
        );
    }

    #[test]
    fn test_uniform_def_levels_all_null() {
        // All-null column: def_level=0 (null) for every slot, no values written.
        let max_def_level = 1;
        let count = 100;
        let expected_def_levels = vec![0i16; count];
        ColumnRoundTripUniform::<Int32Type>::new()
            .with_def_levels(LevelDataRef::Uniform { value: 0, count })
            .with_max_def_level(max_def_level)
            .with_expected_def_levels(&expected_def_levels)
            .run();
    }

    #[test]
    fn test_uniform_def_levels_all_valid() {
        // All-valid column: def_level=max for every slot, all values written.
        let max_def_level = 1;
        let values: Vec<i32> = (0..50).collect();
        let expected_def_levels = vec![max_def_level; values.len()];
        ColumnRoundTripUniform::<Int32Type>::new()
            .with_values(&values)
            .with_def_levels(LevelDataRef::Uniform {
                value: max_def_level,
                count: values.len(),
            })
            .with_max_def_level(max_def_level)
            .with_expected_values(&values)
            .with_expected_def_levels(&expected_def_levels)
            .run();
    }

    #[test]
    fn test_uniform_def_and_rep_levels() {
        // Simulates a list column where every row is null:
        // def=0, rep=0 for each row (one row = one entry with no child values).
        let max_def_level = 2;
        let max_rep_level = 1;
        let count = 200;
        let expected_def_levels = vec![0i16; count];
        let expected_rep_levels = vec![0i16; count];
        ColumnRoundTripUniform::<Int32Type>::new()
            .with_def_levels(LevelDataRef::Uniform { value: 0, count })
            .with_rep_levels(LevelDataRef::Uniform { value: 0, count })
            .with_max_def_level(max_def_level)
            .with_max_rep_level(max_rep_level)
            .with_expected_def_levels(&expected_def_levels)
            .with_expected_rep_levels(&expected_rep_levels)
            .run();
    }

    #[test]
    fn test_uniform_levels_v1_and_v2() {
        // Verify uniform levels work identically for both Parquet writer versions.
        for version in [WriterVersion::PARQUET_1_0, WriterVersion::PARQUET_2_0] {
            let props = WriterProperties::builder()
                .set_writer_version(version)
                .build();
            let max_def = 1;
            let count = 100;
            let expected_def_levels = vec![0i16; count];
            ColumnRoundTripUniform::<Int32Type>::new()
                .with_props(props)
                .with_def_levels(LevelDataRef::Uniform { value: 0, count })
                .with_max_def_level(max_def)
                .with_expected_def_levels(&expected_def_levels)
                .run();
        }
    }
}