Cassandra performance: Replace sequence ids with time-based UUIDs

Makes the schema changes in a separate migration step, so that data in the repair_unit and repair_schedule tables can be migrated over. ref: - #99 - #94 - #99 (comment)
thelastpickle · Jun 23, 2017 · 2ef929c · 2ef929c
1 parent 07887ee
commit 2ef929c
Show file tree

Hide file tree

Showing 7 changed files with 22 additions and 25 deletions.
diff --git a/src/main/java/com/spotify/reaper/core/RepairSchedule.java b/src/main/java/com/spotify/reaper/core/RepairSchedule.java
@@ -16,6 +16,7 @@
 import com.google.common.collect.ImmutableList;
 import com.spotify.reaper.core.RepairSegment.State;
 import com.spotify.reaper.storage.postgresql.LongCollectionSQLType;
+import java.util.UUID;
 
 import java.util.List;
 import java.util.UUID;

diff --git a/src/main/java/com/spotify/reaper/core/RepairSegment.java b/src/main/java/com/spotify/reaper/core/RepairSegment.java
@@ -106,7 +106,6 @@ public enum State {
 
   public static class Builder {
 
-
     public final RingRange tokenRange;
     private final UUID repairUnitId;
     private UUID runId;
@@ -117,7 +116,8 @@ public static class Builder {
     private DateTime startTime;
     private DateTime endTime;
 
-    public Builder(RingRange tokenRange, UUID repairUnitId) {
+    public Builder(UUID runId, RingRange tokenRange, UUID repairUnitId) {
+      this.runId = runId;
       this.repairUnitId = repairUnitId;
       this.tokenRange = tokenRange;
       this.failCount = 0;

diff --git a/src/main/java/com/spotify/reaper/storage/CassandraStorage.java b/src/main/java/com/spotify/reaper/storage/CassandraStorage.java
@@ -295,7 +295,6 @@ private Collection<RepairRun> getRepairRunsAsync(List<ResultSetFuture> repairRun
 
   @Override
   public Collection<RepairRun> getRepairRunsWithState(RunState runState) {
-
     return getClusters().stream()
             // Grab all ids for the given cluster name
             .map(cluster -> getRepairRunIdsForCluster(cluster.getName()))
@@ -409,13 +408,14 @@ private boolean segmentIsWithinRange(RepairSegment segment, RingRange range) {
 
   }
 
-  private static RepairSegment createRepairSegmentFromRow(Row segmentRow){
+  private RepairSegment createRepairSegmentFromRow(Row segmentRow){
+    return createRepairSegmentFromRow(segmentRow, segmentRow.getUUID("id"));
+  }
+  private RepairSegment createRepairSegmentFromRow(Row segmentRow, UUID segmentId){
     return new RepairSegment.Builder(
-            new RingRange(
-                    new BigInteger(segmentRow.getVarint("start_token") +""),
-                    new BigInteger(segmentRow.getVarint("end_token")+"")),
+            segmentRow.getUUID("run_id"),
+            new RingRange(new BigInteger(segmentRow.getVarint("start_token") +""), new BigInteger(segmentRow.getVarint("end_token")+"")),
             segmentRow.getUUID("repair_unit_id"))
-        .withRunId(segmentRow.getUUID("id"))
         .coordinatorHost(segmentRow.getString("coordinator_host"))
         .endTime(new DateTime(segmentRow.getTimestamp("segment_end_time")))
         .failCount(segmentRow.getInt("fail_count"))
@@ -675,12 +675,10 @@ public Collection<RepairRunStatus> getClusterRunStatuses(String clusterName, int
   @Override
   public Collection<RepairScheduleStatus> getClusterScheduleStatuses(String clusterName) {
     Collection<RepairSchedule> repairSchedules = getRepairSchedulesForCluster(clusterName);
-
     Collection<RepairScheduleStatus> repairScheduleStatuses = repairSchedules
             .stream()
             .map(sched -> new RepairScheduleStatus(sched, getRepairUnit(sched.getRepairUnitId()).get()))
             .collect(Collectors.toList());
-
     return repairScheduleStatuses;
   }
 

diff --git a/src/main/java/com/spotify/reaper/storage/IStorage.java b/src/main/java/com/spotify/reaper/storage/IStorage.java
@@ -91,7 +91,6 @@ public interface IStorage {
   Optional<RepairUnit> getRepairUnit(String cluster, String keyspace,
       Set<String> columnFamilyNames);
 
-
   boolean updateRepairSegment(RepairSegment newRepairSegment);
 
   Optional<RepairSegment> getRepairSegment(UUID runId, UUID segmentId);

diff --git a/src/main/java/com/spotify/reaper/storage/postgresql/LongCollectionSQLType.java b/src/main/java/com/spotify/reaper/storage/postgresql/LongCollectionSQLType.java
@@ -10,13 +10,13 @@
  */
 public class LongCollectionSQLType {
 
-  private Collection<Long> collection;
+  private Collection<UUID> collection;
 
-  public LongCollectionSQLType(Collection<Long> collection) {
+  public LongCollectionSQLType(Collection<UUID> collection) {
     this.collection = collection;
   }
 
-  public Collection<Long> getValue() {
+  public Collection<UUID> getValue() {
     if (this.collection == null) {
       return Lists.newArrayList();
     } else {

diff --git a/src/main/java/com/spotify/reaper/storage/postgresql/RepairSegmentMapper.java b/src/main/java/com/spotify/reaper/storage/postgresql/RepairSegmentMapper.java
@@ -29,15 +29,15 @@ public class RepairSegmentMapper implements ResultSetMapper<RepairSegment> {
   public RepairSegment map(int index, ResultSet r, StatementContext ctx) throws SQLException {
     RingRange range = new RingRange(r.getBigDecimal("start_token").toBigInteger(),
                                     r.getBigDecimal("end_token").toBigInteger());
-    return
-        new RepairSegment.Builder(range, fromSequenceId(r.getLong("repair_unit_id")))
-                .withRunId(fromSequenceId(r.getLong("run_id")))
-                .state(RepairSegment.State.values()[r.getInt("state")])
-                .coordinatorHost(r.getString("coordinator_host"))
-                .startTime(RepairRunMapper.getDateTimeOrNull(r, "start_time"))
-                .endTime(RepairRunMapper.getDateTimeOrNull(r, "end_time"))
-                .failCount(r.getInt("fail_count"))
-                .build(fromSequenceId(r.getLong("id")));
+    RepairSegment.Builder repairSegmentBuilder =
+        new RepairSegment.Builder(fromSequenceId(r.getLong("run_id")), range, fromSequenceId(r.getLong("repair_unit_id")));
+    return repairSegmentBuilder
+        .state(RepairSegment.State.values()[r.getInt("state")])
+        .coordinatorHost(r.getString("coordinator_host"))
+        .startTime(RepairRunMapper.getDateTimeOrNull(r, "start_time"))
+        .endTime(RepairRunMapper.getDateTimeOrNull(r, "end_time"))
+        .failCount(r.getInt("fail_count"))
+        .build(fromSequenceId(r.getLong("id")));
   }
 
   private static UUID fromSequenceId(long insertedId) {

diff --git a/src/main/resources/db/cassandra/003_switch_to_uuids.cql b/src/main/resources/db/cassandra/003_switch_to_uuids.cql
@@ -76,7 +76,6 @@ CREATE TABLE IF NOT EXISTS repair_run_by_unit (
 
 
 
-
 CREATE TABLE IF NOT EXISTS repair_schedule_v1 (
   id                 timeuuid PRIMARY KEY,
   repair_unit_id     timeuuid,
@@ -109,4 +108,4 @@ CREATE TABLE IF NOT EXISTS repair_schedule_by_cluster_and_keyspace (
 
 DROP TABLE IF EXISTS repair_segment;
 DROP TABLE IF EXISTS repair_segment_by_run_id;
-DROP TABLE IF EXISTS repair_id;
+DROP TABLE IF EXISTS repair_id;