Add duplicated to ExtensionArray api

Author: tehunter

pandas/core/arrays/base.py

@@ -31,6 +31,7 @@
     ArrayLike,
     AstypeArg,
     AxisInt,
+    DropKeep,
     Dtype,
     FillnaOptions,
     PositionalIndexer,
@@ -78,6 +79,7 @@
     roperator,
 )
 from pandas.core.algorithms import (
+    duplicated,
     factorize_array,
     isin,
     mode,
@@ -888,6 +890,32 @@ def shift(self, periods: int = 1, fill_value: object = None) -> ExtensionArray:
             b = empty
         return self._concat_same_type([a, b])
 
+    def duplicated(
+        self: ExtensionArrayT,
+        keep: DropKeep = "first",
+    ) -> npt.NDArray[np.bool_]:
+        """
+        Indicate duplicate ExtensionArray values.
+
+        Duplicated values are indicated as ``True`` values in the resulting
+        array. Either all duplicates, all except the first or all except the
+        last occurrence of duplicates can be indicated.
+
+        The base implementation uses ``pd.core.algorithms.duplicated``
+
+        Parameters
+        ----------
+        keep : DropKeep, optional
+            See ``Series.duplicated``, by default "first"
+
+        Returns
+        -------
+        npt.NDArray[np.bool_]
+            A numpy boolean array indicating duplicates as determined by ``keep``
+            argument
+        """
+        return duplicated(self, keep=keep)
+
     def unique(self: ExtensionArrayT) -> ExtensionArrayT:
         """
         Compute the ExtensionArray of unique values.

pandas/core/base.py

@@ -1316,7 +1316,12 @@ def drop_duplicates(self, keep: DropKeep = "first"):
 
     @final
     def _duplicated(self, keep: DropKeep = "first") -> npt.NDArray[np.bool_]:
-        return duplicated(self._values, keep=keep)
+        values = self._values
+        if not isinstance(values, np.ndarray):
+            # Going through EA.duplicated directly can improve performance GH#48424
+            return values.duplicated(keep=keep)
+
+        return duplicated(values, keep=keep)
 
     def _arith_method(self, other, op):
         res_name = ops.get_op_result_name(self, other)