Repository: dask/dask
Pull request: Generalize `from_dask_array`
Fixes: https://github.com/rapidsai/cudf/issues/5263

- [x] Tests added / passed
- [x] Passes `black dask` / `flake8 dask`

cc: @kkraus14 @quasiben 

Problem statement:
[BUG] Adding a random int column in dask cudf dataframe results in a string BINARY_OP error
**Describe the bug**
I want to add a new column of random integers to a dask cudf dataframe. A strange BINARY_OP error occured.

**Steps/Code to reproduce bug**
```
import cudf as gd
import dask_cudf
df = gd.DataFrame(list(range(100)))
ddf = dask_cudf.from_cudf(df,npartitions=4)
ddf['fold'] = 0
ddf['fold'] = ddf['fold'].map_partitions(lambda cudf_df: cp.random.randint(0,4,len(cudf_df)))
```

Error message:
```
---------------------------------------------------------------------------
RuntimeError                              Traceback (most recent call last)
<ipython-input-11-ce13f9bab713> in <module>
      2 ddf = dask_cudf.from_cudf(df,npartitions=4)
      3 ddf['fold'] = 0
----> 4 ddf['fold'] = ddf['fold'].map_partitions(lambda cudf_df: cp.random.randint(0,4,len(cudf_df)))

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/dask/dataframe/core.py in __setitem__(self, key, value)
   3414             df = self.assign(**{k: value for k in key})
   3415         else:
-> 3416             df = self.assign(**{key: value})
   3417 
   3418         self.dask = df.dask

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/dask/dataframe/core.py in assign(self, **kwargs)
   3679 
   3680         # Figure out columns of the output
-> 3681         df2 = self._meta_nonempty.assign(**_extract_meta(kwargs, nonempty=True))
   3682         return elemwise(methods.assign, self, *pairs, meta=df2)
   3683 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/dataframe.py in assign(self, **kwargs)
    729         new = self.copy()
    730         for k, v in kwargs.items():
--> 731             new[k] = v
    732         return new
    733 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/contextlib.py in inner(*args, **kwds)
     72         def inner(*args, **kwds):
     73             with self._recreate_cm():
---> 74                 return func(*args, **kwds)
     75         return inner
     76 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/dataframe.py in __setitem__(self, arg, value)
    553                             how="right",
    554                             sort=False,
--> 555                             allow_non_unique=True,
    556                         )
    557                     if is_scalar(value):

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/series.py in _align_to_index(self, index, how, sort, allow_non_unique)
   3247 
   3248         index = as_index(index)
-> 3249         if self.index.equals(index):
   3250             return self
   3251         if not allow_non_unique:

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/contextlib.py in inner(*args, **kwds)
     72         def inner(*args, **kwds):
     73             with self._recreate_cm():
---> 74                 return func(*args, **kwds)
     75         return inner
     76 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/index.py in equals(self, other)
    303             return bool(val)
    304         else:
--> 305             result = self == other
    306             if isinstance(result, bool):
    307                 return result

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/index.py in __eq__(self, other)
    273 
    274     def __eq__(self, other):
--> 275         return self._apply_op("__eq__", other)
    276 
    277     def __ne__(self, other):

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/index.py in _apply_op(self, fn, other)
    215         op = getattr(idx_series, fn)
    216         if other is not None:
--> 217             return as_index(op(other))
    218         else:
    219             return as_index(op())

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/series.py in __eq__(self, other)
    979 
    980     def __eq__(self, other):
--> 981         return self._binaryop(other, "eq")
    982 
    983     def equals(self, other):

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/contextlib.py in inner(*args, **kwds)
     72         def inner(*args, **kwds):
     73             with self._recreate_cm():
---> 74                 return func(*args, **kwds)
     75         return inner
     76 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/series.py in _binaryop(self, other, fn, fill_value, reflect)
    635                     rhs = rhs.fillna(fill_value)
    636 
--> 637         outcol = lhs._column.binary_operator(fn, rhs, reflect=reflect)
    638         result = lhs._copy_construct(data=outcol, name=result_name)
    639         return result

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/column/string.py in binary_operator(self, op, rhs, reflect)
   2172             return lhs.str().cat(others=rhs)
   2173         elif op in ("eq", "ne", "gt", "lt", "ge", "le"):
-> 2174             return _string_column_binop(self, rhs, op=op, out_dtype="bool")
   2175         else:
   2176             msg = "{!r} operator not supported between {} and {}"

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/contextlib.py in inner(*args, **kwds)
     72         def inner(*args, **kwds):
     73             with self._recreate_cm():
---> 74                 return func(*args, **kwds)
     75         return inner
     76 

/raid/data/ml/anaconda3/envs/rapids/lib/python3.7/site-packages/cudf/core/column/string.py in _string_column_binop(lhs, rhs, op, out_dtype)
   2212 @annotate("BINARY_OP", color="orange", domain="cudf_python")
   2213 def _string_column_binop(lhs, rhs, op, out_dtype):
-> 2214     out = libcudf.binaryop.binaryop(lhs=lhs, rhs=rhs, op=op, dtype=out_dtype)
   2215     return out
   2216 

cudf/_lib/binaryop.pyx in cudf._lib.binaryop.binaryop()

cudf/_lib/binaryop.pyx in cudf._lib.binaryop.binaryop_v_v()

RuntimeError: cuDF failure at: /conda/conda-bld/libcudf_1587234413731/work/cpp/src/binaryop/binaryop.cpp:322: Invalid/Unsupported lhs datatype
```

