CDM-28 adding Counter tests

mieslep · mieslep · commit b4b85610924d · 2023-04-05T21:54:01.000+01:00
diff --git a/SIT/smoke/04_counters/breakData.cql b/SIT/smoke/04_counters/breakData.cql
@@ -0,0 +1,18 @@
+
+UPDATE source.smoke_counters SET col1=col1+1 WHERE key='record2';
+
+-- CDM-29 discovered that this record will "zombie"...
+DELETE FROM target.smoke_counters where key='record3';
+
+-- change by same amount on both sides, records should match
+UPDATE source.smoke_counters SET col2=col2+1 WHERE key='record4';
+UPDATE target.smoke_counters SET col2=col2+1 WHERE key='record4';
+
+-- change by different amounts, they shouuld be brought into alignment with 
+-- source, even if that means going down
+UPDATE source.smoke_counters SET col2=col2+1 WHERE key='record5';
+UPDATE target.smoke_counters SET col2=col2+2 WHERE key='record5';
+
+SELECT * FROM source.smoke_counters;
+SELECT * FROM target.smoke_counters;
+
diff --git a/SIT/smoke/04_counters/cdm.fixData.assert b/SIT/smoke/04_counters/cdm.fixData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 5
+Mismatch Record Count: 2
+Corrected Mismatch Record Count: 2
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 2
+Skipped Record Count: 0
diff --git a/SIT/smoke/04_counters/cdm.migrateData.assert b/SIT/smoke/04_counters/cdm.migrateData.assert
@@ -0,0 +1,4 @@
+Read Record Count: 5
+Skipped Record Count: 0
+Write Record Count: 5
+Error Record Count: 0
diff --git a/SIT/smoke/04_counters/cdm.txt b/SIT/smoke/04_counters/cdm.txt
@@ -0,0 +1,3 @@
+migrateData datastax.astra.migrate.Migrate migrate.properties
+validateData datastax.astra.migrate.DiffData migrate.properties
+fixData datastax.astra.migrate.DiffData fix.properties
diff --git a/SIT/smoke/04_counters/cdm.validateData.assert b/SIT/smoke/04_counters/cdm.validateData.assert
@@ -0,0 +1,7 @@
+Read Record Count: 5
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 0
+Corrected Missing Record Count: 0
+Valid Record Count: 5
+Skipped Record Count: 0
diff --git a/SIT/smoke/04_counters/execute.sh b/SIT/smoke/04_counters/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/smoke/04_counters/expected.cql b/SIT/smoke/04_counters/expected.cql
@@ -0,0 +1 @@
+SELECT key,col1,col2 FROM target.smoke_counters;
diff --git a/SIT/smoke/04_counters/expected.out b/SIT/smoke/04_counters/expected.out
@@ -0,0 +1,10 @@
+
+ key     | col1  | col2
+---------+-------+-------
+ record3 |  2000 |  4000
+ record5 |   500 |   501
+ record2 |    11 |    20
+ record1 |     1 |     2
+ record4 | 10000 | 20001
+
+(5 rows)
diff --git a/SIT/smoke/04_counters/fix.properties b/SIT/smoke/04_counters/fix.properties
@@ -0,0 +1,23 @@
+spark.origin.host                                 cdm-sit-cass
+spark.origin.port                                 9042
+spark.origin.username                             cassandra
+spark.origin.password                             cassandra
+spark.origin.keyspaceTable                        source.smoke_counters
+
+spark.target.host                                 cdm-sit-cass
+spark.target.port                                 9042
+spark.target.username                             cassandra
+spark.target.password                             cassandra
+spark.target.keyspaceTable                        target.smoke_counters
+
+spark.numSplits                                   1
+
+spark.query.origin                                key,col1,col2
+spark.query.types                                 0,2,2
+spark.query.origin.partitionKey                   key
+spark.query.target.id                             key
+
+spark.counterTable.cql.index                      1,2
+
+spark.target.autocorrect.missing                  true
+spark.target.autocorrect.mismatch                 true
diff --git a/SIT/smoke/04_counters/migrate.properties b/SIT/smoke/04_counters/migrate.properties
@@ -0,0 +1,20 @@
+spark.origin.host                                 cdm-sit-cass
+spark.origin.port                                 9042
+spark.origin.username                             cassandra
+spark.origin.password                             cassandra
+spark.origin.keyspaceTable                        source.smoke_counters
+
+spark.target.host                                 cdm-sit-cass
+spark.target.port                                 9042
+spark.target.username                             cassandra
+spark.target.password                             cassandra
+spark.target.keyspaceTable                        target.smoke_counters
+
+spark.numSplits                                   1
+
+spark.query.origin                                key,col1,col2
+spark.query.types                                 0,2,2
+spark.query.origin.partitionKey                   key
+spark.query.target.id                             key
+
+spark.counterTable.cql.index                      1,2
diff --git a/SIT/smoke/04_counters/setup.cql b/SIT/smoke/04_counters/setup.cql
@@ -0,0 +1,13 @@
+DROP TABLE IF EXISTS source.smoke_counters;
+CREATE TABLE source.smoke_counters(key text,col1 counter,col2 counter, PRIMARY KEY (key));
+
+UPDATE source.smoke_counters set col1=col1+1, col2=col2+2 where key='record1';
+UPDATE source.smoke_counters set col1=col1+10, col2=col2+20 where key='record2';
+UPDATE source.smoke_counters set col1=col1+1000, col2=col2+2000 where key='record3';
+UPDATE source.smoke_counters set col1=col1+10000, col2=col2+20000 where key='record4';
+UPDATE source.smoke_counters set col1=col1+500, col2=col2+500 where key='record5';
+
+DROP TABLE IF EXISTS target.smoke_counters;
+CREATE TABLE target.smoke_counters(key text,col1 counter,col2 counter, PRIMARY KEY (key));
+
+SELECT * FROM source.smoke_counters;
diff --git a/src/main/java/datastax/astra/migrate/cql/CqlHelper.java b/src/main/java/datastax/astra/migrate/cql/CqlHelper.java
@@ -164,7 +164,8 @@ public boolean hasRandomPartitioner() {
     }
 
     public boolean isCounterTable() {
-        return propertyHelper.getBoolean(KnownProperties.ORIGIN_IS_COUNTER);
+        List<Integer> rtn = propertyHelper.getIntegerList(KnownProperties.ORIGIN_COUNTER_INDEXES);
+        return (null != rtn && rtn.size() > 0);
     }
 
     //--------------- TTL & Writetime Feature ---------------
