add complete() logic to UpdatedReferenceDatasetCollectionTask, including check globals against config (#669)

* first pass at update task

* move stuff back to /lib file

* add tests for former combine file

* add tests for former combine file

* comment w/ logic for complete methjod

* task unit test

* add missing test cases and default luigi datasets param

* delete_reference_data_ht removed

* pseudocode

* implement match check functions

* handle dataset not in joined globals?

* oops

* tests for all fns except version

* add version tests

* test validate function some refactor

* add logging and tests for task

* refactor without custom selects

* eval in validate function, add dataclass

* move checking dataset in ht into compare file, simplify complete logic

* get_ht_path

* raise error on dataset config version mismatch

* rethink selects compare, use existing get field code

* typo

* review comments

* review comments

Author: jklugherz

v03_pipeline/bin/write_cached_reference_dataset_query_ht.py

@@ -15,6 +15,9 @@
     valid_reference_dataset_collection_path,
 )
 from v03_pipeline.lib.reference_data.config import CONFIG
+from v03_pipeline.lib.reference_data.dataset_table_operations import (
+    import_ht_from_config_path,
+)
 
 
 def get_ht(
@@ -25,11 +28,7 @@ def get_ht(
     # If the query is defined over an uncombined reference dataset, use the combiner config.
     if query.reference_dataset:
         config = CONFIG[query.reference_dataset][reference_genome.v02_value]
-        return (
-            config['custom_import'](config['source_path'], reference_genome)
-            if 'custom_import' in config
-            else hl.read_table(config['path'])
-        )
+        return import_ht_from_config_path(config, reference_genome)
     return hl.read_table(
         valid_reference_dataset_collection_path(
             reference_genome,

v03_pipeline/lib/reference_data/compare_globals.py

@@ -0,0 +1,145 @@
+import logging
+from dataclasses import dataclass
+
+import hail as hl
+
+from v03_pipeline.lib.model import ReferenceGenome
+from v03_pipeline.lib.reference_data.config import CONFIG
+from v03_pipeline.lib.reference_data.dataset_table_operations import (
+    get_all_select_fields,
+    get_ht_path,
+    import_ht_from_config_path,
+    parse_dataset_version,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ReferenceDataGlobals:
+    paths: dict[str]
+    versions: dict[str]
+    enums: dict[str, dict[str, list[str]]]
+
+    def __init__(self, globals_struct: hl.Struct):
+        self.paths = self._struct_to_dict(globals_struct.paths)
+        self.versions = self._struct_to_dict(globals_struct.versions)
+        self.enums = self._struct_to_dict(globals_struct.enums)
+
+    def _struct_to_dict(self, struct: hl.Struct) -> dict:
+        result_dict = {}
+        for field in struct:
+            if isinstance(struct[field], hl.Struct):
+                result_dict[field] = self._struct_to_dict(struct[field])
+            else:
+                result_dict[field] = struct[field]
+        return result_dict
+
+
+def get_datasets_to_update(
+    joined_ht: hl.Table,
+    datasets: list[str],
+    reference_genome: ReferenceGenome,
+) -> list[str]:
+    joined_ht_globals = ReferenceDataGlobals(hl.eval(joined_ht.index_globals()))
+    datasets_to_update = []
+    for dataset in datasets:
+        if dataset not in joined_ht.row:
+            datasets_to_update.append(dataset)
+            continue
+
+        if not validate_joined_ht_globals_match_config(
+            joined_ht,
+            joined_ht_globals,
+            dataset,
+            reference_genome,
+        ):
+            datasets_to_update.append(dataset)
+    return datasets_to_update
+
+
+def validate_joined_ht_globals_match_config(
+    joined_ht: hl.Table,
+    joined_ht_globals: ReferenceDataGlobals,
+    dataset: str,
+    reference_genome: ReferenceGenome,
+) -> bool:
+    dataset_config = CONFIG[dataset][reference_genome.v02_value]
+    dataset_ht = import_ht_from_config_path(dataset_config, reference_genome)
+    checks = {
+        'version': ht_version_matches_config(
+            joined_ht_globals,
+            dataset,
+            dataset_config,
+            dataset_ht,
+        ),
+        'path': ht_path_matches_config(joined_ht_globals, dataset, dataset_config),
+        'enum': ht_enums_match_config(joined_ht_globals, dataset, dataset_config),
+        'select': ht_selects_match_config(
+            joined_ht,
+            dataset,
+            dataset_config,
+            dataset_ht,
+        ),
+    }
+
+    results = []
+    for check, result in checks.items():
+        if result is False:
+            logger.info(f'{check} mismatch for {dataset}')
+        results.append(result)
+    return all(results)
+
+
+def ht_version_matches_config(
+    joined_ht_globals: ReferenceDataGlobals,
+    dataset: str,
+    dataset_config: dict,
+    dataset_ht: hl.Table,
+) -> bool:
+    joined_ht_version = joined_ht_globals.versions.get(dataset)
+    if joined_ht_version is None:
+        return False
+
+    config_or_dataset_version = hl.eval(
+        parse_dataset_version(
+            dataset_ht,
+            dataset,
+            dataset_config,
+        ),
+    )
+    return joined_ht_version == config_or_dataset_version
+
+
+def ht_path_matches_config(
+    joined_ht_globals: ReferenceDataGlobals,
+    dataset: str,
+    dataset_config: dict,
+) -> bool:
+    joined_ht_path = joined_ht_globals.paths.get(dataset)
+    if joined_ht_path is None:
+        return False
+
+    config_path = get_ht_path(dataset_config)
+    return joined_ht_path == config_path
+
+
+def ht_enums_match_config(
+    joined_ht_globals: ReferenceDataGlobals,
+    dataset: str,
+    dataset_config: dict,
+) -> bool:
+    joined_ht_enums = joined_ht_globals.enums.get(dataset, {})
+    config_enums = dataset_config.get('enum_select', {})
+    return joined_ht_enums == config_enums
+
+
+def ht_selects_match_config(
+    joined_ht: hl.Table,
+    dataset: str,
+    dataset_config: dict,
+    dataset_ht: hl.Table,
+) -> bool:
+    joined_ht_selects = set(joined_ht[dataset])
+    config_selects = set(get_all_select_fields(dataset_ht, dataset_config).keys())
+    return len(config_selects.symmetric_difference(joined_ht_selects)) == 0