Introduced BigDecimalConverter (#4557)

* Introduced BigDecimalConverter that users can use as part of convert_entry_type processor that currently exists. Optionally, users can also specify required scaling needed on the converted

Signed-off-by: Santhosh Gandhe <gandheaz@amazon.com>

* Added Test case for the newly introduced class. Removed * imports as per the review comment

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* Avoiding using a deprecated method. Added additional test cases

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* Additional tests to increase the coverage

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* removed "scale" being the state of BigDecimal converter. We are now passing the scale while converting the instance only when the instance is BigDecimalConverter

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* test case fix to be inline with the previous commit

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* test case fix to be inline with the previous commit

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* addressing review comments

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* renaming bigdecimal to big_decimal

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* Introduced ConverterArguments as a way to pass additional arguments to the converter and avoided conditional statement for calling converter methods

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* Added additional override convert method to reduce the changes across the code

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* additional Test cases to increase the coverage

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

* added additional tests for converter cases

Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

---------

Signed-off-by: Santhosh Gandhe <gandheaz@amazon.com>
Signed-off-by: Santhosh Gandhe <1909520+san81@users.noreply.github.com>

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/DataType.java

@@ -7,6 +7,7 @@
 
 import com.fasterxml.jackson.annotation.JsonCreator;
 
+import java.math.BigDecimal;
 import java.util.Arrays;
 import java.util.ArrayList;
 import java.util.Map;
@@ -49,6 +50,13 @@ public enum DataType {
      */
     DOUBLE("double"),
 
+    /**
+     * Type of <i>BigDecimal</i>. No precision loss possible type. Compatible with the Java <b>BigDecimal</b> primitive data type.
+     *
+     * @since 2.8
+     */
+    BIG_DECIMAL("big_decimal"),
+
     /**
      * Type of <i>map</i>. Compatible with the Java <b>map</b> primitive data type.
      *
@@ -96,20 +104,22 @@ public static boolean isSameType(final Object object, final String option) {
         if (type == null)
             throw new IllegalArgumentException("Unknown DataType");
         switch (type) {
-          case MAP:
-            return (object instanceof Map);
-          case ARRAY:
-            return (object instanceof ArrayList || object.getClass().isArray());
-          case DOUBLE:
-            return (object instanceof Double);
-          case BOOLEAN:
-            return (object instanceof Boolean);
-          case INTEGER:
-            return (object instanceof Integer);
-          case LONG:
-            return (object instanceof Long);
-          default: // STRING
-            return (object instanceof String);
+            case MAP:
+                return (object instanceof Map);
+            case ARRAY:
+                return (object instanceof ArrayList || object.getClass().isArray());
+            case DOUBLE:
+                return (object instanceof Double);
+            case BOOLEAN:
+                return (object instanceof Boolean);
+            case INTEGER:
+                return (object instanceof Integer);
+            case LONG:
+                return (object instanceof Long);
+            case BIG_DECIMAL:
+                return (object instanceof BigDecimal);
+            default: // STRING
+                return (object instanceof String);
         }
     }
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/JacksonEvent.java

@@ -11,6 +11,8 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.databind.cfg.JsonNodeFeature;
+import com.fasterxml.jackson.databind.json.JsonMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.fasterxml.jackson.databind.type.TypeFactory;
@@ -61,11 +63,14 @@ public class JacksonEvent implements Event {
 
     private static final String SEPARATOR = "/";
 
-    private static final ObjectMapper mapper = new ObjectMapper()
+    private static final ObjectMapper mapper = JsonMapper.builder()
+            .disable(JsonNodeFeature.STRIP_TRAILING_BIGDECIMAL_ZEROES)
+            .build()
             .registerModule(new JavaTimeModule())
             .registerModule(new Jdk8Module()); // required for using Optional with Jackson. Ref: https://github.com/FasterXML/jackson-modules-java8
 
-    private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE = new TypeReference<Map<String, Object>>() {
+
+    private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE = new TypeReference<>() {
     };
 
     private final EventMetadata eventMetadata;

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/source/coordinator/SourceCoordinator.java

@@ -122,6 +122,7 @@ public interface SourceCoordinator<T> {
     /**
      * Should be called by the source when it is shutting down to indicate that it will no longer be able to perform work on partitions,
      * or can be called to give up ownership of its partitions in order to pick up new ones with {@link #getNextPartition(Function)} ()}.
+     * @param partitionKey - Key used as the partition key.
      * @param priorityTimestamp - A timestamp that will determine the order that UNASSIGNED partitions are acquired after they are given up.
      * @throws org.opensearch.dataprepper.model.source.coordinator.exceptions.PartitionUpdateException if the partition could not be given up due to some failure
      * @since 2.8

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/BigDecimalConverter.java

@@ -0,0 +1,58 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+ package org.opensearch.dataprepper.typeconverter;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+
+/**
+ * Converter class for BigDecimal data type. By default, it applies zero scaling keeping the original value as it is.
+ * If required, the scale can be set using the setScale method.
+ */
+public class BigDecimalConverter implements TypeConverter<BigDecimal> {
+
+    public BigDecimal convert(Object source) throws IllegalArgumentException {
+        return this.convert(source, 0);
+    }
+
+    public BigDecimal convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+        return this.convert(source, arguments.getScale());
+    }
+
+    public BigDecimal convert(Object source, int scale) throws IllegalArgumentException {
+        BigDecimal result = null;
+        if (source instanceof String) {
+            result = new BigDecimal((String)source);
+        }
+        else if (source instanceof Float) {
+            result = BigDecimal.valueOf((Float)source);
+        }
+        else if (source instanceof Double) {
+            result = BigDecimal.valueOf((Double)source);
+        }
+        else if (source instanceof Boolean) {
+            result = ((Boolean)source) ? BigDecimal.valueOf(1L) : BigDecimal.valueOf(0L);
+        }
+        else if (source instanceof Integer) {
+            result = BigDecimal.valueOf((Integer)source);
+        }
+        else if (source instanceof Long) {
+            result = BigDecimal.valueOf((Long)source);
+        }
+        else if (source instanceof BigDecimal) {
+            result = ((BigDecimal)source);
+        }
+
+        if(result!=null) {
+            if(scale!=0) {
+                result = result.setScale(scale, RoundingMode.HALF_EVEN);
+            }
+            return result;
+        }
+        throw new IllegalArgumentException("Unsupported type conversion. From Source class: " + source.getClass() + " to BigDecimal");
+    }
+}
+

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/BooleanConverter.java

@@ -6,18 +6,23 @@
 package org.opensearch.dataprepper.typeconverter;
 
 public class BooleanConverter implements TypeConverter<Boolean> {
+
+    public Boolean convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+        return this.convert(source);
+    }
+
     public Boolean convert(Object source) throws IllegalArgumentException {
         if (source instanceof String) {
             return Boolean.parseBoolean((String)source);
         }
         if (source instanceof Number) {
             Number number = (Number)source;
-            return ((number.intValue() != 0) ||
-                    (number.longValue() != 0) ||
+            return ((number instanceof Integer && number.intValue() != 0) ||
+                    (number instanceof Long && number.longValue() != 0) ||
+                    (number instanceof Short && number.shortValue() != 0) ||
+                    (number instanceof Byte && number.byteValue() != 0)) ||
                     (number.floatValue() != 0) ||
-                    (number.doubleValue() != 0) ||
-                    (number.shortValue() != 0) ||
-                    (number.byteValue() != 0)) ? true : false;
+                    (number.doubleValue() != 0);
         }
         if (source instanceof Boolean) {
             return (Boolean)source;

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/ConverterArguments.java

@@ -0,0 +1,10 @@
+package org.opensearch.dataprepper.typeconverter;
+
+/**
+ * Interface for arguments passed to the {@link TypeConverter}
+ *
+ * @since 1.2
+ */
+public interface ConverterArguments {
+    int getScale();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/DoubleConverter.java

@@ -6,6 +6,11 @@
 package org.opensearch.dataprepper.typeconverter;
 
 public class DoubleConverter implements TypeConverter<Double> {
+
+    public Double convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+        return convert(source);
+    }
+
     public Double convert(Object source) throws IllegalArgumentException {
         if (source instanceof String) {
             return Double.parseDouble((String)source);
@@ -17,7 +22,7 @@ public Double convert(Object source) throws IllegalArgumentException {
             return (((Number)source).doubleValue());
         }
         if (source instanceof Boolean) {
-            return (double)(((Boolean)source) ? 1.0 : 0.0);
+            return ((Boolean)source) ? 1.0 : 0.0;
         }
         throw new IllegalArgumentException("Unsupported type conversion. Source class: " + source.getClass());
     }

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/IntegerConverter.java

@@ -7,6 +7,11 @@
 
 import java.math.BigDecimal;
 public class IntegerConverter implements TypeConverter<Integer> {
+
+    public Integer convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+        return convert(source);
+    }
+
     public Integer convert(Object source) throws IllegalArgumentException {
         if (source instanceof String) {
             return Integer.parseInt((String)source);

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/LongConverter.java

@@ -7,6 +7,11 @@
 
  import java.math.BigDecimal;
  public class LongConverter implements TypeConverter<Long> {
+
+     public Long convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+         return this.convert(source);
+     }
+
      public Long convert(Object source) throws IllegalArgumentException {
          if (source instanceof String) {
              return Long.parseLong((String)source);

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/StringConverter.java

@@ -6,6 +6,11 @@
 package org.opensearch.dataprepper.typeconverter;
 
 public class StringConverter implements TypeConverter<String> {
+
+    public String convert(Object source, ConverterArguments arguments) throws IllegalArgumentException {
+        return this.convert(source);
+    }
+
     public String convert(Object source) throws IllegalArgumentException {
         if (source instanceof Number || source instanceof Boolean || source instanceof String)  {
             return source.toString();

data-prepper-api/src/main/java/org/opensearch/dataprepper/typeconverter/TypeConverter.java

@@ -7,4 +7,5 @@
 
 public interface TypeConverter<T> {
   T convert(Object source);
+  T convert(Object source, ConverterArguments arguments);
 }

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/DataTypeTest.java

@@ -15,6 +15,9 @@
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.stream.Stream;
 
@@ -37,15 +40,18 @@ void test_isSameType(Object object, String type, boolean expectedResult) {
     }
 
     private static Stream<Arguments> getSameTypeTestData() {
-        int testArray[] = {1,2};
+        int[] testArray = {1,2};
+        List<Integer> testList = new ArrayList<>();
         return Stream.of(
             Arguments.of(2, "integer", true),
             Arguments.of("testString", "string", true),
             Arguments.of(2L, "long", true),
             Arguments.of(2.0, "double", true),
+            Arguments.of(BigDecimal.valueOf(2.34567), "big_decimal", true),
             Arguments.of(true, "boolean", true),
             Arguments.of(Map.of("k","v"), "map", true),
             Arguments.of(testArray, "array", true),
+            Arguments.of(testList, "array", true),
             Arguments.of(2.0, "integer", false),
             Arguments.of(2, "string", false),
             Arguments.of("testString", "long", false),

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/DefaultEventMetadataTest.java

@@ -176,6 +176,20 @@ public void testAttributes_without_attributes_is_empty() {
 
     }
 
+    @Test
+    public void testAttributes_with_attributes_of_not_map_type() {
+        Object v1 = new Object();
+        eventMetadata = DefaultEventMetadata.builder()
+                .withEventType(testEventType)
+                .withTimeReceived(testTimeReceived)
+                .withAttributes(Map.of("key1", v1))
+                .build();
+        assertThat(eventMetadata.getAttribute("key1"), equalTo(v1));
+        assertThat(eventMetadata.getAttribute("key1/key2/"), equalTo(null));
+        assertThat(eventMetadata.getAttribute("key3"), equalTo(null));
+
+    }
+
     @Test
     public void testBuild_withoutTimeReceived() {
 
@@ -303,6 +317,40 @@ void setUp() {
         @Test
         void equals_returns_false_for_null() {
             assertThat(event.equals(null), equalTo(false));
+            assertThat(event.equals(new Object()), equalTo(false));
+        }
+
+        @Test
+        void equals_returns_false_when_timeinstance_not_match() {
+            DefaultEventMetadata newEvent = DefaultEventMetadata.builder()
+                    .withEventType(eventType)
+                    .withTimeReceived(Instant.now())
+                    .withAttributes(Collections.singletonMap(attributeKey, attributeValue))
+                    .build();
+            assertThat(event.equals(newEvent), equalTo(false));
+        }
+
+        @Test
+        void equals_returns_false_when_attributes_not_match() {
+            String newAttributeKey = UUID.randomUUID().toString();
+            String newAttributeValue = UUID.randomUUID().toString();
+            DefaultEventMetadata newEvent = DefaultEventMetadata.builder()
+                    .withEventType(eventType)
+                    .withTimeReceived(timeReceived)
+                    .withAttributes(Collections.singletonMap(newAttributeKey, newAttributeValue))
+                    .build();
+            assertThat(event.equals(newEvent), equalTo(false));
+        }
+
+        @Test
+        void equals_returns_false_when_tags_not_match() {
+            DefaultEventMetadata newEvent = DefaultEventMetadata.builder()
+                    .withEventType(eventType)
+                    .withTimeReceived(timeReceived)
+                    .withAttributes(Collections.singletonMap(attributeKey, attributeValue))
+                    .withTags(Set.of("some","new","tag"))
+                    .build();
+            assertThat(event.equals(newEvent), equalTo(false));
         }
 
         @Test