Merge branch 'main' into kwannoel/test-rate-limit

.cargo/config.toml

@@ -34,7 +34,7 @@ rustflags = [
 rustflags = [
     "--cfg",
     "tokio_unstable",
-    # uncomment the following two lines to enable `TaskLocalAlloc`
+    # Uncomment the following two lines to enable `TaskLocalAlloc`.
     # "--cfg",
     # "enable_task_local_alloc",
 ]

.vscode/settings.json.example

@@ -9,16 +9,29 @@
         "src/risedevtool/schemas/risedev.json": "risedev.yml",
         "src/risedevtool/schemas/risedev-profiles.user.json": "risedev-profiles.user.yml"
     },
-    // // ============================================================================
+
+    // Optional: check with custom lints using `cargo dylint`
+    //
+    // "rust-analyzer.check.overrideCommand": [
+    //     "cargo",
+    //     "dylint",
+    //     "--all",
+    //     "--workspace",
+    //     "--",
+    //     "--all-targets",
+    //     "--message-format=json"
+    // ],
+
+    // // ===========================================================================
     // // BEGIN SECTION: configurations for the development of deterministic testing
-    // // ============================================================================
+    // // ===========================================================================
     // "rust-analyzer.cargo.extraEnv": {
     //     "RUSTFLAGS": "--cfg tokio_unstable --cfg madsim"
     // },
     // "rust-analyzer.runnableEnv": {
     //     "RUSTFLAGS": "--cfg tokio_unstable --cfg madsim"
     // },
-    // "rust-analyzer.checkOnSave.overrideCommand": [
+    // "rust-analyzer.check.overrideCommand": [
     //     "cargo",
     //     "check",
     //     "-p",

Cargo.toml

@@ -114,7 +114,7 @@ hashbrown = { version = "0.14.0", features = [
   "nightly",
 ] }
 criterion = { version = "0.5", features = ["async_futures"] }
