fix some more comments

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)
 }

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

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)
 	}