gnomad_qc.v5.annotations.batch_fanout

Hail Batch relay fan-out for the v5 annotation scripts.

A Query-on-Batch run has one driver, and a driver crash loses everything it was computing. The scripts therefore cut the genome into chunks and run each chunk from its own small, non-spot Batch job (a “relay”) that starts a QoB driver inside its container. A crash then costs one chunk, and a rerun skips the chunks that already have a _SUCCESS marker. The orchestrator that submits the relays never starts Hail.

Chunk outputs are stored under a hash of the chunk layout. The layout is sampled without a fixed seed, so each regeneration moves the cut points, and a chunk left behind by an older layout must not be taken for a finished chunk of the new one.

Module Functions

gnomad_qc.v5.annotations.batch_fanout.RelayJobSpec(...)	One relay job for submit_relay_batch().
gnomad_qc.v5.annotations.batch_fanout.resolve_commit()	Return the gnomad_qc commit the relay containers should check out.
gnomad_qc.v5.annotations.batch_fanout.build_setup_command(commit)	Return the shell prefix every relay runs before the script.
gnomad_qc.v5.annotations.batch_fanout.relay_context(...)	Return what every relay submission needs: setup command, backend kwargs, script.
gnomad_qc.v5.annotations.batch_fanout.submit_relay_batch(...)	Submit one Hail Batch of relay jobs and wait for it.
gnomad_qc.v5.annotations.batch_fanout.chunk_intervals_hash(data)	Return a 16-hex-char content hash of a chunk layout.
gnomad_qc.v5.annotations.batch_fanout.test_region_hash(...)	Return the layout hash of a --test-region run, which has no layout JSON.
gnomad_qc.v5.annotations.batch_fanout.interval_to_list(iv)	Serialize a locus interval for the chunk-intervals JSON.
gnomad_qc.v5.annotations.batch_fanout.interval_from_list(t, ...)	Rebuild a locus interval from its chunk-intervals JSON form.
gnomad_qc.v5.annotations.batch_fanout.parse_region_interval(...)	Parse contig:start-end into a half-open locus interval.
gnomad_qc.v5.annotations.batch_fanout.spans_sex_chromosome(...)	Return whether a read scope can include chrX or chrY.
gnomad_qc.v5.annotations.batch_fanout.select_chunks_for_contig(...)	Return the chunk indices on chrom, or every index when chrom is None.
gnomad_qc.v5.annotations.batch_fanout.apply_path_suffix(...)	Insert _<suffix> before the .ht extension; unchanged when suffix is empty.
gnomad_qc.v5.annotations.batch_fanout.combine_suffix(*parts)	Join the non-empty parts with underscores, so each scope writes its own path.
gnomad_qc.v5.annotations.batch_fanout.chunk_path(...)	Return the per-chunk HT path beside the final HT.
gnomad_qc.v5.annotations.batch_fanout.group_path(...)	Return a merge-tree intermediate HT path.
gnomad_qc.v5.annotations.batch_fanout.failed_chunks_path(...)	Return the failed-chunk manifest path for one layout.
gnomad_qc.v5.annotations.batch_fanout.list_present_chunk_indices(...)	Return the chunk indices that have a _SUCCESS marker.
gnomad_qc.v5.annotations.batch_fanout.new_run_id()	Return a UTC timestamp id for one orchestrator run.
gnomad_qc.v5.annotations.batch_fanout.write_failed_chunks_manifest(...)	Record which dispatched chunks did not land.
gnomad_qc.v5.annotations.batch_fanout.dispatch_in_waves(...)	Submit pending chunks in sequential waves and report the ones that did not land.
gnomad_qc.v5.annotations.batch_fanout.orchestrate_tree_merge(...)	Union inputs into final_output_path with a tree of relay merge jobs.
gnomad_qc.v5.annotations.batch_fanout.union_and_write_hts(...)	Union HTs that share a schema and globals, and write the result.

Hail Batch relay fan-out for the v5 annotation scripts.

