Download MCPD: optimized ElementCollection fields loading

src/main/java/org/genesys2/server/service/impl/DownloadServiceImpl.java

@@ -273,7 +273,7 @@ public class DownloadServiceImpl implements DownloadService {
 	public void writeXlsxMCPD(JPQLQuery<Long> queryAccessionId, OutputStream outputStream, String shortFilter, String dataSource) throws IOException {
 		writeXlsxMCPD((action) -> {
 			try {
-				accessionProcessor.process(queryAccessionId, action, null);
+				accessionProcessor.processMCPD(queryAccessionId, action, null);
 				LOG.warn("Done streaming MCPD rows");
 			} catch (Exception e) {
 				LOG.warn("Error generating: {}", e.getMessage());

src/main/java/org/genesys2/server/service/worker/AccessionProcessor.java

@@ -16,19 +16,32 @@
 package org.genesys2.server.service.worker;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
 
 import javax.persistence.EntityManager;
 
+import com.querydsl.core.types.EntityPath;
+import com.querydsl.core.types.dsl.NumberPath;
+import com.querydsl.core.types.dsl.SetPath;
+import com.querydsl.core.types.dsl.StringPath;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.time.StopWatch;
 import org.genesys2.server.model.genesys.Accession;
+import org.genesys2.server.model.genesys.AccessionCollect;
+import org.genesys2.server.model.genesys.AccessionData;
 import org.genesys2.server.model.genesys.QAccession;
+import org.genesys2.server.model.genesys.QAccessionCollect;
+import org.genesys2.server.model.genesys.QAccessionId;
 import org.genesys2.server.persistence.AccessionRepository;
 import org.genesys2.server.service.AccessionService;
 import org.genesys2.server.service.AccessionService.IAccessionBatchAction;
 import org.genesys2.server.service.ElasticsearchService;
 import org.genesys2.server.service.filter.AccessionFilter;
+import org.hibernate.Hibernate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -178,6 +191,108 @@ public class AccessionProcessor {
 		}
 	}
 
+	@Transactional(readOnly = true, propagation = Propagation.REQUIRES_NEW)
+	public void processMCPD(JPQLQuery<Long> accessionIdQuery, IAccessionBatchAction action, Long maxSize) throws Exception {
+		StopWatch stopWatch = new StopWatch();
+		stopWatch.start();
+
+		List<Long> results = accessionIdQuery.fetch();
+		var count = results.size();
+
+		for (var pos = 0; pos < count; pos += batchSize) {
+			stopWatch.split();
+			LOG.debug("Reading Accessions. Stopwatch={}s {}+{} of {}. Processing at {} accessions/s", stopWatch.getSplitTime() / 1000, pos, batchSize, count,
+				(double) (batchSize) / (stopWatch.getSplitTime() / 1000));
+			loadAndProcessForMCPD(results.subList(pos, Math.min(pos + batchSize, count)), action);
+			em.clear();
+		}
+
+		stopWatch.stop();
+		LOG.info("Processing Accessions took {}ms", stopWatch.getTime());
+	}
+
+	private void loadAndProcessForMCPD(List<Long> accessionIds, IAccessionBatchAction action) throws Exception {
+		if (! CollectionUtils.isEmpty(accessionIds)) {
+			List<Accession> accessions = accessionRepository.findAllById(accessionIds);
+
+			var accessionMap = accessions.stream().collect(Collectors.toMap(AccessionData::getId, accession -> accession));
+
+			if (!accessions.isEmpty()) {
+
+				var acceWithCollMap = accessions.stream()
+					.filter(accession -> Objects.nonNull(accession.getAccessionId().getColl()))
+					.collect(Collectors.toMap(AccessionData::getId, accession -> accession));
+
+				if (!acceWithCollMap.isEmpty()) {
+					acceWithCollMap.values().forEach(accession -> {
+						var collection = Hibernate.unproxy(accession.getAccessionId().getColl(), AccessionCollect.class);
+						collection.setCollCode(new HashSet<>());
+						collection.setCollName(new HashSet<>());
+						collection.setCollInstAddress(new HashSet<>());
+						accession.getAccessionId().setColl(collection);
+					});
+
+					fetchAndAddStrings(QAccessionCollect.accessionCollect, QAccessionCollect.accessionCollect.accession.id, QAccessionCollect.accessionCollect.collCode, acceWithCollMap.keySet(), (id, value) -> {
+						var target = acceWithCollMap.get(id).getAccessionId().getColl();
+						target.getCollCode().add(value);
+					});
+					fetchAndAddStrings(QAccessionCollect.accessionCollect, QAccessionCollect.accessionCollect.accession.id, QAccessionCollect.accessionCollect.collName, acceWithCollMap.keySet(), (id, value) -> {
+						var target = acceWithCollMap.get(id).getAccessionId().getColl();
+						target.getCollName().add(value);
+					});
+					fetchAndAddStrings(QAccessionCollect.accessionCollect, QAccessionCollect.accessionCollect.accession.id, QAccessionCollect.accessionCollect.collInstAddress, acceWithCollMap.keySet(), (id, value) -> {
+						var target = acceWithCollMap.get(id).getAccessionId().getColl();
+						target.getCollInstAddress().add(value);
+					});
+				}
+
+				accessionMap.values().forEach(accession -> {
+					accession.getAccessionId().setBreederCode(new HashSet<>());
+					accession.getAccessionId().setBreederName(new HashSet<>());
+					accession.getAccessionId().setDuplSite(new HashSet<>());
+					accession.getAccessionId().setStorage(new HashSet<>());
+				});
+				fetchAndAddStrings(QAccessionId.accessionId, QAccessionId.accessionId.id, QAccessionId.accessionId.breederCode, acceWithCollMap.keySet(), (id, value) -> {
+					var target = accessionMap.get(id).getAccessionId();
+					target.getBreederCode().add(value);
+				});
+				fetchAndAddStrings(QAccessionId.accessionId, QAccessionId.accessionId.id, QAccessionId.accessionId.breederName, acceWithCollMap.keySet(), (id, value) -> {
+					var target = accessionMap.get(id).getAccessionId();
+					target.getBreederName().add(value);
+				});
+				fetchAndAddStrings(QAccessionId.accessionId, QAccessionId.accessionId.id, QAccessionId.accessionId.duplSite, acceWithCollMap.keySet(), (id, value) -> {
+					var target = accessionMap.get(id).getAccessionId();
+					target.getDuplSite().add(value);
+				});
+				fetchAndAddNumbers(QAccessionId.accessionId, QAccessionId.accessionId.id, QAccessionId.accessionId.storage, acceWithCollMap.keySet(), (id, value) -> {
+					var target = accessionMap.get(id).getAccessionId();
+					target.getStorage().add(value);
+				});
+			}
+
+			action.apply(accessions);
+		}
+	}
+
+	private void fetchAndAddStrings(EntityPath<?> entityPath, NumberPath<Long> entityPathId, SetPath<String, StringPath> setPath, Collection<Long> identifiers, IAddStuff<String> addToCollection) {
+		var oneToMany = jpaQueryFactory.from(entityPath)
+			.select(entityPathId, setPath.any()).where(entityPathId.in(identifiers)).fetch();
+
+		oneToMany.forEach(combo -> addToCollection.accept(combo.get(entityPathId), combo.get(setPath.any())));
+	}
+
+	private <T extends Number & Comparable<?>> void fetchAndAddNumbers(EntityPath<?> entityPath, NumberPath<Long> entityPathId, SetPath<T, NumberPath<T>> setPath, Collection<Long> identifiers, IAddStuff<T> addToCollection) {
+		var oneToMany = jpaQueryFactory.from(entityPath)
+			.select(entityPathId, setPath.any()).where(entityPathId.in(identifiers)).fetch();
+
+		oneToMany.forEach(combo -> addToCollection.accept(combo.get(entityPathId), combo.get(setPath.any())));
+	}
+
+	public interface IAddStuff<T> {
+
+		void accept(Long id, T value);
+	}
+
 	/**
 	 * Apply action on accessions matching the provided filter.
 	 *

src/test/java/org/genesys/test/server/api/v1/AccessionControllerTest.java

@@ -32,6 +32,7 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
 import java.io.StringReader;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -39,6 +40,7 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import com.querydsl.jpa.impl.JPAQueryFactory;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.lang3.RandomUtils;
@@ -48,7 +50,9 @@ import org.genesys2.server.api.v1.AccessionController;
 import org.genesys2.server.api.v1.AccessionUploadController;
 import org.genesys2.server.model.genesys.Accession;
 import org.genesys2.server.model.genesys.AccessionAlias.AliasType;
+import org.genesys2.server.model.genesys.AccessionCollect;
 import org.genesys2.server.model.genesys.AccessionId;
+import org.genesys2.server.model.genesys.QAccession;
 import org.genesys2.server.model.impl.Crop;
 import org.genesys2.server.model.impl.DiversityTree;
 import org.genesys2.server.model.impl.DiversityTreeAccessionRef;
@@ -56,9 +60,11 @@ import org.genesys2.server.model.impl.DownloadLog;
 import org.genesys2.server.model.json.Api1Constants;
 import org.genesys2.server.service.filter.AccessionFilter;
 import org.genesys2.server.service.worker.AccessionOpResponse;
+import org.genesys2.server.service.worker.AccessionProcessor;
 import org.genesys2.spring.CSVMessageConverter;
 import org.hamcrest.Matchers;
 import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.MediaType;
 import org.springframework.security.test.context.support.WithMockUser;
 import org.springframework.test.web.servlet.MvcResult;
@@ -79,6 +85,12 @@ import com.opencsv.CSVReaderBuilder;
  */
 public class AccessionControllerTest extends AbstractAccessionControllerTest {
 
+	@Autowired
+	private AccessionProcessor accessionProcessor;
+
+	@Autowired
+	private JPAQueryFactory jpaQueryFactory;
+
 	@Test
 	public void createAccessionTest() throws Exception {
 		ObjectNode accessionJson = setUpAccession(true);
@@ -292,6 +304,54 @@ public class AccessionControllerTest extends AbstractAccessionControllerTest {
 		assertThat(downloadLog.getDate(), is(notNullValue()));
 	}
 
+	@Test
+	public void accessionProcessMCPD() throws Exception {
+		assertThat(accessionRepository.count(), is(0L));
+
+		List<Accession> accessions = new ArrayList<>();
+
+		for (int i = 0; i < 10; i++) {
+			var accession = addAccessionInDB(false, addTaxonomy("Musa", "banksii", null));
+			var collection = new AccessionCollect();
+			collection.setCollCode(Sets.newHashSet("MEX002", "IND002"));
+			collection.setCollName(Sets.newHashSet("NAME1", "NAME2"));
+			collection.setCollInstAddress(Sets.newHashSet("INST_ADDRESS"));
+			accession.getAccessionId().setColl(collection);
+			accession.getAccessionId().setBreederCode(Sets.newHashSet("CIV033"));
+			accession.getAccessionId().setBreederName(Sets.newHashSet("BreederName"));
+			accession.getAccessionId().setBreederName(Sets.newHashSet("BreederName"));
+			accession.getAccessionId().setDuplSite(Sets.newHashSet("USA005"));
+			accession.getAccessionId().setStorage(Sets.newHashSet(20));
+			accessions.add(accession);
+		}
+		accessions = accessionRepository.saveAll(accessions);
+		assertThat(accessionRepository.count(), is(10L));
+
+		var query = jpaQueryFactory.select(QAccession.accession.id).from(QAccession.accession);
+		accessionProcessor.processMCPD(query, resultAccessions -> {
+			resultAccessions.forEach(accession -> {
+				var accessionId = accession.getAccessionId();
+				var coll = accessionId.getColl();
+				assertThat(coll.getCollCode(), hasSize(2));
+				assertThat(coll.getCollCode(), hasItems("MEX002", "IND002"));
+				assertThat(coll.getCollName(), hasSize(2));
+				assertThat(coll.getCollName(), hasItems("NAME1", "NAME2"));
+				assertThat(coll.getCollInstAddress(), hasSize(1));
+				assertThat(coll.getCollInstAddress(), hasItem("INST_ADDRESS"));
+				assertThat(accessionId.getBreederCode(), hasSize(1));
+				assertThat(accessionId.getBreederCode(), hasItem("CIV033"));
+				assertThat(accessionId.getBreederName(), hasSize(1));
+				assertThat(accessionId.getBreederName(), hasItem("BreederName"));
+				assertThat(accessionId.getDuplSite(), hasSize(1));
+				assertThat(accessionId.getDuplSite(), hasItem("USA005"));
+				assertThat(accessionId.getStorage(), hasSize(1));
+				assertThat(accessionId.getStorage(), hasItem(20));
+			});
+			return resultAccessions;
+		} , null);
+
+	}
+
 	@Test
 	public void filterAccessionsByPartnerTest() throws Exception {
 		Partner partner = institute.getOwner();