sink(ticdc): fix incorrect default field (#12038)

close #12037

Author: wk989898

cdc/entry/mounter_test.go

@@ -1629,13 +1629,13 @@ func TestNewDMRowChange(t *testing.T) {
 		cdcTableInfo := model.WrapTableInfo(0, "test", 0, originTI)
 		cols := []*model.Column{
 			{
-				Name: "id", Type: 3, Charset: "binary", Flag: 65, Value: 1, Default: nil,
+				Name: "id", Type: 3, Charset: "binary", Flag: 65, Value: 1,
 			},
 			{
-				Name: "a1", Type: 3, Charset: "binary", Flag: 51, Value: 1, Default: nil,
+				Name: "a1", Type: 3, Charset: "binary", Flag: 51, Value: 1,
 			},
 			{
-				Name: "a3", Type: 3, Charset: "binary", Flag: 51, Value: 2, Default: nil,
+				Name: "a3", Type: 3, Charset: "binary", Flag: 51, Value: 2,
 			},
 		}
 		recoveredTI := model.BuildTiDBTableInfo(cdcTableInfo.TableName.Table, cols, cdcTableInfo.IndexColumnsOffset)

cdc/model/codec/codec.go

@@ -223,7 +223,6 @@ func columnFromV1(c *codecv1.Column) *model.Column {
 		Charset:          c.Charset,
 		Flag:             c.Flag,
 		Value:            c.Value,
-		Default:          c.Default,
 		ApproximateBytes: c.ApproximateBytes,
 	}
 }

cdc/model/schema_storage.go

@@ -21,7 +21,6 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tidb/pkg/table/tables"
-	datumTypes "github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/rowcodec"
 	"go.uber.org/zap"
 )
@@ -497,13 +496,3 @@ func (ti *TableInfo) GetPrimaryKeyColumnNames() []string {
 	}
 	return result
 }
-
-// GetColumnDefaultValue returns the default definition of a column.
-func GetColumnDefaultValue(col *model.ColumnInfo) interface{} {
-	defaultValue := col.GetDefaultValue()
-	if defaultValue == nil {
-		defaultValue = col.GetOriginDefaultValue()
-	}
-	defaultDatum := datumTypes.NewDatum(defaultValue)
-	return defaultDatum.GetValue()
-}

cdc/model/sink.go

@@ -480,7 +480,6 @@ func columnData2Column(col *ColumnData, tableInfo *TableInfo) *Column {
 		Collation: colInfo.GetCollate(),
 		Flag:      *tableInfo.ColumnsFlag[colID],
 		Value:     col.Value,
-		Default:   GetColumnDefaultValue(colInfo),
 	}
 }
 
