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());
+    }
+
 }