add ut

pkg/sink/codec/open/open_protocol_encoder_test.go

@@ -225,6 +225,7 @@ func TestEncodeDecodeE2E(t *testing.T) {
 	topic := "test"
 
 	codecConfig := common.NewConfig(config.ProtocolOpen)
+	codecConfig.OpenOutputOldValue = false
 	builder, err := NewBatchEncoderBuilder(ctx, codecConfig)
 	require.NoError(t, err)
 	encoder := builder.Build()
@@ -422,3 +423,41 @@ func TestE2EClaimCheckMessage(t *testing.T) {
 		require.Equal(t, column.Value, decodedColumn.Value, colName)
 	}
 }
+
+func TestOutputOldValueFalse(t *testing.T) {
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	_ = helper.DDL2Event(`create table test.t(a varchar(10) primary key, b varchar(10))`)
+	event := helper.DML2Event(`insert into test.t values ("aa", "bb")`, "test", "t")
+	event.PreColumns = event.Columns
+
+	ctx := context.Background()
+	topic := "test"
+
+	codecConfig := common.NewConfig(config.ProtocolOpen)
+	codecConfig.OpenOutputOldValue = false
+	builder, err := NewBatchEncoderBuilder(ctx, codecConfig)
+	require.NoError(t, err)
+	encoder := builder.Build()
+
+	err = encoder.AppendRowChangedEvent(ctx, topic, event, func() {})
+	require.NoError(t, err)
+
+	message := encoder.Build()[0]
+
+	decoder, err := NewBatchDecoder(ctx, codecConfig, nil)
+	require.NoError(t, err)
+
+	err = decoder.AddKeyValue(message.Key, message.Value)
+	require.NoError(t, err)
+
+	messageType, hasNext, err := decoder.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, messageType, model.MessageTypeRow)
+
+	decoded, err := decoder.NextRowChangedEvent()
+	require.NoError(t, err)
+	require.Nil(t, decoded.PreColumns)
+}