Merge branch 'main' into kanzhen/alter-db-owner-connect

Cargo.toml

@@ -128,7 +128,7 @@ arrow-flight = "49"
 arrow-select = "49"
 arrow-ord = "49"
 arrow-row = "49"
-thiserror-ext = "0.0.7"
+thiserror-ext = "0.0.8"
 tikv-jemalloc-ctl = { git = "https://github.com/risingwavelabs/jemallocator.git", rev = "64a2d9" }
 tikv-jemallocator = { git = "https://github.com/risingwavelabs/jemallocator.git", features = [
   "profiling",

ci/scripts/deterministic-recovery-test.sh

@@ -19,18 +19,17 @@ export LOGDIR=.risingwave/log
 
 mkdir -p $LOGDIR
 
-# FIXME(kwannoel): Why is this failing?
-# echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, background_ddl"
-# seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/background_ddl/sim/basic.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
+echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, background_ddl"
+seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/background_ddl/sim/basic.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, ddl"
-seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/ddl/\*\*/\*.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
+seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} --background-ddl-rate=${BACKGROUND_DDL_RATE} ./e2e_test/ddl/\*\*/\*.slt 2> $LOGDIR/recovery-ddl-{}.log && rm $LOGDIR/recovery-ddl-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, streaming"
-seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/streaming/\*\*/\*.slt 2> $LOGDIR/recovery-streaming-{}.log && rm $LOGDIR/recovery-streaming-{}.log'
+seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} --background-ddl-rate=${BACKGROUND_DDL_RATE} ./e2e_test/streaming/\*\*/\*.slt 2> $LOGDIR/recovery-streaming-{}.log && rm $LOGDIR/recovery-streaming-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, batch"
-seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} ./e2e_test/batch/\*\*/\*.slt 2> $LOGDIR/recovery-batch-{}.log && rm $LOGDIR/recovery-batch-{}.log'
+seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} --background-ddl-rate=${BACKGROUND_DDL_RATE} ./e2e_test/batch/\*\*/\*.slt 2> $LOGDIR/recovery-batch-{}.log && rm $LOGDIR/recovery-batch-{}.log'
 
 echo "--- deterministic simulation e2e, ci-3cn-2fe-3meta, recovery, kafka source,sink"
 seq $TEST_NUM | parallel MADSIM_TEST_SEED={} './risingwave_simulation --kill --kill-rate=${KILL_RATE} --kafka-datadir=./scripts/source/test_data ./e2e_test/source/basic/kafka\*.slt 2> $LOGDIR/recovery-source-{}.log && rm $LOGDIR/recovery-source-{}.log'

ci/workflows/main-cron.yml

@@ -293,7 +293,24 @@ steps:
     retry: *auto-retry
 
   - label: "recovery test (deterministic simulation)"
-    command: "TEST_NUM=12 KILL_RATE=1.0 timeout 55m ci/scripts/deterministic-recovery-test.sh"
+    command: "TEST_NUM=12 KILL_RATE=1.0 BACKGROUND_DDL_RATE=0.0 timeout 55m ci/scripts/deterministic-recovery-test.sh"
+    if: |
+      !(build.pull_request.labels includes "ci/main-cron/skip-ci") && build.env("CI_STEPS") == null
+      || build.pull_request.labels includes "ci/run-recovery-test-deterministic-simulation"
+      || build.env("CI_STEPS") =~ /(^|,)recovery-tests?-deterministic-simulation(,|$$)/
+    depends_on: "build-simulation"
+    plugins:
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 60
+    retry: *auto-retry
+
+  # Ddl statements will randomly run with background_ddl.
+  - label: "background_ddl recovery test (deterministic simulation)"
+    command: "TEST_NUM=12 KILL_RATE=1.0 BACKGROUND_DDL_RATE=0.8 timeout 55m ci/scripts/deterministic-recovery-test.sh"
     if: |
       !(build.pull_request.labels includes "ci/main-cron/skip-ci") && build.env("CI_STEPS") == null
       || build.pull_request.labels includes "ci/run-recovery-test-deterministic-simulation"

ci/workflows/pull-request.yml

@@ -410,7 +410,7 @@ steps:
     retry: *auto-retry
 
   - label: "recovery test (deterministic simulation)"
-    command: "TEST_NUM=8 KILL_RATE=0.5 ci/scripts/deterministic-recovery-test.sh"
+    command: "TEST_NUM=8 KILL_RATE=0.5 BACKGROUND_DDL_RATE=0.0 ci/scripts/deterministic-recovery-test.sh"
     if: |
       !(build.pull_request.labels includes "ci/skip-ci") && build.env("CI_STEPS") == null
       || build.pull_request.labels includes "ci/run-recovery-test-deterministic-simulation"

e2e_test/background_ddl/sim/basic.slt

@@ -13,6 +13,9 @@ INSERT INTO t select * from generate_series(1, 200000);
 statement ok
 FLUSH;
 
