*: implement IterReverse for tikvSnapshot and use desc scan to get latest N ddl history jobs.

meta/meta.go

@@ -651,6 +651,26 @@ func (m *Meta) GetAllHistoryDDLJobs() ([]*model.Job, error) {
 	return jobs, nil
 }
 
+// GetLastNHistoryDDLJobs gets latest N history ddl jobs.
+func (m *Meta) GetLastNHistoryDDLJobs(num int) ([]*model.Job, error) {
+	pairs, err := m.txn.HGetLastN(mDDLJobHistoryKey, num)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	var jobs []*model.Job
+	for _, pair := range pairs {
+		job := &model.Job{}
+		err = job.Decode(pair.Value)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		jobs = append(jobs, job)
+	}
+	sorter := &jobsSorter{jobs: jobs}
+	sort.Sort(sorter)
+	return jobs, nil
+}
+
 // jobsSorter implements the sort.Interface interface.
 type jobsSorter struct {
 	jobs []*model.Job

store/mockstore/mocktikv/rpc.go

@@ -238,10 +238,21 @@ func (h *rpcHandler) handleKvScan(req *kvrpcpb.ScanRequest) *kvrpcpb.ScanRespons
 		panic("KvScan: startKey not in region")
 	}
 	endKey := h.endKey
-	if len(req.EndKey) > 0 && (len(endKey) == 0 || bytes.Compare(req.EndKey, endKey) < 0) {
-		endKey = req.EndKey
+	var pairs []Pair
+	if !req.Reverse {
+		if len(req.EndKey) > 0 && (len(endKey) == 0 || bytes.Compare(req.EndKey, endKey) < 0) {
+			endKey = req.EndKey
+		}
+		pairs = h.mvccStore.Scan(req.GetStartKey(), endKey, int(req.GetLimit()), req.GetVersion(), h.isolationLevel)
+	} else {
+		// TiKV use range [end_key, start_key) for reverse scan.
+		// So the req.StartKey actually is the end_key.
+		if len(req.StartKey) > 0 && (len(endKey) == 0 || bytes.Compare(req.StartKey, endKey) < 0) {
+			endKey = req.StartKey
+		}
+		pairs = h.mvccStore.ReverseScan(req.EndKey, endKey, int(req.GetLimit()), req.GetVersion(), h.isolationLevel)
 	}
-	pairs := h.mvccStore.Scan(req.GetStartKey(), endKey, int(req.GetLimit()), req.GetVersion(), h.isolationLevel)
+
 	return &kvrpcpb.ScanResponse{
 		Pairs: convertToPbPairs(pairs),
 	}

store/tikv/scan.go

@@ -35,9 +35,13 @@ type Scanner struct {
 	nextStartKey []byte
 	endKey       []byte
 	eof          bool
+
+	// Use for reverse scan.
+	desc       bool
+	nextEndKey []byte
 }
 
-func newScanner(snapshot *tikvSnapshot, startKey []byte, endKey []byte, batchSize int) (*Scanner, error) {
+func newScanner(snapshot *tikvSnapshot, startKey []byte, endKey []byte, batchSize int, desc bool) (*Scanner, error) {
 	// It must be > 1. Otherwise scanner won't skipFirst.
 	if batchSize <= 1 {
 		batchSize = scanBatchSize
@@ -48,6 +52,8 @@ func newScanner(snapshot *tikvSnapshot, startKey []byte, endKey []byte, batchSiz
 		valid:        true,
 		nextStartKey: startKey,
 		endKey:       endKey,
+		desc:         desc,
+		nextEndKey:   endKey,
 	}
 	err := scanner.Next()
 	if kv.IsErrNotFound(err) {
@@ -83,14 +89,15 @@ func (s *Scanner) Next() error {
 	if !s.valid {
 		return errors.New("scanner iterator is invalid")
 	}
+	var err error
 	for {
 		s.idx++
 		if s.idx >= len(s.cache) {
 			if s.eof {
 				s.Close()
 				return nil
 			}
-			err := s.getData(bo)
+			err = s.getData(bo)
 			if err != nil {
 				s.Close()
 				return errors.Trace(err)
@@ -101,7 +108,8 @@ func (s *Scanner) Next() error {
 		}
 
 		current := s.cache[s.idx]
-		if len(s.endKey) > 0 && kv.Key(current.Key).Cmp(kv.Key(s.endKey)) >= 0 {
+		if (!s.desc && (len(s.endKey) > 0 && kv.Key(current.Key).Cmp(kv.Key(s.endKey)) >= 0)) ||
+			(s.desc && len(s.nextStartKey) > 0 && kv.Key(current.Key).Cmp(kv.Key(s.nextStartKey)) < 0) {
 			s.eof = true
 			s.Close()
 			return nil
@@ -147,18 +155,33 @@ func (s *Scanner) resolveCurrentLock(bo *Backoffer, current *pb.KvPair) error {
 func (s *Scanner) getData(bo *Backoffer) error {
 	logutil.Logger(context.Background()).Debug("txn getData",
 		zap.Binary("nextStartKey", s.nextStartKey),
+		zap.Binary("nextEndKey", s.nextEndKey),
+		zap.Bool("desc", s.desc),
 		zap.Uint64("txnStartTS", s.startTS()))
 	sender := NewRegionRequestSender(s.snapshot.store.regionCache, s.snapshot.store.client)
-
+	var locKey, reqEndKey, reqStartKey []byte
 	for {
-		loc, err := s.snapshot.store.regionCache.LocateKey(bo, s.nextStartKey)
+		if !s.desc {
+			locKey = s.nextStartKey
+		} else {
+			locKey = s.nextEndKey
+		}
+		loc, err := s.snapshot.store.regionCache.LocateKey(bo, locKey)
 		if err != nil {
 			return errors.Trace(err)
 		}
 
-		reqEndKey := s.endKey
-		if len(reqEndKey) > 0 && len(loc.EndKey) > 0 && bytes.Compare(loc.EndKey, reqEndKey) < 0 {
-			reqEndKey = loc.EndKey
+		if !s.desc {
+			reqEndKey = s.endKey
+			if len(reqEndKey) > 0 && len(loc.EndKey) > 0 && bytes.Compare(loc.EndKey, reqEndKey) < 0 {
+				reqEndKey = loc.EndKey
+			}
+		} else {
+			reqStartKey = s.nextStartKey
+			if len(reqStartKey) == 0 ||
+				(len(loc.StartKey) > 0 && bytes.Compare(loc.StartKey, reqStartKey) > 0) {
+				reqStartKey = loc.StartKey
+			}
 		}
 
 		req := &tikvrpc.Request{
@@ -175,6 +198,11 @@ func (s *Scanner) getData(bo *Backoffer) error {
 				NotFillCache: s.snapshot.notFillCache,
 			},
 		}
+		if s.desc {
+			req.Scan.StartKey = s.nextEndKey
+			req.Scan.EndKey = reqStartKey
+			req.Scan.Reverse = true
+		}
 		resp, err := sender.SendReq(bo, req, loc.Region, ReadTimeoutMedium)
 		if err != nil {
 			return errors.Trace(err)
@@ -218,8 +246,13 @@ func (s *Scanner) getData(bo *Backoffer) error {
 		if len(kvPairs) < s.batchSize {
 			// No more data in current Region. Next getData() starts
 			// from current Region's endKey.
-			s.nextStartKey = loc.EndKey
-			if len(loc.EndKey) == 0 || (len(s.endKey) > 0 && kv.Key(s.nextStartKey).Cmp(kv.Key(s.endKey)) >= 0) {
+			if !s.desc {
+				s.nextStartKey = loc.EndKey
+			} else {
+				s.nextEndKey = reqStartKey
+			}
+			if (!s.desc && (len(loc.EndKey) == 0 || (len(s.endKey) > 0 && kv.Key(s.nextStartKey).Cmp(kv.Key(s.endKey)) >= 0))) ||
+				(s.desc && (len(loc.StartKey) == 0 || (len(s.nextStartKey) > 0 && kv.Key(s.nextStartKey).Cmp(kv.Key(s.nextEndKey)) >= 0))) {
 				// Current Region is the last one.
 				s.eof = true
 			}
@@ -230,7 +263,11 @@ func (s *Scanner) getData(bo *Backoffer) error {
 		// may get an empty response if the Region in fact does not have
 		// more data.
 		lastKey := kvPairs[len(kvPairs)-1].GetKey()
-		s.nextStartKey = kv.Key(lastKey).Next()
+		if !s.desc {
+			s.nextStartKey = kv.Key(lastKey).Next()
+		} else {
+			s.nextEndKey = kv.Key(lastKey)
+		}
 		return nil
 	}
 }

store/tikv/scan_mock_test.go

@@ -42,19 +42,52 @@ func (s *testScanMockSuite) TestScanMultipleRegions(c *C) {
 	txn, err = store.Begin()
 	c.Assert(err, IsNil)
 	snapshot := newTiKVSnapshot(store, kv.Version{Ver: txn.StartTS()})
-	scanner, err := newScanner(snapshot, []byte("a"), nil, 10)
+	scanner, err := newScanner(snapshot, []byte("a"), nil, 10, false)
 	c.Assert(err, IsNil)
 	for ch := byte('a'); ch <= byte('z'); ch++ {
 		c.Assert([]byte{ch}, BytesEquals, []byte(scanner.Key()))
 		c.Assert(scanner.Next(), IsNil)
 	}
 	c.Assert(scanner.Valid(), IsFalse)
 
-	scanner, err = newScanner(snapshot, []byte("a"), []byte("i"), 10)
+	scanner, err = newScanner(snapshot, []byte("a"), []byte("i"), 10, false)
 	c.Assert(err, IsNil)
 	for ch := byte('a'); ch <= byte('h'); ch++ {
 		c.Assert([]byte{ch}, BytesEquals, []byte(scanner.Key()))
 		c.Assert(scanner.Next(), IsNil)
 	}
 	c.Assert(scanner.Valid(), IsFalse)
 }
+
+func (s *testScanMockSuite) TestDescScan(c *C) {
+	store := NewTestStore(c).(*tikvStore)
+	defer store.Close()
+
+	txn, err := store.Begin()
+	c.Assert(err, IsNil)
+	for ch := byte('a'); ch <= byte('z'); ch++ {
+		err = txn.Set([]byte{ch}, []byte{ch})
+		c.Assert(err, IsNil)
+	}
+	err = txn.Commit(context.Background())
+	c.Assert(err, IsNil)
+
+	txn, err = store.Begin()
+	c.Assert(err, IsNil)
+	snapshot := newTiKVSnapshot(store, kv.Version{Ver: txn.StartTS()})
+	scanner, err := newScanner(snapshot, []byte("a"), nil, 10, true)
+	c.Assert(err, IsNil)
+	for ch := byte('z'); ch >= byte('a'); ch-- {
+		c.Assert(string([]byte{ch}), Equals, string([]byte(scanner.Key())))
+		c.Assert(scanner.Next(), IsNil)
+	}
+	c.Assert(scanner.Valid(), IsFalse)
+
+	scanner, err = newScanner(snapshot, []byte("a"), []byte("i"), 10, true)
+	c.Assert(err, IsNil)
+	for ch := byte('h'); ch >= byte('a'); ch-- {
+		c.Assert(string([]byte{ch}), Equals, string([]byte(scanner.Key())))
+		c.Assert(scanner.Next(), IsNil)
+	}
+	c.Assert(scanner.Valid(), IsFalse)
+}

store/tikv/snapshot.go

@@ -289,13 +289,14 @@ func (s *tikvSnapshot) get(bo *Backoffer, k kv.Key) ([]byte, error) {
 
 // Iter return a list of key-value pair after `k`.
 func (s *tikvSnapshot) Iter(k kv.Key, upperBound kv.Key) (kv.Iterator, error) {
-	scanner, err := newScanner(s, k, upperBound, scanBatchSize)
+	scanner, err := newScanner(s, k, upperBound, scanBatchSize, false)
 	return scanner, errors.Trace(err)
 }
 
 // IterReverse creates a reversed Iterator positioned on the first entry which key is less than k.
-func (s *tikvSnapshot) IterReverse(k kv.Key) (kv.Iterator, error) {
-	return nil, kv.ErrNotImplemented
+func (s *tikvSnapshot) IterReverse(endKey kv.Key) (kv.Iterator, error) {
+	scanner, err := newScanner(s, nil, endKey, scanBatchSize, true)
+	return scanner, errors.Trace(err)
 }
 
 func extractLockFromKeyErr(keyErr *pb.KeyError) (*Lock, error) {

structure/hash.go

@@ -216,6 +216,23 @@ func (t *TxStructure) HGetAll(key []byte) ([]HashPair, error) {
 	return res, errors.Trace(err)
 }
 
+// HGetLastN gets latest N fields and values in hash.
+func (t *TxStructure) HGetLastN(key []byte, num int) ([]HashPair, error) {
+	res := make([]HashPair, 0, num)
+	err := t.iterReverseHash(key, func(field []byte, value []byte) (bool, error) {
+		pair := HashPair{
+			Field: append([]byte{}, field...),
+			Value: append([]byte{}, value...),
+		}
+		res = append(res, pair)
+		if len(res) >= num {
+			return false, nil
+		}
+		return true, nil
+	})
+	return res, errors.Trace(err)
+}
+
 // HClear removes the hash value of the key.
 func (t *TxStructure) HClear(key []byte) error {
 	metaKey := t.encodeHashMetaKey(key)
@@ -268,6 +285,37 @@ func (t *TxStructure) iterateHash(key []byte, fn func(k []byte, v []byte) error)
 	return nil
 }
 
+func (t *TxStructure) iterReverseHash(key []byte, fn func(k []byte, v []byte) (bool, error)) error {
+	dataPrefix := t.hashDataKeyPrefix(key)
+	it, err := t.reader.IterReverse(dataPrefix.PrefixNext())
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	var field []byte
+	for it.Valid() {
+		if !it.Key().HasPrefix(dataPrefix) {
+			break
+		}
+
+		_, field, err = t.decodeHashDataKey(it.Key())
+		if err != nil {
+			return errors.Trace(err)
+		}
+
+		more, err := fn(field, it.Value())
+		if !more || err != nil {
+			return errors.Trace(err)
+		}
+
+		err = it.Next()
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return nil
+}
+
 func (t *TxStructure) loadHashMeta(metaKey []byte) (hashMeta, error) {
 	v, err := t.reader.Get(metaKey)
 	if kv.ErrNotExist.Equal(err) {

util/admin/admin.go

@@ -226,7 +226,7 @@ const DefNumHistoryJobs = 10
 // The maximum count of history jobs is num.
 func GetHistoryDDLJobs(txn kv.Transaction, maxNumJobs int) ([]*model.Job, error) {
 	t := meta.NewMeta(txn)
-	jobs, err := t.GetAllHistoryDDLJobs()
+	jobs, err := t.GetLastNHistoryDDLJobs(maxNumJobs)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}