From 9ef40ba54a3cfce9f4946440ef78c8020b1b72f5 Mon Sep 17 00:00:00 2001
From: xxchan <xxchan22f@gmail.com>
Date: Fri, 11 Oct 2024 19:14:46 +0800
Subject: [PATCH] test: move kafka alter source test to source_inline (#18867)

Signed-off-by: xxchan <xxchan22f@gmail.com>
---
 ci/scripts/e2e-source-test.sh                 |  14 -
 e2e_test/source/basic/alter/kafka.slt         | 139 ---------
 .../basic/alter/kafka_after_new_data.slt      |  82 ------
 .../basic/alter/kafka_after_new_data_2.slt    |  14 -
 .../source_inline/kafka/alter/add_column.slt  | 271 ++++++++++++++++++
 .../kafka}/alter/rate_limit_source_kafka.slt  |  16 +-
 .../kafka}/alter/rate_limit_table_kafka.slt   |  14 +-
 scripts/source/alter_data/kafka_alter.2       |   1 -
 scripts/source/alter_data/kafka_alter.3       |   1 -
 scripts/source/prepare_data_after_alter.sh    |  11 -
 scripts/source/test_data/kafka_alter.1        |   1 -
 11 files changed, 286 insertions(+), 278 deletions(-)
 delete mode 100644 e2e_test/source/basic/alter/kafka.slt
 delete mode 100644 e2e_test/source/basic/alter/kafka_after_new_data.slt
 delete mode 100644 e2e_test/source/basic/alter/kafka_after_new_data_2.slt
 create mode 100644 e2e_test/source_inline/kafka/alter/add_column.slt
 rename e2e_test/{source/basic => source_inline/kafka}/alter/rate_limit_source_kafka.slt (87%)
 rename e2e_test/{source/basic => source_inline/kafka}/alter/rate_limit_table_kafka.slt (85%)
 delete mode 100644 scripts/source/alter_data/kafka_alter.2
 delete mode 100644 scripts/source/alter_data/kafka_alter.3
 delete mode 100644 scripts/source/prepare_data_after_alter.sh
 delete mode 100644 scripts/source/test_data/kafka_alter.1

diff --git a/ci/scripts/e2e-source-test.sh b/ci/scripts/e2e-source-test.sh
index 38e59a7da6c85..f10ee3ebef091 100755
--- a/ci/scripts/e2e-source-test.sh
+++ b/ci/scripts/e2e-source-test.sh
@@ -155,20 +155,6 @@ risedev ci-start ci-kafka
 ./scripts/source/prepare_ci_kafka.sh
 risedev slt './e2e_test/source/basic/*.slt'
 risedev slt './e2e_test/source/basic/old_row_format_syntax/*.slt'
-risedev slt './e2e_test/source/basic/alter/kafka.slt'
-
-echo "--- e2e, kafka alter source rate limit"
-risedev slt './e2e_test/source/basic/alter/rate_limit_source_kafka.slt'
-risedev slt './e2e_test/source/basic/alter/rate_limit_table_kafka.slt'
-
-echo "--- e2e, kafka alter source"
-chmod +x ./scripts/source/prepare_data_after_alter.sh
-./scripts/source/prepare_data_after_alter.sh 2
-risedev slt './e2e_test/source/basic/alter/kafka_after_new_data.slt'
-
-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 "--- Run CH-benCHmark"
 risedev slt './e2e_test/ch_benchmark/batch/ch_benchmark.slt'
