You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

276 lines
7.0 KiB

// Copyright 2016 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.
package tikv
import (
"bytes"
"context"
"github.com/pingcap/errors"
pb "github.com/pingcap/kvproto/pkg/kvrpcpb"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/store/tikv/tikvrpc"
"github.com/pingcap/tidb/util/logutil"
"go.uber.org/zap"
)
// Scanner support tikv scan
type Scanner struct {
snapshot *tikvSnapshot
batchSize int
cache []*pb.KvPair
idx int
nextStartKey kv.Key
endKey kv.Key
// Use for reverse scan.
nextEndKey kv.Key
reverse bool
valid bool
eof bool
}
func newScanner(snapshot *tikvSnapshot, startKey []byte, endKey []byte, batchSize int, reverse bool) (*Scanner, error) {
// It must be > 1. Otherwise scanner won't skipFirst.
if batchSize <= 1 {
batchSize = scanBatchSize
}
scanner := &Scanner{
snapshot: snapshot,
batchSize: batchSize,
valid: true,
nextStartKey: startKey,
endKey: endKey,
reverse: reverse,
nextEndKey: endKey,
}
err := scanner.Next()
if kv.IsErrNotFound(err) {
return scanner, nil
}
return scanner, errors.Trace(err)
}
// Valid return valid.
func (s *Scanner) Valid() bool {
return s.valid
}
// Key return key.
func (s *Scanner) Key() kv.Key {
if s.valid {
return s.cache[s.idx].Key
}
return nil
}
// Value return value.
func (s *Scanner) Value() []byte {
if s.valid {
return s.cache[s.idx].Value
}
return nil
}
// Next return next element.
func (s *Scanner) Next() error {
bo := NewBackofferWithVars(context.WithValue(context.Background(), txnStartKey, s.snapshot.version.Ver), scannerNextMaxBackoff, s.snapshot.vars)
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)
if err != nil {
s.Close()
return errors.Trace(err)
}
if s.idx >= len(s.cache) {
continue
}
}
current := s.cache[s.idx]
if (!s.reverse && (len(s.endKey) > 0 && kv.Key(current.Key).Cmp(kv.Key(s.endKey)) >= 0)) ||
(s.reverse && len(s.nextStartKey) > 0 && kv.Key(current.Key).Cmp(kv.Key(s.nextStartKey)) < 0) {
s.eof = true
s.Close()
return nil
}
// Try to resolve the lock
if current.GetError() != nil {
// 'current' would be modified if the lock being resolved
if err := s.resolveCurrentLock(bo, current); err != nil {
s.Close()
return errors.Trace(err)
}
// The check here does not violate the KeyOnly semantic, because current's value
// is filled by resolveCurrentLock which fetches the value by snapshot.get, so an empty
// value stands for NotExist
if len(current.Value) == 0 {
continue
}
}
return nil
}
}
// Close close iterator.
func (s *Scanner) Close() {
s.valid = false
}
func (s *Scanner) startTS() uint64 {
return s.snapshot.version.Ver
}
func (s *Scanner) resolveCurrentLock(bo *Backoffer, current *pb.KvPair) error {
ctx := context.Background()
val, err := s.snapshot.get(ctx, bo, current.Key)
if err != nil {
return errors.Trace(err)
}
current.Error = nil
current.Value = val
return nil
}
func (s *Scanner) getData(bo *Backoffer) error {
logutil.BgLogger().Debug("txn getData",
zap.Stringer("nextStartKey", s.nextStartKey),
zap.Stringer("nextEndKey", s.nextEndKey),
zap.Bool("reverse", s.reverse),
zap.Uint64("txnStartTS", s.startTS()))
sender := NewRegionRequestSender(s.snapshot.store.regionCache, s.snapshot.store.client)
var reqEndKey, reqStartKey []byte
var loc *KeyLocation
var err error
for {
if !s.reverse {
loc, err = s.snapshot.store.regionCache.LocateKey(bo, s.nextStartKey)
} else {
loc, err = s.snapshot.store.regionCache.LocateEndKey(bo, s.nextEndKey)
}
if err != nil {
return errors.Trace(err)
}
if !s.reverse {
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
}
}
sreq := &pb.ScanRequest{
StartKey: s.nextStartKey,
EndKey: reqEndKey,
Limit: uint32(s.batchSize),
Version: s.startTS(),
KeyOnly: s.snapshot.keyOnly,
}
if s.reverse {
sreq.StartKey = s.nextEndKey
sreq.EndKey = reqStartKey
sreq.Reverse = true
}
s.snapshot.mu.RLock()
req := tikvrpc.NewReplicaReadRequest(tikvrpc.CmdScan, sreq, s.snapshot.mu.replicaRead, &s.snapshot.replicaReadSeed, pb.Context{
Priority: s.snapshot.priority,
NotFillCache: s.snapshot.notFillCache,
TaskId: s.snapshot.mu.taskID,
})
s.snapshot.mu.RUnlock()
resp, err := sender.SendReq(bo, req, loc.Region, ReadTimeoutMedium)
if err != nil {
return errors.Trace(err)
}
regionErr, err := resp.GetRegionError()
if err != nil {
return errors.Trace(err)
}
if regionErr != nil {
logutil.BgLogger().Debug("scanner getData failed",
zap.Stringer("regionErr", regionErr))
err = bo.Backoff(BoRegionMiss, errors.New(regionErr.String()))
if err != nil {
return errors.Trace(err)
}
continue
}
if resp.Resp == nil {
return errors.Trace(ErrBodyMissing)
}
cmdScanResp := resp.Resp.(*pb.ScanResponse)
err = s.snapshot.store.CheckVisibility(s.startTS())
if err != nil {
return errors.Trace(err)
}
kvPairs := cmdScanResp.Pairs
// Check if kvPair contains error, it should be a Lock.
for _, pair := range kvPairs {
if keyErr := pair.GetError(); keyErr != nil {
lock, err := extractLockFromKeyErr(keyErr)
if err != nil {
return errors.Trace(err)
}
pair.Key = lock.Key
}
}
s.cache, s.idx = kvPairs, 0
if len(kvPairs) < s.batchSize {
// No more data in current Region. Next getData() starts
// from current Region's endKey.
if !s.reverse {
s.nextStartKey = loc.EndKey
} else {
s.nextEndKey = reqStartKey
}
if (!s.reverse && (len(loc.EndKey) == 0 || (len(s.endKey) > 0 && kv.Key(s.nextStartKey).Cmp(kv.Key(s.endKey)) >= 0))) ||
(s.reverse && (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
}
return nil
}
// next getData() starts from the last key in kvPairs (but skip
// it by appending a '\x00' to the key). Note that next getData()
// may get an empty response if the Region in fact does not have
// more data.
lastKey := kvPairs[len(kvPairs)-1].GetKey()
if !s.reverse {
s.nextStartKey = kv.Key(lastKey).Next()
} else {
s.nextEndKey = kv.Key(lastKey)
}
return nil
}
}