Adding Cassandra Type Options to IT test

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/cassandra/iowrapper/CassandraIOWrapperHelper.java

@@ -31,6 +31,7 @@
 import com.google.cloud.teleport.v2.spanner.migrations.schema.SourceColumnType;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import java.io.FileNotFoundException;
 import java.util.List;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -72,8 +73,18 @@ static ImmutableList<String> getTablesToRead(
       tablesToRead = schemaDiscovery.discoverTables(dataSource, sourceSchemaReference);
       LOG.info("Auto Discovered SourceTables = {}, Tables = {}", sourceTables, tablesToRead);
     } else {
-      tablesToRead = ImmutableList.copyOf(sourceTables);
-      LOG.info("Using passed SourceTables = {}", sourceTables);
+      ImmutableSet<String> existingTables =
+          schemaDiscovery.discoverTables(dataSource, sourceSchemaReference).stream()
+              .collect(ImmutableSet.toImmutableSet());
+      tablesToRead =
+          sourceTables.stream()
+              .filter(t -> existingTables.contains(t))
+              .collect(ImmutableList.toImmutableList());
+      LOG.info(
+          "Using tables from passed SourceTables = {}, existing tables = {}, tables to read = {}. Tables not present on source shall be ignored",
+          sourceTables,
+          existingTables,
+          tablesToRead);
     }
     return tablesToRead;
   }

v2/sourcedb-to-spanner/src/main/java/com/google/cloud/teleport/v2/source/reader/io/cassandra/mappings/CassandraMappingsProvider.java

@@ -175,12 +175,7 @@ public class CassandraMappingsProvider {
               Row::getString,
               valuePassThrough,
               String.class)
-          .put(
-              "TIME",
-              UnifiedMappingProvider.Type.INTERVAL_NANO,
-              Row::getTime,
-              cassandraTimeToIntervalNano,
-              Long.class)
+          .put("TIME", UnifiedMappingProvider.Type.LONG, Row::getTime, valuePassThrough, Long.class)
           .put(
               "TIMESTAMP",
               UnifiedMappingProvider.Type.TIMESTAMP,

v2/sourcedb-to-spanner/src/test/java/com/google/cloud/teleport/v2/source/reader/io/cassandra/iowrapper/CassandraIOWrapperHelperTest.java

@@ -147,6 +147,14 @@ public void testTablesToRead() {
                 CassandraIOWrapperHelper.buildSchemaDiscovery(),
                 cassandraSchemaReference))
         .isEqualTo(List.of(BASIC_TEST_TABLE));
+
+    assertThat(
+            CassandraIOWrapperHelper.getTablesToRead(
+                List.of(BASIC_TEST_TABLE, "Non-existing-table"),
+                dataSource,
+                CassandraIOWrapperHelper.buildSchemaDiscovery(),
+                cassandraSchemaReference))
+        .isEqualTo(List.of(BASIC_TEST_TABLE));
   }
 
   @Test

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

