func (o *stringRow) storeObject(b *Rpdb, bt *store.Batch, expireat uint64, obj interface{}) error {
	value, ok := obj.(rdb.String)
	if !ok || len(value) == 0 {
		return errors.Trace(ErrObjectValue)
	}

	o.ExpireAt, o.Value = expireat, value
	bt.Set(o.DataKey(), o.DataValue())
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *setRow) deleteObject(b *Rpdb, bt *store.Batch) error {
	it := b.getIterator()
	defer b.putIterator(it)
	for pfx := it.SeekTo(o.DataKeyPrefix()); it.Valid(); it.Next() {
		key := it.Key()
		if !bytes.HasPrefix(key, pfx) {
			break
		}
		bt.Del(key)
	}
	bt.Del(o.MetaKey())
	return it.Error()
}

func (b *Rpdb) commit(bt *store.Batch, fw *Forward) error {
	if bt.Len() == 0 {
		return nil
	}
	if err := b.db.Commit(bt); err != nil {
		log.WarnErrorf(err, "rpdb commit failed")
		return err
	}
	for i := b.itlist.Len(); i != 0; i-- {
		v := b.itlist.Remove(b.itlist.Front()).(*rpdbIterator)
		v.Close()
	}
	b.serial++
	return nil
}

func (o *setRow) storeObject(b *Rpdb, bt *store.Batch, expireat uint64, obj interface{}) error {
	set, ok := obj.(rdb.Set)
	if !ok || len(set) == 0 {
		return errors.Trace(ErrObjectValue)
	}
	for i, m := range set {
		if len(m) != 0 {
			continue
		}
		return errArguments("set[%d], len(member) = %d", i, len(m))
	}

	ms := &markSet{}
	for _, o.Member = range set {
		ms.Set(o.Member)
		bt.Set(o.DataKey(), o.DataValue())
	}
	o.Size, o.ExpireAt = ms.Len(), expireat
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *listRow) storeObject(b *Rpdb, bt *store.Batch, expireat uint64, obj interface{}) error {
	list, ok := obj.(rdb.List)
	if !ok || len(list) == 0 {
		return errors.Trace(ErrObjectValue)
	}
	for i, e := range list {
		if len(e) != 0 {
			continue
		}
		return errArguments("list[%d], len(value) = %d", i, len(e))
	}

	for i, value := range list {
		o.Index, o.Value = int64(i), value
		bt.Set(o.DataKey(), o.DataValue())
	}
	o.Lindex, o.Rindex = 0, int64(len(list))
	o.ExpireAt = expireat
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *hashRow) storeObject(b *Rpdb, bt *store.Batch, expireat uint64, obj interface{}) error {
	hash, ok := obj.(rdb.Hash)
	if !ok || len(hash) == 0 {
		return errors.Trace(ErrObjectValue)
	}
	for i, e := range hash {
		if e == nil {
			return errArguments("hash[%d] is nil", i)
		}
		if len(e.Field) == 0 || len(e.Value) == 0 {
			return errArguments("hash[%d], len(field) = %d, len(value) = %d", i, len(e.Field), len(e.Value))
		}
	}

	ms := &markSet{}
	for _, e := range hash {
		o.Field, o.Value = e.Field, e.Value
		ms.Set(o.Field)
		bt.Set(o.DataKey(), o.DataValue())
	}
	o.Size, o.ExpireAt = ms.Len(), expireat
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *zsetRow) storeObject(b *Rpdb, bt *store.Batch, expireat uint64, obj interface{}) error {
	zset, ok := obj.(rdb.ZSet)
	if !ok || len(zset) == 0 {
		return errors.Trace(ErrObjectValue)
	}
	for i, e := range zset {
		if e == nil {
			return errArguments("zset[%d] is nil", i)
		}
		if len(e.Member) == 0 {
			return errArguments("zset[%d], len(member) = %d", i, len(e.Member))
		}
	}

	ms := &markSet{}
	for _, e := range zset {
		o.Member, o.Score = e.Member, e.Score
		ms.Set(o.Member)
		bt.Set(o.DataKey(), o.DataValue())
	}
	o.Size, o.ExpireAt = ms.Len(), expireat
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *stringRow) deleteObject(b *Rpdb, bt *store.Batch) error {
	bt.Del(o.DataKey())
	bt.Del(o.MetaKey())
	return nil
}