Repository: dask/dask
Pull request: Fix comparison against pd.Series
fixes #4596 

- [x] Tests added / passed
- [x] Passes `flake8 dask`


Problem statement:
dask.dataframe.fillna fails with "ValueError: cannot reindex from a duplicate axis"
Starting with Dask 1.1.0, `dask.dataframe.fillna` fails when trying to fill based on a series from the same dataframe if the index is not unique. The following example works under Dask 1.0.0, but fails with more recent versions:

```
import dask
import dask.dataframe as dd 
import numpy as np 
import pandas as pd


@dask.delayed 
def f(): 
    return pd.DataFrame(dict(a=[1.0], b=[np.NaN])) 

ddf=dd.from_delayed([f(), f()], meta=dict(a=float, b=float))  
ddf.b=ddf.b.fillna(ddf.a)
ddf.compute()
```

Hints from issue discussion:
Thanks. For reference can you attach the stacktrace?

Did you also change pandas versions when upgrading from dask 1.0 to 1.1?
```Traceback (most recent call last):
  File "test.py", line 13, in <module>
    ddf.compute()
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/base.py", line 156, in compute
    (result,) = compute(self, traverse=False, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/base.py", line 398, in compute
    results = schedule(dsk, keys, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/threaded.py", line 76, in get
    pack_exception=pack_exception, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/local.py", line 459, in get_async
    raise_exception(exc, tb)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/compatibility.py", line 112, in reraise
    raise exc
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/local.py", line 230, in execute_task
    result = _execute_task(task, data)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/core.py", line 119, in _execute_task
    return func(*args2)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/optimization.py", line 942, in __call__
    dict(zip(self.inkeys, args)))
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/core.py", line 149, in get
    result = _execute_task(task, cache)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/core.py", line 119, in _execute_task
    return func(*args2)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/compatibility.py", line 93, in apply
    return func(*args, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/dataframe/core.py", line 3778, in apply_and_enforce
    df = func(*args, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/dask/utils.py", line 697, in __call__
    return getattr(obj, self.method)(*args, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/series.py", line 3834, in fillna
    **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/generic.py", line 6104, in fillna
    downcast=downcast)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/internals/managers.py", line 525, in fillna
    return self.apply('fillna', **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/internals/managers.py", line 393, in apply
    copy=align_copy)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/series.py", line 3734, in reindex
    return super(Series, self).reindex(index=index, **kwargs)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/generic.py", line 4356, in reindex
    fill_value, copy).__finalize__(self)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/generic.py", line 4374, in _reindex_axes
    copy=copy, allow_dups=False)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/generic.py", line 4490, in _reindex_with_indexers
    copy=copy)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/internals/managers.py", line 1224, in reindex_indexer
    self.axes[axis]._can_reindex(indexer)
  File "/home/amerkel/venv/lib/python3.6/site-packages/pandas/core/indexes/base.py", line 3086, in _can_reindex
    raise ValueError("cannot reindex from a duplicate axis")
ValueError: cannot reindex from a duplicate axis
```
Pandas version was not changed. The trace above is with pandas 0.24.1, but the same error occurs with 0.23 and 0.22.
Thanks for checking. Any interest in debugging why we seem to take a
different code path? Or perhaps git bisecting between 1.0.0 and 1.1.10?

On Fri, Mar 15, 2019 at 3:18 AM amerkel2 <notifications@github.com> wrote:

> Pandas version was not changed. The trace above is with pandas 0.24.1, but
> the same error occurs with 0.23 and 0.22.
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/issues/4596#issuecomment-473197153>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/ABQHIjziVBwJFSJOeiAzVErY_21be6x3ks5vW1dPgaJpZM4b0Z0t>
> .
>

Sure. What I've been able to find out so far is that the ddf has a non-unique index both with Dask 1.0 and 1.1. So before the fillna, it looks like 
```
     a   b
0  1.0 NaN
0  1.0 NaN
```
So my first question is: is this how it's supposed to be?
> So my first question is: is this how it's supposed to be?

You're asking about the input, right? If so, yes, since each of your input data frames have 0 for the single index label.
> > So my first question is: is this how it's supposed to be?
> 
> You're asking about the input, right? If so, yes, since each of your input data frames have 0 for the single index label.

OK, thx.
Bisecting revealed that https://github.com/dask/dask/commit/5793d9742ebc7e87677a0e2c631649e2c293088d introduced the changed behavior.
Thanks for bisecting.

I notice that `dd.utils.is_series_like` returns True for `dd.Index` types.

```python
In [9]: a = dd.from_pandas(pd.DataFrame({"a": [1, 2]}), 2)

In [10]: dd.utils.is_series_like(a.index)
Out[10]: True
```

The following fixes that I think

```diff
diff --git a/dask/dataframe/utils.py b/dask/dataframe/utils.py
index 05ff2262..ed2237b1 100644
--- a/dask/dataframe/utils.py
+++ b/dask/dataframe/utils.py
@@ -504,13 +504,16 @@ def is_dataframe_like(df):
 
 def is_series_like(s):
     """ Looks like a Pandas Series """
-    return set(dir(s)) > {'name', 'dtype', 'groupby', 'head'} and not isinstance(s, type)
+    attrs = set(dir(s))
+    return (attrs > {'name', 'dtype', 'groupby', 'head'} and
+            'to_series' not in attrs and
+            not isinstance(s, type))
 
 
 def is_index_like(s):
     """ Looks like a Pandas Index """
     attrs = set(dir(s))
-    return attrs > {'name', 'dtype'} and 'head' not in attrs and not isinstance(s, type)
+    return attrs > {'name', 'dtype', 'to_series'} and not isinstance(s, type)
 
 
 def check_meta(x, meta, funcname=None, numeric_equal=True):
```

unfortunately, that doesn't fix your original issue. So we may be on to something, or at a dead end :) Won't have any time to look further right now.
OK, thanks anyway! I'll try to investigate further.

Commit message: Fix comparison against pd.Series.

Patch to review:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 609f7c9cc95..1102153d69f 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -1024,7 +1024,7 @@ def fillna(self, value=None, method=None, limit=None, axis=None):
             # Control whether or not dask's partition alignment happens.
             # We don't want for a pandas Series.
             # We do want it for a dask Series
-            if is_series_like(value):
+            if isinstance(value, pd.Series):
                 args = ()
                 kwargs = {'value': value}
             else:
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index f0136f5a5a0..05cdeeb2a3f 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -1482,6 +1482,16 @@ def test_fillna():
               ddf.fillna(method='pad', limit=3))
 
 
+def test_fillna_duplicate_index():
+    @dask.delayed
+    def f():
+        return pd.DataFrame(dict(a=[1.0], b=[np.NaN]))
+
+    ddf = dd.from_delayed([f(), f()], meta=dict(a=float, b=float))
+    ddf.b = ddf.b.fillna(ddf.a)
+    ddf.compute()
+
+
 def test_fillna_multi_dataframe():
     df = tm.makeMissingDataframe(0.8, 42)
     ddf = dd.from_pandas(df, npartitions=5, sort=False)


Code review comments:
diff_hunk: @@ -1024,7 +1024,7 @@ def fillna(self, value=None, method=None, limit=None, axis=None):
             # Control whether or not dask's partition alignment happens.
             # We don't want for a pandas Series.
             # We do want it for a dask Series
-            if is_series_like(value):
+            if isinstance(value, pd.Series):
path: dask/dataframe/core.py
text: @user1:
```suggestion
            if is_series_like(value) and not is_dask_collection(value):
```

How about this then?

@author:
Thanks, this works!