Match Spark 4.2 date_trunc overflow at Long.MinValue [databricks] (NVIDIA#15416)

Fixes NVIDIA#15382.

### Description

- Match Spark 4.2 (`SPARK-56663`) `date_trunc` semantics at
`Long.MinValue` micros: truncation underflow must raise
`ArithmeticException` instead of returning a wrapped positive timestamp
on GPU.
- Route the check through `TruncTimestampShims` so Spark 4.2 enables the
overflow check while older supported Spark lines keep the previous no-op
behavior, without runtime version compares in common code.
- Add `test_date_trunc_long_min_value_overflow` for `YEAR` and
`MILLISECOND` on Spark 4.2+, comparing CPU/GPU exception behavior and
validating the GPU plan includes `GpuProjectExec`.
- Validated with:
- `mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml -pl
sql-plugin -Dbuildver=420 -Dcuda.version=cuda13 -DskipTests clean
compile`
- `mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml -pl
sql-plugin -Dbuildver=413 -Dcuda.version=cuda13 -DskipTests clean
compile`
- `mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml
-Dbuildver=420 -Dcuda.version=cuda13 -DskipTests validate`
- `mvn -s /home/liangcail/.m2/settings_art.xml -f scala2.13/pom.xml
-Dbuildver=413 -Dcuda.version=cuda13 -DskipTests validate`

### Checklists

Documentation
- [ ] Updated for new or modified user-facing features or behaviors
- [x] No user-facing change

Testing
- [x] Added or modified tests to cover new code paths
- [ ] Covered by existing tests
(Please provide the names of the existing tests in the PR description.)
- [ ] Not required

Performance
- [ ] Tests ran and results are added in the PR description
- [ ] Issue filed with a link in the PR description
- [x] Not required

---------

Signed-off-by: Firestarman <firestarmanllc@gmail.com>

Author: firestarman

integration_tests/src/main/python/date_time_test.py

@@ -20,7 +20,7 @@
 from dateutil import tz
 from marks import allow_non_gpu, approximate_float, datagen_overrides, disable_ansi_mode, ignore_order, incompat, tz_sensitive_test
 from pyspark.sql.types import *
-from spark_session import with_cpu_session, is_before_spark_350, is_before_spark_400
+from spark_session import with_cpu_session, is_before_spark_350, is_before_spark_400, is_spark_420_or_later
 import pyspark.sql.functions as f
 from timezones import all_timezones, fixed_offset_timezones, fixed_offset_timezones_iana, variable_offset_timezones, variable_offset_timezones_iana
 
@@ -1174,3 +1174,37 @@ def test_trunc_timestamp_single_format(data_gen):
             'date_trunc("MILLISECOND", a)',
             'date_trunc("MICROSECOND", a)',
             'date_trunc("invalid", a)'))
+
+_LONG_MIN_TIMESTAMP_MICROS = -9223372036854775808
+
+# SPARK-56663: Spark 4.2+ throws ArithmeticException for date_trunc at Long.MinValue micros.
+_date_trunc_long_min_overflow_formats = [
+    pytest.param('YEAR', id='YEAR'),
+    pytest.param('MILLISECOND', id='MILLISECOND'),
+]
+
+@allow_non_gpu(*non_utc_tz_allow)
+@pytest.mark.skipif(not is_spark_420_or_later(),
+                    reason='date_trunc Long.MinValue overflow is supported on Spark 4.2+')
+@pytest.mark.parametrize('trunc_format', _date_trunc_long_min_overflow_formats)
+def test_date_trunc_long_min_value_overflow(trunc_format):
+    def run(spark):
+        spark.conf.set('spark.rapids.sql.test.validateExecsInGpuPlan', 'GpuProjectExec')
+        return spark.sql(
+            "select date_trunc('{0}', timestamp_micros({1}L))".format(
+                trunc_format, _LONG_MIN_TIMESTAMP_MICROS)).collect()
+    assert_gpu_and_cpu_error(run, conf={}, error_message='ArithmeticException')
+
+@allow_non_gpu(*non_utc_tz_allow)
+@pytest.mark.skipif(not is_spark_420_or_later(),
+                    reason='date_trunc Long.MinValue overflow is supported on Spark 4.2+')
+def test_date_trunc_long_min_value_overflow_column_format():
+    # Exercise the scalar-timestamp / column-format overload so overflow checks compare
+    # equal-length columns instead of a one-row timestamp against a multi-row result.
+    def run(spark):
+        spark.conf.set('spark.rapids.sql.test.validateExecsInGpuPlan', 'GpuProjectExec')
+        return spark.sql(
+            "select date_trunc(fmt, timestamp_micros({0}L)) "
+            "from values ('YEAR'), ('MILLISECOND') as t(fmt)".format(
+                _LONG_MIN_TIMESTAMP_MICROS)).collect()
+    assert_gpu_and_cpu_error(run, conf={}, error_message='ArithmeticException')

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/datetimeExpressions.scala

