Repository: dask/dask
Pull request #11524 (2024-11-14 22:02:06)

Ensure that from_array creates a copy
- [ ] Closes #xxxx
- [ ] Tests added / passed
- [ ] Passes `pre-commit run --all-files`

Yikes, I ran into this when working on linear fusion. Testing that gave me a bunch of weird collision errors which are explainable by the missing copy here

davidhassell: Hello - is there an issue associated with this? I hope I haven't missed a discussion, but I'm not sure that this is a useful change. Please allow me to try and explain ...

The proposed change to `from_array` looks like it will _always_ copy an array_like input. But what if we don't want to do such a copy, for memory usage and speed concerns, or because we want to leverage in-place changes at some stage down the line? Is it not better leave any copying to the user (e.g. `da.from_array(a.copy())`), or we could provide a `copy=<bool>` keyword argument (defaults to False to preserve backwards compatability) to `from_array`, which will try to call the input's `copy` method, doing nothing if an `AttributeError` is raised? That said, I realise that it's cleaner, more general, and less typing to just pass in a copy and not have the keyword!

If the input is, e.g., a `list`, then it is already copied when it gets converted to a numpy array (which is fine), so we don't want to copy it again, in any event, I think. Some information in the docstring about copying might be all that's required? (I guess something along the lines of "the input is not copied, unless it is a list or tuple, which gets copied into a numpy array").

Thanks,
David
phofl: This is a pretty bad bug (but no issue associated)

> because we want to leverage in-place changes at some stage down the line?

This is a very strong -1 from me, anything that changes an array inplace that was passed into a Dask Collection is a very very strong anti pattern and **should** never be relied upon, especially since we say our collections are immutable.

>  for memory usage and speed concerns

A copy is pretty cheap and everything that you do afterwards will have a lot more overhead than just copying the array once. If one copy really screws up the performance of your computation, then Dask is probably not the best tool anyway. Additionally, you really shouldn't pass large arrays into from_array anyway, you should load the data from storage.
davidhassell: Hi. Thanks for responding.

If I have a 10Gb numpy array and I make a copy, on my 16GB laptop, then everything is killed, that can't be right! Isn't Dask exactly the right tool to use my 10GB data? A copy on that sized array on my laptop would take ~1 second. I have thousands of these, so we're really adding up the time.

I appreciate that the in-place change I described is unusual, and could be not recommended, but I can think of workflows that leverage: Instantiate two Dask arrays from the same numpy array, build up different graphs on each, have the option of changing the source in both prior to computation (i.e. before anything has been passed to the scheduler). I don't want to do that at the moment, but I don't think we should disallow it.

It's rightly up to Dask to choose its own copying strategy (fully endorsed by the Python Array API), but whatever its default is, I believe that the user should have options. 

I think that the proposed solution not only takes our options away, but also produces inconsistent results between different input types: Input types without a copy method, input types whose copy method is "shallow", and in input types whose copy method is "deep" would all give different outcomes - i.e. there's no guarantee that a copy is always going into the Dask graph. Better for the user to pass in a copy if they want to (which they can today), guided by better documentation (which could exist tomorrow)

Cheers,
David
davidhassell: (P.S. Would it be OK to open an issue for this?)
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     13 files  ± 0       13 suites  ±0   3h 21m 55s ⏱️ - 4m 15s
 13 290 tests + 1   12 228 ✅ + 1   1 062 💤 ±0  0 ❌ ±0 
138 687 runs  +11  119 696 ✅ +13  18 991 💤  - 2  0 ❌ ±0 

Results for commit d5395135. ± Comparison against base commit 98331028.

[test-results]:data:application/gzip;base64,H4sIAL2HNmcC/02MSw7DIAwFrxKx7iKGAnYvUyWQSKj5VARWUe9e8oMsZ549K+vd0C3sVYF4VGyJLhSy0TfBzdPGHEAmldZw7Jzqi99LNGY/4hyL/LjvJmvFs+sbNyRXPjvvZ38aH6cjjQr1yTkNpEhleaaRCC53S+98L5t5HF1IwKwUJEFIY/FJQmqwLXXWSC1MjZSCCnnTIrDfHyxtwHgaAQAA

phofl: > I appreciate that the in-place change I described is unusual, and could be not recommended, but I can think of workflows that leverage: Instantiate two Dask arrays from the same numpy array, build up different graphs on each, have the option of changing the source in both prior to computation (i.e. before anything has been passed to the scheduler). I don't want to do that at the moment, but I don't think we should disallow it.

