From dea4ea82d1fafae0bc3c9bac62f457b45cb80650 Mon Sep 17 00:00:00 2001
From: Vytenis Darulis <vytenis@uber.com>
Date: Tue, 9 Feb 2021 00:54:27 -0500
Subject: [PATCH] [aggregator] Checking if metadata is set to default should
 not cause copying

---
 src/metrics/metadata/metadata.go              | 14 ++++++++----
 .../metadata/metadata_benchmark_test.go       | 22 +++++++++++++++----
 src/metrics/metadata/metadata_test.go         |  5 ++++-
 src/metrics/pipeline/applied/type.go          |  2 +-
 4 files changed, 33 insertions(+), 10 deletions(-)

diff --git a/src/metrics/metadata/metadata.go b/src/metrics/metadata/metadata.go
index f93e928cec..0995677a2b 100644
--- a/src/metrics/metadata/metadata.go
+++ b/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
diff --git a/src/metrics/metadata/metadata_benchmark_test.go b/src/metrics/metadata/metadata_benchmark_test.go
index 8577a448db..ff505e28ae 100644
--- a/src/metrics/metadata/metadata_benchmark_test.go
+++ b/src/metrics/metadata/metadata_benchmark_test.go
@@ -23,6 +23,9 @@ package metadata
 import (
 	"runtime"
 	"testing"
+
+	"github.com/m3db/m3/src/metrics/aggregation"
+	"github.com/m3db/m3/src/metrics/policy"
 )
 
 func isDefault(m StagedMetadatas) bool {
@@ -30,11 +33,22 @@ func isDefault(m StagedMetadatas) bool {
 }
 
 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()
 		}
 	}
diff --git a/src/metrics/metadata/metadata_test.go b/src/metrics/metadata/metadata_test.go
index 25a4120803..7376747a07 100644
--- a/src/metrics/metadata/metadata_test.go
+++ b/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())
+		})
 	}
 }
 
diff --git a/src/metrics/pipeline/applied/type.go b/src/metrics/pipeline/applied/type.go
index a6db631164..c1c57d6a70 100644
--- a/src/metrics/pipeline/applied/type.go
+++ b/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] }