Implement preallocated buffer

db.go

@@ -858,7 +858,7 @@ func writeLevel0Table(ft flushTask, f io.Writer, bopts table.Options) error {
 		}
 		b.Add(iter.Key(), iter.Value())
 	}
-	_, err := f.Write(b.Finish())
+	_, err := b.Finish().WriteTo(f)
 	return err
 }
 

db2_test.go

@@ -534,7 +534,7 @@ func createTableWithRange(t *testing.T, db *DB, start, end int) *table.Table {
 	fd, err := y.CreateSyncedFile(table.NewFilename(fileID, db.opt.Dir), true)
 	require.NoError(t, err)
 
-	_, err = fd.Write(b.Finish())
+	_, err = b.Finish().WriteTo(fd)
 	require.NoError(t, err, "unable to write to file")
 
 	opts := table.Options{LoadingMode: options.LoadToRAM, ChkMode: options.NoVerification}

levels.go

@@ -571,15 +571,19 @@ func (s *levelsController) compactBuildTables(
 		s.kv.opt.Debugf("LOG Compact. Added %d keys. Skipped %d keys. Iteration took: %v",
 			numKeys, numSkips, time.Since(timeStart))
 		build := func(fileID uint64) (*table.Table, error) {
-			fd, err := y.CreateSyncedFile(table.NewFilename(fileID, s.kv.opt.Dir), true)
+			fd, err := y.CreateSyncedFile(table.NewFilename(fileID, s.kv.opt.Dir), false)
 			if err != nil {
 				return nil, errors.Wrapf(err, "While opening new table: %d", fileID)
 			}
 
-			if _, err := fd.Write(builder.Finish()); err != nil {
+			if _, err := builder.Finish().WriteTo(fd); err != nil {
 				return nil, errors.Wrapf(err, "Unable to write to file: %d", fileID)
 			}
 
+			if err := fd.Sync(); err != nil {
+				return nil, errors.Wrapf(err, "Unable to sync file: %d", fileID)
+			}
+
 			opts := table.Options{
 				LoadingMode: s.kv.opt.TableLoadingMode,
 				ChkMode:     s.kv.opt.ChecksumVerificationMode,

manifest_test.go

@@ -150,7 +150,7 @@ func buildTable(t *testing.T, keyValues [][]string) *os.File {
 			UserMeta: 0,
 		})
 	}
-	f.Write(b.Finish())
+	y.Check2(b.Finish().WriteTo(f))
 	f.Close()
 	f, _ = y.OpenSyncedFile(filename, true)
 	return f

stream_writer.go

@@ -297,8 +297,8 @@ func (w *sortedWriter) send() error {
 		return err
 	}
 	go func(builder *table.Builder) {
-		data := builder.Finish()
-		err := w.createTable(data)
+		buf := builder.Finish()
+		err := w.createTable(buf)
 		w.throttle.Done(err)
 	}(w.builder)
 
@@ -319,16 +319,19 @@ func (w *sortedWriter) Done() error {
 	return w.send()
 }
 
-func (w *sortedWriter) createTable(data []byte) error {
-	if len(data) == 0 {
+func (w *sortedWriter) createTable(buf *y.Buffer) error {
+	if buf.Len() == 0 {
 		return nil
 	}
 	fileID := w.db.lc.reserveFileID()
-	fd, err := y.CreateSyncedFile(table.NewFilename(fileID, w.db.opt.Dir), true)
+	fd, err := y.CreateSyncedFile(table.NewFilename(fileID, w.db.opt.Dir), false)
 	if err != nil {
 		return err
 	}
-	if _, err := fd.Write(data); err != nil {
+	if _, err := buf.WriteTo(fd); err != nil {
+		return err
+	}
+	if err := fd.Sync(); err != nil {
 		return err
 	}
 	opts := table.Options{

table/builder.go

@@ -60,7 +60,7 @@ func (h header) Size() int { return headerSize }
 // Builder is used in building a table.
 type Builder struct {
 	// Typically tens or hundreds of meg. This is for one single file.
-	buf *bytes.Buffer
+	buf *y.Buffer
 
 	baseKey      []byte   // Base key for the current block.
 	baseOffset   uint32   // Offset for the current block.
@@ -75,7 +75,7 @@ type Builder struct {
 // NewTableBuilder makes a new TableBuilder.
 func NewTableBuilder(opts Options) *Builder {
 	return &Builder{
-		buf:        newBuffer(1 << 20),
+		buf:        y.NewBuffer(1 << 20),
 		tableIndex: &pb.TableIndex{},
 		keyHashes:  make([]uint64, 0, 1024), // Avoid some malloc calls.
 		opt:        &opts,
@@ -126,7 +126,7 @@ func (b *Builder) addHelper(key []byte, v y.ValueStruct) {
 	b.buf.Write(h.Encode())
 	b.buf.Write(diffKey) // We only need to store the key difference.
 
-	v.EncodeTo(b.buf)
+	v.EncodeToYBuf(b.buf)
 }
 
 /*
@@ -144,7 +144,7 @@ func (b *Builder) finishBlock() {
 	b.buf.Write(y.U32SliceToBytes(b.entryOffsets))
 	b.buf.Write(y.U32ToBytes(uint32(len(b.entryOffsets))))
 
-	blockBuf := b.buf.Bytes()[b.baseOffset:] // Store checksum for current block.
+	blockBuf := b.buf.ReadAt(int(b.baseOffset), -1)
 	b.writeChecksum(blockBuf)
 
 	// TODO(Ashish):Add padding: If we want to make block as multiple of OS pages, we can
@@ -222,7 +222,7 @@ The table structure looks like
 | Index   | Index Size | Checksum  | Checksum Size |
 +---------+------------+-----------+---------------+
 */
-func (b *Builder) Finish() []byte {
+func (b *Builder) Finish() *y.Buffer {
 	bf := z.NewBloomFilter(float64(len(b.keyHashes)), b.opt.BloomFalsePositive)
 	for _, h := range b.keyHashes {
 		bf.Add(h)
@@ -244,7 +244,7 @@ func (b *Builder) Finish() []byte {
 	y.Check(err)
 
 	b.writeChecksum(index)
-	return b.buf.Bytes()
+	return b.buf
 }
 
 func (b *Builder) writeChecksum(data []byte) {

table/builder_test.go

@@ -23,10 +23,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/require"
-
 	"github.com/dgraph-io/badger/options"
 	"github.com/dgraph-io/badger/y"
+	"github.com/stretchr/testify/require"
 )
 
 func TestTableIndex(t *testing.T) {
@@ -63,7 +62,8 @@ func TestTableIndex(t *testing.T) {
 			}
 			builder.Add(k, vs)
 		}
-		f.Write(builder.Finish())
+		_, err = builder.Finish().WriteTo(f)
+		require.NoError(t, err, "unable to write to file")
 
 		opts = Options{LoadingMode: options.LoadToRAM, ChkMode: options.OnTableAndBlockRead}
 		tbl, err := OpenTable(f, opts)
@@ -76,3 +76,38 @@ func TestTableIndex(t *testing.T) {
 		}
 	})
 }
+
+func BenchmarkBuilder(b *testing.B) {
+	rand.Seed(time.Now().Unix())
+	key := func(i int) []byte {
+		return []byte(fmt.Sprintf("%032d", i))
+	}
+
+	val := make([]byte, 32)
+	rand.Read(val)
+	vs := y.ValueStruct{Value: []byte(val)}
+
+	keysCount := 1300000
+	for i := 0; i < b.N; i++ {
+		opts := Options{BlockSize: 4 * 1024, BloomFalsePositive: 0.01}
+		builder := NewTableBuilder(opts)
+		// filename := fmt.Sprintf("%s%c%d.sst", os.TempDir(), os.PathSeparator, rand.Int63())
+		// f, err := y.OpenSyncedFile(filename, false)
+		// require.NoError(b, err)
+
+		for i := 0; i < keysCount; i++ {
+			builder.Add(key(i), vs)
+		}
+
+		_ = builder.Finish()
+		// b.Logf("data size: %d\n", len(data))
+		// bo := bufio.NewWriterSize(f, 100<<20)
+		// f.Write(builder.Finish())
+		// f.Write(builder.Finish())
+		// fmt.Println(builder.buf.Len())
+		// io.Copy(f, builder.Finish())
+		// f.Sync()
+		// bo.Flush()
+		// _, err = builder.Finish().WriteTo(f)
+	}
+}

table/table_test.go

@@ -77,7 +77,8 @@ func buildTable(t *testing.T, keyValues [][]string) *os.File {
 		y.AssertTrue(len(kv) == 2)
 		b.Add(y.KeyWithTs([]byte(kv[0]), 0), y.ValueStruct{Value: []byte(kv[1]), Meta: 'A', UserMeta: 0})
 	}
-	f.Write(b.Finish())
+	_, err = b.Finish().WriteTo(f)
+	require.NoError(t, err, "writing to file failed")
 	f.Close()
 	f, _ = y.OpenSyncedFile(filename, true)
 	return f
@@ -666,7 +667,8 @@ func TestTableBigValues(t *testing.T) {
 		builder.Add(key, vs)
 	}
 
-	f.Write(builder.Finish())
+	_, err = builder.Finish().WriteTo(f)
+	require.NoError(t, err, "unable to write to file")
 	opts = Options{LoadingMode: options.LoadToRAM, ChkMode: options.OnTableAndBlockRead}
 	tbl, err := OpenTable(f, opts)
 	require.NoError(t, err, "unable to open table")
@@ -761,7 +763,8 @@ func BenchmarkReadMerged(b *testing.B) {
 			v := fmt.Sprintf("%d", id)
 			builder.Add([]byte(k), y.ValueStruct{Value: []byte(v), Meta: 123, UserMeta: 0})
 		}
-		f.Write(builder.Finish())
+		_, err = builder.Finish().WriteTo(f)
+		require.NoError(b, err, "unable to write to file")
 		opts = Options{LoadingMode: options.LoadToRAM, ChkMode: options.OnTableAndBlockRead}
 		tbl, err := OpenTable(f, opts)
 		y.Check(err)
@@ -847,9 +850,39 @@ func getTableForBenchmarks(b *testing.B, count int) *Table {
 		builder.Add([]byte(k), y.ValueStruct{Value: []byte(v)})
 	}
 
-	f.Write(builder.Finish())
+	_, err = builder.Finish().WriteTo(f)
+	require.NoError(b, err, "unable to write to file")
 	opts = Options{LoadingMode: options.LoadToRAM, ChkMode: options.NoVerification}
 	tbl, err := OpenTable(f, opts)
 	require.NoError(b, err, "unable to open table")
 	return tbl
 }
+
+func BenchmarkFilewrite(b *testing.B) {
+	a := bytes.Repeat([]byte("a"), 1024)
+	rand.Seed(time.Now().Unix())
+	b.Run("sync-flag", func(b *testing.B) {
+		filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Int63())
+		f, err := y.OpenSyncedFile(filename, true)
+		require.NoError(b, err)
+		for i := 0; i < b.N; i++ {
+			for i := 0; i < 100; i++ {
+				_, err := f.Write(a)
+				require.NoError(b, err)
+			}
+		}
+	})
+
+	b.Run("fysnc", func(b *testing.B) {
+		filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Int63())
+		f, err := y.OpenSyncedFile(filename, false)
+		require.NoError(b, err)
+		for i := 0; i < b.N; i++ {
+			for i := 0; i < 100; i++ {
+				_, err := f.Write(a)
+				require.NoError(b, err)
+			}
+			require.NoError(b, f.Sync())
+		}
+	})
+}

y/iterator.go

@@ -86,6 +86,18 @@ func (v *ValueStruct) EncodeTo(buf *bytes.Buffer) {
 	buf.Write(v.Value)
 }
 
+// EncodeTo should be kept in sync with the Encode function above. The reason
+// this function exists is to avoid creating byte arrays per key-value pair in
+// table/builder.go.
+func (v *ValueStruct) EncodeToYBuf(buf *Buffer) {
+	buf.WriteByte(v.Meta)
+	buf.WriteByte(v.UserMeta)
+	var enc [binary.MaxVarintLen64]byte
+	sz := binary.PutUvarint(enc[:], v.ExpiresAt)
+	buf.Write(enc[:sz])
+	buf.Write(v.Value)
+}
+
 // Iterator is an interface for a basic iterator.
 type Iterator interface {
 	Next()