move

.gitattributes

@@ -1,6 +1,6 @@
 # regression test data
 src/tests/regress/data/** linguist-vendored
 # source test data
-scripts/source/test_data/** linguist-vendored
+e2e_test/source_legacy/basic/scripts/test_data/** linguist-vendored
 # generated grafana dashboard
 grafana/risingwave-dashboard.json linguist-generated

ci/scripts/deterministic-e2e-test.sh

@@ -11,7 +11,7 @@ download-and-decompress-artifact risingwave_simulation .
 chmod +x ./risingwave_simulation
 
 echo "--- Extract data for Kafka"
-pushd ./scripts/source/
+pushd ./e2e_test/source_legacy/basic/scripts/
 mkdir -p ./test_data
 unzip -o test_data.zip -d .
 popd
@@ -39,7 +39,7 @@ echo "--- deterministic simulation e2e, ci-3cn-2fe, batch"
 seq "$TEST_NUM" | parallel './risingwave_simulation ./e2e_test/batch/\*\*/\*.slt 2> $LOGDIR/batch-{}.log && rm $LOGDIR/batch-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe, kafka source"
-seq "$TEST_NUM" | parallel './risingwave_simulation --kafka-datadir=./scripts/source/test_data ./e2e_test/source_legacy/basic/kafka\*.slt 2> $LOGDIR/source-{}.log && rm $LOGDIR/source-{}.log'
+seq "$TEST_NUM" | parallel './risingwave_simulation --kafka-datadir=./e2e_test/source_legacy/basic/scripts/test_data ./e2e_test/source_legacy/basic/kafka\*.slt 2> $LOGDIR/source-{}.log && rm $LOGDIR/source-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe, parallel, streaming"
 seq "$TEST_NUM" | parallel './risingwave_simulation -j 16 ./e2e_test/streaming/\*\*/\*.slt 2> $LOGDIR/parallel-streaming-{}.log && rm $LOGDIR/parallel-streaming-{}.log'

ci/scripts/deterministic-recovery-test.sh

@@ -93,6 +93,6 @@ echo "--- deterministic simulation e2e, ci-3cn-2fe-1meta, recovery, kafka source
 seq "$TEST_NUM" | parallel './risingwave_simulation \
 --kill \
 --kill-rate=${KILL_RATE} \
---kafka-datadir=./scripts/source/test_data \
+--kafka-datadir=./e2e_test/source_legacy/basic/scripts/test_data \
 ${EXTRA_ARGS:-} \
 ./e2e_test/source_legacy/basic/kafka\*.slt 2> $LOGDIR/recovery-source-{}.log && rm $LOGDIR/recovery-source-{}.log'

ci/scripts/e2e-source-test.sh

@@ -148,11 +148,11 @@ echo "--- Kill cluster"
 risedev ci-kill
 export RISINGWAVE_CI=true
 
-echo "--- e2e, ci-kafka-plus-pubsub, kafka and pubsub source"
+echo "--- e2e, ci-kafka-plus-pubsub, legacy kafka tests"
 export RUST_MIN_STACK=4194304
 RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
 risedev ci-start ci-kafka
-./scripts/source/prepare_ci_kafka.sh
+./e2e_test/source_legacy/basic/scripts/prepare_ci_kafka.sh
 risedev slt './e2e_test/source_legacy/basic/*.slt'
 risedev slt './e2e_test/source_legacy/basic/old_row_format_syntax/*.slt'
 

e2e_test/source_inline/kafka/avro/name_strategy.slt

