refactor(error): move RwError from common to frontend

src/batch/clippy.toml

@@ -1,11 +1,6 @@
-disallowed-methods = [
-]
+disallowed-methods = []
 
-disallowed-types = [
-    { path = "risingwave_common::error::ErrorCode", reason = "Please use per-crate error type instead." },
-    { path = "risingwave_common::error::RwError", reason = "Please use per-crate error type instead." },
-    { path = "risingwave_common::error::Result", reason = "Please use per-crate error type instead." },
-]
+disallowed-types = []
 
 doc-valid-idents = [
     "RisingWave",
@@ -16,7 +11,7 @@ doc-valid-idents = [
     "PostgreSQL",
     "MySQL",
     "TopN",
-    "VNode"
+    "VNode",
 ]
 avoid-breaking-exported-api = false
 upper-case-acronyms-aggressive = true

src/batch/src/error.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 
 pub use anyhow::anyhow;
 use risingwave_common::array::ArrayError;
-use risingwave_common::error::{BoxedError, ErrorCode, RwError};
+use risingwave_common::error::BoxedError;
 use risingwave_common::util::value_encoding::error::ValueEncodingError;
 use risingwave_dml::error::DmlError;
 use risingwave_expr::ExprError;
@@ -145,19 +145,6 @@ impl From<tonic::Status> for BatchError {
     }
 }
 
-impl From<BatchError> for RwError {
-    fn from(s: BatchError) -> Self {
-        ErrorCode::BatchError(Box::new(s)).into()
-    }
-}
-
-// TODO(error-handling): remove after eliminating RwError from connector.
-impl From<RwError> for BatchError {
-    fn from(s: RwError) -> Self {
-        Self::Internal(anyhow!(s))
-    }
-}
-
 impl<'a> From<&'a BatchError> for Status {
     fn from(err: &'a BatchError) -> Self {
         err.to_status(tonic::Code::Internal, "batch")

src/bench/Cargo.toml

@@ -8,6 +8,7 @@ license = { workspace = true }
 repository = { workspace = true }
 
 [dependencies]
+anyhow = "1"
 async-trait = "0.1"
 aws-config = { workspace = true }
 aws-sdk-s3 = { workspace = true }

src/bench/s3_bench/main.rs

@@ -28,7 +28,6 @@ use futures::stream::{self, StreamExt};
 use futures::{future, Future, FutureExt};
 use itertools::Itertools;
 use rand::{Rng, SeedableRng};
-use risingwave_common::error::RwError;
 use tokio::join;
 use tokio::sync::RwLock;
 use tracing::debug;
@@ -233,7 +232,7 @@ async fn multi_part_upload(
             let part_t = Instant::now();
             let result = a.send().await.unwrap();
             let part_ttl = part_t.elapsed();
-            Ok::<_, RwError>((result, part_ttl))
+            Ok::<_, anyhow::Error>((result, part_ttl))
         })
         .collect_vec();
     let ttfb = t.elapsed();
@@ -318,7 +317,7 @@ async fn multi_part_get(
         .into_iter()
         .map(create_part_get)
         .map(|resp| async move {
-            let result: Result<(usize, Duration), RwError> = Ok((
+            let result: anyhow::Result<(usize, Duration)> = Ok((
                 resp.await
                     .unwrap()
                     .body
@@ -381,7 +380,7 @@ async fn run_case(
     cfg: Arc<Config>,
     client: Arc<Client>,
     objs: Arc<RwLock<ObjPool>>,
-) -> Result<(), RwError> {
+) -> anyhow::Result<()> {
     let (name, analysis) = match case.clone() {
         Case::Put {
             name,

src/bench/sink_bench/main.rs

@@ -19,6 +19,7 @@
 use core::str::FromStr;
 use std::collections::HashMap;
 
+use anyhow::anyhow;
 use clap::Parser;
 use futures::prelude::future::Either;
 use futures::prelude::stream::{BoxStream, PollNext};
@@ -27,7 +28,6 @@ use futures::{FutureExt, StreamExt, TryStreamExt};
 use futures_async_stream::try_stream;
 use risingwave_common::buffer::Bitmap;
 use risingwave_common::catalog::ColumnId;
-use risingwave_common::error::anyhow_error;
 use risingwave_connector::dispatch_sink;
 use risingwave_connector::parser::{
     EncodingProperties, ParserConfig, ProtocolProperties, SpecificParserConfig,
@@ -82,7 +82,7 @@ impl LogReader for MockRangeLogReader {
                         .take()
                         .unwrap()
                         .send(self.throughput_metric.take().unwrap())
-                        .map_err(|_| anyhow_error!("Can't send throughput_metric"))?;
+                        .map_err(|_| anyhow!("Can't send throughput_metric"))?;
                     futures::future::pending().await
                 },
             item = self.upstreams.next() => {
@@ -108,7 +108,7 @@ impl LogReader for MockRangeLogReader {
                             },
                         ))
                     }
-                    _ => Err(anyhow_error!("Can't assert message type".to_string())),
+                    _ => Err(anyhow!("Can't assert message type".to_string())),
                 }
             }
         }
@@ -390,7 +390,7 @@ fn mock_from_legacy_type(
             SINK_TYPE_APPEND_ONLY => SinkFormat::AppendOnly,
             SINK_TYPE_UPSERT => SinkFormat::Upsert,
             _ => {
-                return Err(SinkError::Config(risingwave_common::array::error::anyhow!(
+                return Err(SinkError::Config(anyhow!(
                     "sink type unsupported: {}",
                     r#type
                 )))