**Environment overview (please complete the following information)**
 - Environment location: Bare-metal
 - Method of cuDF install: conda

**Environment details**
cuda 10.2
cudf 0.14.0a+3339.g57ddd5e.dirty
dask_cudf 0.14.0a+3339.g57ddd5e.dirty

**Additional context**
Add any other context about the problem here.

Hints from issue discussion:

Commit message: typos

Patch to review:
diff --git a/dask/array/core.py b/dask/array/core.py
index ef897009182..7b55c1dea28 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -1430,7 +1430,7 @@ def to_hdf5(self, filename, datapath, **kwargs):
         """
         return to_hdf5(filename, datapath, self, **kwargs)
 
-    def to_dask_dataframe(self, columns=None, index=None):
+    def to_dask_dataframe(self, columns=None, index=None, meta_df=None):
         """ Convert dask Array to dask Dataframe
 
         Parameters
@@ -1455,7 +1455,7 @@ def to_dask_dataframe(self, columns=None, index=None):
         """
         from ..dataframe import from_dask_array
 
-        return from_dask_array(self, columns=columns, index=index)
+        return from_dask_array(self, columns=columns, index=index, meta_df=meta_df)
 
     def __bool__(self):
         if self.size > 1:
diff --git a/dask/dataframe/core.py b/dask/dataframe/core.py
index 4316967d230..db67655a924 100644
--- a/dask/dataframe/core.py
+++ b/dask/dataframe/core.py
@@ -3781,7 +3781,7 @@ def assign(self, **kwargs):
                             v.npartitions, self.npartitions
                         )
                     )
-                kwargs[k] = from_dask_array(v, index=self.index)
+                kwargs[k] = from_dask_array(v, index=self.index, meta_df=self._meta)
 
         pairs = list(sum(kwargs.items(), ()))
 
diff --git a/dask/dataframe/io/io.py b/dask/dataframe/io/io.py
index 6903e035d24..16b029422d1 100644
--- a/dask/dataframe/io/io.py
+++ b/dask/dataframe/io/io.py
@@ -13,14 +13,14 @@
 
 from ..core import DataFrame, Series, Index, new_dd_object, has_parallel_type
 from ..shuffle import set_partition
-from ..utils import insert_meta_param_description, check_meta, make_meta
+from ..utils import insert_meta_param_description, check_meta, make_meta, is_series_like
 
 from ...utils import M, ensure_dict
 
 lock = Lock()
 
 
-def _meta_from_array(x, columns=None, index=None):
+def _meta_from_array(x, columns=None, index=None, meta_df=None):
     """ Create empty pd.DataFrame or pd.Series which has correct dtype """
 
     if x.ndim > 2:
@@ -34,6 +34,8 @@ def _meta_from_array(x, columns=None, index=None):
             raise ValueError("'index' must be an instance of dask.dataframe.Index")
         index = index._meta
 
+    if meta_df is None:
+        meta_df = pd.DataFrame()
     if getattr(x.dtype, "names", None) is not None:
         # record array has named columns
         if columns is None:
@@ -47,9 +49,11 @@ def _meta_from_array(x, columns=None, index=None):
         dtypes = [fields[n][0] if n in fields else "f8" for n in columns]
     elif x.ndim == 1:
         if np.isscalar(columns) or columns is None:
-            return pd.Series([], name=columns, dtype=x.dtype, index=index)
+            return meta_df._constructor_sliced(
+                [], name=columns, dtype=x.dtype, index=index
+            )
         elif len(columns) == 1:
-            return pd.DataFrame(
+            return meta_df._constructor(
                 np.array([], dtype=x.dtype), columns=columns, index=index
             )
         raise ValueError(
@@ -69,11 +73,11 @@ def _meta_from_array(x, columns=None, index=None):
         dtypes = [x.dtype] * len(columns)
 
     data = {c: np.array([], dtype=dt) for (c, dt) in zip(columns, dtypes)}
-    return pd.DataFrame(data, columns=columns, index=index)
+    return meta_df._constructor(data, columns=columns, index=index)
 
 
-def from_array(x, chunksize=50000, columns=None):
-    """ Read any slicable array into a Dask Dataframe
+def from_array(x, chunksize=50000, columns=None, meta_df=None):
+    """ Read any sliceable array into a Dask Dataframe
 
     Uses getitem syntax to pull slices out of the array.  The array need not be
     a NumPy array but must support slicing syntax
@@ -90,9 +94,9 @@ def from_array(x, chunksize=50000, columns=None):
 
     """
     if isinstance(x, da.Array):
