Merge branch 'main' into yiming/merge-sink-exec-write-logic

.github/workflows/doc.yml

@@ -4,11 +4,12 @@ on:
   push:
     branches:
       - main
+  workflow_dispatch:
 
 env:
   SCCACHE_GHA_ENABLED: "true"
   RUSTC_WRAPPER: "sccache"
-  RUSTDOCFLAGS: "--cfg docsrs --markdown-css rust.css --markdown-no-toc --index-page /home/runner/work/risingwave/risingwave/docs/rustdoc/index.md -Zunstable-options"
+  RUSTDOCFLAGS: "--markdown-css rust.css --markdown-no-toc --index-page /home/runner/work/risingwave/risingwave/docs/rustdoc/index.md -Zunstable-options"
 
 jobs:
   build:

Cargo.toml

@@ -11,6 +11,7 @@ members = [
   "src/common/heap_profiling",
   "src/compute",
   "src/connector",
+  "src/connector/with_options",
   "src/ctl",
   "src/error",
   "src/expr/core",
@@ -49,12 +50,12 @@ members = [
   "src/tests/simulation",
   "src/tests/sqlsmith",
   "src/tests/state_cleaning_test",
+  "src/utils/delta_btree_map",
   "src/utils/local_stats_alloc",
   "src/utils/pgwire",
   "src/utils/runtime",
   "src/utils/sync-point",
   "src/utils/variables",
-  "src/utils/with_options",
   "src/utils/workspace-config",
   "src/workspace-hack",
 ]
@@ -136,7 +137,9 @@ arrow-cast-deltalake = { package = "arrow-cast", version = "48.0.1" }
 arrow-schema-deltalake = { package = "arrow-schema", version = "48.0.1" }
 # Use a forked version which removes the dependencies on dynamo db to reduce
 # compile time and binary size.
-deltalake = { git = "https://github.com/risingwavelabs/delta-rs", rev = "5c2dccd4640490202ffe98adbd13b09cef8e007b", features = ["s3-no-concurrent-write"] }
+deltalake = { git = "https://github.com/risingwavelabs/delta-rs", rev = "5c2dccd4640490202ffe98adbd13b09cef8e007b", features = [
+  "s3-no-concurrent-write",
+] }
 parquet = "49"
 thiserror-ext = "0.0.10"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }
@@ -204,8 +207,14 @@ explicit_iter_loop = "warn"
 inconsistent_struct_constructor = "warn"
 unused_async = "warn"
 map_flatten = "warn"
-no_effect_underscore_binding = "warn"
 await_holding_lock = "warn"
+# For private code, this is a little pedantic and not worth fixing.
+# &Vec or &String is acceptable
+ptr_arg = "allow"
+# a little pedantic
+get_first = "allow"
+# https://github.com/rust-lang/rust-clippy/issues/12016
+blocks_in_conditions = "allow"
 
 [workspace.lints.rustdoc]
 private_intra_doc_links = "allow"

ci/build-ci-image.sh

@@ -9,7 +9,8 @@ cd "$DIR"
 cat ../rust-toolchain
 # shellcheck disable=SC2155
 
-export BUILD_ENV_VERSION=v20231109
+# REMEMBER TO ALSO UPDATE ci/docker-compose.yml
+export BUILD_ENV_VERSION=v20231226
 
 export BUILD_TAG="public.ecr.aws/x5u3w5h6/rw-build-env:${BUILD_ENV_VERSION}"
 

ci/docker-compose.yml

@@ -71,7 +71,7 @@ services:
       retries: 5
 
   source-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231109
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231226
     depends_on:
       - mysql
       - db
@@ -81,7 +81,7 @@ services:
       - ..:/risingwave
 
   sink-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231109
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231226
     depends_on:
       - mysql
       - db
@@ -93,12 +93,12 @@ services:
       - ..:/risingwave
 
   rw-build-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231109
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231226
     volumes:
       - ..:/risingwave
 
   ci-flamegraph-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231109
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231226
     # NOTE(kwannoel): This is used in order to permit
     # syscalls for `nperf` (perf_event_open),
     # so it can do CPU profiling.
@@ -109,7 +109,7 @@ services:
       - ..:/risingwave
 
   regress-test-env:
