Repository: dask/dask
Pull request: Fix map(series) for unsorted base series index
Found a bug where tests were not actually catching this, fixes https://github.com/dask/dask/issues/5458

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask`


Problem statement:
Bug in MapSeries for unsorted index base series
Tests were not actually testing the intended base-series index being unsorted case.

Code to show error in current dask:
```python
import pandas as pd
import dask.dataframe as dd

ser1 = dd.from_pandas(pd.Series([0,1,2,3], index=["a", "c", "b", "d"]), npartitions=1)
ser1 = ser1.map_partitions(lambda x: x.sort_values())
ser2 = dd.from_pandas(pd.Series([0,4,8,12], index=[0,1,2,3]), npartitions=1)
print(ser1.compute().map(ser2.compute()))
print(ser1.map(ser2).compute())
```

Errored output:
```
a     0
c     4
b     8
d    12
dtype: int64
a     0
c     8
b     4
d    12
dtype: int64
```

Hints from issue discussion:

Commit message: Update tests to actually catch the not sorted case and fix.

Patch to review:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 97da856a617..8e727e400f8 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -6098,7 +6098,7 @@ def mapseries(base_chunk, concat_map):
 
 def mapseries_combine(index, concat_result):
     final_series = concat_result.sort_index()
-    final_series.index = index
+    final_series = pd.Series(index, index=index).map(final_series)
     return final_series
 
 
diff --git a/dask/dataframe/tests/test_dataframe.py b/dask/dataframe/tests/test_dataframe.py
index 8a73df559f9..58a730302fc 100644
--- a/dask/dataframe/tests/test_dataframe.py
+++ b/dask/dataframe/tests/test_dataframe.py
@@ -4046,8 +4046,8 @@ def test_series_map(base_npart, map_npart, sorted_index, sorted_map_index):
         np.random.shuffle(map_index)
         mapper.index = map_index
     expected = base.map(mapper)
-    dask_base = dd.from_pandas(base, npartitions=base_npart)
-    dask_map = dd.from_pandas(mapper, npartitions=map_npart)
+    dask_base = dd.from_pandas(base, npartitions=base_npart, sort=False)
+    dask_map = dd.from_pandas(mapper, npartitions=map_npart, sort=False)
     result = dask_base.map(dask_map)
     dd.utils.assert_eq(expected, result)
 


Code review comments:
diff_hunk: @@ -6098,7 +6098,7 @@ def mapseries(base_chunk, concat_map):
 
 def mapseries_combine(index, concat_result):
     final_series = concat_result.sort_index()
-    final_series.index = index
+    final_series = pd.Series(index, index=index).map(final_series)
path: dask/dataframe/core.py
text: @user1:
We've been trying to avoid using `pd.Series` explicitly, in order to improve support for other dataframe libraries that are similar enough to pandas (notably RAPIDS cudf).  Can you think of a way that would produce a Series object that don't explicitly use the pandas module?

@author:
I'll work on this a bit and see what I can do. First few attempts somehow ended up passing tests, but functionally gave wrong results, so I'm gonna update some tests too to make this cleaner. 

@author:
Actually, just found that `index.to_series()` is nearly identical to `pd.Series(index, index=index)`, so just use this `to_series()` method and everything works!