feat: add sql migration script and related ci tests (#16000) (#16052)

Co-authored-by: August <[email protected]>
Co-authored-by: Huangjw <[email protected]>

.typos.toml

@@ -10,6 +10,7 @@ steam = "stream"        # You played with Steam games too much.
 ot = "ot"
 bui = "bui"
 mosquitto = "mosquitto" # This is a MQTT broker.
+abd = "abd"
 
 [default.extend-identifiers]
 

Cargo.toml

@@ -160,6 +160,12 @@ opentelemetry = "0.21"
 opentelemetry-otlp = "0.14"
 opentelemetry_sdk = { version = "0.21", default-features = false }
 opentelemetry-semantic-conventions = "0.13"
+sea-orm = { version = "0.12.14", features = [
+  "sqlx-mysql",
+  "sqlx-postgres",
+  "sqlx-sqlite",
+  "runtime-tokio-native-tls",
+] }
 tokio-util = "0.7"
 tracing-opentelemetry = "0.22"
 

ci/scripts/e2e-sql-migration.sh

@@ -0,0 +1,89 @@
+#!/usr/bin/env bash
+
+# Exits as soon as any line fails.
+set -euo pipefail
+
+export RW_PREFIX=$PWD/.risingwave
+export RW_PREFIX_DATA=$RW_PREFIX/data
+
+source ci/scripts/common.sh
+
+wait_for_recovery() {
+  set +e
+  timeout 20s bash -c '
+    while true; do
+      echo "Polling every 1s to check if the recovery is complete for 20s"
+      if psql -h localhost -p 4566 -d dev -U root -c "FLUSH;" </dev/null
+      then exit 0;
+      else sleep 1;
+      fi
+    done
+  '
+  STATUS=$?
+  set -e
+  if [[ $STATUS -ne 0 ]]; then
+    echo "Cluster failed to get recovered: $STATUS"
+    exit 1
+  else
+    echo "Cluster is recovered"
+  fi
+}
+
+while getopts 'p:' opt; do
+    case ${opt} in
+        p )
+            profile=$OPTARG
+            ;;
+        \? )
+            echo "Invalid Option: -$OPTARG" 1>&2
+            exit 1
+            ;;
+        : )
+            echo "Invalid option: $OPTARG requires an argument" 1>&2
+            ;;
+    esac
+done
+shift $((OPTIND -1))
+
+download_and_prepare_rw "$profile" common
+
+echo "--- starting risingwave cluster, ci-1cn-1fe-with-recovery"
+RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+cargo make ci-start ci-1cn-1fe-with-recovery
+
+echo "--- init cluster with some data & DDL"
+sqllogictest -d dev -h localhost -p 4566 './e2e_test/sql_migration/prepare.slt'
+
+echo "--- kill cluster"
+cargo make ci-kill
+
+echo "--- restart etcd"
+cargo make dev ci-meta-etcd-for-migration
+
+echo "--- run migration"
+mkdir -p "${RW_PREFIX_DATA}/sqlite/"
+./target/debug/risingwave risectl \
+meta \
+migration \
+--etcd-endpoints localhost:2388 \
+--sql-endpoint sqlite://"${RW_PREFIX_DATA}/sqlite/metadata.db"\?mode=rwc \
+-f
+
+echo "--- kill etcd"
+cargo make ci-kill
+
+echo "--- starting risingwave cluster, meta-1cn-1fe-sqlite"
+cargo make dev meta-1cn-1fe-sqlite
+
+echo "--- wait for recovery"
+wait_for_recovery
+
+echo "--- run check"
+sqllogictest -d dev -h localhost -p 4566 './e2e_test/sql_migration/check.slt'
+
+echo "--- kill cluster"
+cargo make kill
+
+echo "--- clean data"
+cargo make clean-data
+

ci/workflows/pull-request.yml

@@ -715,6 +715,22 @@ steps:
     timeout_in_minutes: 30
     retry: *auto-retry
 
+  - label: "e2e sql migration test"
+    command: "ci/scripts/e2e-sql-migration.sh -p ci-dev"
+    if: build.pull_request.labels includes "ci/run-e2e-sql-migration-tests" || build.env("CI_STEPS") =~ /(^|,)e2e-sql-migration-tests?(,|$$)/
+    depends_on:
+      - "build"
+      - "build-other"
+      - "docslt"
+    plugins:
+      - docker-compose#v5.1.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 10
+    retry: *auto-retry
+
   # FIXME(kwannoel): Let the github PR labeller label it, if sqlsmith source files has changes.
   - label: "fuzz test"
     command: "ci/scripts/pr-fuzz-test.sh -p ci-dev"

e2e_test/sql_migration/check.slt

