CDM-54 file-driven jobs now as SIT tests

mieslep · mieslep · commit bd145c0406b4 · 2023-05-23T09:27:02.000+01:00
diff --git a/SIT/features/06_partition_range/breakData.cql b/SIT/features/06_partition_range/breakData.cql
@@ -0,0 +1,5 @@
+
+DELETE FROM target.feature_partition_range WHERE key='key1';
+UPDATE target.feature_partition_range SET value='value999' WHERE key='key2';
+SELECT * FROM target.feature_partition_range;
+
diff --git a/SIT/features/06_partition_range/cdm.txt b/SIT/features/06_partition_range/cdm.txt
@@ -0,0 +1,2 @@
+migrateData com.datastax.cdm.job.MigratePartitionsFromFile migrate.properties
+validateData com.datastax.cdm.job.DiffPartitionsFromFile migrate.properties
diff --git a/SIT/features/06_partition_range/execute.sh b/SIT/features/06_partition_range/execute.sh
@@ -0,0 +1,11 @@
+#!/bin/bash -e
+
+workingDir="$1"
+cd "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s migrateData -d "$workingDir"
+
+cqlsh -u $CASS_USERNAME -p $CASS_PASSWORD $CASS_CLUSTER -f $workingDir/breakData.cql > $workingDir/other.breakData.out 2> $workingDir/other.breakData.err
+
+/local/cdm.sh -f cdm.txt -s validateData -d "$workingDir"
+
diff --git a/SIT/features/06_partition_range/expected.cql b/SIT/features/06_partition_range/expected.cql
@@ -0,0 +1 @@
+SELECT * FROM target.feature_partition_range;
diff --git a/SIT/features/06_partition_range/expected.out b/SIT/features/06_partition_range/expected.out
@@ -0,0 +1,7 @@
+
+ key  | value
+------+--------
+ key1 | valueA
+ key2 | valueB
+
+(2 rows)
diff --git a/SIT/features/06_partition_range/migrate.properties b/SIT/features/06_partition_range/migrate.properties
@@ -0,0 +1,10 @@
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.feature_partition_range
+spark.cdm.schema.target.keyspaceTable             target.feature_partition_range
+spark.cdm.perfops.numParts                        1
+
+spark.cdm.autocorrect.missing                     true
+spark.cdm.autocorrect.mismatch                    true
+
diff --git a/SIT/features/06_partition_range/partitions.csv b/SIT/features/06_partition_range/partitions.csv
@@ -0,0 +1,2 @@
+0,2000000000000000000
+8100000000000000000,8500000000000000000
diff --git a/SIT/features/06_partition_range/setup.cql b/SIT/features/06_partition_range/setup.cql
@@ -0,0 +1,8 @@
+DROP TABLE IF EXISTS origin.feature_partition_range;
+CREATE TABLE origin.feature_partition_range(key text, value text, PRIMARY KEY (key));
+INSERT INTO origin.feature_partition_range(key,value) VALUES ('key1','valueA');
+INSERT INTO origin.feature_partition_range(key,value) VALUES ('key2','valueB');
+INSERT INTO origin.feature_partition_range(key,value) VALUES ('key3','valueC');
+
+DROP TABLE IF EXISTS target.feature_partition_range;
+CREATE TABLE target.feature_partition_range(key text, value text, PRIMARY KEY (key));
diff --git a/SIT/features/07_migrate_rows/cdm.txt b/SIT/features/07_migrate_rows/cdm.txt
@@ -0,0 +1 @@
+migrateData com.datastax.cdm.job.MigrateRowsFromFile migrate.properties
diff --git a/SIT/features/07_migrate_rows/execute.sh b/SIT/features/07_migrate_rows/execute.sh
@@ -0,0 +1,8 @@
+#!/bin/bash -e
+
+workingDir="$1"
+cd "$workingDir"
+
+/local/cdm.sh -f cdm.txt -s migrateData -d "$workingDir"
+
+
diff --git a/SIT/features/07_migrate_rows/expected.cql b/SIT/features/07_migrate_rows/expected.cql
@@ -0,0 +1 @@
+SELECT * FROM target.feature_migrate_rows;
diff --git a/SIT/features/07_migrate_rows/expected.out b/SIT/features/07_migrate_rows/expected.out
@@ -0,0 +1,7 @@
+
+ key  | key2 | value
+------+------+--------
+ key1 |    A | valueA
+ key3 |    C | valueC
+
+(2 rows)
diff --git a/SIT/features/07_migrate_rows/migrate.properties b/SIT/features/07_migrate_rows/migrate.properties
@@ -0,0 +1,8 @@
+spark.cdm.origin.connect.host                     cdm-sit-cass
+spark.cdm.target.connect.host                     cdm-sit-cass
+
+spark.cdm.schema.origin.keyspaceTable             origin.feature_migrate_rows
+spark.cdm.schema.target.keyspaceTable             target.feature_migrate_rows
+spark.cdm.perfops.numParts                        1
+
+
diff --git a/SIT/features/07_migrate_rows/primary_key_rows.csv b/SIT/features/07_migrate_rows/primary_key_rows.csv
@@ -0,0 +1,2 @@
+key1 %% A
+key3 %% C
diff --git a/SIT/features/07_migrate_rows/setup.cql b/SIT/features/07_migrate_rows/setup.cql
@@ -0,0 +1,8 @@
+DROP TABLE IF EXISTS origin.feature_migrate_rows;
+CREATE TABLE origin.feature_migrate_rows(key text, key2 text, value text, PRIMARY KEY (key,key2));
+INSERT INTO origin.feature_migrate_rows(key,key2,value) VALUES ('key1','A','valueA');
+INSERT INTO origin.feature_migrate_rows(key,key2,value) VALUES ('key2','B','valueB');
+INSERT INTO origin.feature_migrate_rows(key,key2,value) VALUES ('key3','C','valueC');
+
+DROP TABLE IF EXISTS target.feature_migrate_rows;
+CREATE TABLE target.feature_migrate_rows(key text, key2 text,value text, PRIMARY KEY (key,key2));
diff --git a/src/main/java/com/datastax/cdm/data/EnhancedPK.java b/src/main/java/com/datastax/cdm/data/EnhancedPK.java
@@ -79,6 +79,7 @@ public Integer getTTL() {
 
     private void validate() {
         if (null==values || null== classes || values.isEmpty() || values.size() != classes.size()) {
+            if (null==this.messages) this.messages = new ArrayList<>();
             this.messages.add("ERROR: types and/or values are null and/or empty, or are not the same size");
             this.errorState = true;
             return;
diff --git a/src/main/java/com/datastax/cdm/job/CopyPKJobSession.java b/src/main/java/com/datastax/cdm/job/CopyPKJobSession.java
@@ -108,6 +108,7 @@ public void printCounts(boolean isFinal) {
     private EnhancedPK toEnhancedPK(String rowString) {
         String[] pkFields = rowString.split(" %% ");
         List<Object> values = new ArrayList<>(originPKClasses.size());
+        if (logger.isDebugEnabled()) logger.debug("rowString={}, pkFields={}", rowString, pkFields);
         for (int i=0; i<pkFields.length; i++) {
             PropertyEditor editor = PropertyEditorManager.findEditor(originPKClasses.get(i));
             editor.setAsText(pkFields[i]);
diff --git a/src/main/java/com/datastax/cdm/job/SplitPartitions.java b/src/main/java/com/datastax/cdm/job/SplitPartitions.java
@@ -115,10 +115,12 @@ private static List<Partition> getSubPartitions(int numSplits, BigInteger min, B
     }
 
     public static class PKRows implements Serializable {
+        private static final long serialVersionUID = 1L;
+
         List<String> pkRows;
 
         public PKRows(List<String> rows) {
-            pkRows = rows;
+            pkRows = new ArrayList<>(rows);
         }
     }
 
diff --git a/src/main/scala/com/datastax/cdm/job/BaseJob.scala b/src/main/scala/com/datastax/cdm/job/BaseJob.scala
@@ -20,7 +20,7 @@ abstract class BaseJob[T: ClassTag] extends App {
 
   var jobName: String = _
   var jobFactory: IJobSessionFactory[T] = _
-  var inputFilename: String = ""
+  var fileName: String = ""
 
   var spark: SparkSession = _
   var sContext: SparkContext = _
@@ -39,8 +39,8 @@ abstract class BaseJob[T: ClassTag] extends App {
   var originConnection: CassandraConnector = _
   var targetConnection: CassandraConnector = _
 
-  def setup(jobName: String, jobFactory: IJobSessionFactory[T], inputFilename: String): Unit = {
-    this.inputFilename = inputFilename
+  def setup(jobName: String, jobFactory: IJobSessionFactory[T], fileName: String): Unit = {
+    this.fileName = fileName
     setup(jobName, jobFactory)
   }
 
@@ -69,7 +69,7 @@ abstract class BaseJob[T: ClassTag] extends App {
     this.parts = getParts(numSplits)
     this.slices = sContext.parallelize(parts.asScala.toSeq, parts.size);
     abstractLogger.info("PARAM Calculated -- Total Partitions: " + parts.size())
-    abstractLogger.info("Spark parallelize created : " + slices.count() + " parts!");
+    abstractLogger.info("Spark parallelize created : " + slices.count() + " slices!");
 
     val connectionFetcher = new ConnectionFetcher(sContext, propertyHelper)
     originConnection = connectionFetcher.getConnection("ORIGIN", consistencyLevel)
diff --git a/src/main/scala/com/datastax/cdm/job/BasePKJob.scala b/src/main/scala/com/datastax/cdm/job/BasePKJob.scala
@@ -6,6 +6,6 @@ abstract class BasePKJob extends BaseJob[SplitPartitions.PKRows] {
   override def getParts(pieces: Int): util.Collection[SplitPartitions.PKRows] = {
     // This takes a file with N rows and divides it into pieces of size N/pieces
     // Each PKRows object contains a list of Strings that contain the PK to be parsed
-    SplitPartitions.getRowPartsFromFile(pieces, this.inputFilename)
+    SplitPartitions.getRowPartsFromFile(pieces, this.fileName)
   }
 }
diff --git a/src/main/scala/com/datastax/cdm/job/BasePartitionJob.scala b/src/main/scala/com/datastax/cdm/job/BasePartitionJob.scala
@@ -4,11 +4,11 @@ import java.util
 
 abstract class BasePartitionJob extends BaseJob[SplitPartitions.Partition] {
   override def getParts(pieces: Int): util.Collection[SplitPartitions.Partition] = {
-    if ("".equals(this.inputFilename)) {
+    if ("".equals(this.fileName)) {
       abstractLogger.info("PARAM -- Coverage Percent: " + coveragePercent)
       SplitPartitions.getRandomSubPartitions(pieces, minPartition, maxPartition, coveragePercent)
     } else {
-      SplitPartitions.getSubPartitionsFromFile(pieces, this.inputFilename)
+      SplitPartitions.getSubPartitionsFromFile(pieces, this.fileName)
     }
   }
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+migrateData com.datastax.cdm.job.MigratePartitionsFromFile migrate.properties`
	`2`	`+validateData com.datastax.cdm.job.DiffPartitionsFromFile migrate.properties`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+SELECT * FROM target.feature_partition_range;`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+0,2000000000000000000`
	`2`	`+8100000000000000000,8500000000000000000`