[optimization] read family tables directly from project table. (#769)

* Delete project tasks

* cleanup

* ruff format

* well

* rename

* hacking away

* almost there!

* ruff

* Fix missing updates change

* ruff

* Remove debug code

* remove bad merge

* more precision in test

* project table

* allow for missing project

* remove some unnecessary checks

* test already deleted family

* Lots of renames

* More updates

* Sketch

* Flesh out test

* fix paths

* Rename base hail table

* a bunch more renames

* delete project table

* Add delete project families

* is it that simple?

* add comment

* test it!

* Fix

* add dep

Author: bpblanken

v03_pipeline/lib/tasks/write_family_table.py

@@ -1,16 +1,11 @@
 import hail as hl
 import luigi
 
-from v03_pipeline.lib.annotations.fields import get_fields
-from v03_pipeline.lib.misc.family_entries import compute_callset_family_entries_ht
-from v03_pipeline.lib.misc.io import import_pedigree
-from v03_pipeline.lib.misc.pedigree import parse_pedigree_ht_to_families
-from v03_pipeline.lib.misc.sample_ids import subset_samples
 from v03_pipeline.lib.paths import family_table_path
 from v03_pipeline.lib.tasks.base.base_write import BaseWriteTask
 from v03_pipeline.lib.tasks.files import GCSorLocalTarget
-from v03_pipeline.lib.tasks.write_remapped_and_subsetted_callset import (
-    WriteRemappedAndSubsettedCallsetTask,
+from v03_pipeline.lib.tasks.update_project_table import (
+    UpdateProjectTableTask,
 )
 
 
@@ -56,50 +51,28 @@ def complete(self) -> bool:
         )
 
     def requires(self) -> luigi.Task:
-        return WriteRemappedAndSubsettedCallsetTask(
+        return UpdateProjectTableTask(
             self.reference_genome,
             self.dataset_type,
             self.sample_type,
-            self.callset_path,
             self.project_guid,
+            self.callset_path,
             self.project_remap_path,
             self.project_pedigree_path,
             self.ignore_missing_samples_when_subsetting,
             self.ignore_missing_samples_when_remapping,
             self.validate,
             False,
+            self.is_new_gcnv_joint_call,
         )
 
     def create_table(self) -> hl.Table:
-        callset_mt = hl.read_matrix_table(self.input().path)
-        pedigree_ht = import_pedigree(self.project_pedigree_path)
-        families = parse_pedigree_ht_to_families(pedigree_ht)
-        family = next(
-            iter(
-                family for family in families if family.family_guid == self.family_guid
-            ),
-        )
-        callset_mt = subset_samples(
-            callset_mt,
-            hl.Table.parallelize(
-                [{'s': sample_id} for sample_id in family.samples],
-                hl.tstruct(s=hl.dtype('str')),
-                key='s',
-            ),
-            False,
-        )
-        ht = compute_callset_family_entries_ht(
-            self.dataset_type,
-            callset_mt,
-            get_fields(
-                callset_mt,
-                self.dataset_type.genotype_entry_annotation_fns,
-                **self.param_kwargs,
-            ),
-        )
-        ht = ht.transmute(
-            entries=hl.flatten(ht.family_entries),
+        project_ht = hl.read_table(self.input().path)
+        family_i = project_ht.globals.family_guids.index(self.family_guid)
+        ht = project_ht.transmute(
+            entries=project_ht.family_entries[family_i],
         )
+        ht = ht.filter(hl.is_defined(ht.entries))
         return ht.select_globals(
             sample_ids=ht.family_samples[self.family_guid],
             sample_type=self.sample_type.value,

v03_pipeline/lib/tasks/write_project_family_tables.py

@@ -2,10 +2,8 @@
 import luigi
 
 from v03_pipeline.lib.tasks.base.base_hail_table import BaseHailTableTask
+from v03_pipeline.lib.tasks.update_project_table import UpdateProjectTableTask
 from v03_pipeline.lib.tasks.write_family_table import WriteFamilyTableTask
-from v03_pipeline.lib.tasks.write_remapped_and_subsetted_callset import (
-    WriteRemappedAndSubsettedCallsetTask,
-)
 
 
 class WriteProjectFamilyTablesTask(BaseHailTableTask):
@@ -50,22 +48,23 @@ def complete(self) -> bool:
 
     def run(self):
         # https://luigi.readthedocs.io/en/stable/tasks.html#dynamic-dependencies
-        rmsct_output: luigi.Target = yield WriteRemappedAndSubsettedCallsetTask(
+        update_project_table_task: luigi.Target = yield UpdateProjectTableTask(
             self.reference_genome,
             self.dataset_type,
             self.sample_type,
-            self.callset_path,
             self.project_guid,
+            self.callset_path,
             self.project_remap_path,
             self.project_pedigree_path,
             self.ignore_missing_samples_when_subsetting,
             self.ignore_missing_samples_when_remapping,
             self.validate,
             False,
+            self.is_new_gcnv_joint_call,
         )
-        callset_mt = hl.read_matrix_table(rmsct_output.path)
-        family_samples = hl.eval(callset_mt.globals.family_samples)
-        for family_guid in family_samples:
+        project_ht = hl.read_table(update_project_table_task.path)
+        family_guids = hl.eval(project_ht.globals.family_guids)
+        for family_guid in family_guids:
             self.dynamic_write_family_table_tasks.add(
                 WriteFamilyTableTask(
                     **self.param_kwargs,