@@ -25,7 +25,7 @@ create source s1 () with (
 # Currently we are abusing this test case to also test data types.
 
 system ok
-python3 scripts/source/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "topic" "avro"
+python3 e2e_test/source_legacy/basic/scripts/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "topic" "avro"
 
 statement ok
 CREATE TABLE t_topic ( primary key (rw_key) )
@@ -44,7 +44,7 @@ FORMAT UPSERT ENCODE AVRO (schema.registry = '${RISEDEV_SCHEMA_REGISTRY_URL}');
 
 ## topic: upsert_avro_json-record, key subject: string, value subject: CPLM.OBJ_ATTRIBUTE_VALUE
 system ok
-python3 scripts/source/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "record" "avro"
+python3 e2e_test/source_legacy/basic/scripts/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "record" "avro"
 
 
 statement error key\.message
@@ -80,7 +80,7 @@ create table t_record_format_plain () with (
 ## key subject: upsert_avro_json-topic-record-string
 ## value subject: upsert_avro_json-topic-record-CPLM.OBJ_ATTRIBUTE_VALUE
 system ok
-python3 scripts/source/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "topic-record" "avro"
+python3 e2e_test/source_legacy/basic/scripts/schema_registry_producer.py  "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" e2e_test/source_inline/kafka/avro/upsert_avro_json "topic-record" "avro"
 
 
 

e2e_test/source_legacy/README.md

@@ -7,8 +7,8 @@
 
 Test in this directory needs some prior setup.
 
-See also `ci/scripts/e2e-source-test.sh`, and `scripts/source`
+See also `ci/scripts/e2e-source-test.sh`, and `e2e_test/source_legacy/basic/scripts`
 
 ## Kafka
 
-`scripts/source/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
+`e2e_test/source_legacy/basic/scripts/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.

e2e_test/source_legacy/basic/scripts/README.md

@@ -0,0 +1,12 @@
+This folder contains scripts to prepare data for testing sources.
+
+## Kafka
+
+`e2e_test/source_legacy/basic/scripts/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
+
+- If `<topic_name>` ends with `bin`, the whole file is a message with binary data.
+- If `<topic_name>` ends with `avro_json` or `json_schema`:
+    - The first line is the schema. Key and value are separated by `^`.
+    - The rest of the lines are messages in JSON format. Key and value are separated by `^`.
+    - Produced to Kafka with `schema_registry_producer.py` (serialized to Avro or JSON)
+- Otherwise, each line is a message, and key/value is separated by `^`.

scripts/source/prepare_ci_kafka.sh → e2e_test/source_legacy/basic/scripts/prepare_ci_kafka.sh

@@ -2,18 +2,19 @@
 
 # Exits as soon as any line fails.
 set -e
+export CARGO_MAKE_PRINT_TIME_SUMMARY=false
 
 SCRIPT_PATH="$(cd "$(dirname "$0")" >/dev/null 2>&1 && pwd)"
-cd "$SCRIPT_PATH/.." || exit 1
-# cwd is /scripts
+# SCRIPT_PATH is e2e_test/source_legacy/kafka/script/
+# cwd is e2e_test/source_legacy/kafka/
 
 echo "$SCRIPT_PATH"
+cd "$SCRIPT_PATH"
 
-source ../.risingwave/config/risedev-env
+source ../../../../.risingwave/config/risedev-env
 
 if [ "$1" == "compress" ]; then
   echo "Compress test_data/ into test_data.zip"
-  cd ./source
   zip_file=test_data.zip
   if [ -f "$zip_file" ]; then
     rm "$zip_file"
@@ -23,10 +24,8 @@ if [ "$1" == "compress" ]; then
 fi
 
 echo "--- Extract data for Kafka"
-cd ./source/
 mkdir -p ./test_data/ch_benchmark/
 unzip -o test_data.zip -d .
-cd ..
 
 echo "path:${SCRIPT_PATH}/test_data/**/*"
 
@@ -58,9 +57,9 @@ for filename in $kafka_data_files; do
     if [[ "$topic" = *bin ]]; then
         kcat -P -b "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" -t "$topic" "$filename"
     elif [[ "$topic" = *avro_json ]]; then
-        python3 source/schema_registry_producer.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "$filename" "topic" "avro"
+        python3 schema_registry_producer.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "$filename" "topic" "avro"
     elif [[ "$topic" = *json_schema ]]; then
-        python3 source/schema_registry_producer.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "$filename" "topic" "json"
+        python3 schema_registry_producer.py "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" "${RISEDEV_SCHEMA_REGISTRY_URL}" "$filename" "topic" "json"
     else
         cat "$filename" | kcat -P -K ^  -b "${RISEDEV_KAFKA_BOOTSTRAP_SERVERS}" -t "$topic"
     fi

scripts/source/README.md

@@ -2,7 +2,7 @@ This folder contains scripts to prepare data for testing sources.
 
 ## Kafka
 
-`scripts/source/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
+`e2e_test/source_legacy/basic/scripts/test_data` contains the data. Filename's convention is `<topic_name>.<n_partitions>`.
 
 - If `<topic_name>` ends with `bin`, the whole file is a message with binary data.
 - If `<topic_name>` ends with `avro_json` or `json_schema`:

src/connector/codec/tests/integration_tests/avro.rs

@@ -119,7 +119,7 @@ fn check(
     expected_risingwave_data.assert_eq(&format!("{}", data_str.iter().format("\n----\n")));
 }
 
-// This corresponds to legacy `scripts/source/test_data/avro_simple_schema_bin.1`. TODO: remove that file.
+// This corresponds to legacy `e2e_test/source_legacy/basic/scripts/test_data/avro_simple_schema_bin.1`. TODO: remove that file.
 #[test]
 fn test_simple() {
     check(