First implementation of weighted update for KllDoubles.

src/main/java/org/apache/datasketches/common/Util.java

@@ -729,6 +729,19 @@ public static boolean isOdd(final long n) {
     return (n & 1L) == 1L;
   }
 
+  //Other
+
+  /**
+   * Returns a one if the bit at bitPos is a one, otherwise zero.
+   * @param number the number to examine
+   * @param bitPos the given zero-based bit position, where the least significant
+   * bit is at position zero.
+   * @return a one if the bit at bitPos is a one, otherwise zero.
+   */
+  public static final int bitAt(final long number, final int bitPos) {
+    return (number & (1L << bitPos)) > 0 ? 1 : 0;
+  }
+
   /**
    * Computes the number of decimal digits of the number n
    * @param n the given number
@@ -756,6 +769,8 @@ public static String intToFixedLengthString(final int number, final int length)
     return characterPad(num, length, ' ', false);
   }
 
+  //Generic tests
+
   /**
    * Finds the minimum of two generic items
    * @param <T> the type

src/main/java/org/apache/datasketches/kll/KllDoublesHelper.java

@@ -39,6 +39,20 @@
 //
 final class KllDoublesHelper {
 
+  /**
+   * Create Items Array from given item and weight.
+   * Used with weighted update only.
+   * @param item the given item
+   * @param weight the given weight
+   * @return the Items Array.
+   */
+  static double[] createItemsArray(final double item, final int weight) {
+    final int itemsArrLen = Integer.bitCount(weight);
+    final double[] itemsArr = new double[itemsArrLen];
+    Arrays.fill(itemsArr, item);
+    return itemsArr;
+  }
+
   /**
    * The following code is only valid in the special case of exactly reaching capacity while updating.
    * It cannot be used while merging, while reducing k, or anything else.
@@ -135,12 +149,12 @@ static void mergeDoubleImpl(final KllDoublesSketch mySketch,
       }
     }
 
-    //After the level 0 update, we capture the intermediate state of levels and items arrays...
+    //After the level 0 update, we capture the intermediate state of my levels and items arrays...
     final int myCurNumLevels = mySketch.getNumLevels();
     final int[] myCurLevelsArr = mySketch.levelsArr;
     final double[] myCurDoubleItemsArr = mySketch.getDoubleItemsArray();
 
-    // then rename them and initialize in case there are no higher levels
+    // create aliases in case there are no higher levels
     int myNewNumLevels = myCurNumLevels;
     int[] myNewLevelsArr = myCurLevelsArr;
     double[] myNewDoubleItemsArr = myCurDoubleItemsArr;
@@ -150,12 +164,13 @@ static void mergeDoubleImpl(final KllDoublesSketch mySketch,
       final int tmpSpaceNeeded = mySketch.getNumRetained()
           + KllHelper.getNumRetainedAboveLevelZero(otherNumLevels, otherLevelsArr);
       final double[] workbuf = new double[tmpSpaceNeeded];
-      final int ub = KllHelper.ubOnNumLevels(finalN);
-      final int[] worklevels = new int[ub + 2]; // ub+1 does not work
-      final int[] outlevels  = new int[ub + 2];
 
       final int provisionalNumLevels = max(myCurNumLevels, otherNumLevels);
 
+      final int ub = max(KllHelper.ubOnNumLevels(finalN), provisionalNumLevels);
+      final int[] worklevels = new int[ub + 2]; // ub+1 does not work
+      final int[] outlevels  = new int[ub + 2];
+
       populateDoubleWorkArrays(workbuf, worklevels, provisionalNumLevels,
           myCurNumLevels, myCurLevelsArr, myCurDoubleItemsArr,
           otherNumLevels, otherLevelsArr, otherDoubleItemsArr);
@@ -199,7 +214,7 @@ static void mergeDoubleImpl(final KllDoublesSketch mySketch,
             KllHelper.memorySpaceMgmt(mySketch, myNewLevelsArr.length, myNewDoubleItemsArr.length);
         mySketch.setWritableMemory(wmem);
       }
-    }
+    } //end of updating levels above level 0
 
     //Update Preamble:
     mySketch.setN(finalN);
@@ -225,7 +240,7 @@ static void mergeDoubleImpl(final KllDoublesSketch mySketch,
     assert KllHelper.sumTheSampleWeights(mySketch.getNumLevels(), mySketch.levelsArr) == mySketch.getN();
   }
 
-  private static void mergeSortedDoubleArrays(
+  private static void mergeSortedDoubleArrays( //only bufC is modified
       final double[] bufA, final int startA, final int lenA,
       final double[] bufB, final int startB, final int lenB,
       final double[] bufC, final int startC) {
@@ -299,8 +314,7 @@ private static void randomlyHalveUpDoubles(final double[] buf, final int start,
   }
 
   //Called from KllDoublesSketch::update and this
-  static void updateDouble(final KllDoublesSketch dblSk,
-      final double item) {
+  static void updateDouble(final KllDoublesSketch dblSk, final double item) {
     if (Double.isNaN(item)) { return; } //ignore
     if (dblSk.isEmpty()) {
       dblSk.setMinItem(item);
@@ -445,32 +459,36 @@ private static int[] generalDoublesCompress(
     return new int[] {numLevels, targetItemCount, currentItemCount};
   }
 
-  private static void populateDoubleWorkArrays(
-      final double[] workbuf, final int[] worklevels, final int provisionalNumLevels,
+  private static void populateDoubleWorkArrays( //workBuf and workLevels are modified
+      final double[] workBuf, final int[] workLevels, final int provisionalNumLevels,
       final int myCurNumLevels, final int[] myCurLevelsArr, final double[] myCurDoubleItemsArr,
       final int otherNumLevels, final int[] otherLevelsArr, final double[] otherDoubleItemsArr) {
 
-    worklevels[0] = 0;
+    workLevels[0] = 0;
 
-    // Note: the level zero data from "other" was already inserted into "self"
+    // Note: the level zero data from "other" was already inserted into "self",
+    // This copies into workbuf.
     final int selfPopZero = KllHelper.currentLevelSizeItems(0, myCurNumLevels, myCurLevelsArr);
-    System.arraycopy(myCurDoubleItemsArr, myCurLevelsArr[0], workbuf, worklevels[0], selfPopZero);
-    worklevels[1] = worklevels[0] + selfPopZero;
+    System.arraycopy(myCurDoubleItemsArr, myCurLevelsArr[0], workBuf, workLevels[0], selfPopZero);
+    workLevels[1] = workLevels[0] + selfPopZero;
 
     for (int lvl = 1; lvl < provisionalNumLevels; lvl++) {
       final int selfPop = KllHelper.currentLevelSizeItems(lvl, myCurNumLevels, myCurLevelsArr);
       final int otherPop = KllHelper.currentLevelSizeItems(lvl, otherNumLevels, otherLevelsArr);
-      worklevels[lvl + 1] = worklevels[lvl] + selfPop + otherPop;
-
-      if (selfPop > 0 && otherPop == 0) {
-        System.arraycopy(myCurDoubleItemsArr, myCurLevelsArr[lvl], workbuf, worklevels[lvl], selfPop);
-      } else if (selfPop == 0 && otherPop > 0) {
-        System.arraycopy(otherDoubleItemsArr, otherLevelsArr[lvl], workbuf, worklevels[lvl], otherPop);
-      } else if (selfPop > 0 && otherPop > 0) {
-        mergeSortedDoubleArrays(
+      workLevels[lvl + 1] = workLevels[lvl] + selfPop + otherPop;
+
+      if (selfPop == 0 && otherPop == 0) { continue; }
+      else if (selfPop > 0 && otherPop == 0) {
+        System.arraycopy(myCurDoubleItemsArr, myCurLevelsArr[lvl], workBuf, workLevels[lvl], selfPop);
+      }
+      else if (selfPop == 0 && otherPop > 0) {
+        System.arraycopy(otherDoubleItemsArr, otherLevelsArr[lvl], workBuf, workLevels[lvl], otherPop);
+      }
+      else if (selfPop > 0 && otherPop > 0) {
+        mergeSortedDoubleArrays( //only workbuf is modified
             myCurDoubleItemsArr, myCurLevelsArr[lvl], selfPop,
             otherDoubleItemsArr, otherLevelsArr[lvl], otherPop,
-            workbuf, worklevels[lvl]);
+            workBuf, workLevels[lvl]);
       }
     }
   }

src/main/java/org/apache/datasketches/kll/KllDoublesSketch.java

@@ -276,8 +276,8 @@ public QuantilesDoublesSketchIterator iterator() {
   @Override
   public final void merge(final KllSketch other) {
     if (readOnly || sketchStructure != UPDATABLE) { throw new SketchesArgumentException(TGT_IS_READ_ONLY_MSG); }
-    final KllDoublesSketch othDblSk = (KllDoublesSketch)other;
-    if (othDblSk.isEmpty()) { return; }
+    final KllDoublesSketch othDblSk = (KllDoublesSketch)other; //check cast first
+    if (othDblSk.isEmpty()) { return; } //then check empty
     KllDoublesHelper.mergeDoubleImpl(this, othDblSk);
     kllDoublesSV = null;
   }
@@ -324,6 +324,19 @@ public void update(final double item) {
     kllDoublesSV = null;
   }
 
+  /**
+   * Updates this sketch with the given item the number of times specified by the given weight.
+   * @param item the item to be repeated. NaNs are ignored.
+   * @param weight the number of times the update of item is to be repeated. It must be ≥ one.
+   */
+  public void weightedUpdate(final double item, final int weight) {
+    if (readOnly) { throw new SketchesArgumentException(TGT_IS_READ_ONLY_MSG); }
+    if (weight < 1) { throw new SketchesArgumentException("Weight is less than one."); }
+    if (Double.isNaN(item)) { return; } //ignore
+    KllHeapDoublesSketch.weightedUpdateDouble(this, item, weight);
+    kllDoublesSV = null;
+  }
+
   //restricted
 
   /**

src/main/java/org/apache/datasketches/kll/KllHeapDoublesSketch.java

@@ -81,6 +81,25 @@ final class KllHeapDoublesSketch extends KllDoublesSketch {
     this.doubleItems = new double[k];
   }
 
+  /**
+   * Used for creating a temporary sketch for use with weighted updates.
+   */
+  KllHeapDoublesSketch(final int k, final int m, final double item, final int weight) {
+    super(UPDATABLE);
+    KllHelper.checkM(m);
+    KllHelper.checkK(k, m);
+    this.levelsArr = KllHelper.createLevelsArray(weight);
+    this.readOnly = false;
+    this.k = k;
+    this.m = m;
+    this.n = weight;
+    this.minK = k;
+    this.isLevelZeroSorted = false;
+    this.minDoubleItem = item;
+    this.maxDoubleItem = item;
+    this.doubleItems = KllDoublesHelper.createItemsArray(item, weight);
+  }
+
   /**
    * Heapify constructor.
    * @param srcMem Memory object that contains data serialized by this sketch.
@@ -282,4 +301,13 @@ void setNumLevels(final int numLevels) {
   @Override
   void setWritableMemory(final WritableMemory wmem) { }
 
+  static void weightedUpdateDouble(final KllDoublesSketch dblSk, final double item, final int weight) {
+    if (weight < dblSk.getLevelsArray(UPDATABLE)[0]) {
+      for (int i = 0; i < weight; i++) { dblSk.update(item); }
+    } else {
+      final KllHeapDoublesSketch tmpSk = new KllHeapDoublesSketch(dblSk.getK(), DEFAULT_M, item, weight);
+      dblSk.merge(tmpSk);
+    }
+  }
+
 }

src/main/java/org/apache/datasketches/kll/KllHelper.java

@@ -28,6 +28,7 @@
 import static java.lang.Math.pow;
 import static java.lang.Math.round;
 import static org.apache.datasketches.common.Family.KLL;
+import static org.apache.datasketches.common.Util.bitAt;
 import static org.apache.datasketches.common.Util.floorPowerOf2;
 import static org.apache.datasketches.kll.KllPreambleUtil.DATA_START_ADR;
 import static org.apache.datasketches.kll.KllPreambleUtil.EMPTY_BIT_MASK;
@@ -93,7 +94,7 @@ static class LevelStats {
   /**
    * This is the exact powers of 3 from 3^0 to 3^30 where the exponent is the index
    */
