[processor/delta_to_cumulative] Fix exponential histogram bucket limit rounding

.chloggen/delta-to-cumulative-bucket-limit-rounding.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: processor/delta_to_cumulative
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix exponential histogram downscaling exceeding the bucket limit due to incorrect index rounding.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [50923]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

processor/deltatocumulativeprocessor/internal/data/expo/scale.go

@@ -139,10 +139,11 @@ func Limit(maxBuckets int, scale Scale, arel, brel pmetric.ExponentialHistogramD
 		up--
 	}
 
-	// Keep downscaling until the number of buckets is within the limit.
+	// Match Collapse: round the lower index down and the exclusive upper index
+	// up, including negative indexes, so the merged range stays within the limit.
 	for up-lo > maxBuckets {
-		lo /= 2
-		up /= 2
+		lo >>= 1
+		up = (up + 1) >> 1
 		scale--
 	}
 

processor/deltatocumulativeprocessor/internal/data/expo/scale_test.go

@@ -10,11 +10,65 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/deltatocumulativeprocessor/internal/data/expo"
 )
 
+func TestLimit(t *testing.T) {
+	t.Parallel()
+
+	const maxBuckets = 160
+	cases := []struct {
+		name        string
+		lo, hi      int32
+		wantScale   expo.Scale
+		wantBuckets int
+	}{
+		{name: "within_limit", lo: 1, hi: 160, wantScale: 0, wantBuckets: 160},
+		{name: "aligned", lo: 0, hi: 319, wantScale: -1, wantBuckets: 160},
+		{name: "positive_odd_upper", lo: 0, hi: 320, wantScale: -2, wantBuckets: 81},
+		{name: "positive_odd_bounds", lo: 1, hi: 320, wantScale: -2, wantBuckets: 81},
+		{name: "negative_odd_lower", lo: -321, hi: -2, wantScale: -2, wantBuckets: 81},
+		{name: "negative_odd_upper", lo: -320, hi: -2, wantScale: -1, wantBuckets: 160},
+		{name: "cross_zero", lo: -161, hi: 160, wantScale: -2, wantBuckets: 82},
+		{name: "multiple_scales", lo: 1, hi: 1280, wantScale: -4, wantBuckets: 81},
+	}
+	for _, cs := range cases {
+		t.Run(cs.name, func(t *testing.T) {
+			t.Parallel()
+
+			// Two observations at the endpoints require all intervening buckets.
+			a := pmetric.NewExponentialHistogramDataPointBuckets()
+			a.SetOffset(cs.lo)
+			a.BucketCounts().Append(1)
+			b := pmetric.NewExponentialHistogramDataPointBuckets()
+			b.SetOffset(cs.hi)
+			b.BucketCounts().Append(1)
+
+			to := expo.Limit(maxBuckets, 0, a, b)
+			assert.Equal(t, cs.wantScale, to)
+			assert.Equal(t, to, expo.Limit(maxBuckets, 0, b, a))
+
+			expo.Downscale(a, 0, to)
+			expo.Downscale(b, 0, to)
+			expo.Merge(a, b)
+
+			counts := a.BucketCounts()
+			assert.LessOrEqual(t, counts.Len(), maxBuckets)
+			require.Equal(t, cs.wantBuckets, counts.Len())
+			assert.Equal(t, uint64(1), counts.At(0))
+			assert.Equal(t, uint64(1), counts.At(counts.Len()-1))
+			var total uint64
+			for _, count := range counts.All() {
+				total += count
+			}
+			assert.Equal(t, uint64(2), total)
+		})
+	}
+}
+
 func TestDownscale(t *testing.T) {
 	type Repr[T any] struct {
 		scale expo.Scale