level_handler.go - external/github.com/dgraph-io/badger - Git at Google

 /*
  * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
  * SPDX-License-Identifier: Apache-2.0
  */

 package badger

 import (
 	"fmt"
 	"sort"
 	"sync"

 	"github.com/dgraph-io/badger/v4/table"
 	"github.com/dgraph-io/badger/v4/y"
 )

 type levelHandler struct {
 	// Guards tables, totalSize.
 	sync.RWMutex

 	// 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
 	totalStaleSize int64

 	// l0stall is a condition variable used to signal the (single) flush goroutine
 	// waiting in addLevel0Table that the L0 table count has dropped below the stall
 	// threshold (or that the DB is closing). It is only used for level 0 and its
 	// Locker is this handler's write lock (see RWMutex above), so the stall
 	// predicate (len(tables) >= NumLevelZeroTablesStall) is always evaluated and
 	// the wait is entered atomically under the same lock that compaction uses to
 	// remove L0 tables. This eliminates lost wakeups without a second mutex.
 	l0stall *sync.Cond

 	// The following are initialized once and const.
 	level    int
 	strLevel string
 	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()
 	return s.totalSize
 }

 // initTables replaces s.tables with given tables. This is done during loading.
 func (s *levelHandler) initTables(tables []*table.Table) {
 	s.Lock()
 	defer s.Unlock()

 	s.tables = tables
 	s.totalSize = 0
 	s.totalStaleSize = 0
 	for _, t := range tables {
 		s.addSize(t)
 	}

 	if s.level == 0 {
 		// Key range will overlap. Just sort by fileID in ascending order
 		// because newer tables are at the end of level 0.
 		sort.Slice(s.tables, func(i, j int) bool {
 			return s.tables[i].ID() < s.tables[j].ID()
 		})
 	} else {
 		// Sort tables by keys.
 		sort.Slice(s.tables, func(i, j int) bool {
 			return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
 		})
 	}
 }

 // deleteTables remove tables idx0, ..., idx1-1.
 func (s *levelHandler) deleteTables(toDel []*table.Table) error {
 	s.Lock() // s.Unlock() below

 	toDelMap := make(map[uint64]struct{})
 	for _, t := range toDel {
 		toDelMap[t.ID()] = struct{}{}
 	}

 	// Make a copy as iterators might be keeping a slice of tables.
 	var newTables []*table.Table
 	for _, t := range s.tables {
 		_, found := toDelMap[t.ID()]
 		if !found {
 			newTables = append(newTables, t)
 			continue
 		}
 		s.subtractSize(t)
 	}
 	s.tables = newTables

 	s.Unlock() // Unlock s _before_ we DecrRef our tables, which can be slow.

 	return decrRefs(toDel)
 }

 // 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(toDel, toAdd []*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.)
 	s.Lock() // We s.Unlock() below.

 	toDelMap := make(map[uint64]struct{})
 	for _, t := range toDel {
 		toDelMap[t.ID()] = struct{}{}
 	}
 	var newTables []*table.Table
 	for _, t := range s.tables {
 		_, found := toDelMap[t.ID()]
 		if !found {
 			newTables = append(newTables, t)
 			continue
 		}
 		s.subtractSize(t)
 	}

 	// Increase totalSize first.
 	for _, t := range toAdd {
 		s.addSize(t)
 		t.IncrRef()
 		newTables = append(newTables, t)
 	}

 	// Assign tables.
 	s.tables = newTables
 	sort.Slice(s.tables, func(i, j int) bool {
 		return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
 	})
 	s.Unlock() // s.Unlock before we DecrRef tables -- that can be slow.
 	return decrRefs(toDel)
 }

 // addTable adds toAdd table to levelHandler. Normally when we add tables to levelHandler, we sort
 // tables based on table.Smallest. This is required for correctness of the system. But in case of
 // stream writer this can be avoided. We can just add tables to levelHandler's table list
 // and after all addTable calls, we can sort table list(check sortTable method).
 // NOTE: levelHandler.sortTables() should be called after call addTable calls are done.
 func (s *levelHandler) addTable(t *table.Table) {
 	s.Lock()
 	defer s.Unlock()

 	s.addSize(t) // Increase totalSize first.
 	t.IncrRef()
 	s.tables = append(s.tables, t)
 }

 // sortTables sorts tables of levelHandler based on table.Smallest.
 // Normally it should be called after all addTable calls.
 func (s *levelHandler) sortTables() {
 	s.Lock()
 	defer s.Unlock()

 	sort.Slice(s.tables, func(i, j int) bool {
 		return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
 	})
 }

 func decrRefs(tables []*table.Table) error {
 	for _, table := range tables {
 		if err := table.DecrRef(); err != nil {
 			return err
 		}
 	}
 	return nil
 }

 func newLevelHandler(db *DB, level int) *levelHandler {
 	s := &levelHandler{
 		level:    level,
 		strLevel: fmt.Sprintf("l%d", level),
 		db:       db,
 	}
 	if level == 0 {
 		// The cond's Locker is the handler's write lock (RWMutex.Lock/Unlock).
 		s.l0stall = sync.NewCond(&s.RWMutex)
 	}
 	return s
 }

 // signalL0Drained wakes any flush goroutine waiting in addLevel0Table. It must be
 // called whenever the number of L0 tables may have decreased (e.g. after an
 // L0->Lbase or L0->L0 compaction removes tables) or when the DB is closing. The
 // waiter re-checks the stall predicate under the lock in a loop, so it is safe to
 // Broadcast without holding the lock and without the count having actually changed
 // (spurious broadcasts are harmless).
 func (s *levelHandler) signalL0Drained() {
 	if s.l0stall != nil {
 		s.l0stall.Broadcast()
 	}
 }

 // numTablesLocked returns the number of tables. Caller must hold s.Lock() (write
 // lock), which is the same lock used as the l0stall cond's Locker.
 func (s *levelHandler) numTablesLocked() int {
 	return len(s.tables)
 }

 // addLevel0TableLocked appends t to L0 unconditionally. Caller must hold s.Lock()
 // and must have asserted s.level == 0. This is the lock-held variant of the body
 // of tryAddLevel0Table, used by addLevel0Table which already holds the lock to
 // wait on the stall cond.
 func (s *levelHandler) addLevel0TableLocked(t *table.Table) {
 	s.tables = append(s.tables, t)
 	t.IncrRef()
 	s.addSize(t)
 }

 // tryAddLevel0Table returns true if ok and no stalling.
 func (s *levelHandler) tryAddLevel0Table(t *table.Table) bool {
 	y.AssertTrue(s.level == 0)
 	// Need lock as we may be deleting the first table during a level 0 compaction.
 	s.Lock()
 	defer s.Unlock()
 	// Stall (by returning false) if we are above the specified stall setting for L0.
 	if len(s.tables) >= s.db.opt.NumLevelZeroTablesStall {
 		return false
 	}

 	s.tables = append(s.tables, t)
 	t.IncrRef()
 	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()
 	return len(s.tables)
 }

 func (s *levelHandler) close() error {
 	s.RLock()
 	defer s.RUnlock()
 	var err error
 	for _, t := range s.tables {
 		if closeErr := t.Close(-1); closeErr != nil && err == nil {
 			err = closeErr
 		}
 	}
 	return y.Wrap(err, "levelHandler.close")
 }

 // getTableForKey acquires a read-lock to access s.tables. It returns a list of tableHandlers.
 func (s *levelHandler) getTableForKey(key []byte) ([]*table.Table, func() error) {
 	s.RLock()
 	defer s.RUnlock()

 	if s.level == 0 {
 		// For level 0, we need to check every table. Remember to make a copy as s.tables may change
 		// once we exit this function, and we don't want to lock s.tables while seeking in tables.
 		// CAUTION: Reverse the tables.
 		out := make([]*table.Table, 0, len(s.tables))
 		for i := len(s.tables) - 1; i >= 0; i-- {
 			out = append(out, s.tables[i])
 			s.tables[i].IncrRef()
 		}
 		return out, func() error {
 			for _, t := range out {
 				if err := t.DecrRef(); err != nil {
 					return err
 				}
 			}
 			return nil
 		}
 	}
 	// For level >= 1, we can do a binary search as key range does not overlap.
 	idx := sort.Search(len(s.tables), func(i int) bool {
 		return y.CompareKeys(s.tables[i].Biggest(), key) >= 0
 	})
 	if idx >= len(s.tables) {
 		// Given key is strictly > than every element we have.
 		return nil, func() error { return nil }
 	}
 	tbl := s.tables[idx]
 	tbl.IncrRef()
 	return []*table.Table{tbl}, tbl.DecrRef
 }

 // checkInsideIteator checks if the key is present in the iterator or not. It updates maxVs if the value is
 // found.
 func (s *levelHandler) checkInsideIterator(key []byte, it *table.Iterator, maxVs *y.ValueStruct) {
 	y.NumLSMGetsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
 	it.Seek(key)
 	if !it.Valid() {
 		return
 	}
 	if !y.SameKey(key, it.Key()) {
 		return
 	}
 	if version := y.ParseTs(it.Key()); maxVs.Version < version {
 		*maxVs = it.ValueCopy()
 		maxVs.Version = version
 	}
 }

 func (s *levelHandler) getBatch(keys [][]byte, keysRead []bool) ([]y.ValueStruct, error) {
 	// Find the table for which the key is in, and then seek it. There's a good chance that they next key to be
 	// searched, is in the same table as well. Hence, we store the iterators found. If we don't find the results
 	// in the given table, we would need to search again. Worst case, this function could be a little worse than
 	// getting the n keys, in n different get calls.
 	createIteratorsForEachTable := func(key []byte) (y.ValueStruct, func() error, []*table.Iterator) {
 		tables, decr := s.getTableForKey(key)
 		keyNoTs := y.ParseKey(key)
 		itrs := make([]*table.Iterator, 0)

 		hash := y.Hash(keyNoTs)
 		var maxVs y.ValueStruct
 		for _, th := range tables {
 			// At level 0 tables overlap, and the iterator set built here is REUSED for
 			// the other keys of the batch. Bloom-skipping a table that merely lacks THIS
 			// key would hide it from later keys whose newest version lives there — a
 			// reused set must cover every L0 table. Levels >= 1 are non-overlapping (a
 			// key's versions live in exactly one table per level), so skipping is safe:
 			// a later key not present in the reused table falls back via the
 			// not-found path below.
 			if s.level != 0 && th.DoesNotHave(hash) {
 				y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
 				continue
 			}

 			it := th.NewIterator(0)
 			itrs = append(itrs, it)
 			s.checkInsideIterator(key, it, &maxVs)
 		}

 		return maxVs, decr, itrs
 	}

 	// Use old results from createIteratorsForEachTable and find in those tables.
 	findInIterators := func(key []byte, itrs []*table.Iterator) y.ValueStruct {
 		var maxVs y.ValueStruct
 		for _, it := range itrs {
 			s.checkInsideIterator(key, it, &maxVs)
 		}
 		return maxVs
 	}

 	results := make([]y.ValueStruct, len(keys))

 	decr := func() error { return nil }
 	var itrs []*table.Iterator

 	close_iters := func() {
 		for _, itr := range itrs {
 			itr.Close()
 		}
 	}

 	defer close_iters()

 	for i := 0; i < len(keys); i++ {
 		if keysRead[i] {
 			continue
 		}
 		// If there are no iterators present, create new iterators. Release the table
 		// references of any previous create first — a create can return zero iterators
 		// (bloom missed in every table) while still holding table refs through decr.
 		if len(itrs) == 0 {
 			if err := decr(); err != nil {
 				return nil, err
 			}
 			results[i], decr, itrs = createIteratorsForEachTable(keys[i])
 		} else {
 			results[i] = findInIterators(keys[i], itrs)
 			// Version == 0 means no version of this key was seen in the reused tables
 			// (an empty Value alone is not "not found": tombstones and empty writes are
 			// real answers). The key may live in other tables: close, release, recreate.
 			if results[i].Version == 0 {
 				close_iters()
 				itrs = itrs[:0]
 				if err := decr(); err != nil {
 					return nil, err
 				}
 				results[i], decr, itrs = createIteratorsForEachTable(keys[i])
 			}
 		}
 	}

 	return results, decr()
 }

 // get returns value for a given key or the key after that. If not found, return nil.
 func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
 	tables, decr := s.getTableForKey(key)
 	keyNoTs := y.ParseKey(key)

 	hash := y.Hash(keyNoTs)
 	var maxVs y.ValueStruct
 	for _, th := range tables {
 		if th.DoesNotHave(hash) {
 			y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
 			continue
 		}

 		it := th.NewIterator(0)
 		defer it.Close()

 		s.checkInsideIterator(key, it, &maxVs)
 	}
 	return maxVs, decr()
 }

 // appendIterators appends iterators to an array of iterators, for merging.
 // Note: This obtains references for the table handlers. Remember to close these iterators.
 func (s *levelHandler) appendIterators(iters []y.Iterator, opt *IteratorOptions) []y.Iterator {
 	s.RLock()
 	defer s.RUnlock()

 	var topt int
 	if opt.Reverse {
 		topt = table.REVERSED
 	}
 	if s.level == 0 {
 		// Remember to add in reverse order!
 		// The newer table at the end of s.tables should be added first as it takes precedence.
 		// Level 0 tables are not in key sorted order, so we need to consider them one by one.
 		var out []*table.Table
 		for _, t := range s.tables {
 			if opt.pickTable(t) {
 				out = append(out, t)
 			}
 		}
 		return appendIteratorsReversed(iters, out, topt)
 	}

 	tables := opt.pickTables(s.tables)
 	if len(tables) == 0 {
 		return iters
 	}
 	return append(iters, table.NewConcatIterator(tables, topt))
 }

 type levelHandlerRLocked struct{}

 // overlappingTables returns the tables that intersect with key range. Returns a half-interval.
 // This function should already have acquired a read lock, and this is so important the caller must
 // pass an empty parameter declaring such.
 func (s *levelHandler) overlappingTables(_ levelHandlerRLocked, kr keyRange) (int, int) {
 	if len(kr.left) == 0 || len(kr.right) == 0 {
 		return 0, 0
 	}
 	left := sort.Search(len(s.tables), func(i int) bool {
 		return y.CompareKeys(kr.left, s.tables[i].Biggest()) <= 0
 	})
 	right := sort.Search(len(s.tables), func(i int) bool {
 		return y.CompareKeys(kr.right, s.tables[i].Smallest()) < 0
 	})
 	return left, right
 }
	/*
	* SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
	* SPDX-License-Identifier: Apache-2.0
	*/

	package badger

	import (
	"fmt"
	"sort"
	"sync"

	"github.com/dgraph-io/badger/v4/table"
	"github.com/dgraph-io/badger/v4/y"
	)

	type levelHandler struct {
	// Guards tables, totalSize.
	sync.RWMutex

	// 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
	totalStaleSize int64

	// l0stall is a condition variable used to signal the (single) flush goroutine
	// waiting in addLevel0Table that the L0 table count has dropped below the stall
	// threshold (or that the DB is closing). It is only used for level 0 and its
	// Locker is this handler's write lock (see RWMutex above), so the stall
	// predicate (len(tables) >= NumLevelZeroTablesStall) is always evaluated and
	// the wait is entered atomically under the same lock that compaction uses to
	// remove L0 tables. This eliminates lost wakeups without a second mutex.
	l0stall *sync.Cond

	// The following are initialized once and const.
	level int
	strLevel string
	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()
	return s.totalSize
	}

	// initTables replaces s.tables with given tables. This is done during loading.
	func (s levelHandler) initTables(tables []table.Table) {
	s.Lock()
	defer s.Unlock()

	s.tables = tables
	s.totalSize = 0
	s.totalStaleSize = 0
	for _, t := range tables {
	s.addSize(t)
	}

	if s.level == 0 {
	// Key range will overlap. Just sort by fileID in ascending order
	// because newer tables are at the end of level 0.
	sort.Slice(s.tables, func(i, j int) bool {
	return s.tables[i].ID() < s.tables[j].ID()
	})
	} else {
	// Sort tables by keys.
	sort.Slice(s.tables, func(i, j int) bool {
	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
	})
	}
	}

	// deleteTables remove tables idx0, ..., idx1-1.
	func (s levelHandler) deleteTables(toDel []table.Table) error {
	s.Lock() // s.Unlock() below

	toDelMap := make(map[uint64]struct{})
	for _, t := range toDel {
	toDelMap[t.ID()] = struct{}{}
	}

	// Make a copy as iterators might be keeping a slice of tables.
	var newTables []*table.Table
	for _, t := range s.tables {
	_, found := toDelMap[t.ID()]
	if !found {
	newTables = append(newTables, t)
	continue
	}
	s.subtractSize(t)
	}
	s.tables = newTables

	s.Unlock() // Unlock s _before_ we DecrRef our tables, which can be slow.

	return decrRefs(toDel)
	}

	// 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(toDel, toAdd []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.)
	s.Lock() // We s.Unlock() below.

	toDelMap := make(map[uint64]struct{})
	for _, t := range toDel {
	toDelMap[t.ID()] = struct{}{}
	}
	var newTables []*table.Table
	for _, t := range s.tables {
	_, found := toDelMap[t.ID()]
	if !found {
	newTables = append(newTables, t)
	continue
	}
	s.subtractSize(t)
	}

	// Increase totalSize first.
	for _, t := range toAdd {
	s.addSize(t)
	t.IncrRef()
	newTables = append(newTables, t)
	}

	// Assign tables.
	s.tables = newTables
	sort.Slice(s.tables, func(i, j int) bool {
	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
	})
	s.Unlock() // s.Unlock before we DecrRef tables -- that can be slow.
	return decrRefs(toDel)
	}

	// addTable adds toAdd table to levelHandler. Normally when we add tables to levelHandler, we sort
	// tables based on table.Smallest. This is required for correctness of the system. But in case of
	// stream writer this can be avoided. We can just add tables to levelHandler's table list
	// and after all addTable calls, we can sort table list(check sortTable method).
	// NOTE: levelHandler.sortTables() should be called after call addTable calls are done.
	func (s levelHandler) addTable(t table.Table) {
	s.Lock()
	defer s.Unlock()

	s.addSize(t) // Increase totalSize first.
	t.IncrRef()
	s.tables = append(s.tables, t)
	}

	// sortTables sorts tables of levelHandler based on table.Smallest.
	// Normally it should be called after all addTable calls.
	func (s *levelHandler) sortTables() {
	s.Lock()
	defer s.Unlock()

	sort.Slice(s.tables, func(i, j int) bool {
	return y.CompareKeys(s.tables[i].Smallest(), s.tables[j].Smallest()) < 0
	})
	}

	func decrRefs(tables []*table.Table) error {
	for _, table := range tables {
	if err := table.DecrRef(); err != nil {
	return err
	}
	}
	return nil
	}

	func newLevelHandler(db DB, level int) levelHandler {
	s := &levelHandler{
	level: level,
	strLevel: fmt.Sprintf("l%d", level),
	db: db,
	}
	if level == 0 {
	// The cond's Locker is the handler's write lock (RWMutex.Lock/Unlock).
	s.l0stall = sync.NewCond(&s.RWMutex)
	}
	return s
	}

	// signalL0Drained wakes any flush goroutine waiting in addLevel0Table. It must be
	// called whenever the number of L0 tables may have decreased (e.g. after an
	// L0->Lbase or L0->L0 compaction removes tables) or when the DB is closing. The
	// waiter re-checks the stall predicate under the lock in a loop, so it is safe to
	// Broadcast without holding the lock and without the count having actually changed
	// (spurious broadcasts are harmless).
	func (s *levelHandler) signalL0Drained() {
	if s.l0stall != nil {
	s.l0stall.Broadcast()
	}
	}

	// numTablesLocked returns the number of tables. Caller must hold s.Lock() (write
	// lock), which is the same lock used as the l0stall cond's Locker.
	func (s *levelHandler) numTablesLocked() int {
	return len(s.tables)
	}

	// addLevel0TableLocked appends t to L0 unconditionally. Caller must hold s.Lock()
	// and must have asserted s.level == 0. This is the lock-held variant of the body
	// of tryAddLevel0Table, used by addLevel0Table which already holds the lock to
	// wait on the stall cond.
	func (s levelHandler) addLevel0TableLocked(t table.Table) {
	s.tables = append(s.tables, t)
	t.IncrRef()
	s.addSize(t)
	}

	// tryAddLevel0Table returns true if ok and no stalling.
	func (s levelHandler) tryAddLevel0Table(t table.Table) bool {
	y.AssertTrue(s.level == 0)
	// Need lock as we may be deleting the first table during a level 0 compaction.
	s.Lock()
	defer s.Unlock()
	// Stall (by returning false) if we are above the specified stall setting for L0.
	if len(s.tables) >= s.db.opt.NumLevelZeroTablesStall {
	return false
	}

	s.tables = append(s.tables, t)
	t.IncrRef()
	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()
	return len(s.tables)
	}

	func (s *levelHandler) close() error {
	s.RLock()
	defer s.RUnlock()
	var err error
	for _, t := range s.tables {
	if closeErr := t.Close(-1); closeErr != nil && err == nil {
	err = closeErr
	}
	}
	return y.Wrap(err, "levelHandler.close")
	}

	// getTableForKey acquires a read-lock to access s.tables. It returns a list of tableHandlers.
	func (s levelHandler) getTableForKey(key []byte) ([]table.Table, func() error) {
	s.RLock()
	defer s.RUnlock()

	if s.level == 0 {
	// For level 0, we need to check every table. Remember to make a copy as s.tables may change
	// once we exit this function, and we don't want to lock s.tables while seeking in tables.
	// CAUTION: Reverse the tables.
	out := make([]*table.Table, 0, len(s.tables))
	for i := len(s.tables) - 1; i >= 0; i-- {
	out = append(out, s.tables[i])
	s.tables[i].IncrRef()
	}
	return out, func() error {
	for _, t := range out {
	if err := t.DecrRef(); err != nil {
	return err
	}
	}
	return nil
	}
	}
	// For level >= 1, we can do a binary search as key range does not overlap.
	idx := sort.Search(len(s.tables), func(i int) bool {
	return y.CompareKeys(s.tables[i].Biggest(), key) >= 0
	})
	if idx >= len(s.tables) {
	// Given key is strictly > than every element we have.
	return nil, func() error { return nil }
	}
	tbl := s.tables[idx]
	tbl.IncrRef()
	return []*table.Table{tbl}, tbl.DecrRef
	}

	// checkInsideIteator checks if the key is present in the iterator or not. It updates maxVs if the value is
	// found.
	func (s levelHandler) checkInsideIterator(key []byte, it table.Iterator, maxVs *y.ValueStruct) {
	y.NumLSMGetsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
	it.Seek(key)
	if !it.Valid() {
	return
	}
	if !y.SameKey(key, it.Key()) {
	return
	}
	if version := y.ParseTs(it.Key()); maxVs.Version < version {
	*maxVs = it.ValueCopy()
	maxVs.Version = version
	}
	}

	func (s *levelHandler) getBatch(keys [][]byte, keysRead []bool) ([]y.ValueStruct, error) {
	// Find the table for which the key is in, and then seek it. There's a good chance that they next key to be
	// searched, is in the same table as well. Hence, we store the iterators found. If we don't find the results
	// in the given table, we would need to search again. Worst case, this function could be a little worse than
	// getting the n keys, in n different get calls.
	createIteratorsForEachTable := func(key []byte) (y.ValueStruct, func() error, []*table.Iterator) {
	tables, decr := s.getTableForKey(key)
	keyNoTs := y.ParseKey(key)
	itrs := make([]*table.Iterator, 0)

	hash := y.Hash(keyNoTs)
	var maxVs y.ValueStruct
	for _, th := range tables {
	// At level 0 tables overlap, and the iterator set built here is REUSED for
	// the other keys of the batch. Bloom-skipping a table that merely lacks THIS
	// key would hide it from later keys whose newest version lives there — a
	// reused set must cover every L0 table. Levels >= 1 are non-overlapping (a
	// key's versions live in exactly one table per level), so skipping is safe:
	// a later key not present in the reused table falls back via the
	// not-found path below.
	if s.level != 0 && th.DoesNotHave(hash) {
	y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
	continue
	}

	it := th.NewIterator(0)
	itrs = append(itrs, it)
	s.checkInsideIterator(key, it, &maxVs)
	}

	return maxVs, decr, itrs
	}

	// Use old results from createIteratorsForEachTable and find in those tables.
	findInIterators := func(key []byte, itrs []*table.Iterator) y.ValueStruct {
	var maxVs y.ValueStruct
	for _, it := range itrs {
	s.checkInsideIterator(key, it, &maxVs)
	}
	return maxVs
	}

	results := make([]y.ValueStruct, len(keys))

	decr := func() error { return nil }
	var itrs []*table.Iterator

	close_iters := func() {
	for _, itr := range itrs {
	itr.Close()
	}
	}

	defer close_iters()

	for i := 0; i < len(keys); i++ {
	if keysRead[i] {
	continue
	}
	// If there are no iterators present, create new iterators. Release the table
	// references of any previous create first — a create can return zero iterators
	// (bloom missed in every table) while still holding table refs through decr.
	if len(itrs) == 0 {
	if err := decr(); err != nil {
	return nil, err
	}
	results[i], decr, itrs = createIteratorsForEachTable(keys[i])
	} else {
	results[i] = findInIterators(keys[i], itrs)
	// Version == 0 means no version of this key was seen in the reused tables
	// (an empty Value alone is not "not found": tombstones and empty writes are
	// real answers). The key may live in other tables: close, release, recreate.
	if results[i].Version == 0 {
	close_iters()
	itrs = itrs[:0]
	if err := decr(); err != nil {
	return nil, err
	}
	results[i], decr, itrs = createIteratorsForEachTable(keys[i])
	}
	}
	}

	return results, decr()
	}

	// get returns value for a given key or the key after that. If not found, return nil.
	func (s *levelHandler) get(key []byte) (y.ValueStruct, error) {
	tables, decr := s.getTableForKey(key)
	keyNoTs := y.ParseKey(key)

	hash := y.Hash(keyNoTs)
	var maxVs y.ValueStruct
	for _, th := range tables {
	if th.DoesNotHave(hash) {
	y.NumLSMBloomHitsAdd(s.db.opt.MetricsEnabled, s.strLevel, 1)
	continue
	}

	it := th.NewIterator(0)
	defer it.Close()

	s.checkInsideIterator(key, it, &maxVs)
	}
	return maxVs, decr()
	}

	// appendIterators appends iterators to an array of iterators, for merging.
	// Note: This obtains references for the table handlers. Remember to close these iterators.
	func (s levelHandler) appendIterators(iters []y.Iterator, opt IteratorOptions) []y.Iterator {
	s.RLock()
	defer s.RUnlock()

	var topt int
	if opt.Reverse {
	topt = table.REVERSED
	}
	if s.level == 0 {
	// Remember to add in reverse order!
	// The newer table at the end of s.tables should be added first as it takes precedence.
	// Level 0 tables are not in key sorted order, so we need to consider them one by one.
	var out []*table.Table
	for _, t := range s.tables {
	if opt.pickTable(t) {
	out = append(out, t)
	}
	}
	return appendIteratorsReversed(iters, out, topt)
	}

	tables := opt.pickTables(s.tables)
	if len(tables) == 0 {
	return iters
	}
	return append(iters, table.NewConcatIterator(tables, topt))
	}

	type levelHandlerRLocked struct{}

	// overlappingTables returns the tables that intersect with key range. Returns a half-interval.
	// This function should already have acquired a read lock, and this is so important the caller must
	// pass an empty parameter declaring such.
	func (s *levelHandler) overlappingTables(_ levelHandlerRLocked, kr keyRange) (int, int) {
	if len(kr.left) == 0 \|\| len(kr.right) == 0 {
	return 0, 0
	}
	left := sort.Search(len(s.tables), func(i int) bool {
	return y.CompareKeys(kr.left, s.tables[i].Biggest()) <= 0
	})
	right := sort.Search(len(s.tables), func(i int) bool {
	return y.CompareKeys(kr.right, s.tables[i].Smallest()) < 0
	})
	return left, right
	}