Support non-UTC timezone for casting from date type to timestamp type [databricks]

integration_tests/src/main/python/cast_test.py

@@ -726,8 +726,6 @@ def test_cast_int_to_string_not_UTC():
         {"spark.sql.session.timeZone": "+08"})
 
 not_utc_fallback_test_params = [(timestamp_gen, 'STRING'),
-        # python does not like year 0, and with time zones the default start date can become year 0 :(
-        (DateGen(start=date(1, 1, 1)), 'TIMESTAMP'),
         (SetValuesGen(StringType(), ['2023-03-20 10:38:50', '2023-03-20 10:39:02']), 'TIMESTAMP')]
 
 @allow_non_gpu('ProjectExec')

integration_tests/src/main/python/date_time_test.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 import pytest
-from asserts import assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect, assert_gpu_and_cpu_error
+from asserts import assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect, assert_gpu_and_cpu_error, assert_gpu_and_cpu_are_equal_sql 
 from conftest import is_utc, is_supported_time_zone, get_test_tz
 from data_gen import *
 from datetime import date, datetime, timezone
@@ -671,3 +671,14 @@ def test_timestamp_millis_long_overflow():
 def test_timestamp_micros(data_gen):
     assert_gpu_and_cpu_are_equal_collect(
         lambda spark : unary_op_df(spark, data_gen).selectExpr("timestamp_micros(a)"))
+
+
+@pytest.mark.parametrize('parser_policy', ["LEGACY", "CORRECTED", "EXCEPTION"], ids=idfn)
+def test_date_to_timestamp(parser_policy):
+    assert_gpu_and_cpu_are_equal_sql(
+        lambda spark : unary_op_df(spark, date_gen),
+        "tab",
+        "SELECT cast(a as timestamp) from tab",
+        conf = {
+            "spark.sql.legacy.timeParserPolicy": "{}".format(parser_policy),
+            "spark.rapids.sql.incompatibleDateFormats.enabled": True})

sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuCast.scala

@@ -90,6 +90,7 @@ abstract class CastExprMetaBase[INPUT <: UnaryExpression with TimeZoneAwareExpre
   override def isTimeZoneSupported: Boolean = {
     (fromType, toType) match {
       case (TimestampType, DateType) => true // this is for to_date(...)
+      case (DateType, TimestampType) => true
       case _ => false
     }
   }
@@ -631,6 +632,11 @@ object GpuCast {
             zoneId.normalized())) {
           shifted => shifted.castTo(GpuColumnVector.getNonNestedRapidsType(toDataType))
         }
+      case (DateType, TimestampType) if options.timeZoneId.isDefined =>
+        val zoneId = DateTimeUtils.getZoneId(options.timeZoneId.get)
+        withResource(input.castTo(GpuColumnVector.getNonNestedRapidsType(toDataType))) { cv =>
+          GpuTimeZoneDB.fromTimestampToUtcTimestamp(cv, zoneId.normalized())
+        }
       case _ =>
         input.castTo(GpuColumnVector.getNonNestedRapidsType(toDataType))
     }