From 922457a3d6a1747be2f0d4ec67ad29fb290ffe86 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 29 May 2026 09:00:50 +0000 Subject: [PATCH 1/8] init Signed-off-by: wk989898 --- pkg/config/sink.go | 1 + pkg/sink/codec/avro/arvo.go | 89 ++++++++- pkg/sink/codec/avro/avro_test.go | 188 ++++++++++++++++++ pkg/sink/codec/avro/decoder.go | 186 +++++++++++------ pkg/sink/codec/avro/helper.go | 4 + pkg/sink/codec/common/config.go | 7 + pkg/sink/codec/common/config_test.go | 51 +++++ .../avro_basic/data/data.sql | 11 + tests/integration_tests/avro_basic/run.sh | 3 +- 9 files changed, 478 insertions(+), 62 deletions(-) create mode 100644 pkg/sink/codec/common/config_test.go diff --git a/pkg/config/sink.go b/pkg/config/sink.go index d939b5c281..b25aa133e9 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -436,6 +436,7 @@ type CodecConfig struct { AvroEnableWatermark *bool `toml:"avro-enable-watermark" json:"avro-enable-watermark"` AvroDecimalHandlingMode *string `toml:"avro-decimal-handling-mode" json:"avro-decimal-handling-mode,omitempty"` AvroBigintUnsignedHandlingMode *string `toml:"avro-bigint-unsigned-handling-mode" json:"avro-bigint-unsigned-handling-mode,omitempty"` + AvroIncludeBeforeValue *bool `toml:"avro-include-before-value" json:"avro-include-before-value,omitempty"` EncodingFormat *string `toml:"encoding-format" json:"encoding-format,omitempty"` OutputRowKey *bool `toml:"output-row-key" json:"output-row-key,omitempty"` } diff --git a/pkg/sink/codec/avro/arvo.go b/pkg/sink/codec/avro/arvo.go index 515de064d9..b24267d9d0 100644 --- a/pkg/sink/codec/avro/arvo.go +++ b/pkg/sink/codec/avro/arvo.go @@ -126,7 +126,7 @@ func (a *BatchEncoder) encodeKey(ctx context.Context, topic string, e *event.Row } func (a *BatchEncoder) encodeValue(ctx context.Context, topic string, e *event.RowEvent) ([]byte, error) { - if e.IsDelete() { + if e.IsDelete() && !a.config.AvroIncludeBeforeValue { if !a.config.EnableTiDBExtension || !a.config.AvroEnableWatermark { return nil, nil } @@ -139,7 +139,11 @@ func (a *BatchEncoder) encodeValue(ctx context.Context, topic string, e *event.R } return buf.Bytes(), nil } - length := e.GetRows().Len() + row := e.GetRows() + if e.IsDelete() { + row = e.GetPreRows() + } + length := row.Len() if length == 0 { return nil, nil } @@ -148,7 +152,7 @@ func (a *BatchEncoder) encodeValue(ctx context.Context, topic string, e *event.R index[i] = i } input := &avroEncodeInput{ - row: e.GetRows(), + row: row, colInfos: e.TableInfo.GetColumns(), index: index, columnselector: e.ColumnSelector, @@ -163,6 +167,15 @@ func (a *BatchEncoder) encodeValue(ctx context.Context, topic string, e *event.R log.Error("avro: converting input to native failed", zap.Error(err)) return nil, errors.Trace(err) } + if a.config.AvroIncludeBeforeValue { + native[ticdcBefore] = goavro.Union("null", nil) + if e.IsUpdate() || e.IsDelete() { + native, err = a.nativeValueWithBeforeValue(native, &targetTableName, e) + if err != nil { + return nil, errors.Trace(err) + } + } + } if a.config.EnableTiDBExtension { native = a.nativeValueWithExtension(native, e) } @@ -193,13 +206,55 @@ func (a *BatchEncoder) nativeValueWithExtension( native[tidbPhysicalTime] = oracle.ExtractPhysical(e.CommitTs) if a.config.EnableRowChecksum && e.Checksum != nil { - native[tidbRowLevelChecksum] = strconv.FormatUint(uint64(e.Checksum.Current), 10) + checksum := e.Checksum.Current + if e.IsDelete() { + checksum = e.Checksum.Previous + } + native[tidbRowLevelChecksum] = strconv.FormatUint(uint64(checksum), 10) native[tidbCorrupted] = e.Checksum.Corrupted native[tidbChecksumVersion] = e.Checksum.Version } return native } +func beforeValueRecordName(tableName *commonType.TableName) string { + return common.SanitizeName(tableName.Table) + "_before" +} + +func (a *BatchEncoder) beforeValueRecordFullName(tableName *commonType.TableName) string { + namespace := getAvroNamespace(a.keyspace, tableName.Schema) + if namespace == "" { + return beforeValueRecordName(tableName) + } + return namespace + "." + beforeValueRecordName(tableName) +} + +func (a *BatchEncoder) nativeValueWithBeforeValue( + native map[string]any, + tableName *commonType.TableName, + e *event.RowEvent, +) (map[string]any, error) { + row := e.GetPreRows() + length := row.Len() + index := make([]int, length) + for i := range length { + index[i] = i + } + input := &avroEncodeInput{ + row: row, + colInfos: e.TableInfo.GetColumns(), + index: index, + columnselector: e.ColumnSelector, + } + before, err := a.columns2AvroData(input) + if err != nil { + log.Error("avro: converting before value to native failed", zap.Error(err)) + return nil, errors.Trace(err) + } + native[ticdcBefore] = goavro.Union(a.beforeValueRecordFullName(tableName), before) + return native, nil +} + func routedTableName(tableInfo *commonType.TableInfo) commonType.TableName { tableName := tableInfo.TableName tableName.Schema = tableInfo.GetTargetSchemaName() @@ -250,6 +305,25 @@ func (a *BatchEncoder) schemaWithExtension( return top } +func (a *BatchEncoder) schemaWithBeforeValue( + top *avroSchemaTop, + tableName *commonType.TableName, + input *avroEncodeInput, +) (*avroSchemaTop, error) { + beforeValue, err := a.columns2AvroSchema(tableName, input) + if err != nil { + return nil, err + } + beforeValue.Name = beforeValueRecordName(tableName) + + top.Fields = append(top.Fields, map[string]any{ + "name": ticdcBefore, + "type": []any{"null", beforeValue}, + "default": nil, + }) + return top, nil +} + func (a *BatchEncoder) getDefaultValue(col *model.ColumnInfo) (any, error) { defaultVal := col.GetDefaultValue() if defaultVal == nil { @@ -418,6 +492,13 @@ func (a *BatchEncoder) value2AvroSchema( return "", err } + if a.config.AvroIncludeBeforeValue { + top, err = a.schemaWithBeforeValue(top, tableName, input) + if err != nil { + return "", err + } + } + if a.config.EnableTiDBExtension { top = a.schemaWithExtension(top) } diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index 798bc611d8..081be273a5 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -18,13 +18,68 @@ import ( "testing" "github.com/linkedin/goavro/v2" + "github.com/pingcap/ticdc/downstreamadapter/sink/columnselector" commonType "github.com/pingcap/ticdc/pkg/common" + commonEvent "github.com/pingcap/ticdc/pkg/common/event" "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/integrity" "github.com/pingcap/ticdc/pkg/sink/codec/common" "github.com/pingcap/ticdc/pkg/uuid" + timodel "github.com/pingcap/tidb/pkg/meta/model" + "github.com/pingcap/tidb/pkg/parser/ast" + "github.com/pingcap/tidb/pkg/parser/mysql" + parserTypes "github.com/pingcap/tidb/pkg/parser/types" + "github.com/pingcap/tidb/pkg/util/chunk" "github.com/stretchr/testify/require" ) +func newAvroRowEventForTest( + tableInfo *commonType.TableInfo, + commitTs uint64, + preRow chunk.Row, + row chunk.Row, +) *commonEvent.RowEvent { + return &commonEvent.RowEvent{ + PhysicalTableID: tableInfo.TableName.TableID, + StartTs: commitTs, + CommitTs: commitTs, + TableInfo: tableInfo, + Event: commonEvent.RowChange{ + PreRow: preRow, + Row: row, + }, + ColumnSelector: columnselector.NewDefaultColumnSelector(), + } +} + +func newAvroTableInfoForTest() *commonType.TableInfo { + idFieldType := parserTypes.NewFieldType(mysql.TypeLong) + idFieldType.SetFlag(mysql.PriKeyFlag | mysql.NotNullFlag) + ageFieldType := parserTypes.NewFieldType(mysql.TypeLong) + + return commonType.WrapTableInfo("test", &timodel.TableInfo{ + ID: 20, + Name: ast.NewCIStr("person"), + UpdateTS: 100, + Columns: []*timodel.ColumnInfo{ + { + ID: 1, + Name: ast.NewCIStr("id"), + FieldType: *idFieldType, + State: timodel.StatePublic, + Offset: 0, + }, + { + ID: 2, + Name: ast.NewCIStr("age"), + FieldType: *ageFieldType, + State: timodel.StatePublic, + Offset: 1, + }, + }, + }) +} + func TestAvroEncode4EnableChecksum(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolAvro) codecConfig.EnableTiDBExtension = true @@ -68,6 +123,50 @@ func TestAvroEncode4EnableChecksum(t *testing.T) { require.True(t, found) } +func TestAvroEncodeDeleteChecksum(t *testing.T) { + codecConfig := common.NewConfig(config.ProtocolAvro) + codecConfig.EnableTiDBExtension = true + codecConfig.EnableRowChecksum = true + codecConfig.AvroIncludeBeforeValue = true + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) + defer TeardownEncoderAndSchemaRegistry4Testing() + require.NoError(t, err) + require.NotNil(t, encoder) + + tableInfo := newAvroTableInfoForTest() + event := newAvroRowEventForTest( + tableInfo, + 1024, + chunk.MutRowFromValues(int64(1), int64(18)).ToRow(), + chunk.Row{}, + ) + event.Checksum = &integrity.Checksum{ + Current: 11, + Previous: 22, + } + + topic := "default" + bin, err := encoder.encodeValue(ctx, topic, event) + require.NoError(t, err) + + cid, data, err := extractConfluentSchemaIDAndBinaryData(bin) + require.NoError(t, err) + + avroValueCodec, err := encoder.schemaM.Lookup(ctx, topic, schemaID{confluentSchemaID: cid}) + require.NoError(t, err) + + res, _, err := avroValueCodec.NativeFromBinary(data) + require.NoError(t, err) + m, ok := res.(map[string]interface{}) + require.True(t, ok) + require.Equal(t, deleteOperation, m[tidbOp]) + require.Equal(t, "22", m[tidbRowLevelChecksum]) +} + func TestAvroEncode(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolAvro) codecConfig.EnableTiDBExtension = true @@ -124,6 +223,95 @@ func TestAvroEncode(t *testing.T) { } } +func TestAvroEncodeIncludeBeforeValue(t *testing.T) { + codecConfig := common.NewConfig(config.ProtocolAvro) + codecConfig.EnableTiDBExtension = true + codecConfig.AvroIncludeBeforeValue = true + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) + defer TeardownEncoderAndSchemaRegistry4Testing() + require.NoError(t, err) + require.NotNil(t, encoder) + + tableInfo := newAvroTableInfoForTest() + beforeRow := chunk.MutRowFromValues(int64(1), int64(18)).ToRow() + afterRow := chunk.MutRowFromValues(int64(1), int64(20)).ToRow() + + testCases := []struct { + name string + event *commonEvent.RowEvent + op string + hasBefore bool + }{ + { + name: "insert", + event: newAvroRowEventForTest(tableInfo, 1024, chunk.Row{}, beforeRow), + op: insertOperation, + }, + { + name: "update", + event: newAvroRowEventForTest(tableInfo, 1025, beforeRow, afterRow), + op: updateOperation, + hasBefore: true, + }, + { + name: "delete", + event: newAvroRowEventForTest(tableInfo, 1026, afterRow, chunk.Row{}), + op: deleteOperation, + hasBefore: true, + }, + } + + topic := "default" + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + bin, err := encoder.encodeValue(ctx, topic, tc.event) + require.NoError(t, err) + require.NotNil(t, bin) + + cid, data, err := extractConfluentSchemaIDAndBinaryData(bin) + require.NoError(t, err) + + avroValueCodec, err := encoder.schemaM.Lookup(ctx, topic, schemaID{confluentSchemaID: cid}) + require.NoError(t, err) + + res, _, err := avroValueCodec.NativeFromBinary(data) + require.NoError(t, err) + require.NotNil(t, res) + + m, ok := res.(map[string]interface{}) + require.True(t, ok) + require.Equal(t, int64(tc.event.CommitTs), m[tidbCommitTs]) + require.Equal(t, tc.op, m[tidbOp]) + if tc.hasBefore { + require.NotNil(t, m[ticdcBefore]) + } else { + require.Nil(t, m[ticdcBefore]) + } + + key, err := encoder.encodeKey(ctx, topic, tc.event) + require.NoError(t, err) + decoder := NewDecoder(codecConfig, 0, encoder.schemaM, topic, nil) + decoder.AddKeyValue(key, bin) + + messageType, exist := decoder.HasNext() + require.True(t, exist) + require.Equal(t, common.MessageTypeRow, messageType) + + decoded := decoder.NextDMLEvent() + require.NotNil(t, decoded) + require.Equal(t, tc.event.CommitTs, decoded.CommitTs) + + decodedRow, ok := decoded.GetNextRow() + require.True(t, ok) + common.CompareRow(t, tc.event.Event, tc.event.TableInfo, decodedRow, decoded.TableInfo) + }) + } +} + func TestAvroEncodeDeleteEventUsesPreRowForKey(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolAvro) codecConfig.EnableTiDBExtension = true diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go index 8d6ae80506..b30977eb8c 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -127,13 +127,14 @@ func (d *decoder) NextDMLEvent() *commonEvent.DMLEvent { log.Panic("decode key failed", zap.Error(err)) } - // for the delete event, only have key part, it holds primary key or the unique key columns. - // for the insert / update, extract the value part, it holds all columns. - isDelete := len(d.value) == 0 || d.isDeleteValue() + isDeleteValue := d.isDeleteValue() + hasValue := len(d.value) != 0 && !isDeleteValue + isDelete := !hasValue deleteCommitTs := uint64(0) - if isDelete { - // delete event only have key part, treat it as the value part also. - if d.isDeleteValue() { + if !hasValue { + // Legacy delete event only has the key payload or a delete marker value. + // It can only be decoded as a delete row with key columns in PreRow. + if isDeleteValue { deleteCommitTs = d.decodeDeleteCommitTs() } valueMap = keyMap @@ -143,9 +144,12 @@ func (d *decoder) NextDMLEvent() *commonEvent.DMLEvent { if err != nil { log.Panic("decode value failed", zap.Error(err)) } + if d.config.AvroIncludeBeforeValue { + isDelete = valueMap[tidbOp] == deleteOperation + } } - event, err := assembleEvent(keyMap, valueMap, valueSchema, isDelete) + event, err := assembleEvent(keyMap, valueMap, valueSchema, isDelete, hasValue) if err != nil { log.Panic("assemble event failed", zap.Error(err)) } @@ -154,22 +158,23 @@ func (d *decoder) NextDMLEvent() *commonEvent.DMLEvent { event.CommitTs = deleteCommitTs } - // Delete event only has Primary Key Columns, but the checksum is calculated based on the whole row columns, - // checksum verification cannot be done here, so skip it. - if isDelete { + if !hasValue { return event } expectedChecksum, found := extractExpectedChecksum(valueMap) corrupted := isCorrupted(valueMap) if found { - event.Checksum = []*integrity.Checksum{{ - Current: uint32(expectedChecksum), - Corrupted: corrupted, - }} + checksum := &integrity.Checksum{Corrupted: corrupted} + if isDelete { + checksum.Previous = uint32(expectedChecksum) + } else { + checksum.Current = uint32(expectedChecksum) + } + event.Checksum = []*integrity.Checksum{checksum} } - if isCorrupted(valueMap) { + if corrupted { log.Warn("row data is corrupted", zap.String("topic", d.topic), zap.Uint64("checksum", expectedChecksum)) for _, col := range event.TableInfo.GetColumns() { @@ -210,15 +215,115 @@ func (d *decoder) decodeDeleteCommitTs() uint64 { // assembleEvent return a row changed event // keyMap hold primary key or unique key columns -// valueMap hold all columns information +// valueMap holds all columns for insert/update and before-value delete. +// For legacy delete, valueMap is keyMap and only contains handle columns. // schema is corresponding to the valueMap, it can be used to decode the valueMap to construct columns. func assembleEvent( - keyMap, valueMap, schema map[string]interface{}, isDelete bool, + keyMap, valueMap, schema map[string]interface{}, isDelete bool, hasValue bool, ) (*commonEvent.DMLEvent, error) { fields, ok := schema["fields"].([]interface{}) if !ok { - return nil, errors.New("schema fields should be a map") + return nil, errors.ErrCodecDecode.GenWithStack("schema fields should be a map") + } + + columns, data, err := avroData2Columns(valueMap, fields) + if err != nil { + return nil, errors.Trace(err) + } + + beforeMap, hasBefore, err := extractBeforeValueMap(valueMap) + if err != nil { + return nil, errors.Trace(err) + } + var beforeData map[string]interface{} + if hasBefore { + _, beforeData, err = avroData2Columns(beforeMap, fields) + if err != nil { + return nil, errors.Trace(err) + } + } + + // "namespace.schema" + namespace := schema["namespace"].(string) + schemaName := strings.Split(namespace, ".")[1] + tableName := schema["name"].(string) + + var commitTs int64 + if hasValue { + o, ok := valueMap[tidbCommitTs] + if !ok { + return nil, errors.ErrCodecDecode.GenWithStack("commit ts not found") + } + commitTs = o.(int64) + } + + event := new(commonEvent.DMLEvent) + event.TableInfo = queryTableInfo(schemaName, tableName, columns, keyMap) + event.StartTs = uint64(commitTs) + event.CommitTs = uint64(commitTs) + event.PhysicalTableID = event.TableInfo.TableName.TableID + event.Rows = chunk.NewChunkFromPoolWithCapacity(event.TableInfo.GetFieldSlice(), chunk.InitialCapacity) + event.AddPostFlushFunc(func() { + event.Rows.Destroy(chunk.InitialCapacity, event.TableInfo.GetFieldSlice()) + }) + event.Length++ + if isDelete { + if hasValue { + if !hasBefore { + return nil, errors.ErrCodecDecode.GenWithStack("before value not found for delete event") + } + common.AppendRow2Chunk(beforeData, event.TableInfo.GetColumns(), event.Rows) + } else { + common.AppendRow2Chunk(data, event.TableInfo.GetColumns(), event.Rows) + } + event.RowTypes = append(event.RowTypes, commonType.RowTypeDelete) + } else if hasBefore { + common.AppendRow2Chunk(beforeData, event.TableInfo.GetColumns(), event.Rows) + common.AppendRow2Chunk(data, event.TableInfo.GetColumns(), event.Rows) + event.RowTypes = append(event.RowTypes, commonType.RowTypeUpdate) + } else { + common.AppendRow2Chunk(data, event.TableInfo.GetColumns(), event.Rows) + event.RowTypes = append(event.RowTypes, commonType.RowTypeInsert) + } + return event, nil +} + +func isAvroExtensionField(name string) bool { + switch name { + case tidbOp, tidbCommitTs, tidbPhysicalTime, tidbRowLevelChecksum, + tidbChecksumVersion, tidbCorrupted, ticdcBefore: + return true + default: + return false + } +} + +func extractBeforeValueMap(valueMap map[string]interface{}) (map[string]interface{}, bool, error) { + rawBefore, ok := valueMap[ticdcBefore] + if !ok || rawBefore == nil { + return nil, false, nil + } + + beforeUnion, ok := rawBefore.(map[string]interface{}) + if !ok { + return nil, false, errors.ErrCodecDecode.GenWithStack("before value should be a map") } + for unionName, value := range beforeUnion { + if unionName == "null" || value == nil { + return nil, false, nil + } + before, ok := value.(map[string]interface{}) + if !ok { + return nil, false, errors.ErrCodecDecode.GenWithStack("before record should be a map") + } + return before, true, nil + } + return nil, false, nil +} + +func avroData2Columns( + valueMap map[string]interface{}, fields []interface{}, +) ([]*timodel.ColumnInfo, map[string]interface{}, error) { columns := make([]*timodel.ColumnInfo, 0, len(valueMap)) data := make(map[string]interface{}, 0) // fields is ordered by the column id, so iterate over it to build columns @@ -226,12 +331,11 @@ func assembleEvent( for idx, item := range fields { field, ok := item.(map[string]interface{}) if !ok { - return nil, errors.New("schema field should be a map") + return nil, nil, errors.ErrCodecDecode.GenWithStack("schema field should be a map") } - // `tidbOp` is the first extension field in the schema, - // it's not real columns, so break here. + // Extension fields are not real columns, so break here. colName := field["name"].(string) - if colName == tidbOp { + if isAvroExtensionField(colName) { break } // query the field to get `tidbType`, and get the mysql type from it. @@ -255,11 +359,11 @@ func assembleEvent( flag := flagFromTiDBType(tidbType) value, ok := valueMap[colName] if !ok { - return nil, errors.New("value not found") + return nil, nil, errors.ErrCodecDecode.GenWithStack("value not found") } value, err := getColumnValue(value, holder, mysqlType, flag) if err != nil { - return nil, errors.Trace(err) + return nil, nil, errors.Trace(err) } data[colName] = value @@ -272,39 +376,7 @@ func assembleEvent( tiCol.SetFlag(flag) columns = append(columns, tiCol) } - - // "namespace.schema" - namespace := schema["namespace"].(string) - schemaName := strings.Split(namespace, ".")[1] - tableName := schema["name"].(string) - - var commitTs int64 - if !isDelete { - o, ok := valueMap[tidbCommitTs] - if !ok { - return nil, errors.New("commit ts not found") - } - commitTs = o.(int64) - } - - event := new(commonEvent.DMLEvent) - event.TableInfo = queryTableInfo(schemaName, tableName, columns, keyMap) - event.StartTs = uint64(commitTs) - event.CommitTs = uint64(commitTs) - event.PhysicalTableID = event.TableInfo.TableName.TableID - event.Rows = chunk.NewChunkFromPoolWithCapacity(event.TableInfo.GetFieldSlice(), chunk.InitialCapacity) - event.AddPostFlushFunc(func() { - event.Rows.Destroy(chunk.InitialCapacity, event.TableInfo.GetFieldSlice()) - }) - event.Length++ - common.AppendRow2Chunk(data, event.TableInfo.GetColumns(), event.Rows) - - rowType := commonType.RowTypeInsert - if isDelete { - rowType = commonType.RowTypeDelete - } - event.RowTypes = append(event.RowTypes, rowType) - return event, nil + return columns, data, nil } func queryTableInfo(schemaName, tableName string, columns []*timodel.ColumnInfo, keyMap map[string]interface{}) *commonType.TableInfo { diff --git a/pkg/sink/codec/avro/helper.go b/pkg/sink/codec/avro/helper.go index 8c3d861cef..4b0b3d409e 100644 --- a/pkg/sink/codec/avro/helper.go +++ b/pkg/sink/codec/avro/helper.go @@ -33,6 +33,7 @@ const ( tidbOp = "_tidb_op" tidbCommitTs = "_tidb_commit_ts" tidbPhysicalTime = "_tidb_commit_physical_time" + ticdcBefore = "_ticdc_before" // row level checksum related fields tidbRowLevelChecksum = "_tidb_row_level_checksum" @@ -43,6 +44,7 @@ const ( const ( insertOperation = "c" updateOperation = "u" + deleteOperation = "d" ) const ( @@ -149,6 +151,8 @@ func getOperation(e *commonEvent.RowEvent) string { return insertOperation } else if e.IsUpdate() { return updateOperation + } else if e.IsDelete() { + return deleteOperation } return "" } diff --git a/pkg/sink/codec/common/config.go b/pkg/sink/codec/common/config.go index 83486f3da3..ab85570fc3 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -60,6 +60,7 @@ type Config struct { AvroDecimalHandlingMode string AvroBigintUnsignedHandlingMode string AvroGlueSchemaRegistry *config.GlueSchemaRegistryConfig + AvroIncludeBeforeValue bool // EnableWatermarkEvent set to true, avro encode DDL and checkpoint event // and send to the downstream kafka, they cannot be consumed by the confluent official consumer // and would cause error, so this is only used for ticdc internal testing purpose, should not be @@ -124,6 +125,7 @@ func NewConfig(protocol config.Protocol) *Config { AvroConfluentSchemaRegistry: "", AvroDecimalHandlingMode: "precise", AvroBigintUnsignedHandlingMode: "long", + AvroIncludeBeforeValue: false, AvroEnableWatermark: false, OnlyOutputUpdatedColumns: false, @@ -167,6 +169,7 @@ type urlConfig struct { MaxMessageBytes *int `form:"max-message-bytes"` AvroDecimalHandlingMode *string `form:"avro-decimal-handling-mode"` AvroBigintUnsignedHandlingMode *string `form:"avro-bigint-unsigned-handling-mode"` + AvroIncludeBeforeValue *bool `form:"avro-include-before-value"` // AvroEnableWatermark is the option for enabling watermark in avro protocol // only used for internal testing, do not set this in the production environment since the @@ -224,6 +227,9 @@ func (c *Config) Apply(sinkURI *url.URL, sinkConfig *config.SinkConfig) error { *urlParameter.AvroBigintUnsignedHandlingMode != "" { c.AvroBigintUnsignedHandlingMode = *urlParameter.AvroBigintUnsignedHandlingMode } + if urlParameter.AvroIncludeBeforeValue != nil && c.Protocol == config.ProtocolAvro { + c.AvroIncludeBeforeValue = *urlParameter.AvroIncludeBeforeValue + } if urlParameter.AvroEnableWatermark != nil { if c.EnableTiDBExtension && c.Protocol == config.ProtocolAvro { c.AvroEnableWatermark = *urlParameter.AvroEnableWatermark @@ -325,6 +331,7 @@ func mergeConfig( dest.AvroEnableWatermark = codecConfig.AvroEnableWatermark dest.AvroDecimalHandlingMode = codecConfig.AvroDecimalHandlingMode dest.AvroBigintUnsignedHandlingMode = codecConfig.AvroBigintUnsignedHandlingMode + dest.AvroIncludeBeforeValue = codecConfig.AvroIncludeBeforeValue dest.EncodingFormatType = codecConfig.EncodingFormat } } diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go new file mode 100644 index 0000000000..80112b3dbd --- /dev/null +++ b/pkg/sink/codec/common/config_test.go @@ -0,0 +1,51 @@ +// Copyright 2026 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package common + +import ( + "net/url" + "testing" + + "github.com/pingcap/ticdc/pkg/config" + "github.com/pingcap/ticdc/pkg/util" + "github.com/stretchr/testify/require" +) + +func TestAvroIncludeBeforeValueConfig(t *testing.T) { + cfg := NewConfig(config.ProtocolAvro) + require.False(t, cfg.AvroIncludeBeforeValue) + + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=avro&avro-include-before-value=true") + require.NoError(t, err) + + err = cfg.Apply(sinkURI, &config.SinkConfig{}) + require.NoError(t, err) + require.True(t, cfg.AvroIncludeBeforeValue) +} + +func TestAvroIncludeBeforeValueConfigFile(t *testing.T) { + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=avro") + require.NoError(t, err) + + cfg := NewConfig(config.ProtocolAvro) + err = cfg.Apply(sinkURI, &config.SinkConfig{ + KafkaConfig: &config.KafkaConfig{ + CodecConfig: &config.CodecConfig{ + AvroIncludeBeforeValue: util.AddressOf(true), + }, + }, + }) + require.NoError(t, err) + require.True(t, cfg.AvroIncludeBeforeValue) +} diff --git a/tests/integration_tests/avro_basic/data/data.sql b/tests/integration_tests/avro_basic/data/data.sql index 53b2b78256..dfc8b6b981 100644 --- a/tests/integration_tests/avro_basic/data/data.sql +++ b/tests/integration_tests/avro_basic/data/data.sql @@ -186,6 +186,17 @@ insert into t(c_tinyint, c_mediumint, c_int, c_bigint, a) values (4, 5, 6, 7, 8) alter table t modify c_mediumint varchar(10) null; insert into t(c_tinyint, c_mediumint, c_int, c_bigint, a) values (5, "234", 6, 7, 8); +create table t2( + id int primary key, + v int, + note varchar(32) +); + +insert into t2 values(1, 10, 'before-delete'); +insert into t2 values(2, 20, 'keep'); +update t2 set v = 11, note = 'updated-before-delete' where id = 1; +delete from t2 where id = 1; + create table finish_mark ( id int PRIMARY KEY diff --git a/tests/integration_tests/avro_basic/run.sh b/tests/integration_tests/avro_basic/run.sh index 9bf31fe5c5..6dccf7a8af 100755 --- a/tests/integration_tests/avro_basic/run.sh +++ b/tests/integration_tests/avro_basic/run.sh @@ -42,7 +42,8 @@ function run() { run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY - SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=avro&enable-tidb-extension=true&avro-enable-watermark=true&avro-decimal-handling-mode=string&avro-bigint-unsigned-handling-mode=string" + # Cover Avro insert, update, and delete events with _ticdc_before through the real Kafka consumer. + SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=avro&enable-tidb-extension=true&avro-enable-watermark=true&avro-decimal-handling-mode=string&avro-bigint-unsigned-handling-mode=string&avro-include-before-value=true" schema_registry_uri="http://127.0.0.1:8088" cdc_cli_changefeed create --start-ts=$start_ts --sink-uri=$SINK_URI --config=$CUR/conf/changefeed.toml --schema-registry=$schema_registry_uri From 67596ee928efc6a13c4a100f95135c6af42c8c1b Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 29 May 2026 09:01:17 +0000 Subject: [PATCH 2/8] update Signed-off-by: wk989898 --- api/v2/model.go | 3 +++ api/v2/model_test.go | 30 ++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/api/v2/model.go b/api/v2/model.go index 1aee8e00c8..abb3d307ce 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -415,6 +415,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( AvroEnableWatermark: oldConfig.AvroEnableWatermark, AvroDecimalHandlingMode: oldConfig.AvroDecimalHandlingMode, AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode, + AvroIncludeBeforeValue: oldConfig.AvroIncludeBeforeValue, EncodingFormat: oldConfig.EncodingFormat, } } @@ -744,6 +745,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { AvroEnableWatermark: oldConfig.AvroEnableWatermark, AvroDecimalHandlingMode: oldConfig.AvroDecimalHandlingMode, AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode, + AvroIncludeBeforeValue: oldConfig.AvroIncludeBeforeValue, EncodingFormat: oldConfig.EncodingFormat, } } @@ -1410,6 +1412,7 @@ type CodecConfig struct { AvroEnableWatermark *bool `json:"avro_enable_watermark,omitempty"` AvroDecimalHandlingMode *string `json:"avro_decimal_handling_mode,omitempty"` AvroBigintUnsignedHandlingMode *string `json:"avro_bigint_unsigned_handling_mode,omitempty"` + AvroIncludeBeforeValue *bool `json:"avro_include_before_value,omitempty"` EncodingFormat *string `json:"encoding_format,omitempty"` } diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 03de49437a..168d2cacbf 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -201,3 +201,33 @@ func TestReplicaConfigConversionRedoBatchField(t *testing.T) { require.NotNil(t, apiCfgBack.Consistent.EventCollectorBatchCount) require.Equal(t, 4096, *apiCfgBack.Consistent.EventCollectorBatchCount) } + +func TestReplicaConfigCodecConfigConversion(t *testing.T) { + t.Parallel() + + apiCfg := &ReplicaConfig{ + Sink: &SinkConfig{ + KafkaConfig: &KafkaConfig{ + CodecConfig: &CodecConfig{ + EnableTiDBExtension: util.AddressOf(true), + MaxBatchSize: util.AddressOf(16), + AvroEnableWatermark: util.AddressOf(true), + AvroDecimalHandlingMode: util.AddressOf("string"), + AvroBigintUnsignedHandlingMode: util.AddressOf("string"), + AvroIncludeBeforeValue: util.AddressOf(true), + EncodingFormat: util.AddressOf("avro"), + }, + }, + }, + } + + internalCfg := apiCfg.ToInternalReplicaConfig() + require.NotNil(t, internalCfg.Sink.KafkaConfig) + require.NotNil(t, internalCfg.Sink.KafkaConfig.CodecConfig) + require.True(t, util.GetOrZero(internalCfg.Sink.KafkaConfig.CodecConfig.AvroIncludeBeforeValue)) + + apiCfgBack := ToAPIReplicaConfig(internalCfg) + require.NotNil(t, apiCfgBack.Sink.KafkaConfig) + require.NotNil(t, apiCfgBack.Sink.KafkaConfig.CodecConfig) + require.True(t, util.GetOrZero(apiCfgBack.Sink.KafkaConfig.CodecConfig.AvroIncludeBeforeValue)) +} From 36b17f7139912eafadb4d79985afb72d98b833ec Mon Sep 17 00:00:00 2001 From: wk989898 Date: Tue, 4 Aug 2026 06:45:49 +0000 Subject: [PATCH 3/8] update Signed-off-by: wk989898 --- pkg/sink/codec/avro/avro_test.go | 2 +- pkg/sink/codec/avro/decoder.go | 23 ++++++++++++++++++----- 2 files changed, 19 insertions(+), 6 deletions(-) diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index 42933ba1e1..1a8b6487dd 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -301,7 +301,7 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { require.True(t, exist) require.Equal(t, common.MessageTypeRow, messageType) - decoded := decoder.NextDMLEvent() + decoded := decoder.NextDMLMessage().ToDMLEvent() require.NotNil(t, decoded) require.Equal(t, tc.event.CommitTs, decoded.CommitTs) diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go index 3d562be894..d0a226ef6f 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -115,7 +115,7 @@ func (d *decoder) NextResolvedEvent() uint64 { // NextDMLMessage returns the next row changed message if exists func (d *decoder) NextDMLMessage() *common.DMLMessage { - keyMap, valueMap, valueSchema, isDelete, deleteCommitTs := d.decodeDMLPayload() + keyMap, valueMap, valueSchema, isDelete, hasValue, deleteCommitTs := d.decodeDMLPayload() schemaName, tableName := schemaAndTableName(valueSchema) commitTs := deleteCommitTs if commitTs == 0 && !isDelete { @@ -124,10 +124,12 @@ func (d *decoder) NextDMLMessage() *common.DMLMessage { rowType := commonType.RowTypeInsert if isDelete { rowType = commonType.RowTypeDelete + } else if d.config.AvroIncludeBeforeValue && valueMap[tidbOp] == updateOperation { + rowType = commonType.RowTypeUpdate } tableID := tableIDAllocator.Allocate(schemaName, tableName) return common.NewDMLMessage(tableID, schemaName, tableName, commitTs, rowType, func() *commonEvent.DMLEvent { - return d.assembleDMLEventFromDecoded(keyMap, valueMap, valueSchema, isDelete, deleteCommitTs) + return d.assembleDMLEventFromDecoded(keyMap, valueMap, valueSchema, isDelete, hasValue, deleteCommitTs) }) } @@ -136,6 +138,7 @@ func (d *decoder) decodeDMLPayload() ( valueMap map[string]any, valueSchema map[string]any, isDelete bool, + hasValue bool, deleteCommitTs uint64, ) { var ( @@ -150,9 +153,8 @@ func (d *decoder) decodeDMLPayload() ( } isDeleteValue := d.isDeleteValue() - hasValue := len(d.value) != 0 && !isDeleteValue - isDelete := !hasValue - deleteCommitTs := uint64(0) + hasValue = len(d.value) != 0 && !isDeleteValue + isDelete = !hasValue if !hasValue { // Legacy delete event only has the key payload or a delete marker value. // It can only be decoded as a delete row with key columns in PreRow. @@ -171,6 +173,17 @@ func (d *decoder) decodeDMLPayload() ( } } + return keyMap, valueMap, valueSchema, isDelete, hasValue, deleteCommitTs +} + +func (d *decoder) assembleDMLEventFromDecoded( + keyMap map[string]any, + valueMap map[string]any, + valueSchema map[string]any, + isDelete bool, + hasValue bool, + deleteCommitTs uint64, +) *commonEvent.DMLEvent { event, err := assembleEvent(keyMap, valueMap, valueSchema, isDelete, hasValue) if err != nil { log.Panic("assemble event failed", zap.Error(err)) From 3beaa28e15e43bef6481ac076625373e3181ce92 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 6 Aug 2026 08:06:59 +0000 Subject: [PATCH 4/8] chore Signed-off-by: wk989898 --- pkg/sink/codec/avro/avro_test.go | 18 +++++++++--------- pkg/sink/codec/avro/decoder.go | 14 +++++++------- 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index 1a8b6487dd..f288206678 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -110,7 +110,7 @@ func TestAvroEncode4EnableChecksum(t *testing.T) { require.NoError(t, err) require.NotNil(t, res) - m, ok := res.(map[string]interface{}) + m, ok := res.(map[string]any) require.True(t, ok) _, found := m[tidbRowLevelChecksum] @@ -161,7 +161,7 @@ func TestAvroEncodeDeleteChecksum(t *testing.T) { res, _, err := avroValueCodec.NativeFromBinary(data) require.NoError(t, err) - m, ok := res.(map[string]interface{}) + m, ok := res.(map[string]any) require.True(t, ok) require.Equal(t, deleteOperation, m[tidbOp]) require.Equal(t, "22", m[tidbRowLevelChecksum]) @@ -193,12 +193,12 @@ func TestAvroEncode(t *testing.T) { res, _, err := avroKeyCodec.NativeFromBinary(data) require.NoError(t, err) require.NotNil(t, res) - for k := range res.(map[string]interface{}) { + for k := range res.(map[string]any) { if k == "_tidb_commit_ts" || k == "_tidb_op" || k == "_tidb_commit_physical_time" { require.Fail(t, "key shall not include extension fields") } } - require.Equal(t, int32(127), res.(map[string]interface{})["tu1"]) + require.Equal(t, int32(127), res.(map[string]any)["tu1"]) bin, err = encoder.encodeValue(ctx, topic, event) require.NoError(t, err) @@ -213,7 +213,7 @@ func TestAvroEncode(t *testing.T) { require.NoError(t, err) require.NotNil(t, res) - for k, v := range res.(map[string]interface{}) { + for k, v := range res.(map[string]any) { if k == "_tidb_op" { require.Equal(t, "c", v.(string)) } @@ -282,7 +282,7 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { require.NoError(t, err) require.NotNil(t, res) - m, ok := res.(map[string]interface{}) + m, ok := res.(map[string]any) require.True(t, ok) require.Equal(t, int64(tc.event.CommitTs), m[tidbCommitTs]) require.Equal(t, tc.op, m[tidbOp]) @@ -400,7 +400,7 @@ func TestAvroEnvelope(t *testing.T) { require.NoError(t, err) - testNativeData := make(map[string]interface{}) + testNativeData := make(map[string]any) testNativeData["id"] = 7 bin, err := avroCodec.BinaryFromNative(nil, testNativeData) @@ -422,7 +422,7 @@ func TestAvroEnvelope(t *testing.T) { require.NoError(t, err) require.NotNil(t, parsed) - id, exists := parsed.(map[string]interface{})["id"] + id, exists := parsed.(map[string]any)["id"] require.True(t, exists) require.Equal(t, int32(7), id) @@ -442,7 +442,7 @@ func TestAvroEnvelope(t *testing.T) { parsed, _, err = avroCodec.NativeFromBinary(evlp[18:]) require.NoError(t, err) require.NotNil(t, parsed) - id, exists = parsed.(map[string]interface{})["id"] + id, exists = parsed.(map[string]any)["id"] require.True(t, exists) require.Equal(t, int32(7), id) } diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go index d0a226ef6f..78d50ef2d4 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -254,7 +254,7 @@ func (d *decoder) decodeDeleteCommitTs() uint64 { // For legacy delete, valueMap is keyMap and only contains handle columns. // schema is corresponding to the valueMap, it can be used to decode the valueMap to construct columns. func assembleEvent( - keyMap, valueMap, schema map[string]interface{}, isDelete bool, hasValue bool, + keyMap, valueMap, schema map[string]any, isDelete bool, hasValue bool, ) (*commonEvent.DMLEvent, error) { fields, ok := schema["fields"].([]any) if !ok { @@ -270,7 +270,7 @@ func assembleEvent( if err != nil { return nil, errors.Trace(err) } - var beforeData map[string]interface{} + var beforeData map[string]any if hasBefore { _, beforeData, err = avroData2Columns(beforeMap, fields) if err != nil { @@ -333,13 +333,13 @@ func isAvroExtensionField(name string) bool { } } -func extractBeforeValueMap(valueMap map[string]interface{}) (map[string]interface{}, bool, error) { +func extractBeforeValueMap(valueMap map[string]any) (map[string]any, bool, error) { rawBefore, ok := valueMap[ticdcBefore] if !ok || rawBefore == nil { return nil, false, nil } - beforeUnion, ok := rawBefore.(map[string]interface{}) + beforeUnion, ok := rawBefore.(map[string]any) if !ok { return nil, false, errors.ErrCodecDecode.GenWithStack("before value should be a map") } @@ -347,7 +347,7 @@ func extractBeforeValueMap(valueMap map[string]interface{}) (map[string]interfac if unionName == "null" || value == nil { return nil, false, nil } - before, ok := value.(map[string]interface{}) + before, ok := value.(map[string]any) if !ok { return nil, false, errors.ErrCodecDecode.GenWithStack("before record should be a map") } @@ -357,8 +357,8 @@ func extractBeforeValueMap(valueMap map[string]interface{}) (map[string]interfac } func avroData2Columns( - valueMap map[string]interface{}, fields []interface{}, -) ([]*timodel.ColumnInfo, map[string]interface{}, error) { + valueMap map[string]any, fields []any, +) ([]*timodel.ColumnInfo, map[string]any, error) { columns := make([]*timodel.ColumnInfo, 0, len(valueMap)) data := make(map[string]any, 0) // fields is ordered by the column id, so iterate over it to build columns From 140b23b07c1cd2c49c2cca59d44c7720f70abd3f Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 6 Aug 2026 08:23:56 +0000 Subject: [PATCH 5/8] update Signed-off-by: wk989898 --- pkg/sink/codec/avro/avro_test.go | 42 ++++++++++++++++++++++++++++---- pkg/sink/codec/avro/decoder.go | 17 +++++++------ 2 files changed, 46 insertions(+), 13 deletions(-) diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index f288206678..26467f34bd 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -244,23 +244,27 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { name string event *commonEvent.RowEvent op string + rowType commonType.RowType hasBefore bool }{ { - name: "insert", - event: newAvroRowEventForTest(tableInfo, 1024, chunk.Row{}, beforeRow), - op: insertOperation, + name: "insert", + event: newAvroRowEventForTest(tableInfo, 1024, chunk.Row{}, beforeRow), + op: insertOperation, + rowType: commonType.RowTypeInsert, }, { name: "update", event: newAvroRowEventForTest(tableInfo, 1025, beforeRow, afterRow), op: updateOperation, + rowType: commonType.RowTypeUpdate, hasBefore: true, }, { name: "delete", event: newAvroRowEventForTest(tableInfo, 1026, afterRow, chunk.Row{}), op: deleteOperation, + rowType: commonType.RowTypeDelete, hasBefore: true, }, } @@ -294,14 +298,18 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { key, err := encoder.encodeKey(ctx, topic, tc.event) require.NoError(t, err) - decoder := NewDecoder(codecConfig, 0, encoder.schemaM, topic, nil) + decoderConfig := *codecConfig + decoderConfig.AvroIncludeBeforeValue = false + decoder := NewDecoder(&decoderConfig, 0, encoder.schemaM, topic, nil) decoder.AddKeyValue(key, bin) messageType, exist := decoder.HasNext() require.True(t, exist) require.Equal(t, common.MessageTypeRow, messageType) - decoded := decoder.NextDMLMessage().ToDMLEvent() + message := decoder.NextDMLMessage() + require.Equal(t, tc.rowType, message.RowType) + decoded := message.ToDMLEvent() require.NotNil(t, decoded) require.Equal(t, tc.event.CommitTs, decoded.CommitTs) @@ -312,6 +320,30 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { } } +func TestSchemaAndTableName(t *testing.T) { + testCases := []struct { + name string + namespace string + schemaName string + }{ + {name: "keyspace and schema", namespace: "keyspace.schema", schemaName: "schema"}, + {name: "empty keyspace", namespace: ".schema", schemaName: "schema"}, + {name: "empty schema", namespace: "keyspace", schemaName: ""}, + {name: "empty namespace", namespace: "", schemaName: ""}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + schemaName, tableName := schemaAndTableName(map[string]any{ + "namespace": tc.namespace, + "name": "table", + }) + require.Equal(t, tc.schemaName, schemaName) + require.Equal(t, "table", tableName) + }) + } +} + func TestAvroEncodeDeleteEventUsesPreRowForKey(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolAvro) codecConfig.EnableTiDBExtension = true diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go index 78d50ef2d4..1a3f569e30 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -124,7 +124,7 @@ func (d *decoder) NextDMLMessage() *common.DMLMessage { rowType := commonType.RowTypeInsert if isDelete { rowType = commonType.RowTypeDelete - } else if d.config.AvroIncludeBeforeValue && valueMap[tidbOp] == updateOperation { + } else if operation, ok := valueMap[tidbOp]; ok && operation == updateOperation { rowType = commonType.RowTypeUpdate } tableID := tableIDAllocator.Allocate(schemaName, tableName) @@ -168,8 +168,8 @@ func (d *decoder) decodeDMLPayload() ( if err != nil { log.Panic("decode value failed", zap.Error(err)) } - if d.config.AvroIncludeBeforeValue { - isDelete = valueMap[tidbOp] == deleteOperation + if operation, ok := valueMap[tidbOp]; ok { + isDelete = operation == deleteOperation } } @@ -278,10 +278,7 @@ func assembleEvent( } } - // "namespace.schema" - namespace := schema["namespace"].(string) - schemaName := strings.Split(namespace, ".")[1] - tableName := schema["name"].(string) + schemaName, tableName := schemaAndTableName(schema) var commitTs int64 if hasValue { @@ -416,7 +413,11 @@ func avroData2Columns( func schemaAndTableName(schema map[string]any) (string, string) { namespace := schema["namespace"].(string) - return strings.Split(namespace, ".")[1], schema["name"].(string) + parts := strings.SplitN(namespace, ".", 2) + if len(parts) < 2 { + return "", schema["name"].(string) + } + return parts[1], schema["name"].(string) } func queryTableInfo(schemaName, tableName string, columns []*timodel.ColumnInfo, keyMap map[string]any) *commonType.TableInfo { From 7bd574798ed2571b7218073854a18105a0380daf Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 6 Aug 2026 09:08:46 +0000 Subject: [PATCH 6/8] update Signed-off-by: wk989898 --- api/v2/changefeed_toml_test.go | 46 ++++++++++++++++++++++++ api/v2/model.go | 14 ++++---- pkg/sink/codec/avro/arvo.go | 19 +++++++--- pkg/sink/codec/avro/avro_test.go | 53 ++++++++++++++++++++++++++++ pkg/sink/codec/avro/decoder.go | 2 +- pkg/sink/codec/common/config_test.go | 4 +++ 6 files changed, 125 insertions(+), 13 deletions(-) diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 4a8539ef4a..25490277ad 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -131,6 +131,52 @@ func TestChangeFeedInfoTOMLRoundTripToInternal(t *testing.T) { require.Equal(t, "eventual", util.GetOrZero(wrapper.Config.Consistent.Level)) } +func TestCodecConfigTOMLRoundTripToInternal(t *testing.T) { + t.Parallel() + + cfg := &ReplicaConfig{ + Sink: &SinkConfig{ + KafkaConfig: &KafkaConfig{ + CodecConfig: &CodecConfig{ + EnableTiDBExtension: util.AddressOf(true), + MaxBatchSize: util.AddressOf(32), + AvroEnableWatermark: util.AddressOf(true), + AvroDecimalHandlingMode: util.AddressOf("string"), + AvroBigintUnsignedHandlingMode: util.AddressOf("string"), + AvroIncludeBeforeValue: util.AddressOf(true), + EncodingFormat: util.AddressOf("avro"), + }, + }, + }, + } + + var buf bytes.Buffer + require.NoError(t, toml.NewEncoder(&buf).Encode(cfg)) + out := buf.String() + require.Contains(t, out, "enable-tidb-extension = true") + require.Contains(t, out, "max-batch-size = 32") + require.Contains(t, out, "avro-enable-watermark = true") + require.Contains(t, out, `avro-decimal-handling-mode = "string"`) + require.Contains(t, out, `avro-bigint-unsigned-handling-mode = "string"`) + require.Contains(t, out, "avro-include-before-value = true") + require.Contains(t, out, `encoding-format = "avro"`) + + var internalCfg config.ReplicaConfig + meta, err := toml.Decode(out, &internalCfg) + require.NoError(t, err) + require.Empty(t, meta.Undecoded()) + require.NotNil(t, internalCfg.Sink.KafkaConfig) + require.NotNil(t, internalCfg.Sink.KafkaConfig.CodecConfig) + codecCfg := internalCfg.Sink.KafkaConfig.CodecConfig + require.True(t, util.GetOrZero(codecCfg.EnableTiDBExtension)) + require.Equal(t, 32, util.GetOrZero(codecCfg.MaxBatchSize)) + require.True(t, util.GetOrZero(codecCfg.AvroEnableWatermark)) + require.Equal(t, "string", util.GetOrZero(codecCfg.AvroDecimalHandlingMode)) + require.Equal(t, "string", util.GetOrZero(codecCfg.AvroBigintUnsignedHandlingMode)) + require.True(t, util.GetOrZero(codecCfg.AvroIncludeBeforeValue)) + require.Equal(t, "avro", util.GetOrZero(codecCfg.EncodingFormat)) +} + // TestDefaultConfigTOMLRoundTripToInternal encodes the full default replica // config to TOML and decodes it into the internal config.ReplicaConfig, then // asserts that no config-section key is left undecoded. This proves every TOML diff --git a/api/v2/model.go b/api/v2/model.go index 02fdeeb6df..82ecc137cd 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -1438,13 +1438,13 @@ type Capture struct { // CodecConfig represents a MQ codec configuration type CodecConfig struct { - EnableTiDBExtension *bool `json:"enable_tidb_extension,omitempty"` - MaxBatchSize *int `json:"max_batch_size,omitempty"` - AvroEnableWatermark *bool `json:"avro_enable_watermark,omitempty"` - AvroDecimalHandlingMode *string `json:"avro_decimal_handling_mode,omitempty"` - AvroBigintUnsignedHandlingMode *string `json:"avro_bigint_unsigned_handling_mode,omitempty"` - AvroIncludeBeforeValue *bool `json:"avro_include_before_value,omitempty"` - EncodingFormat *string `json:"encoding_format,omitempty"` + EnableTiDBExtension *bool `json:"enable_tidb_extension,omitempty" toml:"enable-tidb-extension,omitempty"` + MaxBatchSize *int `json:"max_batch_size,omitempty" toml:"max-batch-size,omitempty"` + AvroEnableWatermark *bool `json:"avro_enable_watermark,omitempty" toml:"avro-enable-watermark,omitempty"` + AvroDecimalHandlingMode *string `json:"avro_decimal_handling_mode,omitempty" toml:"avro-decimal-handling-mode,omitempty"` + AvroBigintUnsignedHandlingMode *string `json:"avro_bigint_unsigned_handling_mode,omitempty" toml:"avro-bigint-unsigned-handling-mode,omitempty"` + AvroIncludeBeforeValue *bool `json:"avro_include_before_value,omitempty" toml:"avro-include-before-value,omitempty"` + EncodingFormat *string `json:"encoding_format,omitempty" toml:"encoding-format,omitempty"` } // PulsarConfig represents a pulsar sink configuration diff --git a/pkg/sink/codec/avro/arvo.go b/pkg/sink/codec/avro/arvo.go index 71d431f298..95d3e2c669 100644 --- a/pkg/sink/codec/avro/arvo.go +++ b/pkg/sink/codec/avro/arvo.go @@ -178,6 +178,8 @@ func (a *BatchEncoder) encodeValue(ctx context.Context, topic string, e *event.R } if a.config.EnableTiDBExtension { native = a.nativeValueWithExtension(native, e) + } else if a.config.AvroIncludeBeforeValue { + native[tidbOp] = getOperation(e) } bin, err := avroCodec.BinaryFromNative(nil, native) @@ -265,12 +267,8 @@ func routedTableName(tableInfo *commonType.TableInfo) commonType.TableName { func (a *BatchEncoder) schemaWithExtension( top *avroSchemaTop, ) *avroSchemaTop { + top = schemaWithOperation(top) top.Fields = append(top.Fields, - map[string]any{ - "name": tidbOp, - "type": "string", - "default": "", - }, map[string]any{ "name": tidbCommitTs, "type": "long", @@ -305,6 +303,15 @@ func (a *BatchEncoder) schemaWithExtension( return top } +func schemaWithOperation(top *avroSchemaTop) *avroSchemaTop { + top.Fields = append(top.Fields, map[string]any{ + "name": tidbOp, + "type": "string", + "default": "", + }) + return top +} + func (a *BatchEncoder) schemaWithBeforeValue( top *avroSchemaTop, tableName *commonType.TableName, @@ -501,6 +508,8 @@ func (a *BatchEncoder) value2AvroSchema( if a.config.EnableTiDBExtension { top = a.schemaWithExtension(top) + } else if a.config.AvroIncludeBeforeValue { + top = schemaWithOperation(top) } str, err := json.Marshal(top) diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index 26467f34bd..53e76283bb 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -309,6 +309,7 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { message := decoder.NextDMLMessage() require.Equal(t, tc.rowType, message.RowType) + require.Equal(t, tc.event.CommitTs, message.GetCommitTs()) decoded := message.ToDMLEvent() require.NotNil(t, decoded) require.Equal(t, tc.event.CommitTs, decoded.CommitTs) @@ -320,6 +321,58 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { } } +func TestAvroEncodeIncludeBeforeValueWithoutTiDBExtension(t *testing.T) { + codecConfig := common.NewConfig(config.ProtocolAvro) + codecConfig.AvroIncludeBeforeValue = true + + ctx := t.Context() + encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) + defer TeardownEncoderAndSchemaRegistry4Testing() + require.NoError(t, err) + + tableInfo := newAvroTableInfoForTest() + beforeRow := chunk.MutRowFromValues(int64(1), int64(18)).ToRow() + afterRow := chunk.MutRowFromValues(int64(1), int64(20)).ToRow() + testCases := []struct { + name string + event *commonEvent.RowEvent + op string + }{ + { + name: "update", + event: newAvroRowEventForTest(tableInfo, 1025, beforeRow, afterRow), + op: updateOperation, + }, + { + name: "delete", + event: newAvroRowEventForTest(tableInfo, 1026, afterRow, chunk.Row{}), + op: deleteOperation, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + bin, err := encoder.encodeValue(ctx, "default", tc.event) + require.NoError(t, err) + require.NotNil(t, bin) + + cid, data, err := extractConfluentSchemaIDAndBinaryData(bin) + require.NoError(t, err) + avroValueCodec, err := encoder.schemaM.Lookup(ctx, "default", schemaID{confluentSchemaID: cid}) + require.NoError(t, err) + res, _, err := avroValueCodec.NativeFromBinary(data) + require.NoError(t, err) + + valueMap, ok := res.(map[string]any) + require.True(t, ok) + require.Equal(t, tc.op, valueMap[tidbOp]) + require.NotNil(t, valueMap[ticdcBefore]) + require.NotContains(t, valueMap, tidbCommitTs) + require.NotContains(t, valueMap, tidbPhysicalTime) + }) + } +} + func TestSchemaAndTableName(t *testing.T) { testCases := []struct { name string diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go index 1a3f569e30..d3c9a60337 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -118,7 +118,7 @@ func (d *decoder) NextDMLMessage() *common.DMLMessage { keyMap, valueMap, valueSchema, isDelete, hasValue, deleteCommitTs := d.decodeDMLPayload() schemaName, tableName := schemaAndTableName(valueSchema) commitTs := deleteCommitTs - if commitTs == 0 && !isDelete { + if hasValue { commitTs = uint64(valueMap[tidbCommitTs].(int64)) } rowType := commonType.RowTypeInsert diff --git a/pkg/sink/codec/common/config_test.go b/pkg/sink/codec/common/config_test.go index 80112b3dbd..c85209638f 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -31,7 +31,10 @@ func TestAvroIncludeBeforeValueConfig(t *testing.T) { err = cfg.Apply(sinkURI, &config.SinkConfig{}) require.NoError(t, err) + require.False(t, cfg.EnableTiDBExtension) require.True(t, cfg.AvroIncludeBeforeValue) + cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" + require.NoError(t, cfg.Validate()) } func TestAvroIncludeBeforeValueConfigFile(t *testing.T) { @@ -47,5 +50,6 @@ func TestAvroIncludeBeforeValueConfigFile(t *testing.T) { }, }) require.NoError(t, err) + require.False(t, cfg.EnableTiDBExtension) require.True(t, cfg.AvroIncludeBeforeValue) } From c432cf7a4b0c1bca7b53f14c5d0703e9d21d961c Mon Sep 17 00:00:00 2001 From: wk989898 Date: Fri, 7 Aug 2026 08:07:48 +0000 Subject: [PATCH 7/8] update Signed-off-by: wk989898 --- pkg/sink/codec/avro/avro_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/pkg/sink/codec/avro/avro_test.go b/pkg/sink/codec/avro/avro_test.go index 53e76283bb..f9d23e5ac0 100644 --- a/pkg/sink/codec/avro/avro_test.go +++ b/pkg/sink/codec/avro/avro_test.go @@ -87,7 +87,7 @@ func TestAvroEncode4EnableChecksum(t *testing.T) { codecConfig.AvroDecimalHandlingMode = "string" codecConfig.AvroBigintUnsignedHandlingMode = "string" - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) @@ -129,7 +129,7 @@ func TestAvroEncodeDeleteChecksum(t *testing.T) { codecConfig.EnableRowChecksum = true codecConfig.AvroIncludeBeforeValue = true - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) @@ -171,7 +171,7 @@ func TestAvroEncode(t *testing.T) { codecConfig := common.NewConfig(config.ProtocolAvro) codecConfig.EnableTiDBExtension = true - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) @@ -228,7 +228,7 @@ func TestAvroEncodeIncludeBeforeValue(t *testing.T) { codecConfig.EnableTiDBExtension = true codecConfig.AvroIncludeBeforeValue = true - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) From 81bccf67b73c56dbd78372b07c1700a65a0825d3 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Mon, 10 Aug 2026 02:53:50 +0000 Subject: [PATCH 8/8] update comment Signed-off-by: wk989898 --- utils/chann/drainable_chann.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/utils/chann/drainable_chann.go b/utils/chann/drainable_chann.go index 61003e17ac..e2b6c0c4b5 100644 --- a/utils/chann/drainable_chann.go +++ b/utils/chann/drainable_chann.go @@ -17,8 +17,6 @@ package chann // It is a wrapper of Chann. // NOTICE: Please make sure that it is safe to drain rest elements in the channel // before closing the channel. -// -// Deprecated: Just Don't Use It. Use a channel please. type DrainableChann[T any] struct { inner *Chann[T] }