From 47018f6451f1b0dd76b7679feb7398163d372b41 Mon Sep 17 00:00:00 2001 From: MyonKeminta <9948422+MyonKeminta@users.noreply.github.com> Date: Thu, 8 Nov 2018 18:21:40 +0800 Subject: [PATCH] tikvclient: Add endKey param to Scanner (#8178) Signed-off-by: MyonKeminta --- ddl/ddl_db_test.go | 2 +- ddl/delete_range.go | 2 +- ddl/index.go | 2 +- kv/buffer_store.go | 16 ++++---- kv/buffer_store_test.go | 2 +- kv/kv.go | 10 +++-- kv/mem_buffer_test.go | 20 +++++----- kv/memdb_buffer.go | 16 +++----- kv/mock.go | 12 +++--- kv/mock_test.go | 4 +- kv/union_store.go | 8 ++-- kv/union_store_test.go | 18 ++++----- session/txn.go | 16 ++++---- store/mockstore/mocktikv/mock_tikv_test.go | 24 ++++++++++- store/mockstore/mocktikv/rpc.go | 6 ++- store/store_test.go | 28 ++++++------- store/tikv/lock_test.go | 2 +- store/tikv/safepoint_test.go | 2 +- store/tikv/scan.go | 7 +++- store/tikv/scan_mock_test.go | 10 ++++- store/tikv/scan_test.go | 46 ++++++++++++++-------- store/tikv/snapshot.go | 10 ++--- store/tikv/snapshot_test.go | 4 +- store/tikv/store_test.go | 2 +- store/tikv/ticlient_test.go | 4 +- store/tikv/txn.go | 10 ++--- structure/hash.go | 2 +- table/tables/index.go | 6 +-- table/tables/tables.go | 4 +- util/admin/admin.go | 2 +- util/prefix_helper.go | 4 +- 31 files changed, 173 insertions(+), 128 deletions(-) diff --git a/ddl/ddl_db_test.go b/ddl/ddl_db_test.go index bff4d4bedc588..51b9d52da646d 100644 --- a/ddl/ddl_db_test.go +++ b/ddl/ddl_db_test.go @@ -1499,7 +1499,7 @@ func (s *testDBSuite) TestTruncateTable(c *C) { hasOldTableData := true for i := 0; i < waitForCleanDataRound; i++ { err = kv.RunInNewTxn(s.store, false, func(txn kv.Transaction) error { - it, err1 := txn.Seek(tablePrefix) + it, err1 := txn.Iter(tablePrefix, nil) if err1 != nil { return err1 } diff --git a/ddl/delete_range.go b/ddl/delete_range.go index bd1a0785eb5de..bd76b868fdb99 100644 --- a/ddl/delete_range.go +++ b/ddl/delete_range.go @@ -155,7 +155,7 @@ func (dr *delRange) doTask(ctx sessionctx.Context, r util.DelRangeTask) error { finish := true dr.keys = dr.keys[:0] err := kv.RunInNewTxn(dr.d.store, false, func(txn kv.Transaction) error { - iter, err := txn.Seek(oldStartKey) + iter, err := txn.Iter(oldStartKey, nil) if err != nil { return errors.Trace(err) } diff --git a/ddl/index.go b/ddl/index.go index fba45b8cbb20e..934a384d278c0 100644 --- a/ddl/index.go +++ b/ddl/index.go @@ -1009,7 +1009,7 @@ func iterateSnapshotRows(store kv.Storage, t table.Table, version uint64, seekHa return errors.Trace(err) } firstKey := t.RecordKey(seekHandle) - it, err := snap.Seek(firstKey) + it, err := snap.Iter(firstKey, nil) if err != nil { return errors.Trace(err) } diff --git a/kv/buffer_store.go b/kv/buffer_store.go index be2535e57c4ea..84342c71d28b8 100644 --- a/kv/buffer_store.go +++ b/kv/buffer_store.go @@ -74,26 +74,26 @@ func (s *BufferStore) Get(k Key) ([]byte, error) { return val, nil } -// Seek implements the Retriever interface. -func (s *BufferStore) Seek(k Key) (Iterator, error) { - bufferIt, err := s.MemBuffer.Seek(k) +// Iter implements the Retriever interface. +func (s *BufferStore) Iter(k Key, upperBound Key) (Iterator, error) { + bufferIt, err := s.MemBuffer.Iter(k, upperBound) if err != nil { return nil, errors.Trace(err) } - retrieverIt, err := s.r.Seek(k) + retrieverIt, err := s.r.Iter(k, upperBound) if err != nil { return nil, errors.Trace(err) } return NewUnionIter(bufferIt, retrieverIt, false) } -// SeekReverse implements the Retriever interface. -func (s *BufferStore) SeekReverse(k Key) (Iterator, error) { - bufferIt, err := s.MemBuffer.SeekReverse(k) +// IterReverse implements the Retriever interface. +func (s *BufferStore) IterReverse(k Key) (Iterator, error) { + bufferIt, err := s.MemBuffer.IterReverse(k) if err != nil { return nil, errors.Trace(err) } - retrieverIt, err := s.r.SeekReverse(k) + retrieverIt, err := s.r.IterReverse(k) if err != nil { return nil, errors.Trace(err) } diff --git a/kv/buffer_store_test.go b/kv/buffer_store_test.go index af84a9294cd24..1af716011c2a4 100644 --- a/kv/buffer_store_test.go +++ b/kv/buffer_store_test.go @@ -53,7 +53,7 @@ func (s testBufferStoreSuite) TestSaveTo(c *C) { err := bs.SaveTo(mutator) c.Check(err, IsNil) - iter, err := mutator.Seek(nil) + iter, err := mutator.Iter(nil, nil) c.Check(err, IsNil) for iter.Valid() { cmp := bytes.Compare(iter.Key(), iter.Value()) diff --git a/kv/kv.go b/kv/kv.go index 6d4c4b36e1975..cc986ac565a31 100644 --- a/kv/kv.go +++ b/kv/kv.go @@ -77,15 +77,17 @@ type Retriever interface { // Get gets the value for key k from kv store. // If corresponding kv pair does not exist, it returns nil and ErrNotExist. Get(k Key) ([]byte, error) - // Seek creates an Iterator positioned on the first entry that k <= entry's key. + // Iter creates an Iterator positioned on the first entry that k <= entry's key. // If such entry is not found, it returns an invalid Iterator with no error. + // It yields only keys that < upperBound. If upperBound is nil, it means the upperBound is unbounded. // The Iterator must be Closed after use. - Seek(k Key) (Iterator, error) + Iter(k Key, upperBound Key) (Iterator, error) - // SeekReverse creates a reversed Iterator positioned on the first entry which key is less than k. + // IterReverse creates a reversed Iterator positioned on the first entry which key is less than k. // The returned iterator will iterate from greater key to smaller key. // If k is nil, the returned iterator will be positioned at the last key. - SeekReverse(k Key) (Iterator, error) + // TODO: Add lower bound limit + IterReverse(k Key) (Iterator, error) } // Mutator is the interface wraps the basic Set and Delete methods. diff --git a/kv/mem_buffer_test.go b/kv/mem_buffer_test.go index 92807b528da2d..e7ec5a1f4f9e6 100644 --- a/kv/mem_buffer_test.go +++ b/kv/mem_buffer_test.go @@ -76,7 +76,7 @@ func valToStr(c *C, iter Iterator) string { func checkNewIterator(c *C, buffer MemBuffer) { for i := startIndex; i < testCount; i++ { val := encodeInt(i * indexStep) - iter, err := buffer.Seek(val) + iter, err := buffer.Iter(val, nil) c.Assert(err, IsNil) c.Assert([]byte(iter.Key()), BytesEquals, val) c.Assert(decodeInt([]byte(valToStr(c, iter))), Equals, i*indexStep) @@ -86,7 +86,7 @@ func checkNewIterator(c *C, buffer MemBuffer) { // Test iterator Next() for i := startIndex; i < testCount-1; i++ { val := encodeInt(i * indexStep) - iter, err := buffer.Seek(val) + iter, err := buffer.Iter(val, nil) c.Assert(err, IsNil) c.Assert([]byte(iter.Key()), BytesEquals, val) c.Assert(valToStr(c, iter), Equals, string(val)) @@ -102,7 +102,7 @@ func checkNewIterator(c *C, buffer MemBuffer) { } // Non exist and beyond maximum seek test - iter, err := buffer.Seek(encodeInt(testCount * indexStep)) + iter, err := buffer.Iter(encodeInt(testCount*indexStep), nil) c.Assert(err, IsNil) c.Assert(iter.Valid(), IsFalse) @@ -110,7 +110,7 @@ func checkNewIterator(c *C, buffer MemBuffer) { // it returns the smallest key that larger than the one we are seeking inBetween := encodeInt((testCount-1)*indexStep - 1) last := encodeInt((testCount - 1) * indexStep) - iter, err = buffer.Seek(inBetween) + iter, err = buffer.Iter(inBetween, nil) c.Assert(err, IsNil) c.Assert(iter.Valid(), IsTrue) c.Assert([]byte(iter.Key()), Not(BytesEquals), inBetween) @@ -140,7 +140,7 @@ func (s *testKVSuite) TestNewIterator(c *C) { defer testleak.AfterTest(c)() for _, buffer := range s.bs { // should be invalid - iter, err := buffer.Seek(nil) + iter, err := buffer.Iter(nil, nil) c.Assert(err, IsNil) c.Assert(iter.Valid(), IsFalse) @@ -155,7 +155,7 @@ func (s *testKVSuite) TestIterNextUntil(c *C) { buffer := NewMemDbBuffer(DefaultTxnMembufCap) insertData(c, buffer) - iter, err := buffer.Seek(nil) + iter, err := buffer.Iter(nil, nil) c.Assert(err, IsNil) err = NextUntil(iter, func(k Key) bool { @@ -168,7 +168,7 @@ func (s *testKVSuite) TestIterNextUntil(c *C) { func (s *testKVSuite) TestBasicNewIterator(c *C) { defer testleak.AfterTest(c)() for _, buffer := range s.bs { - it, err := buffer.Seek([]byte("2")) + it, err := buffer.Iter([]byte("2"), nil) c.Assert(err, IsNil) c.Assert(it.Valid(), IsFalse) } @@ -193,7 +193,7 @@ func (s *testKVSuite) TestNewIteratorMin(c *C) { } cnt := 0 - it, err := buffer.Seek(nil) + it, err := buffer.Iter(nil, nil) c.Assert(err, IsNil) for it.Valid() { cnt++ @@ -201,7 +201,7 @@ func (s *testKVSuite) TestNewIteratorMin(c *C) { } c.Assert(cnt, Equals, 6) - it, err = buffer.Seek([]byte("DATA_test_main_db_tbl_tbl_test_record__00000000000000000000")) + it, err = buffer.Iter([]byte("DATA_test_main_db_tbl_tbl_test_record__00000000000000000000"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "DATA_test_main_db_tbl_tbl_test_record__00000000000000000001") } @@ -294,7 +294,7 @@ func benchIterator(b *testing.B, buffer MemBuffer) { } b.ResetTimer() for i := 0; i < b.N; i++ { - iter, err := buffer.Seek(nil) + iter, err := buffer.Iter(nil, nil) if err != nil { b.Error(err) } diff --git a/kv/memdb_buffer.go b/kv/memdb_buffer.go index 7e0047b74cb9c..9a5e5455f5b12 100644 --- a/kv/memdb_buffer.go +++ b/kv/memdb_buffer.go @@ -50,14 +50,10 @@ func NewMemDbBuffer(cap int) MemBuffer { } } -// Seek creates an Iterator. -func (m *memDbBuffer) Seek(k Key) (Iterator, error) { - var i Iterator - if k == nil { - i = &memDbIter{iter: m.db.NewIterator(&util.Range{}), reverse: false} - } else { - i = &memDbIter{iter: m.db.NewIterator(&util.Range{Start: []byte(k)}), reverse: false} - } +// Iter creates an Iterator. +func (m *memDbBuffer) Iter(k Key, upperBound Key) (Iterator, error) { + i := &memDbIter{iter: m.db.NewIterator(&util.Range{Start: []byte(k), Limit: []byte(upperBound)}), reverse: false} + err := i.Next() if err != nil { return nil, errors.Trace(err) @@ -69,7 +65,7 @@ func (m *memDbBuffer) SetCap(cap int) { } -func (m *memDbBuffer) SeekReverse(k Key) (Iterator, error) { +func (m *memDbBuffer) IterReverse(k Key) (Iterator, error) { var i *memDbIter if k == nil { i = &memDbIter{iter: m.db.NewIterator(&util.Range{}), reverse: true} @@ -161,7 +157,7 @@ func (i *memDbIter) Close() { // WalkMemBuffer iterates all buffered kv pairs in memBuf func WalkMemBuffer(memBuf MemBuffer, f func(k Key, v []byte) error) error { - iter, err := memBuf.Seek(nil) + iter, err := memBuf.Iter(nil, nil) if err != nil { return errors.Trace(err) } diff --git a/kv/mock.go b/kv/mock.go index 25719ed500e64..b4dae37671de3 100644 --- a/kv/mock.go +++ b/kv/mock.go @@ -68,11 +68,11 @@ func (t *mockTxn) Get(k Key) ([]byte, error) { return nil, nil } -func (t *mockTxn) Seek(k Key) (Iterator, error) { +func (t *mockTxn) Iter(k Key, upperBound Key) (Iterator, error) { return nil, nil } -func (t *mockTxn) SeekReverse(k Key) (Iterator, error) { +func (t *mockTxn) IterReverse(k Key) (Iterator, error) { return nil, nil } @@ -207,10 +207,10 @@ func (s *mockSnapshot) BatchGet(keys []Key) (map[string][]byte, error) { return m, nil } -func (s *mockSnapshot) Seek(k Key) (Iterator, error) { - return s.store.Seek(k) +func (s *mockSnapshot) Iter(k Key, upperBound Key) (Iterator, error) { + return s.store.Iter(k, upperBound) } -func (s *mockSnapshot) SeekReverse(k Key) (Iterator, error) { - return s.store.SeekReverse(k) +func (s *mockSnapshot) IterReverse(k Key) (Iterator, error) { + return s.store.IterReverse(k) } diff --git a/kv/mock_test.go b/kv/mock_test.go index 93dc59490b4a4..2c1db84cf8a97 100644 --- a/kv/mock_test.go +++ b/kv/mock_test.go @@ -46,8 +46,8 @@ func (s testMockSuite) TestInterface(c *C) { if transaction.IsReadOnly() { transaction.Get(Key("lock")) transaction.Set(Key("lock"), []byte{}) - transaction.Seek(Key("lock")) - transaction.SeekReverse(Key("lock")) + transaction.Iter(Key("lock"), nil) + transaction.IterReverse(Key("lock")) } transaction.Commit(context.Background()) diff --git a/kv/union_store.go b/kv/union_store.go index 7823528ff4136..56c830e502e13 100644 --- a/kv/union_store.go +++ b/kv/union_store.go @@ -127,18 +127,18 @@ func (lmb *lazyMemBuffer) Delete(k Key) error { return lmb.mb.Delete(k) } -func (lmb *lazyMemBuffer) Seek(k Key) (Iterator, error) { +func (lmb *lazyMemBuffer) Iter(k Key, upperBound Key) (Iterator, error) { if lmb.mb == nil { return invalidIterator{}, nil } - return lmb.mb.Seek(k) + return lmb.mb.Iter(k, upperBound) } -func (lmb *lazyMemBuffer) SeekReverse(k Key) (Iterator, error) { +func (lmb *lazyMemBuffer) IterReverse(k Key) (Iterator, error) { if lmb.mb == nil { return invalidIterator{}, nil } - return lmb.mb.SeekReverse(k) + return lmb.mb.IterReverse(k) } func (lmb *lazyMemBuffer) Size() int { diff --git a/kv/union_store_test.go b/kv/union_store_test.go index c79c2207663a6..c93db701c6ae2 100644 --- a/kv/union_store_test.go +++ b/kv/union_store_test.go @@ -63,46 +63,46 @@ func (s *testUnionStoreSuite) TestSeek(c *C) { s.store.Set([]byte("2"), []byte("2")) s.store.Set([]byte("3"), []byte("3")) - iter, err := s.us.Seek(nil) + iter, err := s.us.Iter(nil, nil) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("1"), []byte("2"), []byte("3")}, [][]byte{[]byte("1"), []byte("2"), []byte("3")}) - iter, err = s.us.Seek([]byte("2")) + iter, err = s.us.Iter([]byte("2"), nil) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("3")}, [][]byte{[]byte("2"), []byte("3")}) s.us.Set([]byte("4"), []byte("4")) - iter, err = s.us.Seek([]byte("2")) + iter, err = s.us.Iter([]byte("2"), nil) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("3"), []byte("4")}, [][]byte{[]byte("2"), []byte("3"), []byte("4")}) s.us.Delete([]byte("3")) - iter, err = s.us.Seek([]byte("2")) + iter, err = s.us.Iter([]byte("2"), nil) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("4")}, [][]byte{[]byte("2"), []byte("4")}) } -func (s *testUnionStoreSuite) TestSeekReverse(c *C) { +func (s *testUnionStoreSuite) TestIterReverse(c *C) { defer testleak.AfterTest(c)() s.store.Set([]byte("1"), []byte("1")) s.store.Set([]byte("2"), []byte("2")) s.store.Set([]byte("3"), []byte("3")) - iter, err := s.us.SeekReverse(nil) + iter, err := s.us.IterReverse(nil) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("3"), []byte("2"), []byte("1")}, [][]byte{[]byte("3"), []byte("2"), []byte("1")}) - iter, err = s.us.SeekReverse([]byte("3")) + iter, err = s.us.IterReverse([]byte("3")) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("1")}, [][]byte{[]byte("2"), []byte("1")}) s.us.Set([]byte("0"), []byte("0")) - iter, err = s.us.SeekReverse([]byte("3")) + iter, err = s.us.IterReverse([]byte("3")) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("1"), []byte("0")}, [][]byte{[]byte("2"), []byte("1"), []byte("0")}) s.us.Delete([]byte("1")) - iter, err = s.us.SeekReverse([]byte("3")) + iter, err = s.us.IterReverse([]byte("3")) c.Assert(err, IsNil) checkIterator(c, iter, [][]byte{[]byte("2"), []byte("0")}, [][]byte{[]byte("2"), []byte("0")}) } diff --git a/session/txn.go b/session/txn.go index 5a2d17a055024..e03c1cf2c47d6 100644 --- a/session/txn.go +++ b/session/txn.go @@ -159,26 +159,26 @@ func (st *TxnState) Delete(k kv.Key) error { return st.buf.Delete(k) } -// Seek overrides the Transaction interface. -func (st *TxnState) Seek(k kv.Key) (kv.Iterator, error) { - bufferIt, err := st.buf.Seek(k) +// Iter overrides the Transaction interface. +func (st *TxnState) Iter(k kv.Key, upperBound kv.Key) (kv.Iterator, error) { + bufferIt, err := st.buf.Iter(k, upperBound) if err != nil { return nil, errors.Trace(err) } - retrieverIt, err := st.Transaction.Seek(k) + retrieverIt, err := st.Transaction.Iter(k, upperBound) if err != nil { return nil, errors.Trace(err) } return kv.NewUnionIter(bufferIt, retrieverIt, false) } -// SeekReverse overrides the Transaction interface. -func (st *TxnState) SeekReverse(k kv.Key) (kv.Iterator, error) { - bufferIt, err := st.buf.SeekReverse(k) +// IterReverse overrides the Transaction interface. +func (st *TxnState) IterReverse(k kv.Key) (kv.Iterator, error) { + bufferIt, err := st.buf.IterReverse(k) if err != nil { return nil, errors.Trace(err) } - retrieverIt, err := st.Transaction.SeekReverse(k) + retrieverIt, err := st.Transaction.IterReverse(k) if err != nil { return nil, errors.Trace(err) } diff --git a/store/mockstore/mocktikv/mock_tikv_test.go b/store/mockstore/mocktikv/mock_tikv_test.go index 4c7585fe533ee..c72d9a6912c1c 100644 --- a/store/mockstore/mocktikv/mock_tikv_test.go +++ b/store/mockstore/mocktikv/mock_tikv_test.go @@ -118,7 +118,11 @@ func (s *testMockTiKVSuite) mustDeleteOK(c *C, key string, startTS, commitTS uin } func (s *testMockTiKVSuite) mustScanOK(c *C, start string, limit int, ts uint64, expect ...string) { - pairs := s.store.Scan([]byte(start), nil, limit, ts, kvrpcpb.IsolationLevel_SI) + s.mustRangeScanOK(c, start, "", limit, ts, expect...) +} + +func (s *testMockTiKVSuite) mustRangeScanOK(c *C, start, end string, limit int, ts uint64, expect ...string) { + pairs := s.store.Scan([]byte(start), []byte(end), limit, ts, kvrpcpb.IsolationLevel_SI) c.Assert(len(pairs)*2, Equals, len(expect)) for i := 0; i < len(pairs); i++ { c.Assert(pairs[i].Err, IsNil) @@ -128,7 +132,11 @@ func (s *testMockTiKVSuite) mustScanOK(c *C, start string, limit int, ts uint64, } func (s *testMockTiKVSuite) mustReverseScanOK(c *C, end string, limit int, ts uint64, expect ...string) { - pairs := s.store.ReverseScan(nil, []byte(end), limit, ts, kvrpcpb.IsolationLevel_SI) + s.mustRangeReverseScanOK(c, "", end, limit, ts, expect...) +} + +func (s *testMockTiKVSuite) mustRangeReverseScanOK(c *C, start, end string, limit int, ts uint64, expect ...string) { + pairs := s.store.ReverseScan([]byte(start), []byte(end), limit, ts, kvrpcpb.IsolationLevel_SI) c.Assert(len(pairs)*2, Equals, len(expect)) for i := 0; i < len(pairs); i++ { c.Assert(pairs[i].Err, IsNil) @@ -227,6 +235,9 @@ func (s *testMockTiKVSuite) TestReverseScan(c *C) { s.mustReverseScanOK(c, "C\x00", 3, 10, "C", "C10", "A", "A10") s.mustReverseScanOK(c, "C\x00", 4, 10, "C", "C10", "A", "A10") s.mustReverseScanOK(c, "B", 1, 10, "A", "A10") + s.mustRangeReverseScanOK(c, "", "E", 5, 10, "C", "C10", "A", "A10") + s.mustRangeReverseScanOK(c, "", "C\x00", 5, 10, "C", "C10", "A", "A10") + s.mustRangeReverseScanOK(c, "A\x00", "C", 5, 10) } checkV10() @@ -238,6 +249,9 @@ func (s *testMockTiKVSuite) TestReverseScan(c *C) { s.mustReverseScanOK(c, "Z", 5, 20, "E", "E10", "D", "D20", "C", "C10", "B", "B20", "A", "A10") s.mustReverseScanOK(c, "C\x00", 5, 20, "C", "C10", "B", "B20", "A", "A10") s.mustReverseScanOK(c, "A\x00", 1, 20, "A", "A10") + s.mustRangeReverseScanOK(c, "B", "D", 5, 20, "C", "C10", "B", "B20") + s.mustRangeReverseScanOK(c, "B", "D\x00", 5, 20, "D", "D20", "C", "C10", "B", "B20") + s.mustRangeReverseScanOK(c, "B\x00", "D\x00", 5, 20, "D", "D20", "C", "C10") } checkV10() checkV20() @@ -286,6 +300,9 @@ func (s *testMockTiKVSuite) TestScan(c *C) { s.mustScanOK(c, "A\x00", 3, 10, "C", "C10", "E", "E10") s.mustScanOK(c, "C", 4, 10, "C", "C10", "E", "E10") s.mustScanOK(c, "F", 1, 10) + s.mustRangeScanOK(c, "", "E", 5, 10, "A", "A10", "C", "C10") + s.mustRangeScanOK(c, "", "C\x00", 5, 10, "A", "A10", "C", "C10") + s.mustRangeScanOK(c, "A\x00", "C", 5, 10) } checkV10() @@ -297,6 +314,9 @@ func (s *testMockTiKVSuite) TestScan(c *C) { s.mustScanOK(c, "", 5, 20, "A", "A10", "B", "B20", "C", "C10", "D", "D20", "E", "E10") s.mustScanOK(c, "C", 5, 20, "C", "C10", "D", "D20", "E", "E10") s.mustScanOK(c, "D\x00", 1, 20, "E", "E10") + s.mustRangeScanOK(c, "B", "D", 5, 20, "B", "B20", "C", "C10") + s.mustRangeScanOK(c, "B", "D\x00", 5, 20, "B", "B20", "C", "C10", "D", "D20") + s.mustRangeScanOK(c, "B\x00", "D\x00", 5, 20, "C", "C10", "D", "D20") } checkV10() checkV20() diff --git a/store/mockstore/mocktikv/rpc.go b/store/mockstore/mocktikv/rpc.go index 3a512d7a6230f..592f52aeb9d6b 100644 --- a/store/mockstore/mocktikv/rpc.go +++ b/store/mockstore/mocktikv/rpc.go @@ -227,7 +227,11 @@ func (h *rpcHandler) handleKvScan(req *kvrpcpb.ScanRequest) *kvrpcpb.ScanRespons if !h.checkKeyInRegion(req.GetStartKey()) { panic("KvScan: startKey not in region") } - pairs := h.mvccStore.Scan(req.GetStartKey(), h.endKey, int(req.GetLimit()), req.GetVersion(), h.isolationLevel) + endKey := h.endKey + 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) return &kvrpcpb.ScanResponse{ Pairs: convertToPbPairs(pairs), } diff --git a/store/store_test.go b/store/store_test.go index b69f92dbb7014..6fdb11ab58338 100644 --- a/store/store_test.go +++ b/store/store_test.go @@ -97,7 +97,7 @@ func valToStr(c *C, iter kv.Iterator) string { func checkSeek(c *C, txn kv.Transaction) { for i := startIndex; i < testCount; i++ { val := encodeInt(i * indexStep) - iter, err := txn.Seek(val) + iter, err := txn.Iter(val, nil) c.Assert(err, IsNil) c.Assert([]byte(iter.Key()), BytesEquals, val) c.Assert(decodeInt([]byte(valToStr(c, iter))), Equals, i*indexStep) @@ -107,7 +107,7 @@ func checkSeek(c *C, txn kv.Transaction) { // Test iterator Next() for i := startIndex; i < testCount-1; i++ { val := encodeInt(i * indexStep) - iter, err := txn.Seek(val) + iter, err := txn.Iter(val, nil) c.Assert(err, IsNil) c.Assert([]byte(iter.Key()), BytesEquals, val) c.Assert(valToStr(c, iter), Equals, string(val)) @@ -123,7 +123,7 @@ func checkSeek(c *C, txn kv.Transaction) { } // Non exist and beyond maximum seek test - iter, err := txn.Seek(encodeInt(testCount * indexStep)) + iter, err := txn.Iter(encodeInt(testCount*indexStep), nil) c.Assert(err, IsNil) c.Assert(iter.Valid(), IsFalse) @@ -131,7 +131,7 @@ func checkSeek(c *C, txn kv.Transaction) { // it returns the smallest key that larger than the one we are seeking inBetween := encodeInt((testCount-1)*indexStep - 1) last := encodeInt((testCount - 1) * indexStep) - iter, err = txn.Seek(inBetween) + iter, err = txn.Iter(inBetween, nil) c.Assert(err, IsNil) c.Assert(iter.Valid(), IsTrue) c.Assert([]byte(iter.Key()), Not(BytesEquals), inBetween) @@ -278,7 +278,7 @@ func (s *testKVSuite) TestDelete2(c *C) { txn, err = s.s.Begin() c.Assert(err, IsNil) - it, err := txn.Seek([]byte("DATA_test_tbl_department_record__0000000001_0003")) + it, err := txn.Iter([]byte("DATA_test_tbl_department_record__0000000001_0003"), nil) c.Assert(err, IsNil) for it.Valid() { err = txn.Delete([]byte(it.Key())) @@ -290,7 +290,7 @@ func (s *testKVSuite) TestDelete2(c *C) { txn, err = s.s.Begin() c.Assert(err, IsNil) - it, _ = txn.Seek([]byte("DATA_test_tbl_department_record__000000000")) + it, _ = txn.Iter([]byte("DATA_test_tbl_department_record__000000000"), nil) c.Assert(it.Valid(), IsFalse) txn.Commit(context.Background()) } @@ -312,7 +312,7 @@ func (s *testKVSuite) TestBasicSeek(c *C) { c.Assert(err, IsNil) defer txn.Commit(context.Background()) - it, err := txn.Seek([]byte("2")) + it, err := txn.Iter([]byte("2"), nil) c.Assert(err, IsNil) c.Assert(it.Valid(), Equals, false) txn.Delete([]byte("1")) @@ -333,30 +333,30 @@ func (s *testKVSuite) TestBasicTable(c *C) { err = txn.Set([]byte("1"), []byte("1")) c.Assert(err, IsNil) - it, err := txn.Seek([]byte("0")) + it, err := txn.Iter([]byte("0"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "1") err = txn.Set([]byte("0"), []byte("0")) c.Assert(err, IsNil) - it, err = txn.Seek([]byte("0")) + it, err = txn.Iter([]byte("0"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "0") err = txn.Delete([]byte("0")) c.Assert(err, IsNil) txn.Delete([]byte("1")) - it, err = txn.Seek([]byte("0")) + it, err = txn.Iter([]byte("0"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "2") err = txn.Delete([]byte("3")) c.Assert(err, IsNil) - it, err = txn.Seek([]byte("2")) + it, err = txn.Iter([]byte("2"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "2") - it, err = txn.Seek([]byte("3")) + it, err = txn.Iter([]byte("3"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "4") err = txn.Delete([]byte("2")) @@ -411,13 +411,13 @@ func (s *testKVSuite) TestSeekMin(c *C) { txn.Set([]byte(row.key), []byte(row.value)) } - it, err := txn.Seek(nil) + it, err := txn.Iter(nil, nil) for it.Valid() { fmt.Printf("%s, %s\n", it.Key(), it.Value()) it.Next() } - it, err = txn.Seek([]byte("DATA_test_main_db_tbl_tbl_test_record__00000000000000000000")) + it, err = txn.Iter([]byte("DATA_test_main_db_tbl_tbl_test_record__00000000000000000000"), nil) c.Assert(err, IsNil) c.Assert(string(it.Key()), Equals, "DATA_test_main_db_tbl_tbl_test_record__00000000000000000001") diff --git a/store/tikv/lock_test.go b/store/tikv/lock_test.go index f544053f6dbb6..e805abbe7d0b8 100644 --- a/store/tikv/lock_test.go +++ b/store/tikv/lock_test.go @@ -116,7 +116,7 @@ func (s *testLockSuite) TestScanLockResolveWithSeek(c *C) { txn, err := s.store.Begin() c.Assert(err, IsNil) - iter, err := txn.Seek([]byte("a")) + iter, err := txn.Iter([]byte("a"), nil) c.Assert(err, IsNil) for ch := byte('a'); ch <= byte('z'); ch++ { c.Assert(iter.Valid(), IsTrue) diff --git a/store/tikv/safepoint_test.go b/store/tikv/safepoint_test.go index f2d5858bd848d..d4936dcbd9180 100644 --- a/store/tikv/safepoint_test.go +++ b/store/tikv/safepoint_test.go @@ -101,7 +101,7 @@ func (s *testSafePointSuite) TestSafePoint(c *C) { s.waitUntilErrorPlugIn(txn3.startTS) - _, seekerr := txn3.Seek(encodeKey(s.prefix, "")) + _, seekerr := txn3.Iter(encodeKey(s.prefix, ""), nil) c.Assert(seekerr, NotNil) isFallBehind = terror.ErrorEqual(errors.Cause(geterr2), ErrGCTooEarly) isMayFallBehind = terror.ErrorEqual(errors.Cause(geterr2), ErrPDServerTimeout.GenByArgs("start timestamp may fall behind safe point")) diff --git a/store/tikv/scan.go b/store/tikv/scan.go index 0ccf297666ee6..f2ed062b583fa 100644 --- a/store/tikv/scan.go +++ b/store/tikv/scan.go @@ -30,10 +30,11 @@ type Scanner struct { cache []*pb.KvPair idx int nextStartKey []byte + endKey []byte eof bool } -func newScanner(snapshot *tikvSnapshot, startKey []byte, batchSize int) (*Scanner, error) { +func newScanner(snapshot *tikvSnapshot, startKey []byte, endKey []byte, batchSize int) (*Scanner, error) { // It must be > 1. Otherwise scanner won't skipFirst. if batchSize <= 1 { batchSize = scanBatchSize @@ -43,6 +44,7 @@ func newScanner(snapshot *tikvSnapshot, startKey []byte, batchSize int) (*Scanne batchSize: batchSize, valid: true, nextStartKey: startKey, + endKey: endKey, } err := scanner.Next() if kv.IsErrNotFound(err) { @@ -142,6 +144,7 @@ func (s *Scanner) getData(bo *Backoffer) error { Type: tikvrpc.CmdScan, Scan: &pb.ScanRequest{ StartKey: s.nextStartKey, + EndKey: s.endKey, Limit: uint32(s.batchSize), Version: s.startTS(), }, @@ -193,7 +196,7 @@ func (s *Scanner) getData(bo *Backoffer) error { // No more data in current Region. Next getData() starts // from current Region's endKey. s.nextStartKey = loc.EndKey - if len(loc.EndKey) == 0 { + if len(loc.EndKey) == 0 || (len(s.endKey) > 0 && kv.Key(s.nextStartKey).Cmp(kv.Key(s.endKey)) >= 0) { // Current Region is the last one. s.eof = true } diff --git a/store/tikv/scan_mock_test.go b/store/tikv/scan_mock_test.go index af453a985ff32..94b720314f2bd 100644 --- a/store/tikv/scan_mock_test.go +++ b/store/tikv/scan_mock_test.go @@ -41,11 +41,19 @@ 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"), 10) + scanner, err := newScanner(snapshot, []byte("a"), nil, 10) 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) + 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) } diff --git a/store/tikv/scan_test.go b/store/tikv/scan_test.go index 8cacd1e1dd26c..4b2de717681f8 100644 --- a/store/tikv/scan_test.go +++ b/store/tikv/scan_test.go @@ -18,6 +18,7 @@ import ( "time" . "github.com/pingcap/check" + "github.com/pingcap/tidb/kv" "golang.org/x/net/context" ) @@ -39,7 +40,7 @@ func (s *testScanSuite) SetUpSuite(c *C) { func (s *testScanSuite) TearDownSuite(c *C) { txn := s.beginTxn(c) - scanner, err := txn.Seek(encodeKey(s.prefix, "")) + scanner, err := txn.Iter(encodeKey(s.prefix, ""), nil) c.Assert(err, IsNil) c.Assert(scanner, NotNil) for scanner.Valid() { @@ -61,7 +62,25 @@ func (s *testScanSuite) beginTxn(c *C) *tikvTxn { return txn.(*tikvTxn) } -func (s *testScanSuite) TestSeek(c *C) { +func (s *testScanSuite) TestScan(c *C) { + check := func(c *C, scan kv.Iterator, rowNum int, keyOnly bool) { + for i := 0; i < rowNum; i++ { + k := scan.Key() + c.Assert([]byte(k), BytesEquals, encodeKey(s.prefix, s08d("key", i))) + if !keyOnly { + v := scan.Value() + c.Assert(v, BytesEquals, valueBytes(i)) + } + // Because newScan return first item without calling scan.Next() just like go-hbase, + // for-loop count will decrease 1. + if i < rowNum-1 { + scan.Next() + } + } + scan.Next() + c.Assert(scan.Valid(), IsFalse) + } + for _, rowNum := range s.rowNums { txn := s.beginTxn(c) for i := 0; i < rowNum; i++ { @@ -75,21 +94,14 @@ func (s *testScanSuite) TestSeek(c *C) { val, err := txn2.Get(encodeKey(s.prefix, s08d("key", 0))) c.Assert(err, IsNil) c.Assert(val, BytesEquals, valueBytes(0)) - scan, err := txn2.Seek(encodeKey(s.prefix, "")) + // Test scan without upperBound + scan, err := txn2.Iter(encodeKey(s.prefix, ""), nil) c.Assert(err, IsNil) - - for i := 0; i < rowNum; i++ { - k := scan.Key() - c.Assert([]byte(k), BytesEquals, encodeKey(s.prefix, s08d("key", i))) - v := scan.Value() - c.Assert(v, BytesEquals, valueBytes(i)) - // Because newScan return first item without calling scan.Next() just like go-hbase, - // for-loop count will decrease 1. - if i < rowNum-1 { - scan.Next() - } - } - scan.Next() - c.Assert(scan.Valid(), IsFalse) + check(c, scan, rowNum, false) + // Test scan with upperBound + upperBound := rowNum / 2 + scan, err = txn2.Iter(encodeKey(s.prefix, ""), encodeKey(s.prefix, s08d("key", upperBound))) + c.Assert(err, IsNil) + check(c, scan, upperBound, false) } } diff --git a/store/tikv/snapshot.go b/store/tikv/snapshot.go index ce8614a28d7b3..950e9a35892a0 100644 --- a/store/tikv/snapshot.go +++ b/store/tikv/snapshot.go @@ -275,14 +275,14 @@ func (s *tikvSnapshot) get(bo *Backoffer, k kv.Key) ([]byte, error) { } } -// Seek return a list of key-value pair after `k`. -func (s *tikvSnapshot) Seek(k kv.Key) (kv.Iterator, error) { - scanner, err := newScanner(s, k, scanBatchSize) +// 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) return scanner, errors.Trace(err) } -// SeekReverse creates a reversed Iterator positioned on the first entry which key is less than k. -func (s *tikvSnapshot) SeekReverse(k kv.Key) (kv.Iterator, error) { +// 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 } diff --git a/store/tikv/snapshot_test.go b/store/tikv/snapshot_test.go index bdf633f6414a2..f3f6892667d67 100644 --- a/store/tikv/snapshot_test.go +++ b/store/tikv/snapshot_test.go @@ -41,7 +41,7 @@ func (s *testSnapshotSuite) SetUpSuite(c *C) { func (s *testSnapshotSuite) TearDownSuite(c *C) { txn := s.beginTxn(c) - scanner, err := txn.Seek(encodeKey(s.prefix, "")) + scanner, err := txn.Iter(encodeKey(s.prefix, ""), nil) c.Assert(err, IsNil) c.Assert(scanner, NotNil) for scanner.Valid() { @@ -69,7 +69,7 @@ func (s *testSnapshotSuite) checkAll(keys []kv.Key, c *C) { m, err := snapshot.BatchGet(keys) c.Assert(err, IsNil) - scan, err := txn.Seek(encodeKey(s.prefix, "")) + scan, err := txn.Iter(encodeKey(s.prefix, ""), nil) c.Assert(err, IsNil) cnt := 0 for scan.Valid() { diff --git a/store/tikv/store_test.go b/store/tikv/store_test.go index c4f7b60521b9f..bfce16ed7c125 100644 --- a/store/tikv/store_test.go +++ b/store/tikv/store_test.go @@ -211,7 +211,7 @@ func (s *testStoreSuite) TestRequestPriority(c *C) { // Cover Seek request. client.priority = pb.CommandPri_High txn.SetOption(kv.Priority, kv.PriorityHigh) - iter, err := txn.Seek([]byte("key")) + iter, err := txn.Iter([]byte("key"), nil) c.Assert(err, IsNil) for iter.Valid() { c.Assert(iter.Next(), IsNil) diff --git a/store/tikv/ticlient_test.go b/store/tikv/ticlient_test.go index 3efe78a4b15bb..e8a52e836588d 100644 --- a/store/tikv/ticlient_test.go +++ b/store/tikv/ticlient_test.go @@ -61,7 +61,7 @@ func clearStorage(store kv.Storage) error { if err != nil { return errors.Trace(err) } - iter, err := txn.Seek(nil) + iter, err := txn.Iter(nil, nil) if err != nil { return errors.Trace(err) } @@ -93,7 +93,7 @@ func (s *testTiclientSuite) SetUpSuite(c *C) { func (s *testTiclientSuite) TearDownSuite(c *C) { // Clean all data, or it may pollute other data. txn := s.beginTxn(c) - scanner, err := txn.Seek(encodeKey(s.prefix, "")) + scanner, err := txn.Iter(encodeKey(s.prefix, ""), nil) c.Assert(err, IsNil) c.Assert(scanner, NotNil) for scanner.Valid() { diff --git a/store/tikv/txn.go b/store/tikv/txn.go index 9168fc1a2e7ac..c7e3c286c7401 100644 --- a/store/tikv/txn.go +++ b/store/tikv/txn.go @@ -109,23 +109,23 @@ func (txn *tikvTxn) String() string { return fmt.Sprintf("%d", txn.StartTS()) } -func (txn *tikvTxn) Seek(k kv.Key) (kv.Iterator, error) { +func (txn *tikvTxn) Iter(k kv.Key, upperBound kv.Key) (kv.Iterator, error) { metrics.TiKVTxnCmdCounter.WithLabelValues("seek").Inc() start := time.Now() defer func() { metrics.TiKVTxnCmdHistogram.WithLabelValues("seek").Observe(time.Since(start).Seconds()) }() - return txn.us.Seek(k) + return txn.us.Iter(k, upperBound) } -// SeekReverse creates a reversed Iterator positioned on the first entry which key is less than k. -func (txn *tikvTxn) SeekReverse(k kv.Key) (kv.Iterator, error) { +// IterReverse creates a reversed Iterator positioned on the first entry which key is less than k. +func (txn *tikvTxn) IterReverse(k kv.Key) (kv.Iterator, error) { metrics.TiKVTxnCmdCounter.WithLabelValues("seek_reverse").Inc() start := time.Now() defer func() { metrics.TiKVTxnCmdHistogram.WithLabelValues("seek_reverse").Observe(time.Since(start).Seconds()) }() - return txn.us.SeekReverse(k) + return txn.us.IterReverse(k) } func (txn *tikvTxn) Delete(k kv.Key) error { diff --git a/structure/hash.go b/structure/hash.go index ff9c3e4cd8ad8..d99809f845980 100644 --- a/structure/hash.go +++ b/structure/hash.go @@ -238,7 +238,7 @@ func (t *TxStructure) HClear(key []byte) error { func (t *TxStructure) iterateHash(key []byte, fn func(k []byte, v []byte) error) error { dataPrefix := t.hashDataKeyPrefix(key) - it, err := t.reader.Seek(dataPrefix) + it, err := t.reader.Iter(dataPrefix, nil) if err != nil { return errors.Trace(err) } diff --git a/table/tables/index.go b/table/tables/index.go index 808894035a5ed..f371ca80bdff2 100644 --- a/table/tables/index.go +++ b/table/tables/index.go @@ -232,7 +232,7 @@ func (c *index) Delete(sc *stmtctx.StatementContext, m kv.Mutator, indexedValues // Drop removes the KV index from store. func (c *index) Drop(rm kv.RetrieverMutator) error { - it, err := rm.Seek(c.prefix) + it, err := rm.Iter(c.prefix, nil) if err != nil { return errors.Trace(err) } @@ -261,7 +261,7 @@ func (c *index) Seek(sc *stmtctx.StatementContext, r kv.Retriever, indexedValues if err != nil { return nil, false, errors.Trace(err) } - it, err := r.Seek(key) + it, err := r.Iter(key, nil) if err != nil { return nil, false, errors.Trace(err) } @@ -275,7 +275,7 @@ func (c *index) Seek(sc *stmtctx.StatementContext, r kv.Retriever, indexedValues // SeekFirst returns an iterator which points to the first entry of the KV index. func (c *index) SeekFirst(r kv.Retriever) (iter table.IndexIterator, err error) { - it, err := r.Seek(c.prefix) + it, err := r.Iter(c.prefix, nil) if err != nil { return nil, errors.Trace(err) } diff --git a/table/tables/tables.go b/table/tables/tables.go index 66b4104cceb90..7002e67468899 100644 --- a/table/tables/tables.go +++ b/table/tables/tables.go @@ -701,7 +701,7 @@ func (t *Table) buildIndexForRow(ctx sessionctx.Context, rm kv.RetrieverMutator, // IterRecords implements table.Table IterRecords interface. func (t *Table) IterRecords(ctx sessionctx.Context, startKey kv.Key, cols []*table.Column, fn table.RecordIterFunc) error { - it, err := ctx.Txn().Seek(startKey) + it, err := ctx.Txn().Iter(startKey, nil) if err != nil { return errors.Trace(err) } @@ -831,7 +831,7 @@ func (t *Table) RebaseAutoID(ctx sessionctx.Context, newBase int64, isSetStep bo // Seek implements table.Table Seek interface. func (t *Table) Seek(ctx sessionctx.Context, h int64) (int64, bool, error) { seekKey := tablecodec.EncodeRowKeyWithHandle(t.ID, h) - iter, err := ctx.Txn().Seek(seekKey) + iter, err := ctx.Txn().Iter(seekKey, nil) if !iter.Valid() || !iter.Key().HasPrefix(t.RecordPrefix()) { // No more records in the table, skip to the end. return 0, false, nil diff --git a/util/admin/admin.go b/util/admin/admin.go index f603afe8f9df2..4434a2698aea4 100644 --- a/util/admin/admin.go +++ b/util/admin/admin.go @@ -541,7 +541,7 @@ func rowWithCols(sessCtx sessionctx.Context, txn kv.Retriever, t table.Table, h func iterRecords(sessCtx sessionctx.Context, retriever kv.Retriever, t table.Table, startKey kv.Key, cols []*table.Column, fn table.RecordIterFunc) error { - it, err := retriever.Seek(startKey) + it, err := retriever.Iter(startKey, nil) if err != nil { return errors.Trace(err) } diff --git a/util/prefix_helper.go b/util/prefix_helper.go index da91c87362fa9..015f60aa60760 100644 --- a/util/prefix_helper.go +++ b/util/prefix_helper.go @@ -26,7 +26,7 @@ import ( // ScanMetaWithPrefix scans metadata with the prefix. func ScanMetaWithPrefix(retriever kv.Retriever, prefix kv.Key, filter func(kv.Key, []byte) bool) error { - iter, err := retriever.Seek(prefix) + iter, err := retriever.Iter(prefix, nil) if err != nil { return errors.Trace(err) } @@ -56,7 +56,7 @@ func ScanMetaWithPrefix(retriever kv.Retriever, prefix kv.Key, filter func(kv.Ke // DelKeyWithPrefix deletes keys with prefix. func DelKeyWithPrefix(rm kv.RetrieverMutator, prefix kv.Key) error { var keys []kv.Key - iter, err := rm.Seek(prefix) + iter, err := rm.Iter(prefix, nil) if err != nil { return errors.Trace(err) }