-        return from_dask_array(x, columns=columns)
+        return from_dask_array(x, columns=columns, meta_df=meta_df)
 
-    meta = _meta_from_array(x, columns)
+    meta = _meta_from_array(x, columns, meta_df=meta_df)
 
     divisions = tuple(range(0, len(x), chunksize))
     divisions = divisions + (len(x) - 1,)
@@ -102,10 +106,10 @@ def from_array(x, chunksize=50000, columns=None):
     dsk = {}
     for i in range(0, int(ceil(len(x) / chunksize))):
         data = (getitem, x, slice(i * chunksize, (i + 1) * chunksize))
-        if isinstance(meta, pd.Series):
-            dsk[name, i] = (pd.Series, data, None, meta.dtype, meta.name)
+        if is_series_like(meta):
+            dsk[name, i] = (type(meta), data, None, meta.dtype, meta.name)
         else:
-            dsk[name, i] = (pd.DataFrame, data, None, meta.columns)
+            dsk[name, i] = (type(meta), data, None, meta.columns)
     return new_dd_object(dsk, name, meta, divisions)
 
 
@@ -372,7 +376,7 @@ def dataframe_from_ctable(x, slc, columns=None, categories=None, lock=lock):
     return result
 
 
-def from_dask_array(x, columns=None, index=None):
+def from_dask_array(x, columns=None, index=None, meta_df=None):
     """ Create a Dask DataFrame from a Dask Array.
 
     Converts a 2d array into a DataFrame and a 1d array into a Series.
@@ -388,7 +392,7 @@ def from_dask_array(x, columns=None, index=None):
         The default output index depends on whether `x` has any unknown
         chunks. If there are any unknown chunks, the output has ``None``
         for all the divisions (one per chunk). If all the chunks are known,
-        a default index with known divsions is created.
+        a default index with known divisions is created.
 
         Specifying `index` can be useful if you're conforming a Dask Array
         to an existing dask Series or DataFrame, and you would like the
@@ -413,7 +417,7 @@ def from_dask_array(x, columns=None, index=None):
     dask.dataframe._Frame.values: Reverse conversion
     dask.dataframe._Frame.to_records: Reverse conversion
     """
-    meta = _meta_from_array(x, columns, index)
+    meta = _meta_from_array(x, columns, index, meta_df=meta_df)
 
     if x.ndim == 2 and len(x.chunks[1]) > 1:
         x = x.rechunk({1: x.shape[1]})
