test(integration-test): increase the amount of data for mysql-cdc (#9904

)

ci/scripts/integration-tests.sh

@@ -22,6 +22,9 @@ while getopts 'c:f:' opt; do
 done
 shift $((OPTIND -1))
 
+echo "--- ghcr login"
+echo "$GHCR_TOKEN" | docker login ghcr.io -u "$GHCR_USERNAME" --password-stdin
+
 echo "--- install postgresql"
 sudo yum install -y postgresql
 

ci/workflows/integration-tests.yml

@@ -2,6 +2,11 @@ steps:
   - label: "Run Demos {{matrix.testcase}} {{matrix.format}}"
     command: "ci/scripts/integration-tests.sh -c {{matrix.testcase}} -f {{matrix.format}}"
     timeout_in_minutes: 30
+    plugins:
+      - seek-oss/aws-sm#v2.3.1:
+          env:
+            GHCR_USERNAME: ghcr-username
+            GHCR_TOKEN: ghcr-token
     matrix:
       setup:
         testcase:
@@ -76,6 +81,11 @@ steps:
   - label: "Run Demos {{matrix.testcase}} {{matrix.format}}"
     command: "ci/scripts/integration-tests.sh -c {{matrix.testcase}} -f {{matrix.format}}"
     timeout_in_minutes: 30
+    plugins:
+      - seek-oss/aws-sm#v2.3.1:
+          env:
+            GHCR_USERNAME: ghcr-username
+            GHCR_TOKEN: ghcr-token
     matrix:
       setup:
         testcase:

integration_tests/mysql-cdc/create_mv.sql

@@ -1,8 +1,5 @@
-CREATE MATERIALIZED VIEW product_count AS
+CREATE MATERIALIZED VIEW lineitem_count AS
 SELECT
-    product_id,
-    COUNT(*) as product_count
+    COUNT(*) as cnt
 FROM
-    orders
-GROUP BY
-    product_id;
+    lineitem_rw;

integration_tests/mysql-cdc/create_source.sql

@@ -1,18 +1,28 @@
-create table orders (
-    order_id int,
-    order_date bigint,
-    customer_name varchar,
-    price decimal,
-    product_id int,
-    order_status smallint,
-    PRIMARY KEY (order_id)
-) with (
+CREATE TABLE lineitem_rw (
+   L_ORDERKEY BIGINT,
+   L_PARTKEY BIGINT,
+   L_SUPPKEY BIGINT,
+   L_LINENUMBER BIGINT,
+   L_QUANTITY DECIMAL,
+   L_EXTENDEDPRICE DECIMAL,
+   L_DISCOUNT DECIMAL,
+   L_TAX DECIMAL,
+   L_RETURNFLAG VARCHAR,
+   L_LINESTATUS VARCHAR,
+   L_SHIPDATE DATE,
+   L_COMMITDATE DATE,
+   L_RECEIPTDATE DATE,
+   L_SHIPINSTRUCT VARCHAR,
+   L_SHIPMODE VARCHAR,
+   L_COMMENT VARCHAR,
+   PRIMARY KEY(L_ORDERKEY, L_LINENUMBER)
+) WITH (
     connector = 'mysql-cdc',
     hostname = 'mysql',
     port = '3306',
     username = 'root',
     password = '123456',
     database.name = 'mydb',
-    table.name = 'orders',
-    server.id = '1'
+    table.name = 'lineitem',
+    server.id = '2'
 );

integration_tests/mysql-cdc/data_check

@@ -1 +1 @@
-orders,product_count
+lineitem_count

integration_tests/mysql-cdc/docker-compose.yml

@@ -36,7 +36,7 @@ services:
   mysql:
     image: mysql:8.0
     ports:
-      - "3306:3306"
+      - "8306:3306"
     environment:
       - MYSQL_ROOT_PASSWORD=123456
       - MYSQL_USER=mysqluser
@@ -53,15 +53,10 @@ services:
       file: ../../docker/docker-compose.yml
       service: connector-node
   datagen:
-    image: mysql:8.0
+    image: ghcr.io/risingwavelabs/go-tpc:v0.1
     depends_on: 
       - mysql
-    command:
-      - /bin/sh
-      - -c
-      - "mysql -p123456 -h mysql mydb < mysql_prepare.sql"
-    volumes:
-      - "./mysql_prepare.sql:/mysql_prepare.sql"      
+    command: tpch prepare --sf 1 -H mysql -U root -p '123456' -D mydb -P 3306
     container_name: datagen
     restart: on-failure
 volumes:

integration_tests/mysql-cdc/query.sql

@@ -1,6 +1,6 @@
 SELECT
     *
 FROM
-    orders
+    lineitem_count
 LIMIT
     10;

integration_tests/scripts/run_demos.py

@@ -19,14 +19,14 @@ def run_sql_file(f: str, dir: str):
         sys.exit(1)
 
 
-def run_demo(demo: str, format: str):
+def run_demo(demo: str, format: str, wait_time = 40):
     file_dir = dirname(abspath(__file__))
     project_dir = dirname(file_dir)
     demo_dir = os.path.join(project_dir, demo)
     print("Running demo: {}".format(demo))
 
     subprocess.run(["docker", "compose", "up", "-d"], cwd=demo_dir, check=True)
-    sleep(40)
+    sleep(wait_time)
 
     sql_files = ['create_source.sql', 'create_mv.sql', 'query.sql']
     for fname in sql_files:

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/MySqlValidator.java

@@ -140,7 +140,7 @@ private void validateTableSchema() throws SQLException {
                 }
 
                 var srcColName = tableSchema.getColumnNames()[index++];
-                if (!srcColName.equals(field)) {
+                if (!srcColName.equalsIgnoreCase(field)) {
                     throw ValidatorUtils.invalidArgument(
                             String.format("column name mismatch: %s, [%s]", field, srcColName));
                 }
@@ -150,7 +150,8 @@ private void validateTableSchema() throws SQLException {
                             String.format("incompatible data type of column %s", srcColName));
                 }
                 if (key.equalsIgnoreCase("PRI")) {
-                    pkFields.add(field);
+                    // RisingWave always use lower case for column name
+                    pkFields.add(field.toLowerCase());
                 }
             }
 

java/connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/source/common/PostgresValidator.java

@@ -124,7 +124,8 @@ private void validateTableSchema() throws SQLException {
             var pkFields = new HashSet<String>();
             while (res.next()) {
                 var name = res.getString(1);
-                pkFields.add(name);
+                // RisingWave always use lower case for column name
+                pkFields.add(name.toLowerCase());
             }
 
             if (!ValidatorUtils.isPrimaryKeyMatch(tableSchema, pkFields)) {
@@ -146,7 +147,7 @@ private void validateTableSchema() throws SQLException {
                 }
 
                 var srcColName = tableSchema.getColumnNames()[index++];
-                if (!srcColName.equals(field)) {
+                if (!srcColName.equalsIgnoreCase(field)) {
                     throw ValidatorUtils.invalidArgument(
                             "table column defined in the source mismatches upstream column "
                                     + field);