Add unit test for MySQL DataType handler

Signed-off-by: Dinu John <[email protected]>

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

@@ -167,6 +167,8 @@ public long getUnsignedMask() {
                 return 0xFFFFFFL;
             case INT_UNSIGNED:
                 return 0xFFFFFFFFL;
+            case BIGINT_UNSIGNED:
+                return 0xFFFFFFFFFFFFFFFFL;
             default:
                 throw new UnsupportedOperationException("No mask for non-unsigned type: " + this);
         }

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

@@ -4,17 +4,35 @@
 import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
 import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
 
+import java.math.BigInteger;
+
 public class NumericTypeHandler implements DataTypeHandler {
 
     @Override
     public String handle(final MySQLDataType columnType, final String columnName, final Object value,
                          final TableMetadata metadata) {
-        return handleNumericType(columnType, value);
+        if (value == null) {
+            return null;
+        }
+
+        if (!columnType.isNumeric()) {
+            throw new IllegalArgumentException("ColumnType is not numeric: " + columnType);
+        }
+
+        if (!(value instanceof Number)) {
+            throw new IllegalArgumentException("Value is not a number: " + value);
+        }
+
+        return handleNumericType(columnType, (Number) value);
     }
 
-    private String handleNumericType(final MySQLDataType columnType, final Object value) {
+    private String handleNumericType(final MySQLDataType columnType, final Number value) {
         if (columnType.isUnsigned()) {
-            return handleUnsignedNumber((Number) value, columnType.getUnsignedMask());
+            if(columnType == MySQLDataType.BIGINT_UNSIGNED) {
+                return handleUnsignedDouble(value);
+            } else {
+                return handleUnsignedNumber(value, columnType.getUnsignedMask());
+            }
         }
         return value.toString();
     }
@@ -23,4 +41,15 @@ private String handleUnsignedNumber(final Number value, final long mask) {
         final long longVal = value.longValue();
         return String.valueOf(longVal < 0 ? longVal & mask : longVal);
     }
+
+    private String handleUnsignedDouble(final Number value) {
+        long longVal = value.longValue();
+        if (longVal < 0) {
+            return BigInteger.valueOf(longVal & Long.MAX_VALUE)
+                    .add(BigInteger.valueOf(Long.MAX_VALUE))
+                    .add(BigInteger.ONE)
+                    .toString();
+        }
+        return String.valueOf(longVal);
+    }
 }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/impl/SpatialTypeHandler.java

@@ -9,10 +9,7 @@ public class SpatialTypeHandler implements DataTypeHandler {
     @Override
     public String handle(final MySQLDataType columnType, final String columnName, final Object value,
                          final TableMetadata metadata) {
-        // Geometry types are typically returned as WKB (Well-Known Binary)
-        // Convert to WKT (Well-Known Text) or handle according to your needs
-        //return Base64.getEncoder().encodeToString((byte[]) value);
-        //return value.toString();
+        // TODO: Implement the transformation
         return new String((byte[]) value);
     }
 }

data-prepper-plugins/rds-source/src/main/java/org/opensearch/dataprepper/plugins/source/rds/datatype/impl/TemporalTypeHandler.java

@@ -4,42 +4,22 @@
 import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
 import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
 
-import java.time.LocalDate;
-import java.time.LocalTime;
-import java.time.format.DateTimeFormatter;
-
 public class TemporalTypeHandler implements DataTypeHandler {
 
     @Override
     public String handle(final MySQLDataType columnType, final String columnName, final Object value,
                          final TableMetadata metadata) {
         // Date and Time types
         switch (columnType) {
-            // Date and Time types
+            // TODO: Implement the transformation
             case DATE:
-                return value instanceof LocalDate ?
-                        ((LocalDate) value).format(DateTimeFormatter.ISO_LOCAL_DATE) :
-                        value.toString();
-
             case TIME:
-                return value instanceof LocalTime ?
-                        ((LocalTime) value).format(DateTimeFormatter.ISO_LOCAL_TIME) :
-                        value.toString();
-
             case TIMESTAMP:
             case DATETIME:
-                /*if (value instanceof Timestamp) {
-                    return ((LocalDateTime) value).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-                } else if (value instanceof LocalDateTime) {
-                    return ((LocalDateTime) value).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
-                }*/
-                return value.toString();
-
             case YEAR:
                 return value.toString();
             default:
                 throw new IllegalArgumentException("Unsupported temporal data type: " + columnType);
         }
-
     }
 }

data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/datatype/impl/BinaryTypeHandlerTest.java

@@ -0,0 +1,31 @@
+package org.opensearch.dataprepper.plugins.source.rds.datatype.impl;
+
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.DataTypeHandler;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
+import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.util.Base64;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+public class BinaryTypeHandlerTest {
+
+    @Test
+    public void test_handle() {
+        final DataTypeHandler handler = new BinaryTypeHandler();
+        final MySQLDataType columnType = MySQLDataType.BINARY;
+        final String columnName = "binaryColumn";
+        final byte[] testData = "Test binary data".getBytes();
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(), List.of(columnName), List.of(columnName),
+                Collections.emptyMap(), Collections.emptyMap());
+        String result = handler.handle(columnType, columnName, testData, metadata);
+
+        assertThat(result, is(Base64.getEncoder().encodeToString(testData)));
+    }
+}

