Repository: dask/dask
Pull request: Implement complete dask.array.tile function
This implements a complete `dask.array.tile` function, based on this https://github.com/dask/dask/issues/2447#issuecomment-500614803.

For example, 
```python
>>> from dask import array as da
>>> a = da.asarray([1, 2, 3])
>>> da.tile(a, [2, 1]).compute()
array([[1, 2, 3],
       [1, 2, 3]])
```
now works as expected.

- [ ] Tests added / passed
- [ ] Passes `black dask` / `flake8 dask`

Closes https://github.com/dask/dask/issues/2447

Problem statement:
New da.tile implementation
Someone privately sent me this da.tile implementation.  

```python
def tile(A, reps):
    try:
        tup = tuple(reps)
    except TypeError:
        tup = (reps,)
    d = len(tup)
    if (d < A.ndim):
        tup = (1,)*(A.ndim-d) + tup
    nnd = len(tup) - A.ndim # Number of new dimensions
    final_shape = tup[:nnd]+tuple(i*di for i,di in zip(tup[nnd:],A.shape))
    if 0 in final_shape:
        XXXXX # how to make empty array?
    tup = tup[nnd:]
    # Don't tile along dimensions that can be broadcasted
    tup = tuple(1 if l == 1 else r for l, r in zip(A.shape, tup))
    if all(i==1 for i in tup):
        return da.broadcast_to(A, final_shape)
    name = 'tile-'+dask.base.tokenize('tile', A.name, tup)
    dsk = dict(zip(itertools.product([name], *tuple(range(r*len(c)) for r,c in zip(tup, A.chunks))),
                   itertools.product([A.name], *tuple(r*range(len(c)) for r,c in zip(tup, A.chunks)))))
    ary =  da.Array(sharedict.merge(A.dask, dsk), name, tuple(r*c for r,c in zip(tup, A.chunks)), A.dtype)
    return da.broadcast_to(ary, final_shape)
```

I haven't yet gone through it.  If anyone feels like an easy contribution this could be checked and tests could be added

Hints from issue discussion:
What does this offer over the existing `tile` implementation?
This version using broadcasting (with `broadcast_to`) instead of copying (with `np.tile`) when possible. Presumably this should be significantly faster when tiling dimensions of length 1.
@mrocklin - I wanted to checkout + test this implementation for a PR.
In which file is the current `da.tile` function implementation?
```
mrocklin@carbon:~/workspace/dask$ git grep "def tile"
dask/array/creation.py:def tile(A, reps):
```
Sorry for the delay, but I checked out this implementation, but it appears to be failing some of the existing tests, due to a value error in the normalize_chunks function.

I've included the terminal output of one specific test here:
```
_________________________ test_tile[0-shape0-chunks0] _________________________

shape = (10,), chunks = (1,), reps = 0

    @pytest.mark.parametrize('shape, chunks', [
        ((10,), (1,)),
        ((10, 11, 13), (4, 5, 3)),
    ])
    @pytest.mark.parametrize('reps', [0, 1, 2, 3, 5])
    def test_tile(shape, chunks, reps):
        x = np.random.random(shape)
        d = da.from_array(x, chunks=chunks)
        print("shape = {}".format(shape))
        print("chunks = {}".format(chunks))
        print("reps = {}".format(reps))
        print("x = {}".format(x))
        print("d = {}".format(d))
>       assert_eq(np.tile(x, reps), da.tile(d, reps))

dask\array\tests\test_creation.py:538: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
dask\array\creation.py:769: in tile
    ary =  Array(HighLevelGraph.merge(A.dask, dsk), name, tuple(r*c for r,c in zip(tup, A.chunks)), A.dtype)
dask\array\core.py:878: in __new__
    self._chunks = normalize_chunks(chunks, shape, dtype=self.dtype)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

chunks = ((),), shape = None, limit = None, dtype = dtype('float64')
previous_chunks = None

    def normalize_chunks(chunks, shape=None, limit=None, dtype=None,
                         previous_chunks=None):
.
.
.
        for c in chunks:
            if not c:
>               raise ValueError("Empty tuples are not allowed in chunks. Express "
                                 "zero length dimensions with 0(s) in chunks")
E               ValueError: Empty tuples are not allowed in chunks. Express zero length dimensions with 0(s) in chunks

dask\array\core.py:2084: ValueError
---------------------------- Captured stdout call -----------------------------
shape = (10,)
chunks = (1,)
reps = 0
x = [1.80644123e-01 2.71394459e-01 8.07915574e-01 2.56170310e-01
 6.64009049e-01 4.36492273e-02 6.53426686e-01 9.35922465e-01
 3.99060915e-01 8.38937790e-04]
d = dask.array<array, shape=(10,), dtype=float64, chunksize=(1,)>
```

Any advice on how to proceed?
Not from me.  It looks like for some reason there is an empty tuple passed as a chunk element.
I was also looking over some of the other failed testcases, and wanted to ask why pytest raises a "NotImplementedError" when calling the da.tile function.

As I understand it, the NotImplemented Error should only be called when the function is either not implemented or is an abstract method pending implementation in a child class.

```
_________________ test_tile_array_reps[reps0-shape0-chunks0] __________________

shape = (10,), chunks = (1,), reps = [1]

    @pytest.mark.parametrize('shape, chunks', [
        ((10,), (1,)),
        ((10, 11, 13), (4, 5, 3)),
    ])
    @pytest.mark.parametrize('reps', [[1], [1, 2]])
    def test_tile_array_reps(shape, chunks, reps):
        x = np.random.random(shape)
        d = da.from_array(x, chunks=chunks)
        print("shape = {}".format(shape))
        print("chunks = {}".format(chunks))
        print("reps = {}".format(reps))
        print("x = {}".format(x))
        print("d = {}".format(d))
        with pytest.raises(NotImplementedError):
>           da.tile(d, reps)
E           Failed: DID NOT RAISE <class 'NotImplementedError'>

dask\array\tests\test_creation.py:572: Failed
---------------------------- Captured stdout call -----------------------------
shape = (10,)
chunks = (1,)
reps = [1]
x = [0.2742034  0.18548361 0.01524323 0.75336014 0.64558162 0.80801986
 0.30105168 0.16652774 0.19331923 0.40458339]
d = dask.array<array, shape=(10,), dtype=float64, chunksize=(1,)>
_________________ test_tile_array_reps[reps0-shape1-chunks1] __________________

shape = (10, 11, 13), chunks = (4, 5, 3), reps = [1]

    @pytest.mark.parametrize('shape, chunks', [
        ((10,), (1,)),
        ((10, 11, 13), (4, 5, 3)),
    ])
    @pytest.mark.parametrize('reps', [[1], [1, 2]])
    def test_tile_array_reps(shape, chunks, reps):
        x = np.random.random(shape)
        d = da.from_array(x, chunks=chunks)
        print("shape = {}".format(shape))
        print("chunks = {}".format(chunks))
        print("reps = {}".format(reps))
        print("x = {}".format(x))
        print("d = {}".format(d))
        with pytest.raises(NotImplementedError):
>           da.tile(d, reps)
E           Failed: DID NOT RAISE <class 'NotImplementedError'>

dask\array\tests\test_creation.py:572: Failed
---------------------------- Captured stdout call -----------------------------
shape = (10, 11, 13)
chunks = (4, 5, 3)
reps = [1]
x = [[[0.66558733 0.2010031  0.12845087 ... 0.09263792 0.10583748 0.99209543]
  [0.36256797 0.0970691  0.07406676 ... 0.75285652 0.67651216 0.50109876]
  [0.96387529 0.24592683 0.12651464 ... 0.73664226 0.44816201 0.12381766]
  ...
  [0.87078875 0.58406969 0.15834242 ... 0.7337691  0.60768864 0.31679159]
  [0.64517704 0.33423197 0.49616654 ... 0.82066559 0.36135677 0.62351153]
  [0.85151602 0.44429418 0.28333741 ... 0.60677065 0.16604884 0.90253504]]]
d = dask.array<array, shape=(10, 11, 13), dtype=float64, chunksize=(4, 5, 3)>
_________________ test_tile_array_reps[reps1-shape0-chunks0] __________________

shape = (10,), chunks = (1,), reps = [1, 2]

    @pytest.mark.parametrize('shape, chunks', [
        ((10,), (1,)),
        ((10, 11, 13), (4, 5, 3)),
    ])
    @pytest.mark.parametrize('reps', [[1], [1, 2]])
    def test_tile_array_reps(shape, chunks, reps):
        x = np.random.random(shape)
        d = da.from_array(x, chunks=chunks)
        print("shape = {}".format(shape))
        print("chunks = {}".format(chunks))
        print("reps = {}".format(reps))
        print("x = {}".format(x))
        print("d = {}".format(d))
        with pytest.raises(NotImplementedError):
>           da.tile(d, reps)
E           Failed: DID NOT RAISE <class 'NotImplementedError'>

dask\array\tests\test_creation.py:572: Failed
---------------------------- Captured stdout call -----------------------------
shape = (10,)
chunks = (1,)
reps = [1, 2]
x = [0.1840209  0.62584515 0.24485925 0.60199409 0.63640462 0.9670016
 0.38364673 0.19144075 0.38382034 0.65983511]
d = dask.array<array, shape=(10,), dtype=float64, chunksize=(1,)>
_________________ test_tile_array_reps[reps1-shape1-chunks1] __________________

shape = (10, 11, 13), chunks = (4, 5, 3), reps = [1, 2]

    @pytest.mark.parametrize('shape, chunks', [
        ((10,), (1,)),
        ((10, 11, 13), (4, 5, 3)),
    ])
    @pytest.mark.parametrize('reps', [[1], [1, 2]])
    def test_tile_array_reps(shape, chunks, reps):
        x = np.random.random(shape)
        d = da.from_array(x, chunks=chunks)
        print("shape = {}".format(shape))
        print("chunks = {}".format(chunks))
        print("reps = {}".format(reps))
        print("x = {}".format(x))
        print("d = {}".format(d))
        with pytest.raises(NotImplementedError):
>           da.tile(d, reps)
E           Failed: DID NOT RAISE <class 'NotImplementedError'>

dask\array\tests\test_creation.py:572: Failed
---------------------------- Captured stdout call -----------------------------
shape = (10, 11, 13)
chunks = (4, 5, 3)
reps = [1, 2]
x = [[[0.05916106 0.26252344 0.51955179 ... 0.96466021 0.28817599 0.64046503]
  [0.55449445 0.40671171 0.28311431 ... 0.43253649 0.0784907  0.42775786]
  [0.33116566 0.38865136 0.36295301 ... 0.77037302 0.04693195 0.7349799 ]
  ...
  [0.13449347 0.05496916 0.26934709 ... 0.45868004 0.34420579 0.13782773]
  [0.56956731 0.9025545  0.5451738  ... 0.65950547 0.59102771 0.33183651]
  [0.42573379 0.86143516 0.66970509 ... 0.07749012 0.70770808 0.52280939]]]
d = dask.array<array, shape=(10, 11, 13), dtype=float64, chunksize=(4, 5, 3)>

```
Here is a link to the current implementation

