refactor(error): eliminate most RwError usages in common crate

src/batch/src/error.rs

@@ -94,6 +94,13 @@ pub enum BatchError {
         BoxedError,
     ),
 
+    #[error("Failed to read from system table: {0}")]
+    SystemTable(
+        #[from]
+        #[backtrace]
+        BoxedError,
+    ),
+
     // Make the ref-counted type to be a variant for easier code structuring.
     #[error(transparent)]
     Shared(

src/batch/src/executor/sys_row_seq_scan.rs

@@ -107,7 +107,11 @@ impl Executor for SysRowSeqScanExecutor {
 impl SysRowSeqScanExecutor {
     #[try_stream(boxed, ok = DataChunk, error = BatchError)]
     async fn do_executor(self: Box<Self>) {
-        let rows = self.sys_catalog_reader.read_table(&self.table_id).await?;
+        let rows = self
+            .sys_catalog_reader
+            .read_table(&self.table_id)
+            .await
+            .map_err(BatchError::SystemTable)?;
         let filtered_rows = rows
             .iter()
             .map(|row| {

src/common/Cargo.toml

@@ -89,6 +89,7 @@ strum = "0.25"
 strum_macros = "0.25"
 sysinfo = { version = "0.29", default-features = false }
 thiserror = "1"
+thiserror-ext = { workspace = true }
 tinyvec = { version = "1", features = ["rustc_1_55", "grab_spare_slice"] }
 tokio = { version = "0.2", package = "madsim-tokio", features = [
     "rt",

src/common/common_service/Cargo.toml

@@ -22,6 +22,8 @@ prometheus = { version = "0.13" }
 risingwave_common = { workspace = true }
 risingwave_pb = { workspace = true }
 risingwave_rpc_client = { workspace = true }
+thiserror = "1"
+thiserror-ext = { workspace = true }
 tokio = { version = "0.2", package = "madsim-tokio", features = ["rt", "rt-multi-thread", "sync", "macros", "time", "signal"] }
 tonic = { workspace = true }
 tower = { version = "0.4", features = ["util", "load-shed"] }

src/common/common_service/src/lib.rs

@@ -16,6 +16,7 @@
 
 #![feature(lint_reasons)]
 #![feature(impl_trait_in_assoc_type)]
+#![feature(error_generic_member_access)]
 
 pub mod metrics_manager;
 pub mod observer_manager;

src/common/common_service/src/observer_manager.rs

@@ -14,8 +14,6 @@
 
 use std::time::Duration;
 
-use risingwave_common::bail;
-use risingwave_common::error::Result;
 use risingwave_pb::meta::subscribe_response::Info;
 use risingwave_pb::meta::{SubscribeResponse, SubscribeType};
 use risingwave_rpc_client::error::RpcError;
@@ -80,6 +78,26 @@ impl<S: ObserverState> ObserverManager<RpcNotificationClient, S> {
     }
 }
 
+/// Error type for [`ObserverManager`].
+#[derive(thiserror::Error, Debug)]
+pub enum ObserverError {
+    #[error("notification channel closed")]
+    ChannelClosed,
+
+    #[error(transparent)]
+    Rpc(
+        #[from]
+        #[backtrace]
+        RpcError,
+    ),
+}
+
+impl From<tonic::Status> for ObserverError {
+    fn from(value: tonic::Status) -> Self {
+        Self::Rpc(value.into())
+    }
+}
+
 impl<T, S> ObserverManager<T, S>
 where
     T: NotificationClient,
@@ -97,24 +115,19 @@ where
         }
     }
 
-    async fn wait_init_notification(&mut self) -> Result<()> {
+    async fn wait_init_notification(&mut self) -> Result<(), ObserverError> {
         let mut notification_vec = Vec::new();
         let init_notification = loop {
             // notification before init notification must be received successfully.
-            match self.rx.message().await {
-                Ok(Some(notification)) => {
+            match self.rx.message().await? {
+                Some(notification) => {
                     if !matches!(notification.info.as_ref().unwrap(), &Info::Snapshot(_)) {
                         notification_vec.push(notification);
                     } else {
                         break notification;
                     }
                 }
-                Ok(None) => {
-                    bail!("notification channel from meta is closed");
-                }
-                Err(err) => {
-                    bail!("receives meta's notification err: {:?}", err);
-                }
+                None => return Err(ObserverError::ChannelClosed),
             }
         };
 
@@ -231,7 +244,10 @@ impl<T: Send + 'static> Channel for Streaming<T> {
 #[async_trait::async_trait]
 pub trait NotificationClient: Send + Sync + 'static {
     type Channel: Channel<Item = SubscribeResponse>;
-    async fn subscribe(&self, subscribe_type: SubscribeType) -> Result<Self::Channel>;
+    async fn subscribe(
+        &self,
+        subscribe_type: SubscribeType,
+    ) -> Result<Self::Channel, ObserverError>;
 }
 
 pub struct RpcNotificationClient {
@@ -248,10 +264,13 @@ impl RpcNotificationClient {
 impl NotificationClient for RpcNotificationClient {
     type Channel = Streaming<SubscribeResponse>;
 
-    async fn subscribe(&self, subscribe_type: SubscribeType) -> Result<Self::Channel> {
+    async fn subscribe(
+        &self,
+        subscribe_type: SubscribeType,
+    ) -> Result<Self::Channel, ObserverError> {
         self.meta_client
             .subscribe(subscribe_type)
             .await
-            .map_err(RpcError::into)
+            .map_err(Into::into)
     }
 }

src/common/heap_profiling/Cargo.toml

@@ -22,6 +22,8 @@ chrono = { version = "0.4", default-features = false, features = [
 ] }
 parking_lot = "0.12"
 risingwave_common = { workspace = true }
+thiserror = "1"
+thiserror-ext = { workspace = true }
 tikv-jemalloc-ctl = { workspace = true }
 tokio = { version = "0.2", package = "madsim-tokio" }
 tracing = "0.1"

src/common/heap_profiling/src/jeprof.rs

@@ -14,12 +14,25 @@
 
 use std::path::Path;
 use std::process::Command;
+use std::result::Result;
 use std::{env, fs};
 
-use anyhow::anyhow;
-use risingwave_common::error::Result;
+/// Error type for running `jeprof`.
+#[derive(thiserror::Error, Debug, thiserror_ext::ContextInto)]
+pub enum JeprofError {
+    #[error(transparent)]
+    IoError(#[from] std::io::Error),
 
-pub async fn run(profile_path: String, collapsed_path: String) -> Result<()> {
+    #[error("jeprof exit with an error (stdout: {stdout}, stderr: {stderr}): {inner}")]
+    ExitError {
+        #[source]
+        inner: std::process::ExitStatusError,
+        stdout: String,
+        stderr: String,
+    },
+}
+
+pub async fn run(profile_path: String, collapsed_path: String) -> Result<(), JeprofError> {
     let executable_path = env::current_exe()?;
 
     let prof_cmd = move || {
@@ -29,20 +42,15 @@ pub async fn run(profile_path: String, collapsed_path: String) -> Result<()> {
             .arg(Path::new(&profile_path))
             .output()
     };
-    match tokio::task::spawn_blocking(prof_cmd).await.unwrap() {
-        Ok(output) => {
-            if output.status.success() {
-                fs::write(Path::new(&collapsed_path), &output.stdout)?;
-                Ok(())
-            } else {
-                Err(anyhow!(
-                    "jeprof exit with an error. stdout: {}, stderr: {}",
-                    String::from_utf8_lossy(&output.stdout),
-                    String::from_utf8_lossy(&output.stderr)
-                )
-                .into())
-            }
-        }
-        Err(e) => Err(e.into()),
-    }
+
+    let output = tokio::task::spawn_blocking(prof_cmd).await.unwrap()?;
+
+    output.status.exit_ok().into_exit_error(
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr),
+    )?;
+
+    fs::write(Path::new(&collapsed_path), &output.stdout)?;
+
+    Ok(())
 }

src/common/heap_profiling/src/lib.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![feature(exit_status_error)]
+
 pub const MANUALLY_DUMP_SUFFIX: &str = "manual.heap";
 pub const AUTO_DUMP_SUFFIX: &str = "auto.heap";
 pub const COLLAPSED_SUFFIX: &str = "collapsed";

src/common/src/array/proto_reader.rs

@@ -256,13 +256,12 @@ mod tests {
         DecimalArray, DecimalArrayBuilder, I32Array, I32ArrayBuilder, TimeArray, TimeArrayBuilder,
         TimestampArray, TimestampArrayBuilder, Utf8Array, Utf8ArrayBuilder,
     };
-    use crate::error::Result;
     use crate::types::{Date, Decimal, Time, Timestamp};
 
     // Convert a column to protobuf, then convert it back to column, and ensures the two are
     // identical.
     #[test]
-    fn test_column_protobuf_conversion() -> Result<()> {
+    fn test_column_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = I32ArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -283,11 +282,10 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 
     #[test]
-    fn test_bool_column_protobuf_conversion() -> Result<()> {
+    fn test_bool_column_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = BoolArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -306,11 +304,10 @@ mod tests {
             1 => assert_eq!(Some(true), x),
             _ => assert_eq!(None, x),
         });
-        Ok(())
     }
 
     #[test]
-    fn test_utf8_column_conversion() -> Result<()> {
+    fn test_utf8_column_conversion() {
         let cardinality = 2048;
         let mut builder = Utf8ArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -330,11 +327,10 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 
     #[test]
-    fn test_decimal_protobuf_conversion() -> Result<()> {
+    fn test_decimal_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = DecimalArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -355,11 +351,10 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 
     #[test]
-    fn test_date_protobuf_conversion() -> Result<()> {
+    fn test_date_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = DateArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -380,11 +375,10 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 
     #[test]
-    fn test_time_protobuf_conversion() -> Result<()> {
+    fn test_time_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = TimeArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -410,11 +404,10 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 
     #[test]
-    fn test_timestamp_protobuf_conversion() -> Result<()> {
+    fn test_timestamp_protobuf_conversion() {
         let cardinality = 2048;
         let mut builder = TimestampArrayBuilder::new(cardinality);
         for i in 0..cardinality {
@@ -440,6 +433,5 @@ mod tests {
                 assert!(x.is_none());
             }
         });
-        Ok(())
     }
 }

src/common/src/catalog/column.rs

@@ -21,7 +21,6 @@ use risingwave_pb::plan_common::{PbColumnCatalog, PbColumnDesc};
 
 use super::row_id_column_desc;
 use crate::catalog::{cdc_table_name_column_desc, offset_column_desc, Field, ROW_ID_COLUMN_ID};
-use crate::error::ErrorCode;
 use crate::types::DataType;
 
 /// Column ID is the unique identifier of a column in a table. Different from table ID, column ID is
@@ -161,24 +160,6 @@ impl ColumnDesc {
         descs
     }
 
-    /// Find `column_desc` in `field_descs` by name.
-    pub fn field(&self, name: &String) -> crate::error::Result<(ColumnDesc, i32)> {
-        if let DataType::Struct { .. } = self.data_type {
-            for (index, col) in self.field_descs.iter().enumerate() {
-                if col.name == *name {
-                    return Ok((col.clone(), index as i32));
-                }
-            }
-            Err(ErrorCode::ItemNotFound(format!("Invalid field name: {}", name)).into())
-        } else {
-            Err(ErrorCode::ItemNotFound(format!(
-                "Cannot get field from non nested column: {}",
-                self.name
-            ))
-            .into())
-        }
-    }
-
     pub fn new_atomic(data_type: DataType, name: &str, column_id: i32) -> Self {
         Self {
             data_type,