Skip to content

version set in ingest pipeline #27573

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 15 commits into from
Feb 21, 2018
Merged

version set in ingest pipeline #27573

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
4e1b507
support document version set in ingest pipelines
PnPie Nov 28, 2017
180d8c1
remove wildcard imports
PnPie Nov 28, 2017
590bd72
change line width
PnPie Nov 28, 2017
f84d796
fix precommit
rjernst Dec 22, 2017
e42a8de
fix more precommit
rjernst Dec 22, 2017
47c3733
Merge branch 'master' of github.com:elastic/elasticsearch into ingest…
PnPie Jan 10, 2018
7981402
Merge branch 'master' of github.com:elastic/elasticsearch into ingest…
PnPie Jan 16, 2018
d93d6bb
Merge branch 'master' of github.com:elastic/elasticsearch into ingest…
PnPie Feb 8, 2018
5595f42
Merge branch 'master' of github.com:elastic/elasticsearch into ingest…
PnPie Feb 9, 2018
ee9a6df
update
PnPie Feb 10, 2018
4b1ad02
Update
PnPie Feb 13, 2018
380fced
Merge remote-tracking branch 'es/master' into ingest_pipeline_version…
martijnvg Feb 16, 2018
10e41bd
Threat version_type as a string, so that we can (de)serialize it with…
martijnvg Feb 16, 2018
1e5c3f9
Merge remote-tracking branch 'es/master' into ingest_pipeline_version…
martijnvg Feb 19, 2018
8799107
removed unneeded changes
martijnvg Feb 20, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 28 additions & 10 deletions ...les/ingest-common/src/test/java/org/elasticsearch/ingest/common/AppendProcessorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.elasticsearch.ingest.common;

import org.elasticsearch.index.VersionType;
import org.elasticsearch.ingest.IngestDocument;
import org.elasticsearch.ingest.Processor;
import org.elasticsearch.ingest.RandomDocumentPicks;
Expand Down Expand Up @@ -123,21 +124,38 @@ public void testConvertScalarToList() throws Exception {
}

public void testAppendMetadata() throws Exception {
//here any metadata field value becomes a list, which won't make sense in most of the cases,
// here any metadata field value becomes a list, which won't make sense in most of the cases,
// but support for append is streamlined like for set so we test it
IngestDocument.MetaData randomMetaData = randomFrom(IngestDocument.MetaData.values());
List<String> values = new ArrayList<>();
List<Object> values = new ArrayList<>();
Processor appendProcessor;
if (randomBoolean()) {
String value = randomAlphaOfLengthBetween(1, 10);
values.add(value);
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), value);
int valuesSize = randomIntBetween(0, 10);
if (randomMetaData == IngestDocument.MetaData.VERSION) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can version and version_type be tested in a dedicated test? I think that makes these tests easier to read.

if (randomBoolean()) {
Long version = randomLong();
values.add(version);
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), version);
} else {
for (int i = 0; i < valuesSize; i++) {
values.add(randomLong());
}
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), values);
}
} else if (randomMetaData == IngestDocument.MetaData.VERSION_TYPE) {
String versionType = randomFrom("internal", "external", "external_gt", "external_gte");
values.add(VersionType.fromString(versionType));
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), versionType);
} else {
int valuesSize = randomIntBetween(0, 10);
for (int i = 0; i < valuesSize; i++) {
values.add(randomAlphaOfLengthBetween(1, 10));
if (randomBoolean()) {
String value = randomAlphaOfLengthBetween(1, 10);
values.add(value);
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), value);
} else {
for (int i = 0; i < valuesSize; i++) {
values.add(randomAlphaOfLengthBetween(1, 10));
}
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), values);
}
appendProcessor = createAppendProcessor(randomMetaData.getFieldName(), values);
}

IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
Expand Down
23 changes: 21 additions & 2 deletions modules/ingest-common/src/test/java/org/elasticsearch/ingest/common/SetProcessorTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.elasticsearch.ingest.common;

