func (o *stringRow) storeObject(s *Store, bt *engine.Batch, expireat int64, 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 *hashRow) deleteObject(s *Store, bt *engine.Batch) error {
	it := s.getIterator()
	defer s.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 (o *listRow) storeObject(s *Store, bt *engine.Batch, expireat int64, 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 *setRow) storeObject(s *Store, bt *engine.Batch, expireat int64, 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 (s *Store) commit(bt *engine.Batch, fw *Forward) error {
	if bt.Len() == 0 {
		return nil
	}

	s.travelPreCommitHandlers(fw)

	if err := s.db.Commit(bt); err != nil {
		log.Warningf("store commit failed - %s", err)
		return errors.Trace(err)
	}
	for i := s.itlist.Len(); i != 0; i-- {
		v := s.itlist.Remove(s.itlist.Front()).(*storeIterator)
		v.Close()
	}
	s.serial++

	s.travelPostCommitHandlers(fw)

	return nil
}

func (o *zsetRow) storeObject(s *Store, bt *engine.Batch, expireat int64, 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
		if math.IsNaN(o.Score) {
			return errors.Errorf("invalid nan score")
		}

		ms.Set(o.Member)
		bt.Set(o.DataKey(), o.DataValue())
		bt.Set(o.IndexKey(), o.IndexValue())
	}
	o.Size, o.ExpireAt = ms.Len(), expireat
	bt.Set(o.MetaKey(), o.MetaValue())
	return nil
}

func (o *hashRow) storeObject(s *Store, bt *engine.Batch, expireat int64, 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 *stringRow) deleteObject(s *Store, bt *engine.Batch) error {
	bt.Del(o.DataKey())
	bt.Del(o.MetaKey())
	return nil
}