[SPARK-52617][SQL]Cast TIME to/from TIMESTAMP_NTZ

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/RewriteTimeCastToTimestampNTZ.scala

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.sql.catalyst.expressions.{Cast, CurrentDate, Expression, MakeTimestampNTZ}
+import org.apache.spark.sql.types.{TimestampNTZType, TimeType}
+
+/**
+ * Rewrites Cast from TIME -> TIMESTAMP_NTZ to MakeTimestampNTZ(CurrentDate(), TIME)
+ *
+ * Example: CAST(TIME '15:30:00' AS TIMESTAMP_NTZ) => MakeTimestampNTZ(CurrentDate(),
+ * Literal(15:30:00))
+ */
+object RewriteTimeCastToTimestampNTZ {
+  def rewrite(expr: Expression): Expression = expr match {
+    case c @ Cast(child, TimestampNTZType, _, _)
+        if child.resolved && child.dataType.isInstanceOf[TimeType] =>
+      MakeTimestampNTZ(CurrentDate(), child)
+    case other =>
+      other
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/TimezoneAwareExpressionResolver.scala

@@ -54,8 +54,12 @@ class TimezoneAwareExpressionResolver(expressionResolver: ExpressionResolver)
   override def resolve(unresolvedTimezoneExpression: TimeZoneAwareExpression): Expression = {
     val expressionWithResolvedChildren =
       withResolvedChildren(unresolvedTimezoneExpression, expressionResolver.resolve _)
+
+    val expressionWithResolvedCast =
+      RewriteTimeCastToTimestampNTZ.rewrite(expressionWithResolvedChildren)
+
     val expressionWithResolvedChildrenAndTimeZone = TimezoneAwareExpressionResolver.resolveTimezone(
-      expressionWithResolvedChildren,
+      expressionWithResolvedCast,
       traversals.current.sessionLocalTimeZone
     )
     coerceExpressionTypes(

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

@@ -122,6 +122,7 @@ object Cast extends QueryErrorsBase {
     case (_: StringType, _: TimeType) => true
     case (TimestampType, DateType) => true
     case (TimestampNTZType, DateType) => true
+    case (TimestampNTZType, _: TimeType) => true
 
     case (_: NumericType, _: NumericType) => true
     case (_: StringType, _: NumericType) => true
@@ -135,6 +136,7 @@ object Cast extends QueryErrorsBase {
     case (_, VariantType) => variant.VariantGet.checkDataType(from, allowStructsAndMaps = false)
 
     case (_: TimeType, _: TimeType) => true
+    case (_: TimeType, TimestampNTZType) => true
 
     // non-null variants can generate nulls even in ANSI mode
     case (ArrayType(fromType, fn), ArrayType(toType, tn)) =>
@@ -229,6 +231,7 @@ object Cast extends QueryErrorsBase {
     case (_: StringType, _: TimeType) => true
     case (TimestampType, DateType) => true
     case (TimestampNTZType, DateType) => true
+    case (TimestampNTZType, _: TimeType) => true
 
     case (_: StringType, CalendarIntervalType) => true
     case (_: StringType, _: DayTimeIntervalType) => true
@@ -254,6 +257,7 @@ object Cast extends QueryErrorsBase {
     case (_, VariantType) => variant.VariantGet.checkDataType(from, allowStructsAndMaps = false)
 
     case (_: TimeType, _: TimeType) => true
+    case (_: TimeType, TimestampNTZType) => true
 
     case (ArrayType(fromType, fn), ArrayType(toType, tn)) =>
       canCast(fromType, toType) &&
@@ -746,6 +750,13 @@ case class Cast(
       }
     case _: TimeType =>
       buildCast[Long](_, nanos => DateTimeUtils.truncateTimeToPrecision(nanos, to.precision))
+    case _: TimestampNTZType =>
+      buildCast[Long](
+        _,
+        micros => {
+          val nanosInDay = DateTimeUtils.getNanosInADay(micros)
+          DateTimeUtils.truncateTimeToPrecision(nanosInDay, to.precision)
+        })
   }
 
   // IntervalConverter

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

@@ -834,4 +834,19 @@ object DateTimeUtils extends SparkDateTimeUtils {
   def makeTimestampNTZ(days: Int, nanos: Long): Long = {
     localDateTimeToMicros(LocalDateTime.of(daysToLocalDate(days), nanosToLocalTime(nanos)))
   }
+
+  /**
+   * Returns the number of nanoseconds past midnight for a given timestamp in microseconds.
+   *
+   * This method uses `toJulianDay`, which splits the timestamp into (Julian day, nanoseconds in
+   * day), and returns only the nanosecond component.
+   *
+   * @param micros
+   *   The timestamp in microseconds since the epoch.
+   * @return
+   *   The number of nanoseconds past midnight on that day.
+   */
+  def getNanosInADay(micros: Long): Long = {
+    toJulianDay(micros)._2
+  }
 }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/resolver/RewriteTimeCastToTimestampNTZSuite.scala

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.analysis.resolver
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.sql.catalyst.expressions.{Cast, CurrentDate, Literal, MakeTimestampNTZ}
+import org.apache.spark.sql.types.{TimestampNTZType, TimeType}
+
+class RewriteTimeCastToTimestampNTZSuite extends SparkFunSuite {
+
+  test("SPARK-52617: rewrite TIME -> TIMESTAMP_NTZ cast to MakeTimestampNTZ") {
+    // TIME: 15:30:00 -> seconds = 15*3600 + 30*60 = 55800
+    val nanos = 55800L * 1_000_000_000L
+    val timeLiteral = Literal(nanos, TimeType(6))
+
+    val castExpr = Cast(timeLiteral, TimestampNTZType)
+    val rewrittenExpr = RewriteTimeCastToTimestampNTZ.rewrite(castExpr)
+
+    val expectedExpr = MakeTimestampNTZ(CurrentDate(), timeLiteral)
+
+    assert(
+      rewrittenExpr.semanticEquals(expectedExpr),
+      s"""
+         |Expected:
+         |  $expectedExpr
+         |But got:
+         |  $rewrittenExpr
+         |""".stripMargin)
+  }
+
+  test("should not rewrite non-time casts") {
+    val literal = Literal(42)
+    val castExpr = Cast(literal, TimestampNTZType)
+
+    val rewrittenExpr = RewriteTimeCastToTimestampNTZ.rewrite(castExpr)
+    assert(rewrittenExpr eq castExpr)
+  }
+}

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/CastSuiteBase.scala

@@ -31,7 +31,7 @@ import org.apache.spark.sql.catalyst.expressions.codegen.CodegenContext
 import org.apache.spark.sql.catalyst.util.DateTimeConstants._
 import org.apache.spark.sql.catalyst.util.DateTimeTestUtils._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils._
-import org.apache.spark.sql.catalyst.util.IntervalUtils
+import org.apache.spark.sql.catalyst.util.{DateTimeUtils, IntervalUtils}
 import org.apache.spark.sql.catalyst.util.IntervalUtils.microsToDuration
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -1507,4 +1507,32 @@ abstract class CastSuiteBase extends SparkFunSuite with ExpressionEvalHelper {
       }
     }
   }
+
+  test("SPARK-52617: cast TimestampNTZType to time") {
+    specialTs.foreach { s =>
+      val ldt = LocalDateTime.parse(s) // parsed as local timestamp
+      val micros = DateTimeUtils.localDateTimeToMicros(ldt)
+
+      val nanosOfDay = ldt.toLocalTime().toNanoOfDay
+      val expected = DateTimeUtils.truncateTimeToPrecision(nanosOfDay, TimeType.DEFAULT_PRECISION)
+
+      checkEvaluation(Cast(Literal(micros, TimestampNTZType), TimeType(0)), expected)
+    }
+  }
+
+  test("SPARK-52617: cast time to TimestampNTZType") {
+    val testCases = Seq(
+      ("2023-01-01T15:30:00.123456789", 9),
+      ("2023-01-01T15:30:00.123456", 6),
+      ("2023-01-01T15:30:00", 0))
+
+    testCases.foreach { case (s, precision) =>
+      val ldt = LocalDateTime.parse(s)
+      val micros = DateTimeUtils.localDateTimeToMicros(ldt)
+      val nanosOfDay = ldt.toLocalTime().toNanoOfDay
+      val expected = DateTimeUtils.truncateTimeToPrecision(nanosOfDay, precision)
+
+      checkEvaluation(Cast(Literal(micros, TimestampNTZType), TimeType(precision)), expected)
+    }
+  }
 }

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

@@ -1250,4 +1250,27 @@ class DateTimeUtilsSuite extends SparkFunSuite with Matchers with SQLHelper {
     assert(truncateTimeToPrecision(localTime(23, 59, 59, 987654), 1) ==
       localTime(23, 59, 59, 900000))
   }
+
+  test("get nanos in a day") {
+    def toMicros(hours: Int, minutes: Int, seconds: Int): Long = {
+      val microsInSecond = 1_000_000L
+      (hours * 3600L + minutes * 60 + seconds) * microsInSecond
+    }
+
+    // Midnight: 00:00:00
+    val midnightMicros = toMicros(0, 0, 0)
+    assert(getNanosInADay(midnightMicros) === 0L)
+
+    // Noon: 12:00:00
+    val noonMicros = toMicros(12, 0, 0)
+    assert(getNanosInADay(noonMicros) === 12L * 3600 * 1_000_000_000L)
+
+    // 23:59:59
+    val endOfDayMicros = toMicros(23, 59, 59)
+    assert(getNanosInADay(endOfDayMicros) === ((23L * 3600 + 59 * 60 + 59) * 1_000_000_000L))
+
+    // 01:30:15
+    val earlyMorningMicros = toMicros(1, 30, 15)
+    assert(getNanosInADay(earlyMorningMicros) === ((1L * 3600 + 30 * 60 + 15) * 1_000_000_000L))
+  }
 }