Merge pull request #122 from datastax/issue/CDM-35

Issue/cdm 35

Author: mieslep

src/main/java/datastax/cdm/cql/statement/AbstractOriginSelectStatement.java

@@ -23,6 +23,7 @@ public abstract class AbstractOriginSelectStatement extends BaseCdmStatement {
     private final Boolean writeTimestampFilterEnabled;
     private final Long minWriteTimeStampFilter;
     private final Long maxWriteTimeStampFilter;
+    private final Long customWritetime;
 
     private final Boolean filterColumnEnabled;
     private final Integer filterColumnIndex;
@@ -46,6 +47,11 @@ public AbstractOriginSelectStatement(PropertyHelper propertyHelper, CqlHelper cq
                     Instant.ofEpochMilli(getMaxWriteTimeStampFilter() / 1000));
         }
 
+        customWritetime = getCustomWritetime();
+        if (customWritetime > 0)
+            logger.info("PARAM -- {}: {} datetime is {} ", KnownProperties.TRANSFORM_CUSTOM_WRITETIME, customWritetime,
+                    Instant.ofEpochMilli(customWritetime / 1000));
+
         filterColumnString = getFilterColumnString();
         filterColumnIndex = getFilterColumnIndex();
         filterColumnType = getFilterColumnType();
@@ -107,6 +113,7 @@ public boolean shouldFilterRecord(Record record) {
     }
 
     public Long getLargestWriteTimeStamp(Row row) {
+        if (customWritetime > 0) return customWritetime;
         if (null==writeTimestampIndexes || writeTimestampIndexes.isEmpty()) return null;
         OptionalLong max = writeTimestampIndexes.stream()
                 .mapToLong(row::getLong)
@@ -172,6 +179,11 @@ public boolean hasWriteTimestampFilter() {
                 null != writetimeCols && !writetimeCols.isEmpty());
     }
 
+    private Long getCustomWritetime() {
+        Long rtn = propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME);
+        return (null==rtn || rtn < 0) ? 0L : rtn;
+    }
+
     private String getFilterColumnString() {
         String rtn = propertyHelper.getString(KnownProperties.FILTER_COLUMN_VALUE);
         if (null!=rtn) return rtn.trim();

src/main/java/datastax/cdm/cql/statement/AbstractTargetUpsertStatement.java

@@ -116,7 +116,9 @@ private void setTTLAndWriteTimeNames() {
         List<Integer> ttlColumnNames = propertyHelper.getIntegerList(KnownProperties.ORIGIN_TTL_INDEXES);
         usingTTL = null!= ttlColumnNames && !ttlColumnNames.isEmpty();
         List<Integer> writeTimeColumnNames = propertyHelper.getIntegerList(KnownProperties.ORIGIN_WRITETIME_INDEXES);
-        usingWriteTime = null!= writeTimeColumnNames && !writeTimeColumnNames.isEmpty();
+        Long customWritetime = propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME);
+        usingWriteTime = (null!= writeTimeColumnNames && !writeTimeColumnNames.isEmpty()
+                || null != customWritetime && customWritetime > 0);
     }
 
     private void setCounterIndexes() {

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

@@ -173,16 +173,15 @@ public enum PropertyType {
     // Guardrails and Transformations
     //==========================================================================
     public static final String TRANSFORM_REPLACE_MISSING_TS     = "spark.cdm.transform.missing.key.ts.replace.value";
+    public static final String TRANSFORM_CUSTOM_WRITETIME       = "spark.cdm.transform.custom.writetime";
 
-//    TODO: this was previously in the configuration but not referenced in 3.3.0 other than in config
-//    public static final String TRANSFORM_WRITETIME_CUSTOM_VALUE = "spark.target.custom.writeTime";  // 0
 //    TODO: 3.3.0 refactored how guardrails are handled, this needs to be merged forward
 //    public static final String GUARDRAIL_FIELD_LIMIT_MB         = "spark.guardrail.colSizeInKB"; //10
 
     static {
         types.put(TRANSFORM_REPLACE_MISSING_TS, PropertyType.NUMBER);
-//        types.put(TRANSFORM_WRITETIME_CUSTOM_VALUE, PropertyType.NUMBER);
-//        defaults.put(TRANSFORM_WRITETIME_CUSTOM_VALUE, "0");
+        types.put(TRANSFORM_CUSTOM_WRITETIME, PropertyType.NUMBER);
+     defaults.put(TRANSFORM_CUSTOM_WRITETIME, "0");
 //           types.put(GUARDRAIL_FIELD_LIMIT_MB, PropertyType.NUMBER);
 //        defaults.put(GUARDRAIL_FIELD_LIMIT_MB, "0");
     }

src/resources/sparkConf.properties

@@ -228,8 +228,15 @@ spark.cdm.perfops.writeRateLimit                  40000
 #                                    MigrateData operation would fail. This parameter allows a crude 
 #                                    constant value to be used in its place, separate from the Constant 
 #                                    Values feature.
+#   .custom.writetime                Default is 0 (diabled). Timestamp value in microseconds to use as the
+#                                    WRITETIME for the target record. This is useful when the WRITETIME of
+#                                    the record in Origin cannot be determined (such as the only non-key
+#                                    columns are collections). This parameter allows a crude constant value
+#                                    to be used in its place, and overrides
+#                                    .schema.origin.column.writetime.indexes.
 #-----------------------------------------------------------------------------------------------------------
 #spark.cdm.transform.missing.key.ts.replace.value
+#spark.cdm.transform.custom.writetime            0
 
 #===========================================================================================================
 # Cassandra Filters are applied on the coordinator node. Note that, depending on the filter, the coordinator