Handling mysql decimal data types with precision 19 or higher

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

Author: divbok

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/mysql/MySQLDataType.java

@@ -129,6 +129,8 @@ public boolean isBigIntUnsigned() {
         return this == MySQLDataType.BIGINT_UNSIGNED;
     }
 
+    public boolean isDecimal() { return this == MySQLDataType.DECIMAL; }
+
     public boolean isBit() {
         return this == MySQLDataType.BIT;
     }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/mysql/handler/NumericTypeHandler.java

@@ -4,6 +4,7 @@
 import org.opensearch.dataprepper.plugins.source.rds.datatype.mysql.MySQLDataType;
 import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
 
+import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.util.ArrayList;
 import java.util.BitSet;
@@ -39,7 +40,13 @@ private Number handleNumericType(final MySQLDataType columnType, final Object va
         }
 
         if (value instanceof Number) {
-            return (Number)value;
+            return (Number) value;
+        }
+
+        if (columnType.isDecimal()) {
+            if (value instanceof byte[] || value instanceof Map || value instanceof ArrayList<?>) {
+                return handleByteArray(value);
+            }
         }
 
         throw new IllegalArgumentException("Unsupported value type. The value is of type: " + value.getClass());
@@ -107,4 +114,29 @@ private static BigInteger bitSetToBigInteger(BitSet bitSet) {
         }
         return result;
     }
+
+    private Number handleByteArray(final Object value) {
+        if (value instanceof byte[]) {
+            return new BigDecimal(new BigInteger((byte[]) value));
+        }
+
+        if (value instanceof Map) {
+            Object data = ((Map<?, ?>)value).get(BYTES_KEY);
+            if (data instanceof byte[]) {
+                return new BigDecimal(new BigInteger((byte[]) data));
+            }
+        }
+
+        if (value instanceof ArrayList<?>) {
+            ArrayList<?> list = (ArrayList<?>) value;
+            byte[] bytes = new byte[list.size()];
+            for (int i = 0; i < list.size(); i++) {
+                bytes[i] = ((Number) list.get(i)).byteValue();
+            }
+            return new BigDecimal(new BigInteger(bytes));
+        }
+
+        throw new IllegalArgumentException("Unsupported byte array value type: " + value.getClass());
+    }
+
 }

data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/datatype/mysql/handler/NumericTypeHandlerTest.java

@@ -113,7 +113,12 @@ private static Stream<Arguments> provideNumericTypeData() {
 
                 // BIT tests
                 Arguments.of(MySQLDataType.BIT, "bit_col", BitSet.valueOf(new byte[]{ 4, 3, 2, 1 }), new BigInteger("16909060")), // BitSet interprets the bytes in little-endian order
-                Arguments.of(MySQLDataType.BIT, "bit_col", Map.of("bytes", new byte[]{ 1, 2, 3, 4 }), new BigInteger("16909060")) // Direct BigInteger interprets the bytes in big-endian order.
+                Arguments.of(MySQLDataType.BIT, "bit_col", Map.of("bytes", new byte[]{ 1, 2, 3, 4 }), new BigInteger("16909060")), // Direct BigInteger interprets the bytes in big-endian order.
+
+                //DECIMAL tests represented as byte arrays
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new ArrayList<>(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16)), new BigDecimal("1339673755198158349044581307228491536")),
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", Map.of("bytes", new byte[]{1, 2, 3, 4}), new BigDecimal("16909060")),
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new byte[]{1, 35, 69, 103, -119, -85, -51, -17}, new BigDecimal("81985529216486895"))
         );
     }