feat: pipeline dispatcher part 2: execution

src/frontend/src/instance/otlp.rs

@@ -20,11 +20,11 @@ use common_telemetry::tracing;
 use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
 use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
 use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
-use pipeline::PipelineWay;
+use pipeline::{GreptimePipelineParams, PipelineWay};
 use servers::error::{self, AuthSnafu, InFlightWriteBytesExceededSnafu, Result as ServerResult};
 use servers::interceptor::{OpenTelemetryProtocolInterceptor, OpenTelemetryProtocolInterceptorRef};
 use servers::otlp;
-use servers::query_handler::OpenTelemetryProtocolHandler;
+use servers::query_handler::{OpenTelemetryProtocolHandler, PipelineHandlerRef};
 use session::context::QueryContextRef;
 use snafu::ResultExt;
 
@@ -112,8 +112,10 @@ impl OpenTelemetryProtocolHandler for Instance {
     #[tracing::instrument(skip_all)]
     async fn logs(
         &self,
+        pipeline_handler: PipelineHandlerRef,
         request: ExportLogsServiceRequest,
         pipeline: PipelineWay,
+        pipeline_params: GreptimePipelineParams,
         table_name: String,
         ctx: QueryContextRef,
     ) -> ServerResult<Output> {
@@ -128,7 +130,15 @@ impl OpenTelemetryProtocolHandler for Instance {
             .get::<OpenTelemetryProtocolInterceptorRef<servers::error::Error>>();
         interceptor_ref.pre_execute(ctx.clone())?;
 
-        let (requests, rows) = otlp::logs::to_grpc_insert_requests(request, pipeline, table_name)?;
+        let (requests, rows) = otlp::logs::to_grpc_insert_requests(
+            request,
+            pipeline,
+            pipeline_params,
+            table_name,
+            &ctx,
+            pipeline_handler,
+        )
+        .await?;
 
         let _guard = if let Some(limiter) = &self.limiter {
             let result = limiter.limit_row_inserts(&requests);

src/pipeline/benches/processor.rs

@@ -13,21 +13,22 @@
 // limitations under the License.
 
 use criterion::{black_box, criterion_group, criterion_main, Criterion};
-use pipeline::{parse, Content, GreptimeTransformer, Pipeline, Result};
+use pipeline::{json_to_intermediate_state, parse, Content, GreptimeTransformer, Pipeline, Result};
 use serde_json::{Deserializer, Value};
 
 fn processor_mut(
     pipeline: &Pipeline<GreptimeTransformer>,
     input_values: Vec<Value>,
 ) -> Result<Vec<greptime_proto::v1::Row>> {
-    let mut payload = pipeline.init_intermediate_state();
     let mut result = Vec::with_capacity(input_values.len());
 
     for v in input_values {
-        pipeline.prepare(v, &mut payload)?;
-        let r = pipeline.exec_mut(&mut payload)?;
+        let mut payload = json_to_intermediate_state(v).unwrap();
+        let r = pipeline
+            .exec_mut(&mut payload)?
+            .into_transformed()
+            .expect("expect transformed result ");
         result.push(r);
-        pipeline.reset_intermediate_state(&mut payload);
     }
 
     Ok(result)

src/pipeline/src/dispatcher.rs

@@ -12,12 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
+
+use common_telemetry::debug;
 use snafu::OptionExt;
 use yaml_rust::Yaml;
 
-use crate::etl::error::{Error, Result};
-use crate::etl_error::{
-    FieldRequiredForDispatcherSnafu, TablePartRequiredForDispatcherRuleSnafu,
+use crate::etl::error::{
+    Error, FieldRequiredForDispatcherSnafu, Result, TablePartRequiredForDispatcherRuleSnafu,
     ValueRequiredForDispatcherRuleSnafu,
 };
 use crate::Value;
@@ -84,6 +86,7 @@ impl TryFrom<&Yaml> for Dispatcher {
                         .as_str()
                         .map(|s| s.to_string())
                         .context(TablePartRequiredForDispatcherRuleSnafu)?;
+
                     let pipeline = rule[PIPELINE].as_str().map(|s| s.to_string());
 
                     if rule[VALUE].is_badvalue() {
@@ -105,3 +108,21 @@ impl TryFrom<&Yaml> for Dispatcher {
         Ok(Dispatcher { field, rules })
     }
 }
+
+impl Dispatcher {
+    /// execute dispatcher and returns matched rule if any
+    pub(crate) fn exec(&self, data: &BTreeMap<String, Value>) -> Option<&Rule> {
+        if let Some(value) = data.get(&self.field) {
+            for rule in &self.rules {
+                if rule.value == *value {
+                    return Some(rule);
+                }
+            }
+
+            None
+        } else {
+            debug!("field {} not found in keys {:?}", &self.field, data.keys());
+            None
+        }
+    }
+}