ALS-7737: Better support for splitting genomic data by chromosome

data/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/data/genotype/BucketIndexBySample.java

@@ -159,7 +159,11 @@ public Set<String> filterVariantSetForPatientSet(Set<String> variantSet, Collect
 			String bucketKey = variantSpec.split(",")[0] + ":" + (Integer.parseInt(variantSpec.split(",")[1])/1000);
 
 			//testBit uses inverted indexes include +2 offset for bookends
-			return _bucketMask.testBit(maxIndex - Collections.binarySearch(bucketList, bucketKey)  + 2);
+			int bucketKeyIndex = Collections.binarySearch(bucketList, bucketKey);
+			if (bucketKeyIndex < 0) {
+				return false;
+			}
+			return _bucketMask.testBit(maxIndex - bucketKeyIndex + 2);
 		}).collect(Collectors.toSet());
 	}
 

data/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/data/genotype/VariantMetadataIndex.java

@@ -34,7 +34,7 @@ public class VariantMetadataIndex implements Serializable {
 	private final Map<String,  FileBackedByteIndexedStorage<Integer, ConcurrentHashMap<String, String[]>> > indexMap = new HashMap<>();
 
 	public static final String VARIANT_METADATA_STORAGE_FILE_PREFIX = "VariantMetadataStorage";
-	private static String fileStoragePrefix = "/opt/local/hpds/all/" + VARIANT_METADATA_STORAGE_FILE_PREFIX;
+	private String fileStoragePrefix = "/opt/local/hpds/all/" + VARIANT_METADATA_STORAGE_FILE_PREFIX;
 
 	/**
 	 * This map allows us to load millions of variants without re-writing the fbbis each time (which would blow up the disk space).

docker/pic-sure-hpds-etl/Dockerfile

@@ -21,6 +21,7 @@ COPY --from=build /app/docker/pic-sure-hpds-etl/CSVLoader-jar-with-dependencies.
 COPY --from=build /app/docker/pic-sure-hpds-etl/CSVLoaderNewSearch-jar-with-dependencies.jar .
 COPY --from=build /app/docker/pic-sure-hpds-etl/CSVDumper-jar-with-dependencies.jar .
 COPY --from=build /app/docker/pic-sure-hpds-etl/VCFLocalLoader-jar-with-dependencies.jar .
+COPY --from=build /app/docker/pic-sure-hpds-etl/SplitChromosomeVcfLoader-jar-with-dependencies.jar .
 COPY --from=build /app/docker/pic-sure-hpds-etl/VariantMetadataLoader-jar-with-dependencies.jar .
 COPY --from=build /app/docker/pic-sure-hpds-etl/UnifiedVCFLocalLoader-jar-with-dependencies.jar .
 COPY --from=build /app/docker/pic-sure-hpds-etl/MultialleleCounter-jar-with-dependencies.jar .

etl/pom.xml

@@ -110,6 +110,26 @@
 							<goal>single</goal>
 						</goals>
 					</execution>
+					<execution>
+						<id>buildSplitChromosomeVcfLoader</id>
+						<configuration>
+							<archive>
+								<manifest>
+									<mainClass>edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.SplitChromosomeVcfLoader</mainClass>
+								</manifest>
+							</archive>
+							<outputDirectory>${project.basedir}/../docker/pic-sure-hpds-etl</outputDirectory>
+							<descriptorRefs>
+								<descriptorRef>jar-with-dependencies</descriptorRef>
+							</descriptorRefs>
+							<classifier>SplitChromosomeVcfLoader</classifier>
+							<finalName>SplitChromosomeVcfLoader</finalName>
+						</configuration>
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
+					</execution>
 					<execution>
 						<id>buildPerPatientUnifiedVCFLocalLoader</id>
 						<configuration>
@@ -275,7 +295,7 @@
 						<configuration>
 							<archive>
 								<manifest>
-									<mainClass>edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.VariantMetadataLoader</mainClass>
+									<mainClass>edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.SplitChromosomeVariantMetadataLoader</mainClass>
 								</manifest>
 							</archive>
 							<outputDirectory>${project.basedir}/../docker/pic-sure-hpds-etl</outputDirectory>

etl/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/data/genotype/util/ConvertMergedInfoStoresToFBBIS.java

@@ -1,14 +1,15 @@
 package edu.harvard.hms.dbmi.avillach.hpds.data.genotype.util;
+import edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.NewVCFLoader;
+
 import java.io.FileNotFoundException;
 import java.io.IOException;
 
-import static edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.NewVCFLoader.convertInfoStoresToByteIndexed;
 
 public class ConvertMergedInfoStoresToFBBIS {
 
-	public static void main(String[] args) throws FileNotFoundException, IOException {
+	public static void main(String[] args) throws IOException {
 
-		convertInfoStoresToByteIndexed();
+		new NewVCFLoader().convertInfoStoresToByteIndexed();
 
 	}
 

etl/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/data/genotype/util/ReSplitMergeInfoStores.java

@@ -1,19 +1,19 @@
 package edu.harvard.hms.dbmi.avillach.hpds.data.genotype.util;
 
-import static edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.NewVCFLoader.splitInfoStoresByColumn;
+import edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.NewVCFLoader;
 
-import java.io.FileNotFoundException;
 import java.io.IOException;
 
-import static edu.harvard.hms.dbmi.avillach.hpds.etl.genotype.NewVCFLoader.convertInfoStoresToByteIndexed;
 
 public class ReSplitMergeInfoStores {
 
-	public static void main(String[] args) throws FileNotFoundException, IOException {
+	public static void main(String[] args) throws IOException {
+
+		NewVCFLoader newVCFLoader = new NewVCFLoader();
 
-		splitInfoStoresByColumn();
+		newVCFLoader.splitInfoStoresByColumn();
 
-		convertInfoStoresToByteIndexed();
+		newVCFLoader.convertInfoStoresToByteIndexed();
 
 	}
 

etl/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/etl/genotype/NewVCFLoader.java

@@ -22,57 +22,77 @@
 
 public class NewVCFLoader {
 
+	public static final String DEFAULT_VCF_INDEX_FILE = "/opt/local/hpds/vcfIndex.tsv";
+	public static final String DEFAULT_STORAGE_DIR = "/opt/local/hpds/all";
+	public static final String DEFAULT_MERGED_DIR = "/opt/local/hpds/merged";
 	private static Logger logger = LoggerFactory.getLogger(NewVCFLoader.class);
-	private static File storageDir = null;
-	private static String storageDirStr = "/opt/local/hpds/all";
-	private static String mergedDirStr = "/opt/local/hpds/merged";
 
-	private static VariantIndexBuilder variantIndexBuilder = new VariantIndexBuilder();
+	protected File indexFile;
+	protected File storageDir;
+	protected String storageDirStr;
+	protected String mergedDirStr;
+
+	protected VariantIndexBuilder variantIndexBuilder;
 
 	// DO NOT CHANGE THIS unless you want to reload all the data everywhere.
-	private static int CHUNK_SIZE = 1000;
+	protected static int CHUNK_SIZE = 1000;
 
 	/**
 	 * @param args - if testing, this should be an array ['vcfIndexFile path', 'output storage dir', 'merged dir'].  
 	 * by default this will be [/opt/local/hpds/vcfIndex.tsv,  "/opt/local/hpds/all", "/opt/local/hpds/merged" ]. 
 	 */
-	public static void main(String[] args) throws FileNotFoundException, IOException {
-		
-		File indexFile;
+	public static void main(String[] args) throws IOException {
+
+		NewVCFLoader vcfLoader;
 		if(args != null && args.length >= 3) {
-			logger.info("Reading parameters from input - this is a test");
-			indexFile = new File(args[0]);
-			storageDirStr = args[1];
-			storageDir = new File(args[1]);
-			mergedDirStr = args[2];
+			logger.info("Reading parameters from input");
+			vcfLoader = new NewVCFLoader(new File(args[0]), args[1], args[2]);
 		} else {
-			indexFile = new File("/opt/local/hpds/vcfIndex.tsv");
-			storageDir = new File(storageDirStr);
+			logger.info(args.length + " arguments provided");
+			logger.info("Using default values");
+			vcfLoader = new NewVCFLoader();
 		}
-		loadVCFs(indexFile);
+		vcfLoader.loadAndMerge();
 	}
 
-	private static ExecutorService chunkWriteEx = Executors.newFixedThreadPool(1);
+	protected void loadAndMerge() throws IOException {
+		createWalkers();
+		loadVCFs();
+	}
 
-	private static ConcurrentHashMap<String, InfoStore> infoStoreMap = new ConcurrentHashMap<String, InfoStore>();
+	public NewVCFLoader() {
+		this.indexFile = new File(DEFAULT_VCF_INDEX_FILE);
+		this.storageDirStr = DEFAULT_STORAGE_DIR;
+		this.storageDir = new File(DEFAULT_STORAGE_DIR);
+		this.mergedDirStr = DEFAULT_MERGED_DIR;
+		this.variantIndexBuilder = new VariantIndexBuilder();
+	}
 
-	private static HashMap<String, char[][]> zygosityMaskStrings;
+	public NewVCFLoader(File indexFile, String storageDir, String mergedDirStr) {
+		this.indexFile = indexFile;
+		this.storageDirStr = storageDir;
+		this.storageDir = new File(storageDir);
+		this.mergedDirStr = mergedDirStr;
+		this.variantIndexBuilder = new VariantIndexBuilder();
+	}
 
-	private static TreeMap<String, FileBackedJsonIndexStorage<Integer, ConcurrentHashMap<String, VariableVariantMasks>>> variantMaskStorage = new TreeMap<>();
+	protected ExecutorService chunkWriteEx = Executors.newFixedThreadPool(1);
 
-	private static long startTime;
+	protected ConcurrentHashMap<String, InfoStore> infoStoreMap = new ConcurrentHashMap<String, InfoStore>();
 
-	private static List<VCFWalker> walkers = new ArrayList<>();
+	protected HashMap<String, char[][]> zygosityMaskStrings;
 
-	private static boolean contigIsHemizygous;
+	protected TreeMap<String, FileBackedJsonIndexStorage<Integer, ConcurrentHashMap<String, VariableVariantMasks>>> variantMaskStorage = new TreeMap<>();
 
-	private static void loadVCFs(File indexFile) throws IOException {
+	protected long startTime;
+
+	protected List<VCFWalker> walkers = new ArrayList<>();
+
+	private boolean contigIsHemizygous;
+
+	protected void loadVCFs() throws IOException {
 		startTime = System.currentTimeMillis();
-		List<VCFIndexLine> vcfIndexLines = parseVCFIndex(indexFile);
-		for (VCFIndexLine line : vcfIndexLines) {
-			walkers.add(new VCFWalker(line));
-		}
-		TreeSet<Integer> allPatientIds = new TreeSet<Integer>();
+		TreeSet<Integer> allPatientIds = new TreeSet<>();
 
 		// Pull the INFO columns out of the headers for each walker and add all patient ids
 		walkers.stream().forEach(walker -> {
@@ -232,7 +252,14 @@ private static void loadVCFs(File indexFile) throws IOException {
 		saveVariantStore(store, variantMaskStorage);
 	}
 
-	private static String sampleIdsForMask(String[] sampleIds, VariantMask variantMask) {
+	private void createWalkers() {
+		List<VCFIndexLine> vcfIndexLines = parseVCFIndex(indexFile);
+		for (VCFIndexLine line : vcfIndexLines) {
+			walkers.add(new VCFWalker(line));
+		}
+	}
+
+	protected String sampleIdsForMask(String[] sampleIds, VariantMask variantMask) {
 		StringBuilder idList = new StringBuilder();
 		if (variantMask != null) {
 			if (variantMask instanceof VariantMaskBitmaskImpl) {
@@ -251,7 +278,7 @@ private static String sampleIdsForMask(String[] sampleIds, VariantMask variantMa
 		return idList.toString();
 	}
 
-	private static void flipChunk(String lastContigProcessed, int lastChunkProcessed, int currentChunk,
+	protected void flipChunk(String lastContigProcessed, int lastChunkProcessed, int currentChunk,
 			String currentContig, boolean isLastChunk, String currentLine) throws IOException, FileNotFoundException {
 		if (!currentContig.contentEquals(lastContigProcessed) || isLastChunk) {
 			if (infoStoreFlipped.get(lastContigProcessed) == null || !infoStoreFlipped.get(lastContigProcessed)) {
@@ -310,7 +337,7 @@ private static void flipChunk(String lastContigProcessed, int lastChunkProcessed
 		}
 	}
 
-	private static void saveVariantStore(VariantStore store,
+	protected void saveVariantStore(VariantStore store,
 			TreeMap<String, FileBackedJsonIndexStorage<Integer, ConcurrentHashMap<String, VariableVariantMasks>>> variantMaskStorage)
 			throws IOException, FileNotFoundException {
 		store.setVariantMaskStorage(variantMaskStorage);
@@ -323,7 +350,7 @@ private static void saveVariantStore(VariantStore store,
 		logger.debug("Done saving variant masks.");
 	}
 
-	private static void saveInfoStores() throws IOException, FileNotFoundException {
+	protected void saveInfoStores() throws IOException, FileNotFoundException {
 		logger.debug("Saving info" + (System.currentTimeMillis() - startTime) + " seconds");
 		try (FileOutputStream fos = new FileOutputStream(new File(storageDir, "infoStores.javabin"));
 				GZIPOutputStream gzos = new GZIPOutputStream(fos);
@@ -334,7 +361,7 @@ private static void saveInfoStores() throws IOException, FileNotFoundException {
 		logger.info("completed load in " + (System.currentTimeMillis() - startTime) + " seconds");
 	}
 
-	public static void splitInfoStoresByColumn() throws FileNotFoundException, IOException {
+	public void splitInfoStoresByColumn() throws FileNotFoundException, IOException {
 		logger.debug("Splitting" + (System.currentTimeMillis() - startTime) + " seconds");
 		try {
 			VCFPerPatientInfoStoreSplitter.splitAll(storageDir,  new File(mergedDirStr));
@@ -344,7 +371,7 @@ public static void splitInfoStoresByColumn() throws FileNotFoundException, IOExc
 		logger.debug("Split" + (System.currentTimeMillis() - startTime) + " seconds");
 	}
 
-	public static void convertInfoStoresToByteIndexed() throws FileNotFoundException, IOException {
+	public void convertInfoStoresToByteIndexed() throws FileNotFoundException, IOException {
 		logger.debug("Converting" + (System.currentTimeMillis() - startTime) + " seconds");
 		try {
 			VCFPerPatientInfoStoreToFBBIISConverter.convertAll(mergedDirStr, storageDirStr);
@@ -354,7 +381,7 @@ public static void convertInfoStoresToByteIndexed() throws FileNotFoundException
 		logger.debug("Converted " + ((System.currentTimeMillis() - startTime) / 1000) + " seconds");
 	}
 
-	private static void shutdownChunkWriteExecutor() {
+	protected void shutdownChunkWriteExecutor() {
 		chunkWriteEx.shutdown();
 		while (!chunkWriteEx.isTerminated()) {
 			try {
@@ -377,16 +404,16 @@ private static ConcurrentHashMap<String, VariableVariantMasks> convertLoadingMap
 
 	static TreeMap<String, Boolean> infoStoreFlipped = new TreeMap<String, Boolean>();
 
-	private static class VCFWalker implements Comparable<VCFWalker> {
+	protected class VCFWalker implements Comparable<VCFWalker> {
 
-		private List<Integer> indices;
-		private Integer[] vcfOffsets;
-		private Integer[] bitmaskOffsets;
-		private HashMap<Integer, Integer> vcfIndexLookup;
-		private String currentLine;
-		private String[] currentLineSplit;
-		private BufferedReader vcfReader;
-		private VCFIndexLine vcfIndexLine;
+		protected List<Integer> indices;
+		protected Integer[] vcfOffsets;
+		protected Integer[] bitmaskOffsets;
+		protected HashMap<Integer, Integer> vcfIndexLookup;
+		protected String currentLine;
+		protected String[] currentLineSplit;
+		protected BufferedReader vcfReader;
+		protected VCFIndexLine vcfIndexLine;
 		boolean hasNext = true;
 		String currentContig;
 		Integer currentPosition;
@@ -469,7 +496,7 @@ private void setMasksForSample(char[][] zygosityMaskStrings, int index, int star
 			zygosityMaskStrings[patientZygosityIndex][bitmaskOffsets[index]] = '1';
 		}
 
-		private String currentSpecNotation() {
+		protected String currentSpecNotation() {
 			String[] variantInfo = currentLineSplit[7].split("[=;]");
 			String gene = "NULL";
 			String consequence = "NULL";
@@ -616,7 +643,7 @@ public int compareTo(VCFWalker o) {
 	private static final int SAMPLE_RELATIONSHIPS_COLUMN = 6;
 	private static final int RELATED_SAMPLE_IDS_COLUMN = 7;
 
-	private static class VCFIndexLine implements Comparable<VCFIndexLine> {
+	protected static class VCFIndexLine implements Comparable<VCFIndexLine> {
 		String vcfPath;
 		String contig;
 		boolean isAnnotated;