https://github.com/dask/dask/blob/4d3a48ee6d5e2b63fef1b0e6740b6478261cdaea/dask/array/creation.py#L758-L770

Perhaps looking at that will help?  I personally don't know why that's a NotImplementedError, but hopefully someone is able to look at the intent of `np.tile` and figure out what is appropriate here.
With NumPy, one can do the following.

```python
In [1]: import numpy as np                                                      

In [2]: a = np.array([0, 1, 2])                                                 

In [3]: np.tile(a, [3, 4])                                                      
Out[3]: 
array([[0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2],
       [0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2],
       [0, 1, 2, 0, 1, 2, 0, 1, 2, 0, 1, 2]])
```

We have not yet implemented this behavior in Dask (hence the `NotImplementedError`). If someone wanted to come along and implement this, that would be welcome.
Should add pytest is not raising `NotImplementedError`. It is checking that `tile` properly raises `NotImplementedError` in those unsupported cases. If a new implementation correctly handled those cases, the `NotImplementedError` test could be replaced by an actual test for the expected behavior.

Commit message: Implement complete tile function

Patch to review:
diff --git a/dask/array/creation.py b/dask/array/creation.py
index 0d17b1ae984..1b10a2585b7 100644
--- a/dask/array/creation.py
+++ b/dask/array/creation.py
@@ -796,17 +796,28 @@ def repeat(a, repeats, axis=None):
 
 @derived_from(np)
 def tile(A, reps):
-    if not isinstance(reps, Integral):
-        raise NotImplementedError("Only integer valued `reps` supported.")
-
-    if reps < 0:
+    try:
+        tup = tuple(reps)
+    except TypeError:
+        tup = (reps,)
+    if any(i < 0 for i in tup):
         raise ValueError("Negative `reps` are not allowed.")
-    elif reps == 0:
-        return A[..., :0]
-    elif reps == 1:
-        return A
-
-    return concatenate(reps * [A], axis=-1)
+    c = asarray(A)
+
+    if all(tup):
+        for nrep in tup[::-1]:
+            c = nrep * [c]
+        return block(c)
+
+    d = len(tup)
+    if d < c.ndim:
+        tup = (1,) * (c.ndim - d) + tup
+    if c.ndim < d:
+        shape = (1,) * (d - c.ndim) + c.shape
+    else:
+        shape = c.shape
+    shape_out = tuple(s * t for s, t in zip(shape, tup))
+    return empty(shape=shape_out, dtype=c.dtype)
 
 
 def expand_pad_value(array, pad_value):
diff --git a/dask/array/tests/test_creation.py b/dask/array/tests/test_creation.py
index 030277a15d5..b936f213238 100644
--- a/dask/array/tests/test_creation.py
+++ b/dask/array/tests/test_creation.py
@@ -571,9 +571,15 @@ def test_repeat():
         assert all(concat(d.repeat(r).chunks))
 
 