diff --git a/src/test/java/datastax/astra/migrate/cql/features/CounterTest.java b/src/test/java/datastax/astra/migrate/cql/features/CounterTest.java
@@ -0,0 +1,79 @@
+package datastax.astra.migrate.cql.features;
+
+import datastax.astra.migrate.cql.CqlHelper;
+import datastax.astra.migrate.properties.KnownProperties;
+import datastax.astra.migrate.properties.PropertyHelper;
+import org.apache.spark.SparkConf;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class CounterTest {
+
+    PropertyHelper helper;
+    SparkConf validSparkConf;
+
+    @BeforeEach
+    public void setup() {
+        helper = PropertyHelper.getInstance();
+        validSparkConf = new SparkConf();
+    }
+
+    @AfterEach
+    public void tearDown() {
+        PropertyHelper.destroyInstance();
+        validSparkConf = null;
+    }
+
+    private void setValidSparkConf() {
+        validSparkConf.set(KnownProperties.ORIGIN_COLUMN_NAMES, "key,col1,col2");
+        validSparkConf.set(KnownProperties.ORIGIN_COLUMN_TYPES, "4,2,2");
+        validSparkConf.set(KnownProperties.ORIGIN_COUNTER_INDEXES, "1,2");
+
+        validSparkConf.set(KnownProperties.TARGET_PRIMARY_KEY, "key");
+    }
+
+    @Test
+    public void smokeTest_initialize() {
+        setValidSparkConf();
+        helper.initializeSparkConf(validSparkConf);
+        assertAll(
+                () -> assertEquals(Arrays.asList(1,2), helper.getIntegerList(KnownProperties.ORIGIN_COUNTER_INDEXES), "ORIGIN_COUNTER_INDEXES")
+        );
+    }
+
+    @Test
+    public void smokeCQL() {
+        SparkConf sparkConf = new SparkConf();
+        sparkConf.set(KnownProperties.ORIGIN_CONNECT_HOST, "localhost");
+        sparkConf.set(KnownProperties.ORIGIN_KEYSPACE_TABLE, "origin.tab1");
+        sparkConf.set(KnownProperties.ORIGIN_COLUMN_NAMES, "key,col1,col2");
+        sparkConf.set(KnownProperties.ORIGIN_COLUMN_TYPES, "4,2,2");
+        sparkConf.set(KnownProperties.ORIGIN_PARTITION_KEY, "key");
+        sparkConf.set(KnownProperties.ORIGIN_COUNTER_INDEXES, "1,2");
+
+        sparkConf.set(KnownProperties.TARGET_PRIMARY_KEY, "key");
+        sparkConf.set(KnownProperties.TARGET_KEYSPACE_TABLE, "target.tab1");
+
+        helper.initializeSparkConf(sparkConf);
+        CqlHelper cqlHelper = new CqlHelper();
+        cqlHelper.initialize();
+
+        String originSelect = "SELECT key,col1,col2 FROM origin.tab1 WHERE TOKEN(key) >= ? AND TOKEN(key) <= ? ALLOW FILTERING";
+        String originSelectByPK = "SELECT key,col1,col2 FROM origin.tab1 WHERE key=?";
+        String targetUpdate = "UPDATE target.tab1 SET col1=col1+?,col2=col2+? WHERE key=?";
+        String targetSelect = "SELECT key,col1,col2 FROM target.tab1 WHERE key=?";
+
+        assertAll(
+                () -> assertEquals(originSelect, cqlHelper.getOriginSelectByPartitionRangeStatement().getCQL().replaceAll("\\s+"," ")),
+                () -> assertEquals(originSelectByPK, cqlHelper.getOriginSelectByPKStatement().getCQL().replaceAll("\\s+"," ")),
+                () -> assertEquals(targetUpdate, cqlHelper.getTargetUpdateStatement().getCQL().replaceAll("\\s+"," ")),
+                () -> assertEquals(targetSelect, cqlHelper.getTargetSelectByPKStatement().getCQL().replaceAll("\\s+"," "))
+        );
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+migrateData datastax.astra.migrate.Migrate migrate.properties`
	`2`	`+validateData datastax.astra.migrate.DiffData migrate.properties`
	`3`	`+fixData datastax.astra.migrate.DiffData fix.properties`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+SELECT key,col1,col2 FROM target.smoke_counters;`