everything working
MODIFY
pangeo_forge_recipes/recipes/xarray_zarr.py
pangeo_forge_recipes/recipes/xarray_zarr.py
@@ -251,7 +251,9 @@ def region_and_conflicts_for_chunk(
this_chunk_conflicts = set()
for idx in range(start_idx, stop_idx):
conflict = all_chunk_conflicts[idx]
+ if conflict:
this_chunk_conflicts.add(conflict)
+ logger.debug(f"ADDING CHUNK CONFLICT {conflict}")
return {concat_dim: region_slice}, this_chunk_conflicts

MODIFY
pangeo_forge_recipes/utils.py
pangeo_forge_recipes/utils.py
@@ -108,7 +108,7 @@ def lock_for_conflicts(conflicts, base_name="pangeo-forge", timeout=None):
logger.warning("Failed to acquire lock %s before timeout %s", lock.name, timeout)
raise ValueError(f"Failed to acquire lock {lock.name} before timeout {timeout}")
logger.debug(f"Acquired lock {lock.name}")
- else:
+ elif conflicts:
logger.debug(f"Asked to lock {conflicts} but no Dask client found.")
try:
yield

MODIFY
tests/conftest.py
tests/conftest.py
@@ -313,7 +313,9 @@ def execute_recipe(request, dask_cluster):
elif request.param == "prefect":
def execute(recipe):
- return recipe.to_prefect().run()
+ state = recipe.to_prefect().run()
+ if state.is_failed():
+ raise ValueError(f"Prefect flow run failed with message {state.message}")
else:
assert request.param == "prefect-dask"
@@ -321,7 +323,9 @@ def execute_recipe(request, dask_cluster):
def execute(recipe):
flow = recipe.to_prefect()
executor = DaskExecutor(address=dask_cluster.scheduler_address)
- flow.run(executor=executor)
+ state = flow.run(executor=executor)
+ if state.is_failed():
+ raise ValueError(f"Prefect flow run failed with message {state.message}")
execute.param = request.param
return execute

MODIFY
tests/test_recipes.py
tests/test_recipes.py
@@ -17,6 +17,11 @@ all_recipes = [
lazy_fixture("netCDFtoZarr_sequential_subset_recipe"),
]
+recipes_no_subset = [
+ lazy_fixture("netCDFtoZarr_sequential_recipe"),
+ lazy_fixture("netCDFtoZarr_sequential_multi_variable_recipe"),
+]
+
def test_to_pipelines_warns(netCDFtoZarr_sequential_recipe):
RecipeClass, file_pattern, kwargs, ds_expected, target = netCDFtoZarr_sequential_recipe
@@ -111,9 +116,9 @@ def test_process(recipe_fixture, execute_recipe, process_input, process_chunk):
xr.testing.assert_identical(ds_actual, ds_expected)
-@pytest.mark.parametrize("inputs_per_chunk", [1, 2])
+@pytest.mark.parametrize("inputs_per_chunk,subset_inputs", [(1, {}), (1, {"time": 2}), (2, {})])
@pytest.mark.parametrize(
- "target_chunks,specify_nitems_per_input,chunk_expectation",
+ "target_chunks,specify_nitems_per_input,error_expectation",
[
({}, True, does_not_raise()),
({"lon": 12}, True, does_not_raise()),
@@ -127,21 +132,27 @@ def test_process(recipe_fixture, execute_recipe, process_input, process_chunk):
({"lon": 12}, False, pytest.raises(ValueError)),
],
)
-@pytest.mark.parametrize("recipe_fixture", all_recipes)
+@pytest.mark.parametrize("recipe_fixture", recipes_no_subset)
def test_chunks(
recipe_fixture,
execute_recipe,
inputs_per_chunk,
target_chunks,
- chunk_expectation,
+ error_expectation,
+ subset_inputs,
specify_nitems_per_input,
):
"""Check that chunking of datasets works as expected."""
RecipeClass, file_pattern, kwargs, ds_expected, target = recipe_fixture
+ for cdim in file_pattern.combine_dims:
+ if hasattr(cdim, "nitems_per_file"):
+ nitems_per_file = cdim.nitems_per_file
+
kwargs["target_chunks"] = target_chunks
kwargs["inputs_per_chunk"] = inputs_per_chunk
+ kwargs["subset_inputs"] = subset_inputs
if specify_nitems_per_input:
kwargs["metadata_cache"] = None
else:
@@ -154,19 +165,31 @@ def test_chunks(
new_combine_dims.append(cdim)
file_pattern = FilePattern(file_pattern.format_function, *new_combine_dims)
- with chunk_expectation as excinfo:
+ with error_expectation as excinfo:
rec = RecipeClass(file_pattern, **kwargs)
if excinfo:
# don't continue if we got an exception
return
+ # we should get a runtime error if we try to subset by a factor of 2
+ # when the file is only 1 item long
+ subset_factor = kwargs.get("subset_inputs", {}).get("time", 1)
+ if nitems_per_file == 1 and subset_factor > 1:
+ subset_error_expectation = pytest.raises(ValueError)
+ else:
+ subset_error_expectation = does_not_raise()
+ with subset_error_expectation as excinfo:
+ # error is raised at execution stage because we don't generally know a priori how
+ # many items in each file
execute_recipe(rec)
+ if excinfo:
+ # don't continue if we got an exception
+ return
# chunk validation
ds_actual = xr.open_zarr(target.get_mapper(), consolidated=True)
sequence_chunks = ds_actual.chunks["time"]
nitems_per_input = list(file_pattern.nitems_per_input.values())[0]
- subset_factor = kwargs.get("subset_inputs", {}).get("time", 1)
seq_chunk_len = target_chunks.get("time", None) or (
nitems_per_input * inputs_per_chunk // subset_factor
)

everything working