@@ -0,0 +1,105 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+query T rowsort
+show databases;
+----
+db1
+dev
+
+query T rowsort
+show schemas;
+----
+information_schema
+pg_catalog
+public
+rw_catalog
+schema1
+
+query T
+SELECT setting FROM pg_catalog.pg_settings where name = 'max_concurrent_creating_streaming_jobs';
+----
+4
+
+query T rowsort
+select name, relation_type from rw_relations where relation_type != 'system table' AND relation_type != 'view';
+----
+ddl_subscription_table subscription
+idx1 index
+m_simple table
+mv1 materialized view
+mv2 materialized view
+s_simple_1 sink
+sink sink
+src source
+t1 table
+t_simple table
+
+query T
+show views;
+----
+v1
+
+query T
+select name, type_, provider from rw_connections;
+----
+conn0 MOCK PRIVATELINK
+
+query TTTTT
+show functions;
+----
+int_42 (empty) integer javascript NULL
+
+statement ok
+insert into t1 select * from generate_series(1, 1000);
+
+query I
+select count(*) from t1;
+----
+2000
+
+statement ok
+create materialized view mv3 as select * from mv2;
+
+statement ok
+REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA schema1 FROM user1;
+
+statement error Permission denied
+drop source src;
+
+statement ok
+drop source src cascade;
+
+statement ok
+drop connection conn0;
+
+statement ok
+drop function int_42;
+
+statement ok
+drop sink s_simple_1;
+
+statement error Permission denied
+drop table t1;
+
+statement ok
+drop table t1 cascade;
+
+statement ok
+drop table t_simple;
+
+statement ok
+drop table m_simple;
+
+statement ok
+drop user user1;
+
+statement ok
+drop schema schema1;
+
+statement ok
+drop database db1;
+
+query T
+select name, relation_type from rw_relations where relation_type != 'system table' AND relation_type != 'view';
+----

e2e_test/sql_migration/prepare.slt

@@ -0,0 +1,68 @@
+statement ok
+SET RW_IMPLICIT_FLUSH TO true;
+
+statement ok
+create database db1;
+
+statement ok
+create schema schema1;
+
+statement ok
+ALTER SYSTEM SET max_concurrent_creating_streaming_jobs TO 4;
+
+statement ok
+create source src (v int) with (
+    connector = 'datagen',
+    fields.v.kind = 'sequence',
+    fields.v.start = '1',
+    fields.v.end  = '10',
+    datagen.rows.per.second='15',
+    datagen.split.num = '1'
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+create table t1(v1 int);
+
+statement ok
+create materialized view mv1 as select * from t1;
+
+statement ok
+create materialized view mv2 as select * from src;
+
+statement ok
+create view v1 as select * from mv1;
+
+statement ok
+CREATE SINK sink FROM mv2 WITH (connector='blackhole');
+
+statement ok
+create user user1;
+
+statement ok
+GRANT ALL PRIVILEGES ON ALL TABLES IN SCHEMA schema1 TO user1;
+
+statement ok
+CREATE CONNECTION conn0 WITH (type = 'privatelink', provider = 'mock');
+
+statement ok
+create index idx1 on t1(v1);
+
+statement ok
+create table t_simple (v1 int, v2 int);
+
+statement ok
+create table m_simple (v1 int primary key, v2 int);
+
+statement ok
+create sink s_simple_1 into m_simple as select v1, v2 from t_simple;
+
+statement ok
+create subscription ddl_subscription_table from mv2 with(retention = '1D');
+
+statement ok
+insert into t1 select * from generate_series(1, 1000);
+
+statement ok
+create function int_42() returns int language javascript as $$
+    return 42;
+$$;

risedev.yml

@@ -898,6 +898,11 @@ profile:
       - use: etcd
       - use: minio
 
+  ci-meta-etcd-for-migration:
+    config-path: src/config/ci.toml
+    steps:
+      - use: etcd
+
   ci-iceberg-test:
     config-path: src/config/ci-iceberg-test.toml
     steps:

src/ctl/Cargo.toml

@@ -24,18 +24,22 @@ futures = { version = "0.3", default-features = false, features = ["alloc"] }
 hex = "0.4"
 inquire = "0.7.0"
 itertools = "0.12"
+memcomparable = "0.2"
 prost = { workspace = true }
 regex = "1.10.0"
 risingwave_common = { workspace = true }
 risingwave_connector = { workspace = true }
 risingwave_frontend = { workspace = true }
 risingwave_hummock_sdk = { workspace = true }
 risingwave_meta = { workspace = true }
+risingwave_meta_model_migration = { workspace = true }
+risingwave_meta_model_v2 = { workspace = true }
 risingwave_object_store = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
 risingwave_storage = { workspace = true }
 risingwave_stream = { workspace = true }
+sea-orm = { workspace = true }
 serde = "1"
 serde_json = "1"
 serde_yaml = "0.9.25"

src/ctl/src/cmd_impl/meta.rs

@@ -15,13 +15,15 @@
 mod backup_meta;
 mod cluster_info;
 mod connection;
+mod migration;
 mod pause_resume;
 mod reschedule;
 mod serving;
 
 pub use backup_meta::*;
 pub use cluster_info::*;
 pub use connection::*;
+pub use migration::*;
 pub use pause_resume::*;
 pub use reschedule::*;
 pub use serving::*;