A Query-on-Batch run has one driver, and a driver crash loses everything it was computing. The scripts therefore cut the genome into chunks and run each chunk from its own small, non-spot Batch job (a “relay”) that starts a QoB driver inside its container. A crash then costs one chunk, and a rerun skips the chunks that already have a _SUCCESS marker. The orchestrator that submits the relays never starts Hail.

Chunk outputs are stored under a hash of the chunk layout. The layout is sampled without a fixed seed, so each regeneration moves the cut points, and a chunk left behind by an older layout must not be taken for a finished chunk of the new one.

class gnomad_qc.v5.annotations.batch_fanout.RelayJobSpec(name: str, cpu: float, memory: str, storage: str, attempts: int, command: str)

One relay job for submit_relay_batch().

name: str

Alias for field number 0

cpu: float

Alias for field number 1

memory: str

Alias for field number 2

storage: str

Alias for field number 3

attempts: int

Alias for field number 4

command: str

Alias for field number 5

gnomad_qc.v5.annotations.batch_fanout.resolve_commit()

Return the gnomad_qc commit the relay containers should check out.

The GNOMAD_QC_COMMIT environment variable takes precedence so an orchestrator that itself runs inside a Batch job, where the checkout is a tarball with no .git directory, can still pin the relays.

Return type:

str

Returns:

Full commit hash.

gnomad_qc.v5.annotations.batch_fanout.build_setup_command(commit, gcp_billing_project='broad-mpg-gnomad', methods_branch='main', hail_version=None)

Return the shell prefix every relay runs before the script.

Both repos change faster than the image, so the relay downloads them at the pinned commit and branch when it starts. The Hail config file sets the Batch billing project and scratch directory for the nested QoB driver, and together with the quota_project_id patch gives Hail’s GCS client a requester-pays project, which a bare container has no gcloud config to supply.

Parameters:

• commit (str) – gnomad_qc commit to check out.

• gcp_billing_project (str) – Requester-pays project.

• methods_branch (str) – gnomad_methods branch or commit to check out.

• hail_version (Optional[str]) – If set, reinstall this Hail version in the container; otherwise the image’s Hail is used. The relay’s Python Hail sets the QoB JAR, so this pins everything. WARNING: the pin is also a floor for every HT the pipeline reads. Hail’s table format is not backward compatible, so lowering the pin without rewriting the input HTs breaks the fan-out.

Return type:

str

Returns:

Shell command string ending in a newline.

gnomad_qc.v5.annotations.batch_fanout.relay_context(commit, methods_branch, batch_billing_project, batch_remote_tmpdir, script_name, gcp_billing_project='broad-mpg-gnomad', hail_version=None)

Return what every relay submission needs: setup command, backend kwargs, script.

Bundles the three values from one caller-supplied commit so every relay an orchestrator submits checks out the same code. The fan-out and the merge are separate orchestrator runs and each resolves its own commit.

Parameters:

• commit (str) – gnomad_qc commit the relays check out.

• methods_branch (str) – gnomad_methods branch or commit the relays check out.

• batch_billing_project (str) – Hail Batch billing project for the relay jobs.

• batch_remote_tmpdir (Optional[str]) – hb.ServiceBackend scratch, or None for Hail’s configured one.

• script_name (str) – Script file name under gnomad_qc/v5/annotations.

• gcp_billing_project (str) – Requester-pays project for the relays.

• hail_version (Optional[str]) – Optional Hail pin for build_setup_command().

Return type:

Tuple[str, Dict[str, str], str]

Returns:

(setup_cmd, backend_kwargs, script_command).

gnomad_qc.v5.annotations.batch_fanout.submit_relay_batch(batch_image, backend_kwargs, batch_name, job_specs, log_label, *, dry_run)

Submit one Hail Batch of relay jobs and wait for it.

Relays are non-spot because a preempted relay leaves the QoB batch it was waiting on running with nobody to collect it. For the same reason callers normally ask for one attempt: a Batch retry cannot cancel that orphaned batch and would run alongside it.

Parameters:

• batch_image (str) – Docker image for the relay containers.

