Added filtering code to CopyJob. Takes a flag then proceeds to find column and value from config file and filter data from origin to target.

Author: mfmaher2

pom.xml

@@ -3,7 +3,7 @@
 
   <groupId>datastax.astra.migrate</groupId>
   <artifactId>cassandra-data-migrator</artifactId>
-  <version>2.6</version>
+  <version>2.7</version>
   <packaging>jar</packaging>
 
   <properties>

src/main/java/datastax/astra/migrate/BaseJobSession.java

@@ -48,7 +48,25 @@ public abstract class BaseJobSession {
     protected String astraKeyspaceTable;
 
     protected Boolean hasRandomPartitioner;
+    protected Boolean filterData;
+    protected String filterColName;
+    protected String filterColType;
+    protected Integer filterColIndex;
+    protected String filterColValue;
 
+    public String getKey(Row sourceRow) {
+        StringBuffer key = new StringBuffer();
+        for (int index = 0; index < idColTypes.size(); index++) {
+            MigrateDataType dataType = idColTypes.get(index);
+            if (index == 0) {
+                key.append(getData(dataType, index, sourceRow));
+            } else {
+                key.append(" %% " + getData(dataType, index, sourceRow));
+            }
+        }
+
+        return key.toString();
+    }
     public List<MigrateDataType> getTypes(String types) {
         List<MigrateDataType> dataTypes = new ArrayList<MigrateDataType>();
         for (String type : types.split(",")) {

src/main/java/datastax/astra/migrate/CopyJobSession.java

@@ -22,6 +22,11 @@ public class CopyJobSession extends AbstractJobSession {
 
     protected CopyJobSession(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
         super(sourceSession, astraSession, sc);
+        filterData = Boolean.parseBoolean(sc.get("spark.origin.FilterData", "false"));
+        filterColName = sc.get("spark.origin.FilterColumn");
+        filterColType = sc.get("spark.origin.FilterColumnType");
+        filterColIndex = Integer.parseInt(sc.get("spark.origin.FilterColumnIndex", "0"));
+        filterColValue = sc.get("spark.origin.FilterColumnValue");
     }
 
     public static CopyJobSession getInstance(CqlSession sourceSession, CqlSession astraSession, SparkConf sc) {
@@ -52,6 +57,13 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                     for (Row sourceRow : resultSet) {
                         readLimiter.acquire(1);
 
+                        if (filterData) {
+                            String col = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
+                            if (col.trim().equalsIgnoreCase(filterColValue)) {
+                                logger.warn("Row larger than 10 MB found filtering out: " + getKey(sourceRow));
+                                continue;
+                            }
+                        }
                         if (writeTimeStampFilter) {
                             // only process rows greater than writeTimeStampFilter
                             Long sourceWriteTimeStamp = getLargestWriteTimeStamp(sourceRow);
@@ -92,6 +104,15 @@ public void getDataAndInsert(BigInteger min, BigInteger max) {
                         if (readCounter.incrementAndGet() % printStatsAfter == 0) {
                             printCounts(false);
                         }
+
+                        if (filterData) {
+                            String colValue = (String) getData(new MigrateDataType(filterColType), filterColIndex, sourceRow);
+                            if (colValue.trim().equalsIgnoreCase(filterColValue)) {
+                                logger.warn("Row larger than 10 MB found filtering out: " + getKey(sourceRow));
+                                continue;
+                            }
+                        }
+
                         batchStatement = batchStatement.add(bindInsert(astraInsertStatement, sourceRow, null));
 
                         // if batch threshold is met, send the writes and clear the batch

src/main/java/datastax/astra/migrate/DiffJobSession.java

@@ -186,18 +186,4 @@ private String isDifferent(Row sourceRow, Row astraRow) {
         return diffData.toString();
     }
 
-    private String getKey(Row sourceRow) {
-        StringBuffer key = new StringBuffer();
-        for (int index = 0; index < idColTypes.size(); index++) {
-            MigrateDataType dataType = idColTypes.get(index);
-            if (index == 0) {
-                key.append(getData(dataType, index, sourceRow));
-            } else {
-                key.append(" %% " + getData(dataType, index, sourceRow));
-            }
-        }
-
-        return key.toString();
-    }
-
 }

src/main/java/datastax/astra/migrate/OriginCountJobSession.java

@@ -23,9 +23,6 @@ public class OriginCountJobSession extends BaseJobSession {
     protected List<Integer> updateSelectMapping = new ArrayList<Integer>();
     protected Boolean checkTableforColSize;
     protected String checkTableforselectCols;
-    protected String filterColName;
-    protected String filterColType;
-    protected Integer filterColIndex;
     protected Integer fieldGuardraillimitMB;
     protected List<MigrateDataType> checkTableforColSizeTypes = new ArrayList<MigrateDataType>();
 

src/resources/sparkConf.properties

@@ -7,9 +7,11 @@ spark.origin.keyspaceTable                        test.a1
 spark.origin.checkTableforColSize                 false
 spark.origin.checkTableforColSize.cols            partition-key,clustering-key
 spark.origin.checkTableforColSize.cols.types      9,1
+spark.origin.FilterData                           true
 spark.origin.FilterColumn                         test
 spark.origin.FilterColumnIndex                    2
 spark.origin.FilterColumnType                     6%16
+spark.origin.FilterColumnValue                    test
 
 spark.target.isAstra                              true
 spark.target.scb                                  file:///aaa/bbb/secure-connect-enterprise.zip