From 10aafae8e925e11dc7f8c96d93947314a09342f6 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 19 Aug 2026 23:23:27 -0700 Subject: [PATCH] fix: honor Parquet byte-array statistics ordering --- .../benches/parquet_metadata_statistics.rs | 9 +- datafusion/datasource-parquet/src/metadata.rs | 68 +- datafusion/datasource-parquet/src/mod.rs | 2 + .../datasource-parquet/src/opener/mod.rs | 5 +- .../datasource-parquet/src/page_filter.rs | 15 + .../datasource-parquet/src/push_decoder.rs | 5 + .../src/row_group_filter.rs | 121 ++- .../src/statistics_order_tests.rs | 717 ++++++++++++++++++ 8 files changed, 918 insertions(+), 24 deletions(-) create mode 100644 datafusion/datasource-parquet/src/statistics_order_tests.rs diff --git a/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs b/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs index 46ebd100fde88..82147abfd9a01 100644 --- a/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs +++ b/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs @@ -28,6 +28,7 @@ use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main}; use datafusion_datasource_parquet::metadata::DFParquetMetadata; use parquet::arrow::ArrowSchemaConverter; +use parquet::basic::ColumnOrder; use parquet::data_type::ByteArray; use parquet::file::metadata::{ ColumnChunkMetaData, FileMetaData, ParquetMetaData, RowGroupMetaData, @@ -169,13 +170,19 @@ fn make_synthetic_metadata( }) .collect::>(); + // Model a modern writer so byte-array bounds have a trustworthy order. + let column_orders = schema_descr + .columns() + .iter() + .map(|column| ColumnOrder::TYPE_DEFINED_ORDER(column.sort_order())) + .collect(); let file_metadata = FileMetaData::new( 1, (spec.row_groups * ROWS_PER_GROUP) as i64, Some("datafusion parquet metadata benchmark".to_string()), None, schema_descr, - None, + Some(column_orders), ); ParquetMetaData::new(file_metadata, row_groups) diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index c01811c087386..8423894eaba84 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -43,12 +43,13 @@ use object_store::{ObjectMeta, ObjectStore}; use parquet::DecodeResult; use parquet::arrow::arrow_reader::statistics::StatisticsConverter; use parquet::arrow::{parquet_column, parquet_to_arrow_schema}; +use parquet::basic::{ColumnOrder, SortOrder, Type as PhysicalType}; use parquet::file::metadata::{ PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, ParquetMetaDataReader, RowGroupMetaData, SortingColumn, }; use parquet::file::statistics::Statistics as ParquetStatistics; -use parquet::schema::types::SchemaDescriptor; +use parquet::schema::types::{ColumnDescriptor, SchemaDescriptor}; use std::any::Any; use std::sync::Arc; @@ -57,6 +58,54 @@ use std::sync::Arc; /// would be too unreliable otherwise. const PARTIAL_NDV_THRESHOLD: f64 = 0.75; +fn requires_unsigned_byte_array_order(column: &ColumnDescriptor) -> bool { + matches!( + column.physical_type(), + PhysicalType::BYTE_ARRAY | PhysicalType::FIXED_LEN_BYTE_ARRAY + ) && column.sort_order() != SortOrder::SIGNED +} + +/// Whether byte-array bounds lack a recognized unsigned comparison order. +/// +/// The deprecated Parquet `min`/`max` fields use signed comparison, unlike +/// Arrow's string and binary comparisons. Even the modern bounds cannot be +/// interpreted without the corresponding footer `column_orders` entry. +/// Signed logical types, such as decimals, retain their existing behavior. +pub(crate) fn has_untrusted_byte_array_order( + parquet_schema: &SchemaDescriptor, + column_orders: Option<&[ColumnOrder]>, + parquet_column_index: usize, +) -> bool { + let column = parquet_schema.column(parquet_column_index); + requires_unsigned_byte_array_order(&column) + && (column.sort_order() != SortOrder::UNSIGNED + || column_orders + .and_then(|orders| orders.get(parquet_column_index)) + .copied() + != Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED))) +} + +/// Whether any row group provides byte-array bounds in the deprecated signed +/// order, or the column's logical order is undefined. +pub(crate) fn has_untrusted_byte_array_stats<'a>( + parquet_schema: &SchemaDescriptor, + parquet_column_index: Option, + row_groups: impl IntoIterator, +) -> bool { + parquet_column_index.is_some_and(|index| { + let column = parquet_schema.column(index); + requires_unsigned_byte_array_order(&column) + && (column.sort_order() != SortOrder::UNSIGNED + || row_groups.into_iter().any(|group| { + group.column(index).statistics().is_some_and(|stats| { + stats.is_min_max_deprecated() + && (stats.min_bytes_opt().is_some() + || stats.max_bytes_opt().is_some()) + }) + })) + }) +} + /// Handles fetching Parquet file schema, metadata and statistics /// from object store. /// @@ -496,6 +545,23 @@ impl<'a> DFParquetMetadata<'a> { file_metadata.schema_descr(), ) { Ok(stats_converter) => { + let parquet_index = stats_converter.parquet_column_index(); + if parquet_index.is_some_and(|index| { + has_untrusted_byte_array_order( + file_metadata.schema_descr(), + file_metadata.column_orders().map(Vec::as_slice), + index, + ) + }) || has_untrusted_byte_array_stats( + file_metadata.schema_descr(), + parquet_index, + row_groups_metadata, + ) { + // The remaining row groups cannot establish bounds + // for the whole file. Keep unrelated statistics. + min_accs[idx] = None; + max_accs[idx] = None; + } let mut accumulators = StatisticsAccumulators { min_accs: &mut min_accs, max_accs: &mut max_accs, diff --git a/datafusion/datasource-parquet/src/mod.rs b/datafusion/datasource-parquet/src/mod.rs index 35f831230b305..23348f94aae8f 100644 --- a/datafusion/datasource-parquet/src/mod.rs +++ b/datafusion/datasource-parquet/src/mod.rs @@ -42,6 +42,8 @@ mod schema_coercion; mod sink; mod sort; pub mod source; +#[cfg(test)] +mod statistics_order_tests; mod supported_predicates; #[cfg(test)] mod test_util; diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b3ce024d66f1f..0b02fcb71d663 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -1105,10 +1105,9 @@ impl FiltersPreparedParquetOpen { // If there is a predicate that can be evaluated against the metadata if let Some(predicate) = self.pruning_predicate.as_ref().map(|p| p.as_ref()) { if prepared.enable_row_group_stats_pruning { - row_groups.prune_by_statistics( + row_groups.prune_by_statistics_with_metadata( &prepared.physical_file_schema, - loaded.reader_metadata.parquet_schema(), - rg_metadata, + &file_metadata, predicate, &prepared.file_metrics, ); diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs index 6bc1aca667981..75acd431263e1 100644 --- a/datafusion/datasource-parquet/src/page_filter.rs +++ b/datafusion/datasource-parquet/src/page_filter.rs @@ -22,6 +22,7 @@ use std::sync::Arc; use super::metrics::ParquetFileMetrics; use crate::ParquetAccessPlan; +use crate::metadata::has_untrusted_byte_array_order; use arrow::array::BooleanArray; use arrow::{ @@ -490,6 +491,7 @@ struct PagesPruningStatistics<'a> { column_index: &'a ParquetColumnIndex, offset_index: &'a ParquetOffsetIndex, page_offsets: &'a Vec, + trusted_min_max: bool, } impl<'a> PagesPruningStatistics<'a> { @@ -529,6 +531,12 @@ impl<'a> PagesPruningStatistics<'a> { return None; }; let page_offsets = offset_index_metadata.page_locations(); + let file_metadata = parquet_metadata.file_metadata(); + let trusted_min_max = !has_untrusted_byte_array_order( + file_metadata.schema_descr(), + file_metadata.column_orders().map(Vec::as_slice), + parquet_column_index, + ); Some(Self { row_group_index, @@ -537,6 +545,7 @@ impl<'a> PagesPruningStatistics<'a> { column_index, offset_index, page_offsets, + trusted_min_max, }) } @@ -563,6 +572,9 @@ impl<'a> PagesPruningStatistics<'a> { } impl PruningStatistics for PagesPruningStatistics<'_> { fn min_values(&self, _column: &datafusion_common::Column) -> Option { + if !self.trusted_min_max { + return None; + } match self.converter.data_page_mins( self.column_index, self.offset_index, @@ -577,6 +589,9 @@ impl PruningStatistics for PagesPruningStatistics<'_> { } fn max_values(&self, _column: &datafusion_common::Column) -> Option { + if !self.trusted_min_max { + return None; + } match self.converter.data_page_maxes( self.column_index, self.offset_index, diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 74d8997198872..d91d86a657902 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -211,6 +211,11 @@ impl RowGroupPruner { .collect::>(); let stats = RowGroupPruningStatistics { parquet_schema: self.parquet_metadata.file_metadata().schema_descr(), + column_orders: self + .parquet_metadata + .file_metadata() + .column_orders() + .map(Vec::as_slice), row_group_metadatas, arrow_schema: self.arrow_schema.as_ref(), // Match the existing static row-group pruning behavior: when a diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index 9231d0232566b..c363a4ed47953 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -20,7 +20,9 @@ use std::sync::Arc; use super::{ParquetAccessPlan, ParquetFileMetrics, RowGroupAccess}; use crate::bloom_filter::BloomFilterStatistics; +use crate::metadata::{has_untrusted_byte_array_order, has_untrusted_byte_array_stats}; use arrow::array::{ArrayRef, BooleanArray, UInt64Array}; +use arrow::compute::nullif; use arrow::datatypes::Schema; use datafusion_common::pruning::PruningStatistics; use datafusion_common::{Column, Result, ScalarValue}; @@ -31,7 +33,8 @@ use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr::{PhysicalExpr, PhysicalExprSimplifier}; use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder}; use parquet::arrow::arrow_reader::statistics::StatisticsConverter; -use parquet::file::metadata::RowGroupMetaData; +use parquet::basic::ColumnOrder; +use parquet::file::metadata::{ParquetMetaData, RowGroupMetaData}; use parquet::schema::types::SchemaDescriptor; /// Reduces the [`ParquetAccessPlan`] based on row group level metadata. @@ -252,8 +255,10 @@ impl RowGroupAccessPlanFilter { /// /// Updates this set to mark row groups that should not be scanned /// - /// Note: This method currently ignores ColumnOrder - /// + /// This method has no file footer, so it cannot establish the sort order + /// of string or binary min/max statistics. Such bounds are ignored. Use + /// [`Self::prune_by_statistics_with_metadata`] when the full metadata is + /// available. Null counts and other columns' statistics remain usable. /// /// # Panics /// if `groups.len() != self.len()` @@ -264,6 +269,51 @@ impl RowGroupAccessPlanFilter { groups: &[RowGroupMetaData], predicate: &PruningPredicate, metrics: &ParquetFileMetrics, + ) { + self.prune_by_statistics_inner( + arrow_schema, + parquet_schema, + groups, + None, + predicate, + metrics, + ); + } + + /// Prune row groups using statistics and the comparison orders recorded + /// in the Parquet file footer. + /// + /// Unlike [`Self::prune_by_statistics`], this method can use string and + /// binary bounds when their ordering is known to match Arrow's ordering. + /// + /// # Panics + /// if `metadata.num_row_groups() != self.len()` + pub fn prune_by_statistics_with_metadata( + &mut self, + arrow_schema: &Schema, + metadata: &ParquetMetaData, + predicate: &PruningPredicate, + metrics: &ParquetFileMetrics, + ) { + let file_metadata = metadata.file_metadata(); + self.prune_by_statistics_inner( + arrow_schema, + file_metadata.schema_descr(), + metadata.row_groups(), + file_metadata.column_orders().map(Vec::as_slice), + predicate, + metrics, + ); + } + + fn prune_by_statistics_inner( + &mut self, + arrow_schema: &Schema, + parquet_schema: &SchemaDescriptor, + groups: &[RowGroupMetaData], + column_orders: Option<&[ColumnOrder]>, + predicate: &PruningPredicate, + metrics: &ParquetFileMetrics, ) { // scoped timer updates on drop let _timer_guard = metrics.statistics_eval_time.timer(); @@ -278,6 +328,7 @@ impl RowGroupAccessPlanFilter { let pruning_stats = RowGroupPruningStatistics { parquet_schema, + column_orders, row_group_metadatas, arrow_schema, // Preserve the existing row-group pruning behavior. This path only @@ -304,8 +355,7 @@ impl RowGroupAccessPlanFilter { // Check if any of the matched row groups are fully contained by the predicate self.identify_fully_matched_row_groups( &fully_contained_candidates_original_idx, - arrow_schema, - parquet_schema, + &pruning_stats, groups, predicate, metrics, @@ -330,8 +380,7 @@ impl RowGroupAccessPlanFilter { fn identify_fully_matched_row_groups( &mut self, candidate_row_group_indices: &[usize], - arrow_schema: &Schema, - parquet_schema: &SchemaDescriptor, + pruning_stats: &RowGroupPruningStatistics<'_>, groups: &[RowGroupMetaData], predicate: &PruningPredicate, metrics: &ParquetFileMetrics, @@ -339,6 +388,7 @@ impl RowGroupAccessPlanFilter { if candidate_row_group_indices.is_empty() { return; } + let arrow_schema = pruning_stats.arrow_schema; let mut inverted_expr: Arc = Arc::new(NotExpr::new(Arc::clone(predicate.orig_expr()))); @@ -381,7 +431,8 @@ impl RowGroupAccessPlanFilter { }; let inverted_pruning_stats = RowGroupPruningStatistics { - parquet_schema, + parquet_schema: pruning_stats.parquet_schema, + column_orders: pruning_stats.column_orders, row_group_metadatas: candidate_row_group_indices .iter() .map(|&i| &groups[i]) @@ -472,6 +523,7 @@ impl RowGroupAccessPlanFilter { /// duplicating the statistics-to-`PruningStatistics` plumbing. pub(crate) struct RowGroupPruningStatistics<'a> { pub(crate) parquet_schema: &'a SchemaDescriptor, + pub(crate) column_orders: Option<&'a [ColumnOrder]>, pub(crate) row_group_metadatas: Vec<&'a RowGroupMetaData>, pub(crate) arrow_schema: &'a Schema, pub(crate) missing_null_counts_as_zero: bool, @@ -479,14 +531,11 @@ pub(crate) struct RowGroupPruningStatistics<'a> { impl<'a> RowGroupPruningStatistics<'a> { /// Return an iterator over the row group metadata - fn metadata_iter(&'a self) -> impl Iterator + 'a { + fn metadata_iter(&self) -> impl Iterator + '_ { self.row_group_metadatas.iter().copied() } - fn statistics_converter<'b>( - &'a self, - column: &'b Column, - ) -> Result> { + fn statistics_converter(&self, column: &Column) -> Result> { Ok(StatisticsConverter::try_new( &column.name, self.arrow_schema, @@ -494,19 +543,53 @@ impl<'a> RowGroupPruningStatistics<'a> { )? .with_missing_null_counts_as_zero(self.missing_null_counts_as_zero)) } + + fn min_max_statistics_converter( + &self, + column: &Column, + ) -> Option> { + let converter = self.statistics_converter(column).ok()?; + let parquet_index = converter.parquet_column_index(); + if parquet_index.is_some_and(|index| { + has_untrusted_byte_array_order(self.parquet_schema, self.column_orders, index) + }) { + return None; + } + Some(converter) + } + + fn mask_untrusted_byte_array_stats( + &self, + parquet_index: Option, + values: ArrayRef, + ) -> Option { + if !has_untrusted_byte_array_stats( + self.parquet_schema, + parquet_index, + self.metadata_iter(), + ) { + return Some(values); + } + // A file may mix legacy and modern row-group statistics. Keep the + // modern bounds usable rather than discarding this entire column. + let mask = BooleanArray::from_iter(self.metadata_iter().map(|group| { + has_untrusted_byte_array_stats(self.parquet_schema, parquet_index, [group]) + })); + nullif(values.as_ref(), &mask).ok() + } } impl PruningStatistics for RowGroupPruningStatistics<'_> { fn min_values(&self, column: &Column) -> Option { - self.statistics_converter(column) - .and_then(|c| Ok(c.row_group_mins(self.metadata_iter())?)) - .ok() + let converter = self.min_max_statistics_converter(column)?; + let values = converter.row_group_mins(self.metadata_iter()).ok()?; + self.mask_untrusted_byte_array_stats(converter.parquet_column_index(), values) } fn max_values(&self, column: &Column) -> Option { - self.statistics_converter(column) - .and_then(|c| Ok(c.row_group_maxes(self.metadata_iter())?)) - .ok() + let converter = self.min_max_statistics_converter(column)?; + let values = converter.row_group_maxes(self.metadata_iter()).ok()?; + self.mask_untrusted_byte_array_stats(converter.parquet_column_index(), values) } fn num_containers(&self) -> usize { diff --git a/datafusion/datasource-parquet/src/statistics_order_tests.rs b/datafusion/datasource-parquet/src/statistics_order_tests.rs new file mode 100644 index 0000000000000..f678fa6b47ee2 --- /dev/null +++ b/datafusion/datasource-parquet/src/statistics_order_tests.rs @@ -0,0 +1,717 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Regression tests for interpreting Parquet byte-array statistics orders. + +use std::io::Write; +use std::sync::Arc; + +use arrow::array::{BooleanArray, record_batch}; +use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use bytes::Bytes; +use datafusion_common::pruning::{PrunableStatistics, PruningStatistics}; +use datafusion_common::stats::Precision; +use datafusion_common::{Column, ScalarValue, Statistics}; +use datafusion_expr::{Expr, col, lit}; +use datafusion_physical_expr::PhysicalExpr; +use datafusion_physical_expr::planner::logical2physical; +use datafusion_physical_plan::metrics::{Count, ExecutionPlanMetricsSet}; +use datafusion_pruning::{MAX_IN_LIST_SIZE, PruningPredicate, PruningPredicateBuilder}; +use parquet::arrow::ArrowWriter; +use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; +use parquet::basic::{ColumnOrder, LogicalType, SortOrder, Type as PhysicalType}; +use parquet::data_type::{ByteArray, FixedLenByteArray}; +use parquet::file::metadata::{ + ColumnChunkMetaData, ColumnIndexBuilder, FileMetaData, OffsetIndexBuilder, + PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter, + RowGroupMetaData, +}; +use parquet::file::properties::{EnabledStatistics, WriterProperties}; +use parquet::file::statistics::Statistics as ParquetStatistics; +use parquet::file::writer::TrackedWrite; +use parquet::schema::types::{SchemaDescriptor, Type as ParquetType}; + +use crate::RowGroupAccessPlanFilter; +use crate::metadata::DFParquetMetadata; +use crate::push_decoder::RowGroupPruner; +use crate::row_group_filter::RowGroupPruningStatistics; +use crate::{PagePruningAccessPlanFilter, ParquetAccessPlan, ParquetFileMetrics}; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StatisticsOrder { + Modern, + Deprecated, + Missing, + Unknown, +} + +struct TestFile { + bytes: Bytes, + schema: SchemaRef, + metadata: Arc, +} + +impl TestFile { + fn new(order: StatisticsOrder) -> Self { + let batch = record_batch!( + ( + "s", + Utf8, + vec![ + Some("aé"), + Some("az"), + Some("b"), + None, + None, + None, + Some("d"), + Some("e"), + Some("f"), + ] + ), + ("n", Int32, [1, 2, 3, 10, 11, 12, 20, 21, 22]) + ) + .unwrap(); + let schema = batch.schema(); + let properties = WriterProperties::builder() + .set_max_row_group_row_count(Some(3)) + .set_data_page_row_count_limit(3) + .set_write_batch_size(3) + .set_dictionary_enabled(false) + .set_statistics_enabled(EnabledStatistics::Page) + .build(); + let mut original = Vec::new(); + let mut writer = + ArrowWriter::try_new(&mut original, Arc::clone(&schema), Some(properties)) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let original = Bytes::from(original); + let metadata = read_metadata(&original); + assert_eq!(metadata.num_row_groups(), 3); + + if order == StatisticsOrder::Modern { + return Self { + bytes: original, + schema, + metadata: Arc::new(metadata), + }; + } + + // Signed-byte comparison gives ["aé", "b"] for the first row + // group's ["aé", "az", "b"]. The endpoints are not inverted in + // unsigned order, but the interval wrongly excludes "az". + let mut row_groups = metadata.row_groups().to_vec(); + let mut columns = row_groups[0].columns().to_vec(); + columns[0] = columns[0] + .clone() + .into_builder() + .set_statistics(ParquetStatistics::byte_array( + Some(ByteArray::from("aé")), + Some(ByteArray::from("b")), + None, + Some(0), + order == StatisticsOrder::Deprecated, + )) + .build() + .unwrap(); + row_groups[0] = row_groups[0] + .clone() + .into_builder() + .set_column_metadata(columns) + .build() + .unwrap(); + + let mut column_index = metadata.column_index().unwrap().clone(); + if matches!(order, StatisticsOrder::Missing | StatisticsOrder::Unknown) { + let mut index = ColumnIndexBuilder::new(PhysicalType::BYTE_ARRAY); + index.append(false, "aé".as_bytes().to_vec(), b"b".to_vec(), 0); + column_index[0][0] = index.build().unwrap(); + } + let metadata = metadata + .into_builder() + .set_row_groups(row_groups) + .set_column_index(Some(column_index)) + .build(); + + // Keep the real data pages, and serialize the replacement statistics + // and page indexes at their actual file offsets. + let mut bytes = Vec::new(); + let mut tracked = TrackedWrite::new(&mut bytes); + tracked + .write_all(&original[..footer_start(&original)]) + .unwrap(); + ParquetMetaDataWriter::new_with_tracked(tracked, &metadata) + .finish() + .unwrap(); + + // parquet-rs always writes TYPEORDER and cannot write an unknown + // ColumnOrder. The final Thrift field is column_orders (field 7): a + // two-element list of unions, followed by the FileMetaData STOP. + // Alter just that field to model old and future writers, then verify + // the decoded footer below. No data or page-index offsets change. + if matches!(order, StatisticsOrder::Missing | StatisticsOrder::Unknown) { + let end = bytes.len() - 8; + let encoded_orders = [0x19, 0x2c, 0x1c, 0, 0, 0x1c, 0, 0, 0]; + let start = end - encoded_orders.len(); + assert_eq!(&bytes[start..end], &encoded_orders); + let metadata_start = footer_start(&bytes); + if order == StatisticsOrder::Missing { + bytes.drain(start..end - 1); + let new_end = bytes.len() - 8; + let metadata_len = (new_end - metadata_start) as u32; + bytes[new_end..new_end + 4].copy_from_slice(&metadata_len.to_le_bytes()); + } else { + // Change the first union member from field 1 (TYPEORDER) to + // an unrecognized field 2. The numeric column stays known. + bytes[start + 2] = 0x2c; + } + } + + let bytes = Bytes::from(bytes); + let metadata = read_metadata(&bytes); + let expected_order = match order { + StatisticsOrder::Missing => ColumnOrder::UNDEFINED, + StatisticsOrder::Unknown => ColumnOrder::UNKNOWN, + _ => ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED), + }; + assert_eq!(metadata.file_metadata().column_order(0), expected_order); + assert_eq!( + metadata + .row_group(0) + .column(0) + .statistics() + .unwrap() + .is_min_max_deprecated(), + order == StatisticsOrder::Deprecated, + ); + Self { + bytes, + schema, + metadata: Arc::new(metadata), + } + } + + fn predicate(&self, expr: &Expr) -> (Arc, PruningPredicate) { + let physical = logical2physical(expr, &self.schema); + let pruning = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&self.schema)) + .try_build(Arc::clone(&physical)) + .unwrap(); + (physical, pruning) + } + + fn statistics(&self) -> Statistics { + DFParquetMetadata::statistics_from_parquet_metadata(&self.metadata, &self.schema) + .unwrap() + } + + fn file_matches(&self, predicate: &PruningPredicate) -> bool { + let stats = PrunableStatistics::new( + vec![Arc::new(self.statistics())], + Arc::clone(&self.schema), + ); + predicate.prune(&stats).unwrap()[0] + } + + fn row_group_plan(&self, predicate: &PruningPredicate) -> ParquetAccessPlan { + let mut filter = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all( + self.metadata.num_row_groups(), + )); + filter.prune_by_statistics_with_metadata( + &self.schema, + &self.metadata, + predicate, + &metrics(), + ); + filter.build() + } + + fn page_plan( + &self, + physical: &Arc, + plan: ParquetAccessPlan, + ) -> ParquetAccessPlan { + PagePruningAccessPlanFilter::new(physical, Arc::clone(&self.schema)) + .prune_plan_with_page_index( + plan, + &self.schema, + self.metadata.file_metadata().schema_descr(), + &self.metadata, + &metrics(), + ) + } + + fn matching_rows( + &self, + physical: &Arc, + plan: ParquetAccessPlan, + ) -> usize { + let mut builder = ParquetRecordBatchReaderBuilder::try_new(self.bytes.clone()) + .unwrap() + .with_row_groups(plan.row_group_indexes()); + if let Some(selection) = plan + .into_overall_row_selection(self.metadata.row_groups()) + .unwrap() + { + builder = builder.with_row_selection(selection); + } + builder + .build() + .unwrap() + .map(|batch| { + let batch = batch.unwrap(); + let matches = physical + .evaluate(&batch) + .unwrap() + .into_array(batch.num_rows()) + .unwrap(); + matches + .as_any() + .downcast_ref::() + .unwrap() + .true_count() + }) + .sum() + } +} + +fn footer_start(bytes: &[u8]) -> usize { + let end = bytes.len() - 8; + let metadata_len = u32::from_le_bytes(bytes[end..end + 4].try_into().unwrap()); + end - metadata_len as usize +} + +fn read_metadata(bytes: &Bytes) -> ParquetMetaData { + ParquetMetaDataReader::new() + .with_page_index_policy(PageIndexPolicy::Required) + .parse_and_finish(bytes) + .unwrap() +} + +fn metrics() -> ParquetFileMetrics { + ParquetFileMetrics::new( + 0, + "statistics-order.parquet", + &ExecutionPlanMetricsSet::new(), + ) +} + +#[test] +fn byte_array_order_preserves_matching_rows_at_every_pruning_level() { + for order in [ + StatisticsOrder::Modern, + StatisticsOrder::Deprecated, + StatisticsOrder::Missing, + StatisticsOrder::Unknown, + ] { + let file = TestFile::new(order); + let (physical, predicate) = file.predicate(&col("s").eq(lit("az"))); + let all = ParquetAccessPlan::new_all(file.metadata.num_row_groups()); + assert_eq!(file.matching_rows(&physical, all.clone()), 1); + assert!(file.file_matches(&predicate), "order={order:?}"); + let row_groups = file.row_group_plan(&predicate); + assert!(row_groups.should_scan(0), "order={order:?}"); + assert!(!row_groups.is_fully_matched(0), "order={order:?}"); + assert_eq!( + row_groups.row_group_indexes(), + if matches!(order, StatisticsOrder::Modern | StatisticsOrder::Deprecated) { + vec![0] + } else { + vec![0, 2] + }, + ); + assert_eq!(file.matching_rows(&physical, row_groups.clone()), 1); + assert_eq!( + file.matching_rows(&physical, file.page_plan(&physical, all)), + 1 + ); + assert_eq!( + file.matching_rows(&physical, file.page_plan(&physical, row_groups)), + 1, + "order={order:?}", + ); + + let mut runtime_pruner = RowGroupPruner::new( + physical, + Arc::clone(&file.schema), + Arc::clone(&file.metadata), + Count::new(), + Count::new(), + MAX_IN_LIST_SIZE, + ); + assert!(!runtime_pruner.should_prune(&[0]), "order={order:?}"); + assert!(runtime_pruner.should_prune(&[1]), "all-null row group"); + } +} + +#[test] +fn byte_array_order_keeps_null_counts_and_unrelated_column_bounds() { + for order in [ + StatisticsOrder::Deprecated, + StatisticsOrder::Missing, + StatisticsOrder::Unknown, + ] { + let file = TestFile::new(order); + let statistics = file.statistics(); + let string = &statistics.column_statistics[0]; + assert_eq!(string.min_value, Precision::Absent, "order={order:?}"); + assert_eq!(string.max_value, Precision::Absent, "order={order:?}"); + assert_eq!(string.null_count, Precision::Exact(3)); + assert_eq!( + statistics.column_statistics[1].min_value, + Precision::Exact(ScalarValue::Int32(Some(1))), + ); + assert_eq!( + statistics.column_statistics[1].max_value, + Precision::Exact(ScalarValue::Int32(Some(22))), + ); + + let (physical, predicate) = file.predicate(&col("s").is_null()); + assert_eq!(file.row_group_plan(&predicate).row_group_indexes(), vec![1]); + let page_plan = file.page_plan(&physical, ParquetAccessPlan::new_all(3)); + assert_eq!(page_plan.row_group_indexes(), vec![1]); + assert_eq!(file.matching_rows(&physical, page_plan), 3); + + let expr = col("s").eq(lit("az")).and(col("n").eq(lit(99))); + let (physical, predicate) = file.predicate(&expr); + assert!(!file.file_matches(&predicate)); + assert!( + file.row_group_plan(&predicate) + .row_group_indexes() + .is_empty() + ); + assert!( + file.page_plan(&physical, ParquetAccessPlan::new_all(3)) + .row_group_indexes() + .is_empty(), + ); + } +} + +#[test] +fn byte_array_order_modern_bounds_and_null_only_groups_remain_usable() { + let file = TestFile::new(StatisticsOrder::Modern); + let statistics = file.statistics(); + assert_eq!( + statistics.column_statistics[0].min_value, + Precision::Exact(ScalarValue::Utf8(Some("az".to_owned()))), + ); + let (physical, predicate) = file.predicate(&col("s").eq(lit("az"))); + assert_eq!(file.row_group_plan(&predicate).row_group_indexes(), vec![0]); + assert_eq!( + file.page_plan(&physical, ParquetAccessPlan::new_all(3)) + .row_group_indexes(), + vec![0], + ); + + // The compatibility API has no footer. It still uses null counts, but + // cannot assume that even modern-looking byte-array bounds are unsigned. + let mut filter = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(3)); + filter.prune_by_statistics( + &file.schema, + file.metadata.file_metadata().schema_descr(), + file.metadata.row_groups(), + &predicate, + &metrics(), + ); + assert_eq!(filter.build().row_group_indexes(), vec![0, 2]); +} + +#[test] +fn byte_array_order_guard_follows_parquet_type_not_arrow_representation() { + let file = TestFile::new(StatisticsOrder::Deprecated); + let metadata = file.metadata.file_metadata(); + let column = Column::from_name("s"); + for data_type in [ + DataType::Utf8, + DataType::LargeUtf8, + DataType::Utf8View, + DataType::Binary, + DataType::LargeBinary, + DataType::BinaryView, + DataType::Dictionary(Box::new(DataType::Int8), Box::new(DataType::Utf8)), + ] { + let schema = Schema::new(vec![Field::new("s", data_type, true)]); + let stats = RowGroupPruningStatistics { + parquet_schema: metadata.schema_descr(), + column_orders: metadata.column_orders().map(Vec::as_slice), + row_group_metadatas: file.metadata.row_groups().iter().collect(), + arrow_schema: &schema, + missing_null_counts_as_zero: true, + }; + for values in [stats.min_values(&column), stats.max_values(&column)] { + let values = values.unwrap(); + assert!(values.is_null(0)); + assert!(values.is_null(1)); + assert!(!values.is_null(2)); + } + assert!(stats.null_counts(&column).is_some()); + } +} + +fn single_column_metadata( + parquet_type: ParquetType, + statistics: ParquetStatistics, + order: Option, +) -> ParquetMetaData { + let physical_type = parquet_type.get_physical_type(); + let schema = Arc::new(SchemaDescriptor::new(Arc::new( + ParquetType::group_type_builder("schema") + .with_fields(vec![Arc::new(parquet_type)]) + .build() + .unwrap(), + ))); + let mut column_index = ColumnIndexBuilder::new(physical_type); + column_index.append( + false, + statistics.min_bytes_opt().unwrap().to_vec(), + statistics.max_bytes_opt().unwrap().to_vec(), + 0, + ); + let mut offset_index = OffsetIndexBuilder::new(); + offset_index.append_row_count(3); + offset_index.append_offset_and_size(0, 1); + let column = ColumnChunkMetaData::builder(schema.column(0)) + .set_num_values(3) + .set_statistics(statistics) + .build() + .unwrap(); + let group = RowGroupMetaData::builder(Arc::clone(&schema)) + .set_num_rows(3) + .set_column_metadata(vec![column]) + .build() + .unwrap(); + ParquetMetaData::new( + FileMetaData::new(1, 3, None, None, schema, order.map(|order| vec![order])), + vec![group], + ) + .into_builder() + .set_column_index(Some(vec![vec![column_index.build().unwrap()]])) + .set_offset_index(Some(vec![vec![offset_index.build()]])) + .build() +} + +#[test] +fn fixed_byte_array_and_uuid_orders_guard_bounds_but_not_null_counts() { + let schema = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::FixedSizeBinary(16), + true, + )])); + let padded = |prefix: &[u8]| { + let mut value = vec![0; 16]; + value[..prefix.len()].copy_from_slice(prefix); + value + }; + let min = padded("aé".as_bytes()); + let max = padded(b"b"); + let value = ScalarValue::FixedSizeBinary(16, Some(padded(b"az"))); + let physical = logical2physical(&col("s").eq(lit(value)), &schema); + let predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(Arc::clone(&physical)) + .unwrap(); + + for logical_type in [None, Some(LogicalType::Uuid)] { + for deprecated in [false, true] { + for order in [ + None, + Some(ColumnOrder::UNKNOWN), + Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED)), + Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED)), + ] { + let parquet_type = ParquetType::primitive_type_builder( + "s", + PhysicalType::FIXED_LEN_BYTE_ARRAY, + ) + .with_length(16) + .with_logical_type(logical_type.clone()) + .build() + .unwrap(); + let statistics = ParquetStatistics::fixed_len_byte_array( + Some(FixedLenByteArray::from(min.clone())), + Some(FixedLenByteArray::from(max.clone())), + None, + Some(0), + deprecated, + ); + let metadata = single_column_metadata(parquet_type, statistics, order); + let trusted = + order == Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED)); + let statistics = DFParquetMetadata::statistics_from_parquet_metadata( + &metadata, &schema, + ) + .unwrap(); + assert_eq!( + statistics.column_statistics[0].min_value, + if trusted && !deprecated { + Precision::Exact(ScalarValue::FixedSizeBinary( + 16, + Some(min.clone()), + )) + } else { + Precision::Absent + }, + ); + assert_eq!( + statistics.column_statistics[0].null_count, + Precision::Exact(0), + ); + + let mut row_groups = + RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(1)); + row_groups.prune_by_statistics_with_metadata( + &schema, + &metadata, + &predicate, + &metrics(), + ); + assert_eq!(row_groups.build().should_scan(0), !trusted || deprecated); + let pages = + PagePruningAccessPlanFilter::new(&physical, Arc::clone(&schema)) + .prune_plan_with_page_index( + ParquetAccessPlan::new_all(1), + &schema, + metadata.file_metadata().schema_descr(), + &metadata, + &metrics(), + ); + // Modern page indexes are independent of legacy row-group + // bounds, but still need a recognized footer order. + assert_eq!(pages.should_scan(0), !trusted); + } + } + } +} + +#[test] +fn signed_decimal_byte_array_statistics_remain_usable() { + let schema = Arc::new(Schema::new(vec![Field::new( + "s", + DataType::Decimal128(10, 0), + true, + )])); + for physical_type in [PhysicalType::BYTE_ARRAY, PhysicalType::FIXED_LEN_BYTE_ARRAY] { + let parquet_type = ParquetType::primitive_type_builder("s", physical_type) + .with_length(16) + .with_logical_type(Some(LogicalType::decimal(0, 10))) + .with_precision(10) + .with_scale(0) + .build() + .unwrap(); + let min = (-2_i128).to_be_bytes().to_vec(); + let max = 3_i128.to_be_bytes().to_vec(); + let statistics = match physical_type { + PhysicalType::BYTE_ARRAY => ParquetStatistics::byte_array( + Some(ByteArray::from(min)), + Some(ByteArray::from(max)), + None, + Some(0), + true, + ), + _ => ParquetStatistics::fixed_len_byte_array( + Some(FixedLenByteArray::from(min)), + Some(FixedLenByteArray::from(max)), + None, + Some(0), + true, + ), + }; + let metadata = single_column_metadata(parquet_type, statistics, None); + let statistics = + DFParquetMetadata::statistics_from_parquet_metadata(&metadata, &schema) + .unwrap(); + assert_eq!( + statistics.column_statistics[0].min_value, + Precision::Exact(ScalarValue::Decimal128(Some(-2), 10, 0)), + ); + let physical = logical2physical( + &col("s").eq(lit(ScalarValue::Decimal128(Some(4), 10, 0))), + &schema, + ); + let predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(physical) + .unwrap(); + let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(1)); + row_groups.prune_by_statistics_with_metadata( + &schema, + &metadata, + &predicate, + &metrics(), + ); + assert!(!row_groups.build().should_scan(0)); + } +} + +#[test] +fn undefined_logical_byte_array_order_is_not_a_bound() { + let parquet_type = ParquetType::primitive_type_builder("s", PhysicalType::BYTE_ARRAY) + .with_logical_type(Some(LogicalType::_Unknown { field_id: 100 })) + .build() + .unwrap(); + let statistics = ParquetStatistics::byte_array( + Some(ByteArray::from("aé")), + Some(ByteArray::from("b")), + None, + Some(0), + false, + ); + let metadata = single_column_metadata( + parquet_type, + statistics, + Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED)), + ); + let schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Utf8, true)])); + let file_statistics = + DFParquetMetadata::statistics_from_parquet_metadata(&metadata, &schema).unwrap(); + assert_eq!( + file_statistics.column_statistics[0].min_value, + Precision::Absent + ); + assert_eq!( + file_statistics.column_statistics[0].max_value, + Precision::Absent + ); + assert_eq!( + file_statistics.column_statistics[0].null_count, + Precision::Exact(0) + ); + let physical = logical2physical(&col("s").eq(lit("az")), &schema); + let predicate = PruningPredicateBuilder::new() + .with_file_schema(Arc::clone(&schema)) + .try_build(Arc::clone(&physical)) + .unwrap(); + let mut row_groups = RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(1)); + row_groups.prune_by_statistics_with_metadata( + &schema, + &metadata, + &predicate, + &metrics(), + ); + assert!(row_groups.build().should_scan(0)); + let pages = PagePruningAccessPlanFilter::new(&physical, Arc::clone(&schema)) + .prune_plan_with_page_index( + ParquetAccessPlan::new_all(1), + &schema, + metadata.file_metadata().schema_descr(), + &metadata, + &metrics(), + ); + assert!(pages.should_scan(0)); +}