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. 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" 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; 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