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

close #12208

Author: ti-chi-bot

pkg/sink/codec/debezium/codec.go

@@ -49,6 +49,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 {
@@ -1088,6 +1091,9 @@ func (c *dbzCodec) EncodeValue(
 						}
 						colInfos := e.TableInfo.GetColInfosForRowChangedEvent()
 						for i, col := range validCols {
+							if col == nil {
+								continue
+							}
 							colx := model.GetColumnDataX(col, e.TableInfo)
 							c.writeDebeziumFieldSchema(fieldsWriter, colx, colInfos[i].Ft)
 						}

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

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

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
-);

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
-);

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

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
-);

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
-);

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

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
-);