@@ -28,7 +28,8 @@ import com.nvidia.spark.rapids.ExprMeta
 import com.nvidia.spark.rapids.GpuOverrides.{extractStringLit, getTimeParserPolicy}
 import com.nvidia.spark.rapids.RapidsPluginImplicits._
 import com.nvidia.spark.rapids.jni.{Arithmetic, CastStrings, DateTimeUtils, GpuTimeZoneDB}
-import com.nvidia.spark.rapids.shims.{NullIntolerantShim, ShimBinaryExpression, ShimExpression}
+import com.nvidia.spark.rapids.shims.{NullIntolerantShim, ShimBinaryExpression, ShimExpression,
+  TruncTimestampShims}
 
 import org.apache.spark.sql.catalyst.expressions.{BinaryExpression, ExpectsInputTypes, Expression, FromUnixTime, FromUTCTimestamp, ImplicitCastInputTypes, MonthsBetween, TimeZoneAwareExpression, ToUTCTimestamp, TruncDate, TruncTimestamp}
 import org.apache.spark.sql.catalyst.util.DateTimeConstants
@@ -1534,6 +1535,59 @@ case class GpuTruncTimestamp(fmt: Expression, timestamp: Expression, timeZoneId:
 
   override def prettyName: String = "date_trunc"
 
+  override protected def truncate(datetimeCol: GpuColumnVector, fmtCol: GpuColumnVector)
+      : ColumnVector = {
+    closeOnExcept(DateTimeUtils.truncate(datetimeCol.getBase, fmtCol.getBase)) { truncated =>
+      TruncTimestampShims.checkOverflow(datetimeCol.getBase, truncated)
+      truncated
+    }
+  }
+
+  override protected def truncate(datetimeVal: GpuScalar, fmtCol: GpuColumnVector): ColumnVector = {
+    // Keep the one-row scalar broadcast into JNI truncate, and compare the multi-row result
+    // against the scalar via TruncTimestampShims.checkOverflow(Scalar, ColumnVector).
+    withResource(ColumnVector.fromScalar(datetimeVal.getBase, 1)) { datetimeCol =>
+      closeOnExcept(DateTimeUtils.truncate(datetimeCol, fmtCol.getBase)) { truncated =>
+        TruncTimestampShims.checkOverflow(datetimeVal.getBase, truncated)
+        truncated
+      }
+    }
+  }
+
+  override protected def truncate(datetimeCol: GpuColumnVector, fmtVal: GpuScalar): ColumnVector = {
+    fmtStr match {
+      case Some(fmt) =>
+        closeOnExcept(DateTimeUtils.truncate(datetimeCol.getBase, fmt)) { truncated =>
+          TruncTimestampShims.checkOverflow(datetimeCol.getBase, truncated)
+          truncated
+        }
+      case None =>
+        GpuColumnVector.columnVectorFromNull(datetimeCol.getRowCount.toInt, dataType)
+    }
+  }
+
+  override protected def truncate(numRows: Int, datetimeVal: GpuScalar, fmtVal: GpuScalar)
+      : ColumnVector = {
+    fmtStr match {
+      case Some(fmt) =>
+        withResource(ColumnVector.fromScalar(datetimeVal.getBase, 1)) { datetimeCol =>
+          closeOnExcept(DateTimeUtils.truncate(datetimeCol, fmt)) { truncated =>
+            TruncTimestampShims.checkOverflow(datetimeVal.getBase, truncated)
+            if (numRows == 1) {
+              truncated
+            } else {
+              withResource(truncated) { _ =>
+                withResource(truncated.getScalarElement(0)) { truncatedScalar =>
+                  ColumnVector.fromScalar(truncatedScalar, numRows)
+                }
+              }
+            }
+          }
+        }
+      case None => GpuColumnVector.columnVectorFromNull(numRows, dataType)
+    }
+  }
+
   // Since the input order of this class is opposite compared to the `GpuTruncDate` class,
   // we need to switch `lhs` and `rhs` in the `doColumnar` methods below.
 

sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/TruncTimestampShims.scala

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+/*** spark-rapids-shim-json-lines
+{"spark": "330"}
+{"spark": "331"}
+{"spark": "332"}
+{"spark": "333"}
+{"spark": "334"}
+{"spark": "340"}
+{"spark": "341"}
+{"spark": "342"}
+{"spark": "343"}
+{"spark": "344"}
+{"spark": "350"}
+{"spark": "350db143"}
+{"spark": "351"}
+{"spark": "352"}
+{"spark": "353"}
+{"spark": "354"}
+{"spark": "355"}
+{"spark": "356"}
+{"spark": "357"}
+{"spark": "358"}
+{"spark": "359"}
+{"spark": "400"}
+{"spark": "400db173"}
+{"spark": "401"}
+{"spark": "402"}
+{"spark": "403"}
+{"spark": "404"}
+{"spark": "411"}
+{"spark": "412"}
+{"spark": "413"}
+spark-rapids-shim-json-lines ***/
+
+package com.nvidia.spark.rapids.shims
+
+import ai.rapids.cudf.{ColumnVector, Scalar}
+
+object TruncTimestampShims {
+  def checkOverflow(datetimeCol: ColumnVector, truncated: ColumnVector): Unit = {}
+
+  def checkOverflow(datetime: Scalar, truncated: ColumnVector): Unit = {}
+}

sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/TruncTimestampShims.scala

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed 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.
+ */
+/*** spark-rapids-shim-json-lines
+{"spark": "420"}
+spark-rapids-shim-json-lines ***/
+
+package com.nvidia.spark.rapids.shims
+
+import ai.rapids.cudf.{ColumnVector, Scalar}
+import com.nvidia.spark.rapids.Arm.withResource
+
+object TruncTimestampShims {
+  /**
+   * SPARK-56663 uses checked subtraction and throws when truncation would underflow. The JNI
+   * kernel uses unchecked chrono arithmetic, so detect a wrapped result by verifying that
+   * truncation never moves a timestamp forward.
+   */
+  def checkOverflow(datetimeCol: ColumnVector, truncated: ColumnVector): Unit = {
+    withResource(truncated.greaterThan(datetimeCol)) { overflow =>
+      checkAnyOverflow(overflow)
+    }
+  }
+
+  def checkOverflow(datetime: Scalar, truncated: ColumnVector): Unit = {
+    withResource(truncated.greaterThan(datetime)) { overflow =>
+      checkAnyOverflow(overflow)
+    }
+  }
+
+  private def checkAnyOverflow(overflow: ColumnVector): Unit = {
+    withResource(overflow.any()) { any =>
+      if (any.isValid && any.getBoolean) {
+        throw new ArithmeticException("long overflow")
+      }
+    }
+  }
+}