Cassandra DML Generator - Spanner to SourceDB

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/schema/Schema.java

@@ -77,6 +77,8 @@ public Schema(
     this.srcSchema = srcSchema;
     this.toSpanner = toSpanner;
     this.toSource = toSource;
+    this.srcToID = toSource;
+    this.spannerToID = toSpanner;
     this.empty = (spSchema == null || srcSchema == null);
   }
 

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java

@@ -75,7 +75,6 @@ public class Constants {
   public static final String DEFAULT_SHARD_ID = "single_shard";
 
   public static final String SOURCE_MYSQL = "mysql";
-
   public static final String SOURCE_CASSANDRA = "cassandra";
 
   // Message written to the file for filtered records

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

@@ -19,10 +19,10 @@
 import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 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.DMLGeneratorResponse;
 import com.google.cloud.teleport.v2.templates.models.PreparedStatementGeneratedResponse;
-import com.google.cloud.teleport.v2.templates.models.PreparedStatementValueObject;
 
 public class CassandraDao implements IDao<DMLGeneratorResponse> {
   private final String cassandraUrl;
@@ -38,22 +38,25 @@ public CassandraDao(
 
   @Override
   public void write(DMLGeneratorResponse dmlGeneratorResponse) throws Exception {
-    try (CqlSession session =
-        (CqlSession)
-            connectionHelper.getConnection(this.cassandraUrl)) { // Ensure connection is obtained
-      if (session == null) {
-        throw new ConnectionException("Connection is null");
-      }
-      PreparedStatementGeneratedResponse preparedStatementGeneratedResponse =
-          (PreparedStatementGeneratedResponse) dmlGeneratorResponse;
-      String dmlStatement = preparedStatementGeneratedResponse.getDmlStatement();
-      PreparedStatement preparedStatement = session.prepare(dmlStatement);
-      BoundStatement boundStatement =
-          preparedStatement.bind(
-              preparedStatementGeneratedResponse.getValues().stream()
-                  .map(PreparedStatementValueObject::value)
-                  .toArray());
-      session.execute(boundStatement);
+    CqlSession session = (CqlSession) connectionHelper.getConnection(this.cassandraUrl);
+    if (session == null) {
+      throw new ConnectionException("Connection is null");
     }
+    PreparedStatementGeneratedResponse preparedStatementGeneratedResponse =
+        (PreparedStatementGeneratedResponse) dmlGeneratorResponse;
+    String dmlStatement = preparedStatementGeneratedResponse.getDmlStatement();
+    PreparedStatement preparedStatement = session.prepare(dmlStatement);
+    BoundStatement boundStatement =
+        preparedStatement.bind(
+            preparedStatementGeneratedResponse.getValues().stream()
+                .map(
+                    v -> {
+                      if (v.value() == CassandraTypeHandler.NullClass.INSTANCE) {
+                        return null;
+                      }
+                      return CassandraTypeHandler.castToExpectedType(v.dataType(), v.value());
+                    })
+                .toArray());
+    session.execute(boundStatement);
   }
 }