SPARK-31710: TIMESTAMP_SECONDS, TIMESTAMP_MILLISECONDS and TIMESTAMP_…

…MICROSECONDS to timestamp transfer

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

@@ -424,6 +424,9 @@ object FunctionRegistry {
     expression[MakeInterval]("make_interval"),
     expression[DatePart]("date_part"),
     expression[Extract]("extract"),
+    expression[SecondsToTimestamp]("timestamp_seconds"),
+    expression[MilliSecondsToTimestamp]("timestamp_milliseconds"),
+    expression[MicroSecondsToTimestamp]("timestamp_microseconds"),
 
     // collection functions
     expression[CreateArray]("array"),

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

@@ -401,6 +401,92 @@ case class DayOfYear(child: Expression) extends UnaryExpression with ImplicitCas
   }
 }
 
+@ExpressionDescription(
+  usage = "_FUNC_(date) - Returns timestamp from seconds.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(1230219000);
+       "2008-12-25 07:30:00.0"
+  """,
+  group = "datetime_funcs",
+  since = "3.1.0")
+case class SecondsToTimestamp(child: Expression)
+  extends NumberToTimestampBase {
+
+  override def upScaleFactor: SQLTimestamp = MICROS_PER_SECOND
+
+  override def prettyName: String = "timestamp_seconds"
+}
+
+@ExpressionDescription(
+  usage = "_FUNC_(date) - Returns timestamp from milliseconds.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(1230219000000);
+       "2008-12-25 07:30:00.0"
+  """,
+  group = "datetime_funcs",
+  since = "3.1.0")
+case class MilliSecondsToTimestamp(child: Expression)
+  extends NumberToTimestampBase {
+
+  override def upScaleFactor: SQLTimestamp = MICROS_PER_MILLIS
+
+  override def prettyName: String = "timestamp_milliseconds"
+}
+
+@ExpressionDescription(
+  usage = "_FUNC_(date) - Returns timestamp from microseconds.",
+  examples = """
+    Examples:
+      > SELECT _FUNC_(1230219000000000);
+       "2008-12-25 07:30:00.0"
+  """,
+  group = "datetime_funcs",
+  since = "3.1.0")
+case class MicroSecondsToTimestamp(child: Expression)
+  extends NumberToTimestampBase {
+
+  override def upScaleFactor: SQLTimestamp = 1L
+
+  override def prettyName: String = "timestamp_microseconds"
+}
+
+abstract class NumberToTimestampBase extends UnaryExpression
+  with ImplicitCastInputTypes{
+
+  protected def upScaleFactor: Long
+
+  override def inputTypes: Seq[AbstractDataType] = Seq(LongType, IntegerType)
+
+  override def dataType: DataType = TimestampType
+
+  override def eval(input: InternalRow): Any = {
+    val t = child.eval(input)
+    if (t == null) {
+      null
+    } else {
+      child.dataType match {
+        case IntegerType =>
+          Math.multiplyExact(t.asInstanceOf[Int].toLong, upScaleFactor)
+        case LongType =>
+          Math.multiplyExact(t.asInstanceOf[Long], upScaleFactor)
+      }
+    }
+  }
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    child.dataType match {
+      case IntegerType =>
+        defineCodeGen(ctx, ev, _ => s"java.lang.Math.multiplyExact(" +
+          s"${ev.value.asInstanceOf[Integer].toLong}, ${upScaleFactor})")
+      case LongType =>
+        defineCodeGen(ctx, ev, _ => s"java.lang.Math.multiplyExact(" +
+          s"${ev.value.asInstanceOf[Long]}, ${upScaleFactor})")
+    }
+  }
+}
+
 @ExpressionDescription(
   usage = "_FUNC_(date) - Returns the year component of the date/timestamp.",
   examples = """

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -3495,6 +3495,28 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
     assert(df4.schema.head.name === "randn(1)")
     checkIfSeedExistsInExplain(df2)
   }
+
+  test("SPARK-31710: " +
+    "TIMESTAMP_SECONDS, TIMESTAMP_MILLISECONDS and TIMESTAMP_MICROSECONDS to timestamp transfer") {
+    val df1 = sql("select TIMESTAMP_SECONDS(1230219000) as timestamp")
+    checkAnswer(df1, Row(Timestamp.valueOf("2008-12-25 07:30:00.0")))
+
+    val df2 = sql("select TIMESTAMP_MILLISECONDS(1230219000000) as timestamp")
+    checkAnswer(df2, Row(Timestamp.valueOf("2008-12-25 07:30:00.0")))
+
+    val df3 = sql("select TIMESTAMP_MICROSECONDS(1230219000000000) as timestamp")
+    checkAnswer(df3, Row(Timestamp.valueOf("2008-12-25 07:30:00.0")))
+
+    val df4 = sql("select TIMESTAMP_SECONDS(-1230219000) as timestamp")
+    checkAnswer(df4, Row(Timestamp.valueOf("1931-01-07 00:30:00.0")))
+
+    val df5 = sql("select TIMESTAMP_MILLISECONDS(-1230219000000) as timestamp")
+    checkAnswer(df5, Row(Timestamp.valueOf("1931-01-07 00:30:00.0")))
+
+    val df6 = sql("select TIMESTAMP_MICROSECONDS(-1230219000000000) as timestamp")
+    checkAnswer(df6, Row(Timestamp.valueOf("1931-01-07 00:30:00.0")))
+
+  }
 }
 
 case class Foo(bar: Option[String])