parquet/arrow/arrow_reader/

statistics.rs

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for additional information regarding copyright ownership
// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements

//! [`StatisticsConverter`] to convert statistics in parquet format to arrow [`ArrayRef`].

/// Notice that all the corresponding tests are in
/// `arrow-rs/parquet/tests/arrow_reader/statistics.rs`.
use crate::arrow::buffer::bit_util::sign_extend_be;
use crate::arrow::parquet_column;
use crate::basic::{ColumnOrder, Type as PhysicalType};
use crate::data_type::Int96;
use crate::errors::{ParquetError, Result};
use crate::file::metadata::RowGroupMetaData;
use crate::file::metadata::page_index::PageIndexProvider;
use crate::file::page_index::column_index::ColumnIndexMetaData;
use crate::file::statistics::Statistics as ParquetStatistics;
use crate::schema::types::SchemaDescriptor;
use arrow_array::builder::{
    BinaryBuilder, BinaryViewBuilder, BooleanBuilder, Date32Builder, Date64Builder,
    Decimal32Builder, Decimal64Builder, FixedSizeBinaryBuilder, Float16Builder, Float32Builder,
    Float64Builder, Int8Builder, Int16Builder, Int32Builder, Int64Builder, LargeBinaryBuilder,
    LargeStringBuilder, StringBuilder, StringViewBuilder, Time32MillisecondBuilder,
    Time32SecondBuilder, Time64MicrosecondBuilder, Time64NanosecondBuilder,
    TimestampMicrosecondBuilder, TimestampMillisecondBuilder, TimestampNanosecondBuilder,
    TimestampSecondBuilder, UInt8Builder, UInt16Builder, UInt32Builder, UInt64Builder,
};
use arrow_array::{
    ArrayRef, BinaryArray, BooleanArray, Date32Array, Date64Array, Decimal32Array, Decimal64Array,
    Decimal128Array, Decimal256Array, Float16Array, Float32Array, Float64Array, Int8Array,
    Int16Array, Int32Array, Int64Array, LargeBinaryArray, Time32MillisecondArray,
    Time32SecondArray, Time64MicrosecondArray, Time64NanosecondArray, TimestampMicrosecondArray,
    TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt8Array,
    UInt16Array, UInt32Array, UInt64Array, new_null_array,
};
use arrow_buffer::{NullBufferBuilder, i256};
use arrow_schema::{DataType, Field, Schema, TimeUnit};
use half::f16;
use std::sync::Arc;

// Convert the bytes array to i32.
// The endian of the input bytes array must be big-endian.
pub(crate) fn from_bytes_to_i32(b: &[u8]) -> i32 {
    // The bytes array are from parquet file and must be the big-endian.
    // The endian is defined by parquet format, and the reference document
    // https://github.com/apache/parquet-format/blob/54e53e5d7794d383529dd30746378f19a12afd58/src/main/thrift/parquet.thrift#L66
    i32::from_be_bytes(sign_extend_be::<4>(b))
}

// Convert the bytes array to i64.
// The endian of the input bytes array must be big-endian.
pub(crate) fn from_bytes_to_i64(b: &[u8]) -> i64 {
    i64::from_be_bytes(sign_extend_be::<8>(b))
}

// Convert the bytes array to i128.
// The endian of the input bytes array must be big-endian.
pub(crate) fn from_bytes_to_i128(b: &[u8]) -> i128 {
    i128::from_be_bytes(sign_extend_be::<16>(b))
}

// Convert the bytes array to i256.
// The endian of the input bytes array must be big-endian.
pub(crate) fn from_bytes_to_i256(b: &[u8]) -> i256 {
    i256::from_be_bytes(sign_extend_be::<32>(b))
}

// Convert the bytes array to f16
pub(crate) fn from_bytes_to_f16(b: &[u8]) -> Option<f16> {
    match b {
        [low, high] => Some(f16::from_be_bytes([*high, *low])),
        _ => None,
    }
}

/// Convert both bounds together: if either wraps in the reader's timestamp unit,
/// neither bound can safely describe the decoded values.
fn int96_statistics(min: &Int96, max: &Int96, unit: &TimeUnit) -> Option<(i64, i64)> {
    let convert = |value: &Int96| {
        const NANOS_PER_DAY: u64 = 86_400_000_000_000;
        let words = value.data();
        let nanos = (u64::from(words[1]) << 32) | u64::from(words[0]);
        if nanos >= NANOS_PER_DAY {
            return None;
        }
        let divisor = match unit {
            TimeUnit::Second => 1_000_000_000,
            TimeUnit::Millisecond => 1_000_000,
            TimeUnit::Microsecond => 1_000,
            TimeUnit::Nanosecond => 1,
        };
        // Canonical INT96 values fit in seconds. Reuse the reader's signed-day
        // conversion, then scale in i128 so a valid i64::MIN is not rejected
        // before adding the fractional second.
        let timestamp = i128::from(value.to_seconds()) * i128::from(1_000_000_000 / divisor)
            + i128::from((nanos % 1_000_000_000) / divisor);
        i64::try_from(timestamp).ok()
    };
    Some((convert(min)?, convert(max)?))
}

/// Define an adapter iterator for extracting statistics from an iterator of
/// `ParquetStatistics`
///
///
/// Handles checking if the statistics are present and valid with the correct type.
///
/// Parameters:
/// * `$iterator_type` is the name of the iterator type (e.g. `MinBooleanStatsIterator`)
/// * `$func` is the function to call to get the value (e.g. `min` or `max`)
/// * `$parquet_statistics_type` is the type of the statistics (e.g. `ParquetStatistics::Boolean`)
/// * `$stat_value_type` is the type of the statistics value (e.g. `bool`)
macro_rules! make_stats_iterator {
    ($iterator_type:ident, $func:ident, $parquet_statistics_type:path, $stat_value_type:ty) => {
        /// Maps an iterator of `ParquetStatistics` into an iterator of
        /// `&$stat_value_type`
        ///
        /// Yielded elements:
        /// * Some(stats) if valid
        /// * None if the statistics are not present, not valid, or not $stat_value_type
        struct $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            iter: I,
        }

        impl<'a, I> $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            /// Create a new iterator to extract the statistics
            fn new(iter: I) -> Self {
                Self { iter }
            }
        }

        /// Implement the Iterator trait for the iterator
        impl<'a, I> Iterator for $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            type Item = Option<&'a $stat_value_type>;

            /// return the next statistics value
            fn next(&mut self) -> Option<Self::Item> {
                let next = self.iter.next();
                next.map(|x| {
                    x.and_then(|stats| match stats {
                        $parquet_statistics_type(s) => s.$func(),
                        _ => None,
                    })
                })
            }

            fn size_hint(&self) -> (usize, Option<usize>) {
                self.iter.size_hint()
            }
        }
    };
}

