Merge branch 'main' into yiming/remote-log-sinker-refactor

ci/scripts/deterministic-it-test.sh

@@ -16,7 +16,6 @@ mv target/ci-sim target/sim
 echo "--- Run integration tests in deterministic simulation mode"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} NEXTEST_PROFILE=ci-sim \
  cargo nextest run \
- --no-capture \
  --no-fail-fast \
  --cargo-metadata target/nextest/cargo-metadata.json \
  --binaries-metadata target/nextest/binaries-metadata.json \

ci/scripts/e2e-source-test.sh

@@ -66,6 +66,11 @@ sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.load.slt'
 sleep 10
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.check.slt'
 
+# cdc share stream test cases
+export MYSQL_HOST=mysql MYSQL_TCP_PORT=3306 MYSQL_PWD=123456
+sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.share_stream.slt'
+
+
 # kill cluster and the connector node
 cargo make kill
 echo "cluster killed "

e2e_test/source/cdc/cdc.share_stream.slt

@@ -0,0 +1,80 @@
+control substitution on
+
+# create database and tables in mysql
+system ok
+mysql --protocol=tcp -u root -e "DROP DATABASE IF EXISTS mytest; CREATE DATABASE mytest;"
+
+system ok
+mysql --protocol=tcp -u root mytest < e2e_test/source/cdc/mysql_create.sql
+
+# enable cdc backfill in ci
+statement ok
+set cdc_backfill='true';
+
+# create a cdc source job, which format fixed to `FORMAT PLAIN ENCODE JSON`
+statement ok
+create source mysql_mytest with (
+ connector = 'mysql-cdc',
+ hostname = '${MYSQL_HOST:localhost}',
+ port = '${MYSQL_TCP_PORT:8306}',
+ username = 'dbz',
+ password = '${MYSQL_PWD:}',
+ database.name = 'mytest',
+ server.id = '5601'
+);
+
+statement ok
+create table products_test ( id INT,
+ name STRING,
+ description STRING,
+ PRIMARY KEY (id)
+) from mysql_mytest table 'mytest.products';
+
+statement ok
+create table orders_test (
+   order_id int,
+   order_date timestamp,
+   customer_name string,
+   price decimal,
+   product_id int,
+   order_status smallint,
+   PRIMARY KEY (order_id)
+) from mysql_mytest table 'mytest.orders';
+
+statement ok
+create materialized view products_test_cnt as select count(*) as cnt from products_test;
+
+statement ok
+create materialized view orders_test_cnt as select count(*) as cnt from orders_test;
+
+
+# generate data to mysql
+system ok
+mysql --protocol=tcp -u root mytest < e2e_test/source/cdc/mysql_init_data.sql
+
+sleep 5s
+
+# check ingestion results
+query I
+SELECT * from products_test_cnt
+----
+9
+
+query I
+SELECT * from orders_test_cnt
+----
+3
+
+query ITT
+SELECT * FROM products_test order by id limit 3
+----
+101 scooter Small 2-wheel scooter
+102 car battery 12V car battery
+103 12-pack drill bits 12-pack of drill bits with sizes ranging from #40 to #3
+
+query ITTT
+SELECT order_id,order_date,customer_name,product_id FROM orders_test order by order_id limit 3
+----
+10001 2020-07-30 10:08:22 Jark 102
+10002 2020-07-30 10:11:09 Sally 105
+10003 2020-07-30 12:00:30 Edward 106

e2e_test/source/cdc/cdc.validate.mysql.slt

@@ -68,6 +68,22 @@ create table products ( id INT,
  server.id = '5085'
 );
 
