Live and Bulk Retry: Data conversion errors when NULL is passed in non-textual datatype

Alternative implementation:
instead of manually checking in each toDatatype function, we could add this check to containsValue function that is used by all datatypes

Reason for rejection: a user can pass "NULL" string purposely to a string column - this should not be converted to null by my logic. containsValue function is called in method toString and thus this conversion cannot be added to the common function and needs to be added for each datatype seperately wherever relevant.
This would not change the existing behaviour for string data type. It would only change NULL handling for other datatypes

Author: aasthabharill

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertor.java

@@ -62,6 +62,9 @@ public static Boolean toBoolean(JsonNode changeEvent, String key, boolean requir
        */
       JsonNode node = changeEvent.get(key);
       if (node.isTextual()) {
+        if (node.asText().equalsIgnoreCase("NULL")) {
+          return null;
+        }
         return BooleanUtils.toBoolean(node.asText());
       }
       return Boolean.valueOf(node.asBoolean());
@@ -80,6 +83,9 @@ public static Long toLong(JsonNode changeEvent, String key, boolean requiredFiel
     try {
       JsonNode node = changeEvent.get(key);
       if (node.isTextual()) {
+        if (node.asText().equalsIgnoreCase("NULL")) {
+          return null;
+        }
         return Long.valueOf(node.asText());
       }
       return node.asLong();
@@ -98,6 +104,9 @@ public static Double toDouble(JsonNode changeEvent, String key, boolean required
     try {
       JsonNode node = changeEvent.get(key);
       if (node.isTextual()) {
+        if (node.asText().equalsIgnoreCase("NULL")) {
+          return null;
+        }
         return Double.valueOf(node.asText());
       }
       return Double.valueOf(node.asDouble());
@@ -115,6 +124,9 @@ public static Float toFloat(JsonNode changeEvent, String key, boolean requiredFi
     try {
       JsonNode node = changeEvent.get(key);
       if (node.isTextual()) {
+        if (node.asText().equalsIgnoreCase("NULL")) {
+          return null;
+        }
         return Float.valueOf(node.asText());
       }
       return new Float(node.asDouble());
@@ -155,6 +167,9 @@ public static ByteArray toByteArray(JsonNode changeEvent, String key, boolean re
       // For data with Spanner type as BYTES, Datastream returns a hex encoded string. We need to
       // decode it before returning to ensure data correctness.
       String s = node.asText();
+      if (s.equalsIgnoreCase("NULL")) {
+        return null;
+      }
       // Make an odd length hex string even by appending a 0 in the beginning.
       if (s.length() % 2 == 1) {
         s = "0" + s;
@@ -179,6 +194,9 @@ public static Timestamp toTimestamp(JsonNode changeEvent, String key, boolean re
     }
     try {
       String timeString = changeEvent.get(key).asText();
+      if (timeString.equalsIgnoreCase("NULL")) {
+        return null;
+      }
       Instant instant = parseTimestamp(timeString);
       return Timestamp.ofTimeSecondsAndNanos(instant.getEpochSecond(), instant.getNano());
     } catch (Exception e) {
@@ -195,6 +213,9 @@ public static Date toDate(JsonNode changeEvent, String key, boolean requiredFiel
     }
     try {
       String dateString = changeEvent.get(key).asText();
+      if (dateString.equalsIgnoreCase("NULL")) {
+        return null;
+      }
       return Date.fromJavaUtilDate(parseLenientDate(dateString));
     } catch (Exception e) {
       throw new ChangeEventConvertorException("Unable to convert field " + key + " to Date", e);
@@ -227,6 +248,9 @@ public static BigDecimal toNumericBigDecimal(
     if (value == null) {
       return null;
     }
+    if (value.equalsIgnoreCase("NULL")) {
+      return null;
+    }
     if (NumberUtils.isCreatable(value) || NumberUtils.isParsable(value) || isNumeric(value)) {
       return new BigDecimal(value).setScale(9, RoundingMode.HALF_UP);
     }

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/convertors/ChangeEventTypeConvertorTest.java

@@ -746,4 +746,30 @@ public void cannotConvertNonExistentRequiredFieldToDate() throws Exception {
         ChangeEventTypeConvertor.toDate(ce, "field1", /* requiredField= */ true),
         Date.parseDate("2020-12-30"));
   }
+
+  @Test
+  public void canConvertNullStrings() throws Exception {
+    JSONObject changeEvent = new JSONObject();
+    changeEvent.put("long_field", "NULL");
+    changeEvent.put("double_field", "null");
+    changeEvent.put("float_field", "Null");
+    changeEvent.put("numeric_field", "NULL");
+    changeEvent.put("boolean_field", "NULL");
+    changeEvent.put("byte_field", "NULL");
+    changeEvent.put("timestamp_field", "NULL");
+    changeEvent.put("date_field", "NULL");
+    changeEvent.put("string_field", "NULL"); // this should not be converted to null
+
+    JsonNode ce = getJsonNode(changeEvent.toString());
+
+    assertNull(ChangeEventTypeConvertor.toLong(ce, "long_field", true));
+    assertNull(ChangeEventTypeConvertor.toDouble(ce, "double_field", true));
+    assertNull(ChangeEventTypeConvertor.toFloat(ce, "float_field", true));
+    assertNull(ChangeEventTypeConvertor.toNumericBigDecimal(ce, "numeric_field", true));
+    assertNull(ChangeEventTypeConvertor.toBoolean(ce, "boolean_field", true));
+    assertNull(ChangeEventTypeConvertor.toByteArray(ce, "byte_field", true));
+    assertNull(ChangeEventTypeConvertor.toTimestamp(ce, "timestamp_field", true));
+    assertNull(ChangeEventTypeConvertor.toDate(ce, "date_field", true));
+    assertEquals("NULL", ChangeEventTypeConvertor.toString(ce, "string_field", true));
+  }
 }