sink: fix codec json encoding with non binary string (#2764) (#2781)

cdc/sink/codec/json.go

@@ -64,7 +64,15 @@ func (c *column) FromSinkColumn(col *model.Column) {
 	}
 	switch col.Type {
 	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
-		str := string(col.Value.([]byte))
+		var str string
+		switch col.Value.(type) {
+		case []byte:
+			str = string(col.Value.([]byte))
+		case string:
+			str = col.Value.(string)
+		default:
+			log.Panic("invalid column value, please report a bug", zap.Any("col", col))
+		}
 		if c.Flag.IsBinary() {
 			str = strconv.Quote(str)
 			str = str[1 : len(str)-1]

cdc/sink/codec/json_test.go

@@ -387,6 +387,28 @@ func (s *columnSuite) TestFormatCol(c *check.C) {
 	c.Assert(row2, check.DeepEquals, row)
 }
 
+func (s *columnSuite) TestNonBinaryStringCol(c *check.C) {
+	defer testleak.AfterTest(c)()
+	col := &model.Column{
+		Name:  "test",
+		Type:  mysql.TypeString,
+		Value: "value",
+	}
+	jsonCol := column{}
+	jsonCol.FromSinkColumn(col)
+	row := &mqMessageRow{Update: map[string]column{"test": jsonCol}}
+	rowEncode, err := row.Encode()
+	c.Assert(err, check.IsNil)
+	row2 := new(mqMessageRow)
+	err = row2.Decode(rowEncode)
+	c.Assert(err, check.IsNil)
+	c.Assert(row2, check.DeepEquals, row)
+	jsonCol2 := row2.Update["test"]
+	col2 := jsonCol2.ToSinkColumn("test")
+	col2.Value = string(col2.Value.([]byte))
+	c.Assert(col2, check.DeepEquals, col)
+}
+
 func (s *columnSuite) TestVarBinaryCol(c *check.C) {
 	defer testleak.AfterTest(c)()
 	col := &model.Column{