Merge branch 'main' into xzhseh/sql-udf-semantic-check

ci/scripts/backfill-test.sh

@@ -32,4 +32,4 @@ download_and_prepare_rw "$profile" source
 
 ################ TESTS
 
-./ci/scripts/run-backfill-tests.sh
+profile=$profile ./ci/scripts/run-backfill-tests.sh

ci/scripts/integration-tests.sh

@@ -79,13 +79,6 @@ docker ps
 echo "--- check if the ingestion is successful"
 # extract the type of upstream source,e.g. mysql,postgres,etc
 upstream=$(echo ${case} | cut -d'-' -f 1)
-if [ "${upstream}" == "mysql" ]; then
-  echo "install mysql"
-  sudo rpm -Uvh https://dev.mysql.com/get/mysql80-community-release-el9-1.noarch.rpm
-  sudo dnf -y install mysql-community-server
-fi
-
-export PGPASSWORD=123456
 python3 check_data.py ${case} ${upstream}
 
 echo "--- clean Demos"

ci/scripts/run-backfill-tests.sh

@@ -4,7 +4,7 @@
 
 # USAGE:
 # ```sh
-# ./ci/scripts/run-backfill-tests.sh
+# profile=(ci-release|ci-dev) ./ci/scripts/run-backfill-tests.sh
 # ```
 # Example progress:
 # dev=> select * from rw_catalog.rw_ddl_progress;
@@ -23,6 +23,11 @@ BACKGROUND_DDL_DIR=$TEST_DIR/background_ddl
 COMMON_DIR=$BACKGROUND_DDL_DIR/common
 
 CLUSTER_PROFILE='ci-1cn-1fe-kafka-with-recovery'