+statement ok
+SET STREAMING_RATE_LIMIT=4000;
+
 statement ok
 CREATE MATERIALIZED VIEW m1 as SELECT * FROM t;
 
@@ -26,23 +29,20 @@ CREATE MATERIALIZED VIEW m3 as SELECT * FROM t;
 statement error
 CREATE MATERIALIZED VIEW m3 as SELECT * FROM t;
 
-# Wait for background ddl to finish
-sleep 30s
-
 query I
 select count(*) from m1;
 ----
-10000000
+200000
 
 query I
 select count(*) from m2;
 ----
-10000000
+200000
 
 query I
 select count(*) from m3;
 ----
-10000000
+200000
 
 statement ok
 DROP MATERIALIZED VIEW m1;

e2e_test/batch/basic/null_range_scan.slt.part

@@ -16,6 +16,43 @@ query II rowsort
 SELECT * FROM t0 WHERE (c1 > 1);
 ----
 
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 > 9223372036854775808) IS NULL;
+----
+1 NULL
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 < 9223372036854775808) IS NULL;
+----
+1 NULL
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 >= -9223372036854775808) IS NULL;
+----
+1 NULL
+
+query II rowsort
+SELECT * FROM t0 WHERE (c1 <= -9223372036854775808) IS NULL;
+----
+1 NULL
+
+query II rowsort
+SELECT * FROM t0 WHERE c1 > 9223372036854775808;
+----
+
+query II rowsort
+SELECT * FROM t0 WHERE c1 < 9223372036854775808;
+----
+
+query II rowsort
+SELECT * FROM t0 WHERE c1 >= -9223372036854775808;
+----
+
+query II rowsort
+SELECT * FROM t0 WHERE c1 <= -9223372036854775808;
+----
+
 statement ok
 create materialized view mv as select * from t0 order by c1 desc nulls last;
 

e2e_test/error_ui/simple/main.slt

@@ -58,3 +58,26 @@ db error: ERROR: Failed to run the query
 Caused by these errors (recent errors listed first):
   1: Expr error
   2: Division by zero
+
+
+statement error
+set rw_implicit_flush to maybe;
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Failed to get/set session config
+  2: Invalid value `maybe` for `rw_implicit_flush`
+  3: provided string was not `true` or `false`
+
+
+statement error
+set transaction_isolation_level to read_committed;
+----
+db error: ERROR: Failed to run the query
+
+Caused by these errors (recent errors listed first):
+  1: Failed to get/set session config
+  2: Invalid value `read_committed` for `transaction_isolation_level`
+  3: Feature is not yet implemented: isolation level
+Tracking issue: https://github.com/risingwavelabs/risingwave/issues/10736

src/common/src/error.rs

@@ -23,6 +23,7 @@ use memcomparable::Error as MemComparableError;
 use risingwave_error::tonic::{ToTonicStatus, TonicStatusWrapper};
 use risingwave_pb::PbFieldNotFound;
 use thiserror::Error;
+use thiserror_ext::Macro;
 use tokio::task::JoinError;
 
 use crate::array::ArrayError;
@@ -36,7 +37,7 @@ pub type BoxedError = Box<dyn Error>;
 
 pub use anyhow::anyhow as anyhow_error;
 
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, Default)]
 pub struct TrackingIssue(Option<u32>);
 
 impl TrackingIssue {
@@ -70,6 +71,15 @@ impl Display for TrackingIssue {
     }
 }
 
