Repository: dgraph-io/badger
Pull request #1629 (2020-12-24 17:17:00)

debug(iterator): Add debug logs for iterator crash
This PR adds debug logs for the following crash:

```
panic: runtime error: slice bounds out of range [:527064] with capacity 4028

goroutine 3198697 [running]:
github.com/dgraph-io/badger/v2/table.(*blockIterator).setIdx(0xc1066a8010, 0x3b)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/iterator.go:87 +0x52c
github.com/dgraph-io/badger/v2/table.(*blockIterator).next(...)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/iterator.go:154
github.com/dgraph-io/badger/v2/table.(*Iterator).next(0xc1066a8000)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/iterator.go:322 +0x1b5
github.com/dgraph-io/badger/v2/table.(*Iterator).Next(0xc1066a8000)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/iterator.go:382 +0x3e
github.com/dgraph-io/badger/v2/table.(*ConcatIterator).Next(0xc1c2a40910)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/iterator.go:505 +0x33
github.com/dgraph-io/badger/v2/table.(*node).next(0xc18ff847d0)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/merge_iterator.go:81 +0x3d
github.com/dgraph-io/badger/v2/table.(*MergeIterator).Next(0xc18ff84790)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/table/merge_iterator.go:157 +0x38
github.com/dgraph-io/badger/v2.(*levelsController).subcompact.func3(0xc166c6f480)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/levels.go:646 +0x164
github.com/dgraph-io/badger/v2.(*levelsController).subcompact(0xc00011a0e0, 0x20719a0, 0xc18ff84790, 0xc1c26904c0, 0x1e, 0x1e, 0xc1c2690500, 0x1e, 0x1e, 0x0, ...)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/levels.go:765 +0x547
github.com/dgraph-io/badger/v2.(*levelsController).compactBuildTables.func3(0xc154706100, 0xc1547060c0, 0xc00011a0e0, 0xc2e581ab60, 0xc1c266e060, 0xc1c26904c0, 0x1e, 0x1e, 0xc1c2690500, 0x1e, ...)
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/levels.go:868 +0x1c9
created by github.com/dgraph-io/badger/v2.(*levelsController).compactBuildTables
        /usr/local/go/pkg/mod/github.com/dgraph-io/badger/v2@v2.0.1-rc1.0.20201112000450-e002a9d4a500/levels.go:864 +0x620
```

I had to add two new fields (blockID and tableID) in `blockIterator` for the debug logs.

<!-- 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/1629)
<!-- Reviewable:end -->



Patch:
diff --git a/.travis.yml b/.travis.yml
index fbcefba..b671868 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -41,12 +41,8 @@ script: >-
     if [ $TRAVIS_OS_NAME = "linux" ] && [ $go_32 ]; then
       uname -a
       GOOS=linux GOARCH=arm go test -v ./...
-      # Another round of tests after turning off mmap.
-      GOOS=linux GOARCH=arm go test -v -vlog_mmap=false github.com/dgraph-io/badger
     else
       go test -v ./...
-      # Another round of tests after turning off mmap.
-      go test -v -vlog_mmap=false github.com/dgraph-io/badger
       # Cross-compile for Plan 9
       GOOS=plan9 go build ./...
     fi
diff --git a/appveyor.yml b/appveyor.yml
index ac3a950..1842b11 100644
--- a/appveyor.yml
+++ b/appveyor.yml
@@ -36,7 +36,6 @@ test_script:
   # Unit tests
   - ps: Add-AppveyorTest "Unit Tests" -Outcome Running
   - go test -v github.com/dgraph-io/badger/...
-  - go test -v -vlog_mmap=false github.com/dgraph-io/badger/...
   - ps: Update-AppveyorTest "Unit Tests" -Outcome Passed
 
 notifications:
