Repository: modin-project/modin
Pull request: PERF-#5554: Implement `drop_duplicates` via new `duplicated`
Signed-off-by: Igoshev, Iaroslav <iaroslav.igoshev@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #5554  <!-- issue must be created for each patch -->
- [x] tests added and passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->

**PERF on 8 cores with Ray engine**

*DataFrame.duplicated()*

| pandas | Modin master | PR 5587 |
|---|---|---|
| 0.07 s | 11.84 s | 0.03 s |

*DataFrame.drop_duplicates()*

| pandas | Modin master | PR 5587 |
|---|---|---|
| 0.07 s | 11.90 s | 0.05 s |

Problem statement:
`.drop_duplicates()` works much slower on Modin than on Pandas
Modin's `.drop_duplicates` implementation works approximately 10x-100x times slower than pandas. We'd like to find some room for improvement here.

```python
import modin.pandas as pd
import pandas
import numpy as np
import timeit

NROWS = 1_000_000
RAND_LOW = 0
RAND_HIGH = 10_000_000

data = {
    "col0": np.tile(
        np.concatenate([np.random.randint(RAND_LOW, RAND_HIGH, NROWS // 100 - 1), [1]]),
        100,
    ),
    "col1": np.tile(
        np.concatenate([np.random.randint(RAND_LOW, RAND_HIGH, NROWS // 100 - 1), [1]]),
        100,
    ),
}

md_df = pd.DataFrame(data)
pd_df = pandas.DataFrame(data)

# drop_duplicate: 
#   input_shape = (1_000_000, 2)
#   output_shape = (10_000, 2)

NRUNS = 5
md_res = timeit.timeit(lambda: md_df.drop_duplicates(), number=NRUNS)
pd_res = timeit.timeit(lambda: pd_df.drop_duplicates(), number=NRUNS)

print(f"Cumulative time of {NRUNS} runs for modin: {md_res}s.") # ~10s.
print(f"Cumulative time of {NRUNS} runs for pandas: {pd_res}s.") # ~0.3s
```

We may use this issue as an epic to reference all of the perf problems with `.drop_duplicates()`

Hints from issue discussion:

Commit message: PERF-#5554: Implement `drop_duplicates` via new `duplicated`

Signed-off-by: Igoshev, Iaroslav <iaroslav.igoshev@intel.com>

Patch to review:
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 3a65e12b6c8..e0aa7093e9b 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -1962,6 +1962,23 @@ def dropna(self, **kwargs):  # noqa: PR02
         """
         return DataFrameDefault.register(pandas.DataFrame.dropna)(self, **kwargs)
 
+    @doc_utils.add_refer_to("DataFrame.duplicated")
+    def duplicated(self, **kwargs):
+        """
+        Return boolean Series denoting duplicate rows.
+
+        Parameters
+        ----------
+        **kwargs : dict
+            Additional keyword arguments to be passed in to `pandas.DataFrame.duplicated`.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            New QueryCompiler containing boolean Series denoting duplicate rows.
+        """
+        return DataFrameDefault.register(pandas.DataFrame.duplicated)(self, **kwargs)
+
     @doc_utils.add_refer_to("DataFrame.nlargest")
     def nlargest(self, n=5, columns=None, keep="first"):
         """
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index fd825302130..9cb6c84bc74 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -38,6 +38,7 @@
 from collections.abc import Iterable
 from typing import List, Hashable
 import warnings
+import hashlib
 
 from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
 from modin.config import Engine
@@ -2363,6 +2364,47 @@ def drop(self, index=None, columns=None, errors: str = "raise"):
 
     # END Drop/Dropna
 
