Protect against duplicate quality metric import, such as failed/restarted job

bbimber · bbimber · commit 565e95a12183 · 2020-08-21T13:47:56.000-07:00
diff --git a/SequenceAnalysis/resources/schemas/dbscripts/postgresql/SequenceAnalysis-12.322-12.323.sql b/SequenceAnalysis/resources/schemas/dbscripts/postgresql/SequenceAnalysis-12.322-12.323.sql
@@ -0,0 +1,20 @@
+UPDATE sequenceanalysis.quality_metrics SET metricname = 'Mean Read Length' WHERE metricname = 'Avg Read Length';
+
+UPDATE sequenceanalysis.quality_metrics SET category = 'Readset' WHERE category IS NULL and metricname IN (
+  'Total Reads',
+  'Min Read Length',
+  'Max Read Length',
+  'Mean Read Length',
+  'Total Bases',
+  'Total MBases',
+  'Total GBases',
+  'Total Q10 Bases',
+  'Total Q20 Bases',
+  'Total Q30 Bases',
+  'Total Q40 Bases',
+  'Pct Q10',
+  'Pct Q20',
+  'Pct Q30',
+  'Pct Q40'
+);
+
diff --git a/SequenceAnalysis/resources/schemas/dbscripts/sqlserver/SequenceAnalysis-12.322-12.323.sql b/SequenceAnalysis/resources/schemas/dbscripts/sqlserver/SequenceAnalysis-12.322-12.323.sql
@@ -0,0 +1,20 @@
+UPDATE sequenceanalysis.quality_metrics SET metricname = 'Mean Read Length' WHERE metricname = 'Avg Read Length';
+
+UPDATE sequenceanalysis.quality_metrics SET category = 'Readset' WHERE category IS NULL and metricname IN (
+  'Total Reads',
+  'Min Read Length',
+  'Max Read Length',
+  'Mean Read Length',
+  'Total Bases',
+  'Total MBases',
+  'Total GBases',
+  'Total Q10 Bases',
+  'Total Q20 Bases',
+  'Total Q30 Bases',
+  'Total Q40 Bases',
+  'Pct Q10',
+  'Pct Q20',
+  'Pct Q30',
+  'Pct Q40'
+);
+
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/SequenceAnalysisModule.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/SequenceAnalysisModule.java
@@ -160,7 +160,7 @@ public String getName()
     @Override
     public Double getSchemaVersion()
     {
-        return 12.322;
+        return 12.323;
     }
 
     @Override
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/analysis/CellHashingHandler.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/analysis/CellHashingHandler.java
@@ -11,6 +11,7 @@
 import org.apache.log4j.Logger;
 import org.jetbrains.annotations.Nullable;
 import org.json.JSONObject;
+import org.labkey.api.data.CompareType;
 import org.labkey.api.data.Container;
 import org.labkey.api.data.SimpleFilter;
 import org.labkey.api.data.Sort;
