diff --git a/datafusion/core/tests/parquet/file_statistics.rs b/datafusion/core/tests/parquet/file_statistics.rs index e171389b8fef6..14f23115a6647 100644 --- a/datafusion/core/tests/parquet/file_statistics.rs +++ b/datafusion/core/tests/parquet/file_statistics.rs @@ -3,8 +3,10 @@ // SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements use std::fs; +use std::io::Write; use std::sync::Arc; +use arrow::array::{Array, Float64Array, RecordBatch}; use arrow::datatypes::{DataType, Field, Schema}; use datafusion::datasource::TableProvider; use datafusion::datasource::file_format::parquet::ParquetFormat; @@ -33,8 +35,130 @@ use datafusion_physical_optimizer::filter_pushdown::FilterPushdown; use datafusion_physical_plan::ExecutionPlan; use datafusion_physical_plan::filter::FilterExec; use datafusion_physical_plan::statistics::{StatisticsArgs, StatisticsContext}; +use parquet::arrow::ArrowWriter; +use parquet::basic::{ColumnOrder, SortOrder}; +use parquet::file::metadata::{ + ParquetMetaData, ParquetMetaDataReader, ParquetMetaDataWriter, +}; +use parquet::file::properties::{EnabledStatistics, WriterProperties}; +use parquet::file::statistics::{Statistics, ValueStatistics}; +use parquet::file::writer::TrackedWrite; use tempfile::tempdir; +#[tokio::test] +async fn min_max_preserve_signed_zero_with_type_order() { + for zero in [0.0_f64, -0.0_f64] { + // Arrange + let temp_dir = tempdir().unwrap(); + let path = temp_dir.path().join("zeros.parquet"); + let schema = + Arc::new(Schema::new(vec![Field::new("x", DataType::Float64, false)])); + let batch = RecordBatch::try_new( + schema, + vec![Arc::new(Float64Array::from(vec![zero; 3]))], + ) + .unwrap(); + let properties = WriterProperties::builder() + .set_statistics_enabled(EnabledStatistics::Chunk) + .build(); + let mut original = Vec::new(); + let mut writer = + ArrowWriter::try_new(&mut original, batch.schema(), Some(properties)) + .unwrap(); + writer.write(&batch).unwrap(); + let metadata = writer.close().unwrap(); + + // TYPE_ORDER can normalize zero bounds even when only one sign occurs. + // A zero NaN count keeps the NaN guard from masking the order check. + let column = metadata + .row_group(0) + .column(0) + .clone() + .into_builder() + .set_statistics(Statistics::Double( + ValueStatistics::new(Some(-0.0), Some(0.0), None, Some(0), false) + .with_nan_count(Some(0)), + )) + .build() + .unwrap(); + let row_group = metadata + .row_group(0) + .clone() + .into_builder() + .set_column_metadata(vec![column]) + .build() + .unwrap(); + let metadata = + ParquetMetaData::new(metadata.file_metadata().clone(), vec![row_group]); + let footer_len = u32::from_le_bytes( + original[original.len() - 8..original.len() - 4] + .try_into() + .unwrap(), + ); + original.truncate(original.len() - 8 - footer_len as usize); + let mut bytes = Vec::new(); + let mut output = TrackedWrite::new(&mut bytes); + output.write_all(&original).unwrap(); + ParquetMetaDataWriter::new_with_tracked(output, &metadata) + .finish() + .unwrap(); + + // parquet-rs writes IEEE order for floats. Change only the final + // column_orders union tag from field 2 (IEEE) to field 1 (TYPE_ORDER). + let end = bytes.len() - 8; + let encoded_orders = [0x19, 0x1c, 0x2c, 0, 0, 0]; + let start = end - encoded_orders.len(); + assert_eq!(&bytes[start..end], &encoded_orders); + bytes[start + 2] = 0x1c; + assert_eq!(&bytes[..4], b"PAR1"); + assert_eq!(&bytes[bytes.len() - 4..], b"PAR1"); + fs::write(&path, bytes).unwrap(); + + let metadata = ParquetMetaDataReader::new() + .parse_and_finish(&fs::File::open(&path).unwrap()) + .unwrap(); + assert_eq!(metadata.num_row_groups(), 1); + assert_eq!( + metadata.file_metadata().column_order(0), + ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED) + ); + let stats = metadata.row_group(0).column(0).statistics().unwrap(); + assert!(!stats.is_min_max_deprecated()); + assert_eq!(stats.nan_count_opt(), Some(0)); + let Statistics::Double(stats) = stats else { + panic!("expected Float64 statistics"); + }; + assert_eq!(stats.min_opt().unwrap().to_bits(), (-0.0_f64).to_bits()); + assert_eq!(stats.max_opt().unwrap().to_bits(), 0.0_f64.to_bits()); + + let ctx = SessionContext::new_with_config( + SessionConfig::new().with_collect_statistics(true), + ); + ctx.register_parquet("t", path.to_str().unwrap(), ParquetReadOptions::default()) + .await + .unwrap(); + + // Act + let batches = ctx + .sql("SELECT MIN(x), MAX(x) FROM t") + .await + .unwrap() + .collect() + .await + .unwrap(); + + // Assert + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 1); + assert_eq!(batches[0].num_columns(), 2); + for column in batches[0].columns() { + let result = column.as_any().downcast_ref::().unwrap(); + assert!(!result.is_null(0)); + assert_eq!(result.value(0).to_bits(), zero.to_bits()); + } + } +} + #[tokio::test] async fn check_stats_precision_with_filter_pushdown() { let testdata = datafusion::test_util::parquet_test_data(); diff --git a/datafusion/core/tests/parquet/schema_coercion.rs b/datafusion/core/tests/parquet/schema_coercion.rs index b191bf8439722..abebe51e6bd53 100644 --- a/datafusion/core/tests/parquet/schema_coercion.rs +++ b/datafusion/core/tests/parquet/schema_coercion.rs @@ -5,16 +5,16 @@ use std::sync::Arc; use arrow::array::{ - ArrayRef, DictionaryArray, Float32Array, Int64Array, RecordBatch, StringArray, - types::Int32Type, + ArrayRef, DictionaryArray, Float32Array, Float64Array, Int64Array, RecordBatch, + StringArray, types::Int32Type, }; use arrow::datatypes::{DataType, Field, Schema}; use datafusion::datasource::physical_plan::ParquetSource; use datafusion::physical_plan::collect; -use datafusion::prelude::SessionContext; +use datafusion::prelude::{ParquetReadOptions, SessionConfig, SessionContext}; use datafusion::test::object_store::local_unpartitioned_file; -use datafusion_common::Result; use datafusion_common::test_util::batches_to_sort_string; +use datafusion_common::{Result, ScalarValue}; use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_datasource::file_scan_config::FileScanConfigBuilder; @@ -22,7 +22,9 @@ use datafusion_datasource::source::DataSourceExec; use insta::assert_snapshot; use object_store::ObjectMeta; use parquet::arrow::ArrowWriter; +use parquet::basic::ColumnOrder; use parquet::file::properties::WriterProperties; +use parquet::file::reader::{FileReader, SerializedFileReader}; use tempfile::NamedTempFile; /// Test for reading data from multiple parquet files with different schemas and coercing them into a single schema. @@ -132,6 +134,69 @@ async fn multi_parquet_coercion_projection() { "); } +#[tokio::test] +async fn parquet_max_with_mixed_float_schemas() { + // Arrange + let cast_values: ArrayRef = + Arc::new(Float64Array::from(vec![1.00000001, 2.00000003])); + let native_values: ArrayRef = Arc::new(Float32Array::from(vec![0.5])); + let (_, files) = store_parquet(vec![ + RecordBatch::try_from_iter(vec![("x", cast_values)]).unwrap(), + RecordBatch::try_from_iter(vec![("x", native_values)]).unwrap(), + ]) + .unwrap(); + for file in &files { + let reader = SerializedFileReader::new(file.reopen().unwrap()).unwrap(); + let metadata = reader.metadata(); + assert_eq!( + metadata.file_metadata().column_order(0), + ColumnOrder::IEEE_754_TOTAL_ORDER + ); + let stats = metadata.row_group(0).column(0).statistics().unwrap(); + assert_eq!(stats.nan_count_opt(), Some(0)); + assert!(!stats.is_min_max_deprecated()); + assert!(stats.min_is_exact()); + assert!(stats.max_is_exact()); + } + + let output_schema = Schema::new(vec![Field::new("x", DataType::Float32, true)]); + let ctx = SessionContext::new_with_config( + SessionConfig::new().with_collect_statistics(true), + ); + let paths: Vec<_> = files + .iter() + .map(|file| file.path().to_str().unwrap()) + .collect(); + let df = ctx + .read_parquet( + paths, + ParquetReadOptions::default() + .file_extension("") + .schema(&output_schema), + ) + .await + .unwrap(); + ctx.register_table("t", df.into_view()).unwrap(); + + // Act + let batches = ctx + .sql("SELECT MAX(x) FROM t") + .await + .unwrap() + .collect() + .await + .unwrap(); + + // Assert + // The Float64 file must still contribute after its values are cast. + assert_eq!(batches.len(), 1); + assert_eq!(batches[0].num_rows(), 1); + assert_eq!( + ScalarValue::try_from_array(batches[0].column(0), 0).unwrap(), + ScalarValue::Float32(Some(2.0)) + ); +} + /// Writes `batches` to a temporary parquet file pub fn store_parquet( batches: Vec, diff --git a/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs b/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs index 91b0e252a78c4..2e98993cca4c8 100644 --- a/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs +++ b/datafusion/datasource-parquet/benches/parquet_metadata_statistics.rs @@ -15,7 +15,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::basic::{ColumnOrder, SortOrder}; use parquet::data_type::ByteArray; use parquet::file::metadata::{ ColumnChunkMetaData, FileMetaData, ParquetMetaData, RowGroupMetaData, @@ -159,11 +159,14 @@ fn make_synthetic_metadata( }) .collect::>(); - // Model a modern writer so byte-array bounds have a trustworthy order. + // Model a modern writer so byte-array and floating bounds have a known order. let column_orders = schema_descr .columns() .iter() - .map(|column| ColumnOrder::TYPE_DEFINED_ORDER(column.sort_order())) + .map(|column| match column.sort_order() { + SortOrder::TOTAL_ORDER => ColumnOrder::IEEE_754_TOTAL_ORDER, + order => ColumnOrder::TYPE_DEFINED_ORDER(order), + }) .collect(); let file_metadata = FileMetaData::new( 1, @@ -236,6 +239,7 @@ fn statistics( let max = max_row.map(|row| value(column_idx, row_group, row) as f64 * 1.5); ParquetStatistics::Double( ValueStatistics::new(min, max, None, null_count, false) + .with_nan_count(Some(0)) .with_min_is_exact(min_exact) .with_max_is_exact(max_exact), ) diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index 1a3778b31ee72..be8490cfc43f4 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -31,7 +31,7 @@ use object_store::{ObjectMeta, ObjectStore}; use parquet::DecodeResult; use parquet::arrow::arrow_reader::statistics::StatisticsConverter; use parquet::arrow::parquet_to_arrow_schema; -use parquet::basic::{ColumnOrder, SortOrder, Type as PhysicalType}; +use parquet::basic::{ColumnOrder, LogicalType, SortOrder, Type as PhysicalType}; use parquet::file::metadata::{ PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, ParquetMetaDataReader, RowGroupMetaData, SortingColumn, @@ -46,11 +46,22 @@ use std::sync::Arc; /// would be too unreliable otherwise. const PARTIAL_NDV_THRESHOLD: f64 = 0.75; +pub(crate) fn is_parquet_float(column: &ColumnDescriptor) -> bool { + // Float16 uses byte-array storage, so identify it by its logical type. + matches!( + column.physical_type(), + PhysicalType::FLOAT | PhysicalType::DOUBLE + ) || column.logical_type_ref() == Some(&LogicalType::Float16) +} + fn requires_unsigned_byte_array_order(column: &ColumnDescriptor) -> bool { + // Byte arrays can also store decimals (signed numeric order) and Float16 + // (IEEE float order), so their bounds must bypass the unsigned byte check. matches!( column.physical_type(), PhysicalType::BYTE_ARRAY | PhysicalType::FIXED_LEN_BYTE_ARRAY ) && column.sort_order() != SortOrder::SIGNED + && !is_parquet_float(column) } /// Whether a column's min/max bounds lack a comparison order matching Arrow's. @@ -62,12 +73,32 @@ fn requires_unsigned_byte_array_order(column: &ColumnDescriptor) -> bool { /// Columns with undefined sort orders, such as `INT96`, never have usable /// min/max bounds regardless of their physical type. The `INT96` check is /// defensive because parquet-rs does not currently expose those bounds. +/// Float bounds require IEEE total order and a matching Arrow value type. pub(crate) fn has_untrusted_min_max_order( parquet_schema: &SchemaDescriptor, column_orders: Option<&[ColumnOrder]>, parquet_column_index: usize, + arrow_type: &DataType, ) -> bool { let column = parquet_schema.column(parquet_column_index); + let arrow_type = min_max_aggregate_data_type(arrow_type); + if is_parquet_float(&column) { + // TYPE_ORDER may report min=-0.0 for a column containing only +0.0. + // That is a bound, but not an exact minimum under IEEE total order. + // Casts can change values: Float64 1.00000001 becomes Float32 1.0. + // Reusing stored bounds after a cast needs validation and conversion. + // For now, require IEEE ordering and matching stored/output float types. + return !matches!( + (column.physical_type(), arrow_type), + (PhysicalType::FLOAT, DataType::Float32) + | (PhysicalType::DOUBLE, DataType::Float64) + | (PhysicalType::FIXED_LEN_BYTE_ARRAY, DataType::Float16) + ) || column_orders.and_then(|orders| orders.get(parquet_column_index)) + != Some(&ColumnOrder::IEEE_754_TOTAL_ORDER); + } + if arrow_type.is_floating() { + return true; + } // As of arrow 60, INT96 columns report `SortOrder::INT96_TIMESTAMP` // rather than `UNDEFINED`; keep treating their min/max as untrusted. // until @@ -85,15 +116,23 @@ pub(crate) fn has_untrusted_min_max_order( != 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>( +/// Whether any row group's bounds use a legacy byte-array order or omit +/// possible NaNs. A missing NaN count is not proof of zero NaNs. +pub(crate) fn has_untrusted_min_max_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); + if is_parquet_float(&column) { + return row_groups.into_iter().any(|group| { + group.num_rows() != 0 + && group.column(index).statistics().is_none_or(|stats| { + stats.nan_count_opt() != Some(0) || stats.is_min_max_deprecated() + }) + }); + } requires_unsigned_byte_array_order(&column) && (column.sort_order() != SortOrder::UNSIGNED || row_groups.into_iter().any(|group| { @@ -631,8 +670,9 @@ impl<'a> DFParquetMetadata<'a> { file_metadata.schema_descr(), file_metadata.column_orders().map(Vec::as_slice), index, + field.data_type(), ) - }) || has_untrusted_byte_array_stats( + }) || has_untrusted_min_max_stats( file_metadata.schema_descr(), parquet_index, row_groups_metadata, diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs index 0e2962e43e451..555894bda0e69 100644 --- a/datafusion/datasource-parquet/src/page_filter.rs +++ b/datafusion/datasource-parquet/src/page_filter.rs @@ -8,11 +8,12 @@ use std::collections::HashSet; use std::sync::Arc; use super::metrics::ParquetFileMetrics; -use crate::metadata::has_untrusted_min_max_order; +use crate::metadata::{has_untrusted_min_max_order, is_parquet_float}; use crate::pruning::build_inverted_predicate; use crate::{ParquetAccessPlan, RowGroupAccess}; -use arrow::array::BooleanArray; +use arrow::array::{BooleanArray, UInt64Array}; +use arrow::compute::nullif; use arrow::{ array::ArrayRef, datatypes::{Schema, SchemaRef}, @@ -706,6 +707,7 @@ struct PagesPruningStatistics<'a> { page_index: &'a dyn PageIndexProvider, page_offsets: &'a Vec, trusted_min_max: bool, + untrusted_nan_counts: Option, } impl<'a> PagesPruningStatistics<'a> { @@ -748,7 +750,30 @@ impl<'a> PagesPruningStatistics<'a> { file_metadata.schema_descr(), file_metadata.column_orders().map(Vec::as_slice), parquet_column_index, + converter.arrow_field().data_type(), ); + let untrusted_nan_counts = if trusted_min_max + && is_parquet_float( + &file_metadata.schema_descr().column(parquet_column_index), + ) { + let group_count = row_group_metadatas[row_group_index] + .column(parquet_column_index) + .statistics() + .and_then(|stats| stats.nan_count_opt()); + let counts = converter + .data_page_nan_counts(page_index, [&row_group_index]) + .ok() + .filter(|counts| counts.len() == page_offsets.len()) + .unwrap_or_else(|| UInt64Array::new_null(page_offsets.len())); + // A known-zero group count also covers pages with no count. + // Nullable iteration preserves the distinction between unknown and zero. + let mask = BooleanArray::from_iter( + counts.iter().map(|count| count.or(group_count) != Some(0)), + ); + (mask.true_count() != 0).then_some(mask) + } else { + None + }; Some(Self { row_group_index, @@ -757,9 +782,17 @@ impl<'a> PagesPruningStatistics<'a> { page_index, page_offsets, trusted_min_max, + untrusted_nan_counts, }) } + fn mask_untrusted_nan_counts(&self, values: ArrayRef) -> Option { + match &self.untrusted_nan_counts { + Some(mask) => nullif(values.as_ref(), mask).ok(), + None => Some(values), + } + } + /// return the row counts in each data page, if possible. fn page_row_counts(&self) -> Option> { let row_group_metadata = self @@ -790,7 +823,7 @@ impl PruningStatistics for PagesPruningStatistics<'_> { .converter .data_page_mins(self.page_index, [&self.row_group_index]) { - Ok(min_values) => Some(min_values), + Ok(min_values) => self.mask_untrusted_nan_counts(min_values), Err(e) => { debug!("Error evaluating data page min values {e}"); None @@ -806,7 +839,7 @@ impl PruningStatistics for PagesPruningStatistics<'_> { .converter .data_page_maxes(self.page_index, [&self.row_group_index]) { - Ok(min_values) => Some(min_values), + Ok(max_values) => self.mask_untrusted_nan_counts(max_values), Err(e) => { debug!("Error evaluating data page max values {e}"); None diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index 23291099ecac7..ec633f5c7fa37 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -7,7 +7,7 @@ use std::sync::Arc; use super::{ParquetAccessPlan, ParquetFileMetrics, RowGroupAccess}; use crate::bloom_filter::BloomFilterStatistics; -use crate::metadata::{has_untrusted_byte_array_stats, has_untrusted_min_max_order}; +use crate::metadata::{has_untrusted_min_max_order, has_untrusted_min_max_stats}; use crate::pruning::build_inverted_predicate; use arrow::array::{ArrayRef, BooleanArray, UInt64Array}; use arrow::compute::nullif; @@ -537,29 +537,33 @@ impl<'a> RowGroupPruningStatistics<'a> { let converter = self.statistics_converter(column).ok()?; let parquet_index = converter.parquet_column_index(); if parquet_index.is_some_and(|index| { - has_untrusted_min_max_order(self.parquet_schema, self.column_orders, index) + has_untrusted_min_max_order( + self.parquet_schema, + self.column_orders, + index, + converter.arrow_field().data_type(), + ) }) { return None; } Some(converter) } - fn mask_untrusted_byte_array_stats( + fn mask_untrusted_min_max_stats( &self, parquet_index: Option, values: ArrayRef, ) -> Option { - if !has_untrusted_byte_array_stats( + if !has_untrusted_min_max_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. + // Keep usable row-group bounds when other groups have unsafe statistics. let mask = BooleanArray::from_iter(self.metadata_iter().map(|group| { - has_untrusted_byte_array_stats(self.parquet_schema, parquet_index, [group]) + has_untrusted_min_max_stats(self.parquet_schema, parquet_index, [group]) })); nullif(values.as_ref(), &mask).ok() } @@ -569,13 +573,13 @@ impl PruningStatistics for RowGroupPruningStatistics<'_> { fn min_values(&self, column: &Column) -> Option { 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) + self.mask_untrusted_min_max_stats(converter.parquet_column_index(), values) } fn max_values(&self, column: &Column) -> Option { 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) + self.mask_untrusted_min_max_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 index f3da8bae9484d..0eb0029f142d3 100644 --- a/datafusion/datasource-parquet/src/statistics_order_tests.rs +++ b/datafusion/datasource-parquet/src/statistics_order_tests.rs @@ -2,13 +2,15 @@ // 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 -//! Regression tests for interpreting Parquet byte-array statistics orders. +//! Regression tests for interpreting Parquet statistics orders and NaN counts. use std::io::Write; use std::sync::Arc; -use arrow::array::{BooleanArray, record_batch}; -use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; +use arrow::array::{ArrayRef, BooleanArray, Float64Array, Int32Array, record_batch}; +use arrow::compute::cast; +use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit}; +use arrow::record_batch::RecordBatch; use bytes::Bytes; use datafusion_common::pruning::{PrunableStatistics, PruningStatistics}; use datafusion_common::stats::Precision; @@ -54,6 +56,42 @@ struct TestFile { } impl TestFile { + fn floats(values: &[Option]) -> Self { + let row_count = values.len(); + let values = Float64Array::from(values.to_vec()); + let batch = RecordBatch::try_from_iter([ + ("f16", cast(&values, &DataType::Float16).unwrap()), + ("f32", cast(&values, &DataType::Float32).unwrap()), + ("f64", Arc::new(values) as ArrayRef), + ( + "n", + Arc::new(Int32Array::from_iter_values(0..row_count as i32)) as ArrayRef, + ), + ]) + .unwrap(); + let schema = batch.schema(); + let properties = WriterProperties::builder() + .set_max_row_group_row_count(Some(6)) + .set_data_page_row_count_limit(3) + .set_write_batch_size(3) + .set_dictionary_enabled(false) + .set_statistics_enabled(EnabledStatistics::Page) + .build(); + let mut bytes = Vec::new(); + let mut writer = + ArrowWriter::try_new(&mut bytes, Arc::clone(&schema), Some(properties)) + .unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let bytes = Bytes::from(bytes); + let metadata = Arc::new(read_metadata(&bytes)); + Self { + bytes, + schema, + metadata, + } + } + fn new(order: StatisticsOrder) -> Self { let batch = record_batch!( ( @@ -312,6 +350,296 @@ fn metrics() -> ParquetFileMetrics { ) } +#[test] +fn float_nan_counts_preserve_matching_rows_at_every_pruning_level() { + // Arrange + let file = TestFile::floats(&[ + Some(-f64::NAN), + Some(5.0), + Some(f64::NAN), + Some(5.0), + Some(5.0), + Some(5.0), + Some(8.0), + Some(9.0), + None, + Some(10.0), + Some(11.0), + None, + ]); + let metadata = file.metadata.file_metadata(); + let row_group_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: &file.schema, + }; + + // Act + let statistics = file.statistics(); + + // Assert + for column_index in 0..3 { + let field = file.schema.field(column_index); + let raw = file + .metadata + .row_group(0) + .column(column_index) + .statistics() + .unwrap(); + assert_eq!(raw.nan_count_opt(), Some(2)); + assert_eq!(raw.null_count_opt(), Some(0)); + assert_eq!(raw.min_bytes_opt(), raw.max_bytes_opt()); + assert_eq!( + metadata.column_order(column_index), + ColumnOrder::IEEE_754_TOTAL_ORDER + ); + for output_type in [ + field.data_type().clone(), + DataType::Dictionary( + Box::new(DataType::Int8), + Box::new(field.data_type().clone()), + ), + DataType::Int64, + ] { + for order in [ + None, + Some(ColumnOrder::UNKNOWN), + Some(ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED)), + Some(ColumnOrder::IEEE_754_TOTAL_ORDER), + ] { + let orders = order.map(|order| vec![order; 3]); + assert_eq!( + has_untrusted_min_max_order( + metadata.schema_descr(), + orders.as_deref(), + column_index, + &output_type, + ), + order != Some(ColumnOrder::IEEE_754_TOTAL_ORDER) + || output_type == DataType::Int64, + "{output_type:?}, {order:?}", + ); + } + } + let index = file + .metadata + .page_index() + .unwrap() + .column_index(0, column_index) + .unwrap(); + assert_eq!(index.nan_counts(), Some(&vec![2, 0])); + let stats = &statistics.column_statistics[column_index]; + assert_eq!(stats.min_value, Precision::Absent); + assert_eq!(stats.max_value, Precision::Absent); + assert_eq!(stats.null_count, Precision::Exact(2)); + let column = Column::from_name(field.name()); + for bounds in [ + row_group_stats.min_values(&column), + row_group_stats.max_values(&column), + ] { + let bounds = bounds.unwrap(); + assert!(bounds.is_null(0)); + assert!(!bounds.is_null(1)); + } + + let value = |v| { + lit(ScalarValue::Float64(Some(v)) + .cast_to(field.data_type()) + .unwrap()) + }; + for (expr, expected, selected) in [ + (col(field.name()).lt(value(0.0)), 1, 3), + (col(field.name()).gt(value(100.0)), 1, 3), + (col(field.name()).not_eq(value(5.0)), 6, 9), + (col(field.name()).eq(value(5.0)), 4, 6), + (col(field.name()).is_null(), 2, 6), + ] { + let (physical, predicate) = file.predicate(&expr); + let all = ParquetAccessPlan::new_all(2); + assert_eq!( + file.matching_rows(&physical, all.clone()), + expected, + "{expr}" + ); + assert!(file.file_matches(&predicate), "{expr}"); + let row_groups = file.row_group_plan(&predicate); + assert!(!row_groups.is_fully_matched(0), "{expr}"); + assert_eq!( + file.matching_rows(&physical, row_groups), + expected, + "{expr}" + ); + // Start from all groups so row-group pruning cannot hide a page error. + let pages = file.page_plan(&physical, all); + assert_eq!( + file.matching_rows(&physical, pages.clone()), + expected, + "{expr}" + ); + assert_eq!( + pages + .into_inner() + .iter() + .zip(file.metadata.row_groups()) + .map(|(access, group)| match access { + crate::RowGroupAccess::Skip => 0, + crate::RowGroupAccess::Scan => group.num_rows() as usize, + crate::RowGroupAccess::Selection(selection) => { + selection.row_count() + } + }) + .sum::(), + selected, + "{expr}", + ); + } + } + assert_eq!( + statistics.column_statistics[3].min_value, + Precision::Exact(ScalarValue::Int32(Some(0))), + ); + assert_eq!( + statistics.column_statistics[3].max_value, + Precision::Exact(ScalarValue::Int32(Some(11))), + ); + let (physical, predicate) = file.predicate(&col("n").gt(lit(20))); + assert!(!file.file_matches(&predicate)); + assert!( + file.row_group_plan(&predicate) + .row_group_indexes() + .is_empty() + ); + assert!( + file.page_plan(&physical, ParquetAccessPlan::new_all(2)) + .row_group_indexes() + .is_empty() + ); +} + +#[test] +fn float_nan_counts_require_file_group_and_page_proof() { + for omit_counts in [false, true] { + // Arrange + let mut values = [Some(5.0); 18]; + if omit_counts { + values[0] = Some(f64::NAN); + } + let mut file = TestFile::floats(&values); + if omit_counts { + let mut groups = file.metadata.row_groups().to_vec(); + for row_group in [0, 2] { + let mut columns = groups[row_group].columns().to_vec(); + for column in &mut columns[..3] { + let statistics = match column.statistics().unwrap().clone() { + ParquetStatistics::Float(s) => s.with_nan_count(None).into(), + ParquetStatistics::Double(s) => s.with_nan_count(None).into(), + ParquetStatistics::FixedLenByteArray(s) => { + s.with_nan_count(None).into() + } + _ => unreachable!(), + }; + *column = column + .clone() + .into_builder() + .set_statistics(statistics) + .build() + .unwrap(); + } + groups[row_group] = groups[row_group] + .clone() + .into_builder() + .set_column_metadata(columns) + .build() + .unwrap(); + } + let indexes = file.metadata.page_index().unwrap(); + let mut pages = indexes + .as_any() + .downcast_ref::() + .unwrap() + .clone() + .into_builder(); + for row_group in [0, 1] { + for column in 0..3 { + let original = indexes.column_index(row_group, column).unwrap(); + let chunk = file.metadata.row_group(row_group).column(column); + let stats = chunk.statistics().unwrap(); + let mut index = ColumnIndexBuilder::new(chunk.column_type()); + index.set_boundary_order(original.get_boundary_order().unwrap()); + // NaNs are excluded, so every page has finite bounds [5, 5]. + for page in 0..original.num_pages() as usize { + index.append( + original.is_null_page(page), + stats.min_bytes_opt().unwrap().to_vec(), + stats.max_bytes_opt().unwrap().to_vec(), + original.null_count(page).unwrap(), + None, + ); + } + pages.put_column_index(index.build().unwrap(), row_group, column); + } + } + let metadata = file + .metadata + .as_ref() + .clone() + .into_builder() + .set_row_groups(groups) + .set_page_index(Some(Arc::new(pages.build()))) + .build(); + // Serialize the missing counts while keeping the real data pages. + let mut bytes = Vec::new(); + let mut tracked = TrackedWrite::new(&mut bytes); + tracked + .write_all(&file.bytes[..footer_start(&file.bytes)]) + .unwrap(); + ParquetMetaDataWriter::new_with_tracked(tracked, &metadata) + .finish() + .unwrap(); + file.bytes = Bytes::from(bytes); + file.metadata = Arc::new(read_metadata(&file.bytes)); + } + + // Act + let statistics = file.statistics(); + + // Assert + for column_index in 0..3 { + let field = file.schema.field(column_index); + let five = ScalarValue::Float64(Some(5.0)) + .cast_to(field.data_type()) + .unwrap(); + let stats = &statistics.column_statistics[column_index]; + let expected = if omit_counts { + Precision::Absent + } else { + Precision::Exact(five.clone()) + }; + assert_eq!(stats.min_value, expected); + assert_eq!(stats.max_value, expected); + assert_eq!(stats.null_count, Precision::Exact(0)); + let (physical, predicate) = + file.predicate(&col(field.name()).not_eq(lit(five))); + assert_eq!(file.file_matches(&predicate), omit_counts); + assert_eq!( + file.row_group_plan(&predicate).row_group_indexes(), + if omit_counts { vec![0, 2] } else { vec![] }, + ); + // RG1 proves its pages safe; RG2's pages prove themselves safe. + let pages = file.page_plan(&physical, ParquetAccessPlan::new_all(3)); + assert_eq!( + pages.row_group_indexes(), + if omit_counts { vec![0] } else { vec![] } + ); + assert_eq!( + file.matching_rows(&physical, pages), + usize::from(omit_counts), + ); + } + } +} + #[test] fn byte_array_order_preserves_matching_rows_at_every_pruning_level() { for order in [ @@ -757,7 +1085,8 @@ fn undefined_int96_order_is_never_trusted() { has_untrusted_min_max_order( &schema, order.as_ref().map(std::slice::from_ref), - 0 + 0, + &DataType::Timestamp(TimeUnit::Nanosecond, None), ), "order={order:?}", ); diff --git a/datafusion/sqllogictest/test_files/parquet.slt b/datafusion/sqllogictest/test_files/parquet.slt index b02672e50ac9c..d024814eacfd5 100644 --- a/datafusion/sqllogictest/test_files/parquet.slt +++ b/datafusion/sqllogictest/test_files/parquet.slt @@ -367,6 +367,46 @@ NULL statement ok DROP TABLE single_nan; +# NaN values are excluded from Parquet min/max statistics. +# Arrange: write columns with finite min=max and no nulls. +query I +COPY ( + SELECT x::REAL AS f32, x::DOUBLE AS f64 + FROM (VALUES ('NaN'), ('5')) AS t(x) +) +TO 'test_files/scratch/parquet/nan_statistics.parquet' +STORED AS PARQUET; +---- +2 + +# Act: read the file with its stored statistics. +statement ok +CREATE EXTERNAL TABLE nan_statistics +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet/nan_statistics.parquet'; + +# Assert: equal finite bounds must not replace NaN values. +query RR rowsort +SELECT f32, f64 FROM nan_statistics; +---- +5 5 +NaN NaN + +# Equality must filter out NaN. +query RR +SELECT f32, f64 FROM nan_statistics WHERE f32 = 5 AND f64 = 5; +---- +5 5 + +# NaN sorts above finite values and must not be pruned. +query RR +SELECT f32, f64 FROM nan_statistics WHERE f32 > 5 AND f64 > 5; +---- +NaN NaN + +statement ok +DROP TABLE nan_statistics; + statement ok CREATE EXTERNAL TABLE list_columns STORED AS PARQUET