Enable convert processor to support Long and Double. (#27957)

Closes #23085

docs/reference/ingest/ingest-node.asciidoc

@@ -778,16 +778,17 @@ Accepts a single value or an array of values.
 Converts an existing field's value to a different type, such as converting a string to an integer.
 If the field value is an array, all members will be converted.
 
-The supported types include: `integer`, `float`, `string`, `boolean`, and `auto`.
+The supported types include: `integer`, `long`, `float`, `double`, `string`, `boolean`, and `auto`.
 
 Specifying `boolean` will set the field to true if its string value is equal to `true` (ignore case), to
 false if its string value is equal to `false` (ignore case), or it will throw an exception otherwise.
 
 Specifying `auto` will attempt to convert the string-valued `field` into the closest non-string type.
-For example, a field whose value is `"true"` will be converted to its respective boolean type: `true`. And
-a value of `"242.15"` will "automatically" be converted to `242.15` of type `float`. If a provided field cannot
-be appropriately converted, the Convert Processor will still process successfully and leave the field value as-is. In
-such a case, `target_field` will still be updated with the unconverted field value.
+For example, a field whose value is `"true"` will be converted to its respective boolean type: `true`. Do note
+that float takes precedence of double in `auto`. A value of `"242.15"` will "automatically" be converted to
+`242.15` of type `float`. If a provided field cannot be appropriately converted, the Convert Processor will
+still process successfully and leave the field value as-is. In such a case, `target_field` will
+still be updated with the unconverted field value.
 
 [[convert-options]]
 .Convert Options

modules/ingest-common/src/main/java/org/elasticsearch/ingest/common/ConvertProcessor.java

@@ -48,6 +48,24 @@ public final class ConvertProcessor extends AbstractProcessor {
                 }
 
             }
+        }, LONG {
+            @Override
+            public Object convert(Object value) {
+                try {
+                    return Long.parseLong(value.toString());
+                } catch(NumberFormatException e) {
+                    throw new IllegalArgumentException("unable to convert [" + value + "] to long", e);
+                }
+            }
+        }, DOUBLE {
+            @Override
+            public Object convert(Object value) {
+                try {
+                    return Double.parseDouble(value.toString());
+                } catch(NumberFormatException e) {
+                    throw new IllegalArgumentException("unable to convert [" + value + "] to double", e);
+                }
+            }
         }, FLOAT {
             @Override
             public Object convert(Object value) {
@@ -81,13 +99,19 @@ public final class ConvertProcessor extends AbstractProcessor {
                 }
                 try {
                     return BOOLEAN.convert(value);
-                } catch (IllegalArgumentException e) { }
+                } catch (IllegalArgumentException e) {}
                 try {
                     return INTEGER.convert(value);
                 } catch (IllegalArgumentException e) {}
+                try {
+                    return LONG.convert(value);
+                } catch (IllegalArgumentException e) {}
                 try {
                     return FLOAT.convert(value);
                 } catch (IllegalArgumentException e) {}
+                try {
+                    return DOUBLE.convert(value);
+                } catch (IllegalArgumentException e) {}
                 return value;
             }
         };

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

@@ -36,6 +36,7 @@ import static org.elasticsearch.ingest.common.ConvertProcessor.Type;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.sameInstance;
+import static org.hamcrest.Matchers.not;
 
 public class ConvertProcessorTests extends ESTestCase {
 
@@ -79,6 +80,92 @@ public class ConvertProcessorTests extends ESTestCase {
         }
     }
 
