CDM-52 : implement writetimeIncrement feature which can avoid autocorrect creating duplicate list entries

Author: mieslep

src/main/java/com/datastax/cdm/feature/WritetimeTTL.java

@@ -27,6 +27,7 @@ public class WritetimeTTL extends AbstractFeature  {
     private Long filterMin;
     private Long filterMax;
     private boolean hasWriteTimestampFilter;
+    private Long writetimeIncrement;
 
     @Override
     public boolean loadProperties(IPropertyHelper propertyHelper) {
@@ -49,6 +50,8 @@ public boolean loadProperties(IPropertyHelper propertyHelper) {
             this.autoWritetimeNames = false;
         }
 
+        this.writetimeIncrement = propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT);
+
         this.filterMin = getMinFilter(propertyHelper);
         this.filterMax = getMaxFilter(propertyHelper);
         this.hasWriteTimestampFilter = (null != filterMin && null != filterMax && filterMin > 0 && filterMax > 0 && filterMax > filterMin);
@@ -74,6 +77,12 @@ protected boolean validateProperties() {
         validateFilterRangeProperties();
         validateTTLNames();
         validateWritetimeNames();
+
+        if (null==this.writetimeIncrement || this.writetimeIncrement < 0L) {
+            logger.error(KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT + " must be set to a value greater than or equal to zero");
+            isValid = false;
+        }
+
         return isValid;
     }
 
@@ -120,6 +129,12 @@ public boolean initializeAndValidate(CqlTable originTable, CqlTable targetTable)
             isValid = false;
         }
 
+        if (this.writetimeIncrement == 0L && (null!=writetimeNames && !writetimeNames.isEmpty()) && originTable.hasUnfrozenList()) {
+            logger.warn("Writetime is configured, but the origin table at least one unfrozen List, and there is a zero-value increment configured at "+
+                    KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT+"; this may result in duplicate list entries when "+
+                    KnownProperties.AUTOCORRECT_MISMATCH+" is enabled.");
+        }
+
         if (!isValid) isEnabled = false;
         logger.info("Feature {} is {}", this.getClass().getSimpleName(), isEnabled?"enabled":"disabled");
         return isValid;