+@pytest.mark.parametrize("a", [da.asarray([0, 1, 2]), [[1, 2], [3, 4]]])
+@pytest.mark.parametrize("reps", [2, (2, 2), (1, 2), (2, 1), (2, 3, 4, 0)])
+def test_tile_basic(a, reps):
+    assert_eq(np.tile(a, reps), da.tile(a, reps))
+
+
 @pytest.mark.parametrize("shape, chunks", [((10,), (1,)), ((10, 11, 13), (4, 5, 3))])
-@pytest.mark.parametrize("reps", [0, 1, 2, 3, 5])
-def test_tile(shape, chunks, reps):
+@pytest.mark.parametrize("reps", [0, 1, 2, 3, 5, (1,), (1, 2)])
+def test_tile_chunks(shape, chunks, reps):
     x = np.random.random(shape)
     d = da.from_array(x, chunks=chunks)
 
@@ -591,13 +597,29 @@ def test_tile_neg_reps(shape, chunks, reps):
 
 
 @pytest.mark.parametrize("shape, chunks", [((10,), (1,)), ((10, 11, 13), (4, 5, 3))])
-@pytest.mark.parametrize("reps", [[1], [1, 2]])
-def test_tile_array_reps(shape, chunks, reps):
+@pytest.mark.parametrize("reps", [0, (0,), (2, 0), (0, 3, 0, 4)])
+def test_tile_zero_reps(shape, chunks, reps):
     x = np.random.random(shape)
     d = da.from_array(x, chunks=chunks)
 
-    with pytest.raises(NotImplementedError):
-        da.tile(d, reps)
+    assert_eq(np.tile(x, reps), da.tile(d, reps))
+
+
+@pytest.mark.parametrize("a", [da.asarray([[[]]]), da.asarray([[], []])])
+@pytest.mark.parametrize("reps", [2, (3, 2, 5)])
+def test_tile_empty_array(a, reps):
+    assert_eq(np.tile(a, reps), da.tile(a, reps))
+
+
+@pytest.mark.parametrize(
+    "shape", [(3,), (2, 3), (3, 4, 3), (3, 2, 3), (4, 3, 2, 4), (2, 2)]
+)
+@pytest.mark.parametrize("reps", [(2,), (1, 2), (2, 1), (2, 2), (2, 3, 2), (3, 2)])
+def test_tile_np_kroncompare_examples(shape, reps):
+    x = np.random.random(shape)
+    d = da.asarray(x)
+
+    assert_eq(np.tile(x, reps), da.tile(d, reps))
 
 
 skip_stat_length = pytest.mark.xfail(_numpy_117, reason="numpy-14061")


Code review comments:
diff_hunk: @@ -591,13 +597,29 @@ def test_tile_neg_reps(shape, chunks, reps):
 
 
 @pytest.mark.parametrize("shape, chunks", [((10,), (1,)), ((10, 11, 13), (4, 5, 3))])
-@pytest.mark.parametrize("reps", [[1], [1, 2]])
-def test_tile_array_reps(shape, chunks, reps):
+@pytest.mark.parametrize("reps", [0, (0,), (2, 0), (0, 3, 0, 4)])
+def test_tile_zero_reps(shape, chunks, reps):
     x = np.random.random(shape)
     d = da.from_array(x, chunks=chunks)
 
-    with pytest.raises(NotImplementedError):
-        da.tile(d, reps)
+    assert_eq(np.tile(x, reps), da.tile(d, reps))
+
+
+@pytest.mark.parametrize("a", [da.asarray([[[]]]), da.asarray([[], []])])
+@pytest.mark.parametrize("reps", [2, (3, 2, 5)])
+def test_tile_empty_array(a, reps):
+    assert_eq(np.tile(a, reps), da.tile(a, reps))
path: dask/array/tests/test_creation.py
text: @user1:
```suggestion
    assert_eq(np.tile(a.compute(), reps), da.tile(a, reps))
```

@user1:
As of Numpy 1.17 `np.tilie` will just dispatch to `da.tile` if given a dask array as an input, so let's coerce to numpy here first.

@author:
That's a cool feature I didn't know about! Changed now.

diff_hunk: @@ -571,9 +571,15 @@ def test_repeat():
         assert all(concat(d.repeat(r).chunks))
 
 
+@pytest.mark.parametrize("a", [da.asarray([0, 1, 2]), [[1, 2], [3, 4]]])
path: dask/array/tests/test_creation.py
text: @user1:
I have a slight preference to avoid creating dask array objects at import time.  This is probably fine, but I suspect that some future bugs might interfere more easily here.  Is it easy for you to bring this into the function?  Maybe in this one case we drop the parametrization.

@author:
Thanks, dropped the parametrization.