hierarchy

blobmap.hierarchy

Reading a store: metadata objects for structure, LIST for sizes.

Array roots come from where the metadata objects are, never from the shape of chunk keys. That is exact rather than heuristic: an array root is a prefix holding a zarr.json with node_type: array, or a .zarray. Guessing from key shape, by looking for a c segment or a numeric one, breaks on a group legitimately named c, on v2 flat keys, and on datatrees.

Nothing here reads a chunk

Only metadata objects are fetched, and those are in hot_always. So partitioning a store cannot trigger a tape restore, and cannot feed its own event loop.

Attributes:

Name	Type	Description
V3_MARKER		Basename identifying a zarr v3 node.
V2_GROUP		Basename identifying a zarr v2 group.
V2_ARRAY		Basename identifying a zarr v2 array.

Example

from obstore.store import S3Store
from blobmap import read_arrays

arrays = read_arrays(S3Store(bucket="cordex"), "nukleus/eur11.zarr")
for a in arrays:
    print(a.path, a.nobjects, a.total_bytes, a.key_encoding)

NotAZarrStore

Bases: ValueError

No zarr metadata was found under the given prefix.

excluded

excluded(
    key: str, exclude: Sequence[str] = DEFAULT_EXCLUDE
) -> bool

Whether a key sits under an excluded path segment.

Matches whole segments, so .sgwtmp skips .sgwtmp/multipart/x but a file merely named data.sgwtmp.nc is kept.

Parameters:

Name	Type	Description	Default
key	str	Object key.	required
exclude	Sequence[str]	Path segments to skip.	DEFAULT_EXCLUDE

Returns:

Type	Description
bool	True if any segment of the key is excluded.

Example

excluded(".sgwtmp/multipart/staging-abc") True excluded("healpix/mean.zarr/tas/0/0") False excluded("healpix/data.sgwtmp.nc") False

Source code in src/blobmap/hierarchy.py

def excluded(key: str, exclude: Sequence[str] = DEFAULT_EXCLUDE) -> bool:
    """Whether a key sits under an excluded path segment.

    Matches whole segments, so `.sgwtmp` skips `.sgwtmp/multipart/x` but a
    file merely *named* `data.sgwtmp.nc` is kept.

    Args:
        key: Object key.
        exclude: Path segments to skip.

    Returns:
        True if any segment of the key is excluded.

    Example:
        >>> excluded(".sgwtmp/multipart/staging-abc")
        True
        >>> excluded("healpix/mean.zarr/tas/0/0")
        False
        >>> excluded("healpix/data.sgwtmp.nc")
        False
    """
    return any(part in exclude for part in key.split("/"))

detect_format

detect_format(store: Store, scope: str) -> str | None

Identify the zarr format at a prefix.

Parameters:

Name	Type	Description	Default
store	Store	A storage handle.	required
scope	str	Prefix to inspect.	required

Returns:

Type	Description
str | None	"v3", "v2", or None when this is not a store root. One
str | None	delimited LIST, no recursion, so it is cheap enough to call at every
str | None	level of a scan.

Source code in src/blobmap/hierarchy.py

def detect_format(store: Store, scope: str) -> str | None:
    """Identify the zarr format at a prefix.

    Args:
        store: A storage handle.
        scope: Prefix to inspect.

    Returns:
        `"v3"`, `"v2"`, or `None` when this is not a store root. One
        delimited LIST, no recursion, so it is cheap enough to call at every
        level of a scan.
    """
    names = list_names(store, _prefix(scope))
    if V3_MARKER in names:
        return "v3"
    if names & {V2_GROUP, V2_ARRAY}:
        return "v2"
    return None

find_store_root

find_store_root(
    store: Store, prefix: str, ceiling: str = ""
) -> str | None

Climb to the outermost prefix that still holds zarr metadata.

A metadata write tells you a node changed, not which store it belongs to. Debouncing on the node would fragment the pending set per variable and, worse, hand the partitioner a sub-array as if it were a store root. Pass this as root_of to EventPoller.

