Spool Chunking

Spool.chunk runs in two stages: a planner decides everything from metadata alone, and assembly loads, trims, and combines patch data only when a patch is requested. The code cells below execute against the real machinery, so this note fails the doc build if it drifts from the implementation.

Plans

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.

import dascore as dc

spool = dc.get_example_spool("random_das")
plan = spool.chunk_plan(time=3)
assert len(plan.outputs) == len(spool.chunk(time=3))
assert {"output_id", "_patch_row", "_modified"}.issubset(plan.members.columns)
# Resolved parameters are recorded on the plan, never left in config.
assert isinstance(plan.params["group"], tuple)
assert plan.params["sampling_group_tolerance"] == dc.get_config().sampling_group_tolerance

# Chaining partitions both dimensions: the second chunk plans over the
# first one's outputs, not the original members.
chained = dc.get_example_spool("random_das").chunk(time=None).chunk(distance=100)
assert all(p.shape[p.get_axis("distance")] <= 100 for p in chained)

Plans are deterministic: the same spool produces the same plan regardless of metadata row order, and members with _modified=False load whole (no per-patch selection cost).

Partitioning

Patches may only combine when they agree on all of:

Kind — the config option patch_kind_attrs by default (conventional categorical identity: acquisition key, data category, and tag; see Patch Compatibility), overridden per call with group=. Differing values are never an error; the patches simply land in separate outputs. A missing value is a value: patches which never recorded an attribute partition together, and apart from those which did. Explicitly passed names must exist somewhere in the spool; config names are best-effort.
Structure — the dimensions tuple, the coordinate identity of every non-chunked dimension, and the chunked dimension’s units (a metre patch can never plan into one output with a seconds patch, or a unitful with a unitless one; compatible spellings such as metres and feet are normalized to one unit per dimensionality before partitioning, so they plan together, and assembly converts each member to the first member’s units).
Sampling — step magnitudes within the relative tolerance config.sampling_group_tolerance (default 5%) of the group’s smallest member, with matching orientation (ascending never merges with descending; contiguous descending patches merge with each other).
Continuity — patches within tolerance samples of each other, evaluated within each group so unrelated patches can never bridge a gap. Each run of a patch whose coordinate is segmented into runs of one step (up to 256) plans as a trimmed member of its own, so a hole inside a patch is judged like a gap between patches; consecutive runs landing in one output are read back as one member. tolerance may also be a quantity (1 * s, 2 * m) or a timedelta, which bounds the excess over one step in the coordinate’s own units rather than counting steps: a spacing is a gap when it exceeds step + tolerance, so adjacent patches stay contiguous under any tolerance and a zero excess admits exactly one step. Since it needs no sampling interval, an absolute tolerance also measures gaps between patches whose step is unknown (there the step counts as nothing) — patches a sample-count tolerance has nothing to scale, and so never splits. Both spellings are one object, GapTolerance, and its one predicate is what coordinates’ get_discontinuities and the waterfall’s gap_factor use, so the three agree on every spacing given the same object; a bare number to get_discontinuities is an absolute excess, not a count.

Spool.get_gaps and Spool.get_coverage read this same partition: a get_coverage row is one cell of criteria 1-3, and a get_gaps row is a boundary criterion 4 declines to close. So on a spool of unplanned patches the counts tie — merging yields one patch per coverage row plus one per gap reported. The reports read the rows a spool presents rather than a plan’s members, so a spool that has already been chunked reports on the patches it now holds, and re-chunking it need not reproduce those counts.

import numpy as np

t0 = np.datetime64("2020-01-01", "ns")
p1 = dc.get_example_patch(time_min=t0, acquisition_key="XX1.R2D1..RAW")
time = p1.get_coord("time")
p2 = dc.get_example_patch(time_min=time.max() + time.step, acquisition_key="XX1.R2D1..RAW")
source = "XX2.R2D1..RAW"
p3, p4 = p1.update_attrs(acquisition_key=source), p2.update_attrs(acquisition_key=source)

# Two data sources, each with two contiguous patches: two outputs, no error.
spool = dc.spool([p1, p2, p3, p4])
merged = spool.chunk(time=None)
assert len(merged) == 2

# The reports partition the same way, so the counts tie.
assert len(merged) == len(spool.get_gaps()) + len(spool.get_coverage())

Remaining (non-group, non-dimensional) attributes must hold equal values within a partition, policed by conflict: "raise" (default), "drop", or "keep_first". A missing value (null or "") is a value like any other — an attribute one member recorded and another did not is a conflict, since the output can only state one thing. "keep_first" states what the first member did, stated or not; "drop" leaves the attribute out. History and the ids are never policed — the output carries the first member’s — but merging members whose histories differ (processed beside unprocessed data) warns.

import pytest
from dascore.exceptions import CoordMergeError

typed = p1.update_attrs(data_type="velocity")
with pytest.raises(CoordMergeError, match="data_type"):
    dc.spool([typed, p2]).chunk(time=None)

merged = dc.spool([typed, p2]).chunk(time=None, conflict="keep_first")
assert len(merged) == 1
assert merged.get_contents()["data_type"].iloc[0] == "velocity"
assert merged[0].attrs.data_type == "velocity"

Chunking is defined on dimensions: a patch that lacks the chunked dimension — including one that carries the name only as a non-dimensional coordinate, which cannot be trimmed or merged along it — raises by default, and missing_dim="drop" excludes it instead. Losing patches silently would be data loss, so it requires the explicit opt-in.

import pytest
from dascore.exceptions import ChunkError

no_time = [dc.get_example_patch().mean("time") for _ in range(2)]
with pytest.raises(ChunkError, match="missing_dim"):
    dc.spool(no_time).chunk(time=None)
