Skip to content

Commit

Permalink
Rename ingest processor supports overriding target field if exists (#…
Browse files Browse the repository at this point in the history
…12990)

* Rename ingest processor supports overriding target field if exists

Signed-off-by: Gao Binlong <[email protected]>

* Modify change log

Signed-off-by: Gao Binlong <[email protected]>

---------

Signed-off-by: Gao Binlong <[email protected]>
(cherry picked from commit 41a3055)
Signed-off-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
  • Loading branch information
github-actions[bot] committed Apr 23, 2024
1 parent 3956365 commit 6939a3d
Show file tree
Hide file tree
Showing 6 changed files with 96 additions and 26 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
- Allow setting KEYSTORE_PASSWORD through env variable ([#12865](https://github.com/opensearch-project/OpenSearch/pull/12865))
- Add a counter to node stat (and _cat/shards) api to track shard going from idle to non-idle ([#12768](https://github.com/opensearch-project/OpenSearch/pull/12768))
- [Concurrent Segment Search] Disable concurrent segment search for system indices and throttled requests ([#12954](https://github.com/opensearch-project/OpenSearch/pull/12954))
- Rename ingest processor supports overriding target field if exists ([#12990](https://github.com/opensearch-project/OpenSearch/pull/12990))
- [Tiered Caching] Make took time caching policy setting dynamic ([#13063](https://github.com/opensearch-project/OpenSearch/pull/13063))
- Detect breaking changes on pull requests ([#9044](https://github.com/opensearch-project/OpenSearch/pull/9044))
- Add cluster primary balance contraint for rebalancing with buffer ([#12656](https://github.com/opensearch-project/OpenSearch/pull/12656))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -52,18 +52,21 @@ public final class RenameProcessor extends AbstractProcessor {
private final TemplateScript.Factory field;
private final TemplateScript.Factory targetField;
private final boolean ignoreMissing;
private final boolean overrideTarget;

RenameProcessor(
String tag,
String description,
TemplateScript.Factory field,
TemplateScript.Factory targetField,
boolean ignoreMissing
boolean ignoreMissing,
boolean overrideTarget
) {
super(tag, description);
this.field = field;
this.targetField = targetField;
this.ignoreMissing = ignoreMissing;
this.overrideTarget = overrideTarget;
}

TemplateScript.Factory getField() {
Expand All @@ -78,6 +81,10 @@ boolean isIgnoreMissing() {
return ignoreMissing;
}

boolean isOverrideTarget() {
return overrideTarget;
}

@Override
public IngestDocument execute(IngestDocument document) {
String path = document.renderTemplate(field);
Expand All @@ -94,9 +101,10 @@ public IngestDocument execute(IngestDocument document) {
// We fail here if the target field point to an array slot that is out of range.
// If we didn't do this then we would fail if we set the value in the target_field
// and then on failure processors would not see that value we tried to rename as we already
// removed it.
// removed it. If the target field is out of range, we throw the exception no matter
// what the parameter overrideTarget is.
String target = document.renderTemplate(targetField);
if (document.hasField(target, true)) {
if (document.hasField(target, true) && !overrideTarget) {
throw new IllegalArgumentException("field [" + target + "] already exists");
}

Expand Down Expand Up @@ -143,7 +151,8 @@ public RenameProcessor create(
scriptService
);
boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_missing", false);
return new RenameProcessor(processorTag, description, fieldTemplate, targetFieldTemplate, ignoreMissing);
boolean overrideTarget = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "override_target", false);
return new RenameProcessor(processorTag, description, fieldTemplate, targetFieldTemplate, ignoreMissing, overrideTarget);
}
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,7 @@ public void testEscapeFields_valueField() throws Exception {
null,
new TestTemplateService.MockTemplateScript.Factory("foo"),
new TestTemplateService.MockTemplateScript.Factory("foo.bar"),
false,
false
);
processor.execute(document);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,19 @@ public void testCreateWithIgnoreMissing() throws Exception {
assertThat(renameProcessor.isIgnoreMissing(), equalTo(true));
}

public void testCreateWithOverrideTarget() throws Exception {
Map<String, Object> config = new HashMap<>();
config.put("field", "old_field");
config.put("target_field", "new_field");
config.put("override_target", true);
String processorTag = randomAlphaOfLength(10);
RenameProcessor renameProcessor = factory.create(null, processorTag, null, config);
assertThat(renameProcessor.getTag(), equalTo(processorTag));
assertThat(renameProcessor.getField().newInstance(Collections.emptyMap()).execute(), equalTo("old_field"));
assertThat(renameProcessor.getTargetField().newInstance(Collections.emptyMap()).execute(), equalTo("new_field"));
assertThat(renameProcessor.isOverrideTarget(), equalTo(true));
}

public void testCreateNoFieldPresent() throws Exception {
Map<String, Object> config = new HashMap<>();
config.put("target_field", "new_field");
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,7 @@ public void testRename() throws Exception {
do {
newFieldName = RandomDocumentPicks.randomFieldName(random());
} while (RandomDocumentPicks.canAddField(newFieldName, ingestDocument) == false || newFieldName.equals(fieldName));
Processor processor = createRenameProcessor(fieldName, newFieldName, false);
Processor processor = createRenameProcessor(fieldName, newFieldName, false, false);
processor.execute(ingestDocument);
assertThat(ingestDocument.getFieldValue(newFieldName, Object.class), equalTo(fieldValue));
}
Expand All @@ -77,7 +77,7 @@ public void testRenameArrayElement() throws Exception {
document.put("one", one);
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), document);

Processor processor = createRenameProcessor("list.0", "item", false);
Processor processor = createRenameProcessor("list.0", "item", false, false);
processor.execute(ingestDocument);
Object actualObject = ingestDocument.getSourceAndMetadata().get("list");
assertThat(actualObject, instanceOf(List.class));
Expand All @@ -90,7 +90,7 @@ public void testRenameArrayElement() throws Exception {
assertThat(actualObject, instanceOf(String.class));
assertThat(actualObject, equalTo("item1"));

processor = createRenameProcessor("list.0", "list.3", false);
processor = createRenameProcessor("list.0", "list.3", false, randomBoolean());
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
Expand All @@ -105,7 +105,7 @@ public void testRenameArrayElement() throws Exception {
public void testRenameNonExistingField() throws Exception {
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
String fieldName = RandomDocumentPicks.randomFieldName(random());
Processor processor = createRenameProcessor(fieldName, RandomDocumentPicks.randomFieldName(random()), false);
Processor processor = createRenameProcessor(fieldName, RandomDocumentPicks.randomFieldName(random()), false, false);
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
Expand All @@ -114,7 +114,7 @@ public void testRenameNonExistingField() throws Exception {
}

// when using template snippet, the resolved field path maybe empty
processor = createRenameProcessor("", RandomDocumentPicks.randomFieldName(random()), false);
processor = createRenameProcessor("", RandomDocumentPicks.randomFieldName(random()), false, false);
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
Expand All @@ -127,38 +127,44 @@ public void testRenameNonExistingFieldWithIgnoreMissing() throws Exception {
IngestDocument originalIngestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
IngestDocument ingestDocument = new IngestDocument(originalIngestDocument);
String fieldName = RandomDocumentPicks.randomFieldName(random());
Processor processor = createRenameProcessor(fieldName, RandomDocumentPicks.randomFieldName(random()), true);
Processor processor = createRenameProcessor(fieldName, RandomDocumentPicks.randomFieldName(random()), true, false);
processor.execute(ingestDocument);
assertIngestDocument(originalIngestDocument, ingestDocument);

// when using template snippet, the resolved field path maybe empty
processor = createRenameProcessor("", RandomDocumentPicks.randomFieldName(random()), true);
processor = createRenameProcessor("", RandomDocumentPicks.randomFieldName(random()), true, false);
processor.execute(ingestDocument);
assertIngestDocument(originalIngestDocument, ingestDocument);
}

public void testRenameNewFieldAlreadyExists() throws Exception {
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
String fieldName = RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument);
Processor processor = createRenameProcessor(
RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument),
fieldName,
false
);
String field = RandomDocumentPicks.randomExistingFieldName(random(), ingestDocument);
Object fieldValue = ingestDocument.getFieldValue(field, Object.class);
String targetField = RandomDocumentPicks.addRandomField(random(), ingestDocument, RandomDocumentPicks.randomFieldValue(random()));

Processor processor = createRenameProcessor(field, targetField, false, false);
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
} catch (IllegalArgumentException e) {
assertThat(e.getMessage(), equalTo("field [" + fieldName + "] already exists"));
assertThat(e.getMessage(), equalTo("field [" + targetField + "] already exists"));
}

Processor processorWithOverrideTarget = createRenameProcessor(field, targetField, false, true);

processorWithOverrideTarget.execute(ingestDocument);
assertThat(ingestDocument.hasField(field), equalTo(false));
assertThat(ingestDocument.hasField(targetField), equalTo(true));
assertThat(ingestDocument.getFieldValue(targetField, Object.class), equalTo(fieldValue));
}

public void testRenameExistingFieldNullValue() throws Exception {
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
String fieldName = RandomDocumentPicks.randomFieldName(random());
ingestDocument.setFieldValue(fieldName, null);
String newFieldName = randomValueOtherThanMany(ingestDocument::hasField, () -> RandomDocumentPicks.randomFieldName(random()));
Processor processor = createRenameProcessor(fieldName, newFieldName, false);
Processor processor = createRenameProcessor(fieldName, newFieldName, false, false);
processor.execute(ingestDocument);
if (newFieldName.startsWith(fieldName + '.')) {
assertThat(ingestDocument.getFieldValue(fieldName, Object.class), instanceOf(Map.class));
Expand All @@ -182,7 +188,7 @@ public Object put(String key, Object value) {
source.put("list", Collections.singletonList("item"));

IngestDocument ingestDocument = new IngestDocument(source, Collections.emptyMap());
Processor processor = createRenameProcessor("list", "new_field", false);
Processor processor = createRenameProcessor("list", "new_field", false, false);
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
Expand All @@ -206,7 +212,7 @@ public Object remove(Object key) {
source.put("list", Collections.singletonList("item"));

IngestDocument ingestDocument = new IngestDocument(source, Collections.emptyMap());
Processor processor = createRenameProcessor("list", "new_field", false);
Processor processor = createRenameProcessor("list", "new_field", false, false);
try {
processor.execute(ingestDocument);
fail("processor execute should have failed");
Expand All @@ -221,12 +227,12 @@ public void testRenameLeafIntoBranch() throws Exception {
Map<String, Object> source = new HashMap<>();
source.put("foo", "bar");
IngestDocument ingestDocument = new IngestDocument(source, Collections.emptyMap());
Processor processor1 = createRenameProcessor("foo", "foo.bar", false);
Processor processor1 = createRenameProcessor("foo", "foo.bar", false, false);
processor1.execute(ingestDocument);
assertThat(ingestDocument.getFieldValue("foo", Map.class), equalTo(Collections.singletonMap("bar", "bar")));
assertThat(ingestDocument.getFieldValue("foo.bar", String.class), equalTo("bar"));

Processor processor2 = createRenameProcessor("foo.bar", "foo.bar.baz", false);
Processor processor2 = createRenameProcessor("foo.bar", "foo.bar.baz", false, false);
processor2.execute(ingestDocument);
assertThat(
ingestDocument.getFieldValue("foo", Map.class),
Expand All @@ -236,18 +242,19 @@ public void testRenameLeafIntoBranch() throws Exception {
assertThat(ingestDocument.getFieldValue("foo.bar.baz", String.class), equalTo("bar"));

// for fun lets try to restore it (which don't allow today)
Processor processor3 = createRenameProcessor("foo.bar.baz", "foo", false);
Processor processor3 = createRenameProcessor("foo.bar.baz", "foo", false, false);
Exception e = expectThrows(IllegalArgumentException.class, () -> processor3.execute(ingestDocument));
assertThat(e.getMessage(), equalTo("field [foo] already exists"));
}

private RenameProcessor createRenameProcessor(String field, String targetField, boolean ignoreMissing) {
private RenameProcessor createRenameProcessor(String field, String targetField, boolean ignoreMissing, boolean overrideTarget) {
return new RenameProcessor(
randomAlphaOfLength(10),
null,
new TestTemplateService.MockTemplateScript.Factory(field),
new TestTemplateService.MockTemplateScript.Factory(targetField),
ignoreMissing
ignoreMissing,
overrideTarget
);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -64,3 +64,42 @@ teardown:
index: test
id: 1
- match: { _source.message: "foo bar baz" }

---
"Test rename processor with override_target":
- skip:
version: " - 2.13.99"
reason: "introduced in 2.14.0"
- do:
ingest.put_pipeline:
id: "my_pipeline"
body: >
{
"description": "_description",
"processors": [
{
"rename" : {
"field" : "foo",
"target_field" : "bar",
"override_target" : true
}
}
]
}
- match: { acknowledged: true }

- do:
index:
index: test
id: 1
pipeline: "my_pipeline"
body: {
foo: "foo",
bar: "bar"
}

- do:
get:
index: test
id: 1
- match: { _source: { "bar": "foo" } }

0 comments on commit 6939a3d

Please sign in to comment.