Merge pull request #8 from shuiyisong/refactor/log

refactor: minor update

src/frontend/src/error.rs

@@ -297,29 +297,6 @@ pub enum Error {
         source: table::error::Error,
     },
 
-    #[snafu(display("Failed to insert pipeline to pipeline table, name: {}", name))]
-    InsertPipeline {
-        name: String,
-        #[snafu(implicit)]
-        location: Location,
-        source: BoxedError,
-    },
-
-    #[snafu(display("Failed to parse pipeline"))]
-    ParsePipeline {
-        #[snafu(implicit)]
-        location: Location,
-        source: BoxedError,
-    },
-
-    #[snafu(display("Failed to get pipeline to pipeline table, name: {}", name))]
-    GetPipeline {
-        name: String,
-        #[snafu(implicit)]
-        location: Location,
-        source: BoxedError,
-    },
-
     #[snafu(display("Unsupported format: {:?}", format))]
     UnsupportedFormat {
         #[snafu(implicit)]
@@ -406,9 +383,6 @@ impl ErrorExt for Error {
             | Error::EmptyData { .. }
             | Error::ColumnNoneDefaultValue { .. }
             | Error::IncompleteGrpcRequest { .. }
-            | Error::InsertPipeline { .. }
-            | Error::ParsePipeline { .. }
-            | Error::GetPipeline { .. }
             | Error::InvalidTlsConfig { .. } => StatusCode::InvalidArguments,
 
             Error::NotSupported { .. } => StatusCode::Unsupported,

src/frontend/src/instance.rs

@@ -49,6 +49,7 @@ use meta_client::MetaClientOptions;
 use operator::delete::DeleterRef;
 use operator::insert::InserterRef;
 use operator::statement::StatementExecutor;
+use pipeline::pipeline_operator::PipelineOperator;
 use prometheus::HistogramTimer;
 use query::metrics::OnDone;
 use query::parser::{PromQuery, QueryLanguageParser, QueryStatement};
@@ -89,7 +90,6 @@ use crate::error::{
 };
 use crate::frontend::FrontendOptions;
 use crate::heartbeat::HeartbeatTask;
-use crate::pipeline::PipelineOperator;
 use crate::script::ScriptExecutor;
 
 #[async_trait]

src/frontend/src/instance/builder.rs

@@ -30,6 +30,7 @@ use operator::request::Requester;
 use operator::statement::{StatementExecutor, StatementExecutorRef};
 use operator::table::TableMutationOperator;
 use partition::manager::PartitionRuleManager;
+use pipeline::pipeline_operator::PipelineOperator;
 use query::QueryEngineFactory;
 use servers::server::ServerHandlers;
 use snafu::OptionExt;
@@ -38,7 +39,6 @@ use crate::error::{self, Result};
 use crate::heartbeat::HeartbeatTask;
 use crate::instance::region_query::FrontendRegionQueryHandler;
 use crate::instance::Instance;
-use crate::pipeline::PipelineOperator;
 use crate::script::ScriptExecutor;
 
 /// The frontend [`Instance`] builder.

src/frontend/src/instance/log_handler.rs

@@ -18,7 +18,7 @@ use auth::{PermissionChecker, PermissionCheckerRef, PermissionReq};
 use client::Output;
 use common_error::ext::BoxedError;
 use pipeline::{GreptimeTransformer, Pipeline};
-use servers::error::{AuthSnafu, ExecuteGrpcRequestSnafu, Result as ServerResult};
+use servers::error::{AuthSnafu, ExecuteGrpcRequestSnafu, PipelineSnafu, Result as ServerResult};
 use servers::query_handler::LogHandler;
 use session::context::QueryContextRef;
 use snafu::ResultExt;
@@ -49,8 +49,7 @@ impl LogHandler for Instance {
         self.pipeline_operator
             .get_pipeline(query_ctx, name)
             .await
-            .map_err(BoxedError::new)
-            .context(servers::error::GetPipelineSnafu { name })
+            .context(PipelineSnafu)
     }
 
     async fn insert_pipeline(
@@ -63,6 +62,7 @@ impl LogHandler for Instance {
         self.pipeline_operator
             .insert_pipeline(name, content_type, pipeline, query_ctx)
             .await
+            .context(PipelineSnafu)
     }
 
     async fn delete_pipeline(&self, _name: &str, _query_ctx: QueryContextRef) -> ServerResult<()> {

src/frontend/src/lib.rs

@@ -19,7 +19,6 @@ pub mod frontend;
 pub mod heartbeat;
 pub mod instance;
 pub(crate) mod metrics;
-mod pipeline;
 mod script;
 pub mod server;
 pub mod service_config;

src/pipeline/src/etl/README.md

@@ -0,0 +1,13 @@
+# pipeline
+
+ETL capability
+
+## processors
+
+refer [elastic ingest processor][elastic-ingest-processor] for detail
+
+### Example
+
+Go to [pipeline](../../tests/pipeline.rs)
+
+[elastic-ingest-processor]: https://www.elastic.co/guide/en/elasticsearch/reference/current/processors.html