[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,9 +20,11 @@ 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 scala.util.control.NonFatal
+
 import com.google.common.cache.CacheBuilder
 
 import org.apache.spark.SparkUpgradeException
@@ -31,26 +33,92 @@ import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.internal.SQLConf.LegacyBehaviorPolicy._
 
 trait DateTimeFormatterHelper {
-  private def getOrDefault(accessor: TemporalAccessor, field: ChronoField, default: Int): Int = {
+  private def getFieldValue(
+      accessor: TemporalAccessor,
+      field: TemporalField): Option[Int] = {
     if (accessor.isSupported(field)) {
-      accessor.get(field)
+      try {
+        Option(accessor.get(field))
+      } catch {
+        case NonFatal(_) => None
+      }
     } else {
-      default
+      None
     }
   }
 
-  protected def toLocalDate(accessor: TemporalAccessor): LocalDate = {
+  @throws[DateTimeException]
+  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 all the date fields are resolved(yMd or Ywu), return the local date directly.
     if (localDate != null) return localDate
 
-    // 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 month = getOrDefault(accessor, ChronoField.MONTH_OF_YEAR, 1)
-    val day = getOrDefault(accessor, ChronoField.DAY_OF_MONTH, 1)
-    LocalDate.of(year, month, day)
+    var res = LocalDate.of(1970, 1, 1)
+
+    val weekFields = WeekFields.of(locale)
+    val weekBasedYearField = weekFields.weekBasedYear
+    var weekBasedYearEnabled = false
+
+    val year = getFieldValue(accessor, weekBasedYearField).map { y =>
+      weekBasedYearEnabled = true
+      y
+    }.orElse {
+      getFieldValue(accessor, ChronoField.YEAR)
+    }
+
+    val week = getFieldValue(accessor, weekFields.weekOfWeekBasedYear())
+    val dayOfWeek = getFieldValue(accessor, weekFields.dayOfWeek())
+
+    // TODO: How to check 'W' week-of-month field, not like other week-based field, it always throw
+    // UnsupportedTemporalTypeException to get it even `accessor.isSupported` passed.
+
+    if (weekBasedYearEnabled) {
+      // If the week-based-year field exists, only the week-based fields matters.
+      res.`with`(weekFields.weekOfWeekBasedYear, week.getOrElse(1).toLong)
+        .`with`(weekFields.dayOfWeek(), dayOfWeek.getOrElse(1).toLong)
+        .`with`(weekBasedYearField, year.get)
+    } else {
+      if (year.isDefined) {
+        res = res.withYear(year.get)
+      }
+
+      val month = getFieldValue(accessor, ChronoField.MONTH_OF_YEAR)
+      val day = getFieldValue(accessor, ChronoField.DAY_OF_MONTH)
+
+      if (month.isDefined && week.isDefined) {
+        // check the week fall into the correct month of the year
+        res = res.`with`(weekFields.weekOfWeekBasedYear(), week.get)
+        if (res.getMonthValue != month.get) {
+          throw new DateTimeException(
+            s"week-of-week-based-year value: ${week.get} conflicts with month-of-year value:" +
+              s" ${month.get} which should be ${res.getMonthValue} instead.")
+        }
+      } else if (month.isDefined) {
+        res = res.withMonth(month.get)
+      } else if (week.isDefined) {
+        if (day.isDefined) {
+          throw new DateTimeException(
+            s"Can not use week-of-week-based-year and day-of-month together in non-week-based" +
+              s" mode.")
+        }
+        res = res.`with`(weekFields.weekOfWeekBasedYear, week.get)
+      }
+
+      if (dayOfWeek.isDefined && day.isDefined) {
+        // check whether the days matches
+        res = res.`with`(weekFields.dayOfWeek(), dayOfWeek.get)
+        if (res.getDayOfMonth != day.get) {
+          throw new DateTimeException(
+            s"day-of-week value: ${dayOfWeek.get} conflicts with day-of-month value:" +
+              s" ${day.get} which should be ${res.getDayOfMonth} instead.")
+        }
+      } else if (day.isDefined) {
+        res = res.withDayOfMonth(day.get)
+      } else if (dayOfWeek.isDefined) {
+        res = res.`with`(weekFields.dayOfWeek(), dayOfWeek.get)
+      }
+      res
+    }
   }
 
   private def toLocalTime(accessor: TemporalAccessor): LocalTime = {
@@ -66,16 +134,19 @@ trait DateTimeFormatterHelper {
     } else {
       0
     }
-    val minute = getOrDefault(accessor, ChronoField.MINUTE_OF_HOUR, 0)
-    val second = getOrDefault(accessor, ChronoField.SECOND_OF_MINUTE, 0)
-    val nanoSecond = getOrDefault(accessor, ChronoField.NANO_OF_SECOND, 0)
-    LocalTime.of(hour, minute, second, nanoSecond)
+    val minute = getFieldValue(accessor, ChronoField.MINUTE_OF_HOUR)
+    val second = getFieldValue(accessor, ChronoField.SECOND_OF_MINUTE)
+    val nanoSecond = getFieldValue(accessor, ChronoField.NANO_OF_SECOND)
+    LocalTime.of(hour, minute.getOrElse(0), second.getOrElse(0), nanoSecond.getOrElse(0))
   }
 
   // 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,65 @@ 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") {
+    // TODO: Locale.US is Sunday started day of week, which affects the new formatter,
+    // while in 2.4 it's Monday first.
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-11-17"), Literal("YYYY-MM-dd")).child,
+      Timestamp.valueOf("2017-12-31 00:00:00.0"))
+
+    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("1969 1 2"), Literal("YYYY w u")).child,
+      Timestamp.valueOf("1968-12-30 00:00:00.0"))
+
+    // the existence of 'W' is not for generating the timestamp but likely for checking whether
+    // the timestamp falling into it.
+    checkEvaluation(
+      new ParseToTimestamp(Literal("1969 5 1 2"), Literal("YYYY W w u")).child,
+      Timestamp.valueOf("1968-12-30 00:00:00.0"))
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("1969 5 2"), Literal("YYYY W u")).child,
+      Timestamp.valueOf("1968-12-30 00:00:00.0"))
+
+//    // the legacy parser does not support 'W' and results null, so SparkUpgradeException will come
+//    checkExceptionInExpression[SparkUpgradeException](
+//      new ParseToTimestamp(Literal("1969 4 2"), Literal("YYYY W u")).child, "3.0")
+//    checkExceptionInExpression[SparkUpgradeException](
+//      new ParseToTimestamp(Literal("5"), Literal("W")).child, "3.0")
+
+    // https://bugs.openjdk.java.net/browse/JDK-8145633
+    // Adjacent value parsing not supported for Localized Patterns
+    checkExceptionInExpression[SparkUpgradeException](
+      new ParseToTimestamp(Literal("196940"), Literal("YYYYww")).child, "3.0")
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2020 1 3 2"), Literal("yyyy M w u")).child,
+      Timestamp.valueOf("2020-01-13 00:00:00.0"))
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-46-7 13:33:33"), Literal("YYYY-ww-u HH:mm:ss")).child,
+      Timestamp.valueOf("2018-11-17 13:33:33.0"))
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-11-17"), Literal("YYYY-ww-dd")).child,
+      Timestamp.valueOf("2018-03-11 00:00:00.0"))
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("2018-11-2-17"), Literal("yyyy-ww-dd")).child,
+      null)
+
+    // problem of first day of week change
+    checkExceptionInExpression[SparkUpgradeException](
+      new ParseToTimestamp(Literal("1969 1 6 1"), Literal("yyyy M d u")).child, "3.0")
+
+    checkEvaluation(
+      new ParseToTimestamp(Literal("1969 5 11 11"), Literal("YYYY M ww dd")).child,
+      Timestamp.valueOf("1969-03-09 00:00:00.0"))
+  }
 }

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

