func (i *iterator) Next() bool {
	i.d.DB.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(bucketName)
		c := b.Cursor()
		var key []byte
		var value []byte

		if !i.seeked {
			i.seeked = true
			if i.key == nil {
				key, value = c.First()
			} else {
				key, value = c.Seek(i.key)
			}
		} else {
			c.Seek(i.key)
			key, value = c.Next()
		}

		i.key = bytes.CloneBytes(key)
		i.value = bytes.CloneBytes(value)

		return nil
	})

	return i.key != nil
}

func (d *db) SeekReverse(startKey []byte) ([]byte, []byte, error) {
	var key, value []byte
	err := d.DB.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(bucketName)
		c := b.Cursor()
		var k, v []byte
		if startKey == nil {
			k, v = c.Last()
		} else {
			c.Seek(startKey)
			k, v = c.Prev()
		}
		if k != nil {
			key, value = bytes.CloneBytes(k), bytes.CloneBytes(v)
		}
		return nil
	})

	if err != nil {
		return nil, nil, errors.Trace(err)
	}
	if key == nil {
		return nil, nil, errors.Trace(engine.ErrNotFound)
	}
	return key, value, nil
}

func (d *db) MultiSeek(keys [][]byte) []*engine.MSeekResult {
	res := make([]*engine.MSeekResult, 0, len(keys))
	d.DB.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(bucketName)
		c := b.Cursor()
		for _, key := range keys {
			var k, v []byte
			if key == nil {
				k, v = c.First()
			} else {
				k, v = c.Seek(key)
			}

			r := &engine.MSeekResult{}
			if k == nil {
				r.Err = engine.ErrNotFound
			} else {
				r.Key, r.Value, r.Err = bytes.CloneBytes(k), bytes.CloneBytes(v), nil
			}

			res = append(res, r)
		}
		return nil
	})

	return res
}

func (i *iterator) Seek(startKey []byte) bool {
	i.d.DB.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(bucketName)
		c := b.Cursor()
		var key []byte
		var value []byte

		key, value = c.Seek(startKey)

		i.key = bytes.CloneBytes(key)
		i.value = bytes.CloneBytes(value)

		return nil
	})

	return i.key != nil
}

func (it *dbIter) Next() (kv.Iterator, error) {
	encKey := codec.EncodeBytes(nil, it.startKey)
	var retErr error
	var engineIter engine.Iterator
	for {
		var err error
		engineIter, err = it.s.internalSeek(encKey)
		if err != nil {
			it.valid = false
			retErr = err
			break
		}

		metaKey := engineIter.Key()
		// Get real key from metaKey
		key, _, err := MvccDecode(metaKey)
		if err != nil {
			// It's not a valid metaKey, maybe overflow (other data).
			it.valid = false
			break
		}
		// Get kv pair.
		val, err := it.s.MvccGet(key, it.exceptedVersion)
		if err != nil && !errors2.ErrorEqual(err, kv.ErrNotExist) {
			// Get this version error
			it.valid = false
			retErr = err
			break
		}
		if val != nil {
			it.k = bytes.CloneBytes(key)
			it.v = bytes.CloneBytes(val)
			it.startKey = key.Next()
			break
		}
		// Current key's all versions are deleted, just go next key.
		encKey = codec.EncodeBytes(nil, key.Next())
	}
	return it, errors.Trace(retErr)
}

func (d *db) Get(key []byte) ([]byte, error) {
	var value []byte

	err := d.DB.View(func(tx *bolt.Tx) error {
		b := tx.Bucket(bucketName)
		v := b.Get(key)
		if v == nil {
			return errors.Trace(engine.ErrNotFound)
		}
		value = bytes.CloneBytes(v)
		return nil
	})

	return value, errors.Trace(err)
}

func (gc *localstoreCompactor) getAllVersions(k kv.Key) ([]kv.EncodedKey, error) {
	startKey := MvccEncodeVersionKey(k, kv.MaxVersion)
	endKey := MvccEncodeVersionKey(k, kv.MinVersion)

	it, err := gc.db.Seek(startKey)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer it.Release()

	var ret []kv.EncodedKey
	for it.Next() {
		if kv.EncodedKey(it.Key()).Cmp(endKey) < 0 {
			ret = append(ret, bytes.CloneBytes(kv.EncodedKey(it.Key())))
		}
	}
	return ret, nil
}

func (gc *localstoreCompactor) getAllVersions(key kv.Key) ([]kv.EncodedKey, error) {
	var keys []kv.EncodedKey
	k := key
	for ver := kv.MaxVersion; ver.Ver > 0; ver.Ver-- {
		mvccK, _, err := gc.db.Seek(MvccEncodeVersionKey(key, ver))
		if terror.ErrorEqual(err, engine.ErrNotFound) {
			break
		}
		if err != nil {
			return nil, errors.Trace(err)
		}
		k, ver, err = MvccDecode(mvccK)
		if k.Cmp(key) != 0 {
			break
		}
		if err != nil {
			return nil, errors.Trace(err)
		}
		keys = append(keys, bytes.CloneBytes(mvccK))
	}
	return keys, nil
}

func (s *dbSnapshot) MvccGet(k kv.Key, ver kv.Version) ([]byte, error) {
	// engine Snapshot return nil, nil for value not found,
	// so here we will check nil and return kv.ErrNotExist.
	// get newest version, (0, MaxUint64)
	// Key arrangement:
	// Key -> META
	// ...
	// Key_ver
	// Key_ver-1
	// Key_ver-2
	// ...
	// Key_ver-n
	// Key_0
	// NextKey -> META
	// NextKey_xxx
	startKey := MvccEncodeVersionKey(k, ver)
	endKey := MvccEncodeVersionKey(k, kv.MinVersion)

	it, err := s.internalSeek(startKey)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var rawKey []byte
	var v []byte
	// Check if the scan is not exceed this key's all versions and the value is not
	// tombstone.
	if kv.EncodedKey(it.Key()).Cmp(endKey) < 0 && !isTombstone(it.Value()) {
		rawKey = it.Key()
		v = it.Value()
	}
	// No such key (or it's tombstone).
	if rawKey == nil {
		return nil, kv.ErrNotExist
	}
	return bytes.CloneBytes(v), nil
}