diff --git a/db.go b/db.go index 2b4723f..66dd570 100644 --- a/db.go +++ b/db.go @@ -733,11 +733,28 @@ func (db *DB) getMemTables() ([]*memTable, func()) { } } -// get returns the value in memtable or disk for given key. -// Note that value will include meta byte. -// -// getBatch would return the values of list of keys in order -// +func (db *DB) checkKeyInMemtables(tables []*memTable, key []byte, maxVs *y.ValueStruct, version uint64) bool { + y.NumGetsAdd(db.opt.MetricsEnabled, 1) + for i := 0; i < len(tables); i++ { + vs := tables[i].sl.Get(key) + y.NumMemtableGetsAdd(db.opt.MetricsEnabled, 1) + if vs.Meta == 0 && vs.Value == nil { + continue + } + // Found the required version of the key, mark as done, no need to process + // it further + if vs.Version == version { + y.NumGetsWithResultsAdd(db.opt.MetricsEnabled, 1) + *maxVs = vs + return true + } + if maxVs.Version < vs.Version { + *maxVs = vs + } + } + return false +} + // IMPORTANT: We should never write an entry with an older timestamp for the same key, We need to // maintain this invariant to search for the latest value of a key, or else we need to search in all // tables and find the max version among them. To maintain this invariant, we also need to ensure @@ -749,44 +766,32 @@ func (db *DB) getMemTables() ([]*memTable, func()) { // do that. For every get("fooX") call where X is the version, we will search // for "fooX" in all the levels of the LSM tree. This is expensive but it // removes the overhead of handling move keys completely. -func (db *DB) getBatch(keys [][]byte, keysRead []bool) ([]y.ValueStruct, error) { +// +// getBatch would return the values of list of keys in order +// Note that value will include meta byte. +func (db *DB) getBatch(keys [][]byte, keysRead []bool, version uint64) ([]y.ValueStruct, error) { if db.IsClosed() { - return []y.ValueStruct{}, ErrDBClosed + return nil, ErrDBClosed } tables, decr := db.getMemTables() // Lock should be released. defer decr() maxVs := make([]y.ValueStruct, len(keys)) - y.NumGetsAdd(db.opt.MetricsEnabled, 1) // For memtable, we need to check every memtable each time for j, key := range keys { if keysRead[j] { continue } - version := y.ParseTs(key) - for i := 0; i < len(tables); i++ { - vs := tables[i].sl.Get(key) - y.NumMemtableGetsAdd(db.opt.MetricsEnabled, 1) - if vs.Meta == 0 && vs.Value == nil { - continue - } - // Found the required version of the key, mark as done, no need to process - // it further - if vs.Version == version { - y.NumGetsWithResultsAdd(db.opt.MetricsEnabled, 1) - maxVs[j] = vs - keysRead[j] = true - break - } - if maxVs[j].Version < vs.Version { - maxVs[j] = vs - } + if db.checkKeyInMemtables(tables, key, &maxVs[j], version) { + keysRead[j] = true } } - return db.lc.getBatch(keys, maxVs, 0, keysRead) + return db.lc.getBatch(keys, maxVs, 0, keysRead, version) } +// get returns the value in memtable or disk for given key. +// Note that value will include meta byte. func (db *DB) get(key []byte) (y.ValueStruct, error) { if db.opt.useGetBatch { done := make([]bool, 1) @@ -805,22 +810,8 @@ func (db *DB) get(key []byte) (y.ValueStruct, error) { var maxVs y.ValueStruct version := y.ParseTs(key) - - y.NumGetsAdd(db.opt.MetricsEnabled, 1) - for i := 0; i < len(tables); i++ { - vs := tables[i].sl.Get(key) - y.NumMemtableGetsAdd(db.opt.MetricsEnabled, 1) - if vs.Meta == 0 && vs.Value == nil { - continue - } - // Found the required version of the key, return immediately. - if vs.Version == version { - y.NumGetsWithResultsAdd(db.opt.MetricsEnabled, 1) - return vs, nil - } - if maxVs.Version < vs.Version { - maxVs = vs - } + if db.checkKeyInMemtables(tables, key, &maxVs, version) { + return maxVs, nil } return db.lc.get(key, maxVs, 0) } diff --git a/levels.go b/levels.go index 9c6aa6a..7abef98 100644 --- a/levels.go +++ b/levels.go @@ -1594,8 +1594,8 @@ func (s *levelsController) close() error { return y.Wrap(err, "levelsController.Close") } -func (s *levelsController) getBatch(keys [][]byte, maxVs []y.ValueStruct, startLevel int, keysRead []bool) ( - []y.ValueStruct, error) { +func (s *levelsController) getBatch(keys [][]byte, + maxVs []y.ValueStruct, startLevel int, keysRead []bool, version uint64) ([]y.ValueStruct, error) { if s.kv.IsClosed() { return []y.ValueStruct{}, ErrDBClosed } @@ -1609,13 +1609,13 @@ func (s *levelsController) getBatch(keys [][]byte, maxVs []y.ValueStruct, startL if h.level < startLevel { continue } - vs, err := h.getBatch(keys, keysRead) // Calls h.RLock() and h.RUnlock(). + vs, err := h.getBatch(keys, keysRead, version) // Calls h.RLock() and h.RUnlock(). if err != nil { return []y.ValueStruct{}, y.Wrapf(err, "get keys: %q", keys) } for i, v := range vs { - // Done is only update by this function or one in db. levelhandler will + // keysRead is only update by this function or one in db. levelhandler will // not update done. No need to do anything is done is set. if keysRead[i] { continue @@ -1624,7 +1624,6 @@ func (s *levelsController) getBatch(keys [][]byte, maxVs []y.ValueStruct, startL continue } y.NumBytesReadsLSMAdd(s.kv.opt.MetricsEnabled, int64(len(v.Value))) - version := y.ParseTs(keys[i]) if v.Version == version { maxVs[i] = v keysRead[i] = true diff --git a/txn.go b/txn.go index e046e76..0f0d961 100644 --- a/txn.go +++ b/txn.go @@ -452,7 +452,8 @@ func (txn *Txn) GetBatch(keys [][]byte) (items []*Item, rerr error) { item := items[i] if e, has := txn.pendingWrites[string(key)]; has && bytes.Equal(key, e.Key) { if isDeletedOrExpired(e.meta, e.ExpiresAt) { - return nil, ErrKeyNotFound + items[i] = nil + continue } // Fulfill from cache. item.meta = e.meta @@ -479,7 +480,7 @@ func (txn *Txn) GetBatch(keys [][]byte) (items []*Item, rerr error) { for i, key := range keys { seeks[i] = y.KeyWithTs(key, txn.readTs) } - vss, err := txn.db.getBatch(seeks, done) + vss, err := txn.db.getBatch(seeks, done, txn.readTs) if err != nil { return nil, y.Wrapf(err, "DB::Get keys: %q", keys) }