That would screw with a bunch of different things since the content of your array goes into the graph generation and influences the keys. Again: You really should **not** do that under any circumstances

>If I have a 10Gb numpy array and I make a copy, on my 16GB laptop, then everything is killed, that can't be right! Isn't Dask exactly the right tool to use my 10GB data? A copy on that sized array on my laptop would take ~1 second. I have thousands of these, so we're really adding up the time.

Where are you getting these arrays from? You should use Dask directly to create them instead of putting them into the graph.

> It's rightly up to Dask to choose its own copying strategy (fully endorsed by the Python Array API), but whatever its default is, I believe that the user should have options.

I disagree, mutating the thing inplace has so many potential pitfalls that barely any user will be aware off.
davidhassell: Hi,

> Where are you getting these arrays from?You should use Dask directly to create them instead of putting them into the graph.

I don't follow this, I'm afraid. Dask is not the only softwre library out there, and my 10GB array could have been made by another piece of software that outputs (say) numpy arrays, I still want to use Dask to process that output (to perform a reduction, for instance).

 > I disagree, mutating the thing inplace has so many potential pitfalls that barely any user will be aware off.

So let the user be aware of these! Numpy has benefits and dangers from it's view/copy approach - they're well advertised and we're all good with that. I really don't see a problem with making it clear ... perhaps the issue that prompted this wouldn't have arisen if there was a description of the input array in the `from_array` docstring. Currently [there is none](https://github.com/dask/dask/blob/2024.11.2/dask/array/core.py#L3474), but how about:

```
        x : array_like
            The array to be converted to a dask array. `list` and `tuple` inputs are copied but 
            for all other input types, any in-place changes made subsequently to *x* will  
            also affect the returned dask array.
```

