diff --git a/datafusion/core/tests/parquet/page_pruning.rs b/datafusion/core/tests/parquet/page_pruning.rs index 372a7a601d492..a8fdad0eb38d0 100644 --- a/datafusion/core/tests/parquet/page_pruning.rs +++ b/datafusion/core/tests/parquet/page_pruning.rs @@ -765,12 +765,13 @@ async fn prune_f64_lt() { #[tokio::test] async fn prune_f64_scalar_fun_and_gt() { // result of sql "SELECT * FROM t where abs(f - 1) <= 0.000001 and f >= 0.1" - // only use "f >= 0" to prune + // only "f >= 0.1" could prune, and only by the float maximum, which Parquet statistics + // compute without NaN (NaN sorts above every number), so no page is pruned test_prune( Scenario::Float64, "SELECT * FROM t where abs(f - 1) <= 0.000001 and f >= 0.1", Some(0), - Some(10), + Some(0), 1, 5, ) diff --git a/datafusion/core/tests/parquet/row_group_pruning.rs b/datafusion/core/tests/parquet/row_group_pruning.rs index 0721715921909..d1ff34ddec16a 100644 --- a/datafusion/core/tests/parquet/row_group_pruning.rs +++ b/datafusion/core/tests/parquet/row_group_pruning.rs @@ -836,15 +836,16 @@ async fn prune_f64_lt() { #[tokio::test] async fn prune_f64_scalar_fun_and_gt() { // result of sql "SELECT * FROM t where abs(f - 1) <= 0.000001 and f >= 0.1" - // only use "f >= 0" to prune + // only "f >= 0.1" could prune, and only by the float maximum, which Parquet statistics + // compute without NaN (NaN sorts above every number), so no row group is pruned RowGroupPruningTest::new() .with_scenario(Scenario::Float64) .with_query("SELECT * FROM t where abs(f - 1) <= 0.000001 and f >= 0.1") .with_expected_errors(Some(0)) - .with_matched_by_stats(Some(2)) - .with_pruned_by_stats(Some(2)) + .with_matched_by_stats(Some(4)) + .with_pruned_by_stats(Some(0)) .with_pruned_files(Some(0)) - .with_matched_by_bloom_filter(Some(2)) + .with_matched_by_bloom_filter(Some(4)) .with_pruned_by_bloom_filter(Some(0)) .with_expected_rows(1) .test_row_group_prune() diff --git a/datafusion/datasource-parquet/src/metadata.rs b/datafusion/datasource-parquet/src/metadata.rs index 38db1e22ee6c0..7665025c9c218 100644 --- a/datafusion/datasource-parquet/src/metadata.rs +++ b/datafusion/datasource-parquet/src/metadata.rs @@ -584,6 +584,27 @@ fn min_max_aggregate_data_type(input_type: &DataType) -> &DataType { } } +/// Whether a column's Parquet maximum may sit below some of its values. +/// +/// Parquet writers leave NaN out of floating-point statistics (the Parquet spec asks them to), +/// while a NaN with its sign bit clear (the `'NaN'` literal) sorts above every other value. A +/// float column's maximum therefore only bounds the non-NaN values: it must not prune `col > x` +/// or stand for the column's `MAX`. The minimum stays a valid lower bound for such a NaN. +pub(crate) fn max_statistic_may_omit_nan(data_type: &DataType) -> bool { + min_max_aggregate_data_type(data_type).is_floating() +} + +/// Whether a float minimum is a zero, which Parquet writers record as `-0.0` whatever the sign +/// of the stored zeros: it bounds the column but is not its exact minimum. +fn min_statistic_is_signed_zero(min: &ScalarValue) -> bool { + match min { + ScalarValue::Float16(Some(value)) => value.to_f32() == 0.0, + ScalarValue::Float32(Some(value)) => *value == 0.0, + ScalarValue::Float64(Some(value)) => *value == 0.0, + _ => false, + } +} + fn create_max_min_accs( schema: &Schema, ) -> (Vec>, Vec>) { @@ -591,6 +612,9 @@ fn create_max_min_accs( .fields() .iter() .map(|field| { + if max_statistic_may_omit_nan(field.data_type()) { + return None; + } MaxAccumulator::try_new(min_max_aggregate_data_type(field.data_type())).ok() }) .collect(); @@ -637,7 +661,13 @@ impl StatisticsAccumulators<'_> { self.is_min_value_exact.get(i).unwrap(), ) { (Some(min_value), Some(true)) => { - min_value.evaluate().ok().map(Precision::Exact) + min_value.evaluate().ok().map(|min| { + if min_statistic_is_signed_zero(&min) { + Precision::Inexact(min) + } else { + Precision::Exact(min) + } + }) } (Some(min_value), Some(false)) | (Some(min_value), None) => { min_value.evaluate().ok().map(Precision::Inexact) @@ -1138,6 +1168,43 @@ mod tests { Ok(()) } + #[test] + fn float_maximum_is_absent_from_file_statistics() -> Result<()> { + use arrow::array::{Float64Array, RecordBatch}; + use parquet::arrow::ArrowWriter; + let schema = Arc::new(Schema::new(vec![ + Field::new("f", DataType::Float64, true), + Field::new("i", DataType::Int32, true), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Float64Array::from(vec![1.0, f64::NAN])), + Arc::new(Int32Array::from(vec![1, 2])), + ], + )?; + let mut buffer = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut buffer, Arc::clone(&schema), None)?; + writer.write(&batch)?; + writer.close()?; + let metadata = + ParquetMetaDataReader::new().parse_and_finish(&bytes::Bytes::from(buffer))?; + let statistics = + DFParquetMetadata::statistics_from_parquet_metadata(&metadata, &schema)?; + // The writer left NaN out of the float statistics, so max 1.0 would hide the NaN row. + let float = &statistics.column_statistics[0]; + assert_eq!(float.max_value, Precision::Absent); + assert_eq!( + float.min_value.get_value(), + Some(&ScalarValue::Float64(Some(1.0))) + ); + assert_eq!( + statistics.column_statistics[1].max_value.get_value(), + Some(&ScalarValue::Int32(Some(2))) + ); + Ok(()) + } + #[test] fn test_has_any_exact_match() { // Case 1: Mixed exact and inexact matches diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index a9f4662f23c94..81a9deace6ec8 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -2489,8 +2489,9 @@ mod test { assert_eq!(num_batches, 1); assert_eq!(num_rows, 3); - // A filter on `b = 5.0` should exclude all rows - let expr = col("b").eq(lit(ScalarValue::Float32(Some(5.0)))); + // A filter on `b = 0.5` should exclude all rows (by the minimum: a float maximum is not a + // bound, since Parquet statistics leave NaN out) + let expr = col("b").eq(lit(ScalarValue::Float32(Some(0.5)))); let predicate = logical2physical(&expr, &schema); let opener = make_opener(predicate); let stream = open_file(&opener, file).await.unwrap(); @@ -2627,7 +2628,7 @@ mod test { assert_eq!(num_rows, 0); // Should prune based on file statistics but not partition value - let expr = col("part").eq(lit(1)).and(col("b").eq(lit(7.0))); + let expr = col("part").eq(lit(1)).and(col("b").eq(lit(0.5))); let predicate = logical2physical(&expr, &table_schema); let opener = make_opener(predicate); let stream = open_file(&opener, file.clone()).await.unwrap(); @@ -2636,7 +2637,7 @@ mod test { assert_eq!(num_rows, 0); // Should prune based on both partition value and file statistics - let expr = col("part").eq(lit(2)).and(col("b").eq(lit(7.0))); + let expr = col("part").eq(lit(2)).and(col("b").eq(lit(0.5))); let predicate = logical2physical(&expr, &table_schema); let opener = make_opener(predicate); let stream = open_file(&opener, file).await.unwrap(); diff --git a/datafusion/datasource-parquet/src/page_filter.rs b/datafusion/datasource-parquet/src/page_filter.rs index 6bc1aca667981..63696118aef5e 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::max_statistic_may_omit_nan; use arrow::array::BooleanArray; use arrow::{ @@ -577,6 +578,9 @@ impl PruningStatistics for PagesPruningStatistics<'_> { } fn max_values(&self, _column: &datafusion_common::Column) -> Option { + if max_statistic_may_omit_nan(self.converter.arrow_field().data_type()) { + return None; + } match self.converter.data_page_maxes( self.column_index, self.offset_index, diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index ddf71bb7e6d95..debe7473be9c8 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -20,6 +20,7 @@ use std::sync::Arc; use super::{ParquetAccessPlan, ParquetFileMetrics, RowGroupAccess}; use crate::bloom_filter::BloomFilterStatistics; +use crate::metadata::max_statistic_may_omit_nan; use arrow::array::{ArrayRef, BooleanArray, UInt64Array}; use arrow::datatypes::Schema; use datafusion_common::pruning::PruningStatistics; @@ -495,9 +496,11 @@ impl PruningStatistics for RowGroupPruningStatistics<'_> { } 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.statistics_converter(column).ok()?; + if max_statistic_may_omit_nan(converter.arrow_field().data_type()) { + return None; + } + converter.row_group_maxes(self.metadata_iter()).ok() } fn num_containers(&self) -> usize { @@ -660,6 +663,52 @@ mod tests { assert_pruned(row_groups, ExpectedPruning::Some(vec![1])) } + #[test] + fn row_group_float_maximum_does_not_prune() { + use datafusion_expr::{col, lit}; + // Parquet statistics leave NaN out, and NaN sorts above every number: a row group + // holding 1.0 and NaN has max 1.0 but a row matching `c1 > 15`. The minimum still prunes. + let schema = + Arc::new(Schema::new(vec![Field::new("c1", DataType::Float64, true)])); + let field = PrimitiveTypeField::new("c1", PhysicalType::DOUBLE); + let schema_descr = get_test_schema_descr(vec![field]); + let row_group = |min: f64, max: f64| { + get_row_group_meta_data( + &schema_descr, + vec![ParquetStatistics::double( + Some(min), + Some(max), + None, + Some(0), + false, + )], + ) + }; + let row_groups_for = |expr: datafusion_expr::Expr| { + let expr = logical2physical(&expr, &schema); + let pruning_predicate = + build_test_pruning_predicate(expr, Arc::clone(&schema)); + let mut row_groups = + RowGroupAccessPlanFilter::new(ParquetAccessPlan::new_all(2)); + row_groups.prune_by_statistics( + &schema, + &schema_descr, + &[row_group(1.0, 10.0), row_group(11.0, 20.0)], + &pruning_predicate, + &parquet_file_metrics(), + ); + row_groups + }; + assert_pruned( + row_groups_for(col("c1").gt(lit(15.0))), + ExpectedPruning::None, + ); + assert_pruned( + row_groups_for(col("c1").lt(lit(5.0))), + ExpectedPruning::Some(vec![0]), + ); + } + #[test] fn row_group_fully_matched_requires_known_non_null_predicate_columns() { use datafusion_expr::{col, lit}; diff --git a/datafusion/sqllogictest/test_files/parquet_float_nan_statistics.slt b/datafusion/sqllogictest/test_files/parquet_float_nan_statistics.slt new file mode 100644 index 0000000000000..f04deb7bacbd9 --- /dev/null +++ b/datafusion/sqllogictest/test_files/parquet_float_nan_statistics.slt @@ -0,0 +1,87 @@ +# 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. + +# NaN rows of a float column must survive Parquet statistics. +# +# Parquet writers leave NaN out of float min/max statistics, while NaN sorts above every +# number. A file holding 1.0 and NaN therefore has min = max = 1.0; trusting that maximum +# prunes the file for `column1 > 2`, answers MAX with 1.0, and reads the column as the +# constant 1.0. + +query I +COPY (VALUES (1.0), ('NaN'::double), (1.0)) +TO 'test_files/scratch/parquet_float_nan_statistics/t/0.parquet' +STORED AS PARQUET; +---- +3 + +statement ok +CREATE EXTERNAL TABLE t +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_float_nan_statistics/t'; + +query I +SELECT count(*) FROM t WHERE column1 > 2; +---- +1 + +query R +SELECT max(column1) FROM t; +---- +NaN + +query R +SELECT column1 FROM t ORDER BY column1; +---- +1 +1 +NaN + +query R +SELECT column1 FROM t ORDER BY column1 DESC LIMIT 1; +---- +NaN + +# The minimum is still a bound: the 'NaN' literal is never below it. +query I +SELECT count(*) FROM t WHERE column1 < 1; +---- +0 + +statement ok +DROP TABLE t; + +# Parquet writers record a zero minimum as -0.0, so MIN must not be answered from it. +query I +COPY (VALUES (0.0), (1.0)) +TO 'test_files/scratch/parquet_float_nan_statistics/zero/0.parquet' +STORED AS PARQUET; +---- +2 + +statement ok +CREATE EXTERNAL TABLE zero +STORED AS PARQUET +LOCATION 'test_files/scratch/parquet_float_nan_statistics/zero'; + +query T +SELECT CAST(min(column1) AS VARCHAR) FROM zero; +---- +0.0 + +statement ok +DROP TABLE zero; diff --git a/datafusion/sqllogictest/test_files/parquet_statistics.slt b/datafusion/sqllogictest/test_files/parquet_statistics.slt index 9cf6b1e0381d1..3382e1001df9d 100644 --- a/datafusion/sqllogictest/test_files/parquet_statistics.slt +++ b/datafusion/sqllogictest/test_files/parquet_statistics.slt @@ -170,18 +170,18 @@ query TT EXPLAIN SELECT f32 FROM typed_table WHERE f32 = 2.5; ---- physical_plan -01)FilterExec: CAST(f32@0 AS Float64) = 2.5, statistics=[Rows=Inexact(1), Bytes=Inexact(1), [(Col[0]: Min=Exact(Float32(2.5)) Max=Exact(Float32(2.5)) Null=Inexact(0) Distinct=Exact(1) ScanBytes=Inexact(1))]] -02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, statistics=[Rows=Inexact(5), Bytes=Inexact(20), [(Col[0]: Min=Inexact(Float32(1.5)) Max=Inexact(Float32(5.5)) Null=Inexact(0) ScanBytes=Inexact(20))]] -03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_statistics/typed_table.parquet]]}, projection=[f32], file_type=parquet, predicate=CAST(f32@2 AS Float64) = 2.5, pruning_predicate=f32_null_count@2 != row_count@3 AND CAST(f32_min@0 AS Float64) <= 2.5 AND 2.5 <= CAST(f32_max@1 AS Float64), required_guarantees=[], statistics=[Rows=Inexact(5), Bytes=Inexact(20), [(Col[0]: Min=Inexact(Float32(1.5)) Max=Inexact(Float32(5.5)) Null=Inexact(0) ScanBytes=Inexact(20))]] +01)FilterExec: CAST(f32@0 AS Float64) = 2.5, statistics=[Rows=Inexact(5), Bytes=Inexact(20), [(Col[0]: Min=Exact(Float32(2.5)) Max=Exact(Float32(2.5)) Null=Inexact(0) Distinct=Exact(1) ScanBytes=Inexact(20))]] +02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, statistics=[Rows=Inexact(5), Bytes=Inexact(20), [(Col[0]: Min=Inexact(Float32(1.5)) Null=Inexact(0) ScanBytes=Inexact(20))]] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_statistics/typed_table.parquet]]}, projection=[f32], file_type=parquet, predicate=CAST(f32@2 AS Float64) = 2.5, pruning_predicate=f32_null_count@2 != row_count@3 AND CAST(f32_min@0 AS Float64) <= 2.5 AND 2.5 <= CAST(f32_max@1 AS Float64), required_guarantees=[], statistics=[Rows=Inexact(5), Bytes=Inexact(20), [(Col[0]: Min=Inexact(Float32(1.5)) Null=Inexact(0) ScanBytes=Inexact(20))]] # Reversed operand order: literal = column (Float64) query TT EXPLAIN SELECT f64 FROM typed_table WHERE 2.5 = f64; ---- physical_plan -01)FilterExec: f64@0 = 2.5, statistics=[Rows=Inexact(1), Bytes=Inexact(1), [(Col[0]: Min=Exact(Float64(2.5)) Max=Exact(Float64(2.5)) Null=Exact(0) Distinct=Exact(1) ScanBytes=Inexact(1))]] -02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, statistics=[Rows=Inexact(5), Bytes=Inexact(40), [(Col[0]: Min=Inexact(Float64(1.5)) Max=Inexact(Float64(5.5)) Null=Inexact(0) ScanBytes=Inexact(40))]] -03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_statistics/typed_table.parquet]]}, projection=[f64], file_type=parquet, predicate=f64@3 = 2.5, pruning_predicate=f64_null_count@2 != row_count@3 AND f64_min@0 <= 2.5 AND 2.5 <= f64_max@1, required_guarantees=[f64 in (2.5)], statistics=[Rows=Inexact(5), Bytes=Inexact(40), [(Col[0]: Min=Inexact(Float64(1.5)) Max=Inexact(Float64(5.5)) Null=Inexact(0) ScanBytes=Inexact(40))]] +01)FilterExec: f64@0 = 2.5, statistics=[Rows=Inexact(5), Bytes=Inexact(40), [(Col[0]: Min=Exact(Float64(2.5)) Max=Exact(Float64(2.5)) Null=Exact(0) Distinct=Exact(1) ScanBytes=Inexact(40))]] +02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, statistics=[Rows=Inexact(5), Bytes=Inexact(40), [(Col[0]: Min=Inexact(Float64(1.5)) Null=Inexact(0) ScanBytes=Inexact(40))]] +03)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_statistics/typed_table.parquet]]}, projection=[f64], file_type=parquet, predicate=f64@3 = 2.5, pruning_predicate=f64_null_count@2 != row_count@3 AND f64_min@0 <= 2.5 AND 2.5 <= f64_max@1, required_guarantees=[f64 in (2.5)], statistics=[Rows=Inexact(5), Bytes=Inexact(40), [(Col[0]: Min=Inexact(Float64(1.5)) Null=Inexact(0) ScanBytes=Inexact(40))]] statement ok DROP TABLE typed_table;