cdclog: fix lost delete (#1069)

pkg/kv/kv.go

@@ -19,6 +19,7 @@ import (
 	"sort"
 
 	"github.com/pingcap/errors"
+	sst "github.com/pingcap/kvproto/pkg/import_sstpb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
@@ -57,6 +58,10 @@ type Iter interface {
 	Value() []byte
 	// Close close this iter.
 	Close() error
+	// OpType represents operations of pair. currently we have two types.
+	// 1. Put
+	// 2. Delete
+	OpType() sst.Pair_OP
 }
 
 // IterProducer produces iterator with given range.
@@ -171,6 +176,14 @@ func (s *SimpleKVIter) Close() error {
 	return nil
 }
 
+// OpType implements Iter.KeyIsDelete.
+func (s *SimpleKVIter) OpType() sst.Pair_OP {
+	if s.Valid() && s.pairs[s.index].IsDelete {
+		return sst.Pair_Delete
+	}
+	return sst.Pair_Put
+}
+
 // Encoder encodes a row of SQL values into some opaque type which can be
 // consumed by OpenEngine.WriteEncoded.
 type Encoder interface {

pkg/restore/ingester.go

@@ -393,6 +393,7 @@ func (i *Ingester) writeToTiKV(
 			pair := &sst.Pair{
 				Key:   bytesBuf.AddBytes(iter.Key()),
 				Value: bytesBuf.AddBytes(iter.Value()),
+				Op:    iter.OpType(),
 			}
 			pairs = append(pairs, pair)
 		} else {

tests/br_log_restore/run.sh

@@ -141,6 +141,14 @@ if [ "$row_count" -ne "1" ]; then
     echo "TEST: [$TEST_NAME] fail on recover ts range test."
 fi
 
+# record a=3 should be deleted
+row_count=$(run_sql "SELECT COUNT(*) FROM ${DB}_DDL2.t2 WHERE a=3;" | awk '/COUNT/{print $2}')
+if [ "$row_count" -ne "0" ]; then
+    fail=true
+    echo "TEST: [$TEST_NAME] fail on key not deleted."
+fi
+
+
 for i in $(seq $DB_COUNT); do
     if [ "${row_count_ori[i]}" != "${row_count_new[i]}" ];then
         fail=true