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

.cargo/config.toml

@@ -34,7 +34,13 @@ rustflags = [
 rustflags = [
     "--cfg",
     "tokio_unstable",
-    # uncomment the following two lines to enable `TaskLocalAlloc`
+
+    # Register the `rw` tool for referencing the custom lints in attributes like `#[allow(rw::xx)]`.
+    # This is essentially the same as `#![..]` in each crate root.
+    "-Zcrate-attr=feature(register_tool)",
+    "-Zcrate-attr=register_tool(rw)",
+
+    # Uncomment the following two lines to enable `TaskLocalAlloc`.
     # "--cfg",
     # "enable_task_local_alloc",
 ]

.github/workflows/doc.yml

@@ -8,7 +8,7 @@ on:
 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: "--cfg docsrs --markdown-css rust.css --markdown-no-toc --index-page /home/runner/work/risingwave/risingwave/docs/rustdoc/index.md -Zunstable-options -Zcrate-attr=feature(register_tool) -Zcrate-attr=register_tool(rw)"
 
 jobs:
   build:

ci/scripts/check.sh

@@ -35,7 +35,7 @@ sccache --show-stats
 sccache --zero-stats
 
 echo "--- Build documentation"
-RUSTDOCFLAGS="-Dwarnings" cargo doc --document-private-items --no-deps
+RUSTDOCFLAGS="-Dwarnings -Zcrate-attr=feature(register_tool) -Zcrate-attr=register_tool(rw)" cargo doc --document-private-items --no-deps
 
 echo "--- Show sccache stats"
 sccache --show-stats

lints/src/format_error.rs

@@ -17,7 +17,7 @@ use clippy_utils::macros::{
     find_format_arg_expr, find_format_args, is_format_macro, macro_backtrace,
 };
 use clippy_utils::ty::implements_trait;
-use clippy_utils::{is_trait_method, match_function_call};
+use clippy_utils::{is_in_cfg_test, is_in_test_function, is_trait_method, match_function_call};
 use rustc_ast::FormatArgsPiece;
 use rustc_hir::{Expr, ExprKind};
 use rustc_lint::{LateContext, LateLintPass};
@@ -63,6 +63,11 @@ const TRACING_FIELD_DISPLAY: [&str; 3] = ["tracing_core", "field", "display"];
 
 impl<'tcx> LateLintPass<'tcx> for FormatError {
     fn check_expr(&mut self, cx: &LateContext<'tcx>, expr: &'tcx Expr<'_>) {
+        // Ignore if in test code.
+        if is_in_cfg_test(cx.tcx, expr.hir_id) || is_in_test_function(cx.tcx, expr.hir_id) {
+            return;
+        }
+
         // `%err`, `?err` in tracing events and spans.
         if let Some(args) = match_function_call(cx, expr, &TRACING_FIELD_DEBUG)
             .or_else(|| match_function_call(cx, expr, &TRACING_FIELD_DISPLAY))

src/batch/src/executor/aggregation/orderby.rs

@@ -14,7 +14,7 @@
 
 use std::ops::Range;
 
-use anyhow::anyhow;
+use anyhow::Context;
 use risingwave_common::array::{Op, RowRef, StreamChunk};
 use risingwave_common::estimate_size::EstimateSize;
 use risingwave_common::row::{OwnedRow, Row, RowExt};
@@ -25,7 +25,7 @@ use risingwave_common::util::sort_util::{ColumnOrder, OrderType};
 use risingwave_expr::aggregate::{
     AggStateDyn, AggregateFunction, AggregateState, BoxedAggregateFunction,
 };
-use risingwave_expr::{ExprError, Result};
+use risingwave_expr::Result;
 
 /// `ProjectionOrderBy` is a wrapper of `AggregateFunction` that sorts rows by given columns and
 /// then projects columns.
@@ -77,7 +77,7 @@ impl ProjectionOrderBy {
     fn push_row(&self, state: &mut State, row: RowRef<'_>) -> Result<()> {
         let key =
             memcmp_encoding::encode_row(row.project(&self.order_col_indices), &self.order_types)
-                .map_err(|e| ExprError::Internal(anyhow!("failed to encode row, error: {}", e)))?;
+                .context("failed to encode row")?;
         let projected_row = row.project(&self.arg_indices).to_owned_row();
 
         state.unordered_values_estimated_heap_size +=

src/batch/src/executor/insert.rs

@@ -15,7 +15,7 @@
 use std::iter::repeat;
 use std::sync::Arc;
 
-use anyhow::anyhow;
+use anyhow::Context;
 use futures_async_stream::try_stream;
 use itertools::Itertools;
 use risingwave_common::array::{
@@ -231,8 +231,8 @@ impl BoxedExecutorBuilder for InsertExecutor {
                 .map(|IndexAndExpr { index: i, expr: e }| {
                     Ok((
                         i as usize,
-                        build_from_prost(&e.ok_or_else(|| anyhow!("expression is None"))?)
-                            .map_err(|e| anyhow!("failed to build expression: {}", e))?,
+                        build_from_prost(&e.context("expression is None")?)
+                            .context("failed to build expression")?,
                     ))
                 })
                 .collect::<Result<Vec<_>>>()?;

src/batch/src/executor/mod.rs

@@ -41,7 +41,7 @@ mod update;
 mod utils;
 mod values;
 
-use anyhow::anyhow;
+use anyhow::Context;
 use async_recursion::async_recursion;
 pub use delete::*;
 pub use expand::*;
@@ -69,6 +69,7 @@ pub use sort_agg::*;
 pub use sort_over_window::SortOverWindowExecutor;
 pub use source::*;
 pub use table_function::*;
+use thiserror_ext::AsReport;
 pub use top_n::TopNExecutor;
 pub use union::*;
 pub use update::*;
@@ -183,12 +184,14 @@ impl<'a, C: Clone> ExecutorBuilder<'a, C> {
 
 impl<'a, C: BatchTaskContext> ExecutorBuilder<'a, C> {
     pub async fn build(&self) -> Result<BoxedExecutor> {
-        self.try_build().await.map_err(|e| {
-            let err_msg = format!("Failed to build executor: {e}");
-            let plan_node_body = self.plan_node.get_node_body();
-            error!("{err_msg}, plan node is: \n {plan_node_body:?}");
-            anyhow!(err_msg).into()
-        })
+        self.try_build()
+            .await
+            .inspect_err(|e| {
+                let plan_node = self.plan_node.get_node_body();
+                error!(error = %e.as_report(), ?plan_node, "failed to build executor");
+            })
+            .context("failed to build executor")
+            .map_err(Into::into)
     }
 
     #[async_recursion]

src/batch/src/rpc/service/task_service.rs

@@ -22,6 +22,7 @@ use risingwave_pb::task_service::{
     CancelTaskRequest, CancelTaskResponse, CreateTaskRequest, ExecuteRequest, GetDataResponse,
     TaskInfoResponse,
 };
+use thiserror_ext::AsReport;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{Request, Response, Status};
 
@@ -93,7 +94,7 @@ impl TaskService for BatchServiceImpl {
                 state_rx,
             ))),
             Err(e) => {
-                error!("failed to fire task {}", e);
+                error!(error = %e.as_report(), "failed to fire task");
                 Err(e.into())
             }
         }
@@ -146,8 +147,9 @@ impl TaskService for BatchServiceImpl {
             .await
         {
             error!(
-                "failed to build executors and trigger execution of Task {:?}: {}",
-                task_id, e
+                error = %e.as_report(),
+                ?task_id,
+                "failed to build executors and trigger execution"
             );
             return Err(e.into());
         }
@@ -158,12 +160,12 @@ impl TaskService for BatchServiceImpl {
             // therefore we would only have one data output.
             output_id: 0,
         };
-        let mut output = task.get_task_output(&pb_task_output_id).map_err(|e| {
+        let mut output = task.get_task_output(&pb_task_output_id).inspect_err(|e| {
             error!(
-                "failed to get task output of Task {:?} in local execution mode",
-                task_id
+                error = %e.as_report(),
+                ?task_id,
+                "failed to get task output in local execution mode",
             );
-            e
         })?;
         let mut writer = GrpcExchangeWriter::new(tx.clone());
         // Always spawn a task and do not block current function.

src/batch/src/task/task_execution.rs

@@ -34,6 +34,7 @@ use risingwave_pb::plan_common::CapturedExecutionContext;
 use risingwave_pb::task_service::task_info_response::TaskStatus;
 use risingwave_pb::task_service::{GetDataResponse, TaskInfoResponse};
 use risingwave_pb::PbFieldNotFound;
+use thiserror_ext::AsReport;
 use tokio::select;
 use tokio::task::JoinHandle;
 use tokio_metrics::TaskMonitor;
@@ -644,7 +645,7 @@ impl<C: BatchTaskContext> BatchTaskExecution<C> {
                             ShutdownMsg::Init => {
                                 // There is no message received from shutdown channel, which means it caused
                                 // task failed.
-                                error!("Batch task failed: {:?}", e);
+                                error!(error = %e.as_report(), "Batch task failed");
                                 error = Some(e);
                                 state = TaskStatus::Failed;
                                 break;
@@ -671,7 +672,7 @@ impl<C: BatchTaskContext> BatchTaskExecution<C> {
 
         let error = error.map(Arc::new);
         *self.failure.lock() = error.clone();
-        let err_str = error.as_ref().map(|e| e.to_string());
+        let err_str = error.as_ref().map(|e| e.to_report_string());
         if let Err(e) = sender.close(error).await {
             match e {
                 SenderError => {
@@ -689,8 +690,8 @@ impl<C: BatchTaskContext> BatchTaskExecution<C> {
 
         if let Err(e) = self.change_state_notify(state, state_tx, err_str).await {
             warn!(
-                "The status receiver in FE has closed so the status push is failed {:}",
-                e
+                error = %e.as_report(),
+                "The status receiver in FE has closed so the status push is failed",
             );
         }
 

src/common/common_service/src/metrics_manager.rs

@@ -19,10 +19,11 @@ use std::sync::OnceLock;
 use hyper::{Body, Request, Response};
 use prometheus::{Encoder, Registry, TextEncoder};
 use risingwave_common::monitor::GLOBAL_METRICS_REGISTRY;
+use thiserror_ext::AsReport;
 use tower::make::Shared;
 use tower::ServiceBuilder;
 use tower_http::add_extension::AddExtensionLayer;
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 
 pub struct MetricsManager {}
 
@@ -46,7 +47,7 @@ impl MetricsManager {
                 let serve_future =
                     hyper::Server::bind(&listen_socket_addr).serve(Shared::new(service));
                 if let Err(err) = serve_future.await {
-                    eprintln!("server error: {}", err);
+                    error!(error = %err.as_report(), "metrics service exited with error");
                 }
             });
             listen_addr_clone

src/common/common_service/src/observer_manager.rs

@@ -18,6 +18,7 @@ use risingwave_pb::meta::subscribe_response::Info;
 use risingwave_pb::meta::{SubscribeResponse, SubscribeType};
 use risingwave_rpc_client::error::RpcError;
 use risingwave_rpc_client::MetaClient;
+use thiserror_ext::AsReport;
 use tokio::task::JoinHandle;
 use tonic::{Status, Streaming};
 
@@ -175,7 +176,7 @@ where
     /// call the `handle_initialization_notification` and `handle_notification` to update node data.
     pub async fn start(mut self) -> JoinHandle<()> {
         if let Err(err) = self.wait_init_notification().await {
-            tracing::warn!("Receives meta's notification err {:?}", err);
+            tracing::warn!(error = %err.as_report(), "Receives meta's notification err");
             self.re_subscribe().await;
         }
 
@@ -190,8 +191,8 @@ where
                         }
                         self.observer_states.handle_notification(resp.unwrap());
                     }
-                    Err(e) => {
-                        tracing::error!("Receives meta's notification err {:?}", e);
+                    Err(err) => {
+                        tracing::warn!(error = %err.as_report(), "Receives meta's notification err");
                         self.re_subscribe().await;
                     }
                 }
@@ -211,7 +212,7 @@ where
                     tracing::debug!("re-subscribe success");
                     self.rx = rx;
                     if let Err(err) = self.wait_init_notification().await {
-                        tracing::warn!("Receives meta's notification err {:?}", err);
+                        tracing::warn!(error = %err.as_report(), "Receives meta's notification err");
                         continue;
                     } else {
                         break;

src/common/src/array/arrow.rs

@@ -45,15 +45,14 @@ pub fn to_record_batch_with_schema(
             if column.data_type() == field.data_type() {
                 Ok(column)
             } else {
-                cast(&column, field.data_type())
-                    .map_err(|err| ArrayError::FromArrow(err.to_string()))
+                cast(&column, field.data_type()).map_err(ArrayError::from_arrow)
             }
         })
         .try_collect::<_, _, ArrayError>()?;
 
     let opts = arrow_array::RecordBatchOptions::default().with_row_count(Some(chunk.capacity()));
     arrow_array::RecordBatch::try_new_with_options(schema, columns, &opts)
-        .map_err(|err| ArrayError::ToArrow(err.to_string()))
+        .map_err(ArrayError::to_arrow)
 }
 
 // Implement bi-directional `From` between `DataChunk` and `arrow_array::RecordBatch`.
@@ -84,7 +83,7 @@ impl TryFrom<&DataChunk> for arrow_array::RecordBatch {
         let opts =
             arrow_array::RecordBatchOptions::default().with_row_count(Some(chunk.capacity()));
         arrow_array::RecordBatch::try_new_with_options(schema, columns, &opts)
-            .map_err(|err| ArrayError::ToArrow(err.to_string()))
+            .map_err(ArrayError::to_arrow)
     }
 }
 
@@ -129,7 +128,7 @@ macro_rules! converts_generic {
                             .unwrap()
                             .try_into()?,
                     )),)*
-                    t => Err(ArrayError::FromArrow(format!("unsupported data type: {t:?}"))),
+                    t => Err(ArrayError::from_arrow(format!("unsupported data type: {t:?}"))),
                 }
             }
         }
@@ -252,8 +251,8 @@ impl TryFrom<&DataType> for arrow_schema::DataType {
                 datatype.as_ref().try_into()?,
                 true,
             )))),
-            DataType::Serial => Err(ArrayError::ToArrow(
-                "Serial type is not supported to convert to arrow".to_string(),
+            DataType::Serial => Err(ArrayError::to_arrow(
+                "Serial type is not supported to convert to arrow",
             )),
         }
     }
@@ -485,9 +484,9 @@ impl TryFrom<&arrow_array::LargeBinaryArray> for DecimalArray {
             .map(|o| {
                 o.map(|s| {
                     let s = std::str::from_utf8(s)
-                        .map_err(|_| ArrayError::FromArrow(format!("invalid decimal: {s:?}")))?;
+                        .map_err(|_| ArrayError::from_arrow(format!("invalid decimal: {s:?}")))?;
                     s.parse()
-                        .map_err(|_| ArrayError::FromArrow(format!("invalid decimal: {s:?}")))
+                        .map_err(|_| ArrayError::from_arrow(format!("invalid decimal: {s:?}")))
                 })
                 .transpose()
             })
@@ -521,7 +520,7 @@ impl TryFrom<&arrow_array::LargeStringArray> for JsonbArray {
             .map(|o| {
                 o.map(|s| {
                     s.parse()
-                        .map_err(|_| ArrayError::FromArrow(format!("invalid json: {s}")))
+                        .map_err(|_| ArrayError::from_arrow(format!("invalid json: {s}")))
                 })
                 .transpose()
             })