diff --git a/api/v2/changefeed_toml_test.go b/api/v2/changefeed_toml_test.go index 459d26fc42..4bfdeee940 100644 --- a/api/v2/changefeed_toml_test.go +++ b/api/v2/changefeed_toml_test.go @@ -133,6 +133,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 b96788e467..9083c92a7d 100644 --- a/api/v2/model.go +++ b/api/v2/model.go @@ -437,6 +437,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig( AvroEnableWatermark: oldConfig.AvroEnableWatermark, AvroDecimalHandlingMode: oldConfig.AvroDecimalHandlingMode, AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode, + AvroIncludeBeforeValue: oldConfig.AvroIncludeBeforeValue, EncodingFormat: oldConfig.EncodingFormat, } } @@ -770,6 +771,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig { AvroEnableWatermark: oldConfig.AvroEnableWatermark, AvroDecimalHandlingMode: oldConfig.AvroDecimalHandlingMode, AvroBigintUnsignedHandlingMode: oldConfig.AvroBigintUnsignedHandlingMode, + AvroIncludeBeforeValue: oldConfig.AvroIncludeBeforeValue, EncodingFormat: oldConfig.EncodingFormat, } } @@ -1448,6 +1450,7 @@ type CodecConfig struct { 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"` } diff --git a/api/v2/model_test.go b/api/v2/model_test.go index 3a194a3f9f..ef48ced8be 100644 --- a/api/v2/model_test.go +++ b/api/v2/model_test.go @@ -230,3 +230,33 @@ func TestReplicaConfigConversionMySQLAsyncDDLTimeout(t *testing.T) { require.NotNil(t, apiCfgBack.Sink.MySQLConfig) require.Equal(t, "45m", util.GetOrZero(apiCfgBack.Sink.MySQLConfig.AsyncDDLTimeout)) } + +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)) +} diff --git a/pkg/config/sink.go b/pkg/config/sink.go index a0f415a34d..55645a073d 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -453,6 +453,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 cd15233bc8..95d3e2c669 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,8 +167,19 @@ 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) + } else if a.config.AvroIncludeBeforeValue { + native[tidbOp] = getOperation(e) } bin, err := avroCodec.BinaryFromNative(nil, native) @@ -193,13 +208,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() @@ -210,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", @@ -250,6 +303,34 @@ 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, + 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,8 +499,17 @@ 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) + } 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 c4e5883e7c..f9d23e5ac0 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 @@ -32,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) @@ -55,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] @@ -68,11 +123,55 @@ 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(t.Context()) + 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]any) + 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 - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(t.Context()) defer cancel() encoder, err := SetupEncoderAndSchemaRegistry4Testing(ctx, codecConfig) @@ -94,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) @@ -114,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)) } @@ -124,6 +223,180 @@ 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(t.Context()) + 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 + rowType commonType.RowType + hasBefore bool + }{ + { + 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, + }, + } + + 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]any) + 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) + 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) + + 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) + + decodedRow, ok := decoded.GetNextRow() + require.True(t, ok) + common.CompareRow(t, tc.event.Event, tc.event.TableInfo, decodedRow, decoded.TableInfo) + }) + } +} + +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 + 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 @@ -212,7 +485,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) @@ -234,7 +507,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) @@ -254,7 +527,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 ee074954fd..d3c9a60337 100644 --- a/pkg/sink/codec/avro/decoder.go +++ b/pkg/sink/codec/avro/decoder.go @@ -115,19 +115,21 @@ 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 { + if hasValue { commitTs = uint64(valueMap[tidbCommitTs].(int64)) } rowType := commonType.RowTypeInsert if isDelete { rowType = commonType.RowTypeDelete + } else if operation, ok := valueMap[tidbOp]; ok && operation == 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 ( @@ -149,12 +152,13 @@ func (d *decoder) decodeDMLPayload() ( 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() - if isDelete { - // delete event only have key part, treat it as the value part also. - if d.isDeleteValue() { + isDeleteValue := d.isDeleteValue() + 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. + if isDeleteValue { deleteCommitTs = d.decodeDeleteCommitTs() } valueMap = keyMap @@ -164,9 +168,12 @@ func (d *decoder) decodeDMLPayload() ( if err != nil { log.Panic("decode value failed", zap.Error(err)) } + if operation, ok := valueMap[tidbOp]; ok { + isDelete = operation == deleteOperation + } } - return keyMap, valueMap, valueSchema, isDelete, deleteCommitTs + return keyMap, valueMap, valueSchema, isDelete, hasValue, deleteCommitTs } func (d *decoder) assembleDMLEventFromDecoded( @@ -174,9 +181,10 @@ func (d *decoder) assembleDMLEventFromDecoded( valueMap map[string]any, valueSchema map[string]any, isDelete bool, + hasValue bool, deleteCommitTs uint64, ) *commonEvent.DMLEvent { - 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)) } @@ -185,22 +193,23 @@ func (d *decoder) assembleDMLEventFromDecoded( 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() { @@ -241,15 +250,112 @@ 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]any, isDelete bool, + keyMap, valueMap, schema map[string]any, isDelete bool, hasValue bool, ) (*commonEvent.DMLEvent, error) { fields, ok := schema["fields"].([]any) 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]any + if hasBefore { + _, beforeData, err = avroData2Columns(beforeMap, fields) + if err != nil { + return nil, errors.Trace(err) + } + } + + schemaName, tableName := schemaAndTableName(schema) + + 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]any) (map[string]any, bool, error) { + rawBefore, ok := valueMap[ticdcBefore] + if !ok || rawBefore == nil { + return nil, false, nil + } + + beforeUnion, ok := rawBefore.(map[string]any) + 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]any) + 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]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 @@ -257,12 +363,11 @@ func assembleEvent( for idx, item := range fields { field, ok := item.(map[string]any) 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. @@ -286,11 +391,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 @@ -303,41 +408,16 @@ func assembleEvent( tiCol.SetFlag(flag) columns = append(columns, tiCol) } - - schemaName, tableName := schemaAndTableName(schema) - - 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 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 { 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 4f5cee429d..dbb584a8ad 100644 --- a/pkg/sink/codec/common/config.go +++ b/pkg/sink/codec/common/config.go @@ -64,6 +64,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 @@ -129,6 +130,7 @@ func NewConfig(protocol config.Protocol) *Config { AvroConfluentSchemaRegistry: "", AvroDecimalHandlingMode: "precise", AvroBigintUnsignedHandlingMode: "long", + AvroIncludeBeforeValue: false, AvroEnableWatermark: false, OnlyOutputUpdatedColumns: false, @@ -172,6 +174,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 and debezium-avro protocol // only used for internal testing, do not set this in the production environment since the @@ -229,6 +232,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.Protocol == config.ProtocolDebeziumAvro) { @@ -332,6 +338,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 index 4369de7216..c85209638f 100644 --- a/pkg/sink/codec/common/config_test.go +++ b/pkg/sink/codec/common/config_test.go @@ -18,143 +18,38 @@ import ( "testing" "github.com/pingcap/ticdc/pkg/config" - "github.com/pingcap/ticdc/pkg/errors" "github.com/pingcap/ticdc/pkg/util" "github.com/stretchr/testify/require" ) -func TestApplyReturnsSinkInvalidConfigForQueryBindingError(t *testing.T) { - cfg := NewConfig(config.ProtocolOpen) - sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?max-batch-size=invalid") - require.NoError(t, err) - - err = cfg.Apply(sinkURI, config.GetDefaultReplicaConfig().Sink) - errCode, ok := errors.RFCCode(err) - require.True(t, ok, err) - require.Equal(t, errors.ErrSinkInvalidConfig.RFCCode(), errCode) -} +func TestAvroIncludeBeforeValueConfig(t *testing.T) { + cfg := NewConfig(config.ProtocolAvro) + require.False(t, cfg.AvroIncludeBeforeValue) -func TestValidateMaxBatchMessageBytes(t *testing.T) { - tests := []struct { - name string - adjust func(*Config) - expected string - }{ - { - name: "non-positive max message bytes", - adjust: func(cfg *Config) { - cfg.MaxMessageBytes = 0 - }, - expected: "invalid max-message-bytes 0", - }, - { - name: "negative max batched bytes", - adjust: func(cfg *Config) { - cfg.MaxBatchedBytes = -1 - }, - expected: "invalid max-batch-message-bytes -1", - }, - { - name: "max batched bytes exceeds max message bytes", - adjust: func(cfg *Config) { - cfg.MaxMessageBytes = 100 - cfg.MaxBatchedBytes = 101 - }, - expected: "max-batch-message-bytes 101 cannot be greater than max-message-bytes 100", - }, - { - name: "non-positive max batch size", - adjust: func(cfg *Config) { - cfg.MaxBatchSize = 0 - }, - expected: "invalid max-batch-size 0", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - cfg := NewConfig(config.ProtocolOpen) - test.adjust(cfg) - - err := cfg.Validate() - require.ErrorContains(t, err, test.expected) - errCode, ok := errors.RFCCode(err) - require.True(t, ok, err) - require.Equal(t, errors.ErrCodecInvalidConfig.RFCCode(), errCode) - }) - } -} - -func TestDebeziumAvroSchemaRegistryConfig(t *testing.T) { - t.Parallel() - - cfg := NewConfig(config.ProtocolDebeziumAvro) - cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" - require.NoError(t, cfg.Validate()) - - cfg = NewConfig(config.ProtocolDebeziumAvro) - cfg.AvroGlueSchemaRegistry = &config.GlueSchemaRegistryConfig{ - RegistryName: "test-registry", - Region: "us-east-1", - } - require.NoError(t, cfg.Validate()) - - cfg = NewConfig(config.ProtocolDebeziumAvro) - require.ErrorContains( - t, - cfg.Validate(), - `Debezium Avro protocol requires parameter "schema-registry" or "glue-schema-registry"`, - ) - - cfg = NewConfig(config.ProtocolDebeziumAvro) - cfg.AvroGlueSchemaRegistry = &config.GlueSchemaRegistryConfig{} - cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" - require.ErrorContains( - t, - cfg.Validate(), - `Debezium Avro protocol requires only one of "schema-registry" or "glue-schema-registry"`, - ) - - cfg = NewConfig(config.ProtocolDebezium) - cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" - require.ErrorContains(t, cfg.Validate(), `Debezium protocol does not support schema registry`) -} - -func TestDebeziumAvroGlueSchemaRegistryConfig(t *testing.T) { - t.Parallel() - - cfg := NewConfig(config.ProtocolDebeziumAvro) - sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium-avro") + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=avro&avro-include-before-value=true") require.NoError(t, err) - glueSchemaRegistryConfig := &config.GlueSchemaRegistryConfig{ - RegistryName: "test-registry", - Region: "us-east-1", - } - sinkConfig := config.GetDefaultReplicaConfig().Sink - sinkConfig.KafkaConfig = &config.KafkaConfig{ - GlueSchemaRegistryConfig: glueSchemaRegistryConfig, - } - - err = cfg.Apply(sinkURI, sinkConfig) + err = cfg.Apply(sinkURI, &config.SinkConfig{}) require.NoError(t, err) - require.Same(t, glueSchemaRegistryConfig, cfg.AvroGlueSchemaRegistry) - require.Empty(t, cfg.AvroConfluentSchemaRegistry) + require.False(t, cfg.EnableTiDBExtension) + require.True(t, cfg.AvroIncludeBeforeValue) + cfg.AvroConfluentSchemaRegistry = "http://127.0.0.1:8081" require.NoError(t, cfg.Validate()) } -func TestDebeziumAvroWatermarkConfig(t *testing.T) { - t.Parallel() - - cfg := NewConfig(config.ProtocolDebeziumAvro) - sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=debezium-avro&enable-tidb-extension=true&avro-enable-watermark=true") +func TestAvroIncludeBeforeValueConfigFile(t *testing.T) { + sinkURI, err := url.Parse("kafka://127.0.0.1:9092/topic?protocol=avro") require.NoError(t, err) - sinkConfig := config.GetDefaultReplicaConfig().Sink - sinkConfig.SchemaRegistry = util.AddressOf("http://127.0.0.1:8081") - err = cfg.Apply(sinkURI, sinkConfig) + 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.EnableTiDBExtension) - require.True(t, cfg.AvroEnableWatermark) - require.Equal(t, "http://127.0.0.1:8081", cfg.AvroConfluentSchemaRegistry) + require.False(t, cfg.EnableTiDBExtension) + 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 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] }