From 7810ad2525f2ffdd1d7649aa7f1d5d71d0c49b3d Mon Sep 17 00:00:00 2001
From: Ling Jin <7138436+3AceShowHand@users.noreply.github.com>
Date: Wed, 21 Jun 2023 15:37:18 +0800
Subject: [PATCH] mounter(ticdc): enum type default value set explicitly to
 avoid wrong enum value fetched by the encoder (#9262)

close pingcap/tiflow#9259
---
 cdc/entry/mounter.go        |  7 ++++++-
 cdc/entry/mounter_test.go   | 13 ++++++++++++-
 pkg/sink/codec/avro/avro.go | 14 +++++++++++---
 3 files changed, 29 insertions(+), 5 deletions(-)

diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index dcb012498b5..ed411e6165d 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -723,7 +723,12 @@ func getDefaultOrZeroValue(col *timodel.ColumnInfo) (types.Datum, any, int, stri
 		case mysql.TypeEnum:
 			// For enum type, if no default value and not null is set,
 			// the default value is the first element of the enum list
-			d = types.NewDatum(col.FieldType.GetElem(0))
+			name := col.FieldType.GetElem(0)
+			enumValue, err := types.ParseEnumName(col.FieldType.GetElems(), name, col.GetCollate())
+			if err != nil {
+				return d, nil, 0, "", errors.Trace(err)
+			}
+			d = types.NewMysqlEnumDatum(enumValue)
 		case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
 			return d, emptyBytes, sizeOfEmptyBytes, "", nil
 		default:
diff --git a/cdc/entry/mounter_test.go b/cdc/entry/mounter_test.go
index 5eb7891bfc1..3782151669c 100644
--- a/cdc/entry/mounter_test.go
+++ b/cdc/entry/mounter_test.go
@@ -615,6 +615,9 @@ func TestGetDefaultZeroValue(t *testing.T) {
 	ftTypeEnumNotNull.SetFlag(mysql.NotNullFlag)
 	ftTypeEnumNotNull.SetElems([]string{"e0", "e1"})
 
+	// mysql.TypeEnum + null
+	ftTypeEnumNull := types.NewFieldType(mysql.TypeEnum)
+
 	// mysql.TypeSet + notnull
 	ftTypeSetNotNull := types.NewFieldType(mysql.TypeSet)
 	ftTypeSetNotNull.SetFlag(mysql.NotNullFlag)
@@ -959,7 +962,7 @@ func TestGetDefaultZeroValue(t *testing.T) {
 			ColInfo: timodel.ColumnInfo{FieldType: *ftTypeEnumNotNull},
 			// TypeEnum value will be a string and then translate to []byte
 			// NotNull && no default will choose first element
-			Res:     uint64(0),
+			Res:     uint64(1),
 			Default: nil,
 		},
 		// mysql.TypeEnum + notnull + default
@@ -973,6 +976,14 @@ func TestGetDefaultZeroValue(t *testing.T) {
 			Res:     "e1",
 			Default: "e1",
 		},
+		// mysql.TypeEnum + null
+		{
+			Name: "mysql.TypeEnum + null",
+			ColInfo: timodel.ColumnInfo{
+				FieldType: *ftTypeEnumNull,
+			},
+			Res: nil,
+		},
 		// mysql.TypeSet + notnull
 		{
 			Name:    "mysql.TypeSet + notnull",
diff --git a/pkg/sink/codec/avro/avro.go b/pkg/sink/codec/avro/avro.go
index ec53b98fea2..a26ed6cf71b 100644
--- a/pkg/sink/codec/avro/avro.go
+++ b/pkg/sink/codec/avro/avro.go
@@ -913,8 +913,11 @@ func columnToAvroData(
 		if v, ok := col.Value.(string); ok {
 			return v, "string", nil
 		}
-		enumVar, err := types.ParseEnumValue(ft.GetElems(), col.Value.(uint64))
+		elements := ft.GetElems()
+		number := col.Value.(uint64)
+		enumVar, err := types.ParseEnumValue(elements, number)
 		if err != nil {
+			log.Info("avro encoder parse enum value failed", zap.Strings("elements", elements), zap.Uint64("number", number))
 			return nil, "", cerror.WrapError(cerror.ErrAvroEncodeFailed, err)
 		}
 		return enumVar.Name, "string", nil
@@ -922,8 +925,12 @@ func columnToAvroData(
 		if v, ok := col.Value.(string); ok {
 			return v, "string", nil
 		}
-		setVar, err := types.ParseSetValue(ft.GetElems(), col.Value.(uint64))
+		elements := ft.GetElems()
+		number := col.Value.(uint64)
+		setVar, err := types.ParseSetValue(elements, number)
 		if err != nil {
+			log.Info("avro encoder parse set value failed",
+				zap.Strings("elements", elements), zap.Uint64("number", number), zap.Error(err))
 			return nil, "", cerror.WrapError(cerror.ErrAvroEncodeFailed, err)
 		}
 		return setVar.Name, "string", nil
@@ -935,13 +942,14 @@ func columnToAvroData(
 		if v, ok := col.Value.(string); ok {
 			n, err := strconv.ParseInt(v, 10, 32)
 			if err != nil {
+				log.Info("avro encoder parse year value failed", zap.String("value", v), zap.Error(err))
 				return nil, "", cerror.WrapError(cerror.ErrAvroEncodeFailed, err)
 			}
 			return int32(n), "int", nil
 		}
 		return int32(col.Value.(int64)), "int", nil
 	default:
-		log.Error("unknown mysql type", zap.Any("mysqlType", col.Type))
+		log.Error("unknown mysql type", zap.Any("value", col.Value), zap.Any("mysqlType", col.Type))
 		return nil, "", cerror.ErrAvroEncodeFailed.GenWithStack("unknown mysql type")
 	}
 }