CASSANALYTICS-197: Source token ranges from Cassandra for mutation tracked keyspaces

CHANGES.txt

@@ -1,5 +1,8 @@
 0.5.0
 -----
+ * Source token ranges from Cassandra for mutation tracked keyspaces (CASSANALYTICS-197)
+ * Parse <replicas>/<transient> replication factor for witness-enabled keyspaces (CASSANALYTICS-194)
+ * Determine whether mutation tracking is enabled for keyspace for bulk writes (CASSANALYTICS-160)
  * Upgrade sidecar version to 0.4.0
  * Exclude IP address from RingInstance equality so node replacement does not fail bulk write jobs (CASSANALYTICS-175)
  * Regenerate bloom filters for CQLSSTableWriter (CASSANALYTICS-167)

cassandra-analytics-common/src/main/java/org/apache/cassandra/spark/utils/CqlUtils.java

@@ -52,6 +52,9 @@ public final class CqlUtils
                                                                                         "max_index_interval"
                                                                                         );
     private static final Pattern REPLICATION_FACTOR_PATTERN = Pattern.compile("WITH REPLICATION = (\\{[^\\}]*\\})");
+    private static final String TRACKED_REPLICATION_TYPE = "tracked";
+    private static final Pattern REPLICATION_TYPE_PATTERN = Pattern.compile("replication_type\\s*=\\s*'(\\w+)'",
+                                                                             Pattern.CASE_INSENSITIVE);
     // Initialize a mapper allowing single quotes to process the RF string from the CREATE KEYSPACE statement
     private static final ObjectMapper MAPPER = new ObjectMapper().configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
     private static final Pattern ESCAPED_WHITESPACE_PATTERN = Pattern.compile("(\\\\r|\\\\n|\\\\r\\n)+");
@@ -173,9 +176,18 @@ public static ReplicationFactor extractReplicationFactor(@NotNull String schemaS
             throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace), exception);
         }
 
-        String className = map.remove("class");
-        ReplicationFactor.ReplicationStrategy strategy = ReplicationFactor.ReplicationStrategy.getEnum(className);
-        return new ReplicationFactor(strategy, map.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, v -> Integer.parseInt(v.getValue()))));
+        // Values may use the <replicas>/<transient> form for witness replicas, so delegate parsing to
+        // ReplicationFactor. parseStrict reports an unparseable value directly instead of dropping the
+        // datacenter, which would otherwise surface later as a confusing "DC not found" error.
+        try
+        {
+            return ReplicationFactor.parseStrict(map);
+        }
+        catch (IllegalArgumentException exception)
+        {
+            throw new RuntimeException(String.format("Unable to parse replication factor for keyspace: %s", keyspace),
+                                       exception);
+        }
     }
 
     public static String extractTableSchema(@NotNull String schemaStr, @NotNull String keyspace, @NotNull String table)
@@ -296,4 +308,33 @@ public static boolean isTimeRangeFilterSupported(String compactionStrategy)
     {
         return compactionStrategy == null || compactionStrategy.endsWith("TimeWindowCompactionStrategy");
     }
+
+    /**
+     * Extracts replication type from create schema statement
+     *
+     * @param schemaStr full cluster schema string as returned by Sidecar
+     * @param keyspace  name of the keyspace to check
+     * @return {@code true} if keyspace is tracked {@code false} otherwise
+     */
+    public static String extractReplicationType(@NotNull String schemaStr, @NotNull String keyspace)
+    {
+        String createKeyspaceSchema = extractKeyspaceSchema(schemaStr, keyspace);
+        Matcher matcher = REPLICATION_TYPE_PATTERN.matcher(createKeyspaceSchema);
+        if (matcher.find())
+        {
+            return matcher.group(1);
+        }
+        return null;
+    }
+
+    /**
+     * Returns {@code true} if {@code replication_type = 'tracked'} in create statement otherwise {@code false}
+     *
+     * @param replicationType replication type extracted from create statement
+     * @return {@code true} if replication type is tracked {@code false} otherwise
+     */
+    public static boolean isTracked(String replicationType)
+    {
+        return TRACKED_REPLICATION_TYPE.equalsIgnoreCase(replicationType);
+    }
 }