Update test.

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch/processors/GoldenResourceAnnotatingProcessor.java

@@ -24,13 +24,14 @@ import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.context.RuntimeSearchParam;
 import ca.uhn.fhir.fhirpath.IFhirPath;
 import ca.uhn.fhir.jpa.batch.log.Logs;
+import ca.uhn.fhir.jpa.bulk.job.BulkExportJobConfig;
 import ca.uhn.fhir.jpa.dao.mdm.MdmExpansionCacheSvc;
 import ca.uhn.fhir.util.ExtensionUtil;
 import ca.uhn.fhir.util.SearchParameterUtil;
-import org.hl7.fhir.instance.model.api.IBase;
+import org.apache.commons.lang3.StringUtils;
+import org.hl7.fhir.instance.model.api.IBaseExtension;
 import org.hl7.fhir.instance.model.api.IBaseReference;
 import org.hl7.fhir.instance.model.api.IBaseResource;
-import org.hl7.fhir.instance.model.api.IIdType;
 import org.slf4j.Logger;
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -55,7 +56,11 @@ public class GoldenResourceAnnotatingProcessor implements ItemProcessor<List<IBa
 	@Autowired
 	private MdmExpansionCacheSvc myMdmExpansionCacheSvc;
 
+	@Value("#{jobParameters['" + BulkExportJobConfig.EXPAND_MDM_PARAMETER+ "'] ?: false}")
+	private boolean myMdmEnabled;
+
 	private RuntimeSearchParam myRuntimeSearchParam;
+	private String myPatientFhirPath;
 	private IFhirPath myFhirPath;
 
 
