fix: pipeline condition node incorrect evaluation (#5219)

fix #5218 

Backend:
- [x] allow condition node to evaluate `null` and `""`, empty, for both
`=` and `!=` conditions
- [x] stop ingestion when resolved dynamic stream name is empty

**How to Use(In ConditionNode):**
To evaluate field value is not empty:
`app_name != ""` 
To evaluate field value is not null:
`app_name != null`

Author: taimingl

.github/ISSUE_TEMPLATE/2-feature-request.yaml

@@ -1,4 +1,4 @@
-name: 'Feature Request'
+name: "Feature Request"
 description: Suggest a feature for OpenObserve
 labels: ["✏️ Feature"]
 
@@ -18,7 +18,7 @@ body:
         - ingestion
         - alerts
         - localization
-        - routing
+        - pipeline
         - metrics
         - traces
         - other
@@ -43,4 +43,4 @@ body:
     attributes:
       label: Alternatives considered
     validations:
-      required: true
+      required: true

src/config/src/meta/stream.rs

@@ -822,13 +822,14 @@ impl RoutingCondition {
         let val = match row.get(&self.column) {
             Some(val) => val,
             None => {
+                // field not found -> dropped
                 return false;
             }
         };
         match val {
             Value::String(v) => {
                 let val = v.as_str();
-                let con_val = self.value.as_str().unwrap_or_default();
+                let con_val = self.value.as_str().unwrap_or_default().trim_matches('"'); // "" is interpreted as empty string
                 match self.operator {
                     Operator::EqualTo => val == con_val,
                     Operator::NotEqualTo => val != con_val,
@@ -878,6 +879,10 @@ impl RoutingCondition {
                     _ => false,
                 }
             }
+            Value::Null => {
+                matches!(self.operator, Operator::EqualTo)
+                    && matches!(&self.value, Value::String(v) if v == "null")
+            }
             _ => false,
         }
     }

src/service/pipeline/batch_execution.rs

@@ -242,20 +242,7 @@ impl ExecutablePipeline {
             log::debug!("[Pipeline]: starts result collecting job");
             let mut count: usize = 0;
             let mut results = HashMap::new();
-            while let Some((idx, mut stream_params, record)) = result_receiver.recv().await {
-                if stream_params.stream_name.contains("{") {
-                    match resolve_stream_name(&stream_params.stream_name, &record) {
-                        Ok(stream_name) => {
-                            stream_params.stream_name = stream_name.into();
-                        }
-                        Err(e) => {
-                            log::error!(
-                                "[Pipeline]: dynamic stream name detected in destination, but failed to resolve due to {e}. Record dropped."
-                            );
-                            continue;
-                        }
-                    }
-                }
+            while let Some((idx, stream_params, record)) = result_receiver.recv().await {
                 results
                     .entry(stream_params)
                     .or_insert(Vec::new())
@@ -528,9 +515,36 @@ async fn process_node(
                         };
                     }
 
-                    if let Err(send_err) = result_sender
-                        .send((idx, stream_params.clone(), record))
-                        .await
+                    let mut destination_stream = stream_params.clone();
+                    if destination_stream.stream_name.contains("{") {
+                        match resolve_stream_name(&destination_stream.stream_name, &record) {
+                            Ok(stream_name) if !stream_name.is_empty() => {
+                                destination_stream.stream_name = stream_name.into();
+                            }
+                            resolve_res => {
+                                let err_msg = if let Err(e) = resolve_res {
+                                    format!("Dynamic stream name detected in destination, but failed to resolve due to {e}. Record dropped")
+                                } else {
+                                    "Dynamic Stream Name resolved to empty. Record dropped"
+                                        .to_string()
+                                };
+                                log::error!("{err_msg}");
+                                if let Err(send_err) = error_sender
+                                    .send((node.id.to_string(), node.node_type(), err_msg))
+                                    .await
+                                {
+                                    log::error!(
+                                        "[Pipeline]: LeafNode failed sending errors for collection caused by: {send_err}"
+                                    );
+                                    break;
+                                }
+                                continue;
+                            }
+                        }
+                    }
+
+                    if let Err(send_err) =
+                        result_sender.send((idx, destination_stream, record)).await
                     {
                         log::error!(
                             "[Pipeline]: LeafNode errors sending result for collection caused by: {send_err}"