Updated code for better caching and reruning the indexer in place

agr_indexer/src/main/java/org/alliancegenome/indexer/config/IndexerConfig.java

@@ -23,6 +23,7 @@ public enum IndexerConfig {
 	GoIndexer("go", GoIndexer.class, 4, 914, 914, 8, 1),				// (44087,	262, 460247, 1093)
 	ModelIndexer("model", ModelIndexer.class, 4, 1426, 1426, 8, 1),		   // (132447, 314,	 18593,	 701)
 	DiseaseAnnotationMlIndexer("diseaseAnnotation", DiseaseAnnotationCurationIndexer.class, 1, 1426, 1426, 2, 1),
+	//PhenotypeAnnotationMlIndexer("phenotypeAnnotation", PhenotypeCurationIndexer.class, 1, 1426, 1426, 2, 1),
 	// still not implemented  VariantIndexer("variant", VariantIndexer.class, 2, 3000, 400, 4, 4)
 	;
 

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/AlleleIndexer.java

@@ -14,6 +14,7 @@
 import com.fasterxml.jackson.databind.MapperFeature;
 
 import lombok.extern.slf4j.Slf4j;
+import net.nilosplace.process_display.ProcessDisplayHelper;
 
 @Slf4j
 public class AlleleIndexer extends Indexer {
@@ -30,9 +31,11 @@ public AlleleIndexer(IndexerConfig config) {
 	public void index() {
 		try {
 			repo = new AlleleIndexerRepository();
+			log.info("Loading Doc Cache");
 			alleleDocumentCache = repo.getAlleleDocumentCache();
+			log.info("Loading Popularity");
 			alleleDocumentCache.setPopularity(popularityScore);
-
+			log.info("Loading The Queue");
 			LinkedBlockingDeque<String> queue = new LinkedBlockingDeque<>(alleleDocumentCache.getAlleleMap().keySet());
 
 			initiateThreading(queue);
@@ -48,7 +51,9 @@ public void index() {
 	protected void startSingleThread(LinkedBlockingDeque<String> queue) {
 		ArrayList<Allele> list = new ArrayList<>();
 		AlleleTranslator alleleTranslator = new AlleleTranslator();
+		ProcessDisplayHelper ph = new ProcessDisplayHelper(10000);
 		while (true) {
+			ph.startProcess("Allele Indexer", list.size());
 			try {
 				if (list.size() >= indexerConfig.getBufferSize()) {
 					Iterable<AlleleVariantSequence> avsDocs = alleleTranslator.translateEntities(list);
@@ -66,10 +71,12 @@ protected void startSingleThread(LinkedBlockingDeque<String> queue) {
 						repo.clearCache();
 						list.clear();
 					}
+					ph.finishProcess();
 					return;
 				}
 
 				String key = queue.takeFirst();
+				ph.progressProcess();
 				Allele allele = alleleDocumentCache.getAlleleMap().get(key);
 				if (allele != null) {
 					list.add(allele);

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/GeneIndexer.java

@@ -48,22 +48,22 @@ public void index() {
 
 	@Override
 	protected void startSingleThread(LinkedBlockingDeque<String> queue) {
-		ArrayList<Gene> list = new ArrayList<>();
+		ArrayList<Gene> bucket = new ArrayList<>();
 		GeneTranslator geneTrans = new GeneTranslator();
 		while (true) {
 			try {
-				if (list.size() >= indexerConfig.getBufferSize()) {
-					Iterable<SearchableItemDocument> geneDocuments = geneTrans.translateEntities(list);
+				if (bucket.size() >= indexerConfig.getBufferSize()) {
+					Iterable<SearchableItemDocument> geneDocuments = geneTrans.translateEntities(bucket);
 					geneDocumentCache.addCachedFields(geneDocuments);
 					indexDocuments(geneDocuments);
-					list.clear();
+					bucket.clear();
 				}
 				if (queue.isEmpty()) {
-					if (list.size() > 0) {
-						Iterable<SearchableItemDocument> geneDocuments = geneTrans.translateEntities(list);
+					if (bucket.size() > 0) {
+						Iterable<SearchableItemDocument> geneDocuments = geneTrans.translateEntities(bucket);
 						geneDocumentCache.addCachedFields(geneDocuments);
 						indexDocuments(geneDocuments);
-						list.clear();
+						bucket.clear();
 					}
 					return;
 				}
@@ -72,7 +72,7 @@ protected void startSingleThread(LinkedBlockingDeque<String> queue) {
 				Gene gene = geneDocumentCache.getGeneMap().get(key);
 
 				if (gene != null) {
-					list.add(gene);
+					bucket.add(gene);
 				} else {
 					log.debug("No gene found for " + key);
 				}

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/GoIndexer.java

@@ -30,7 +30,9 @@ public void index() {
 		log.info("Pulling All Terms Finished");
 
 		Iterable<SearchableItemDocument> docs = goTrans.translateEntities(terms);
-		docs.forEach(doc -> doc.setPopularity(popularityScore.get(doc.getPrimaryKey())));
+		for (SearchableItemDocument doc : docs) {
+			doc.setPopularity(popularityScore.get(doc.getPrimaryKey()));
+		}
 
 		log.info("Translation Done");
 
@@ -40,6 +42,7 @@ public void index() {
 
 	}
 
+	@Override
 	protected void startSingleThread(LinkedBlockingDeque<String> queue) {
 		// No need to multithread this
 	}

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/Indexer.java

@@ -1,6 +1,5 @@
 package org.alliancegenome.indexer.indexers;
 
-
 import java.io.IOException;
 import java.net.URL;
 import java.nio.file.Files;
@@ -41,7 +40,6 @@
 
 import lombok.extern.slf4j.Slf4j;
 
-
 @Slf4j
 public abstract class Indexer extends Thread {
 
@@ -99,7 +97,9 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
 		builder.setConcurrentRequests(indexerConfig.getConcurrentRequests());
 		builder.setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueSeconds(1L), 60));
 		bulkProcessor = builder.build();
-		//bulkProcessor = BulkProcessor.builder((request, bulkListener) -> searchClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener).build();
+		// bulkProcessor = BulkProcessor.builder((request, bulkListener) ->
+		// searchClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
+		// listener).build();
 
 	}
 
@@ -168,6 +168,10 @@ public <D extends ESDocument> void indexDocuments(Iterable<D> docs, Class<?> vie
 				} else {
 					json = om.writeValueAsString(doc);
 				}
+				if (json.length() > 19_000_000) {
+					log.warn("Document is too large for ES skipping: " + json.length());
+					continue;
+				}
 				stats.addDocument(json);
 				bulkProcessor.add(new IndexRequest(indexName).source(json, XContentType.JSON));
 				display.progressProcess();
@@ -185,6 +189,7 @@ public void initiateThreading(LinkedBlockingDeque<String> queue) throws Interrup
 		List<Thread> threads = new ArrayList<Thread>();
 		for (int i = 0; i < numberOfThreads; i++) {
 			Thread t = new Thread(new Runnable() {
+				@Override
 				public void run() {
 					startSingleThread(queue);
 				}

agr_indexer/src/main/java/org/alliancegenome/indexer/rest/interfaces/EcoTermRESTInterface.java → agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/curation/interfaces/EcoTermRESTInterface.java

@@ -1,4 +1,4 @@
-package org.alliancegenome.indexer.rest.interfaces;
+package org.alliancegenome.indexer.indexers.curation.interfaces;
 
 import org.alliancegenome.curation_api.model.entities.ontology.ECOTerm;
 import org.alliancegenome.curation_api.response.ObjectResponse;

agr_indexer/src/main/java/org/alliancegenome/indexer/rest/interfaces/VocabularyRESTInterface.java → agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/curation/interfaces/VocabularyRESTInterface.java

@@ -1,4 +1,4 @@
-package org.alliancegenome.indexer.rest.interfaces;
+package org.alliancegenome.indexer.indexers.curation.interfaces;
 
 import java.util.HashMap;
 
@@ -16,13 +16,13 @@
 import jakarta.ws.rs.QueryParam;
 
 @Path("/vocabularyterm")
-@Produces({"application/json"})
-@Consumes({"application/json"})
+@Produces({ "application/json" })
+@Consumes({ "application/json" })
 public interface VocabularyRESTInterface {
 
 	@POST
 	@Path("/find")
-	@JsonView({View.FieldsAndLists.class})
+	@JsonView({ View.FieldsAndLists.class })
 	SearchResponse<VocabularyTerm> find(@DefaultValue("0") @QueryParam("page") Integer page, @DefaultValue("10") @QueryParam("limit") Integer limit, HashMap<String, Object> params);
 
 }

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/curation/service/EcoTermService.java

@@ -4,7 +4,7 @@
 import org.alliancegenome.curation_api.model.entities.ontology.ECOTerm;
 import org.alliancegenome.curation_api.response.ObjectResponse;
 import org.alliancegenome.indexer.RestConfig;
-import org.alliancegenome.indexer.rest.interfaces.EcoTermRESTInterface;
+import org.alliancegenome.indexer.indexers.curation.interfaces.EcoTermRESTInterface;
 
 import si.mazi.rescu.RestProxyFactory;
 

agr_indexer/src/main/java/org/alliancegenome/indexer/indexers/curation/service/VocabularyService.java

@@ -6,7 +6,7 @@
 import org.alliancegenome.curation_api.model.entities.VocabularyTerm;
 import org.alliancegenome.curation_api.response.SearchResponse;
 import org.alliancegenome.indexer.RestConfig;
-import org.alliancegenome.indexer.rest.interfaces.VocabularyRESTInterface;
+import org.alliancegenome.indexer.indexers.curation.interfaces.VocabularyRESTInterface;
 
 import si.mazi.rescu.RestProxyFactory;