• backend_kwargs (Dict[str, str]) – kwargs for hb.ServiceBackend.

• batch_name (str) – Hail Batch name.

• job_specs (Sequence[RelayJobSpec]) – Jobs to submit; nothing is submitted when empty.

• log_label (str) – Noun for log messages (“chunk”, “merge”).

• dry_run (bool) – Validate the batch without running it.

Return type:

Optional[int]

Returns:

Batch id, or None when nothing ran.

gnomad_qc.v5.annotations.batch_fanout.chunk_intervals_hash(data)

Return a 16-hex-char content hash of a chunk layout.

Every regeneration of the layout moves the cut points, and keying outputs by this hash keeps each layout’s chunks apart. An embedded intervals_hash key is dropped before hashing, so a JSON that carries its own hash re-hashes to the same value.

Parameters:

data (Dict[str, Any]) – Parsed chunk-intervals JSON.

Return type:

str

Returns:

First 16 hex chars of the SHA-256 of its canonical serialization.

gnomad_qc.v5.annotations.batch_fanout.test_region_hash(test_region)

Return the layout hash of a --test-region run, which has no layout JSON.

Hashing the region strings keeps two region tests under the same output path from seeing each other’s chunk as already present. The sub-interval count is deliberately left out: the merge and validate steps do not pass it, and it does not change the chunk’s contents.

Parameters:

test_region (Sequence[str]) – Region strings as passed on the command line.

Return type:

str

Returns:

test_region_ plus a 16-hex-char hash.

gnomad_qc.v5.annotations.batch_fanout.interval_to_list(iv)

Serialize a locus interval for the chunk-intervals JSON.

Parameters:

iv (Interval) – Locus interval.

Return type:

list

Returns:

[start_contig, start_pos, end_contig, end_pos, includes_start, includes_end].

gnomad_qc.v5.annotations.batch_fanout.interval_from_list(t, reference_genome)

Rebuild a locus interval from its chunk-intervals JSON form.

Parameters:

• t (Sequence) – Output of interval_to_list().

• reference_genome (str) – Reference genome name.

Return type:

Interval

Returns:

Locus interval.

gnomad_qc.v5.annotations.batch_fanout.parse_region_interval(interval_s, reference_genome='GRCh38')

Parse contig:start-end into a half-open locus interval.

Half-open so adjacent regions never share a locus.

Parameters:

• interval_s (str) – Interval string; commas in positions are allowed.

• reference_genome (str) – Reference genome name.

Return type:

Interval

Returns:

[start, end) locus interval.

gnomad_qc.v5.annotations.batch_fanout.spans_sex_chromosome(intervals, chrom=None)

Return whether a read scope can include chrX or chrY.

The sex-ploidy adjustment changes nothing on autosomes, so a scope that cannot touch chrX or chrY may skip it. A read with no scope counts as spanning.

Parameters:

• intervals (Optional[Sequence[Interval]]) – Read intervals, or None. When given, they decide the answer and chrom is ignored.

• chrom (Optional[str]) – Single contig, or None. Used only when intervals is empty; this path assumes GRCh38 contig names.

Return type:

bool

Returns:

True unless the scope is known to exclude both sex contigs.

gnomad_qc.v5.annotations.batch_fanout.select_chunks_for_contig(chunk_contigs, chrom)

Return the chunk indices on chrom, or every index when chrom is None.

Parameters:

• chunk_contigs (Sequence[Optional[str]]) – Contig of each chunk, by index.

• chrom (Optional[str]) – Contig to keep, or None.

Return type:

List[int]

Returns:

Eligible chunk indices.

gnomad_qc.v5.annotations.batch_fanout.apply_path_suffix(path, suffix)

Insert _<suffix> before the .ht extension; unchanged when suffix is empty.

Parameters:

• path (str) – HT path ending in .ht.

• suffix (Optional[str]) – Suffix without the leading underscore.

Return type:

str

Returns:

Suffixed path.

gnomad_qc.v5.annotations.batch_fanout.combine_suffix(*parts)

