add duplicate and empty check

Author: nikhilsinhaparseable

src/handlers/http/modal/utils/ingest_utils.rs

@@ -16,15 +16,15 @@
  *
  */
 
-use std::collections::HashMap;
-
 use actix_web::HttpRequest;
 use chrono::Utc;
 use http::header::USER_AGENT;
 use opentelemetry_proto::tonic::{
     logs::v1::LogsData, metrics::v1::MetricsData, trace::v1::TracesData,
 };
 use serde_json::Value;
+use std::collections::HashMap;
+use tracing::warn;
 
 use crate::{
     event::{
@@ -164,7 +164,14 @@ pub fn get_custom_fields_from_header(req: HttpRequest) -> HashMap<String, String
         if header_name.starts_with("x-p-") && !IGNORE_HEADERS.contains(&header_name) {
             if let Ok(value) = header_value.to_str() {
                 let key = header_name.trim_start_matches("x-p-");
-                p_custom_fields.insert(key.to_string(), value.to_string());
+                if !key.is_empty() {
+                    p_custom_fields.insert(key.to_string(), value.to_string());
+                } else {
+                    warn!(
+                        "Ignoring header with empty key after prefix: {}",
+                        header_name
+                    );
+                }
             }
         }
 
@@ -221,4 +228,17 @@ mod tests {
         assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "TestUserAgent");
         assert_eq!(custom_fields.get(FORMAT_KEY).unwrap(), "otel-logs");
     }
+
+    #[test]
+    fn test_get_custom_fields_empty_header_after_prefix() {
+        let req = TestRequest::default()
+            .insert_header(("x-p-", "empty"))
+            .to_http_request();
+
+        let custom_fields = get_custom_fields_from_header(req);
+
+        assert_eq!(custom_fields.len(), 2);
+        assert_eq!(custom_fields.get(USER_AGENT_KEY).unwrap(), "");
+        assert_eq!(custom_fields.get(SOURCE_IP_KEY).unwrap(), "");
+    }
 }

src/utils/arrow/mod.rs

@@ -40,7 +40,10 @@
 //! }
 //! ```
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 
 use arrow_array::{
     Array, ArrayRef, RecordBatch, StringArray, TimestampMillisecondArray, UInt64Array,
@@ -162,6 +165,8 @@ pub fn add_parseable_fields(
         .iter()
         .map(|f| f.as_ref().clone())
         .collect_vec();
+    let mut field_names: HashSet<String> = fields.iter().map(|f| f.name().to_string()).collect();
+
     fields.insert(
         0,
         Field::new(
@@ -170,23 +175,24 @@ pub fn add_parseable_fields(
             true,
         ),
     );
-    fields.extend(
-        sorted_keys
-            .iter()
-            .map(|k| Field::new(*k, DataType::Utf8, true)),
-    );
-
     let mut columns = rb.columns().iter().map(Arc::clone).collect_vec();
     columns.insert(
         0,
         Arc::new(get_timestamp_array(p_timestamp, row_count)) as ArrayRef,
     );
-    columns.extend(sorted_keys.iter().map(|k| {
-        let value = p_custom_fields.get(*k).unwrap();
-        Arc::new(StringArray::from_iter_values(
-            std::iter::repeat(value).take(row_count),
-        )) as ArrayRef
-    }));
+
+    //ignore the duplicate fields, no need to add them again
+    for key in sorted_keys {
+        if !field_names.contains(key) {
+            fields.push(Field::new(key, DataType::Utf8, true));
+            field_names.insert(key.to_string());
+
+            let value = p_custom_fields.get(key).unwrap();
+            columns.push(Arc::new(StringArray::from_iter_values(
+                std::iter::repeat(value).take(row_count),
+            )) as ArrayRef);
+        }
+    }
 
     // Create the new schema and batch
     let new_schema = Arc::new(Schema::new(fields));