fix: coerce timestamp string predicates with Spark semantics

crates/sail-common-datafusion/Cargo.toml

@@ -10,14 +10,14 @@ workspace = true
 sail-common = { path = "../sail-common" }
 
 chrono = { workspace = true }
-ryu = { workspace = true }
 datafusion = { workspace = true }
 datafusion-common = { workspace = true }
 datafusion-expr = { workspace = true }
 arrow-schema = { workspace = true }
 thiserror = { workspace = true }
 either = { workspace = true }
 lexical-core = { workspace = true }
+num-bigint = { workspace = true }
 futures = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }

crates/sail-common-datafusion/src/display.rs

@@ -494,21 +494,23 @@ macro_rules! primitive_display {
     };
 }
 
+/// Formats a `f64` with the notation used by Java `Double.toString` and Spark casts.
+pub fn spark_f64_to_string(value: f64) -> String {
+    crate::java_float::format_f64(value)
+}
+
+/// Formats an `f32` with the notation used by Java `Float.toString` and Spark casts.
+pub fn spark_f32_to_string(value: f32) -> String {
+    crate::java_float::format_f32(value)
+}
+
 macro_rules! primitive_display_float {
-    ($($t:ty),+) => {
+    ($($t:ty => $formatter:ident),+) => {
         $(impl<'a> DisplayIndex for &'a PrimitiveArray<$t>
         {
             fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
                 let value = self.value(idx);
-                let mut buffer = ryu::Buffer::new();
-                if value.is_infinite() {
-                    if !value.is_sign_positive() {
-                        f.write_str("-")?;
-                    }
-                    f.write_str("Infinity")?;
-                } else {
-                    f.write_str(buffer.format(value))?;
-                }
+                f.write_str(&$formatter(value))?;
                 Ok(())
             }
         })+
@@ -517,7 +519,10 @@ macro_rules! primitive_display_float {
 
 primitive_display!(Int8Type, Int16Type, Int32Type, Int64Type);
 primitive_display!(UInt8Type, UInt16Type, UInt32Type, UInt64Type);
-primitive_display_float!(Float32Type, Float64Type);
+primitive_display_float!(
+    Float32Type => spark_f32_to_string,
+    Float64Type => spark_f64_to_string
+);
 
 impl DisplayIndex for &PrimitiveArray<Float16Type> {
     fn write(&self, idx: usize, f: &mut dyn Write) -> FormatResult {
@@ -1028,6 +1033,43 @@ mod tests {
         assert_eq!(TEST_CONST_OPTIONS.date_format, TimeFormat::Custom("foo"));
     }
 
+    #[test]
+    fn test_spark_float_to_string() {
+        assert_eq!(spark_f64_to_string(1e18), "1.0E18");
+        assert_eq!(spark_f64_to_string(1e-7), "1.0E-7");
+        assert_eq!(spark_f64_to_string(f64::from_bits(1)), "4.9E-324");
+        assert_eq!(spark_f64_to_string(-f64::from_bits(1)), "-4.9E-324");
+        assert_eq!(spark_f64_to_string(-0.0), "-0.0");
+        assert_eq!(spark_f32_to_string(10_000_000.0), "1.0E7");
+        assert_eq!(spark_f32_to_string(f32::from_bits(1)), "1.4E-45");
+        assert_eq!(spark_f32_to_string(-f32::from_bits(1)), "-1.4E-45");
+        assert_eq!(spark_f32_to_string(-0.0), "-0.0");
+        assert_eq!(
+            spark_f64_to_string(f64::from_bits(0x43d0_0000_0000_0000)),
+            "4.6116860184273879E18"
+        );
+        assert_eq!(
+            spark_f64_to_string(f64::from_bits(0xc36c_415f_f381_377c)),
+            "-6.3625437687430112E16"
+        );
+        assert_eq!(
+            spark_f64_to_string(f64::from_bits(0xc395_327f_1b7a_a2d1)),
+            "-3.8185574490758867E17"
+        );
+        assert_eq!(
+            spark_f32_to_string(f32::from_bits(0xd75b_54b0)),
+            "-2.41156777E14"
+        );
+        assert_eq!(
+            spark_f32_to_string(f32::from_bits(0x4d32_a0e4)),
+            "1.87305536E8"
+        );
+        assert_eq!(
+            spark_f32_to_string(f32::from_bits(0xddbd_6f50)),
+            "-1.70627712E18"
+        );
+    }
+
     #[expect(clippy::unwrap_used)]
     #[test]
     fn test_map_array_to_string() {

crates/sail-common-datafusion/src/formatter.rs

@@ -8,6 +8,7 @@ use datafusion::arrow::temporal_conversions::{
     time32ms_to_time, time32s_to_time, time64ns_to_time, time64us_to_time,
 };
 use datafusion_common::arrow::temporal_conversions::{date32_to_datetime, date64_to_datetime};
+use sail_common::spec::{DayTimeIntervalField, YearMonthIntervalField};
 
 pub struct BinaryFormatter<'a>(pub &'a [u8]);
 
@@ -234,11 +235,45 @@ pub struct IntervalYearMonthFormatter(pub i32);
 
 impl Display for IntervalYearMonthFormatter {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        let months = self.0;
-        let years = months / 12;
-        let prepend = if years == 0 && months < 0 { "-" } else { "" };
-        let months = (months % 12).abs();
-        write!(f, "INTERVAL '{prepend}{years}-{months}' YEAR TO MONTH")
+        SparkYearMonthIntervalFormatter(
+            self.0,
+            YearMonthIntervalField::Year,
+            YearMonthIntervalField::Month,
+        )
+        .fmt(f)
+    }
+}
+
+pub struct SparkYearMonthIntervalFormatter(
+    pub i32,
+    pub YearMonthIntervalField,
+    pub YearMonthIntervalField,
+);
+
+impl Display for SparkYearMonthIntervalFormatter {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        let Self(value, start, end) = *self;
+        let magnitude = value.unsigned_abs();
+        let sign = if value < 0 { "-" } else { "" };
+        let body = match (start, end) {
+            (YearMonthIntervalField::Year, YearMonthIntervalField::Year) => {
+                (magnitude / 12).to_string()
+            }
+            (YearMonthIntervalField::Year, YearMonthIntervalField::Month) => {
+                format!("{}-{}", magnitude / 12, magnitude % 12)
+            }
+            (YearMonthIntervalField::Month, YearMonthIntervalField::Month) => magnitude.to_string(),
+            (YearMonthIntervalField::Month, YearMonthIntervalField::Year) => {
+                return Err(std::fmt::Error);
+            }
+        };
+        write_qualified_interval(
+            f,
+            sign,
+            &body,
+            year_month_field_name(start),
+            year_month_field_name(end),
+        )
     }
 }
 
@@ -357,24 +392,107 @@ pub struct DurationMicrosecondFormatter(pub i64);
 
 impl Display for DurationMicrosecondFormatter {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        let microseconds = self.0;
-        let days = microseconds / 86_400_000_000;
-        let prepend = if days == 0 && microseconds < 0 {
-            "-"
-        } else {
-            ""
+        SparkDayTimeIntervalFormatter(
+            self.0,
+            DayTimeIntervalField::Day,
+            DayTimeIntervalField::Second,
+        )
+        .fmt(f)
+    }
+}
+
+pub struct SparkDayTimeIntervalFormatter(
+    pub i64,
+    pub DayTimeIntervalField,
+    pub DayTimeIntervalField,
+);
+
+impl Display for SparkDayTimeIntervalFormatter {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        const MICROSECONDS_PER_SECOND: u64 = 1_000_000;
+        const MICROSECONDS_PER_MINUTE: u64 = 60 * MICROSECONDS_PER_SECOND;
+        const MICROSECONDS_PER_HOUR: u64 = 60 * MICROSECONDS_PER_MINUTE;
+        const MICROSECONDS_PER_DAY: u64 = 24 * MICROSECONDS_PER_HOUR;
+
+        let Self(value, start, end) = *self;
+        if start > end {
+            return Err(std::fmt::Error);
+        }
+        let mut magnitude = value.unsigned_abs();
+        let sign = if value < 0 { "-" } else { "" };
+        let leading_unit = match start {
+            DayTimeIntervalField::Day => MICROSECONDS_PER_DAY,
+            DayTimeIntervalField::Hour => MICROSECONDS_PER_HOUR,
+            DayTimeIntervalField::Minute => MICROSECONDS_PER_MINUTE,
+            DayTimeIntervalField::Second => MICROSECONDS_PER_SECOND,
         };
-        let hours = ((microseconds % 86_400_000_000) / 3_600_000_000).abs();
-        let minutes = ((microseconds % 3_600_000_000) / 60_000_000).abs();
-        let seconds = ((microseconds % 60_000_000) / 1_000_000).abs();
-        let fraction = FractionFormatter::<6>((microseconds % 1_000_000).unsigned_abs() as u32);
-        write!(
+        let leading = magnitude / leading_unit;
+        magnitude %= leading_unit;
+        let mut body = leading.to_string();
+
+        if start < DayTimeIntervalField::Hour && end >= DayTimeIntervalField::Hour {
+            write!(body, " {:02}", magnitude / MICROSECONDS_PER_HOUR)?;
+            magnitude %= MICROSECONDS_PER_HOUR;
+        }
+        if start < DayTimeIntervalField::Minute && end >= DayTimeIntervalField::Minute {
+            write!(body, ":{:02}", magnitude / MICROSECONDS_PER_MINUTE)?;
+            magnitude %= MICROSECONDS_PER_MINUTE;
+        }
+        if start < DayTimeIntervalField::Second && end == DayTimeIntervalField::Second {
+            write!(
+                body,
+                ":{:02}{}",
+                magnitude / MICROSECONDS_PER_SECOND,
+                FractionFormatter::<6>((magnitude % MICROSECONDS_PER_SECOND) as u32)
+            )?;
+        } else if start == DayTimeIntervalField::Second {
+            write!(
+                body,
+                "{}",
+                FractionFormatter::<6>((magnitude % MICROSECONDS_PER_SECOND) as u32)
+            )?;
+        }
+
+        write_qualified_interval(
             f,
-            "INTERVAL '{prepend}{days} {hours:02}:{minutes:02}:{seconds:02}{fraction}' DAY TO SECOND",
+            sign,
+            &body,
+            day_time_field_name(start),
+            day_time_field_name(end),
         )
     }
 }
 
+fn write_qualified_interval(
+    f: &mut Formatter<'_>,
+    sign: &str,
+    body: &str,
+    start: &str,
+    end: &str,
+) -> std::fmt::Result {
+    if start == end {
+        write!(f, "INTERVAL '{sign}{body}' {start}")
+    } else {
+        write!(f, "INTERVAL '{sign}{body}' {start} TO {end}")
+    }
+}
+
+fn year_month_field_name(field: YearMonthIntervalField) -> &'static str {
+    match field {
+        YearMonthIntervalField::Year => "YEAR",
+        YearMonthIntervalField::Month => "MONTH",
+    }
+}
+
+fn day_time_field_name(field: DayTimeIntervalField) -> &'static str {
+    match field {
+        DayTimeIntervalField::Day => "DAY",
+        DayTimeIntervalField::Hour => "HOUR",
+        DayTimeIntervalField::Minute => "MINUTE",
+        DayTimeIntervalField::Second => "SECOND",
+    }
+}
+
 pub struct DurationNanosecondFormatter(pub i64);
 
 impl Display for DurationNanosecondFormatter {