diff --git a/gortools/src/main/scala/gorsat/Analysis/SortAnalysis.scala b/gortools/src/main/scala/gorsat/Analysis/SortAnalysis.scala index 970df4ee5..9650e2c0d 100644 --- a/gortools/src/main/scala/gorsat/Analysis/SortAnalysis.scala +++ b/gortools/src/main/scala/gorsat/Analysis/SortAnalysis.scala @@ -49,6 +49,10 @@ case class SortState(sortStep: SortGenome) extends BinState { sortStep.finish() sortStep.reinit() } + + override def close(): Unit = { + sortStep.discard() + } } case class SortFactory(header: String, session: GorSession, sortInfo: Array[SortInfo], div: Int) extends BinFactory { diff --git a/gortools/src/main/scala/gorsat/Analysis/SortGenome.scala b/gortools/src/main/scala/gorsat/Analysis/SortGenome.scala index c351564aa..5a05b162f 100644 --- a/gortools/src/main/scala/gorsat/Analysis/SortGenome.scala +++ b/gortools/src/main/scala/gorsat/Analysis/SortGenome.scala @@ -67,6 +67,26 @@ case class SortGenome(header: String, session: GorSession, sortInfo: Array[Row.S alreadySorted = true } + /** + * Drops buffered rows and deletes spilled files without emitting anything. + */ + def discard(): Unit = { + deleteSpillFiles() + reinit() + } + + private def deleteSpillFiles(): Unit = { + ordFileList.foreach(x => { + try { + val f = new java.io.File(x) + f.delete + } catch { + case _: Exception => + // Do nothing + } + }) + } + private def flushToDisk(): Unit = { ensureSorted() val outputArray = inputArray.take(lines) @@ -125,15 +145,7 @@ case class SortGenome(header: String, session: GorSession, sortInfo: Array[Row.S // Do nothing } } - ordFileList.foreach(x => { - try { - val f = new java.io.File(x) - f.delete - } catch { - case _: Exception => - // Do nothing - } - }) + deleteSpillFiles() } } else { ensureSorted() diff --git a/gortools/src/test/java/gorsat/UTestBufferedAnalysisErrorState.java b/gortools/src/test/java/gorsat/UTestBufferedAnalysisErrorState.java new file mode 100644 index 000000000..8349465e7 --- /dev/null +++ b/gortools/src/test/java/gorsat/UTestBufferedAnalysisErrorState.java @@ -0,0 +1,143 @@ +/* + * BEGIN_COPYRIGHT + * + * Copyright (C) 2011-2013 deCODE genetics Inc. + * Copyright (C) 2013-2019 WuXi NextCode Inc. + * All Rights Reserved. + * + * GORpipe is free software: you can redistribute it and/or modify + * it under the terms of the AFFERO GNU General Public License as published by + * the Free Software Foundation. + * + * GORpipe is distributed "AS-IS" AND WITHOUT ANY WARRANTY OF ANY KIND, + * INCLUDING ANY IMPLIED WARRANTY OF MERCHANTABILITY, + * NON-INFRINGEMENT, OR FITNESS FOR A PARTICULAR PURPOSE. See + * the AFFERO GNU General Public License for the complete license terms. + * + * You should have received a copy of the AFFERO GNU General Public License + * along with GORpipe. If not, see + * + * END_COPYRIGHT + */ + +package gorsat; + +import gorsat.process.CLIGorExecutionEngine; +import gorsat.process.PipeOptions; +import org.gorpipe.exceptions.GorDataException; +import org.junit.Assert; +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * Buffering analysis steps (group, sort, ...) must not flush their buffered rows at finish + * when an upstream step has thrown. Otherwise a failing query emits a partial result row + * before the error is reported. + */ +public class UTestBufferedAnalysisErrorState { + + private static class Outcome { + List rows; + Throwable error; + } + + /** + * Runs the query the way the command line does, pushing rows through the pipeline into an output + * step, and returns the data rows written (header excluded) along with any error. + */ + private static Outcome run(String query) { + Outcome outcome = new Outcome(); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + PipeOptions options = PipeOptions.parseInputArguments(new String[]{query}); + try { + new CLIGorExecutionEngine(options, null, null, out).execute(); + } catch (Throwable t) { + outcome.error = t; + } + String[] lines = out.toString(StandardCharsets.UTF_8).split("\n"); + outcome.rows = Arrays.stream(lines).skip(1).filter(l -> !l.isEmpty()).toList(); + return outcome; + } + + private static void assertOnlyError(String query) { + Outcome outcome = run(query); + Assert.assertNotNull("Expected an error for: " + query, outcome.error); + Assert.assertTrue("Expected GorDataException but got " + outcome.error, + hasCause(outcome.error, GorDataException.class)); + Assert.assertEquals("No data rows expected for: " + query, List.of(), outcome.rows); + } + + private static boolean hasCause(Throwable t, Class type) { + for (Throwable c = t; c != null; c = c.getCause()) { + if (type.isInstance(c)) return true; + } + return false; + } + + @Test + public void groupCountEmitsNothingWhenUpstreamThrows() { + assertOnlyError("nor <(norrows 5) | throwif #1 = 3 | group -count"); + } + + @Test + public void groupListEmitsNothingWhenUpstreamThrows() { + assertOnlyError("nor <(norrows 5) | throwif #1 = 3 | group -lis -sc #1"); + } + + @Test + public void sortEmitsNothingWhenUpstreamThrows() { + assertOnlyError("nor <(norrows 5) | throwif #1 = 3 | sort -c #1:nr"); + } + + @Test + public void gorGroupEmitsNothingWhenUpstreamThrows() { + assertOnlyError("gorrows -p chr1:1-6 | throwif pos = 3 | group chrom -count"); + } + + @Test + public void gorSortEmitsNothingWhenUpstreamThrows() { + assertOnlyError("gorrows -p chr1:1-6 | throwif pos = 3 | sort chrom -c pos:nr"); + } + + @Test + public void sortRemovesSpillFilesWhenUpstreamThrows() { + String oldBatchSize = System.getProperty("gor.sort.batchSize"); + System.setProperty("gor.sort.batchSize", "24"); + try { + Set before = sortSpillFiles(); + assertOnlyError("gorrows -p chr1:1-100 | throwif pos = 90 | sort chrom -c pos:nr"); + Set left = sortSpillFiles(); + left.removeAll(before); + Assert.assertEquals("Sort spill files left behind", Set.of(), left); + } finally { + if (oldBatchSize == null) System.clearProperty("gor.sort.batchSize"); + else System.setProperty("gor.sort.batchSize", oldBatchSize); + } + } + + private static Set sortSpillFiles() { + String[] names = new File(System.getProperty("java.io.tmpdir")).list((dir, name) -> name.startsWith("gorsort")); + return names == null ? new HashSet<>() : new HashSet<>(Arrays.asList(names)); + } + + @Test + public void groupOutputUnchangedWithoutError() { + Outcome outcome = run("nor <(norrows 5) | throwif #1 = 99 | group -count -lis -sc #1"); + Assert.assertNull(outcome.error); + Assert.assertEquals(List.of("5\t0,1,2,3,4"), outcome.rows); + } + + @Test + public void sortOutputUnchangedWithoutError() { + Outcome outcome = run("nor <(norrows 3) | throwif #1 = 99 | sort -c #1:nr"); + Assert.assertNull(outcome.error); + Assert.assertEquals(List.of("2", "1", "0"), outcome.rows); + } +} diff --git a/model/src/main/scala/gorsat/Commands/BinAggregator.scala b/model/src/main/scala/gorsat/Commands/BinAggregator.scala index 1d6522829..436e021b4 100644 --- a/model/src/main/scala/gorsat/Commands/BinAggregator.scala +++ b/model/src/main/scala/gorsat/Commands/BinAggregator.scala @@ -120,5 +120,21 @@ case class BinAggregator(binFactory: BinFactory, numBins: Int, window: Int, useK } currentKey = null } + + /** + * Releases the bins without sending their content to the next processor, used when the pipeline has failed. + */ + def discard(): Unit = { + for (mID <- bins.indices) { + binInfo(mID).used = false + try { + bins(mID).close() + } catch { + case _: Exception => + // ignore close error + } + } + currentKey = null + } } diff --git a/model/src/main/scala/gorsat/Commands/BinAnalysis.scala b/model/src/main/scala/gorsat/Commands/BinAnalysis.scala index eb06cad84..2074b1755 100644 --- a/model/src/main/scala/gorsat/Commands/BinAnalysis.scala +++ b/model/src/main/scala/gorsat/Commands/BinAnalysis.scala @@ -34,6 +34,8 @@ abstract class BinAnalysis(rowHandler: RowHandler, binAggregator: BinAggregator) } override def finish(): Unit = { - binAggregator.finalFlush() + // An upstream step has failed, the buffered bins are partial and must not be emitted. + if (isInErrorState) binAggregator.discard() + else binAggregator.finalFlush() } }