Merge pull request #499 from pingcap/c4pt0r/refactor-iterator

*: refactor iterator interface

kv/btree_buffer.go

@@ -16,6 +16,8 @@
 package kv
 
 import (
+	"github.com/juju/errors"
+	"github.com/ngaut/log"
 	"github.com/pingcap/tidb/kv/memkv"
 	"github.com/pingcap/tidb/util/types"
 )
@@ -73,7 +75,11 @@ func (b *btreeBuffer) NewIterator(param interface{}) Iterator {
 	}
 	iter := &btreeIter{e: e}
 	// the initial push...
-	iter.Next()
+	err = iter.Next()
+	if err != nil {
+		log.Error(err)
+		return &btreeIter{ok: false}
+	}
 	return iter
 }
 
@@ -93,10 +99,14 @@ func (i *btreeIter) Value() []byte {
 }
 
 // Next implements Iterator Next.
-func (i *btreeIter) Next() (Iterator, error) {
+func (i *btreeIter) Next() error {
 	k, v, err := i.e.Next()
-	i.k, i.v, i.ok = string(fromIfaces(k)), fromIfaces(v), err == nil
-	return i, err
+	if err != nil {
+		i.ok = false
+		return errors.Trace(err)
+	}
+	i.k, i.v, i.ok = string(fromIfaces(k)), fromIfaces(v), true
+	return nil
 }
 
 // Valid implements Iterator Valid.

kv/index_iter.go

@@ -87,11 +87,10 @@ func (c *indexIter) Next() (k []interface{}, h int64, err error) {
 		k = vv
 	}
 	// update new iter to next
-	newIt, err := c.it.Next()
+	err = c.it.Next()
 	if err != nil {
 		return nil, 0, errors.Trace(err)
 	}
-	c.it = newIt
 	return
 }
 
@@ -209,7 +208,7 @@ func (c *kvIndex) Drop(txn Transaction) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
-		it, err = it.Next()
+		err = it.Next()
 		if err != nil {
 			return errors.Trace(err)
 		}

kv/iter.go

