From 6ab0c3140ff887392004fc02590b844412147361 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 16:50:50 -0400 Subject: [PATCH 01/53] get variant annotations from clickhouse not hail --- loading_pipeline/lib/misc/clickhouse.py | 28 ++++++++--- loading_pipeline/lib/tasks/exports/fields.py | 24 ++-------- .../exports/write_new_entries_parquet.py | 47 +------------------ 3 files changed, 27 insertions(+), 72 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 480356abbc..1f79e0ed1e 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -655,9 +655,13 @@ def insert_new_entries( ) ] common, overrides = [c for c in dst_cols if c in src_cols], {} - if 'geneId_ids' in dst_cols and 'geneIds' in src_cols: - common = [c for c in common if c not in ('geneId_ids', 'geneIds')] + if 'xpos' not in common: + common.append('xpos') + overrides['xpos'] = 'v.xpos' + + if 'geneId_ids' in dst_cols: common.append('geneId_ids') + gene_list_field = 'sortedGeneConsequences' if table_name_builder.dataset_type == DatasetType.SV else 'sortedTranscriptConsequences' overrides['geneId_ids'] = f""" arrayFilter( x -> x IS NOT NULL, @@ -667,7 +671,7 @@ def insert_new_entries( 'seqrdb_id', g ), - geneIds + arrayDistinct(v.{gene_list_field}.geneId) ) ) """ @@ -681,13 +685,25 @@ def insert_new_entries( dictGetOrDefault({ClickhouseReferenceDataset.GNOMAD_GENOMES.search_path(table_name_builder)}, 'filter_af', key, 0) > 0.05 """ + if table_name_builder.dataset_type == DatasetType.GCNV: + overrides + dst_list = ', '.join(common) - src_list = ', '.join([overrides.get(c, c) for c in common]) + src_list = ', '.join([overrides.get(c, f'e.{c}') for c in common]) logged_query( f""" INSERT INTO {table_name_builder.staging_dst_table(ClickHouseTable.ENTRIES)} ({dst_list}) - SELECT {src_list} - FROM {table_name_builder.src_table(ClickHouseTable.ENTRIES)} + SELECT e.key, {src_list} + FROM ( + SELECT + dst.key, + COLUMNS('.*') EXCEPT(variantId, key) + FROM {table_name_builder.src_table(ClickHouseTable.ENTRIES)} src + INNER JOIN {table_name_builder.dst_table(ClickHouseTable.KEY_LOOKUP)} dst + ON {ClickHouseTable.KEY_LOOKUP.join_condition} + ) e + INNER JOIN {table_name_builder.dst_table(ClickHouseTable.VARIANTS_MEMORY)} v + ON assumeNotNull(e.key) = v.key """, # nosec B608 ) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index a29ba5b95c..e08a8a7ba1 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -1,5 +1,6 @@ import hail as hl +from loading_pipeline.lib.annotations.shared import xpos from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType from loading_pipeline.lib.tasks.exports.misc import ( reformat_transcripts_for_export, @@ -100,7 +101,7 @@ def get_existing_variants_export_field(dataset_type: DatasetType) -> str: return f'key AS key_, variantId AS variant_id {dt_fields}' -def get_entries_call_annotations_fields( +def _get_entries_call_annotations_fields( dataset_type: DatasetType, ): if dataset_type == DatasetType.GCNV: @@ -154,7 +155,7 @@ def _get_calls_export_fields( getattr(fe, f'sample_{field}'), getattr(ht, field), ) - for field in get_entries_call_annotations_fields(dataset_type) + for field in _get_entries_call_annotations_fields(dataset_type) }, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, @@ -163,20 +164,6 @@ def _get_calls_export_fields( }[dataset_type](fe) -def get_entries_annotations_export_fields(dataset_type: DatasetType): - fields = { - 'key_': lambda ht: ht.key_, - 'xpos': lambda ht: hl.int64(ht.xpos), - } - if dataset_type in {DatasetType.SV, DatasetType.SNV_INDEL}: - fields['geneIds'] = lambda ht: ( - hl.set(ht.sorted_gene_consequences.gene_id) - if dataset_type == DatasetType.SV - else hl.set(ht.sorted_transcript_consequences.gene_id) - ) - return fields - - def get_entries_export_fields( ht: hl.Table, dataset_type: DatasetType, @@ -185,13 +172,10 @@ def get_entries_export_fields( return { 'project_guid': ht.family_entries.project_guid[0], 'family_guid': ht.family_entries.family_guid[0], - **{ - field: getattr(ht, field) - for field in get_entries_annotations_export_fields(dataset_type) - }, **( { 'sample_type': sample_type.value, + 'xpos': xpos(ht), } if dataset_type in {DatasetType.SNV_INDEL, DatasetType.MITO} else {} diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py index 810c8e56f8..dfd4402803 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py @@ -3,13 +3,11 @@ import luigi.util from loading_pipeline.lib.annotations.fields import get_fields -from loading_pipeline.lib.annotations.shared import xpos from loading_pipeline.lib.misc.family_entries import ( compute_callset_family_entries_ht, deduplicate_by_most_non_ref_calls, deglobalize_ids, ) -from loading_pipeline.lib.misc.io import import_parquet from loading_pipeline.lib.paths import ( new_entries_parquet_path, ) @@ -18,17 +16,9 @@ ) from loading_pipeline.lib.tasks.base.base_write_parquet import BaseWriteParquetTask from loading_pipeline.lib.tasks.exports.fields import ( - get_entries_annotations_export_fields, - get_entries_call_annotations_fields, get_entries_export_fields, ) from loading_pipeline.lib.tasks.files import GCSorLocalTarget -from loading_pipeline.lib.tasks.write_existing_variants_parquet import ( - WriteExistingVariantsParquetTask, -) -from loading_pipeline.lib.tasks.write_new_variants_table import ( - WriteNewVariantsTableTask, -) from loading_pipeline.lib.tasks.write_remapped_and_subsetted_callset import ( WriteRemappedAndSubsettedCallsetTask, ) @@ -47,45 +37,11 @@ def output(self) -> luigi.Target: def requires(self) -> list[luigi.Task]: return [ - self.clone(WriteExistingVariantsParquetTask), - self.clone(WriteNewVariantsTableTask), self.clone(WriteRemappedAndSubsettedCallsetTask), ] def create_table(self) -> hl.Table: - annotations_ht = hl.read_table(self.input()[1].path) - annotation_selects = { - field: func(annotations_ht) - for field, func in { - **get_entries_annotations_export_fields(self.dataset_type), - **get_entries_call_annotations_fields(self.dataset_type), - }.items() - } - annotations_ht = annotations_ht.select(**annotation_selects) - - existing_annotations_ht = import_parquet( - self.input()[0].path, - self.reference_genome, - self.dataset_type, - ) - if 'xpos' not in existing_annotations_ht.row: - existing_annotations_ht = existing_annotations_ht.annotate( - xpos=hl.int64(xpos(existing_annotations_ht)), - ) - if 'gene_ids' in existing_annotations_ht.row: - existing_annotations_ht = existing_annotations_ht.annotate( - gene_ids=hl.set(existing_annotations_ht.gene_ids), - ) - if 'geneIds' in existing_annotations_ht.row: - existing_annotations_ht = existing_annotations_ht.annotate( - geneIds=hl.set(existing_annotations_ht.geneIds), - ) - - annotations_ht = annotations_ht.union( - existing_annotations_ht.select(*annotation_selects), - ) - - mt = hl.read_matrix_table(self.input()[2].path) + mt = hl.read_matrix_table(self.input()[0].path) ht = compute_callset_family_entries_ht( self.dataset_type, mt, @@ -97,7 +53,6 @@ def create_table(self) -> hl.Table: ) ht = deglobalize_ids(ht) ht = deduplicate_by_most_non_ref_calls(ht) - ht = ht.join(annotations_ht) # the family entries ht will contain rows # where at least one family is defined... after explosion, From 36fcca9eb5f5743fd359756b83e00c2fae46d01d Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 16:58:48 -0400 Subject: [PATCH 02/53] export variant id --- loading_pipeline/lib/tasks/exports/fields.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index e08a8a7ba1..b10fccc70e 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -1,6 +1,6 @@ import hail as hl -from loading_pipeline.lib.annotations.shared import xpos +from loading_pipeline.lib.annotations.shared import variant_id, xpos from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType from loading_pipeline.lib.tasks.exports.misc import ( reformat_transcripts_for_export, @@ -175,10 +175,11 @@ def get_entries_export_fields( **( { 'sample_type': sample_type.value, + 'variantId': variant_id(ht), 'xpos': xpos(ht), } if dataset_type in {DatasetType.SNV_INDEL, DatasetType.MITO} - else {} + else {'variantId': ht.variant_id} ), 'filters': ht.filters, 'calls': hl.sorted(ht.family_entries, key=lambda fe: fe.s).map( From cea65620be90db917717f82ba7b3d8c1c0703b08 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 17:05:20 -0400 Subject: [PATCH 03/53] clean up export fields --- loading_pipeline/lib/tasks/exports/fields.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index b10fccc70e..6d890f84c8 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -92,11 +92,12 @@ def get_dataset_type_specific_variants_annotations( def get_existing_variants_export_field(dataset_type: DatasetType) -> str: + # TODO should really be querying from key lookup table dt_fields = { - DatasetType.SNV_INDEL: ', transcripts.geneId AS geneIds', + DatasetType.SNV_INDEL: '', DatasetType.MITO: '', - DatasetType.SV: ', CAST(xpos AS Int64) AS xpos, end, endChrom, sortedGeneConsequences.geneId AS geneIds', - DatasetType.GCNV: ', CAST(xpos AS Int64) AS xpos, pos AS start, end, numExon as num_exon, sortedGeneConsequences.geneId AS gene_ids', + DatasetType.SV: ', end, endChrom', + DatasetType.GCNV: '', }[dataset_type] return f'key AS key_, variantId AS variant_id {dt_fields}' From f58717bff679791908b11c8a3d093a2d50d0fbfb Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 17:23:16 -0400 Subject: [PATCH 04/53] write existing cleanup --- loading_pipeline/lib/misc/clickhouse.py | 12 ++++-------- loading_pipeline/lib/tasks/exports/fields.py | 11 ----------- .../lib/tasks/write_existing_variants_parquet.py | 3 --- 3 files changed, 4 insertions(+), 22 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 1f79e0ed1e..5f230d5a17 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -968,28 +968,24 @@ def export_existing_variants_to_parquet( reference_genome: ReferenceGenome, dataset_type: DatasetType, run_id: str, - export_select_fields: str, ) -> None: table_name_builder = TableNameBuilder( reference_genome, dataset_type, run_id, ) - variants_table = table_name_builder.dst_table( - ClickHouseTable.VARIANT_DETAILS - if dataset_type.should_write_new_variant_details - else ClickHouseTable.VARIANTS_MEMORY, - ) + variants_table = table_name_builder.dst_table(ClickHouseTable.VARIANTS_MEMORY) export_table = table_name_builder.src_table( ClickHouseTable.EXISTING_VARIANTS, ).replace( - '/*.parquet', + '/*.parquet.gz', '', ) + dt_fields = ', end, endChrom' if dataset_type == DatasetType.SV else '' logged_query( f""" INSERT INTO FUNCTION {export_table} - SELECT {export_select_fields} + SELECT key AS key_, variantId AS variant_id {dt_fields} FROM {variants_table} """, # nosec B608 ) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 6d890f84c8..020b92915f 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -91,17 +91,6 @@ def get_dataset_type_specific_variants_annotations( }[dataset_type](ht) -def get_existing_variants_export_field(dataset_type: DatasetType) -> str: - # TODO should really be querying from key lookup table - dt_fields = { - DatasetType.SNV_INDEL: '', - DatasetType.MITO: '', - DatasetType.SV: ', end, endChrom', - DatasetType.GCNV: '', - }[dataset_type] - return f'key AS key_, variantId AS variant_id {dt_fields}' - - def _get_entries_call_annotations_fields( dataset_type: DatasetType, ): diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet.py index da89e0ce79..d108128428 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet.py @@ -6,7 +6,6 @@ from loading_pipeline.lib.tasks.base.base_loading_run_params import ( BaseLoadingRunParams, ) -from loading_pipeline.lib.tasks.exports.fields import get_existing_variants_export_field from loading_pipeline.lib.tasks.files import ( GCSorLocalTarget, ) @@ -27,10 +26,8 @@ def complete(self) -> bool: return self.output().exists() def run(self): - export_select_fields = get_existing_variants_export_field(self.dataset_type) export_existing_variants_to_parquet( self.reference_genome, self.dataset_type, self.run_id, - export_select_fields, ) From 7dc09fb7ff80de23788bc56bd9b3ddcf29f68f37 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 17:25:56 -0400 Subject: [PATCH 05/53] correct table source; --- loading_pipeline/lib/misc/clickhouse.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 5f230d5a17..01ba7c6f7f 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -974,7 +974,11 @@ def export_existing_variants_to_parquet( dataset_type, run_id, ) - variants_table = table_name_builder.dst_table(ClickHouseTable.VARIANTS_MEMORY) + variants_table = variants_table = table_name_builder.dst_table( + ClickHouseTable.KEY_LOOKUP + if dataset_type.should_write_new_variant_details + else ClickHouseTable.VARIANTS_MEMORY, + ) export_table = table_name_builder.src_table( ClickHouseTable.EXISTING_VARIANTS, ).replace( From 0d3f6163a10490e684a0bfc17ad530ff90ab2226 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 17:26:15 -0400 Subject: [PATCH 06/53] clean up --- loading_pipeline/lib/misc/clickhouse.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 01ba7c6f7f..a04fe1df5a 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -974,7 +974,7 @@ def export_existing_variants_to_parquet( dataset_type, run_id, ) - variants_table = variants_table = table_name_builder.dst_table( + variants_table = table_name_builder.dst_table( ClickHouseTable.KEY_LOOKUP if dataset_type.should_write_new_variant_details else ClickHouseTable.VARIANTS_MEMORY, From f2d15ef012f7224482e54d8761278a0055ff3345 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Fri, 11 Sep 2026 17:32:18 -0400 Subject: [PATCH 07/53] clean up --- loading_pipeline/lib/misc/clickhouse.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index a04fe1df5a..3a5189a876 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -661,7 +661,11 @@ def insert_new_entries( if 'geneId_ids' in dst_cols: common.append('geneId_ids') - gene_list_field = 'sortedGeneConsequences' if table_name_builder.dataset_type == DatasetType.SV else 'sortedTranscriptConsequences' + gene_list_field = ( + 'sortedGeneConsequences' + if table_name_builder.dataset_type == DatasetType.SV + else 'sortedTranscriptConsequences' + ) overrides['geneId_ids'] = f""" arrayFilter( x -> x IS NOT NULL, @@ -685,9 +689,6 @@ def insert_new_entries( dictGetOrDefault({ClickhouseReferenceDataset.GNOMAD_GENOMES.search_path(table_name_builder)}, 'filter_af', key, 0) > 0.05 """ - if table_name_builder.dataset_type == DatasetType.GCNV: - overrides - dst_list = ', '.join(common) src_list = ', '.join([overrides.get(c, f'e.{c}') for c in common]) logged_query( From a50cf6ca7704b84e9f501be4ee5784b8d6c717ce Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 13:40:51 -0400 Subject: [PATCH 08/53] update diagram --- loading_pipeline/docs/Diagrams.md | 51 ++++++++++--------- .../exports/write_new_entries_parquet.py | 4 ++ 2 files changed, 30 insertions(+), 25 deletions(-) diff --git a/loading_pipeline/docs/Diagrams.md b/loading_pipeline/docs/Diagrams.md index 95541d3b3a..e05ea1f382 100644 --- a/loading_pipeline/docs/Diagrams.md +++ b/loading_pipeline/docs/Diagrams.md @@ -26,32 +26,33 @@ WriteRemappedAndSubsettedCallsetTask | v - WriteMetadataForRunTask WriteExistingVariantsParquetTask - |___________________________________________| + WriteMetadataForRunTask | - v - WriteNewVariantsTableTask - | - ______________________+_______________________ - | | | - v v v - WriteNewEntries... WriteNewVariants... WriteNewVariantDetails... - ParquetTask ParquetTask ParquetTask - | | (optional) - |___________________________|_______________________| - | - v - RunPipelineTask - (all parquets ready) - | - v - WriteSuccessFileTask - | - v - LoadClickhouseVariants - | - v - LoadClickhouseEntries + _________________+__________ WriteExistingVariantsParquetTask + | | | + | |______________________| + | | + v v + WriteNewEntriesParquetTask WriteNewVariantsTableTask + | | + | ____________+____________ + | | | + | v v + | WriteNewVariantsParquetTask WriteNewVariantDetailsParquetTask + | | (optional) + |__________________________|_______________________| + | + v + RunPipelineTask (all parquets ready) + | + v + WriteSuccessFileTask + | + v + LoadClickhouseVariants + | + v + LoadClickhouseEntries ``` ClickHouse LSM-Tree diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py index dfd4402803..35bb6cad41 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py @@ -19,6 +19,9 @@ get_entries_export_fields, ) from loading_pipeline.lib.tasks.files import GCSorLocalTarget +from loading_pipeline.lib.tasks.write_metadata_for_run import ( + WriteMetadataForRunTask, +) from loading_pipeline.lib.tasks.write_remapped_and_subsetted_callset import ( WriteRemappedAndSubsettedCallsetTask, ) @@ -38,6 +41,7 @@ def output(self) -> luigi.Target: def requires(self) -> list[luigi.Task]: return [ self.clone(WriteRemappedAndSubsettedCallsetTask), + self.clone(WriteMetadataForRunTask), ] def create_table(self) -> hl.Table: From 4c1737c3230add90311761cc619967f194e52830 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 15:35:54 -0400 Subject: [PATCH 09/53] fix gz for parquet --- loading_pipeline/lib/misc/clickhouse.py | 5 ++++- loading_pipeline/lib/paths.py | 2 +- loading_pipeline/lib/paths_test.py | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 3a5189a876..44a95eb547 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -983,8 +983,11 @@ def export_existing_variants_to_parquet( export_table = table_name_builder.src_table( ClickHouseTable.EXISTING_VARIANTS, ).replace( - '/*.parquet.gz', + '/*.parquet', '', + ).replace( + '.parquet', + '.parquet.gz', ) dt_fields = ', end, endChrom' if dataset_type == DatasetType.SV else '' logged_query( diff --git a/loading_pipeline/lib/paths.py b/loading_pipeline/lib/paths.py index df59e6cc39..1409638623 100644 --- a/loading_pipeline/lib/paths.py +++ b/loading_pipeline/lib/paths.py @@ -320,7 +320,7 @@ def existing_variants_parquet_path( dataset_type, ), run_id, - 'existing_variants.parquet', + 'existing_variants.parquet.gz', ) diff --git a/loading_pipeline/lib/paths_test.py b/loading_pipeline/lib/paths_test.py index 3803bdb92a..c36fdc2d4f 100644 --- a/loading_pipeline/lib/paths_test.py +++ b/loading_pipeline/lib/paths_test.py @@ -72,7 +72,7 @@ def test_existing_variants_parquet_path(self) -> None: DatasetType.GCNV, 'manual__2023-06-26T18:30:09.349671+00:00', ), - '/var/seqr/pipeline-data/GRCh38/GCNV/runs/manual__2023-06-26T18:30:09.349671+00:00/existing_variants.parquet', + '/var/seqr/pipeline-data/GRCh38/GCNV/runs/manual__2023-06-26T18:30:09.349671+00:00/existing_variants.parquet.gz', ) def test_remapped_and_subsetted_callset_path(self) -> None: From e3d9e9525763b34b4a026153ea08e4e5be58d0b1 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 15:58:40 -0400 Subject: [PATCH 10/53] update tests --- .../exports/write_new_entries_parquet_test.py | 77 ++----------------- 1 file changed, 7 insertions(+), 70 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index 05b73e2086..2f4092801b 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -1,6 +1,3 @@ -from typing import ClassVar - -import hail as hl import luigi.worker import pandas as pd @@ -12,14 +9,10 @@ from loading_pipeline.lib.misc.validation import ALL_VALIDATIONS from loading_pipeline.lib.paths import ( new_entries_parquet_path, - new_variants_table_path, ) from loading_pipeline.lib.tasks.exports.write_new_entries_parquet import ( WriteNewEntriesParquetTask, ) -from loading_pipeline.lib.test.clickhouse_schema_testcase import ( - ClickhouseSchemaTestCase, -) from loading_pipeline.lib.test.misc import ( convert_ndarray_to_list, copy_project_pedigree_to_mocked_dir, @@ -36,61 +29,11 @@ TEST_MITO_CALLSET = 'loading_pipeline/var/test/callsets/mito_1.mt' TEST_SV_VCF_2 = 'loading_pipeline/var/test/callsets/sv_2.vcf' TEST_GCNV_BED_FILE = 'loading_pipeline/var/test/callsets/gcnv_1.tsv' -TEST_SNV_INDEL_ANNOTATIONS = ( - 'loading_pipeline/var/test/exports/GRCh38/SNV_INDEL/annotations.ht' -) -TEST_MITO_ANNOTATIONS = 'loading_pipeline/var/test/exports/GRCh38/MITO/annotations.ht' -TEST_SV_ANNOTATIONS = 'loading_pipeline/var/test/exports/GRCh38/SV/annotations.ht' -TEST_GCNV_ANNOTATIONS = 'loading_pipeline/var/test/exports/GRCh38/GCNV/annotations.ht' TEST_RUN_ID = 'manual__2024-04-03' -class WriteNewEntriesParquetTest(MockedDatarootTestCase, ClickhouseSchemaTestCase): - fixtures: ClassVar = ['clickhouse_test'] - - def setUp(self) -> None: - super().setUp() - ht = hl.read_table( - TEST_SNV_INDEL_ANNOTATIONS, - ) - ht = ht.filter(ht.variant_id != '1-878314-G-C') - ht.write( - new_variants_table_path( - ReferenceGenome.GRCh38, - DatasetType.SNV_INDEL, - TEST_RUN_ID, - ), - ) - ht = hl.read_table( - TEST_MITO_ANNOTATIONS, - ) - ht.write( - new_variants_table_path( - ReferenceGenome.GRCh38, - DatasetType.MITO, - TEST_RUN_ID, - ), - ) - ht = hl.read_table( - TEST_SV_ANNOTATIONS, - ) - ht.write( - new_variants_table_path( - ReferenceGenome.GRCh38, - DatasetType.SV, - TEST_RUN_ID, - ), - ) - ht = hl.read_table(TEST_GCNV_ANNOTATIONS) - ht.write( - new_variants_table_path( - ReferenceGenome.GRCh38, - DatasetType.GCNV, - TEST_RUN_ID, - ), - ) - +class WriteNewEntriesParquetTest(MockedDatarootTestCase): def test_write_new_entries_parquet(self): copy_project_pedigree_to_mocked_dir( TEST_PEDIGREE_3_REMAP, @@ -152,12 +95,11 @@ def test_write_new_entries_parquet(self): [export_json[0], export_json[9], export_json[15]], [ { - 'key': 0, 'project_guid': 'R0114_project4', 'family_guid': '123_1', 'sample_type': 'WGS', + 'variantId': '1-876499-A-G', 'xpos': 1000876499, - 'geneIds': ['ENSG00000187634'], 'filters': [], 'calls': [ { @@ -171,12 +113,11 @@ def test_write_new_entries_parquet(self): 'sign': 1, }, { - 'key': 0, 'project_guid': 'R0113_test_project', 'family_guid': 'abc_1', 'sample_type': 'WGS', + 'variantId': '1-876499-A-G', 'xpos': 1000876499, - 'geneIds': ['ENSG00000187634'], 'filters': [], 'calls': [ { @@ -204,12 +145,11 @@ def test_write_new_entries_parquet(self): 'sign': 1, }, { - 'key': 1, 'project_guid': 'R0113_test_project', 'family_guid': 'abc_1', 'sample_type': 'WGS', + 'variantId': '1-878314-G-C', 'xpos': 1000878314, - 'geneIds': ['ENSG00000177000'], 'filters': ['VQSRTrancheSNP99.00to99.90'], 'calls': [ { @@ -267,10 +207,10 @@ def test_mito_write_new_entries_parquet(self): export_json, [ { - 'key': 998, 'project_guid': 'R0116_test_project3', 'family_guid': 'family_1', 'sample_type': 'WGS', + 'variantId': 'M-8-G-T', 'xpos': 25000000008, 'filters': [], 'calls': [ @@ -322,11 +262,9 @@ def test_sv_write_new_entries_parquet(self): export_json, [ { - 'key': 727, 'project_guid': 'R0115_test_project2', 'family_guid': 'family_2_1', - 'xpos': 1001025886, - 'geneIds': ['ENSG00000188157'], + 'variantId': 'BND_chr1_6', 'filters': ['HIGH_SR_BACKGROUND', 'UNRESOLVED'], 'calls': [ { @@ -405,10 +343,9 @@ def test_gcnv_write_new_entries_parquet(self): export_json, [ { - 'key': 0, 'project_guid': 'R0115_test_project2', 'family_guid': 'family_2_1', - 'xpos': 1000939203, + 'variantId': 'suffix_16456_DEL', 'filters': [], 'calls': [ { From 94046e4d7952d855581f5ef45395eccd438f7ae0 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 16:06:04 -0400 Subject: [PATCH 11/53] fix path --- loading_pipeline/lib/misc/clickhouse.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 44a95eb547..53036e1cae 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -985,9 +985,6 @@ def export_existing_variants_to_parquet( ).replace( '/*.parquet', '', - ).replace( - '.parquet', - '.parquet.gz', ) dt_fields = ', end, endChrom' if dataset_type == DatasetType.SV else '' logged_query( From c6084099131f8ad7dfec8a4fdfe502a1ea7d8861 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 16:22:24 -0400 Subject: [PATCH 12/53] moc sleep t speed up tests --- loading_pipeline/lib/misc/clickhouse_test.py | 1 + 1 file changed, 1 insertion(+) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 1126671b17..0bf3f46424 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -193,6 +193,7 @@ } +@patch('time.sleep', return_value=None) class ClickhouseTest(MockedDatarootTestCase, ClickhouseSchemaTestCase): fixtures: ClassVar = ['clickhouse_test'] From 8135f2492d97181f636b13a2320f7eb4e412baf2 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 16:51:23 -0400 Subject: [PATCH 13/53] fix gcnv overrides --- loading_pipeline/lib/tasks/exports/fields.py | 27 +++++--------------- 1 file changed, 7 insertions(+), 20 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 020b92915f..5a9f31642d 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -91,19 +91,6 @@ def get_dataset_type_specific_variants_annotations( }[dataset_type](ht) -def _get_entries_call_annotations_fields( - dataset_type: DatasetType, -): - if dataset_type == DatasetType.GCNV: - return { - 'start': lambda ht: hl.int64(ht.start_locus.position), - 'end': lambda ht: hl.int64(ht.end_locus.position), - 'num_exon': lambda ht: ht.num_exon, - 'gene_ids': lambda ht: hl.set(ht.sorted_gene_consequences.gene_id), - } - return {} - - def _get_calls_export_fields( ht: hl.Table, fe: hl.Struct, @@ -140,13 +127,13 @@ def _get_calls_export_fields( cn=fe.CN, qs=fe.QS, defragged=fe.defragged, - **{ - snake_to_camelcase(field): hl.or_else( - getattr(fe, f'sample_{field}'), - getattr(ht, field), - ) - for field in _get_entries_call_annotations_fields(dataset_type) - }, + start=hl.or_else(fe.sample_start, ht.start_locus.position), + end=hl.or_else(fe.sample_end, ht.end_locus.position), + numExon=hl.or_else(fe.sample_num_exon, ht.num_exon), + geneIds=hl.or_else( + fe.sample_gene_ids, + hl.set(ht.sorted_gene_consequences.gene_id), + ), newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, prevOverlap=fe.concordance.prev_overlap, From 96fac211a91dffdf3ffba1e158dcf7dbf35ccbc6 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 16:52:11 -0400 Subject: [PATCH 14/53] fix tests --- .../exports/write_new_entries_parquet_test.py | 75 +++++++++---------- 1 file changed, 37 insertions(+), 38 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index 2f4092801b..b25e0b6060 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -71,28 +71,28 @@ def test_write_new_entries_parquet(self): ), ) export_json = convert_ndarray_to_list(df.to_dict('records')) - self.assertEqual(len(export_json), 16) + self.assertEqual(len(export_json), 181) self.assertEqual( df['family_guid'].value_counts().to_dict(), { - 'abc_1': 2, - '345_1': 2, - '123_1': 1, - '234_1': 1, - '456_1': 1, - '567_1': 1, - '678_1': 1, - '789_1': 1, - '890_1': 1, - '901_1': 1, - 'bcd_1': 1, - 'cde_1': 1, - 'def_1': 1, - 'efg_1': 1, + 'abc_1': 16, + '789_1': 15, + '890_1': 14, + '901_1': 14, + 'bcd_1': 14, + '345_1': 13, + '456_1': 13, + '567_1': 13, + 'def_1': 13, + '123_1': 12, + '234_1': 11, + '678_1': 11, + 'cde_1': 11, + 'efg_1': 11, }, ) self.assertEqual( - [export_json[0], export_json[9], export_json[15]], + [export_json[2], export_json[11], export_json[17]], [ { 'project_guid': 'R0114_project4', @@ -203,29 +203,28 @@ def test_mito_write_new_entries_parquet(self): ), ) export_json = convert_ndarray_to_list(df.to_dict('records')) + self.assertEqual(len(export_json), 3) self.assertEqual( - export_json, - [ - { - 'project_guid': 'R0116_test_project3', - 'family_guid': 'family_1', - 'sample_type': 'WGS', - 'variantId': 'M-8-G-T', - 'xpos': 25000000008, - 'filters': [], - 'calls': [ - { - 'sampleId': 'RGP_1270_2', - 'gt': 2, - 'dp': 4216, - 'hl': 0.999, - 'mitoCn': 224, - 'contamination': 0.0, - }, - ], - 'sign': 1, - }, - ], + export_json[0], + { + 'project_guid': 'R0116_test_project3', + 'family_guid': 'family_1', + 'sample_type': 'WGS', + 'variantId': 'M-8-G-T', + 'xpos': 25000000008, + 'filters': [], + 'calls': [ + { + 'sampleId': 'RGP_1270_2', + 'gt': 2, + 'dp': 4216, + 'hl': 0.999, + 'mitoCn': 224, + 'contamination': 0.0, + }, + ], + 'sign': 1, + }, ) def test_sv_write_new_entries_parquet(self): From 72b76be77f2a4525e5dd7a48361f6c169cc73c5f Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 17:03:08 -0400 Subject: [PATCH 15/53] use real gcnv ht fields --- loading_pipeline/lib/misc/clickhouse_test.py | 32 +++++++++++--------- loading_pipeline/lib/tasks/exports/fields.py | 6 ++-- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 0bf3f46424..61f864bf11 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -303,7 +303,6 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): # New Entries Parquet df = pd.DataFrame( { - 'key': [10, 3, 4], 'project_guid': [ 'project_d', 'project_d', @@ -314,20 +313,25 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): 'family_d2', 'family_d3', ], - 'xpos': [ - 123456789, - 123456789, - 123456789, - ], 'sample_type': [ 'WES', 'WES', 'WES', ], - 'geneIds': [ + 'variantId': [ + '1-3-A-C', + 'Y-19-A-C', + 'M-12-C-G', + ], + 'xpos': [ + 123456789, + 123456789, + 123456789, + ], + 'filters': [ + [], + [], [], - ['GENE1', 'GENE2'], - ['GENE3'], ], 'calls': [ [('sample_d1', 0), ('sample_d11', 2)], @@ -343,12 +347,12 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): ) schema = pa.schema( [ - ('key', pa.int64()), ('project_guid', pa.string()), ('family_guid', pa.string()), - ('xpos', pa.int64()), ('sample_type', pa.string()), - ('geneIds', pa.list_(pa.string())), + ('variantId', pa.string()), + ('xpos', pa.int64()), + ('filters', pa.list_(pa.string())), ( 'calls', pa.list_( @@ -368,13 +372,13 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): schema, ) write_test_parquet( - df.drop('geneIds', axis=1), + df.drop(['sample_type', 'xpos'], axis=1), new_entries_parquet_path( ReferenceGenome.GRCh38, DatasetType.GCNV, TEST_RUN_ID, ), - schema.remove(5).remove(5), + pa.schema([f for f in schema if f.name not in ('sample_type', 'xpos')]), ) def test_get_clickhouse_client(self): diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 5a9f31642d..cce3d9acf9 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -127,12 +127,12 @@ def _get_calls_export_fields( cn=fe.CN, qs=fe.QS, defragged=fe.defragged, - start=hl.or_else(fe.sample_start, ht.start_locus.position), - end=hl.or_else(fe.sample_end, ht.end_locus.position), + start=hl.or_else(fe.sample_start, ht.start), + end=hl.or_else(fe.sample_end, ht.end), numExon=hl.or_else(fe.sample_num_exon, ht.num_exon), geneIds=hl.or_else( fe.sample_gene_ids, - hl.set(ht.sorted_gene_consequences.gene_id), + hl.set(ht.gene_ids), ), newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, From 70c67137df177a6a2ef2b7aea0cd7d97b8d3934d Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 17:11:59 -0400 Subject: [PATCH 16/53] fix test load entries setup --- loading_pipeline/lib/misc/clickhouse_test.py | 22 +++++++------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 61f864bf11..51dc6589e0 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -313,6 +313,11 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): 'family_d2', 'family_d3', ], + 'xpos': [ + 123456789, + 123456789, + 123456789, + ], 'sample_type': [ 'WES', 'WES', @@ -323,16 +328,6 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): 'Y-19-A-C', 'M-12-C-G', ], - 'xpos': [ - 123456789, - 123456789, - 123456789, - ], - 'filters': [ - [], - [], - [], - ], 'calls': [ [('sample_d1', 0), ('sample_d11', 2)], [('sample_d2', 0)], @@ -349,10 +344,9 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): [ ('project_guid', pa.string()), ('family_guid', pa.string()), + ('xpos', pa.int64()), ('sample_type', pa.string()), ('variantId', pa.string()), - ('xpos', pa.int64()), - ('filters', pa.list_(pa.string())), ( 'calls', pa.list_( @@ -372,13 +366,13 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): schema, ) write_test_parquet( - df.drop(['sample_type', 'xpos'], axis=1), + df.drop(['xpos', 'sample_type'], axis=1), new_entries_parquet_path( ReferenceGenome.GRCh38, DatasetType.GCNV, TEST_RUN_ID, ), - pa.schema([f for f in schema if f.name not in ('sample_type', 'xpos')]), + schema.remove(2).remove(2), ) def test_get_clickhouse_client(self): From 9fcf4cc58b0936c53f702a13302ff6676f78ee8a Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 17:48:53 -0400 Subject: [PATCH 17/53] fix existing parquet test --- .../write_existing_variants_parquet_test.py | 26 +++++++++---------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 1323c0b775..3225690f1d 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -1,3 +1,4 @@ +import gzip from typing import ClassVar import luigi.worker @@ -40,26 +41,23 @@ def _run_task( worker.run() self.assertTrue(task.output().exists()) self.assertTrue(task.complete()) - return pd.read_parquet( + with gzip.open( existing_variants_parquet_path(reference_genome, dataset_type, TEST_RUN_ID), - ) + ) as f: + return pd.read_parquet(f) def test_snv_indel(self): df = self._run_task(DatasetType.SNV_INDEL) - self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) df = df.sort_values('key_').reset_index(drop=True) self.assertEqual( convert_ndarray_to_list( - df[['key_', 'variant_id', 'geneIds']].to_dict('records'), + df[['key_', 'variant_id']].to_dict('records'), ), [ - { - 'key_': 1, - 'variant_id': '1-878314-G-C', - 'geneIds': ['ENSG00000177000'], - }, - {'key_': 7, 'variant_id': '7-1234567-AGT-A', 'geneIds': []}, - {'key_': 10, 'variant_id': '10-987654-G-A', 'geneIds': []}, + {'key_': 1, 'variant_id': '1-878314-G-C'}, + {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, + {'key_': 10, 'variant_id': '10-987654-G-A'}, ], ) @@ -68,7 +66,7 @@ def test_grch37_snv_indel(self): DatasetType.SNV_INDEL, reference_genome=ReferenceGenome.GRCh37, ) - self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) self.assertEqual(len(df), 0) def test_mito(self): @@ -80,7 +78,7 @@ def test_sv(self): df = self._run_task(DatasetType.SV) self.assertEqual( list(df.columns), - ['key_', 'variant_id', 'xpos', 'end', 'endChrom', 'geneIds'], + ['key_', 'variant_id', 'end', 'endChrom'], ) self.assertEqual(len(df), 0) @@ -88,6 +86,6 @@ def test_gcnv(self): df = self._run_task(DatasetType.GCNV) self.assertEqual( list(df.columns), - ['key_', 'variant_id', 'xpos', 'start', 'end', 'num_exon', 'gene_ids'], + ['key_', 'variant_id'], ) self.assertEqual(len(df), 0) From 978fc3db7312447d2aa8b40c1a7320bf54b818ba Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Mon, 14 Sep 2026 17:55:40 -0400 Subject: [PATCH 18/53] ruff --- loading_pipeline/lib/tasks/exports/fields.py | 1 - 1 file changed, 1 deletion(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index cce3d9acf9..b35dea1f2b 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -4,7 +4,6 @@ from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType from loading_pipeline.lib.tasks.exports.misc import ( reformat_transcripts_for_export, - snake_to_camelcase, ) STANDARD_CONTIGS = hl.set( From 4665067902cb92703cb5bca3d10d95c0f4e9293a Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 09:25:22 -0400 Subject: [PATCH 19/53] update sleep patch --- loading_pipeline/lib/misc/clickhouse_test.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 51dc6589e0..397ec3b270 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -193,7 +193,6 @@ } -@patch('time.sleep', return_value=None) class ClickhouseTest(MockedDatarootTestCase, ClickhouseSchemaTestCase): fixtures: ClassVar = ['clickhouse_test'] @@ -212,6 +211,10 @@ def setUpClass(cls): def setUp(self): super().setUp() + sleep_patch = patch('time.sleep') + sleep_patch.start() + self.addCleanup(sleep_patch.stop) + base_path = runs_path( ReferenceGenome.GRCh38, DatasetType.SNV_INDEL, From c736a6741419261e9a5944506c8de1cc8aa65411 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 09:35:19 -0400 Subject: [PATCH 20/53] update fixture data --- .../lib/test/fixtures/clickhouse_test.json | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/loading_pipeline/lib/test/fixtures/clickhouse_test.json b/loading_pipeline/lib/test/fixtures/clickhouse_test.json index 9de4810176..d7726ed9dc 100644 --- a/loading_pipeline/lib/test/fixtures/clickhouse_test.json +++ b/loading_pipeline/lib/test/fixtures/clickhouse_test.json @@ -302,5 +302,29 @@ "sorted_motif_feature_consequences": [], "sorted_regulatory_feature_consequences": [] } + }, + { + "model": "clickhouse_search.keylookupsnvindel", + "pk": "1-878314-G-C", + "fields": { + "variant_id": "1-878314-G-C", + "key": 1 + } + }, + { + "model": "clickhouse_search.keylookupsnvindel", + "pk": "7-1234567-AGT-A", + "fields": { + "variant_id": "7-1234567-AGT-A", + "key": 7 + } + }, + { + "model": "clickhouse_search.keylookupsnvindel", + "pk": "10-987654-G-A", + "fields": { + "variant_id": "10-987654-G-A", + "key": 10 + } } ] From 180b57b824221cf5b2a0fabb80b127c15aad911f Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 09:52:04 -0400 Subject: [PATCH 21/53] fix test setu --- .../exports/write_new_entries_parquet_test.py | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index b25e0b6060..c2ae4ee420 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -1,3 +1,5 @@ +import os + import luigi.worker import pandas as pd @@ -8,6 +10,7 @@ ) from loading_pipeline.lib.misc.validation import ALL_VALIDATIONS from loading_pipeline.lib.paths import ( + existing_variants_parquet_path, new_entries_parquet_path, ) from loading_pipeline.lib.tasks.exports.write_new_entries_parquet import ( @@ -235,6 +238,20 @@ def test_sv_write_new_entries_parquet(self): SampleType.WGS, 'R0115_test_project2', ) + existing_variants_path = existing_variants_parquet_path( + ReferenceGenome.GRCh38, + DatasetType.SV, + TEST_RUN_ID, + ) + os.makedirs(os.path.dirname(existing_variants_path), exist_ok=True) + pd.DataFrame( + { + 'variant_id': ['BND_chr1_6'], + 'key_': [727], + 'end': [180928], + 'endChrom': ['chr5'], + }, + ).to_parquet(existing_variants_path) worker = luigi.worker.Worker() task = WriteNewEntriesParquetTask( reference_genome=ReferenceGenome.GRCh38, From 127a1430508d35f1c14ce61158572e3cddb754da Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 09:56:39 -0400 Subject: [PATCH 22/53] gcnv call fields --- loading_pipeline/lib/tasks/exports/fields.py | 28 +++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index b35dea1f2b..020b92915f 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -4,6 +4,7 @@ from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType from loading_pipeline.lib.tasks.exports.misc import ( reformat_transcripts_for_export, + snake_to_camelcase, ) STANDARD_CONTIGS = hl.set( @@ -90,6 +91,19 @@ def get_dataset_type_specific_variants_annotations( }[dataset_type](ht) +def _get_entries_call_annotations_fields( + dataset_type: DatasetType, +): + if dataset_type == DatasetType.GCNV: + return { + 'start': lambda ht: hl.int64(ht.start_locus.position), + 'end': lambda ht: hl.int64(ht.end_locus.position), + 'num_exon': lambda ht: ht.num_exon, + 'gene_ids': lambda ht: hl.set(ht.sorted_gene_consequences.gene_id), + } + return {} + + def _get_calls_export_fields( ht: hl.Table, fe: hl.Struct, @@ -126,13 +140,13 @@ def _get_calls_export_fields( cn=fe.CN, qs=fe.QS, defragged=fe.defragged, - start=hl.or_else(fe.sample_start, ht.start), - end=hl.or_else(fe.sample_end, ht.end), - numExon=hl.or_else(fe.sample_num_exon, ht.num_exon), - geneIds=hl.or_else( - fe.sample_gene_ids, - hl.set(ht.gene_ids), - ), + **{ + snake_to_camelcase(field): hl.or_else( + getattr(fe, f'sample_{field}'), + getattr(ht, field), + ) + for field in _get_entries_call_annotations_fields(dataset_type) + }, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, prevOverlap=fe.concordance.prev_overlap, From 5ddc35803a2c7440c49352c8237396c84c942ad1 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 10:47:57 -0400 Subject: [PATCH 23/53] actually fix gcnv --- loading_pipeline/lib/core/dataset_type.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/loading_pipeline/lib/core/dataset_type.py b/loading_pipeline/lib/core/dataset_type.py index e9bccc99f6..6d6430ba14 100644 --- a/loading_pipeline/lib/core/dataset_type.py +++ b/loading_pipeline/lib/core/dataset_type.py @@ -324,6 +324,10 @@ def genotype_entry_annotation_fns(self) -> list[Callable[..., hl.Expression]]: gcnv.CN, gcnv.GT, gcnv.QS, + gcnv.start_locus, + gcnv.end_locus, + gcnv.num_exon, + gcnv.sorted_gene_consequences, ], }[self] From 72b58fa3aaaa9dd47530770a956e389e57f4ca1c Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 10:51:42 -0400 Subject: [PATCH 24/53] fix key handling for entries insert --- loading_pipeline/lib/misc/clickhouse.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 53036e1cae..6216da98a2 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -655,6 +655,7 @@ def insert_new_entries( ) ] common, overrides = [c for c in dst_cols if c in src_cols], {} + common.insert(0, 'key') if 'xpos' not in common: common.append('xpos') overrides['xpos'] = 'v.xpos' @@ -694,7 +695,7 @@ def insert_new_entries( logged_query( f""" INSERT INTO {table_name_builder.staging_dst_table(ClickHouseTable.ENTRIES)} ({dst_list}) - SELECT e.key, {src_list} + SELECT {src_list} FROM ( SELECT dst.key, From 94eb9a39bcded1def65abb32fe29fc583e38411b Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 10:58:24 -0400 Subject: [PATCH 25/53] fix test --- .../exports/write_new_entries_parquet_test.py | 95 +++++++++---------- 1 file changed, 47 insertions(+), 48 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index c2ae4ee420..13e852a627 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -274,55 +274,54 @@ def test_sv_write_new_entries_parquet(self): ), ) export_json = convert_ndarray_to_list(df.to_dict('records')) + self.assertEqual(len(export_json), 2) self.assertEqual( - export_json, - [ - { - 'project_guid': 'R0115_test_project2', - 'family_guid': 'family_2_1', - 'variantId': 'BND_chr1_6', - 'filters': ['HIGH_SR_BACKGROUND', 'UNRESOLVED'], - 'calls': [ - { - 'sampleId': 'RGP_164_1', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_2', - 'gt': 1, - 'cn': None, - 'gq': 31, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_3', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_4', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - ], - 'sign': 1, - }, - ], + export_json[0], + { + 'project_guid': 'R0115_test_project2', + 'family_guid': 'family_2_1', + 'variantId': 'BND_chr1_6', + 'filters': ['HIGH_SR_BACKGROUND', 'UNRESOLVED'], + 'calls': [ + { + 'sampleId': 'RGP_164_1', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_2', + 'gt': 1, + 'cn': None, + 'gq': 31, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_3', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_4', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + ], + 'sign': 1, + }, ) def test_gcnv_write_new_entries_parquet(self): From b06d2caf8e2e89043627fa180c2d4f110a41b2e5 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 11:53:22 -0400 Subject: [PATCH 26/53] correct funcs for gcnv overrides --- loading_pipeline/lib/tasks/exports/fields.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 020b92915f..aa8445d127 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -143,9 +143,9 @@ def _get_calls_export_fields( **{ snake_to_camelcase(field): hl.or_else( getattr(fe, f'sample_{field}'), - getattr(ht, field), + get_value(ht), ) - for field in _get_entries_call_annotations_fields(dataset_type) + for field, get_value in _get_entries_call_annotations_fields(dataset_type).items() }, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, From 1f5a264a7076c1908dedd0103aa3f0849f20211b Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 11:57:39 -0400 Subject: [PATCH 27/53] load entires test dependencies --- loading_pipeline/lib/misc/clickhouse_test.py | 25 +++++++++++++++++++- 1 file changed, 24 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 397ec3b270..087abc10d3 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -464,6 +464,12 @@ def test_direct_insert_all_keys(self): return_value=[ClickhouseReferenceDataset.CLINVAR], ) def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): + load_run_variants( + ReferenceGenome.GRCh38, + DatasetType.SNV_INDEL, + TEST_RUN_ID, + ) + # Tests individual components of the atomic_insert_entries # to validate the state after each step. cursor = connections['clickhouse_write'].cursor() @@ -892,7 +898,18 @@ def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): ], ) - def test_load_run_entries_snv_indel(self): + @patch.object( + ClickhouseReferenceDataset, + 'for_reference_genome_dataset_type', + return_value=[], + ) + def test_load_run_entries_snv_indel(self, mock_for_reference_genome_dataset_type): + load_run_variants( + ReferenceGenome.GRCh38, + DatasetType.SNV_INDEL, + TEST_RUN_ID, + ) + load_run_entries( ReferenceGenome.GRCh38, DatasetType.SNV_INDEL, @@ -1060,6 +1077,12 @@ def test_load_run_variants_gcnv(self): self.assertEqual(key_lookup_count, 4) def test_load_run_entries_gcnv(self): + load_run_variants( + ReferenceGenome.GRCh38, + DatasetType.GCNV, + TEST_RUN_ID, + ) + load_run_entries( ReferenceGenome.GRCh38, DatasetType.GCNV, From c2df84b2571abdd18c6b125943a8bf304e6e7402 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 12:11:07 -0400 Subject: [PATCH 28/53] test existing variants parquets with hail --- .../write_existing_variants_parquet_test.py | 83 +++++++++++-------- 1 file changed, 49 insertions(+), 34 deletions(-) diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 3225690f1d..9df9899e74 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -1,10 +1,10 @@ -import gzip from typing import ClassVar +import hail as hl import luigi.worker -import pandas as pd from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType +from loading_pipeline.lib.misc.io import import_parquet from loading_pipeline.lib.paths import existing_variants_parquet_path from loading_pipeline.lib.tasks.write_existing_variants_parquet import ( WriteExistingVariantsParquetTask, @@ -12,7 +12,6 @@ from loading_pipeline.lib.test.clickhouse_schema_testcase import ( ClickhouseSchemaTestCase, ) -from loading_pipeline.lib.test.misc import convert_ndarray_to_list from loading_pipeline.lib.test.mocked_dataroot_testcase import MockedDatarootTestCase TEST_RUN_ID = 'manual__2024-04-03' @@ -28,7 +27,7 @@ def _run_task( self, dataset_type: DatasetType, reference_genome: ReferenceGenome = ReferenceGenome.GRCh38, - ) -> pd.DataFrame: + ) -> hl.Table: worker = luigi.worker.Worker() task = WriteExistingVariantsParquetTask( reference_genome=reference_genome, @@ -41,51 +40,67 @@ def _run_task( worker.run() self.assertTrue(task.output().exists()) self.assertTrue(task.complete()) - with gzip.open( + return import_parquet( existing_variants_parquet_path(reference_genome, dataset_type, TEST_RUN_ID), - ) as f: - return pd.read_parquet(f) + reference_genome, + dataset_type, + ) def test_snv_indel(self): - df = self._run_task(DatasetType.SNV_INDEL) - self.assertEqual(list(df.columns), ['key_', 'variant_id']) - df = df.sort_values('key_').reset_index(drop=True) + ht = self._run_task(DatasetType.SNV_INDEL) + self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) self.assertEqual( - convert_ndarray_to_list( - df[['key_', 'variant_id']].to_dict('records'), - ), + ht.collect(), [ - {'key_': 1, 'variant_id': '1-878314-G-C'}, - {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, - {'key_': 10, 'variant_id': '10-987654-G-A'}, + hl.Struct( + key_=1, + locus=hl.Locus( + contig='chr1', + position=878314, + reference_genome='GRCh38', + ), + alleles=['G', 'C'], + ), + hl.Struct( + key_=7, + locus=hl.Locus( + contig='chr7', + position=1234567, + reference_genome='GRCh38', + ), + alleles=['AGT', 'A'], + ), + hl.Struct( + key_=10, + locus=hl.Locus( + contig='chr10', + position=987654, + reference_genome='GRCh38', + ), + alleles=['G', 'A'], + ), ], ) def test_grch37_snv_indel(self): - df = self._run_task( + ht = self._run_task( DatasetType.SNV_INDEL, reference_genome=ReferenceGenome.GRCh37, ) - self.assertEqual(list(df.columns), ['key_', 'variant_id']) - self.assertEqual(len(df), 0) + self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) + self.assertEqual(ht.count(), 0) def test_mito(self): - df = self._run_task(DatasetType.MITO) - self.assertEqual(list(df.columns), ['key_', 'variant_id']) - self.assertEqual(len(df), 0) + ht = self._run_task(DatasetType.MITO) + self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) + self.assertEqual(ht.count(), 0) def test_sv(self): - df = self._run_task(DatasetType.SV) - self.assertEqual( - list(df.columns), - ['key_', 'variant_id', 'end', 'endChrom'], - ) - self.assertEqual(len(df), 0) + ht = self._run_task(DatasetType.SV) + self.assertEqual(list(ht.row), ['key_', 'variant_id', 'end', 'endChrom']) + self.assertEqual(ht.count(), 0) def test_gcnv(self): - df = self._run_task(DatasetType.GCNV) - self.assertEqual( - list(df.columns), - ['key_', 'variant_id'], - ) - self.assertEqual(len(df), 0) + ht = self._run_task(DatasetType.GCNV) + self.assertEqual(list(ht.row), ['key_', 'variant_id']) + self.assertEqual(ht.count(), 0) From cb048cbf41cdccd1fb02c9354178a0bb5b2f8b25 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 12:14:10 -0400 Subject: [PATCH 29/53] ruff --- loading_pipeline/lib/tasks/exports/fields.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index aa8445d127..3c8bc1c8be 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -145,7 +145,9 @@ def _get_calls_export_fields( getattr(fe, f'sample_{field}'), get_value(ht), ) - for field, get_value in _get_entries_call_annotations_fields(dataset_type).items() + for field, get_value in _get_entries_call_annotations_fields( + dataset_type, + ).items() }, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, From 57a5c9da02cbd544c98e7c6ca938160b3a5dcdbe Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 13:49:46 -0400 Subject: [PATCH 30/53] fix test setup --- loading_pipeline/lib/misc/clickhouse_test.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 087abc10d3..010dada172 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -21,6 +21,7 @@ delete_existing_families_from_staging_entries, delete_family_guids, direct_insert_all_keys, + drop_staging_db, exchange_tables, get_clickhouse_client, insert_new_entries, @@ -469,6 +470,7 @@ def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): DatasetType.SNV_INDEL, TEST_RUN_ID, ) + drop_staging_db() # Tests individual components of the atomic_insert_entries # to validate the state after each step. From 8ad4d875a30b37f0ef2bab18f863f61f3c1f14ca Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 14:20:19 -0400 Subject: [PATCH 31/53] revert gzipping chnge --- loading_pipeline/lib/misc/clickhouse.py | 1 + loading_pipeline/lib/paths.py | 2 +- loading_pipeline/lib/paths_test.py | 2 +- 3 files changed, 3 insertions(+), 2 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index 6216da98a2..f1fcf0a3c9 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -993,6 +993,7 @@ def export_existing_variants_to_parquet( INSERT INTO FUNCTION {export_table} SELECT key AS key_, variantId AS variant_id {dt_fields} FROM {variants_table} + SETTINGS output_format_parquet_use_custom_encoder=1 """, # nosec B608 ) diff --git a/loading_pipeline/lib/paths.py b/loading_pipeline/lib/paths.py index 1409638623..df59e6cc39 100644 --- a/loading_pipeline/lib/paths.py +++ b/loading_pipeline/lib/paths.py @@ -320,7 +320,7 @@ def existing_variants_parquet_path( dataset_type, ), run_id, - 'existing_variants.parquet.gz', + 'existing_variants.parquet', ) diff --git a/loading_pipeline/lib/paths_test.py b/loading_pipeline/lib/paths_test.py index c36fdc2d4f..3803bdb92a 100644 --- a/loading_pipeline/lib/paths_test.py +++ b/loading_pipeline/lib/paths_test.py @@ -72,7 +72,7 @@ def test_existing_variants_parquet_path(self) -> None: DatasetType.GCNV, 'manual__2023-06-26T18:30:09.349671+00:00', ), - '/var/seqr/pipeline-data/GRCh38/GCNV/runs/manual__2023-06-26T18:30:09.349671+00:00/existing_variants.parquet.gz', + '/var/seqr/pipeline-data/GRCh38/GCNV/runs/manual__2023-06-26T18:30:09.349671+00:00/existing_variants.parquet', ) def test_remapped_and_subsetted_callset_path(self) -> None: From 08244e434665df80d99112ee1f796163411fe921 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 14:53:57 -0400 Subject: [PATCH 32/53] properly keep override annotations --- loading_pipeline/lib/core/dataset_type.py | 4 ---- loading_pipeline/lib/misc/family_entries.py | 2 ++ loading_pipeline/lib/tasks/exports/fields.py | 8 +++----- .../lib/tasks/exports/write_new_entries_parquet.py | 2 ++ 4 files changed, 7 insertions(+), 9 deletions(-) diff --git a/loading_pipeline/lib/core/dataset_type.py b/loading_pipeline/lib/core/dataset_type.py index 6d6430ba14..e9bccc99f6 100644 --- a/loading_pipeline/lib/core/dataset_type.py +++ b/loading_pipeline/lib/core/dataset_type.py @@ -324,10 +324,6 @@ def genotype_entry_annotation_fns(self) -> list[Callable[..., hl.Expression]]: gcnv.CN, gcnv.GT, gcnv.QS, - gcnv.start_locus, - gcnv.end_locus, - gcnv.num_exon, - gcnv.sorted_gene_consequences, ], }[self] diff --git a/loading_pipeline/lib/misc/family_entries.py b/loading_pipeline/lib/misc/family_entries.py index 8bc71af451..67f0339130 100644 --- a/loading_pipeline/lib/misc/family_entries.py +++ b/loading_pipeline/lib/misc/family_entries.py @@ -7,6 +7,7 @@ def compute_callset_family_entries_ht( dataset_type: DatasetType, mt: hl.MatrixTable, entries_fields: dict[str, hl.Expression], + additional_selects: dict | None = None, ) -> hl.Table: sample_id_to_family_guid = hl.dict( { @@ -47,6 +48,7 @@ def compute_callset_family_entries_ht( lambda fe: fe[0].family_guid, ) ), + **{field: get_value(mt) for field, get_value in (additional_selects or {}).items()}, ).rows() # NB: globalize before we set families to missing ht = globalize_ids(ht) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 3c8bc1c8be..87bd28bb37 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -91,7 +91,7 @@ def get_dataset_type_specific_variants_annotations( }[dataset_type](ht) -def _get_entries_call_annotations_fields( +def get_entries_call_annotations_fields( dataset_type: DatasetType, ): if dataset_type == DatasetType.GCNV: @@ -143,11 +143,9 @@ def _get_calls_export_fields( **{ snake_to_camelcase(field): hl.or_else( getattr(fe, f'sample_{field}'), - get_value(ht), + getattr(ht, field), ) - for field, get_value in _get_entries_call_annotations_fields( - dataset_type, - ).items() + for field in get_entries_call_annotations_fields(dataset_type) }, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py index 35bb6cad41..a4f043866e 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py @@ -16,6 +16,7 @@ ) from loading_pipeline.lib.tasks.base.base_write_parquet import BaseWriteParquetTask from loading_pipeline.lib.tasks.exports.fields import ( + get_entries_call_annotations_fields, get_entries_export_fields, ) from loading_pipeline.lib.tasks.files import GCSorLocalTarget @@ -54,6 +55,7 @@ def create_table(self) -> hl.Table: self.dataset_type.genotype_entry_annotation_fns, **self.param_kwargs, ), + get_entries_call_annotations_fields(self.dataset_type), ) ht = deglobalize_ids(ht) ht = deduplicate_by_most_non_ref_calls(ht) From 7188bc63c2345c1eedfa16a42556b6ad994c4a98 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:02:33 -0400 Subject: [PATCH 33/53] fix clickhouse test fixtures --- loading_pipeline/lib/misc/clickhouse_test.py | 19 ++------ .../lib/test/fixtures/clickhouse_test.json | 46 +++++++++++++++++++ 2 files changed, 49 insertions(+), 16 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 010dada172..ac053fed4e 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -328,9 +328,9 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): 'WES', ], 'variantId': [ - '1-3-A-C', - 'Y-19-A-C', - 'M-12-C-G', + '10-987654-G-A', + '3-133456789-A-G', + '4-133456789-C-T', ], 'calls': [ [('sample_d1', 0), ('sample_d11', 2)], @@ -465,13 +465,6 @@ def test_direct_insert_all_keys(self): return_value=[ClickhouseReferenceDataset.CLINVAR], ) def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): - load_run_variants( - ReferenceGenome.GRCh38, - DatasetType.SNV_INDEL, - TEST_RUN_ID, - ) - drop_staging_db() - # Tests individual components of the atomic_insert_entries # to validate the state after each step. cursor = connections['clickhouse_write'].cursor() @@ -906,12 +899,6 @@ def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): return_value=[], ) def test_load_run_entries_snv_indel(self, mock_for_reference_genome_dataset_type): - load_run_variants( - ReferenceGenome.GRCh38, - DatasetType.SNV_INDEL, - TEST_RUN_ID, - ) - load_run_entries( ReferenceGenome.GRCh38, DatasetType.SNV_INDEL, diff --git a/loading_pipeline/lib/test/fixtures/clickhouse_test.json b/loading_pipeline/lib/test/fixtures/clickhouse_test.json index d7726ed9dc..d205e72da2 100644 --- a/loading_pipeline/lib/test/fixtures/clickhouse_test.json +++ b/loading_pipeline/lib/test/fixtures/clickhouse_test.json @@ -326,5 +326,51 @@ "variant_id": "10-987654-G-A", "key": 10 } + }, + { + "model": "clickhouse_search.keylookupsnvindel", + "pk": "3-133456789-A-G", + "fields": { + "variant_id": "3-133456789-A-G", + "key": 3 + } + }, + { + "model": "clickhouse_search.keylookupsnvindel", + "pk": "4-133456789-C-T", + "fields": { + "variant_id": "4-133456789-C-T", + "key": 4 + } + }, + { + "model": "clickhouse_search.variantssnvindel", + "pk": 10, + "fields": { + "key": 10, + "sorted_transcript_consequences": [], + "sorted_motif_feature_consequences": [], + "sorted_regulatory_feature_consequences": [] + } + }, + { + "model": "clickhouse_search.variantssnvindel", + "pk": 3, + "fields": { + "key": 3, + "sorted_transcript_consequences": [], + "sorted_motif_feature_consequences": [], + "sorted_regulatory_feature_consequences": [] + } + }, + { + "model": "clickhouse_search.variantssnvindel", + "pk": 4, + "fields": { + "key": 4, + "sorted_transcript_consequences": [], + "sorted_motif_feature_consequences": [], + "sorted_regulatory_feature_consequences": [] + } } ] From 8c526c69944bab184e8c30a40d9e489b6e79ca14 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:03:28 -0400 Subject: [PATCH 34/53] unused import --- loading_pipeline/lib/misc/clickhouse_test.py | 1 - 1 file changed, 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index ac053fed4e..6d0ae5d5ec 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -21,7 +21,6 @@ delete_existing_families_from_staging_entries, delete_family_guids, direct_insert_all_keys, - drop_staging_db, exchange_tables, get_clickhouse_client, insert_new_entries, From 67cb75d547ece84f126e9cfe67f6088e2875da09 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:06:38 -0400 Subject: [PATCH 35/53] ruff --- loading_pipeline/lib/misc/clickhouse_test.py | 7 +------ loading_pipeline/lib/misc/family_entries.py | 5 ++++- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 6d0ae5d5ec..da1a0a5449 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -892,12 +892,7 @@ def test_entries_insert_flow(self, mock_for_reference_genome_dataset_type): ], ) - @patch.object( - ClickhouseReferenceDataset, - 'for_reference_genome_dataset_type', - return_value=[], - ) - def test_load_run_entries_snv_indel(self, mock_for_reference_genome_dataset_type): + def test_load_run_entries_snv_indel(self): load_run_entries( ReferenceGenome.GRCh38, DatasetType.SNV_INDEL, diff --git a/loading_pipeline/lib/misc/family_entries.py b/loading_pipeline/lib/misc/family_entries.py index 67f0339130..5b9cd529ae 100644 --- a/loading_pipeline/lib/misc/family_entries.py +++ b/loading_pipeline/lib/misc/family_entries.py @@ -48,7 +48,10 @@ def compute_callset_family_entries_ht( lambda fe: fe[0].family_guid, ) ), - **{field: get_value(mt) for field, get_value in (additional_selects or {}).items()}, + **{ + field: get_value(mt) + for field, get_value in (additional_selects or {}).items() + }, ).rows() # NB: globalize before we set families to missing ht = globalize_ids(ht) From 6e3e7202aa48824a19926d445e3afe9b0da2ae69 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:06:44 -0400 Subject: [PATCH 36/53] Revert "test existing variants parquets with hail" This reverts commit c2df84b2571abdd18c6b125943a8bf304e6e7402. --- .../write_existing_variants_parquet_test.py | 83 ++++++++----------- 1 file changed, 34 insertions(+), 49 deletions(-) diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 9df9899e74..3225690f1d 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -1,10 +1,10 @@ +import gzip from typing import ClassVar -import hail as hl import luigi.worker +import pandas as pd from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType -from loading_pipeline.lib.misc.io import import_parquet from loading_pipeline.lib.paths import existing_variants_parquet_path from loading_pipeline.lib.tasks.write_existing_variants_parquet import ( WriteExistingVariantsParquetTask, @@ -12,6 +12,7 @@ from loading_pipeline.lib.test.clickhouse_schema_testcase import ( ClickhouseSchemaTestCase, ) +from loading_pipeline.lib.test.misc import convert_ndarray_to_list from loading_pipeline.lib.test.mocked_dataroot_testcase import MockedDatarootTestCase TEST_RUN_ID = 'manual__2024-04-03' @@ -27,7 +28,7 @@ def _run_task( self, dataset_type: DatasetType, reference_genome: ReferenceGenome = ReferenceGenome.GRCh38, - ) -> hl.Table: + ) -> pd.DataFrame: worker = luigi.worker.Worker() task = WriteExistingVariantsParquetTask( reference_genome=reference_genome, @@ -40,67 +41,51 @@ def _run_task( worker.run() self.assertTrue(task.output().exists()) self.assertTrue(task.complete()) - return import_parquet( + with gzip.open( existing_variants_parquet_path(reference_genome, dataset_type, TEST_RUN_ID), - reference_genome, - dataset_type, - ) + ) as f: + return pd.read_parquet(f) def test_snv_indel(self): - ht = self._run_task(DatasetType.SNV_INDEL) - self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) + df = self._run_task(DatasetType.SNV_INDEL) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) + df = df.sort_values('key_').reset_index(drop=True) self.assertEqual( - ht.collect(), + convert_ndarray_to_list( + df[['key_', 'variant_id']].to_dict('records'), + ), [ - hl.Struct( - key_=1, - locus=hl.Locus( - contig='chr1', - position=878314, - reference_genome='GRCh38', - ), - alleles=['G', 'C'], - ), - hl.Struct( - key_=7, - locus=hl.Locus( - contig='chr7', - position=1234567, - reference_genome='GRCh38', - ), - alleles=['AGT', 'A'], - ), - hl.Struct( - key_=10, - locus=hl.Locus( - contig='chr10', - position=987654, - reference_genome='GRCh38', - ), - alleles=['G', 'A'], - ), + {'key_': 1, 'variant_id': '1-878314-G-C'}, + {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, + {'key_': 10, 'variant_id': '10-987654-G-A'}, ], ) def test_grch37_snv_indel(self): - ht = self._run_task( + df = self._run_task( DatasetType.SNV_INDEL, reference_genome=ReferenceGenome.GRCh37, ) - self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) - self.assertEqual(ht.count(), 0) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) + self.assertEqual(len(df), 0) def test_mito(self): - ht = self._run_task(DatasetType.MITO) - self.assertEqual(list(ht.row), ['key_', 'locus', 'alleles']) - self.assertEqual(ht.count(), 0) + df = self._run_task(DatasetType.MITO) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) + self.assertEqual(len(df), 0) def test_sv(self): - ht = self._run_task(DatasetType.SV) - self.assertEqual(list(ht.row), ['key_', 'variant_id', 'end', 'endChrom']) - self.assertEqual(ht.count(), 0) + df = self._run_task(DatasetType.SV) + self.assertEqual( + list(df.columns), + ['key_', 'variant_id', 'end', 'endChrom'], + ) + self.assertEqual(len(df), 0) def test_gcnv(self): - ht = self._run_task(DatasetType.GCNV) - self.assertEqual(list(ht.row), ['key_', 'variant_id']) - self.assertEqual(ht.count(), 0) + df = self._run_task(DatasetType.GCNV) + self.assertEqual( + list(df.columns), + ['key_', 'variant_id'], + ) + self.assertEqual(len(df), 0) From 014a449d9ace087c3eb564fbcbfd39b878616b92 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:08:23 -0400 Subject: [PATCH 37/53] Revert "fix existing parquet test" This reverts commit 9fcf4cc58b0936c53f702a13302ff6676f78ee8a. --- .../write_existing_variants_parquet_test.py | 26 ++++++++++--------- 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 3225690f1d..1323c0b775 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -1,4 +1,3 @@ -import gzip from typing import ClassVar import luigi.worker @@ -41,23 +40,26 @@ def _run_task( worker.run() self.assertTrue(task.output().exists()) self.assertTrue(task.complete()) - with gzip.open( + return pd.read_parquet( existing_variants_parquet_path(reference_genome, dataset_type, TEST_RUN_ID), - ) as f: - return pd.read_parquet(f) + ) def test_snv_indel(self): df = self._run_task(DatasetType.SNV_INDEL) - self.assertEqual(list(df.columns), ['key_', 'variant_id']) + self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) df = df.sort_values('key_').reset_index(drop=True) self.assertEqual( convert_ndarray_to_list( - df[['key_', 'variant_id']].to_dict('records'), + df[['key_', 'variant_id', 'geneIds']].to_dict('records'), ), [ - {'key_': 1, 'variant_id': '1-878314-G-C'}, - {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, - {'key_': 10, 'variant_id': '10-987654-G-A'}, + { + 'key_': 1, + 'variant_id': '1-878314-G-C', + 'geneIds': ['ENSG00000177000'], + }, + {'key_': 7, 'variant_id': '7-1234567-AGT-A', 'geneIds': []}, + {'key_': 10, 'variant_id': '10-987654-G-A', 'geneIds': []}, ], ) @@ -66,7 +68,7 @@ def test_grch37_snv_indel(self): DatasetType.SNV_INDEL, reference_genome=ReferenceGenome.GRCh37, ) - self.assertEqual(list(df.columns), ['key_', 'variant_id']) + self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) self.assertEqual(len(df), 0) def test_mito(self): @@ -78,7 +80,7 @@ def test_sv(self): df = self._run_task(DatasetType.SV) self.assertEqual( list(df.columns), - ['key_', 'variant_id', 'end', 'endChrom'], + ['key_', 'variant_id', 'xpos', 'end', 'endChrom', 'geneIds'], ) self.assertEqual(len(df), 0) @@ -86,6 +88,6 @@ def test_gcnv(self): df = self._run_task(DatasetType.GCNV) self.assertEqual( list(df.columns), - ['key_', 'variant_id'], + ['key_', 'variant_id', 'xpos', 'start', 'end', 'num_exon', 'gene_ids'], ) self.assertEqual(len(df), 0) From 38a22d676edc33d87a32decdd44182d2d124847a Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:09:48 -0400 Subject: [PATCH 38/53] actual update existing variants test --- .../write_existing_variants_parquet_test.py | 18 +++++++----------- 1 file changed, 7 insertions(+), 11 deletions(-) diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 1323c0b775..52e9a1bdc2 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -46,20 +46,16 @@ def _run_task( def test_snv_indel(self): df = self._run_task(DatasetType.SNV_INDEL) - self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) df = df.sort_values('key_').reset_index(drop=True) self.assertEqual( convert_ndarray_to_list( - df[['key_', 'variant_id', 'geneIds']].to_dict('records'), + df[['key_', 'variant_id']].to_dict('records'), ), [ - { - 'key_': 1, - 'variant_id': '1-878314-G-C', - 'geneIds': ['ENSG00000177000'], - }, - {'key_': 7, 'variant_id': '7-1234567-AGT-A', 'geneIds': []}, - {'key_': 10, 'variant_id': '10-987654-G-A', 'geneIds': []}, + {'key_': 1, 'variant_id': '1-878314-G-C'}, + {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, + {'key_': 10, 'variant_id': '10-987654-G-A'}, ], ) @@ -68,7 +64,7 @@ def test_grch37_snv_indel(self): DatasetType.SNV_INDEL, reference_genome=ReferenceGenome.GRCh37, ) - self.assertEqual(list(df.columns), ['key_', 'variant_id', 'geneIds']) + self.assertEqual(list(df.columns), ['key_', 'variant_id']) self.assertEqual(len(df), 0) def test_mito(self): @@ -80,7 +76,7 @@ def test_sv(self): df = self._run_task(DatasetType.SV) self.assertEqual( list(df.columns), - ['key_', 'variant_id', 'xpos', 'end', 'endChrom', 'geneIds'], + ['key_', 'variant_id', 'end', 'endChrom'], ) self.assertEqual(len(df), 0) From 98ff3444e4019549e106f66a4c542aa83a095d24 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:36:38 -0400 Subject: [PATCH 39/53] fix gcnv annotations --- loading_pipeline/lib/tasks/exports/fields.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 87bd28bb37..0e10fbd55a 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -96,10 +96,10 @@ def get_entries_call_annotations_fields( ): if dataset_type == DatasetType.GCNV: return { - 'start': lambda ht: hl.int64(ht.start_locus.position), - 'end': lambda ht: hl.int64(ht.end_locus.position), + 'start': lambda ht: hl.int64(ht.start), + 'end': lambda ht: hl.int64(ht.end), 'num_exon': lambda ht: ht.num_exon, - 'gene_ids': lambda ht: hl.set(ht.sorted_gene_consequences.gene_id), + 'gene_ids': lambda ht: hl.set(ht.gene_ids), } return {} From ef2e7e60bfa3a8d0978ae18319c2f1a7b157c929 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:39:47 -0400 Subject: [PATCH 40/53] fixture test up[dates --- loading_pipeline/lib/misc/clickhouse_test.py | 4 ++++ .../lib/tasks/write_existing_variants_parquet_test.py | 4 +++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index da1a0a5449..dbfdd1fe13 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -981,6 +981,8 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_memory, [ + (3, [], [], []), + (4, [], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), @@ -998,6 +1000,8 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_disk, [ + (3, [], [], []), + (4, [], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), diff --git a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py index 52e9a1bdc2..d3d8af7dd8 100644 --- a/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py +++ b/loading_pipeline/lib/tasks/write_existing_variants_parquet_test.py @@ -54,6 +54,8 @@ def test_snv_indel(self): ), [ {'key_': 1, 'variant_id': '1-878314-G-C'}, + {'key_': 3, 'variant_id': '3-133456789-A-G'}, + {'key_': 4, 'variant_id': '4-133456789-C-T'}, {'key_': 7, 'variant_id': '7-1234567-AGT-A'}, {'key_': 10, 'variant_id': '10-987654-G-A'}, ], @@ -84,6 +86,6 @@ def test_gcnv(self): df = self._run_task(DatasetType.GCNV) self.assertEqual( list(df.columns), - ['key_', 'variant_id', 'xpos', 'start', 'end', 'num_exon', 'gene_ids'], + ['key_', 'variant_id'], ) self.assertEqual(len(df), 0) From 6ff864fa9cec4bb3064d5a59375647814bd1fc8c Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:44:41 -0400 Subject: [PATCH 41/53] update gen id test fixtures --- loading_pipeline/lib/misc/clickhouse_test.py | 6 +++++- loading_pipeline/lib/test/fixtures/clickhouse_test.json | 9 +++++++-- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index dbfdd1fe13..decb169e8c 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -203,7 +203,11 @@ def setUpClass(cls): with connections['clickhouse_write'].cursor() as cursor: cursor.execute( f'INSERT INTO {Env.CLICKHOUSE_DATABASE}.`seqrdb_gene_ids_src` VALUES', - [('GENE1', 123), ('GENE2', 12), ('GENE3', 1)], + [ + ('ENSG00000141510', 123), + ('ENSG00000012048', 12), + ('ENSG00000139618', 1), + ], ) cursor.execute( f'SYSTEM RELOAD DICTIONARY {Env.CLICKHOUSE_DATABASE}.`seqrdb_gene_ids`', diff --git a/loading_pipeline/lib/test/fixtures/clickhouse_test.json b/loading_pipeline/lib/test/fixtures/clickhouse_test.json index d205e72da2..c3d10c054e 100644 --- a/loading_pipeline/lib/test/fixtures/clickhouse_test.json +++ b/loading_pipeline/lib/test/fixtures/clickhouse_test.json @@ -358,7 +358,10 @@ "pk": 3, "fields": { "key": 3, - "sorted_transcript_consequences": [], + "sorted_transcript_consequences": [ + [null, null, [], null, null, "ENSG00000141510", null], + [null, null, [], null, null, "ENSG00000012048", null] + ], "sorted_motif_feature_consequences": [], "sorted_regulatory_feature_consequences": [] } @@ -368,7 +371,9 @@ "pk": 4, "fields": { "key": 4, - "sorted_transcript_consequences": [], + "sorted_transcript_consequences": [ + [null, null, [], null, null, "ENSG00000139618", null] + ], "sorted_motif_feature_consequences": [], "sorted_regulatory_feature_consequences": [] } From 7bd4fa4be9202336c337f7c75372ca654cc06d51 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 16:54:01 -0400 Subject: [PATCH 42/53] better gcnv test --- loading_pipeline/lib/misc/clickhouse_test.py | 104 ++++++++++++++++++- 1 file changed, 101 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index decb169e8c..3908b33903 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -298,8 +298,19 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): TEST_RUN_ID, ), ) + gcnv_variants_df = pd.DataFrame( + { + 'key': [10, 11, 12, 13], + 'variantId': [ + 'suffix_1000_DEL', + 'suffix_1001_DUP', + 'suffix_1002_DEL', + 'suffix_1003_DUP', + ], + }, + ) write_test_parquet( - df, + gcnv_variants_df, new_variants_parquet_path( ReferenceGenome.GRCh38, DatasetType.GCNV, @@ -372,14 +383,101 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): ), schema, ) + gcnv_entries_df = pd.DataFrame( + { + 'project_guid': [ + 'project_d', + 'project_d', + 'project_d', + ], + 'family_guid': [ + 'family_d1', + 'family_d2', + 'family_d3', + ], + 'variantId': [ + 'suffix_1000_DEL', + 'suffix_1002_DEL', + 'suffix_1003_DUP', + ], + 'calls': [ + [ + { + 'sampleId': 'sample_d1', + 'gt': 0, + 'cn': 2, + 'qs': 4, + 'defragged': False, + 'start': 100006937, + 'end': 100007881, + 'numExon': 2, + 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], + 'newCall': False, + 'prevCall': True, + 'prevOverlap': False, + }, + { + 'sampleId': 'sample_d11', + 'gt': 2, + 'cn': 0, + 'qs': 30, + 'defragged': False, + 'start': 100006937, + 'end': 100007881, + 'numExon': 2, + 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], + 'newCall': True, + 'prevCall': False, + 'prevOverlap': False, + }, + ], + [ + { + 'sampleId': 'sample_d2', + 'gt': 0, + 'cn': 2, + 'qs': 5, + 'defragged': False, + 'start': 100017585, + 'end': 100023213, + 'numExon': 1, + 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], + 'newCall': False, + 'prevCall': True, + 'prevOverlap': False, + }, + ], + [ + { + 'sampleId': 'sample_d3', + 'gt': 1, + 'cn': 1, + 'qs': 20, + 'defragged': False, + 'start': 100017585, + 'end': 100023213, + 'numExon': 1, + 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], + 'newCall': True, + 'prevCall': False, + 'prevOverlap': False, + }, + ], + ], + 'sign': [ + 1, + 1, + 1, + ], + }, + ) write_test_parquet( - df.drop(['xpos', 'sample_type'], axis=1), + gcnv_entries_df, new_entries_parquet_path( ReferenceGenome.GRCh38, DatasetType.GCNV, TEST_RUN_ID, ), - schema.remove(2).remove(2), ) def test_get_clickhouse_client(self): From eabfa8aacd63ab9dd903b4609799d41cfd65ca49 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 17:51:53 -0400 Subject: [PATCH 43/53] update test from fixture --- loading_pipeline/lib/misc/clickhouse_test.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 3908b33903..f3c743ad3e 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -1083,8 +1083,8 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_memory, [ - (3, [], [], []), - (4, [], [], []), + (3, [(None, None, [], None, None, 'ENSG00000141510', None), (None, None, [], None, None, 'ENSG00000012048', None)], [], []), + (4, [(None, None, [], None, None, 'ENSG00000139618', None)], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), @@ -1102,8 +1102,8 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_disk, [ - (3, [], [], []), - (4, [], [], []), + (3, [(None, None, [], None, None, 'ENSG00000141510', None), (None, None, [], None, None, 'ENSG00000012048', None)], [], []), + (4, [(None, None, [], None, None, 'ENSG00000139618', None)], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), From 46a5cea8fe6c59b675e5629889ee84c151944d0d Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 18:09:55 -0400 Subject: [PATCH 44/53] directly define gcnv entry overrides --- loading_pipeline/lib/annotations/gcnv.py | 15 ++++++++--- loading_pipeline/lib/misc/family_entries.py | 5 ---- loading_pipeline/lib/tasks/exports/fields.py | 27 ++++--------------- .../exports/write_new_entries_parquet.py | 2 -- 4 files changed, 16 insertions(+), 33 deletions(-) diff --git a/loading_pipeline/lib/annotations/gcnv.py b/loading_pipeline/lib/annotations/gcnv.py index d9e2a8a603..1e81f0375e 100644 --- a/loading_pipeline/lib/annotations/gcnv.py +++ b/loading_pipeline/lib/annotations/gcnv.py @@ -104,28 +104,35 @@ def rg37_locus_end( def sample_end(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.or_missing( + return hl.sample_end( ~_start_and_end_equal(mt), mt.sample_end, + mt.end, ) def sample_gene_ids(mt: hl.MatrixTable, **_: Any) -> hl.Expression: parsed_genes = parse_gcnv_genes(mt.genes_any_overlap_Ensemble_ID) - return hl.or_missing(parsed_genes != mt.gene_ids, parsed_genes) + return hl.if_else( + parsed_genes != mt.gene_ids, + parsed_genes, + mt.gene_ids, + ) def sample_start(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.or_missing( + return hl.if_else( ~_start_and_end_equal(mt), mt.sample_start, + mt.start, ) def sample_num_exon(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.or_missing( + return hl.if_else( mt.genes_any_overlap_totalExons != mt.num_exon, mt.genes_any_overlap_totalExons, + mt.num_exon, ) diff --git a/loading_pipeline/lib/misc/family_entries.py b/loading_pipeline/lib/misc/family_entries.py index 5b9cd529ae..8bc71af451 100644 --- a/loading_pipeline/lib/misc/family_entries.py +++ b/loading_pipeline/lib/misc/family_entries.py @@ -7,7 +7,6 @@ def compute_callset_family_entries_ht( dataset_type: DatasetType, mt: hl.MatrixTable, entries_fields: dict[str, hl.Expression], - additional_selects: dict | None = None, ) -> hl.Table: sample_id_to_family_guid = hl.dict( { @@ -48,10 +47,6 @@ def compute_callset_family_entries_ht( lambda fe: fe[0].family_guid, ) ), - **{ - field: get_value(mt) - for field, get_value in (additional_selects or {}).items() - }, ).rows() # NB: globalize before we set families to missing ht = globalize_ids(ht) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 0e10fbd55a..4d162e032f 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -91,21 +91,7 @@ def get_dataset_type_specific_variants_annotations( }[dataset_type](ht) -def get_entries_call_annotations_fields( - dataset_type: DatasetType, -): - if dataset_type == DatasetType.GCNV: - return { - 'start': lambda ht: hl.int64(ht.start), - 'end': lambda ht: hl.int64(ht.end), - 'num_exon': lambda ht: ht.num_exon, - 'gene_ids': lambda ht: hl.set(ht.gene_ids), - } - return {} - - def _get_calls_export_fields( - ht: hl.Table, fe: hl.Struct, dataset_type: DatasetType, ): @@ -140,13 +126,10 @@ def _get_calls_export_fields( cn=fe.CN, qs=fe.QS, defragged=fe.defragged, - **{ - snake_to_camelcase(field): hl.or_else( - getattr(fe, f'sample_{field}'), - getattr(ht, field), - ) - for field in get_entries_call_annotations_fields(dataset_type) - }, + start=fe.sample_start, + end=fe.sample_end, + numExon=fe.sample_num_exon, + gene_ids=fe.sample_gene_ids, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, prevOverlap=fe.concordance.prev_overlap, @@ -173,7 +156,7 @@ def get_entries_export_fields( ), 'filters': ht.filters, 'calls': hl.sorted(ht.family_entries, key=lambda fe: fe.s).map( - lambda fe: _get_calls_export_fields(ht, fe, dataset_type), + lambda fe: _get_calls_export_fields(fe, dataset_type), ), 'sign': 1, } diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py index a4f043866e..35bb6cad41 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet.py @@ -16,7 +16,6 @@ ) from loading_pipeline.lib.tasks.base.base_write_parquet import BaseWriteParquetTask from loading_pipeline.lib.tasks.exports.fields import ( - get_entries_call_annotations_fields, get_entries_export_fields, ) from loading_pipeline.lib.tasks.files import GCSorLocalTarget @@ -55,7 +54,6 @@ def create_table(self) -> hl.Table: self.dataset_type.genotype_entry_annotation_fns, **self.param_kwargs, ), - get_entries_call_annotations_fields(self.dataset_type), ) ht = deglobalize_ids(ht) ht = deduplicate_by_most_non_ref_calls(ht) From 5f9aebda2ef9af70a61efd5f25ef6c39c5baf583 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 18:14:56 -0400 Subject: [PATCH 45/53] clean up --- loading_pipeline/lib/annotations/gcnv.py | 2 +- loading_pipeline/lib/misc/clickhouse_test.py | 6 ++---- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/loading_pipeline/lib/annotations/gcnv.py b/loading_pipeline/lib/annotations/gcnv.py index 1e81f0375e..01f24f0c2e 100644 --- a/loading_pipeline/lib/annotations/gcnv.py +++ b/loading_pipeline/lib/annotations/gcnv.py @@ -104,7 +104,7 @@ def rg37_locus_end( def sample_end(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.sample_end( + return hl.if_else( ~_start_and_end_equal(mt), mt.sample_end, mt.end, diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index f3c743ad3e..7cb0855aa3 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -1077,14 +1077,13 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ SELECT * FROM {Env.CLICKHOUSE_DATABASE}.`GRCh38/SNV_INDEL/variants_memory` + WHERE key > 5 """, # nosec B608 ) variants_memory = cursor.fetchall() self.assertCountEqual( variants_memory, [ - (3, [(None, None, [], None, None, 'ENSG00000141510', None), (None, None, [], None, None, 'ENSG00000012048', None)], [], []), - (4, [(None, None, [], None, None, 'ENSG00000139618', None)], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), @@ -1096,14 +1095,13 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ SELECT * FROM {Env.CLICKHOUSE_DATABASE}.`GRCh38/SNV_INDEL/variants_disk` + WHERE key > 5 """, # nosec B608 ) variants_disk = cursor.fetchall() self.assertCountEqual( variants_disk, [ - (3, [(None, None, [], None, None, 'ENSG00000141510', None), (None, None, [], None, None, 'ENSG00000012048', None)], [], []), - (4, [(None, None, [], None, None, 'ENSG00000139618', None)], [], []), (10, [], [], []), (11, [], [], []), (12, [], [], []), From a1d15b08bed7a70812045df44765ef003fbe3d9d Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 18:19:46 -0400 Subject: [PATCH 46/53] unused import --- loading_pipeline/lib/tasks/exports/fields.py | 1 - 1 file changed, 1 deletion(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 4d162e032f..96e7a3a3b1 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -4,7 +4,6 @@ from loading_pipeline.lib.core import DatasetType, ReferenceGenome, SampleType from loading_pipeline.lib.tasks.exports.misc import ( reformat_transcripts_for_export, - snake_to_camelcase, ) STANDARD_CONTIGS = hl.set( From 7739f6666c4d82d672a1d6c88304e695b0f75409 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Tue, 15 Sep 2026 18:37:44 -0400 Subject: [PATCH 47/53] debug tests --- loading_pipeline/lib/misc/clickhouse_test.py | 6 +++--- .../lib/tasks/exports/write_new_entries_parquet_test.py | 1 + 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 7cb0855aa3..7a6405b115 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -281,7 +281,7 @@ def write_test_parquet(df: pd.DataFrame, parquet_path: str, schema=None): # New Variants parquet. df = pd.DataFrame( { - 'key': [10, 11, 12, 13], + 'key': [20, 11, 12, 13], 'variantId': [ '1-3-A-C', '2-4-A-T', @@ -1084,10 +1084,10 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_memory, [ - (10, [], [], []), (11, [], [], []), (12, [], [], []), (13, [], [], []), + (20, [], [], []), ], ) cursor.execute( @@ -1102,10 +1102,10 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_disk, [ - (10, [], [], []), (11, [], [], []), (12, [], [], []), (13, [], [], []), + (20, [], [], []), ], ) cursor.execute( diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index 13e852a627..7efb21b018 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -353,6 +353,7 @@ def test_gcnv_write_new_entries_parquet(self): TEST_RUN_ID, ), ) + self.maxDiff = None export_json = convert_ndarray_to_list(df.to_dict('records')) self.assertEqual( export_json, From 7476efa441022860b34a83125481981b5ba70283 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 11:11:49 -0400 Subject: [PATCH 48/53] fix new entries test --- loading_pipeline/lib/tasks/exports/fields.py | 2 +- .../exports/write_new_entries_parquet_test.py | 142 +++++++++--------- 2 files changed, 74 insertions(+), 70 deletions(-) diff --git a/loading_pipeline/lib/tasks/exports/fields.py b/loading_pipeline/lib/tasks/exports/fields.py index 96e7a3a3b1..7e27ae9d70 100644 --- a/loading_pipeline/lib/tasks/exports/fields.py +++ b/loading_pipeline/lib/tasks/exports/fields.py @@ -128,7 +128,7 @@ def _get_calls_export_fields( start=fe.sample_start, end=fe.sample_end, numExon=fe.sample_num_exon, - gene_ids=fe.sample_gene_ids, + geneIds=fe.sample_gene_ids, newCall=fe.concordance.new_call, prevCall=fe.concordance.prev_call, prevOverlap=fe.concordance.prev_overlap, diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index 7efb21b018..ceacecda7a 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -208,26 +208,28 @@ def test_mito_write_new_entries_parquet(self): export_json = convert_ndarray_to_list(df.to_dict('records')) self.assertEqual(len(export_json), 3) self.assertEqual( - export_json[0], - { - 'project_guid': 'R0116_test_project3', - 'family_guid': 'family_1', - 'sample_type': 'WGS', - 'variantId': 'M-8-G-T', - 'xpos': 25000000008, - 'filters': [], - 'calls': [ - { - 'sampleId': 'RGP_1270_2', - 'gt': 2, - 'dp': 4216, - 'hl': 0.999, - 'mitoCn': 224, - 'contamination': 0.0, - }, - ], - 'sign': 1, - }, + export_json[:1], + [ + { + 'project_guid': 'R0116_test_project3', + 'family_guid': 'family_1', + 'sample_type': 'WGS', + 'variantId': 'M-8-G-T', + 'xpos': 25000000008, + 'filters': [], + 'calls': [ + { + 'sampleId': 'RGP_1270_2', + 'gt': 2, + 'dp': 4216, + 'hl': 0.999, + 'mitoCn': 224, + 'contamination': 0.0, + }, + ], + 'sign': 1, + }, + ] ) def test_sv_write_new_entries_parquet(self): @@ -276,52 +278,54 @@ def test_sv_write_new_entries_parquet(self): export_json = convert_ndarray_to_list(df.to_dict('records')) self.assertEqual(len(export_json), 2) self.assertEqual( - export_json[0], - { - 'project_guid': 'R0115_test_project2', - 'family_guid': 'family_2_1', - 'variantId': 'BND_chr1_6', - 'filters': ['HIGH_SR_BACKGROUND', 'UNRESOLVED'], - 'calls': [ - { - 'sampleId': 'RGP_164_1', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_2', - 'gt': 1, - 'cn': None, - 'gq': 31, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_3', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - { - 'sampleId': 'RGP_164_4', - 'gt': 0, - 'cn': None, - 'gq': 99, - 'newCall': True, - 'prevCall': False, - 'prevNumAlt': None, - }, - ], - 'sign': 1, - }, + export_json[:1], + [ + { + 'variantId': 'BND_chr1_6', + 'project_guid': 'R0115_test_project2', + 'family_guid': 'family_2_1', + 'filters': ['HIGH_SR_BACKGROUND', 'UNRESOLVED'], + 'calls': [ + { + 'sampleId': 'RGP_164_1', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_2', + 'gt': 1, + 'cn': None, + 'gq': 31, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_3', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + { + 'sampleId': 'RGP_164_4', + 'gt': 0, + 'cn': None, + 'gq': 99, + 'newCall': True, + 'prevCall': False, + 'prevNumAlt': None, + }, + ], + 'sign': 1, + }, + ], ) def test_gcnv_write_new_entries_parquet(self): @@ -353,15 +357,15 @@ def test_gcnv_write_new_entries_parquet(self): TEST_RUN_ID, ), ) - self.maxDiff = None export_json = convert_ndarray_to_list(df.to_dict('records')) + self.assertEqual(len(export_json), 2) self.assertEqual( - export_json, + export_json[:1], [ { + 'variantId': 'suffix_16456_DEL', 'project_guid': 'R0115_test_project2', 'family_guid': 'family_2_1', - 'variantId': 'suffix_16456_DEL', 'filters': [], 'calls': [ { From 3c27bde7886867ff67abddbfe232e398fda2252b Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 11:13:22 -0400 Subject: [PATCH 49/53] fix test --- loading_pipeline/lib/misc/clickhouse_test.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 7a6405b115..874ebd4c81 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -1084,6 +1084,7 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_memory, [ + (10, [], [], []), (11, [], [], []), (12, [], [], []), (13, [], [], []), @@ -1102,6 +1103,7 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_disk, [ + (10, [], [], []), (11, [], [], []), (12, [], [], []), (13, [], [], []), From 61aa3d7e8be57858dd29be00c68d57bbdf18ba2e Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 11:16:34 -0400 Subject: [PATCH 50/53] ruff --- .../lib/tasks/exports/write_new_entries_parquet_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index ceacecda7a..c005add9b5 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -229,7 +229,7 @@ def test_mito_write_new_entries_parquet(self): ], 'sign': 1, }, - ] + ], ) def test_sv_write_new_entries_parquet(self): From ca9fe13894e1f2f6a756851295dbcf74bd24778c Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 11:37:24 -0400 Subject: [PATCH 51/53] fix test --- loading_pipeline/lib/misc/clickhouse_test.py | 3 +-- .../lib/tasks/exports/write_new_entries_parquet_test.py | 1 + 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/loading_pipeline/lib/misc/clickhouse_test.py b/loading_pipeline/lib/misc/clickhouse_test.py index 874ebd4c81..bd2cb9a2ee 100644 --- a/loading_pipeline/lib/misc/clickhouse_test.py +++ b/loading_pipeline/lib/misc/clickhouse_test.py @@ -1103,11 +1103,10 @@ def test_load_run_variants_snv_indel(self, mock_for_reference_genome_dataset_typ self.assertCountEqual( variants_disk, [ - (10, [], [], []), + (20, [], [], []), (11, [], [], []), (12, [], [], []), (13, [], [], []), - (20, [], [], []), ], ) cursor.execute( diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index c005add9b5..bd45fe0cd5 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -359,6 +359,7 @@ def test_gcnv_write_new_entries_parquet(self): ) export_json = convert_ndarray_to_list(df.to_dict('records')) self.assertEqual(len(export_json), 2) + self.maxDiff = None self.assertEqual( export_json[:1], [ From 90ea0ba89d1324bd872f73bbd05db53b5ada40a5 Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 12:39:26 -0400 Subject: [PATCH 52/53] clean up gcnv sample overrides --- loading_pipeline/lib/annotations/gcnv.py | 29 +++---------------- .../exports/write_new_entries_parquet_test.py | 5 ++-- 2 files changed, 6 insertions(+), 28 deletions(-) diff --git a/loading_pipeline/lib/annotations/gcnv.py b/loading_pipeline/lib/annotations/gcnv.py index 01f24f0c2e..76696fd54c 100644 --- a/loading_pipeline/lib/annotations/gcnv.py +++ b/loading_pipeline/lib/annotations/gcnv.py @@ -11,10 +11,6 @@ from loading_pipeline.lib.misc.gcnv import parse_gcnv_genes -def _start_and_end_equal(mt: hl.MatrixTable) -> hl.BooleanExpression: - return (mt.sample_start == mt.start) & (mt.sample_end == mt.end) - - def CN(mt: hl.MatrixTable, **_: Any) -> hl.Expression: # noqa: N802 return mt.CN @@ -104,36 +100,19 @@ def rg37_locus_end( def sample_end(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.if_else( - ~_start_and_end_equal(mt), - mt.sample_end, - mt.end, - ) + return mt.sample_end def sample_gene_ids(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - parsed_genes = parse_gcnv_genes(mt.genes_any_overlap_Ensemble_ID) - return hl.if_else( - parsed_genes != mt.gene_ids, - parsed_genes, - mt.gene_ids, - ) + return parse_gcnv_genes(mt.genes_any_overlap_Ensemble_ID) def sample_start(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.if_else( - ~_start_and_end_equal(mt), - mt.sample_start, - mt.start, - ) + return mt.sample_start def sample_num_exon(mt: hl.MatrixTable, **_: Any) -> hl.Expression: - return hl.if_else( - mt.genes_any_overlap_totalExons != mt.num_exon, - mt.genes_any_overlap_totalExons, - mt.num_exon, - ) + return mt.genes_any_overlap_totalExons def sorted_gene_consequences( diff --git a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py index bd45fe0cd5..b8908c1f8f 100644 --- a/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py +++ b/loading_pipeline/lib/tasks/exports/write_new_entries_parquet_test.py @@ -359,7 +359,6 @@ def test_gcnv_write_new_entries_parquet(self): ) export_json = convert_ndarray_to_list(df.to_dict('records')) self.assertEqual(len(export_json), 2) - self.maxDiff = None self.assertEqual( export_json[:1], [ @@ -391,7 +390,7 @@ def test_gcnv_write_new_entries_parquet(self): 'defragged': False, 'start': 100017585, 'end': 100023213, - 'numExon': 1, + 'numExon': 3, 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], 'newCall': False, 'prevCall': False, @@ -405,7 +404,7 @@ def test_gcnv_write_new_entries_parquet(self): 'defragged': False, 'start': 100017585, 'end': 100023213, - 'numExon': 1, + 'numExon': 3, 'geneIds': ['ENSG00000117620', 'ENSG00000283761'], 'newCall': False, 'prevCall': True, From 753e1dd68e42529b1eab0ede57b0f80c292e456d Mon Sep 17 00:00:00 2001 From: Hana Snow Date: Wed, 16 Sep 2026 13:28:27 -0400 Subject: [PATCH 53/53] unambiguous key source --- loading_pipeline/lib/misc/clickhouse.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/loading_pipeline/lib/misc/clickhouse.py b/loading_pipeline/lib/misc/clickhouse.py index f1fcf0a3c9..f35df5d3f5 100644 --- a/loading_pipeline/lib/misc/clickhouse.py +++ b/loading_pipeline/lib/misc/clickhouse.py @@ -687,7 +687,7 @@ def insert_new_entries( ): common.append('is_gnomad_gt_5_percent') overrides['is_gnomad_gt_5_percent'] = f""" - dictGetOrDefault({ClickhouseReferenceDataset.GNOMAD_GENOMES.search_path(table_name_builder)}, 'filter_af', key, 0) > 0.05 + dictGetOrDefault({ClickhouseReferenceDataset.GNOMAD_GENOMES.search_path(table_name_builder)}, 'filter_af', e.key, 0) > 0.05 """ dst_list = ', '.join(common)