storage: coalesce results with the same labels into one row

Change-Id: Ia9d62651ba23794cd64dfcf2ea2e529914f3b100
Reviewed-on: https://go-review.googlesource.com/35674Reviewed-by: Russ Cox <rsc@golang.org>

storage/benchfmt/benchfmt.go

@@ -9,6 +9,7 @@ package benchfmt
 
 import (
 	"bufio"
+	"bytes"
 	"fmt"
 	"io"
 	"sort"
@@ -53,7 +54,7 @@ func NewReader(r io.Reader) *Reader {
 // AddLabels adds additional labels as if they had been read from the header of a file.
 // It must be called before the first call to r.Next.
 func (r *Reader) AddLabels(labels Labels) {
-	r.permLabels = labels.copy()
+	r.permLabels = labels.Copy()
 	for k, v := range labels {
 		r.labels[k] = v
 	}
@@ -61,6 +62,7 @@ func (r *Reader) AddLabels(labels Labels) {
 
 // Result represents a single line from a benchmark file.
 // All information about that line is self-contained in the Result.
+// A Result is immutable once created.
 type Result struct {
 	// Labels is the set of persistent labels that apply to the result.
 	// Labels must not be modified.
@@ -75,10 +77,29 @@ type Result struct {
 	Content string
 }
 
+// SameLabels reports whether r and b have the same labels.
+func (r *Result) SameLabels(b *Result) bool {
+	return r.Labels.Equal(b.Labels) && r.NameLabels.Equal(b.NameLabels)
+}
+
 // Labels is a set of key-value strings.
 type Labels map[string]string
 
-// TODO(quentin): Add String and Equal methods to Labels?
+// String returns the labels formatted as a comma-separated
+// list enclosed in braces.
+func (l Labels) String() string {
+	var out bytes.Buffer
+	out.WriteString("{")
+	for k, v := range l {
+		fmt.Fprintf(&out, "%q: %q, ", k, v)
+	}
+	if out.Len() > 1 {
+		// Remove extra ", "
+		out.Truncate(out.Len() - 2)
+	}
+	out.WriteString("}")
+	return out.String()
+}
 
 // Keys returns a sorted list of the keys in l.
 func (l Labels) Keys() []string {
@@ -90,6 +111,19 @@ func (l Labels) Keys() []string {
 	return out
 }
 
+// Equal reports whether l and b have the same keys and values.
+func (l Labels) Equal(b Labels) bool {
+	if len(l) != len(b) {
+		return false
+	}
+	for k := range l {
+		if l[k] != b[k] {
+			return false
+		}
+	}
+	return true
+}
+
 // A Printer prints a sequence of benchmark results.
 type Printer struct {
 	w      io.Writer
@@ -178,9 +212,9 @@ func (r *Reader) newResult(labels Labels, lineNum int, name, content string) *Re
 	return res
 }
 
-// copy returns a new copy of the labels map, to protect against
+// Copy returns a new copy of the labels map, to protect against
 // future modifications to labels.
-func (l Labels) copy() Labels {
+func (l Labels) Copy() Labels {
 	new := make(Labels)
 	for k, v := range l {
 		new[k] = v
@@ -207,7 +241,7 @@ func (r *Reader) Next() bool {
 			}
 			if !copied {
 				copied = true
-				r.labels = r.labels.copy()
+				r.labels = r.labels.Copy()
 			}
 			// TODO(quentin): Spec says empty value is valid, but
 			// we need a way to cancel previous labels, so we'll
@@ -222,7 +256,7 @@ func (r *Reader) Next() bool {
 		// Blank line delimits the header. If we find anything else, the file must not have a header.
 		if !havePerm {
 			if line == "" {
-				r.permLabels = r.labels.copy()
+				r.permLabels = r.labels.Copy()
 			} else {
 				r.permLabels = Labels{}
 			}

storage/db/db.go

@@ -167,6 +167,7 @@ type Upload struct {
 	// pending arguments for flush
 	insertRecordArgs []interface{}
 	insertLabelArgs  []interface{}
+	lastResult       *benchfmt.Result
 }
 
 // now is a hook for testing
@@ -267,11 +268,19 @@ func (db *DB) NewUpload(ctx context.Context) (*Upload, error) {
 // InsertRecord inserts a single record in an existing upload.
 // If InsertRecord returns a non-nil error, the Upload has failed and u.Abort() must be called.
 func (u *Upload) InsertRecord(r *benchfmt.Result) error {
+	if u.lastResult != nil && u.lastResult.SameLabels(r) {
+		data := u.insertRecordArgs[len(u.insertRecordArgs)-1].([]byte)
+		data = append(data, r.Content...)
+		data = append(data, '\n')
+		u.insertRecordArgs[len(u.insertRecordArgs)-1] = data
+		return nil
+	}
 	// TODO(quentin): Support multiple lines (slice of results?)
 	var buf bytes.Buffer
 	if err := benchfmt.NewPrinter(&buf).Print(r); err != nil {
 		return err
 	}
+	u.lastResult = r
 	u.insertRecordArgs = append(u.insertRecordArgs, u.ID, u.recordid, buf.Bytes())
 	for _, k := range r.Labels.Keys() {
 		if err := u.insertLabel(k, r.Labels[k]); err != nil {
@@ -331,6 +340,7 @@ func (u *Upload) flush() error {
 		}
 		u.insertLabelArgs = nil
 	}
+	u.lastResult = nil
 	return nil
 }
 

storage/db/db_test.go

@@ -114,24 +114,25 @@ func TestReplaceUpload(t *testing.T) {
 
 	ctx := context.Background()
 
-	r := &benchfmt.Result{
-		benchfmt.Labels{"key": "value"},
-		nil,
-		1,
-		"BenchmarkName 1 ns/op",
-	}
+	labels := benchfmt.Labels{"key": "value"}
+
 	u, err := db.NewUpload(ctx)
 	if err != nil {
 		t.Fatalf("NewUpload: %v", err)
 	}
-	r.Labels["uploadid"] = u.ID
+	labels["uploadid"] = u.ID
 	for _, num := range []string{"1", "2"} {
-		r.Labels["num"] = num
+		labels["num"] = num
 		for _, num2 := range []int{1, 2} {
-			r.Content = fmt.Sprintf("BenchmarkName %d ns/op", num2)
-			if err := u.InsertRecord(r); err != nil {
+			if err := u.InsertRecord(&benchfmt.Result{
+				labels,
+				nil,
+				1,
+				fmt.Sprintf("BenchmarkName %d ns/op", num2),
+			}); err != nil {
 				t.Fatalf("InsertRecord: %v", err)
 			}
+			labels = labels.Copy()
 		}
 	}
 
@@ -150,18 +151,23 @@ BenchmarkName 1 ns/op
 BenchmarkName 2 ns/op
 `)
 
-	r.Labels["num"] = "3"
-	r.Content = "BenchmarkName 3 ns/op"
+	labels["num"] = "3"
 
 	for _, uploadid := range []string{u.ID, "new"} {
 		u, err := db.ReplaceUpload(uploadid)
 		if err != nil {
 			t.Fatalf("ReplaceUpload: %v", err)
 		}
-		r.Labels["uploadid"] = u.ID
-		if err := u.InsertRecord(r); err != nil {
+		labels["uploadid"] = u.ID
+		if err := u.InsertRecord(&benchfmt.Result{
+			labels,
+			nil,
+			1,
+			"BenchmarkName 3 ns/op",
+		}); err != nil {
 			t.Fatalf("InsertRecord: %v", err)
 		}
+		labels = labels.Copy()
 
 		if err := u.Commit(); err != nil {
 			t.Fatalf("Commit: %v", err)
@@ -193,15 +199,16 @@ func TestNewUpload(t *testing.T) {
 	br := benchfmt.NewReader(strings.NewReader(`
 key: value
 BenchmarkName 1 ns/op
+BenchmarkName 2 ns/op
 `))
-	if !br.Next() {
-		t.Fatalf("unable to read test string: %v", br.Err())
+	for br.Next() {
+		if err := u.InsertRecord(br.Result()); err != nil {
+			t.Fatalf("InsertRecord: %v", err)
+		}
 	}
-
-	if err := u.InsertRecord(br.Result()); err != nil {
-		t.Fatalf("InsertRecord: %v", err)
+	if err := br.Err(); err != nil {
+		t.Fatalf("Err: %v", err)
 	}
-
 	if err := u.Commit(); err != nil {
 		t.Fatalf("Commit: %v", err)
 	}