reject event on key collision, log error on ingestion failure

Author: nikhilsinhaparseable

src/event/format/json.rs

@@ -27,7 +27,10 @@ use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
 use datafusion::arrow::util::bit_util::round_upto_multiple_of_64;
 use itertools::Itertools;
 use serde_json::Value;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 use tracing::error;
 
 use super::EventFormat;
@@ -74,6 +77,10 @@ impl EventFormat for Event {
             _ => unreachable!("flatten would have failed beforehand"),
         };
 
+        // Rename JSON keys starting with '@' to '_' to match the schema
+        // Reject event if renaming would cause a key collision
+        let value_arr = rename_json_keys(value_arr)?;
+
         // collect all the keys from all the json objects in the request body
         let fields =
             collect_keys(value_arr.iter()).expect("fields can be collected from array of objects");
@@ -118,9 +125,6 @@ impl EventFormat for Event {
             ));
         }
 
-        // Rename JSON keys starting with '@' to '_' to match the schema
-        let value_arr = rename_json_keys(value_arr);
-
         Ok((value_arr, schema, is_first))
     }
 
@@ -260,22 +264,44 @@ fn collect_keys<'a>(values: impl Iterator<Item = &'a Value>) -> Result<Vec<&'a s
     Ok(keys)
 }
 
-/// Renames JSON keys to match the schema transformation using normalize_field_name
-fn rename_json_keys(values: Vec<Value>) -> Vec<Value> {
+/// Renames JSON keys to match the schema transformation using normalize_field_name.
+/// Returns an error if renaming would cause a key collision.
+fn rename_json_keys(values: Vec<Value>) -> Result<Vec<Value>, anyhow::Error> {
     values
         .into_iter()
         .map(|value| {
             if let Value::Object(map) = value {
+                // Collect original keys to check for collisions
+                let original_keys: HashSet<String> = map.keys().cloned().collect();
+
+                // Check for collisions before renaming
+                for key in map.keys() {
+                    if key.starts_with('@') {
+                        let mut normalized_key = key.clone();
+                        super::normalize_field_name(&mut normalized_key);
+                        if original_keys.contains(&normalized_key) {
+                            return Err(anyhow!(
+                                "Key collision detected: '{}' and '{}' would both map to '{}'",
+                                key,
+                                normalized_key,
+                                normalized_key
+                            ));
+                        }
+                    }
+                }
+
                 let new_map: serde_json::Map<String, Value> = map
                     .into_iter()
                     .map(|(mut key, val)| {
-                        super::normalize_field_name(&mut key);
+                        if key.starts_with('@') {
+                            super::normalize_field_name(&mut key);
+                        }
                         (key, val)
                     })
                     .collect();
-                Value::Object(new_map)
+                Ok(Value::Object(new_map))
             } else {
-                value
+                Ok(value)
             }
         })
         .collect()
@@ -291,10 +317,7 @@ fn fields_mismatch(
         if val.is_null() {
             continue;
         }
-        // Normalize field name to match schema transformation
-        let mut lookup_name = name.to_string();
-        super::normalize_field_name(&mut lookup_name);
-        let Some(field) = get_field(schema, &lookup_name) else {
+        let Some(field) = get_field(schema, name) else {
             return true;
         };
         if !valid_type(field, val, schema_version, static_schema_flag) {

src/handlers/http/ingest.rs

@@ -24,6 +24,7 @@ use actix_web::{HttpRequest, HttpResponse, http::header::ContentType};
 use arrow_array::RecordBatch;
 use bytes::Bytes;
 use chrono::Utc;
+use tracing::error;
 
 use crate::event::error::EventError;
 use crate::event::format::known_schema::{self, KNOWN_SCHEMA_LIST};
@@ -543,6 +544,7 @@ impl actix_web::ResponseError for PostError {
     }
 
     fn error_response(&self) -> actix_web::HttpResponse<actix_web::body::BoxBody> {
+        error!("{self}");
         match self {
             PostError::MetastoreError(metastore_error) => {
                 actix_web::HttpResponse::build(metastore_error.status_code())