diff --git a/badger/cmd/info.go b/badger/cmd/info.go
index f7c4002..8b47d13 100644
--- a/badger/cmd/info.go
+++ b/badger/cmd/info.go
@@ -256,8 +256,9 @@ func tableInfo(dir, valueDir string, db *badger.DB) {
 	y.Check(err)
 
 	fmt.Println()
-	fmt.Println("SSTable [Li, Id, Total Keys including internal keys] " +
-		"[Compression Ratio, Uncompressed Size, Index Size, BF Size] " +
+	// Total keys includes the internal keys as well.
+	fmt.Println("SSTable [Li, Id, Total Keys] " +
+		"[Compression Ratio, StaleData Ratio, Uncompressed Size, Index Size, BF Size] " +
 		"[Left Key, Version -> Right Key, Version]")
 	totalIndex := uint64(0)
 	totalBloomFilter := uint64(0)
@@ -268,9 +269,11 @@ func tableInfo(dir, valueDir string, db *badger.DB) {
 
 		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)
+		staleDataRatio := float64(t.StaleDataSize) / float64(t.UncompressedSize)
+		fmt.Printf("SSTable [L%d, %03d, %07d] [%.2f, %.2f, %s, %s, %s] [%20X, v%d -> %20X, v%d]\n",
+			t.Level, t.ID, t.KeyCount, compressionRatio, staleDataRatio,
+			hbytes(int64(t.UncompressedSize)), hbytes(int64(t.IndexSz)),
+			hbytes(int64(t.BloomFilterSize)), lk, lt, rk, rt)
 		totalIndex += uint64(t.IndexSz)
 		totalBloomFilter += uint64(t.BloomFilterSize)
 		totalCompressionRatio += compressionRatio
diff --git a/compaction.go b/compaction.go
index 56ce645..a492ebd 100644
--- a/compaction.go
+++ b/compaction.go
@@ -50,6 +50,10 @@ func (r keyRange) equals(dst keyRange) bool {
 }
 
 func (r *keyRange) extend(kr keyRange) {
+	// TODO(ibrahim): Is this needed?
+	if kr.isEmpty() {
+		return
+	}
 	if r.isEmpty() {
 		*r = kr
 	}
@@ -69,6 +73,11 @@ func (r keyRange) overlapsWith(dst keyRange) bool {
 	if r.isEmpty() {
 		return true
 	}
+	// TODO(ibrahim): Do you need this?
+	// Empty dst doesn't overlap with anything.
+	if dst.isEmpty() {
+		return false
+	}
 	if r.inf || dst.inf {
 		return true
 	}
@@ -176,7 +185,7 @@ func (cs *compactStatus) compareAndAdd(_ thisAndNextLevelRLocked, cd compactDef)
 	defer cs.Unlock()
 
 	tl := cd.thisLevel.level
-	y.AssertTruef(tl < len(cs.levels)-1, "Got level %d. Max levels: %d", tl, len(cs.levels))
+	y.AssertTruef(tl < len(cs.levels), "Got level %d. Max levels: %d", tl, len(cs.levels))
 	thisLevel := cs.levels[cd.thisLevel.level]
 	nextLevel := cs.levels[cd.nextLevel.level]
 
@@ -205,14 +214,17 @@ func (cs *compactStatus) delete(cd compactDef) {
 	defer cs.Unlock()
 
 	tl := cd.thisLevel.level
-	y.AssertTruef(tl < len(cs.levels)-1, "Got level %d. Max levels: %d", tl, len(cs.levels))
+	y.AssertTruef(tl < len(cs.levels), "Got level %d. Max levels: %d", tl, len(cs.levels))
 
 	thisLevel := cs.levels[cd.thisLevel.level]
 	nextLevel := cs.levels[cd.nextLevel.level]
 
 	thisLevel.delSize -= cd.thisSize
 	found := thisLevel.remove(cd.thisRange)
-	if !cd.nextRange.isEmpty() {
+	// The following check makes sense only if we're compacting more than one
+	// table. In case of the max level, we might rewrite a single table to
+	// remove stale data.
+	if cd.thisLevel != cd.nextLevel && !cd.nextRange.isEmpty() {
 		found = nextLevel.remove(cd.nextRange) && found
 	}
 
diff --git a/db.go b/db.go
index 6cbeb29..d165cd7 100644
--- a/db.go
+++ b/db.go
@@ -1888,9 +1888,10 @@ func (db *DB) LevelsToString() string {
 	for _, li := range levels {
 		b.WriteString(fmt.Sprintf(
 			"Level %d [%s]: NumTables: %02d. Size: %s of %s. Score: %.2f->%.2f"+
-				" Target FileSize: %s\n",
+				" StaleData: %s Target FileSize: %s\n",
 			li.Level, base(li.IsBaseLevel), li.NumTables,
-			h(li.Size), h(li.TargetSize), li.Score, li.Adjusted, h(li.TargetFileSize)))
+			h(li.Size), h(li.TargetSize), li.Score, li.Adjusted, h(li.StaleDatSize),
+			h(li.TargetFileSize)))
 	}
 	b.WriteString("Level Done\n")
 	return b.String()
diff --git a/fb/TableIndex.go b/fb/TableIndex.go
index 6c5d794..7b4074b 100644
--- a/fb/TableIndex.go
+++ b/fb/TableIndex.go
@@ -128,8 +128,20 @@ func (rcv *TableIndex) MutateOnDiskSize(n uint32) bool {
 	return rcv._tab.MutateUint32Slot(14, n)
 }
 
+func (rcv *TableIndex) StaleDataSize() uint32 {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(16))
+	if o != 0 {
+		return rcv._tab.GetUint32(o + rcv._tab.Pos)
+	}
+	return 0
+}
+
+func (rcv *TableIndex) MutateStaleDataSize(n uint32) bool {
+	return rcv._tab.MutateUint32Slot(16, n)
+}
+
 func TableIndexStart(builder *flatbuffers.Builder) {
-	builder.StartObject(6)
+	builder.StartObject(7)
 }
 func TableIndexAddOffsets(builder *flatbuffers.Builder, offsets flatbuffers.UOffsetT) {
 	builder.PrependUOffsetTSlot(0, flatbuffers.UOffsetT(offsets), 0)
@@ -155,6 +167,9 @@ func TableIndexAddUncompressedSize(builder *flatbuffers.Builder, uncompressedSiz
 func TableIndexAddOnDiskSize(builder *flatbuffers.Builder, onDiskSize uint32) {
 	builder.PrependUint32Slot(5, onDiskSize, 0)
 }
+func TableIndexAddStaleDataSize(builder *flatbuffers.Builder, staleDataSize uint32) {
+	builder.PrependUint32Slot(6, staleDataSize, 0)
+}
 func TableIndexEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
 	return builder.EndObject()
 }
diff --git a/fb/flatbuffer.fbs b/fb/flatbuffer.fbs
index c76fa7d..af559dd 100644
--- a/fb/flatbuffer.fbs
+++ b/fb/flatbuffer.fbs
@@ -23,6 +23,7 @@ table TableIndex {
   key_count:uint32;
   uncompressed_size:uint32;
   on_disk_size:uint32;
+  stale_data_size:uint32;
 }
 
 table BlockOffset {
diff --git a/go.mod b/go.mod
index 7714dbf..3e6e3d1 100644
--- a/go.mod
+++ b/go.mod
@@ -7,7 +7,7 @@ go 1.12
 require (
 	github.com/DataDog/zstd v1.4.1
 	github.com/cespare/xxhash v1.1.0
-	github.com/dgraph-io/ristretto v0.0.4-0.20201205013540-bafef7527542
+	github.com/dgraph-io/ristretto v0.0.4-0.20201224172411-e860a6c48e8a
 	github.com/dustin/go-humanize v1.0.0
 	github.com/golang/protobuf v1.3.1
 	github.com/golang/snappy v0.0.1
diff --git a/go.sum b/go.sum
index ca11151..203bcd8 100644
--- a/go.sum
+++ b/go.sum
@@ -17,8 +17,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3
 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.20201205013540-bafef7527542 h1:Yya3DAZRF8ibfavFQQysegFbPJQ8plpta39uev0lw1I=
-github.com/dgraph-io/ristretto v0.0.4-0.20201205013540-bafef7527542/go.mod h1:tv2ec8nA7vRpSYX7/MbP52ihrUMXIHit54CQMq8npXQ=
+github.com/dgraph-io/ristretto v0.0.4-0.20201224172411-e860a6c48e8a h1:jbJJPy2En2HoeNzt3C8Pc03O6+WRmle0O+xTNiORZW0=
+github.com/dgraph-io/ristretto v0.0.4-0.20201224172411-e860a6c48e8a/go.mod h1:tv2ec8nA7vRpSYX7/MbP52ihrUMXIHit54CQMq8npXQ=
 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=
@@ -77,7 +77,6 @@ github.com/spf13/pflag v0.0.0-20170417173400-9e4c21054fa1/go.mod h1:DYY7MBk1bdzu
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
-github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
@@ -105,7 +104,6 @@ golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73r
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974 h1:IX6qOQeG5uLjB/hjjwjedwfjND0hgjPMMyO1RoIXQNI=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
diff --git a/level_handler.go b/level_handler.go
index 3836be2..06feb76 100644
--- a/level_handler.go
+++ b/level_handler.go
@@ -32,8 +32,9 @@ type levelHandler struct {
 	// For level >= 1, tables are sorted by key ranges, which do not overlap.
 	// For level 0, tables are sorted by time.
 	// For level 0, newest table are at the back. Compact the oldest one first, which is at the front.
-	tables    []*table.Table
-	totalSize int64
+	tables         []*table.Table
+	totalSize      int64
+	totalStaleSize int64
 
 	// The following are initialized once and const.
 	level    int
@@ -41,6 +42,16 @@ type levelHandler struct {
 	db       *DB
 }
 
+func (s *levelHandler) isLastLevel() bool {
+	return s.level == s.db.opt.MaxLevels-1
+}
+
+func (s *levelHandler) getTotalStaleSize() int64 {
+	s.RLock()
+	defer s.RUnlock()
+	return s.totalStaleSize
+}
+
 func (s *levelHandler) getTotalSize() int64 {
 	s.RLock()
 	defer s.RUnlock()
@@ -54,8 +65,9 @@ func (s *levelHandler) initTables(tables []*table.Table) {
 
 	s.tables = tables
 	s.totalSize = 0
+	s.totalStaleSize = 0
 	for _, t := range tables {
-		s.totalSize += t.Size()
+		s.addSize(t)
 	}
 
 	if s.level == 0 {
@@ -89,7 +101,7 @@ func (s *levelHandler) deleteTables(toDel []*table.Table) error {
 			newTables = append(newTables, t)
 			continue
 		}
-		s.totalSize -= t.Size()
+		s.subtractSize(t)
 	}
 	s.tables = newTables
 
@@ -117,12 +129,12 @@ func (s *levelHandler) replaceTables(toDel, toAdd []*table.Table) error {
 			newTables = append(newTables, t)
 			continue
 		}
-		s.totalSize -= t.Size()
+		s.subtractSize(t)
 	}
 
 	// Increase totalSize first.
 	for _, t := range toAdd {
-		s.totalSize += t.Size()
+		s.addSize(t)
 		t.IncrRef()
 		newTables = append(newTables, t)
 	}
@@ -145,7 +157,7 @@ func (s *levelHandler) addTable(t *table.Table) {
 	s.Lock()
 	defer s.Unlock()
 
-	s.totalSize += t.Size() // Increase totalSize first.
+	s.addSize(t) // Increase totalSize first.
 	t.IncrRef()
 	s.tables = append(s.tables, t)
 }
@@ -191,11 +203,22 @@ func (s *levelHandler) tryAddLevel0Table(t *table.Table) bool {
 
 	s.tables = append(s.tables, t)
 	t.IncrRef()
-	s.totalSize += t.Size()
+	s.addSize(t)
 
 	return true
 }
 
+// This should be called while holding the lock on the level.
+func (s *levelHandler) addSize(t *table.Table) {
+	s.totalSize += t.Size()
+	s.totalStaleSize += int64(t.StaleDataSize())
+}
+
+// This should be called while holding the lock on the level.
+func (s *levelHandler) subtractSize(t *table.Table) {
+	s.totalSize -= t.Size()
+	s.totalStaleSize -= int64(t.StaleDataSize())
+}
 func (s *levelHandler) numTables() int {
 	s.RLock()
 	defer s.RUnlock()
diff --git a/levels.go b/levels.go
index 84fd290..984583d 100644
--- a/levels.go
+++ b/levels.go
@@ -445,6 +445,18 @@ func (s *levelsController) runCompactor(id int, lc *z.Closer) {
 		return prios
 	}
 
+	run := func(p compactionPriority) bool {
+		err := s.doCompact(id, p)
+		switch err {
+		case nil:
+			return true
+		case errFillTables:
+			// pass
+		default:
+			s.kv.opt.Warningf("While running doCompact: %v\n", err)
+		}
+		return false
+	}
 	runOnce := func() bool {
 		prios := s.pickCompactLevels()
 		if id == 0 {
@@ -457,20 +469,23 @@ func (s *levelsController) runCompactor(id int, lc *z.Closer) {
 			} else if p.adjusted < 1.0 {
 				break
 			}
-
-			err := s.doCompact(id, p)
-			switch err {
-			case nil:
+			if run(p) {
 				return true
-			case errFillTables:
-				// pass
-			default:
-				s.kv.opt.Warningf("While running doCompact: %v\n", err)
 			}
 		}
+
 		return false
 	}
 
+	tryLmaxToLmaxCompaction := func() {
+		p := compactionPriority{
+			level: s.lastLevel().level,
+			t:     s.levelTargets(),
+		}
+		run(p)
+
+	}
+	count := 0
 	ticker := time.NewTicker(50 * time.Millisecond)
 	defer ticker.Stop()
 	var backOff int
@@ -478,6 +493,7 @@ func (s *levelsController) runCompactor(id int, lc *z.Closer) {
 		select {
 		// Can add a done channel or other stuff.
 		case <-ticker.C:
+			count++
 			if z.NumAllocBytes() > 16<<30 {
 				// Back off. We're already using a lot of memory.
 				backOff++
@@ -486,7 +502,13 @@ func (s *levelsController) runCompactor(id int, lc *z.Closer) {
 				}
 				break
 			}
-			runOnce()
+			// Each ticker is 50ms so 50*200=10seconds.
+			if s.kv.opt.LmaxCompaction && id == 2 && count >= 200 {
+				tryLmaxToLmaxCompaction()
+				count = 0
+			} else {
+				runOnce()
+			}
 		case <-lc.HasBeenClosed():
 			return
 		}
@@ -642,8 +664,13 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 		return r-l >= 10
 	}
 
-	var lastKey, skipKey []byte
-	var numBuilds, numVersions int
+	var (
+		lastKey, skipKey       []byte
+		numBuilds, numVersions int
+		// Denotes if the first key is a series of duplicate keys had
+		// "DiscardEarlierVersions" set
+		firstKeyHasDiscardSet bool
+	)
 
 	addKeys := func(builder *table.Builder) {
 		timeStart := time.Now()
@@ -670,6 +697,7 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 			}
 
 			if !y.SameKey(it.Key(), lastKey) {
+				firstKeyHasDiscardSet = false
 				if len(kr.right) > 0 && y.CompareKeys(it.Key(), kr.right) >= 0 {
 					break
 				}
@@ -681,6 +709,7 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 				}
 				lastKey = y.SafeCopy(lastKey, it.Key())
 				numVersions = 0
+				firstKeyHasDiscardSet = it.Value().Meta&bitDiscardEarlierVersions > 0
 
 				if len(tableKr.left) == 0 {
 					tableKr.left = y.SafeCopy(tableKr.left, it.Key())
@@ -703,6 +732,9 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 
 			vs := it.Value()
 			version := y.ParseTs(it.Key())
+
+			isExpired := isDeletedOrExpired(vs.Meta, vs.ExpiresAt)
+
 			// Do not discard entries inserted by merge operator. These entries will be
 			// discarded once they're merged
 			if version <= discardTs && vs.Meta&bitMergeEntry == 0 {
@@ -710,7 +742,6 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 				// versions which are below the minReadTs, otherwise, we might end up discarding the
 				// only valid version for a running transaction.
 				numVersions++
-
 				// Keep the current version and discard all the next versions if
 				// - The `discardEarlierVersions` bit is set OR
 				// - We've already processed `NumVersionsToKeep` number of versions
@@ -718,8 +749,6 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 				lastValidVersion := vs.Meta&bitDiscardEarlierVersions > 0 ||
 					numVersions == s.kv.opt.NumVersionsToKeep
 
-				isExpired := isDeletedOrExpired(vs.Meta, vs.ExpiresAt)
-
 				if isExpired || lastValidVersion {
 					// If this version of the key is deleted or expired, skip all the rest of the
 					// versions. Ensure that we're only removing versions below readTs.
@@ -748,10 +777,22 @@ func (s *levelsController) subcompact(it y.Iterator, kr keyRange, cd compactDef,
 			if vs.Meta&bitValuePointer > 0 {
 				vp.Decode(vs.Value)
 			}
-			builder.Add(it.Key(), vs, vp.Len)
+			switch {
+			case firstKeyHasDiscardSet:
+				// This key is same as the last key which had "DiscardEarlierVersions" set. The
+				// the next compactions will drop this key if its ts >
+				// discardTs (of the next compaction).
+				builder.AddStaleKey(it.Key(), vs, vp.Len)
+			case isExpired:
+				// If the key is expired, the next compaction will drop it if
+				// its ts > discardTs (of the next compaction).
+				builder.AddStaleKey(it.Key(), vs, vp.Len)
+			default:
+				builder.Add(it.Key(), vs, vp.Len)
+			}
 		}
-		s.kv.opt.Debugf("LOG Compact. Added %d keys. Skipped %d keys. Iteration took: %v",
-			numKeys, numSkips, time.Since(timeStart).Round(time.Millisecond))
+		s.kv.opt.Debugf("[%d] LOG Compact. Added %d keys. Skipped %d keys. Iteration took: %v",
+			cd.compactorId, numKeys, numSkips, time.Since(timeStart).Round(time.Millisecond))
 	} // End of function: addKeys
 
 	if len(kr.left) > 0 {
@@ -1189,6 +1230,18 @@ func (s *levelsController) fillTablesL0(cd *compactDef) bool {
 	return s.fillTablesL0ToL0(cd)
 }
 
+// sortByStaleData sorts tables based on the amount of stale data they have.
+// This is useful in removing tombstones.
+func (s *levelsController) sortByStaleDataSize(tables []*table.Table, cd *compactDef) {
+	if len(tables) == 0 || cd.nextLevel == nil {
+		return
+	}
+
+	sort.Slice(tables, func(i, j int) bool {
+		return tables[i].StaleDataSize() > tables[j].StaleDataSize()
+	})
+}
+
 // sortByHeuristic sorts tables in increasing order of MaxVersion, so we
 // compact older tables first.
 func (s *levelsController) sortByHeuristic(tables []*table.Table, cd *compactDef) {
@@ -1202,6 +1255,91 @@ func (s *levelsController) sortByHeuristic(tables []*table.Table, cd *compactDef
 	})
 }
 
+// This function should be called with lock on levels.
+func (s *levelsController) fillMaxLevelTables(tables []*table.Table, cd *compactDef) bool {
+	sortedTables := make([]*table.Table, len(tables))
+	copy(sortedTables, tables)
+	s.sortByStaleDataSize(sortedTables, cd)
+
+	if len(sortedTables) > 0 && sortedTables[0].StaleDataSize() == 0 {
+		// This is a maxLevel to maxLevel compaction and we don't have any stale data.
+		return false
+	}
+	cd.bot = []*table.Table{}
+	collectBotTables := func(t *table.Table, needSz int64) {
+		totalSize := t.Size()
+
+		j := sort.Search(len(tables), func(i int) bool {
+			return y.CompareKeys(tables[i].Smallest(), t.Smallest()) >= 0
+		})
+		y.AssertTrue(tables[j].ID() == t.ID())
+		j++
+		// Collect tables until we reach the the required size.
+		for j < len(tables) {
+			newT := tables[j]
+			totalSize += newT.Size()
+
+			if totalSize >= needSz {
+				break
+			}
+			cd.bot = append(cd.bot, newT)
+			cd.nextRange.extend(getKeyRange(newT))
+			j++
+		}
+	}
+	now := time.Now()
+	for _, t := range sortedTables {
+		// If the maxVersion is above the discardTs, we won't clean anything in
+		// the compaction. So skip this table.
+		if t.MaxVersion() > s.kv.orc.discardAtOrBelow() {
+			continue
+		}
+		if now.Sub(t.CreatedAt) < time.Hour {
+			// Just created it an hour ago. Don't pick for compaction.
+			continue
+		}
+		// If the stale data size is less than 10 MB, it might not be worth
+		// rewriting the table. Skip it.
+		if t.StaleDataSize() < 10<<20 {
+			continue
+		}
+
+		cd.thisSize = t.Size()
+		cd.thisRange = getKeyRange(t)
+		// Set the next range as the same as the current range. If we don't do
+		// this, we won't be able to run more than one max level compactions.
+		cd.nextRange = cd.thisRange
+		// If we're already compacting this range, don't do anything.
+		if s.cstatus.overlapsWith(cd.thisLevel.level, cd.thisRange) {
+			continue
+		}
+
+		// Found a valid table!
+		cd.top = []*table.Table{t}
+
+		needFileSz := cd.t.fileSz[cd.thisLevel.level]
+		// The table size is what we want so no need to collect more tables.
+		if t.Size() >= needFileSz {
+			break
+		}
+		// TableSize is less than what we want. Collect more tables for compaction.
+		// If the level has multiple small tables, we collect all of them
+		// together to form a bigger table.
+		collectBotTables(t, needFileSz)
+		if !s.cstatus.compareAndAdd(thisAndNextLevelRLocked{}, *cd) {
+			cd.bot = cd.bot[:0]
+			cd.nextRange = keyRange{}
+			continue
+		}
+		return true
+	}
+	if len(cd.top) == 0 {
+		return false
+	}
+
+	return s.cstatus.compareAndAdd(thisAndNextLevelRLocked{}, *cd)
+}
+
 func (s *levelsController) fillTables(cd *compactDef) bool {
 	cd.lockLevels()
 	defer cd.unlockLevels()
@@ -1211,6 +1349,10 @@ func (s *levelsController) fillTables(cd *compactDef) bool {
 	if len(tables) == 0 {
 		return false
 	}
+	// We're doing a maxLevel to maxLevel compaction. Pick tables based on the stale data size.
+	if cd.thisLevel.isLastLevel() {
+		return s.fillMaxLevelTables(tables, cd)
+	}
 	// We pick tables, so we compact older tables first. This is similar to
 	// kOldestLargestSeqFirst in RocksDB.
 	s.sortByHeuristic(tables, cd)
@@ -1225,18 +1367,6 @@ func (s *levelsController) fillTables(cd *compactDef) bool {
 		cd.top = []*table.Table{t}
 		left, right := cd.nextLevel.overlappingTables(levelHandlerRLocked{}, cd.thisRange)
 
-		// Sometimes below line(make([]*table.Table, right-left)) panics with error
-		// (runtime error: makeslice: len out of range). One of the reason for this can be when
-		// right < left. We don't know how to reproduce it as of now. We are just logging it so
-		// that we can get more context.
-		if right < left {
-			s.kv.opt.Errorf("right: %d is less than left: %d in overlappingTables for current "+
-				"level: %d, next level: %d, key range(%s, %s)", right, left, cd.thisLevel.level,
-				cd.nextLevel.level, cd.thisRange.left, cd.thisRange.right)
-
-			continue
-		}
-
 		cd.bot = make([]*table.Table, right-left)
 		copy(cd.bot, cd.nextLevel.tables[left:right])
 
@@ -1271,8 +1401,8 @@ func (s *levelsController) runCompactDef(id, l int, cd compactDef) (err error) {
 	nextLevel := cd.nextLevel
 
 	y.AssertTrue(len(cd.splits) == 0)
-	if thisLevel.level == 0 && nextLevel.level == 0 {
-		// don't do anything for L0 -> L0.
+	if thisLevel.level == nextLevel.level {
+		// don't do anything for L0 -> L0 and Lmax -> Lmax.
 	} else {
 		s.addSplits(&cd)
 	}
@@ -1314,7 +1444,6 @@ func (s *levelsController) runCompactDef(id, l int, cd compactDef) (err error) {
 
 	from := append(tablesToString(cd.top), tablesToString(cd.bot)...)
 	to := tablesToString(newTables)
-
 	if dur := time.Since(timeStart); dur > 2*time.Second {
 		var expensive string
 		if dur > time.Second {
@@ -1350,7 +1479,7 @@ var errFillTables = errors.New("Unable to fill tables")
 // doCompact picks some table on level l and compacts it away to the next level.
 func (s *levelsController) doCompact(id int, p compactionPriority) error {
 	l := p.level
-	y.AssertTrue(l+1 < s.kv.opt.MaxLevels) // Sanity check.
+	y.AssertTrue(l < s.kv.opt.MaxLevels) // Sanity check.
 	if p.t.baseLevel == 0 {
 		p.t = s.levelTargets()
 	}
@@ -1375,7 +1504,11 @@ func (s *levelsController) doCompact(id int, p compactionPriority) error {
 			return errFillTables
 		}
 	} else {
-		cd.nextLevel = s.levels[l+1]
+		cd.nextLevel = cd.thisLevel
+		// We're not compacting the last level so pick the next level.
+		if !cd.thisLevel.isLastLevel() {
+			cd.nextLevel = s.levels[l+1]
+		}
 		if !s.fillTables(&cd) {
 			return errFillTables
 		}
@@ -1494,6 +1627,7 @@ type TableInfo struct {
 	Right            []byte
 	KeyCount         uint32 // Number of keys in the table
 	OnDiskSize       uint32
+	StaleDataSize    uint32
 	UncompressedSize uint32
 	MaxVersion       uint64
 	IndexSz          int
@@ -1511,6 +1645,7 @@ func (s *levelsController) getTableInfo() (result []TableInfo) {
 				Right:            t.Biggest(),
 				KeyCount:         t.KeyCount(),
 				OnDiskSize:       t.OnDiskSize(),
+				StaleDataSize:    t.StaleDataSize(),
 				IndexSz:          t.IndexSize(),
 				BloomFilterSize:  t.BloomFilterSize(),
 				UncompressedSize: t.UncompressedSize(),
@@ -1538,6 +1673,7 @@ type LevelInfo struct {
 	IsBaseLevel    bool
 	Score          float64
 	Adjusted       float64
+	StaleDatSize   int64
 }
 
 func (s *levelsController) getLevelInfo() []LevelInfo {
@@ -1549,6 +1685,8 @@ func (s *levelsController) getLevelInfo() []LevelInfo {
 		result[i].Level = i
 		result[i].Size = l.totalSize
 		result[i].NumTables = len(l.tables)
+		result[i].StaleDatSize = l.totalStaleSize
+
 		l.RUnlock()
 
 		result[i].TargetSize = t.targetSz[i]
diff --git a/options.go b/options.go
index 8d78934..5b75cfe 100644
--- a/options.go
+++ b/options.go
@@ -75,6 +75,7 @@ type Options struct {
 
 	NumCompactors        int
 	CompactL0OnClose     bool
+	LmaxCompaction       bool
 	ZSTDCompressionLevel int
 
 	// When set, checksum will be validated for each entry read from the value log file.
diff --git a/stream_writer.go b/stream_writer.go
index 6302d89..925313d 100644
--- a/stream_writer.go
+++ b/stream_writer.go
@@ -360,6 +360,7 @@ func (w *sortedWriter) Add(key []byte, vs y.ValueStruct) error {
 	}
 
 	sameKey := y.SameKey(key, w.lastKey)
+
 	// Same keys should go into the same SSTable.
 	if !sameKey && w.builder.ReachedCapacity() {
 		if err := w.send(false); err != nil {
@@ -372,6 +373,7 @@ func (w *sortedWriter) Add(key []byte, vs y.ValueStruct) error {
 	if vs.Meta&bitValuePointer > 0 {
 		vp.Decode(vs.Value)
 	}
+
 	w.builder.Add(key, vs, vp.Len)
 	return nil
 }
diff --git a/table/builder.go b/table/builder.go
index 5f40038..8b03c7f 100644
--- a/table/builder.go
+++ b/table/builder.go
@@ -89,6 +89,7 @@ type Builder struct {
 	opts          *Options
 	maxVersion    uint64
 	onDiskSize    uint32
+	staleDataSize int
 
 	// Used to concurrently compress/encrypt blocks.
 	wg        sync.WaitGroup
@@ -317,9 +318,26 @@ func (b *Builder) shouldFinishBlock(key []byte, value y.ValueStruct) bool {
 	return estimatedSize > uint32(b.opts.BlockSize)
 }
 
+// AddStaleKey is same is Add function but it also increments the internal
+// staleDataSize counter. This value will be used to prioritize this table for
+// compaction.
+func (b *Builder) AddStaleKey(key []byte, v y.ValueStruct, valueLen uint32) {
+	// Rough estimate based on how much space it will occupy in the SST.
+	b.staleDataSize += len(key) + len(v.Value) + 4 /* entry offset */ + 4 /* header size */
+	b.addInternal(key, v, valueLen, true)
+}
+
 // Add adds a key-value pair to the block.
 func (b *Builder) Add(key []byte, value y.ValueStruct, valueLen uint32) {
+	b.addInternal(key, value, valueLen, false)
+}
+
+func (b *Builder) addInternal(key []byte, value y.ValueStruct, valueLen uint32, isStale bool) {
 	if b.shouldFinishBlock(key, value) {
+		if isStale {
+			// This key will be added to tableIndex and it is stale.
+			b.staleDataSize += len(key) + 4 /* len */ + 4 /* offset */
+		}
 		b.finishBlock()
 		// Create a new block and start writing.
 		b.curBlock = &bblock{
@@ -530,6 +548,7 @@ func (b *Builder) buildIndex(bloom []byte) ([]byte, uint32) {
 	fb.TableIndexAddUncompressedSize(builder, b.uncompressedSize)
 	fb.TableIndexAddKeyCount(builder, uint32(len(b.keyHashes)))
 	fb.TableIndexAddOnDiskSize(builder, b.onDiskSize)
+	fb.TableIndexAddStaleDataSize(builder, uint32(b.staleDataSize))
 	builder.Finish(fb.TableIndexEnd(builder))
 
 	buf := builder.FinishedBytes()
diff --git a/table/iterator.go b/table/iterator.go
index 2730568..7c48c68 100644
--- a/table/iterator.go
+++ b/table/iterator.go
@@ -18,6 +18,7 @@ package table
 
 import (
 	"bytes"
+	"fmt"
 	"io"
 	"sort"
 
@@ -35,6 +36,8 @@ type blockIterator struct {
 	entryOffsets []uint32
 	block        *block
 
+	tableID uint64
+	blockID int
 	// prevOverlap stores the overlap of the previous key with the base key.
 	// This avoids unnecessary copy of base key when the overlap is same for multiple keys.
 	prevOverlap uint16
@@ -83,6 +86,17 @@ func (itr *blockIterator) setIdx(i int) {
 		// EndOffset of the current entry is the start offset of the next entry.
 		endOffset = int(itr.entryOffsets[itr.idx+1])
 	}
+	defer func() {
+		if r := recover(); r != nil {
+			var debugBuf bytes.Buffer
+			fmt.Fprintf(&debugBuf, "==== Recovered====\n")
+			fmt.Fprintf(&debugBuf, "Table ID: %d\nBlock ID: %d\nEntry Idx: %d\nData len: %d\n"+
+				"StartOffset: %d\nEndOffset: %d\nEntryOffsets len: %d\nEntryOffsets: %v\n",
+				itr.tableID, itr.blockID, itr.idx, len(itr.data), startOffset, endOffset,
+				len(itr.entryOffsets), itr.entryOffsets)
+			panic(debugBuf.String())
+		}
+	}()
 
 	entryData := itr.data[startOffset:endOffset]
 	var h header
@@ -209,6 +223,8 @@ func (itr *Iterator) seekToFirst() {
 		itr.err = err
 		return
 	}
+	itr.bi.tableID = itr.t.id
+	itr.bi.blockID = itr.bpos
 	itr.bi.setBlock(block)
 	itr.bi.seekToFirst()
 	itr.err = itr.bi.Error()
@@ -226,6 +242,8 @@ func (itr *Iterator) seekToLast() {
 		itr.err = err
 		return
 	}
+	itr.bi.tableID = itr.t.id
+	itr.bi.blockID = itr.bpos
 	itr.bi.setBlock(block)
 	itr.bi.seekToLast()
 	itr.err = itr.bi.Error()
@@ -238,6 +256,8 @@ func (itr *Iterator) seekHelper(blockIdx int, key []byte) {
 		itr.err = err
 		return
 	}
+	itr.bi.tableID = itr.t.id
+	itr.bi.blockID = itr.bpos
 	itr.bi.setBlock(block)
 	itr.bi.seek(key, origin)
 	itr.err = itr.bi.Error()
@@ -313,6 +333,8 @@ func (itr *Iterator) next() {
 			itr.err = err
 			return
 		}
+		itr.bi.tableID = itr.t.id
+		itr.bi.blockID = itr.bpos
 		itr.bi.setBlock(block)
 		itr.bi.seekToFirst()
 		itr.err = itr.bi.Error()
@@ -341,6 +363,8 @@ func (itr *Iterator) prev() {
 			itr.err = err
 			return
 		}
+		itr.bi.tableID = itr.t.id
+		itr.bi.blockID = itr.bpos
 		itr.bi.setBlock(block)
 		itr.bi.seekToLast()
 		itr.err = itr.bi.Error()
diff --git a/table/table.go b/table/table.go
index 22d3f71..2fe3232 100644
--- a/table/table.go
+++ b/table/table.go
@@ -107,10 +107,8 @@ type Table struct {
 	smallest, biggest []byte // Smallest and largest keys (with timestamps).
 	id                uint64 // file id, part of filename
 
-	Checksum  []byte
-	CreatedAt time.Time
-	// Stores the total size of key-values stored in this table (including the size on vlog).
-	onDiskSize     uint32
+	Checksum       []byte
+	CreatedAt      time.Time
 	indexStart     int
 	indexLen       int
 	hasBloomFilter bool
@@ -343,6 +341,60 @@ func OpenInMemoryTable(data []byte, id uint64, opt *Options) (*Table, error) {
 }
 
 func (t *Table) initBiggestAndSmallest() error {
+	// This defer will help gathering debugging info incase initIndex crashes.
+	defer func() {
+		if r := recover(); r != nil {
+			// Use defer for printing info because there may be an intermediate panic.
+			var debugBuf bytes.Buffer
+			defer func() {
+				panic(fmt.Sprintf("%s\n== Recovered ==\n", debugBuf.String()))
+			}()
+
+			// Get the count of null bytes at the end of file. This is to make sure if there was an
+			// issue with mmap sync or file copy.
+			count := 0
+			for i := len(t.Data) - 1; i >= 0; i-- {
+				if t.Data[i] != 0 {
+					break
+				}
+				count++
+			}
+
+			fmt.Fprintf(&debugBuf, "\n== Recovering from initIndex crash ==\n")
+			fmt.Fprintf(&debugBuf, "File Info: [ID: %d, Size: %d, Zeros: %d]\n",
+				t.id, t.tableSize, count)
+
+			fmt.Fprintf(&debugBuf, "isEnrypted: %v ", t.shouldDecrypt())
+
+			readPos := t.tableSize
+
+			// Read checksum size.
+			readPos -= 4
+			buf := t.readNoFail(readPos, 4)
+			checksumLen := int(y.BytesToU32(buf))
+			fmt.Fprintf(&debugBuf, "checksumLen: %d ", checksumLen)
+
+			// Read checksum.
+			checksum := &pb.Checksum{}
+			readPos -= checksumLen
+			buf = t.readNoFail(readPos, checksumLen)
+			proto.Unmarshal(buf, checksum)
+			fmt.Fprintf(&debugBuf, "checksum: %+v ", checksum)
+
+			// Read index size from the footer.
+			readPos -= 4
+			buf = t.readNoFail(readPos, 4)
+			indexLen := int(y.BytesToU32(buf))
+			fmt.Fprintf(&debugBuf, "indexLen: %d ", indexLen)
+
+			// Read index.
+			readPos -= t.indexLen
+			t.indexStart = readPos
+			indexData := t.readNoFail(readPos, t.indexLen)
+			fmt.Fprintf(&debugBuf, "index: %v ", indexData)
+		}
+	}()
+
 	var err error
 	var ko *fb.BlockOffset
 	if ko, err = t.initIndex(); err != nil {
@@ -608,6 +660,9 @@ func (t *Table) IndexSize() int {
 // Size is its file size in bytes
 func (t *Table) Size() int64 { return int64(t.tableSize) }
 
+// StaleDataSize is the amount of stale data (that can be dropped by a compaction )in this SST.
+func (t *Table) StaleDataSize() uint32 { return t.fetchIndex().StaleDataSize() }
+
 // Smallest is its smallest key, or nil if there are none
 func (t *Table) Smallest() []byte { return t.smallest }
 
diff --git a/util.go b/util.go
index e8f710e..4d12d58 100644
--- a/util.go
+++ b/util.go
@@ -53,9 +53,10 @@ func (s *levelHandler) validate() error {
 
 		if y.CompareKeys(s.tables[j-1].Biggest(), s.tables[j].Smallest()) >= 0 {
 			return errors.Errorf(
-				"Inter: Biggest(j-1) \n%s\n vs Smallest(j): \n%s\n: level=%d j=%d numTables=%d",
-				hex.Dump(s.tables[j-1].Biggest()), hex.Dump(s.tables[j].Smallest()),
-				s.level, j, numTables)
+				"Inter: Biggest(j-1)[%d] \n%s\n vs Smallest(j)[%d]: \n%s\n: "+
+					"level=%d j=%d numTables=%d",
+				s.tables[j-1].ID(), hex.Dump(s.tables[j-1].Biggest()), s.tables[j].ID(),
+				hex.Dump(s.tables[j].Smallest()), s.level, j, numTables)
 		}
 
 		if y.CompareKeys(s.tables[j].Smallest(), s.tables[j].Biggest()) > 0 {
diff --git a/y/bloom.go b/y/bloom.go
index 390a89f..806770b 100644
--- a/y/bloom.go
+++ b/y/bloom.go
@@ -119,7 +119,7 @@ func Hash(b []byte) uint32 {
 		seed = 0xbc9f1d34
 		m    = 0xc6a4a793
 	)
-	h := uint32(seed) ^ uint32(len(b)*m)
+	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


Test patch:
diff --git a/levels_test.go b/levels_test.go
index 76746ff..5969d9f 100644
--- a/levels_test.go
+++ b/levels_test.go
@@ -1053,6 +1053,79 @@ func TestKeyVersions(t *testing.T) {
 	})
 }
 
+func TestSameLevel(t *testing.T) {
+	opt := DefaultOptions("")
+	opt.NumCompactors = 0
+	opt.NumVersionsToKeep = math.MaxInt32
+	opt.managedTxns = true
+	opt.LmaxCompaction = true
+	runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
+		l6 := []keyValVersion{
+			{"A", "bar", 4, bitDiscardEarlierVersions}, {"A", "bar", 3, 0},
+			{"A", "bar", 2, 0}, {"Afoo", "baz", 2, 0},
+		}
+		l61 := []keyValVersion{
+			{"B", "bar", 4, bitDiscardEarlierVersions}, {"B", "bar", 3, 0},
+			{"B", "bar", 2, 0}, {"Bfoo", "baz", 2, 0},
+		}
+		l62 := []keyValVersion{
+			{"C", "bar", 4, bitDiscardEarlierVersions}, {"C", "bar", 3, 0},
+			{"C", "bar", 2, 0}, {"Cfoo", "baz", 2, 0},
+		}
+		createAndOpen(db, l6, 6)
+		createAndOpen(db, l61, 6)
+		createAndOpen(db, l62, 6)
+		require.NoError(t, db.lc.validate())
+
+		getAllAndCheck(t, db, []keyValVersion{
+			{"A", "bar", 4, bitDiscardEarlierVersions}, {"A", "bar", 3, 0},
+			{"A", "bar", 2, 0}, {"Afoo", "baz", 2, 0},
+			{"B", "bar", 4, bitDiscardEarlierVersions}, {"B", "bar", 3, 0},
+			{"B", "bar", 2, 0}, {"Bfoo", "baz", 2, 0},
+			{"C", "bar", 4, bitDiscardEarlierVersions}, {"C", "bar", 3, 0},
+			{"C", "bar", 2, 0}, {"Cfoo", "baz", 2, 0},
+		})
+
+		cdef := compactDef{
+			thisLevel: db.lc.levels[6],
+			nextLevel: db.lc.levels[6],
+			top:       []*table.Table{db.lc.levels[6].tables[0]},
+			bot:       db.lc.levels[6].tables[1:],
+			t:         db.lc.levelTargets(),
+		}
+		cdef.t.baseLevel = 1
+		// Set dicardTs to 3. foo2 and foo1 should be dropped.
+		db.SetDiscardTs(3)
+		require.NoError(t, db.lc.runCompactDef(-1, 6, cdef))
+		getAllAndCheck(t, db, []keyValVersion{
+			{"A", "bar", 4, bitDiscardEarlierVersions}, {"A", "bar", 3, 0},
+			{"A", "bar", 2, 0}, {"Afoo", "baz", 2, 0},
+			{"B", "bar", 4, bitDiscardEarlierVersions}, {"B", "bar", 3, 0},
+			{"B", "bar", 2, 0}, {"Bfoo", "baz", 2, 0},
+			{"C", "bar", 4, bitDiscardEarlierVersions}, {"C", "bar", 3, 0},
+			{"C", "bar", 2, 0}, {"Cfoo", "baz", 2, 0},
+		})
+
+		require.NoError(t, db.lc.validate())
+		// Set dicardTs to 7.
+		db.SetDiscardTs(7)
+		cdef = compactDef{
+			thisLevel: db.lc.levels[6],
+			nextLevel: db.lc.levels[6],
+			top:       []*table.Table{db.lc.levels[6].tables[0]},
+			bot:       db.lc.levels[6].tables[1:],
+			t:         db.lc.levelTargets(),
+		}
+		cdef.t.baseLevel = 1
+		require.NoError(t, db.lc.runCompactDef(-1, 6, cdef))
+		getAllAndCheck(t, db, []keyValVersion{
+			{"A", "bar", 4, bitDiscardEarlierVersions}, {"Afoo", "baz", 2, 0},
+			{"B", "bar", 4, bitDiscardEarlierVersions}, {"Bfoo", "baz", 2, 0},
+			{"C", "bar", 4, bitDiscardEarlierVersions}, {"Cfoo", "baz", 2, 0}})
+		require.NoError(t, db.lc.validate())
+	})
+}
+
 func TestTableContainsPrefix(t *testing.T) {
 	opts := table.Options{
 		BlockSize:          4 * 1024,
@@ -1091,3 +1164,61 @@ func TestTableContainsPrefix(t *testing.T) {
 	require.False(t, containsPrefix(tbl, []byte("key323")))
 	require.False(t, containsPrefix(tbl, []byte("key5")))
 }
+
+func TestStaleDataCleanup(t *testing.T) {
+	opt := DefaultOptions("")
+	opt.managedTxns = true
+	opt.LmaxCompaction = true
+	runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
+		opts := table.Options{
+			BlockSize:          4 * 1024,
+			BloomFalsePositive: 0.01,
+		}
+		buildStaleTable := func(prefix byte) *table.Table {
+			filename := table.NewFilename(db.lc.reserveFileID(), db.opt.Dir)
+			b := table.NewTableBuilder(opts)
+			defer b.Close()
+			key := make([]byte, 100)
+			val := make([]byte, 500)
+			rand.Read(val)
+
+			copy(key, []byte{prefix})
+			count := uint64(40000)
+			for i := count; i > 0; i-- {
+				var meta byte
+				if i == 0 {
+					meta = bitDiscardEarlierVersions
+				}
+				b.AddStaleKey(y.KeyWithTs(key, i), y.ValueStruct{Meta: meta, Value: val}, 0)
+			}
+			tbl, err := table.CreateTable(filename, b)
+			require.NoError(t, err)
+			return tbl
+		}
+
+		level := 6
+		lh := db.lc.levels[level]
+		for i := byte(1); i < 5; i++ {
+			tab := buildStaleTable(i)
+			require.NoError(t, db.manifest.addChanges([]*pb.ManifestChange{
+				newCreateChange(tab.ID(), level, 0, tab.CompressionType()),
+			}))
+			tab.CreatedAt = time.Now().Add(-10 * time.Hour)
+			// Add table to the given level.
+			lh.addTable(tab)
+		}
+		require.NoError(t, db.lc.validate())
+
+		require.NotZero(t, lh.getTotalStaleSize())
+
+		db.SetDiscardTs(1 << 30)
+		// Modify the target file size so that we can compact all tables at once.
+		tt := db.lc.levelTargets()
+		tt.fileSz[6] = 1 << 30
+		prio := compactionPriority{level: 6, t: tt}
+		require.NoError(t, db.lc.doCompact(-1, prio))
+
+		require.Zero(t, lh.getTotalStaleSize())
+
+	})
+}