Morton decimal shard ids + hive-partitioned output layout (issue 199)

.github/scripts/run_benchmark.py

@@ -169,7 +169,9 @@ def run_target(
             catalog=str(shardmap_path),
             store=store,
             backend="lambda",
-            morton_cell=str(shard_key),
+            # morton_cell takes the external shard label (the decimal morton
+            # string for HEALPix — issue #199), not the raw packed-word digits.
+            morton_cell=grid.shard_label(shard_key),
             region=region,
             function_name=function_name,
             overwrite=True,

docs/hive_layout.md

@@ -0,0 +1,152 @@
+# Hive store layout (morton-hive/1)
+
+zagg can write each dispatch shard as its **own self-describing leaf zarr**
+under a morton digit tree, instead of into one shared flat store. The layout is
+the write-side half of the sparse-coverage design record
+([`docs/design/sparse_coverage.md`](design/sparse_coverage.md) §2–§3, decisions
+D1–D6); the convention itself is owned by the mortie spec and versioned as
+`morton-hive/1`.
+
+It is opt-in (default `flat`, today's single shared store) and — in this first
+round — wired to the **local backend only** (see [Status](#status)).
+
+## Layout
+
+```
+{store_root}/
+  morton_hive.json               <- static manifest (root-only exception)
+  {sign+base}/{d1}/.../{d_n}/    <- one decimal digit per level (D2)
+    {full_id}.zarr/              <- vanilla zarr v3 leaf, one per shard (D3)
+```
+
+- **Ids are morton decimal strings** (D1): sign + base digit (`1..6` /
+  `-1..-6`), then one digit per order, digits `1..4`, never `0`. A string
+  prefix *is* the spatial ancestor, so cross-resolution containment is
+  `fine_id.startswith(coarse_id)` — arithmetic, not I/O.
+- **One digit per path component** (D2), so shards at mixed orders nest
+  naturally: every order is a legal node.
+- **Full id at the leaf** (D3): `.../-5/1/1/2/3/3/3/-5112333.zarr` is
+  self-describing without parsing its directory chain — greppable in
+  inventories, unambiguous if moved. Each leaf is a completely vanilla zarr v3
+  store: the same group/array template as the flat layout, sized to one shard
+  (dense arrays hold `cells_per_shard` cells; `resolution: chunk` companions
+  hold the shard's K inner chunks; CSR ragged subgroups sit under the same
+  group path, named by the shard label at K==1).
+- **Node invariant** (D5): below the root, a node contains *only* digit
+  children (`[1-4]/`) and `*.zarr` objects — zero zarr metadata above the
+  leaf, no shared mutable state across workers. The root alone also carries
+  the manifest (and, in a follow-on, `coverage.moc`). `zagg.hive`
+  re-checks every computed leaf path against this invariant before writing.
+
+## Config
+
+```yaml
+output:
+  store: s3://bucket/product        # becomes the hive root
+  store_layout: hive                # default: flat
+  grid:
+    type: healpix                   # hive is HEALPix-only (morton digit tree)
+    parent_order: 9                 # shard order -> tree depth
+    child_order: 13                 # cell order
+```
+
+Validation rejects `hive` with a rectilinear grid (node names are morton
+digits), with `sharded: true` (a leaf is a vanilla zarr v3 store), and with
+`consolidate_metadata: true` (there is no store-root zarr hierarchy to
+consolidate — D5/D12).
+
+## The manifest (`morton_hive.json`)
+
+Written **once at template time**, before any shard dispatches; never touched
+during a run (D6). With it, every shard path is computable arithmetically with
+zero requests:
+
+```json
+{
+ "spec": "morton-hive/1",
+ "dataset": {"short_name": "ATL03", "version": "007"},
+ "cell_order": 13,
+ "shard_order": 9,
+ "split_schedule": [1, 1, 1, 1, 1, 1, 1, 1, 1],
+ "pyramid": {"orders": [], "aggregation": {}},
+ "generated_at": "2026-07-10T12:00:00+00:00"
+}
+```
+
+`split_schedule` is implicit under D2 (one digit per level down to the shard
+order) but recorded explicitly for forward compatibility. `pyramid` is
+declared-only in round one: overview zarrs are generated by a later
+post-process sweep (D11), never at fan-out time.
+
+A rerun into an existing root verifies the manifest's **frozen keys** match
+the run's own configuration (`spec`, `dataset`, `cell_order`, `shard_order`,
+`split_schedule`) and fails loudly on a mismatch — the hive analogue of the
+flat layout's shard-map signature guard. The sweep-mutable `pyramid` block and
+the `generated_at` timestamp are deliberately excluded, so a swept store still
+resumes (and the sweep's pyramid declaration is preserved, not clobbered).
+
+**Re-templating does not remove existing leaves.** Overwriting the manifest
+replaces *only* the JSON — committed leaves written under the old
+configuration would survive, stamped and walker-discoverable, and (because
+mixed shard orders are legal under D2) indistinguishable from intentional
+data. The writer therefore refuses an overwrite that changes the frozen keys
+while the digit tree has any `{sign+base}` children (one delimiter-LIST):
+clear the store root, or pick a new one, before writing with a different
+configuration.
+
+## The commit stamp
+
+S3 has no empty directories and LIST is strongly consistent, so **absence is
+trustworthy**: a delimiter-LIST with no digit children means nothing finer
+exists. **Presence is not** — a worker that dies mid-shard has already created
+the `.zarr/` prefix. So the shard's *final* write is a root
+`group.attrs.update(...)` recording completion (D4):
+
+```json
+"morton_hive_commit": {
+  "spec": "morton-hive/1",
+  "complete": true,
+  "cells_with_data": 412,
+  "granule_count": 17,
+  "written_at": "2026-07-10T12:03:41+00:00"
+}
+```
+
+A leaf whose root metadata lacks the stamp is **debris**: incomplete,
+ignorable, safe to overwrite on retry (the writer re-emits the leaf template
+with `overwrite=True`, so retries are idempotent). This is *not* consolidated
+metadata — one small PUT rewriting the root `zarr.json`, which the leaf
+template creates anyway. A shard that errors, or streams no chunks (no data),
+leaves no stamp; a fully empty shard leaves no `.zarr/` prefix at all (the
+leaf is created lazily on the first chunk write).
+
+## Reading a hive store
+
+There is no store-root `zarr.open()` (deliberately — D12; a root hierarchy can
+be added later by the sweep as a derived artifact). Readers:
+
+1. GET `morton_hive.json` (once, cacheable) → `shard_order`, `cell_order`.
+2. Compute a shard's leaf path by string arithmetic on its decimal id
+   (`zagg.hive.shard_leaf_path`), open the leaf zarr, and **check the commit
+   stamp** (`zagg.hive.read_commit`) before trusting the contents.
+3. Discovery without a shard list is a delimiter-LIST walk: recurse on
+   `[1-4]/` children; a `*.zarr` entry is data at that node; no digit children
+   ⇒ nothing finer. Never LIST per observation in a join loop (D10).
+
+The `coverage.moc` domain declaration (§4 of the design record) is a follow-on
+issue and will remove the walk from the discovery path too.
+
+## Status
+
+- **Both backends** write hive stores end-to-end through the same
+  `zagg.hive.process_and_write_hive` code path. On **Lambda**
+  ([issue #199](https://github.com/englacial/zagg/issues/199) phase 3)
+  `mode: "setup"` writes only the manifest — the orchestrator still needs no
+  S3 access — and each worker derives its leaf path from its `shard_key` +
+  the event config's orders, emits its own leaf template, and stamps
+  completion as its final PUT. The async status channel stays at the flat
+  sibling prefix (`{store_root}.status/<run_id>/…`), outside the digit tree.
+- The store-level `coverage.moc` (and any per-shard MOC stamping) waits on
+  the design in [issue #200](https://github.com/englacial/zagg/issues/200).
+- Write-throughput validation at fleet scale is tracked with the benchmark
+  machinery in [issue #202](https://github.com/englacial/zagg/issues/202).

mkdocs.yml

@@ -26,6 +26,7 @@ nav:
       - "Architecture": "design/architecture.md"
       - "Schema": "design/schema.md"
       - "Sharded output": "sharding.md"
+      - "Hive store layout": "hive_layout.md"
       - "Typed morton boundary": "morton_arrow.md"
   - "Deployment":
       - "Standing Up the Backend": "deployment/standup.md"

pyproject.toml

@@ -35,7 +35,10 @@ dependencies = [
     # register from it). 0.2.0 is not on PyPI yet — resolution fails until
     # upstream publishes it.
     "h5coro-hidefix>=0.3.1",
-    "mortie>=0.8.5",
+    # 0.9.0 ships the decimal-emit surface shard ids route through (issue #199):
+    # decimal_repr / to_decimal, the _decimal_to_word parse-back, hive_path, and
+    # the MortonIndexScalar decimal display.
+    "mortie>=0.9.0",
     "earthaccess",
     "boto3",
     "fastparquet",

src/zagg/__main__.py

@@ -29,29 +29,50 @@ def main():
 """,
     )
     parser.add_argument("--config", required=True, help="Path to pipeline config YAML")
-    parser.add_argument("--catalog", default=None, help="Path to granule catalog JSON (overrides config)")
+    parser.add_argument(
+        "--catalog", default=None, help="Path to granule catalog JSON (overrides config)"
+    )
     parser.add_argument("--store", default=None, help="Output store path (overrides config)")
-    parser.add_argument("--backend", default="local", choices=["local", "lambda"],
-                        help="Execution backend (default: local)")
-    parser.add_argument("--driver", default=None, choices=["s3", "https"],
-                        help="Data access driver (default: from config, or s3)")
-    parser.add_argument("--max-cells", type=int, default=None, help="Limit number of cells (for testing)")
-    parser.add_argument("--morton-cell", type=str, default=None, help="Process a specific morton cell")
+    parser.add_argument(
+        "--backend",
+        default="local",
+        choices=["local", "lambda"],
+        help="Execution backend (default: local)",
+    )
+    parser.add_argument(
+        "--driver",
+        default=None,
+        choices=["s3", "https"],
+        help="Data access driver (default: from config, or s3)",
+    )
+    parser.add_argument(
+        "--max-cells", type=int, default=None, help="Limit number of cells (for testing)"
+    )
+    parser.add_argument(
+        "--morton-cell",
+        type=str,
+        default=None,
+        help="Process a single shard: its decimal morton id (e.g. -31123) for "
+        "HEALPix, or the shard-key int for other grids",
+    )
     parser.add_argument("--max-workers", type=int, default=None, help="Max concurrent workers")
     parser.add_argument("--overwrite", action="store_true", help="Overwrite existing Zarr template")
     parser.add_argument("--dry-run", action="store_true", help="Show what would be processed")
     parser.add_argument(
-        "--profile", action="store_true",
+        "--profile",
+        action="store_true",
         help="Emit per-phase worker timings (read/index/aggregate) on the lambda "
-             "backend. Off by default to avoid the per-worker probe tax (issue #100).",
+        "backend. Off by default to avoid the per-worker probe tax (issue #100).",
     )
     parser.add_argument("--region", default="us-west-2", help="AWS region (default: us-west-2)")
     parser.add_argument(
-        "--output-creds", default=None, metavar="PATH",
+        "--output-creds",
+        default=None,
+        metavar="PATH",
         help="Path to a JSON file with credentials for writing the output store "
-             "(keys: accessKeyId, secretAccessKey, optional sessionToken/"
-             "endpointUrl/region; camelCase, snake_case, or STS PascalCase "
-             "spellings accepted). Omit to use the ambient/execution-role creds.",
+        "(keys: accessKeyId, secretAccessKey, optional sessionToken/"
+        "endpointUrl/region; camelCase, snake_case, or STS PascalCase "
+        "spellings accepted). Omit to use the ambient/execution-role creds.",
     )
     parser.add_argument(
         "--function-name",
@@ -94,17 +115,23 @@ def main():
 
     if args.dry_run:
         print(f"\n[DRY RUN] Would process {results['total_cells']} cells")
-        print(f"  Granules per cell: min={results['granules_per_cell_min']}, "
-              f"max={results['granules_per_cell_max']}, "
-              f"avg={results['granules_per_cell_avg']:.1f}")
+        print(
+            f"  Granules per cell: min={results['granules_per_cell_min']}, "
+            f"max={results['granules_per_cell_max']}, "
+            f"avg={results['granules_per_cell_avg']:.1f}"
+        )
         print(f"  Output: {results['store_path']}")
     else:
-        print(f"\nDone: {results['cells_with_data']} cells with data, "
-              f"{results['total_obs']:,} obs, {results['cells_error']} errors, "
-              f"{results['wall_time_s']:.1f}s")
+        print(
+            f"\nDone: {results['cells_with_data']} cells with data, "
+            f"{results['total_obs']:,} obs, {results['cells_error']} errors, "
+            f"{results['wall_time_s']:.1f}s"
+        )
         if "estimated_cost_usd" in results:
-            print(f"Lambda compute: {results['lambda_time_s']:.0f}s total, "
-                  f"{results['gb_seconds']:.0f} GB-s, ~${results['estimated_cost_usd']:.2f}")
+            print(
+                f"Lambda compute: {results['lambda_time_s']:.0f}s total, "
+                f"{results['gb_seconds']:.0f} GB-s, ~${results['estimated_cost_usd']:.2f}"
+            )
         if results.get("worker_phase_max"):
             breakdown = ", ".join(
                 f"{phase} {secs:.0f}s" for phase, secs in results["worker_phase_max"].items()

src/zagg/config.py

@@ -318,6 +318,33 @@ def validate_config(config: PipelineConfig) -> None:
             f"output.consolidate_metadata must be a boolean (got {consolidate_metadata!r})"
         )
 
+    # Optional store layout (issue #199 phase 2): "flat" (default, today's single
+    # shared store) or "hive" (one leaf zarr per shard under a morton digit tree —
+    # docs/design/sparse_coverage.md D1-D6). Hive ids are morton decimal strings,
+    # so the layout is HEALPix-only; the sharded (ShardingCodec) write path and
+    # metadata consolidation both assume the single shared store, so they are
+    # rejected with hive rather than silently mis-writing.
+    store_layout = config.output.get("store_layout")
+    if store_layout is not None and store_layout not in ("flat", "hive"):
+        raise ValueError(f"output.store_layout must be 'flat' or 'hive' (got {store_layout!r})")
+    if store_layout == "hive":
+        if (grid or {}).get("type", "healpix") != "healpix":
+            raise ValueError(
+                "output.store_layout: hive requires a healpix grid (hive node names "
+                f"are morton decimal digits; grid type is {(grid or {}).get('type')!r})"
+            )
+        if (grid or {}).get("sharded"):
+            raise ValueError(
+                "output.store_layout: hive does not support sharded (ShardingCodec) "
+                "output yet — each hive leaf is a vanilla zarr v3 store (D3); drop "
+                "sharded or use the flat layout"
+            )
+        if consolidate_metadata:
+            raise ValueError(
+                "output.store_layout: hive has no store-root zarr hierarchy to "
+                "consolidate (D5/D12) — drop output.consolidate_metadata"
+            )
+
     # Validate bounds structure (optional)
     if config.bounds is not None:
         allowed_keys = {"temporal", "spatial"}
@@ -1507,6 +1534,26 @@ def get_store_path(config: PipelineConfig) -> str | None:
     return config.output.get("store")
 
 
+def get_store_layout(config: PipelineConfig) -> str:
+    """Return the STORE layout — flat single store vs morton hive (issue #199).
+
+    Distinct from ``output.grid.layout`` (the HEALPix *array* layout inside one
+    store): ``output.store_layout`` selects how shard output is arranged under
+    ``output.store``. ``"flat"`` (default) is today's single shared zarr store.
+    ``"hive"`` writes one self-describing leaf zarr per shard at
+    ``{store}/{sign+base}/{d1}/.../{full_id}.zarr`` with a root
+    ``morton_hive.json`` manifest and a per-leaf commit stamp — see
+    ``docs/design/sparse_coverage.md`` (D1-D6) and :mod:`zagg.hive`. A
+    present-but-null key falls back to the default, like ``layout``.
+
+    Returns
+    -------
+    str
+        ``"flat"`` (default) or ``"hive"``.
+    """
+    return config.output.get("store_layout") or "flat"
+
+
 def get_aoi_mask(config: PipelineConfig) -> bool:
     """Whether the optional strict-AOI cell mask is enabled (issue #101).
 

src/zagg/grids/base.py

@@ -258,6 +258,17 @@ def block_index(self, shard_key: ShardKey) -> tuple[int, ...]:
         """
         ...
 
+    def shard_label(self, shard_key: ShardKey) -> str:
+        """External string form of a shard key (issue #199).
+
+        Used wherever a shard id surfaces outside the process — CSR subgroup
+        names, async ``.status`` object keys, log lines. HEALPix renders the
+        packed word as its decimal morton string (D1 in
+        ``docs/design/sparse_coverage.md``); rectilinear keeps the packed tile
+        int's decimal digits.
+        """
+        ...
+
     def shard_footprint(self, shard_key: ShardKey):
         """Shard polygon in WGS84 (shapely.Geometry)."""
         ...
@@ -302,10 +313,23 @@ def nests_with(self, other: "OutputGrid") -> bool:
         ...
 
 
+def shard_label(grid, shard_key) -> str:
+    """External string form of ``shard_key`` under ``grid`` (issue #199).
+
+    Dispatches to ``grid.shard_label`` (decimal morton string for HEALPix,
+    plain int digits for rectilinear) and falls back to ``str(int(...))`` for
+    minimal grid stand-ins that don't implement the method (the same tolerance
+    the worker extends to ``iter_chunks``).
+    """
+    fn = getattr(grid, "shard_label", None)
+    return fn(shard_key) if fn is not None else str(int(shard_key))
+
+
 __all__ = [
     "OutputGrid",
     "ShardKey",
     "InconsistentShardError",
+    "shard_label",
     "vector_array_spec",
     "chunk_array_spec",
     "sharded_array_spec",