-  private static long[] powersOfThree =
+  static long[] powersOfThree =
       new long[] {1, 3, 9, 27, 81, 243, 729, 2187, 6561, 19683, 59049, 177147, 531441,
   1594323, 4782969, 14348907, 43046721, 129140163, 387420489, 1162261467,
   3486784401L, 10460353203L, 31381059609L, 94143178827L, 282429536481L,
@@ -150,6 +151,23 @@ public static long convertToCumulative(final long[] array) {
     return subtotal;
   }
 
+  /**
+   * Create the Levels Array from given weight
+   * Used with weighted update only.
+   * @param weight the given weight
+   * @return the Levels Array
+   */
+  static int[] createLevelsArray(final int weight) {
+    final int numLevels = 32 - Integer.numberOfLeadingZeros(weight);
+    final int[] levelsArr = new int[numLevels + 1]; //always one more than numLevels
+    int itemsArrIndex = 0;
+    levelsArr[0] = itemsArrIndex;
+    for (int level = 0; level < numLevels; level++) {
+      levelsArr[level + 1] = itemsArrIndex += bitAt(weight, level);
+    }
+    return levelsArr;
+  }
+
   static int currentLevelSizeItems(final int level, final int numLevels, final int[] levels) {
     if (level >= numLevels) { return 0; }
     return levels[level + 1] - levels[level];
@@ -180,7 +198,9 @@ static LevelStats getFinalSketchStatsAtNumLevels(
       printf("%6s %8s %12s %18s %18s\n", "Level", "Items", "CumItems", "N at Level", "CumN");
     }
     for (int level = 0; level < numLevels; level++) {
-      final LevelStats lvlStats = getLevelCapacityItems(k, m, numLevels, level);
+      final int items = KllHelper.levelCapacity(k, numLevels, level, m);
+      final long n = (long)items << level;
+      final LevelStats lvlStats = new LevelStats(n, numLevels, items);
       cumItems += lvlStats.numItems;
       cumN += lvlStats.n;
       if (printSketchStructure) {
@@ -257,24 +277,6 @@ static int getKFromEpsilon(final double epsilon, final boolean pmf) {
     return max(KllSketch.MIN_M, min(KllSketch.MAX_K, k));
   }
 
-  /**
-   * Given k, m, numLevels, this computes the item capacity of a single level.
-   * @param k the given user sketch configuration parameter
-   * @param m the given user sketch configuration parameter
-   * @param numLevels the given number of levels of the sketch
-   * @param level the specific level to compute its item capacity
-   * @return LevelStats with the computed N and items for the given level.
-   */
-  static LevelStats getLevelCapacityItems(
-      final int k,
-      final int m,
-      final int numLevels,
-      final int level) {
-    final int items = KllHelper.levelCapacity(k, numLevels, level, m);
-    final long n = (long)items << level;
-    return new LevelStats(n, numLevels, items);
-  }
-
   /**
    * Gets the normalized rank error given k and pmf.
    * Static method version of the <i>getNormalizedRankError(boolean)</i>.
@@ -696,7 +698,7 @@ static int findLevelToCompact(final int k, final int m, final int numLevels, fin
    * @param depth the zero-based index of the level being computed.
    * @return the actual capacity of a given level given its depth index.
    */
-  private static long intCapAux(final int k, final int depth) {
+  static long intCapAux(final int k, final int depth) {
     if (depth <= 30) { return intCapAuxAux(k, depth); }
     final int half = depth / 2;
     final int rest = depth - half;
@@ -710,7 +712,7 @@ private static long intCapAux(final int k, final int depth) {
    * @param depth the zero-based index of the level being computed. The max depth is 30!
    * @return the actual capacity of a given level given its depth index.
    */
-  private static long intCapAuxAux(final long k, final int depth) {
+  static long intCapAuxAux(final long k, final int depth) {
     final long twok = k << 1; // for rounding at the end, pre-multiply by 2 here, divide by 2 during rounding.
     final long tmp = ((twok << depth) / powersOfThree[depth]); //2k* (2/3)^depth. 2k also keeps the fraction larger.
     final long result = ((tmp + 1L) >>> 1); // (tmp + 1)/2. If odd, round up. This guarantees an integer.

src/main/java/org/apache/datasketches/kll/KllSketch.java

@@ -113,7 +113,7 @@ public abstract class KllSketch implements QuantilesAPI {
   final SketchType sketchType;
   final SketchStructure sketchStructure;
   boolean readOnly;
-  int[] levelsArr; //Always writable form
+  int[] levelsArr; //Always updatable form
 
   /**
    * Constructor for on-heap and off-heap.

src/main/java/org/apache/datasketches/quantilescommon/QuantilesDoublesAPI.java

@@ -293,7 +293,7 @@ default double[] getRanks(double[] quantiles) {
 
   /**
    * Updates this sketch with the given item.
-   * @param item from a stream of quantiles. NaNs are ignored.
+   * @param item from a stream of items. NaNs are ignored.
    */
   void update(double item);