Repository: dask/dask

Issue:
`from_array` no longer detects dask arrays
## Problem

Not sure if this needs to be supported, but I had some old code that accidentally did a `da.from_array(existing_dask_arr)`. This is obviously unnecessary, but the code ran fine. Starting with dask 2.2 this no longer seems to work and will result in `map_blocks` passing a dask array to the underlying function instead of a numpy array.

## Example Code

```
import numpy as np; import dask.array as da

a = np.zeros((5, 5), dtype=np.float)
a = da.from_array(a, chunks=2)
b = da.from_array(a, chunks=2)

def test(arr): 
    print(type(arr)) 
    return arr 

print("Running a")
a.map_blocks(test, dtype=a.dtype).compute() 
print("Running b")
b.map_blocks(test, dtype=b.dtype).compute() 
```

## Output

```
Running a
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
Running b
<class 'numpy.ndarray'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
<class 'dask.array.core.Array'>
```

## Expected

```
Running a
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
Running b
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
<class 'numpy.ndarray'>
```

Discussion:
djhoese: @jakirkham Let me know if you want a test. I don't see any `from_array` specific tests so not sure where the best place is.
jakirkham: > Let me know if you want a test.

That would be great. Could you please add one in [`dask/array/tests/test_array_core.py`]( https://github.com/dask/dask/blob/281fd90c1ebdda7b3fb6d067e632368376969b6d/dask/array/tests/test_array_core.py )?

> I don't see any `from_array` specific tests so not sure where the best place is.

Hmm...that's concerning. Could you please raise an issue for this?
djhoese: Ah I didn't realize the tests were separated that way. I was only looking in `dask/tests`. I'm sure there are `from_array` tests in `dask/array/tests/`.
jakirkham: Yeah, if you search for `def test_from_array_*`, there are a few different cases tested. They are a bit scattered though.
djhoese: So test was added but there is one failure in `dask/array/tests/test_optimization.py:test_gh3937`. The issue is that `asarray` is given a list of dask arrays. `asarray` looks like:

```
    if isinstance(a, Array):
        return a
    elif hasattr(a, "to_dask_array"):
        return a.to_dask_array()
    elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
        a = stack(a)
    elif not isinstance(getattr(a, "shape", None), Iterable):
        a = np.asarray(a)
    return from_array(a, chunks=a.shape, getitem=getter_inline, **kwargs)
```

This will hit the `stack` if statement which will convert everything to a dask array which it then passes to `from_array`.
jakirkham: Where is it calling `from_array` in `stack`?
jakirkham: Could we just move that `from_array` into the `else` case? I don't think any other case needs it.
djhoese: We could, but then `chunks`, `getitem`, and `**kwargs` are not being used.
jakirkham: They don't seem to be used by anything else anyways.
jrbourbeau: The following seems to pass tests for me locally

```diff
diff --git a/dask/array/core.py b/dask/array/core.py
index bb40549c..286dbcbc 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3580,7 +3580,7 @@ def asarray(a, **kwargs):
     elif hasattr(a, "to_dask_array"):
         return a.to_dask_array()
     elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-        a = stack(a)
+        return stack(a)
     elif not isinstance(getattr(a, "shape", None), Iterable):
         a = np.asarray(a)
     return from_array(a, chunks=a.shape, getitem=getter_inline, **kwargs)
```
jakirkham: Yeah I think the application of `from_array` to anything other than the result of `np.asarray` here is unintentional. It's probably best for us to go ahead and clean that up now to avoid similar mistakes in the future.
djhoese: Sounds good. Done. Thanks @jrbourbeau.

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 974f61640..286dbcbc6 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -2634,6 +2634,10 @@ def from_array(
     >>> a = da.from_array(x, chunks='100 MiB')  # doctest: +SKIP
     >>> a = da.from_array(x)  # doctest: +SKIP
     """
+    if isinstance(x, Array):
+        raise ValueError(
+            "Array is already a dask array. Use 'asarray' or " "'rechunk' instead."
+        )
     if isinstance(x, (list, tuple, memoryview) + np.ScalarType):
         x = np.array(x)
 
@@ -3576,7 +3580,7 @@ def asarray(a, **kwargs):
     elif hasattr(a, "to_dask_array"):
         return a.to_dask_array()
     elif isinstance(a, (list, tuple)) and any(isinstance(i, Array) for i in a):
-        a = stack(a)
+        return stack(a)
     elif not isinstance(getattr(a, "shape", None), Iterable):
         a = np.asarray(a)
     return from_array(a, chunks=a.shape, getitem=getter_inline, **kwargs)


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index b321eecf4..48143d181 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2325,6 +2325,13 @@ def test_from_array_copy():
     assert y.compute() is not y_c.compute()
 
 
+def test_from_array_dask_array():
+    x = np.array([[1, 2], [3, 4]])
+    dx = da.from_array(x, chunks=(1, 2))
+    with pytest.raises(ValueError):
+        da.from_array(dx)
+
+
 def test_asarray():
     assert_eq(da.asarray([1, 2, 3]), np.asarray([1, 2, 3]))