-tonic = { package = "madsim-tonic", version = "0.4.0" }
+tonic = { package = "madsim-tonic", version = "0.4.1" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }
 prost = { version = "0.12" }
 icelake = { git = "https://github.com/icelake-io/icelake", rev = "5cdcdffd24f4624a0a43f92c5f368988169a799b", features = [
@@ -173,6 +173,7 @@ risingwave_udf = { path = "./src/udf" }
 risingwave_variables = { path = "./src/utils/variables" }
 risingwave_java_binding = { path = "./src/java_binding" }
 risingwave_jni_core = { path = "src/jni_core" }
+tokio-util = "0.7"
 
 [workspace.lints.rust]
 # `forbid` will also prevent the misuse of `#[allow(unused)]`

Makefile.toml

@@ -1223,7 +1223,7 @@ script = """
 scripts/check/check-trailing-spaces.sh --fix
 """
 
-[tasks.check]
+[tasks.check-fast]
 category = "RiseDev - Check"
 dependencies = [
   "warn-on-missing-tools",
@@ -1232,9 +1232,12 @@ dependencies = [
   "check-fmt",
   "check-trailing-spaces",
   "check-typos",
-  "check-clippy",
-  "check-java",
 ]
+description = "Perform part of the pre-CI checks that are fast to run"
+
+[tasks.check]
+category = "RiseDev - Check"
+dependencies = ["check-fast", "check-clippy", "check-java"]
 script = """
 #!/usr/bin/env bash
 
@@ -1250,11 +1253,7 @@ alias = "check"
 category = "RiseDev - Check"
 dependencies = [
   "warn-on-missing-tools",
-  "check-hakari",
-  "check-dep-sort",
-  "check-fmt",
-  "check-trailing-spaces",
-  "check-typos",
+  "check-fast",
   "check-clippy-fix",
   "check-java-fix",
 ]

backwards-compat-tests/scripts/utils.sh

@@ -118,6 +118,15 @@ seed_json_kafka() {
   insert_json_kafka '{"timestamp": "2023-07-28 06:54:00", "user_id": 9, "page_id": 4, "action": "yjtyjtyyy"}'
 }
 
+# https://stackoverflow.com/a/4024263
+version_le() {
+    printf '%s\n' "$1" "$2" | sort -C -V
+}
+
+version_lt() {
+    ! version_le "$2" "$1"
+}
+
 ################################### Entry Points
 
 # Get $OLD_VERSION and $NEW_VERSION for Risingwave
@@ -150,14 +159,13 @@ get_rw_versions() {
   NEW_VERSION=$(cat Cargo.toml | grep "\[workspace\.package\]" -A 5 | sed -n 's/version = \"\([0-9]*\.[0-9]*\.[0-9]*\).*/\1/p' | tr -d ' ')
 
   # Then we assert that `$OLD_VERSION` < `$NEW_VERSION`.
-  local TOP=$(echo -e "$OLD_VERSION\n$NEW_VERSION" | sort -t '.' -n | tail -1)
-  if [[ "$TOP" != "$NEW_VERSION" ]]
+  if version_lt "$OLD_VERSION" "$NEW_VERSION"
   then
-    echo "ERROR: $OLD_VERSION > $NEW_VERSION"
-    exit 1
-  else
     echo "OLD_VERSION: $OLD_VERSION"
     echo "NEW_VERSION: $NEW_VERSION"
+  else
+    echo "ERROR: $OLD_VERSION >= $NEW_VERSION"
+    exit 1
   fi
 }
 
@@ -211,6 +219,18 @@ seed_old_cluster() {
   echo "--- KAFKA TEST: Validating old cluster"
   sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/kafka/validate_original.slt"
 
+  # Test invalid WITH options, if OLD_VERSION <= 1.5.0
+  if version_le "$OLD_VERSION" "1.5.0" ; then
+    echo "--- KAFKA TEST (invalid options): Seeding old cluster with data"
+    sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/kafka/invalid_options/seed.slt"
+
+    echo "--- KAFKA TEST (invalid options): wait 5s for kafka to process data"
+    sleep 5
+
+    echo "--- KAFKA TEST (invalid options): Validating old cluster"
+    sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/kafka/invalid_options/validate_original.slt"
+  fi
+
   echo "--- Killing cluster"
   kill_cluster
   echo "--- Killed cluster"
@@ -240,5 +260,11 @@ validate_new_cluster() {
   echo "--- KAFKA TEST: Validating new cluster"
   sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/kafka/validate_restart.slt"
 
+  # Test invalid WITH options, if OLD_VERSION <= 1.5.0
+  if version_le "$OLD_VERSION" "1.5.0" ; then
+    echo "--- KAFKA TEST (invalid options): Validating new cluster"
+    sqllogictest -d dev -h localhost -p 4566 "$TEST_DIR/kafka/invalid_options/validate_restart.slt"
+  fi
+
   kill_cluster
 }

backwards-compat-tests/slt/kafka/invalid_options/seed.slt

@@ -0,0 +1,23 @@
+# In older versions of RisingWave, invalid WITH options are accepted and persisted.
+# We use this test to ensure that existing sources with invalid options can still be used.
+
+statement ok
+CREATE SOURCE IF NOT EXISTS kafka_source_with_invalid_option
+(
+ action varchar,
+ user_id integer,
+ obj_id integer,
+ name varchar,
+ page_id integer,
+ age integer
+)
+WITH (
+ connector='kafka',
+ topic='backwards_compat_test_kafka_source',
+ properties.bootstrap.server='localhost:29092',
+ scan.startup.mode='earliest',
+ invalid_option='oops'
+) FORMAT PLAIN ENCODE JSON;
+
+statement ok
+CREATE MATERIALIZED VIEW kafka_mv1_with_invalid_option as SELECT * FROM kafka_source_with_invalid_option;

backwards-compat-tests/slt/kafka/invalid_options/validate_original.slt

@@ -0,0 +1,12 @@
+query I rowsort
+SELECT * FROM kafka_mv1_with_invalid_option;
+----
+6786745ge 6 NULL NULL 3 NULL
+erwerhghj 4 NULL NULL 2 NULL
+fgbgfnyyy 7 NULL NULL 3 NULL
+fsdfgerrg 2 NULL NULL 1 NULL
+gtrgretrg 1 NULL NULL 1 NULL
+kiku7ikkk 5 NULL NULL 2 NULL
+sdfergtth 3 NULL NULL 1 NULL
+werwerwwe 8 NULL NULL 4 NULL
+yjtyjtyyy 9 NULL NULL 4 NULL

backwards-compat-tests/slt/kafka/invalid_options/validate_restart.slt

@@ -0,0 +1,51 @@
+# create a new mv on source, it should retrieve all records
+# and match mv1.
+statement ok
+CREATE MATERIALIZED VIEW kafka_mv2_with_invalid_option as SELECT * FROM kafka_source_with_invalid_option;
+
+sleep 5s
+
+query I rowsort
+SELECT * FROM kafka_mv2_with_invalid_option;
+----
+6786745ge 6 NULL NULL 3 NULL
+6786745ge 6 NULL NULL 3 NULL
+erwerhghj 4 NULL NULL 2 NULL
+erwerhghj 4 NULL NULL 2 NULL
+fgbgfnyyy 7 NULL NULL 3 NULL
+fgbgfnyyy 7 NULL NULL 3 NULL
+fsdfgerrg 2 NULL NULL 1 NULL
+fsdfgerrg 2 NULL NULL 1 NULL
+gtrgretrg 1 NULL NULL 1 NULL
+gtrgretrg 1 NULL NULL 1 NULL
+kiku7ikkk 5 NULL NULL 2 NULL
+kiku7ikkk 5 NULL NULL 2 NULL
+sdfergtth 3 NULL NULL 1 NULL
+sdfergtth 3 NULL NULL 1 NULL
+werwerwwe 8 NULL NULL 4 NULL
+werwerwwe 8 NULL NULL 4 NULL
+yjtyjtyyy 9 NULL NULL 4 NULL
+yjtyjtyyy 9 NULL NULL 4 NULL
+
+# MV1 should also have new records
+query I rowsort
+SELECT * FROM kafka_mv1_with_invalid_option;
+----
+6786745ge 6 NULL NULL 3 NULL
+6786745ge 6 NULL NULL 3 NULL
+erwerhghj 4 NULL NULL 2 NULL
+erwerhghj 4 NULL NULL 2 NULL
+fgbgfnyyy 7 NULL NULL 3 NULL
+fgbgfnyyy 7 NULL NULL 3 NULL
+fsdfgerrg 2 NULL NULL 1 NULL
+fsdfgerrg 2 NULL NULL 1 NULL
+gtrgretrg 1 NULL NULL 1 NULL
+gtrgretrg 1 NULL NULL 1 NULL
+kiku7ikkk 5 NULL NULL 2 NULL
+kiku7ikkk 5 NULL NULL 2 NULL
+sdfergtth 3 NULL NULL 1 NULL
+sdfergtth 3 NULL NULL 1 NULL
+werwerwwe 8 NULL NULL 4 NULL
+werwerwwe 8 NULL NULL 4 NULL
+yjtyjtyyy 9 NULL NULL 4 NULL
+yjtyjtyyy 9 NULL NULL 4 NULL