Parquet Files¶

# 00-latest = the newest monthly full; swap for a dated month (e.g. .../parquet/2026-06) to pin a release.
BASE="https://pub-f0ad0e0dac0345408dcc95bda20beb42.r2.dev/datasets/parquet/00-latest"
mkdir -p clinvar-gkm-parquet && cd clinvar-gkm-parquet

# Download specific sections
curl -O "${BASE}/scv.parquet"
curl -O "${BASE}/varcond-proposition.parquet"
curl -O "${BASE}/condition.parquet"

# Or download all 22 sections
for section in sequenceReference location allele copyNumberCount copyNumberChange \
               gene variation condition conditionSet therapy therapyGroup submitter \
               varcond-proposition vartumor-proposition vartherapy-proposition varcustom-proposition \
               evidenceLine vcv_evidenceLine rcv_evidenceLine \
               scv vcv rcv; do
  curl -O "${BASE}/${section}.parquet"
done

# Install DuckDB
brew install duckdb   # macOS
# or: pip install duckdb

# Query SCV statements
duckdb -c "
  SELECT id, classification.name AS classification, direction, strength.name AS strength, quality.name AS quality
  FROM 'scv.parquet'
  WHERE classification.name = 'Pathogenic'
  LIMIT 10;
"

# Count classifications across all SCVs
duckdb -c "
  SELECT classification.name AS classification, direction, COUNT(*) as n
  FROM 'scv.parquet'
  GROUP BY classification.name, direction
  ORDER BY n DESC;
"

# Join SCVs with propositions to find pathogenic variants for a specific condition
duckdb -c "
  SELECT s.id, s.classification.name AS classification, p.predicate, p.object_condition_id
  FROM 'scv.parquet' s
  JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
  WHERE s.classification.name = 'Pathogenic'
    AND p.object_condition_id LIKE '%clinvar.trait:9580%'
  LIMIT 10;
"

# Inspect the schema of any section
duckdb -c "DESCRIBE SELECT * FROM 'scv.parquet';"

import duckdb

df = duckdb.sql("""
    SELECT id, classification.name AS classification, direction, strength.name AS strength, quality.name AS quality
    FROM 'scv.parquet'
    WHERE classification.name = 'Pathogenic'
    LIMIT 100
""").df()

print(df)

import pandas as pd

# Load a section into a DataFrame
scv = pd.read_parquet("scv.parquet")

# Filter pathogenic SCVs — classification is a MappableConcept struct (dict), so read .name
pathogenic = scv[scv["classification"].map(lambda c: c and c.get("name")) == "Pathogenic"]
print(f"{len(pathogenic)} pathogenic SCVs")

# Access the full JSON when you need nested fields
import json
record = json.loads(pathogenic.iloc[0]["data"])
print(record["proposition"])

-- SCVs for BRCA1: classification, review status, condition
SELECT
    s.id AS scv_id,
    s.classification.name AS classification,
    s.direction,
    s.strength.name AS strength,
    s.quality.name AS review_status,
    p.gene_context_name AS gene,
    c.name AS condition_name,
    c.primaryCoding.code AS condition_code
FROM 'scv.parquet' s
JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
LEFT JOIN 'condition.parquet' c ON p.object_condition_id = c.id
WHERE p.gene_context_name = 'BRCA1'
ORDER BY s.classification.name;

-- Count SCVs by classification and review status for BRCA2
SELECT
    p.gene_context_name AS gene,
    s.classification.name AS classification,
    s.quality.name AS review_status,
    s.direction,
    COUNT(*) AS scv_count
FROM 'scv.parquet' s
JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
WHERE p.gene_context_name = 'BRCA2'
GROUP BY ALL
ORDER BY scv_count DESC;

-- Only expert panel and criteria-provided SCVs for a gene
SELECT
    s.id AS scv_id,
    s.classification.name AS classification,
    s.quality.name AS review_status,
    c.name AS condition_name
FROM 'scv.parquet' s
JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
LEFT JOIN 'condition.parquet' c ON p.object_condition_id = c.id
WHERE p.gene_context_name = 'TP53'
  AND s.quality.name IN ('criteria provided', 'reviewed by expert panel')
ORDER BY s.classification.name;

