Repository: dgraph-io/badger
Pull request #1546 (2020-09-29 19:14:55)

[BREAKING] feat(index): Use flatbuffers instead of protobuf
This PR 
- Uses flatbuffers instead of protobufs for table index and directly stores byte slices in the cache.
- Uses MaxVersion to pick the oldest tables for compaction first.
- Uses leveldb/bloom so that we can test it without unmarshal
- Adds uncompressed size and key count in table index.
- Updates write bench tool to use managed mode.

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/dgraph-io/badger/1546)
<!-- Reviewable:end -->


CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/dgraph-io/badger?pullRequest=1546) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our [Contributor License Agreement](https://cla-assistant.io/dgraph-io/badger?pullRequest=1546) before we can accept your contribution.<br/>**3** out of **4** committers have signed the CLA.<br/><br/>:white_check_mark: jarifibrahim<br/>:white_check_mark: manishrjain<br/>:white_check_mark: ahsanbarkati<br/>:x: darkn3rd<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/dgraph-io/badger?pullRequest=1546) it.</sub>

Patch:
diff --git a/badger/cmd/info.go b/badger/cmd/info.go
index 7769a7f..ba1c5eb 100644
--- a/badger/cmd/info.go
+++ b/badger/cmd/info.go
@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"path"
 	"path/filepath"
 	"sort"
 	"strings"
@@ -239,27 +240,46 @@ func dur(src, dst time.Time) string {
 	return humanize.RelTime(dst, src, "earlier", "later")
 }
 
+func getInfo(fileInfos []os.FileInfo, tid uint64) int64 {
+	fileName := table.IDToFilename(tid)
+	for _, fi := range fileInfos {
+		if path.Base(fi.Name()) == fileName {
+			return fi.Size()
+		}
+	}
+	return 0
+}
+
 func tableInfo(dir, valueDir string, db *badger.DB) {
 	// we want all tables with keys count here.
-	tables := db.Tables(true)
+	tables := db.Tables()
+	fileInfos, err := ioutil.ReadDir(dir)
+	y.Check(err)
+
 	fmt.Println()
 	fmt.Println("SSTable [Li, Id, Total Keys including internal keys] " +
-		"[Left Key, Version -> Right Key, Version] [Index Size] [BF Size]")
+		"[Compression Ratio, Uncompressed Size, Index Size, BF Size] " +
+		"[Left Key, Version -> Right Key, Version]")
 	totalIndex := uint64(0)
 	totalBloomFilter := uint64(0)
+	totalCompressionRatio := float64(0.0)
 	for _, t := range tables {
 		lk, lt := y.ParseKey(t.Left), y.ParseTs(t.Left)
 		rk, rt := y.ParseKey(t.Right), y.ParseTs(t.Right)
 
+		compressionRatio := float64(t.UncompressedSize) /
+			float64(getInfo(fileInfos, t.ID)-int64(t.IndexSz))
+		fmt.Printf("SSTable [L%d, %03d, %07d] [%.2f, %s, %s, %s] [%20X, v%d -> %20X, v%d]\n",
+			t.Level, t.ID, t.KeyCount, compressionRatio, hbytes(int64(t.UncompressedSize)),
+			hbytes(int64(t.IndexSz)), hbytes(int64(t.BloomFilterSize)), lk, lt, rk, rt)
 		totalIndex += uint64(t.IndexSz)
 		totalBloomFilter += uint64(t.BloomFilterSize)
-		fmt.Printf("SSTable [L%d, %03d, %07d] [%20X, v%d -> %20X, v%d] [%s] [%s] \n",
-			t.Level, t.ID, t.KeyCount, lk, lt, rk, rt, hbytes(int64(t.IndexSz)),
-			hbytes(int64(t.BloomFilterSize)))
+		totalCompressionRatio += compressionRatio
 	}
 	fmt.Println()
 	fmt.Printf("Total Index Size: %s\n", hbytes(int64(totalIndex)))
 	fmt.Printf("Total BloomFilter Size: %s\n", hbytes(int64(totalIndex)))
+	fmt.Printf("Mean Compression Ratio: %.2f\n", totalCompressionRatio/float64(len(tables)))
 	fmt.Println()
 }
 
diff --git a/badger/cmd/read_bench.go b/badger/cmd/read_bench.go
index 1c0e094..8b34511 100644
--- a/badger/cmd/read_bench.go
+++ b/badger/cmd/read_bench.go
@@ -19,6 +19,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"math"
 	"math/rand"
 	"strings"
 	"sync/atomic"
@@ -120,7 +121,6 @@ func readBench(cmd *cobra.Command, args []string) error {
 		return y.Wrapf(err, "unable to open DB")
 	}
 	defer db.Close()
-	now := time.Now()
 
 	fmt.Println("*********************************************************")
 	fmt.Println("Starting to benchmark Reads")
@@ -131,32 +131,7 @@ func readBench(cmd *cobra.Command, args []string) error {
 		fullScanDB(db)
 		return nil
 	}
-	keys, err := getSampleKeys(db)
-	if err != nil {
-		return y.Wrapf(err, "error while sampling keys")
-	}
-	fmt.Println("*********************************************************")
-	fmt.Printf("Total Sampled Keys: %d, read in time: %s\n", len(keys), time.Since(now))
-	fmt.Println("*********************************************************")
-
-	if len(keys) == 0 {
-		fmt.Println("DB is empty, hence returning")
-		return nil
-	}
-	c := z.NewCloser(0)
-	startTime = time.Now()
-	for i := 0; i < numGoroutines; i++ {
-		c.AddRunning(1)
-		go readKeys(db, c, keys)
-	}
-
-	// also start printing stats
-	c.AddRunning(1)
-	go printStats(c)
-
-	<-time.After(dur)
-	c.SignalAndWait()
-
+	readTest(db, dur)
 	return nil
 }
 
