Skip to content

Commit 13e1cf6

Browse files
original-brownbearoriginal-brownbear
authored
ingest: Add ignore_missing property to foreach filter (#22147) (#31578)
1 parent 26a927a commit 13e1cf6

File tree

4 files changed

+70
-23
lines changed

4 files changed

+70
-23
lines changed

‎docs/reference/ingest/ingest-node.asciidoc

+4-3
Original file line numberDiff line numberDiff line change
@@ -1075,9 +1075,10 @@ then it aborts the execution and leaves the array unmodified.
10751075
.Foreach Options
10761076
[options="header"]
10771077
|======
1078-
| Name | Required | Default | Description
1079-
| `field` | yes | - | The array field
1080-
| `processor` | yes | - | The processor to execute against each field
1078+
| Name | Required | Default | Description
1079+
| `field` | yes | - | The array field
1080+
| `processor` | yes | - | The processor to execute against each field
1081+
| `ignore_missing` | no | false | If `true` and `field` does not exist or is `null`, the processor quietly exits without modifying the document
10811082
|======
10821083

10831084
Assume the following document:

‎modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ForEachProcessor.java

+17-3
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.util.Set;
3131

3232
import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
33+
import static org.elasticsearch.ingest.ConfigurationUtils.readBooleanProperty;
3334
import static org.elasticsearch.ingest.ConfigurationUtils.readMap;
3435
import static org.elasticsearch.ingest.ConfigurationUtils.readStringProperty;
3536

@@ -47,16 +48,28 @@ public final class ForEachProcessor extends AbstractProcessor {
4748

4849
private final String field;
4950
private final Processor processor;
51+
private final boolean ignoreMissing;
5052

51-
ForEachProcessor(String tag, String field, Processor processor) {
53+
ForEachProcessor(String tag, String field, Processor processor, boolean ignoreMissing) {
5254
super(tag);
5355
this.field = field;
5456
this.processor = processor;
57+
this.ignoreMissing = ignoreMissing;
58+
}
59+
60+
boolean isIgnoreMissing() {
61+
return ignoreMissing;
5562
}
5663

5764
@Override
5865
public void execute(IngestDocument ingestDocument) throws Exception {
59-
List values = ingestDocument.getFieldValue(field, List.class);
66+
List values = ingestDocument.getFieldValue(field, List.class, ignoreMissing);
67+
if (values == null) {
68+
if (ignoreMissing) {
69+
return;
70+
}
71+
throw new IllegalArgumentException("field [" + field + "] is null, cannot loop over its elements.");
72+
}
6073
List<Object> newValues = new ArrayList<>(values.size());
6174
for (Object value : values) {
6275
Object previousValue = ingestDocument.getIngestMetadata().put("_value", value);
@@ -87,14 +100,15 @@ public static final class Factory implements Processor.Factory {
87100
public ForEachProcessor create(Map<String, Processor.Factory> factories, String tag,
88101
Map<String, Object> config) throws Exception {
89102
String field = readStringProperty(TYPE, tag, config, "field");
103+
boolean ignoreMissing = readBooleanProperty(TYPE, tag, config, "ignore_missing", false);
90104
Map<String, Map<String, Object>> processorConfig = readMap(TYPE, tag, config, "processor");
91105
Set<Map.Entry<String, Map<String, Object>>> entries = processorConfig.entrySet();
92106
if (entries.size() != 1) {
93107
throw newConfigurationException(TYPE, tag, "processor", "Must specify exactly one processor type");
94108
}
95109
Map.Entry<String, Map<String, Object>> entry = entries.iterator().next();
96110
Processor processor = ConfigurationUtils.readProcessor(factories, entry.getKey(), entry.getValue());
97-
return new ForEachProcessor(tag, field, processor);
111+
return new ForEachProcessor(tag, field, processor, ignoreMissing);
98112
}
99113
}
100114
}

‎modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorFactoryTests.java

+18
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,24 @@ public void testCreate() throws Exception {
4646
assertThat(forEachProcessor, Matchers.notNullValue());
4747
assertThat(forEachProcessor.getField(), equalTo("_field"));
4848
assertThat(forEachProcessor.getProcessor(), Matchers.sameInstance(processor));
49+
assertFalse(forEachProcessor.isIgnoreMissing());
50+
}
51+
52+
public void testSetIgnoreMissing() throws Exception {
53+
Processor processor = new TestProcessor(ingestDocument -> { });
54+
Map<String, Processor.Factory> registry = new HashMap<>();
55+
registry.put("_name", (r, t, c) -> processor);
56+
ForEachProcessor.Factory forEachFactory = new ForEachProcessor.Factory();
57+
58+
Map<String, Object> config = new HashMap<>();
59+
config.put("field", "_field");
60+
config.put("processor", Collections.singletonMap("_name", Collections.emptyMap()));
61+
config.put("ignore_missing", true);
62+
ForEachProcessor forEachProcessor = forEachFactory.create(registry, null, config);
63+
assertThat(forEachProcessor, Matchers.notNullValue());
64+
assertThat(forEachProcessor.getField(), equalTo("_field"));
65+
assertThat(forEachProcessor.getProcessor(), Matchers.sameInstance(processor));
66+
assertTrue(forEachProcessor.isIgnoreMissing());
4967
}
5068

5169
public void testCreateWithTooManyProcessorTypes() throws Exception {

‎modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/ForEachProcessorTests.java

+31-17
Original file line numberDiff line numberDiff line change
@@ -19,22 +19,22 @@
1919

2020
package org.elasticsearch.ingest.common;
2121

22-
import org.elasticsearch.ingest.CompoundProcessor;
23-
import org.elasticsearch.ingest.IngestDocument;
24-
import org.elasticsearch.ingest.Processor;
25-
import org.elasticsearch.ingest.TestProcessor;
26-
import org.elasticsearch.ingest.TestTemplateService;
27-
import org.elasticsearch.script.TemplateScript;
28-
import org.elasticsearch.test.ESTestCase;
29-
3022
import java.util.ArrayList;
3123
import java.util.Arrays;
3224
import java.util.Collections;
3325
import java.util.HashMap;
3426
import java.util.List;
3527
import java.util.Locale;
3628
import java.util.Map;
29+
import org.elasticsearch.ingest.CompoundProcessor;
30+
import org.elasticsearch.ingest.IngestDocument;
31+
import org.elasticsearch.ingest.Processor;
32+
import org.elasticsearch.ingest.TestProcessor;
33+
import org.elasticsearch.ingest.TestTemplateService;
34+
import org.elasticsearch.script.TemplateScript;
35+
import org.elasticsearch.test.ESTestCase;
3736

37+
import static org.elasticsearch.ingest.IngestDocumentMatcher.assertIngestDocument;
3838
import static org.hamcrest.Matchers.equalTo;
3939

4040
public class ForEachProcessorTests extends ESTestCase {
@@ -49,7 +49,8 @@ public void testExecute() throws Exception {
4949
);
5050

5151
ForEachProcessor processor = new ForEachProcessor(
52-
"_tag", "values", new UppercaseProcessor("_tag", "_ingest._value", false, "_ingest._value")
52+
"_tag", "values", new UppercaseProcessor("_tag", "_ingest._value", false, "_ingest._value"),
53+
false
5354
);
5455
processor.execute(ingestDocument);
5556

@@ -69,7 +70,7 @@ public void testExecuteWithFailure() throws Exception {
6970
throw new RuntimeException("failure");
7071
}
7172
});
72-
ForEachProcessor processor = new ForEachProcessor("_tag", "values", testProcessor);
73+
ForEachProcessor processor = new ForEachProcessor("_tag", "values", testProcessor, false);
7374
try {
7475
processor.execute(ingestDocument);
7576
fail("exception expected");
@@ -89,7 +90,8 @@ public void testExecuteWithFailure() throws Exception {
8990
});
9091
Processor onFailureProcessor = new TestProcessor(ingestDocument1 -> {});
9192
processor = new ForEachProcessor(
92-
"_tag", "values", new CompoundProcessor(false, Arrays.asList(testProcessor), Arrays.asList(onFailureProcessor))
93+
"_tag", "values", new CompoundProcessor(false, Arrays.asList(testProcessor), Arrays.asList(onFailureProcessor)),
94+
false
9395
);
9496
processor.execute(ingestDocument);
9597
assertThat(testProcessor.getInvokedCounter(), equalTo(3));
@@ -109,7 +111,7 @@ public void testMetaDataAvailable() throws Exception {
109111
id.setFieldValue("_ingest._value.type", id.getSourceAndMetadata().get("_type"));
110112
id.setFieldValue("_ingest._value.id", id.getSourceAndMetadata().get("_id"));
111113
});
112-
ForEachProcessor processor = new ForEachProcessor("_tag", "values", innerProcessor);
114+
ForEachProcessor processor = new ForEachProcessor("_tag", "values", innerProcessor, false);
113115
processor.execute(ingestDocument);
114116

115117
assertThat(innerProcessor.getInvokedCounter(), equalTo(2));
@@ -137,7 +139,7 @@ public void testRestOfTheDocumentIsAvailable() throws Exception {
137139
ForEachProcessor processor = new ForEachProcessor(
138140
"_tag", "values", new SetProcessor("_tag",
139141
new TestTemplateService.MockTemplateScript.Factory("_ingest._value.new_field"),
140-
(model) -> model.get("other")));
142+
(model) -> model.get("other")), false);
141143
processor.execute(ingestDocument);
142144

143145
assertThat(ingestDocument.getFieldValue("values.0.new_field", String.class), equalTo("value"));
@@ -174,7 +176,7 @@ public String getTag() {
174176
"_index", "_type", "_id", null, null, null, Collections.singletonMap("values", values)
175177
);
176178

177-
ForEachProcessor processor = new ForEachProcessor("_tag", "values", innerProcessor);
179+
ForEachProcessor processor = new ForEachProcessor("_tag", "values", innerProcessor, false);
178180
processor.execute(ingestDocument);
179181
@SuppressWarnings("unchecked")
180182
List<String> result = ingestDocument.getFieldValue("values", List.class);
@@ -199,7 +201,7 @@ public void testModifyFieldsOutsideArray() throws Exception {
199201
"_tag", "values", new CompoundProcessor(false,
200202
Collections.singletonList(new UppercaseProcessor("_tag_upper", "_ingest._value", false, "_ingest._value")),
201203
Collections.singletonList(new AppendProcessor("_tag", template, (model) -> (Collections.singletonList("added"))))
202-
));
204+
), false);
203205
processor.execute(ingestDocument);
204206