make_stats_iterator!(
    MinBooleanStatsIterator,
    min_opt,
    ParquetStatistics::Boolean,
    bool
);
make_stats_iterator!(
    MaxBooleanStatsIterator,
    max_opt,
    ParquetStatistics::Boolean,
    bool
);
make_stats_iterator!(
    MinInt32StatsIterator,
    min_opt,
    ParquetStatistics::Int32,
    i32
);
make_stats_iterator!(
    MaxInt32StatsIterator,
    max_opt,
    ParquetStatistics::Int32,
    i32
);
make_stats_iterator!(
    MinInt64StatsIterator,
    min_opt,
    ParquetStatistics::Int64,
    i64
);
make_stats_iterator!(
    MaxInt64StatsIterator,
    max_opt,
    ParquetStatistics::Int64,
    i64
);
make_stats_iterator!(
    MinFloatStatsIterator,
    min_opt,
    ParquetStatistics::Float,
    f32
);
make_stats_iterator!(
    MaxFloatStatsIterator,
    max_opt,
    ParquetStatistics::Float,
    f32
);
make_stats_iterator!(
    MinDoubleStatsIterator,
    min_opt,
    ParquetStatistics::Double,
    f64
);
make_stats_iterator!(
    MaxDoubleStatsIterator,
    max_opt,
    ParquetStatistics::Double,
    f64
);
make_stats_iterator!(
    MinByteArrayStatsIterator,
    min_bytes_opt,
    ParquetStatistics::ByteArray,
    [u8]
);
make_stats_iterator!(
    MaxByteArrayStatsIterator,
    max_bytes_opt,
    ParquetStatistics::ByteArray,
    [u8]
);
make_stats_iterator!(
    MinFixedLenByteArrayStatsIterator,
    min_bytes_opt,
    ParquetStatistics::FixedLenByteArray,
    [u8]
);
make_stats_iterator!(
    MaxFixedLenByteArrayStatsIterator,
    max_bytes_opt,
    ParquetStatistics::FixedLenByteArray,
    [u8]
);

/// Special iterator adapter for extracting i128 values from an iterator of
/// `ParquetStatistics`
///
/// Handles checking if the statistics are present and valid with the correct type.
///
/// Depending on the parquet file, the statistics for `Decimal128` can be stored as
/// `Int32`, `Int64` or `ByteArray` or `FixedSizeByteArray` :mindblown:
///
/// This iterator handles all cases, extracting the values
/// and converting it to `stat_value_type`.
///
/// Parameters:
/// * `$iterator_type` is the name of the iterator type (e.g. `MinBooleanStatsIterator`)
/// * `$func` is the function to call to get the value (e.g. `min` or `max`)
/// * `$bytes_func` is the function to call to get the value as bytes (e.g. `min_bytes` or `max_bytes`)
/// * `$stat_value_type` is the type of the statistics value (e.g. `i128`)
/// * `convert_func` is the function to convert the bytes to stats value (e.g. `from_bytes_to_i128`)
macro_rules! make_decimal_stats_iterator {
    ($iterator_type:ident, $func:ident, $bytes_func:ident, $stat_value_type:ident, $convert_func: ident) => {
        struct $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            iter: I,
        }

        impl<'a, I> $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            fn new(iter: I) -> Self {
                Self { iter }
            }
        }

        impl<'a, I> Iterator for $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            type Item = Option<$stat_value_type>;

            fn next(&mut self) -> Option<Self::Item> {
                let next = self.iter.next();
                next.map(|x| {
                    x.and_then(|stats| match stats {
                        ParquetStatistics::Int32(s) => {
                            s.$func().map(|x| $stat_value_type::from(*x))
                        }
                        ParquetStatistics::Int64(s) => s
                            .$func()
                            .map(|x| $stat_value_type::try_from(*x).ok())
                            .flatten(),
                        ParquetStatistics::ByteArray(s) => s.$bytes_func().map($convert_func),
                        ParquetStatistics::FixedLenByteArray(s) => {
                            s.$bytes_func().map($convert_func)
                        }
                        _ => None,
                    })
                })
            }

            fn size_hint(&self) -> (usize, Option<usize>) {
                self.iter.size_hint()
            }
        }
    };
}

make_decimal_stats_iterator!(
    MinDecimal32StatsIterator,
    min_opt,
    min_bytes_opt,
    i32,
    from_bytes_to_i32
);
make_decimal_stats_iterator!(
    MaxDecimal32StatsIterator,
    max_opt,
    max_bytes_opt,
    i32,
    from_bytes_to_i32
);
make_decimal_stats_iterator!(
    MinDecimal64StatsIterator,
    min_opt,
    min_bytes_opt,
    i64,
    from_bytes_to_i64
);
make_decimal_stats_iterator!(
    MaxDecimal64StatsIterator,
    max_opt,
    max_bytes_opt,
    i64,
    from_bytes_to_i64
);
make_decimal_stats_iterator!(
    MinDecimal128StatsIterator,
    min_opt,
    min_bytes_opt,
    i128,
    from_bytes_to_i128
);
make_decimal_stats_iterator!(
    MaxDecimal128StatsIterator,
    max_opt,
    max_bytes_opt,
    i128,
    from_bytes_to_i128
);
make_decimal_stats_iterator!(
    MinDecimal256StatsIterator,
    min_opt,
    min_bytes_opt,
    i256,
    from_bytes_to_i256
);
make_decimal_stats_iterator!(
    MaxDecimal256StatsIterator,
    max_opt,
    max_bytes_opt,
    i256,
    from_bytes_to_i256
);

/// Extract timestamp bounds from INT64 or non-deprecated INT96 statistics.
macro_rules! make_timestamp_stats_iterator {
    ($iterator_type:ident, $func:ident, $bound:tt) => {
        struct $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            iter: I,
            unit: TimeUnit,
        }

        impl<'a, I> $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            fn new(iter: I, unit: TimeUnit) -> Self {
                Self { iter, unit }
            }
        }

        impl<'a, I> Iterator for $iterator_type<'a, I>
        where
            I: Iterator<Item = Option<&'a ParquetStatistics>>,
        {
            type Item = Option<i64>;

            fn next(&mut self) -> Option<Self::Item> {
                self.iter.next().map(|statistics| {
                    let statistics = statistics?;
                    match statistics {
                        ParquetStatistics::Int64(values) => values.$func().copied(),
                        // Deprecated min/max fields use signed comparison, not
                        // the timestamp order advertised by the file footer.
                        ParquetStatistics::Int96(values) if !statistics.is_min_max_deprecated() => {
                            int96_statistics(values.min_opt()?, values.max_opt()?, &self.unit)
                                .map(|bounds| bounds.$bound)
                        }
                        _ => None,
                    }
                })
            }

            fn size_hint(&self) -> (usize, Option<usize>) {
                self.iter.size_hint()
            }
        }
    };
}

make_timestamp_stats_iterator!(MinTimestampStatsIterator, min_opt, 0);
make_timestamp_stats_iterator!(MaxTimestampStatsIterator, max_opt, 1);

