feat: implement exponential histogram merge

packages/sdk-metrics/src/aggregator/ExponentialHistogram.ts

@@ -149,6 +149,10 @@ export class ExponentialHistogramAccumulation implements Accumulation {
     return this._count;
   }
 
+  zeroCount(): number {
+    return this._zeroCount;
+  }
+
   scale(): number {
     if (this._count === this._zeroCount) {
       // all zeros! scale doesn't matter, use zero
@@ -213,9 +217,65 @@ export class ExponentialHistogramAccumulation implements Accumulation {
 
   /**
    * Merge combines data from other into self
-   * @param other
+   * @param {ExponentialHistogramAccumulation} other
    */
-  mergeFrom(_other: ExponentialHistogramAccumulation) {}
+  merge(other: ExponentialHistogramAccumulation) {
+    if(this._count === 0) {
+      this._min = other.min();
+      this._max = other.max();
+    } else if ( other.count() !== 0 ) {
+      if(other.min() < this.min()) {
+        this._min = other.min();
+      }
+      if(other.max() > this.max()) {
+        this._max = other.max();
+      }
+    }
+
+    this._sum += other.sum();
+    this._count += other.count();
+    this._zeroCount += other.zeroCount();
+
+    let minScale = Math.min(this.scale(), other.scale());
+
+    const ourHighLowPos = this.highLowAtScale(this.positive(), minScale);
+    const otherHighLowPos = other.highLowAtScale(other.positive(), minScale);
+    const highLowPos = {
+      low: Math.min(ourHighLowPos.low, otherHighLowPos.low),
+      high: Math.max(ourHighLowPos.high, otherHighLowPos.high),
+    };
+
+    const ourHighLowNeg = this.highLowAtScale(this.negative(), minScale);
+    const otherHighLowNeg = other.highLowAtScale(other.negative(), minScale);
+    const highLowNeg = {
+      low: Math.min(ourHighLowNeg.low, otherHighLowNeg.low),
+      high: Math.max(ourHighLowNeg.high, otherHighLowNeg.high),
+    };
+
+    minScale = Math.min(
+      minScale - this._changeScale(highLowPos, this._maxSize),
+      minScale - this._changeScale(highLowNeg, this._maxSize),
+    );
+
+    this._downscale(this.scale() - minScale);
+
+    this._mergeBuckets(this.positive(), other, other.positive(), minScale);
+    this._mergeBuckets(this.negative(), other, other.negative(), minScale);
+  }
+
+  highLowAtScale(buckets: Buckets, scale: number): HighLow {
+    if(buckets.length() === 0) {
+      return {
+        low: 0,
+        high: -1,
+      };
+    }
+    const shift = this.scale() - scale;
+    return {
+      low: util.rightShift(buckets.indexStart, shift),
+      high: util.rightShift(buckets.indexEnd, shift),
+    };
+  }
 
   // todo: rename?
   private _update(buckets: Buckets, value: number, increment: number) {
@@ -330,6 +390,28 @@ export class ExponentialHistogramAccumulation implements Accumulation {
     }
   }
 
+  private _mergeBuckets(
+    ours: Buckets,
+    other: ExponentialHistogramAccumulation,
+    theirs: Buckets,
+    scale: number
+  ) {
+    const theirOffset = theirs.offset();
+    const theirChange = other.scale() - scale;
+
+    for(let i=0; i < theirs.length(); i++) {
+      const result = this._incrementIndexby(
+        ours,
+        util.rightShift(theirOffset + i, theirChange),
+        theirs.at(i),
+      );
+      if( result.success ) {
+        // this should not happen
+        // todo: log
+      }
+    }
+  }
+
   // todo: delete, for debugging
   printBuckets() {
     console.log('positive:');

packages/sdk-metrics/test/aggregator/ExponentialHistogram.test.ts

@@ -20,6 +20,7 @@ import { Mapping } from '../../src/aggregator/exponential-histogram//mapping/typ
 import { ExponentMapping } from '../../src/aggregator/exponential-histogram//mapping/ExponentMapping';
 import { LogarithmMapping } from '../../src/aggregator/exponential-histogram/mapping/LogarithmMapping';
 import * as assert from 'assert';
+import { assertInEpsilon, assertInDelta } from './exponential-histogram/helpers';
 
 describe('ExponentialHistogramAccumulation', () => {
   describe('record', () => {
@@ -202,6 +203,122 @@ describe('ExponentialHistogramAccumulation', () => {
       });
     });
   });
+  describe('merge', () => {
+    it('handles simple (even) case', () => {
+      const acc0 = new ExponentialHistogramAccumulation([0, 0], 4);
+      const acc1 = new ExponentialHistogramAccumulation([0, 0], 4);
+      const acc2 = new ExponentialHistogramAccumulation([0, 0], 4);
+
+      for(let i = 0; i < 4; i++) {
+        const v1 = 2 << i;
+        const v2 = 1 / (1 << i);
+
+        acc0.record(v1);
+        acc1.record(v2);
+        acc2.record(v1);
+        acc2.record(v2);
+      }
+
+      assert.strictEqual(acc0.scale(), 0);
+      assert.strictEqual(acc1.scale(), 0);
+      assert.strictEqual(acc2.scale(), -1);
+
+      assert.strictEqual(acc0.positive().offset(), 0);
+      assert.strictEqual(acc1.positive().offset(), -4);
+      assert.strictEqual(acc2.positive().offset(), -2);
+
+      assert.deepStrictEqual(getCounts(acc0.positive()), [1, 1, 1, 1]);
+      assert.deepStrictEqual(getCounts(acc1.positive()), [1, 1, 1, 1]);
+      assert.deepStrictEqual(getCounts(acc2.positive()), [2, 2, 2, 2]);
+
+      acc0.merge(acc1);
+
+      assert.strictEqual(acc0.scale(), -1);
+      assert.strictEqual(acc2.scale(), -1);
+
+      assertHistogramsEqual(acc0, acc2);
+    });
+
+    it('handles simple (odd) case', () => {
+      const acc0 = new ExponentialHistogramAccumulation([0, 0], 4);
+      const acc1 = new ExponentialHistogramAccumulation([0, 0], 4);
+      const acc2 = new ExponentialHistogramAccumulation([0, 0], 4);
+
+      for(let i = 0; i < 4; i++) {
+        const v1 = 2 << i;
+        const v2 = 2 / (1 << i);
+
+        acc0.record(v1);
+        acc1.record(v2);
+        acc2.record(v1);
+        acc2.record(v2);
+      }
+
+      assert.strictEqual(acc0.count(), 4);
+      assert.strictEqual(acc1.count(), 4);
+      assert.strictEqual(acc2.count(), 8);
+
+      assert.strictEqual(acc0.scale(), 0);
+      assert.strictEqual(acc1.scale(), 0);
+      assert.strictEqual(acc2.scale(), -1);
+
+      assert.strictEqual(acc0.positive().offset(), 0);
+      assert.strictEqual(acc1.positive().offset(), -3);
+      assert.strictEqual(acc2.positive().offset(), -2);
+
+      assert.deepStrictEqual(getCounts(acc0.positive()), [1, 1, 1, 1]);
+      assert.deepStrictEqual(getCounts(acc1.positive()), [1, 1, 1, 1]);
+      assert.deepStrictEqual(getCounts(acc2.positive()), [1, 2, 3, 2]);
+
+      acc0.merge(acc1);
+
+      assert.strictEqual(acc0.scale(), -1);
+      assert.strictEqual(acc2.scale(), -1);
+
+      assertHistogramsEqual(acc0, acc2);
+    });
+
+    it('handles exhaustive test case', () => {
+      const testMergeExhaustive =  (a: number[], b: number[], size: number, incr: number) => {
+        const aHist = new ExponentialHistogramAccumulation([0, 0], size);
+        const bHist = new ExponentialHistogramAccumulation([0, 0], size);
+        const cHist = new ExponentialHistogramAccumulation([0, 0], size);
+
+        a.forEach(av => {
+          aHist.updateByIncrement(av, incr);
+          cHist.updateByIncrement(av, incr);
+        });
+        b.forEach(bv => {
+          bHist.updateByIncrement(bv, incr);
+          cHist.updateByIncrement(bv, incr);
+        });
+
+        aHist.merge(bHist);
+
+        assertHistogramsEqual(aHist, cHist);
+      }
+
+      const factor = 1024.0;
+      const count = 16;
+      const means = [0, factor];
+      const stddevs = [1, factor];
+      const sizes = [2, 6, 8, 9, 16];
+      const increments = [1, 0x100, 0x10000, 0x100000000];
+
+      for(let mean of means) {
+        for(let stddev of stddevs) {
+          const values = Array.from({length: count}, () => mean + Math.random() * stddev);
+          for(let part = 1; part < count; part++) {
+            for(let size of sizes) {
+              for (let incr of increments) {
+                testMergeExhaustive(values.slice(0, part), values.slice(part, count), size, incr);
+              }
+            }
+          }
+        }
+      }
+    });
+  });
 });
 
 function getCounts(buckets: Buckets): Array<number> {
@@ -225,3 +342,40 @@ function getMapping(scale: number): Mapping {
     return LogarithmMapping.get(scale);
   }
 }
+
+function assertHistogramsEqual(
+  actual: ExponentialHistogramAccumulation,
+  expected: ExponentialHistogramAccumulation,
+) {
+  const actualSum = actual.sum();
+  const expectedSum = expected.sum();
+
+  if(actualSum === 0 || expectedSum === 0) {
+    assertInDelta(actualSum, expectedSum, 1e-6);
+  } else {
+    assertInEpsilon(actualSum, expectedSum, 1e-6);
+  }
+
+  assert.strictEqual(actual.count(), expected.count());
+  assert.strictEqual(actual.zeroCount(), expected.zeroCount());
+  assert.strictEqual(actual.scale(), expected.scale());
+
+  assert.strictEqual(
+    bucketsToString(actual.positive()),
+    bucketsToString(expected.positive())
+  );
+
+  assert.strictEqual(
+    bucketsToString(actual.negative()),
+    bucketsToString(expected.negative())
+  );
+}
+
+function bucketsToString(buckets: Buckets): string {
+  let str = `[@${buckets.offset()}`;
+  for(let i = 0; i < buckets.length(); i++) {
+    str += buckets.at(i).toString();
+  }
+  str += ']\n';
+  return str;
+}

packages/sdk-metrics/test/aggregator/exponential-histogram/helpers.ts

@@ -31,3 +31,14 @@ export function assertInEpsilon(
     `expected relative error: ${relErr} to be < ${epsilon}`
   );
 }
+
+export function assertInDelta(
+  actual: number,
+  expected: number,
+  delta: number) {
+  const actualDelta = Math.abs(expected - actual);
+  assert.ok(
+    actualDelta < delta,
+    `expected delta: ${delta} to be < ${actualDelta}`
+  );
+}