Parameters:

Name	Type	Description	Default
store	Store	A storage handle.	required
prefix	str	Where to start climbing, usually a metadata object's parent.	required
ceiling	str	Do not climb above this prefix.	''

Returns:

Type	Description
str | None	The store root, or None if nothing on the way up looks like zarr.
str | None	Costs one delimited LIST per level.

Source code in src/blobmap/hierarchy.py

def find_store_root(store: Store, prefix: str, ceiling: str = "") -> str | None:
    """Climb to the outermost prefix that still holds zarr metadata.

    A metadata write tells you a *node* changed, not which store it belongs
    to. Debouncing on the node would fragment the pending set per variable
    and, worse, hand the partitioner a sub-array as if it were a store root.
    Pass this as `root_of` to
    [`EventPoller`][blobmap.discover.events.EventPoller].

    Args:
        store: A storage handle.
        prefix: Where to start climbing, usually a metadata object's parent.
        ceiling: Do not climb above this prefix.

    Returns:
        The store root, or `None` if nothing on the way up looks like zarr.
        Costs one delimited LIST per level.
    """
    parts = [p for p in prefix.strip("/").split("/") if p]
    floor = len([p for p in ceiling.strip("/").split("/") if p])
    found: str | None = None
    for i in range(len(parts), floor, -1):
        candidate = "/".join(parts[:i])
        if detect_format(store, candidate) is not None:
            found = candidate
    return found

read_arrays

read_arrays(
    store: Store,
    scope: str,
    *,
    exclude: Sequence[str] = DEFAULT_EXCLUDE
) -> list[Array]

Describe every array under a scope, sized from a single LIST.

Parameters:

Name	Type	Description	Default
store	Store	A storage handle for the data.	required
scope	str	Prefix to read, usually a store root.	required
exclude	Sequence[str]	Path segments to skip, defaulting to DEFAULT_EXCLUDE. Relevant when scanning a gateway's backing filesystem, which exposes staging directories the S3 API hides.	DEFAULT_EXCLUDE

Note

The listing is walked twice: once for metadata keys, once to accumulate sizes. That keeps memory proportional to the number of arrays rather than the number of objects, which matters at HEALPix scale where a single store holds millions of chunks. It also means the two passes could disagree if the store is written concurrently; an object appearing between them is reported as belonging to no array. Debouncing before partitioning is what avoids that.

Returns:

Type	Description
list[Array]	One Array per array found, sorted by path.
list[Array]	Arrays whose metadata cannot be read are skipped with a warning
list[Array]	rather than failing the whole run.

Raises:

Type	Description
NotAZarrStore	If no zarr metadata is found at all.

Note

Logs a warning when data objects belong to no array. Those resolve as unmanaged and stay hot, which is safe but means storage nobody is tiering.

Source code in src/blobmap/hierarchy.py

