chore: merge error files under pipeline crate (#5738)

src/servers/src/error.rs

@@ -151,7 +151,7 @@ pub enum Error {
     #[snafu(display("Failed to describe statement"))]
     DescribeStatement { source: BoxedError },
 
-    #[snafu(display("Pipeline management api error"))]
+    #[snafu(display("Pipeline error"))]
     Pipeline {
         #[snafu(source)]
         source: pipeline::error::Error,
@@ -159,14 +159,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Pipeline transform error"))]
-    PipelineTransform {
-        #[snafu(source)]
-        source: pipeline::etl_error::Error,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
     #[snafu(display("Not supported: {}", feat))]
     NotSupported { feat: String },
 
@@ -661,7 +653,6 @@ impl ErrorExt for Error {
             | CheckDatabaseValidity { source, .. } => source.status_code(),
 
             Pipeline { source, .. } => source.status_code(),
-            PipelineTransform { source, .. } => source.status_code(),
 
             NotSupported { .. }
             | InvalidParameter { .. }

src/servers/src/http/event.rs

@@ -32,7 +32,6 @@ use common_telemetry::{error, warn};
 use datatypes::value::column_data_to_json;
 use headers::ContentType;
 use lazy_static::lazy_static;
-use pipeline::error::PipelineTransformSnafu;
 use pipeline::util::to_pipeline_version;
 use pipeline::{GreptimePipelineParams, GreptimeTransformer, PipelineDefinition, PipelineVersion};
 use serde::{Deserialize, Serialize};
@@ -284,9 +283,7 @@ async fn dryrun_pipeline_inner(
         &pipeline_handler,
         PipelineDefinition::Resolved(pipeline),
         &params,
-        pipeline::json_array_to_intermediate_state(value)
-            .context(PipelineTransformSnafu)
-            .context(PipelineSnafu)?,
+        pipeline::json_array_to_intermediate_state(value).context(PipelineSnafu)?,
         "dry_run".to_owned(),
         query_ctx,
         true,
@@ -636,9 +633,7 @@ pub(crate) async fn ingest_logs_inner(
             &state,
             PipelineDefinition::from_name(&pipeline_name, version),
             &pipeline_params,
-            pipeline::json_array_to_intermediate_state(request.values)
-                .context(PipelineTransformSnafu)
-                .context(PipelineSnafu)?,
+            pipeline::json_array_to_intermediate_state(request.values).context(PipelineSnafu)?,
             request.table,
             &query_ctx,
             true,

src/servers/src/otlp/logs.rs

@@ -32,7 +32,7 @@ use snafu::{ensure, ResultExt};
 use super::trace::attributes::OtlpAnyValue;
 use super::utils::{bytes_to_hex_string, key_value_to_jsonb};
 use crate::error::{
-    IncompatibleSchemaSnafu, NotSupportedSnafu, PipelineTransformSnafu, Result,
+    IncompatibleSchemaSnafu, NotSupportedSnafu, PipelineSnafu, Result,
     UnsupportedJsonDataTypeForTagSnafu,
 };
 use crate::pipeline::run_pipeline;
@@ -72,8 +72,7 @@ pub async fn to_grpc_insert_requests(
         }
         PipelineWay::Pipeline(pipeline_def) => {
             let data = parse_export_logs_service_request(request);
-            let array =
-                pipeline::json_array_to_intermediate_state(data).context(PipelineTransformSnafu)?;
+            let array = pipeline::json_array_to_intermediate_state(data).context(PipelineSnafu)?;
 
             let inserts = run_pipeline(
                 &pipeline_handler,

src/servers/src/pipeline.rs

@@ -23,7 +23,7 @@ use pipeline::{
 use session::context::QueryContextRef;
 use snafu::ResultExt;
 
-use crate::error::{CatalogSnafu, PipelineTransformSnafu, Result};
+use crate::error::{CatalogSnafu, PipelineSnafu, Result};
 use crate::metrics::{
     METRIC_FAILURE_VALUE, METRIC_HTTP_LOGS_TRANSFORM_ELAPSED, METRIC_SUCCESS_VALUE,
 };
@@ -74,7 +74,7 @@ pub(crate) async fn run_pipeline(
                     table_name,
                 }]
             })
-            .context(PipelineTransformSnafu)
+            .context(PipelineSnafu)
     } else {
         let pipeline = get_pipeline(pipeline_definition, state, query_ctx).await?;
 
@@ -91,7 +91,7 @@ pub(crate) async fn run_pipeline(
                         .with_label_values(&[db.as_str(), METRIC_FAILURE_VALUE])
                         .observe(transform_timer.elapsed().as_secs_f64());
                 })
-                .context(PipelineTransformSnafu)?;
+                .context(PipelineSnafu)?;
 
             match r {
                 PipelineExecOutput::Transformed(row) => {