add new source tests

risingwavelabs · Apr 5, 2024 · f280603 · f280603
1 parent 0e3ace1
commit f280603
Show file tree

Hide file tree

Showing 8 changed files with 157 additions and 56 deletions.
diff --git a/Makefile.toml b/Makefile.toml
@@ -15,6 +15,7 @@ extend = [
   { path = "src/storage/backup/integration_tests/Makefile.toml" },
   { path = "src/java_binding/make-java-binding.toml" },
   { path = "src/stream/tests/integration_tests/integration_test.toml" },
+  { path = "e2e_test/source_new/commands.toml"}
 ]
 
 env_files = ["./risedev-components.user.env"]
@@ -558,8 +559,8 @@ env_files = ["${PREFIX_CONFIG}/risedev-env"]
 script = '''
 #!/usr/bin/env bash
 cat <<EOF > "${PREFIX_CONFIG}/psql-env"
-export PGHOST=$RW_FRONTEND_LISTEN_ADDRESS
-export PGPORT=$RW_FRONTEND_PORT
+export PGHOST=$RISEDEV_FRONTEND_LISTEN_ADDRESS
+export PGPORT=$RISEDEV_FRONTEND_PORT
 export PGUSER=root
 export PGDATABASE=dev
 EOF
@@ -576,7 +577,7 @@ dependencies = ["check-risedev-env-file"]
 env_files = ["${PREFIX_CONFIG}/risedev-env"]
 script = '''
 #!/usr/bin/env bash
-psql -h $RW_FRONTEND_LISTEN_ADDRESS -p $RW_FRONTEND_PORT -U root -d dev "$@"
+psql -h $RISEDEV_FRONTEND_LISTEN_ADDRESS -p $RISEDEV_FRONTEND_PORT -U root -d dev "$@"
 '''
 
 [tasks.ctl]
@@ -883,7 +884,7 @@ cargo build \
 """
 
 [tasks.stest]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Run unit tests in deterministic simulation mode"
 dependencies = ["install-nextest"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -908,7 +909,7 @@ cargo nextest run \
 """
 
 [tasks.sit-test]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Run integration tests in deterministic simulation mode"
 dependencies = ["install-nextest"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -923,7 +924,7 @@ cargo nextest run \
 """
 
 [tasks.sarchive-it-test]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Archive integration tests in deterministic simulation mode"
 dependencies = ["install-nextest"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -939,7 +940,7 @@ cargo nextest archive \
 """
 
 [tasks.scheck]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Run cargo check in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -964,7 +965,7 @@ cargo check \
 """
 
 [tasks.sslt]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Run e2e tests in deterministic simulation mode"
 dependencies = ["warn-on-missing-tools"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -978,7 +979,7 @@ cargo run \
 """
 
 [tasks.sslt-build-all]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Build deterministic simulation runner and tests"
 dependencies = ["warn-on-missing-tools"]
 env = { CARGO_TARGET_DIR = "target/sim" }
@@ -993,7 +994,7 @@ cargo build \
 """
 
 [tasks.sslt-cov]
-category = "RiseDev - Deterministic Simulation"
+category = "RiseDev - Test - Deterministic Simulation"
 description = "Run e2e tests in deterministic simulation mode and report code coverage"
 dependencies = ["install-llvm-cov"]
 env = { CARGO_TARGET_DIR = "target/sim-cov" }
@@ -1290,42 +1291,42 @@ echo "All processes has exited."
 """
 
 [tasks.slt]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 install_crate = { version = "0.19.1", crate_name = "sqllogictest-bin", binary = "sqllogictest", test_arg = [
   "--help",
 ], install_command = "binstall" }
 dependencies = ["check-risedev-env-file"]
 env_files = ["${PREFIX_CONFIG}/risedev-env"]
-env = { SLT_HOST = "${RW_FRONTEND_LISTEN_ADDRESS}", SLT_PORT = "${RW_FRONTEND_PORT}", SLT_DB = "dev" }
+env = { SLT_HOST = "${RISEDEV_FRONTEND_LISTEN_ADDRESS}", SLT_PORT = "${RISEDEV_FRONTEND_PORT}", SLT_DB = "dev" }
 command = "sqllogictest"
 args = ["${@}"]
 description = "🌟 Run SQLLogicTest"
 
 [tasks.slt-streaming]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 extend = "slt"
 args = ["${@}", "./e2e_test/streaming/**/*.slt"]
 description = "Run all streaming e2e tests"
 
 [tasks.slt-batch]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 extend = "slt"
 args = ["${@}", "./e2e_test/batch/*.slt"]
 description = "Run all batch e2e tests"
 
 [tasks.slt-generated]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 extend = "slt"
 args = ["${@}", "./e2e_test/generated/**/*.slt"]
 description = "Run all generated e2e tests"
 
 [tasks.slt-all]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 run_task = { name = ["slt-streaming", "slt-batch", "slt-generated"] }
 description = "Run all e2e tests"
 
 [tasks.docslt]
-category = "RiseDev - SQLLogicTest"
+category = "RiseDev - Test - SQLLogicTest"
 description = "Extract SQL examples written in SQLLogicTest syntax from Rust doc comments"
 script = '''
 #!/usr/bin/env bash
