add e2e tests

e2e_test/source_legacy/cdc/cdc.share_stream.slt

@@ -4,9 +4,20 @@ control substitution on
 system ok
 mysql --protocol=tcp -u root -e "DROP DATABASE IF EXISTS mytest; CREATE DATABASE mytest;"
 
+system ok
+mysql --protocol=tcp -u root -e "DROP DATABASE IF EXISTS kdb; CREATE DATABASE kdb;"
+
+system ok
+mysql --protocol=tcp -u root -e "DROP DATABASE IF EXISTS kdb; CREATE DATABASE kdb;"
+
 system ok
 mysql --protocol=tcp -u root mytest < e2e_test/source_legacy/cdc/mysql_create.sql
 
+system ok
+mysql --protocol=tcp -u root kdb -e " CREATE TABLE kt1 (id int primary key, v1 varchar);
+                                      INSERT INTO kt1 VALUES (1,'aaa'),(2,'bbb');
+"
+
 # generate data to mysql
 system ok
 mysql --protocol=tcp -u root mytest < e2e_test/source_legacy/cdc/mysql_init_data.sql
@@ -24,7 +35,7 @@ create source mysql_mytest with (
  port = '${MYSQL_TCP_PORT:8306}',
  username = 'rwcdc',
  password = secret mysql_pwd,
- database.name = 'mytest',
+ database.name = 'mytest,kdb',
  server.id = '5601'
 );
 
@@ -112,6 +123,10 @@ create table orders_test (
    PRIMARY KEY (order_id)
 ) from mysql_mytest table 'mytest.orders';
 
+
+statement ok
+create table kt1 (*) from mysql_mytest table 'kdb.kt1';
+
 statement ok
 create materialized view products_test_cnt as select count(*) as cnt from rw.products_test;
 
@@ -121,6 +136,9 @@ create materialized view orders_test_cnt as select count(*) as cnt from orders_t
 system ok
 mysql --protocol=tcp -u root mytest -e "INSERT INTO products VALUES(default, 'Juice', '100ml Juice');"
 
+system ok
+mysql --protocol=tcp -u root kdb -e "INSERT INTO kt1 VALUES (3, 'ccc'),(4, 'ddd');"
+
 system ok
 mysql --protocol=tcp -u root mytest -e "FLUSH LOGS"
 
@@ -146,6 +164,16 @@ select count(*) from orders_no_backfill
 ----
 0
 
+
+query IT
+select * from kt1 order by id;
+----
+1 aaa
+2 bbb
+3 ccc
+4 ddd
+
+
 # check ingestion results
 query I
 SELECT * from products_test_cnt

e2e_test/source_legacy/cdc/mysql_cdc.sql

@@ -62,3 +62,7 @@ FLUSH PRIVILEGES;
 CREATE TABLE tt3 (v1 int primary key, v2 timestamp);
 INSERT INTO tt3 VALUES (1, '2020-07-30 10:08:22');
 INSERT INTO tt3 VALUES (2, '2020-07-31 10:09:22');
+
+USE `kdb`;
+CREATE TABLE kt1 (id int primary key, v1 varchar);
+INSERT INTO kt1 VALUES (1,'aaa'),(2,'bbb');