test: integrate mysql (user-managed) into risedev and inline source tests

Makefile.toml

@@ -1308,7 +1308,6 @@ echo "All processes has exited."
 """
 
 [tasks.slt]
-env = { SLT_HOST = "${RISEDEV_RW_FRONTEND_LISTEN_ADDRESS}", SLT_PORT = "${RISEDEV_RW_FRONTEND_PORT}", SLT_DB = "dev" }
 category = "RiseDev - Test - SQLLogicTest"
 install_crate = { version = "0.20.1", crate_name = "sqllogictest-bin", binary = "sqllogictest", test_arg = [
   "--help",
@@ -1319,6 +1318,19 @@ command = "sqllogictest"
 args = ["${@}"]
 description = "🌟 Run SQLLogicTest"
 
+[tasks.slt.env]
+SLT_HOST = "${RISEDEV_RW_FRONTEND_LISTEN_ADDRESS}"
+SLT_PORT = "${RISEDEV_RW_FRONTEND_PORT}"
+SLT_DB = "dev"
+PATH = "${PWD}/e2e_test/commands:${PATH}"
+
+[tasks.slt-clean]
+category = "RiseDev - Test - SQLLogicTest"
+dependencies = ["clean-kafka", "reset-rw"]
+description = "Run SQLLogicTest with a clean environment"
+run_task = "slt"
+args = ["${@}"]
+
 [tasks.slt-streaming]
 category = "RiseDev - Test - SQLLogicTest"
 extend = "slt"
@@ -1433,3 +1445,14 @@ script = """
 vars = dump_variables
 echo ${vars}
 """
+
+[tasks.show-risedev-env]
+description = "Show risedev-env environment variables"
+dependencies = ["check-risedev-env-file"]
+script = '''
+#!/usr/bin/env bash
+set -euo pipefail
+cat ${PREFIX_CONFIG}/risedev-env
+echo "Hint: To load the environment variables into the shell, you may run:"
+echo "$(tput setaf 4)\set -a; source ${PREFIX_CONFIG}/risedev-env; set +a$(tput sgr0)"
+'''

ci/scripts/e2e-source-test.sh

@@ -31,6 +31,12 @@ buildkite-agent artifact download risingwave-connector.tar.gz ./
 mkdir ./connector-node
 tar xf ./risingwave-connector.tar.gz -C ./connector-node
 
+echo "--- e2e, inline test"
+risedev ci-start ci-inline-source-test
+risedev slt './e2e_test/source_inline/**/*.slt'
+echo "--- Kill cluster"
+risedev ci-kill
+
 echo "--- Prepare data"
 cp src/connector/src/test_data/simple-schema.avsc ./avro-simple-schema.avsc
 cp src/connector/src/test_data/complex-schema.avsc ./avro-complex-schema.avsc
@@ -82,15 +88,6 @@ echo "--- cdc share source test"
 export MYSQL_HOST=mysql MYSQL_TCP_PORT=3306 MYSQL_PWD=123456
 risedev slt './e2e_test/source/cdc/cdc.share_stream.slt'
 
-# create a share source and check whether heartbeat message is received
-risedev slt './e2e_test/source/cdc/cdc.create_source_job.slt'
-table_id=$(psql -U root -h localhost -p 4566 -d dev -t -c "select id from rw_internal_tables where name like '%mysql_source%';" | xargs);
-table_count=$(psql -U root -h localhost -p 4566 -d dev -t -c "select count(*) from rw_table(${table_id}, public);" | xargs);
-if [ "$table_count" -eq 0 ]; then
-    echo "ERROR: internal table of cdc share source is empty!"
-    exit 1
-fi
-
 echo "--- mysql & postgres load and check"
 risedev slt './e2e_test/source/cdc/cdc.load.slt'
 # wait for cdc loading
@@ -168,9 +165,6 @@ echo "--- e2e, kafka alter source again"
 ./scripts/source/prepare_data_after_alter.sh 3
 risedev slt './e2e_test/source/basic/alter/kafka_after_new_data_2.slt'
 
-echo "--- e2e, inline test"
-risedev slt './e2e_test/source_inline/**/*.slt'
-
 echo "--- Run CH-benCHmark"
 risedev slt './e2e_test/ch_benchmark/batch/ch_benchmark.slt'
 risedev slt './e2e_test/ch_benchmark/streaming/*.slt'

e2e_test/commands/README.md

@@ -0,0 +1,8 @@
+# "built-in" Sqllogictest system commands
+
+Scripts (executables) in this directory are expected to be used as `system` commands in sqllogictests. You can use any language like bash, python, zx.
+
+They will be loaded in `PATH` by `risedev slt`, and thus function as kind of "built-in" commands.
+
+Only general commands should be put here.
+If the script is ad-hoc (only used for one test), it's better to put next to the test file.

e2e_test/commands/internal_table.mjs

@@ -0,0 +1,67 @@
+#!/usr/bin/env zx
+
+// zx: A tool for writing better scripts
+// https://google.github.io/zx/
+
+const {
+  name: job_name,
+  type: table_type,
+  count: count,
+} = minimist(process.argv.slice(3), {
+  string: ["name", "type"],
+  boolean: ["count"],
+});
+
+// Return an array of CSV string
+async function psql(query) {
+  return (
+    await $`
+psql -h $RISEDEV_RW_FRONTEND_LISTEN_ADDRESS -p $RISEDEV_RW_FRONTEND_PORT -U root -d dev \
+--csv --tuples-only -c ${query}
+`
+  )
+    .toString()
+    .trim()
+    .split("\n")
+    .filter((line) => line.trim() != "");
+}
+
+// If `table_type` is null, return all internal tables for the job.
+// If `job_name` is null, return all jobs' internal tables.
+async function select_internal_table(job_name, table_type) {
+  // Note: if we have `t1`, and `t1_balabala`, the latter one will also be matched 😄.
+  const internal_tables = await psql(
+    `select name from rw_internal_tables where name like '__internal_${job_name}_%_${table_type}_%'`
+  );
+  if (internal_tables.length == 0) {
+    throw new Error(
+      `No internal tables found for the pattern '__internal_${job_name}_%_${table_type}_%'`
+    );
+  }
+
+  const res = new Map(
+    await Promise.all(
+      internal_tables.map(async (t) => {
+        let rows = await psql(`select * from ${t}`);
+        return [t, rows];
+      })
+    )
+  );
+  return res;
+}
+
+const tables = await select_internal_table(job_name, table_type);
+for (const [table_name, rows] of tables) {
+  if (tables.size > 1) {
+    console.log(`Table: ${table_name}`);
+  }
+  if (count) {
+    console.log(`count: ${rows.length}`);
+  } else {
+    if (rows.length == 0) {
+      console.log("(empty)");
+    } else {
+      console.log(`${rows.join("\n")}`);
+    }
+  }
+}

e2e_test/source/cdc/mysql_cdc.sql

@@ -49,9 +49,11 @@ VALUES (1,1,'no'),
        (3,3,'no'),
        (4,4,'no');
 
+-- This user is for non-shared CDC
 CREATE USER 'dbz'@'%' IDENTIFIED BY '123456';
 GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'dbz'@'%';
 
+-- This user is for shared CDC
 CREATE USER 'rwcdc'@'%' IDENTIFIED BY '123456';
 GRANT SELECT, RELOAD, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'rwcdc'@'%';
 

e2e_test/source_inline/README.md

@@ -26,5 +26,5 @@ risedev slt 'e2e_test/source_inline/**/*.slt'
 To write tests, please ensure each file is self-contained and does not depend on running external scripts to setup the environment.
 
 Use `system` command to setup instead.
-For simple cases, you can directly write a bash command;
-For more complex cases, you can write a test script (with any language like bash, python, zx), and invoke it in the `system` command.
+- For simple cases, you can directly write a bash command;
+- For more complex cases, you can write a test script. See also [e2e_test/commands/README.md](../commands/README.md)