fix: bugs around ingestion and OTel JSON (#1788)

* fix: bugs around ingestion and OTel JSON

1. remove restriction of dataset field count limit, default still remains 1000
2. infer timestamp to false in stats calculation
3. keep RFC 2822 value as Utf8 because arrow JSON decoder rejects and fails with error -
`Request: Failed to create recordbatch due to JsonError("whilst decoding field 'zeek.sip.date': failed to parse \"Tue, 22 Sep 2026 08:04:47 GMT\" as Timestamp(ms): Parser error: Error parsing timestamp from 'Tue, 22 Sep 2026 08:04:47 GMT': error parsing date")`

4. try deserialize an OTLP/JSON payload and optionally normalise by omitting empty `values` fields
in `ArrayValue` and `KeyValuelist`
5. gate this behaviour via env `P_NORMALISE_OTEL_PAYLOAD` default to false

* size>0 check

Author: nikhilsinhaparseable

src/cli.rs

@@ -618,6 +618,14 @@ pub struct Options {
     )]
     pub max_event_payload_size: usize,
 
+    #[arg(
+        long,
+        env = "P_NORMALISE_OTEL_PAYLOAD",
+        default_value = "false",
+        help = "Normalise missing OTLP collection values and retry when JSON deserialization fails"
+    )]
+    pub normalise_otel_payload: bool,
+
     // the duration during which local sync should be completed
     #[arg(
         long,

src/event/format/mod.rs

@@ -24,6 +24,7 @@ use std::{
 };
 
 use anyhow::{Error as AnyError, anyhow};
+use arrow::compute::kernels::cast_utils::string_to_datetime;
 use arrow_array::RecordBatch;
 use arrow_schema::{DataType, Field, Schema, TimeUnit};
 use chrono::{DateTime, Utc};
@@ -58,6 +59,15 @@ static TIME_FIELD_NAME_PARTS: [&str; 11] = [
     "dt",
 ];
 
+/// Returns whether Arrow's JSON timestamp decoder can parse this value.
+///
+/// Timestamp inference must use the same parser as decoding. Chrono accepts
+/// formats such as RFC 2822 that Arrow rejects, which otherwise produces a
+/// schema Arrow cannot use to decode the original JSON value.
+fn is_arrow_timestamp(value: &str) -> bool {
+    string_to_datetime(&Utc, value).is_ok()
+}
+
 type EventSchema = Vec<Arc<Field>>;
 
 /// Normalizes a field name by replacing leading '@' with '_'.
