Ingest: Enforce _version metadata not null in sourceAndMetadata map

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

@@ -351,7 +351,7 @@ private void testCommunityIdProcessor(Map<String, Object> source, int seed, Stri
             ignoreMissing
         );
 
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         String hash = output.getFieldValue(DEFAULT_TARGET, String.class, ignoreMissing);

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

@@ -26,15 +26,15 @@ public class DotExpanderProcessorTests extends ESTestCase {
     public void testEscapeFields() throws Exception {
         Map<String, Object> source = new HashMap<>();
         source.put("foo.bar", "baz1");
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         DotExpanderProcessor processor = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
         assertThat(document.getFieldValue("foo.bar", String.class), equalTo("baz1"));
 
         source = new HashMap<>();
         source.put("foo.bar.baz", "value");
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar.baz");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -44,21 +44,23 @@ public void testEscapeFields() throws Exception {
         source = new HashMap<>();
         source.put("foo.bar", "baz1");
         source.put("foo", new HashMap<>(Collections.singletonMap("bar", "baz2")));
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         processor.execute(document);
-        assertThat(document.getSourceAndMetadata().size(), equalTo(1));
+        assertThat(document.getSource().size(), equalTo(1));
+        assertThat(document.getMetadata().size(), equalTo(1)); // the default version
         assertThat(document.getFieldValue("foo.bar", List.class).size(), equalTo(2));
         assertThat(document.getFieldValue("foo.bar.0", String.class), equalTo("baz2"));
         assertThat(document.getFieldValue("foo.bar.1", String.class), equalTo("baz1"));
 
         source = new HashMap<>();
         source.put("foo.bar", "2");
         source.put("foo", new HashMap<>(Collections.singletonMap("bar", 1)));
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         processor.execute(document);
-        assertThat(document.getSourceAndMetadata().size(), equalTo(1));
+        assertThat(document.getSource().size(), equalTo(1));
+        assertThat(document.getMetadata().size(), equalTo(1)); // the default version
         assertThat(document.getFieldValue("foo.bar", List.class).size(), equalTo(2));
         assertThat(document.getFieldValue("foo.bar.0", Integer.class), equalTo(1));
         assertThat(document.getFieldValue("foo.bar.1", String.class), equalTo("2"));
@@ -68,15 +70,15 @@ public void testEscapeFields_valueField() throws Exception {
         Map<String, Object> source = new HashMap<>();
         source.put("foo.bar", "baz1");
         source.put("foo", "baz2");
-        IngestDocument document1 = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document1 = TestIngestDocument.withDefaultVersion(source);
         Processor processor1 = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         // foo already exists and if a leaf field and therefor can't be replaced by a map field:
         Exception e = expectThrows(IllegalArgumentException.class, () -> processor1.execute(document1));
         assertThat(e.getMessage(), equalTo("cannot expand [foo.bar], because [foo] is not an object field, but a value field"));
 
         // so because foo is no branch field but a value field the `foo.bar` field can't be expanded
         // into [foo].[bar], so foo should be renamed first into `[foo].[bar]:
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         Processor processor = new RenameProcessor(
             "_tag",
             null,
@@ -93,7 +95,7 @@ public void testEscapeFields_valueField() throws Exception {
 
         source = new HashMap<>();
         source.put("foo.bar", "baz1");
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -102,7 +104,7 @@ public void testEscapeFields_valueField() throws Exception {
         source = new HashMap<>();
         source.put("foo.bar.baz", "baz1");
         source.put("foo", new HashMap<>(Collections.singletonMap("bar", new HashMap<>())));
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar.baz");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -112,7 +114,7 @@ public void testEscapeFields_valueField() throws Exception {
         source = new HashMap<>();
         source.put("foo.bar.baz", "baz1");
         source.put("foo", new HashMap<>(Collections.singletonMap("bar", "baz2")));
-        IngestDocument document2 = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document2 = TestIngestDocument.withDefaultVersion(source);
         Processor processor2 = new DotExpanderProcessor("_tag", null, null, "foo.bar.baz");
         e = expectThrows(IllegalArgumentException.class, () -> processor2.execute(document2));
         assertThat(e.getMessage(), equalTo("cannot expand [foo.bar.baz], because [foo.bar] is not an object field, but a value field"));
@@ -121,7 +123,7 @@ public void testEscapeFields_valueField() throws Exception {
     public void testEscapeFields_path() throws Exception {
         Map<String, Object> source = new HashMap<>();
         source.put("foo", new HashMap<>(Collections.singletonMap("bar.baz", "value")));
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         DotExpanderProcessor processor = new DotExpanderProcessor("_tag", null, "foo", "bar.baz");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -130,7 +132,7 @@ public void testEscapeFields_path() throws Exception {
 
         source = new HashMap<>();
         source.put("field", new HashMap<>(Collections.singletonMap("foo.bar.baz", "value")));
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, "field", "foo.bar.baz");
         processor.execute(document);
         assertThat(document.getFieldValue("field.foo", Map.class).size(), equalTo(1));
@@ -142,7 +144,7 @@ public void testEscapeFields_doNothingIfFieldNotInSourceDoc() throws Exception {
         // asking to expand a (literal) field that is not present in the source document
         Map<String, Object> source = new HashMap<>();
         source.put("foo.bar", "baz1");
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         // abc.def does not exist in source, so don't mutate document
         DotExpanderProcessor processor = new DotExpanderProcessor("_tag", null, null, "abc.def");
         processor.execute(document);
@@ -160,7 +162,7 @@ public void testEscapeFields_doNothingIfFieldNotInSourceDoc() throws Exception {
         Map<String, Object> inner = new HashMap<>();
         inner.put("bar", "baz1");
         source.put("foo", inner);
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         // foo.bar, the literal value (as opposed to nested value) does not exist in source, so don't mutate document
         processor = new DotExpanderProcessor("_tag", null, null, "foo.bar");
         processor.execute(document);
@@ -178,7 +180,7 @@ public void testOverride() throws Exception {
         inner.put("qux", "quux");
         source.put("foo", inner);
         source.put("foo.bar", "baz2");
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         DotExpanderProcessor processor = new DotExpanderProcessor("_tag", null, null, "foo.bar", true);
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(2));
@@ -190,7 +192,7 @@ public void testWildcard() throws Exception {
         Map<String, Object> source = new HashMap<>();
         source.put("foo.bar", "baz");
         source.put("qux.quux", "corge");
-        IngestDocument document = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument document = TestIngestDocument.withDefaultVersion(source);
         DotExpanderProcessor processor = new DotExpanderProcessor("_tag", null, null, "*");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -202,7 +204,7 @@ public void testWildcard() throws Exception {
         Map<String, Object> inner = new HashMap<>();
         inner.put("bar.baz", "qux");
         source.put("foo", inner);
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, "foo", "*");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));
@@ -213,7 +215,7 @@ public void testWildcard() throws Exception {
         inner = new HashMap<>();
         inner.put("bar.baz", "qux");
         source.put("foo", inner);
-        document = TestIngestDocument.ofSourceAndMetadata(source);
+        document = TestIngestDocument.withDefaultVersion(source);
         processor = new DotExpanderProcessor("_tag", null, null, "*");
         processor.execute(document);
         assertThat(document.getFieldValue("foo", Map.class).size(), equalTo(1));

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

@@ -227,7 +227,7 @@ private String doTestFingerprint(
         MessageDigest md = MessageDigest.getInstance(FingerprintProcessor.Factory.DEFAULT_METHOD);
         expectedBytes = md.digest(expectedBytes);
 
-        var input = TestIngestDocument.ofSourceAndMetadata(inputMap);
+        var input = TestIngestDocument.withDefaultVersion(inputMap);
         var output = fp.execute(input);
         assertTrue(output.hasField("fingerprint"));
         String fingerprint = output.getFieldValue("fingerprint", String.class);
@@ -257,7 +257,7 @@ public void testMethod() throws Exception {
             config.put("method", FingerprintProcessor.Factory.SUPPORTED_DIGESTS[k]);
 
             FingerprintProcessor fp = factory.create(null, randomAlphaOfLength(10), null, config);
-            var input = TestIngestDocument.ofSourceAndMetadata(inputMap);
+            var input = TestIngestDocument.withDefaultVersion(inputMap);
             var output = fp.execute(input);
             assertTrue(output.hasField("fingerprint"));
             String fingerprint = output.getFieldValue("fingerprint", String.class);
@@ -394,7 +394,7 @@ private void doTestObjectTraversal(Map<String, Object> inputMap, List<String> fi
             expectedBytes = concatBytes(expectedBytes, toBytes(value));
         }
 
-        var input = TestIngestDocument.ofSourceAndMetadata(inputMap);
+        var input = TestIngestDocument.withDefaultVersion(inputMap);
         var output = fp.execute(input);
         var hasher = (TestHasher) threadLocalHasher.get();
         assertThat(hasher.getBytesSeen(), equalTo(expectedBytes));

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

@@ -150,7 +150,7 @@ public void testReadFromField() throws Exception {
             null,
             config
         );
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
         String hash = output.getFieldValue(DEFAULT_TARGET, String.class);
         assertThat(hash, equalTo("external"));
@@ -196,7 +196,7 @@ private void testNetworkDirectionProcessor(
             config
         );
 
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         String hash = output.getFieldValue(DEFAULT_TARGET, String.class, ignoreMissing);

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

@@ -80,7 +80,7 @@ public void testUseRoot() throws Exception {
 
         var processor = new RegisteredDomainProcessor(null, null, "domain", "", false);
 
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         String domain = output.getFieldValue(domainField, String.class);
@@ -131,7 +131,7 @@ private void testRegisteredDomainProcessor(
 
         var processor = new RegisteredDomainProcessor(null, null, "domain", "url", ignoreMissing);
 
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         String domain = output.getFieldValue(domainField, String.class, expectedDomain == null);

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

@@ -179,7 +179,7 @@ public void testRenameAtomicOperationRemoveFails() throws Exception {
     public void testRenameLeafIntoBranch() throws Exception {
         Map<String, Object> source = new HashMap<>();
         source.put("foo", "bar");
-        IngestDocument ingestDocument = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument ingestDocument = TestIngestDocument.withDefaultVersion(source);
         Processor processor1 = createRenameProcessor("foo", "foo.bar", false);
         processor1.execute(ingestDocument);
         assertThat(ingestDocument.getFieldValue("foo", Map.class), equalTo(Collections.singletonMap("bar", "bar")));

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

@@ -100,7 +100,7 @@ public void testSplitAppendable() throws Exception {
         Processor splitProcessor = (new SplitProcessor.Factory()).create(null, null, null, splitConfig);
         Map<String, Object> source = new HashMap<>();
         source.put("flags", "new|hot|super|fun|interesting");
-        IngestDocument ingestDocument = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument ingestDocument = TestIngestDocument.withDefaultVersion(source);
         splitProcessor.execute(ingestDocument);
         @SuppressWarnings("unchecked")
         List<String> flags = (List<String>) ingestDocument.getFieldValue("flags", List.class);

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

@@ -186,7 +186,7 @@ public void testRemoveIfSuccessfulDoesNotRemoveTargetField() throws Exception {
 
         Map<String, Object> source = new HashMap<>();
         source.put(field, "http://www.google.com");
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         Map<String, Object> expectedSourceAndMetadata = new HashMap<>();
@@ -202,7 +202,7 @@ public void testInvalidUri() {
 
         Map<String, Object> source = new HashMap<>();
         source.put("field", uri);
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
 
         IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> processor.execute(input));
         assertThat(e.getMessage(), containsString("unable to parse URI [" + uri + "]"));
@@ -218,7 +218,7 @@ private void testUriParsing(boolean keepOriginal, boolean removeIfSuccessful, St
 
         Map<String, Object> source = new HashMap<>();
         source.put("field", uri);
-        IngestDocument input = TestIngestDocument.ofSourceAndMetadata(source);
+        IngestDocument input = TestIngestDocument.withDefaultVersion(source);
         IngestDocument output = processor.execute(input);
 
         Map<String, Object> expectedSourceAndMetadata = new HashMap<>();

server/src/main/java/org/elasticsearch/ingest/IngestDocument.java

@@ -97,7 +97,13 @@ public IngestDocument(Map<String, Object> sourceAndMetadata, Map<String, Object>
             IngestSourceAndMetadata.getTimestamp(ingestMetadata),
             IngestSourceAndMetadata.VALIDATORS
         );
-        this.ingestMetadata = ingestMetadata;
+        this.ingestMetadata = new HashMap<>(ingestMetadata);
+        this.ingestMetadata.computeIfPresent(TIMESTAMP, (k, v) -> {
+            if (v instanceof String) {
+                return this.sourceAndMetadata.getTimestamp();
+            }
+            return v;
+        });
     }
 
     /**

server/src/main/java/org/elasticsearch/ingest/IngestSourceAndMetadata.java

@@ -54,7 +54,7 @@ class IngestSourceAndMetadata extends AbstractMap<String, Object> {
         IngestDocument.Metadata.ROUTING.getFieldName(),
         IngestSourceAndMetadata::stringValidator,
         IngestDocument.Metadata.VERSION.getFieldName(),
-        IngestSourceAndMetadata::longValidator,
+        IngestSourceAndMetadata::versionValidator,
         IngestDocument.Metadata.VERSION_TYPE.getFieldName(),
         IngestSourceAndMetadata::versionTypeValidator,
         IngestDocument.Metadata.DYNAMIC_TEMPLATES.getFieldName(),
@@ -151,8 +151,11 @@ public static ZonedDateTime getTimestamp(Map<String, Object> ingestMetadata) {
         if (ingestMetadata == null) {
             return null;
         }
-        if (ingestMetadata.get(IngestDocument.TIMESTAMP)instanceof ZonedDateTime timestamp) {
+        Object ts = ingestMetadata.get(IngestDocument.TIMESTAMP);
+        if (ts instanceof ZonedDateTime timestamp) {
             return timestamp;
+        } else if (ts instanceof String str) {
+            return ZonedDateTime.parse(str);
         }
         return null;
     }
@@ -532,6 +535,16 @@ protected static void longValidator(String key, Object value) {
         );
     }
 
+    /**
+     * Version must be non-null and representable as a long without loss of precision
+     */
+    protected static void versionValidator(String key, Object value) {
+        if (value == null) {
+            throw new IllegalArgumentException(key + " cannot be null");
+        }
+        longValidator(key, value);
+    }
+
     /**
      * Allow lower case Strings that map to VersionType values, or null
      */