[aggregator] Checking if metadata is set to default should not cause …

…copying (#3198)

src/metrics/metadata/metadata.go

@@ -100,10 +100,10 @@ func (m PipelineMetadata) Equal(other PipelineMetadata) bool {
 
 // IsDefault returns whether this is the default standard pipeline metadata.
 func (m PipelineMetadata) IsDefault() bool {
-	return m.AggregationID.IsDefault() &&
-		m.StoragePolicies.IsDefault() &&
+	return m.AggregationID == aggregation.DefaultID &&
+		len(m.StoragePolicies) == 0 &&
 		m.Pipeline.IsEmpty() &&
-		m.DropPolicy.IsDefault()
+		m.DropPolicy == policy.DefaultDropPolicy
 }
 
 // IsMappingRule returns whether this is a rollup rule pipeline metadata.
@@ -435,7 +435,13 @@ func (sms StagedMetadatas) Equal(other StagedMetadatas) bool {
 
 // IsDefault determines whether the list of staged metadata is a default list.
 func (sms StagedMetadatas) IsDefault() bool {
-	return len(sms) == 1 && sms[0].IsDefault()
+	// very ugly but need to help out the go compiler here, as function calls have a high inlining cost
+	return len(sms) == 1 && len(sms[0].Pipelines) == 1 &&
+		sms[0].CutoverNanos == 0 && !sms[0].Tombstoned &&
+		sms[0].Pipelines[0].AggregationID == aggregation.DefaultID &&
+		len(sms[0].Pipelines[0].StoragePolicies) == 0 &&
+		sms[0].Pipelines[0].Pipeline.IsEmpty() &&
+		sms[0].Pipelines[0].DropPolicy == policy.DefaultDropPolicy
 }
 
 // IsDropPolicyApplied returns whether the list of staged metadata is the

src/metrics/metadata/metadata_benchmark_test.go

@@ -23,18 +23,32 @@ package metadata
 import (
 	"runtime"
 	"testing"
+
+	"github.com/m3db/m3/src/metrics/aggregation"
+	"github.com/m3db/m3/src/metrics/policy"
 )
 
 func isDefault(m StagedMetadatas) bool {
 	return m.IsDefault()
 }
 
 func BenchmarkMetadata_IsDefault(b *testing.B) {
-	m := testLargeStagedMetadatas
-	m = append(m, testLargeStagedMetadatas...)
+	m := StagedMetadatas{
+		StagedMetadata{
+			CutoverNanos: 0,
+			Tombstoned:   false,
+			Metadata: Metadata{
+				Pipelines: []PipelineMetadata{
+					{
+						AggregationID:   aggregation.DefaultID,
+						StoragePolicies: []policy.StoragePolicy{},
+					},
+				},
+			},
+		},
+	}
 	for i := 0; i < b.N; i++ {
-		m[0].CutoverNanos = int64(i)
-		if isDefault(m) {
+		if !isDefault(m) {
 			b.Fail()
 		}
 	}

src/metrics/metadata/metadata_test.go

@@ -700,7 +700,10 @@ func TestStagedMetadatasIsDefault(t *testing.T) {
 	}
 
 	for _, input := range inputs {
-		require.Equal(t, input.expected, input.metadatas.IsDefault())
+		input := input
+		t.Run(fmt.Sprintf("%v", input.metadatas), func(t *testing.T) {
+			require.Equal(t, input.expected, input.metadatas.IsDefault())
+		})
 	}
 }
 

src/metrics/pipeline/applied/type.go

@@ -180,7 +180,7 @@ func NewPipeline(ops []OpUnion) Pipeline {
 func (p Pipeline) Len() int { return len(p.operations) }
 
 // IsEmpty determines whether a pipeline is empty.
-func (p Pipeline) IsEmpty() bool { return p.Len() == 0 }
+func (p Pipeline) IsEmpty() bool { return len(p.operations) == 0 }
 
 // At returns the operation at a given step.
 func (p Pipeline) At(i int) OpUnion { return p.operations[i] }