@@ -388,7 +398,8 @@ pub fn override_data_type(
                             .any(|part| field_name.to_lowercase().contains(part))
                         && field.data_type() == &DataType::Utf8
                         && (DateTime::parse_from_rfc3339(s).is_ok()
-                            || DateTime::parse_from_rfc2822(s).is_ok()) =>
+                            || DateTime::parse_from_rfc2822(s).is_ok())
+                        && is_arrow_timestamp(s) =>
                 {
                     Field::new(
                         field_name,
@@ -456,12 +467,13 @@ fn value_compatible_with_type(
             // Timestamps can accept strings that parse as datetime or numbers
             match value {
                 Value::String(s) => {
-                    chrono::DateTime::parse_from_rfc3339(s).is_ok()
+                    (chrono::DateTime::parse_from_rfc3339(s).is_ok()
                         || chrono::DateTime::parse_from_rfc2822(s).is_ok()
                         || chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S%.f").is_ok()
                         || chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%dT%H:%M:%S").is_ok()
                         || chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S%.f").is_ok()
-                        || chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S").is_ok()
+                        || chrono::NaiveDateTime::parse_from_str(s, "%Y-%m-%d %H:%M:%S").is_ok())
+                        && is_arrow_timestamp(s)
                 }
                 // Arrow JSON decoder with coerce_primitive(false) cannot decode
                 // a JSON number into a Timestamp field — it expects a string.
@@ -941,6 +953,31 @@ mod tests {
         assert!(conflicts.is_empty());
     }
 
+    #[test]
+    fn test_detect_schema_conflicts_rfc2822_timestamp() {
+        let mut existing_schema: HashMap<String, Arc<Field>> = HashMap::new();
+        existing_schema.insert(
+            "date".to_string(),
+            Arc::new(Field::new(
+                "date",
+                DataType::Timestamp(TimeUnit::Millisecond, None),
+                true,
+            )),
+        );
+
+        let inferred_schema = Schema::new(vec![Field::new("date", DataType::Utf8, true)]);
+        let values = vec![json!({"date": "Tue, 22 Sep 2026 08:04:47 GMT"})];
+
+        let conflicts = detect_schema_conflicts(
+            &inferred_schema,
+            &existing_schema,
+            &values,
+            SchemaVersion::V1,
+        );
+
+        assert_eq!(conflicts.get("date"), Some(&"date_utf8".to_string()));
+    }
+
     #[test]
     fn test_detect_schema_conflicts_number_to_utf8() {
         // Existing schema has request_body as Utf8
@@ -1062,6 +1099,16 @@ mod tests {
         assert_eq!(updated.field(0).data_type(), &DataType::Utf8);
     }
 
+    #[test]
+    fn override_data_type_keeps_rfc2822_as_utf8() {
+        let inferred = Arc::new(Schema::new(vec![Field::new("date", DataType::Utf8, true)]));
+        let log = json!({"date": "Tue, 22 Sep 2026 08:04:47 GMT"});
+
+        let updated = override_data_type(inferred, log, SchemaVersion::V1, true);
+
+        assert_eq!(updated.field(0).data_type(), &DataType::Utf8);
+    }
+
     #[test]
     fn override_data_type_still_casts_numbers_when_infer_disabled() {
         // The flag must only affect string-to-timestamp inference, not

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

@@ -41,7 +41,10 @@ use crate::{
             kinesis::{Message, flatten_kinesis_logs},
         },
     },
-    otel::{logs::flatten_otel_logs, metrics::flatten_otel_metrics, traces::flatten_otel_traces},
+    otel::{
+        logs::flatten_otel_logs, metrics::flatten_otel_metrics, otel_utils::deserialize_otel_json,
+        traces::flatten_otel_traces,
+    },
     parseable::{DEFAULT_TENANT, PARSEABLE},
     storage::StreamType,
     utils::{
@@ -223,7 +226,8 @@ pub fn flatten_and_push_logs(
             )?;
         }
         LogSource::OtelLogs => {
-            let logs: LogsData = serde_json::from_value(json)?;
+            let logs: LogsData =
+                deserialize_otel_json(json, PARSEABLE.options.normalise_otel_payload)?;
             let records = flatten_otel_logs(&logs, tenant_str);
             if !records.is_empty() {
                 push_logs(
@@ -238,7 +242,8 @@ pub fn flatten_and_push_logs(
             }
         }
         LogSource::OtelTraces => {
-            let traces: TracesData = serde_json::from_value(json)?;
+            let traces: TracesData =
+                deserialize_otel_json(json, PARSEABLE.options.normalise_otel_payload)?;
             let records = flatten_otel_traces(&traces, tenant_str);
             if !records.is_empty() {
                 push_logs(
@@ -253,7 +258,8 @@ pub fn flatten_and_push_logs(
             }
         }
         LogSource::OtelMetrics => {
-            let metrics: MetricsData = serde_json::from_value(json)?;
+            let metrics: MetricsData =
+                deserialize_otel_json(json, PARSEABLE.options.normalise_otel_payload)?;
             let records = flatten_otel_metrics(metrics, tenant_str);
             if !records.is_empty() {
                 push_logs(

src/option.rs

@@ -87,16 +87,14 @@ impl From<Compression> for parquet::basic::Compression {
 }
 
 pub mod validation {
+    use super::{Compression, Mode};
+    use path_clean::PathClean;
     use std::{
         env, io,
         net::ToSocketAddrs,
         path::{Path, PathBuf},
     };
 
-    use super::{Compression, Mode};
-    use crate::cli::DATASET_FIELD_COUNT_LIMIT;
-    use path_clean::PathClean;
-
     pub fn file_path(s: &str) -> Result<PathBuf, String> {
         if s.is_empty() {
             return Err("empty path".to_owned());
@@ -196,16 +194,15 @@ pub mod validation {
         }
     }
     pub fn validate_dataset_fields_allowed_limit(s: &str) -> Result<usize, String> {
-        if let Ok(size) = s.parse::<usize>() {
-            if (1..=DATASET_FIELD_COUNT_LIMIT).contains(&size) {
-                Ok(size)
-            } else {
-                Err(format!(
-                    "Invalid value for P_DATASET_FIELD_COUNT_LIMIT. It should be between 1 and {DATASET_FIELD_COUNT_LIMIT}"
-                ))
-            }
+        if let Ok(size) = s.parse::<usize>()
+            && size > 0
+        {
+            Ok(size)
         } else {
-            Err("Invalid value for P_DATASET_FIELD_COUNT_LIMIT. It should be given as integer value".to_string())
+            Err(
+                "Invalid value for P_DATASET_FIELD_COUNT_LIMIT. It should be a positive integer"
+                    .to_string(),
+            )
         }
     }
 

src/otel/otel_utils.rs

@@ -20,8 +20,65 @@ use chrono::DateTime;
 use opentelemetry_proto::tonic::common::v1::{
     AnyValue, ArrayValue, KeyValue, KeyValueList, any_value::Value as OtelValue,
 };
+use serde::de::DeserializeOwned;
 use serde_json::{Map, Value};
 
+/// Deserializes an OTLP/JSON payload and optionally retries after restoring
+/// omitted empty `values` fields in `ArrayValue` and `KeyValueList` messages.
+pub fn deserialize_otel_json<T>(
+    mut json: Value,
+    normalise_on_failure: bool,
+) -> Result<T, serde_json::Error>
+where
+    T: DeserializeOwned,
+{
+    if !normalise_on_failure {
+        return serde_json::from_value(json);
+    }
+
+    match T::deserialize(&json) {
+        Ok(value) => Ok(value),
+        Err(original_error) => {
+            if !normalise_missing_otel_collection_values(&mut json) {
+                return Err(original_error);
+            }
+
+            serde_json::from_value(json)
+        }
+    }
+}
+
+fn normalise_missing_otel_collection_values(value: &mut Value) -> bool {
+    match value {
+        Value::Object(map) => {
+            let mut changed = false;
+
+            for collection_key in ["arrayValue", "kvlistValue"] {
+                if let Some(Value::Object(collection)) = map.get_mut(collection_key)
+                    && !collection.contains_key("values")
+                {
+                    collection.insert("values".to_string(), Value::Array(Vec::new()));
+                    changed = true;
+                }
+            }
+
+            for child in map.values_mut() {
+                changed |= normalise_missing_otel_collection_values(child);
+            }
+
+            changed
+        }
+        Value::Array(values) => {
+            let mut changed = false;
+            for child in values {
+                changed |= normalise_missing_otel_collection_values(child);
+            }
+            changed
+        }
+        _ => false,
+    }
+}
+
 // Value can be one of types - String, Bool, Int, Double, ArrayValue, AnyValue, KeyValueList, Byte
 pub fn collect_json_from_value(key: &str, value: OtelValue) -> Map<String, Value> {
     let mut value_json: Map<String, Value> = Map::new();
@@ -216,6 +273,7 @@ pub fn convert_epoch_nano_to_timestamp(epoch_ns: i64) -> String {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use opentelemetry_proto::tonic::logs::v1::LogsData;
 
     #[test]
     fn zero_epoch_uses_current_utc_time() {
@@ -237,4 +295,58 @@ mod tests {
             "1970-01-01T00:00:01.000000000Z"
         );
     }
+
+    fn logs_with_omitted_empty_collection_values() -> Value {
+        serde_json::json!({
+            "resourceLogs": [{
+                "scopeLogs": [{
+                    "logRecords": [{
+                        "attributes": [
+                            {
+                                "key": "input",
+                                "value": { "kvlistValue": {} }
+                            },
+                            {
+                                "key": "tags",
+                                "value": { "arrayValue": {} }
+                            }
+                        ]
+                    }]
+                }]
+            }]
+        })
+    }
+
+    #[test]
+    fn otel_payload_normalisation_is_disabled_by_default() {
+        let error =
+            deserialize_otel_json::<LogsData>(logs_with_omitted_empty_collection_values(), false)
+                .unwrap_err();
+
+        assert!(error.to_string().contains("missing field"));
+        assert!(error.to_string().contains("values"));
+    }
+
+    #[test]
+    fn otel_payload_normalisation_retries_failed_deserialization() {
+        let logs =
+            deserialize_otel_json::<LogsData>(logs_with_omitted_empty_collection_values(), true)
+                .unwrap();
+
+        let attributes = &logs.resource_logs[0].scope_logs[0].log_records[0].attributes;
+        assert!(matches!(
+            attributes[0]
+                .value
+                .as_ref()
+                .and_then(|value| value.value.as_ref()),
+            Some(OtelValue::KvlistValue(KeyValueList { values })) if values.is_empty()
+        ));
+        assert!(matches!(
+            attributes[1]
+                .value
+                .as_ref()
+                .and_then(|value| value.value.as_ref()),
+            Some(OtelValue::ArrayValue(ArrayValue { values })) if values.is_empty()
+        ));
+    }
 }

src/storage/field_stats.rs

@@ -203,7 +203,7 @@ async fn calculate_field_stats_inner(
             &p_custom_fields,
             TelemetryType::Logs,
             tenant_id,
-            true,
+            false,
         )?
         .process()?;
     }