Fix non-UTC ORC timestamp reads

Match Apache ORC timezone, rounding, and historical offset semantics for physical timestamps and schema-evolution casts.

Fixes NVIDIA#15449.

Signed-off-by: Chong Gao <chongg@nvidia.com>

Author: Chong Gao

integration_tests/src/main/python/orc_cast_test.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2020-2025, NVIDIA CORPORATION.
+# Copyright (c) 2020-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.
@@ -128,6 +128,39 @@ def test_casting_from_double_to_timestamp(spark_tmp_path, data_gen):
     )
 
 
+@pytest.mark.skipif(not is_not_utc(), reason="non-UTC ORC timestamp regression test")
+@allow_non_gpu(*non_utc_allow_orc_scan)
+def test_non_utc_timestamp_regressions(spark_tmp_path):
+    integer_path = spark_tmp_path + '/orc_integer_timestamp_regression'
+    double_path = spark_tmp_path + '/orc_double_timestamp_regression'
+    physical_path = spark_tmp_path + '/orc_physical_timestamp_regression'
+
+    with_cpu_session(
+        lambda spark: spark.createDataFrame([(514952012,)], "a long")
+            .write.orc(integer_path)
+    )
+    with_cpu_session(
+        lambda spark: spark.createDataFrame(
+            [(0.0,), (-8589934591.999999,), (-7953731124.723491,)], "a double")
+            .write.orc(double_path)
+    )
+    with_cpu_session(
+        lambda spark: spark.range(1)
+            .selectExpr("timestamp_micros(-2957649381472612L) AS a")
+            .write.orc(physical_path)
+    )
+
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: spark.read.schema("a timestamp").orc(integer_path)
+    )
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: spark.read.schema("a timestamp").orc(double_path)
+    )
+    assert_gpu_and_cpu_are_equal_collect(
+        lambda spark: spark.read.orc(physical_path)
+    )
+
+
 @allow_non_gpu(*non_utc_allow_for_test_casting_from_overflow_long)
 def test_casting_from_overflow_double_to_timestamp(spark_tmp_path):
     orc_path = spark_tmp_path + '/orc_casting_from_overflow_double_to_timestamp'

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOrcScan.scala

