issue_comments

I will be on the lookout for any changes that may be required.

Both sound good to me.

Your code for limit looks OK though I didn't look closely. It looks very similar to https://github.com/pydata/xarray/blob/3960ea3ba08f81d211899827612550f6ac2de804/xarray/core/missing.py#L30-L34

Two questions: 1. Is possible to set the array used for the test_push_dask as np.array([np.nan, 1, 2, 3, np.nan, np.nan, np.nan, np.nan, 4, 5, np.nan, 6])?, using that array you can validate the test case that I put on this issue without creating another array (It's the original array but permuted). 2. Can I erase the conditional that checks for the case where all the chunks have size 1?, I think that with the new method that is not necessary. py # I think this is only necessary due to the use of the map_overlap of the previous method. if all(c == 1 for c in array.chunks[axis]): array = array.rechunk({axis: 2})

yes, of course, by the way, it would be possible to add something like the following code for the case that there is a limit? I know this code generates like 4x more tasks but at least it does the job so, probably a warning could be sufficient. (If it is not good enough to be added there is no problem, probably building the graph manually will be a better option than using this algorithm for the forward fill with limits).

```py def ffill(x: xr.DataArray, dim: str, limit=None):

def _fill_with_last_one(a, b):
    # cumreduction apply the push func over all the blocks first so, 
    # the only missing part is filling the missing values using
    # the last data for every one of them
    if isinstance(a, np.ma.masked_array) or isinstance(b, np.ma.masked_array):
        a = np.ma.getdata(a)
        b = np.ma.getdata(b)
        values = np.where(~np.isnan(b), b, a)
        return np.ma.masked_array(values, mask=np.ma.getmaskarray(b))

    return np.where(~np.isnan(b), b, a)


from bottleneck import push


def _ffill(arr):
    return xr.DataArray(
        da.reductions.cumreduction(
            func=push,
            binop=_fill_with_last_one,
            ident=np.nan,
            x=arr.data,
            axis=arr.dims.index(dim),
            dtype=arr.dtype,
            method="sequential",
        ),
        dims=x.dims,
        coords=x.coords
    )

if limit is not None:
    axis = x.dims.index(dim)
    arange = xr.DataArray(
        da.broadcast_to(
            da.arange(
                x.shape[axis],
                chunks=x.chunks[axis],
                dtype=x.dtype
            ).reshape(
                tuple(size if i == axis else 1 for i, size in enumerate(x.shape))
            ),
            x.shape,
            x.chunks
        ),
        coords=x.coords,
        dims=x.dims
    )
    valid_limits = (arange - _ffill(arange.where(x.notnull(), np.nan))) <= limit
    return _ffill(arr).where(valid_limits, np.nan)

return _ffill(arr)

```

Thanks @josephnowak . This is a great idea! 👏🏾 👏🏾 Can you send in a pull request please? We'll need to add the example from your first post as a test.

I think you can replace this dask_array_ops.push with your version: https://github.com/pydata/xarray/blob/3960ea3ba08f81d211899827612550f6ac2de804/xarray/core/dask_array_ops.py#L56-L80

This function is expected to return a dask array, so you can just return the result of cumreduction instead of wrapping it up in a DataArray.

Probably using the logic of the cumsum and cumprod of dask you can implement the forward fill. I check a little bit the dask code that is on Xarray and apparently none of them use the HighLevelGraph so if the idea is to avoid building the graph manually I think that you can use the cumreduction function of dask to make the work (Probably there is a better dask function for doing this kind of computations but I haven't find it).

```py def ffill(x: xr.DataArray, dim: str, limit=None):

def _fill_with_last_one(a, b):
    # cumreduction apply the push func over all the blocks first so, 
    # the only missing part is filling the missing values using
    # the last data for every one of them
    if isinstance(a, np.ma.masked_array) or isinstance(b, np.ma.masked_array):
        a = np.ma.getdata(a)
        b = np.ma.getdata(b)
        values = np.where(~np.isnan(b), b, a)
        return np.ma.masked_array(values, mask=np.ma.getmaskarray(b))

    return np.where(~np.isnan(b), b, a)


from bottleneck import push

return xr.DataArray(
    da.reductions.cumreduction(
        func=push,
        binop=_fill_with_last_one,
        ident=np.nan,
        x=x.data,
        axis=x.dims.index(dim),
        dtype=x.dtype,
        method="sequential",
    ),
    dims=x.dims,
    coords=x.coords
)

```