add TimestampType int handling in EvaluatePython, remove casting to long in Python

Author: JasonMWhite

python/pyspark/sql/tests.py

@@ -1438,8 +1438,10 @@ def test_time_with_timezone(self):
     # regression test for SPARK-19561
     def test_datetime_at_epoch(self):
         epoch = datetime.datetime.fromtimestamp(0)
-        df = self.spark.createDataFrame([Row(date=epoch)])
-        self.assertEqual(df.first()['date'], epoch)
+        df = self.spark.createDataFrame([Row(date=epoch)]).select('date', lit(epoch).alias('lit_date'))
+        first = df.first()
+        self.assertEqual(first['date'], epoch)
+        self.assertEqual(first['lit_date'], epoch)
 
     def test_decimal(self):
         from decimal import Decimal

python/pyspark/sql/types.py

@@ -189,7 +189,7 @@ def toInternal(self, dt):
         if dt is not None:
             seconds = (calendar.timegm(dt.utctimetuple()) if dt.tzinfo
                        else time.mktime(dt.timetuple()))
-            return long(seconds) * 1000000 + dt.microsecond
+            return int(seconds) * 1000000 + dt.microsecond
 
     def fromInternal(self, ts):
         if ts is not None:

sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala

@@ -112,6 +112,7 @@ object EvaluatePython {
     case (c: Int, DateType) => c
 
     case (c: Long, TimestampType) => c
+    case (c: Int, TimestampType) => c.toLong
 
     case (c, StringType) => UTF8String.fromString(c.toString)