minimal new variants changes (#1214)

* minimal new variants changes

* missed one

Author: bpblanken

v03_pipeline/lib/core/constants.py

@@ -1,6 +1,7 @@
 import os
 
 MIGRATION_RUN_ID = 'hail_search_to_clickhouse_migration'
+VARIANTS_MIGRATION_RUN_ID = 'annotations_to_variants_tables'
 GRCH37_TO_GRCH38_LIFTOVER_REF_PATH = (
     'gs://hail-common/references/grch37_to_grch38.over.chain.gz'
     if os.environ.get('HAIL_DATAPROC') == '1'

v03_pipeline/lib/core/dataset_type.py

@@ -370,6 +370,10 @@ def export_vcf_annotation_fns(self) -> list[Callable[..., hl.Expression]]:
     def should_write_new_transcripts(self):
         return self == DatasetType.SNV_INDEL
 
+    @property
+    def should_write_new_variant_details(self):
+        return self == DatasetType.SNV_INDEL
+
     @property
     def overwrite_male_non_par_calls(self) -> None:
         return self == DatasetType.SV

v03_pipeline/lib/paths.py

@@ -353,6 +353,21 @@ def new_transcripts_parquet_path(
     )
 
 
+def new_variant_details_parquet_path(
+    reference_genome: ReferenceGenome,
+    dataset_type: DatasetType,
+    run_id: str,
+) -> str:
+    return os.path.join(
+        runs_path(
+            reference_genome,
+            dataset_type,
+        ),
+        run_id,
+        'new_variant_details.parquet',
+    )
+
+
 def new_variants_parquet_path(
     reference_genome: ReferenceGenome,
     dataset_type: DatasetType,

v03_pipeline/lib/tasks/dataproc/base_run_job_on_dataproc.py

@@ -1,3 +1,4 @@
+import re
 import time
 
 import google.api_core.exceptions
@@ -45,7 +46,7 @@ def task(self):
 
     @property
     def job_id(self):
-        return f'{self.task.task_family}-{self.reference_genome[-2:]}-{self.run_id}-{self.attempt_id}'
+        return f'{self.task.task_family}-{self.reference_genome[-2:]}-{re.sub("_", "-", self.dataset_type.value)}-{self.run_id}-{self.attempt_id}'
 
     def requires(self) -> [luigi.Task]:
         return [self.clone(CreateDataprocClusterTask)]

v03_pipeline/lib/tasks/dataproc/misc.py

@@ -3,16 +3,16 @@
 import luigi
 
 from v03_pipeline.lib.core import Env, ReferenceGenome
-from v03_pipeline.lib.core.constants import (
-    MIGRATION_RUN_ID,
-)
+from v03_pipeline.lib.core.constants import MIGRATION_RUN_ID, VARIANTS_MIGRATION_RUN_ID
 
 CLUSTER_NAME_PREFIX = 'pipeline-runner'
 
 
 def get_cluster_name(reference_genome: ReferenceGenome, run_id: str):
     if MIGRATION_RUN_ID in run_id:
         return f'{Env.DEPLOYMENT_TYPE}-{CLUSTER_NAME_PREFIX}-{reference_genome.value.lower()}-hs-to-clckhse-mgrtn'
+    if VARIANTS_MIGRATION_RUN_ID in run_id:
+        return f'{Env.DEPLOYMENT_TYPE}-{CLUSTER_NAME_PREFIX}-{reference_genome.value.lower()}-vrnts-mgrtn'
     return f'{Env.DEPLOYMENT_TYPE}-{CLUSTER_NAME_PREFIX}-{reference_genome.value.lower()}-{run_id}'
 
 

v03_pipeline/lib/tasks/dataproc/run_pipeline_on_dataproc_test.py

@@ -54,7 +54,7 @@ def test_job_already_exists_failed(
         mock_logger.error.assert_has_calls(
             [
                 call(
-                    'Job RunPipelineTask-38-manual__2024-04-03-0 entered ERROR state',
+                    'Job RunPipelineTask-38-SNV-INDEL-manual__2024-04-03-0 entered ERROR state',
                 ),
             ],
         )
@@ -128,7 +128,7 @@ def test_job_failed(
         mock_logger.info.assert_has_calls(
             [
                 call(
-                    'Waiting for Job completion RunPipelineTask-38-manual__2024-04-05-1',
+                    'Waiting for Job completion RunPipelineTask-38-SNV-INDEL-manual__2024-04-05-1',
                 ),
             ],
         )
@@ -175,8 +175,10 @@ def test_job_success(
         mock_logger.info.assert_has_calls(
             [
                 call(
-                    'Waiting for Job completion RunPipelineTask-38-manual__2024-04-06-0',
+                    'Waiting for Job completion RunPipelineTask-38-SNV-INDEL-manual__2024-04-06-0',
+                ),
+                call(
+                    'Job RunPipelineTask-38-SNV-INDEL-manual__2024-04-06-0 is complete',
                 ),
-                call('Job RunPipelineTask-38-manual__2024-04-06-0 is complete'),
             ],
         )