import org.elasticsearch.index.VersionType;
import org.elasticsearch.ingest.IngestDocument;
import org.elasticsearch.ingest.Processor;
import org.elasticsearch.ingest.RandomDocumentPicks;
Expand Down Expand Up @@ -101,10 +102,28 @@ public void testSetExistingNullFieldWithOverrideDisabled() throws Exception {

public void testSetMetadata() throws Exception {
IngestDocument.MetaData randomMetaData = randomFrom(IngestDocument.MetaData.values());
Processor processor = createSetProcessor(randomMetaData.getFieldName(), "_value", true);
Processor processor;
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same as previous comment.

long version = 0L;
String versionType = new String();
if (randomMetaData == IngestDocument.MetaData.VERSION) {
version = randomLong();
processor = createSetProcessor(randomMetaData.getFieldName(), version, true);
} else if (randomMetaData == IngestDocument.MetaData.VERSION_TYPE) {
versionType = randomFrom("internal", "external", "external_gt", "external_gte");
processor = createSetProcessor(randomMetaData.getFieldName(), versionType, true);
} else {
processor = createSetProcessor(randomMetaData.getFieldName(), "_value", true);
}
IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
processor.execute(ingestDocument);
assertThat(ingestDocument.getFieldValue(randomMetaData.getFieldName(), String.class), Matchers.equalTo("_value"));
if (randomMetaData == IngestDocument.MetaData.VERSION) {
assertThat(ingestDocument.getFieldValue(randomMetaData.getFieldName(), Long.class), Matchers.equalTo(version));
} else if (randomMetaData == IngestDocument.MetaData.VERSION_TYPE) {
assertThat(ingestDocument.getFieldValue(randomMetaData.getFieldName(), VersionType.class),
Matchers.equalTo(VersionType.fromString(versionType)));
} else {
assertThat(ingestDocument.getFieldValue(randomMetaData.getFieldName(), String.class), Matchers.equalTo("_value"));
}
}

private static Processor createSetProcessor(String fieldName, Object fieldValue, boolean overrideEnabled) {
Expand Down
11 changes: 10 additions & 1 deletion server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.elasticsearch.common.io.stream.Writeable;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.ingest.ConfigurationUtils;
import org.elasticsearch.ingest.IngestDocument;
import org.elasticsearch.ingest.Pipeline;
Expand Down Expand Up @@ -195,8 +196,16 @@ private static List<IngestDocument> parseDocs(Map<String, Object> config) {
dataMap, MetaData.ROUTING.getFieldName());
String parent = ConfigurationUtils.readOptionalStringOrIntProperty(null, null,
dataMap, MetaData.PARENT.getFieldName());
Long version = null;
if (dataMap.containsKey(MetaData.VERSION.getFieldName())) {
version = (Long) ConfigurationUtils.readObject(null, null, dataMap, MetaData.VERSION.getFieldName());
}
VersionType versionType = null;
if (dataMap.containsKey(MetaData.VERSION_TYPE.getFieldName())) {
versionType = (VersionType) ConfigurationUtils.readObject(null, null, dataMap, MetaData.VERSION_TYPE.getFieldName());
}
IngestDocument ingestDocument =
new IngestDocument(index, type, id, routing, parent, document);
new IngestDocument(index, type, id, routing, parent, version, versionType, document);
ingestDocumentList.add(ingestDocument);
}
return ingestDocumentList;
Expand Down
6 changes: 3 additions & 3 deletions server/src/main/java/org/elasticsearch/action/ingest/WriteableIngestDocument.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,10 @@ IngestDocument getIngestDocument() {
@Override
public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
builder.startObject("doc");
Map<IngestDocument.MetaData, String> metadataMap = ingestDocument.extractMetadata();
for (Map.Entry<IngestDocument.MetaData, String> metadata : metadataMap.entrySet()) {
Map<IngestDocument.MetaData, Object> metadataMap = ingestDocument.extractMetadata();
for (Map.Entry<IngestDocument.MetaData, Object> metadata : metadataMap.entrySet()) {
if (metadata.getValue() != null) {
builder.field(metadata.getKey().getFieldName(), metadata.getValue());
builder.field(metadata.getKey().getFieldName(), metadata.getValue().toString());
}
}
builder.field("_source", ingestDocument.getSourceAndMetadata());
Expand Down
4 changes: 3 additions & 1 deletion server/src/main/java/org/elasticsearch/index/mapper/VersionFieldMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,9 @@ public MetadataFieldMapper getDefault(MappedFieldType fieldType, ParserContext c
}
}

static final class VersionFieldType extends MappedFieldType {
public static final class VersionFieldType extends MappedFieldType {

public static final String NAME = "_version_type";

VersionFieldType() {
}
Expand Down
35 changes: 29 additions & 6 deletions server/src/main/java/org/elasticsearch/ingest/IngestDocument.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,14 +20,14 @@
package org.elasticsearch.ingest;

import org.elasticsearch.common.Strings;
import org.elasticsearch.common.settings.Setting;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.mapper.IdFieldMapper;
import org.elasticsearch.index.mapper.IndexFieldMapper;
import org.elasticsearch.index.mapper.ParentFieldMapper;
import org.elasticsearch.index.mapper.RoutingFieldMapper;
import org.elasticsearch.index.mapper.SourceFieldMapper;
import org.elasticsearch.index.mapper.TypeFieldMapper;
import org.elasticsearch.script.ScriptService;
import org.elasticsearch.index.mapper.VersionFieldMapper;
import org.elasticsearch.script.TemplateScript;

import java.time.ZoneOffset;
Expand Down Expand Up @@ -73,6 +73,17 @@ public IngestDocument(String index, String type, String id, String routing, Stri
this.ingestMetadata.put(TIMESTAMP, ZonedDateTime.now(ZoneOffset.UTC));
}

public IngestDocument(String index, String type, String id, String routing,String parent,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of adding a new constructor, can the existing constructor be modified to also accept version and versionType? I know this requires changing more tests, but I think this class should only have two constructors.

Long version, VersionType versionType,Map<String, Object> source) {
this(index, type, id, routing, parent, source);
if (version != null) {
sourceAndMetadata.put(MetaData.VERSION.getFieldName(), version);
}
if (versionType != null) {
sourceAndMetadata.put(MetaData.VERSION_TYPE.getFieldName(), versionType);
}
}

/**
* Copy constructor that creates a new {@link IngestDocument} which has exactly the same properties as the one provided as argument
*/
Expand Down Expand Up @@ -457,6 +468,9 @@ private void setFieldValue(String path, Object value, boolean append) {
}

String leafKey = fieldPath.pathElements[fieldPath.pathElements.length - 1];
if (leafKey == MetaData.VERSION_TYPE.getFieldName()) {
value = VersionType.fromString(value.toString());
}
if (context == null) {
throw new IllegalArgumentException("cannot set [" + leafKey + "] with null parent as part of path [" + path + "]");
}
Expand Down Expand Up @@ -559,10 +573,17 @@ private Map<String, Object> createTemplateModel() {
* one time operation that extracts the metadata fields from the ingest document and returns them.
* Metadata fields that used to be accessible as ordinary top level fields will be removed as part of this call.
*/
public Map<MetaData, String> extractMetadata() {
Map<MetaData, String> metadataMap = new EnumMap<>(MetaData.class);
public Map<MetaData, Object> extractMetadata() {
Map<MetaData, Object> metadataMap = new EnumMap<>(MetaData.class);
for (MetaData metaData : MetaData.values()) {
metadataMap.put(metaData, cast(metaData.getFieldName(), sourceAndMetadata.remove(metaData.getFieldName()), String.class));
if (metaData == MetaData.VERSION) {
metadataMap.put(metaData, cast(metaData.getFieldName(), sourceAndMetadata.remove(metaData.getFieldName()), Long.class));
} else if (metaData == MetaData.VERSION_TYPE) {
metadataMap.put(metaData,
cast(metaData.getFieldName(), sourceAndMetadata.remove(metaData.getFieldName()), VersionType.class));
} else {
metadataMap.put(metaData, cast(metaData.getFieldName(), sourceAndMetadata.remove(metaData.getFieldName()), String.class));
}
}
return metadataMap;
}
Expand Down Expand Up @@ -651,7 +672,9 @@ public enum MetaData {
TYPE(TypeFieldMapper.NAME),
ID(IdFieldMapper.NAME),
ROUTING(RoutingFieldMapper.NAME),
PARENT(ParentFieldMapper.NAME);
PARENT(ParentFieldMapper.NAME),
VERSION(VersionFieldMapper.NAME),
VERSION_TYPE(VersionFieldMapper.VersionFieldType.NAME);

private final String fieldName;

Expand Down
19 changes: 12 additions & 7 deletions server/src/main/java/org/elasticsearch/ingest/PipelineExecutionService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.elasticsearch.common.metrics.CounterMetric;
import org.elasticsearch.common.metrics.MeanMetric;
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.threadpool.ThreadPool;

import java.util.Collections;
Expand Down Expand Up @@ -164,18 +165,22 @@ private void innerExecute(IndexRequest indexRequest, Pipeline pipeline) throws E
String id = indexRequest.id();
String routing = indexRequest.routing();
String parent = indexRequest.parent();
Long version = indexRequest.version();
VersionType versionType = indexRequest.versionType();
Map<String, Object> sourceAsMap = indexRequest.sourceAsMap();
IngestDocument ingestDocument = new IngestDocument(index, type, id, routing, parent, sourceAsMap);
IngestDocument ingestDocument = new IngestDocument(index, type, id, routing, parent, version, versionType, sourceAsMap);
pipeline.execute(ingestDocument);

Map<IngestDocument.MetaData, String> metadataMap = ingestDocument.extractMetadata();
Map<IngestDocument.MetaData, Object> metadataMap = ingestDocument.extractMetadata();
//it's fine to set all metadata fields all the time, as ingest document holds their starting values
//before ingestion, which might also get modified during ingestion.
indexRequest.index(metadataMap.get(IngestDocument.MetaData.INDEX));
indexRequest.type(metadataMap.get(IngestDocument.MetaData.TYPE));
indexRequest.id(metadataMap.get(IngestDocument.MetaData.ID));
indexRequest.routing(metadataMap.get(IngestDocument.MetaData.ROUTING));
indexRequest.parent(metadataMap.get(IngestDocument.MetaData.PARENT));
indexRequest.index((String) metadataMap.get(IngestDocument.MetaData.INDEX));
indexRequest.type((String) metadataMap.get(IngestDocument.MetaData.TYPE));
indexRequest.id((String) metadataMap.get(IngestDocument.MetaData.ID));
indexRequest.routing((String) metadataMap.get(IngestDocument.MetaData.ROUTING));
indexRequest.parent((String) metadataMap.get(IngestDocument.MetaData.PARENT));
indexRequest.version((Long) metadataMap.get(IngestDocument.MetaData.VERSION));
indexRequest.versionType((VersionType) metadataMap.get(IngestDocument.MetaData.VERSION_TYPE));
indexRequest.source(ingestDocument.getSourceAndMetadata());
} catch (Exception e) {
totalStats.ingestFailed();
Expand Down
32 changes: 23 additions & 9 deletions ...er/src/test/java/org/elasticsearch/action/ingest/SimulatePipelineRequestParsingTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import java.util.List;
import java.util.Map;

import org.elasticsearch.index.VersionType;
import org.elasticsearch.ingest.CompoundProcessor;
import org.elasticsearch.ingest.IngestDocument;
import org.elasticsearch.ingest.Pipeline;
Expand All @@ -44,6 +45,8 @@
import static org.elasticsearch.ingest.IngestDocument.MetaData.PARENT;
import static org.elasticsearch.ingest.IngestDocument.MetaData.ROUTING;
import static org.elasticsearch.ingest.IngestDocument.MetaData.TYPE;
import static org.elasticsearch.ingest.IngestDocument.MetaData.VERSION;
import static org.elasticsearch.ingest.IngestDocument.MetaData.VERSION_TYPE;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.nullValue;
import static org.mockito.Mockito.mock;
Expand Down Expand Up @@ -98,7 +101,7 @@ public void testParseUsingPipelineStore() throws Exception {
Iterator<Map<String, Object>> expectedDocsIterator = expectedDocs.iterator();
for (IngestDocument ingestDocument : actualRequest.getDocuments()) {
Map<String, Object> expectedDocument = expectedDocsIterator.next();
Map<IngestDocument.MetaData, String> metadataMap = ingestDocument.extractMetadata();
Map<IngestDocument.MetaData, Object> metadataMap = ingestDocument.extractMetadata();
assertThat(metadataMap.get(INDEX), equalTo(expectedDocument.get(INDEX.getFieldName())));
assertThat(metadataMap.get(TYPE), equalTo(expectedDocument.get(TYPE.getFieldName())));
assertThat(metadataMap.get(ID), equalTo(expectedDocument.get(ID.getFieldName())));
Expand All @@ -120,17 +123,26 @@ public void testParseWithProvidedPipeline() throws Exception {
for (int i = 0; i < numDocs; i++) {
Map<String, Object> doc = new HashMap<>();
Map<String, Object> expectedDoc = new HashMap<>();
List<IngestDocument.MetaData> fields = Arrays.asList(INDEX, TYPE, ID, ROUTING, PARENT);
List<IngestDocument.MetaData> fields = Arrays.asList(INDEX, TYPE, ID, ROUTING, PARENT, VERSION, VERSION_TYPE);
for(IngestDocument.MetaData field : fields) {
if(randomBoolean()) {
String value = randomAlphaOfLengthBetween(1, 10);
if (field == VERSION) {
Long value = randomLong();
doc.put(field.getFieldName(), value);
expectedDoc.put(field.getFieldName(), value);
}
else {
Integer value = randomIntBetween(1, 1000000);
} else if (field == VERSION_TYPE) {
VersionType value = randomFrom(VersionType.INTERNAL, VersionType.EXTERNAL, VersionType.EXTERNAL_GTE);
doc.put(field.getFieldName(), value);
expectedDoc.put(field.getFieldName(), String.valueOf(value));
expectedDoc.put(field.getFieldName(), value);
} else {
if (randomBoolean()) {
String value = randomAlphaOfLengthBetween(1, 10);
doc.put(field.getFieldName(), value);
expectedDoc.put(field.getFieldName(), value);
} else {
Integer value = randomIntBetween(1, 1000000);
doc.put(field.getFieldName(), value);
expectedDoc.put(field.getFieldName(), String.valueOf(value));
}
}
}
String fieldName = randomAlphaOfLengthBetween(1, 10);
Expand Down Expand Up @@ -175,12 +187,14 @@ public void testParseWithProvidedPipeline() throws Exception {
Iterator<Map<String, Object>> expectedDocsIterator = expectedDocs.iterator();
for (IngestDocument ingestDocument : actualRequest.getDocuments()) {
Map<String, Object> expectedDocument = expectedDocsIterator.next();
Map<IngestDocument.MetaData, String> metadataMap = ingestDocument.extractMetadata();
Map<IngestDocument.MetaData, Object> metadataMap = ingestDocument.extractMetadata();
assertThat(metadataMap.get(INDEX), equalTo(expectedDocument.get(INDEX.getFieldName())));
assertThat(metadataMap.get(TYPE), equalTo(expectedDocument.get(TYPE.getFieldName())));
assertThat(metadataMap.get(ID), equalTo(expectedDocument.get(ID.getFieldName())));
assertThat(metadataMap.get(ROUTING), equalTo(expectedDocument.get(ROUTING.getFieldName())));
assertThat(metadataMap.get(PARENT), equalTo(expectedDocument.get(PARENT.getFieldName())));
assertThat(metadataMap.get(VERSION), equalTo(expectedDocument.get(VERSION.getFieldName())));
assertThat(metadataMap.get(VERSION_TYPE), equalTo(expectedDocument.get(VERSION_TYPE.getFieldName())));
assertThat(ingestDocument.getSourceAndMetadata(), equalTo(expectedDocument.get(Fields.SOURCE)));
}

Expand Down
Loading