+    def duplicated(self, **kwargs):
+        def _compute_hash(df):
+            return df.apply(
+                lambda s: hashlib.new("md5", str(tuple(s)).encode()).hexdigest(), axis=1
+            ).to_frame()
+
+        def _compute_duplicated(df):
+            return df.duplicated(**kwargs).to_frame()
+
+        new_index = self._modin_frame._index_cache
+        new_columns = [MODIN_UNNAMED_SERIES_LABEL]
+        if len(self.columns) > 1:
+            # if the number of columns we are checking for duplicates is larger than 1,
+            # we must hash them to generate a single value that can be compared across rows.
+            hashed_modin_frame = self._modin_frame.apply_full_axis(
+                1,
+                _compute_hash,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.dtype("O"),
+            )
+            new_modin_frame = hashed_modin_frame.apply_full_axis(
+                0,
+                _compute_duplicated,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.bool_,
+            )
+        else:
+            new_modin_frame = self._modin_frame.apply_full_axis(
+                0,
+                _compute_duplicated,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.bool_,
+            )
+        return self.__constructor__(new_modin_frame, shape_hint="column")
+
     # Insert
     # This method changes the shape of the resulting data. In Pandas, this
     # operation is always inplace, but this object is immutable, so we just
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index f9087be81e7..9e30b3ae9bf 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -1384,18 +1384,19 @@ def drop_duplicates(
         Return `BasePandasDataset` with duplicate rows removed.
         """
         inplace = validate_bool_kwarg(inplace, "inplace")
-        subset = kwargs.get("subset", None)
         ignore_index = kwargs.get("ignore_index", False)
+        subset = kwargs.get("subset", None)
         if subset is not None:
             if is_list_like(subset):
                 if not isinstance(subset, list):
                     subset = list(subset)
             else:
                 subset = [subset]
-            duplicates = self.duplicated(keep=keep, subset=subset)
+            df = self[subset]
         else:
-            duplicates = self.duplicated(keep=keep)
-        result = self[~duplicates]
+            df = self
+        duplicated = df.duplicated(keep=keep)
+        result = self[~duplicated]
         if ignore_index:
             result.index = pandas.RangeIndex(stop=len(result))
         if inplace:
diff --git a/modin/pandas/dataframe.py b/modin/pandas/dataframe.py
index f603489be10..2980be08a4d 100644
--- a/modin/pandas/dataframe.py
+++ b/modin/pandas/dataframe.py
@@ -334,19 +334,10 @@ def duplicated(self, subset=None, keep="first"):  # noqa: PR01, RT01, D200
         """
         Return boolean ``Series`` denoting duplicate rows.
         """
-        import hashlib
-
         df = self[subset] if subset is not None else self
-        # if the number of columns we are checking for duplicates is larger than 1, we must
-        # hash them to generate a single value that can be compared across rows.
-        if len(df.columns) > 1:
-            hashed = df.apply(
-                lambda s: hashlib.new("md5", str(tuple(s)).encode()).hexdigest(), axis=1
-            ).to_frame()
-        else:
-            hashed = df
-        duplicates = hashed.apply(lambda s: s.duplicated(keep=keep)).squeeze(axis=1)
-        # remove Series name which was assigned automatically by .apply
+        new_qc = df._query_compiler.duplicated(keep=keep)
+        duplicates = self._reduce_dimension(new_qc)
+        # remove Series name which was assigned automatically by .apply in QC
         duplicates.name = None
         return duplicates
 


Code review comments:
diff_hunk: @@ -2363,6 +2364,47 @@ def drop(self, index=None, columns=None, errors: str = "raise"):
 
     # END Drop/Dropna
 
+    def duplicated(self, **kwargs):
+        def _compute_hash(df):
+            return df.apply(
+                lambda s: hashlib.new("md5", str(tuple(s)).encode()).hexdigest(), axis=1
+            ).to_frame()
+
+        def _compute_duplicated(df):
+            return df.duplicated(**kwargs).to_frame()
+
+        new_index = self._modin_frame._index_cache
+        new_columns = [MODIN_UNNAMED_SERIES_LABEL]
+        if len(self.columns) > 1:
+            # if the number of columns we are checking for duplicates is larger than 1,
+            # we must hash them to generate a single value that can be compared across rows.
+            hashed_modin_frame = self._modin_frame.apply_full_axis(
+                1,
+                _compute_hash,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.dtype("O"),
+            )
+            new_modin_frame = hashed_modin_frame.apply_full_axis(
+                0,
+                _compute_duplicated,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.bool_,
+            )
+        else:
+            new_modin_frame = self._modin_frame.apply_full_axis(
+                0,
+                _compute_duplicated,
+                new_index=new_index,
+                new_columns=new_columns,
+                keep_partitioning=False,
+                dtypes=np.bool_,
+            )
path: modin/core/storage_formats/pandas/query_compiler.py
text: @user1:
```suggestion
        else:
            hashed_modin_frame = self._modin_frame
        new_modin_frame = hashed_modin_frame.apply_full_axis(
            0,
            _compute_duplicated,
            new_index=new_index,
            new_columns=new_columns,
            keep_partitioning=False,
            dtypes=np.bool_,
        )
```