fix: when rewrite complex type, use overlay rather than append or merge (#932)

internal/rdb/types/hash.go

@@ -24,6 +24,7 @@ func (o *HashObject) LoadFromBuffer(rd io.Reader, key string, typeByte byte) {
 func (o *HashObject) Rewrite() <-chan RedisCmd {
 	go func() {
 		defer close(o.cmdC)
+		o.cmdC <- RedisCmd{"del", o.key}
 		switch o.typeByte {
 		case rdbTypeHash:
 			o.readHash()

internal/rdb/types/list.go

@@ -30,6 +30,7 @@ func (o *ListObject) LoadFromBuffer(rd io.Reader, key string, typeByte byte) {
 func (o *ListObject) Rewrite() <-chan RedisCmd {
 	go func() {
 		defer close(o.cmdC)
+		o.cmdC <- RedisCmd{"del", o.key}
 		switch o.typeByte {
 		case rdbTypeList:
 			o.readList()

internal/rdb/types/mbbloom.go

@@ -140,6 +140,7 @@ func (o *BloomObject) Rewrite() <-chan RedisCmd {
 		} else {
 			h = getEncodedHeader(&o.sb, true, true)
 		}
+		cmdC <- RedisCmd{"del", o.key}
 		cmd := RedisCmd{"BF.LOADCHUNK", o.key, "1", h}
 		cmdC <- cmd
 		curIter := uint64(1)

internal/rdb/types/set.go

@@ -24,6 +24,7 @@ func (o *SetObject) LoadFromBuffer(rd io.Reader, key string, typeByte byte) {
 func (o *SetObject) Rewrite() <-chan RedisCmd {
 	go func() {
 		defer close(o.cmdC)
+		o.cmdC <- RedisCmd{"del", o.key}
 		switch o.typeByte {
 		case rdbTypeSet:
 			o.readSet()

internal/rdb/types/set_test.go

@@ -19,6 +19,9 @@ func testOne(t *testing.T, typeByte byte, setData string, values []string) {
 	cmdC := o.Rewrite()
 	var elements []string
 	for cmd := range cmdC {
+		if cmd[0] == "del" {
+			continue
+		}
 		elements = append(elements, cmd[2])
 	}
 	if len(elements) != len(values) {

internal/rdb/types/stream.go

@@ -57,6 +57,7 @@ func (o *StreamObject) LoadFromBuffer(rd io.Reader, key string, typeByte byte) {
 func (o *StreamObject) Rewrite() <-chan RedisCmd {
 	go func() {
 		defer close(o.cmdC)
+		o.cmdC <- RedisCmd{"del", o.key}
 		switch o.typeByte {
 		case rdbTypeStreamListpacks:
 			o.readStream()

internal/rdb/types/tairhash.go

@@ -26,6 +26,7 @@ func (o *TairHashObject) Rewrite() <-chan RedisCmd {
 		dictSizeStr := structure.ReadModuleUnsigned(rd)
 		key := structure.ReadModuleString(rd)
 		size, _ := strconv.Atoi(dictSizeStr)
+		cmdC <- RedisCmd{"del", key}
 		for i := 0; i < size; i++ {
 			skey := structure.ReadModuleString(rd)
 			version := structure.ReadModuleUnsigned(rd)

internal/rdb/types/tairzset.go

@@ -25,6 +25,7 @@ func (o *TairZsetObject) Rewrite() <-chan RedisCmd {
 	cmdC := o.cmdC
 	go func() {
 		defer close(cmdC)
+		cmdC <- RedisCmd{"del", o.key}
 		length, _ := strconv.Atoi(structure.ReadModuleUnsigned(rd))
 		scoreNum, _ := strconv.Atoi(structure.ReadModuleUnsigned(rd))
 		for i := 0; i < length; i++ {

internal/rdb/types/zset.go

@@ -25,6 +25,7 @@ func (o *ZsetObject) LoadFromBuffer(rd io.Reader, key string, typeByte byte) {
 func (o *ZsetObject) Rewrite() <-chan RedisCmd {
 	go func() {
 		defer close(o.cmdC)
+		o.cmdC <- RedisCmd{"del", o.key}
 		switch o.typeByte {
 		case rdbTypeZSet:
 			o.readZset()