Skip to content
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

Fix data page statistics when all rows are null in a data page #11295

Merged
merged 7 commits into from
Jul 7, 2024
Merged

Fix data page statistics when all rows are null in a data page #11295

Show file tree
Hide file tree
Changes from 5 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c8926f2
Adds tests for data page statistics when all values on the page are n…
Jul 5, 2024
4fad730
Fix handling of data page statistics for FixedBinaryArray using a bui…
Jul 5, 2024
663c504
Fix data page all nulls stats test for Dictionary DataType.
Jul 5, 2024
829faf8
Fixes handling of None statistics for Decimal128 and Decimal256.
Jul 5, 2024
ecb8376
Consolidate make_data_page_stats_iterator uses.
Jul 5, 2024
a41a56b
Fix linting error.
Jul 6, 2024
a4f6a9e
Remove unnecessary collect.
Jul 7, 2024
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
130 changes: 69 additions & 61 deletions datafusion/core/src/datasource/physical_plan/parquet/statistics.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

// TODO: potentially move this to arrow-rs: https://github.com/apache/arrow-rs/issues/4328

use arrow::array::builder::FixedSizeBinaryBuilder;
use arrow::datatypes::i256;
use arrow::{array::ArrayRef, datatypes::DataType};
use arrow_array::{
Expand Down Expand Up @@ -600,6 +601,31 @@ make_data_page_stats_iterator!(
Index::DOUBLE,
f64
);
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just consolidating these together.

make_data_page_stats_iterator!(
MinByteArrayDataPageStatsIterator,
|x: &PageIndex<ByteArray>| { x.min.clone() },
Index::BYTE_ARRAY,
ByteArray
);
make_data_page_stats_iterator!(
MaxByteArrayDataPageStatsIterator,
|x: &PageIndex<ByteArray>| { x.max.clone() },
Index::BYTE_ARRAY,
ByteArray
);
make_data_page_stats_iterator!(
MaxFixedLenByteArrayDataPageStatsIterator,
|x: &PageIndex<FixedLenByteArray>| { x.max.clone() },
Index::FIXED_LEN_BYTE_ARRAY,
FixedLenByteArray
);

make_data_page_stats_iterator!(
MinFixedLenByteArrayDataPageStatsIterator,
|x: &PageIndex<FixedLenByteArray>| { x.min.clone() },
Index::FIXED_LEN_BYTE_ARRAY,
FixedLenByteArray
);

macro_rules! get_decimal_page_stats_iterator {
($iterator_type: ident, $func: ident, $stat_value_type: ident, $convert_func: ident) => {
Expand Down Expand Up @@ -634,9 +660,7 @@ macro_rules! get_decimal_page_stats_iterator {
.indexes
.iter()
.map(|x| {
Some($stat_value_type::from(
x.$func.unwrap_or_default(),
))
x.$func.and_then(|x| Some($stat_value_type::from(x)))
})
.collect::<Vec<_>>(),
),
Expand All @@ -645,9 +669,7 @@ macro_rules! get_decimal_page_stats_iterator {
.indexes
.iter()
.map(|x| {
Some($stat_value_type::from(
x.$func.unwrap_or_default(),
))
x.$func.and_then(|x| Some($stat_value_type::from(x)))
})
.collect::<Vec<_>>(),
),
Expand All @@ -656,9 +678,9 @@ macro_rules! get_decimal_page_stats_iterator {
.indexes
.iter()
.map(|x| {
Some($convert_func(
x.clone().$func.unwrap_or_default().data(),
))
x.clone()
.$func
.and_then(|x| Some($convert_func(x.data())))
})
.collect::<Vec<_>>(),
),
Expand All @@ -667,9 +689,9 @@ macro_rules! get_decimal_page_stats_iterator {
.indexes
.iter()
.map(|x| {
Some($convert_func(
x.clone().$func.unwrap_or_default().data(),
))
x.clone()
.$func
.and_then(|x| Some($convert_func(x.data())))
})
.collect::<Vec<_>>(),
),
Expand Down Expand Up @@ -713,32 +735,6 @@ get_decimal_page_stats_iterator!(
i256,
from_bytes_to_i256
);
make_data_page_stats_iterator!(
MinByteArrayDataPageStatsIterator,
|x: &PageIndex<ByteArray>| { x.min.clone() },
Index::BYTE_ARRAY,
ByteArray
);
make_data_page_stats_iterator!(
MaxByteArrayDataPageStatsIterator,
|x: &PageIndex<ByteArray>| { x.max.clone() },
Index::BYTE_ARRAY,
ByteArray
);

make_data_page_stats_iterator!(
MaxFixedLenByteArrayDataPageStatsIterator,
|x: &PageIndex<FixedLenByteArray>| { x.max.clone() },
Index::FIXED_LEN_BYTE_ARRAY,
FixedLenByteArray
);

make_data_page_stats_iterator!(
MinFixedLenByteArrayDataPageStatsIterator,
|x: &PageIndex<FixedLenByteArray>| { x.min.clone() },
Index::FIXED_LEN_BYTE_ARRAY,
FixedLenByteArray
);

macro_rules! get_data_page_statistics {
($stat_type_prefix: ident, $data_type: ident, $iterator: ident) => {
Expand All @@ -757,7 +753,7 @@ macro_rules! get_data_page_statistics {
UInt8Array::from_iter(
[<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| u8::try_from(x).ok())
})
})
Expand All @@ -768,7 +764,7 @@ macro_rules! get_data_page_statistics {
UInt16Array::from_iter(
[<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| u16::try_from(x).ok())
})
})
Expand All @@ -779,7 +775,7 @@ macro_rules! get_data_page_statistics {
UInt32Array::from_iter(
[<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| Some(x as u32))
})
})
Expand All @@ -789,7 +785,7 @@ macro_rules! get_data_page_statistics {
UInt64Array::from_iter(
[<$stat_type_prefix Int64DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| Some(x as u64))
})
})
Expand All @@ -799,7 +795,7 @@ macro_rules! get_data_page_statistics {
Int8Array::from_iter(
[<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| i8::try_from(x).ok())
})
})
Expand All @@ -810,7 +806,7 @@ macro_rules! get_data_page_statistics {
Int16Array::from_iter(
[<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| i16::try_from(x).ok())
})
})
Expand All @@ -823,11 +819,11 @@ macro_rules! get_data_page_statistics {
Float16Array::from_iter(
[<$stat_type_prefix Float16DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter().filter_map(|x| {
x.and_then(|x| Some(from_bytes_to_f16(x.data())))
x.into_iter().map(|x| {
x.and_then(|x| from_bytes_to_f16(x.data()))
})
})
.flatten()
.flatten().collect::<Vec<_>>()
efredine marked this conversation as resolved.
Show resolved Hide resolved
)
)),
Some(DataType::Float32) => Ok(Arc::new(Float32Array::from_iter([<$stat_type_prefix Float32DataPageStatsIterator>]::new($iterator).flatten()))),
Expand All @@ -836,7 +832,7 @@ macro_rules! get_data_page_statistics {
Some(DataType::LargeBinary) => Ok(Arc::new(LargeBinaryArray::from_iter([<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).flatten()))),
Some(DataType::Utf8) => Ok(Arc::new(StringArray::from(
[<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| {
let res = std::str::from_utf8(x.data()).map(|s| s.to_string()).ok();
if res.is_none() {
Expand All @@ -849,7 +845,7 @@ macro_rules! get_data_page_statistics {
))),
Some(DataType::LargeUtf8) => Ok(Arc::new(LargeStringArray::from(
[<$stat_type_prefix ByteArrayDataPageStatsIterator>]::new($iterator).map(|x| {
x.into_iter().filter_map(|x| {
x.into_iter().map(|x| {
x.and_then(|x| {
let res = std::str::from_utf8(x.data()).map(|s| s.to_string()).ok();
if res.is_none() {
Expand Down Expand Up @@ -878,10 +874,10 @@ macro_rules! get_data_page_statistics {
Date64Array::from([<$stat_type_prefix Int32DataPageStatsIterator>]::new($iterator)
.map(|x| {
x.into_iter()
.filter_map(|x| {
.map(|x| {
x.and_then(|x| i64::try_from(x).ok())
.map(|x| x * 24 * 60 * 60 * 1000)
})
.map(|x| x * 24 * 60 * 60 * 1000)
}).flatten().collect::<Vec<_>>()
)
)
Expand Down Expand Up @@ -919,16 +915,28 @@ macro_rules! get_data_page_statistics {
})
},
Some(DataType::FixedSizeBinary(size)) => {
Ok(Arc::new(
FixedSizeBinaryArray::try_from_iter(
[<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator)
.flat_map(|x| x.into_iter())
.filter_map(|x| x)
).unwrap_or_else(|e| {
log::debug!("FixedSizeBinary statistics is invalid: {}", e);
FixedSizeBinaryArray::new(*size, vec![].into(), None)
})
))
let mut builder = FixedSizeBinaryBuilder::new(*size);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this looks like a nice change to me

let iterator = [<$stat_type_prefix FixedLenByteArrayDataPageStatsIterator>]::new($iterator);
for x in iterator {
for x in x.into_iter() {
let Some(x) = x else {
builder.append_null(); // no statistics value
continue;
};

if x.len() == *size as usize {
let _ = builder.append_value(x.data());
} else {
log::debug!(
"FixedSizeBinary({}) statistics is a binary of size {}, ignoring it.",
size,
x.len(),
);
builder.append_null();
}
}
}
Ok(Arc::new(builder.finish()))
},
_ => unimplemented!()
}
Expand Down
Loading
Loading