version set in ingest pipeline

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

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestDocument.MetaData;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.TestTemplateService;
@@ -122,10 +124,10 @@ 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,
+    public void testAppendMetadataExceptVersion() throws Exception {
+        // 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());
+        MetaData randomMetaData = randomFrom(MetaData.INDEX, MetaData.TYPE, MetaData.ID, MetaData.ROUTING, MetaData.PARENT);
         List<String> values = new ArrayList<>();
         Processor appendProcessor;
         if (randomBoolean()) {

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

@@ -38,7 +38,7 @@ public void testJodaPattern() throws Exception {
                 "events-", "y", "yyyyMMdd"
         );
 
-        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null,
+        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null, null, null,
                 Collections.singletonMap("_field", "2016-04-25T12:24:20.101Z"));
         processor.execute(document);
         assertThat(document.getSourceAndMetadata().get("_index"), equalTo("<events-{20160425||/y{yyyyMMdd|UTC}}>"));
@@ -48,7 +48,7 @@ public void testTAI64N()throws Exception {
         Function<String, DateTime> function = DateFormat.Tai64n.getFunction(null, DateTimeZone.UTC, null);
         DateIndexNameProcessor dateProcessor = new DateIndexNameProcessor("_tag", "_field", Collections.singletonList(function),
                 DateTimeZone.UTC, "events-", "m", "yyyyMMdd");
-        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null,
+        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null, null, null,
                 Collections.singletonMap("_field", (randomBoolean() ? "@" : "") + "4000000050d506482dbdf024"));
         dateProcessor.execute(document);
         assertThat(document.getSourceAndMetadata().get("_index"), equalTo("<events-{20121222||/m{yyyyMMdd|UTC}}>"));
@@ -58,12 +58,12 @@ public void testUnixMs()throws Exception {
         Function<String, DateTime> function = DateFormat.UnixMs.getFunction(null, DateTimeZone.UTC, null);
         DateIndexNameProcessor dateProcessor = new DateIndexNameProcessor("_tag", "_field", Collections.singletonList(function),
                 DateTimeZone.UTC, "events-", "m", "yyyyMMdd");
-        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null,
+        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null, null, null,
                 Collections.singletonMap("_field", "1000500"));
         dateProcessor.execute(document);
         assertThat(document.getSourceAndMetadata().get("_index"), equalTo("<events-{19700101||/m{yyyyMMdd|UTC}}>"));
 
-        document = new IngestDocument("_index", "_type", "_id", null, null,
+        document = new IngestDocument("_index", "_type", "_id", null, null, null, null,
                 Collections.singletonMap("_field", 1000500L));
         dateProcessor.execute(document);
         assertThat(document.getSourceAndMetadata().get("_index"), equalTo("<events-{19700101||/m{yyyyMMdd|UTC}}>"));
@@ -73,7 +73,7 @@ public void testUnix()throws Exception {
         Function<String, DateTime> function = DateFormat.Unix.getFunction(null, DateTimeZone.UTC, null);
         DateIndexNameProcessor dateProcessor = new DateIndexNameProcessor("_tag", "_field", Collections.singletonList(function),
                 DateTimeZone.UTC, "events-", "m", "yyyyMMdd");
-        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null,
+        IngestDocument document = new IngestDocument("_index", "_type", "_id", null, null, null, null,
                 Collections.singletonMap("_field", "1000.5"));
         dateProcessor.execute(document);
         assertThat(document.getSourceAndMetadata().get("_index"), equalTo("<events-{19700101||/m{yyyyMMdd|UTC}}>"));

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

@@ -45,7 +45,7 @@ public void testExecute() throws Exception {
         values.add("bar");
         values.add("baz");
         IngestDocument ingestDocument = new IngestDocument(
-            "_index", "_type", "_id", null, null, Collections.singletonMap("values", values)
+            "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values", values)
         );
 
         ForEachProcessor processor = new ForEachProcessor(
@@ -61,7 +61,7 @@ public void testExecute() throws Exception {
 
     public void testExecuteWithFailure() throws Exception {
         IngestDocument ingestDocument = new IngestDocument(
-            "_index", "_type", "_id", null, null, Collections.singletonMap("values", Arrays.asList("a", "b", "c"))
+            "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values", Arrays.asList("a", "b", "c"))
         );
 
         TestProcessor testProcessor = new TestProcessor(id -> {
@@ -101,7 +101,7 @@ public void testMetaDataAvailable() throws Exception {
         values.add(new HashMap<>());
         values.add(new HashMap<>());
         IngestDocument ingestDocument = new IngestDocument(
-            "_index", "_type", "_id", null, null, Collections.singletonMap("values", values)
+            "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values", values)
         );
 
         TestProcessor innerProcessor = new TestProcessor(id -> {
@@ -132,7 +132,7 @@ public void testRestOfTheDocumentIsAvailable() throws Exception {
         document.put("values", values);
         document.put("flat_values", new ArrayList<>());
         document.put("other", "value");
-        IngestDocument ingestDocument = new IngestDocument("_index", "_type", "_id", null, null, document);
+        IngestDocument ingestDocument = new IngestDocument("_index", "_type", "_id", null, null, null, null, document);
 
         ForEachProcessor processor = new ForEachProcessor(
             "_tag", "values", new SetProcessor("_tag",
@@ -171,7 +171,7 @@ public String getTag() {
             values.add("");
         }
         IngestDocument ingestDocument = new IngestDocument(
-            "_index", "_type", "_id", null, null, Collections.singletonMap("values", values)
+            "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values", values)
         );
 
         ForEachProcessor processor = new ForEachProcessor("_tag", "values", innerProcessor);
@@ -190,7 +190,7 @@ public void testModifyFieldsOutsideArray() throws Exception {
         values.add(1);
         values.add(null);
         IngestDocument ingestDocument = new IngestDocument(
-                "_index", "_type", "_id", null, null, Collections.singletonMap("values", values)
+                "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values", values)
         );
 
         TemplateScript.Factory template = new TestTemplateService.MockTemplateScript.Factory("errors");
@@ -220,7 +220,7 @@ public void testScalarValueAllowsUnderscoreValueFieldToRemainAccessible() throws
         source.put("_value", "new_value");
         source.put("values", values);
         IngestDocument ingestDocument = new IngestDocument(
-                "_index", "_type", "_id", null, null, source
+                "_index", "_type", "_id", null, null, null, null, source
         );
 
         TestProcessor processor = new TestProcessor(doc -> doc.setFieldValue("_ingest._value",
@@ -251,7 +251,7 @@ public void testNestedForEach() throws Exception {
         values.add(value);
 
         IngestDocument ingestDocument = new IngestDocument(
-                "_index", "_type", "_id", null, null, Collections.singletonMap("values1", values)
+                "_index", "_type", "_id", null, null, null, null, Collections.singletonMap("values1", values)
         );
 
         TestProcessor testProcessor = new TestProcessor(

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

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.ingest.common;
 
+import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.ingest.IngestDocument.MetaData;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.RandomDocumentPicks;
 import org.elasticsearch.ingest.TestTemplateService;
@@ -99,14 +101,30 @@ public void testSetExistingNullFieldWithOverrideDisabled() throws Exception {
         assertThat(ingestDocument.getFieldValue(fieldName, Object.class), equalTo(newValue));
     }
 
-    public void testSetMetadata() throws Exception {
-        IngestDocument.MetaData randomMetaData = randomFrom(IngestDocument.MetaData.values());
+    public void testSetMetadataExceptVersion() throws Exception {
+        MetaData randomMetaData = randomFrom(MetaData.INDEX, MetaData.TYPE, MetaData.ID, MetaData.ROUTING, MetaData.PARENT);
         Processor processor = createSetProcessor(randomMetaData.getFieldName(), "_value", true);
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         processor.execute(ingestDocument);
         assertThat(ingestDocument.getFieldValue(randomMetaData.getFieldName(), String.class), Matchers.equalTo("_value"));
     }
 
+    public void testSetMetadataVersion() throws Exception {
+        long version = randomNonNegativeLong();
+        Processor processor = createSetProcessor(MetaData.VERSION.getFieldName(), version, true);
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(MetaData.VERSION.getFieldName(), Long.class), Matchers.equalTo(version));
+    }
+
+    public void testSetMetadataVersionType() throws Exception {
+        String versionType = randomFrom("internal", "external", "external_gte");
+        Processor processor = createSetProcessor(MetaData.VERSION_TYPE.getFieldName(), versionType, true);
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(MetaData.VERSION_TYPE.getFieldName(), String.class), Matchers.equalTo(versionType));
+    }
+
     private static Processor createSetProcessor(String fieldName, Object fieldValue, boolean overrideEnabled) {
         return new SetProcessor(randomAlphaOfLength(10), new TestTemplateService.MockTemplateScript.Factory(fieldName),
                 ValueSource.wrap(fieldValue, TestTemplateService.instance()), overrideEnabled);

modules/ingest-common/src/test/resources/rest-api-spec/test/ingest/170_version.yml

@@ -0,0 +1,76 @@
+---
+teardown:
+  - do:
+      ingest.delete_pipeline:
+        id: "my_pipeline"
+        ignore: 404
+
+---
+"Test set document version & version type":
+  - do:
+      cluster.health:
+          wait_for_status: green
+
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline1"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "set" : {
+                  "field" : "_version",
+                  "value": 1
+                }
+              },
+              {
+                "set" : {
+                  "field" : "_version_type",
+                  "value": "internal"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      ingest.put_pipeline:
+        id: "my_pipeline2"
+        body:  >
+          {
+            "description": "_description",
+            "processors": [
+              {
+                "set" : {
+                  "field" : "_version",
+                  "value": 1
+                }
+              },
+              {
+                "set" : {
+                  "field" : "_version_type",
+                  "value": "external"
+                }
+              }
+            ]
+          }
+  - match: { acknowledged: true }
+
+  - do:
+      catch: conflict
+      index:
+        index: test
+        type: test
+        id: 1
+        pipeline: "my_pipeline1"
+        body: {}
+
+  - do:
+      index:
+        index: test
+        type: test
+        id: 1
+        pipeline: "my_pipeline2"
+        body: {}
+  - match: { _version: 1 }

qa/smoke-test-ingest-with-all-dependencies/src/test/java/org/elasticsearch/ingest/IngestDocumentMustacheIT.java

@@ -33,7 +33,7 @@ public class IngestDocumentMustacheIT extends AbstractScriptTestCase {
     public void testAccessMetaDataViaTemplate() {
         Map<String, Object> document = new HashMap<>();
         document.put("foo", "bar");
-        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, document);
+        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, null, null, document);
         ingestDocument.setFieldValue(compile("field1"), ValueSource.wrap("1 {{foo}}", scriptService));
         assertThat(ingestDocument.getFieldValue("field1", String.class), equalTo("1 bar"));
 
@@ -48,7 +48,7 @@ public void testAccessMapMetaDataViaTemplate() {
         innerObject.put("baz", "hello baz");
         innerObject.put("qux", Collections.singletonMap("fubar", "hello qux and fubar"));
         document.put("foo", innerObject);
-        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, document);
+        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, null, null, document);
         ingestDocument.setFieldValue(compile("field1"),
                 ValueSource.wrap("1 {{foo.bar}} {{foo.baz}} {{foo.qux.fubar}}", scriptService));
         assertThat(ingestDocument.getFieldValue("field1", String.class), equalTo("1 hello bar hello baz hello qux and fubar"));
@@ -67,7 +67,7 @@ public void testAccessListMetaDataViaTemplate() {
         list.add(value);
         list.add(null);
         document.put("list2", list);
-        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, document);
+        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, null, null, document);
         ingestDocument.setFieldValue(compile("field1"), ValueSource.wrap("1 {{list1.0}} {{list2.0}}", scriptService));
         assertThat(ingestDocument.getFieldValue("field1", String.class), equalTo("1 foo {field=value}"));
     }
@@ -77,7 +77,7 @@ public void testAccessIngestMetadataViaTemplate() {
         Map<String, Object> ingestMap = new HashMap<>();
         ingestMap.put("timestamp", "bogus_timestamp");
         document.put("_ingest", ingestMap);
-        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, document);
+        IngestDocument ingestDocument = new IngestDocument("index", "type", "id", null, null, null, null, document);
         ingestDocument.setFieldValue(compile("ingest_timestamp"),
                 ValueSource.wrap("{{_ingest.timestamp}} and {{_source._ingest.timestamp}}", scriptService));
         assertThat(ingestDocument.getFieldValue("ingest_timestamp", String.class),

qa/smoke-test-ingest-with-all-dependencies/src/test/java/org/elasticsearch/ingest/ValueSourceMustacheIT.java

@@ -64,7 +64,7 @@ public void testValueSourceWithTemplates() {
     }
 
     public void testAccessSourceViaTemplate() {
-        IngestDocument ingestDocument = new IngestDocument("marvel", "type", "id", null, null, new HashMap<>());
+        IngestDocument ingestDocument = new IngestDocument("marvel", "type", "id", null, null, null, null, new HashMap<>());
         assertThat(ingestDocument.hasField("marvel"), is(false));
         ingestDocument.setFieldValue(compile("{{_index}}"), ValueSource.wrap("{{_index}}", scriptService));
         assertThat(ingestDocument.getFieldValue("marvel", String.class), equalTo("marvel"));

server/src/main/java/org/elasticsearch/action/ingest/SimulatePipelineRequest.java

@@ -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;
@@ -195,8 +196,17 @@ 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.fromString(ConfigurationUtils.readStringProperty(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;

server/src/main/java/org/elasticsearch/action/ingest/WriteableIngestDocument.java

@@ -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());