Merge pull request #1467 from jamesagnew/dm-add-pre-fetch-for-expansion-of-large-valuesets

Performance enhancements for large ValueSet expansion.

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/dao/data/ITermValueSetConceptDao.java

@@ -40,8 +40,8 @@ public interface ITermValueSetConceptDao extends JpaRepository<TermValueSetConce
 	@Modifying
 	void deleteByTermValueSetId(@Param("pid") Long theValueSetId);
 
-	@Query("SELECT vsc from TermValueSetConcept vsc WHERE vsc.myValueSet.myId = :pid")
+	@Query("SELECT vsc from TermValueSetConcept vsc LEFT OUTER JOIN FETCH vsc.myDesignations WHERE vsc.myValueSet.myId = :pid")
-	Slice<TermValueSetConcept> findByTermValueSetId(Pageable thePage, @Param("pid") Long theValueSetId);
+	Slice<TermValueSetConcept> findByTermValueSetIdAndPreFetchDesignations(Pageable thePage, @Param("pid") Long theValueSetId);
 
 	@Query("SELECT vsc FROM TermValueSetConcept vsc WHERE vsc.myValueSet.myId = :pid AND vsc.mySystem = :system_url AND vsc.myCode = :codeval")
 	Optional<TermValueSetConcept> findByTermValueSetIdSystemAndCode(@Param("pid") Long theValueSetId, @Param("system_url") String theSystem, @Param("codeval") String theCode);

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/entity/TermValueSet.java

@@ -26,6 +26,7 @@ import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
+import org.hibernate.annotations.ColumnDefault;
 
 import javax.annotation.Nonnull;
 import javax.persistence.*;
@@ -69,6 +70,14 @@ public class TermValueSet implements Serializable {
 	@OneToMany(mappedBy = "myValueSet")
 	private List<TermValueSetConcept> myConcepts;
 
+	@Column(name = "TOTAL_CONCEPTS", nullable = false)
+	@ColumnDefault("0")
+	private Long myTotalConcepts;
+
+	@Column(name = "TOTAL_CONCEPT_DESIGNATIONS", nullable = false)
+	@ColumnDefault("0")
+	private Long myTotalConceptDesignations;
+
 	@Enumerated(EnumType.STRING)
 	@Column(name = "EXPANSION_STATUS", nullable = false, length = MAX_EXPANSION_STATUS_LENGTH)
 	private TermValueSetPreExpansionStatusEnum myExpansionStatus;
@@ -76,6 +85,8 @@ public class TermValueSet implements Serializable {
 	public TermValueSet() {
 		super();
 		myExpansionStatus = TermValueSetPreExpansionStatusEnum.NOT_EXPANDED;
+		myTotalConcepts = 0L;
+		myTotalConceptDesignations = 0L;
 	}
 
 	public Long getId() {
@@ -120,6 +131,48 @@ public class TermValueSet implements Serializable {
 		return myConcepts;
 	}
 
+	public Long getTotalConcepts() {
+		return myTotalConcepts;
+	}
+
+	public TermValueSet setTotalConcepts(Long theTotalConcepts) {
+		myTotalConcepts = theTotalConcepts;
+		return this;
+	}
+
+	public TermValueSet decrementTotalConcepts() {
+		if (myTotalConcepts > 0) {
+			myTotalConcepts--;
+		}
+		return this;
+	}
+
+	public TermValueSet incrementTotalConcepts() {
+		myTotalConcepts++;
+		return this;
+	}
+
+	public Long getTotalConceptDesignations() {
+		return myTotalConceptDesignations;
+	}
+
+	public TermValueSet setTotalConceptDesignations(Long theTotalConceptDesignations) {
+		myTotalConceptDesignations = theTotalConceptDesignations;
+		return this;
+	}
+
+	public TermValueSet decrementTotalConceptDesignations() {
+		if (myTotalConceptDesignations > 0) {
+			myTotalConceptDesignations--;
+		}
+		return this;
+	}
+
+	public TermValueSet incrementTotalConceptDesignations() {
+		myTotalConceptDesignations++;
+		return this;
+	}
+
 	public TermValueSetPreExpansionStatusEnum getExpansionStatus() {
 		return myExpansionStatus;
 	}
@@ -157,6 +210,8 @@ public class TermValueSet implements Serializable {
 			.append("myResourcePid", myResourcePid)
 			.append("myName", myName)
 			.append(myConcepts != null ? ("myConcepts - size=" + myConcepts.size()) : ("myConcepts=(null)"))
+			.append("myTotalConcepts", myTotalConcepts)
+			.append("myTotalConceptDesignations", myTotalConceptDesignations)
 			.append("myExpansionStatus", myExpansionStatus)
 			.toString();
 	}

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/BaseHapiTerminologySvcImpl.java

@@ -28,6 +28,7 @@ import ca.uhn.fhir.jpa.dao.data.*;
 import ca.uhn.fhir.jpa.entity.*;
 import ca.uhn.fhir.jpa.entity.TermConceptParentChildLink.RelationshipTypeEnum;
 import ca.uhn.fhir.jpa.model.entity.ResourceTable;
+import ca.uhn.fhir.jpa.search.SearchCoordinatorSvcImpl;
 import ca.uhn.fhir.jpa.util.ScrollableResultsIterator;
 import ca.uhn.fhir.rest.api.server.RequestDetails;
 import ca.uhn.fhir.rest.server.exceptions.InternalErrorException;
@@ -372,13 +373,6 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 
 			myConceptMapDao.deleteTermConceptMapById(existingTermConceptMap.getId());
 			ourLog.info("Done deleting existing TermConceptMap[{}] and its children.", existingTermConceptMap.getId());
-
-			ourLog.info("Flushing...");
-			myConceptMapGroupElementTargetDao.flush();
-			myConceptMapGroupElementDao.flush();
-			myConceptMapGroupDao.flush();
-			myConceptMapDao.flush();
-			ourLog.info("Done flushing.");
 		}
 	}
 
@@ -400,12 +394,6 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 			myValueSetConceptDao.deleteByTermValueSetId(existingTermValueSet.getId());
 			myValueSetDao.deleteByTermValueSetId(existingTermValueSet.getId());
 			ourLog.info("Done deleting existing TermValueSet[{}] and its children.", existingTermValueSet.getId());
-
-			ourLog.info("Flushing...");
-			myValueSetConceptDesignationDao.flush();
-			myValueSetConceptDao.flush();
-			myValueSetDao.flush();
-			ourLog.info("Done flushing.");
 		}
 	}
 
