Merge pull request #14 from Ankitp1342/feature/diff_only

Add new flag to DiffData to only show the diff and not insert into de…

Author: Ankitp1342

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

@@ -36,6 +36,7 @@ public class DiffJobSession extends CopyJobSession {
     private AtomicLong skippedCounter = new AtomicLong(0);
 
     protected List<MigrateDataType> selectColTypes = new ArrayList<MigrateDataType>();
+    protected Boolean isDiffOnly = false;
 
 
     public static DiffJobSession getInstance(CqlSession sourceSession, CqlSession astraSession, SparkConf sparkConf) {
@@ -54,6 +55,7 @@ private DiffJobSession(CqlSession sourceSession, CqlSession astraSession, SparkC
         super(sourceSession, astraSession, sparkConf);
 
         selectColTypes = getTypes(sparkConf.get("spark.migrate.diff.select.types"));
+        isDiffOnly = Boolean.parseBoolean(sparkConf.get("spark.migrate.isDiffOnly", "false"));
     }
 
     public void getDataAndDiff(BigInteger min, BigInteger max) {
@@ -120,9 +122,12 @@ private void diff(Row sourceRow, Row astraRow) {
             missingCounter.incrementAndGet();
             logger.error("Data is missing in Astra: " + getKey(sourceRow));
             //correct data
-            astraSession.execute(bindInsert(astraInsertStatement, sourceRow));
-            correctedMissingCounter.incrementAndGet();
-            logger.error("Corrected missing data in Astra: " + getKey(sourceRow));
+
+            if (!isDiffOnly) {
+                astraSession.execute(bindInsert(astraInsertStatement, sourceRow));
+                correctedMissingCounter.incrementAndGet();
+                logger.error("Corrected missing data in Astra: " + getKey(sourceRow));
+            }
             return;
         }
 

src/resources/autoSparkConfig.py

@@ -55,6 +55,12 @@
   'inet':     '0'
 
 }
+system_keyspace = ['OpsCenter','dse_insights_local','solr_admin','test','dse_system','dse_analytics','system_auth','system_traces','system','dse_system_local','system_distributed','system_schema','dse_perf','dse_insights','dse_security','dse_system','killrvideo','dse_leases','dsefs_c4z','HiveMetaStore','dse_analytics','dsefs','spark_system']
+tp_tbl_data = {
+    'Materialized Views':{},
+    'Secondary Indexes':{},
+    'Storage-Attached Indexes':{}
+}
 
 def field_type_comment(tbl,fieldName,fieldType):
   if fieldType=='date':
@@ -99,6 +105,18 @@ def field_type_comment(tbl,fieldName,fieldType):
 cfg_array = {}
 schema_name = 'schema'
 
+def add_tp_tbl(gr,ks,tbl,src_ks,src_tbl):
+  if src_ks not in system_keyspace:
+    try:
+      type(tp_tbl_data[gr][src_ks])
+    except:
+      tp_tbl_data[gr][src_ks]={}
+    try:
+      type(tp_tbl_data[gr][src_ks][src_tbl])
+    except:
+      tp_tbl_data[gr][src_ks][src_tbl] = []
+    if (ks+'.'+tbl) not in tp_tbl_data[gr][src_ks][src_tbl]:
+      tp_tbl_data[gr][src_ks][src_tbl].append(ks+'.'+tbl)
 
 def process_field(tbl,fieldName,fieldType,cql=''):
   if 'map<' in fieldType:
@@ -113,6 +131,12 @@ def process_field(tbl,fieldName,fieldType,cql=''):
     for mapType in mapData:
       fieldValue += '%' + field_type_array[mapType.strip()]
       field_type_comment(tbl,fieldName,mapType.strip())
+  elif 'list<' in fieldType:
+    mapData = fieldType.split('<')[1].split('>')[0].split(',')
+    fieldValue = field_type_array['list']
+    for mapType in mapData:
+      fieldValue += '%' + field_type_array[mapType.strip()]
+      field_type_comment(tbl,fieldName,mapType.strip())
   elif fieldType in field_types:
     fieldValue = field_type_array[fieldType]
   else: