Support dotted field notations in the reroute processor

docs/changelog/96243.yaml

@@ -0,0 +1,5 @@
+pr: 96243
+summary: Support dotted field notations in the reroute processor
+area: Ingest Node
+type: bug
+issues: []

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/RerouteProcessor.java

@@ -94,10 +94,15 @@ public IngestDocument execute(IngestDocument ingestDocument) throws Exception {
         ingestDocument.setFieldValue(DATA_STREAM_TYPE, type);
         ingestDocument.setFieldValue(DATA_STREAM_DATASET, dataset);
         ingestDocument.setFieldValue(DATA_STREAM_NAMESPACE, namespace);
-        if (ingestDocument.hasField(EVENT_DATASET)) {
+        // remove fields with dotted field names
+        ingestDocument.getCtxMap().remove(DATA_STREAM_TYPE);
+        ingestDocument.getCtxMap().remove(DATA_STREAM_DATASET);
+        ingestDocument.getCtxMap().remove(DATA_STREAM_NAMESPACE);
+        if (ingestDocument.getCtxMap().containsKey(EVENT_DATASET) || ingestDocument.hasField(EVENT_DATASET)) {
             // ECS specifies that "event.dataset should have the same value as data_stream.dataset"
             // not eagerly set event.dataset but only if the doc contains it already to ensure it's consistent with data_stream.dataset
             ingestDocument.setFieldValue(EVENT_DATASET, dataset);
+            ingestDocument.getCtxMap().remove(EVENT_DATASET);
         }
         return ingestDocument;
     }
@@ -253,10 +258,33 @@ private DataStreamValueSource(String value, Function<String, String> sanitizer)
         @Nullable
         public String resolve(IngestDocument ingestDocument) {
             if (fieldReference != null) {
-                return sanitizer.apply(ingestDocument.getFieldValue(fieldReference, String.class, true));
+                String value = ingestDocument.getFieldValue(fieldReference, String.class, true);
+                if (value == null) {
+                    value = getStringFieldValueInDottedNotation(ingestDocument);
+                }
+                return sanitizer.apply(value);
             } else {
                 return value;
             }
         }
+
+        private String getStringFieldValueInDottedNotation(IngestDocument ingestDocument) {
+            String value = null;
+            Object valueObject = ingestDocument.getCtxMap().get(fieldReference);
+            if (valueObject instanceof String) {
+                value = (String) valueObject;
+            } else if (valueObject != null) {
+                throw new IllegalArgumentException(
+                    "field ["
+                        + fieldReference
+                        + "] of type ["
+                        + valueObject.getClass().getName()
+                        + "] cannot be cast to ["
+                        + String.class.getName()
+                        + "]"
+                );
+            }
+            return value;
+        }
     }
 }

modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/RerouteProcessorTests.java

@@ -41,6 +41,17 @@ public void testEventDataset() throws Exception {
         assertThat(ingestDocument.getFieldValue("event.dataset", String.class), equalTo("foo"));
     }
 
+    public void testEventDatasetDottedFieldName() throws Exception {
+        IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
+        ingestDocument.getCtxMap().put("event.dataset", "foo");
+
+        RerouteProcessor processor = createRerouteProcessor(List.of("{{event.dataset}}"), List.of());
+        processor.execute(ingestDocument);
+        assertDataSetFields(ingestDocument, "logs", "foo", "default");
+        assertNull(ingestDocument.getCtxMap().get("event.dataset"));
+        assertThat(ingestDocument.getFieldValue("event.dataset", String.class), equalTo("foo"));
+    }
+
     public void testNoDataset() throws Exception {
         IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
         ingestDocument.setFieldValue("ds", "foo");
@@ -81,6 +92,16 @@ public void testDataStreamFieldsFromDocument() throws Exception {
         assertDataSetFields(ingestDocument, "logs", "foo", "bar");
     }
 
+    public void testDataStreamFieldsFromDocumentDottedNotation() throws Exception {
+        IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
+        ingestDocument.getCtxMap().put("data_stream.dataset", "foo");
+        ingestDocument.getCtxMap().put("data_stream.namespace", "bar");
+
+        RerouteProcessor processor = createRerouteProcessor(List.of(), List.of());
+        processor.execute(ingestDocument);
+        assertDataSetFields(ingestDocument, "logs", "foo", "bar");
+    }
+
     public void testInvalidDataStreamFieldsFromDocument() throws Exception {
         IngestDocument ingestDocument = createIngestDocument("logs-generic-default");
         ingestDocument.setFieldValue("data_stream.dataset", "foo-bar");
@@ -253,6 +274,9 @@ private void assertDataSetFields(IngestDocument ingestDocument, String type, Str
         assertThat(ingestDocument.getFieldValue("data_stream.type", String.class), equalTo(type));
         assertThat(ingestDocument.getFieldValue("data_stream.dataset", String.class), equalTo(dataset));
         assertThat(ingestDocument.getFieldValue("data_stream.namespace", String.class), equalTo(namespace));
+        assertNull(ingestDocument.getCtxMap().get("data_stream.type"));
+        assertNull(ingestDocument.getCtxMap().get("data_stream.dataset"));
+        assertNull(ingestDocument.getCtxMap().get("data_stream.namespace"));
         assertThat(ingestDocument.getFieldValue("_index", String.class), equalTo(type + "-" + dataset + "-" + namespace));
         if (ingestDocument.hasField("event.dataset")) {
             assertThat(