assert len(dc.spool(no_time).chunk(time=None, missing_dim="drop")) == 0

# A name carried only as a non-dimensional coordinate counts as missing.
base = dc.get_example_patch()
aux = base.update_coords(
    sensor=("distance", np.arange(base.shape[base.get_axis("distance")], dtype=float))
)
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 a compatible unit (time=10 * s, distance=100 * m, converted to the partition’s stated coordinate units — a spool recording no units for the coordinate raises, since there is nothing to convert to) or a quantity of information (time=25 * megabytes), which sizes each output by its data array. A bare length means the coordinate’s own units, like every other bare number. 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.

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.

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"]
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)
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 fused with bounded error — no coordinate value moves more than tolerance * step, or more than the tolerance itself when it states the limit in the coordinate’s own units — and never at the cost of the sampling step. Fusing therefore absorbs the sub-sample jitter of labels rounded on their way to a file, but a gap the tolerance was wide enough to merge across stays a run boundary: its samples are missing, which is not the same as a slower sampling rate. fill_value is what writes those samples, and its output is one evenly sampled run.

# Contiguous members: exact fuse to one evenly sampled run.
patch = dc.spool([p1, p2]).chunk(time=None)[0]
assert patch.get_coord("time").evenly_sampled

# A gap forced together by a loose tolerance warns, and the output keeps
# the exact coordinate, holes and all: those samples are absent either way.
gap_start = time.max() + 3 * time.step
p_gap = dc.get_example_patch(time_min=gap_start, acquisition_key="XX1.R2D1..RAW")
with pytest.warns(UserWarning, match="force merging"):
    forced = dc.spool([p1, p_gap]).chunk(time=None, tolerance=5)[0]
coord = forced.get_coord("time")
assert coord.runs_count == 2
assert len(coord.get_discontinuities("gaps")) == 1

# snap_coords only ever absorbs sub-sample jitter, so here it changes
# nothing: no label moves, whether it runs or not.
with pytest.warns(UserWarning, match="force merging"):
    exact = dc.spool([p1, p_gap]).chunk(
        time=None, tolerance=5, snap_coords=False
    )[0]
assert np.array_equal(exact.get_coord("time").values, coord.values)

# An absolute tolerance -- a quantity, or a timedelta -- spans the same
# gap, stated in the coordinate's units rather than in samples.
with pytest.warns(UserWarning, match="force merging"):
    absolute = dc.spool([p1, p_gap]).chunk(time=None, tolerance=5 * time.step)[0]
assert absolute.get_coord("time") == coord

# fill_value writes the missing samples, so the merge comes back evenly
# sampled rather than segmented.
filled = dc.spool([p1, p_gap]).chunk(time=None, tolerance=5, fill_value=np.nan)[0]
assert filled.get_coord("time").evenly_sampled
assert np.isnan(filled.data).any()

Coordinates with holes are an in-memory representation only: a written patch must be contiguous, so saving raises unless split=True (or patch.split_gaps() is used first) to write each contiguous run as its own patch.

Union spools

spool + spool produces a lazy spool over the union of both spools’ metadata; chunking works across the seam, so contiguous patches from different spools (even file-backed and in-memory mixed) merge into one.

combined = dc.spool([p1]) + dc.spool([p2])
assert len(combined.chunk(time=None)) == 1

# Spools have set semantics by patch instance: the same patch (or a copy
# of it) appears once; operations mint distinct instances.
assert len(dc.spool([p1, p1])) == 1
assert len(dc.spool([p1, p1.new()])) == 2
assert len(dc.spool([p1]) + dc.spool([p1])) == 1

Explicit output windows

Pass a rectangular (n, 2) array as the chunked dimension to request bounded absolute coordinate windows. Each row is independent and inclusive. Input order is preserved, including overlapping and duplicate rows; compatible source groups within a row follow deterministic group order. chunk_plan accepts the same form and retains original request-row numbers in private _request_row plan metadata. One selected range may yield several source pieces while the same chunk window assembles them into one patch.

An explicit request is complete when its inward-aligned first and last sampled positions are supplied by one compatible output under the existing continuity and tolerance rules. A tolerated internal gap can remain segmented; fill_value may fill a permitted hole, even when a window is entirely inside it. Filling does not extend an acquisition beyond its overall envelope. Unknown or uneven sample steps use the source coordinates where available. overlap cannot be combined with explicit windows. Bare numeric bounds use the planner’s normalized coordinate units, while quantity bounds convert as absolute points. An empty (0, 2) array returns an empty plan or spool.

By default, an unmet request raises ChunkError. Set on_incomplete="warn" to report and skip unmet request/group outputs, or "ignore" to skip them silently. With keep_partial=True, nonempty available pieces within a request are retained first; the policy then handles requests still wholly absent. These rules apply to each applicable compatibility group whose overall envelope intersects a window. The conflict setting remains the separate policy for attributes of members that actually contribute to an accepted output. Planning uses coordinate metadata and keeps measurement-array loading deferred; source changes and unrelated read failures remain runtime errors.

import numpy as np
import pytest
from dascore.exceptions import ChunkError

patch = dc.get_example_patch().select(distance=(0, 9))
source = dc.spool(patch)
windows = np.array([[2, 4], [8, 10]])
with pytest.raises(ChunkError):
    source.chunk(distance=windows)
partial = source.chunk(distance=windows, keep_partial=True)
assert [(p.get_coord("distance").min(), p.get_coord("distance").max()) for p in partial] == [(2, 4), (8, 9)]
accepted = source.chunk(distance=windows, on_incomplete="ignore")
assert len(accepted) == 1
assert len(source.chunk_plan(distance=windows, on_incomplete="ignore").outputs) == 1