add copy ingest processor

docs/reference/ingest/ingest-node.asciidoc

@@ -824,6 +824,7 @@ include::processors/append.asciidoc[]
 include::processors/bytes.asciidoc[]
 include::processors/circle.asciidoc[]
 include::processors/convert.asciidoc[]
+include::processors/copy.asciidoc[]
 include::processors/csv.asciidoc[]
 include::processors/date.asciidoc[]
 include::processors/date-index-name.asciidoc[]

docs/reference/ingest/processors/copy.asciidoc

@@ -0,0 +1,82 @@
+[[copy-processor]]
+=== COPY Processor
+Copy the value of an existed field to other field or the root of the document.
+
+[[copy-options]]
+.Copy Options
+[options="header"]
+|======
+| Name             | Required  | Default    | Description
+| `field`          | yes       | -          | The field copied from
+| `target_field`   | no        | -          | The field copied to, `add_to_root` cannot be set to `true` when this field is set
+| `add_to_root`    | no        | `false`    | Flag that copy the value of `field` to the top level of the document. `target_field` must not be set when this option is chosen.
+| `ignore_missing` | no        | `false`    | If `true` and `field` does not exist or is `null`, the processor quietly exits without modifying the document
+include::common-options.asciidoc[]
+|======
+
+The supported types which can be copied are `boolean`, `number`, `array`, `object`, `string`, `date`.
+
+We can copy the sub-field of an object to another field:
+
+[source,js]
+--------------------------------------------------
+{
+  "copy" : {
+    "field" : "foo.bar",
+    "target_field" : "zoo"
+  }
+}
+--------------------------------------------------
+// NOTCONSOLE
+
+If the following document is processed:
+
+[source,js]
+--------------------------------------------------
+{
+  "foo": {
+    "bar": 1
+  }
+}
+--------------------------------------------------
+// NOTCONSOLE
+
+after the `copy` processor operates on it, it will look like:
+
+[source,js]
+--------------------------------------------------
+{
+  "foo": {
+    "bar": 1
+  },
+  "zoo": 1
+}
+--------------------------------------------------
+// NOTCONSOLE
+
+We can set `add_to_root` to `true` if we want to copy one field to the top level of the document:
+[source,js]
+--------------------------------------------------
+{
+  "copy" : {
+    "field" : "foo",
+    "add_to_root": true
+  }
+}
+--------------------------------------------------
+// NOTCONSOLE
+
+then after the `copy` processor operates on the document above, it will look like:
+
+[source,js]
+--------------------------------------------------
+{
+  "foo": {
+    "bar": 1
+  },
+  "bar": 1
+}
+--------------------------------------------------
+// NOTCONSOLE
+
+Note that we can only copy an `object` field to the root of the document.

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

@@ -0,0 +1,132 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest.common;
+
+import org.elasticsearch.ingest.AbstractProcessor;
+import org.elasticsearch.ingest.ConfigurationUtils;
+import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.Processor;
+
+import java.util.Map;
+
+import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
+
+/**
+ * Processor that adds a new field with value copied from other existed field.
+ */
+public class CopyProcessor extends AbstractProcessor {
+
+    public static final String TYPE = "copy";
+
+    private final String field;
+    private final String targetField;
+    private final boolean addToRoot;
+    private final boolean ignoreMissing;
+
+    CopyProcessor(String tag, String field, String targetField, boolean addToRoot, boolean ignoreMissing)  {
+        super(tag);
+        this.field = field;
+        this.targetField = targetField;
+        this.addToRoot = addToRoot;
+        this.ignoreMissing = ignoreMissing;
+    }
+
+    public String getField() {
+        return field;
+    }
+
+    public String getTargetField() {
+        return targetField;
+    }
+
+    public boolean isAddToRoot() {
+        return addToRoot;
+    }
+
+    public boolean isIgnoreMissing() {
+        return ignoreMissing;
+    }
+
+    private static void apply(Map<String, Object> ctx, Object value) {
+        if (value instanceof Map) {
+            @SuppressWarnings("unchecked")
+            Map<String, Object> map = (Map<String, Object>) value;
+            ctx.putAll(map);
+        } else {
+            throw new IllegalArgumentException("cannot add non-map fields to root of document");
+        }
+    }
+
+    @Override
+    public IngestDocument execute(IngestDocument document) {
+        Object fieldValue = document.getFieldValue(field, Object.class, ignoreMissing);
+
+        if (fieldValue == null && ignoreMissing) {
+            return document;
+        } else if (fieldValue == null) {
+            throw new IllegalArgumentException("field [" + field + "] is null, cannot be copied");
+        }
+
+        if (field.equals(targetField)) {
+            return document;
+        }
+
+        if (addToRoot) {
+            apply(document.getSourceAndMetadata(), fieldValue);
+        } else {
+            document.setFieldValue(targetField, IngestDocument.deepCopy(fieldValue));
+        }
+
+        return document;
+    }
+
+    @Override
+    public String getType() {
+        return TYPE;
+    }
+
+    public static final class Factory implements Processor.Factory {
+        @Override
+        public CopyProcessor create(
+            Map<String, Processor.Factory> registry, String processorTag,
+            Map<String, Object> config) throws Exception {
+            String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
+            String targetField = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "target_field");
+            boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_missing", false);
+            boolean addToRoot = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "add_to_root", false);
+
+            if (addToRoot == false && targetField == null) {
+                throw newConfigurationException(TYPE, processorTag, "target_field",
+                    "either `target_field` or `add_to_root` must be set");
+            }
+            if (addToRoot && targetField != null) {
+                throw newConfigurationException(TYPE, processorTag, "target_field",
+                    "cannot set `target_field` while also setting `add_to_root` to true");
+            }
+
+            return new CopyProcessor(
+                processorTag,
+                field,
+                targetField,
+                addToRoot,
+                ignoreMissing);
+        }
+    }
+}

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

