test(source): add tidb-cdc compatible test (#16206)

integration_tests/tidb-cdc-sink/create_source.sql

@@ -56,6 +56,34 @@ create table datatype (
       scan.startup.mode='earliest'
 ) FORMAT CANAL ENCODE JSON;
 
+CREATE TABLE tidb_cdc_types (
+    id integer PRIMARY KEY,
+    c_boolean boolean,
+    c_bit boolean,
+    c_tinyint smallint,
+    c_smallint smallint,
+    c_mediumint integer,
+    c_integer integer,
+    c_bigint bigint,
+    c_decimal decimal,
+    c_float real,
+    c_double double,
+    c_char_255 varchar,
+    c_varchar_10000 varchar,
+    c_binary_255 bytea,
+    c_varbinary_10000 bytea,
+    c_date date,
+    c_time time,
+    c_datetime timestamp,
+    c_timestamp timestamp,
+    c_json jsonb
+) WITH (
+      connector='kafka',
+      topic='ticdc_test_tidb_cdc_types',
+      properties.bootstrap.server='kafka:9092',
+      scan.startup.mode='earliest'
+) FORMAT CANAL ENCODE JSON;
+
 create table tidb_sink_datatypes (
   id integer PRIMARY KEY,
   c_boolean boolean,

integration_tests/tidb-cdc-sink/data_check

@@ -1 +1 @@
-tweet,user,datatype
+tweet,user,datatype,tidb_cdc_types

integration_tests/tidb-cdc-sink/tidb_create_tables.sql

@@ -77,3 +77,26 @@ CREATE TABLE tidb_sink_datatypes (
   c_double_precision_array LONGTEXT,
   c_varchar_array LONGTEXT
 );
+
+CREATE TABLE tidb_cdc_types (
+  id integer PRIMARY KEY,
+  c_boolean boolean,
+  c_bit bit,
+  c_tinyint tinyint,
+  c_smallint smallint,
+  c_mediumint mediumint,
+  c_integer integer,
+  c_bigint bigint,
+  c_decimal decimal,
+  c_float float,
+  c_double double,
+  c_char_255 char(255),
+  c_varchar_10000 varchar(10000),
+  c_binary_255 binary(255),
+  c_varbinary_10000 varbinary(10000),
+  c_date date,
+  c_time time(6),
+  c_datetime datetime(6),
+  c_timestamp timestamp(6),
+  c_json JSON
+);