205207
List result = ingestDocument.getFieldValue("values", List.class);
@@ -225,7 +227,7 @@ public void testScalarValueAllowsUnderscoreValueFieldToRemainAccessible() throws
225227

226228
TestProcessor processor = new TestProcessor(doc -> doc.setFieldValue("_ingest._value",
227229
doc.getFieldValue("_source._value", String.class)));
228-
ForEachProcessor forEachProcessor = new ForEachProcessor("_tag", "values", processor);
230+
ForEachProcessor forEachProcessor = new ForEachProcessor("_tag", "values", processor, false);
229231
forEachProcessor.execute(ingestDocument);
230232

231233
List result = ingestDocument.getFieldValue("values", List.class);
@@ -258,7 +260,7 @@ public void testNestedForEach() throws Exception {
258260
doc -> doc.setFieldValue("_ingest._value", doc.getFieldValue("_ingest._value", String.class).toUpperCase(Locale.ENGLISH))
259261
);
260262
ForEachProcessor processor = new ForEachProcessor(
261-
"_tag", "values1", new ForEachProcessor("_tag", "_ingest._value.values2", testProcessor));
263+
"_tag", "values1", new ForEachProcessor("_tag", "_ingest._value.values2", testProcessor, false), false);
262264
processor.execute(ingestDocument);
263265

264266
List result = ingestDocument.getFieldValue("values1.0.values2", List.class);
@@ -270,4 +272,16 @@ public void testNestedForEach() throws Exception {
270272
assertThat(result.get(1), equalTo("JKL"));
271273
}
272274

275+
public void testIgnoreMissing() throws Exception {
276+
IngestDocument originalIngestDocument = new IngestDocument(
277+
"_index", "_type", "_id", null, null, null, Collections.emptyMap()
278+
);
279+
IngestDocument ingestDocument = new IngestDocument(originalIngestDocument);
280+
TestProcessor testProcessor = new TestProcessor(doc -> {});
281+
ForEachProcessor processor = new ForEachProcessor("_tag", "_ingest._value", testProcessor, true);
282+
processor.execute(ingestDocument);
283+
assertIngestDocument(originalIngestDocument, ingestDocument);
284+
assertThat(testProcessor.getInvokedCounter(), equalTo(0));
285+
}
286+
273287
}

0 commit comments

Comments
 (0)