From 899310656aeabb27bc9fad1fd3afcef532a82cb2 Mon Sep 17 00:00:00 2001
From: paomian <xpaomian@gmail.com>
Date: Tue, 10 Sep 2024 12:03:16 +0800
Subject: [PATCH 1/2] fix: pipeline dissert error is returned directly to the
 user, instead of printing a warn log

---
 src/pipeline/src/etl/processor/dissect.rs | 14 +++++---------
 1 file changed, 5 insertions(+), 9 deletions(-)

diff --git a/src/pipeline/src/etl/processor/dissect.rs b/src/pipeline/src/etl/processor/dissect.rs
index 9a4b8a966e6..dca88d38430 100644
--- a/src/pipeline/src/etl/processor/dissect.rs
+++ b/src/pipeline/src/etl/processor/dissect.rs
@@ -817,16 +817,12 @@ impl Processor for DissectProcessor {
         for field in self.fields.iter() {
             let index = field.input_index();
             match val.get(index) {
-                Some(Value::String(val_str)) => match self.process(val_str) {
-                    Ok(r) => {
-                        for (k, v) in r {
-                            val[k] = v;
-                        }
-                    }
-                    Err(e) => {
-                        warn!("dissect processor: {}", e);
+                Some(Value::String(val_str)) => {
+                    let r = self.process(val_str)?;
+                    for (k, v) in r {
+                        val[k] = v;
                     }
-                },
+                }
                 Some(Value::Null) | None => {
                     if !self.ignore_missing {
                         return Err(format!(

From c852ccd2565f39703f5fedeeee6cdb3d8483eeff Mon Sep 17 00:00:00 2001
From: paomian <xpaomian@gmail.com>
Date: Tue, 10 Sep 2024 15:18:30 +0800
Subject: [PATCH 2/2] chore: add more test for pipeline

---
 src/pipeline/src/etl.rs       | 21 ++++++++++------
 src/pipeline/tests/dissect.rs | 34 ++++++++++++++++++++++++++
 src/pipeline/tests/regex.rs   | 46 +++++++++++++++++++++++++++++++++++
 3 files changed, 93 insertions(+), 8 deletions(-)

diff --git a/src/pipeline/src/etl.rs b/src/pipeline/src/etl.rs
index de4c544a011..f6b3efd6e6f 100644
--- a/src/pipeline/src/etl.rs
+++ b/src/pipeline/src/etl.rs
@@ -21,7 +21,7 @@ pub mod value;
 
 use ahash::HashSet;
 use common_telemetry::debug;
-use itertools::{merge, Itertools};
+use itertools::Itertools;
 use processor::{Processor, ProcessorBuilder, Processors};
 use transform::{TransformBuilders, Transformer, Transforms};
 use value::Value;
@@ -91,13 +91,18 @@ where
             debug!("required_keys: {:?}", required_keys);
 
             // intermediate keys are the keys that all processor and transformer required
-            let ordered_intermediate_keys: Vec<String> =
-                merge(processors_required_keys, transforms_required_keys)
-                    .cloned()
-                    .collect::<HashSet<String>>()
-                    .into_iter()
-                    .sorted()
-                    .collect();
+            let ordered_intermediate_keys: Vec<String> = [
+                processors_required_keys,
+                transforms_required_keys,
+                processors_output_keys,
+            ]
+            .iter()
+            .flat_map(|l| l.iter())
+            .collect::<HashSet<&String>>()
+            .into_iter()
+            .sorted()
+            .cloned()
+            .collect_vec();
 
             let mut final_intermediate_keys = Vec::with_capacity(ordered_intermediate_keys.len());
             let mut intermediate_keys_exclude_original =
diff --git a/src/pipeline/tests/dissect.rs b/src/pipeline/tests/dissect.rs
index 82ce63399c6..22cf14c46bb 100644
--- a/src/pipeline/tests/dissect.rs
+++ b/src/pipeline/tests/dissect.rs
@@ -247,3 +247,37 @@ transform:
         Some(StringValue("key1_key2".to_string()))
     );
 }
+
+#[test]
+fn test_parse_failure() {
+    let input_str = r#"
+{
+    "str": "key1 key2"
+}"#;
+
+    let pipeline_yaml = r#"
+processors:
+  - dissect:
+      field: str
+      patterns:
+        - "%{key1} %{key2} %{key3}"
+
+transform:
+  - fields:
+      - key1
+    type: string
+"#;
+
+    let input_value = serde_json::from_str::<serde_json::Value>(input_str).unwrap();
+
+    let yaml_content = pipeline::Content::Yaml(pipeline_yaml.into());
+    let pipeline: pipeline::Pipeline<pipeline::GreptimeTransformer> =
+        pipeline::parse(&yaml_content).expect("failed to parse pipeline");
+    let mut result = pipeline.init_intermediate_state();
+
+    pipeline.prepare(input_value, &mut result).unwrap();
+    let row = pipeline.exec_mut(&mut result);
+
+    assert!(row.is_err());
+    assert_eq!(row.err().unwrap(), "No matching pattern found");
+}
diff --git a/src/pipeline/tests/regex.rs b/src/pipeline/tests/regex.rs
index 5be60c98752..a8a7daaf5c6 100644
--- a/src/pipeline/tests/regex.rs
+++ b/src/pipeline/tests/regex.rs
@@ -122,3 +122,49 @@ transform:
 
     assert_eq!(output.rows[0].values[0].value_data, None);
 }
+
+#[test]
+fn test_unuse_regex_group() {
+    let input_value_str = r#"
+  [
+    {
+      "str": "123 456"
+    }
+  ]
+"#;
+
+    let pipeline_yaml = r#"
+processors:
+- regex:
+    fields:
+      - str
+    pattern: "(?<id1>\\d+) (?<id2>\\d+)"
+
+transform:
+- field: str_id1
+  type: string
+"#;
+
+    let output = common::parse_and_exec(input_value_str, pipeline_yaml);
+
+    assert_eq!(
+        output.schema,
+        vec![
+            common::make_column_schema(
+                "str_id1".to_string(),
+                ColumnDataType::String,
+                SemanticType::Field,
+            ),
+            common::make_column_schema(
+                "greptime_timestamp".to_string(),
+                ColumnDataType::TimestampNanosecond,
+                SemanticType::Timestamp,
+            ),
+        ]
+    );
+
+    assert_eq!(
+        output.rows[0].values[0].value_data,
+        Some(StringValue("123".to_string()))
+    );
+}