[SPARK-31939][SQL][test-java11] Fix Parsing day of year when year field pattern is missing

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

@@ -39,6 +39,18 @@ trait DateTimeFormatterHelper {
     }
   }
 
+  private def verifyLocalDate(
+      accessor: TemporalAccessor, field: ChronoField, candidate: LocalDate): Unit = {
+    if (accessor.isSupported(field) && candidate.isSupported(field)) {
+      val actual = accessor.get(field)
+      val expected = candidate.get(field)
+      if (actual != expected) {
+        throw new DateTimeException(s"Conflict found: Field $field $actual differs from" +
+          s" $field $expected derived from $candidate")
+      }
+    }
+  }
+
   protected def toLocalDate(accessor: TemporalAccessor): LocalDate = {
     val localDate = accessor.query(TemporalQueries.localDate())
     // If all the date fields are specified, return the local date directly.
@@ -48,9 +60,17 @@ trait DateTimeFormatterHelper {
     // 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)
+    if (accessor.isSupported(ChronoField.DAY_OF_YEAR)) {
+      val dayOfYear = accessor.get(ChronoField.DAY_OF_YEAR)
+      val date = LocalDate.ofYearDay(year, dayOfYear)
+      verifyLocalDate(accessor, ChronoField.MONTH_OF_YEAR, date)
+      verifyLocalDate(accessor, ChronoField.DAY_OF_MONTH, date)
+      date
+    } else {
+      val month = getOrDefault(accessor, ChronoField.MONTH_OF_YEAR, 1)
+      val day = getOrDefault(accessor, ChronoField.DAY_OF_MONTH, 1)
+      LocalDate.of(year, month, day)
+    }
   }
 
   private def toLocalTime(accessor: TemporalAccessor): LocalTime = {

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

@@ -258,19 +258,19 @@ class TimestampFormatterSuite extends DatetimeFormatterSuite {
     }
   }
 
+  def assertParsingError(f: => Unit): Unit = {
+    intercept[Exception](f) match {
+      case e: SparkUpgradeException =>
+        assert(e.getCause.isInstanceOf[DateTimeException])
+      case e =>
+        assert(e.isInstanceOf[DateTimeException])
+    }
+  }
+
   test("SPARK-30958: parse timestamp with negative year") {
     val formatter1 = TimestampFormatter("yyyy-MM-dd HH:mm:ss", UTC, isParsing = true)
     assert(formatter1.parse("-1234-02-22 02:22:22") === date(-1234, 2, 22, 2, 22, 22))
 
-    def assertParsingError(f: => Unit): Unit = {
-      intercept[Exception](f) match {
-        case e: SparkUpgradeException =>
-          assert(e.getCause.isInstanceOf[DateTimeException])
-        case e =>
-          assert(e.isInstanceOf[DateTimeException])
-      }
-    }
-
     // "yyyy" with "G" can't parse negative year or year 0000.
     val formatter2 = TimestampFormatter("G yyyy-MM-dd HH:mm:ss", UTC, isParsing = true)
     assertParsingError(formatter2.parse("BC -1234-02-22 02:22:22"))
@@ -433,4 +433,35 @@ class TimestampFormatterSuite extends DatetimeFormatterSuite {
       assert(formatter.format(date(1970, 4, 10)) == "100")
     }
   }
+
+  test("SPARK-31939: Fix Parsing day of year when year field pattern is missing") {
+    // resolved to queryable LocaleDate or fail directly
+    val f0 = TimestampFormatter("yyyy-dd-DD", UTC, isParsing = true)
+    assert(f0.parse("2020-29-60") === date(2020, 2, 29))
+    assertParsingError(f0.parse("2020-02-60"))
+    val f1 = TimestampFormatter("yyyy-MM-DD", UTC, isParsing = true)
+    assert(f1.parse("2020-02-60") === date(2020, 2, 29))
+    assertParsingError(f1.parse("2020-03-60"))
+    val f2 = TimestampFormatter("yyyy-MM-dd-DD", UTC, isParsing = true)
+    assert(f2.parse("2020-02-29-60") === date(2020, 2, 29))
+    assertParsingError(f2.parse("2020-03-01-60"))
+    val f3 = TimestampFormatter("yyyy-DDD", UTC, isParsing = true)
+    assert(f3.parse("2020-366") === date(2020, 12, 31))
+    assertParsingError(f3.parse("2019-366"))
+
+    // unresolved and need to check manually(SPARK-31939 fixed)
+    val f4 = TimestampFormatter("DDD", UTC, isParsing = true)
+    assert(f4.parse("365") === date(1970, 12, 31))
+    assertParsingError(f4.parse("366")) // 1970 is not a leap year
+    val f5 = TimestampFormatter("MM-DD", UTC, isParsing = true)
+    assert(f5.parse("03-60") === date(1970, 3, 1))
+    assertParsingError(f5.parse("02-60"))
+    val f6 = TimestampFormatter("MM-dd-DD", UTC, isParsing = true)
+    assert(f6.parse("02-28-59") === date(1970, 2, 28))
+    assertParsingError(f6.parse("02-28-60"))
+    assertParsingError(f6.parse("02-28-58"))
+    val f7 = TimestampFormatter("dd-DD", UTC, isParsing = true)
+    assert(f7.parse("28-59") === date(1970, 2, 28))
+    assertParsingError(f7.parse("27-59"))
+  }
 }