fix: [branch-1.0] backport five bug fixes for 1.0.1

.github/workflows/pr_build_linux.yml

@@ -325,6 +325,7 @@ jobs:
               org.apache.comet.exec.CometShuffle4_0Suite
               org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
+              org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite
               org.apache.comet.exec.CometAsyncShuffleSuite

.github/workflows/pr_build_macos.yml

@@ -141,6 +141,7 @@ jobs:
               org.apache.comet.exec.CometShuffle4_0Suite
               org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
+              org.apache.spark.sql.comet.execution.shuffle.CometNativeShuffleInputRDDSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite
               org.apache.comet.exec.CometAsyncShuffleSuite

docs/source/user-guide/latest/compatibility/index.md

@@ -134,10 +134,10 @@ so users hunting an unexpected value have a single place to check:
   `DECIMAL(1, 1)`) throws `NUMERIC_VALUE_OUT_OF_RANGE` regardless of the eval mode. Spark returns
   `NULL` under legacy and try mode, and only throws under ANSI
   ([#5068](https://github.com/apache/datafusion-comet/issues/5068)).
-- `CAST(string AS timestamp)`, `CAST(string AS timestamp_ntz)`, `to_time` and `try_to_time` trim
-  Unicode whitespace. Spark trims only the bytes `0x00`-`0x20` and `0x7F`, so a value padded with
-  an ASCII control byte parses in Spark and returns `NULL` in Comet, while a value padded with
-  non-ASCII whitespace such as `U+3000` returns `NULL` in Spark and parses in Comet
+- `CAST(string AS timestamp)` and `CAST(string AS timestamp_ntz)` trim Unicode whitespace.
+  Spark trims only the bytes `0x00`-`0x20` and `0x7F`, so a value padded with an ASCII control byte
+  parses in Spark and returns `NULL` in Comet, while a value padded with non-ASCII whitespace such
+  as `U+3000` returns `NULL` in Spark and parses in Comet
   ([#5149](https://github.com/apache/datafusion-comet/issues/5149)).
 - Native `RANGE` window frames with an explicit `PRECEDING` / `FOLLOWING` offset diverge from
   Spark when the boundary arithmetic overflows for `DATE` or `DECIMAL` `ORDER BY` columns

native/spark-expr/benches/cast_from_boolean.rs

@@ -69,20 +69,7 @@ fn criterion_benchmark(c: &mut Criterion) {
         None,
         None,
     );
-    let cast_to_str = Cast::new(
-        expr.clone(),
-        DataType::Utf8,
-        spark_cast_options.clone(),
-        None,
-        None,
-    );
-    let cast_to_decimal = Cast::new(
-        expr,
-        DataType::Decimal128(10, 4),
-        spark_cast_options,
-        None,
-        None,
-    );
+    let cast_to_str = Cast::new(expr, DataType::Utf8, spark_cast_options, None, None);
 
     let mut group = c.benchmark_group("cast_bool".to_string());
     group.bench_function("i8", |b| {
@@ -106,9 +93,6 @@ fn criterion_benchmark(c: &mut Criterion) {
     group.bench_function("str", |b| {
         b.iter(|| cast_to_str.evaluate(&boolean_batch).unwrap());
     });
-    group.bench_function("decimal", |b| {
-        b.iter(|| cast_to_decimal.evaluate(&boolean_batch).unwrap());
-    });
 }
 
 fn create_boolean_batch() -> RecordBatch {

native/spark-expr/src/array_funcs/arrays_overlap.rs

@@ -272,10 +272,9 @@ fn range_has_null(nulls: Option<&NullBuffer>, range: Range<usize>) -> bool {
     nulls.is_some_and(|n| n.null_count() > 0 && n.slice(range.start, range.len()).null_count() > 0)
 }
 
-/// Projects a native value onto a hashable key whose equality matches the Arrow compare kernels:
-/// the value itself for integral types, the bit pattern for floats. Arrow orders floats by total
-/// order rather than IEEE semantics (NaN equals NaN, and 0.0 does not equal -0.0), which is
-/// exactly bit equality.
+/// Projects a native value onto a Spark-compatible hashable key. Floating-point keys canonicalize
+/// every NaN representation while preserving the distinct bit patterns of positive and negative
+/// zero.
 trait OverlapKey: Copy {
     type Key: Hash + Eq + Copy;
 
@@ -299,15 +298,23 @@ impl OverlapKey for f32 {
     type Key = u32;
 
     fn overlap_key(self) -> u32 {
-        self.to_bits()
+        if self.is_nan() {
+            f32::NAN.to_bits()
+        } else {
+            self.to_bits()
+        }
     }
 }
 
 impl OverlapKey for f64 {
     type Key = u64;
 
     fn overlap_key(self) -> u64 {
-        self.to_bits()
+        if self.is_nan() {
+            f64::NAN.to_bits()
+        } else {
+            self.to_bits()
+        }
     }
 }
 
@@ -554,6 +561,132 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_flat_float32_nan_payloads_and_signed_zero() -> Result<()> {
+        let positive_nan = f32::from_bits(0x7fc0_0001);
+        let negative_nan = f32::from_bits(0xffc0_0002);
+        let signaling_nan = f32::from_bits(0x7f80_0001);
+
+        let hash_nan_left = (1..=16)
+            .map(|value| Some(value as f32))
+            .chain([Some(positive_nan)])
+            .collect::<Vec<_>>();
+        let hash_nan_right = (17..=32)
+            .map(|value| Some(value as f32))
+            .chain([Some(negative_nan)])
+            .collect::<Vec<_>>();
+        let hash_zero_left = (1..=16)
+            .map(|value| Some(value as f32))
+            .chain([Some(0.0)])
+            .collect::<Vec<_>>();
+        let hash_zero_right = (17..=32)
+            .map(|value| Some(value as f32))
+            .chain([Some(-0.0)])
+            .collect::<Vec<_>>();
+
+        let left = ListArray::from_iter_primitive::<Float32Type, _, _>([
+            Some(vec![Some(positive_nan)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(signaling_nan)]),
+            Some(vec![Some(0.0)]),
+            Some(vec![Some(-0.0)]),
+            Some(vec![Some(positive_nan), None]),
+            Some(vec![Some(0.0), None]),
+            Some(hash_nan_left),
+            Some(hash_zero_left),
+        ]);
+        let right = ListArray::from_iter_primitive::<Float32Type, _, _>([
+            Some(vec![Some(f32::NAN)]),
+            Some(vec![Some(positive_nan)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(-0.0)]),
+            Some(vec![Some(0.0)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(-0.0)]),
+            Some(hash_nan_right),
+            Some(hash_zero_right),
+        ]);
+
+        let result = arrays_overlap_list::<i32>(&left, &right)?;
+        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
+        let expected = BooleanArray::from(vec![
+            Some(true),
+            Some(true),
+            Some(true),
+            Some(false),
+            Some(false),
+            Some(true),
+            None,
+            Some(true),
+            Some(false),
+        ]);
+        assert_eq!(result, &expected);
+        Ok(())
+    }
+
+    #[test]
+    fn test_flat_float64_nan_payloads_and_signed_zero() -> Result<()> {
+        let positive_nan = f64::from_bits(0x7ff8_0000_0000_0001);
+        let negative_nan = f64::from_bits(0xfff8_0000_0000_0002);
+        let signaling_nan = f64::from_bits(0x7ff0_0000_0000_0001);
+
+        let hash_nan_left = (1..=16)
+            .map(|value| Some(value as f64))
+            .chain([Some(positive_nan)])
+            .collect::<Vec<_>>();
+        let hash_nan_right = (17..=32)
+            .map(|value| Some(value as f64))
+            .chain([Some(negative_nan)])
+            .collect::<Vec<_>>();
+        let hash_zero_left = (1..=16)
+            .map(|value| Some(value as f64))
+            .chain([Some(0.0)])
+            .collect::<Vec<_>>();
+        let hash_zero_right = (17..=32)
+            .map(|value| Some(value as f64))
+            .chain([Some(-0.0)])
+            .collect::<Vec<_>>();
+
+        let left = ListArray::from_iter_primitive::<Float64Type, _, _>([
+            Some(vec![Some(positive_nan)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(signaling_nan)]),
+            Some(vec![Some(0.0)]),
+            Some(vec![Some(-0.0)]),
+            Some(vec![Some(positive_nan), None]),
+            Some(vec![Some(0.0), None]),
+            Some(hash_nan_left),
+            Some(hash_zero_left),
+        ]);
+        let right = ListArray::from_iter_primitive::<Float64Type, _, _>([
+            Some(vec![Some(f64::NAN)]),
+            Some(vec![Some(positive_nan)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(-0.0)]),
+            Some(vec![Some(0.0)]),
+            Some(vec![Some(negative_nan)]),
+            Some(vec![Some(-0.0)]),
+            Some(hash_nan_right),
+            Some(hash_zero_right),
+        ]);
+
+        let result = arrays_overlap_list::<i32>(&left, &right)?;
+        let result = result.as_any().downcast_ref::<BooleanArray>().unwrap();
+        let expected = BooleanArray::from(vec![
+            Some(true),
+            Some(true),
+            Some(true),
+            Some(false),
+            Some(false),
+            Some(true),
+            None,
+            Some(true),
+            Some(false),
+        ]);
+        assert_eq!(result, &expected);
+        Ok(())
+    }
+
     #[test]
     fn test_null_only_overlap() -> Result<()> {
         // [1, NULL] vs [NULL, 2] => null (no definite overlap, but nulls present)

native/spark-expr/src/bloom_filter/spark_bit_array.rs

@@ -77,8 +77,8 @@ impl SparkBitArray {
     pub fn merge_be_words(&mut self, incoming: &[u8]) {
         debug_assert_eq!(self.data.len() * 8, incoming.len());
         let mut bit_count: usize = 0;
-        for (word, chunk) in self.data.iter_mut().zip(incoming.chunks_exact(8)) {
-            *word |= u64::from_be_bytes(chunk.try_into().unwrap());
+        for (word, chunk) in self.data.iter_mut().zip(incoming.as_chunks::<8>().0) {
+            *word |= u64::from_be_bytes(*chunk);
             bit_count += word.count_ones() as usize;
         }
         self.bit_count = bit_count;

native/spark-expr/src/conversion_funcs/boolean.rs

@@ -15,11 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::{SparkError, SparkResult};
-use arrow::array::{Array, ArrayRef, AsArray, Decimal128Array, TimestampMicrosecondBuilder};
+use crate::SparkResult;
+use arrow::array::{Array, ArrayRef, AsArray, TimestampMicrosecondBuilder};
 use arrow::datatypes::DataType;
 use std::sync::Arc;
 
+/// Boolean -> Decimal is intentionally absent: it has no native path and is routed through the
+/// JVM codegen dispatcher on the Scala side (see `CometCast.canCastFromBoolean`).
 pub fn is_df_cast_from_bool_spark_compatible(to_type: &DataType) -> bool {
     use DataType::*;
     matches!(
@@ -28,35 +30,6 @@ pub fn is_df_cast_from_bool_spark_compatible(to_type: &DataType) -> bool {
     )
 }
 
-pub fn cast_boolean_to_decimal(
-    array: &ArrayRef,
-    precision: u8,
-    scale: i8,
-) -> SparkResult<ArrayRef> {
-    let bool_array = array.as_boolean();
-    let scaled_val = 10_i128.pow(scale as u32);
-    let result: Decimal128Array = bool_array
-        .iter()
-        .map(|v| v.map(|b| if b { scaled_val } else { 0 }))
-        .collect();
-
-    // Convert Arrow decimal overflow errors to SparkError
-    let decimal_array = result
-        .with_precision_and_scale(precision, scale)
-        .map_err(|e| {
-            if matches!(e, arrow::error::ArrowError::InvalidArgumentError(_))
-                && e.to_string().contains("too large to store in a Decimal128")
-            {
-                // Use the scaled value as it's the only non-zero value that could overflow
-                crate::error::decimal_overflow_error(scaled_val, precision, scale)
-            } else {
-                SparkError::Arrow(Arc::new(e))
-            }
-        })?;
-
-    Ok(Arc::new(decimal_array))
-}
-
 pub(crate) fn cast_boolean_to_timestamp(
     array_ref: &ArrayRef,
     target_tz: &Option<Arc<str>>,
@@ -212,20 +185,20 @@ mod tests {
     }
 
     #[test]
-    fn test_bool_to_decimal_cast() {
-        let result = cast_array(
+    fn test_bool_to_decimal_cast_is_not_supported() {
+        // Boolean -> Decimal has no native path; the Scala planner routes it through the JVM
+        // codegen dispatcher, so reaching the native cast at all is a bug.
+        let err = cast_array(
             test_input_bool_array(),
             &Decimal128(10, 4),
             &test_input_spark_opts(),
         )
-        .unwrap();
-        let expected_arr = Decimal128Array::from(vec![10000_i128, 0_i128])
-            .with_precision_and_scale(10, 4)
-            .unwrap();
-        let arr = result.as_any().downcast_ref::<Decimal128Array>().unwrap();
-        assert_eq!(arr.value(0), expected_arr.value(0));
-        assert_eq!(arr.value(1), expected_arr.value(1));
-        assert!(arr.is_null(2));
+        .expect_err("expected boolean -> decimal to be unsupported");
+        assert!(
+            err.to_string()
+                .contains("Native cast invoked for unsupported cast"),
+            "unexpected error: {err}"
+        );
     }
 
     #[test]

native/spark-expr/src/conversion_funcs/cast.rs

@@ -16,7 +16,7 @@
 // under the License.
 
 use crate::conversion_funcs::boolean::{
-    cast_boolean_to_decimal, cast_boolean_to_timestamp, is_df_cast_from_bool_spark_compatible,
+    cast_boolean_to_timestamp, is_df_cast_from_bool_spark_compatible,
 };
 use crate::conversion_funcs::numeric::{
     cast_decimal128_to_utf8, cast_decimal_to_timestamp, cast_float32_to_decimal128,
@@ -438,9 +438,6 @@ pub(crate) fn cast_array(
         (Int64, Binary) if (eval_mode == Legacy) => {
             cast_whole_num_to_binary!(&array, Int64Array, 8)
         }
-        (Boolean, Decimal128(precision, scale)) => {
-            cast_boolean_to_decimal(&array, *precision, *scale)
-        }
         (Int8 | Int16 | Int32 | Int64, Timestamp(_, tz)) => cast_int_to_timestamp(&array, tz),
         (Float32 | Float64, Timestamp(_, tz)) => cast_float_to_timestamp(&array, tz, eval_mode),
         (Boolean, Timestamp(_, tz)) => cast_boolean_to_timestamp(&array, tz),

native/spark-expr/src/conversion_funcs/mod.rs

@@ -20,5 +20,7 @@ pub mod cast;
 mod numeric;
 mod string;
 mod temporal;
-mod trim;
+pub(crate) mod trim;
 mod utils;
+
+pub(crate) use string::ymd_to_epoch_day;

native/spark-expr/src/conversion_funcs/string.rs

@@ -1234,7 +1234,7 @@ fn is_leap_year(year: i64) -> bool {
 /// Days since 1970-01-01 for a proleptic Gregorian year/month/day, or `None` when the
 /// combination is not a real calendar date. Unlike `NaiveDate::from_ymd_opt`, this accepts
 /// any year that fits in `i64`.
-fn ymd_to_epoch_day(year: i64, month: i64, day: i64) -> Option<i64> {
+pub(crate) fn ymd_to_epoch_day(year: i64, month: i64, day: i64) -> Option<i64> {
     const DAYS_IN_MONTH: [i64; 12] = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31];
     let mut max_day = *DAYS_IN_MONTH.get(usize::try_from(month.checked_sub(1)?).ok()?)?;
     if month == 2 && is_leap_year(year) {

native/spark-expr/src/conversion_funcs/trim.rs

@@ -31,9 +31,11 @@
 //! `Double.parseDouble` trims before parsing, and `Decimal.stringToJavaBigDecimal` does
 //! `str.toString.trim`.
 //!
+//! `to_time` and `try_to_time` also use [`trim_all`] after detecting an optional AM/PM suffix;
+//! suffix detection itself first removes ASCII spaces only, matching Spark's `stringToTime`.
+//!
 //! \* `timestamp` and `timestamp_ntz` are listed for what Spark does; the Comet parsers for those
-//! two targets have not been migrated to these helpers and still use `str::trim`, as do `to_time`
-//! and `try_to_time` in `datetime_funcs::to_time`
+//! two targets have not been migrated to these helpers and still use `str::trim`
 //! (<https://github.com/apache/datafusion-comet/issues/5149>).
 
 /// True for the bytes trimmed by `org.apache.spark.unsafe.types.UTF8String.trimAll`, i.e. the