FloatHistogram.Add/Sub: handle any schema change

Signed-off-by: Linas Medziunas <linas.medziunas@gmail.com>

model/histogram/float_histogram.go

@@ -231,11 +231,8 @@ func (h *FloatHistogram) Div(scalar float64) *FloatHistogram {
 // resulting histogram might have buckets with a population of zero or directly
 // adjacent spans (offset=0). To normalize those, call the Compact method.
 //
-// The method reconciles differences in the zero threshold and in the schema,
+// The method reconciles differences in the zero threshold and in the schema, and
-// but the schema of the other histogram must be ≥ the schema of the receiving
+// changes them if needed. The other histogram will not be modified in any case.
-// histogram (i.e. must have an equal or higher resolution). This means that the
-// schema of the receiving histogram won't change. Its zero threshold, however,
-// will change if needed. The other histogram will not be modified in any case.
 //
 // This method returns a pointer to the receiving histogram for convenience.
 func (h *FloatHistogram) Add(other *FloatHistogram) *FloatHistogram {
@@ -269,21 +266,30 @@ func (h *FloatHistogram) Add(other *FloatHistogram) *FloatHistogram {
 	h.Sum += other.Sum
 
 	var (
+		hPositiveSpans   = h.PositiveSpans
+		hPositiveBuckets = h.PositiveBuckets
+		hNegativeSpans   = h.NegativeSpans
+		hNegativeBuckets = h.NegativeBuckets
+
 		otherPositiveSpans   = other.PositiveSpans
 		otherPositiveBuckets = other.PositiveBuckets
 		otherNegativeSpans   = other.NegativeSpans
 		otherNegativeBuckets = other.NegativeBuckets
 	)
 
-	if other.Schema < h.Schema {
+	switch {
-		panic(fmt.Errorf("cannot add histogram with schema %d to %d", other.Schema, h.Schema))
+	case other.Schema < h.Schema:
-	} else if other.Schema > h.Schema {
+		hPositiveSpans, hPositiveBuckets = reduceResolution(hPositiveSpans, hPositiveBuckets, h.Schema, other.Schema, false, true)
+		hNegativeSpans, hNegativeBuckets = reduceResolution(hNegativeSpans, hNegativeBuckets, h.Schema, other.Schema, false, true)
+		h.Schema = other.Schema
+
+	case other.Schema > h.Schema:
 		otherPositiveSpans, otherPositiveBuckets = reduceResolution(otherPositiveSpans, otherPositiveBuckets, other.Schema, h.Schema, false, false)
 		otherNegativeSpans, otherNegativeBuckets = reduceResolution(otherNegativeSpans, otherNegativeBuckets, other.Schema, h.Schema, false, false)
 	}
 
-	h.PositiveSpans, h.PositiveBuckets = addBuckets(h.Schema, h.ZeroThreshold, false, h.PositiveSpans, h.PositiveBuckets, otherPositiveSpans, otherPositiveBuckets)
+	h.PositiveSpans, h.PositiveBuckets = addBuckets(h.Schema, h.ZeroThreshold, false, hPositiveSpans, hPositiveBuckets, otherPositiveSpans, otherPositiveBuckets)
-	h.NegativeSpans, h.NegativeBuckets = addBuckets(h.Schema, h.ZeroThreshold, false, h.NegativeSpans, h.NegativeBuckets, otherNegativeSpans, otherNegativeBuckets)
+	h.NegativeSpans, h.NegativeBuckets = addBuckets(h.Schema, h.ZeroThreshold, false, hNegativeSpans, hNegativeBuckets, otherNegativeSpans, otherNegativeBuckets)
 
 	return h
 }
@@ -296,21 +302,30 @@ func (h *FloatHistogram) Sub(other *FloatHistogram) *FloatHistogram {
 	h.Sum -= other.Sum
 
 	var (
+		hPositiveSpans   = h.PositiveSpans
+		hPositiveBuckets = h.PositiveBuckets
+		hNegativeSpans   = h.NegativeSpans
+		hNegativeBuckets = h.NegativeBuckets
+
 		otherPositiveSpans   = other.PositiveSpans
 		otherPositiveBuckets = other.PositiveBuckets
 		otherNegativeSpans   = other.NegativeSpans
 		otherNegativeBuckets = other.NegativeBuckets
 	)
 
-	if other.Schema < h.Schema {
+	switch {
-		panic(fmt.Errorf("cannot subtract histigram with schema %d to %d", other.Schema, h.Schema))
+	case other.Schema < h.Schema:
-	} else if other.Schema > h.Schema {
+		hPositiveSpans, hPositiveBuckets = reduceResolution(hPositiveSpans, hPositiveBuckets, h.Schema, other.Schema, false, true)
+		hNegativeSpans, hNegativeBuckets = reduceResolution(hNegativeSpans, hNegativeBuckets, h.Schema, other.Schema, false, true)
+		h.Schema = other.Schema
+
+	case other.Schema > h.Schema:
 		otherPositiveSpans, otherPositiveBuckets = reduceResolution(otherPositiveSpans, otherPositiveBuckets, other.Schema, h.Schema, false, false)
 		otherNegativeSpans, otherNegativeBuckets = reduceResolution(otherNegativeSpans, otherNegativeBuckets, other.Schema, h.Schema, false, false)
 	}
 
-	h.PositiveSpans, h.PositiveBuckets = addBuckets(h.Schema, h.ZeroThreshold, true, h.PositiveSpans, h.PositiveBuckets, otherPositiveSpans, otherPositiveBuckets)
+	h.PositiveSpans, h.PositiveBuckets = addBuckets(h.Schema, h.ZeroThreshold, true, hPositiveSpans, hPositiveBuckets, otherPositiveSpans, otherPositiveBuckets)
-	h.NegativeSpans, h.NegativeBuckets = addBuckets(h.Schema, h.ZeroThreshold, true, h.NegativeSpans, h.NegativeBuckets, otherNegativeSpans, otherNegativeBuckets)
+	h.NegativeSpans, h.NegativeBuckets = addBuckets(h.Schema, h.ZeroThreshold, true, hNegativeSpans, hNegativeBuckets, otherNegativeSpans, otherNegativeBuckets)
 
 	return h
 }

model/histogram/float_histogram_test.go

@@ -1242,7 +1242,7 @@ func TestFloatHistogramAdd(t *testing.T) {
 				Sum:             1.234,
 				PositiveSpans:   []Span{{0, 2}, {3, 3}},
 				PositiveBuckets: []float64{5, 4, 2, 3, 6},
-				NegativeSpans:   []Span{{-9, 2}, {3, 2}},
+				NegativeSpans:   []Span{{-6, 2}, {1, 2}},
 				NegativeBuckets: []float64{1, 1, 4, 4},
 			},
 			&FloatHistogram{
@@ -1262,7 +1262,7 @@ func TestFloatHistogramAdd(t *testing.T) {
 				Sum:             3.579,
 				PositiveSpans:   []Span{{-2, 2}, {0, 5}, {0, 3}},
 				PositiveBuckets: []float64{1, 0, 5, 4, 3, 4, 7, 2, 3, 6},
-				NegativeSpans:   []Span{{-9, 2}, {3, 2}, {5, 2}, {3, 2}},
+				NegativeSpans:   []Span{{-6, 2}, {1, 2}, {4, 2}, {3, 2}},
 				NegativeBuckets: []float64{1, 1, 4, 4, 3, 1, 5, 6},
 			},
 		},
@@ -1573,16 +1573,33 @@ func TestFloatHistogramAdd(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			in2Copy := c.in2.Copy()
+			testHistogramAdd(t, c.in1, c.in2, c.expected)
-			require.Equal(t, c.expected, c.in1.Add(c.in2))
+			testHistogramAdd(t, c.in2, c.in1, c.expected)
-			// Has it also happened in-place?
-			require.Equal(t, c.expected, c.in1)
-			// Check that the argument was not mutated.
-			require.Equal(t, in2Copy, c.in2)
 		})
 	}
 }
 
+func testHistogramAdd(t *testing.T, a, b, expected *FloatHistogram) {
+	var (
+		aCopy        = a.Copy()
+		bCopy        = b.Copy()
+		expectedCopy = expected.Copy()
+	)
+
+	res := aCopy.Add(bCopy)
+
+	res.Compact(0)
+	expectedCopy.Compact(0)
+
+	require.Equal(t, expectedCopy, res)
+
+	// Has it also happened in-place?
+	require.Equal(t, expectedCopy, aCopy)
+
+	// Check that the argument was not mutated.
+	require.Equal(t, b, bCopy)
+}
+
 func TestFloatHistogramSub(t *testing.T) {
 	// This has fewer test cases than TestFloatHistogramAdd because Add and
 	// Sub share most of the trickier code.
@@ -1662,16 +1679,35 @@ func TestFloatHistogramSub(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			in2Copy := c.in2.Copy()
+			testFloatHistogramSub(t, c.in1, c.in2, c.expected)
-			require.Equal(t, c.expected, c.in1.Sub(c.in2))
+
-			// Has it also happened in-place?
+			expectedNegative := c.expected.Copy().Mul(-1)
-			require.Equal(t, c.expected, c.in1)
+			testFloatHistogramSub(t, c.in2, c.in1, expectedNegative)
-			// Check that the argument was not mutated.
-			require.Equal(t, in2Copy, c.in2)
 		})
 	}
 }
 
+func testFloatHistogramSub(t *testing.T, a, b, expected *FloatHistogram) {
+	var (
+		aCopy        = a.Copy()
+		bCopy        = b.Copy()
+		expectedCopy = expected.Copy()
+	)
+
+	res := aCopy.Sub(bCopy)
+
+	res.Compact(0)
+	expectedCopy.Compact(0)
+
+	require.Equal(t, expectedCopy, res)
+
+	// Has it also happened in-place?
+	require.Equal(t, expectedCopy, aCopy)
+
+	// Check that the argument was not mutated.
+	require.Equal(t, b, bCopy)
+}
+
 func TestFloatHistogramCopyToSchema(t *testing.T) {
 	cases := []struct {
 		name         string