Skip to content

Commit b17ce85

Browse files
gaobinlonggaobinlong
authored
Add copy_from parameter for set ingest processor (#63540)
1 parent 39a8643 commit b17ce85

File tree

7 files changed

+240
-21
lines changed

7 files changed

+240
-21
lines changed

‎docs/reference/ingest/processors/set.asciidoc‎

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,8 @@ its value will be replaced with the provided one.
1313
|======
1414
| Name | Required | Default | Description
1515
| `field` | yes | - | The field to insert, upsert, or update. Supports <<accessing-template-fields,template snippets>>.
16-
| `value` | yes | - | The value to be set for the field. Supports <<accessing-template-fields,template snippets>>.
16+
| `value` | yes* | - | The value to be set for the field. Supports <<accessing-template-fields,template snippets>>. May specify only one of `value` or `copy_from`.
17+
| `copy_from` | no | - | The origin field which will be copied to `field`, cannot set `value` simultaneously. Supported data types are `boolean`, `number`, `array`, `object`, `string`, `date`, etc.
1718
| `override` | no | true | If processor will update fields with pre-existing non-null-valued field. When set to `false`, such fields will not be touched.
1819
| `ignore_empty_value` | no | `false` | If `true` and `value` is a <<accessing-template-fields,template snippet>> that evaluates to `null` or the empty string, the processor quietly exits without modifying the document
1920
include::common-options.asciidoc[]
@@ -87,3 +88,54 @@ Result:
8788
}
8889
--------------------------------------------------
8990
// TESTRESPONSE[s/2019-03-11T21:54:37.909224Z/$body.docs.0.doc._ingest.timestamp/]
91+
The contents of a field including complex values such as arrays and objects can be copied to another field using `copy_from`:
92+
[source,console]
93+
--------------------------------------------------
94+
PUT _ingest/pipeline/set_bar
95+
{
96+
"description": "sets the value of bar from the field foo",
97+
"processors": [
98+
{
99+
"set": {
100+
"field": "bar",
101+
"copy_from": "foo"
102+
}
103+
}
104+
]
105+
}
106+
107+
POST _ingest/pipeline/set_bar/_simulate
108+
{
109+
"docs": [
110+
{
111+
"_source": {
112+
"foo": ["foo1", "foo2"]
113+
}
114+
}
115+
]
116+
}
117+
--------------------------------------------------
118+
119+
Result:
120+
121+
[source,console-result]
122+
--------------------------------------------------
123+
{
124+
"docs" : [
125+
{
126+
"doc" : {
127+
"_index" : "_index",
128+
"_id" : "_id",
129+
"_source" : {
130+
"bar": ["foo1", "foo2"],
131+
"foo": ["foo1", "foo2"]
132+
},
133+
"_ingest" : {
134+
"timestamp" : "2020-09-30T12:55:17.742795Z"
135+
}
136+
}
137+
}
138+
]
139+
}
140+
--------------------------------------------------
141+
// TESTRESPONSE[s/2020-09-30T12:55:17.742795Z/$body.docs.0.doc._ingest.timestamp/]

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

Lines changed: 33 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929

3030
import java.util.Map;
3131

