func (s *levelsController) compactBuildTables(
	lev int, cd compactDef) ([]*table.Table, func() error, error) {
	topTables := cd.top
	botTables := cd.bot

	var hasOverlap bool
	{
		kr := getKeyRange(cd.top)
		for i, lh := range s.levels {
			if i <= lev { // Skip upper levels.
				continue
			}
			lh.RLock()
			left, right := lh.overlappingTables(levelHandlerRLocked{}, kr)
			lh.RUnlock()
			if right-left > 0 {
				hasOverlap = true
				break
			}
		}
	}

	// Try to collect stats so that we can inform value log about GC. That would help us find which
	// value log file should be GCed.
	discardStats := make(map[uint32]int64)
	updateStats := func(vs y.ValueStruct) {
		if vs.Meta&bitValuePointer > 0 {
			var vp valuePointer
			vp.Decode(vs.Value)
			discardStats[vp.Fid] += int64(vp.Len)
		}
	}

	// Create iterators across all the tables involved first.
	var iters []y.Iterator
	if lev == 0 {
		iters = appendIteratorsReversed(iters, topTables, false)
	} else if len(topTables) > 0 {
		y.AssertTrue(len(topTables) == 1)
		iters = []y.Iterator{topTables[0].NewIterator(false)}
	}

	// Next level has level>=1 and we can use ConcatIterator as key ranges do not overlap.
	var valid []*table.Table
	for _, table := range botTables {
		if len(cd.dropPrefix) > 0 &&
			bytes.HasPrefix(table.Smallest(), cd.dropPrefix) &&
			bytes.HasPrefix(table.Biggest(), cd.dropPrefix) {
			// All the keys in this table have the dropPrefix. So, this table does not need to be
			// in the iterator and can be dropped immediately.
			continue
		}
		valid = append(valid, table)
	}
	iters = append(iters, table.NewConcatIterator(valid, false))
	it := y.NewMergeIterator(iters, false)
	defer it.Close() // Important to close the iterator to do ref counting.

	it.Rewind()

	// Pick a discard ts, so we can discard versions below this ts. We should
	// never discard any versions starting from above this timestamp, because
	// that would affect the snapshot view guarantee provided by transactions.
	discardTs := s.kv.orc.discardAtOrBelow()

	// Start generating new tables.
	type newTableResult struct {
		table *table.Table
		err   error
	}
	resultCh := make(chan newTableResult)
	var numBuilds, numVersions int
	var lastKey, skipKey []byte
	for it.Valid() {
		timeStart := time.Now()
		builder := table.NewTableBuilder()
		var numKeys, numSkips uint64
		for ; it.Valid(); it.Next() {
			// See if we need to skip the prefix.
			if len(cd.dropPrefix) > 0 && bytes.HasPrefix(it.Key(), cd.dropPrefix) {
				numSkips++
				updateStats(it.Value())
				continue
			}

			// See if we need to skip this key.
			if len(skipKey) > 0 {
				if y.SameKey(it.Key(), skipKey) {
					numSkips++
					updateStats(it.Value())
					continue
				} else {
					skipKey = skipKey[:0]
				}
			}

			if !y.SameKey(it.Key(), lastKey) {
				if builder.ReachedCapacity(s.kv.opt.MaxTableSize) {
					// 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.
					break
				}
				lastKey = y.SafeCopy(lastKey, it.Key())
				numVersions = 0
			}

			vs := it.Value()
			version := y.ParseTs(it.Key())
			if version <= discardTs {
				// Keep track of the number of versions encountered for this key. Only consider the
				// versions which are below the minReadTs, otherwise, we might end up discarding the
				// only valid version for a running transaction.
				numVersions++
				lastValidVersion := vs.Meta&bitDiscardEarlierVersions > 0
				if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) ||
					numVersions > s.kv.opt.NumVersionsToKeep ||
					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.
					skipKey = y.SafeCopy(skipKey, it.Key())

					if lastValidVersion {
						// Add this key. We have set skipKey, so the following key versions
						// would be skipped.
					} else if hasOverlap {
						// If this key range has overlap with lower levels, then keep the deletion
						// marker with the latest version, discarding the rest. We have set skipKey,
						// so the following key versions would be skipped.
					} else {
						// If no overlap, we can skip all the versions, by continuing here.
						numSkips++
						updateStats(vs)
						continue // Skip adding this key.
					}
				}
			}
			numKeys++
			y.Check(builder.Add(it.Key(), it.Value()))
		}
		// It was true that it.Valid() at least once in the loop above, which means we
		// called Add() at least once, and builder is not Empty().
		s.kv.opt.Debugf("LOG Compact. Added %d keys. Skipped %d keys. Iteration took: %v",
			numKeys, numSkips, time.Since(timeStart))
		if !builder.Empty() {
			numBuilds++
			fileID := s.reserveFileID()
			go func(builder *table.Builder) {
				defer builder.Close()

				fd, err := y.CreateSyncedFile(table.NewFilename(fileID, s.kv.opt.Dir), true)
				if err != nil {
					resultCh <- newTableResult{nil, errors.Wrapf(err, "While opening new table: %d", fileID)}
					return
				}

				if _, err := fd.Write(builder.Finish()); err != nil {
					resultCh <- newTableResult{nil, errors.Wrapf(err, "Unable to write to file: %d", fileID)}
					return
				}

				tbl, err := table.OpenTable(fd, s.kv.opt.TableLoadingMode, nil)
				// decrRef is added below.
				resultCh <- newTableResult{tbl, errors.Wrapf(err, "Unable to open table: %q", fd.Name())}
			}(builder)
		}
	}

	newTables := make([]*table.Table, 0, 20)
	// Wait for all table builders to finish.
	var firstErr error
	for x := 0; x < numBuilds; x++ {
		res := <-resultCh
		newTables = append(newTables, res.table)
		if firstErr == nil {
			firstErr = res.err
		}
	}

	if firstErr == nil {
		// Ensure created files' directory entries are visible.  We don't mind the extra latency
		// from not doing this ASAP after all file creation has finished because this is a
		// background operation.
		firstErr = syncDir(s.kv.opt.Dir)
	}

	if firstErr != nil {
		// An error happened.  Delete all the newly created table files (by calling DecrRef
		// -- we're the only holders of a ref).
		for j := 0; j < numBuilds; j++ {
			if newTables[j] != nil {
				newTables[j].DecrRef()
			}
		}
		errorReturn := errors.Wrapf(firstErr, "While running compaction for: %+v", cd)
		return nil, nil, errorReturn
	}

	sort.Slice(newTables, func(i, j int) bool {
		return y.CompareKeys(newTables[i].Biggest(), newTables[j].Biggest()) < 0
	})
	s.kv.vlog.updateDiscardStats(discardStats)
	s.kv.opt.Debugf("Discard stats: %v", discardStats)
	return newTables, func() error { return decrRefs(newTables) }, nil
}
func (s *levelsController) compactBuildTables(
	lev int, cd compactDef) ([]*table.Table, func() error, error) {
	topTables := cd.top
	botTables := cd.bot

	var hasOverlap bool
	{
		kr := getKeyRange(cd.top)
		for i, lh := range s.levels {
			if i <= lev { // Skip upper levels.
				continue
			}
			lh.RLock()
			left, right := lh.overlappingTables(levelHandlerRLocked{}, kr)
			lh.RUnlock()
			if right-left > 0 {
				hasOverlap = true
				break
			}
		}
	}

	// Try to collect stats so that we can inform value log about GC. That would help us find which
	// value log file should be GCed.
	discardStats := make(map[uint32]int64)
	updateStats := func(vs y.ValueStruct) {
		if vs.Meta&bitValuePointer > 0 {
			var vp valuePointer
			vp.Decode(vs.Value)
			discardStats[vp.Fid] += int64(vp.Len)
		}
	}

	// Create iterators across all the tables involved first.
	var iters []y.Iterator
	if lev == 0 {
		iters = appendIteratorsReversed(iters, topTables, false)
	} else if len(topTables) > 0 {
		y.AssertTrue(len(topTables) == 1)
		iters = []y.Iterator{topTables[0].NewIterator(false)}
	}

	// Next level has level>=1 and we can use ConcatIterator as key ranges do not overlap.
	var valid []*table.Table
	for _, table := range botTables {
		if len(cd.dropPrefix) > 0 &&
			bytes.HasPrefix(table.Smallest(), cd.dropPrefix) &&
			bytes.HasPrefix(table.Biggest(), cd.dropPrefix) {
			// All the keys in this table have the dropPrefix. So, this table does not need to be
			// in the iterator and can be dropped immediately.
			continue
		}
		valid = append(valid, table)
	}
	iters = append(iters, table.NewConcatIterator(valid, false))
	it := y.NewMergeIterator(iters, false)
	defer it.Close() // Important to close the iterator to do ref counting.

	it.Rewind()

	// Pick a discard ts, so we can discard versions below this ts. We should
	// never discard any versions starting from above this timestamp, because
	// that would affect the snapshot view guarantee provided by transactions.
	discardTs := s.kv.orc.discardAtOrBelow()

	// Start generating new tables.
	type newTableResult struct {
		table *table.Table
		err   error
	}
	resultCh := make(chan newTableResult)
	var numBuilds, numVersions int
	var lastKey, skipKey []byte
	for it.Valid() {
		timeStart := time.Now()
		builder := table.NewTableBuilder()
		var numKeys, numSkips uint64
		for ; it.Valid(); it.Next() {
			// See if we need to skip the prefix.
			if len(cd.dropPrefix) > 0 && bytes.HasPrefix(it.Key(), cd.dropPrefix) {
				numSkips++
				updateStats(it.Value())
				continue
			}

			// See if we need to skip this key.
			if len(skipKey) > 0 {
				if y.SameKey(it.Key(), skipKey) {
					numSkips++
					updateStats(it.Value())
					continue
				} else {
					skipKey = skipKey[:0]
				}
			}

			if !y.SameKey(it.Key(), lastKey) {
				if builder.ReachedCapacity(s.kv.opt.MaxTableSize) {
					// 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.
					break
				}
				lastKey = y.SafeCopy(lastKey, it.Key())
				numVersions = 0
			}

			vs := it.Value()
			version := y.ParseTs(it.Key())
			if version <= discardTs {
				// Keep track of the number of versions encountered for this key. Only consider the
				// versions which are below the minReadTs, otherwise, we might end up discarding the
				// only valid version for a running transaction.
				numVersions++
				lastValidVersion := vs.Meta&bitDiscardEarlierVersions > 0
				if isDeletedOrExpired(vs.Meta, vs.ExpiresAt) ||
					numVersions > s.kv.opt.NumVersionsToKeep ||
					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.
					skipKey = y.SafeCopy(skipKey, it.Key())

					if lastValidVersion {
						// Add this key. We have set skipKey, so the following key versions
						// would be skipped.
					} else if hasOverlap {
						// If this key range has overlap with lower levels, then keep the deletion
						// marker with the latest version, discarding the rest. We have set skipKey,
						// so the following key versions would be skipped.
					} else {
						// If no overlap, we can skip all the versions, by continuing here.
						numSkips++
						updateStats(vs)
						continue // Skip adding this key.
					}
				}
			}
			numKeys++
			y.Check(builder.Add(it.Key(), it.Value()))
		}
		// It was true that it.Valid() at least once in the loop above, which means we
		// called Add() at least once, and builder is not Empty().
		s.kv.opt.Debugf("LOG Compact. Added %d keys. Skipped %d keys. Iteration took: %v",
			numKeys, numSkips, time.Since(timeStart))
		if !builder.Empty() {
			numBuilds++
			fileID := s.reserveFileID()
			go func(builder *table.Builder) {
				defer builder.Close()

				fd, err := y.CreateSyncedFile(table.NewFilename(fileID, s.kv.opt.Dir), true)
				if err != nil {
					resultCh <- newTableResult{nil, errors.Wrapf(err, "While opening new table: %d", fileID)}
					return
				}

				if _, err := fd.Write(builder.Finish()); err != nil {
					resultCh <- newTableResult{nil, errors.Wrapf(err, "Unable to write to file: %d", fileID)}
					return
				}

				tbl, err := table.OpenTable(fd, s.kv.opt.TableLoadingMode, nil)
				// decrRef is added below.
				resultCh <- newTableResult{tbl, errors.Wrapf(err, "Unable to open table: %q", fd.Name())}
			}(builder)
		}
	}

	newTables := make([]*table.Table, 0, 20)
	// Wait for all table builders to finish.
	var firstErr error
	for x := 0; x < numBuilds; x++ {
		res := <-resultCh
		newTables = append(newTables, res.table)
		if firstErr == nil {
			firstErr = res.err
		}
	}

	if firstErr == nil {
		// Ensure created files' directory entries are visible.  We don't mind the extra latency
		// from not doing this ASAP after all file creation has finished because this is a
		// background operation.
		firstErr = syncDir(s.kv.opt.Dir)
	}

	if firstErr != nil {
		// An error happened.  Delete all the newly created table files (by calling DecrRef
		// -- we're the only holders of a ref).
		for j := 0; j < numBuilds; j++ {
			if newTables[j] != nil {
				newTables[j].DecrRef()
			}
		}
		errorReturn := errors.Wrapf(firstErr, "While running compaction for: %+v", cd)
		return nil, nil, errorReturn
	}

	sort.Slice(newTables, func(i, j int) bool {
		return y.CompareKeys(newTables[i].Biggest(), newTables[j].Biggest()) < 0
	})
	s.kv.vlog.updateDiscardStats(discardStats)
	s.kv.opt.Debugf("Discard stats: %v", discardStats)
	return newTables, func() error { return decrRefs(newTables) }, nil
}
// compactBuildTables merge topTables and botTables to form a list of new tables.