Conversation
…ion join Introduces a full seqr-loader entry point (synthetic_full_workflow.py) that runs the synthetic-proband pre-workflow -> isolated combiner -> densify -> row-join against the standard workflow's global annotate_cohort.mt -> per-dataset annotate -> ES index load. Skips the standard annotation stack (VEP / VQSR / gnomAD / clinvar joins) so synthetic samples never pollute AC/AN/AF for other projects. New stages (in synthetic_proband_stages.py): - CombineGvcfsIntoVdsFromManifest: consumes the pre-workflow manifest via inputs.as_path, builds a fresh VDS every run (no incremental machinery). - AnnotateFromGlobalCallset: metamist-queries the latest AnnotateCohort MT, row-joins onto the densified MT. Fails loud in queue_jobs if no source MT is found. Manual override via annotate_from_global_callset.source_mt. - SubsetMtToDatasetFromGlobalCallset, AnnotateDatasetFromGlobalCallset, ExportMtAsEsIndexFromGlobalCallset: thin subclasses of the standard stages that swap the upstream annotation source. Standard-workflow changes (stages.py): - Extracted _get_cohort_mt / _get_annotated_mt_path helper methods on SubsetMtToDatasetWithHail, AnnotateDataset, ExportMtAsEsIndex. Behaviour is identical - the extractions are extension seams for the subclasses above. - Made CreateDenseMtFromVdsWithHail's hps_vcf_dir + separate_header_vcf_dir outputs optional, gated on combiner.emit_sites_only_vcf_fragments (defaults True, so standard workflow is unchanged). Synthetic workflow sets it false. Cleanup: - Removed the config-gated synthetic_gvcf_text_file branch from the standard CombineGvcfsIntoVds job - the new manifest-driven stage replaces it. Metamist helper (utils.py): - query_for_latest_annotate_cohort_mt filters LATEST_ANALYSIS_QUERY by meta.stage == 'AnnotateCohort' (excluding legacy AnnotateCohortSmallVariantsWithHailQuery runs from production-pipelines) and meta.sequencing_type. Design decision worth reviewing (Matt/Ed): Chose Option A (helper-method extract + subclass) over duplicating AnnotateDataset in synthetic_proband_stages.py. Applied the same pattern to SubsetMtToDatasetWithHail and ExportMtAsEsIndex so the synthetic path can support only_families config on ravenscroft-rpl datasets in the future without further stages.py refactors. Alternative considered: config-driven required_stages on AnnotateDataset, rejected because required_stages is class-level and can't be conditional per workflow. The four _get_* extraction points are intentional extension seams - their docstrings name the AnnotateFromGlobalCallset variants as the intended overriders. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
The synthetic workflow's DAG was broken. AnnotateFromGlobalCallset required CreateDenseMtFromVdsWithHail, whose own required_stages is the standard CombineGvcfsIntoVds - so cpg-flow would pull the standard combiner into the synthetic DAG and skip our new CombineGvcfsIntoVdsFromManifest (and by extension the pre-workflow stages) entirely. The full workflow would attempt to consume gVCFs from metamist and never register the synthetic ones (which are typed `synthetic_gvcf`, not `gvcf`). Fix: subclass CreateDenseMtFromVdsWithHail as CreateDenseMtFromVdsWithHail- NoFragments in synthetic_proband_stages.py, redecorated to require CombineGvcfsIntoVdsFromManifest. Overrides _get_input_vds (new extraction seam on the base class) to read from the manifest combiner, and overrides expected_outputs to omit the four sites-only VCF-fragment keys that only VQSR / VEP care about. AnnotateFromGlobalCallset now requires the subclass. Also removes the emit_sites_only_vcf_fragments config flag added in the prior commit. The flag was a config-driven proxy for what the subclass does more directly, and nobody but the synthetic workflow would ever set it. The subclass approach is consistent with the pattern already used for AnnotateDatasetFromGlobalCallset et al. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
cpg-flow's @stage.stage decorator wraps its target class in a function (via functools.wraps), which cannot be subclassed - Python raises "TypeError: function() argument 'code' must be code, not str" when it tries to build a class body inheriting from a function. The first commit on this branch tried to inherit from the decorated stages (CreateDenseMtFromVdsWithHail, SubsetMtToDatasetWithHail, AnnotateDataset, ExportMtAsEsIndex) and hit this error at import time. Split each of those four stages into an undecorated Base class carrying the logic (expected_outputs, queue_jobs, extension-seam methods) and a tiny decorated wrapper that inherits from it and holds only the @stage.stage(...) registration. Standard workflow behaviour is unchanged because the decorated wrappers preserve the original names, decorators, and required_stages. The four synthetic-workflow subclasses in synthetic_proband_stages.py now inherit from the Base classes, which are real classes and can be subclassed normally. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Step 2's change (b5a4baf) put the two optional --sites_only / --separate_header args on their own line at the end of the command's multi-line f-string. When both resolve to '' (the synthetic-workflow case, via CreateDenseMtFromVdsWithHailNoFragments), the previous line's trailing `\` continues into a whitespace-only line, which broke bash parsing inside Hail Batch's wrapper script (`syntax error: unexpected end of file`). Move the conditional args onto the same line as --checkpoint so the trailing line of the multi-line command always has real content, matching the pattern every other job file in the repo uses. Same behaviour when either arg is non-empty, no longer crashes when both are empty. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
sparse_split_multi (in densify_VDS_to_MT) leaves non-minimal padding on split alleles. Two independently-processed MTs (this cohort's synthetic MT vs the global annotate_cohort.mt) end up with the same biological variant under different (locus, alleles) keys, and the exact-match annotate_rows join in AnnotateFromGlobalCallset misses them. Investigation (see analysis-notebooks explore-bucket-mts.ipynb, ravenscroft- rpl test run 2026-09-17) found 124/11.6M synthetic rows unmatched, and 124/124 recoverable by trimming the global's rows to their minimal form and matching against the synthetic. Every mismatch was a normalisation-padding artefact, not a real variant difference. Fix: after the exact anti-join, driver-side loop over each unmatched row does a partition-indexed locus-window search on the global, trims each candidate's alleles, and finds the one whose trimmed form matches the input row. Builds a small keyed rewrite table (Hail lookup broadcast) and re-keys the input MT's affected rows to the global's non-minimal key so annotate_rows lands. Cost is bounded by mismatch count (small - scales with input MT size, not the much larger global MT). No shuffle of the global (only partition-indexed reads), so production runs against the full global annotate_cohort.mt stay tractable. Input MT re-key is a shuffle but on the small side. Output MT keys are all sourced from the global (either exact-matched or key-rewritten to global's form). This means seqr will see variants under the global's representation - matches the collaborator requirement that this workflow should not invent variant representations. The invariant check moves after recovery: it now only fires for genuinely unmatched rows (create_synthetic_proband_gvcf.py invention, or temporal drift between the global's build date and the parental gVCFs' current state). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Previously AnnotateFromGlobalCallset did a second anti_join against the global AFTER the key-recovery step, as a belt-and-braces sanity check that recovery worked. That second anti_join triggered a distributed sort on the re-keyed input MT which failed on Hail-on-Batch with an unhelpful "unknown error, zero partition errors" (likely worker preemption during the shuffle). The recovery loop already knows per-row whether it found a match. Track the unrecovered rows during the loop and raise ValueError directly from the recovery function if any are unrecovered - same fail-loud contract, no redundant anti_join computation, no redundant shuffle. Also fails earlier (before the re-key work) if recovery is incomplete. Delete _assert_every_row_present_in_global; its contract is now inline in _recover_mismatched_keys_from_global. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Previous approach re-keyed the entire input MT after finding ~124 rows that
needed to move to the global's non-minimal keys. That triggered a full
distributed sort over 11.6M rows, which hung indefinitely (17+ hours on one
partition) and eventually crashed with "unknown error, zero partition errors"
- data skew during the sort left one worker with an outsized chunk it
couldn't finish.
Restructure: split the input into two halves.
- already_ok: rows whose keys already match global directly. Kept at their
original keys (no re-key), joined with annotate_rows normally.
- needs_rewrite: tiny subset (~100s of rows) that need re-keying to the
global's non-minimal form. Coalesced to one partition, then re-keyed -
trivial shuffle at that scale.
Then union_rows the two halves. Both are sorted by (locus, alleles), so
union_rows does a linear merge with no distributed sort.
Net cost: no shuffle on the bulk 11.6M rows. Sort work reduced to the ~100s
of rewritten rows, which comfortably fits in a single partition. Output MT
keys are still all sourced from the global (either exact-matched or
key-rewritten), preserving the invariant that seqr sees only global keys.
Rename _recover_mismatched_keys_from_global -> _build_rewrite_table (now
returns the lookup table rather than a modified MT) and add
_annotate_via_split_union (owns the split-union join). Docstring updated.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Chaining needs_rewrite.naive_coalesce(1).key_rows_by(...) with rewrite_tmp field references inside the key_rows_by call binds those field expressions to the pre-coalesce MT identity. When key_rows_by then evaluates against the post-coalesce MT, Hail refuses to mix expressions from different-identity sources - even though the schema is identical: ExpressionException: Cannot combine expressions from different source objects Same fix as the notebook: assign the coalesced MT to a variable, then reference rewrite_tmp through that variable so all expressions bind to the same identity. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…ed sort Every approach that produced an output MT with the global's non-minimal keys for the ~124 recovered rows required a distributed sort (full-MT re-key OR union_rows on two independently-keyed halves) and kept failing on Hail Query-on-Batch. Investigation of Hail's Scala backend (ServiceBackend.scala:159-165) showed the QoB workers are hardcoded as preemptible=true with no Python-side override, so preemption during the sort kept producing the "unknown error, zero partition errors" backend failure. New approach: compute an "effective lookup key" per row (rewrite's non-minimal key for the ~124 recovered rows, original key for everything else) and index global_rows by that computed key. Global annotations still land on every input row - but the input MT is never re-keyed, no union is required, and no distributed sort happens anywhere. Tradeoff: output MT keeps min-rep keys throughout, including for the 124 recovered rows. Seqr will index those 124 variants under min-rep form rather than the global's non-minimal padding. This is arguably more canonical (matches ClinVar / gnomAD conventions) and no variants are lost. Rename _annotate_via_split_union -> _annotate_via_effective_key. Update the module docstring to explain the tradeoff and its motivation. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
…nobs
Ed's feedback on the LowerDistributedSort failures: preemption of QoB workers
isn't the root cause (Hail handles it robustly). Two suggestions worth trying:
1. Add a checkpoint before the failing join step - splits Hail's query graph
so QoB doesn't have to fuse the whole pipeline into one shuffle, and lets
us recover cheaply from partial failures.
2. Tune QoB resourcing via driver_cores / worker_cores on init_batch. Start
with 2/1, escalate to 2/2 or 4/2 if the first attempt still fails.
Changes:
- scripts/annotate_from_global_callset.py: accept --checkpoint CLI arg. In
_annotate_via_effective_key, checkpoint the input MT (with lookup-key fields
added) before the annotation join. Pass driver_cores / worker_cores to
init_batch via new config keys under `annotate_from_global_callset`,
defaulting to 2/1.
- jobs/AnnotateFromGlobalCallset.py: pass the checkpoint path through to the
script.
- synthetic_proband_stages.py: compute checkpoint path from self.tmp_prefix
so it lives in the -tmp bucket (auto-cleaned by lifecycle rules; we don't
want stale checkpoints in main after successful runs).
Also updates the module docstring to correct the earlier (incorrect) claim
that QoB workers are preemption-vulnerable - Ed confirmed this isn't the
failure mode.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
Every previous approach (re-key + union, computed-key lookup with the input
MT's real row_key hidden inside an hl.if_else) tripped Hail's optimizer into
a full distributed sort, which then hung on partition 0 for hours. Gemini
suggested and Ed's guidance supports: split the join into two parts.
1. Direct join `global_rows[input_mt.row_key]` for the bulk. Uses the input
MT's actual row key so Hail can prove alignment and potentially stream
without shuffling. If it does need to co-partition, it's at least the
best-optimised join case.
2. For the ~124 rows whose exact keys don't appear in global (sparse_split-
_multi padding), collect their global annotations driver-side during the
recovery step, then broadcast a small hl.literal dict of overrides.
3. hl.coalesce(direct_join_result, broadcast_override) picks the direct-join
annotation where it landed, else the broadcast override.
Consequence: the ~124 rewrites are handled by a Python dict broadcast (a few
MB), not by a Hail-side lookup on a computed key. No shuffle for the
override path.
Renamed _build_rewrite_table -> _build_rewrite_list (now returns list of dicts
including the collected global annotation struct, not a Hail Table).
_annotate_via_effective_key -> _annotate_via_direct_join_with_broadcast.
Removed the pre-join checkpoint added in f269a0d - no longer needed since the
join no longer runs on the hangy code path. Also removed the checkpoint_path
plumbing from the job wrapper + stage.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
annotate_cohort.mtfor row annotations so synthetic samples never enter the global AC/AN/AF stats.synthetic_proband_stages.py:CombineGvcfsIntoVdsFromManifest,AnnotateFromGlobalCallset, and three thin*FromGlobalCallsetsubclasses ofSubsetMtToDatasetWithHail/AnnotateDataset/ExportMtAsEsIndexthat swap the upstream annotation source without duplicating logic.stages.pyextractions of_get_cohort_mt/_get_annotated_mt_pathare extension seams for the subclasses;combiner.emit_sites_only_vcf_fragmentsdefaultsTrueso densify still emits VCFs for the standard path.Design decision worth reviewing (Matt / Ed)
Chose helper-method extract + subclass over duplicating
AnnotateDataset. Applied the same pattern toSubsetMtToDatasetWithHailandExportMtAsEsIndexso this workflow can eventually supportonly_familieson RPL datasets without furtherstages.pyrefactors. Config-drivenrequired_stageswas rejected because that field is class-level and can't be conditional per workflow. See the commit message for the fuller writeup.Stacked PR
Base:
dummy_gvcf_stage_code(PR #56). Do not merge before that base merges.Test plan
synthetic_full_workflow.pyagainst a ravenscroft-rpl-shaped test config; confirm the resolved DAG contains:GenerateSyntheticProbandGvcfs → GenerateSyntheticProbandCombinerInputs → CombineGvcfsIntoVdsFromManifest → CreateDenseMtFromVdsWithHail → AnnotateFromGlobalCallset → AnnotateDatasetFromGlobalCallset → ExportMtAsEsIndexFromGlobalCallsetand nothing from the standard VEP/VQSR chain.combiner.emit_sites_only_vcf_fragments = falseis set in the synthetic workflow's config; densify should skip thehps_vcf_dirandseparate_header_vcf_diroutputs.query_for_latest_annotate_cohort_mt('seqr')returns the expected latest genomeannotate_cohort.mtfrom metamist (2026-08-12 entry as of writing).annotate_cohort.mtdescribe (DP, GQ, GP, PG, SB, RGQ, MIN_DP, PS, PID, GT, PGT, AD, PL).AnnotateFromGlobalCallsetinvariant check does not fire (no synthetic-only variants) on the end-to-end run.first_workflow.pyandfull_workflow.pyand confirm their DAG shapes are unchanged.🤖 Generated with Claude Code