partition

blobmap.partition ¶

The decisions: arrays in, Manifest out.

Pure functions with no I/O. Everything that actually matters lives here, and it is testable with hand-written Array objects in milliseconds.

The cut proceeds top down:

Coordinates and arrays under t_hot_bytes are pinned hot.
If the whole scope fits in t_max_bytes, it becomes one blob. This is the common case, and the best outcome for tape: small stores are read whole, so one mount beats one per variable.
Arrays over t_max_bytes are bucketed, which is the only way to cut inside an array.
What is left is coalesced with its neighbours until each group clears t_min_bytes, so a wide tree of small variables does not cost a tape mount per gigabyte.

Example

from blobmap.model import Array, GiB, Policy arrays = [Array("time", (400,), (400,), 8, is_coordinate=True), ... Array("tas", (400, 4, 4), (10, 4, 4), 4, ... stored_bytes=400 * GiB)] manifest = partition("cordex/a.zarr", arrays) sorted(b.id for b in manifest.blobs) ['b_tas'] "time/**" in manifest.hot_always True

Diff `dataclass` ¶

Diff(
    added: tuple[Blob, ...] = (),
    removed: tuple[Blob, ...] = (),
    modified: tuple[tuple[Blob, Blob], ...] = (),
)

What a repartition changed.

Attributes:

Name	Type	Description
`added`	`tuple[Blob, ...]`	Blobs that did not exist before. Always safe.
`removed`	`tuple[Blob, ...]`	Blobs that disappeared. Any tape copy held against these ids is now orphaned.
`modified`	`tuple[tuple[Blob, Blob], ...]`	Pairs of (old, new) for blobs whose definition changed. The id still resolves but now covers different objects, so the tape copy no longer matches.

Example

from blobmap.model import Blob, Manifest old = Manifest("s", (Blob("a", ("a",)),), ()) new = Manifest("s", (Blob("a", ("a",)), Blob("b", ("b",))), ()) diff(old, new).is_additive True

is_additive `property` ¶

is_additive: bool

Whether no existing blob definition changed or disappeared.

Anything else invalidates ids that blobtier holds tape addresses for.

Returns:

Type	Description
`bool`	True when only additions were made.

is_empty `property` ¶

is_empty: bool

Whether nothing changed at all.

Returns:

Type	Description
`bool`	True when there is nothing to write.

describe ¶

describe() -> str

Render for a log line or --dry-run output.

Returns:

Type	Description
`str`	One line per change, prefixed `+`, `-` or `~`, or `(no change)`.

Source code in src/blobmap/partition.py

def describe(self) -> str:
    """Render for a log line or `--dry-run` output.

    Returns:
        One line per change, prefixed `+`, `-` or `~`, or `(no change)`.
    """
    lines = [f"+ {b.id} {list(b.prefixes)}" for b in self.added]
    lines += [f"- {b.id} {list(b.prefixes)}" for b in self.removed]
    lines += [f"~ {o.id}: {o.to_json()} -> {n.to_json()}"
              for o, n in self.modified]
    return "\n".join(lines) or "(no change)"

bucket_width ¶

bucket_width(a: Array, policy: Policy) -> int

Choose how many objects along dimension 0 go into one blob.

Two bounds apply:

ideal hits t_max_bytes at the compression observed today
hard keeps the blob under width_clamp * t_max_bytes even if compression degrades to nothing, computed from uncompressed object size, which cannot change without rewriting the array

The result is rounded down to a power of two so a ratio drifting from 2.0 to 2.3 does not move the width and renumber every blob. Rounding is skipped below pow2_floor, where it would throw away too much of the target. That is the common case for sharded arrays, which have far fewer and larger objects.

Parameters:

Name	Type	Description	Default
`a`	`Array`	The array to size. Its `object_shape` is the unit, so a sharded array is measured in shards.	required
`policy`	`Policy`	Thresholds to apply.	required

Returns:

Type	Description
`int`	Objects per blob, at least 1.

Note

Logs a warning when a single object already exceeds t_max_bytes, since there is then nothing left to cut.

Example

from blobmap.model import Array, GiB, Policy plain = Array("tas", (1_314_000, 412, 424), (128, 412, 424), 4, ... stored_bytes=459 * GiB) bucket_width(plain, Policy()) 2048

The same bytes in a 10x sharded array is a tenth of the objects, so the width falls with it. Reading chunks instead of shards here would give both arrays the same width, and blobs 10x the target:

sharded = Array("tas", (1_314_000, 412, 424), (128, 412, 424), 4, ... shards=(1280, 412, 424), stored_bytes=459 * GiB) bucket_width(sharded, Policy()) 128

Source code in src/blobmap/partition.py

