Skip to content

fix: prevent incorrect Parquet pruning with NaNs #24804

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
sunchao wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

fix: prevent incorrect Parquet pruning with NaNs #24804

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
11 changes: 6 additions & 5 deletions datafusion/core/tests/parquet/page_pruning.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -742,11 +742,12 @@ uint_tests!(
// page-2 0 0.0 4.0
// page-3 0 5.0 9.0
async fn prune_f64_lt() {
// Parquet floating bounds omit possible NaNs, so they cannot prune pages.
test_prune(
Scenario::Float64,
"SELECT * FROM t where f < 1",
Some(0),
Some(5),
Some(0),
11,
5,
)
Expand All @@ -755,7 +756,7 @@ async fn prune_f64_lt() {
Scenario::Float64,
"SELECT * FROM t where -f > -1",
Some(0),
Some(5),
Some(0),
11,
5,
)
Expand All @@ -764,13 +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
// The scalar function is unsupported for pruning, and the floating bounds
// for f >= 0.1 omit possible NaNs. Neither condition can prune pages.
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
24 changes: 13 additions & 11 deletions datafusion/core/tests/parquet/row_group_pruning.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -807,14 +807,16 @@ async fn prune_uint32_eq_large_in_list() {

#[tokio::test]
async fn prune_f64_lt() {
// Parquet floating bounds omit possible NaNs, so all row groups reach the
// Bloom filter stage, which cannot prune these range predicates.
RowGroupPruningTest::new()
.with_scenario(Scenario::Float64)
.with_query("SELECT * FROM t where f < 1")
.with_expected_errors(Some(0))
.with_matched_by_stats(Some(3))
.with_pruned_by_stats(Some(1))
.with_matched_by_stats(Some(4))
.with_pruned_by_stats(Some(0))
.with_pruned_files(Some(0))
.with_matched_by_bloom_filter(Some(3))
.with_matched_by_bloom_filter(Some(4))
.with_pruned_by_bloom_filter(Some(0))
.with_expected_rows(11)
.test_row_group_prune()
Expand All @@ -823,10 +825,10 @@ async fn prune_f64_lt() {
.with_scenario(Scenario::Float64)
.with_query("SELECT * FROM t where -f > -1")
.with_expected_errors(Some(0))
.with_matched_by_stats(Some(3))
.with_pruned_by_stats(Some(1))
.with_matched_by_stats(Some(4))
.with_pruned_by_stats(Some(0))
.with_pruned_files(Some(0))
.with_matched_by_bloom_filter(Some(3))
.with_matched_by_bloom_filter(Some(4))
.with_pruned_by_bloom_filter(Some(0))
.with_expected_rows(11)
.test_row_group_prune()
Expand All @@ -835,16 +837,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
// The scalar function is unsupported for pruning, and the floating bounds
// for f >= 0.1 omit possible NaNs. Neither condition can prune row groups.
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
15 changes: 12 additions & 3 deletions datafusion/datasource-parquet/src/metadata.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ 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::basic::{ColumnOrder, LogicalType, SortOrder, Type as PhysicalType};
use parquet::file::metadata::{
PageIndexPolicy, ParquetMetaData, ParquetMetaDataPushDecoder, ParquetMetaDataReader,
RowGroupMetaData, SortingColumn,
Expand All @@ -70,17 +70,26 @@ fn requires_unsigned_byte_array_order(column: &ColumnDescriptor) -> bool {
/// 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.
/// Non-floating signed logical types, such as decimals, retain their behavior.
/// 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.
/// Floating-point bounds omit NaNs, which Arrow orders below or above all
/// finite values depending on their sign. Without NaN-absence statistics,
/// neither endpoint bounds every non-null value in the column.
pub(crate) fn has_untrusted_min_max_order(
parquet_schema: &SchemaDescriptor,
column_orders: Option<&[ColumnOrder]>,
parquet_column_index: usize,
) -> bool {
let column = parquet_schema.column(parquet_column_index);
if column.sort_order() == SortOrder::UNDEFINED {
if column.sort_order() == SortOrder::UNDEFINED
|| matches!(
column.physical_type(),
PhysicalType::FLOAT | PhysicalType::DOUBLE
)
|| matches!(column.logical_type_ref(), Some(LogicalType::Float16))
{
return true;
}
requires_unsigned_byte_array_order(&column)
Expand Down
27 changes: 14 additions & 13 deletions datafusion/datasource-parquet/src/opener/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2823,9 +2823,10 @@ mod test {
async fn test_prune_on_statistics() {
let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;

// Integer bounds remain usable; Parquet floating bounds exclude possible NaNs.
let batch = record_batch!(
("a", Int32, vec![Some(1), Some(2), Some(2)]),
("b", Float32, vec![Some(1.0), Some(2.0), None])
("b", Int32, vec![Some(1), Some(2), None])
)
.unwrap();

Expand All @@ -2842,8 +2843,8 @@ mod test {
.add_column_statistics(ColumnStatistics::new_unknown())
.add_column_statistics(
ColumnStatistics::new_unknown()
.with_min_value(Precision::Exact(ScalarValue::Float32(Some(1.0))))
.with_max_value(Precision::Exact(ScalarValue::Float32(Some(2.0))))
.with_min_value(Precision::Exact(ScalarValue::Int32(Some(1))))
.with_max_value(Precision::Exact(ScalarValue::Int32(Some(2))))
.with_null_count(Precision::Exact(1)),
),
));
Expand All @@ -2867,8 +2868,8 @@ 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 = 5` should exclude all rows
let expr = col("b").eq(lit(ScalarValue::Int32(Some(5))));
let predicate = logical2physical(&expr, &schema);
let opener = make_opener(predicate);
let stream = open_file(&opener, file).await.unwrap();
Expand Down Expand Up @@ -2943,7 +2944,7 @@ mod test {

let batch = record_batch!(
("a", Int32, vec![Some(1), Some(2), Some(3)]),
("b", Float64, vec![Some(1.0), Some(2.0), None])
("b", Int64, vec![Some(1), Some(2), None])
)
.unwrap();
let data_size =
Expand All @@ -2959,15 +2960,15 @@ mod test {
.add_column_statistics(ColumnStatistics::new_unknown())
.add_column_statistics(
ColumnStatistics::new_unknown()
.with_min_value(Precision::Exact(ScalarValue::Float64(Some(1.0))))
.with_max_value(Precision::Exact(ScalarValue::Float64(Some(2.0))))
.with_min_value(Precision::Exact(ScalarValue::Int64(Some(1))))
.with_max_value(Precision::Exact(ScalarValue::Int64(Some(2))))
.with_null_count(Precision::Exact(1)),
),
));
let table_schema = Arc::new(Schema::new(vec![
Field::new("part", DataType::Int32, false),
Field::new("a", DataType::Int32, false),
Field::new("b", DataType::Float32, true),
Field::new("b", DataType::Int32, true),
]));
let table_schema_for_opener = TableSchemaBuilder::from(&file_schema)
.with_table_partition_cols(vec![Arc::new(Field::new(
Expand All @@ -2987,7 +2988,7 @@ mod test {
};

// Filter should match the partition value and file statistics
let expr = col("part").eq(lit(1)).and(col("b").eq(lit(1.0)));
let expr = col("part").eq(lit(1)).and(col("b").eq(lit(1i64)));
let predicate = logical2physical(&expr, &table_schema);
let opener = make_opener(predicate);
let stream = open_file(&opener, file.clone()).await.unwrap();
Expand All @@ -2996,7 +2997,7 @@ mod test {
assert_eq!(num_rows, 3);

// Should prune based on partition value but not file statistics
let expr = col("part").eq(lit(2)).and(col("b").eq(lit(1.0)));
let expr = col("part").eq(lit(2)).and(col("b").eq(lit(1i64)));
let predicate = logical2physical(&expr, &table_schema);
let opener = make_opener(predicate);
let stream = open_file(&opener, file.clone()).await.unwrap();
Expand All @@ -3005,7 +3006,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(7i64)));
let predicate = logical2physical(&expr, &table_schema);
let opener = make_opener(predicate);
let stream = open_file(&opener, file.clone()).await.unwrap();
Expand All @@ -3014,7 +3015,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(7i64)));
let predicate = logical2physical(&expr, &table_schema);
let opener = make_opener(predicate);
let stream = open_file(&opener, file).await.unwrap();
Expand Down
Loading