diff --git a/benchmarks/test_spool_benchmarks.py b/benchmarks/test_spool_benchmarks.py index 3a95dbdb2..71d5165ce 100644 --- a/benchmarks/test_spool_benchmarks.py +++ b/benchmarks/test_spool_benchmarks.py @@ -268,3 +268,39 @@ def test_merge_many_patches(self, many_patches): merged = dc.spool(many_patches).chunk(time=None) assert len(merged) == 1 assert isinstance(merged[0], dc.Patch) + + +class TestLargePlanBenchmarks: + """ + Benchmarks for plan construction at the scale an archive reaches. + + The other plan benchmarks stop at 100 patches, where per-member cost + is invisible. These run at thousands so a change in how an output's + metadata is decided shows up as time rather than noise. Planning is + lazy, so nothing here loads data: `len` realizes the plan and stops. + """ + + @pytest.fixture(scope="class") + def large_spool(self): + """A spool of several thousand patches which meet end to end.""" + return dc.spool(_make_contiguous_patches(4000, shape=(10, 20), time_step=0.01)) + + @pytest.mark.benchmark + def test_merge_plan_many_members(self, large_spool): + """Time planning one output out of thousands of members.""" + assert len(large_spool.chunk(time=None)) == 1 + + @pytest.mark.benchmark + def test_segment_plan_many_outputs(self, large_spool): + """Time planning many outputs, each cut from the merged span.""" + assert len(large_spool.chunk(time=0.4)) > 100 + + @pytest.mark.benchmark + def test_concatenate_plan_many_members(self, large_spool): + """Time planning a concatenation which joins every member.""" + assert len(large_spool.concatenate(time=None)) == 1 + + @pytest.mark.benchmark + def test_concatenate_plan_many_outputs(self, large_spool): + """Time planning a concatenation which groups members in pairs.""" + assert len(large_spool.concatenate(time=2)) == 2000 diff --git a/dascore/core/coord_join.py b/dascore/core/coord_join.py new file mode 100644 index 000000000..0d3757798 --- /dev/null +++ b/dascore/core/coord_join.py @@ -0,0 +1,177 @@ +""" +Predicting what joining coordinates will produce, from their summaries. + +A lazy plan describes each output patch before any patch is loaded. What +that description says about a coordinate must be what assembly will +actually build, or the catalog and the patch tell different stories. + +The way to guarantee that is to decide it *once*: this module rebuilds +each member's coordinate from the summary the index stored and runs the +same [`concat_coords`](`dascore.core.coords.concat_coords`) call assembly +runs, then states the result as a summary again. Nothing here reimplements +a joining rule; where a rule cannot be applied to summaries alone the +answer is None, which means "claim nothing", never a guess. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import numpy as np + +from dascore.core.coords import CoordSummary, concat_coords, get_coord +from dascore.exceptions import CoordError +from dascore.units import get_quantity +from dascore.utils.misc import get_middle_value + + +def join_summaries( + summaries: Sequence[CoordSummary], + *, + snap_tolerance: float | None = None, +) -> CoordSummary | None: + """ + Return the summary of the coordinate joining these members would give. + + Parameters + ---------- + summaries + The members' coordinate summaries, in the order they will be + joined. They are trusted to describe validated coordinates, as + the index's do. + snap_tolerance + Multiplied by the step to bound how far + [`simplify`](`dascore.core.coords.BaseCoord.simplify`) may move a + value when absorbing a seam, matching what the assembler passes. + None performs only exact simplifications. + + Returns + ------- + The joined summary, or None when the join cannot be decided from + summaries alone — a member which states no step (an array, a + segmented or value-less coordinate, labels), members of different + kinds or units, or values which overlap. The caller then states only + what it can prove of its own accord. + + Examples + -------- + >>> from dascore.core.coords import get_coord + >>> from dascore.core.coord_join import join_summaries + >>> + >>> first = get_coord(start=0.0, stop=10.0, step=1.0) + >>> second = get_coord(start=10.0, stop=20.0, step=1.0) + >>> joined = join_summaries([first.to_summary(), second.to_summary()]) + >>> assert joined.min == 0.0 and joined.max == 19.0 + """ + if not summaries: + return None + if len(summaries) == 1: + return summaries[0] + if not all(x.is_range_like and x.len for x in summaries): + # Values the summary does not carry cannot be joined without + # reading them, and reading them is what laziness avoids. + return None + spellings = {get_quantity(x.units) for x in summaries if x.units is not None} + if len(spellings) > 1: + # One physical coordinate spelled two ways: which spelling the + # output speaks is assembly's choice, made on the values. + return None + coords = [x.to_coord(on_grid=True) for x in summaries] + joining = coords + if spellings and any(x.units is None for x in summaries): + # A member stating no units is not a member disagreeing about + # them: `_concatenate_group` picks a spelling and every unitless + # member adopts it, its numbers unchanged. Only the copies being + # joined adopt it, so each member is still checked against the + # summary it was actually made from. + spoken = next(x.units for x in summaries if x.units is not None) + joining = [x if x.units is not None else x.set_units(spoken) for x in coords] + try: + joined = concat_coords(*joining) + except CoordError: + # Overlapping, contradictory, or otherwise unjoinable members; + # loading them will raise, and the row must not pretend otherwise. + return None + if snap_tolerance: + step = joined.step if joined.step is not None else _middle_step(joining) + if step is not None: + joined = joined.simplify(snap_tolerance * np.abs(step)) + stated = joined.to_summary() + if not _rebuilt_faithfully(summaries, coords): + # A member which does not rebuild into the coordinate it was made + # from — one written at a precision the index does not store, say + # — cannot have the join's identity computed from it. The step + # goes with the identity: a summary which still looked evenly + # sampled would have the identity recomputed from the very values + # which did not survive. The envelope holds either way. + stated = stated.model_copy(update=dict(fingerprint=None, step=None, len=None)) + return stated + + +def raw_join_summary( + summaries: Sequence[CoordSummary], joined: CoordSummary +) -> CoordSummary: + """ + What laying the members' own values end to end actually gives. + + A raw concatenation hands those values to + [`get_coord`](`dascore.core.coords.get_coord`), which reads the step + back off them, while a fused range is generated from a single start. + The two agree in exact arithmetic, so integer and datetime grids are + returned untouched; floating members generated from their own starts + can drift from the fused grid inside their span even where every + boundary matches, and the coordinate which loads is then a different + one with a different step and identity. + + Widths promote the same way: a fused range takes the first member's + dtype, while `np.concatenate` gives an int32 laid before an int64 + back as int64. Where either happens the values are already in hand, + so the answer is built from them rather than guessed at. + """ + promoted = _promoted_dtype(summaries) + if joined.step is None or not joined.len: + # nothing structural is claimed either way, but the width still + # is: what loads is the concatenation, whatever shape it has + return joined.model_copy(update=dict(dtype=str(promoted))) + same_dtype = promoted == np.dtype(joined.dtype) + if same_dtype and np.dtype(joined.dtype).kind != "f": + return joined # computed exactly, so there is nothing to check + raw = np.concatenate([x.to_coord(on_grid=True).values for x in summaries]).astype( + promoted + ) + if same_dtype and np.array_equal(raw, joined.to_coord(on_grid=True).values): + return joined + stated = get_coord(values=raw, units=joined.units).to_summary() + return stated.model_copy(update=dict(dims=joined.dims)) + + +def _promoted_dtype(summaries: Sequence[CoordSummary]): + """ + The dtype `np.concatenate` gives these members. + + A fused range takes the first member's dtype; laying the values end + to end promotes them, so an int32 followed by an int64 loads as + int64 -- a different coordinate with a different identity. + """ + # every member here is range-like, and a range states its dtype + return np.result_type(*[np.dtype(x.dtype) for x in summaries]) + + +def _rebuilt_faithfully(summaries, coords) -> bool: + """Whether every member came back as the coordinate it was made from.""" + return all( + x.fingerprint is None or x.fingerprint == y.fingerprint() + for x, y in zip(summaries, coords, strict=True) + ) + + +def _middle_step(coords): + """ + The step a tolerance is scaled by when the join has none of its own. + + The middle of the members' steps, which is what the assembler scales + by (`dascore.utils.patch._middle_step`); a wider one would absorb a + seam here that the loaded patch keeps. + """ + steps = [x.step for x in coords if x.step is not None] + return get_middle_value(steps) if steps else None diff --git a/dascore/core/coords.py b/dascore/core/coords.py index ac4d3b4fa..e04ffb294 100644 --- a/dascore/core/coords.py +++ b/dascore/core/coords.py @@ -218,6 +218,35 @@ def _scalar_dtype(dtype: np.dtype, name: str) -> np.dtype: return np.dtype(f"timedelta64[{unit}]") if name == "step" else dtype +def _grid_range(start, step, length: int, units, fields_set=None, cls=None): + """ + Build a CoordRange on an exactly known grid, skipping re-validation. + + `CoordRange.validate_start_stop_step_len` exists to *derive* shape and a + normalized stop from loosely specified inputs. Callers here already know + the sample count exactly, so re-deriving costs ~60us and can only + reproduce what is passed in. Only use this where start/step/length come + from an already-validated coordinate; anything taking user input must go + through the validating constructor. + """ + # Mirror check_time_units, which forces time-like coords to seconds. + # Note it tests `start` for truthiness, so a coord starting at exactly + # zero is left alone; that quirk is reproduced here deliberately. + if start and (is_timedelta64(start) or is_datetime64(start)): + units = _second_quantity() + return (cls or CoordRange).model_construct( + # copy; model_construct stores the set by reference. + _fields_set=set(fields_set) if fields_set else {"start", "stop", "step"}, + units=units, + step=step, + shape=(length,), + # matches what the validator stores for dtype. + dtype=np.asarray(start + step).dtype, + start=start, + stop=start + step * length, + ) + + class CoordSummary(DascoreBaseModel): """ A summary for coordinates. @@ -281,13 +310,29 @@ def _derive_dtype_if_unset(self) -> Self: object.__setattr__(self, "dtype", str(dtype).split("[")[0]) return self - def to_coord(self) -> CoordRange: - """Convert to coord range, if possible.""" + def to_coord(self, *, on_grid: bool = False) -> CoordRange: + """ + Convert to coord range, if possible. + + Parameters + ---------- + on_grid + When True the summary is trusted to describe a grid exactly — + `len` samples of `step` starting at `min` — and the range is + built without re-deriving what it already states, which costs + about 60us a call (see `CoordRange._new_grid`). Only pass this + for a summary which came from a validated coordinate, such as + one the index stored; anything taking user input must not. + """ if not self.is_range_like: msg = "Cannot convert summary which is not evenly sampled to coord." raise CoordError(msg) step = self.step assert step is not None # is_range_like above rules out a null step + if on_grid and self.len: + # a reverse coord runs from its max + start = self.max if np.sign(step) == -1 else self.min + return _grid_range(start, step, self.len, self.units) # this is a reverse coord if np.sign(step) == -1: start, stop = self.max, self.min + step @@ -1509,22 +1554,8 @@ def _new_grid(self, start, step, length: int) -> Self: CoordRange and length is computed from indices; anything taking user input must go through the validating constructor. """ - units = self.units - # Mirror check_time_units, which forces time-like coords to seconds. - # Note it tests `start` for truthiness, so a coord starting at exactly - # zero is left alone; that quirk is reproduced here deliberately. - if start and (is_timedelta64(start) or is_datetime64(start)): - units = _second_quantity() - return self.model_construct( - # copy; model_construct stores the set by reference. - _fields_set=set(self.model_fields_set), - units=units, - step=step, - shape=(length,), - # matches what the validator stores for dtype. - dtype=np.asarray(start + step).dtype, - start=start, - stop=start + step * length, + return _grid_range( + start, step, length, self.units, self.model_fields_set, type(self) ) @model_validator(mode="before") @@ -2118,27 +2149,64 @@ def _maybe_promote_segment(seg: BaseCoord) -> BaseCoord: return seg +def _reproduces(fused: CoordRange, run: list) -> bool: + """Whether one grid lands on every boundary the pieces state.""" + offset = 0 + for piece in run: + if fused[offset] != piece.start or fused.stop != run[-1].stop: + return False + offset += len(piece) + return True + + def _fuse_segments(segments: tuple[BaseCoord, ...]) -> tuple[BaseCoord, ...]: - """Fuse adjacent segments that continue exactly (normal form).""" - out = [segments[0]] + """ + Fuse adjacent segments that continue exactly (normal form). + + Runs are gathered before anything is built. A merge of many + contiguous pieces is one run, and the coordinate covering it is + constructed once rather than once per piece, which a long merge + would otherwise pay thousands of times. + """ + out: list[BaseCoord] = [] + run: list[BaseCoord] = [segments[0]] + + def flush() -> None: + """Add the run gathered so far as a single coordinate.""" + if len(run) == 1: + out.append(run[0]) + elif isinstance(run[0], CoordRange): + # Every piece sits on one grid, so the whole run does too and + # its length is theirs added up. Floating point addition is + # not associative, though, so the rebuilt grid is checked + # against the boundaries it must reproduce; where it cannot, + # the pieces stay as they were rather than being altered. + length = sum(len(x) for x in run) + fused = run[0]._new_grid(run[0].start, run[0].step, length) + if _reproduces(fused, run): + out.append(fused) + else: + out.extend(run) + else: + values = np.concatenate([x.values for x in run]) + out.append(CoordMonotonicArray(values=values, units=run[0].units)) + run.clear() + for seg in segments[1:]: - prev = out[-1] + prev = run[-1] both_ranges = isinstance(prev, CoordRange) and isinstance(seg, CoordRange) - if both_ranges and prev.step == seg.step and prev.stop == seg.start: - out[-1] = CoordRange( - start=prev.start, stop=seg.stop, step=prev.step, units=prev.units - ) - continue + continues = both_ranges and prev.step == seg.step and prev.stop == seg.start + # Adjacent irregular arrays carry no sampling expectation, so the + # boundary between them has no meaning; fuse for canonical form. both_arrays = isinstance(prev, CoordMonotonicArray) and isinstance( seg, CoordMonotonicArray ) - if both_arrays: - # Adjacent irregular arrays carry no sampling expectation, so the - # boundary between them has no meaning; fuse for canonical form. - values = np.concatenate([prev.values, seg.values]) - out[-1] = CoordMonotonicArray(values=values, units=prev.units) + if continues or both_arrays: + run.append(seg) continue - out.append(seg) + flush() + run.append(seg) + flush() return tuple(out) @@ -2161,10 +2229,15 @@ def _validate_segment_compat(segments: tuple[BaseCoord, ...]) -> None: dtypes = {np.dtype(s.dtype) for s in segments} msg = f"Segments must share compatible dtypes, got {dtypes}." raise CoordError(msg) - units = {get_quantity(s.units) for s in segments} - if len(units) > 1: - msg = "All segments must have the same units." - raise CoordError(msg) + # Hashing a pint Quantity is expensive and a long merge would do it + # once per segment, so the objects are compared first: segments which + # share one (or state none) agree without normalizing anything. + spellings = {id(s.units) for s in segments} + if len(spellings) > 1: + units = {get_quantity(s.units) for s in segments} + if len(units) > 1: + msg = "All segments must have the same units." + raise CoordError(msg) def _validate_segment_chain(segments: tuple[BaseCoord, ...]) -> None: diff --git a/dascore/io/index/backend.py b/dascore/io/index/backend.py index 6e126c241..c105993ea 100644 --- a/dascore/io/index/backend.py +++ b/dascore/io/index/backend.py @@ -1259,6 +1259,30 @@ def attr_stated_ids(self, name: str, patch_ids=None) -> set[int]: params.append(json.dumps([int(x) for x in patch_ids])) return {int(x) for x in self._fetch_df(sql, params)["patch_id"]} + def coord_frame(self, patch_ids) -> pd.DataFrame: + """ + Return one row per (patch, coordinate) for the given patches. + + Unlike the pivoted relation, which flattens a coordinate's envelope + onto the patch row, this keeps every coordinate's own record: the + dims it rides *for that patch* (`coord_dims_map` collapses those to + one per name), its dtype, its length, and its typed envelope. That + is what a summary needs to be rebuilt without loading anything. + """ + columns = ( + "pc.patch_id, pc.coord_name, pc.coord_dims, cd.def_key, cd.fingerprint, " + "cd.value_kind, cd.dtype, cd.length, cd.units, " + "cd.min_num, cd.max_num, cd.step_num, " + "cd.min_ns, cd.max_ns, cd.step_ns, cd.min_str, cd.max_str, " + "cd.is_relative" + ) + base = ( + f"SELECT {columns} FROM patch_coords pc " + "JOIN coord_defs cd ON cd.coord_def_id = pc.coord_def_id" + ) + ids = [int(x) for x in patch_ids] + return self._fetch_in(base, "pc.patch_id", ids) + def coord_dims_map(self) -> dict[str, str]: """Return each coord name's dims string (first observed wins).""" df = self._fetch_df("SELECT DISTINCT coord_name, coord_dims FROM patch_coords") diff --git a/dascore/io/index/ingest.py b/dascore/io/index/ingest.py index 5e6a3f60e..d4d258db6 100644 --- a/dascore/io/index/ingest.py +++ b/dascore/io/index/ingest.py @@ -17,13 +17,16 @@ import json import re import warnings -from collections.abc import Hashable +from collections.abc import Hashable, Mapping +from contextlib import suppress from dataclasses import dataclass, field, fields, replace -from typing import SupportsInt, TypedDict, cast +from functools import partial +from typing import Any, SupportsInt, TypedDict, cast import numpy as np import pandas as pd +from dascore.core.coords import CoordSummary from dascore.core.summary import PatchSummary, normalize_source_patch_key from dascore.exceptions import InvalidInventoryError from dascore.io.index.schema import ( @@ -387,6 +390,109 @@ def dump_path_attrs(path_attrs: dict[str, str] | None) -> str | None: return json.dumps(path_attrs, sort_keys=True) if path_attrs else None +def coord_summary(row: Mapping) -> CoordSummary | None: + """ + Rebuild a coordinate summary from one stored coordinate row. + + The inverse of `_coord_record`: a row of `patch_coords` joined to its + `coord_defs` definition (see `SQLiteIndexBackend.coord_frame`) states + everything a summary carries, so a member's coordinate can be + described without loading the patch it belongs to. + + Returns None for a row whose value kind the index does not represent. + + Built without re-validating: every value is converted here to the type + a validator would coerce it to, and this runs per coordinate per member + while a plan is made, where validation measured ~28us a call. + """ + kind = row.get("value_kind") + units = row.get("units") + units = None if units is None or pd.isnull(units) else units + fingerprint = row.get("fingerprint") + fingerprint = None if pd.isnull(fingerprint) else str(fingerprint) + length = row.get("length") + length = None if length is None or pd.isnull(length) else int(length) + dims = str(row.get("coord_dims") or "") + common: dict[str, Any] = dict( + dtype=str(row.get("dtype") or "").split("[")[0], + units=units, + dims=tuple(x for x in dims.split(",") if x), + len=length, + fingerprint=fingerprint, + ) + if kind == "time": + # stored as integer nanoseconds; a relative coord is a duration + stamp = _ns_timedelta if row.get("is_relative") else _ns_datetime + step = row.get("step_ns") + return CoordSummary.model_construct( + min=stamp(row.get("min_ns")), + max=stamp(row.get("max_ns")), + step=None if pd.isnull(step) else _ns_timedelta(step), + **common, + ) + if kind == "num": + # The envelope is stored as a float whatever the coordinate is, so + # it is cast back: an integer coordinate rebuilt from floats would + # be a float64 coordinate, with a different identity from the one + # the index recorded. An envelope nobody stated is NaN, as + # validation would make it; only a step is genuinely absent. + dtype = np.dtype(common["dtype"]) if common["dtype"] else np.dtype("float64") + low, high = _opt_float(row.get("min_num")), _opt_float(row.get("max_num")) + step = _opt_float(row.get("step_num")) + cast = partial(_as_dtype, dtype=dtype) + return CoordSummary.model_construct( + min=np.nan if low is None else cast(low), + max=np.nan if high is None else cast(high), + step=None if step is None else cast(step), + **common, + ) + if kind == "str": + low, high = row.get("min_str"), row.get("max_str") + return CoordSummary.model_construct( + min=None if low is None else str(low), + max=None if high is None else str(high), + **common, + ) + return None + + +def _as_dtype(value: float, dtype: np.dtype) -> Any: + """ + A stored float as the numeric type the coordinate states. + + Envelopes are stored as float64 whatever the coordinate is, so a + float32 or an unsigned one would rebuild as something else and carry + a different identity from the patch. The cast is kept only where it + round-trips, since restoring a dtype must not alter a value. + """ + if not np.issubdtype(dtype, np.number): + return value + with suppress(TypeError, ValueError, OverflowError): + cast = dtype.type(value) + if float(cast) == float(value): + return cast + return value + + +def _opt_float(value) -> float | None: + """A stored number, or None where the row states none.""" + return None if value is None or pd.isnull(value) else float(value) + + +def _ns_datetime(value) -> np.datetime64: + """A stored nanosecond count as a datetime, NaT when the row states none.""" + if value is None or pd.isnull(value): + return np.datetime64("NaT", "ns") + return np.datetime64(int(value), "ns") + + +def _ns_timedelta(value) -> np.timedelta64: + """A stored nanosecond count as a duration, NaT when the row states none.""" + if value is None or pd.isnull(value): + return np.timedelta64("NaT", "ns") + return np.timedelta64(int(value), "ns") + + def _coord_record(name: str, summary) -> CoordRecord | None: """Convert one CoordSummary into a CoordRecord.""" fingerprint = getattr(summary, "fingerprint", None) @@ -438,10 +544,12 @@ def _coord_record(name: str, summary) -> CoordRecord | None: **common, ) if dtype.kind in "USO": + # a summary which states no labels has none to store; stringifying + # the missing value would write the label "nan" return CoordRecord( value_kind="str", - min_str=str(summary.min), - max_str=str(summary.max), + min_str=None if pd.isnull(summary.min) else str(summary.min), + max_str=None if pd.isnull(summary.max) else str(summary.max), **common, ) return None # unsupported coord representation: skip, per design diff --git a/dascore/io/index/planned.py b/dascore/io/index/planned.py index f61baac5d..681291676 100644 --- a/dascore/io/index/planned.py +++ b/dascore/io/index/planned.py @@ -17,13 +17,16 @@ from __future__ import annotations import secrets -from collections.abc import Mapping +from collections.abc import Mapping, Sequence +from contextlib import suppress import numpy as np import pandas as pd import dascore as dc +from dascore.core.coord_join import join_summaries, raw_join_summary from dascore.core.coords import CoordSummary +from dascore.exceptions import UnitError from dascore.io.index.backend import get_backend from dascore.io.index.catalog import ( CompositeResolver, @@ -38,18 +41,19 @@ PatchRecord, SourceRecord, _coord_record, + coord_summary, typed_value, ) -from dascore.units import get_quantity +from dascore.units import convert_units, get_quantity, get_quantity_str from dascore.utils.chunk_plan import ( _SOURCE_COLUMNS, - _concatenated_steps, _ensure_patch_id, ) from dascore.utils.misc import _CanonicalRange, is_range from dascore.utils.patch import concatenate_planned from dascore.utils.patch_assembly import PatchAssembler from dascore.utils.pd import adjust_segments +from dascore.utils.time import to_float # Row columns which name dc.read's own keyword arguments; passing one along # as a trim hint would collide with the value the loader already supplies. @@ -248,13 +252,477 @@ def _extrema(grouped, how: str) -> np.ndarray: return np.array(values, dtype=object) +def _member_summaries(backend, members: pd.DataFrame) -> dict: + """Every member's coordinates, as the index recorded them.""" + if not len(members) or backend is None: + return {} + ids = [int(x) for x in members["_patch_id"].dropna().unique()] + assert ids, "a plan's members name the patches they load" + out: dict[int, dict[str, CoordSummary]] = {} + # one coordinate definition serves every member which shares it: a + # coordinate riding along unchanged is stored once and read once, + # which is most of them + seen: dict[tuple, CoordSummary | None] = {} + for row in backend.coord_frame(ids).to_dict("records"): + name = str(row["coord_name"]) + # keyed by the stored definition, not the fingerprint: a + # fingerprint normalizes units, so metres and feet holding the + # same physical values share one while their stored envelopes and + # spellings differ, and reusing either for the other would + # misreport what the patch holds + key = (name, row.get("def_key"), row.get("coord_dims")) + if key[1] is None or key not in seen: + seen[key] = coord_summary(row) + summary = seen[key] + if summary is not None: + out.setdefault(int(row["patch_id"]), {})[name] = summary + if set(out) != set(ids): + # Members this backend does not name describe other patches, so + # nothing here says what the outputs hold and the plan's own rows + # answer instead. Bare integers from two indexes also overlap by + # chance, which this cannot see: the collapse that would ask such + # a question is recognized by provenance in `derived_catalog`, + # which passes no backend at all. + return {} + return out + + +def _is_cut(stored: Mapping, row: Mapping, plan_dim: str) -> bool: + """Whether this member loads less than the whole of its dimension.""" + if row.get("_modified"): + return True + summary = stored.get(int(row["_patch_id"]), {}).get(plan_dim) + low, high = row.get(f"{plan_dim}_min"), row.get(f"{plan_dim}_max") + if summary is None or (pd.isnull(low) and pd.isnull(high)): + return False + # The planner states every member in one unit; the index kept each + # member's own spelling. Compared as they stand, a member restated + # from centimetres into metres looks trimmed to a hundredth of + # itself, and everything riding the dimension loses its envelope. + stated = row.get(f"_{plan_dim}_units") + first, last = summary.min, summary.max + if stated is not None and not pd.isnull(stated) and summary.units is not None: + with suppress(TypeError, ValueError, UnitError): + first = convert_units(first, stated, summary.units) + last = convert_units(last, stated, summary.units) + return bool(low != first or high != last) + + +def _trimmed_summary(summary: CoordSummary, row: Mapping, name: str) -> CoordSummary: + """ + The member's summary as its trim leaves it. + + A trimmed member holds fewer samples than the index recorded, so the + stored envelope, length and identity all describe something the + output will not contain; the plan's own range replaces them. + """ + low, high = row.get(f"{name}_min"), row.get(f"{name}_max") + if pd.isnull(low) and pd.isnull(high): + return summary + # The bounds are only the same bounds if they are said in the same + # unit: two single-sample members at zero read as equal whatever + # they are measured in, and returning each one's native spelling + # leaves the join with mixed units and nothing it can state. + stated = row.get(f"_{name}_units") + spelled_alike = ( + stated is None + or pd.isnull(stated) + or summary.units is None + or get_quantity(stated) == get_quantity(summary.units) + ) + if spelled_alike and low == summary.min and high == summary.max: + return summary # the whole of it, so its identity still holds + step = row.get(f"{name}_step", summary.step) + step = summary.step if pd.isnull(step) else step + length = None + if step is not None and not pd.isnull(step): + with suppress(TypeError, ValueError, ZeroDivisionError): + span = to_float(high) - to_float(low) + length = round(abs(span / to_float(step))) + 1 + # built rather than copied: these values come from the plan's frame, + # so they need the conforming a validated summary does (a pandas + # Timestamp where the rest of the join speaks numpy would not compare) + units = row.get(f"_{name}_units", summary.units) + units = summary.units if units is None or pd.isnull(units) else units + # A member restated in another unit is converted when it loads, and + # scaling an integer grid by a fraction gives floats: the row's own + # numbers already say so, and the record must agree with them or the + # identity names a coordinate nobody will load. + dtype = summary.dtype if spelled_alike else _stated_dtype(low, summary) + return CoordSummary( + dtype=dtype, + min=low, + max=high, + step=step, + units=units, + dims=summary.dims, + len=length, + ) + + +def _stated_dtype(value, summary: CoordSummary) -> str: + """The dtype of a bound the plan restated, falling back to the summary.""" + with suppress(TypeError, ValueError): + restated = np.asarray(value).dtype + if restated.kind == np.dtype(summary.dtype).kind or restated.kind == "f": + return str(restated) + return summary.dtype + + +def _union_summary(summaries: Sequence[CoordSummary]) -> CoordSummary: + """ + What can be said of members which cannot be joined from summaries. + + The envelope spans them all — that much any join preserves — and + nothing else is claimed: no step, and no identity. + """ + stated = [x for x in summaries if not pd.isnull(x.min)] + # a member which states nothing describes nothing, not even its dtype + template = stated[0] if stated else summaries[0] + blank = dict(step=None, len=None, fingerprint=None) + kinds = {_summary_kind(x) for x in stated} + spellings = {get_quantity(x.units) for x in stated} + silent = len(stated) != len(summaries) + if len(kinds) > 1 or len(spellings) > 1 or (silent and "str" in kinds): + # No envelope covers these members. Two spellings of one + # coordinate (2000 milliseconds beside 3 seconds) cannot be + # compared until one is chosen, which only the loaded patch does; + # two kinds of value cannot be compared at all; and a member + # which states no labels cannot be given one, since text has no + # missing value to stand in. + null = _null_like(template.min) + return template.model_copy(update=dict(min=null, max=null, **blank)) + # both ends come from the members the checks above inspected: a + # member stating only one of them was never compared for kind, and + # letting its max through would compare a moment with a number + lows = [x.min for x in stated] + highs = [x.max for x in stated if not pd.isnull(x.max)] + # assembly hands these arrays to numpy, which promotes them: an + # int32 laid beside a float64 comes back float64, and a record + # naming the first member's dtype would describe neither + dtype = template.dtype + with suppress(TypeError, ValueError): + dtype = str(np.result_type(*[np.dtype(x.dtype) for x in stated if x.dtype])) + return template.model_copy( + update=dict( + min=min(lows) if lows else template.min, + max=max(highs) if highs else template.max, + dtype=dtype, + **blank, + ) + ) + + +def _unvouched(trimmed: bool) -> dict: + """ + What a summary may still say once its values are not vouched for. + + A residual selection trims these values when the patch loads, so the + step and the sample count describe something the output will not + contain — and a summary which still looked evenly sampled would have + its identity recomputed from those very values by `_coord_record`. + The envelope stays: it still bounds where the output lies. + """ + void: dict = {"fingerprint": None} + if trimmed: + void.update(step=None, len=None) + return void + + +def _joins_in_member_order( + summaries: Sequence[CoordSummary], joined: CoordSummary +) -> bool: + """Whether the members already lie in the order the join put them in.""" + if joined.step is None: + # nothing structural is claimed of a join with no step, so the + # order the blocks lie in changes nothing this can protect + return True + # a summary's min is its smallest value whichever way it runs, so the + # direction comes from the step; `step - step` is its own zero + descending = joined.step < joined.step - joined.step + lows = [x.min for x in summaries] + return lows == sorted(lows, reverse=descending) + + +def _cut_rider( + summary: CoordSummary, + whole: CoordSummary | None, + row: Mapping, + plan_dim: str, +) -> CoordSummary | None: + """ + A rider as the cut leaves it, where the members say enough to tell. + + A coordinate riding the dimension being cut is sliced along with it, + sample for sample, so where both are evenly sampled the slice is + exact. Working it out matters: a row which states nothing about a + coordinate is not a candidate for any range over it, and the patch it + would have loaded is real. + """ + if whole is None or not (whole.is_range_like and whole.len): + return None + if not (summary.is_range_like and summary.len) or summary.len != whole.len: + return None + low, high = row.get(f"{plan_dim}_min"), row.get(f"{plan_dim}_max") + if pd.isnull(low) or pd.isnull(high): + return None + # both are evenly sampled and the same length, so the trim which + # slices one slices the other at the same samples + _, indexer = whole.to_coord(on_grid=True).select((low, high)) + sliced = summary.to_coord(on_grid=True)[indexer] + return sliced.to_summary().model_copy(update=dict(dims=summary.dims)) + + +def _summary_kind(summary: CoordSummary) -> str: + """Whether a summary holds times, numbers or labels.""" + kind = np.dtype(summary.dtype).kind if summary.dtype else "" + if kind in "mM": + # a moment and a duration are both spelled with time units and + # cannot be compared with each other, so they are not one kind + return "datetime" if kind == "M" else "timedelta" + return "str" if kind in "USO" else "num" + + +def _null_like(value): + """The missing value of whatever kind this one is.""" + if isinstance(value, np.datetime64 | pd.Timestamp): + return np.datetime64("NaT", "ns") + if isinstance(value, np.timedelta64 | pd.Timedelta): + return np.timedelta64("NaT", "ns") + return np.nan + + +def predicted_coords( + backend, + members: pd.DataFrame, + plan_dim: str, + *, + trimmed_dims: frozenset[str] = frozenset(), + snap_tolerance: float | None = None, + mode: str = "chunk", + drop_conflicting: bool = False, +) -> dict[int, dict[str, CoordSummary | None]]: + """ + Per output, the summary of every coordinate its members hold. + + A name mapped to None is one the members hold which the output will + not: the assembly drops it, so nothing is claimed about it, but the + envelope columns bearing its name are still its own. + + This is what the plan claims about an output, and it is decided by + running the *real* join over the members' summaries + ([`join_summaries`](`dascore.core.coord_join.join_summaries`)), so a + row cannot describe a coordinate differently from the patch assembly + will build. Where the join cannot be decided from summaries alone the + envelope still spans the members and nothing else is claimed. + + Parameters + ---------- + backend + The parent index, which holds the members' coordinate rows. + members + The plan's member table: which patches feed which output, with + each member's trim. + plan_dim + The dimension being chunked or concatenated. Coordinates riding + it are joined along it; the others must already agree. + trimmed_dims + Dimensions a residual selection trims at load. A coordinate on + one of them describes untrimmed values, so it keeps no identity. + snap_tolerance + Passed to the join, bounding how far a seam may be absorbed. + mode + Which assembly will build these outputs. A merge drops a + coordinate its members do not all share; a concatenation joins + raw values, so it can be identified only where they form a range. + drop_conflicting + Whether the merge drops non-dimensional coordinates its members + disagree about, rather than refusing to build the patch. + """ + stored = _member_summaries(backend, members) + if not stored: + return {} + out: dict[int, dict[str, CoordSummary | None]] = {} + # the whole table is turned into rows once: doing it per output costs + # pandas' fixed overhead thousands of times over on a segment plan + by_output: dict[int, list[dict]] = {} + for row in members.to_dict("records"): + by_output.setdefault(int(row["output_id"]), []).append(row) + for output_id, records in sorted(by_output.items()): + names: dict[str, None] = {} # an ordered set + for row in records: + names.update(dict.fromkeys(stored.get(int(row["_patch_id"]), {}))) + described: dict[str, CoordSummary | None] = {} + # the plan's member rows are what will be *loaded*: they carry each + # member's trim, in the unit the plan settled on, so along the + # planned dimension they outrank what the index recorded + cut = any(_is_cut(stored, row, plan_dim) for row in records) + for name in names: + summaries = [] + for row in records: + summary = stored.get(int(row["_patch_id"]), {}).get(name) + if summary is None: + continue + if name == plan_dim: + summary = _trimmed_summary(summary, row, name) + elif cut and plan_dim in summary.dims: + # A coordinate riding a dimension being cut keeps only + # the values inside the cut. Where both are evenly + # sampled that slice is exact; where it cannot be + # worked out the envelope goes with the step and the + # identity, rather than advertising values the patch + # does not hold. + sliced = _cut_rider( + summary, + stored.get(int(row["_patch_id"]), {}).get(plan_dim), + row, + plan_dim, + ) + if sliced is None: + null = _null_like(summary.min) + sliced = summary.model_copy( + update=dict( + min=null, + max=null, + step=None, + len=None, + fingerprint=None, + ) + ) + summary = sliced + summaries.append(summary) + assert summaries, "a name comes from the members which state it" + stated = _describe( + name, + summaries, + plan_dim, + trimmed_dims, + snap_tolerance, + mode=mode, + every_member=len(summaries) == len(records), + drop_conflicting=drop_conflicting, + ) + # None records that the members hold it and the output will + # not: no coordinate is written, and the envelope columns it + # owns are still recognized as its own rather than as attrs + described[name] = stated + out[output_id] = described + return out + + +def _describe( + name: str, + summaries: Sequence[CoordSummary], + plan_dim: str, + trimmed_dims: frozenset[str], + snap_tolerance: float | None, + *, + mode: str = "chunk", + every_member: bool = True, + drop_conflicting: bool = False, +) -> CoordSummary | None: + """ + State one coordinate of one output, claiming only what holds. + + None means the output does not carry it at all. + """ + first = summaries[0] + if plan_dim == name and name not in first.dims: + # the members hold this as an ordinary coordinate and the + # concatenation replaces it with a dimension of its own, so + # nothing the members say about it survives + return None + rides = plan_dim == name or plan_dim in first.dims + trimmed = bool(set(first.dims) & trimmed_dims) + if mode == "chunk" and len({tuple(x.dims) for x in summaries}) > 1: + # merge_coord_managers intersects (name, dims), so a coordinate + # the members hang on different dimensions is dropped rather than + # reconciled -- and where their values agree, nothing else + # notices in time to say so + return None + if mode == "chunk" and not every_member: + # A merge keeps only what every member states: + # merge_coord_managers drops the rest (see + # _drop_unshared_coordinates), so describing it would advertise a + # coordinate the patch will not carry. This holds for a rider as + # much as for a coordinate standing outside the merged dimension. + return None + if not rides: + # What assembly compares is the coordinate as each member holds + # it, so agreement takes the fingerprint *and* the units it is + # stated in: a fingerprint is normalized, and metres beside + # centimetres share one while `merge_coord_managers` finds them + # unequal and drops them. An unidentified coordinate is not + # thereby the same one in every member either -- a plan which + # could not vouch for its values stored no fingerprint, and two + # of those are unknown, not equal. A lone member has nothing to + # agree with and keeps what it says. + identities = {(x.fingerprint, get_quantity(x.units)) for x in summaries} + agreed = len(summaries) == 1 or ( + not any(fingerprint is None for fingerprint, _ in identities) + and len(identities) == 1 + ) + if mode == "chunk" and not (agreed or not drop_conflicting): + # the members disagree, and a merge told to drop conflicts + # will drop this one rather than choose between them + return None + if agreed and not trimmed: + return first + # members which disagree leave nothing to vouch for: a step and a + # sample count are enough for `_coord_record` to work an identity + # back out, so they go with the fingerprint + return first.model_copy(update=_unvouched(True)) + blank = all(pd.isnull(x.min) and pd.isnull(x.max) for x in summaries) + if blank and name == plan_dim: + # Nobody states any values along the dimension being joined, so + # there is nothing to join and nothing to say the plan's own row + # does not already say: it carries the identity the planner works + # out for such a dimension (see _member_key_digests). An + # auxiliary coordinate has no such row, so it is still described. + return None + # Only the merged dimension is snapped: `_get_merged_coord` simplifies + # it, while a rider is raw-concatenated through `get_coord`, which + # absorbs no seam. Snapping one here would claim a step the loaded + # coordinate does not have. + tolerance = snap_tolerance if name == plan_dim else None + joined = join_summaries(summaries, snap_tolerance=tolerance) + if joined is None: + return _union_summary(summaries) + raw_join = mode == "concat" or name != plan_dim + if raw_join and not _joins_in_member_order(summaries, joined): + # The join sorted these blocks; the concatenation will not. Their + # values interleave or run backwards once laid end to end, so the + # result is an array whose order -- and identity -- the sorted + # join does not describe. + joined = joined.model_copy(update=dict(step=None)) + if raw_join: + # The join generated one grid from a single start; the + # concatenation lays each member's own values end to end, which + # for floats is not always the same coordinate. + joined = raw_join_summary(summaries, joined) + if raw_join and joined.step is None: + # Only the merged dimension is built by the join this predicts + # with. A concatenation, and a rider on either path, has its raw + # values concatenated and handed to get_coord, which for anything + # but a single range gives an array whose identity is those + # values — unknowable from summaries. + joined = joined.model_copy(update=dict(fingerprint=None, len=None)) + if trimmed and name != plan_dim: + # the planned dimension's own trim is already in the member rows + # this joined; any other coordinate is cut at load instead + joined = joined.model_copy(update=_unvouched(True)) + return joined.model_copy(update=dict(dims=first.dims)) + + def _aux_coord_info( source_rows: pd.DataFrame, members: pd.DataFrame, plan_dim: str, coord_dims_map: Mapping[str, str], trimmed_dims: frozenset[str] = frozenset(), - concat: bool = False, + *, + mode: str = "chunk", + drop_conflicting: bool = False, ) -> dict[int, dict[str, dict]]: """ Aggregate per-output envelope info for auxiliary coordinates. @@ -268,9 +736,9 @@ def _aux_coord_info( always aggregate — the catalog contract is candidacy, with exact values re-established at load. - A concatenation joins a rider's segments rather than merging them, so - a rider whose members share one step and meet end to end keeps that - step (its values still differ member by member, so not its identity). + Used where an output's coordinates cannot be predicted from the + members' own summaries — a re-plan whose members this index does not + know — so the member rows are all there is to describe them with. """ out: dict[int, dict[str, dict]] = {} @@ -319,36 +787,37 @@ def _aux_coord_info( if step_col in joined.columns: step_ok = keep & (grouped[step_col].nunique().to_numpy() == 1) step_first = grouped[step_col].first().to_numpy() - if rides and concat: - # the joined coordinate is a range when the segments are - # (the members are already in the order they join in) - order = {v: i for i, v in enumerate(output_ids)} - codes = joined["output_id"].map(order).to_numpy() - step_first = _concatenated_steps(joined, codes, name) - step_ok = ~pd.isnull(step_first) unit_ok, unit_first = no_gate, None if unit_col in joined.columns: unit_ok = grouped[unit_col].nunique().to_numpy() == 1 unit_first = grouped[unit_col].first().to_numpy() - # members spelling one coordinate two ways (seconds beside - # milliseconds) have no envelope in common: the magnitudes mean - # different things, and only the loaded patch says which - # spelling wins. No units at all is one spelling, not two. - undecided = grouped[unit_col].nunique().to_numpy() > 1 - if undecided.any(): - lows = np.array( - [None if u else v for v, u in zip(lows, undecided)], dtype=object - ) - highs = np.array( - [None if u else v for v, u in zip(highs, undecided)], dtype=object - ) - step_ok = step_ok & ~undecided # a coordinate no member holds contributes nothing; one the members # do hold is always named, even when nothing about its values can # be stated — the patch will have it, so the catalog says so held = grouped[cmin].count().to_numpy() > 0 if key_col in joined.columns: held = held | (grouped[key_col].count().to_numpy() > 0) + if mode == "chunk": + # A merge keeps only what every member states, and only what + # they agree on when told to drop conflicts: assembly drops + # the rest, so naming it here would advertise a coordinate + # the patch will not carry. + stated = grouped[cmin].count().to_numpy() + if key_col in joined.columns: + stated = np.maximum(stated, grouped[key_col].count().to_numpy()) + dropped = stated != grouped.size().to_numpy() + lone = grouped.size().to_numpy() == 1 + if drop_conflicting and not rides and key_col in joined.columns: + # A rider holds a different segment in every member, so + # its definitions differ by design; assembly joins those + # values rather than comparing them, and only a + # coordinate standing outside the merge is a conflict. + # `nunique` counts no nulls, so a lone member holding an + # unidentified coordinate counts zero of them; it has + # nothing to disagree with either way + disagree = grouped[key_col].nunique().to_numpy() != 1 + dropped = dropped | (disagree & ~lone) + held = held & ~dropped absent = ~held for index in np.flatnonzero(~absent): step = step_first[index] if step_first is not None else None @@ -366,17 +835,123 @@ def _aux_coord_info( return out +def _clear_dropped_aux( + outputs: pd.DataFrame, + aux_info: Mapping[int, Mapping[str, Mapping]], + coord_dims_map: Mapping[str, str], + plan_dim: str, +) -> pd.DataFrame: + """ + Blank the envelope columns of coordinates the output will not carry. + + A coordinate `_aux_coord_info` does not describe is one assembly + drops. Its envelope columns are carried from the members and would + otherwise survive as ordinary metadata, so the row would still state + values for a coordinate the patch does not hold. + """ + names = [x for x in coord_dims_map if x != plan_dim] + if not names: + return outputs + out = outputs.copy(deep=False) + ids = [int(x) for x in out["output_id"]] + for name in names: + gone = [name not in aux_info.get(x, {}) for x in ids] + if not any(gone): + continue + columns = [f"{name}_min", f"{name}_max", f"{name}_step", f"_{name}_units"] + for column in (x for x in columns if x in out.columns): + kept = out[column].to_numpy(dtype=object, copy=True) + kept[np.array(gone)] = None + out[column] = pd.Series(kept, index=out.index, dtype=object) + return out + + +def _trimmed_envelopes( + predicted: Mapping[int, Mapping[str, CoordSummary | None]], + outputs: pd.DataFrame, + trimmed_dims: frozenset[str], +) -> dict[int, dict[str, CoordSummary | None]]: + """ + Restate a trimmed coordinate's envelope from the row holding the trim. + + A residual selection is applied when the patch loads, and the plan's + own row is where its adjusted bounds live: the members' stored + summaries still describe the whole of what the index recorded. + Candidacy is answered from the coordinate record, so a record which + kept the untrimmed envelope would keep the row a candidate for values + it will not return. + """ + # every described output is one of these rows: both come from the + # same plan, and the members were grouped by these very ids + rows = {int(x["output_id"]): x for x in outputs.to_dict("records")} + out: dict[int, dict[str, CoordSummary | None]] = {} + for output_id, described in predicted.items(): + row = rows[int(output_id)] + out[output_id] = { + name: summary + if summary is None or not (set(summary.dims) & trimmed_dims) + else _trimmed_summary(summary, row, name) + for name, summary in described.items() + } + return out + + +def _apply_predictions( + outputs: pd.DataFrame, + predicted: Mapping[int, Mapping[str, CoordSummary | None]], + name: str, +) -> pd.DataFrame: + """ + Restate the planned dimension's envelope from what the join predicts. + + The frame's envelope columns feed selection and the patches table, so + they must say what the records say; otherwise the row and its own + coordinate would disagree, which is the drift this predicts away. + """ + if not predicted: + return outputs + min_name, max_name, step_name = f"{name}_min", f"{name}_max", f"{name}_step" + unit_col = f"_{name}_units" + out = outputs.copy(deep=False) + described = [predicted.get(int(x), {}).get(name) for x in out["output_id"]] + if not any(x is not None for x in described): + return out + fields = { + min_name: lambda x: x.min, + max_name: lambda x: x.max, + step_name: lambda x: x.step, + unit_col: lambda x: None if x.units is None else get_quantity_str(x.units), + } + for column, read in fields.items(): + if column not in out.columns: + continue + # an output the join could not describe keeps what the row said; + # only what was predicted is restated + kept = out[column].to_numpy(dtype=object, copy=True) + for index, summary in enumerate(described): + if summary is not None: + kept[index] = read(summary) + out[column] = pd.Series(kept, index=out.index, dtype=object) + return out + + def _output_records( outputs: pd.DataFrame, token: str, aux_info: Mapping[int, Mapping[str, Mapping]] | None = None, + predicted: Mapping[int, Mapping[str, CoordSummary | None]] | None = None, ) -> list[SourceRecord]: """ Convert plan output rows into ingestible source records. + `predicted` states what an output's coordinates will be, decided by + joining its members' summaries. A coordinate it describes is written + from that summary; the row is consulted only for what it cannot know, + such as a dimension the plan creates out of the member count. """ records = [] aux_info = aux_info or {} + predicted = predicted or {} # Envelope columns belong to coordinates actually present in a row; # an attr that merely looks envelope-shaped (channel_step with no # channel coord) is ordinary metadata and must be preserved. The @@ -394,25 +969,38 @@ def _output_records( dims = str(row.get("dims") or "") dim_names = [d for d in dims.split(",") if d] aux = aux_info.get(output_id, {}) + known = predicted.get(output_id, {}) coords = [] for name in dim_names: - record = _coord_record_from_row(row, name) + summary = known.get(name) + if summary is not None: + record = _coord_record( + name, summary.model_copy(update={"dims": (name,)}) + ) + else: + record = _coord_record_from_row(row, name) + if record is not None: + coords.append(record) + for name, summary in known.items(): + if name in dim_names or summary is None: + continue + record = _coord_record(name, summary) if record is not None: coords.append(record) # auxiliary (non-dimension) coordinates remain on the assembled # patches, so the catalog must keep describing them for name, info in aux.items(): - if name in dim_names: + if name in dim_names or name in known: continue record = _coord_record_from_row( info, name, dims=info["dims"], name_is_held=True ) if record is not None: coords.append(record) - cache_key = (dims, tuple(aux)) + cache_key = (dims, tuple(aux), tuple(known)) envelope_keys = envelope_cache.get(cache_key) if envelope_keys is None: - coord_names = set(dim_names) | set(aux) | base_names + coord_names = set(dim_names) | set(aux) | set(known) | base_names envelope_keys = { f"{name}_{sfx}" for name in coord_names @@ -760,10 +1348,48 @@ def derived_catalog( stale_keys = stale_def_keys(parent_residuals, coord_dims_map, outputs.columns) if stale_keys: outputs = outputs.drop(columns=stale_keys) - aux_info = _aux_coord_info( - sources, trims, name, coord_dims_map, trimmed_dims, concat=mode == "concat" + # what an output will hold is decided by joining its members' + # summaries, the same join assembly runs on their values + snap = ( + merge_kwargs.get("tolerance") if merge_kwargs.get("snap_coords", True) else None + ) + # Re-planning a derived view on its own dimension collapses to the + # *grandparent's* members (see `collapse_working_df`), which this + # backend holds no coordinates for: its ids name the derived outputs, + # and asking it about a member id would describe the wrong patch. + collapsed = ( + parent is not None + and isinstance(parent.resolver, PlanResolver) + and parent.resolver.dim == name + and not parent.resolver.lossy ) - records = _output_records(outputs, token, aux_info=aux_info) + predicted = predicted_coords( + None if parent is None or collapsed else parent.backend, + trims, + name, + trimmed_dims=trimmed_dims, + snap_tolerance=snap, + mode=mode, + drop_conflicting=merge_kwargs.get("conflict") in {"drop", "keep_first"}, + ) + if predicted and trimmed_dims: + predicted = _trimmed_envelopes(predicted, outputs, trimmed_dims) + outputs = _apply_predictions(outputs, predicted, name) + aux_info = {} + if not predicted: + # a re-plan whose members this index does not know: the auxiliary + # coordinates are described from the member rows, as before + aux_info = _aux_coord_info( + sources, + trims, + name, + coord_dims_map, + trimmed_dims, + mode=mode, + drop_conflicting=merge_kwargs.get("conflict") in {"drop", "keep_first"}, + ) + outputs = _clear_dropped_aux(outputs, aux_info, coord_dims_map, name) + records = _output_records(outputs, token, aux_info=aux_info, predicted=predicted) backend.write_sources(records) return PatchCatalog(backend=backend, resolver=resolver) diff --git a/docs/notes/spool_chunking.qmd b/docs/notes/spool_chunking.qmd index b417f995a..48ab223c1 100644 --- a/docs/notes/spool_chunking.qmd +++ b/docs/notes/spool_chunking.qmd @@ -8,6 +8,8 @@ title: Spool Chunking The planner consumes the spool's flat metadata relation and produces a `ChunkPlan`: an `outputs` table (one row per patch the chunked spool will contain) and a `members` table (which slice of which source patch feeds each output). `Spool.chunk_plan` exposes the plan `chunk` would execute, with the same arguments. The chunked spool *is* a fresh in-memory catalog whose patch rows are the plan outputs; a plan resolver loads each output's members through the parent's resolver and executes the assembly engine (`dascore.utils.patch_assembly`) row by row. Chunking is one dimension per call; multi-dimensional chunking chains. Re-chunking a chunked spool along the *same* dimension re-plans from the current view's members (the collapse rule); chunking a *different* dimension plans over the spool's current output rows, preserving the boundaries the earlier operation assembled — `chunk(time=2).chunk(distance=100)` partitions both dimensions. `concatenate` is the same machinery with order-based grouping instead of continuity: `build_concat_plan` partitions as a chunk plan does (kind, dimensions, the identity of every other dimension, the dimension's units), polices the remaining attributes with `conflict` the same way (coordinates are settled when the output loads, not by the plan), then groups each partition's rows by the requested count in the order of the dimension with no sampling or gap test, and the assembled patches execute the plan rather than re-deciding it. +What an output's row *says* about its coordinates is not computed twice. The plan describes each output by running the same join the assembler runs — `dascore.core.coord_join.join_summaries` rebuilds each member's coordinate from the summary the index stored and calls `concat_coords`, exactly as `_get_merged_coord` does on real values — and states the result back as a summary. A row and the patch it describes therefore cannot disagree about an envelope, a step, a unit or an identity. Where summaries alone cannot settle the join (a member which states no step, members spelled in two units, values which overlap) the row claims only the envelope spanning its members, and where the members are not this index's at all (re-planning a derived view collapses to its *grandparent's* members) the plan's own rows describe the outputs as they did before. + ```{python} import dascore as dc diff --git a/tests/conftest.py b/tests/conftest.py index c880d46e7..a4f5ff719 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -730,3 +730,52 @@ def brady_hs_das_dts_coords(): coord_table = coord_table.iloc[51:] coord_table = coord_table.astype(float) return coord_table + + +def _assert_contents_match_patches(spool, skip=()): + """ + Assert a spool's catalog says what its patches actually hold. + + A lazy spool describes each patch before loading it. This compares + that description against re-indexing the patches themselves, which + is the only definition of the row being right. + + Columns which legitimately differ are skipped: provenance (a plan + row names a plan, a live patch names nothing), history, and any the + caller adds. + """ + described = spool.get_contents() + actual = dc.spool(list(spool)).get_contents() + common = set(described.columns) & set(actual.columns) + ignored = { + "history", + "source_path", + "source_format", + "source_version", + "source_patch_key", + *skip, + } + claimed = sorted( + c + for c in set(described.columns) - set(actual.columns) - ignored + if described[c].notnull().any() + ) + if claimed: + msg = f"the catalog states columns its patches do not hold: {claimed}" + raise AssertionError(msg) + columns = sorted(common - ignored) + left, right = described[columns], actual[columns] + same = (left == right) | (pd.isnull(left) & pd.isnull(right)) + if not same.all().all(): + bad = [c for c in columns if not same[c].all()] + msg = "\n".join( + f" {c}: row says {left[c].tolist()}, patches say {right[c].tolist()}" + for c in bad + ) + raise AssertionError(f"the catalog disagrees with its patches:\n{msg}") + + +@pytest.fixture(scope="session") +def assert_contents_match(): + """A callable asserting a spool's rows say what its patches hold.""" + return _assert_contents_match_patches diff --git a/tests/test_core/test_coord_join.py b/tests/test_core/test_coord_join.py new file mode 100644 index 000000000..f5fd8f2cc --- /dev/null +++ b/tests/test_core/test_coord_join.py @@ -0,0 +1,130 @@ +"""Tests for predicting a join from coordinate summaries.""" + +from __future__ import annotations + +import numpy as np +import pytest + +import dascore as dc +from dascore.core.coord_join import join_summaries +from dascore.core.coords import concat_coords, get_coord + + +def _range(start, stop, step=1.0, **kwargs): + """A range coordinate, for brevity.""" + return get_coord(start=start, stop=stop, step=step, **kwargs) + + +def _joined(coords, tolerance=None): + """What the real join gives, as a summary.""" + joined = concat_coords(*coords) + if tolerance: + step = joined.step + if step is None: + step = max(abs(x.step) for x in coords if x.step is not None) + joined = joined.simplify(tolerance * np.abs(step)) + return joined.to_summary() + + +class TestAgreesWithTheRealJoin: + """The prediction is the real join's answer, or nothing.""" + + @pytest.mark.parametrize( + "coords", + [ + [_range(0.0, 10.0), _range(10.0, 20.0)], # contiguous + [_range(0.0, 10.0), _range(15.0, 25.0)], # gapped + [_range(10.0, 20.0), _range(0.0, 10.0)], # supplied out of order + [_range(0.0, 10.0), _range(10.0, 20.0), _range(20.0, 30.0)], + [_range(10.0, 0.0, -1.0), _range(0.0, -10.0, -1.0)], # descending + [_range(0.0, 10.0, units="m"), _range(10.0, 20.0, units="m")], + ], + ) + def test_matches(self, coords): + """Every field, fingerprint included, is what the join produces.""" + predicted = join_summaries([x.to_summary() for x in coords]) + assert predicted == _joined(coords) + + def test_time_coords_match(self): + """Datetime members join the same way.""" + t0 = np.datetime64("2020-01-01", "ns") + step = np.timedelta64(4_000_000, "ns") + coords = [ + get_coord(start=t0, stop=t0 + 100 * step, step=step), + get_coord(start=t0 + 100 * step, stop=t0 + 200 * step, step=step), + ] + assert join_summaries([x.to_summary() for x in coords]) == _joined(coords) + + def test_patch_coords_match(self): + """Coordinates as the index stores them predict exactly.""" + patch = dc.get_example_patch() + time = patch.get_coord("time") + half = len(time) // 2 + first, second = time[:half], time[half:] + predicted = join_summaries([first.to_summary(), second.to_summary()]) + assert predicted == _joined([first, second]) + assert predicted.fingerprint == time.fingerprint() + + def test_a_coarser_step_is_not_vouched_for(self): + """ + A step spelled in coarser units leaves the join unidentified. + + The index stores nanoseconds, so a coordinate written at another + precision does not rebuild into itself; the envelope still spans + the members, but nothing is claimed about how they sample it. + """ + t0 = np.datetime64("2020-01-01", "ms") + step = np.timedelta64(4, "ms") + coords = [ + get_coord(start=t0, stop=t0 + 100 * step, step=step), + get_coord(start=t0 + 100 * step, stop=t0 + 200 * step, step=step), + ] + predicted = join_summaries([x.to_summary() for x in coords]) + assert predicted.fingerprint is None + assert predicted.step is None + assert predicted.len is None + # the envelope is still the one the join produces + expected = _joined(coords) + assert predicted.min == expected.min + assert predicted.max == expected.max + + +class TestClaimsNothingWhenItCannotTell: + """Where summaries are not enough, the answer is None.""" + + def test_no_summaries(self): + """Nothing in, nothing claimed.""" + assert join_summaries([]) is None + + def test_one_summary_passes_through(self): + """A lone member is its own answer, untouched.""" + summary = _range(0.0, 10.0).to_summary() + assert join_summaries([summary]) is summary + + def test_member_without_a_step(self): + """An array member's values are not in its summary.""" + array = get_coord(values=np.array([0.0, 1.0, 3.0])) + summaries = [array.to_summary(), _range(10.0, 20.0).to_summary()] + assert join_summaries(summaries) is None + + def test_value_less_member(self): + """A coordinate which states no values cannot be joined.""" + blank = dc.get_example_patch().mean("time").get_coord("time") + summaries = [blank.to_summary(), blank.to_summary()] + assert join_summaries(summaries) is None + + def test_members_spelled_two_ways(self): + """Which spelling wins is decided on the values, not here.""" + coords = [_range(0.0, 10.0, units="m"), _range(10.0, 20.0, units="cm")] + assert join_summaries([x.to_summary() for x in coords]) is None + + def test_overlapping_members(self): + """Overlapping members will raise at load; the row claims nothing.""" + coords = [_range(0.0, 10.0), _range(5.0, 15.0)] + assert join_summaries([x.to_summary() for x in coords]) is None + + def test_string_members(self): + """Label coordinates carry no step, so they take the same path.""" + labels = get_coord(values=np.array(["a", "b", "c"])) + summaries = [labels.to_summary(), labels.to_summary()] + assert join_summaries(summaries) is None diff --git a/tests/test_core/test_coords.py b/tests/test_core/test_coords.py index 83872ea4b..f833abc34 100644 --- a/tests/test_core/test_coords.py +++ b/tests/test_core/test_coords.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +import itertools import pickle import re from collections.abc import Mapping @@ -28,6 +29,8 @@ CoordString, CoordSummary, _get_coord_kind, + _reproduces, + concat_coords, get_coord, ) from dascore.exceptions import CoordError, ParameterError @@ -2903,3 +2906,131 @@ def test_the_hook_says_so_when_it_is_not_implemented(self, evenly_sampled_coord) """A class which did not implement it gets an error naming itself.""" with pytest.raises(NotImplementedError, match="unit conversion"): BaseCoord._convert_units(evenly_sampled_coord, "m") + + +class TestFusedRangeConstruction: + """A fused range must equal the one full validation would build.""" + + @pytest.mark.parametrize( + "start,stop,step", + [ + (0.0, 10.0, 1.0), + (10.0, 0.0, -1.0), + (-5.0, 5.0, 0.5), + ], + ) + def test_numeric_fuse_matches_validated(self, start, stop, step): + """Fusing two ranges gives the range spanning both.""" + first = get_coord(start=start, stop=stop, step=step) + second = get_coord(start=stop, stop=stop + (stop - start), step=step) + fused = concat_coords(first, second) + expected = CoordRange(start=start, stop=stop + (stop - start), step=step) + assert fused == expected + assert fused.fingerprint() == expected.fingerprint() + assert np.array_equal(fused.values, expected.values) + + def test_time_fuse_keeps_units_and_values(self): + """A datetime fuse states seconds, as the validating path does.""" + t0 = np.datetime64("2020-01-01", "ns") + step = np.timedelta64(4, "ms") + first = get_coord(start=t0, stop=t0 + 100 * step, step=step) + second = get_coord(start=t0 + 100 * step, stop=t0 + 200 * step, step=step) + fused = concat_coords(first, second) + expected = CoordRange(start=t0, stop=t0 + 200 * step, step=step) + assert fused == expected + assert fused.units == expected.units + assert fused.fingerprint() == expected.fingerprint() + assert np.array_equal(fused.values, expected.values) + + def test_unitful_fuse_keeps_the_unit(self): + """The fused range speaks the unit its segments spoke.""" + first = get_coord(start=0.0, stop=10.0, step=1.0, units="m") + second = get_coord(start=10.0, stop=20.0, step=1.0, units="m") + fused = concat_coords(first, second) + assert fused.units == get_quantity("m") + assert len(fused) == 20 + + def test_a_run_which_cannot_be_rebuilt_keeps_its_pieces(self): + """ + Fusing must not move a value, even by an ulp. + + Floating point addition is not associative, so a grid rebuilt + from the first start and the summed length need not land on the + boundaries its pieces state; where it does not, the pieces stay + as they were rather than being quietly altered. + """ + pieces, start = [], 0.1 + for _ in range(4): + piece = get_coord(start=start, stop=start + 0.3, step=0.1) + pieces.append(piece) + start = piece.stop + # the pieces meet exactly, so they form one run + assert all(a.stop == b.start for a, b in itertools.pairwise(pieces)) + joined = concat_coords(*pieces) + assert len(joined) == sum(len(x) for x in pieces) + assert np.array_equal(joined.values, np.concatenate([x.values for x in pieces])) + + def test_reproduces_rejects_a_grid_which_misses_a_boundary(self): + """The check itself answers both ways.""" + pieces, start = [], 0.1 + for _ in range(4): + piece = get_coord(start=start, stop=start + 0.3, step=0.1) + pieces.append(piece) + start = piece.stop + length = sum(len(x) for x in pieces) + drifting = pieces[0]._new_grid(pieces[0].start, pieces[0].step, length) + assert not _reproduces(drifting, pieces) + assert _reproduces(pieces[0], [pieces[0]]) + + def test_many_segments_fuse_to_one_range(self): + """A long run of contiguous ranges collapses to a single range.""" + pieces = [ + get_coord(start=i * 10.0, stop=(i + 1) * 10.0, step=1.0) for i in range(50) + ] + fused = concat_coords(*pieces) + assert isinstance(fused, CoordRange) + assert len(fused) == 500 + assert fused.min() == 0.0 and fused.max() == 499.0 + + +class TestSummaryOnGrid: + """A summary trusted to describe a grid builds the same coord.""" + + @pytest.mark.parametrize( + "kwargs", + [ + dict(start=0.0, stop=100.0, step=1.0), + dict(start=100.0, stop=0.0, step=-1.0), + dict(start=-5.0, stop=5.0, step=0.5, units="m"), + ], + ) + def test_matches_the_validated_conversion(self, kwargs): + """on_grid=True is a shortcut, never a different answer.""" + summary = get_coord(**kwargs).to_summary() + slow, fast = summary.to_coord(), summary.to_coord(on_grid=True) + assert slow == fast + assert slow.fingerprint() == fast.fingerprint() + assert slow.units == fast.units + assert np.array_equal(slow.values, fast.values) + + def test_time_summary_matches(self): + """Time-like coords keep the seconds their validated twin states.""" + t0 = np.datetime64("2020-01-01", "ns") + step = np.timedelta64(4, "ms") + summary = get_coord(start=t0, stop=t0 + 100 * step, step=step).to_summary() + slow, fast = summary.to_coord(), summary.to_coord(on_grid=True) + assert slow == fast + assert slow.units == fast.units + assert np.array_equal(slow.values, fast.values) + + def test_without_a_length_falls_back(self): + """A summary which does not state its length is validated as before.""" + summary = CoordSummary(dtype="float64", min=0.0, max=9.0, step=1.0) + assert summary.len is None + assert summary.to_coord(on_grid=True) == summary.to_coord() + + def test_still_refuses_a_summary_without_a_step(self): + """The shortcut does not make an unsampled summary convertible.""" + summary = CoordSummary(dtype="float64", min=0.0, max=9.0) + with pytest.raises(CoordError, match="evenly sampled"): + summary.to_coord(on_grid=True) diff --git a/tests/test_core/test_patch_chunk.py b/tests/test_core/test_patch_chunk.py index a5dba7047..eaf158311 100644 --- a/tests/test_core/test_patch_chunk.py +++ b/tests/test_core/test_patch_chunk.py @@ -53,32 +53,9 @@ def test_chunk_doesnt_modify_original(self, random_spool): second = random_spool.get_contents().copy() assert first.equals(second) - def test_patches_match_df_contents(self, random_spool): + def test_patches_match_df_contents(self, random_spool, assert_contents_match): """Ensure the patch content matches the dataframe.""" - new = random_spool.chunk(time=2) - # get contents of chunked spool - chunk_df = new.get_contents() - new_patches = list(new) - new_spool = dc.spool(new_patches) - # get content of spool created from patches in chunked spool. - new_content = new_spool.get_contents() - # these should be (nearly) identical. - common = set(chunk_df.columns) & set(new_content.columns) - # len fields may differ by ±1 between summary-based and data-based - # counts; identity/provenance columns legitimately differ between - # plan rows and re-scanned live patches - skip = { - "history", - "source_path", - "source_format", - "source_version", - "source_patch_key", - } - skip |= {c for c in common if c.endswith("_len")} - cols = sorted(common - skip) - comp1, comp2 = chunk_df[cols], new_content[cols] - equal_cols = (comp1 == comp2) | (pd.isnull(comp1) & pd.isnull(comp2)) - assert equal_cols.all().all() + assert_contents_match(random_spool.chunk(time=2)) def test_merge_empty_spool(self, tmp_path_factory): """Ensure merge doesn't raise on empty spools.""" @@ -1465,3 +1442,59 @@ def test_array_size_raises(self, random_spool): """A chunk length is one value, not an array of them.""" with pytest.raises(ParameterError, match="single quantity"): random_spool.chunk(time=np.array([1.0, 2.0]) * dc.units.MB) + + +class TestRowsMatchPatches: + """A plan's rows say what its patches hold, whatever the plan.""" + + @pytest.fixture(scope="class") + def spool_with_aux(self): + """A spool whose patches carry a non-dimensional coordinate.""" + spool = dc.get_example_spool("random_das") + patches = [] + for patch in spool: + n = patch.shape[patch.get_axis("distance")] + patches.append( + patch.update_coords(latitude=("distance", np.arange(n) * 1.0)) + ) + return dc.spool(patches) + + @pytest.mark.parametrize( + "operation", + [ + lambda x: x.chunk(time=None), + lambda x: x.chunk(time=2), + lambda x: x.chunk(time=2, overlap=0.5), + lambda x: x.concatenate(time=None), + lambda x: x.concatenate(time=2), + lambda x: x.chunk(time=None).chunk(time=4), + ], + ) + def test_plain_spool(self, random_spool, operation, assert_contents_match): + """Every plan kind describes its outputs as they turn out.""" + assert_contents_match(operation(random_spool)) + + @pytest.mark.parametrize( + "operation", + [ + lambda x: x.chunk(time=None), + lambda x: x.chunk(time=2), + lambda x: x.concatenate(time=None), + ], + ) + def test_with_an_auxiliary_coordinate( + self, spool_with_aux, operation, assert_contents_match + ): + """A coordinate riding another dimension is described as it lands.""" + assert_contents_match(operation(spool_with_aux)) + + def test_after_a_selection(self, random_spool, assert_contents_match): + """A selection applied at load leaves the row honest.""" + time = random_spool[0].get_coord("time") + window = (time.min(), time.min() + (time.max() - time.min()) / 2) + selected = random_spool.select(time=window).chunk(time=None) + # time_max is excluded: a selection's upper bound is a request, and + # the row states the request while the patch ends at the last + # sample inside it. That predates this machinery -- `dev` states + # the same pair -- and belongs to how residuals clamp envelopes. + assert_contents_match(selected, skip={"time_max"}) diff --git a/tests/test_io/test_index/test_ingest.py b/tests/test_io/test_index/test_ingest.py new file mode 100644 index 000000000..0ae41ccc8 --- /dev/null +++ b/tests/test_io/test_index/test_ingest.py @@ -0,0 +1,137 @@ +"""Tests for turning stored coordinate rows back into summaries.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +import dascore as dc +from dascore.core.coords import get_coord +from dascore.io.index.ingest import _as_dtype, coord_summary + + +class TestCoordSummaryFromRow: + """A stored coordinate row states everything its summary carries.""" + + @pytest.fixture() + def indexed(self): + """A spool whose patch has a numeric, a time and a label coord.""" + patch = dc.get_example_patch() + n = patch.shape[patch.get_axis("distance")] + labels = np.array([f"s{i:03d}" for i in range(n)]) + patch = patch.update_coords( + latitude=("distance", np.arange(n) * 1.0), + station=("distance", labels), + ) + return patch, dc.spool([patch]) + + def _rows(self, spool): + """Every stored coordinate row of the spool's one patch.""" + backend = spool._catalog.backend + frame = backend._fetch_df("SELECT patch_id FROM patches") + ids = [int(x) for x in frame["patch_id"]] + return backend.coord_frame(ids).to_dict("records") + + def test_describes_every_coordinate(self, indexed): + """Each row rebuilds the coordinate the patch holds.""" + patch, spool = indexed + rows = {x["coord_name"]: x for x in self._rows(spool)} + assert set(rows) == set(patch.coords.coord_map) + for name, row in rows.items(): + summary = coord_summary(row) + coord = patch.get_coord(name) + assert summary.min == coord.min() + assert summary.max == coord.max() + assert summary.len == len(coord) + assert summary.dims == patch.coords.dim_map[name] + assert summary.fingerprint == coord.fingerprint() + + def test_range_coords_rebuild_exactly(self, indexed): + """A sampled coordinate rebuilds into the coordinate it describes.""" + patch, spool = indexed + for row in self._rows(spool): + summary = coord_summary(row) + if not summary.is_range_like: + continue + rebuilt = summary.to_coord(on_grid=True) + coord = patch.get_coord(row["coord_name"]) + assert rebuilt.fingerprint() == coord.fingerprint() + assert np.array_equal(rebuilt.values, coord.values) + + @pytest.mark.parametrize("dtype", ["float32", "float64", "int64", "uint16"]) + def test_every_numeric_dtype_rebuilds_as_itself(self, dtype): + """A stored envelope comes back as the kind of number it was.""" + patch = dc.get_example_patch() + n = patch.shape[patch.get_axis("distance")] + coord = get_coord(values=np.arange(n, dtype=dtype)) + spool = dc.spool([patch.update_coords(distance=coord)]) + row = next(x for x in self._rows(spool) if x["coord_name"] == "distance") + rebuilt = coord_summary(row).to_coord(on_grid=True) + assert rebuilt.dtype == coord.dtype + assert rebuilt.fingerprint() == coord.fingerprint() + + def test_a_cast_which_would_change_a_value_is_not_made(self): + """Restoring a dtype is a change of type, never of value.""" + # a fractional value cannot be an integer, so it stays as it is + assert _as_dtype(1.5, np.dtype("int64")) == 1.5 + # a dtype which is not a number is left alone + assert _as_dtype(1.5, np.dtype("str")) == 1.5 + + def test_relative_time_is_a_duration(self): + """A relative time coordinate rebuilds as a duration, not a date.""" + row = { + "value_kind": "time", + "is_relative": True, + "min_ns": 0, + "max_ns": 1_000_000_000, + "step_ns": 1_000_000, + "dtype": "timedelta64", + "coord_dims": "time", + "length": 1001, + "units": None, + "fingerprint": None, + } + summary = coord_summary(row) + assert summary.max == np.timedelta64(1, "s") + assert summary.step == np.timedelta64(1, "ms") + + def test_a_row_stating_no_values_is_described(self): + """A null envelope gives a summary which claims nothing.""" + row = { + "value_kind": "num", + "min_num": np.nan, + "max_num": np.nan, + "step_num": None, + "dtype": "float64", + "coord_dims": "rank", + "length": None, + "units": None, + "fingerprint": "abc", + } + summary = coord_summary(row) + assert not summary.is_range_like + assert summary.fingerprint == "abc" + + @pytest.mark.parametrize("relative", [True, False]) + def test_a_time_row_without_values(self, relative): + """A time coordinate stating no values summarizes as NaT.""" + row = { + "value_kind": "time", + "is_relative": relative, + "min_ns": None, + "max_ns": None, + "step_ns": None, + "dtype": "timedelta64" if relative else "datetime64", + "coord_dims": "time", + "length": None, + "units": None, + "fingerprint": None, + } + summary = coord_summary(row) + assert pd.isnull(summary.min) and pd.isnull(summary.max) + assert not summary.is_range_like + + def test_unknown_kind_is_skipped(self): + """A value kind the index does not represent describes nothing.""" + assert coord_summary({"value_kind": "something_else"}) is None diff --git a/tests/test_io/test_index/test_planned.py b/tests/test_io/test_index/test_planned.py index 5a7606a1f..04ee5152d 100644 --- a/tests/test_io/test_index/test_planned.py +++ b/tests/test_io/test_index/test_planned.py @@ -12,18 +12,32 @@ import pytest import dascore as dc -from dascore.exceptions import MissingPatchError, ParameterError +from dascore.core.coords import get_coord +from dascore.exceptions import CoordMergeError, MissingPatchError, ParameterError +from dascore.io.index.catalog import PatchCatalog from dascore.io.index.planned import ( PlanResolver, + _apply_predictions, _aux_coord_info, _coord_record_from_row, + _cut_rider, + _describe, + _extrema, _ns, + _null_like, + _stated_dtype, _stated_units, collapse_working_df, derived_catalog, + predicted_coords, ) from dascore.units import m -from dascore.utils.chunk_plan import ChunkPlan, samples_adjusted_envelopes +from dascore.utils.chunk_plan import ( + ChunkPlan, + build_concat_plan, + samples_adjusted_envelopes, +) +from dascore.utils.patch import concatenate_patches @pytest.fixture(scope="module") @@ -397,3 +411,797 @@ def test_absent_units_read_as_none(self, value): def test_stated_units_pass_through(self): """A real spelling survives as a string.""" assert _stated_units("ft") == "ft" + + +class TestPredictedCoords: + """What a plan claims about an output, decided by the real join.""" + + @pytest.fixture() + def pair(self): + """Two patches which meet end to end along time.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + return first, second + + def _plan_and_backend(self, patches, **kwargs): + """A concat plan over the patches, and the spool's backend.""" + spool = dc.spool(list(patches)) + frame = PatchCatalog.from_patches(list(patches)).to_df() + plan = build_concat_plan(frame, **kwargs) + return plan, spool._catalog.backend + + def test_joined_dimension_is_the_real_join(self, pair): + """The concatenated dimension is described as assembly builds it.""" + plan, backend = self._plan_and_backend(pair, time=None) + described = predicted_coords(backend, plan.members, "time")[0] + time = described["time"] + whole = concatenate_patches(list(pair), time=None)[0].get_coord("time") + assert time.min == whole.min() + assert time.max == whole.max() + assert time.len == len(whole) + assert time.step == whole.step + assert time.fingerprint == whole.fingerprint() + + def test_other_coordinates_keep_their_identity(self, pair): + """A coordinate every member shares is described as it stands.""" + plan, backend = self._plan_and_backend(pair, time=None) + described = predicted_coords(backend, plan.members, "time")[0] + distance = pair[0].get_coord("distance") + assert described["distance"].fingerprint == distance.fingerprint() + assert described["distance"].len == len(distance) + + def test_a_trimmed_dimension_claims_no_identity(self, pair): + """Values a residual trims at load are not vouched for.""" + plan, backend = self._plan_and_backend(pair, time=None) + described = predicted_coords( + backend, plan.members, "time", trimmed_dims=frozenset({"distance"}) + )[0] + assert described["distance"].fingerprint is None + + def test_members_which_cannot_be_joined_span_the_envelope(self): + """Labels have no step, so only the envelope is claimed.""" + base = dc.get_example_patch() + n = base.shape[base.get_axis("distance")] + labels = np.array([f"s{i:03d}" for i in range(n)]) + renamed = base.rename_coords(distance="range") + first = renamed.update_coords(range=labels) + second = renamed.update_coords(range=np.array([f"t{i:03d}" for i in range(n)])) + plan, backend = self._plan_and_backend([first, second], range=None) + described = predicted_coords(backend, plan.members, "range")[0] + assert described["range"].step is None + assert described["range"].fingerprint is None + assert described["range"].min == "s000" + assert described["range"].max == f"t{n - 1:03d}" + + def test_a_trim_of_the_joined_dimension_is_already_counted(self, pair): + """The planned dimension's own trim rides in the rows being joined.""" + plan, backend = self._plan_and_backend(pair, time=None) + described = predicted_coords( + backend, plan.members, "time", trimmed_dims=frozenset({"time"}) + )[0] + # the members state what will be loaded, so the join still holds + assert described["time"].fingerprint is not None + assert described["time"].len is not None + + def test_a_rider_of_a_trimmed_dimension_is_voided(self, pair): + """A coordinate joined along a dimension the load trims says less.""" + first, second = pair + nt = first.shape[first.get_axis("time")] + patches = [ + x.update_coords(clock=("time", np.arange(nt) * 1.0 + i * nt)) + for i, x in enumerate((first, second)) + ] + plan, backend = self._plan_and_backend(patches, time=None) + described = predicted_coords( + backend, plan.members, "time", trimmed_dims=frozenset({"time"}) + )[0] + assert described["clock"].fingerprint is None + assert described["clock"].step is None + # the envelope still bounds where the rider lies + assert described["clock"].min == 0.0 + + def test_a_trim_of_another_dimension_voids_what_rides_it(self, pair): + """A coordinate cut at load claims neither step nor identity.""" + first, second = pair + n = first.shape[first.get_axis("distance")] + patches = [ + x.update_coords(latitude=("distance", np.arange(n) * 1.0)) + for x in (first, second) + ] + plan, backend = self._plan_and_backend(patches, time=None) + described = predicted_coords( + backend, plan.members, "time", trimmed_dims=frozenset({"distance"}) + )[0] + assert described["latitude"].fingerprint is None + assert described["latitude"].step is None + + def test_a_trimmed_member_is_described_by_its_trim(self, pair): + """A member which loads part of its patch states that part.""" + first, second = pair + plan, backend = self._plan_and_backend(pair, time=None) + members = plan.members.copy() + time = first.get_coord("time") + cut = time.min() + (time.max() - time.min()) / 2 + members["_modified"] = [True, False] + members.loc[members.index[0], "time_max"] = cut + described = predicted_coords(backend, members, "time")[0] + # the trimmed member's own range bounds the join, not its source + assert described["time"].min == time.min() + assert described["time"].max == second.get_coord("time").max() + + def test_a_trim_which_states_no_range_is_left_alone(self, pair): + """A member marked modified but stating no range keeps its summary.""" + plan, backend = self._plan_and_backend(pair, time=None) + members = plan.members.copy() + members["_modified"] = True + members["time_min"] = pd.NaT + members["time_max"] = pd.NaT + described = predicted_coords(backend, members, "time")[0] + whole = concatenate_patches(list(pair), time=None)[0].get_coord("time") + assert described["time"].max == whole.max() + + def test_a_coordinate_one_member_lacks(self, pair): + """What a partly-held coordinate becomes depends on the assembly.""" + first, second = pair + n = first.shape[first.get_axis("distance")] + lat = second.update_coords(latitude=("distance", np.arange(n) * 1.0)) + plan, backend = self._plan_and_backend([first, lat], time=None) + # a concatenation carries it over from the member which has it + described = predicted_coords(backend, plan.members, "time", mode="concat")[0] + assert "latitude" in described + assert described["latitude"].len == n + # a merge drops what its members do not all share, so nothing is said + merged = predicted_coords(backend, plan.members, "time", mode="chunk")[0] + # named, but stated as nothing: the merge will not carry it + assert merged["latitude"] is None + + def test_a_merge_drops_what_its_members_disagree_about(self, pair): + """Under drop, a coordinate the members differ on is not described.""" + first, second = pair + n = first.shape[first.get_axis("distance")] + patches = [ + first.update_coords(latitude=("distance", np.arange(n) * 1.0)), + second.update_coords(latitude=("distance", np.ones(n))), + ] + plan, backend = self._plan_and_backend(patches, time=None) + dropped = predicted_coords( + backend, plan.members, "time", mode="chunk", drop_conflicting=True + )[0] + assert dropped["latitude"] is None + # refusing instead of dropping, the output either matches or raises + kept = predicted_coords(backend, plan.members, "time", mode="chunk")[0] + assert "latitude" in kept + + def test_a_coordinate_nobody_states_is_left_to_the_row(self, pair): + """A blank dimension is described by the plan, not predicted.""" + first, _ = pair + blanks = [first.mean("time"), first.new().mean("time")] + plan, backend = self._plan_and_backend(blanks, time=None) + described = predicted_coords(backend, plan.members, "time")[0] + assert described["time"] is None # the row states its identity + assert "distance" in described # everything else is still described + + def test_null_of_each_kind(self): + """The missing value of a kind is of that kind.""" + assert pd.isnull(_null_like(np.datetime64("2020-01-01", "ns"))) + assert isinstance(_null_like(np.timedelta64(1, "s")), np.timedelta64) + assert isinstance(_null_like(pd.Timedelta(1, "s")), np.timedelta64) + assert pd.isnull(_null_like(1.0)) + + def test_predictions_skip_columns_the_frame_lacks(self, pair): + """Restating an envelope touches only columns the frame has.""" + plan, backend = self._plan_and_backend(pair, time=None) + described = predicted_coords(backend, plan.members, "time") + outputs = plan.outputs.drop(columns=["time_step"]) + applied = _apply_predictions(outputs, described, "time") + assert "time_step" not in applied.columns + assert applied["time_max"].iloc[0] == described[0]["time"].max + + def test_a_replanned_view_falls_back_to_its_rows(self): + """Members this index does not know are described from the plan.""" + spool = dc.get_example_spool("random_das") + patches = [ + x.update_coords( + sensor=("distance", np.arange(x.shape[x.get_axis("distance")]) * 1.0) + ) + for x in spool + ] + joined = dc.spool(patches).concatenate(time=None) + # the re-plan collapses to members of the *original* spool, whose + # ids this derived index does not use + again = joined.chunk(time=None) + row = again.get_contents().iloc[0] + assert row["sensor_min"] == 0.0 + assert row["sensor_max"] == patches[0].get_coord("sensor").max() + assert "sensor" in again[0].coords.coord_map + + def test_a_label_coordinate_falls_back_to_its_row(self): + """A string coordinate is described from the row when predicting cannot.""" + row = { + "station_min": "a000", + "station_max": "a299", + "_station_def_key": "fp:" + "b" * 32, + } + record = _coord_record_from_row(row, "station", dims=("distance",)) + assert record is not None + assert record.value_kind == "str" + assert record.min_str == "a000" + assert record.coord_hash == "b" * 32 + + def test_a_rider_falls_back_without_its_identity(self): + """In the fallback a rider keeps identity only when alone and whole.""" + spool = dc.get_example_spool("random_das") + patches = [ + x.update_coords( + clock=("time", np.arange(x.shape[x.get_axis("time")]) * 1.0) + ) + for x in spool + ] + joined = dc.spool(patches).concatenate(time=None) + again = joined.chunk(time=None) # re-plan: members are unknown here + frame = again._catalog.to_df() + assert not str(frame["_clock_def_key"].iloc[0]).startswith("fp:") + assert "clock" in again[0].coords.coord_map + + def test_extrema_of_values_which_do_not_compare(self): + """A group holding two kinds of value has no envelope.""" + frame = pd.DataFrame({"code": [0, 1], "value": ["a", 2.0]}) + grouped = frame.groupby("code")["value"] + assert list(_extrema(grouped, "min")) == ["a", 2.0] + mixed = pd.DataFrame({"code": [0, 0], "value": ["a", 2.0]}) + assert list(_extrema(mixed.groupby("code")["value"], "min")) == [None] + + def test_no_members_describes_nothing(self, pair): + """An empty member table claims nothing.""" + plan, backend = self._plan_and_backend(pair, time=None) + empty = plan.members.iloc[:0] + assert predicted_coords(backend, empty, "time") == {} + assert predicted_coords(None, plan.members, "time") == {} + + +class TestWhatAMergeWillNotCarry: + """A row states a coordinate only where the patch will hold it.""" + + @pytest.fixture() + def pair(self): + """Two patches meeting end to end along time.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + return first, second + + def test_rider_only_one_member_states(self, pair, assert_contents_match): + """A merge drops a coordinate its members do not all hold.""" + first, second = pair + samples = first.shape[first.get_axis("time")] + held = first.update_coords(bar=("time", np.arange(float(samples)))) + spool = dc.spool([held, second]) + # a value beside no value is a conflict, so a plain merge refuses + with pytest.raises(CoordMergeError): + spool.chunk(time=None) + merged = spool.chunk(time=None, conflict="drop") + assert "bar" not in merged[0].coords.coord_map + assert "bar_min" not in merged.get_contents().columns + assert_contents_match(merged) + + def test_conflicting_rider_is_dropped_in_the_fallback( + self, pair, assert_contents_match + ): + """A re-plan describes no coordinate the merge drops for conflicting.""" + first, second = pair + axis = first.get_axis("distance") + values = np.arange(float(first.shape[axis])) + left = first.update_coords( + depth=("distance", get_coord(values=values, units="m")) + ) + right = second.update_coords( + depth=("distance", get_coord(values=values, units="ft")) + ) + time = first.get_coord("time") + step = (time.max() - time.min()) / 3 + spool = dc.spool([left, right]).chunk(time=step, conflict="drop") + # the subdivision keeps it wherever an output has one member; the + # output spanning the seam merges two spellings and drops it + assert "depth" in spool[0].coords.coord_map + stated = spool.get_contents()["depth_min"] + assert stated.notnull().any() and stated.isnull().any() + # merging them back drops it, and the row must not still state it + again = spool.chunk(time=None, conflict="drop") + assert "depth" not in again[0].coords.coord_map + assert "depth_min" not in again.get_contents().columns + + +class TestRidersKeepMemberOrder: + """A concatenation lays members end to end; the join sorts them.""" + + def _pair(self, low, high): + """Two contiguous patches carrying a rider on time.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + samples = first.shape[first.get_axis("time")] + return ( + first.update_coords(foo=("time", np.arange(*low, dtype=float))), + second.update_coords(foo=("time", np.arange(*high, dtype=float))), + ), samples + + def test_reversed_rider_claims_no_structure(self, assert_contents_match): + """Blocks running backwards concatenate into an array, not a range.""" + first = dc.get_example_patch() + samples = first.shape[first.get_axis("time")] + (pair, _) = self._pair((samples, 2 * samples), (0, samples)) + spool = dc.spool(list(pair)).concatenate(time=None) + row = spool.get_contents().iloc[0] + assert pd.isnull(row["foo_step"]) + frame = spool._catalog.to_df() + assert not str(frame["_foo_def_key"].iloc[0]).startswith("fp:") + assert_contents_match(spool) + + def test_ordered_rider_keeps_its_identity(self, assert_contents_match): + """Blocks already in order do join into one range.""" + first = dc.get_example_patch() + samples = first.shape[first.get_axis("time")] + (pair, _) = self._pair((0, samples), (samples, 2 * samples)) + spool = dc.spool(list(pair)).concatenate(time=None) + assert spool.get_contents().iloc[0]["foo_step"] == 1.0 + assert_contents_match(spool) + + +class TestMomentsAndDurations: + """A datetime and a timedelta are not two spellings of one kind.""" + + def test_mixed_time_kinds_state_no_envelope(self): + """Neither bounds the other, so the row states neither.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + samples = first.shape[first.get_axis("time")] + moment = get_coord(values=time.values.copy(), units="s") + duration = get_coord( + values=np.arange(samples).astype("timedelta64[s]"), units="s" + ) + spool = dc.spool( + [ + first.update_coords(stamp=("time", moment)), + second.update_coords(stamp=("time", duration)), + ] + ).concatenate(time=None) + assert pd.isnull(spool.get_contents().iloc[0]["stamp_min"]) + + +class TestAgreementNeedsIdentity: + """Two coordinates nobody can identify are unknown, not equal.""" + + def _summary(self, values, fingerprint): + """A distance-riding summary stating (or not) an identity.""" + summary = get_coord(values=values).to_summary() + return summary.model_copy( + update=dict(dims=("distance",), fingerprint=fingerprint) + ) + + def test_unidentified_members_do_not_agree(self): + """A merge told to drop conflicts drops what it cannot compare.""" + left = self._summary(np.arange(4.0), None) + right = self._summary(np.arange(4.0) + 10, None) + described = _describe( + "rough", + [left, right], + "time", + frozenset(), + None, + mode="chunk", + drop_conflicting=True, + ) + assert described is None + + def test_a_lone_member_keeps_what_it_says(self): + """With nothing to agree with, an unidentified member still counts.""" + only = self._summary(np.arange(4.0), None) + described = _describe( + "rough", + [only], + "time", + frozenset(), + None, + mode="chunk", + drop_conflicting=True, + ) + assert described is not None + assert described.min == only.min + + def test_dims_must_match_to_survive_a_merge(self, assert_contents_match): + """A coordinate hung on different dimensions is dropped, not merged.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + values = np.arange(float(first.shape[first.get_axis("distance")])) + # matching values, so no envelope conflict raises first + left = first.update_coords(baz=("distance", values)) + right = second.update_coords( + baz=("time", np.resize(values, first.shape[first.get_axis("time")])) + ) + merged = dc.spool([left, right]).chunk(time=None, conflict="drop") + assert "baz" not in merged[0].coords.coord_map + assert "baz_min" not in merged.get_contents().columns + assert_contents_match(merged) + + def test_one_fingerprint_two_spellings_do_not_agree(self, assert_contents_match): + """A fingerprint is normalized; what a merge compares is not.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + values = np.arange(float(first.shape[first.get_axis("distance")])) + metres = get_coord(values=values, units="m") + centimetres = get_coord(values=values * 100.0, units="cm") + assert metres.to_summary().fingerprint == centimetres.to_summary().fingerprint + left = first.update_coords(depth=("distance", metres)) + right = second.update_coords(depth=("distance", centimetres)) + merged = dc.spool([left, right]).chunk(time=None, conflict="drop") + assert "depth" not in merged[0].coords.coord_map + assert "depth_min" not in merged.get_contents().columns + assert_contents_match(merged) + + def test_one_spelling_still_agrees(self, assert_contents_match): + """Members stating one coordinate the same way keep it.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + values = np.arange(float(first.shape[first.get_axis("distance")])) + coord = get_coord(values=values, units="m") + merged = dc.spool( + [ + first.update_coords(depth=("distance", coord)), + second.update_coords(depth=("distance", coord)), + ] + ).chunk(time=None) + assert "depth" in merged[0].coords.coord_map + assert merged.get_contents().iloc[0]["depth_min"] == 0.0 + assert_contents_match(merged) + + +class TestTrimmedCoordsStayTrimmed: + """A residual trims at load; the record must not outrun it.""" + + def test_a_samples_residual_survives_a_union(self): + """Candidacy is answered from the record, so it holds the trim.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + selected = dc.spool([first]).select(distance=(5, 50), samples=True) + union = selected + dc.spool([second]) + # the trimmed patch holds distance 5..49 and must not be a + # candidate for values only the untrimmed source ever held + elsewhere = union.select(distance=(60, 80)) + assert len(elsewhere) == len(elsewhere.get_contents()) == 1 + assert len(list(elsewhere)) == 1 + loaded = elsewhere[0].get_coord("distance") + assert loaded.min() == 60 and loaded.max() == 80 + + def test_the_record_says_what_the_patch_holds(self): + """The stored envelope matches the trimmed patch, not its source.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + selected = dc.spool([first]).select(distance=(5, 50), samples=True) + union = selected + dc.spool([second]) + frame = union._catalog.backend.coord_frame([1, 2]) + distance = frame[frame["coord_name"] == "distance"] + stated = distance[distance["patch_id"] == 1].iloc[0] + held = union[0].get_coord("distance") + assert stated["min_num"] == held.min() + assert stated["max_num"] == held.max() + assert stated["length"] == len(held) + + +class TestRidersSurviveTheirMerge: + """A rider is joined, not compared, and never snapped.""" + + @pytest.fixture() + def riding(self): + """Two contiguous patches carrying a clock on time.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + samples = first.shape[first.get_axis("time")] + return first, second, samples + + def test_a_replan_keeps_the_rider_it_holds(self, riding): + """Differing definitions are how a rider works, not a conflict.""" + first, second, samples = riding + time = first.get_coord("time") + left = first.update_coords(clock=("time", np.arange(float(samples)))) + right = second.update_coords( + clock=("time", np.arange(float(samples), 2.0 * samples)) + ) + step = (time.max() - time.min()) / 2 + spool = dc.spool([left, right]).chunk(time=step, conflict="drop") + again = spool.chunk(time=None, conflict="drop") + assert "clock" in again[0].coords.coord_map + assert again.get_contents().iloc[0]["clock_min"] == 0.0 + + def test_an_irregular_rider_is_not_snapped(self, riding, assert_contents_match): + """Assembly simplifies the merged dimension and nothing else.""" + first, second, samples = riding + left = first.update_coords(clock=("time", np.arange(float(samples)))) + # the second block starts half a step late: a seam a tolerant + # snap would absorb on the merged dimension, but not here + right = second.update_coords( + clock=("time", np.arange(float(samples)) + samples + 0.5) + ) + merged = dc.spool([left, right]).chunk(time=None, conflict="keep_first") + assert pd.isnull(merged.get_contents().iloc[0]["clock_step"]) + assert_contents_match(merged) + + def test_a_contiguous_rider_keeps_its_step(self, riding, assert_contents_match): + """Not snapping is not the same as claiming nothing.""" + first, second, samples = riding + left = first.update_coords(clock=("time", np.arange(float(samples)))) + right = second.update_coords( + clock=("time", np.arange(float(samples)) + samples) + ) + merged = dc.spool([left, right]).chunk(time=None, conflict="keep_first") + assert merged.get_contents().iloc[0]["clock_step"] == 1.0 + assert_contents_match(merged) + + def test_a_restated_member_is_not_a_trimmed_one(self, assert_contents_match): + """The planner's unit and the index's spelling describe one member.""" + first = dc.get_example_patch() + size = first.shape[first.get_axis("distance")] + values = np.arange(float(size)) + left = first.update_coords( + distance=get_coord(values=values, units="m"), + rider=("distance", values), + ) + right = first.update_coords( + distance=get_coord(values=(values + size) * 100.0, units="cm"), + rider=("distance", values + size), + ) + merged = dc.spool([left, right]).chunk(distance=None, conflict="keep_first") + row = merged.get_contents().iloc[0] + assert row["rider_min"] == 0.0 and row["rider_max"] == 2 * size - 1 + assert_contents_match(merged) + + +class TestFallbackAgreement: + """The fallback mirrors the rules prediction applies.""" + + def _frames(self, keys): + """Source rows and members for one output holding `keys`.""" + count = len(keys) + sources = pd.DataFrame( + { + "_patch_id": list(range(1, count + 1)), + "depth_min": [0.0] * count, + "depth_max": [9.0] * count, + "_depth_def_key": keys, + } + ) + members = pd.DataFrame( + { + "output_id": [0] * count, + "_patch_id": list(range(1, count + 1)), + "_modified": [False] * count, + } + ) + return sources, members + + def test_a_lone_unidentified_member_is_still_described(self): + """`nunique` counts no nulls, and one member disagrees with nobody.""" + sources, members = self._frames([None]) + described = _aux_coord_info( + sources, + members, + "time", + {"depth": "distance"}, + mode="chunk", + drop_conflicting=True, + ) + assert "depth" in described[0] + + def test_members_which_disagree_are_still_dropped(self): + """The exemption is for having nobody to disagree with.""" + sources, members = self._frames(["fp:a", "fp:b"]) + described = _aux_coord_info( + sources, + members, + "time", + {"depth": "distance"}, + mode="chunk", + drop_conflicting=True, + ) + assert "depth" not in described.get(0, {}) + + +class TestWhatTheUnionCarriesOver: + """The fallback describes what raw concatenation will produce.""" + + def test_mixed_numeric_dtypes_promote(self): + """Numpy promotes what it concatenates, so the record must too.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + samples = first.shape[first.get_axis("time")] + rng = np.random.default_rng(0) + left = first.update_coords( + rider=("time", np.sort(rng.integers(0, 100, samples)).astype("int32")) + ) + right = second.update_coords( + rider=("time", np.sort(rng.uniform(200, 300, samples)).astype("float64")) + ) + spool = dc.spool([left, right]).concatenate(time=None) + frame = spool._catalog.backend.coord_frame([0, 1, 2]) + stated = frame[frame["coord_name"] == "rider"]["dtype"].iloc[0] + assert stated == str(spool[0].get_coord("rider").dtype) == "float64" + + def test_one_dtype_is_left_alone(self): + """Promotion is not an excuse to restate what already agrees.""" + first = dc.get_example_patch() + time = first.get_coord("time") + second = dc.get_example_patch(time_min=time.max() + time.step) + samples = first.shape[first.get_axis("time")] + rng = np.random.default_rng(0) + values = np.sort(rng.integers(0, 100, samples)).astype("int32") + spool = dc.spool( + [ + first.update_coords(rider=("time", values)), + second.update_coords(rider=("time", values + 100)), + ] + ).concatenate(time=None) + frame = spool._catalog.backend.coord_frame([0, 1, 2]) + assert frame[frame["coord_name"] == "rider"]["dtype"].iloc[0] == "int32" + + def test_zero_in_two_units_is_not_one_bound(self, assert_contents_match): + """Equal numbers in different spellings are not the same bounds.""" + first = dc.get_example_patch() + time = first.get_coord("time") + data = np.random.default_rng(0).random((1, len(time))) + + def sample(units): + """A one-sample patch whose distance sits at zero.""" + distance = get_coord(values=np.array([0.0]), units=units) + return dc.Patch( + data=data, + coords={"distance": distance, "time": time}, + dims=("distance", "time"), + ) + + spool = dc.spool([sample("m"), sample("cm")]).concatenate(distance=None) + row = spool.get_contents().iloc[0] + assert row["distance_min"] == 0.0 and row["distance_max"] == 0.0 + # the output is real, so a selection over it must keep it + assert len(spool.select(distance=(-1, 1))) == 1 + assert_contents_match(spool) + + +class TestRawJoinsUseRawValues: + """A concatenation lays values end to end; a range regenerates them.""" + + def _pieces(self, lengths, start=-10.0, step=0.1): + """Patches whose distance ranges meet but drift inside their spans.""" + time = dc.get_example_patch().get_coord("time") + patches = [] + for length in lengths: + distance = get_coord(start=start, step=step, stop=start + step * length) + data = np.random.default_rng(0).random((length, len(time))) + patches.append( + dc.Patch( + data=data, + coords={"distance": distance, "time": time}, + dims=("distance", "time"), + ) + ) + start = start + step * length + return patches + + def test_drifting_floats_state_the_step_they_will_have(self, assert_contents_match): + """Boundaries can match while interior samples do not.""" + spool = dc.spool(self._pieces((2, 3, 4))).concatenate(distance=None) + loaded = spool[0].get_coord("distance") + assert spool.get_contents().iloc[0]["distance_step"] == loaded.step + frame = spool._catalog.backend.coord_frame([0, 1, 2, 3]) + stated = frame[frame["coord_name"] == "distance"]["fingerprint"].iloc[0] + assert stated == loaded.fingerprint() + assert_contents_match(spool) + + def test_floats_which_do_not_drift_keep_the_fused_range( + self, assert_contents_match + ): + """Rebuilding from values is a correction, not a policy.""" + spool = dc.spool(self._pieces((2, 3, 4), start=0.0, step=1.0)).concatenate( + distance=None + ) + assert spool.get_contents().iloc[0]["distance_step"] == 1.0 + assert_contents_match(spool) + + def test_integer_widths_promote(self): + """A fused range takes the first dtype; concatenation promotes.""" + time = dc.get_example_patch().get_coord("time") + + def piece(values): + """A patch whose distance holds exactly these values.""" + data = np.random.default_rng(0).random((len(values), len(time))) + return dc.Patch( + data=data, + coords={"distance": get_coord(values=values), "time": time}, + dims=("distance", "time"), + ) + + spool = dc.spool( + [ + piece(np.arange(0, 5, dtype="int32")), + piece(np.arange(5, 10, dtype="int64")), + ] + ).concatenate(distance=None) + loaded = spool[0].get_coord("distance") + frame = spool._catalog.backend.coord_frame([0, 1, 2]) + stated = frame[frame["coord_name"] == "distance"] + assert stated["dtype"].iloc[0] == str(loaded.dtype) == "int64" + assert stated["fingerprint"].iloc[0] == loaded.fingerprint() + + +class TestCutRiders: + """A coordinate riding a cut dimension is sliced along with it.""" + + def _summaries(self, samples=100): + """A whole dimension and a rider of the same length.""" + start = np.datetime64("2020-01-01") + step = np.timedelta64(1, "s") + whole = get_coord(start=start, step=step, stop=start + step * samples) + rider = get_coord(values=np.arange(float(samples))) + return whole.to_summary(), rider.to_summary() + + def test_the_slice_is_exact(self): + """Evenly sampled members give the cut exactly.""" + whole, rider = self._summaries() + start = np.datetime64("2020-01-01") + row = { + "time_min": start + np.timedelta64(10, "s"), + "time_max": start + np.timedelta64(20, "s"), + } + sliced = _cut_rider(rider, whole, row, "time") + assert sliced.min == 10.0 and sliced.max == 20.0 and sliced.len == 11 + + def test_an_unmeasured_dimension_says_nothing(self): + """Without a grid on both sides the slice cannot be worked out.""" + whole, rider = self._summaries() + row = {"time_min": np.datetime64("2020-01-01"), "time_max": None} + assert _cut_rider(rider, None, row, "time") is None + assert _cut_rider(rider, whole, row, "time") is None + stepless = rider.model_copy(update=dict(step=None)) + row = { + "time_min": np.datetime64("2020-01-01"), + "time_max": np.datetime64("2020-01-01") + np.timedelta64(5, "s"), + } + assert _cut_rider(stepless, whole, row, "time") is None + + def test_an_array_rider_keeps_no_envelope(self, assert_contents_match): + """What cannot be sliced is not guessed at.""" + first = dc.get_example_patch() + time = first.get_coord("time") + samples = first.shape[first.get_axis("time")] + rng = np.random.default_rng(0) + rough = first.update_coords(rough=("time", np.sort(rng.uniform(0, 1, samples)))) + spool = dc.spool([rough]).chunk( + time=(time.max() - time.min()) / 2, conflict="keep_first" + ) + assert spool.get_contents()["rough_min"].isnull().all() + assert "rough" in spool[0].coords.coord_map + + +class TestRestatedDtype: + """A bound the plan restated says what the coordinate becomes.""" + + def _summary(self): + """An integer summary in centimetres.""" + values = np.arange(0, 500, 100, dtype="int32") + return get_coord(values=values, units="cm").to_summary() + + def test_a_scaled_integer_becomes_a_float(self): + """Converting an integer grid by a fraction gives floats.""" + assert _stated_dtype(np.float64(1.0), self._summary()) == "float64" + + def test_a_bound_which_says_nothing_changes_nothing(self): + """A value of no kind cannot restate the coordinate's own.""" + summary = self._summary() + assert _stated_dtype(None, summary) == summary.dtype