@@ -322,8 +322,7 @@ object GpuOrcScan {
       case (DType.BOOL8 | DType.INT8 | DType.INT16 | DType.INT32 | DType.INT64,
       DType.TIMESTAMP_MICROSECONDS) =>
         withResource(OrcCastingShims.castIntegerToTimestamp(col, fromDt)) { timestamp =>
-          GpuTimeZoneDB.fromTimestampToUtcTimestamp(
-            timestamp, ZoneId.systemDefault().normalized())
+          GpuTimeZoneDB.convertOrcFromUtc(timestamp, ZoneId.systemDefault().getId)
         }
 
       // float to bool/integral
@@ -382,16 +381,19 @@ object GpuOrcScan {
         // Math.round half up can be implemented in terms of floor
         // Math.round(x) = n iff x is in [n-0.5, n+0.5) iff x+0.5 is in [n,n+1) iff floor(x+0.5) = n
         //
-        val milliseconds = withResource(Scalar.fromDouble(DateTimeConstants.MILLIS_PER_SECOND)) {
-          thousand =>
-          // ORC assumes value is in seconds
-          withResource(col.mul(thousand, DType.FLOAT64)) { doubleMillis =>
-            withResource(Scalar.fromDouble(0.5)) { half =>
-              withResource(doubleMillis.add(half)) { doubleMillisPlusHalf =>
-                withResource(doubleMillisPlusHalf.floor()) { millis =>
-                  withResource(getOverflowFlags(doubleMillis, millis)) { overflowFlags =>
-                    withResource(Scalar.fromNull(millis.getType)) { nullVal =>
-                      overflowFlags.ifElse(millis, nullVal)
+        val milliseconds = withResource(col.castTo(DType.FLOAT64)) { doubleSeconds =>
+          withResource(convertOrcFloatingPointSeconds(doubleSeconds)) { convertedSeconds =>
+            withResource(Scalar.fromDouble(DateTimeConstants.MILLIS_PER_SECOND)) { thousand =>
+              // ORC applies timezone conversion while the value is still in seconds.
+              withResource(convertedSeconds.mul(thousand, DType.FLOAT64)) { doubleMillis =>
+                withResource(Scalar.fromDouble(0.5)) { half =>
+                  withResource(doubleMillis.add(half)) { doubleMillisPlusHalf =>
+                    withResource(doubleMillisPlusHalf.floor()) { millis =>
+                      withResource(getOverflowFlags(doubleMillis, millis)) { overflowFlags =>
+                        withResource(Scalar.fromNull(millis.getType)) { nullVal =>
+                          overflowFlags.ifElse(millis, nullVal)
+                        }
+                      }
                     }
                   }
                 }
@@ -419,12 +421,11 @@ object GpuOrcScan {
           }
           withResource(Scalar.fromDouble(DateTimeConstants.MICROS_PER_MILLIS)) { thousand =>
             withResource(milliseconds.mul(thousand)) { microseconds =>
-                withResource(microseconds.castTo(DType.INT64)) { longVec =>
-                  withResource(longVec.castTo(DType.TIMESTAMP_MICROSECONDS)) { timestamp =>
-                    GpuTimeZoneDB.fromTimestampToUtcTimestamp(
-                      timestamp, ZoneId.systemDefault().normalized())
-                  }
+              withResource(microseconds.castTo(DType.INT64)) { longVec =>
+                withResource(longVec.castTo(DType.TIMESTAMP_MICROSECONDS)) { timestamp =>
+                  timestamp.incRefCount()
                 }
+              }
             }
           }
         }
@@ -444,6 +445,35 @@ object GpuOrcScan {
     }
   }
 
+  /**
+   * Apply Spark's java.time timezone rules before ORC's millisecond rounding and overflow check.
+   * Looking up the offset with a microsecond timestamp preserves the original floating-point
+   * value: only the integral timezone delta is taken from the converted timestamp.
+   */
+  private def convertOrcFloatingPointSeconds(seconds: ColumnView): ColumnVector = {
+    withResource(Scalar.fromDouble(DateTimeConstants.MICROS_PER_SECOND)) { microsPerSecond =>
+      withResource(seconds.mul(microsPerSecond, DType.FLOAT64)) { doubleMicros =>
+        withResource(doubleMicros.castTo(DType.INT64)) { localMicros =>
+          withResource(localMicros.castTo(DType.TIMESTAMP_MICROSECONDS)) { localTimestamp =>
+            withResource(GpuTimeZoneDB.fromTimestampToUtcTimestamp(
+                localTimestamp, ZoneId.systemDefault().normalized())) { utcTimestamp =>
+              withResource(utcTimestamp.castTo(DType.INT64)) { utcMicros =>
+                withResource(utcMicros.sub(localMicros)) { offsetMicros =>
+                  withResource(offsetMicros.castTo(DType.FLOAT64)) { doubleOffsetMicros =>
+                    withResource(doubleOffsetMicros.div(microsPerSecond, DType.FLOAT64)) {
+                      offsetSeconds =>
+                      seconds.add(offsetSeconds, DType.FLOAT64)
+                    }
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
   /**
    * Whether the type casting is supported by GPU ORC reading.
    *

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOrcTimezoneUtils.scala

@@ -21,7 +21,7 @@ import java.util.Optional
 
 import scala.collection.mutable.ArrayBuffer
 
-import ai.rapids.cudf.{ColumnView, DType, Table}
+import ai.rapids.cudf.{ColumnView, DType, Scalar, Table}
 import com.nvidia.spark.rapids.Arm.withResource
 import com.nvidia.spark.rapids.RapidsPluginImplicits.AutoCloseableProducingSeq
 import com.nvidia.spark.rapids.jni.GpuTimeZoneDB
@@ -77,16 +77,17 @@ object GpuOrcTimezoneUtils {
    */
   private def rebaseWithWriterTimezone(
       input: Table, writerTz: String, readerTz: String): Table = {
+    val readerZone = ZoneId.of(readerTz, ZoneId.SHORT_IDS)
     withResource(input) { _ =>
       withResource(GpuTimeZoneDB.buildOrcTimezoneContext(writerTz, readerTz)) { tzCtx =>
         val newColumns = (0 until input.getNumberOfColumns).safeMap { colIdx =>
           val col = input.getColumn(colIdx)
           val dType = col.getType
           if (dType.hasTimeResolution) {
-            GpuTimeZoneDB.convertOrcTimezones(col, tzCtx)
+            convertOrcTimestamp(col, tzCtx, readerZone)
           } else if (dType == DType.LIST || dType == DType.STRUCT) {
             withResource(new ArrayBuffer[ColumnView]) { toClose =>
-              val rebased = rebaseNestedWithWriterTimezone(col, tzCtx, toClose)
+              val rebased = rebaseNestedWithWriterTimezone(col, tzCtx, readerZone, toClose)
               if (rebased eq col) {
                 col.incRefCount()
               } else {
@@ -105,18 +106,61 @@ object GpuOrcTimezoneUtils {
     }
   }
 
+  /**
+   * Match the full Spark ORC timestamp path. Apache ORC uses java.util.TimeZone while decoding,
+   * but Spark materializes the resulting java.sql.Timestamp using java.time rules. Those rule
+   * sets can differ for historical and projected timestamps.
+   */
+  private def convertOrcTimestamp(
+      col: ColumnView,
+      tzCtx: GpuTimeZoneDB.OrcTimezoneContext,
+      readerZone: ZoneId): ai.rapids.cudf.ColumnVector = {
+    withResource(GpuTimeZoneDB.convertOrcTimezones(col, tzCtx)) { orcTimestamp =>
+      val firstTransitionUs = tzCtx.getReaderFirstTransitionUs
+      if (firstTransitionUs == Long.MinValue) {
+        orcTimestamp.incRefCount()
+      } else {
+        withResource(GpuTimeZoneDB.convertOrcFromUtc(orcTimestamp, tzCtx)) { utilUtc =>
+          withResource(GpuTimeZoneDB.fromTimestampToUtcTimestamp(
+              orcTimestamp, readerZone.normalized())) { zoneUtc =>
+            withResource(orcTimestamp.castTo(DType.INT64)) { orcMicros =>
+              withResource(utilUtc.castTo(DType.INT64)) { utilMicros =>
+                withResource(zoneUtc.castTo(DType.INT64)) { zoneMicros =>
+                  withResource(zoneMicros.sub(utilMicros)) { ruleCorrection =>
+                    withResource(orcMicros.add(ruleCorrection)) { corrected =>
+                      withResource(corrected.castTo(DType.TIMESTAMP_MICROSECONDS)) {
+                        correctedTimestamp =>
+                        withResource(Scalar.timestampFromLong(
+                            DType.TIMESTAMP_MICROSECONDS, firstTransitionUs)) { firstTransition =>
+                          withResource(orcTimestamp.lessThan(firstTransition)) { needsCorrection =>
+                            needsCorrection.ifElse(correctedTimestamp, orcTimestamp)
+                          }
+                        }
+                      }
+                    }
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+
   private def rebaseNestedWithWriterTimezone(
       col: ColumnView,
       tzCtx: GpuTimeZoneDB.OrcTimezoneContext,
+      readerZone: ZoneId,
       toClose: ArrayBuffer[ColumnView]): ColumnView = {
     val addToClose = (v: ColumnView) => { toClose += v; v }
     val dType = col.getType
 
     if (dType.hasTimeResolution) {
-      GpuTimeZoneDB.convertOrcTimezones(col, tzCtx)
+      convertOrcTimestamp(col, tzCtx, readerZone)
     } else if (dType == DType.LIST) {
       val child = addToClose(col.getChildColumnView(0))
-      val newChild = rebaseNestedWithWriterTimezone(child, tzCtx, toClose)
+      val newChild = rebaseNestedWithWriterTimezone(child, tzCtx, readerZone, toClose)
       if (newChild ne child) {
         col.replaceListChild(addToClose(newChild))
       } else {
@@ -125,7 +169,7 @@ object GpuOrcTimezoneUtils {
     } else if (dType == DType.STRUCT) {
       val newViews = (0 until col.getNumChildren).map { i =>
         val child = addToClose(col.getChildColumnView(i))
-        val newChild = rebaseNestedWithWriterTimezone(child, tzCtx, toClose)
+        val newChild = rebaseNestedWithWriterTimezone(child, tzCtx, readerZone, toClose)
         if (newChild ne child) addToClose(newChild)
         newChild
       }