/// Special macro to combine the statistics iterators for min and max.
/// This is used to avoid repeating the same code for min and max statistics extractions
///
/// Parameters:
/// data_type: The data type of the statistics (e.g. `DataType::Int32`)
/// iterator: The iterator of [`ParquetStatistics`] to extract the statistics from.
macro_rules! get_statistics {
    (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
        get_statistics!(
            $data_type,
            $iterator,
            $physical_type,
            MinBooleanStatsIterator,
            MinInt32StatsIterator,
            MinInt64StatsIterator,
            MinTimestampStatsIterator,
            MinFloatStatsIterator,
            MinDoubleStatsIterator,
            MinByteArrayStatsIterator,
            MinFixedLenByteArrayStatsIterator,
            MinDecimal32StatsIterator,
            MinDecimal64StatsIterator,
            MinDecimal128StatsIterator,
            MinDecimal256StatsIterator,
            min_statistics
        )
    };
    (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
        get_statistics!(
            $data_type,
            $iterator,
            $physical_type,
            MaxBooleanStatsIterator,
            MaxInt32StatsIterator,
            MaxInt64StatsIterator,
            MaxTimestampStatsIterator,
            MaxFloatStatsIterator,
            MaxDoubleStatsIterator,
            MaxByteArrayStatsIterator,
            MaxFixedLenByteArrayStatsIterator,
            MaxDecimal32StatsIterator,
            MaxDecimal64StatsIterator,
            MaxDecimal128StatsIterator,
            MaxDecimal256StatsIterator,
            max_statistics
        )
    };
    (
        $data_type: ident,
        $iterator: ident,
        $physical_type: ident,
        $boolean_iter: ident,
        $int32_iter: ident,
        $int64_iter: ident,
        $timestamp_iter: ident,
        $float_iter: ident,
        $double_iter: ident,
        $byte_array_iter: ident,
        $fixed_len_byte_array_iter: ident,
        $decimal32_iter: ident,
        $decimal64_iter: ident,
        $decimal128_iter: ident,
        $decimal256_iter: ident,
        $dictionary_statistics: ident
    ) => {
        match $data_type {
            DataType::Boolean => Ok(Arc::new(BooleanArray::from_iter(
                $boolean_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::Int8 => Ok(Arc::new(Int8Array::from_iter(
                $int32_iter::new($iterator).map(|x| {
                    x.and_then(|x| i8::try_from(*x).ok())
                }),
            ))),
            DataType::Int16 => Ok(Arc::new(Int16Array::from_iter(
                $int32_iter::new($iterator).map(|x| {
                    x.and_then(|x| i16::try_from(*x).ok())
                }),
            ))),
            DataType::Int32 => Ok(Arc::new(Int32Array::from_iter(
                $int32_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::Int64 => Ok(Arc::new(Int64Array::from_iter(
                $int64_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::UInt8 => Ok(Arc::new(UInt8Array::from_iter(
                $int32_iter::new($iterator).map(|x| {
                    x.and_then(|x| u8::try_from(*x).ok())
                }),
            ))),
            DataType::UInt16 => Ok(Arc::new(UInt16Array::from_iter(
                $int32_iter::new($iterator).map(|x| {
                    x.and_then(|x| u16::try_from(*x).ok())
                }),
            ))),
            DataType::UInt32 => Ok(Arc::new(UInt32Array::from_iter(
                $int32_iter::new($iterator).map(|x| x.map(|x| *x as u32)),
            ))),
            DataType::UInt64 => Ok(Arc::new(UInt64Array::from_iter(
                $int64_iter::new($iterator).map(|x| x.map(|x| *x as u64)),
            ))),
            DataType::Float16 => Ok(Arc::new(Float16Array::from_iter(
                $fixed_len_byte_array_iter::new($iterator).map(|x| x.and_then(|x| {
                    from_bytes_to_f16(x)
                })),
            ))),
            DataType::Float32 => Ok(Arc::new(Float32Array::from_iter(
                $float_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::Float64 => Ok(Arc::new(Float64Array::from_iter(
                $double_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::Date32 => Ok(Arc::new(Date32Array::from_iter(
                $int32_iter::new($iterator).map(|x| x.copied()),
            ))),
            DataType::Date64 if $physical_type == Some(PhysicalType::INT32) => Ok(Arc::new(Date64Array::from_iter(
                $int32_iter::new($iterator)
                    .map(|x| x.map(|x| i64::from(*x) * 24 * 60 * 60 * 1000))))),
            DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => Ok(Arc::new(Date64Array::from_iter(
                $int64_iter::new($iterator).map(|x| x.copied()),))),
            DataType::Timestamp(unit, timezone) =>{
                let iter = $timestamp_iter::new($iterator, *unit);
                Ok(match unit {
                    TimeUnit::Second => Arc::new(TimestampSecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
                    TimeUnit::Millisecond => Arc::new(TimestampMillisecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
                    TimeUnit::Microsecond => Arc::new(TimestampMicrosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
                    TimeUnit::Nanosecond => Arc::new(TimestampNanosecondArray::from_iter(iter).with_timezone_opt(timezone.clone())),
                })
            },
            DataType::Time32(unit) => {
                Ok(match unit {
                    TimeUnit::Second =>  Arc::new(Time32SecondArray::from_iter(
                        $int32_iter::new($iterator).map(|x| x.copied()),
                    )),
                    TimeUnit::Millisecond => Arc::new(Time32MillisecondArray::from_iter(
                        $int32_iter::new($iterator).map(|x| x.copied()),
                    )),
                    _ => {
                        let len = $iterator.count();
                        // don't know how to extract statistics, so return a null array
                        new_null_array($data_type, len)
                    }
                })
            },
            DataType::Time64(unit) => {
                Ok(match unit {
                    TimeUnit::Microsecond =>  Arc::new(Time64MicrosecondArray::from_iter(
                        $int64_iter::new($iterator).map(|x| x.copied()),
                    )),
                    TimeUnit::Nanosecond => Arc::new(Time64NanosecondArray::from_iter(
                        $int64_iter::new($iterator).map(|x| x.copied()),
                    )),
                    _ => {
                        let len = $iterator.count();
                        // don't know how to extract statistics, so return a null array
                        new_null_array($data_type, len)
                    }
                })
            },
            DataType::Binary => Ok(Arc::new(BinaryArray::from_iter(
                $byte_array_iter::new($iterator)
            ))),
            DataType::LargeBinary => Ok(Arc::new(LargeBinaryArray::from_iter(
                $byte_array_iter::new($iterator)
            ))),
            DataType::Utf8 => {
                let iterator = $byte_array_iter::new($iterator);
                let mut builder = StringBuilder::new();
                for x in iterator {
                    let Some(x) = x else {
                        builder.append_null(); // no statistics value
                        continue;
                    };

                    let Ok(x) = std::str::from_utf8(x) else {
                        builder.append_null();
                        continue;
                    };

                    builder.append_value(x);
                }
                Ok(Arc::new(builder.finish()))
            },
            DataType::LargeUtf8 => {
                let iterator = $byte_array_iter::new($iterator);
                let mut builder = LargeStringBuilder::new();
                for x in iterator {
                    let Some(x) = x else {
                        builder.append_null(); // no statistics value
                        continue;
                    };

                    let Ok(x) = std::str::from_utf8(x) else {
                        builder.append_null();
                        continue;
                    };

                    builder.append_value(x);
                }
                Ok(Arc::new(builder.finish()))
            },
            DataType::FixedSizeBinary(size) => {
                let iterator = $fixed_len_byte_array_iter::new($iterator);
                let mut builder = FixedSizeBinaryBuilder::new(*size);
                for x in iterator {
                    let Some(x) = x else {
                        builder.append_null(); // no statistics value
                        continue;
                    };

                    // ignore invalid values
                    if x.len().try_into() != Ok(*size){
                        builder.append_null();
                        continue;
                    }

                    builder.append_value(x).expect("ensure to append successfully here, because size have been checked before");
                }
                Ok(Arc::new(builder.finish()))
            },
            DataType::Decimal32(precision, scale) => {
                let arr = Decimal32Array::from_iter(
                    $decimal32_iter::new($iterator)
                ).with_precision_and_scale(*precision, *scale)?;
                Ok(Arc::new(arr))
            },
            DataType::Decimal64(precision, scale) => {
                let arr = Decimal64Array::from_iter(
                    $decimal64_iter::new($iterator)
                ).with_precision_and_scale(*precision, *scale)?;
                Ok(Arc::new(arr))
            },
            DataType::Decimal128(precision, scale) => {
                let arr = Decimal128Array::from_iter(
                    $decimal128_iter::new($iterator)
                ).with_precision_and_scale(*precision, *scale)?;
                Ok(Arc::new(arr))
            },
            DataType::Decimal256(precision, scale) => {
                let arr = Decimal256Array::from_iter(
                    $decimal256_iter::new($iterator)
                ).with_precision_and_scale(*precision, *scale)?;
                Ok(Arc::new(arr))
            },
            DataType::Dictionary(_, value_type) => {
                $dictionary_statistics(value_type, $iterator, $physical_type)
            },
            DataType::Utf8View => {
                let iterator = $byte_array_iter::new($iterator);
                let mut builder = StringViewBuilder::new();
                for x in iterator {
                    let Some(x) = x else {
                        builder.append_null(); // no statistics value
                        continue;
                    };

                    let Ok(x) = std::str::from_utf8(x) else {
                        builder.append_null();
                        continue;
                    };

                    builder.append_value(x);
                }
                Ok(Arc::new(builder.finish()))
            },
            DataType::BinaryView => {
                let iterator = $byte_array_iter::new($iterator);
                let mut builder = BinaryViewBuilder::new();
                for x in iterator {
                    let Some(x) = x else {
                        builder.append_null(); // no statistics value
                        continue;
                    };

                    builder.append_value(x);
                }
                Ok(Arc::new(builder.finish()))
            }

            DataType::Map(_,_) |
            DataType::Duration(_) |
            DataType::Interval(_) |
            DataType::Date64 |  // required to cover $physical_type match guard
            DataType::Null |
            DataType::List(_) |
            DataType::ListView(_) |
            DataType::FixedSizeList(_, _) |
            DataType::LargeList(_) |
            DataType::LargeListView(_) |
            DataType::Struct(_) |
            DataType::Union(_, _) |
            DataType::RunEndEncoded(_, _) => {
                let len = $iterator.count();
                // don't know how to extract statistics, so return a null array
                Ok(new_null_array($data_type, len))
            }
        }
    };
}

macro_rules! get_data_page_statistics {
    (Min, $data_type: ident, $iterator: ident, $physical_type: ident) => {
        get_data_page_statistics!(
            $data_type,
            $iterator,
            $physical_type,
            min_values_iter,
            true,
            min_page_statistics
        )
    };
    (Max, $data_type: ident, $iterator: ident, $physical_type: ident) => {
        get_data_page_statistics!(
            $data_type,
            $iterator,
            $physical_type,
            max_values_iter,
            false,
            max_page_statistics
        )
    };
    (
        $data_type: ident,
        $iterator: ident,
        $physical_type: ident,
        $values_iter: ident,
        $is_min: expr,
        $page_statistics: ident
    ) => {{
        let chunks: Vec<(usize, Option<&ColumnIndexMetaData>)> = $iterator.collect();
        let capacity: usize = chunks.iter().map(|c| c.0).sum();
        match $data_type {
                DataType::Boolean => {
                    let mut b = BooleanBuilder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BOOLEAN(index)) => {
                                for val in index.$values_iter() {
                                    b.append_option(val.copied());
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::UInt8 => {
                    let mut b = UInt8Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.and_then(|&x| u8::try_from(x).ok())),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::UInt16 => {
                    let mut b = UInt16Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                     index.$values_iter()
                                        .map(|val| val.and_then(|&x| u16::try_from(x).ok())),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::UInt32 => {
                    let mut b = UInt32Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|&x| x as u32)),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::UInt64 => {
                    let mut b = UInt64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|&x| x as u64)),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Int8 => {
                    let mut b = Int8Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.and_then(|&x| i8::try_from(x).ok())),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Int16 => {
                    let mut b = Int16Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.and_then(|&x| i16::try_from(x).ok())),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Int32 => {
                    let mut b = Int32Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Int64 => {
                    let mut b = Int64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Float16 => {
                    let mut b = Float16Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.and_then(|x| from_bytes_to_f16(x))),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Float32 => {
                    let mut b = Float32Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::FLOAT(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Float64 => {
                    let mut b = Float64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::DOUBLE(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Binary => {
                    let mut b = BinaryBuilder::with_capacity(capacity, capacity * 10);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    b.append_option(val.map(|x| x.as_ref()));
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::LargeBinary => {
                    let mut b = LargeBinaryBuilder::with_capacity(capacity, capacity * 10);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    b.append_option(val.map(|x| x.as_ref()));
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Utf8 => {
                    let mut b = StringBuilder::with_capacity(capacity, capacity * 10);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    match val {
                                        Some(x) => match std::str::from_utf8(x.as_ref()) {
                                            Ok(s) => b.append_value(s),
                                            _ => b.append_null(),
                                        }
                                        None => b.append_null(),
                                    }
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::LargeUtf8 => {
                    let mut b = LargeStringBuilder::with_capacity(capacity, capacity * 10);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    match val {
                                        Some(x) => match std::str::from_utf8(x.as_ref()) {
                                            Ok(s) => b.append_value(s),
                                            _ => b.append_null(),
                                        }
                                        None => b.append_null(),
                                    }
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Dictionary(_, value_type) => {
                    $page_statistics(value_type, chunks.into_iter(), $physical_type)
                },
                DataType::Timestamp(unit, timezone) => {
                    match unit {
                        TimeUnit::Second => {
                            let mut b = TimestampSecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    Some(ColumnIndexMetaData::INT96(index)) => {
                                        b.extend_from_iter_option(
                                            index.min_values_iter().zip(index.max_values_iter())
                                                .map(|(min, max)| {
                                                    let (min, max) = int96_statistics(min?, max?, unit)?;
                                                    Some(if $is_min { min } else { max })
                                                }),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
                        }
                        TimeUnit::Millisecond => {
                            let mut b = TimestampMillisecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    Some(ColumnIndexMetaData::INT96(index)) => {
                                        b.extend_from_iter_option(
                                            index.min_values_iter().zip(index.max_values_iter())
                                                .map(|(min, max)| {
                                                    let (min, max) = int96_statistics(min?, max?, unit)?;
                                                    Some(if $is_min { min } else { max })
                                                }),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
                        }
                        TimeUnit::Microsecond => {
                            let mut b = TimestampMicrosecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    Some(ColumnIndexMetaData::INT96(index)) => {
                                        b.extend_from_iter_option(
                                            index.min_values_iter().zip(index.max_values_iter())
                                                .map(|(min, max)| {
                                                    let (min, max) = int96_statistics(min?, max?, unit)?;
                                                    Some(if $is_min { min } else { max })
                                                }),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
                        }
                        TimeUnit::Nanosecond => {
                            let mut b = TimestampNanosecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    Some(ColumnIndexMetaData::INT96(index)) => {
                                        b.extend_from_iter_option(
                                            index.min_values_iter().zip(index.max_values_iter())
                                                .map(|(min, max)| {
                                                    let (min, max) = int96_statistics(min?, max?, unit)?;
                                                    Some(if $is_min { min } else { max })
                                                }),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish().with_timezone_opt(timezone.clone())))
                        }
                    }
                },
                DataType::Date32 => {
                    let mut b = Date32Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Date64 if $physical_type == Some(PhysicalType::INT32)=> {
                    let mut b = Date64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|&x| (x as i64) * 24 * 60 * 60 * 1000)),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Date64 if $physical_type == Some(PhysicalType::INT64) => {
                    let mut b = Date64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Decimal32(precision, scale) => {
                    let mut b = Decimal32Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.and_then(|&x| i32::try_from(x).ok())),
                                );
                            }
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
                                );
                            }
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i32(x.as_ref()))),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
                },
                DataType::Decimal64(precision, scale) => {
                    let mut b = Decimal64Builder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| *x as i64)),
                                );
                            }
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.copied()),
                                );
                            }
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
                                );
                            }
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i64(x.as_ref()))),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
                },
                DataType::Decimal128(precision, scale) => {
                    let mut b = Decimal128Array::builder(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| *x as i128)),
                                );
                            }
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| *x as i128)),
                                );
                            }
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
                                );
                            }
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i128(x.as_ref()))),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
                },
                DataType::Decimal256(precision, scale) => {
                    let mut b = Decimal256Array::builder(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::INT32(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| i256::from_i128(*x as i128))),
                                );
                            }
                            Some(ColumnIndexMetaData::INT64(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| i256::from_i128(*x as i128))),
                                );
                            }
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
                                );
                            }
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                b.extend_from_iter_option(
                                    index.$values_iter()
                                        .map(|val| val.map(|x| from_bytes_to_i256(x.as_ref()))),
                                );
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.with_precision_and_scale(*precision, *scale)?.finish()))
                },
                DataType::Time32(unit) => {
                    match unit {
                        TimeUnit::Second => {
                            let mut b = Time32SecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT32(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish()))
                        }
                        TimeUnit::Millisecond => {
                            let mut b = Time32MillisecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT32(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish()))
                        }
                        _ => {
                            Ok(new_null_array($data_type, capacity))
                        }
                    }
                }
                DataType::Time64(unit) => {
                    match unit {
                        TimeUnit::Microsecond => {
                            let mut b = Time64MicrosecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish()))
                        }
                        TimeUnit::Nanosecond => {
                            let mut b = Time64NanosecondBuilder::with_capacity(capacity);
                            for (len, index) in chunks {
                                match index {
                                    Some(ColumnIndexMetaData::INT64(index)) => {
                                        b.extend_from_iter_option(
                                            index.$values_iter()
                                                .map(|val| val.copied()),
                                        );
                                    }
                                    _ => b.append_nulls(len),
                                }
                            }
                            Ok(Arc::new(b.finish()))
                        }
                        _ => {
                            Ok(new_null_array($data_type, capacity))
                        }
                    }
                },
                DataType::FixedSizeBinary(size) => {
                    let mut b = FixedSizeBinaryBuilder::with_capacity(capacity, *size);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    match val {
                                        Some(v) => {
                                           if v.len() == *size as usize {
                                               let _ = b.append_value(v.as_ref())?;
                                           } else {
                                               b.append_null();
                                           }
                                       }
                                        None => b.append_null(),
                                    }
                                }
                            }
                            _ => b.append_nulls(len),
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Utf8View => {
                    let mut b = StringViewBuilder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    match val {
                                        Some(x) => match std::str::from_utf8(x.as_ref()) {
                                            Ok(s) => b.append_value(s),
                                            _ => b.append_null(),
                                        }
                                        None => b.append_null(),
                                    }
                                }
                            }
                            _ => {
                                for _ in 0..len { b.append_null(); }
                            }
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::BinaryView => {
                    let mut b = BinaryViewBuilder::with_capacity(capacity);
                    for (len, index) in chunks {
                        match index {
                            Some(ColumnIndexMetaData::BYTE_ARRAY(index)) => {
                                for val in index.$values_iter() {
                                    match val {
                                        Some(v) => b.append_value(v.as_ref()),
                                        None => b.append_null(),
                                    }
                                }
                            }
                            _ => {
                                for _ in 0..len { b.append_null(); }
                            }
                        }
                    }
                    Ok(Arc::new(b.finish()))
                },
                DataType::Date64 |  // required to cover $physical_type match guard
                DataType::Null |
                DataType::Duration(_) |
                DataType::Interval(_) |
                DataType::List(_) |
                DataType::ListView(_) |
                DataType::FixedSizeList(_, _) |
                DataType::LargeList(_) |
                DataType::LargeListView(_) |
                DataType::Struct(_) |
                DataType::Union(_, _) |
                DataType::Map(_, _) |
                DataType::RunEndEncoded(_, _) => {
                    // don't know how to extract statistics, so return a null array
                    Ok(new_null_array($data_type, capacity))
                },
            }
        }
    };
}
/// Extracts the min statistics from an iterator of [`ParquetStatistics`] to an
/// [`ArrayRef`]
///
/// This is an internal helper -- see [`StatisticsConverter`] for public API
fn min_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
    data_type: &DataType,
    iterator: I,
    physical_type: Option<PhysicalType>,
) -> Result<ArrayRef> {
    get_statistics!(Min, data_type, iterator, physical_type)
}

