Full-coverage sidecar write-back (issue #190)

src/zagg/index/inline.py

@@ -25,8 +25,13 @@
 (``<store>/<granule_id>.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

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__":