From 8a9c4b461ceece498df4e9c6dcd6b61f5b204fa2 Mon Sep 17 00:00:00 2001 From: Conrad Date: Tue, 27 Oct 2026 10:42:52 -0400 Subject: [PATCH 1/4] build: Gate the DuckDB dump dependency behind a duckdb extra The dump is an offline batch tool most deployments never run, so its only dependency stays out of the API image: pip install cfdb[duckdb]. The dev extra carries it so the test suite can exercise the loader, and package-data ships the dump schema DDL inside the wheel, which src-layout auto-discovery would otherwise omit. --- pyproject.toml | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 9ef7e15..eb18a77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,7 +58,12 @@ readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.12" [project.optional-dependencies] -dev = ["allpairspy", "cloudpickle", "cryptography", "debugpy", "httpx", "hypothesis", "mongomock-motor", "moto[ecs,s3]", "pytest-asyncio", "pytest-mock", "ruff"] +dev = ["allpairspy", "cloudpickle", "cryptography", "debugpy", "duckdb", "httpx", "hypothesis", "mongomock-motor", "moto[ecs,s3]", "pytest-asyncio", "pytest-mock", "ruff"] +# The DuckDB dump (issue #123) is an offline batch tool most deployments +# never run, so its only dependency stays behind an extra rather than +# growing the API image: +# ``pip install "cfdb[duckdb] @ git+https://github.com/abdenlab/cfdb.git"``. +duckdb = ["duckdb"] [tool.uv.sources] # Pinned to a sha, not a branch: clodius considers ``tiles_v2`` unstable @@ -101,6 +106,12 @@ markers = [ "integration: slow or tool-dependent tests (e.g. samtools, htslib, real wool WorkerPool)", ] +# Ship the dump schema DDL inside the wheel — src-layout auto-discovery +# includes only ``*.py`` without this, and ``cfdb.dump`` reads +# ``schema.sql`` via importlib.resources at run time. +[tool.setuptools.package-data] +"cfdb.dump" = ["*.sql"] + [tool.setuptools_scm] local_scheme = "dirty-tag" version_scheme = "release-branch-semver" From ad7fb318f5ed93be3118b1f0cbe92848d28675a6 Mon Sep 17 00:00:00 2001 From: Conrad Date: Tue, 27 Oct 2026 10:43:03 -0400 Subject: [PATCH 2/4] feat: Dump the files collection to a flattened relational DuckDB file Add cfdb dump duckdb, which streams the materialized files collection into a single DuckDB database whose schema of record is the checked-in src/cfdb/dump/schema.sql: C2M2-mirroring entity and junction tables on natural (id_namespace, local_id) keys, promoted cross-consortium common columns, per-consortium extension tables carrying only the unpromoted residue, three convenience views, and provenance tables (dump_info, dump_stat, dump_conflict) that make every dump self-describing. The loader validates each document through FileMetadataModel, exactly as the GraphQL resolvers hydrate the same documents, so the model's read validators apply identically on both paths. Entity rows repeated across documents are first-write-wins with divergence counted and sampled into dump_conflict; junction rows take the union over all copies so a relationship carried by only one copy is never dropped. A key resolving to two submissions aborts the dump, since the global natural keys rest on id_namespaces being DCC-scoped. High-cardinality rows stream through constraint-free TEMP staging tables so memory is bounded by the entity registries rather than the file count, and foreign keys are declared up front because DuckDB has no ALTER TABLE ADD CONSTRAINT, with finalize inserting parents before children. --- src/cfdb/cli.py | 153 +++++- src/cfdb/dump/__init__.py | 14 + src/cfdb/dump/loader.py | 1071 +++++++++++++++++++++++++++++++++++++ src/cfdb/dump/schema.sql | 576 ++++++++++++++++++++ 4 files changed, 1813 insertions(+), 1 deletion(-) create mode 100644 src/cfdb/dump/__init__.py create mode 100644 src/cfdb/dump/loader.py create mode 100644 src/cfdb/dump/schema.sql diff --git a/src/cfdb/cli.py b/src/cfdb/cli.py index 9de9f73..282bfff 100644 --- a/src/cfdb/cli.py +++ b/src/cfdb/cli.py @@ -4,6 +4,8 @@ import requests from pymongo import MongoClient +from cfdb.dcc_registry import get_all_dcc_names + __client__ = None logging.basicConfig(level=logging.INFO) @@ -23,7 +25,7 @@ def debug(ctx, param, value): import debugpy - debugpy.listen(5678) + debugpy.listen(value) logging.debug("Waiting for debugger to attach...") debugpy.wait_for_client() logging.debug("Debugger attached") @@ -110,5 +112,154 @@ def sync(dcc_names: tuple[str, ...], api_url: str, api_key: str): raise SystemExit(1) +@cli.group("dump") +def dump(): + """Export cfdb data to offline analytic formats.""" + + +@dump.command("duckdb") +@click.argument("output", type=click.Path(dir_okay=False, writable=True)) +@click.option( + "--database-url", + default="mongodb://127.0.0.1:27017", + envvar="DATABASE_URL", + help="MongoDB connection string", + show_default=True, +) +@click.option( + "--database-name", + default="cfdb", + envvar="DATABASE_NAME", + help="MongoDB database holding the materialized files collection", + show_default=True, +) +@click.option( + "--submission", + "-s", + "submissions", + multiple=True, + type=click.Choice(get_all_dcc_names(), case_sensitive=False), + help="Restrict the dump to these consortia. Repeatable; omit for all.", +) +@click.option( + "--to-parquet", + "to_parquet", + type=click.Path(file_okay=False), + default=None, + help="Also export every table as Parquet into this directory.", +) +@click.option( + "--force", + is_flag=True, + help="Overwrite OUTPUT if it already exists.", +) +@click.option( + "--debug", + "-d", + callback=debug, + expose_value=False, + help="Run with debugger listening on the specified port.", + is_eager=True, + type=int, +) +def dump_duckdb( + output: str, + database_url: str, + database_name: str, + submissions: tuple[str, ...], + to_parquet: str | None, + force: bool, +): + """ + Dump the materialized files collection to a relational DuckDB database. + + Streams every document in ``files`` (optionally restricted by + --submission) and writes OUTPUT, a single .duckdb file holding the + flattened schema defined in src/cfdb/dump/schema.sql. Requires the + ``duckdb`` extra: + pip install "cfdb[duckdb] @ git+https://github.com/abdenlab/cfdb.git". + + Examples: + + cfdb dump duckdb cfdb.duckdb + + cfdb dump duckdb cfdb.duckdb -s 4dn -s encode --to-parquet ./parquet + """ + try: + import duckdb + except ImportError: + click.echo( + "Error: the DuckDB dump requires the 'duckdb' extra: " + 'pip install "cfdb[duckdb] @ git+https://github.com/abdenlab/cfdb.git"', + err=True, + ) + raise SystemExit(1) + + from pathlib import Path + + from cfdb.db import mongo_client_kwargs + from cfdb.dump import DumpError, dump_files + + output_path = Path(output) + if output_path.exists(): + if not force: + click.echo( + f"Error: {output_path} already exists (pass --force to overwrite)", + err=True, + ) + raise SystemExit(1) + output_path.unlink() + + normalized = [s.lower() for s in submissions] + query = {"submission": {"$in": normalized}} if normalized else {} + client = MongoClient(database_url, **mongo_client_kwargs()) + cursor = client[database_name]["files"].find(query, batch_size=1000) + + con = duckdb.connect(str(output_path)) + completed = False + try: + report = dump_files( + cursor, + con, + source_database=database_name, + submissions=normalized or None, + ) + completed = True + except DumpError as e: + click.echo(f"Error: {e}", err=True) + raise SystemExit(1) + finally: + client.close() + if not completed: + # A partial dump is worse than no dump: a finally block (rather + # than except) runs this cleanup on every exit path, including + # KeyboardInterrupt. + con.close() + output_path.unlink(missing_ok=True) + + # From here the dump is complete and self-describing; a failure in the + # secondary parquet export must not destroy it. + export_error: Exception | None = None + if to_parquet: + parquet_dir = str(Path(to_parquet)).replace("'", "''") + try: + con.execute(f"EXPORT DATABASE '{parquet_dir}' (FORMAT PARQUET)") + except Exception as e: + export_error = e + con.close() + + click.echo(report.summary()) + click.echo(f"Wrote {output_path}") + if export_error is not None: + click.echo( + f"Error: Parquet export to {to_parquet} failed ({export_error}); " + f"{output_path} is complete and was kept", + err=True, + ) + raise SystemExit(1) + if to_parquet: + click.echo(f"Exported Parquet to {to_parquet}") + + if __name__ == "__main__": cli() diff --git a/src/cfdb/dump/__init__.py b/src/cfdb/dump/__init__.py new file mode 100644 index 0000000..f5f6d8b --- /dev/null +++ b/src/cfdb/dump/__init__.py @@ -0,0 +1,14 @@ +"""Relational DuckDB dump of the materialized ``files`` collection. + +See :mod:`cfdb.dump.loader` for the flattening semantics and +``schema.sql`` for the schema of record (issue #123). +""" + +from cfdb.dump.loader import ( + SCHEMA_VERSION, + DumpError, + DumpReport, + dump_files, +) + +__all__ = ["SCHEMA_VERSION", "DumpError", "DumpReport", "dump_files"] diff --git a/src/cfdb/dump/loader.py b/src/cfdb/dump/loader.py new file mode 100644 index 0000000..cf24528 --- /dev/null +++ b/src/cfdb/dump/loader.py @@ -0,0 +1,1071 @@ +"""Flatten the materialized ``files`` collection into a DuckDB database. + +The schema of record is ``schema.sql`` next to this module (issue #123): +C2M2-mirroring entity and junction tables keyed by natural +``(id_namespace, local_id)`` composites, promoted cross-consortium common +columns, per-consortium ``*_extra_*`` extension tables, three convenience +views, and provenance tables that make every dump self-describing. + +The loader consumes an iterable of raw ``files`` documents — the CLI +wires in a pymongo cursor; tests feed plain dicts — and validates each +through :class:`cfdb.models.FileMetadataModel`, exactly as the GraphQL +resolvers hydrate the same documents, so the model's read validators +(empty-string-to-None, 4DN CV-token coercion) apply here as they do on +the API read path. ``submission`` is not a model field and is read from +the raw document. + +Dedup policy. Entity rows repeat across file documents and nothing +guarantees the repeats are identical — enrichment passes update in +place — so entity rows are first-write-wins, with every divergent later +copy counted and a bounded sample written to ``dump_conflict``: +divergence is reported, never silently resolved. Junction rows instead +take the union over all copies (``SELECT DISTINCT`` at finalize), +because a relationship dropped by first-wins would be silent data loss +rather than a visible conflict. A key that resolves to two different +submissions aborts the dump: the natural keys are only globally unique +because id_namespaces are DCC-scoped, and that assumption is asserted +rather than trusted. + +Memory model. Low-cardinality entities (dcc, project, ontology terms, +collections, biosamples, subjects, and their 1:1 extension rows) are +accumulated in Python registries so divergence can be compared column by +column. High-cardinality rows — files, their 1:1 extras, and every +junction — stream to constraint-free TEMP staging tables in batches, so +memory is bounded by the entity registries rather than the file count. +DuckDB has no ``ALTER TABLE ... ADD CONSTRAINT``, so the real tables +declare their foreign keys up front and finalize inserts parents before +children: dcc -> project -> ontology -> entities -> file -> junctions +and extensions. +""" + +from __future__ import annotations + +import json +import logging +import re +import subprocess +from collections import Counter +from dataclasses import dataclass, field +from datetime import datetime, timezone +from importlib import metadata, resources +from pathlib import Path +from typing import TYPE_CHECKING, Any, Iterable, Optional + +from pydantic import ValidationError + +from cfdb.models import Biosample, Collection, FileMetadataModel, Subject + +if TYPE_CHECKING: # duckdb is an optional extra; the connection is passed in. + import duckdb + +logger = logging.getLogger(__name__) + +_SCHEMA_SQL = (resources.files(__package__) / "schema.sql").read_text() + +#: Parsed from the ``-- schema_version: N`` header so the Python constant +#: cannot drift from the checked-in DDL. +_VERSION_MATCH = re.search(r"--\s*schema_version:\s*(\d+)", _SCHEMA_SQL) +if _VERSION_MATCH is None: + raise RuntimeError( + "schema.sql carries no '-- schema_version: N' header line; " + "cfdb.dump parses it at import, so restore the header" + ) +SCHEMA_VERSION = int(_VERSION_MATCH.group(1)) + +#: Registry-backed tables, in foreign-key insert order. +_REGISTRY_INSERT_ORDER = ( + "dcc", + "project", + "file_format", + "data_type", + "assay_type", + "anatomy", + "ncbi_taxonomy", + "collection", + "biosample", + "subject", + "collection_extra_fourdn", + "collection_extra_encode", + "collection_extra_hubmap", + "biosample_extra_encode", + "subject_extra_hubmap", +) + +#: Streamed tables, staged in TEMP shadows and inserted at finalize — the +#: file table first (its extras and junctions reference it). +_STAGED_FILE_TABLES = ( + "file", + "file_extra_fourdn", + "file_extra_encode", + "file_extra_hubmap", + "file_extra_fourdn_extra_file", +) +_JUNCTION_TABLES = ( + "file_in_collection", + "biosample_in_collection", + "subject_in_collection", + "biosample_from_subject", + "collection_anatomy", + "subject_race", +) +_STAGED_TABLES = _STAGED_FILE_TABLES + _JUNCTION_TABLES + + +class DumpError(RuntimeError): + """A condition that invalidates the dump as a whole.""" + + +@dataclass +class DumpReport: + """What a finished dump did, for the CLI summary and tests.""" + + file_count: int + rows: dict[str, int] + conflicts: dict[str, int] + stats: dict[str, int] + + def summary(self) -> str: + lines = [f"files dumped: {self.file_count}"] + lines += [ + f" {table}: {count} rows" for table, count in sorted(self.rows.items()) + ] + if self.conflicts: + lines.append("divergent entity copies (first write kept):") + lines += [f" {t}: {n}" for t, n in sorted(self.conflicts.items())] + counters = {k: v for k, v in self.stats.items() if v} + if counters: + lines.append("counters:") + lines += [f" {k}: {v}" for k, v in sorted(counters.items())] + return "\n".join(lines) + + +@dataclass +class _Registry: + """First-write-wins row store for one low-cardinality table.""" + + table: str + key_cols: tuple[str, ...] = ("id_namespace", "local_id") + rows: dict[tuple, dict] = field(default_factory=dict) + #: Keys registered as placeholders (a project known only through a + #: foreign key). A real row replaces a stub silently; a stub never + #: overwrites anything and never counts as a conflict. + stubs: set[tuple] = field(default_factory=set) + + +def _utc_naive(value: datetime) -> datetime: + """Normalize to UTC and drop tzinfo for DuckDB's TIMESTAMP.""" + if value.tzinfo is not None: + value = value.astimezone(timezone.utc).replace(tzinfo=None) + return value + + +def _cfdb_version() -> Optional[str]: + try: + return metadata.version("cfdb") + except metadata.PackageNotFoundError: + return None + + +def _cfdb_commit() -> Optional[str]: + """Best-effort HEAD sha when run from a cfdb checkout; None elsewhere. + + Anchored to this module's directory, never the process CWD, so the + probe resolves the repository containing the installed package. The + trackedness check rejects a site-packages that merely sits inside + some other repository (a venv created in an analysis repo): only an + editable checkout has this file tracked by the repo that contains it. + """ + module_path = Path(__file__).resolve() + try: + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", module_path.name], + capture_output=True, + text=True, + timeout=5, + cwd=module_path.parent, + ) + if tracked.returncode != 0: + return None + result = subprocess.run( + ["git", "rev-parse", "HEAD"], + capture_output=True, + text=True, + timeout=5, + cwd=module_path.parent, + ) + except (OSError, subprocess.SubprocessError): + return None + sha = result.stdout.strip() + return sha if result.returncode == 0 and sha else None + + +def _table_columns(con: duckdb.DuckDBPyConnection, table: str) -> list[str]: + rows = con.execute(f'PRAGMA table_info("{table}")').fetchall() + return [row[1] for row in rows] + + +class _Dumper: + """Single-use orchestrator behind :func:`dump_files`.""" + + def __init__( + self, + con: duckdb.DuckDBPyConnection, + *, + source_database: str, + source_collection: str, + submissions: Optional[list[str]], + conflict_sample_cap: int, + batch_size: int, + ) -> None: + self.con = con + self.source_database = source_database + self.source_collection = source_collection + self.requested_submissions = list(submissions) if submissions else None + self.conflict_sample_cap = conflict_sample_cap + self.batch_size = batch_size + + self.stats: Counter[str] = Counter() + self.conflicts: Counter[str] = Counter() + self.conflict_samples: list[tuple] = [] + self._sample_counts: Counter[str] = Counter() + self.seen_submissions: set[str] = set() + self._dcc_id_by_submission: dict[str, str] = {} + self.file_count = 0 + + self.registries: dict[str, _Registry] = { + "dcc": _Registry("dcc", key_cols=("id",)), + "project": _Registry("project"), + "file_format": _Registry("file_format", key_cols=("id",)), + "data_type": _Registry("data_type", key_cols=("id",)), + "assay_type": _Registry("assay_type", key_cols=("id",)), + "anatomy": _Registry("anatomy", key_cols=("id",)), + "ncbi_taxonomy": _Registry("ncbi_taxonomy", key_cols=("id",)), + "collection": _Registry("collection"), + "biosample": _Registry("biosample"), + "subject": _Registry("subject"), + "collection_extra_fourdn": _Registry("collection_extra_fourdn"), + "collection_extra_encode": _Registry("collection_extra_encode"), + "collection_extra_hubmap": _Registry("collection_extra_hubmap"), + "biosample_extra_encode": _Registry("biosample_extra_encode"), + "subject_extra_hubmap": _Registry("subject_extra_hubmap"), + } + self._columns: dict[str, list[str]] = {} + self._buffers: dict[str, list[dict]] = {table: [] for table in _STAGED_TABLES} + + # ── schema and staging ────────────────────────────────────────────── + + def bootstrap(self) -> None: + self.con.execute(_SCHEMA_SQL) + for table in _REGISTRY_INSERT_ORDER + _STAGED_TABLES: + self._columns[table] = _table_columns(self.con, table) + for table in _STAGED_TABLES: + # LIMIT 0 copies the column shape without the constraints, so + # staging accepts rows in any order and duplicates survive + # until finalize dedupes or rejects them. + self.con.execute( + f'CREATE TEMP TABLE "stage_{table}" AS SELECT * FROM "{table}" LIMIT 0' + ) + + def _stage(self, table: str, row: dict) -> None: + buffer = self._buffers[table] + buffer.append(row) + if len(buffer) >= self.batch_size: + self._flush(table) + + def _flush(self, table: str) -> None: + buffer = self._buffers[table] + if not buffer: + return + self._insert_rows(f"stage_{table}", self._columns[table], buffer) + buffer.clear() + + def _insert_rows(self, table: str, columns: list[str], rows: list[dict]) -> None: + column_list = ", ".join(f'"{c}"' for c in columns) + placeholders = ", ".join("?" for _ in columns) + self.con.executemany( + f'INSERT INTO "{table}" ({column_list}) VALUES ({placeholders})', + [tuple(row.get(c) for c in columns) for row in rows], + ) + + # ── registries ────────────────────────────────────────────────────── + + def _register(self, table: str, row: dict, *, stub: bool = False) -> None: + registry = self.registries[table] + key = tuple(row[c] for c in registry.key_cols) + existing = registry.rows.get(key) + if existing is None: + registry.rows[key] = row + if stub: + registry.stubs.add(key) + return + # The two-submission assertion runs before the stub short-circuits: + # a stub carries the referencing document's real submission, so the + # comparison is well-defined in every existing/incoming × real/stub + # combination, and only the attribute upgrade stays silent. + if "submission" in existing and existing["submission"] != row.get("submission"): + key_names = ", ".join(registry.key_cols) + raise DumpError( + f"{table} key {key!r} appears under two submissions " + f"({existing['submission']!r} and {row.get('submission')!r}); " + f"({key_names}) is no longer globally unique and the " + "dump schema's keying assumption does not hold" + ) + if stub: + return + if key in registry.stubs: + registry.rows[key] = row + registry.stubs.discard(key) + return + diffs = [c for c in existing if existing.get(c) != row.get(c)] + if not diffs: + return + self.conflicts[table] += 1 + natural_key = "|".join(str(part) for part in key) + for column in diffs: + if self._sample_counts[table] >= self.conflict_sample_cap: + break + self._sample_counts[table] += 1 + kept = existing.get(column) + discarded = row.get(column) + self.conflict_samples.append( + ( + table, + natural_key, + column, + None if kept is None else str(kept), + None if discarded is None else str(discarded), + ) + ) + + # ── coercions ─────────────────────────────────────────────────────── + + def _coerce_timestamp( + self, value: Optional[str], counter: str + ) -> Optional[datetime]: + if not value: + return None + try: + return _utc_naive(datetime.fromisoformat(value.replace("Z", "+00:00"))) + except ValueError: + self.stats[counter] += 1 + return None + + def _precoerce_int(self, container: Any, key: str, counter: str) -> None: + """Coerce a known string-typed numeric field on the raw document. + + 4DN and HuBMAP load ``size_in_bytes`` (and 4DN its sidecar + ``file_size``) as strings through the C2M2 TSV path. Left alone, + a non-numeric value would fail the whole document's model + validation; coerced here, the failure costs one NULL and one + counted coercion instead of the document. + """ + if not isinstance(container, dict): + return + value = container.get(key) + if value is None or isinstance(value, int) and not isinstance(value, bool): + return + if isinstance(value, str): + text = value.strip() + if not text: + container[key] = None + return + try: + container[key] = int(text) + except ValueError: + container[key] = None + self.stats[counter] += 1 + return + if isinstance(value, float) and value.is_integer(): + container[key] = int(value) + return + container[key] = None + self.stats[counter] += 1 + + def _precoerce(self, raw: dict) -> None: + self._precoerce_int(raw, "size_in_bytes", "size_in_bytes_uncoercible") + extra = raw.get("extra") + fourdn = extra.get("fourdn") if isinstance(extra, dict) else None + extra_files = fourdn.get("extra_files") if isinstance(fourdn, dict) else None + if isinstance(extra_files, list): + for entry in extra_files: + self._precoerce_int(entry, "file_size", "extra_file_size_uncoercible") + + # ── document processing ───────────────────────────────────────────── + + def _skip_invalid(self, raw: dict, reason: str) -> None: + self.stats["documents_invalid"] += 1 + logger.warning( + "Skipping document %s/%s: %s", + raw.get("submission"), + raw.get("local_id"), + reason, + ) + + def process(self, raw: dict) -> None: + self._precoerce(raw) + try: + model = FileMetadataModel.model_validate(raw) + except ValidationError as exc: + self._skip_invalid(raw, f"model validation failed: {exc.errors()[:3]}") + return + submission = raw.get("submission") + if not submission or not isinstance(submission, str): + self._skip_invalid(raw, "no submission field") + return + if not model.id_namespace or not model.local_id: + self._skip_invalid(raw, "missing (id_namespace, local_id) key") + return + if not model.dcc.id: + self._skip_invalid(raw, "dcc carries no id") + return + + self.seen_submissions.add(submission) + self.file_count += 1 + + self._register_dcc(model, submission) + project_fk = self._project_fk( + model.project_id_namespace, model.project_local_id, submission, model + ) + self._stage_file(model, submission, project_fk) + self._stage_file_extras(model) + + for collection in model.collections: + self._process_collection(model, collection, submission) + + def _register_dcc(self, model: FileMetadataModel, submission: str) -> None: + dcc = model.dcc + # dcc.submission is declared UNIQUE, so the id <-> submission mapping + # must be 1:1. _register's submission check asserts one direction + # (one id under two submissions); this asserts the other, so the + # violation is a one-line DumpError instead of a finalize traceback. + known_id = self._dcc_id_by_submission.setdefault(submission, dcc.id) + if known_id != dcc.id: + raise DumpError( + f"submission {submission!r} carries two dcc ids " + f"({known_id!r} and {dcc.id!r}); dcc.submission is declared " + "UNIQUE and the schema's 1:1 mapping assumption does not hold" + ) + self._register( + "dcc", + { + "id": dcc.id, + "submission": submission, + "dcc_name": dcc.dcc_name, + "dcc_abbreviation": dcc.dcc_abbreviation, + "dcc_description": dcc.dcc_description, + "contact_email": dcc.contact_email or None, + "contact_name": dcc.contact_name or None, + "dcc_url": dcc.dcc_url or None, + "project_id_namespace": dcc.project_id_namespace or None, + "project_local_id": dcc.project_local_id or None, + }, + ) + + def _project_fk( + self, + id_namespace: Optional[str], + local_id: Optional[str], + submission: str, + model: Optional[FileMetadataModel] = None, + ) -> tuple[Optional[str], Optional[str]]: + """Normalize a project FK pair and guarantee its target row exists. + + The joined project object and the FK columns travel separately in + the source documents, so an FK can name a project the join never + materialized — and biosamples and subjects carry project FKs with + no joined object at all. A stub row (all-NULL attributes) keeps + the foreign key satisfiable without inventing data; the real + object upgrades the stub whenever a document carries it. + """ + id_namespace = id_namespace or None + local_id = local_id or None + if (id_namespace is None) != (local_id is None): + self.stats["project_fk_incomplete"] += 1 + return None, None + if id_namespace is None: + return None, None + project = model.project if model is not None else None + if ( + project is not None + and project.id_namespace == id_namespace + and project.local_id == local_id + ): + self._register( + "project", + { + "id_namespace": id_namespace, + "local_id": local_id, + "submission": submission, + "name": project.name or None, + "abbreviation": project.abbreviation, + "description": project.description, + "persistent_id": project.persistent_id, + }, + ) + else: + registry = self.registries["project"] + if (id_namespace, local_id) not in registry.rows: + self.stats["project_stubs_created"] += 1 + self._register( + "project", + { + "id_namespace": id_namespace, + "local_id": local_id, + "submission": submission, + "name": None, + "abbreviation": None, + "description": None, + "persistent_id": None, + }, + stub=True, + ) + return id_namespace, local_id + + def _term_fk(self, table: str, term: Any) -> Optional[str]: + """Register a CV term and return its id for an FK column, or None. + + The term models default both ``id`` and ``name`` to ``""`` and + pass validation that way, so either can arrive blank on a + model-valid document. A term without an id cannot be keyed and a + term without a name cannot satisfy the NOT NULL name column, so + both are skipped and counted here — nulling the referencing FK + column in the same decision, which is what keeps a dirty term a + counter in ``dump_stat`` rather than a constraint crash at + finalize. + """ + if term is None: + return None + if not term.id: + self.stats["term_id_missing"] += 1 + return None + if not term.name: + self.stats["term_name_missing"] += 1 + return None + row = { + "id": term.id, + "name": term.name, + "description": term.description, + } + if table == "ncbi_taxonomy": + row["clade"] = term.clade + self._register(table, row) + return term.id + + @staticmethod + def _any_value(row: dict, key_cols: tuple[str, ...]) -> bool: + return any( + value is not None for column, value in row.items() if column not in key_cols + ) + + def _stage_file( + self, + model: FileMetadataModel, + submission: str, + project_fk: tuple[Optional[str], Optional[str]], + ) -> None: + self._stage( + "file", + { + "id_namespace": model.id_namespace, + "local_id": model.local_id, + "submission": submission, + "accession_id": model.accession_id, + "filename": model.filename or None, + "persistent_id": model.persistent_id, + "creation_time": self._coerce_timestamp( + model.creation_time, "creation_time_unparseable" + ), + "size_in_bytes": model.size_in_bytes, + "sha256": model.sha256, + "md5": model.md5, + "file_format": self._term_fk("file_format", model.file_format), + "compression_format": model.compression_format, + "data_type": self._term_fk("data_type", model.data_type), + "assay_type": self._term_fk("assay_type", model.assay_type), + "analysis_type": model.analysis_type, + "mime_type": model.mime_type, + "project_id_namespace": project_fk[0], + "project_local_id": project_fk[1], + "bundle_collection_id_namespace": model.bundle_collection_id_namespace, + "bundle_collection_local_id": model.bundle_collection_local_id, + "dbgap_study_id": model.dbgap_study_id, + "access_url": model.access_url, + "status": model.status, + "data_access_level": model.data_access_level, + "genome_assembly": model.genome_assembly, + "genome_annotation": model.genome_annotation, + "output_type": model.output_type, + "output_type_detail": model.output_type_detail, + "biological_replicates": model.biological_replicates, + "technical_replicates": model.technical_replicates, + "assay_info": model.assay_info, + "condition": model.condition, + }, + ) + + def _stage_file_extras(self, model: FileMetadataModel) -> None: + extra = model.extra + if extra is None: + return + key = {"id_namespace": model.id_namespace, "local_id": model.local_id} + key_cols = ("id_namespace", "local_id") + if extra.fourdn is not None: + fourdn = extra.fourdn + row = { + **key, + "enriched_file_format": fourdn.enriched_file_format, + "dataset": fourdn.dataset, + "biosource_name": fourdn.biosource_name, + "replicate_info": fourdn.replicate_info, + "cell_line_tier": fourdn.cell_line_tier, + } + if self._any_value(row, key_cols): + self._stage("file_extra_fourdn", row) + for ordinal, entry in enumerate(fourdn.extra_files or []): + self._stage( + "file_extra_fourdn_extra_file", + { + **key, + "ordinal": ordinal, + "href": entry.href, + "md5sum": entry.md5sum, + "file_size": entry.file_size, + "file_format": entry.file_format, + }, + ) + if extra.encode is not None: + encode = extra.encode + row = { + **key, + "annotation_type": encode.annotation_type, + "organism": encode.organism, + "controlled_by": encode.controlled_by, + "s3_uri": encode.s3_uri, + "azure_url": encode.azure_url, + "file_analysis_title": encode.file_analysis_title, + "file_analysis_status": encode.file_analysis_status, + "read_length": encode.read_length, + "mapped_read_length": encode.mapped_read_length, + "run_type": encode.run_type, + "paired_end": encode.paired_end, + "paired_with": encode.paired_with, + "index_of": encode.index_of, + "derived_from": encode.derived_from, + "audit_warning": encode.audit_warning, + "audit_not_compliant": encode.audit_not_compliant, + "audit_error": encode.audit_error, + } + if self._any_value(row, key_cols): + self._stage("file_extra_encode", row) + if extra.hubmap is not None: + hubmap = extra.hubmap + row = { + **key, + "rel_path": hubmap.rel_path, + "is_data_product": hubmap.is_data_product, + } + if self._any_value(row, key_cols): + self._stage("file_extra_hubmap", row) + + def _process_collection( + self, model: FileMetadataModel, collection: Collection, submission: str + ) -> None: + if not collection.id_namespace or not collection.local_id: + self.stats["entities_missing_key"] += 1 + return + self._register( + "collection", + { + "id_namespace": collection.id_namespace, + "local_id": collection.local_id, + "submission": submission, + "accession_id": collection.accession_id, + "persistent_id": collection.persistent_id, + "creation_time": self._coerce_timestamp( + collection.creation_time, "creation_time_unparseable" + ), + "abbreviation": collection.abbreviation, + "name": collection.name or None, + "description": collection.description, + "lab": collection.lab, + "experiment_type": collection.experiment_type, + "experiment_target": collection.experiment_target, + "analyte_class": collection.analyte_class, + }, + ) + self._register_collection_extras(collection) + collection_key = { + "collection_id_namespace": collection.id_namespace, + "collection_local_id": collection.local_id, + } + self._stage( + "file_in_collection", + { + "file_id_namespace": model.id_namespace, + "file_local_id": model.local_id, + **collection_key, + }, + ) + for anatomy in collection.anatomy: + anatomy_id = self._term_fk("anatomy", anatomy) + if anatomy_id is None: + continue + self._stage("collection_anatomy", {**collection_key, "anatomy": anatomy_id}) + for subject in collection.subjects: + if not self._register_subject(subject, submission): + continue + self._stage( + "subject_in_collection", + { + "subject_id_namespace": subject.id_namespace, + "subject_local_id": subject.local_id, + **collection_key, + }, + ) + for biosample in collection.biosamples: + if not self._register_biosample(biosample, submission): + continue + self._stage( + "biosample_in_collection", + { + "biosample_id_namespace": biosample.id_namespace, + "biosample_local_id": biosample.local_id, + **collection_key, + }, + ) + + def _register_collection_extras(self, collection: Collection) -> None: + extra = collection.extra + if extra is None: + return + key = { + "id_namespace": collection.id_namespace, + "local_id": collection.local_id, + } + key_cols = ("id_namespace", "local_id") + if extra.fourdn is not None: + fourdn = extra.fourdn + row = { + **key, + "display_title": fourdn.display_title, + "targeted_factor": fourdn.targeted_factor, + "digestion_enzyme": fourdn.digestion_enzyme, + "crosslinking_method": fourdn.crosslinking_method, + "crosslinking_temperature": fourdn.crosslinking_temperature, + "crosslinking_time": fourdn.crosslinking_time, + "ligation_temperature": fourdn.ligation_temperature, + "ligation_volume": fourdn.ligation_volume, + "ligation_time": fourdn.ligation_time, + "digestion_temperature": fourdn.digestion_temperature, + "digestion_time": fourdn.digestion_time, + "tagging_method": fourdn.tagging_method, + "fragmentation_method": fourdn.fragmentation_method, + "biotin_removed": fourdn.biotin_removed, + "library_prep_kit": fourdn.library_prep_kit, + "average_fragment_size": fourdn.average_fragment_size, + "fragment_size_range": fourdn.fragment_size_range, + "status": fourdn.status, + "date_created": self._coerce_timestamp( + fourdn.date_created, "date_created_unparseable" + ), + } + if self._any_value(row, key_cols): + self._register("collection_extra_fourdn", row) + if extra.encode is not None: + encode = extra.encode + row = { + **key, + "project": encode.project, + "platform": encode.platform, + "dbxrefs": encode.dbxrefs, + "rbns_protein_concentration": encode.rbns_protein_concentration, + "annotation_type": encode.annotation_type, + "software_used": encode.software_used, + "encyclopedia_version": encode.encyclopedia_version, + } + if self._any_value(row, key_cols): + self._register("collection_extra_encode", row) + if extra.hubmap is not None: + hubmap = extra.hubmap + row = { + **key, + "pipeline": hubmap.pipeline, + "processing": hubmap.processing, + "group_name": hubmap.group_name, + "visualization": hubmap.visualization, + "vitessce_hints": hubmap.vitessce_hints, + "metadata": ( + # sort_keys so two equal, key-reordered copies compare + # equal instead of registering a false divergence. + None + if hubmap.metadata is None + else json.dumps(hubmap.metadata, sort_keys=True) + ), + } + if self._any_value(row, key_cols): + self._register("collection_extra_hubmap", row) + + def _register_biosample(self, biosample: Biosample, submission: str) -> bool: + if not biosample.id_namespace or not biosample.local_id: + self.stats["entities_missing_key"] += 1 + return False + anatomy_id = self._term_fk("anatomy", biosample.anatomy) + project_fk = self._project_fk( + biosample.project_id_namespace, biosample.project_local_id, submission + ) + self._register( + "biosample", + { + "id_namespace": biosample.id_namespace, + "local_id": biosample.local_id, + "submission": submission, + "project_id_namespace": project_fk[0], + "project_local_id": project_fk[1], + "persistent_id": biosample.persistent_id, + "creation_time": self._coerce_timestamp( + biosample.creation_time, "creation_time_unparseable" + ), + "sample_prep_method": biosample.sample_prep_method, + "anatomy": anatomy_id, + "biofluid": biosample.biofluid, + }, + ) + if biosample.extra is not None and biosample.extra.encode is not None: + encode = biosample.extra.encode + row = { + "id_namespace": biosample.id_namespace, + "local_id": biosample.local_id, + "biosample_type": encode.biosample_type, + "life_stage": encode.life_stage, + "age": encode.age, + "age_units": encode.age_units, + "biosample_treatments": encode.biosample_treatments, + "biosample_treatments_amount": encode.biosample_treatments_amount, + "biosample_treatments_duration": encode.biosample_treatments_duration, + "biosample_genetic_modifications": ( + encode.biosample_genetic_modifications + ), + "library_made_from": encode.library_made_from, + "library_depleted_in": encode.library_depleted_in, + "library_extraction_method": encode.library_extraction_method, + "library_lysis_method": encode.library_lysis_method, + "library_crosslinking_method": encode.library_crosslinking_method, + "library_strand_specific": encode.library_strand_specific, + "library_fragmentation_method": encode.library_fragmentation_method, + "library_size_range": encode.library_size_range, + } + if self._any_value(row, ("id_namespace", "local_id")): + self._register("biosample_extra_encode", row) + for subject in biosample.subjects: + if not self._register_subject(subject, submission): + continue + self._stage( + "biosample_from_subject", + { + "biosample_id_namespace": biosample.id_namespace, + "biosample_local_id": biosample.local_id, + "subject_id_namespace": subject.id_namespace, + "subject_local_id": subject.local_id, + }, + ) + return True + + def _register_subject(self, subject: Subject, submission: str) -> bool: + if not subject.id_namespace or not subject.local_id: + self.stats["entities_missing_key"] += 1 + return False + taxonomy_id = self._term_fk("ncbi_taxonomy", subject.taxonomy) + project_fk = self._project_fk( + subject.project_id_namespace, subject.project_local_id, submission + ) + self._register( + "subject", + { + "id_namespace": subject.id_namespace, + "local_id": subject.local_id, + "submission": submission, + "project_id_namespace": project_fk[0], + "project_local_id": project_fk[1], + "persistent_id": subject.persistent_id, + "creation_time": self._coerce_timestamp( + subject.creation_time, "creation_time_unparseable" + ), + "granularity": subject.granularity, + "sex": subject.sex, + "ethnicity": subject.ethnicity, + "age_at_enrollment": subject.age_at_enrollment, + "age_at_sampling": subject.age_at_sampling, + "taxonomy": taxonomy_id, + }, + ) + for race in subject.race: + if not race: + continue + self._stage( + "subject_race", + { + "subject_id_namespace": subject.id_namespace, + "subject_local_id": subject.local_id, + "race": race, + }, + ) + if subject.extra is not None and subject.extra.hubmap is not None: + hubmap = subject.extra.hubmap + row = { + "id_namespace": subject.id_namespace, + "local_id": subject.local_id, + "age_value": hubmap.age_value, + "age_unit": hubmap.age_unit, + "sex": hubmap.sex, + "race": hubmap.race, + "body_mass_index_value": hubmap.body_mass_index_value, + "body_mass_index_unit": hubmap.body_mass_index_unit, + "cause_of_death": hubmap.cause_of_death, + "death_event": hubmap.death_event, + "mechanism_of_injury": hubmap.mechanism_of_injury, + "medical_history": hubmap.medical_history, + "social_history": hubmap.social_history, + "height_value": hubmap.height_value, + "height_unit": hubmap.height_unit, + "weight_value": hubmap.weight_value, + "weight_unit": hubmap.weight_unit, + } + if self._any_value(row, ("id_namespace", "local_id")): + self._register("subject_extra_hubmap", row) + return True + + # ── finalize ──────────────────────────────────────────────────────── + + def _assert_unique_file_keys(self) -> None: + offenders = self.con.execute( + """ + SELECT id_namespace, local_id, count(*) AS copies, + count(DISTINCT submission) AS submissions + FROM stage_file + GROUP BY id_namespace, local_id + HAVING count(*) > 1 + LIMIT 5 + """ + ).fetchall() + if offenders: + detail = "; ".join( + f"({ns!r}, {lid!r}) x{copies} across {subs} submission(s)" + for ns, lid, copies, subs in offenders + ) + raise DumpError( + f"duplicate file keys in the source (first 5 shown): {detail}" + ) + + def finalize(self) -> DumpReport: + for table in _STAGED_TABLES: + self._flush(table) + self._assert_unique_file_keys() + + # A live connection was handed in, so duckdb resolves at runtime even + # though the module only imports it for type checking. + import duckdb + + table = "" + try: + for table in _REGISTRY_INSERT_ORDER: + registry = self.registries[table] + if registry.rows: + self._insert_rows( + table, self._columns[table], list(registry.rows.values()) + ) + + for table in _STAGED_FILE_TABLES: + self.con.execute( + f'INSERT INTO "{table}" SELECT * FROM "stage_{table}"' + ) + for table in _JUNCTION_TABLES: + self.con.execute( + f'INSERT INTO "{table}" SELECT DISTINCT * FROM "stage_{table}"' + ) + except duckdb.ConstraintException as e: + raise DumpError( + f"finalize insert into {table!r} violated a schema constraint " + f"({e}); the source breaks an assumption the schema declares" + ) from e + for table in _STAGED_TABLES: + self.con.execute(f'DROP TABLE "stage_{table}"') + + if self.conflict_samples: + self.con.executemany( + "INSERT INTO dump_conflict VALUES (?, ?, ?, ?, ?)", + self.conflict_samples, + ) + + rows: dict[str, int] = {} + for table in _REGISTRY_INSERT_ORDER + _STAGED_TABLES: + rows[table] = self.con.execute( + f'SELECT count(*) FROM "{table}"' + ).fetchone()[0] + + surviving_stubs = len(self.registries["project"].stubs) + if surviving_stubs: + self.stats["project_stubs_surviving"] = surviving_stubs + stats = dict(self.stats) + stat_rows = [(f"rows_{t}", n) for t, n in sorted(rows.items())] + stat_rows += [(f"conflicts_{t}", n) for t, n in sorted(self.conflicts.items())] + stat_rows += sorted(stats.items()) + self.con.executemany("INSERT INTO dump_stat VALUES (?, ?)", stat_rows) + + submissions = self.requested_submissions or sorted(self.seen_submissions) + self.con.execute( + "INSERT INTO dump_info VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + [ + SCHEMA_VERSION, + _utc_naive(datetime.now(timezone.utc)), + _cfdb_version(), + _cfdb_commit(), + self.source_database, + self.source_collection, + submissions, + self.file_count, + ], + ) + + return DumpReport( + file_count=self.file_count, + rows=rows, + conflicts=dict(self.conflicts), + stats=stats, + ) + + +def dump_files( + docs: Iterable[dict], + con: duckdb.DuckDBPyConnection, + *, + source_database: str, + source_collection: str = "files", + submissions: Optional[list[str]] = None, + conflict_sample_cap: int = 1000, + batch_size: int = 10_000, +) -> DumpReport: + """Flatten ``files`` documents into an empty DuckDB database. + + Executes ``schema.sql`` on ``con`` — which must therefore be a fresh + database — streams ``docs`` through the flattening described in the + module docstring, and returns a :class:`DumpReport`. Raises + :class:`DumpError` when the source violates an assumption the schema + depends on: duplicate file keys, or one key under two submissions. + + The precoercion of string-typed numerics (``size_in_bytes``, + ``extra_files[].file_size``) writes into the supplied dicts in place + — immaterial for a pymongo cursor, visible to a caller that keeps + the dicts. + """ + dumper = _Dumper( + con, + source_database=source_database, + source_collection=source_collection, + submissions=submissions, + conflict_sample_cap=conflict_sample_cap, + batch_size=batch_size, + ) + dumper.bootstrap() + for raw in docs: + dumper.process(raw) + return dumper.finalize() diff --git a/src/cfdb/dump/schema.sql b/src/cfdb/dump/schema.sql new file mode 100644 index 0000000..1b6074a --- /dev/null +++ b/src/cfdb/dump/schema.sql @@ -0,0 +1,576 @@ +-- cfdb DuckDB dump schema (issue #123). +-- +-- schema_version: 2 +-- +-- The schema of record for the flattened relational dump of the +-- materialized ``files`` collection. The design rules, the per-consortium +-- source mapping for every promoted common column, and the loader's dedup +-- and coercion policies are documented in the issue and in the README's +-- "DuckDB dump" section. Bump the schema_version header above whenever +-- this file changes shape; ``cfdb.dump`` parses it at import. +-- +-- DuckDB has no ALTER TABLE ... ADD CONSTRAINT, so every foreign key is +-- declared here and the loader inserts parents before children: +-- dcc -> project -> ontology tables -> entities -> junctions -> extensions. + +-- ─── provenance ──────────────────────────────────────────────────────────── + +CREATE TABLE dump_info ( + schema_version INTEGER NOT NULL, + generated_at TIMESTAMP NOT NULL, + cfdb_version VARCHAR, + cfdb_commit VARCHAR, + source_database VARCHAR NOT NULL, + source_collection VARCHAR NOT NULL, + submissions VARCHAR[] NOT NULL, -- submissions included in this dump + file_count BIGINT NOT NULL +); + +-- Row counts per table, plus coercion-failure counters +-- (size_in_bytes_uncoercible, creation_time_unparseable, ...). +CREATE TABLE dump_stat ( + name VARCHAR PRIMARY KEY, + value BIGINT NOT NULL +); + +-- Sampled divergences where repeated copies of one entity disagreed. +CREATE TABLE dump_conflict ( + table_name VARCHAR NOT NULL, + natural_key VARCHAR NOT NULL, + column_name VARCHAR NOT NULL, + kept_value VARCHAR, + discarded_value VARCHAR +); + +-- ─── reference ───────────────────────────────────────────────────────────── + +CREATE TABLE dcc ( + id VARCHAR PRIMARY KEY, -- CFDE-issued, e.g. cfde_registry_dcc:4dn + submission VARCHAR NOT NULL UNIQUE, -- 4dn | encode | hubmap + dcc_name VARCHAR NOT NULL, + dcc_abbreviation VARCHAR NOT NULL, + dcc_description VARCHAR, + contact_email VARCHAR, + contact_name VARCHAR, + dcc_url VARCHAR, + project_id_namespace VARCHAR, + project_local_id VARCHAR +); + +CREATE TABLE project ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + submission VARCHAR NOT NULL REFERENCES dcc (submission), + name VARCHAR, + abbreviation VARCHAR, + description VARCHAR, + persistent_id VARCHAR, + PRIMARY KEY (id_namespace, local_id) +); + +-- Ontology tables are deliberately submission-free: a term is the same term in +-- every consortium, and a dump exists to be joined across them. The raw Mongo +-- collections are scoped UNIQUE (submission, id), so the loader dedupes on id +-- and records any name/description divergence in dump_conflict. + +CREATE TABLE file_format ( -- EDAM format: terms, or minted cfdb: tokens + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + description VARCHAR +); + +CREATE TABLE data_type ( -- EDAM data: terms + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + description VARCHAR +); + +CREATE TABLE assay_type ( -- OBI terms + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + description VARCHAR +); + +CREATE TABLE anatomy ( -- UBERON terms + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + description VARCHAR +); + +CREATE TABLE ncbi_taxonomy ( + id VARCHAR PRIMARY KEY, + name VARCHAR NOT NULL, + clade VARCHAR, + description VARCHAR +); + +-- ─── entities ────────────────────────────────────────────────────────────── + +CREATE TABLE file ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + submission VARCHAR NOT NULL REFERENCES dcc (submission), + accession_id VARCHAR, -- case-folded; NULL for HuBMAP + filename VARCHAR, + persistent_id VARCHAR, + creation_time TIMESTAMP, + size_in_bytes BIGINT, + sha256 VARCHAR, + md5 VARCHAR, + file_format VARCHAR REFERENCES file_format (id), + compression_format VARCHAR, -- '' means "nothing recorded" + data_type VARCHAR REFERENCES data_type (id), + assay_type VARCHAR REFERENCES assay_type (id), + analysis_type VARCHAR, + mime_type VARCHAR, + project_id_namespace VARCHAR, + project_local_id VARCHAR, + bundle_collection_id_namespace VARCHAR, + bundle_collection_local_id VARCHAR, + dbgap_study_id VARCHAR, + access_url VARCHAR, + status VARCHAR, + data_access_level VARCHAR, -- public | consortium | protected + -- promoted common enrichment + genome_assembly VARCHAR, + genome_annotation VARCHAR, + output_type VARCHAR, + output_type_detail VARCHAR, + biological_replicates VARCHAR, + technical_replicates VARCHAR, + assay_info VARCHAR, + "condition" VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (project_id_namespace, project_local_id) + REFERENCES project (id_namespace, local_id) +); + +CREATE TABLE collection ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + submission VARCHAR NOT NULL REFERENCES dcc (submission), + accession_id VARCHAR, + persistent_id VARCHAR, + creation_time TIMESTAMP, + abbreviation VARCHAR, + name VARCHAR, + description VARCHAR, + -- promoted common enrichment + lab VARCHAR, + experiment_type VARCHAR, + experiment_target VARCHAR, + analyte_class VARCHAR, + PRIMARY KEY (id_namespace, local_id) +); + +CREATE TABLE biosample ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + submission VARCHAR NOT NULL REFERENCES dcc (submission), + project_id_namespace VARCHAR, + project_local_id VARCHAR, + persistent_id VARCHAR, + creation_time TIMESTAMP, + sample_prep_method VARCHAR, + anatomy VARCHAR REFERENCES anatomy (id), + biofluid VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (project_id_namespace, project_local_id) + REFERENCES project (id_namespace, local_id) +); + +CREATE TABLE subject ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + submission VARCHAR NOT NULL REFERENCES dcc (submission), + project_id_namespace VARCHAR, + project_local_id VARCHAR, + persistent_id VARCHAR, + creation_time TIMESTAMP, + granularity VARCHAR, + sex VARCHAR, + ethnicity VARCHAR, + age_at_enrollment DOUBLE, + age_at_sampling DOUBLE, + -- The files projection collapses C2M2's subject_role_taxonomy junction to a + -- single term, so the role is not recoverable from this source. + taxonomy VARCHAR REFERENCES ncbi_taxonomy (id), + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (project_id_namespace, project_local_id) + REFERENCES project (id_namespace, local_id) +); + +-- ─── C2M2 junctions ──────────────────────────────────────────────────────── + +CREATE TABLE file_in_collection ( + file_id_namespace VARCHAR NOT NULL, + file_local_id VARCHAR NOT NULL, + collection_id_namespace VARCHAR NOT NULL, + collection_local_id VARCHAR NOT NULL, + PRIMARY KEY (file_id_namespace, file_local_id, + collection_id_namespace, collection_local_id), + FOREIGN KEY (file_id_namespace, file_local_id) + REFERENCES file (id_namespace, local_id), + FOREIGN KEY (collection_id_namespace, collection_local_id) + REFERENCES collection (id_namespace, local_id) +); + +CREATE TABLE biosample_in_collection ( + biosample_id_namespace VARCHAR NOT NULL, + biosample_local_id VARCHAR NOT NULL, + collection_id_namespace VARCHAR NOT NULL, + collection_local_id VARCHAR NOT NULL, + PRIMARY KEY (biosample_id_namespace, biosample_local_id, + collection_id_namespace, collection_local_id), + FOREIGN KEY (biosample_id_namespace, biosample_local_id) + REFERENCES biosample (id_namespace, local_id), + FOREIGN KEY (collection_id_namespace, collection_local_id) + REFERENCES collection (id_namespace, local_id) +); + +CREATE TABLE subject_in_collection ( + subject_id_namespace VARCHAR NOT NULL, + subject_local_id VARCHAR NOT NULL, + collection_id_namespace VARCHAR NOT NULL, + collection_local_id VARCHAR NOT NULL, + PRIMARY KEY (subject_id_namespace, subject_local_id, + collection_id_namespace, collection_local_id), + FOREIGN KEY (subject_id_namespace, subject_local_id) + REFERENCES subject (id_namespace, local_id), + FOREIGN KEY (collection_id_namespace, collection_local_id) + REFERENCES collection (id_namespace, local_id) +); + +CREATE TABLE biosample_from_subject ( + biosample_id_namespace VARCHAR NOT NULL, + biosample_local_id VARCHAR NOT NULL, + subject_id_namespace VARCHAR NOT NULL, + subject_local_id VARCHAR NOT NULL, + PRIMARY KEY (biosample_id_namespace, biosample_local_id, + subject_id_namespace, subject_local_id), + FOREIGN KEY (biosample_id_namespace, biosample_local_id) + REFERENCES biosample (id_namespace, local_id), + FOREIGN KEY (subject_id_namespace, subject_local_id) + REFERENCES subject (id_namespace, local_id) +); + +CREATE TABLE collection_anatomy ( + collection_id_namespace VARCHAR NOT NULL, + collection_local_id VARCHAR NOT NULL, + anatomy VARCHAR NOT NULL REFERENCES anatomy (id), + PRIMARY KEY (collection_id_namespace, collection_local_id, anatomy), + FOREIGN KEY (collection_id_namespace, collection_local_id) + REFERENCES collection (id_namespace, local_id) +); + +CREATE TABLE subject_race ( + subject_id_namespace VARCHAR NOT NULL, + subject_local_id VARCHAR NOT NULL, + race VARCHAR NOT NULL, + PRIMARY KEY (subject_id_namespace, subject_local_id, race), + FOREIGN KEY (subject_id_namespace, subject_local_id) + REFERENCES subject (id_namespace, local_id) +); + +-- ─── per-consortium extensions (1:1 with parent) ─────────────────────────── + +-- genome_assembly, condition, assay_info, file_type and file_type_detailed are +-- promoted verbatim and omitted here; replicate_info is kept because the +-- replicate columns are regex-extracted from it. +CREATE TABLE file_extra_fourdn ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + enriched_file_format VARCHAR, + dataset VARCHAR, + biosource_name VARCHAR, + replicate_info VARCHAR, + cell_line_tier VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) REFERENCES file (id_namespace, local_id) +); + +CREATE TABLE file_extra_fourdn_extra_file ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + ordinal INTEGER NOT NULL, -- position in extra_files[] + href VARCHAR, + md5sum VARCHAR, + file_size BIGINT, + file_format VARCHAR, -- 4DN token ("pairs_px2", "bai"), not EDAM + PRIMARY KEY (id_namespace, local_id, ordinal), + FOREIGN KEY (id_namespace, local_id) REFERENCES file (id_namespace, local_id) +); + +-- assembly, genome_annotation, output_type, file_format_type and both replicate +-- columns are promoted verbatim and omitted here. +CREATE TABLE file_extra_encode ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + annotation_type VARCHAR, + organism VARCHAR, + controlled_by VARCHAR, + s3_uri VARCHAR, + azure_url VARCHAR, + file_analysis_title VARCHAR, + file_analysis_status VARCHAR, + read_length VARCHAR, + mapped_read_length VARCHAR, + run_type VARCHAR, + paired_end VARCHAR, + paired_with VARCHAR, + index_of VARCHAR, + derived_from VARCHAR, + audit_warning VARCHAR, + audit_not_compliant VARCHAR, + audit_error VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) REFERENCES file (id_namespace, local_id) +); + +CREATE TABLE file_extra_hubmap ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + rel_path VARCHAR, + is_data_product BOOLEAN, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) REFERENCES file (id_namespace, local_id) +); + +-- experiment_type and lab are promoted and omitted here. +CREATE TABLE collection_extra_fourdn ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + display_title VARCHAR, + targeted_factor VARCHAR[], + digestion_enzyme VARCHAR, + crosslinking_method VARCHAR, + crosslinking_temperature VARCHAR, -- numeric upstream, stringified on ingest + crosslinking_time VARCHAR, + ligation_temperature VARCHAR, + ligation_volume VARCHAR, + ligation_time VARCHAR, + digestion_temperature VARCHAR, + digestion_time VARCHAR, + tagging_method VARCHAR, + fragmentation_method VARCHAR, + biotin_removed VARCHAR, + library_prep_kit VARCHAR, + average_fragment_size VARCHAR, + fragment_size_range VARCHAR, -- a genuine range string, e.g. "200-400" + status VARCHAR, + date_created TIMESTAMP, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) + REFERENCES collection (id_namespace, local_id) +); + +-- experiment_target and analyte_class are promoted and omitted here. +CREATE TABLE collection_extra_encode ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + project VARCHAR, + platform VARCHAR, + dbxrefs VARCHAR, + rbns_protein_concentration VARCHAR, + annotation_type VARCHAR, + software_used VARCHAR, + encyclopedia_version VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) + REFERENCES collection (id_namespace, local_id) +); + +-- dataset_type (-> experiment_type) and analyte_class are promoted and omitted. +CREATE TABLE collection_extra_hubmap ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + pipeline VARCHAR, + processing VARCHAR, + group_name VARCHAR, + visualization BOOLEAN, + vitessce_hints VARCHAR[], + metadata JSON, -- open-ended assay-specific dict; no fixed columns + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) + REFERENCES collection (id_namespace, local_id) +); + +CREATE TABLE biosample_extra_encode ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + biosample_type VARCHAR, + life_stage VARCHAR, + age VARCHAR, -- free text upstream ("unknown") + age_units VARCHAR, + biosample_treatments VARCHAR, + biosample_treatments_amount VARCHAR, + biosample_treatments_duration VARCHAR, + biosample_genetic_modifications VARCHAR, + library_made_from VARCHAR, + library_depleted_in VARCHAR, + library_extraction_method VARCHAR, + library_lysis_method VARCHAR, + library_crosslinking_method VARCHAR, + library_strand_specific VARCHAR, + library_fragmentation_method VARCHAR, + library_size_range VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) + REFERENCES biosample (id_namespace, local_id) +); + +CREATE TABLE subject_extra_hubmap ( + id_namespace VARCHAR NOT NULL, + local_id VARCHAR NOT NULL, + age_value DOUBLE, + age_unit VARCHAR, + sex VARCHAR, + race VARCHAR, + body_mass_index_value DOUBLE, + body_mass_index_unit VARCHAR, + cause_of_death VARCHAR, + death_event VARCHAR, + mechanism_of_injury VARCHAR, + medical_history VARCHAR[], + social_history VARCHAR[], + height_value DOUBLE, + height_unit VARCHAR, + weight_value DOUBLE, + weight_unit VARCHAR, + PRIMARY KEY (id_namespace, local_id), + FOREIGN KEY (id_namespace, local_id) + REFERENCES subject (id_namespace, local_id) +); + +-- ─── convenience views ───────────────────────────────────────────────────── + +-- One row per file with the ontology term names and every per-consortium +-- file extension column inlined, so the common read needs no joins. +CREATE VIEW v_file AS +SELECT + f.*, + d.dcc_abbreviation, + ff.name AS file_format_name, + dt.name AS data_type_name, + aty.name AS assay_type_name, + x4.enriched_file_format, + x4.dataset, + x4.biosource_name, + x4.replicate_info, + x4.cell_line_tier, + xe.annotation_type, + xe.organism, + xe.controlled_by, + xe.s3_uri, + xe.azure_url, + xe.file_analysis_title, + xe.file_analysis_status, + xe.read_length, + xe.mapped_read_length, + xe.run_type, + xe.paired_end, + xe.paired_with, + xe.index_of, + xe.derived_from, + xe.audit_warning, + xe.audit_not_compliant, + xe.audit_error, + xh.rel_path, + xh.is_data_product +FROM file AS f +JOIN dcc AS d ON d.submission = f.submission +LEFT JOIN file_format AS ff ON ff.id = f.file_format +LEFT JOIN data_type AS dt ON dt.id = f.data_type +LEFT JOIN assay_type AS aty ON aty.id = f.assay_type +LEFT JOIN file_extra_fourdn AS x4 + ON x4.id_namespace = f.id_namespace AND x4.local_id = f.local_id +LEFT JOIN file_extra_encode AS xe + ON xe.id_namespace = f.id_namespace AND xe.local_id = f.local_id +LEFT JOIN file_extra_hubmap AS xh + ON xh.id_namespace = f.id_namespace AND xh.local_id = f.local_id; + +-- The file x collection x biosample x subject grain -- "which files came from +-- heart tissue". Explicitly a fan-out view: a file row repeats once per +-- biosample/subject path, so never SUM(size_in_bytes) over it. +CREATE VIEW v_file_biosample AS +SELECT + f.submission, + f.id_namespace AS file_id_namespace, + f.local_id AS file_local_id, + f.filename, + f.accession_id AS file_accession_id, + c.id_namespace AS collection_id_namespace, + c.local_id AS collection_local_id, + c.name AS collection_name, + c.experiment_type, + b.id_namespace AS biosample_id_namespace, + b.local_id AS biosample_local_id, + a.id AS anatomy_id, + a.name AS anatomy_name, + s.id_namespace AS subject_id_namespace, + s.local_id AS subject_local_id, + t.name AS taxonomy_name +FROM file AS f +JOIN file_in_collection AS fic + ON fic.file_id_namespace = f.id_namespace AND fic.file_local_id = f.local_id +JOIN collection AS c + ON c.id_namespace = fic.collection_id_namespace + AND c.local_id = fic.collection_local_id +LEFT JOIN biosample_in_collection AS bic + ON bic.collection_id_namespace = c.id_namespace + AND bic.collection_local_id = c.local_id +LEFT JOIN biosample AS b + ON b.id_namespace = bic.biosample_id_namespace + AND b.local_id = bic.biosample_local_id +LEFT JOIN anatomy AS a ON a.id = b.anatomy +LEFT JOIN biosample_from_subject AS bfs + ON bfs.biosample_id_namespace = b.id_namespace + AND bfs.biosample_local_id = b.local_id +LEFT JOIN subject AS s + ON s.id_namespace = bfs.subject_id_namespace + AND s.local_id = bfs.subject_local_id +LEFT JOIN ncbi_taxonomy AS t ON t.id = s.taxonomy; + +-- Subject demographics coalesced across the C2M2 columns and HuBMAP's +-- parallel enrichment, with a *_source column per value so provenance +-- survives the coalesce. The loader never invents values; this view is +-- where the triplicated facts are reconciled, visibly. +CREATE VIEW v_subject_demographics AS +SELECT + s.id_namespace, + s.local_id, + s.submission, + COALESCE(s.sex, xh.sex) AS sex, + CASE + WHEN s.sex IS NOT NULL THEN 'subject' + WHEN xh.sex IS NOT NULL THEN 'subject_extra_hubmap' + END AS sex_source, + COALESCE(s.age_at_enrollment, s.age_at_sampling, xh.age_value) AS age, + CASE + WHEN s.age_at_enrollment IS NOT NULL THEN 'subject.age_at_enrollment' + WHEN s.age_at_sampling IS NOT NULL THEN 'subject.age_at_sampling' + WHEN xh.age_value IS NOT NULL THEN 'subject_extra_hubmap.age_value' + END AS age_source, + -- The unit belongs to HuBMAP's age_value alone; a C2M2-sourced age + -- (years by definition) must not be paired with it. + CASE + WHEN s.age_at_enrollment IS NULL AND s.age_at_sampling IS NULL + THEN xh.age_unit + END AS age_unit, + COALESCE(r.races, CASE WHEN xh.race IS NOT NULL THEN [xh.race] END) AS race, + CASE + WHEN r.races IS NOT NULL THEN 'subject_race' + WHEN xh.race IS NOT NULL THEN 'subject_extra_hubmap' + END AS race_source, + s.ethnicity +FROM subject AS s +LEFT JOIN subject_extra_hubmap AS xh + ON xh.id_namespace = s.id_namespace AND xh.local_id = s.local_id +LEFT JOIN ( + SELECT subject_id_namespace, subject_local_id, + list(race ORDER BY race) AS races + FROM subject_race + GROUP BY subject_id_namespace, subject_local_id +) AS r + ON r.subject_id_namespace = s.id_namespace + AND r.subject_local_id = s.local_id; From 911b420a2ce24a4eea42f94f1f80220c3ec86af2 Mon Sep 17 00:00:00 2001 From: Conrad Date: Tue, 27 Oct 2026 10:43:12 -0400 Subject: [PATCH 3/4] test: Cover the DuckDB dump loader, schema, views, and CLI Exercise the public surface end to end against a real in-memory DuckDB: schema bootstrap, flattening and junction recovery, first-wins entity dedup with divergence sampling, junction union semantics, both DumpError abort conditions, coercion and provenance accounting, stub project upgrades, JSON and list column fidelity, all three views, and the full CLI option surface including partial-output cleanup. Two Hypothesis properties pin the cardinality invariants under arbitrary file/collection sharing and byte-exact text roundtrips through model validation and parameterized insert. --- tests/test_cli.py | 215 +++++++ tests/test_dump.py | 1433 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 1648 insertions(+) create mode 100644 tests/test_cli.py create mode 100644 tests/test_dump.py diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..e797cbc --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,215 @@ +"""Tests for the cfdb CLI commands.""" + +from __future__ import annotations + +import duckdb +from click.testing import CliRunner + +import cfdb.cli as cli_module +from tests.test_dump import make_fourdn_doc + + +def patch_mongo(mocker, docs: list[dict]): + """Patch the CLI's MongoClient with a client whose find() yields docs.""" + fake_collection = mocker.MagicMock() + fake_collection.find.return_value = docs + fake_client = mocker.MagicMock() + fake_client.__getitem__.return_value = {"files": fake_collection} + mocker.patch.object(cli_module, "MongoClient", return_value=fake_client) + return fake_collection + + +def test_dump_duckdb_should_fail_when_duckdb_missing(mocker, tmp_path): + """Test the CLI's missing-extra diagnostic. + + Given: + An environment where the duckdb module is not importable. + When: + cfdb dump duckdb is invoked. + Then: + It should exit nonzero and name the cfdb[duckdb] extra. + """ + # Arrange + mocker.patch.dict("sys.modules", {"duckdb": None}) + runner = CliRunner() + + # Act + result = runner.invoke(cli_module.cli, ["dump", "duckdb", str(tmp_path / "out.duckdb")]) + + # Assert + assert result.exit_code == 1 + assert "cfdb[duckdb]" in result.output + + +def test_dump_duckdb_should_write_database_when_mongo_returns_documents(mocker, tmp_path): + """Test the CLI wiring from Mongo cursor to DuckDB file. + + Given: + A MongoClient whose files collection yields one document. + When: + cfdb dump duckdb is invoked with a submission filter. + Then: + It should query Mongo with that filter and write a DuckDB file + holding the document. + """ + # Arrange + fake_collection = patch_mongo(mocker, [make_fourdn_doc()]) + output = tmp_path / "out.duckdb" + runner = CliRunner() + + # Act + result = runner.invoke( + cli_module.cli, ["dump", "duckdb", str(output), "-s", "4DN"] + ) + + # Assert + assert result.exit_code == 0, result.output + fake_collection.find.assert_called_once_with( + {"submission": {"$in": ["4dn"]}}, batch_size=1000 + ) + con = duckdb.connect(str(output)) + assert con.execute("SELECT count(*) FROM file").fetchone()[0] == 1 + con.close() + + +def test_dump_duckdb_should_refuse_existing_output_without_force(tmp_path): + """Test the CLI's overwrite guard. + + Given: + An OUTPUT path that already exists. + When: + cfdb dump duckdb is invoked without --force. + Then: + It should exit nonzero without touching the file. + """ + # Arrange + output = tmp_path / "out.duckdb" + output.write_bytes(b"not a database") + runner = CliRunner() + + # Act + result = runner.invoke(cli_module.cli, ["dump", "duckdb", str(output)]) + + # Assert + assert result.exit_code == 1 + assert "--force" in result.output + assert output.read_bytes() == b"not a database" + + +def test_dump_duckdb_should_overwrite_existing_output_when_forced(mocker, tmp_path): + """Test the --force overwrite path. + + Given: + An OUTPUT path that already exists and a Mongo yielding one + document. + When: + cfdb dump duckdb is invoked with --force. + Then: + It should replace the file with a valid DuckDB holding the + document. + """ + # Arrange + patch_mongo(mocker, [make_fourdn_doc()]) + output = tmp_path / "out.duckdb" + output.write_bytes(b"not a database") + runner = CliRunner() + + # Act + result = runner.invoke(cli_module.cli, ["dump", "duckdb", str(output), "--force"]) + + # Assert + assert result.exit_code == 0, result.output + con = duckdb.connect(str(output)) + assert con.execute("SELECT count(*) FROM file").fetchone()[0] == 1 + con.close() + + +def test_dump_duckdb_should_export_parquet_when_requested(mocker, tmp_path): + """Test the --to-parquet export path. + + Given: + A Mongo yielding one document and a --to-parquet directory. + When: + cfdb dump duckdb is invoked. + Then: + It should name the export in its output and leave Parquet files in + the directory. + """ + # Arrange + patch_mongo(mocker, [make_fourdn_doc()]) + output = tmp_path / "out.duckdb" + parquet_dir = tmp_path / "parquet" + runner = CliRunner() + + # Act + result = runner.invoke( + cli_module.cli, + ["dump", "duckdb", str(output), "--to-parquet", str(parquet_dir)], + ) + + # Assert + assert result.exit_code == 0, result.output + assert "Exported Parquet" in result.output + assert list(parquet_dir.glob("*.parquet")) + + +def test_dump_duckdb_should_keep_database_when_parquet_export_fails(mocker, tmp_path): + """Test that a failed export never destroys the completed dump. + + Given: + A Mongo yielding one document and a --to-parquet target inside an + unwritable directory. + When: + cfdb dump duckdb is invoked. + Then: + It should exit nonzero naming the export failure while the + completed .duckdb stays on disk. + """ + # Arrange + patch_mongo(mocker, [make_fourdn_doc()]) + output = tmp_path / "out.duckdb" + readonly_parent = tmp_path / "readonly" + readonly_parent.mkdir() + readonly_parent.chmod(0o500) + parquet_target = readonly_parent / "parquet" + runner = CliRunner() + + # Act + try: + result = runner.invoke( + cli_module.cli, + ["dump", "duckdb", str(output), "--to-parquet", str(parquet_target)], + ) + finally: + readonly_parent.chmod(0o700) + + # Assert + assert result.exit_code != 0 + assert "Parquet export" in result.output + assert "was kept" in result.output + assert output.exists() + + +def test_dump_duckdb_should_remove_partial_output_when_dump_fails(mocker, tmp_path): + """Test partial-output cleanup on a failed dump. + + Given: + A Mongo yielding two documents with duplicate file keys. + When: + cfdb dump duckdb is invoked. + Then: + It should exit nonzero, print the DumpError message, and remove the + partial output file. + """ + # Arrange + patch_mongo(mocker, [make_fourdn_doc("file-1"), make_fourdn_doc("file-1")]) + output = tmp_path / "out.duckdb" + runner = CliRunner() + + # Act + result = runner.invoke(cli_module.cli, ["dump", "duckdb", str(output)]) + + # Assert + assert result.exit_code == 1 + assert "duplicate file keys" in result.output + assert not output.exists() diff --git a/tests/test_dump.py b/tests/test_dump.py new file mode 100644 index 0000000..9daad2b --- /dev/null +++ b/tests/test_dump.py @@ -0,0 +1,1433 @@ +"""Tests for the relational DuckDB dump of the ``files`` collection.""" + +from __future__ import annotations + +import json +import subprocess +from datetime import datetime + +import duckdb +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from cfdb.dump import SCHEMA_VERSION, DumpError, DumpReport, dump_files + +#: Every table schema.sql creates, by tier. +PROVENANCE_TABLES = {"dump_info", "dump_stat", "dump_conflict"} +ENTITY_TABLES = { + "dcc", + "project", + "file_format", + "data_type", + "assay_type", + "anatomy", + "ncbi_taxonomy", + "file", + "collection", + "biosample", + "subject", +} +JUNCTION_TABLES = { + "file_in_collection", + "biosample_in_collection", + "subject_in_collection", + "biosample_from_subject", + "collection_anatomy", + "subject_race", +} +EXTENSION_TABLES = { + "file_extra_fourdn", + "file_extra_fourdn_extra_file", + "file_extra_encode", + "file_extra_hubmap", + "collection_extra_fourdn", + "collection_extra_encode", + "collection_extra_hubmap", + "biosample_extra_encode", + "subject_extra_hubmap", +} +ALL_TABLES = PROVENANCE_TABLES | ENTITY_TABLES | JUNCTION_TABLES | EXTENSION_TABLES +ALL_VIEWS = {"v_file", "v_file_biosample", "v_subject_demographics"} + +FOURDN_NS = "tag:4dn.org,2015:" +ENCODE_NS = "tag:encode.org,2020:" +HUBMAP_NS = "tag:hubmapconsortium.org,2023:" + + +def make_dcc(submission: str) -> dict: + """Build the embedded dcc object for a document.""" + return { + "id": f"cfde_registry_dcc:{submission}", + "dcc_name": f"{submission} name", + "dcc_abbreviation": submission.upper(), + "contact_email": "contact@example.org", + "contact_name": "Contact", + "dcc_url": f"https://{submission}.example.org", + "project_id_namespace": "", + "project_local_id": "", + } + + +def make_subject(ns: str, local_id: str, **overrides) -> dict: + """Build a nested subject object.""" + subject = { + "id_namespace": ns, + "local_id": local_id, + "sex": "female", + "race": ["asian"], + "taxonomy": { + "id": "NCBI:txid9606", + "name": "Homo sapiens", + "clade": "species", + }, + } + subject.update(overrides) + return subject + + +def make_collection(ns: str, local_id: str, **overrides) -> dict: + """Build a nested collection object with one biosample and one subject.""" + collection = { + "id_namespace": ns, + "local_id": local_id, + "name": f"Collection {local_id}", + "lab": "Lab A", + "experiment_type": "in situ Hi-C", + "anatomy": [{"id": "UBERON:0000948", "name": "heart"}], + "subjects": [make_subject(ns, f"{local_id}-subj")], + "biosamples": [ + { + "id_namespace": ns, + "local_id": f"{local_id}-bio", + "anatomy": {"id": "UBERON:0000948", "name": "heart"}, + "subjects": [make_subject(ns, f"{local_id}-subj")], + } + ], + } + collection.update(overrides) + return collection + + +def make_fourdn_doc(local_id: str = "file-1", **overrides) -> dict: + """Build a complete 4DN files document.""" + doc = { + "submission": "4dn", + "id_namespace": FOURDN_NS, + "local_id": local_id, + "accession_id": "4dnfiabc123", + "filename": f"{local_id}.mcool", + "creation_time": "2021-03-04T05:06:07+00:00", + "size_in_bytes": "123456", + "md5": "aa", + "project_id_namespace": FOURDN_NS, + "project_local_id": "proj-1", + "genome_assembly": "GRCh38", + "output_type": "contact matrix", + "output_type_detail": "contact matrix (mcool)", + "assay_info": "Dilution Hi-C", + "condition": "control", + "dcc": make_dcc("4dn"), + "project": { + "id_namespace": FOURDN_NS, + "local_id": "proj-1", + "name": "Project One", + }, + "file_format": {"id": "format:3590", "name": "HDF5", "description": "d"}, + "extra": { + "fourdn": { + "enriched_file_format": "mcool", + "file_type": "contact matrix", + "file_type_detailed": "contact matrix (mcool)", + "genome_assembly": "GRCh38", + "replicate_info": "Biorep 1, Techrep 1", + "extra_files": [ + { + "href": "/x", + "md5sum": "bb", + "file_size": "42", + "file_format": {"display_title": "pairs_px2"}, + } + ], + } + }, + "collections": [ + make_collection( + FOURDN_NS, + "exp-1", + accession_id="4dnexabc", + extra={ + "fourdn": { + "display_title": "Exp 1", + "targeted_factor": ["CTCF protein"], + "digestion_enzyme": "DpnII", + "date_created": "2020-01-02T03:04:05+00:00", + } + }, + ) + ], + } + doc.update(overrides) + return doc + + +def make_encode_doc(local_id: str = "ENCFF001AAA", **overrides) -> dict: + """Build a complete ENCODE files document.""" + doc = { + "submission": "encode", + "id_namespace": ENCODE_NS, + "local_id": local_id, + "accession_id": local_id.lower(), + "filename": f"{local_id}.bed.gz", + "size_in_bytes": 6262125716, + "sha256": "cc", + "genome_assembly": "GRCh38", + "genome_annotation": "V29", + "output_type": "peaks", + "output_type_detail": "narrowPeak", + "biological_replicates": "1", + "technical_replicates": "1_1", + "dcc": make_dcc("encode"), + "file_format": {"id": "format:3003", "name": "BED"}, + "extra": { + "encode": { + "assembly": "GRCh38", + "output_type": "peaks", + "s3_uri": "s3://encode-public/x.bed.gz", + "run_type": "single-ended", + "audit_warning": "mild to moderate bottlenecking", + } + }, + "collections": [ + make_collection( + ENCODE_NS, + "ENCSR001AAA", + experiment_type="TF ChIP-seq", + experiment_target="CTCF", + analyte_class="DNA", + extra={ + "encode": { + "project": "ENCODE", + "platform": "Illumina NovaSeq 6000", + } + }, + biosamples=[ + { + "id_namespace": ENCODE_NS, + "local_id": "ENCBS001AAA", + "anatomy": {"id": "UBERON:0002107", "name": "liver"}, + "subjects": [make_subject(ENCODE_NS, "ENCDO001AAA")], + "extra": { + "encode": { + "biosample_type": "tissue", + "life_stage": "adult", + "library_made_from": "DNA", + } + }, + } + ], + ) + ], + } + doc.update(overrides) + return doc + + +def make_hubmap_doc(local_id: str = "hm-file-1", **overrides) -> dict: + """Build a complete HuBMAP files document.""" + doc = { + "submission": "hubmap", + "id_namespace": HUBMAP_NS, + "local_id": local_id, + "filename": f"{local_id}.tsv", + "size_in_bytes": "999", + "md5": "dd", + "status": "Published", + "data_access_level": "public", + "genome_assembly": "GRCh38", + "dcc": make_dcc("hubmap"), + "extra": { + "hubmap": { + "rel_path": "expr/some.tsv", + "is_data_product": True, + } + }, + "collections": [ + make_collection( + HUBMAP_NS, + "HBM123.ABCD.456", + experiment_type="RNAseq", + analyte_class="RNA", + extra={ + "hubmap": { + "pipeline": "salmon-rnaseq", + "processing": "processed", + "visualization": True, + "vitessce_hints": ["is_sc", "rna"], + "metadata": {"assay_category": "sequence"}, + } + }, + subjects=[ + make_subject( + HUBMAP_NS, + "donor-1", + sex=None, + race=[], + extra={ + "hubmap": { + "sex": "Male", + "race": "White", + "age_value": 61.0, + "age_unit": "years", + } + }, + ) + ], + biosamples=[ + { + "id_namespace": HUBMAP_NS, + "local_id": "HBM123.ABCD.456-bio", + "anatomy": {"id": "UBERON:0002113", "name": "kidney"}, + "subjects": [ + make_subject( + HUBMAP_NS, + "donor-1", + sex=None, + race=[], + extra={ + "hubmap": { + "sex": "Male", + "race": "White", + "age_value": 61.0, + "age_unit": "years", + } + }, + ) + ], + } + ], + ) + ], + } + doc.update(overrides) + return doc + + +def run_dump(docs: list[dict], **kwargs) -> tuple[duckdb.DuckDBPyConnection, DumpReport]: + """Dump documents into a fresh in-memory DuckDB and return it.""" + con = duckdb.connect(":memory:") + report = dump_files(docs, con, source_database="cfdb", **kwargs) + return con, report + + +class TestDumpReport: + def test_summary_should_render_sections_when_conflicts_and_counters_present(self): + """Test the populated summary rendering. + + Given: + A DumpReport with row counts, a non-empty conflict tally, and a + non-zero counter. + When: + summary() is called. + Then: + It should render the per-table row lines plus the divergence and + counter sections with their values. + """ + # Arrange + report = DumpReport( + file_count=2, + rows={"file": 2, "collection": 1}, + conflicts={"collection": 1}, + stats={"documents_invalid": 3}, + ) + + # Act + summary = report.summary() + + # Assert + assert "files dumped: 2" in summary + assert "file: 2 rows" in summary + assert "divergent entity copies (first write kept):" in summary + assert "collection: 1" in summary + assert "counters:" in summary + assert "documents_invalid: 3" in summary + + def test_summary_should_omit_sections_when_no_conflicts_or_counters(self): + """Test empty-section suppression in the summary. + + Given: + A DumpReport with no conflicts and only zero-valued counters. + When: + summary() is called. + Then: + It should omit both the divergence and counter section headers. + """ + # Arrange + report = DumpReport( + file_count=1, + rows={"file": 1}, + conflicts={}, + stats={"documents_invalid": 0}, + ) + + # Act + summary = report.summary() + + # Assert + assert "divergent" not in summary + assert "counters:" not in summary + + +def test_dump_files_should_create_all_tables(): + """Test schema bootstrap. + + Given: + No documents at all. + When: + dump_files runs against a fresh in-memory DuckDB. + Then: + It should create every table and view the schema of record defines. + """ + # Act + con, _ = run_dump([]) + + # Assert + tables = {row[0] for row in con.execute("SELECT table_name FROM duckdb_tables()").fetchall()} + views = { + row[0] + for row in con.execute( + "SELECT view_name FROM duckdb_views() WHERE NOT internal" + ).fetchall() + } + assert tables == ALL_TABLES + assert ALL_VIEWS <= views + + +def test_dump_files_should_flatten_nested_entities_when_doc_spans_all_levels(): + """Test flattening of one fully nested document. + + Given: + One 4DN document carrying a collection with anatomy, a biosample, + and a subject shared between the collection and the biosample. + When: + The document is dumped. + Then: + It should recover one row per entity and one row per C2M2 junction. + """ + # Arrange + doc = make_fourdn_doc() + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.file_count == 1 + for table in ( + "file", + "collection", + "biosample", + "subject", + "file_in_collection", + "biosample_in_collection", + "subject_in_collection", + "biosample_from_subject", + "collection_anatomy", + "subject_race", + "file_extra_fourdn", + "file_extra_fourdn_extra_file", + "collection_extra_fourdn", + ): + assert report.rows[table] == 1, table + + +def test_dump_files_should_collapse_repeated_entities_when_shared_across_files(): + """Test entity dedup across documents. + + Given: + Two 4DN documents embedding byte-identical copies of the same + collection. + When: + Both documents are dumped. + Then: + It should write one collection row, two file rows, and two + file_in_collection rows, with no conflicts recorded. + """ + # Arrange + docs = [make_fourdn_doc("file-1"), make_fourdn_doc("file-2")] + + # Act + con, report = run_dump(docs) + + # Assert + assert report.rows["file"] == 2 + assert report.rows["collection"] == 1 + assert report.rows["file_in_collection"] == 2 + assert report.conflicts == {} + + +def test_dump_files_should_record_conflict_when_repeated_entity_diverges(): + """Test divergence reporting on repeated entities. + + Given: + Two documents embedding the same collection key with different + lab values. + When: + Both documents are dumped. + Then: + It should keep the first copy's value and record the divergence + in dump_conflict rather than silently resolving it. + """ + # Arrange + first = make_fourdn_doc("file-1") + second = make_fourdn_doc("file-2") + second["collections"][0]["lab"] = "Lab B" + + # Act + con, report = run_dump([first, second]) + + # Assert + assert report.conflicts["collection"] == 1 + assert con.execute("SELECT lab FROM collection").fetchone()[0] == "Lab A" + conflict = con.execute( + "SELECT table_name, column_name, kept_value, discarded_value FROM dump_conflict" + ).fetchone() + assert conflict == ("collection", "lab", "Lab A", "Lab B") + assert con.execute( + "SELECT value FROM dump_stat WHERE name = 'conflicts_collection'" + ).fetchone()[0] == 1 + + +def test_dump_files_should_union_junction_rows_when_copies_disagree(): + """Test union semantics for junction rows. + + Given: + Two documents whose copies of one collection carry different + biosample lists. + When: + Both documents are dumped. + Then: + It should keep the union of biosample_in_collection rows, so a + relationship present in only one copy is not silently dropped. + """ + # Arrange + first = make_fourdn_doc("file-1") + second = make_fourdn_doc("file-2") + second["collections"][0]["biosamples"].append( + { + "id_namespace": FOURDN_NS, + "local_id": "exp-1-bio-extra", + "anatomy": {"id": "UBERON:0000948", "name": "heart"}, + "subjects": [], + } + ) + + # Act + con, report = run_dump([first, second]) + + # Assert + assert report.rows["biosample"] == 2 + assert report.rows["biosample_in_collection"] == 2 + + +def test_dump_files_should_raise_when_key_maps_to_two_submissions(): + """Test the key-uniqueness assertion. + + Given: + Two documents embedding the same collection key under two + different submissions. + When: + Both documents are dumped. + Then: + It should raise DumpError naming both submissions, because the + schema's global natural keys no longer hold. + """ + # Arrange + first = make_fourdn_doc("file-1") + second = make_encode_doc("ENCFF001AAA") + second["collections"][0]["id_namespace"] = FOURDN_NS + second["collections"][0]["local_id"] = "exp-1" + + # Act & assert + with pytest.raises(DumpError, match="two submissions"): + run_dump([first, second]) + + +def test_dump_files_should_raise_when_duplicate_file_keys(): + """Test the duplicate-file-key assertion. + + Given: + Two documents carrying the same (id_namespace, local_id) file key. + When: + Both documents are dumped. + Then: + It should raise DumpError rather than write an ambiguous file row. + """ + # Arrange + docs = [make_fourdn_doc("file-1"), make_fourdn_doc("file-1")] + + # Act & assert + with pytest.raises(DumpError, match="duplicate file keys"): + run_dump(docs) + + +def test_dump_files_should_enforce_foreign_keys_when_junction_names_absent_parent(): + """Test that the schema's declared foreign keys are enforced. + + Given: + A completed dump and a file_in_collection row naming a collection + no document carried. + When: + The row is inserted directly. + Then: + It should raise ConstraintException, because the fails-at-write-time + promise rests on enforcement rather than declaration. + """ + # Arrange + con, _ = run_dump([make_fourdn_doc()]) + + # Act & assert + with pytest.raises(duckdb.ConstraintException): + con.execute( + "INSERT INTO file_in_collection VALUES (?, ?, ?, ?)", + [FOURDN_NS, "file-1", FOURDN_NS, "no-such-collection"], + ) + + +def test_dump_files_should_raise_when_stub_upgraded_across_submissions(): + """Test the key-uniqueness assertion on the stub-upgrade path. + + Given: + A document stubbing a project via a biosample FK, followed by a + document from another submission carrying the real project object + for the same key. + When: + Both documents are dumped. + Then: + It should raise DumpError naming both submissions instead of + silently upgrading the stub across consortia. + """ + # Arrange + first = make_fourdn_doc(project=None, project_id_namespace="", project_local_id="") + first["collections"][0]["biosamples"][0]["project_id_namespace"] = FOURDN_NS + first["collections"][0]["biosamples"][0]["project_local_id"] = "proj-x" + second = make_encode_doc( + project={"id_namespace": FOURDN_NS, "local_id": "proj-x", "name": "P"}, + project_id_namespace=FOURDN_NS, + project_local_id="proj-x", + ) + + # Act & assert + with pytest.raises(DumpError, match="two submissions"): + run_dump([first, second]) + + +def test_dump_files_should_raise_when_stub_references_real_row_across_submissions(): + """Test the key-uniqueness assertion on the incoming-stub path. + + Given: + A document carrying a real project object, followed by a document + from another submission whose biosample FK stubs the same key. + When: + Both documents are dumped. + Then: + It should raise DumpError naming both submissions instead of + silently dropping the cross-consortium stub. + """ + # Arrange + first = make_fourdn_doc() + second = make_encode_doc() + second["collections"][0]["biosamples"][0]["project_id_namespace"] = FOURDN_NS + second["collections"][0]["biosamples"][0]["project_local_id"] = "proj-1" + + # Act & assert + with pytest.raises(DumpError, match="two submissions"): + run_dump([first, second]) + + +def test_dump_files_should_coerce_string_size_when_source_is_tsv_typed(): + """Test numeric coercion of string-typed sizes. + + Given: + A 4DN document whose size_in_bytes and sidecar file_size are + strings, as the C2M2 TSV path loads them. + When: + The document is dumped. + Then: + It should store both as integers with no coercion failures + counted. + """ + # Arrange + doc = make_fourdn_doc() + + # Act + con, report = run_dump([doc]) + + # Assert + assert con.execute("SELECT size_in_bytes FROM file").fetchone()[0] == 123456 + assert ( + con.execute("SELECT file_size FROM file_extra_fourdn_extra_file").fetchone()[0] + == 42 + ) + assert "size_in_bytes_uncoercible" not in report.stats + + +def test_dump_files_should_count_failure_when_size_uncoercible(): + """Test coercion-failure accounting. + + Given: + A document whose size_in_bytes is a non-numeric string. + When: + The document is dumped. + Then: + It should store NULL, keep the document, and count the failure in + dump_stat. + """ + # Arrange + doc = make_fourdn_doc(size_in_bytes="N/A") + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.file_count == 1 + assert con.execute("SELECT size_in_bytes FROM file").fetchone()[0] is None + assert report.stats["size_in_bytes_uncoercible"] == 1 + assert con.execute( + "SELECT value FROM dump_stat WHERE name = 'size_in_bytes_uncoercible'" + ).fetchone()[0] == 1 + + +def test_dump_files_should_null_term_fk_when_term_id_empty(): + """Test the term-id guard on the file-level FK columns. + + Given: + A document whose file_format term carries an empty id. + When: + The document is dumped. + Then: + It should complete with file.file_format NULL, no file_format row, + and the skip counted in dump_stat. + """ + # Arrange + doc = make_fourdn_doc(file_format={"id": "", "name": "HDF5"}) + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.file_count == 1 + assert con.execute("SELECT file_format FROM file").fetchone()[0] is None + assert con.execute("SELECT count(*) FROM file_format").fetchone()[0] == 0 + assert report.stats["term_id_missing"] == 1 + + +def test_dump_files_should_skip_term_and_null_fk_when_term_name_empty(): + """Test the nameless-term guard. + + Given: + A document whose file_format term carries an id but a blank name. + When: + The document is dumped. + Then: + It should complete with the term skipped, file.file_format NULL, + and the skip counted in dump_stat, because a nameless term cannot + satisfy the NOT NULL name column. + """ + # Arrange + doc = make_fourdn_doc(file_format={"id": "format:3590", "name": ""}) + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.file_count == 1 + assert con.execute("SELECT file_format FROM file").fetchone()[0] is None + assert con.execute("SELECT count(*) FROM file_format").fetchone()[0] == 0 + assert report.stats["term_name_missing"] == 1 + + +def test_dump_files_should_populate_promoted_columns_when_consortium_fills_them(): + """Test the promoted cross-consortium common columns. + + Given: + One document per consortium, each filling the common slots its + ingest promotes. + When: + All three are dumped together. + Then: + It should populate each file and collection common column from + that consortium's source field. + """ + # Arrange + docs = [make_fourdn_doc(), make_encode_doc(), make_hubmap_doc()] + + # Act + con, report = run_dump(docs) + + # Assert + rows = { + submission: (assembly, annotation, output_type, condition) + for submission, assembly, annotation, output_type, condition in con.execute( + 'SELECT submission, genome_assembly, genome_annotation, output_type, "condition" FROM file' + ).fetchall() + } + assert rows["4dn"] == ("GRCh38", None, "contact matrix", "control") + assert rows["encode"] == ("GRCh38", "V29", "peaks", None) + assert rows["hubmap"] == ("GRCh38", None, None, None) + collections = { + submission: (experiment_type, experiment_target, analyte_class) + for submission, experiment_type, experiment_target, analyte_class in con.execute( + "SELECT submission, experiment_type, experiment_target, analyte_class FROM collection" + ).fetchall() + } + assert collections["4dn"] == ("in situ Hi-C", None, None) + assert collections["encode"] == ("TF ChIP-seq", "CTCF", "DNA") + assert collections["hubmap"] == ("RNAseq", None, "RNA") + + +def test_dump_files_should_omit_promoted_fields_from_extension_tables(): + """Test that extension tables carry only the unpromoted residue. + + Given: + The dumped schema. + When: + The extension tables' columns are inspected. + Then: + It should omit verbatim-promoted fields and keep replicate_info, + whose promotion is derived rather than verbatim. + """ + # Arrange + con, _ = run_dump([make_fourdn_doc()]) + + # Act + columns = { + row[0] + for row in con.execute( + "SELECT column_name FROM duckdb_columns() WHERE table_name = 'file_extra_fourdn'" + ).fetchall() + } + + # Assert + assert "replicate_info" in columns + assert "file_type" not in columns + assert "file_type_detailed" not in columns + assert "genome_assembly" not in columns + + +def test_dump_files_should_skip_and_count_invalid_documents(): + """Test invalid-document handling. + + Given: + One valid document and one document with no dcc. + When: + Both are dumped. + Then: + It should load the valid document, skip the invalid one, and + count the skip in dump_stat. + """ + # Arrange + invalid = make_fourdn_doc("file-2") + del invalid["dcc"] + + # Act + con, report = run_dump([make_fourdn_doc("file-1"), invalid]) + + # Assert + assert report.file_count == 1 + assert report.stats["documents_invalid"] == 1 + assert con.execute( + "SELECT value FROM dump_stat WHERE name = 'documents_invalid'" + ).fetchone()[0] == 1 + + +def test_dump_files_should_write_provenance(): + """Test the dump_info provenance row. + + Given: + One document and an explicit submissions filter. + When: + The document is dumped. + Then: + It should write one dump_info row carrying the schema version, + the filter, and the file count. + """ + # Arrange + doc = make_fourdn_doc() + + # Act + con, _ = run_dump([doc], submissions=["4dn"]) + + # Assert + row = con.execute( + "SELECT schema_version, source_database, source_collection, submissions, file_count" + " FROM dump_info" + ).fetchone() + assert row == (SCHEMA_VERSION, "cfdb", "files", ["4dn"], 1) + + +def test_dump_files_should_not_record_foreign_repo_commit_when_run_from_one( + tmp_path, monkeypatch +): + """Test the cfdb_commit provenance anchor. + + Given: + The process CWD pointed at an unrelated git repository with a + known HEAD. + When: + A document is dumped. + Then: + It should record either cfdb's own commit or NULL in + dump_info.cfdb_commit, never the unrelated repository's HEAD. + """ + # Arrange + env = {"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"} + subprocess.run(["git", "init", "-q"], cwd=tmp_path, check=True) + subprocess.run( + ["git", "commit", "-q", "--allow-empty", "-m", "scratch"], + cwd=tmp_path, + check=True, + env={**env, "HOME": str(tmp_path), "GIT_CONFIG_GLOBAL": "/dev/null"}, + ) + scratch_sha = subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=tmp_path, capture_output=True, text=True + ).stdout.strip() + monkeypatch.chdir(tmp_path) + + # Act + con, _ = run_dump([make_fourdn_doc()]) + + # Assert + recorded = con.execute("SELECT cfdb_commit FROM dump_info").fetchone()[0] + assert recorded != scratch_sha + + +def test_v_subject_demographics_should_coalesce_with_source_when_hubmap_extra_present(): + """Test provenance-carrying coalescing of subject demographics. + + Given: + A HuBMAP subject whose C2M2 sex and race are empty but whose + enrichment carries both. + When: + v_subject_demographics is queried. + Then: + It should surface the enrichment values with source columns + naming subject_extra_hubmap. + """ + # Arrange + con, _ = run_dump([make_hubmap_doc()]) + + # Act + row = con.execute( + "SELECT sex, sex_source, race, race_source, age, age_source" + " FROM v_subject_demographics" + ).fetchone() + + # Assert + assert row == ( + "Male", + "subject_extra_hubmap", + ["White"], + "subject_extra_hubmap", + 61.0, + "subject_extra_hubmap.age_value", + ) + + +def test_v_subject_demographics_should_null_age_unit_when_age_from_c2m2_column(): + """Test the source-consistency of the age_unit column. + + Given: + A subject carrying both a C2M2 age_at_enrollment and divergent + HuBMAP enrichment age_value and age_unit. + When: + v_subject_demographics is queried. + Then: + It should pair the C2M2 age with a NULL age_unit, because the + unit belongs to HuBMAP's age_value alone. + """ + # Arrange + doc = make_hubmap_doc() + for subject in ( + doc["collections"][0]["subjects"][0], + doc["collections"][0]["biosamples"][0]["subjects"][0], + ): + subject["age_at_enrollment"] = 30.0 + subject["extra"]["hubmap"]["age_value"] = 360.0 + subject["extra"]["hubmap"]["age_unit"] = "months" + con, _ = run_dump([doc]) + + # Act + row = con.execute( + "SELECT age, age_source, age_unit FROM v_subject_demographics" + ).fetchone() + + # Assert + assert row == (30.0, "subject.age_at_enrollment", None) + + +@settings(max_examples=25, deadline=None, suppress_health_check=[HealthCheck.too_slow]) +@given( + memberships=st.lists( + st.sets(st.integers(min_value=0, max_value=3), min_size=1, max_size=3), + min_size=1, + max_size=8, + ) +) +def test_dump_files_should_preserve_cardinalities_with_arbitrary_sharing(memberships): + """Test flattening invariants over arbitrary file/collection sharing. + + Given: + Generated batches of files, each a member of an arbitrary + non-empty subset of a shared collection pool. + When: + The batch is dumped. + Then: + It should write exactly one row per distinct file and collection, + and one file_in_collection row per membership. + """ + # Arrange + docs = [] + for index, collections in enumerate(memberships): + doc = make_fourdn_doc(f"file-{index}") + doc["collections"] = [ + make_collection(FOURDN_NS, f"exp-{c}") for c in sorted(collections) + ] + docs.append(doc) + expected_collections = {c for collections in memberships for c in collections} + expected_memberships = sum(len(collections) for collections in memberships) + + # Act + con, report = run_dump(docs) + + # Assert + assert report.rows["file"] == len(memberships) + assert report.rows["collection"] == len(expected_collections) + assert report.rows["file_in_collection"] == expected_memberships + + +def test_dump_files_should_null_and_count_when_timestamps_unparseable(): + """Test the timestamp coercion-failure policy. + + Given: + A 4DN document whose file creation_time and collection-extra + date_created are non-ISO strings. + When: + The document is dumped. + Then: + It should store NULL in both columns, keep the document, and count + each failure in the report stats and dump_stat. + """ + # Arrange + doc = make_fourdn_doc(creation_time="not-a-date") + doc["collections"][0]["extra"]["fourdn"]["date_created"] = "also-not-a-date" + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.file_count == 1 + assert con.execute("SELECT creation_time FROM file").fetchone()[0] is None + assert ( + con.execute("SELECT date_created FROM collection_extra_fourdn").fetchone()[0] + is None + ) + assert report.stats["creation_time_unparseable"] == 1 + assert report.stats["date_created_unparseable"] == 1 + counters = dict( + con.execute( + "SELECT name, value FROM dump_stat WHERE name LIKE '%_unparseable'" + ).fetchall() + ) + assert counters == { + "creation_time_unparseable": 1, + "date_created_unparseable": 1, + } + + +def test_dump_files_should_normalize_to_utc_when_creation_time_offset(): + """Test timezone normalization of TIMESTAMP columns. + + Given: + A document whose creation_time carries a +05:00 offset. + When: + The document is dumped. + Then: + It should store the UTC-equivalent naive timestamp. + """ + # Arrange + doc = make_fourdn_doc(creation_time="2021-03-04T10:06:07+05:00") + + # Act + con, _ = run_dump([doc]) + + # Assert + stored = con.execute("SELECT creation_time FROM file").fetchone()[0] + assert stored == datetime(2021, 3, 4, 5, 6, 7) + + +def test_dump_files_should_dedupe_terms_globally_when_descriptions_diverge(): + """Test ontology global dedup and divergence reporting. + + Given: + A 4DN and an ENCODE document sharing one file_format id with + divergent descriptions. + When: + Both documents are dumped. + Then: + It should write one globally deduped term row keeping the first + description, and record the divergence in dump_conflict rather + than raising. + """ + # Arrange + first = make_fourdn_doc( + file_format={"id": "format:3003", "name": "BED", "description": "A"} + ) + second = make_encode_doc() + second["file_format"] = {"id": "format:3003", "name": "BED", "description": "B"} + + # Act + con, report = run_dump([first, second]) + + # Assert + assert report.rows["file_format"] == 1 + assert con.execute("SELECT description FROM file_format").fetchone()[0] == "A" + assert report.conflicts["file_format"] == 1 + conflict = con.execute( + "SELECT table_name, natural_key, column_name, kept_value, discarded_value" + " FROM dump_conflict" + ).fetchone() + assert conflict == ("file_format", "format:3003", "description", "A", "B") + + +def test_dump_files_should_write_project_stub_when_fk_has_no_object(): + """Test the project stub-row policy. + + Given: + A document with no joined project object whose biosample carries a + project foreign key. + When: + The document is dumped. + Then: + It should write an all-NULL-attribute stub project row keyed by the + FK, count it, and keep foreign-key integrity. + """ + # Arrange + doc = make_fourdn_doc(project=None, project_id_namespace="", project_local_id="") + doc["collections"][0]["biosamples"][0]["project_id_namespace"] = FOURDN_NS + doc["collections"][0]["biosamples"][0]["project_local_id"] = "proj-stub" + + # Act + con, report = run_dump([doc]) + + # Assert + assert report.stats["project_stubs_created"] == 1 + assert con.execute("SELECT local_id, name FROM project").fetchall() == [ + ("proj-stub", None) + ] + assert ( + con.execute("SELECT project_local_id FROM biosample").fetchone()[0] + == "proj-stub" + ) + assert con.execute("SELECT project_local_id FROM file").fetchone()[0] is None + + +def test_dump_files_should_upgrade_project_stub_when_real_object_arrives(): + """Test the stub-upgrade path. + + Given: + A stub-creating document followed by a document carrying the real + project object for the same key. + When: + Both documents are dumped in that order. + Then: + It should upgrade the stub to the real row with no conflict + recorded. + """ + # Arrange + first = make_fourdn_doc("file-1", project=None) + second = make_fourdn_doc("file-2") + + # Act + con, report = run_dump([first, second]) + + # Assert + assert report.rows["project"] == 1 + assert con.execute("SELECT name FROM project").fetchone()[0] == "Project One" + assert report.conflicts == {} + assert report.stats["project_stubs_created"] == 1 + + +def test_dump_files_should_bound_conflict_samples_when_cap_reached(): + """Test the conflict sample cap. + + Given: + Three documents embedding one collection whose lab diverges in each + copy, dumped with conflict_sample_cap=1. + When: + The documents are dumped. + Then: + It should write exactly one dump_conflict row while the conflict + tally keeps counting both divergent copies. + """ + # Arrange + docs = [make_fourdn_doc(f"file-{i}") for i in range(3)] + docs[1]["collections"][0]["lab"] = "Lab B" + docs[2]["collections"][0]["lab"] = "Lab C" + + # Act + con, report = run_dump(docs, conflict_sample_cap=1) + + # Assert + assert report.conflicts["collection"] == 2 + assert con.execute("SELECT count(*) FROM dump_conflict").fetchone()[0] == 1 + + +def test_dump_files_should_record_seen_submissions_when_filter_omitted(): + """Test the dump_info submissions fallback. + + Given: + One document per consortium and no submissions argument. + When: + The documents are dumped. + Then: + It should record the sorted seen submissions in dump_info. + """ + # Arrange + docs = [make_hubmap_doc(), make_fourdn_doc(), make_encode_doc()] + + # Act + con, _ = run_dump(docs) + + # Assert + assert con.execute("SELECT submissions FROM dump_info").fetchone()[0] == [ + "4dn", + "encode", + "hubmap", + ] + + +def test_dump_files_should_skip_extension_rows_when_all_fields_none(): + """Test all-None extension-row suppression. + + Given: + A document whose file-level and collection-level extra submodels + carry no values at all. + When: + The document is dumped. + Then: + It should write no extension rows for either level. + """ + # Arrange + doc = make_fourdn_doc(extra={"fourdn": {}}) + doc["collections"][0]["extra"] = {"fourdn": {}} + + # Act + _, report = run_dump([doc]) + + # Assert + assert report.rows["file_extra_fourdn"] == 0 + assert report.rows["file_extra_fourdn_extra_file"] == 0 + assert report.rows["collection_extra_fourdn"] == 0 + + +def test_dump_files_should_roundtrip_json_and_list_columns(): + """Test JSON and list column fidelity. + + Given: + A HuBMAP document whose collection enrichment carries an assay + metadata dict and a vitessce_hints list. + When: + The document is dumped. + Then: + It should round-trip the dict through the JSON column and the list + through the VARCHAR[] column unchanged. + """ + # Arrange + doc = make_hubmap_doc() + + # Act + con, _ = run_dump([doc]) + + # Assert + metadata, hints = con.execute( + "SELECT metadata, vitessce_hints FROM collection_extra_hubmap" + ).fetchone() + assert json.loads(metadata) == {"assay_category": "sequence"} + assert hints == ["is_sc", "rna"] + + +def test_dump_files_should_order_extra_files_by_source_position(): + """Test sidecar child-table ordinal ordering. + + Given: + A 4DN document with two extra_files sidecars. + When: + The document is dumped. + Then: + It should write one row per sidecar with ordinals matching source + order. + """ + # Arrange + doc = make_fourdn_doc() + doc["extra"]["fourdn"]["extra_files"].append( + { + "href": "/y", + "md5sum": "cc", + "file_size": "43", + "file_format": {"display_title": "bai"}, + } + ) + + # Act + con, _ = run_dump([doc]) + + # Assert + rows = con.execute( + "SELECT ordinal, href FROM file_extra_fourdn_extra_file ORDER BY ordinal" + ).fetchall() + assert rows == [(0, "/x"), (1, "/y")] + + +def test_dump_files_should_preserve_counts_when_batch_size_forces_flushes(): + """Test the staging flush path. + + Given: + Three documents sharing one collection, dumped with batch_size=1. + When: + The documents are dumped. + Then: + It should produce the same row counts as the default batch size. + """ + # Arrange + docs = [make_fourdn_doc(f"file-{i}") for i in range(3)] + + # Act + _, report = run_dump(docs, batch_size=1) + + # Assert + assert report.rows["file"] == 3 + assert report.rows["collection"] == 1 + assert report.rows["file_in_collection"] == 3 + + +def test_dump_files_should_keep_empty_compression_format(): + """Test compression_format empty-string semantics. + + Given: + A document whose compression_format is the empty string, meaning + "nothing recorded" rather than "uncompressed". + When: + The document is dumped. + Then: + It should store the empty string, not NULL. + """ + # Arrange + doc = make_fourdn_doc(compression_format="") + + # Act + con, _ = run_dump([doc]) + + # Assert + assert con.execute("SELECT compression_format FROM file").fetchone()[0] == "" + + +def test_v_file_should_inline_extension_columns_per_consortium(): + """Test v_file extension and term inlining. + + Given: + One ENCODE and one HuBMAP document. + When: + v_file is queried. + Then: + It should inline each row's own extension columns, term names, and + dcc abbreviation, with the other consortium's extension columns + NULL. + """ + # Arrange + con, _ = run_dump([make_encode_doc(), make_hubmap_doc()]) + + # Act + rows = { + submission: columns + for submission, *columns in con.execute( + "SELECT submission, s3_uri, rel_path, file_format_name, dcc_abbreviation" + " FROM v_file" + ).fetchall() + } + + # Assert + assert rows["encode"] == ["s3://encode-public/x.bed.gz", None, "BED", "ENCODE"] + assert rows["hubmap"] == [None, "expr/some.tsv", None, "HUBMAP"] + + +def test_v_file_biosample_should_resolve_names_at_fanout_grain(): + """Test the v_file_biosample join grain. + + Given: + One fully nested 4DN document. + When: + v_file_biosample is queried. + Then: + It should return the file-collection-biosample-subject grain with + anatomy and taxonomy names resolved. + """ + # Arrange + con, _ = run_dump([make_fourdn_doc()]) + + # Act + rows = con.execute( + "SELECT filename, collection_name, anatomy_name, taxonomy_name" + " FROM v_file_biosample" + ).fetchall() + + # Assert + assert rows == [("file-1.mcool", "Collection exp-1", "heart", "Homo sapiens")] + + +@settings(max_examples=25, deadline=None, suppress_health_check=[HealthCheck.too_slow]) +@given( + values=st.tuples( + *( + [ + st.text( + alphabet=st.characters( + min_codepoint=1, exclude_categories=("Cs",) + ), + min_size=1, + max_size=50, + ) + ] + * 4 + ) + ) +) +def test_dump_files_should_roundtrip_text_with_arbitrary_values(values): + """Test text fidelity across the document-to-DuckDB path. + + Given: + Arbitrary non-empty surrogate-free text assigned to the + pass-through file columns status, mime_type, analysis_type, and + condition. + When: + The document is dumped. + Then: + It should round-trip every value byte-exact through model + validation and parameterized insert. + """ + # Arrange + status, mime_type, analysis_type, condition = values + doc = make_fourdn_doc( + status=status, + mime_type=mime_type, + analysis_type=analysis_type, + condition=condition, + ) + + # Act + con, _ = run_dump([doc]) + + # Assert + row = con.execute( + 'SELECT status, mime_type, analysis_type, "condition" FROM file' + ).fetchone() + assert row == values From 06cbb4c35d68e4cb1844514bd39f050e03630a41 Mon Sep 17 00:00:00 2001 From: Conrad Date: Tue, 27 Oct 2026 10:43:12 -0400 Subject: [PATCH 4/4] docs: Document the DuckDB dump command, schema, and loader policies --- README.md | 49 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/README.md b/README.md index 55fe576..1405bfc 100644 --- a/README.md +++ b/README.md @@ -1240,3 +1240,52 @@ cfdb sync 4dn hubmap - `--api-url` - cfdb API base URL (default: `http://localhost:8000`, env: `CFDB_API_URL`) - `--api-key` - API key for sync endpoint (env: `SYNC_API_KEY`) - `--debug` / `-d` - Enable debugpy debugging + +### DuckDB dump + +`cfdb dump duckdb` flattens the materialized `files` collection into a single relational DuckDB file, so corpus-level questions — counts by assembly and consortium, every contact map from a given tissue, `output_type` coverage — become SQL anyone can run on a laptop instead of aggregation pipelines run by whoever holds Mongo credentials. It requires the `duckdb` extra: `pip install "cfdb[duckdb] @ git+https://github.com/abdenlab/cfdb.git"` (or `uv sync --extra dev` in a checkout). + +```bash +# Dump everything +cfdb dump duckdb cfdb.duckdb + +# One or more consortia, plus a Parquet export of every table +cfdb dump duckdb cfdb.duckdb -s 4dn -s encode --to-parquet ./parquet +``` + +**Options:** +- `--database-url` - MongoDB connection string (default: `mongodb://127.0.0.1:27017`, env: `DATABASE_URL`) +- `--database-name` - Database holding the `files` collection (default: `cfdb`, env: `DATABASE_NAME`) +- `--submission` / `-s` - Restrict to these consortia (`4dn`, `encode`, `hubmap`); repeatable, omit for all +- `--to-parquet` - Also `EXPORT DATABASE` every table as Parquet into this directory +- `--force` - Overwrite the output file if it exists + +**Schema.** The schema of record is [`src/cfdb/dump/schema.sql`](src/cfdb/dump/schema.sql) — a checked-in artifact in the same spirit as `schema.graphql`, versioned by its `schema_version` header and designed in [#123](https://github.com/abdenlab/cfdb/issues/123). The tables mirror C2M2 where C2M2 has a table (`file`, `collection`, `biosample`, `subject`, and the six junctions, all keyed by natural `(id_namespace, local_id)` composites, with `submission` as a non-key column on every entity); enrichment with a cross-consortium slot lives in promoted common columns (`genome_assembly`, `output_type`, `experiment_type`, `analyte_class`, …); the per-consortium residue lives in 1:1 `*_extra_*` extension tables, which omit verbatim-promoted fields but keep any field whose promotion is derived rather than verbatim (`replicate_info` survives because the replicate columns are regex-extracted from it). Three views carry the joins analysts would otherwise retype: `v_file` (one row per file with term names and extension columns inlined), `v_file_biosample` (the file × collection × biosample × subject fan-out grain), and `v_subject_demographics` (sex/race/age coalesced across the C2M2 columns and HuBMAP's enrichment, with a `*_source` column per value so provenance survives the coalesce). + +**Common-column mapping.** Each promoted column and the upstream field that fills it, per consortium. A column populated by one consortium today is still a common column, not a 4DN or ENCODE column: it names a slot any consortium could fill, which is what makes the next DCC's ingest a mapping exercise rather than a schema change. + +`file`: + +| Column | 4DN source | ENCODE source | HuBMAP source | +|---|---|---|---| +| `genome_assembly` | `genome_assembly` | `File assembly` | dataset `genome_assembly` | +| `genome_annotation` | — | `Genome annotation` | — | +| `output_type` | `file_type` | `Output type` | — | +| `output_type_detail` | `file_type_detailed` | `File format type` | — | +| `biological_replicates` | parsed from `replicate_info` | `Biological replicate(s)` | — | +| `technical_replicates` | parsed from `replicate_info` | `Technical replicate(s)` | — | +| `assay_info` | `assay_info` | — | — | +| `condition` | `condition` | — | — | + +`collection`: + +| Column | 4DN source | ENCODE source | HuBMAP source | +|---|---|---|---| +| `lab` | `lab` | `Lab` | — | +| `experiment_type` | `experiment_type` | assay | `dataset_type` | +| `experiment_target` | — | `Experiment target` / `Targets` | — | +| `analyte_class` | — | `Library made from` | `analyte_class` | + +Enrichment with no common slot today — ENCODE's `annotation_type`, `organism`, `software_used`, `encyclopedia_version`, `life_stage`, `age`, `age_units` among it — stays in the extension tables under its upstream name, and promoting such a field later is a schema change the `schema_version` header records. + +**Loader policies.** Entity rows repeated across documents are first-write-wins, and every divergent later copy is counted and sampled into `dump_conflict` — divergence is reported, never silently resolved. Junction rows take the union over all copies instead, so a relationship present in only one copy is never dropped. A `(id_namespace, local_id)` key appearing under two submissions aborts the dump: the natural keys are only globally unique because id_namespaces are DCC-scoped, and the loader asserts that rather than trusting it (duplicate file keys abort likewise). String-typed numbers from the C2M2 TSV path (`size_in_bytes` on 4DN/HuBMAP) and unparseable timestamps coerce to NULL with a counter, a document failing model validation is skipped with a counter, and every counter plus per-table row counts land in `dump_stat`, next to `dump_info` (schema version, generation time, cfdb version/commit, source database, submissions, file count) — so a dump is self-describing about which code produced it and where the source disagreed with itself. Counters measure occurrences, not entities: a divergent or uncoercible value on an entity embedded by many documents counts once per embedding document, so a large count can be one bad shared entity rather than many. Foreign keys are declared and enforced, so a dump that violates referential integrity fails at write time rather than surprising a reader. An aborted or interrupted run deletes its partial output, and the backstop if one ever survives is `dump_info`: the row is written last, so a database without one is a partial dump.