@@ -1400,7 +1401,7 @@ UPDATE_EXPECT=1 cargo test -p risingwave_connector tests::test_with_options_yaml
 '''
 
 [tasks.backwards-compat-test]
-category = "RiseDev - Backwards Compatibility Test"
+category = "RiseDev - Test - Backwards Compatibility Test"
 description = "Run backwards compatibility test"
 script = "./backwards-compat-tests/scripts/run_local.sh"
 

diff --git a/e2e_test/source/README.md b/e2e_test/source/README.md
@@ -1,3 +1,7 @@
+> [!NOTE]
+>
+> Please write new tests according to the style in `e2e_test/source_new`.
+
 Test in this directory needs some prior setup.
 
 See also `ci/scripts/e2e-source-test.sh`, and `scripts/source`

diff --git a/e2e_test/source_new/README.md b/e2e_test/source_new/README.md
@@ -0,0 +1,12 @@
+Compared with prior source tests (`e2e_test/source`), tests in this directory are expected to be easy to run locally and easy to write.
+
+To run locally, use `risdev d` to start services (including external systems like Kafka and Postgres).
+Then use `risedev slt` to run the tests, which will load the environment variables (ports, etc.)
+according to the services started by `risdev d`.
+
+```sh
+risedev slt 'e2e_test/source-new/**/*.slt'
+```
+
+To write tests, please ensure each file is self-contained and does not depend on external scripts to setup the environment.
+Use `system` command to setup instead.
diff --git a/e2e_test/source_new/commands.toml b/e2e_test/source_new/commands.toml
@@ -0,0 +1,65 @@
+# This file contains commands used by the tests.
+
+[tasks.source-test-hook]
+private = true
+dependencies = ["check-risedev-env-file"]
+env_files = ["${PREFIX_CONFIG}/risedev-env"]
+
+[tasks.check-kafka-env]
+dependencies = ["check-risedev-env-file"]
+env_files = ["${PREFIX_CONFIG}/risedev-env"]
+script = '''
+#!/usr/bin/env bash
+set -e
+
+if [ ! -d "${PREFIX_BIN}/kafka" ]; then
+    echo "Kafka is not installed in ${PREFIX_BIN}/kafka. Did you enable Kafka using $(tput setaf 4)\`./risedev configure\`$(tput sgr0)?"
+    exit 1
+fi
+
+# TODO: we may support risedev-env.override so that we can connect to a Kafka not started by risedev-dev.
+if [ -z "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" ]; then
+    echo "RISEDEV_KAFKA_BOOTSTRAP_SERVERS is not set in risedev-env file. Did you start Kafka using $(tput setaf 4)\`./risedev d\`$(tput sgr0)?"
+    exit 1
+fi
+'''
+
+[tasks.kafka-topics]
+category = "RiseDev - Test - Source Test - Kafka"
+description = "This tool helps to create, delete, describe, or change a topic. "
+dependencies = ["check-kafka-env"]
+script = """
+#!/usr/bin/env bash
+set -e
+${PREFIX_BIN}/kafka/bin/kafka-topics.sh --bootstrap-server ${RISEDEV_KAFKA_BOOTSTRAP_SERVERS} "$@"
+"""
+
+[tasks.kafka-produce]
+category = "RiseDev - Test - Source Test - Kafka"
+description = "This tool helps to read data from standard input and publish it to Kafka."
+dependencies = ["check-kafka-env"]
+script = """
+#!/usr/bin/env bash
+set -e
+${PREFIX_BIN}/kafka/bin/kafka-console-producer.sh --bootstrap-server ${RISEDEV_KAFKA_BOOTSTRAP_SERVERS} "$@"
+"""
+
+[tasks.kafka-consume]
+category = "RiseDev - Test - Source Test - Kafka"
+description = "This tool helps to read data from Kafka topics and outputs it to standard output."
+dependencies = ["check-kafka-env"]
+script = """
+#!/usr/bin/env bash
+set -e
+${PREFIX_BIN}/kafka/bin/kafka-console-consumer.sh --bootstrap-server ${RISEDEV_KAFKA_BOOTSTRAP_SERVERS} "$@"
+"""
+
+[tasks.kafka-consumer-groups]
+category = "RiseDev - Test - Source Test - Kafka"
+description = "This tool helps to read data from Kafka topics and outputs it to standard output."
+dependencies = ["check-kafka-env"]
+script = """
+#!/usr/bin/env bash
+set -e
+${PREFIX_BIN}/kafka/bin/kafka-consumer-groups.sh --bootstrap-server ${RISEDEV_KAFKA_BOOTSTRAP_SERVERS} "$@"
+"""
diff --git a/e2e_test/source_new/kafka/test_1.slt b/e2e_test/source_new/kafka/test_1.slt
@@ -0,0 +1,9 @@
+system ok
+./risedev kafka-topics --topic source_new_test_1 --create --partitions 3
+
+system ok
+echo "1" | ./risedev kafka-produce --topic source_new_test_1
+
+
+system ok
+./risedev kafka-topics --topic source_new_test_1 --delete
diff --git a/scripts/source/prepare_ci_kafka.sh b/scripts/source/prepare_ci_kafka.sh
@@ -13,8 +13,6 @@ fi
 SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
 cd "$SCRIPT_PATH/.." || exit 1
 
-KAFKA_BIN="$SCRIPT_PATH/../../.risingwave/bin/kafka/bin"
-
 echo "$SCRIPT_PATH"
 
 if [ "$1" == "compress" ]; then
@@ -36,6 +34,11 @@ cd ..
 
 echo "path:${SCRIPT_PATH}/test_data/**/*"
 
+# Cheat risedev that the kafka is started by risedev
+cat << EOF > "$RW_PREFIX"/config/risedev-env
+RISEDEV_KAFKA_BOOTSTRAP_SERVERS="message_queue:29092"
+EOF
+
 echo "Create topics"
 kafka_data_files=$(find "$SCRIPT_PATH"/test_data -type f)
 for filename in $kafka_data_files; do
@@ -46,10 +49,10 @@ for filename in $kafka_data_files; do
 
     # always ok
     echo "Drop topic $topic"
-    "$KAFKA_BIN"/kafka-topics.sh --bootstrap-server message_queue:29092 --topic "$topic" --delete || true
+    risedev kafka-topics --topic "$topic" --delete || true
 
     echo "Recreate topic $topic with partition $partition"
-    "$KAFKA_BIN"/kafka-topics.sh --bootstrap-server message_queue:29092 --topic "$topic" --create --partitions "$partition") &
+    risedev kafka-topics --topic "$topic" --create --partitions "$partition") &
 done
 wait
 

diff --git a/src/cmd_all/scripts/e2e-full-standalone-demo.sh b/src/cmd_all/scripts/e2e-full-standalone-demo.sh
@@ -67,8 +67,8 @@ sleep 15
 # FIXME: Integrate standalone into risedev, so we can reuse risedev-env functionality here.
 cat << EOF > "$RW_PREFIX"/config/risedev-env
 RW_META_ADDR="http://0.0.0.0:5690"
-RW_FRONTEND_LISTEN_ADDRESS="0.0.0.0"
-RW_FRONTEND_PORT="4566"
+RISEDEV_FRONTEND_LISTEN_ADDRESS="0.0.0.0"
+RISEDEV_FRONTEND_PORT="4566"
 EOF
 
 echo "--- Setting up table"

diff --git a/src/risedevtool/src/risedev_env.rs b/src/risedevtool/src/risedev_env.rs
@@ -25,50 +25,57 @@ use crate::{add_hummock_backend, HummockInMemoryStrategy, ServiceConfig};
 pub fn generate_risedev_env(services: &Vec<ServiceConfig>) -> String {
     let mut env = String::new();
     for item in services {
-        if let ServiceConfig::ComputeNode(c) = item {
-            // RW_HUMMOCK_URL
-            // If the cluster is launched without a shared storage, we will skip this.
-            {
-                let mut cmd = Command::new("compute-node");
-                if add_hummock_backend(
-                    "dummy",
-                    c.provide_opendal.as_ref().unwrap(),
-                    c.provide_minio.as_ref().unwrap(),
-                    c.provide_aws_s3.as_ref().unwrap(),
-                    HummockInMemoryStrategy::Disallowed,
-                    &mut cmd,
-                )
-                .is_ok()
+        match item {
+            ServiceConfig::ComputeNode(c) => {
+                // RW_HUMMOCK_URL
+                // If the cluster is launched without a shared storage, we will skip this.
+                {
+                    let mut cmd = Command::new("compute-node");
+                    if add_hummock_backend(
+                        "dummy",
+                        c.provide_opendal.as_ref().unwrap(),
+                        c.provide_minio.as_ref().unwrap(),
+                        c.provide_aws_s3.as_ref().unwrap(),
+                        HummockInMemoryStrategy::Disallowed,
+                        &mut cmd,
+                    )
+                    .is_ok()
+                    {
+                        writeln!(
+                            env,
+                            "RW_HUMMOCK_URL=\"{}\"",
+                            cmd.get_args().nth(1).unwrap().to_str().unwrap()
+                        )
+                        .unwrap();
+                    }
+                }
+
+                // RW_META_ADDR
                 {
+                    let meta_node = &c.provide_meta_node.as_ref().unwrap()[0];
                     writeln!(
                         env,
-                        "RW_HUMMOCK_URL=\"{}\"",
-                        cmd.get_args().nth(1).unwrap().to_str().unwrap()
+                        "RW_META_ADDR=\"http://{}:{}\"",
+                        meta_node.address, meta_node.port
                     )
                     .unwrap();
                 }
             }
-
-            // RW_META_ADDR
-            {
-                let meta_node = &c.provide_meta_node.as_ref().unwrap()[0];
+            ServiceConfig::Frontend(c) => {
+                let listen_address = &c.listen_address;
+                writeln!(env, "RISEDEV_FRONTEND_LISTEN_ADDRESS=\"{listen_address}\"",).unwrap();
+                let port = &c.port;
+                writeln!(env, "RISEDEV_FRONTEND_PORT=\"{port}\"",).unwrap();
+            }
+            ServiceConfig::Kafka(c) => {
                 writeln!(
                     env,
-                    "RW_META_ADDR=\"http://{}:{}\"",
-                    meta_node.address, meta_node.port
+                    "RISEDEV_KAFKA_BOOTSTRAP_SERVERS=\"{}:{}\"",
+                    c.address, c.port
                 )
                 .unwrap();
             }
-            break;
-        }
-    }
-    for item in services {
-        if let ServiceConfig::Frontend(c) = item {
-            let listen_address = &c.listen_address;
-            writeln!(env, "RW_FRONTEND_LISTEN_ADDRESS=\"{listen_address}\"",).unwrap();
-            let port = &c.port;
-            writeln!(env, "RW_FRONTEND_PORT=\"{port}\"",).unwrap();
-            break;
+            _ => {}
         }
     }
     env