fix: match Spark 4.2 date_trunc overflow for SECOND and MILLISECOND

docs/source/contributor-guide/expression-audits/datetime_funcs.md

@@ -35,6 +35,7 @@
 
 ## date_trunc
 
+- Spark 4.2.0 raises `long overflow` when SECOND or MILLISECOND truncation falls below the smallest timestamp, like every other unit (SPARK-56663). Earlier versions wrap the Long subtraction. The serde tells the native kernel which Spark it runs against, so on 4.2 those two units also keep dictionary key masking for values near the lower bound.
 - Performance (tuned 2026-10-06, PR [#5956](https://github.com/apache/datafusion-comet/pull/5956)): scalar timestamp truncation skips dictionary key masking for infallible fine units and safe UTC/NTZ values with substantial repetition. A values-only range check retains masking near the lower timestamp bound; non-UTC calendar paths and high-cardinality coarse batches retain masking to avoid errors and unnecessary calendar work on unused entries. About 19x faster for MICROSECOND with 65,536 keys and 32 distinct values. Matched Criterion coverage includes sparse/dense/all-NULL keys, high-cardinality YEAR batches, and used/unused overflow values. Benchmark: `benches/timestamp_trunc.rs`.
 
 ## dayname

native/core/src/execution/expressions/temporal.rs

@@ -158,7 +158,12 @@ impl ExpressionBuilder for TruncTimestampBuilder {
         let format = planner.create_expr(expr.format.as_ref().unwrap(), input_schema)?;
         let timezone = expr.timezone.clone();
 
-        Ok(Arc::new(TimestampTruncExpr::new(child, format, timezone)))
+        Ok(Arc::new(TimestampTruncExpr::new(
+            child,
+            format,
+            timezone,
+            expr.wrap_second_millisecond_overflow,
+        )))
     }
 }
 

native/proto/src/proto/expr.proto

@@ -597,6 +597,11 @@ message TruncTimestamp {
   Expr format = 1;
   Expr child = 2;
   string timezone = 3;
+  // Whether SECOND and MILLISECOND truncation wraps when the result would fall below the
+  // smallest timestamp. Spark before 4.2.0 wraps the Long subtraction; 4.2.0 (SPARK-56663)
+  // raises `long overflow` like every other unit. This tracks the Spark version Comet is
+  // running against.
+  bool wrap_second_millisecond_overflow = 4;
 }
 
 message Subquery {

native/spark-expr/benches/timestamp_trunc.rs

@@ -56,6 +56,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                 Arc::new(Column::new("a", 0)),
                 lit(format),
                 timezone.to_string(),
+                true,
             );
             for rows in ROW_COUNTS {
                 for (null_ratio, tag) in NULL_RATIOS.into_iter().chain([(0.875, "dense")]) {
@@ -110,6 +111,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                     Arc::new(Column::new("a", 0)),
                     lit(format),
                     "UTC".into(),
+                    true,
                 );
                 let cardinality_tag = if cardinality == rows { "high" } else { "low" };
                 group.bench_with_input(
@@ -136,8 +138,12 @@ fn criterion_benchmark(c: &mut Criterion) {
                 true,
             )]));
             let batch = RecordBatch::try_new(schema, vec![Arc::new(input)]).unwrap();