data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/datatype/impl/JsonTypeHandlerTest.java

@@ -0,0 +1,31 @@
+package org.opensearch.dataprepper.plugins.source.rds.datatype.impl;
+
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.DataTypeHandler;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
+import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class JsonTypeHandlerTest {
+
+    @Test
+    public void test_handle() {
+        final DataTypeHandler handler = new JsonTypeHandler();
+        final MySQLDataType columnType = MySQLDataType.JSON;
+        final String columnName = "jsonColumn";
+        final String jsonValue = "{\"key\":\"value\"}";
+        final byte[] testData = jsonValue.getBytes();
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(), List.of(columnName), List.of(columnName),
+                Collections.emptyMap(), Collections.emptyMap());
+        String result = handler.handle(columnType, columnName, testData, metadata);
+
+        assertThat(result, is(jsonValue));
+    }
+}

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

@@ -0,0 +1,123 @@
+package org.opensearch.dataprepper.plugins.source.rds.datatype.impl;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.DataTypeHandler;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
+import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+public class NumericTypeHandlerTest {
+    @ParameterizedTest
+    @MethodSource("provideNumericTypeData")
+    public void test_handle(final MySQLDataType mySQLDataType, final String columnName, final Object value, final Object expectedValue) {
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(), List.of(columnName), List.of(columnName),
+                Collections.emptyMap(), Collections.emptyMap());
+        final DataTypeHandler numericTypeHandler = new NumericTypeHandler();
+        String result = numericTypeHandler.handle(mySQLDataType, columnName, value, metadata);
+
+        assertThat(result, is(expectedValue));
+    }
+
+    private static Stream<Arguments> provideNumericTypeData() {
+        return Stream.of(
+                // TINYINT tests (signed: -128 to 127)
+                Arguments.of(MySQLDataType.TINYINT, "tinyint_col", (byte)1, "1"),
+                Arguments.of(MySQLDataType.TINYINT, "tinyint_col", (byte)-128, "-128"),
+                Arguments.of(MySQLDataType.TINYINT, "tinyint_col", (byte)127, "127"),
+                Arguments.of(MySQLDataType.TINYINT, "tinyint_col", null, null),
+
+                // TINYINT UNSIGNED tests (0 to 255)
+                Arguments.of(MySQLDataType.TINYINT_UNSIGNED, "tinyint_unsigned_col", (short)0, "0"),
+                Arguments.of(MySQLDataType.TINYINT_UNSIGNED, "tinyint_unsigned_col", (short)255, "255"),
+                Arguments.of(MySQLDataType.TINYINT_UNSIGNED, "tinyint_unsigned_col", (short)128, "128"),
+
+                // SMALLINT tests (signed: -32,768 to 32,767)
+                Arguments.of(MySQLDataType.SMALLINT, "smallint_col", (short)32767, "32767"),
+                Arguments.of(MySQLDataType.SMALLINT, "smallint_col", (short)-32768, "-32768"),
+                Arguments.of(MySQLDataType.SMALLINT, "smallint_col", (short)0, "0"),
+
+                // SMALLINT UNSIGNED tests (0 to 65,535)
+                Arguments.of(MySQLDataType.SMALLINT_UNSIGNED, "smallint_unsigned_col", 0, "0"),
+                Arguments.of(MySQLDataType.SMALLINT_UNSIGNED, "smallint_unsigned_col", 65535, "65535"),
+                Arguments.of(MySQLDataType.SMALLINT_UNSIGNED, "smallint_unsigned_col", 32768, "32768"),
+
+                // INTEGER/INT tests (signed: -2,147,483,648 to 2,147,483,647)
+                Arguments.of(MySQLDataType.INT, "int_col", 2147483647, "2147483647"),
+                Arguments.of(MySQLDataType.INT, "int_col", -2147483648, "-2147483648"),
+                Arguments.of(MySQLDataType.INT, "int_col", 0, "0"),
+
+                // INTEGER/INT UNSIGNED tests (0 to 4,294,967,295)
+                Arguments.of(MySQLDataType.INT_UNSIGNED, "int_unsigned_col", 4294967295L, "4294967295"),
+                Arguments.of(MySQLDataType.INT_UNSIGNED, "int_unsigned_col", 0L, "0"),
+                Arguments.of(MySQLDataType.INT_UNSIGNED, "int_unsigned_col", 2147483648L, "2147483648"),
+
+                // BIGINT tests (signed: -9,223,372,036,854,775,808 to 9,223,372,036,854,775,807)
+                Arguments.of(MySQLDataType.BIGINT, "bigint_col", 9223372036854775807L, "9223372036854775807"),
+                Arguments.of(MySQLDataType.BIGINT, "bigint_col", -9223372036854775808L, "-9223372036854775808"),
+                Arguments.of(MySQLDataType.BIGINT, "bigint_col", 0L, "0"),
+
+                // BIGINT UNSIGNED tests (0 to 18,446,744,073,709,551,615)
+                Arguments.of(MySQLDataType.BIGINT_UNSIGNED, "bigint_unsigned_col", new BigInteger("18446744073709551615"), "18446744073709551615"),
+                Arguments.of(MySQLDataType.BIGINT_UNSIGNED, "bigint_unsigned_col", BigInteger.ZERO, "0"),
+                Arguments.of(MySQLDataType.BIGINT_UNSIGNED, "bigint_unsigned_col", new BigInteger("9223372036854775808"), "9223372036854775808"),
+
+                // DECIMAL/NUMERIC tests
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new BigDecimal("123.45"), "123.45"),
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new BigDecimal("-123.45"), "-123.45"),
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new BigDecimal("0.0"), "0.0"),
+                Arguments.of(MySQLDataType.DECIMAL, "decimal_col", new BigDecimal("999999.99"), "999999.99"),
+
+                // FLOAT tests
+                Arguments.of(MySQLDataType.FLOAT, "float_col", 123.45f, "123.45"),
+                Arguments.of(MySQLDataType.FLOAT, "float_col", -123.45f, "-123.45"),
+                Arguments.of(MySQLDataType.FLOAT, "float_col", 0.0f, "0.0"),
+                Arguments.of(MySQLDataType.FLOAT, "float_col", Float.MAX_VALUE, String.valueOf(Float.MAX_VALUE)),
+
+                // DOUBLE tests
+                Arguments.of(MySQLDataType.DOUBLE, "double_col", 123.45678901234, "123.45678901234"),
+                Arguments.of(MySQLDataType.DOUBLE, "double_col", -123.45678901234, "-123.45678901234"),
+                Arguments.of(MySQLDataType.DOUBLE, "double_col", 0.0, "0.0"),
+                Arguments.of(MySQLDataType.DOUBLE, "double_col", Double.MAX_VALUE, String.valueOf(Double.MAX_VALUE))
+        );
+    }
+
+    @Test
+    public void test_handleInvalidType() {
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(),
+                List.of("invalid_col"), List.of("invalid_col"),
+                Collections.emptyMap(), Collections.emptyMap());
+        final DataTypeHandler numericTypeHandler = new NumericTypeHandler();
+
+        assertThrows(IllegalArgumentException.class, () -> {
+            numericTypeHandler.handle(MySQLDataType.INT_UNSIGNED, "invalid_col", "not_a_number", metadata);
+        });
+    }
+
+    @Test
+    public void test_handleInvalidValue() {
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(),
+                List.of("int_col"), List.of("int_col"),
+                Collections.emptyMap(), Collections.emptyMap());
+        final DataTypeHandler numericTypeHandler = new NumericTypeHandler();
+
+        assertThrows(IllegalArgumentException.class, () -> {
+            numericTypeHandler.handle(MySQLDataType.INT, "int_col", "not_a_number", metadata);
+        });
+    }
+}

data-prepper-plugins/rds-source/src/test/java/org/opensearch/dataprepper/plugins/source/rds/datatype/impl/SpatialTypeHandlerTest.java

@@ -0,0 +1,30 @@
+package org.opensearch.dataprepper.plugins.source.rds.datatype.impl;
+
+import org.junit.jupiter.api.Test;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.DataTypeHandler;
+import org.opensearch.dataprepper.plugins.source.rds.datatype.MySQLDataType;
+import org.opensearch.dataprepper.plugins.source.rds.model.TableMetadata;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class SpatialTypeHandlerTest {
+
+    @Test
+    public void test_handle() {
+        final DataTypeHandler handler = new SpatialTypeHandler();
+        final MySQLDataType columnType = MySQLDataType.GEOMETRY;
+        final String columnName = "geometryColumn";
+        final String value = UUID.randomUUID().toString();
+        final TableMetadata metadata = new TableMetadata(
+                UUID.randomUUID().toString(), UUID.randomUUID().toString(), List.of(columnName), List.of(columnName),
+                Collections.emptyMap(), Collections.emptyMap());
+        String result = handler.handle(columnType, columnName, value.getBytes(), metadata);
+
+        assertThat(result, is(value));
+    }
+}