fix(parquet): validate float bounds using NaN counts

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::<Float64Array>().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();

datafusion/core/tests/parquet/schema_coercion.rs

@@ -5,24 +5,26 @@
 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;
 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<RecordBatch>,

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::<Vec<_>>();
 
-    // 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),
             )

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 <https://github.com/apache/datafusion/issues/25484>
@@ -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<usize>,
     row_groups: impl IntoIterator<Item = &'a RowGroupMetaData>,
 ) -> 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,