Could we please open an issue on this, as per the [development guidlines](https://docs.dask.org/en/latest/develop.html#issues)?

Many thanks,
David
fjetter: > Hi,
> 
> > Where are you getting these arrays from?You should use Dask directly to create them instead of putting them into the graph.
> 
> I don't follow this, I'm afraid. Dask is not the only softwre library out there, and my 10GB array could have been made by another piece of software that outputs (say) numpy arrays, I still want to use Dask to process that output (to perform a reduction, for instance).
> 

This is possible of course and this is also what this API is for. However, you will achieve better performance if you let dask do the generation/data loading. The data copy that is being introduced here is one example why that is better. There are other reasons that are more complex to explain (TLDR if we use a specialized API we can make certain assumption we can't make for the generic `from_array`)

> > I disagree, mutating the thing inplace has so many potential pitfalls that barely any user will be aware off.
> 
> So let the user be aware of these! Numpy has benefits and dangers from it's view/copy approach - they're well advertised and we're all good with that. I really don't see a problem with making it clear ... perhaps the issue that prompted this wouldn't have arisen if there was a description of the input array in the `from_array` docstring. Currently [there is none](https://github.com/dask/dask/blob/2024.11.2/dask/array/core.py#L3474), but how about:

I think there is a conceptual misunderstanding. In the numpy case, these "dangers" or "pitfalls" are mostly there because the API is sometimes not clear and it is easy to make a mistake. Experienced users can still navigate and leverage this.

Dask operates differently. We don't have the concept/distinction of views vs copies. There is no way for users to guarantee that inplace mutations are safe. This mostly comes from us executing things in parallel. The fact that intermediate results are not mutated is a core assumption the dask architecture builds on.


davidhassell: Hello,

> Dask operates differently. We don't have the concept/distinction of views vs copies. There is no way for users to guarantee that inplace mutations are safe. This mostly comes from us executing things in parallel. The fact that intermediate results are not mutated is a core assumption the dask architecture builds on.

OK, I understand that, but it is not enforced anywhere else in Dask, so why should it be in `from_array`? (notwithstanding the out-of-memory and speed issues I have already highlighted). Consider the following, which quite happily embeds a non-copy of a numpy array in a dask graph:

```python
In [25]: import dask.array as da

In [26]: import numpy as np

In [27]: dx = da.arange(3)

In [28]: x = np.array([4, 5, 6])

In [29]: dy = dx + x

In [30]: dy.compute()
Out[30]: array([4, 6, 8])

In [31]: x[0] = 999

In [32]: dy.compute()
array([999,   6,   8])
```
Dask can't always account for what we choose to put inside a graph. Rather it should it make it clear to users that what they put in (by whatever means) is not a copy, and they should put copies in if they need to.

By automatically copying, you're going to prevent Dask being used tomorrow in cases for which it works perfectly well today, and there won't be a nice error message, rather the session will just get killed when it runs out of memory.

Thanks,
David
davidhassell: P.P.S. surely it's time for an issue to be opened on this :)
phofl: > Consider the following, which quite happily embeds a non-copy of a numpy array in a dask graph:

This is incorrect, we are calling from_array under the hood
davidhassell: > This is incorrect, we are calling from_array under the hood

Interesting - and I get point. _If_ you were to make `from_array` copy numpy arrays (and variously some other things), you can propagate this copying through all operations. Which makes me even more worried :) If `x` is automatically copied, how could I do `dy = dx + x` for a large enough `x`?

> However, you will achieve better performance if you let dask do the generation/data loading

Only if dask is capable of the generation/data loading - it can't do everything :)

If no one else wants to start an issue on this, I shall ...

Patch:
diff --git a/dask/array/core.py b/dask/array/core.py
index 83d877514..10736af6f 100644
--- a/dask/array/core.py
+++ b/dask/array/core.py
@@ -3622,6 +3622,9 @@ def from_array(
     if isinstance(x, (list, tuple, memoryview) + np.ScalarType):
         x = np.array(x)
 
+    if is_arraylike(x) and hasattr(x, "copy"):
+        x = x.copy()
+
     if asarray is None:
         asarray = not hasattr(x, "__array_function__")
 


Test patch:
diff --git a/dask/array/tests/test_array_core.py b/dask/array/tests/test_array_core.py
index f4e886aa5..5a49c19be 100644
--- a/dask/array/tests/test_array_core.py
+++ b/dask/array/tests/test_array_core.py
@@ -2579,7 +2579,8 @@ def test_from_array_ndarray_onechunk(x):
     dx = da.from_array(x, chunks=-1)
     assert_eq(x, dx)
     assert len(dx.dask) == 1
-    assert dx.dask[(dx.name,) + (0,) * dx.ndim] is x
+    assert not dx.dask[(dx.name,) + (0,) * dx.ndim] is x
+    assert_eq(dx.dask[(dx.name,) + (0,) * dx.ndim], x)
 
 
 def test_from_array_ndarray_getitem():
@@ -2706,7 +2707,8 @@ def test_from_array_inline():
 
     a = np.array([1, 2, 3]).view(MyArray)
     dsk = dict(da.from_array(a, name="my-array", inline_array=False).dask)
-    assert dsk["original-my-array"] is a
+    assert not dsk["original-my-array"] is a
+    assert_eq(dsk["original-my-array"], a)
 
     dsk = dict(da.from_array(a, name="my-array", inline_array=True).dask)
     assert "original-my-array" not in dsk
@@ -4379,7 +4381,7 @@ def test_blockwise_with_numpy_arrays():
     assert_eq(x + y, x + x)
 
     s = da.sum(x)
-    assert any(x is v for v in s.dask.values())
+    assert any(isinstance(v, np.ndarray) for v in s.dask.values())
 
 
 @pytest.mark.parametrize("chunks", (100, 6))
@@ -5383,6 +5385,15 @@ def test_to_backend():
             x.to_backend("missing")
 
 
+def test_from_array_copies():
+    x = np.arange(60).reshape((6, 10))
+    original_array = x.copy()
+    chunks = (2, 3)
+    dx = da.from_array(x, chunks=chunks)
+    x[2:4, x[0] > 3] = -5
+    assert_eq(original_array, dx)
+
+
 def test_load_store_chunk():
     actual = np.array([0, 0, 0, 0, 0, 0])
     load_store_chunk(
diff --git a/dask/array/tests/test_optimization.py b/dask/array/tests/test_optimization.py
index 7ddb4ecc1..fc9bd5b79 100644
--- a/dask/array/tests/test_optimization.py
+++ b/dask/array/tests/test_optimization.py
@@ -392,7 +392,6 @@ def test_fuse_getter_with_asarray(chunks):
     y = da.ones(10, chunks=chunks)
     z = x + y
     dsk = z.__dask_optimize__(z.dask, z.__dask_keys__())
-    assert any(v is x for v in dsk.values())
     for v in dsk.values():
         s = str(v)
         assert s.count("getitem") + s.count("getter") <= 1