leveldb: reduce compaction table set of target level

leveldb/session_compaction.go

@@ -144,6 +144,7 @@ func newCompaction(s *session, v *version, sourceLevel int, t0 tFiles, typ int)
 		tPtrs:         make([]int, len(v.levels)),
 	}
 	c.expand()
+	c.reduce()
 	c.save()
 	return c
 }
@@ -157,6 +158,8 @@ type compaction struct {
 	sourceLevel   int
 	levels        [2]tFiles
 	maxGPOverlaps int64
+	skips         tFiles
+	skipIndex     int
 
 	gp                tFiles
 	gpi               int
@@ -170,20 +173,23 @@ type compaction struct {
 	snapSeenKey           bool
 	snapGPOverlappedBytes int64
 	snapTPtrs             []int
+	snapSkipIndex         int
 }
 
 func (c *compaction) save() {
 	c.snapGPI = c.gpi
 	c.snapSeenKey = c.seenKey
 	c.snapGPOverlappedBytes = c.gpOverlappedBytes
 	c.snapTPtrs = append(c.snapTPtrs[:0], c.tPtrs...)
+	c.snapSkipIndex = c.skipIndex
 }
 
 func (c *compaction) restore() {
 	c.gpi = c.snapGPI
 	c.seenKey = c.snapSeenKey
 	c.gpOverlappedBytes = c.snapGPOverlappedBytes
 	c.tPtrs = append(c.tPtrs[:0], c.snapTPtrs...)
+	c.skipIndex = c.snapSkipIndex
 }
 
 func (c *compaction) release() {
@@ -245,6 +251,66 @@ func (c *compaction) expand() {
 	c.imin, c.imax = imin, imax
 }
 
+// reduce tries to reduce table set of target level; need external synchronization.
+func (c *compaction) reduce() {
+	t0, t1 := c.levels[0], c.levels[1]
+	if len(t1) <= 2 {
+		return
+	}
+
+	// Options.
+	ro := &opt.ReadOptions{
+		DontFillCache: true,
+		Strict:        opt.StrictOverride,
+	}
+	strict := c.s.o.GetStrict(opt.StrictCompaction)
+	if strict {
+		ro.Strict |= opt.StrictReader
+	}
+
+	// the source level iterator
+	var it iterator.Iterator
+	if c.sourceLevel == 0 {
+		its := make([]iterator.Iterator, 0, len(t0))
+		for _, t := range t0 {
+			its = append(its, c.s.tops.newIterator(t, nil, ro))
+		}
+		it = iterator.NewMergedIterator(its, c.s.icmp, strict)
+	} else {
+		it = iterator.NewIndexedIterator(t0.newIndexIterator(c.s.tops, c.s.icmp, nil, ro), strict)
+	}
+	defer it.Release()
+
+	ft := make(tFiles, 0, len(t1))
+	skips := make(tFiles, 0, len(t1)-2)
+	targetTableSize := c.s.o.GetCompactionTableSize(c.sourceLevel + 1)
+	for i, t := range t1 {
+		// the first, the last and small tables are ignored.
+		ignore := i == 0 || i == len(t1)-1 || t.size < int64(targetTableSize/10)
+
+		// If no key in source level falls within this table, this table can be safely skipped.
+		if !ignore && it.Seek(t.imin) {
+			// It's important to compare ukeys here, to prevent ukey from hopping across tables
+			// after compaction done.
+			if c.s.icmp.uCompare(internalKey(it.Key()).ukey(), t.imax.ukey()) > 0 {
+				if it.Prev() {
+					if c.s.icmp.uCompare(internalKey(it.Key()).ukey(), t.imin.ukey()) < 0 {
+						skips = append(skips, t)
+						continue
+					}
+				}
+			}
+		}
+		ft = append(ft, t)
+	}
+
+	if it.Error() == nil && len(skips) > 0 {
+		c.s.logf("table@compaction reducing L%d -> L%d F·-%d S·-%s", c.sourceLevel, c.sourceLevel+1, len(skips), shortenb(skips.size()))
+		c.levels[1] = ft
+		c.skips = skips
+	}
+}
+
 // Check whether compaction is trivial.
 func (c *compaction) trivial() bool {
 	return len(c.levels[0]) == 1 && len(c.levels[1]) == 0 && c.gp.size() <= c.maxGPOverlaps
@@ -269,7 +335,7 @@ func (c *compaction) baseLevelForKey(ukey []byte) bool {
 	return true
 }
 
-func (c *compaction) shouldStopBefore(ikey internalKey) bool {
+func (c *compaction) shouldStopBefore(ikey internalKey) (shouldStop bool) {
 	for ; c.gpi < len(c.gp); c.gpi++ {
 		gp := c.gp[c.gpi]
 		if c.s.icmp.Compare(ikey, gp.imax) <= 0 {
@@ -284,9 +350,18 @@ func (c *compaction) shouldStopBefore(ikey internalKey) bool {
 	if c.gpOverlappedBytes > c.maxGPOverlaps {
 		// Too much overlap for current output; start new output.
 		c.gpOverlappedBytes = 0
-		return true
+		shouldStop = true
+	}
+
+	for c.skipIndex < len(c.skips) {
+		if c.s.icmp.uCompare(ikey.ukey(), c.skips[c.skipIndex].imin.ukey()) < 0 {
+			break
+		}
+		// hop across the current skipped table; start new output.
+		c.skipIndex++
+		shouldStop = true
 	}
-	return false
+	return
 }
 
 // Creates an iterator.

leveldb/version.go

@@ -516,10 +516,16 @@ func (p *versionStaging) finish(trivial bool) *version {
 					index := nt.searchNumLess(added[len(added)-1].fd.Num)
 					nt = append(nt[:index], append(added, nt[index:]...)...)
 				} else {
+					imin, imax := added.getRange(p.base.s.icmp)
+					i1 := nt.searchMin(p.base.s.icmp, imin)
+					i2 := nt.searchMin(p.base.s.icmp, imax)
+
+					// join with overlaps, and sort
+					added = append(added, nt[i1:i2]...)
 					added.sortByKey(p.base.s.icmp)
-					_, amax := added.getRange(p.base.s.icmp)
-					index := nt.searchMin(p.base.s.icmp, amax)
-					nt = append(nt[:index], append(added, nt[index:]...)...)
+
+					added = append(added, nt[i2:]...)
+					nt = append(nt[:i1], added...)
 				}
 				nv.levels[level] = nt
 				continue