@@ -160,3 +160,29 @@ 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('2018-11-17 13:33:33', 'YYYY-MM-dd HH:mm:ss'); -- only the week-based year matters for the date part
+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');
+select to_timestamp('1969 1 1', 'yyyy w u');
+select to_timestamp('1969 1 1', 'yyyy M u');
+select to_timestamp('1 1969 1', 'M YYYY u');
+select to_timestamp('1969 1 1', 'YYYY M u');
+select to_timestamp('1969 1 6 1', 'yyyy M d u');
+select to_timestamp('1969 1 5 1', 'yyyy M d u');
+-- YYYY-ww-dd
+select to_timestamp('2018 11 17', 'YYYY ww dd');
+select to_timestamp('1969 2 11', 'YYYY W dd');
+select to_timestamp('1969 5 11 11', 'YYYY M ww dd');
+
+select to_timestamp('2020 1 3 2', 'yyyy M w u');
+select to_timestamp('2018-11-2-17', 'yyyy-ww-W-dd');
+select to_timestamp('2018-11-2-11', 'yyyy-ww-W-dd');
+select to_timestamp('2018-11-3-12', 'yyyy-ww-W-dd');
+select to_timestamp('2018-11-10', 'yyyy-ww-dd');
+select to_timestamp('1', 'u');
+select to_timestamp('5 2', 'u d');
+select to_timestamp('5 3', 'u d');