[SPARK-50870][SQL] Add the timezone when casting to timestamp in V2ScanRelationPushDown

changgyoopark-db · cloud-fan · commit 24abb0fbb1db · 2025-01-21T12:02:04.000+08:00
### What changes were proposed in this pull request? Add the timezone information to a cast expression when the destination type requires it. ### Why are the changes needed? When current_timestamp() is materialized as a string, the timezone information is gone (e.g., 2024-12-27 10:26:27.684158) which prohibits further optimization rules from being applied to the affected data source. For example, ``` Project [1735900357973433#10 AS current_timestamp()#6] +- 'Project [cast(2025-01-03 10:32:37.973433#11 as timestamp) AS 1735900357973433#10] +- RelationV2[2025-01-03 10:32:37.973433#11] xxx ``` -> This query fails to execute because the injected cast expression lacks the timezone information. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Existing tests. ### Was this patch authored or co-authored using generative AI tooling? No. Closes #49549 from changgyoopark-db/SPARK-50870. Authored-by: changgyoopark-db <changgyoo.park@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala
@@ -328,7 +328,12 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
     if (expression.dataType == expectedDataType) {
       expression
     } else {
-      Cast(expression, expectedDataType)
+      val cast = Cast(expression, expectedDataType)
+      if (cast.timeZoneId.isEmpty && cast.needsTimeZone) {
+        cast.withTimeZone(conf.sessionLocalTimeZone)
+      } else {
+        cast
+      }
     }
 
   def buildScanWithPushedAggregate(plan: LogicalPlan): LogicalPlan = plan.transform {
