[ALS-6818] Data Dictionary Loader Job from HPDS (#130)

* Add CSV loader for phenotype data

Introduced CSVLoaderNewSearch class to facilitate loading phenotype data from CSV files. Enhancements were made to the LoadingStore class, including writing column metadata to a CSV file. These changes improve data processing and metadata management functionality within the ETL workflow.

* Add CSVLoaderNewSearch jar and refine Java formatting

Added the CSVLoaderNewSearch jar to the Dockerfile and updated the Maven build process in pom.xml to include its packaging. Additionally, cleaned up Java code formatting by removing an unnecessary semicolon and adjusting spacing for consistency. These changes enhance the ETL process and maintain code readability.

* Add dumpStatsAndColumnMeta call to LoadingStore

Moved the dumpStatsAndColumnMeta execution from CSVLoaderNewSearch to LoadingStore for better encapsulation and responsibility alignment. This ensures metadata and statistics dumping is consistently handled within the store's lifecycle.

data/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/data/phenotype/ColumnMeta.java

@@ -2,14 +2,13 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 
 import java.io.Serializable;
 import java.util.List;
 
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
-public class ColumnMeta implements Serializable{
+public class ColumnMeta implements Serializable {
 
 	private static final long serialVersionUID = -124111104912063811L;
 	private String name;

docker/pic-sure-hpds-etl/Dockerfile

@@ -7,6 +7,7 @@ RUN apk add --no-cache --purge -uU curl wget unzip gnupg openssl
 ADD create_key.sh .
 ADD SQLLoader-jar-with-dependencies.jar .
 ADD CSVLoader-jar-with-dependencies.jar .
+ADD CSVLoaderNewSearch-jar-with-dependencies.jar .
 ADD CSVDumper-jar-with-dependencies.jar .
 ADD VCFLocalLoader-jar-with-dependencies.jar .
 ADD VariantMetadataLoader-jar-with-dependencies.jar .

etl/pom.xml

@@ -130,6 +130,26 @@
 							<goal>single</goal>
 						</goals>
 					</execution>
+					<execution>
+						<id>buildCSVLoaderNewSearch</id>
+						<configuration>
+							<archive>
+								<manifest>
+									<mainClass>edu.harvard.hms.dbmi.avillach.hpds.etl.phenotype.CSVLoaderNewSearch</mainClass>
+								</manifest>
+							</archive>
+							<outputDirectory>${project.basedir}/../docker/pic-sure-hpds-etl</outputDirectory>
+							<descriptorRefs>
+								<descriptorRef>jar-with-dependencies</descriptorRef>
+							</descriptorRefs>
+							<classifier>CSVLoaderNewSearch</classifier>
+							<finalName>CSVLoaderNewSearch</finalName>
+						</configuration>
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
+					</execution>
 					<execution>
 						<id>buildCSVLoader</id>
 						<configuration>

etl/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/etl/phenotype/CSVLoaderNewSearch.java

@@ -0,0 +1,118 @@
+package edu.harvard.hms.dbmi.avillach.hpds.etl.phenotype;
+
+import edu.harvard.hms.dbmi.avillach.hpds.crypto.Crypto;
+import edu.harvard.hms.dbmi.avillach.hpds.data.phenotype.PhenoCube;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.Date;
+
+@SuppressWarnings({"unchecked", "rawtypes"})
+public class CSVLoaderNewSearch {
+
+    private static final LoadingStore store = new LoadingStore();
+
+    private static final Logger log = LoggerFactory.getLogger(CSVLoaderNewSearch.class);
+
+    private static final int PATIENT_NUM = 0;
+
+    private static final int CONCEPT_PATH = 1;
+
+    private static final int NUMERIC_VALUE = 2;
+
+    private static final int TEXT_VALUE = 3;
+
+    private static final int DATETIME = 4;
+
+    private static boolean DO_VARNAME_ROLLUP = false;
+
+    private static final String HPDS_DIRECTORY = "/opt/local/hpds/";
+
+    public static void main(String[] args) throws IOException {
+        if (args.length > 1) {
+            if (args[0].equalsIgnoreCase("NO_ROLLUP")) {
+                log.info("NO_ROLLUP SET.");
+                DO_VARNAME_ROLLUP = false;
+            }
+        }
+        store.allObservationsStore = new RandomAccessFile(HPDS_DIRECTORY + "allObservationsStore.javabin", "rw");
+        initialLoad();
+        store.saveStore(HPDS_DIRECTORY);
+    }
+
+    private static void initialLoad() throws IOException {
+        Crypto.loadDefaultKey();
+        Reader in = new FileReader(HPDS_DIRECTORY + "allConcepts.csv");
+        Iterable<CSVRecord> records = CSVFormat.DEFAULT.withSkipHeaderRecord().withFirstRecordAsHeader().parse(new BufferedReader(in, 1024 * 1024));
+
+        final PhenoCube[] currentConcept = new PhenoCube[1];
+        for (CSVRecord record : records) {
+            processRecord(currentConcept, record);
+        }
+    }
+
+    private static void processRecord(final PhenoCube[] currentConcept, CSVRecord record) {
+        if (record.size() < 4) {
+            log.info("Record number {} had less records than we expected so we are skipping it.", record.getRecordNumber());
+            return;
+        }
+
+        String conceptPath = getSanitizedConceptPath(record);
+        String numericValue = record.get(NUMERIC_VALUE);
+        boolean isAlpha = (numericValue == null || numericValue.isEmpty());
+        String value = isAlpha ? record.get(TEXT_VALUE) : numericValue;
+        currentConcept[0] = getPhenoCube(currentConcept[0], conceptPath, isAlpha);
+
+        if (value != null && !value.trim().isEmpty() &&
+            ((isAlpha && currentConcept[0].vType == String.class) || (!isAlpha && currentConcept[0].vType == Double.class))) {
+            value = value.trim();
+            currentConcept[0].setColumnWidth(isAlpha ? Math.max(currentConcept[0].getColumnWidth(), value.getBytes().length) : Double.BYTES);
+            int patientId = Integer.parseInt(record.get(PATIENT_NUM));
+            Date date = null;
+            if (record.size() > 4 && record.get(DATETIME) != null && !record.get(DATETIME).isEmpty()) {
+                date = new Date(Long.parseLong(record.get(DATETIME)));
+            }
+            currentConcept[0].add(patientId, isAlpha ? value : Double.parseDouble(value), date);
+            store.allIds.add(patientId);
+        }
+    }
+
+    private static PhenoCube getPhenoCube(PhenoCube currentConcept, String conceptPath, boolean isAlpha) {
+        if (currentConcept == null || !currentConcept.name.equals(conceptPath)) {
+            currentConcept = store.store.getIfPresent(conceptPath);
+            if (currentConcept == null) {
+                currentConcept = new PhenoCube(conceptPath, isAlpha ? String.class : Double.class);
+                store.store.put(conceptPath, currentConcept);
+            }
+        }
+
+        return currentConcept;
+    }
+
+    private static String getSanitizedConceptPath(CSVRecord record) {
+        String conceptPathFromRow = record.get(CONCEPT_PATH);
+        String[] segments = conceptPathFromRow.split("\\\\");
+        for (int x = 0; x < segments.length; x++) {
+            segments[x] = segments[x].trim();
+        }
+        conceptPathFromRow = String.join("\\", segments) + "\\";
+        conceptPathFromRow = conceptPathFromRow.replaceAll("\\ufffd", "");
+        String textValueFromRow = record.get(TEXT_VALUE) == null ? null : record.get(TEXT_VALUE).trim();
+        if (textValueFromRow != null) {
+            textValueFromRow = textValueFromRow.replaceAll("\\ufffd", "");
+        }
+        String conceptPath;
+
+        if (DO_VARNAME_ROLLUP) {
+            conceptPath = conceptPathFromRow.endsWith("\\" + textValueFromRow + "\\") ? conceptPathFromRow.replaceAll("\\\\[^\\\\]*\\\\$", "\\\\") : conceptPathFromRow;
+        } else {
+            conceptPath = conceptPathFromRow;
+        }
+        return conceptPath;
+    }
+
+
+}

etl/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/etl/phenotype/LoadingStore.java

@@ -1,11 +1,17 @@
 package edu.harvard.hms.dbmi.avillach.hpds.etl.phenotype;
 
 import java.io.*;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
 
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVPrinter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -106,18 +112,19 @@ public PhenoCube load(String key) throws Exception {
 
 	public TreeSet<Integer> allIds = new TreeSet<Integer>();
 
-	public void saveStore(String hpdsDirectory) throws FileNotFoundException, IOException {
+	public void saveStore(String hpdsDirectory) throws IOException {
 		System.out.println("Invalidating store");
 		store.invalidateAll();
 		store.cleanUp();
 		System.out.println("Writing metadata");
-		ObjectOutputStream metaOut = new ObjectOutputStream(new GZIPOutputStream(new FileOutputStream(new File(hpdsDirectory + "columnMeta.javabin"))));
+		ObjectOutputStream metaOut = new ObjectOutputStream(new GZIPOutputStream(new FileOutputStream(hpdsDirectory + "columnMeta.javabin")));
 		metaOut.writeObject(metadataMap);
 		metaOut.writeObject(allIds);
 		metaOut.flush();
 		metaOut.close();
 		System.out.println("Closing Store");
 		allObservationsStore.close();
+		dumpStatsAndColumnMeta(hpdsDirectory);
 	}
 
 	public void dumpStats() {
@@ -149,5 +156,55 @@ public void dumpStats() {
 		}
 	}
 
+	/**
+	 * This method will display counts for the objects stored in the metadata.
+	 * This will also write out a csv file used by the data dictionary importer.
+	 */
+	public void dumpStatsAndColumnMeta(String hpdsDirectory) {
+		try (ObjectInputStream objectInputStream =
+					 new ObjectInputStream(new GZIPInputStream(new FileInputStream(hpdsDirectory + "columnMeta.javabin")))){
+			TreeMap<String, ColumnMeta> metastore = (TreeMap<String, ColumnMeta>) objectInputStream.readObject();
+			try(BufferedWriter writer = Files.newBufferedWriter(Paths.get(hpdsDirectory + "columnMeta.csv"), StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING)) {
+				CSVPrinter printer = new CSVPrinter(writer, CSVFormat.DEFAULT);
+				for(String key : metastore.keySet()) {
+					ColumnMeta columnMeta = metastore.get(key);
+					Object[] columnMetaOut = new Object[11];
+
+					StringBuilder listQuoted = new StringBuilder();
+					AtomicInteger x = new AtomicInteger(1);
+
+					if(columnMeta.getCategoryValues() != null){
+						if(!columnMeta.getCategoryValues().isEmpty()) {
+							columnMeta.getCategoryValues().forEach(string -> {
+								listQuoted.append(string);
+								if(x.get() != columnMeta.getCategoryValues().size()) listQuoted.append("µ");
+								x.incrementAndGet();
+							});
+						}
+					}
+
+					columnMetaOut[0] = columnMeta.getName();
+					columnMetaOut[1] = String.valueOf(columnMeta.getWidthInBytes());
+					columnMetaOut[2] = String.valueOf(columnMeta.getColumnOffset());
+					columnMetaOut[3] = String.valueOf(columnMeta.isCategorical());
+					// this should nest the list of values in a list inside the String array.
+					columnMetaOut[4] = listQuoted;
+					columnMetaOut[5] = String.valueOf(columnMeta.getMin());
+					columnMetaOut[6] = String.valueOf(columnMeta.getMax());
+					columnMetaOut[7] = String.valueOf(columnMeta.getAllObservationsOffset());
+					columnMetaOut[8] = String.valueOf(columnMeta.getAllObservationsLength());
+					columnMetaOut[9] = String.valueOf(columnMeta.getObservationCount());
+					columnMetaOut[10] = String.valueOf(columnMeta.getPatientCount());
+
+					printer.printRecord(columnMetaOut);
+				}
+
+				writer.flush();
+            }
+
+		} catch (IOException | ClassNotFoundException e) {
+			throw new RuntimeException("Could not load metastore", e);
+		}
+	}
 
 }