sink(ticdc): add ut case for open protocol (#11013)

close #11012
pingcap · Jun 21, 2024 · c001638 · c001638
1 parent 4f7b74c
commit c001638
Show file tree

Hide file tree

Showing 2 changed files with 33 additions and 1 deletion.
diff --git a/pkg/sink/codec/open/open_protocol_encoder_test.go b/pkg/sink/codec/open/open_protocol_encoder_test.go
@@ -461,3 +461,35 @@ func TestOutputOldValueFalse(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, decoded.PreColumns)
 }
+
+func TestNoHandleKeys(t *testing.T) {
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.ForceReplicate = true
+	helper := entry.NewSchemaTestHelperWithReplicaConfig(t, replicaConfig)
+	defer helper.Close()
+
+	// insert
+	_ = helper.DDL2Event(`create table test.t(a varchar(10), b varchar(10))`)
+	event := helper.DML2Event(`insert into test.t values ("aa", "bb")`, "test", "t")
+	codecConfig := common.NewConfig(config.ProtocolOpen)
+	codecConfig.OpenOutputOldValue = false
+	key, value, err := rowChangeToMsg(event, codecConfig, true)
+	require.Error(t, err)
+	require.Nil(t, value)
+	require.Nil(t, key)
+
+	// update
+	event.PreColumns = event.Columns
+	key, value, err = rowChangeToMsg(event, codecConfig, true)
+	require.Error(t, err)
+	require.Nil(t, value)
+	require.Nil(t, key)
+
+	// delete
+	event.PreColumns = event.Columns
+	event.Columns = nil
+	key, value, err = rowChangeToMsg(event, codecConfig, true)
+	require.Error(t, err)
+	require.Nil(t, value)
+	require.Nil(t, key)
+}
diff --git a/pkg/sink/codec/open/open_protocol_message.go b/pkg/sink/codec/open/open_protocol_message.go
@@ -150,7 +150,7 @@ func rowChangeToMsg(
 			value.PreColumns = rowChangeColumns2CodecColumns(e.GetPreColumns(), largeMessageOnlyHandleKeyColumns)
 		}
 		if largeMessageOnlyHandleKeyColumns && (len(value.Update) == 0 ||
-			(len(value.PreColumns) == 0 && !config.OpenOutputOldValue)) {
+			(len(value.PreColumns) == 0 && config.OpenOutputOldValue)) {
 			return nil, nil, cerror.ErrOpenProtocolCodecInvalidData.GenWithStack("not found handle key columns for the update event")
 		}
 		if config.OnlyOutputUpdatedColumns {