func (t *TxStructure) encodeHashDataKey(key []byte, field []byte) kv.Key {
	ek := make([]byte, 0, len(t.prefix)+len(key)+len(field)+30)
	ek = append(ek, t.prefix...)
	ek = codec.EncodeBytes(ek, key)
	ek = codec.EncodeUint(ek, uint64(HashData))
	return codec.EncodeBytes(ek, field)
}

func encodeRegionKey(r *metapb.Region) *metapb.Region {
	if r.StartKey != nil {
		r.StartKey = codec.EncodeBytes(nil, r.StartKey)
	}
	if r.EndKey != nil {
		r.EndKey = codec.EncodeBytes(nil, r.EndKey)
	}
	return r
}

func (it *dbIter) Next(fn kv.FnKeyCmp) (kv.Iterator, error) {
	encKey := codec.EncodeBytes(nil, it.startKey)
	// max key
	encEndKey := codec.EncodeBytes(nil, []byte{0xff, 0xff})
	var retErr error
	var engineIter engine.Iterator
	for {
		engineIter, retErr = it.s.db.Seek(encKey)
		if retErr != nil {
			return nil, errors.Trace(retErr)
		}
		// Check if overflow
		if !engineIter.Next() {
			it.valid = false
			break
		}

		metaKey := engineIter.Key()
		// Check if meet the end of table.
		if bytes.Compare(metaKey, encEndKey) >= 0 {
			it.valid = false
			break
		}
		// 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 = key
			it.v = val
			it.startKey = key.Next()
			break
		}
		// Release the iterator, and update key
		engineIter.Release()
		// Current key's all versions are deleted, just go next key.
		encKey = codec.EncodeBytes(nil, key.Next())
	}
	engineIter.Release()
	return it, errors.Trace(retErr)
}

// GetRegion encodes the key before send requests to pd-server and decodes the
// returned StartKey && EndKey from pd-server.
func (c *codecPDClient) GetRegion(key []byte) (*metapb.Region, *metapb.Peer, error) {
	encodedKey := codec.EncodeBytes([]byte(nil), key)
	region, peer, err := c.Client.GetRegion(encodedKey)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}
	if region == nil {
		return nil, nil, nil
	}
	if len(region.StartKey) != 0 {
		_, decoded, err := codec.DecodeBytes(region.StartKey)
		if err != nil {
			return nil, nil, errors.Trace(err)
		}
		region.StartKey = decoded
	}
	if len(region.EndKey) != 0 {
		_, decoded, err := codec.DecodeBytes(region.EndKey)
		if err != nil {
			return nil, nil, errors.Trace(err)
		}
		region.EndKey = decoded
	}
	return region, peer, nil
}

func (t *TxStructure) encodeListDataKey(key []byte, index int64) kv.Key {
	ek := make([]byte, 0, len(t.prefix)+len(key)+36)
	ek = append(ek, t.prefix...)
	ek = codec.EncodeBytes(ek, key)
	ek = codec.EncodeUint(ek, uint64(ListData))
	return codec.EncodeInt(ek, index)
}

// Both lock and unlock are used for simulating scenario of percolator papers.
func (s *dbStore) tryConditionLockKey(tid uint64, key string, snapshotVal []byte) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if _, ok := s.keysLocked[key]; ok {
		return errors.Trace(kv.ErrLockConflict)
	}

	metaKey := codec.EncodeBytes(nil, []byte(key))
	currValue, err := s.db.Get(metaKey)
	if errors2.ErrorEqual(err, kv.ErrNotExist) || currValue == nil {
		// If it's a new key, we won't need to check its version
		return nil
	}
	if err != nil {
		return errors.Trace(err)
	}
	_, ver, err := codec.DecodeUint(currValue)
	if err != nil {
		return errors.Trace(err)
	}

	// If there's newer version of this key, returns error.
	if ver > tid {
		log.Warnf("txn:%d, tryLockKey condition not match for key %s, currValue:%q, snapshotVal:%q", tid, key, currValue, snapshotVal)
		return errors.Trace(kv.ErrConditionNotMatch)
	}

	s.keysLocked[key] = tid

	return nil
}

func (t *TxStructure) encodeStringDataKey(key []byte) kv.Key {
	// for codec Encode, we may add extra bytes data, so here and following encode
	// we will use extra length like 4 for a little optimization.
	ek := make([]byte, 0, len(t.prefix)+len(key)+24)
	ek = append(ek, t.prefix...)
	ek = codec.EncodeBytes(ek, key)
	return codec.EncodeUint(ek, uint64(StringData))
}