@@ -162,7 +177,7 @@ public Long getLargestWriteTimeStamp(Row row) {
                 .mapToLong(row::getLong)
                 .filter(Objects::nonNull)
                 .max();
-        return max.isPresent() ? max.getAsLong() : null;
+        return max.isPresent() ? max.getAsLong() + this.writetimeIncrement : null;
     }
 
     public Integer getLargestTTL(Row row) {

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

@@ -142,6 +142,7 @@ public enum PropertyType {
     //==========================================================================
     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";
+    public static final String TRANSFORM_CUSTOM_WRITETIME_INCREMENT            = "spark.cdm.transform.custom.writetime.incrementBy";
     public static final String TRANSFORM_CODECS                                = "spark.cdm.transform.codecs";
     public static final String TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT        = "spark.cdm.transform.codecs.timestamp.string.format";
     public static final String TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT_ZONE   = "spark.cdm.transform.codecs.timestamp.string.zone";
@@ -155,6 +156,8 @@ public enum PropertyType {
         types.put(TRANSFORM_REPLACE_MISSING_TS, PropertyType.NUMBER);
         types.put(TRANSFORM_CUSTOM_WRITETIME, PropertyType.NUMBER);
      defaults.put(TRANSFORM_CUSTOM_WRITETIME, "0");
+        types.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, PropertyType.NUMBER);
+     defaults.put(TRANSFORM_CUSTOM_WRITETIME_INCREMENT, "0");
         types.put(TRANSFORM_CODECS, PropertyType.STRING_LIST);
         types.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, PropertyType.STRING);
      defaults.put(TRANSFORM_CODECS_TIMESTAMP_STRING_FORMAT, "yyyyMMddHHmmss");

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

@@ -342,6 +342,14 @@ public boolean isWritetimeTTLColumn(String columnName) {
         return this.writetimeTTLColumns.contains(columnName);
     }
 
+    public boolean hasUnfrozenList() {
+        return this.cqlAllColumns.stream()
+                .filter(columnMetadata ->
+                        columnNames.contains(columnMetadata.getName().asInternal()) &&
+                        columnMetadata.getType() instanceof ListType)
+                .anyMatch(columnMetadata -> !CqlData.isFrozen(columnMetadata.getType()));
+    }
+
     private static ConsistencyLevel mapToConsistencyLevel(String level) {
         ConsistencyLevel retVal = ConsistencyLevel.LOCAL_QUORUM;
         if (StringUtils.isNotEmpty(level)) {

src/resources/sparkConf.properties

@@ -191,12 +191,18 @@ 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
+#   .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.
+#      .writetime.incrementBy        Default is 0. This is useful when you have a List that is not frozen,
+#                                    and are updating this via the autocorrect feature. Lists are not idempotent,
+#                                    and subsequent UPSERTs would add duplicates to the list. Future versions
+#                                    of CDM may tombstone the previous list, but for now this solution is
+#                                    viable and, crucially, more performant.
 #   .codecs                          Default is empty. A comma-separated list of additional codecs to
 #                                    enable. Current codecs are:
 #                                        INT_STRING                : int stored in a String
@@ -216,7 +222,8 @@ spark.cdm.perfops.writeRateLimit                  40000
 #     .string.zone                 Default is UTC ; Must be in ZoneRulesProvider.getAvailableZoneIds()
 #-----------------------------------------------------------------------------------------------------------
 #spark.cdm.transform.missing.key.ts.replace.value
-#spark.cdm.transform.custom.writetime            0
+#spark.cdm.transform.custom.writetime                  0
+#spark.cdm.transform.custom.writetime.incrementBy      0
 #spark.cdm.transform.codecs
 #spark.cdm.transform.codecs.timestamp.string.format    yyyyMMddHHmmss
 #spark.cdm.transform.codecs.timestamp.string.zone      UTC

src/test/java/com/datastax/cdm/feature/TTLAndWritetimeTest.java

@@ -428,4 +428,38 @@ public void testOriginIsTarget() {
         feature.loadProperties(propertyHelper);
         assertThrows(IllegalArgumentException.class, () -> feature.initializeAndValidate(originTable, targetTable));
     }
+
+    @Test
+    public void writetimeIncrementTest() {
+        when(propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME)).thenReturn(0L);
+        when(propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT)).thenReturn(314L);
+        when(originTable.indexOf("WRITETIME("+writetimeColumnName+")")).thenReturn(100);
+        when(originRow.getLong(eq(100))).thenReturn(1000L);
+        when(originTable.indexOf("WRITETIME("+writetimeTTLColumnName+")")).thenReturn(101);
+        when(originRow.getLong(eq(101))).thenReturn(3000L);
+
+        feature.loadProperties(propertyHelper);
+        feature.initializeAndValidate(originTable, targetTable);
+
+        Long largestWritetime = feature.getLargestWriteTimeStamp(originRow);
+        assertEquals(3314L, largestWritetime);
+    }
+
+    @Test
+    public void invalidWritetimeIncrement() {
+        when(propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT)).thenReturn(-1L);
+        feature.loadProperties(propertyHelper);
+        assertFalse(feature.isEnabled());
+    }
+
+    @Test
+    public void testZeroIncrementWithUnfrozenList() {
+        when(propertyHelper.getLong(KnownProperties.TRANSFORM_CUSTOM_WRITETIME_INCREMENT)).thenReturn(0L);
+        when(originTable.hasUnfrozenList()).thenReturn(true);
+
+        feature.loadProperties(propertyHelper);
+        feature.initializeAndValidate(originTable, targetTable);
+        assertTrue(feature.isEnabled());
+    }
+
 }