fix(ENGKNOW-2060): Add MetricsMonitor and fix pipe headers.

drivers/src/test/java/org/gorpipe/oci/driver/ITestOCIBvlTestSuite.java

@@ -13,7 +13,7 @@
 @Category(IntegrationTests.class)
 public class ITestOCIBvlTestSuite extends BvlTestSuite {
 
-    private String bucketName = "gdb_gor_test_data_dev";
+    private String bucketName = "gdb-gor-test-data-dev";
 
     private static String OCI_TENANT;
     private static String OCI_USER;

drivers/src/test/java/org/gorpipe/oci/driver/ITestOCICommonFilesTests.java

@@ -19,7 +19,7 @@
 @Category(IntegrationTests.class)
 public class ITestOCICommonFilesTests extends CommonFilesTests {
 
-    private String bucketName = "gdb_gor_test_data_dev";
+    private String bucketName = "gdb-gor-test-data-dev";
 
     private static String OCI_TENANT;
     private static String OCI_USER;

drivers/src/test/java/org/gorpipe/oci/driver/ITestOCICommonStreamTests.java

@@ -26,7 +26,7 @@
 @Category(IntegrationTests.class)
 public class ITestOCICommonStreamTests extends CommonStreamTests {
 
-    private String bucketName = "gdb_gor_test_data_dev";
+    private String bucketName = "gdb-gor-test-data-dev";
 
     private static String OCI_TENANT;
     private static String OCI_USER;

drivers/src/test/java/org/gorpipe/oci/driver/ITestOCISource.java

@@ -25,7 +25,7 @@
 @Category(IntegrationTests.class)
 public class ITestOCISource {
 
-    private String bucketName = "gdb_gor_test_data_dev";
+    private String bucketName = "gdb-gor-test-data-dev";
 
     private static String OCI_TENANT;
     private static String OCI_USER;

drivers/src/test/java/org/gorpipe/oci/driver/UTestOCIObjectStorageSourceType.java

@@ -42,29 +42,29 @@ static public void setUpClass() {
     @Test
     public void testResolveGorOciUrl() throws IOException {
         var source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("oci://gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null, false));
+                new SourceReference("oci://gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null, false));
         assertEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
 
         source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("oc://gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null, false));
+                new SourceReference("oc://gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null, false));
         assertEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
     }
 
     @Test
     public void testResolveOciHttpUrl() throws IOException {
         var source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("https://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null,  false));
+                new SourceReference("https://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null,  false));
         assertEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
 
         source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("http://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null, false));
+                new SourceReference("http://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null, false));
         assertEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
     }
 
     @Test
     public void testResolveOciNativeHttpUrl() throws IOException {
         var source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("https://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/n/namespace/b/gdb_gor_test_data_dev/o/the/path.dat", securityContext(), null, null, null, false));
+                new SourceReference("https://namespace.objectstorage.us-ashburn-1.oci.customer-oci.com/n/namespace/b/gdb-gor-test-data-dev/o/the/path.dat", securityContext(), null, null, null, false));
         assertEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
     }
 
@@ -82,11 +82,11 @@ public void testResolveNonOciHttpUrl() throws IOException {
     @Test
     public void testResolveS3HttpUrl() throws IOException {
         var source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("https://s3.us-east1.amazonaws.com/gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null,  false));
+                new SourceReference("https://s3.us-east1.amazonaws.com/gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null,  false));
         assertNotEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
 
         source = GorDriverFactory.fromConfig().resolveDataSource(
-                new SourceReference("http://s3.us-east1.amazonaws.com/gdb_gor_test_data_dev/the/path.dat", securityContext(), null, null, null,  false));
+                new SourceReference("http://s3.us-east1.amazonaws.com/gdb-gor-test-data-dev/the/path.dat", securityContext(), null, null, null,  false));
         assertNotEquals(OCIObjectStorageSourceType.OCI_OBJECT_STORAGE, source.getSourceType());
     }
 

drivers/src/testFixtures/java/org/gorpipe/utils/DriverUtils.java

@@ -86,7 +86,7 @@ public static String awsSecurityContext(String key, String secret) {
     public static String ociSecurityContext(String tenant, String user, String secret, String fingerprint) {
         // Credentials for gor_unittest user in nextcode OCI account
         Credentials cred = new Credentials.Builder().service("oci")
-                .lookupKey("gdb_gor_test_data_dev")
+                .lookupKey("gdb-gor-test-data-dev")
                 .set(Credentials.Attr.API_ENDPOINT, DEFAULT_OCI_ENDPOINT)
                 .set(Credentials.Attr.REALM, tenant)
                 .set(Credentials.Attr.KEY, fingerprint)

gorscripts/src/main/java/org/gorpipe/gor/cli/GorConfigDoc.java

@@ -122,7 +122,7 @@ public String toString() {
 
         try (Formatter formatter = new Formatter()) {
             for (String[] row : rows) {
-                formatter.format(format, row);
+                formatter.format(format, (Object)row);
             }
 
             return header + "\n" + formatter.out().toString();

gortools/src/main/scala/gorsat/Analysis/ForkWrite.scala

@@ -22,7 +22,6 @@
 
 package gorsat.Analysis
 
-import java.nio.file.{Files, Path, Paths, StandardOpenOption}
 import java.util.zip.Deflater
 import gorsat.Commands.{Analysis, Output, RowHeader}
 import gorsat.Outputs.OutFile
@@ -158,8 +157,9 @@ case class ForkWrite(forkCol: Int,
     * @return
     */
   def createOutFile(name: String, skipHeader: Boolean): Output = {
-    if (rowHeader==null || useFork) OutFile.driver(name, session.getProjectContext.getFileReader, header, skipHeader, options)
-    else {
+    if (rowHeader == null || useFork) {
+      OutFile.driver(name, session.getProjectContext.getFileReader, header, skipHeader, options)
+    } else {
       if (!rowHeader.toString.equals(header)) {
         rowHeader = RowHeader(header, rowHeader.columnTypes)
       }
@@ -284,6 +284,15 @@ case class ForkWrite(forkCol: Int,
     }
   }
 
+  // Available after finish.
+  def getMd5: String = {
+    if (!useFork && singleFileHolder.out != null) {
+      singleFileHolder.out.getMeta.getMd5
+    } else {
+      ""
+    }
+  }
+
   private def extractLink(fileName: String) : (String,String) = {
     var linkFile = options.linkFile
     var linkFileContent = ""

gortools/src/main/scala/gorsat/Analysis/OrderedMapAnalysis.scala

@@ -161,7 +161,7 @@ case class OrderedMapAnalysis(session: GorSession,
   }
 
   private def validateKeyOrder(r: Row, key: String, prevKey: String, source: String): Unit = {
-    if (prevKey != null && !prevKey.isEmpty && prevKey > key) {
+    if (prevKey != null && !prevKey.isEmpty && prevKey.compareTo(key) > 0) {
       throw new GorDataException(
         String.format("%s source is not ordered, as required if the -ordered options is used.  " +
           "Row '%s' is out of order.  Key/Prevkey was '%s'/'%s'.",

gortools/src/main/scala/gorsat/Monitors/CancelMonitor.scala

@@ -27,6 +27,8 @@ import org.gorpipe.gor.model.Row
 import org.gorpipe.gor.monitor.GorMonitor
 
 case class CancelMonitor(gm : GorMonitor) extends Analysis {
+  override def isTypeInformationMaintained: Boolean = true
+
   override def process(r : Row): Unit = {
     if (gm.isCancelled()) {
       reportWantsNoMore()

gortools/src/main/scala/gorsat/Monitors/MemoryMonitor.scala

@@ -35,6 +35,8 @@ case class MemoryMonitor(logname: String,
     throw new GorLowMemoryException(msg)
   }, minFreeMemMB, minFreeMemRatio)
 
+  override def isTypeInformationMaintained: Boolean = true
+
   override def process(r: Row): Unit = {
     mmu.check(r)
     super.process(r)

gortools/src/main/scala/gorsat/Monitors/MonitorLog.scala

@@ -28,6 +28,9 @@ import org.gorpipe.gor.monitor.GorMonitor
 
 case class MonitorLog(logname : String, n : Int, gm : GorMonitor) extends Analysis {
   var m = 0L
+
+  override def isTypeInformationMaintained: Boolean = true
+
   override def process(r : Row): Unit = {
     m += 1; if ((m % n) == 0) gm.log(logname+"> ("+m+") "+r.toColString)
     super.process(r)

gortools/src/main/scala/gorsat/Monitors/MonitorProgress.scala

@@ -31,6 +31,9 @@ case class MonitorProgress(milliSec : Int, gm : GorMonitor) extends Analysis {
   var t = System.currentTimeMillis
   var lastRowChr : String = ""
   var lastRowPos : Int = 0
+
+  override def isTypeInformationMaintained: Boolean = true
+
   override def process(r : Row): Unit = {
     m += 1
     if (r.chr != lastRowChr) {

gortools/src/main/scala/gorsat/Monitors/StatsMonitor.scala

@@ -0,0 +1,55 @@
+/*
+ *  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 <http://www.gnu.org/licenses/agpl-3.0.html>
+ *
+ *  END_COPYRIGHT
+ */
+
+package gorsat.Monitors
+
+import gorsat.Commands.{Analysis, RowHeader}
+import org.gorpipe.gor.model.Row
+
+/**
+ * Collect basic stats about the analysis stream.
+ */
+case class StatsMonitor() extends Analysis {
+  var startTime: Long = System.currentTimeMillis
+  var stopTime = 0L
+
+  var rowCount = 0L
+  var bytesCount = 0L
+
+  override def isTypeInformationMaintained: Boolean = true
+
+  def elapsedTime(): Long =  if (stopTime > 0)  stopTime - startTime else System.currentTimeMillis - startTime
+
+  override def process(r : Row): Unit = {
+    bytesCount += r.length()
+    rowCount += 1
+
+
+    super.process(r)
+  }
+  override def finish(): Unit = {
+    if (rowHeader == null) {
+      rowHeader = new RowHeader(Array(), Array())
+    }
+    stopTime = System.currentTimeMillis;
+  }
+}

gortools/src/main/scala/gorsat/Monitors/TimeoutMonitor.scala

@@ -39,7 +39,6 @@ case class TimeoutMonitor() extends Analysis {
   // Check for timeout every (checkEveryXRows) rows, and also at the first call to process on the instance
   var resettingRowCounter = checkEveryXRows
 
-
   override def isTypeInformationMaintained: Boolean = true
 
   override def process(r: Row): Unit = {

gortools/src/test/java/gorsat/external/plink/UTestPlinkRegression.java

@@ -30,6 +30,10 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 
+/* Can be run in container, something like:
+docker run -it --rm --name gor-test -v .:/opt/gor/src/  -w /opt/gor/src/ us-ashburn-1.ocir.io/id5mlxoq0dmt/genedx/gdb-gor-services:v13.1.3 bash
+export GRADLE_USER_HOME=/tmp/.gradle
+ */
 @Ignore("Needs plink2 installed")
 public class UTestPlinkRegression {
     String vcfheader = "#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\ta\tb\tc\td\te\tf\tg\ti\tj\n";

gortools/src/test/java/gorsat/monitors/StatsMonitorTest.java

@@ -0,0 +1,107 @@
+package gorsat.monitors;
+
+import gorsat.Analysis.ForkWrite;
+import gorsat.Analysis.OutputOptions;
+import gorsat.Monitors.StatsMonitor;
+import gorsat.process.PipeInstance;
+import org.gorpipe.gor.binsearch.GorIndexType;
+import org.gorpipe.gor.monitor.GorMonitor;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import scala.Option;
+
+import java.nio.file.Path;
+import java.util.zip.Deflater;
+
+import static gorsat.TestUtils.createPipeInstance;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class StatsMonitorTest{
+
+    @Rule
+    public TemporaryFolder workDir = new TemporaryFolder();
+    private Path workDirPath;
+
+    @Before
+    public void setupTest() {
+        workDirPath = workDir.getRoot().toPath();
+    }
+
+    @Test
+    public void testRowCountAndBytesCount() {
+
+        var statsMonitor = new StatsMonitor();
+        try (PipeInstance pipe = createPipeInstance(false)) {
+            pipe.init("gorrows -p chr1:1-1000 | calc a 'abc'", null);
+            pipe.lastStep().$bar(statsMonitor);
+
+            while (pipe.hasNext()) {
+                pipe.next();
+            }
+        }
+
+        assertEquals("chrom\tpos\ta", statsMonitor.getHeader());
+        assertEquals(999, statsMonitor.rowCount());
+        assertEquals(11880, statsMonitor.bytesCount());
+    }
+
+    @Test
+    public void testElapsedTime() throws InterruptedException {
+        StatsMonitor monitor = new StatsMonitor();
+        Thread.sleep(100);
+        monitor.finish();
+
+        long elapsedTime = monitor.elapsedTime();
+        assertTrue(elapsedTime >= 100);
+    }
+
+    @Test
+    public void testRowCountAndBytesCountForSelfWriting() {
+
+        var statsMonitor = new StatsMonitor();
+        try (PipeInstance pipe = createPipeInstance(false)) {
+            pipe.init("gorrows -p chr1:1-1000 | calc a 'abc' | write " + workDirPath.resolve("test.gor").toString(), new GorMonitor());
+            pipe.lastStep().$bar(statsMonitor);
+
+            while (pipe.hasNext()) {
+                pipe.next();
+            }
+        }
+
+        assertEquals("", statsMonitor.getHeader());
+        assertEquals(0, statsMonitor.rowCount());
+        assertEquals(0, statsMonitor.bytesCount());
+    }
+
+    @Test
+    public void testRowCountAndBytesCountForAddedWrite() {
+
+        var statsMonitor = new StatsMonitor();
+        ForkWrite forkWrite = null;
+        try (PipeInstance pipe = createPipeInstance(true)) {
+            pipe.init("gorrows -p chr1:1-1000 | calc a 'abc'", new GorMonitor());
+            pipe.lastStep().$bar(statsMonitor);
+
+            var outputOptions = new OutputOptions(false, false, true, false,
+                    false,  GorIndexType.NONE, new String[0], new String[0], Option.empty(),  Option.empty(), Deflater.BEST_SPEED,
+                    Option.empty(), false, false, null, "", null, false, false);
+            forkWrite = new ForkWrite(-1, workDirPath.resolve("test.gor").toString(), pipe.getSession(), pipe.getHeader(), outputOptions);
+
+            pipe.lastStep().$bar(forkWrite);
+
+            while (pipe.hasNext()) {
+                pipe.next();
+            }
+        }
+
+        assertEquals("chrom\tpos\ta", statsMonitor.getHeader());
+        assertEquals(999, statsMonitor.rowCount());
+        assertEquals(11880, statsMonitor.bytesCount());
+
+        assertEquals("c84f18441d00c7ea79d233efdaf8f07b", forkWrite.getMd5());
+    }
+
+}

model/src/main/scala/gorsat/Commands/Analysis.scala

@@ -33,7 +33,7 @@ abstract class Analysis() extends Processor with Cloneable {
   var nextProcessor: Processor = _
   var alreadyFinished = false
   var isInErrorState = false
-  var rowHeader: RowHeader = _
+  var rowHeader: RowHeader = _ // Input row header.
   var cloned : Analysis = _
   var isCloned = false
 
@@ -124,6 +124,9 @@ abstract class Analysis() extends Processor with Cloneable {
       pipeTo = to
       nextProcessor = to
       to.from(this)
+      if (rowHeader != null) {
+        setRowHeader(rowHeader)
+      }
     }
     this
   }
@@ -140,8 +143,9 @@ abstract class Analysis() extends Processor with Cloneable {
     null
   }
 
+  // Input header string
   def getHeader(): String = {
-    null
+    if (rowHeader != null) rowHeader.toString else null
   }
 
   def setup() : Unit = {}