+#[derive(Error, Debug, Macro)]
+#[error("Feature is not yet implemented: {feature}\n{issue}")]
+#[thiserror_ext(macro(path = "crate::error"))]
+pub struct NotImplemented {
+    #[message]
+    pub feature: String,
+    pub issue: TrackingIssue,
+}
+
 #[derive(Error, Debug)]
 pub enum ErrorCode {
     #[error("internal error: {0}")]
@@ -87,8 +97,8 @@ pub enum ErrorCode {
         #[backtrace]
         BoxedError,
     ),
-    #[error("Feature is not yet implemented: {0}\n{1}")]
-    NotImplemented(String, TrackingIssue),
+    #[error(transparent)]
+    NotImplemented(#[from] NotImplemented),
     // Tips: Use this only if it's intended to reject the query
     #[error("Not supported: {0}\nHINT: {1}")]
     NotSupported(String, String),
@@ -200,6 +210,13 @@ pub enum ErrorCode {
     ),
 }
 
+// TODO(error-handling): automatically generate this impl.
+impl From<NotImplemented> for RwError {
+    fn from(value: NotImplemented) -> Self {
+        ErrorCode::from(value).into()
+    }
+}
+
 pub fn internal_error(msg: impl Into<String>) -> RwError {
     ErrorCode::InternalError(msg.into()).into()
 }
@@ -479,14 +496,8 @@ macro_rules! ensure_eq {
 
 #[macro_export]
 macro_rules! bail {
-    ($msg:literal $(,)?) => {
-        return Err($crate::error::anyhow_error!($msg).into())
-    };
-    ($err:expr $(,)?) => {
-        return Err($crate::error::anyhow_error!($err).into())
-    };
-    ($fmt:expr, $($arg:tt)*) => {
-        return Err($crate::error::anyhow_error!($fmt, $($arg)*).into())
+    ($($arg:tt)*) => {
+        return Err($crate::error::anyhow_error!($($arg)*).into())
     };
 }
 
@@ -616,7 +627,7 @@ mod tests {
         check_grpc_error(ErrorCode::TaskNotFound, Code::Internal);
         check_grpc_error(ErrorCode::InternalError(String::new()), Code::Internal);
         check_grpc_error(
-            ErrorCode::NotImplemented(String::new(), None.into()),
+            ErrorCode::NotImplemented(not_implemented!("test")),
             Code::Internal,
         );
     }

src/common/src/session_config/transaction_isolation_level.rs

@@ -15,6 +15,8 @@
 use std::fmt::Formatter;
 use std::str::FromStr;
 
+use crate::error::{bail_not_implemented, NotImplemented};
+
 #[derive(Copy, Default, Debug, Clone, PartialEq, Eq)]
 // Some variants are never constructed so allow dead code here.
 #[allow(dead_code)]
@@ -27,10 +29,10 @@ pub enum IsolationLevel {
 }
 
 impl FromStr for IsolationLevel {
-    type Err = &'static str;
+    type Err = NotImplemented;
 
     fn from_str(_s: &str) -> Result<Self, Self::Err> {
-        Err("isolation level is not yet supported")
+        bail_not_implemented!(issue = 10736, "isolation level");
     }
 }
 

src/common/src/types/to_binary.rs

@@ -16,16 +16,16 @@ use bytes::{Bytes, BytesMut};
 use postgres_types::{ToSql, Type};
 
 use super::{DataType, DatumRef, ScalarRefImpl, F32, F64};
-use crate::error::TrackingIssue;
+use crate::error::NotImplemented;
 
 /// Error type for [`ToBinary`] trait.
 #[derive(thiserror::Error, Debug)]
 pub enum ToBinaryError {
     #[error(transparent)]
     ToSql(Box<dyn std::error::Error + Send + Sync>),
 
-    #[error("Feature is not yet implemented: {0}\n{1}")]
-    NotImplemented(String, TrackingIssue),
+    #[error(transparent)]
+    NotImplemented(#[from] NotImplemented),
 }
 
 pub type Result<T> = std::result::Result<T, ToBinaryError>;
@@ -87,15 +87,10 @@ impl ToBinary for ScalarRefImpl<'_> {
             ScalarRefImpl::Time(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Bytea(v) => v.to_binary_with_type(ty),
             ScalarRefImpl::Jsonb(v) => v.to_binary_with_type(ty),
-            ScalarRefImpl::Struct(_) | ScalarRefImpl::List(_) => {
-                Err(ToBinaryError::NotImplemented(
-                    format!(
-                        "the pgwire extended-mode encoding for {} is unsupported",
-                        ty
-                    ),
-                    Some(7949).into(),
-                ))
-            }
+            ScalarRefImpl::Struct(_) | ScalarRefImpl::List(_) => bail_not_implemented!(
+                issue = 7949,
+                "the pgwire extended-mode encoding for {ty} is unsupported"
+            ),
         }
     }
 }

src/compute/src/memory/config.rs

@@ -21,7 +21,7 @@ pub const MIN_COMPUTE_MEMORY_MB: usize = 512;
 /// overhead, network buffer, etc.) in megabytes.
 pub const MIN_SYSTEM_RESERVED_MEMORY_MB: usize = 512;
 
-const SYSTEM_RESERVED_MEMORY_PROPORTION: f64 = 0.2;
+const SYSTEM_RESERVED_MEMORY_PROPORTION: f64 = 0.3;
 
 const STORAGE_MEMORY_PROPORTION: f64 = 0.3;
 
@@ -157,14 +157,14 @@ mod tests {
     #[test]
     fn test_reserve_memory_bytes() {
         // at least 512 MB
-        let (reserved, non_reserved) = reserve_memory_bytes(2 << 30);
+        let (reserved, non_reserved) = reserve_memory_bytes(1536 << 20);
         assert_eq!(reserved, 512 << 20);
-        assert_eq!(non_reserved, 1536 << 20);
+        assert_eq!(non_reserved, 1024 << 20);
 
         // reserve based on proportion
         let (reserved, non_reserved) = reserve_memory_bytes(10 << 30);
-        assert_eq!(reserved, 2 << 30);
-        assert_eq!(non_reserved, 8 << 30);
+        assert_eq!(reserved, 3 << 30);
+        assert_eq!(non_reserved, 7 << 30);
     }
 
     #[test]