Adds RewriteTimeCastToTimestampNTZ rule

Author: subham611

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

@@ -19,22 +19,14 @@ package org.apache.spark.sql.catalyst.analysis
 
 import java.util
 import java.util.Locale
-
 import scala.collection.mutable
 import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._
 import scala.util.{Failure, Random, Success, Try}
-
 import org.apache.spark.{SparkException, SparkThrowable, SparkUnsupportedOperationException}
 import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst._
-import org.apache.spark.sql.catalyst.analysis.resolver.{
-  AnalyzerBridgeState,
-  HybridAnalyzer,
-  Resolver => OperatorResolver,
-  ResolverExtension,
-  ResolverGuard
-}
+import org.apache.spark.sql.catalyst.analysis.resolver.{AnalyzerBridgeState, HybridAnalyzer, ResolverExtension, ResolverGuard, Resolver => OperatorResolver}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.encoders.OuterScopes
 import org.apache.spark.sql.catalyst.expressions._
@@ -49,7 +41,7 @@ import org.apache.spark.sql.catalyst.trees.AlwaysProcess
 import org.apache.spark.sql.catalyst.trees.CurrentOrigin.withOrigin
 import org.apache.spark.sql.catalyst.trees.TreePattern._
 import org.apache.spark.sql.catalyst.types.DataTypeUtils
-import org.apache.spark.sql.catalyst.util.{toPrettySQL, trimTempResolvedColumn, CharVarcharUtils}
+import org.apache.spark.sql.catalyst.util.{CharVarcharUtils, toPrettySQL, trimTempResolvedColumn}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
 import org.apache.spark.sql.connector.catalog.{View => _, _}
 import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
@@ -483,6 +475,7 @@ 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/RewriteTimeCastToTimestampNTZ.scala

@@ -0,0 +1,52 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.expressions.{Cast, CurrentDate, MakeTimestampNTZ}
+import org.apache.spark.sql.catalyst.optimizer.ComputeCurrentTime
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.types.{TimestampNTZType, TimeType}
+
+/**
+ * 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))
+ * }}}
+ *
+ * This transformation must happen during resolution, before expression evaluation or
+ * optimization.
+ */
+object RewriteTimeCastToTimestampNTZ extends Rule[LogicalPlan] {
+  override def apply(plan: LogicalPlan): LogicalPlan = plan transformAllExpressions {
+    case c @ Cast(child, TimestampNTZType, _, _) if child.dataType.isInstanceOf[TimeType] =>
+      MakeTimestampNTZ(CurrentDate(), child)
+  }
+}

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

@@ -705,16 +705,6 @@ case class Cast(
       buildCast[Int](_, d => daysToMicros(d, ZoneOffset.UTC))
     case TimestampType =>
       buildCast[Long](_, ts => convertTz(ts, ZoneOffset.UTC, zoneId))
-    case _: TimeType =>
-      val currentDay = DateTimeUtils.currentDate(zoneId)
-      buildCast[Long](
-        _,
-        nanos => {
-          val NANOS_PER_DAY = 86_400_000_000_000L  // 24 * 60 * 60 * 1_000_000_000
-          val nanosOfDay = ((nanos % NANOS_PER_DAY) + NANOS_PER_DAY) % NANOS_PER_DAY
-          DateTimeUtils.makeTimestampNTZ(currentDay, nanosOfDay)
-        }
-      )
   }
 
   private[this] def decimalToTimestamp(d: Decimal): Long = {

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

@@ -0,0 +1,42 @@
+/*
+ * 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
+
+import org.apache.spark.sql.catalyst.expressions.{Alias, Cast, CurrentDate, Literal, MakeTimestampNTZ}
+import org.apache.spark.sql.catalyst.plans.logical.{OneRowRelation, Project}
+import org.apache.spark.sql.types.{TimestampNTZType, TimeType}
+
+class RewriteTimeCastToTimestampNTZSuite extends AnalysisTest {
+
+  test("SPARK-52617: RewriteTimeCastToTimestampNTZ rewrites") {
+
+    // TIME: 15:30:00 => nanos = 15 * 3600 + 30 * 60 = 55800s => nanos
+    val nanos = 55800L * 1_000_000_000L
+    val timeLiteral = Literal(nanos, TimeType(6))
+
+    val originalPlan =
+      Project(Seq(Alias(Cast(timeLiteral, TimestampNTZType), "ts")()), OneRowRelation())
+    val expectedPlan = Project(
+      Seq(
+        Alias(MakeTimestampNTZ(CurrentDate(), Literal.create(timeLiteral, TimeType(6))), "ts")()),
+      OneRowRelation())
+
+    val rewrittenPlan = RewriteTimeCastToTimestampNTZ(originalPlan)
+    comparePlans(rewrittenPlan, expectedPlan)
+  }
+}