codec: fix a panic in Debezium when configuring column-selector (#12210)

wk989898 · web-flow · commit 9bc740d55a9d · 2025-07-04T08:19:30.000Z
close #12208
diff --git a/pkg/sink/codec/debezium/codec.go b/pkg/sink/codec/debezium/codec.go
@@ -50,6 +50,9 @@ func (c *dbzCodec) writeDebeziumFieldValues(
 	colInfos := tableInfo.GetColInfosForRowChangedEvent()
 	writer.WriteObjectField(fieldName, func() {
 		for i, col := range cols {
+			if col == nil {
+				continue
+			}
 			colx := model.GetColumnDataX(col, tableInfo)
 			err = c.writeDebeziumFieldValue(writer, colx, colInfos[i].Ft)
 			if err != nil {
@@ -1038,6 +1041,9 @@ func (c *dbzCodec) EncodeValue(
 							validCols = e.Columns
 						}
 						for _, col := range validCols {
+							if col == nil {
+								continue
+							}
 							colx := model.GetColumnDataX(col, e.TableInfo)
 							ft := &e.TableInfo.GetColumnByID(colx.ColumnID).FieldType
 							c.writeDebeziumFieldSchema(fieldsWriter, colx, ft)
diff --git a/tests/integration_tests/_utils/check_table_not_exists b/tests/integration_tests/_utils/check_table_not_exists
@@ -19,11 +19,11 @@ while [ $i -lt $check_time ]; do
 		break
 	fi
 	((i++))
-	echo "table $1 not exists for $i-th check, retry later"
+	echo "table $1 exists for $i-th check, retry later"
 	sleep 2
 done
 
 if [ $i -ge $check_time ]; then
-	echo "table $1 not exists at last check"
+	echo "table $1 exists at last check"
 	exit 1
 fi
diff --git a/tests/integration_tests/canal_json_adapter_compatibility/run.sh b/tests/integration_tests/canal_json_adapter_compatibility/run.sh
@@ -33,11 +33,13 @@ function run() {
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.binary_columns ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 600
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml 600
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.binary_columns ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 600
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml 600
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/canal_json_basic/data/data.sql b/tests/integration_tests/canal_json_basic/data/data.sql
@@ -323,8 +323,3 @@ CREATE TABLE 表1 (
 RENAME TABLE 表1 TO 表2;
 
 DROP TABLE 表2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/canal_json_basic/data/data_gbk.sql b/tests/integration_tests/canal_json_basic/data/data_gbk.sql
@@ -94,8 +94,3 @@ CREATE TABLE 
 ) ENGINE = InnoDB;
 
 RENAME TABLE ��1 TO ��2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/canal_json_basic/run.sh b/tests/integration_tests/canal_json_basic/run.sh
@@ -50,11 +50,13 @@ function run() {
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/canal_json_content_compatible/data/data.sql b/tests/integration_tests/canal_json_content_compatible/data/data.sql
@@ -323,8 +323,3 @@ CREATE TABLE 表1 (
 RENAME TABLE 表1 TO 表2;
 
 DROP TABLE 表2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/canal_json_content_compatible/data/data_gbk.sql b/tests/integration_tests/canal_json_content_compatible/data/data_gbk.sql
@@ -94,8 +94,3 @@ CREATE TABLE 
 ) ENGINE = InnoDB;
 
 RENAME TABLE ��1 TO ��2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/canal_json_content_compatible/run.sh b/tests/integration_tests/canal_json_content_compatible/run.sh
@@ -50,11 +50,13 @@ function run() {
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/debezium_basic/conf/changefeed.toml b/tests/integration_tests/debezium_basic/conf/changefeed.toml
@@ -0,0 +1,4 @@
+[sink]
+column-selectors = [
+    {matcher = ['test.t1'], columns = ['account_id']},
+]
diff --git a/tests/integration_tests/debezium_basic/data/data.sql b/tests/integration_tests/debezium_basic/data/data.sql
@@ -323,8 +323,3 @@ CREATE TABLE 表1 (
 RENAME TABLE 表1 TO 表2;
 
 DROP TABLE 表2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/debezium_basic/data/data_gbk.sql b/tests/integration_tests/debezium_basic/data/data_gbk.sql
@@ -94,8 +94,3 @@ CREATE TABLE 
 ) ENGINE = InnoDB;
 
 RENAME TABLE ��1 TO ��2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/debezium_basic/data/test.sql b/tests/integration_tests/debezium_basic/data/test.sql
@@ -0,0 +1,7 @@
+DROP DATABASE IF EXISTS test;
+CREATE DATABASE test;
+USE test;
+
+create table t1 (id int primary key, account_id int not null);
+alter table t1 add unique key(account_id);
+insert into t1 values (12,34);
diff --git a/tests/integration_tests/debezium_basic/run.sh b/tests/integration_tests/debezium_basic/run.sh
@@ -28,22 +28,32 @@ function run() {
 
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=debezium&enable-tidb-extension=true"
 
-	run_cdc_cli changefeed create --sink-uri="$SINK_URI"
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI" --config=$CUR/conf/changefeed.toml
 	sleep 5 # wait for changefeed to start
 	# determine the sink uri and run corresponding consumer
-	# currently only kafka and pulsar are supported
-	run_kafka_consumer $WORK_DIR $SINK_URI
+	run_kafka_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml
 
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-
-	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
+	run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "CREATE TABLE test.finish_mark3 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark3 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	echo "Starting build checksum checker..."
+	cd $CUR/../../utils/checksum_checker
+	if [ ! -f ./checksum_checker ]; then
+		GO111MODULE=on go build
+	fi
+	./checksum_checker --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" --databases "test" --config="$CUR/conf/changefeed.toml"
+
 	cleanup_process $CDC_BINARY
 }
 
diff --git a/tests/integration_tests/kafka_simple_basic/data/data.sql b/tests/integration_tests/kafka_simple_basic/data/data.sql
@@ -407,8 +407,3 @@ CREATE TABLE 表1 (
 RENAME TABLE 表1 TO 表2;
 
 DROP TABLE 表2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/kafka_simple_basic/data/data_gbk.sql b/tests/integration_tests/kafka_simple_basic/data/data_gbk.sql
@@ -94,8 +94,3 @@ CREATE TABLE 
 ) ENGINE = InnoDB;
 
 RENAME TABLE ��1 TO ��2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/kafka_simple_basic/run.sh b/tests/integration_tests/kafka_simple_basic/run.sh
@@ -50,11 +50,13 @@ function run() {
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/kafka_simple_basic_avro/data/data.sql b/tests/integration_tests/kafka_simple_basic_avro/data/data.sql
@@ -456,8 +456,3 @@ CREATE TABLE 表1 (
 RENAME TABLE 表1 TO 表2;
 
 DROP TABLE 表2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/kafka_simple_basic_avro/data/data_gbk.sql b/tests/integration_tests/kafka_simple_basic_avro/data/data_gbk.sql
@@ -94,8 +94,3 @@ CREATE TABLE 
 ) ENGINE = InnoDB;
 
 RENAME TABLE ��1 TO ��2;
-
-create table finish_mark
-(
-    id int PRIMARY KEY
-);
diff --git a/tests/integration_tests/kafka_simple_basic_avro/run.sh b/tests/integration_tests/kafka_simple_basic_avro/run.sh
@@ -38,11 +38,13 @@ function run() {
 	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark1 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	run_sql_file $CUR/data/data_gbk.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	check_table_exists test.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
+	run_sql "CREATE TABLE test.finish_mark2 (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	check_table_exists test.finish_mark2 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 200
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY

-Original file line number
+Diff line change
@@ @@ -0,0 +1,4 @@ @@
 +[sink]
 +column-selectors = [
 +    {matcher = ['test.t1'], columns = ['account_id']},
 +]