@@ -450,10 +454,10 @@ def from_dask_array(x, columns=None, index=None):
     for i, (chunk, ind) in enumerate(zip(x.__dask_keys__(), index)):
         if x.ndim == 2:
             chunk = chunk[0]
-        if isinstance(meta, pd.Series):
-            dsk[name, i] = (pd.Series, chunk, ind, x.dtype, meta.name)
+        if is_series_like(meta):
+            dsk[name, i] = (type(meta), chunk, ind, x.dtype, meta.name)
         else:
-            dsk[name, i] = (pd.DataFrame, chunk, ind, meta.columns)
+            dsk[name, i] = (type(meta), chunk, ind, meta.columns)
 
     to_merge.extend([ensure_dict(x.dask), dsk])
     return new_dd_object(merge(*to_merge), name, meta, divisions)


Code review comments:
diff_hunk: @@ -1430,7 +1430,7 @@ def to_hdf5(self, filename, datapath, **kwargs):
         """
         return to_hdf5(filename, datapath, self, **kwargs)
 
-    def to_dask_dataframe(self, columns=None, index=None):
+    def to_dask_dataframe(self, columns=None, index=None, meta_df=None):
path: dask/array/core.py
text: @user1:
I don't think that this name immediately explains the concept it's capturing. Can you think of alternatives?

Perhaps `dataframe_class`? Or does that not convey that this is the concrete dataframe that make up the partitions?

You'll need to add this parameter to the docstring.

@user1:
Does this need to be instantiated, or just the class?

@author:
Updated to `target_dataframe`, we'll need an instantiated object to actually access the `_constructor` & `_constructor_sliced` properties.

@user2:
I don't find `target_dataframe` a meaningful kwarg name - my first thought from that name was this was the class of the chunk type.

Since this is an instantiated object that will be used as the metadata for the target, something with `meta` would be more consistent with the rest of the api. I'd probably go with `meta` here (I realize this contradicts Tom's suggestion above, sorry Tom :/). Since it's an instantiated object used as metadata it seems fitting to keep the existing verbage - many of our other conversion routines to dataframes also have this kwarg.

@author:
@user1 are you also okay with `meta`?
Or one more option that comes to my mind is `target_meta`/`meta_target`?

@user1:
My only hesitation with using `meta` is that it overloads (or expands) the definition of meta a bit. Right now it's used to identify the structure of the dataset: columns, index(es), and their dtypes. If we also allow it to mean the concrete container type, then what does that imply? If I have something like

```python
df = dd.from_pandas(pd.DataFrame({"A": [1, 2, 3]}), 2)
df.map_partitions(lambda x: x + 1, meta=cudf.DataFrame({"A": int}))`
```

would you expect that to copy from host to device memory by calling `cudf.DataFrame._constructor` on the result of the `lambda`? According to `meta`, the concrete type should be a `cudf.DataFrame`. But the function doesn't do anything with host -> device, so it would need to be done implicitly.

And to be clear: I'm not against defining that behavior, but we should do it intentionally.

@user2:
> If we also allow it to mean the concrete container type, then what does that imply?

Don't we already define that behavior? If I'm using `cudf` with dask dataframe, isn't `meta` a cudf object?

> would you expect that to copy from host to device memory by calling cudf.DataFrame._constructor on the result of the lambda?

Currently we trust the user that `meta` is correct, and sometimes warn/error if that isn't true. In this case we'd trust the user that the output is a `cudf` type, but wouldn't do the conversion ourselves. I believe this matches current conventions, and is the least-surprising choice. `meta` is there to hint to dask-dataframe the structure of the underlying dataframe when inference may not be able to handle it - it's a type hint, not a type conversion.

@user1:
> Don't we already define that behavior? If I'm using cudf with dask dataframe, isn't meta a cudf object?

I'm not sure.

Good point about us trusting `meta`. That's consistent with not implicitly calling `meta._constructor` on the output, since we're trusting that it's correct.

So I'm probably fine with using `meta` here.

@author:
Okay, updated the param name to be `meta`.