@@ -89,7 +89,8 @@ public Map<String, Processor.Factory> getProcessors(Processor.Parameters paramet
                 entry(DissectProcessor.TYPE, new DissectProcessor.Factory()),
                 entry(DropProcessor.TYPE, new DropProcessor.Factory()),
                 entry(HtmlStripProcessor.TYPE, new HtmlStripProcessor.Factory()),
-                entry(CsvProcessor.TYPE, new CsvProcessor.Factory()));
+                entry(CsvProcessor.TYPE, new CsvProcessor.Factory()),
+                entry(CopyProcessor.TYPE, new CopyProcessor.Factory()));
     }
 
     @Override

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

@@ -0,0 +1,102 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.ingest.common;
+
+import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.test.ESTestCase;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+
+public class CopyProcessorFactoryTests extends ESTestCase {
+
+    private static final CopyProcessor.Factory FACTORY = new CopyProcessor.Factory();
+
+    public void testCreate() throws Exception {
+        String processorTag = randomAlphaOfLength(10);
+        String randomField = randomAlphaOfLength(10);
+        String randomTargetField = randomAlphaOfLength(5);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        config.put("target_field", randomTargetField);
+        CopyProcessor copyProcessor = FACTORY.create(null, processorTag, config);
+        assertThat(copyProcessor.getTag(), equalTo(processorTag));
+        assertThat(copyProcessor.getField(), equalTo(randomField));
+        assertThat(copyProcessor.getTargetField(), equalTo(randomTargetField));
+    }
+
+    public void testCreateWithAddToRoot() throws Exception {
+        String processorTag = randomAlphaOfLength(10);
+        String randomField = randomAlphaOfLength(10);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        config.put("add_to_root", true);
+        CopyProcessor copyProcessor = FACTORY.create(null, processorTag, config);
+        assertThat(copyProcessor.getTag(), equalTo(processorTag));
+        assertThat(copyProcessor.getField(), equalTo(randomField));
+        assertTrue(copyProcessor.isAddToRoot());
+    }
+
+    public void testCreateWithMissingField() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        String processorTag = randomAlphaOfLength(10);
+        ElasticsearchException exception = expectThrows(
+            ElasticsearchParseException.class,
+            () -> FACTORY.create(null, processorTag, config));
+        assertThat(exception.getMessage(), equalTo("[field] required property is missing"));
+    }
+
+    public void testCreateWithIgnoreMissing() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", "old_field");
+        config.put("target_field", "new_field");
+        config.put("ignore_missing", true);
+        String processorTag = randomAlphaOfLength(10);
+        CopyProcessor copyProcessor = FACTORY.create(null, processorTag, config);
+        assertThat(copyProcessor.getTag(), equalTo(processorTag));
+        assertThat(copyProcessor.getField(), equalTo("old_field"));
+        assertThat(copyProcessor.getTargetField(), equalTo("new_field"));
+        assertThat(copyProcessor.isIgnoreMissing(), equalTo(true));
+    }
+
+    public void testCreateWithoutTargetFieldAndAddToRoot() throws Exception {
+        String randomField = randomAlphaOfLength(10);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        ElasticsearchException exception = expectThrows(ElasticsearchParseException.class,
+            () -> FACTORY.create(null, randomAlphaOfLength(10), config));
+        assertThat(exception.getMessage(), equalTo("[target_field] either `target_field` or `add_to_root` must be set"));
+    }
+
+    public void testCreateWithBothTargetFieldAndAddToRoot() throws Exception {
+        String randomField = randomAlphaOfLength(10);
+        String randomTargetField = randomAlphaOfLength(5);
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", randomField);
+        config.put("target_field", randomTargetField);
+        config.put("add_to_root", true);
+        ElasticsearchException exception = expectThrows(ElasticsearchParseException.class,
+            () -> FACTORY.create(null, randomAlphaOfLength(10), config));
+        assertThat(exception.getMessage(), equalTo("[target_field] cannot set `target_field` while also setting `add_to_root` to true"));
+    }
+}