Use the label.Set.Equivalent value instead of an encoding in the batcher (#658)

exporters/metric/prometheus/prometheus.go

@@ -159,7 +159,7 @@ func NewExportPipeline(config Config, period time.Duration) (*push.Controller, h
 	// it could try again on the next scrape and no data would be lost, only resolution.
 	//
 	// Gauges (or LastValues) and Summaries are an exception to this and have different behaviors.
-	batcher := ungrouped.New(selector, label.DefaultEncoder(), true)
+	batcher := ungrouped.New(selector, true)
 	pusher := push.New(batcher, exporter, period)
 	pusher.Start()
 

exporters/metric/stdout/stdout.go

@@ -137,7 +137,7 @@ func NewExportPipeline(config Config, period time.Duration) (*push.Controller, e
 	if err != nil {
 		return nil, err
 	}
-	batcher := ungrouped.New(selector, exporter.config.LabelEncoder, true)
+	batcher := ungrouped.New(selector, true)
 	pusher := push.New(batcher, exporter, period)
 	pusher.Start()
 

exporters/otlp/otlp_integration_test.go

@@ -28,7 +28,6 @@ import (
 
 	"go.opentelemetry.io/otel/api/core"
 	"go.opentelemetry.io/otel/api/key"
-	"go.opentelemetry.io/otel/api/label"
 	"go.opentelemetry.io/otel/api/metric"
 	metricapi "go.opentelemetry.io/otel/api/metric"
 	"go.opentelemetry.io/otel/exporters/otlp"
@@ -112,7 +111,7 @@ func newExporterEndToEndTest(t *testing.T, additionalOpts []otlp.ExporterOption)
 	}
 
 	selector := simple.NewWithExactMeasure()
-	batcher := ungrouped.New(selector, label.DefaultEncoder(), true)
+	batcher := ungrouped.New(selector, true)
 	pusher := push.New(batcher, exp, 60*time.Second)
 	pusher.Start()
 

sdk/metric/batcher/ungrouped/ungrouped.go

@@ -26,15 +26,14 @@ import (
 
 type (
 	Batcher struct {
-		selector     export.AggregationSelector
-		batchMap     batchMap
-		stateful     bool
-		labelEncoder label.Encoder
+		selector export.AggregationSelector
+		batchMap batchMap
+		stateful bool
 	}
 
 	batchKey struct {
 		descriptor *metric.Descriptor
-		encoded    string
+		distinct   label.Distinct
 	}
 
 	batchValue struct {
@@ -48,12 +47,11 @@ type (
 var _ export.Batcher = &Batcher{}
 var _ export.CheckpointSet = batchMap{}
 
-func New(selector export.AggregationSelector, labelEncoder label.Encoder, stateful bool) *Batcher {
+func New(selector export.AggregationSelector, stateful bool) *Batcher {
 	return &Batcher{
-		selector:     selector,
-		batchMap:     batchMap{},
-		stateful:     stateful,
-		labelEncoder: labelEncoder,
+		selector: selector,
+		batchMap: batchMap{},
+		stateful: stateful,
 	}
 }
 
@@ -63,10 +61,9 @@ func (b *Batcher) AggregatorFor(descriptor *metric.Descriptor) export.Aggregator
 
 func (b *Batcher) Process(_ context.Context, record export.Record) error {
 	desc := record.Descriptor()
-	encoded := record.Labels().Encoded(b.labelEncoder)
 	key := batchKey{
 		descriptor: desc,
-		encoded:    encoded,
+		distinct:   record.Labels().Equivalent(),
 	}
 	agg := record.Aggregator()
 	value, ok := b.batchMap[key]

sdk/metric/batcher/ungrouped/ungrouped_test.go

@@ -30,7 +30,7 @@ import (
 
 func TestUngroupedStateless(t *testing.T) {
 	ctx := context.Background()
-	b := ungrouped.New(test.NewAggregationSelector(), test.SdkEncoder, false)
+	b := ungrouped.New(test.NewAggregationSelector(), false)
 
 	// Set initial lastValue values
 	_ = b.Process(ctx, test.NewLastValueRecord(&test.LastValueADesc, test.Labels1, 10))
@@ -92,7 +92,7 @@ func TestUngroupedStateless(t *testing.T) {
 
 func TestUngroupedStateful(t *testing.T) {
 	ctx := context.Background()
-	b := ungrouped.New(test.NewAggregationSelector(), test.SdkEncoder, true)
+	b := ungrouped.New(test.NewAggregationSelector(), true)
 
 	counterA := test.NewCounterRecord(&test.CounterADesc, test.Labels1, 10)
 	caggA := counterA.Aggregator()