+if [[ -n "${BUILDKITE:-}" ]]; then
+  RUNTIME_CLUSTER_PROFILE='ci-3cn-1fe-with-monitoring'
+else
+  RUNTIME_CLUSTER_PROFILE='ci-3cn-1fe'
+fi
 export RUST_LOG="info,risingwave_meta::barrier::progress=debug,risingwave_meta::rpc::ddl_controller=debug"
 
 run_sql_file() {
@@ -200,67 +205,49 @@ test_sink_backfill_recovery() {
   wait
 }
 
-test_no_shuffle_backfill_runtime() {
-  echo "--- e2e, test_no_shuffle_backfill_runtime"
-  cargo make ci-start $CLUSTER_PROFILE
+test_arrangement_backfill_snapshot_and_upstream_runtime() {
+  echo "--- e2e, test_backfill_snapshot_and_upstream_runtime"
+  cargo make ci-start $RUNTIME_CLUSTER_PROFILE
   sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_table.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
-
-  # Provide updates ...
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt' 2>&1 1>out.log &
-
-  # ... and concurrently create mv.
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_no_shuffle_mv.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert_snapshot.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert_upstream.slt' 2>&1 1>out.log &
+  echo "[INFO] Upstream is ingesting in background"
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_arrangement_backfill_mv.slt'
 
   wait
 
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows_arrangement.slt'
 
   cargo make kill
   cargo make wait-processes-exit
 }
 
-test_arrangement_backfill_runtime() {
-  echo "--- e2e, test_arrangement_backfill_runtime"
-  cargo make ci-start $CLUSTER_PROFILE
+test_no_shuffle_backfill_snapshot_and_upstream_runtime() {
+  echo "--- e2e, test_backfill_snapshot_and_upstream_runtime"
+  cargo make ci-start $RUNTIME_CLUSTER_PROFILE
   sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_table.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
-
-  # Provide updates ...
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt' 2>&1 1>out.log &
-
-  # ... and concurrently create mv.
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_arrangement_backfill_mv.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert_snapshot.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert_upstream.slt' 2>&1 1>out.log &
+  echo "[INFO] Upstream is ingesting in background"
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_no_shuffle_mv.slt'
 
   wait
 
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows_no_shuffle.slt'
 
   cargo make kill
   cargo make wait-processes-exit
 }
 
-test_no_shuffle_backfill_snapshot_only_runtime() {
-  echo "--- e2e, test_no_shuffle_backfill_snapshot_only_runtime"
-  cargo make ci-start $CLUSTER_PROFILE
+test_backfill_snapshot_runtime() {
+  echo "--- e2e, test_backfill_snapshot_runtime"
+  cargo make ci-start $RUNTIME_CLUSTER_PROFILE
   sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_table.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_no_shuffle_mv.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows.slt'
-
-  cargo make kill
-  cargo make wait-processes-exit
-}
-
-test_arrangement_backfill_snapshot_only_runtime() {
-  echo "--- e2e, test_arrangement_backfill_snapshot_only_runtime"
-  cargo make ci-start $CLUSTER_PROFILE
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_table.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/insert_snapshot.slt'
   sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_arrangement_backfill_mv.slt'
-  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/create_no_shuffle_mv.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows_no_shuffle.slt'
+  sqllogictest -p 4566 -d dev 'e2e_test/backfill/runtime/validate_rows_arrangement.slt'
 
   cargo make kill
   cargo make wait-processes-exit
@@ -272,10 +259,21 @@ main() {
   test_backfill_tombstone
   test_replication_with_column_pruning
   test_sink_backfill_recovery
-  test_no_shuffle_backfill_runtime
-  test_arrangement_backfill_runtime
-  test_no_shuffle_backfill_snapshot_only_runtime
-  test_arrangement_backfill_snapshot_only_runtime
+
+  # Only if profile is "ci-release", run it.
+  if [[ ${profile:-} == "ci-release" ]]; then
+    echo "--- Using release profile, running backfill performance tests."
+    # Need separate tests, we don't want to backfill concurrently.
+    # It's difficult to measure the time taken for each backfill if we do so.
+    test_no_shuffle_backfill_snapshot_and_upstream_runtime
+    test_arrangement_backfill_snapshot_and_upstream_runtime
+
+    # Backfill will happen in sequence here.
+    test_backfill_snapshot_runtime
+
+    # No upstream only tests, because if there's no snapshot,
+    # Backfill will complete almost immediately.
+  fi
 }
 
 main

ci/workflows/pull-request.yml

@@ -96,7 +96,7 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
-    timeout_in_minutes: 15
+    timeout_in_minutes: 16
     retry: *auto-retry
 
   - label: "end-to-end test (parallel)"

e2e_test/backfill/runtime/create_arrangement_backfill_mv.slt

@@ -2,4 +2,4 @@ statement ok
 SET STREAMING_ENABLE_ARRANGEMENT_BACKFILL=true;
 
 statement ok
-CREATE MATERIALIZED VIEW m1 AS SELECT * FROM t;
+CREATE MATERIALIZED VIEW arrangement_backfill AS SELECT * FROM t;

e2e_test/backfill/runtime/create_no_shuffle_mv.slt

@@ -2,4 +2,4 @@ statement ok
 SET STREAMING_ENABLE_ARRANGEMENT_BACKFILL=false;
 
 statement ok
-CREATE MATERIALIZED VIEW m1 AS SELECT * FROM t;
+CREATE MATERIALIZED VIEW no_shuffle_backfill AS SELECT * FROM t;

e2e_test/backfill/runtime/create_table.slt

@@ -1,2 +1,2 @@
 statement ok
-CREATE TABLE t (v1 int, v2 varchar, v3 bigint);
+CREATE TABLE t (v1 int primary key, v2 varchar, v3 bigint);

e2e_test/backfill/runtime/insert_snapshot.slt

@@ -0,0 +1,5 @@
+statement ok
+INSERT INTO t select generate_series, 'jakbj2khbe2', 22222222222 from generate_series(4000001, 8000000);
+
+statement ok
+flush;

e2e_test/backfill/runtime/insert_upstream.slt

@@ -0,0 +1,5 @@
+statement ok
+INSERT INTO t select generate_series, 'jakbj2khbe2', 22222222222 from generate_series(1, 4000000);
+
+statement ok
+flush;

e2e_test/backfill/runtime/validate_rows_arrangement.slt

@@ -0,0 +1,4 @@
+query I
+select (select count(*) from arrangement_backfill) = (select count(*) from t);
+----
+t

e2e_test/backfill/runtime/validate_rows_no_shuffle.slt

@@ -0,0 +1,4 @@
+query I
+select (select count(*) from no_shuffle_backfill) = (select count(*) from t);
+----
+t

e2e_test/iceberg/start_spark_connect_server.sh

@@ -16,6 +16,7 @@ tar -xzf $SPARK_FILE --no-same-owner
   --master local[3] \
   --conf spark.driver.bindAddress=0.0.0.0 \
   --conf spark.sql.catalog.demo=org.apache.iceberg.spark.SparkCatalog \
+  --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
   --conf spark.sql.catalog.demo.type=hadoop \
   --conf spark.sql.catalog.demo.warehouse=s3a://icebergdata/demo \
   --conf spark.sql.catalog.demo.hadoop.fs.s3a.endpoint=http://127.0.0.1:9301 \

e2e_test/iceberg/test_case/cdc/load.slt

@@ -24,8 +24,8 @@ CREATE SINK s1 AS select * from products WITH (
     connector = 'iceberg',
     type = 'upsert',
     force_append_only = 'false',
-    database.name = 'demo',
-    table.name = 'demo_db.demo_table',
+    database.name = 'demo_db',
+    table.name = 'demo_table',
     catalog.type = 'storage',
     warehouse.path = 's3://icebergdata/demo',
     s3.endpoint = 'http://127.0.0.1:9301',
@@ -38,6 +38,8 @@ CREATE SINK s1 AS select * from products WITH (
 statement ok
 flush;
 
+sleep 20s
+
 query I
 select count(*) from products;
 ----

e2e_test/iceberg/test_case/iceberg_sink_append_only.slt

@@ -23,8 +23,8 @@ CREATE SINK s6 AS select * from mv6 WITH (
     connector = 'iceberg',
     type = 'append-only',
     force_append_only = 'true',
-    database.name = 'demo',
-    table.name = 'demo_db.demo_table',
+    database.name = 'demo_db',
+    table.name = 'demo_table',
     catalog.type = 'storage',
     warehouse.path = 's3://icebergdata/demo',
     s3.endpoint = 'http://127.0.0.1:9301',

e2e_test/iceberg/test_case/iceberg_sink_upsert.slt

@@ -12,8 +12,8 @@ CREATE SINK s6 AS select mv6.id as id, mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3
     connector = 'iceberg',
     type = 'upsert',
     force_append_only = 'false',
-    database.name = 'demo',
-    table.name = 'demo_db.demo_table',
+    database.name = 'demo_db',
+    table.name = 'demo_table',
     catalog.type = 'storage',
     warehouse.path = 's3://icebergdata/demo',
     s3.endpoint = 'http://127.0.0.1:9301',

e2e_test/iceberg/test_case/no_partition_append_only.toml

@@ -13,7 +13,7 @@ init_sqls = [
     v_date date,
     v_timestamp timestamp,
     v_ts_ntz timestamp_ntz
-    ) TBLPROPERTIES ('format-version'='2');
+    ) USING iceberg TBLPROPERTIES ('format-version'='2');
     '''
 ]
 

e2e_test/sink/iceberg_sink.slt

@@ -15,8 +15,8 @@ CREATE SINK s6 AS select mv6.v1 as v1, mv6.v2 as v2, mv6.v3 as v3 from mv6 WITH
     s3.secret.key = 'hummockadmin',
     s3.region = 'us-east-1',
     catalog.type = 'storage',
-    database.name='demo',
-    table.name='demo_db.demo_table'
+    database.name='demo_db',
+    table.name='demo_table'
 );
 
 statement ok

e2e_test/source/basic/kafka.slt

@@ -293,8 +293,8 @@ with (
   properties.bootstrap.server = 'message_queue:29092'
 ) FORMAT PLAIN ENCODE AVRO (schema.registry = 'http://127.0.0.1:8081');
 
-# we cannot create debezium source with generated column
-statement error
+# create debezium source with generated column
+statement ok
 create table s20 (
   id integer primary key,
   first_name varchar,
@@ -847,6 +847,9 @@ drop source s17
 statement ok
 drop source s18
 
+statement ok
+drop table s20
+
 statement ok
 drop table s21
 

e2e_test/source/basic/old_row_format_syntax/kafka.slt

@@ -280,8 +280,8 @@ with (
 ) row format avro
 row schema location confluent schema registry 'http://127.0.0.1:8081'
 
-# we cannot create debezium source with generated column
-statement error
+# create debezium source with generated column
+statement ok
 create table s20 (
   id integer primary key,
   first_name varchar,
@@ -736,6 +736,9 @@ drop source s17
 statement ok
 drop source s18
 
+statement ok
+drop table s20
+
 statement ok
 drop table s21