diff --git a/docs/specification.md b/docs/specification.md index 2f0515b30..b866d208b 100644 --- a/docs/specification.md +++ b/docs/specification.md @@ -144,7 +144,9 @@ A vlen array without a well-formed `element` declaration is **not** a `zagg-ragged/1` array; a reader MUST refuse it with a pointed error rather than decode under a guessed layout (pre-issue-209 CSR stores are a hard break). The `ragged` attrs key is reserved: config-declared field attrs MUST -NOT shadow it (enforced at config validation). A located field's provenance +NOT shadow it (enforced at config validation). The §2.0 `weights` key is +likewise spec-owned on a ragged payload array — writer-stamped from the +field's declaration, never author-transcribed. A located field's provenance attrs (e.g. `stratum`, `signal_threshold` — §3.3) land on the **payload array only**; the `{field}_locations` sibling carries no user attrs. @@ -239,15 +241,48 @@ A t-digest field is a `zagg-ragged/1` (or `/2`) array whose element declaration is `{"dtype": "float32", "shape": [-1, 2]}`. Source of truth in code: `zagg.stats.tdigest`. +### 2.0 The `weights` declaration + +**Contract** ([issue #422](https://github.com/englacial/zagg/issues/422)). +A digest payload array declares the semantics of its weight column under the +**`weights`** attrs key — a **sibling** of the §1.2 `ragged` block on the +payload array, never a key inside it (the `ragged` block is retired wholesale +under `/2` — §1.6/§6.3 — so a sibling key survives that metadata-only +migration untouched). Two values are defined: + +- **`"counts"`** — weights are observation counts: integers ≥ 1 whose sum is + the cell's exact observation count, per §2.1. **An absent `weights` key + MUST be read as `"counts"`** — every store written before this revision is + conformant verbatim, no byte rewritten. +- **`"flux"`** — weights are calibrated flux: positive finite float32 reals + (a zero-weight observation carries no flux and MUST NOT produce a row); + `sum(weights)` estimates the cell's detected **photoelectrons**, not an + observation count. A flux-declared array MUST record its calibration + provenance in the same attrs: a `gain` key carrying at minimum the gain + constant's `name` and `version` (the operating point of any write-time + clip rides alongside, writer-defined). + +A reader MUST strict-check the value: an unknown declaration is a future +revision of this section and MUST be refused, never read as either defined +value. **Merges are legal only between payloads carrying the same +declaration** (counts with counts, flux with flux — an absent key is +`"counts"` for this rule too): a mixed merge would produce a weight column +whose sum means neither thing, so a merging reader or writer MUST refuse it. +The declaration rides the payload array only; a located field's +`{field}_locations` sibling carries no `weights` key (§1.2's no-user-attrs +rule for siblings is unchanged). + ### 2.1 Centroid array **Contract.** A populated cell's decoded payload is a `(k, 2)` **float32** array of weighted centroids: -- column 0 is the centroid **mean**; column 1 is the centroid **weight** - (the number of observations merged into it, ≥ 1); +- column 0 is the centroid **mean**; column 1 is the centroid **weight** — + under the `"counts"` declaration (§2.0, the default) the number of + observations merged into it, an integer ≥ 1; under `"flux"` a positive + real per §2.0; - rows MUST be sorted **ascending by mean**; -- `sum(weights)` MUST equal the cell's **exact** observation count — the +- under `"counts"`, `sum(weights)` MUST equal the cell's **exact** observation count — the number of finite `source` values the digest was built over (non-finite source rows are dropped before building) — **while that count is representable in float32, i.e. `<= 2^24` (16,777,216)**; above that bound @@ -258,6 +293,9 @@ array of weighted centroids: watch at coarse overview orders (§4.4). For a stratified product (§3) each stratum digest's total weight is the exact stratum count, under the same bound; +- under `"flux"` (§2.0) `sum(weights)` is a float32 photoelectron estimate, + not a count: the exact-count recovery above (and §3.3's) is undefined for + a flux payload, and no integrality holds; - an absent cell decodes as the zero-length `(0, 2)` array (the `b""` fill). ### 2.2 The location channel @@ -823,7 +861,16 @@ staged sweep's finisher. method on an excluded field would declare a t-digest array that does not exist. `exact`/`approximate` entries carry the fold `method`, any further fold provenance (an `exact` fold's `nan_policy`), and enough dtype/shape - metadata to know the overview array's form up front. This map is the + metadata to know the overview array's form up front. An `approximate` + entry MAY additionally carry `overview_delta` — the compression budget + overview folds run at when it is split from the leaf `delta` + ([issue #424](https://github.com/englacial/zagg/issues/424); both budgets + are fold algebra, informative per §2.3) — and, for a non-default §2.0 + declaration, `weights` together with the `gain` calibration provenance + §2.0 requires beside it (the overview writer reconstructs its arrays from + this entry alone, so a declaration recorded here without its provenance + would write an overview whose calibration is unrecoverable); a reader MUST + tolerate entry keys it does not bind. This map is the **all-fields** view; the per-overview `zagg_overview.fields` attrs map (§4.3) is the materialized subset. - **`all_time`** — whether the `all.zarr` all-time fold is materialized at @@ -974,7 +1021,8 @@ re-invoking the idempotent leaf, never a sweep-side fold from raw cells. "window": "all", "fields": {"count": {"class": "exact", "method": "sum", "nan_policy": "skip"}, "h_tdigest": {"class": "approximate", "method": "tdigest_kway", - "delta": 16, "dtype": "float32", "inner_shape": [2]}}, + "delta": 16, "overview_delta": 16, + "dtype": "float32", "inner_shape": [2]}}, "groups": {"5": {"regime": "leaf-column", "merges_from_raw": 1, "n_cells": 4}, "4": {"regime": "leaf-column", "merges_from_raw": 1, "n_cells": 1}}, "cells_with_data_order": 5, @@ -989,7 +1037,12 @@ re-invoking the idempotent leaf, never a sweep-side fold from raw cells. commit stamp's `cells_with_data` records (the finest group). `fields` follows §4.3's materialized-fields contract (approximate entries additionally carry `dtype`/`inner_shape`/`delta` — enough to decode - without the manifest). `groups` carries the per-group provenance slots: + without the manifest — and `overview_delta`, the budget this column's fold + actually compressed at, which is the split pyramid-fold budget rather than + the leaf `delta` ([issue #424](https://github.com/englacial/zagg/issues/424); + both are fold algebra, informative per §2.3). A reader MUST tolerate entry + keys it does not bind, exactly as in §4.5). `groups` carries the per-group + provenance slots: the fold **regime** (`"leaf-column"` — folded from the leaf's own resident cells; `source_children` never rides this regime, its source is complete by construction), the `merges_from_raw` integer, and `n_cells` @@ -1322,13 +1375,20 @@ drift fails zagg's own suite (`tests/test_spec_conformance.py`) on whichever side moved. moczarr vendors the same fixtures for its parity gates (espg/moczarr#19/#20). -Three tiny single-shard hive stores plus one manifest-only declaration, all +Four tiny single-shard hive stores plus one manifest-only declaration, all on the same deliberately small geometry — shard order 4, inner-chunk order 5, cell order 6 (16 cells, K = 4 inner chunks of 4 cells), sharded (the hive default): - **`minimal/`** — one *unlocated* digest field (`h_tdigest`) plus `count`. The smallest thing that is a conforming store. +- **`flux/`** — the §2.0 `weights` declaration surface: one flux-declared + digest field (`rx_flux`, `weights: "flux"` stamped beside the `ragged` + block, `gain` provenance attrs) plus `count`. Its payloads carry + fractional positive weights whose per-cell sums are **not** integers — + the pin that a flux reader must not round-trip weights through counts — + while `minimal/` (committed before this revision, unregenerated) pins the + absent-key ⇒ `"counts"` default. - **`kitchen_sink/`** — the full stratified-product surface: located signal/noise digest strata (payload + `{field}_locations` siblings, `stratum`/`signal_threshold` provenance attrs), the `composition` word diff --git a/src/zagg/column.py b/src/zagg/column.py index 1378174ad..634a6b25f 100644 --- a/src/zagg/column.py +++ b/src/zagg/column.py @@ -167,7 +167,7 @@ def fold_column(slabs: dict, fields: dict, *, cell_order: int, resolutions: list a fractional fold factor, which no guard downstream can read as a divisor (both classes would surface it as an opaque numpy failure instead). """ - from zagg.sweep_overview import decode_digest, fold_dense, fold_digests + from zagg.sweep_overview import decode_digest, fold_dense, fold_digests, overview_fold_delta cell_order = int(cell_order) fields = composable_fields(fields) @@ -192,7 +192,7 @@ def fold_column(slabs: dict, fields: dict, *, cell_order: int, resolutions: list else: dtype = meta.get("dtype") or "float32" inner = tuple(meta.get("inner_shape") or (2,)) - delta = int(meta.get("delta") or 512) + delta = overview_fold_delta(meta) if slab.shape[0] % factor: raise ValueError( f"cannot fold {slab.shape[0]} cells {factor}-to-one for {name!r}" @@ -223,11 +223,14 @@ def _column_provenance(meta: dict) -> dict: shared helper — the overview's identical gap is a spec call for the issue #383 phase 4 section, not a reason to leave this artifact short. """ - from zagg.sweep_overview import _field_provenance + from zagg.sweep_overview import _field_provenance, overview_fold_delta entry = dict(_field_provenance(meta)) if meta.get("class") == "approximate": entry["delta"] = int(meta.get("delta") or 512) + # The budget the column fold actually compressed at (issue #424): + # the split overview_delta, not the leaf δ. + entry["overview_delta"] = overview_fold_delta(meta) entry["dtype"] = meta.get("dtype") or "float32" entry["inner_shape"] = list(meta.get("inner_shape") or (2,)) return entry diff --git a/src/zagg/config.py b/src/zagg/config.py index 215c8cd90..feec2463d 100644 --- a/src/zagg/config.py +++ b/src/zagg/config.py @@ -565,13 +565,22 @@ def validate_config(config: PipelineConfig) -> None: if attrs is not None: if not isinstance(attrs, dict) or not all(isinstance(k, str) for k in attrs): raise ValueError(f"Variable '{name}': attrs must be a mapping with string keys") - from zagg.grids.base import RAGGED_ELEMENT_ATTR + from zagg.grids.base import RAGGED_ELEMENT_ATTR, WEIGHTS_ATTR if RAGGED_ELEMENT_ATTR in attrs: raise ValueError( f"Variable '{name}': attrs key {RAGGED_ELEMENT_ATTR!r} is reserved " f"for the ragged layout block (issue #209)" ) + # The §2.0 weights key is spec-owned on ragged payload arrays: + # the template stamps it from the field-level ``weights:`` + # declaration, so an attrs transcription could silently disagree. + if meta.get("kind") == "ragged" and WEIGHTS_ATTR in attrs: + raise ValueError( + f"Variable '{name}': attrs key {WEIGHTS_ATTR!r} is spec-owned on a " + f"ragged payload array — declare the field-level 'weights:' key " + f"instead (spec §2.0, issue #424)" + ) import json try: @@ -1473,7 +1482,11 @@ def _validate_output_kind(name: str, meta: dict) -> None: ``ragged``). ``scalar`` fields need neither and stay the default path. ``vector`` and ``ragged`` fields may be driven by either ``function`` or ``expression``; ``len``/``count`` are rejected for both (they short-circuit - to a scalar count). See issue #29 (vector) and issue #48 (ragged). + to a scalar count). See issue #29 (vector) and issue #48 (ragged). A ragged + field may additionally declare ``weights`` (the spec §2.0 counts/flux + payload declaration; flux requires ``gain`` provenance attrs) and + ``overview_delta`` (the split pyramid-fold budget) — both issue #424, + both rejected on other kinds. A field may also declare ``resolution`` (``cell`` default, or ``chunk``). A ``resolution: chunk`` field (issue #30 item 2) is written ONCE per chunk @@ -1514,6 +1527,16 @@ def _validate_output_kind(name: str, meta: dict) -> None: f"Variable '{name}': 'location' is only valid for kind 'ragged', not '{kind}'" ) + # ``weights`` (spec §2.0, issue #424) declares a digest payload's + # weight-column semantics; ``overview_delta`` (issue #424) is the split + # pyramid-fold compression budget. Both describe a ragged digest payload — + # nothing else has a weight column or an overview digest fold. + for key in ("weights", "overview_delta"): + if key in meta and kind != "ragged": + raise ValueError( + f"Variable '{name}': '{key}' is only valid for kind 'ragged', not '{kind}'" + ) + # resolution (cell default, or chunk). A chunk-resolution field stores one # value per chunk in a companion array (issue #30 item 2). ``scalar`` and # ``vector`` chunk companions are wired (issue #82): a scalar companion is a @@ -1576,6 +1599,44 @@ def _validate_output_kind(name: str, meta: dict) -> None: raise ValueError(f"Variable '{name}': kind 'ragged' requires 'inner_shape'") _validate_trailing_shape(name, meta["inner_shape"], key_name="inner_shape") + # The §2.0 weights declaration (issue #424): "counts" (integer weights, + # sum exact — the absent-key default) or "flux" (positive reals, sum an + # estimate of detected photoelectrons). A flux field MUST carry its + # calibration provenance — the spec requires a ``gain`` attrs mapping + # naming at minimum the gain constant's name and version — so a store + # never holds calibrated weights whose calibration is unrecoverable. + weights = meta.get("weights") + if weights is not None: + from zagg.grids.base import WEIGHTS_KINDS + + if weights not in WEIGHTS_KINDS: + raise ValueError( + f"Variable '{name}': weights {weights!r} is not one of {WEIGHTS_KINDS} (spec §2.0)" + ) + if weights == "flux": + gain = (meta.get("attrs") or {}).get("gain") + if not isinstance(gain, dict) or not {"name", "version"} <= set(gain): + raise ValueError( + f"Variable '{name}': weights 'flux' requires calibration provenance " + f"in attrs — a 'gain' mapping with at least 'name' and 'version' " + f"(spec §2.0, issue #424)" + ) + + # ``overview_delta`` (issue #424): the pyramid/overview fold budget, split + # from the leaf ``params.delta`` (leaf δ is the loss-free bound, overview δ + # the ~1/δ accuracy bound). A top-level key, NOT a params entry — params + # values are forwarded to the reducer as kwargs, which never take it. + overview_delta = meta.get("overview_delta") + if overview_delta is not None: + if not isinstance(overview_delta, int) or isinstance(overview_delta, bool): + raise ValueError( + f"Variable '{name}': overview_delta must be a positive int (got {overview_delta!r})" + ) + if overview_delta < 1: + raise ValueError( + f"Variable '{name}': overview_delta must be a positive int (got {overview_delta!r})" + ) + # Same restriction as vector: ``len``/``count`` produce a scalar count. if meta.get("function") in ("len", "count"): raise ValueError( @@ -2166,6 +2227,10 @@ def get_output_signature(meta: dict) -> dict: # Ragged location channel (issue #87): the per-observation morton column # the reducer folds per centroid; ``None`` for unlocated fields. "location": meta.get("location"), + # Ragged weights declaration (spec §2.0, issue #424): "counts"/"flux", + # or ``None`` when undeclared (the spec reads absence as counts, so + # ``None`` keeps pre-#424 templates and signatures byte-identical). + "weights": meta.get("weights"), } @@ -2204,6 +2269,10 @@ def output_field_signature(config: PipelineConfig) -> list[dict]: # so existing shard-map signatures are byte-identical. if sig["location"] is not None: entry["location"] = sig["location"] + # A weights declaration changes the payload semantics (spec §2.0, + # issue #424) — same keyed-only-when-set discipline as location. + if sig["weights"] is not None: + entry["weights"] = sig["weights"] fields.append(entry) return sorted(fields, key=lambda f: f["name"]) diff --git a/src/zagg/configs/atl03_tdigest_healpix.yaml b/src/zagg/configs/atl03_tdigest_healpix.yaml index 71850f684..a753a3918 100644 --- a/src/zagg/configs/atl03_tdigest_healpix.yaml +++ b/src/zagg/configs/atl03_tdigest_healpix.yaml @@ -68,7 +68,7 @@ aggregation: h_tdigest: # Per-cell t-digest of photon heights: a ragged (vlen-bytes) field whose payload # is an (n_centroids, 2) array of (mean, weight) centroids. ``build_tdigest`` - # is called as build_tdigest(h_ph_values, delta=256); ``inner_shape: [2]`` + # is called as build_tdigest(h_ph_values, delta=8192); ``inner_shape: [2]`` # declares the per-element centroid width. To store ONE digest per chunk # instead of per cell, add ``resolution: chunk`` (one payload per inner # chunk under the chunk-uniform contract). @@ -77,7 +77,15 @@ aggregation: source: h_ph inner_shape: [2] params: - delta: 256 + # Leaf centroid budget (issues #414/#424): loss-free while a cell's + # count stays <= delta. 8192 covers every real surface cell in the + # statewide CA scan (only atmospheric storm artifacts exceed it); + # digest size scales with actual counts, so typical cells are unmoved. + delta: 8192 + # Pyramid/overview folds compress at this split budget instead (~1/512 + # quantile accuracy) — overviews are summaries, not loss-free carriers, + # and the cap bounds the sweep's k-way fold buffers (issue #424). + overview_delta: 512 dtype: float32 fill_value: 0 diff --git a/src/zagg/configs/atl03_tdigest_healpix_hive.yaml b/src/zagg/configs/atl03_tdigest_healpix_hive.yaml index 2b3c6cf30..3aedd3e90 100644 --- a/src/zagg/configs/atl03_tdigest_healpix_hive.yaml +++ b/src/zagg/configs/atl03_tdigest_healpix_hive.yaml @@ -66,7 +66,8 @@ aggregation: source: h_ph inner_shape: [2] params: - delta: 256 # centroid budget (accuracy knob) + delta: 8192 # leaf centroid budget: loss-free bound (issues #414/#424) + overview_delta: 512 # pyramid-fold budget: ~1/512 accuracy (issue #424) dtype: float32 fill_value: 0 diff --git a/src/zagg/configs/atl03_tdigest_located_healpix.yaml b/src/zagg/configs/atl03_tdigest_located_healpix.yaml index e1f350c18..4869727d5 100644 --- a/src/zagg/configs/atl03_tdigest_located_healpix.yaml +++ b/src/zagg/configs/atl03_tdigest_located_healpix.yaml @@ -68,7 +68,8 @@ aggregation: location: leaf_id inner_shape: [2] params: - delta: 256 + delta: 8192 # leaf centroid budget: loss-free bound (issues #414/#424) + overview_delta: 512 # pyramid-fold budget: ~1/512 accuracy (issue #424) dtype: float32 fill_value: 0 diff --git a/src/zagg/configs/atl03_tdigest_strata_healpix.yaml b/src/zagg/configs/atl03_tdigest_strata_healpix.yaml index a101a5708..5d3617521 100644 --- a/src/zagg/configs/atl03_tdigest_strata_healpix.yaml +++ b/src/zagg/configs/atl03_tdigest_strata_healpix.yaml @@ -97,11 +97,12 @@ aggregation: location: leaf_id inner_shape: [2] params: - delta: 256 + delta: 8192 # leaf centroid budget: loss-free bound (issues #414/#424) where: >- (signal_conf_land >= 2) | (signal_conf_ocean >= 2) | (signal_conf_sea_ice >= 2) | (signal_conf_land_ice >= 2) | (signal_conf_inland_water >= 2) + overview_delta: 512 # pyramid-fold budget: ~1/512 accuracy (issue #424) dtype: float32 fill_value: 0 attrs: @@ -114,11 +115,12 @@ aggregation: location: leaf_id inner_shape: [2] params: - delta: 256 + delta: 8192 # leaf centroid budget: loss-free bound (issues #414/#424) where: >- ~((signal_conf_land >= 2) | (signal_conf_ocean >= 2) | (signal_conf_sea_ice >= 2) | (signal_conf_land_ice >= 2) | (signal_conf_inland_water >= 2)) + overview_delta: 512 # pyramid-fold budget: ~1/512 accuracy (issue #424) dtype: float32 fill_value: 0 attrs: diff --git a/src/zagg/grids/base.py b/src/zagg/grids/base.py index cfa154ed4..22df47d9e 100644 --- a/src/zagg/grids/base.py +++ b/src/zagg/grids/base.py @@ -58,6 +58,39 @@ #: produce identical objects across workers. RAGGED_ZSTD_LEVEL = 3 +#: Array-attrs key declaring a digest payload's weight-column semantics +#: (spec §2.0, issue #424): a SIBLING of :data:`RAGGED_ELEMENT_ATTR` on the +#: payload array — never a key inside the versioned ``ragged`` block, which +#: is retired wholesale under ``zagg-ragged/2`` (a sibling key survives that +#: metadata-only migration untouched; espg ruling on issue #422). Spec-owned: +#: stamped at template time from the field's ``weights:`` declaration, +#: reserved against config-declared attrs. +WEIGHTS_ATTR = "weights" + +#: The defined §2.0 declarations. ``counts`` — integer weights ≥ 1 summing to +#: the exact observation count (the default: an ABSENT key reads as counts, +#: keeping every pre-#424 store conformant verbatim). ``flux`` — positive +#: reals whose sum estimates detected photoelectrons (calibration provenance +#: required in attrs). Merges are legal only between matching declarations. +WEIGHTS_KINDS = ("counts", "flux") + + +def weights_declaration(attrs) -> str: + """The §2.0 weights declaration recorded in a payload array's attrs. + + Absent key reads as ``"counts"`` (spec §2.0 — existing stores are + conformant verbatim); an unknown value is a future spec revision and + raises rather than half-parsing (the strict-check discipline every + versioned convention on the spec page gets). + """ + value = dict(attrs or {}).get(WEIGHTS_ATTR, "counts") + if value not in WEIGHTS_KINDS: + raise ValueError( + f"unknown weights declaration {value!r} (spec §2.0 defines {WEIGHTS_KINDS}); " + f"refusing to guess weight semantics for a future spec revision" + ) + return value + def apply_field_attrs(spec, meta: dict): """Merge a variable's config-declared ``attrs`` onto its array spec. @@ -135,6 +168,7 @@ def ragged_array_spec( element_dtype, inner_shape=(), locations=None, + weights=None, ): """Vlen-bytes ``ArraySpec`` for a ``kind: ragged`` field (issue #209). @@ -180,6 +214,13 @@ def ragged_array_spec( declared in the payload array's attrs so a reader binds the channel by METADATA, not by reconstructing the naming convention (review, PR #211). ``None`` (unlocated) records nothing. + weights : str, optional + The field's §2.0 weights declaration (issue #424), stamped as the + :data:`WEIGHTS_ATTR` SIBLING key beside the ``ragged`` block (never + inside it — the block is retired under ``/2``). ``None`` (undeclared) + records nothing: the spec reads an absent key as ``"counts"``, so + pre-#424 configs emit byte-identical templates. Pass it for the + payload array only — a located sibling carries no user/spec attrs. Returns ------- @@ -215,8 +256,15 @@ def ragged_array_spec( ragged_meta: dict = {"spec": RAGGED_SPEC, "element": element} if locations is not None: ragged_meta["locations"] = str(locations) + attributes: dict = {RAGGED_ELEMENT_ATTR: ragged_meta} + if weights is not None: + if weights not in WEIGHTS_KINDS: + raise ValueError( + f"weights declaration {weights!r} is not one of {WEIGHTS_KINDS} (spec §2.0)" + ) + attributes[WEIGHTS_ATTR] = str(weights) return ArraySpec( - attributes={RAGGED_ELEMENT_ATTR: ragged_meta}, + attributes=attributes, shape=tuple(int(s) for s in shape), dimension_names=tuple(dims), data_type="variable_length_bytes", @@ -584,6 +632,9 @@ def shard_label(grid, shard_key) -> str: "RAGGED_ELEMENT_ATTR", "RAGGED_SPEC", "RAGGED_ZSTD_LEVEL", + "WEIGHTS_ATTR", + "WEIGHTS_KINDS", + "weights_declaration", "ShardKey", "InconsistentShardError", "shard_label", diff --git a/src/zagg/grids/healpix.py b/src/zagg/grids/healpix.py index 577f8c857..96d69a98c 100644 --- a/src/zagg/grids/healpix.py +++ b/src/zagg/grids/healpix.py @@ -725,6 +725,7 @@ def _shard(arr): element_dtype=sig["dtype"] or "float32", inner_shape=sig["inner_shape"], locations=located, + weights=sig.get("weights"), **rag_kw, ), meta, diff --git a/src/zagg/grids/rectilinear.py b/src/zagg/grids/rectilinear.py index fe809e166..704753c4c 100644 --- a/src/zagg/grids/rectilinear.py +++ b/src/zagg/grids/rectilinear.py @@ -654,6 +654,7 @@ def _shard(arr): element_dtype=sig["dtype"] or "float32", inner_shape=sig["inner_shape"], locations=located, + weights=sig.get("weights"), **rag_kw, ), meta, diff --git a/src/zagg/pyramid.py b/src/zagg/pyramid.py index 1506957ab..61bc80577 100644 --- a/src/zagg/pyramid.py +++ b/src/zagg/pyramid.py @@ -158,7 +158,7 @@ def declared_fields(config) -> tuple[dict, list]: """ from zagg.config import get_agg_fields from zagg.semantics import EXACT_MERGE_LAWS, _fold_function_name, composability_classes - from zagg.sweep_overview import EXACT_NAN_POLICY, TDIGEST_LAW + from zagg.sweep_overview import EXACT_NAN_POLICY, TDIGEST_LAW, overview_fold_delta agg = get_agg_fields(config) fields: dict = {} @@ -175,13 +175,36 @@ def declared_fields(config) -> tuple[dict, list]: } elif cls == "approximate": inner = meta.get("inner_shape") or (2,) + delta = int((meta.get("params") or {}).get("delta", 512)) fields[name] = { "class": "approximate", "method": TDIGEST_LAW, "dtype": meta.get("dtype", "float32"), "inner_shape": [int(inner)] if isinstance(inner, int) else [int(x) for x in inner], - "delta": int((meta.get("params") or {}).get("delta", 512)), + "delta": delta, + # The split pyramid-fold budget (issue #424), recorded RESOLVED + # so the manifest is self-describing: the sweep folds at this + # value. Pre-#424 manifests lack the key — the reader-side + # capped fallback in overview_fold_delta reproduces their + # historical fold-at-leaf-δ behavior exactly (all carried + # δ ≤ 512). + "overview_delta": overview_fold_delta( + {"delta": delta, "overview_delta": meta.get("overview_delta")} + ), } + # The §2.0 weights declaration, keyed only when non-default so + # existing manifests stay byte-identical; the sweep's fold gate + # compares it against the stored arrays (issue #424). Its + # calibration provenance rides along, because the manifest is the + # ONLY thing the overview writer reconstructs a field from + # (``sweep_overview._overview_config``) and §2.0 makes ``gain`` + # REQUIRED on every flux-declared array — without it the overview + # would be stamped flux with its calibration unrecoverable. + if meta.get("weights") not in (None, "counts"): + fields[name]["weights"] = meta["weights"] + gain = (meta.get("attrs") or {}).get("gain") + if gain is not None: + fields[name]["gain"] = gain else: fields[name] = {"class": "none"} excluded.append(name) diff --git a/src/zagg/readers/tdigest_tensor.py b/src/zagg/readers/tdigest_tensor.py index 458f2ee71..e5a081ee4 100644 --- a/src/zagg/readers/tdigest_tensor.py +++ b/src/zagg/readers/tdigest_tensor.py @@ -80,7 +80,7 @@ import zarr from zarr.abc.store import Store -from zagg.grids.base import RAGGED_ELEMENT_ATTR, RAGGED_SPEC +from zagg.grids.base import RAGGED_ELEMENT_ATTR, RAGGED_SPEC, weights_declaration from zagg.readers._layout import ( normalize_subtree, rank_to_rowcol, @@ -265,6 +265,13 @@ def _open_ragged(store: Store, field: str, zarr_format) -> tuple: readers decode by (never a hardcoded dtype). Raises a pointed error when the attrs are missing/malformed: silently guessing an element layout would misinterpret every payload (pre-issue-209 CSR stores are a hard break). + + The sibling §2.0 ``weights`` declaration is strict-checked here too — the + reader is the party that spec sentence's MUST is addressed to, and this + is the seam gating the other spec markers. Both DEFINED values open: + what the readers bind of flux semantics is issue #426's read-validation + phase, and this only refuses the undefined ones (a future revision of + §2.0, which must never be read as either defined value). """ arr = zarr.open_array(store, path=field, mode="r", zarr_format=zarr_format) raw_meta = arr.attrs.get(RAGGED_ELEMENT_ATTR) @@ -286,6 +293,7 @@ def _open_ragged(store: Store, field: str, zarr_format) -> tuple: f"understands {RAGGED_SPEC!r} only — a newer writer's layout must be " f"adopted deliberately, not half-parsed" ) + weights_declaration(dict(arr.attrs)) # §2.0: refuse an unknown declaration if arr.ndim != 1: raise ValueError( f"{field!r} has {arr.ndim} dimensions; the t-digest tensor readers " @@ -695,6 +703,11 @@ def read_tensors( counts to the nearest integer; ``float32`` keeps fractional counts. A per-bin count exceeding the dtype's max (65535 for ``uint16``) wraps on cast — keep ``uint32`` for dense cells with many observations per bin. + On a field declaring ``weights: "flux"`` (spec §2.0) the weights are + positive reals, not counts, so an integer dtype rounds each bin's + photoelectron estimate away — pass ``"float32"`` for a flux field. + (Rasterization does not otherwise bind flux semantics yet; the flux + read path is issue #426.) block_order : int, optional HEALPix order of the emitted blocks (default ``None`` — one block per read chunk). Must be at or coarser than the chunk order; a block is @@ -761,7 +774,9 @@ def read_tensors( ------ ValueError On an unknown ``dtype``/``fit``, a store missing the ragged element - attrs or the ``morton`` sibling, an out-of-range ``block_order``, a + attrs or the ``morton`` sibling, a payload declaring a §2.0 + ``weights`` value this zagg does not define, an out-of-range + ``block_order``, a block tensor over ``max_block_bytes``, a corrupt/misaligned occupancy sidecar, a ``subtree`` finer than the read chunks (or malformed / too-deep), or (with ``fit="raise"``) a block whose trimmed range diff --git a/src/zagg/semantics.py b/src/zagg/semantics.py index 3f63a4564..b8b4287b9 100644 --- a/src/zagg/semantics.py +++ b/src/zagg/semantics.py @@ -30,7 +30,11 @@ ``chunk_inner``), ``sharded``, store layout/path, ``emit_cell_ids`` (the issue #304 transition hatch), worker sizing, streaming mode, read knobs, catalog/bounds (run inputs, recorded per-run — catalog identity lives in the -D20 sidecar, never the product identity). +D20 sidecar, never the product identity), and the per-variable +``overview_delta`` (issue #424 — the pyramid-fold budget shapes overview +artifacts only, and the overview family is already packaging). A variable's +``weights: "counts"`` normalizes away as the spec §2.0 absent-key default; +``weights: "flux"`` is output-defining and hashes. Canonical form: the core dict serialized as sorted-key, compact, ASCII-escaped JSON — so YAML comments, whitespace, and key order can never @@ -60,6 +64,12 @@ #: (issue #132) transports identical values either way. AGGREGATION_PACKAGING_KEYS = ("handoff",) +#: Per-variable aggregation keys that are packaging (issue #424): +#: ``overview_delta`` shapes the pyramid/overview fold budget only — the +#: overview family is packaging already (``output.pyramid`` never enters the +#: core), so the split budget must not move a base product's identity either. +VARIABLE_PACKAGING_KEYS = ("overview_delta",) + #: Display length of :func:`semantic_fingerprint` (12 hex = 48 bits; the #: birthday bound puts same-store collision odds around 1e-8 at 1e4 products #: — recorded rationale on the issue #299 thread). @@ -171,6 +181,32 @@ def _without(mapping: dict, keys: tuple[str, ...]) -> dict: return {k: v for k, v in (mapping or {}).items() if k not in keys} +def _normalize_variables(aggregation: dict) -> dict: + """Drop per-variable packaging keys and default-valued declarations (#424). + + Two normalizations, both hash-stability obligations: + + - :data:`VARIABLE_PACKAGING_KEYS` (``overview_delta``) drop outright — + overview artifacts are packaging, so declaring the split fold budget + hashes identically to omitting it; + - ``weights: "counts"`` drops because it is the spec §2.0 **absent-key + default** — the explicit and absent spellings mean the same bytes, so + they must be the same product. ``weights: "flux"`` stays: the stored + weight column means something else, which is exactly output-defining. + """ + variables = (aggregation or {}).get("variables") + if not variables: + return aggregation + out = {} + for name, meta in variables.items(): + if isinstance(meta, dict): + meta = {k: v for k, v in meta.items() if k not in VARIABLE_PACKAGING_KEYS} + if meta.get("weights") == "counts": + meta = {k: v for k, v in meta.items() if k != "weights"} + out[name] = meta + return {**aggregation, "variables": out} + + def _prune_nulls(obj): """Recursively drop ``None``-valued keys from every dict in ``obj``. @@ -209,7 +245,9 @@ def semantic_core(config: PipelineConfig) -> dict: if key in grid_cfg: grid[key] = grid_cfg[key] core: dict = { - "aggregation": _without(config.aggregation, AGGREGATION_PACKAGING_KEYS), + "aggregation": _normalize_variables( + _without(config.aggregation, AGGREGATION_PACKAGING_KEYS) + ), "data_source": _without(config.data_source, DATA_SOURCE_PACKAGING_KEYS), "grid": grid, # pipeline.type is output-defining (espg-ruled on the PR #316 @@ -255,6 +293,7 @@ def semantic_fingerprint(digest: str) -> str: "AGGREGATION_PACKAGING_KEYS", "COMPOSABILITY_CLASSES", "DATA_SOURCE_PACKAGING_KEYS", + "VARIABLE_PACKAGING_KEYS", "EXACT_MERGE_LAWS", "FINGERPRINT_HEX", "GRID_SPATIAL_KEYS", diff --git a/src/zagg/sweep_overview.py b/src/zagg/sweep_overview.py index d7c82de02..6b5e028f2 100644 --- a/src/zagg/sweep_overview.py +++ b/src/zagg/sweep_overview.py @@ -191,6 +191,55 @@ def encode_digest(digest: np.ndarray, dtype) -> bytes: return np.ascontiguousarray(np.asarray(digest, dtype=dt)).tobytes() +#: Default cap on the pyramid-fold compression budget (issue #424). Overview +#: digests are governed by the ~1/δ quantile-accuracy bound (512 → ~0.2%), +#: not the leaf's loss-free bound — and the cap also bounds the sweep's +#: chunk-batched k-way fold buffers (4 children × δ × cells), which saturate +#: toward ~1 GB per slab at a δ=8,192 leaf budget vs ~33 MB here. +OVERVIEW_DELTA_CAP = 512 + + +def overview_fold_delta(meta: dict) -> int: + """The δ an overview/pyramid/column fold compresses at (issue #424). + + A declared ``overview_delta`` (manifest field entry / config field key) + wins. Absent — every pre-#424 manifest — the leaf ``delta`` capped at + :data:`OVERVIEW_DELTA_CAP`: identical to the historical fold-at-leaf-δ + behavior for every manifest ever written (all carried δ ≤ 512), while a + raised leaf δ no longer saturates the fold buffers. Leaf-side builds and + streaming/spill folds are NOT overview folds and keep the leaf δ — only + the first fold level ever sees leaf-δ inputs, and those are bounded by + actual observations, not δ. + """ + declared = meta.get("overview_delta") + if declared is not None: + return int(declared) + return min(int(meta.get("delta") or OVERVIEW_DELTA_CAP), OVERVIEW_DELTA_CAP) + + +def check_weights_match(attrs, meta: dict, field: str) -> None: + """Refuse a digest fold across mismatched §2.0 weights declarations (#424). + + Merges are legal only between payloads carrying the same declaration + (spec §2.0): folding a counts payload under a flux declaration (or vice + versa) would produce a weight column whose sum means neither thing. The + manifest's declared value (absent ⇒ counts, like the attrs key itself) is + the store-wide truth; a source array disagreeing with it raises here, + which the per-leaf/per-child fold guards turn into a loud skip rather + than a silent mixed merge. + """ + from zagg.grids.base import weights_declaration + + stored = weights_declaration(dict(attrs or {})) + declared = meta.get("weights") or "counts" + if stored != declared: + raise ValueError( + f"field {field!r}: stored weights declaration {stored!r} does not match " + f"the manifest's {declared!r} — merges are legal only between matching " + f"declarations (spec §2.0, issue #424)" + ) + + def fold_digests(cell_digests: list, *, delta: int, dtype="float32") -> bytes: """Merge one overview cell's accumulated t-digests into its payload bytes. @@ -702,8 +751,15 @@ def _field_drift(group, name, meta) -> str | None: with a worse error. ``zagg-ragged/2`` is a live migration path (issue #210 moves the element declaration into the zarr data type), so this is not hypothetical (espg-ruled, issue #358). + + The same branch gates the §2.0 ``weights`` declaration for the same + reason (issue #424): it is what :func:`check_weights_match` refuses a + fold over, so a disagreement caught here is the retrofit refusing, and + one missed here is every leaf warning at fold time. A stored value this + zagg does not define RAISES out of the probe rather than reporting drift + — that store is unreadable, not merely mis-declared. """ - from zagg.grids.base import RAGGED_ELEMENT_ATTR, RAGGED_SPEC + from zagg.grids.base import RAGGED_ELEMENT_ATTR, RAGGED_SPEC, weights_declaration try: arr = group[name] @@ -739,6 +795,20 @@ def _field_drift(group, name, meta) -> str | None: declared_inner = [int(s) for s in meta.get("inner_shape") or [2]] if stored_inner != declared_inner: return f"field {name!r}: ragged inner_shape {stored_inner} != declared {declared_inner}" + # The §2.0 weights declaration is one more thing a leaf can falsify, + # and it is stamped on the array this probe already opened: absent it + # here, ``declare_pyramid`` installs a flux declaration over a counts + # store cleanly, and every leaf then fails ``check_weights_match`` at + # FOLD time as a per-leaf warning (review finding, issue #424). + # Absent on either side is counts, exactly as the fold gate reads it. + stored_weights = weights_declaration(dict(arr.attrs)) + declared_weights = meta.get("weights") or "counts" + if stored_weights != declared_weights: + return ( + f"field {name!r}: stored weights declaration {stored_weights!r} != " + f"declared {declared_weights!r} — merges are legal only between " + f"matching declarations (spec §2.0)" + ) elif meta["class"] == "exact": declared_dt = np.dtype(meta.get("dtype") or "float32") if arr.dtype != declared_dt: @@ -1306,7 +1376,7 @@ def _fold_node( cell_digests: dict = {} for name, meta in fields.items(): try: - values = group[name][:] + arr = group[name] except KeyError: # Schema evolution: the field postdates this leaf — it # contributes fill, exactly what re-running would write. @@ -1314,9 +1384,13 @@ def _fold_node( continue if meta["class"] == "exact": partials[name] = fold_dense( - values, fold_factor, meta.get("method"), meta.get("fill_value", "NaN") + arr[:], fold_factor, meta.get("method"), meta.get("fill_value", "NaN") ) else: + # Mismatched §2.0 weights declarations refuse to merge + # (issue #424); the enclosing guard skips the leaf loudly. + check_weights_match(dict(arr.attrs), meta, name) + values = arr[:] dtype = meta.get("dtype") or "float32" inner = tuple(meta.get("inner_shape") or (2,)) cell_digests[name] = [ @@ -1350,7 +1424,7 @@ def _fold_node( for j, cell in enumerate(acc): if cell: slab[j] = fold_digests( - cell, delta=int(meta.get("delta") or 512), dtype=meta.get("dtype") or "float32" + cell, delta=overview_fold_delta(meta), dtype=meta.get("dtype") or "float32" ) slabs[name] = slab stamps = [t for t in timestamps if t is not None] @@ -1534,18 +1608,22 @@ def _fold_child(group, fields, factor, span, path) -> dict: partials: dict = {} for name, meta in fields.items(): try: - values = group[name][:] + arr = group[name] except KeyError: logger.debug(f"sweep[overview]: overview {path} lacks field {name!r}") continue if meta["class"] == "exact": partials[name] = fold_dense( - values, factor, meta.get("method"), meta.get("fill_value", "NaN") + arr[:], factor, meta.get("method"), meta.get("fill_value", "NaN") ) continue + # Mismatched §2.0 weights declarations refuse to merge (issue #424); + # the enclosing per-child guard skips the child loudly. + check_weights_match(dict(arr.attrs), meta, name) + values = arr[:] dtype = meta.get("dtype") or "float32" inner = tuple(meta.get("inner_shape") or (2,)) - delta = int(meta.get("delta") or 512) + delta = overview_fold_delta(meta) folded = np.full(span, b"", dtype=object) for j in range(span): cell = [ @@ -1641,6 +1719,15 @@ def _overview_config(fields): The overview zarr reuses the leaf template machinery (``HealpixGrid.emit_shard_template``) so structure — dtypes, fills, the D18 ragged attrs, the D16 dggs attrs — cannot drift from source leaves. + + The §2.0 ``weights`` declaration (and the ``gain`` provenance §2.0 makes + REQUIRED beside it) is carried through from the manifest field entry + (:func:`zagg.pyramid.declared_fields`), because the overview's payload IS + the fold of its sources' weights: a bare template would declare the fold + of a flux store as counts, and :func:`check_weights_match` would then + refuse every child of the cascade (review finding, issue #424). Both keys + are absent on a counts field, so a counts store's template bytes are + unchanged. """ from zagg.config import PipelineConfig @@ -1660,6 +1747,10 @@ def _overview_config(fields): "dtype": meta.get("dtype", "float32"), "fill_value": 0, } + if meta.get("weights") not in (None, "counts"): + variables[name]["weights"] = meta["weights"] + if meta.get("gain") is not None: + variables[name]["attrs"] = {"gain": meta["gain"]} return PipelineConfig( aggregation={ "coordinates": {"morton": {"dtype": "uint64", "fill_value": 0}}, diff --git a/src/zagg/sweep_stage.py b/src/zagg/sweep_stage.py index cd544f990..06ff089a6 100644 --- a/src/zagg/sweep_stage.py +++ b/src/zagg/sweep_stage.py @@ -457,6 +457,7 @@ def _merge_slabs( decode_digest, fold_dense, fold_digests, + overview_fold_delta, ) windows = next((len(row) for row in rows if row is not None), 1) @@ -487,7 +488,7 @@ def _merge_slabs( continue dtype = meta.get("dtype") or "float32" inner = tuple(meta.get("inner_shape") or (2,)) - delta = int(meta.get("delta") or 512) + delta = overview_fold_delta(meta) out = np.full(n_out, b"", dtype=object) pending: dict[int, list] = {} for i, row in enumerate(rows): diff --git a/tests/data/spec/column.expected.json b/tests/data/spec/column.expected.json index ee817c5f9..0f934b7b2 100644 --- a/tests/data/spec/column.expected.json +++ b/tests/data/spec/column.expected.json @@ -233,6 +233,7 @@ "class": "approximate", "method": "tdigest_kway", "delta": 16, + "overview_delta": 16, "dtype": "float32", "inner_shape": [ 2 @@ -252,7 +253,7 @@ } }, "cells_with_data_order": 5, - "generated_at": "2026-08-09T07:45:46+00:00" + "generated_at": "2026-08-11T00:43:29+00:00" }, "commit": { "spec": "morton-hive/1", diff --git a/tests/data/spec/column/1/1/2/1/3/all.pyramid.zarr/zarr.json b/tests/data/spec/column/1/1/2/1/3/all.pyramid.zarr/zarr.json index 29981245e..b5b63d535 100644 --- a/tests/data/spec/column/1/1/2/1/3/all.pyramid.zarr/zarr.json +++ b/tests/data/spec/column/1/1/2/1/3/all.pyramid.zarr/zarr.json @@ -17,6 +17,7 @@ "class": "approximate", "method": "tdigest_kway", "delta": 16, + "overview_delta": 16, "dtype": "float32", "inner_shape": [ 2 @@ -36,14 +37,14 @@ } }, "cells_with_data_order": 5, - "generated_at": "2026-08-09T07:45:46+00:00" + "generated_at": "2026-08-11T00:43:29+00:00" }, "morton_hive_commit": { "spec": "morton-hive/1", "complete": true, "cells_with_data": 3, "granule_count": 1, - "written_at": "2026-08-09T07:45:46+00:00" + "written_at": "2026-08-11T00:43:29+00:00" } }, "zarr_format": 3, diff --git a/tests/data/spec/column/morton_hive.json b/tests/data/spec/column/morton_hive.json index df8d3a2b0..8248027f2 100644 --- a/tests/data/spec/column/morton_hive.json +++ b/tests/data/spec/column/morton_hive.json @@ -67,10 +67,11 @@ "inner_shape": [ 2 ], - "delta": 16 + "delta": 16, + "overview_delta": 16 } } } }, - "generated_at": "2026-08-09T07:45:46+00:00" + "generated_at": "2026-08-11T00:43:29+00:00" } \ No newline at end of file diff --git a/tests/data/spec/flux.expected.json b/tests/data/spec/flux.expected.json new file mode 100644 index 000000000..289460020 --- /dev/null +++ b/tests/data/spec/flux.expected.json @@ -0,0 +1,223 @@ +{ + "shard": "11213", + "leaf": "1/1/2/1/3/11213.zarr", + "group": "6", + "shard_order": 4, + "chunk_order": 5, + "cell_order": 6, + "cells_per_chunk": 4, + "chunks_per_shard": 4, + "empty_chunk": 2, + "delta": 16, + "cells": [ + { + "index": 0, + "morton": "1233986297899515910", + "count": 40, + "rx_flux": [ + [ + 21.400999069213867, + 20.584632873535156 + ], + [ + 22.049999237060547, + 28.51854133605957 + ], + [ + 22.729999542236328, + 27.028770446777344 + ], + [ + 23.6932430267334, + 39.50696563720703 + ], + [ + 25.29913330078125, + 36.73384094238281 + ], + [ + 25.936399459838867, + 47.27507019042969 + ], + [ + 26.602724075317383, + 64.17291259765625 + ], + [ + 27.72444725036621, + 56.30500793457031 + ], + [ + 28.679750442504883, + 47.046207427978516 + ], + [ + 29.173053741455078, + 49.60882568359375 + ], + [ + 29.992942810058594, + 39.97758483886719 + ], + [ + 32.01917266845703, + 58.849674224853516 + ], + [ + 33.08859634399414, + 49.66215515136719 + ], + [ + 34.4471549987793, + 35.73512268066406 + ], + [ + 34.94689178466797, + 30.71946144104004 + ], + [ + 36.68199920654297, + 23.74026107788086 + ], + [ + 37.43299865722656, + 29.927345275878906 + ], + [ + 38.334999084472656, + 9.647828102111816 + ] + ], + "flux_sum": 695.040207862854 + }, + { + "index": 2, + "morton": "1234549247852937222", + "count": 1, + "rx_flux": [ + [ + 35.612998962402344, + 27.34147834777832 + ] + ], + "flux_sum": 27.34147834777832 + }, + { + "index": 5, + "morton": "1235393672783069190", + "count": 5, + "rx_flux": [ + [ + 22.231000900268555, + 0.7394490242004395 + ], + [ + 30.42300033569336, + 6.947474956512451 + ], + [ + 31.219999313354492, + 24.131603240966797 + ], + [ + 35.95800018310547, + 27.82486343383789 + ], + [ + 37.22100067138672, + 8.305474281311035 + ] + ], + "flux_sum": 67.94886493682861 + }, + { + "index": 15, + "morton": "1238208422550175750", + "count": 300, + "rx_flux": [ + [ + 16.290128707885742, + 35.331485748291016 + ], + [ + 19.59087371826172, + 116.78459167480469 + ], + [ + 21.737735748291016, + 172.67822265625 + ], + [ + 23.6151065826416, + 242.83253479003906 + ], + [ + 24.977148056030273, + 298.5267333984375 + ], + [ + 26.191791534423828, + 357.3218994140625 + ], + [ + 27.391592025756836, + 389.447998046875 + ], + [ + 28.501562118530273, + 400.43853759765625 + ], + [ + 29.88144302368164, + 421.423583984375 + ], + [ + 30.997642517089844, + 400.68658447265625 + ], + [ + 32.06013107299805, + 379.5057067871094 + ], + [ + 33.377960205078125, + 333.33795166015625 + ], + [ + 34.73054885864258, + 288.2079772949219 + ], + [ + 37.06555938720703, + 228.9956817626953 + ], + [ + 39.2369384765625, + 156.27883911132812 + ], + [ + 41.98875045776367, + 76.70658874511719 + ], + [ + 43.2875862121582, + 30.34041976928711 + ] + ], + "flux_sum": 4328.8453369140625 + } + ], + "content_hashes": { + "arrays": { + "6/count": "29458cad5550071cc942b81809aa26becc67c09a3e2e209e4876d8782224757b", + "6/morton": "f6282635e373d534ef4d91166d306441049e7bbed3d7d2ec8add306f62274d06", + "6/rx_flux": "3ba141cb29b1771dee4c4c3f2aadec42b7de69e075d4334d6d7187301dc59eb8" + }, + "combined": "910a9b12d34b4d072f454b2dd1cc232a6a9a92b536a8d26f5395154a1c02bc32" + }, + "weights": "flux", + "gain": { + "name": "spec-fixture-gain", + "version": "1" + } +} diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/c/0 b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/c/0 new file mode 100644 index 000000000..c8436f879 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/zarr.json b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/zarr.json new file mode 100644 index 000000000..5e651672e --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/count/zarr.json @@ -0,0 +1,58 @@ +{ + "shape": [ + 16 + ], + "data_type": "int32", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 16 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "sharding_indexed", + "configuration": { + "chunk_shape": [ + 4 + ], + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "index_codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + }, + { + "name": "crc32c" + } + ], + "index_location": "end" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/c/0 b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/c/0 new file mode 100644 index 000000000..a14256bfb Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/zarr.json b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/zarr.json new file mode 100644 index 000000000..0b83eaa9a --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/morton/zarr.json @@ -0,0 +1,58 @@ +{ + "shape": [ + 16 + ], + "data_type": "uint64", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 16 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "sharding_indexed", + "configuration": { + "chunk_shape": [ + 4 + ], + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "index_codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + }, + { + "name": "crc32c" + } + ], + "index_location": "end" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/c/0 b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/c/0 new file mode 100644 index 000000000..00e4c9834 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/zarr.json b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/zarr.json new file mode 100644 index 000000000..fc89ad231 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/rx_flux/zarr.json @@ -0,0 +1,79 @@ +{ + "shape": [ + 16 + ], + "data_type": "variable_length_bytes", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 16 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": "", + "codecs": [ + { + "name": "sharding_indexed", + "configuration": { + "chunk_shape": [ + 4 + ], + "codecs": [ + { + "name": "vlen-bytes", + "configuration": {} + }, + { + "name": "zstd", + "configuration": { + "level": 3, + "checksum": false + } + } + ], + "index_codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + }, + { + "name": "crc32c" + } + ], + "index_location": "end" + } + } + ], + "attributes": { + "ragged": { + "spec": "zagg-ragged/1", + "element": { + "dtype": "float32", + "shape": [ + -1, + 2 + ] + } + }, + "weights": "flux", + "gain": { + "name": "spec-fixture-gain", + "version": "1" + } + }, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/zarr.json b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/zarr.json new file mode 100644 index 000000000..c87640057 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/6/zarr.json @@ -0,0 +1,34 @@ +{ + "attributes": { + "zarr_conventions": [ + { + "schema_url": "https://raw.githubusercontent.com/zarr-conventions/dggs/refs/tags/v1/schema.json", + "spec_url": "https://github.com/zarr-conventions/dggs/blob/v1/README.md", + "uuid": "7b255807-140c-42ca-97f6-7a1cfecdbc38", + "name": "dggs", + "description": "Discrete Global Grid Systems convention for zarr" + }, + { + "schema_url": "https://github.com/espg/mortie/blob/main/docs/specification.md#dggs-attrs", + "spec_url": "https://github.com/espg/mortie/blob/main/docs/specification.md", + "uuid": "3e22156d-ea9e-4e01-95fe-e3809a4b41e7", + "name": "morton-dggs", + "description": "Packed-u64 morton (HEALPix) DGGS convention" + } + ], + "dggs": { + "name": "morton", + "refinement_level": 6, + "spatial_dimension": "cells", + "ellipsoid": { + "name": "WGS84", + "semimajor_axis": 6378137.0, + "inverse_flattening": 298.257223563 + }, + "compression": "none", + "coordinate": "morton" + } + }, + "zarr_format": 3, + "node_type": "group" +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/coverage.moc b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/coverage.moc new file mode 100644 index 000000000..e9144276b Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/coverage.moc differ diff --git a/tests/data/spec/flux/1/1/2/1/3/11213.zarr/zarr.json b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/zarr.json new file mode 100644 index 000000000..7deae99d9 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/11213.zarr/zarr.json @@ -0,0 +1,28 @@ +{ + "attributes": { + "morton_hive_commit": { + "spec": "morton-hive/1", + "complete": true, + "cells_with_data": 4, + "granule_count": 1, + "written_at": "2026-08-11T00:11:08+00:00", + "coverage": { + "spec": "morton-moc/1", + "box": [ + "112131", + "1121322", + "1121344", + null + ], + "cell_order": 6, + "source": "worker", + "encoding": "bitmap", + "sidecar": "coverage.moc", + "nbytes": 11, + "raw_nbytes": 2 + } + } + }, + "zarr_format": 3, + "node_type": "group" +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.stats.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.stats.json new file mode 100644 index 000000000..081205eb0 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.stats.json @@ -0,0 +1 @@ +{"schema_version": 1, "shard_key": 1233986297899515908, "window": null, "run_id": null, "semantic_hash": null, "zagg_version": "0.43.1.dev270+g1509fee07", "n_shards": 1, "n_granules": 1, "granules_sha256": null, "content_hashes": {"arrays": {"4/count": "71f269d578d413efdf624d699f2d8430be995bd75dba859bd9a782e7c89c3118", "4/morton": "92cedd52796c6c0844af1ce3e7e7636ed28804f6bf8f9d10ceab93edf1a29805", "4/rx_flux": "426567515966563f313618e5551f4c6c66fa925af305dbf8fff53541aaf57170", "5/count": "6380596d3ca826d3ccda1b07f368ca3b43b486b7aaa0ca17e055ac8bf0b2c064", "5/morton": "2d7a07b3e2db265b2050fedfd968fb6763f5b012c87033526852bffc90b059d2", "5/rx_flux": "02b03bafbfdd26d6f51611513cba41bc01c64f37684184cf0a6f39ddcff4c9f4"}, "combined": "06663de3e1b4cc32c28644c79c69c18eb62831431371b202da16ccbcec1e0403"}, "n_obs": 0, "n_obs_read": null, "cells_with_data": 3, "phase_timings": {}, "duration_s": 0.0, "spill_bytes": null, "spill_blocks_closed": null, "raster_bytes_read": null, "raster_px_decoded": null, "raster_px_sampled": null, "leaf_column": null, "gb_seconds": null, "est_cost_usd": null, "max_memory_mb": null, "container_hwm_mb": null, "lambda": null, "timestamp": "2026-08-11T00:11:08+00:00", "success": true, "error": null, "invoked_by": null} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/c/0 new file mode 100644 index 000000000..541655da1 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/zarr.json new file mode 100644 index 000000000..e07607926 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/count/zarr.json @@ -0,0 +1,36 @@ +{ + "shape": [ + 1 + ], + "data_type": "int32", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 1 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/c/0 new file mode 100644 index 000000000..d3f352aa5 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/zarr.json new file mode 100644 index 000000000..f8043e85a --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/morton/zarr.json @@ -0,0 +1,36 @@ +{ + "shape": [ + 1 + ], + "data_type": "uint64", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 1 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/c/0 new file mode 100644 index 000000000..de1e85732 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/zarr.json new file mode 100644 index 000000000..68ece70d0 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/rx_flux/zarr.json @@ -0,0 +1,57 @@ +{ + "shape": [ + 1 + ], + "data_type": "variable_length_bytes", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 1 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": "", + "codecs": [ + { + "name": "vlen-bytes", + "configuration": {} + }, + { + "name": "zstd", + "configuration": { + "level": 3, + "checksum": false + } + } + ], + "attributes": { + "ragged": { + "spec": "zagg-ragged/1", + "element": { + "dtype": "float32", + "shape": [ + -1, + 2 + ] + } + }, + "weights": "flux", + "gain": { + "name": "spec-fixture-gain", + "version": "1" + } + }, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/zarr.json new file mode 100644 index 000000000..bcf2c16e4 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/4/zarr.json @@ -0,0 +1,34 @@ +{ + "attributes": { + "zarr_conventions": [ + { + "schema_url": "https://raw.githubusercontent.com/zarr-conventions/dggs/refs/tags/v1/schema.json", + "spec_url": "https://github.com/zarr-conventions/dggs/blob/v1/README.md", + "uuid": "7b255807-140c-42ca-97f6-7a1cfecdbc38", + "name": "dggs", + "description": "Discrete Global Grid Systems convention for zarr" + }, + { + "schema_url": "https://github.com/espg/mortie/blob/main/docs/specification.md#dggs-attrs", + "spec_url": "https://github.com/espg/mortie/blob/main/docs/specification.md", + "uuid": "3e22156d-ea9e-4e01-95fe-e3809a4b41e7", + "name": "morton-dggs", + "description": "Packed-u64 morton (HEALPix) DGGS convention" + } + ], + "dggs": { + "name": "morton", + "refinement_level": 4, + "spatial_dimension": "cells", + "ellipsoid": { + "name": "WGS84", + "semimajor_axis": 6378137.0, + "inverse_flattening": 298.257223563 + }, + "compression": "none", + "coordinate": "morton" + } + }, + "zarr_format": 3, + "node_type": "group" +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/c/0 new file mode 100644 index 000000000..19cc6c8c8 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/zarr.json new file mode 100644 index 000000000..98dc8036c --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/count/zarr.json @@ -0,0 +1,36 @@ +{ + "shape": [ + 4 + ], + "data_type": "int32", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 4 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/c/0 new file mode 100644 index 000000000..b9d8dc2f4 Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/zarr.json new file mode 100644 index 000000000..87f93bddf --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/morton/zarr.json @@ -0,0 +1,36 @@ +{ + "shape": [ + 4 + ], + "data_type": "uint64", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 4 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": 0, + "codecs": [ + { + "name": "bytes", + "configuration": { + "endian": "little" + } + } + ], + "attributes": {}, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/c/0 b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/c/0 new file mode 100644 index 000000000..bb8ecfd7b Binary files /dev/null and b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/c/0 differ diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/zarr.json new file mode 100644 index 000000000..1193e2897 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/rx_flux/zarr.json @@ -0,0 +1,57 @@ +{ + "shape": [ + 4 + ], + "data_type": "variable_length_bytes", + "chunk_grid": { + "name": "regular", + "configuration": { + "chunk_shape": [ + 4 + ] + } + }, + "chunk_key_encoding": { + "name": "default", + "configuration": { + "separator": "/" + } + }, + "fill_value": "", + "codecs": [ + { + "name": "vlen-bytes", + "configuration": {} + }, + { + "name": "zstd", + "configuration": { + "level": 3, + "checksum": false + } + } + ], + "attributes": { + "ragged": { + "spec": "zagg-ragged/1", + "element": { + "dtype": "float32", + "shape": [ + -1, + 2 + ] + } + }, + "weights": "flux", + "gain": { + "name": "spec-fixture-gain", + "version": "1" + } + }, + "dimension_names": [ + "cells" + ], + "zarr_format": 3, + "node_type": "array", + "storage_transformers": [] +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/zarr.json new file mode 100644 index 000000000..140deffad --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/5/zarr.json @@ -0,0 +1,34 @@ +{ + "attributes": { + "zarr_conventions": [ + { + "schema_url": "https://raw.githubusercontent.com/zarr-conventions/dggs/refs/tags/v1/schema.json", + "spec_url": "https://github.com/zarr-conventions/dggs/blob/v1/README.md", + "uuid": "7b255807-140c-42ca-97f6-7a1cfecdbc38", + "name": "dggs", + "description": "Discrete Global Grid Systems convention for zarr" + }, + { + "schema_url": "https://github.com/espg/mortie/blob/main/docs/specification.md#dggs-attrs", + "spec_url": "https://github.com/espg/mortie/blob/main/docs/specification.md", + "uuid": "3e22156d-ea9e-4e01-95fe-e3809a4b41e7", + "name": "morton-dggs", + "description": "Packed-u64 morton (HEALPix) DGGS convention" + } + ], + "dggs": { + "name": "morton", + "refinement_level": 5, + "spatial_dimension": "cells", + "ellipsoid": { + "name": "WGS84", + "semimajor_axis": 6378137.0, + "inverse_flattening": 298.257223563 + }, + "compression": "none", + "coordinate": "morton" + } + }, + "zarr_format": 3, + "node_type": "group" +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/zarr.json b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/zarr.json new file mode 100644 index 000000000..429364d15 --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/all.pyramid.zarr/zarr.json @@ -0,0 +1,52 @@ +{ + "attributes": { + "role": "column", + "zagg_column": { + "spec": "zagg-column/1", + "node": "11213", + "order": 4, + "source_cell_order": 6, + "window": "all", + "fields": { + "count": { + "class": "exact", + "method": "sum", + "nan_policy": "skip" + }, + "rx_flux": { + "class": "approximate", + "method": "tdigest_kway", + "delta": 16, + "overview_delta": 16, + "dtype": "float32", + "inner_shape": [ + 2 + ] + } + }, + "groups": { + "5": { + "regime": "leaf-column", + "merges_from_raw": 1, + "n_cells": 4 + }, + "4": { + "regime": "leaf-column", + "merges_from_raw": 1, + "n_cells": 1 + } + }, + "cells_with_data_order": 5, + "generated_at": "2026-08-11T00:41:18+00:00" + }, + "morton_hive_commit": { + "spec": "morton-hive/1", + "complete": true, + "cells_with_data": 3, + "granule_count": 1, + "written_at": "2026-08-11T00:41:18+00:00" + } + }, + "zarr_format": 3, + "node_type": "group" +} \ No newline at end of file diff --git a/tests/data/spec/flux/1/1/2/1/3/granules.json b/tests/data/spec/flux/1/1/2/1/3/granules.json new file mode 100644 index 000000000..cb3e11dea --- /dev/null +++ b/tests/data/spec/flux/1/1/2/1/3/granules.json @@ -0,0 +1 @@ +{"spec": "zagg-granule-ids/1", "granules_sha256": "037fe2f1b68f3d5199685428fe5be1550fe7aaccd4afd0b3feff6ea07c5f05dd", "granule_ids": ["s3://fixture/a.h5"]} \ No newline at end of file diff --git a/tests/data/spec/flux/morton_hive.json b/tests/data/spec/flux/morton_hive.json new file mode 100644 index 000000000..86a94914d --- /dev/null +++ b/tests/data/spec/flux/morton_hive.json @@ -0,0 +1,82 @@ +{ + "spec": "morton-hive/1", + "dataset": { + "short_name": "SPEC_FIXTURE", + "version": "1" + }, + "semantic_hash": "c86c276252bdda6f5d9857df06076ba0dcc6602d108dee36e71bdf4e41140962", + "cell_order": 6, + "shard_order": 4, + "split_schedule": [ + 1, + 1, + 1, + 1 + ], + "path_grouping": 1, + "pyramid": { + "spec": "zagg-pyramid/2", + "overviews": [ + { + "node": 4, + "cells": [ + 5 + ] + }, + { + "node": 3, + "cells": [ + 4 + ] + }, + { + "node": 2, + "cells": [ + 3 + ] + }, + { + "node": 1, + "cells": [ + 2 + ] + }, + { + "node": 0, + "cells": [ + 1 + ] + } + ], + "overview": { + "all_time": false, + "fold_source": "cascade", + "exact_levels": 1, + "fields": { + "count": { + "class": "exact", + "method": "sum", + "nan_policy": "skip", + "dtype": "int32", + "fill_value": 0 + }, + "rx_flux": { + "class": "approximate", + "method": "tdigest_kway", + "dtype": "float32", + "inner_shape": [ + 2 + ], + "delta": 16, + "overview_delta": 16, + "weights": "flux", + "gain": { + "name": "spec-fixture-gain", + "version": "1" + } + } + } + } + }, + "generated_at": "2026-08-11T00:41:18+00:00" +} \ No newline at end of file diff --git a/tests/test_column.py b/tests/test_column.py index 9fffee5fd..d0f3bd369 100644 --- a/tests/test_column.py +++ b/tests/test_column.py @@ -431,11 +431,14 @@ def test_role_and_provenance_attrs(self, tmp_path): assert set(attrs["fields"]) == {"count", "h_min", "h_tdigest"} assert attrs["fields"]["count"] == {"class": "exact", "method": "sum", "nan_policy": "skip"} # An approximate entry also carries what decided its centroid bytes: - # a #370 k-way gather cannot recover δ from the leaf. + # a #370 k-way gather cannot recover δ from the leaf. Since issue #424 + # that is the split overview_delta (here the DELTA≤512 fallback), the + # budget fold_column actually compressed at; delta stays the leaf's. assert attrs["fields"]["h_tdigest"] == { "class": "approximate", "method": "tdigest_kway", "delta": DELTA, + "overview_delta": DELTA, "dtype": "float32", "inner_shape": [2], } diff --git a/tests/test_config.py b/tests/test_config.py index 30ee5d42a..094d8f57e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1852,6 +1852,7 @@ def test_scalar_signature(self): "dtype": "float32", "resolution": "cell", "location": None, + "weights": None, } def test_scalar_default_dtype_none(self): @@ -1863,6 +1864,7 @@ def test_scalar_default_dtype_none(self): "dtype": None, "resolution": "cell", "location": None, + "weights": None, } def test_vector_int_signature(self): @@ -1874,6 +1876,7 @@ def test_vector_int_signature(self): "dtype": "float32", "resolution": "cell", "location": None, + "weights": None, } def test_vector_list_signature(self): @@ -2135,6 +2138,7 @@ def test_get_output_signature_ragged(self): "dtype": "float32", "resolution": "cell", "location": None, + "weights": None, } def test_ragged_inner_shape_int_normalized(self): @@ -2236,6 +2240,106 @@ def test_output_field_signature_scalar_inner_shape_empty(self, atl06_config): assert e["inner_shape"] == [], f"{e['name']!r} has non-empty inner_shape" +# --------------------------------------------------------------------------- +# Packaged δ = 8,192 raise (issues #414/#424) +# --------------------------------------------------------------------------- + + +class TestPackagedDeltaRaise: + """The four shipped ATL03 t-digest configs carry the split budgets. + + Leaf δ = 8,192 is the measured loss-free bound (issue #422's statewide CA + scan); overview_delta = 512 is the pyramid-fold accuracy budget. The + packaged configs are EXPLICIT so ``_DEFAULT_DELTA`` never decides an + output value for them (the silent-change-under-stable-hash trap). + """ + + @pytest.mark.parametrize( + "name", + [ + "atl03_tdigest_healpix", + "atl03_tdigest_healpix_hive", + "atl03_tdigest_located_healpix", + "atl03_tdigest_strata_healpix", + ], + ) + def test_digest_fields_declare_the_split_budgets(self, name): + cfg = default_config(name) + ragged = { + f: meta + for f, meta in cfg.aggregation["variables"].items() + if meta.get("kind") == "ragged" + } + assert ragged # every one of these templates carries digest fields + for meta in ragged.values(): + assert meta["params"]["delta"] == 8192 + assert meta["overview_delta"] == 512 + + +# --------------------------------------------------------------------------- +# Ragged weights declaration + overview_delta (spec §2.0, issue #424) +# --------------------------------------------------------------------------- + + +_FLUX_GAIN = {"gain": {"name": "test_gain", "version": "1"}} + + +class TestWeightsDeclaration: + def test_counts_validates(self): + validate_config(_ragged_cfg(inner_shape=[2], weights="counts")) + + def test_flux_with_gain_provenance_validates(self): + validate_config(_ragged_cfg(inner_shape=[2], weights="flux", attrs=dict(_FLUX_GAIN))) + + def test_flux_requires_gain_provenance(self): + with pytest.raises(ValueError, match="requires calibration provenance"): + validate_config(_ragged_cfg(inner_shape=[2], weights="flux")) + + def test_flux_gain_requires_name_and_version(self): + with pytest.raises(ValueError, match="requires calibration provenance"): + validate_config( + _ragged_cfg(inner_shape=[2], weights="flux", attrs={"gain": {"name": "g"}}) + ) + + def test_unknown_value_rejected(self): + with pytest.raises(ValueError, match="is not one of"): + validate_config(_ragged_cfg(inner_shape=[2], weights="photons")) + + def test_weights_rejected_on_non_ragged_kinds(self): + with pytest.raises(ValueError, match="'weights' is only valid for kind 'ragged'"): + _validate_output_kind("f", {"function": "min", "weights": "counts"}) + with pytest.raises(ValueError, match="'weights' is only valid for kind 'ragged'"): + _validate_output_kind("f", {"kind": "vector", "trailing_shape": 4, "weights": "counts"}) + + def test_attrs_weights_key_is_spec_owned_on_ragged(self): + # The template stamps the §2.0 key from the field declaration; an + # attrs transcription could silently disagree, so it is reserved. + with pytest.raises(ValueError, match="spec-owned"): + validate_config(_ragged_cfg(inner_shape=[2], attrs={"weights": "flux"})) + + def test_signature_carries_weights_only_when_set(self): + entries = output_field_signature( + _ragged_cfg(inner_shape=[2], weights="flux", attrs=dict(_FLUX_GAIN)) + ) + assert entries[0]["weights"] == "flux" + # Undeclared: keyed-only-when-set, so existing signatures are stable. + assert "weights" not in output_field_signature(_ragged_cfg(inner_shape=[2]))[0] + + +class TestOverviewDelta: + def test_valid_overview_delta_validates(self): + validate_config(_ragged_cfg(inner_shape=[2], overview_delta=512)) + + @pytest.mark.parametrize("bad", [0, -1, 2.5, "512", True]) + def test_invalid_overview_delta_rejected(self, bad): + with pytest.raises(ValueError, match="overview_delta must be a positive int"): + validate_config(_ragged_cfg(inner_shape=[2], overview_delta=bad)) + + def test_overview_delta_rejected_on_non_ragged_kinds(self): + with pytest.raises(ValueError, match="'overview_delta' is only valid for kind 'ragged'"): + _validate_output_kind("f", {"function": "min", "overview_delta": 512}) + + # --------------------------------------------------------------------------- # Ragged location channel (issue #87) # --------------------------------------------------------------------------- diff --git a/tests/test_grids.py b/tests/test_grids.py index cf7ab60e4..ae6945af0 100644 --- a/tests/test_grids.py +++ b/tests/test_grids.py @@ -1587,6 +1587,73 @@ def test_undeclared_fields_unchanged(self): assert dict(m["count"].attributes) == {} +class TestWeightsStamp: + """The §2.0 weights declaration is template-stamped (issue #424).""" + + def _grid(self, meta_extra): + from zagg.config import PipelineConfig + + cfg = PipelineConfig( + aggregation={ + "variables": { + "rx_flux": { + "kind": "ragged", + "function": "zagg.stats.tdigest.build_tdigest", + "source": "h_li", + "inner_shape": [2], + "dtype": "float32", + "params": {"delta": 64}, + **meta_extra, + }, + } + } + ) + return HealpixGrid(parent_order=4, child_order=8, chunk_inner=6, sharded=True, config=cfg) + + def test_flux_stamped_as_sibling_of_the_ragged_block(self): + attrs = dict( + self._grid({"weights": "flux", "attrs": {"gain": {"name": "g", "version": "1"}}}) + .spec() + .members["rx_flux"] + .attributes + ) + assert attrs["weights"] == "flux" + # A sibling key, never inside the versioned block (the /2 migration + # retires the block wholesale — espg ruling, issue #422). + assert "weights" not in attrs["ragged"] + assert attrs["ragged"]["spec"] == "zagg-ragged/1" + # The declared provenance attrs ride alongside. + assert attrs["gain"] == {"name": "g", "version": "1"} + + def test_counts_stamped_when_declared(self): + attrs = dict(self._grid({"weights": "counts"}).spec().members["rx_flux"].attributes) + assert attrs["weights"] == "counts" + + def test_undeclared_emits_no_key(self): + # Absent config declaration -> absent attrs key: pre-#424 configs + # produce byte-identical templates (spec §2.0 reads absence as counts). + attrs = dict(self._grid({}).spec().members["rx_flux"].attributes) + assert "weights" not in attrs + + def test_located_sibling_carries_no_weights_key(self): + attrs = ( + self._grid({"weights": "counts", "location": "leaf_id"}) + .spec() + .members["rx_flux_locations"] + .attributes + ) + assert "weights" not in dict(attrs) + + def test_weights_declaration_helper(self): + from zagg.grids.base import weights_declaration + + assert weights_declaration({}) == "counts" + assert weights_declaration(None) == "counts" + assert weights_declaration({"weights": "flux"}) == "flux" + with pytest.raises(ValueError, match="unknown weights declaration"): + weights_declaration({"weights": "photons"}) + + class TestCoverageBbox: """Shard-complete fetch regions (the AOI-bbox under-fetch gap).""" diff --git a/tests/test_readers.py b/tests/test_readers.py index 9c79cfd0a..c6b6e8618 100644 --- a/tests/test_readers.py +++ b/tests/test_readers.py @@ -510,6 +510,30 @@ def test_unknown_spec_raises_pointed(self): with pytest.raises(ValueError, match="understands 'zagg-ragged/1' only"): list(read_tensors(store, "12/h_tdigest")) + def test_unknown_weights_declaration_raises(self): + # Spec §2.0 (issue #424): an undefined declaration is a future + # revision of that section and MUST be refused, never read as either + # defined value. The reader is the party that MUST is addressed to. + store, _g, _w = _build_store({_KEY_A: {0: np.array([1.0, 2.0])}}) + arr = zarr.open_array(store, path="12/h_tdigest", mode="r+") + arr.attrs["weights"] = "photons" + with pytest.raises(ValueError, match="unknown weights declaration"): + list(read_tensors(store, "12/h_tdigest")) + with pytest.raises(ValueError, match="unknown weights declaration"): + read_cell(store, "12/h_tdigest", 0) + + def test_defined_weights_declarations_open(self): + # Both DEFINED values read: what the readers BIND of flux semantics + # is issue #426's read-validation phase, so the gate refuses only the + # undefined ones (an absent key is "counts" — every other test here). + rng = np.random.default_rng(19) + for declared in ("counts", "flux"): + store, _g, _w = _build_store({_KEY_A: {0: rng.uniform(0.0, 30.0, 500)}}) + arr = zarr.open_array(store, path="12/h_tdigest", mode="r+") + arr.attrs["weights"] = declared + blocks = list(read_tensors(store, "12/h_tdigest", dtype="float32")) + assert len(blocks) == 1 + def test_garbage_element_dtype_raises_pointed(self): # A corrupt dtype string surfaces as this layout's pointed error, not # a bare numpy TypeError with no mention of the field or contract. diff --git a/tests/test_semantics.py b/tests/test_semantics.py index 6250eb220..0a6b72811 100644 --- a/tests/test_semantics.py +++ b/tests/test_semantics.py @@ -233,3 +233,52 @@ def test_pipeline_type_is_semantic(self): def test_fingerprint_rejects_short_input(self): with pytest.raises(ValueError, match="not a semantic hash"): semantic_fingerprint("abc") + + +def _digest_cfg(**meta_extra) -> PipelineConfig: + """A minimal config carrying one ragged digest field ``d``.""" + return PipelineConfig( + data_source={ + "reader": "h5coro", + "groups": ["g"], + "variables": {"h": "/p"}, + "coordinates": {"latitude": "/lat", "longitude": "/lon"}, + }, + aggregation={ + "variables": { + "d": { + "kind": "ragged", + "function": "zagg.stats.tdigest.build_tdigest", + "source": "h", + "inner_shape": [2], + "dtype": "float32", + "params": {"delta": 8192}, + **meta_extra, + } + } + }, + output={"grid": {"type": "healpix", "parent_order": 6, "child_order": 12}}, + ) + + +class TestWeightsAndOverviewDeltaHashing: + """Issue #424: default weights and overview_delta never move a hash.""" + + def test_weights_counts_hashes_as_absent(self): + # The §2.0 absent-key default: explicit and absent spellings are the + # same bytes, so they must be the same product identity. + assert semantic_hash(_digest_cfg(weights="counts")) == semantic_hash(_digest_cfg()) + + def test_weights_flux_is_semantic(self): + gain = {"gain": {"name": "g", "version": "1"}} + base = _digest_cfg(attrs=dict(gain)) + flux = _digest_cfg(weights="flux", attrs=dict(gain)) + assert semantic_hash(flux) != semantic_hash(base) + assert semantic_core(flux)["aggregation"]["variables"]["d"]["weights"] == "flux" + + def test_overview_delta_is_packaging(self): + # The pyramid-fold budget shapes overview artifacts only; the base + # product identity must not move when it is declared (issue #424). + assert semantic_hash(_digest_cfg(overview_delta=512)) == semantic_hash(_digest_cfg()) + core = semantic_core(_digest_cfg(overview_delta=512)) + assert "overview_delta" not in core["aggregation"]["variables"]["d"] diff --git a/tests/test_spec_conformance.py b/tests/test_spec_conformance.py index 4c427aed7..a41d84188 100644 --- a/tests/test_spec_conformance.py +++ b/tests/test_spec_conformance.py @@ -50,6 +50,7 @@ ("kitchen_sink", "h_tdigest_noise", "float32", (2,)), ("kitchen_sink", "h_tdigest_signal_locations", "uint64", ()), ("kitchen_sink", "h_tdigest_noise_locations", "uint64", ()), + ("flux", "rx_flux", "float32", (2,)), ] SENTINEL = 2**64 - 1 @@ -66,6 +67,9 @@ # column/'s LEAF is minimal's, byte for byte — the same literal is the # cross-fixture pin that writing a column perturbs nothing. "column": "2f4ff37de621de05962ab720cec05fd643757977f1afbd0e859ca588a143b72e", + # The §2.0 flux fixture (issue #424) — asserted by TestFluxDeclaration + # (flux is not in FIXTURES: the leaf-shaped suite hardcodes h_tdigest). + "flux": "910a9b12d34b4d072f454b2dd1cc232a6a9a92b536a8d26f5395154a1c02bc32", } #: The same pin over the §4.6 COLUMN artifact of the ``column/`` fixture (not #: its leaf, which is ``minimal``'s): the only committed store whose §5 key @@ -84,6 +88,7 @@ "kitchen_sink", "6/composition", ): "04886a9dfb60c8a53de48202dc3d5ac694698864825426f084be961aa678acd5", + ("flux", "6/rx_flux"): "3ba141cb29b1771dee4c4c3f2aadec42b7de69e075d4334d6d7187301dc59eb8", } @@ -306,6 +311,94 @@ def test_read_locations_binds_through_attrs(self): assert got == want +class TestFluxDeclaration: + """§2.0 — the weights declaration, pinned on the committed `flux/` store. + + The counts side of the contract needs no new fixture: `minimal/` predates + §2.0 and is deliberately not regenerated, so its absent `weights` key IS + the committed absent-key ⇒ counts pin (issue #424). + """ + + def test_weights_key_is_a_sibling_of_the_ragged_block(self): + exp = _expected("flux") + attrs = _array_meta("flux", exp, "rx_flux")["attributes"] + assert attrs["weights"] == "flux" + # A sibling key, never inside the versioned block: the block is + # retired wholesale under /2, a sibling survives the migration. + assert "weights" not in attrs["ragged"] + assert attrs["ragged"]["spec"] == "zagg-ragged/1" + + def test_calibration_provenance_recorded(self): + exp = _expected("flux") + attrs = _array_meta("flux", exp, "rx_flux")["attributes"] + assert attrs["gain"] == exp["gain"] + assert {"name", "version"} <= set(attrs["gain"]) + + def test_fold_arrays_re_declare_weights_and_gain(self): + """A fold's own payload arrays carry §2.0 through (review, issue #424). + + The overview/column writer reconstructs its template from the manifest + field entry alone, so a declaration that stops at the leaf leaves every + folded array reading as ``counts`` — and the fold gate then refuses the + whole cascade above the finest level. + """ + exp = _expected("flux") + column = _leaf_dir("flux", exp).parent / "all.pyramid.zarr" + groups = sorted(p.name for p in column.iterdir() if p.is_dir()) + assert groups # the fixture's §4.6 column, folded from the flux leaf + for group in groups: + meta = json.loads((column / group / "rx_flux" / "zarr.json").read_text()) + attrs = meta["attributes"] + assert attrs["weights"] == "flux" + assert attrs["gain"] == exp["gain"] + assert "weights" not in attrs["ragged"] # sibling key, never inside + + def test_absent_key_reads_as_counts_on_the_committed_minimal(self): + from zagg.grids.base import weights_declaration + + exp = _expected("minimal") + attrs = _array_meta("minimal", exp, "h_tdigest")["attributes"] + assert "weights" not in attrs + assert weights_declaration(attrs) == "counts" + + def test_flux_weights_are_positive_fractional_reals(self): + exp = _expected("flux") + store = _leaf_store("flux", exp) + non_integer = 0 + for cell in exp["cells"]: + got = read_cell(store, f"{exp['group']}/rx_flux", cell["index"]) + want = np.array(cell["rx_flux"], dtype=np.float32).reshape(-1, 2) + np.testing.assert_array_equal(got, want) + assert np.all(got[:, 1] > 0) # §2.0: positive, no zero-weight rows + assert np.all(np.diff(got[:, 0]) >= 0) # §2.1 ordering unchanged + total = float(got[:, 1].astype(np.float64).sum()) + assert total == cell["flux_sum"] + non_integer += int(total != round(total)) + assert non_integer # flux sums are photoelectron estimates, not counts + + def test_leaf_is_stamped_and_manifest_marked(self): + exp = _expected("flux") + attrs = json.loads((_leaf_dir("flux", exp) / "zarr.json").read_text())["attributes"] + assert attrs["morton_hive_commit"]["complete"] is True + manifest = json.loads((SPEC_DATA / "flux" / "morton_hive.json").read_text()) + assert manifest["spec"] == "morton-hive/1" + + def test_per_array_hashes_match_golden(self): + exp = _expected("flux") + got = TestContentHashes._hash_leaf(_leaf_dir("flux", exp)) + assert got == exp["content_hashes"]["arrays"] + + def test_frozen_digests_pin_the_recipe(self): + # flux is not in FIXTURES (the leaf-shaped suite hardcodes h_tdigest), + # so its frozen literals are asserted here. + exp = _expected("flux") + assert exp["content_hashes"]["combined"] == FROZEN_COMBINED["flux"] + arrays = exp["content_hashes"]["arrays"] + assert arrays["6/rx_flux"] == FROZEN_ARRAYS[("flux", "6/rx_flux")] + combined = hashlib.sha256("\n".join(sorted(arrays.values())).encode()).hexdigest() + assert combined == exp["content_hashes"]["combined"] + + class TestCoordinateAndDense: """§1.1 — the `morton` coordinate and the dense `count` field, off the store. @@ -763,6 +856,14 @@ def test_basename_role_and_attrs_grammar(self): assert block["node"] == exp["shard"] and block["order"] == exp["shard_order"] assert block["source_cell_order"] == exp["cell_order"] assert block["window"] == "all" + # §4.6's decode-without-the-manifest keys on an approximate entry — + # including ``overview_delta``, the budget this column's fold actually + # compressed at (issue #424), which the committed bytes must carry for + # a spec-and-fixtures reader to see it. + entry = block["fields"]["h_tdigest"] + assert entry["class"] == "approximate" and entry["method"] == "tdigest_kway" + assert {"delta", "overview_delta", "dtype", "inner_shape"} <= set(entry) + assert entry["overview_delta"] == exp["delta"] def test_stamp_is_present_and_field_pinned(self): exp = _expected(COLUMN) diff --git a/tests/test_spill.py b/tests/test_spill.py index 56e63c2b4..fd5ca8c6c 100644 --- a/tests/test_spill.py +++ b/tests/test_spill.py @@ -459,6 +459,14 @@ def test_removed_arena_knobs_rejected_with_spill(self): with pytest.raises(ValueError, match="unknown key"): get_streaming(_config(streaming={"mode": "spill", "state_layout": "arena"})) + def test_spill_fold_state_carries_the_declared_delta(self): + # Issue #424: the spill fold honors the field's declared δ (the 8,192 + # leaf budget), never the module default. + cfg = _config(streaming={"mode": "spill"}, variables=_base_variables(delta=8192)) + agg = SpillAggregator(cfg, _grid(cfg), "pandas", 25) + assert agg._digest_fields["h_tdigest"].delta == 8192 + agg.close() + def test_non_mergeable_config_is_accepted_by_spill(self): # The whole point of spill: reducers merge-mode validation rejects are # exact in the single-block regime. diff --git a/tests/test_streaming.py b/tests/test_streaming.py index cc2a8719f..d4fbe4385 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -404,6 +404,15 @@ def test_occupied_out_fed_under_streaming(self, monkeypatch): np.testing.assert_array_equal(np.sort(occ["merge"]), np.sort(occ["pooled"])) +class TestDeclaredDeltaPropagation: + """Issue #424: streaming folds honor the field's declared δ, not the default.""" + + def test_streaming_state_carries_the_declared_delta(self): + cfg = _config(streaming={"buffer_granules": 2}, delta=8192) + agg = StreamingAggregator(cfg, _grid(cfg), "pandas", 2) + assert agg._digest_fields["h_tdigest"] == ("h_ph", 8192) + + class TestStreamingReviewFolds: """Folds from the phase-4 adversarial review.""" diff --git a/tests/test_sweep_overview.py b/tests/test_sweep_overview.py index bbbc97693..6840e908a 100644 --- a/tests/test_sweep_overview.py +++ b/tests/test_sweep_overview.py @@ -352,6 +352,162 @@ def test_merge_is_permutation_stable(self): assert forward == backward # the issue #279 order-independent law +class TestOverviewFoldDelta: + """Issue #424: the split leaf-δ / overview-δ fold budget.""" + + def test_declared_budget_wins(self): + from zagg.sweep_overview import overview_fold_delta + + assert overview_fold_delta({"delta": 8192, "overview_delta": 512}) == 512 + assert overview_fold_delta({"delta": 64, "overview_delta": 1024}) == 1024 + + def test_absent_reproduces_every_historical_manifest(self): + # Every manifest ever written carried δ ≤ 512, so the capped fallback + # is byte-identical to the old fold-at-leaf-δ behavior for all of them. + from zagg.sweep_overview import overview_fold_delta + + assert overview_fold_delta({"delta": 256}) == 256 + assert overview_fold_delta({"delta": 16}) == 16 + assert overview_fold_delta({}) == 512 + + def test_absent_caps_a_raised_leaf_delta(self): + # A δ=8,192 leaf must not saturate the sweep's k-way fold buffers + # through an old-style manifest: the cap bounds it. + from zagg.sweep_overview import OVERVIEW_DELTA_CAP, overview_fold_delta + + assert overview_fold_delta({"delta": 8192}) == OVERVIEW_DELTA_CAP == 512 + + def test_declared_fields_records_the_resolved_budget(self): + from zagg.config import PipelineConfig + from zagg.pyramid import declared_fields + + def cfg(**extra): + return PipelineConfig( + data_source={ + "variables": {"h": "/p"}, + "coordinates": {"latitude": "/lat", "longitude": "/lon"}, + }, + aggregation={ + "variables": { + "d": { + "kind": "ragged", + "function": "zagg.stats.tdigest.build_tdigest", + "source": "h", + "inner_shape": [2], + "params": {"delta": 8192}, + **extra, + } + } + }, + output={"grid": {"type": "healpix", "parent_order": 6, "child_order": 12}}, + ) + + declared = declared_fields(cfg(overview_delta=256))[0]["d"] + assert declared["delta"] == 8192 and declared["overview_delta"] == 256 + # Undeclared resolves to the capped fallback, recorded explicitly so + # the manifest is self-describing. + resolved = declared_fields(cfg())[0]["d"] + assert resolved["delta"] == 8192 and resolved["overview_delta"] == 512 + + +class TestWeightsGate: + """Spec §2.0 (issue #424): merges refuse across mismatched declarations.""" + + def test_matching_defaults_pass(self): + from zagg.sweep_overview import check_weights_match + + # Absent on both sides reads as counts on both sides. + check_weights_match({}, {"class": "approximate"}, "d") + check_weights_match(None, {}, "d") + check_weights_match({"weights": "flux"}, {"weights": "flux"}, "d") + + def test_mismatch_refuses_both_ways(self): + from zagg.sweep_overview import check_weights_match + + with pytest.raises(ValueError, match="matching\\s+declarations"): + check_weights_match({"weights": "flux"}, {"class": "approximate"}, "d") + with pytest.raises(ValueError, match="matching\\s+declarations"): + check_weights_match({}, {"weights": "flux"}, "d") + + def test_unknown_stored_declaration_refuses(self): + from zagg.sweep_overview import check_weights_match + + with pytest.raises(ValueError, match="unknown weights declaration"): + check_weights_match({"weights": "photons"}, {}, "d") + + def test_declared_fields_records_flux_only(self): + from zagg.config import PipelineConfig + from zagg.pyramid import declared_fields + + def cfg(**extra): + return PipelineConfig( + data_source={ + "variables": {"h": "/p"}, + "coordinates": {"latitude": "/lat", "longitude": "/lon"}, + }, + aggregation={ + "variables": { + "d": { + "kind": "ragged", + "function": "zagg.stats.tdigest.build_tdigest", + "source": "h", + "inner_shape": [2], + "params": {"delta": 64}, + **extra, + } + } + }, + output={"grid": {"type": "healpix", "parent_order": 6, "child_order": 12}}, + ) + + # counts (explicit or absent) records nothing — existing manifests + # stay byte-identical; flux is recorded for the sweep's fold gate. + assert "weights" not in declared_fields(cfg())[0]["d"] + assert "weights" not in declared_fields(cfg(weights="counts"))[0]["d"] + flux = cfg(weights="flux", attrs={"gain": {"name": "g", "version": "1"}}) + assert declared_fields(flux)[0]["d"]["weights"] == "flux" + # §2.0 makes gain REQUIRED beside a flux declaration, and the manifest + # entry is all the overview writer reconstructs a field from. + assert declared_fields(flux)[0]["d"]["gain"] == {"name": "g", "version": "1"} + + def test_overview_template_stamps_the_flux_declaration(self, tmp_path): + """The reconstructed overview template carries §2.0 through (#424). + + Without it a flux store's overview declares counts, and the cascade's + per-child gate then refuses every child (review finding). + """ + from zagg.grids.healpix import HealpixGrid + from zagg.sweep_overview import _overview_config, check_weights_match + + gain = {"name": "rx_gain", "version": "3"} + base = { + "class": "approximate", + "method": "tdigest_kway", + "dtype": "float32", + "inner_shape": [2], + "delta": 64, + "overview_delta": 64, + } + fields = { + "flux_d": {**base, "weights": "flux", "gain": gain}, + "counts_d": dict(base), + } + grid = HealpixGrid(2, 4, config=_overview_config(fields), sharded=True) + grid.emit_shard_template(open_store(str(tmp_path / "ov.zarr")), overwrite=True) + group = zarr.open_group( + open_store(str(tmp_path / "ov.zarr")), path="4", mode="r", zarr_format=3 + ) + assert group["flux_d"].attrs["weights"] == "flux" + assert dict(group["flux_d"].attrs["gain"]) == gain + # A counts field declares nothing — a counts store's template bytes + # are unchanged by this path. + assert "weights" not in dict(group["counts_d"].attrs) + assert "gain" not in dict(group["counts_d"].attrs) + # And the fold gate now accepts the overview as a cascade source. + check_weights_match(dict(group["flux_d"].attrs), fields["flux_d"], "flux_d") + check_weights_match(dict(group["counts_d"].attrs), fields["counts_d"], "counts_d") + + class TestPyramidBlock: """The manifest declaration (Phase C): template time + config grammar.""" @@ -400,7 +556,10 @@ def test_tdigest_field_declaration_carries_delta(self): block = build_pyramid_block(cfg, shard_order=9) entry = block["overview"]["fields"]["h_tdigest"] assert entry["class"] == "approximate" and entry["method"] == "tdigest_kway" - assert entry["inner_shape"] == [2] and entry["delta"] == 256 + # The packaged split budgets (issues #414/#424): leaf δ 8,192 + # (loss-free bound), overview folds at 512 (accuracy bound). + assert entry["inner_shape"] == [2] and entry["delta"] == 8192 + assert entry["overview_delta"] == 512 def test_explicit_orders_and_all_time(self): from zagg.sweep_overview import build_pyramid_block @@ -1973,6 +2132,42 @@ def test_ragged_inner_shape_drift_refuses(self, tmp_path): with pytest.raises(ValueError, match="h_tdigest.*inner_shape"): declare_pyramid(str(tmp_path), _leaf_cfg()) + def test_flux_declaration_over_a_counts_store_refuses(self, tmp_path): + # The operator-facing case the gate exists for (review, issue #424): + # without it the retrofit PUTs the flux declaration clean, and the + # next sweep logs "skipping unreadable leaf" for every leaf. + self._pre_declaration_store(tmp_path) + cfg = _leaf_cfg() + cfg.aggregation["variables"]["h_tdigest"].update( + weights="flux", attrs={"gain": {"name": "g", "version": "1"}} + ) + with pytest.raises(ValueError, match="h_tdigest.*weights declaration"): + declare_pyramid(str(tmp_path), cfg) + assert "pyramid" not in read_manifest(str(tmp_path)) # nothing written + + def test_store_side_weights_drift_refuses(self, tmp_path): + # The other direction: a flux-stamped leaf under a counts config. + from zarr import open_array + + self._pre_declaration_store(tmp_path) + leaf = open_store(shard_leaf_path(str(tmp_path), morton_word("-311"))) + arr = open_array(leaf, path=f"{CELL_ORDER}/h_tdigest", mode="r+", zarr_format=3) + arr.attrs["weights"] = "flux" + with pytest.raises(ValueError, match="h_tdigest.*weights declaration"): + declare_pyramid(str(tmp_path), _leaf_cfg()) + assert "pyramid" not in read_manifest(str(tmp_path)) + + def test_undefined_stored_weights_raises(self, tmp_path): + # A value §2.0 does not define: unreadable, not merely mis-declared. + from zarr import open_array + + self._pre_declaration_store(tmp_path) + leaf = open_store(shard_leaf_path(str(tmp_path), morton_word("-311"))) + arr = open_array(leaf, path=f"{CELL_ORDER}/h_tdigest", mode="r+", zarr_format=3) + arr.attrs["weights"] = "photons" + with pytest.raises(ValueError, match="unknown weights declaration"): + declare_pyramid(str(tmp_path), _leaf_cfg()) + def test_unreadable_ragged_spec_revision_refuses(self, tmp_path): # espg-ruled (issue #358): a store whose ragged layout THIS zagg cannot # read must be refused at declaration, not discovered at fold time. diff --git a/tests/test_tdigest.py b/tests/test_tdigest.py index b8a8f1981..44402f9a9 100644 --- a/tests/test_tdigest.py +++ b/tests/test_tdigest.py @@ -788,3 +788,50 @@ def test_located_output_identical_to_standard(self): d_s, l_s = build_tdigest(vals, delta=32, locations=locs) np.testing.assert_array_equal(d_p, d_s) np.testing.assert_array_equal(l_p, l_s) + + +class TestSingletonPreservation: + """Issue #424: loss-free below δ, pinned at the exact boundary n = δ. + + The δ = 8,192 raise's whole justification is "no original observation is + merged away while a cell's count stays ≤ δ" — these tests make that + load-bearing for the build AND both merge paths, so a future ``_compress`` + change that eagerly re-compresses below saturation fails here. + """ + + def _assert_all_singletons(self, digest, source_values): + assert digest.shape == (len(source_values), 2) + np.testing.assert_array_equal(digest[:, 1], np.ones(len(source_values), dtype=np.float32)) + np.testing.assert_array_equal(digest[:, 0], np.sort(source_values).astype(np.float32)) + + def test_build_is_loss_free_at_n_equals_delta(self): + rng = np.random.default_rng(424) + values = rng.normal(0.0, 100.0, 512) + self._assert_all_singletons(build_tdigest(values, delta=512), values) + + def test_pairwise_merge_keeps_every_singleton_at_combined_delta(self): + # Two loss-free digests over overlapping ranges, combined n == δ: the + # merge re-runs the same greedy rule over 512 unit sub-centroids, so + # no eager re-compression below saturation may exist. + rng = np.random.default_rng(4242) + a_vals = rng.normal(0.0, 100.0, 300) + b_vals = rng.normal(0.0, 100.0, 212) + a = build_tdigest(a_vals, delta=512) + b = build_tdigest(b_vals, delta=512) + merged = merge_tdigests(a, b, delta=512) + self._assert_all_singletons(merged, np.concatenate([a_vals, b_vals])) + + def test_kway_merge_keeps_every_singleton_at_combined_delta(self): + rng = np.random.default_rng(42424) + parts = [rng.normal(0.0, 100.0, 128) for _ in range(4)] + digests = [build_tdigest(p, delta=512) for p in parts] + merged = merge_tdigests_kway(digests, delta=512) + self._assert_all_singletons(merged, np.concatenate(parts)) + + def test_default_delta_unchanged(self): + # Raising the default would silently change output values under an + # unchanged semantic hash for any config omitting ``delta`` — the + # packaged configs are explicit instead (issue #424, plan Q2 ruling). + from zagg.stats.tdigest import _DEFAULT_DELTA + + assert _DEFAULT_DELTA == 512 diff --git a/tools/generate_spec_fixtures.py b/tools/generate_spec_fixtures.py index 25b482005..e03a299d1 100644 --- a/tools/generate_spec_fixtures.py +++ b/tools/generate_spec_fixtures.py @@ -32,6 +32,14 @@ (the `atl03_tdigest_strata_healpix.yaml` field shapes), including a single-photon cell that packs the §3.1 golden word `0xFF000000FF0000FF` and a noise-only cell whose signal payload is the empty ``(0, 2)`` array. +- ``flux/`` — the §2.0 weights-declaration surface (issue #424): one + flux-declared digest field (`rx_flux`, ``weights: flux`` stamped as the + SIBLING attrs key beside the ``ragged`` block, ``gain`` provenance attrs) + + `count`. Payload weights are fractional positive reals built through the + real merge algebra, so per-cell weight sums are NOT integers — the pin + that a flux reader must not round-trip weights through counts. The + committed ``minimal/`` (which predates §2.0 and is not regenerated) is the + absent-key ⇒ counts pin. - ``column/`` — the ``minimal`` inputs plus an explicit ``output.pyramid.overviews: 5`` knob, so the SAME worker invocation also writes the §4.6 leaf column (issue #383): ``all.pyramid.zarr`` beside the @@ -133,6 +141,11 @@ #: ``{order: fold_source}`` shape the production bookkeeping writer takes. PYRAMID_V1_ACTUALS = {1: "leaves", 0: "cascade"} +#: The ``flux/`` fixture's §2.0 calibration provenance (issue #424): flux +#: weights are meaningless without the gain constant that produced them, so +#: the spec requires name + version in the payload array's attrs. +FLUX_GAIN = {"name": "spec-fixture-gain", "version": "1"} + def _cell_photons(rng, n, *, signal_frac=0.6): """Synthetic photons: heights + 5 confidence columns, ``n`` rows.""" @@ -161,13 +174,49 @@ def _point_words(grid, cell_word, n, rng): return morton_words(MortonIndexArray.from_latlon(lats, lons, points=True)) -def _config(kitchen_sink: bool, pyramid: dict | None = None): +def _flux_digest(h, rng): + """A §2.0 flux payload built through the real merge algebra (issue #424). + + Fractional positive weights (photoelectron-scale uniforms), sorted + sub-centroids, k-way merged at :data:`DELTA` — so the committed payload + exercises `_compress` over WEIGHTED sub-centroids and its per-cell weight + sums are not integers. A single sample passes through uncompressed (the + kway single-contributor law), pinning the singleton fractional weight. + """ + from zagg.stats.tdigest import merge_tdigests_kway + + h = np.asarray(h, dtype=np.float64) + weights = rng.uniform(0.5, 30.0, len(h)) + order = np.argsort(h, kind="stable") + sub = np.column_stack([h[order], weights[order]]).astype(np.float32) + if len(sub) < 2: + return sub + # Two sorted halves through the k-way fold: a real weighted compression. + return merge_tdigests_kway([sub[0::2], sub[1::2]], delta=DELTA) + + +def _config(kitchen_sink: bool, pyramid: dict | None = None, flux: bool = False): from zagg.config import PipelineConfig variables: dict = { "count": {"function": "len", "source": "h", "dtype": "int32", "fill_value": 0} } - if kitchen_sink: + if flux: + variables["rx_flux"] = { + "kind": "ragged", + # The generator feeds precomputed payloads through the write path + # (the fake process_shard below), so the declared reducer is never + # exercised here — the real flux transform lands with issue #425. + "function": "zagg.stats.tdigest.build_tdigest", + "source": "h", + "inner_shape": [2], + "dtype": "float32", + "fill_value": 0, + "params": {"delta": DELTA}, + "weights": "flux", + "attrs": {"gain": dict(FLUX_GAIN)}, + } + elif kitchen_sink: for stratum in ("signal", "noise"): variables[f"h_tdigest_{stratum}"] = { "kind": "ragged", @@ -236,7 +285,7 @@ def _config(kitchen_sink: bool, pyramid: dict | None = None): ) -def _build_cells(grid, shard, kitchen_sink: bool): +def _build_cells(grid, shard, kitchen_sink: bool, flux: bool = False): """Per-cell synthetic inputs and their expected decoded values. Returns ``(by_chunk, expected_cells)`` where ``by_chunk`` maps a chunk @@ -266,7 +315,14 @@ def _build_cells(grid, shard, kitchen_sink: bool): conf[:] = rng.integers(-2, 2, conf.shape) record: dict = {"index": cell_index, "morton": str(cell_word), "count": n} fields: dict = {"count": n} - if kitchen_sink: + if flux: + digest = _flux_digest(h, rng) + fields["rx_flux"] = digest + record["rx_flux"] = [[float(m), float(w)] for m, w in digest] + # The weight total (float64 sum over the float32 weights), recorded + # so the non-integer pin is a committed expectation, not derived. + record["flux_sum"] = float(digest[:, 1].astype(np.float64).sum()) + elif kitchen_sink: words = _point_words(grid, cell_word, n, rng) signal = (conf >= 2).any(axis=1) for stratum, mask in (("signal", signal), ("noise", ~signal)): @@ -295,7 +351,7 @@ def _build_cells(grid, shard, kitchen_sink: bool): return by_chunk, expected -def _fake_process_shard(grid, by_chunk, kitchen_sink: bool): +def _fake_process_shard(grid, by_chunk, kitchen_sink: bool, ragged_field: str = "h_tdigest"): """A ``process_shard`` stand-in feeding the REAL sharded leaf write path.""" def fake(g, shard_key, urls, **kwargs): @@ -325,7 +381,7 @@ def fake(g, shard_key, urls, **kwargs): ) else: ids = sorted(cells) - ragged["h_tdigest"] = ([cells[i]["h_tdigest"] for i in ids], ids) + ragged[ragged_field] = ([cells[i][ragged_field] for i in ids], ids) kwargs["chunk_results"].append((block, df, ragged)) occupied.extend(int(children[i]) for i in sorted(cells)) kwargs["occupied_out"].append(np.asarray(occupied, dtype=np.uint64)) @@ -393,16 +449,16 @@ def _o11_hashes(leaf_path: str) -> dict: return {"arrays": dict(sorted(hashes.items())), "combined": combined} -def build(out: Path, kitchen_sink: bool, pyramid: dict | None = None) -> None: +def build(out: Path, kitchen_sink: bool, pyramid: dict | None = None, flux: bool = False) -> None: import zagg.processing as processing from zagg import hive from zagg.grids import HealpixGrid from zagg.grids.morton import morton_word - cfg = _config(kitchen_sink, pyramid=pyramid) + cfg = _config(kitchen_sink, pyramid=pyramid, flux=flux) grid = HealpixGrid(4, 6, layout="fullsphere", config=cfg, chunk_inner=5, sharded=True) shard = morton_word(SHARD_KEY) - by_chunk, expected_cells = _build_cells(grid, shard, kitchen_sink) + by_chunk, expected_cells = _build_cells(grid, shard, kitchen_sink, flux=flux) assert EMPTY_CHUNK not in by_chunk if out.exists(): @@ -414,7 +470,9 @@ def build(out: Path, kitchen_sink: bool, pyramid: dict | None = None) -> None: hive.build_manifest(grid, dataset={"short_name": "SPEC_FIXTURE", "version": "1"}), ) original = processing.process_shard - processing.process_shard = _fake_process_shard(grid, by_chunk, kitchen_sink) + processing.process_shard = _fake_process_shard( + grid, by_chunk, kitchen_sink, ragged_field="rx_flux" if flux else "h_tdigest" + ) try: meta = hive.process_and_write_hive( shard, @@ -444,6 +502,11 @@ def build(out: Path, kitchen_sink: bool, pyramid: dict | None = None) -> None: "cells": expected_cells, "content_hashes": _o11_hashes(str(out / leaf_rel)), } + if flux: + # The §2.0 declaration + provenance the conformance tests assert + # against the committed array attrs (issue #424). + expected["weights"] = "flux" + expected["gain"] = dict(FLUX_GAIN) if pyramid is not None: # The §4.6 column fixture (issue #383): the SAME worker invocation # wrote the leaf's column; record the raw knob, the decoded groups, @@ -614,11 +677,27 @@ def main() -> None: type=Path, default=Path(__file__).resolve().parent.parent / "tests" / "data" / "spec", ) + parser.add_argument( + "--only", + nargs="*", + default=None, + help="fixture names to (re)generate (default: all). The older-era " + "fixtures are stale by design (see module docstring) — regenerate " + "only what you mean to commit.", + ) args = parser.parse_args() - build(args.out / "minimal", kitchen_sink=False) - build(args.out / "kitchen_sink", kitchen_sink=True) - build(args.out / "column", kitchen_sink=False, pyramid={"overviews": 5}) - build_pyramid(args.out / "pyramid") + builders = { + "minimal": lambda: build(args.out / "minimal", kitchen_sink=False), + "kitchen_sink": lambda: build(args.out / "kitchen_sink", kitchen_sink=True), + "column": lambda: build(args.out / "column", kitchen_sink=False, pyramid={"overviews": 5}), + "pyramid": lambda: build_pyramid(args.out / "pyramid"), + "flux": lambda: build(args.out / "flux", kitchen_sink=False, flux=True), + } + unknown = set(args.only or ()) - set(builders) + if unknown: + parser.error(f"unknown fixture name(s) {sorted(unknown)} (known: {sorted(builders)})") + for name in args.only if args.only is not None else builders: + builders[name]() if __name__ == "__main__":