CDM-53 writetime and ttl columns are now automatically detected and configured, or can be manually configured.

mieslep · msmygit · commit 96ec01bc4004 · 2023-05-09T13:53:25.000-04:00
diff --git a/SIT/cdm.sh b/SIT/cdm.sh
@@ -97,8 +97,8 @@ fi
 
 spark-submit --properties-file "${PROPERTIES}" \
   --master "local[*]" \
-  --conf "spark.driver.extraJavaOptions=-Dlog4j.configurationFile=file:///local/log4j.xml -Dcom.datastax.cdm.log.level=TRACE" \
-  --conf "spark.executor.extraJavaOptions=-Dlog4j.configurationFile=file:///local/log4j.xml -Dcom.datastax.cdm.log.level=TRACE" \
+  --conf "spark.driver.extraJavaOptions=-Dlog4j.configurationFile=file:///local/log4j.xml -Dcom.datastax.cdm.log.level=DEBUG" \
+  --conf "spark.executor.extraJavaOptions=-Dlog4j.configurationFile=file:///local/log4j.xml -Dcom.datastax.cdm.log.level=DEBUG" \
   --class ${CLASS} \
   /local/cassandra-data-migrator.jar
 
diff --git a/SIT/smoke/03_ttl_writetime/breakData.cql b/SIT/smoke/03_ttl_writetime/breakData.cql
@@ -21,7 +21,33 @@ INSERT INTO origin.smoke_ttl_writetime(key, w_col3)  VALUES ('record6','C---') U
 -- tw_col2 is TTL and Writetime, so both changes should be propatated by the diff
 INSERT INTO origin.smoke_ttl_writetime(key, tw_col2) VALUES ('record7','B---') USING TTL 700000 AND TIMESTAMP 1087384200000000; -- 11:10:00
 
-
-
-SELECT * FROM target.smoke_ttl_writetime;
+SELECT key
+      ,t_col1
+      ,WRITETIME(t_col1) AS t_col1_wt
+      ,TTL(t_col1) AS t_col1_ttl
+      ,tw_col2
+      ,WRITETIME(tw_col2) AS tw_col2_wt
+      ,TTL(tw_col2) AS tw_col2_ttl
+      ,w_col3
+      ,WRITETIME(w_col3) AS w_col3_wt
+      ,TTL(w_col3) AS w_col3_ttl
+      ,col4
+      ,WRITETIME(col4) AS col4_wt
+      ,TTL(col4) AS col4_ttl
+  FROM origin.smoke_ttl_writetime;
+
+SELECT key
+      ,t_col1
+      ,WRITETIME(t_col1) AS t_col1_wt
+      ,TTL(t_col1) AS t_col1_ttl
+      ,tw_col2
+      ,WRITETIME(tw_col2) AS tw_col2_wt
+      ,TTL(tw_col2) AS tw_col2_ttl
+      ,w_col3
+      ,WRITETIME(w_col3) AS w_col3_wt
+      ,TTL(w_col3) AS w_col3_ttl
+      ,col4
+      ,WRITETIME(col4) AS col4_wt
+      ,TTL(col4) AS col4_ttl
+  FROM target.smoke_ttl_writetime;
 
diff --git a/SIT/smoke/03_ttl_writetime/setup.cql b/SIT/smoke/03_ttl_writetime/setup.cql
@@ -51,19 +51,19 @@ INSERT INTO origin.smoke_ttl_writetime(key, w_col3) VALUES ('record8','CCCC') US
 INSERT INTO origin.smoke_ttl_writetime(key,   col4) VALUES ('record8','DDDD') USING TTL 700000 AND TIMESTAMP 1087383780000000; -- 11:00:03
 
 