-- Compare pathogenicity classification distributions across genes
SELECT
    p.gene_context_name AS gene,
    s.classification.name AS classification,
    COUNT(*) AS n
FROM 'scv.parquet' s
JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
WHERE p.gene_context_name IN ('BRCA1', 'BRCA2', 'TP53', 'MLH1')
  AND s.quality.name = 'criteria provided'
GROUP BY gene, s.classification.name
ORDER BY gene, n DESC;

-- SCVs with submitter name and assertion method (from JSON)
SELECT
    s.id AS scv_id,
    s.classification.name AS classification,
    p.gene_context_name AS gene,
    json_extract_string(s.data, '$.contributions[0].agent.name') AS submitter,
    json_extract_string(s.data, '$.specifiedBy.name') AS method
FROM 'scv.parquet' s
JOIN 'varcond-proposition.parquet' p ON s.proposition_id = p.id
WHERE p.gene_context_name = 'BRCA1'
  AND s.classification.name = 'Pathogenic'
  AND s.quality.name = 'reviewed by expert panel'
LIMIT 20;

BASE="https://pub-f0ad0e0dac0345408dcc95bda20beb42.r2.dev"

# 1. The checkpoint monthly full named by the delta's manifest (e.g. 2026-06).
curl -s "${BASE}/deltas/00-latest/manifest.json" -o manifest.json
CKPT=$(python3 -c "import json;print(json.load(open('manifest.json'))['checkpoint_full']['release'])")

# 2. The checkpoint full Parquet set -> full/, and the delta set -> delta/.
mkdir -p full delta
for section in sequenceReference location allele copyNumberCount copyNumberChange \
               gene variation condition conditionSet therapy therapyGroup submitter \
               varcond-proposition vartumor-proposition vartherapy-proposition varcustom-proposition \
               evidenceLine vcv_evidenceLine rcv_evidenceLine scv vcv rcv; do
  curl -sf "${BASE}/datasets/parquet/${CKPT}/${section}.parquet" -o "full/${section}.parquet"  || true
  curl -sf "${BASE}/deltas/00-latest/parquet/${section}.parquet" -o "delta/${section}.parquet" || true
done

import duckdb, json, os, shutil

con = duckdb.connect()
manifest = json.load(open("manifest.json"))

# Start from a copy of the checkpoint full set; overwrite only the changed sections.
shutil.copytree("full", "updated", dirs_exist_ok=True)

for section, info in manifest["sections"].items():
    full    = f"full/{section}.parquet"      # checkpoint (baseline) section
    delta   = f"delta/{section}.parquet"     # added + updated rows (may be absent)
    out     = f"updated/{section}.parquet"
    deleted = info["deleted"]                # ids removed this release
    has_delta = os.path.exists(delta)

    if not os.path.exists(full):
        # Section new since the checkpoint: the delta rows ARE the section.
        if has_delta:
            shutil.copyfile(delta, out)
        continue

    if not has_delta:
        # Deletes only — filter the baseline, nothing to append.
        con.execute("""
            COPY (SELECT * FROM read_parquet(?) WHERE id NOT IN (SELECT unnest(?)))
            TO ? (FORMAT PARQUET)
        """, [full, deleted, out])
    else:
        # Drop updated + deleted keys from the baseline, then append the delta rows.
        con.execute("""
            COPY (
                SELECT * FROM read_parquet(?)                        -- baseline full
                WHERE id NOT IN (SELECT id FROM read_parquet(?))     -- drop updated keys
                  AND id NOT IN (SELECT unnest(?))                   -- drop deleted keys
                UNION ALL BY NAME
                SELECT * FROM read_parquet(?)                        -- append added + updated
            ) TO ? (FORMAT PARQUET)
        """, [full, delta, deleted, delta, out])

# `updated/<section>.parquet` now reflects the release the manifest names.

import duckdb, json, os, urllib.request

BASE = "https://pub-f0ad0e0dac0345408dcc95bda20beb42.r2.dev"
TARGET_WEEK = "2026-07-20"   # ClinVar release date (YYYY-MM-DD) to reconstruct up to (inclusive)

