Migration service + double writing (#120)

Delta block writing + service to migrate existing delta columns

Author: bdevore17

common/dropwizard/src/main/java/com/bazaarvoice/emodb/common/dropwizard/service/EmoServiceMode.java

@@ -22,6 +22,8 @@
  *      A typical use case for this is running command-line tools, such as reports.</li>
  * <li>SCANNER:
  *      Only services, tasks, and resources required to support the scan and upload server.</li>
+ * <li>SCANNER:
+ *      Only services, tasks, and resources required to migrate all delta from old tables to new "blocked" tables.</li>
  * </ul>
  */
 public enum EmoServiceMode {
@@ -95,6 +97,18 @@ public enum EmoServiceMode {
             Aspect.scanner,
             Aspect.security,
             Aspect.full_consistency
+    ),
+
+    DELTA_MIGRATOR(
+            Aspect.web,
+            Aspect.cache,
+            Aspect.leader_control,
+            Aspect.dataCenter,
+            Aspect.dataStore_module,
+            Aspect.blobStore_module, // needed for permission resolver
+            Aspect.delta_migrator,
+            Aspect.security,
+            Aspect.full_consistency
     );
 
     private final EnumSet<Aspect> aspects;
@@ -156,6 +170,7 @@ public enum Aspect {
         security,
         invalidation_cache_listener, // This makes sure the node is registered in zookeeper to invalidate its caches
         scanner(false),
+        delta_migrator(false),
         swagger,
         uac;
 

sdk/src/main/resources/emodb-default-config-ddl.yaml

@@ -7,6 +7,7 @@ systemOfRecord:
                keyspace: ugc_global
                table.audit: ugc_audit
                table.delta: ugc_delta
+               table.delta_v2: ugc_delta_v2
                table.history: ugc_history
       catalog_global:
          replicationFactor: 1
@@ -15,6 +16,7 @@ systemOfRecord:
                keyspace: catalog_global
                table.audit: cat_audit
                table.delta: cat_delta
+               table.delta_v2: cat_delta_v2
                table.history: cat_history
       app_global:
          replicationFactor: 1
@@ -23,11 +25,13 @@ systemOfRecord:
                keyspace: app_global
                table.audit: default_audit
                table.delta: default_delta
+               table.delta_v2: default_delta_v2
                table.history: default_history
             sys:
                keyspace: app_global
                table.audit: sys_audit
                table.delta: sys_delta
+               table.delta_v2: sys_delta_v2
                table.history: sys_history
 
 databus:

sor/src/main/java/com/bazaarvoice/emodb/sor/DataStoreConfiguration.java

@@ -62,6 +62,20 @@ public class DataStoreConfiguration {
     @JsonProperty("stashRoot")
     private Optional<String> _stashRoot = Optional.absent();
 
+    @Valid
+    @NotNull
+    @JsonProperty("migrationPhase")
+    private DeltaMigrationPhase _migrationPhase = DeltaMigrationPhase.PRE_MIGRATION;
+
+
+    /*
+    Only temporarily configurable during the migration period
+    */
+    @Valid
+    @NotNull
+    @JsonProperty
+    private int _deltaBlockSizeInKb = 64;
+
     public String getSystemTablePlacement() {
         return _systemTablePlacement;
     }
@@ -155,4 +169,12 @@ public DataStoreConfiguration setDeltaEncodingVersion(int deltaEncodingVersion)
         _deltaEncodingVersion = deltaEncodingVersion;
         return this;
     }
+
+    public DeltaMigrationPhase getMigrationPhase() {
+        return _migrationPhase;
+    }
+
+    public int getDeltaBlockSizeInKb() {
+        return _deltaBlockSizeInKb;
+    }
 }

sor/src/main/java/com/bazaarvoice/emodb/sor/DataStoreModule.java

@@ -24,16 +24,7 @@
 import com.bazaarvoice.emodb.datacenter.api.KeyspaceDiscovery;
 import com.bazaarvoice.emodb.sor.admin.RowKeyTask;
 import com.bazaarvoice.emodb.sor.api.DataStore;
-import com.bazaarvoice.emodb.sor.core.AuditStore;
-import com.bazaarvoice.emodb.sor.core.DataProvider;
-import com.bazaarvoice.emodb.sor.core.DataStoreProviderProxy;
-import com.bazaarvoice.emodb.sor.core.DataTools;
-import com.bazaarvoice.emodb.sor.core.DefaultAuditStore;
-import com.bazaarvoice.emodb.sor.core.DefaultDataStore;
-import com.bazaarvoice.emodb.sor.core.DeltaHistoryTtl;
-import com.bazaarvoice.emodb.sor.core.LocalDataStore;
-import com.bazaarvoice.emodb.sor.core.StashRoot;
-import com.bazaarvoice.emodb.sor.core.SystemDataStore;
+import com.bazaarvoice.emodb.sor.core.*;
 import com.bazaarvoice.emodb.sor.db.astyanax.DAOModule;
 import com.bazaarvoice.emodb.sor.db.astyanax.DeltaPlacementFactory;
 import com.bazaarvoice.emodb.sor.db.cql.CqlForMultiGets;
@@ -237,6 +228,9 @@ protected void configure() {
         // Data tools used to generate reports
         bind(DataTools.class).to(DefaultDataStore.class);
         expose(DataTools.class);
+
+        bind(MigratorTools.class).to(DefaultMigratorTools.class);
+        expose(MigratorTools.class);
     }
 
     @Provides @Singleton

sor/src/main/java/com/bazaarvoice/emodb/sor/DeltaMigrationPhase.java

@@ -0,0 +1,30 @@
+package com.bazaarvoice.emodb.sor;
+
+public enum DeltaMigrationPhase {
+    PRE_MIGRATION(true, true, false),
+    DOUBLE_WRITE_LEGACY_READ(true, true, true),
+    DOUBLE_WRITE_BLOCKED_READ(false, true, true),
+    FULLY_MIGRATED(false, false, true);
+
+    private final boolean _readFromLegacyDeltaTables;
+    private final boolean _writeToLegacyDeltaTables;
+    private final boolean _writeToBlockedDeltaTables;
+
+    DeltaMigrationPhase(boolean readFromLegacyDeltaTables, boolean writeToLegacyDeltaTables, boolean writeToBlockedDeltaTables) {
+        _readFromLegacyDeltaTables = readFromLegacyDeltaTables;
+        _writeToLegacyDeltaTables = writeToLegacyDeltaTables;
+        _writeToBlockedDeltaTables = writeToBlockedDeltaTables;
+    }
+
+    public boolean isReadFromLegacyDeltaTables() {
+        return _readFromLegacyDeltaTables;
+    }
+
+    public boolean isWriteToLegacyDeltaTables() {
+        return _writeToLegacyDeltaTables;
+    }
+
+    public boolean isWriteToBlockedDeltaTables() {
+        return _writeToBlockedDeltaTables;
+    }
+}

sor/src/main/java/com/bazaarvoice/emodb/sor/core/DataTools.java

@@ -50,4 +50,5 @@ public interface DataTools {
 
     /** Return a consistent TableSet view of all tables in the system. */
     TableSet createTableSet();
+
 }

sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultDataStore.java

@@ -823,8 +823,6 @@ private void checkValidTableOrAttributeName(String tableOrAttribute, String name
 
     }
 
-
-
     @Override
     public URI getStashRoot()
             throws StashNotAvailableException {

sor/src/main/java/com/bazaarvoice/emodb/sor/core/DefaultMigratorTools.java

@@ -0,0 +1,36 @@
+package com.bazaarvoice.emodb.sor.core;
+
+import com.bazaarvoice.emodb.sor.db.MigrationScanResult;
+import com.bazaarvoice.emodb.sor.db.MigratorReaderDAO;
+import com.bazaarvoice.emodb.sor.db.MigratorWriterDAO;
+import com.bazaarvoice.emodb.sor.db.ScanRange;
+import com.google.inject.Inject;
+
+import java.util.Iterator;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+public class DefaultMigratorTools implements MigratorTools {
+
+    private final MigratorReaderDAO _migratorReaderDao;
+    private final MigratorWriterDAO _migratorWriterDao;
+
+    @Inject
+    public DefaultMigratorTools(MigratorReaderDAO migratorReaderDAO, MigratorWriterDAO migratorWriterDAO) {
+        _migratorReaderDao = checkNotNull(migratorReaderDAO, "migratorReaderDao");
+        _migratorWriterDao = checkNotNull(migratorWriterDAO, "migratorWriterDao");
+    }
+
+    public void writeRows(String placement, Iterator<MigrationScanResult> results, int maxConcurrentWrites) {
+        checkNotNull(placement, "placement");
+        checkNotNull(results, "rows");
+        _migratorWriterDao.writeRows(placement, results, maxConcurrentWrites);
+    }
+
+    public Iterator<MigrationScanResult> readRows(String placement, ScanRange scanRange) {
+        checkNotNull(placement, "placement");
+        checkNotNull(scanRange, scanRange);
+        return _migratorReaderDao.readRows(placement, scanRange);
+    }
+
+}

sor/src/main/java/com/bazaarvoice/emodb/sor/core/MigratorTools.java

@@ -0,0 +1,15 @@
+package com.bazaarvoice.emodb.sor.core;
+
+
+import com.bazaarvoice.emodb.sor.db.MigrationScanResult;
+import com.bazaarvoice.emodb.sor.db.ScanRange;
+
+import java.util.Iterator;
+
+// interface to migrate deltas from old tables to new tables with blocking
+public interface MigratorTools {
+
+    void writeRows(String placement, Iterator<MigrationScanResult> results, int maxConcurrentWrites);
+
+    Iterator<MigrationScanResult> readRows(String placement, ScanRange scanRange);
+}

sor/src/main/java/com/bazaarvoice/emodb/sor/core/test/InMemoryDataStore.java

@@ -1,7 +1,7 @@
 package com.bazaarvoice.emodb.sor.core.test;
 
 import com.bazaarvoice.emodb.sor.core.DefaultDataStore;
-import com.bazaarvoice.emodb.sor.db.test.InMemoryDataDAO;
+import com.bazaarvoice.emodb.sor.db.test.InMemoryDataReaderDAO;
 import com.bazaarvoice.emodb.sor.log.NullSlowQueryLog;
 import com.bazaarvoice.emodb.table.db.test.InMemoryTableDAO;
 import com.codahale.metrics.MetricRegistry;
@@ -17,14 +17,14 @@
 public class InMemoryDataStore extends DefaultDataStore {
 
     public InMemoryDataStore(MetricRegistry metricRegistry) {
-        this(new InMemoryDataDAO(), metricRegistry);
+        this(new InMemoryDataReaderDAO(), metricRegistry);
     }
 
-    public InMemoryDataStore(InMemoryDataDAO dataDao, MetricRegistry metricRegistry) {
+    public InMemoryDataStore(InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry) {
         this(new EventBus(), dataDao, metricRegistry);
     }
 
-    public InMemoryDataStore(EventBus eventBus, InMemoryDataDAO dataDao, MetricRegistry metricRegistry) {
+    public InMemoryDataStore(EventBus eventBus, InMemoryDataReaderDAO dataDao, MetricRegistry metricRegistry) {
         super(eventBus, new InMemoryTableDAO(), dataDao, dataDao,
                 new NullSlowQueryLog(), MoreExecutors.sameThreadExecutor(), new InMemoryAuditStore(),
                 Optional.<URI>absent(), metricRegistry);