Merge pull request #185 from DataSketches/kll-floats-sketch

KllFloatsSketch

Author: leerho

src/main/java/com/yahoo/sketches/ByteArrayUtil.java

@@ -2,6 +2,19 @@
 
 public final class ByteArrayUtil {
 
+  /**
+   * Put a given short value into a given byte array at a given offset.
+   * Assumes little-endian byte order.
+   * @param bytes destination byte array
+   * @param offset destination offset
+   * @param value source value
+   */
+  public static void putShort(final byte[] bytes, final int offset, final short value) {
+    for (int i = 0; i < Short.BYTES; i++) {
+      bytes[offset + i] = (byte) ((value >>> (8 * i)) & 0xff);
+    }
+  }
+
   /**
    * Put a given integer value into a given byte array at a given offset.
    * Assumes little-endian byte order.

src/main/java/com/yahoo/sketches/Family.java

@@ -111,8 +111,12 @@ public enum Family {
   /**
    * The VarOpt family of sketches. (Not part of TSF.)
    */
-  VAROPT_UNION(14, "VAROPT_UNION", 1, 4);
+  VAROPT_UNION(14, "VAROPT_UNION", 1, 4),
 
+  /**
+   * KLL quanliles sketch
+   */
+  KLL(15, "KLL", 1, 2);
 
   private static final Map<Integer, Family> lookupID = new HashMap<Integer, Family>();
   private static final Map<String, Family> lookupFamName = new HashMap<String, Family>();

src/main/java/com/yahoo/sketches/QuantilesHelper.java

@@ -0,0 +1,85 @@
+package com.yahoo.sketches;
+
+/**
+ * Common static methods for quantiles sketches
+ */
+public class QuantilesHelper {
+
+  /**
+   * Convert the weights into totals of the weights preceding each item
+   * @param array of weights
+   * @return total weight
+   */
+  public static long convertToPrecedingCummulative(final long[] array) {
+    long subtotal = 0;
+    for (int i = 0; i < array.length; i++) {
+      final long newSubtotal = subtotal + array[i];
+      array[i] = subtotal;
+      subtotal = newSubtotal;
+    }
+    return subtotal;
+  }
+
+  /**
+   * Returns the zero-based index (position) of a value in the hypothetical sorted stream of
+   * values of size n.
+   * @param phi the fractional position where: 0 &le; &#966; &le; 1.0.
+   * @param n the size of the stream
+   * @return the index, a value between 0 and n-1.
+   */
+  public static long posOfPhi(final double phi, final long n) {
+    final long pos = (long) Math.floor(phi * n);
+    return (pos == n) ? n - 1 : pos;
+  }
+
+  /**
+   * This is written in terms of a plain array to facilitate testing.
+   * @param arr the chunk containing the position
+   * @param pos the position
+   * @return the index of the chunk containing the position
+   */
+  public static int chunkContainingPos(final long[] arr, final long pos) {
+    final int nominalLength = arr.length - 1; /* remember, arr contains an "extra" position */
+    assert nominalLength > 0;
+    final long n = arr[nominalLength];
+    assert 0 <= pos;
+    assert pos < n;
+    final int l = 0;
+    final int r = nominalLength;
+    // the following three asserts should probably be retained since they ensure
+    // that the necessary invariants hold at the beginning of the search
+    assert l < r;
+    assert arr[l] <= pos;
+    assert pos < arr[r];
+    return searchForChunkContainingPos(arr, pos, l, r);
+  }
+
+  // Let m_i denote the minimum position of the length=n "full" sorted sequence
+  //   that is represented in slot i of the length = n "chunked" sorted sequence.
+  //
+  // Note that m_i is the same thing as auxCumWtsArr_[i]
+  //
+  // Then the answer to a positional query 0 <= q < n is l, where 0 <= l < len,
+  // A)  m_l <= q
+  // B)   q  < m_r
+  // C)   l+1 = r
+  //
+  // A) and B) provide the invariants for our binary search.
+  // Observe that they are satisfied by the initial conditions:  l = 0 and r = len.
+  private static int searchForChunkContainingPos(final long[] arr, final long pos, final int l, final int r) {
+    // the following three asserts can probably go away eventually, since it is fairly clear
+    // that if these invariants hold at the beginning of the search, they will be maintained
+    assert l < r;
+    assert arr[l] <= pos;
+    assert pos < arr[r];
+    if (l + 1 == r) {
+      return l;
+    }
+    final int m = l + (r - l) / 2;
+    if (arr[m] <= pos) {
+      return (searchForChunkContainingPos(arr, pos, m, r));
+    }
+    return (searchForChunkContainingPos(arr, pos, l, m));
+  }
+
+}

src/main/java/com/yahoo/sketches/kll/KllFloatsQuantileCalculator.java

@@ -0,0 +1,136 @@
+/*
+ * Copyright 2018, Yahoo! Inc. Licensed under the terms of the
+ * Apache License 2.0. See LICENSE file at the project root for terms.
+ */
+
+package com.yahoo.sketches.kll;
+
+import java.util.Arrays;
+
+import com.yahoo.sketches.QuantilesHelper;
+
+/**
+ * Data structure for answering quantile queries based on the samples from KllSketch
+ * @author Kevin Lang
+ * @author Alexander Saydakov
+ */
+final class KllFloatsQuantileCalculator {
+
+  private long n_;
+  private float[] items_;
+  private long[] weights_;
+  private int[] levels_;
+  private int numLevels_;
+
+  // assumes that all levels are sorted including level 0
+  KllFloatsQuantileCalculator(final float[] items, final int[] levels, final int numLevels,
+      final long n) {
+    n_ = n;
+    final int numItems = levels[numLevels] - levels[0];
+    items_ = new float[numItems];
+    weights_ = new long[numItems + 1]; // one more is intentional
+    levels_ = new int[numLevels + 1];
+    populateFromSketch(items, levels, numLevels, numItems);
+    blockyTandemMergeSort(items_, weights_, levels_, numLevels_);
+    QuantilesHelper.convertToPrecedingCummulative(weights_);
+  }
+
+  float getQuantile(final double phi) {
+    final long pos = QuantilesHelper.posOfPhi(phi, n_);
+    return approximatelyAnswerPositonalQuery(pos);
+  }
+
+  private float approximatelyAnswerPositonalQuery(final long pos) {
+    assert pos >= 0;
+    assert pos < n_;
+    final int index = QuantilesHelper.chunkContainingPos(weights_, pos);
+    return items_[index];
+  }
+
+  private void populateFromSketch(final float[] srcItems, final int[] srcLevels,
+      final int numLevels, final int numItems) {
+    final int offset = srcLevels[0];
+    System.arraycopy(srcItems, offset, items_, 0, numItems);
+    int srcLevel = 0;
+    int dstLevel = 0;
+    int weight = 1;
+    while (srcLevel < numLevels) {
+      final int fromIndex = srcLevels[srcLevel] - offset;
+      final int toIndex = srcLevels[srcLevel + 1] - offset; // exclusive
+      if (fromIndex < toIndex) { // skip empty levels
+        Arrays.fill(weights_, fromIndex, toIndex, weight);
+        levels_[dstLevel] = fromIndex;
+        levels_[dstLevel + 1] = toIndex;
+        dstLevel++;
+      }
+      srcLevel++;
+      weight *= 2;
+    }
+    weights_[numItems] = 0;
+    numLevels_ = dstLevel;
+  }
+
+  private static void blockyTandemMergeSort(final float[] items, final long[] weights,
+      final int[] levels, final int numLevels) {
+    if (numLevels == 1) { return; }
+
+    // duplicate the input in preparation for the "ping-pong" copy reduction strategy.
+    final float[] itemsTmp = Arrays.copyOf(items, items.length);
+    final long[] weightsTmp = Arrays.copyOf(weights, items.length); // don't need the extra one here
+
+    blockyTandemMergeSortRecursion(itemsTmp, weightsTmp, items, weights, levels, 0, numLevels);
+  }
+
+  private static void blockyTandemMergeSortRecursion(final float[] itemsSrc, final long[] weightsSrc,
+      final float[] itemsDst, final long[] weightsDst, final int[] levels, final int startingLevel,
+      final int numLevels) {
+    if (numLevels == 1) { return; }
+    final int numLevels1 = numLevels / 2;
+    final int numLevels2 = numLevels - numLevels1;
+    assert numLevels1 >= 1;
+    assert numLevels2 >= numLevels1;
+    final int startingLevel1 = startingLevel;
+    final int startingLevel2 = startingLevel + numLevels1;
+    // swap roles of src and dst
+    blockyTandemMergeSortRecursion(itemsDst, weightsDst, itemsSrc, weightsSrc, levels,
+        startingLevel1, numLevels1);
+    blockyTandemMergeSortRecursion(itemsDst, weightsDst, itemsSrc, weightsSrc, levels,
+        startingLevel2, numLevels2);
+    tandemMerge(itemsSrc, weightsSrc, itemsDst, weightsDst, levels, startingLevel1, numLevels1,
+        startingLevel2, numLevels2);
+  }
+
+  private static void tandemMerge(final float[] itemsSrc, final long[] weightsSrc,
+      final float[] itemsDst, final long[] weightsDst,
+      final int[] levelStarts, final int startingLevel1, final int numLevels1,
+      final int startingLevel2, final int numLevels2) {
+    final int fromIndex1 = levelStarts[startingLevel1];
+    final int toIndex1 = levelStarts[startingLevel1 + numLevels1]; // exclusive
+    final int fromIndex2 = levelStarts[startingLevel2];
+    final int toIndex2 = levelStarts[startingLevel2 + numLevels2]; // exclusive
+    int iSrc1 = fromIndex1;
+    int iSrc2 = fromIndex2;
+    int iDst = fromIndex1;
+
+    while ((iSrc1 < toIndex1) && (iSrc2 < toIndex2)) {
+      if (itemsSrc[iSrc1] < itemsSrc[iSrc2]) {
+        itemsDst[iDst] = itemsSrc[iSrc1];
+        weightsDst[iDst] = weightsSrc[iSrc1];
+        iSrc1++;
+      } else {
+        itemsDst[iDst] = itemsSrc[iSrc2];
+        weightsDst[iDst] = weightsSrc[iSrc2];
+        iSrc2++;
+      }
+      iDst++;
+    }
+    if (iSrc1 < toIndex1) {
+      System.arraycopy(itemsSrc, iSrc1, itemsDst, iDst, toIndex1 - iSrc1);
+      System.arraycopy(weightsSrc, iSrc1, weightsDst, iDst, toIndex1 - iSrc1);
+    } else if (iSrc2 < toIndex2) {
+      System.arraycopy(itemsSrc, iSrc2, itemsDst, iDst, toIndex2 - iSrc2);
+      System.arraycopy(weightsSrc, iSrc2, weightsDst, iDst, toIndex2 - iSrc2);
+    }
+  }
+
+}