[SPARK-23715][SQL] the input of to/from_utc_timestamp can not have timezone

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TypeCoercion.scala

@@ -782,6 +782,22 @@ object TypeCoercion {
       // Skip nodes who's children have not been resolved yet.
       case e if !e.childrenResolved => e
 
+      // Special rules for `to/from_utc_timestamp`. `to/from_utc_timestamp` assumes its input is
+      // in UTC timezone, and if input is string, it should not contain timezone.
+      // TODO: We should move the type coercion logic to expressions instead of a central
+      // place to put all the rules.
+      case e: FromUTCTimestamp if e.left.dataType == StringType =>
+        e.copy(left = StringToTimestampWithoutTimezone(e.left))
+
+      case e: FromUTCTimestamp if e.left.dataType == DateType =>
+        e.copy(left = Cast(e.left, TimestampType))
+
+      case e: ToUTCTimestamp if e.left.dataType == StringType =>
+        e.copy(left = StringToTimestampWithoutTimezone(e.left))
+
+      case e: ToUTCTimestamp if e.left.dataType == DateType =>
+        e.copy(left = Cast(e.left, TimestampType))
+
       case b @ BinaryOperator(left, right) if left.dataType != right.dataType =>
         findTightestCommonType(left.dataType, right.dataType).map { commonType =>
           if (b.inputType.acceptsType(commonType)) {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala

@@ -1016,6 +1016,47 @@ case class TimeAdd(start: Expression, interval: Expression, timeZoneId: Option[S
   }
 }
 
+/**
+ * A special expression used to convert the string input of `to/from_utc_timestamp` to timestamp,
+ * which requires the timestamp string to not have timezone information, otherwise null is returned.
+ */
+case class StringToTimestampWithoutTimezone(child: Expression, timeZoneId: Option[String] = None)
+  extends UnaryExpression with TimeZoneAwareExpression with ExpectsInputTypes {
+
+  override def withTimeZone(timeZoneId: String): TimeZoneAwareExpression =
+    copy(timeZoneId = Option(timeZoneId))
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(StringType)
+  override def dataType: DataType = TimestampType
+  override def nullable: Boolean = true
+  override def prettyName: String = "string_to_timestamp"
+
+  override def nullSafeEval(input: Any): Any = {
+    DateTimeUtils.stringToTimestamp(
+      input.asInstanceOf[UTF8String], timeZone, forceTimezone = true).orNull
+  }
+
+  override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val dtu = DateTimeUtils.getClass.getName.stripSuffix("$")
+    val tz = ctx.addReferenceObj("timeZone", timeZone)
+    val longOpt = ctx.freshName("longOpt")
+    val eval = child.genCode(ctx)
+    val code = s"""
+       |${eval.code}
+       |${CodeGenerator.JAVA_BOOLEAN} ${ev.isNull} = true;
+       |${CodeGenerator.JAVA_LONG} ${ev.value} = ${CodeGenerator.defaultValue(TimestampType)};
+       |if (!${eval.isNull}) {
+       |  scala.Option<Long> $longOpt = $dtu.stringToTimestamp(${eval.value}, $tz, true);
+       |  if ($longOpt.isDefined()) {
+       |    ${ev.value} = ((Long) $longOpt.get()).longValue();
+       |    ${ev.isNull} = false;
+       |  }
+       |}
+     """.stripMargin
+    ev.copy(code = code)
+  }
+}
+
 /**
  * Given a timestamp like '2017-07-14 02:40:00.0', interprets it as a time in UTC, and renders
  * that time as a timestamp in the given time zone. For example, 'GMT+1' would yield
@@ -1032,7 +1073,7 @@ case class TimeAdd(start: Expression, interval: Expression, timeZoneId: Option[S
   since = "1.5.0")
 // scalastyle:on line.size.limit
 case class FromUTCTimestamp(left: Expression, right: Expression)
-  extends BinaryExpression with ImplicitCastInputTypes {
+  extends BinaryExpression with ExpectsInputTypes {
 
   override def inputTypes: Seq[AbstractDataType] = Seq(TimestampType, StringType)
   override def dataType: DataType = TimestampType
@@ -1221,7 +1262,7 @@ case class MonthsBetween(
   since = "1.5.0")
 // scalastyle:on line.size.limit
 case class ToUTCTimestamp(left: Expression, right: Expression)
-  extends BinaryExpression with ImplicitCastInputTypes {
+  extends BinaryExpression with ExpectsInputTypes {
 
   override def inputTypes: Seq[AbstractDataType] = Seq(TimestampType, StringType)
   override def dataType: DataType = TimestampType

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

@@ -296,10 +296,27 @@ object DateTimeUtils {
    * `T[h]h:[m]m:[s]s.[ms][ms][ms][us][us][us]+[h]h:[m]m`
    */
   def stringToTimestamp(s: UTF8String): Option[SQLTimestamp] = {
-    stringToTimestamp(s, defaultTimeZone())
+    stringToTimestamp(s, defaultTimeZone(), forceTimezone = false)
   }
 
   def stringToTimestamp(s: UTF8String, timeZone: TimeZone): Option[SQLTimestamp] = {
+    stringToTimestamp(s, timeZone, forceTimezone = false)
+  }
+
+  /**
+   * Converts a timestamp string to microseconds from the unix epoch, w.r.t. the given timezone.
+   * Returns None if the input string is not a valid timestamp format.
+   *
+   * @param s the input timestamp string.
+   * @param timeZone the timezone of the timestamp string, will be ignored if the timestamp string
+   *                 already contains timezone information and `forceTimezone` is false.
+   * @param forceTimezone if true, force to apply the given timezone to the timestamp string. If the
+   *                      timestamp string already contains timezone, return None.
+   */
+  def stringToTimestamp(
+      s: UTF8String,
+      timeZone: TimeZone,
+      forceTimezone: Boolean): Option[SQLTimestamp] = {
     if (s == null) {
       return None
     }
@@ -417,6 +434,8 @@ object DateTimeUtils {
       return None
     }
 
+    if (tz.isDefined && forceTimezone) return None
+
     val c = if (tz.isEmpty) {
       Calendar.getInstance(timeZone)
     } else {

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

@@ -27,3 +27,8 @@ select current_date = current_date(), current_timestamp = current_timestamp(), a
 select a, b from ttf2 order by a, current_date;
 
 select weekday('2007-02-03'), weekday('2009-07-30'), weekday('2017-05-27'), weekday(null), weekday('1582-10-15 13:10:15');
+
+-- SPARK-23715: the input of to/from_utc_timestamp can not have timezone
+select from_utc_timestamp('2000-10-10 00:00:00+00:00', 'PST');
+
+select to_utc_timestamp('2000-10-10 00:00:00+00:00', 'PST');

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

@@ -1,5 +1,5 @@
 -- Automatically generated by SQLQueryTestSuite
--- Number of queries: 10
+-- Number of queries: 12
 
 
 -- !query 0
@@ -82,9 +82,26 @@ struct<a:int,b:int>
 1	2
 2	3
 
+
 -- !query 9
 select weekday('2007-02-03'), weekday('2009-07-30'), weekday('2017-05-27'), weekday(null), weekday('1582-10-15 13:10:15')
--- !query 3 schema
+-- !query 9 schema
 struct<weekday(CAST(2007-02-03 AS DATE)):int,weekday(CAST(2009-07-30 AS DATE)):int,weekday(CAST(2017-05-27 AS DATE)):int,weekday(CAST(NULL AS DATE)):int,weekday(CAST(1582-10-15 13:10:15 AS DATE)):int>
--- !query 3 output
+-- !query 9 output
 5	3	5	NULL	4
+
+
+-- !query 10
+select from_utc_timestamp('2000-10-10 00:00:00+00:00', 'PST')
+-- !query 10 schema
+struct<from_utc_timestamp(string_to_timestamp(2000-10-10 00:00:00+00:00), PST):timestamp>
+-- !query 10 output
+NULL
+
+
+-- !query 11
+select to_utc_timestamp('2000-10-10 00:00:00+00:00', 'PST')
+-- !query 11 schema
+struct<to_utc_timestamp(string_to_timestamp(2000-10-10 00:00:00+00:00), PST):timestamp>
+-- !query 11 output
+NULL