refactor(error): clean-up direct error formatting (part 2)

src/common/src/error.rs

@@ -154,6 +154,7 @@ pub enum ErrorCode {
         BoxedError,
     ),
     // TODO: use a new type for bind error
+    // TODO: error-handling
     #[error("Bind error: {0}")]
     BindError(String),
     // TODO: only keep this one

src/connector/src/lib.rs

@@ -32,6 +32,9 @@
 #![feature(iterator_try_collect)]
 #![feature(try_blocks)]
 #![feature(error_generic_member_access)]
+#![feature(register_tool)]
+#![register_tool(rw)]
+#![allow(rw::format_error)] // TODO: remove this
 
 use std::time::Duration;
 

src/ctl/Cargo.toml

@@ -38,6 +38,7 @@ serde = "1"
 serde_json = "1"
 serde_yaml = "0.9.25"
 size = "0.4"
+thiserror-ext = { workspace = true }
 tokio = { version = "0.2", package = "madsim-tokio", features = [
     "rt",
     "rt-multi-thread",

src/ctl/src/cmd_impl/meta/reschedule.rs

@@ -25,6 +25,7 @@ use risingwave_pb::meta::table_fragments::ActorStatus;
 use risingwave_pb::meta::{GetClusterInfoResponse, GetReschedulePlanResponse, Reschedule};
 use serde::{Deserialize, Serialize};
 use serde_yaml;
+use thiserror_ext::AsReport;
 
 use crate::CtlContext;
 
@@ -250,7 +251,7 @@ pub async fn unregister_workers(
     } = match meta_client.get_cluster_info().await {
         Ok(info) => info,
         Err(e) => {
-            println!("Failed to get cluster info: {}", e);
+            println!("Failed to get cluster info: {}", e.as_report());
             exit(1);
         }
     };
@@ -359,7 +360,7 @@ pub async fn unregister_workers(
 
         println!("Unregistering worker #{}, address: {:?}", id, host);
         if let Err(e) = meta_client.delete_worker_node(host).await {
-            println!("Failed to delete worker #{}: {}", id, e);
+            println!("Failed to delete worker #{}: {}", id, e.as_report());
         };
     }
 

src/ctl/src/cmd_impl/profile.rs

@@ -19,6 +19,7 @@ use futures::future::try_join_all;
 use risingwave_pb::common::WorkerType;
 use risingwave_pb::monitor_service::ProfilingResponse;
 use risingwave_rpc_client::ComputeClientPool;
+use thiserror_ext::AsReport;
 use tokio::fs::{create_dir_all, File};
 use tokio::io::AsyncWriteExt;
 
@@ -64,9 +65,9 @@ pub async fn cpu_profile(context: &CtlContext, sleep_s: u64) -> anyhow::Result<(
                 }
                 Err(err) => {
                     tracing::error!(
-                        "Failed to get profiling result from {} with error {}",
-                        node_name,
-                        err.to_string()
+                        error = %err.as_report(),
+                        %node_name,
+                        "Failed to get profiling result",
                     );
                 }
             }
@@ -113,9 +114,9 @@ pub async fn heap_profile(context: &CtlContext, dir: Option<String>) -> anyhow::
 
             if let Err(err) = response {
                 tracing::error!(
-                    "Failed to dump profile on {} with error {}",
-                    node_name,
-                    err.to_string()
+                    error = %err.as_report(),
+                    %node_name,
+                    "Failed to dump profile",
                 );
             }
             Ok::<_, anyhow::Error>(())

src/ctl/src/cmd_impl/scale/resize.rs

@@ -25,6 +25,7 @@ use risingwave_pb::meta::update_worker_node_schedulability_request::Schedulabili
 use risingwave_pb::meta::{GetClusterInfoResponse, GetReschedulePlanResponse};
 use risingwave_stream::task::FragmentId;
 use serde_yaml;
+use thiserror_ext::AsReport;
 
 use crate::cmd_impl::meta::ReschedulePayload;
 use crate::common::CtlContext;
@@ -119,7 +120,7 @@ pub async fn resize(ctl_ctx: &CtlContext, scale_ctx: ScaleCommandContext) -> any
     } = match meta_client.get_cluster_info().await {
         Ok(resp) => resp,
         Err(e) => {
-            fail!("Failed to fetch cluster info: {}", e);
+            fail!("Failed to fetch cluster info: {}", e.as_report());
         }
     };
 
