Handle TimestampType separately when convert to pandas' dtype. (#798)

When the initialization of pandas in pyarrow is not done yet, it can't convert `pa.TimestampType` to pandas' dtype. In that case, the following example raises an error:

```py
from datetime import datetime
import databricks.koalas as ks

kdf = ks.DataFrame({'t': [datetime(2019, 1, 1, 0, 0, 0), datetime(2019, 1, 2, 0, 0, 0), datetime(2019, 1, 3, 0, 0, 0)]})
kdf[kdf['t'] != kdf['t']]
```

```py
>>> kdf[kdf['t'] != kdf['t']]
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/ueshin/workspace/databricks-koalas/master/databricks/koalas/frame.py", line 6646, in __repr__
    pdf = self.head(max_display_count + 1)._to_internal_pandas()
  File "/Users/ueshin/workspace/databricks-koalas/master/databricks/koalas/frame.py", line 6639, in _to_internal_pandas
    return self._internal.pandas_df
  File "/Users/ueshin/workspace/databricks-koalas/master/databricks/koalas/utils.py", line 338, in _lazy_property
    setattr(self, attr_name, fn(self))
  File "/Users/ueshin/workspace/databricks-koalas/master/databricks/koalas/internal.py", line 638, in pandas_df
    for field in sdf.schema})
  File "/Users/ueshin/workspace/databricks-koalas/master/databricks/koalas/internal.py", line 638, in <dictcomp>
    for field in sdf.schema})
  File "pyarrow/types.pxi", line 404, in pyarrow.lib.TimestampType.to_pandas_dtype
  File "/Users/ueshin/workspace/databricks-koalas/miniconda/envs/databricks-koalas_3.6/lib/python3.6/site-packages/pyarrow/pandas_compat.py", line 625, in make_datetimetz
    return _pandas_api.datetimetz_type('ns', tz=tz)
TypeError: 'NoneType' object is not callable
```

We know the dtype should be `np.dtype('datetime64[ns]')`, so we don't need to rely on pyarrow's implementation.

```py
>>> kdf[kdf['t'] != kdf['t']]
Empty DataFrame
Columns: [t]
Index: []
```

Resolves #772.

Author: ueshin

databricks/koalas/base.py

@@ -26,12 +26,11 @@
 from pandas.api.types import is_list_like
 from pyspark import sql as spark
 from pyspark.sql import functions as F, Window
-from pyspark.sql.types import DoubleType, FloatType, LongType, StringType, TimestampType, \
-    to_arrow_type
+from pyspark.sql.types import DoubleType, FloatType, LongType, StringType, TimestampType
 
 from databricks import koalas as ks  # For running doctests and reference resolution in PyCharm.
 from databricks.koalas.internal import _InternalFrame
-from databricks.koalas.typedef import pandas_wraps
+from databricks.koalas.typedef import pandas_wraps, spark_type_to_pandas_dtype
 from databricks.koalas.utils import align_diff_series, scol_for
 
 
@@ -219,10 +218,7 @@ def dtype(self):
         >>> s.rename("a").to_frame().set_index("a").index.dtype
         dtype('<M8[ns]')
         """
-        if type(self.spark_type) == TimestampType:
-            return np.dtype('datetime64[ns]')
-        else:
-            return np.dtype(to_arrow_type(self.spark_type).to_pandas_dtype())
+        return spark_type_to_pandas_dtype(self.spark_type)
 
     @property
     def empty(self):

databricks/koalas/internal.py

@@ -32,7 +32,7 @@
 
 from databricks import koalas as ks  # For running doctests and reference resolution in PyCharm.
 from databricks.koalas.config import get_option
-from databricks.koalas.typedef import infer_pd_series_spark_type
+from databricks.koalas.typedef import infer_pd_series_spark_type, spark_type_to_pandas_dtype
 from databricks.koalas.utils import column_index_level, default_session, lazy_property, scol_for
 
 
@@ -634,7 +634,7 @@ def pandas_df(self):
         sdf = self.spark_internal_df
         pdf = sdf.toPandas()
         if len(pdf) == 0 and len(sdf.schema) > 0:
-            pdf = pdf.astype({field.name: to_arrow_type(field.dataType).to_pandas_dtype()
+            pdf = pdf.astype({field.name: spark_type_to_pandas_dtype(field.dataType)
                               for field in sdf.schema})
 
         index_columns = self.index_columns

databricks/koalas/tests/test_dataframe.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 #
 
+from datetime import date, datetime
 import inspect
 
 import numpy as np
@@ -1642,3 +1643,11 @@ def test_transform(self):
 
         with self.assertRaisesRegex(AssertionError, "the first argument should be a callable"):
             kdf.transform(1)
+
+    def test_empty_timestamp(self):
+        pdf = pd.DataFrame({'t': [datetime(2019, 1, 1, 0, 0, 0),
+                                  datetime(2019, 1, 2, 0, 0, 0),
+                                  datetime(2019, 1, 3, 0, 0, 0)]})
+        kdf = ks.from_pandas(pdf)
+        self.assert_eq(kdf[kdf['t'] != kdf['t']], pdf[pdf['t'] != pdf['t']])
+        self.assert_eq(kdf[kdf['t'] != kdf['t']].dtypes, pdf[pdf['t'] != pdf['t']].dtypes)

databricks/koalas/typedef.py

@@ -145,6 +145,14 @@ def as_spark_type(tpe) -> types.DataType:
     return _known_types.get(tpe, None)
 
 
+def spark_type_to_pandas_dtype(spark_type):
+    """ Return the given Spark DataType to pandas dtype. """
+    if isinstance(spark_type, types.TimestampType):
+        return np.dtype('datetime64[ns]')
+    else:
+        return np.dtype(types.to_arrow_type(spark_type).to_pandas_dtype())
+
+
 def as_python_type(spark_tpe):
     return _py_conversions.get(spark_tpe, None)