forked from talent-plan/tinykv
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Nick Cameron <nrc@ncameron.org>
- Loading branch information
Showing
8 changed files
with
327 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package commands | ||
|
|
||
| import ( | ||
| "github.com/pingcap-incubator/tinykv/kv/tikv/storage/kvstore" | ||
| "github.com/pingcap-incubator/tinykv/proto/pkg/kvrpcpb" | ||
| ) | ||
|
|
||
| type Scan struct { | ||
| request *kvrpcpb.ScanRequest | ||
| response *kvrpcpb.ScanResponse | ||
| } | ||
|
|
||
| func NewScan(request *kvrpcpb.ScanRequest) Scan { | ||
| return Scan{request, &kvrpcpb.ScanResponse{}} | ||
| } | ||
|
|
||
| func (s *Scan) BuildTxn(txn *kvstore.MvccTxn) error { | ||
| txn.StartTS = &s.request.Version | ||
|
|
||
| iter := kvstore.NewScanner(s.request.StartKey, txn) | ||
| limit := s.request.Limit | ||
| for { | ||
| if limit == 0 { | ||
| // We've scanned up to the requested limit. | ||
| return nil | ||
| } | ||
| limit -= 1 | ||
| key, value, err := iter.Next() | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if key == nil { | ||
| // Reached the end of the DB | ||
| return nil | ||
| } | ||
|
|
||
| pair := kvrpcpb.KvPair{} | ||
| pair.Key = key | ||
| pair.Value = value | ||
| s.response.Pairs = append(s.response.Pairs, &pair) | ||
| } | ||
| } | ||
|
|
||
| func (s *Scan) Context() *kvrpcpb.Context { | ||
| return s.request.Context | ||
| } | ||
|
|
||
| func (s *Scan) Response() interface{} { | ||
| return s.response | ||
| } | ||
|
|
||
| func (s *Scan) HandleError(err error) interface{} { | ||
| if err == nil { | ||
| return nil | ||
| } | ||
|
|
||
| if regionErr := extractRegionError(err); regionErr != nil { | ||
| s.response.RegionError = regionErr | ||
| return &s.response | ||
| } | ||
|
|
||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package kvstore | ||
|
|
||
| import ( | ||
| "github.com/pingcap-incubator/tinykv/kv/util/engine_util" | ||
| ) | ||
|
|
||
| // Scanner is used for reading multiple sequential key/value pairs from the storage layer. It is aware of the implementation | ||
| // of the storage layer and returns results suitable for users. | ||
| // Invariant: either the scanner is finished and can not be used, or it is ready to return a value immediately. | ||
| type Scanner struct { | ||
| writeIter engine_util.DBIterator | ||
| txn *MvccTxn | ||
| } | ||
|
|
||
| // NewScanner creates a new scanner ready to read from the snapshot in txn. | ||
| func NewScanner(startKey []byte, txn *MvccTxn) *Scanner { | ||
| writeIter := txn.Reader.IterCF(engine_util.CfWrite) | ||
| writeIter.Seek(EncodeKey(startKey, TsMax)) | ||
| return &Scanner{ | ||
| writeIter: writeIter, | ||
| txn: txn, | ||
| } | ||
| } | ||
|
|
||
| // Next returns the next key/value pair from the scanner. If the scanner is exhausted, then it will return `nil, nil, nil`. | ||
| func (scan *Scanner) Next() ([]byte, []byte, error) { | ||
| // Search for the next relevant key/value. | ||
| for { | ||
| if !scan.writeIter.Valid() { | ||
| // The underlying iterator is exhausted - we've reached the end of the DB. | ||
| return nil, nil, nil | ||
| } | ||
|
|
||
| item := scan.writeIter.Item() | ||
| userKey := decodeUserKey(item.Key()) | ||
| commitTs := decodeTimestamp(item.Key()) | ||
|
|
||
| if commitTs >= *scan.txn.StartTS { | ||
| // The key was not committed before our transaction started, find an earlier key. | ||
| scan.writeIter.Seek(EncodeKey(userKey, commitTs-1)) | ||
| continue | ||
| } | ||
|
|
||
| // Note: we might check if userKey is locked (since we should not read an uncommitted transaction). However, | ||
| // because we are iterating over writes, we are guaranteed never to get a locked key at our timestamp (i.e., if | ||
| // the key were locked, then we would use the older value, which is what we will get via the write in any case). | ||
|
|
||
| writeValue, err := item.Value() | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
| write, err := ParseWrite(writeValue) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
| if write.Kind != WriteKindPut { | ||
| // Key is removed, go to next key. | ||
| scan.writeIter.Seek(EncodeKey(userKey, 0)) | ||
| continue | ||
| } | ||
|
|
||
| value, err := scan.txn.GetValue(userKey, write.StartTS) | ||
| if err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| scan.writeIter.Next() | ||
|
|
||
| return userKey, value, nil | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.