/// Extracts the max statistics from an iterator of [`ParquetStatistics`] to an [`ArrayRef`]
///
/// This is an internal helper -- see [`StatisticsConverter`] for public API
fn max_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
    data_type: &DataType,
    iterator: I,
    physical_type: Option<PhysicalType>,
) -> Result<ArrayRef> {
    get_statistics!(Max, data_type, iterator, physical_type)
}

/// Extracts the min statistics from an iterator
/// of parquet page [`ColumnIndexMetaData`]'s to an [`ArrayRef`]
pub(crate) fn min_page_statistics<'a, I>(
    data_type: &DataType,
    iterator: I,
    physical_type: Option<PhysicalType>,
) -> Result<ArrayRef>
where
    I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
{
    get_data_page_statistics!(Min, data_type, iterator, physical_type)
}

/// Extracts the max statistics from an iterator
/// of parquet page [`ColumnIndexMetaData`]'s to an [`ArrayRef`]
pub(crate) fn max_page_statistics<'a, I>(
    data_type: &DataType,
    iterator: I,
    physical_type: Option<PhysicalType>,
) -> Result<ArrayRef>
where
    I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
{
    get_data_page_statistics!(Max, data_type, iterator, physical_type)
}

/// Extracts the null count statistics from an iterator
/// of parquet page [`ColumnIndexMetaData`]'s to an [`ArrayRef`]
///
/// The returned Array is an [`UInt64Array`]
pub(crate) fn null_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
where
    I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
{
    let chunks: Vec<_> = iterator.collect();
    let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
    let mut values = Vec::with_capacity(total_capacity);
    let mut nulls = NullBufferBuilder::new(total_capacity);
    for (len, index) in chunks {
        match index {
            Some(index) if index.null_counts().is_some() => {
                values.extend(index.null_counts().unwrap().iter().map(|&x| x as u64));
                nulls.append_n_non_nulls(len);
            }
            _ => {
                values.resize(values.len() + len, 0);
                nulls.append_n_nulls(len);
            }
        }
    }
    let null_buffer = nulls.build();
    let array = UInt64Array::new(values.into(), null_buffer);
    Ok(array)
}

/// Extracts the NaN count statistics from an iterator
/// of parquet page [`ColumnIndexMetaData`]'s to an [`ArrayRef`]
///
/// The returned Array is an [`UInt64Array`]
pub(crate) fn nan_counts_page_statistics<'a, I>(iterator: I) -> Result<UInt64Array>
where
    I: Iterator<Item = (usize, Option<&'a ColumnIndexMetaData>)>,
{
    let chunks: Vec<_> = iterator.collect();
    let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
    let mut values = Vec::with_capacity(total_capacity);
    let mut nulls = NullBufferBuilder::new(total_capacity);
    for (len, index) in chunks {
        match index {
            Some(index) if index.nan_counts().is_some() => {
                values.extend(index.nan_counts().unwrap().iter().map(|&x| x as u64));
                nulls.append_n_non_nulls(len);
            }
            _ => {
                values.resize(values.len() + len, 0);
                nulls.append_n_nulls(len);
            }
        }
    }
    let null_buffer = nulls.build();
    let array = UInt64Array::new(values.into(), null_buffer);
    Ok(array)
}

/// Extracts Parquet statistics as Arrow arrays
///
/// This is used to convert Parquet statistics to Arrow [`ArrayRef`], with
/// proper type conversions. This information can be used for pruning Parquet
/// files, row groups, and data pages based on the statistics embedded in
/// Parquet metadata.
///
/// # Statistics ordering
///
/// INT96 timestamp bounds are returned only when [`Self::with_column_orders`]
/// supplies [`ColumnOrder::INT96_TIMESTAMP_ORDER`] for the selected Parquet leaf
/// column. Without this order, INT96 minimum and maximum values are null, since
/// legacy bounds may use a different order and are not safe for timestamp pruning.
/// Null counts are unaffected.
///
/// # Schemas
///
/// The converter uses the schema of the Parquet file and the Arrow schema to
/// convert the underlying statistics value (stored as a parquet value) into the
/// corresponding Arrow value. For example, Decimals are stored as binary in
/// parquet files and this structure handles mapping them to the `i128`
/// representation used in Arrow.
///
/// Note: The Parquet schema and Arrow schema do not have to be identical (for
/// example, the columns may be in different orders and one or the other schemas
/// may have additional columns). The function [`parquet_column`] is used to
/// match the column in the Parquet schema to the column in the Arrow schema
/// when using [`Self::try_new`]. For nested fields (e.g., struct fields),
/// where `parquet_column` does not support schema resolution, use
/// [`Self::from_column_index`] instead with a pre-resolved leaf column index.
#[derive(Debug)]
pub struct StatisticsConverter<'a> {
    /// the index of the matched column in the Parquet schema
    parquet_column_index: Option<usize>,
    /// The field (with data type) of the column in the Arrow schema
    arrow_field: &'a Field,
    /// treat missing null_counts as 0 nulls
    missing_null_counts_as_zero: bool,
    /// The physical type of the matched column in the Parquet schema
    physical_type: Option<PhysicalType>,
    /// The column order of the matched Parquet leaf column, if provided
    column_order: Option<ColumnOrder>,
}

impl<'a> StatisticsConverter<'a> {
    /// Return the index of the column in the Parquet schema, if any
    ///
    /// Returns `None` if the column is was present in the Arrow schema, but not
    /// present in the parquet file
    pub fn parquet_column_index(&self) -> Option<usize> {
        self.parquet_column_index
    }

    /// Return the arrow schema's [`Field]` of the column in the Arrow schema
    pub fn arrow_field(&self) -> &'a Field {
        self.arrow_field
    }

    /// Set the statistics converter to treat missing null counts as missing
    ///
    /// By default, the converter will treat missing null counts as though
    /// the null count is known to be `0`.
    ///
    /// Note that parquet files written by parquet-rs currently do not store
    /// null counts even when it is known there are zero nulls, and the reader
    /// will return 0 for the null counts in that instance. This behavior may
    /// change in a future release.
    ///
    /// Both parquet-java and parquet-cpp store null counts as 0 when there are
    /// no nulls, and don't write unknown values to the null count field.
    pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
        self.missing_null_counts_as_zero = missing_null_counts_as_zero;
        self
    }

    /// Set the file's column orders, as returned by
    /// [`FileMetaData::column_orders`](crate::file::metadata::FileMetaData::column_orders).
    ///
    /// The orders must correspond to the Parquet schema's leaf columns. This
    /// converter selects the order using its resolved Parquet column index.
    /// INT96 min/max statistics are returned only when that order is
    /// [`ColumnOrder::INT96_TIMESTAMP_ORDER`]. Missing or other orders leave
    /// INT96 bounds unknown. Statistics for other physical types and null counts
    /// are unaffected.
    ///
    /// Only the selected order is copied, so the slice need not outlive the converter.
    pub fn with_column_orders(mut self, column_orders: Option<&[ColumnOrder]>) -> Self {
        self.column_order = self
            .parquet_column_index
            .and_then(|index| column_orders?.get(index).copied());
        self
    }

    /// Returns a [`UInt64Array`] with row counts for each row group
    ///
    /// # Return Value
    ///
    /// The returned array has no nulls, and has one value for each row group.
    /// Each value is the number of rows in the row group.
    ///
    /// # Example
    /// ```no_run
    /// # use arrow::datatypes::Schema;
    /// # use arrow_array::{ArrayRef, UInt64Array};
    /// # use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
    /// # use parquet::file::metadata::ParquetMetaData;
    /// # fn get_parquet_metadata() -> ParquetMetaData { unimplemented!() }
    /// # fn get_arrow_schema() -> Schema { unimplemented!() }
    /// // Given the metadata for a parquet file and the arrow schema
    /// let metadata: ParquetMetaData = get_parquet_metadata();
    /// let arrow_schema: Schema = get_arrow_schema();
    /// let parquet_schema = metadata.file_metadata().schema_descr();
    /// // create a converter
    /// let converter = StatisticsConverter::try_new("foo", &arrow_schema, parquet_schema)
    ///   .unwrap();
    /// // get the row counts for each row group
    /// let row_counts = converter.row_group_row_counts(metadata
    ///   .row_groups()
    ///   .iter()
    /// ).unwrap();
    /// // file had 2 row groups, with 1024 and 23 rows respectively
    /// assert_eq!(row_counts, Some(UInt64Array::from(vec![1024, 23])));
    /// ```
    pub fn row_group_row_counts<I>(&self, metadatas: I) -> Result<Option<UInt64Array>>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(_) = self.parquet_column_index else {
            return Ok(None);
        };

        let mut builder = UInt64Array::builder(10);
        for metadata in metadatas {
            let row_count = metadata.num_rows();
            let row_count: u64 = row_count.try_into().map_err(|e| {
                arrow_err!(format!(
                    "Parquet row count {row_count} too large to convert to u64: {e}"
                ))
            })?;
            builder.append_value(row_count);
        }
        Ok(Some(builder.finish()))
    }

    /// Create a new `StatisticsConverter` to extract statistics for a column
    ///
    /// Note if there is no corresponding column in the parquet file, the returned
    /// arrays will be null. This can happen if the column is in the arrow
    /// schema but not in the parquet schema due to schema evolution.
    ///
    /// This constructor only supports top-level, non-nested columns. For nested
    /// fields (e.g., fields within a struct), use [`Self::from_column_index`].
    ///
    /// See example on [`Self::row_group_mins`] for usage
    ///
    /// # Errors
    ///
    /// * If the column is not found in the arrow schema
    pub fn try_new<'b>(
        column_name: &'b str,
        arrow_schema: &'a Schema,
        parquet_schema: &'a SchemaDescriptor,
    ) -> Result<Self> {
        // ensure the requested column is in the arrow schema
        let Some((_idx, arrow_field)) = arrow_schema.column_with_name(column_name) else {
            return Err(arrow_err!(format!(
                "Column '{}' not found in schema for statistics conversion",
                column_name
            )));
        };

        // find the column in the parquet schema, if not, return a null array
        let parquet_index = match parquet_column(parquet_schema, arrow_schema, column_name) {
            Some((parquet_idx, matched_field)) => {
                // sanity check that matching field matches the arrow field
                if matched_field.as_ref() != arrow_field {
                    return Err(arrow_err!(format!(
                        "Matched column '{:?}' does not match original matched column '{:?}'",
                        matched_field, arrow_field
                    )));
                }
                Some(parquet_idx)
            }
            None => None,
        };

        Ok(Self {
            parquet_column_index: parquet_index,
            arrow_field,
            missing_null_counts_as_zero: true,
            physical_type: parquet_index.map(|idx| parquet_schema.column(idx).physical_type()),
            column_order: None,
        })
    }

    /// Create a new `StatisticsConverter` from a Parquet leaf column index directly.
    ///
    /// Unlike [`Self::try_new`], this constructor bypasses schema resolution and
    /// accepts a Parquet column index directly. This is useful for nested fields
    /// (e.g., struct fields) where the caller has already resolved the mapping
    /// from the Arrow field to the Parquet leaf column.
    ///
    /// # Arguments
    ///
    /// * `parquet_column_index` - The index of the leaf column in the Parquet schema
    /// * `arrow_field` - The Arrow field describing the column's data type
    /// * `parquet_schema` - The Parquet schema descriptor (used to look up the physical type)
    ///
    /// The caller must ensure that `arrow_field` describes the same leaf column as
    /// `parquet_column_index`. This mapping is not validated by the converter; if
    /// the Arrow type does not match the Parquet column statistics, extraction
    /// returns null statistics values rather than an error.
    ///
    /// # Errors
    ///
    /// * If the `parquet_column_index` is out of bounds
    pub fn from_column_index(
        parquet_column_index: usize,
        arrow_field: &'a Field,
        parquet_schema: &'a SchemaDescriptor,
    ) -> Result<Self> {
        if parquet_column_index >= parquet_schema.columns().len() {
            return Err(arrow_err!(format!(
                "Parquet column index {} out of bounds, column count {}",
                parquet_column_index,
                parquet_schema.columns().len()
            )));
        }

        let physical_type = parquet_schema.column(parquet_column_index).physical_type();

        Ok(Self {
            parquet_column_index: Some(parquet_column_index),
            arrow_field,
            missing_null_counts_as_zero: true,
            physical_type: Some(physical_type),
            column_order: None,
        })
    }

    /// Extract the minimum values from row group statistics in [`RowGroupMetaData`]
    ///
    /// # Return Value
    ///
    /// The returned array contains 1 value for each row group, in the same order as `metadatas`
    ///
    /// Each value is either
    /// * the minimum value for the column
    /// * a null value, if the statistics can not be extracted
    ///
    /// Note that a null value does NOT mean the min value was actually
    /// `null` it means it the requested statistic is unknown
    ///
    /// # Errors
    ///
    /// Reasons for not being able to extract the statistics include:
    /// * the column is not present in the parquet file
    /// * statistics for the column are not present in the row group
    /// * the stored statistic value can not be converted to the requested type
    ///
    /// # Example
    /// ```no_run
    /// # use std::sync::Arc;
    /// # use arrow::datatypes::Schema;
    /// # use arrow_array::{ArrayRef, Float64Array};
    /// # use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
    /// # use parquet::file::metadata::ParquetMetaData;
    /// # fn get_parquet_metadata() -> ParquetMetaData { unimplemented!() }
    /// # fn get_arrow_schema() -> Schema { unimplemented!() }
    /// // Given the metadata for a parquet file and the arrow schema
    /// let metadata: ParquetMetaData = get_parquet_metadata();
    /// let arrow_schema: Schema = get_arrow_schema();
    /// let parquet_schema = metadata.file_metadata().schema_descr();
    /// // create a converter
    /// let converter = StatisticsConverter::try_new("foo", &arrow_schema, parquet_schema)
    ///   .unwrap();
    /// // get the minimum value for the column "foo" in the parquet file
    /// let min_values: ArrayRef = converter
    ///   .row_group_mins(metadata.row_groups().iter())
    ///   .unwrap();
    /// // if "foo" is a Float64 value, the returned array will contain Float64 values
    /// assert_eq!(min_values, Arc::new(Float64Array::from(vec![Some(1.0), Some(2.0)])) as _);
    /// ```
    pub fn row_group_mins<I>(&self, metadatas: I) -> Result<ArrayRef>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let data_type = self.arrow_field.data_type();

        let Some(parquet_index) = self.parquet_column_index else {
            return Ok(self.make_null_array(data_type, metadatas));
        };

        let iter = metadatas.into_iter().map(|x| {
            x.column(parquet_index)
                .statistics()
                .filter(|_| self.is_min_max_order_supported())
        });
        min_statistics(data_type, iter, self.physical_type)
    }

    /// Extract the maximum values from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_mins`] for details
    pub fn row_group_maxes<I>(&self, metadatas: I) -> Result<ArrayRef>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let data_type = self.arrow_field.data_type();

        let Some(parquet_index) = self.parquet_column_index else {
            return Ok(self.make_null_array(data_type, metadatas));
        };

        let iter = metadatas.into_iter().map(|x| {
            x.column(parquet_index)
                .statistics()
                .filter(|_| self.is_min_max_order_supported())
        });
        max_statistics(data_type, iter, self.physical_type)
    }

    /// Extract the `is_max_value_exact` flags from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_maxes`] for details
    pub fn row_group_is_max_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = metadatas.into_iter().count();
            return Ok(BooleanArray::from_iter(std::iter::repeat_n(
                None,
                num_row_groups,
            )));
        };

        let is_max_value_exact = metadatas
            .into_iter()
            .map(|x| x.column(parquet_index).statistics())
            .map(|s| s.map(|s| s.max_is_exact()));
        Ok(BooleanArray::from_iter(is_max_value_exact))
    }

    /// Extract the `is_min_value_exact` flags from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_mins`] for details
    pub fn row_group_is_min_value_exact<I>(&self, metadatas: I) -> Result<BooleanArray>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = metadatas.into_iter().count();
            return Ok(BooleanArray::from_iter(std::iter::repeat_n(
                None,
                num_row_groups,
            )));
        };

        let is_min_value_exact = metadatas
            .into_iter()
            .map(|x| x.column(parquet_index).statistics())
            .map(|s| s.map(|s| s.min_is_exact()));
        Ok(BooleanArray::from_iter(is_min_value_exact))
    }

    /// Extract the null counts from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_mins`] for details
    pub fn row_group_null_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = metadatas.into_iter().count();
            return Ok(UInt64Array::from_iter(std::iter::repeat_n(
                None,
                num_row_groups,
            )));
        };

        let null_counts = metadatas
            .into_iter()
            .map(|x| x.column(parquet_index).statistics())
            .map(|s| {
                let s = s?;
                if self.missing_null_counts_as_zero {
                    Some(s.null_count_opt().unwrap_or(0))
                } else {
                    s.null_count_opt()
                }
            });
        Ok(UInt64Array::from_iter(null_counts))
    }

    /// Extract the NaN counts from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_mins`] for details
    pub fn row_group_nan_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = metadatas.into_iter().count();
            return Ok(UInt64Array::from_iter(std::iter::repeat_n(
                None,
                num_row_groups,
            )));
        };

        let nan_counts = metadatas
            .into_iter()
            .map(|x| x.column(parquet_index).statistics())
            .map(|s| s?.nan_count_opt());
        Ok(UInt64Array::from_iter(nan_counts))
    }

    /// Extract the distinct counts from row group statistics in [`RowGroupMetaData`]
    ///
    /// See docs on [`Self::row_group_mins`] for details
    pub fn row_group_distinct_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
    where
        I: IntoIterator<Item = &'a RowGroupMetaData>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = metadatas.into_iter().count();
            return Ok(UInt64Array::from_iter(std::iter::repeat_n(
                None,
                num_row_groups,
            )));
        };

        let distinct_counts = metadatas
            .into_iter()
            .map(|x| x.column(parquet_index).statistics())
            .map(|s| s?.distinct_count_opt());
        Ok(UInt64Array::from_iter(distinct_counts))
    }

    /// Extract the minimum values from Data Page statistics.
    ///
    /// In Parquet files, in addition to the Column Chunk level statistics
    /// (stored for each column for each row group) there are also
    /// optional statistics stored for each data page, as part of
    /// the [`PageIndex`].
    ///
    /// Since a single Column Chunk is stored as one or more pages,
    /// page level statistics can prune at a finer granularity.
    ///
    /// However since they are stored in a separate metadata
    /// structure ([`ColumnIndexMetaData`]) there is different code to extract them as
    /// compared to arrow statistics.
    ///
    /// # Parameters:
    ///
    /// * `page_index`: The parquet page indices, read from `ParquetMetaData`
    ///
    /// * `row_group_indices`: The indices of the row groups, that are used to
    ///   extract the column page index and offset index on a per row group
    ///   per column basis.
    ///
    /// # Return Value
    ///
    /// The returned array contains 1 value for each `NativeIndex`
    /// in the underlying `Index`es, in the same order as they appear
    /// in `metadatas`.
    ///
    /// For example, if there are two `Index`es in `metadatas`:
    /// 1. the first having `3` `PageIndex` entries
    /// 2. the second having `2` `PageIndex` entries
    ///
    /// The returned array would have 5 rows.
    ///
    /// Each value is either:
    /// * the minimum value for the page
    /// * a null value, if the statistics can not be extracted
    ///
    /// Note that a null value does NOT mean the min value was actually
    /// `null` it means it the requested statistic is unknown
    ///
    /// # Errors
    ///
    /// Reasons for not being able to extract the statistics include:
    /// * the column is not present in the parquet file
    /// * statistics for the pages are not present in the row group
    /// * the stored statistic value can not be converted to the requested type
    ///
    /// [`PageIndex`]: crate::file::metadata::page_index::PageIndex
    pub fn data_page_mins<I>(
        &self,
        page_index: &dyn PageIndexProvider,
        row_group_indices: I,
    ) -> Result<ArrayRef>
    where
        I: IntoIterator<Item = &'a usize>,
    {
        let data_type = self.arrow_field.data_type();

        let Some(parquet_index) = self.parquet_column_index else {
            return Ok(self.make_null_array(data_type, row_group_indices));
        };

        let iter = row_group_indices.into_iter().map(|rg_index| {
            let column_page_index_per_row_group_per_column = page_index
                .column_index(*rg_index, parquet_index)
                .filter(|_| self.is_min_max_order_supported());
            let num_data_pages = page_index
                .num_data_pages(*rg_index, parquet_index)
                .unwrap_or(0);

            (num_data_pages, column_page_index_per_row_group_per_column)
        });

        min_page_statistics(data_type, iter, self.physical_type)
    }

    /// Extract the maximum values from Data Page statistics.
    ///
    /// See docs on [`Self::data_page_mins`] for details.
    pub fn data_page_maxes<I>(
        &self,
        page_index: &dyn PageIndexProvider,
        row_group_indices: I,
    ) -> Result<ArrayRef>
    where
        I: IntoIterator<Item = &'a usize>,
    {
        let data_type = self.arrow_field.data_type();

        let Some(parquet_index) = self.parquet_column_index else {
            return Ok(self.make_null_array(data_type, row_group_indices));
        };

        let iter = row_group_indices.into_iter().map(|rg_index| {
            let column_page_index_per_row_group_per_column = page_index
                .column_index(*rg_index, parquet_index)
                .filter(|_| self.is_min_max_order_supported());
            let num_data_pages = page_index
                .num_data_pages(*rg_index, parquet_index)
                .unwrap_or(0);

            (num_data_pages, column_page_index_per_row_group_per_column)
        });

        max_page_statistics(data_type, iter, self.physical_type)
    }

    /// Returns a [`UInt64Array`] with null counts for each data page.
    ///
    /// See docs on [`Self::data_page_mins`] for details.
    pub fn data_page_null_counts<I>(
        &self,
        page_index: &dyn PageIndexProvider,
        row_group_indices: I,
    ) -> Result<UInt64Array>
    where
        I: IntoIterator<Item = &'a usize>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = row_group_indices.into_iter().count();
            return Ok(UInt64Array::new_null(num_row_groups));
        };

        let iter = row_group_indices.into_iter().map(|rg_index| {
            let column_page_index_per_row_group_per_column =
                page_index.column_index(*rg_index, parquet_index);
            let num_data_pages = page_index
                .num_data_pages(*rg_index, parquet_index)
                .unwrap_or(0);

            (num_data_pages, column_page_index_per_row_group_per_column)
        });
        null_counts_page_statistics(iter)
    }

    /// Returns a [`UInt64Array`] with NaN counts for each data page.
    ///
    /// See docs on [`Self::data_page_mins`] for details.
    pub fn data_page_nan_counts<I>(
        &self,
        page_index: &dyn PageIndexProvider,
        row_group_indices: I,
    ) -> Result<UInt64Array>
    where
        I: IntoIterator<Item = &'a usize>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            let num_row_groups = row_group_indices.into_iter().count();
            return Ok(UInt64Array::new_null(num_row_groups));
        };

        let iter = row_group_indices.into_iter().map(|rg_index| {
            let column_page_index_per_row_group_per_column =
                page_index.column_index(*rg_index, parquet_index);
            let num_data_pages = page_index
                .num_data_pages(*rg_index, parquet_index)
                .unwrap_or(0);

            (num_data_pages, column_page_index_per_row_group_per_column)
        });
        nan_counts_page_statistics(iter)
    }

    /// Returns a [`UInt64Array`] with row counts for each data page.
    ///
    /// This function iterates over the given row group indexes and computes
    /// the row count for each page in the specified column.
    ///
    /// # Parameters:
    ///
    /// * `column_offset_index`: The parquet column offset indices, read from
    ///   `ParquetMetaData` offset_index
    ///
    /// * `row_group_metadatas`: The metadata slice of the row groups, read
    ///   from `ParquetMetaData` row_groups
    ///
    /// * `row_group_indices`: The indices of the row groups, that are used to
    ///   extract the column offset index on a per row group per column basis.
    ///
    /// See docs on [`Self::data_page_mins`] for details.
    pub fn data_page_row_counts<I>(
        &self,
        page_index: &dyn PageIndexProvider,
        row_group_metadatas: &'a [RowGroupMetaData],
        row_group_indices: I,
    ) -> Result<Option<UInt64Array>>
    where
        I: IntoIterator<Item = &'a usize>,
    {
        let Some(parquet_index) = self.parquet_column_index else {
            // no matching column found in parquet_index;
            // thus we cannot extract page_locations in order to determine
            // the row count on a per DataPage basis.
            return Ok(None);
        };

        let mut row_counts = Vec::new();
        let mut nulls = NullBufferBuilder::new(0);
        for rg_idx in row_group_indices {
            let Some(offset_index) = page_index.offset_index(*rg_idx, parquet_index) else {
                // A partial page index may contain the ColumnIndex but not the OffsetIndex.
                // Preserve page-wise alignment with data_page_{mins,maxes,null_counts} by
                // emitting one null row count for every page known from the ColumnIndex.
                let num_pages = page_index
                    .column_index(*rg_idx, parquet_index)
                    .map(|index| index.num_pages() as usize)
                    .unwrap_or(0);
                row_counts.resize(row_counts.len() + num_pages, 0);
                nulls.append_n_nulls(num_pages);
                continue;
            };
            let page_locations = offset_index.page_locations();

            let row_count_per_page = page_locations
                .windows(2)
                .map(|loc| Some(loc[1].first_row_index as u64 - loc[0].first_row_index as u64));

            // append the last page row count
            let num_rows_in_row_group = &row_group_metadatas[*rg_idx].num_rows();
            let row_count_per_page = row_count_per_page.chain(std::iter::once(Some(
                *num_rows_in_row_group as u64
                    - page_locations.last().unwrap().first_row_index as u64,
            )));

            row_counts.extend(row_count_per_page.clone().map(|x| x.unwrap_or(0)));
            for val in row_count_per_page {
                if val.is_some() {
                    nulls.append_non_null();
                } else {
                    nulls.append_null();
                }
            }
        }

        Ok(Some(UInt64Array::new(row_counts.into(), nulls.build())))
    }

    fn is_min_max_order_supported(&self) -> bool {
        self.physical_type != Some(PhysicalType::INT96)
            || self.column_order == Some(ColumnOrder::INT96_TIMESTAMP_ORDER)
    }

    /// Returns a null array of data_type with one element per row group
    fn make_null_array<I, A>(&self, data_type: &DataType, metadatas: I) -> ArrayRef
    where
        I: IntoIterator<Item = A>,
    {
        // column was in the arrow schema but not in the parquet schema, so return a null array
        let num_row_groups = metadatas.into_iter().count();
        new_null_array(data_type, num_row_groups)
    }
}

// See tests in parquet/tests/arrow_reader/statistics.rs