diff --git a/src/zagg/index/inline.py b/src/zagg/index/inline.py index 87abaa4..a0e8e26 100644 --- a/src/zagg/index/inline.py +++ b/src/zagg/index/inline.py @@ -25,8 +25,13 @@ (``/.parquet``, the PR #159 offsets schema plus the per-dataset decode metadata) after its last group is read. That is how the sidecar store gets populated before a ``sidecar`` backend can serve it (the -issue's deployment progression); coverage is lazy — the datasets this run's -planned reads actually touched. +issue's deployment progression). Coverage defaults to **full** (issue #190): +the manifest carries chunk maps for *every* decodable dataset in the granule, +enumerated from the same front-of-file metadata the read already cached (no +extra chunk GETs), so one build serves any downstream config — the #163 store +design intent. ``write_back_coverage: lazy`` restores the old behavior +(persist only the datasets this run's planned reads touched) for a private +run that wants the smaller manifest. Known h5coro quirk this backend must sidestep: a hyperslice starting exactly on an interior chunk boundary (``k * chunk_len``, ``k > 0``) trips h5coro's @@ -190,12 +195,26 @@ def build_chunk_map(h5obj, path: str) -> ChunkMap: never raises on its own — it just leaves default metadata) and ``ValueError`` for layouts without file-offset storage (compact). """ - from h5coro.h5dataset import INVALID_VALUE, H5Dataset - from h5coro.h5metadata import H5Metadata + from h5coro.h5dataset import H5Dataset ds = H5Dataset(h5obj, path, earlyExit=True, metaOnly=True, enableAttributes=False) if ds.meta.typeSize == 0: raise KeyError(path) + return _chunk_map_from_dataset(h5obj, ds, path) + + +def _chunk_map_from_dataset(h5obj, ds, path: str) -> ChunkMap: + """Assemble a :class:`ChunkMap` from an already-parsed ``H5Dataset``. + + The metadata-only body of :func:`build_chunk_map`, factored out so the + full-coverage walk (:func:`_iter_datasets`, issue #190) can map each + dataset from the single ``H5Dataset`` it already parsed to classify it — + avoiding a second object-header parse per dataset. ``ds`` must be a + non-group (``typeSize != 0``) metadata-only dataset. + """ + from h5coro.h5dataset import INVALID_VALUE, H5Dataset + from h5coro.h5metadata import H5Metadata + dims = tuple(int(x) for x in ds.meta.dimensions or ()) try: dtype = np.dtype( @@ -265,6 +284,85 @@ def _empty() -> ChunkMap: raise ValueError(f"{path}: unsupported storage layout {ds.meta.layout!r} for chunk indexing") +def _iter_datasets(h5obj): + """Yield ``(path, H5Dataset)`` for every dataset in the granule. + + Descends the group tree with h5coro (``metaOnly``, attributes off), so the + whole enumeration is served from the front-of-file metadata block h5coro + already cached — one ranged GET, no chunk data (issue #190; the full walk + stays inside the single ~4 MiB cache line the config's read already paid + for, measured on real ATL03). A node whose ``typeSize == 0`` is a group + (or a typeless link) and is descended; anything else is a dataset yielded + with the ``H5Dataset`` its classification already parsed, so the caller + can build its chunk map without re-parsing the object header. Each node's + object header is parsed exactly once: parsing a group path populates its + *direct* children into ``h5obj.pathAddresses`` (h5coro traverses one level + to resolve the path), so the classification parse doubles as the group's + enumeration parse. A ``seen`` set of object-header addresses bounds the + recursion against a (non-ICESat-2) hard-link cycle. Paths are absolute + (leading ``/``) and visited in sorted order for determinism. + """ + from h5coro.h5dataset import H5Dataset + + def child_paths(path: str) -> list[str]: + prefix = "" if path == "/" else path.lstrip("/") + "/" + kids = set() + for pp in list(h5obj.pathAddresses.keys()): + if pp.startswith(prefix): + rel = pp[len(prefix) :] + if rel and "/" not in rel: # a direct child of ``path`` + kids.add("/" + prefix + rel) + return sorted(kids) + + seen: set[int] = set() + + def visit(path: str): + for kp in child_paths(path): + ds = H5Dataset(h5obj, kp, earlyExit=False, metaOnly=True, enableAttributes=False) + if ds.meta.typeSize == 0: # group (or typeless link): descend + addr = h5obj.pathAddresses.get(kp.lstrip("/")) + if addr in seen: + continue # hard-link cycle guard + seen.add(addr) + yield from visit(kp) # kp's children are already in pathAddresses + else: + yield kp, ds + + # Parse the root once to populate its direct children, then descend. + H5Dataset(h5obj, "/", earlyExit=False, metaOnly=True, enableAttributes=False) + yield from visit("/") + + +def full_granule_maps(h5obj, existing: dict[str, ChunkMap]) -> dict[str, ChunkMap]: + """Chunk maps for **every** decodable dataset in the granule (issue #190). + + The write-back full-coverage payload: enumerate the whole group tree + (:func:`_iter_datasets`) and map each dataset, **reusing** any map already + in ``existing`` (the config's read set, built during the read) so those + stay byte-identical to the read path and aren't re-walked. Undecodable + dtypes (strings/compounds) are recorded with a blank ``dtype`` and skipped + at read by the sidecar consumer (``datasets_from_manifest``) — the + include-and-skip choice, matching that consumer. COMPACT-layout datasets + (data in the object header, no file offset) raise :class:`ValueError` and + are dropped: a ranged read could never serve them. + """ + maps = dict(existing) + for path, ds in _iter_datasets(h5obj): + if path in maps: + continue + try: + maps[path] = _chunk_map_from_dataset(h5obj, ds, path) + except ValueError: + continue # COMPACT (or other offset-less layout): not chunk-indexable + except Exception as exc: + # Match the read path's per-dataset tolerance (a bad B-tree or an + # odd object header degrades that dataset to the h5coro decoder + # there): drop just this dataset from the manifest instead of + # sinking the whole granule's write-back. + logger.warning(f" write-back: no chunk map for {path} ({exc}); skipping") + return maps + + def granule_manifest(maps: dict[str, ChunkMap]) -> pd.DataFrame: """Assemble one granule's write-back manifest from its chunk maps. @@ -341,11 +439,27 @@ class InlineIndex(VirtualIndex): """ name = "inline" - config_keys = frozenset({"write_back", "store"}) + config_keys = frozenset({"write_back", "store", "write_back_coverage"}) + + #: Accepted ``write_back_coverage`` values (issue #190). + _COVERAGE = frozenset({"full", "lazy"}) - def __init__(self, write_back: bool = False, store: str | None = None): + def __init__( + self, + write_back: bool = False, + store: str | None = None, + write_back_coverage: str = "full", + ): self.write_back = bool(write_back) self.store = store + # Write-back coverage (issue #190): ``full`` (default, the #163 store + # design intent) persists chunk maps for EVERY decodable dataset in the + # granule — extracted from the metadata the read already cached, so any + # downstream config can be served from the manifest; ``lazy`` persists + # only the config's read set (the pre-#190 behavior), an escape hatch + # for a private run that wants the smaller manifest. Neither changes + # what the READ path fetches. + self.write_back_coverage = write_back_coverage # Chunk maps accumulated across each in-flight granule's groups, # keyed full resource URL -> {dataset path -> ChunkMap} (issue #180: # the worker may hold ``data_source.granule_workers`` granules in @@ -373,6 +487,17 @@ def validate_index_config(cls, index_cfg: dict, data_source: dict | None = None) "index.store is only meaningful for backend 'inline' with " "write_back: true (inline never reads the store)" ) + if "write_back_coverage" in index_cfg and not write_back: + raise ValueError( + "index.write_back_coverage is only meaningful for backend 'inline' " + "with write_back: true" + ) + coverage = index_cfg.get("write_back_coverage", "full") + if coverage not in cls._COVERAGE: + raise ValueError( + f"index.write_back_coverage must be one of {sorted(cls._COVERAGE)} " + f"(got {coverage!r})" + ) # Both read routes accept this backend (issue #170 phase 2): sources # with read_plan.spatial_index take the planned route, read-plan-less # (flat) sources the full-read route -- same compiled addressing seam. @@ -392,6 +517,7 @@ def from_index_config(cls, index_cfg: dict) -> "InlineIndex": return cls( write_back=index_cfg.get("write_back", False), store=index_cfg.get("store"), + write_back_coverage=index_cfg.get("write_back_coverage", "full"), ) def _pending_for(self, h5obj) -> dict[str, ChunkMap]: @@ -460,6 +586,13 @@ def finish_granule(self, h5obj, granule_url: str) -> None: maps = self._pending.pop(granule_url, {}) if not self.write_back: return + if self.write_back_coverage == "full": + # Widen the persisted set to every decodable dataset in the granule + # (issue #190): the read already cached the metadata, so this is a + # metadata-only walk (no extra chunk GETs) that lets the manifest + # serve any downstream config, not just this run's read set. The + # read path is untouched — only what write-back PERSISTS grows. + maps = full_granule_maps(h5obj, maps) maps = {path: cm for path, cm in maps.items() if cm.raw} if not maps: return diff --git a/tests/data/index/atl03_mini.h5 b/tests/data/index/atl03_mini.h5 index 0d40393..61ffd2d 100644 Binary files a/tests/data/index/atl03_mini.h5 and b/tests/data/index/atl03_mini.h5 differ diff --git a/tests/data/index/make_fixture.py b/tests/data/index/make_fixture.py index 223542f..d2734fd 100644 --- a/tests/data/index/make_fixture.py +++ b/tests/data/index/make_fixture.py @@ -18,6 +18,19 @@ ph_index_beg,segment_ph_cnt}`` — small, **contiguous** (exercises the pseudo-chunk path of ``build_chunk_map``). +Datasets the shipped config never reads, present so the write-back +full-coverage walk (issue #190) has something beyond the read set to cover: + +- ``/{beam}/heights/delta_time`` — chunked (a non-read chunked dataset), + ``/{beam}/geolocation/segment_id`` — contiguous (a non-read pseudo-chunk). +- ``/ancillary_data/atlas_sdp_gps_epoch`` — a top-level (non-beam) group the + read never descends into. +- ``/ancillary_data/data_start_utc`` — a fixed-length **string** dataset: an + undecodable dtype the full walk records (``dtype == ""``) and the sidecar + consumer skips at read (include-and-skip, issue #190). +- ``/ancillary_data/control`` — a **compact-layout** dataset (data lives in + the object header, no file offset): the walk skips it at write. + Geometry is deterministic: 20 segments of 128 photons each, except segment 8 which is EMPTY (``cnt == 0`` and ``ph_index_beg == 0``, the real ATL03 empty-segment marker — issue #116), for 2432 photons total → 10 chunks of @@ -64,9 +77,26 @@ def beam_arrays(lat_offset: int): "geolocation/reference_photon_lon": 0.001 * ibeg + lat_offset, "geolocation/ph_index_beg": ibeg, "geolocation/segment_ph_cnt": counts, + # Not in the shipped read set (issue #190 full-coverage walk fodder): + # a chunked dataset and a contiguous one the config never touches. + "heights/delta_time": (0.0001 * i).astype(np.float64), + "geolocation/segment_id": np.arange(N_SEG, dtype=np.int32) + 1, } +def _add_compact_dataset(f, name: str, arr: np.ndarray) -> None: + """Create a COMPACT-layout dataset (data in the object header, no file + offset) via h5py's low-level API — the write-back walk skips these.""" + import h5py + + space = h5py.h5s.create_simple(arr.shape) + dcpl = h5py.h5p.create(h5py.h5p.DATASET_CREATE) + dcpl.set_layout(h5py.h5d.COMPACT) + tid = h5py.h5t.py_create(arr.dtype) + dsid = h5py.h5d.create(f.id, name.encode(), tid, space, dcpl) + dsid.write(h5py.h5s.ALL, h5py.h5s.ALL, np.ascontiguousarray(arr)) + + def main(out_path: str = "tests/data/index/atl03_mini.h5") -> None: import h5py @@ -82,6 +112,17 @@ def main(out_path: str = "tests/data/index/atl03_mini.h5") -> None: compression_opts=(4 if chunked else None), shuffle=chunked, ) + # A top-level (non-beam) group the read never descends into, holding + # the edge dtypes/layouts the full-coverage walk must handle (#190): + # a plain contiguous array, an undecodable fixed-length string, and a + # compact-layout dataset. + f.create_dataset( + "ancillary_data/atlas_sdp_gps_epoch", data=np.array([1.198800018e9], dtype=np.float64) + ) + f.create_dataset( + "ancillary_data/data_start_utc", data=np.array(["2018-12-25T02:42:52Z"], dtype="S20") + ) + _add_compact_dataset(f, "ancillary_data/control", np.arange(4, dtype=np.int32)) if __name__ == "__main__": diff --git a/tests/test_index.py b/tests/test_index.py index cad23ee..20633d8 100644 --- a/tests/test_index.py +++ b/tests/test_index.py @@ -541,6 +541,22 @@ def test_missing_dataset_raises_keyerror(self): with pytest.raises(KeyError, match="nope"): build_chunk_map(h5obj, "/gt1l/heights/nope") + def test_string_dataset_recorded_with_blank_dtype(self): + # Undecodable dtype (issue #190): the chunk map is still built (valid + # file offsets) but dtype is blank -- the sidecar consumer skips it at + # read while the manifest keeps it covered. + h5obj = _open_fixture() + cm = build_chunk_map(h5obj, "/ancillary_data/data_start_utc") + assert cm.dtype == "" + assert len(cm.raw) == 1 # contiguous string -> one pseudo-chunk + + def test_compact_dataset_raises_valueerror(self): + # COMPACT layout stores data in the object header (no file offset), so + # it cannot be chunk-indexed -- the full-coverage walk drops it. + h5obj = _open_fixture() + with pytest.raises(ValueError, match="unsupported storage layout"): + build_chunk_map(h5obj, "/ancillary_data/control") + def test_starts_on_boundary(self): h5obj = _open_fixture() cm = build_chunk_map(h5obj, "/gt1l/heights/h_ph") @@ -1100,6 +1116,40 @@ def test_valid_write_back_block(self, tmp_path): assert isinstance(backend, InlineIndex) assert backend.write_back is True assert backend.store == str(tmp_path) + assert backend.write_back_coverage == "full" # #190 default + + def test_coverage_defaults_full(self): + assert InlineIndex(write_back=True, store="s3://b/p/").write_back_coverage == "full" + + def test_coverage_bad_value_rejected(self): + with pytest.raises(ValueError, match="write_back_coverage must be one of"): + validate_index_config( + { + "backend": "inline", + "write_back": True, + "store": "s3://b/p/", + "write_back_coverage": "partial", + } + ) + + def test_coverage_without_write_back_rejected(self): + with pytest.raises(ValueError, match="write_back_coverage is only meaningful"): + validate_index_config({"backend": "inline", "write_back_coverage": "full"}) + + def test_lazy_coverage_from_config(self, tmp_path): + backend = index_from_config( + PipelineConfig( + data_source=_fixture_data_source( + index={ + "backend": "inline", + "write_back": True, + "store": str(tmp_path), + "write_back_coverage": "lazy", + } + ) + ) + ) + assert backend.write_back_coverage == "lazy" class TestGranuleManifest: @@ -1200,6 +1250,24 @@ def _expected_datasets(self, beams=("gt1l", "gt2l")): f"/{beam}/geolocation/{name}" for beam in beams for name in geoloc } + def _full_datasets(self, beams=("gt1l", "gt2l")): + # Full write-back coverage (issue #190): the read set PLUS every other + # decodable dataset in the fixture — the non-read chunked/contiguous + # arrays, a top-level ancillary group, and an undecodable string + # (recorded, skipped at read). The COMPACT ``control`` dataset has no + # file offset and is NOT in the manifest. + extra = {f"/{beam}/heights/delta_time" for beam in beams} | { + f"/{beam}/geolocation/segment_id" for beam in beams + } + return ( + self._expected_datasets(beams) + | extra + | { + "/ancillary_data/atlas_sdp_gps_epoch", + "/ancillary_data/data_start_utc", + } + ) + def test_round_trip_to_local_store(self, tmp_path): store = tmp_path / "zagg-index" / "ATL03" / "007" # created by open_object_store df_out, meta = self._run( @@ -1210,7 +1278,19 @@ def test_round_trip_to_local_store(self, tmp_path): assert manifest_path.is_file() df = pd.read_parquet(manifest_path, engine="fastparquet") assert list(df.columns) == list(MANIFEST_DTYPES) - assert set(df["dataset"]) == self._expected_datasets() + # Full coverage (issue #190): every decodable dataset, not just the + # config's read set — which is a strict subset now. + covered = set(df["dataset"]) + assert covered == self._full_datasets() + assert self._expected_datasets() < covered # strict superset of the read set + # Spot-check datasets the config never reads are present... + assert {"/gt1l/heights/delta_time", "/gt2l/geolocation/segment_id"} <= covered + assert "/ancillary_data/atlas_sdp_gps_epoch" in covered + # ...the undecodable string is recorded with a blank dtype (skipped at + # read by the sidecar consumer)... + assert set(df[df["dataset"] == "/ancillary_data/data_start_utc"]["dtype"]) == {""} + # ...and the COMPACT dataset (no file offset) is dropped. + assert "/ancillary_data/control" not in covered h5obj = _open_fixture() cm = build_chunk_map(h5obj, "/gt1l/heights/h_ph") got = df[df["dataset"] == "/gt1l/heights/h_ph"].sort_values("chunk_idx") @@ -1250,15 +1330,20 @@ def test_coverage_deterministic_for_empty_groups(self, tmp_path): ) assert meta["error"] is None df = pd.read_parquet(store / "atl03_mini.parquet", engine="fastparquet") - assert set(df["dataset"]) == self._expected_datasets() + # gt2l's read returns None before the read seam, but full-coverage + # write-back enumerates the whole granule at finish_granule, so both + # beams (and the shared ancillary datasets) are covered regardless of + # which shard matched — concurrent shards write identical manifests. + assert set(df["dataset"]) == self._full_datasets() def test_manifest_byte_identical_to_direct_chunk_maps(self, tmp_path): - # Pins the sidecar-build product (issue #185): the persisted manifest - # is byte-identical to one assembled directly from build_chunk_map - # over the deterministic per-group coverage — the selection datasets - # included — so the default path's skip-the-walk heuristic cannot - # erode what write_back runs persist. - from zagg.index.inline import write_manifest + # Pins the sidecar-build product (issues #185/#190): the persisted + # manifest is byte-identical to one assembled directly — the read set + # from build_chunk_map (as the read path builds it) widened to full + # coverage by full_granule_maps — so neither the default path's + # skip-the-walk heuristic nor the full-coverage walk erode what + # write_back runs persist. + from zagg.index.inline import full_granule_maps, write_manifest store = tmp_path / "via-backend" df_out, meta = self._run( @@ -1266,7 +1351,8 @@ def test_manifest_byte_identical_to_direct_chunk_maps(self, tmp_path): ) assert meta["error"] is None h5obj = _open_fixture() - direct = {p: build_chunk_map(h5obj, p) for p in self._expected_datasets()} + read_set = {p: build_chunk_map(h5obj, p) for p in self._expected_datasets()} + direct = full_granule_maps(h5obj, read_set) write_manifest(granule_manifest(direct), str(tmp_path / "direct"), "atl03_mini") assert (store / "atl03_mini.parquet").read_bytes() == ( tmp_path / "direct" / "atl03_mini.parquet" @@ -1298,6 +1384,93 @@ def test_finish_granule_drains_pending_when_off(self): backend.finish_granule(object(), "s3://b/g.h5") assert backend._pending == {} + def test_lazy_coverage_persists_only_read_set(self, tmp_path): + # The escape hatch (issue #190): write_back_coverage: lazy reproduces + # the pre-#190 behavior -- the manifest carries exactly the config's + # read set, none of the non-read datasets. + store = tmp_path / "lazy" + df_out, meta = self._run( + tmp_path, + { + "backend": "inline", + "write_back": True, + "store": str(store), + "write_back_coverage": "lazy", + }, + ) + assert meta["error"] is None + df = pd.read_parquet(store / "atl03_mini.parquet", engine="fastparquet") + assert set(df["dataset"]) == self._expected_datasets() + assert "/gt1l/heights/delta_time" not in set(df["dataset"]) + + +class TestFullCoverageWalk: + """Issue #190: write-back full-coverage enumeration + assembly.""" + + def test_iter_datasets_enumerates_whole_granule(self): + from zagg.index.inline import _iter_datasets + + h5obj = _open_fixture() + paths = {p for p, _ in _iter_datasets(h5obj)} + # Every dataset in the fixture -- both beams, the ancillary group, the + # string and compact datasets (groups are descended, not yielded). + assert "/gt1l/heights/delta_time" in paths + assert "/gt2l/geolocation/segment_id" in paths + assert {"/ancillary_data/atlas_sdp_gps_epoch", "/ancillary_data/control"} <= paths + assert "/ancillary_data/data_start_utc" in paths + assert len(paths) == 23 # 16 read set + 7 extras (control incl.) + + def test_iter_datasets_yields_usable_datasets(self): + # The H5Dataset yielded alongside each path builds a chunk map + # byte-identical to a fresh build_chunk_map (no second parse needed). + from zagg.index.inline import _chunk_map_from_dataset, _iter_datasets + + h5obj = _open_fixture() + ref = _open_fixture() + for path, ds in _iter_datasets(h5obj): + if path == "/ancillary_data/control": + continue # compact: raises in both routes + a = _chunk_map_from_dataset(h5obj, ds, path) + b = build_chunk_map(ref, path) + assert a.raw == b.raw and a.dtype == b.dtype and a.dims == b.dims + + def test_full_granule_maps_reuses_and_widens(self): + from zagg.index.inline import full_granule_maps + + h5obj = _open_fixture() + read = {"/gt1l/heights/h_ph": build_chunk_map(h5obj, "/gt1l/heights/h_ph")} + maps = full_granule_maps(h5obj, read) + # The prebuilt map is reused verbatim... + assert maps["/gt1l/heights/h_ph"] is read["/gt1l/heights/h_ph"] + # ...the walk widens to the non-read datasets... + assert "/gt1l/heights/delta_time" in maps + assert "/ancillary_data/data_start_utc" in maps # string kept (blank dtype) + # ...and the COMPACT dataset is dropped (no file offset). + assert "/ancillary_data/control" not in maps + + def test_full_granule_maps_tolerates_one_bad_dataset(self, monkeypatch, caplog): + # A single dataset whose map build raises (not a compact ValueError) + # must not sink the whole granule's write-back -- it degrades like the + # read path (log + skip), and every other dataset is still covered. + import zagg.index.inline as inline_mod + from zagg.index.inline import full_granule_maps + + real = inline_mod._chunk_map_from_dataset + + def flaky(h5obj, ds, path): + if path == "/gt2l/heights/h_ph": + raise RuntimeError("boom") + return real(h5obj, ds, path) + + monkeypatch.setattr(inline_mod, "_chunk_map_from_dataset", flaky) + h5obj = _open_fixture() + with caplog.at_level("WARNING", logger="zagg.index.inline"): + maps = full_granule_maps(h5obj, {}) + assert "/gt2l/heights/h_ph" not in maps # the flaky one dropped + assert "/gt1l/heights/h_ph" in maps # everything else covered + assert "/gt2l/heights/lat_ph" in maps + assert any("boom" in r.message for r in caplog.records) + class TestInlineInterleavedGranules: """Issue #180 phase 1: with granule-level read concurrency the worker may @@ -1450,3 +1623,113 @@ def test_same_basename_granules_do_not_collide(self, tmp_path, caplog): writes = [r for r in caplog.records if "inline write-back" in r.message] assert len(writes) == 2 assert (store / "granule.parquet").is_file() + + +# --------------------------------------------------------------------------- +# Real-granule validation (issue #190): the full-coverage walk on an actual +# ATL03 granule -- ~1,000+ datasets vs the ~48 lazy read set, previously +# covered datasets byte-identical, and the build-time delta. Gated behind the +# slow marker AND the presence of a local granule (the ~/ignore files are not +# in the repo), so CI (which has neither) skips it. +# --------------------------------------------------------------------------- + +_REAL_ATL03_DIRS = ( + Path.home() / "ignore" / "atl03_1336_r05", + Path.home() / "ignore" / "zagg_neon_atl03_test_shard" / "granules", +) + + +def _real_atl03_granule(): + for d in _REAL_ATL03_DIRS: + hits = sorted(d.glob("*.h5")) if d.is_dir() else [] + if hits: + return hits[0] + return None + + +_REAL_GRANULE = _real_atl03_granule() + +# The ATL03 tdigest/gain_bias read set: 8 datasets x 6 beams == 48 (issue #190). +_ATL03_BEAMS = ("gt1l", "gt1r", "gt2l", "gt2r", "gt3l", "gt3r") +_ATL03_READ_SET = tuple( + f"/{b}/{d}" + for b in _ATL03_BEAMS + for d in ( + "heights/lat_ph", + "heights/lon_ph", + "heights/h_ph", + "heights/signal_conf_ph", + "geolocation/ph_index_beg", + "geolocation/reference_photon_lat", + "geolocation/reference_photon_lon", + "geolocation/segment_ph_cnt", + ) +) + + +@pytest.mark.slow +@pytest.mark.skipif(_REAL_GRANULE is None, reason="no local ~/ignore ATL03 granule") +class TestFullCoverageRealGranule: + def _open(self): + from h5coro import filedriver + from h5coro import h5coro as h5c + + return h5c.H5Coro(str(_REAL_GRANULE), filedriver.FileDriver, errorChecking=True) + + def test_full_covers_thousand_plus_and_read_set_byte_identical(self): + import time + + from zagg.index.inline import full_granule_maps + + # Lazy read set (the pre-#190 coverage): time it and keep the maps. + h_lazy = self._open() + t0 = time.perf_counter() + read_set = {p: build_chunk_map(h_lazy, p) for p in _ATL03_READ_SET} + t_lazy = time.perf_counter() - t0 + assert len(read_set) == 48 + + # Full coverage, reusing the read set (as finish_granule does). + h_full = self._open() + read_reused = {p: build_chunk_map(h_full, p) for p in _ATL03_READ_SET} + t0 = time.perf_counter() + full = full_granule_maps(h_full, read_reused) + t_full_walk = time.perf_counter() - t0 + + persisted = {p for p, cm in full.items() if cm.raw} + # ~1,000+ datasets vs 48 -- a strict, order-of-magnitude widening. + assert len(persisted) > 1000 + assert set(_ATL03_READ_SET) < persisted + + # The previously-covered datasets are byte-identical (no regression): + # full reuses the read-set maps verbatim, so the manifest rows match a + # standalone build_chunk_map exactly. + ref = self._open() + for p in _ATL03_READ_SET: + a = full[p] + b = build_chunk_map(ref, p) + assert a.byte_offset.tolist() == b.byte_offset.tolist() + assert a.nbytes.tolist() == b.nbytes.tolist() + assert a.dtype == b.dtype + + # Build-time delta: the incremental walk cost over the lazy read set. + # Loose upper bound only (timing is machine-dependent); the measured + # value is reported in the PR body. + print( + f"\n[#190] full={len(persisted)} datasets vs lazy=48; " + f"lazy build {t_lazy * 1000:.1f} ms, full-walk add {t_full_walk * 1000:.1f} ms" + ) + assert t_full_walk < 5.0 + + def test_manifest_roundtrips_and_stays_within_one_cache_line(self): + # The full walk is metadata-only: it must not pull chunk data. h5coro + # caches in fixed-size lines; the whole enumeration + mapping should + # stay within the front-of-file block the read already fetched. + from zagg.index.inline import full_granule_maps + + h5obj = self._open() + full = full_granule_maps(h5obj, {}) + df = granule_manifest(full) + assert list(df.columns) == list(MANIFEST_DTYPES) + assert df["dataset"].nunique() > 1000 + # One 4 MiB line covers all dataset metadata (measured on real ATL03). + assert len(h5obj.cache) <= 2