Make legacy date formatters independent from default JVM time zone

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateFormatter.scala

@@ -19,8 +19,7 @@ package org.apache.spark.sql.catalyst.util
 
 import java.text.SimpleDateFormat
 import java.time.{LocalDate, ZoneId}
-import java.time.format.DateTimeFormatter
-import java.util.{Date, Locale}
+import java.util.{Date, Locale, TimeZone}
 
 import org.apache.commons.lang3.time.FastDateFormat
 
@@ -83,31 +82,42 @@ class Iso8601DateFormatter(
 
 trait LegacyDateFormatter extends DateFormatter {
   def parseToDate(s: String): Date
+  def zoneId: ZoneId
 
   override def parse(s: String): Int = {
-    fromJavaDate(new java.sql.Date(parseToDate(s).getTime))
+    fromJavaDate(new java.sql.Date(parseToDate(s).getTime), TimeZone.getTimeZone(zoneId))
   }
 
   override def format(days: Int): String = {
-    format(DateTimeUtils.toJavaDate(days))
+    format(DateTimeUtils.toJavaDate(days, TimeZone.getTimeZone(zoneId)))
   }
 
   override def format(localDate: LocalDate): String = {
     format(localDateToDays(localDate))
   }
 }
 
-class LegacyFastDateFormatter(pattern: String, locale: Locale) extends LegacyDateFormatter {
+class LegacyFastDateFormatter(
+    pattern: String,
+    override val zoneId: ZoneId,
+    locale: Locale) extends LegacyDateFormatter {
   @transient
-  private lazy val fdf = FastDateFormat.getInstance(pattern, locale)
+  private lazy val fdf = FastDateFormat.getInstance(pattern, TimeZone.getTimeZone(zoneId), locale)
   override def parseToDate(s: String): Date = fdf.parse(s)
   override def format(d: Date): String = fdf.format(d)
   override def validatePatternString(): Unit = fdf
 }
 
-class LegacySimpleDateFormatter(pattern: String, locale: Locale) extends LegacyDateFormatter {
+class LegacySimpleDateFormatter(
+    pattern: String,
+    override val zoneId: ZoneId,
+    locale: Locale) extends LegacyDateFormatter {
   @transient
-  private lazy val sdf = new SimpleDateFormat(pattern, locale)
+  private lazy val sdf = {
+    val formatter = new SimpleDateFormat(pattern, locale)
+    formatter.setTimeZone(TimeZone.getTimeZone(zoneId))
+    formatter
+  }
   override def parseToDate(s: String): Date = sdf.parse(s)
   override def format(d: Date): String = sdf.format(d)
   override def validatePatternString(): Unit = sdf
@@ -143,9 +153,9 @@ object DateFormatter {
       legacyFormat: LegacyDateFormat): DateFormatter = {
     legacyFormat match {
       case FAST_DATE_FORMAT =>
-        new LegacyFastDateFormatter(pattern, locale)
+        new LegacyFastDateFormatter(pattern, zoneId, locale)
       case SIMPLE_DATE_FORMAT | LENIENT_SIMPLE_DATE_FORMAT =>
-        new LegacySimpleDateFormatter(pattern, locale)
+        new LegacySimpleDateFormatter(pattern, zoneId, locale)
     }
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala

@@ -103,8 +103,12 @@ object DateTimeUtils {
    * @return The number of days since epoch from java.sql.Date.
    */
   def fromJavaDate(date: Date): SQLDate = {
+    fromJavaDate(date, TimeZone.getDefault)
+  }
+
+  def fromJavaDate(date: Date, timeZone: TimeZone): SQLDate = {
     val millisUtc = date.getTime
-    val millisLocal = millisUtc + TimeZone.getDefault.getOffset(millisUtc)
+    val millisLocal = millisUtc + timeZone.getOffset(millisUtc)
     val julianDays = Math.toIntExact(Math.floorDiv(millisLocal, MILLIS_PER_DAY))
     rebaseJulianToGregorianDays(julianDays)
   }
@@ -123,9 +127,13 @@ object DateTimeUtils {
    * @return A `java.sql.Date` from number of days since epoch.
    */
   def toJavaDate(daysSinceEpoch: SQLDate): Date = {
+    toJavaDate(daysSinceEpoch, TimeZone.getDefault)
+  }
+
+  def toJavaDate(daysSinceEpoch: SQLDate, timeZone: TimeZone): Date = {
     val rebasedDays = rebaseGregorianToJulianDays(daysSinceEpoch)
     val localMillis = Math.multiplyExact(rebasedDays, MILLIS_PER_DAY)
-    val timeZoneOffset = TimeZone.getDefault match {
+    val timeZoneOffset = timeZone match {
       case zoneInfo: ZoneInfo => zoneInfo.getOffsetsByWall(localMillis, null)
       case timeZone: TimeZone => timeZone.getOffset(localMillis - timeZone.getRawOffset)
     }

sql/catalyst/src/test/scala/org/apache/spark/sql/util/DateFormatterSuite.scala

@@ -17,7 +17,8 @@
 
 package org.apache.spark.sql.util
 
-import java.time.{DateTimeException, LocalDate}
+import java.time.{DateTimeException, LocalDate, ZoneId}
+import java.util.{Calendar, TimeZone}
 
 import org.apache.spark.{SparkFunSuite, SparkUpgradeException}
 import org.apache.spark.sql.catalyst.plans.SQLHelper
@@ -28,26 +29,35 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy
 
 class DateFormatterSuite extends SparkFunSuite with SQLHelper {
-  test("parsing dates") {
-    outstandingTimezonesIds.foreach { timeZone =>
-      withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
-        val formatter = DateFormatter(getZoneId(timeZone))
-        val daysSinceEpoch = formatter.parse("2018-12-02")
-        assert(daysSinceEpoch === 17867)
+  private def withOutstandingZoneIds(f: ZoneId => Unit): Unit = {
+    for {
+      jvmZoneId <- outstandingZoneIds
+      sessionZoneId <- outstandingZoneIds
+    } {
+      withDefaultTimeZone(jvmZoneId) {
+        withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> sessionZoneId.getId) {
+          f(sessionZoneId)
+        }
       }
     }
   }
 
+  test("parsing dates") {
+    withOutstandingZoneIds { zoneId =>
+      val formatter = DateFormatter(zoneId)
+      val daysSinceEpoch = formatter.parse("2018-12-02")
+      assert(daysSinceEpoch === 17867)
+    }
+  }
+
   test("format dates") {
-    outstandingTimezonesIds.foreach { timeZone =>
-      withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
-        val formatter = DateFormatter(getZoneId(timeZone))
-        val (days, expected) = (17867, "2018-12-02")
-        val date = formatter.format(days)
-        assert(date === expected)
-        assert(formatter.format(daysToLocalDate(days)) === expected)
-        assert(formatter.format(toJavaDate(days)) === expected)
-      }
+    withOutstandingZoneIds { zoneId =>
+      val formatter = DateFormatter(zoneId)
+      val (days, expected) = (17867, "2018-12-02")
+      val date = formatter.format(days)
+      assert(date === expected)
+      assert(formatter.format(daysToLocalDate(days)) === expected)
+      assert(formatter.format(toJavaDate(days, TimeZone.getTimeZone(zoneId))) === expected)
     }
   }
 
@@ -66,18 +76,16 @@ class DateFormatterSuite extends SparkFunSuite with SQLHelper {
             "2018-12-12",
             "2038-01-01",
             "5010-11-17").foreach { date =>
-            outstandingTimezonesIds.foreach { timeZone =>
-              withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
-                val formatter = DateFormatter(
-                  DateFormatter.defaultPattern,
-                  getZoneId(timeZone),
-                  DateFormatter.defaultLocale,
-                  legacyFormat)
-                val days = formatter.parse(date)
-                assert(date === formatter.format(days))
-                assert(date === formatter.format(daysToLocalDate(days)))
-                assert(date === formatter.format(toJavaDate(days)))
-              }
+            withOutstandingZoneIds { zoneId =>
+              val formatter = DateFormatter(
+                DateFormatter.defaultPattern,
+                zoneId,
+                DateFormatter.defaultLocale,
+                legacyFormat)
+              val days = formatter.parse(date)
+              assert(date === formatter.format(days))
+              assert(date === formatter.format(daysToLocalDate(days)))
+              assert(date === formatter.format(toJavaDate(days, TimeZone.getTimeZone(zoneId))))
             }
           }
         }
@@ -100,17 +108,15 @@ class DateFormatterSuite extends SparkFunSuite with SQLHelper {
             17877,
             24837,
             1110657).foreach { days =>
-            outstandingTimezonesIds.foreach { timeZone =>
-              withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
-                val formatter = DateFormatter(
-                  DateFormatter.defaultPattern,
-                  getZoneId(timeZone),
-                  DateFormatter.defaultLocale,
-                  legacyFormat)
-                val date = formatter.format(days)
-                val parsed = formatter.parse(date)
-                assert(days === parsed)
-              }
+            withOutstandingZoneIds { zoneId =>
+              val formatter = DateFormatter(
+                DateFormatter.defaultPattern,
+                zoneId,
+                DateFormatter.defaultLocale,
+                legacyFormat)
+              val date = formatter.format(days)
+              val parsed = formatter.parse(date)
+              assert(days === parsed)
             }
           }
         }
@@ -167,18 +173,21 @@ class DateFormatterSuite extends SparkFunSuite with SQLHelper {
   test("SPARK-31557: rebasing in legacy formatters/parsers") {
     withSQLConf(SQLConf.LEGACY_TIME_PARSER_POLICY.key -> LegacyBehaviorPolicy.LEGACY.toString) {
       LegacyDateFormats.values.foreach { legacyFormat =>
-        outstandingTimezonesIds.foreach { timeZone =>
-          withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> timeZone) {
-            val formatter = DateFormatter(
-              DateFormatter.defaultPattern,
-              getZoneId(timeZone),
-              DateFormatter.defaultLocale,
-              legacyFormat)
-            assert(LocalDate.ofEpochDay(formatter.parse("1000-01-01")) === LocalDate.of(1000, 1, 1))
-            assert(formatter.format(LocalDate.of(1000, 1, 1)) === "1000-01-01")
-            assert(formatter.format(localDateToDays(LocalDate.of(1000, 1, 1))) === "1000-01-01")
-            assert(formatter.format(java.sql.Date.valueOf("1000-01-01")) === "1000-01-01")
-          }
+        withOutstandingZoneIds { zoneId =>
+          val formatter = DateFormatter(
+            DateFormatter.defaultPattern,
+            zoneId,
+            DateFormatter.defaultLocale,
+            legacyFormat)
+          assert(LocalDate.ofEpochDay(formatter.parse("1000-01-01")) === LocalDate.of(1000, 1, 1))
+          assert(formatter.format(LocalDate.of(1000, 1, 1)) === "1000-01-01")
+          assert(formatter.format(localDateToDays(LocalDate.of(1000, 1, 1))) === "1000-01-01")
+          val cal = new Calendar.Builder()
+            .setCalendarType("gregory")
+            .setTimeZone(TimeZone.getTimeZone(zoneId))
+            .setDate(1000, 0, 1)
+            .build()
+          assert(formatter.format(cal.getTime) === "1000-01-01")
         }
       }
     }