Join the non-empty parts with underscores, so each scope writes its own path.

Parameters:

parts (Optional[str]) – Suffix fragments; empty ones are skipped.

Return type:

Optional[str]

Returns:

Joined suffix, or None when every part is empty.

gnomad_qc.v5.annotations.batch_fanout.chunk_path(ht_path, idx, intervals_hash)

Return the per-chunk HT path beside the final HT.

Parameters:

• ht_path (str) – Final HT path the chunks will be merged into.

• idx (int) – Chunk index.

• intervals_hash (str) – Layout hash namespacing the chunk directory.

Return type:

str

Returns:

<ht>_chunks/<hash>/<idx:08d>.chunk.ht.

gnomad_qc.v5.annotations.batch_fanout.group_path(ht_path, level, group_idx, tree_tag, intervals_hash)

Return a merge-tree intermediate HT path.

Level, tree shape and layout hash are all in the directory so a rerun with a different tree or layout writes fresh instead of reusing stale groups.

Parameters:

• ht_path (str) – Final HT path.

• level (int) – Merge-tree level, 1-indexed.

• group_idx (int) – Group index within the level.

• tree_tag (str) – Tree-shape tag the caller builds from its own tree parameters, e.g. coverage passes pp3_gs500 (partitions per chunk, group size). Two consumers may use different tags; each must keep its own tag stable or its finished groups stop being found.

• intervals_hash (str) – Layout hash.

Return type:

str

Returns:

<ht>_merge_groups_<tag>/<hash>/L<level>_<group>.ht.

gnomad_qc.v5.annotations.batch_fanout.failed_chunks_path(ht_path, intervals_hash)

Return the failed-chunk manifest path for one layout.

Parameters:

• ht_path (str) – Final HT path.

• intervals_hash (str) – Layout hash.

Return type:

str

Returns:

<ht>_chunks/<hash>/_failed_chunks.json.

gnomad_qc.v5.annotations.batch_fanout.list_present_chunk_indices(ht_path, intervals_hash)

Return the chunk indices that have a _SUCCESS marker.

One listing is cheaper than one GCS stat per chunk, and looking for _SUCCESS rather than the chunk directory means a half-written chunk is rerun. A missing directory lists as empty.

Parameters:

• ht_path (str) – Final HT path.

• intervals_hash (str) – Layout hash.

Return type:

Set[int]

Returns:

Completed chunk indices.

gnomad_qc.v5.annotations.batch_fanout.new_run_id()

Return a UTC timestamp id for one orchestrator run.

Return type:

str

Returns:

run-YYYYmmddTHHMMSSZ.

gnomad_qc.v5.annotations.batch_fanout.write_failed_chunks_manifest(ht_path, intervals_hash, failed, n_dispatched, run_id, commit, app_name, waves)

Record which dispatched chunks did not land.

A rerun of the fan-out resumes from the missing chunks on its own. The manifest is a durable record of what ran and what failed, which the logs are not, so it is written even when every chunk landed (n_failed 0 with the run id and batch ids). Successive runs on one layout overwrite the same file.

Parameters:

• ht_path (str) – Final HT path.

• intervals_hash (str) – Layout hash.

• failed (Sequence[int]) – Dispatched chunk indices with no _SUCCESS.

• n_dispatched (int) – Chunks dispatched by this run.

• run_id (str) – This orchestrator run’s id.

• commit (str) – gnomad_qc commit the relays ran.

• app_name (Optional[str]) – --app-name the relays used.

• waves (Sequence[Dict[str, Any]]) – Per-wave records: wave number, batch id, chunks dispatched and chunks that failed.

Return type:

str

Returns:

Manifest path.

gnomad_qc.v5.annotations.batch_fanout.dispatch_in_waves(pending, wave_size, submit_wave, present_indices, *, dry_run, on_wave_complete=None, retry_passes=0)

Submit pending chunks in sequential waves and report the ones that did not land.

Waves cap how many relays, and so how many nested QoB drivers, run at once. A batch run does not raise when a job fails, so the chunk directory is listed again after each wave to find the chunks that did not land.

