[SPARK-31868][SQL] Restore the parsing behaviour week-based-year for 2.4

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

@@ -56,7 +56,7 @@ class Iso8601DateFormatter(
     val specialDate = convertSpecialDate(s.trim, zoneId)
     specialDate.getOrElse {
       try {
-        val localDate = toLocalDate(formatter.parse(s))
+        val localDate = toLocalDate(formatter.parse(s), locale)
         localDateToDays(localDate)
       } catch checkDiffResult(s, legacyFormatter.parse)
     }

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

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst.util
 import java.time._
 import java.time.chrono.IsoChronology
 import java.time.format.{DateTimeFormatter, DateTimeFormatterBuilder, ResolverStyle}
-import java.time.temporal.{ChronoField, TemporalAccessor, TemporalQueries}
+import java.time.temporal._
 import java.util.Locale
 
 import com.google.common.cache.CacheBuilder
@@ -39,15 +39,22 @@ trait DateTimeFormatterHelper {
     }
   }
 
-  protected def toLocalDate(accessor: TemporalAccessor): LocalDate = {
+  protected def toLocalDate(accessor: TemporalAccessor, locale: Locale): LocalDate = {
     val localDate = accessor.query(TemporalQueries.localDate())
     // If all the date fields are specified, return the local date directly.
     if (localDate != null) return localDate
 
+    lazy val weekBasedYearField = WeekFields.of(locale).weekBasedYear()
     // Users may want to parse only a few datetime fields from a string and extract these fields
     // later, and we should provide default values for missing fields.
     // To be compatible with Spark 2.4, we pick 1970 as the default value of year.
-    val year = getOrDefault(accessor, ChronoField.YEAR, 1970)
+    val year = if (accessor.isSupported(ChronoField.YEAR)) {
+      accessor.get(ChronoField.YEAR)
+    } else if (accessor.isSupported(weekBasedYearField)) {
+      val year = accessor.get(weekBasedYearField) - 1
+      return LocalDate.of(year, 12, 1).`with`(TemporalAdjusters.lastInMonth(DayOfWeek.SUNDAY))
+    } else 1970
+
     val month = getOrDefault(accessor, ChronoField.MONTH_OF_YEAR, 1)
     val day = getOrDefault(accessor, ChronoField.DAY_OF_MONTH, 1)
     LocalDate.of(year, month, day)
@@ -74,8 +81,11 @@ trait DateTimeFormatterHelper {
 
   // Converts the parsed temporal object to ZonedDateTime. It sets time components to zeros
   // if they does not exist in the parsed object.
-  protected def toZonedDateTime(accessor: TemporalAccessor, zoneId: ZoneId): ZonedDateTime = {
-    val localDate = toLocalDate(accessor)
+  protected def toZonedDateTime(
+      accessor: TemporalAccessor,
+      zoneId: ZoneId,
+      locale: Locale): ZonedDateTime = {
+    val localDate = toLocalDate(accessor, locale)
     val localTime = toLocalTime(accessor)
     ZonedDateTime.of(localDate, localTime, zoneId)
   }

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

@@ -79,7 +79,7 @@ class Iso8601TimestampFormatter(
         val parsed = formatter.parse(s)
         val parsedZoneId = parsed.query(TemporalQueries.zone())
         val timeZoneId = if (parsedZoneId == null) zoneId else parsedZoneId
-        val zonedDateTime = toZonedDateTime(parsed, timeZoneId)
+        val zonedDateTime = toZonedDateTime(parsed, timeZoneId, locale)
         val epochSeconds = zonedDateTime.toEpochSecond
         val microsOfSecond = zonedDateTime.get(MICRO_OF_SECOND)
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/DateExpressionsSuite.scala

@@ -1168,4 +1168,13 @@ class DateExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     checkExceptionInExpression[ArithmeticException](
       MillisToTimestamp(Literal(-92233720368547758L)), "long overflow")
   }
+
+  test("SPARK-31868: Restore the behaviour week-based-year for 2.4") {
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-11-17 13:33:33"), Literal("YYYY-MM-dd HH:mm:ss")).child,
+      Timestamp.valueOf("2017-12-31 13:33:33.0"))
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-11-17"), Literal("YYYY-MM-dd")).child,
+      Timestamp.valueOf("2017-12-31 00:00:00.0"))
+  }
 }

sql/core/src/test/resources/sql-tests/inputs/datetime.sql

@@ -160,3 +160,9 @@ select from_json('{"time":"26/October/2015"}', 'time Timestamp', map('timestampF
 select from_json('{"date":"26/October/2015"}', 'date Date', map('dateFormat', 'dd/MMMMM/yyyy'));
 select from_csv('26/October/2015', 'time Timestamp', map('timestampFormat', 'dd/MMMMM/yyyy'));
 select from_csv('26/October/2015', 'date Date', map('dateFormat', 'dd/MMMMM/yyyy'));
+
+-- SPARK-31868: Restore the behaviour week-based-year for 2.4
+select to_timestamp('1969-01-01', 'YYYY-MM-dd');
+select to_timestamp('1969-12-31', 'YYYY-MM-dd');
+select to_timestamp('2018-01-01', 'YYYY-MM-dd');
+select to_timestamp('2018-11-17 13:33:33', 'YYYY-MM-dd HH:mm:ss');

sql/core/src/test/resources/sql-tests/results/ansi/datetime.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 116
+-- Number of queries: 120
 
 
 -- !query
@@ -999,3 +999,35 @@ struct<>
 -- !query output
 org.apache.spark.SparkUpgradeException
 You may get a different result due to the upgrading of Spark 3.0: Fail to recognize 'dd/MMMMM/yyyy' pattern in the DateTimeFormatter. 1) You can set spark.sql.legacy.timeParserPolicy to LEGACY to restore the behavior before Spark 3.0. 2) You can form a valid datetime pattern with the guide from https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html
+
+
+-- !query
+select to_timestamp('1969-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('1969-12-31', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-12-31, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('2018-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(2018-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+2017-12-31 00:00:00
+
+
+-- !query
+select to_timestamp('2018-11-17 13:33:33', 'YYYY-MM-dd HH:mm:ss')
+-- !query schema
+struct<to_timestamp(2018-11-17 13:33:33, YYYY-MM-dd HH:mm:ss):timestamp>
+-- !query output
+2017-12-31 13:33:33

sql/core/src/test/resources/sql-tests/results/datetime-legacy.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 116
+-- Number of queries: 120
 
 
 -- !query
@@ -956,3 +956,35 @@ select from_csv('26/October/2015', 'date Date', map('dateFormat', 'dd/MMMMM/yyyy
 struct<from_csv(26/October/2015):struct<date:date>>
 -- !query output
 {"date":2015-10-26}
+
+
+-- !query
+select to_timestamp('1969-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('1969-12-31', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-12-31, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('2018-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(2018-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+2017-12-31 00:00:00
+
+
+-- !query
+select to_timestamp('2018-11-17 13:33:33', 'YYYY-MM-dd HH:mm:ss')
+-- !query schema
+struct<to_timestamp(2018-11-17 13:33:33, YYYY-MM-dd HH:mm:ss):timestamp>
+-- !query output
+2017-12-31 13:33:33

sql/core/src/test/resources/sql-tests/results/datetime.sql.out

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 116
+-- Number of queries: 120
 
 
 -- !query
@@ -971,3 +971,35 @@ struct<>
 -- !query output
 org.apache.spark.SparkUpgradeException
 You may get a different result due to the upgrading of Spark 3.0: Fail to recognize 'dd/MMMMM/yyyy' pattern in the DateTimeFormatter. 1) You can set spark.sql.legacy.timeParserPolicy to LEGACY to restore the behavior before Spark 3.0. 2) You can form a valid datetime pattern with the guide from https://spark.apache.org/docs/latest/sql-ref-datetime-pattern.html
+
+
+-- !query
+select to_timestamp('1969-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('1969-12-31', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(1969-12-31, YYYY-MM-dd):timestamp>
+-- !query output
+1968-12-29 00:00:00
+
+
+-- !query
+select to_timestamp('2018-01-01', 'YYYY-MM-dd')
+-- !query schema
+struct<to_timestamp(2018-01-01, YYYY-MM-dd):timestamp>
+-- !query output
+2017-12-31 00:00:00
+
+
+-- !query
+select to_timestamp('2018-11-17 13:33:33', 'YYYY-MM-dd HH:mm:ss')
+-- !query schema
+struct<to_timestamp(2018-11-17 13:33:33, YYYY-MM-dd HH:mm:ss):timestamp>
+-- !query output
+2017-12-31 13:33:33