@@ -199,16 +174,20 @@ func readKeys(db *badger.DB, c *z.Closer, keys [][]byte) {
 
 func lookupForKey(db *badger.DB, key []byte) (sz uint64) {
 	err := db.View(func(txn *badger.Txn) error {
-		itm, err := txn.Get(key)
-		y.Check(err)
-
-		if keysOnly {
-			sz = uint64(itm.KeySize())
-		} else {
-			y.Check2(itm.ValueCopy(nil))
-			sz = uint64(itm.EstimatedSize())
+		iopt := badger.DefaultIteratorOptions
+		iopt.AllVersions = true
+		it := txn.NewKeyIterator(key, iopt)
+		defer it.Close()
+
+		cnt := 0
+		for it.Seek(key); it.Valid(); it.Next() {
+			itm := it.Item()
+			sz += uint64(itm.EstimatedSize())
+			cnt++
+			if cnt == 10 {
+				break
+			}
 		}
-
 		return nil
 	})
 	y.Check(err)
@@ -219,7 +198,7 @@ func lookupForKey(db *badger.DB, key []byte) (sz uint64) {
 func getSampleKeys(db *badger.DB) ([][]byte, error) {
 	var keys [][]byte
 	count := 0
-	stream := db.NewStream()
+	stream := db.NewStreamAt(math.MaxUint64)
 
 	// overide stream.KeyToList as we only want keys. Also
 	// we can take only first version for the key.
diff --git a/badger/cmd/write_bench.go b/badger/cmd/write_bench.go
index 39581cb..25cb0ca 100644
--- a/badger/cmd/write_bench.go
+++ b/badger/cmd/write_bench.go
@@ -21,6 +21,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"log"
+	"math"
 	"math/rand"
 	"os"
 	"path/filepath"
@@ -88,7 +89,7 @@ const (
 func init() {
 	benchCmd.AddCommand(writeBenchCmd)
 	writeBenchCmd.Flags().IntVarP(&keySz, "key-size", "k", 32, "Size of key")
-	writeBenchCmd.Flags().IntVarP(&valSz, "val-size", "v", 128, "Size of value")
+	writeBenchCmd.Flags().IntVar(&valSz, "val-size", 128, "Size of value")
 	writeBenchCmd.Flags().Float64VarP(&numKeys, "keys-mil", "m", 10.0,
 		"Number of keys to add in millions")
 	writeBenchCmd.Flags().BoolVar(&syncWrites, "sync", true,
@@ -96,12 +97,12 @@ func init() {
 	writeBenchCmd.Flags().BoolVarP(&force, "force-compact", "f", true,
 		"Force compact level 0 on close.")
 	writeBenchCmd.Flags().BoolVarP(&sorted, "sorted", "s", false, "Write keys in sorted order.")
-	writeBenchCmd.Flags().BoolVarP(&showLogs, "logs", "l", false, "Show Badger logs.")
+	writeBenchCmd.Flags().BoolVarP(&showLogs, "verbose", "v", false, "Show Badger logs.")
 	writeBenchCmd.Flags().IntVarP(&valueThreshold, "value-th", "t", 1<<10, "Value threshold")
 	writeBenchCmd.Flags().IntVarP(&numVersions, "num-version", "n", 1, "Number of versions to keep")
-	writeBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache", 0,
+	writeBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache-mb", 0,
 		"Size of block cache in MB")
-	writeBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache", 0,
+	writeBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache-mb", 0,
 		"Size of index cache in MB.")
 	writeBenchCmd.Flags().Uint32Var(&vlogMaxEntries, "vlog-maxe", 1000000, "Value log Max Entries")
 	writeBenchCmd.Flags().StringVarP(&encryptionKey, "encryption-key", "e", "",
@@ -112,7 +113,7 @@ func init() {
 		"Load Bloom filter on DB open.")
 	writeBenchCmd.Flags().BoolVar(&detectConflicts, "conficts", true,
 		"If true, it badger will detect the conflicts")
-	writeBenchCmd.Flags().BoolVar(&compression, "compression", false,
+	writeBenchCmd.Flags().BoolVar(&compression, "compression", true,
 		"If true, badger will use ZSTD mode")
 	writeBenchCmd.Flags().BoolVar(&showDir, "show-dir", false,
 		"If true, the report will include the directory contents")
@@ -133,7 +134,7 @@ func writeRandom(db *badger.DB, num uint64) error {
 	y.Check2(rand.Read(value))
 
 	es := uint64(keySz + valSz) // entry size is keySz + valSz
-	batch := db.NewWriteBatch()
+	batch := db.NewManagedWriteBatch()
 
 	ttlPeriod, errParse := time.ParseDuration(ttlDuration)
 	y.Check(errParse)
@@ -146,11 +147,11 @@ func writeRandom(db *badger.DB, num uint64) error {
 		if ttlPeriod != 0 {
 			e.WithTTL(ttlPeriod)
 		}
-		err := batch.SetEntry(e)
+		err := batch.SetEntryAt(e, 1)
 		for err == badger.ErrBlockedWrites {
 			time.Sleep(time.Second)
-			batch = db.NewWriteBatch()
-			err = batch.SetEntry(e)
+			batch = db.NewManagedWriteBatch()
+			err = batch.SetEntryAt(e, 1)
 		}
 		if err != nil {
 			panic(err)
@@ -162,6 +163,34 @@ func writeRandom(db *badger.DB, num uint64) error {
 	return batch.Flush()
 }
 
+func readTest(db *badger.DB, dur time.Duration) {
+	now := time.Now()
+	keys, err := getSampleKeys(db)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Println("*********************************************************")
+	fmt.Printf("Total Sampled Keys: %d, read in time: %s\n", len(keys), time.Since(now))
+	fmt.Println("*********************************************************")
+
+	if len(keys) == 0 {
+		fmt.Println("DB is empty, hence returning")
+		return
+	}
+	c := z.NewCloser(0)
+	readStartTime := time.Now()
+	for i := 0; i < numGoroutines; i++ {
+		c.AddRunning(1)
+		go readKeys(db, c, keys)
+	}
+
+	// also start printing stats
+	c.AddRunning(1)
+	go printReadStats(c, readStartTime)
+	<-time.After(dur)
+	c.SignalAndWait()
+}
+
 func writeSorted(db *badger.DB, num uint64) error {
 	value := make([]byte, valSz)
 	y.Check2(rand.Read(value))
@@ -257,7 +286,7 @@ func writeBench(cmd *cobra.Command, args []string) error {
 	}
 
 	fmt.Printf("Opening badger with options = %+v\n", opt)
-	db, err := badger.Open(opt)
+	db, err := badger.OpenManaged(opt)
 	if err != nil {
 		return err
 	}
@@ -282,6 +311,16 @@ func writeBench(cmd *cobra.Command, args []string) error {
 	if sorted {
 		err = writeSorted(db, num)
 	} else {
+		go func() {
+			for {
+				select {
+				case <-c.HasBeenClosed():
+					return
+				case <-time.After(30 * time.Second):
+					readTest(db, 5*time.Minute)
+				}
+			}
+		}()
 		err = writeRandom(db, num)
 	}
 
@@ -296,7 +335,7 @@ func showKeysStats(db *badger.DB) {
 		validKeyCount    uint32
 	)
 
-	txn := db.NewTransaction(false)
+	txn := db.NewTransactionAt(math.MaxUint64, false)
 	defer txn.Discard()
 
 	iopt := badger.DefaultIteratorOptions
@@ -363,7 +402,7 @@ func reportStats(c *z.Closer, db *badger.DB) {
 			entries := atomic.LoadUint64(&entriesWritten)
 			bytesRate := sz / uint64(dur.Seconds())
 			entriesRate := entries / uint64(dur.Seconds())
-			fmt.Printf("Time elapsed: %s, bytes written: %s, speed: %s/sec, "+
+			fmt.Printf("[WRITE] Time elapsed: %s, bytes written: %s, speed: %s/sec, "+
 				"entries written: %d, speed: %d/sec, gcSuccess: %d\n", y.FixedDuration(time.Since(startTime)),
 				humanize.Bytes(sz), humanize.Bytes(bytesRate), entries, entriesRate, gcSuccess)
 		}
@@ -453,3 +492,25 @@ func dropPrefix(c *z.Closer, db *badger.DB) {
 		}
 	}
 }
+
+func printReadStats(c *z.Closer, startTime time.Time) {
+	defer c.Done()
+
+	t := time.NewTicker(time.Second)
+	defer t.Stop()
+	for {
+		select {
+		case <-c.HasBeenClosed():
+			return
+		case <-t.C:
+			dur := time.Since(startTime)
+			sz := atomic.LoadUint64(&sizeRead)
+			entries := atomic.LoadUint64(&entriesRead)
+			bytesRate := sz / uint64(dur.Seconds())
+			entriesRate := entries / uint64(dur.Seconds())
+			fmt.Printf("[READ] Time elapsed: %s, bytes read: %s, speed: %s/sec, "+
+				"entries read: %d, speed: %d/sec\n", y.FixedDuration(time.Since(startTime)),
+				humanize.Bytes(sz), humanize.Bytes(bytesRate), entries, entriesRate)
+		}
+	}
+}
diff --git a/compaction.go b/compaction.go
index 0372b8b..4495950 100644
--- a/compaction.go
+++ b/compaction.go
@@ -23,8 +23,6 @@ import (
 	"math"
 	"sync"
 
-	"golang.org/x/net/trace"
-
 	"github.com/dgraph-io/badger/v2/table"
 	"github.com/dgraph-io/badger/v2/y"
 )
@@ -128,19 +126,6 @@ type compactStatus struct {
 	levels []*levelCompactStatus
 }
 
-func (cs *compactStatus) toLog(tr trace.Trace) {
-	cs.RLock()
-	defer cs.RUnlock()
-
-	tr.LazyPrintf("Compaction status:")
-	for i, l := range cs.levels {
-		if l.debug() == "" {
-			continue
-		}
-		tr.LazyPrintf("[%d] %s", i, l.debug())
-	}
-}
-
 func (cs *compactStatus) overlapsWith(level int, this keyRange) bool {
 	cs.RLock()
 	defer cs.RUnlock()
diff --git a/db.go b/db.go
index dad4812..c0d14b9 100644
--- a/db.go
+++ b/db.go
@@ -54,12 +54,13 @@ const (
 )
 
 type closers struct {
-	updateSize *z.Closer
-	compactors *z.Closer
-	memtable   *z.Closer
-	writes     *z.Closer
-	valueGC    *z.Closer
-	pub        *z.Closer
+	updateSize  *z.Closer
+	compactors  *z.Closer
+	memtable    *z.Closer
+	writes      *z.Closer
+	valueGC     *z.Closer
+	pub         *z.Closer
+	cacheHealth *z.Closer
 }
 
 // DB provides the various functions required to interact with Badger.
@@ -249,8 +250,7 @@ func checkAndSetOptions(opt *Options) error {
 
 	needCache := (opt.Compression != options.None) || (len(opt.EncryptionKey) > 0)
 	if needCache && opt.BlockCacheSize == 0 {
-		opt.Warningf("BlockCacheSize should be set " +
-			"since compression/encryption are enabled")
+		panic("BlockCacheSize should be set since compression/encryption are enabled")
 	}
 	return nil
 }
@@ -362,6 +362,10 @@ func Open(opt Options) (db *DB, err error) {
 			return nil, errors.Wrap(err, "failed to create bf cache")
 		}
 	}
+
+	db.closers.cacheHealth = z.NewCloser(1)
+	go db.monitorCache(db.closers.cacheHealth)
+
 	if db.opt.InMemory {
 		db.opt.SyncWrites = false
 		// If badger is running in memory mode, push everything into the LSM Tree.
@@ -437,6 +441,42 @@ func Open(opt Options) (db *DB, err error) {
 	return db, nil
 }
 
+func (db *DB) monitorCache(c *z.Closer) {
+	defer c.Done()
+	if db.blockCache == nil {
+		return
+	}
+	count := 0
+	analyze := func(name string, metrics *ristretto.Metrics) {
+		// If the mean life expectancy is less than 10 seconds, the cache
+		// might be too small.
+		le := metrics.LifeExpectancySeconds()
+		lifeTooShort := le.Count > 0 && float64(le.Sum)/float64(le.Count) < 10
+		hitRatioTooLow := metrics.Ratio() > 0 && metrics.Ratio() < 0.4
+		if lifeTooShort && hitRatioTooLow {
+			db.opt.Warningf("%s might be too small. Metrics: %s\n", name, metrics)
+			db.opt.Warningf("Cache life expectancy (in seconds): %+v\n", le)
+
+		} else if le.Count > 1000 && count%5 == 0 {
+			db.opt.Infof("%s metrics: %s\n", name, metrics)
+		}
+	}
+
+	ticker := time.NewTicker(1 * time.Minute)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-c.HasBeenClosed():
+			return
+		case <-ticker.C:
+		}
+
+		count++
+		analyze("Block cache", db.BlockCacheMetrics())
+		analyze("Index cache", db.IndexCacheMetrics())
+	}
+}
+
 // getHead prints all the head pointer in the DB and return the max value.
 func (db *DB) getHead() (valuePointer, uint64) {
 	// This is a hack. If we use newTransaction(..) we'll end up in deadlock
@@ -562,6 +602,7 @@ func (db *DB) close() (err error) {
 	close(db.writeCh)
 
 	db.closers.pub.SignalAndWait()
+	db.closers.cacheHealth.Signal()
 
 	// Now close the value log.
 	if vlogErr := db.vlog.Close(); vlogErr != nil {
@@ -1450,15 +1491,15 @@ func (db *DB) GetSequence(key []byte, bandwidth uint64) (*Sequence, error) {
 
 // Tables gets the TableInfo objects from the level controller. If withKeysCount
 // is true, TableInfo objects also contain counts of keys for the tables.
-func (db *DB) Tables(withKeysCount bool) []TableInfo {
-	return db.lc.getTableInfo(withKeysCount)
+func (db *DB) Tables() []TableInfo {
+	return db.lc.getTableInfo()
 }
 
 // KeySplits can be used to get rough key ranges to divide up iteration over
 // the DB.
 func (db *DB) KeySplits(prefix []byte) []string {
 	var splits []string
-	tables := db.Tables(false)
+	tables := db.Tables()
 
 	// We just want table ranges here and not keys count.
 	for _, ti := range tables {
diff --git a/fb/BlockOffset.go b/fb/BlockOffset.go
new file mode 100644
index 0000000..6ef437e
--- /dev/null
+++ b/fb/BlockOffset.go
@@ -0,0 +1,104 @@
+// Code generated by the FlatBuffers compiler. DO NOT EDIT.
+
+package fb
+
+import (
+	flatbuffers "github.com/google/flatbuffers/go"
+)
+
+type BlockOffset struct {
+	_tab flatbuffers.Table
+}
+
+func GetRootAsBlockOffset(buf []byte, offset flatbuffers.UOffsetT) *BlockOffset {
+	n := flatbuffers.GetUOffsetT(buf[offset:])
+	x := &BlockOffset{}
+	x.Init(buf, n+offset)
+	return x
+}
+
+func (rcv *BlockOffset) Init(buf []byte, i flatbuffers.UOffsetT) {
+	rcv._tab.Bytes = buf
+	rcv._tab.Pos = i
+}
+
+func (rcv *BlockOffset) Table() flatbuffers.Table {
+	return rcv._tab
+}
+
+func (rcv *BlockOffset) Key(j int) byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		a := rcv._tab.Vector(o)
+		return rcv._tab.GetByte(a + flatbuffers.UOffsetT(j*1))
+	}
+	return 0
+}
+
+func (rcv *BlockOffset) KeyLength() int {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		return rcv._tab.VectorLen(o)
+	}
+	return 0
+}
+
+func (rcv *BlockOffset) KeyBytes() []byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		return rcv._tab.ByteVector(o + rcv._tab.Pos)
+	}
+	return nil
+}
+
+func (rcv *BlockOffset) MutateKey(j int, n byte) bool {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		a := rcv._tab.Vector(o)
+		return rcv._tab.MutateByte(a+flatbuffers.UOffsetT(j*1), n)
+	}
+	return false
+}
+
+func (rcv *BlockOffset) Offset() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *BlockOffset) MutateOffset(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(6, n)
+}
+
+func (rcv *BlockOffset) Len() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *BlockOffset) MutateLen(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(8, n)
+}
+
+func BlockOffsetStart(builder *flatbuffers.Builder) {
+	builder.StartObject(3)
+}
+func BlockOffsetAddKey(builder *flatbuffers.Builder, key flatbuffers.UOffsetT) {
+	builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(key), 0)
+}
+func BlockOffsetStartKeyVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+	return builder.StartVector(1, numElems, 1)
+}
+func BlockOffsetAddOffset(builder *flatbuffers.Builder, offset uint32) {
+	builder.PrependUint32Slot(1, offset, 0)
+}
+func BlockOffsetAddLen(builder *flatbuffers.Builder, len uint32) {
+	builder.PrependUint32Slot(2, len, 0)
+}
+func BlockOffsetEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+	return builder.EndObject()
+}
diff --git a/fb/TableIndex.go b/fb/TableIndex.go
new file mode 100644
index 0000000..72bea12
--- /dev/null
+++ b/fb/TableIndex.go
@@ -0,0 +1,160 @@
+// Code generated by the FlatBuffers compiler. DO NOT EDIT.
+
+package fb
+
+import (
+	flatbuffers "github.com/google/flatbuffers/go"
+)
+
+type TableIndex struct {
+	_tab flatbuffers.Table
+}
+
+func GetRootAsTableIndex(buf []byte, offset flatbuffers.UOffsetT) *TableIndex {
+	n := flatbuffers.GetUOffsetT(buf[offset:])
+	x := &TableIndex{}
+	x.Init(buf, n+offset)
+	return x
+}
+
+func (rcv *TableIndex) Init(buf []byte, i flatbuffers.UOffsetT) {
+	rcv._tab.Bytes = buf
+	rcv._tab.Pos = i
+}
+
+func (rcv *TableIndex) Table() flatbuffers.Table {
+	return rcv._tab
+}
+
+func (rcv *TableIndex) Offsets(obj *BlockOffset, j int) bool {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		x := rcv._tab.Vector(o)
+		x += flatbuffers.UOffsetT(j) * 4
+		x = rcv._tab.Indirect(x)
+		obj.Init(rcv._tab.Bytes, x)
+		return true
+	}
+	return false
+}
+
+func (rcv *TableIndex) OffsetsLength() int {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(4))
+	if o != 0 {
+		return rcv._tab.VectorLen(o)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) BloomFilter(j int) byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		a := rcv._tab.Vector(o)
+		return rcv._tab.GetByte(a + flatbuffers.UOffsetT(j*1))
+	}
+	return 0
+}
+
+func (rcv *TableIndex) BloomFilterLength() int {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		return rcv._tab.VectorLen(o)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) BloomFilterBytes() []byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		return rcv._tab.ByteVector(o + rcv._tab.Pos)
+	}
+	return nil
+}
+
+func (rcv *TableIndex) MutateBloomFilter(j int, n byte) bool {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(6))
+	if o != 0 {
+		a := rcv._tab.Vector(o)
+		return rcv._tab.MutateByte(a+flatbuffers.UOffsetT(j*1), n)
+	}
+	return false
+}
+
+func (rcv *TableIndex) EstimatedSize() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(8))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) MutateEstimatedSize(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(8, n)
+}
+
+func (rcv *TableIndex) MaxVersion() uint64 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
+	if o != 0 {
+		return rcv._tab.GetUint64(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) MutateMaxVersion(n uint64) bool {
+	return rcv._tab.MutateUint64Slot(10, n)
+}
+
+func (rcv *TableIndex) UncompressedSize() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) MutateUncompressedSize(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(12, n)
+}
+
+func (rcv *TableIndex) KeyCount() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) MutateKeyCount(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(14, n)
+}
+
+func TableIndexStart(builder *flatbuffers.Builder) {
+	builder.StartObject(6)
+}
+func TableIndexAddOffsets(builder *flatbuffers.Builder, offsets flatbuffers.UOffsetT) {
+	builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(offsets), 0)
+}
+func TableIndexStartOffsetsVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+	return builder.StartVector(4, numElems, 4)
+}
+func TableIndexAddBloomFilter(builder *flatbuffers.Builder, bloomFilter flatbuffers.UOffsetT) {
+	builder.PrependUOffsetTSlot(1, flatbuffers.UOffsetT(bloomFilter), 0)
+}
+func TableIndexStartBloomFilterVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
+	return builder.StartVector(1, numElems, 1)
+}
+func TableIndexAddEstimatedSize(builder *flatbuffers.Builder, estimatedSize uint32) {
+	builder.PrependUint32Slot(2, estimatedSize, 0)
+}
+func TableIndexAddMaxVersion(builder *flatbuffers.Builder, maxVersion uint64) {
+	builder.PrependUint64Slot(3, maxVersion, 0)
+}
+func TableIndexAddUncompressedSize(builder *flatbuffers.Builder, uncompressedSize uint32) {
+	builder.PrependUint32Slot(4, uncompressedSize, 0)
+}
+func TableIndexAddKeyCount(builder *flatbuffers.Builder, keyCount uint32) {
+	builder.PrependUint32Slot(5, keyCount, 0)
+}
+func TableIndexEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
+	return builder.EndObject()
+}
diff --git a/fb/flatbuffer.fbs b/fb/flatbuffer.fbs
new file mode 100644
index 0000000..580f183
--- /dev/null
+++ b/fb/flatbuffer.fbs
@@ -0,0 +1,35 @@
+/*
+ * Copyright 2020 Dgraph Labs, Inc. and Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+namespace fb;
+
+table TableIndex {
+  offsets:[BlockOffset];
+  bloom_filter:[ubyte];
+  estimated_size:uint32;
+  max_version:uint64;
+  uncompressed_size:uint32;
+  key_count:uint32;
+}
+
+table BlockOffset {
+  key:[ubyte];
+  offset:uint;
+  len:uint;
+}
+
+root_type TableIndex;
+root_type BlockOffset;
diff --git a/fb/gen.sh b/fb/gen.sh
new file mode 100755
index 0000000..90d88b4
--- /dev/null
+++ b/fb/gen.sh
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+
+set -e
+
+## Install flatc if not present
+## ref. https://google.github.io/flatbuffers/flatbuffers_guide_building.html
+command -v flatc > /dev/null || { ./install_flatbuffers.sh ; }
+
+flatc --go flatbuffer.fbs
+# Move files to the correct directory.
+mv fb/* ./
+rmdir fb
diff --git a/fb/install_flatbuffers.sh b/fb/install_flatbuffers.sh
new file mode 100755
index 0000000..ae07da7
--- /dev/null
+++ b/fb/install_flatbuffers.sh
@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+
+set -e
+
+install_mac() {
+  command -v brew > /dev/null || \
+    { echo "[ERROR]: 'brew' command not not found. Exiting" 1>&2; exit 1; }
+  brew install flatbuffers
+}
+
+install_linux() {
+  for CMD in curl cmake g++ make; do
+    command -v $CMD > /dev/null || \
+      { echo "[ERROR]: '$CMD' command not not found. Exiting" 1>&2; exit 1; }
+  done
+
+  ## Create Temp Build Directory
+  BUILD_DIR=$(mktemp -d)
+  pushd $BUILD_DIR
+
+  ## Fetch Latest Tarball
+  LATEST_VERSION=$(curl -s https://api.github.com/repos/google/flatbuffers/releases/latest | grep -oP '(?<=tag_name": ")[^"]+')
+  curl -sLO   https://github.com/google/flatbuffers/archive/$LATEST_VERSION.tar.gz
+  tar xf $LATEST_VERSION.tar.gz
+
+  ## Build Binaries
+  cd flatbuffers-${LATEST_VERSION#v}
+  cmake -G "Unix Makefiles" -DCMAKE_BUILD_TYPE=Release
+  make
+  ./flattests
+  cp flatc /usr/local/bin/flatc
+
+  ## Cleanup Temp Build Directory
+  popd
+  rm -rf $BUILD_DIR
+}
+
+SYSTEM=$(uname -s)
+
+case ${SYSTEM,,} in
+  linux)
+    sudo bash -c "$(declare -f install_linux); install_linux"
+    ;;
+  darwin)
+    install_mac
+    ;;
+esac
diff --git a/go.mod b/go.mod
index 33db830..dfa7aae 100644
--- a/go.mod
+++ b/go.mod
@@ -2,14 +2,16 @@ module github.com/dgraph-io/badger/v2
 
 go 1.12
 
+// replace github.com/dgraph-io/ristretto => /home/mrjn/go/src/github.com/dgraph-io/ristretto
+
 require (
 	github.com/DataDog/zstd v1.4.1
 	github.com/cespare/xxhash v1.1.0
-	github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6
-	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2
+	github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8
 	github.com/dustin/go-humanize v1.0.0
 	github.com/golang/protobuf v1.3.1
 	github.com/golang/snappy v0.0.1
+	github.com/google/flatbuffers v1.12.0
 	github.com/kr/pretty v0.1.0 // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
diff --git a/go.sum b/go.sum
index bbb99f9..63935e6 100644
--- a/go.sum
+++ b/go.sum
@@ -13,8 +13,8 @@ github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwc
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6 h1:maqkH6zkcUx1PP+LpbLMef2IOSz2/mLYT3RqecFcdDI=
-github.com/dgraph-io/ristretto v0.0.4-0.20200930150433-e1609c8d4ca6/go.mod h1:bDI4cDaalvYSji3vBVDKrn9ouDZrwN974u8ZO/AhYXs=
+github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8 h1:LBKxHZx6GX+vkwoXO7Zd69/VSvyB3BMDBilnqQ4c/Ws=
+github.com/dgraph-io/ristretto v0.0.4-0.20201003140040-36af8a316fc8/go.mod h1:bDI4cDaalvYSji3vBVDKrn9ouDZrwN974u8ZO/AhYXs=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
@@ -24,6 +24,8 @@ github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
 github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/google/flatbuffers v1.12.0 h1:/PtAHvnBY4Kqnx/xCQ3OIV9uYcSFGScBsWI3Oogeh6w=
+github.com/google/flatbuffers v1.12.0/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
diff --git a/iterator.go b/iterator.go
index 71045d1..ff9e9be 100644
--- a/iterator.go
+++ b/iterator.go
@@ -27,7 +27,6 @@ import (
 
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/table"
-	"github.com/dgryski/go-farm"
 
 	"github.com/dgraph-io/badger/v2/y"
 )
@@ -332,7 +331,8 @@ func (opt *IteratorOptions) pickTable(t table.TableInterface) bool {
 	}
 	// Bloom filter lookup would only work if opt.Prefix does NOT have the read
 	// timestamp as part of the key.
-	if opt.prefixIsKey && t.DoesNotHave(farm.Fingerprint64(opt.Prefix)) {
+	if opt.prefixIsKey && t.DoesNotHave(y.Hash(opt.Prefix)) {
+		y.NumLSMBloomHits.Add("pickTable", 1)
 		return false
 	}
 	return true
@@ -365,19 +365,10 @@ func (opt *IteratorOptions) pickTables(all []*table.Table) []*table.Table {
 	}
 
 	var out []*table.Table
-	hash := farm.Fingerprint64(opt.Prefix)
 	for _, t := range filtered {
-		// When we encounter the first table whose smallest key is higher than
-		// opt.Prefix, we can stop.
-		if opt.compareToPrefix(t.Smallest()) > 0 {
-			return out
+		if opt.pickTable(t) {
+			out = append(out, t)
 		}
-		// opt.Prefix is actually the key. So, we can run bloom filter checks
-		// as well.
-		if t.DoesNotHave(hash) {
-			continue
-		}
-		out = append(out, t)
 	}
 	return out
 }
diff --git a/level_handler.go b/level_handler.go
index dfae862..68a77b0 100644
--- a/level_handler.go
+++ b/level_handler.go
@@ -21,8 +21,6 @@ import (
 	"sort"
 	"sync"
 
-	"github.com/dgryski/go-farm"
-
 	"github.com/dgraph-io/badger/v2/table"
 	"github.com/dgraph-io/badger/v2/y"
 	"github.com/pkg/errors"
@@ -259,7 +257,7 @@ func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
 	tables, decr := s.getTableForKey(key)
 	keyNoTs := y.ParseKey(key)
 
-	hash := farm.Fingerprint64(keyNoTs)
+	hash := y.Hash(keyNoTs)
 	var maxVs y.ValueStruct
 	for _, th := range tables {
 		if th.DoesNotHave(hash) {
diff --git a/levels.go b/levels.go
index 823569c..c07c645 100644
--- a/levels.go
+++ b/levels.go
@@ -18,6 +18,7 @@ package badger
 
 import (
 	"bytes"
+	"encoding/hex"
 	"fmt"
 	"math/rand"
 	"os"
@@ -841,23 +842,16 @@ func (s *levelsController) fillTablesL0(cd *compactDef) bool {
 	return true
 }
 
-// sortByOverlap sorts tables in increasing order of overlap with next level.
-func (s *levelsController) sortByOverlap(tables []*table.Table, cd *compactDef) {
+// sortByHeuristic sorts tables in increasing order of MaxVersion, so we
+// compact older tables first.
+func (s *levelsController) sortByHeuristic(tables []*table.Table, cd *compactDef) {
 	if len(tables) == 0 || cd.nextLevel == nil {
 		return
 	}
 
-	tableOverlap := make([]int, len(tables))
-	for i := range tables {
-		// get key range for table
-		tableRange := getKeyRange(tables[i])
-		// get overlap with next level
-		left, right := cd.nextLevel.overlappingTables(levelHandlerRLocked{}, tableRange)
-		tableOverlap[i] = right - left
-	}
-
+	// Sort tables by max version. This is what RocksDB does.
 	sort.Slice(tables, func(i, j int) bool {
-		return tableOverlap[i] < tableOverlap[j]
+		return tables[i].MaxVersion() < tables[j].MaxVersion()
 	})
 }
 
@@ -870,15 +864,14 @@ func (s *levelsController) fillTables(cd *compactDef) bool {
 	if len(tables) == 0 {
 		return false
 	}
-
-	// We want to pick files from current level in order of increasing overlap with next level
-	// tables. Idea here is to first compact file from current level which has least overlap with
-	// next level. This provides us better write amplification.
-	s.sortByOverlap(tables, cd)
+	// We pick tables, so we compact older tables first. This is similar to
+	// kOldestLargestSeqFirst in RocksDB.
+	s.sortByHeuristic(tables, cd)
 
 	for _, t := range tables {
 		cd.thisSize = t.Size()
 		cd.thisRange = getKeyRange(t)
+		// If we're already compacting this range, don't do anything.
 		if s.cstatus.overlapsWith(cd.thisLevel.level, cd.thisRange) {
 			continue
 		}
@@ -962,6 +955,13 @@ func (s *levelsController) runCompactDef(l int, cd compactDef) (err error) {
 	s.kv.opt.Infof("LOG Compact %d->%d, del %d tables, add %d tables, took %v\n",
 		thisLevel.level, nextLevel.level, len(cd.top)+len(cd.bot),
 		len(newTables), time.Since(timeStart))
+
+	if cd.thisLevel.level != 0 && len(newTables) > 2*s.kv.opt.LevelSizeMultiplier {
+		s.kv.opt.Infof("This Range (numTables: %d)\nLeft:\n%s\nRight:\n%s\n",
+			len(cd.top), hex.Dump(cd.thisRange.left), hex.Dump(cd.thisRange.right))
+		s.kv.opt.Infof("Next Range (numTables: %d)\nLeft:\n%s\nRight:\n%s\n",
+			len(cd.bot), hex.Dump(cd.nextRange.left), hex.Dump(cd.nextRange.right))
+	}
 	return nil
 }
 
@@ -999,14 +999,12 @@ func (s *levelsController) doCompact(id int, p compactionPriority) error {
 
 	s.kv.opt.Infof("[Compactor: %d] Running compaction: %+v for level: %d\n",
 		id, p, cd.thisLevel.level)
-	s.cstatus.toLog(cd.elog)
 	if err := s.runCompactDef(l, cd); err != nil {
 		// This compaction couldn't be done successfully.
 		s.kv.opt.Warningf("[Compactor: %d] LOG Compact FAILED with error: %+v: %+v", id, err, cd)
 		return err
 	}
 
-	s.cstatus.toLog(cd.elog)
 	s.kv.opt.Infof("[Compactor: %d] Compaction for level: %d DONE", id, cd.thisLevel.level)
 	return nil
 }
@@ -1128,38 +1126,31 @@ func (s *levelsController) appendIterators(
 
 // TableInfo represents the information about a table.
 type TableInfo struct {
-	ID              uint64
-	Level           int
-	Left            []byte
-	Right           []byte
-	KeyCount        uint64 // Number of keys in the table
-	EstimatedSz     uint64
-	IndexSz         int
-	BloomFilterSize int
+	ID               uint64
+	Level            int
+	Left             []byte
+	Right            []byte
+	KeyCount         uint32 // Number of keys in the table
+	EstimatedSz      uint32
+	UncompressedSize uint32
+	IndexSz          int
+	BloomFilterSize  int
 }
 
-func (s *levelsController) getTableInfo(withKeysCount bool) (result []TableInfo) {
+func (s *levelsController) getTableInfo() (result []TableInfo) {
 	for _, l := range s.levels {
 		l.RLock()
 		for _, t := range l.tables {
-			var count uint64
-			if withKeysCount {
-				it := t.NewIterator(table.NOCACHE)
-				for it.Rewind(); it.Valid(); it.Next() {
-					count++
-				}
-				it.Close()
-			}
-
 			info := TableInfo{
-				ID:              t.ID(),
-				Level:           l.level,
-				Left:            t.Smallest(),
-				Right:           t.Biggest(),
-				KeyCount:        count,
-				EstimatedSz:     t.EstimatedSize(),
-				IndexSz:         t.IndexSize(),
-				BloomFilterSize: t.BloomFilterSize(),
+				ID:               t.ID(),
+				Level:            l.level,
+				Left:             t.Smallest(),
+				Right:            t.Biggest(),
+				KeyCount:         t.KeyCount(),
+				EstimatedSz:      t.EstimatedSize(),
+				IndexSz:          t.IndexSize(),
+				BloomFilterSize:  t.BloomFilterSize(),
+				UncompressedSize: t.UncompressedSize(),
 			}
 			result = append(result, info)
 		}
diff --git a/pb/badgerpb2.pb.go b/pb/badgerpb2.pb.go
index edacec7..ba6f93b 100644
--- a/pb/badgerpb2.pb.go
+++ b/pb/badgerpb2.pb.go
@@ -91,7 +91,7 @@ func (x Checksum_Algorithm) String() string {
 }
 
 func (Checksum_Algorithm) EnumDescriptor() ([]byte, []int) {
-	return fileDescriptor_e63e84f9f0d3998c, []int{6, 0}
+	return fileDescriptor_e63e84f9f0d3998c, []int{4, 0}
 }
 
 type KV struct {
@@ -381,132 +381,6 @@ func (m *ManifestChange) GetCompression() uint32 {
 	return 0
 }
 
-type BlockOffset struct {
-	Key                  []byte   `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
-	Offset               uint32   `protobuf:"varint,2,opt,name=offset,proto3" json:"offset,omitempty"`
-	Len                  uint32   `protobuf:"varint,3,opt,name=len,proto3" json:"len,omitempty"`
-	XXX_NoUnkeyedLiteral struct{} `json:"-"`
-	XXX_unrecognized     []byte   `json:"-"`
-	XXX_sizecache        int32    `json:"-"`
-}
-
-func (m *BlockOffset) Reset()         { *m = BlockOffset{} }
-func (m *BlockOffset) String() string { return proto.CompactTextString(m) }
-func (*BlockOffset) ProtoMessage()    {}
-func (*BlockOffset) Descriptor() ([]byte, []int) {
-	return fileDescriptor_e63e84f9f0d3998c, []int{4}
-}
-func (m *BlockOffset) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *BlockOffset) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_BlockOffset.Marshal(b, m, deterministic)
-	} else {
-		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
-		if err != nil {
-			return nil, err
-		}
-		return b[:n], nil
-	}
-}
-func (m *BlockOffset) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_BlockOffset.Merge(m, src)
-}
-func (m *BlockOffset) XXX_Size() int {
-	return m.Size()
-}
-func (m *BlockOffset) XXX_DiscardUnknown() {
-	xxx_messageInfo_BlockOffset.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_BlockOffset proto.InternalMessageInfo
-
-func (m *BlockOffset) GetKey() []byte {
-	if m != nil {
-		return m.Key
-	}
-	return nil
-}
-
-func (m *BlockOffset) GetOffset() uint32 {
-	if m != nil {
-		return m.Offset
-	}
-	return 0
-}
-
-func (m *BlockOffset) GetLen() uint32 {
-	if m != nil {
-		return m.Len
-	}
-	return 0
-}
-
-type TableIndex struct {
-	Offsets              []*BlockOffset `protobuf:"bytes,1,rep,name=offsets,proto3" json:"offsets,omitempty"`
-	BloomFilter          []byte         `protobuf:"bytes,2,opt,name=bloom_filter,json=bloomFilter,proto3" json:"bloom_filter,omitempty"`
-	EstimatedSize        uint64         `protobuf:"varint,3,opt,name=estimated_size,json=estimatedSize,proto3" json:"estimated_size,omitempty"`
-	XXX_NoUnkeyedLiteral struct{}       `json:"-"`
-	XXX_unrecognized     []byte         `json:"-"`
-	XXX_sizecache        int32          `json:"-"`
-}
-
-func (m *TableIndex) Reset()         { *m = TableIndex{} }
-func (m *TableIndex) String() string { return proto.CompactTextString(m) }
-func (*TableIndex) ProtoMessage()    {}
-func (*TableIndex) Descriptor() ([]byte, []int) {
-	return fileDescriptor_e63e84f9f0d3998c, []int{5}
-}
-func (m *TableIndex) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *TableIndex) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_TableIndex.Marshal(b, m, deterministic)
-	} else {
-		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
-		if err != nil {
-			return nil, err
-		}
-		return b[:n], nil
-	}
-}
-func (m *TableIndex) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_TableIndex.Merge(m, src)
-}
-func (m *TableIndex) XXX_Size() int {
-	return m.Size()
-}
-func (m *TableIndex) XXX_DiscardUnknown() {
-	xxx_messageInfo_TableIndex.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_TableIndex proto.InternalMessageInfo
-
-func (m *TableIndex) GetOffsets() []*BlockOffset {
-	if m != nil {
-		return m.Offsets
-	}
-	return nil
-}
-
-func (m *TableIndex) GetBloomFilter() []byte {
-	if m != nil {
-		return m.BloomFilter
-	}
-	return nil
-}
-
-func (m *TableIndex) GetEstimatedSize() uint64 {
-	if m != nil {
-		return m.EstimatedSize
-	}
-	return 0
-}
-
 type Checksum struct {
 	Algo                 Checksum_Algorithm `protobuf:"varint,1,opt,name=algo,proto3,enum=badgerpb2.Checksum_Algorithm" json:"algo,omitempty"`
 	Sum                  uint64             `protobuf:"varint,2,opt,name=sum,proto3" json:"sum,omitempty"`
@@ -519,7 +393,7 @@ func (m *Checksum) Reset()         { *m = Checksum{} }
 func (m *Checksum) String() string { return proto.CompactTextString(m) }
 func (*Checksum) ProtoMessage()    {}
 func (*Checksum) Descriptor() ([]byte, []int) {
-	return fileDescriptor_e63e84f9f0d3998c, []int{6}
+	return fileDescriptor_e63e84f9f0d3998c, []int{4}
 }
 func (m *Checksum) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -576,7 +450,7 @@ func (m *DataKey) Reset()         { *m = DataKey{} }
 func (m *DataKey) String() string { return proto.CompactTextString(m) }
 func (*DataKey) ProtoMessage()    {}
 func (*DataKey) Descriptor() ([]byte, []int) {
-	return fileDescriptor_e63e84f9f0d3998c, []int{7}
+	return fileDescriptor_e63e84f9f0d3998c, []int{5}
 }
 func (m *DataKey) XXX_Unmarshal(b []byte) error {
 	return m.Unmarshal(b)
@@ -641,8 +515,6 @@ func init() {
 	proto.RegisterType((*KVList)(nil), "badgerpb2.KVList")
 	proto.RegisterType((*ManifestChangeSet)(nil), "badgerpb2.ManifestChangeSet")
 	proto.RegisterType((*ManifestChange)(nil), "badgerpb2.ManifestChange")
-	proto.RegisterType((*BlockOffset)(nil), "badgerpb2.BlockOffset")
-	proto.RegisterType((*TableIndex)(nil), "badgerpb2.TableIndex")
 	proto.RegisterType((*Checksum)(nil), "badgerpb2.Checksum")
 	proto.RegisterType((*DataKey)(nil), "badgerpb2.DataKey")
 }
@@ -650,51 +522,44 @@ func init() {
 func init() { proto.RegisterFile("badgerpb2.proto", fileDescriptor_e63e84f9f0d3998c) }
 
 var fileDescriptor_e63e84f9f0d3998c = []byte{
-	// 689 bytes of a gzipped FileDescriptorProto
-	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x54, 0xcd, 0x6e, 0xda, 0x40,
-	0x10, 0xc6, 0xc6, 0xe1, 0x67, 0x08, 0x84, 0xae, 0xda, 0xc8, 0x51, 0x15, 0x4a, 0x1c, 0x45, 0x45,
-	0x95, 0x0a, 0x2d, 0x54, 0xbd, 0x13, 0x42, 0x15, 0x44, 0x22, 0xa4, 0x4d, 0x14, 0x45, 0xbd, 0xa0,
-	0xc5, 0x1e, 0xc0, 0xc2, 0x7f, 0xf2, 0x2e, 0x56, 0xc8, 0x13, 0xf4, 0xd2, 0x7b, 0x1f, 0xa9, 0xc7,
-	0x1e, 0xfa, 0x00, 0x55, 0xfa, 0x22, 0x95, 0xd7, 0x86, 0x82, 0xd4, 0xde, 0x66, 0xbe, 0xf9, 0x76,
-	0x67, 0xbf, 0x6f, 0xc6, 0x86, 0x83, 0x09, 0xb3, 0x66, 0x18, 0x06, 0x93, 0x76, 0x33, 0x08, 0x7d,
-	0xe1, 0x93, 0xe2, 0x06, 0x30, 0x7e, 0x2a, 0xa0, 0x0e, 0xef, 0x48, 0x15, 0xb2, 0x0b, 0x5c, 0xe9,
-	0x4a, 0x5d, 0x69, 0xec, 0xd3, 0x38, 0x24, 0xcf, 0x61, 0x2f, 0x62, 0xce, 0x12, 0x75, 0x55, 0x62,
-	0x49, 0x42, 0x5e, 0x42, 0x71, 0xc9, 0x31, 0x1c, 0xbb, 0x28, 0x98, 0x9e, 0x95, 0x95, 0x42, 0x0c,
-	0x5c, 0xa3, 0x60, 0x44, 0x87, 0x7c, 0x84, 0x21, 0xb7, 0x7d, 0x4f, 0xd7, 0xea, 0x4a, 0x43, 0xa3,
-	0xeb, 0x94, 0x1c, 0x03, 0xe0, 0x43, 0x60, 0x87, 0xc8, 0xc7, 0x4c, 0xe8, 0x7b, 0xb2, 0x58, 0x4c,
-	0x91, 0xae, 0x20, 0x04, 0x34, 0x79, 0x61, 0x4e, 0x5e, 0x28, 0xe3, 0xb8, 0x13, 0x17, 0x21, 0x32,
-	0x77, 0x6c, 0x5b, 0x3a, 0xd4, 0x95, 0x46, 0x99, 0x16, 0x12, 0x60, 0x60, 0x91, 0x57, 0x50, 0x4a,
-	0x8b, 0x96, 0xef, 0xa1, 0x5e, 0xaa, 0x2b, 0x8d, 0x02, 0x85, 0x04, 0xba, 0xf0, 0x3d, 0x34, 0x5e,
-	0x43, 0x6e, 0x78, 0x77, 0x65, 0x73, 0x41, 0x8e, 0x41, 0x5d, 0x44, 0xba, 0x52, 0xcf, 0x36, 0x4a,
-	0xed, 0x72, 0xf3, 0xaf, 0x13, 0xc3, 0x3b, 0xaa, 0x2e, 0x22, 0xe3, 0x12, 0x9e, 0x5d, 0x33, 0xcf,
-	0x9e, 0x22, 0x17, 0xbd, 0x39, 0xf3, 0x66, 0x78, 0x83, 0x82, 0x74, 0x20, 0x6f, 0xca, 0x84, 0xa7,
-	0x07, 0x8f, 0xb6, 0x0e, 0xee, 0xd2, 0xe9, 0x9a, 0x69, 0x7c, 0x55, 0xa1, 0xb2, 0x5b, 0x23, 0x15,
-	0x50, 0x07, 0x96, 0x34, 0x55, 0xa3, 0xea, 0xc0, 0x22, 0x1d, 0x50, 0x47, 0x81, 0x34, 0xb4, 0xd2,
-	0x3e, 0xfd, 0xef, 0x95, 0xcd, 0x51, 0x80, 0x21, 0x13, 0xb6, 0xef, 0x51, 0x75, 0x14, 0xc4, 0x83,
-	0xb8, 0xc2, 0x08, 0x1d, 0x69, 0x77, 0x99, 0x26, 0x09, 0x79, 0x01, 0xb9, 0x05, 0xae, 0x62, 0x6f,
-	0x12, 0xab, 0xf7, 0x16, 0xb8, 0x1a, 0x58, 0xe4, 0x1c, 0x0e, 0xd0, 0x33, 0xc3, 0x55, 0x10, 0x1f,
-	0x1f, 0x33, 0x67, 0xe6, 0x4b, 0xb7, 0x2b, 0x3b, 0x0a, 0xfa, 0x1b, 0x46, 0xd7, 0x99, 0xf9, 0xb4,
-	0x82, 0x3b, 0x39, 0xa9, 0x43, 0xc9, 0xf4, 0xdd, 0x20, 0x44, 0x2e, 0x47, 0x99, 0x93, 0x6d, 0xb7,
-	0x21, 0xe3, 0x14, 0x8a, 0x9b, 0x37, 0x12, 0x80, 0x5c, 0x8f, 0xf6, 0xbb, 0xb7, 0xfd, 0x6a, 0x26,
-	0x8e, 0x2f, 0xfa, 0x57, 0xfd, 0xdb, 0x7e, 0x55, 0x31, 0x06, 0x50, 0x3a, 0x77, 0x7c, 0x73, 0x31,
-	0x9a, 0x4e, 0x39, 0x8a, 0x7f, 0x6c, 0xd8, 0x21, 0xe4, 0x7c, 0x59, 0x93, 0x8e, 0x94, 0x69, 0x9a,
-	0xc5, 0x4c, 0x07, 0xbd, 0x54, 0x6e, 0x1c, 0x1a, 0x5f, 0x14, 0x80, 0x5b, 0x36, 0x71, 0x70, 0xe0,
-	0x59, 0xf8, 0x40, 0xde, 0x41, 0x3e, 0xa1, 0xae, 0xc7, 0x73, 0xb8, 0x25, 0x6e, 0xab, 0x27, 0x5d,
-	0xd3, 0xc8, 0x09, 0xec, 0x4f, 0x1c, 0xdf, 0x77, 0xc7, 0x53, 0xdb, 0x11, 0x18, 0xa6, 0x3b, 0x5d,
-	0x92, 0xd8, 0x27, 0x09, 0x91, 0x33, 0xa8, 0x20, 0x17, 0xb6, 0xcb, 0x04, 0x5a, 0x63, 0x6e, 0x3f,
-	0xa2, 0x7c, 0x80, 0x46, 0xcb, 0x1b, 0xf4, 0xc6, 0x7e, 0x44, 0x23, 0x82, 0x42, 0x6f, 0x8e, 0xe6,
-	0x82, 0x2f, 0x5d, 0xf2, 0x1e, 0x34, 0xe9, 0xb0, 0x22, 0x1d, 0x3e, 0xde, 0x7a, 0xc4, 0x9a, 0xd2,
-	0x8c, 0x0d, 0x0d, 0x6d, 0x31, 0x77, 0xa9, 0xa4, 0xc6, 0xda, 0xf8, 0xd2, 0x95, 0xfd, 0x35, 0x1a,
-	0x87, 0xc6, 0x19, 0x14, 0x37, 0xa4, 0xc4, 0xcb, 0x5e, 0xa7, 0xdd, 0xab, 0x66, 0xc8, 0x3e, 0x14,
-	0xee, 0xef, 0x2f, 0x19, 0x9f, 0x7f, 0xfc, 0x50, 0x55, 0x0c, 0x13, 0xf2, 0x17, 0x4c, 0xb0, 0x21,
-	0xae, 0xb6, 0x46, 0xaf, 0x6c, 0x8f, 0x9e, 0x80, 0x66, 0x31, 0xc1, 0x52, 0x6d, 0x32, 0x8e, 0x17,
-	0xd0, 0x8e, 0xd2, 0xef, 0x54, 0xb5, 0xa3, 0xf8, 0x3b, 0x34, 0x43, 0x94, 0x12, 0x99, 0x90, 0x9b,
-	0x93, 0xa5, 0xc5, 0x14, 0xe9, 0x8a, 0x37, 0x47, 0x50, 0xd9, 0xdd, 0x0d, 0x92, 0x87, 0x2c, 0x43,
-	0x5e, 0xcd, 0x9c, 0x77, 0xbe, 0x3f, 0xd5, 0x94, 0x1f, 0x4f, 0x35, 0xe5, 0xd7, 0x53, 0x4d, 0xf9,
-	0xf6, 0xbb, 0x96, 0xf9, 0x7c, 0x32, 0xb3, 0xc5, 0x7c, 0x39, 0x69, 0x9a, 0xbe, 0xdb, 0xb2, 0x66,
-	0x21, 0x0b, 0xe6, 0x6f, 0x6d, 0xbf, 0x95, 0x78, 0xd0, 0x8a, 0xda, 0xad, 0x60, 0x32, 0xc9, 0xc9,
-	0xdf, 0x4d, 0xe7, 0x4f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x88, 0x22, 0x82, 0x98, 0x81, 0x04, 0x00,
-	0x00,
+	// 589 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x74, 0x93, 0xcf, 0x6e, 0xda, 0x4e,
+	0x10, 0xc7, 0x59, 0xe3, 0xf0, 0x67, 0x48, 0x08, 0xbf, 0xd5, 0xaf, 0x92, 0xa3, 0x2a, 0x94, 0x3a,
+	0xaa, 0x8a, 0x2a, 0x15, 0x54, 0xa8, 0x7a, 0x27, 0x04, 0x29, 0x88, 0x44, 0x91, 0xb6, 0x51, 0x14,
+	0xf5, 0x82, 0x16, 0x7b, 0x0a, 0x16, 0xf8, 0x8f, 0x76, 0x17, 0xab, 0x3c, 0x44, 0xef, 0x7d, 0xa4,
+	0x1e, 0x7b, 0xe8, 0x03, 0x54, 0xe9, 0x8b, 0x54, 0xbb, 0x36, 0x14, 0x0e, 0xbd, 0xcd, 0x7c, 0xe7,
+	0xbb, 0x3b, 0xe3, 0xcf, 0x8e, 0xe1, 0x74, 0xc6, 0xfd, 0x39, 0x8a, 0x64, 0xd6, 0xeb, 0x24, 0x22,
+	0x56, 0x31, 0xad, 0xee, 0x04, 0xf7, 0x27, 0x01, 0x6b, 0xf2, 0x40, 0x1b, 0x50, 0x5c, 0xe2, 0xc6,
+	0x21, 0x2d, 0xd2, 0x3e, 0x66, 0x3a, 0xa4, 0xff, 0xc3, 0x51, 0xca, 0x57, 0x6b, 0x74, 0x2c, 0xa3,
+	0x65, 0x09, 0x7d, 0x0e, 0xd5, 0xb5, 0x44, 0x31, 0x0d, 0x51, 0x71, 0xa7, 0x68, 0x2a, 0x15, 0x2d,
+	0xdc, 0xa2, 0xe2, 0xd4, 0x81, 0x72, 0x8a, 0x42, 0x06, 0x71, 0xe4, 0xd8, 0x2d, 0xd2, 0xb6, 0xd9,
+	0x36, 0xa5, 0xe7, 0x00, 0xf8, 0x25, 0x09, 0x04, 0xca, 0x29, 0x57, 0xce, 0x91, 0x29, 0x56, 0x73,
+	0x65, 0xa0, 0x28, 0x05, 0xdb, 0x5c, 0x58, 0x32, 0x17, 0x9a, 0x58, 0x77, 0x92, 0x4a, 0x20, 0x0f,
+	0xa7, 0x81, 0xef, 0x40, 0x8b, 0xb4, 0x4f, 0x58, 0x25, 0x13, 0xc6, 0x3e, 0x7d, 0x01, 0xb5, 0xbc,
+	0xe8, 0xc7, 0x11, 0x3a, 0xb5, 0x16, 0x69, 0x57, 0x18, 0x64, 0xd2, 0x55, 0x1c, 0xa1, 0xfb, 0x1a,
+	0x4a, 0x93, 0x87, 0x9b, 0x40, 0x2a, 0x7a, 0x0e, 0xd6, 0x32, 0x75, 0x48, 0xab, 0xd8, 0xae, 0xf5,
+	0x4e, 0x3a, 0x7f, 0x49, 0x4c, 0x1e, 0x98, 0xb5, 0x4c, 0xdd, 0x6b, 0xf8, 0xef, 0x96, 0x47, 0xc1,
+	0x67, 0x94, 0x6a, 0xb8, 0xe0, 0xd1, 0x1c, 0x3f, 0xa2, 0xa2, 0x7d, 0x28, 0x7b, 0x26, 0x91, 0xf9,
+	0xc1, 0xb3, 0xbd, 0x83, 0x87, 0x76, 0xb6, 0x75, 0xba, 0x5f, 0x2d, 0xa8, 0x1f, 0xd6, 0x68, 0x1d,
+	0xac, 0xb1, 0x6f, 0xa0, 0xda, 0xcc, 0x1a, 0xfb, 0xb4, 0x0f, 0xd6, 0x5d, 0x62, 0x80, 0xd6, 0x7b,
+	0x17, 0xff, 0xbc, 0xb2, 0x73, 0x97, 0xa0, 0xe0, 0x2a, 0x88, 0x23, 0x66, 0xdd, 0x25, 0xfa, 0x21,
+	0x6e, 0x30, 0xc5, 0x95, 0xc1, 0x7d, 0xc2, 0xb2, 0x84, 0x3e, 0x83, 0xd2, 0x12, 0x37, 0x9a, 0x4d,
+	0x86, 0xfa, 0x68, 0x89, 0x9b, 0xb1, 0x4f, 0x2f, 0xe1, 0x14, 0x23, 0x4f, 0x6c, 0x12, 0x7d, 0x7c,
+	0xca, 0x57, 0xf3, 0xd8, 0xd0, 0xae, 0x1f, 0x7c, 0xc1, 0x68, 0xe7, 0x18, 0xac, 0xe6, 0x31, 0xab,
+	0xe3, 0x41, 0x4e, 0x5b, 0x50, 0xf3, 0xe2, 0x30, 0x11, 0x28, 0xcd, 0x53, 0x96, 0x4c, 0xdb, 0x7d,
+	0xc9, 0xbd, 0x80, 0xea, 0x6e, 0x46, 0x0a, 0x50, 0x1a, 0xb2, 0xd1, 0xe0, 0x7e, 0xd4, 0x28, 0xe8,
+	0xf8, 0x6a, 0x74, 0x33, 0xba, 0x1f, 0x35, 0x88, 0x9b, 0x42, 0x65, 0xb8, 0x40, 0x6f, 0x29, 0xd7,
+	0x21, 0x7d, 0x07, 0xb6, 0x99, 0x85, 0x98, 0x59, 0xce, 0xf7, 0x66, 0xd9, 0x5a, 0x3a, 0xba, 0xb5,
+	0x08, 0xd4, 0x22, 0x64, 0xc6, 0xaa, 0x37, 0x52, 0xae, 0x43, 0x03, 0xcb, 0x66, 0x3a, 0x74, 0x5f,
+	0x41, 0x75, 0x67, 0xca, 0xba, 0x0e, 0xfb, 0xbd, 0x61, 0xa3, 0x40, 0x8f, 0xa1, 0xf2, 0xf8, 0x78,
+	0xcd, 0xe5, 0xe2, 0xc3, 0xfb, 0x06, 0x71, 0x3d, 0x28, 0x5f, 0x71, 0xc5, 0x27, 0xb8, 0xd9, 0x83,
+	0x44, 0xf6, 0x21, 0x51, 0xb0, 0x7d, 0xae, 0x78, 0xbe, 0xd9, 0x26, 0xd6, 0x4f, 0x15, 0xa4, 0xf9,
+	0x46, 0x5b, 0x41, 0xaa, 0x37, 0xd6, 0x13, 0xc8, 0x15, 0xfa, 0x7a, 0x63, 0x35, 0xe3, 0x22, 0xab,
+	0xe6, 0xca, 0x40, 0xbd, 0x39, 0x83, 0xfa, 0x21, 0x45, 0x5a, 0x86, 0x22, 0x47, 0xd9, 0x28, 0x5c,
+	0xf6, 0xbf, 0x3f, 0x35, 0xc9, 0x8f, 0xa7, 0x26, 0xf9, 0xf5, 0xd4, 0x24, 0xdf, 0x7e, 0x37, 0x0b,
+	0x9f, 0x5e, 0xce, 0x03, 0xb5, 0x58, 0xcf, 0x3a, 0x5e, 0x1c, 0x76, 0xfd, 0xb9, 0xe0, 0xc9, 0xe2,
+	0x6d, 0x10, 0x77, 0x33, 0x06, 0xdd, 0xb4, 0xd7, 0x4d, 0x66, 0xb3, 0x92, 0xf9, 0x31, 0xfb, 0x7f,
+	0x02, 0x00, 0x00, 0xff, 0xff, 0x1d, 0xc3, 0x11, 0xa3, 0xab, 0x03, 0x00, 0x00,
 }
 
 func (m *KV) Marshal() (dAtA []byte, err error) {
@@ -916,103 +781,6 @@ func (m *ManifestChange) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	return len(dAtA) - i, nil
 }
 
-func (m *BlockOffset) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *BlockOffset) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *BlockOffset) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.Len != 0 {
-		i = encodeVarintBadgerpb2(dAtA, i, uint64(m.Len))
-		i--
-		dAtA[i] = 0x18
-	}
-	if m.Offset != 0 {
-		i = encodeVarintBadgerpb2(dAtA, i, uint64(m.Offset))
-		i--
-		dAtA[i] = 0x10
-	}
-	if len(m.Key) > 0 {
-		i -= len(m.Key)
-		copy(dAtA[i:], m.Key)
-		i = encodeVarintBadgerpb2(dAtA, i, uint64(len(m.Key)))
-		i--
-		dAtA[i] = 0xa
-	}
-	return len(dAtA) - i, nil
-}
-
-func (m *TableIndex) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *TableIndex) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *TableIndex) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if m.XXX_unrecognized != nil {
-		i -= len(m.XXX_unrecognized)
-		copy(dAtA[i:], m.XXX_unrecognized)
-	}
-	if m.EstimatedSize != 0 {
-		i = encodeVarintBadgerpb2(dAtA, i, uint64(m.EstimatedSize))
-		i--
-		dAtA[i] = 0x18
-	}
-	if len(m.BloomFilter) > 0 {
-		i -= len(m.BloomFilter)
-		copy(dAtA[i:], m.BloomFilter)
-		i = encodeVarintBadgerpb2(dAtA, i, uint64(len(m.BloomFilter)))
-		i--
-		dAtA[i] = 0x12
-	}
-	if len(m.Offsets) > 0 {
-		for iNdEx := len(m.Offsets) - 1; iNdEx >= 0; iNdEx-- {
-			{
-				size, err := m.Offsets[iNdEx].MarshalToSizedBuffer(dAtA[:i])
-				if err != nil {
-					return 0, err
-				}
-				i -= size
-				i = encodeVarintBadgerpb2(dAtA, i, uint64(size))
-			}
-			i--
-			dAtA[i] = 0xa
-		}
-	}
-	return len(dAtA) - i, nil
-}
-
 func (m *Checksum) Marshal() (dAtA []byte, err error) {
 	size := m.Size()
 	dAtA = make([]byte, size)
@@ -1218,53 +986,6 @@ func (m *ManifestChange) Size() (n int) {
 	return n
 }
 
-func (m *BlockOffset) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	l = len(m.Key)
-	if l > 0 {
-		n += 1 + l + sovBadgerpb2(uint64(l))
-	}
-	if m.Offset != 0 {
-		n += 1 + sovBadgerpb2(uint64(m.Offset))
-	}
-	if m.Len != 0 {
-		n += 1 + sovBadgerpb2(uint64(m.Len))
-	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
-	}
-	return n
-}
-
-func (m *TableIndex) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if len(m.Offsets) > 0 {
-		for _, e := range m.Offsets {
-			l = e.Size()
-			n += 1 + l + sovBadgerpb2(uint64(l))
-		}
-	}
-	l = len(m.BloomFilter)
-	if l > 0 {
-		n += 1 + l + sovBadgerpb2(uint64(l))
-	}
-	if m.EstimatedSize != 0 {
-		n += 1 + sovBadgerpb2(uint64(m.EstimatedSize))
-	}
-	if m.XXX_unrecognized != nil {
-		n += len(m.XXX_unrecognized)
-	}
-	return n
-}
-
 func (m *Checksum) Size() (n int) {
 	if m == nil {
 		return 0
@@ -1926,273 +1647,6 @@ func (m *ManifestChange) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *BlockOffset) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowBadgerpb2
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: BlockOffset: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: BlockOffset: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Key", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Key = append(m.Key[:0], dAtA[iNdEx:postIndex]...)
-			if m.Key == nil {
-				m.Key = []byte{}
-			}
-			iNdEx = postIndex
-		case 2:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Offset", wireType)
-			}
-			m.Offset = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.Offset |= uint32(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Len", wireType)
-			}
-			m.Len = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.Len |= uint32(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		default:
-			iNdEx = preIndex
-			skippy, err := skipBadgerpb2(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if skippy < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if (iNdEx + skippy) < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
-func (m *TableIndex) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowBadgerpb2
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: TableIndex: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: TableIndex: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Offsets", wireType)
-			}
-			var msglen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				msglen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if msglen < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			postIndex := iNdEx + msglen
-			if postIndex < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Offsets = append(m.Offsets, &BlockOffset{})
-			if err := m.Offsets[len(m.Offsets)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil {
-				return err
-			}
-			iNdEx = postIndex
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field BloomFilter", wireType)
-			}
-			var byteLen int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				byteLen |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			if byteLen < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			postIndex := iNdEx + byteLen
-			if postIndex < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.BloomFilter = append(m.BloomFilter[:0], dAtA[iNdEx:postIndex]...)
-			if m.BloomFilter == nil {
-				m.BloomFilter = []byte{}
-			}
-			iNdEx = postIndex
-		case 3:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field EstimatedSize", wireType)
-			}
-			m.EstimatedSize = 0
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowBadgerpb2
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				m.EstimatedSize |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-		default:
-			iNdEx = preIndex
-			skippy, err := skipBadgerpb2(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if skippy < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if (iNdEx + skippy) < 0 {
-				return ErrInvalidLengthBadgerpb2
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
 func (m *Checksum) Unmarshal(dAtA []byte) error {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/pb/badgerpb2.proto b/pb/badgerpb2.proto
index 00fad8c..862e147 100644
--- a/pb/badgerpb2.proto
+++ b/pb/badgerpb2.proto
@@ -61,18 +61,6 @@ message ManifestChange {
   uint32 compression = 6;   // Only used for CREATE Op.
 }
 
-message BlockOffset {
-  bytes key = 1;
-  uint32 offset = 2;
-  uint32 len = 3;
-}
-
-message TableIndex {
-  repeated BlockOffset offsets = 1;
-  bytes bloom_filter = 2;
-  uint64 estimated_size = 3;
-}
-
 message Checksum {
   enum Algorithm {
     CRC32C = 0;
diff --git a/table/builder.go b/table/builder.go
index 8cfba19..7c845fc 100644
--- a/table/builder.go
+++ b/table/builder.go
@@ -24,9 +24,10 @@ import (
 	"sync/atomic"
 	"unsafe"
 
-	"github.com/dgryski/go-farm"
+	"github.com/dgraph-io/badger/v2/fb"
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/snappy"
+	fbs "github.com/google/flatbuffers/go"
 	"github.com/pkg/errors"
 
 	"github.com/dgraph-io/badger/v2/options"
@@ -66,6 +67,7 @@ func (h *header) Decode(buf []byte) {
 	copy(((*[headerSize]byte)(unsafe.Pointer(h))[:]), buf[:headerSize])
 }
 
+// bblock represents a block that is being compressed/encrypted in the background.
 type bblock struct {
 	data  []byte
 	start uint32 // Points to the starting offset of the block.
@@ -80,12 +82,15 @@ type Builder struct {
 	bufLock    sync.Mutex // This lock guards the buf. We acquire lock when we resize the buf.
 	actualSize uint32     // Used to store the sum of sizes of blocks after compression/encryption.
 
-	baseKey      []byte   // Base key for the current block.
-	baseOffset   uint32   // Offset for the current block.
-	entryOffsets []uint32 // Offsets of entries present in current block.
-	tableIndex   *pb.TableIndex
-	keyHashes    []uint64 // Used for building the bloomfilter.
-	opt          *Options
+	baseKey    []byte // Base key for the current block.
+	baseOffset uint32 // Offset for the current block.
+
+	entryOffsets  []uint32 // Offsets of entries present in current block.
+	offsets       *z.Buffer
+	estimatedSize uint32
+	keyHashes     []uint32 // Used for building the bloomfilter.
+	opt           *Options
+	maxVersion    uint64
 
 	// Used to concurrently compress/encrypt blocks.
 	wg        sync.WaitGroup
@@ -98,10 +103,9 @@ func NewTableBuilder(opts Options) *Builder {
 	b := &Builder{
 		// Additional 16 MB to store index (approximate).
 		// We trim the additional space in table.Finish().
-		buf:        z.Calloc(int(opts.TableSize + 16*MB)),
-		tableIndex: &pb.TableIndex{},
-		keyHashes:  make([]uint64, 0, 1024), // Avoid some malloc calls.
-		opt:        &opts,
+		buf:     z.Calloc(int(opts.TableSize + 16*MB)),
+		opt:     &opts,
+		offsets: z.NewBuffer(1 << 20),
 	}
 
 	// If encryption or compression is not enabled, do not start compression/encryption goroutines
@@ -168,6 +172,7 @@ func (b *Builder) handleBlock() {
 
 // Close closes the TableBuilder.
 func (b *Builder) Close() {
+	b.offsets.Release()
 	z.Free(b.buf)
 }
 
@@ -185,8 +190,12 @@ func (b *Builder) keyDiff(newKey []byte) []byte {
 	return newKey[i:]
 }
 
-func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint64) {
-	b.keyHashes = append(b.keyHashes, farm.Fingerprint64(y.ParseKey(key)))
+func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint32) {
+	b.keyHashes = append(b.keyHashes, y.Hash(y.ParseKey(key)))
+
+	if version := y.ParseTs(key); version > b.maxVersion {
+		b.maxVersion = version
+	}
 
 	// diffKey stores the difference of key with baseKey.
 	var diffKey []byte
@@ -221,9 +230,9 @@ func (b *Builder) addHelper(key []byte, v y.ValueStruct, vpLen uint64) {
 	b.sz += v.Encode(b.buf[b.sz:])
 
 	// Size of KV on SST.
-	sstSz := uint64(uint32(headerSize) + uint32(len(diffKey)) + v.EncodedSize())
+	sstSz := uint32(headerSize) + uint32(len(diffKey)) + v.EncodedSize()
 	// Total estimated size = size on SST + size on vlog (length of value pointer).
-	b.tableIndex.EstimatedSize += (sstSz + vpLen)
+	b.estimatedSize += (sstSz + vpLen)
 }
 
 // grow increases the size of b.buf by atleast 50%.
@@ -297,12 +306,19 @@ func (b *Builder) finishBlock() {
 func (b *Builder) addBlockToIndex() {
 	blockBuf := b.buf[b.baseOffset:b.sz]
 	// Add key to the block index.
-	bo := &pb.BlockOffset{
-		Key:    y.Copy(b.baseKey),
-		Offset: b.baseOffset,
-		Len:    uint32(len(blockBuf)),
-	}
-	b.tableIndex.Offsets = append(b.tableIndex.Offsets, bo)
+	builder := fbs.NewBuilder(64)
+	off := builder.CreateByteVector(b.baseKey)
+
+	fb.BlockOffsetStart(builder)
+	fb.BlockOffsetAddKey(builder, off)
+	fb.BlockOffsetAddOffset(builder, b.baseOffset)
+	fb.BlockOffsetAddLen(builder, uint32(len(blockBuf)))
+	uoff := fb.BlockOffsetEnd(builder)
+	builder.Finish(uoff)
+
+	out := builder.FinishedBytes()
+	dst := b.offsets.SliceAllocate(len(out))
+	copy(dst, out)
 }
 
 func (b *Builder) shouldFinishBlock(key []byte, value y.ValueStruct) bool {
@@ -342,7 +358,7 @@ func (b *Builder) Add(key []byte, value y.ValueStruct, valueLen uint32) {
 		b.baseOffset = uint32((b.sz))
 		b.entryOffsets = b.entryOffsets[:0]
 	}
-	b.addHelper(key, value, uint64(valueLen))
+	b.addHelper(key, value, valueLen)
 }
 
 // TODO: vvv this was the comment on ReachedCapacity.
@@ -358,9 +374,10 @@ func (b *Builder) ReachedCapacity(capacity uint64) bool {
 		4 + // count of all entry offsets
 		8 + // checksum bytes
 		4 // checksum length
+
 	estimateSz := blocksSize +
 		4 + // Index length
-		5*(uint32(len(b.tableIndex.Offsets))) // approximate index size
+		uint32(b.offsets.Len())
 
 	return uint64(estimateSz) > capacity
 }
@@ -378,15 +395,6 @@ The table structure looks like
 */
 // In case the data is encrypted, the "IV" is added to the end of the index.
 func (b *Builder) Finish(allocate bool) []byte {
-	if b.opt.BloomFalsePositive > 0 {
-		bf := z.NewBloomFilter(float64(len(b.keyHashes)), b.opt.BloomFalsePositive)
-		for _, h := range b.keyHashes {
-			bf.Add(h)
-		}
-		// Add bloom filter to the index.
-		b.tableIndex.BloomFilter = bf.JSONMarshal()
-	}
-
 	b.finishBlock() // This will never start a new block.
 
 	if b.blockChan != nil {
@@ -395,32 +403,44 @@ func (b *Builder) Finish(allocate bool) []byte {
 	// Wait for block handler to finish.
 	b.wg.Wait()
 
+	// We have added padding after each block so we should minus the
+	// padding from the actual table size. len(blocklist) would be zero if
+	// there is no compression/encryption.
+	uncompressedSize := b.sz - uint32(padding*len(b.blockList))
 	dst := b.buf
 	// Fix block boundaries. This includes moving the blocks so that we
 	// don't have any interleaving space between them.
+	bo, next := []byte{}, 1
 	if len(b.blockList) > 0 {
 		dstLen := uint32(0)
-		for i, bl := range b.blockList {
-			off := b.tableIndex.Offsets[i]
+		for _, bl := range b.blockList {
+			bo, next = b.offsets.Slice(next)
 			// Length of the block is end minus the start.
-			off.Len = bl.end - bl.start
+			fbo := fb.GetRootAsBlockOffset(bo, 0)
+			fbo.MutateLen(bl.end - bl.start)
 			// New offset of the block is the point in the main buffer till
 			// which we have written data.
-			off.Offset = dstLen
+			fbo.MutateOffset(dstLen)
 
 			copy(dst[dstLen:], b.buf[bl.start:bl.end])
 
 			// New length is the start of the block plus its length.
-			dstLen = off.Offset + off.Len
+			dstLen = fbo.Offset() + fbo.Len()
 		}
+		y.AssertTrue(next == 0)
 		// Start writing to the buffer from the point until which we have valid data.
 		// Fix the length because append and writeChecksum also rely on it.
 		b.sz = dstLen
 	}
 
-	index, err := proto.Marshal(b.tableIndex)
-	y.Check(err)
+	var f y.Filter
+	if b.opt.BloomFalsePositive > 0 {
+		bits := y.BloomBitsPerKey(len(b.keyHashes), b.opt.BloomFalsePositive)
+		f = y.NewFilter(b.keyHashes, bits)
+	}
+	index := b.buildIndex(f, uncompressedSize)
 
+	var err error
 	if b.shouldEncrypt() {
 		index, err = b.encrypt(index, false)
 		y.Check(err)
@@ -518,3 +538,64 @@ func (b *Builder) compressData(data []byte) ([]byte, error) {
 	}
 	return nil, errors.New("Unsupported compression type")
 }
+
+func (b *Builder) buildIndex(bloom []byte, tableSz uint32) []byte {
+	builder := fbs.NewBuilder(3 << 20)
+
+	boList := b.writeBlockOffsets(builder)
+	// Write block offset vector the the idxBuilder.
+	fb.TableIndexStartOffsetsVector(builder, len(boList))
+
+	// Write individual block offsets.
+	for i := 0; i < len(boList); i++ {
+		builder.PrependUOffsetT(boList[i])
+	}
+	boEnd := builder.EndVector(len(boList))
+
+	var bfoff fbs.UOffsetT
+	// Write the bloom filter.
+	if len(bloom) > 0 {
+		bfoff = builder.CreateByteVector(bloom)
+	}
+
+	fb.TableIndexStart(builder)
+	fb.TableIndexAddOffsets(builder, boEnd)
+	fb.TableIndexAddBloomFilter(builder, bfoff)
+	fb.TableIndexAddEstimatedSize(builder, b.estimatedSize)
+	fb.TableIndexAddMaxVersion(builder, b.maxVersion)
+	fb.TableIndexAddUncompressedSize(builder, tableSz)
+	fb.TableIndexAddKeyCount(builder, uint32(len(b.keyHashes)))
+	builder.Finish(fb.TableIndexEnd(builder))
+
+	return builder.FinishedBytes()
+}
+
+// writeBlockOffsets writes all the blockOffets in b.offsets and returns the
+// offsets for the newly written items.
+func (b *Builder) writeBlockOffsets(builder *fbs.Builder) []fbs.UOffsetT {
+	so := b.offsets.SliceOffsets()
+	var uoffs []fbs.UOffsetT
+	for i := len(so) - 1; i >= 0; i-- {
+		// We add these in reverse order.
+		data, _ := b.offsets.Slice(so[i])
+		uoff := b.writeBlockOffset(builder, data)
+		uoffs = append(uoffs, uoff)
+	}
+	return uoffs
+}
+
+// writeBlockOffset writes the given key,offset,len triple to the indexBuilder.
+// It returns the offset of the newly written blockoffset.
+func (b *Builder) writeBlockOffset(builder *fbs.Builder, data []byte) fbs.UOffsetT {
+	// Write the key to the buffer.
+	bo := fb.GetRootAsBlockOffset(data, 0)
+
+	k := builder.CreateByteVector(bo.KeyBytes())
+
+	// Build the blockOffset.
+	fb.BlockOffsetStart(builder)
+	fb.BlockOffsetAddKey(builder, k)
+	fb.BlockOffsetAddOffset(builder, bo.Offset())
+	fb.BlockOffsetAddLen(builder, bo.Len())
+	return fb.BlockOffsetEnd(builder)
+}
diff --git a/table/iterator.go b/table/iterator.go
index f46ac74..0270ec9 100644
--- a/table/iterator.go
+++ b/table/iterator.go
@@ -21,6 +21,7 @@ import (
 	"io"
 	"sort"
 
+	"github.com/dgraph-io/badger/v2/fb"
 	"github.com/dgraph-io/badger/v2/y"
 	"github.com/pkg/errors"
 )
@@ -198,7 +199,7 @@ func (itr *Iterator) useCache() bool {
 }
 
 func (itr *Iterator) seekToFirst() {
-	numBlocks := itr.t.noOfBlocks
+	numBlocks := itr.t.offsetsLength()
 	if numBlocks == 0 {
 		itr.err = io.EOF
 		return
@@ -215,7 +216,7 @@ func (itr *Iterator) seekToFirst() {
 }
 
 func (itr *Iterator) seekToLast() {
-	numBlocks := itr.t.noOfBlocks
+	numBlocks := itr.t.offsetsLength()
 	if numBlocks == 0 {
 		itr.err = io.EOF
 		return
@@ -252,9 +253,11 @@ func (itr *Iterator) seekFrom(key []byte, whence int) {
 	case current:
 	}
 
-	idx := sort.Search(itr.t.noOfBlocks, func(idx int) bool {
-		ko := itr.t.blockOffsets()[idx]
-		return y.CompareKeys(ko.Key, key) > 0
+	var ko fb.BlockOffset
+	idx := sort.Search(itr.t.offsetsLength(), func(idx int) bool {
+		// Offsets should never return false since we're iterating within the OffsetsLength.
+		y.AssertTrue(itr.t.offsets(&ko, idx))
+		return y.CompareKeys(ko.KeyBytes(), key) > 0
 	})
 	if idx == 0 {
 		// The smallest key in our table is already strictly > key. We can return that.
@@ -272,7 +275,7 @@ func (itr *Iterator) seekFrom(key []byte, whence int) {
 	itr.seekHelper(idx-1, key)
 	if itr.err == io.EOF {
 		// Case 1. Need to visit block[idx].
-		if idx == itr.t.noOfBlocks {
+		if idx == itr.t.offsetsLength() {
 			// If idx == len(itr.t.blockIndex), then input key is greater than ANY element of table.
 			// There's nothing we can do. Valid() should return false as we seek to end of table.
 			return
@@ -300,7 +303,7 @@ func (itr *Iterator) seekForPrev(key []byte) {
 func (itr *Iterator) next() {
 	itr.err = nil
 
-	if itr.bpos >= itr.t.noOfBlocks {
+	if itr.bpos >= itr.t.offsetsLength() {
 		itr.err = io.EOF
 		return
 	}
diff --git a/table/table.go b/table/table.go
index c1730ef..1b26d4b 100644
--- a/table/table.go
+++ b/table/table.go
@@ -36,6 +36,7 @@ import (
 	"github.com/golang/snappy"
 	"github.com/pkg/errors"
 
+	"github.com/dgraph-io/badger/v2/fb"
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/y"
@@ -46,13 +47,6 @@ import (
 const fileSuffix = ".sst"
 const intSize = int(unsafe.Sizeof(int(0)))
 
-// 1 word = 8 bytes
-// sizeOfOffsetStruct is the size of pb.BlockOffset
-const sizeOfOffsetStruct int64 = 3*8 + // key array take 3 words
-	1*8 + // offset and len takes 1 word
-	3*8 + // XXX_unrecognized array takes 3 word.
-	1*8 // so far 7 words, in order to round the slab we're adding one more word.
-
 // Options contains configurable options for Table/Builder.
 type Options struct {
 	// Options for Opening/Building Table.
@@ -80,6 +74,7 @@ type Options struct {
 	// Compression indicates the compression algorithm used for block compression.
 	Compression options.CompressionType
 
+	// Block cache is used to cache decompressed and decrypted blocks.
 	BlockCache *ristretto.Cache
 	IndexCache *ristretto.Cache
 
@@ -95,7 +90,7 @@ type Options struct {
 type TableInterface interface {
 	Smallest() []byte
 	Biggest() []byte
-	DoesNotHave(hash uint64) bool
+	DoesNotHave(hash uint32) bool
 }
 
 // Table represents a loaded table file with the info we have about it.
@@ -104,11 +99,9 @@ type Table struct {
 
 	fd        *os.File // Own fd.
 	tableSize int      // Initialized in OpenTable, using fd.Stat().
-	bfLock    sync.Mutex
 
-	blockOffset []*pb.BlockOffset
-	ref         int32    // For file garbage collection. Atomic.
-	bf          *z.Bloom // Nil if index cache in enabled.
+	index *fb.TableIndex // Nil if encryption is enabled.
+	ref   int32          // For file garbage collection. Atomic.
 
 	mmap []byte // Memory mapped.
 
@@ -118,15 +111,18 @@ type Table struct {
 
 	Checksum []byte
 	// Stores the total size of key-values stored in this table (including the size on vlog).
-	estimatedSize  uint64
+	estimatedSize  uint32
 	indexStart     int
 	indexLen       int
 	hasBloomFilter bool
 
 	IsInmemory bool // Set to true if the table is on level 0 and opened in memory.
 	opt        *Options
+}
 
-	noOfBlocks int // Total number of blocks.
+// MaxVersion returns the maximum version across all keys stored in this table.
+func (t *Table) MaxVersion() uint64 {
+	return t.fetchIndex().MaxVersion()
 }
 
 // CompressionType returns the compression algorithm used for block compression.
@@ -146,6 +142,11 @@ func (t *Table) DecrRef() error {
 		// We can safely delete this file, because for all the current files, we always have
 		// at least one reference pointing to them.
 
+		// Delete all blocks from the cache.
+		for i := 0; i < t.offsetsLength(); i++ {
+			t.opt.BlockCache.Del(t.blockCacheKey(i))
+		}
+
 		// It's necessary to delete windows files.
 		if t.opt.LoadingMode == options.MemoryMap {
 			if err := y.Munmap(t.mmap); err != nil {
@@ -169,13 +170,6 @@ func (t *Table) DecrRef() error {
 		if err := os.Remove(filename); err != nil {
 			return err
 		}
-		// Delete all blocks from the cache.
-		for i := 0; i < t.noOfBlocks; i++ {
-			t.opt.BlockCache.Del(t.blockCacheKey(i))
-		}
-		// Delete bloom filter and indices from the cache.
-		t.opt.IndexCache.Del(t.blockOffsetsCacheKey())
-		t.opt.IndexCache.Del(t.bfCacheKey())
 	}
 	return nil
 }
@@ -355,12 +349,12 @@ func OpenInMemoryTable(data []byte, id uint64, opt *Options) (*Table, error) {
 
 func (t *Table) initBiggestAndSmallest() error {
 	var err error
-	var ko *pb.BlockOffset
+	var ko *fb.BlockOffset
 	if ko, err = t.initIndex(); err != nil {
 		return errors.Wrapf(err, "failed to read index.")
 	}
 
-	t.smallest = ko.Key
+	t.smallest = ko.KeyBytes()
 
 	it2 := t.NewIterator(REVERSED | NOCACHE)
 	defer it2.Close()
@@ -372,7 +366,7 @@ func (t *Table) initBiggestAndSmallest() error {
 	return nil
 }
 
-// Close closes the open table.  (Releases resources back to the OS.)
+// Close closes the open table. (Releases resources back to the OS.)
 func (t *Table) Close() error {
 	if t.opt.LoadingMode == options.MemoryMap {
 		if err := y.Munmap(t.mmap); err != nil {
@@ -409,7 +403,7 @@ func (t *Table) readNoFail(off, sz int) []byte {
 
 // initIndex reads the index and populate the necessary table fields and returns
 // first block offset
-func (t *Table) initIndex() (*pb.BlockOffset, error) {
+func (t *Table) initIndex() (*fb.BlockOffset, error) {
 	readPos := t.tableSize
 
 	// Read checksum len from the last 4 bytes.
@@ -448,36 +442,22 @@ func (t *Table) initIndex() (*pb.BlockOffset, error) {
 	}
 
 	if t.opt.Compression == options.None {
-		t.estimatedSize = index.EstimatedSize
+		t.estimatedSize = index.EstimatedSize()
 	} else {
 		// Due to compression the real size on disk is much
 		// smaller than what we estimate from index.EstimatedSize.
-		t.estimatedSize = uint64(t.tableSize)
+		t.estimatedSize = uint32(t.tableSize)
 	}
-	t.hasBloomFilter = len(index.BloomFilter) > 0
-	t.noOfBlocks = len(index.Offsets)
+	t.hasBloomFilter = len(index.BloomFilterBytes()) > 0
 
 	// No cache
 	if t.opt.IndexCache == nil {
-		// Keep blooms in memory.
-		if t.hasBloomFilter && t.opt.LoadBloomsOnOpen {
-			bf, err := z.JSONUnmarshal(index.BloomFilter)
-			if err != nil {
-				return nil,
-					errors.Wrapf(err, "failed to unmarshal bloomfilter for table:%d", t.id)
-			}
-
-			t.bfLock.Lock()
-			t.bf = bf
-			t.bfLock.Unlock()
-		}
-		// Keep block offsets in memory since there is no cache.
-		t.blockOffset = index.Offsets
+		// If there's no encryption, this points to the mmap'ed buffer.
+		t.index = index
 	}
-
-	// We don't need to put anything in the indexCache here. Table.Open will
-	// create an iterator and that iterator will push the indices in cache.
-	return index.Offsets[0], nil
+	var bo fb.BlockOffset
+	y.AssertTrue(index.Offsets(&bo, 0))
+	return &bo, nil
 }
 
 // KeySplits splits the table into at least n ranges based on the block offsets.
@@ -486,56 +466,47 @@ func (t *Table) KeySplits(n int, prefix []byte) []string {
 		return nil
 	}
 
-	var res []string
-	offsets := t.blockOffsets()
-	jump := len(offsets) / n
+	oLen := t.offsetsLength()
+	jump := oLen / n
 	if jump == 0 {
 		jump = 1
 	}
 
-	for i := 0; i < len(offsets); i += jump {
-		if i >= len(offsets) {
-			i = len(offsets) - 1
+	var bo fb.BlockOffset
+	var res []string
+	for i := 0; i < oLen; i += jump {
+		if i >= oLen {
+			i = oLen - 1
 		}
-		if bytes.HasPrefix(offsets[i].Key, prefix) {
-			res = append(res, string(offsets[i].Key))
+		y.AssertTrue(t.offsets(&bo, i))
+		if bytes.HasPrefix(bo.KeyBytes(), prefix) {
+			res = append(res, string(bo.KeyBytes()))
 		}
 	}
 	return res
 }
 
-// blockOffsets returns block offsets of this table.
-func (t *Table) blockOffsets() []*pb.BlockOffset {
+func (t *Table) fetchIndex() *fb.TableIndex {
 	if t.opt.IndexCache == nil {
-		return t.blockOffset
+		return t.index
 	}
 
-	if val, ok := t.opt.IndexCache.Get(t.blockOffsetsCacheKey()); ok && val != nil {
-		return val.([]*pb.BlockOffset)
+	if val, ok := t.opt.IndexCache.Get(t.indexKey()); ok && val != nil {
+		return val.(*fb.TableIndex)
 	}
 
 	index, err := t.readTableIndex()
 	y.Check(err)
-	t.opt.IndexCache.Set(
-		t.blockOffsetsCacheKey(),
-		index.Offsets,
-		calculateOffsetsSize(index.Offsets))
-
-	return index.Offsets
+	t.opt.IndexCache.Set(t.indexKey(), index, int64(t.indexLen))
+	return index
 }
 
-// calculateOffsetsSize returns the size of *pb.BlockOffset array
-func calculateOffsetsSize(offsets []*pb.BlockOffset) int64 {
-	totalSize := sizeOfOffsetStruct * int64(len(offsets))
+func (t *Table) offsetsLength() int {
+	return t.fetchIndex().OffsetsLength()
+}
 
-	for _, ko := range offsets {
-		// add key size.
-		totalSize += int64(cap(ko.Key))
-		// add XXX_unrecognized size.
-		totalSize += int64(cap(ko.XXX_unrecognized))
-	}
-	// Add three words for array size.
-	return totalSize + 3*8
+func (t *Table) offsets(ko *fb.BlockOffset, i int) bool {
+	return t.fetchIndex().Offsets(ko, i)
 }
 
 // block function return a new block. Each block holds a ref and the byte
@@ -543,7 +514,7 @@ func calculateOffsetsSize(offsets []*pb.BlockOffset) int64 {
 // caller should release the block by calling block.decrRef() on it.
 func (t *Table) block(idx int, useCache bool) (*block, error) {
 	y.AssertTruef(idx >= 0, "idx=%d", idx)
-	if idx >= t.noOfBlocks {
+	if idx >= t.offsetsLength() {
 		return nil, errors.New("block out of index")
 	}
 	if t.opt.BlockCache != nil {
@@ -559,19 +530,20 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 		}
 	}
 
-	// Read the block index if it's nil
-	ko := t.blockOffsets()[idx]
+	var ko fb.BlockOffset
+	y.AssertTrue(t.offsets(&ko, idx))
 	blk := &block{
-		offset: int(ko.Offset),
+		offset: int(ko.Offset()),
 		ref:    1,
 	}
 	defer blk.decrRef() // Deal with any errors, where blk would not be returned.
 	atomic.AddInt32(&NumBlocks, 1)
 
 	var err error
-	if blk.data, err = t.read(blk.offset, int(ko.Len)); err != nil {
+	if blk.data, err = t.read(blk.offset, int(ko.Len())); err != nil {
 		return nil, errors.Wrapf(err,
-			"failed to read from file: %s at offset: %d, len: %d", t.fd.Name(), blk.offset, ko.Len)
+			"failed to read from file: %s at offset: %d, len: %d",
+			t.fd.Name(), blk.offset, ko.Len())
 	}
 
 	if t.shouldDecrypt() {
@@ -584,7 +556,7 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 	if err = t.decompress(blk); err != nil {
 		return nil, errors.Wrapf(err,
 			"failed to decode compressed data in file: %s at offset: %d, len: %d",
-			t.fd.Name(), blk.offset, ko.Len)
+			t.fd.Name(), blk.offset, ko.Len())
 	}
 
 	// Read meta data related to block.
@@ -639,18 +611,6 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 	return blk, nil
 }
 
-// bfCacheKey returns the cache key for bloom filter. Bloom filters are stored in index cache.
-func (t *Table) bfCacheKey() []byte {
-	y.AssertTrue(t.id < math.MaxUint32)
-	buf := make([]byte, 6)
-	// Without the "bf" prefix, we will have conflict with the blockCacheKey.
-	buf[0] = 'b'
-	buf[1] = 'f'
-
-	binary.BigEndian.PutUint32(buf[2:], uint32(t.id))
-	return buf
-}
-
 // blockCacheKey is used to store blocks in the block cache.
 func (t *Table) blockCacheKey(idx int) []byte {
 	y.AssertTrue(t.id < math.MaxUint32)
@@ -663,31 +623,35 @@ func (t *Table) blockCacheKey(idx int) []byte {
 	return buf
 }
 
-// blockOffsetsCacheKey returns the cache key for block offsets. blockOffsets
+// indexKey returns the cache key for block offsets. blockOffsets
 // are stored in the index cache.
-func (t *Table) blockOffsetsCacheKey() uint64 {
+func (t *Table) indexKey() uint64 {
 	return t.id
 }
 
-// IndexSize returns the size of table index in bytes stored in the memory. The
-// size of on-disk representation would be less than the in-memory representation.
+// UncompressedSize is the size of table index in bytes.
+func (t *Table) UncompressedSize() uint32 {
+	return t.index.UncompressedSize()
+}
+
+// KeyCount is the total number of keys in this table.
+func (t *Table) KeyCount() uint32 {
+	return t.index.KeyCount()
+}
+
+// IndexSize is the size of table index in bytes.
 func (t *Table) IndexSize() int {
-	indexSz := 0
-	for _, bi := range t.blockOffsets() {
-		indexSz += bi.Size()
-	}
-	return indexSz
+	return t.indexLen
 }
 
-// BloomFilterSize returns the size of the bloom filter in bytes stored in memory. The
-// size of on-disk representation would be less than the in-memory representation.
+// BloomFilterSize returns the size of the bloom filter in bytes stored in memory.
 func (t *Table) BloomFilterSize() int {
-	return t.bf.TotalSize()
+	return t.index.BloomFilterLength()
 }
 
 // EstimatedSize returns the total size of key-values stored in this table (including the
 // disk space occupied on the value log).
-func (t *Table) EstimatedSize() uint64 { return t.estimatedSize }
+func (t *Table) EstimatedSize() uint32 { return t.estimatedSize }
 
 // Size is its file size in bytes
 func (t *Table) Size() int64 { return int64(t.tableSize) }
@@ -706,49 +670,29 @@ func (t *Table) ID() uint64 { return t.id }
 
 // DoesNotHave returns true if and only if the table does not have the key hash.
 // It does a bloom filter lookup.
-func (t *Table) DoesNotHave(hash uint64) bool {
+func (t *Table) DoesNotHave(hash uint32) bool {
 	if !t.hasBloomFilter {
 		return false
 	}
 
-	// Return fast if the cache is absent.
-	if t.opt.IndexCache == nil {
-		t.bfLock.Lock()
-		if t.bf == nil {
-			y.AssertTrue(!t.opt.LoadBloomsOnOpen)
-			// Load bloomfilter into memory since the cache is absent.
-			t.bf, _ = t.readBloomFilter()
-		}
-		t.bfLock.Unlock()
-		return !t.bf.Has(hash)
-	}
-
-	// Check if the bloom filter exists in the cache.
-	if bf, ok := t.opt.IndexCache.Get(t.bfCacheKey()); bf != nil && ok {
-		return !bf.(*z.Bloom).Has(hash)
-	}
-
-	bf, sz := t.readBloomFilter()
-	t.opt.IndexCache.Set(t.bfCacheKey(), bf, int64(sz))
-	return !bf.Has(hash)
+	index := t.fetchIndex()
+	bf := index.BloomFilterBytes()
+	return !y.Filter(bf).MayContain(hash)
 }
 
 // readBloomFilter reads the bloom filter from the SST and returns its length
 // along with the bloom filter.
 func (t *Table) readBloomFilter() (*z.Bloom, int) {
-	// Read bloom filter from the SST.
-	index, err := t.readTableIndex()
-	y.Check(err)
-
-	bf, err := z.JSONUnmarshal(index.BloomFilter)
+	index := t.fetchIndex()
+	// Read bloom filter from the index.
+	bf, err := z.JSONUnmarshal(index.BloomFilterBytes())
 	y.Check(err)
-	return bf, len(index.BloomFilter)
+	return bf, index.BloomFilterLength()
 }
 
 // readTableIndex reads table index from the sst and returns its pb format.
-func (t *Table) readTableIndex() (*pb.TableIndex, error) {
+func (t *Table) readTableIndex() (*fb.TableIndex, error) {
 	data := t.readNoFail(t.indexStart, t.indexLen)
-	index := pb.TableIndex{}
 	var err error
 	// Decrypt the table index if it is encrypted.
 	if t.shouldDecrypt() {
@@ -757,18 +701,18 @@ func (t *Table) readTableIndex() (*pb.TableIndex, error) {
 				"Error while decrypting table index for the table %d in readTableIndex", t.id)
 		}
 	}
-	y.Check(proto.Unmarshal(data, &index))
-	return &index, nil
+	return fb.GetRootAsTableIndex(data, 0), nil
 }
 
 // VerifyChecksum verifies checksum for all blocks of table. This function is called by
 // OpenTable() function. This function is also called inside levelsController.VerifyChecksum().
 func (t *Table) VerifyChecksum() error {
-	for i, os := range t.blockOffsets() {
+	ti := t.fetchIndex()
+	for i := 0; i < ti.OffsetsLength(); i++ {
 		b, err := t.block(i, true)
 		if err != nil {
 			return y.Wrapf(err, "checksum validation failed for table: %s, block: %d, offset:%d",
-				t.Filename(), i, os.Offset)
+				t.Filename(), i, b.offset)
 		}
 		// We should not call incrRef here, because the block already has one ref when created.
 		defer b.decrRef()
@@ -778,7 +722,7 @@ func (t *Table) VerifyChecksum() error {
 			if err = b.verifyCheckSum(); err != nil {
 				return y.Wrapf(err,
 					"checksum validation failed for table: %s, block: %d, offset:%d",
-					t.Filename(), i, os.Offset)
+					t.Filename(), i, b.offset)
 			}
 		}
 	}
diff --git a/y/bloom.go b/y/bloom.go
new file mode 100644
index 0000000..390a89f
--- /dev/null
+++ b/y/bloom.go
@@ -0,0 +1,170 @@
+// Copyright 2013 The LevelDB-Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package y
+
+import "math"
+
+// Filter is an encoded set of []byte keys.
+type Filter []byte
+
+func (f Filter) MayContainKey(k []byte) bool {
+	return f.MayContain(Hash(k))
+}
+
+// MayContain returns whether the filter may contain given key. False positives
+// are possible, where it returns true for keys not in the original set.
+func (f Filter) MayContain(h uint32) bool {
+	if len(f) < 2 {
+		return false
+	}
+	k := f[len(f)-1]
+	if k > 30 {
+		// This is reserved for potentially new encodings for short Bloom filters.
+		// Consider it a match.
+		return true
+	}
+	nBits := uint32(8 * (len(f) - 1))
+	delta := h>>17 | h<<15
+	for j := uint8(0); j < k; j++ {
+		bitPos := h % nBits
+		if f[bitPos/8]&(1<<(bitPos%8)) == 0 {
+			return false
+		}
+		h += delta
+	}
+	return true
+}
+
+// NewFilter returns a new Bloom filter that encodes a set of []byte keys with
+// the given number of bits per key, approximately.
+//
+// A good bitsPerKey value is 10, which yields a filter with ~ 1% false
+// positive rate.
+func NewFilter(keys []uint32, bitsPerKey int) Filter {
+	return Filter(appendFilter(nil, keys, bitsPerKey))
+}
+
+// BloomBitsPerKey returns the bits per key required by bloomfilter based on
+// the false positive rate.
+func BloomBitsPerKey(numEntries int, fp float64) int {
+	size := -1 * float64(numEntries) * math.Log(fp) / math.Pow(float64(0.69314718056), 2)
+	locs := math.Ceil(float64(0.69314718056) * size / float64(numEntries))
+	return int(locs)
+}
+
+func appendFilter(buf []byte, keys []uint32, bitsPerKey int) []byte {
+	if bitsPerKey < 0 {
+		bitsPerKey = 0
+	}
+	// 0.69 is approximately ln(2).
+	k := uint32(float64(bitsPerKey) * 0.69)
+	if k < 1 {
+		k = 1
+	}
+	if k > 30 {
+		k = 30
+	}
+
+	nBits := len(keys) * int(bitsPerKey)
+	// For small len(keys), we can see a very high false positive rate. Fix it
+	// by enforcing a minimum bloom filter length.
+	if nBits < 64 {
+		nBits = 64
+	}
+	nBytes := (nBits + 7) / 8
+	nBits = nBytes * 8
+	buf, filter := extend(buf, nBytes+1)
+
+	for _, h := range keys {
+		delta := h>>17 | h<<15
+		for j := uint32(0); j < k; j++ {
+			bitPos := h % uint32(nBits)
+			filter[bitPos/8] |= 1 << (bitPos % 8)
+			h += delta
+		}
+	}
+	filter[nBytes] = uint8(k)
+
+	return buf
+}
+
+// extend appends n zero bytes to b. It returns the overall slice (of length
+// n+len(originalB)) and the slice of n trailing zeroes.
+func extend(b []byte, n int) (overall, trailer []byte) {
+	want := n + len(b)
+	if want <= cap(b) {
+		overall = b[:want]
+		trailer = overall[len(b):]
+		for i := range trailer {
+			trailer[i] = 0
+		}
+	} else {
+		// Grow the capacity exponentially, with a 1KiB minimum.
+		c := 1024
+		for c < want {
+			c += c / 4
+		}
+		overall = make([]byte, want, c)
+		trailer = overall[len(b):]
+		copy(overall, b)
+	}
+	return overall, trailer
+}
+
+// hash implements a hashing algorithm similar to the Murmur hash.
+func Hash(b []byte) uint32 {
+	const (
+		seed = 0xbc9f1d34
+		m    = 0xc6a4a793
+	)
+	h := uint32(seed) ^ uint32(len(b)*m)
+	for ; len(b) >= 4; b = b[4:] {
+		h += uint32(b[0]) | uint32(b[1])<<8 | uint32(b[2])<<16 | uint32(b[3])<<24
+		h *= m
+		h ^= h >> 16
+	}
+	switch len(b) {
+	case 3:
+		h += uint32(b[2]) << 16
+		fallthrough
+	case 2:
+		h += uint32(b[1]) << 8
+		fallthrough
+	case 1:
+		h += uint32(b[0])
+		h *= m
+		h ^= h >> 24
+	}
+	return h
+}
+
+// FilterPolicy implements the db.FilterPolicy interface from the leveldb/db
+// package.
+//
+// The integer value is the approximate number of bits used per key. A good
+// value is 10, which yields a filter with ~ 1% false positive rate.
+//
+// It is valid to use the other API in this package (leveldb/bloom) without
+// using this type or the leveldb/db package.
+
+// type FilterPolicy int
+
+// // Name implements the db.FilterPolicy interface.
+// func (p FilterPolicy) Name() string {
+// 	// This string looks arbitrary, but its value is written to LevelDB .ldb
+// 	// files, and should be this exact value to be compatible with those files
+// 	// and with the C++ LevelDB code.
+// 	return "leveldb.BuiltinBloomFilter2"
+// }
+
+// // AppendFilter implements the db.FilterPolicy interface.
+// func (p FilterPolicy) AppendFilter(dst []byte, keys [][]byte) []byte {
+// 	return appendFilter(dst, keys, int(p))
+// }
+
+// // MayContain implements the db.FilterPolicy interface.
+// func (p FilterPolicy) MayContain(filter, key []byte) bool {
+// 	return Filter(filter).MayContain(key)
+// }


Test patch:
diff --git a/db2_test.go b/db2_test.go
index f0e6e59..c87669f 100644
--- a/db2_test.go
+++ b/db2_test.go
@@ -495,26 +495,29 @@ func TestCompactionFilePicking(t *testing.T) {
 	}
 
 	tables := db.lc.levels[2].tables
-	db.lc.sortByOverlap(tables, cdef)
+	db.lc.sortByHeuristic(tables, cdef)
 
 	var expKey [8]byte
-	// First table should be with smallest and biggest keys as 1 and 4.
+	// First table should be with smallest and biggest keys as 1 and 4 which
+	// has the lowest version.
 	binary.BigEndian.PutUint64(expKey[:], uint64(1))
 	require.Equal(t, expKey[:], y.ParseKey(tables[0].Smallest()))
 	binary.BigEndian.PutUint64(expKey[:], uint64(4))
 	require.Equal(t, expKey[:], y.ParseKey(tables[0].Biggest()))
 
-	// Second table should be with smallest and biggest keys as 13 and 18.
+	// Second table should be with smallest and biggest keys as 13 and 18
+	// which has the second lowest version.
 	binary.BigEndian.PutUint64(expKey[:], uint64(13))
-	require.Equal(t, expKey[:], y.ParseKey(tables[1].Smallest()))
+	require.Equal(t, expKey[:], y.ParseKey(tables[2].Smallest()))
 	binary.BigEndian.PutUint64(expKey[:], uint64(18))
-	require.Equal(t, expKey[:], y.ParseKey(tables[1].Biggest()))
+	require.Equal(t, expKey[:], y.ParseKey(tables[2].Biggest()))
 
-	// Third table should be with smallest and biggest keys as 5 and 12.
+	// Third table should be with smallest and biggest keys as 5 and 12 which
+	// has the maximum version.
 	binary.BigEndian.PutUint64(expKey[:], uint64(5))
-	require.Equal(t, expKey[:], y.ParseKey(tables[2].Smallest()))
+	require.Equal(t, expKey[:], y.ParseKey(tables[1].Smallest()))
 	binary.BigEndian.PutUint64(expKey[:], uint64(12))
-	require.Equal(t, expKey[:], y.ParseKey(tables[2].Biggest()))
+	require.Equal(t, expKey[:], y.ParseKey(tables[1].Biggest()))
 }
 
 // addToManifest function is used in TestCompactionFilePicking. It adds table to db manifest.
diff --git a/db_test.go b/db_test.go
index c3aa8dc..fd3bbe5 100644
--- a/db_test.go
+++ b/db_test.go
@@ -380,7 +380,7 @@ func TestStreamDB(t *testing.T) {
 	dir, err := ioutil.TempDir("", "badger-test")
 	require.NoError(t, err)
 	defer removeDir(dir)
-	opts := getTestOptions(dir).WithCompression(options.ZSTD)
+	opts := getTestOptions(dir).WithCompression(options.ZSTD).WithBlockCacheSize(100 << 20)
 
 	db, err := OpenManaged(opts)
 	require.NoError(t, err)
@@ -838,6 +838,7 @@ func TestLoad(t *testing.T) {
 		require.NoError(t, err)
 		opt := getTestOptions("")
 		opt.EncryptionKey = key
+		opt.BlockCacheSize = 100 << 20
 		opt.Compression = options.None
 		testLoad(t, opt)
 	})
@@ -848,11 +849,13 @@ func TestLoad(t *testing.T) {
 		opt := getTestOptions("")
 		opt.EncryptionKey = key
 		opt.Compression = options.ZSTD
+		opt.BlockCacheSize = 100 << 20
 		testLoad(t, opt)
 	})
 	t.Run("TestLoad without Encryption and with compression", func(t *testing.T) {
 		opt := getTestOptions("")
 		opt.Compression = options.ZSTD
+		opt.BlockCacheSize = 100 << 20
 		testLoad(t, opt)
 	})
 }
diff --git a/iterator_test.go b/iterator_test.go
index cdd3276..47e6554 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -38,7 +38,7 @@ type tableMock struct {
 
 func (tm *tableMock) Smallest() []byte             { return tm.left }
 func (tm *tableMock) Biggest() []byte              { return tm.right }
-func (tm *tableMock) DoesNotHave(hash uint64) bool { return false }
+func (tm *tableMock) DoesNotHave(hash uint32) bool { return false }
 
 func TestPickTables(t *testing.T) {
 	opt := DefaultIteratorOptions
diff --git a/stream_writer_test.go b/stream_writer_test.go
index 1a1c2be..8542490 100644
--- a/stream_writer_test.go
+++ b/stream_writer_test.go
@@ -344,7 +344,7 @@ func TestStreamWriter6(t *testing.T) {
 		require.NoError(t, sw.Write(list), "sw.Write() failed")
 		require.NoError(t, sw.Flush(), "sw.Flush() failed")
 
-		tables := db.Tables(true)
+		tables := db.Tables()
 		require.Equal(t, 4, len(tables), "Count of tables not matching")
 		for _, tab := range tables {
 			if tab.Level > 0 {
diff --git a/table/builder_test.go b/table/builder_test.go
index 2bb1866..73d54b5 100644
--- a/table/builder_test.go
+++ b/table/builder_test.go
@@ -23,9 +23,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/dgryski/go-farm"
 	"github.com/stretchr/testify/require"
 
+	"github.com/dgraph-io/badger/v2/fb"
 	"github.com/dgraph-io/badger/v2/options"
 	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/y"
@@ -95,7 +95,7 @@ func TestTableIndex(t *testing.T) {
 			blockFirstKeys := make([][]byte, 0)
 			blockCount := 0
 			for i := 0; i < keysCount; i++ {
-				k := []byte(fmt.Sprintf("%016x", i))
+				k := y.KeyWithTs([]byte(fmt.Sprintf("%016x", i)), uint64(i+1))
 				v := fmt.Sprintf("%d", i)
 				vs := y.ValueStruct{Value: []byte(v)}
 				if i == 0 { // This is first key for first block.
@@ -119,12 +119,15 @@ func TestTableIndex(t *testing.T) {
 			}
 
 			// Ensure index is built correctly
-			require.Equal(t, blockCount, tbl.noOfBlocks)
+			require.Equal(t, blockCount, tbl.offsetsLength())
 			idx, err := tbl.readTableIndex()
 			require.NoError(t, err)
-			for i, ko := range idx.Offsets {
-				require.Equal(t, ko.Key, blockFirstKeys[i])
+			for i := 0; i < idx.OffsetsLength(); i++ {
+				var bo fb.BlockOffset
+				require.True(t, idx.Offsets(&bo, i))
+				require.Equal(t, blockFirstKeys[i], bo.KeyBytes())
 			}
+			require.Equal(t, keysCount, int(tbl.MaxVersion()))
 			f.Close()
 			require.NoError(t, os.RemoveAll(filename))
 		})
@@ -232,7 +235,7 @@ func TestBloomfilter(t *testing.T) {
 		c := 0
 		for it.Rewind(); it.Valid(); it.Next() {
 			c++
-			hash := farm.Fingerprint64(y.ParseKey(it.Key()))
+			hash := y.Hash(y.ParseKey(it.Key()))
 			require.False(t, tab.DoesNotHave(hash))
 		}
 		require.Equal(t, keyCount, c)
@@ -242,13 +245,13 @@ func TestBloomfilter(t *testing.T) {
 		c = 0
 		for it.Rewind(); it.Valid(); it.Next() {
 			c++
-			hash := farm.Fingerprint64(y.ParseKey(it.Key()))
+			hash := y.Hash(y.ParseKey(it.Key()))
 			require.False(t, tab.DoesNotHave(hash))
 		}
 		require.Equal(t, keyCount, c)
 
 		// Ensure tab.DoesNotHave works
-		hash := farm.Fingerprint64([]byte("foo"))
+		hash := y.Hash([]byte("foo"))
 		require.Equal(t, withBlooms, tab.DoesNotHave(hash))
 	}
 
diff --git a/table/table_test.go b/table/table_test.go
index 88b7bd9..6dc2f7a 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -31,7 +31,6 @@ import (
 
 	"github.com/cespare/xxhash"
 	"github.com/dgraph-io/badger/v2/options"
-	"github.com/dgraph-io/badger/v2/pb"
 	"github.com/dgraph-io/badger/v2/y"
 	"github.com/dgraph-io/ristretto"
 	"github.com/stretchr/testify/require"
@@ -80,7 +79,8 @@ func buildTable(t *testing.T, keyValues [][]string, opts Options) *os.File {
 	})
 	for _, kv := range keyValues {
 		y.AssertTrue(len(kv) == 2)
-		b.Add(y.KeyWithTs([]byte(kv[0]), 0), y.ValueStruct{Value: []byte(kv[1]), Meta: 'A', UserMeta: 0}, 0)
+		b.Add(y.KeyWithTs([]byte(kv[0]), 0),
+			y.ValueStruct{Value: []byte(kv[1]), Meta: 'A', UserMeta: 0}, 0)
 	}
 	_, err = f.Write(b.Finish(false))
 	require.NoError(t, err, "writing to file failed")
@@ -699,7 +699,7 @@ func TestTableBigValues(t *testing.T) {
 		TableSize: uint64(n) * 1 << 20}
 	builder := NewTableBuilder(opts)
 	for i := 0; i < n; i++ {
-		key := y.KeyWithTs([]byte(key("", i)), 0)
+		key := y.KeyWithTs([]byte(key("", i)), uint64(i+1))
 		vs := y.ValueStruct{Value: value(i)}
 		builder.Add(key, vs, 0)
 	}
@@ -721,6 +721,7 @@ func TestTableBigValues(t *testing.T) {
 	}
 	require.False(t, itr.Valid(), "table iterator should be invalid now")
 	require.Equal(t, n, count)
+	require.Equal(t, n, int(tbl.MaxVersion()))
 }
 
 // This test is for verifying checksum failure during table open.
@@ -739,10 +740,13 @@ func TestTableChecksum(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, n, len(rb))
 
-	_, err = OpenTable(f, opts)
-	if err == nil || !strings.Contains(err.Error(), "checksum") {
-		t.Fatal("Test should have been failed with checksum mismatch error")
-	}
+	require.Panics(t, func() {
+		// Either OpenTable will panic on corrupted data or the checksum verification will fail.
+		_, err = OpenTable(f, opts)
+		if strings.Contains(err.Error(), "checksum") {
+			panic("checksum mismatch")
+		}
+	})
 }
 
 var cacheConfig = ristretto.Config{
@@ -932,7 +936,7 @@ func TestOpenKVSize(t *testing.T) {
 		require.NoError(t, err)
 
 		// The estimated size is same as table size in case compression is enabled.
-		require.Equal(t, uint64(table.tableSize), table.EstimatedSize())
+		require.Equal(t, uint32(table.tableSize), table.EstimatedSize())
 	})
 
 	t.Run("no compressin", func(t *testing.T) {
@@ -945,7 +949,7 @@ func TestOpenKVSize(t *testing.T) {
 
 		stat, err := table.fd.Stat()
 		require.NoError(t, err)
-		require.Less(t, table.EstimatedSize(), uint64(stat.Size()))
+		require.Less(t, table.EstimatedSize(), uint32(stat.Size()))
 	})
 }
 
@@ -961,28 +965,30 @@ func TestDoesNotHaveRace(t *testing.T) {
 	wg.Add(5)
 	for i := 0; i < 5; i++ {
 		go func() {
-			require.True(t, table.DoesNotHave(uint64(1237882)))
+			require.True(t, table.DoesNotHave(uint32(1237882)))
 			wg.Done()
 		}()
 	}
 	wg.Wait()
 }
 
-var ko *pb.BlockOffset
+func TestMaxVersion(t *testing.T) {
+	opt := getTestTableOptions()
+	b := NewTableBuilder(opt)
+	defer b.Close()
 
-// Use this benchmark to manually verify block offset size calculation
-func BenchmarkBlockOffsetSizeCalculation(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		ko = &pb.BlockOffset{
-			Key: []byte{1, 23},
-		}
-	}
-}
+	filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Uint32())
+	f, err := y.CreateSyncedFile(filename, true)
+	require.NoError(t, err)
 
-func TestBlockOffsetSizeCalculation(t *testing.T) {
-	// Empty struct testing.
-	require.Equal(t, calculateOffsetsSize([]*pb.BlockOffset{&pb.BlockOffset{}}), int64(88))
-	// Testing with key bytes
-	require.Equal(t, calculateOffsetsSize([]*pb.BlockOffset{&pb.BlockOffset{Key: []byte{1, 1}}}),
-		int64(90))
+	N := 1000
+	for i := 0; i < N; i++ {
+		b.Add(y.KeyWithTs([]byte(fmt.Sprintf("foo:%d", i)), uint64(i+1)), y.ValueStruct{}, 0)
+	}
+	_, err = f.Write(b.Finish(false))
+	require.NoError(t, err, "writing to file failed")
+	f.Close()
+	f, _ = y.OpenSyncedFile(filename, false)
+	table, err := OpenTable(f, opt)
+	require.Equal(t, N, int(table.MaxVersion()))
 }
diff --git a/y/bloom_test.go b/y/bloom_test.go
new file mode 100644
index 0000000..dcb3dac
--- /dev/null
+++ b/y/bloom_test.go
@@ -0,0 +1,147 @@
+// Copyright 2013 The LevelDB-Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package y
+
+import (
+	"testing"
+)
+
+func (f Filter) String() string {
+	s := make([]byte, 8*len(f))
+	for i, x := range f {
+		for j := 0; j < 8; j++ {
+			if x&(1<<uint(j)) != 0 {
+				s[8*i+j] = '1'
+			} else {
+				s[8*i+j] = '.'
+			}
+		}
+	}
+	return string(s)
+}
+
+func TestSmallBloomFilter(t *testing.T) {
+	var hash []uint32
+	for _, word := range [][]byte{
+		[]byte("hello"),
+		[]byte("world"),
+	} {
+		hash = append(hash, Hash(word))
+	}
+
+	f := NewFilter(hash, 10)
+	got := f.String()
+	// The magic want string comes from running the C++ leveldb code's bloom_test.cc.
+	want := "1...1.........1.........1.....1...1...1.....1.........1.....1....11....."
+	if got != want {
+		t.Fatalf("bits:\ngot  %q\nwant %q", got, want)
+	}
+
+	m := map[string]bool{
+		"hello": true,
+		"world": true,
+		"x":     false,
+		"foo":   false,
+	}
+	for k, want := range m {
+		got := f.MayContainKey([]byte(k))
+		if got != want {
+			t.Errorf("MayContain: k=%q: got %v, want %v", k, got, want)
+		}
+	}
+}
+
+func TestBloomFilter(t *testing.T) {
+	nextLength := func(x int) int {
+		if x < 10 {
+			return x + 1
+		}
+		if x < 100 {
+			return x + 10
+		}
+		if x < 1000 {
+			return x + 100
+		}
+		return x + 1000
+	}
+	le32 := func(i int) []byte {
+		b := make([]byte, 4)
+		b[0] = uint8(uint32(i) >> 0)
+		b[1] = uint8(uint32(i) >> 8)
+		b[2] = uint8(uint32(i) >> 16)
+		b[3] = uint8(uint32(i) >> 24)
+		return b
+	}
+
+	nMediocreFilters, nGoodFilters := 0, 0
+loop:
+	for length := 1; length <= 10000; length = nextLength(length) {
+		keys := make([][]byte, 0, length)
+		for i := 0; i < length; i++ {
+			keys = append(keys, le32(i))
+		}
+		var hashes []uint32
+		for _, key := range keys {
+			hashes = append(hashes, Hash(key))
+		}
+		f := NewFilter(hashes, 10)
+
+		if len(f) > (length*10/8)+40 {
+			t.Errorf("length=%d: len(f)=%d is too large", length, len(f))
+			continue
+		}
+
+		// All added keys must match.
+		for _, key := range keys {
+			if !f.MayContainKey(key) {
+				t.Errorf("length=%d: did not contain key %q", length, key)
+				continue loop
+			}
+		}
+
+		// Check false positive rate.
+		nFalsePositive := 0
+		for i := 0; i < 10000; i++ {
+			if f.MayContainKey(le32(1e9 + i)) {
+				nFalsePositive++
+			}
+		}
+		if nFalsePositive > 0.02*10000 {
+			t.Errorf("length=%d: %d false positives in 10000", length, nFalsePositive)
+			continue
+		}
+		if nFalsePositive > 0.0125*10000 {
+			nMediocreFilters++
+		} else {
+			nGoodFilters++
+		}
+	}
+
+	if nMediocreFilters > nGoodFilters/5 {
+		t.Errorf("%d mediocre filters but only %d good filters", nMediocreFilters, nGoodFilters)
+	}
+}
+
+func TestHash(t *testing.T) {
+	// The magic want numbers come from running the C++ leveldb code in hash.cc.
+	testCases := []struct {
+		s    string
+		want uint32
+	}{
+		{"", 0xbc9f1d34},
+		{"g", 0xd04a8bda},
+		{"go", 0x3e0b0745},
+		{"gop", 0x0c326610},
+		{"goph", 0x8c9d6390},
+		{"gophe", 0x9bfd4b0a},
+		{"gopher", 0xa78edc7c},
+		{"I had a dream it would end this way.", 0xe14a9db9},
+	}
+	for _, tc := range testCases {
+		if got := Hash([]byte(tc.s)); got != tc.want {
+			t.Errorf("s=%q: got 0x%08x, want 0x%08x", tc.s, got, tc.want)
+		}
+	}
+}