Fix is_unique to respect the current Spark column (#981)

Author: ueshin

databricks/koalas/series.py

@@ -911,18 +911,17 @@ def is_unique(self):
         >>> ks.Series([1, 2, 3, None]).is_unique
         True
         """
-        sdf = self._kdf._sdf.select(self._scol)
-        col = self._scol
+        scol = self._scol
 
         # Here we check:
         #   1. the distinct count without nulls and count without nulls for non-null values
         #   2. count null values and see if null is a distinct value.
         #
         # This workaround is in order to calculate the distinct count including nulls in
         # single pass. Note that COUNT(DISTINCT expr) in Spark is designed to ignore nulls.
-        return sdf.select(
-            (F.count(col) == F.countDistinct(col)) &
-            (F.count(F.when(col.isNull(), 1).otherwise(None)) <= 1)
+        return self._kdf._sdf.select(
+            (F.count(scol) == F.countDistinct(scol)) &
+            (F.count(F.when(scol.isNull(), 1).otherwise(None)) <= 1)
         ).collect()[0][0]
 
     def reset_index(self, level=None, drop=False, name=None, inplace=False):

databricks/koalas/tests/test_series.py

@@ -434,18 +434,22 @@ def test_is_unique(self):
         pser = pd.Series([1, 2, 2, None, None])
         kser = ks.from_pandas(pser)
         self.assertEqual(False, kser.is_unique)
+        self.assertEqual(False, (kser + 1).is_unique)
 
         pser = pd.Series([1, None, None])
         kser = ks.from_pandas(pser)
         self.assertEqual(False, kser.is_unique)
+        self.assertEqual(False, (kser + 1).is_unique)
 
         pser = pd.Series([1])
         kser = ks.from_pandas(pser)
         self.assertEqual(pser.is_unique, kser.is_unique)
+        self.assertEqual((pser + 1).is_unique, (kser + 1).is_unique)
 
         pser = pd.Series([1, 1, 1])
         kser = ks.from_pandas(pser)
         self.assertEqual(pser.is_unique, kser.is_unique)
+        self.assertEqual((pser + 1).is_unique, (kser + 1).is_unique)
 
     def test_to_list(self):
         if LooseVersion(pd.__version__) >= LooseVersion("0.24.0"):