32+
import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationException;
33+
3234
/**
3335
* Processor that adds new fields with their corresponding values. If the field is already present, its value
3436
* will be replaced with the provided one.
@@ -40,18 +42,20 @@ public final class SetProcessor extends AbstractProcessor {
4042
private final boolean overrideEnabled;
4143
private final TemplateScript.Factory field;
4244
private final ValueSource value;
45+
private final String copyFrom;
4346
private final boolean ignoreEmptyValue;
4447

45-
SetProcessor(String tag, String description, TemplateScript.Factory field, ValueSource value) {
46-
this(tag, description, field, value, true, false);
48+
SetProcessor(String tag, String description, TemplateScript.Factory field, ValueSource value, String copyFrom) {
49+
this(tag, description, field, value, copyFrom, true, false);
4750
}
4851

49-
SetProcessor(String tag, String description, TemplateScript.Factory field, ValueSource value, boolean overrideEnabled,
52+
SetProcessor(String tag, String description, TemplateScript.Factory field, ValueSource value, String copyFrom, boolean overrideEnabled,
5053
boolean ignoreEmptyValue) {
5154
super(tag, description);
5255
this.overrideEnabled = overrideEnabled;
5356
this.field = field;
5457
this.value = value;
58+
this.copyFrom = copyFrom;
5559
this.ignoreEmptyValue = ignoreEmptyValue;
5660
}
5761

@@ -67,14 +71,23 @@ public ValueSource getValue() {
6771
return value;
6872
}
6973

74+
public String getCopyFrom() {
75+
return copyFrom;
76+
}
77+
7078
public boolean isIgnoreEmptyValue() {
7179
return ignoreEmptyValue;
7280
}
7381

7482
@Override
7583
public IngestDocument execute(IngestDocument document) {
7684
if (overrideEnabled || document.hasField(field) == false || document.getFieldValue(field, Object.class) == null) {
77-
document.setFieldValue(field, value, ignoreEmptyValue);
85+
if (copyFrom != null) {
86+
Object fieldValue = document.getFieldValue(copyFrom, Object.class, ignoreEmptyValue);
87+
document.setFieldValue(field, fieldValue, ignoreEmptyValue);
88+
} else {
89+
document.setFieldValue(field, value, ignoreEmptyValue);
90+
}
7891
}
7992
return document;
8093
}
@@ -96,16 +109,30 @@ public Factory(ScriptService scriptService) {
96109
public SetProcessor create(Map<String, Processor.Factory> registry, String processorTag,
97110
String description, Map<String, Object> config) throws Exception {
98111
String field = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
99-
Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
112+
String copyFrom = ConfigurationUtils.readOptionalStringProperty(TYPE, processorTag, config, "copy_from");
113+
ValueSource valueSource = null;
114+
if (copyFrom == null) {
115+
Object value = ConfigurationUtils.readObject(TYPE, processorTag, config, "value");
116+
valueSource = ValueSource.wrap(value, scriptService);
117+
} else {
118+
Object value = config.remove("value");
119+
if (value != null) {
120+
throw newConfigurationException(TYPE, processorTag, "copy_from",
121+
"cannot set both `copy_from` and `value` in the same processor");
122+
}
123+
}
124+
100125
boolean overrideEnabled = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "override", true);
101126
TemplateScript.Factory compiledTemplate = ConfigurationUtils.compileTemplate(TYPE, processorTag,
102127
"field", field, scriptService);
103128
boolean ignoreEmptyValue = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_empty_value", false);
129+
104130
return new SetProcessor(
105131
processorTag,
106132
description,
107133
compiledTemplate,
108-
ValueSource.wrap(value, scriptService),
134+
valueSource,
135+
copyFrom,
109136
overrideEnabled,
110137
ignoreEmptyValue);
111138
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -140,7 +140,7 @@ public void testRestOfTheDocumentIsAvailable() throws Exception {
140140
ForEachProcessor processor = new ForEachProcessor(
141141
"_tag", null, "values", new SetProcessor("_tag",
142142
null, new TestTemplateService.MockTemplateScript.Factory("_ingest._value.new_field"),
143-
(model) -> model.get("other")), false);
143+
(model) -> model.get("other"), null), false);
144144
processor.execute(ingestDocument, (result, e) -> {});
145145

146146
assertThat(ingestDocument.getFieldValue("values.0.new_field", String.class), equalTo("value"));

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

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,4 +112,25 @@ public void testInvalidMustacheTemplate() throws Exception {
112112
assertThat(exception.getMetadata("es.processor_tag").get(0), equalTo(processorTag));
113113
}
114114

115+
public void testCreateWithCopyFrom() throws Exception {
116+
Map<String, Object> config = new HashMap<>();
117+
config.put("field", "field1");
118+
config.put("copy_from", "field2");
119+
String processorTag = randomAlphaOfLength(10);
120+
SetProcessor setProcessor = factory.create(null, processorTag, null, config);
121+
assertThat(setProcessor.getTag(), equalTo(processorTag));
122+
assertThat(setProcessor.getField().newInstance(Collections.emptyMap()).execute(), equalTo("field1"));
123+
assertThat(setProcessor.getCopyFrom(), equalTo("field2"));
124+
}
125+
126+
public void testCreateWithCopyFromAndValue() throws Exception {
127+
Map<String, Object> config = new HashMap<>();
128+
config.put("field", "field1");
129+
config.put("copy_from", "field2");
130+
config.put("value", "value1");
131+
String processorTag = randomAlphaOfLength(10);
132+
ElasticsearchException exception = expectThrows(ElasticsearchException.class,
133+
() -> factory.create(null, processorTag, null, config));
134+
assertThat(exception.getMessage(), equalTo("[copy_from] cannot set both `copy_from` and `value` in the same processor"));
135+
}
115136
}

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

Lines changed: 29 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.hamcrest.Matchers;
3030

3131
import java.util.HashMap;
32+
import java.util.Map;
3233

3334
import static org.hamcrest.Matchers.equalTo;
3435

@@ -38,7 +39,7 @@ public void testSetExistingFields() throws Exception {
3839
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
3940
String fieldName = RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument);
4041
Object fieldValue = RandomDocumentPicks.randomFieldValue(random());
41-
Processor processor = createSetProcessor(fieldName, fieldValue, true, false);
42+
Processor processor = createSetProcessor(fieldName, fieldValue, null, true, false);
4243
processor.execute(ingestDocument);
4344
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
4445
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(fieldValue));
@@ -50,7 +51,7 @@ public void testSetNewFields() throws Exception {
5051
IngestDocument testIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
5152
Object fieldValue = RandomDocumentPicks.randomFieldValue(random());
5253
String fieldName = RandomDocumentPicks.addRandomField(random(), testIngestDocument, fieldValue);
53-
Processor processor = createSetProcessor(fieldName, fieldValue, true, false);
54+
Processor processor = createSetProcessor(fieldName, fieldValue, null, true, false);
5455
processor.execute(ingestDocument);
5556
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
5657
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(fieldValue));
@@ -59,7 +60,7 @@ public void testSetNewFields() throws Exception {
5960
public void testSetFieldsTypeMismatch() throws Exception {
6061
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
6162
ingestDocument.setFieldValue("field", "value");
62-
Processor processor = createSetProcessor("field.inner", "value", true, false);
63+
Processor processor = createSetProcessor("field.inner", "value", null, true, false);
6364
try {
6465
processor.execute(ingestDocument);
6566
fail("processor execute should have failed");
@@ -73,7 +74,7 @@ public void testSetNewFieldWithOverrideDisabled() throws Exception {
7374
IngestDocument ingestDocument = new IngestDocument(new HashMap<>(), new HashMap<>());
7475
String fieldName = RandomDocumentPicks.randomFieldName(random());
7576
Object fieldValue = RandomDocumentPicks.randomFieldValue(random());
76-
Processor processor = createSetProcessor(fieldName, fieldValue, false, false);
77+
Processor processor = createSetProcessor(fieldName, fieldValue, null, false, false);
7778
processor.execute(ingestDocument);
7879
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
7980
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(fieldValue));
@@ -83,7 +84,7 @@ public void testSetExistingFieldWithOverrideDisabled() throws Exception {
8384
IngestDocument ingestDocument = new IngestDocument(new HashMap<>(), new HashMap<>());
8485
Object fieldValue = "foo";
8586
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, fieldValue);
86-
Processor processor = createSetProcessor(fieldName, "bar", false, false);
87+
Processor processor = createSetProcessor(fieldName, "bar", null, false, false);
8788
processor.execute(ingestDocument);
8889
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
8990
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(fieldValue));
@@ -94,54 +95,69 @@ public void testSetExistingNullFieldWithOverrideDisabled() throws Exception {
9495
Object fieldValue = null;
9596
Object newValue = "bar";
9697
String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, fieldValue);
97-
Processor processor = createSetProcessor(fieldName, newValue, false, false);
98+
Processor processor = createSetProcessor(fieldName, newValue, null, false, false);
9899
processor.execute(ingestDocument);
99100
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
100101
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(newValue));
101102
}
102103

103104
public void testSetMetadataExceptVersion() throws Exception {
104105
Metadata randomMetadata = randomFrom(Metadata.INDEX, Metadata.ID, Metadata.ROUTING);
105-
Processor processor = createSetProcessor(randomMetadata.getFieldName(), "_value", true, false);
106+
Processor processor = createSetProcessor(randomMetadata.getFieldName(), "_value", null, true, false);
106107
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
107108
processor.execute(ingestDocument);
108109
assertThat(ingestDocument.getFieldValue(randomMetadata.getFieldName(), String.class), Matchers.equalTo("_value"));
109110
}
110111

111112
public void testSetMetadataVersion() throws Exception {
112113
long version = randomNonNegativeLong();
113-
Processor processor = createSetProcessor(Metadata.VERSION.getFieldName(), version, true, false);
114+
Processor processor = createSetProcessor(Metadata.VERSION.getFieldName(), version, null, true, false);
114115
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
115116
processor.execute(ingestDocument);
116117
assertThat(ingestDocument.getFieldValue(Metadata.VERSION.getFieldName(), Long.class), Matchers.equalTo(version));
117118
}
118119

119120
public void testSetMetadataVersionType() throws Exception {
120121
String versionType = randomFrom("internal", "external", "external_gte");
121-
Processor processor = createSetProcessor(Metadata.VERSION_TYPE.getFieldName(), versionType, true, false);
122+
Processor processor = createSetProcessor(Metadata.VERSION_TYPE.getFieldName(), versionType, null, true, false);
122123
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
123124
processor.execute(ingestDocument);
124125
assertThat(ingestDocument.getFieldValue(Metadata.VERSION_TYPE.getFieldName(), String.class), Matchers.equalTo(versionType));
125126
}
126127

127128
public void testSetMetadataIfSeqNo() throws Exception {
128129
long ifSeqNo = randomNonNegativeLong();
129-
Processor processor = createSetProcessor(Metadata.IF_SEQ_NO.getFieldName(), ifSeqNo, true, false);
130+
Processor processor = createSetProcessor(Metadata.IF_SEQ_NO.getFieldName(), ifSeqNo, null, true, false);
130131
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
131132
processor.execute(ingestDocument);
132133
assertThat(ingestDocument.getFieldValue(Metadata.IF_SEQ_NO.getFieldName(), Long.class), Matchers.equalTo(ifSeqNo));
133134
}
134135

135136
public void testSetMetadataIfPrimaryTerm() throws Exception {
136137
long ifPrimaryTerm = randomNonNegativeLong();
137-
Processor processor = createSetProcessor(Metadata.IF_PRIMARY_TERM.getFieldName(), ifPrimaryTerm, true, false);
138+
Processor processor = createSetProcessor(Metadata.IF_PRIMARY_TERM.getFieldName(), ifPrimaryTerm, null, true, false);
138139
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
139140
processor.execute(ingestDocument);
140141
assertThat(ingestDocument.getFieldValue(Metadata.IF_PRIMARY_TERM.getFieldName(), Long.class), Matchers.equalTo(ifPrimaryTerm));
141142
}
142143

143-
private static Processor createSetProcessor(String fieldName, Object fieldValue, boolean overrideEnabled, boolean ignoreEmptyValue) {
144+
public void testCopyFromOtherField() throws Exception {
145+
Map<String, Object> document = new HashMap<>();
146+
Object fieldValue = RandomDocumentPicks.randomFieldValue(random());
147+
document.put("field", fieldValue);
148+
149+
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);
150+
String fieldName = RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument);
151+
152+
Processor processor = createSetProcessor(fieldName, null, "field", true, false);
153+
processor.execute(ingestDocument);
154+
assertThat(ingestDocument.hasField(fieldName), equalTo(true));
155+
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(fieldValue));
156+
}
157+
158+
private static Processor createSetProcessor(String fieldName, Object fieldValue, String copyFrom, boolean overrideEnabled,
159+
boolean ignoreEmptyValue) {
144160
return new SetProcessor(randomAlphaOfLength(10), null, new TestTemplateService.MockTemplateScript.Factory(fieldName),
145-
ValueSource.wrap(fieldValue, TestTemplateService.instance()), overrideEnabled, ignoreEmptyValue);
161+
ValueSource.wrap(fieldValue, TestTemplateService.instance()), copyFrom, overrideEnabled, ignoreEmptyValue);
146162
}
147163
}

0 commit comments

Comments
 (0)