@@ -652,7 +651,6 @@ type Column struct {
 	Collation string         `msg:"collation"`
 	Flag      ColumnFlagType `msg:"-"`
 	Value     interface{}    `msg:"-"`
-	Default   interface{}    `msg:"-"`
 
 	// ApproximateBytes is approximate bytes consumed by the column.
 	ApproximateBytes int `msg:"-"`
@@ -1360,7 +1358,7 @@ func (x ColumnDataX) GetFlag() ColumnFlagType {
 
 // GetDefaultValue return default value.
 func (x ColumnDataX) GetDefaultValue() interface{} {
-	return GetColumnDefaultValue(x.info)
+	return x.info.GetDefaultValue()
 }
 
 // GetColumnInfo returns column info.
@@ -1388,7 +1386,6 @@ func Columns2ColumnDataForTest(columns []*Column) ([]*ColumnData, *TableInfo) {
 		info.Columns[i].SetType(column.Type)
 		info.Columns[i].SetCharset(column.Charset)
 		info.Columns[i].SetCollate(column.Collation)
-		info.Columns[i].DefaultValue = column.Default
 
 		info.ColumnsFlag[columnID] = new(ColumnFlagType)
 		*info.ColumnsFlag[columnID] = column.Flag

pkg/sink/cloudstorage/table_definition.go

@@ -77,7 +77,7 @@ func (t *TableCol) FromTiColumnInfo(col *timodel.ColumnInfo, outputColumnID bool
 	if mysql.HasNotNullFlag(col.GetFlag()) {
 		t.Nullable = "false"
 	}
-	t.Default = model.GetColumnDefaultValue(col)
+	t.Default = col.GetDefaultValue()
 
 	switch col.GetType() {
 	case mysql.TypeTimestamp, mysql.TypeDatetime, mysql.TypeDuration:

pkg/sink/codec/debezium/codec.go

@@ -505,7 +505,10 @@ func (c *dbzCodec) writeDebeziumFieldValue(
 	col model.ColumnDataX,
 	ft *types.FieldType,
 ) error {
-	value := getValue(col)
+	value := col.Value
+	if value == nil {
+		value = col.GetDefaultValue()
+	}
 	if value == nil {
 		writer.WriteNullField(col.GetName())
 		return nil

pkg/sink/codec/debezium/helper.go

@@ -43,9 +43,6 @@ func (v *visiter) Leave(n ast.Node) (node ast.Node, ok bool) {
 	switch col := n.(type) {
 	case *ast.ColumnDef:
 		c := v.columnsMap[col.Name.Name]
-		if col.Options != nil {
-			parseOptions(col.Options, c)
-		}
 		if col.Tp != nil {
 			parseType(c, col)
 		}
@@ -67,16 +64,13 @@ func extractValue(expr ast.ExprNode) any {
 func parseType(c *timodel.ColumnInfo, col *ast.ColumnDef) {
 	ft := col.Tp
 	switch ft.GetType() {
-	case mysql.TypeDatetime, mysql.TypeDuration, mysql.TypeTimestamp:
-		c.SetDecimal(ft.GetDecimal())
-		if c.OriginDefaultValue != nil {
-			c.SetDefaultValue(c.OriginDefaultValue)
-		}
-	case mysql.TypeYear:
-		c.SetFlen(ft.GetFlen())
-		if c.OriginDefaultValue != nil {
-			c.SetDefaultValue(c.OriginDefaultValue)
+	case mysql.TypeDatetime, mysql.TypeDuration, mysql.TypeTimestamp, mysql.TypeYear:
+		if ft.GetType() == mysql.TypeYear {
+			c.SetFlen(ft.GetFlen())
+		} else {
+			c.SetDecimal(ft.GetDecimal())
 		}
+		parseOptions(col.Options, c)
 	default:
 	}
 }
@@ -89,7 +83,7 @@ func parseOptions(options []*ast.ColumnOption, c *timodel.ColumnInfo) {
 			if defaultValue == nil {
 				continue
 			}
-			if err := c.SetOriginDefaultValue(defaultValue); err != nil {
+			if err := c.SetDefaultValue(defaultValue); err != nil {
 				log.Error("failed to set default value")
 			}
 		}
@@ -243,13 +237,6 @@ func getBitFromUint64(n int, v uint64) []byte {
 	return buf[:numBytes]
 }
 
-func getValue(col model.ColumnDataX) any {
-	if col.Value == nil {
-		return col.GetDefaultValue()
-	}
-	return col.Value
-}
-
 func getDBTableName(e *model.DDLEvent) (string, string) {
 	if e.TableInfo == nil {
 		return "", ""

pkg/sink/codec/debezium/helper_test.go

@@ -20,7 +20,6 @@ import (
 	pmodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/types"
-	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/stretchr/testify/require"
 )
 
@@ -59,18 +58,6 @@ func TestGetColumns(t *testing.T) {
 	require.Equal(t, columnInfos[4].Comment, "")
 }
 
-func TestGetValue(t *testing.T) {
-	column := &model.Column{
-		Default: 1,
-	}
-	data := model.Column2ColumnDataXForTest(column)
-	v := getValue(data)
-	require.Equal(t, v, int64(1))
-	data.Value = 2
-	v = getValue(data)
-	require.Equal(t, v, 2)
-}
-
 func TestGetSchemaTopicName(t *testing.T) {
 	namespace := "default"
 	schema := "1A.B"

pkg/sink/codec/simple/avro.go

@@ -109,7 +109,7 @@ func newTableSchemaMap(tableInfo *model.TableInfo) interface{} {
 			"nullable": !mysql.HasNotNullFlag(col.GetFlag()),
 			"default":  nil,
 		}
-		defaultValue := model.GetColumnDefaultValue(col)
+		defaultValue := col.GetDefaultValue()
 		if defaultValue != nil {
 			// according to TiDB source code, the default value is converted to string if not nil.
 			column["default"] = map[string]interface{}{

pkg/sink/codec/simple/encoder_test.go

@@ -1487,15 +1487,15 @@ func TestEncodeLargeEventsNormal(t *testing.T) {
 
 			obtainedDefaultValues := make(map[string]interface{}, len(obtainedDDL.TableInfo.Columns))
 			for _, col := range obtainedDDL.TableInfo.Columns {
-				obtainedDefaultValues[col.Name.O] = model.GetColumnDefaultValue(col)
+				obtainedDefaultValues[col.Name.O] = col.GetDefaultValue()
 				switch col.GetType() {
 				case mysql.TypeFloat, mysql.TypeDouble:
 					require.Equal(t, 0, col.GetDecimal())
 				default:
 				}
 			}
 			for _, col := range ddlEvent.TableInfo.Columns {
-				expected := model.GetColumnDefaultValue(col)
+				expected := col.GetDefaultValue()
 				obtained := obtainedDefaultValues[col.Name.O]
 				require.Equal(t, expected, obtained)
 			}

pkg/sink/codec/simple/message.go

@@ -145,7 +145,7 @@ func newColumnSchema(col *timodel.ColumnInfo) *columnSchema {
 		tp.Decimal = col.GetDecimal()
 	}
 
-	defaultValue := model.GetColumnDefaultValue(col)
+	defaultValue := col.GetDefaultValue()
 	if defaultValue != nil && col.GetType() == mysql.TypeBit {
 		defaultValue = common.MustBinaryLiteralToInt([]byte(defaultValue.(string)))
 	}