CDM-37 adding codec feature and seeding with some String-to-numeric codecs

mieslep · mieslep · commit 84bf24458614 · 2023-04-12T22:38:41.000+01:00
diff --git a/SIT/features/03_codec/breakData.cql b/SIT/features/03_codec/breakData.cql
@@ -0,0 +1,4 @@
+
+DELETE FROM target.codec WHERE key='key2';
+UPDATE target.codec SET val_int=654321, val_bigint=314159200000000, val_timestamp='2024-04-16 10:30:00+0000' WHERE key='key3';
+SELECT * FROM target.codec;
diff --git a/SIT/features/03_codec/cdm.fixData.assert b/SIT/features/03_codec/cdm.fixData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 3
+Mismatch Record Count: 1
+Corrected Mismatch Record Count: 1
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 1
+Skipped Record Count: 0
diff --git a/SIT/features/03_codec/cdm.migrateData.assert b/SIT/features/03_codec/cdm.migrateData.assert
@@ -0,0 +1,4 @@
+Read Record Count: 3
+Skipped Record Count: 0
+Write Record Count: 3
+Error Record Count: 0
diff --git a/SIT/features/03_codec/cdm.txt b/SIT/features/03_codec/cdm.txt
@@ -0,0 +1,3 @@
+migrateData datastax.cdm.job.Migrate migrate.properties
+validateData datastax.cdm.job.DiffData migrate.properties
+fixData datastax.cdm.job.DiffData fix.properties
diff --git a/SIT/features/03_codec/cdm.validateData.assert b/SIT/features/03_codec/cdm.validateData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 3
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 0
+Corrected Missing Record Count: 0
+Valid Record Count: 3
+Skipped Record Count: 0
diff --git a/SIT/features/03_codec/execute.sh b/SIT/features/03_codec/execute.sh
@@ -0,0 +1,15 @@
+#!/bin/bash -e
+
+workingDir="$1"
+cd "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s migrateData -d "$workingDir" > cdm.migrateData.out 2>cdm.migrateData.err
+/local/cdm-assert.sh -f cdm.migrateData.out -a cdm.migrateData.assert -d "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s validateData -d "$workingDir" > cdm.validateData.out 2>cdm.validateData.err
+/local/cdm-assert.sh -f cdm.validateData.out -a cdm.validateData.assert -d "$workingDir"
+
+cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD $CASS_CLUSTER -f $workingDir/breakData.cql > $workingDir/breakData.out 2> $workingDir/breakData.err
+
+/local/cdm.sh -f cdm.txt -s fixData -d "$workingDir" > cdm.fixData.out 2>cdm.fixData.err
+/local/cdm-assert.sh -f cdm.fixData.out -a cdm.fixData.assert -d "$workingDir"
diff --git a/SIT/features/03_codec/expected.cql b/SIT/features/03_codec/expected.cql
@@ -0,0 +1 @@
+SELECT * FROM target.codec;
diff --git a/SIT/features/03_codec/expected.out b/SIT/features/03_codec/expected.out
@@ -0,0 +1,8 @@
+
+ key  | val_bigint          | val_int | val_timestamp
+------+---------------------+---------+---------------------------------
+ key1 | 9223372036854775807 |    1234 | 2004-06-16 11:00:00.000000+0000
+ key3 |    3141592653589793 |  123456 | 2004-06-16 11:30:00.000000+0000
+ key2 |          2147483648 |   12345 | 2004-06-16 11:15:00.000000+0000
+
+(3 rows)
diff --git a/SIT/features/03_codec/fix.properties b/SIT/features/03_codec/fix.properties
@@ -0,0 +1,17 @@
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.codec
+spark.cdm.schema.target.keyspaceTable             target.codec
+spark.cdm.perfops.numParts                        1
+
+spark.cdm.schema.origin.column.names              key,val_int,val_bigint,val_timestamp
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           key
+spark.cdm.schema.origin.column.types              0,0,0,0
+
+spark.cdm.transform.codecs                        CQL_INT_TO_STRING,CQL_BIGINT_TO_STRING,CQL_TIMESTAMP_TO_STRING
+
+spark.cdm.autocorrect.missing                     true
+spark.cdm.autocorrect.mismatch                    true
+
diff --git a/SIT/features/03_codec/migrate.properties b/SIT/features/03_codec/migrate.properties
@@ -0,0 +1,15 @@
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.codec
+spark.cdm.schema.target.keyspaceTable             target.codec
+spark.cdm.perfops.numParts                        1
+
+spark.cdm.schema.origin.column.names              key,val_int,val_bigint,val_timestamp
+spark.cdm.schema.origin.column.partition.names    key
+spark.cdm.schema.target.column.id.names           key
+spark.cdm.schema.origin.column.types              0,0,0,0
+
+spark.cdm.transform.codecs                        CQL_INT_TO_STRING,CQL_BIGINT_TO_STRING,CQL_TIMESTAMP_TO_STRING
+
+
diff --git a/SIT/features/03_codec/setup.cql b/SIT/features/03_codec/setup.cql
@@ -0,0 +1,9 @@
+DROP TABLE IF EXISTS origin.codec;
+CREATE TABLE origin.codec(key text, val_int text, val_bigint text, val_timestamp text, PRIMARY KEY (key));
+INSERT INTO origin.codec(key,val_int,val_bigint,val_timestamp) VALUES ('key1','1234'  ,'9223372036854775807','1087383600000');
+INSERT INTO origin.codec(key,val_int,val_bigint,val_timestamp) VALUES ('key2','12345' ,'2147483648'         ,'1087384500000');
+INSERT INTO origin.codec(key,val_int,val_bigint,val_timestamp) VALUES ('key3','123456','3141592653589793'   ,'1087385400000');
+SELECT * FROM origin.codec;
+
+DROP TABLE IF EXISTS target.codec;
+CREATE TABLE target.codec(key text, val_int int, val_bigint bigint, val_timestamp timestamp, PRIMARY KEY (key));
diff --git a/pom.xml b/pom.xml
@@ -16,6 +16,7 @@
     <connector.version>3.2.0</connector.version>
     <cassandra.version>3.11.13</cassandra.version>
     <junit.version>5.9.1</junit.version>
