From 36fb851950e6e9970d1490a624ac53c5b7a6a735 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Houpert?= <10154151+lhoupert@users.noreply.github.com> Date: Fri, 17 Jul 2026 10:22:09 +0100 Subject: [PATCH 1/3] fix(conversion): open sources with native-chunk-aligned reads and an atomic zarr CacheStore (#220) Some S2 conversions fail with "RuntimeError: error during blosc decompression: -1". Root cause is an fsspec bug open since 2021 (fsspec/filesystem_spec#639): simplecache writes a download straight to its final cache filename, so a concurrent read of the same key can observe a half-written file. Products whose quality/atmosphere aot/wvp are stored as one whole-array 10980x10980 chunk trigger it because dask (chunks="auto") splits that single object into ~9 read tasks fetching it concurrently (EOPF-Explorer/data-pipeline#339). open_source_datatree() (conversion/open_source.py) is the safe way to open a source: - opens with dask chunks matching the on-disk zarr chunks, so each object is fetched by exactly one task (removes the race trigger and the ~9x redundant downloads), and - an optional cache_dir replaces the pipeline's simplecache layer with zarr's built-in CacheStore over a LocalStore (suggested by @d-v-b in EOPF-Explorer/data-pipeline#339). LocalStore writes atomically (temp file + rename), so the half-written-file race cannot occur. Cache entries are namespaced per source URL to prevent cross-product collisions. The convert-s2-optimized CLI command now opens sources through this helper (generic convert/info/validate open sites are left for a follow-up). zarr is bounded to >=3.2.0,<3.3.0 while we depend on the experimental CacheStore API; the import is lazy and guarded so only the cache_dir path is affected if zarr moves it. History note: this branch previously carried a hand-rolled atomic fsspec cache (atomiccache) and an output chunk clamp; the former was replaced by the CacheStore wiring, the latter is split out to its own PR. Synthesized from commits acfd142, f950067, cf1a5d2, cd3eb00, da2f2c1. Co-authored-by: Claude Fable 5 --- pyproject.toml | 4 +- src/eopf_geozarr/cli.py | 6 +- src/eopf_geozarr/conversion/__init__.py | 2 + src/eopf_geozarr/conversion/open_source.py | 90 +++++++++++++ tests/test_open_source.py | 145 +++++++++++++++++++++ uv.lock | 2 +- 6 files changed, 243 insertions(+), 6 deletions(-) create mode 100644 src/eopf_geozarr/conversion/open_source.py create mode 100644 tests/test_open_source.py diff --git a/pyproject.toml b/pyproject.toml index 2e2ce41d..20cf7d5b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,9 @@ requires-python = ">=3.12" dependencies = [ "pydantic-zarr>=0.8.0", "pydantic>=2.12", - "zarr[cast-value-rs]>=3.2.0", + # <3.3.0: open_source_datatree uses zarr's experimental CacheStore, whose + # location/API may change between minor releases; bump deliberately. + "zarr[cast-value-rs]>=3.2.0,<3.3.0", "xarray>=2025.7.1", "dask[array,distributed]>=2026.1.0", "numpy>=2.3.1", diff --git a/src/eopf_geozarr/cli.py b/src/eopf_geozarr/cli.py index 5e4d95da..6a2f23e7 100755 --- a/src/eopf_geozarr/cli.py +++ b/src/eopf_geozarr/cli.py @@ -24,6 +24,7 @@ validate_s3_access, ) from .conversion.geozarr import get_zarr_group +from .conversion.open_source import open_source_datatree if TYPE_CHECKING: from dask.distributed import Client @@ -1243,10 +1244,7 @@ def convert_s2_optimized_command(args: argparse.Namespace) -> None: try: # Load input dataset log.info("Loading Sentinel-2 dataset from", input_path=args.input_path) - storage_options = get_storage_options(str(args.input_path)) - dt_input = xr.open_datatree( - str(args.input_path), engine="zarr", chunks="auto", storage_options=storage_options - ) + dt_input = open_source_datatree(str(args.input_path)) # Convert convert_s2_optimized( diff --git a/src/eopf_geozarr/conversion/__init__.py b/src/eopf_geozarr/conversion/__init__.py index 1a0e448a..1a292070 100644 --- a/src/eopf_geozarr/conversion/__init__.py +++ b/src/eopf_geozarr/conversion/__init__.py @@ -17,6 +17,7 @@ iterative_copy, setup_datatree_metadata_geozarr_spec_compliant, ) +from .open_source import open_source_datatree from .utils import ( calculate_aligned_chunk_size, downsample_2d_array, @@ -36,6 +37,7 @@ "is_s3_path", "iterative_copy", "open_s3_zarr_group", + "open_source_datatree", "parse_s3_path", "s3_path_exists", "setup_datatree_metadata_geozarr_spec_compliant", diff --git a/src/eopf_geozarr/conversion/open_source.py b/src/eopf_geozarr/conversion/open_source.py new file mode 100644 index 00000000..3cf5302b --- /dev/null +++ b/src/eopf_geozarr/conversion/open_source.py @@ -0,0 +1,90 @@ +"""Open source EOPF zarr stores with dask chunks aligned to native chunks. + +With ``chunks="auto"``, dask may split one on-disk zarr chunk into several +read tasks that each fetch and decompress the same object key — multiplying +egress and racing on shared cache files (EOPF-Explorer/data-pipeline#339). +``open_source_datatree`` opens with ``chunks={}`` (the store's native chunk +grid), so each zarr chunk is read by exactly one dask task; downstream +conversion code rechunks to its own output encoding anyway. Note that one +task then materializes one whole native chunk (~241 MB raw for cpm_v270 +aot/wvp) — under tight memory budgets, lower dask task concurrency rather +than sub-splitting stored chunks. +""" + +import hashlib +from pathlib import Path +from typing import Any + +import xarray as xr + +from .fs_utils import S3FsOptions, get_storage_options + + +def open_source_datatree( + path: str, + *, + storage_options: S3FsOptions | dict[str, Any] | None = None, + cache_dir: str | None = None, + engine: str = "zarr", +) -> xr.DataTree: + """Open a source datatree with dask chunks matching the native zarr chunks. + + Parameters + ---------- + path : str + Source store URL (local path, s3://, or https://). + storage_options : dict, optional + fsspec storage options. Defaults to ``get_storage_options(path)`` + (S3 credentials/endpoint for s3:// paths, None otherwise). + cache_dir : str, optional + Local directory for an on-disk read cache of source objects, backed + by zarr's ``CacheStore`` over a ``LocalStore``. LocalStore writes are + atomic (temp file + rename), so a concurrent read never sees a + partially downloaded object. Entries are namespaced per source URL + (CacheStore keys entries by relative zarr key, so different sources + sharing a directory would otherwise collide) and never expire; use an + ephemeral directory. + engine : str + xarray backend engine, default ``"zarr"``. + + Returns + ------- + xr.DataTree + Datatree whose dask arrays are chunked exactly on the store's native + chunk grid: each on-disk zarr chunk is read by exactly one dask task. + """ + if storage_options is None: + storage_options = get_storage_options(path) + if cache_dir is None: + return xr.open_datatree( + path, + engine=engine, + chunks={}, + storage_options=storage_options, + ) + # Imported lazily: CacheStore is zarr's experimental API, so a relocation + # in a future zarr release must not break importing this package. + try: + from zarr.experimental.cache_store import CacheStore + except ImportError as exc: + raise ImportError( + "cache_dir requires zarr.experimental.cache_store.CacheStore " + "(present in zarr 3.2.x). Your zarr version no longer provides " + "it; pin zarr accordingly or open without cache_dir." + ) from exc + from zarr.storage import FsspecStore, LocalStore + + source = FsspecStore.from_url( + path, + storage_options=dict(storage_options) if storage_options else None, + read_only=True, + ) + # CacheStore keys entries by relative zarr key ("aot/0.0" is the same key + # for every S2 product), so namespace the cache per source URL. Trailing + # slashes are stripped so equivalent URLs share a namespace. + cache_key = hashlib.sha256(path.rstrip("/").encode()).hexdigest()[:16] + source_cache = Path(cache_dir) / cache_key + cached = CacheStore(source, cache_store=LocalStore(source_cache)) + # xarray's open_datatree accepts a zarr store at runtime, but its stub does + # not list Store among the accepted input types. + return xr.open_datatree(cached, engine=engine, chunks={}) # pyright: ignore[reportArgumentType] diff --git a/tests/test_open_source.py b/tests/test_open_source.py new file mode 100644 index 00000000..89484e36 --- /dev/null +++ b/tests/test_open_source.py @@ -0,0 +1,145 @@ +"""Tests for open_source_datatree: native-chunk-aligned reads and the +per-source cache (EOPF-Explorer/data-pipeline#339).""" + +import os +import threading +from collections.abc import Iterator +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer + +import numcodecs +import numpy as np +import pytest +import zarr + +from eopf_geozarr.conversion.open_source import open_source_datatree + + +def _build_source_store(store_path: str, seed: int = 42) -> str: + """Write a zarr v2 store mimicking a cpm_v270 source group. + + - ``aot``: ONE whole-array chunk (the pathological v270 layout) + - ``b02``: regular 32x32 tiles + + ``seed`` varies the values so two stores can share a key layout but differ + in content. + """ + group = zarr.open_group(store_path, mode="w", zarr_format=2) + rng = np.random.default_rng(seed) + + aot = group.create_array( + "aot", + shape=(128, 128), + chunks=(128, 128), + dtype="f8", + compressors=numcodecs.Blosc(cname="lz4", clevel=5, shuffle=1), + ) + aot[:] = rng.random((128, 128)) + aot.attrs["_ARRAY_DIMENSIONS"] = ["y", "x"] + + b02 = group.create_array( + "b02", + shape=(128, 128), + chunks=(32, 32), + dtype="f8", + compressors=numcodecs.Blosc(cname="lz4", clevel=5, shuffle=1), + ) + b02[:] = rng.random((128, 128)) + b02.attrs["_ARRAY_DIMENSIONS"] = ["y", "x"] + + zarr.consolidate_metadata(group.store) + return store_path + + +@pytest.fixture +def source_store(tmp_path) -> str: + return _build_source_store(str(tmp_path / "source.zarr")) + + +def test_single_chunk_array_is_one_dask_task(source_store) -> None: + dt = open_source_datatree(source_store) + assert dt["aot"].data.chunks == ((128,), (128,)) + + +def test_tiled_array_chunks_match_native_grid(source_store) -> None: + dt = open_source_datatree(source_store) + assert dt["aot"].data.chunks == ((128,), (128,)) + assert dt["b02"].data.chunks == ((32,) * 4, (32,) * 4) + + +def test_values_roundtrip(source_store) -> None: + dt = open_source_datatree(source_store) + aot_expected = np.asarray(zarr.open_array(source_store, mode="r", path="aot")[:]) + np.testing.assert_array_equal(dt["aot"].values, aot_expected) + + +def test_explicit_storage_options_are_forwarded_to_xarray(monkeypatch) -> None: + captured: dict[str, object] = {} + + def fake_open_datatree(path: object, **kwargs: object) -> None: + captured.update(kwargs) + raise InterruptedError("stop before any I/O") + + monkeypatch.setattr("xarray.open_datatree", fake_open_datatree) + sentinel = {"endpoint_url": "https://example.invalid"} + with pytest.raises(InterruptedError): + open_source_datatree("/some/store.zarr", storage_options=sentinel) + assert captured["storage_options"] == sentinel + assert captured["chunks"] == {} + + +@pytest.fixture +def http_source(source_store) -> Iterator[str]: + """Serve the source store over local HTTP (mimics the EODC https source).""" + root = os.path.dirname(source_store) + name = os.path.basename(source_store) + + class Handler(SimpleHTTPRequestHandler): + def __init__(self, *args: object, **kwargs: object) -> None: + super().__init__(*args, directory=root, **kwargs) # type: ignore[arg-type] + + def log_message(self, format: str, *args: object) -> None: + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + yield f"http://127.0.0.1:{server.server_address[1]}/{name}" + server.shutdown() + + +def test_cache_dir_reads_correct_data_over_http(http_source, source_store, tmp_path) -> None: + """cache_dir reads return correct data and populate the cache, with no + leftover temp files. Atomicity of the cache writes themselves is zarr + LocalStore's contract (temp file + rename).""" + cache_dir = str(tmp_path / "source-cache") + dt = open_source_datatree(http_source, cache_dir=cache_dir) + + aot_expected = np.asarray(zarr.open_array(source_store, mode="r", path="aot")[:]) + np.testing.assert_array_equal(dt["aot"].values, aot_expected) + assert dt["aot"].data.chunks == ((128,), (128,)) # native alignment preserved + + cached_files = [f for _, _, fs in os.walk(cache_dir) for f in fs] + assert cached_files, "cache directory should be populated after reads" + assert not [f for f in cached_files if f.endswith(".partial")] + + +def test_two_sources_can_share_one_cache_dir(tmp_path) -> None: + """A shared cache_dir must never let one source serve another's bytes. + + CacheStore keys entries by zarr key ("aot/0.0" is identical for every S2 + product), so open_source_datatree namespaces the cache per source URL. + """ + first = _build_source_store(str(tmp_path / "first.zarr"), seed=1) + second = _build_source_store(str(tmp_path / "second.zarr"), seed=2) + shared_cache = str(tmp_path / "shared-cache") + + dt_first = open_source_datatree(first, cache_dir=shared_cache) + np.testing.assert_array_equal( + dt_first["aot"].values, + np.asarray(zarr.open_array(first, mode="r", path="aot")[:]), + ) + + dt_second = open_source_datatree(second, cache_dir=shared_cache) + np.testing.assert_array_equal( + dt_second["aot"].values, + np.asarray(zarr.open_array(second, mode="r", path="aot")[:]), + ) diff --git a/uv.lock b/uv.lock index a7c71ef4..bbeb7613 100644 --- a/uv.lock +++ b/uv.lock @@ -854,7 +854,7 @@ requires-dist = [ { name = "structlog", specifier = ">=25.5.0" }, { name = "typing-extensions", specifier = ">=4.15.0" }, { name = "xarray", specifier = ">=2025.7.1" }, - { name = "zarr", extras = ["cast-value-rs"], specifier = ">=3.2.0" }, + { name = "zarr", extras = ["cast-value-rs"], specifier = ">=3.2.0,<3.3.0" }, { name = "zarr-cm", specifier = ">=0.4.1" }, ] From cc8766411e681e5513cf7f6b447c29a1a66c6247 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 17 Jul 2026 10:22:54 +0100 Subject: [PATCH 2/3] chore: release 0.10.2 (#187) Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com> --- .release-please-manifest.json | 2 +- CHANGELOG.md | 10 ++++++++++ pyproject.toml | 2 +- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/.release-please-manifest.json b/.release-please-manifest.json index e1a24428..31ed71b1 100644 --- a/.release-please-manifest.json +++ b/.release-please-manifest.json @@ -1,3 +1,3 @@ { - ".": "0.10.1" + ".": "0.10.2" } \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 20c5c3be..7319ccb4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## 0.10.2 (2026-07-17) + +## What's Changed +* ci(deps): bump the actions group with 3 updates by @dependabot[bot] in https://github.com/EOPF-Explorer/data-model/pull/185 +* bump zarr conventions; migrate type checking to pyright by @d-v-b in https://github.com/EOPF-Explorer/data-model/pull/199 +* fix(conversion): concurrent-safe source opening (native-chunk-aligned reads + atomic CacheStore) for data-pipeline#339 by @lhoupert in https://github.com/EOPF-Explorer/data-model/pull/220 + + +**Full Changelog**: https://github.com/EOPF-Explorer/data-model/compare/v0.10.1...v0.10.2 + ## 0.10.1 (2026-06-09) ## What's Changed diff --git a/pyproject.toml b/pyproject.toml index 20cf7d5b..d908faf3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "eopf-geozarr" -version = "0.10.1" +version = "0.10.2" description = "GeoZarr compliant data model for EOPF datasets" readme = "README.md" license = { text = "Apache-2.0" } From a4a6973c1db992cdee1a88a9d0825fd4c0da5e6b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Houpert?= <10154151+lhoupert@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:52:34 +0100 Subject: [PATCH 3/3] chore: re-lock uv.lock after merging main Co-Authored-By: Claude Fable 5 --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index 49379c2e..6a1eaacc 100644 --- a/uv.lock +++ b/uv.lock @@ -1109,7 +1109,7 @@ wheels = [ [[package]] name = "eopf-geozarr" -version = "0.10.1" +version = "0.10.2" source = { editable = "." } dependencies = [ { name = "aiohttp" },