Fix JsonProcessor nested field access

Author: jbaiera

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

@@ -149,6 +149,10 @@ public static void apply(
         boolean strictJsonParsing
     ) {
         Object value = apply(ctx.get(fieldName), allowDuplicateKeys, strictJsonParsing);
+        mergeParsedJson(ctx, value, conflictStrategy);
+    }
+
+    private static void mergeParsedJson(Map<String, Object> ctx, Object value, ConflictStrategy conflictStrategy) {
         if (value instanceof Map) {
             @SuppressWarnings("unchecked")
             Map<String, Object> map = (Map<String, Object>) value;
@@ -185,7 +189,8 @@ public static void recursiveMerge(Map<String, Object> target, Map<String, Object
     @Override
     public IngestDocument execute(IngestDocument document) throws Exception {
         if (addToRoot) {
-            apply(document.getSourceAndMetadata(), field, allowDuplicateKeys, addToRootConflictStrategy, strictJsonParsing);
+            Object value = apply(document.getFieldValue(field, Object.class), allowDuplicateKeys, strictJsonParsing);
+            mergeParsedJson(document.getSourceAndMetadata(), value, addToRootConflictStrategy);
         } else {
             document.setFieldValue(targetField, apply(document.getFieldValue(field, Object.class), allowDuplicateKeys, strictJsonParsing));
         }

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

@@ -12,6 +12,8 @@
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestPipelineFieldAccessPattern;
+import org.elasticsearch.ingest.IngestPipelineTestUtils;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentBuilder;
@@ -166,6 +168,28 @@ public void testAddToRoot() throws Exception {
         assertEquals("see", sourceAndMetadata.get("c"));
     }
 
+    public void testAddToRootNestedField() throws Exception {
+        String processorTag = randomAlphaOfLength(3);
+        String randomTargetField = randomAlphaOfLength(2);
+        JsonProcessor jsonProcessor = new JsonProcessor(processorTag, null, "a.b", randomTargetField, true, REPLACE, false);
+
+        String json = "{\"a\": 1, \"b\": 2}";
+        Map<String, Object> subfield = new HashMap<>();
+        subfield.put("b", json);
+
+        Map<String, Object> document = new HashMap<>();
+        document.put("a", subfield);
+        document.put("c", "see");
+
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
+        ingestDocument = IngestPipelineTestUtils.runWithRandomAccessPattern(ingestDocument, jsonProcessor);
+
+        Map<String, Object> sourceAndMetadata = ingestDocument.getSourceAndMetadata();
+        assertEquals(1, sourceAndMetadata.get("a"));
+        assertEquals(2, sourceAndMetadata.get("b"));
+        assertEquals("see", sourceAndMetadata.get("c"));
+    }
+
     public void testDuplicateKeys() throws Exception {
         String processorTag = randomAlphaOfLength(3);
         JsonProcessor lenientJsonProcessor = new JsonProcessor(processorTag, null, "a", null, true, REPLACE, true);