Merge pull request #57 from uc-cdis/use_separate_view_for_observation_queries

Feat: use new observation_continuous view to speed up queries

Author: pieterlukasse

README.md

@@ -101,6 +101,7 @@ TABLE results.COHORT
 TABLE omop.person
 TABLE omop.observation
 TABLE omop.concept
+VIEW omop.observation_continuous
 ```
 
 

models/cohortdata.go

@@ -68,7 +68,7 @@ func (h CohortData) RetrieveDataBySourceIdAndCohortIdAndConceptIdsOrderedByPerso
 
 	// get the observations for the subjects and the concepts, to build up the data rows to return:
 	var cohortData []*PersonConceptAndValue
-	meta_result := omopDataSource.Db.Model(&Observation{}).
+	meta_result := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("observation.person_id, observation.observation_concept_id as concept_id, observation.value_as_string as concept_value_as_string, observation.value_as_number as concept_value_as_number, observation.value_as_concept_id as concept_value_as_concept_id").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as cohort ON cohort.subject_id = observation.person_id").
 		Where("cohort.cohort_definition_id = ?", cohortDefinitionId).
@@ -87,7 +87,7 @@ func (h CohortData) RetrieveHistogramDataBySourceIdAndCohortIdAndConceptIdsAndCo
 
 	// get the observations for the subjects and the concepts, to build up the data rows to return:
 	var cohortData []*PersonConceptAndValue
-	query := omopDataSource.Db.Model(&Observation{}).
+	query := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("distinct(observation.person_id), observation.observation_concept_id as concept_id, observation.value_as_number as concept_value_as_number").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as cohort ON cohort.subject_id = observation.person_id").
 		Where("cohort.cohort_definition_id = ?", cohortDefinitionId).
@@ -111,7 +111,7 @@ func (h CohortData) RetrieveCohortOverlapStats(sourceId int, caseCohortId int, c
 
 	// count persons that are in the intersection of both case and control cohorts, filtering on filterConceptValue:
 	var cohortOverlapStats CohortOverlapStats
-	query := omopDataSource.Db.Model(&Observation{}).
+	query := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("count(distinct(observation.person_id)) as case_control_overlap").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as case_cohort ON case_cohort.subject_id = observation.person_id").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as control_cohort ON control_cohort.subject_id = case_cohort.subject_id"). // this one allows for the intersection between case and control and the assessment of the overlap
@@ -136,7 +136,7 @@ func (h CohortData) RetrieveCohortOverlapStatsWithoutFilteringOnConceptValue(sou
 
 	// count persons that are in the intersection of both case and control cohorts, filtering on filterConceptValue:
 	var cohortOverlapStats CohortOverlapStats
-	query := omopDataSource.Db.Model(&Observation{}).
+	query := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("count(distinct(observation.person_id)) as case_control_overlap").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as case_cohort ON case_cohort.subject_id = observation.person_id").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as control_cohort ON control_cohort.subject_id = case_cohort.subject_id"). // this one allows for the intersection between case and control and the assessment of the overlap
@@ -173,7 +173,7 @@ func (h CohortData) ValidateObservationData(observationConceptIdsToCheck []int64
 		log.Printf("INFO: checking if no duplicate data is found for concept ids %v in `observation` table of data source %d...",
 			observationConceptIdsToCheck, source.SourceId)
 		var personConceptAndCount []*PersonConceptAndCount
-		query := omopDataSource.Db.Model(&Observation{}).
+		query := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 			Select("observation.person_id, observation.observation_concept_id as concept_id, count(*)").
 			Where("observation.observation_concept_id in (?)", observationConceptIdsToCheck).
 			Group("observation.person_id, observation.observation_concept_id").

models/concept.go

@@ -161,7 +161,7 @@ func (h Concept) RetrieveStatsBySourceIdAndCohortIdAndConceptIds(sourceId int, c
 	// no value for this concept by first finding the ones that do have some value and
 	// then subtracting them from cohort size before dividing:
 	var conceptsAndPersonsWithData []*ConceptAndPersonsWithDataStats
-	meta_result = omopDataSource.Db.Model(&Observation{}).
+	meta_result = omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("observation_concept_id as concept_id, count(distinct(person_id)) as nperson_ids").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as cohort ON cohort.subject_id = observation.person_id").
 		Where("cohort.cohort_definition_id = ?", cohortDefinitionId).
@@ -220,7 +220,7 @@ func (h Concept) RetrieveBreakdownStatsBySourceIdAndCohortIdAndConceptIdsAndCoho
 	// count persons, grouping by concept value:
 	var breakdownValueFieldName = "observation.value_as_" + getConceptValueType(breakdownConceptId)
 	var conceptBreakdownList []*ConceptBreakdown
-	query := omopDataSource.Db.Model(&Observation{}).
+	query := omopDataSource.Db.Table(omopDataSource.Schema+".observation_continuous as observation").
 		Select("observation.value_as_concept_id, count(distinct(observation.person_id)) as npersons_in_cohort_with_value").
 		Joins("INNER JOIN "+resultsDataSource.Schema+".cohort as cohort ON cohort.subject_id = observation.person_id").
 		Where("cohort.cohort_definition_id = ?", cohortDefinitionId).

models/helper.go

@@ -8,13 +8,23 @@ import (
 	"gorm.io/gorm"
 )
 
+// Helper function that adds extra filter clauses to the query, joining on the right set of tables, excluding data where necessary, etc.
+// * It was added here to make it reusable, given these filters need to be added to many of the queries that take in
+//   a list of filters in the form of concept ids and cohort pairs. The one assumption it makes is that the given `query` object already contains
+//   a basic query on a table or view that have been named or aliased as "observation" (see comments in code). This assumption is
+//   checked at the start.
 func QueryFilterByConceptIdsAndCohortPairsHelper(query *gorm.DB, filterConceptIds []int64, filterCohortPairs []utils.CustomDichotomousVariableDef, omopSchemaName string, resultSchemaName string) *gorm.DB {
+	// Validate assumption of a table or view aliased as "observation":
+	if query.Statement.Table != "observation" {
+		panic("Error: this QueryFilterByConceptIdsAndCohortPairsHelper is meant for adding extra filters to a query on a table or view with the alias name `observation`")
+	}
+
 	// iterate over the filterConceptIds, adding a new INNER JOIN and filters for each, so that the resulting set is the
 	// set of persons that have a non-null value for each and every one of the concepts:
 	for i, filterConceptId := range filterConceptIds {
 		observationTableAlias := fmt.Sprintf("observation_filter_%d", i)
 		log.Printf("Adding extra INNER JOIN with alias %s", observationTableAlias)
-		query = query.Joins("INNER JOIN "+omopSchemaName+".observation as "+observationTableAlias+" ON "+observationTableAlias+".person_id = observation.person_id").
+		query = query.Joins("INNER JOIN "+omopSchemaName+".observation_continuous as "+observationTableAlias+" ON "+observationTableAlias+".person_id = observation.person_id"). // assumption: there is a table or view named or aliased as "observation"
 			Where(observationTableAlias+".observation_concept_id = ?", filterConceptId).
 			Where("(" + observationTableAlias + ".value_as_string is not null or " + observationTableAlias + ".value_as_number is not null)") // TODO - improve performance by only filtering on type according to getConceptValueType()
 	}
@@ -33,7 +43,7 @@ func QueryFilterByConceptIdsAndCohortPairsHelper(query *gorm.DB, filterConceptId
 				"  EXCEPT "+ //now use EXCEPT to exclude the part where both cohorts INTERSECT
 				"  Select "+cohortTableAlias2+".subject_id FROM "+resultSchemaName+".cohort as "+cohortTableAlias2+
 				"  INNER JOIN "+resultSchemaName+".cohort as "+cohortTableAlias3+" ON "+cohortTableAlias3+".subject_id = "+cohortTableAlias2+".subject_id "+
-				"  where "+cohortTableAlias2+".cohort_definition_id = ? AND "+cohortTableAlias3+".cohort_definition_id =? ) AS "+unionExceptAlias+" ON "+unionExceptAlias+".subject_id = observation.person_id",
+				"  where "+cohortTableAlias2+".cohort_definition_id = ? AND "+cohortTableAlias3+".cohort_definition_id =? ) AS "+unionExceptAlias+" ON "+unionExceptAlias+".subject_id = observation.person_id", // assumption: there is a table or view named or aliased as "observation"
 			filterCohortPair.CohortId1, filterCohortPair.CohortId2, filterCohortPair.CohortId1, filterCohortPair.CohortId2)
 	}
 

tests/models_tests/models_test.go

@@ -3,6 +3,7 @@ package models_tests
 import (
 	"log"
 	"os"
+	"strings"
 	"testing"
 
 	"github.com/uc-cdis/cohort-middleware/config"
@@ -366,6 +367,46 @@ func TestRetrieveHistogramDataBySourceIdAndCohortIdAndConceptIdsAndCohortPairs(t
 	}
 }
 
+func TestQueryFilterByConceptIdsAndCohortPairsHelper(t *testing.T) {
+	// This test checks whether the query succeeds when there is a table or
+	// view aliased as "observation" and whether it fails otherwise.
+
+	setUp(t)
+	omopDataSource := tests.GetOmopDataSource()
+	filterConceptIds := []int64{1, 2, 3}
+	filterCohortPairs := []utils.CustomDichotomousVariableDef{} // empty / not really needed for test
+	var personIds []struct {
+		PersonId int64
+	}
+
+	// Subtest1: correct alias "observation":
+	query := omopDataSource.Db.Table(omopDataSource.Schema + ".observation_continuous as observation").
+		Select("observation.person_id")
+	query = models.QueryFilterByConceptIdsAndCohortPairsHelper(query, filterConceptIds, filterCohortPairs, omopDataSource.Schema, "")
+	meta_result := query.Scan(&personIds)
+	if meta_result.Error != nil {
+		t.Errorf("Did NOT expect an error")
+	}
+	// Subtest2: incorrect alias "observationWRONG"...should fail/panic:
+	defer func() {
+		panicMessage := recover()
+
+		if panicMessage == nil {
+			t.Errorf("The code did not panic")
+		}
+		panicMessageStr, isString := panicMessage.(string)
+
+		if !isString || !strings.HasPrefix(panicMessageStr, "Error: this QueryFilterByConceptIdsAndCohortPairsHelper is meant for ") {
+			t.Errorf("The code did not panic with expected error message")
+		}
+
+	}()
+	query = omopDataSource.Db.Table(omopDataSource.Schema + ".observation_continuous as observationWRONG").
+		Select("*")
+	query = models.QueryFilterByConceptIdsAndCohortPairsHelper(query, filterConceptIds, filterCohortPairs, omopDataSource.Schema, "")
+	query.Scan(&personIds)
+}
+
 func TestRetrieveDataBySourceIdAndCohortIdAndConceptIdsOrderedByPersonId(t *testing.T) {
 	setUp(t)
 	cohortDefinitions, _ := cohortDefinitionModel.GetAllCohortDefinitionsAndStatsOrderBySizeDesc(testSourceId)
@@ -404,18 +445,19 @@ func TestRetrieveDataBySourceIdAndCohortIdAndConceptIdsOrderedByPersonId(t *test
 func TestErrorForRetrieveDataBySourceIdAndCohortIdAndConceptIdsOrderedByPersonId(t *testing.T) {
 	// Tests if the method returns an error when query fails.
 
-	// break something in the omop schema to cause a query failure in the next method:
-	tests.BreakSomething(models.Omop, "observation", "person_id")
+	cohortDefinitions, _ := cohortDefinitionModel.GetAllCohortDefinitionsAndStatsOrderBySizeDesc(testSourceId)
+
+	// break something in the Results schema to cause a query failure in the next method:
+	tests.BreakSomething(models.Results, "cohort", "cohort_definition_id")
 	// set last action to restore back:
 	// run test:
-	cohortDefinitions, _ := cohortDefinitionModel.GetAllCohortDefinitionsAndStatsOrderBySizeDesc(testSourceId)
 	_, error := cohortDataModel.RetrieveDataBySourceIdAndCohortIdAndConceptIdsOrderedByPersonId(
 		testSourceId, cohortDefinitions[0].Id, allConceptIds)
 	if error == nil {
 		t.Errorf("Expected error")
 	}
 	// revert the broken part:
-	tests.FixSomething(models.Omop, "observation", "person_id")
+	tests.FixSomething(models.Results, "cohort", "cohort_definition_id")
 }
 
 // for given source and cohort, counts how many persons have the given HARE value

tests/setup_local_db/ddl_results_and_cdm.sql

@@ -94,3 +94,9 @@ CREATE TABLE omop.concept
     valid_end_date date NOT NULL DEFAULT DATE('2099-01-01'),
     invalid_reason character varying(1) COLLATE pg_catalog."default"
 );
+
+CREATE VIEW omop.OBSERVATION_CONTINUOUS AS
+SELECT ob.person_id, ob.observation_concept_id, ob.value_as_string, ob.value_as_number, ob.value_as_concept_id
+FROM omop.observation ob
+INNER JOIN omop.concept concept ON concept.CONCEPT_ID=ob.OBSERVATION_CONCEPT_ID
+WHERE concept.CONCEPT_CLASS_ID='MVP Continuous' or concept.CONCEPT_ID=2000007027

tests/setup_local_db/test_data_results_and_cdm.sql

@@ -5,10 +5,10 @@
 insert into omop.concept
 (concept_id,concept_name,domain_id,vocabulary_id,concept_class_id,standard_concept,concept_code,valid_start_date,valid_end_date,invalid_reason)
 values
-    (2000006885,'Average height ','Measurement','Measurement','Measurement','S','F','1970-01-01','2099-12-31',NULL),
-    (2000000323,'MVP Age Group','Person','Person','Person','S','F','1970-01-01','2099-12-31',NULL),
-    (2000000324,'Sex, indicated by the subject','Person','Person','Observation Type',NULL,'OMOP4822310','1970-01-01','2099-12-31',NULL),
-    (2000000280,'BMI at enrollment','Measurement','Measurement','Measurement','S','2','1970-01-01','2099-12-31',NULL)
+    (2000006885,'Average height ','Measurement','Measurement','MVP Continuous','S','F','1970-01-01','2099-12-31',NULL),
+    (2000000323,'MVP Age Group','Person','Person','MVP Continuous','S','F','1970-01-01','2099-12-31',NULL),
+    (2000000324,'Some continuous value, indicated by the subject','Person','Person','MVP Continuous',NULL,'OMOP4822310','1970-01-01','2099-12-31',NULL),
+    (2000000280,'BMI at enrollment','Measurement','Measurement','MVP Continuous','S','2','1970-01-01','2099-12-31',NULL)
 ;
 
 -- TODO - add concept_type_class "concept_class_id" "MVP Continuous" to better reflect real queries?
@@ -99,12 +99,12 @@ For HARE info, see https://pubmed.ncbi.nlm.nih.gov/31564439/.
 insert into omop.concept
 (concept_id,concept_code,concept_name,domain_id,vocabulary_id,concept_class_id,standard_concept,valid_start_date,valid_end_date,invalid_reason)
 values
-    (2000007027,'HARE_CODE','HARE',        'Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL),
-    (2000007028,'HIS', 'Hispanic',         'Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL),
-    (2000007029,'ASN','non-Hispanic Asian','Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL),
-    (2000007030,'AFR','non-Hispanic Black','Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL),
-    (2000007031,'EUR','non-Hispanic White','Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL),
-    (2000007032,'OTH','Other',             'Person','Person','Observation Type','S','1970-01-01','2099-12-31',NULL)
+    (2000007027,'HARE_CODE','HARE',        'Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL),
+    (2000007028,'HIS', 'Hispanic',         'Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL),
+    (2000007029,'ASN','non-Hispanic Asian','Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL),
+    (2000007030,'AFR','non-Hispanic Black','Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL),
+    (2000007031,'EUR','non-Hispanic White','Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL),
+    (2000007032,'OTH','Other',             'Person','Person','MVP Ordinal','S','1970-01-01','2099-12-31',NULL)
 ;
 
 -- insert `observation` records: