compaction skip non-overlapping bottom tables

level_handler.go

@@ -80,9 +80,7 @@ func (s *levelHandler) initTables(tables []*table.Table) {
 		})
 	} else {
 		// Sort tables by keys.
-		sort.Slice(s.tables, func(i, j int) bool {
-			return y.CompareKeysWithVer(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
-		})
+		sortTables(s.tables)
 	}
 }
 
@@ -135,9 +133,15 @@ func assertTablesOrder(tables []*table.Table) {
 	}
 }
 
+func sortTables(tables []*table.Table) {
+	sort.Slice(tables, func(i, j int) bool {
+		return y.CompareKeysWithVer(tables[i].Smallest(), tables[j].Smallest()) < 0
+	})
+}
+
 // replaceTables will replace tables[left:right] with newTables. Note this EXCLUDES tables[right].
 // You must call decr() to delete the old tables _after_ writing the update to the manifest.
-func (s *levelHandler) replaceTables(newTables []*table.Table) error {
+func (s *levelHandler) replaceTables(newTables []*table.Table, skippedTbls []*table.Table) error {
 	// Need to re-search the range of tables in this level to be replaced as other goroutines might
 	// be changing it as well.  (They can't touch our tables, but if they add/remove other tables,
 	// the indices get shifted around.)
@@ -160,30 +164,35 @@ func (s *levelHandler) replaceTables(newTables []*table.Table) error {
 		right: newTables[len(newTables)-1].Biggest(),
 	}
 	left, right := s.overlappingTables(levelHandlerRLocked{}, kr)
-
-	toDecr := make([]*table.Table, right-left)
+	toDecr := make([]*table.Table, 0, right-left)
 	// Update totalSize and reference counts.
 	for i := left; i < right; i++ {
 		tbl := s.tables[i]
-		s.totalSize -= tbl.Size()
-		toDecr[i-left] = tbl
-	}
-
-	// To be safe, just make a copy. TODO: Be more careful and avoid copying.
-	numDeleted := right - left
-	numAdded := len(newTables)
-	tables := make([]*table.Table, len(s.tables)-numDeleted+numAdded)
-	y.Assert(left == copy(tables, s.tables[:left]))
-	t := tables[left:]
-	y.Assert(numAdded == copy(t, newTables))
-	t = t[numAdded:]
-	y.Assert(len(s.tables[right:]) == copy(t, s.tables[right:]))
-	assertTablesOrder(tables)
+		if !isSkippedTable(tbl, skippedTbls) {
+			s.totalSize -= tbl.Size()
+			toDecr = append(toDecr, tbl)
+		}
+	}
+	tables := make([]*table.Table, 0, left+len(newTables)+len(skippedTbls)+(len(s.tables)-right))
+	tables = append(tables, s.tables[:left]...)
+	tables = append(tables, newTables...)
+	tables = append(tables, skippedTbls...)
+	tables = append(tables, s.tables[right:]...)
+	sortTables(tables)
 	s.tables = tables
 	s.Unlock() // s.Unlock before we DecrRef tables -- that can be slow.
 	return decrRefs(toDecr)
 }
 
