Merge branch 'main' into jinser/sink-DynamoDB

ci/scripts/run-backfill-tests.sh

@@ -23,7 +23,7 @@ TEST_DIR=$PWD/e2e_test
 BACKGROUND_DDL_DIR=$TEST_DIR/background_ddl
 COMMON_DIR=$BACKGROUND_DDL_DIR/common
 
-CLUSTER_PROFILE='ci-1cn-1fe-kafka-with-recovery'
+CLUSTER_PROFILE='ci-1cn-1fe-user-kafka-with-recovery'
 echo "--- Configuring cluster profiles"
 if [[ -n "${BUILDKITE:-}" ]]; then
   echo "Running in buildkite"
@@ -187,14 +187,14 @@ test_sink_backfill_recovery() {
 
   # Restart
   restart_cluster
-  sleep 3
+  sleep 5
 
   # Sink back into rw
   run_sql "CREATE TABLE table_kafka (v1 int primary key)
     WITH (
       connector = 'kafka',
       topic = 's_kafka',
-      properties.bootstrap.server = 'localhost:29092',
+      properties.bootstrap.server = 'message_queue:29092',
   ) FORMAT DEBEZIUM ENCODE JSON;"
 
   sleep 10

ci/workflows/main-cron.yml

@@ -704,7 +704,7 @@ steps:
       - "build"
     plugins:
       - docker-compose#v5.1.0:
-          run: rw-build-env
+          run: source-test-env
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs

ci/workflows/pull-request.yml

@@ -700,7 +700,7 @@ steps:
       - "build"
     plugins:
       - docker-compose#v5.1.0:
-          run: rw-build-env
+          run: source-test-env
           config: ci/docker-compose.yml
           mount-buildkite-agent: true
       - ./ci/plugins/upload-failure-logs

e2e_test/backfill/sink/create_sink.slt

@@ -20,7 +20,7 @@ from t x join t y
     on x.v1 = y.v1
 with (
     connector='kafka',
-    properties.bootstrap.server='localhost:29092',
+    properties.bootstrap.server='message_queue:29092',
     topic='s_kafka',
     primary_key='v1',
     allow.auto.create.topics=true,

e2e_test/error_ui/extended/main.slt

@@ -4,8 +4,9 @@ selet 1;
 db error: ERROR: Failed to prepare the statement
 
 Caused by:
-  sql parser error: Expected an SQL statement, found: selet at line:1, column:6
-Near "selet"
+  sql parser error: expected an SQL statement, found: selet at line 1, column 1
+LINE 1: selet 1;
+             ^
 
 
 query error

e2e_test/error_ui/simple/main.slt

@@ -4,8 +4,9 @@ selet 1;
 db error: ERROR: Failed to run the query
 
 Caused by:
-  sql parser error: Expected an SQL statement, found: selet at line:1, column:6
-Near "selet"
+  sql parser error: expected an SQL statement, found: selet at line 1, column 1
+LINE 1: selet 1;
+             ^
 
 
 statement error

e2e_test/source/basic/datagen.slt

@@ -186,9 +186,9 @@ statement ok
 drop table s1;
 
 # Do NOT allow With clause to contain a comma only.
-statement error Expected identifier.*
+statement error expected identifier.*
 create table s1 (v1 int) with (,) FORMAT PLAIN ENCODE JSON;
 
 # Do NOT allow an empty With clause.
-statement error Expected identifier.*
+statement error expected identifier.*
 create table s1 (v1 int) with () FORMAT PLAIN ENCODE JSON;

e2e_test/source/basic/ddl.slt

@@ -4,8 +4,9 @@ create source s;
 db error: ERROR: Failed to run the query
 
 Caused by:
-  sql parser error: Expected description of the format, found: ; at line:1, column:17
-Near "create source s"
+  sql parser error: expected description of the format, found: ; at line 1, column 16
+LINE 1: create source s;
+                       ^
 
 
 statement error missing WITH clause

e2e_test/source/basic/old_row_format_syntax/datagen.slt

@@ -182,9 +182,9 @@ statement ok
 drop table s1;
 
 # Do NOT allow With clause to contain a comma only.
-statement error Expected identifier.*
+statement error expected identifier.*
 create table s1 (v1 int) with (,) ROW FORMAT JSON;
 
 # Do NOT allow an empty With clause.
-statement error Expected identifier.*
+statement error expected identifier.*
 create table s1 (v1 int) with () ROW FORMAT JSON;

e2e_test/source/cdc/cdc.share_stream.slt

@@ -27,10 +27,9 @@ statement error Should not create MATERIALIZED VIEW or SELECT directly on shared
 create materialized view mv as select * from mysql_mytest;
 
 statement error The upstream table name must contain database name prefix*
-create table products_test ( id INT,
+create table products_test ( id INT PRIMARY KEY,
    name STRING,
-   description STRING,
-   PRIMARY KEY (id)
+   description STRING
 ) from mysql_mytest table 'products';
 
 statement ok
@@ -233,12 +232,11 @@ CREATE TABLE IF NOT EXISTS postgres_all_types(
 
 statement error The upstream table name must contain schema name prefix*
 CREATE TABLE person_new (
-    id int,
+    id int PRIMARY KEY,
     name varchar,
     email_address varchar,
     credit_card varchar,
-    city varchar,
-    PRIMARY KEY (id)
+    city varchar
 ) FROM pg_source TABLE 'person';
 
 statement ok

e2e_test/udf/sql_udf.slt

@@ -360,7 +360,7 @@ In SQL UDF definition: `select a + b + c + not_be_displayed(c)`
                                        ^
 
 
-statement error Expected end of statement, found: 💩
+statement error expected end of statement, found: 💩
 create function call_regexp_replace() returns varchar language sql as 'select regexp_replace('💩💩💩💩💩foo🤔️bar亲爱的😭baz这不是爱情❤️‍🔥', 'baz(...)', '这是🥵', 'ic')';
 
 # Recursive definition can NOT be accepted at present due to semantic check
@@ -401,7 +401,7 @@ statement error return type mismatch detected
 create function type_mismatch(INT) returns varchar language sql as 'select $1 + 114514 + $1';
 
 # Invalid function body syntax
-statement error Expected an expression:, found: EOF at the end
+statement error expected an expression:, found: EOF at the end
 create function add_error(INT, INT) returns int language sql as $$select $1 + $2 +$$;
 
 ######################################################################

risedev.yml

@@ -950,7 +950,7 @@ profile:
       - use: frontend
       - use: compactor
 
-  ci-1cn-1fe-kafka-with-recovery:
+  ci-1cn-1fe-user-kafka-with-recovery:
     config-path: src/config/ci-recovery.toml
     steps:
       - use: minio
@@ -962,7 +962,9 @@ profile:
       - use: frontend
       - use: compactor
       - use: kafka
-        persist-data: true
+        user-managed: true
+        address: message_queue
+        port: 29092
 
   ci-meta-backup-test-etcd:
     config-path: src/config/ci-meta-backup-test.toml

src/batch/Cargo.toml

@@ -20,6 +20,7 @@ arrow-schema = { workspace = true }
 assert_matches = "1"
 async-recursion = "1"
 async-trait = "0.1"
+bytes = "1"
 either = "1"
 foyer = { workspace = true }
 futures = { version = "0.3", default-features = false, features = ["alloc"] }
@@ -30,9 +31,11 @@ hytra = "0.1.2"
 icelake = { workspace = true }
 itertools = { workspace = true }
 memcomparable = "0.2"
+opendal = "0.45.1"
 parking_lot = { workspace = true }
 paste = "1"
 prometheus = { version = "0.13", features = ["process"] }
+prost = "0.12"
 rand = { workspace = true }
 risingwave_common = { workspace = true }
 risingwave_common_estimate_size = { workspace = true }
@@ -62,6 +65,8 @@ tokio-stream = "0.1"
 tokio-util = { workspace = true }
 tonic = { workspace = true }
 tracing = "0.1"
+twox-hash = "1"
+uuid = { version = "1", features = ["v4"] }
 
 [target.'cfg(not(madsim))'.dependencies]
 workspace-hack = { path = "../workspace-hack" }

src/batch/benches/hash_agg.rs

@@ -13,6 +13,8 @@
 // limitations under the License.
 pub mod utils;
 
+use std::sync::Arc;
+
 use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion};
 use itertools::Itertools;
 use risingwave_batch::executor::aggregation::build as build_agg;
@@ -96,14 +98,15 @@ fn create_hash_agg_executor(
     let schema = Schema { fields };
 
     Box::new(HashAggExecutor::<hash::Key64>::new(
-        agg_init_states,
+        Arc::new(agg_init_states),
         group_key_columns,
         group_key_types,
         schema,
         input,
         "HashAggExecutor".to_string(),
         CHUNK_SIZE,
         MemoryContext::none(),
+        false,
         ShutdownToken::empty(),
     ))
 }

src/batch/src/error.rs

@@ -139,6 +139,13 @@ pub enum BatchError {
 
     #[error("Not enough memory to run this query, batch memory limit is {0} bytes")]
     OutOfMemory(u64),
+
+    #[error("Failed to spill out to disk")]
+    Spill(
+        #[from]
+        #[backtrace]
+        opendal::Error,
+    ),
 }
 
 // Serialize/deserialize error.