Add copy_from option to the Append processor (#132003)

Author: chrisberkhout

docs/changelog/132003.yaml

@@ -0,0 +1,5 @@
+pr: 132003
+summary: Add `copy_from` option to the Append processor
+area: Ingest Node
+type: enhancement
+issues: []

docs/reference/enrich-processor/append-processor.md

@@ -14,9 +14,10 @@ $$$append-options$$$
 | Name | Required | Default | Description |
 | --- | --- | --- | --- |
 | `field` | yes | - | The field to be appended to. Supports [template snippets](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#template-snippets). |
-| `value` | yes | - | The value to be appended. Supports [template snippets](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#template-snippets). |
+| `value` | yes* | - | The value to be appended. Supports [template snippets](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#template-snippets). May specify only one of `value` or `copy_from`. |
+| `copy_from` {applies_to}`stack: ga 9.2.0` | no | - | The origin field which will be appended to `field`, cannot set `value` simultaneously. |
 | `allow_duplicates` | no | true | If `false`, the processor does not appendvalues already present in the field. |
-| `media_type` | no | `application/json` | The media type for encoding `value`. Applies only when `value` is a[template snippet](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#template-snippets). Must be one of `application/json`, `text/plain`, or`application/x-www-form-urlencoded`. |
+| `media_type` | no | `application/json` | The media type for encoding `value`. Applies only when `value` is a [template snippet](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#template-snippets). Must be one of `application/json`, `text/plain`, or`application/x-www-form-urlencoded`. |
 | `description` | no | - | Description of the processor. Useful for describing the purpose of the processor or its configuration. |
 | `if` | no | - | Conditionally execute the processor. See [Conditionally run a processor](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#conditionally-run-processor). |
 | `ignore_failure` | no | `false` | Ignore failures for the processor. See [Handling pipeline failures](docs-content://manage-data/ingest/transform-enrich/ingest-pipelines.md#handling-pipeline-failures). |

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

@@ -21,6 +21,8 @@
 
 import java.util.Map;
 
+import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
+
 /**
  * Processor that appends value or values to existing lists. If the field is not present a new list holding the
  * provided values will be added. If the field is a scalar it will be converted to a single item list and the provided
@@ -32,12 +34,21 @@ public final class AppendProcessor extends AbstractProcessor {
 
     private final TemplateScript.Factory field;
     private final ValueSource value;
+    private final String copyFrom;
     private final boolean allowDuplicates;
 
-    AppendProcessor(String tag, String description, TemplateScript.Factory field, ValueSource value, boolean allowDuplicates) {
+    AppendProcessor(
+        String tag,
+        String description,
+        TemplateScript.Factory field,
+        ValueSource value,
+        String copyFrom,
+        boolean allowDuplicates
+    ) {
         super(tag, description);
         this.field = field;
         this.value = value;
+        this.copyFrom = copyFrom;
         this.allowDuplicates = allowDuplicates;
     }
 
@@ -49,10 +60,19 @@ public ValueSource getValue() {
         return value;
     }
 
+    public String getCopyFrom() {
+        return copyFrom;
+    }
+
     @Override
     public IngestDocument execute(IngestDocument document) throws Exception {
         String path = document.renderTemplate(field);
-        document.appendFieldValue(path, value, allowDuplicates);
+        if (copyFrom != null) {
+            Object fieldValue = document.getFieldValue(copyFrom, Object.class);
+            document.appendFieldValue(path, IngestDocument.deepCopy(fieldValue), allowDuplicates);
+        } else {
+            document.appendFieldValue(path, value, allowDuplicates);
+        }
         return document;
     }
 
@@ -78,17 +98,27 @@ public AppendProcessor create(
             ProjectId projectId
         ) throws Exception {
             String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
-            Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
+            String copyFrom = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "copy_from");
+            String mediaType = ConfigurationUtils.readMediaTypeProperty(TYPE, processorTag, config, "media_type", "application/json");
+            ValueSource valueSource = null;
+            if (copyFrom == null) {
+                Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
+                valueSource = ValueSource.wrap(value, scriptService, Map.of(Script.CONTENT_TYPE_OPTION, mediaType));
+            } else {
+                Object value = config.remove("value");
+                if (value != null) {
+                    throw newConfigurationException(
+                        TYPE,
+                        processorTag,
+                        "copy_from",
+                        "cannot set both `copy_from` and `value` in the same processor"
+                    );
+                }
+            }
             boolean allowDuplicates = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "allow_duplicates", true);
             TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag, "field", field, scriptService);
-            String mediaType = ConfigurationUtils.readMediaTypeProperty(TYPE, processorTag, config, "media_type", "application/json");
-            return new AppendProcessor(
-                processorTag,
-                description,
-                compiledTemplate,
-                ValueSource.wrap(value, scriptService, Map.of(Script.CONTENT_TYPE_OPTION, mediaType)),
-                allowDuplicates
-            );
+
+            return new AppendProcessor(processorTag, description, compiledTemplate, valueSource, copyFrom, allowDuplicates);
         }
     }
 }

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

@@ -123,4 +123,28 @@ public void testMediaType() throws Exception {
         );
         assertThat(e.getMessage(), containsString("property does not contain a supported media type [" + expectedMediaType + "]"));
     }
+
+    public void testCreateWithCopyFrom() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "field1");
+        config.put("copy_from", "field2");
+        String processorTag = randomAlphaOfLength(10);
+        AppendProcessor appendProcessor = factory.create(null, processorTag, null, config, null);
+        assertThat(appendProcessor.getTag(), equalTo(processorTag));
+        assertThat(appendProcessor.getField().newInstance(Map.of()).execute(), equalTo("field1"));
+        assertThat(appendProcessor.getCopyFrom(), equalTo("field2"));
+    }
+
+    public void testCreateWithCopyFromAndValue() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "field1");
+        config.put("copy_from", "field2");
+        config.put("value", "value1");
+        String processorTag = randomAlphaOfLength(10);
+        ElasticsearchException exception = expectThrows(
+            ElasticsearchException.class,
+            () -> factory.create(null, processorTag, null, config, null)
+        );
+        assertThat(exception.getMessage(), equalTo("[copy_from] cannot set both `copy_from` and `value` in the same processor"));
+    }
 }