feat(pipeline): gsub prosessor

src/pipeline/src/etl/processor/gsub.rs

@@ -0,0 +1,229 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use regex::Regex;
+
+use crate::etl::field::{Field, Fields};
+use crate::etl::processor::{
+ yaml_bool, yaml_field, yaml_fields, yaml_string, Processor, FIELDS_NAME, FIELD_NAME,
+ IGNORE_MISSING_NAME,
+};
+use crate::etl::value::{Array, Map, Value};
+
+pub(crate) const PROCESSOR_GSUB: &str = "gsub";
+
+const REPLACEMENT_NAME: &str = "replacement";
+const PATTERN_NAME: &str = "pattern";
+
+/// A processor to replace all matches of a pattern in string by a replacement, only support string value, and array string value
+#[derive(Debug, Default)]
+pub struct GsubProcessor {
+ fields: Fields,
+ pattern: Option<Regex>,
+ replacement: Option<String>,
+ ignore_missing: bool,
+}
+
+impl GsubProcessor {
+ fn with_fields(&mut self, fields: Fields) {
+ self.fields = fields;
+ }
+
+ fn with_ignore_missing(&mut self, ignore_missing: bool) {
+ self.ignore_missing = ignore_missing;
+ }
+
+ fn try_pattern(&mut self, pattern: &str) -> Result<(), String> {
+ self.pattern = Some(Regex::new(pattern).map_err(|e| e.to_string())?);
+ Ok(())
+ }
+
+ fn with_replacement(&mut self, replacement: impl Into<String>) {
+ self.replacement = Some(replacement.into());
+ }
+
+ fn check(self) -> Result<Self, String> {
+ if self.pattern.is_none() {
+ return Err("pattern is required".to_string());
+ }
+
+ if self.replacement.is_none() {
+ return Err("replacement is required".to_string());
+ }
+
+ Ok(self)
+ }
+
+ fn process_string_field(&self, val: &str, field: &Field) -> Result<Map, String> {
+ let replacement = self.replacement.as_ref().unwrap();
+ let new_val = self
+ .pattern
+ .as_ref()
+ .unwrap()
+ .replace_all(val, replacement)
+ .to_string();
+ let val = Value::String(new_val);
+
+ let key = match field.target_field {
+ Some(ref target_field) => target_field,
+ None => field.get_field(),
+ };
+
+ Ok(Map::one(key, val))
+ }
+
+ fn process_array_field(&self, arr: &Array, field: &Field) -> Result<Map, String> {
+ let key = match field.target_field {
+ Some(ref target_field) => target_field,
+ None => field.get_field(),
+ };
+
+ let re = self.pattern.as_ref().unwrap();
+ let replacement = self.replacement.as_ref().unwrap();
+
+ let mut result = Array::default();
+ for val in arr.iter() {
+ match val {
+ Value::String(haystack) => {
+ let new_val = re.replace_all(haystack, replacement).to_string();
+ result.push(Value::String(new_val));
+ }
+ _ => {
+ return Err(format!(
+ "{} processor: expect string or array string, but got {val:?}",
+ self.kind()
+ ))
+ }
+ }
+ }
+
+ Ok(Map::one(key, Value::Array(result)))
+ }
+}
+
+impl TryFrom<&yaml_rust::yaml::Hash> for GsubProcessor {
+ type Error = String;
+
+ fn try_from(value: &yaml_rust::yaml::Hash) -> Result<Self, Self::Error> {
+ let mut processor = GsubProcessor::default();
+
+ for (k, v) in value.iter() {
+ let key = k
+ .as_str()
+ .ok_or(format!("key must be a string, but got {k:?}"))?;
+ match key {
+ FIELD_NAME => {
+ processor.with_fields(Fields::one(yaml_field(v, FIELD_NAME)?));
+ }
+ FIELDS_NAME => {
+ processor.with_fields(yaml_fields(v, FIELDS_NAME)?);
+ }
+ PATTERN_NAME => {
+ processor.try_pattern(&yaml_string(v, PATTERN_NAME)?)?;
+ }
+ REPLACEMENT_NAME => {
+ processor.with_replacement(yaml_string(v, REPLACEMENT_NAME)?);
+ }
+
+ IGNORE_MISSING_NAME => {
+ processor.with_ignore_missing(yaml_bool(v, IGNORE_MISSING_NAME)?);
+ }
+
+ _ => {}
+ }
+ }
+
+ processor.check()
+ }
+}
+
+impl crate::etl::processor::Processor for GsubProcessor {
+ fn kind(&self) -> &str {
+ PROCESSOR_GSUB
+ }
+
+ fn ignore_missing(&self) -> bool {
+ self.ignore_missing
+ }
+
+ fn fields(&self) -> &Fields {
+ &self.fields
+ }
+
+ fn exec_field(&self, val: &Value, field: &Field) -> Result<Map, String> {
+ match val {
+ Value::String(val) => self.process_string_field(val, field),
+ Value::Array(arr) => self.process_array_field(arr, field),
+ _ => Err(format!(
+ "{} processor: expect string or array string, but got {val:?}",
+ self.kind()
+ )),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::etl::field::Field;
+ use crate::etl::processor::gsub::GsubProcessor;
+ use crate::etl::processor::Processor;
+ use crate::etl::value::{Map, Value};
+
+ #[test]
+ fn test_string_value() {
+ let mut processor = GsubProcessor::default();
+ processor.try_pattern(r"\d+").unwrap();
+ processor.with_replacement("xxx");
+
+ let field = Field::new("message");
+ let val = Value::String("123".to_string());
+ let result = processor.exec_field(&val, &field).unwrap();
+
+ assert_eq!(
+ result,
+ Map::one("message", Value::String("xxx".to_string()))
+ );
+ }
+
+ #[test]
+ fn test_array_string_value() {
+ let mut processor = GsubProcessor::default();
+ processor.try_pattern(r"\d+").unwrap();
+ processor.with_replacement("xxx");
+
+ let field = Field::new("message");
+ let val = Value::Array(
+ vec![
+ Value::String("123".to_string()),
+ Value::String("456".to_string()),
+ ]
+ .into(),
+ );
+ let result = processor.exec_field(&val, &field).unwrap();
+
+ assert_eq!(
+ result,
+ Map::one(
+ "message",
+ Value::Array(
+ vec![
+ Value::String("xxx".to_string()),
+ Value::String("xxx".to_string())
+ ]
+ .into()
+ )
+ )
+ );
+ }
+}

src/pipeline/src/etl/processor/mod.rs

@@ -17,6 +17,7 @@ pub mod csv;
 pub mod date;
 pub mod dissect;
 pub mod epoch;
+pub mod gsub;
 pub mod letter;
 pub mod regex;
 pub mod urlencoding;
@@ -29,6 +30,7 @@ use csv::CsvProcessor;
 use date::DateProcessor;
 use dissect::DissectProcessor;
 use epoch::EpochProcessor;
+use gsub::GsubProcessor;
 use letter::LetterProcessor;
 use regex::RegexProcessor;
 use urlencoding::UrlEncodingProcessor;
@@ -163,6 +165,7 @@ fn parse_processor(doc: &yaml_rust::Yaml) -> Result<Arc<dyn Processor>, String>
  date::PROCESSOR_DATE => Arc::new(DateProcessor::try_from(value)?),
  dissect::PROCESSOR_DISSECT => Arc::new(DissectProcessor::try_from(value)?),
  epoch::PROCESSOR_EPOCH => Arc::new(EpochProcessor::try_from(value)?),
+ gsub::PROCESSOR_GSUB => Arc::new(GsubProcessor::try_from(value)?),
  letter::PROCESSOR_LETTER => Arc::new(LetterProcessor::try_from(value)?),
  regex::PROCESSOR_REGEX => Arc::new(RegexProcessor::try_from(value)?),
  urlencoding::PROCESSOR_URL_ENCODING => Arc::new(UrlEncodingProcessor::try_from(value)?),

src/pipeline/src/etl/value/array.rs

@@ -45,6 +45,12 @@ impl std::ops::Deref for Array {
  }
 }
 
+impl std::ops::DerefMut for Array {
+ fn deref_mut(&mut self) -> &mut Self::Target {
+ &mut self.values
+ }
+}
+
 impl IntoIterator for Array {
  type Item = Value;
 
@@ -54,3 +60,9 @@ impl IntoIterator for Array {
  self.values.into_iter()
  }
 }
+
+impl From<Vec<Value>> for Array {
+ fn from(values: Vec<Value>) -> Self {
+ Array { values }
+ }
+}

src/pipeline/tests/gsub.rs

@@ -0,0 +1,70 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use greptime_proto::v1::value::ValueData::TimestampMillisecondValue;
+use greptime_proto::v1::{ColumnDataType, ColumnSchema, SemanticType};
+use pipeline::{parse, Content, GreptimeTransformer, Pipeline, Value};
+
+#[test]
+fn test_gsub() {
+ let input_value_str = r#"
+ [
+ {
+ "reqTimeSec": "1573840000.000"
+ }
+ ]
+"#;
+ let input_value: Value = serde_json::from_str::<serde_json::Value>(input_value_str)
+ .expect("failed to parse input value")
+ .try_into()
+ .expect("failed to convert input value");
+
+ let pipeline_yaml = r#"
+---
+description: Pipeline for Akamai DataStream2 Log
+
+processors:
+ - gsub:
+ field: reqTimeSec
+ pattern: "\\."
+ replacement: ""
+ - epoch:
+ field: reqTimeSec
+ resolution: millisecond
+ ignore_missing: true
+
+transform:
+ - field: reqTimeSec
+ type: epoch, millisecond
+ index: timestamp
+"#;
+
+ let yaml_content = Content::Yaml(pipeline_yaml.into());
+ let pipeline: Pipeline<GreptimeTransformer> =
+ parse(&yaml_content).expect("failed to parse pipeline");
+ let output = pipeline.exec(input_value).expect("failed to exec pipeline");
+
+ let expected_schema = vec![ColumnSchema {
+ column_name: "reqTimeSec".to_string(),
+ datatype: ColumnDataType::TimestampMillisecond.into(),
+ semantic_type: SemanticType::Timestamp.into(),
+ datatype_extension: None,
+ }];
+
+ assert_eq!(output.schema, expected_schema);
+ assert_eq!(
+ output.rows[0].values[0].value_data,
+ Some(TimestampMillisecondValue(1573840000000))
+ );
+}

src/pipeline/tests/pipeline.rs

@@ -19,10 +19,6 @@ use greptime_proto::v1::value::ValueData::{
 use greptime_proto::v1::Value as GreptimeValue;
 use pipeline::{parse, Content, GreptimeTransformer, Pipeline, Value};
 
-// use pipeline::transform::GreptimeTransformer;
-// use pipeline::value::Value;
-// use pipeline::{parse, Content, Pipeline};
-
 #[test]
 fn main() {
  let input_value_str = r#"