-SELECT key,
-       t_col1,
-       writetime(t_col1)  as  wt_t_col1, 
-             ttl(t_col1)  as ttl_t_col1,
-       tw_col2,
-       writetime(tw_col2) as  wt_tw_col2,
-             ttl(tw_col2) as ttl_tw_col2,
-       w_col3,
-       writetime(w_col3)  as  wt_w_col3,
-             ttl(w_col3)  as ttl_w_col3,
-       col4, 
-       writetime(col4)    as  wt_col4,
-             ttl(col4)    as ttl_col4
+SELECT key
+      ,t_col1
+      ,WRITETIME(t_col1) AS t_col1_wt
+      ,TTL(t_col1) AS t_col1_ttl
+      ,tw_col2
+      ,WRITETIME(tw_col2) AS tw_col2_wt
+      ,TTL(tw_col2) AS tw_col2_ttl
+      ,w_col3
+      ,WRITETIME(w_col3) AS w_col3_wt
+      ,TTL(w_col3) AS w_col3_ttl
+      ,col4
+      ,WRITETIME(col4) AS col4_wt
+      ,TTL(col4) AS col4_ttl
   FROM origin.smoke_ttl_writetime;
 
 DROP TABLE IF EXISTS target.smoke_ttl_writetime;
diff --git a/src/main/java/com/datastax/cdm/data/CqlData.java b/src/main/java/com/datastax/cdm/data/CqlData.java
@@ -63,6 +63,16 @@ public static boolean isCollection(DataType dataType) {
         return false;
     }
 
