preventing bit from going negative (#3038)

Author: VardhanThigle

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/jdbc/rowmapper/provider/MysqlJdbcValueMappings.java

@@ -27,7 +27,6 @@
 import com.google.re2j.Matcher;
 import com.google.re2j.Pattern;
 import java.math.BigDecimal;
-import java.math.BigInteger;
 import java.math.RoundingMode;
 import java.nio.ByteBuffer;
 import java.sql.Blob;
@@ -53,10 +52,6 @@ public class MysqlJdbcValueMappings implements JdbcValueMappingsProvider {
    */
   private static final ResultSetValueMapper<?> valuePassThrough = (value, schema) -> value;
 
-  /** Map the bytes returned by BIT(N) type to long. MySql Bit(N) has max 64 bits. */
-  private static final ResultSetValueMapper<byte[]> bytesToLong =
-      (value, schema) -> new BigInteger(value).longValue();
-
   /** Map BigDecimal type to Byte array. */
   private static final ResultSetValueMapper<BigDecimal> bigDecimalToByteArray =
       (value, schema) ->
@@ -197,7 +192,7 @@ private static long instantToMicro(Instant instant) {
           .put("BIGINT", Pair.of(ResultSet::getLong, valuePassThrough))
           .put("BIGINT UNSIGNED", Pair.of(ResultSet::getBigDecimal, bigDecimalToAvroNumber))
           .put("BINARY", Pair.of(ResultSet::getBytes, bytesToHexString))
-          .put("BIT", Pair.of(ResultSet::getBytes, bytesToLong))
+          .put("BIT", Pair.of(ResultSet::getLong, valuePassThrough))
           .put("BLOB", Pair.of(ResultSet::getBlob, blobToHexString))
           .put("BOOL", Pair.of(ResultSet::getInt, valuePassThrough))
           .put("CHAR", Pair.of(ResultSet::getString, valuePassThrough))

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/reader/io/jdbc/rowmapper/JdbcSourceRowMapperTest.java

@@ -291,9 +291,9 @@ private static ImmutableList<Column> mySQLColumns() {
                 .build())
         .add(
             Column.builder()
-                .derbyColumnType("CHAR(8) FOR BIT DATA") // Derby mapping for Bit
+                .derbyColumnType("BIGINT") // Derby mapping for Bit
                 .sourceColumnType("BIT")
-                .inputValue(ByteBuffer.allocate(8).putLong(5L).array())
+                .inputValue(5L)
                 .mappedValue(5L)
                 .build())
         .add(

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/MySQLDataTypesIT.java

@@ -158,6 +158,8 @@ private Map<String, List<Map<String, Object>>> getExpectedData() {
         "binary",
         createRows("binary", "eDU4MD" + repeatString("A", 334), repeatString("/", 340), "NULL"));
     expectedData.put("bit", createRows("bit", "f/////////8=", "NULL"));
+    expectedData.put("bit8", createRows("bit8", "0", "255", "NULL"));
+    expectedData.put("bit1", createRows("bit1", "false", "true", "NULL"));
     expectedData.put("blob", createRows("blob", "eDU4MDA=", repeatString("/", 87380), "NULL"));
     expectedData.put("bool", createRows("bool", "false", "true", "NULL"));
     expectedData.put("boolean", createRows("boolean", "false", "true", "NULL"));

v2/sourcedb-to-spanner/src/test/resources/DataTypesIT/mysql-data-types.sql

@@ -152,6 +152,16 @@ CREATE TABLE `bit_table` (
     `bit_col` BIT(64) DEFAULT NULL
 );
 
+CREATE TABLE `bit8_table` (
+     `id` INT PRIMARY KEY,
+     `bit8_col` BIT(8) DEFAULT NULL
+);
+
+CREATE TABLE `bit1_table` (
+     `id` INT PRIMARY KEY,
+     `bit1_col` BIT(1) DEFAULT NULL
+);
+
 CREATE TABLE `boolean_table` (
     `id` INT PRIMARY KEY,
     `boolean_col` TINYINT(1) DEFAULT NULL
@@ -276,6 +286,8 @@ ALTER TABLE `bigint_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `bigint_unsigned_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `binary_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `bit_table` MODIFY `id` INT AUTO_INCREMENT;
+ALTER TABLE `bit8_table` MODIFY `id` INT AUTO_INCREMENT;
+ALTER TABLE `bit1_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `blob_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `bool_table` MODIFY `id` INT AUTO_INCREMENT;
 ALTER TABLE `boolean_table` MODIFY `id` INT AUTO_INCREMENT;
@@ -318,6 +330,8 @@ INSERT INTO `bigint_unsigned_table` (`bigint_unsigned_col`) VALUES (184467440737
 INSERT INTO `binary_table` (`binary_col`) VALUES (x'7835383030000000000000000000000000000000');
 INSERT INTO `binary_table` (`binary_col`) VALUES (REPEAT(X'FF', 255));
 INSERT INTO `bit_table` (`bit_col`) VALUES (b'0111111111111111111111111111111111111111111111111111111111111111');
+INSERT INTO `bit8_table` (`bit8_col`) VALUES (b'0'), (b'11111111');
+INSERT INTO `bit1_table` (`bit1_col`) VALUES (b'0'), (b'1');
 INSERT INTO `blob_table` (`blob_col`) VALUES (X'7835383030');
 INSERT INTO `blob_table` (`blob_col`) VALUES (REPEAT(X'FF', 65535));
 INSERT INTO `bool_table` (`bool_col`) VALUES (0);
@@ -427,6 +441,8 @@ INSERT INTO `bigint_table` (`bigint_col`) VALUES (NULL);
 INSERT INTO `bigint_unsigned_table` (`bigint_unsigned_col`) VALUES (NULL);
 INSERT INTO `binary_table` (`binary_col`) VALUES (NULL);
 INSERT INTO `bit_table` (`bit_col`) VALUES (NULL);
+INSERT INTO `bit8_table` (`bit8_col`) VALUES (NULL);
+INSERT INTO `bit1_table` (`bit1_col`) VALUES (NULL);
 INSERT INTO `blob_table` (`blob_col`) VALUES (NULL);
 INSERT INTO `bool_table` (`bool_col`) VALUES (NULL);
 INSERT INTO `boolean_table` (`boolean_col`) VALUES (NULL);

v2/sourcedb-to-spanner/src/test/resources/DataTypesIT/mysql-spanner-schema.sql

@@ -18,6 +18,18 @@ CREATE TABLE IF NOT EXISTS bit_table (
   bit_col BYTES(MAX),
 ) PRIMARY KEY(id);
 
+CREATE TABLE IF NOT EXISTS bit8_table (
+  id INT64 NOT NULL,
+  -- testing mapping bit to int64
+  bit8_col INT64,
+) PRIMARY KEY(id);
+
+CREATE TABLE IF NOT EXISTS bit1_table (
+  id INT64 NOT NULL,
+  -- testing mapping bit to Bool
+  bit1_col BOOL,
+) PRIMARY KEY(id);
+
 CREATE TABLE IF NOT EXISTS blob_table (
   id INT64 NOT NULL,
   blob_col BYTES(MAX),