Improve scatter/gather jobs when there is no primary output generated

Author: bbimber

SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/ProcessVariantsHandler.java

@@ -411,6 +411,7 @@ public static File processVCF(File input, Integer libraryId, JobContext ctx, Res
             resumer.getFileManager().addIntermediateFile(outputFileIdx);
         }
 
+        File effectiveInput = currentVCF; //this will be tested at the end to determine if a new file was actually created
         for (PipelineStepCtx<VariantProcessingStep> stepCtx : providers)
         {
             ctx.getLogger().info("Starting to run: " + stepCtx.getProvider().getLabel());
@@ -484,7 +485,7 @@ public static File processVCF(File input, Integer libraryId, JobContext ctx, Res
             resumer.setStepComplete(stepIdx, input.getPath(), action, currentVCF);
         }
 
-        if (currentVCF.exists())
+        if (currentVCF != null && currentVCF.exists() && !currentVCF.equals(effectiveInput))
         {
             resumer.getFileManager().removeIntermediateFile(currentVCF);
             resumer.getFileManager().removeIntermediateFile(new File(currentVCF.getPath() + ".tbi"));

SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/VariantProcessingRemoteMergeTask.java

@@ -3,6 +3,7 @@
 import htsjdk.samtools.util.Interval;
 import org.apache.commons.lang3.StringUtils;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 import org.labkey.api.pipeline.AbstractTaskFactory;
 import org.labkey.api.pipeline.AbstractTaskFactorySettings;
 import org.labkey.api.pipeline.PipelineJob;
@@ -127,13 +128,16 @@ private VariantProcessingJob getPipelineJob()
         if (handler instanceof SequenceOutputHandler.TracksVCF)
         {
             Set<SequenceOutputFile> outputs = new HashSet<>();
-            scatterOutputs.values().forEach(f -> outputs.addAll(getPipelineJob().getOutputsToCreate().stream().filter(x -> f.equals(x.getFile())).collect(Collectors.toSet())));
+            scatterOutputs.values().forEach(f -> outputs.addAll(getPipelineJob().getOutputsToCreate().stream().filter(x -> x != null && f.equals(x.getFile())).collect(Collectors.toSet())));
             getJob().getLogger().debug("Total component outputs created: " + outputs.size());
             getPipelineJob().getOutputsToCreate().removeAll(outputs);
             getJob().getLogger().debug("Total SequenceOutputFiles on job after remove: " + getPipelineJob().getOutputsToCreate().size());
 
-            SequenceOutputFile finalOutput = ((SequenceOutputHandler.TracksVCF)getPipelineJob().getHandler()).createFinalSequenceOutput(getJob(), finalOut, getPipelineJob().getFiles());
-            manager.addSequenceOutput(finalOutput);
+            if (finalOut != null)
+            {
+                SequenceOutputFile finalOutput = ((SequenceOutputHandler.TracksVCF) getPipelineJob().getHandler()).createFinalSequenceOutput(getJob(), finalOut, getPipelineJob().getFiles());
+                manager.addSequenceOutput(finalOutput);
+            }
         }
         else
         {
@@ -152,14 +156,15 @@ private VariantProcessingJob getPipelineJob()
         return new RecordedActionSet(action);
     }
 
-    private File runDefaultVariantMerge(JobContextImpl ctx, TaskFileManagerImpl manager, RecordedAction action, SequenceOutputHandler<SequenceOutputHandler.SequenceOutputProcessor> handler) throws PipelineJobException
+    private @Nullable File runDefaultVariantMerge(JobContextImpl ctx, TaskFileManagerImpl manager, RecordedAction action, SequenceOutputHandler<SequenceOutputHandler.SequenceOutputProcessor> handler) throws PipelineJobException
     {
         Map<String, List<Interval>> jobToIntervalMap = getPipelineJob().getJobToIntervalMap();
         getJob().setStatus(PipelineJob.TaskStatus.running, "Combining Per-Contig VCFs: " + jobToIntervalMap.size());
 
         Map<String, File> scatterOutputs = getPipelineJob().getScatterJobOutputs();
         List<File> toConcat = new ArrayList<>();
         Set<File> missing = new HashSet<>();
+        int totalNull = 0;
         for (String name : jobToIntervalMap.keySet())
         {
             if (!scatterOutputs.containsKey(name))
@@ -168,45 +173,29 @@ private File runDefaultVariantMerge(JobContextImpl ctx, TaskFileManagerImpl mana
             }
 
             File vcf = scatterOutputs.get(name);
-            if (!vcf.exists())
+            if (scatterOutputs.get(name) == null)
             {
-                missing.add(vcf);
-            }
-
-            // NOTE: this was added to fix a one-time issue where -L was dropped from some upstream GenotypeGVCFs.
-            // Under normal conditions this would never be necessary.
-            boolean ensureOutputsWithinIntervals = getPipelineJob().getParameterJson().optBoolean("variantCalling.GenotypeGVCFs.ensureOutputsWithinIntervalsOnMerge", false);
-            if (ensureOutputsWithinIntervals)
-            {
-                getJob().getLogger().debug("Ensuring ensure scatter outputs respect intervals");
-
-                File subsetVcf = new File(vcf.getParentFile(), SequenceAnalysisService.get().getUnzippedBaseName(vcf.getName()) + ".subset.vcf.gz");
-                File subsetVcfIdx = new File(subsetVcf.getPath() + ".tbi");
-                manager.addIntermediateFile(subsetVcf);
-                manager.addIntermediateFile(subsetVcfIdx);
-
-                if (subsetVcfIdx.exists())
-                {
-                    getJob().getLogger().debug("Index exists, will not re-subset the VCF: " + subsetVcf.getName());
-                }
-                else
-                {
-                    OutputVariantsStartingInIntervalsStep.Wrapper wrapper = new OutputVariantsStartingInIntervalsStep.Wrapper(getJob().getLogger());
-                    wrapper.execute(vcf, subsetVcf, getPipelineJob().getIntervalsForTask());
-                }
-
-                toConcat.add(subsetVcf);
+                totalNull++;
+                continue;
             }
-            else
+            else if (!vcf.exists())
             {
-                toConcat.add(vcf);
+                missing.add(vcf);
+                continue;
             }
 
+            toConcat.add(vcf);
+
             manager.addInput(action, "Input VCF", vcf);
             manager.addIntermediateFile(vcf);
             manager.addIntermediateFile(new File(vcf.getPath() + ".tbi"));
         }
 
+        if (totalNull > 0 && !toConcat.isEmpty())
+        {
+            throw new PipelineJobException("The scatter jobs returned a mixture of null and non-null outputs");
+        }
+
         Set<Integer> genomeIds = new HashSet<>();
         getPipelineJob().getFiles().forEach(x -> genomeIds.add(x.getLibrary_id()));
         if (genomeIds.size() != 1)
@@ -217,23 +206,27 @@ private File runDefaultVariantMerge(JobContextImpl ctx, TaskFileManagerImpl mana
         ReferenceGenome genome = getPipelineJob().getSequenceSupport().getCachedGenome(genomeIds.iterator().next());
 
         String basename = SequenceAnalysisService.get().getUnzippedBaseName(toConcat.get(0).getName());
-        File combined = new File(getPipelineJob().getAnalysisDirectory(), basename + ".vcf.gz");
-        File combinedIdx = new File(combined.getPath() + ".tbi");
-        if (combinedIdx.exists())
+        File combined = null;
+        if (!toConcat.isEmpty())
         {
-            getJob().getLogger().info("VCF exists, will not recreate: " + combined.getPath());
-        }
-        else
-        {
-            if (!missing.isEmpty())
+            combined = new File(getPipelineJob().getAnalysisDirectory(), basename + ".vcf.gz");
+            File combinedIdx = new File(combined.getPath() + ".tbi");
+            if (combinedIdx.exists())
             {
-                throw new PipelineJobException("Missing one of more VCFs: " + missing.stream().map(File::getPath).collect(Collectors.joining(",")));
+                getJob().getLogger().info("VCF exists, will not recreate: " + combined.getPath());
             }
+            else
+            {
+                if (!missing.isEmpty())
+                {
+                    throw new PipelineJobException("Missing one of more VCFs: " + missing.stream().map(File::getPath).collect(Collectors.joining(",")));
+                }
 
-            boolean sortAfterMerge = handler instanceof VariantProcessingStep.SupportsScatterGather && ((VariantProcessingStep.SupportsScatterGather)handler).doSortAfterMerge();
-            combined = SequenceAnalysisService.get().combineVcfs(toConcat, combined, genome, getJob().getLogger(), true, null, sortAfterMerge);
+                boolean sortAfterMerge = handler instanceof VariantProcessingStep.SupportsScatterGather && ((VariantProcessingStep.SupportsScatterGather) handler).doSortAfterMerge();
+                combined = SequenceAnalysisService.get().combineVcfs(toConcat, combined, genome, getJob().getLogger(), true, null, sortAfterMerge);
+            }
+            manager.addOutput(action, "Merged VCF", combined);
         }
-        manager.addOutput(action, "Merged VCF", combined);
 
         if (handler instanceof VariantProcessingStep.SupportsScatterGather)
         {

SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/VariantProcessingRemoteSplitTask.java

@@ -117,8 +117,13 @@ private VariantProcessingJob getPipelineJob()
                 {
                     output = ((SequenceOutputHandler.TracksVCF)handler).finalizeScatterJobOutput(ctx, output);
 
-                    // If the output is still under the work dir, translate path. Otherwise it was already copied to the the source dir
-                    if (output.getPath().startsWith(_wd.getDir().getPath()))
+                    // If the output is still under the work dir, translate path. Otherwise it was already copied to the source dir
+                    if (output == null)
+                    {
+                        ctx.getLogger().debug("No output produced, adding null to scatter outputs");
+                        getPipelineJob().getScatterJobOutputs().put(getPipelineJob().getIntervalSetName(), null);
+                    }
+                    else if (output.getPath().startsWith(_wd.getDir().getPath()))
                     {
                         //NOTE: the VCF will be copied back to the source dir, so translate paths
                         String path = _wd.getRelativePath(output);