+# empty table name
+statement error
+create table products ( id INT,
+ name STRING,
+ description STRING,
+ PRIMARY KEY (id)
+) with (
+ connector = 'mysql-cdc',
+ hostname = 'mysql',
+ port = '3306',
+ username = 'root',
+ password = '123456',
+ database.name = 'my@db',
+ server.id = '5085'
+);
+
 # invalid primary key
 statement error
 create table orders (

e2e_test/source/cdc/mysql_create.sql

@@ -0,0 +1,26 @@
+CREATE TABLE products (
+    id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    name VARCHAR(255) NOT NULL,
+    description VARCHAR(512)
+) AUTO_INCREMENT = 101;
+
+CREATE TABLE orders (
+    order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+    order_date DATETIME NOT NULL,
+    customer_name VARCHAR(255) NOT NULL,
+    price DECIMAL(10, 5) NOT NULL,
+    product_id INTEGER NOT NULL,
+    order_status BOOLEAN NOT NULL -- Whether order has been placed
+) AUTO_INCREMENT = 10001;
+
+CREATE TABLE mytable (
+    v1 INTEGER NOT NULL PRIMARY KEY,
+    v2 INTEGER NOT NULL,
+    v3 VARCHAR(255) NOT NULL
+);
+
+DROP USER IF EXISTS 'dbz'@'%';
+CREATE USER 'dbz'@'%' IDENTIFIED BY '123456';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'dbz'@'%';
+
+CREATE TABLE tt3 (v1 int primary key, v2 timestamp);

e2e_test/source/cdc/mysql_init_data.sql

@@ -0,0 +1,28 @@
+-- USE `my@db`;
+
+INSERT INTO products
+VALUES (default,"scooter","Small 2-wheel scooter"),
+       (default,"car battery","12V car battery"),
+       (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
+       (default,"hammer","12oz carpenter's hammer"),
+       (default,"hammer","14oz carpenter's hammer"),
+       (default,"hammer","16oz carpenter's hammer"),
+       (default,"rocks","box of assorted rocks"),
+       (default,"jacket","water resistent black wind breaker"),
+       (default,"spare tire","24 inch spare tire");
+
+
+INSERT INTO orders
+VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
+       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
+       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
+
+
+INSERT INTO mytable
+VALUES (1,1,'no'),
+       (2,2,'no'),
+       (3,3,'no'),
+       (4,4,'no');
+
+INSERT INTO tt3 VALUES (1, '2020-07-30 10:08:22');
+INSERT INTO tt3 VALUES (2, '2020-07-31 10:09:22');

grafana/risingwave-dev-dashboard.dashboard.py

@@ -1518,6 +1518,10 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_compute_error_count')}) by (error_type, error_msg, fragment_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: fragment_id={{fragment_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_compute_error')}) by (error_type, error_msg, fragment_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: fragment_id={{fragment_id}})",
+                        ),
                     ],
                 ),
                 panels.timeseries_count(
@@ -1528,6 +1532,10 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_source_error_count')}) by (error_type, error_msg, fragment_id, table_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: table_id={{table_id}}, fragment_id={{fragment_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_source_error')}) by (error_type, error_msg, fragment_id, table_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: table_id={{table_id}}, fragment_id={{fragment_id}})",
+                        ),
                     ],
                 ),
                 panels.timeseries_count(
@@ -1538,6 +1546,20 @@ def section_streaming_errors(outer_panels):
                             f"sum({metric('user_source_reader_error_count')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
                             "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
                         ),
+                        panels.target(
+                            f"sum({metric('user_source_reader_error')}) by (error_type, error_msg, actor_id, source_id, executor_name)",
+                            "{{error_type}}: {{error_msg}} ({{executor_name}}: actor_id={{actor_id}}, source_id={{source_id}})",
+                        ),
+                    ],
+                ),
+                panels.timeseries_count(
+                    "Sink by Connector",
+                    "",
+                    [
+                        panels.target(
+                            f"sum({metric('user_sink_error')}) by (connector_name, executor_id, error_msg)",
+                            "{{connector_name}}: {{error_msg}} ({{executor_id}})",
+                        ),
                     ],
                 ),
             ],