This is an automated email from the ASF dual-hosted git repository.
etseidl pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new efb6d1816e Implement PARQUET-2249: Introduce IEEE 754 total order
(#9619)
efb6d1816e is described below
commit efb6d1816e659b481fd69a2a7e50cdef858b50ab
Author: Ed Seidl <[email protected]>
AuthorDate: Mon Aug 3 10:51:47 2026 -0700
Implement PARQUET-2249: Introduce IEEE 754 total order (#9619)
# Which issue does this PR close?
- Closes #8156.
- Closes https://github.com/apache/arrow-rs/pull/8158
# Rationale for this change
This takes the implementation done by @Xuanwo (#8158) and updates it to
the new thrift format and recent changes to the original proposal
(https://github.com/apache/parquet-format/pull/514).
# What changes are included in this PR?
Adds needed thrift structures as well as NaN counts for pages and column
chunks.
# Are these changes tested?
Yes, new tests added (more may be needed).
# Are there any user-facing changes?
Yes. This adds the `IEEE_754_TOTAL_ORDER` variant to the public
`ColumnOrder` enum, and `TOTAL_ORDER` to the public `SortOrder` enum,
which are technically breaking API changes. This also adds a `nan_count`
argument to the public function `ColumnIndexBuilder::append`.
Non-breaking changes include adding `nan_count` fields to the
`ValueStatistics` and `ColumnIndex` structs.
Behaviorally, this will now order all floating point statistics using
IEEE754 total order, and all floating point columns will use the new
`IEEE_754_TOTAL_ORDER` column order in the `FileMetaData.column_orders`
list. Newer readers that recognize the new column order may now safely
use the statistics for chunk and page pruning. Older readers will not
recognize this column order and should continue to ignore the
statistics.
---
parquet/Cargo.toml | 5 +
parquet/src/arrow/arrow_reader/statistics.rs | 80 +++++
parquet/src/arrow/arrow_writer/byte_array.rs | 2 +
parquet/src/arrow/arrow_writer/mod.rs | 115 ++++++-
parquet/src/basic.rs | 126 ++++++--
parquet/src/column/writer/encoder.rs | 103 +++---
parquet/src/column/writer/mod.rs | 432 +++++++++++++++++++-------
parquet/src/file/metadata/mod.rs | 82 ++++-
parquet/src/file/metadata/thrift/mod.rs | 50 ++-
parquet/src/file/metadata/writer.rs | 9 +-
parquet/src/file/page_index/column_index.rs | 58 +++-
parquet/src/file/page_index/index_reader.rs | 1 +
parquet/src/file/statistics.rs | 218 ++++++++++++-
parquet/src/file/writer.rs | 16 +-
parquet/src/schema/types.rs | 42 ++-
parquet/tests/arrow_reader/statistics.rs | 10 +-
parquet/tests/ieee754_nan_interop.rs | 448 +++++++++++++++++++++++++++
17 files changed, 1545 insertions(+), 252 deletions(-)
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index b559b4bbc9..d34fbb5e52 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -178,6 +178,11 @@ name = "arrow_writer"
required-features = ["arrow"]
path = "./tests/arrow_writer.rs"
+[[test]]
+name = "ieee754_nan_interop"
+required-features = ["arrow"]
+path = "./tests/ieee754_nan_interop.rs"
+
[[test]]
name = "encryption"
required-features = ["arrow"]
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs
b/parquet/src/arrow/arrow_reader/statistics.rs
index c096448b6b..66d437243f 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -1408,6 +1408,35 @@ where
Ok(array)
}
+/// Extracts the NaN count statistics from an iterator
+/// of parquet page [`ColumnIndexMetaData`]'s to an [`ArrayRef`]
+///
+/// The returned Array is an [`UInt64Array`]
+pub(crate) fn nan_counts_page_statistics<'a, I>(iterator: I) ->
Result<UInt64Array>
+where
+ I: Iterator<Item = (usize, &'a ColumnIndexMetaData)>,
+{
+ let chunks: Vec<_> = iterator.collect();
+ let total_capacity: usize = chunks.iter().map(|(len, _)| *len).sum();
+ let mut values = Vec::with_capacity(total_capacity);
+ let mut nulls = NullBufferBuilder::new(total_capacity);
+ for (len, index) in chunks {
+ match index.nan_counts() {
+ Some(counts) => {
+ values.extend(counts.iter().map(|&x| x as u64));
+ nulls.append_n_non_nulls(len);
+ }
+ None => {
+ values.resize(values.len() + len, 0);
+ nulls.append_n_nulls(len);
+ }
+ }
+ }
+ let null_buffer = nulls.build();
+ let array = UInt64Array::new(values.into(), null_buffer);
+ Ok(array)
+}
+
/// Extracts Parquet statistics as Arrow arrays
///
/// This is used to convert Parquet statistics to Arrow [`ArrayRef`], with
@@ -1770,6 +1799,28 @@ impl<'a> StatisticsConverter<'a> {
Ok(UInt64Array::from_iter(null_counts))
}
+ /// Extract the NaN counts from row group statistics in
[`RowGroupMetaData`]
+ ///
+ /// See docs on [`Self::row_group_mins`] for details
+ pub fn row_group_nan_counts<I>(&self, metadatas: I) -> Result<UInt64Array>
+ where
+ I: IntoIterator<Item = &'a RowGroupMetaData>,
+ {
+ let Some(parquet_index) = self.parquet_column_index else {
+ let num_row_groups = metadatas.into_iter().count();
+ return Ok(UInt64Array::from_iter(std::iter::repeat_n(
+ None,
+ num_row_groups,
+ )));
+ };
+
+ let nan_counts = metadatas
+ .into_iter()
+ .map(|x| x.column(parquet_index).statistics())
+ .map(|s| s.and_then(|s| s.nan_count_opt()));
+ Ok(UInt64Array::from_iter(nan_counts))
+ }
+
/// Extract the minimum values from Data Page statistics.
///
/// In Parquet files, in addition to the Column Chunk level statistics
@@ -1909,6 +1960,35 @@ impl<'a> StatisticsConverter<'a> {
null_counts_page_statistics(iter)
}
+ /// Returns a [`UInt64Array`] with NaN counts for each data page.
+ ///
+ /// See docs on [`Self::data_page_mins`] for details.
+ pub fn data_page_nan_counts<I>(
+ &self,
+ column_page_index: &ParquetColumnIndex,
+ column_offset_index: &ParquetOffsetIndex,
+ row_group_indices: I,
+ ) -> Result<UInt64Array>
+ where
+ I: IntoIterator<Item = &'a usize>,
+ {
+ let Some(parquet_index) = self.parquet_column_index else {
+ let num_row_groups = row_group_indices.into_iter().count();
+ return Ok(UInt64Array::new_null(num_row_groups));
+ };
+
+ let iter = row_group_indices.into_iter().map(|rg_index| {
+ let column_page_index_per_row_group_per_column =
+ &column_page_index[*rg_index][parquet_index];
+ let num_data_pages = &column_offset_index[*rg_index][parquet_index]
+ .page_locations()
+ .len();
+
+ (*num_data_pages, column_page_index_per_row_group_per_column)
+ });
+ nan_counts_page_statistics(iter)
+ }
+
/// Returns a [`UInt64Array`] with row counts for each data page.
///
/// This function iterates over the given row group indexes and computes
diff --git a/parquet/src/arrow/arrow_writer/byte_array.rs
b/parquet/src/arrow/arrow_writer/byte_array.rs
index 145431c264..ea45edeea2 100644
--- a/parquet/src/arrow/arrow_writer/byte_array.rs
+++ b/parquet/src/arrow/arrow_writer/byte_array.rs
@@ -293,6 +293,7 @@ impl FallbackEncoder {
encoding,
min_value,
max_value,
+ nan_count: None,
variable_length_bytes,
})
}
@@ -415,6 +416,7 @@ impl DictEncoder {
encoding: Encoding::RLE_DICTIONARY,
min_value,
max_value,
+ nan_count: None,
variable_length_bytes,
}
}
diff --git a/parquet/src/arrow/arrow_writer/mod.rs
b/parquet/src/arrow/arrow_writer/mod.rs
index 8b9aea2e12..2a96170adc 100644
--- a/parquet/src/arrow/arrow_writer/mod.rs
+++ b/parquet/src/arrow/arrow_writer/mod.rs
@@ -1945,6 +1945,7 @@ fn chunk_contiguous_vec(arena: Vec<u8>, chunk_size:
usize) -> Vec<FixedLenByteAr
#[cfg(test)]
mod tests {
use super::*;
+ use std::cmp::Ordering;
use std::collections::HashMap;
use std::fs::File;
@@ -3417,10 +3418,120 @@ mod tests {
for column in row_group.columns() {
assert!(column.offset_index_offset().is_some());
assert!(column.offset_index_length().is_some());
- assert!(column.column_index_offset().is_none());
- assert!(column.column_index_length().is_none());
+ assert!(column.column_index_offset().is_some());
+ assert!(column.column_index_length().is_some());
}
}
+ assert!(file_meta_data.column_index().is_some());
+ if let Some(col_indexes) = file_meta_data.column_index() {
+ for rg_idx in col_indexes {
+ for idx in rg_idx {
+ assert!(idx.nan_counts().is_some());
+ let float_idx = match idx {
+ ColumnIndexMetaData::DOUBLE(idx) => idx,
+ _ => panic!("expected double statistics"),
+ };
+ for i in 0..idx.num_pages() as usize {
+ assert_eq!(float_idx.nan_count(i), Some(10));
+ assert_eq!(
+
f64::NAN.total_cmp(float_idx.min_value(i).unwrap()),
+ Ordering::Equal
+ );
+ assert_eq!(
+
f64::NAN.total_cmp(float_idx.max_value(i).unwrap()),
+ Ordering::Equal
+ );
+ }
+ }
+ }
+ }
+ }
+
+ #[test]
+ fn check_page_offset_index_with_mixed_nan() {
+ let schema = Arc::new(Schema::new(vec![Field::new(
+ "col",
+ DataType::Float64,
+ true,
+ )]));
+
+ let mut out = Vec::with_capacity(1024);
+ let props = WriterProperties::builder()
+ .set_data_page_row_count_limit(10)
+ .build();
+ let mut writer = ArrowWriter::try_new(&mut out, schema.clone(),
Some(props))
+ .expect("Unable to write file");
+
+ // write a page of all NaN (since batch min and max are NaN, global
min/max are NaN)
+ let values = Arc::new(Float64Array::from(vec![f64::NAN; 10]));
+ let batch = RecordBatch::try_new(schema.clone(),
vec![values]).unwrap();
+ writer.write(&batch).unwrap();
+
+ // write a page of all -NaN (batch min/max is -NaN, should update
global min to -NaN)
+ let values = Arc::new(Float64Array::from(vec![-f64::NAN; 10]));
+ let batch = RecordBatch::try_new(schema.clone(),
vec![values]).unwrap();
+ writer.write(&batch).unwrap();
+
+ // write a page of all 0 (non-NaN should override global min/max, now
0/0)
+ let values = Arc::new(Float64Array::from(vec![0_f64; 10]));
+ let batch = RecordBatch::try_new(schema.clone(),
vec![values]).unwrap();
+ writer.write(&batch).unwrap();
+
+ // write a mixed page (should now have min -1, max 1)
+ let values = Arc::new(Float64Array::from(vec![
+ -1.0,
+ 0.0,
+ f64::NAN,
+ -f64::NAN,
+ 1.0,
+ ]));
+ let batch = RecordBatch::try_new(schema.clone(),
vec![values]).unwrap();
+ writer.write(&batch).unwrap();
+
+ let file_meta_data = writer.close().unwrap();
+
+ // check the column chunk stats are correct
+ let col_stats = file_meta_data
+ .row_group(0)
+ .column(0)
+ .statistics()
+ .expect("missing column chunk statistics");
+
+ assert_eq!(col_stats.nan_count_opt(), Some(22));
+ assert_eq!(col_stats.min_bytes_opt(), Some((-1.0f64).as_bytes()));
+ assert_eq!(col_stats.max_bytes_opt(), Some(1.0f64.as_bytes()));
+
+ assert!(file_meta_data.column_index().is_some());
+ let col_idx = &file_meta_data.column_index().as_ref().unwrap()[0][0];
+ assert_eq!(col_idx.num_pages(), 4);
+
+ // test each page
+ let float_idx = match col_idx {
+ ColumnIndexMetaData::DOUBLE(idx) => idx,
+ _ => panic!("expected double statistics"),
+ };
+
+ assert_eq!(float_idx.nan_counts, Some(vec![10, 10, 0, 2]));
+ assert_eq!(
+ f64::NAN.total_cmp(float_idx.min_value(0).unwrap()),
+ Ordering::Equal
+ );
+ assert_eq!(
+ f64::NAN.total_cmp(float_idx.max_value(0).unwrap()),
+ Ordering::Equal
+ );
+ assert_eq!(
+ (-f64::NAN).total_cmp(float_idx.min_value(1).unwrap()),
+ Ordering::Equal
+ );
+ assert_eq!(
+ (-f64::NAN).total_cmp(float_idx.max_value(1).unwrap()),
+ Ordering::Equal
+ );
+ assert_eq!(float_idx.min_value(2), Some(&0.0));
+ assert_eq!(float_idx.max_value(2), Some(&0.0));
+ assert_eq!(float_idx.min_value(3), Some(&-1.0));
+ assert_eq!(float_idx.max_value(3), Some(&1.0));
}
#[test]
diff --git a/parquet/src/basic.rs b/parquet/src/basic.rs
index b4f18f3117..19b89738a0 100644
--- a/parquet/src/basic.rs
+++ b/parquet/src/basic.rs
@@ -985,6 +985,8 @@ pub enum SortOrder {
UNSIGNED,
/// Comparison is undefined.
UNDEFINED,
+ /// Use IEEE 754 total order.
+ TOTAL_ORDER,
}
impl SortOrder {
@@ -1005,6 +1007,8 @@ pub enum ColumnOrder {
/// Column uses the order defined by its logical or physical type
/// (if there is no logical type), parquet-format 2.4.0+.
TYPE_DEFINED_ORDER(SortOrder),
+ /// Column ordering to use for floating point types.
+ IEEE_754_TOTAL_ORDER,
// The following are not defined in the Parquet spec and should always be
last.
/// Undefined column order, means legacy behaviour before parquet-format
2.4.0.
/// Sort order is always SIGNED.
@@ -1025,14 +1029,36 @@ impl ColumnOrder {
converted_type: ConvertedType,
physical_type: Type,
) -> SortOrder {
- Self::sort_order_for_type(logical_type.as_ref(), converted_type,
physical_type)
+ Self::column_order_for_type(logical_type.as_ref(), converted_type,
physical_type)
+ .sort_order()
+ }
+
+ /// Returns the `ColumnOrder` for a physical/logical type.
+ pub fn column_order_for_type(
+ logical_type: Option<&LogicalType>,
+ converted_type: ConvertedType,
+ physical_type: Type,
+ ) -> ColumnOrder {
+ if Some(&LogicalType::Float16) == logical_type
+ || matches!(physical_type, Type::FLOAT | Type::DOUBLE)
+ {
+ ColumnOrder::IEEE_754_TOTAL_ORDER
+ } else {
+ let sort_order =
+ Self::sort_order_for_type(logical_type, converted_type,
physical_type, true);
+ ColumnOrder::TYPE_DEFINED_ORDER(sort_order)
+ }
}
/// Returns sort order for a physical/logical type.
+ ///
+ /// `is_type_defined` indicates whether the column order for this type is
+ /// [`ColumnOrder::TYPE_DEFINED_ORDER`].
pub fn sort_order_for_type(
logical_type: Option<&LogicalType>,
converted_type: ConvertedType,
physical_type: Type,
+ is_type_defined: bool,
) -> SortOrder {
match logical_type {
Some(logical) => match logical {
@@ -1050,18 +1076,28 @@ impl ColumnOrder {
LogicalType::Timestamp(_) => SortOrder::SIGNED,
LogicalType::Unknown => SortOrder::UNDEFINED,
LogicalType::Uuid => SortOrder::UNSIGNED,
- LogicalType::Float16 => SortOrder::SIGNED,
+ LogicalType::Float16 => {
+ if is_type_defined {
+ SortOrder::SIGNED
+ } else {
+ SortOrder::TOTAL_ORDER
+ }
+ }
LogicalType::Variant(_)
| LogicalType::Geometry(_)
| LogicalType::Geography(_)
| LogicalType::_Unknown { .. } => SortOrder::UNDEFINED,
},
// Fall back to converted type
- None => Self::get_converted_sort_order(converted_type,
physical_type),
+ None => Self::get_converted_sort_order(converted_type,
physical_type, is_type_defined),
}
}
- fn get_converted_sort_order(converted_type: ConvertedType, physical_type:
Type) -> SortOrder {
+ fn get_converted_sort_order(
+ converted_type: ConvertedType,
+ physical_type: Type,
+ is_type_defined: bool,
+ ) -> SortOrder {
match converted_type {
// Unsigned byte-wise comparison.
ConvertedType::UTF8
@@ -1096,24 +1132,35 @@ impl ColumnOrder {
}
// Fall back to physical type.
- ConvertedType::NONE => Self::get_default_sort_order(physical_type),
+ ConvertedType::NONE => Self::get_default_sort_order(physical_type,
is_type_defined),
}
}
/// Returns default sort order based on physical type.
- fn get_default_sort_order(physical_type: Type) -> SortOrder {
+ fn get_default_sort_order(physical_type: Type, is_type_defined: bool) ->
SortOrder {
match physical_type {
// Order: false, true
Type::BOOLEAN => SortOrder::UNSIGNED,
Type::INT32 | Type::INT64 => SortOrder::SIGNED,
Type::INT96 => SortOrder::UNDEFINED,
// Notes to remember when comparing float/double values:
- // If the min is a NaN, it should be ignored.
- // If the max is a NaN, it should be ignored.
- // If the min is +0, the row group may contain -0 values as well.
- // If the max is -0, the row group may contain +0 values as well.
- // When looking for NaN values, min and max should be ignored.
- Type::FLOAT | Type::DOUBLE => SortOrder::SIGNED,
+ // If legacy TYPE_DEFINED_ORDER is specified:
+ // If the min is a NaN, it should be ignored.
+ // If the max is a NaN, it should be ignored.
+ // If the min is +0, the row group may contain -0 values as well.
+ // If the max is -0, the row group may contain +0 values as well.
+ // When looking for NaN values, min and max should be ignored.
+ // If IEEE_754_TOTAL_ORDER:
+ // Examine nan_count to see if NaNs are present.
+ // If min/max are NaN, that means only NaNs are present.
+ // If min/max are not NaN, they are ordered according to total
order.
+ Type::FLOAT | Type::DOUBLE => {
+ if is_type_defined {
+ SortOrder::SIGNED
+ } else {
+ SortOrder::TOTAL_ORDER
+ }
+ }
// Unsigned byte-wise comparison
Type::BYTE_ARRAY | Type::FIXED_LEN_BYTE_ARRAY =>
SortOrder::UNSIGNED,
}
@@ -1123,6 +1170,7 @@ impl ColumnOrder {
pub fn sort_order(&self) -> SortOrder {
match *self {
ColumnOrder::TYPE_DEFINED_ORDER(order) => order,
+ ColumnOrder::IEEE_754_TOTAL_ORDER => SortOrder::TOTAL_ORDER,
ColumnOrder::UNDEFINED => SortOrder::SIGNED,
ColumnOrder::UNKNOWN => SortOrder::UNDEFINED,
}
@@ -1141,6 +1189,10 @@ impl<'a, R: ThriftCompactInputProtocol<'a>>
ReadThrift<'a, R> for ColumnOrder {
prot.skip_empty_struct()?;
Self::TYPE_DEFINED_ORDER(SortOrder::SIGNED)
}
+ 2 => {
+ prot.skip_empty_struct()?;
+ Self::IEEE_754_TOTAL_ORDER
+ }
_ => {
prot.skip(field_ident.field_type)?;
Self::UNKNOWN
@@ -1165,6 +1217,10 @@ impl WriteThrift for ColumnOrder {
writer.write_field_begin(FieldType::Struct, 1, 0)?;
writer.write_struct_end()?;
}
+ Self::IEEE_754_TOTAL_ORDER => {
+ writer.write_field_begin(FieldType::Struct, 2, 0)?;
+ writer.write_struct_end()?;
+ }
_ => return Err(general_err!("Attempt to write undefined
ColumnOrder")),
}
// write end of struct for this union
@@ -1964,6 +2020,7 @@ mod tests {
assert_eq!(SortOrder::SIGNED.to_string(), "SIGNED");
assert_eq!(SortOrder::UNSIGNED.to_string(), "UNSIGNED");
assert_eq!(SortOrder::UNDEFINED.to_string(), "UNDEFINED");
+ assert_eq!(SortOrder::TOTAL_ORDER.to_string(), "TOTAL_ORDER");
}
#[test]
@@ -1980,6 +2037,10 @@ mod tests {
ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED).to_string(),
"TYPE_DEFINED_ORDER(UNDEFINED)"
);
+ assert_eq!(
+ ColumnOrder::IEEE_754_TOTAL_ORDER.to_string(),
+ "IEEE_754_TOTAL_ORDER"
+ );
assert_eq!(ColumnOrder::UNDEFINED.to_string(), "UNDEFINED");
}
@@ -1996,7 +2057,12 @@ mod tests {
fn check_sort_order(types: Vec<LogicalType>, expected_order:
SortOrder) {
for tpe in types {
assert_eq!(
- ColumnOrder::get_sort_order(Some(tpe),
ConvertedType::NONE, Type::BYTE_ARRAY),
+ ColumnOrder::column_order_for_type(
+ Some(&tpe),
+ ConvertedType::NONE,
+ Type::BYTE_ARRAY
+ )
+ .sort_order(),
expected_order
);
}
@@ -2030,10 +2096,12 @@ mod tests {
LogicalType::timestamp(false, TimeUnit::MILLIS),
LogicalType::timestamp(false, TimeUnit::MICROS),
LogicalType::timestamp(true, TimeUnit::NANOS),
- LogicalType::Float16,
];
check_sort_order(signed, SortOrder::SIGNED);
+ let float = vec![LogicalType::Float16];
+ check_sort_order(float, SortOrder::TOTAL_ORDER);
+
// Undefined comparison
let undefined = vec![
LogicalType::List,
@@ -2052,7 +2120,7 @@ mod tests {
fn check_sort_order(types: Vec<ConvertedType>, expected_order:
SortOrder) {
for tpe in types {
assert_eq!(
- ColumnOrder::get_sort_order(None, tpe, Type::BYTE_ARRAY),
+ ColumnOrder::column_order_for_type(None, tpe,
Type::BYTE_ARRAY).sort_order(),
expected_order
);
}
@@ -2104,35 +2172,43 @@ mod tests {
fn test_column_order_get_default_sort_order() {
// Comparison based on physical type
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::BOOLEAN),
+ ColumnOrder::get_default_sort_order(Type::BOOLEAN, true),
SortOrder::UNSIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::INT32),
+ ColumnOrder::get_default_sort_order(Type::INT32, true),
SortOrder::SIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::INT64),
+ ColumnOrder::get_default_sort_order(Type::INT64, true),
SortOrder::SIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::INT96),
+ ColumnOrder::get_default_sort_order(Type::INT96, true),
SortOrder::UNDEFINED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::FLOAT),
+ ColumnOrder::get_default_sort_order(Type::FLOAT, false),
+ SortOrder::TOTAL_ORDER
+ );
+ assert_eq!(
+ ColumnOrder::get_default_sort_order(Type::DOUBLE, false),
+ SortOrder::TOTAL_ORDER
+ );
+ assert_eq!(
+ ColumnOrder::get_default_sort_order(Type::FLOAT, true),
SortOrder::SIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::DOUBLE),
+ ColumnOrder::get_default_sort_order(Type::DOUBLE, true),
SortOrder::SIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::BYTE_ARRAY),
+ ColumnOrder::get_default_sort_order(Type::BYTE_ARRAY, true),
SortOrder::UNSIGNED
);
assert_eq!(
- ColumnOrder::get_default_sort_order(Type::FIXED_LEN_BYTE_ARRAY),
+ ColumnOrder::get_default_sort_order(Type::FIXED_LEN_BYTE_ARRAY,
true),
SortOrder::UNSIGNED
);
}
@@ -2151,6 +2227,10 @@ mod tests {
ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED).sort_order(),
SortOrder::UNDEFINED
);
+ assert_eq!(
+ ColumnOrder::IEEE_754_TOTAL_ORDER.sort_order(),
+ SortOrder::TOTAL_ORDER
+ );
assert_eq!(ColumnOrder::UNDEFINED.sort_order(), SortOrder::SIGNED);
}
diff --git a/parquet/src/column/writer/encoder.rs
b/parquet/src/column/writer/encoder.rs
index d9adacff41..094c7c26a0 100644
--- a/parquet/src/column/writer/encoder.rs
+++ b/parquet/src/column/writer/encoder.rs
@@ -16,7 +16,6 @@
// under the License.
use bytes::Bytes;
-use half::f16;
use crate::basic::{ConvertedType, Encoding, LogicalType, Type};
use crate::bloom_filter::Sbbf;
@@ -30,7 +29,7 @@ use crate::errors::{ParquetError, Result};
use crate::file::properties::{EnabledStatistics, WriterProperties};
use crate::geospatial::accumulator::{GeoStatsAccumulator,
try_new_geo_stats_accumulator};
use crate::geospatial::statistics::GeospatialStatistics;
-use crate::schema::types::{ColumnDescPtr, ColumnDescriptor};
+use crate::schema::types::{BasicTypeInfo, ColumnDescPtr};
/// A collection of [`ParquetValueType`] encoded by a [`ColumnValueEncoder`]
pub trait ColumnValues {
@@ -65,6 +64,7 @@ pub struct DataPageValues<T> {
pub encoding: Encoding,
pub min_value: Option<T>,
pub max_value: Option<T>,
+ pub nan_count: Option<u64>,
pub variable_length_bytes: Option<i64>,
}
@@ -173,6 +173,7 @@ pub struct ColumnValueEncoderImpl<T: DataType> {
statistics_enabled: EnabledStatistics,
min_value: Option<T::T>,
max_value: Option<T::T>,
+ nan_count: Option<u64>,
bloom_filter: Option<Sbbf>,
bloom_filter_target_fpp: f64,
variable_length_bytes: Option<i64>,
@@ -180,11 +181,9 @@ pub struct ColumnValueEncoderImpl<T: DataType> {
}
impl<T: DataType> ColumnValueEncoderImpl<T> {
- fn min_max(&self, values: &[T::T], value_indices: Option<&[usize]>) ->
Option<(T::T, T::T)> {
- match value_indices {
- Some(indices) => get_min_max(&self.descr, indices.iter().map(|x|
&values[*x])),
- None => get_min_max(&self.descr, values.iter()),
- }
+ fn is_floating_point_column(&self) -> bool {
+ matches!(self.descr.physical_type(), Type::FLOAT | Type::DOUBLE)
+ || self.descr.logical_type_ref() == Some(&LogicalType::Float16)
}
fn write_slice(&mut self, slice: &[T::T]) -> Result<()> {
@@ -194,9 +193,14 @@ impl<T: DataType> ColumnValueEncoderImpl<T> {
{
if let Some(accumulator) =
self.geo_stats_accumulator.as_deref_mut() {
update_geo_stats_accumulator(accumulator, slice.iter());
- } else if let Some((min, max)) = self.min_max(slice, None) {
+ } else if let Some((min, max, nan_count)) =
+ get_min_max(self.descr.get_basic_info(), slice.iter())
+ {
update_min(&self.descr, &min, &mut self.min_value);
update_max(&self.descr, &max, &mut self.max_value);
+ if self.is_floating_point_column() {
+ *self.nan_count.get_or_insert(0) += nan_count;
+ }
}
if let Some(var_bytes) = T::T::variable_length_bytes(slice) {
@@ -258,6 +262,7 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
bloom_filter_target_fpp,
min_value: None,
max_value: None,
+ nan_count: None,
variable_length_bytes: None,
geo_stats_accumulator,
})
@@ -386,6 +391,7 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
num_values: std::mem::take(&mut self.num_values),
min_value: self.min_value.take(),
max_value: self.max_value.take(),
+ nan_count: self.nan_count.take(),
variable_length_bytes: self.variable_length_bytes.take(),
})
}
@@ -395,63 +401,52 @@ impl<T: DataType> ColumnValueEncoder for
ColumnValueEncoderImpl<T> {
}
}
-fn get_min_max<'a, T, I>(descr: &ColumnDescriptor, mut iter: I) -> Option<(T,
T)>
+// Get min and max values for all values in `iter`.
+//
+// For floating point we need to compare NaN values until we encounter a
non-NaN
+// value which then becomes the new min/max. After this, only non-NaN values
are
+// evaluated. If all values are NaN, then the min/max NaNs as determined by
+// IEEE 754 total order are returned.
+fn get_min_max<'a, T, I>(basic_type_info: &BasicTypeInfo, mut iter: I) ->
Option<(T, T, u64)>
where
T: ParquetValueType + 'a,
I: Iterator<Item = &'a T>,
{
- let first = loop {
- let next = iter.next()?;
- if !is_nan(descr, next) {
- break next;
- }
- };
+ let first = iter.next()?;
+ let mut min_max_nan = is_nan(basic_type_info, first);
+ let mut nan_count = min_max_nan as u64;
let mut min = first;
let mut max = first;
for val in iter {
- if is_nan(descr, val) {
- continue;
- }
- if compare_greater(descr, min, val) {
- min = val;
- }
- if compare_greater(descr, val, max) {
- max = val;
+ match (min_max_nan, is_nan(basic_type_info, val)) {
+ // skip NaNs if we've encounter non-NaN
+ (false, true) => {
+ nan_count += 1;
+ continue;
+ }
+ // if min/max are NaN, check for non-NaN and reset
+ (true, false) => {
+ min = val;
+ max = val;
+ min_max_nan = false;
+ continue;
+ }
+ // both are NaN or non-NaN, so do the comparison
+ (_, val_is_nan) => {
+ nan_count += val_is_nan as u64;
+ // we've already initialized min and max, so a single value
can't be both
+ // extremes
+ if compare_greater(basic_type_info, min, val) {
+ min = val;
+ } else if compare_greater(basic_type_info, val, max) {
+ max = val;
+ }
+ }
}
}
- // Float/Double statistics have special case for zero.
- //
- // If computed min is zero, whether negative or positive,
- // the spec states that the min should be written as -0.0
- // (negative zero)
- //
- // For max, it has similar logic but will be written as 0.0
- // (positive zero)
- let min = replace_zero(min, descr, -0.0);
- let max = replace_zero(max, descr, 0.0);
-
- Some((min, max))
-}
-
-#[inline]
-fn replace_zero<T: ParquetValueType>(val: &T, descr: &ColumnDescriptor,
replace: f32) -> T {
- match T::PHYSICAL_TYPE {
- Type::FLOAT if f32::from_le_bytes(val.as_bytes().try_into().unwrap())
== 0.0 => {
- T::try_from_le_slice(&f32::to_le_bytes(replace)).unwrap()
- }
- Type::DOUBLE if f64::from_le_bytes(val.as_bytes().try_into().unwrap())
== 0.0 => {
- T::try_from_le_slice(&f64::to_le_bytes(replace as f64)).unwrap()
- }
- Type::FIXED_LEN_BYTE_ARRAY
- if descr.logical_type_ref() == Some(LogicalType::Float16).as_ref()
- && f16::from_le_bytes(val.as_bytes().try_into().unwrap()) ==
f16::NEG_ZERO =>
- {
-
T::try_from_le_slice(&f16::to_le_bytes(f16::from_f32(replace))).unwrap()
- }
- _ => val.clone(),
- }
+ Some((min.clone(), max.clone(), nan_count))
}
/// Creates a bloom filter sized for the column's configured NDV, returning
the filter
diff --git a/parquet/src/column/writer/mod.rs b/parquet/src/column/writer/mod.rs
index aa9cef16c5..d0a092a9c5 100644
--- a/parquet/src/column/writer/mod.rs
+++ b/parquet/src/column/writer/mod.rs
@@ -23,12 +23,13 @@ use half::f16;
use crate::bloom_filter::Sbbf;
use crate::file::page_index::column_index::ColumnIndexMetaData;
use crate::file::page_index::offset_index::OffsetIndexMetaData;
+use std::cmp::Ordering;
use std::collections::{BTreeSet, VecDeque};
use std::str;
use crate::basic::{
- BoundaryOrder, Compression, ConvertedType, Encoding, EncodingMask,
IntType, LogicalType,
- PageType, Type,
+ BoundaryOrder, Compression, ConvertedType, Encoding, EncodingMask,
LogicalType, PageType,
+ SortOrder, Type,
};
use crate::column::page::{CompressedPage, Page, PageWriteSpec, PageWriter};
use crate::column::writer::encoder::{ColumnValueEncoder,
ColumnValueEncoderImpl, ColumnValues};
@@ -47,7 +48,7 @@ use crate::file::properties::{
EnabledStatistics, WriterProperties, WriterPropertiesPtr, WriterVersion,
};
use crate::file::statistics::{Statistics, ValueStatistics};
-use crate::schema::types::{ColumnDescPtr, ColumnDescriptor};
+use crate::schema::types::{BasicTypeInfo, ColumnDescPtr, ColumnDescriptor};
mod byte_budget_chunker;
pub(crate) mod encoder;
@@ -246,6 +247,7 @@ struct PageMetrics {
num_buffered_values: u32,
num_buffered_rows: u32,
num_page_nulls: u64,
+ num_page_nans: Option<u64>,
repetition_level_histogram: Option<LevelHistogram>,
definition_level_histogram: Option<LevelHistogram>,
}
@@ -273,6 +275,7 @@ impl PageMetrics {
self.num_buffered_values = 0;
self.num_buffered_rows = 0;
self.num_page_nulls = 0;
+ self.num_page_nans = None;
self.repetition_level_histogram
.as_mut()
.map(LevelHistogram::reset);
@@ -295,6 +298,7 @@ struct ColumnMetrics<T: Default> {
min_column_value: Option<T>,
max_column_value: Option<T>,
num_column_nulls: u64,
+ num_column_nans: Option<u64>,
column_distinct_count: Option<u64>,
variable_length_bytes: Option<i64>,
repetition_level_histogram: Option<LevelHistogram>,
@@ -1058,6 +1062,27 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
Ok(())
}
+ // For float columns, always provide Some(n), even if n is 0
+ // For non-float columns, always provide None
+ fn get_nan_count<T: ParquetValueType>(&self) -> Option<i64> {
+ let nan_count = || {
+ let nan_count = self.page_metrics.num_page_nans.unwrap_or(0);
+ match i64::try_from(nan_count) {
+ Ok(count) => Some(count),
+ _ => Some(i64::MAX),
+ }
+ };
+ match T::PHYSICAL_TYPE {
+ Type::FLOAT | Type::DOUBLE => nan_count(),
+ Type::FIXED_LEN_BYTE_ARRAY
+ if matches!(self.descr.logical_type_ref(),
Some(LogicalType::Float16)) =>
+ {
+ nan_count()
+ }
+ _ => None,
+ }
+ }
+
/// Update the column index and offset index when adding the data page
fn update_column_offset_index(
&mut self,
@@ -1075,6 +1100,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
vec![],
vec![],
self.page_metrics.num_page_nulls as i64,
+ self.get_nan_count::<E::T>(),
);
} else if self.column_index_builder.valid() {
// from page statistics
@@ -1088,10 +1114,11 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
let new_min = stat.min_opt().unwrap();
let new_max = stat.max_opt().unwrap();
if let Some((last_min, last_max)) =
&self.last_non_null_data_page_min_max {
+ let basic_info = self.descr.get_basic_info();
if self.data_page_boundary_ascending {
// If last min/max are greater than new min/max
then not ascending anymore
- let not_ascending = compare_greater(&self.descr,
last_min, new_min)
- || compare_greater(&self.descr, last_max,
new_max);
+ let not_ascending = compare_greater(basic_info,
last_min, new_min)
+ || compare_greater(basic_info, last_max,
new_max);
if not_ascending {
self.data_page_boundary_ascending = false;
}
@@ -1099,8 +1126,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
if self.data_page_boundary_descending {
// If new min/max are greater than last min/max
then not descending anymore
- let not_descending = compare_greater(&self.descr,
new_min, last_min)
- || compare_greater(&self.descr, new_max,
last_max);
+ let not_descending = compare_greater(basic_info,
new_min, last_min)
+ || compare_greater(basic_info, new_max,
last_max);
if not_descending {
self.data_page_boundary_descending = false;
}
@@ -1122,6 +1149,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
)
.0,
self.page_metrics.num_page_nulls as i64,
+ self.get_nan_count::<E::T>(),
);
} else {
self.column_index_builder.append(
@@ -1129,6 +1157,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
stat.min_bytes_opt().unwrap().to_vec(),
stat.max_bytes_opt().unwrap().to_vec(),
self.page_metrics.num_page_nulls as i64,
+ self.get_nan_count::<E::T>(),
);
}
}
@@ -1297,6 +1326,11 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
self.column_metrics.num_column_nulls +=
self.page_metrics.num_page_nulls;
+ if let Some(nan_count) = values_data.nan_count {
+ *self.column_metrics.num_column_nans.get_or_insert(0) += nan_count;
+ self.page_metrics.num_page_nans = Some(nan_count);
+ }
+
let page_statistics = match (values_data.min_value,
values_data.max_value) {
(Some(min), Some(max)) => {
// Update chunk level statistics
@@ -1310,7 +1344,8 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
None,
Some(self.page_metrics.num_page_nulls),
false,
- ),
+ )
+ .with_nan_count(values_data.nan_count),
)
}
_ => None,
@@ -1494,6 +1529,7 @@ impl<'a, E: ColumnValueEncoder> GenericColumnWriter<'a,
E> {
Some(self.column_metrics.num_column_nulls),
false,
)
+ .with_nan_count(self.column_metrics.num_column_nans)
.with_backwards_compatible_min_max(backwards_compatible_min_max)
.into();
@@ -1642,80 +1678,109 @@ impl<'a, E: ColumnValueEncoder>
GenericColumnWriter<'a, E> {
}
fn update_min<T: ParquetValueType>(descr: &ColumnDescriptor, val: &T, min:
&mut Option<T>) {
- update_stat::<T, _>(descr, val, min, |cur| compare_greater(descr, cur,
val))
+ match min {
+ None => *min = Some(val.clone()),
+ Some(min) => {
+ let basic_type_info = descr.get_basic_info();
+ let is_min_nan = is_nan(basic_type_info, min);
+ let is_val_nan = is_nan(basic_type_info, val);
+ match (is_min_nan, is_val_nan) {
+ // current min is not NaN, but incoming is NaN: skip
+ (false, true) => {}
+ // current min is NaN, but incoming is not: assign val to min
+ (true, false) => *min = val.clone(),
+ // both NaN or non-NaN, safe to call update_stat()
+ _ => {
+ update_stat::<T, _>(val, min, |cur|
compare_greater(basic_type_info, cur, val))
+ }
+ }
+ }
+ }
}
fn update_max<T: ParquetValueType>(descr: &ColumnDescriptor, val: &T, max:
&mut Option<T>) {
- update_stat::<T, _>(descr, val, max, |cur| compare_greater(descr, val,
cur))
+ match max {
+ None => *max = Some(val.clone()),
+ Some(max) => {
+ let basic_type_info = descr.get_basic_info();
+ let is_max_nan = is_nan(basic_type_info, max);
+ let is_val_nan = is_nan(basic_type_info, val);
+ match (is_max_nan, is_val_nan) {
+ // current max is not NaN, but incoming is NaN: skip
+ (false, true) => {}
+ // current max is NaN, but incoming is not: assign val to max
+ (true, false) => *max = val.clone(),
+ // both NaN or non-NaN, safe to call update_stat()
+ _ => {
+ update_stat::<T, _>(val, max, |cur|
compare_greater(basic_type_info, val, cur))
+ }
+ }
+ }
+ }
}
#[inline]
#[allow(clippy::eq_op)]
-fn is_nan<T: ParquetValueType>(descr: &ColumnDescriptor, val: &T) -> bool {
+fn is_nan<T: ParquetValueType>(basic_type_info: &BasicTypeInfo, val: &T) ->
bool {
match T::PHYSICAL_TYPE {
Type::FLOAT | Type::DOUBLE => val != val,
- Type::FIXED_LEN_BYTE_ARRAY if descr.logical_type_ref() ==
Some(&LogicalType::Float16) => {
+ Type::FIXED_LEN_BYTE_ARRAY
+ if matches!(basic_type_info.sort_order(), SortOrder::TOTAL_ORDER)
=>
+ {
+ // taken from f16 impl, but skips creating f16. just compare the
bits as u16.
let val = val.as_bytes();
- let val = f16::from_le_bytes([val[0], val[1]]);
- val.is_nan()
+ // Float16 is stored little endian
+ let uval = (val[1] as u16) << 8 | val[0] as u16;
+ uval & 0x7FFFu16 > 0x7C00u16
}
_ => false,
}
}
-/// Perform a conditional update of `cur`, skipping any NaN values
+/// Perform a conditional update of `cur`
///
-/// If `cur` is `None`, sets `cur` to `Some(val)`, otherwise calls
`should_update` with
-/// the value of `cur`, and updates `cur` to `Some(val)` if it returns `true`
-fn update_stat<T: ParquetValueType, F>(
- descr: &ColumnDescriptor,
- val: &T,
- cur: &mut Option<T>,
- should_update: F,
-) where
+/// Calls `should_update` with the value of `cur`, and updates `cur` to
`Some(val)` if it
+/// returns `true`. `cur` must not be `None` or this will panic.
+fn update_stat<T: ParquetValueType, F>(val: &T, cur: &mut T, should_update: F)
+where
F: Fn(&T) -> bool,
{
- if is_nan(descr, val) {
- return;
- }
-
- if cur.as_ref().is_none_or(should_update) {
- *cur = Some(val.clone());
+ if should_update(cur) {
+ *cur = val.clone();
}
}
/// Evaluate `a > b` according to underlying logical type.
-fn compare_greater<T: ParquetValueType>(descr: &ColumnDescriptor, a: &T, b:
&T) -> bool {
+fn compare_greater<T: ParquetValueType>(basic_type_info: &BasicTypeInfo, a:
&T, b: &T) -> bool {
match T::PHYSICAL_TYPE {
- Type::INT32 | Type::INT64 => {
- if let Some(LogicalType::Integer(IntType {
- is_signed: false, ..
- })) = descr.logical_type_ref()
- {
- // need to compare unsigned
- return compare_greater_unsigned_int(a, b);
- }
-
- match descr.converted_type() {
- ConvertedType::UINT_8
- | ConvertedType::UINT_16
- | ConvertedType::UINT_32
- | ConvertedType::UINT_64 => {
- return compare_greater_unsigned_int(a, b);
- }
- _ => {}
- };
+ Type::FLOAT => {
+ let a = f32::from_le_bytes(a.as_bytes().try_into().unwrap());
+ let b = f32::from_le_bytes(b.as_bytes().try_into().unwrap());
+ return a.total_cmp(&b) == Ordering::Greater;
}
- Type::FIXED_LEN_BYTE_ARRAY | Type::BYTE_ARRAY => {
- if let Some(LogicalType::Decimal(_)) = descr.logical_type_ref() {
- return compare_greater_byte_array_decimals(a.as_bytes(),
b.as_bytes());
- }
- if let ConvertedType::DECIMAL = descr.converted_type() {
- return compare_greater_byte_array_decimals(a.as_bytes(),
b.as_bytes());
- }
- if let Some(LogicalType::Float16) = descr.logical_type_ref() {
- return compare_greater_f16(a.as_bytes(), b.as_bytes());
- }
+ Type::DOUBLE => {
+ let a = f64::from_le_bytes(a.as_bytes().try_into().unwrap());
+ let b = f64::from_le_bytes(b.as_bytes().try_into().unwrap());
+ return a.total_cmp(&b) == Ordering::Greater;
+ }
+ Type::INT32 | Type::INT64
+ if matches!(basic_type_info.sort_order(), SortOrder::UNSIGNED) =>
+ {
+ return compare_greater_unsigned_int(a, b);
+ }
+ Type::FIXED_LEN_BYTE_ARRAY
+ if matches!(basic_type_info.sort_order(), SortOrder::TOTAL_ORDER)
=>
+ {
+ return compare_greater_f16(a.as_bytes(), b.as_bytes());
+ }
+ Type::FIXED_LEN_BYTE_ARRAY | Type::BYTE_ARRAY
+ if matches!(basic_type_info.converted_type(),
ConvertedType::DECIMAL)
+ || matches!(
+ basic_type_info.logical_type_ref(),
+ Some(LogicalType::Decimal(_))
+ ) =>
+ {
+ return compare_greater_byte_array_decimals(a.as_bytes(),
b.as_bytes());
}
_ => {}
@@ -1764,7 +1829,7 @@ fn compare_greater_unsigned_int<T: ParquetValueType>(a:
&T, b: &T) -> bool {
fn compare_greater_f16(a: &[u8], b: &[u8]) -> bool {
let a = f16::from_le_bytes(a.try_into().unwrap());
let b = f16::from_le_bytes(b.try_into().unwrap());
- a > b
+ a.total_cmp(&b) == Ordering::Greater
}
/// Signed comparison of bytes arrays
@@ -3207,7 +3272,7 @@ mod tests {
#[test]
fn test_float_statistics() {
let stats = statistics_roundtrip::<FloatType>(&[-1.0, 3.0, -2.0, 2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-2.0);
assert_eq!(stats.max_opt().unwrap(), &3.0);
@@ -3219,7 +3284,7 @@ mod tests {
#[test]
fn test_double_statistics() {
let stats = statistics_roundtrip::<DoubleType>(&[-1.0, 3.0, -2.0,
2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-2.0);
assert_eq!(stats.max_opt().unwrap(), &3.0);
@@ -3264,6 +3329,137 @@ mod tests {
}
}
+ #[test]
+ fn test_ieee754_total_order_float() {
+ // Test IEEE 754 total order for f32
+ // Order should be: -NaN < -Inf < -1.0 < -0.0 < +0.0 < 1.0 < +Inf <
+NaN
+ let neg_nan = f32::from_bits(0xffc00000); // a NaN with the sign bit
set
+ let neg_inf = f32::NEG_INFINITY;
+ let neg_one = -1.0_f32;
+ let neg_zero = -0.0_f32;
+ let pos_zero = 0.0_f32;
+ let pos_one = 1.0_f32;
+ let pos_inf = f32::INFINITY;
+ let pos_nan = f32::from_bits(0x7fc00000); // a NaN with the sign bit
unset
+
+ let values = vec![
+ pos_nan, neg_zero, pos_inf, neg_one, neg_nan, pos_one, neg_inf,
pos_zero,
+ ];
+
+ let stats = statistics_roundtrip::<FloatType>(&values);
+ if let Statistics::Float(stats) = stats {
+ // With IEEE 754 total order, min should be -NaN, max should be
+NaN
+ // But since we filter out NaN values, min should be -Inf, max
should be +Inf
+ assert_eq!(stats.min_opt().unwrap(), &neg_inf);
+ assert_eq!(stats.max_opt().unwrap(), &pos_inf);
+ assert_eq!(stats.nan_count_opt(), Some(2)); // neg_nan and pos_nan
+ } else {
+ panic!("Expected float statistics");
+ }
+ }
+
+ #[test]
+ fn test_ieee754_total_order_float_only_nan() {
+ // Test IEEE 754 total order for various NaN representations
+ // They should be ordered by the significand
+ let neg_nan1 = f32::from_bits(0xffc00000); // sign bit set,
significand x400000
+ let neg_nan2 = f32::from_bits(0xffc00001); // sign bit set,
significand x400001
+ let neg_nan3 = f32::from_bits(0xffc00002); // sign bit set,
significand x400002
+ let pos_nan1 = f32::from_bits(0x7fc00000); // sign bit unset,
significand x400000
+ let pos_nan2 = f32::from_bits(0x7fc00001); // sign bit unset,
significand x400001
+ let pos_nan3 = f32::from_bits(0x7fc00002); // sign bit unset,
significand x400002
+
+ let values = vec![neg_nan1, neg_nan2, neg_nan3, pos_nan1, pos_nan2,
pos_nan3];
+
+ let stats = statistics_roundtrip::<FloatType>(&values);
+ if let Statistics::Float(stats) = stats {
+ // With IEEE 754 total order, min should be `neg_nan3`, max
`pos_nan3`
+ assert_eq!(
+ stats.min_opt().unwrap().total_cmp(&neg_nan3),
+ Ordering::Equal
+ );
+ assert_eq!(
+ stats.max_opt().unwrap().total_cmp(&pos_nan3),
+ Ordering::Equal
+ );
+ assert_eq!(stats.nan_count_opt(), Some(6));
+ } else {
+ panic!("Expected float statistics");
+ }
+ }
+
+ #[test]
+ fn test_ieee754_total_order_double() {
+ // Test IEEE 754 total order for f64
+ let neg_nan = f64::from_bits(0xfff8000000000000);
+ let neg_inf = f64::NEG_INFINITY;
+ let neg_one = -1.0_f64;
+ let neg_zero = -0.0_f64;
+ let pos_zero = 0.0_f64;
+ let pos_one = 1.0_f64;
+ let pos_inf = f64::INFINITY;
+ let pos_nan = f64::from_bits(0x7ff8000000000000);
+
+ let values = vec![
+ pos_nan, neg_zero, pos_inf, neg_one, neg_nan, pos_one, neg_inf,
pos_zero,
+ ];
+
+ let stats = statistics_roundtrip::<DoubleType>(&values);
+ if let Statistics::Double(stats) = stats {
+ // With IEEE 754 total order, and NaN filtering
+ assert_eq!(stats.min_opt().unwrap(), &neg_inf);
+ assert_eq!(stats.max_opt().unwrap(), &pos_inf);
+ assert_eq!(stats.nan_count_opt(), Some(2));
+ } else {
+ panic!("Expected double statistics");
+ }
+ }
+
+ #[test]
+ fn test_ieee754_total_order_double_only_nan() {
+ // Test IEEE 754 total order for various NaN representations
+ // They should be ordered by the significand
+ let neg_nan1 = f64::from_bits(0xfff8000000000000);
+ let neg_nan2 = f64::from_bits(0xfff8000000000001);
+ let neg_nan3 = f64::from_bits(0xfff8000000000002);
+ let pos_nan1 = f64::from_bits(0x7ff8000000000000);
+ let pos_nan2 = f64::from_bits(0x7ff8000000000001);
+ let pos_nan3 = f64::from_bits(0x7ff8000000000002);
+
+ let values = vec![neg_nan1, neg_nan2, neg_nan3, pos_nan1, pos_nan2,
pos_nan3];
+
+ let stats = statistics_roundtrip::<DoubleType>(&values);
+ if let Statistics::Double(stats) = stats {
+ // With IEEE 754 total order, min should be `neg_nan3`, max
`pos_nan3`
+ assert_eq!(
+ stats.min_opt().unwrap().total_cmp(&neg_nan3),
+ Ordering::Equal
+ );
+ assert_eq!(
+ stats.max_opt().unwrap().total_cmp(&pos_nan3),
+ Ordering::Equal
+ );
+ assert_eq!(stats.nan_count_opt(), Some(6));
+ } else {
+ panic!("Expected float statistics");
+ }
+ }
+
+ #[test]
+ fn test_ieee754_total_order_zeros() {
+ // Test that -0.0 and +0.0 are handled correctly
+ let values = vec![-0.0_f32, 0.0_f32, -0.0_f32, 0.0_f32];
+
+ let stats = statistics_roundtrip::<FloatType>(&values);
+ if let Statistics::Float(stats) = stats {
+ // With IEEE 754 total order, -0.0 < +0.0
+ assert_eq!(stats.min_opt().unwrap().to_bits(),
(-0.0_f32).to_bits());
+ assert_eq!(stats.max_opt().unwrap().to_bits(), 0.0_f32.to_bits());
+ } else {
+ panic!("Expected float statistics");
+ }
+ }
+
#[test]
fn test_column_writer_check_float16_min_max() {
let input = [
@@ -3277,7 +3473,7 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(
stats.min_opt().unwrap(),
&ByteArray::from(-f16::from_f32(2.0))
@@ -3296,12 +3492,13 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
assert_eq!(
stats.max_opt().unwrap(),
&ByteArray::from(f16::ONE + f16::ONE)
);
+ assert_eq!(stats.nan_count_opt(), Some(1));
}
#[test]
@@ -3312,12 +3509,13 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
assert_eq!(
stats.max_opt().unwrap(),
&ByteArray::from(f16::ONE + f16::ONE)
);
+ assert_eq!(stats.nan_count_opt(), Some(1));
}
#[test]
@@ -3328,12 +3526,13 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ONE));
assert_eq!(
stats.max_opt().unwrap(),
&ByteArray::from(f16::ONE + f16::ONE)
);
+ assert_eq!(stats.nan_count_opt(), Some(1));
}
#[test]
@@ -3344,9 +3543,16 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.min_bytes_opt().is_none());
- assert!(stats.max_bytes_opt().is_none());
- assert!(stats.is_min_max_backwards_compatible());
+ assert_eq!(
+ stats.min_bytes_opt(),
+ Some(ByteArray::from(f16::NAN).as_bytes())
+ );
+ assert_eq!(
+ stats.max_bytes_opt(),
+ Some(ByteArray::from(f16::NAN).as_bytes())
+ );
+ assert!(!stats.is_min_max_backwards_compatible());
+ assert_eq!(stats.nan_count_opt(), Some(2));
}
#[test]
@@ -3357,8 +3563,8 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
- assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
+ assert!(!stats.is_min_max_backwards_compatible());
+ assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ZERO));
assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
}
@@ -3370,9 +3576,9 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
- assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
+ assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
}
#[test]
@@ -3383,8 +3589,8 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
- assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
+ assert!(!stats.is_min_max_backwards_compatible());
+ assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(f16::ZERO));
assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::PI));
}
@@ -3396,18 +3602,19 @@ mod tests {
.collect::<Vec<_>>();
let stats = float16_statistics_roundtrip(&input);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
assert_eq!(stats.min_opt().unwrap(), &ByteArray::from(-f16::PI));
- assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::ZERO));
+ assert_eq!(stats.max_opt().unwrap(), &ByteArray::from(f16::NEG_ZERO));
}
#[test]
fn test_float_statistics_nan_middle() {
let stats = statistics_roundtrip::<FloatType>(&[1.0, f32::NAN, 2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &1.0);
assert_eq!(stats.max_opt().unwrap(), &2.0);
+ assert_eq!(stats.nan_count_opt(), Some(1))
} else {
panic!("expecting Statistics::Float");
}
@@ -3416,10 +3623,11 @@ mod tests {
#[test]
fn test_float_statistics_nan_start() {
let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, 1.0, 2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &1.0);
assert_eq!(stats.max_opt().unwrap(), &2.0);
+ assert_eq!(stats.nan_count_opt(), Some(1))
} else {
panic!("expecting Statistics::Float");
}
@@ -3428,19 +3636,20 @@ mod tests {
#[test]
fn test_float_statistics_nan_only() {
let stats = statistics_roundtrip::<FloatType>(&[f32::NAN, f32::NAN]);
- assert!(stats.min_bytes_opt().is_none());
- assert!(stats.max_bytes_opt().is_none());
- assert!(stats.is_min_max_backwards_compatible());
+ assert_eq!(stats.min_bytes_opt(), Some(f32::NAN.as_bytes()));
+ assert_eq!(stats.max_bytes_opt(), Some(f32::NAN.as_bytes()));
+ assert_eq!(stats.nan_count_opt(), Some(2));
+ assert!(!stats.is_min_max_backwards_compatible());
assert!(matches!(stats, Statistics::Float(_)));
}
#[test]
fn test_float_statistics_zero_only() {
let stats = statistics_roundtrip::<FloatType>(&[0.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
- assert_eq!(stats.min_opt().unwrap(), &-0.0);
- assert!(stats.min_opt().unwrap().is_sign_negative());
+ assert_eq!(stats.min_opt().unwrap(), &0.0);
+ assert!(stats.min_opt().unwrap().is_sign_positive());
assert_eq!(stats.max_opt().unwrap(), &0.0);
assert!(stats.max_opt().unwrap().is_sign_positive());
} else {
@@ -3451,12 +3660,12 @@ mod tests {
#[test]
fn test_float_statistics_neg_zero_only() {
let stats = statistics_roundtrip::<FloatType>(&[-0.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-0.0);
assert!(stats.min_opt().unwrap().is_sign_negative());
- assert_eq!(stats.max_opt().unwrap(), &0.0);
- assert!(stats.max_opt().unwrap().is_sign_positive());
+ assert_eq!(stats.max_opt().unwrap(), &-0.0);
+ assert!(stats.max_opt().unwrap().is_sign_negative());
} else {
panic!("expecting Statistics::Float");
}
@@ -3465,10 +3674,10 @@ mod tests {
#[test]
fn test_float_statistics_zero_min() {
let stats = statistics_roundtrip::<FloatType>(&[0.0, 1.0, f32::NAN,
2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
- assert_eq!(stats.min_opt().unwrap(), &-0.0);
- assert!(stats.min_opt().unwrap().is_sign_negative());
+ assert_eq!(stats.min_opt().unwrap(), &0.0);
+ assert!(stats.min_opt().unwrap().is_sign_positive());
assert_eq!(stats.max_opt().unwrap(), &2.0);
} else {
panic!("expecting Statistics::Float");
@@ -3478,11 +3687,11 @@ mod tests {
#[test]
fn test_float_statistics_neg_zero_max() {
let stats = statistics_roundtrip::<FloatType>(&[-0.0, -1.0, f32::NAN,
-2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Float(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-2.0);
- assert_eq!(stats.max_opt().unwrap(), &0.0);
- assert!(stats.max_opt().unwrap().is_sign_positive());
+ assert_eq!(stats.max_opt().unwrap(), &-0.0);
+ assert!(stats.max_opt().unwrap().is_sign_negative());
} else {
panic!("expecting Statistics::Float");
}
@@ -3491,10 +3700,11 @@ mod tests {
#[test]
fn test_double_statistics_nan_middle() {
let stats = statistics_roundtrip::<DoubleType>(&[1.0, f64::NAN, 2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &1.0);
assert_eq!(stats.max_opt().unwrap(), &2.0);
+ assert_eq!(stats.nan_count_opt(), Some(1))
} else {
panic!("expecting Statistics::Double");
}
@@ -3503,10 +3713,11 @@ mod tests {
#[test]
fn test_double_statistics_nan_start() {
let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, 1.0, 2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &1.0);
assert_eq!(stats.max_opt().unwrap(), &2.0);
+ assert_eq!(stats.nan_count_opt(), Some(1))
} else {
panic!("expecting Statistics::Double");
}
@@ -3515,19 +3726,20 @@ mod tests {
#[test]
fn test_double_statistics_nan_only() {
let stats = statistics_roundtrip::<DoubleType>(&[f64::NAN, f64::NAN]);
- assert!(stats.min_bytes_opt().is_none());
- assert!(stats.max_bytes_opt().is_none());
+ assert_eq!(stats.min_bytes_opt(), Some(f64::NAN.as_bytes()));
+ assert_eq!(stats.max_bytes_opt(), Some(f64::NAN.as_bytes()));
+ assert_eq!(stats.nan_count_opt(), Some(2));
assert!(matches!(stats, Statistics::Double(_)));
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
}
#[test]
fn test_double_statistics_zero_only() {
let stats = statistics_roundtrip::<DoubleType>(&[0.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
- assert_eq!(stats.min_opt().unwrap(), &-0.0);
- assert!(stats.min_opt().unwrap().is_sign_negative());
+ assert_eq!(stats.min_opt().unwrap(), &0.0);
+ assert!(stats.min_opt().unwrap().is_sign_positive());
assert_eq!(stats.max_opt().unwrap(), &0.0);
assert!(stats.max_opt().unwrap().is_sign_positive());
} else {
@@ -3538,12 +3750,12 @@ mod tests {
#[test]
fn test_double_statistics_neg_zero_only() {
let stats = statistics_roundtrip::<DoubleType>(&[-0.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-0.0);
assert!(stats.min_opt().unwrap().is_sign_negative());
- assert_eq!(stats.max_opt().unwrap(), &0.0);
- assert!(stats.max_opt().unwrap().is_sign_positive());
+ assert_eq!(stats.max_opt().unwrap(), &-0.0);
+ assert!(stats.max_opt().unwrap().is_sign_negative());
} else {
panic!("expecting Statistics::Double");
}
@@ -3552,10 +3764,10 @@ mod tests {
#[test]
fn test_double_statistics_zero_min() {
let stats = statistics_roundtrip::<DoubleType>(&[0.0, 1.0, f64::NAN,
2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
- assert_eq!(stats.min_opt().unwrap(), &-0.0);
- assert!(stats.min_opt().unwrap().is_sign_negative());
+ assert_eq!(stats.min_opt().unwrap(), &0.0);
+ assert!(stats.min_opt().unwrap().is_sign_positive());
assert_eq!(stats.max_opt().unwrap(), &2.0);
} else {
panic!("expecting Statistics::Double");
@@ -3565,11 +3777,11 @@ mod tests {
#[test]
fn test_double_statistics_neg_zero_max() {
let stats = statistics_roundtrip::<DoubleType>(&[-0.0, -1.0, f64::NAN,
-2.0]);
- assert!(stats.is_min_max_backwards_compatible());
+ assert!(!stats.is_min_max_backwards_compatible());
if let Statistics::Double(stats) = stats {
assert_eq!(stats.min_opt().unwrap(), &-2.0);
- assert_eq!(stats.max_opt().unwrap(), &0.0);
- assert!(stats.max_opt().unwrap().is_sign_positive());
+ assert_eq!(stats.max_opt().unwrap(), &-0.0);
+ assert!(stats.max_opt().unwrap().is_sign_negative());
} else {
panic!("expecting Statistics::Double");
}
diff --git a/parquet/src/file/metadata/mod.rs b/parquet/src/file/metadata/mod.rs
index 646438d2e9..cf0eb6d27f 100644
--- a/parquet/src/file/metadata/mod.rs
+++ b/parquet/src/file/metadata/mod.rs
@@ -1458,6 +1458,7 @@ pub struct ColumnIndexBuilder {
min_values: Vec<Vec<u8>>,
max_values: Vec<Vec<u8>>,
null_counts: Vec<i64>,
+ nan_counts: Vec<Option<i64>>,
boundary_order: BoundaryOrder,
/// contains the concatenation of the histograms of all pages
repetition_level_histograms: Option<Vec<i64>>,
@@ -1482,6 +1483,7 @@ impl ColumnIndexBuilder {
min_values: Vec::new(),
max_values: Vec::new(),
null_counts: Vec::new(),
+ nan_counts: Vec::new(),
boundary_order: BoundaryOrder::UNORDERED,
repetition_level_histograms: None,
definition_level_histograms: None,
@@ -1490,17 +1492,24 @@ impl ColumnIndexBuilder {
}
/// Append statistics for the next page
+ ///
+ /// For floating-point columns (FLOAT, DOUBLE, or FLOAT16), `nan_count`
must always
+ /// be `Some(n)`, even if n is 0. For non-floating-point columns,
`nan_count` must
+ /// always be `None`. This requirement ensures correct serialization
according to
+ /// the Parquet specification.
pub fn append(
&mut self,
null_page: bool,
min_value: Vec<u8>,
max_value: Vec<u8>,
null_count: i64,
+ nan_count: Option<i64>,
) {
self.null_pages.push(null_page);
self.min_values.push(min_value);
self.max_values.push(max_value);
self.null_counts.push(null_count);
+ self.nan_counts.push(nan_count);
}
/// Append the given page-level histograms to the [`ColumnIndex`]
histograms.
@@ -1548,51 +1557,79 @@ impl ColumnIndexBuilder {
pub fn build(self) -> Result<ColumnIndexMetaData> {
Ok(match self.column_type {
Type::BOOLEAN => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(false)?;
ColumnIndexMetaData::BOOLEAN(index)
}
Type::INT32 => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(false)?;
ColumnIndexMetaData::INT32(index)
}
Type::INT64 => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(false)?;
ColumnIndexMetaData::INT64(index)
}
Type::INT96 => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(false)?;
ColumnIndexMetaData::INT96(index)
}
Type::FLOAT => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(true)?;
ColumnIndexMetaData::FLOAT(index)
}
Type::DOUBLE => {
- let index = self.build_page_index()?;
+ let index = self.build_page_index(true)?;
ColumnIndexMetaData::DOUBLE(index)
}
Type::BYTE_ARRAY => {
- let index = self.build_byte_array_index()?;
+ let index = self.build_byte_array_index(false)?;
ColumnIndexMetaData::BYTE_ARRAY(index)
}
Type::FIXED_LEN_BYTE_ARRAY => {
- let index = self.build_byte_array_index()?;
+ let index = self.build_byte_array_index(true)?;
ColumnIndexMetaData::FIXED_LEN_BYTE_ARRAY(index)
}
})
}
- fn build_page_index<T>(self) -> Result<PrimitiveColumnIndex<T>>
+ fn build_nan_counts(nan_counts: &[Option<i64>]) -> Option<Vec<i64>> {
+ let has_some = nan_counts.iter().any(|x| x.is_some());
+ let has_none = nan_counts.iter().any(|x| x.is_none());
+
+ if has_some && !has_none {
+ Some(nan_counts.iter().map(|x| x.unwrap()).collect())
+ } else if !has_some && has_none {
+ None
+ } else {
+ debug_assert!(
+ false,
+ "Mixed Some/None in nan_counts - caller should provide
consistent values"
+ );
+ Some(nan_counts.iter().map(|x| x.unwrap_or(0)).collect())
+ }
+ }
+
+ fn build_page_index<T>(self, may_have_nan: bool) ->
Result<PrimitiveColumnIndex<T>>
where
T: ParquetValueType,
{
let min_values: Vec<&[u8]> = self.min_values.iter().map(|v|
v.as_slice()).collect();
let max_values: Vec<&[u8]> = self.max_values.iter().map(|v|
v.as_slice()).collect();
+ // Parquet spec requires nan_counts to be either present for all pages
or absent entirely.
+ // Callers must ensure consistency:
+ // - For floating-point columns: all pages must have Some(n)
+ // - For non-floating-point columns: all pages must have None
+ let nan_counts = if may_have_nan && !self.nan_counts.is_empty() {
+ Self::build_nan_counts(&self.nan_counts)
+ } else {
+ None
+ };
+
PrimitiveColumnIndex::try_new(
self.null_pages,
self.boundary_order,
Some(self.null_counts),
+ nan_counts,
self.repetition_level_histograms,
self.definition_level_histograms,
min_values,
@@ -1600,14 +1637,25 @@ impl ColumnIndexBuilder {
)
}
- fn build_byte_array_index(self) -> Result<ByteArrayColumnIndex> {
+ fn build_byte_array_index(self, may_have_nan: bool) ->
Result<ByteArrayColumnIndex> {
let min_values: Vec<&[u8]> = self.min_values.iter().map(|v|
v.as_slice()).collect();
let max_values: Vec<&[u8]> = self.max_values.iter().map(|v|
v.as_slice()).collect();
+ // Parquet spec requires nan_counts to be either present for all pages
or absent entirely.
+ // Callers must ensure consistency:
+ // - For floating-point columns: all pages must have Some(n)
+ // - For non-floating-point columns: all pages must have None
+ let nan_counts = if may_have_nan && !self.nan_counts.is_empty() {
+ Self::build_nan_counts(&self.nan_counts)
+ } else {
+ None
+ };
+
ByteArrayColumnIndex::try_new(
self.null_pages,
self.boundary_order,
Some(self.null_counts),
+ nan_counts,
self.repetition_level_histograms,
self.definition_level_histograms,
min_values,
@@ -2047,14 +2095,14 @@ mod tests {
.build();
#[cfg(not(feature = "encryption"))]
- let base_expected_size = 2734;
+ let base_expected_size = 2798;
#[cfg(feature = "encryption")]
- let base_expected_size = 2902;
+ let base_expected_size = 2966;
assert_eq!(parquet_meta.memory_size(), base_expected_size);
let mut column_index = ColumnIndexBuilder::new(Type::BOOLEAN);
- column_index.append(false, vec![1u8], vec![2u8, 3u8], 4);
+ column_index.append(false, vec![1u8], vec![2u8, 3u8], 4, None);
let column_index = column_index.build().unwrap();
let native_index = match column_index {
ColumnIndexMetaData::BOOLEAN(index) => index,
@@ -2078,9 +2126,9 @@ mod tests {
.build();
#[cfg(not(feature = "encryption"))]
- let bigger_expected_size = 3160;
+ let bigger_expected_size = 3248;
#[cfg(feature = "encryption")]
- let bigger_expected_size = 3328;
+ let bigger_expected_size = 3416;
// more set fields means more memory usage
assert!(bigger_expected_size > base_expected_size);
@@ -2127,7 +2175,7 @@ mod tests {
.set_row_groups(row_group_meta.clone())
.build();
- let base_expected_size = 2042;
+ let base_expected_size = 2074;
assert_eq!(parquet_meta_data.memory_size(), base_expected_size);
let footer_key = "0123456789012345".as_bytes();
@@ -2153,7 +2201,7 @@ mod tests {
.set_file_decryptor(Some(decryptor))
.build();
- let expected_size_with_decryptor = 3056;
+ let expected_size_with_decryptor = 3088;
assert!(expected_size_with_decryptor > base_expected_size);
assert_eq!(
diff --git a/parquet/src/file/metadata/thrift/mod.rs
b/parquet/src/file/metadata/thrift/mod.rs
index d5a0112a5e..29b7dffc07 100644
--- a/parquet/src/file/metadata/thrift/mod.rs
+++ b/parquet/src/file/metadata/thrift/mod.rs
@@ -113,6 +113,7 @@ struct Statistics<'a> {
6: optional binary<'a> min_value;
7: optional bool is_max_value_exact;
8: optional bool is_min_value_exact;
+ 9: optional i64 nan_count;
}
);
@@ -207,6 +208,19 @@ fn convert_stats(
.transpose()?;
// Generic distinct count (count of distinct values occurring)
let distinct_count = stats.distinct_count.map(|value| value as
u64);
+ // Generic nan count for floating point types
+ let nan_count = stats
+ .nan_count
+ .map(|nan_count| {
+ if nan_count < 0 {
+ return Err(general_err!(
+ "Statistics NaN count is negative {}",
+ nan_count
+ ));
+ }
+ Ok(nan_count as u64)
+ })
+ .transpose()?;
// Whether or not statistics use deprecated min/max fields.
let old_format = stats.min_value.is_none() &&
stats.max_value.is_none();
// Generic min value as bytes.
@@ -291,19 +305,25 @@ fn convert_stats(
};
FStatistics::int96(min, max, distinct_count, null_count,
old_format)
}
- Type::FLOAT => FStatistics::float(
- min.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
- max.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
- distinct_count,
- null_count,
- old_format,
+ Type::FLOAT => FStatistics::Float(
+ ValueStatistics::new(
+ min.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
+ max.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
+ distinct_count,
+ null_count,
+ old_format,
+ )
+ .with_nan_count(nan_count),
),
- Type::DOUBLE => FStatistics::double(
- min.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
- max.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
- distinct_count,
- null_count,
- old_format,
+ Type::DOUBLE => FStatistics::Double(
+ ValueStatistics::new(
+ min.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
+ max.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
+ distinct_count,
+ null_count,
+ old_format,
+ )
+ .with_nan_count(nan_count),
),
Type::BYTE_ARRAY => FStatistics::ByteArray(
ValueStatistics::new(
@@ -324,6 +344,7 @@ fn convert_stats(
null_count,
old_format,
)
+ .with_nan_count(nan_count)
.with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
.with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
),
@@ -884,6 +905,7 @@ pub(crate) fn parquet_metadata_from_bytes(
column.logical_type_ref(),
column.converted_type(),
column.physical_type(),
+ true,
);
cos[i] = ColumnOrder::TYPE_DEFINED_ORDER(sort_order);
}
@@ -1001,6 +1023,7 @@ pub(crate) struct PageStatistics {
6: optional binary min_value;
7: optional bool is_max_value_exact;
8: optional bool is_min_value_exact;
+ 9: optional i64 nan_count;
}
);
@@ -1898,6 +1921,7 @@ pub(crate) mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
let decoded_none = super::convert_stats(&column_descr,
Some(none_null_count))
.unwrap()
@@ -1913,6 +1937,7 @@ pub(crate) mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
let decoded_zero = super::convert_stats(&column_descr,
Some(zero_null_count))
.unwrap()
@@ -1943,6 +1968,7 @@ pub(crate) mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
let err = super::convert_stats(&column_descr,
Some(make_stats(Some(&invalid), None)))
diff --git a/parquet/src/file/metadata/writer.rs
b/parquet/src/file/metadata/writer.rs
index cd5d617f93..4b88077d58 100644
--- a/parquet/src/file/metadata/writer.rs
+++ b/parquet/src/file/metadata/writer.rs
@@ -205,21 +205,16 @@ impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
let offset_indexes = self.finalize_offset_indexes()?;
// We only include ColumnOrder for leaf nodes.
- // Currently only supported ColumnOrder is TypeDefinedOrder so we set
this
- // for all leaf nodes.
- // Even if the column has an undefined sort order, such as INTERVAL,
this
- // is still technically the defined TYPEORDER so it should still be
set.
let column_orders = self
.schema_descr
.columns()
.iter()
.map(|col| {
- let sort_order = ColumnOrder::sort_order_for_type(
+ ColumnOrder::column_order_for_type(
col.logical_type_ref(),
col.converted_type(),
col.physical_type(),
- );
- ColumnOrder::TYPE_DEFINED_ORDER(sort_order)
+ )
})
.collect();
diff --git a/parquet/src/file/page_index/column_index.rs
b/parquet/src/file/page_index/column_index.rs
index 2f90b3d8e5..63b94f28df 100644
--- a/parquet/src/file/page_index/column_index.rs
+++ b/parquet/src/file/page_index/column_index.rs
@@ -43,6 +43,7 @@ pub struct ColumnIndex {
pub(crate) null_counts: Option<Vec<i64>>,
pub(crate) repetition_level_histograms: Option<Vec<i64>>,
pub(crate) definition_level_histograms: Option<Vec<i64>>,
+ pub(crate) nan_counts: Option<Vec<i64>>,
}
impl ColumnIndex {
@@ -58,6 +59,13 @@ impl ColumnIndex {
self.null_counts.as_ref().map(|nc| nc[idx])
}
+ /// Returns the number of NaN values in the page indexed by `idx`
+ ///
+ /// Returns `None` if no NaN counts have been set in the index
+ pub fn nan_count(&self, idx: usize) -> Option<i64> {
+ self.nan_counts.as_ref().map(|nc| nc[idx])
+ }
+
/// Returns the repetition level histogram for the page indexed by `idx`
pub fn repetition_level_histogram(&self, idx: usize) -> Option<&[i64]> {
if let Some(rep_hists) = self.repetition_level_histograms.as_ref() {
@@ -95,10 +103,12 @@ pub struct PrimitiveColumnIndex<T> {
}
impl<T: ParquetValueType> PrimitiveColumnIndex<T> {
+ #[allow(clippy::too_many_arguments)]
pub(crate) fn try_new(
null_pages: Vec<bool>,
boundary_order: BoundaryOrder,
null_counts: Option<Vec<i64>>,
+ nan_counts: Option<Vec<i64>>,
repetition_level_histograms: Option<Vec<i64>>,
definition_level_histograms: Option<Vec<i64>>,
min_bytes: Vec<&[u8]>,
@@ -160,6 +170,7 @@ impl<T: ParquetValueType> PrimitiveColumnIndex<T> {
null_counts,
repetition_level_histograms,
definition_level_histograms,
+ nan_counts,
},
min_values,
max_values,
@@ -171,6 +182,7 @@ impl<T: ParquetValueType> PrimitiveColumnIndex<T> {
index.null_pages,
index.boundary_order,
index.null_counts,
+ index.nan_counts,
index.repetition_level_histograms,
index.definition_level_histograms,
index.min_values,
@@ -286,7 +298,11 @@ impl<T: ParquetValueType> WriteThrift for
PrimitiveColumnIndex<T> {
repetition_level_histograms.write_thrift_field(writer, 6,
last_field_id)?;
}
if let Some(definition_level_histograms) =
&self.definition_level_histograms {
- definition_level_histograms.write_thrift_field(writer, 7,
last_field_id)?;
+ last_field_id =
+ definition_level_histograms.write_thrift_field(writer, 7,
last_field_id)?;
+ }
+ if let Some(nan_counts) = &self.nan_counts {
+ nan_counts.write_thrift_field(writer, 8, last_field_id)?;
}
writer.write_struct_end()
}
@@ -304,10 +320,12 @@ pub struct ByteArrayColumnIndex {
}
impl ByteArrayColumnIndex {
+ #[allow(clippy::too_many_arguments)]
pub(crate) fn try_new(
null_pages: Vec<bool>,
boundary_order: BoundaryOrder,
null_counts: Option<Vec<i64>>,
+ nan_counts: Option<Vec<i64>>,
repetition_level_histograms: Option<Vec<i64>>,
definition_level_histograms: Option<Vec<i64>>,
min_values: Vec<&[u8]>,
@@ -383,6 +401,7 @@ impl ByteArrayColumnIndex {
null_pages,
boundary_order,
null_counts,
+ nan_counts,
repetition_level_histograms,
definition_level_histograms,
},
@@ -398,6 +417,7 @@ impl ByteArrayColumnIndex {
index.null_pages,
index.boundary_order,
index.null_counts,
+ index.nan_counts,
index.repetition_level_histograms,
index.definition_level_histograms,
index.min_values,
@@ -485,7 +505,11 @@ impl WriteThrift for ByteArrayColumnIndex {
repetition_level_histograms.write_thrift_field(writer, 6,
last_field_id)?;
}
if let Some(definition_level_histograms) =
&self.definition_level_histograms {
- definition_level_histograms.write_thrift_field(writer, 7,
last_field_id)?;
+ last_field_id =
+ definition_level_histograms.write_thrift_field(writer, 7,
last_field_id)?;
+ }
+ if let Some(nan_counts) = &self.nan_counts {
+ nan_counts.write_thrift_field(writer, 8, last_field_id)?;
}
writer.write_struct_end()
}
@@ -588,7 +612,7 @@ impl ColumnIndexMetaData {
/// Returns array of null counts, one per page.
///
- /// Returns `None` if now null counts have been set in the index
+ /// Returns `None` if no null counts have been set in the index
pub fn null_counts(&self) -> Option<&Vec<i64>> {
match self {
Self::NONE => None,
@@ -603,6 +627,23 @@ impl ColumnIndexMetaData {
}
}
+ /// Returns array of NaN counts, one per page.
+ ///
+ /// Returns `None` if no NaN counts have been set in the index
+ pub fn nan_counts(&self) -> Option<&Vec<i64>> {
+ match self {
+ Self::NONE => None,
+ Self::BOOLEAN(index) => index.nan_counts.as_ref(),
+ Self::INT32(index) => index.nan_counts.as_ref(),
+ Self::INT64(index) => index.nan_counts.as_ref(),
+ Self::INT96(index) => index.nan_counts.as_ref(),
+ Self::FLOAT(index) => index.nan_counts.as_ref(),
+ Self::DOUBLE(index) => index.nan_counts.as_ref(),
+ Self::BYTE_ARRAY(index) => index.nan_counts.as_ref(),
+ Self::FIXED_LEN_BYTE_ARRAY(index) => index.nan_counts.as_ref(),
+ }
+ }
+
/// Returns the number of pages
pub fn num_pages(&self) -> u64 {
colidx_enum_func!(self, num_pages)
@@ -615,6 +656,13 @@ impl ColumnIndexMetaData {
colidx_enum_func!(self, null_count, idx)
}
+ /// Returns the number of NaN values in the page indexed by `idx`
+ ///
+ /// Returns `None` if no NaN counts have been set in the index
+ pub fn nan_count(&self, idx: usize) -> Option<i64> {
+ colidx_enum_func!(self, nan_count, idx)
+ }
+
/// Returns the repetition level histogram for the page indexed by `idx`
pub fn repetition_level_histogram(&self, idx: usize) -> Option<&[i64]> {
colidx_enum_func!(self, repetition_level_histogram, idx)
@@ -716,6 +764,7 @@ mod tests {
null_pages: vec![false],
boundary_order: BoundaryOrder::ASCENDING,
null_counts: Some(vec![0]),
+ nan_counts: None,
repetition_level_histograms: Some(vec![1, 2]),
definition_level_histograms: Some(vec![1, 2, 3]),
},
@@ -740,6 +789,7 @@ mod tests {
null_pages: vec![true],
boundary_order: BoundaryOrder::ASCENDING,
null_counts: Some(vec![1]),
+ nan_counts: None,
repetition_level_histograms: None,
definition_level_histograms: Some(vec![1, 0]),
},
@@ -767,6 +817,7 @@ mod tests {
&[], // this shouldn't be empty as null_pages[1] is false
],
null_counts: None,
+ nan_counts: None,
repetition_level_histograms: None,
definition_level_histograms: None,
boundary_order: BoundaryOrder::UNORDERED,
@@ -791,6 +842,7 @@ mod tests {
repetition_level_histograms: None,
definition_level_histograms: None,
boundary_order: BoundaryOrder::UNORDERED,
+ nan_counts: None,
};
// ColumnIndex arrays must align with the number of pages
(null_pages.len()).
diff --git a/parquet/src/file/page_index/index_reader.rs
b/parquet/src/file/page_index/index_reader.rs
index f0e40f7fdd..1c8c607554 100644
--- a/parquet/src/file/page_index/index_reader.rs
+++ b/parquet/src/file/page_index/index_reader.rs
@@ -66,6 +66,7 @@ pub(super) struct ThriftColumnIndex<'a> {
5: optional list<i64> null_counts
6: optional list<i64> repetition_level_histograms;
7: optional list<i64> definition_level_histograms;
+ 8: optional list<i64> nan_counts
}
);
diff --git a/parquet/src/file/statistics.rs b/parquet/src/file/statistics.rs
index 9682fd54b8..61edaa6129 100644
--- a/parquet/src/file/statistics.rs
+++ b/parquet/src/file/statistics.rs
@@ -139,6 +139,18 @@ pub(crate) fn from_thrift_page_stats(
.transpose()?;
// Generic distinct count (count of distinct values occurring)
let distinct_count = stats.distinct_count.map(|value| value as
u64);
+ // Generic nan count for floating point types
+ let nan_count = stats
+ .nan_count
+ .map(|nan_count| {
+ if nan_count < 0 {
+ return Err(ParquetError::General(format!(
+ "Statistics NaN count is negative {nan_count}",
+ )));
+ }
+ Ok(nan_count as u64)
+ })
+ .transpose()?;
// Whether or not statistics use deprecated min/max fields.
let old_format = stats.min_value.is_none() &&
stats.max_value.is_none();
// Generic min value as bytes.
@@ -230,19 +242,29 @@ pub(crate) fn from_thrift_page_stats(
};
Statistics::int96(min, max, distinct_count, null_count,
old_format)
}
- Type::FLOAT => Statistics::float(
- min.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
- max.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
- distinct_count,
- null_count,
- old_format,
+ Type::FLOAT => Statistics::Float(
+ ValueStatistics::new(
+ min.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
+ max.map(|data|
f32::from_le_bytes(data[..4].try_into().unwrap())),
+ distinct_count,
+ null_count,
+ old_format,
+ )
+ .with_nan_count(nan_count)
+
.with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
+
.with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
),
- Type::DOUBLE => Statistics::double(
- min.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
- max.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
- distinct_count,
- null_count,
- old_format,
+ Type::DOUBLE => Statistics::Double(
+ ValueStatistics::new(
+ min.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
+ max.map(|data|
f64::from_le_bytes(data[..8].try_into().unwrap())),
+ distinct_count,
+ null_count,
+ old_format,
+ )
+ .with_nan_count(nan_count)
+
.with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
+
.with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
),
Type::BYTE_ARRAY => Statistics::ByteArray(
ValueStatistics::new(
@@ -263,6 +285,12 @@ pub(crate) fn from_thrift_page_stats(
null_count,
old_format,
)
+ // Note: We set nan_count here even though we can't verify
if this is Float16.
+ // The spec says nan_count should only be set for Float16
logical type,
+ // but this function doesn't have access to logical type
information.
+ // Writers should only set nan_count for Float16, and
readers should
+ // handle this gracefully.
+ .with_nan_count(nan_count)
.with_max_is_exact(stats.is_max_value_exact.unwrap_or(false))
.with_min_is_exact(stats.is_min_value_exact.unwrap_or(false)),
),
@@ -288,6 +316,11 @@ pub(crate) fn page_stats_to_thrift(stats:
Option<&Statistics>) -> Option<PageSta
.distinct_count_opt()
.and_then(|value| i64::try_from(value).ok());
+ // record nan count if it can fit in i64
+ let nan_count = stats
+ .nan_count_opt()
+ .and_then(|value| i64::try_from(value).ok());
+
let mut thrift_stats = PageStatistics {
max: None,
min: None,
@@ -297,6 +330,7 @@ pub(crate) fn page_stats_to_thrift(stats:
Option<&Statistics>) -> Option<PageSta
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count,
};
// Get min/max if set.
@@ -448,6 +482,11 @@ impl Statistics {
statistics_enum_func![self, null_count_opt]
}
+ /// Returns NaN count for floating point types, if known.
+ pub fn nan_count_opt(&self) -> Option<u64> {
+ statistics_enum_func![self, nan_count_opt]
+ }
+
/// Returns `true` if the min value is set, and is an exact min value.
pub fn min_is_exact(&self) -> bool {
statistics_enum_func![self, min_is_exact]
@@ -511,6 +550,8 @@ pub struct ValueStatistics<T> {
// Distinct count could be omitted in some cases
distinct_count: Option<u64>,
null_count: Option<u64>,
+ // NaN count for floating point types
+ nan_count: Option<u64>,
// Whether or not the min or max values are exact, or truncated.
is_max_value_exact: bool,
@@ -541,6 +582,7 @@ impl<T> ValueStatistics<T> {
max,
distinct_count,
null_count,
+ nan_count: None,
is_min_max_deprecated,
is_min_max_backwards_compatible: is_min_max_deprecated,
}
@@ -580,6 +622,16 @@ impl<T> ValueStatistics<T> {
}
}
+ /// Returns NaN count for floating point types.
+ pub fn nan_count_opt(&self) -> Option<u64> {
+ self.nan_count
+ }
+
+ /// Set the NaN count for floating point types.
+ pub fn with_nan_count(self, nan_count: Option<u64>) -> Self {
+ Self { nan_count, ..self }
+ }
+
/// Returns min value of the statistics, if known.
pub fn min_opt(&self) -> Option<&T> {
self.min.as_ref()
@@ -698,6 +750,8 @@ impl<T: ParquetValueType> fmt::Debug for ValueStatistics<T>
{
#[cfg(test)]
mod tests {
+ use core::f32;
+
use super::*;
#[test]
@@ -729,6 +783,7 @@ mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
from_thrift_page_stats(Type::INT32, Some(thrift_stats)).unwrap();
@@ -1050,6 +1105,7 @@ mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
let err = from_thrift_page_stats(Type::BOOLEAN,
Some(tstatistics)).unwrap_err();
assert_eq!(
@@ -1103,6 +1159,7 @@ mod tests {
min_value: None,
is_max_value_exact: None,
is_min_value_exact: None,
+ nan_count: None,
};
let err = from_thrift_page_stats(Type::INT96,
Some(thrift_stats.clone())).unwrap_err();
@@ -1141,4 +1198,141 @@ mod tests {
_ => unreachable!(),
}
}
+
+ #[test]
+ fn test_nan_count_float() {
+ // Test NaN count for f32
+ let stats = Statistics::Float(
+ ValueStatistics::new(Some(1.0_f32), Some(5.0_f32), None, Some(0),
false)
+ .with_nan_count(Some(3)),
+ );
+
+ assert_eq!(stats.nan_count_opt(), Some(3));
+
+ // Verify round-trip through thrift
+ let thrift_stats = page_stats_to_thrift(Some(&stats)).unwrap();
+ assert_eq!(thrift_stats.nan_count, Some(3));
+
+ let round_tripped = from_thrift_page_stats(Type::FLOAT,
Some(thrift_stats))
+ .unwrap()
+ .unwrap();
+ assert_eq!(round_tripped.nan_count_opt(), Some(3));
+ }
+
+ #[test]
+ fn test_nan_count_double() {
+ // Test NaN count for f64
+ let stats = Statistics::Double(
+ ValueStatistics::new(Some(1.0_f64), Some(5.0_f64), None, Some(0),
false)
+ .with_nan_count(Some(5)),
+ );
+
+ assert_eq!(stats.nan_count_opt(), Some(5));
+
+ // Verify round-trip through thrift
+ let thrift_stats = page_stats_to_thrift(Some(&stats)).unwrap();
+ assert_eq!(thrift_stats.nan_count, Some(5));
+
+ let round_tripped = from_thrift_page_stats(Type::DOUBLE,
Some(thrift_stats))
+ .unwrap()
+ .unwrap();
+ assert_eq!(round_tripped.nan_count_opt(), Some(5));
+ }
+
+ #[test]
+ fn test_nan_count_none_for_non_float() {
+ // NaN count should not be set for non-floating point types
+ let stats = Statistics::int32(Some(1), Some(100), None, Some(0),
false);
+ assert_eq!(stats.nan_count_opt(), None);
+
+ let thrift_stats = page_stats_to_thrift(Some(&stats)).unwrap();
+ assert_eq!(thrift_stats.nan_count, None);
+ }
+
+ #[test]
+ fn test_nan_count_backwards_compatible() {
+ // Test that missing nan_count field is handled correctly
+ let thrift_stats = PageStatistics {
+ min: None,
+ max: None,
+ min_value: Some(vec![0, 0, 0, 0]), // 0.0_f32 in bytes
+ max_value: Some(vec![0, 0, 128, 63]), // 1.0_f32 in bytes
+ null_count: Some(0),
+ distinct_count: None,
+ nan_count: None, // Not set
+ is_min_value_exact: None,
+ is_max_value_exact: None,
+ };
+
+ let stats = from_thrift_page_stats(Type::FLOAT, Some(thrift_stats))
+ .unwrap()
+ .unwrap();
+
+ // nan_count should be None when not provided
+ assert_eq!(stats.nan_count_opt(), None);
+ }
+
+ #[test]
+ fn test_statistics_with_nan_min_max() {
+ // Test that when there are only NaN values, min/max are NaN
+ let stats = Statistics::Float(
+ ValueStatistics::new(
+ Some(f32::NAN), // min and max should have NaN values
+ Some(f32::NAN),
+ None,
+ Some(0),
+ false,
+ )
+ .with_nan_count(Some(10)), // All values are NaN
+ );
+
+ assert_eq!(stats.min_bytes_opt(), Some(f32::NAN.as_bytes()));
+ assert_eq!(stats.max_bytes_opt(), Some(f32::NAN.as_bytes()));
+ assert_eq!(stats.nan_count_opt(), Some(10));
+
+ // Verify serialization handles this case
+ let thrift_stats = page_stats_to_thrift(Some(&stats)).unwrap();
+ assert_eq!(thrift_stats.min_value, Some(f32::NAN.as_bytes().to_vec()));
+ assert_eq!(thrift_stats.max_value, Some(f32::NAN.as_bytes().to_vec()));
+ assert_eq!(thrift_stats.nan_count, Some(10));
+ }
+
+ #[test]
+ fn test_nan_count_too_large() {
+ // Test that nan_count larger than i64::MAX is not serialized
+ let stats = Statistics::Float(
+ ValueStatistics::new(Some(1.0_f32), Some(2.0_f32), None, Some(0),
false)
+ .with_nan_count(Some(u64::MAX)),
+ );
+
+ let thrift_stats = page_stats_to_thrift(Some(&stats)).unwrap();
+ // u64::MAX can't fit in i64, so it should be None
+ assert_eq!(thrift_stats.nan_count, None);
+ }
+
+ #[test]
+ fn test_nan_counts_in_column_index() {
+ // Test that nan_counts are properly collected in page index
+ use crate::file::metadata::ColumnIndexBuilder;
+
+ // Test for floating-point column - all pages must have Some(n)
+ let mut float_builder = ColumnIndexBuilder::new(Type::FLOAT);
+ float_builder.append(false, vec![0u8; 4], vec![255u8; 4], 0, Some(5));
+ float_builder.append(false, vec![0u8; 4], vec![255u8; 4], 2, Some(3));
+ float_builder.append(false, vec![0u8; 4], vec![255u8; 4], 0, Some(0));
// No NaN but still Some(0)
+
+ let float_column_index = float_builder.build().unwrap();
+ // Verify nan_counts field is properly set for float column
+ assert_eq!(float_column_index.nan_counts(), Some(&vec![5, 3, 0]));
+
+ // Test for non-floating-point column - all pages must have None
+ let mut int_builder = ColumnIndexBuilder::new(Type::INT32);
+ int_builder.append(false, vec![0u8; 4], vec![255u8; 4], 0, None);
+ int_builder.append(false, vec![0u8; 4], vec![255u8; 4], 2, None);
+ int_builder.append(false, vec![0u8; 4], vec![255u8; 4], 0, None);
+
+ let int_column_index = int_builder.build().unwrap();
+ // Verify nan_counts field is None for non-float column
+ assert_eq!(int_column_index.nan_counts(), None);
+ }
}
diff --git a/parquet/src/file/writer.rs b/parquet/src/file/writer.rs
index b62d8886cd..cb94483200 100644
--- a/parquet/src/file/writer.rs
+++ b/parquet/src/file/writer.rs
@@ -1303,6 +1303,16 @@ mod tests {
.build()
.unwrap(),
),
+ Arc::new(
+ types::Type::primitive_type_builder("col5",
Type::FLOAT)
+ .build()
+ .unwrap(),
+ ),
+ Arc::new(
+ types::Type::primitive_type_builder("col6",
Type::DOUBLE)
+ .build()
+ .unwrap(),
+ ),
])
.build()
.unwrap(),
@@ -1321,9 +1331,13 @@ mod tests {
// INTERVAL
ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNDEFINED),
// Float16
- ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED),
+ ColumnOrder::IEEE_754_TOTAL_ORDER,
// String
ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::UNSIGNED),
+ // FLOAT
+ ColumnOrder::IEEE_754_TOTAL_ORDER,
+ // DOUBLE
+ ColumnOrder::IEEE_754_TOTAL_ORDER,
];
let actual = reader.metadata().file_metadata().column_orders();
diff --git a/parquet/src/schema/types.rs b/parquet/src/schema/types.rs
index 1f9b8590fc..2b2c3f1bb0 100644
--- a/parquet/src/schema/types.rs
+++ b/parquet/src/schema/types.rs
@@ -314,12 +314,19 @@ impl<'a> PrimitiveTypeBuilder<'a> {
/// Creates a new `PrimitiveType` instance from the collected attributes.
/// Returns `Err` in case of any building conditions are not met.
pub fn build(self) -> Result<Type> {
+ let sort_order = ColumnOrder::column_order_for_type(
+ self.logical_type.as_ref(),
+ self.converted_type,
+ self.physical_type,
+ )
+ .sort_order();
let mut basic_info = BasicTypeInfo {
name: String::from(self.name),
repetition: Some(self.repetition),
converted_type: self.converted_type,
logical_type: self.logical_type.clone(),
id: self.id,
+ sort_order,
};
// Check length before logical type, since it is used for logical type
validation.
@@ -651,6 +658,7 @@ impl<'a> GroupTypeBuilder<'a> {
converted_type: self.converted_type,
logical_type: self.logical_type.clone(),
id: self.id,
+ sort_order: SortOrder::UNDEFINED,
};
// Populate the converted type if only the logical type is populated
if self.logical_type.is_some() && self.converted_type ==
ConvertedType::NONE {
@@ -672,6 +680,7 @@ pub struct BasicTypeInfo {
converted_type: ConvertedType,
logical_type: Option<LogicalType>,
id: Option<i32>,
+ sort_order: SortOrder,
}
impl HeapSize for BasicTypeInfo {
@@ -733,6 +742,11 @@ impl BasicTypeInfo {
assert!(self.id.is_some());
self.id.unwrap()
}
+
+ /// Returns [`SortOrder`] for the type.
+ pub fn sort_order(&self) -> SortOrder {
+ self.sort_order
+ }
}
// ----------------------------------------------------------------------
@@ -928,6 +942,11 @@ impl ColumnDescriptor {
self.primitive_type.clone()
}
+ /// Returns [`BasicTypeInfo`] information for this leaf column.
+ pub fn get_basic_info(&self) -> &BasicTypeInfo {
+ self.primitive_type.get_basic_info()
+ }
+
/// Returns column name.
pub fn name(&self) -> &str {
self.primitive_type.name()
@@ -994,13 +1013,24 @@ impl ColumnDescriptor {
}
}
- /// Returns the sort order for this column
+ /// Returns the sort order for this column as currently defined for the
logical or
+ /// physical type.
+ ///
+ /// Returns `SortOrder::UNDEFINED` for non-primitive types.
pub fn sort_order(&self) -> SortOrder {
- ColumnOrder::sort_order_for_type(
- self.logical_type_ref(),
- self.converted_type(),
- self.physical_type(),
- )
+ match self.primitive_type.as_ref() {
+ Type::PrimitiveType {
+ basic_info,
+ physical_type,
+ ..
+ } => ColumnOrder::column_order_for_type(
+ basic_info.logical_type_ref(),
+ basic_info.converted_type(),
+ *physical_type,
+ )
+ .sort_order(),
+ _ => SortOrder::UNDEFINED,
+ }
}
}
diff --git a/parquet/tests/arrow_reader/statistics.rs
b/parquet/tests/arrow_reader/statistics.rs
index bbb891dfd2..173361f56f 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -781,7 +781,7 @@ async fn test_float_16() {
expected_min: Arc::new(Float16Array::from(vec![
f16::from_f32(-5.),
f16::from_f32(-4.),
- f16::from_f32(-0.),
+ f16::from_f32(0.),
f16::from_f32(5.),
])),
// maxes are [-1, 0, 4, 9]
@@ -817,7 +817,7 @@ async fn test_float_32() {
Test {
reader: &reader,
// mins are [-5, -4, 0, 5]
- expected_min: Arc::new(Float32Array::from(vec![-5., -4., -0., 5.0])),
+ expected_min: Arc::new(Float32Array::from(vec![-5., -4., 0., 5.0])),
// maxes are [-1, 0, 4, 9]
expected_max: Arc::new(Float32Array::from(vec![-1., 0., 4., 9.])),
// nulls are [0, 0, 0, 0]
@@ -846,7 +846,7 @@ async fn test_float_64() {
Test {
reader: &reader,
// mins are [-5, -4, 0, 5]
- expected_min: Arc::new(Float64Array::from(vec![-5., -4., -0., 5.0])),
+ expected_min: Arc::new(Float64Array::from(vec![-5., -4., 0., 5.0])),
// maxes are [-1, 0, 4, 9]
expected_max: Arc::new(Float64Array::from(vec![-1., 0., 4., 9.])),
// nulls are [0, 0, 0, 0]
@@ -1897,7 +1897,7 @@ async fn test_float64() {
Test {
reader: &reader,
- expected_min: Arc::new(Float64Array::from(vec![-5.0, -4.0, -0.0,
5.0])),
+ expected_min: Arc::new(Float64Array::from(vec![-5.0, -4.0, 0.0, 5.0])),
expected_max: Arc::new(Float64Array::from(vec![-1.0, 0.0, 4.0, 9.0])),
expected_null_counts: UInt64Array::from(vec![0, 0, 0, 0]),
expected_row_counts: Some(UInt64Array::from(vec![5, 5, 5, 5])),
@@ -1925,7 +1925,7 @@ async fn test_float16() {
Test {
reader: &reader,
expected_min: Arc::new(Float16Array::from(
- vec![-5.0, -4.0, -0.0, 5.0]
+ vec![-5.0, -4.0, 0.0, 5.0]
.into_iter()
.map(f16::from_f32)
.collect::<Vec<_>>(),
diff --git a/parquet/tests/ieee754_nan_interop.rs
b/parquet/tests/ieee754_nan_interop.rs
new file mode 100644
index 0000000000..cc3326dae5
--- /dev/null
+++ b/parquet/tests/ieee754_nan_interop.rs
@@ -0,0 +1,448 @@
+// 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.
+
+//! Interoperability test for
https://github.com/apache/parquet-format/pull/514.
+//! Demonstrate reading NaN statstics and counts from a file generated with
+//! parquet-java, and show that on write we produce the same statistics.
+
+use bytes::Bytes;
+use core::f32;
+use half::f16;
+use std::{path::PathBuf, sync::Arc};
+
+use arrow::util::test_util::parquet_test_data;
+use arrow_array::{Array, Float16Array, Float32Array, Float64Array,
RecordBatch, UInt64Array};
+use arrow_schema::{DataType, Field, Schema};
+use parquet::{
+ arrow::{
+ ArrowWriter,
+ arrow_reader::{ArrowReaderBuilder, ArrowReaderOptions,
statistics::StatisticsConverter},
+ },
+ errors::Result,
+ file::{metadata::ParquetMetaData, properties::WriterProperties},
+ schema::types::SchemaDescriptor,
+};
+
+const NAN_COUNTS: [u64; 5] = [0, 4, 10, 0, 0];
+
+const FLOAT_NEG_NAN_SMALL: f32 = f32::from_bits(0xffffffff);
+const FLOAT_NEG_NAN_LARGE: f32 = f32::from_bits(0xfff00001);
+const FLOAT_NAN_SMALL: f32 = f32::from_bits(0x7fc00001);
+const FLOAT_NAN_LARGE: f32 = f32::from_bits(0x7fffffff);
+
+const FLOAT_MINS: [f32; 5] = [-2.0, -2.0, FLOAT_NEG_NAN_SMALL, 0.0, -5.0];
+const FLOAT_MAXS: [f32; 5] = [5.0, 3.0, FLOAT_NAN_LARGE, 5.0, -0.0];
+
+fn validate_float_metadata(
+ metadata: &ParquetMetaData,
+ arrow_schema: &Schema,
+ parquet_schema: &SchemaDescriptor,
+) -> Result<()> {
+ let converter = StatisticsConverter::try_new("float_ieee754",
arrow_schema, parquet_schema)?;
+ let row_group_indices: Vec<_> = (0..metadata.num_row_groups()).collect();
+
+ // verify column statistics mins
+ let exp: Arc<dyn Array> =
Arc::new(Float32Array::from(FLOAT_MINS.to_vec()));
+ let mins = converter.row_group_mins(metadata.row_groups())?;
+ assert_eq!(&mins, &exp);
+
+ // verify page mins (should be 1 page per row group, so should be same)
+ let page_mins = converter.data_page_mins(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_mins, &exp);
+
+ let exp: Arc<dyn Array> =
Arc::new(Float32Array::from(FLOAT_MAXS.to_vec()));
+ let maxs = converter.row_group_maxes(metadata.row_groups())?;
+ assert_eq!(&maxs, &exp);
+
+ // verify page maxs (should be 1 page per row group, so should be same)
+ let page_maxs = converter.data_page_maxes(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_maxs, &exp);
+
+ let exp = UInt64Array::from(NAN_COUNTS.to_vec());
+ let nans = converter.row_group_nan_counts(metadata.row_groups())?;
+ assert_eq!(&nans, &exp);
+
+ let page_nans = converter.data_page_nan_counts(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_nans, &exp);
+
+ Ok(())
+}
+
+const DOUBLE_NEG_NAN_SMALL: f64 = f64::from_bits(0xffffffffffffffff);
+const DOUBLE_NEG_NAN_LARGE: f64 = f64::from_bits(0xfff0000000000001);
+const DOUBLE_NAN_SMALL: f64 = f64::from_bits(0x7ff0000000000001);
+const DOUBLE_NAN_LARGE: f64 = f64::from_bits(0x7fffffffffffffff);
+
+const DOUBLE_MINS: [f64; 5] = [-2.0, -2.0, DOUBLE_NEG_NAN_SMALL, 0.0, -5.0];
+const DOUBLE_MAXS: [f64; 5] = [5.0, 3.0, DOUBLE_NAN_LARGE, 5.0, -0.0];
+
+fn validate_double_metadata(
+ metadata: &ParquetMetaData,
+ arrow_schema: &Schema,
+ parquet_schema: &SchemaDescriptor,
+) -> Result<()> {
+ let converter = StatisticsConverter::try_new("double_ieee754",
arrow_schema, parquet_schema)?;
+ let row_group_indices: Vec<_> = (0..metadata.num_row_groups()).collect();
+
+ // verify column statistics mins
+ let exp: Arc<dyn Array> =
Arc::new(Float64Array::from(DOUBLE_MINS.to_vec()));
+ let mins = converter.row_group_mins(metadata.row_groups())?;
+ assert_eq!(&mins, &exp);
+
+ // verify page mins (should be 1 page per row group, so should be same)
+ let page_mins = converter.data_page_mins(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_mins, &exp);
+
+ let exp: Arc<dyn Array> =
Arc::new(Float64Array::from(DOUBLE_MAXS.to_vec()));
+ let maxs = converter.row_group_maxes(metadata.row_groups())?;
+ assert_eq!(&maxs, &exp);
+
+ // verify page maxs (should be 1 page per row group, so should be same)
+ let page_maxs = converter.data_page_maxes(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_maxs, &exp);
+
+ let exp = UInt64Array::from(NAN_COUNTS.to_vec());
+ let nans = converter.row_group_nan_counts(metadata.row_groups())?;
+ assert_eq!(&nans, &exp);
+
+ let page_nans = converter.data_page_nan_counts(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_nans, &exp);
+
+ Ok(())
+}
+
+const FLOAT16_NEG_NAN_SMALL: f16 = f16::from_bits(0xffff);
+const FLOAT16_NEG_NAN_LARGE: f16 = f16::from_bits(0xfc01);
+const FLOAT16_NAN_SMALL: f16 = f16::from_bits(0x7c01);
+const FLOAT16_NAN_LARGE: f16 = f16::from_bits(0x7fff);
+
+const FLOAT16_MINS: [f16; 5] = [
+ f16::from_bits(0xc000),
+ f16::from_bits(0xc000),
+ FLOAT16_NEG_NAN_SMALL,
+ f16::from_bits(0x0000),
+ f16::from_bits(0xc500),
+];
+const FLOAT16_MAXS: [f16; 5] = [
+ f16::from_bits(0x4500),
+ f16::from_bits(0x4200),
+ FLOAT16_NAN_LARGE,
+ f16::from_bits(0x4500),
+ f16::from_bits(0x8000),
+];
+
+fn validate_float16_metadata(
+ metadata: &ParquetMetaData,
+ arrow_schema: &Schema,
+ parquet_schema: &SchemaDescriptor,
+) -> Result<()> {
+ let converter = StatisticsConverter::try_new("float16_ieee754",
arrow_schema, parquet_schema)?;
+ let row_group_indices: Vec<_> = (0..metadata.num_row_groups()).collect();
+
+ // verify column statistics mins
+ let exp: Arc<dyn Array> =
Arc::new(Float16Array::from(FLOAT16_MINS.to_vec()));
+ let mins = converter.row_group_mins(metadata.row_groups())?;
+ assert_eq!(&mins, &exp);
+
+ // verify page mins (should be 1 page per row group, so should be same)
+ let page_mins = converter.data_page_mins(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_mins, &exp);
+
+ let exp: Arc<dyn Array> =
Arc::new(Float16Array::from(FLOAT16_MAXS.to_vec()));
+ let maxs = converter.row_group_maxes(metadata.row_groups())?;
+ assert_eq!(&maxs, &exp);
+
+ // verify page maxs (should be 1 page per row group, so should be same)
+ let page_maxs = converter.data_page_maxes(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_maxs, &exp);
+
+ let exp = UInt64Array::from(NAN_COUNTS.to_vec());
+ let nans = converter.row_group_nan_counts(metadata.row_groups())?;
+ assert_eq!(&nans, &exp);
+
+ let page_nans = converter.data_page_nan_counts(
+ metadata.column_index().unwrap(),
+ metadata.offset_index().unwrap(),
+ &row_group_indices,
+ )?;
+ assert_eq!(&page_nans, &exp);
+
+ Ok(())
+}
+
+fn validate_metadata(
+ metadata: &ParquetMetaData,
+ arrow_schema: &Schema,
+ parquet_schema: &SchemaDescriptor,
+) -> Result<()> {
+ validate_float_metadata(metadata, arrow_schema, parquet_schema)?;
+ validate_double_metadata(metadata, arrow_schema, parquet_schema)?;
+ validate_float16_metadata(metadata, arrow_schema, parquet_schema)
+}
+
+#[test]
+fn test_ieee754_interop() {
+ // 1) read interop file
+ // 2) validate stats are as expected
+ // 3) rewrite file, check validate metadata from writer
+ // 4) re-read what we've written, again validate metadata
+ let parquet_testing_data = parquet_test_data();
+ let path =
PathBuf::from(parquet_testing_data).join("floating_orders_nan_count.parquet");
+ println!("Reading file: {path:?}");
+
+ let file = std::fs::File::open(&path).unwrap();
+ let options = ArrowReaderOptions::new()
+
.with_page_index_policy(parquet::file::metadata::PageIndexPolicy::Required);
+ let builder = ArrowReaderBuilder::try_new_with_options(file,
options).unwrap();
+ let file_metadata = builder.metadata().clone();
+ let schema = builder.schema().clone();
+ let parquet_schema = builder.parquet_schema().clone();
+
+ println!("validate interop file");
+ validate_metadata(file_metadata.as_ref(), schema.as_ref(), &parquet_schema)
+ .expect("validate read metadata");
+
+ let reader = builder.build().unwrap();
+ let mut outbuf = Vec::new();
+ {
+ let writer_options = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(10))
+ .build();
+ let mut writer = ArrowWriter::try_new(&mut outbuf, schema.clone(),
Some(writer_options))
+ .expect("create arrow writer");
+ for maybe_batch in reader {
+ let batch = maybe_batch.expect("reading batch");
+ writer.write(&batch).expect("writing data");
+ }
+ let write_meta = writer.close().expect("closing file");
+ println!("validate writer output");
+ validate_metadata(&write_meta, schema.as_ref(), &parquet_schema)
+ .expect("validate written metadata");
+ }
+
+ //fs::write("output.pq", outbuf.clone()).unwrap();
+
+ // now re-validate the bit we've written
+ let options = ArrowReaderOptions::new()
+
.with_page_index_policy(parquet::file::metadata::PageIndexPolicy::Required);
+ let builder =
ArrowReaderBuilder::try_new_with_options(Bytes::from(outbuf), options).unwrap();
+ let file_metadata = builder.metadata().clone();
+ let schema = builder.schema().clone();
+ let parquet_schema = builder.parquet_schema().clone();
+
+ println!("validate from rust output");
+ validate_metadata(file_metadata.as_ref(), schema.as_ref(), &parquet_schema)
+ .expect("validate re-read metadata");
+}
+
+// This test replicates the data produced by the parquet-java code that
generated
+// parquet-testing/data/floating_orders_nan_count.parquet
+#[test]
+fn test_ieee754_interop2() {
+ // define schema
+ let schema = Schema::new(vec![
+ Field::new("float_ieee754", DataType::Float32, false),
+ Field::new("double_ieee754", DataType::Float64, false),
+ Field::new("float16_ieee754", DataType::Float16, false),
+ ]);
+ let schema = Arc::new(schema);
+
+ let mut outbuf = Vec::new();
+ {
+ let writer_options = WriterProperties::builder()
+ .set_max_row_group_row_count(Some(10))
+ .build();
+ let mut writer = ArrowWriter::try_new(&mut outbuf, schema.clone(),
Some(writer_options))
+ .expect("create arrow writer");
+
+ // this only works for non-NaN cases
+ let make_batch = |data: &[f32]| -> RecordBatch {
+ let arr1 = Float32Array::from(data.to_vec());
+ let arr2 = Float64Array::from(data.iter().map(|v| *v as
f64).collect::<Vec<_>>());
+ let arr3 =
+ Float16Array::from(data.iter().map(|v|
f16::from_f32(*v)).collect::<Vec<_>>());
+
+ RecordBatch::try_new(
+ schema.clone(),
+ vec![Arc::new(arr1), Arc::new(arr2), Arc::new(arr3)],
+ )
+ .unwrap()
+ };
+
+ // batch 1: no NaNs
+ let batch = make_batch(&[-2.0f32, -1.0, -0.0, 0.0, 0.5, 1.0, 2.0, 3.0,
4.0, 5.0]);
+ writer.write(&batch).expect("writing batch1");
+
+ // batch 2: mixed
+ let float_data = vec![
+ FLOAT_NEG_NAN_SMALL,
+ -2.0,
+ FLOAT_NEG_NAN_LARGE,
+ -1.0,
+ -0.0,
+ 0.0,
+ 1.0,
+ FLOAT_NAN_SMALL,
+ 3.0,
+ FLOAT_NAN_LARGE,
+ ];
+ let double_data = vec![
+ DOUBLE_NEG_NAN_SMALL,
+ -2.0,
+ DOUBLE_NEG_NAN_LARGE,
+ -1.0,
+ -0.0,
+ 0.0,
+ 1.0,
+ DOUBLE_NAN_SMALL,
+ 3.0,
+ DOUBLE_NAN_LARGE,
+ ];
+ let float16_data = vec![
+ FLOAT16_NEG_NAN_SMALL,
+ f16::from_f32(-2.0),
+ FLOAT16_NEG_NAN_LARGE,
+ f16::from_f32(-1.0),
+ f16::from_f32(-0.0),
+ f16::from_f32(0.0),
+ f16::from_f32(1.0),
+ FLOAT16_NAN_SMALL,
+ f16::from_f32(3.0),
+ FLOAT16_NAN_LARGE,
+ ];
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Float32Array::from(float_data)),
+ Arc::new(Float64Array::from(double_data)),
+ Arc::new(Float16Array::from(float16_data)),
+ ],
+ )
+ .unwrap();
+ writer.write(&batch).expect("writing batch2");
+
+ // batch 3: all NaN
+ let float_data = vec![
+ FLOAT_NEG_NAN_SMALL,
+ FLOAT_NEG_NAN_LARGE,
+ FLOAT_NAN_SMALL,
+ FLOAT_NAN_LARGE,
+ FLOAT_NEG_NAN_SMALL,
+ FLOAT_NEG_NAN_LARGE,
+ FLOAT_NAN_SMALL,
+ FLOAT_NAN_LARGE,
+ FLOAT_NEG_NAN_SMALL,
+ FLOAT_NAN_LARGE,
+ ];
+ let double_data = vec![
+ DOUBLE_NEG_NAN_SMALL,
+ DOUBLE_NEG_NAN_LARGE,
+ DOUBLE_NAN_SMALL,
+ DOUBLE_NAN_LARGE,
+ DOUBLE_NEG_NAN_SMALL,
+ DOUBLE_NEG_NAN_LARGE,
+ DOUBLE_NAN_SMALL,
+ DOUBLE_NAN_LARGE,
+ DOUBLE_NEG_NAN_SMALL,
+ DOUBLE_NAN_LARGE,
+ ];
+ let float16_data = vec![
+ FLOAT16_NEG_NAN_SMALL,
+ FLOAT16_NEG_NAN_LARGE,
+ FLOAT16_NAN_SMALL,
+ FLOAT16_NAN_LARGE,
+ FLOAT16_NEG_NAN_SMALL,
+ FLOAT16_NEG_NAN_LARGE,
+ FLOAT16_NAN_SMALL,
+ FLOAT16_NAN_LARGE,
+ FLOAT16_NEG_NAN_SMALL,
+ FLOAT16_NAN_LARGE,
+ ];
+ let batch = RecordBatch::try_new(
+ schema.clone(),
+ vec![
+ Arc::new(Float32Array::from(float_data)),
+ Arc::new(Float64Array::from(double_data)),
+ Arc::new(Float16Array::from(float16_data)),
+ ],
+ )
+ .unwrap();
+ writer.write(&batch).expect("writing batch3");
+
+ // batch 4: 0 min
+ let batch = make_batch(&[0.0f32, 0.0, 0.0, 0.5, 1.0, 1.5, 2.0, 3.0,
4.0, 5.0]);
+ writer.write(&batch).expect("writing batch4");
+
+ // batch 5: -0 max
+ let batch = make_batch(&[
+ -5.0f32, -4.0, -3.0, -2.0, -1.5, -1.0, -0.5, -0.0, -0.0, -0.0,
+ ]);
+ writer.write(&batch).expect("writing batch5");
+
+ let write_meta = writer.close().expect("closing file");
+ let parquet_schema = write_meta.file_metadata().schema_descr();
+ println!("validate writer output");
+ validate_metadata(&write_meta, schema.as_ref(), parquet_schema)
+ .expect("validate written metadata");
+ }
+
+ //fs::write("output2.pq", outbuf.clone()).unwrap();
+
+ // now re-validate the bit we've written
+ let options = ArrowReaderOptions::new()
+
.with_page_index_policy(parquet::file::metadata::PageIndexPolicy::Required);
+ let builder =
ArrowReaderBuilder::try_new_with_options(Bytes::from(outbuf), options).unwrap();
+ let file_metadata = builder.metadata().clone();
+ let schema = builder.schema().clone();
+ let parquet_schema = builder.parquet_schema().clone();
+
+ println!("validate from rust output");
+ validate_metadata(file_metadata.as_ref(), schema.as_ref(), &parquet_schema)
+ .expect("validate re-read metadata");
+}
