sink: add before field for avro protocol

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

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"`
 }
 

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))
+}

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"`
 }

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)