Repository: modin-project/modin
Pull request: FIX-#4467: 'drop_duplicates' no longer removes items based on index value

Signed-off-by: Joshua Sacher <joshuasacher@gmail.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?

Fix a case where, if index values were not unique, multiple values could be incorrectly removed when using `drop_duplicates`

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [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 #4467  <!-- issue must be created for each patch -->
- [ ] 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 -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
drop_duplicates after pd.concat not working as expected
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: MacOS 12.3.1
- **Modin version** (`modin.__version__`): 0.14.1
- **Python version**: Python 3.9.12
- **Code we can use to reproduce**:

```python
import pandas as pd
import modin.pandas as mpd
import numpy as np

# Pandas
np.random.seed(0)
df1 = pd.DataFrame(np.random.randint(0, 1000, size=1000), columns=["A"])
np.random.seed(1)
df2 = pd.DataFrame(np.random.randint(0, 1000, size=1000), columns=["A"])

df = pd.concat([df1, df2])
print(len(df))  # 2000
print(len(df.drop_duplicates()))  # 874

# Modin
np.random.seed(0)
df1 = mpd.DataFrame(np.random.randint(0, 1000, size=1000), columns=["A"])
np.random.seed(1)
df2 = mpd.DataFrame(np.random.randint(0, 1000, size=1000), columns=["A"])

df = mpd.concat([df1, df2])
print(len(df))  # 2000
print(len(df.drop_duplicates()))  # 322 ?!?

# Issue caused by index? Use ignore_index=True to fix
df = mpd.concat([df1, df2], ignore_index=True)
print(len(df))  # 2000
print(len(df.drop_duplicates()))  # 874
```

### Describe the problem

After combining DataFrames using `mpd.concat()`, dropping duplicate values removes WAY too many items. I believe this is related to the indexing of the combined DataFrames, as using `ignore_index=True` in the `concat()` call seems to fix the issue.

In regular pandas, the result is the same whether or not `ignore_index=True` is used.

The above example is a toy one. I noticed this when combining large CSV files. Happy to share those, but this takes much less time to run!

Hints from issue discussion:
Tested with both Ray and Dask as the engine. Both give the same results.
Think I found the issue in `BasePandasDataset` class in `base.py`

```python
    def drop_duplicates(
        self, keep="first", inplace=False, **kwargs
    ):  # noqa: PR01, RT01, D200
        """
        Return `BasePandasDataset` with duplicate rows removed.
        """
        inplace = validate_bool_kwarg(inplace, "inplace")
        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)
        else:
            duplicates = self.duplicated(keep=keep)
        indices = duplicates.values.nonzero()[0]
        return self.drop(index=self.index[indices], inplace=inplace)
```

The `return` statement is using the index values, but the indices are duplicated after the `concat`. This leads to extra rows getting dropped erroneously.

My solution may be overly simplistic, but instead of this:
```python
        indices = duplicates.values.nonzero()[0]
        return self.drop(index=self.index[indices], inplace=inplace)
```
could you do something like what [Pandas does](https://github.com/pandas-dev/pandas/blob/a160caff845db846516ff5f973c5c0a09a605d8d/pandas/core/frame.py#L6363) (shown below)?
```python
        result = self[-duplicated]
        if ignore_index:
            result.index = default_index(len(result))

        if inplace:
            self._update_inplace(result)
            return None
        else:
            return result
```

(sorry if this isn't useful -- I'm much better at breaking things than fixing them!)
@jrsacher thank you for reporting this bug! I was able to reproduce it at commit cf0eb393daa41abdd8cdf32b52ceee938cdcbe1a. If I see a quick fix, I'll assign the issue to myself and make a PR.
@jrsacher I'm sorry, but I missed your last message. I think you've diagnosed the bug correctly. I think making the Modin implementation more like the pandas implementation as you suggested should work. 

Would you be interested in contributing this fix yourself? Members of \@modin-project/modin-contributors, including me, can help you make the fix and get it merged. You could follow the guide [here](https://github.com/modin-project/modin/blob/cf0eb393daa41abdd8cdf32b52ceee938cdcbe1a/contributing/contributing.md) to get set up for Modin development.
@mvashishtha Thanks for the push to tackle this myself. I gave it a shot. Looking forward to getting feedback!

Commit message: FIX-#4467: 'drop_duplicates' no longer removes items based on index value

Signed-off-by: Joshua Sacher <joshuasacher@gmail.com>

Patch to review:
diff --git a/docs/release_notes/release_notes-0.15.0.rst b/docs/release_notes/release_notes-0.15.0.rst
index 52d41b5b8f1..224b378af7b 100644
--- a/docs/release_notes/release_notes-0.15.0.rst
+++ b/docs/release_notes/release_notes-0.15.0.rst
@@ -18,6 +18,7 @@ Key Features and Updates
   * FIX-#4394: Fix issue with multiindex metadata desync (#4395)
   * FIX-#4425: Add parameters to groupby pct_change (#4429)
   * FIX-#4461: Fix S3 CSV data path (#4462)
+  * FIX-#4467: `drop_duplicates` no longer removes items based on index values (#4468)
 * Performance enhancements
   * FEAT-#4320: Add connectorx as an alternative engine for read_sql (#4346)
 * Benchmarking enhancements
@@ -61,3 +62,4 @@ Contributors
 @anmyachev
 @dchigarev
 @devin-petersohn
+@jrsacher
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index 603894d10db..727f3b3b6cf 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -1387,6 +1387,7 @@ def drop_duplicates(
         """
         inplace = validate_bool_kwarg(inplace, "inplace")
         subset = kwargs.get("subset", None)
+        ignore_index = kwargs.get("ignore_index", False)
         if subset is not None:
             if is_list_like(subset):
                 if not isinstance(subset, list):
@@ -1396,8 +1397,15 @@ def drop_duplicates(
             duplicates = self.duplicated(keep=keep, subset=subset)
         else:
             duplicates = self.duplicated(keep=keep)
-        indices = duplicates.values.nonzero()[0]
-        return self.drop(index=self.index[indices], inplace=inplace)
+
+        result = self[-duplicates]
+        if ignore_index:
+            result.index = pandas.RangeIndex(stop=len(result))
+        if inplace:
+            self._update_inplace(result)
+            return None
+        else:
+            return result
 
     def eq(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """


Code review comments:
diff_hunk: @@ -1396,8 +1397,15 @@ def drop_duplicates(
             duplicates = self.duplicated(keep=keep, subset=subset)
         else:
             duplicates = self.duplicated(keep=keep)
-        indices = duplicates.values.nonzero()[0]
-        return self.drop(index=self.index[indices], inplace=inplace)
+
+        result = self[-duplicates]
path: modin/pandas/base.py
text: @user1:
2 Nits: less vertical whitespace and use logical `not` rather than `negative`
```suggestion
        result = self[~duplicates]
```

diff_hunk: @@ -1396,8 +1397,15 @@ def drop_duplicates(
             duplicates = self.duplicated(keep=keep, subset=subset)
         else:
             duplicates = self.duplicated(keep=keep)
-        indices = duplicates.values.nonzero()[0]
-        return self.drop(index=self.index[indices], inplace=inplace)
+
+        result = self[-duplicates]
+        if ignore_index:
+            result.index = pandas.RangeIndex(stop=len(result))
+        if inplace:
+            self._update_inplace(result)
+            return None
path: modin/pandas/base.py
text: @user1:
```suggestion
            self._update_inplace(result)
```