Merge integration to main for 0.9.12.2 hotfix

README.md

@@ -90,7 +90,7 @@ To compile the .mermaid source files to .png's, install the [Mermaid CLI](http:/
 
 ## Pre-requisites
 
-1. [Java 8](http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html)
+1. [Java 11](https://www.oracle.com/java/technologies/javase-downloads.html#JDK11)
 2. [SBT](http://www.scala-sbt.org/) to build
 3. [Apache Cassandra](http://cassandra.apache.org/) 2.x or 3.x (We prefer using [CCM](https://github.com/pcmanus/ccm) for local testing)
     - For testing, install a single node C* cluster, like this:  `ccm create v39_single -v 3.9 -n 1 -s`

memory/src/main/scala/filodb.memory/format/vectors/DoubleVector.scala

@@ -322,7 +322,8 @@ extends DoubleVectorDataReader {
     val it = iterate(acc, vect, 0)
     var last = Double.MinValue
     cforRange { 0 until len } { pos =>
-      val nextVal = it.next
+      var nextVal = it.next
+      if (nextVal.isNaN) nextVal = 0  // explicit counter reset due to end of time series marker
       if (nextVal < last) {   // reset!
         _correction += last
         _drops += pos
@@ -442,8 +443,10 @@ class DoubleCounterAppender(addr: BinaryRegion.NativePointer, maxBytes: Int, dis
 extends DoubleAppendingVector(addr, maxBytes, dispose) {
   private var last = Double.MinValue
   override final def addData(data: Double): AddResponse = {
-    if (data < last) PrimitiveVectorReader.markDrop(MemoryAccessor.nativePtrAccessor, addr)
-    last = data
+    if (!data.isNaN && data < last)
+      PrimitiveVectorReader.markDrop(MemoryAccessor.nativePtrAccessor, addr)
+    if (!data.isNaN)
+      last = data
     super.addData(data)
   }
 

query/src/main/scala/filodb/query/exec/PeriodicSamplesMapper.scala

@@ -444,7 +444,8 @@ class BufferableCounterCorrectionIterator(iter: Iterator[RowReader]) extends Ite
   override def hasNext: Boolean = iter.hasNext
   override def next(): TransientRow = {
     val next = iter.next()
-    val nextVal = next.getDouble(1)
+    var nextVal = next.getDouble(1)
+    if (nextVal.isNaN) nextVal = 0 // explicit counter reset due to end of time series marker
     if (nextVal < prevVal) {
       correction += prevVal
     }

query/src/test/scala/filodb/query/exec/WindowIteratorSpec.scala

@@ -252,7 +252,35 @@ class WindowIteratorSpec extends RawDataWindowingSpec {
     chunkedIt.foreach { v =>
       windowResults.find(a => a._1 == v.getLong(0)).foreach(b => v.getDouble(1) shouldEqual b._2 +- 0.0000000001)
     }
+  }
+
+  it("should deal with NaN end of time series marker during counter correction") {
+    val samples = Seq(
+      1614821996000L -> Double.NaN,
+      1614821996100L -> 489.0,
+      1614821997000L -> Double.NaN,
+      1614822566000L -> 19.0,
+      1614822596000L -> 26.0,
+      1614822626000L -> 26.0,
+      1614822656000L -> 26.0,
+      1614822686000L -> 26.0,
+      1614822716000L -> 26.0,
+      1614822717000L -> Double.NaN,
+      1614822866000L -> 5.0
+    )
+    val rawRows = samples.map(s => new TransientRow(s._1, s._2))
+    import filodb.core.query.NoCloseCursor._
+    val slidingWinIterator = new SlidingWindowIterator(rawRows.iterator, 1614822880000L, 15000, 1614822880000L, 900000L,
+      RangeFunction(tsResSchema,
+        Some(InternalRangeFunction.Rate), ColumnType.DoubleColumn, queryConfig,
+        useChunked = false).asSliding, queryConfig)
+      slidingWinIterator.next().getDouble(1) shouldEqual 0.5870753512132821
 
+    val rv = timeValueRVPk(samples)
+    val chunkedIt = new ChunkedWindowIteratorD(rv, 1614822880000L, 15000, 1614822880000L, 900000L,
+      RangeFunction(tsResSchema,
+        Some(Rate), ColumnType.DoubleColumn, queryConfig, useChunked = true).asChunkedD, querySession)
+    chunkedIt.next().getDouble(1) shouldEqual 0.5870753512132821
   }
 
   it("should calculate the rate for instant queries where step is 0") {