[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

@@ -41,6 +41,7 @@ import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.unsafe.types.UTF8String
 import org.apache.spark.util.{SerializableConfiguration, Utils}
 
@@ -227,7 +228,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(SQLConf.get.sessionLocalTimeZone)).eval(null)
         }
       }
 

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

@@ -37,6 +37,7 @@ import org.apache.spark.sql.execution._
 import org.apache.spark.sql.execution.metric.SQLMetrics
 import org.apache.spark.sql.hive._
 import org.apache.spark.sql.hive.client.HiveClientImpl
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.{BooleanType, DataType}
 import org.apache.spark.util.Utils
 
@@ -104,7 +105,7 @@ case class HiveTableScanExec(
     hadoopConf)
 
   private def castFromString(value: String, dataType: DataType) = {
-    Cast(Literal(value), dataType).eval(null)
+    Cast(Literal(value), dataType, Option(SQLConf.get.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")
+  }
+
 }