def bucket_width(a: Array, policy: Policy) -> int:
    """Choose how many objects along dimension 0 go into one blob.

    Two bounds apply:

    * *ideal* hits `t_max_bytes` at the compression observed today
    * *hard* keeps the blob under `width_clamp * t_max_bytes` even if
      compression degrades to nothing, computed from uncompressed object
      size, which cannot change without rewriting the array

    The result is rounded down to a power of two so a ratio drifting from 2.0
    to 2.3 does not move the width and renumber every blob. Rounding is
    skipped below `pow2_floor`, where it would throw away too much of the
    target. That is the common case for sharded arrays, which have far fewer
    and larger objects.

    Args:
        a: The array to size. Its
            [`object_shape`][blobmap.model.Array.object_shape] is the unit,
            so a sharded array is measured in shards.
        policy: Thresholds to apply.

    Returns:
        Objects per blob, at least 1.

    Note:
        Logs a warning when a single object already exceeds `t_max_bytes`,
        since there is then nothing left to cut.

    Example:
        >>> from blobmap.model import Array, GiB, Policy
        >>> plain = Array("tas", (1_314_000, 412, 424), (128, 412, 424), 4,
        ...               stored_bytes=459 * GiB)
        >>> bucket_width(plain, Policy())
        2048

        The same bytes in a 10x sharded array is a tenth of the objects, so
        the width falls with it. Reading `chunks` instead of `shards` here
        would give both arrays the same width, and blobs 10x the target:

        >>> sharded = Array("tas", (1_314_000, 412, 424), (128, 412, 424), 4,
        ...                 shards=(1280, 412, 424), stored_bytes=459 * GiB)
        >>> bucket_width(sharded, Policy())
        128
    """
    ideal = max(1, policy.t_max_bytes // a.avg_object_bytes)
    hard = max(1, (policy.width_clamp * policy.t_max_bytes)
               // a.uncompressed_object_bytes)
    width = max(1, min(ideal, hard))
    if width >= policy.pow2_floor:
        width = 1 << (width.bit_length() - 1)
    if width == 1 and a.avg_object_bytes > policy.t_max_bytes:
        log.warning("%s: a single object is %.1f GiB, above t_max -- cannot "
                    "cut finer than one object", a.path,
                    a.avg_object_bytes / GiB)
    return width

partition ¶

partition(
    scope: str,
    arrays: list[Array],
    *,
    policy: Policy | None = None,
    previous: Manifest | None = None
) -> Manifest

Compute the blob definitions for one scope.

When previous is given its blobs are pinned: carried over byte identical, with new cuts only in regions no existing blob claims. That is what keeps blob ids, and the tape addresses blobtier holds against them, valid across a repartition. Growth along a bucketed dimension needs no repartition at all, since the id is arithmetic.

Because pinning is unconditional, a policy change alone has no effect on an existing scope. Cuts are frozen once made. Pass previous=None to recompute from scratch, which is the only way to move a blob and the only way to orphan a tape copy.

hot_always is always recomputed, since it follows from the store's structure: hot data is never archived, so there is no state to invalidate, and an array that grew past t_hot_bytes should stop being held back. pinned is carried over untouched, because it follows from intent rather than structure and can only be changed deliberately.

Parameters:

Name	Type	Description	Default
`scope`	`str`	Prefix these definitions apply to, relative to the data store.	required
`arrays`	`list[Array]`	Every array under the scope, from `read_arrays` or written by hand.	required
`policy`	`Policy \| None`	Thresholds. Defaults to `previous.policy` when repartitioning, else to `Policy` defaults.	`None`
`previous`	`Manifest \| None`	The manifest currently in effect, whose blobs are pinned.	`None`

Returns:

Type	Description
`Manifest`	A validated `Manifest`. The epoch is carried over unchanged; callers
`Manifest`	that intend to write should use
`Manifest`	`bumped`.

Raises:

Type	Description
`ValueError`	If the result fails validation, which would indicate a bug in the cut rather than bad input.

Example

from blobmap.model import Array, GiB first = partition("s", [Array("tas", (400, 4, 4), (10, 4, 4), 4, ... stored_bytes=400 * GiB)]) grown = partition("s", [Array("tas", (800, 4, 4), (10, 4, 4), 4, ... stored_bytes=800 * GiB)], ... previous=first) diff(first, grown).is_empty # an append changes nothing True

Source code in src/blobmap/partition.py

def partition(
    scope: str,
    arrays: list[Array],
    *,
    policy: Policy | None = None,
    previous: Manifest | None = None,
) -> Manifest:
    """Compute the blob definitions for one scope.

    When `previous` is given its blobs are *pinned*: carried over byte
    identical, with new cuts only in regions no existing blob claims. That is
    what keeps blob ids, and the tape addresses blobtier holds against them,
    valid across a repartition. Growth along a bucketed dimension needs no
    repartition at all, since the id is arithmetic.

    Because pinning is unconditional, a policy change alone has no effect on
    an existing scope. Cuts are frozen once made. Pass `previous=None` to
    recompute from scratch, which is the only way to move a blob and the only
    way to orphan a tape copy.

    `hot_always` is always recomputed, since it follows from the store's
    structure: hot data is never archived, so there is no state to invalidate,
    and an array that grew past `t_hot_bytes` should stop being held back.
    `pinned` is carried over untouched, because it follows from intent rather
    than structure and can only be changed deliberately.

    Args:
        scope: Prefix these definitions apply to, relative to the data store.
        arrays: Every array under the scope, from
            [`read_arrays`][blobmap.hierarchy.read_arrays] or written by hand.
        policy: Thresholds. Defaults to `previous.policy` when repartitioning,
            else to [`Policy`][blobmap.model.Policy] defaults.
        previous: The manifest currently in effect, whose blobs are pinned.

    Returns:
        A validated `Manifest`. The epoch is carried over unchanged; callers
        that intend to write should use
        [`bumped`][blobmap.model.Manifest.bumped].

    Raises:
        ValueError: If the result fails validation, which would indicate a
            bug in the cut rather than bad input.

    Example:
        >>> from blobmap.model import Array, GiB
        >>> first = partition("s", [Array("tas", (400, 4, 4), (10, 4, 4), 4,
        ...                                stored_bytes=400 * GiB)])
        >>> grown = partition("s", [Array("tas", (800, 4, 4), (10, 4, 4), 4,
        ...                                stored_bytes=800 * GiB)],
        ...                   previous=first)
        >>> diff(first, grown).is_empty          # an append changes nothing
        True
    """
    policy = policy or (previous.policy if previous else Policy())

    # Pins are intent, so they survive a repartition. hot_always is derived
    # from structure, so it is recomputed.
    pinned: tuple[Pin, ...] = previous.pinned if previous else ()

    hot: list[str] = default_hot_always()
    payload: list[Array] = []
    for a in arrays:
        if a.is_coordinate or a.total_bytes < policy.t_hot_bytes:
            hot.append(f"{a.path}/**" if a.path else "**")
        elif any(pin.covers(a.path) for pin in pinned):
            # deliberately held hot; giving it a blob would let the tiering
            # policy archive it the moment the pin is forgotten
            hot.append(f"{a.path}/**" if a.path else "**")
        else:
            payload.append(a)

    carried: tuple[Blob, ...] = previous.blobs if previous else ()
    unclaimed = [a for a in payload if not _claimed(a, carried)]
    taken = {b.id for b in carried}
    blobs = list(carried) + _cut(unclaimed, policy, taken, fresh=not carried)

    manifest = Manifest(
        scope=scope,
        blobs=tuple(blobs),
        hot_always=tuple(hot),
        pinned=pinned,
        policy=policy,
        epoch=previous.epoch if previous else 1,
        generated_at=now(),
        provenance={
            a.path or ".": {
                "objects": a.nobjects,
                "objects_seen": a.nobjects_seen,
                "stored_bytes": a.total_bytes,
                "uncompressed_object_bytes": a.uncompressed_object_bytes,
                "sharded": a.shards is not None,
                "key_encoding": a.key_encoding,
            }
            for a in arrays
        },
    )
    manifest.validate()
    return manifest

diff ¶

diff(old: Manifest | None, new: Manifest) -> Diff

Compare two manifests by blob id.

Parameters:

Name	Type	Description	Default
`old`	`Manifest \| None`	The manifest previously in effect, or `None` for a first run.	required
`new`	`Manifest`	The freshly computed manifest.	required

Returns:

Type	Description
`Diff`	A `Diff`. When `old` is `None` everything counts as added.

Source code in src/blobmap/partition.py

def diff(old: Manifest | None, new: Manifest) -> Diff:
    """Compare two manifests by blob id.

    Args:
        old: The manifest previously in effect, or `None` for a first run.
        new: The freshly computed manifest.

    Returns:
        A `Diff`. When `old` is `None` everything counts as added.
    """
    if old is None:
        return Diff(added=new.blobs)
    a, b = old.by_id(), new.by_id()
    return Diff(
        added=tuple(b[k] for k in b if k not in a),
        removed=tuple(a[k] for k in a if k not in b),
        modified=tuple((a[k], b[k]) for k in a if k in b and a[k] != b[k]),
    )

partition

blobmap.partition ¶

Diff dataclass ¶

is_additive property ¶

is_empty property ¶

describe ¶

bucket_width ¶

partition ¶

diff ¶

Diff `dataclass` ¶

is_additive `property` ¶

is_empty `property` ¶