@@ -524,7 +512,7 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 	}
 
 	private void populateExpansionComponent(ValueSet.ValueSetExpansionComponent theExpansionComponent, TermValueSet theTermValueSet, int theOffset, int theCount) {
-		int total = myValueSetConceptDao.countByTermValueSetId(theTermValueSet.getId());
+		int total = theTermValueSet.getTotalConcepts().intValue();
 		theExpansionComponent.setTotal(total);
 		theExpansionComponent.setOffset(theOffset);
 		theExpansionComponent.addParameter().setName("offset").setValue(new IntegerType(theOffset));
@@ -539,13 +527,11 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 
 	private void expandConcepts(ValueSet.ValueSetExpansionComponent theExpansionComponent, TermValueSet theTermValueSet, int theOffset, int theCount) {
 		int conceptsExpanded = 0;
-		for (int i = theOffset; i < (theOffset + theCount); i++) {
+		int toIndex = theOffset + theCount;
-			final int page = i;
+		Slice<TermValueSetConcept> slice = myValueSetConceptDao.findByTermValueSetIdAndPreFetchDesignations(SearchCoordinatorSvcImpl.toPage(theOffset, toIndex), theTermValueSet.getId());
-			Supplier<Slice<TermValueSetConcept>> loader = () -> myValueSetConceptDao.findByTermValueSetId(PageRequest.of(page, 1), theTermValueSet.getId());
-
-			Slice<TermValueSetConcept> slice = loader.get();
 		if (!slice.hasContent()) {
-				break;
+			logConceptsExpanded(theTermValueSet, conceptsExpanded);
+			return;
 		}
 
 		for (TermValueSetConcept concept : slice.getContent()) {
@@ -558,33 +544,22 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 			expandDesignations(theTermValueSet, concept, containsComponent);
 
 			if (++conceptsExpanded % 250 == 0) {
-					ourLog.info("Have expanded {} concepts in ValueSet[{}]", conceptsExpanded, theTermValueSet.getUrl());
+				logConceptsExpanded(theTermValueSet, conceptsExpanded);
 			}
 		}
 
-			if (!slice.hasNext()) {
+		logConceptsExpanded(theTermValueSet, conceptsExpanded);
-				break;
-			}
 	}
 
-		if (conceptsExpanded > 0) {
+	private void logConceptsExpanded(TermValueSet theTermValueSet, int theConceptsExpanded) {
-			ourLog.info("Have expanded {} concepts in ValueSet[{}]", conceptsExpanded, theTermValueSet.getUrl());
+		if (theConceptsExpanded > 0) {
+			ourLog.info("Have expanded {} concepts in ValueSet[{}]", theConceptsExpanded, theTermValueSet.getUrl());
 		}
 	}
 
 	private void expandDesignations(TermValueSet theValueSet, TermValueSetConcept theConcept, ValueSet.ValueSetExpansionContainsComponent theContainsComponent) {
 		int designationsExpanded = 0;
-		int index = 0;
+		for (TermValueSetConceptDesignation designation : theConcept.getDesignations()) {
-		while (true) {
-			final int page = index++;
-			Supplier<Slice<TermValueSetConceptDesignation>> loader = () -> myValueSetConceptDesignationDao.findByTermValueSetConceptId(PageRequest.of(page, 1000), theConcept.getId());
-
-			Slice<TermValueSetConceptDesignation> slice = loader.get();
-			if (!slice.hasContent()) {
-				break;
-			}
-
-			for (TermValueSetConceptDesignation designation : slice.getContent()) {
 			ValueSet.ConceptReferenceDesignationComponent designationComponent = theContainsComponent.addDesignation();
 			designationComponent.setLanguage(designation.getLanguage());
 			designationComponent.setUse(new Coding(
@@ -594,17 +569,16 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 			designationComponent.setValue(designation.getValue());
 
 			if (++designationsExpanded % 250 == 0) {
-					ourLog.info("Have expanded {} designations for Concept[{}|{}] in ValueSet[{}]", designationsExpanded, theConcept.getSystem(), theConcept.getCode(), theValueSet.getUrl());
+				logDesignationsExpanded(theValueSet, theConcept, designationsExpanded);
 			}
 		}
 
-			if (!slice.hasNext()) {
+		logDesignationsExpanded(theValueSet, theConcept, designationsExpanded);
-				break;
-			}
 	}
 
-		if (designationsExpanded > 0) {
+	private void logDesignationsExpanded(TermValueSet theValueSet, TermValueSetConcept theConcept, int theDesignationsExpanded) {
-			ourLog.info("Have expanded {} designations for Concept[{}|{}] in ValueSet[{}]", designationsExpanded, theConcept.getSystem(), theConcept.getCode(), theValueSet.getUrl());
+		if (theDesignationsExpanded > 0) {
+			ourLog.info("Have expanded {} designations for Concept[{}|{}] in ValueSet[{}]", theDesignationsExpanded, theConcept.getSystem(), theConcept.getCode(), theValueSet.getUrl());
 		}
 	}
 
@@ -1874,7 +1848,7 @@ public abstract class BaseHapiTerminologySvcImpl implements IHapiTerminologySvc,
 					TermValueSet refreshedValueSetToExpand = myValueSetDao.findById(valueSetToExpand.getId()).get();
 					return getValueSetFromResourceTable(refreshedValueSetToExpand.getResource());
 				});
-				expandValueSet(valueSet, new ValueSetConceptAccumulator(valueSetToExpand, myValueSetConceptDao, myValueSetConceptDesignationDao));
+				expandValueSet(valueSet, new ValueSetConceptAccumulator(valueSetToExpand, myValueSetDao, myValueSetConceptDao, myValueSetConceptDesignationDao));
 
 				// We are done with this ValueSet.
 				txTemplate.execute(t -> {

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/term/ValueSetConceptAccumulator.java

@@ -22,6 +22,7 @@ package ca.uhn.fhir.jpa.term;
 
 import ca.uhn.fhir.jpa.dao.data.ITermValueSetConceptDao;
 import ca.uhn.fhir.jpa.dao.data.ITermValueSetConceptDesignationDao;
+import ca.uhn.fhir.jpa.dao.data.ITermValueSetDao;
 import ca.uhn.fhir.jpa.entity.TermConceptDesignation;
 import ca.uhn.fhir.jpa.entity.TermValueSet;
 import ca.uhn.fhir.jpa.entity.TermValueSetConcept;
@@ -38,13 +39,15 @@ public class ValueSetConceptAccumulator implements IValueSetConceptAccumulator {
 	private static final org.slf4j.Logger ourLog = org.slf4j.LoggerFactory.getLogger(ValueSetConceptAccumulator.class);
 
 	private TermValueSet myTermValueSet;
+	private ITermValueSetDao myValueSetDao;
 	private ITermValueSetConceptDao myValueSetConceptDao;
 	private ITermValueSetConceptDesignationDao myValueSetConceptDesignationDao;
 	private int myConceptsSaved;
 	private int myDesignationsSaved;
 
-	public ValueSetConceptAccumulator(@Nonnull TermValueSet theTermValueSet, @Nonnull ITermValueSetConceptDao theValueSetConceptDao, @Nonnull ITermValueSetConceptDesignationDao theValueSetConceptDesignationDao) {
+	public ValueSetConceptAccumulator(@Nonnull TermValueSet theTermValueSet, @Nonnull ITermValueSetDao theValueSetDao, @Nonnull ITermValueSetConceptDao theValueSetConceptDao, @Nonnull ITermValueSetConceptDesignationDao theValueSetConceptDesignationDao) {
 		myTermValueSet = theTermValueSet;
+		myValueSetDao = theValueSetDao;
 		myValueSetConceptDao = theValueSetConceptDao;
 		myValueSetConceptDesignationDao = theValueSetConceptDesignationDao;
 		myConceptsSaved = 0;
@@ -79,14 +82,12 @@ public class ValueSetConceptAccumulator implements IValueSetConceptAccumulator {
 			ourLog.info("Excluding [{}|{}] from ValueSet[{}]", concept.getSystem(), concept.getCode(), myTermValueSet.getUrl());
 			for (TermValueSetConceptDesignation designation : concept.getDesignations()) {
 				myValueSetConceptDesignationDao.deleteById(designation.getId());
+				myTermValueSet.decrementTotalConceptDesignations();
 			}
 			myValueSetConceptDao.deleteById(concept.getId());
+			myTermValueSet.decrementTotalConcepts();
+			myValueSetDao.save(myTermValueSet);
 			ourLog.info("Done excluding [{}|{}] from ValueSet[{}]", concept.getSystem(), concept.getCode(), myTermValueSet.getUrl());
-
-			ourLog.info("Flushing...");
-			myValueSetConceptDesignationDao.flush();
-			myValueSetConceptDao.flush();
-			ourLog.info("Done flushing.");
 		}
 	}
 
@@ -102,6 +103,7 @@ public class ValueSetConceptAccumulator implements IValueSetConceptAccumulator {
 			concept.setDisplay(theDisplay);
 		}
 		myValueSetConceptDao.save(concept);
+		myValueSetDao.save(myTermValueSet.incrementTotalConcepts());
 
 		if (myConceptsSaved++ % 250 == 0) { // TODO: DM 2019-08-23 - This message never appears in the log. Fix it!
 			ourLog.info("Have pre-expanded {} concepts in ValueSet[{}]", myConceptsSaved, myTermValueSet.getUrl());
@@ -126,6 +128,7 @@ public class ValueSetConceptAccumulator implements IValueSetConceptAccumulator {
 		}
 		designation.setValue(theDesignation.getValue());
 		myValueSetConceptDesignationDao.save(designation);
+		myValueSetDao.save(myTermValueSet.incrementTotalConceptDesignations());
 
 		if (myDesignationsSaved++ % 250 == 0) { // TODO: DM 2019-08-23 - This message never appears in the log. Fix it!
 			ourLog.info("Have pre-expanded {} designations for Concept[{}|{}] in ValueSet[{}]", myDesignationsSaved, theConcept.getSystem(), theConcept.getCode(), myTermValueSet.getUrl());

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/search/SearchCoordinatorSvcImplTest.java

@@ -312,6 +312,7 @@ public class SearchCoordinatorSvcImplTest {
 	public void testGetPage() {
 		Pageable page = SearchCoordinatorSvcImpl.toPage(50, 73);
 		assertEquals(50, page.getOffset());
+		assertEquals(23, page.getPageSize());
 	}
 
 	@Test

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/term/TerminologySvcImplR4Test.java

@@ -36,6 +36,7 @@ import java.util.Optional;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.empty;
 import static org.junit.Assert.*;
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.times;
@@ -643,9 +644,12 @@ public class TerminologySvcImplR4Test extends BaseJpaR4Test {
 		ValueSet expandedValueSet = myTermSvc.expandValueSet(valueSet, myDaoConfig.getPreExpandValueSetsDefaultOffsetExperimental(), myDaoConfig.getPreExpandValueSetsDefaultCountExperimental());
 		ourLog.info("Expanded ValueSet:\n" + myFhirCtx.newJsonParser().setPrettyPrint(true).encodeResourceToString(expandedValueSet));
 
-		Long termValueSetId = myTermValueSetDao.findByResourcePid(myExtensionalVsIdOnResourceTable).get().getId();
+		TermValueSet termValueSet = myTermValueSetDao.findByResourcePid(myExtensionalVsIdOnResourceTable).get();
+		Long termValueSetId = termValueSet.getId();
 		assertEquals(3, myTermValueSetConceptDesignationDao.countByTermValueSetId(termValueSetId).intValue());
+		assertEquals(3, termValueSet.getTotalConceptDesignations().intValue());
 		assertEquals(24, myTermValueSetConceptDao.countByTermValueSetId(termValueSetId).intValue());
+		assertEquals(24, termValueSet.getTotalConcepts().intValue());
 
 		new TransactionTemplate(myTxManager).execute(new TransactionCallbackWithoutResult() {
 			@Override
@@ -677,9 +681,12 @@ public class TerminologySvcImplR4Test extends BaseJpaR4Test {
 		ValueSet expandedValueSet = myTermSvc.expandValueSet(valueSet, myDaoConfig.getPreExpandValueSetsDefaultOffsetExperimental(), myDaoConfig.getPreExpandValueSetsDefaultCountExperimental());
 		ourLog.info("Expanded ValueSet:\n" + myFhirCtx.newJsonParser().setPrettyPrint(true).encodeResourceToString(expandedValueSet));
 
-		Long termValueSetId = myTermValueSetDao.findByResourcePid(myExtensionalVsIdOnResourceTable).get().getId();
+		TermValueSet termValueSet = myTermValueSetDao.findByResourcePid(myExtensionalVsIdOnResourceTable).get();
+		Long termValueSetId = termValueSet.getId();
 		assertEquals(3, myTermValueSetConceptDesignationDao.countByTermValueSetId(termValueSetId).intValue());
+		assertEquals(3, termValueSet.getTotalConceptDesignations().intValue());
 		assertEquals(24, myTermValueSetConceptDao.countByTermValueSetId(termValueSetId).intValue());
+		assertEquals(24, termValueSet.getTotalConcepts().intValue());
 
 		new TransactionTemplate(myTxManager).execute(new TransactionCallbackWithoutResult() {
 			@Override
@@ -741,9 +748,17 @@ public class TerminologySvcImplR4Test extends BaseJpaR4Test {
 
 		myTermSvc.preExpandDeferredValueSetsToTerminologyTables();
 
+		myCaptureQueriesListener.clear();
+
 		ValueSet expandedValueSet = myTermSvc.expandValueSet(valueSet, myDaoConfig.getPreExpandValueSetsDefaultOffsetExperimental(), myDaoConfig.getPreExpandValueSetsDefaultCountExperimental());
 		ourLog.info("Expanded ValueSet:\n" + myFhirCtx.newJsonParser().setPrettyPrint(true).encodeResourceToString(expandedValueSet));
 
+		myCaptureQueriesListener.logSelectQueriesForCurrentThread();
+		assertEquals(3, myCaptureQueriesListener.getSelectQueriesForCurrentThread().size());
+		assertThat(myCaptureQueriesListener.getUpdateQueriesForCurrentThread(), empty());
+		assertThat(myCaptureQueriesListener.getInsertQueriesForCurrentThread(), empty());
+		assertThat(myCaptureQueriesListener.getDeleteQueriesForCurrentThread(), empty());
+
 		assertEquals(codeSystem.getConcept().size(), expandedValueSet.getExpansion().getTotal());
 		assertEquals(myDaoConfig.getPreExpandValueSetsDefaultOffsetExperimental(), expandedValueSet.getExpansion().getOffset());
 		assertEquals(2, expandedValueSet.getExpansion().getParameter().size());

hapi-fhir-jpaserver-migrate/src/main/java/ca/uhn/fhir/jpa/migrate/tasks/HapiFhirJpaMigrationTasks.java

@@ -87,6 +87,12 @@ public class HapiFhirJpaMigrationTasks extends BaseMigrationTasks<VersionEnum> {
 			.addForeignKey("FK_TRM_VSCD_VS_PID")
 			.toColumn("VALUESET_PID")
 			.references("TRM_VALUESET", "PID");
+
+		// TermValueSet
+		version.startSectionWithMessage("Processing table: TRM_VALUESET");
+		Builder.BuilderWithTableName termValueSetTable = version.onTable("TRM_VALUESET");
+		termValueSetTable.addColumn("TOTAL_CONCEPTS").nonNullable().type(BaseTableColumnTypeTask.ColumnTypeEnum.LONG);
+		termValueSetTable.addColumn("TOTAL_CONCEPT_DESIGNATIONS").nonNullable().type(BaseTableColumnTypeTask.ColumnTypeEnum.LONG);
 	}
 
 	protected void init400() {