@@ -172,7 +172,7 @@ private Map<String, String> getAllDataTypeNullRow() {
         .put("varchar_set_col", "[]")
         .put("timestamp_col", "NULL")
         .put("tinyint_set_col", "[]")
-        .put("time_col", "P0D")
+        .put("time_col", "0")
         .put("bigint_boolean_map_col", "{}")
         .put("text_list_col", "[]")
         .put("boolean_list_col", "[]")
@@ -259,7 +259,7 @@ private Map<String, String> getAllDataTypeMinRow() {
         .put("varchar_set_col", "[]")
         .put("timestamp_col", "1970-01-01T00:00:00Z")
         .put("tinyint_set_col", "[]")
-        .put("time_col", "P0D")
+        .put("time_col", "0")
         .put("bigint_boolean_map_col", "{}")
         .put("text_list_col", "[]")
         .put("boolean_list_col", "[]")
@@ -345,7 +345,7 @@ private Map<String, String> getAllDataTypeMaxRow() {
             "{\"-128\":\"184467440000000000000\",\"127\":\"-184467440000000000000\"}")
         .put("text_set_col", "[G,KNOWS,LEONARDO,LER,MATHEMATICIAN,O,OG]")
         .put("double_set_col", "[-Infinity,-1.7976931348623157E308,0.0,3.1415926,Infinity,NaN]")
-        .put("time_list_col", "[P0D,P0DT23H59M59.999999999S]")
+        .put("time_list_col", "[0,86399999999999]")
         .put("frozen_ascii_list_col", "[a,b]")
         .put("int_list_col", "[-2147483648,0,2147483647]")
         .put("ascii_list_col", "[a,~]")
@@ -355,27 +355,23 @@ private Map<String, String> getAllDataTypeMaxRow() {
             "timestamp_set_col",
             "[1970-01-01T00:00:00Z,1970-01-01T00:00:00Z,1969-12-31T23:59:59.999000000Z]")
         .put("ascii_col", "~~~~~~~")
-        .put(
-            "time_tinyint_map_col",
-            "{\"{\\\"years\\\": 0, \\\"months\\\": 0, \\\"days\\\": 0, \\\"hours\\\": 0, \\\"minutes\\\": 0, \\\"seconds\\\": 0, \\\"nanos\\\": 0}\":\"127\",\"{\\\"years\\\": 0, \\\"months\\\": 0, \\\"days\\\": 0, \\\"hours\\\": 0, \\\"minutes\\\": 0, \\\"seconds\\\": 0, \\\"nanos\\\": 86399999999999}\":\"-128\"}")
+        .put("time_tinyint_map_col", "{\"0\":\"127\",\"86399999999999\":\"-128\"}")
         .put("float_col", "3.4028235E38")
         .put("bigint_set_col", "[-9223372036854775808,0,9223372036854775807]")
         .put("varchar_set_col", "[a,~]")
         .put("timestamp_col", "1969-12-31T23:59:59.999000000Z")
         .put("tinyint_set_col", "[-128,0,127]")
-        .put("time_col", "P0DT23H59M59.999999999S")
+        .put("time_col", "86399999999999")
         .put("bigint_boolean_map_col", "{\"42\":\"true\",\"84\":\"false\"}")
         .put("text_list_col", "[G,O,OG,LER,KNOWS,LEONARDO,MATHEMATICIAN]")
         .put("boolean_list_col", "[true,false]")
         .put("blob_list_col", "[AAAAAAAAAAA=,////////]")
         .put(
             "timeuuid_set_col",
             "[00000000-0000-1000-9000-000000000000,88888888-8888-1888-9888-888888888888,ffffffff-ffff-1fff-9fff-ffffffffffff]")
-        .put(
-            "int_time_map_col",
-            "{\"1\":\"{\\\"years\\\": 0, \\\"months\\\": 0, \\\"days\\\": 0, \\\"hours\\\": 0, \\\"minutes\\\": 0, \\\"seconds\\\": 0, \\\"nanos\\\": 0}\",\"42\":\"{\\\"years\\\": 0, \\\"months\\\": 0, \\\"days\\\": 0, \\\"hours\\\": 0, \\\"minutes\\\": 0, \\\"seconds\\\": 0, \\\"nanos\\\": 2400}\"}")
+        .put("int_time_map_col", "{\"1\":\"0\",\"42\":\"2400\"}")
         .put("timeuuid_col", "ffffffff-ffff-1fff-9fff-ffffffffffff")
-        .put("time_set_col", "[P0D,P0DT23H59M59.999999999S]")
+        .put("time_set_col", "[0,86399999999999]")
         .put("boolean_set_col", "[false,true]")
         .put("bigint_col", "9223372036854775807")
         .put("float_set_col", "[-Infinity,-3.4028235E38,0.0,3.4028235E38,Infinity,NaN]")

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

@@ -46,6 +46,7 @@
 import org.apache.beam.it.jdbc.PostgresResourceManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
 
 /**
  * Base class for SourceDbToSpanner integration tests. It provides helper functions related to
@@ -174,7 +175,11 @@ public void createSpannerDDL(SpannerResourceManager spannerResourceManager, Stri
     LOG.info("Creating spanner DDL");
     String ddl =
         String.join(
-            " ", Resources.readLines(Resources.getResource(resourceName), StandardCharsets.UTF_8));
+            " ",
+            Resources.readLines(Resources.getResource(resourceName), StandardCharsets.UTF_8)
+                .stream()
+                .map(line -> line.replaceAll("\\s*--.*$", ""))
+                .collect(ImmutableList.toImmutableList()));
     ddl = ddl.trim();
     List<String> ddls =
         Arrays.stream(ddl.split(";")).filter(d -> !d.isBlank()).collect(Collectors.toList());