diff --git a/src/datajoint/__init__.py b/src/datajoint/__init__.py index 2e242ce84..9728d2781 100644 --- a/src/datajoint/__init__.py +++ b/src/datajoint/__init__.py @@ -284,7 +284,7 @@ def FreeTable(conn_or_name, full_table_name: str | None = None) -> _FreeTable: "diagram": (".diagram", None), # Return the module itself # cli imports click "cli": (".cli", "cli"), - # gc — exposed lazily so `dj.gc.scan(...)` works as documented in gc.py + # gc — exposed lazily so `dj.gc.GarbageCollector(...)` works as documented in gc.py # and in the user docs (how-to/garbage-collection.md). "gc": (".gc", None), # Return the module itself } diff --git a/src/datajoint/builtin_codecs/hash.py b/src/datajoint/builtin_codecs/hash.py index bb3a3852f..4873c073f 100644 --- a/src/datajoint/builtin_codecs/hash.py +++ b/src/datajoint/builtin_codecs/hash.py @@ -21,7 +21,7 @@ class HashCodec(Codec): The database column stores JSON metadata: ``{hash, store, size}``. Duplicate content is automatically deduplicated across all tables. - Deletion: Requires garbage collection via ``dj.gc.collect()``. + Deletion: Requires garbage collection via ``dj.gc.GarbageCollector``. External only - requires @ modifier. diff --git a/src/datajoint/builtin_codecs/npy.py b/src/datajoint/builtin_codecs/npy.py index 54853437b..3094dfb90 100644 --- a/src/datajoint/builtin_codecs/npy.py +++ b/src/datajoint/builtin_codecs/npy.py @@ -272,7 +272,7 @@ class Recording(dj.Manual): - Path: ``{schema}/{table}/{pk}/{attribute}.npy`` - Database column: JSON with ``{path, store, dtype, shape}`` - Deletion: Requires garbage collection via ``dj.gc.collect()``. + Deletion: Requires garbage collection via ``dj.gc.GarbageCollector``. See Also -------- diff --git a/src/datajoint/builtin_codecs/object.py b/src/datajoint/builtin_codecs/object.py index 1c0d8c673..08ef0e188 100644 --- a/src/datajoint/builtin_codecs/object.py +++ b/src/datajoint/builtin_codecs/object.py @@ -53,7 +53,7 @@ def make(self, key): {store_root}/{schema}/{table}/{pk}/{field}/ - Deletion: Requires garbage collection via ``dj.gc.collect()``. + Deletion: Requires garbage collection via ``dj.gc.GarbageCollector``. Comparison with hash-addressed:: diff --git a/src/datajoint/builtin_codecs/schema.py b/src/datajoint/builtin_codecs/schema.py index c8cc0759d..11c6c5750 100644 --- a/src/datajoint/builtin_codecs/schema.py +++ b/src/datajoint/builtin_codecs/schema.py @@ -114,7 +114,7 @@ def _build_path( Build schema-addressed storage path. Constructs a path that mirrors the database schema structure: - ``{schema}/{table}/{pk_values}/{field}{ext}`` + ``{schema_prefix}/{schema}/{table}/{pk_values}/{field}{ext}`` Supports partitioning if configured in the store. @@ -150,6 +150,7 @@ def _build_path( spec = config.get_store_spec(store_name) partition_pattern = spec.get("partition_pattern") token_length = spec.get("token_length", 8) + schema_prefix = spec["schema_prefix"] # always present: settings applies the default return build_object_path( schema=schema, @@ -159,6 +160,7 @@ def _build_path( ext=ext, partition_pattern=partition_pattern, token_length=token_length, + schema_prefix=schema_prefix, ) def _get_backend(self, store_name: str | None = None, config=None): diff --git a/src/datajoint/codecs.py b/src/datajoint/codecs.py index 2719e9509..c68e33413 100644 --- a/src/datajoint/codecs.py +++ b/src/datajoint/codecs.py @@ -38,6 +38,7 @@ class MyTable(dj.Manual): from __future__ import annotations +import json import logging from abc import ABC, abstractmethod from typing import Any @@ -219,6 +220,45 @@ def validate(self, value: Any) -> None: """ pass + def referenced_paths(self, stored: Any) -> list[tuple[str, str | None]]: + """ + Return the external store paths this stored value references. + + Garbage collection and delete-time cleanup call this on the *stored* + (encoded) representation of a column value — the JSON/dict already in + the database — so discovery stays metadata-only (no download or decode). + + The default recognizes DataJoint's standard external-storage metadata: + a dict (or JSON string) carrying a ``path`` and optional ``store``. This + covers hash-addressed (````/````/````) and + schema-addressed (````/```` and any :class:`SchemaCodec` + subclass) storage with no extra work, and returns an empty list for + values that are not externally stored (e.g. in-table ````). + + Override only if your codec stores external references in a non-standard + shape. Returning the paths here is what lets garbage collection see a + custom codec's files as *referenced* rather than orphaned (see #1469). + + Parameters + ---------- + stored : any + The stored (encoded) value as read from the database column. + + Returns + ------- + list[tuple[str, str | None]] + ``(path, store_name)`` for each external artifact referenced. + """ + value = stored + if isinstance(value, str): + try: + value = json.loads(value) + except (json.JSONDecodeError, TypeError, ValueError): + return [] + if isinstance(value, dict) and "path" in value: + return [(value["path"], value.get("store"))] + return [] + def __repr__(self) -> str: return f"<{self.__class__.__name__}(name={self.name!r})>" diff --git a/src/datajoint/gc.py b/src/datajoint/gc.py index 8c87efd84..30ce884a7 100644 --- a/src/datajoint/gc.py +++ b/src/datajoint/gc.py @@ -1,33 +1,40 @@ """ Garbage collection for object storage. -This module provides utilities to identify and remove orphaned items -from object storage. Storage items become orphaned when all database rows -referencing them are deleted. +This module identifies and removes orphaned items from object storage. Storage +items become orphaned when all database rows referencing them are deleted. DataJoint uses two object storage patterns: Hash-addressed storage Types: ````, ````, ```` - Path: ``_hash/{schema}/{hash}`` (with optional subfolding) + Path: ``{hash_prefix}/{schema}/{hash}`` (with optional subfolding; + ``hash_prefix`` defaults to ``_hash``) Deduplication: Per-schema (identical data within a schema shares storage) Deletion: Requires garbage collection Schema-addressed storage Types: ````, ```` - Path: ``{schema}/{table}/{pk}/{field}/`` + Path: ``{schema_prefix}/{schema}/{table}/{pk}/...`` (``schema_prefix`` + defaults to ``_schema``; pre-2.3.1 stores hold root-level ``{schema}/...`` paths) Deduplication: None (each entity has unique path) Deletion: Requires garbage collection -Usage:: +Garbage collection is **store-specific** and handles one store at a time. The +:class:`GarbageCollector` is bound to its schemas and store at construction; +neither is threaded through each call (the config defaults to the schemas'):: import datajoint as dj - # Scan schemas and find orphaned items - stats = dj.gc.scan(schema1, schema2, store_name='mystore') + collector = dj.gc.GarbageCollector(schema1, schema2, store="mystore") + stats = collector.collect() # read-only report (dry_run=True default) + stats = collector.collect(dry_run=False) # actually delete the orphans - # Remove orphaned items (dry_run=False to actually delete) - stats = dj.gc.collect(schema1, schema2, store_name='mystore', dry_run=True) +Both sections embed the schema name in every path, so garbage collection is +**per-schema**: each schema is scanned against its own subtree only. Orphan +detection is therefore confined to the collector's schemas — any subset of the +schemas sharing a store may be collected safely, and GC never touches another +schema's objects or user-managed content elsewhere in the store. See Also -------- @@ -36,675 +43,355 @@ from __future__ import annotations -import json import logging +import re from typing import TYPE_CHECKING, Any -from .hash_registry import delete_path, get_store_backend from .errors import DataJointError +from .hash_registry import delete_path, get_store_backend if TYPE_CHECKING: from .schemas import _Schema as Schema logger = logging.getLogger(__name__.split(".")[0]) +# Base32 content-hash filename: 26 lowercase alphanumeric chars. +_BASE32_HASH = re.compile(r"^[a-z2-7]{26}$") -def _uses_hash_storage(attr) -> bool: - """ - Check if an attribute uses hash-addressed storage. - - Hash-addressed types use content deduplication via MD5/Base32 hashing: - - - ```` - raw hash storage - - ```` - chains to ```` - - ```` - chains to ```` - Parameters - ---------- - attr : Attribute - Attribute from table heading. - - Returns - ------- - bool - True if the attribute uses hash-addressed storage. +def _is_covered(path: str, referenced: set[str]) -> bool: """ - if not attr.codec: - return False + Return True if a stored file is accounted for by a referenced object path. - codec_name = getattr(attr.codec, "name", "") - store = getattr(attr, "store", None) + A stored file is covered when: - # always uses hash-addressed storage (external only) - if codec_name == "hash": - return True + - it IS a referenced path (single-file object, e.g. ``field_token.npy``); + - it lies UNDER a referenced path (directory-valued object, e.g. a Zarr + store ``field_token/`` whose stored form is many chunk files); or + - it is the ``.manifest.json`` sidecar of a referenced object (written + alongside folder-valued objects). - # and use hash-addressed storage when external - if codec_name in ("blob", "attach") and store is not None: + Uses O(path-depth) ancestor-prefix lookups against the referenced set. + """ + if path.endswith(".manifest.json"): + path = path[: -len(".manifest.json")] + if path in referenced: return True - + idx = path.rfind("/") + while idx > 0: + if path[:idx] in referenced: + return True + idx = path.rfind("/", 0, idx) return False -def _uses_schema_storage(attr) -> bool: - """ - Check if an attribute uses schema-addressed storage. - - Schema-addressed types store data at paths derived from the schema structure: - - - ```` - arbitrary objects (pickled or native formats) - - ```` - NumPy arrays with lazy loading - - Parameters - ---------- - attr : Attribute - Attribute from table heading. - - Returns - ------- - bool - True if the attribute uses schema-addressed storage. - """ - if not attr.codec: - return False - - codec_name = getattr(attr.codec, "name", "") - return codec_name in ("object", "npy") - - -def _extract_hash_refs(value: Any) -> list[tuple[str, str | None]]: - """ - Extract path references from hash-addressed storage metadata. - - Hash-addressed storage stores metadata as JSON with ``path`` and ``hash`` keys. - The path is used for file operations; the hash is for integrity verification. - - Parameters - ---------- - value : Any - The stored value (JSON string or dict). - - Returns - ------- - list[tuple[str, str | None]] - List of (path, store_name) tuples. - """ - refs = [] - - if value is None: - return refs - - # Parse JSON if string - if isinstance(value, str): - try: - value = json.loads(value) - except (json.JSONDecodeError, TypeError): - return refs - - # Extract path from dict (path is required for new data, hash for legacy) - if isinstance(value, dict) and "path" in value: - refs.append((value["path"], value.get("store"))) - - return refs - - -def _extract_schema_refs(value: Any) -> list[tuple[str, str | None]]: - """ - Extract schema-addressed path references from a stored value. - - Schema-addressed storage stores metadata as JSON with a ``path`` key. - - Parameters - ---------- - value : Any - The stored value (JSON string or dict). - - Returns - ------- - list[tuple[str, str | None]] - List of (path, store_name) tuples. - """ - refs = [] - - if value is None: - return refs - - # Parse JSON if string - if isinstance(value, str): - try: - value = json.loads(value) - except (json.JSONDecodeError, TypeError): - return refs - - # Extract path from dict - if isinstance(value, dict) and "path" in value: - refs.append((value["path"], value.get("store"))) - - return refs - - -def scan_hash_references( - *schemas: "Schema", - store_name: str | None = None, - verbose: bool = False, -) -> set[str]: +class GarbageCollector: """ - Scan schemas for hash-addressed storage references. + Store-specific garbage collector — one store, a fixed set of schemas. - Examines all tables in the given schemas and extracts storage paths - from columns that use hash-addressed storage (````, ````, - ````). + The schemas to collect and the store are bound at construction, so neither + is threaded through individual operations. The store is resolved eagerly + (an unknown/misconfigured store raises immediately). Storage-side work (the + stored-file listings, deletion) uses this store; database metadata is read + through each schema's own connection. Parameters ---------- *schemas : Schema - Schema instances to scan. - store_name : str, optional - Only include references to this store (None = all stores). - verbose : bool, optional - Print progress information. - - Returns - ------- - set[str] - Set of storage paths that are referenced. + The schemas to collect. At least one is required. Every managed path + embeds its schema, so collection is per-schema and confined to these + schemas: objects of any other schema sharing the store are never listed + and never at risk — pass any subset safely. + store : str, optional + Store name (None = the default store), keyword-only. + config : Config, optional + Config that defines the store. Defaults to the first schema's + connection config (``schemas[0].connection._config``). """ - referenced: set[str] = set() - - for schema in schemas: - if verbose: - logger.info(f"Scanning schema: {schema.database}") - # Get all tables in schema - for table_name in schema.list_tables(): - try: - # Get table class - table = schema.get_table(table_name) - - # Check each attribute for hash-addressed storage - for attr_name, attr in table.heading.attributes.items(): - if not _uses_hash_storage(attr): + def __init__(self, *schemas: "Schema", store: str | None = None, config=None) -> None: + if not schemas: + raise DataJointError("At least one schema must be provided") + self.schemas = schemas + self.store = store + self.config = config if config is not None else schemas[0].connection._config + # Resolve the store eagerly: validates it exists and pins the backend + # and section prefixes for this collector's lifetime (one store only). + self.backend = get_store_backend(store, config=self.config) + spec = self.config.get_store_spec(store) + self._hash_prefix = spec["hash_prefix"].strip("/") # settings applies the "_hash" default + self._schema_prefix = spec["schema_prefix"].strip("/") # ... "_schema" default + + # ------------------------------------------------------------------ # + # References — extracted from database metadata (per-schema connection) + # ------------------------------------------------------------------ # + def hash_references(self, verbose: bool = False) -> set[str]: + """ + Full relative store paths of hash-addressed objects referenced by live + rows across this collector's schemas. + + Reads columns classified by ``Heading.hash_objects`` (````, and + external ````/````). Paths are taken verbatim from each + row's metadata (independent of the store's current prefixes) and + filtered to this collector's store. + """ + return self._references("hash_objects", verbose) + + def schema_references(self, verbose: bool = False) -> set[str]: + """ + Full relative store paths of schema-addressed objects referenced by live + rows across this collector's schemas. + + Reads columns classified by ``Heading.schema_objects`` — any + ``SchemaCodec`` (````, ````, custom subclasses, + recognized by type per #1469). For a directory-valued object the + recorded path is the directory prefix, not its individual files. + """ + return self._references("schema_objects", verbose) + + def _references(self, heading_attr: str, verbose: bool) -> set[str]: + referenced: set[str] = set() + for schema in self.schemas: + if verbose: + logger.info(f"Scanning schema {schema.database} for {heading_attr}") + for table_name in schema.list_tables(): + try: + table = schema.get_table(table_name) + # Classification lives on the heading — one source of truth. + for attr_name in getattr(table.heading, heading_attr): + attr = table.heading.attributes[attr_name] + if verbose: + logger.info(f" Scanning {table_name}.{attr_name}") + # Read raw JSON metadata via cursor — bypasses + # decode_attribute, so we get the stored dict + # (PostgreSQL/JSONB) or JSON string (MySQL). The codec's + # referenced_paths() extracts paths and handles both + # shapes (codec-driven discovery, #1469). + try: + for row in table.proj(attr_name).cursor(as_dict=True): + for path, ref_store in attr.codec.referenced_paths(row[attr_name]): + if self.store is None or ref_store == self.store: + referenced.add(path) + except Exception as e: + logger.warning(f"Error scanning {table_name}.{attr_name}: {e}") + except Exception as e: + logger.warning(f"Error accessing table {table_name}: {e}") + return referenced + + # ------------------------------------------------------------------ # + # Stored listings — walk this store, one schema's subtree + # ------------------------------------------------------------------ # + def list_hash_paths(self, schema_name: str) -> dict[str, int]: + """ + List one schema's hash-addressed items: ``{hash_prefix}/{schema}/...``. + + Every hash path embeds the schema and deduplication is per-schema, so + this listing is complete for that schema and independent of every other + schema on the store. Only the currently configured prefix is walked; + objects under a former prefix are not listed (they remain readable via + their metadata paths). + + Returns a dict mapping each object's full relative store path to size. + """ + hp = self._hash_prefix + section = f"{hp}/{schema_name}/" if hp else f"{schema_name}/" + full_root = self.backend._full_path("") + stored: dict[str, int] = {} + try: + for root, _dirs, files in self.backend.fs.walk(self.backend._full_path(section)): + for filename in files: + if filename.endswith(".manifest.json"): + continue # folder-object sidecar, not a hash file + if not _BASE32_HASH.match(filename): continue - - if verbose: - logger.info(f" Scanning {table_name}.{attr_name}") - - # Read raw JSON metadata via cursor — bypasses decode_attribute - # so we get the stored dict (PostgreSQL/JSONB) or JSON string - # (MySQL), not the decoded codec output. _extract_hash_refs - # handles both shapes. + file_path = f"{root}/{filename}" try: - cursor = table.proj(attr_name).cursor(as_dict=True) - for row in cursor: - for path, ref_store in _extract_hash_refs(row[attr_name]): - # Filter by store if specified - if store_name is None or ref_store == store_name: - referenced.add(path) - except Exception as e: - logger.warning(f"Error scanning {table_name}.{attr_name}: {e}") - - except Exception as e: - logger.warning(f"Error accessing table {table_name}: {e}") - - return referenced - - -def scan_schema_references( - *schemas: "Schema", - store_name: str | None = None, - verbose: bool = False, -) -> set[str]: - """ - Scan schemas for schema-addressed storage references. - - Examines all tables in the given schemas and extracts paths from columns - that use schema-addressed storage (````, ````). - - Parameters - ---------- - *schemas : Schema - Schema instances to scan. - store_name : str, optional - Only include references to this store (None = all stores). - verbose : bool, optional - Print progress information. - - Returns - ------- - set[str] - Set of storage paths that are referenced. - """ - referenced: set[str] = set() - - for schema in schemas: - if verbose: - logger.info(f"Scanning schema for schema-addressed storage: {schema.database}") - - # Get all tables in schema - for table_name in schema.list_tables(): - try: - # Get table class - table = schema.get_table(table_name) - - # Check each attribute for schema-addressed storage - for attr_name, attr in table.heading.attributes.items(): - if not _uses_schema_storage(attr): - continue - - if verbose: - logger.info(f" Scanning {table_name}.{attr_name}") - - # Read raw JSON metadata via cursor — bypasses decode_attribute - # so we get the stored dict (PostgreSQL/JSONB) or JSON string - # (MySQL), not the decoded codec output. _extract_schema_refs - # handles both shapes. + relative_path = file_path.replace(full_root, "").lstrip("/") + stored[relative_path] = self.backend.fs.size(file_path) + except Exception: + pass + except FileNotFoundError: + pass # the hash section does not exist yet + except Exception as e: + logger.warning(f"Error listing stored hashes: {e}") + return stored + + def list_schema_paths(self, schema_name: str) -> dict[str, int]: + """ + List one schema's schema-addressed object files: ``{schema_prefix}/{schema}/...``. + + Enumerates individual files: a single-file object is one file at + ``{schema_prefix}/{schema}/{table}/{pk}/{field}_{token}[.ext]``; a + directory-valued object (e.g. a Zarr store) is many files under that + prefix, plus a ``{path}.manifest.json`` sidecar — all listed. Orphan + detection matches them against references via :func:`_is_covered`. + + The walk is confined to the schema's own section, so it never enters the + hash or filepath sections (mutually-exclusive top-level prefixes, + enforced by store validation) — GC's blast radius is exactly + ``{schema_prefix}/{schema}/``; user ```` content and other + schemas' objects are structurally out of reach. + + Legacy layout: DataJoint 2.3.0 and earlier wrote schema-addressed + objects at root level ``{schema}/...`` (``schema_prefix`` ignored, see + datajoint/datajoint-python#1487). Such a store should set + ``schema_prefix=""`` so writes and this walk use the root-level layout + consistently; otherwise those objects remain readable but outside the + scanned section. + + Returns a dict mapping each object file's relative path to size. + """ + sp = self._schema_prefix + rel = f"{sp}/{schema_name}" if sp else schema_name + full_root = self.backend._full_path("") + stored: dict[str, int] = {} + try: + for root, _dirs, files in self.backend.fs.walk(self.backend._full_path(rel)): + for filename in files: + # Manifest sidecars are INCLUDED: owned by their object and + # reclaimed with it. _is_covered() keeps a live object's + # manifest out of the orphan set. + file_path = f"{root}/{filename}" + relative_path = file_path.replace(full_root, "").lstrip("/") try: - cursor = table.proj(attr_name).cursor(as_dict=True) - for row in cursor: - for path, ref_store in _extract_schema_refs(row[attr_name]): - # Filter by store if specified - if store_name is None or ref_store == store_name: - referenced.add(path) - except Exception as e: - logger.warning(f"Error scanning {table_name}.{attr_name}: {e}") - - except Exception as e: - logger.warning(f"Error accessing table {table_name}: {e}") - - return referenced - - -def list_stored_hashes(store_name: str | None = None, config=None) -> dict[str, int]: - """ - List all hash-addressed items in storage. - - Scans the ``_hash/`` directory in the specified store and returns - all storage paths found. These correspond to ````, ````, - and ```` types. - - Parameters - ---------- - store_name : str, optional - Store to scan (None = default store). - config : Config, optional - Config instance. If None, falls back to global settings.config. - - Returns - ------- - dict[str, int] - Dict mapping storage path to size in bytes. - """ - import re - - backend = get_store_backend(store_name, config=config) - stored: dict[str, int] = {} - - # Hash-addressed storage: _hash/{schema}/{subfolders...}/{hash} - hash_prefix = "_hash/" - # Base32 pattern: 26 lowercase alphanumeric chars - base32_pattern = re.compile(r"^[a-z2-7]{26}$") - - try: - full_prefix = backend._full_path(hash_prefix) - - for root, dirs, files in backend.fs.walk(full_prefix): - for filename in files: - # Skip manifest files - if filename.endswith(".manifest.json"): - continue - - # The filename is the base32 hash - content_hash = filename - - # Validate it looks like a base32 hash - if base32_pattern.match(content_hash): + stored[relative_path] = self.backend.fs.size(file_path) + except Exception: + stored[relative_path] = 0 + except FileNotFoundError: + pass # this schema's section does not exist yet + except Exception as e: + logger.warning(f"Error listing stored schemas: {e}") + return stored + + # ------------------------------------------------------------------ # + # Deletion + # ------------------------------------------------------------------ # + def delete_schema_path(self, path: str) -> bool: + """ + Delete a single schema-addressed object file (not a directory). + + ``path`` is an individual file (as enumerated by + :meth:`list_schema_paths`), so only the orphaned token file is removed, + leaving other versions/fields under the same primary-key directory + intact. Empty parent directories are pruned best-effort. Returns True if + deleted, False if not found. + """ + try: + full_path = self.backend._full_path(path) + if self.backend.fs.exists(full_path): + self.backend.fs.rm(full_path) + logger.debug(f"Deleted schema object: {path}") + # Best-effort: prune now-empty parent directories to store root. + root = self.backend._full_path("").rstrip("/") + parent = full_path.rsplit("/", 1)[0] + while parent and parent != root and parent.startswith(root): try: - file_path = f"{root}/{filename}" - size = backend.fs.size(file_path) - # Build relative path for comparison with stored metadata - # Path format: _hash/{schema}/{subfolders...}/{hash} - relative_path = file_path.replace(backend._full_path(""), "").lstrip("/") - stored[relative_path] = size + if self.backend.fs.ls(parent): + break # not empty + self.backend.fs.rmdir(parent) except Exception: - pass - - except FileNotFoundError: - # No _hash/ directory exists yet - pass - except Exception as e: - logger.warning(f"Error listing stored hashes: {e}") - - return stored - - -def list_schema_paths(store_name: str | None = None, config=None) -> dict[str, int]: - """ - List all schema-addressed items in storage. - - Scans for directories matching the schema-addressed storage pattern: - ``{schema}/{table}/{pk}/{field}/`` - - Parameters - ---------- - store_name : str, optional - Store to scan (None = default store). - config : Config, optional - Config instance. If None, falls back to global settings.config. - - Returns - ------- - dict[str, int] - Dict mapping storage path to size in bytes. - """ - backend = get_store_backend(store_name, config=config) - stored: dict[str, int] = {} - - try: - # Walk the storage looking for schema-addressed paths - full_prefix = backend._full_path("") - - for root, dirs, files in backend.fs.walk(full_prefix): - # Skip _hash directory (hash-addressed storage) - if "_hash" in root: - continue - - # Look for schema-addressed pattern (has files, not in _hash) - # Schema-addressed paths: {schema}/{table}/{pk}/{field}/ - relative_path = root.replace(full_prefix, "").lstrip("/") - - # Skip empty paths and root-level directories - if not relative_path or relative_path.count("/") < 2: - continue - - # Calculate total size of this directory - total_size = 0 - for file in files: - try: - file_path = f"{root}/{file}" - total_size += backend.fs.size(file_path) - except Exception: - pass - - # Only count directories with files (actual objects) - if total_size > 0 or files: - stored[relative_path] = total_size - - except FileNotFoundError: - pass - except Exception as e: - logger.warning(f"Error listing stored schemas: {e}") - - return stored - - -def delete_schema_path(path: str, store_name: str | None = None, config=None) -> bool: - """ - Delete a schema-addressed directory from storage. - - Parameters - ---------- - path : str - Storage path (relative to store root). - store_name : str, optional - Store name (None = default store). - config : Config, optional - Config instance. If None, falls back to global settings.config. - - Returns - ------- - bool - True if deleted, False if not found. - """ - backend = get_store_backend(store_name, config=config) - - try: - full_path = backend._full_path(path) - if backend.fs.exists(full_path): - # Remove entire directory tree - backend.fs.rm(full_path, recursive=True) - logger.debug(f"Deleted schema path: {path}") - return True - except Exception as e: - logger.warning(f"Error deleting schema path {path}: {e}") - - return False - - -def scan( - *schemas: "Schema", - store_name: str | None = None, - verbose: bool = False, -) -> dict[str, Any]: - """ - Scan for orphaned storage items without deleting. - - Scans both hash-addressed storage (for ````, ````, ````) - and schema-addressed storage (for ````, ````). - - Parameters - ---------- - *schemas : Schema - Schema instances to scan. - store_name : str, optional - Store to check (None = default store). - verbose : bool, optional - Print progress information. - - Returns - ------- - dict[str, Any] - Dict with scan statistics: - - - hash_referenced: Number of hash items referenced in database - - hash_stored: Number of hash items in storage - - hash_orphaned: Number of unreferenced hash items - - hash_orphaned_bytes: Total size of orphaned hashes - - orphaned_hashes: List of orphaned content hashes - - schema_paths_referenced: Number of schema items referenced in database - - schema_paths_stored: Number of schema items in storage - - schema_paths_orphaned: Number of unreferenced schema items - - schema_paths_orphaned_bytes: Total size of orphaned schema items - - orphaned_paths: List of orphaned schema paths - """ - if not schemas: - raise DataJointError("At least one schema must be provided") - - # Extract config from the first schema's connection - _config = schemas[0].connection._config if schemas else None - - # --- Hash-addressed storage --- - hash_referenced = scan_hash_references(*schemas, store_name=store_name, verbose=verbose) - hash_stored = list_stored_hashes(store_name, config=_config) - orphaned_hashes = set(hash_stored.keys()) - hash_referenced - hash_orphaned_bytes = sum(hash_stored.get(h, 0) for h in orphaned_hashes) - - # --- Schema-addressed storage --- - schema_paths_referenced = scan_schema_references(*schemas, store_name=store_name, verbose=verbose) - schema_paths_stored = list_schema_paths(store_name, config=_config) - orphaned_paths = set(schema_paths_stored.keys()) - schema_paths_referenced - schema_paths_orphaned_bytes = sum(schema_paths_stored.get(p, 0) for p in orphaned_paths) - - return { - # Hash-addressed storage stats - "hash_referenced": len(hash_referenced), - "hash_stored": len(hash_stored), - "hash_orphaned": len(orphaned_hashes), - "hash_orphaned_bytes": hash_orphaned_bytes, - "orphaned_hashes": sorted(orphaned_hashes), - # Schema-addressed storage stats - "schema_paths_referenced": len(schema_paths_referenced), - "schema_paths_stored": len(schema_paths_stored), - "schema_paths_orphaned": len(orphaned_paths), - "schema_paths_orphaned_bytes": schema_paths_orphaned_bytes, - "orphaned_paths": sorted(orphaned_paths), - # Combined totals - "referenced": len(hash_referenced) + len(schema_paths_referenced), - "stored": len(hash_stored) + len(schema_paths_stored), - "orphaned": len(orphaned_hashes) + len(orphaned_paths), - "orphaned_bytes": hash_orphaned_bytes + schema_paths_orphaned_bytes, - } - - -def collect( - *schemas: "Schema", - store_name: str | None = None, - dry_run: bool = True, - verbose: bool = False, -) -> dict[str, Any]: - """ - Remove orphaned storage items. - - Scans the given schemas for storage references, then removes any - items that are not referenced. - - Parameters - ---------- - *schemas : Schema - Schema instances to scan. - store_name : str, optional - Store to clean (None = default store). - dry_run : bool, optional - If True, report what would be deleted without deleting. Default True. - verbose : bool, optional - Print progress information. - - Returns - ------- - dict[str, Any] - Dict with collection statistics: - - - referenced: Total items referenced in database - - stored: Total items in storage - - orphaned: Total unreferenced items - - hash_deleted: Number of hash items deleted - - schema_paths_deleted: Number of schema items deleted - - deleted: Total items deleted (0 if dry_run) - - bytes_freed: Bytes freed (0 if dry_run) - - errors: Number of deletion errors - """ - # First scan to find orphaned items - stats = scan(*schemas, store_name=store_name, verbose=verbose) - - # Extract config from the first schema's connection - _config = schemas[0].connection._config if schemas else None - - hash_deleted = 0 - schema_paths_deleted = 0 - bytes_freed = 0 - errors = 0 - - if not dry_run: - # Delete orphaned hashes - if stats["hash_orphaned"] > 0: - hash_stored = list_stored_hashes(store_name, config=_config) + break + parent = parent.rsplit("/", 1)[0] + return True + except Exception as e: + logger.warning(f"Error deleting schema object {path}: {e}") + return False - for path in stats["orphaned_hashes"]: + # ------------------------------------------------------------------ # + # Orchestration + # ------------------------------------------------------------------ # + def collect(self, dry_run: bool = True, verbose: bool = False) -> dict[str, Any]: + """ + Report — and, unless ``dry_run``, remove — orphaned storage. + + ``dry_run=True`` (the default) is a **read-only scan**: it reports what + would be removed and deletes nothing. ``dry_run=False`` additionally + deletes the orphans. The report is the same either way, so the safe + default doubles as the inspection call. + + Every managed path embeds its schema, so a schema's stored files can + only match that schema's references — orphan detection is confined to + this collector's schemas; objects of any other schema on the store are + never listed, deleted, or at risk. Objects under a *former* prefix stay + readable via their metadata paths but are not reclamation candidates + until the setting is restored. + + Returns a dict with per-section stats and the deletion outcome: + + - hash_paths_referenced / hash_paths_stored / hash_paths_orphaned / hash_paths_orphaned_bytes + - orphaned_hash_paths: orphaned hash items as full relative store paths + (NOT bare hashes — passed as-is to the deleter) + - schema_paths_referenced / schema_paths_stored / schema_paths_orphaned + / schema_paths_orphaned_bytes + - orphaned_schema_paths: orphaned schema files as full relative store paths + - hash_paths_deleted / schema_paths_deleted / deleted / bytes_freed (0 when + dry_run) / errors / dry_run + """ + # References can be gathered across all schemas at once: because paths + # embed the schema, a file under schema X is only ever covered by an X + # reference, so combined coverage equals per-schema coverage. + hash_paths_referenced = self.hash_references(verbose=verbose) + schema_paths_referenced = self.schema_references(verbose=verbose) + + hash_paths_stored: dict[str, int] = {} + schema_paths_stored: dict[str, int] = {} + for schema in self.schemas: + hash_paths_stored.update(self.list_hash_paths(schema.database)) + schema_paths_stored.update(self.list_schema_paths(schema.database)) + + orphaned_hash_paths = sorted(set(hash_paths_stored.keys()) - hash_paths_referenced) + # Coverage, not exact set difference: a referenced path may be a + # directory-valued object (many files + a manifest sidecar). + orphaned_schema_paths = sorted(p for p in schema_paths_stored if not _is_covered(p, schema_paths_referenced)) + + hash_paths_deleted = 0 + schema_paths_deleted = 0 + bytes_freed = 0 + errors = 0 + + if not dry_run: + # The size maps from the scan above are reused for the byte tally — + # no second listing pass. + for path in orphaned_hash_paths: try: - size = hash_stored.get(path, 0) - if delete_path(path, store_name, config=_config): - hash_deleted += 1 - bytes_freed += size + if delete_path(path, self.store, config=self.config): + hash_paths_deleted += 1 + bytes_freed += hash_paths_stored.get(path, 0) if verbose: - logger.info(f"Deleted: {path} ({size} bytes)") + logger.info(f"Deleted: {path} ({hash_paths_stored.get(path, 0)} bytes)") except Exception as e: errors += 1 logger.warning(f"Failed to delete {path}: {e}") - # Delete orphaned schema paths - if stats["schema_paths_orphaned"] > 0: - schema_paths_stored = list_schema_paths(store_name, config=_config) - - for path in stats["orphaned_paths"]: + for path in orphaned_schema_paths: try: - size = schema_paths_stored.get(path, 0) - if delete_schema_path(path, store_name, config=_config): + if self.delete_schema_path(path): schema_paths_deleted += 1 - bytes_freed += size + bytes_freed += schema_paths_stored.get(path, 0) if verbose: - logger.info(f"Deleted schema path: {path} ({size} bytes)") + logger.info(f"Deleted schema path: {path} ({schema_paths_stored.get(path, 0)} bytes)") except Exception as e: errors += 1 logger.warning(f"Failed to delete schema path {path}: {e}") - return { - "referenced": stats["referenced"], - "stored": stats["stored"], - "orphaned": stats["orphaned"], - "hash_deleted": hash_deleted, - "schema_paths_deleted": schema_paths_deleted, - "deleted": hash_deleted + schema_paths_deleted, - "bytes_freed": bytes_freed, - "errors": errors, - "dry_run": dry_run, - # Include detailed stats - "hash_orphaned": stats["hash_orphaned"], - "schema_paths_orphaned": stats["schema_paths_orphaned"], - } - - -def format_stats(stats: dict[str, Any]) -> str: - """ - Format GC statistics as a human-readable string. - - Parameters - ---------- - stats : dict[str, Any] - Statistics dict from scan() or collect(). - - Returns - ------- - str - Formatted string. - """ - lines = ["Object Storage Statistics:"] - - # Show hash-addressed storage stats if present - if "hash_referenced" in stats: - lines.append("") - lines.append("Hash-Addressed Storage (, , ):") - lines.append(f" Referenced: {stats['hash_referenced']}") - lines.append(f" Stored: {stats['hash_stored']}") - lines.append(f" Orphaned: {stats['hash_orphaned']}") - if "hash_orphaned_bytes" in stats: - size_mb = stats["hash_orphaned_bytes"] / (1024 * 1024) - lines.append(f" Orphaned size: {size_mb:.2f} MB") - - # Show schema-addressed storage stats if present - if "schema_paths_referenced" in stats: - lines.append("") - lines.append("Schema-Addressed Storage (, ):") - lines.append(f" Referenced: {stats['schema_paths_referenced']}") - lines.append(f" Stored: {stats['schema_paths_stored']}") - lines.append(f" Orphaned: {stats['schema_paths_orphaned']}") - if "schema_paths_orphaned_bytes" in stats: - size_mb = stats["schema_paths_orphaned_bytes"] / (1024 * 1024) - lines.append(f" Orphaned size: {size_mb:.2f} MB") - - # Show totals - lines.append("") - lines.append("Totals:") - lines.append(f" Referenced in database: {stats['referenced']}") - lines.append(f" Stored in backend: {stats['stored']}") - lines.append(f" Orphaned (unreferenced): {stats['orphaned']}") - - if "orphaned_bytes" in stats: - size_mb = stats["orphaned_bytes"] / (1024 * 1024) - lines.append(f" Orphaned size: {size_mb:.2f} MB") - - # Show deletion results if this is from collect() - if "deleted" in stats: - lines.append("") - if stats.get("dry_run", True): - lines.append(" [DRY RUN - no changes made]") - else: - lines.append(f" Deleted: {stats['deleted']}") - if "hash_deleted" in stats: - lines.append(f" Hash items: {stats['hash_deleted']}") - if "schema_paths_deleted" in stats: - lines.append(f" Schema paths: {stats['schema_paths_deleted']}") - freed_mb = stats["bytes_freed"] / (1024 * 1024) - lines.append(f" Bytes freed: {freed_mb:.2f} MB") - if stats.get("errors", 0) > 0: - lines.append(f" Errors: {stats['errors']}") - - return "\n".join(lines) + return { + # Hash-addressed storage stats + "hash_paths_referenced": len(hash_paths_referenced), + "hash_paths_stored": len(hash_paths_stored), + "hash_paths_orphaned": len(orphaned_hash_paths), + "hash_paths_orphaned_bytes": sum(hash_paths_stored.get(h, 0) for h in orphaned_hash_paths), + "orphaned_hash_paths": orphaned_hash_paths, + # Schema-addressed storage stats + "schema_paths_referenced": len(schema_paths_referenced), + "schema_paths_stored": len(schema_paths_stored), + "schema_paths_orphaned": len(orphaned_schema_paths), + "schema_paths_orphaned_bytes": sum(schema_paths_stored.get(p, 0) for p in orphaned_schema_paths), + "orphaned_schema_paths": orphaned_schema_paths, + # Deletion outcome (all zero when dry_run) + "hash_paths_deleted": hash_paths_deleted, + "schema_paths_deleted": schema_paths_deleted, + "deleted": hash_paths_deleted + schema_paths_deleted, + "bytes_freed": bytes_freed, + "errors": errors, + "dry_run": dry_run, + } diff --git a/src/datajoint/hash_registry.py b/src/datajoint/hash_registry.py index d33c916ba..fb6f22818 100644 --- a/src/datajoint/hash_registry.py +++ b/src/datajoint/hash_registry.py @@ -5,11 +5,11 @@ codec. Content is identified by a Base32-encoded MD5 hash and stored with per-schema isolation:: - _hash/{schema}/{hash} + {hash_prefix}/{schema}/{hash} (hash_prefix defaults to _hash) With optional subfolding (configured per-store):: - _hash/{schema}/{fold1}/{fold2}/{hash} + {hash_prefix}/{schema}/{fold1}/{fold2}/{hash} Subfolding creates directory hierarchies to improve performance on filesystems that struggle with large directories (ext3, FAT32, NFS). Modern filesystems @@ -25,7 +25,7 @@ Hash-addressed storage is used by ````, ````, and ```` types. Deduplication occurs within each schema. Deletion requires garbage collection -via ``dj.gc.collect()``. +via ``dj.gc.GarbageCollector``. See Also -------- @@ -97,17 +97,19 @@ def build_hash_path( content_hash: str, schema_name: str, subfolding: tuple[int, ...] | None = None, + *, + hash_prefix: str, ) -> str: """ Build the storage path for hash-addressed storage. Path structure without subfolding:: - _hash/{schema}/{hash} + {hash_prefix}/{schema}/{hash} (hash_prefix defaults to _hash) Path structure with subfolding (e.g., (2, 2)):: - _hash/{schema}/{fold1}/{fold2}/{hash} + {hash_prefix}/{schema}/{fold1}/{fold2}/{hash} Parameters ---------- @@ -117,6 +119,10 @@ def build_hash_path( Database/schema name for isolation. subfolding : tuple[int, ...], optional Subfolding pattern from store config. None means flat (no subfolding). + hash_prefix : str + Section prefix from the store's ``hash_prefix`` setting. There is no + fallback here by design: settings (``get_store_spec``) is the single + source of the ``"_hash"`` default, applied to every store spec. Returns ------- @@ -127,12 +133,14 @@ def build_hash_path( if not (len(content_hash) == 26 and content_hash.isalnum() and content_hash.islower()): raise DataJointError(f"Invalid content hash (expected 26-char lowercase base32): {content_hash}") + prefix = hash_prefix.strip("/") + section = f"{prefix}/" if prefix else "" if subfolding: folds = _subfold(content_hash, subfolding) fold_path = "/".join(folds) - return f"_hash/{schema_name}/{fold_path}/{content_hash}" + return f"{section}{schema_name}/{fold_path}/{content_hash}" else: - return f"_hash/{schema_name}/{content_hash}" + return f"{section}{schema_name}/{content_hash}" def get_store_backend(store_name: str | None = None, config: Config | None = None) -> StorageBackend: @@ -217,8 +225,17 @@ def put_hash( Metadata dict with keys: hash, path, schema, store, size. """ content_hash = compute_hash(data) - subfolding = get_store_subfolding(store_name, config=config) - path = build_hash_path(content_hash, schema_name, subfolding) + if config is None: + from .settings import config # type: ignore[assignment] + assert config is not None + spec = config.get_store_spec(store_name) + subfolding = tuple(spec["subfolding"]) if spec.get("subfolding") else None + path = build_hash_path( + content_hash, + schema_name, + subfolding, + hash_prefix=spec["hash_prefix"], # always present: settings applies the default + ) backend = get_store_backend(store_name, config=config) diff --git a/src/datajoint/heading.py b/src/datajoint/heading.py index 8bd91ad3e..722c412f4 100644 --- a/src/datajoint/heading.py +++ b/src/datajoint/heading.py @@ -168,6 +168,41 @@ def original_name(self) -> str: assert self.attribute_expression.startswith(("`", '"')) return self.attribute_expression.strip('`"') + @property + def is_hash_object(self) -> bool: + """ + True if the value is stored in external hash-addressed storage. + + Covers ```` (external-only) and ````/```` *with* + a store. Inline ````/```` (no store) return False — their + bytes live in the column, so there is nothing in object storage to + collect. The ``store is not None`` test mirrors the framework's own + ``is_store`` determination (see ``declare``/``Heading`` compilation). + """ + if not self.codec: + return False + name = getattr(self.codec, "name", "") + if name == "hash": + return True # is external-only (get_dtype raises without @) + # / are dual-mode: external only when a store is set. + return name in ("blob", "attach") and self.store is not None + + @property + def is_schema_object(self) -> bool: + """ + True if the value is stored in schema-addressed storage. + + Covers ````, ````, and any custom ``SchemaCodec`` + subclass — recognized by codec *type*, not name, so third-party codecs + are included (see #1469). Schema-addressed storage is external-only, so + no store guard is needed. + """ + if not self.codec: + return False + from .builtin_codecs.schema import SchemaCodec # local import avoids import cycle + + return isinstance(self.codec, SchemaCodec) + class Heading: """ @@ -277,6 +312,18 @@ def non_blobs(self) -> list[str]: """Attributes that are not blobs or JSON.""" return [k for k, v in self.attributes.items() if not (v.is_blob or v.json)] + @property + def hash_objects(self) -> list[str]: + """Names of attributes stored in external hash-addressed storage (````, + external ````/````). See :attr:`Attribute.is_hash_object`.""" + return [k for k, v in self.attributes.items() if v.is_hash_object] + + @property + def schema_objects(self) -> list[str]: + """Names of attributes stored in schema-addressed storage (````, + ````, custom ``SchemaCodec``). See :attr:`Attribute.is_schema_object`.""" + return [k for k, v in self.attributes.items() if v.is_schema_object] + @property def new_attributes(self) -> list[str]: """Attributes with computed expressions (projections).""" diff --git a/src/datajoint/settings.py b/src/datajoint/settings.py index 6ae23478b..40167697e 100644 --- a/src/datajoint/settings.py +++ b/src/datajoint/settings.py @@ -571,6 +571,19 @@ def _validate_prefix_separation( """ Validate that storage section prefixes don't overlap. + These keys **control the store layout** (see + how-to/configure-storage in the documentation): ``hash_prefix`` + (default ``_hash``) is where hash-addressed objects are written and + scanned; ``schema_prefix`` (default ``_schema``) is where + schema-addressed objects are written; ``filepath_prefix`` (default + unrestricted) is the namespace ```` paths must stay in. + Writers, garbage collection, and ```` validation all + consume the same values, so the sections may be relocated per store + without the components drifting apart. Changing a prefix on a store + that already holds data is not recommended: existing objects remain + readable (their metadata stores full paths) but reclamation scans + only the currently configured sections. + Parameters ---------- store_name : str diff --git a/src/datajoint/staged_insert.py b/src/datajoint/staged_insert.py index ffbe8a8f2..b3d82ac6c 100644 --- a/src/datajoint/staged_insert.py +++ b/src/datajoint/staged_insert.py @@ -120,6 +120,7 @@ def _resolve_field(self, field: str, ext: str) -> tuple[str, "StorageBackend"]: ext=ext if ext else None, partition_pattern=partition_pattern, token_length=token_length, + schema_prefix=spec["schema_prefix"], # always present: settings applies the default ) self._staged_objects[field] = { diff --git a/src/datajoint/storage.py b/src/datajoint/storage.py index 6a8260163..39a21ce46 100644 --- a/src/datajoint/storage.py +++ b/src/datajoint/storage.py @@ -196,6 +196,8 @@ def build_object_path( ext: str | None, partition_pattern: str | None = None, token_length: int = 8, + *, + schema_prefix: str, ) -> tuple[str, str]: """ Build the storage path for an object attribute. @@ -216,6 +218,10 @@ def build_object_path( Partition pattern with ``{attr}`` placeholders. token_length : int, optional Length of random token suffix. Default 8. + schema_prefix : str + Section prefix from the store's ``schema_prefix`` setting. No fallback + by design: settings (``get_store_spec``) is the single source of the + ``"_schema"`` default, applied to every store spec. Returns ------- @@ -256,8 +262,11 @@ def build_object_path( pk_parts.append(f"{attr}={encode_pk_value(value)}") # Construct full path - # Pattern: {partition_attrs}/{schema}/{table}/{remaining_pk}/{filename} + # Pattern: {schema_prefix}/{partition_attrs}/{schema}/{table}/{remaining_pk}/{filename} parts = [] + prefix = schema_prefix.strip("/") + if prefix: + parts.append(prefix) if partition_parts: parts.extend(partition_parts) parts.append(schema) diff --git a/tests/integration/test_gc.py b/tests/integration/test_gc.py index c9ea741bd..42150d832 100644 --- a/tests/integration/test_gc.py +++ b/tests/integration/test_gc.py @@ -2,6 +2,7 @@ Tests for garbage collection (gc.py). """ +from contextlib import ExitStack from unittest.mock import MagicMock, patch import numpy as np @@ -9,9 +10,21 @@ import datajoint as dj from datajoint import gc +from datajoint.builtin_codecs.object import ObjectCodec +from datajoint.codecs import get_codec from datajoint.errors import DataJointError +class GcCustomObjectCodec(ObjectCodec): + """A custom schema-addressed codec — a ``SchemaCodec`` subclass with a + non-built-in name, mirroring the user's NetCDF codec in #1469. Registered + at import via ``Codec.__init_subclass__``. GC must recognize it by type + (not by hardcoded name) and treat its files as referenced, not orphaned. + """ + + name = "gc_custom_object" + + # Tables used by TestScanWithLiveData. Defined at module scope so dj.Schema's # context resolution can find them by class name; bound to a schema inside # each fixture (see schema(...) calls below). @@ -41,347 +54,209 @@ class GcObjectTest(dj.Manual): """ -class TestUsesHashStorage: - """Tests for _uses_hash_storage helper function.""" +class GcCustomCodecTest(dj.Manual): + definition = """ + rid : int + --- + payload : + """ + - def test_returns_false_for_no_adapter(self): - """Test that False is returned when attribute has no codec.""" - attr = MagicMock() - attr.codec = None +def _make_attr(**overrides): + """Build a real heading.Attribute with defaults, overriding named fields.""" + from datajoint.heading import Attribute, default_attribute_properties - assert gc._uses_hash_storage(attr) is False + return Attribute(**{**default_attribute_properties, **overrides}) - def test_returns_true_for_hash_type(self): - """Test that True is returned for type.""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "hash" - attr.store = "mystore" - assert gc._uses_hash_storage(attr) is True +class TestAttributeIsHashObject: + """Attribute.is_hash_object classifies external hash-addressed storage + (moved from gc._uses_hash_storage — the classification now lives on the + heading so GC and other modules share one source of truth).""" - def test_returns_true_for_blob_external(self): - """Test that True is returned for type (external).""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "blob" - attr.store = "mystore" + def test_false_for_no_codec(self): + assert _make_attr(codec=None).is_hash_object is False - assert gc._uses_hash_storage(attr) is True + def test_true_for_hash_type(self): + # is external-only; True regardless of store attribute value + assert _make_attr(codec=get_codec("hash"), store="mystore").is_hash_object is True - def test_returns_true_for_attach_external(self): - """Test that True is returned for type (external).""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "attach" - attr.store = "mystore" + def test_true_for_blob_external(self): + assert _make_attr(codec=get_codec("blob"), store="mystore").is_hash_object is True - assert gc._uses_hash_storage(attr) is True + def test_true_for_attach_external(self): + assert _make_attr(codec=get_codec("attach"), store="mystore").is_hash_object is True - def test_returns_false_for_blob_internal(self): - """Test that False is returned for internal storage.""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "blob" - attr.store = None + def test_false_for_blob_internal(self): + # inline (no store) lives in the column — not hash-addressed + assert _make_attr(codec=get_codec("blob"), store=None).is_hash_object is False - assert gc._uses_hash_storage(attr) is False + def test_false_for_schema_codec(self): + assert _make_attr(codec=get_codec("object"), store="mystore").is_hash_object is False -class TestExtractHashRefs: - """Tests for _extract_hash_refs helper function.""" +class TestHashReferencedPaths: + """Reference extraction for hash-addressed metadata via the codec API + (Codec.referenced_paths — the production path used by gc.scan).""" def test_returns_empty_for_none(self): - """Test that empty list is returned for None value.""" - assert gc._extract_hash_refs(None) == [] + assert get_codec("hash").referenced_paths(None) == [] def test_parses_json_string(self): - """Test parsing JSON string with path.""" value = '{"path": "_hash/schema/abc123", "hash": "abc123", "store": "mystore"}' - refs = gc._extract_hash_refs(value) + refs = get_codec("hash").referenced_paths(value) - assert len(refs) == 1 - assert refs[0] == ("_hash/schema/abc123", "mystore") + assert refs == [("_hash/schema/abc123", "mystore")] def test_parses_dict_directly(self): - """Test parsing dict with path.""" value = {"path": "_hash/schema/def456", "hash": "def456", "store": None} - refs = gc._extract_hash_refs(value) + refs = get_codec("hash").referenced_paths(value) - assert len(refs) == 1 - assert refs[0] == ("_hash/schema/def456", None) + assert refs == [("_hash/schema/def456", None)] def test_returns_empty_for_invalid_json(self): - """Test that empty list is returned for invalid JSON.""" - assert gc._extract_hash_refs("not json") == [] + assert get_codec("hash").referenced_paths("not json") == [] def test_returns_empty_for_dict_without_path(self): - """Test that empty list is returned for dict without path key.""" - assert gc._extract_hash_refs({"hash": "abc123"}) == [] - - -class TestUsesSchemaStorage: - """Tests for _uses_schema_storage helper function.""" + assert get_codec("hash").referenced_paths({"hash": "abc123"}) == [] - def test_returns_false_for_no_adapter(self): - """Test that False is returned when attribute has no codec.""" - attr = MagicMock() - attr.codec = None - assert gc._uses_schema_storage(attr) is False +class TestAttributeIsSchemaObject: + """Attribute.is_schema_object classifies schema-addressed storage by codec + TYPE (so custom SchemaCodec subclasses are included, #1469).""" - def test_returns_true_for_object_type(self): - """Test that True is returned for type.""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "object" + def test_false_for_no_codec(self): + assert _make_attr(codec=None).is_schema_object is False - assert gc._uses_schema_storage(attr) is True + def test_true_for_object_type(self): + assert _make_attr(codec=get_codec("object")).is_schema_object is True - def test_returns_true_for_npy_type(self): - """Test that True is returned for type.""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "npy" + def test_true_for_npy_type(self): + assert _make_attr(codec=get_codec("npy")).is_schema_object is True - assert gc._uses_schema_storage(attr) is True + def test_true_for_custom_schema_subclass(self): + # by type, not name — a custom ObjectCodec subclass counts (#1469) + assert _make_attr(codec=get_codec("gc_custom_object")).is_schema_object is True - def test_returns_false_for_other_types(self): - """Test that False is returned for non-schema-addressed types.""" - attr = MagicMock() - attr.codec = MagicMock() - attr.codec.name = "blob" + def test_false_for_hash_codec(self): + assert _make_attr(codec=get_codec("blob"), store="mystore").is_schema_object is False - assert gc._uses_schema_storage(attr) is False - -class TestExtractSchemaRefs: - """Tests for _extract_schema_refs helper function.""" +class TestSchemaReferencedPaths: + """Reference extraction for schema-addressed metadata via the codec API + (inherited by every SchemaCodec subclass, including custom codecs).""" def test_returns_empty_for_none(self): - """Test that empty list is returned for None value.""" - assert gc._extract_schema_refs(None) == [] + assert get_codec("object").referenced_paths(None) == [] def test_parses_json_string(self): - """Test parsing JSON string with path.""" value = '{"path": "schema/table/pk/field", "store": "mystore"}' - refs = gc._extract_schema_refs(value) + refs = get_codec("object").referenced_paths(value) - assert len(refs) == 1 - assert refs[0] == ("schema/table/pk/field", "mystore") + assert refs == [("schema/table/pk/field", "mystore")] def test_parses_dict_directly(self): - """Test parsing dict with path.""" - value = {"path": "test/path", "store": None} - refs = gc._extract_schema_refs(value) + refs = get_codec("object").referenced_paths({"path": "test/path", "store": None}) - assert len(refs) == 1 - assert refs[0] == ("test/path", None) + assert refs == [("test/path", None)] + + def test_custom_subclass_inherits_extraction(self): + refs = get_codec("gc_custom_object").referenced_paths({"path": "a/b/c", "store": "local"}) + + assert refs == [("a/b/c", "local")] def test_returns_empty_for_dict_without_path(self): - """Test that empty list is returned for dict without path key.""" - assert gc._extract_schema_refs({"other": "data"}) == [] + assert get_codec("object").referenced_paths({"other": "data"}) == [] + + +def _mock_collector(store="test_store"): + """A GarbageCollector wired to mocks (no real store), for logic-only tests. + The backend is resolved eagerly at construction, so patch it out.""" + cfg = MagicMock() + cfg.get_store_spec.return_value = {"hash_prefix": "_hash", "schema_prefix": "_schema"} + with patch("datajoint.gc.get_store_backend", return_value=MagicMock()): + return gc.GarbageCollector(MagicMock(), store=store, config=cfg) + + +def _gc(schema, store="local"): + """One store-bound collector per live-data test (config from the schema).""" + return gc.GarbageCollector(schema, store=store) -class TestScan: - """Tests for scan function.""" +class TestGarbageCollectorConstruction: + """The collector is store-bound and resolves its store eagerly.""" + + def test_resolves_store_at_construction(self): + cfg = MagicMock() + cfg.get_store_spec.return_value = {"hash_prefix": "_hash", "schema_prefix": "_schema"} + with patch("datajoint.gc.get_store_backend", return_value="BACKEND") as mock_backend: + collector = gc.GarbageCollector(MagicMock(), store="mystore", config=cfg) + mock_backend.assert_called_once_with("mystore", config=cfg) # eager + assert collector.store == "mystore" + assert collector.backend == "BACKEND" + assert collector._hash_prefix == "_hash" and collector._schema_prefix == "_schema" + + +class TestCollect: + """Tests for GarbageCollector.collect (dry_run=True is the read-only scan).""" + + # Fixtures patched onto the four data-gathering methods so the orphan set is + # fixed: hashes path1,path2 referenced but path1,path3 stored → path3 + # orphaned; schema pk1 referenced but pk1,pk2 stored → pk2 orphaned. + def _patch_data(self): + return [ + patch.object(gc.GarbageCollector, "hash_references", return_value={"_hash/s/path1", "_hash/s/path2"}), + patch.object(gc.GarbageCollector, "schema_references", return_value={"s/t/pk1/f"}), + patch.object(gc.GarbageCollector, "list_hash_paths", return_value={"_hash/s/path1": 100, "_hash/s/path3": 200}), + patch.object(gc.GarbageCollector, "list_schema_paths", return_value={"s/t/pk1/f": 500, "s/t/pk2/f": 300}), + ] def test_requires_at_least_one_schema(self): - """Test that at least one schema is required.""" + """At least one schema is required, checked before the store is resolved.""" with pytest.raises(DataJointError, match="At least one schema must be provided"): - gc.scan() - - @patch("datajoint.gc.scan_schema_references") - @patch("datajoint.gc.list_schema_paths") - @patch("datajoint.gc.scan_hash_references") - @patch("datajoint.gc.list_stored_hashes") - def test_returns_stats(self, mock_list_hashes, mock_scan_hash, mock_list_schemas, mock_scan_schema): - """Test that scan returns proper statistics.""" - # Mock hash-addressed storage (now uses paths) - mock_scan_hash.return_value = {"_hash/schema/path1", "_hash/schema/path2"} - mock_list_hashes.return_value = { - "_hash/schema/path1": 100, - "_hash/schema/path3": 200, # orphaned - } - - # Mock schema-addressed storage - mock_scan_schema.return_value = {"schema/table/pk1/field"} - mock_list_schemas.return_value = { - "schema/table/pk1/field": 500, - "schema/table/pk2/field": 300, # orphaned - } - - mock_schema = MagicMock() - stats = gc.scan(mock_schema, store_name="test_store") - - # Hash stats - assert stats["hash_referenced"] == 2 - assert stats["hash_stored"] == 2 - assert stats["hash_orphaned"] == 1 - assert "_hash/schema/path3" in stats["orphaned_hashes"] - - # Schema stats - assert stats["schema_paths_referenced"] == 1 - assert stats["schema_paths_stored"] == 2 - assert stats["schema_paths_orphaned"] == 1 - assert "schema/table/pk2/field" in stats["orphaned_paths"] - - # Combined totals - assert stats["referenced"] == 3 - assert stats["stored"] == 4 - assert stats["orphaned"] == 2 - assert stats["orphaned_bytes"] == 500 # 200 hash + 300 schema + gc.GarbageCollector(store="test_store") + def test_dry_run_reports_stats_and_deletes_nothing(self): + """dry_run=True (default) returns the full orphan report without deleting.""" + with ExitStack() as es: + for p in self._patch_data(): + es.enter_context(p) + mock_delete = es.enter_context(patch("datajoint.gc.delete_path")) + stats = _mock_collector().collect() # dry_run defaults True -class TestCollect: - """Tests for collect function.""" - - @patch("datajoint.gc.scan") - def test_dry_run_does_not_delete(self, mock_scan): - """Test that dry_run=True doesn't delete anything.""" - mock_scan.return_value = { - "referenced": 1, - "stored": 2, - "orphaned": 1, - "orphaned_bytes": 100, - "orphaned_hashes": ["_hash/schema/orphan_path"], - "orphaned_paths": [], - "hash_orphaned": 1, - "schema_paths_orphaned": 0, - } - - mock_schema = MagicMock() - stats = gc.collect(mock_schema, store_name="test_store", dry_run=True) - - assert stats["deleted"] == 0 - assert stats["bytes_freed"] == 0 assert stats["dry_run"] is True - - @patch("datajoint.gc.delete_path") - @patch("datajoint.gc.list_stored_hashes") - @patch("datajoint.gc.scan") - def test_deletes_orphaned_hashes(self, mock_scan, mock_list_stored, mock_delete): - """Test that orphaned hashes are deleted when dry_run=False.""" - mock_scan.return_value = { - "referenced": 1, - "stored": 2, - "orphaned": 1, - "orphaned_bytes": 100, - "orphaned_hashes": ["_hash/schema/orphan_path"], - "orphaned_paths": [], - "hash_orphaned": 1, - "schema_paths_orphaned": 0, - } - mock_list_stored.return_value = {"_hash/schema/orphan_path": 100} - mock_delete.return_value = True - - mock_schema = MagicMock() - stats = gc.collect(mock_schema, store_name="test_store", dry_run=False) - - assert stats["deleted"] == 1 - assert stats["hash_deleted"] == 1 - assert stats["bytes_freed"] == 100 + assert stats["deleted"] == 0 and stats["bytes_freed"] == 0 + mock_delete.assert_not_called() + # full report is present + assert stats["hash_paths_orphaned"] == 1 and stats["orphaned_hash_paths"] == ["_hash/s/path3"] + assert stats["hash_paths_orphaned_bytes"] == 200 + assert stats["schema_paths_orphaned"] == 1 and stats["orphaned_schema_paths"] == ["s/t/pk2/f"] + assert stats["schema_paths_orphaned_bytes"] == 300 + # no combined totals + for k in ("referenced", "stored", "orphaned", "orphaned_bytes"): + assert k not in stats + + def test_delete_removes_orphans_via_bound_store(self): + """dry_run=False deletes each section's orphans, sized from the same scan.""" + with ExitStack() as es: + for p in self._patch_data(): + es.enter_context(p) + mock_delete_hash = es.enter_context(patch("datajoint.gc.delete_path", return_value=True)) + mock_delete_schema = es.enter_context(patch.object(gc.GarbageCollector, "delete_schema_path", return_value=True)) + collector = _mock_collector() + stats = collector.collect(dry_run=False) + + assert stats["hash_paths_deleted"] == 1 and stats["schema_paths_deleted"] == 1 + assert stats["deleted"] == 2 + assert stats["bytes_freed"] == 500 # 200 (hash path3) + 300 (schema pk2) assert stats["dry_run"] is False - mock_delete.assert_called_once_with("_hash/schema/orphan_path", "test_store", config=mock_schema.connection._config) - - @patch("datajoint.gc.delete_schema_path") - @patch("datajoint.gc.list_schema_paths") - @patch("datajoint.gc.scan") - def test_deletes_orphaned_schemas(self, mock_scan, mock_list_schemas, mock_delete): - """Test that orphaned schema paths are deleted when dry_run=False.""" - mock_scan.return_value = { - "referenced": 1, - "stored": 2, - "orphaned": 1, - "orphaned_bytes": 500, - "orphaned_hashes": [], - "orphaned_paths": ["schema/table/pk/field"], - "hash_orphaned": 0, - "schema_paths_orphaned": 1, - } - mock_list_schemas.return_value = {"schema/table/pk/field": 500} - mock_delete.return_value = True - - mock_schema = MagicMock() - stats = gc.collect(mock_schema, store_name="test_store", dry_run=False) - - assert stats["deleted"] == 1 - assert stats["schema_paths_deleted"] == 1 - assert stats["bytes_freed"] == 500 - assert stats["dry_run"] is False - mock_delete.assert_called_once_with("schema/table/pk/field", "test_store", config=mock_schema.connection._config) - - -class TestFormatStats: - """Tests for format_stats function.""" - - def test_formats_scan_stats(self): - """Test formatting scan statistics.""" - stats = { - "referenced": 10, - "stored": 15, - "orphaned": 5, - "orphaned_bytes": 1024 * 1024, # 1 MB - "hash_referenced": 6, - "hash_stored": 8, - "hash_orphaned": 2, - "hash_orphaned_bytes": 512 * 1024, - "schema_paths_referenced": 4, - "schema_paths_stored": 7, - "schema_paths_orphaned": 3, - "schema_paths_orphaned_bytes": 512 * 1024, - } - - result = gc.format_stats(stats) - - assert "Referenced in database: 10" in result - assert "Stored in backend: 15" in result - assert "Orphaned (unreferenced): 5" in result - assert "1.00 MB" in result - # Check for detailed sections - assert "Hash-Addressed Storage" in result - assert "Schema-Addressed Storage" in result - - def test_formats_collect_stats_dry_run(self): - """Test formatting collect statistics with dry_run.""" - stats = { - "referenced": 10, - "stored": 15, - "orphaned": 5, - "deleted": 0, - "bytes_freed": 0, - "dry_run": True, - } - - result = gc.format_stats(stats) - - assert "DRY RUN" in result - - def test_formats_collect_stats_actual(self): - """Test formatting collect statistics after actual deletion.""" - stats = { - "referenced": 10, - "stored": 15, - "orphaned": 5, - "deleted": 3, - "hash_deleted": 2, - "schema_paths_deleted": 1, - "bytes_freed": 2 * 1024 * 1024, # 2 MB - "errors": 2, - "dry_run": False, - } - - result = gc.format_stats(stats) - - assert "Deleted: 3" in result - assert "Hash items: 2" in result - assert "Schema paths: 1" in result - assert "2.00 MB" in result - assert "Errors: 2" in result + # hash deleted via the collector's own store/config (not threaded params) + mock_delete_hash.assert_called_once_with("_hash/s/path3", collector.store, config=collector.config) + mock_delete_schema.assert_called_once_with("s/t/pk2/f") class TestScanWithLiveData: - """End-to-end tests for gc.scan() against real schemas with external storage. + """End-to-end tests for GarbageCollector.scan() against real schemas with external storage. Exercises the full production path: scan_*_references → table.proj(attr).cursor() → raw JSON metadata. @@ -428,28 +303,30 @@ def schema_object(self, connection_test, prefix, mock_stores): schema.drop() def test_scan_finds_active_blob_reference(self, schema_blob): - """scan() must report hash_referenced >= 1 for a populated column. + """scan() must report hash_paths_referenced >= 1 for a populated column. Decoded value type returned by BlobCodec.decode is numpy.ndarray, which - does not satisfy `_extract_hash_refs`'s dict/JSON-string check — this + does not satisfy the raw-metadata dict/JSON-string check in referenced_paths — this test fails before the cursor-based fix in scan_hash_references. """ GcBlobTest.insert1({"rid": 1, "payload": np.arange(64, dtype="uint8")}) - stats = gc.scan(schema_blob, store_name="local") + collector = _gc(schema_blob) + stats = collector.collect() - assert stats["hash_referenced"] >= 1, f"scan should find the active reference; got {stats}" + assert stats["hash_paths_referenced"] >= 1, f"scan should find the active reference; got {stats}" def test_scan_finds_active_npy_reference(self, schema_npy): """scan() must report schema_paths_referenced >= 1 for a populated column. Decoded value type returned by NpyCodec.decode is NpyRef (lazy handle), - which does not satisfy `_extract_schema_refs`'s dict check — this test + which does not satisfy the raw-metadata dict check in referenced_paths — this test fails before the cursor-based fix in scan_schema_references. """ GcNpyTest.insert1({"rid": 1, "waveform": np.arange(64, dtype="float32")}) - stats = gc.scan(schema_npy, store_name="local") + collector = _gc(schema_npy) + stats = collector.collect() assert stats["schema_paths_referenced"] >= 1, f"scan should find the active reference; got {stats}" @@ -457,11 +334,431 @@ def test_scan_finds_active_object_reference(self, schema_object): """scan() must report schema_paths_referenced >= 1 for a populated column. Decoded value type returned by ObjectCodec.decode is ObjectRef (lazy - handle), which does not satisfy `_extract_schema_refs`'s dict check — + handle), which does not satisfy the raw-metadata dict check in referenced_paths — this test fails before the cursor-based fix in scan_schema_references. """ GcObjectTest.insert1({"rid": 1, "results": b"hello-gc-test"}) - stats = gc.scan(schema_object, store_name="local") + collector = _gc(schema_object) + stats = collector.collect() assert stats["schema_paths_referenced"] >= 1, f"scan should find the active reference; got {stats}" + + @pytest.fixture + def schema_custom(self, connection_test, prefix, mock_stores): + schema_name = f"{prefix}_test_gc_e2e_custom" + schema = dj.Schema( + schema_name, + context={"GcCustomCodecTest": GcCustomCodecTest}, + connection=connection_test, + ) + schema(GcCustomCodecTest) + yield schema + schema.drop() + + def test_custom_codec_reference_not_orphaned(self, schema_custom): + """#1469: a live custom SchemaCodec value must be recognized as + referenced and its file path must NOT be flagged as an orphan. Before + the fix, schema-addressed classification keyed on the hardcoded names + object/npy, so this codec was never scanned and its live file was + reported orphaned. Now Attribute.is_schema_object dispatches on type. + + Asserts on the specific live path (not global counts) so it is robust to + other tests sharing the same ``local`` store. + """ + GcCustomCodecTest.insert1({"rid": 1, "payload": b"live-payload"}) + + collector = _gc(schema_custom) + refs = collector.schema_references() + assert refs, "custom codec's live reference must be discovered (#1469)" + live_path = next(iter(refs)) + + stats = collector.collect() + assert live_path not in stats["orphaned_schema_paths"], f"live custom-codec file wrongly flagged orphan: {live_path}" + + def test_custom_codec_survives_collect(self, schema_custom): + """#1469 end-to-end data-loss guard: collect() must delete only the + deleted row's file and keep the surviving row's file (checked by exact + path, robust to a shared store).""" + GcCustomCodecTest.insert1({"rid": 1, "payload": b"row-one"}) + GcCustomCodecTest.insert1({"rid": 2, "payload": b"row-two"}) + + collector = _gc(schema_custom) + refs_all = collector.schema_references() + assert len(refs_all) == 2, f"both live rows should be referenced; got {refs_all}" + + # Delete one row — its file becomes a genuine orphan (delete-then-GC). + (GcCustomCodecTest & {"rid": 1}).delete(prompt=False) + + refs_live = collector.schema_references() + assert len(refs_live) == 1, f"one live row should remain referenced; got {refs_live}" + live_path = next(iter(refs_live)) + deleted_paths = refs_all - refs_live + + collector.collect(dry_run=False) + + stored_after = set(collector.list_schema_paths(schema_custom.database)) + assert live_path in stored_after, f"collect() deleted the live custom-codec file {live_path} (#1469)" + assert not (deleted_paths & stored_after), f"deleted row's orphan not reclaimed: {deleted_paths & stored_after}" + + +class TestDirectoryObjects: + """Directory-valued objects (e.g. Zarr stores) — the referenced metadata + path is a directory PREFIX whose stored form is many chunk files plus a + `{path}.manifest.json` sidecar. Orphan matching must be coverage-based + (exact / ancestor-prefix / manifest), not exact set difference: with exact + matching, every chunk of a LIVE store is misclassified as an orphan and + collect() deletes live data.""" + + @pytest.fixture + def schema_dirobj(self, connection_test, prefix, mock_stores): + schema = dj.Schema( + f"{prefix}_test_gc_dirobj", + context={"GcObjectTest": GcObjectTest}, + connection=connection_test, + ) + schema(GcObjectTest) + yield schema + schema.drop() + + @staticmethod + def _make_store_dir(tmp_path, name, n_chunks=3): + d = tmp_path / name + d.mkdir() + (d / ".zarray").write_bytes(b"{}") + for i in range(n_chunks): + (d / f"0.{i}").write_bytes(f"chunk{i}".encode()) + return d + + def test_live_directory_object_not_orphaned(self, schema_dirobj, tmp_path): + """A live folder object's chunk files and manifest must be covered by + its referenced prefix — none may appear in orphaned_schema_paths.""" + GcObjectTest.insert1({"rid": 1, "results": str(self._make_store_dir(tmp_path, "live.zarr"))}) + + collector = _gc(schema_dirobj) + refs = collector.schema_references() + assert len(refs) == 1 + ref = next(iter(refs)) + + stored = collector.list_schema_paths(schema_dirobj.database) + chunk_files = [p for p in stored if p.startswith(ref + "/")] + assert len(chunk_files) >= 4, f"expected the folder's files under {ref}; got {sorted(stored)}" + assert f"{ref}.manifest.json" in stored, "manifest sidecar must be listed (it is reclaimable state)" + + stats = collector.collect() + flagged = [ + p for p in stats["orphaned_schema_paths"] if p == ref or p.startswith(ref + "/") or p == f"{ref}.manifest.json" + ] + assert not flagged, f"live directory object misclassified as orphaned: {flagged}" + + def test_orphaned_directory_object_fully_reclaimed(self, schema_dirobj, tmp_path): + """After deleting one row, collect() must reclaim that folder object's + chunks AND manifest while leaving the surviving row's folder intact.""" + GcObjectTest.insert1({"rid": 1, "results": str(self._make_store_dir(tmp_path, "gone.zarr"))}) + GcObjectTest.insert1({"rid": 2, "results": str(self._make_store_dir(tmp_path, "kept.zarr"))}) + + collector = _gc(schema_dirobj) + refs_all = collector.schema_references() + assert len(refs_all) == 2 + + (GcObjectTest & {"rid": 1}).delete(prompt=False) + refs_live = collector.schema_references() + assert len(refs_live) == 1 + live_ref = next(iter(refs_live)) + dead_ref = next(iter(refs_all - refs_live)) + + collector.collect(dry_run=False) + + stored_after = set(collector.list_schema_paths(schema_dirobj.database)) + live_files = {p for p in stored_after if p.startswith(live_ref + "/")} + assert len(live_files) >= 4, f"live folder object lost files: {sorted(stored_after)}" + assert f"{live_ref}.manifest.json" in stored_after, "live object's manifest must survive collect()" + + dead_files = {p for p in stored_after if p == dead_ref or p.startswith(dead_ref + "/")} + assert not dead_files, f"orphaned folder object not reclaimed: {sorted(dead_files)}" + assert f"{dead_ref}.manifest.json" not in stored_after, "orphaned object's manifest must be reclaimed" + + +class GcProbeHash(dj.Manual): + """Table whose name (`gc_probe_hash`) contains the substring `_hash` — + regression guard for the schema-path walk's hash-subtree skip.""" + + definition = """ + rid : int + --- + results : + """ + + +class TestHashSubstringTableName: + """A table whose name contains the substring `_hash` (e.g. `gc_probe_hash`) + is listed and reclaimed normally. Per-schema scoping walks only the + schema's own section (`{schema_prefix}/{schema}/`), so such a table's + objects are ordinary schema-addressed files — there is no store-wide + hash-subtree skip that a substring could trip over.""" + + @pytest.fixture + def schema_probe(self, connection_test, prefix, mock_stores): + schema = dj.Schema( + f"{prefix}_test_gc_hashname", + context={"GcProbeHash": GcProbeHash}, + connection=connection_test, + ) + schema(GcProbeHash) + yield schema + schema.drop() + + def test_hash_substring_table_listed_and_reclaimed(self, schema_probe): + GcProbeHash.insert1({"rid": 1, "results": b"keep"}) + GcProbeHash.insert1({"rid": 2, "results": b"drop"}) + + collector = _gc(schema_probe) + refs = collector.schema_references() + assert len(refs) == 2 + stored = set(collector.list_schema_paths(schema_probe.database)) + missing = refs - stored + assert not missing, ( + f"files of a table whose name contains '_hash' must be listed; missing {missing} " + "(substring-based skip would hide them from GC forever)" + ) + + (GcProbeHash & {"rid": 2}).delete(prompt=False) + refs_live = collector.schema_references() + dead_ref = next(iter(refs - refs_live)) + + collector.collect(dry_run=False) + stored_after = set(collector.list_schema_paths(schema_probe.database)) + assert dead_ref not in stored_after, "orphan under a '_hash'-substring table name must be reclaimed" + assert next(iter(refs_live)) in stored_after, "live file must survive" + + +class TestUserFilesOutsideSchemaSectionUntouched: + """Per-schema scoping walks only `{schema_prefix}/{schema}/`, so any file + outside that section — including user-managed content anywhere + else in the store — is structurally never a GC candidate. The old + cohabitation hazard (user files misread as orphans) is gone by + construction, with or without a declared filepath_prefix.""" + + @pytest.fixture + def schema_fp(self, connection_test, prefix, mock_stores): + schema = dj.Schema( + f"{prefix}_test_gc_userfiles", + context={"GcObjectTest": GcObjectTest}, + connection=connection_test, + ) + schema(GcObjectTest) + yield schema + schema.drop() + + def test_user_file_outside_section_never_listed_or_collected(self, schema_fp): + from pathlib import Path + + cfg = schema_fp.connection._config + location = Path(cfg.get_store_spec("local")["location"]) + user_file = location / "userfiles" / "deep" / "important.bin" + user_file.parent.mkdir(parents=True, exist_ok=True) + user_file.write_bytes(b"user-managed") + + GcObjectTest.insert1({"rid": 1, "results": b"managed-object"}) + + # The user file lives outside {schema_prefix}/{schema}/, so the scoped + # walk never lists it — no filepath_prefix declaration needed. + collector = _gc(schema_fp) + stored = collector.list_schema_paths(schema_fp.database) + assert "userfiles/deep/important.bin" not in stored + assert stored, "the schema's own managed object must still be listed" + + collector.collect(dry_run=False) + assert user_file.exists(), "collect() must never touch files outside the schema section" + + +class TestPrefixSettingsHonored: + """The per-store hash_prefix/schema_prefix settings control the layout + (docs: how-to/configure-storage): writers place objects under the + configured sections and GC scans the same sections — relocation without + drift between components.""" + + @pytest.fixture + def schema_prefixed(self, connection_test, prefix, mock_stores): + cfg = connection_test._config + cfg.stores["local"]["hash_prefix"] = "content" + cfg.stores["local"]["schema_prefix"] = "objects" + schema = dj.Schema( + f"{prefix}_test_gc_prefixes", + context={"GcBlobTest": GcBlobTest, "GcObjectTest": GcObjectTest}, + connection=connection_test, + ) + schema(GcBlobTest) + schema(GcObjectTest) + yield schema + schema.drop() + cfg.stores["local"].pop("hash_prefix", None) + cfg.stores["local"].pop("schema_prefix", None) + + def test_custom_sections_written_scanned_and_collected(self, schema_prefixed): + GcBlobTest.insert1({"rid": 1, "payload": np.arange(16, dtype="uint8")}) + GcObjectTest.insert1({"rid": 1, "results": b"payload-one"}) + GcObjectTest.insert1({"rid": 2, "results": b"payload-two"}) + + # Writers honored the configured sections (metadata records the real paths) + collector = _gc(schema_prefixed) + hash_refs = collector.hash_references() + assert hash_refs and all(r.startswith("content/") for r in hash_refs), hash_refs + obj_refs = collector.schema_references() + assert len(obj_refs) == 2 and all(r.startswith("objects/") for r in obj_refs), obj_refs + + # GC scans the same sections: everything referenced, nothing falsely orphaned + stats = collector.collect() + assert stats["hash_paths_referenced"] >= 1 + assert not (hash_refs & set(stats["orphaned_hash_paths"])) + assert not (obj_refs & set(stats["orphaned_schema_paths"])) + + # And reclamation works within the configured sections + (GcObjectTest & {"rid": 2}).delete(prompt=False) + collector.collect(dry_run=False) + stored_after = set(collector.list_schema_paths(schema_prefixed.database)) + live = collector.schema_references() + assert live <= stored_after, "live object under custom section must survive" + assert not ((obj_refs - live) & stored_after), "orphan under custom section must be reclaimed" + + +class TestHashObjectsOutsideCurrentSection: + """A live hash object is protected from GC even if it lives OUTSIDE the + currently configured hash section — e.g. after `hash_prefix` is changed on + a populated store. Reads always work (metadata records the full path); GC + must not misclassify such objects as schema-addressed orphans and delete + live data. Regression for the prefix-change corner.""" + + @pytest.fixture + def schema_pfx(self, connection_test, prefix, mock_stores): + schema = dj.Schema( + f"{prefix}_test_gc_hashmove", + context={"GcBlobTest": GcBlobTest, "GcObjectTest": GcObjectTest}, + connection=connection_test, + ) + schema(GcBlobTest) + schema(GcObjectTest) + yield schema + schema.drop() + + def test_live_hash_object_survives_prefix_change(self, schema_pfx): + cfg = schema_pfx.connection._config + # Written under the default _hash/ section. + GcBlobTest.insert1({"rid": 1, "payload": np.arange(8, dtype="uint8")}) + GcObjectTest.insert1({"rid": 1, "results": b"schema-obj"}) + + # Relocate the hash section: new writes would go to content/, but the + # existing object stays at _hash/ (metadata keeps its full path). + cfg.stores["local"]["hash_prefix"] = "content" + try: + # Construct AFTER the flip so the collector pins the current + # (content) prefix. The reference is the metadata path, recorded + # when the object was written under _hash — unaffected by the flip. + collector = _gc(schema_pfx) + hash_ref = next(iter(collector.hash_references())) + assert hash_ref.startswith("_hash/") + + # Read still works via the metadata path. + assert (GcBlobTest & {"rid": 1}).fetch1("payload") is not None + + stats = collector.collect() + assert hash_ref not in set( + stats["orphaned_schema_paths"] + ), "live hash object outside the current hash section must not be a schema orphan" + assert hash_ref not in set(stats["orphaned_hash_paths"]) + + # End-to-end: collect must not destroy the live hash object. + collector.collect(dry_run=False) + from datajoint.hash_registry import get_store_backend + + backend = get_store_backend("local", config=cfg) + assert backend.exists(hash_ref), "collect() deleted a live hash object after prefix change" + assert (GcBlobTest & {"rid": 1}).fetch1("payload") is not None + finally: + cfg.stores["local"].pop("hash_prefix", None) + + +class TestPerSchemaScoping: + """GC operates per schema: because every object path embeds the schema and + hash dedup is per-schema, scanning/collecting one schema is confined to + that schema's subfolder and can never see or delete another schema's + objects — even when both share one store. Removes the old 'pass every + schema or lose data' hazard.""" + + @pytest.fixture + def two_schemas(self, connection_test, prefix, mock_stores): + # Unique names per test: the shared `local` store persists objects + # across tests, but per-schema scoping means leftovers under other + # schema names are invisible here — so unique names give clean counts. + import time + + uid = str(int(time.time() * 1000))[-9:] + a = dj.Schema( + f"{prefix}_gcsc_a{uid}", + context={"GcBlobTest": GcBlobTest, "GcObjectTest": GcObjectTest}, + connection=connection_test, + ) + b = dj.Schema( + f"{prefix}_gcsc_b{uid}", + connection=connection_test, + ) + a(GcBlobTest) + a(GcObjectTest) + # Distinct classes bound to schema b (same definitions, second schema). + b_blob = type("GcBlobTest", (dj.Manual,), {"definition": GcBlobTest.definition}) + b_obj = type("GcObjectTest", (dj.Manual,), {"definition": GcObjectTest.definition}) + b(b_blob) + b(b_obj) + yield a, b, b_blob, b_obj + b.drop() + a.drop() + + def test_list_scoped_to_schema(self, two_schemas): + a, b, b_blob, b_obj = two_schemas + collector = _gc(a) + GcBlobTest.insert1({"rid": 1, "payload": np.arange(4, dtype="uint8")}) + GcObjectTest.insert1({"rid": 1, "results": b"a-obj"}) + b_blob.insert1({"rid": 1, "payload": np.arange(4, dtype="uint8")}) + b_obj.insert1({"rid": 1, "results": b"b-obj"}) + + a_hashes = set(collector.list_hash_paths(a.database)) + a_paths = set(collector.list_schema_paths(a.database)) + assert a_hashes and all(f"/{a.database}/" in h for h in a_hashes), a_hashes + assert a_paths and all(p.split("/")[0] == a.database or f"/{a.database}/" in p for p in a_paths), a_paths + # nothing from schema b leaked into schema a's scoped listings + assert not any(b.database in p for p in a_hashes | a_paths) + + def test_scan_one_schema_ignores_the_other(self, two_schemas): + a, b, b_blob, b_obj = two_schemas + collector = _gc(a) + GcBlobTest.insert1({"rid": 1, "payload": np.arange(4, dtype="uint8")}) + b_blob.insert1({"rid": 1, "payload": np.arange(4, dtype="uint8")}) + b_obj.insert1({"rid": 1, "results": b"b-obj"}) + + stats = collector.collect() + # b's live objects are outside a's subtree → not counted, not orphaned + assert stats["hash_paths_orphaned"] == 0 and stats["schema_paths_orphaned"] == 0 + assert not any(b.database in p for p in stats["orphaned_schema_paths"] + stats["orphaned_hash_paths"]) + + def test_collect_one_schema_never_touches_the_other(self, two_schemas): + a, b, b_blob, b_obj = two_schemas + collector = _gc(a) + # a has an orphan (insert then delete); b has only live data + GcObjectTest.insert1({"rid": 1, "results": b"a-doomed"}) + b_blob.insert1({"rid": 1, "payload": np.arange(4, dtype="uint8")}) + b_obj.insert1({"rid": 1, "results": b"b-live"}) + + a_orphan = next(iter(collector.schema_references())) + (GcObjectTest & {"rid": 1}).delete(prompt=False) + + b_hashes_before = set(collector.list_hash_paths(b.database)) + b_paths_before = set(collector.list_schema_paths(b.database)) + assert b_hashes_before and b_paths_before + + collector.collect(dry_run=False) + + # a's orphan reclaimed; b entirely intact + assert a_orphan not in set(collector.list_schema_paths(a.database)) + assert set(collector.list_hash_paths(b.database)) == b_hashes_before + assert set(collector.list_schema_paths(b.database)) == b_paths_before + assert len(b_obj()) == 1 # b's row (and its object) untouched diff --git a/tests/integration/test_hash_storage.py b/tests/integration/test_hash_storage.py index bc1c61a4d..42283493b 100644 --- a/tests/integration/test_hash_storage.py +++ b/tests/integration/test_hash_storage.py @@ -58,33 +58,33 @@ class TestBuildHashPath: def test_builds_flat_path(self): """Test that path is built as _hash/{schema}/{hash}.""" test_hash = "abcdefghijklmnopqrstuvwxyz"[:26] # 26 char base32 - result = build_hash_path(test_hash, "my_schema") + result = build_hash_path(test_hash, "my_schema", hash_prefix="_hash") assert result == f"_hash/my_schema/{test_hash}" def test_builds_subfolded_path(self): """Test path with subfolding.""" test_hash = "abcdefghijklmnopqrstuvwxyz"[:26] - result = build_hash_path(test_hash, "my_schema", subfolding=(2, 2)) + result = build_hash_path(test_hash, "my_schema", subfolding=(2, 2), hash_prefix="_hash") assert result == f"_hash/my_schema/ab/cd/{test_hash}" def test_rejects_invalid_hash(self): """Test that invalid hash raises error.""" with pytest.raises(DataJointError, match="Invalid content hash"): - build_hash_path("not-a-hash", "my_schema") + build_hash_path("not-a-hash", "my_schema", hash_prefix="_hash") with pytest.raises(DataJointError, match="Invalid content hash"): - build_hash_path("a" * 64, "my_schema") # Too long + build_hash_path("a" * 64, "my_schema", hash_prefix="_hash") # Too long with pytest.raises(DataJointError, match="Invalid content hash"): - build_hash_path("ABCDEFGHIJKLMNOPQRSTUVWXYZ"[:26], "my_schema") # Uppercase + build_hash_path("ABCDEFGHIJKLMNOPQRSTUVWXYZ"[:26], "my_schema", hash_prefix="_hash") # Uppercase def test_real_hash_path(self): """Test path building with a real computed hash.""" data = b"test content" content_hash = compute_hash(data) - path = build_hash_path(content_hash, "test_schema") + path = build_hash_path(content_hash, "test_schema", hash_prefix="_hash") # Verify structure: _hash/{schema}/{hash} parts = path.split("/") @@ -98,17 +98,24 @@ def test_real_hash_path(self): class TestPutHash: """Tests for put_hash function.""" - @patch("datajoint.hash_registry.get_store_subfolding") @patch("datajoint.hash_registry.get_store_backend") - def test_stores_new_content(self, mock_get_backend, mock_get_subfolding): - """Test storing new content.""" + def test_stores_new_content(self, mock_get_backend): + """Test storing new content (spec-driven: put_hash reads the store + spec for subfolding and hash_prefix).""" + import datajoint as dj + mock_backend = MagicMock() mock_backend.exists.return_value = False mock_get_backend.return_value = mock_backend - mock_get_subfolding.return_value = None - data = b"new content" - result = put_hash(data, schema_name="test_schema", store_name="test_store") + original_stores = dj.config.stores.copy() + dj.config.stores["test_store"] = {"protocol": "file", "location": "/tmp/test_hash_store"} + try: + data = b"new content" + result = put_hash(data, schema_name="test_schema", store_name="test_store") + finally: + dj.config.stores.clear() + dj.config.stores.update(original_stores) # Verify return value includes hash and path assert "hash" in result @@ -122,17 +129,23 @@ def test_stores_new_content(self, mock_get_backend, mock_get_subfolding): # Verify backend was called mock_backend.put_buffer.assert_called_once() - @patch("datajoint.hash_registry.get_store_subfolding") @patch("datajoint.hash_registry.get_store_backend") - def test_deduplicates_existing_content(self, mock_get_backend, mock_get_subfolding): + def test_deduplicates_existing_content(self, mock_get_backend): """Test that existing content is not re-uploaded.""" + import datajoint as dj + mock_backend = MagicMock() mock_backend.exists.return_value = True # Content already exists mock_get_backend.return_value = mock_backend - mock_get_subfolding.return_value = None - data = b"existing content" - result = put_hash(data, schema_name="test_schema", store_name="test_store") + original_stores = dj.config.stores.copy() + dj.config.stores["test_store"] = {"protocol": "file", "location": "/tmp/test_hash_store"} + try: + data = b"existing content" + result = put_hash(data, schema_name="test_schema", store_name="test_store") + finally: + dj.config.stores.clear() + dj.config.stores.update(original_stores) # Verify return value is still correct assert result["hash"] == compute_hash(data) diff --git a/tests/integration/test_object.py b/tests/integration/test_object.py index c03540b58..ec40f40b4 100644 --- a/tests/integration/test_object.py +++ b/tests/integration/test_object.py @@ -80,6 +80,7 @@ def test_build_object_path_basic(self): field="data_file", primary_key={"id": 123}, ext=".dat", + schema_prefix="_schema", ) assert "myschema" in path assert "MyTable" in path @@ -96,6 +97,7 @@ def test_build_object_path_no_extension(self): field="data_folder", primary_key={"id": 456}, ext=None, + schema_prefix="_schema", ) assert not path.endswith(".") assert "data_folder_" in path @@ -108,6 +110,7 @@ def test_build_object_path_multiple_pk(self): field="raw_data", primary_key={"subject_id": 1, "session_id": 2}, ext=".zarr", + schema_prefix="_schema", ) assert "subject_id=1" in path assert "session_id=2" in path @@ -121,9 +124,11 @@ def test_build_object_path_with_partition(self): primary_key={"subject_id": 1, "session_id": 2}, ext=".dat", partition_pattern="{subject_id}", + schema_prefix="_schema", ) # subject_id should be at the beginning due to partition - assert path.startswith("subject_id=1") + # section prefix first, then partition attrs (schema_prefix default "_schema") + assert path.startswith("_schema/subject_id=1") class TestObjectRef: diff --git a/tests/unit/test_codecs.py b/tests/unit/test_codecs.py index 56445419d..764ed2d31 100644 --- a/tests/unit/test_codecs.py +++ b/tests/unit/test_codecs.py @@ -484,6 +484,44 @@ def test_filepath_rejects_hash_section(self): dj.config.stores.clear() dj.config.stores.update(original_stores) + def test_filepath_reservation_follows_hash_prefix_override(self): + """hash_prefix controls the layout: overriding it moves the reserved + section along with the writer (both consume the same setting), so the + override is reserved and the former default is released.""" + from unittest.mock import MagicMock, patch + + import datajoint as dj + + filepath_codec = get_codec("filepath") + + original_stores = dj.config.stores.copy() + try: + dj.config.stores["test_store"] = { + "protocol": "file", + "location": "/tmp/test", + "hash_prefix": "content", # override; hash objects still land in _hash/ + } + + with patch("datajoint.hash_registry.get_store_backend") as mock_get_backend: + mock_backend = MagicMock() + mock_backend.exists.return_value = True + mock_get_backend.return_value = mock_backend + + # the overridden section is reserved + with pytest.raises( + ValueError, + match=r" cannot use reserved section 'content'", + ): + filepath_codec.encode("content/file.dat", store_name="test_store") + + # the former default is released — hash objects now live under + # the override, so `_hash/` is ordinary user namespace here + result = filepath_codec.encode("_hash/schema/file.dat", store_name="test_store") + assert result["path"] == "_hash/schema/file.dat" + finally: + dj.config.stores.clear() + dj.config.stores.update(original_stores) + def test_filepath_rejects_schema_section(self): """Test that filepath rejects paths starting with default schema prefix.""" from unittest.mock import MagicMock, patch