address comments

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/csv/UnivocityParser.scala

@@ -182,7 +182,7 @@ class UnivocityParser(
           case NonFatal(e) =>
             // If fails to parse, then tries the way used in 2.0 and 1.x for backwards
             // compatibility.
-            val str = UTF8String.fromString(datum)
+            val str = UTF8String.fromString(DateTimeUtils.cleanLegacyTimestampStr(datum))
             DateTimeUtils.stringToTimestamp(str, options.zoneId).getOrElse(throw e)
         }
       }
@@ -195,7 +195,7 @@ class UnivocityParser(
           case NonFatal(e) =>
             // If fails to parse, then tries the way used in 2.0 and 1.x for backwards
             // compatibility.
-            val str = UTF8String.fromString(datum)
+            val str = UTF8String.fromString(DateTimeUtils.cleanLegacyTimestampStr(datum))
             DateTimeUtils.stringToDate(str, options.zoneId).getOrElse(throw e)
         }
       }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -235,7 +235,7 @@ class JacksonParser(
             case NonFatal(e) =>
               // If fails to parse, then tries the way used in 2.0 and 1.x for backwards
               // compatibility.
-              val str = UTF8String.fromString(parser.getText)
+              val str = UTF8String.fromString(DateTimeUtils.cleanLegacyTimestampStr(parser.getText))
               DateTimeUtils.stringToTimestamp(str, options.zoneId).getOrElse(throw e)
           }
 
@@ -252,7 +252,7 @@ class JacksonParser(
             case NonFatal(e) =>
               // If fails to parse, then tries the way used in 2.0 and 1.x for backwards
               // compatibility.
-              val str = UTF8String.fromString(parser.getText)
+              val str = UTF8String.fromString(DateTimeUtils.cleanLegacyTimestampStr(parser.getText))
               DateTimeUtils.stringToDate(str, options.zoneId).getOrElse {
                 // In Spark 1.5.0, we store the data as number of days since epoch in string.
                 // So, we just convert it to Int.

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

@@ -163,6 +163,21 @@ object DateTimeUtils {
     instantToMicros(localDateTime.atZone(zoneId).toInstant)
   }
 
+  // A method called by JSON/CSV parser to clean up the legacy timestamp string by removing the
+  // "GMT" string.
+  def cleanLegacyTimestampStr(s: String): String = {
+    val indexOfGMT = s.indexOf("GMT")
+    if (indexOfGMT != -1) {
+      // ISO8601 with a weird time zone specifier (2000-01-01T00:00GMT+01:00)
+      val s0 = s.substring(0, indexOfGMT)
+      val s1 = s.substring(indexOfGMT + 3)
+      // Mapped to 2000-01-01T00:00+01:00
+      s0 + s1
+    } else {
+      s
+    }
+  }
+
   /**
    * Trim and parse a given UTF8 date string to the corresponding a corresponding [[Long]] value.
    * The return type is [[Option]] in order to distinguish between 0L and null. The following

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/csv/UnivocityParserSuite.scala

@@ -341,6 +341,15 @@ class UnivocityParserSuite extends SparkFunSuite with SQLHelper {
         days(2020, 1, 12, 0, 0, 0))
       assert(parser.makeConverter("t", DateType).apply("2020-1-12 xyz") ==
         days(2020, 1, 12, 0, 0, 0))
+      // The legacy format ignores the "GMT" from the string
+      assert(parser.makeConverter("t", TimestampType).apply("2020-1-12 12:3:45GMT") ==
+        date(2020, 1, 12, 12, 3, 45, 0))
+      assert(parser.makeConverter("t", TimestampType).apply("GMT2020-1-12 12:3:45") ==
+        date(2020, 1, 12, 12, 3, 45, 0))
+      assert(parser.makeConverter("t", DateType).apply("2020-1-12GMT") ==
+        days(2020, 1, 12, 0, 0, 0))
+      assert(parser.makeConverter("t", DateType).apply("GMT2020-1-12") ==
+        days(2020, 1, 12, 0, 0, 0))
     }
 
     val options = new CSVOptions(Map.empty[String, String], false, "UTC")