codec(ticdc): correct default value for debezium (#12475)

close #12474

Author: wk989898

pkg/sink/codec/debezium/codec.go

@@ -518,37 +518,43 @@ func (c *dbzCodec) writeDebeziumFieldValue(
 		return nil
 
 	case mysql.TypeEnum:
-		v, ok := value.(uint64)
-		if !ok {
+		switch v := value.(type) {
+		case uint64:
+			enumVar, err := types.ParseEnumValue(ft.GetElems(), v)
+			if err != nil {
+				// Invalid enum value inserted in non-strict mode.
+				writer.WriteStringField(col.GetName(), "")
+				return nil
+			}
+			writer.WriteStringField(col.GetName(), enumVar.Name)
+		case string:
+			writer.WriteStringField(col.GetName(), v)
+		default:
 			return cerror.ErrDebeziumEncodeFailed.GenWithStack(
 				"unexpected column value type %T for enum column %s",
 				value,
 				col.GetName())
 		}
-		enumVar, err := types.ParseEnumValue(ft.GetElems(), v)
-		if err != nil {
-			// Invalid enum value inserted in non-strict mode.
-			writer.WriteStringField(col.GetName(), "")
-			return nil
-		}
-		writer.WriteStringField(col.GetName(), enumVar.Name)
 		return nil
 
 	case mysql.TypeSet:
-		v, ok := value.(uint64)
-		if !ok {
+		switch v := value.(type) {
+		case uint64:
+			setVar, err := types.ParseSetValue(ft.GetElems(), v)
+			if err != nil {
+				// Invalid enum value inserted in non-strict mode.
+				writer.WriteStringField(col.GetName(), "")
+				return nil
+			}
+			writer.WriteStringField(col.GetName(), setVar.Name)
+		case string:
+			writer.WriteStringField(col.GetName(), v)
+		default:
 			return cerror.ErrDebeziumEncodeFailed.GenWithStack(
 				"unexpected column value type %T for set column %s",
 				value,
 				col.GetName())
 		}
-		setVar, err := types.ParseSetValue(ft.GetElems(), v)
-		if err != nil {
-			// Invalid enum value inserted in non-strict mode.
-			writer.WriteStringField(col.GetName(), "")
-			return nil
-		}
-		writer.WriteStringField(col.GetName(), setVar.Name)
 		return nil
 
 	case mysql.TypeNewDecimal:

tests/integration_tests/debezium/sql/data_types.sql

@@ -390,8 +390,8 @@ INSERT INTO t_json VALUES ('["foo"]', 1);
 */
 
 CREATE TABLE t_enum(
-  col_e   ENUM('a', 'b', 'c'),
-  col_s   SET('a', 'b', 'c'),
+  col_e   ENUM('a', 'b', 'c') DEFAULT 'a',
+  col_s   SET('a', 'b', 'c') DEFAULT 'a',
   pk      INT PRIMARY KEY
 );
 
@@ -400,3 +400,5 @@ INSERT INTO t_enum VALUES ('a', 'c', 1);
 SET sql_mode='';
 INSERT INTO t_enum VALUES ('d', 'e', 2);
 SET sql_mode='strict_trans_tables';
+
+INSERT INTO t_enum VALUES (null, null, 3);

tests/integration_tests/debezium_basic/data/test.sql

@@ -5,3 +5,7 @@ USE test;
 create table t1 (id int primary key, account_id int not null);
 alter table t1 add unique key(account_id);
 insert into t1 values (12,34);
+
+CREATE TABLE test.vec(id int primary key, data VECTOR(5));
+INSERT INTO test.vec(id, data) VALUES (1, "[1,2,3,4,5]");
+