Adds type coersion rule

Author: subham611

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

@@ -93,6 +93,7 @@ object AnsiTypeCoercion extends TypeCoercionBase {
       StackCoercion ::
       Division ::
       IntegralDivision ::
+      RewriteTimeCastToTimestampNTZ ::
       ImplicitTypeCasts ::
       DateTimeOperations ::
       WindowFrameCoercion ::

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

@@ -66,6 +66,7 @@ object TypeCoercion extends TypeCoercionBase {
       StackCoercion ::
       Division ::
       IntegralDivision ::
+      RewriteTimeCastToTimestampNTZ ::
       ImplicitTypeCasts ::
       DateTimeOperations ::
       WindowFrameCoercion ::

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

@@ -19,39 +19,15 @@ package org.apache.spark.sql.catalyst.analysis
 
 import scala.annotation.tailrec
 import scala.collection.mutable
-
-import org.apache.spark.sql.catalyst.expressions.{
-  Alias,
-  CaseWhen,
-  Cast,
-  Concat,
-  Elt,
-  Expression,
-  MapZipWith,
-  Stack,
-  WindowSpecDefinition
-}
-import org.apache.spark.sql.catalyst.plans.logical.{
-  AddColumns,
-  AlterColumns,
-  Call,
-  CreateTable,
-  Except,
-  Intersect,
-  LogicalPlan,
-  Project,
-  ReplaceTable,
-  Union,
-  UnionLoop,
-  Unpivot
-}
+import org.apache.spark.sql.catalyst.expressions.{Alias, CaseWhen, Cast, Concat, CurrentDate, Elt, Expression, MakeTimestampNTZ, MapZipWith, Stack, WindowSpecDefinition}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumns, AlterColumns, Call, CreateTable, Except, Intersect, LogicalPlan, Project, ReplaceTable, Union, UnionLoop, Unpivot}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
 import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure
 import org.apache.spark.sql.errors.DataTypeErrors.cannotMergeIncompatibleDataTypesError
-import org.apache.spark.sql.types.DataType
+import org.apache.spark.sql.types.{DataType, TimeType, TimestampNTZType}
 
 abstract class TypeCoercionBase extends TypeCoercionHelper {
 
@@ -480,4 +456,35 @@ abstract class TypeCoercionBase extends TypeCoercionHelper {
       case withChildrenResolved => StringLiteralTypeCoercion(withChildrenResolved)
     }
   }
+
+  /**
+   * Rewrites a cast from [[TimeType]] to [[TimestampNTZType]] into a [[MakeTimestampNTZ]]
+   * expression.
+   *
+   * The conversion from TIME to TIMESTAMP_NTZ requires a date component, which TIME itself does
+   * not provide. This rule injects [[CurrentDate]] as the implicit date part, effectively
+   * treating the TIME value as a time of day on the current date. This rewrite ensures that all
+   * such casts within a query use a consistent date, as required by the [[ComputeCurrentTime]]
+   * rule which replaces [[CurrentDate]] with a fixed value during analysis.
+   *
+   * For example, the following SQL:
+   * {{{
+   *   SELECT CAST(make_time(12, 30, 0) AS TIMESTAMP_NTZ)
+   * }}}
+   * will be rewritten to:
+   * {{{
+   *   SELECT make_timestamp_ntz(current_date, make_time(12, 30, 0))
+   * }}}
+   */
+  object RewriteTimeCastToTimestampNTZ extends TypeCoercionRule {
+    override def transform: PartialFunction[Expression, Expression] = {
+      case c @ Cast(child, TimestampNTZType, _, _) if child.resolved =>
+        child.dataType match {
+          case _: TimeType =>
+            // Convert TIME -> TIMESTAMP_NTZ using MakeTimestampNTZ(CurrentDate(), time)
+            MakeTimestampNTZ(CurrentDate(), child)
+          case _ => c
+        }
+    }
+  }
 }

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

@@ -1772,6 +1772,13 @@ class TypeCoercionSuite extends TypeCoercionSuiteBase {
     assert(wp1.isInstanceOf[Project])
     assert(wp1.expressions.forall(!_.exists(_ == t1.output.head)))
   }
+
+  test("SPARK-52617: RewriteTimeCastToTimestampNTZ: TIME to TIMESTAMP_NTZ coercion") {
+    val expr = Cast(Literal.create(123456789L, TimeType(6)), TimestampNTZType)
+    val coerced = RewriteTimeCastToTimestampNTZ.transform.apply(expr)
+    val expected = MakeTimestampNTZ(CurrentDate(), Literal.create(123456789L, TimeType(6)))
+    assert(coerced.semanticEquals(expected))
+  }
 }