Repository: dgraph-io/badger
Pull request #1473 (2020-08-21 12:48:18)

Rework Block and Index cache
This PR separates the block cache and the index cache. Earlier we had a mix of both the caches. 
Block Cache -> Stores block. Should be used when running badger with compression/encryption
Index Cache -> Used to limit the memory used by table indices and bloom filters. 

This PR also remove the `KeepBlocksInCache` and `KeepBlockIndicesInCache` options.
<!-- 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/1473)
<!-- Reviewable:end -->



Patch:
diff --git a/badger/cmd/bank.go b/badger/cmd/bank.go
index a312ca9..83529df 100644
--- a/badger/cmd/bank.go
+++ b/badger/cmd/bank.go
@@ -365,7 +365,7 @@ func runTest(cmd *cobra.Command, args []string) error {
 		WithValueThreshold(1). // Make all values go to value log
 		WithCompression(options.ZSTD).
 		WithKeepL0InMemory(false).
-		WithMaxCacheSize(10 << 20)
+		WithBlockCacheSize(10 << 20)
 
 	if mmap {
 		opts = opts.WithTableLoadingMode(options.MemoryMap)
diff --git a/badger/cmd/info.go b/badger/cmd/info.go
index 8c6581d..14d932f 100644
--- a/badger/cmd/info.go
+++ b/badger/cmd/info.go
@@ -244,13 +244,13 @@ func tableInfo(dir, valueDir string, db *badger.DB) {
 	tables := db.Tables(true)
 	fmt.Println()
 	fmt.Println("SSTable [Li, Id, Total Keys including internal keys] " +
-		"[Left Key, Version -> Right Key, Version]")
+		"[Left Key, Version -> Right Key, Version]" + "[Index Size]")
 	for _, t := range tables {
 		lk, lt := y.ParseKey(t.Left), y.ParseTs(t.Left)
 		rk, rt := y.ParseKey(t.Right), y.ParseTs(t.Right)
 
-		fmt.Printf("SSTable [L%d, %03d, %07d] [%20X, v%d -> %20X, v%d]\n",
-			t.Level, t.ID, t.KeyCount, lk, lt, rk, rt)
+		fmt.Printf("SSTable [L%d, %03d, %07d] [%20X, v%d -> %20X, v%d] [%s]\n",
+			t.Level, t.ID, t.KeyCount, lk, lt, rk, rt, hbytes(int64(t.IndexSz)))
 	}
 	fmt.Println()
 }
diff --git a/badger/cmd/read_bench.go b/badger/cmd/read_bench.go
index f09b5f5..6a622cd 100644
--- a/badger/cmd/read_bench.go
+++ b/badger/cmd/read_bench.go
@@ -46,10 +46,8 @@ var (
 	entriesRead uint64    // will store entries read till now
 	startTime   time.Time // start time of read benchmarking
 
-	sizeMaxCache      int64
-	keepBlockIdxCache bool
-	keepBlocksCache   bool
-	sizeMaxBfCache    int64
+	blockCacheSize int64
+	indexCacheSize int64
 
 	sampleSize  int
 	loadingMode string
@@ -75,13 +73,8 @@ func init() {
 			"Valid loading modes are fileio and mmap.")
 	readBenchCmd.Flags().BoolVar(
 		&fullScan, "full-scan", false, "If true, full db will be scanned using iterators.")
-	readBenchCmd.Flags().Int64VarP(&sizeMaxCache, "max-cache", "C", 0, "Max size of cache in MB")
-	readBenchCmd.Flags().BoolVarP(&keepBlockIdxCache, "keep-bidx", "b", false,
-		"Keep block indices in cache")
-	readBenchCmd.Flags().BoolVarP(&keepBlocksCache, "keep-blocks", "B", false,
-		"Keep blocks in cache")
-	readBenchCmd.Flags().Int64VarP(&sizeMaxBfCache, "max-bf-cache", "c", 0,
-		"Maximum Bloom Filter Cache Size in MB")
+	readBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache", 0, "Max size of block cache in MB")
+	readBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache", 0, "Max size of index cache in MB")
 }
 
 // Scan the whole database using the iterators
@@ -118,10 +111,8 @@ func readBench(cmd *cobra.Command, args []string) error {
 		WithReadOnly(readOnly).
 		WithTableLoadingMode(mode).
 		WithValueLogLoadingMode(mode).
-		WithMaxCacheSize(sizeMaxCache << 20).
-		WithKeepBlockIndicesInCache(keepBlockIdxCache).
-		WithKeepBlocksInCache(keepBlocksCache).
-		WithMaxBfCacheSize(sizeMaxBfCache << 20)
+		WithBlockCacheSize(blockCacheSize << 20).
+		WithIndexCacheSize(indexCacheSize << 20)
 	fmt.Printf("Opening badger with options = %+v\n", opt)
 	db, err := badger.Open(opt)
 	if err != nil {
diff --git a/badger/cmd/write_bench.go b/badger/cmd/write_bench.go
index ec003f9..e4cac53 100644
--- a/badger/cmd/write_bench.go
+++ b/badger/cmd/write_bench.go
@@ -59,19 +59,15 @@ var (
 	sizeWritten    uint64
 	entriesWritten uint64
 
-	valueThreshold      int
-	numVersions         int
-	maxCacheSize        int64
-	keepBlockIdxInCache bool
-	keepBlocksInCache   bool
-	maxBfCacheSize      int64
-	vlogMaxEntries      uint32
-	loadBloomsOnOpen    bool
-	detectConflicts     bool
-	compression         bool
-	showDir             bool
-	ttlDuration         string
-	showKeysCount       bool
+	valueThreshold   int
+	numVersions      int
+	vlogMaxEntries   uint32
+	loadBloomsOnOpen bool
+	detectConflicts  bool
+	compression      bool
+	showDir          bool
+	ttlDuration      string
+	showKeysCount    bool
 
 	sstCount  uint32
 	vlogCount uint32
@@ -102,13 +98,10 @@ func init() {
 	writeBenchCmd.Flags().BoolVarP(&showLogs, "logs", "l", 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().Int64VarP(&maxCacheSize, "max-cache", "C", 0, "Max size of cache in MB")
-	writeBenchCmd.Flags().BoolVarP(&keepBlockIdxInCache, "keep-bidx", "b", false,
-		"Keep block indices in cache")
-	writeBenchCmd.Flags().BoolVarP(&keepBlocksInCache, "keep-blocks", "B", false,
-		"Keep blocks in cache")
-	writeBenchCmd.Flags().Int64VarP(&maxBfCacheSize, "max-bf-cache", "c", 0,
-		"Maximum Bloom Filter Cache Size in MB")
+	writeBenchCmd.Flags().Int64Var(&blockCacheSize, "block-cache", 0,
+		"Size of block cache in MB")
+	writeBenchCmd.Flags().Int64Var(&indexCacheSize, "index-cache", 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", "",
 		"If it is true, badger will encrypt all the data stored on the disk.")
@@ -249,10 +242,8 @@ func writeBench(cmd *cobra.Command, args []string) error {
 		WithCompactL0OnClose(force).
 		WithValueThreshold(valueThreshold).
 		WithNumVersionsToKeep(numVersions).
-		WithMaxCacheSize(maxCacheSize << 20).
-		WithKeepBlockIndicesInCache(keepBlockIdxInCache).
-		WithKeepBlocksInCache(keepBlocksInCache).
-		WithMaxBfCacheSize(maxBfCacheSize << 20).
+		WithBlockCacheSize(blockCacheSize << 20).
+		WithIndexCacheSize(indexCacheSize << 20).
 		WithValueLogMaxEntries(vlogMaxEntries).
 		WithTableLoadingMode(mode).
 		WithEncryptionKey([]byte(encryptionKey)).
diff --git a/db.go b/db.go
index e485400..aaacec4 100644
--- a/db.go
+++ b/db.go
@@ -90,7 +90,7 @@ type DB struct {
 	pub        *publisher
 	registry   *KeyRegistry
 	blockCache *ristretto.Cache
-	bfCache    *ristretto.Cache
+	indexCache *ristretto.Cache
 }
 
 const (
@@ -210,12 +210,6 @@ func Open(opt Options) (db *DB, err error) {
 			maxValueThreshold)
 	}
 
-	// We need a cache if KeepBlocksInCache or KeepBlockIndices is set to true.
-	if (opt.KeepBlocksInCache || opt.KeepBlockIndicesInCache) && opt.MaxCacheSize <= 0 {
-		return nil, errors.New("Cannot use MaxCacheSize=0 with KeepBlocksInCache " +
-			"or KeepBlockIndicesInCache set.")
-	}
-
 	// If ValueThreshold is greater than opt.maxBatchSize, we won't be able to push any data using
 	// the transaction APIs. Transaction batches entries into batches of size opt.maxBatchSize.
 	if int64(opt.ValueThreshold) > opt.maxBatchSize {
@@ -317,11 +311,11 @@ func Open(opt Options) (db *DB, err error) {
 		}
 	}()
 
-	if opt.MaxCacheSize > 0 {
+	if opt.BlockCacheSize > 0 {
 		config := ristretto.Config{
 			// Use 5% of cache memory for storing counters.
-			NumCounters: int64(float64(opt.MaxCacheSize) * 0.05 * 2),
-			MaxCost:     int64(float64(opt.MaxCacheSize) * 0.95),
+			NumCounters: int64(float64(opt.BlockCacheSize) * 0.05 * 2),
+			MaxCost:     int64(float64(opt.BlockCacheSize) * 0.95),
 			BufferItems: 64,
 			Metrics:     true,
 			OnExit:      table.BlockEvictHandler,
@@ -332,15 +326,15 @@ func Open(opt Options) (db *DB, err error) {
 		}
 	}
 
-	if opt.MaxBfCacheSize > 0 {
+	if opt.IndexCacheSize > 0 {
 		config := ristretto.Config{
 			// Use 5% of cache memory for storing counters.
-			NumCounters: int64(float64(opt.MaxBfCacheSize) * 0.05 * 2),
-			MaxCost:     int64(float64(opt.MaxBfCacheSize) * 0.95),
+			NumCounters: int64(float64(opt.IndexCacheSize) * 0.05 * 2),
+			MaxCost:     int64(float64(opt.IndexCacheSize) * 0.95),
 			BufferItems: 64,
 			Metrics:     true,
 		}
-		db.bfCache, err = ristretto.NewCache(&config)
+		db.indexCache, err = ristretto.NewCache(&config)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to create bf cache")
 		}
@@ -437,7 +431,7 @@ func (db *DB) cleanup() {
 	db.stopCompactions()
 
 	db.blockCache.Close()
-	db.bfCache.Close()
+	db.indexCache.Close()
 	if db.closers.updateSize != nil {
 		db.closers.updateSize.Signal()
 	}
@@ -455,18 +449,18 @@ func (db *DB) cleanup() {
 	db.vlog.Close()
 }
 
-// DataCacheMetrics returns the metrics for the underlying data cache.
-func (db *DB) DataCacheMetrics() *ristretto.Metrics {
+// BlockCacheMetrics returns the metrics for the underlying block cache.
+func (db *DB) BlockCacheMetrics() *ristretto.Metrics {
 	if db.blockCache != nil {
 		return db.blockCache.Metrics
 	}
 	return nil
 }
 
-// BfCacheMetrics returns the metrics for the underlying bloom filter cache.
-func (db *DB) BfCacheMetrics() *ristretto.Metrics {
-	if db.bfCache != nil {
-		return db.bfCache.Metrics
+// IndexCacheMetrics returns the metrics for the underlying index cache.
+func (db *DB) IndexCacheMetrics() *ristretto.Metrics {
+	if db.indexCache != nil {
+		return db.indexCache.Metrics
 	}
 	return nil
 }
@@ -560,7 +554,7 @@ func (db *DB) close() (err error) {
 	db.closers.updateSize.SignalAndWait()
 	db.orc.Stop()
 	db.blockCache.Close()
-	db.bfCache.Close()
+	db.indexCache.Close()
 
 	if db.opt.InMemory {
 		return
@@ -1050,8 +1044,8 @@ func (db *DB) handleFlushTask(ft flushTask) error {
 	bopts := buildTableOptions(db.opt)
 	bopts.DataKey = dk
 	// Builder does not need cache but the same options are used for opening table.
-	bopts.Cache = db.blockCache
-	bopts.BfCache = db.bfCache
+	bopts.BlockCache = db.blockCache
+	bopts.IndexCache = db.indexCache
 	tableData := buildL0Table(ft, bopts)
 
 	fileID := db.lc.reserveFileID()
@@ -1634,7 +1628,7 @@ func (db *DB) dropAll() (func(), error) {
 	db.lc.nextFileID = 1
 	db.opt.Infof("Deleted %d value log files. DropAll done.\n", num)
 	db.blockCache.Clear()
-	db.bfCache.Clear()
+	db.indexCache.Clear()
 
 	return resume, nil
 }
diff --git a/levels.go b/levels.go
index 7de857e..bfb95df 100644
--- a/levels.go
+++ b/levels.go
@@ -158,8 +158,8 @@ func newLevelsController(db *DB, mf *Manifest) (*levelsController, error) {
 			// Set compression from table manifest.
 			topt.Compression = tf.Compression
 			topt.DataKey = dk
-			topt.Cache = db.blockCache
-			topt.BfCache = db.bfCache
+			topt.BlockCache = db.blockCache
+			topt.IndexCache = db.indexCache
 			t, err := table.OpenTable(fd, topt)
 			if err != nil {
 				if strings.HasPrefix(err.Error(), "CHECKSUM_MISMATCH:") {
@@ -570,8 +570,8 @@ nextTable:
 		bopts := buildTableOptions(s.kv.opt)
 		bopts.DataKey = dk
 		// Builder does not need cache but the same options are used for opening table.
-		bopts.Cache = s.kv.blockCache
-		bopts.BfCache = s.kv.bfCache
+		bopts.BlockCache = s.kv.blockCache
+		bopts.IndexCache = s.kv.indexCache
 		builder := table.NewTableBuilder(bopts)
 		var numKeys, numSkips uint64
 		for ; it.Valid(); it.Next() {
@@ -1139,6 +1139,7 @@ type TableInfo struct {
 	Right       []byte
 	KeyCount    uint64 // Number of keys in the table
 	EstimatedSz uint64
+	IndexSz     int
 }
 
 func (s *levelsController) getTableInfo(withKeysCount bool) (result []TableInfo) {
@@ -1161,6 +1162,7 @@ func (s *levelsController) getTableInfo(withKeysCount bool) (result []TableInfo)
 				Right:       t.Biggest(),
 				KeyCount:    count,
 				EstimatedSz: t.EstimatedSize(),
+				IndexSz:     t.IndexSize(),
 			}
 			result = append(result, info)
 		}
diff --git a/options.go b/options.go
index 9fa4ca2..1e68acd 100644
--- a/options.go
+++ b/options.go
@@ -62,8 +62,8 @@ type Options struct {
 	BlockSize          int
 	BloomFalsePositive float64
 	KeepL0InMemory     bool
-	MaxCacheSize       int64
-	MaxBfCacheSize     int64
+	BlockCacheSize     int64
+	IndexCacheSize     int64
 	LoadBloomsOnOpen   bool
 
 	NumLevelZeroTables      int
@@ -98,12 +98,6 @@ type Options struct {
 	// conflict detection is disabled.
 	DetectConflicts bool
 
-	// KeepBlockIndicesInCache decides whether to keep the block offsets in the cache or not.
-	KeepBlockIndicesInCache bool
-
-	// KeepBlocksInCache decides whether to keep the sst blocks in the cache or not.
-	KeepBlocksInCache bool
-
 	// Transaction start and commit timestamps are managed by end-user.
 	// This is only useful for databases built on top of Badger (like Dgraph).
 	// Not recommended for most users.
@@ -141,9 +135,10 @@ func DefaultOptions(path string) Options {
 		KeepL0InMemory:          false,
 		VerifyValueChecksum:     false,
 		Compression:             options.None,
-		MaxCacheSize:            0,
-		MaxBfCacheSize:          0,
+		BlockCacheSize:          0,
+		IndexCacheSize:          0,
 		LoadBloomsOnOpen:        true,
+
 		// The following benchmarks were done on a 4 KB block size (default block size). The
 		// compression is ratio supposed to increase with increasing compression level but since the
 		// input for compression algorithm is small (4 KB), we don't get significant benefit at
@@ -169,23 +164,19 @@ func DefaultOptions(path string) Options {
 		EncryptionKey:                 []byte{},
 		EncryptionKeyRotationDuration: 10 * 24 * time.Hour, // Default 10 days.
 		DetectConflicts:               true,
-		KeepBlocksInCache:             false,
-		KeepBlockIndicesInCache:       false,
 	}
 }
 
 func buildTableOptions(opt Options) table.Options {
 	return table.Options{
-		TableSize:               uint64(opt.MaxTableSize),
-		BlockSize:               opt.BlockSize,
-		BloomFalsePositive:      opt.BloomFalsePositive,
-		LoadBloomsOnOpen:        opt.LoadBloomsOnOpen,
-		LoadingMode:             opt.TableLoadingMode,
-		ChkMode:                 opt.ChecksumVerificationMode,
-		Compression:             opt.Compression,
-		ZSTDCompressionLevel:    opt.ZSTDCompressionLevel,
-		KeepBlockIndicesInCache: opt.KeepBlockIndicesInCache,
-		KeepBlocksInCache:       opt.KeepBlocksInCache,
+		TableSize:            uint64(opt.MaxTableSize),
+		BlockSize:            opt.BlockSize,
+		BloomFalsePositive:   opt.BloomFalsePositive,
+		LoadBloomsOnOpen:     opt.LoadBloomsOnOpen,
+		LoadingMode:          opt.TableLoadingMode,
+		ChkMode:              opt.ChecksumVerificationMode,
+		Compression:          opt.Compression,
+		ZSTDCompressionLevel: opt.ZSTDCompressionLevel,
 	}
 }
 
@@ -503,7 +494,7 @@ func (opt Options) WithEncryptionKey(key []byte) Options {
 	return opt
 }
 
-// WithEncryptionRotationDuration returns new Options value with the duration set to
+// WithEncryptionKeyRotationDuration returns new Options value with the duration set to
 // the given value.
 //
 // Key Registry will use this duration to create new keys. If the previous generated
@@ -562,18 +553,18 @@ func (opt Options) WithChecksumVerificationMode(cvMode options.ChecksumVerificat
 	return opt
 }
 
-// WithMaxCacheSize returns a new Options value with MaxCacheSize set to the given value.
+// WithBlockCacheSize returns a new Options value with BlockCacheSize set to the given value.
 //
-// This value specifies how much data cache should hold in memory. A small size of cache means lower
-// memory consumption and lookups/iterations would take longer.
-// It is recommended to use a cache if you're using compression or encryption.
+// This value specifies how much data cache should hold in memory. A small size
+// of cache means lower memory consumption and lookups/iterations would take
+// longer. It is recommended to use a cache if you're using compression or encryption.
 // If compression and encryption both are disabled, adding a cache will lead to
-// unnecessary overhead which will affect the read performance. Setting size to zero disables the
-// cache altogether.
+// unnecessary overhead which will affect the read performance. Setting size to
+// zero disables the cache altogether.
 //
-// Default value of MaxCacheSize is zero.
-func (opt Options) WithMaxCacheSize(size int64) Options {
-	opt.MaxCacheSize = size
+// Default value of BlockCacheSize is zero.
+func (opt Options) WithBlockCacheSize(size int64) Options {
+	opt.BlockCacheSize = size
 	return opt
 }
 
@@ -622,22 +613,6 @@ func (opt Options) WithBypassLockGuard(b bool) Options {
 	return opt
 }
 
-// WithMaxBfCacheSize returns a new Options value with MaxBfCacheSize set to the given value.
-//
-// This value specifies how much memory should be used by the bloom filters.
-// Badger uses bloom filters to speed up lookups. Each table has its own bloom
-// filter and each bloom filter is approximately of 5 MB.
-//
-// Zero value for BfCacheSize means all the bloom filters will be kept in
-// memory and the cache is disabled.
-//
-// The default value of MaxBfCacheSize is 0 which means all bloom filters will
-// be kept in memory.
-func (opt Options) WithMaxBfCacheSize(size int64) Options {
-	opt.MaxBfCacheSize = size
-	return opt
-}
-
 // WithLoadBloomsOnOpen returns a new Options value with LoadBloomsOnOpen set to the given value.
 //
 // Badger uses bloom filters to speed up key lookups. When LoadBloomsOnOpen is set
@@ -651,6 +626,24 @@ func (opt Options) WithLoadBloomsOnOpen(b bool) Options {
 	return opt
 }
 
+// WithIndexCacheSize returns a new Options value with IndexCacheSize set to
+// the given value.
+//
+// This value specifies how much memory should be used by table indices. These
+// indices include the block offsets and the bloomfilters. Badger uses bloom
+// filters to speed up lookups. Each table has its own bloom
+// filter and each bloom filter is approximately of 5 MB.
+//
+// Zero value for IndexCacheSize means all the indices will be kept in
+// memory and the cache is disabled.
+//
+// The default value of IndexCacheSize is 0 which means all indices are kept in
+// memory.
+func (opt Options) WithIndexCacheSize(size int64) Options {
+	opt.IndexCacheSize = size
+	return opt
+}
+
 // WithDetectConflicts returns a new Options value with DetectConflicts set to the given value.
 //
 // Detect conflicts options determines if the transactions would be checked for
@@ -664,31 +657,3 @@ func (opt Options) WithDetectConflicts(b bool) Options {
 	opt.DetectConflicts = b
 	return opt
 }
-
-// WithKeepBlockIndicesInCache returns a new Option value with KeepBlockOffsetInCache set to the
-// given value.
-//
-// When this option is set badger will store the block offsets in a cache along with the blocks.
-// The size of the cache is determined by the MaxCacheSize option. When indices
-// are stored in the cache, the read performance might be affected but the
-// cache limits the amount of memory used by the indices.
-//
-// The default value of KeepBlockOffsetInCache is false.
-func (opt Options) WithKeepBlockIndicesInCache(val bool) Options {
-	opt.KeepBlockIndicesInCache = val
-	return opt
-}
-
-// WithKeepBlocksInCache returns a new Option value with KeepBlocksInCache set to the
-// given value.
-//
-// When this option is set badger will store table blocks in the cache. The
-// size of the cache is determined by the MaxCacheSize option. It is not
-// recommended to enable this option if you're not using compression or
-// encryption in badger.
-//
-// The default value of KeepBlocksInCache is false.
-func (opt Options) WithKeepBlocksInCache(val bool) Options {
-	opt.KeepBlocksInCache = val
-	return opt
-}
diff --git a/stream_writer.go b/stream_writer.go
index 4e37934..394ee0f 100644
--- a/stream_writer.go
+++ b/stream_writer.go
@@ -418,8 +418,8 @@ func (w *sortedWriter) createTable(builder *table.Builder) error {
 	fileID := w.db.lc.reserveFileID()
 	opts := buildTableOptions(w.db.opt)
 	opts.DataKey = builder.DataKey()
-	opts.Cache = w.db.blockCache
-	opts.BfCache = w.db.bfCache
+	opts.BlockCache = w.db.blockCache
+	opts.IndexCache = w.db.indexCache
 	var tbl *table.Table
 	if w.db.opt.InMemory {
 		var err error
diff --git a/table/table.go b/table/table.go
index b9b0948..dec3f0f 100644
--- a/table/table.go
+++ b/table/table.go
@@ -79,8 +79,8 @@ type Options struct {
 	// Compression indicates the compression algorithm used for block compression.
 	Compression options.CompressionType
 
-	Cache   *ristretto.Cache
-	BfCache *ristretto.Cache
+	BlockCache *ristretto.Cache
+	IndexCache *ristretto.Cache
 
 	// ZSTDCompressionLevel is the ZSTD compression level used for compressing blocks.
 	ZSTDCompressionLevel int
@@ -88,12 +88,6 @@ type Options struct {
 	// When LoadBloomsOnOpen is set, bloom filters will be read only when they are accessed.
 	// Otherwise they will be loaded on table open.
 	LoadBloomsOnOpen bool
-
-	// KeepBlockIndicesInCache decides whether to keep the block offsets in the cache or not.
-	KeepBlockIndicesInCache bool
-
-	// KeepBlocksInCache decides whether to keep the block in the cache or not.
-	KeepBlocksInCache bool
 }
 
 // TableInterface is useful for testing.
@@ -111,9 +105,9 @@ type Table struct {
 	tableSize int      // Initialized in OpenTable, using fd.Stat().
 	bfLock    sync.Mutex
 
-	blockIndex []*pb.BlockOffset
-	ref        int32    // For file garbage collection. Atomic.
-	bf         *z.Bloom // Nil if BfCache is set.
+	blockOffset []*pb.BlockOffset
+	ref         int32    // For file garbage collection. Atomic.
+	bf          *z.Bloom // Nil if index cache in enabled.
 
 	mmap []byte // Memory mapped.
 
@@ -176,11 +170,11 @@ func (t *Table) DecrRef() error {
 		}
 		// Delete all blocks from the cache.
 		for i := 0; i < t.noOfBlocks; i++ {
-			t.opt.Cache.Del(t.blockCacheKey(i))
+			t.opt.BlockCache.Del(t.blockCacheKey(i))
 		}
-		// Delete bloom filter from the cache.
-		t.opt.BfCache.Del(t.bfCacheKey())
-
+		// Delete bloom filter and indices from the cache.
+		t.opt.IndexCache.Del(t.blockOffsetsCacheKey())
+		t.opt.IndexCache.Del(t.bfCacheKey())
 	}
 	return nil
 }
@@ -361,7 +355,7 @@ func OpenInMemoryTable(data []byte, id uint64, opt *Options) (*Table, error) {
 func (t *Table) initBiggestAndSmallest() error {
 	var err error
 	var ko *pb.BlockOffset
-	if ko, err = t.readIndex(); err != nil {
+	if ko, err = t.initIndex(); err != nil {
 		return errors.Wrapf(err, "failed to read index.")
 	}
 
@@ -412,9 +406,9 @@ func (t *Table) readNoFail(off, sz int) []byte {
 	return res
 }
 
-// readIndex reads the index and populate the necessary table fields and returns
+// initIndex reads the index and populate the necessary table fields and returns
 // first block offset
-func (t *Table) readIndex() (*pb.BlockOffset, error) {
+func (t *Table) initIndex() (*pb.BlockOffset, error) {
 	readPos := t.tableSize
 
 	// Read checksum len from the last 4 bytes.
@@ -447,55 +441,56 @@ func (t *Table) readIndex() (*pb.BlockOffset, error) {
 		return nil, y.Wrapf(err, "failed to verify checksum for table: %s", t.Filename())
 	}
 
-	index := pb.TableIndex{}
-	// Decrypt the table index if it is encrypted.
-	if t.shouldDecrypt() {
-		var err error
-		if data, err = t.decrypt(data); err != nil {
-			return nil, y.Wrapf(err,
-				"Error while decrypting table index for the table %d in Table.readIndex", t.id)
-		}
+	index, err := t.readTableIndex()
+	if err != nil {
+		return nil, err
 	}
-	err := proto.Unmarshal(data, &index)
-	y.Check(err)
 
 	t.estimatedSize = index.EstimatedSize
 	t.hasBloomFilter = len(index.BloomFilter) > 0
 	t.noOfBlocks = len(index.Offsets)
 
-	if t.hasBloomFilter && t.opt.LoadBloomsOnOpen {
-		t.bfLock.Lock()
-		t.bf, _ = t.readBloomFilter()
-		t.bfLock.Unlock()
-	}
-
-	if t.opt.KeepBlockIndicesInCache && t.opt.Cache != nil {
-		t.opt.Cache.Set(
-			t.blockOffsetsCacheKey(),
-			index.Offsets,
-			calculateOffsetsSize(index.Offsets))
+	// 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)
+			}
 
-		return index.Offsets[0], nil
+			t.bfLock.Lock()
+			t.bf = bf
+			t.bfLock.Unlock()
+		}
+		// Keep block offsets in memory since there is no cache.
+		t.blockOffset = index.Offsets
 	}
 
-	t.blockIndex = index.Offsets
+	// 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
 }
 
 // blockOffsets returns block offsets of this table.
 func (t *Table) blockOffsets() []*pb.BlockOffset {
-	if !t.opt.KeepBlockIndicesInCache || t.opt.Cache == nil {
-		return t.blockIndex
+	if t.opt.IndexCache == nil {
+		return t.blockOffset
 	}
 
-	if val, ok := t.opt.Cache.Get(t.blockOffsetsCacheKey()); ok && val != nil {
+	if val, ok := t.opt.IndexCache.Get(t.blockOffsetsCacheKey()); ok && val != nil {
 		return val.([]*pb.BlockOffset)
 	}
 
-	ti := t.readTableIndex()
+	index, err := t.readTableIndex()
+	y.Check(err)
+	t.opt.IndexCache.Set(
+		t.blockOffsetsCacheKey(),
+		index.Offsets,
+		calculateOffsetsSize(index.Offsets))
 
-	t.opt.Cache.Set(t.blockOffsetsCacheKey(), ti.Offsets, calculateOffsetsSize(ti.Offsets))
-	return ti.Offsets
+	return index.Offsets
 }
 
 // calculateOffsetsSize returns the size of *pb.BlockOffset array
@@ -520,9 +515,9 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 	if idx >= t.noOfBlocks {
 		return nil, errors.New("block out of index")
 	}
-	if t.opt.Cache != nil && t.opt.KeepBlocksInCache {
+	if t.opt.BlockCache != nil {
 		key := t.blockCacheKey(idx)
-		blk, ok := t.opt.Cache.Get(key)
+		blk, ok := t.opt.BlockCache.Get(key)
 		if ok && blk != nil {
 			// Use the block only if the increment was successful. The block
 			// could get evicted from the cache between the Get() call and the
@@ -597,14 +592,14 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 	}
 
 	blk.incrRef()
-	if useCache && t.opt.Cache != nil && t.opt.KeepBlocksInCache {
+	if useCache && t.opt.BlockCache != nil {
 		key := t.blockCacheKey(idx)
 		// incrRef should never return false here because we're calling it on a
 		// new block with ref=1.
 		y.AssertTrue(blk.incrRef())
 
 		// Decrement the block ref if we could not insert it in the cache.
-		if !t.opt.Cache.Set(key, blk, blk.size()) {
+		if !t.opt.BlockCache.Set(key, blk, blk.size()) {
 			blk.decrRef()
 		}
 		// We have added an OnReject func in our cache, which gets called in case the block is not
@@ -613,7 +608,7 @@ func (t *Table) block(idx int, useCache bool) (*block, error) {
 	return blk, nil
 }
 
-// bfCacheKey returns the cache key for bloom filter.
+// 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)
@@ -625,6 +620,7 @@ func (t *Table) bfCacheKey() []byte {
 	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)
 	y.AssertTrue(uint32(idx) < math.MaxUint32)
@@ -636,15 +632,19 @@ func (t *Table) blockCacheKey(idx int) []byte {
 	return buf
 }
 
-// blockOffsetsCacheKey returns the cache key for block offsets.
-func (t *Table) blockOffsetsCacheKey() []byte {
-	y.AssertTrue(t.id < math.MaxUint32)
-	buf := make([]byte, 6)
-	buf[0] = 'b'
-	buf[1] = 'o'
+// blockOffsetsCacheKey returns the cache key for block offsets. blockOffsets
+// are stored in the index cache.
+func (t *Table) blockOffsetsCacheKey() uint64 {
+	return t.id
+}
 
-	binary.BigEndian.PutUint32(buf[2:], uint32(t.id))
-	return buf
+// 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
 }
 
 // EstimatedSize returns the total size of key-values stored in this table (including the
@@ -672,28 +672,26 @@ func (t *Table) DoesNotHave(hash uint64) bool {
 	if !t.hasBloomFilter {
 		return false
 	}
-	var bf *z.Bloom
 
 	// Return fast if the cache is absent.
-	if t.opt.BfCache == nil {
+	if t.opt.IndexCache == nil {
 		t.bfLock.Lock()
-		// Load bloomfilter into memory if the cache is absent.
 		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 bloomfilter exists in the cache.
-	if b, ok := t.opt.BfCache.Get(t.bfCacheKey()); b != nil && ok {
-		bf = b.(*z.Bloom)
-		return !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.BfCache.Set(t.bfCacheKey(), bf, int64(sz))
+	t.opt.IndexCache.Set(t.bfCacheKey(), bf, int64(sz))
 	return !bf.Has(hash)
 }
 
@@ -701,24 +699,28 @@ func (t *Table) DoesNotHave(hash uint64) bool {
 // along with the bloom filter.
 func (t *Table) readBloomFilter() (*z.Bloom, int) {
 	// Read bloom filter from the SST.
-	index := t.readTableIndex()
+	index, err := t.readTableIndex()
+	y.Check(err)
+
 	bf, err := z.JSONUnmarshal(index.BloomFilter)
 	y.Check(err)
 	return bf, len(index.BloomFilter)
 }
 
 // readTableIndex reads table index from the sst and returns its pb format.
-func (t *Table) readTableIndex() *pb.TableIndex {
+func (t *Table) readTableIndex() (*pb.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() {
-		data, err = t.decrypt(data)
-		y.Check(err)
+		if data, err = t.decrypt(data); err != nil {
+			return nil, y.Wrapf(err,
+				"Error while decrypting table index for the table %d in readTableIndex", t.id)
+		}
 	}
 	y.Check(proto.Unmarshal(data, &index))
-	return &index
+	return &index, nil
 }
 
 // VerifyChecksum verifies checksum for all blocks of table. This function is called by


Test patch:
diff --git a/db_test.go b/db_test.go
index 78b4f29..febf6d1 100644
--- a/db_test.go
+++ b/db_test.go
@@ -72,8 +72,7 @@ func getTestOptions(dir string) Options {
 	opt := DefaultOptions(dir).
 		WithMaxTableSize(1 << 15). // Force more compaction.
 		WithLevelOneSize(4 << 15). // Force more compaction.
-		WithSyncWrites(false).
-		WithMaxCacheSize(10 << 20)
+		WithSyncWrites(false)
 	if !*mmap {
 		return opt.WithValueLogLoadingMode(options.FileIO)
 	}
@@ -287,12 +286,11 @@ func TestGet(t *testing.T) {
 		test(t, db)
 		require.NoError(t, db.Close())
 	})
-	t.Run("cache disabled", func(t *testing.T) {
-		opts := DefaultOptions("").WithInMemory(true).WithMaxCacheSize(0)
-		db, err := Open(opts)
-		require.NoError(t, err)
-		test(t, db)
-		require.NoError(t, db.Close())
+	t.Run("cache enabled", func(t *testing.T) {
+		opts := DefaultOptions("").WithBlockCacheSize(10 << 20)
+		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
+			test(t, db)
+		})
 	})
 }
 
diff --git a/iterator_test.go b/iterator_test.go
index 0fa45cd..cdd3276 100644
--- a/iterator_test.go
+++ b/iterator_test.go
@@ -210,7 +210,7 @@ func TestIteratePrefix(t *testing.T) {
 	t.Run("With Block Offsets in Cache", func(t *testing.T) {
 		t.Parallel()
 		opts := getTestOptions("")
-		opts = opts.WithKeepBlockIndicesInCache(true)
+		opts.IndexCacheSize = 100 << 20
 		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
 			testIteratorPrefix(t, db)
 		})
@@ -219,7 +219,8 @@ func TestIteratePrefix(t *testing.T) {
 	t.Run("With Block Offsets and Blocks in Cache", func(t *testing.T) {
 		t.Parallel()
 		opts := getTestOptions("")
-		opts = opts.WithKeepBlockIndicesInCache(true).WithKeepBlocksInCache(true)
+		opts.BlockCacheSize = 100 << 20
+		opts.IndexCacheSize = 100 << 20
 		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
 			testIteratorPrefix(t, db)
 		})
@@ -228,7 +229,7 @@ func TestIteratePrefix(t *testing.T) {
 	t.Run("With Blocks in Cache", func(t *testing.T) {
 		t.Parallel()
 		opts := getTestOptions("")
-		opts = opts.WithKeepBlocksInCache(true)
+		opts.BlockCacheSize = 100 << 20
 		runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
 			testIteratorPrefix(t, db)
 		})
diff --git a/table/builder_test.go b/table/builder_test.go
index 522e02e..2bb1866 100644
--- a/table/builder_test.go
+++ b/table/builder_test.go
@@ -120,7 +120,9 @@ func TestTableIndex(t *testing.T) {
 
 			// Ensure index is built correctly
 			require.Equal(t, blockCount, tbl.noOfBlocks)
-			for i, ko := range tbl.readTableIndex().Offsets {
+			idx, err := tbl.readTableIndex()
+			require.NoError(t, err)
+			for i, ko := range idx.Offsets {
 				require.Equal(t, ko.Key, blockFirstKeys[i])
 			}
 			f.Close()
diff --git a/table/table_test.go b/table/table_test.go
index 99d3f1e..b9f6c98 100644
--- a/table/table_test.go
+++ b/table/table_test.go
@@ -781,7 +781,7 @@ func BenchmarkReadAndBuild(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		func() {
 			opts := Options{Compression: options.ZSTD, BlockSize: 4 * 0124, BloomFalsePositive: 0.01}
-			opts.Cache = cache
+			opts.BlockCache = cache
 			newBuilder := NewTableBuilder(opts)
 			it := tbl.NewIterator(0)
 			defer it.Close()
@@ -807,7 +807,7 @@ func BenchmarkReadMerged(b *testing.B) {
 	for i := 0; i < m; i++ {
 		filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Int63())
 		opts := Options{Compression: options.ZSTD, BlockSize: 4 * 1024, BloomFalsePositive: 0.01}
-		opts.Cache = cache
+		opts.BlockCache = cache
 		builder := NewTableBuilder(opts)
 		f, err := y.OpenSyncedFile(filename, true)
 		y.Check(err)
@@ -899,7 +899,7 @@ func getTableForBenchmarks(b *testing.B, count int, cache *ristretto.Cache) *Tab
 		cache, err = ristretto.NewCache(&cacheConfig)
 		require.NoError(b, err)
 	}
-	opts.Cache = cache
+	opts.BlockCache = cache
 	builder := NewTableBuilder(opts)
 	filename := fmt.Sprintf("%s%s%d.sst", os.TempDir(), string(os.PathSeparator), rand.Int63())
 	f, err := y.OpenSyncedFile(filename, true)