From 0494c2a175c510367c84c317d1463735704eea39 Mon Sep 17 00:00:00 2001 From: Test Date: Wed, 7 Oct 2026 20:38:07 +0000 Subject: [PATCH] fix(ENGKNOW-4002): do not flush group/sort buffers when pipeline is in error state When a step upstream of a BinAnalysis (group, sort, rank, atmax, ...) threw, finish() still flushed the buffered bins, so a failing query wrote a partial aggregate row before the error was reported. BinAnalysis.finish() now discards the bins when isInErrorState is set, matching LeftWhereAnalysis. Discarding closes the bins, and SortState.close() deletes any sort spill files so they are not left behind on error. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../scala/gorsat/Analysis/SortAnalysis.scala | 4 + .../scala/gorsat/Analysis/SortGenome.scala | 30 ++-- .../UTestBufferedAnalysisErrorState.java | 143 ++++++++++++++++++ .../scala/gorsat/Commands/BinAggregator.scala | 16 ++ .../scala/gorsat/Commands/BinAnalysis.scala | 4 +- 5 files changed, 187 insertions(+), 10 deletions(-) create mode 100644 gortools/src/test/java/gorsat/UTestBufferedAnalysisErrorState.java 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() } }