Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions datafusion/core/tests/parquet/page_pruning.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Expand Down
9 changes: 5 additions & 4 deletions datafusion/core/tests/parquet/row_group_pruning.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
69 changes: 68 additions & 1 deletion datafusion/datasource-parquet/src/metadata.rs
Original file line number Diff line number Diff line change
Expand Up @@ -584,13 +584,37 @@ 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<Option<MaxAccumulator>>, Vec<Option<MinAccumulator>>) {
let max_values: Vec<Option<MaxAccumulator>> = schema
.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();
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
9 changes: 5 additions & 4 deletions datafusion/datasource-parquet/src/opener/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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();
Expand All @@ -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();
Expand Down
4 changes: 4 additions & 0 deletions datafusion/datasource-parquet/src/page_filter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -577,6 +578,9 @@ impl PruningStatistics for PagesPruningStatistics<'_> {
}

fn max_values(&self, _column: &datafusion_common::Column) -> Option<ArrayRef> {
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,
Expand Down
55 changes: 52 additions & 3 deletions datafusion/datasource-parquet/src/row_group_filter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -495,9 +496,11 @@ impl PruningStatistics for RowGroupPruningStatistics<'_> {
}

fn max_values(&self, column: &Column) -> Option<ArrayRef> {
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 {
Expand Down Expand Up @@ -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};
Expand Down
Original file line number Diff line number Diff line change
@@ -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;
12 changes: 6 additions & 6 deletions datafusion/sqllogictest/test_files/parquet_statistics.slt
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
Loading