+    <mockito.version>4.11.0</mockito.version>
   </properties>
 
   <distributionManagement>
@@ -124,6 +125,12 @@
       </exclusions>
     </dependency>
     <!-- Force Guava 18 on Test CP, Required for Cassandra -->
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <version>${mockito.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
diff --git a/src/main/java/datastax/cdm/cql/CqlHelper.java b/src/main/java/datastax/cdm/cql/CqlHelper.java
@@ -3,6 +3,10 @@
 import com.datastax.oss.driver.api.core.ConsistencyLevel;
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.cql.Row;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry;
+import datastax.cdm.cql.codec.CodecFactory;
+import datastax.cdm.cql.codec.Codecset;
 import datastax.cdm.job.MigrateDataType;
 import datastax.cdm.data.PKFactory;
 import datastax.cdm.feature.Feature;
@@ -64,6 +68,8 @@ public boolean initialize() {
                 feature.alterProperties(this.propertyHelper, this.pkFactory);
         }
 
+        registerTargetCodecs();
+
         originSelectByPartitionRangeStatement = new OriginSelectByPartitionRangeStatement(propertyHelper,this);
         originSelectByPKStatement = new OriginSelectByPKStatement(propertyHelper,this);
         targetInsertStatement = new TargetInsertStatement(propertyHelper,this);
@@ -105,6 +111,23 @@ public Boolean isFeatureEnabled(Featureset featureEnum) {
     public TargetUpdateStatement getTargetUpdateStatement() {return targetUpdateStatement;}
     public TargetSelectByPKStatement getTargetSelectByPKStatement() {return targetSelectByPKStatement;}
 
+    // ----------------- Codec Functions --------------
+    private void registerTargetCodecs() {
+        List<String> codecList = propertyHelper.getStringList(KnownProperties.TRANSFORM_CODECS);
+        if (null!=codecList && !codecList.isEmpty()) {
+            MutableCodecRegistry registry =
+                    (MutableCodecRegistry) targetSession.getContext().getCodecRegistry();
+
+            StringBuilder sb = new StringBuilder("PARAM -- Codecs Enabled: ");
+            for (String codec : codecList) {
+                TypeCodec<?> typeCodec = CodecFactory.getCodec(Codecset.valueOf(codec));
+                registry.register(typeCodec);
+                sb.append(codec).append(" ");
+            }
+            logger.info(sb.toString());
+        }
+    }
+
 
     // --------------- Session and Performance -------------------------
     public void setOriginSession(CqlSession originSession) {
diff --git a/src/main/java/datastax/cdm/cql/codec/CodecFactory.java b/src/main/java/datastax/cdm/cql/codec/CodecFactory.java
@@ -0,0 +1,15 @@
+package datastax.cdm.cql.codec;
+
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+
+public class CodecFactory {
+    public static TypeCodec<?> getCodec(Codecset codec) {
+        switch (codec) {
+            case CQL_INT_TO_STRING: return new CqlIntToStringCodec();
+            case CQL_BIGINT_TO_STRING: return new CqlBigintToStringCodec();
+            case CQL_TIMESTAMP_TO_STRING: return new CqlTimestampToStringCodec();
+            default:
+                throw new IllegalArgumentException("Unknown codec: " + codec);
+        }
+    }
+}
diff --git a/src/main/java/datastax/cdm/cql/codec/Codecset.java b/src/main/java/datastax/cdm/cql/codec/Codecset.java
@@ -0,0 +1,7 @@
+package datastax.cdm.cql.codec;
+
+public enum Codecset {
+    CQL_INT_TO_STRING,
+    CQL_BIGINT_TO_STRING,
+    CQL_TIMESTAMP_TO_STRING
+}
diff --git a/src/main/java/datastax/cdm/cql/codec/CqlBigintToStringCodec.java b/src/main/java/datastax/cdm/cql/codec/CqlBigintToStringCodec.java
@@ -0,0 +1,56 @@
+package datastax.cdm.cql.codec;
+
+import com.datastax.oss.driver.api.core.ProtocolVersion;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import org.jetbrains.annotations.NotNull;
+
+import java.nio.ByteBuffer;
+
+/**
+ * This codec converts a CQL BigInt to a Java String.
+ */
+public class CqlBigintToStringCodec implements TypeCodec<String> {
+
+    @Override
+    public @NotNull GenericType<String> getJavaType() {
+        return GenericType.STRING;
+    }
+
+    @Override
+    public @NotNull DataType getCqlType() {
+        return DataTypes.BIGINT;
+    }
+
+    @Override
+    public ByteBuffer encode(String value, @NotNull ProtocolVersion protocolVersion) {
+        if (value == null) {
+            return null;
+        } else {
+            long longValue = Long.parseLong(value);
+            return TypeCodecs.BIGINT.encode(longValue, protocolVersion);
+        }
+    }
+
+    @Override
+    public String decode(ByteBuffer bytes, @NotNull ProtocolVersion protocolVersion) {
+        Long longValue = TypeCodecs.BIGINT.decode(bytes, protocolVersion);
+        return longValue.toString();
+    }
+
+    @Override
+    public @NotNull String format(String value) {
+        long longValue = Long.parseLong(value);
+        return TypeCodecs.BIGINT.format(longValue);
+    }
+
+    @Override
+    public String parse(String value) {
+        Long longValue = TypeCodecs.BIGINT.parse(value);
+        return longValue == null ? null : longValue.toString();
+    }
+}
+
diff --git a/src/main/java/datastax/cdm/cql/codec/CqlIntToStringCodec.java b/src/main/java/datastax/cdm/cql/codec/CqlIntToStringCodec.java
@@ -0,0 +1,56 @@
+package datastax.cdm.cql.codec;
+
+import com.datastax.oss.driver.api.core.ProtocolVersion;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import org.jetbrains.annotations.NotNull;
+
+import java.nio.ByteBuffer;
+
+/**
+ * This codec converts a CQL INT to a Java String.
+ */
+public class CqlIntToStringCodec implements TypeCodec<String> {
+
+    @Override
+    public @NotNull GenericType<String> getJavaType() {
+        return GenericType.STRING;
+    }
+
+    @Override
+    public @NotNull DataType getCqlType() {
+        return DataTypes.INT;
+    }
+
+    @Override
+    public ByteBuffer encode(String value, @NotNull ProtocolVersion protocolVersion) {
+        if (value == null) {
+            return null;
+        } else {
+            int intValue = Integer.parseInt(value);
+            return TypeCodecs.INT.encode(intValue, protocolVersion);
+        }
+    }
+
+    @Override
+    public String decode(ByteBuffer bytes, @NotNull ProtocolVersion protocolVersion) {
+        Integer intValue = TypeCodecs.INT.decode(bytes, protocolVersion);
+        return intValue.toString();
+    }
+
+    @Override
+    public @NotNull String format(String value) {
+        int intValue = Integer.parseInt(value);
+        return TypeCodecs.INT.format(intValue);
+    }
+
+    @Override
+    public String parse(String value) {
+        Integer intValue = TypeCodecs.INT.parse(value);
+        return intValue == null ? null : intValue.toString();
+    }
+}
+
diff --git a/src/main/java/datastax/cdm/cql/codec/CqlTimestampToStringCodec.java b/src/main/java/datastax/cdm/cql/codec/CqlTimestampToStringCodec.java
@@ -0,0 +1,56 @@
+package datastax.cdm.cql.codec;
+
+import com.datastax.oss.driver.api.core.ProtocolVersion;
+import com.datastax.oss.driver.api.core.type.DataTypes;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodec;
+import com.datastax.oss.driver.api.core.type.codec.TypeCodecs;
+import com.datastax.oss.driver.api.core.type.DataType;
+import com.datastax.oss.driver.api.core.type.reflect.GenericType;
+import org.jetbrains.annotations.NotNull;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+
+/**
+ * This codec converts a CQL TIMESTAMP to a Java String.
+ */
+public class CqlTimestampToStringCodec implements TypeCodec<String> {
+
+    @Override
+    public @NotNull GenericType<String> getJavaType() {
+        return GenericType.STRING;
+    }
+
+    @Override
+    public @NotNull DataType getCqlType() {
+        return DataTypes.TIMESTAMP;
+    }
+
+    @Override
+    public ByteBuffer encode(String value, @NotNull ProtocolVersion protocolVersion) {
+        if (value == null) {
+            return null;
+        } else {
+            Instant instantValue = Instant.ofEpochMilli(Long.parseLong(value));
+            return TypeCodecs.TIMESTAMP.encode(instantValue, protocolVersion);
+        }
+    }
+
+    @Override
+    public String decode(ByteBuffer bytes, @NotNull ProtocolVersion protocolVersion) {
+        Instant instantValue = TypeCodecs.TIMESTAMP.decode(bytes, protocolVersion);
+        return instantValue == null ? null : String.valueOf(instantValue.toEpochMilli());
+    }
+
+    @Override
+    public @NotNull String format(String value) {
+        Instant instantValue = Instant.ofEpochMilli(Long.parseLong(value));
+        return TypeCodecs.TIMESTAMP.format(instantValue);
+    }
+
+    @Override
+    public String parse(String value) {
+        Instant instantValue = TypeCodecs.TIMESTAMP.parse(value);
+        return instantValue == null ? null : String.valueOf(instantValue.toEpochMilli());
+    }
+}
diff --git a/src/main/java/datastax/cdm/properties/KnownProperties.java b/src/main/java/datastax/cdm/properties/KnownProperties.java
@@ -174,6 +174,8 @@ 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_CODECS                 = "spark.cdm.transform.codecs";
+
 
 //    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
@@ -182,7 +184,9 @@ 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(GUARDRAIL_FIELD_LIMIT_MB, PropertyType.NUMBER);
+        types.put(TRANSFORM_CODECS, PropertyType.STRING_LIST);
+
+     //           types.put(GUARDRAIL_FIELD_LIMIT_MB, PropertyType.NUMBER);
 //        defaults.put(GUARDRAIL_FIELD_LIMIT_MB, "0");
     }
 
diff --git a/src/resources/sparkConf.properties b/src/resources/sparkConf.properties
@@ -234,9 +234,13 @@ spark.cdm.perfops.writeRateLimit                  40000
 #                                    columns are collections). This parameter allows a crude constant value
 #                                    to be used in its place, and overrides
 #                                    .schema.origin.column.writetime.indexes.
+#   .codecs                          Default is empty. A comma-separated list of additional codecs to
+#                                    enable. These correspond with the Enum values in
+#                                    datastax.cdm.cql.codec.Codecset .
 #-----------------------------------------------------------------------------------------------------------
 #spark.cdm.transform.missing.key.ts.replace.value
 #spark.cdm.transform.custom.writetime            0
+#spark.cdm.transform.codecs
 
 #===========================================================================================================
 # Cassandra Filters are applied on the coordinator node. Note that, depending on the filter, the coordinator
diff --git a/src/test/java/datastax/cdm/cql/codec/CqlBigintToStringCodecTest.java b/src/test/java/datastax/cdm/cql/codec/CqlBigintToStringCodecTest.java
diff --git a/src/test/java/datastax/cdm/cql/codec/CqlIntToStringCodecTest.java b/src/test/java/datastax/cdm/cql/codec/CqlIntToStringCodecTest.java
diff --git a/src/test/java/datastax/cdm/cql/codec/CqlTimestampToStringCodecTest.java b/src/test/java/datastax/cdm/cql/codec/CqlTimestampToStringCodecTest.java

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+migrateData datastax.cdm.job.Migrate migrate.properties`
	`2`	`+validateData datastax.cdm.job.DiffData migrate.properties`
	`3`	`+fixData datastax.cdm.job.DiffData fix.properties`