Merge pull request #3744 from benbjohnson/bz1-bench

Add bz1 size benchmarks

tsdb/engine/bz1/bz1.go

@@ -360,6 +360,7 @@ func (e *Engine) writeIndex(tx *bolt.Tx, key string, a [][]byte) error {
 	// This is the optimized fast path. Otherwise we need to merge the points
 	// with existing blocks on disk and rewrite all the blocks for that range.
 	if k, v := c.Last(); k == nil || int64(btou64(v[0:8])) < tmin {
+		bkt.FillPercent = 1.0
 		if err := e.writeBlocks(bkt, a); err != nil {
 			return fmt.Errorf("append blocks: %s", err)
 		}
@@ -541,6 +542,18 @@ func (e *Engine) Stats() (stats Stats, err error) {
 	return stats, err
 }
 
+// SeriesBucketStats returns internal BoltDB stats for a series bucket.
+func (e *Engine) SeriesBucketStats(key string) (stats bolt.BucketStats, err error) {
+	err = e.db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket([]byte("points")).Bucket([]byte(key))
+		if bkt != nil {
+			stats = bkt.Stats()
+		}
+		return nil
+	})
+	return stats, err
+}
+
 // Stats represents internal engine statistics.
 type Stats struct {
 	Size int64 // BoltDB data size

tsdb/engine/bz1/bz1_test.go

@@ -14,8 +14,10 @@ import (
 	"testing/quick"
 	"time"
 
+	"github.com/influxdb/influxdb/influxql"
 	"github.com/influxdb/influxdb/tsdb"
 	"github.com/influxdb/influxdb/tsdb/engine/bz1"
+	"github.com/influxdb/influxdb/tsdb/engine/wal"
 )
 
 // Ensure the engine can write series metadata and reload it.
@@ -302,6 +304,62 @@ func TestEngine_WriteIndex_Quick(t *testing.T) {
 	}, nil)
 }
 
+func BenchmarkEngine_WriteIndex_512b(b *testing.B)  { benchmarkEngine_WriteIndex(b, 512) }
+func BenchmarkEngine_WriteIndex_1KB(b *testing.B)   { benchmarkEngine_WriteIndex(b, 1*1024) }
+func BenchmarkEngine_WriteIndex_4KB(b *testing.B)   { benchmarkEngine_WriteIndex(b, 4*1024) }
+func BenchmarkEngine_WriteIndex_16KB(b *testing.B)  { benchmarkEngine_WriteIndex(b, 16*1024) }
+func BenchmarkEngine_WriteIndex_32KB(b *testing.B)  { benchmarkEngine_WriteIndex(b, 32*1024) }
+func BenchmarkEngine_WriteIndex_64KB(b *testing.B)  { benchmarkEngine_WriteIndex(b, 64*1024) }
+func BenchmarkEngine_WriteIndex_128KB(b *testing.B) { benchmarkEngine_WriteIndex(b, 128*1024) }
+func BenchmarkEngine_WriteIndex_256KB(b *testing.B) { benchmarkEngine_WriteIndex(b, 256*1024) }
+
+func benchmarkEngine_WriteIndex(b *testing.B, blockSize int) {
+	// Skip small iterations.
+	if b.N < 1000000 {
+		return
+	}
+
+	// Create a simple engine.
+	e := OpenDefaultEngine()
+	e.BlockSize = blockSize
+	defer e.Close()
+
+	// Create codec.
+	codec := tsdb.NewFieldCodec(map[string]*tsdb.Field{
+		"value": {
+			ID:   uint8(1),
+			Name: "value",
+			Type: influxql.Float,
+		},
+	})
+
+	// Generate points.
+	a := make(map[string][][]byte)
+	a["cpu"] = make([][]byte, b.N)
+	for i := 0; i < b.N; i++ {
+		a["cpu"][i] = wal.MarshalEntry(int64(i), MustEncodeFields(codec, tsdb.Fields{"value": float64(i)}))
+	}
+
+	b.ResetTimer()
+
+	// Insert into engine.
+	if err := e.WriteIndex(a, nil, nil); err != nil {
+		b.Fatal(err)
+	}
+
+	// Calculate on-disk size per point.
+	bs, _ := e.SeriesBucketStats("cpu")
+	stats, err := e.Stats()
+	if err != nil {
+		b.Fatal(err)
+	}
+	b.Logf("pts=%9d  bytes/pt=%4.01f  leaf-util=%3.0f%%",
+		b.N,
+		float64(stats.Size)/float64(b.N),
+		(float64(bs.LeafInuse)/float64(bs.LeafAlloc))*100.0,
+	)
+}
+
 // Engine represents a test wrapper for bz1.Engine.
 type Engine struct {
 	*bz1.Engine
@@ -432,6 +490,15 @@ func MergePoints(a []Points) Points {
 	return m
 }
 
+// MustEncodeFields encodes fields with codec. Panic on error.
+func MustEncodeFields(codec *tsdb.FieldCodec, fields tsdb.Fields) []byte {
+	b, err := codec.EncodeFields(fields)
+	if err != nil {
+		panic(err)
+	}
+	return b
+}
+
 // copyBytes returns a copy of a byte slice.
 func copyBytes(b []byte) []byte {
 	if b == nil {