def read_arrays(
    store: Store, scope: str, *, exclude: Sequence[str] = DEFAULT_EXCLUDE
) -> list[Array]:
    """Describe every array under a scope, sized from a single LIST.

    Args:
        store: A storage handle for the data.
        scope: Prefix to read, usually a store root.
        exclude: Path segments to skip, defaulting to
            `DEFAULT_EXCLUDE`. Relevant
            when scanning a gateway's backing filesystem, which exposes
            staging directories the S3 API hides.

    Note:
        The listing is walked twice: once for metadata keys, once to
        accumulate sizes. That keeps memory proportional to the number of
        arrays rather than the number of objects, which matters at HEALPix
        scale where a single store holds millions of chunks. It also means the
        two passes could disagree if the store is written concurrently; an
        object appearing between them is reported as belonging to no array.
        Debouncing before partitioning is what avoids that.

    Returns:
        One [`Array`][blobmap.model.Array] per array found, sorted by path.
        Arrays whose metadata cannot be read are skipped with a warning
        rather than failing the whole run.

    Raises:
        NotAZarrStore: If no zarr metadata is found at all.

    Note:
        Logs a warning when data objects belong to no array. Those resolve as
        unmanaged and stay hot, which is safe but means storage nobody is
        tiering.
    """
    prefix = _prefix(scope)

    # Pass one keeps only metadata keys. Retaining every data entry costs
    # roughly 200 bytes each, which for a HEALPix store at zoom 9 is hundreds
    # of megabytes to gigabytes -- and looks like a hang rather than a
    # slowdown, because nothing is printed while it accumulates.
    metadata_keys: list[str] = []
    seen = 0
    for entry in list_all(store, prefix):
        seen += 1
        if seen % PROGRESS_EVERY == 0:
            log.info(
                "%s: %d objects listed, %d metadata",
                scope or ".",
                seen,
                len(metadata_keys),
            )
        if excluded(_relative(entry.key, prefix), exclude):
            continue
        if posixpath.basename(entry.key) in METADATA_BASENAMES:
            metadata_keys.append(entry.key)

    if not metadata_keys:
        raise NotAZarrStore(f"{scope}: no zarr metadata found")
    log.debug(
        "%s: %d objects, %d metadata objects", scope or ".", seen, len(metadata_keys)
    )

    # v2 keeps attributes in a separate .zattrs object, so _ARRAY_DIMENSIONS
    # is invisible unless we read it. Without this every v2 coordinate looks
    # like a plain array, and only the size rule keeps it off tape.
    attributes: dict[str, dict[str, Any]] = {}
    for key in metadata_keys:
        if posixpath.basename(key) != V2_ATTRS:
            continue
        loaded = _load(store, key)
        if loaded is not None:
            attributes[_relative(posixpath.dirname(key), prefix)] = loaded

    arrays: dict[str, dict[str, Any]] = {}
    for key in metadata_keys:
        base = posixpath.basename(key)
        if base not in (V3_MARKER, V2_ARRAY):
            continue
        meta = _load(store, key)
        if meta is None or not _is_array(meta):
            continue
        root = _relative(posixpath.dirname(key), prefix)
        if base == V2_ARRAY and root in attributes:
            # normalise onto the v3 shape so _is_coordinate has one code path
            meta = {**meta, "attributes": attributes[root]}
        arrays[root] = meta

    # Pass two streams the listing again, accumulating per array rather than
    # per object, so memory is bounded by the number of arrays. Two walks cost
    # less than holding the first one.
    nodes: dict[str, _Node] = {root: _Node() for root in arrays}
    unassigned = 0
    counted = 0
    for entry in list_all(store, prefix):
        relative = _relative(entry.key, prefix)
        if excluded(relative, exclude):
            continue
        if posixpath.basename(entry.key) in METADATA_BASENAMES:
            continue
        counted += 1
        if counted % PROGRESS_EVERY == 0:
            log.info("%s: %d objects sized", scope or ".", counted)
        owner = _owner(relative, nodes)
        if owner is None:
            unassigned += 1
            continue
        node = nodes[owner]
        node.stored_bytes += entry.size
        node.nobjects += 1
        if len(node.sample_keys) < 4:
            node.sample_keys.append(relative)
    if unassigned:
        log.warning(
            "%s: %d objects belong to no array; they will resolve as "
            "unmanaged and stay hot",
            scope,
            unassigned,
        )

    out: list[Array] = []
    for root in sorted(arrays):
        array = _to_array(root, arrays[root], nodes[root])
        if array is None:
            continue
        if array.nobjects_seen and array.nobjects_seen > array.nobjects:
            # the declared grid says fewer objects than are actually stored,
            # so something is left over from an append or a rechunk. This
            # skews avg_object_bytes and therefore the chosen bucket width.
            log.warning(
                "%s: %d objects stored but the declared grid holds %d "
                "(shape=%s, object shape=%s) -- stale chunks from an append "
                "or rechunk will skew sizing",
                array.path or ".",
                array.nobjects_seen,
                array.nobjects,
                array.shape,
                array.object_shape,
            )
        out.append(array)
    return out