+    public static boolean isFrozen(DataType dataType) {
+        if (isPrimitive(dataType)) return false;
+        if (dataType instanceof UserDefinedType) return ((UserDefinedType) dataType).isFrozen();
+        if (dataType instanceof ListType) return ((ListType) dataType).isFrozen();
+        if (dataType instanceof SetType) return ((SetType) dataType).isFrozen();
+        if (dataType instanceof MapType) return ((MapType) dataType).isFrozen();
+        if (dataType instanceof TupleType) return dataType.asCql(true, false).contains("frozen<");
+        return false;
+    }
+
     public static Class getBindClass(DataType dataType) {
         Class primitiveClass = primitiveDataTypeToJavaClassMap.get(dataType);
         if (primitiveClass != null) return primitiveClass;
diff --git a/src/main/java/com/datastax/cdm/feature/WritetimeTTL.java b/src/main/java/com/datastax/cdm/feature/WritetimeTTL.java
@@ -18,7 +18,9 @@ public class WritetimeTTL extends AbstractFeature  {
     private final boolean logDebug = logger.isDebugEnabled();;
 
     private List<String> ttlNames;
+    private boolean autoTTLNames;
     private List<String> writetimeNames;
+    private boolean autoWritetimeNames;
     private Long customWritetime = 0L;
     private List<Integer> ttlSelectColumnIndexes = null;
     private List<Integer> writetimeSelectColumnIndexes = null;
@@ -33,14 +35,18 @@ public boolean loadProperties(IPropertyHelper propertyHelper) {
             logger.info("PARAM -- TTLCols: {}", ttlNames);
         }
 
+        this.autoTTLNames = propertyHelper.getBoolean(KnownProperties.ORIGIN_TTL_AUTO);
         this.writetimeNames = getWritetimeNames(propertyHelper);
         if (null!=this.writetimeNames && !this.writetimeNames.isEmpty()) {
             logger.info("PARAM -- WriteTimestampCols: {}", writetimeNames);
+            this.autoTTLNames = false;
         }
 
+        this.autoWritetimeNames = propertyHelper.getBoolean(KnownProperties.ORIGIN_WRITETIME_AUTO);
         this.customWritetime = getCustomWritetime(propertyHelper);
         if (this.customWritetime > 0) {
             logger.info("PARAM -- {}: {} datetime is {} ", KnownProperties.TRANSFORM_CUSTOM_WRITETIME, customWritetime, Instant.ofEpochMilli(customWritetime / 1000));
+            this.autoWritetimeNames = false;
         }
 
         this.filterMin = getMinFilter(propertyHelper);
@@ -80,15 +86,40 @@ public boolean initializeAndValidate(CqlTable originTable, CqlTable targetTable)
             throw new IllegalArgumentException("Origin table is not an origin table");
         }
 
+        if (originTable.isCounterTable()) {
+            if (isEnabled) {
+                logger.error("Counter table cannot specify TTL or WriteTimestamp columns as they cannot set on write");
+                isValid = false;
+                isEnabled = false;
+                return false;
+            }
+
+            logger.info("Counter table does not support TTL or WriteTimestamp columns as they cannot set on write, so feature is disabled");
+            return true;
+        }
+
         isValid = true;
         if (!validateProperties()) {
             isEnabled = false;
             return false;
         }
 
+        if (autoTTLNames) {
+            this.ttlNames = originTable.getWritetimeTTLColumns();
+        }
+
+        if (autoWritetimeNames) {
+            this.writetimeNames = originTable.getWritetimeTTLColumns();
+        }
+
         validateTTLColumns(originTable);
         validateWritetimeColumns(originTable);
 
+        if (hasWriteTimestampFilter && (null==writetimeNames || writetimeNames.isEmpty())) {
+            logger.error("WriteTimestamp filter is configured but no WriteTimestamp columns are defined");
+            isValid = false;
+        }
+
         if (!isValid) isEnabled = false;
         logger.info("Feature {} is {}", this.getClass().getSimpleName(), isEnabled?"enabled":"disabled");
         return isValid;
@@ -117,8 +148,8 @@ public static Long getMaxFilter(IPropertyHelper propertyHelper) {
 
     public Long getCustomWritetime() { return customWritetime; }
     public boolean hasWriteTimestampFilter() { return isEnabled && hasWriteTimestampFilter; }
-    public Long getMinWriteTimeStampFilter() { return this.hasWriteTimestampFilter ? this.filterMin : Long.MIN_VALUE; }
-    public Long getMaxWriteTimeStampFilter() { return this.hasWriteTimestampFilter ? this.filterMax : Long.MAX_VALUE; }
+    public Long getMinWriteTimeStampFilter() { return (this.hasWriteTimestampFilter && null!=this.filterMin) ? this.filterMin : Long.MIN_VALUE; }
+    public Long getMaxWriteTimeStampFilter() { return (this.hasWriteTimestampFilter && null!=this.filterMax) ? this.filterMax : Long.MAX_VALUE; }
 
     public boolean hasTTLColumns() { return null!=this.ttlSelectColumnIndexes && !this.ttlSelectColumnIndexes.isEmpty(); }
     public boolean hasWritetimeColumns() { return customWritetime>0 || null!=this.writetimeSelectColumnIndexes && !this.writetimeSelectColumnIndexes.isEmpty(); }
@@ -157,6 +188,12 @@ private void validateTTLColumns(CqlTable originTable) {
                 logger.error("TTL column {} is not present on origin table {}", ttlName, originTable.getKeyspaceName());
                 isValid = false;
                 return;
+            } else {
+                if (!originTable.isWritetimeTTLColumn(ttlName)) {
+                    logger.error("TTL column {} is not a column which can provide a TTL on origin table {}", ttlName, originTable.getKeyspaceName());
+                    isValid = false;
+                    return;
+                }
             }
 
             newColumnNames.add("TTL(" + ttlName + ")");
@@ -183,6 +220,12 @@ private void validateWritetimeColumns(CqlTable originTable) {
                 logger.error("Writetime column {} is not configured for origin table {}", writetimeName, originTable.getKeyspaceName());
                 isValid = false;
                 return;
+            } else {
+                if (!originTable.isWritetimeTTLColumn(writetimeName)) {
+                    logger.error("Writetime column {} is not a column which can provide a WRITETIME on origin table {}", writetimeName, originTable.getKeyspaceName());
+                    isValid = false;
+                    return;
+                }
             }
 
             newColumnNames.add("WRITETIME(" + writetimeName + ")");
diff --git a/src/main/java/com/datastax/cdm/properties/KnownProperties.java b/src/main/java/com/datastax/cdm/properties/KnownProperties.java
@@ -60,17 +60,22 @@ public enum PropertyType {
     // Properties that describe the origin schema
     //==========================================================================
     public static final String ORIGIN_KEYSPACE_TABLE         = "spark.cdm.schema.origin.keyspaceTable";
+    public static final String ORIGIN_TTL_AUTO               = "spark.cdm.schema.origin.column.ttl.automatic";
     public static final String ORIGIN_TTL_NAMES              = "spark.cdm.schema.origin.column.ttl.names";
+    public static final String ORIGIN_WRITETIME_AUTO         = "spark.cdm.schema.origin.column.writetime.automatic";
     public static final String ORIGIN_WRITETIME_NAMES        = "spark.cdm.schema.origin.column.writetime.names";
 
     public static final String ORIGIN_COLUMN_NAMES_TO_TARGET = "spark.cdm.schema.origin.column.names.to.target";
 
-
     static {
            types.put(ORIGIN_KEYSPACE_TABLE, PropertyType.STRING);
         required.add(ORIGIN_KEYSPACE_TABLE);
            types.put(ORIGIN_TTL_NAMES, PropertyType.STRING_LIST);
+           types.put(ORIGIN_TTL_AUTO, PropertyType.BOOLEAN);
+        defaults.put(ORIGIN_TTL_AUTO, "true");
            types.put(ORIGIN_WRITETIME_NAMES, PropertyType.STRING_LIST);
+           types.put(ORIGIN_WRITETIME_AUTO, PropertyType.BOOLEAN);
+        defaults.put(ORIGIN_WRITETIME_AUTO, "true");
 
            types.put(ORIGIN_COLUMN_NAMES_TO_TARGET, PropertyType.STRING_LIST);
     }
diff --git a/src/main/java/com/datastax/cdm/schema/CqlTable.java b/src/main/java/com/datastax/cdm/schema/CqlTable.java
@@ -45,6 +45,7 @@ public class CqlTable extends BaseTable {
     private List<ColumnMetadata> cqlAllColumns;
     private Map<String,DataType> columnNameToCqlTypeMap;
     private final List<Class> bindClasses;
+    private List<String> writetimeTTLColumns;
 
     private CqlTable otherCqlTable;
     private List<Integer> correspondingIndexes;
@@ -306,13 +307,41 @@ private void setCqlMetadata(CqlSession cqlSession) {
                 .filter(md -> !this.cqlAllColumns.contains(md))
                 .collect(Collectors.toCollection(() -> this.cqlAllColumns));
 
+        this.writetimeTTLColumns = tableMetadata.getColumns().values().stream()
+                .filter(columnMetadata -> canColumnHaveTTLorWritetime(tableMetadata, columnMetadata))
+                .map(ColumnMetadata::getName)
+                .map(CqlIdentifier::asInternal)
+                .collect(Collectors.toList());
+
         this.columnNameToCqlTypeMap = this.cqlAllColumns.stream()
                 .collect(Collectors.toMap(
                         columnMetadata -> columnMetadata.getName().asInternal(),
                         ColumnMetadata::getType
                 ));
     }
 
+    private boolean canColumnHaveTTLorWritetime(TableMetadata tableMetadata, ColumnMetadata columnMetadata) {
+        DataType dataType = columnMetadata.getType();
+        boolean isKeyColumn = tableMetadata.getPartitionKey().contains(columnMetadata) ||
+                              tableMetadata.getClusteringColumns().containsKey(columnMetadata);
+
+        if (isKeyColumn) return false;
+        if (CqlData.isPrimitive(dataType)) return true;
+        if (dataType instanceof TupleType) return true; // TODO: WRITETIME and TTL functions are very slow on Tuples in cqlsh...should they be supported here?
+        if (CqlData.isFrozen(dataType)) return true;
+        return false;
+    }
+
+    public List<String> getWritetimeTTLColumns() {
+        return this.writetimeTTLColumns.stream()
+                .filter(columnName -> this.columnNames.contains(columnName))
+                .collect(Collectors.toList());
+    }
+
+    public boolean isWritetimeTTLColumn(String columnName) {
+        return this.writetimeTTLColumns.contains(columnName);
+    }
+
     private static ConsistencyLevel mapToConsistencyLevel(String level) {
         ConsistencyLevel retVal = ConsistencyLevel.LOCAL_QUORUM;
         if (StringUtils.isNotEmpty(level)) {
diff --git a/src/resources/sparkConf.properties b/src/resources/sparkConf.properties
@@ -53,14 +53,29 @@ spark.cdm.target.connect.password      cassandra
 # Recommended Parameters:
 #  spark.cdm.schema.origin
 #    .column
-#       .ttl.names      : Default is empty. Names from .column.names to be combined using the MAX
-#                           function to determine the TTL of the entire migrated record. Will use target 
-#                           table default when not set. The names cannot include any columns listed in
-#                           partition-key,clustering-key.
-#       .writetime.names: Default is empty. Names from .column.names to be combined using the MAX
-#                           function to determine the TIMESTAMP of the entire migrated record. Will use 
-#                           target table default when not set. The names cannot include any columns
-#                           listed in the primary key e.g. partition-key,clustering-key
+#       .ttl
+#          .automatic     : Default is true, unless .ttl.names is specified. When true, the TTL of the
+#                           target record will be determined by finding the maxiumum TTL of
+#                           all origin columns that can have TTL set (which excludes partition key,
+#                           clustering key, collections/UDT/tuple, and frozen columns). When false, and
+#                           .names is not set, the target record will have the TTL determined by the target
+#                           table configuration.
+#          .names         : Default is empty, meaning they will be determined automatically if that is set
+#                           (see above). Specify a subset of eligible columns that are used to calculate
+#                           the TTL of the target record.
+#       .writetime
+#          .automatic     : Default is true, unless .writetime.names is specified. When true, the WRITETIME of
+#                           the target record will be determined by finding the maxiumum WRITETIME of
+#                           all origin columns that can have WRITETIME set (which excludes partition key,
+#                           clustering key, collections/UDT/tuple, and frozen columns). When false, and
+#                           .names is not set, the target record will have the WRITETIME determined by the target
+#                           table configuration.
+#
+#                           *** Note spark.cdm.transform.custom.writetime overrides this setting ***
+#
+#          .names         : Default is empty, meaning they will be determined automatically if that is set
+#                           (see above). Specify a subset of eligible columns that are used to calculate
+#                           the WRITETIME of the target record.
 #
 # Other Parameters:
 #  spark.cdm.schema.origin
@@ -70,10 +85,12 @@ spark.cdm.target.connect.password      cassandra
 #                           origin_column_name:target_column_name. The list is comma-separated. Only renamed
 #                           columns need to be listed.
 #-----------------------------------------------------------------------------------------------------------
-spark.cdm.schema.origin.keyspaceTable             keyspace_name.table_name
-spark.cdm.schema.origin.column.ttl.names          data_col1,data_col2,...
-spark.cdm.schema.origin.column.writetime.names    data_col1,data_col2,...
-#spark.cdm.schema.origin.column.names.to.target    partition_col1:partition_col_1,partition_col2:partition_col_2,...
+spark.cdm.schema.origin.keyspaceTable                keyspace_name.table_name
+#spark.cdm.schema.origin.column.ttl.automatic         true
+#spark.cdm.schema.origin.column.ttl.names             data_col1,data_col2,...
+#spark.cdm.schema.origin.column.writetime.automatic   true
+#spark.cdm.schema.origin.column.writetime.names       data_col1,data_col2,...
+#spark.cdm.schema.origin.column.names.to.target       partition_col1:partition_col_1,partition_col2:partition_col_2,...
 
 #===========================================================================================================
 # Details about the Target Schema
diff --git a/src/test/java/com/datastax/cdm/cql/CommonMocks.java b/src/test/java/com/datastax/cdm/cql/CommonMocks.java
@@ -259,6 +259,7 @@ public void setOriginTableWhens() {
                 });
         when(originTable.getColumnCqlTypes()).thenReturn(originColumnTypes);
         when(originTable.getDataType(anyString())).thenAnswer(invocation -> {String name = invocation.getArgument(0,String.class); return originColumnTypes.get(originColumnNames.indexOf(name));});
+        when(originTable.getWritetimeTTLColumns()).thenReturn(originValueColumns);
 
         when(originTable.getOtherCqlTable()).thenReturn(targetTable);
         when(originTable.getCorrespondingIndex(anyInt())).thenAnswer(invocation -> targetColumnNames.indexOf(originColumnNames.get(invocation.getArgument(0, Integer.class))));
diff --git a/src/test/java/com/datastax/cdm/feature/TTLAndWritetimeTest.java b/src/test/java/com/datastax/cdm/feature/TTLAndWritetimeTest.java
