Merge branch 'main' into zj/iceberg

Cargo.toml

@@ -62,7 +62,7 @@ exclude = ["lints"]
 resolver = "2"
 
 [workspace.package]
-version = "1.3.0-alpha"
+version = "1.5.0-alpha"
 edition = "2021"
 homepage = "https://github.com/risingwavelabs/risingwave"
 keywords = ["sql", "database", "streaming"]

backwards-compat-tests/scripts/utils.sh

@@ -129,37 +129,47 @@ version_lt() {
 
 ################################### Entry Points
 
-# Get $OLD_VERSION and $NEW_VERSION for Risingwave
-get_rw_versions() {
-  # For backwards compat test we assume we are testing the latest version of RW (i.e. latest main commit)
-  # against the Nth latest release candidate, where N > 1. N can be larger,
-  # in case some old cluster did not upgrade.
-  local VERSION_OFFSET=4
-
-  # First we obtain a list of versions from git branch names.
-  # Then we normalize them to semver format (MAJOR.MINOR.PATCH).
-  echo "--- git branch origin output"
-  git branch -r | grep origin
-
-  # Extract X.Y.Z tags
-  echo "--- VERSION BRANCHES"
-  local tags=$(git tag | grep -E "^v[0-9]+\.[0-9]+\.[0-9]+$" | tr -d 'v' | tr -d ' ')
-  echo "$tags"
-
-  # Then we sort them in descending order.
-  echo "--- VERSIONS"
-  local sorted_versions=$(echo -e "$tags" | sort -t '.' -n)
-  echo "$sorted_versions"
-
-  # Then we take the Nth latest version.
-  # We set $OLD_VERSION to this.
-  OLD_VERSION=$(echo -e "$sorted_versions" | tail -n $VERSION_OFFSET | head -1)
+get_old_version() {
+   # For backwards compat test we assume we are testing the latest version of RW (i.e. latest main commit)
+   # against the Nth latest release candidate, where N > 1. N can be larger,
+   # in case some old cluster did not upgrade.
+   if [[ -z $VERSION_OFFSET ]]
+   then
+       local VERSION_OFFSET=1
+   fi
+
+   # First we obtain a list of versions from git branch names.
+   # Then we normalize them to semver format (MAJOR.MINOR.PATCH).
+   echo "--- git branch origin output"
+   git branch -r | grep origin
+
+   # Extract X.Y.Z tags
+   echo "--- VERSION BRANCHES"
+   local tags=$(git tag | grep -E "^v[0-9]+\.[0-9]+\.[0-9]+$" | tr -d 'v' | tr -d ' ')
+   echo "$tags"
+
+   # Then we sort them in descending order.
+   echo "--- VERSIONS"
+   local sorted_versions=$(echo -e "$tags" | sort -t '.' -n)
+   echo "$sorted_versions"
+
+   # Then we take the Nth latest version.
+   # We set $OLD_VERSION to this.
+   OLD_VERSION=$(echo -e "$sorted_versions" | tail -n $VERSION_OFFSET | head -1)
+}
 
+get_new_version() {
   # Next, for $NEW_VERSION we just scrape it from `workspace.package.version`.
   NEW_VERSION=$(cat Cargo.toml | grep "\[workspace\.package\]" -A 5 | sed -n 's/version = \"\([0-9]*\.[0-9]*\.[0-9]*\).*/\1/p' | tr -d ' ')
+}
+
+# Get $OLD_VERSION and $NEW_VERSION for Risingwave
+get_rw_versions() {
+  get_old_version
+  get_new_version
 
-  # Then we assert that `$OLD_VERSION` < `$NEW_VERSION`.
-  if version_lt "$OLD_VERSION" "$NEW_VERSION"
+  # Then we assert that `$OLD_VERSION` <= `$NEW_VERSION`.
+  if version_le "$OLD_VERSION" "$NEW_VERSION"
   then
     echo "OLD_VERSION: $OLD_VERSION"
     echo "NEW_VERSION: $NEW_VERSION"

ci/scripts/run-backfill-tests.sh

@@ -96,21 +96,20 @@ restart_cn() {
 test_snapshot_and_upstream_read() {
   echo "--- e2e, ci-backfill, test_snapshot_and_upstream_read"
   cargo make ci-start ci-backfill
-
-  run_sql_file "$PARENT_PATH"/sql/backfill/create_base_table.sql
+  run_sql_file "$PARENT_PATH"/sql/backfill/basic/create_base_table.sql
 
   # Provide snapshot
-  run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql
+  run_sql_file "$PARENT_PATH"/sql/backfill/basic/insert.sql
 
   # Provide updates ...
-  run_sql_file "$PARENT_PATH"/sql/backfill/insert.sql &
+  run_sql_file "$PARENT_PATH"/sql/backfill/basic/insert.sql &
 
   # ... and concurrently create mv.
-  run_sql_file "$PARENT_PATH"/sql/backfill/create_mv.sql &
+  run_sql_file "$PARENT_PATH"/sql/backfill/basic/create_mv.sql &
 
   wait
 
-  run_sql_file "$PARENT_PATH"/sql/backfill/select.sql </dev/null
+  run_sql_file "$PARENT_PATH"/sql/backfill/basic/select.sql </dev/null
 
   cargo make kill
   cargo make wait-processes-exit
@@ -146,7 +145,29 @@ test_backfill_tombstone() {
   ./risedev psql -c "CREATE MATERIALIZED VIEW m1 as select * from tomb;"
   echo "--- Kill cluster"
   kill_cluster
+  cargo make wait-processes-exit
+  wait
+}
+
+test_replication_with_column_pruning() {
+  echo "--- e2e, test_replication_with_column_pruning"
+  cargo make ci-start ci-backfill
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/create_base_table.sql
+  # Provide snapshot
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/insert.sql
+
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/create_mv.sql &
+
+  # Provide upstream updates
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/insert.sql &
+
   wait
+
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/select.sql </dev/null
+  run_sql_file "$PARENT_PATH"/sql/backfill/replication_with_column_pruning/drop.sql
+  echo "--- Kill cluster"
+  cargo make kill
+  cargo make wait-processes-exit
 }
 
 # Test sink backfill recovery
@@ -184,6 +205,7 @@ main() {
   set -euo pipefail
   test_snapshot_and_upstream_read
   test_backfill_tombstone
+  test_replication_with_column_pruning
   test_sink_backfill_recovery
 }
 

ci/scripts/sql/backfill/basic/drop.sql

@@ -0,0 +1,2 @@
+DROP MATERIALIZED VIEW mv1;
+DROP TABLE t1;

ci/scripts/sql/backfill/replication_with_column_pruning/create_base_table.sql

@@ -0,0 +1,2 @@
+SET STREAMING_ENABLE_ARRANGEMENT_BACKFILL=true;
+create table t1 (x int, y int, z int);

ci/scripts/sql/backfill/replication_with_column_pruning/create_mv.sql

@@ -0,0 +1 @@
+create materialized view mv1 as select y, x from t1;

ci/scripts/sql/backfill/replication_with_column_pruning/drop.sql

@@ -0,0 +1,2 @@
+DROP MATERIALIZED VIEW mv1;
+DROP TABLE t1;

ci/scripts/sql/backfill/replication_with_column_pruning/insert.sql

@@ -0,0 +1,2 @@
+insert into t1 select generate_series + 1, generate_series + 2, generate_series + 3 from generate_series(1, 100000);
+flush;

ci/scripts/sql/backfill/replication_with_column_pruning/select.sql

@@ -0,0 +1 @@
+SELECT * FROM mv1 ORDER BY x LIMIT 10;

ci/workflows/main-cron.yml

@@ -575,9 +575,9 @@ steps:
     timeout_in_minutes: 5
 
   # Backwards compatibility tests
-  - label: "Backwards compatibility tests"
+  - label: "Backwards compatibility tests version_offset={{matrix.version_offset}}"
     key: "backwards-compat-tests"
-    command: "RW_COMMIT=$BUILDKITE_COMMIT ci/scripts/backwards-compat-test.sh -p ci-release"
+    command: "VERSION_OFFSET={{matrix.version_offset}} RW_COMMIT=$BUILDKITE_COMMIT ci/scripts/backwards-compat-test.sh -p ci-release"
     if: |
       !(build.pull_request.labels includes "ci/main-cron/skip-ci") && build.env("CI_STEPS") == null
       || build.pull_request.labels includes "ci/run-backwards-compat-tests"
@@ -590,6 +590,21 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
+    matrix:
+      setup:
+        # Test the 4 latest versions against the latest main.
+        # e.g.
+        # 1: 2.0.0
+        # 2: 1.1.1
+        # 3: 1.0.1
+        # 4: 1.0.0
+        # It is ordered by the full version number, rather than minor / major version.
+        # We can change to just be on major version in the future.
+        version_offset:
+          - "1"
+          - "2"
+          - "3"
+          - "4"
     timeout_in_minutes: 30
     retry: *auto-retry
 

ci/workflows/pull-request.yml

@@ -518,7 +518,7 @@ steps:
 
   # Backwards compatibility tests
   - label: "Backwards compatibility tests"
-    command: "RW_COMMIT=$BUILDKITE_COMMIT ci/scripts/backwards-compat-test.sh -p ci-dev"
+    command: "VERSION_OFFSET={{matrix.version_offset}} RW_COMMIT=$BUILDKITE_COMMIT ci/scripts/backwards-compat-test.sh -p ci-dev"
     if: |
       build.pull_request.labels includes "breaking-change" ||
         build.pull_request.labels includes "ci/run-backwards-compat-tests" ||
@@ -531,6 +531,21 @@ steps:
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs
+    matrix:
+      setup:
+        # Test the 4 latest versions against the latest main.
+        # e.g.
+        # 1: 2.0.0
+        # 2: 1.1.1
+        # 3: 1.0.1
+        # 4: 1.0.0
+        # It is ordered by the full version number, rather than minor / major version.
+        # We can change to just be on major version in the future.
+        version_offset:
+          - "1"
+          - "2"
+          - "3"
+          - "4"
     timeout_in_minutes: 25
 
   # Sqlsmith differential testing

docker/docker-compose-with-local-fs.yml

@@ -0,0 +1,102 @@
+---
+version: "3"
+x-image: &image
+  image: ${RW_IMAGE:-risingwavelabs/risingwave:nightly-20231211}
+services:
+  risingwave-standalone:
+    <<: *image
+    command: "standalone --meta-opts=\" \
+                    --listen-addr 0.0.0.0:5690 \
+                    --advertise-addr 0.0.0.0:5690 \
+                    --dashboard-host 0.0.0.0:5691 \
+                    --prometheus-host 0.0.0.0:1250 \
+                    --connector-rpc-endpoint 0.0.0.0:50051 \
+                    --backend etcd \
+                    --etcd-endpoints etcd-0:2388 \
+                    --state-store hummock+fs://<local-path> \
+                    --data-directory hummock_001 \
+                    --config-path /risingwave.toml\" \
+                 --compute-opts=\" \
+                    --config-path /risingwave.toml \
+                    --listen-addr 0.0.0.0:5688 \
+                    --prometheus-listener-addr 0.0.0.0:1222 \
+                    --advertise-addr 0.0.0.0:5688 \
+                    --async-stack-trace verbose \
+                    --connector-rpc-endpoint 0.0.0.0:50051 \
+                    # --parallelism 4 \
+                    --role both \
+                    --meta-address http://0.0.0.0:5690\" \
+                 --frontend-opts=\" \
+                   --config-path /risingwave.toml \
+                   --listen-addr 0.0.0.0:4566 \
+                   --advertise-addr 0.0.0.0:4566 \
+                   --prometheus-listener-addr 0.0.0.0:2222 \
+                   --health-check-listener-addr 0.0.0.0:6786 \
+                   --meta-addr http://0.0.0.0:5690\" \
+                 --compactor-opts=\" \
+                   --listen-addr 0.0.0.0:6660 \
+                   --prometheus-listener-addr 0.0.0.0:1260 \
+                   --advertise-addr 0.0.0.0:6660 \
+                   --meta-address http://0.0.0.0:5690\""
+    expose:
+      - "6660"
+      - "1260"
+      - "4566"
+      - "5688"
+      - "1222"
+      - "5690"
+      - "1250"
+      - "5691"
+      - "2222"
+    ports:
+      - "4566:4566"
+      - "5690:5690"
+      - "5691:5691"
+      - "1222:1222"
+      - "1250:1250"
+      - "1260:1260"
+      - "2222:2222"
+    depends_on:
+      - etcd-0
+    volumes:
+      - "./risingwave.toml:/risingwave.toml"
+    environment:
+      RUST_BACKTRACE: "1"
+      # If ENABLE_TELEMETRY is not set, telemetry will start by default
+      ENABLE_TELEMETRY: ${ENABLE_TELEMETRY:-true}
+    container_name: risingwave-standalone
+    healthcheck:
+      test:
+        - CMD-SHELL
+        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/6660; exit $$?;'
+        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5688; exit $$?;'
+        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/4566; exit $$?;'
+        - bash -c 'printf \"GET / HTTP/1.1\n\n\" > /dev/tcp/127.0.0.1/5690; exit $$?;'
+      interval: 1s
+      timeout: 5s
+    restart: always
+    deploy:
+      resources:
+        limits:
+          memory: <config-the-allocated-memory>
+        reservations:
+          memory: <config-the-allocated-memory>
+  etcd-0:
+    extends:
+      file: docker-compose.yml
+      service: etcd-0
+  grafana-0:
+    extends:
+      file: docker-compose.yml
+      service: grafana-0
+  prometheus-0:
+    extends:
+      file: docker-compose.yml
+      service: prometheus-0
+volumes:
+  etcd-0:
+    external: false
+  grafana-0:
+    external: false
+  prometheus-0:
+    external: false

integration_tests/big-query-sink/README.md

@@ -23,9 +23,9 @@ CREATE table '${project_id}'.'${dataset_id}'.'${table_id}'(
 
 4. Execute the SQL queries in sequence:
 
-- append-only/create_source.sql
-- append-only/create_mv.sql
-- append-only/create_sink.sql
+- create_source.sql
+- create_mv.sql
+- create_sink.sql
 
     1. We need to obtain the JSON file for Google Cloud service accounts, which can be configured here: https://console.cloud.google.com/iam-admin/serviceaccounts.
     2. Because BigQuery has limited support for updates and deletes, we currently only support 'append only'

integration_tests/big-query-sink/prepare.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+
+set -euo pipefail
+
+# set gcloud
+docker compose exec gcloud-cli gcloud auth login --cred-file=/gcp-rwctest.json
+
+docker compose exec gcloud-cli gcloud config set project rwctest
+
+bq_prepare_file='bq_prepare.sql'
+bq_prepare_content=$(cat $bq_prepare_file)
+
+docker compose exec gcloud-cli bq query --use_legacy_sql=false "$bq_prepare_content"
+
+sleep 10

integration_tests/big-query-sink/sink_check.py

@@ -0,0 +1,25 @@
+import json
+import subprocess
+import sys
+
+relations = ['rwctest.bqtest.bq_sink']
+
+failed_cases = []
+for rel in relations:
+    sql = f"SELECT COUNT(*) AS count FROM `{rel}`"
+    print(f"run sql: {sql} on Bigquery")
+    rows = subprocess.check_output(
+        ["docker", "compose", "exec", "gcloud-cli", "bq", "query", "--use_legacy_sql=false", "--format=json", sql],
+    )
+    rows = int(json.loads(rows.decode("utf-8").strip())[0]['count'])
+    print(f"{rows} rows in {rel}")
+    if rows < 1:
+        failed_cases.append(rel)
+
+    drop_sql = f"DROP TABLE IF EXISTS `{rel}`"
+    subprocess.run(["docker", "compose", "exec", "gcloud-cli", "bq", "query", "--use_legacy_sql=false", drop_sql],
+                   check=True)
+
+if len(failed_cases) != 0:
+    print(f"Data check failed for case {failed_cases}")
+    sys.exit(1)