+    public void testConvertLong() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        Map<String, Long> expectedResult = new HashMap<>();
+        long randomLong = randomLong();
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, randomLong);
+        expectedResult.put(fieldName, randomLong);
+
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.LONG, false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(fieldName, Long.class), equalTo(randomLong));
+    }
+
+    public void testConvertLongList() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        int numItems = randomIntBetween(1, 10);
+        List<String> fieldValue = new ArrayList<>();
+        List<Long> expectedList = new ArrayList<>();
+        for (int j = 0; j < numItems; j++) {
+            long randomLong = randomLong();
+            fieldValue.add(Long.toString(randomLong));
+            expectedList.add(randomLong);
+        }
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, fieldValue);
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.LONG, false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(fieldName, List.class), equalTo(expectedList));
+    }
+
+    public void testConvertLongError() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
+        String fieldName = RandomDocumentPicks.randomFieldName(random());
+        String value = "string-" + randomAlphaOfLengthBetween(1, 10);
+        ingestDocument.setFieldValue(fieldName, value);
+
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.LONG, false);
+        try {
+            processor.execute(ingestDocument);
+            fail("processor execute should have failed");
+        } catch(IllegalArgumentException e) {
+            assertThat(e.getMessage(), equalTo("unable to convert [" + value + "] to long"));
+        }
+    }
+
+    public void testConvertDouble() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        Map<String, Double> expectedResult = new HashMap<>();
+        double randomDouble = randomDouble();
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, randomDouble);
+        expectedResult.put(fieldName, randomDouble);
+
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.DOUBLE, false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(fieldName, Double.class), equalTo(randomDouble));
+    }
+
+    public void testConvertDoubleList() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
+        int numItems = randomIntBetween(1, 10);
+        List<String> fieldValue = new ArrayList<>();
+        List<Double> expectedList = new ArrayList<>();
+        for (int j = 0; j < numItems; j++) {
+            double randomDouble = randomDouble();
+            fieldValue.add(Double.toString(randomDouble));
+            expectedList.add(randomDouble);
+        }
+        String fieldName = RandomDocumentPicks.addRandomField(random(), ingestDocument, fieldValue);
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.DOUBLE, false);
+        processor.execute(ingestDocument);
+        assertThat(ingestDocument.getFieldValue(fieldName, List.class), equalTo(expectedList));
+    }
+
+    public void testConvertDoubleError() throws Exception {
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), new HashMap<>());
+        String fieldName = RandomDocumentPicks.randomFieldName(random());
+        String value = "string-" + randomAlphaOfLengthBetween(1, 10);
+        ingestDocument.setFieldValue(fieldName, value);
+
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), fieldName, fieldName, Type.DOUBLE, false);
+        try {
+            processor.execute(ingestDocument);
+            fail("processor execute should have failed");
+        } catch(IllegalArgumentException e) {
+            assertThat(e.getMessage(), equalTo("unable to convert [" + value + "] to double"));
+        }
+    }
+
     public void testConvertFloat() throws Exception {
         IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random());
         Map<String, Float> expectedResult = new HashMap<>();
@@ -231,6 +318,16 @@ public class ConvertProcessorTests extends ESTestCase {
                     randomValue = randomBoolean;
                     randomValueString = Boolean.toString(randomBoolean);
                     break;
+                case 3:
+                    long randomLong = randomLong();
+                    randomValue = randomLong;
+                    randomValueString = Long.toString(randomLong);
+                    break;
+                case 4:
+                    double randomDouble = randomDouble();
+                    randomValue = randomDouble;
+                    randomValueString = Double.toString(randomDouble);
+                    break;
                 default:
                     throw new UnsupportedOperationException();
             }
@@ -342,6 +439,28 @@ public class ConvertProcessorTests extends ESTestCase {
         assertThat(convertedValue, equalTo(randomInt));
     }
 
+    public void testAutoConvertMatchLong() throws Exception {
+        long randomLong = randomLong();
+        String randomString = Long.toString(randomLong);
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.singletonMap("field", randomString));
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), "field", "field", Type.AUTO, false);
+        processor.execute(ingestDocument);
+        Object convertedValue = ingestDocument.getFieldValue("field", Object.class);
+        assertThat(convertedValue, equalTo(randomLong));
+    }
+
+    public void testAutoConvertDoubleNotMatched() throws Exception {
+        double randomDouble = randomDouble();
+        String randomString = Double.toString(randomDouble);
+        float randomFloat  = Float.parseFloat(randomString);
+        IngestDocument ingestDocument = RandomDocumentPicks.randomIngestDocument(random(), Collections.singletonMap("field", randomString));
+        Processor processor = new ConvertProcessor(randomAlphaOfLength(10), "field", "field", Type.AUTO, false);
+        processor.execute(ingestDocument);
+        Object convertedValue = ingestDocument.getFieldValue("field", Object.class);
+        assertThat(convertedValue, not(randomDouble));
+        assertThat(convertedValue, equalTo(randomFloat));
+    }
+
     public void testAutoConvertMatchFloat() throws Exception {
         float randomFloat = randomFloat();
         String randomString = Float.toString(randomFloat);