@@ -182,6 +183,21 @@ public void complete(PipelineJob job, List<Readset> readsets, List<SequenceOutpu
             for (SequenceOutputFile so : outputsCreated)
             {
                 job.getLogger().info("Saving quality metrics for: " + so.getName());
+
+                //NOTE: if this job errored and restarted, we may have duplicate records:
+                SimpleFilter filter = new SimpleFilter(FieldKey.fromString("readset"), so.getReadset());
+                filter.addCondition(FieldKey.fromString("category"), "Cell Hashing", CompareType.EQUAL);
+                filter.addCondition(FieldKey.fromString("dataid"), so.getDataId(), CompareType.EQUAL);
+                filter.addCondition(FieldKey.fromString("container"), job.getContainer().getId(), CompareType.EQUAL);
+                TableSelector ts = new TableSelector(ti, PageFlowUtil.set("rowid"), filter, null);
+                if (ts.exists())
+                {
+                    job.getLogger().info("Deleting existing QC metrics (probably from prior restarted job)");
+                    ts.getArrayList(Integer.class).forEach(rowid -> {
+                        Table.delete(ti, rowid);
+                    });
+                }
+
                 if (so.getFile().getName().endsWith(CALL_EXTENSION))
                 {
                     Map<String, Object> counts = parseOutputTable(job.getLogger(), so.getFile(), getCiteSeqCountUnknownOutput(so.getFile().getParentFile(), _type, null), so.getFile().getParentFile(), null, false, _type);
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/IlluminaImportTask.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/IlluminaImportTask.java
@@ -297,6 +297,7 @@ private void addQualityMetrics(DbSchema schema, int readsetId, Pair<Integer, Int
         Integer count = readCounts.get(key);
 
         Map<String, Object> r = new HashMap<>();
+        r.put("category", "Readset");
         r.put("metricname", "Total Reads");
         r.put("metricvalue", count);
         r.put("dataid", d.getRowId());
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/ReadsetCreationTask.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/ReadsetCreationTask.java
@@ -420,6 +420,7 @@ public static long addQualityMetricsForReadset(Readset rs, int fileId, PipelineJ
             for (String metricName : metricsMap.keySet())
             {
                 Map<String, Object> r = new HashMap<>();
+                r.put("category", "Readset");
                 r.put("metricname", metricName);
                 r.put("metricvalue", metricsMap.get(metricName));
                 r.put("dataid", d.getRowId());
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/TaskFileManagerImpl.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/pipeline/TaskFileManagerImpl.java
@@ -1,6 +1,7 @@
 package org.labkey.sequenceanalysis.pipeline;
 
 import au.com.bytecode.opencsv.CSVReader;
+import au.com.bytecode.opencsv.CSVWriter;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -51,6 +52,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * User: bimber
@@ -420,12 +422,56 @@ public void addIntermediateFiles(Collection<File> files)
     @Override
     public void addPicardMetricsFiles(List<PipelineStepOutput.PicardMetricsOutput> files)
     {
+        //Note: in case there is a restarted job, inspect file for redundancy, and remove previous:
+        Set<String> filePaths = files.stream().map(PipelineStepOutput.PicardMetricsOutput::getMetricFile).map(File::getPath).collect(Collectors.toSet());
+
+        File metricLog = getMetricsLog(false);
+        if (metricLog.exists())
+        {
+            _job.getLogger().info("Inspecting existing metrics file for redundancy");
+
+            List<String[]> lines = new ArrayList<>();
+            boolean needToReplace = false;
+            try (CSVReader reader = new CSVReader(Readers.getReader(metricLog), '\t'))
+            {
+                String[] line;
+                while ((line = reader.readNext()) != null)
+                {
+                    if (filePaths.contains(line[6]))
+                    {
+                        needToReplace = true;
+                    }
+                    else
+                    {
+                        lines.add(line);
+                    }
+                }
+            }
+            catch (IOException e)
+            {
+                throw new RuntimeException(e);
+            }
+
+            if (needToReplace)
+            {
+                _job.getLogger().info("Replacing previously written metrics for these files, which probably indicates this job was restarted:");
+                try (CSVWriter writer = new CSVWriter(PrintWriters.getPrintWriter(metricLog), '\t', CSVWriter.NO_QUOTE_CHARACTER))
+                {
+                    lines.forEach(writer::writeNext);
+                }
+                catch (IOException e)
+                {
+                    throw new RuntimeException(e);
+                }
+            }
+        }
+
         for (PipelineStepOutput.PicardMetricsOutput mf : files)
         {
             _job.getLogger().debug("adding picard metrics file: " + mf.getMetricFile().getPath());
 
             //write to log
-            File metricLog = getMetricsLog(true);
+            metricLog = getMetricsLog(true);
             try (BufferedWriter writer = new BufferedWriter(new FileWriter(metricLog, true)))
             {
                 String bamRelPath = mf.getInputFile() == null ? "" : FilenameUtils.normalize(_wd.getRelativePath(mf.getInputFile()));
@@ -434,7 +480,7 @@ public void addPicardMetricsFiles(List<PipelineStepOutput.PicardMetricsOutput> f
                 List<Map<String, Object>> metricLines = PicardMetricsUtil.processFile(mf.getMetricFile(), _job.getLogger());
                 for (Map<String, Object> line : metricLines)
                 {
-                    writer.write(StringUtils.join(Arrays.asList(mf.getReadsetId(), bamRelPath, type, line.get("category"), line.get("metricname"), line.get("metricvalue")), "\t"));
+                    writer.write(StringUtils.join(Arrays.asList(mf.getReadsetId(), bamRelPath, type, line.get("category"), line.get("metricname"), line.get("metricvalue"), mf.getMetricFile().getPath()), "\t"));
                     writer.write('\n');
                 }
             }
@@ -471,9 +517,9 @@ public void writeMetricsToDb(Map<Integer, Integer> readsetMap, Map<Integer, Map<
                         continue; //header
                     }
 
-                    if (line.length != 6)
+                    if (line.length != 7)
                     {
-                        throw new RuntimeException("Line length is not 6: " + i + "[" + StringUtils.join(line, ";") + "]");
+                        throw new RuntimeException("Line length is not 7: " + i + "[" + StringUtils.join(line, ";") + "]");
                     }
 
                     Map<String, Object> toInsert = new HashMap<>();
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/alignment/CellRangerWrapper.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/alignment/CellRangerWrapper.java
@@ -8,15 +8,19 @@
 import org.jetbrains.annotations.Nullable;
 import org.json.JSONObject;
 import org.labkey.api.collections.CaseInsensitiveHashMap;
+import org.labkey.api.data.CompareType;
 import org.labkey.api.data.ConvertHelper;
 import org.labkey.api.data.DbSchema;
 import org.labkey.api.data.DbSchemaType;
+import org.labkey.api.data.SimpleFilter;
 import org.labkey.api.data.Table;
 import org.labkey.api.data.TableInfo;
+import org.labkey.api.data.TableSelector;
 import org.labkey.api.exp.api.ExpData;
 import org.labkey.api.exp.api.ExperimentService;
 import org.labkey.api.pipeline.PipelineJob;
 import org.labkey.api.pipeline.PipelineJobException;
+import org.labkey.api.query.FieldKey;
 import org.labkey.api.reader.Readers;
 import org.labkey.api.sequenceanalysis.model.AnalysisModel;
 import org.labkey.api.sequenceanalysis.model.ReadData;
@@ -636,6 +640,22 @@ public void complete(SequenceAnalysisJobSupport support, AnalysisModel model) th
                     }
 
                     TableInfo ti = DbSchema.get("sequenceanalysis", DbSchemaType.Module).getTable("quality_metrics");
+
+                    //NOTE: if this job errored and restarted, we may have duplicate records:
+                    SimpleFilter filter = new SimpleFilter(FieldKey.fromString("readset"), model.getReadset());
+                    filter.addCondition(FieldKey.fromString("analysis_id"), model.getRowId(), CompareType.EQUAL);
+                    filter.addCondition(FieldKey.fromString("dataid"), model.getAlignmentFile(), CompareType.EQUAL);
+                    filter.addCondition(FieldKey.fromString("category"), "Cell Ranger", CompareType.EQUAL);
+                    filter.addCondition(FieldKey.fromString("container"), getPipelineCtx().getJob().getContainer().getId(), CompareType.EQUAL);
+                    TableSelector ts = new TableSelector(ti, PageFlowUtil.set("rowid"), filter, null);
+                    if (ts.exists())
+                    {
+                        getPipelineCtx().getLogger().info("Deleting existing QC metrics (probably from prior restarted job)");
+                        ts.getArrayList(Integer.class).forEach(rowid -> {
+                            Table.delete(ti, rowid);
+                        });
+                    }
+
                     for (int j = 0; j < header.length; j++)
                     {
                         Map<String, Object> toInsert = new CaseInsensitiveHashMap<>();
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/analysis/LofreqAnalysis.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/analysis/LofreqAnalysis.java
@@ -30,11 +30,15 @@
 import org.biojava3.core.sequence.io.GenericFastaHeaderParser;
 import org.jetbrains.annotations.Nullable;
 import org.json.JSONObject;
+import org.labkey.api.data.CompareType;
 import org.labkey.api.data.DbSchema;
 import org.labkey.api.data.DbSchemaType;
+import org.labkey.api.data.SimpleFilter;
 import org.labkey.api.data.Table;
 import org.labkey.api.data.TableInfo;
+import org.labkey.api.data.TableSelector;
 import org.labkey.api.pipeline.PipelineJobException;
+import org.labkey.api.query.FieldKey;
 import org.labkey.api.reader.Readers;
 import org.labkey.api.sequenceanalysis.SequenceAnalysisService;
 import org.labkey.api.sequenceanalysis.model.AnalysisModel;
@@ -51,6 +55,7 @@
 import org.labkey.api.sequenceanalysis.run.AbstractCommandWrapper;
 import org.labkey.api.sequenceanalysis.run.SimpleScriptWrapper;
 import org.labkey.api.util.FileUtil;
+import org.labkey.api.util.PageFlowUtil;
 import org.labkey.api.writer.PrintWriters;
 import org.labkey.sequenceanalysis.SequenceAnalysisModule;
 import org.labkey.sequenceanalysis.run.util.DepthOfCoverageWrapper;
@@ -659,6 +664,21 @@ public Output performAnalysisPerSampleLocal(AnalysisModel model, File inputBam,
             getPipelineCtx().getLogger().info("Loading metrics");
             int total = 0;
             TableInfo ti = DbSchema.get("sequenceanalysis", DbSchemaType.Module).getTable("quality_metrics");
+
+            //NOTE: if this job errored and restarted, we may have duplicate records:
+            SimpleFilter filter = new SimpleFilter(FieldKey.fromString("readset"), model.getReadset());
+            filter.addCondition(FieldKey.fromString("dataid"), model.getAlignmentFile(), CompareType.EQUAL);
+            filter.addCondition(FieldKey.fromString("analysis_id"), model.getRowId(), CompareType.EQUAL);
+            filter.addCondition(FieldKey.fromString("container"), getPipelineCtx().getJob().getContainer().getId(), CompareType.EQUAL);
+            TableSelector ts = new TableSelector(ti, PageFlowUtil.set("rowid"), filter, null);
+            if (ts.exists())
+            {
+                getPipelineCtx().getLogger().info("Deleting existing QC metrics (probably from prior restarted job)");
+                ts.getArrayList(Integer.class).forEach(rowid -> {
+                    Table.delete(ti, rowid);
+                });
+            }
+
             try (CSVReader reader = new CSVReader(Readers.getReader(metrics), '\t'))
             {
                 String[] line;
diff --git a/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/preprocessing/TagPcrSummaryStep.java b/SequenceAnalysis/src/org/labkey/sequenceanalysis/run/preprocessing/TagPcrSummaryStep.java
@@ -250,6 +250,22 @@ public Output performAnalysisPerSampleLocal(AnalysisModel model, File inputBam,
             getPipelineCtx().getJob().getLogger().info("Loading metrics");
             AtomicInteger total = new AtomicInteger(0);
             TableInfo ti = DbSchema.get("sequenceanalysis", DbSchemaType.Module).getTable("quality_metrics");
+
+            //NOTE: if this job errored and restarted, we may have duplicate records:
+            SimpleFilter filter = new SimpleFilter(FieldKey.fromString("readset"), model.getReadset());
+            filter.addCondition(FieldKey.fromString("dataid"), model.getAlignmentFile(), CompareType.EQUAL);
+            filter.addCondition(FieldKey.fromString("analysis_id"), model.getRowId(), CompareType.EQUAL);
+            filter.addCondition(FieldKey.fromString("category"), "Tag-PCR", CompareType.EQUAL);
+            filter.addCondition(FieldKey.fromString("container"), getPipelineCtx().getJob().getContainer().getId(), CompareType.EQUAL);
+            TableSelector ts = new TableSelector(ti, PageFlowUtil.set("rowid"), filter, null);
+            if (ts.exists())
+            {
+                getPipelineCtx().getLogger().info("Deleting existing QC metrics (probably from prior restarted job)");
+                ts.getArrayList(Integer.class).forEach(rowid -> {
+                    Table.delete(ti, rowid);
+                });
+            }
+
             Map<String, String> metricsMap = parseMetricFile(metrics);
             metricsMap.forEach((metricname, value) -> {
                     Map<String, Object> r = new HashMap<>();

Original file line number	Diff line number	Diff line change
`@@ -160,7 +160,7 @@ public String getName()`
`160`	`160`	`@Override`
`161`	`161`	`public Double getSchemaVersion()`
`162`	`162`	`{`
`163`		`- return 12.322;`
	`163`	`+ return 12.323;`
`164`	`164`	`}`
`165`	`165`
`166`	`166`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -420,6 +420,7 @@ public static long addQualityMetricsForReadset(Readset rs, int fileId, PipelineJ`
`420`	`420`	`for (String metricName : metricsMap.keySet())`
`421`	`421`	`{`
`422`	`422`	`Map<String, Object> r = new HashMap<>();`
	`423`	`+ r.put("category", "Readset");`
`423`	`424`	`r.put("metricname", metricName);`
`424`	`425`	`r.put("metricvalue", metricsMap.get(metricName));`
`425`	`426`	`r.put("dataid", d.getRowId());`