func genIndexPrefix(indexPrefix, indexName string) string {
	// Use EncodeBytes to guarantee generating different index prefix.
	// e.g, two indices c1 and c with index prefix p, if no EncodeBytes,
	// the index format looks p_c and p_c1, if c has an index value which the first encoded byte is '1',
	// we will meet an error, because p_c1 is for index c1.
	// If EncodeBytes, c1 -> c1\x00\x01 and c -> c\x00\x01, the prefixs are different.
	key := fmt.Sprintf("%s_%s", indexPrefix, indexName)
	return string(codec.EncodeBytes(nil, []byte(key)))
}

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 (txn *dbTxn) doCommit() error {
	b := txn.store.newBatch()
	keysLocked := make([]string, 0, len(txn.snapshotVals))
	defer func() {
		for _, key := range keysLocked {
			txn.store.unLockKeys(key)
		}
	}()

	// check lazy condition pairs
	if err := txn.UnionStore.CheckLazyConditionPairs(); err != nil {
		return errors.Trace(err)
	}

	txn.Snapshot.Release()

	// Check locked keys
	for k := range txn.snapshotVals {
		err := txn.store.tryConditionLockKey(txn.tid, k)
		if err != nil {
			return errors.Trace(err)
		}
		keysLocked = append(keysLocked, k)
	}

	// disable version provider temporarily
	providerMu.Lock()
	defer providerMu.Unlock()

	curVer, err := globalVersionProvider.CurrentVersion()
	if err != nil {
		return errors.Trace(err)
	}
	err = txn.each(func(iter kv.Iterator) error {
		metaKey := codec.EncodeBytes(nil, []byte(iter.Key()))
		// put dummy meta key, write current version
		b.Put(metaKey, codec.EncodeUint(nil, curVer.Ver))
		mvccKey := MvccEncodeVersionKey(kv.Key(iter.Key()), curVer)
		if len(iter.Value()) == 0 { // Deleted marker
			b.Put(mvccKey, nil)
		} else {
			b.Put(mvccKey, iter.Value())
		}
		return nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	// Update commit version.
	txn.version = curVer
	return txn.store.writeBatch(b)
}

// GetRegion encodes the key before send requests to pd-server and decodes the
// returned StartKey && EndKey from pd-server.
func (c *codecPDClient) GetRegion(key []byte) (*metapb.Region, *metapb.Peer, error) {
	encodedKey := codec.EncodeBytes([]byte(nil), key)
	region, peer, err := c.Client.GetRegion(encodedKey)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}
	if region == nil {
		return nil, nil, nil
	}
	err = decodeRegionMetaKey(region)
	if err != nil {
		return nil, nil, errors.Trace(err)
	}
	return region, peer, nil
}

func (txn *dbTxn) doCommit() error {
	b := txn.store.newBatch()
	keysLocked := make([]string, 0, len(txn.snapshotVals))
	defer func() {
		for _, key := range keysLocked {
			txn.store.unLockKeys(key)
		}
	}()
	// Check locked keys
	for k, v := range txn.snapshotVals {
		err := txn.store.tryConditionLockKey(txn.tid, k, v)
		if err != nil {
			return errors.Trace(err)
		}
		keysLocked = append(keysLocked, k)
	}

	// Check dirty store
	curVer, err := globalVersionProvider.CurrentVersion()
	if err != nil {
		return errors.Trace(err)
	}
	err = txn.each(func(iter kv.Iterator) error {
		metaKey := codec.EncodeBytes(nil, []byte(iter.Key()))
		// put dummy meta key, write current version
		b.Put(metaKey, codec.EncodeUint(nil, curVer.Ver))
		mvccKey := MvccEncodeVersionKey(kv.Key(iter.Key()), curVer)
		if len(iter.Value()) == 0 { // Deleted marker
			b.Put(mvccKey, nil)
		} else {
			b.Put(mvccKey, iter.Value())
		}
		return nil
	})
	if err != nil {
		return errors.Trace(err)
	}
	// Update commit version.
	txn.version = curVer
	// Release read lock before write. Workaround for BoltDB.
	txn.Snapshot.Release()
	return txn.store.writeBatch(b)
}

// Both lock and unlock are used for simulating scenario of percolator papers.
func (s *dbStore) tryConditionLockKey(tid uint64, key string) error {
	metaKey := codec.EncodeBytes(nil, []byte(key))

	s.mu.Lock()
	defer s.mu.Unlock()

	if s.closed {
		return errors.Trace(ErrDBClosed)
	}

	if _, ok := s.keysLocked[key]; ok {
		return errors.Trace(kv.ErrLockConflict)
	}

	currValue, err := s.db.Get(metaKey)
	if terror.ErrorEqual(err, kv.ErrNotExist) {
		s.keysLocked[key] = tid
		return nil
	}
	if err != nil {
		return errors.Trace(err)
	}

	// key not exist.
	if currValue == nil {
		s.keysLocked[key] = tid
		return nil
	}
	_, ver, err := codec.DecodeUint(currValue)
	if err != nil {
		return errors.Trace(err)
	}

	// If there's newer version of this key, returns error.
	if ver > tid {
		log.Warnf("txn:%d, tryLockKey condition not match for key %s, currValue:%q", tid, key, currValue)
		return errors.Trace(kv.ErrConditionNotMatch)
	}

	s.keysLocked[key] = tid
	return nil
}

func encodeKey(s string) []byte {
	return codec.EncodeBytes(nil, []byte(s))
}

func (t *TxStructure) hashDataKeyPrefix(key []byte) kv.Key {
	ek := make([]byte, 0, len(t.prefix)+len(key)+24)
	ek = append(ek, t.prefix...)
	ek = codec.EncodeBytes(ek, key)
	return codec.EncodeUint(ek, uint64(HashData))
}

// MvccEncodeVersionKey returns the encoded key
func MvccEncodeVersionKey(key kv.Key, ver kv.Version) kv.EncodedKey {
	b := codec.EncodeBytes(nil, key)
	ret := codec.EncodeUintDesc(b, ver.Ver)
	return ret
}

func encodeKey(prefix, s string) []byte {
	return codec.EncodeBytes(nil, []byte(fmt.Sprintf("%s_%s", prefix, s)))
}

func newMvccKey(key []byte) mvccKey {
	if len(key) == 0 {
		return nil
	}
	return codec.EncodeBytes(nil, key)
}

func (t *TxStructure) encodeHashMetaKey(key []byte) []byte {
	ek := make([]byte, 0, len(t.prefix)+len(key)+24)
	ek = append(ek, t.prefix...)
	ek = codec.EncodeBytes(ek, key)
	return codec.EncodeUint(ek, uint64(HashMeta))
}