add warning log for histogram value drop and add comments.

Author: Yu Zhang

memory/src/main/scala/filodb.memory/format/NibblePack.scala

@@ -319,6 +319,15 @@ object NibblePack {
       val numElems = Math.min(numBuckets - i, 8)
       cforRange { 0 until numElems } { n =>
         if (data(n) < lastHistDeltas(i + n)) valueDropped = true
+        // delta(n) represents the increase in the Nth bucket between measurements.
+        // Each new delta should be greater than or equal to the previous one.
+        // A decrease indicates a dropped or reset value, which is invalid.
+        // Example of an invalid histogram due to a drop:
+        //   time1 = (0, 131, 131)
+        //   time2 = (433, 447, 447)
+        //   -> Bucket 1 increased by 131 at time1, but only by 14 at time2, violating the monotonicity rule.
+        //   Previous deltas:  0, 131, 0  // lastHistDeltas
+        //   Current deltas:   0, 14,  0  // data
         packArray(n) = data(n) - originalDeltas(i + n)
       }
       System.arraycopy(data, 0, lastHistDeltas, i, numElems)

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

@@ -626,7 +626,7 @@ case class EmptyHistogramException(message: String) extends IllegalArgumentExcep
  * A reader for SectDelta encoded histograms, including correction/drop functionality
  */
 class SectDeltaHistogramReader(acc2: MemoryReader, histVect: Ptr.U8)
-      extends RowHistogramReader(acc2, histVect) with CounterHistogramReader {
+      extends RowHistogramReader(acc2, histVect) with CounterHistogramReader with StrictLogging {
   // baseHist is section base histogram; summedHist used to compute base + delta or other sums
   private val summedHist = LongHistogram.empty(buckets)
   private val baseHist = summedHist.copy
@@ -683,11 +683,18 @@ class SectDeltaHistogramReader(acc2: MemoryReader, histVect: Ptr.U8)
   // code to go through and build a list of corrections and corresponding index values.. (dropIndex, correction)
   private lazy val corrections = {
     var index = 0
+    var dropped = false
     // Step 1: build an iterator of (starting-index, section) for each section
-    iterateSections.map { case (s) => val o = (index, s); index += s.numElements(acc); o }.collect {
+    val correctionData = iterateSections.map { case (s) => val o = (index, s); index += s.numElements(acc); o }.collect {
       case (i, s) if i > 0 && s.sectionType(acc) == Section.TypeDrop =>
+        dropped = true
         (i, apply(i - 1).asInstanceOf[LongHistogram].copy)
     }.toBuffer
+    if (dropped) {
+      logger.warn(s"detected counter reset in histogram correction=${corrections}\n" +
+        s"allSections=${dumpAllSections}")
+    }
+    correctionData
   }
 
   def dropPositions(accNotUsed: MemoryReader, vectorNotUsed: BinaryVectorPtr): debox.Buffer[Int] = {