SECTIONS = [
    "sequenceReference", "location", "allele", "copyNumberCount", "copyNumberChange",
    "gene", "variation", "condition", "conditionSet", "therapy", "therapyGroup", "submitter",
    "varcond-proposition", "vartumor-proposition", "vartherapy-proposition", "varcustom-proposition",
    "evidenceLine", "vcv_evidenceLine", "rcv_evidenceLine", "scv", "vcv", "rcv",
]

con = duckdb.connect()


def fetch_json(url):
    with urllib.request.urlopen(url) as r:
        return json.load(r)


# 1. Read the checkpoint from the TARGET_WEEK's OWN manifest — its `checkpoint_full` records the
#    monthly full that was current when that week was uploaded. Do NOT use deltas/00-latest here:
#    its checkpoint advances as new monthly fulls publish, so it reflects today's month, not this
#    week's era, and the delta filter below would then match nothing for an older target week.
y, mo, day = TARGET_WEEK.split("-")
target_dir = f"{y}-{mo}{day}"                # "2026-07-20" -> "2026-0720"
checkpoint = fetch_json(f"{BASE}/deltas/{target_dir}/manifest.json")["checkpoint_full"]
checkpoint_month = checkpoint["release"]     # "2026-06"

# 2. Download that checkpoint's full Parquet set into updated/ as the starting state.
#    Prior-year checkpoints live under archives/<YYYY>/parquet/<release>/.
if checkpoint["path"].startswith("archives/"):
    year = checkpoint["path"].split("/")[1]
    parquet_base = f"{BASE}/archives/{year}/parquet/{checkpoint_month}"
else:
    parquet_base = f"{BASE}/datasets/parquet/{checkpoint_month}"
os.makedirs("updated", exist_ok=True)
for section in SECTIONS:
    try:
        urllib.request.urlretrieve(f"{parquet_base}/{section}.parquet", f"updated/{section}.parquet")
    except Exception:
        pass   # a section absent from the checkpoint first appears when a delta introduces it

# 3. From the index, take the dated weekly deltas AFTER the checkpoint month, oldest -> newest,
#    up to and including TARGET_WEEK.
index = fetch_json(f"{BASE}/index.json")
deltas = sorted(
    (d for d in index["deltas"]
     if d["release"] != "latest"
     and d["release"][:7] > checkpoint_month
     and d["release"] <= TARGET_WEEK),
    key=lambda d: d["release"],
)

# 4. Replay each delta's Parquet (added/updated rows) + manifest deletes, verifying the chain.
prev_compare = None
for d in deltas:
    manifest = fetch_json(f"{BASE}/{d['manifest']}")
    if prev_compare is not None and manifest["baseline_release"] != prev_compare:
        raise SystemExit(
            f"broken chain before {manifest['release']}: re-bootstrap from a monthly full"
        )

    present = set(d.get("parquet") or [])   # sections with a delta Parquet this week
    for section, info in manifest["sections"].items():
        out = f"updated/{section}.parquet"
        deleted = info["deleted"]
        delta_local = None
        if section in present:
            delta_local = f"updated/.delta_{section}.parquet"
            urllib.request.urlretrieve(
                f"{BASE}/{d['path']}parquet/{section}.parquet", delta_local
            )

        if not os.path.exists(out):
            # Section new since the checkpoint: the delta rows ARE the section.
            if delta_local:
                os.replace(delta_local, out)
            continue

        tmp = out + ".tmp"
        if delta_local is None:
            # Deletes only — filter the baseline, nothing to append.
            con.execute("""
                COPY (SELECT * FROM read_parquet(?) WHERE id NOT IN (SELECT unnest(?)))
                TO ? (FORMAT PARQUET)
            """, [out, deleted, tmp])
        else:
            con.execute("""
                COPY (
                    SELECT * FROM read_parquet(?)
                    WHERE id NOT IN (SELECT id FROM read_parquet(?))
                      AND id NOT IN (SELECT unnest(?))
                    UNION ALL BY NAME
                    SELECT * FROM read_parquet(?)
                ) TO ? (FORMAT PARQUET)
            """, [out, delta_local, deleted, delta_local, tmp])
        os.replace(tmp, out)
        if delta_local and os.path.exists(delta_local):
            os.remove(delta_local)

    prev_compare = manifest["compare_release"]

# updated/<section>.parquet now reflects the TARGET_WEEK release.