+func isSkippedTable(tbl *table.Table, skips []*table.Table) bool {
+	for _, skip := range skips {
+		if tbl == skip {
+			return true
+		}
+	}
+	return false
+}
+
 func decrRefs(tables []*table.Table) error {
 	for _, table := range tables {
 		if table == nil {

levels.go

@@ -371,6 +371,7 @@ func (lc *levelsController) compactBuildTables(level int, cd compactDef,
 		filter = lc.kv.opt.CompactionFilterFactory(level+1, cd.smallest(), cd.biggest())
 		guards = filter.Guards()
 	}
+	skippedTbls := cd.skippedTbls
 
 	var lastKey, skipKey []byte
 	var builder *table.Builder
@@ -410,6 +411,13 @@ func (lc *levelsController) compactBuildTables(level int, cd compactDef,
 				// Only break if we are on a different key, and have reached capacity. We want
 				// to ensure that all versions of the key are stored in the same sstable, and
 				// not divided across multiple tables at the same level.
+				if len(skippedTbls) > 0 {
+					skipped := skippedTbls[0]
+					if bytes.Compare(key, skipped.Biggest()) > 0 {
+						skippedTbls = skippedTbls[1:]
+						break
+					}
+				}
 				if shouldFinishFile(key, lastKey, guard, int64(builder.EstimateSize()), lc.kv.opt.MaxTableSize) {
 					break
 				}
@@ -491,10 +499,7 @@ func (lc *levelsController) compactBuildTables(level int, cd compactDef,
 		log.Error(err)
 		return
 	}
-
-	sort.Slice(newTables, func(i, j int) bool {
-		return y.CompareKeysWithVer(newTables[i].Biggest(), newTables[j].Biggest()) < 0
-	})
+	sortTables(newTables)
 	lc.kv.vlog.updateGCStats(discardStats.discardSpaces)
 	log.Infof("Discard stats: %v", discardStats)
 	assertTablesOrder(newTables)
@@ -522,6 +527,8 @@ type compactDef struct {
 	top []*table.Table
 	bot []*table.Table
 
+	skippedTbls []*table.Table
+
 	thisRange keyRange
 	nextRange keyRange
 
@@ -610,8 +617,7 @@ func (lc *levelsController) fillTablesL0(cd *compactDef) bool {
 
 	kr := getKeyRange(cd.top)
 	left, right := cd.nextLevel.overlappingTables(levelHandlerRLocked{}, kr)
-	cd.bot = make([]*table.Table, right-left)
-	copy(cd.bot, cd.nextLevel.tables[left:right])
+	lc.fillBottomTables(cd, cd.nextLevel.tables[left:right])
 
 	if len(cd.bot) == 0 { // the bottom-most level
 		cd.nextRange = kr
@@ -626,6 +632,26 @@ func (lc *levelsController) fillTablesL0(cd *compactDef) bool {
 	return true
 }
 
+func (lc *levelsController) fillBottomTables(cd *compactDef, overlappingTables []*table.Table) {
+	for _, t := range overlappingTables {
+		// If none of the top tables contains the range in an overlapping bottom table,
+		// we can skip it during compaction to reduce write amplification.
+		var added bool
+		for _, topTbl := range cd.top {
+			iter := topTbl.NewIteratorNoRef(false)
+			iter.Seek(t.Smallest())
+			if iter.Valid() && bytes.Compare(iter.Key(), t.Biggest()) < 0 {
+				cd.bot = append(cd.bot, t)
+				added = true
+				break
+			}
+		}
+		if !added {
+			cd.skippedTbls = append(cd.skippedTbls, t)
+		}
+	}
+}
+
 func (lc *levelsController) fillTables(cd *compactDef) bool {
 	cd.lockLevels()
 	defer cd.unlockLevels()
@@ -656,9 +682,7 @@ func (lc *levelsController) fillTables(cd *compactDef) bool {
 		}
 		cd.top = []*table.Table{t}
 		left, right := cd.nextLevel.overlappingTables(levelHandlerRLocked{}, cd.thisRange)
-
-		cd.bot = make([]*table.Table, right-left)
-		copy(cd.bot, cd.nextLevel.tables[left:right])
+		lc.fillBottomTables(cd, cd.nextLevel.tables[left:right])
 
 		if len(cd.bot) == 0 {
 			cd.bot = []*table.Table{}
@@ -780,7 +804,7 @@ func (lc *levelsController) runCompactDef(l int, cd compactDef, limiter *rate.Li
 
 	var newTables []*table.Table
 	var changeSet protos.ManifestChangeSet
-	if l > 0 && len(cd.bot) == 0 {
+	if l > 0 && len(cd.bot) == 0 && len(cd.skippedTbls) == 0 {
 		// skip level 0, since it may has many table overlap with each other
 		newTables = cd.top
 		changeSet = protos.ManifestChangeSet{Changes: []*protos.ManifestChange{
@@ -803,7 +827,7 @@ func (lc *levelsController) runCompactDef(l int, cd compactDef, limiter *rate.Li
 
 	// See comment earlier in this function about the ordering of these ops, and the order in which
 	// we access levels when reading.
-	if err := nextLevel.replaceTables(newTables); err != nil {
+	if err := nextLevel.replaceTables(newTables, cd.skippedTbls); err != nil {
 		return err
 	}
 	if err := thisLevel.deleteTables(cd.top); err != nil {