test(connector): introduce inline e2e test for CDC table schema change (

risingwavelabs#12344)

Signed-off-by: Bugen Zhao <[email protected]>

Makefile.toml

@@ -1292,7 +1292,7 @@ echo "All processes has exited."
 
 [tasks.slt]
 category = "RiseDev - SQLLogicTest"
-install_crate = { version = "0.15.3", crate_name = "sqllogictest-bin", binary = "sqllogictest", test_arg = [
+install_crate = { version = "0.17.0", crate_name = "sqllogictest-bin", binary = "sqllogictest", test_arg = [
   "--help",
 ], install_command = "binstall" }
 command = "sqllogictest"

ci/Dockerfile

@@ -42,7 +42,7 @@ ENV CARGO_REGISTRIES_CRATES_IO_PROTOCOL=sparse
 RUN curl -L --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/cargo-bins/cargo-binstall/main/install-from-binstall-release.sh | bash
 RUN cargo binstall -y --no-symlinks cargo-llvm-cov cargo-nextest cargo-hakari cargo-sort cargo-cache cargo-audit \
     [email protected] \
-    sqllogictest-bin@0.15.3 \
+    sqllogictest-bin@0.17.0 \
     && cargo install sccache \
     && cargo cache -a \
     && rm -rf "/root/.cargo/registry/index" \

ci/build-ci-image.sh

@@ -13,7 +13,7 @@ cat ../rust-toolchain
 # !!! CHANGE THIS WHEN YOU WANT TO BUMP CI IMAGE !!! #
 #          AND ALSO docker-compose.yml               #
 ######################################################
-export BUILD_ENV_VERSION=v20230914
+export BUILD_ENV_VERSION=v20230919
 
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/docker-compose.yml

@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230914
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
     depends_on:
       - mysql
       - db
@@ -81,7 +81,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230914
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
     depends_on:
       - mysql
       - db
@@ -91,12 +91,12 @@ services:
       - ..:/risingwave
 
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230914
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230914
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -107,7 +107,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230914
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20230919
     depends_on:
       db:
         condition: service_healthy

ci/scripts/e2e-source-test.sh

@@ -45,9 +45,9 @@ echo "--- e2e, ci-1cn-1fe, mysql & postgres cdc"
 mysql --host=mysql --port=3306 -u root -p123456 < ./e2e_test/source/cdc/mysql_cdc.sql
 
 # import data to postgres
-export PGPASSWORD='postgres';
-createdb -h db -U postgres cdc_test
-psql -h db -U postgres -d cdc_test < ./e2e_test/source/cdc/postgres_cdc.sql
+export PGHOST=db PGUSER=postgres PGPASSWORD=postgres PGDATABASE=cdc_test
+createdb
+psql < ./e2e_test/source/cdc/postgres_cdc.sql
 
 node_port=50051
 node_timeout=10
@@ -80,6 +80,9 @@ cargo make ci-start ci-1cn-1fe-with-recovery
 echo "waiting for connector node to start"
 wait_for_connector_node_start
 
+echo "--- inline cdc test"
+sqllogictest -p 4566 -d dev './e2e_test/source/cdc_inline/**/*.slt'
+
 echo "--- mysql & postgres cdc validate test"
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.validate.mysql.slt'
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.validate.postgres.slt'
@@ -97,7 +100,7 @@ echo "cluster killed "
 
 # insert new rows
 mysql --host=mysql --port=3306 -u root -p123456 < ./e2e_test/source/cdc/mysql_cdc_insert.sql
-psql -h db -U postgres -d cdc_test < ./e2e_test/source/cdc/postgres_cdc_insert.sql
+psql < ./e2e_test/source/cdc/postgres_cdc_insert.sql
 echo "inserted new rows into mysql and postgres"
 
 # start cluster w/o clean-data

e2e_test/source/cdc_inline/alter/postgres_alter.slt

@@ -0,0 +1,92 @@
+control substitution on
+
+system ok
+psql -c "
+CREATE TABLE alter_test (k BIGINT PRIMARY KEY, v CHARACTER VARYING);
+INSERT INTO alter_test VALUES (1, 'a'), (2, 'b');
+"
+
+statement ok
+CREATE TABLE alter_test (k BIGINT PRIMARY KEY, v CHARACTER VARYING)
+WITH (
+    connector = 'postgres-cdc',
+    hostname = '${PGHOST:localhost}',
+    port = '${PGPORT:5432}',
+    username = '${PGUSER:$USER}',
+    password = '${PGPASSWORD:}',
+    database.name = '${PGDATABASE:postgres}',
+    schema.name = 'public',
+    table.name = 'alter_test',
+    slot.name = 'alter_test'
+)
+
+sleep 5s
+
+query IT
+SELECT * FROM alter_test ORDER BY k
+----
+1 a
+2 b
+
+statement ok
+ALTER TABLE alter_test ADD COLUMN v2 CHARACTER VARYING
+
+system ok
+psql -c "
+INSERT INTO alter_test VALUES (3, 'c');
+"
+
+# FIXME: after schema change in RisingWave, why does it take so long to get the new data?
+sleep 20s
+
+query ITT
+SELECT * FROM alter_test ORDER BY k
+----
+1 a NULL
+2 b NULL
+3 c NULL
+
+system ok
+psql -c "
+ALTER TABLE alter_test ADD COLUMN v2 CHARACTER VARYING;
+INSERT INTO alter_test VALUES (4, 'd', 'dd');
+"
+
+sleep 5s
+
+query ITT
+SELECT * FROM alter_test ORDER BY k
+----
+1 a NULL
+2 b NULL
+3 c NULL
+4 d dd
+
+statement ok
+ALTER TABLE alter_test DROP COLUMN v
+
+system ok
+psql -c "
+ALTER TABLE alter_test DROP COLUMN v;
+INSERT INTO alter_test VALUES (5, 'ee');
+"
+
+# FIXME: after schema change in RisingWave, why does it take so long to get the new data?
+sleep 20s
+
+query IT
+SELECT * FROM alter_test ORDER BY k
+----
+1 NULL
+2 NULL
+3 NULL
+4 dd
+5 ee
+
+statement ok
+DROP TABLE alter_test
+
+system ok
+psql -c "
+DROP TABLE alter_test;
+"

src/tests/simulation/Cargo.toml

@@ -44,7 +44,7 @@ risingwave_sqlsmith = { workspace = true }
 serde = "1.0.188"
 serde_derive = "1.0.188"
 serde_json = "1.0.107"
-sqllogictest = "0.15.3"
+sqllogictest = "0.17.0"
 tempfile = "3"
 tokio = { version = "0.2.23", package = "madsim-tokio" }
 tokio-postgres = "0.7"

src/tests/simulation/src/client.rs

@@ -223,4 +223,10 @@ impl sqllogictest::AsyncDB for RisingWave {
     async fn sleep(dur: Duration) {
         tokio::time::sleep(dur).await
     }
+
+    async fn run_command(
+        _command: std::process::Command,
+    ) -> std::io::Result<std::process::ExitStatus> {
+        unimplemented!("spawning process is not supported in simulation mode")
+    }
 }