Merge pull request #119 from datastax/issue/CDM-28

mieslep · web-flow · commit 750d8223fdc3 · 2023-04-06T16:31:46.000+01:00
Issue/cdm 28
diff --git a/SIT/smoke/04_counters/breakData.cql b/SIT/smoke/04_counters/breakData.cql
@@ -0,0 +1,19 @@
+
+UPDATE source.smoke_counters SET col1=col1+1 WHERE key='record2';
+
+-- CDM-29 discovered that this record will "zombie"...
+-- fixData will not correct it, fixForce will
+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: 0
+Valid Record Count: 2
+Skipped Record Count: 0
diff --git a/SIT/smoke/04_counters/cdm.fixForce.assert b/SIT/smoke/04_counters/cdm.fixForce.assert
@@ -0,0 +1,7 @@
+Read Record Count: 5
+Mismatch Record Count: 0
+Corrected Mismatch Record Count: 0
+Missing Record Count: 1
+Corrected Missing Record Count: 1
+Valid Record Count: 4
+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,4 @@
+migrateData datastax.astra.migrate.Migrate migrate.properties
+validateData datastax.astra.migrate.DiffData migrate.properties
+fixData datastax.astra.migrate.DiffData fix.properties
+fixForce datastax.astra.migrate.DiffData fix-force.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,18 @@
+#!/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"
+
+/local/cdm.sh -f cdm.txt -s fixForce -d "$workingDir" > cdm.fixForce.out 2>cdm.fixForce.err
+/local/cdm-assert.sh -f cdm.fixForce.out -a cdm.fixForce.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-force.properties b/SIT/smoke/04_counters/fix-force.properties
@@ -0,0 +1,24 @@
+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.counterTable.missing.force                  true
+
+spark.target.autocorrect.missing                  true
+spark.target.autocorrect.mismatch                 true
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/DiffJobSession.java b/src/main/java/datastax/astra/migrate/DiffJobSession.java
@@ -41,6 +41,7 @@ public class DiffJobSession extends CopyJobSession {
     private final AtomicLong skippedCounter = new AtomicLong(0);
 
     private final boolean isCounterTable;
+    private final boolean forceCounterWhenMissing;
     private final List<Integer> targetToOriginColumnIndexes;
     private final List<MigrateDataType> targetColumnTypes;
     private final int explodeMapKeyIndex;
@@ -56,6 +57,7 @@ private DiffJobSession(CqlSession originSession, CqlSession targetSession, Spark
         logger.info("PARAM -- Autocorrect Mismatch: {}", autoCorrectMismatch);
 
         this.isCounterTable = cqlHelper.isCounterTable();
+        this.forceCounterWhenMissing = propertyHelper.getBoolean(KnownProperties.ORIGIN_COUNTER_FORCE_WHEN_MISSING);
         this.targetToOriginColumnIndexes = cqlHelper.getPKFactory().getTargetToOriginColumnIndexes();
         this.targetColumnTypes = cqlHelper.getPKFactory().getTargetColumnTypes();
 
@@ -167,6 +169,10 @@ private void diff(Record record) {
         if (targetRow == null) {
             missingCounter.incrementAndGet();
             logger.error("Missing target row found for key: {}", record.getPk());
+            if (autoCorrectMissing && isCounterTable && !forceCounterWhenMissing) {
+                logger.error("{} is true, but not Inserting as {} is not enabled; key : {}", KnownProperties.TARGET_AUTOCORRECT_MISSING, KnownProperties.ORIGIN_COUNTER_FORCE_WHEN_MISSING, record.getPk());
+                return;
+            }
 
             //correct data
             if (autoCorrectMissing) {
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/main/java/datastax/astra/migrate/properties/KnownProperties.java b/src/main/java/datastax/astra/migrate/properties/KnownProperties.java
@@ -71,6 +71,7 @@ public enum PropertyType {
     public static final String ORIGIN_IS_COUNTER      = "spark.counterTable";              // false
     public static final String ORIGIN_COUNTER_CQL     = "spark.counterTable.cql";
     public static final String ORIGIN_COUNTER_INDEXES = "spark.counterTable.cql.index";    // 0
+    public static final String ORIGIN_COUNTER_FORCE_WHEN_MISSING = "spark.counterTable.missing.force";    // false
     public static final String ORIGIN_PRIMARY_KEY        = "spark.query.origin.id";         // Defaults to TARGET_PRIMARY_KEY, same config rules
     public static final String ORIGIN_PRIMARY_KEY_TYPES  = "spark.cdm.cql.origin.id.types"; // Code-managed, not an external property
 
@@ -89,6 +90,8 @@ public enum PropertyType {
         defaults.put(ORIGIN_IS_COUNTER, "false");
            types.put(ORIGIN_COUNTER_CQL, PropertyType.STRING);
            types.put(ORIGIN_COUNTER_INDEXES, PropertyType.NUMBER_LIST);
+           types.put(ORIGIN_COUNTER_FORCE_WHEN_MISSING, PropertyType.BOOLEAN);
+        defaults.put(ORIGIN_COUNTER_FORCE_WHEN_MISSING, "false");
            types.put(ORIGIN_PRIMARY_KEY, PropertyType.STRING_LIST);
         required.add(ORIGIN_PRIMARY_KEY);
            types.put(ORIGIN_PRIMARY_KEY_TYPES, PropertyType.MIGRATION_TYPE_LIST);
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 @@`
	`1`	`+SELECT key,col1,col2 FROM target.smoke_counters;`