-- review fixes

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/dml/CassandraDMLGenerator.java

@@ -125,7 +125,7 @@ public DMLGeneratorResponse getDMLStatement(DMLGeneratorRequest dmlGeneratorRequ
     }
     java.sql.Timestamp timestamp = dmlGeneratorRequest.getCommitTimestamp().toSqlTimestamp();
     String modType = dmlGeneratorRequest.getModType();
-    return generatorDMLResponse(
+    return generateDMLResponse(
         spannerTable, sourceTable, dmlGeneratorRequest, pkColumnNameValues, timestamp, modType);
   }
 
@@ -153,7 +153,7 @@ public DMLGeneratorResponse getDMLStatement(DMLGeneratorRequest dmlGeneratorRequ
    *     calls {@link #getUpsertStatementCQL}. - For "DELETE", calls {@link #getDeleteStatementCQL}.
    *     - For unsupported modType values, logs an error and returns an empty response.
    */
-  private static DMLGeneratorResponse generatorDMLResponse(
+  private static DMLGeneratorResponse generateDMLResponse(
       SpannerTable spannerTable,
       SourceTable sourceTable,
       DMLGeneratorRequest dmlGeneratorRequest,
@@ -326,10 +326,6 @@ private static Map<String, PreparedStatementValueObject<?>> getColumnValues(
             getMappedColumnValue(
                 spannerColDef, sourceColDef, keyValuesJson, sourceDbTimezoneOffset);
       } else if (newValuesJson.has(spannerColumnName)) {
-        // get the value based on Spanner and Source type
-        if (newValuesJson.isNull(spannerColumnName)) {
-          continue;
-        }
         columnValue =
             getMappedColumnValue(
                 spannerColDef, sourceColDef, newValuesJson, sourceDbTimezoneOffset);
@@ -390,10 +386,6 @@ private static Map<String, PreparedStatementValueObject<?>> getPkColumnValues(
             getMappedColumnValue(
                 spannerColDef, sourceColDef, keyValuesJson, sourceDbTimezoneOffset);
       } else if (newValuesJson.has(spannerColumnName)) {
-        // get the value based on Spanner and Source type
-        if (newValuesJson.isNull(spannerColumnName)) {
-          continue;
-        }
         columnValue =
             getMappedColumnValue(
                 spannerColDef, sourceColDef, newValuesJson, sourceDbTimezoneOffset);

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/dml/CassandraTypeHandler.java

@@ -333,7 +333,7 @@ private static Object handleSpannerColumnType(
     try {
       return spannerType.contains("string")
           ? valuesJson.optString(columnName)
-          : valuesJson.opt(columnName);
+          : valuesJson.isNull(columnName) ? null : valuesJson.opt(columnName);
     } catch (Exception e) {
       throw new IllegalArgumentException(
           "Exception Caught During parsing for Spanner column type: " + spannerType);

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/dao/source/CassandraDaoTest.java

@@ -27,6 +27,7 @@
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.cql.Statement;
 import com.google.cloud.teleport.v2.templates.dbutils.connection.IConnectionHelper;
+import com.google.cloud.teleport.v2.templates.dbutils.dml.CassandraTypeHandler;
 import com.google.cloud.teleport.v2.templates.exceptions.ConnectionException;
 import com.google.cloud.teleport.v2.templates.models.PreparedStatementGeneratedResponse;
 import com.google.cloud.teleport.v2.templates.models.PreparedStatementValueObject;
@@ -90,6 +91,27 @@ public void testPreparedStatementExecution() throws Exception {
     verify(mockSession).execute(ArgumentMatchers.eq(mockBoundStatement));
   }
 
+  @Test
+  public void testPreparedStatementExecutionForNullAsValue() throws Exception {
+    String preparedDmlStatement = "INSERT INTO test (id, name) VALUES (?, ?)";
+    List<PreparedStatementValueObject<?>> values =
+        Arrays.asList(
+            PreparedStatementValueObject.create("date", CassandraTypeHandler.NullClass.INSTANCE),
+            PreparedStatementValueObject.create("varchar", "text"));
+
+    when(mockPreparedStatementGeneratedResponse.getDmlStatement()).thenReturn(preparedDmlStatement);
+    when(mockPreparedStatementGeneratedResponse.getValues()).thenReturn(values);
+    when(mockConnectionHelper.getConnection(anyString())).thenReturn(mockSession);
+    when(mockSession.prepare(preparedDmlStatement)).thenReturn(mockPreparedStatement);
+    when(mockPreparedStatement.bind(ArgumentMatchers.any())).thenReturn(mockBoundStatement);
+
+    cassandraDao.write(mockPreparedStatementGeneratedResponse);
+
+    verify(mockSession).prepare(ArgumentMatchers.eq(preparedDmlStatement));
+    verify(mockPreparedStatement).bind(ArgumentMatchers.any());
+    verify(mockSession).execute(ArgumentMatchers.eq(mockBoundStatement));
+  }
+
   @Test
   public void testWriteWithExceptionInPreparedStatement() throws Exception {
     String preparedDmlStatement = "INSERT INTO test (id, name) VALUES (?, ?)";

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/dml/CassandraDMLGeneratorTest.java

@@ -104,6 +104,29 @@ public void tableAndAllColumnNameTypesMatch() {
     assertEquals(3, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
   }
 
+  @Test
+  public void tableAndAllColumnNameTypesForNullValueMatch() {
+    Schema schema = SessionFileReader.read("src/test/resources/cassandraSession.json");
+    String tableName = "sample_table";
+    String newValueStr = "{\"date_column\":null}";
+    JSONObject newValuesJson = new JSONObject(newValueStr);
+    String keyValueString = "{\"id\":\"999\"}";
+    JSONObject keyValuesJson = new JSONObject(keyValueString);
+    String modType = "INSERT";
+    CassandraDMLGenerator cassandraDMLGenerator = new CassandraDMLGenerator();
+    DMLGeneratorResponse dmlGeneratorResponse =
+        cassandraDMLGenerator.getDMLStatement(
+            new DMLGeneratorRequest.Builder(
+                    modType, tableName, newValuesJson, keyValuesJson, "+00:00")
+                .setSchema(schema)
+                .setCommitTimestamp(Timestamp.now())
+                .build());
+    String sql = dmlGeneratorResponse.getDmlStatement();
+
+    assertTrue(sql.contains("id"));
+    assertEquals(3, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
+  }
+
   @Test
   public void tableNameMatchColumnNameTypeMismatch() {
     Schema schema = SessionFileReader.read("src/test/resources/cassandraSession.json");
@@ -270,7 +293,7 @@ public void updateToNull() {
     String sql = dmlGeneratorResponse.getDmlStatement();
 
     assertTrue(sql.contains("SingerId"));
-    assertEquals(2, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
+    assertEquals(3, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
   }
 
   @Test
@@ -292,7 +315,7 @@ public void deleteMultiplePKColumns() {
                 .setCommitTimestamp(Timestamp.now())
                 .build());
     String sql = dmlGeneratorResponse.getDmlStatement();
-    assertEquals(2, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
+    assertEquals(3, ((PreparedStatementGeneratedResponse) dmlGeneratorResponse).getValues().size());
   }
 
   @Test