fix(error): do not directly format BoxedError (#16558)

Signed-off-by: Bugen Zhao <[email protected]>

lints/ui/format_error.rs

@@ -80,4 +80,13 @@ fn main() {
     let _ = anyhow!("some error occurred: {}", err.as_report());
     let _ = anyhow!("{:?}", anyhow_err.as_report());
     let _ = anyhow!("some error occurred: {:?}", anyhow_err.as_report());
+
+    let box_dyn_err_1: Box<dyn Error> = Box::new(err.clone());
+    let box_dyn_err_2: Box<dyn Error + Send> = Box::new(err.clone());
+    let box_dyn_err_3: Box<dyn Error + Send + Sync> = Box::new(err.clone());
+
+    // TODO: fail to lint
+    let _ = format!("{}", box_dyn_err_1);
+    info!("{}", box_dyn_err_2);
+    let _ = box_dyn_err_3.to_string();
 }

src/frontend/src/error.rs

@@ -237,3 +237,13 @@ impl From<JoinError> for RwError {
         ErrorCode::Uncategorized(join_error.into()).into()
     }
 }
+
+// For errors without a concrete type, put them into `Uncategorized`.
+impl From<BoxedError> for RwError {
+    fn from(e: BoxedError) -> Self {
+        // Show that the error is of `BoxedKind`, instead of `AdhocKind` which loses the sources.
+        // This is essentially expanded from `anyhow::anyhow!(e)`.
+        let e = anyhow::__private::kind::BoxedKind::anyhow_kind(&e).new(e);
+        ErrorCode::Uncategorized(e).into()
+    }
+}

src/frontend/src/handler/query.rs

@@ -386,9 +386,7 @@ async fn execute(
                         "no affected rows in output".to_string(),
                     )))
                 }
-                Some(row) => {
-                    row.map_err(|err| RwError::from(ErrorCode::InternalError(format!("{}", err))))?
-                }
+                Some(row) => row?,
             };
             let affected_rows_str = first_row_set[0].values()[0]
                 .as_ref()

src/frontend/src/scheduler/distributed/query_manager.rs

@@ -50,6 +50,7 @@ impl DistributedQueryStream {
 }
 
 impl Stream for DistributedQueryStream {
+    // TODO(error-handling): use a concrete error type.
     type Item = Result<DataChunk, BoxedError>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {

src/frontend/src/scheduler/local.rs

@@ -55,6 +55,7 @@ use crate::scheduler::task_context::FrontendBatchTaskContext;
 use crate::scheduler::{ReadSnapshot, SchedulerError, SchedulerResult};
 use crate::session::{FrontendEnv, SessionImpl};
 
+// TODO(error-handling): use a concrete error type.
 pub type LocalQueryStream = ReceiverStream<Result<DataChunk, BoxedError>>;
 pub struct LocalQueryExecution {
     sql: String,

src/frontend/src/session/cursor_manager.rs

@@ -27,7 +27,7 @@ use risingwave_common::types::DataType;
 use risingwave_sqlparser::ast::{Ident, ObjectName, Statement};
 
 use crate::catalog::subscription_catalog::SubscriptionCatalog;
-use crate::error::{ErrorCode, Result, RwError};
+use crate::error::{ErrorCode, Result};
 use crate::handler::declare_cursor::create_stream_for_cursor;
 use crate::handler::util::{
     convert_epoch_to_logstore_i64, convert_logstore_i64_to_unix_millis,
@@ -78,9 +78,7 @@ impl QueryCursor {
             let rows = self.row_stream.next().await;
             let rows = match rows {
                 None => return Ok(None),
-                Some(row) => {
-                    row.map_err(|err| RwError::from(ErrorCode::InternalError(format!("{}", err))))?
-                }
+                Some(row) => row?,
             };
             self.remaining_rows = rows.into_iter().collect();
         }
@@ -396,9 +394,7 @@ impl SubscriptionCursor {
         if remaining_rows.is_empty()
             && let Some(row_set) = row_stream.next().await
         {
-            remaining_rows.extend(row_set.map_err(|e| {
-                ErrorCode::InternalError(format!("Cursor get next chunk error {:?}", e.to_string()))
-            })?);
+            remaining_rows.extend(row_set?);
         }
         Ok(())
     }