-    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231109
+    image: public.ecr.aws/x5u3w5h6/rw-build-env:v20231226
     depends_on:
       db:
         condition: service_healthy

ci/rust-toolchain

@@ -1,2 +1,7 @@
+# To update toolchain, do the following:
+# 1. update this file
+# 2. update lints/rust-toolchain, lints/Cargo.toml
+# 3. update ci/build-ci-image.sh and ci/docker-compose.yml to build a new CI image
+
 [toolchain]
-channel = "nightly-2023-10-21"
+channel = "nightly-2023-12-26"

ci/scripts/e2e-source-test.sh

@@ -72,8 +72,9 @@ sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.check.slt'
 
 # kill cluster
 cargo make kill
-echo "cluster killed "
+echo "> cluster killed "
 
+echo "--- mysql & postgres recovery check"
 # insert into mytest database (cdc.share_stream.slt)
 mysql --protocol=tcp -u root mytest -e "INSERT INTO products
        VALUES (default,'RisingWave','Next generation Streaming Database'),
@@ -84,16 +85,19 @@ mysql --protocol=tcp -u root mytest -e "INSERT INTO products
 
 # insert new rows
 mysql --host=mysql --port=3306 -u root -p123456 < ./e2e_test/source/cdc/mysql_cdc_insert.sql
+echo "> inserted new rows into mysql"
+
 psql < ./e2e_test/source/cdc/postgres_cdc_insert.sql
-echo "inserted new rows into mysql and postgres"
+echo "> inserted new rows into postgres"
 
 # start cluster w/o clean-data
-RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+unset RISINGWAVE_CI
+export RUST_LOG="events::stream::message::chunk=trace,risingwave_stream=debug,risingwave_batch=info,risingwave_storage=info" \
 
 cargo make dev ci-1cn-1fe-with-recovery
-echo "wait for cluster recovery finish"
+echo "> wait for cluster recovery finish"
 sleep 20
-echo "check mviews after cluster recovery"
+echo "> check mviews after cluster recovery"
 # check results
 sqllogictest -p 4566 -d dev './e2e_test/source/cdc/cdc.check_new_rows.slt'
 
@@ -104,6 +108,7 @@ echo "--- Kill cluster"
 cargo make ci-kill
 
 echo "--- e2e, ci-1cn-1fe, protobuf schema registry"
+export RISINGWAVE_CI=true
 RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
 cargo make ci-start ci-1cn-1fe
 python3 -m pip install requests protobuf confluent-kafka

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

@@ -35,6 +35,7 @@
     'nats': ['json'],
     'doris-sink': ['json'],
     'starrocks-sink': ['json'],
+    'deltalake-sink': ['json'],
 }
 
 def gen_pipeline_steps():

ci/scripts/notify.py

@@ -58,6 +58,7 @@
     "vector-json": ["tao"],
     "doris-sink": ["xinhao"],
     "starrocks-sink": ["xinhao"],
+    "deltalake-sink": ["xinhao"],
 }
 
 def get_failed_tests(get_test_status, test_map):

e2e_test/batch/catalog/pg_attribute.slt.part

@@ -38,9 +38,7 @@ select i.relname, a.attname, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_attribute a on t.oid = a.attrelid and a.attnum = ANY(ix.indkey)
     where t.relname = 'tmp' order by a.attnum;
 ----
-tmp_idx id2 {2,3,4,5}
-tmp_idx id3 {2,3,4,5}
-tmp_idx id4 {2,3,4,5}
+tmp_idx id2 {2}
 
 statement ok
 drop table tmp;

e2e_test/batch/catalog/pg_index.slt.part

@@ -10,7 +10,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id2_idx';
 ----
-2 {2,3}
+1 {2}
 
 statement ok
 create index tmp_id2_idx_include_id1 on tmp(id2) include(id1);
@@ -21,7 +21,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id2_idx_include_id1';
 ----
-3 {2,3,4}
+1 {2}
 
 statement ok
 create index tmp_id1_id2_idx on tmp(id1, id2);
@@ -32,7 +32,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id1_id2_idx';
 ----
-3 {2,3,4}
+2 {1,2}
 
 statement ok
 drop table tmp;