[EPIC] Efficiently and correctly extract parquet statistics into ArrayRefs

[alamb]

Is your feature request related to a problem or challenge?

There are at least three places that parquet statistics are extracted into ArrayRefs today

1. ParquetExec (pruning Row Groups): https://github.com/apache/datafusion/blob/465c89f7f16d48b030d4a384733567b91dab88fa/datafusion/core/src/datasource/physical_plan/parquet/statistics.rs#L18-L17

• ParquetExec (Pruning pages): https://github.com/apache/datafusion/blob/671cef85c550969ab2c86d644968a048cb181c0c/datafusion/core/src/datasource/physical_plan/parquet/page_filter.rs#L393-L392

• ListingTable (pruning files): https://github.com/apache/datafusion/blob/97148bd105fc2102b0444f2d67ef535937da5dfe/datafusion/core/src/datasource/file_format/parquet.rs#L295-L294

Not only are there three copies of the code, they are all subtly different (e.g. #8295) and have varying degrees of testing

Describe the solution you'd like

I would like one API with the following properties:

1. Extracts statistics from one or more parquet files as ArrayRefs suitable to pass to PruningPredicate
2. Does so correctly (applies the appropriate schema coercion / conversion rules)
3. Does so quickly and efficiently (e.g. does not do this once per row group), is suitable for 1000s of parquet files

Describe alternatives you've considered

Some ideas from apache/arrow-rs#4328

Subtasks

• feat: API for collecting statistics/index for metadata of a parquet file + tests #10537
• Implement a benchmark for extracting arrow statistics from parquet #10606
• Incorrect statistics read for i8 i16 columns in parquet #10585
• DataFusion ignores "column order" parquet statistics specification #10586
• DataFusion reads Date32 and Date64 parquet statistics in as Int32Array #10587
• Complete porting tests from datafusion/core/src/datasource/physical_plan/parquet/statistics.rs to datafusion/core/tests/parquet/arrow_statistics.rs
• Convert ParquetExec to use the new API for RowGroup pruning
• Convert ParquetExec to use the new API for PagePruning
• Convert ListingTable to use the new API for file pruning
• Incorrect statistics read for unsigned integer columns in parquet #10604
• Incorrect statistics read for binary columns in parquet  #10605
• Incorrect statistics read for struct array in parquet #10609
• Improve performance of extracting statistics from parquet files #10626
• Minor: Add tests for extracting dictionary parquet statistics #10729
• Extract parquet statistics from timestamps with timezones #10758
• Extract parquet statistics from f16 columns #10757
• Extract parquet statistics from Time32 and Time64 columns #10751
• Extract parquet statistics from Interval columns #10752
• Extract parquet statistics from Duration columns #10754
• Extract parquet statistics from LargeBinary columns #10753
• Extract parquet statistics from LargeUtf8 columns #10756
• Extract parquet statistics from Decimal256 columns #10755

Follow on projects:

• Efficiently and correctly Extract Page Index statistics into ArrayRefs #10806

Here is a proposed API:

/// statistics extracted from `Statistics` as Arrow `ArrayRef`s
///
/// # Note:
/// If the corresponding `Statistics` is not present, or has no information for 
/// a column, a NULL is present in the  corresponding array entry
pub struct ArrowStatistics {
  /// min values
  min: ArrayRef,
  /// max values
  max: ArrayRef,
  /// Row counts (UInt64Array)
  row_count: ArrayRef,
  /// Null Counts (UInt64Array)
  null_count: ArrayRef,
}

// (TODO accessors for min/max/row_count/null_count)

/// Extract `ArrowStatistics` from the  parquet [`Statistics`]
pub fn parquet_stats_to_arrow(
    arrow_datatype: &DataType,
    statistics: impl IntoIterator<Item = Option<&Statistics>>
) -> Result<ArrowStatisics> {
  todo!()
}

Maybe it would make sense to have something more builder style:

struct ParquetStatisticsExtractor {
...
}

// create an extractor that can extract data from parquet files 
let extractor = ParquetStatisticsExtractor::new(arrow_schema, parquet_schema)

// get parquet statistics (one for each row group) somehow:
let parquet_stats: Vec<&Statistics> = ...;

// extract min/max values for column "a" and "b";
let col_a stats = extractor.extract("a", parquet_stats.iter());
let col_b stats = extractor.extract("b", parquet_stats.iter());

(This is similar to the existing API parquet::arrow::parquet_to_arrow_schema)

Note Statistics above is Statistics

There is a version of this code here in DataFusion that could perhaps be adapted:

datafusion/datafusion/core/src/datasource/physical_plan/parquet/statistics.rs

Lines 179 to 186 in accce97

	pub(crate) fn min_statistics<'a, I: Iterator<Item = Option<&'a ParquetStatistics>>>(
	data_type: &DataType,
	iterator: I,
	) -> Result<ArrayRef> {
	let scalars = iterator
	.map(|x| x.and_then(|s| get_statistic!(s, min, min_bytes, Some(data_type))));
	collect_scalars(data_type, scalars)
	}

Testing

I suggest we add a new module to the existing parquet test in https://github.com/apache/datafusion/blob/main/datafusion/core/tests/parquet_exec.rs

The tests should look like:

let record_batch = make_batch_with_relevant_datatype();
// write batch/batches to file
// open file / extract stats from metadata
// compare stats

I can help writing these tests

I personally suggest:

1. Make a PR with the basic API and a single basic types (like Int/UInt or String) and figure out the test pattern (I can definitely help here)
2. Then we can fill out support for the rest of the types in a follow on PR

cc @tustvold in case you have other ideas

Additional context

This code likely eventually would be good to have in the parquet crate -- see apache/arrow-rs#4328. However, I think initially we should do it in DataFusion to iterate faster and figure out the API before moving it up there

There are a bunch of related improvements that I think become much simpler with this feature:

1. Consolidate statistics aggregation #8229

[alamb]

In terms of sequencing of this feature what I would recommend

First PR

Purpose: Sketch out the API, and test framework

1. Create a test framework for this
2. Create the basic API and extract min/max values for Int64 columns

Second PR (draft)

purpose: demonstrate the API can be used in DataFusion, also ensure test coverage is adequate
Update one of the uses of parquet statistics (like ListingTable) to use the new API. @alamb would like to do this if I have time

Third+Fourth+... PRs

Add support for the remaining datatypes, along with tests
This part can be parallelized into multiple PRs

[NGA-TRAN]

I start working on the first PR

[alamb]

After working through an actual example in #10549 I have a new API proposal: NGA-TRAN#118

Here is what the API looks like

/// What type of statistics should be extracted?
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub enum RequestedStatistics {
    /// Minimum Value
    Min,
    /// Maximum Value
    Max,
    /// Null Count, returned as a [`UInt64Array`])
    NullCount,
}

/// Extracts Parquet statistics as Arrow arrays
///
/// This is used to convert Parquet statistics to Arrow arrays, with proper type
/// conversions. This information can be used for pruning parquet files or row
/// groups based on the statistics embedded in parquet files
///
/// # Schemas
///
/// The schema of the parquet file and the arrow schema are used 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.
///
/// 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 file to the column in the arrow schema.
///
/// # Multiple parquet files
///
/// This API is designed to support efficiently extracting statistics from
/// multiple parquet files (hence why the parquet schema is passed in as an
/// argument). This is useful when building an index for a directory of parquet
/// files.
///
#[derive(Debug)]
pub struct StatisticsConverter<'a> {
    /// The name of the column to extract statistics for
    column_name: &'a str,
    /// The type of statistics to extract
    statistics_type: RequestedStatistics,
    /// The arrow schema of the query
    arrow_schema: &'a Schema,
    /// The field (with data type) of the column in the arrow schema
    arrow_field: &'a Field,
}

impl<'a> StatisticsConverter<'a> {
    /// Returns a [`UInt64Array`] with counts for each row group
    ///
    /// 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.
    pub fn row_counts(metadata: &ParquetMetaData) -> Result<UInt64Array> {
...
    }

    /// create an new statistics converter
    pub fn try_new(
        column_name: &'a str,
        statistics_type: RequestedStatistics,
        arrow_schema: &'a Schema,
    ) -> Result<Self> {
...
    }

    /// extract the statistics from a parquet file, given the parquet file's metadata
    ///
    /// The returned array contains 1 value for each row group in the parquet
    /// file in order
    ///
    /// Each value is either
    /// * the requested statistics type for the column
    /// * a null value, if the statistics can not be extracted
    ///
    /// Note that a null value does NOT mean the min or max value was actually
    /// `null` it means it the requested statistic is unknown
    ///
    /// 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
    pub fn extract(&self, metadata: &ParquetMetaData) -> Result<ArrayRef> {
...
    }
}

I am envisioning this API could also easily support

Extract from multiple files in one go

impl<'a> StatisticsConverter<'a> {
..
/// Extract metadata from multiple parquet files into an single arrow array
/// one element per row group per file
fn extract_multi(&self, metadata: impl IntoIterator<Item = &ParquetMetadata>))-> Result<ArrayRef> {
...
}

Extract information from the page index as well

impl<'a> StatisticsConverter<'a> {
..
/// Extract metadata from page indexes across all row groups. The returned array has one element
/// per page across all row groups
fn extract_page(&self, metadata: impl IntoIterator<Item = &ParquetMetadata>))-> Result<ArrayRef> {
...
}