Adds type coersion rule

Author: subham611

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

@@ -483,7 +483,6 @@ class Analyzer(override val catalogManager: CatalogManager) extends RuleExecutor
         ResolveWithCTE,
         ExtractDistributedSequenceID) ++
       Seq(ResolveUpdateEventTimeWatermarkColumn) ++
-      Seq(RewriteTimeCastToTimestampNTZ) ++
       extendedResolutionRules : _*),
     Batch("Remove TempResolvedColumn", Once, RemoveTempResolvedColumn),
     Batch("Post-Hoc Resolution", Once,

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AnsiTypeCoercion.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/RewriteTimeCastToTimestampNTZ.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/TypeCoercion.scala

@@ -25,8 +25,10 @@ import org.apache.spark.sql.catalyst.expressions.{
   CaseWhen,
   Cast,
   Concat,
+  CurrentDate,
   Elt,
   Expression,
+  MakeTimestampNTZ,
   MapZipWith,
   Stack,
   WindowSpecDefinition
@@ -51,7 +53,7 @@ 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, TimestampNTZType, TimeType}
 
 abstract class TypeCoercionBase extends TypeCoercionHelper {
 
@@ -480,4 +482,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/TypeCoercionBase.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))
+  }
 }