codec(ticdc): move canal test util to the utils, it should be shared …

…among all prot… (#10265)

close #10264

pkg/sink/codec/canal/canal_json_message.go

@@ -229,15 +229,6 @@ func canalJSONFormatColumn(value interface{}, name string, mysqlTypeStr string)
 		log.Panic("canal-json encoded message should have type in `string`")
 	}
 
-	if mysqlType == mysql.TypeBit || mysqlType == mysql.TypeSet {
-		val, err := strconv.ParseUint(data, 10, 64)
-		if err != nil {
-			log.Panic("invalid column value for bit", zap.Any("col", result), zap.Error(err))
-		}
-		result.Value = val
-		return result
-	}
-
 	var err error
 	if utils.IsBinaryMySQLType(mysqlTypeStr) {
 		// when encoding the `JavaSQLTypeBLOB`, use `ISO8859_1` decoder, now reverse it back.
@@ -246,6 +237,44 @@ func canalJSONFormatColumn(value interface{}, name string, mysqlTypeStr string)
 		if err != nil {
 			log.Panic("invalid column value, please report a bug", zap.Any("col", result), zap.Error(err))
 		}
+		result.Value = value
+		return result
+	}
+
+	switch mysqlType {
+	case mysql.TypeBit, mysql.TypeSet:
+		value, err = strconv.ParseUint(data, 10, 64)
+		if err != nil {
+			log.Panic("invalid column value for bit", zap.Any("col", result), zap.Error(err))
+		}
+	case mysql.TypeTiny, mysql.TypeShort, mysql.TypeLong, mysql.TypeInt24, mysql.TypeYear:
+		value, err = strconv.ParseInt(data, 10, 64)
+		if err != nil {
+			log.Panic("invalid column value for int", zap.Any("col", result), zap.Error(err))
+		}
+	case mysql.TypeEnum:
+		value, err = strconv.ParseInt(data, 10, 64)
+		if err != nil {
+			log.Panic("invalid column value for enum", zap.Any("col", result), zap.Error(err))
+		}
+	case mysql.TypeLonglong:
+		value, err = strconv.ParseInt(data, 10, 64)
+		if err != nil {
+			value, err = strconv.ParseUint(data, 10, 64)
+			if err != nil {
+				log.Panic("invalid column value for bigint", zap.Any("col", result), zap.Error(err))
+			}
+		}
+	case mysql.TypeFloat:
+		value, err = strconv.ParseFloat(data, 32)
+		if err != nil {
+			log.Panic("invalid column value for float", zap.Any("col", result), zap.Error(err))
+		}
+	case mysql.TypeDouble:
+		value, err = strconv.ParseFloat(data, 64)
+		if err != nil {
+			log.Panic("invalid column value for double", zap.Any("col", result), zap.Error(err))
+		}
 	}
 
 	result.Value = value