@@ -72,7 +72,7 @@ func DecodeValue(data []byte) ([]interface{}, error) {
 func NextUntil(it Iterator, fn FnKeyCmp) (Iterator, error) {
 	var err error
 	for it.Valid() && !fn([]byte(it.Key())) {
-		it, err = it.Next()
+		err = it.Next()
 		if err != nil {
 			return nil, errors.Trace(err)
 		}

kv/kv.go

@@ -175,7 +175,7 @@ type FnKeyCmp func(key Key) bool
 
 // Iterator is the interface for a interator on KV store.
 type Iterator interface {
-	Next() (Iterator, error)
+	Next() error
 	Value() []byte
 	Key() string
 	Valid() bool

kv/mem_buffer_test.go

@@ -90,13 +90,13 @@ func checkNewIterator(c *C, buffer MemBuffer) {
 		c.Assert(iter.Key(), Equals, string(val))
 		c.Assert(valToStr(c, iter), Equals, string(val))
 
-		next, err := iter.Next()
+		err := iter.Next()
 		c.Assert(err, IsNil)
-		c.Assert(next.Valid(), IsTrue)
+		c.Assert(iter.Valid(), IsTrue)
 
 		val = encodeInt((i + 1) * indexStep)
-		c.Assert(next.Key(), Equals, string(val))
-		c.Assert(valToStr(c, next), Equals, string(val))
+		c.Assert(iter.Key(), Equals, string(val))
+		c.Assert(valToStr(c, iter), Equals, string(val))
 		iter.Close()
 	}
 
@@ -181,7 +181,7 @@ func (s *testKVSuite) TestNewIteratorMin(c *C) {
 		it := buffer.NewIterator(nil)
 		for it.Valid() {
 			cnt++
-			it, _ = it.Next()
+			it.Next()
 		}
 		c.Assert(cnt, Equals, 6)
 

kv/memdb_buffer.go

@@ -69,9 +69,9 @@ func (m *memDbBuffer) Release() {
 }
 
 // Next implements the Iterator Next.
-func (i *memDbIter) Next() (Iterator, error) {
+func (i *memDbIter) Next() error {
 	i.iter.Next()
-	return i, nil
+	return nil
 }
 
 // Valid implements the Iterator Valid.

kv/union_iter.go

@@ -44,13 +44,13 @@ func newUnionIter(dirtyIt Iterator, snapshotIt Iterator) *UnionIter {
 
 // Go next and update valid status.
 func (iter *UnionIter) dirtyNext() {
-	iter.dirtyIt, _ = iter.dirtyIt.Next()
+	iter.dirtyIt.Next()
 	iter.dirtyValid = iter.dirtyIt.Valid()
 }
 
 // Go next and update valid status.
 func (iter *UnionIter) snapshotNext() {
-	iter.snapshotIt, _ = iter.snapshotIt.Next()
+	iter.snapshotIt.Next()
 	iter.snapshotValid = iter.snapshotIt.Valid()
 }
 
@@ -115,14 +115,14 @@ func (iter *UnionIter) updateCur() {
 }
 
 // Next implements the Iterator Next interface.
-func (iter *UnionIter) Next() (Iterator, error) {
+func (iter *UnionIter) Next() error {
 	if !iter.curIsDirty {
 		iter.snapshotNext()
 	} else {
 		iter.dirtyNext()
 	}
 	iter.updateCur()
-	return iter, nil
+	return nil
 }
 
 // Value implements the Iterator Value interface.

store/localstore/kv_test.go

@@ -109,13 +109,13 @@ func checkSeek(c *C, txn kv.Transaction) {
 		c.Assert(iter.Key(), Equals, string(val))
 		c.Assert(valToStr(c, iter), Equals, string(val))
 
-		next, err := iter.Next()
+		err = iter.Next()
 		c.Assert(err, IsNil)
-		c.Assert(next.Valid(), IsTrue)
+		c.Assert(iter.Valid(), IsTrue)
 
 		val = encodeInt((i + 1) * indexStep)
-		c.Assert(next.Key(), Equals, string(val))
-		c.Assert(valToStr(c, next), Equals, string(val))
+		c.Assert(iter.Key(), Equals, string(val))
+		c.Assert(valToStr(c, iter), Equals, string(val))
 		iter.Close()
 	}
 
@@ -280,7 +280,7 @@ func (s *testKVSuite) TestDelete2(c *C) {
 	for it.Valid() {
 		err = txn.Delete([]byte(it.Key()))
 		c.Assert(err, IsNil)
-		it, err = it.Next()
+		err = it.Next()
 		c.Assert(err, IsNil)
 	}
 	txn.Commit()
@@ -415,7 +415,7 @@ func (s *testKVSuite) TestSeekMin(c *C) {
 	it, err := txn.Seek(nil)
 	for it.Valid() {
 		fmt.Printf("%s, %s\n", it.Key(), it.Value())
-		it, _ = it.Next()
+		it.Next()
 	}
 
 	it, err = txn.Seek([]byte("DATA_test_main_db_tbl_tbl_test_record__00000000000000000000"))

store/localstore/mvcc_test.go

@@ -142,7 +142,7 @@ func (t *testMvccSuite) TestMvccNext(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(it.Valid(), IsTrue)
 	for it.Valid() {
-		it, err = it.Next()
+		err = it.Next()
 		c.Assert(err, IsNil)
 	}
 	txn.Commit()
@@ -244,7 +244,7 @@ func (t *testMvccSuite) TestMvccSnapshotScan(c *C) {
 			if string(it.Value()) == "new" {
 				found = true
 			}
-			it, err = it.Next()
+			err = it.Next()
 			c.Assert(err, IsNil)
 		}
 		return found
@@ -275,7 +275,7 @@ func (t *testMvccSuite) TestBufferedIterator(c *C) {
 	c.Assert(err, IsNil)
 	cnt := 0
 	for iter.Valid() {
-		iter, err = iter.Next()
+		err = iter.Next()
 		c.Assert(err, IsNil)
 		cnt++
 	}

store/localstore/snapshot.go

@@ -151,7 +151,7 @@ func newDBIter(s *dbSnapshot, startKey kv.Key, exceptedVer kv.Version) *dbIter {
 	return it
 }
 
-func (it *dbIter) Next() (kv.Iterator, error) {
+func (it *dbIter) Next() error {
 	encKey := codec.EncodeBytes(nil, it.startKey)
 	var retErr error
 	var engineIter engine.Iterator
@@ -189,7 +189,7 @@ func (it *dbIter) Next() (kv.Iterator, error) {
 		// Current key's all versions are deleted, just go next key.
 		encKey = codec.EncodeBytes(nil, key.Next())
 	}
-	return it, errors.Trace(retErr)
+	return errors.Trace(retErr)
 }
 
 func (it *dbIter) Valid() bool {

store/localstore/txn.go

@@ -176,7 +176,7 @@ func (txn *dbTxn) Delete(k kv.Key) error {
 func (txn *dbTxn) each(f func(kv.Iterator) error) error {
 	iter := txn.UnionStore.Dirty.NewIterator(nil)
 	defer iter.Close()
-	for ; iter.Valid(); iter, _ = iter.Next() {
+	for ; iter.Valid(); iter.Next() {
 		if err := f(iter); err != nil {
 			return errors.Trace(err)
 		}

structure/hash.go

@@ -249,7 +249,7 @@ func (t *TxStructure) iterateHash(key []byte, fn func(k []byte, v []byte) error)
 			return errors.Trace(err)
 		}
 
-		it, err = it.Next()
+		err = it.Next()
 		if err != nil {
 			return errors.Trace(err)
 		}

util/prefix_helper.go

@@ -45,7 +45,10 @@ func ScanMetaWithPrefix(txn kv.Transaction, prefix string, filter func([]byte, [
 			if !filter([]byte(iter.Key()), iter.Value()) {
 				break
 			}
-			iter, err = iter.Next()
+			err = iter.Next()
+			if err != nil {
+				return errors.Trace(err)
+			}
 		} else {
 			break
 		}
@@ -75,7 +78,10 @@ func DelKeyWithPrefix(ctx context.Context, prefix string) error {
 
 		if iter.Valid() && strings.HasPrefix(iter.Key(), prefix) {
 			keys = append(keys, iter.Key())
-			iter, err = iter.Next()
+			err = iter.Next()
+			if err != nil {
+				return errors.Trace(err)
+			}
 		} else {
 			break
 		}