Implemented skip column feature (#348)

* Implemented skip column feature
* Var naming update

Author: pravinbhat

src/main/java/com/datastax/cdm/properties/KnownProperties.java

@@ -80,6 +80,7 @@ public enum PropertyType {
     public static final String ORIGIN_WRITETIME_NAMES = "spark.cdm.schema.origin.column.writetime.names";
     public static final String ALLOW_COLL_FOR_WRITETIME_TTL_CALC = "spark.cdm.schema.ttlwritetime.calc.useCollections";
 
+    public static final String ORIGIN_COLUMN_NAMES_TO_SKIP = "spark.cdm.schema.origin.column.skip";
     public static final String ORIGIN_COLUMN_NAMES_TO_TARGET = "spark.cdm.schema.origin.column.names.to.target";
 
     static {
@@ -93,6 +94,7 @@ public enum PropertyType {
         defaults.put(ORIGIN_WRITETIME_AUTO, "true");
         types.put(ALLOW_COLL_FOR_WRITETIME_TTL_CALC, PropertyType.BOOLEAN);
         defaults.put(ALLOW_COLL_FOR_WRITETIME_TTL_CALC, "false");
+        types.put(ORIGIN_COLUMN_NAMES_TO_SKIP, PropertyType.STRING_LIST);
         types.put(ORIGIN_COLUMN_NAMES_TO_TARGET, PropertyType.STRING_LIST);
     }
 

src/main/java/com/datastax/cdm/schema/CqlTable.java

@@ -26,6 +26,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -470,7 +471,9 @@ private void setCqlMetadata(CqlSession cqlSession) {
             }
         }
         String columnName = extractColumnName;
+        List<String> skipColumns = propertyHelper.getStringList(KnownProperties.ORIGIN_COLUMN_NAMES_TO_SKIP);
         this.cqlAllColumns = tableMetadata.getColumns().values().stream().filter(md -> !this.cqlAllColumns.contains(md))
+                .filter(md -> CollectionUtils.isEmpty(skipColumns) || !skipColumns.contains(md.getName().asCql(true)))
                 .filter(md -> !extractJsonExclusive || md.getName().asCql(true).endsWith(columnName))
                 .collect(Collectors.toCollection(() -> this.cqlAllColumns));
 

src/resources/cdm-detailed.properties

@@ -96,6 +96,10 @@ spark.cdm.connect.target.password      cassandra
 # Other Parameters:
 #  spark.cdm.schema
 #    .origin.column
+#       .skip             : Default is empty. A comma-separated list of columns that should be skipped.
+#                           Only use this property when both origin and target tables contain the above columns. 
+#                           If the target table does not contain the column, CDM will detect it and auto-skip.
+#
 #       .names.to.target  : Default is empty. If column names are changed between Origin and Target, then
 #                           this map-like list provides a mechanism to associate the two. The format is
 #                           origin_column_name:target_column_name. The list is comma-separated. Only renamed

src/test/java/com/datastax/cdm/schema/CqlTableTest.java

@@ -16,6 +16,7 @@
 package com.datastax.cdm.schema;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 import org.junit.jupiter.api.Test;
 
@@ -39,4 +40,12 @@ void testCL() {
         assertEquals(CqlTable.mapToConsistencyLevel("all"), ConsistencyLevel.ALL);
     }
 
+    @Test
+    void testformatName() {
+        assertNull(CqlTable.formatName(null));
+        assertEquals("", CqlTable.formatName(""));
+        assertEquals("\"KS123ks.T123able\"", CqlTable.formatName("KS123ks.T123able"));
+        assertEquals("\"Ks.Table\"", CqlTable.formatName("\"Ks.Table\""));
+        assertEquals("\"ks.table\"", CqlTable.formatName("ks.table"));
+    }
 }