After the last wave, the chunks still missing are re-dispatched up to retry_passes times. This is safe where Batch-level attempts are not: by the time a pass runs, every relay from the previous dispatch has exited, so a re-dispatch cannot race a live relay’s inner QoB batch.

Parameters:

• pending (Sequence[int]) – Chunk indices to dispatch.

• wave_size (int) – Chunks per wave; <= 0 means one wave.

• submit_wave (Callable[[List[int], Optional[str]], Optional[int]]) – Submits one wave and returns its batch id. It must honor the dry run itself (pass dry_run on to submit_relay_batch()); this function only stops after the first wave. Retry passes get the labels retry1, retry2, …

• present_indices (Callable[[], Set[int]]) – Lists the chunk indices with _SUCCESS.

• dry_run (bool) – Stop after the first wave without checking its outputs.

• on_wave_complete (Optional[Callable[[List[int], List[Dict[str, Any]]], None]]) – Called after each wave and each retry pass with the failed indices and wave records so far, e.g. to rewrite the failed-chunk manifest so it survives an orchestrator death. Not called on a dry run. The lists are this function’s live accumulators: read them, do not keep or change them.

• retry_passes (int) – Re-dispatch passes for missing chunks after the waves.

Return type:

Tuple[List[int], List[Dict[str, Any]]]

Returns:

(failed indices, wave records); the failed list is what is still missing after the last retry pass. Both are empty when nothing was pending and on a dry run, so a caller must check dry_run before treating an empty failed list as “every chunk landed”.

gnomad_qc.v5.annotations.batch_fanout.orchestrate_tree_merge(inputs, final_output_path, group_path_fn, merge_group_size, overwrite, final_coalesce_to, batch_prefix, job_prefix, setup_cmd, script, common_flags, merge_cpu, merge_memory, merge_storage, final_merge_storage, submit, attempts=1)

Union inputs into final_output_path with a tree of relay merge jobs.

With thousands of chunks, one job that unions everything would need a huge input list and too much memory, so the merge is a tree. Each level groups its inputs merge_group_size at a time and submits one --run-merge job per group; a group whose HT already exists is skipped unless overwrite is set. The last level is a single job that writes the final HT, and it always runs and overwrites, so a caller that wants to keep an existing final HT checks for it before calling.

Parameters:

• inputs (Sequence[str]) – Chunk HT paths, in order.

• final_output_path (str) – Final HT path.

• group_path_fn (Callable[[int, int], str]) – (level, group_idx) -> intermediate HT path.

• merge_group_size (int) – Inputs per merge job.

• overwrite (bool) – Recompute intermediate groups that already exist.

• final_coalesce_to (Optional[int]) – --merge-coalesce-to for the final job, or None.

• batch_prefix (str) – Batch name prefix; level and final are appended.

• job_prefix (str) – Job name prefix.

• setup_cmd (str) – Relay shell prefix from build_setup_command().

• script (str) – Script command inside the relay container.

• common_flags (str) – Flags appended to every --run-merge command.

• merge_cpu (float) – CPU per merge job.

• merge_memory (str) – Memory class per merge job.

• merge_storage (str) – Storage per intermediate merge job.

• final_merge_storage (str) – Storage for the final merge job.

• submit (Callable[[str, List[RelayJobSpec], str], Optional[int]]) – (batch_name, job_specs, log_label) -> batch id.

• attempts (int) – Batch attempts per merge job. Default 1.

Return type:

None

Returns:

None.

gnomad_qc.v5.annotations.batch_fanout.union_and_write_hts(input_paths, output_path, coalesce_to=None)

Union HTs that share a schema and globals, and write the result.

Needs a running Hail context. Table.union keeps the globals of the first input, which loses nothing because every chunk was built from the same inputs and carries the same globals.

Parameters:

• input_paths (Sequence[str]) – HT paths to union.

• output_path (str) – Destination path, overwritten.

• coalesce_to (Optional[int]) – naive_coalesce target before writing, or None.

Return type:

None

Returns:

None.