From 578dd449c30ea106a7f9689a58052de7d0de6154 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Fri, 7 Aug 2026 13:13:01 +0200 Subject: [PATCH 1/3] Chunk spools by data size and by unit-bearing lengths Spool.chunk now accepts a pint quantity as the chunk length. A quantity in the coordinate's own units works (time=10*s, distance=100*ft), which previously raised NotImplementedError, and a quantity of information (time=25*megabytes) chunks so each output patch's data array is at most the requested size. Overlap accepts both forms. A size resolves per partition, since the conversion needs that partition's sampling interval, its extent along the other dimensions, and its element dtype: bytes_per_sample = itemsize * prod(other dims' sample counts) n_samples = floor(requested_bytes / bytes_per_sample) The count is floored so the data never exceeds the request, and is computed against the partition's smallest step: steps within sampling_group_tolerance share a partition, so sizing against the median would let a faster-sampled member overshoot. To make this a pure metadata operation, the index now records each patch's element dtype (INDEX_VERSION 5 -> 6) and surfaces it to the spool relation as a private _dtype column. The leading underscore is load-bearing: chunk's merge-compatibility grouping compares all non-private columns, so a public dtype column would raise CoordMergeError on every merge of patches with differing element types. dtype is deliberately not a partition key, so a partition may mix dtypes; the estimate uses np.result_type, matching the upcast assembly performs. Spool.chunk_plan(...).params["size"] reports what a size resolved to. The low_freq_proc recipe now uses the new API instead of hand-computing bytes per second from a loaded patch. --- dascore/core/spool.py | 24 +- dascore/io/index/backend.py | 1 + dascore/io/index/catalog.py | 4 +- dascore/io/index/indexer.py | 4 +- dascore/io/index/ingest.py | 5 +- dascore/io/index/planned.py | 3 + dascore/io/index/schema.py | 11 +- dascore/units.py | 60 +++++ dascore/utils/chunk_plan.py | 223 +++++++++++++++++- docs/changelog.qmd | 1 + docs/notes/spool_chunking.qmd | 33 +++ docs/recipes/low_freq_proc.qmd | 43 ++-- docs/tutorial/spool.qmd | 14 ++ tests/test_core/test_patch_chunk.py | 195 ++++++++++++++- .../test_io/test_index/test_index_contract.py | 26 ++ tests/test_units.py | 68 ++++++ tests/test_utils/test_chunk.py | 105 ++++++++- 17 files changed, 782 insertions(+), 38 deletions(-) diff --git a/dascore/core/spool.py b/dascore/core/spool.py index 23c6817b8..9cbd00410 100644 --- a/dascore/core/spool.py +++ b/dascore/core/spool.py @@ -216,20 +216,35 @@ def chunk( kwargs kwargs are used to specify the dimension along which to chunk, eg: `time=10` chunks along the time axis in 10 second increments. + The value may also be a quantity: one of the coordinate's own + units (`time=10 * s`) or a data size (`time=25 * megabytes`), + which chunks so each patch's data array is about that large. + `overlap` accepts the same forms. Examples -------- >>> import dascore as dc - >>> from dascore.units import s + >>> from dascore.units import s, megabytes >>> >>> spool = dc.get_example_spool("random_das") >>> # get spools with time duration of 10 seconds >>> time_chunked = spool.chunk(time=10, overlap=1) + >>> # the same, with the units stated explicitly + >>> unit_chunked = spool.chunk(time=10 * s) + >>> # get patches whose data arrays are at most ~1 MB + >>> size_chunked = spool.chunk(time=1 * megabytes) >>> # merge along time axis >>> time_merged = spool.chunk(time=...) Notes ----- + A data size measures the patch's data array only; coordinates and + attrs are extra, as are any copies a later processing step makes, + so the patch as a whole is somewhat larger. The sample count is + rounded down, so the data never exceeds the requested size, and a + merge of patches with different dtypes is sized against the dtype + they upcast to. + [`Spool.concatenate`](`dascore.BaseSpool.concatenate`) performs a similar operation but disregards the coordinate values. @@ -818,7 +833,7 @@ def chunk( def concatenate(self, check_behavior: WARN_LEVELS = "warn", **kwargs) -> Self: """{desc}""" from dascore.io.index.planned import derived_catalog - from dascore.utils.chunk_plan import ChunkPlan + from dascore.utils.chunk_plan import ChunkPlan, _combined_dtype if len(kwargs) != 1: msg = ( @@ -848,6 +863,11 @@ def concatenate(self, check_behavior: WARN_LEVELS = "warn", **kwargs) -> Self: ) member_frames.append(members) first = group_rows.iloc[0].to_dict() + if "_dtype" in group_rows.columns: + # concatenation upcasts like a merge does, so the group's + # dtype is what the members combine to, not the first row's + combined = _combined_dtype(group_rows["_dtype"]) + first["_dtype"] = "" if combined is None else str(combined) if has_envelope: first[f"{dim}_min"] = group_rows[f"{dim}_min"].min() first[f"{dim}_max"] = group_rows[f"{dim}_max"].max() diff --git a/dascore/io/index/backend.py b/dascore/io/index/backend.py index 63fb94379..95d0a5385 100644 --- a/dascore/io/index/backend.py +++ b/dascore/io/index/backend.py @@ -483,6 +483,7 @@ def write_sources(self, records: list[SourceRecord]) -> None: patch.n_dims, patch.dims, patch.shape, + patch.dtype, patch.sample_count_total, patch.time_min, patch.time_max, diff --git a/dascore/io/index/catalog.py b/dascore/io/index/catalog.py index dee725bb0..f5bd24e3b 100644 --- a/dascore/io/index/catalog.py +++ b/dascore/io/index/catalog.py @@ -38,7 +38,7 @@ InvalidSpoolQueryError, Query, ) -from dascore.io.index.schema import SPOOL_HIDDEN_COLUMNS +from dascore.io.index.schema import SPOOL_HIDDEN_COLUMNS, SPOOL_PRIVATE_RENAMES from dascore.utils.misc import is_range from dascore.utils.paths import is_memory_uri from dascore.utils.pd import adjust_segments, relative_ranges_to_absolute @@ -922,7 +922,7 @@ def to_df(self) -> pd.DataFrame: "patch_id", key=lambda s: s.map(position), kind="stable" ).reset_index(drop=True) df = df.drop(columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore").rename( - columns={"patch_id": "_patch_id"} + columns=dict(SPOOL_PRIVATE_RENAMES) ) # SQL identifies overlapping source patches. Expose the selected # envelopes, matching spool.get_contents() and the exact trim diff --git a/dascore/io/index/indexer.py b/dascore/io/index/indexer.py index da5f70afe..afb8d02f8 100644 --- a/dascore/io/index/indexer.py +++ b/dascore/io/index/indexer.py @@ -31,7 +31,7 @@ hive_path_attrs, summaries_to_records, ) -from dascore.io.index.schema import SPOOL_HIDDEN_COLUMNS +from dascore.io.index.schema import SPOOL_HIDDEN_COLUMNS, SPOOL_PRIVATE_RENAMES from dascore.utils.misc import _iter_filesystem from dascore.utils.paths import ( coerce_to_local_path, @@ -459,7 +459,7 @@ def get_contents(self, _attrs=None, _coords=None, **kwargs) -> pd.DataFrame: query = resolve_query(self._backend, _attrs=_attrs, _coords=_coords, **kwargs) df = self._backend.query(query) df = df.drop(columns=list(SPOOL_HIDDEN_COLUMNS), errors="ignore") - return df.rename(columns={"patch_id": "_patch_id"}) + return df.rename(columns=dict(SPOOL_PRIVATE_RENAMES)) __call__ = get_contents diff --git a/dascore/io/index/ingest.py b/dascore/io/index/ingest.py index 0ebbc844f..8b3a8919c 100644 --- a/dascore/io/index/ingest.py +++ b/dascore/io/index/ingest.py @@ -127,6 +127,7 @@ class PatchRecord: source_patch_id: str dims: str shape: str + dtype: str n_dims: int sample_count_total: int | None time_min: int | None @@ -395,6 +396,7 @@ def patch_record(summary: PatchSummary) -> PatchRecord: source_patch_id=normalize_source_patch_id(summary.source_patch_id), dims=",".join(summary.dims), shape=",".join(str(x) for x in shape), + dtype=str(summary.dtype or ""), n_dims=len(summary.dims), sample_count_total=int(np.prod(shape)) if shape else None, time_min=time_min, @@ -496,7 +498,7 @@ def summaries_to_records( _PATCH_ROW_FIELDS = tuple( f.name for f in fields(PatchRecord) - if f.name not in ("source_patch_id", "dims", "shape", "attrs", "coords") + if f.name not in ("source_patch_id", "dims", "shape", "dtype", "attrs", "coords") ) # Backends with no boolean type hand these columns back as 0/1, which # def_key would hash differently than the bool a scan produced. @@ -609,6 +611,7 @@ def assemble_source_records( source_patch_id=normalize_source_patch_id(patch.source_patch_id), dims=_py_scalar(patch.dims) or "", shape=_py_scalar(patch.shape) or "", + dtype=_py_scalar(patch.dtype) or "", attrs=typed, coords=tuple(coords), **{f: _py_scalar(getattr(patch, f)) for f in _PATCH_ROW_FIELDS}, diff --git a/dascore/io/index/planned.py b/dascore/io/index/planned.py index 3118a1dd8..f2f458bef 100644 --- a/dascore/io/index/planned.py +++ b/dascore/io/index/planned.py @@ -268,6 +268,9 @@ def _output_records( source_patch_id=str(output_id), dims=dims, shape="", + # the plan carries the element dtype privately so a chained + # chunk can still size patches by their memory footprint + dtype=str(row.get("_dtype") or ""), n_dims=len(dim_names), sample_count_total=None, time_min=_ns(row.get("time_min")), diff --git a/dascore/io/index/schema.py b/dascore/io/index/schema.py index f60022789..07198308f 100644 --- a/dascore/io/index/schema.py +++ b/dascore/io/index/schema.py @@ -25,7 +25,7 @@ from typing import NamedTuple, get_args, get_type_hints # Version of the index schema, independent of dascore's version. -INDEX_VERSION = 5 +INDEX_VERSION = 6 # Identity string so any tool can sanity-check what it opened. WHAT_IS_THIS = "dascore_spool_index" @@ -100,6 +100,7 @@ class PatchRow(NamedTuple): n_dims: int dims: str shape: str + dtype: str # the data array's dtype, eg "float64" sample_count_total: int | None time_min: int | None # epoch ns; NULL for relative-time patches time_max: int | None @@ -265,6 +266,7 @@ def _columns(row_type: type[NamedTuple]) -> MappingProxyType[str, str]: "n_dims", "dims", "shape", + "dtype", "sample_count_total", "coord_def_id", "def_key", @@ -297,6 +299,13 @@ def _columns(row_type: type[NamedTuple]) -> MappingProxyType[str, str]: # non-private columns. SPOOL_HIDDEN_COLUMNS = ("n_dims", "sample_count_total", "shape") +# Structural columns the spool relation carries *privately*. The leading +# underscore is load-bearing, not cosmetic: chunk's merge-compatibility +# grouping and conflict policing both compare all non-private columns, so +# a public `dtype` would raise CoordMergeError on every merge of patches +# with differing element types. +SPOOL_PRIVATE_RENAMES = MappingProxyType({"patch_id": "_patch_id", "dtype": "_dtype"}) + # Explicit secondary indexes. Every other access path is covered by a # PRIMARY KEY or UNIQUE autoindex above — patch_coords(patch_id, # coord_name), sources(base_uri, source_path), patches(source_id, diff --git a/dascore/units.py b/dascore/units.py index 5e3261e24..cd6558629 100644 --- a/dascore/units.py +++ b/dascore/units.py @@ -469,6 +469,66 @@ def is_percent(value: Any) -> bool: return isinstance(value, Quantity) and value.units == get_unit("percent") +def is_data_size(value: Any) -> bool: + """ + Return True if value is a quantity of information (bytes, bits, MB, ...). + + Pint treats information as dimensionless, so a compatibility check + against bytes also passes for percents and bare dimensionless + quantities. The base unit is the only reliable discriminator. + + Parameters + ---------- + value + Any value of any type to test if it is a data size quantity. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.units import is_data_size + >>> + >>> assert is_data_size(25 * dc.units.megabytes) + >>> assert is_data_size(dc.get_quantity("1 MiB")) + >>> + >>> # Percents, strain and plain numbers are not sizes. + >>> assert not is_data_size(dc.get_quantity("50%")) + >>> assert not is_data_size(25) + """ + return isinstance(value, Quantity) and value.to_base_units().units == get_unit( + "bit" + ) + + +def get_byte_count(value: Quantity) -> float: + """ + Return the number of bytes a data size quantity represents. + + Parameters + ---------- + value + A quantity of information (eg 25 * dc.units.megabytes). + + Notes + ----- + Do not use [`to_float`](`dascore.utils.time.to_float`) for this. It + falls back to `float(value)`, and pint converts a dimensionless + quantity to its base units, which for information is *bits*, so a + size would come back eight times too large. + + Examples + -------- + >>> import dascore as dc + >>> from dascore.units import get_byte_count + >>> + >>> assert get_byte_count(25 * dc.units.megabytes) == 25_000_000 + >>> assert get_byte_count(dc.get_quantity("1 MiB")) == 1_048_576 + """ + if not is_data_size(value): + msg = f"Expected a data size quantity (eg '25 MB'), got {value!r}." + raise UnitError(msg) + return value.to("byte").magnitude + + def maybe_convert_percent_to_fraction(obj): """ Iterate an object and convert any percentages to fractions. diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index 1d782f63b..894beeb4c 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -29,7 +29,9 @@ CoordMergeError, InvalidSpoolQueryError, ParameterError, + UnitError, ) +from dascore.units import DimensionalityError, Quantity, get_byte_count, is_data_size from dascore.utils.attrs import validate_conflict from dascore.utils.chunk import get_intervals from dascore.utils.misc import get_middle_value, is_range @@ -317,6 +319,190 @@ def _coerce_length_overlap(value, overlap, start_dtype): return value, overlap +def _needs_partition_resolution(value, overlap) -> bool: + """ + True when a chunk length or overlap can only be resolved per partition. + + Quantities need the partition's units, sampling interval, and (for + data sizes) its element dtype, none of which are known before the + partitions exist. + """ + return isinstance(value, Quantity) or isinstance(overlap, Quantity) + + +def _combined_dtype(dtypes: pd.Series) -> np.dtype | None: + """ + Return the dtype an assembled merge of these members would produce. + + Assembly upcasts a mixed-dtype merge with `np.result_type`, so the + size estimate must use the same rule. Returns None when the relation + carries no usable dtype; planning never fails on bad metadata here, + it fails later with an explanatory error. + """ + values = {str(x) for x in dtypes.dropna().unique() if str(x)} + if not values: + return None + try: + return np.result_type(*(np.dtype(x) for x in sorted(values))) + except TypeError: + return None + + +def _slab_samples(sub: pd.DataFrame, name: str) -> int | None: + """ + Samples in one index-slab along `name`. + + This is the product of the *other* dimensions' sample counts, i.e. + how many elements one step along `name` costs. The widest member of + the partition wins, so the byte estimate bounds every member rather + than just the first. Returns None when any count is underivable. + """ + if "dims" in sub.columns: + dims = str(sub["dims"].iloc[0]).split(",") + else: # bare frames (no dims column) fall back to complete triples + candidates = (x[: -len("_min")] for x in sub.columns if x.endswith("_min")) + dims = [ + x for x in candidates if {f"{x}_max", f"{x}_step"}.issubset(sub.columns) + ] + total = pd.Series(1.0, index=sub.index) + for dim in dims: + if not dim or dim == name: + continue + cols = [f"{dim}_min", f"{dim}_max", f"{dim}_step"] + if not set(cols).issubset(sub.columns): + return None + mins, maxs, steps = (sub[c] for c in cols) + with np.errstate(invalid="ignore", divide="ignore"): + ratio = to_float((maxs - mins).values) / np.abs(to_float(steps.values)) + counts = np.round(ratio) + 1 + if not np.all(np.isfinite(counts)): + return None + total = total * counts + return int(total.max()) + + +def _size_to_length(size_quant, sub, name, size_step): + """ + Convert a data size into a chunk length along `name`. + + The sample count is floored so an output's data never exceeds the request, + and clamped to one sample when a single slab is already larger than + the target (reported back through the returned diagnostics). + """ + dtype = _combined_dtype(sub["_dtype"]) if "_dtype" in sub.columns else None + if dtype is None: + msg = ( + f"Cannot chunk by data size along {name!r}: the patch relation " + "carries no dtype information. Hand-built dataframes and spools " + "indexed by an older DASCore lack it; delete and rebuild the " + "index, or pass an explicit length instead of a size." + ) + raise ChunkError(msg) + slab = _slab_samples(sub, name) + if slab is None: + msg = ( + f"Cannot chunk by data size along {name!r}: the sample count of " + "another dimension cannot be determined from its envelope " + "(missing or non-uniform sampling)." + ) + raise ChunkError(msg) + step_float = np.abs(to_float(size_step)) + if not np.isfinite(step_float) or step_float == 0: + msg = ( + f"Cannot chunk by data size along {name!r}: the sampling " + "interval is unknown." + ) + raise ChunkError(msg) + bytes_per_sample = dtype.itemsize * slab + requested = get_byte_count(size_quant) + samples = int(requested // bytes_per_sample) + clamped = samples < 1 + samples = max(samples, 1) + diagnostics = { + "dtype": str(dtype), + "itemsize": int(dtype.itemsize), + "slab_samples": int(slab), + "bytes_per_sample": int(bytes_per_sample), + "n_samples": samples, + "clamped": clamped, + } + return samples * size_step, diagnostics + + +def _quantity_to_dim_value(quant, sub, name, start_dtype): + """Convert a (non-size) quantity into the chunked dimension's units.""" + if is_datetime64(start_dtype) or is_timedelta64(start_dtype): + try: + seconds = quant.to("s").magnitude + except DimensionalityError: + msg = ( + f"Cannot chunk {name!r} by {quant}: the coordinate is " + "time-like, so the value must have units of time." + ) + raise UnitError(msg) from None + return to_timedelta64(seconds) + units_col = f"_{name}_units" + if units_col in sub.columns: + units = sub[units_col].iloc[0] + if units is None or pd.isnull(units) or units == "": + msg = ( + f"Cannot chunk {name!r} by {quant}: the coordinate has no " + "units, so a unit-bearing length is ambiguous." + ) + raise UnitError(msg) + try: + return quant.to(units).magnitude + except DimensionalityError: + msg = ( + f"Cannot chunk {name!r} by {quant}: incompatible with the " + f"coordinate's units of {units}." + ) + raise UnitError(msg) from None + # bare frames carry no units column; envelopes are canonical SI. + return quant.to_base_units().magnitude + + +def _resolve_partition_length(value, overlap, sub, name, size_step, start_dtype): + """ + Resolve one partition's chunk length and overlap. + + Returns (length, overlap, diagnostics); diagnostics is None unless a + data size was resolved. + """ + diagnostics = None + + def _resolve(val, is_overlap): + nonlocal diagnostics + if not isinstance(val, Quantity): + return val + if is_data_size(val): + # a negative overlap opens a gap; floor the magnitude so the + # gap widens monotonically rather than flipping direction + sign = -1 if val.magnitude < 0 else 1 + length, diag = _size_to_length(abs(val), sub, name, size_step) + if not is_overlap: + diagnostics = diag + return sign * length + if val.dimensionless: + msg = ( + f"Cannot chunk {name!r} by {val}: a dimensionless quantity " + "has no meaning as a length. Use a data size (eg '25 MB') " + "or the coordinate's units." + ) + raise UnitError(msg) + return _quantity_to_dim_value(val, sub, name, start_dtype) + + value_out = _resolve(value, False) + overlap_out = _resolve(overlap, True) + # a size-derived length is already in the dimension's own units; a + # plain number against a time-like dim still needs the usual coercion + if not isinstance(value, Quantity) or not is_data_size(value): + value_out, _ = _coerce_length_overlap(value_out, None, start_dtype) + if not isinstance(overlap, Quantity) or not is_data_size(overlap): + _, overlap_out = _coerce_length_overlap(None, overlap_out, start_dtype) + return value_out, overlap_out, diagnostics + + def _coord_owner(col: str, coord_names: set[str]) -> str | None: """ Return the coordinate owning an envelope column, if any. @@ -376,6 +562,13 @@ def _police_columns(sub: pd.DataFrame, name, conflict) -> dict: col = f"_{coord}_units" if col in sub.columns: carried[col] = sub[col].iloc[0] + # The element dtype carries privately (see SPOOL_PRIVATE_RENAMES): a + # size-based chunk needs it, and a partition may legitimately mix + # dtypes, so the carried value is the one assembly will upcast to. + if "_dtype" in sub.columns: + combined = _combined_dtype(sub["_dtype"]) + if combined is not None: + carried["_dtype"] = str(combined) return carried @@ -560,7 +753,10 @@ def build_chunk_plan( "may be unevenly sampled, or have their sampling rate increased." ) warnings.warn(msg, UserWarning, stacklevel=_user_stacklevel()) - value_c, overlap_c = _coerce_length_overlap(value, overlap, df[min_name].dtype) + per_partition = _needs_partition_resolution(value, overlap) + if not per_partition: + value_c, overlap_c = _coerce_length_overlap(value, overlap, df[min_name].dtype) + size_diagnostics: list[dict] = [] out_frames, member_frames = [], [] next_id = 0 # Deterministic partition order (spec 8): by (partition min, smallest @@ -574,6 +770,17 @@ def build_chunk_plan( start, stop, step = get_interval_columns(sub, name) part_step = get_middle_value(step.values) # D7: one step everywhere g_start, g_stop = start.min(), stop.max() + if per_partition and not merge_mode: + # Size against the partition's *smallest* step, not its median: + # steps within sampling_group_tolerance share a partition, so a + # member sampled faster than the median would otherwise fit more + # samples into the length and overshoot the requested size. + size_step = step.abs().min() + value_c, overlap_c, diag = _resolve_partition_length( + value, overlap, sub, name, size_step, df[min_name].dtype + ) + if diag is not None: + size_diagnostics.append({"first_output_id": next_id, **diag}) if merge_mode: start_stop = np.atleast_2d(np.asarray([g_start, g_stop])) else: @@ -607,6 +814,20 @@ def build_chunk_plan( outputs = outputs[outputs["output_id"].isin(fed)] out_frames.append(outputs) member_frames.append(members) + if size_diagnostics: + params["size"] = { + "requested_bytes": get_byte_count(value), + "partitions": tuple(size_diagnostics), + } + # warn once per call, not once per partition + if clamped := sum(x["clamped"] for x in size_diagnostics): + msg = ( + f"A single sample along {name!r} is larger than the requested " + f"size of {value} in {clamped} partition(s), so those patches " + "contain one sample each and exceed the request. Chunk another " + "dimension first to make them smaller." + ) + warnings.warn(msg, UserWarning, stacklevel=_user_stacklevel()) if not out_frames or all(x.empty for x in out_frames): msg = "Could not chunk. No segments with sufficient length found." raise ChunkError(msg) diff --git a/docs/changelog.qmd b/docs/changelog.qmd index 6b4d909cb..c88ce9eab 100644 --- a/docs/changelog.qmd +++ b/docs/changelog.qmd @@ -4,6 +4,7 @@ The [releases page](https://github.com/DASDAE/dascore/releases) tracks changes f ## Unreleased API Changes +- **`Spool.chunk` accepts quantities as the chunk length.** A quantity in the coordinate's own units works (`chunk(time=10 * dc.units.s)`, `chunk(distance=100 * dc.units.ft)`), where any quantity previously raised `NotImplementedError`. A quantity of information (`chunk(time=25 * dc.units.megabytes)`) chunks so each patch's *data array* is at most the requested size; the sample count is floored, and a partition mixing element types is sized against the dtype assembly upcasts to. `overlap` accepts both forms, and `Spool.chunk_plan(...).params["size"]` reports what a size resolved to. The index schema records each patch's element dtype to make this possible, so its version is bumped and existing indexes must be deleted and rebuilt. An attr named `dtype` is now reserved and stays unindexed. - `Patch.drop_coords` and `CoordManager.drop_coords` accept a sequence of names as well as bare names, so `patch.drop_coords(["latitude", "longitude"])` works alongside `patch.drop_coords("latitude", "longitude")`. Previously a list or set raised `TypeError` and a tuple or generator was silently ignored; a tuple now drops the named coordinates, and one naming a dimension raises `ParameterError` as a bare dimension name always has. - **Runtime configuration is now two-tier.** `dc.set_config(...)` sets the process-wide base permanently (visible from every thread and task) and returns the new config; it is no longer a context manager. Temporary, thread/task-local overrides use the new `dc.config_context(...)` context manager, which restores on exit and isolates concurrent overrides via a `ContextVar`. `Spool.map(...)` captures the config active at the call and re-applies it in each worker, so thread- and process-pool workers observe the caller's config regardless of the pool's start method. Unknown config field names now raise instead of being silently ignored. Migrate `with dc.set_config(...)` to `with dc.config_context(...)`. - PRODML 2.1 now supports writing one raw time-by-distance patch with `dc.write` as a standalone HDF5 file. Full PRODML 2.3 conformance requires EPC/XML packaging, which DASCore does not write. diff --git a/docs/notes/spool_chunking.qmd b/docs/notes/spool_chunking.qmd index 403a3816d..a53d2a5e8 100644 --- a/docs/notes/spool_chunking.qmd +++ b/docs/notes/spool_chunking.qmd @@ -72,6 +72,39 @@ with pytest.raises(ChunkError, match="non-dimensional coordinate"): dc.spool([aux]).chunk(sensor=100) ``` +## Chunking by size + +A chunk length may be a quantity instead of a bare number: either the coordinate's own units (`time=10 * s`, converted to the envelope's canonical units) or a quantity of information (`time=25 * megabytes`), which sizes each output by its data array. `overlap` accepts the same forms. + +A size resolves per partition, because the conversion needs that partition's sampling interval, its extent along the *other* dimensions, and its element dtype: + +``` +bytes_per_sample = itemsize * (product of the other dimensions' sample counts) +n_samples = floor(requested_bytes / bytes_per_sample) +``` + +The count is floored, so an output's data array never exceeds the request, and it is computed against the partition's *smallest* step — steps within `sampling_group_tolerance` share a partition, so sizing against the median would let a faster-sampled member overshoot. Element dtype is deliberately **not** a partition key: making it one would change which patches merge. A partition may therefore mix dtypes, and the estimate uses `np.result_type` over them, matching the upcast assembly performs. A patch relation with no dtype (a hand-built dataframe, or an index predating the column) raises `ChunkError` rather than guessing an itemsize. + +```{python} +from dascore.units import megabytes + +target = 1 * megabytes +sized = dc.get_example_spool("random_das").chunk(time=target) +assert max(p.data.nbytes for p in sized) <= 1e6 + +# The plan explains what the request resolved to. +plan = dc.get_example_spool("random_das").chunk_plan(time=target) +(part,) = plan.params["size"]["partitions"] +assert part["bytes_per_sample"] == part["itemsize"] * part["slab_samples"] +assert part["n_samples"] == int(1e6) // part["bytes_per_sample"] + +# The dtype carries through a chained chunk, so sizes still work. +chained = dc.get_example_spool("random_das").chunk(distance=100).chunk(time=target) +assert max(p.data.nbytes for p in chained) <= 1e6 +``` + +A size measures the data array alone; coordinates, attrs, and copies made by later processing are extra. When a single sample along the chunked dimension is already larger than the request, the outputs hold one sample each and a warning says so — the request cannot be honored below one sample. + ## Merged coordinates Assembly builds the chunked dimension's coordinate by exact concatenation of the member coordinates: contiguous members fuse to a plain evenly sampled range, and every real seam is recorded. When `snap_coords=True` (default) the result is then simplified with **bounded error** — no coordinate value moves more than `tolerance * step`. A within-tolerance gap therefore comes back as an evenly sampled range whose worst label error is about half the gap (never more than the tolerance); this is the honest replacement for the old unconditional snap, whose error was unbounded. With `snap_coords=False`, or when accumulated gaps exceed the bound, the coordinate stays segmented — exactly non-uniform, with every gap queryable. diff --git a/docs/recipes/low_freq_proc.qmd b/docs/recipes/low_freq_proc.qmd index 9f05b0a32..38eacd007 100644 --- a/docs/recipes/low_freq_proc.qmd +++ b/docs/recipes/low_freq_proc.qmd @@ -55,37 +55,26 @@ Notes: 2. The `memory_safety_factor` is optional and helps prevent getting too close to the memory limit. ```{python} -# Get patch's number of bytes per seconds (based on patch's data type) -pa_bytes_per_second = pa.data.nbytes / pa.seconds -# Define processing factor and safety factor -processing_factor = 5 -memory_safety_factor = 1.2 - -# Calculate memory size required for each second of data to get processed -memory_size_per_second = pa_bytes_per_second * processing_factor * memory_safety_factor -memory_size_per_second_MB = memory_size_per_second / 1e6 - -# Calculate chunk size that can be loaded (in seconds) -chunk_size = memory_limit_MB / memory_size_per_second_MB - -# Ensure `chunk_size` does not exceed the spool length -time_step = sp[0].get_coord('time').step -time_min = sp[0].get_coord('time').min() -time_max = sp[-1].get_coord('time').max() -spool_length = dc.to_float((time_max - time_min + time_step)) -if chunk_size > spool_length: - print( - f"Warning: Specified `chunk_size` ({chunk_size:.2f} seconds) exceeds the spool length " - f"({spool_length:.2f} seconds). Adjusting `chunk_size` to match spool length." - ) - chunk_size = spool_length +# Define processing factor and safety factor +processing_factor = 5 +memory_safety_factor = 1.2 + +# How large each loaded patch may be, once the copies made during +# processing are accounted for. `chunk` takes this size directly and +# works out the sample arithmetic from the patch's dtype and shape. +patch_size = ( + memory_limit_MB * dc.units.megabytes / (processing_factor * memory_safety_factor) +) ``` Next, we need to determine the extent of artifacts introduced by low-pass filtering at the edges of each patch. To achieve this, we apply LF processing to a delta function patch, which contains a unit value at the center and zeros elsewhere. The distorted edges are then identified based on a defined threshold. ```{python} # Retrieve a patch of appropriate size for LF processing that fits into memory -pa_chunked_sp = sp.chunk(time=chunk_size, keep_partial=True)[0] +# (`keep_partial` also covers a budget larger than the whole spool). +pa_chunked_sp = sp.chunk(time=patch_size, keep_partial=True)[0] +# The duration that size worked out to, used to validate the edges below. +chunk_size = pa_chunked_sp.seconds # Create a delta patch based on new patch size delta_pa = dc.get_example_patch("delta_patch", dim="time", patch=pa_chunked_sp) @@ -130,8 +119,8 @@ if np.ceil(edge) >= chunk_size / 2: ## Perform low-frequency processing and save results on disk ```{python} -# First we chunk the spool based on the `chunk_size` and `edge` calculated before. -sp_chunked_overlap = sp.chunk(time=chunk_size, overlap=2*edge, keep_partial=True) +# First we chunk the spool based on the `patch_size` and `edge` calculated before. +sp_chunked_overlap = sp.chunk(time=patch_size, overlap=2*edge, keep_partial=True) # Process each patch in the spool and save the result patch lf_patches = [] diff --git a/docs/tutorial/spool.qmd b/docs/tutorial/spool.qmd index 3ada28ed9..b3bc7fd57 100644 --- a/docs/tutorial/spool.qmd +++ b/docs/tutorial/spool.qmd @@ -258,6 +258,20 @@ subspool = spool.chunk(time=3, overlap=1, keep_partial=True) merged_spool = spool.chunk(time=None) ``` +The chunk length can also be a quantity, either in the coordinate's own units or as a data size. A data size chunks so that each patch's data array is about that large, which is the usual way to keep patches inside a memory budget without working out the sample arithmetic by hand. + +```{python} +from dascore.units import s, megabytes + +# The same 3 second chunk, with the units stated explicitly. +unit_chunked = spool.chunk(time=3 * s) + +# Chunk so each patch's data array is at most ~1 MB. +size_chunked = spool.chunk(time=1 * megabytes) +``` + +A data size measures the data array only; coordinates, attrs, and any copies made later during processing are extra, so the patch as a whole is somewhat larger. The sample count is rounded down, so the patch's data never exceeds the requested size, and `MB` is 10^6^ bytes while `MiB` is 2^20^. `overlap` accepts the same forms. + # concatenate Similar to `chunk`, [`Spool.concatenate`](`dascore.BaseSpool.concatenate`) is used to combine patches together. However, `concatenate` doesn't account for coordinate values along the concatenation axis, and can even be used to create new patch dimensions. Like `chunk`, it is available on every spool and produces a lazy, plan-backed result. diff --git a/tests/test_core/test_patch_chunk.py b/tests/test_core/test_patch_chunk.py index ec4cef3c4..5d6cdcdb6 100644 --- a/tests/test_core/test_patch_chunk.py +++ b/tests/test_core/test_patch_chunk.py @@ -15,7 +15,7 @@ import pytest import dascore as dc -from dascore.exceptions import ChunkError, CoordMergeError, ParameterError +from dascore.exceptions import ChunkError, CoordMergeError, ParameterError, UnitError from dascore.utils.misc import get_middle_value from dascore.utils.time import to_timedelta64 @@ -925,6 +925,36 @@ def test_same_dim_rechunk_still_collapses(self): assert len(rechunk) == 8 assert {x.shape for x in rechunk} == {(300, 500)} + def test_size_after_merge(self, random_spool): + """A merged spool still knows its dtype.""" + target = dc.get_quantity("1 MB") + out = random_spool.chunk(time=None).chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_size_after_other_dim_chunk(self, random_spool): + """A derived catalog carries the dtype to the next chunk.""" + target = dc.get_quantity("1 MB") + out = random_spool.chunk(distance=100).chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_size_then_size(self, random_spool): + """Chunking by size twice narrows the plan monotonically. + + This asserts on the plan rather than the assembled patches: a + same-dim rechunk currently mis-assembles one output regardless of + how the length is expressed (a plain `chunk(time=3.332).chunk( + time=1.664)` hits it too). See #832; tighten this to assert + `nbytes` once that is fixed. + """ + first = random_spool.chunk(time=2 * dc.units.MB) + plan = first.chunk_plan(time=1 * dc.units.MB) + (part,) = plan.params["size"]["partitions"] + step = plan.outputs["time_step"].iloc[0] + spans = plan.outputs["time_max"] - plan.outputs["time_min"] + step + samples = (spans / step).round().astype(int) + assert (samples <= part["n_samples"]).all() + assert len(plan.outputs) > len(first) + class TestMatchMergeUnits: """The member unit normalizer's defensive paths.""" @@ -939,3 +969,166 @@ def test_incompatible_units_pass_through(self): out, kept = _match_merge_units(patch, "distance", target) assert out is patch # unconverted assert kept == target + + +class TestUnitChunkValue: + """Chunk lengths carrying the coordinate's own units.""" + + def test_seconds_match_bare(self, random_spool): + """A duration in seconds equals the bare seconds value.""" + quant = random_spool.chunk(time=3 * dc.units.s) + bare = random_spool.chunk(time=3) + assert len(quant) == len(bare) + assert [x.shape for x in quant] == [x.shape for x in bare] + + def test_other_time_unit(self, random_spool): + """Milliseconds convert to the same chunk as seconds.""" + out = random_spool.chunk(time=3000 * dc.units.ms) + assert len(out) == len(random_spool.chunk(time=3)) + + def test_distance_unit_converts(self, random_spool): + """A distance in feet converts to the coordinate's metres.""" + out = random_spool.chunk(distance=100 * dc.units.ft) + assert len(out) == len(random_spool.chunk(distance=30.48)) + + def test_unitless_coord_raises(self, random_spool): + """A unit-bearing length needs a coordinate with units.""" + patches = [x.set_units(distance=None) for x in random_spool] + with pytest.raises(UnitError, match="no units"): + dc.spool(patches).chunk(distance=100 * dc.units.ft) + + def test_wrong_dimensionality_raises(self, random_spool): + """A length cannot chunk time.""" + with pytest.raises(UnitError, match="time-like"): + random_spool.chunk(time=100 * dc.units.ft) + + +class TestSizeChunk: + """Chunk lengths expressed as a data size.""" + + @staticmethod + def _make(dtype, start, distance=50, samples=400): + """A patch of a given dtype starting at a given time.""" + rng = np.random.default_rng(42) + data = rng.random((distance, samples)).astype(dtype) + coords = { + "distance": np.arange(distance) * 1.0, + "time": start + np.arange(samples) * np.timedelta64(4, "ms"), + } + return dc.Patch(data=data, coords=coords, dims=("distance", "time")) + + @pytest.fixture(scope="class") + def mixed_dtype_spool(self): + """Two contiguous patches whose element types differ.""" + start = np.datetime64("2020-01-01T00:00:00") + second = start + np.timedelta64(1600, "ms") + return dc.spool([self._make("float64", start), self._make("float32", second)]) + + @pytest.mark.parametrize( + "size", ("1 MB", "2 MB", "1 MiB", "500 kB"), ids=lambda x: x.replace(" ", "") + ) + def test_never_exceeds_request(self, random_spool, size): + """Every output patch fits inside the requested size.""" + quant = dc.get_quantity(size) + limit = quant.to("byte").magnitude + sizes = [x.data.nbytes for x in random_spool.chunk(time=quant)] + assert max(sizes) <= limit + # and the request is not trivially under-delivered + assert max(sizes) > 0.8 * limit + + def test_binary_and_decimal_prefixes_differ(self, random_spool): + """MiB is 2**20 bytes while MB is 10**6, as pint defines them.""" + decimal = random_spool.chunk(time=1 * dc.units.MB)[0] + binary = random_spool.chunk(time=1 * dc.units.MiB)[0] + assert binary.data.nbytes > decimal.data.nbytes + assert binary.data.nbytes <= 1024**2 + + def test_smaller_dtype_gives_more_samples(self): + """Half the itemsize fits twice the samples in the same bytes.""" + start = np.datetime64("2020-01-01T00:00:00") + target = dc.get_quantity("40 kB") + wide = dc.spool([self._make("float64", start)]).chunk(time=target) + narrow = dc.spool([self._make("float32", start)]).chunk(time=target) + wide_samples = wide[0].shape[wide[0].get_axis("time")] + narrow_samples = narrow[0].shape[narrow[0].get_axis("time")] + assert narrow_samples == 2 * wide_samples + + def test_mixed_dtype_partition_uses_upcast(self, mixed_dtype_spool): + """A mixed partition is sized against the dtype assembly upcasts to.""" + target = dc.get_quantity("100 kB") + plan = mixed_dtype_spool.chunk_plan(time=target) + (part,) = plan.params["size"]["partitions"] + assert part["dtype"] == "float64" + out = mixed_dtype_spool.chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_slab_larger_than_target_warns(self, random_spool): + """One sample is the floor; the request cannot be honored below it.""" + with pytest.warns(UserWarning, match="larger than the requested size"): + out = random_spool.chunk(time=1 * dc.units.kB) + patch = out[0] + assert patch.shape[patch.get_axis("time")] == 1 + + def test_keep_partial_stays_bounded(self, random_spool): + """A partial segment is smaller than the request, never larger.""" + target = dc.get_quantity("2 MB") + out = random_spool.chunk(time=target, keep_partial=True) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_overlap_in_coord_units(self, random_spool): + """Overlap may use the coordinate's units while the length is a size.""" + target = dc.get_quantity("2 MB") + out = random_spool.chunk(time=target, overlap=1 * dc.units.s) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_overlap_as_size(self, random_spool): + """Overlap may also be expressed as a size.""" + target = dc.get_quantity("2 MB") + plain = random_spool.chunk(time=target) + out = random_spool.chunk(time=target, overlap=1 * dc.units.MB) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + assert len(out) > len(plain) # overlap yields more segments + + def test_overlap_larger_than_length_raises(self, random_spool): + """The existing overlap guard applies to sizes too.""" + with pytest.raises(ParameterError): + random_spool.chunk(time=1 * dc.units.MB, overlap=2 * dc.units.MB) + + def test_descending_coord(self, random_spool): + """A reverse-sorted coordinate chunks by size like any other.""" + patches = [ + x.snap_coords("time").sort_coords("time", reverse=True) + for x in random_spool + ] + target = dc.get_quantity("1 MB") + out = dc.spool(patches).chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_one_dimensional_patch(self): + """With no other dims a slab is one element.""" + start = np.datetime64("2020-01-01T00:00:00") + data = np.arange(1000, dtype="float64") + coords = {"time": start + np.arange(1000) * np.timedelta64(1, "s")} + patch = dc.Patch(data=data, coords=coords, dims=("time",)) + plan = dc.spool([patch]).chunk_plan(time=dc.get_quantity("1 kB")) + (part,) = plan.params["size"]["partitions"] + assert part["slab_samples"] == 1 + assert part["n_samples"] == 1000 // 8 + + def test_three_dimensional_patch(self): + """A slab is the product of every other dimension.""" + start = np.datetime64("2020-01-01T00:00:00") + data = np.zeros((3, 5, 100), dtype="float64") + coords = { + "distance": np.arange(3) * 1.0, + "depth": np.arange(5) * 1.0, + "time": start + np.arange(100) * np.timedelta64(1, "s"), + } + patch = dc.Patch(data=data, coords=coords, dims=("distance", "depth", "time")) + plan = dc.spool([patch]).chunk_plan(time=dc.get_quantity("1 kB")) + (part,) = plan.params["size"]["partitions"] + assert part["slab_samples"] == 15 + + def test_merge_mode_unaffected(self, random_spool): + """A merge takes no length, so sizes never apply.""" + assert len(random_spool.chunk(time=None)) == 1 diff --git a/tests/test_io/test_index/test_index_contract.py b/tests/test_io/test_index/test_index_contract.py index e5321be45..da7fd2609 100644 --- a/tests/test_io/test_index/test_index_contract.py +++ b/tests/test_io/test_index/test_index_contract.py @@ -191,6 +191,32 @@ def test_ordering_deterministic(self, backend): assert not nulls[: (~nulls).sum()].any() +class TestElementDtype: + """The data array's dtype is recorded per patch (size-based chunking).""" + + def test_dtype_round_trips(self, backend): + """Each patch keeps the dtype its summary reported.""" + df = backend.query() + expected = {str(x.source_path): x.dtype for x in make_summaries()} + got = {str(k): v for k, v in zip(df["path"], df["dtype"], strict=True)} + assert got == expected + + def test_dtype_is_private_in_flat_relation(self): + """The spool sees `_dtype`, never a public `dtype` column.""" + import dascore as dc + + spool = dc.get_example_spool("random_das") + df = spool.get_contents() + assert "dtype" not in df.columns + assert set(df["_dtype"]) == {str(spool[0].data.dtype)} + + def test_dtype_attr_is_reserved(self): + """An attr named `dtype` cannot shadow the structural column.""" + from dascore.io.index.schema import RESERVED_ATTR_COLUMNS + + assert "dtype" in RESERVED_ATTR_COLUMNS + + class TestAttrPredicates: """Attr predicates are exact at the index.""" diff --git a/tests/test_units.py b/tests/test_units.py index ac4a5b58e..73ba23527 100644 --- a/tests/test_units.py +++ b/tests/test_units.py @@ -13,15 +13,18 @@ Quantity, assert_dtype_compatible_with_units, convert_units, + get_byte_count, get_factor_and_unit, get_filter_units, get_quantity, get_quantity_str, get_unit, invert_quantity, + is_data_size, maybe_convert_percent_to_fraction, quant_sequence_to_quant_array, ) +from dascore.utils.time import to_float class TestUnitInit: @@ -526,3 +529,68 @@ def test_fork_handler_replaces_held_lock(self): assert new_lock is not old_lock finally: units_module._UNIT_LOCK = old_lock + + +class TestDataSize: + """Tests for identifying and measuring data size quantities.""" + + sizes = ("1 byte", "1 bit", "25 MB", "3 kB", "1 MiB", "2 GiB") + not_sizes = ("1 m", "10 s", "50%", "1 strain", "1 dimensionless") + + @pytest.mark.parametrize("value", sizes) + def test_sizes_detected(self, value): + """Quantities of information are data sizes.""" + assert is_data_size(get_quantity(value)) + + @pytest.mark.parametrize("value", not_sizes) + def test_non_sizes_rejected(self, value): + """Percents and dimensionless quantities are not data sizes.""" + assert not is_data_size(get_quantity(value)) + + @pytest.mark.parametrize("value", (25, 1.0, None, "25 MB")) + def test_non_quantities_rejected(self, value): + """Only quantities can be data sizes.""" + assert not is_data_size(value) + + def test_millibarn_is_not_megabytes(self): + """`mb` is millibarn (an area) in pint; only `MB` is megabytes.""" + assert not is_data_size(dc.units.mb) + assert is_data_size(dc.units.MB) + + def test_undefined_byte_alias_raises(self): + """`KB` is not a pint unit; the kilobyte spelling is `kB`.""" + with pytest.raises(pint.UndefinedUnitError): + dc.units.KB + + @pytest.mark.parametrize( + "value,expected", + ( + ("25 MB", 25_000_000), + ("1 MiB", 1_048_576), + ("1 kB", 1_000), + ("8 bit", 1), + ("1 byte", 1), + ), + ) + def test_byte_count(self, value, expected): + """Byte counts follow pint's decimal/binary prefixes.""" + assert get_byte_count(get_quantity(value)) == expected + + @pytest.mark.parametrize("value", not_sizes) + def test_byte_count_requires_size(self, value): + """Non-sizes cannot be measured in bytes.""" + with pytest.raises(UnitError, match="data size"): + get_byte_count(get_quantity(value)) + + def test_byte_count_is_not_to_float(self): + """ + Guard the bits trap. + + `to_float` falls back to `float(quantity)`, and pint converts a + dimensionless quantity to base units, where information is bits. + Any "simplification" of get_byte_count to to_float makes every + size eight times too large. + """ + quant = get_quantity("25 MB") + assert get_byte_count(quant) == 25_000_000 + assert to_float(quant) == 8 * 25_000_000 diff --git a/tests/test_utils/test_chunk.py b/tests/test_utils/test_chunk.py index 92ab76083..6cb6da940 100644 --- a/tests/test_utils/test_chunk.py +++ b/tests/test_utils/test_chunk.py @@ -9,7 +9,7 @@ import pytest import dascore as dc -from dascore.exceptions import ChunkError, ParameterError +from dascore.exceptions import ChunkError, ParameterError, UnitError from dascore.utils.chunk import get_intervals from dascore.utils.chunk_plan import build_chunk_plan from dascore.utils.time import to_timedelta64 @@ -350,3 +350,106 @@ def test_modified_flag_no_chunk(self, contiguous_df): ) assert len(plan.outputs) == len(df) assert not plan.members["_modified"].any() + + +class TestQuantityChunkValues: + """Chunk lengths given as quantities of the dimension's own units.""" + + def test_seconds_match_bare_value(self, contiguous_df): + """A time in seconds equals the bare (seconds) value.""" + quant = build_chunk_plan(contiguous_df, time=5 * dc.units.s) + bare = build_chunk_plan(contiguous_df, time=5) + assert quant.outputs.equals(bare.outputs) + + def test_other_time_unit(self, contiguous_df): + """Any time unit converts to the coordinate's own scale.""" + quant = build_chunk_plan(contiguous_df, time=5000 * dc.units.ms) + bare = build_chunk_plan(contiguous_df, time=5) + assert quant.outputs.equals(bare.outputs) + + def test_numeric_dim_unit(self, contiguous_df): + """Numeric envelopes are canonical SI, so mm converts to m.""" + # one row, so chunking distance has no time envelopes to merge + single = contiguous_df.iloc[[0]] + quant = build_chunk_plan(single, distance=2000 * dc.units.mm) + # a quantity's magnitude is a float, so compare to the float value + bare = build_chunk_plan(single, distance=2.0) + assert quant.outputs.equals(bare.outputs) + + def test_overlap_quantity(self, contiguous_df): + """Overlap accepts the same forms as the chunk length.""" + quant = build_chunk_plan( + contiguous_df, time=5 * dc.units.s, overlap=1000 * dc.units.ms + ) + bare = build_chunk_plan(contiguous_df, time=5, overlap=1) + assert quant.outputs.equals(bare.outputs) + + def test_wrong_dimensionality_raises(self, contiguous_df): + """A length cannot chunk a time coordinate.""" + with pytest.raises(UnitError, match="time-like"): + build_chunk_plan(contiguous_df, time=100 * dc.units.ft) + + def test_percent_raises(self, contiguous_df): + """A percent is dimensionless but is not a data size.""" + with pytest.raises(UnitError, match="dimensionless"): + build_chunk_plan(contiguous_df, time=dc.get_quantity("50%")) + + @pytest.mark.parametrize( + "value", (0 * dc.units.s, -1 * dc.units.s, 0 * dc.units.MB, -1 * dc.units.MB) + ) + def test_non_positive_raises(self, contiguous_df, value): + """The positive-value guard applies to quantities too.""" + with pytest.raises(ParameterError, match="greater than 0"): + build_chunk_plan(contiguous_df, time=value) + + +class TestSizeChunkPlanDF: + """Data-size chunk lengths on hand-built dataframes.""" + + @pytest.fixture() + def sized_df(self, contiguous_df): + """A frame carrying the structural columns a size chunk needs.""" + return contiguous_df.assign(_dtype="float64", dims="distance,time") + + def test_missing_dtype_raises(self, contiguous_df): + """A frame with no dtype cannot answer a size request.""" + with pytest.raises(ChunkError, match="no dtype information"): + build_chunk_plan(contiguous_df, time=1 * dc.units.MB) + + def test_null_dtype_raises(self, sized_df): + """Nulls are as unusable as an absent column.""" + df = sized_df.assign(_dtype=None) + with pytest.raises(ChunkError, match="no dtype information"): + build_chunk_plan(df, time=1 * dc.units.MB) + + def test_unknown_step_raises(self, sized_df): + """An unknown sampling interval makes the sample count undefined.""" + df = sized_df.assign(distance_step=np.nan) + with pytest.raises(ChunkError, match="cannot be determined"): + build_chunk_plan(df, time=1 * dc.units.MB) + + def test_sample_count(self, sized_df): + """The count is floor(bytes / (itemsize * slab samples)).""" + plan = build_chunk_plan(sized_df, time=dc.get_quantity("10 kB")) + (part,) = plan.params["size"]["partitions"] + # 11 distance samples of float64 per time sample + assert part["slab_samples"] == 11 + assert part["bytes_per_sample"] == 88 + assert part["n_samples"] == 10_000 // 88 + + def test_params_records_request(self, sized_df): + """Params explain what the size request resolved to.""" + plan = build_chunk_plan(sized_df, time=dc.get_quantity("10 kB")) + assert plan.params["size"]["requested_bytes"] == 10_000 + assert plan.params["size"]["partitions"][0]["dtype"] == "float64" + + def test_params_absent_without_size(self, sized_df): + """A plain chunk records no size diagnostics.""" + assert "size" not in build_chunk_plan(sized_df, time=5).params + + def test_mixed_dtypes_upcast(self, sized_df): + """A partition mixing dtypes is sized against the upcast dtype.""" + df = sized_df.copy() + df.loc[df.index[0], "_dtype"] = "float32" + plan = build_chunk_plan(df, time=dc.get_quantity("10 kB")) + assert plan.params["size"]["partitions"][0]["dtype"] == "float64" From 6d7d8b4c4c694ef6d8e0d6a35398a2de582e2a18 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Fri, 7 Aug 2026 13:52:34 +0200 Subject: [PATCH 2/3] Fix size-bound and dtype defects found in review Three defects in the size feature, each with a regression test that fails without its fix. The bound could be exceeded by up to 1.84x. An output holds the sum of its members' sample counts, which equals span/step + 1 only when the partition sits on one grid; members separated by less than one sample each contribute their own trailing sample, so near-contiguous files whose boundaries miss the grid pack more samples into a span than the grid allows. Divide the length by the partition's measured packing factor, which is exactly 1.0 when members tile the grid. This also supersedes the min-step rationale: packing is measured in units of the step used, so the two cancel and the length works out to n_samples / max_density either way. The comment and design note now say so instead of claiming the median step would overshoot. Element dtype is now resolved per output rather than per partition. An output drawing only from its float32 members really is float32, so claiming the partition-wide upcast both over-sized a chained chunk and made a chunked spool compare unequal to its own materialized twin -- for plain chunking too, not just sizes. Non-finite and array-valued quantities are rejected up front. A NaN magnitude is null, so `chunk(time=nan*MB)` silently merged the whole spool when the user asked for a size cap, and `inf` raised a bare ValueError from the sample-count division. Also: never write NaN into the derived catalog's dtype column (NaN is truthy, so `or ""` let the string "nan" through and poisoned every later np.dtype of that column); cover np.result_type with an int32 + float32 partition, where max-itemsize silently doubles the budget while float32 + float64 cannot tell the two rules apart; and replace a vacuous merge-mode test that passed with the feature absent and a tautological one that restated a literal from the module under test. --- dascore/io/index/planned.py | 13 +- dascore/units.py | 9 +- dascore/utils/chunk_plan.py | 111 +++++++++++-- docs/notes/spool_chunking.qmd | 7 +- tests/test_core/test_patch_chunk.py | 153 +++++++++++++++++- .../test_io/test_index/test_index_contract.py | 28 +++- tests/test_units.py | 18 ++- tests/test_utils/test_chunk.py | 14 ++ 8 files changed, 322 insertions(+), 31 deletions(-) diff --git a/dascore/io/index/planned.py b/dascore/io/index/planned.py index f2f458bef..df4988ca0 100644 --- a/dascore/io/index/planned.py +++ b/dascore/io/index/planned.py @@ -62,6 +62,13 @@ def _num(value) -> float | None: return float(value) +def _dtype_str(value) -> str: + """Convert a stored element dtype to its string, "" when unknown.""" + if value is None or pd.isnull(value): + return "" + return str(value) + + def _coord_record_from_row( row: Mapping, name: str, dims: tuple[str, ...] | None = None ) -> CoordRecord | None: @@ -269,8 +276,10 @@ def _output_records( dims=dims, shape="", # the plan carries the element dtype privately so a chained - # chunk can still size patches by their memory footprint - dtype=str(row.get("_dtype") or ""), + # chunk can still size patches by their memory footprint. + # NaN is truthy, so `or ""` alone would store the string + # "nan" and poison every later np.dtype() of this column. + dtype=_dtype_str(row.get("_dtype")), n_dims=len(dim_names), sample_count_total=None, time_min=_ns(row.get("time_min")), diff --git a/dascore/units.py b/dascore/units.py index cd6558629..c3cc8a0da 100644 --- a/dascore/units.py +++ b/dascore/units.py @@ -510,10 +510,11 @@ def get_byte_count(value: Quantity) -> float: Notes ----- - Do not use [`to_float`](`dascore.utils.time.to_float`) for this. It - falls back to `float(value)`, and pint converts a dimensionless - quantity to its base units, which for information is *bits*, so a - size would come back eight times too large. + This is the only correct way to get a byte count from a quantity. + In particular [`to_float`](`dascore.utils.time.to_float`) is not: + pint converts a dimensionless quantity to its base units, which for + information is *bits*, so anything routing a size through a plain + `float()` conversion is eight times too large. Examples -------- diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index 894beeb4c..36b7c9903 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -319,6 +319,22 @@ def _coerce_length_overlap(value, overlap, start_dtype): return value, overlap +def _validate_quantity(label: str, quant, name: str) -> None: + """Reject quantity chunk lengths that cannot describe a length.""" + if not isinstance(quant, Quantity): + return + magnitude = np.asarray(quant.magnitude) + if magnitude.ndim: + msg = ( + f"The {label} for {name!r} must be a single quantity, got an " + f"array of {magnitude.size}." + ) + raise ParameterError(msg) + if not np.isfinite(magnitude): + msg = f"The {label} for {name!r} must be finite, got {quant}." + raise ParameterError(msg) + + def _needs_partition_resolution(value, overlap) -> bool: """ True when a chunk length or overlap can only be resolved per partition. @@ -418,15 +434,63 @@ def _size_to_length(size_quant, sub, name, size_step): samples = int(requested // bytes_per_sample) clamped = samples < 1 samples = max(samples, 1) + # An output holds the sum of its members' sample counts, which only + # equals span/step + 1 when the partition sits on a single grid. + # Members separated by less than one sample pack more samples into + # the same span (each contributes its own trailing sample), so the + # length is divided by how much denser than the grid the partition + # actually is. Exactly 1.0 for gridded partitions. + packing = _packing_factor(sub, name, step_float) + span_samples = max(int(samples // packing), 1) diagnostics = { "dtype": str(dtype), "itemsize": int(dtype.itemsize), "slab_samples": int(slab), "bytes_per_sample": int(bytes_per_sample), "n_samples": samples, + "packing": float(packing), "clamped": clamped, } - return samples * size_step, diagnostics + return span_samples * size_step, diagnostics + + +def _packing_factor(sub: pd.DataFrame, name: str, step_float: float) -> float: + """ + How densely a partition's members pack samples, relative to its grid. + + 1.0 when members tile the grid exactly. Greater when consecutive + members are separated by less than one sample, which happens with + near-contiguous files whose boundaries do not land on the grid. + Overlapping members are excluded: `_remove_overlaps` deduplicates + them at member-build time, so they do not add samples. + """ + try: + start, stop, step = get_interval_columns(sub, name) + except Exception: # not a complete envelope; assume the grid + return 1.0 + starts = to_float(start.values) + order = np.argsort(starts, kind="stable") + # relative to the first start: absolute datetimes are ~1e9 seconds, + # where a float64 difference loses the precision a sub-sample gap + # lives in + starts = starts[order] - starts[order][0] + spans = to_float((stop - start).values)[order] + steps = np.abs(to_float(step.values))[order] + with np.errstate(invalid="ignore", divide="ignore"): + counts = np.round(spans / steps) + 1 + # distance to the next member's start; the last member owns its + # own span plus the one step its trailing sample occupies + deltas = np.diff(starts, append=starts[-1] + spans[-1] + steps[-1]) + # an overlap is deduplicated, so it cannot exceed grid density + deltas = np.maximum(deltas, spans) + density = counts / deltas + density = density[np.isfinite(density) & (density > 0)] + if not len(density): + return 1.0 + packing = float(density.max() * step_float) + # residual float noise must not cost a sample on an exactly gridded + # partition, whose packing is 1.0 by construction + return 1.0 if packing < 1 + 1e-9 else packing def _quantity_to_dim_value(quant, sub, name, start_dtype): @@ -562,16 +626,31 @@ def _police_columns(sub: pd.DataFrame, name, conflict) -> dict: col = f"_{coord}_units" if col in sub.columns: carried[col] = sub[col].iloc[0] - # The element dtype carries privately (see SPOOL_PRIVATE_RENAMES): a - # size-based chunk needs it, and a partition may legitimately mix - # dtypes, so the carried value is the one assembly will upcast to. - if "_dtype" in sub.columns: - combined = _combined_dtype(sub["_dtype"]) - if combined is not None: - carried["_dtype"] = str(combined) return carried +def _output_dtypes(sub: pd.DataFrame, members: pd.DataFrame) -> pd.Series | None: + """ + The element dtype each output assembles to, indexed by output_id. + + Carried privately (see SPOOL_PRIVATE_RENAMES) because a size-based + chunk needs it. It is resolved per *output* rather than per + partition: a partition may legitimately mix dtypes, but an output + drawing only from its float32 members really is float32, and + claiming the partition-wide upcast would both over-size a later + chunk and make the plan row disagree with the patch it assembles + (which spool equality compares). + """ + if "_dtype" not in sub.columns or members.empty: + return None + by_patch = sub.drop_duplicates("_patch_id").set_index("_patch_id")["_dtype"] + grouped = members.groupby("output_id")["_patch_id"] + dtypes = grouped.apply(lambda pids: _combined_dtype(by_patch.reindex(pids))) + # "" is the same not-known sentinel ingest writes; never leave NaN, + # which would reach the derived catalog as the string "nan". + return dtypes.map(lambda x: "" if x is None else str(x)) + + def _build_members(sub: pd.DataFrame, outputs: pd.DataFrame, name) -> pd.DataFrame: """ Bind one partition's outputs to trimmed source slices. @@ -671,6 +750,11 @@ def build_chunk_plan( validate_conflict(conflict) ((name, value),) = kwargs.items() value = None if value is Ellipsis else value + # Police quantities before merge_mode is decided: a NaN magnitude is + # null, so a nan-valued size would silently merge the whole spool + # when the user asked for a size *cap*. + for label, quant in (("chunk value", value), ("overlap", overlap)): + _validate_quantity(label, quant, name) merge_mode = pd.isnull(value) if merge_mode and (keep_partial or overlap): msg = ( @@ -771,10 +855,11 @@ def build_chunk_plan( part_step = get_middle_value(step.values) # D7: one step everywhere g_start, g_stop = start.min(), stop.max() if per_partition and not merge_mode: - # Size against the partition's *smallest* step, not its median: - # steps within sampling_group_tolerance share a partition, so a - # member sampled faster than the median would otherwise fit more - # samples into the length and overshoot the requested size. + # The bound itself is enforced by the packing factor, which is + # measured in units of this step and so cancels the choice + # (length works out to n_samples / max density either way). + # The smallest step is still the better unit: it makes the + # flooring granularity the finest the partition allows. size_step = step.abs().min() value_c, overlap_c, diag = _resolve_partition_length( value, overlap, sub, name, size_step, df[min_name].dtype @@ -812,6 +897,8 @@ def build_chunk_plan( # runtime error; it is not surfaced at all. fed = set(members["output_id"]) if not members.empty else set() outputs = outputs[outputs["output_id"].isin(fed)] + if (dtypes := _output_dtypes(sub_sorted, members)) is not None: + outputs["_dtype"] = outputs["output_id"].map(dtypes).fillna("") out_frames.append(outputs) member_frames.append(members) if size_diagnostics: diff --git a/docs/notes/spool_chunking.qmd b/docs/notes/spool_chunking.qmd index a53d2a5e8..1fb7cf779 100644 --- a/docs/notes/spool_chunking.qmd +++ b/docs/notes/spool_chunking.qmd @@ -83,7 +83,11 @@ bytes_per_sample = itemsize * (product of the other dimensions' sample counts) n_samples = floor(requested_bytes / bytes_per_sample) ``` -The count is floored, so an output's data array never exceeds the request, and it is computed against the partition's *smallest* step — steps within `sampling_group_tolerance` share a partition, so sizing against the median would let a faster-sampled member overshoot. Element dtype is deliberately **not** a partition key: making it one would change which patches merge. A partition may therefore mix dtypes, and the estimate uses `np.result_type` over them, matching the upcast assembly performs. A patch relation with no dtype (a hand-built dataframe, or an index predating the column) raises `ChunkError` rather than guessing an itemsize. +The count is floored, so an output's data array never exceeds the request. + +What actually enforces that bound is the **packing factor**, which the length is divided by. An output holds the sum of its members' sample counts, and that equals `span/step + 1` only when the partition sits on one grid: members separated by *less* than one sample each contribute their own trailing sample, so a near-contiguous set of files whose boundaries miss the grid packs more samples into a span than the grid would. The factor is the partition's maximum local sample density measured in units of its step — exactly 1.0 when members tile the grid, and reported in `params["size"]`. (Because it is measured in step units, it also cancels the choice of *which* step the length is expressed in; the smallest is used only to make the flooring granularity as fine as the partition allows.) Overlapping members are excluded from the measurement, since `_remove_overlaps` deduplicates them and they add no samples. + +Element dtype is resolved per *output* rather than per partition — dtype is deliberately **not** a partition key (making it one would change which patches merge), so a partition may mix dtypes, but an output drawing only from its float32 members really is float32; claiming the partition-wide `np.result_type` would over-size a later chunk and make the plan row disagree with the patch it assembles, which spool equality compares. An output that does span members of differing dtypes gets the upcast, matching what assembly produces. A patch relation with no dtype (a hand-built dataframe, or an index predating the column) raises `ChunkError` rather than guessing an itemsize. ```{python} from dascore.units import megabytes @@ -97,6 +101,7 @@ plan = dc.get_example_spool("random_das").chunk_plan(time=target) (part,) = plan.params["size"]["partitions"] assert part["bytes_per_sample"] == part["itemsize"] * part["slab_samples"] assert part["n_samples"] == int(1e6) // part["bytes_per_sample"] +assert part["packing"] == 1.0 # this spool tiles its grid exactly # The dtype carries through a chained chunk, so sizes still work. chained = dc.get_example_spool("random_das").chunk(distance=100).chunk(time=target) diff --git a/tests/test_core/test_patch_chunk.py b/tests/test_core/test_patch_chunk.py index 5d6cdcdb6..7414b7edf 100644 --- a/tests/test_core/test_patch_chunk.py +++ b/tests/test_core/test_patch_chunk.py @@ -1062,6 +1062,63 @@ def test_mixed_dtype_partition_uses_upcast(self, mixed_dtype_spool): out = mixed_dtype_spool.chunk(time=target) assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + def test_mixed_steps_in_one_partition_stay_bounded(self): + """ + Sizing must use the partition's smallest step, not its median. + + Steps within `sampling_group_tolerance` share a partition, so a + member sampled faster than the median fits more samples into the + same length and would overshoot the byte target. + """ + t0 = np.datetime64("2020-01-01T00:00:00") + patches, start = [], t0 + for step_ns in (4_000_000, 4_000_000, 3_850_000): # 3.75% apart + step = np.timedelta64(step_ns, "ns") + times = start + np.arange(500) * step + patches.append( + dc.Patch( + data=np.zeros((25, 500)), + dims=("distance", "time"), + coords={"distance": np.arange(25) * 1.0, "time": times}, + ) + ) + start = times[-1] + step + spool = dc.spool(patches).sort("time") + target = dc.get_quantity("100 kB") + # the steps must actually share one partition or this proves nothing + assert len(spool.chunk_plan(time=target).params["size"]["partitions"]) == 1 + out = spool.chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_int_float_partition_promotes_above_both(self): + """ + A mixed partition is sized by `np.result_type`, not max itemsize. + + int32 and float32 are both 4 bytes but promote to 8-byte + float64, so a max-itemsize estimate would under-count and + produce patches at twice the requested size. + """ + t0 = np.datetime64("2020-01-01T00:00:00") + + def make(dtype, start): + return dc.Patch( + data=np.zeros((50, 250), dtype=dtype), + dims=("distance", "time"), + coords={ + "distance": np.arange(50) * 1.0, + "time": start + np.arange(250) * np.timedelta64(4, "ms"), + }, + ) + + second = t0 + np.timedelta64(1000, "ms") + spool = dc.spool([make("int32", t0), make("float32", second)]).sort("time") + target = dc.get_quantity("100 kB") + (part,) = spool.chunk_plan(time=target).params["size"]["partitions"] + assert part["dtype"] == "float64" + assert part["itemsize"] == 8 + out = spool.chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + def test_slab_larger_than_target_warns(self, random_spool): """One sample is the floor; the request cannot be honored below it.""" with pytest.warns(UserWarning, match="larger than the requested size"): @@ -1078,8 +1135,10 @@ def test_keep_partial_stays_bounded(self, random_spool): def test_overlap_in_coord_units(self, random_spool): """Overlap may use the coordinate's units while the length is a size.""" target = dc.get_quantity("2 MB") + plain = random_spool.chunk(time=target) out = random_spool.chunk(time=target, overlap=1 * dc.units.s) assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + assert len(out) > len(plain) # the overlap is honored, not dropped def test_overlap_as_size(self, random_spool): """Overlap may also be expressed as a size.""" @@ -1129,6 +1188,96 @@ def test_three_dimensional_patch(self): (part,) = plan.params["size"]["partitions"] assert part["slab_samples"] == 15 - def test_merge_mode_unaffected(self, random_spool): - """A merge takes no length, so sizes never apply.""" + def test_sub_sample_gaps_stay_bounded(self): + """ + Members packed closer than one sample must not overshoot. + + Near-contiguous files whose boundaries miss the grid fit more + samples into a span than span/step + 1, because each member + contributes its own trailing sample. + """ + t0 = np.datetime64("2020-01-01T00:00:00") + second = np.timedelta64(1_000_000_000, "ns") + patches, offset = [], 0 + for _ in range(30): + start = t0 + np.timedelta64(offset, "ns") + patches.append( + dc.Patch( + data=np.zeros((50, 20)), + dims=("distance", "time"), + coords={ + "distance": np.arange(50) * 1.0, + "time": start + np.arange(20) * second, + }, + ) + ) + offset += 19 * 1_000_000_000 + 10_000_000 # 10 ms short of the grid + target = dc.get_quantity("100 kB") + out = dc.spool(patches).chunk(time=target) + assert max(x.data.nbytes for x in out) <= target.to("byte").magnitude + + def test_gridded_partition_has_unit_packing(self, random_spool): + """A partition that tiles its grid is sized without correction.""" + plan = random_spool.chunk_plan(time=1 * dc.units.MB) + (part,) = plan.params["size"]["partitions"] + assert part["packing"] == 1.0 + + def test_output_dtype_matches_assembled_patch(self): + """ + A plan row must claim the dtype its patch actually assembles to. + + Claiming the partition-wide upcast would both over-size a later + size chunk and make a chunked spool compare unequal to its own + materialized twin. + """ + t0 = np.datetime64("2020-01-01T00:00:00") + + def make(dtype, start): + rng = np.random.default_rng(1) + return dc.Patch( + data=rng.random((40, 400)).astype(dtype), + dims=("distance", "time"), + coords={ + "distance": np.arange(40) * 1.0, + "time": start + np.arange(400) * np.timedelta64(4, "ms"), + }, + ) + + spool = dc.spool( + [make("float32", t0), make("float64", t0 + np.timedelta64(1600, "ms"))] + ).sort("time") + out = spool.chunk(time=1.0) + claimed = out.get_contents()["_dtype"].tolist() + assert claimed == [str(x.data.dtype) for x in out] + assert out == dc.spool(list(out)) + + def test_merge_mode_records_no_size(self, random_spool): + """A merge takes no length, so no size is ever resolved.""" + plan = random_spool.chunk_plan(time=None) + assert plan.merge_mode + assert "size" not in plan.params assert len(random_spool.chunk(time=None)) == 1 + + def test_merge_mode_rejects_size_overlap(self, random_spool): + """A size overlap is still an overlap, which merging forbids.""" + with pytest.raises(ParameterError, match="keep_partial and overlap"): + random_spool.chunk(time=None, overlap=1 * dc.units.MB) + + @pytest.mark.parametrize( + "value", (float("inf") * dc.units.MB, float("nan") * dc.units.MB) + ) + def test_non_finite_size_raises(self, random_spool, value): + """ + A non-finite size is a bad request, not a merge. + + NaN is null, so without an explicit guard a nan-valued size + would silently merge the whole spool into one patch when the + user asked for a size cap. + """ + with pytest.raises(ParameterError, match="finite"): + random_spool.chunk(time=value) + + 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) diff --git a/tests/test_io/test_index/test_index_contract.py b/tests/test_io/test_index/test_index_contract.py index da7fd2609..e3cdc1184 100644 --- a/tests/test_io/test_index/test_index_contract.py +++ b/tests/test_io/test_index/test_index_contract.py @@ -210,11 +210,29 @@ def test_dtype_is_private_in_flat_relation(self): assert "dtype" not in df.columns assert set(df["_dtype"]) == {str(spool[0].data.dtype)} - def test_dtype_attr_is_reserved(self): - """An attr named `dtype` cannot shadow the structural column.""" - from dascore.io.index.schema import RESERVED_ATTR_COLUMNS - - assert "dtype" in RESERVED_ATTR_COLUMNS + def test_dtype_attr_does_not_shadow_column(self, tmp_path): + """A patch attr named `dtype` is skipped, not written to the column.""" + summary = make_summaries()[0] + shadowed = PatchSummary( + attrs=dict(summary.attrs.model_dump(), dtype="not a dtype"), + coords={k: v.model_dump() for k, v in summary.coords.items()}, + dims=summary.dims, + shape=summary.shape, + dtype=summary.dtype, + source_path=summary.source_path, + source_format=summary.source_format, + source_version=summary.source_version, + ) + path = tmp_path / "shadow.sqlite3" + back = get_backend(path) + try: + with pytest.warns(UserWarning, match="dtype"): + back.write_sources(summaries_to_records([shadowed])) + df = back.query() + # the structural column keeps the element dtype, not the attr + assert df["dtype"].iloc[0] == summary.dtype + finally: + back.close() class TestAttrPredicates: diff --git a/tests/test_units.py b/tests/test_units.py index 73ba23527..4cc6be4b1 100644 --- a/tests/test_units.py +++ b/tests/test_units.py @@ -586,11 +586,19 @@ def test_byte_count_is_not_to_float(self): """ Guard the bits trap. - `to_float` falls back to `float(quantity)`, and pint converts a - dimensionless quantity to base units, where information is bits. - Any "simplification" of get_byte_count to to_float makes every - size eight times too large. + pint converts a dimensionless quantity to base units, and + information's base unit is the bit, so routing a size through + `float()` makes it eight times too large. Assert the byte count + directly rather than the wrong value, so this holds however + `to_float` treats quantities. """ quant = get_quantity("25 MB") assert get_byte_count(quant) == 25_000_000 - assert to_float(quant) == 8 * 25_000_000 + # the trap: what a bare float() conversion would have produced + assert quant.to_base_units().magnitude == 8 * 25_000_000 + # to_float is not a byte converter; it either raises or answers + # in seconds, but must never be mistaken for get_byte_count + try: + assert to_float(quant) != get_byte_count(quant) + except UnitError: + pass diff --git a/tests/test_utils/test_chunk.py b/tests/test_utils/test_chunk.py index 6cb6da940..5105b3169 100644 --- a/tests/test_utils/test_chunk.py +++ b/tests/test_utils/test_chunk.py @@ -447,6 +447,20 @@ def test_params_absent_without_size(self, sized_df): """A plain chunk records no size diagnostics.""" assert "size" not in build_chunk_plan(sized_df, time=5).params + def test_unknown_dtype_partition_is_empty_not_nan(self, sized_df): + """ + A partition with no dtype carries "", never NaN. + + NaN is truthy, so it would reach the derived catalog as the + string "nan" and poison every later np.dtype() of the column. + """ + known = sized_df.assign(station="a") + unknown = sized_df.assign(station="b", _dtype="") + both = pd.concat([known, unknown], ignore_index=True) + outputs = build_chunk_plan(both, time=5).outputs + assert not outputs["_dtype"].isna().any() + assert set(outputs["_dtype"]) == {"float64", ""} + def test_mixed_dtypes_upcast(self, sized_df): """A partition mixing dtypes is sized against the upcast dtype.""" df = sized_df.copy() From 9c88a80f1dc7c4cecbfdeab23cf5bfb96b7e3351 Mon Sep 17 00:00:00 2001 From: Derrick Chambers Date: Fri, 7 Aug 2026 14:21:41 +0200 Subject: [PATCH 3/3] Fix a Windows path assumption and cover the remaining branches test_dtype_round_trips keyed a dict on source_path, which the SQLite round-trip returns with backslashes on Windows while the summary literal uses forward slashes. It failed every Windows job on both the full and min-deps matrices; normalize the separator before comparing. Nothing about the feature is platform-dependent. Codecov also flagged uncovered lines in the new code, all of them real branches rather than dead ones: a frame with no `dims` column falling back to its complete envelopes, an unknown step on the chunked dimension, an unparseable dtype string, a missing envelope on another dimension, and a numeric coordinate rejecting a dimensionally wrong quantity. Each now has a test. Two defensive branches really were unreachable and are gone rather than pragma'd: get_interval_columns cannot fail in _packing_factor because the caller already read those columns for the same partition, and the density array cannot be empty once the step is validated (now a plain assert, which documents the invariant and executes every call). --- dascore/io/index/planned.py | 4 +-- dascore/utils/chunk_plan.py | 11 +++---- .../test_io/test_index/test_index_contract.py | 9 ++++-- tests/test_utils/test_chunk.py | 31 +++++++++++++++++++ 4 files changed, 44 insertions(+), 11 deletions(-) diff --git a/dascore/io/index/planned.py b/dascore/io/index/planned.py index df4988ca0..61dcc1074 100644 --- a/dascore/io/index/planned.py +++ b/dascore/io/index/planned.py @@ -64,9 +64,7 @@ def _num(value) -> float | None: def _dtype_str(value) -> str: """Convert a stored element dtype to its string, "" when unknown.""" - if value is None or pd.isnull(value): - return "" - return str(value) + return "" if value is None or pd.isnull(value) else str(value) def _coord_record_from_row( diff --git a/dascore/utils/chunk_plan.py b/dascore/utils/chunk_plan.py index 36b7c9903..94f594567 100644 --- a/dascore/utils/chunk_plan.py +++ b/dascore/utils/chunk_plan.py @@ -464,10 +464,8 @@ def _packing_factor(sub: pd.DataFrame, name: str, step_float: float) -> float: Overlapping members are excluded: `_remove_overlaps` deduplicates them at member-build time, so they do not add samples. """ - try: - start, stop, step = get_interval_columns(sub, name) - except Exception: # not a complete envelope; assume the grid - return 1.0 + # the caller already read these columns for this partition + start, stop, step = get_interval_columns(sub, name) starts = to_float(start.values) order = np.argsort(starts, kind="stable") # relative to the first start: absolute datetimes are ~1e9 seconds, @@ -485,8 +483,9 @@ def _packing_factor(sub: pd.DataFrame, name: str, step_float: float) -> float: deltas = np.maximum(deltas, spans) density = counts / deltas density = density[np.isfinite(density) & (density > 0)] - if not len(density): - return 1.0 + # a partition whose steps are all unusable is rejected before this, + # so at least one member always yields a density + assert len(density) packing = float(density.max() * step_float) # residual float noise must not cost a sample on an exactly gridded # partition, whose packing is 1.0 by construction diff --git a/tests/test_io/test_index/test_index_contract.py b/tests/test_io/test_index/test_index_contract.py index e3cdc1184..b8894def3 100644 --- a/tests/test_io/test_index/test_index_contract.py +++ b/tests/test_io/test_index/test_index_contract.py @@ -197,8 +197,13 @@ class TestElementDtype: def test_dtype_round_trips(self, backend): """Each patch keeps the dtype its summary reported.""" df = backend.query() - expected = {str(x.source_path): x.dtype for x in make_summaries()} - got = {str(k): v for k, v in zip(df["path"], df["dtype"], strict=True)} + + def _key(path): + """Compare path-independently; Windows round-trips backslashes.""" + return str(path).replace("\\", "/") + + expected = {_key(x.source_path): x.dtype for x in make_summaries()} + got = {_key(k): v for k, v in zip(df["path"], df["dtype"], strict=True)} assert got == expected def test_dtype_is_private_in_flat_relation(self): diff --git a/tests/test_utils/test_chunk.py b/tests/test_utils/test_chunk.py index 5105b3169..1af7cd285 100644 --- a/tests/test_utils/test_chunk.py +++ b/tests/test_utils/test_chunk.py @@ -447,6 +447,37 @@ def test_params_absent_without_size(self, sized_df): """A plain chunk records no size diagnostics.""" assert "size" not in build_chunk_plan(sized_df, time=5).params + def test_frame_without_dims_column(self, contiguous_df): + """A frame with no `dims` falls back to its complete envelopes.""" + df = contiguous_df.assign(_dtype="float64") # deliberately no dims + plan = build_chunk_plan(df, time=dc.get_quantity("10 kB")) + (part,) = plan.params["size"]["partitions"] + assert part["slab_samples"] == 11 # the distance envelope + + def test_unknown_chunk_dim_step_raises(self, sized_df): + """The chunked dimension's own step must be known to size it.""" + df = sized_df.assign(time_step=np.timedelta64("NaT")) + with pytest.raises(ChunkError, match="sampling interval is unknown"): + build_chunk_plan(df, time=dc.get_quantity("10 kB")) + + def test_numeric_dim_incompatible_units_raise(self, sized_df): + """A numeric coordinate rejects a dimensionally wrong quantity.""" + df = sized_df.assign(_distance_units="m") + with pytest.raises(UnitError, match="incompatible with"): + build_chunk_plan(df, distance=5 * dc.units.s) + + def test_unparsable_dtype_raises(self, sized_df): + """A dtype string numpy cannot parse is unusable, not absent.""" + df = sized_df.assign(_dtype="not-a-real-dtype") + with pytest.raises(ChunkError, match="dtype"): + build_chunk_plan(df, time=1 * dc.units.MB) + + def test_missing_other_dim_envelope_raises(self, sized_df): + """A dim without a full envelope has no derivable sample count.""" + df = sized_df.drop(columns=["distance_max"]) + with pytest.raises(ChunkError, match="cannot be determined"): + build_chunk_plan(df, time=1 * dc.units.MB) + def test_unknown_dtype_partition_is_empty_not_nan(self, sized_df): """ A partition with no dtype carries "", never NaN.