@@ -63,25 +68,49 @@ public class GoldenResourceAnnotatingProcessor implements ItemProcessor<List<IBa
 
 	@Override
 	public List<IBaseResource> process(List<IBaseResource> theIBaseResources) throws Exception {
-		if (myRuntimeSearchParam == null) {
+		if (myMdmEnabled) {
-			myRuntimeSearchParam = SearchParameterUtil.getPatientSearchParamForResourceType(myContext, myResourceType);
+			if (myRuntimeSearchParam == null) {
+				myRuntimeSearchParam = SearchParameterUtil.getPatientSearchParamForResourceType(myContext, myResourceType);
+			}
+			String path = runtimeSearchParamFhirPath();
+			theIBaseResources.forEach(iBaseResource -> annotateClinicalResourceWithRelatedGoldenResourcePatient(path, iBaseResource));
 		}
-		String path = runtimeSearchParamFhirPath();
+
-		theIBaseResources
-			.forEach(iBaseResource -> annotateResourceWithRelatedGoldenResourcePatient(path, iBaseResource));
 		return theIBaseResources;
 	}
 
-	private void annotateResourceWithRelatedGoldenResourcePatient(String path, IBaseResource iBaseResource) {
+	private void annotateClinicalResourceWithRelatedGoldenResourcePatient(String path, IBaseResource iBaseResource) {
-		Optional<IBaseReference> evaluate = getFhirParser().evaluateFirst(iBaseResource, path, IBaseReference.class);
+		Optional<String> patientReference = getPatientReference(path, iBaseResource);
-		if (evaluate.isPresent()) {
+		if (patientReference.isPresent()) {
-			String sourceResourceId = evaluate.get().getReferenceElement().getIdPart();
+			addGoldenResourceExtension(iBaseResource, patientReference.get());
-			ExtensionUtil.setExtension(myContext, iBaseResource, ASSOCIATED_GOLDEN_RESOURCE_EXTENSION_URL, myMdmExpansionCacheSvc.getGoldenResourceId(sourceResourceId));
 		} else {
-			ourLog.warn("Failed to find the patient compartment information for resource {}", iBaseResource);
+			ourLog.warn("Failed to find the patient reference information for resource {}", iBaseResource);
 		}
 	}
 
+	private Optional<String> getPatientReference(String path, IBaseResource iBaseResource) {
+		//In the case of patient, we will just use the raw ID.
+		if (myResourceType.equalsIgnoreCase("Patient")) {
+			return Optional.of(iBaseResource.getIdElement().getIdPart());
+		//Otherwise, we will perform evaluation of the fhirPath.
+		} else {
+			Optional<IBaseReference> optionalReference = getFhirParser().evaluateFirst(iBaseResource, path, IBaseReference.class);
+			return optionalReference.map(theIBaseReference -> theIBaseReference.getReferenceElement().getIdPart());
+		}
+	}
+
+	private void addGoldenResourceExtension(IBaseResource iBaseResource, String sourceResourceId) {
+		String goldenResourceId = myMdmExpansionCacheSvc.getGoldenResourceId(sourceResourceId);
+		if (!StringUtils.isBlank(goldenResourceId)) {
+			IBaseExtension<?, ?> extension = ExtensionUtil.getOrCreateExtension(iBaseResource, ASSOCIATED_GOLDEN_RESOURCE_EXTENSION_URL);
+			ExtensionUtil.setExtension(myContext, extension, "reference", prefixPatient(goldenResourceId));
+		}
+	}
+
+	private String prefixPatient(String theResourceId) {
+		return "Patient/" + theResourceId;
+	}
+
 	private IFhirPath getFhirParser() {
 		if (myFhirPath == null) {
 			myFhirPath = myContext.newFhirPath();
@@ -90,9 +119,15 @@ public class GoldenResourceAnnotatingProcessor implements ItemProcessor<List<IBa
 	}
 
 	private String runtimeSearchParamFhirPath() {
-		if (myRuntimeSearchParam == null) {
+		if (myPatientFhirPath == null) {
 			myRuntimeSearchParam = SearchParameterUtil.getPatientSearchParamForResourceType(myContext, myResourceType);
+			myPatientFhirPath = myRuntimeSearchParam.getPath();
+			//Yes this is a stupid hack, but by default this runtime search param will return stuff like
+			// Observation.subject.where(resolve() is Patient) which unfortunately our FHIRpath evaluator doesn't play nicely with.
+			if (myPatientFhirPath.contains(".where")) {
+				myPatientFhirPath = myPatientFhirPath.substring(0, myPatientFhirPath.indexOf(".where"));
+			}
 		}
-		return myRuntimeSearchParam.getPath();
+		return myPatientFhirPath;
 	}
 }

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/bulk/job/BulkExportJobConfig.java

@@ -80,7 +80,7 @@ public class BulkExportJobConfig {
 
 	@Bean
 	@Lazy
-	@StepScope
+	@JobScope
 	public MdmExpansionCacheSvc mdmExpansionCacheSvc() {
 		return new MdmExpansionCacheSvc();
 	}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/bulk/job/GroupBulkItemReader.java

@@ -45,7 +45,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -123,7 +122,6 @@ public class GroupBulkItemReader extends BaseBulkItemReader implements ItemReade
 			IBaseResource group = myDaoRegistry.getResourceDao("Group").read(new IdDt(myGroupId));
 			Long pidOrNull = myIdHelperService.getPidOrNull(group);
 			List<List<Long>> lists = myMdmLinkDao.expandPidsFromGroupPidGivenMatchResult(pidOrNull, MdmMatchResultEnum.MATCH);
-
 			lists.forEach(patientPidsToExport::addAll);
 		}
 		List<ResourcePersistentId> resourcePersistentIds = patientPidsToExport
@@ -161,11 +159,6 @@ public class GroupBulkItemReader extends BaseBulkItemReader implements ItemReade
 		if (myMdmEnabled) {
 			List<List<Long>> goldenPidTargetPidTuples = myMdmLinkDao.expandPidsFromGroupPidGivenMatchResult(pidOrNull, MdmMatchResultEnum.MATCH);
 			//Now lets translate these pids into resource IDs
-
-			System.out.println("ZOOPER");
-			goldenPidTargetPidTuples.stream()
-				.map(Object::toString)
-				.forEach(list -> System.out.println(String.join(", ", list)));
 			Set<Long> uniquePids = new HashSet<>();
 			goldenPidTargetPidTuples.forEach(uniquePids::addAll);
 			Map<Long, Optional<String>> pidToForcedIdMap = myIdHelperService.translatePidsToForcedIds(uniquePids);
@@ -180,22 +173,8 @@ public class GroupBulkItemReader extends BaseBulkItemReader implements ItemReade
 				}
 				goldenResourceToSourcePidMap.get(goldenPid).add(sourcePid);
 			}
-			Map<String, String> sourceResourceIdToGoldenResourceIdMap = new HashMap<>();
+			populateMdmResourceCacheIfNeeded(goldenResourceToSourcePidMap);
 
-			goldenResourceToSourcePidMap.entrySet()
-				.forEach(entry -> {
-					Long key = entry.getKey();
-					String goldenResourceId = myIdHelperService.translatePidIdToForcedId(new ResourcePersistentId(key)).orElse(key.toString());
-
-					Map<Long, Optional<String>> pidsToForcedIds = myIdHelperService.translatePidsToForcedIds(entry.getValue());
-					Set<String> sourceResourceIds = pidsToForcedIds.entrySet().stream()
-						.map(ent -> ent.getValue().isPresent() ? ent.getValue().get() : ent.getKey().toString())
-						.collect(Collectors.toSet());
-
-					sourceResourceIds
-						.forEach(sourceResourceId -> sourceResourceIdToGoldenResourceIdMap.put(sourceResourceId, goldenResourceId));
-				});
-			myMdmExpansionCacheSvc.setCacheContents(sourceResourceIdToGoldenResourceIdMap);
 
 			//If the result of the translation is an empty optional, it means there is no forced id, and we can use the PID as the resource ID.
 			Set<String> resolvedResourceIds = pidToForcedIdMap.entrySet().stream()
@@ -212,6 +191,28 @@ public class GroupBulkItemReader extends BaseBulkItemReader implements ItemReade
 		return expandedIds;
 	}
 
+	private void populateMdmResourceCacheIfNeeded(Map<Long, Set<Long>> goldenResourceToSourcePidMap) {
+		if (myMdmExpansionCacheSvc.hasBeenPopulated()) {
+			return;
+		}
+		Map<String, String> sourceResourceIdToGoldenResourceIdMap = new HashMap<>();
+
+		goldenResourceToSourcePidMap.entrySet()
+			.forEach(entry -> {
+				Long key = entry.getKey();
+				String goldenResourceId = myIdHelperService.translatePidIdToForcedId(new ResourcePersistentId(key)).orElse(key.toString());
+
+				Map<Long, Optional<String>> pidsToForcedIds = myIdHelperService.translatePidsToForcedIds(entry.getValue());
+				Set<String> sourceResourceIds = pidsToForcedIds.entrySet().stream()
+					.map(ent -> ent.getValue().isPresent() ? ent.getValue().get() : ent.getKey().toString())
+					.collect(Collectors.toSet());
+
+				sourceResourceIds
+					.forEach(sourceResourceId -> sourceResourceIdToGoldenResourceIdMap.put(sourceResourceId, goldenResourceId));
+			});
+		myMdmExpansionCacheSvc.setCacheContents(sourceResourceIdToGoldenResourceIdMap);
+	}
+
 	private void queryResourceTypeWithReferencesToPatients(Set<ResourcePersistentId> myReadPids, List<String> idChunk) {
 		//Build SP map
 		//First, inject the _typeFilters and _since from the export job

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/mdm/MdmExpansionCacheSvc.java

@@ -2,15 +2,15 @@ package ca.uhn.fhir.jpa.dao.mdm;
 
 import org.slf4j.Logger;
 
-import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
 public class MdmExpansionCacheSvc {
 	private static final Logger ourLog = getLogger(MdmExpansionCacheSvc.class);
 
-	private Map<String, String> mySourceToGoldenIdCache = new HashMap<>();
+	private ConcurrentHashMap<String, String> mySourceToGoldenIdCache = new ConcurrentHashMap<>();
 
 	public String getGoldenResourceId(String theSourceId) {
 		ourLog.info(buildLog("About to lookup cached resource ID " + theSourceId, true));
@@ -32,6 +32,12 @@ public class MdmExpansionCacheSvc {
 	}
 
 	public void setCacheContents(Map<String, String> theSourceResourceIdToGoldenResourceIdMap) {
-		this.mySourceToGoldenIdCache = theSourceResourceIdToGoldenResourceIdMap;
+		if (mySourceToGoldenIdCache.isEmpty()) {
+			this.mySourceToGoldenIdCache.putAll(theSourceResourceIdToGoldenResourceIdMap);
+		}
+	}
+
+	public boolean hasBeenPopulated() {
+		return !mySourceToGoldenIdCache.isEmpty();
 	}
 }

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/bulk/BulkDataExportSvcImplR4Test.java

@@ -6,6 +6,7 @@ import ca.uhn.fhir.jpa.api.config.DaoConfig;
 import ca.uhn.fhir.jpa.api.model.DaoMethodOutcome;
 import ca.uhn.fhir.jpa.batch.BatchJobsConfig;
 import ca.uhn.fhir.jpa.batch.api.IBatchJobSubmitter;
+import ca.uhn.fhir.jpa.batch.processors.GoldenResourceAnnotatingProcessor;
 import ca.uhn.fhir.jpa.bulk.api.BulkDataExportOptions;
 import ca.uhn.fhir.jpa.bulk.api.IBulkDataExportSvc;
 import ca.uhn.fhir.jpa.bulk.job.BulkExportJobParametersBuilder;
@@ -646,7 +647,7 @@ public class BulkDataExportSvcImplR4Test extends BaseJpaR4Test {
 		// Create a bulk job
 		BulkDataExportOptions bulkDataExportOptions = new BulkDataExportOptions();
 		bulkDataExportOptions.setOutputFormat(null);
-		bulkDataExportOptions.setResourceTypes(Sets.newHashSet("Immunization"));
+		bulkDataExportOptions.setResourceTypes(Sets.newHashSet("Immunization", "Observation", "Patient"));
 		bulkDataExportOptions.setSince(null);
 		bulkDataExportOptions.setFilters(null);
 		bulkDataExportOptions.setGroupId(myPatientGroupId);
@@ -654,22 +655,28 @@ public class BulkDataExportSvcImplR4Test extends BaseJpaR4Test {
 		bulkDataExportOptions.setExportStyle(BulkDataExportOptions.ExportStyle.GROUP);
 		IBulkDataExportSvc.JobInfo jobDetails = myBulkDataExportSvc.submitJob(bulkDataExportOptions);
 
-
 		myBulkDataExportSvc.buildExportFiles();
 		awaitAllBulkJobCompletions();
 
 		IBulkDataExportSvc.JobInfo jobInfo = myBulkDataExportSvc.getJobInfoOrThrowResourceNotFound(jobDetails.getJobId());
 
 		assertThat(jobInfo.getStatus(), equalTo(BulkJobStatusEnum.COMPLETE));
-		assertThat(jobInfo.getFiles().size(), equalTo(1));
+		assertThat(jobInfo.getFiles().size(), equalTo(3));
 		assertThat(jobInfo.getFiles().get(0).getResourceType(), is(equalTo("Immunization")));
 
 		// Iterate over the files
+
 		String nextContents = getBinaryContents(jobInfo, 0);
-
 		assertThat(jobInfo.getFiles().get(0).getResourceType(), is(equalTo("Immunization")));
+		assertThat(nextContents, is(containsString(GoldenResourceAnnotatingProcessor.ASSOCIATED_GOLDEN_RESOURCE_EXTENSION_URL)));
 
-		assertThat(nextContents, is(containsString("subject_golden_resource")));
+		nextContents = getBinaryContents(jobInfo, 1);
+		assertThat(jobInfo.getFiles().get(1).getResourceType(), is(equalTo("Observation")));
+		assertThat(nextContents, is(containsString(GoldenResourceAnnotatingProcessor.ASSOCIATED_GOLDEN_RESOURCE_EXTENSION_URL)));
+
+		nextContents = getBinaryContents(jobInfo, 2);
+		assertThat(jobInfo.getFiles().get(2).getResourceType(), is(equalTo("Patient")));
+		assertThat(nextContents, is(containsString(GoldenResourceAnnotatingProcessor.ASSOCIATED_GOLDEN_RESOURCE_EXTENSION_URL)));
 
 	}