-            let expr =
-                TimestampTruncExpr::new(Arc::new(Column::new("a", 0)), lit("YEAR"), "UTC".into());
+            let expr = TimestampTruncExpr::new(
+                Arc::new(Column::new("a", 0)),
+                lit("YEAR"),
+                "UTC".into(),
+                true,
+            );
             assert_eq!(expr.evaluate(&batch).is_err(), referenced);
             group.bench_with_input(
                 BenchmarkId::from_parameter(format!("YEAR/{rows}/{tag}")),

native/spark-expr/src/datetime_funcs/timestamp_trunc.rs

@@ -41,20 +41,25 @@ pub struct TimestampTruncExpr {
     /// `Arc<str>` so it can be cheaply cloned onto Arrow `Timestamp` data types without
     /// reallocating, and parsed once into a `chrono::TimeZone` per batch.
     timezone: Arc<str>,
+    /// Whether SECOND and MILLISECOND truncation wraps below the smallest timestamp, as Spark
+    /// did before 4.2.0, instead of raising `long overflow` as 4.2.0 does (SPARK-56663).
+    wrap_second_millisecond_overflow: bool,
 }
 
 impl Hash for TimestampTruncExpr {
     fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
         self.child.hash(state);
         self.format.hash(state);
         self.timezone.hash(state);
+        self.wrap_second_millisecond_overflow.hash(state);
     }
 }
 impl PartialEq for TimestampTruncExpr {
     fn eq(&self, other: &Self) -> bool {
         self.child.eq(&other.child)
             && self.format.eq(&other.format)
             && self.timezone.eq(&other.timezone)
+            && self.wrap_second_millisecond_overflow == other.wrap_second_millisecond_overflow
     }
 }
 
@@ -63,11 +68,13 @@ impl TimestampTruncExpr {
         child: Arc<dyn PhysicalExpr>,
         format: Arc<dyn PhysicalExpr>,
         timezone: String,
+        wrap_second_millisecond_overflow: bool,
     ) -> Self {
         TimestampTruncExpr {
             child,
             format,
             timezone: Arc::from(timezone),
+            wrap_second_millisecond_overflow,
         }
     }
 }
@@ -100,6 +107,7 @@ impl PhysicalExpr for TimestampTruncExpr {
         let format = self.format.evaluate(batch)?;
         let output_type = output_type(&timestamp.data_type());
         let tz = &self.timezone;
+        let wrap = self.wrap_second_millisecond_overflow;
         let resolve_tz = |ts: ArrayRef| -> datafusion::common::Result<ArrayRef> {
             // For TimestampNTZ (Timestamp(Microsecond, None)), skip timezone conversion.
             // NTZ values are timezone-independent and truncation should operate directly on the
@@ -121,15 +129,16 @@ impl PhysicalExpr for TimestampTruncExpr {
         };
         match (timestamp, format) {
             (ColumnarValue::Array(ts), ColumnarValue::Scalar(Utf8(Some(format)))) => {
-                let result = timestamp_trunc_dyn(&resolve_tz(ts)?, format)?;
+                let result = timestamp_trunc_dyn(&resolve_tz(ts)?, format, wrap)?;
                 Ok(ColumnarValue::Array(relabel(result)?))
             }
             (ColumnarValue::Array(ts), ColumnarValue::Array(formats)) => {
                 let result = timestamp_trunc_array_fmt_dyn(&resolve_tz(ts)?, &formats)?;
                 Ok(ColumnarValue::Array(relabel(result)?))
             }
             (ColumnarValue::Scalar(ts_scalar), ColumnarValue::Scalar(Utf8(Some(format)))) => {
-                let result = timestamp_trunc_dyn(&resolve_tz(ts_scalar.to_array()?)?, format)?;
+                let result =
+                    timestamp_trunc_dyn(&resolve_tz(ts_scalar.to_array()?)?, format, wrap)?;
                 let scalar = ScalarValue::try_from_array(&relabel(result)?, 0)?;
                 Ok(ColumnarValue::Scalar(scalar))
             }
@@ -153,6 +162,7 @@ impl PhysicalExpr for TimestampTruncExpr {
             Arc::clone(&children[0]),
             Arc::clone(&self.format),
             self.timezone.to_string(),
+            self.wrap_second_millisecond_overflow,
         )))
     }
 }
@@ -195,6 +205,7 @@ mod tests {
             Arc::new(Column::new("ts", 0)),
             Arc::new(Literal::new(Utf8(Some("HOUR".to_string())))),
             "Asia/Kolkata".to_string(),
+            false,
         );
         let declared = expr.data_type(&schema).unwrap();
         let ColumnarValue::Array(result) = expr.evaluate(&batch).unwrap() else {