Allow removing multiple fields in ingest processor (#24750)

* Allow removing multiple fields in ingest processor

* Iteration 2

* Few fixes

docs/reference/ingest/ingest-node.asciidoc

@@ -1651,16 +1651,18 @@ Converts a string to its lowercase equivalent.
 
 [[remove-processor]]
 === Remove Processor
-Removes an existing field. If the field doesn't exist, an exception will be thrown.
+Removes existing fields. If one field doesn't exist, an exception will be thrown.
 
 [[remove-options]]
 .Remove Options
 [options="header"]
 |======
 | Name      | Required  | Default  | Description
-| `field`   | yes       | -        | The field to be removed
+| `field`   | yes       | -        | Fields to be removed
 |======
 
+Here is an example to remove a single field:
+
 [source,js]
 --------------------------------------------------
 {
@@ -1671,6 +1673,18 @@ Removes an existing field. If the field doesn't exist, an exception will be thro
 --------------------------------------------------
 // NOTCONSOLE
 
+To remove multiple fields, you can use the following query:
+
+[source,js]
+--------------------------------------------------
+{
+  "remove": {
+    "field": ["foo", "bar"]
+  }
+}
+--------------------------------------------------
+// NOTCONSOLE
+
 [[rename-processor]]
 === Rename Processor
 Renames an existing field. If the field doesn't exist or the new name is already used, an exception will be thrown.

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

@@ -25,7 +25,10 @@ import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.TemplateService;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * Processor that removes existing fields. Nothing happens if the field is not present.
@@ -34,20 +37,20 @@ public final class RemoveProcessor extends AbstractProcessor {
 
     public static final String TYPE = "remove";
 
-    private final TemplateService.Template field;
+    private final List<TemplateService.Template> fields;
 
-    RemoveProcessor(String tag, TemplateService.Template field) {
+    RemoveProcessor(String tag, List<TemplateService.Template> fields) {
         super(tag);
-        this.field = field;
+        this.fields = new ArrayList<>(fields);
     }
 
-    public TemplateService.Template getField() {
-        return field;
+    public List<TemplateService.Template> getFields() {
+        return fields;
     }
 
     @Override
     public void execute(IngestDocument document) {
-        document.removeField(field);
+       fields.forEach(document::removeField);
     }
 
     @Override
@@ -66,10 +69,18 @@ public final class RemoveProcessor extends AbstractProcessor {
         @Override
         public RemoveProcessor create(Map<String, Processor.Factory> registry, String processorTag,
                                       Map<String, Object> config) throws Exception {
-            String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
-            TemplateService.Template compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag,
-                "field", field, templateService);
-            return new RemoveProcessor(processorTag, compiledTemplate);
+            final List<String> fields = new ArrayList<>();
+            final Object field = ConfigurationUtils.readObject(TYPE, processorTag, config, "field");
+            if (field instanceof List) {
+                fields.addAll((List) field);
+            } else {
+                fields.add((String) field);
+            }
+
+            final List<TemplateService.Template> compiledTemplates = fields.stream()
+                .map(f -> ConfigurationUtils.compileTemplate(TYPE, processorTag, "field", f, templateService))
+                .collect(Collectors.toList());
+            return new RemoveProcessor(processorTag, compiledTemplates);
         }
     }
 }

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

@@ -25,9 +25,11 @@ import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 import org.junit.Before;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 
@@ -46,7 +48,18 @@ public class RemoveProcessorFactoryTests extends ESTestCase {
         String processorTag = randomAlphaOfLength(10);
         RemoveProcessor removeProcessor = factory.create(null, processorTag, config);
         assertThat(removeProcessor.getTag(), equalTo(processorTag));
-        assertThat(removeProcessor.getField().execute(Collections.emptyMap()), equalTo("field1"));
+        assertThat(removeProcessor.getFields().get(0).execute(Collections.emptyMap()), equalTo("field1"));
+    }
+
+    public void testCreateMultipleFields() throws Exception {
+        Map<String, Object> config = new HashMap<>();
+        config.put("field", Arrays.asList("field1", "field2"));
+        String processorTag = randomAlphaOfLength(10);
+        RemoveProcessor removeProcessor = factory.create(null, processorTag, config);
+        assertThat(removeProcessor.getTag(), equalTo(processorTag));
+        assertThat(removeProcessor.getFields().stream()
+            .map(template -> template.execute(Collections.emptyMap()))
+            .collect(Collectors.toList()), equalTo(Arrays.asList("field1", "field2")));
     }
 
     public void testCreateMissingField() throws Exception {

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

@@ -25,6 +25,7 @@ import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.TestTemplateService;
 import org.elasticsearch.test.ESTestCase;
 
+import java.util.Collections;
 import java.util.HashMap;
 
 import static org.hamcrest.Matchers.containsString;
@@ -35,7 +36,8 @@ public class RemoveProcessorTests extends ESTestCase {
     public void testRemoveFields() throws Exception {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         String field = RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument);
-        Processor processor = new RemoveProcessor(randomAlphaOfLength(10), new TestTemplateService.MockTemplate(field));
+        Processor processor = new RemoveProcessor(randomAlphaOfLength(10),
+            Collections.singletonList(new TestTemplateService.MockTemplate(field)));
         processor.execute(ingestDocument);
         assertThat(ingestDocument.hasField(field), equalTo(false));
     }
@@ -43,7 +45,8 @@ public class RemoveProcessorTests extends ESTestCase {
     public void testRemoveNonExistingField() throws Exception {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
         String fieldName = RandomDocumentPicks.randomFieldName(random());
-        Processor processor = new RemoveProcessor(randomAlphaOfLength(10), new TestTemplateService.MockTemplate(fieldName));
+        Processor processor = new RemoveProcessor(randomAlphaOfLength(10),
+            Collections.singletonList(new TestTemplateService.MockTemplate(fieldName)));
         try {
             processor.execute(ingestDocument);
             fail("remove field should have failed");