support full range of datetime

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

@@ -260,6 +260,11 @@ object DateTimeUtils {
     var j = 0
     var digitsMilli = 0
     var justTime = false
+    var sign = 1
+    if (bytes(j) == '-' || bytes(j) == '+') {
+      sign = if (bytes(j) == '-') -1 else 1
+      j += 1
+    }
     while (j < bytes.length) {
       val b = bytes(j)
       val parsedValue = b - '0'.toByte
@@ -269,10 +274,6 @@ object DateTimeUtils {
           i += 3
         } else if (i < 2) {
           if (b == '-') {
-            if (i == 0 && j != 4) {
-              // year should have exact four digits
-              return (Array.empty, None, false)
-            }
             segments(i) = currentSegmentValue
             currentSegmentValue = 0
             i += 1
@@ -339,10 +340,6 @@ object DateTimeUtils {
     }
 
     segments(i) = currentSegmentValue
-    if (!justTime && i == 0 && j != 4) {
-      // year should have exact four digits
-      return (Array.empty, None, false)
-    }
 
     while (digitsMilli < 6) {
       segments(6) *= 10
@@ -401,7 +398,7 @@ object DateTimeUtils {
       val localDate = if (justTime) {
         LocalDate.now(zoneId)
       } else {
-        LocalDate.of(segments(0), segments(1), segments(2))
+        LocalDate.of(sign * segments(0), segments(1), segments(2))
       }
       val localDateTime = LocalDateTime.of(localDate, localTime)
       val zonedDateTime = ZonedDateTime.of(localDateTime, zoneId)
@@ -530,17 +527,18 @@ object DateTimeUtils {
       return None
     }
     val segments: Array[Int] = Array[Int](1, 1, 1)
+    var sign = 1
     var i = 0
     var currentSegmentValue = 0
     val bytes = s.trimAll().getBytes
     var j = 0
+    if (bytes(j) == '-' || bytes(j) == '+') {
+      sign = if (bytes(j) == '-') -1 else 1
+      j += 1
+    }
     while (j < bytes.length && (i < 3 && !(bytes(j) == ' ' || bytes(j) == 'T'))) {
       val b = bytes(j)
       if (i < 2 && b == '-') {
-        if (i == 0 && j != 4) {
-          // year should have exact four digits
-          return None
-        }
         segments(i) = currentSegmentValue
         currentSegmentValue = 0
         i += 1
@@ -554,17 +552,13 @@ object DateTimeUtils {
       }
       j += 1
     }
-    if (i == 0 && j != 4) {
-      // year should have exact four digits
-      return None
-    }
     if (i < 2 && j < bytes.length) {
       // For the `yyyy` and `yyyy-[m]m` formats, entire input must be consumed.
       return None
     }
     segments(i) = currentSegmentValue
     try {
-      val localDate = LocalDate.of(segments(0), segments(1), segments(2))
+      val localDate = LocalDate.of(sign * segments(0), segments(1), segments(2))
       Some(localDateToDays(localDate))
     } catch {
       case NonFatal(_) => None

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

@@ -151,6 +151,21 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers with SQLHelper {
     assert(toDate("1999 08").isEmpty)
   }
 
+  test("SPARK-35780: support full range of date string") {
+    assert(toDate("02015-03-18").get === days(2015, 3, 18))
+    assert(toDate("015-03-18").get === days(15, 3, 18))
+    assert(toDate("015").get === days(15, 1, 1))
+    assert(toDate("02015").get === days(2015, 1, 1))
+    assert(toDate("-02015").get === days(-2015, 1, 1))
+    assert(toDate("999999-1-28").get === days(999999, 1, 28))
+    assert(toDate("-999999-1-28").get === days(-999999, 1, 28))
+    assert(toDate("1-1-28").get === days(1, 1, 28))
+    assert(toDate("5881580-7-11").get === days(5881580, 7, 11))
+    assert(toDate("5881580-7-12").isEmpty)
+    assert(toDate("-5877641-6-23").get === days(-5877641, 6, 23))
+    assert(toDate("-5877641-6-22").isEmpty)
+  }
+
   private def toTimestamp(str: String, zoneId: ZoneId): Option[Long] = {
     stringToTimestamp(UTF8String.fromString(str), zoneId)
   }
@@ -283,6 +298,22 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers with SQLHelper {
     }
   }
 
+  test("SPARK-35780: support full range of timestamp string") {
+    def checkStringToTimestamp(str: String, expected: Option[Long]): Unit = {
+      assert(toTimestamp(str, UTC) === expected)
+    }
+
+    checkStringToTimestamp("-1969-12-31 16:00:00", Option(date(-1969, 12, 31, 16, zid = UTC)))
+    checkStringToTimestamp("02015-03-18 16:00:00", Option(date(2015, 3, 18, 16, zid = UTC)))
+    checkStringToTimestamp("015-03-18 16:00:00", Option(date(15, 3, 18, 16, zid = UTC)))
+    checkStringToTimestamp("000001", Option(date(1, 1, 1, 0, zid = UTC)))
+    checkStringToTimestamp("-000001", Option(date(-1, 1, 1, 0, zid = UTC)))
+    checkStringToTimestamp("99999-03-18T12:03:17",
+      Option(date(99999, 3, 18, 12, 3, 17, zid = UTC)))
+    checkStringToTimestamp("-99999-03-18T12:03:17",
+      Option(date(-99999, 3, 18, 12, 3, 17, zid = UTC)))
+  }
+
   test("SPARK-15379: special invalid date string") {
     // Test stringToDate
     assert(toDate("2015-02-29 00:00:00").isEmpty)