Issue 3441 reindex query being ignored (#3451)

https://github.com/hapifhir/hapi-fhir/issues/3441

Co-authored-by: Tadgh <garygrantgraham@gmail.com>

hapi-fhir-docs/src/main/resources/ca/uhn/hapi/fhir/changelog/6_0_0/3441-reindex-query-being-ignored.yaml

@@ -0,0 +1,6 @@
+---
+type: fix
+issue: 3441
+jira: SMILE-3441
+title: "Reindexing jobs were not respected the passed in date range in SearchParams. We now take date these date ranges
+into account when running re-indexing jobs."

hapi-fhir-jpaserver-base/pom.xml

@@ -771,6 +771,14 @@
 					</execution>
 				</executions>
 			</plugin>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+				<configuration>
+					<source>15</source>
+					<target>15</target>
+				</configuration>
+			</plugin>
 		</plugins>
 		<resources>
 			<resource>

hapi-fhir-jpaserver-base/src/main/java/ca/uhn/fhir/jpa/batch/reader/BaseReverseCronologicalBatchPidReader.java

@@ -35,6 +35,7 @@ import ca.uhn.fhir.rest.api.Constants;
 import ca.uhn.fhir.rest.api.SortOrderEnum;
 import ca.uhn.fhir.rest.api.SortSpec;
 import ca.uhn.fhir.rest.param.DateRangeParam;
+import ca.uhn.fhir.rest.param.ParamPrefixEnum;
 import org.apache.commons.lang3.time.DateUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -155,11 +156,41 @@ public abstract class BaseReverseCronologicalBatchPidReader implements ItemReade
 
 	protected void addDateCountAndSortToSearch(ResourceSearch resourceSearch) {
 		SearchParameterMap map = resourceSearch.getSearchParameterMap();
-		map.setLastUpdated(new DateRangeParam().setUpperBoundInclusive(getCurrentHighThreshold()));
+		DateRangeParam rangeParam = getDateRangeParam(resourceSearch);
+		map.setLastUpdated(rangeParam);
 		map.setLoadSynchronousUpTo(myBatchSize);
 		map.setSort(new SortSpec(Constants.PARAM_LASTUPDATED, SortOrderEnum.DESC));
 	}
 
+	/**
+	 * Evaluates the passed in {@link ResourceSearch} to see if it contains a non-null {@link DateRangeParam}.
+	 *
+	 * If one such {@link DateRangeParam} exists, we use that to determine the upper and lower bounds for the returned
+	 * {@link DateRangeParam}. The {@link DateRangeParam#getUpperBound()} is compared to the
+	 * {@link BaseReverseCronologicalBatchPidReader#getCurrentHighThreshold()}, and the lower of the two date values
+	 * is used.
+	 *
+	 * If no {@link DateRangeParam} is set, we use the local {@link BaseReverseCronologicalBatchPidReader#getCurrentHighThreshold()}
+	 * to create a {@link DateRangeParam}.
+	 * @param resourceSearch The {@link ResourceSearch} to check.
+	 * @return {@link DateRangeParam}
+	 */
+	private DateRangeParam getDateRangeParam(ResourceSearch resourceSearch) {
+		DateRangeParam rangeParam = resourceSearch.getSearchParameterMap().getLastUpdated();
+		if (rangeParam != null) {
+			if (rangeParam.getUpperBound() == null) {
+				rangeParam.setUpperBoundInclusive(getCurrentHighThreshold());
+			} else {
+				Date theUpperBound = (getCurrentHighThreshold() == null || rangeParam.getUpperBound().getValue().before(getCurrentHighThreshold()))
+					? rangeParam.getUpperBound().getValue() : getCurrentHighThreshold();
+				rangeParam.setUpperBoundInclusive(theUpperBound);
+			}
+		} else {
+			rangeParam = new DateRangeParam().setUpperBoundInclusive(getCurrentHighThreshold());
+		}
+		return rangeParam;
+	}
+
 	@Override
 	public void open(ExecutionContext executionContext) throws ItemStreamException {
 		if (executionContext.containsKey(BatchConstants.CURRENT_URL_INDEX)) {

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/delete/job/ReindexJobTest.java

@@ -1,5 +1,6 @@
 package ca.uhn.fhir.jpa.delete.job;
 
+import ca.uhn.fhir.jpa.api.model.DaoMethodOutcome;
 import ca.uhn.fhir.jpa.batch.CommonBatchJobConfig;
 import ca.uhn.fhir.jpa.batch.api.IBatchJobSubmitter;
 import ca.uhn.fhir.jpa.batch.config.BatchConstants;
@@ -7,6 +8,8 @@ import ca.uhn.fhir.jpa.batch.job.MultiUrlJobParameterUtil;
 import ca.uhn.fhir.jpa.batch.reader.CronologicalBatchAllResourcePidReader;
 import ca.uhn.fhir.jpa.dao.r4.BaseJpaR4Test;
 import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
+import ca.uhn.fhir.model.api.TemporalPrecisionEnum;
+import ca.uhn.fhir.model.primitive.DateTimeDt;
 import ca.uhn.fhir.test.utilities.BatchJobHelper;
 import org.apache.commons.lang3.time.DateUtils;
 import org.hl7.fhir.instance.model.api.IIdType;
@@ -28,8 +31,10 @@ import java.util.List;
 import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
 import static org.hamcrest.Matchers.hasSize;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 public class ReindexJobTest extends BaseJpaR4Test {
 	private static final Logger ourLog = LoggerFactory.getLogger(ReindexJobTest.class);
@@ -81,6 +86,79 @@ public class ReindexJobTest extends BaseJpaR4Test {
 		assertEquals(obsFinalId.getIdPart(), alleleObservationIds.get(0));
 	}
 
+	private long generateAndReturnTimeGap() {
+		long start_time = System.currentTimeMillis();
+		sleepUntilTimeChanges();
+		long end_time = System.currentTimeMillis();
+		return end_time - start_time;
+	}
+
+	@Test
+	public void testReindexJobLastUpdatedFilter() throws Exception {
+		// Given
+		DaoMethodOutcome T1_Patient = myReindexTestHelper.createEyeColourPatient(true);
+		long timeGap1 = generateAndReturnTimeGap();
+		DaoMethodOutcome T3_Patient = myReindexTestHelper.createEyeColourPatient(true);
+		long timeGap2 = generateAndReturnTimeGap();
+		DaoMethodOutcome T6_Patient = myReindexTestHelper.createEyeColourPatient(true);
+
+		// Setup cutoff
+		Date firstPatientLastUpdated = T1_Patient.getResource().getMeta().getLastUpdated();
+		Date secondPatientLastUpdated = T3_Patient.getResource().getMeta().getLastUpdated();
+		Date T2_Date = DateUtils.addMilliseconds(firstPatientLastUpdated, (int) (timeGap1 / 2));
+		Date T4_Date = DateUtils.addMilliseconds(secondPatientLastUpdated, (int) (timeGap2 / 2));
+		ourLog.info("Older lastUpdated: {}", firstPatientLastUpdated);
+		ourLog.info("Newer lastUpdated: {}", secondPatientLastUpdated);
+		ourLog.info("Cutoff Lowerbound: {}", T2_Date);
+		ourLog.info("Cutoff Upperbound: {}", T4_Date);
+		assertTrue(T2_Date.after(firstPatientLastUpdated));
+		assertTrue(T2_Date.before(secondPatientLastUpdated));
+		assertTrue(T4_Date.after(secondPatientLastUpdated));
+
+		//Create our new SP.
+		myReindexTestHelper.createEyeColourSearchParameter();
+
+		//There exists 3 patients
+		assertEquals(3, myPatientDao.search(SearchParameterMap.newSynchronous()).size());
+		// The searchparam value is on the patient, but it hasn't been indexed yet, so the call to search for all with eye-colour returns 0
+		assertThat(myReindexTestHelper.getEyeColourPatientIds(), hasSize(0));
+
+		// Only reindex one of them
+		String T2_DateString = new DateTimeDt(T2_Date).setPrecision(TemporalPrecisionEnum.MILLI).getValueAsString();
+		String T4_DateString = new DateTimeDt(T4_Date).setPrecision(TemporalPrecisionEnum.MILLI).getValueAsString();
+		JobParameters T3_Patient_JobParams = MultiUrlJobParameterUtil.buildJobParameters("Patient?_lastUpdated=ge" +
+			T2_DateString + "&_lastUpdated=le" + T4_DateString);
+		JobParameters T1_Patient_JobParams = MultiUrlJobParameterUtil.buildJobParameters("Patient?_lastUpdated=le" + T2_DateString);
+		JobParameters T6_Patient_JobParams = MultiUrlJobParameterUtil.buildJobParameters("Patient?_lastUpdated=ge" + T4_DateString);
+
+		// execute
+		JobExecution jobExecution = myBatchJobSubmitter.runJob(myReindexJob, T3_Patient_JobParams);
+		myBatchJobHelper.awaitJobCompletion(jobExecution);
+
+		// Now one of them should be indexed for the eye colour SP
+		List<String> eyeColourPatientIds = myReindexTestHelper.getEyeColourPatientIds();
+		assertThat(eyeColourPatientIds, hasSize(1));
+		assertEquals(T3_Patient.getId().getIdPart(), eyeColourPatientIds.get(0));
+
+		// execute
+		JobExecution jobExecutionT1 = myBatchJobSubmitter.runJob(myReindexJob, T1_Patient_JobParams);
+		myBatchJobHelper.awaitJobCompletion(jobExecution);
+
+		// Now one of them should be indexed for the eye colour SP
+		eyeColourPatientIds = myReindexTestHelper.getEyeColourPatientIds();
+		assertThat(eyeColourPatientIds, hasSize(2));
+		assertThat(eyeColourPatientIds, hasItem(T3_Patient.getId().getIdPart()));
+
+		// execute
+		JobExecution jobExecutionT6 = myBatchJobSubmitter.runJob(myReindexJob, T6_Patient_JobParams);
+		myBatchJobHelper.awaitJobCompletion(jobExecution);
+
+		// Now one of them should be indexed for the eye colour SP
+		eyeColourPatientIds = myReindexTestHelper.getEyeColourPatientIds();
+		assertThat(eyeColourPatientIds, hasSize(3));
+		assertThat(eyeColourPatientIds, hasItem(T6_Patient.getId().getIdPart()));
+	}
+
 	@Test
 	public void testReindexEverythingJob() throws Exception {
 		// setup

hapi-fhir-jpaserver-base/src/test/java/ca/uhn/fhir/jpa/delete/job/ReindexTestHelper.java

@@ -3,6 +3,7 @@ package ca.uhn.fhir.jpa.delete.job;
 import ca.uhn.fhir.context.FhirContext;
 import ca.uhn.fhir.jpa.api.dao.DaoRegistry;
 import ca.uhn.fhir.jpa.api.dao.IFhirResourceDao;
+import ca.uhn.fhir.jpa.api.model.DaoMethodOutcome;
 import ca.uhn.fhir.jpa.searchparam.SearchParameterMap;
 import ca.uhn.fhir.rest.api.CacheControlDirective;
 import ca.uhn.fhir.rest.api.server.IBundleProvider;
@@ -16,6 +17,7 @@ import org.hl7.fhir.instance.model.api.IBaseResource;
 import org.hl7.fhir.instance.model.api.IIdType;
 import org.hl7.fhir.r4.model.Enumerations;
 import org.hl7.fhir.r4.model.Observation;
+import org.hl7.fhir.r4.model.Patient;
 import org.hl7.fhir.r4.model.SearchParameter;
 import org.hl7.fhir.r4.model.StringType;
 import org.slf4j.Logger;
@@ -26,15 +28,21 @@ import java.util.List;
 
 public class ReindexTestHelper {
 	public static final String ALLELE_EXTENSION_URL = "http://hl7.org/fhir/StructureDefinition/observation-geneticsAlleleName";
+	public static final String EYECOLOUR_EXTENSION_URL = "http://hl7.org/fhir/StructureDefinition/patient-eyeColour";
 	public static final String ALLELE_SP_CODE = "alleleName";
+	public static final String EYECOLOUR_SP_CODE= "eyecolour";
 	private static final Logger ourLog = LoggerFactory.getLogger(ReindexTestHelper.class);
 	private static final String TEST_ALLELE_VALUE = "HERC";
+	private static final String TEST_EYECOLOUR_VALUE = "blue";
 
 	private final FhirContext myFhirContext;
 	private final DaoRegistry myDaoRegistry;
 	private final ISearchParamRegistry mySearchParamRegistry;
 	private final IFhirResourceDao<SearchParameter> mySearchParameterDao;
 	private final IFhirResourceDao<Observation> myObservationDao;
+	private final IFhirResourceDao<Patient> myPatientDao;
+
+
 
 	public ReindexTestHelper(FhirContext theFhirContext, DaoRegistry theDaoRegistry, ISearchParamRegistry theSearchParamRegistry) {
 		myFhirContext = theFhirContext;
@@ -42,13 +50,30 @@ public class ReindexTestHelper {
 		mySearchParamRegistry = theSearchParamRegistry;
 		mySearchParameterDao = myDaoRegistry.getResourceDao(SearchParameter.class);
 		myObservationDao = myDaoRegistry.getResourceDao(Observation.class);
+		myPatientDao = myDaoRegistry.getResourceDao(Patient.class);
 	}
 
 	public void createAlleleSearchParameter() {
 		createAlleleSearchParameter(ALLELE_SP_CODE);
 	}
+	public void createEyeColourSearchParameter() {
+		createEyeColourSearchParameter(EYECOLOUR_SP_CODE);
+	}
 
-	public void createAlleleSearchParameter(String theCode) {
+	public DaoMethodOutcome createEyeColourPatient(boolean theActive) {
+		Patient patient = buildPatientWithEyeColourExtension(theActive);
+		return myPatientDao.create(patient);
+
+	}
+
+	private Patient buildPatientWithEyeColourExtension(boolean theActive) {
+		Patient p = new Patient();
+		p.addExtension().setUrl(EYECOLOUR_EXTENSION_URL).setValue(new StringType(TEST_EYECOLOUR_VALUE));
+		p.setActive(theActive);
+		return p;
+	}
+
+	public DaoMethodOutcome createAlleleSearchParameter(String theCode) {
 		SearchParameter alleleName = new SearchParameter();
 		alleleName.setId("SearchParameter/alleleName");
 		alleleName.setStatus(Enumerations.PublicationStatus.ACTIVE);
@@ -58,10 +83,27 @@ public class ReindexTestHelper {
 		alleleName.setTitle("AlleleName");
 		alleleName.setExpression("Observation.extension('" + ALLELE_EXTENSION_URL + "')");
 		alleleName.setXpathUsage(SearchParameter.XPathUsageType.NORMAL);
-		mySearchParameterDao.create(alleleName);
+		DaoMethodOutcome daoMethodOutcome = mySearchParameterDao.update(alleleName);
 		mySearchParamRegistry.forceRefresh();
+		return daoMethodOutcome;
 	}
 
+	public DaoMethodOutcome createEyeColourSearchParameter(String theCode) {
+		SearchParameter eyeColourSp = new SearchParameter();
+		eyeColourSp.setId("SearchParameter/eye-colour");
+		eyeColourSp.setStatus(Enumerations.PublicationStatus.ACTIVE);
+		eyeColourSp.addBase("Patient");
+		eyeColourSp.setCode(theCode);
+		eyeColourSp.setType(Enumerations.SearchParamType.TOKEN);
+		eyeColourSp.setTitle("Eye Colour");
+		eyeColourSp.setExpression("Patient.extension('" + EYECOLOUR_EXTENSION_URL+ "')");
+		eyeColourSp.setXpathUsage(SearchParameter.XPathUsageType.NORMAL);
+		DaoMethodOutcome daoMethodOutcome = mySearchParameterDao.update(eyeColourSp);
+		mySearchParamRegistry.forceRefresh();
+		return daoMethodOutcome;
+	}
+
+
 	public IIdType createObservationWithAlleleExtension(Observation.ObservationStatus theStatus) {
 		Observation observation = buildObservationWithAlleleExtension(theStatus);
 		return myObservationDao.create(observation).getId();
@@ -75,17 +117,32 @@ public class ReindexTestHelper {
 		return observation;
 	}
 
+	public List<String> getEyeColourPatientIds() {
+		return getEyeColourPatientIds(EYECOLOUR_SP_CODE, null);
+	}
+
 	public List<String> getAlleleObservationIds() {
 		return getAlleleObservationIds(ALLELE_SP_CODE, null);
 	}
 
+	public List<String> getEyeColourPatientIds(String theCode, String theIdentifier) {
+		SearchParameterMap map = SearchParameterMap.newSynchronous();
+		map.add(theCode, new TokenParam(TEST_EYECOLOUR_VALUE));
+		if (theIdentifier != null) {
+			map.add(Observation.SP_IDENTIFIER, new TokenParam(theIdentifier));
+		}
+		ourLog.info("Searching for Patients with url {}", map.toNormalizedQueryString(myFhirContext));
+		IBundleProvider result = myPatientDao.search(map);
+		return result.getAllResourceIds();
+	}
+
 	public List<String> getAlleleObservationIds(String theCode, String theIdentifier) {
 		SearchParameterMap map = SearchParameterMap.newSynchronous();
 		map.add(theCode, new TokenParam(TEST_ALLELE_VALUE));
 		if (theIdentifier != null) {
 			map.add(Observation.SP_IDENTIFIER, new TokenParam(theIdentifier));
 		}
-		ourLog.info("Searching with url {}", map.toNormalizedQueryString(myFhirContext));
+		ourLog.info("Searching for Observations with url {}", map.toNormalizedQueryString(myFhirContext));
 		IBundleProvider result = myObservationDao.search(map);
 		return result.getAllResourceIds();
 	}