GetPileupSummaries streams output rather than storing it in memory

src/main/java/org/broadinstitute/hellbender/tools/walkers/contamination/GetPileupSummaries.java

@@ -13,10 +13,13 @@
 import org.broadinstitute.hellbender.engine.filters.ReadFilterLibrary;
 import org.broadinstitute.hellbender.engine.filters.WellformedReadFilter;
 import org.broadinstitute.hellbender.exceptions.UserException;
+import org.broadinstitute.hellbender.utils.io.IOUtils;
 import org.broadinstitute.hellbender.utils.pileup.ReadPileup;
 import org.broadinstitute.hellbender.utils.read.ReadUtils;
+import org.broadinstitute.hellbender.utils.tsv.TableUtils;
 
 import java.io.File;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -137,11 +140,11 @@ public class GetPileupSummaries extends LocusWalker {
     @Argument(fullName = MIN_MAPPING_QUALITY_LONG_NAME, shortName = MIN_MAPPING_QUALITY_SHORT_NAME, doc = "Minimum read mapping quality", optional = true)
     private int minMappingQuality = DEFAULT_MINIMUM_MAPPING_QUALITY;
 
-    private final List<PileupSummary> pileupSummaries = new ArrayList<>();
-
     private boolean sawVariantsWithoutAlleleFrequency = false;
     private boolean sawVariantsWithAlleleFrequency = false;
 
+    PileupSummary.PileupSummaryTableWriter writer;
+
     @Override
     public boolean requiresReads() {
         return true;
@@ -185,6 +188,15 @@ public void onTraversalStart() {
         if (!alleleFrequencyInHeader) {
             throw new UserException.BadInput("Population vcf does not have an allele frequency (AF) info field in its header.");
         }
+
+        try {
+            writer = new PileupSummary.PileupSummaryTableWriter(IOUtils.fileToPath(outputTable));
+            final String sampleName = ReadUtils.getSamplesFromHeader(getHeaderForReads()).stream().findFirst().get();
+            writer.writeMetadata(TableUtils.SAMPLE_METADATA_TAG, sampleName);
+        } catch (IOException ex) {
+            throw new UserException.CouldNotCreateOutputFile(outputTable, ex);
+        }
+
     }
 
     @Override
@@ -198,7 +210,11 @@ public void apply(AlignmentContext alignmentContext, ReferenceContext referenceC
         if ( vc.isBiallelic() && vc.isSNP() && alleleFrequencyInRange(vc) ) {
             final ReadPileup pileup = alignmentContext.getBasePileup()
                     .makeFilteredPileup(pe -> pe.getRead().getMappingQuality() >= minMappingQuality);
-            pileupSummaries.add(new PileupSummary(vc, pileup));
+            try {
+                writer.writeRecord(new PileupSummary(vc, pileup));
+            } catch (final IOException ex) {
+                throw new UserException(String.format("Encountered an IO exception while writing to %s", outputTable));
+            }
         }
     }
 
@@ -207,11 +223,21 @@ public Object onTraversalSuccess() {
         if (sawVariantsWithoutAlleleFrequency && !sawVariantsWithAlleleFrequency) {
             throw new UserException.BadInput("No variants in population vcf had an allele frequency (AF) field.");
         }
-        final String sampleName = ReadUtils.getSamplesFromHeader(getHeaderForReads()).stream().findFirst().get();
-        PileupSummary.writeToFile(sampleName, pileupSummaries, outputTable);
+
         return "SUCCESS";
     }
 
+    @Override
+    public void closeTool() {
+        try {
+            if (writer != null) {
+                writer.close();
+            }
+        } catch (IOException ex) {
+            throw new UserException(String.format("Encountered an IO exception while closing %s", outputTable));
+        }
+    }
+
     private boolean alleleFrequencyInRange(final VariantContext vc) {
         if (!vc.hasAttribute(VCFConstants.ALLELE_FREQUENCY_KEY)) {
             if (!sawVariantsWithoutAlleleFrequency) {

src/main/java/org/broadinstitute/hellbender/tools/walkers/contamination/PileupSummary.java

@@ -169,8 +169,8 @@ public int compare(PileupSummary ps1, PileupSummary ps2) {
     }
 
     //-------- The following methods are boilerplate for reading and writing pileup summary tables
-    private static class PileupSummaryTableWriter extends TableWriter<PileupSummary> {
-        private PileupSummaryTableWriter(final Path output) throws IOException {
+    public static class PileupSummaryTableWriter extends TableWriter<PileupSummary> {
+        public PileupSummaryTableWriter(final Path output) throws IOException {
             super(output, PileupSummaryTableColumn.COLUMNS);
         }