Add support for schema key aliases in query engine Parsers

rust/experimental/query_engine/engine-recordset-otlp-bridge/src/bridge.rs

@@ -339,15 +339,25 @@ fn build_log_record_schema(
 ) -> Result<(ParserMapSchema, Option<ParserMapSchema>), ParserError> {
     let mut log_record_schema = ParserMapSchema::new()
         .set_default_map_key("Attributes")
+        .with_alias("attributes", "Attributes")
         .with_key_definition("Timestamp", ParserMapKeySchema::DateTime)
+        .with_alias("time_unix_nano", "Timestamp")
         .with_key_definition("ObservedTimestamp", ParserMapKeySchema::DateTime)
+        .with_alias("observed_time_unix_nano", "ObservedTimestamp")
         .with_key_definition("SeverityNumber", ParserMapKeySchema::Integer)
+        .with_alias("severity_number", "SeverityNumber")
         .with_key_definition("SeverityText", ParserMapKeySchema::String)
+        .with_alias("severity_text", "SeverityText")
         .with_key_definition("Body", ParserMapKeySchema::Any)
+        .with_alias("body", "Body")
         .with_key_definition("TraceId", ParserMapKeySchema::Array)
+        .with_alias("trace_id", "TraceId")
         .with_key_definition("SpanId", ParserMapKeySchema::Array)
+        .with_alias("span_id", "SpanId")
         .with_key_definition("TraceFlags", ParserMapKeySchema::Integer)
-        .with_key_definition("EventName", ParserMapKeySchema::String);
+        .with_alias("flags", "TraceFlags")
+        .with_key_definition("EventName", ParserMapKeySchema::String)
+        .with_alias("event_name", "EventName");
 
     if let Some(mut attributes_schema) = attributes_schema {
         let schema = attributes_schema.get_schema_mut();
@@ -363,13 +373,17 @@ fn build_log_record_schema(
             // present in Attributes users might query with ambiguous naming.
             // For example: source | extend Body = 'something' will write to the
             // top-level field and not Attributes.
-            if let Some(removed) = schema.remove(top_level_key)
-                && &removed != top_level_key_schema
-            {
-                return Err(ParserError::SchemaError(format!(
-                    "'{top_level_key}' key cannot be declared as '{}' type",
-                    &removed
-                )));
+
+            // Check both the canonical key and all its aliases
+            for key_name in log_record_schema.get_all_key_names_for_canonical_key(top_level_key) {
+                if let Some(removed) = schema.remove(key_name.as_ref())
+                    && &removed != top_level_key_schema
+                {
+                    return Err(ParserError::SchemaError(format!(
+                        "'{key_name}' key cannot be declared as '{}' type",
+                        &removed
+                    )));
+                }
             }
         }
 
@@ -786,4 +800,127 @@ mod tests {
         assert_eq!(1, e.len());
         assert!(matches!(e[0], ParserError::SchemaError(_)));
     }
+
+    #[test]
+    fn test_parse_kql_query_with_field_aliases() {
+        let run_test = |query: &str| {
+            parse_kql_query_into_pipeline(query, None).unwrap();
+        };
+
+        // Test PascalCase canonical names
+        run_test("source | where SeverityText == 'Info'");
+        run_test("source | extend x = SeverityNumber");
+        run_test("source | project Timestamp, Body");
+
+        // Test snake_case aliases
+        run_test("source | where severity_text == 'Info'");
+        run_test("source | extend x = severity_number");
+        run_test("source | project time_unix_nano, body");
+
+        // Test mixing aliases and canonical names
+        run_test("source | where severity_text == 'Info' and SeverityNumber > 0");
+        run_test("source | extend ts = time_unix_nano, sev = SeverityText");
+
+        // Test default map key alias
+        run_test("source | where attributes.custom_field == 'value'");
+        run_test("source | where Attributes.custom_field == 'value'");
+    }
+
+    #[test]
+    fn test_attributes_schema_removes_top_level_aliases() {
+        // Test that when user provides an attributes schema with aliases
+        // of top-level fields, they get removed properly
+        let result = parse_kql_query_into_pipeline(
+            "source | extend severity_text = 'Debug'",
+            Some(
+                BridgeOptions::new().with_attributes_schema(
+                    ParserMapSchema::new()
+                        // This should be removed because severity_text is an alias for SeverityText
+                        .with_key_definition("severity_text", ParserMapKeySchema::String)
+                        .with_key_definition("custom_field", ParserMapKeySchema::Integer),
+                ),
+            ),
+        );
+
+        // Should succeed - severity_text should write to top-level field
+        assert!(result.is_ok());
+
+        // Test with canonical name too
+        let result = parse_kql_query_into_pipeline(
+            "source | extend SeverityText = 'Debug'",
+            Some(
+                BridgeOptions::new().with_attributes_schema(
+                    ParserMapSchema::new()
+                        .with_key_definition("SeverityText", ParserMapKeySchema::String)
+                        .with_key_definition("custom_field", ParserMapKeySchema::Integer),
+                ),
+            ),
+        );
+
+        assert!(result.is_ok());
+    }
+
+    #[test]
+    fn test_field_aliases_in_log_record_processing() {
+        let create_request = || {
+            ExportLogsServiceRequest::new().with_resource_logs(ResourceLogs::new().with_scope_logs(
+                ScopeLogs::new().with_log_records(vec![
+                    LogRecord::new()
+                        .with_severity_text("Info".into())
+                        .with_severity_number(9),
+                    LogRecord::new()
+                        .with_severity_text("Warning".into())
+                        .with_severity_number(13),
+                ]),
+            ))
+        };
+
+        let pipeline_alias = parse_kql_query_into_pipeline(
+            "source | where severity_text == 'Info' and severity_number == 9",
+            None,
+        )
+        .unwrap();
+        let (result_alias, _) = process_export_logs_service_request_using_pipeline(
+            None,
+            &pipeline_alias,
+            RecordSetEngineDiagnosticLevel::Error,
+            create_request(),
+        )
+        .unwrap();
+
+        let pipeline_canonical = parse_kql_query_into_pipeline(
+            "source | where SeverityText == 'Info' and SeverityNumber == 9",
+            None,
+        )
+        .unwrap();
+        let (result_canonical, _) = process_export_logs_service_request_using_pipeline(
+            None,
+            &pipeline_canonical,
+            RecordSetEngineDiagnosticLevel::Error,
+            create_request(),
+        )
+        .unwrap();
+
+        let alias_logs = &result_alias.unwrap().resource_logs[0].scope_logs[0].log_records;
+        let canonical_logs = &result_canonical.unwrap().resource_logs[0].scope_logs[0].log_records;
+
+        assert_eq!(alias_logs.len(), 1);
+        assert_eq!(canonical_logs.len(), 1);
+        assert_eq!(
+            alias_logs[0].severity_text.as_ref().unwrap().get_value(),
+            canonical_logs[0]
+                .severity_text
+                .as_ref()
+                .unwrap()
+                .get_value()
+        );
+        assert_eq!(
+            alias_logs[0].severity_number.as_ref().unwrap().get_value(),
+            canonical_logs[0]
+                .severity_number
+                .as_ref()
+                .unwrap()
+                .get_value()
+        );
+    }
 }