diff --git a/e2e_test/source/basic/alter/kafka.slt b/e2e_test/source/basic/alter/kafka.slt
deleted file mode 100644
index 45b0cdd860f3e..0000000000000
--- a/e2e_test/source/basic/alter/kafka.slt
+++ /dev/null
@@ -1,139 +0,0 @@
-statement ok
-CREATE SOURCE s1 (v1 int) with (
-  connector = 'kafka',
-  topic = 'kafka_alter',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest'
-) FORMAT PLAIN ENCODE JSON;
-
-statement ok
-CREATE SOURCE s2 (v2 varchar) with (
-  connector = 'kafka',
-  topic = 'kafka_alter',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest'
-) FORMAT PLAIN ENCODE JSON;
-
-statement ok
-CREATE TABLE t (v1 int) with (
-  connector = 'kafka',
-  topic = 'kafka_alter',
-  properties.bootstrap.server = 'message_queue:29092',
-  scan.startup.mode = 'earliest'
-) FORMAT PLAIN ENCODE JSON;
-
-
-statement ok
-create materialized view mv1 as select * from s1;
-
-statement ok
-create materialized view mv2 as select * from s2;
-
-sleep 5s
-
-statement ok
-flush;
-
-query I
-select * from s1;
-----
-1
-
-query T
-select * from s2;
-----
-11
-
-query I
-select * from t;
-----
-1
-
-# alter source
-statement ok
-alter source s1 add column v2 varchar;
-
-# alter source with null column
-statement ok
-alter source s2 add column v4 int;
-
-# alter table with conn
-statement error try to use ALTER TABLE instead
-alter source t add column v0 int;
-
-statement ok
-create materialized view mv3 as select * from s1;
-
-statement ok
-create materialized view mv4 as select * from s2;
-
-statement ok
-alter table t add column v2 varchar;
-
-sleep 5s
-
-statement ok
-flush;
-
-query IT
-select * from s1
-----
-1 11
-
-query TI
-select * from s2
-----
-11 NULL
-
-query I
-select * from mv1
-----
-1
-
-query T
-select * from mv2
-----
-11
-
-query IT
-select * from mv3
-----
-1 11
-
-query TI
-select * from mv4
-----
-11 NULL
-
-query IT
-select * from t
-----
-1 NULL
-
-# alter source again
-statement ok
-alter source s1 add column v3 int;
-
-statement ok
-create materialized view mv5 as select * from s1;
-
-sleep 5s
-
-statement ok
-flush;
-
-query ITI
-select * from s1
-----
-1 11 111
-
-query ITI
-select * from mv5
-----
-1 11 111
-
-# check definition after altering
-query TT
-show create source s1;
-----
-public.s1 CREATE SOURCE s1 (v1 INT, v2 CHARACTER VARYING, v3 INT) WITH (connector = 'kafka', topic = 'kafka_alter', properties.bootstrap.server = 'message_queue:29092', scan.startup.mode = 'earliest') FORMAT PLAIN ENCODE JSON
\ No newline at end of file
diff --git a/e2e_test/source/basic/alter/kafka_after_new_data.slt b/e2e_test/source/basic/alter/kafka_after_new_data.slt
deleted file mode 100644
index 5a73b749079f9..0000000000000
--- a/e2e_test/source/basic/alter/kafka_after_new_data.slt
+++ /dev/null
@@ -1,82 +0,0 @@
-sleep 5s
-
-statement ok
-flush;
-
-query IT rowsort
-select * from s1
-----
-1 11 111
-2 22 222
-
-query I rowsort
-select * from mv1
-----
-1
-2
-
-query IT rowsort
-select * from mv3
-----
-1 11
-2 22
-
-query TI rowsort
-select * from s2
-----
-11 NULL
-22 NULL
-
-query T rowsort
-select * from mv2
-----
-11
-22
-
-query TI rowsort
-select * from mv4
-----
-11 NULL
-22 NULL
-
-query ITI rowsort
-select * from mv5
-----
-1 11 111
-2 22 222
-
-query IT rowsort
-select * from t
-----
-1 NULL
-2 22
-
-statement ok
-alter table t add column v3 int;
-
-query IT rowsort
-select * from t
-----
-1 NULL NULL
-2 22 NULL
-
-statement ok
-drop materialized view mv1
-
-statement ok
-drop materialized view mv2
-
-statement ok
-drop materialized view mv3
-
-statement ok
-drop materialized view mv4
-
-statement ok
-drop materialized view mv5
-
-statement ok
-drop source s1
-
-statement ok
-drop source s2
\ No newline at end of file
diff --git a/e2e_test/source/basic/alter/kafka_after_new_data_2.slt b/e2e_test/source/basic/alter/kafka_after_new_data_2.slt
deleted file mode 100644
index c10634d259138..0000000000000
--- a/e2e_test/source/basic/alter/kafka_after_new_data_2.slt
+++ /dev/null
@@ -1,14 +0,0 @@
-sleep 5s
-
-statement ok
-flush;
-
-query IT rowsort
-select * from t
-----
-1 NULL NULL
-2 22 NULL
-3 33 333
-
-statement ok
-drop table t;
\ No newline at end of file
diff --git a/e2e_test/source_inline/kafka/alter/add_column.slt b/e2e_test/source_inline/kafka/alter/add_column.slt
new file mode 100644
index 0000000000000..cde818f6baf90
--- /dev/null
+++ b/e2e_test/source_inline/kafka/alter/add_column.slt
@@ -0,0 +1,271 @@
+control substitution on
+
+system ok
+rpk topic delete kafka_alter || true
+
+system ok
+rpk topic create kafka_alter -p 1
+
+system ok
+cat <<EOF | rpk topic produce kafka_alter -f "%v\n"
+{"v1": 1, "v2": "11", "v3": 111}
+EOF
+
+statement ok
+CREATE SOURCE s1 (v1 int) with (
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'kafka_alter',
+  scan.startup.mode = 'earliest'
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+CREATE SOURCE s2 (v2 varchar) with (
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'kafka_alter',
+  scan.startup.mode = 'earliest'
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+CREATE TABLE t (v1 int) with (
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'kafka_alter',
+  scan.startup.mode = 'earliest'
+) FORMAT PLAIN ENCODE JSON;
+
+
+statement ok
+create materialized view mv1 as select * from s1;
+
+statement ok
+create materialized view mv2 as select * from s2;
+
+sleep 2s
+
+statement ok
+flush;
+
+query I
+select * from s1;
+----
+1
+
+query T
+select * from s2;
+----
+11
+
+query I
+select * from t;
+----
+1
+
+# alter source
+statement ok
+alter source s1 add column v2 varchar;
+
+# alter source with null column
+statement ok
+alter source s2 add column v4 int;
+
+# alter table with conn
+statement error try to use ALTER TABLE instead
+alter source t add column v0 int;
+
+statement ok
+create materialized view mv3 as select * from s1;
+
+statement ok
+create materialized view mv4 as select * from s2;
+
+statement ok
+alter table t add column v2 varchar;
+
+sleep 2s
+
+statement ok
+flush;
+
+query IT
+select * from s1
+----
+1 11
+
+query TI
+select * from s2
+----
+11 NULL
+
+query I
+select * from mv1
+----
+1
+
+query T
+select * from mv2
+----
+11
+
+query IT
+select * from mv3
+----
+1 11
+
+query TI
+select * from mv4
+----
+11 NULL
+
+query IT
+select * from t
+----
+1 NULL
+
+# alter source again
+statement ok
+alter source s1 add column v3 int;
+
+statement ok
+create materialized view mv5 as select * from s1;
+
+sleep 2s
+
+statement ok
+flush;
+
+query ITI
+select * from s1
+----
+1 11 111
+
+query ITI
+select * from mv5
+----
+1 11 111
+
+# check definition after altering
+query TTTT
+describe s1;
+----
+v1	integer	false	NULL
+_rw_kafka_timestamp	timestamp with time zone	true	NULL
+_row_id	serial	true	NULL
+v2	character varying	false	NULL
+v3	integer	false	NULL
+primary key	_row_id	NULL	NULL
+table description	s1	NULL	NULL
+
+query TT
+show create source s1;
+----
+public.s1 CREATE SOURCE s1 (v1 INT, v2 CHARACTER VARYING, v3 INT) WITH (connector = 'kafka', properties.bootstrap.server = 'message_queue:29092', topic = 'kafka_alter', scan.startup.mode = 'earliest') FORMAT PLAIN ENCODE JSON
+
+
+system ok
+cat <<EOF | rpk topic produce kafka_alter -f "%v\n"
+{"v1": 2, "v2": "22", "v3": 222}
+EOF
+
+
+sleep 2s
+
+statement ok
+flush;
+
+query IT rowsort
+select * from s1
+----
+1 11 111
+2 22 222
+
+query I rowsort
+select * from mv1
+----
+1
+2
+
+query IT rowsort
+select * from mv3
+----
+1 11
+2 22
+
+query TI rowsort
+select * from s2
+----
+11 NULL
+22 NULL
+
+query T rowsort
+select * from mv2
+----
+11
+22
+
+query TI rowsort
+select * from mv4
+----
+11 NULL
+22 NULL
+
+query ITI rowsort
+select * from mv5
+----
+1 11 111
+2 22 222
+
+query IT rowsort
+select * from t
+----
+1 NULL
+2 22
+
+statement ok
+alter table t add column v3 int;
+
+query IT rowsort
+select * from t
+----
+1 NULL NULL
+2 22 NULL
+
+statement ok
+drop materialized view mv1
+
+statement ok
+drop materialized view mv2
+
+statement ok
+drop materialized view mv3
+
+statement ok
+drop materialized view mv4
+
+statement ok
+drop materialized view mv5
+
+statement ok
+drop source s1
+
+statement ok
+drop source s2
+
+
+system ok
+cat <<EOF | rpk topic produce kafka_alter -f "%v\n"
+{"v1": 3, "v2": "33", "v3": 333}
+EOF
+
+sleep 2s
+
+statement ok
+flush;
+
+query IT rowsort
+select * from t
+----
+1 NULL NULL
+2 22 NULL
+3 33 333
+
+statement ok
+drop table t;
diff --git a/e2e_test/source/basic/alter/rate_limit_source_kafka.slt b/e2e_test/source_inline/kafka/alter/rate_limit_source_kafka.slt
similarity index 87%
rename from e2e_test/source/basic/alter/rate_limit_source_kafka.slt
rename to e2e_test/source_inline/kafka/alter/rate_limit_source_kafka.slt
index ac56a7f48e3a5..9bfa7238e3476 100644
--- a/e2e_test/source/basic/alter/rate_limit_source_kafka.slt
+++ b/e2e_test/source_inline/kafka/alter/rate_limit_source_kafka.slt
@@ -1,3 +1,5 @@
+control substitution on
+
 ############## Create kafka seed data
 
 statement ok
@@ -12,11 +14,10 @@ statement ok
 create sink kafka_sink
 from
   kafka_seed_data with (
-    properties.bootstrap.server = 'message_queue:29092',
-    topic = 'kafka_source',
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = 'test_rate_limit',
     type = 'append-only',
-    force_append_only='true',
-    connector = 'kafka'
+    force_append_only='true'
 );
 
 ############## Source from kafka (rate_limit = 0)
@@ -27,9 +28,8 @@ sleep 5s
 
 statement ok
 create source kafka_source (v1 int) with (
-  connector = 'kafka',
-  topic = 'kafka_source',
-  properties.bootstrap.server = 'message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'test_rate_limit',
   scan.startup.mode = 'earliest',
   source_rate_limit = 0,
 ) FORMAT PLAIN ENCODE JSON
@@ -126,4 +126,4 @@ statement ok
 drop sink kafka_sink;
 
 statement ok
-drop table kafka_seed_data;
\ No newline at end of file
+drop table kafka_seed_data;
diff --git a/e2e_test/source/basic/alter/rate_limit_table_kafka.slt b/e2e_test/source_inline/kafka/alter/rate_limit_table_kafka.slt
similarity index 85%
rename from e2e_test/source/basic/alter/rate_limit_table_kafka.slt
rename to e2e_test/source_inline/kafka/alter/rate_limit_table_kafka.slt
index 1a23bbf1009de..ac2a665fd10c0 100644
--- a/e2e_test/source/basic/alter/rate_limit_table_kafka.slt
+++ b/e2e_test/source_inline/kafka/alter/rate_limit_table_kafka.slt
@@ -1,3 +1,5 @@
+control substitution on
+
 ############## Create kafka seed data
 
 statement ok
@@ -12,11 +14,10 @@ statement ok
 create sink kafka_sink
 from
   kafka_seed_data with (
-    properties.bootstrap.server = 'message_queue:29092',
-    topic = 'rate_limit_source_kafka_0',
+    ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+    topic = 'test_rate_limit_for_table',
     type = 'append-only',
     force_append_only='true',
-    connector = 'kafka'
 );
 
 # topic may not be created yet
@@ -26,9 +27,8 @@ sleep 4s
 
 statement ok
 create table kafka_source (v1 int) with (
-  connector = 'kafka',
-  topic = 'rate_limit_source_kafka_0',
-  properties.bootstrap.server = 'message_queue:29092',
+  ${RISEDEV_KAFKA_WITH_OPTIONS_COMMON},
+  topic = 'test_rate_limit_for_table',
   scan.startup.mode = 'earliest',
   source_rate_limit = 0
 ) FORMAT PLAIN ENCODE JSON
@@ -101,4 +101,4 @@ statement ok
 drop sink kafka_sink;
 
 statement ok
-drop table kafka_seed_data;
\ No newline at end of file
+drop table kafka_seed_data;
diff --git a/scripts/source/alter_data/kafka_alter.2 b/scripts/source/alter_data/kafka_alter.2
deleted file mode 100644
index 39676d079c758..0000000000000
--- a/scripts/source/alter_data/kafka_alter.2
+++ /dev/null
@@ -1 +0,0 @@
-{"v1": 2, "v2": "22", "v3": 222}
\ No newline at end of file
diff --git a/scripts/source/alter_data/kafka_alter.3 b/scripts/source/alter_data/kafka_alter.3
deleted file mode 100644
index fb9015ae75caf..0000000000000
--- a/scripts/source/alter_data/kafka_alter.3
+++ /dev/null
@@ -1 +0,0 @@
-{"v1": 3, "v2": "33", "v3": 333}
\ No newline at end of file
diff --git a/scripts/source/prepare_data_after_alter.sh b/scripts/source/prepare_data_after_alter.sh
deleted file mode 100644
index a82bd59328653..0000000000000
--- a/scripts/source/prepare_data_after_alter.sh
+++ /dev/null
@@ -1,11 +0,0 @@
-#!/usr/bin/env bash
-
-# Exits as soon as any line fails.
-set -e
-
-SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
-cd "$SCRIPT_PATH/.." || exit 1
-
-FILE="./source/alter_data/kafka_alter.$1"
-echo "Send data from $FILE"
-cat $FILE | kcat -P -b message_queue:29092 -t kafka_alter
\ No newline at end of file
diff --git a/scripts/source/test_data/kafka_alter.1 b/scripts/source/test_data/kafka_alter.1
deleted file mode 100644
index 3cf65f5d494cf..0000000000000
--- a/scripts/source/test_data/kafka_alter.1
+++ /dev/null
@@ -1 +0,0 @@
-{"v1": 1, "v2": "11", "v3": 111}
\ No newline at end of file