Skip to content

Commit d23b434

Browse files
alex-odysseushernaldourbinaoleg-odysseus
authored
Cohort generation with Demographics (#2402)
Extended cohort generation to include demographic characterization. Add function to get demographic report from cohort definition --------- Co-authored-by: hernaldo.urbina <hernaldo.urbina@odysseusinc.com> Co-authored-by: oleg-odysseus <oleg.himself@gmail.com>
1 parent d4dc8a5 commit d23b434

15 files changed

+638
-81
lines changed

‎src/main/java/org/ohdsi/webapi/Constants.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -82,6 +82,7 @@ interface Params {
8282
String EXECUTABLE_FILE_NAME = "executableFilename";
8383
String GENERATION_ID = "generation_id";
8484
String DESIGN_HASH = "design_hash";
85+
String DEMOGRAPHIC_STATS = "demographic_stats";
8586
}
8687

8788
interface Variables {

‎src/main/java/org/ohdsi/webapi/cohortcharacterization/GenerateLocalCohortTasklet.java‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package org.ohdsi.webapi.cohortcharacterization;
22

33
import org.ohdsi.webapi.cohortdefinition.CohortDefinition;
4+
import org.ohdsi.webapi.cohortdefinition.CohortDefinitionDetails;
45
import org.ohdsi.webapi.cohortdefinition.CohortGenerationRequestBuilder;
56
import org.ohdsi.webapi.cohortdefinition.CohortGenerationUtils;
67
import org.ohdsi.webapi.generationcache.GenerationCacheHelper;
@@ -32,6 +33,7 @@
3233

3334
import static org.ohdsi.webapi.Constants.Params.SOURCE_ID;
3435
import static org.ohdsi.webapi.Constants.Params.TARGET_TABLE;
36+
import static org.ohdsi.webapi.Constants.Params.DEMOGRAPHIC_STATS;
3537

3638
public class GenerateLocalCohortTasklet implements StoppableTasklet {
3739

@@ -89,14 +91,14 @@ public RepeatStatus execute(StepContribution contribution, ChunkContext chunkCon
8991
if (useAsyncCohortGeneration) {
9092
List<CompletableFuture> executions = cohortDefinitions.stream()
9193
.map(cd ->
92-
CompletableFuture.supplyAsync(() -> generateCohort(cd, source, resultSchema, targetTable),
94+
CompletableFuture.supplyAsync(() -> generateCohort(cd, source, resultSchema, targetTable),
9395
Executors.newSingleThreadExecutor()
9496
)
9597
).collect(Collectors.toList());
9698
CompletableFuture.allOf(executions.toArray(new CompletableFuture[]{})).join();
9799
} else {
98100
CompletableFuture.runAsync(() ->
99-
cohortDefinitions.stream().forEach(cd -> generateCohort(cd, source, resultSchema, targetTable)),
101+
cohortDefinitions.stream().forEach(cd -> generateCohort(cd, source, resultSchema, targetTable)),
100102
Executors.newSingleThreadExecutor()
101103
).join();
102104
}
@@ -113,8 +115,8 @@ private Object generateCohort(CohortDefinition cd, Source source, String resultS
113115
sessionId,
114116
resultSchema
115117
);
116-
117-
int designHash = this.generationCacheHelper.computeHash(cd.getDetails().getExpression());
118+
CohortDefinitionDetails details = cd.getDetails();
119+
int designHash = this.generationCacheHelper.computeHash(details.getExpression());
118120
CohortGenerationUtils.insertInclusionRules(cd, source, designHash, resultSchema, sessionId, cancelableJdbcTemplate);
119121

120122
try {

‎src/main/java/org/ohdsi/webapi/cohortcharacterization/converter/BaseCcDTOToCcEntityConverter.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import com.odysseusinc.arachne.commons.utils.ConverterUtils;
44
import org.apache.commons.lang3.StringUtils;
5-
import org.ohdsi.analysis.CohortMetadata;
65
import org.ohdsi.analysis.Utils;
76
import org.ohdsi.analysis.cohortcharacterization.design.CcResultType;
87
import org.ohdsi.webapi.cohortcharacterization.domain.CcStrataConceptSetEntity;
@@ -18,7 +17,6 @@
1817
import org.ohdsi.webapi.tag.domain.Tag;
1918
import org.springframework.beans.factory.annotation.Autowired;
2019

21-
import java.util.List;
2220
import java.util.Objects;
2321
import java.util.Set;
2422
import java.util.stream.Collectors;

‎src/main/java/org/ohdsi/webapi/cohortdefinition/CohortGenerationInfo.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,21 @@ public CohortGenerationInfo(CohortDefinition definition, Integer sourceId)
8484
@ManyToOne(fetch = FetchType.LAZY)
8585
@JoinColumn(name = "created_by_id")
8686
private UserEntity createdBy;
87+
88+
@Column(name = "cc_generate_id")
89+
private Long ccGenerateId;
90+
91+
// If true, then demographic has been selected.
92+
@Column(name = "is_demographic")
93+
private boolean isDemographic;
94+
95+
public boolean isDemographic() {
96+
return isDemographic;
97+
}
98+
99+
public void setIsDemographic(boolean isDemographic) {
100+
this.isDemographic = isDemographic;
101+
}
87102

88103
public CohortGenerationInfoId getId() {
89104
return id;
@@ -187,4 +202,13 @@ public void setCreatedBy(UserEntity createdBy) {
187202
public UserEntity getCreatedBy() {
188203
return createdBy;
189204
}
205+
206+
public Long getCcGenerateId() {
207+
return ccGenerateId;
208+
}
209+
210+
public void setCcGenerateId(Long ccGenerateId) {
211+
this.ccGenerateId = ccGenerateId;
212+
}
213+
190214
}

‎src/main/java/org/ohdsi/webapi/cohortdefinition/CohortGenerationRequest.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,8 @@ public class CohortGenerationRequest {
1111
private String targetSchema;
1212
private Integer targetId;
1313

14-
public CohortGenerationRequest(CohortExpression expression, Source source, String sessionId, Integer targetId, String targetSchema) {
14+
public CohortGenerationRequest(CohortExpression expression, Source source, String sessionId, Integer targetId,
15+
String targetSchema) {
1516

1617
this.expression = expression;
1718
this.source = source;

‎src/main/java/org/ohdsi/webapi/cohortdefinition/CohortGenerationUtils.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -70,7 +70,7 @@ public static String[] buildGenerationSql(CohortGenerationRequest request) {
7070
"results_database_schema.cohort_inclusion_stats",
7171
"results_database_schema.cohort_summary_stats",
7272
"results_database_schema.cohort_censor_stats",
73-
"results_database_schema.cohort_inclusion"
73+
"results_database_schema.cohort_inclusion"
7474
},
7575
new String[] {
7676
COHORT_CACHE,

‎src/main/java/org/ohdsi/webapi/cohortdefinition/GenerateCohortTasklet.java‎

Lines changed: 148 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -16,21 +16,39 @@
1616
package org.ohdsi.webapi.cohortdefinition;
1717

1818
import org.ohdsi.circe.helper.ResourceHelper;
19+
import org.ohdsi.cohortcharacterization.CCQueryBuilder;
20+
import org.ohdsi.sql.BigQuerySparkTranslate;
1921
import org.ohdsi.sql.SqlRender;
2022
import org.ohdsi.sql.SqlSplit;
2123
import org.ohdsi.sql.SqlTranslate;
24+
import org.ohdsi.webapi.cohortcharacterization.domain.CohortCharacterizationEntity;
2225
import org.ohdsi.webapi.common.generation.CancelableTasklet;
26+
import org.ohdsi.webapi.common.generation.GenerationUtils;
27+
import org.ohdsi.webapi.feanalysis.domain.FeAnalysisEntity;
28+
import org.ohdsi.webapi.feanalysis.repository.FeAnalysisEntityRepository;
2329
import org.ohdsi.webapi.generationcache.GenerationCacheHelper;
30+
import org.ohdsi.webapi.shiro.Entities.UserRepository;
2431
import org.ohdsi.webapi.source.Source;
2532
import org.ohdsi.webapi.source.SourceService;
2633
import org.ohdsi.webapi.util.CancelableJdbcTemplate;
2734
import org.ohdsi.webapi.util.SessionUtils;
35+
import org.ohdsi.webapi.util.SourceUtils;
2836
import org.slf4j.LoggerFactory;
2937
import org.springframework.batch.core.scope.context.ChunkContext;
3038
import org.springframework.batch.core.step.tasklet.StoppableTasklet;
39+
import org.springframework.beans.factory.annotation.Autowired;
3140
import org.springframework.transaction.support.TransactionTemplate;
3241

42+
import com.google.common.collect.ImmutableList;
43+
import com.odysseusinc.arachne.commons.types.DBMSType;
44+
45+
import java.sql.SQLException;
46+
import java.util.Arrays;
47+
import java.util.HashSet;
3348
import java.util.Map;
49+
import java.util.Set;
50+
import java.util.stream.Collectors;
51+
import java.util.stream.Stream;
3452

3553
import static org.ohdsi.webapi.Constants.Params.*;
3654

@@ -44,54 +62,151 @@ public class GenerateCohortTasklet extends CancelableTasklet implements Stoppabl
4462
private final GenerationCacheHelper generationCacheHelper;
4563
private final CohortDefinitionRepository cohortDefinitionRepository;
4664
private final SourceService sourceService;
65+
private final FeAnalysisEntityRepository feAnalysisRepository;
66+
67+
public GenerateCohortTasklet(final CancelableJdbcTemplate jdbcTemplate, final TransactionTemplate transactionTemplate,
68+
final GenerationCacheHelper generationCacheHelper,
69+
final CohortDefinitionRepository cohortDefinitionRepository, final SourceService sourceService) {
70+
super(LoggerFactory.getLogger(GenerateCohortTasklet.class), jdbcTemplate, transactionTemplate);
71+
this.generationCacheHelper = generationCacheHelper;
72+
this.cohortDefinitionRepository = cohortDefinitionRepository;
73+
this.sourceService = sourceService;
74+
this.feAnalysisRepository = null;
75+
}
4776

4877
public GenerateCohortTasklet(
4978
final CancelableJdbcTemplate jdbcTemplate,
5079
final TransactionTemplate transactionTemplate,
5180
final GenerationCacheHelper generationCacheHelper,
5281
final CohortDefinitionRepository cohortDefinitionRepository,
53-
final SourceService sourceService
82+
final SourceService sourceService, final FeAnalysisEntityRepository feAnalysisRepository
5483
) {
5584
super(LoggerFactory.getLogger(GenerateCohortTasklet.class), jdbcTemplate, transactionTemplate);
5685
this.generationCacheHelper = generationCacheHelper;
5786
this.cohortDefinitionRepository = cohortDefinitionRepository;
5887
this.sourceService = sourceService;
88+
this.feAnalysisRepository = feAnalysisRepository;
5989
}
6090

6191
@Override
6292
protected String[] prepareQueries(ChunkContext chunkContext, CancelableJdbcTemplate jdbcTemplate) {
93+
Map<String, Object> jobParams = chunkContext.getStepContext().getJobParameters();
94+
95+
String[] defaultQueries = prepareQueriesDefault(jobParams, jdbcTemplate);
96+
97+
Boolean demographicStat = jobParams.get(DEMOGRAPHIC_STATS) == null ? null
98+
: Boolean.valueOf((String) jobParams.get(DEMOGRAPHIC_STATS));
99+
100+
if (demographicStat != null && demographicStat.booleanValue()) {
101+
String[] demographicsQueries = prepareQueriesDemographic(chunkContext, jdbcTemplate);
102+
return Stream.concat(Arrays.stream(defaultQueries), Arrays.stream(demographicsQueries)).toArray(String[]::new);
103+
}
104+
105+
return defaultQueries;
106+
}
107+
108+
private String[] prepareQueriesDemographic(ChunkContext chunkContext, CancelableJdbcTemplate jdbcTemplate) {
109+
Map<String, Object> jobParams = chunkContext.getStepContext().getJobParameters();
110+
CohortCharacterizationEntity cohortCharacterization = new CohortCharacterizationEntity();
111+
112+
Integer cohortDefinitionId = Integer.valueOf(jobParams.get(COHORT_DEFINITION_ID).toString());
113+
CohortDefinition cohortDefinition = cohortDefinitionRepository.findOneWithDetail(cohortDefinitionId);
114+
115+
cohortCharacterization.setCohortDefinitions(new HashSet<>(Arrays.asList(cohortDefinition)));
116+
117+
// Get FE Analysis Demographic (Gender, Age, Race,)
118+
Set<FeAnalysisEntity> feAnalysis = feAnalysisRepository.findByListIds(Arrays.asList(70, 72, 74, 77));
119+
120+
// Set<CcFeAnalysisEntity> ccFeAnalysis = feAnalysis.stream().map(a -> {
121+
// CcFeAnalysisEntity ccA = new CcFeAnalysisEntity();
122+
// ccA.setCohortCharacterization(cohortCharacterization);
123+
// ccA.setFeatureAnalysis(a);
124+
// return ccA;
125+
// }).collect(Collectors.toSet());
126+
127+
cohortCharacterization.setFeatureAnalyses(feAnalysis);
128+
129+
final Long jobId = chunkContext.getStepContext().getStepExecution().getJobExecution().getId();
130+
131+
final Integer sourceId = Integer.valueOf(jobParams.get(SOURCE_ID).toString());
132+
final Source source = sourceService.findBySourceId(sourceId);
133+
134+
final String cohortTable = jobParams.get(TARGET_TABLE).toString();
135+
final String sessionId = jobParams.get(SESSION_ID).toString();
136+
137+
final String tempSchema = SourceUtils.getTempQualifier(source);
138+
139+
boolean includeAnnual = false;
140+
boolean includeTemporal = false;
141+
142+
CCQueryBuilder ccQueryBuilder = new CCQueryBuilder(cohortCharacterization, cohortTable, sessionId,
143+
SourceUtils.getCdmQualifier(source), SourceUtils.getResultsQualifier(source),
144+
SourceUtils.getVocabularyQualifier(source), tempSchema, jobId);
145+
String sql = ccQueryBuilder.build();
146+
147+
/*
148+
* There is an issue with temp tables on sql server: Temp tables scope is
149+
* session or stored procedure. To execute PreparedStatement sql server
150+
* uses stored procedure <i>sp_executesql</i> and this is the reason why
151+
* multiple PreparedStatements cannot share the same local temporary
152+
* table.
153+
*
154+
* On the other side, temp tables cannot be re-used in the same
155+
* PreparedStatement, e.g. temp table cannot be created, used, dropped and
156+
* created again in the same PreparedStatement because sql optimizator
157+
* detects object already exists and fails. When is required to re-use
158+
* temp table it should be separated to several PreparedStatements.
159+
*
160+
* An option to use global temp tables also doesn't work since such tables
161+
* can be not supported / disabled.
162+
*
163+
* Therefore, there are two ways: - either precisely group SQLs into
164+
* statements so that temp tables aren't re-used in a single statement, -
165+
* or use ‘permanent temporary tables’
166+
*
167+
* The second option looks better since such SQL could be exported and
168+
* executed manually, which is not the case with the first option.
169+
*/
170+
if (ImmutableList.of(DBMSType.MS_SQL_SERVER.getOhdsiDB(), DBMSType.PDW.getOhdsiDB())
171+
.contains(source.getSourceDialect())) {
172+
sql = sql.replaceAll("#", tempSchema + "." + sessionId + "_").replaceAll("tempdb\\.\\.", "");
173+
}
174+
if (source.getSourceDialect().equals("spark")) {
175+
try {
176+
sql = BigQuerySparkTranslate.sparkHandleInsert(sql, source.getSourceConnection());
177+
} catch (SQLException e) {
178+
e.printStackTrace();
179+
}
180+
}
181+
182+
final String translatedSql = SqlTranslate.translateSql(sql, source.getSourceDialect(), sessionId, tempSchema);
183+
return SqlSplit.splitSql(translatedSql);
184+
}
185+
186+
private String[] prepareQueriesDefault(Map<String, Object> jobParams, CancelableJdbcTemplate jdbcTemplate) {
187+
Integer cohortDefinitionId = Integer.valueOf(jobParams.get(COHORT_DEFINITION_ID).toString());
188+
Integer sourceId = Integer.parseInt(jobParams.get(SOURCE_ID).toString());
189+
String targetSchema = jobParams.get(TARGET_DATABASE_SCHEMA).toString();
190+
String sessionId = jobParams.getOrDefault(SESSION_ID, SessionUtils.sessionId()).toString();
191+
192+
CohortDefinition cohortDefinition = cohortDefinitionRepository.findOneWithDetail(cohortDefinitionId);
193+
Source source = sourceService.findBySourceId(sourceId);
194+
195+
CohortGenerationRequestBuilder generationRequestBuilder = new CohortGenerationRequestBuilder(sessionId,
196+
targetSchema);
197+
198+
int designHash = this.generationCacheHelper.computeHash(cohortDefinition.getDetails().getExpression());
199+
CohortGenerationUtils.insertInclusionRules(cohortDefinition, source, designHash, targetSchema, sessionId,
200+
jdbcTemplate);
201+
202+
GenerationCacheHelper.CacheResult res = generationCacheHelper.computeCacheIfAbsent(cohortDefinition, source,
203+
generationRequestBuilder,
204+
(resId, sqls) -> generationCacheHelper.runCancelableCohortGeneration(jdbcTemplate, stmtCancel, sqls));
63205

64-
Map<String, Object> jobParams = chunkContext.getStepContext().getJobParameters();
65-
66-
Integer cohortDefinitionId = Integer.valueOf(jobParams.get(COHORT_DEFINITION_ID).toString());
67-
Integer sourceId = Integer.parseInt(jobParams.get(SOURCE_ID).toString());
68-
String targetSchema = jobParams.get(TARGET_DATABASE_SCHEMA).toString();
69-
String sessionId = jobParams.getOrDefault(SESSION_ID, SessionUtils.sessionId()).toString();
70-
71-
CohortDefinition cohortDefinition = cohortDefinitionRepository.findOneWithDetail(cohortDefinitionId);
72-
Source source = sourceService.findBySourceId(sourceId);
73-
74-
CohortGenerationRequestBuilder generationRequestBuilder = new CohortGenerationRequestBuilder(
75-
sessionId,
76-
targetSchema
77-
);
78-
79-
int designHash = this.generationCacheHelper.computeHash(cohortDefinition.getDetails().getExpression());
80-
CohortGenerationUtils.insertInclusionRules(cohortDefinition, source, designHash, targetSchema, sessionId, jdbcTemplate);
81-
82-
GenerationCacheHelper.CacheResult res = generationCacheHelper.computeCacheIfAbsent(
83-
cohortDefinition,
84-
source,
85-
generationRequestBuilder,
86-
(resId, sqls) -> generationCacheHelper.runCancelableCohortGeneration(jdbcTemplate, stmtCancel, sqls)
87-
);
88-
89-
String sql = SqlRender.renderSql(
90-
copyGenerationIntoCohortTableSql,
91-
new String[]{ RESULTS_DATABASE_SCHEMA, COHORT_DEFINITION_ID, DESIGN_HASH },
92-
new String[]{ targetSchema, cohortDefinition.getId().toString(), res.getIdentifier().toString() }
93-
);
94-
sql = SqlTranslate.translateSql(sql, source.getSourceDialect());
95-
return SqlSplit.splitSql(sql);
206+
String sql = SqlRender.renderSql(copyGenerationIntoCohortTableSql,
207+
new String[] { RESULTS_DATABASE_SCHEMA, COHORT_DEFINITION_ID, DESIGN_HASH },
208+
new String[] { targetSchema, cohortDefinition.getId().toString(), res.getIdentifier().toString() });
209+
sql = SqlTranslate.translateSql(sql, source.getSourceDialect());
210+
return SqlSplit.splitSql(sql);
96211
}
97212
}

‎src/main/java/org/ohdsi/webapi/cohortdefinition/GenerationJobExecutionListener.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ public void afterJob(JobExecution je) {
8888
CohortGenerationInfo info = findBySourceId(df, sourceId);
8989
setExecutionDurationIfPossible(je, info);
9090
info.setStatus(GenerationStatus.COMPLETE);
91+
info.setCcGenerateId(je.getId());
9192

9293
if (je.getStatus() == BatchStatus.FAILED || je.getStatus() == BatchStatus.STOPPED) {
9394
info.setIsValid(false);

‎src/main/java/org/ohdsi/webapi/cohortdefinition/InclusionRuleReport.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717

1818
import java.util.List;
1919

20+
import org.ohdsi.webapi.cohortcharacterization.report.Report;
21+
2022
/**
2123
*
2224
* @author Chris Knoll <cknoll@ohdsi.org>
@@ -42,5 +44,10 @@ public static class InclusionRuleStatistic
4244
public Summary summary;
4345
public List<InclusionRuleStatistic> inclusionRuleStats;
4446
public String treemapData;
47+
public List<Report> demographicsStats;
48+
49+
public Float prevalenceThreshold = 0.01f;
50+
public Boolean showEmptyResults = false;
51+
public int count = 0;
4552

4653
}

‎src/main/java/org/ohdsi/webapi/cohortdefinition/converter/CohortGenerationInfoToCohortGenerationInfoDTOConverter.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@ public CohortGenerationInfoDTO convert(CohortGenerationInfo info) {
2222
dto.setStartTime(info.getStartTime());
2323
dto.setStatus(info.getStatus());
2424
dto.setIsValid(info.isIsValid());
25+
dto.setCcGenerateId(info.getCcGenerateId());
26+
dto.setIsDemographic(info.isDemographic());
2527

2628
return dto;
2729
}

0 commit comments

Comments
 (0)