@@ -300,7 +301,7 @@ pub async fn resize(ctl_ctx: &CtlContext, scale_ctx: ScaleCommandContext) -> any
     } = match response {
         Ok(response) => response,
         Err(e) => {
-            fail!("Failed to generate plan: {:?}", e);
+            fail!("Failed to generate plan: {:?}", e.as_report());
         }
     };
 
@@ -364,7 +365,7 @@ pub async fn resize(ctl_ctx: &CtlContext, scale_ctx: ScaleCommandContext) -> any
             match meta_client.reschedule(reschedules, revision, false).await {
                 Ok(response) => response,
                 Err(e) => {
-                    fail!("Failed to execute plan: {:?}", e);
+                    fail!("Failed to execute plan: {:?}", e.as_report());
                 }
             };
 
@@ -391,7 +392,7 @@ pub async fn update_schedulability(
     let GetClusterInfoResponse { worker_nodes, .. } = match meta_client.get_cluster_info().await {
         Ok(resp) => resp,
         Err(e) => {
-            fail!("Failed to get cluster info: {:?}", e);
+            fail!("Failed to get cluster info: {:?}", e.as_report());
         }
     };
 

src/ctl/src/common/context.rs

@@ -15,6 +15,7 @@
 use risingwave_rpc_client::MetaClient;
 use risingwave_storage::hummock::HummockStorage;
 use risingwave_storage::monitor::MonitoredStateStore;
+use thiserror_ext::AsReport;
 use tokio::sync::OnceCell;
 
 use crate::common::hummock_service::{HummockServiceOpts, Metrics};
@@ -68,9 +69,9 @@ impl CtlContext {
                 .await
             {
                 tracing::warn!(
-                    "failed to unregister ctl worker {}: {}",
-                    meta_client.worker_id(),
-                    e
+                    error = %e.as_report(),
+                    worker_id = %meta_client.worker_id(),
+                    "failed to unregister ctl worker",
                 );
             }
         }

src/expr/core/src/error.rs

@@ -63,9 +63,11 @@ pub enum ExprError {
     DivisionByZero,
 
     #[error("Parse error: {0}")]
+    // TODO: error-handling
     Parse(Box<str>),
 
     #[error("Invalid parameter {name}: {reason}")]
+    // TODO: error-handling
     InvalidParam {
         name: &'static str,
         reason: Box<str>,

src/expr/core/src/expr/expr_field.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use anyhow::anyhow;
+use anyhow::{anyhow, Context};
 use risingwave_common::array::{ArrayImpl, ArrayRef, DataChunk};
 use risingwave_common::row::OwnedRow;
 use risingwave_common::types::{DataType, Datum, ScalarImpl};
@@ -90,7 +90,7 @@ impl Build for FieldExpression {
             bail!("Expected Constant as 1st argument");
         };
         let index = Datum::from_protobuf(value, &DataType::Int32)
-            .map_err(|e| anyhow!("Failed to deserialize i32, reason: {:?}", e))?
+            .context("Failed to deserialize i32")?
             .unwrap()
             .as_int32()
             .to_owned();

src/expr/core/src/expr/expr_udf.rs

@@ -25,6 +25,7 @@ use risingwave_common::row::OwnedRow;
 use risingwave_common::types::{DataType, Datum};
 use risingwave_pb::expr::ExprNode;
 use risingwave_udf::ArrowFlightUdfClient;
+use thiserror_ext::AsReport;
 
 use super::{BoxedExpression, Build};
 use crate::expr::Expression;
@@ -182,7 +183,7 @@ impl Build for UdfExpression {
                     Ok(Field::new(
                         "",
                         DataType::from(t).try_into().map_err(|e: ArrayError| {
-                            risingwave_udf::Error::unsupported(e.to_string())
+                            risingwave_udf::Error::unsupported(e.to_report_string())
                         })?,
                         true,
                     ))

src/expr/core/src/expr/wrapper/non_strict.rs

@@ -17,6 +17,7 @@ use auto_impl::auto_impl;
 use risingwave_common::array::{ArrayRef, DataChunk};
 use risingwave_common::row::{OwnedRow, Row};
 use risingwave_common::types::{DataType, Datum};
+use thiserror_ext::AsReport;
 
 use crate::error::Result;
 use crate::expr::{Expression, ValueImpl};
@@ -48,7 +49,7 @@ pub struct LogReport;
 
 impl EvalErrorReport for LogReport {
     fn report(&self, error: ExprError) {
-        tracing::error!(%error, "failed to evaluate expression");
+        tracing::error!(error=%error.as_report(), "failed to evaluate expression");
     }
 }
 

src/expr/core/src/table_function/user_defined.rs

@@ -21,6 +21,7 @@ use futures_util::stream;
 use risingwave_common::array::{ArrayError, DataChunk, I32Array};
 use risingwave_common::bail;
 use risingwave_udf::ArrowFlightUdfClient;
+use thiserror_ext::AsReport;
 
 use super::*;
 
@@ -139,7 +140,7 @@ pub fn new_user_defined(prost: &PbTableFunction, chunk_size: usize) -> Result<Bo
                 Ok(Field::new(
                     "",
                     DataType::from(t).try_into().map_err(|e: ArrayError| {
-                        risingwave_udf::Error::unsupported(e.to_string())
+                        risingwave_udf::Error::unsupported(e.to_report_string())
                     })?,
                     true,
                 ))

src/expr/impl/Cargo.toml

@@ -44,6 +44,7 @@ sha1 = "0.10"
 sha2 = "0.10"
 sql-json-path = { version = "0.1", features = ["jsonbb"] }
 thiserror = "1"
+thiserror-ext = { workspace = true }
 tokio = { version = "0.2", package = "madsim-tokio", features = ["time"] }
 tracing = "0.1"
 

src/expr/impl/benches/expr.rs

@@ -31,6 +31,7 @@ use risingwave_expr::expr::*;
 use risingwave_expr::sig::FUNCTION_REGISTRY;
 use risingwave_expr::ExprError;
 use risingwave_pb::expr::expr_node::PbType;
+use thiserror_ext::AsReport;
 
 criterion_group!(benches, bench_expr, bench_raw);
 criterion_main!(benches);
@@ -398,7 +399,7 @@ fn bench_expr(c: &mut Criterion) {
         }) {
             Ok(agg) => agg,
             Err(e) => {
-                println!("error: {e}");
+                println!("error: {}", e.as_report());
                 continue;
             }
         };

src/expr/impl/src/scalar/cast.rs

@@ -28,6 +28,7 @@ use risingwave_expr::expr::{
 };
 use risingwave_expr::{function, ExprError, Result};
 use risingwave_pb::expr::expr_node::PbType;
+use thiserror_ext::AsReport;
 
 #[function("cast(varchar) -> *int")]
 #[function("cast(varchar) -> decimal")]
@@ -52,7 +53,7 @@ where
 #[function("pgwire_recv(bytea) -> int8")]
 pub fn pgwire_recv(elem: &[u8]) -> Result<i64> {
     let fixed_length =
-        <[u8; 8]>::try_from(elem).map_err(|e| ExprError::Parse(e.to_string().into()))?;
+        <[u8; 8]>::try_from(elem).map_err(|e| ExprError::Parse(e.to_report_string().into()))?;
     Ok(i64::from_be_bytes(fixed_length))
 }
 

src/expr/impl/src/scalar/encdec.rs

@@ -16,6 +16,7 @@ use std::fmt::Write;
 
 use risingwave_common::cast::{parse_bytes_hex, parse_bytes_traditional};
 use risingwave_expr::{function, ExprError, Result};
+use thiserror_ext::AsReport;
 
 const PARSE_BASE64_INVALID_END: &str = "invalid base64 end sequence";
 const PARSE_BASE64_INVALID_PADDING: &str = "unexpected \"=\" while decoding base64 sequence";
@@ -95,7 +96,7 @@ pub fn convert_from(data: &[u8], src_encoding: &str, writer: &mut impl Write) ->
         CharacterSet::Utf8 => {
             let text = String::from_utf8(data.to_vec()).map_err(|e| ExprError::InvalidParam {
                 name: "data",
-                reason: e.to_string().into(),
+                reason: e.to_report_string().into(),
             })?;
             writer.write_str(&text).unwrap();
             Ok(())

src/expr/impl/src/scalar/format.rs

@@ -18,13 +18,14 @@ use std::str::FromStr;
 use risingwave_common::row::Row;
 use risingwave_common::types::{ScalarRefImpl, ToText};
 use risingwave_expr::{function, ExprError, Result};
+use thiserror_ext::AsReport;
 
 use super::string::quote_ident;
 
 /// Formats arguments according to a format string.
 #[function(
     "format(varchar, ...) -> varchar",
-    prebuild = "Formatter::from_str($0).map_err(|e| ExprError::Parse(e.to_string().into()))?"
+    prebuild = "Formatter::from_str($0).map_err(|e| ExprError::Parse(e.to_report_string().into()))?"
 )]
 fn format(formatter: &Formatter, row: impl Row, writer: &mut impl Write) -> Result<()> {
     let mut args = row.iter();

src/expr/impl/src/scalar/int256.rs

@@ -15,7 +15,7 @@
 use risingwave_common::types::Int256;
 use risingwave_expr::ExprError::Parse;
 use risingwave_expr::{function, Result};
-
+use thiserror_ext::AsReport;
 const MAX_AVAILABLE_HEX_STR_LEN: usize = 66;
 
 /// Returns the integer value of the hexadecimal string.
@@ -33,13 +33,13 @@ pub fn hex_to_int256(s: &str) -> Result<Int256> {
     Int256::from_str_hex(s).map_err(|e| {
         Parse(
             if s.len() <= MAX_AVAILABLE_HEX_STR_LEN {
-                format!("failed to parse hex '{}', {}", s, e)
+                format!("failed to parse hex '{}', {}", s, e.as_report())
             } else {
                 format!(
                     "failed to parse hex '{}...'(truncated, total {} bytes), {}",
                     &s[..MAX_AVAILABLE_HEX_STR_LEN],
                     s.len(),
-                    e
+                    e.as_report()
                 )
             }
             .into(),

src/expr/impl/src/scalar/jsonb_path.rs

@@ -16,6 +16,7 @@ use jsonbb::ValueRef;
 use risingwave_common::types::{JsonbRef, JsonbVal};
 use risingwave_expr::{function, ExprError, Result};
 use sql_json_path::{EvalError, JsonPath, ParseError};
+use thiserror_ext::AsReport;
 
 #[function(
     "jsonb_path_exists(jsonb, varchar) -> boolean",
@@ -313,12 +314,12 @@ fn jsonb_path_query_first4(
 }
 
 fn parse_error(e: ParseError) -> ExprError {
-    ExprError::Parse(e.to_string().into())
+    ExprError::Parse(e.to_report_string().into())
 }
 
 fn eval_error(e: EvalError) -> ExprError {
     ExprError::InvalidParam {
         name: "jsonpath",
-        reason: e.to_string().into(),
+        reason: e.to_report_string().into(),
     }
 }