Merge remote-tracking branch 'origin/main' into siyuan/cdc-handle-schema-event

Author: StrikeW

Cargo.lock

@@ -121,19 +121,20 @@ aws-smithy-types = { version = "1", default-features = false, features = [
 aws-endpoint = "0.60"
 aws-types = "1"
 axum = "=0.7.4" # TODO: 0.7.5+ does not work with current toolchain
-etcd-client = { package = "madsim-etcd-client", version = "0.4" }
+etcd-client = { package = "madsim-etcd-client", version = "0.6" }
 futures-async-stream = "0.2.9"
 hytra = "0.1"
 rdkafka = { package = "madsim-rdkafka", version = "0.4.1", features = [
     "cmake-build",
 ] }
 hashbrown = { version = "0.14", features = ["ahash", "inline-more", "nightly"] }
 criterion = { version = "0.5", features = ["async_futures"] }
-tonic = { package = "madsim-tonic", version = "0.4.1" }
-tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }
-otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev = "492c244e0be91feb659c0cd48a624bbd96045a33" }
-prost = { version = "0.12" }
-icelake = { git = "https://github.com/icelake-io/icelake", rev = "07d53893d7788b4e41fc11efad8a6be828405c31", features = [
+tonic = { package = "madsim-tonic", version = "0.5.1" }
+tonic-build = { package = "madsim-tonic-build", version = "0.5" }
+otlp-embedded = { git = "https://github.com/risingwavelabs/otlp-embedded", rev = "e6cd165b9bc85783b42c106e99186b86b73e3507" }
+prost = { version = "0.13" }
+prost-build = { version = "0.13" }
+icelake = { git = "https://github.com/risingwavelabs/icelake.git", rev = "1860eb315183a5f3f72b4097c1e40d49407f8373", features = [
     "prometheus",
 ] }
 arrow-array-iceberg = { package = "arrow-array", version = "52" }
@@ -179,6 +180,7 @@ tikv-jemallocator = { git = "https://github.com/risingwavelabs/jemallocator.git"
     "profiling",
     "stats",
 ], rev = "64a2d9" }
+# TODO(http-bump): bump to use tonic 0.12 once minitrace-opentelemetry is updated
 opentelemetry = "0.23"
 opentelemetry-otlp = "0.16"
 opentelemetry_sdk = { version = "0.23", default-features = false }
@@ -194,6 +196,7 @@ sea-orm = { version = "0.12.14", features = [
     "runtime-tokio-native-tls",
 ] }
 sqlx = "0.7"
+tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "0dd1055", features = ["net", "fs"] }
 tokio-util = "0.7"
 tracing-opentelemetry = "0.24"
 rand = { version = "0.8", features = ["small_rng"] }
@@ -334,7 +337,9 @@ opt-level = 2
 # Patch third-party crates for deterministic simulation.
 quanta = { git = "https://github.com/madsim-rs/quanta.git", rev = "948bdc3" }
 getrandom = { git = "https://github.com/madsim-rs/getrandom.git", rev = "e79a7ae" }
-tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "fe39bb8e" }
+# Don't patch `tokio-stream`, but only use the madsim version for **direct** dependencies.
+# Imagine an unpatched dependency depends on the original `tokio` and the patched `tokio-stream`.
+# tokio-stream = { git = "https://github.com/madsim-rs/tokio.git", rev = "0dd1055" }
 tokio-retry = { git = "https://github.com/madsim-rs/rust-tokio-retry.git", rev = "95e2fd3" }
 tokio-postgres = { git = "https://github.com/madsim-rs/rust-postgres.git", rev = "ac00d88" }
 futures-timer = { git = "https://github.com/madsim-rs/futures-timer.git", rev = "05b33b4" }

Cargo.toml

@@ -56,7 +56,7 @@
 
 RisingWave is a Postgres-compatible SQL engine engineered to provide the <i><b>simplest</b></i> and <i><b>most cost-efficient</b></i> approach for <b>processing</b>, <b>analyzing</b>, and <b>managing</b> real-time event streaming data.
 
