Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions cubed/core/ops.py
Original file line number Diff line number Diff line change
Expand Up @@ -243,6 +243,7 @@ def key_function(out_key):
chunkss=[chunks],
target_stores=[target],
output_blocks=output_blocks,
num_tasks=source.npartitions,
**blockwise_kwargs,
)
from cubed import Array
Expand Down
4 changes: 3 additions & 1 deletion cubed/primitive/blockwise.py
Original file line number Diff line number Diff line change
Expand Up @@ -296,6 +296,7 @@ def general_blockwise(
target_chunks_: Optional[T_RegularChunks] = None,
return_writes_stores: bool = False,
output_blocks: Optional[Iterator[List[int]]] = None,
num_tasks=None,
**kwargs,
) -> PrimitiveOperation:
"""A more general form of ``blockwise`` that uses a function to specify the block
Expand Down Expand Up @@ -418,7 +419,8 @@ def general_blockwise(
output_blocks = map(
list, itertools.product(*[range(len(c)) for c in chunks_normal])
)
num_tasks = math.prod(len(c) for c in chunks_normal)
if num_tasks is None:
num_tasks = math.prod(len(c) for c in chunks_normal)

pipeline = CubedPipeline(
apply_blockwise,
Expand Down
13 changes: 12 additions & 1 deletion cubed/tests/test_core.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,13 @@
from cubed._testing import assert_array_equal
from cubed.array_api.dtypes import _floating_dtypes
from cubed.backend_array_api import namespace as nxp
from cubed.core.ops import general_blockwise, merge_chunks, partial_reduce, tree_reduce
from cubed.core.ops import (
_store_array,
general_blockwise,
merge_chunks,
partial_reduce,
tree_reduce,
)
from cubed.core.optimization import fuse_all_optimize_dag, multiple_inputs_optimize_dag
from cubed.core.plan import ArrayRole
from cubed.storage.store import open_storage_array
Expand Down Expand Up @@ -197,6 +203,11 @@ def test_to_zarr_region(tmp_path, spec, executor):
# note that the same zarr store is overwritten in the following tests

region = (slice(0, 2), slice(0, 2))

# need to use internal _store_array function to access plan
# since to_zarr is eager
assert _store_array(a[:2, :2], z, region=region).plan().num_tasks == 1

cubed.to_zarr(a[:2, :2], z, region=region, executor=executor)
res = open_storage_array(store, mode="r")
assert_array_equal(res[region], np.array([[1, 2], [5, 6]]))
Expand Down
Loading