Repository: databricks/koalas (language: python)

Issue:
Supporting nullable data types for float data, Float32 and Float64
pandas experimentally starts to support [nullable data types for float data](https://pandas.pydata.org/pandas-docs/dev/whatsnew/v1.2.0.html#experimental-nullable-data-types-for-float-data), Float32 and Float64 from pandas 1.2.

The difference from the existing float32 and float64 as below.

```python
# the default NumPy float64 dtype
In [11]: s1 = pd.Series([1.5, None])

In [12]: s1
Out[12]: 
0    1.5
1    NaN
Length: 2, dtype: float64

In [13]: s1 > 1
Out[13]: 
0     True
1    False
Length: 2, dtype: bool
```

```python
# the new nullable float64 dtype
In [14]: s2 = pd.Series([1.5, None], dtype="Float64")

In [15]: s2
Out[15]: 
0     1.5
1    <NA>
Length: 2, dtype: Float64

In [16]: s2 > 1
Out[16]: 
0    True
1    <NA>
Length: 2, dtype: boolean
```


Discussion / hints:
I'm investigating them as well as `Int64`, `Int32`, etc. or `boolean`, `string` types.

Fix patch:
diff --git a/databricks/koalas/base.py b/databricks/koalas/base.py
index eb0c431838..fa8ac82b9e 100644
--- a/databricks/koalas/base.py
+++ b/databricks/koalas/base.py
@@ -25,7 +25,7 @@
 
 import numpy as np
 import pandas as pd  # noqa: F401
-from pandas.api.types import is_list_like
+from pandas.api.types import is_list_like, pandas_dtype
 from pyspark import sql as spark
 from pyspark.sql import functions as F, Window, Column
 from pyspark.sql.types import (
@@ -51,7 +51,12 @@
 )
 from databricks.koalas.spark import functions as SF
 from databricks.koalas.spark.accessors import SparkIndexOpsMethods
-from databricks.koalas.typedef import as_spark_type, spark_type_to_pandas_dtype
+from databricks.koalas.typedef import (
+    Dtype,
+    as_spark_type,
+    extension_dtypes,
+    spark_type_to_pandas_dtype,
+)
 from databricks.koalas.utils import (
     combine_frames,
     same_anchor,
@@ -191,7 +196,7 @@ def align_diff_index_ops(func, this_index_ops: "IndexOpsMixin", *args) -> "Index
                 ).rename(that_series.name)
 
 
-def booleanize_null(left_scol, scol, f) -> Column:
+def booleanize_null(scol, f) -> Column:
     """
     Booleanize Null in Spark Column
     """
@@ -205,12 +210,6 @@ def booleanize_null(left_scol, scol, f) -> Column:
         filler = f == Column.__ne__
         scol = F.when(scol.isNull(), filler).otherwise(scol)
 
-    elif f == Column.__or__:
-        scol = F.when(left_scol.isNull() | scol.isNull(), False).otherwise(scol)
-
-    elif f == Column.__and__:
-        scol = F.when(scol.isNull(), False).otherwise(scol)
-
     return scol
 
 
@@ -239,13 +238,23 @@ def wrapper(self, *args):
             # Same DataFrame anchors
             args = [arg.spark.column if isinstance(arg, IndexOpsMixin) else arg for arg in args]
             scol = f(self.spark.column, *args)
-            scol = booleanize_null(self.spark.column, scol, f)
+
+            spark_type = self._internal.spark_frame.select(scol).schema[0].dataType
+            use_extension_dtypes = any(
+                isinstance(col.dtype, extension_dtypes) for col in [self] + cols
+            )
+            dtype = spark_type_to_pandas_dtype(
+                spark_type, use_extension_dtypes=use_extension_dtypes
+            )
+
+            if not isinstance(dtype, extension_dtypes):
+                scol = booleanize_null(scol, f)
 
             if isinstance(self, Series) or not any(isinstance(col, Series) for col in cols):
-                index_ops = self._with_new_scol(scol)
+                index_ops = self._with_new_scol(scol, dtype=dtype)
             else:
                 kser = next(col for col in cols if isinstance(col, Series))
-                index_ops = kser._with_new_scol(scol)
+                index_ops = kser._with_new_scol(scol, dtype=dtype)
         elif get_option("compute.ops_on_diff_frames"):
             index_ops = align_diff_index_ops(f, self, *args)
         else:
@@ -293,7 +302,7 @@ def _kdf(self) -> DataFrame:
         pass
 
     @abstractmethod
-    def _with_new_scol(self, scol: spark.Column):
+    def _with_new_scol(self, scol: spark.Column, *, dtype=None):
         pass
 
     @property
@@ -603,11 +612,63 @@ def rpow_func(left, right):
 
     # `and`, `or`, `not` cannot be overloaded in Python,
     # so use bitwise operators as boolean operators
-    __and__ = column_op(Column.__and__)
-    __or__ = column_op(Column.__or__)
+    def __and__(self, other) -> Union["Series", "Index"]:
+        if isinstance(self.dtype, extension_dtypes) or (
+            isinstance(other, IndexOpsMixin) and isinstance(other.dtype, extension_dtypes)
+        ):
+
+            def and_func(left, right):
+                if not isinstance(right, spark.Column):
+                    if pd.isna(right):
+                        right = F.lit(None)
+                    else:
+                        right = F.lit(right)
+                return left & right
+
+        else:
+
+            def and_func(left, right):
+                if not isinstance(right, spark.Column):
+                    if pd.isna(right):
+                        right = F.lit(None)
+                    else:
+                        right = F.lit(right)
+                scol = left & right
+                return F.when(scol.isNull(), False).otherwise(scol)
+
+        return column_op(and_func)(self, other)
+
+    def __or__(self, other) -> Union["Series", "Index"]:
+        if isinstance(self.dtype, extension_dtypes) or (
+            isinstance(other, IndexOpsMixin) and isinstance(other.dtype, extension_dtypes)
+        ):
+
+            def or_func(left, right):
+                if not isinstance(right, spark.Column):
+                    if pd.isna(right):
+                        right = F.lit(None)
+                    else:
+                        right = F.lit(right)
+                return left | right
+
+        else:
+
+            def or_func(left, right):
+                if not isinstance(right, spark.Column) and pd.isna(right):
+                    return F.lit(False)
+                else:
+                    scol = left | F.lit(right)
+                    return F.when(left.isNull() | scol.isNull(), False).otherwise(scol)
+
+        return column_op(or_func)(self, other)
+
     __invert__ = column_op(Column.__invert__)
-    __rand__ = column_op(Column.__rand__)
-    __ror__ = column_op(Column.__ror__)
+
+    def __rand__(self, other) -> Union["Series", "Index"]:
+        return self.__and__(other)
+
+    def __ror__(self, other) -> Union["Series", "Index"]:
+        return self.__or__(other)
 
     def __len__(self):
         return len(self._kdf)
@@ -632,7 +693,7 @@ def __array_ufunc__(self, ufunc: Callable, method: str, *inputs: Any, **kwargs:
             raise NotImplementedError("Koalas objects currently do not support %s." % ufunc)
 
     @property
-    def dtype(self) -> np.dtype:
+    def dtype(self) -> Dtype:
         """Return the dtype object of the underlying data.
 
         Examples
@@ -652,7 +713,7 @@ def dtype(self) -> np.dtype:
         >>> s.rename("a").to_frame().set_index("a").index.dtype
         dtype('<M8[ns]')
         """
-        return spark_type_to_pandas_dtype(self.spark.data_type)
+        return self._internal.data_dtypes[0]
 
     @property
     def empty(self) -> bool:
@@ -955,7 +1016,7 @@ def ndim(self) -> int:
         """
         return 1
 
-    def astype(self, dtype) -> Union["Index", "Series"]:
+    def astype(self, dtype: Union[str, type, Dtype]) -> Union["Index", "Series"]:
         """
         Cast a Koalas object to a specified dtype ``dtype``.
 
@@ -989,37 +1050,55 @@ def astype(self, dtype) -> Union["Index", "Series"]:
         >>> ser.rename("a").to_frame().set_index("a").index.astype('int64')
         Int64Index([1, 2], dtype='int64', name='a')
         """
+        dtype = pandas_dtype(dtype)
         spark_type = as_spark_type(dtype)
         if not spark_type:
             raise ValueError("Type {} not understood".format(dtype))
         if isinstance(spark_type, BooleanType):
-            if isinstance(self.spark.data_type, StringType):
-                scol = F.when(self.spark.column.isNull(), F.lit(False)).otherwise(
-                    F.length(self.spark.column) > 0
-                )
-            elif isinstance(self.spark.data_type, (FloatType, DoubleType)):
-                scol = F.when(
-                    self.spark.column.isNull() | F.isnan(self.spark.column), F.lit(True)
-                ).otherwise(self.spark.column.cast(spark_type))
+            if isinstance(dtype, extension_dtypes):
+                scol = self.spark.column.cast(spark_type)
             else:
-                scol = F.when(self.spark.column.isNull(), F.lit(False)).otherwise(
-                    self.spark.column.cast(spark_type)
-                )
+                if isinstance(self.spark.data_type, StringType):
+                    scol = F.when(self.spark.column.isNull(), F.lit(False)).otherwise(
+                        F.length(self.spark.column) > 0
+                    )
+                elif isinstance(self.spark.data_type, (FloatType, DoubleType)):
+                    scol = F.when(
+                        self.spark.column.isNull() | F.isnan(self.spark.column), F.lit(True)
+                    ).otherwise(self.spark.column.cast(spark_type))
+                else:
+                    scol = F.when(self.spark.column.isNull(), F.lit(False)).otherwise(
+                        self.spark.column.cast(spark_type)
+                    )
         elif isinstance(spark_type, StringType):
-            if isinstance(self.spark.data_type, NumericType):
-                null_str = str(np.nan)
-            elif isinstance(self.spark.data_type, (DateType, TimestampType)):
-                null_str = str(pd.NaT)
-            else:
-                null_str = str(None)
-            if isinstance(self.spark.data_type, BooleanType):
-                casted = F.when(self.spark.column, "True").otherwise("False")
+            if isinstance(dtype, extension_dtypes):
+                if isinstance(self.spark.data_type, BooleanType):
+                    scol = F.when(
+                        self.spark.column.isNotNull(),
+                        F.when(self.spark.column, "True").otherwise("False"),
+                    )
+                elif isinstance(self.spark.data_type, TimestampType):
+                    # seems like a pandas' bug?
+                    scol = F.when(self.spark.column.isNull(), str(pd.NaT)).otherwise(
+                        self.spark.column.cast(spark_type)
+                    )
+                else:
+                    scol = self.spark.column.cast(spark_type)
             else:
-                casted = self.spark.column.cast(spark_type)
-            scol = F.when(self.spark.column.isNull(), null_str).otherwise(casted)
+                if isinstance(self.spark.data_type, NumericType):
+                    null_str = str(np.nan)
+                elif isinstance(self.spark.data_type, (DateType, TimestampType)):
+                    null_str = str(pd.NaT)
+                else:
+                    null_str = str(None)
+                if isinstance(self.spark.data_type, BooleanType):
+                    casted = F.when(self.spark.column, "True").otherwise("False")
+                else:
+                    casted = self.spark.column.cast(spark_type)
+                scol = F.when(self.spark.column.isNull(), null_str).otherwise(casted)
         else:
             scol = self.spark.column.cast(spark_type)
-        return self._with_new_scol(scol)
+        return self._with_new_scol(scol, dtype=dtype)
 
     def isin(self, values) -> Union["Series", "Index"]:
         """
diff --git a/databricks/koalas/frame.py b/databricks/koalas/frame.py
index bdd780c433..492599d5d3 100644
--- a/databricks/koalas/frame.py
+++ b/databricks/koalas/frame.py
@@ -795,7 +795,7 @@ def apply_op(kdf, this_column_labels, that_column_labels):
                             .alias(name_like_string(label))
                         )
                         column_labels.append(label)
-                internal = self._internal.with_new_columns(applied, column_labels)
+                internal = self._internal.with_new_columns(applied, column_labels=column_labels)
                 return DataFrame(internal)
         else:
             return self._apply_series_op(lambda kser: getattr(kser, op)(other))
@@ -5504,7 +5504,7 @@ def head(self, n: int = 5) -> "DataFrame":
             sdf = self._internal.resolved_copy.spark_frame
             if get_option("compute.ordered_head"):
                 sdf = sdf.orderBy(NATURAL_ORDER_COLUMN_NAME)
-            return DataFrame(self._internal.with_new_sdf(sdf.limit(n)))
+            return DataFrame(self._internal.with_new_sdf(sdf.limit(n), preserve_dtypes=True))
 
     def pivot_table(
         self, values=None, index=None, columns=None, aggfunc="mean", fill_value=None
@@ -5958,17 +5958,12 @@ def columns(self, columns) -> None:
         else:
             column_label_names = None
 
-        data_columns = [name_like_string(label) for label in column_labels]
-        data_spark_columns = [
-            self._internal.spark_column_for(label).alias(name)
-            for label, name in zip(self._internal.column_labels, data_columns)
+        ksers = [
+            self._kser_for(label).rename(name)
+            for label, name in zip(self._internal.column_labels, column_labels)
         ]
         self._update_internal_frame(
-            self._internal.with_new_columns(
-                data_spark_columns,
-                column_labels=column_labels,
-                column_label_names=column_label_names,
-            )
+            self._internal.with_new_columns(ksers, column_label_names=column_label_names)
         )
 
     @property
@@ -6434,7 +6429,9 @@ def _sort(
         }
         by = [mapper[(asc, na_position)](scol) for scol, asc in zip(by, ascending)]
         sdf = self._internal.resolved_copy.spark_frame.sort(*(by + [NATURAL_ORDER_COLUMN_NAME]))
-        kdf = DataFrame(self._internal.with_new_sdf(sdf))  # type: ks.DataFrame
+        kdf = DataFrame(
+            self._internal.with_new_sdf(sdf, preserve_dtypes=True)
+        )  # type: ks.DataFrame
         if inplace:
             self._update_internal_frame(kdf._internal)
             return None
diff --git a/databricks/koalas/indexes/base.py b/databricks/koalas/indexes/base.py
index 03911b7487..cc00bb9625 100644
--- a/databricks/koalas/indexes/base.py
+++ b/databricks/koalas/indexes/base.py
@@ -159,7 +159,7 @@ def _internal(self) -> InternalFrame:
     def _column_label(self):
         return self._kdf._internal.index_names[0]
 
-    def _with_new_scol(self, scol: spark.Column) -> "Index":
+    def _with_new_scol(self, scol: spark.Column, *, dtype=None) -> "Index":
         """
         Copy Koalas Index with the new Spark Column.
 
@@ -168,8 +168,10 @@ def _with_new_scol(self, scol: spark.Column) -> "Index":
         """
         internal = self._internal.copy(
             index_spark_columns=[scol.alias(SPARK_DEFAULT_INDEX_NAME)],
+            index_dtypes=[dtype],
             column_labels=[],
             data_spark_columns=[],
+            data_dtypes=[],
         )
         return DataFrame(internal).index
 
diff --git a/databricks/koalas/indexes/multi.py b/databricks/koalas/indexes/multi.py
index 2613c571f3..82679af279 100644
--- a/databricks/koalas/indexes/multi.py
+++ b/databricks/koalas/indexes/multi.py
@@ -100,7 +100,7 @@ def _column_label(self):
     def __abs__(self):
         raise TypeError("TypeError: cannot perform __abs__ with this index type: MultiIndex")
 
-    def _with_new_scol(self, scol: spark.Column):
+    def _with_new_scol(self, scol: spark.Column, *, dtype=None):
         raise NotImplementedError("Not supported for type MultiIndex")
 
     def _align_and_column_op(self, f, *args) -> Index:
diff --git a/databricks/koalas/indexing.py b/databricks/koalas/indexing.py
index 99b35c1a99..a2e076404c 100644
--- a/databricks/koalas/indexing.py
+++ b/databricks/koalas/indexing.py
@@ -36,7 +36,12 @@
     SPARK_DEFAULT_SERIES_NAME,
 )
 from databricks.koalas.exceptions import SparkPandasIndexingError, SparkPandasNotImplementedError
-from databricks.koalas.typedef import Scalar
+from databricks.koalas.typedef.typehints import (
+    Dtype,
+    Scalar,
+    extension_dtypes,
+    spark_type_to_pandas_dtype,
+)
 from databricks.koalas.utils import (
     is_name_like_tuple,
     is_name_like_value,
@@ -272,7 +277,9 @@ def _select_rows(
 
     def _select_cols(
         self, cols_sel: Any, missing_keys: Optional[List[Tuple]] = None
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """
         Dispatch the logic for select columns to more specific methods by `cols_sel` argument types.
 
@@ -282,10 +289,11 @@ def _select_cols(
 
         Returns
         -------
-        Tuple of list of column label, list of Spark columns, bool:
+        Tuple of list of column label, list of Spark columns, list of dtypes, bool:
 
             * The column labels selected.
             * The Spark columns selected.
+            * The dtypes selected.
             * The boolean value whether Series should be returned or not.
             * The Series name if needed.
         """
@@ -294,7 +302,8 @@ def _select_cols(
         if cols_sel is None:
             column_labels = self._internal.column_labels
             data_spark_columns = self._internal.data_spark_columns
-            return column_labels, data_spark_columns, False, None
+            data_dtypes = self._internal.data_dtypes
+            return column_labels, data_spark_columns, data_dtypes, False, None
         elif isinstance(cols_sel, Series):
             return self._select_cols_by_series(cols_sel, missing_keys)
         elif isinstance(cols_sel, spark.Column):
@@ -304,7 +313,8 @@ def _select_cols(
                 # If slice is None - select everything, so nothing to do
                 column_labels = self._internal.column_labels
                 data_spark_columns = self._internal.data_spark_columns
-                return column_labels, data_spark_columns, False, None
+                data_dtypes = self._internal.data_dtypes
+                return column_labels, data_spark_columns, data_dtypes, False, None
             return self._select_cols_by_slice(cols_sel, missing_keys)
         elif isinstance(cols_sel, tuple):
             return self._select_cols_else(cols_sel, missing_keys)
@@ -355,35 +365,45 @@ def _select_rows_else(
     @abstractmethod
     def _select_cols_by_series(
         self, cols_sel: "Series", missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """ Select columns by `Series` type key. """
         pass
 
     @abstractmethod
     def _select_cols_by_spark_column(
         self, cols_sel: spark.Column, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """ Select columns by Spark `Column` type key. """
         pass
 
     @abstractmethod
     def _select_cols_by_slice(
         self, cols_sel: slice, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """ Select columns by `slice` type key. """
         pass
 
     @abstractmethod
     def _select_cols_by_iterable(
         self, cols_sel: Iterable, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """ Select columns by `Iterable` type key. """
         pass
 
     @abstractmethod
     def _select_cols_else(
         self, cols_sel: Any, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         """ Select columns by other type key. """
         pass
 
@@ -406,6 +426,7 @@ def __getitem__(self, key) -> Union["Series", "DataFrame"]:
             column_label = self._kdf_or_kser._column_label
             column_labels = [column_label]
             data_spark_columns = [self._internal.spark_column_for(column_label)]
+            data_dtypes = [self._internal.dtype_for(column_label)]
             returns_series = True
             series_name = self._kdf_or_kser.name
         else:
@@ -426,9 +447,13 @@ def __getitem__(self, key) -> Union["Series", "DataFrame"]:
                 return type(self)(kdf)[kdf[temp_col], cols_sel][list(self._kdf_or_kser.columns)]
 
             cond, limit, remaining_index = self._select_rows(rows_sel)
-            column_labels, data_spark_columns, returns_series, series_name = self._select_cols(
-                cols_sel
-            )
+            (
+                column_labels,
+                data_spark_columns,
+                data_dtypes,
+                returns_series,
+                series_name,
+            ) = self._select_cols(cols_sel)
 
             if cond is None and limit is None and returns_series:
                 kser = self._kdf_or_kser._kser_for(column_labels[0])
@@ -439,9 +464,11 @@ def __getitem__(self, key) -> Union["Series", "DataFrame"]:
         if remaining_index is not None:
             index_spark_columns = self._internal.index_spark_columns[-remaining_index:]
             index_names = self._internal.index_names[-remaining_index:]
+            index_dtypes = self._internal.index_dtypes[-remaining_index:]
         else:
             index_spark_columns = self._internal.index_spark_columns
             index_names = self._internal.index_names
+            index_dtypes = self._internal.index_dtypes
 
         if len(column_labels) > 0:
             column_labels = column_labels.copy()
@@ -491,8 +518,10 @@ def __getitem__(self, key) -> Union["Series", "DataFrame"]:
             spark_frame=sdf,
             index_spark_columns=index_spark_columns,
             index_names=index_names,
+            index_dtypes=index_dtypes,
             column_labels=column_labels,
             data_spark_columns=data_spark_columns,
+            data_dtypes=data_dtypes,
             column_label_names=column_label_names,
         )
         kdf = DataFrame(internal)
@@ -649,7 +678,7 @@ def __setitem__(self, key, value):
 
             cond, limit, remaining_index = self._select_rows(rows_sel)
             missing_keys = []
-            _, data_spark_columns, _, _ = self._select_cols(cols_sel, missing_keys=missing_keys)
+            _, data_spark_columns, _, _, _ = self._select_cols(cols_sel, missing_keys=missing_keys)
 
             if cond is None:
                 cond = F.lit(True)
@@ -667,14 +696,22 @@ def __setitem__(self, key, value):
                 value = F.lit(value)
 
             new_data_spark_columns = []
-            for new_scol, spark_column_name in zip(
-                self._internal.data_spark_columns, self._internal.data_spark_column_names
+            new_dtypes = []
+            for new_scol, spark_column_name, new_dtype in zip(
+                self._internal.data_spark_columns,
+                self._internal.data_spark_column_names,
+                self._internal.data_dtypes,
             ):
                 for scol in data_spark_columns:
                     if new_scol._jc.equals(scol._jc):
                         new_scol = F.when(cond, value).otherwise(scol).alias(spark_column_name)
+                        new_dtype = spark_type_to_pandas_dtype(
+                            self._internal.spark_frame.select(new_scol).schema[0].dataType,
+                            use_extension_dtypes=isinstance(new_dtype, extension_dtypes),
+                        )
                         break
                 new_data_spark_columns.append(new_scol)
+                new_dtypes.append(new_dtype)
 
             column_labels = self._internal.column_labels.copy()
             for label in missing_keys:
@@ -692,8 +729,11 @@ def __setitem__(self, key, value):
                     )
                 column_labels.append(label)
                 new_data_spark_columns.append(F.when(cond, value).alias(name_like_string(label)))
+                new_dtypes.append(None)
 
-            internal = self._internal.with_new_columns(new_data_spark_columns, column_labels)
+            internal = self._internal.with_new_columns(
+                new_data_spark_columns, column_labels=column_labels, data_dtypes=new_dtypes
+            )
             self._kdf_or_kser._update_internal_frame(internal, requires_same_anchor=False)
 
 
@@ -1065,7 +1105,7 @@ def _select_rows_else(
 
     def _get_from_multiindex_column(
         self, key, missing_keys, labels=None, recursed=0
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], Any, bool, Optional[Tuple]]:
         """ Select columns from multi-index columns. """
         assert isinstance(key, tuple)
         if labels is None:
@@ -1081,7 +1121,7 @@ def _get_from_multiindex_column(
                     raise KeyError(k)
                 else:
                     missing_keys.append(key)
-                    return [], [], False, None
+                    return [], [], [], False, None
 
         if all(lbl is not None and len(lbl) > 0 and lbl[0] == "" for _, lbl in labels):
             # If the head is '', drill down recursively.
@@ -1095,6 +1135,7 @@ def _get_from_multiindex_column(
                 label = list(labels)[0]
                 column_labels = [label]
                 data_spark_columns = [self._internal.spark_column_for(label)]
+                data_dtypes = [self._internal.dtype_for(label)]
                 if label is None:
                     series_name = None
                 else:
@@ -1106,47 +1147,60 @@ def _get_from_multiindex_column(
                     None if lbl is None or lbl == (None,) else lbl for _, lbl in labels
                 ]
                 data_spark_columns = [self._internal.spark_column_for(label) for label, _ in labels]
+                data_dtypes = [self._internal.dtype_for(label) for label, _ in labels]
                 series_name = None
 
-            return column_labels, data_spark_columns, returns_series, series_name
+            return column_labels, data_spark_columns, data_dtypes, returns_series, series_name
 
     def _select_cols_by_series(
         self, cols_sel: "Series", missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         column_labels = [cols_sel._column_label]
         data_spark_columns = [cols_sel.spark.column]
-        return column_labels, data_spark_columns, True, None
+        data_dtypes = [cols_sel.dtype]
+        return column_labels, data_spark_columns, data_dtypes, True, None
 
     def _select_cols_by_spark_column(
         self, cols_sel: spark.Column, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         column_labels = [
             (self._internal.spark_frame.select(cols_sel).columns[0],)
         ]  # type: List[Tuple]
         data_spark_columns = [cols_sel]
-        return column_labels, data_spark_columns, True, None
+        return column_labels, data_spark_columns, None, True, None
 
     def _select_cols_by_slice(
         self, cols_sel: slice, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         start, stop = self._kdf_or_kser.columns.slice_locs(start=cols_sel.start, end=cols_sel.stop)
         column_labels = self._internal.column_labels[start:stop]
         data_spark_columns = self._internal.data_spark_columns[start:stop]
-        return column_labels, data_spark_columns, False, None
+        data_dtypes = self._internal.data_dtypes[start:stop]
+        return column_labels, data_spark_columns, data_dtypes, False, None
 
     def _select_cols_by_iterable(
         self, cols_sel: Iterable, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         from databricks.koalas.series import Series
 
         if all(isinstance(key, Series) for key in cols_sel):
             column_labels = [key._column_label for key in cols_sel]
             data_spark_columns = [key.spark.column for key in cols_sel]
+            data_dtypes = [key.dtype for key in cols_sel]
         elif all(isinstance(key, spark.Column) for key in cols_sel):
             column_labels = [
                 (self._internal.spark_frame.select(col).columns[0],) for col in cols_sel
             ]
             data_spark_columns = list(cols_sel)
+            data_dtypes = None
         elif any(isinstance(key, tuple) for key in cols_sel) and any(
             not is_name_like_tuple(key) for key in cols_sel
         ):
@@ -1163,12 +1217,14 @@ def _select_cols_by_iterable(
 
             column_labels = []
             data_spark_columns = []
+            data_dtypes = []
             for key in cols_sel:
                 found = False
                 for label in self._internal.column_labels:
                     if label == key or label[0] == key:
                         column_labels.append(label)
                         data_spark_columns.append(self._internal.spark_column_for(label))
+                        data_dtypes.append(self._internal.dtype_for(label))
                         found = True
                 if not found:
                     if missing_keys is None:
@@ -1176,11 +1232,13 @@ def _select_cols_by_iterable(
                     else:
                         missing_keys.append(key)
 
-        return column_labels, data_spark_columns, False, None
+        return column_labels, data_spark_columns, data_dtypes, False, None
 
     def _select_cols_else(
         self, cols_sel: Any, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         if not is_name_like_tuple(cols_sel):
             cols_sel = (cols_sel,)
         return self._get_from_multiindex_column(cols_sel, missing_keys)
@@ -1498,7 +1556,9 @@ def _select_rows_else(
 
     def _select_cols_by_series(
         self, cols_sel: "Series", missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         raise ValueError(
             "Location based indexing can only have [integer, integer slice, "
             "listlike of integers, boolean array] types, got {}".format(cols_sel)
@@ -1506,7 +1566,9 @@ def _select_cols_by_series(
 
     def _select_cols_by_spark_column(
         self, cols_sel: spark.Column, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         raise ValueError(
             "Location based indexing can only have [integer, integer slice, "
             "listlike of integers, boolean array] types, got {}".format(cols_sel)
@@ -1514,12 +1576,16 @@ def _select_cols_by_spark_column(
 
     def _select_cols_by_slice(
         self, cols_sel: slice, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         if all(
             s is None or isinstance(s, int) for s in (cols_sel.start, cols_sel.stop, cols_sel.step)
         ):
             column_labels = self._internal.column_labels[cols_sel]
             data_spark_columns = self._internal.data_spark_columns[cols_sel]
+            data_dtypes = self._internal.data_dtypes[cols_sel]
+            return column_labels, data_spark_columns, data_dtypes, False, None
         else:
             not_none = (
                 cols_sel.start
@@ -1533,29 +1599,34 @@ def _select_cols_by_slice(
                     not_none, type(not_none)
                 )
             )
-        return column_labels, data_spark_columns, False, None
 
     def _select_cols_by_iterable(
         self, cols_sel: Iterable, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         if all(isinstance(s, bool) for s in cols_sel):
             cols_sel = [i for i, s in enumerate(cols_sel) if s]
         if all(isinstance(s, int) for s in cols_sel):
             column_labels = [self._internal.column_labels[s] for s in cols_sel]
             data_spark_columns = [self._internal.data_spark_columns[s] for s in cols_sel]
-            return column_labels, data_spark_columns, False, None
+            data_dtypes = [self._internal.data_dtypes[s] for s in cols_sel]
+            return column_labels, data_spark_columns, data_dtypes, False, None
         else:
             raise TypeError("cannot perform reduce with flexible type")
 
     def _select_cols_else(
         self, cols_sel: Any, missing_keys: Optional[List[Tuple]]
-    ) -> Tuple[List[Tuple], Optional[List[spark.Column]], bool, Optional[Tuple]]:
+    ) -> Tuple[
+        List[Tuple], Optional[List[spark.Column]], Optional[List[Dtype]], bool, Optional[Tuple]
+    ]:
         if isinstance(cols_sel, int):
             if cols_sel > len(self._internal.column_labels):
                 raise KeyError(cols_sel)
             column_labels = [self._internal.column_labels[cols_sel]]
             data_spark_columns = [self._internal.data_spark_columns[cols_sel]]
-            return column_labels, data_spark_columns, True, None
+            data_dtypes = [self._internal.data_dtypes[cols_sel]]
+            return column_labels, data_spark_columns, data_dtypes, True, None
         else:
             raise ValueError(
                 "Location based indexing can only have [integer, integer slice, "
diff --git a/databricks/koalas/internal.py b/databricks/koalas/internal.py
index 54e9631002..e4491a1b21 100644
--- a/databricks/koalas/internal.py
+++ b/databricks/koalas/internal.py
@@ -46,6 +46,9 @@
     from databricks.koalas.series import Series
 from databricks.koalas.config import get_option
 from databricks.koalas.typedef import (
+    Dtype,
+    as_spark_type,
+    extension_dtypes,
     infer_pd_series_spark_type,
     spark_type_to_pandas_dtype,
 )
@@ -122,10 +125,14 @@ class InternalFrame(object):
 
     * `data_spark_columns` represents non-indexing Spark columns
 
+    * `data_dtypes` represents external non-indexing dtypes
+
     * `index_spark_column_names` represents internal index Spark column names
 
     * `index_spark_columns` represents internal index Spark columns
 
+    * `index_dtypes` represents external index dtypes
+
     * `spark_column_names` represents all columns
 
     * `index_names` represents the external index name as a label
@@ -152,6 +159,10 @@ class InternalFrame(object):
     ['__index_level_0__', 'A', 'B', 'C', 'D', 'E']
     >>> internal.index_names
     [None]
+    >>> internal.data_dtypes
+    [dtype('int64'), dtype('int64'), dtype('int64'), dtype('int64'), dtype('int64')]
+    >>> internal.index_dtypes
+    [dtype('int64')]
     >>> internal.to_internal_spark_frame.show()  # doctest: +NORMALIZE_WHITESPACE
     +-----------------+---+---+---+---+---+
     |__index_level_0__|  A|  B|  C|  D|  E|
@@ -207,6 +218,10 @@ class InternalFrame(object):
     ['A', 'B', 'C', 'D', 'E']
     >>> internal.index_names
     [('A',)]
+    >>> internal.data_dtypes
+    [dtype('int64'), dtype('int64'), dtype('int64'), dtype('int64')]
+    >>> internal.index_dtypes
+    [dtype('int64')]
     >>> internal.to_internal_spark_frame.show()  # doctest: +NORMALIZE_WHITESPACE
     +---+---+---+---+---+
     |  A|  B|  C|  D|  E|
@@ -263,6 +278,10 @@ class InternalFrame(object):
     ['__index_level_0__', 'A', 'B', 'C', 'D', 'E']
     >>> internal.index_names
     [None, ('A',)]
+    >>> internal.data_dtypes
+    [dtype('int64'), dtype('int64'), dtype('int64'), dtype('int64')]
+    >>> internal.index_dtypes
+    [dtype('int64'), dtype('int64')]
     >>> internal.to_internal_spark_frame.show()  # doctest: +NORMALIZE_WHITESPACE
     +-----------------+---+---+---+---+---+
     |__index_level_0__|  A|  B|  C|  D|  E|
@@ -344,6 +363,10 @@ class InternalFrame(object):
     ['A', 'B']
     >>> internal.index_names
     [('A',)]
+    >>> internal.data_dtypes
+    [dtype('int64')]
+    >>> internal.index_dtypes
+    [dtype('int64')]
     >>> internal.to_internal_spark_frame.show()  # doctest: +NORMALIZE_WHITESPACE
     +---+---+
     |  A|  B|
@@ -367,8 +390,10 @@ def __init__(
         spark_frame: spark.DataFrame,
         index_spark_columns: Optional[List[spark.Column]],
         index_names: Optional[List[Optional[Tuple]]] = None,
+        index_dtypes: Optional[List[Dtype]] = None,
         column_labels: Optional[List[Tuple]] = None,
         data_spark_columns: Optional[List[spark.Column]] = None,
+        data_dtypes: Optional[List[Dtype]] = None,
         column_label_names: Optional[List[Optional[Tuple]]] = None,
     ) -> None:
         """
@@ -380,11 +405,15 @@ def __init__(
                                     Spark Columns for the index.
         :param index_names: list of tuples
                             the index names.
+        :param index_dtypes: list of dtypes
+                             the index dtypes.
         :param column_labels: list of tuples with the same length
                               The multi-level values in the tuples.
         :param data_spark_columns: list of Spark Column
                                    Spark Columns to appear as columns. If this is None, calculated
                                    from spark_frame.
+        :param data_dtypes: list of dtypes.
+                            the data dtypes.
         :param column_label_names: Names for each of the column index levels.
 
         See the examples below to refer what each parameter means.
@@ -407,7 +436,7 @@ def __init__(
 
         >>> internal = kdf._internal
 
-        >>> internal._sdf.show()  # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
+        >>> internal.spark_frame.show()  # doctest: +NORMALIZE_WHITESPACE +ELLIPSIS
         +-----------------+-----------------+------+------+------+...
         |__index_level_0__|__index_level_1__|(a, x)|(a, y)|(b, z)|...
         +-----------------+-----------------+------+------+------+...
@@ -416,19 +445,25 @@ def __init__(
         |              zoo|              bar|     7|     8|     9|...
         +-----------------+-----------------+------+------+------+...
 
-        >>> internal._index_spark_columns
+        >>> internal.index_spark_columns
         [Column<b'__index_level_0__'>, Column<b'__index_level_1__'>, Column<b'(a, x)'>]
 
-        >>> internal._index_names
+        >>> internal.index_names
         [('row_index_a',), ('row_index_b',), ('a', 'x')]
 
-        >>> internal._column_labels
+        >>> internal.index_dtypes
+        [dtype('O'), dtype('O'), dtype('int64')]
+
+        >>> internal.column_labels
         [('a', 'y'), ('b', 'z')]
 
-        >>> internal._data_spark_columns
+        >>> internal.data_spark_columns
         [Column<b'(a, y)'>, Column<b'(b, z)'>]
 
-        >>> internal._column_label_names
+        >>> internal.data_dtypes
+        [dtype('int64'), dtype('int64')]
+
+        >>> internal.column_label_names
         [('column_labels_a',), ('column_labels_b',)]
         """
 
@@ -465,6 +500,16 @@ def __init__(
                 NATURAL_ORDER_COLUMN_NAME, F.monotonically_increasing_id()
             )
 
+        self._sdf = spark_frame  # type: spark.DataFrame
+
+        # index_spark_columns
+        assert all(
+            isinstance(index_scol, spark.Column) for index_scol in index_spark_columns
+        ), index_spark_columns
+
+        self._index_spark_columns = index_spark_columns  # type: List[spark.Column]
+
+        # index_names
         if not index_names:
             index_names = [None] * len(index_spark_columns)
 
@@ -472,22 +517,37 @@ def __init__(
             len(index_spark_columns),
             len(index_names),
         )
-        assert all(
-            isinstance(index_scol, spark.Column) for index_scol in index_spark_columns
-        ), index_spark_columns
         assert all(
             is_name_like_tuple(index_name, check_type=True) for index_name in index_names
         ), index_names
-        assert data_spark_columns is None or all(
-            isinstance(scol, spark.Column) for scol in data_spark_columns
-        )
 
-        self._sdf = spark_frame  # type: spark.DataFrame
-        self._index_spark_columns = index_spark_columns  # type: List[spark.Column]
         self._index_names = index_names  # type: List[Optional[Tuple]]
 
+        # index_dtypes
+        if not index_dtypes:
+            index_dtypes = [None] * len(index_spark_columns)
+
+        assert len(index_spark_columns) == len(index_dtypes), (
+            len(index_spark_columns),
+            len(index_dtypes),
+        )
+
+        index_dtypes = [
+            spark_type_to_pandas_dtype(spark_frame.select(scol).schema[0].dataType)
+            if dtype is None
+            else dtype
+            for dtype, scol in zip(index_dtypes, index_spark_columns)
+        ]
+
+        assert all(
+            as_spark_type(dtype, raise_error=False) is not None for dtype in index_dtypes
+        ), index_dtypes
+
+        self._index_dtypes = index_dtypes
+
+        # data_spark-columns
         if data_spark_columns is None:
-            self._data_spark_columns = [
+            data_spark_columns = [
                 scol_for(spark_frame, col)
                 for col in spark_frame.columns
                 if all(
@@ -496,9 +556,12 @@ def __init__(
                 )
                 and col not in HIDDEN_COLUMNS
             ]
+            self._data_spark_columns = data_spark_columns
         else:
+            assert all(isinstance(scol, spark.Column) for scol in data_spark_columns)
             self._data_spark_columns = data_spark_columns
 
+        # column_labels
         if column_labels is None:
             self._column_labels = [
                 (col,) for col in spark_frame.select(self._data_spark_columns).columns
@@ -519,6 +582,29 @@ def __init__(
                 assert len(set(len(label) for label in column_labels)) <= 1, column_labels
             self._column_labels = column_labels
 
+        # data_dtypes
+        if not data_dtypes:
+            data_dtypes = [None] * len(data_spark_columns)
+
+        assert len(data_spark_columns) == len(data_dtypes), (
+            len(data_spark_columns),
+            len(data_dtypes),
+        )
+
+        data_dtypes = [
+            spark_type_to_pandas_dtype(spark_frame.select(scol).schema[0].dataType)
+            if dtype is None
+            else dtype
+            for dtype, scol in zip(data_dtypes, data_spark_columns)
+        ]
+
+        assert all(
+            as_spark_type(dtype, raise_error=False) is not None for dtype in data_dtypes
+        ), data_dtypes
+
+        self._data_dtypes = data_dtypes
+
+        # column_label_names
         if column_label_names is None:
             self._column_label_names = [None] * column_labels_level(
                 self._column_labels
@@ -712,7 +798,7 @@ def spark_column_for(self, label: Tuple) -> spark.Column:
         """ Return Spark Column for the given column label. """
         column_labels_to_scol = dict(zip(self.column_labels, self.data_spark_columns))
         if label in column_labels_to_scol:
-            return column_labels_to_scol[label]  # type: ignore
+            return column_labels_to_scol[label]
         else:
             raise KeyError(name_like_string(label))
 
@@ -740,6 +826,14 @@ def spark_column_nullable_for(self, label_or_scol: Union[Tuple, spark.Column]) -
             scol = self.spark_column_for(label_or_scol)
         return self.spark_frame.select(scol).schema[0].nullable
 
+    def dtype_for(self, label: Tuple) -> Dtype:
+        """ Return dtype for the given column label. """
+        column_labels_to_dtype = dict(zip(self.column_labels, self.data_dtypes))
+        if label in column_labels_to_dtype:
+            return column_labels_to_dtype[label]
+        else:
+            raise KeyError(name_like_string(label))
+
     @property
     def spark_frame(self) -> spark.DataFrame:
         """ Return the managed Spark DataFrame. """
@@ -805,6 +899,16 @@ def column_label_names(self) -> List[Optional[Tuple]]:
         """ Return names of the index levels. """
         return self._column_label_names
 
+    @property
+    def index_dtypes(self) -> List[Dtype]:
+        """ Return dtypes for the managed index columns. """
+        return self._index_dtypes
+
+    @property
+    def data_dtypes(self) -> List[Dtype]:
+        """ Return dtypes for the managed columns. """
+        return self._data_dtypes
+
     @lazy_property
     def to_internal_spark_frame(self) -> spark.DataFrame:
         """
@@ -845,8 +949,18 @@ def to_pandas_frame(self) -> pd.DataFrame:
                         )
 
         column_names = []
-        for i, (label, spark_column, column_name) in enumerate(
-            zip(self.column_labels, self.data_spark_columns, self.data_spark_column_names)
+        ext_dtypes = {
+            col: dtype
+            for col, dtype in zip(self.index_spark_column_names, self.index_dtypes)
+            if isinstance(dtype, extension_dtypes)
+        }
+        for i, (label, spark_column, column_name, dtype) in enumerate(
+            zip(
+                self.column_labels,
+                self.data_spark_columns,
+                self.data_spark_column_names,
+                self.data_dtypes,
+            )
         ):
             for index_spark_column_name, index_spark_column in zip(
                 self.index_spark_column_names, self.index_spark_columns
@@ -859,6 +973,11 @@ def to_pandas_frame(self) -> pd.DataFrame:
                 if column_name != name:
                     column_name = name
                 column_names.append(column_name)
+                if isinstance(dtype, extension_dtypes):
+                    ext_dtypes[column_name] = dtype
+
+        if len(ext_dtypes) > 0:
+            pdf = pdf.astype(ext_dtypes, copy=True)
 
         append = False
         for index_field in self.index_spark_column_names:
@@ -868,13 +987,13 @@ def to_pandas_frame(self) -> pd.DataFrame:
         pdf = pdf[column_names]
 
         names = [
-            name if name is None or len(name) > 1 else name[0] for name in self._column_label_names
+            name if name is None or len(name) > 1 else name[0] for name in self.column_label_names
         ]
         if self.column_labels_level > 1:
-            pdf.columns = pd.MultiIndex.from_tuples(self._column_labels, names=names)
+            pdf.columns = pd.MultiIndex.from_tuples(self.column_labels, names=names)
         else:
             pdf.columns = pd.Index(
-                [None if label is None else label[0] for label in self._column_labels],
+                [None if label is None else label[0] for label in self.column_labels],
                 name=names[0],
             )
 
@@ -892,16 +1011,24 @@ def resolved_copy(self) -> "InternalFrame":
             spark_frame=sdf,
             index_spark_columns=[scol_for(sdf, col) for col in self.index_spark_column_names],
             data_spark_columns=[scol_for(sdf, col) for col in self.data_spark_column_names],
+            preserve_dtypes=True,
         )
 
     def with_new_sdf(
-        self, spark_frame: spark.DataFrame, data_columns: Optional[List[str]] = None
+        self,
+        spark_frame: spark.DataFrame,
+        data_columns: Optional[List[str]] = None,
+        *,
+        preserve_dtypes: bool = False
     ) -> "InternalFrame":
         """ Copy the immutable InternalFrame with the updates by the specified Spark DataFrame.
 
         :param spark_frame: the new Spark DataFrame
         :param data_columns: the new column names.
             If None, the original one is used.
+        :param preserve_dtypes: whether the dtypes can be preserved.
+            If True, the original dtypes are used; otherwise the dtypes will be inferred from
+            the new Spark DataFrame.
         :return: the copied InternalFrame.
         """
         if data_columns is None:
@@ -916,22 +1043,28 @@ def with_new_sdf(
             spark_frame=sdf,
             index_spark_columns=[scol_for(sdf, col) for col in self.index_spark_column_names],
             data_spark_columns=[scol_for(sdf, col) for col in data_columns],
+            preserve_dtypes=preserve_dtypes,
         )
 
     def with_new_columns(
         self,
         scols_or_ksers: List[Union[spark.Column, "Series"]],
+        *,
         column_labels: Optional[List[Tuple]] = None,
+        data_dtypes: Optional[List[Dtype]] = None,
         column_label_names: Optional[Union[List[Optional[Tuple]], _NoValueType]] = _NoValue,
-        keep_order: bool = True,
+        keep_order: bool = True
     ) -> "InternalFrame":
         """
         Copy the immutable InternalFrame with the updates by the specified Spark Columns or Series.
 
         :param scols_or_ksers: the new Spark Columns or Series.
         :param column_labels: the new column index.
-            If None, the its column_labels is used when the corresponding `scols_or_ksers` is
-            Series, otherwise the original one is used.
+            If None, the column_labels of the corresponding `scols_or_ksers` is used if it is
+            Series; otherwise the original one is used.
+        :param data_dtypes: the new dtypes.
+            If None, the dtypes of the corresponding `scols_or_ksers` is used if it is Series;
+            otherwise the dtypes will be inferred from the corresponding `scols_or_ksers`.
         :param column_label_names: the new names of the column index levels.
         :return: the copied InternalFrame.
         """
@@ -965,6 +1098,19 @@ def with_new_columns(
                 scol = scol_or_kser
             data_spark_columns.append(scol)
 
+        if data_dtypes is None:
+            data_dtypes = []
+            for scol_or_kser in scols_or_ksers:
+                if isinstance(scol_or_kser, Series):
+                    data_dtypes.append(scol_or_kser.dtype)
+                else:
+                    data_dtypes.append(None)
+        else:
+            assert len(scols_or_ksers) == len(data_dtypes), (
+                len(scols_or_ksers),
+                len(data_dtypes),
+            )
+
         sdf = self.spark_frame
         if not keep_order:
             sdf = self.spark_frame.select(self.index_spark_columns + data_spark_columns)
@@ -983,7 +1129,9 @@ def with_new_columns(
             index_spark_columns=index_spark_columns,
             column_labels=column_labels,
             data_spark_columns=data_spark_columns,
+            data_dtypes=data_dtypes,
             column_label_names=column_label_names,
+            preserve_dtypes=True,
         )
 
     def with_filter(self, pred: Union[spark.Column, "Series"]) -> "InternalFrame":
@@ -1001,10 +1149,12 @@ def with_filter(self, pred: Union[spark.Column, "Series"]) -> "InternalFrame":
             spark_type = self.spark_frame.select(pred).schema[0].dataType
             assert isinstance(spark_type, BooleanType), spark_type
 
-        return self.with_new_sdf(self.spark_frame.filter(pred).select(self.spark_columns))
+        return self.with_new_sdf(
+            self.spark_frame.filter(pred).select(self.spark_columns), preserve_dtypes=True
+        )
 
     def with_new_spark_column(
-        self, column_label: Tuple, scol: spark.Column, keep_order: bool = True
+        self, column_label: Tuple, scol: spark.Column, *, keep_order: bool = True
     ) -> "InternalFrame":
         """
         Copy the immutable InternalFrame with the updates by the specified Spark Column.
@@ -1032,6 +1182,7 @@ def select_column(self, column_label: Tuple) -> "InternalFrame":
         return self.copy(
             column_labels=[column_label],
             data_spark_columns=[self.spark_column_for(column_label)],
+            data_dtypes=[self.dtype_for(column_label)],
             column_label_names=None,
         )
 
@@ -1040,9 +1191,13 @@ def copy(
         spark_frame: Union[spark.DataFrame, _NoValueType] = _NoValue,
         index_spark_columns: Union[List[spark.Column], _NoValueType] = _NoValue,
         index_names: Union[List[Optional[Tuple]], _NoValueType] = _NoValue,
+        index_dtypes: Optional[Union[List[Dtype], _NoValueType]] = _NoValue,
         column_labels: Optional[Union[List[Tuple], _NoValueType]] = _NoValue,
         data_spark_columns: Optional[Union[List[spark.Column], _NoValueType]] = _NoValue,
+        data_dtypes: Optional[Union[List[Dtype], _NoValueType]] = _NoValue,
         column_label_names: Optional[Union[List[Optional[Tuple]], _NoValueType]] = _NoValue,
+        *,
+        preserve_dtypes: bool = False  # TODO: remove eventually.
     ) -> "InternalFrame":
         """ Copy the immutable InternalFrame.
 
@@ -1050,11 +1205,20 @@ def copy(
         :param index_spark_columns: the list of Spark Column.
                                     If not specified, the original ones are used.
         :param index_names: the index names. If not specified, the original ones are used.
+        :param index_dtypes: the index dtypes. If not specified, the original dtyeps are used
+                             if preserve_dtypes is True; otherwise the dtypes will be inferred
+                             from the corresponding `index_spark_columns`.
         :param column_labels: the new column labels. If not specified, the original ones are used.
         :param data_spark_columns: the new Spark Columns.
                                    If not specified, the original ones are used.
+        :param data_dtypes: the data dtypes. If not specified, the original dtyeps are used
+                             if preserve_dtypes is True; otherwise the dtypes will be inferred
+                             from the corresponding `data_spark_columns`.
         :param column_label_names: the new names of the column index levels.
                                    If not specified, the original ones are used.
+        :param preserve_dtypes: whether the dtypes can be preserved.
+                                If True, the original dtypes are used; otherwise the dtypes will be
+                                inferred from the new Spark columns.
         :return: the copied immutable InternalFrame.
         """
         if spark_frame is _NoValue:
@@ -1063,18 +1227,24 @@ def copy(
             index_spark_columns = self.index_spark_columns
         if index_names is _NoValue:
             index_names = self.index_names
+        if index_dtypes is _NoValue:
+            index_dtypes = self.index_dtypes if preserve_dtypes else None
         if column_labels is _NoValue:
             column_labels = self.column_labels
         if data_spark_columns is _NoValue:
             data_spark_columns = self.data_spark_columns
+        if data_dtypes is _NoValue:
+            data_dtypes = self.data_dtypes if preserve_dtypes else None
         if column_label_names is _NoValue:
             column_label_names = self.column_label_names
         return InternalFrame(
             spark_frame=spark_frame,
             index_spark_columns=index_spark_columns,
             index_names=index_names,
+            index_dtypes=index_dtypes,
             column_labels=column_labels,
             data_spark_columns=data_spark_columns,
+            data_dtypes=data_dtypes,
             column_label_names=column_label_names,
         )
 
@@ -1085,6 +1255,8 @@ def from_pandas(pdf: pd.DataFrame) -> "InternalFrame":
         :param pdf: :class:`pd.DataFrame`
         :return: the created immutable DataFrame
         """
+        pdf = pdf.copy()
+
         columns = pdf.columns
         data_columns = [name_like_string(col) for col in columns]
         if isinstance(columns, pd.MultiIndex):
@@ -1095,12 +1267,15 @@ def from_pandas(pdf: pd.DataFrame) -> "InternalFrame":
             name if name is None or isinstance(name, tuple) else (name,) for name in columns.names
         ]
 
+        index = pdf.index
         index_names = [
-            name if name is None or isinstance(name, tuple) else (name,) for name in pdf.index.names
+            name if name is None or isinstance(name, tuple) else (name,) for name in index.names
         ]
         index_columns = [SPARK_INDEX_NAME_FORMAT(i) for i in range(len(index_names))]
 
-        pdf = pdf.copy()
+        index_dtypes = [index.get_level_values(i).dtype for i in range(len(index_names))]
+        data_dtypes = list(pdf.dtypes)
+
         pdf.index.names = index_columns
         reset_index = pdf.reset_index()
         reset_index.columns = index_columns + data_columns
@@ -1122,7 +1297,9 @@ def from_pandas(pdf: pd.DataFrame) -> "InternalFrame":
             spark_frame=sdf,
             index_spark_columns=[scol_for(sdf, col) for col in index_columns],
             index_names=index_names,
+            index_dtypes=index_dtypes,
             column_labels=column_labels,
             data_spark_columns=[scol_for(sdf, col) for col in data_columns],
+            data_dtypes=data_dtypes,
             column_label_names=column_label_names,
         )
diff --git a/databricks/koalas/namespace.py b/databricks/koalas/namespace.py
index c28173ee01..79d220ff31 100644
--- a/databricks/koalas/namespace.py
+++ b/databricks/koalas/namespace.py
@@ -2016,7 +2016,9 @@ def concat(objs, axis=0, join="outer", ignore_index=False, sort=False) -> Union[
                 concat_kdf = DataFrame(
                     concat_kdf._internal.with_new_columns(
                         concat_kdf._internal.data_spark_columns + kdf._internal.data_spark_columns,
-                        concat_kdf._internal.column_labels + kdf._internal.column_labels,
+                        column_labels=(
+                            concat_kdf._internal.column_labels + kdf._internal.column_labels
+                        ),
                     )
                 )
             else:
diff --git a/databricks/koalas/series.py b/databricks/koalas/series.py
index 93b818d2f1..a121d5be29 100644
--- a/databricks/koalas/series.py
+++ b/databricks/koalas/series.py
@@ -372,7 +372,7 @@ def __init__(self, data=None, index=None, dtype=None, name=None, copy=False, fas
                 )
             internal = InternalFrame.from_pandas(pd.DataFrame(s))
             if s.name is None:
-                internal = internal.copy(column_labels=[None])
+                internal = internal.copy(column_labels=[None], preserve_dtypes=True)
             anchor = DataFrame(internal)
 
             self._anchor = anchor
@@ -399,7 +399,7 @@ def _update_anchor(self, kdf: DataFrame):
         self._anchor = kdf
         object.__setattr__(kdf, "_kseries", {self._column_label: self})
 
-    def _with_new_scol(self, scol: spark.Column) -> "Series":
+    def _with_new_scol(self, scol: spark.Column, *, dtype=None) -> "Series":
         """
         Copy Koalas Series with the new Spark Column.
 
@@ -407,7 +407,8 @@ def _with_new_scol(self, scol: spark.Column) -> "Series":
         :return: the copied Series
         """
         internal = self._internal.copy(
-            data_spark_columns=[scol.alias(name_like_string(self._column_label))]
+            data_spark_columns=[scol.alias(name_like_string(self._column_label))],
+            data_dtypes=[dtype],
         )
         return first_series(DataFrame(internal))
 
@@ -1057,8 +1058,11 @@ def rename(self, index=None, **kwargs) -> "Series":
             index = (index,)
         scol = self.spark.column.alias(name_like_string(index))
 
-        internal = self._kdf._internal.copy(
-            column_labels=[index], data_spark_columns=[scol], column_label_names=None
+        internal = self._internal.copy(
+            column_labels=[index],
+            data_spark_columns=[scol],
+            column_label_names=None,
+            preserve_dtypes=True,
         )
         kdf = DataFrame(internal)  # type: DataFrame
 
@@ -4918,7 +4922,7 @@ def dot(self, other: Union["Series", DataFrame]) -> Union[Scalar, "Series"]:
             product_ksers = [other._kser_for(label) * self_kser for label in column_labels]
 
             dot_product_kser = DataFrame(
-                other._internal.with_new_columns(product_ksers, column_labels)
+                other._internal.with_new_columns(product_ksers, column_labels=column_labels)
             ).sum()
 
             return cast(Series, dot_product_kser).rename(self.name)
diff --git a/databricks/koalas/strings.py b/databricks/koalas/strings.py
index 98aa2e1d93..790d13420f 100644
--- a/databricks/koalas/strings.py
+++ b/databricks/koalas/strings.py
@@ -2000,7 +2000,7 @@ def split(self, pat=None, n=-1, expand=False) -> Union["ks.Series", "ks.DataFram
             spark_columns = [scol[i].alias(str(i)) for i in range(n + 1)]
             column_labels = [(i,) for i in range(n + 1)]
             internal = kdf._internal.with_new_columns(
-                spark_columns, cast(Optional[List], column_labels)
+                spark_columns, column_labels=cast(Optional[List], column_labels)
             )
             return DataFrame(internal)
         else:
@@ -2136,7 +2136,7 @@ def rsplit(self, pat=None, n=-1, expand=False) -> Union["ks.Series", "ks.DataFra
             spark_columns = [scol[i].alias(str(i)) for i in range(n + 1)]
             column_labels = [(i,) for i in range(n + 1)]
             internal = kdf._internal.with_new_columns(
-                spark_columns, cast(Optional[List], column_labels)
+                spark_columns, column_labels=cast(Optional[List], column_labels)
             )
             return DataFrame(internal)
         else:
diff --git a/databricks/koalas/typedef/typehints.py b/databricks/koalas/typedef/typehints.py
index 9019df3940..6da10850b4 100644
--- a/databricks/koalas/typedef/typehints.py
+++ b/databricks/koalas/typedef/typehints.py
@@ -24,6 +24,36 @@
 
 import numpy as np
 import pandas as pd
+from pandas.api.extensions import ExtensionDtype
+
+try:
+    from pandas import Int8Dtype, Int16Dtype, Int32Dtype, Int64Dtype
+
+    extension_dtypes_available = True
+    extension_dtypes = (Int8Dtype, Int16Dtype, Int32Dtype, Int64Dtype)  # type: typing.Tuple
+
+    try:
+        from pandas import BooleanDtype, StringDtype
+
+        extension_object_dtypes_available = True
+        extension_dtypes += (BooleanDtype, StringDtype)
+    except ImportError:
+        extension_object_dtypes_available = False
+
+    try:
+        from pandas import Float32Dtype, Float64Dtype
+
+        extension_float_dtypes_available = True
+        extension_dtypes += (Float32Dtype, Float64Dtype)
+    except ImportError:
+        extension_float_dtypes_available = False
+
+except ImportError:
+    extension_dtypes_available = False
+    extension_object_dtypes_available = False
+    extension_float_dtypes_available = False
+    extension_dtypes = ()
+
 import pyarrow as pa
 import pyspark.sql.types as types
 
@@ -41,6 +71,8 @@
     int, float, bool, str, bytes, decimal.Decimal, datetime.date, datetime.datetime, None
 ]
 
+Dtype = typing.Union[np.dtype, ExtensionDtype]
+
 
 # A column of data, with the data type.
 class SeriesType(typing.Generic[T]):
@@ -90,7 +122,9 @@ class NameTypeHolder(object):
     tpe = None
 
 
-def as_spark_type(tpe) -> types.DataType:
+def as_spark_type(
+    tpe: typing.Union[str, type, Dtype], *, raise_error: bool = True
+) -> types.DataType:
     """
     Given a Python type, returns the equivalent spark type.
     Accepts:
@@ -104,8 +138,11 @@ def as_spark_type(tpe) -> types.DataType:
     # ArrayType
     if tpe in (np.ndarray,):
         return types.ArrayType(types.StringType())
-    elif hasattr(tpe, "__origin__") and issubclass(tpe.__origin__, list):
-        return types.ArrayType(as_spark_type(tpe.__args__[0]))
+    elif hasattr(tpe, "__origin__") and issubclass(tpe.__origin__, list):  # type: ignore
+        element_type = as_spark_type(tpe.__args__[0], raise_error=raise_error)  # type: ignore
+        if element_type is None:
+            return None
+        return types.ArrayType(element_type)
     # BinaryType
     elif tpe in (bytes, np.character, np.bytes_, np.string_):
         return types.BinaryType()
@@ -137,12 +174,71 @@ def as_spark_type(tpe) -> types.DataType:
     # TimestampType
     elif tpe in (datetime.datetime, np.datetime64, "datetime64[ns]", "M"):
         return types.TimestampType()
-    else:
+
+    # extension types
+    elif extension_dtypes_available:
+        # IntegralType
+        if isinstance(tpe, Int8Dtype) or (isinstance(tpe, str) and tpe == "Int8"):
+            return types.ByteType()
+        elif isinstance(tpe, Int16Dtype) or (isinstance(tpe, str) and tpe == "Int16"):
+            return types.ShortType()
+        elif isinstance(tpe, Int32Dtype) or (isinstance(tpe, str) and tpe == "Int32"):
+            return types.IntegerType()
+        elif isinstance(tpe, Int64Dtype) or (isinstance(tpe, str) and tpe == "Int64"):
+            return types.LongType()
+
+        if extension_object_dtypes_available:
+            # BooleanType
+            if isinstance(tpe, BooleanDtype) or (isinstance(tpe, str) and tpe == "boolean"):
+                return types.BooleanType()
+            # StringType
+            elif isinstance(tpe, StringDtype) or (isinstance(tpe, str) and tpe == "string"):
+                return types.StringType()
+
+        if extension_float_dtypes_available:
+            # FractionalType
+            if isinstance(tpe, Float32Dtype) or (isinstance(tpe, str) and tpe == "Float32"):
+                return types.FloatType()
+            elif isinstance(tpe, Float64Dtype) or (isinstance(tpe, str) and tpe == "Float64"):
+                return types.DoubleType()
+
+    if raise_error:
         raise TypeError("Type %s was not understood." % tpe)
+    else:
+        return None
 
 
-def spark_type_to_pandas_dtype(spark_type):
+def spark_type_to_pandas_dtype(
+    spark_type: types.DataType, *, use_extension_dtypes: bool = False
+) -> Dtype:
     """ Return the given Spark DataType to pandas dtype. """
+
+    if use_extension_dtypes and extension_dtypes_available:
+        # IntegralType
+        if isinstance(spark_type, types.ByteType):
+            return Int8Dtype()
+        elif isinstance(spark_type, types.ShortType):
+            return Int16Dtype()
+        elif isinstance(spark_type, types.IntegerType):
+            return Int32Dtype()
+        elif isinstance(spark_type, types.LongType):
+            return Int64Dtype()
+
+        if extension_object_dtypes_available:
+            # BooleanType
+            if isinstance(spark_type, types.BooleanType):
+                return BooleanDtype()
+            # StringType
+            elif isinstance(spark_type, types.StringType):
+                return StringDtype()
+
+        # FractionalType
+        if extension_float_dtypes_available:
+            if isinstance(spark_type, types.FloatType):
+                return Float32Dtype()
+            elif isinstance(spark_type, types.DoubleType):
+                return Float64Dtype()
+
     if isinstance(
         spark_type, (types.DateType, types.NullType, types.StructType, types.UserDefinedType)
     ):
diff --git a/databricks/koalas/utils.py b/databricks/koalas/utils.py
index 293b3c7a60..a9d14d91c0 100644
--- a/databricks/koalas/utils.py
+++ b/databricks/koalas/utils.py
@@ -28,13 +28,17 @@
 import pyspark
 from pyspark import sql as spark
 from pyspark.sql import functions as F
-from pyspark.sql.types import FloatType
+from pyspark.sql.types import DoubleType
 import pandas as pd
 from pandas.api.types import is_list_like
 
 # For running doctests and reference resolution in PyCharm.
 from databricks import koalas as ks  # noqa: F401
-from databricks.koalas.typedef.typehints import as_spark_type
+from databricks.koalas.typedef.typehints import (
+    as_spark_type,
+    extension_dtypes,
+    spark_type_to_pandas_dtype,
+)
 
 if TYPE_CHECKING:
     # This is required in old Python 3.5 to prevent circular reference.
@@ -143,16 +147,25 @@ def resolve(internal, side):
                 data_spark_columns=[
                     scol_for(sdf, rename(col)) for col in internal.data_spark_column_names
                 ],
+                preserve_dtypes=True,
             )
 
         this_internal = resolve(this._internal, "this")
         that_internal = resolve(that._internal, "that")
 
         this_index_map = list(
-            zip(this_internal.index_spark_column_names, this_internal.index_names)
+            zip(
+                this_internal.index_spark_column_names,
+                this_internal.index_names,
+                this_internal.index_dtypes,
+            )
         )
         that_index_map = list(
-            zip(that_internal.index_spark_column_names, that_internal.index_names)
+            zip(
+                that_internal.index_spark_column_names,
+                that_internal.index_names,
+                that_internal.index_dtypes,
+            )
         )
         assert len(this_index_map) == len(that_index_map)
 
@@ -168,9 +181,11 @@ def resolve(internal, side):
 
         # If the same named index is found, that's used.
         index_column_names = []
-        for i, ((this_column, this_name), (that_column, that_name)) in enumerate(
-            this_and_that_index_map
-        ):
+        index_use_extension_dtypes = []
+        for (
+            i,
+            ((this_column, this_name, this_dtype), (that_column, that_name, that_dtype)),
+        ) in enumerate(this_and_that_index_map):
             if this_name == that_name:
                 # We should merge the Spark columns into one
                 # to mimic pandas' behavior.
@@ -181,6 +196,9 @@ def resolve(internal, side):
 
                 column_name = SPARK_INDEX_NAME_FORMAT(i)
                 index_column_names.append(column_name)
+                index_use_extension_dtypes.append(
+                    any(isinstance(dtype, extension_dtypes) for dtype in [this_dtype, that_dtype])
+                )
                 merged_index_scols.append(
                     F.when(this_scol.isNotNull(), this_scol).otherwise(that_scol).alias(column_name)
                 )
@@ -209,12 +227,22 @@ def resolve(internal, side):
             + order_column
         )
 
+        index_spark_columns = [scol_for(joined_df, col) for col in index_column_names]
+        index_dtypes = [
+            spark_type_to_pandas_dtype(field.dataType, use_extension_dtypes=use_extension_dtypes)
+            for field, use_extension_dtypes in zip(
+                joined_df.select(index_spark_columns).schema, index_use_extension_dtypes
+            )
+        ]
+
         index_columns = set(index_column_names)
         new_data_columns = [
             col
             for col in joined_df.columns
             if col not in index_columns and col != NATURAL_ORDER_COLUMN_NAME
         ]
+        data_dtypes = this_internal.data_dtypes + that_internal.data_dtypes
+
         level = max(this_internal.column_labels_level, that_internal.column_labels_level)
 
         def fill_label(label):
@@ -232,10 +260,12 @@ def fill_label(label):
         return DataFrame(
             InternalFrame(
                 spark_frame=joined_df,
-                index_spark_columns=[scol_for(joined_df, col) for col in index_column_names],
+                index_spark_columns=index_spark_columns,
                 index_names=this_internal.index_names,
+                index_dtypes=index_dtypes,
                 column_labels=column_labels,
                 data_spark_columns=[scol_for(joined_df, col) for col in new_data_columns],
+                data_dtypes=data_dtypes,
                 column_label_names=column_label_names,
             )
         )
@@ -305,6 +335,8 @@ def align_diff_frames(
         - inner: Same as 'full' mode; however, internally performs inner join instead.
     :return: Aligned DataFrame
     """
+    from databricks.koalas.frame import DataFrame
+
     assert how == "full" or how == "left" or how == "inner"
 
     this_column_labels = this._internal.column_labels
@@ -341,10 +373,10 @@ def align_diff_frames(
                 # is intentional so that `this_columns` and `that_columns` can be paired.
                 additional_that_columns.append(combined_label)
             elif fillna:
-                columns_to_keep.append(F.lit(None).cast(FloatType()).alias(str(combined_label)))
+                columns_to_keep.append(F.lit(None).cast(DoubleType()).alias(str(combined_label)))
                 column_labels_to_keep.append(combined_label)
             else:
-                columns_to_keep.append(combined._internal.spark_column_for(combined_label))
+                columns_to_keep.append(combined._kser_for(combined_label))
                 column_labels_to_keep.append(combined_label)
 
     that_columns_to_apply += additional_that_columns
@@ -355,16 +387,18 @@ def align_diff_frames(
         kser_set, column_labels_applied = zip(
             *resolve_func(combined, this_columns_to_apply, that_columns_to_apply)
         )
-        columns_applied = [c.spark.column for c in kser_set]
+        columns_applied = list(kser_set)
         column_labels_applied = list(column_labels_applied)
     else:
         columns_applied = []
         column_labels_applied = []
 
-    applied = combined[columns_applied + columns_to_keep]
-    applied.columns = pd.MultiIndex.from_tuples(
-        column_labels_applied + column_labels_to_keep, names=combined.columns.names
-    )
+    applied = DataFrame(
+        combined._internal.with_new_columns(
+            columns_applied + columns_to_keep,
+            column_labels=column_labels_applied + column_labels_to_keep,
+        )
+    )  # type: DataFrame
 
     # 3. Restore the names back and deduplicate columns.
     this_labels = OrderedDict()


Test patch:
diff --git a/databricks/koalas/tests/test_dataframe.py b/databricks/koalas/tests/test_dataframe.py
index 5c5ab05587..8dfa0d9fc4 100644
--- a/databricks/koalas/tests/test_dataframe.py
+++ b/databricks/koalas/tests/test_dataframe.py
@@ -25,12 +25,19 @@
 import pyspark
 from pyspark import StorageLevel
 from pyspark.ml.linalg import SparseVector
+from pyspark.sql import functions as F
 
 from databricks import koalas as ks
 from databricks.koalas.config import option_context
 from databricks.koalas.exceptions import PandasNotImplementedError
 from databricks.koalas.frame import CachedDataFrame
 from databricks.koalas.missing.frame import _MissingPandasLikeDataFrame
+from databricks.koalas.typedef.typehints import (
+    extension_dtypes,
+    extension_dtypes_available,
+    extension_float_dtypes_available,
+    extension_object_dtypes_available,
+)
 from databricks.koalas.testing.utils import (
     ReusedSQLTestCase,
     SQLTestUtils,
@@ -88,6 +95,98 @@ def test_dataframe(self):
         index_cols = pdf.columns[column_mask]
         self.assert_eq(kdf[index_cols], pdf[index_cols])
 
+    def _check_extension(self, kdf, pdf):
+        if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+            self.assert_eq(kdf, pdf, check_exact=False)
+            for dtype in kdf.dtypes:
+                self.assertTrue(isinstance(dtype, extension_dtypes))
+        else:
+            self.assert_eq(kdf, pdf)
+
+    @unittest.skipIf(not extension_dtypes_available, "pandas extension dtypes are not available")
+    def test_extension_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "a": pd.Series([1, 2, None, 4], dtype="Int8"),
+                "b": pd.Series([1, None, None, 4], dtype="Int16"),
+                "c": pd.Series([1, 2, None, None], dtype="Int32"),
+                "d": pd.Series([None, 2, None, 4], dtype="Int64"),
+            }
+        )
+        kdf = ks.from_pandas(pdf)
+
+        self._check_extension(kdf, pdf)
+        self._check_extension(kdf + F.lit(1).cast("byte"), pdf + 1)
+        self._check_extension(kdf + kdf, pdf + pdf)
+
+    @unittest.skipIf(not extension_dtypes_available, "pandas extension dtypes are not available")
+    def test_astype_extension_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "a": [1, 2, None, 4],
+                "b": [1, None, None, 4],
+                "c": [1, 2, None, None],
+                "d": [None, 2, None, 4],
+            }
+        )
+        kdf = ks.from_pandas(pdf)
+
+        astype = {"a": "Int8", "b": "Int16", "c": "Int32", "d": "Int64"}
+
+        self._check_extension(kdf.astype(astype), pdf.astype(astype))
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_extension_object_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "a": pd.Series(["a", "b", None, "c"], dtype="string"),
+                "b": pd.Series([True, None, False, True], dtype="boolean"),
+            }
+        )
+        kdf = ks.from_pandas(pdf)
+
+        self._check_extension(kdf, pdf)
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_astype_extension_object_dtypes(self):
+        pdf = pd.DataFrame({"a": ["a", "b", None, "c"], "b": [True, None, False, True]})
+        kdf = ks.from_pandas(pdf)
+
+        astype = {"a": "string", "b": "boolean"}
+
+        self._check_extension(kdf.astype(astype), pdf.astype(astype))
+
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "pandas extension float dtypes are not available"
+    )
+    def test_extension_float_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "a": pd.Series([1.0, 2.0, None, 4.0], dtype="Float32"),
+                "b": pd.Series([1.0, None, 3.0, 4.0], dtype="Float64"),
+            }
+        )
+        kdf = ks.from_pandas(pdf)
+
+        self._check_extension(kdf, pdf)
+        self._check_extension(kdf + 1, pdf + 1)
+        self._check_extension(kdf + kdf, pdf + pdf)
+
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "pandas extension float dtypes are not available"
+    )
+    def test_astype_extension_float_dtypes(self):
+        pdf = pd.DataFrame({"a": [1.0, 2.0, None, 4.0], "b": [1.0, None, 3.0, 4.0]})
+        kdf = ks.from_pandas(pdf)
+
+        astype = {"a": "Float32", "b": "Float64"}
+
+        self._check_extension(kdf.astype(astype), pdf.astype(astype))
+
     def test_insert(self):
         #
         # Basic DataFrame
diff --git a/databricks/koalas/tests/test_ops_on_diff_frames.py b/databricks/koalas/tests/test_ops_on_diff_frames.py
index 36eacc3d5a..9af1f68b17 100644
--- a/databricks/koalas/tests/test_ops_on_diff_frames.py
+++ b/databricks/koalas/tests/test_ops_on_diff_frames.py
@@ -15,6 +15,7 @@
 #
 from distutils.version import LooseVersion
 from itertools import product
+import unittest
 
 import pandas as pd
 import numpy as np
@@ -23,7 +24,14 @@
 
 from databricks import koalas as ks
 from databricks.koalas.config import set_option, reset_option
+from databricks.koalas.frame import DataFrame
 from databricks.koalas.testing.utils import ReusedSQLTestCase, SQLTestUtils
+from databricks.koalas.typedef.typehints import (
+    extension_dtypes,
+    extension_dtypes_available,
+    extension_float_dtypes_available,
+    extension_object_dtypes_available,
+)
 
 
 class OpsOnDiffFramesEnabledTest(ReusedSQLTestCase, SQLTestUtils):
@@ -163,24 +171,61 @@ def test_no_matched_index(self):
             ).set_index("b")
 
     def test_arithmetic(self):
-        kdf1 = self.kdf1
-        kdf2 = self.kdf2
-        pdf1 = self.pdf1
-        pdf2 = self.pdf2
-        kser1 = self.kser1
-        pser1 = self.pser1
-        kser2 = self.kser2
-        pser2 = self.pser2
+        self._test_arithmetic_frame(self.pdf1, self.pdf2, check_extension=False)
+        self._test_arithmetic_series(self.pser1, self.pser2, check_extension=False)
+
+    @unittest.skipIf(not extension_dtypes_available, "pandas extension dtypes are not available")
+    def test_arithmetic_extension_dtypes(self):
+        self._test_arithmetic_frame(
+            self.pdf1.astype("Int64"), self.pdf2.astype("Int64"), check_extension=True
+        )
+        self._test_arithmetic_series(
+            self.pser1.astype(int).astype("Int64"),
+            self.pser2.astype(int).astype("Int64"),
+            check_extension=True,
+        )
+
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "pandas extension float dtypes are not available"
+    )
+    def test_arithmetic_extension_float_dtypes(self):
+        self._test_arithmetic_frame(
+            self.pdf1.astype("Float64"), self.pdf2.astype("Float64"), check_extension=True
+        )
+        self._test_arithmetic_series(
+            self.pser1.astype("Float64"), self.pser2.astype("Float64"), check_extension=True
+        )
+
+    def _test_arithmetic_frame(self, pdf1, pdf2, *, check_extension):
+        kdf1 = ks.from_pandas(pdf1)
+        kdf2 = ks.from_pandas(pdf2)
+
+        def assert_eq(actual, expected):
+            if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+                self.assert_eq(actual, expected, check_exact=not check_extension)
+                if check_extension:
+                    if isinstance(actual, DataFrame):
+                        for dtype in actual.dtypes:
+                            self.assertTrue(isinstance(dtype, extension_dtypes))
+                    else:
+                        self.assertTrue(isinstance(actual.dtype, extension_dtypes))
+            else:
+                self.assert_eq(actual, expected)
 
         # Series
-        self.assert_eq((kdf1.a - kdf2.b).sort_index(), (pdf1.a - pdf2.b).sort_index())
+        assert_eq((kdf1.a - kdf2.b).sort_index(), (pdf1.a - pdf2.b).sort_index())
 
-        self.assert_eq((kdf1.a * kdf2.a).sort_index(), (pdf1.a * pdf2.a).sort_index())
+        assert_eq((kdf1.a * kdf2.a).sort_index(), (pdf1.a * pdf2.a).sort_index())
 
-        self.assert_eq((kdf1["a"] / kdf2["a"]).sort_index(), (pdf1["a"] / pdf2["a"]).sort_index())
+        if check_extension and not extension_float_dtypes_available:
+            self.assert_eq(
+                (kdf1["a"] / kdf2["a"]).sort_index(), (pdf1["a"] / pdf2["a"]).sort_index()
+            )
+        else:
+            assert_eq((kdf1["a"] / kdf2["a"]).sort_index(), (pdf1["a"] / pdf2["a"]).sort_index())
 
         # DataFrame
-        self.assert_eq((kdf1 + kdf2).sort_index(), (pdf1 + pdf2).sort_index())
+        assert_eq((kdf1 + kdf2).sort_index(), (pdf1 + pdf2).sort_index())
 
         # Multi-index columns
         columns = pd.MultiIndex.from_tuples([("x", "a"), ("x", "b")])
@@ -190,65 +235,135 @@ def test_arithmetic(self):
         pdf2.columns = columns
 
         # Series
-        self.assert_eq(
+        assert_eq(
             (kdf1[("x", "a")] - kdf2[("x", "b")]).sort_index(),
             (pdf1[("x", "a")] - pdf2[("x", "b")]).sort_index(),
         )
 
-        self.assert_eq(
+        assert_eq(
             (kdf1[("x", "a")] - kdf2["x"]["b"]).sort_index(),
             (pdf1[("x", "a")] - pdf2["x"]["b"]).sort_index(),
         )
 
-        self.assert_eq(
+        assert_eq(
             (kdf1["x"]["a"] - kdf2[("x", "b")]).sort_index(),
             (pdf1["x"]["a"] - pdf2[("x", "b")]).sort_index(),
         )
 
         # DataFrame
-        self.assert_eq((kdf1 + kdf2).sort_index(), (pdf1 + pdf2).sort_index())
+        assert_eq((kdf1 + kdf2).sort_index(), (pdf1 + pdf2).sort_index())
+
+    def _test_arithmetic_series(self, pser1, pser2, *, check_extension):
+        kser1 = ks.from_pandas(pser1)
+        kser2 = ks.from_pandas(pser2)
+
+        def assert_eq(actual, expected):
+            if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+                self.assert_eq(actual, expected, check_exact=not check_extension)
+                if check_extension:
+                    self.assertTrue(isinstance(actual.dtype, extension_dtypes))
+            else:
+                self.assert_eq(actual, expected)
 
         # MultiIndex Series
-        self.assert_eq((kser1 + kser2).sort_index(), (pser1 + pser2).sort_index())
+        assert_eq((kser1 + kser2).sort_index(), (pser1 + pser2).sort_index())
 
-        self.assert_eq((kser1 - kser2).sort_index(), (pser1 - pser2).sort_index())
+        assert_eq((kser1 - kser2).sort_index(), (pser1 - pser2).sort_index())
 
-        self.assert_eq((kser1 * kser2).sort_index(), (pser1 * pser2).sort_index())
+        assert_eq((kser1 * kser2).sort_index(), (pser1 * pser2).sort_index())
 
-        self.assert_eq((kser1 / kser2).sort_index(), (pser1 / pser2).sort_index())
+        if check_extension and not extension_float_dtypes_available:
+            self.assert_eq((kser1 / kser2).sort_index(), (pser1 / pser2).sort_index())
+        else:
+            assert_eq((kser1 / kser2).sort_index(), (pser1 / pser2).sort_index())
 
     def test_arithmetic_chain(self):
-        kdf1 = self.kdf1
-        kdf2 = self.kdf2
-        kdf3 = self.kdf3
-        pdf1 = self.pdf1
-        pdf2 = self.pdf2
-        pdf3 = self.pdf3
-        kser1 = self.kser1
-        pser1 = self.pser1
-        kser2 = self.kser2
-        pser2 = self.pser2
-        kser3 = self.kser3
-        pser3 = self.pser3
+        self._test_arithmetic_chain_frame(self.pdf1, self.pdf2, self.pdf3, check_extension=False)
+        self._test_arithmetic_chain_series(
+            self.pser1, self.pser2, self.pser3, check_extension=False
+        )
 
-        # Series
-        self.assert_eq(
-            (kdf1.a - kdf2.b - kdf3.c).sort_index(), (pdf1.a - pdf2.b - pdf3.c).sort_index()
+    @unittest.skipIf(not extension_dtypes_available, "pandas extension dtypes are not available")
+    def test_arithmetic_chain_extension_dtypes(self):
+        self._test_arithmetic_chain_frame(
+            self.pdf1.astype("Int64"),
+            self.pdf2.astype("Int64"),
+            self.pdf3.astype("Int64"),
+            check_extension=True,
+        )
+        self._test_arithmetic_chain_series(
+            self.pser1.astype(int).astype("Int64"),
+            self.pser2.astype(int).astype("Int64"),
+            self.pser3.astype(int).astype("Int64"),
+            check_extension=True,
         )
 
-        self.assert_eq(
-            (kdf1.a * (kdf2.a * kdf3.c)).sort_index(), (pdf1.a * (pdf2.a * pdf3.c)).sort_index()
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "pandas extension float dtypes are not available"
+    )
+    def test_arithmetic_chain_extension_float_dtypes(self):
+        self._test_arithmetic_chain_frame(
+            self.pdf1.astype("Float64"),
+            self.pdf2.astype("Float64"),
+            self.pdf3.astype("Float64"),
+            check_extension=True,
+        )
+        self._test_arithmetic_chain_series(
+            self.pser1.astype("Float64"),
+            self.pser2.astype("Float64"),
+            self.pser3.astype("Float64"),
+            check_extension=True,
         )
 
-        self.assert_eq(
-            (kdf1["a"] / kdf2["a"] / kdf3["c"]).sort_index(),
-            (pdf1["a"] / pdf2["a"] / pdf3["c"]).sort_index(),
+    def _test_arithmetic_chain_frame(self, pdf1, pdf2, pdf3, *, check_extension):
+        kdf1 = ks.from_pandas(pdf1)
+        kdf2 = ks.from_pandas(pdf2)
+        kdf3 = ks.from_pandas(pdf3)
+
+        common_columns = set(kdf1.columns).intersection(kdf2.columns).intersection(kdf3.columns)
+
+        def assert_eq(actual, expected):
+            if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+                self.assert_eq(actual, expected, check_exact=not check_extension)
+                if check_extension:
+                    if isinstance(actual, DataFrame):
+                        for column, dtype in zip(actual.columns, actual.dtypes):
+                            if column in common_columns:
+                                self.assertTrue(isinstance(dtype, extension_dtypes))
+                            else:
+                                self.assertFalse(isinstance(dtype, extension_dtypes))
+                    else:
+                        self.assertTrue(isinstance(actual.dtype, extension_dtypes))
+            else:
+                self.assert_eq(actual, expected)
+
+        # Series
+        assert_eq((kdf1.a - kdf2.b - kdf3.c).sort_index(), (pdf1.a - pdf2.b - pdf3.c).sort_index())
+
+        assert_eq(
+            (kdf1.a * (kdf2.a * kdf3.c)).sort_index(), (pdf1.a * (pdf2.a * pdf3.c)).sort_index()
         )
 
+        if check_extension and not extension_float_dtypes_available:
+            self.assert_eq(
+                (kdf1["a"] / kdf2["a"] / kdf3["c"]).sort_index(),
+                (pdf1["a"] / pdf2["a"] / pdf3["c"]).sort_index(),
+            )
+        else:
+            assert_eq(
+                (kdf1["a"] / kdf2["a"] / kdf3["c"]).sort_index(),
+                (pdf1["a"] / pdf2["a"] / pdf3["c"]).sort_index(),
+            )
+
         # DataFrame
-        self.assert_eq(
-            (kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index(), almost=True
-        )
+        if check_extension and (
+            LooseVersion("1.0") <= LooseVersion(pd.__version__) < LooseVersion("1.1")
+        ):
+            self.assert_eq(
+                (kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index(), almost=True
+            )
+        else:
+            assert_eq((kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index())
 
         # Multi-index columns
         columns = pd.MultiIndex.from_tuples([("x", "a"), ("x", "b")])
@@ -260,30 +375,72 @@ def test_arithmetic_chain(self):
         kdf3.columns = columns
         pdf3.columns = columns
 
+        common_columns = set(kdf1.columns).intersection(kdf2.columns).intersection(kdf3.columns)
+
         # Series
-        self.assert_eq(
+        assert_eq(
             (kdf1[("x", "a")] - kdf2[("x", "b")] - kdf3[("y", "c")]).sort_index(),
             (pdf1[("x", "a")] - pdf2[("x", "b")] - pdf3[("y", "c")]).sort_index(),
         )
 
-        self.assert_eq(
+        assert_eq(
             (kdf1[("x", "a")] * (kdf2[("x", "b")] * kdf3[("y", "c")])).sort_index(),
             (pdf1[("x", "a")] * (pdf2[("x", "b")] * pdf3[("y", "c")])).sort_index(),
         )
 
         # DataFrame
-        self.assert_eq(
-            (kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index(), almost=True
-        )
+        if check_extension and (
+            LooseVersion("1.0") <= LooseVersion(pd.__version__) < LooseVersion("1.1")
+        ):
+            self.assert_eq(
+                (kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index(), almost=True
+            )
+        else:
+            assert_eq((kdf1 + kdf2 - kdf3).sort_index(), (pdf1 + pdf2 - pdf3).sort_index())
 
-        # MultiIndex Series
-        self.assert_eq((kser1 + kser2 - kser3).sort_index(), (pser1 + pser2 - pser3).sort_index())
+    def _test_arithmetic_chain_series(self, pser1, pser2, pser3, *, check_extension):
+        kser1 = ks.from_pandas(pser1)
+        kser2 = ks.from_pandas(pser2)
+        kser3 = ks.from_pandas(pser3)
 
-        self.assert_eq((kser1 * kser2 * kser3).sort_index(), (pser1 * pser2 * pser3).sort_index())
+        def assert_eq(actual, expected):
+            if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+                self.assert_eq(actual, expected, check_exact=not check_extension)
+                if check_extension:
+                    self.assertTrue(isinstance(actual.dtype, extension_dtypes))
+            else:
+                self.assert_eq(actual, expected)
 
-        self.assert_eq((kser1 - kser2 / kser3).sort_index(), (pser1 - pser2 / pser3).sort_index())
+        # MultiIndex Series
+        assert_eq((kser1 + kser2 - kser3).sort_index(), (pser1 + pser2 - pser3).sort_index())
+
+        assert_eq((kser1 * kser2 * kser3).sort_index(), (pser1 * pser2 * pser3).sort_index())
+
+        if check_extension and not extension_float_dtypes_available:
+            if LooseVersion(pd.__version__) >= LooseVersion("1.0"):
+                self.assert_eq(
+                    (kser1 - kser2 / kser3).sort_index(), (pser1 - pser2 / pser3).sort_index()
+                )
+            else:
+                expected = pd.Series(
+                    [249.0, np.nan, 0.0, 0.88, np.nan, np.nan, np.nan, np.nan, np.nan, -np.inf]
+                    + [np.nan, np.nan, np.nan, np.nan, np.nan, np.nan, np.nan],
+                    index=pd.MultiIndex(
+                        [
+                            ["cow", "falcon", "koala", "koalas", "lama"],
+                            ["length", "power", "speed", "weight"],
+                        ],
+                        [
+                            [0, 0, 0, 0, 0, 1, 1, 1, 1, 1, 2, 3, 3, 3, 4, 4, 4],
+                            [0, 1, 2, 2, 3, 0, 0, 1, 2, 3, 0, 0, 3, 3, 0, 2, 3],
+                        ],
+                    ),
+                )
+                self.assert_eq((kser1 - kser2 / kser3).sort_index(), expected)
+        else:
+            assert_eq((kser1 - kser2 / kser3).sort_index(), (pser1 - pser2 / pser3).sort_index())
 
-        self.assert_eq((kser1 + kser2 * kser3).sort_index(), (pser1 + pser2 * pser3).sort_index())
+        assert_eq((kser1 + kser2 * kser3).sort_index(), (pser1 + pser2 * pser3).sort_index())
 
     def test_mod(self):
         pser = pd.Series([100, None, -300, None, 500, -700])
@@ -360,6 +517,48 @@ def test_bitwise(self):
         self.assert_eq(pser1 | pser2, (kser1 | kser2).sort_index())
         self.assert_eq(pser1 & pser2, (kser1 & kser2).sort_index())
 
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_bitwise_extension_dtype(self):
+        def assert_eq(actual, expected):
+            if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+                self.assert_eq(actual, expected, check_exact=False)
+                self.assertTrue(isinstance(actual.dtype, extension_dtypes))
+            else:
+                self.assert_eq(actual, expected)
+
+        pser1 = pd.Series(
+            [True, False, True, False, np.nan, np.nan, True, False, np.nan], dtype="boolean"
+        )
+        pser2 = pd.Series(
+            [True, False, False, True, True, False, np.nan, np.nan, np.nan], dtype="boolean"
+        )
+        kser1 = ks.from_pandas(pser1)
+        kser2 = ks.from_pandas(pser2)
+
+        assert_eq((kser1 | kser2).sort_index(), pser1 | pser2)
+        assert_eq((kser1 & kser2).sort_index(), pser1 & pser2)
+
+        pser1 = pd.Series([True, False, np.nan], index=list("ABC"), dtype="boolean")
+        pser2 = pd.Series([False, True, np.nan], index=list("DEF"), dtype="boolean")
+        kser1 = ks.from_pandas(pser1)
+        kser2 = ks.from_pandas(pser2)
+
+        # a pandas bug?
+        # assert_eq((kser1 | kser2).sort_index(), pser1 | pser2)
+        # assert_eq((kser1 & kser2).sort_index(), pser1 & pser2)
+        assert_eq(
+            (kser1 | kser2).sort_index(),
+            pd.Series([True, None, None, None, True, None], index=list("ABCDEF"), dtype="boolean"),
+        )
+        assert_eq(
+            (kser1 & kser2).sort_index(),
+            pd.Series(
+                [None, False, None, False, None, None], index=list("ABCDEF"), dtype="boolean"
+            ),
+        )
+
     def test_concat_column_axis(self):
         pdf1 = pd.DataFrame({"A": [0, 2, 4], "B": [1, 3, 5]}, index=[1, 2, 3])
         pdf1.columns.names = ["AB"]
diff --git a/databricks/koalas/tests/test_series.py b/databricks/koalas/tests/test_series.py
index 90538779fd..8502046aef 100644
--- a/databricks/koalas/tests/test_series.py
+++ b/databricks/koalas/tests/test_series.py
@@ -31,6 +31,7 @@
 import pandas as pd
 import pyspark
 from pyspark.ml.linalg import SparseVector
+from pyspark.sql import functions as F
 
 from databricks import koalas as ks
 from databricks.koalas.testing.utils import (
@@ -40,6 +41,12 @@
 )
 from databricks.koalas.exceptions import PandasNotImplementedError
 from databricks.koalas.missing.series import MissingPandasLikeSeries
+from databricks.koalas.typedef.typehints import (
+    extension_dtypes,
+    extension_dtypes_available,
+    extension_float_dtypes_available,
+    extension_object_dtypes_available,
+)
 
 
 class SeriesTest(ReusedSQLTestCase, SQLTestUtils):
@@ -83,6 +90,59 @@ def test_repr_cache_invalidation(self):
         s.rename("a", inplace=True)
         self.assertEqual(s.__repr__(), s.rename("a").__repr__())
 
+    def _check_extension(self, kser, pser):
+        if LooseVersion("1.1") <= LooseVersion(pd.__version__) < LooseVersion("1.2.2"):
+            self.assert_eq(kser, pser, check_exact=False)
+            self.assertTrue(isinstance(kser.dtype, extension_dtypes))
+        else:
+            self.assert_eq(kser, pser)
+
+    @unittest.skipIf(not extension_dtypes_available, "pandas extension dtypes are not available")
+    def test_extension_dtypes(self):
+        for pser in [
+            pd.Series([1, 2, None, 4], dtype="Int8"),
+            pd.Series([1, 2, None, 4], dtype="Int16"),
+            pd.Series([1, 2, None, 4], dtype="Int32"),
+            pd.Series([1, 2, None, 4], dtype="Int64"),
+        ]:
+            kser = ks.from_pandas(pser)
+
+            self._check_extension(kser, pser)
+            self._check_extension(kser + F.lit(1).cast("byte"), pser + 1)
+            self._check_extension(kser + kser, pser + pser)
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_extension_object_dtypes(self):
+        # string
+        pser = pd.Series(["a", None, "c", "d"], dtype="string")
+        kser = ks.from_pandas(pser)
+
+        self._check_extension(kser, pser)
+
+        # boolean
+        pser = pd.Series([True, False, True, None], dtype="boolean")
+        kser = ks.from_pandas(pser)
+
+        self._check_extension(kser, pser)
+        self._check_extension(kser & kser, pser & pser)
+        self._check_extension(kser | kser, pser | pser)
+
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "pandas extension float dtypes are not available"
+    )
+    def test_extension_float_dtypes(self):
+        for pser in [
+            pd.Series([1.0, 2.0, None, 4.0], dtype="Float32"),
+            pd.Series([1.0, 2.0, None, 4.0], dtype="Float64"),
+        ]:
+            kser = ks.from_pandas(pser)
+
+            self._check_extension(kser, pser)
+            self._check_extension(kser + 1, pser + 1)
+            self._check_extension(kser + kser, pser + pser)
+
     def test_empty_series(self):
         pser_a = pd.Series([], dtype="i1")
         pser_b = pd.Series([], dtype="str")
@@ -288,7 +348,33 @@ def test_or(self):
         )
         kdf = ks.from_pandas(pdf)
 
-        self.assert_eq(pdf["left"] | pdf["right"], kdf["left"] | kdf["right"])
+        self.assert_eq(kdf["left"] | kdf["right"], pdf["left"] | pdf["right"])
+        self.assert_eq(kdf["left"] | True, pdf["left"] | True)
+        self.assert_eq(kdf["left"] | False, pdf["left"] | False)
+        self.assert_eq(kdf["left"] | None, pdf["left"] | None)
+        self.assert_eq(True | kdf["right"], True | pdf["right"])
+        self.assert_eq(False | kdf["right"], False | pdf["right"])
+        self.assert_eq(None | kdf["right"], None | pdf["right"])
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_or_extenstion_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "left": [True, False, True, False, np.nan, np.nan, True, False, np.nan],
+                "right": [True, False, False, True, True, False, np.nan, np.nan, np.nan],
+            }
+        ).astype("boolean")
+        kdf = ks.from_pandas(pdf)
+
+        self._check_extension(kdf["left"] | kdf["right"], pdf["left"] | pdf["right"])
+        self._check_extension(kdf["left"] | True, pdf["left"] | True)
+        self._check_extension(kdf["left"] | False, pdf["left"] | False)
+        self._check_extension(kdf["left"] | pd.NA, pdf["left"] | pd.NA)
+        self._check_extension(True | kdf["right"], True | pdf["right"])
+        self._check_extension(False | kdf["right"], False | pdf["right"])
+        self._check_extension(pd.NA | kdf["right"], pd.NA | pdf["right"])
 
     def test_and(self):
         pdf = pd.DataFrame(
@@ -299,9 +385,33 @@ def test_and(self):
         )
         kdf = ks.from_pandas(pdf)
 
-        self.assert_eq(
-            pdf["left"] & pdf["right"], kdf["left"] & kdf["right"],
-        )
+        self.assert_eq(kdf["left"] & kdf["right"], pdf["left"] & pdf["right"])
+        self.assert_eq(kdf["left"] & True, pdf["left"] & True)
+        self.assert_eq(kdf["left"] & False, pdf["left"] & False)
+        self.assert_eq(kdf["left"] & None, pdf["left"] & None)
+        self.assert_eq(True & kdf["right"], True & pdf["right"])
+        self.assert_eq(False & kdf["right"], False & pdf["right"])
+        self.assert_eq(None & kdf["right"], None & pdf["right"])
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "pandas extension object dtypes are not available"
+    )
+    def test_and_extenstion_dtypes(self):
+        pdf = pd.DataFrame(
+            {
+                "left": [True, False, True, False, np.nan, np.nan, True, False, np.nan],
+                "right": [True, False, False, True, True, False, np.nan, np.nan, np.nan],
+            }
+        ).astype("boolean")
+        kdf = ks.from_pandas(pdf)
+
+        self._check_extension(kdf["left"] & kdf["right"], pdf["left"] & pdf["right"])
+        self._check_extension(kdf["left"] & True, pdf["left"] & True)
+        self._check_extension(kdf["left"] & False, pdf["left"] & False)
+        self._check_extension(kdf["left"] & pd.NA, pdf["left"] & pd.NA)
+        self._check_extension(True & kdf["right"], True & pdf["right"])
+        self._check_extension(False & kdf["right"], False & pdf["right"])
+        self._check_extension(pd.NA & kdf["right"], pd.NA & pdf["right"])
 
     def test_to_numpy(self):
         pser = pd.Series([1, 2, 3, 4, 5, 6, 7], name="x")
@@ -1346,8 +1456,7 @@ def test_shift(self):
         with self.assertRaisesRegex(ValueError, "periods should be an int; however"):
             kser.shift(periods=1.5)
 
-    def test_astype(self):
-        pser = pd.Series([10, 20, 15, 30, 45], name="x")
+    def _test_numeric_astype(self, pser):
         kser = ks.Series(pser)
 
         self.assert_eq(kser.astype(int), pser.astype(int))
@@ -1382,6 +1491,53 @@ def test_astype(self):
         self.assert_eq(kser.astype("str"), pser.astype("str"))
         self.assert_eq(kser.astype("U"), pser.astype("U"))
 
+        if extension_dtypes_available:
+            from pandas import Int8Dtype, Int16Dtype, Int32Dtype, Int64Dtype
+
+            self._check_extension(kser.astype("Int8"), pser.astype("Int8"))
+            self._check_extension(kser.astype("Int16"), pser.astype("Int16"))
+            self._check_extension(kser.astype("Int32"), pser.astype("Int32"))
+            self._check_extension(kser.astype("Int64"), pser.astype("Int64"))
+            self._check_extension(kser.astype(Int8Dtype()), pser.astype(Int8Dtype()))
+            self._check_extension(kser.astype(Int16Dtype()), pser.astype(Int16Dtype()))
+            self._check_extension(kser.astype(Int32Dtype()), pser.astype(Int32Dtype()))
+            self._check_extension(kser.astype(Int64Dtype()), pser.astype(Int64Dtype()))
+
+        if extension_object_dtypes_available:
+            from pandas import StringDtype
+
+            if LooseVersion(pd.__version__) >= LooseVersion("1.1"):
+                self._check_extension(kser.astype("string"), pser.astype("string"))
+                self._check_extension(kser.astype(StringDtype()), pser.astype(StringDtype()))
+            else:
+                self._check_extension(
+                    kser.astype("string"),
+                    pd.Series(["10", "20", "15", "30", "45"], name="x", dtype="string"),
+                )
+                self._check_extension(
+                    kser.astype(StringDtype()),
+                    pd.Series(["10", "20", "15", "30", "45"], name="x", dtype=StringDtype()),
+                )
+
+        if extension_float_dtypes_available:
+            from pandas import Float32Dtype, Float64Dtype
+
+            self._check_extension(kser.astype("Float32"), pser.astype("Float32"))
+            self._check_extension(kser.astype("Float64"), pser.astype("Float64"))
+            self._check_extension(kser.astype(Float32Dtype()), pser.astype(Float32Dtype()))
+            self._check_extension(kser.astype(Float64Dtype()), pser.astype(Float64Dtype()))
+
+    def test_astype(self):
+        psers = [pd.Series([10, 20, 15, 30, 45], name="x")]
+
+        if extension_dtypes_available:
+            psers.append(pd.Series([10, 20, 15, 30, 45], name="x", dtype="Int64"))
+        if extension_float_dtypes_available:
+            psers.append(pd.Series([10, 20, 15, 30, 45], name="x", dtype="Float64"))
+
+        for pser in psers:
+            self._test_numeric_astype(pser)
+
         pser = pd.Series([10, 20, 15, 30, 45, None, np.nan], name="x")
         kser = ks.Series(pser)
 
@@ -1399,12 +1555,37 @@ def test_astype(self):
             self.assert_eq(kser.astype(str), pser.astype(str))
         self.assert_eq(kser.str.strip().astype(bool), pser.str.strip().astype(bool))
 
+        if extension_object_dtypes_available:
+            from pandas import StringDtype
+
+            self._check_extension(kser.astype("string"), pser.astype("string"))
+            self._check_extension(kser.astype(StringDtype()), pser.astype(StringDtype()))
+
         pser = pd.Series([True, False, None], name="x")
         kser = ks.Series(pser)
 
         self.assert_eq(kser.astype(bool), pser.astype(bool))
         self.assert_eq(kser.astype(str), pser.astype(str))
 
+        if extension_object_dtypes_available:
+            from pandas import BooleanDtype, StringDtype
+
+            self._check_extension(kser.astype("boolean"), pser.astype("boolean"))
+            self._check_extension(kser.astype(BooleanDtype()), pser.astype(BooleanDtype()))
+
+            if LooseVersion(pd.__version__) >= LooseVersion("1.1"):
+                self._check_extension(kser.astype("string"), pser.astype("string"))
+                self._check_extension(kser.astype(StringDtype()), pser.astype(StringDtype()))
+            else:
+                self._check_extension(
+                    kser.astype("string"),
+                    pd.Series(["True", "False", None], name="x", dtype="string"),
+                )
+                self._check_extension(
+                    kser.astype(StringDtype()),
+                    pd.Series(["True", "False", None], name="x", dtype=StringDtype()),
+                )
+
         pser = pd.Series(["2020-10-27 00:00:01", None], name="x")
         kser = ks.Series(pser)
 
@@ -1414,6 +1595,16 @@ def test_astype(self):
         self.assert_eq(kser.astype("M").astype(str), pser.astype("M").astype(str))
         self.assert_eq(kser.astype("M").dt.date.astype(str), pser.astype("M").dt.date.astype(str))
 
+        if extension_object_dtypes_available:
+            from pandas import StringDtype
+
+            self._check_extension(
+                kser.astype("M").astype("string"), pser.astype("M").astype("string")
+            )
+            self._check_extension(
+                kser.astype("M").astype(StringDtype()), pser.astype("M").astype(StringDtype())
+            )
+
         with self.assertRaisesRegex(TypeError, "not understood"):
             kser.astype("int63")
 
diff --git a/databricks/koalas/tests/test_typedef.py b/databricks/koalas/tests/test_typedef.py
index 77e57a9a50..41fa95ba9d 100644
--- a/databricks/koalas/tests/test_typedef.py
+++ b/databricks/koalas/tests/test_typedef.py
@@ -40,7 +40,13 @@
     TimestampType,
 )
 
-from databricks.koalas.typedef import infer_return_type, as_spark_type
+from databricks.koalas.typedef import (
+    infer_return_type,
+    as_spark_type,
+    extension_dtypes_available,
+    extension_float_dtypes_available,
+    extension_object_dtypes_available,
+)
 from databricks import koalas as ks
 
 
@@ -277,3 +283,45 @@ def test_as_spark_type(self):
 
         with self.assertRaisesRegex(TypeError, "Type uint64 was not understood."):
             as_spark_type(np.dtype("uint64"))
+
+    @unittest.skipIf(not extension_dtypes_available, "The pandas extension types are not available")
+    def test_as_spark_type_extension_dtypes(self):
+        from pandas import Int8Dtype, Int16Dtype, Int32Dtype, Int64Dtype
+
+        type_mapper = {
+            Int8Dtype(): ByteType(),
+            Int16Dtype(): ShortType(),
+            Int32Dtype(): IntegerType(),
+            Int64Dtype(): LongType(),
+        }
+
+        for extension_dtype, spark_type in type_mapper.items():
+            self.assertEqual(as_spark_type(extension_dtype), spark_type)
+
+    @unittest.skipIf(
+        not extension_object_dtypes_available, "The pandas extension object types are not available"
+    )
+    def test_as_spark_type_extension_object_dtypes(self):
+        from pandas import BooleanDtype, StringDtype
+
+        type_mapper = {
+            BooleanDtype(): BooleanType(),
+            StringDtype(): StringType(),
+        }
+
+        for extension_dtype, spark_type in type_mapper.items():
+            self.assertEqual(as_spark_type(extension_dtype), spark_type)
+
+    @unittest.skipIf(
+        not extension_float_dtypes_available, "The pandas extension float types are not available"
+    )
+    def test_as_spark_type_extension_float_dtypes(self):
+        from pandas import Float32Dtype, Float64Dtype
+
+        type_mapper = {
+            Float32Dtype(): FloatType(),
+            Float64Dtype(): DoubleType(),
+        }
+
+        for extension_dtype, spark_type in type_mapper.items():
+            self.assertEqual(as_spark_type(extension_dtype), spark_type)