fix: rebase from main

Author: nikhilsinhaparseable

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

@@ -98,6 +98,7 @@ pub async fn push_logs(
                 &HashMap::new(),
                 size,
                 schema_version,
+                log_source,
             )
             .await?;
         } else {
@@ -107,6 +108,7 @@ pub async fn push_logs(
                 None,
                 custom_partition.as_ref(),
                 schema_version,
+                log_source,
             )?;
             let custom_partition = custom_partition.unwrap();
             let custom_partition_list = custom_partition.split(',').collect::<Vec<&str>>();
@@ -126,6 +128,7 @@ pub async fn push_logs(
                     &custom_partition_values,
                     size,
                     schema_version,
+                    log_source,
                 )
                 .await?;
             }
@@ -137,6 +140,7 @@ pub async fn push_logs(
             time_partition_limit,
             None,
             schema_version,
+            log_source,
         )?;
         for value in data {
             parsed_timestamp = get_parsed_timestamp(&value, time_partition.as_ref().unwrap())?;
@@ -151,6 +155,7 @@ pub async fn push_logs(
                 &HashMap::new(),
                 size,
                 schema_version,
+                log_source,
             )
             .await?;
         }
@@ -161,6 +166,7 @@ pub async fn push_logs(
             time_partition_limit,
             custom_partition.as_ref(),
             schema_version,
+            log_source,
         )?;
         let custom_partition = custom_partition.unwrap();
         let custom_partition_list = custom_partition.split(',').collect::<Vec<&str>>();
@@ -181,6 +187,7 @@ pub async fn push_logs(
                 &custom_partition_values,
                 size,
                 schema_version,
+                log_source,
             )
             .await?;
         }
@@ -200,6 +207,7 @@ pub async fn create_process_record_batch(
     custom_partition_values: &HashMap<String, String>,
     origin_size: u64,
     schema_version: SchemaVersion,
+    log_source: &LogSource,
 ) -> Result<(), PostError> {
     let (rb, is_first_event) = get_stream_schema(
         stream_name,
@@ -208,7 +216,7 @@ pub async fn create_process_record_batch(
         static_schema_flag,
         time_partition,
         schema_version,
-            log_source,
+        log_source,
     )?;
     event::Event {
         rb,
@@ -234,6 +242,7 @@ pub fn get_stream_schema(
     static_schema_flag: Option<&String>,
     time_partition: Option<&String>,
     schema_version: SchemaVersion,
+    log_source: &LogSource,
 ) -> Result<(arrow_array::RecordBatch, bool), PostError> {
     let hash_map = STREAM_INFO.read().unwrap();
     let schema = hash_map
@@ -248,6 +257,7 @@ pub fn get_stream_schema(
         static_schema_flag,
         time_partition,
         schema_version,
+        log_source,
     )
 }
 

src/utils/json/mod.rs

@@ -39,16 +39,18 @@ pub fn flatten_json_body(
     validation_required: bool,
     log_source: &LogSource,
 ) -> Result<Value, anyhow::Error> {
+    // Flatten the json body only if new schema and has less than 4 levels of nesting
     let mut nested_value = if schema_version == SchemaVersion::V1
+        && !has_more_than_four_levels(&body, 1)
         && matches!(
             log_source,
             LogSource::Json | LogSource::Custom(_) | LogSource::Kinesis
         ) {
-        flatten::generic_flattening(body)?
+        let flattened_json = generic_flattening(&body)?;
+        convert_to_array(flattened_json)?
     } else {
         body
     };
-
     flatten::flatten(
         &mut nested_value,
         "_",
@@ -107,6 +109,8 @@ pub fn convert_to_string(value: &Value) -> Value {
 
 #[cfg(test)]
 mod tests {
+    use crate::event::format::LogSource;
+
     use super::flatten_json_body;
     use serde_json::json;
 
@@ -121,7 +125,8 @@ mod tests {
                 None,
                 None,
                 crate::metadata::SchemaVersion::V1,
-                false
+                false,
+                &LogSource::default()
             )
             .unwrap(),
             expected
@@ -139,7 +144,8 @@ mod tests {
                 None,
                 None,
                 crate::metadata::SchemaVersion::V1,
-                false
+                false,
+                &LogSource::default()
             )
             .unwrap(),
             expected