kosiew commented on code in PR #26011:
URL: https://github.com/apache/datafusion/pull/26011#discussion_r4345846501
##########
datafusion/spark/src/function/array/array_contains.rs:
##########
@@ -77,6 +86,129 @@ impl ScalarUDFImpl for SparkArrayContains {
}
}
+/// Rewrites float values so that bitwise comparison matches Spark's equality:
+/// `-0.0` becomes `0.0` and every NaN becomes the canonical NaN. Applies to a
+/// float argument and to the elements of a list argument; other values are
+/// returned as they are.
+fn canonicalize_floats(value: &ColumnarValue) -> Result<ColumnarValue> {
Review Comment:
Optional simplification: could canonicalize_floats return ColumnarValue
directly? Neither it nor its callees returns a recoverable error, so removing
Result, the outer Ok, and the caller's ? would clarify the failure boundary.
This does not address the reconstruction panic, which Result cannot catch.
##########
datafusion/spark/src/function/array/array_contains.rs:
##########
@@ -77,6 +86,129 @@ impl ScalarUDFImpl for SparkArrayContains {
}
}
+/// Rewrites float values so that bitwise comparison matches Spark's equality:
+/// `-0.0` becomes `0.0` and every NaN becomes the canonical NaN. Applies to a
+/// float argument and to the elements of a list argument; other values are
+/// returned as they are.
+fn canonicalize_floats(value: &ColumnarValue) -> Result<ColumnarValue> {
+ Ok(match value {
+ ColumnarValue::Array(array) => {
+ ColumnarValue::Array(canonicalize_float_array(array))
+ }
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Float32(Some(v)) => {
+
ColumnarValue::Scalar(ScalarValue::Float32(Some(canonical_f32(*v))))
+ }
+ ScalarValue::Float64(Some(v)) => {
+
ColumnarValue::Scalar(ScalarValue::Float64(Some(canonical_f64(*v))))
+ }
+ ScalarValue::List(list) => ColumnarValue::Scalar(ScalarValue::List(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_list::<i32>()
+ .clone()
+ .into(),
+ )),
+ ScalarValue::LargeList(list) => {
+ ColumnarValue::Scalar(ScalarValue::LargeList(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_list::<i64>()
+ .clone()
+ .into(),
+ ))
+ }
+ ScalarValue::FixedSizeList(list) => {
+ ColumnarValue::Scalar(ScalarValue::FixedSizeList(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_fixed_size_list()
+ .clone()
+ .into(),
+ ))
+ }
+ ScalarValue::Struct(list) =>
ColumnarValue::Scalar(ScalarValue::Struct(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_struct()
+ .clone()
+ .into(),
+ )),
+ _ => value.clone(),
+ },
+ })
+}
+
+fn canonical_f32(v: f32) -> f32 {
+ if v.is_nan() {
+ f32::NAN
+ } else if v == 0.0 {
+ 0.0
+ } else {
+ v
+ }
+}
+
+fn canonical_f64(v: f64) -> f64 {
+ if v.is_nan() {
+ f64::NAN
+ } else if v == 0.0 {
+ 0.0
+ } else {
+ v
+ }
+}
+
+fn canonicalize_float_array(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => Arc::new(unary::<Float32Type, _, Float32Type>(
+ array.as_primitive::<Float32Type>(),
+ canonical_f32,
+ )),
+ DataType::Float64 => Arc::new(unary::<Float64Type, _, Float64Type>(
+ array.as_primitive::<Float64Type>(),
+ canonical_f64,
+ )),
+ DataType::List(field) => {
+ let list = array.as_list::<i32>();
+ Arc::new(GenericListArray::<i32>::new(
+ Arc::clone(field),
+ list.offsets().clone(),
+ canonicalize_float_array(list.values()),
+ list.nulls().cloned(),
+ ))
+ }
+ DataType::LargeList(field) => {
+ let list = array.as_list::<i64>();
+ Arc::new(GenericListArray::<i64>::new(
+ Arc::clone(field),
+ list.offsets().clone(),
+ canonicalize_float_array(list.values()),
+ list.nulls().cloned(),
+ ))
+ }
+ DataType::FixedSizeList(field, size) => {
+ let list = array.as_fixed_size_list();
+ Arc::new(FixedSizeListArray::new(
Review Comment:
A zero-width FixedSizeListArray can contain one valid row, but
FixedSizeListArray::new infers length zero, silently dropping that row or
panicking when nested inside a List with offsets [0, 1]. This also affects
non-float children. Please return the original array when size == 0 or use a
length-preserving constructor, and add UDF tests verifying one valid false
result for a non-null scalar needle against an array-valued haystack containing
one zero-width list, plus preserved row counts without panic for a zero-width
list nested inside a List.
##########
datafusion/spark/src/function/array/array_contains.rs:
##########
@@ -77,6 +86,129 @@ impl ScalarUDFImpl for SparkArrayContains {
}
}
+/// Rewrites float values so that bitwise comparison matches Spark's equality:
+/// `-0.0` becomes `0.0` and every NaN becomes the canonical NaN. Applies to a
+/// float argument and to the elements of a list argument; other values are
+/// returned as they are.
+fn canonicalize_floats(value: &ColumnarValue) -> Result<ColumnarValue> {
+ Ok(match value {
+ ColumnarValue::Array(array) => {
+ ColumnarValue::Array(canonicalize_float_array(array))
+ }
+ ColumnarValue::Scalar(scalar) => match scalar {
+ ScalarValue::Float32(Some(v)) => {
+
ColumnarValue::Scalar(ScalarValue::Float32(Some(canonical_f32(*v))))
+ }
+ ScalarValue::Float64(Some(v)) => {
+
ColumnarValue::Scalar(ScalarValue::Float64(Some(canonical_f64(*v))))
+ }
+ ScalarValue::List(list) => ColumnarValue::Scalar(ScalarValue::List(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_list::<i32>()
+ .clone()
+ .into(),
+ )),
+ ScalarValue::LargeList(list) => {
+ ColumnarValue::Scalar(ScalarValue::LargeList(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_list::<i64>()
+ .clone()
+ .into(),
+ ))
+ }
+ ScalarValue::FixedSizeList(list) => {
+ ColumnarValue::Scalar(ScalarValue::FixedSizeList(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_fixed_size_list()
+ .clone()
+ .into(),
+ ))
+ }
+ ScalarValue::Struct(list) =>
ColumnarValue::Scalar(ScalarValue::Struct(
+ canonicalize_float_array(&(Arc::clone(list) as ArrayRef))
+ .as_struct()
+ .clone()
+ .into(),
+ )),
+ _ => value.clone(),
+ },
+ })
+}
+
+fn canonical_f32(v: f32) -> f32 {
+ if v.is_nan() {
+ f32::NAN
+ } else if v == 0.0 {
+ 0.0
+ } else {
+ v
+ }
+}
+
+fn canonical_f64(v: f64) -> f64 {
+ if v.is_nan() {
+ f64::NAN
+ } else if v == 0.0 {
+ 0.0
+ } else {
+ v
+ }
+}
+
+fn canonicalize_float_array(array: &ArrayRef) -> ArrayRef {
+ match array.data_type() {
+ DataType::Float32 => Arc::new(unary::<Float32Type, _, Float32Type>(
+ array.as_primitive::<Float32Type>(),
+ canonical_f32,
+ )),
+ DataType::Float64 => Arc::new(unary::<Float64Type, _, Float64Type>(
+ array.as_primitive::<Float64Type>(),
+ canonical_f64,
+ )),
+ DataType::List(field) => {
Review Comment:
Optional performance follow-up: consider reusing list/struct arrays when
their children are unchanged. Currently even List<Int32> is reconstructed,
while a small slice of a large float list normalizes the entire retained child
buffer. Please benchmark these cases before introducing a type gate or
visible-range optimization; no typical-workload slowdown is established yet.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
