[SPARK-21739][SQL]Cast expression should initialize timezoneId when it is called statically to convert something into TimestampType

sql/hive/src/main/scala/org/apache/spark/sql/hive/TableReader.scala

@@ -227,7 +227,8 @@ class HadoopTableReader(
       def fillPartitionKeys(rawPartValues: Array[String], row: InternalRow): Unit = {
         partitionKeyAttrs.foreach { case (attr, ordinal) =>
           val partOrdinal = partitionKeys.indexOf(attr)
-          row(ordinal) = Cast(Literal(rawPartValues(partOrdinal)), attr.dataType).eval(null)
+          row(ordinal) = Cast(Literal(rawPartValues(partOrdinal)), attr.dataType,
+            Option(sparkSession.sessionState.conf.sessionLocalTimeZone)).eval(null)
         }
       }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/execution/HiveTableScanExec.scala

@@ -104,7 +104,8 @@ case class HiveTableScanExec(
     hadoopConf)
 
   private def castFromString(value: String, dataType: DataType) = {
-    Cast(Literal(value), dataType).eval(null)
+    Cast(Literal(value), dataType,
+      Option(sparkSession.sessionState.conf.sessionLocalTimeZone)).eval(null)
   }
 
   private def addColumnMetadataToConf(hiveConf: Configuration): Unit = {

sql/hive/src/test/scala/org/apache/spark/sql/hive/QueryPartitionSuite.scala

@@ -68,4 +68,25 @@ class QueryPartitionSuite extends QueryTest with SQLTestUtils with TestHiveSingl
       sql("DROP TABLE IF EXISTS createAndInsertTest")
     }
   }
+
+  test("SPARK-21739: Cast expression should initialize timezoneId " +
+    "when it is called statically to convert something into TimestampType") {
+    // create table for test
+    sql("CREATE TABLE table_with_timestamp_partition(value int) PARTITIONED by (ts timestamp)")
+    sql("INSERT OVERWRITE TABLE table_with_timestamp_partition " +
+      "partition (ts = '2010-01-01 00:00:00.000') VALUES (1)")
+    sql("INSERT OVERWRITE TABLE table_with_timestamp_partition " +
+      "partition (ts = '2010-01-02 00:00:00.000') VALUES (2)")
+
+    // test for Cast expression in TableReader
+    checkAnswer(sql("select value from table_with_timestamp_partition"),
+      Seq(Row(1), Row(2)))
+
+    // test for Cast expression in HiveTableScanExec
+    checkAnswer(sql("select value from table_with_timestamp_partition " +
+      "where ts = '2010-01-02 00:00:00.000'"), Row(2))
+
+    sql("DROP TABLE IF EXISTS table_with_timestamp_partition")
+  }
+
 }