diff --git a/README.md b/README.md index da21e5c..1c8d063 100644 --- a/README.md +++ b/README.md @@ -37,31 +37,110 @@ The project implements an ETL conversion of MIMIC IV PhysioNet dataset to OMOP C #### To run the ETL pipeline end-to-end * load the latest standard OMOP vocabularies from http://athena.ohdsi.org * create a working copy of the loaded vocabularies, where custom mapping data will be added to -* get custom mapping vocabulary _delta tables from https://github.com/TuftsCTSI/CVB/tree/main/MIMIC/Ontology -* get other custom mapping vocabulary _delta tables, if needed +* get custom MIMIC mapping vocabulary _delta tables from https://github.com/TuftsCTSI/CVB/tree/main/MIMIC/Ontology +* get other custom mapping vocabulary _delta tables (e.g. https://github.com/TuftsCTSI/CVB/tree/main/WAVEFORM/Ontology), if needed * set variables in vocabulary_refresh/README.md * run vocabulary refresh commands given below from directory "vocabulary_refresh" * set the project variables in `conf/*.etlconf` - * run script "wf_read" to load waveform sample data if needed * run workflow commands below in the given sequence * in the workflow commands is the "environment" name, which equals "dev" for the demo dataset and "full" for the full set * set the project root (location of this file) as the current directory + +#### Waveforms +- Generate your source data in the format of the `data/waveform_files.csv` and `data/waveform_channels.csv` or use those files when performing a dummy build. Upload to BigQuery as waveform_files and waveform_channels tables, respectively. See the Waveform Source Schema section below for more details. +- The waveform build gets run when `python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_waveforms.conf` is executed, as outlined below. Tests should be run with `python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_waveforms_qa.conf` + +Vocabulary note: the standard process for combining Athena and custom vocab (_delta tables) going forward +was set in: https://github.com/OHDSI/MIMIC/pull/37 . However, since the _delta tables are evolving +regularly, you can use temporary process which adds the master Athena tables to a +BigQuery dataset and any _delta tables to their own datasets. You can then combine the Athena and all _delta tables by using a BigQuery view. The current expectation is to use +these sources for your vocabulary: +- Athena tables (Feb 2026) +- CVB MIMIC _delta tables: https://github.com/TuftsCTSI/CVB/tree/main/MIMIC +- CVB WAVEFORM _delta tables: https://github.com/TuftsCTSI/CVB/tree/main/WAVEFORM + +Hardcoded waveform concept dependencies currently used by the ETL: +- 2081500001 — physiological monitoring concept +- 2082499975 — WFDB waveform format concept + +These dependencies are validated by the waveform QA script. + +Timestamp note: timestamps in the waveform staging files are MIMIC date-shifted and represent local clinical time, not UTC. No timezone conversion is applied in the ETL. + +Procedure and device linkage note: the ETL does not currently link to waveforms to procedures or devices. Additional investigation is needed to determine if this can be added to the ETL. ``` cd vocabulary_refresh python vocabulary_refresh.py -s10 python vocabulary_refresh.py -s20 python vocabulary_refresh.py -s30 cd ../ -python scripts/wf_read.py -e conf/.etlconf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_setup.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_ddl.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_staging.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_etl.conf +python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_waveforms.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_ut.conf +python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_waveforms_qa.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_metrics.conf python scripts/run_workflow.py -e conf/.etlconf -c conf/workflow_unload.conf ``` +NOTE: the legacy process for incorporating waveforms into this ETL has been removed since the current recommendation is to populate the official Waveform Extension tables by using workflow_waveforms.conf. If needed, the legacy process can be run with `python scripts/wf_read.py -e conf/.etlconf`. + +#### Waveform Source Schema +These examples describe more detail around what is expected for the source files that are required for the waveform build. + +Some of these columns are only required for debug and have been marked as `YES` under the `Optional` column. + +The `Mode` column indicates whether a given column can have `NULL` values or not as set +by the table requirements. + +Timestamp note: timestamps in the waveform staging files are MIMIC date-shifted and represent local clinical time, not UTC. No timezone conversion is applied in the ETL. + +##### waveform_files: +| Field name | Type | Optional | Mode | Description | +|----------------------------------|-----------|----------|----------|-----------------------------------------------------------------------------------------| +| subject_id | INTEGER | YES | NULLABLE | Globally unique subject identifier from MIMIC | +| person_id | INTEGER | NO | REQUIRED | OMOP person identifier, 1:1 relationship to subject_id | +| hadm_id | INTEGER | YES | NULLABLE | Globablly unique hospital admission identifier from MIMIC | +| visit_occurrence_id | INTEGER | NO | REQUIRED | OMOP encounter identifier, 1:1 relationship to hadm_id | +| visit_detail_id | INTEGER | NO | NULLABLE | OMOP identifier for more specific encounter details (e.g. exact ICU ward) | +| location | STRING | YES | NULLABLE | Location field from WFDB header file which indicates ICU location | +| waveform_folders | STRING | YES | NULLABLE | Path to WFDB files in PhysioNet WFDB project | +| record_id | INTEGER | YES | REQUIRED | Globally unique WFDB recording identifier | +| group_id | INTEGER | NO | REQUIRED | OMOP recording identifier, 1:1 relationship to record_id | +| mimic_start | TIMESTAMP | YES | NULLABLE | The start of the recording, date shifted per MIMIC | +| mimic_end | TIMESTAMP | YES | NULLABLE | The end of the recording, date shifted per MIMIC | +| session_start | TIMESTAMP | NO | REQUIRED | The start of the recording, date shifted per OMOP | +| session_end | TIMESTAMP | NO | REQUIRED | The end of the recording, date shifted per OMOP | +| file_start | TIMESTAMP | NO | REQUIRED | The start time for a given file/segment | +| file_end | TIMESTAMP | NO | REQUIRED | The end time for a given file/segment, calculated based on the segment length | +| src_file | STRING | NO | NULLABLE | The path to the source file (e.g. original MIMIC file structure) | +| trg_file | STRING | NO | REQUIRED | The path to the final file location | + +###### waveform_channels: +| Field name | Type | Optional | Mode | Description | +|-----------------------|-----------|----------|----------|-----------------------------------------------------------------------------------| +| person_id | INTEGER | NO | REQUIRED | OMOP person identifier | +| visit_occurrence_id | INTEGER | NO | REQUIRED | OMOP encounter identifier | +| visit_detail_id | INTEGER | NO | NULLABLE | OMOP identifier for more specific encounter details (e.g. exact ICU ward) | +| group_id | INTEGER | NO | REQUIRED | OMOP recording identifier | +| session_start | TIMESTAMP | NO | REQUIRED | The start of the recording, date shifted per OMOP | +| session_end | TIMESTAMP | NO | REQUIRED | The end of the recording, date shifted per OMOP | +| file_start | TIMESTAMP | NO | REQUIRED | The start time for a given file/segment | +| file_end | TIMESTAMP | NO | REQUIRED | The end time for a given file/segment, calculated based on the segment length | +| src_file | STRING | NO | NULLABLE | The path to the source file (e.g. original MIMIC file structure) | +| trg_file | STRING | NO | REQUIRED | The path to the final file location | +| channel_index | INTEGER | NO | REQUIRED | Stable channel ordinal from WFDB channel order within the file/segment | +| channel_name | STRING | NO | NULLABLE | The channel name (e.g. "II") | +| sample_units | STRING | NO | NULLABLE | The sample amplitude units | +| sample_rate | INTEGER | NO | NULLABLE | The sample rate | +| sample_rate_units | STRING | NO | NULLABLE | The sample rate units | +| gain | FLOAT | NO | NULLABLE | The ADC (analog to digital converter) gain | +| gain_units | STRING | NO | NULLABLE | The ADC gain units | +| segment_length | INTEGER | NO | NULLABLE | The length of the segment as the number of samples | + +Note: the build makes use of the inherent channel_index within the WFDB files/segments to disambiguate if duplicate channel names occur within a file. #### To look at UT and Metrics reports * see metrics dataset name in the corresponding `.etlconf` file diff --git a/conf/workflow_ddl.conf b/conf/workflow_ddl.conf index d6641ba..2ff113e 100644 --- a/conf/workflow_ddl.conf +++ b/conf/workflow_ddl.conf @@ -5,7 +5,8 @@ "scripts": [ - {"script": "etl/ddl/ddl_voc_5_4_2.sql", "comment": ""}, - {"script": "etl/ddl/ddl_cdm_5_4_2.sql", "comment": ""} + {"script": "etl/ddl/ddl_voc_5_4_2.sql", "comment": ""}, + {"script": "etl/ddl/ddl_cdm_5_4_2.sql", "comment": ""}, + {"script": "etl/ddl/ddl_cdm_waveform_extension.sql", "comment": ""} ] } diff --git a/conf/workflow_waveforms.conf b/conf/workflow_waveforms.conf new file mode 100644 index 0000000..cf3b6d3 --- /dev/null +++ b/conf/workflow_waveforms.conf @@ -0,0 +1,12 @@ +{ + "workflow": "waveforms", + "type": "sql_script", + + "scripts": + [ + {"script": "etl/etl/cdm_waveform_occurrence.sql", "comment": ""}, + {"script": "etl/etl/cdm_waveform_registry.sql", "comment": ""}, + {"script": "etl/etl/cdm_waveform_update_num_of_files.sql", "comment": ""}, + {"script": "etl/etl/cdm_waveform_channel_metadata.sql", "comment": ""} + ] +} diff --git a/conf/workflow_waveforms_qa.conf b/conf/workflow_waveforms_qa.conf new file mode 100644 index 0000000..24c5420 --- /dev/null +++ b/conf/workflow_waveforms_qa.conf @@ -0,0 +1,10 @@ +{ + "workflow": "waveforms_qa", + "type": "sql_script", + "scripts": [ + { + "script": "test/qa/qa_wf_extension.sql", + "comment": "Blocking waveform QA checks" + } + ] +} diff --git a/etl/ddl/ddl_cdm_waveform_extension.sql b/etl/ddl/ddl_cdm_waveform_extension.sql new file mode 100644 index 0000000..33b4684 --- /dev/null +++ b/etl/ddl/ddl_cdm_waveform_extension.sql @@ -0,0 +1,76 @@ + + +CREATE OR REPLACE TABLE @etl_project.@etl_dataset.cdm_waveform_occurrence +( + waveform_occurrence_id INT64 not null, + waveform_occurrence_concept_id INT64 not null, + person_id INT64 not null, + waveform_occurrence_start_datetime DATETIME not null, + waveform_occurrence_end_datetime DATETIME not null, + visit_occurrence_id INT64 not null, + visit_detail_id INT64 , + preceding_waveform_occurrence_id INT64 , + waveform_format_concept_id INT64 , + waveform_occurrence_source_value STRING , + num_of_files INT64 , + waveform_format_source_value STRING +) +; + + +CREATE OR REPLACE TABLE @etl_project.@etl_dataset.cdm_waveform_registry +( + waveform_registry_id INT64 not null, + waveform_occurrence_id INT64 not null, + waveform_feature_id INT64 , + person_id INT64 not null, + waveform_file_start_datetime DATETIME not null, + waveform_file_end_datetime DATETIME not null, + visit_occurrence_id INT64 not null, + visit_detail_id INT64 , + file_extension_concept_id INT64 , + file_extension_source_value STRING not null, + waveform_source_file_uri STRING , + waveform_target_file_uri STRING not null +); + + +CREATE OR REPLACE TABLE @etl_project.@etl_dataset.cdm_waveform_channel_metadata +( + waveform_channel_metadata_id INT64 not null, + waveform_registry_id INT64 not null, + procedure_occurrence_id INT64 , + device_exposure_id INT64 , + waveform_channel_source_value STRING , + channel_concept_id INT64 not null, + metadata_source_value STRING not null, + metadata_concept_id INT64 not null, + value_as_number FLOAT64 , + value_as_concept_id INT64 , + value_as_string STRING , + unit_concept_id INT64 , + unit_source_value STRING +); + + +CREATE OR REPLACE TABLE @etl_project.@etl_dataset.cdm_waveform_feature +( + waveform_feature_id INT64 not null, + waveform_occurrence_id INT64 not null, + waveform_registry_id INT64 not null, + waveform_channel_metadata_id INT64 not null, + measurement_id INT64 , + observation_id INT64 , + algorithm_concept_id INT64 not null, + algorithm_source_value STRING , + anatomic_site_concept_id INT64 , + waveform_feature_start_timestamp TIME , + waveform_feature_end_timestamp TIME , + is_feature_overflow BOOLEAN , + value_as_number FLOAT64 , + value_as_concept_id INT64 , + value_as_string STRING , + value_is_a_registry_file BOOLEAN , + unit_concept_id INT64 , + unit_source_value STRING +); diff --git a/etl/etl/cdm_waveform_channel_metadata.sql b/etl/etl/cdm_waveform_channel_metadata.sql new file mode 100644 index 0000000..61675f2 --- /dev/null +++ b/etl/etl/cdm_waveform_channel_metadata.sql @@ -0,0 +1,395 @@ +-- ------------------------------------------------------------------- +-- MIMIC Waveform ETL +-- ------------------------------------------------------------------- + +DECLARE bad_registry_link INT64; +DECLARE person_visit_mismatch INT64; +DECLARE duplicate_channel_index INT64; +DECLARE selected_channel_ambiguity INT64; +DECLARE ambiguous_unit_mappings INT64; +DECLARE missing_metadata_mappings INT64; + +-- Preflight 1: each channel trg_file must resolve to exactly one registry row (1:1) +SET bad_registry_link = ( + SELECT COUNT(*) FROM ( + SELECT meta.trg_file + FROM @etl_project.@etl_dataset.waveform_channels meta + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = meta.trg_file + GROUP BY meta.trg_file + HAVING COUNTIF(r.waveform_registry_id IS NULL) > 0 + OR COUNT(DISTINCT r.waveform_registry_id) != 1 + ) +); +ASSERT bad_registry_link = 0 AS 'each channel trg_file must resolve to exactly one registry row'; + +-- Preflight 2: channel person/visit must be consistent with occurrence via registry +SET person_visit_mismatch = ( + SELECT COUNT(*) FROM ( + SELECT 1 + FROM @etl_project.@etl_dataset.waveform_channels meta + JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = meta.trg_file + JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = r.waveform_occurrence_id + WHERE meta.person_id != o.person_id + OR meta.visit_occurrence_id != o.visit_occurrence_id + ) +); +ASSERT person_visit_mismatch = 0 AS 'channel staging person/visit must match occurrence via registry'; + +---- Preflight 3: channel_index must uniquely identify channels within each registry file +SET duplicate_channel_index = ( + SELECT COUNT(*) FROM ( + SELECT + reg.waveform_registry_id, + m.channel_index + FROM @etl_project.@etl_dataset.waveform_channels m + JOIN @etl_project.@etl_dataset.cdm_waveform_registry reg + ON reg.waveform_target_file_uri = m.trg_file + GROUP BY + reg.waveform_registry_id, + m.channel_index + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_channel_index = 0 AS 'duplicate channel_index within registry file'; + + +-- Build all allowed channel concept candidates once, without applying vocabulary precedence. +-- The tier-specific ASSERTs and final COALESCE below apply WAVEFORM -> MIMIC4 -> Athena precedence. +CREATE TEMP TABLE tmp_channel_candidates_distinct AS +WITH channel_names AS ( + SELECT DISTINCT UPPER(channel_name) AS channel_name_u + FROM @etl_project.@etl_dataset.waveform_channels +), +channel_candidates AS ( + SELECT + cn.channel_name_u, + c.concept_id, + c.vocabulary_id + FROM channel_names cn + JOIN @etl_project.@etl_dataset.voc_concept c + ON UPPER(c.concept_code) = cn.channel_name_u + AND c.domain_id IN ('Waveform Metadata', 'Measurement', 'Observation') + AND c.standard_concept = 'S' + AND c.invalid_reason IS NULL + + UNION ALL + + SELECT + cn.channel_name_u, + c.concept_id, + c.vocabulary_id + FROM channel_names cn + JOIN @etl_project.@etl_dataset.voc_concept c + ON UPPER(c.concept_name) = cn.channel_name_u + AND c.domain_id IN ('Waveform Metadata', 'Measurement', 'Observation') + AND c.standard_concept = 'S' + AND c.invalid_reason IS NULL + + UNION ALL + + SELECT + cn.channel_name_u, + c.concept_id, + c.vocabulary_id + FROM channel_names cn + JOIN @etl_project.@etl_dataset.voc_concept_synonym syn + ON UPPER(syn.concept_synonym_name) = cn.channel_name_u + JOIN @etl_project.@etl_dataset.voc_concept c + ON c.concept_id = syn.concept_id + AND c.domain_id IN ('Waveform Metadata', 'Measurement', 'Observation') + AND c.standard_concept = 'S' + AND c.invalid_reason IS NULL +) +SELECT DISTINCT + channel_name_u, + concept_id, + vocabulary_id +FROM channel_candidates +; + +-- Summarize candidate counts by vocabulary tier. The selected tier is the first tier with any candidates. +CREATE TEMP TABLE tmp_channel_tier_summary AS +SELECT + channel_name_u, + CASE + WHEN vocabulary_id = 'WAVEFORM' THEN 'WAVEFORM' + WHEN vocabulary_id = 'MIMIC4' THEN 'MIMIC4' + ELSE 'ATHENA' + END AS vocabulary_tier, + ARRAY_AGG(DISTINCT concept_id ORDER BY concept_id) AS concept_ids, + COUNT(DISTINCT concept_id) AS concept_count +FROM tmp_channel_candidates_distinct +GROUP BY + channel_name_u, + vocabulary_tier +; + +CREATE TEMP TABLE tmp_channel_selected_tier AS +WITH channel_names AS ( + SELECT DISTINCT UPPER(channel_name) AS channel_name_u + FROM @etl_project.@etl_dataset.waveform_channels +) +SELECT + cn.channel_name_u, + CASE + WHEN COALESCE(waveform.concept_count, 0) > 0 THEN 'WAVEFORM' + WHEN COALESCE(mimic4.concept_count, 0) > 0 THEN 'MIMIC4' + WHEN COALESCE(athena.concept_count, 0) > 0 THEN 'ATHENA' + END AS selected_tier +FROM channel_names cn +LEFT JOIN tmp_channel_tier_summary waveform + ON waveform.channel_name_u = cn.channel_name_u + AND waveform.vocabulary_tier = 'WAVEFORM' +LEFT JOIN tmp_channel_tier_summary mimic4 + ON mimic4.channel_name_u = cn.channel_name_u + AND mimic4.vocabulary_tier = 'MIMIC4' +LEFT JOIN tmp_channel_tier_summary athena + ON athena.channel_name_u = cn.channel_name_u + AND athena.vocabulary_tier = 'ATHENA' +; + +-- Preflight 4: selected vocabulary tier must not contain ambiguous channel mappings +SET selected_channel_ambiguity = ( + SELECT COUNT(*) + FROM tmp_channel_selected_tier selected + JOIN tmp_channel_tier_summary summary + ON summary.channel_name_u = selected.channel_name_u + AND summary.vocabulary_tier = selected.selected_tier + WHERE summary.concept_count > 1 +); +ASSERT selected_channel_ambiguity = 0 AS 'ambiguous selected-tier channel mappings'; + +-- Build valid standard Unit-domain candidates once and fail only on ambiguous source units. +CREATE TEMP TABLE tmp_unit_candidates_distinct AS +WITH unit_values AS ( + SELECT DISTINCT TRIM(sample_units) AS unit_source_value_raw + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_units IS NOT NULL + + UNION DISTINCT + + SELECT DISTINCT TRIM(sample_rate_units) AS unit_source_value_raw + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_rate_units IS NOT NULL + + UNION DISTINCT + + SELECT DISTINCT TRIM(gain_units) AS unit_source_value_raw + FROM @etl_project.@etl_dataset.waveform_channels + WHERE gain_units IS NOT NULL +), +unit_candidates AS ( + SELECT + uv.unit_source_value_raw, + c.concept_id + FROM unit_values uv + JOIN @etl_project.@etl_dataset.voc_concept c + ON UPPER(c.concept_name) = UPPER(uv.unit_source_value_raw) + AND c.vocabulary_id IN ('WAVEFORM', 'MIMIC4', 'UCUM', 'SNOMED') + AND c.domain_id = 'Unit' + AND c.standard_concept = 'S' + AND c.invalid_reason IS NULL + + UNION ALL + + SELECT + uv.unit_source_value_raw, + c.concept_id + FROM unit_values uv + JOIN @etl_project.@etl_dataset.voc_concept c + ON c.concept_code = uv.unit_source_value_raw + AND c.vocabulary_id IN ('WAVEFORM', 'MIMIC4', 'UCUM', 'SNOMED') + AND c.domain_id = 'Unit' + AND c.standard_concept = 'S' + AND c.invalid_reason IS NULL +) +SELECT DISTINCT + unit_source_value_raw, + concept_id +FROM unit_candidates +; + +CREATE TEMP TABLE tmp_unit_summary AS +SELECT + unit_source_value_raw, + ARRAY_AGG(DISTINCT concept_id ORDER BY concept_id) AS concept_ids, + COUNT(DISTINCT concept_id) AS concept_count +FROM tmp_unit_candidates_distinct +GROUP BY unit_source_value_raw +; + +SET ambiguous_unit_mappings = ( + SELECT COUNT(*) + FROM tmp_unit_summary + WHERE concept_count > 1 +); +ASSERT ambiguous_unit_mappings = 0 AS 'ambiguous unit mappings'; + +-- Preflight 5: ETL-controlled metadata_type values must resolve to valid metadata concepts +SET missing_metadata_mappings = ( + WITH metadata_types AS ( + SELECT 'AMPLITUDE' AS metadata_type + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_units IS NOT NULL + + UNION DISTINCT + + SELECT 'SAMPLERATE' AS metadata_type + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_rate IS NOT NULL + + UNION DISTINCT + + SELECT 'RESOLUTION' AS metadata_type + FROM @etl_project.@etl_dataset.waveform_channels + WHERE gain IS NOT NULL + + UNION DISTINCT + + SELECT 'SEGMENTLENGTH' AS metadata_type + FROM @etl_project.@etl_dataset.waveform_channels + WHERE segment_length IS NOT NULL + ) + SELECT COUNT(*) + FROM metadata_types mt + LEFT JOIN ( + SELECT DISTINCT concept_id, UPPER(concept_name) AS concept_name_u + FROM @etl_project.@etl_dataset.voc_concept + WHERE domain_id = 'Waveform Metadata' + AND standard_concept = 'S' + AND invalid_reason IS NULL + QUALIFY ROW_NUMBER() OVER (PARTITION BY UPPER(concept_name) ORDER BY concept_id) = 1 + ) vc_metadata + ON vc_metadata.concept_name_u = UPPER(mt.metadata_type) + WHERE vc_metadata.concept_id IS NULL +); +ASSERT missing_metadata_mappings = 0 AS 'unresolved metadata_type mappings'; + +TRUNCATE TABLE @etl_project.@etl_dataset.cdm_waveform_channel_metadata; + +INSERT INTO @etl_project.@etl_dataset.cdm_waveform_channel_metadata +WITH channel_metadata_unpivoted AS ( + -- channel_index is source-derived from WFDB channel order and used to disambiguate + -- duplicate labels within a file + SELECT + person_id, visit_occurrence_id, group_id, trg_file, channel_index, channel_name, sample_units, + 'AMPLITUDE' AS metadata_type, + CAST(NULL AS FLOAT64) AS value_as_number, + CAST(NULL AS INT64) AS value_as_concept_id, + CAST(NULL AS STRING) AS value_as_string, + sample_units AS unit_source_value + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_units IS NOT NULL + + UNION ALL + + SELECT + person_id, visit_occurrence_id, group_id, trg_file, channel_index, channel_name, sample_units, + 'SAMPLERATE' AS metadata_type, + sample_rate AS value_as_number, + CAST(NULL AS INT64) AS value_as_concept_id, + CAST(NULL AS STRING) AS value_as_string, + sample_rate_units AS unit_source_value + FROM @etl_project.@etl_dataset.waveform_channels + WHERE sample_rate IS NOT NULL + + UNION ALL + + SELECT + person_id, visit_occurrence_id, group_id, trg_file, channel_index, channel_name, sample_units, + 'RESOLUTION' AS metadata_type, + gain AS value_as_number, + CAST(NULL AS INT64) AS value_as_concept_id, + CAST(NULL AS STRING) AS value_as_string, + gain_units AS unit_source_value + FROM @etl_project.@etl_dataset.waveform_channels + WHERE gain IS NOT NULL + + UNION ALL + + SELECT + person_id, visit_occurrence_id, group_id, trg_file, channel_index, channel_name, sample_units, + 'SEGMENTLENGTH' AS metadata_type, + segment_length AS value_as_number, + CAST(NULL AS INT64) AS value_as_concept_id, + CAST(NULL AS STRING) AS value_as_string, + CAST(NULL AS STRING) AS unit_source_value + FROM @etl_project.@etl_dataset.waveform_channels + WHERE segment_length IS NOT NULL +), +channel_map AS ( + SELECT + selected.channel_name_u, + summary.concept_ids[SAFE_OFFSET(0)] AS concept_id + FROM tmp_channel_selected_tier selected + JOIN tmp_channel_tier_summary summary + ON summary.channel_name_u = selected.channel_name_u + AND summary.vocabulary_tier = selected.selected_tier + WHERE summary.concept_count = 1 +), +unit_map AS ( + SELECT + unit_source_value_raw, + concept_ids[SAFE_OFFSET(0)] AS concept_id + FROM tmp_unit_summary + WHERE concept_count = 1 +) +SELECT + `@etl_project.@etl_dataset.obf_id_str`( + TO_JSON_STRING(STRUCT( + r.waveform_registry_id AS waveform_registry_id, + meta.channel_index AS channel_index, + meta.metadata_type AS metadata_type + )), + 64 + ) AS waveform_channel_metadata_id, + r.waveform_registry_id AS waveform_registry_id, + CAST(NULL AS INT64) AS procedure_occurrence_id, + CAST(NULL AS INT64) AS device_exposure_id, + meta.channel_name AS waveform_channel_source_value, + + -- Map channel_name to channel_concept_id using selected vocabulary tier precedence WAVEFORM -> MIMIC4 -> Athena + channel_map.concept_id AS channel_concept_id, + + -- Map metadata_type to metadata_concept_id + meta.metadata_type AS metadata_source_value, + vc_metadata.concept_id AS metadata_concept_id, + + meta.value_as_number AS value_as_number, + meta.value_as_concept_id AS value_as_concept_id, + meta.value_as_string AS value_as_string, + + -- Map unit_source_value to unit_concept_id using valid standard Unit-domain concepts only + unit_map.concept_id AS unit_concept_id, + meta.unit_source_value AS unit_source_value + +FROM + channel_metadata_unpivoted meta + +-- Join 0: Resolve waveform_registry_id from populated WAVEFORM_REGISTRY by target URI +-- (1:1 on trg_file). +JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = meta.trg_file + +-- Join 1: Resolve channel_name to channel_concept_id from selected vocabulary tier +LEFT JOIN channel_map + ON channel_map.channel_name_u = UPPER(meta.channel_name) + +-- Join 2: Map metadata_type (SAMPLERATE, GAIN, etc) to metadata_concept_id +LEFT JOIN + (SELECT DISTINCT concept_id, concept_name + FROM @etl_project.@etl_dataset.voc_concept + WHERE domain_id = 'Waveform Metadata' + AND standard_concept = 'S' + AND invalid_reason IS NULL + QUALIFY ROW_NUMBER() OVER (PARTITION BY UPPER(concept_name) ORDER BY concept_id) = 1 + ) vc_metadata + ON UPPER(vc_metadata.concept_name) = UPPER(meta.metadata_type) + +-- Join 3: Map unit_source_value to unit_concept_id from resolved standard Unit-domain mappings +LEFT JOIN unit_map + ON unit_map.unit_source_value_raw = TRIM(meta.unit_source_value) +; diff --git a/etl/etl/cdm_waveform_occurrence.sql b/etl/etl/cdm_waveform_occurrence.sql new file mode 100644 index 0000000..f3dd211 --- /dev/null +++ b/etl/etl/cdm_waveform_occurrence.sql @@ -0,0 +1,59 @@ +-- ------------------------------------------------------------------- +-- MIMIC Waveform ETL +-- ------------------------------------------------------------------- + +DECLARE bad_groups INT64; +DECLARE bad_time INT64; + +-- Preflight 1: validate group_id grain is unique for person/visit/session interval + +SET bad_groups = ( + SELECT COUNT(*) FROM ( + SELECT group_id + FROM @etl_project.@etl_dataset.waveform_files + GROUP BY group_id + HAVING COUNT(DISTINCT person_id) > 1 + OR COUNT(DISTINCT visit_occurrence_id) > 1 + OR COUNT(DISTINCT visit_detail_id) > 1 + OR COUNT(DISTINCT session_start) > 1 + OR COUNT(DISTINCT session_end) > 1 + ) +); +ASSERT bad_groups = 0 AS 'group_id violates expected grain (person, visit, session_start/end)'; + +-- Preflight 2: occurrence end must be >= start +SET bad_time = ( + SELECT COUNT(*) FROM ( + SELECT + MIN(DATETIME(file_start)) AS s, + MAX(DATETIME(file_end)) AS e + FROM @etl_project.@etl_dataset.waveform_files + GROUP BY group_id + HAVING e < s + ) +); +ASSERT bad_time = 0 AS 'occurrence end < start for at least one group_id'; + +TRUNCATE TABLE @etl_project.@etl_dataset.cdm_waveform_occurrence; + +INSERT INTO @etl_project.@etl_dataset.cdm_waveform_occurrence +SELECT + `@etl_project.@etl_dataset`.obf_id(f.group_id, 32) AS waveform_occurrence_id, + -- Physiological monitoring from TuftsCTSI + CAST(2081500001 AS INT64) AS waveform_occurrence_concept_id, + MIN(f.person_id) AS person_id, + MIN(DATETIME(f.session_start)) AS waveform_occurrence_start_datetime, + MAX(DATETIME(f.session_end)) AS waveform_occurrence_end_datetime, + MIN(f.visit_occurrence_id) AS visit_occurrence_id, + MIN(f.visit_detail_id) AS visit_detail_id, + CAST(NULL AS INT64) AS preceding_waveform_occurrence_id, + -- WFDB from TuftsCTSI + CAST(2082499975 AS INT64) AS waveform_format_concept_id, + CAST(f.group_id AS STRING) AS waveform_occurrence_source_value, + COUNT(*) AS num_of_files, + 'WFDB' AS waveform_format_source_value + + FROM @etl_project.@etl_dataset.waveform_files f + GROUP BY + f.group_id +; diff --git a/etl/etl/cdm_waveform_registry.sql b/etl/etl/cdm_waveform_registry.sql new file mode 100644 index 0000000..4fe8690 --- /dev/null +++ b/etl/etl/cdm_waveform_registry.sql @@ -0,0 +1,94 @@ +-- ------------------------------------------------------------------- +-- MIMIC Waveform ETL +-- ------------------------------------------------------------------- + +DECLARE missing_trg INT64; +DECLARE missing_occ INT64; + +-- Preflight 1: trg_file must be present for all rows +SET missing_trg = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.waveform_files + WHERE trg_file IS NULL OR TRIM(trg_file) = '' +); +ASSERT missing_trg = 0 AS 'staging contains rows with empty trg_file; canonical target URI is required' +; + +-- Preflight 2: every group_id used for registry must resolve to exactly one occurrence +SET missing_occ = ( + SELECT COUNT(*) FROM ( + SELECT `@etl_project.@etl_dataset.obf_id`(group_id, 32) AS occ_id + FROM @etl_project.@etl_dataset.waveform_files + GROUP BY occ_id + HAVING NOT EXISTS ( + SELECT 1 + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence o + WHERE o.waveform_occurrence_id = occ_id + ) + ) +); +ASSERT missing_occ = 0 AS 'registry rows must resolve to an existing occurrence (1:1 by group_id)'; + +TRUNCATE TABLE @etl_project.@etl_dataset.cdm_waveform_registry; + +INSERT INTO @etl_project.@etl_dataset.cdm_waveform_registry +-- Make one registry row per target file (trg_file); preserve raw extension; map via normalized extension +WITH files_with_extensions AS ( + SELECT + f.*, + REGEXP_EXTRACT(TRIM(f.trg_file), r'(\.[^.]+)$') AS raw_trg_ext, + UPPER(REGEXP_EXTRACT(TRIM(f.trg_file), r'\.([^.]+)$')) AS norm_trg_ext + FROM @etl_project.@etl_dataset.waveform_files f +), +file_rows AS ( + SELECT + `@etl_project.@etl_dataset.obf_id_str`(f.trg_file, 32) AS waveform_registry_id, + `@etl_project.@etl_dataset.obf_id`(f.group_id, 32) AS waveform_occurrence_id, + CAST(NULL AS INT64) AS waveform_feature_id, + f.person_id AS person_id, + DATETIME(f.file_start) AS waveform_file_start_datetime, + DATETIME(f.file_end) AS waveform_file_end_datetime, + f.visit_occurrence_id AS visit_occurrence_id, + f.visit_detail_id AS visit_detail_id, + f.raw_trg_ext AS extracted_extension, -- raw (with dot, case-preserving) + f.norm_trg_ext AS normalized_extension, -- uppercase, no dot (for mapping only) + f.src_file AS waveform_source_file_uri, + f.trg_file AS waveform_target_file_uri + FROM files_with_extensions f +) +SELECT + fr.waveform_registry_id AS waveform_registry_id, + fr.waveform_occurrence_id, + fr.waveform_feature_id, + o.person_id, + fr.waveform_file_start_datetime, + fr.waveform_file_end_datetime, + o.visit_occurrence_id, + o.visit_detail_id, + wc1.concept_id AS file_extension_concept_id, + fr.extracted_extension AS file_extension_source_value, + fr.waveform_source_file_uri AS waveform_source_file_uri, + fr.waveform_target_file_uri AS waveform_target_file_uri + + FROM file_rows fr + LEFT JOIN ( + SELECT concept_id, UPPER(concept_name) AS concept_name_u + FROM @etl_project.@etl_dataset.voc_concept + WHERE domain_id = 'Waveform Metadata' + QUALIFY ROW_NUMBER() OVER (PARTITION BY UPPER(concept_name) ORDER BY concept_id) = 1 + ) wc + ON wc.concept_name_u = fr.normalized_extension + LEFT JOIN ( SELECT concept_id_1, concept_id_2 + FROM @etl_project.@etl_dataset.voc_concept_relationship + WHERE relationship_id = 'Maps to' + AND invalid_reason IS NULL + QUALIFY ROW_NUMBER() OVER (PARTITION BY concept_id_1 ORDER BY concept_id_2) = 1 + ) cr1 + ON cr1.concept_id_1 = wc.concept_id + LEFT JOIN @etl_project.@etl_dataset.voc_concept wc1 + ON wc1.concept_id = cr1.concept_id_2 + AND wc1.invalid_reason IS NULL + AND wc1.standard_concept = 'S' + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = fr.waveform_occurrence_id +; diff --git a/etl/etl/cdm_waveform_update_num_of_files.sql b/etl/etl/cdm_waveform_update_num_of_files.sql new file mode 100644 index 0000000..8e3035a --- /dev/null +++ b/etl/etl/cdm_waveform_update_num_of_files.sql @@ -0,0 +1,9 @@ +-- Set num_of_files from actual registry rows +UPDATE @etl_project.@etl_dataset.cdm_waveform_occurrence wo +SET num_of_files = r.cnt +FROM ( + SELECT waveform_occurrence_id, COUNT(*) AS cnt + FROM @etl_project.@etl_dataset.cdm_waveform_registry + GROUP BY waveform_occurrence_id +) r +WHERE wo.waveform_occurrence_id = r.waveform_occurrence_id; \ No newline at end of file diff --git a/scripts/bq_run_waveform_script.py b/scripts/bq_run_waveform_script.py new file mode 100644 index 0000000..8d49940 --- /dev/null +++ b/scripts/bq_run_waveform_script.py @@ -0,0 +1,210 @@ +# +# Run full SQL script files in BigQuery without splitting on semicolons. +# Intended for BigQuery scripting features such as DECLARE, ASSERT, +# and CREATE TEMP TABLE reused later in the same script. +# + +import os +import sys +import getopt +import json +import datetime +import subprocess + +config_default = { + + "variables": { + + "@variable_1": "No project replacement by default", + "@variable_2": "No dataset replacement by default" + }, + + "escaping_chars": { + '"': '\\"' + } +} + + +def read_params(): + + print('Reading params...') + params = { + "etlconf_file": "", + "config_file": "", + "script_files": [], + "files_not_found": [] + } + + try: + opts, args = getopt.getopt(sys.argv[1:],"e:c:",["etlconf=,config="]) + if len(args) == 0: + raise getopt.GetoptError("read_params() error", "Mandatory argument is missing.") + + except getopt.GetoptError as err: + print(err.args) + print("Please indicate correct params:") + print("etlconf_file: optional: indicate '-e' for 'etlconf', global config json file") + print("config_file: optional: indicate '-c' for 'config', local config json file") + print("script_files: [mandatory: indicate at least one script name as unnamed argument]") + sys.exit(2) + + for opt, arg in opts: + if opt == '-e' or opt == '--etlconf': + if os.path.isfile(arg): + params['etlconf_file'] = arg + if opt == '-c' or opt == '--config': + if os.path.isfile(arg): + params['config_file'] = arg + + for arg in args: + if os.path.isfile(arg): + params['script_files'].append(arg) + else: + params['files_not_found'].append(arg) + + print('scripts to run', params) + return params + + +def read_config(etlconf_file, config_file): + + print('Reading config...') + config = {} + config_read = {} + etlconf_read = {} + + if os.path.isfile(etlconf_file): + with open(etlconf_file) as f: + etlconf_read = json.load(f) + + if os.path.isfile(config_file): + with open(config_file) as f: + config_read = json.load(f) + + for k in config_default: + s = etlconf_read.get(k, config_default[k]) + config[k] = s + + for k in config_default: + s = config_read.get(k, config[k]) + config[k] = s + + print(config) + return config + + +def remove_comments(s_query): + + print('Remove_comments()...') + + s_lines_src = s_query.split('\n') + s_result = "" + + for s in s_lines_src: + comment_flag = s.replace(' ', '')[0:2] + + if comment_flag != '--' and len(s) > 0: + s_result = s_result + s + '\n' + + return s_result + + +def format_query(s_query, config): + + print('Formatting query...') + + s_result = s_query + + for var, val in config['escaping_chars'].items(): + s_result = s_result.replace(var, val) + + for var, val in config['variables'].items(): + s_result = s_result.replace(var, val) + + print(s_result) + return s_result + + +def troubleshooting_bqc_format(bqc): + + print('Troubleshooting_bqc_format()...') + + s_lines_src = bqc.split('\n') + s_result = "" + + for s in s_lines_src: + + comment_pos = s.find('--') + if comment_pos > -1: + s = s[0:comment_pos].strip() + + if len(s) > 0: + s_result = s_result + s + ' ' + + return s_result + + +def nice_message(s_filename, status, msg): + time = datetime.datetime.now() + file = s_filename.ljust(35, ' ') + result = 'Done.' if status==0 else 'Error' + message = '' if len(msg)==0 else ': ' + msg if len(msg.split('\n')) == 1 else '\n' + '\n'.join(map(lambda x: ''.ljust(4) + x, msg.split('\n'))) + + return '{0} | {1} | {2}{3}'.format(time, file, result, message) + + +def main(): + + rc = 0 + duration = datetime.datetime.now() + params = read_params() + config = read_config(params['etlconf_file'], params['config_file']) + + if len(params['files_not_found']) > 0: + rc = 2 + for s_filename in params['files_not_found']: + print('No such file or directory: {file}\n'.format(file=s_filename)) + + else: + bq_base_cmd = ["bq", "query", "--use_legacy_sql=false"] + s_done = [] + s_done.append(nice_message('start...', 0, '')) + + for s_filename in params['script_files']: + + print('Run script {file}\n'.format(file=s_filename)) + + s_query = open(s_filename).read() + formatted_query = troubleshooting_bqc_format(format_query(remove_comments(s_query), config)) + cmd = bq_base_cmd + [formatted_query] + + print('Starting query...') + try: + completed = subprocess.run(cmd, capture_output=True, text=True) + rc = completed.returncode + if rc != 0: + print('bq stdout:\n' + completed.stdout) + print('bq stderr:\n' + completed.stderr) + except FileNotFoundError: + rc = 127 + print('Error: bq CLI not found in PATH') + + s_done.append( + nice_message(s_filename, rc, '' if rc==0 else 'See script output above')) + + if rc != 0: + break + + print('\nScripts executed:') + for a in s_done: + print(a) + duration = datetime.datetime.now() - duration + print('Run time: {0}'.format(duration)) + + return rc + + +return_code = main() + +print('bq_run_waveform_script.exit()', return_code) +exit(return_code) diff --git a/scripts/run_workflow.py b/scripts/run_workflow.py index 81ededb..4d32fef 100644 --- a/scripts/run_workflow.py +++ b/scripts/run_workflow.py @@ -22,7 +22,7 @@ "workflow": "ddl, staging, etl etc", "comment": "", - "type": "sql or py", + "type": "sql, sql_script, or py", "variables": { "@etl_project": "target project name", @@ -127,6 +127,13 @@ def main(): config = read_config(params['etlconf_file'], params['config_file']) run_command_bq_script = "python scripts/bq_run_script.py {e} {etlconf_file} {c} {config_file} {script_file}" + run_command_bq_sql_script = "python scripts/bq_run_waveform_script.py {e} {etlconf_file} {c} {config_file} {script_file}" + + if config['type'] == 'sql_script': + run_command_template = run_command_bq_sql_script + else: + run_command_template = run_command_bq_script + to_run = \ config['scripts'] \ @@ -134,7 +141,7 @@ def main(): else params['script_files'] # run all given scripts at a time - run_command = run_command_bq_script.format( + run_command = run_command_template.format( script_file= ' '.join(map( lambda s : s['script'], to_run)), e = ('-e' if len(params['etlconf_file'])> 0 else ''), etlconf_file= params['etlconf_file'], diff --git a/test/qa/qa_wf_extension.sql b/test/qa/qa_wf_extension.sql new file mode 100644 index 0000000..9389049 --- /dev/null +++ b/test/qa/qa_wf_extension.sql @@ -0,0 +1,331 @@ +-- 1. Duplicate primary keys. All three queries must return zero rows. +DECLARE duplicate_waveform_occurrence_ids INT64; +SET duplicate_waveform_occurrence_ids = ( + SELECT COUNT(*) FROM ( + SELECT waveform_occurrence_id + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence + GROUP BY waveform_occurrence_id + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_waveform_occurrence_ids = 0 AS 'duplicate waveform_occurrence_id values'; + +DECLARE duplicate_waveform_registry_ids INT64; +SET duplicate_waveform_registry_ids = ( + SELECT COUNT(*) FROM ( + SELECT waveform_registry_id + FROM @etl_project.@etl_dataset.cdm_waveform_registry + GROUP BY waveform_registry_id + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_waveform_registry_ids = 0 AS 'duplicate waveform_registry_id values'; + +DECLARE duplicate_waveform_channel_metadata_ids INT64; +SET duplicate_waveform_channel_metadata_ids = ( + SELECT COUNT(*) FROM ( + SELECT waveform_channel_metadata_id + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata + GROUP BY waveform_channel_metadata_id + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_waveform_channel_metadata_ids = 0 AS 'duplicate waveform_channel_metadata_id values'; + +-- 2. Missing occurrence foreign keys. Must return zero rows. +DECLARE orphan_registry_rows INT64; +SET orphan_registry_rows = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_registry r + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = r.waveform_occurrence_id + WHERE o.waveform_occurrence_id IS NULL +); +ASSERT orphan_registry_rows = 0 AS 'orphan registry rows'; + +-- 3. Missing registry foreign keys. Must return zero rows. +DECLARE orphan_channel_metadata_rows INT64; +SET orphan_channel_metadata_rows = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata m + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_registry_id = m.waveform_registry_id + WHERE r.waveform_registry_id IS NULL +); +ASSERT orphan_channel_metadata_rows = 0 AS 'orphan channel metadata rows'; + +-- 4. Invalid preceding occurrence references. Must return zero rows. +DECLARE invalid_preceding_occurrence_refs INT64; +SET invalid_preceding_occurrence_refs = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence o + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence p + ON p.waveform_occurrence_id = o.preceding_waveform_occurrence_id + WHERE o.preceding_waveform_occurrence_id IS NOT NULL + AND p.waveform_occurrence_id IS NULL +); +ASSERT invalid_preceding_occurrence_refs = 0 AS 'invalid preceding waveform occurrence references'; + +-- 5. Invalid occurrence intervals. Must return zero rows. +DECLARE invalid_occurrence_intervals INT64; +SET invalid_occurrence_intervals = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence + WHERE waveform_occurrence_start_datetime IS NULL + OR waveform_occurrence_end_datetime IS NULL + OR waveform_occurrence_end_datetime < waveform_occurrence_start_datetime +); +ASSERT invalid_occurrence_intervals = 0 AS 'invalid waveform occurrence intervals'; + +-- 6. Invalid registry intervals. Must return zero rows. +DECLARE invalid_registry_intervals INT64; +SET invalid_registry_intervals = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_registry + WHERE waveform_file_start_datetime IS NULL + OR waveform_file_end_datetime IS NULL + OR waveform_file_end_datetime < waveform_file_start_datetime +); +ASSERT invalid_registry_intervals = 0 AS 'invalid waveform registry intervals'; + +-- 7. Registry timestamps outside the parent occurrence. Must return zero rows unless a documented tolerance or exception policy applies. +DECLARE registry_outside_occurrence_intervals INT64; +SET registry_outside_occurrence_intervals = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_registry r + JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = r.waveform_occurrence_id + WHERE r.waveform_file_start_datetime < o.waveform_occurrence_start_datetime + OR r.waveform_file_end_datetime > o.waveform_occurrence_end_datetime +); +ASSERT registry_outside_occurrence_intervals = 0 AS 'registry timestamps outside parent occurrence window'; + +-- 8. Person and visit inconsistencies. Must return zero rows when registry context is intended to be inherited directly. +DECLARE registry_context_mismatches INT64; +SET registry_context_mismatches = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_registry r + JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = r.waveform_occurrence_id + WHERE r.person_id != o.person_id + OR r.visit_occurrence_id != o.visit_occurrence_id + OR (r.visit_detail_id IS DISTINCT FROM o.visit_detail_id) +); +ASSERT registry_context_mismatches = 0 AS 'registry context mismatches parent occurrence'; + +-- 9. File-count inconsistencies. Must return zero rows. +DECLARE inconsistent_occurrence_file_counts INT64; +SET inconsistent_occurrence_file_counts = ( + SELECT COUNT(*) FROM ( + SELECT + o.waveform_occurrence_id, + o.num_of_files, + COUNT(r.waveform_registry_id) AS registry_file_count + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence o + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_occurrence_id = o.waveform_occurrence_id + GROUP BY o.waveform_occurrence_id, o.num_of_files + HAVING o.num_of_files != COUNT(r.waveform_registry_id) + ) +); +ASSERT inconsistent_occurrence_file_counts = 0 AS 'waveform_occurrence.num_of_files inconsistent with registry row count'; + +-- 10. Duplicate target files. Must return zero rows unless duplicate logical registration of one physical file is explicitly intended and documented. +DECLARE duplicate_target_file_uris INT64; +SET duplicate_target_file_uris = ( + SELECT COUNT(*) FROM ( + SELECT waveform_target_file_uri + FROM @etl_project.@etl_dataset.cdm_waveform_registry + GROUP BY waveform_target_file_uri + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_target_file_uris = 0 AS 'duplicate waveform_target_file_uri values'; + +-- 11a. Missing required occurrence concepts. Must return zero rows for the final production tables. +DECLARE missing_occurrence_required_concepts INT64; +SET missing_occurrence_required_concepts = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence + WHERE waveform_occurrence_concept_id IS NULL + OR waveform_occurrence_concept_id = 0 +); +ASSERT missing_occurrence_required_concepts = 0 AS 'missing required waveform occurrence concepts'; + +-- 11b. Missing required channel concepts. Temporarily disabled pending vocabulary updates: +-- CVB issue #31: https://github.com/TuftsCTSI/CVB/issues/31 +-- DECLARE missing_channel_required_concepts INT64; +-- SET missing_channel_required_concepts = ( +-- SELECT COUNT(*) +-- FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata +-- WHERE channel_concept_id IS NULL +-- OR channel_concept_id = 0 +-- ); +-- ASSERT missing_channel_required_concepts = 0 AS 'missing required channel concepts'; + +-- 11c. Missing required metadata concepts. Temporarily disabled pending vocabulary updates: +-- CVB issue #26: https://github.com/TuftsCTSI/CVB/issues/26 +-- DECLARE missing_metadata_required_concepts INT64; +-- SET missing_metadata_required_concepts = ( +-- SELECT COUNT(*) +-- FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata +-- WHERE metadata_concept_id IS NULL +-- OR metadata_concept_id = 0 +-- ); +-- ASSERT missing_metadata_required_concepts = 0 AS 'missing required metadata concepts'; + +-- 12. Invalid or missing vocabulary concepts. Must return zero rows. +DECLARE invalid_or_missing_vocabulary_concepts INT64; +SET invalid_or_missing_vocabulary_concepts = ( + SELECT COUNT(*) + FROM ( + SELECT DISTINCT x.concept_id + FROM ( + SELECT waveform_occurrence_concept_id AS concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence + + UNION DISTINCT + + SELECT waveform_format_concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_occurrence + WHERE waveform_format_concept_id IS NOT NULL + + UNION DISTINCT + + SELECT file_extension_concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_registry + WHERE file_extension_concept_id IS NOT NULL + + UNION DISTINCT + + SELECT channel_concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata + + UNION DISTINCT + + SELECT metadata_concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata + + UNION DISTINCT + + SELECT unit_concept_id + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata + WHERE unit_concept_id IS NOT NULL + ) x + LEFT JOIN @etl_project.@etl_dataset.voc_concept c + ON c.concept_id = x.concept_id + WHERE x.concept_id != 0 + AND ( + c.concept_id IS NULL + OR c.invalid_reason IS NOT NULL + ) + ) +); +ASSERT invalid_or_missing_vocabulary_concepts = 0 AS 'invalid or missing vocabulary concepts referenced by waveform extension tables'; + +-- 13. Invalid unit concepts. Must return zero rows unless a documented custom-unit exception exists. +DECLARE invalid_unit_concepts INT64; +SET invalid_unit_concepts = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata m + LEFT JOIN @etl_project.@etl_dataset.voc_concept c + ON c.concept_id = m.unit_concept_id + WHERE m.unit_concept_id IS NOT NULL + AND ( + c.concept_id IS NULL + OR c.domain_id != 'Unit' + OR c.standard_concept != 'S' + OR c.invalid_reason IS NOT NULL + ) +); +ASSERT invalid_unit_concepts = 0 AS 'invalid unit_concept_id values'; + +-- 14. Empty channel metadata records. Must return zero rows. +DECLARE empty_channel_metadata_records INT64; +SET empty_channel_metadata_records = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_channel_metadata + WHERE value_as_number IS NULL + AND value_as_concept_id IS NULL + AND value_as_string IS NULL + AND unit_concept_id IS NULL + AND unit_source_value IS NULL +); +ASSERT empty_channel_metadata_records = 0 AS 'empty channel metadata records'; + +-- 15. Duplicate channel metadata grain. Must return zero rows unless repeated metadata values are explicitly modeled. +DECLARE duplicate_channel_metadata_grain INT64; +SET duplicate_channel_metadata_grain = ( + SELECT COUNT(*) FROM ( + WITH staged_channel_metadata AS ( + SELECT + r.waveform_registry_id, + wc.channel_index, + 'AMPLITUDE' AS metadata_source_value + FROM @etl_project.@etl_dataset.waveform_channels wc + JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = wc.trg_file + WHERE wc.sample_units IS NOT NULL + + UNION ALL + + SELECT + r.waveform_registry_id, + wc.channel_index, + 'SAMPLERATE' AS metadata_source_value + FROM @etl_project.@etl_dataset.waveform_channels wc + JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = wc.trg_file + WHERE wc.sample_rate IS NOT NULL + + UNION ALL + + SELECT + r.waveform_registry_id, + wc.channel_index, + 'RESOLUTION' AS metadata_source_value + FROM @etl_project.@etl_dataset.waveform_channels wc + JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = wc.trg_file + WHERE wc.gain IS NOT NULL + + UNION ALL + + SELECT + r.waveform_registry_id, + wc.channel_index, + 'SEGMENTLENGTH' AS metadata_source_value + FROM @etl_project.@etl_dataset.waveform_channels wc + JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_target_file_uri = wc.trg_file + WHERE wc.segment_length IS NOT NULL + ) + SELECT + waveform_registry_id, + channel_index, + metadata_source_value + FROM staged_channel_metadata + GROUP BY + waveform_registry_id, + channel_index, + metadata_source_value + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_channel_metadata_grain = 0 AS 'duplicate channel metadata at intended natural grain'; + +-- 16. Required hardcoded waveform vocabulary dependencies must exist and be valid. +DECLARE invalid_hardcoded_waveform_concepts INT64; +SET invalid_hardcoded_waveform_concepts = ( + SELECT COUNT(*) + FROM ( + SELECT 2081500001 AS concept_id + UNION ALL + SELECT 2082499975 AS concept_id + ) required_concepts + LEFT JOIN @etl_project.@etl_dataset.voc_concept c + ON c.concept_id = required_concepts.concept_id + WHERE c.concept_id IS NULL + OR c.invalid_reason IS NOT NULL +); +ASSERT invalid_hardcoded_waveform_concepts = 0 AS 'missing or invalid hardcoded waveform concept dependencies'; diff --git a/test/qa/qa_wf_feature.sql b/test/qa/qa_wf_feature.sql new file mode 100644 index 0000000..d4f145b --- /dev/null +++ b/test/qa/qa_wf_feature.sql @@ -0,0 +1,46 @@ +-- waveform_feature QA checks. Enable in workflow_waveforms_qa.conf when waveform_feature ETL/output is available. + +-- 1. Duplicate waveform_feature primary keys. Must return zero rows. +DECLARE duplicate_waveform_feature_ids INT64; +SET duplicate_waveform_feature_ids = ( + SELECT COUNT(*) FROM ( + SELECT waveform_feature_id + FROM @etl_project.@etl_dataset.cdm_waveform_feature + GROUP BY waveform_feature_id + HAVING COUNT(*) > 1 + ) +); +ASSERT duplicate_waveform_feature_ids = 0 AS 'duplicate waveform_feature_id values'; + +-- 2. Missing occurrence foreign keys from waveform_feature. Must return zero rows. +DECLARE orphan_feature_occurrence_rows INT64; +SET orphan_feature_occurrence_rows = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_feature f + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_occurrence o + ON o.waveform_occurrence_id = f.waveform_occurrence_id + WHERE o.waveform_occurrence_id IS NULL +); +ASSERT orphan_feature_occurrence_rows = 0 AS 'orphan waveform_feature rows by waveform_occurrence_id'; + +-- 3. Missing registry foreign keys from waveform_feature. Must return zero rows. +DECLARE orphan_feature_registry_rows INT64; +SET orphan_feature_registry_rows = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_feature f + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_registry r + ON r.waveform_registry_id = f.waveform_registry_id + WHERE r.waveform_registry_id IS NULL +); +ASSERT orphan_feature_registry_rows = 0 AS 'orphan waveform_feature rows by waveform_registry_id'; + +-- 4. Missing channel metadata foreign keys from waveform_feature. Must return zero rows. +DECLARE orphan_feature_channel_metadata_rows INT64; +SET orphan_feature_channel_metadata_rows = ( + SELECT COUNT(*) + FROM @etl_project.@etl_dataset.cdm_waveform_feature f + LEFT JOIN @etl_project.@etl_dataset.cdm_waveform_channel_metadata m + ON m.waveform_channel_metadata_id = f.waveform_channel_metadata_id + WHERE m.waveform_channel_metadata_id IS NULL +); +ASSERT orphan_feature_channel_metadata_rows = 0 AS 'orphan waveform_feature rows by waveform_channel_metadata_id';