-![RisingWave](https://github.com/risingwavelabs/risingwave/assets/41638002/10c44404-f78b-43ce-bbd9-3646690acc59)
+![RisingWave](./docs/dev/src/images/architecture_20240814.png)
 
 ## When to use RisingWave?
 RisingWave can ingest millions of events per second, continuously join live data streams with historical tables, and serve ad-hoc queries in real-time. Typical use cases include, but are not limited to:

README.md

@@ -11,7 +11,7 @@ ENV LANG en_US.utf8
 # Use AWS ubuntu mirror
 RUN sed -i 's|http://archive.ubuntu.com/ubuntu|http://us-east-2.ec2.archive.ubuntu.com/ubuntu/|g' /etc/apt/sources.list
 RUN apt-get update -yy && \
-    DEBIAN_FRONTEND=noninteractive apt-get -y install sudo make build-essential cmake protobuf-compiler curl parallel python3 python3-pip python3-venv software-properties-common \
+    DEBIAN_FRONTEND=noninteractive apt-get -y install sudo make build-essential cmake protobuf-compiler curl parallel python3 python3-pip python3-venv software-properties-common psmisc \
     openssl libssl-dev libsasl2-dev libcurl4-openssl-dev pkg-config bash openjdk-17-jdk wget unzip git tmux lld postgresql-client kcat netcat-openbsd mysql-client \
     maven zstd libzstd-dev locales \
     python3.12 python3.12-dev \

ci/Dockerfile

@@ -10,7 +10,7 @@ cat ../rust-toolchain
 # shellcheck disable=SC2155
 
 # REMEMBER TO ALSO UPDATE ci/docker-compose.yml
-export BUILD_ENV_VERSION=v20240731
+export BUILD_ENV_VERSION=v20240812
 
 export BUILD_TAG="public.ecr.aws/w1p7b4n3/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/build-ci-image.sh

@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240731
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
     depends_on:
       - mysql
       - sqlserver-server
@@ -85,7 +85,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240731
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
     depends_on:
       - mysql
       - db
@@ -108,12 +108,12 @@ services:
 
 
   rw-build-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240731
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240731
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -124,7 +124,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240731
+    image: public.ecr.aws/w1p7b4n3/rw-build-env:v20240812
     depends_on:
       db:
         condition: service_healthy
@@ -266,7 +266,6 @@ services:
       SCHEMA_REGISTRY_HOST_NAME: schemaregistry
       SCHEMA_REGISTRY_LISTENERS: http://schemaregistry:8082
       SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: message_queue:29092
-      SCHEMA_REGISTRY_DEBUG: 'true'
 
   pulsar-server:
     container_name: pulsar-server

ci/docker-compose.yml

@@ -97,7 +97,6 @@ function filter_stack_trace() {
   touch tmp
   cat "$1" \
   | sed -E '/  [1-9][0-9]+:/d' \
-  | sed -E '/  [3-9]+:/d' \
   | sed -E '/  at .rustc/d' \
   | sed -E '/  at ...cargo/d' > tmp
   cp tmp "$1"

ci/scripts/common.sh

@@ -9,7 +9,7 @@ echo "--- Download artifacts"
 download-and-decompress-artifact risingwave_simulation .
 chmod +x ./risingwave_simulation
 
-export RUST_LOG="risingwave_meta::barrier::recovery=debug,\
+export RUST_LOG="info,risingwave_meta::barrier::recovery=debug,\
 risingwave_meta::manager::catalog=debug,\
 risingwave_meta::rpc::ddl_controller=debug,\
 risingwave_meta::barrier::mod=debug,\

ci/scripts/deterministic-recovery-test.sh

@@ -65,6 +65,7 @@ def gen_pipeline_steps():
          env:
            GHCR_USERNAME: ghcr-username
            GHCR_TOKEN: ghcr-token
+           RW_LICENSE_KEY: rw-license-key
      - ./ci/plugins/docker-compose-logs
 """
     return pipeline_steps

ci/scripts/gen-integration-test-yaml.py

@@ -77,7 +77,7 @@ mv target/debug/risingwave_e2e_extended_mode_test-"$profile" target/debug/rising
 chmod +x ./target/debug/risingwave_e2e_extended_mode_test
 
 echo "--- e2e, $mode, streaming"
-RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info,risingwave_stream::common::table::state_table=warn" \
 cluster_start
 # Please make sure the regression is expected before increasing the timeout.
 sqllogictest -p 4566 -d dev './e2e_test/streaming/**/*.slt' --junit "streaming-${profile}"

ci/scripts/run-e2e-test.sh

@@ -19,7 +19,7 @@ start_single_node() {
 }
 
 stop_single_node() {
-  pkill risingwave
+  killall --wait risingwave
   rm -rf "$HOME/.risingwave/state_store"
   rm -rf "$HOME/.risingwave/meta_store"
 }
@@ -47,7 +47,6 @@ wait_single_node() {
 
 restart_single_node() {
   stop_single_node
-  sleep 5
   start_single_node "$PREFIX_LOG"/single-node-restarted.log &
   wait_single_node
 }

ci/scripts/single-node-utils.sh

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q0_temporal_filter;

ci/scripts/sql/nexmark/q0-temporal-filter.drop.sql

@@ -0,0 +1,7 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q0_temporal_filter
+AS
+SELECT auction, bidder, price, date_time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q0-temporal-filter.sql

@@ -4,4 +4,4 @@ CREATE SINK nexmark_q0
 AS
 SELECT auction, bidder, price, date_time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q0.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q1_temporal_filter;

ci/scripts/sql/nexmark/q1-temporal-filter.drop.sql

@@ -0,0 +1,10 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q1_temporal_filter
+AS
+SELECT auction,
+       bidder,
+       0.908 * price as price,
+       date_time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q1-temporal-filter.sql

@@ -7,4 +7,4 @@ SELECT auction,
        0.908 * price as price,
        date_time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q1.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q10_temporal_filter;

ci/scripts/sql/nexmark/q10-temporal-filter.drop.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q10_temporal_filter AS
+SELECT auction,
+       bidder,
+       price,
+       date_time,
+       TO_CHAR(date_time, 'YYYY-MM-DD') as date,
+       TO_CHAR(date_time, 'HH:MI')      as time
+FROM bid_filtered
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q10-temporal-filter.sql

@@ -8,4 +8,4 @@ SELECT auction,
        TO_CHAR(date_time, 'YYYY-MM-DD') as date,
        TO_CHAR(date_time, 'HH:MI')      as time
 FROM bid
-WITH ( connector = 'blackhole', type = 'append-only');
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q10.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q101_temporal_filter;

ci/scripts/sql/nexmark/q101-temporal-filter.drop.sql

@@ -0,0 +1,16 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q101_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    b.max_price AS current_highest_bid
+FROM auction a
+LEFT OUTER JOIN (
+    SELECT
+        b1.auction,
+        MAX(b1.price) max_price
+    FROM bid_filtered b1
+    GROUP BY b1.auction
+) b ON a.id = b.auction
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q101-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q102_temporal_filter;

ci/scripts/sql/nexmark/q102-temporal-filter.drop.sql

@@ -0,0 +1,14 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q102_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+HAVING COUNT(b.auction) >= (
+    SELECT COUNT(*) / COUNT(DISTINCT auction) FROM bid_filtered
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q102-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q103_temporal_filter;

ci/scripts/sql/nexmark/q103-temporal-filter.drop.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q103_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name
+FROM auction a
+WHERE a.id IN (
+    SELECT b.auction FROM bid_filtered b
+    GROUP BY b.auction
+    HAVING COUNT(*) >= 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q103-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q104_temporal_filter;

ci/scripts/sql/nexmark/q104-temporal-filter.drop.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q104_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name
+FROM auction a
+WHERE a.id NOT IN (
+    SELECT b.auction FROM bid_filtered b
+    GROUP BY b.auction
+    HAVING COUNT(*) < 20
+)
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q104-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_temporal_filter;

ci/scripts/sql/nexmark/q105-temporal-filter.drop.sql

@@ -0,0 +1,13 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+ORDER BY bid_count DESC
+LIMIT 1000
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_without_limit_temporal_filter;

ci/scripts/sql/nexmark/q105-without-limit-temporal-filter.drop.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_without_limit_temporal_filter AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid_filtered b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-without-limit-temporal-filter.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q105_without_limit;

ci/scripts/sql/nexmark/q105-without-limit.drop.sql

@@ -0,0 +1,11 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q105_without_limit AS
+SELECT
+    a.id AS auction_id,
+    a.item_name AS auction_item_name,
+    COUNT(b.auction) AS bid_count
+FROM auction a
+JOIN bid b ON a.id = b.auction
+GROUP BY a.id, a.item_name
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q105-without-limit.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q106;

ci/scripts/sql/nexmark/q106.drop.sql

@@ -0,0 +1,21 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+CREATE SINK nexmark_q106
+AS
+SELECT
+    MIN(final) AS min_final
+FROM
+    (
+        SELECT
+            auction.id,
+            MAX(price) AS final
+        FROM
+            auction,
+            bid
+        WHERE
+            bid.auction = auction.id
+            AND bid.date_time BETWEEN auction.date_time AND auction.expires
+        GROUP BY
+            auction.id
+    )
+WITH ( connector = 'blackhole', type = 'append-only', force_append_only = 'true');

ci/scripts/sql/nexmark/q106.sql

@@ -0,0 +1,3 @@
+-- noinspection SqlNoDataSourceInspectionForFile
+-- noinspection SqlResolveForFile
+DROP SINK nexmark_q107;