This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24433-dbdc627deb98778dbf23cf7ccc2df7c66e1d9f55 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit f239959350080b5bfe4403d4de5e83bcfc03dc0b Author: Theo Butler <[email protected]> AuthorDate: Thu Aug 20 16:28:03 2026 +0000 fix: expand float fuzz coverage and correct discovered inconsistencies (#24433) ## Which issue does this PR close? - Closes #24431. - Closes #24432. ## Rationale for this change Two floating-point edge cases produced inconsistent results: - `ScalarValue::eq_array` used IEEE equality for floats, while `ScalarValue` equality uses bit representations. An identical NaN therefore compared unequal, while `+0.0` and `-0.0` compared equal. - Grouped floating-point `MIN` and `MAX` used partial ordering. With NaNs or signed zeros, results could depend on input order, batching, or partitioning and disagree with the non-grouped accumulators. Both issues were exposed by generating special floating-point values in fuzz data and adding fuzz coverage for `ScalarValue` array conversions. ## What changes are included in this PR? - Generate `+0.0`, `-0.0`, positive and negative infinities, and NaNs with varied signs and payloads in random Float32 and Float64 test data. - Add fuzz coverage checking consistency and round trips among `ScalarValue::try_from_array`, `eq_array`, `to_array`, and `iter_to_array` across supported array types and sliced arrays. - Make Float16, Float32, and Float64 `ScalarValue::eq_array` comparisons use the same bitwise equality as `ScalarValue::PartialEq`. - Make grouped Float16, Float32, and Float64 `MIN`/`MAX` use total ordering and total-order extrema as their initial accumulator values. ## Are these changes tested? Yes. The existing `scalar_eq_array` test now covers NaNs and signed zeros. New deterministic grouped-accumulator tests cover positive and negative NaNs, signed zeros, and the total-order sentinel values. The scalar conversion and aggregation fuzz tests cover the same paths across randomized arrays and execution configurations. ## Are there any user-facing changes? Yes. `ScalarValue::eq_array` now uses bitwise float equality, and grouped floating-point `MIN` and `MAX` now return deterministic results consistent with DataFusion's existing total-order semantics. There are no API signature changes. --- datafusion/common/src/scalar/mod.rs | 19 ++- datafusion/core/tests/fuzz_cases/mod.rs | 1 + .../core/tests/fuzz_cases/scalar_value_fuzz.rs | 131 +++++++++++++++++++++ datafusion/functions-aggregate/src/min_max.rs | 93 +++++++++++++-- test-utils/src/array_gen/random_data.rs | 29 ++++- 5 files changed, 260 insertions(+), 13 deletions(-) diff --git a/datafusion/common/src/scalar/mod.rs b/datafusion/common/src/scalar/mod.rs index bef46702b8..3ae23d3b74 100644 --- a/datafusion/common/src/scalar/mod.rs +++ b/datafusion/common/src/scalar/mod.rs @@ -1217,6 +1217,17 @@ macro_rules! eq_array_primitive { }}; } +macro_rules! eq_array_float { + ($array:expr, $index:expr, $array_cast:ident, $VALUE:expr) => {{ + let array = $array_cast($array)?; + let is_valid = array.is_valid($index); + Ok::<bool, DataFusionError>(match $VALUE { + Some(val) => is_valid && array.value($index).to_bits() == val.to_bits(), + None => !is_valid, + }) + }}; +} + impl ScalarValue { /// Create a [`Result<ScalarValue>`] with the provided value and datatype /// @@ -4574,13 +4585,13 @@ impl ScalarValue { eq_array_primitive!(array, index, as_boolean_array, val)? } ScalarValue::Float16(val) => { - eq_array_primitive!(array, index, as_float16_array, val)? + eq_array_float!(array, index, as_float16_array, val)? } ScalarValue::Float32(val) => { - eq_array_primitive!(array, index, as_float32_array, val)? + eq_array_float!(array, index, as_float32_array, val)? } ScalarValue::Float64(val) => { - eq_array_primitive!(array, index, as_float64_array, val)? + eq_array_float!(array, index, as_float64_array, val)? } ScalarValue::Int8(val) => { eq_array_primitive!(array, index, as_int8_array, val)? @@ -7874,7 +7885,7 @@ mod tests { } let bool_vals = [Some(true), None, Some(false)]; - let f32_vals = [Some(-1.0), None, Some(1.0)]; + let f32_vals = [Some(-0.0), Some(0.0), Some(f32::NAN), None, Some(1.0)]; let f64_vals = make_typed_vec!(f32_vals, f64); let i8_vals = [Some(-1), None, Some(1)]; diff --git a/datafusion/core/tests/fuzz_cases/mod.rs b/datafusion/core/tests/fuzz_cases/mod.rs index edb53df382..3e425e48c7 100644 --- a/datafusion/core/tests/fuzz_cases/mod.rs +++ b/datafusion/core/tests/fuzz_cases/mod.rs @@ -41,4 +41,5 @@ mod window_fuzz; // Utility modules mod once_exec; mod record_batch_generator; +mod scalar_value_fuzz; mod spilling_fuzz_in_memory_constrained_env; diff --git a/datafusion/core/tests/fuzz_cases/scalar_value_fuzz.rs b/datafusion/core/tests/fuzz_cases/scalar_value_fuzz.rs new file mode 100644 index 0000000000..040a86ea37 --- /dev/null +++ b/datafusion/core/tests/fuzz_cases/scalar_value_fuzz.rs @@ -0,0 +1,131 @@ +// 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. + +use arrow::array::ArrayRef; +use datafusion_common::ScalarValue; +use rand::random; + +use super::record_batch_generator::{RecordBatchGenerator, get_supported_types_columns}; + +const NUM_SEEDS: usize = 32; + +#[test] +fn scalar_value_eq_array_consistency() { + for_each_array(|context, array| { + for (index, expected) in extract_scalars(array, context).iter().enumerate() { + assert!( + expected + .eq_array(array, index) + .unwrap_or_else(|e| panic!("eq_array failed for {context}: {e}")), + "eq_array disagreed with try_from_array at index {index} for {context}" + ); + } + }); +} + +#[test] +fn scalar_value_iter_to_array_roundtrip() { + for_each_array(|context, array| { + let expected = extract_scalars(array, context); + let rebuilt = ScalarValue::iter_to_array(expected.iter().cloned()) + .unwrap_or_else(|e| panic!("iter_to_array failed for {context}: {e}")); + + assert_eq!( + rebuilt.data_type(), + array.data_type(), + "iter_to_array changed the type for {context}" + ); + assert_eq!( + rebuilt.len(), + array.len(), + "iter_to_array changed the length for {context}" + ); + + let actual = extract_scalars(&rebuilt, context); + assert_eq!( + actual, expected, + "iter_to_array changed values for {context}" + ); + }); +} + +#[test] +fn scalar_value_to_array_roundtrip() { + for_each_array(|context, array| { + for (index, expected) in extract_scalars(array, context).iter().enumerate() { + let singleton = expected + .to_array() + .unwrap_or_else(|e| panic!("to_array failed for {context}: {e}")); + assert_eq!( + singleton.len(), + 1, + "to_array returned the wrong length at index {index} for {context}" + ); + + let actual = ScalarValue::try_from_array(&singleton, 0).unwrap_or_else(|e| { + panic!( + "try_from_array failed on to_array output at index {index} for {context}: {e}" + ) + }); + assert_eq!( + &actual, expected, + "to_array roundtrip changed index {index} for {context}" + ); + } + }); +} + +fn for_each_array(check: impl Fn(&str, &ArrayRef)) { + for _ in 0..NUM_SEEDS { + let seed = random(); + let columns = get_supported_types_columns(seed); + let batch = RecordBatchGenerator::new(1, 128, columns) + .with_seed(seed) + .generate() + .unwrap_or_else(|e| panic!("failed to generate batch for seed {seed}: {e}")); + + for (field, array) in batch.schema().fields().iter().zip(batch.columns()) { + let check_array = |array: &ArrayRef| { + let context = format!( + "seed={seed}, column={}, type={}, len={}, offset={}", + field.name(), + array.data_type(), + array.len(), + array.offset() + ); + check(&context, array); + }; + + check_array(array); + + // Also exercise arrays with non-zero offsets. + if array.len() > 2 { + check_array(&array.slice(1, array.len() - 2)); + } + } + } +} + +fn extract_scalars(array: &ArrayRef, context: &str) -> Vec<ScalarValue> { + (0..array.len()) + .map(|index| { + ScalarValue::try_from_array(array, index).unwrap_or_else(|e| { + panic!("try_from_array failed at index {index} for {context}: {e}") + }) + }) + .collect() +} diff --git a/datafusion/functions-aggregate/src/min_max.rs b/datafusion/functions-aggregate/src/min_max.rs index 400956af0e..73a4b84fcf 100644 --- a/datafusion/functions-aggregate/src/min_max.rs +++ b/datafusion/functions-aggregate/src/min_max.rs @@ -130,6 +130,17 @@ macro_rules! primitive_max_accumulator { .with_starting_value($NATIVE::MIN), )) }}; + ($DATA_TYPE:ident, $NATIVE:ident, $PRIMTYPE:ident, total, $BITS:ident) => {{ + Ok(Box::new( + PrimitiveGroupsAccumulator::<$PRIMTYPE, _>::new($DATA_TYPE, |cur, new| { + if new.total_cmp(cur) == Ordering::Greater { + *cur = new + } + }) + // Use the total-order minimum so negative NaNs replace the sentinel. + .with_starting_value($NATIVE::from_bits($BITS::MAX)), + )) + }}; } /// Creates a [`PrimitiveGroupsAccumulator`] for computing `MIN` @@ -153,6 +164,17 @@ macro_rules! primitive_min_accumulator { .with_starting_value($NATIVE::MAX), )) }}; + ($DATA_TYPE:ident, $NATIVE:ident, $PRIMTYPE:ident, total, $BITS:ident) => {{ + Ok(Box::new( + PrimitiveGroupsAccumulator::<$PRIMTYPE, _>::new(&$DATA_TYPE, |cur, new| { + if new.total_cmp(cur) == Ordering::Less { + *cur = new + } + }) + // Use the total-order maximum so positive NaNs replace the sentinel. + .with_starting_value($NATIVE::from_bits($BITS::MAX >> 1)), + )) + }}; } trait FromColumnStatistics { @@ -272,13 +294,13 @@ impl AggregateUDFImpl for Max { UInt32 => primitive_max_accumulator!(data_type, u32, UInt32Type), UInt64 => primitive_max_accumulator!(data_type, u64, UInt64Type), Float16 => { - primitive_max_accumulator!(data_type, f16, Float16Type) + primitive_max_accumulator!(data_type, f16, Float16Type, total, u16) } Float32 => { - primitive_max_accumulator!(data_type, f32, Float32Type) + primitive_max_accumulator!(data_type, f32, Float32Type, total, u32) } Float64 => { - primitive_max_accumulator!(data_type, f64, Float64Type) + primitive_max_accumulator!(data_type, f64, Float64Type, total, u64) } Date32 => primitive_max_accumulator!(data_type, i32, Date32Type), Date64 => primitive_max_accumulator!(data_type, i64, Date64Type), @@ -566,13 +588,13 @@ impl AggregateUDFImpl for Min { UInt32 => primitive_min_accumulator!(data_type, u32, UInt32Type), UInt64 => primitive_min_accumulator!(data_type, u64, UInt64Type), Float16 => { - primitive_min_accumulator!(data_type, f16, Float16Type) + primitive_min_accumulator!(data_type, f16, Float16Type, total, u16) } Float32 => { - primitive_min_accumulator!(data_type, f32, Float32Type) + primitive_min_accumulator!(data_type, f32, Float32Type, total, u32) } Float64 => { - primitive_min_accumulator!(data_type, f64, Float64Type) + primitive_min_accumulator!(data_type, f64, Float64Type, total, u64) } Date32 => primitive_min_accumulator!(data_type, i32, Date32Type), Date64 => primitive_min_accumulator!(data_type, i64, Date64Type), @@ -1014,7 +1036,7 @@ mod tests { use super::*; use arrow::{ array::{ - Array, DictionaryArray, Float32Array, Int8Array, Int32Array, + Array, AsArray, DictionaryArray, Float32Array, Int8Array, Int32Array, IntervalDayTimeArray, IntervalMonthDayNanoArray, IntervalYearMonthArray, PrimitiveArray, StringArray, }, @@ -1023,8 +1045,65 @@ mod tests { IntervalUnit, IntervalYearMonthType, }, }; + use datafusion_expr::EmitTo; use std::sync::Arc; + #[test] + fn grouped_float_min_max_total_order() -> Result<()> { + fn grouped_float32_min() -> Result<Box<dyn GroupsAccumulator>> { + let data_type = &DataType::Float32; + primitive_min_accumulator!(data_type, f32, Float32Type, total, u32) + } + + fn grouped_float32_max() -> Result<Box<dyn GroupsAccumulator>> { + let data_type = &DataType::Float32; + primitive_max_accumulator!(data_type, f32, Float32Type, total, u32) + } + + fn evaluate_grouped_float32( + mut accumulator: Box<dyn GroupsAccumulator>, + values: Vec<f32>, + ) -> f32 { + let group_indices = vec![0; values.len()]; + let values = Arc::new(Float32Array::from(values)) as ArrayRef; + accumulator + .update_batch(&[values], &group_indices, None, 1) + .unwrap(); + accumulator + .evaluate(EmitTo::All) + .unwrap() + .as_primitive::<Float32Type>() + .value(0) + } + + let positive_nan = f32::NAN; + let negative_nan = f32::from_bits(f32::NAN.to_bits() | (1 << 31)); + + let min_cases = [ + (vec![positive_nan, 1.0], 1.0), + (vec![1.0, negative_nan], negative_nan), + (vec![0.0, -0.0], -0.0), + (vec![positive_nan], positive_nan), + ]; + for (values, expected) in min_cases { + let actual = evaluate_grouped_float32(grouped_float32_min()?, values); + assert_eq!(actual.to_bits(), expected.to_bits()); + } + + let max_cases = [ + (vec![positive_nan, 1.0], positive_nan), + (vec![1.0, negative_nan], 1.0), + (vec![-0.0, 0.0], 0.0), + (vec![negative_nan], negative_nan), + ]; + for (values, expected) in max_cases { + let actual = evaluate_grouped_float32(grouped_float32_max()?, values); + assert_eq!(actual.to_bits(), expected.to_bits()); + } + + Ok(()) + } + #[test] fn interval_min_max() { // IntervalYearMonth diff --git a/test-utils/src/array_gen/random_data.rs b/test-utils/src/array_gen/random_data.rs index f341d23417..bcd8e4d24c 100644 --- a/test-utils/src/array_gen/random_data.rs +++ b/test-utils/src/array_gen/random_data.rs @@ -51,6 +51,31 @@ macro_rules! basic_random_data { }; } +macro_rules! float_random_data { + ($ARROW_TYPE:ty, $NATIVE:ty) => { + impl RandomNativeData for $ARROW_TYPE { + fn generate_random_native_data(rng: &mut StdRng) -> Self::Native { + match rng.random_range(0..100) { + 0 => 0.0, + 1 => -0.0, + 2 => <$NATIVE>::INFINITY, + 3 => <$NATIVE>::NEG_INFINITY, + 4 => { + // Generate a NaN with a uniformly selected sign and nonzero payload. + let sign_mask = (-0.0 as $NATIVE).to_bits(); + let exponent_mask = <$NATIVE>::INFINITY.to_bits(); + let payload_mask = !(sign_mask | exponent_mask); + let sign = if rng.random() { sign_mask } else { 0 }; + let payload = rng.random_range(1..=payload_mask); + <$NATIVE>::from_bits(sign | exponent_mask | payload) + } + _ => rng.random(), + } + } + } + }; +} + basic_random_data!(Int8Type); basic_random_data!(Int16Type); basic_random_data!(Int32Type); @@ -59,8 +84,8 @@ basic_random_data!(UInt8Type); basic_random_data!(UInt16Type); basic_random_data!(UInt32Type); basic_random_data!(UInt64Type); -basic_random_data!(Float32Type); -basic_random_data!(Float64Type); +float_random_data!(Float32Type, f32); +float_random_data!(Float64Type, f64); basic_random_data!(Date32Type); basic_random_data!(Time32SecondType); basic_random_data!(Time32MillisecondType); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
