Support floating point observations in Sketch

To support both long and double, copy provided values to an internal
double buffer (possibly with conversion), and build sketch from
that. This also removes the need for the caller to copy the values.

Author: vickenty

dogstatsd-http-core/src/main/java/com/datadoghq/dogstatsd/Sketch.java

@@ -34,6 +34,8 @@ public class Sketch {
     int size;
     int head;
 
+    private double[] values = new double[0];
+
     double min;
     double max;
     double sum;
@@ -110,40 +112,66 @@ public long count() {
     }
 
     /**
-     * Builds the sketch from the given values. The {@code values} array is modified in place
-     * (sorted); callers that need to preserve the original ordering should pass a copy.
+     * Builds the sketch from the given values.
      *
-     * @param values the observations to include in the sketch; sorted in place
-     * @param sampleRate the sampling rate used to collect {@code values}, in {@code (0, 1]}. Each
+     * @param observations the observations to include in the sketch
+     * @param sampleRate the sampling rate used to collect {@code observations}, in {@code (0, 1]}. Each
      *     observation is weighted by {@code 1 / sampleRate} when accumulating counts and sums.
      *     Rates below ~1.08e-19 saturate the per-observation weight; bin counts and the total
      *     {@code count} field saturate at {@link Long#MAX_VALUE} on overflow.
      */
-    public void build(long[] values, double sampleRate) {
-        if (Double.isNaN(sampleRate) || sampleRate <= 0 || sampleRate > 1) {
-            throw new IllegalArgumentException("sampleRate is out of range");
+    public void build(long[] observations, double sampleRate) {
+        validateSampleRate(sampleRate);
+        reset();
+        if (observations == null || observations.length == 0) {
+            return;
+        }
+        ensureCapacity(observations.length);
+        for (int i = 0; i < observations.length; i++) {
+            values[i] = observations[i];
         }
+        buildInner(observations.length, sampleRate);
+    }
 
+    /**
+     * Builds the sketch from the given values.
+     *
+     * @param observations the observations to include in the sketch
+     * @param sampleRate the sampling rate used to collect {@code observations}, in {@code (0, 1]}. Each
+     *     observation is weighted by {@code 1 / sampleRate} when accumulating counts and sums.
+     *     Rates below ~1.08e-19 saturate the per-observation weight; bin counts and the total
+     *     {@code count} field saturate at {@link Long#MAX_VALUE} on overflow.
+     */
+    public void build(double[] observations, double sampleRate) {
+        validateSampleRate(sampleRate);
         reset();
-        buildInner(values, sampleRate);
+        if (observations == null || observations.length == 0) {
+            return;
+        }
+        ensureCapacity(observations.length);
+        System.arraycopy(observations, 0, values, 0, observations.length);
+        buildInner(observations.length, sampleRate);
     }
 
-    private void buildInner(final long[] values, double sampleRate) {
-        if (values == null || values.length == 0) {
-            return;
+    private static void validateSampleRate(double sampleRate) {
+        if (Double.isNaN(sampleRate) || sampleRate <= 0 || sampleRate > 1) {
+            throw new IllegalArgumentException("sampleRate is out of range");
         }
+    }
 
-        Arrays.sort(values);
+    private void buildInner(int length, double sampleRate) {
+        Arrays.sort(values, 0, length);
 
         final long sampleSize = (long) (1 / sampleRate);
         min = values[0];
-        max = values[values.length - 1];
-        count = satMul(sampleSize, values.length);
+        max = values[length - 1];
+        count = satMul(sampleSize, length);
 
         short topKey = negInfKey - 1;
         long topCount = 0;
 
-        for (long val : values) {
+        for (int i = 0; i < length; i++) {
+            double val = values[i];
             sum += val / sampleRate;
 
             short key = key(val);
@@ -162,6 +190,12 @@ private void buildInner(final long[] values, double sampleRate) {
         append(topKey, topCount);
     }
 
+    private void ensureCapacity(int needed) {
+        if (values.length < needed) {
+            values = new double[needed];
+        }
+    }
+
     private void reset() {
         min = 0;
         max = 0;

dogstatsd-http-core/src/test/java/com/datadoghq/dogstatsd/SketchTest.java

@@ -11,6 +11,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
 
+import java.util.Arrays;
 import org.junit.Test;
 import org.junit.function.ThrowingRunnable;
 
@@ -47,7 +48,7 @@ public void keys() {
 
     @Test
     public void basic() {
-        s.build(null, 1);
+        s.build((long[]) null, 1);
         assertEquals(0, s.min(), 0);
         assertEquals(0, s.max(), 0);
         assertEquals(0, s.sum(), 0);
@@ -112,11 +113,13 @@ public void binMerge() {
             values[i] = val;
             values[i + values.length / 2] = -val;
         }
+        long[] sortedValues = values.clone();
+        Arrays.sort(sortedValues);
         s.build(values, 1);
-        assertEquals(values[0], s.min(), 0);
-        assertEquals(values[values.length - 1], s.max(), 0);
+        assertEquals(sortedValues[0], s.min(), 0);
+        assertEquals(sortedValues[sortedValues.length - 1], s.max(), 0);
 
-        final short foldedKey = Sketch.key(values[4]);
+        final short foldedKey = Sketch.key(sortedValues[4]);
 
         s.bins(
                 new Sketch.BinConsumer() {