CEP-45: Initial read path integration and reconciliation implementation

src/java/org/apache/cassandra/batchlog/BatchlogManager.java

@@ -36,6 +36,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.RateLimiter;
 import org.apache.cassandra.concurrent.ScheduledExecutorPlus;
+import org.apache.cassandra.db.MutationId;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.transport.Dispatcher;
@@ -126,10 +127,10 @@ public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedExcep
 
     public static void remove(TimeUUID id)
     {
-        new Mutation(PartitionUpdate.fullPartitionDelete(SystemKeyspace.Batches,
-                                                         id.toBytes(),
-                                                         FBUtilities.timestampMicros(),
-                                                         FBUtilities.nowInSeconds()))
+        new Mutation(MutationId.none(), PartitionUpdate.fullPartitionDelete(SystemKeyspace.Batches,
+                                                                             id.toBytes(),
+                                                                             FBUtilities.timestampMicros(),
+                                                                             FBUtilities.nowInSeconds()))
             .apply();
     }
 

src/java/org/apache/cassandra/config/Config.java

@@ -646,6 +646,8 @@ public static class SSTableConfig
 
     public boolean dynamic_data_masking_enabled = false;
 
+    public boolean mutation_tracking_enabled = false;
+
     /**
      * Time in milliseconds after a warning will be emitted to the log and to the client that a UDF runs too long.
      * (Only valid, if user_defined_functions_threads_enabled==true)

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -5394,6 +5394,20 @@ public static void setDynamicDataMaskingEnabled(boolean enabled)
         }
     }
 
+    public static boolean getMutationTrackingEnabled()
+    {
+        return conf.mutation_tracking_enabled;
+    }
+
+    public static void setMutationTrackingEnabled(boolean enabled)
+    {
+        if (enabled != conf.mutation_tracking_enabled)
+        {
+            logger.info("Setting mutation_tracking_enabled to {}", enabled);
+            conf.mutation_tracking_enabled = enabled;
+        }
+    }
+
     public static OptionalDouble getSeverityDuringDecommission()
     {
         return conf.severity_during_decommission > 0 ?

src/java/org/apache/cassandra/cql3/QueryProcessor.java

@@ -58,7 +58,6 @@
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.ReadQuery;
-import org.apache.cassandra.db.ReadResponse;
 import org.apache.cassandra.db.SinglePartitionReadQuery;
 import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -88,6 +87,7 @@
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.service.pager.QueryPager;
+import org.apache.cassandra.service.reads.IReadResponse;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.transport.Dispatcher;
@@ -514,17 +514,17 @@ else if (readQuery instanceof SinglePartitionReadQuery.Group)
             {
                 throw new IllegalArgumentException("Unable to handle; only expected ReadCommands but given " + readQuery.getClass());
             }
-            Future<List<Message<ReadResponse>>> future = FutureCombiner.allOf(commands.stream()
-                                                                                      .map(rc -> Message.out(rc.verb(), rc))
-                                                                                      .map(m -> MessagingService.instance().<ReadCommand, ReadResponse>sendWithResult(m, address))
-                                                                                      .collect(Collectors.toList()));
+            Future<List<Message<IReadResponse>>> future = FutureCombiner.allOf(commands.stream()
+                                                                                       .map(rc -> Message.out(rc.verb(), rc))
+                                                                                       .map(m -> MessagingService.instance().<ReadCommand, IReadResponse>sendWithResult(m, address))
+                                                                                       .collect(Collectors.toList()));
 
             ResultSetBuilder result = new ResultSetBuilder(select.getResultMetadata(), select.getSelection().newSelectors(options), false);
             return future.map(list -> {
                 int i = 0;
-                for (Message<ReadResponse> m : list)
+                for (Message<IReadResponse> m : list)
                 {
-                    ReadResponse rsp = m.payload;
+                    IReadResponse rsp = m.payload;
                     try (PartitionIterator it = UnfilteredPartitionIterators.filter(rsp.makeIterator(commands.get(i++)), nowInSec))
                     {
                         while (it.hasNext())

src/java/org/apache/cassandra/cql3/statements/BatchUpdatesCollector.java

@@ -32,6 +32,7 @@
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.IMutation;
 import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.MutationId;
 import org.apache.cassandra.db.RegularAndStaticColumns;
 import org.apache.cassandra.db.commitlog.CommitLogSegment;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
@@ -218,12 +219,14 @@ public MutationBuilder add(PartitionUpdate.Builder updateBuilder)
         public Mutation build()
         {
             ImmutableMap.Builder<TableId, PartitionUpdate> updates = new ImmutableMap.Builder<>();
+            boolean createId = false;
             for (Map.Entry<TableId, PartitionUpdate.Builder> updateEntry : modifications.entrySet())
             {
                 PartitionUpdate update = updateEntry.getValue().build();
                 updates.put(updateEntry.getKey(), update);
+                createId |= update.metadata().hasLoggedReplication();
             }
-            return new Mutation(keyspaceName, key, updates.build(), createdAt);
+            return new Mutation(createId ? MutationId.createNext() : MutationId.none(), keyspaceName, key, updates.build(), createdAt);
         }
 
         public PartitionUpdate.Builder get(TableId tableId)

src/java/org/apache/cassandra/cql3/statements/SingleTableUpdatesCollector.java

@@ -30,6 +30,7 @@
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.IMutation;
 import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.MutationId;
 import org.apache.cassandra.db.RegularAndStaticColumns;
 import org.apache.cassandra.db.commitlog.CommitLogSegment;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
@@ -104,9 +105,9 @@ public List<IMutation> toMutations(ClientState state)
             if (metadata.isVirtual())
                 mutation = new VirtualMutation(builder.build());
             else if (metadata.isCounter())
-                mutation = new CounterMutation(new Mutation(builder.build()), counterConsistencyLevel);
+                mutation = new CounterMutation(new Mutation(MutationId.createFor(metadata), builder.build()), counterConsistencyLevel);
             else
-                mutation = new Mutation(builder.build());
+                mutation = new Mutation(MutationId.createFor(metadata), builder.build());
 
             mutation.validateIndexedColumns(state);
             mutation.validateSize(MessagingService.current_version, CommitLogSegment.ENTRY_OVERHEAD_SIZE);

src/java/org/apache/cassandra/cql3/statements/schema/KeyspaceAttributes.java

@@ -25,6 +25,7 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.KeyspaceParams.Option;
+import org.apache.cassandra.schema.ReplicationType;
 import org.apache.cassandra.schema.ReplicationParams;
 
 public final class KeyspaceAttributes extends PropertyDefinitions
@@ -66,17 +67,25 @@ private Map<String, String> getAllReplicationOptions()
     KeyspaceParams asNewKeyspaceParams()
     {
         boolean durableWrites = getBoolean(Option.DURABLE_WRITES.toString(), KeyspaceParams.DEFAULT_DURABLE_WRITES);
-        return KeyspaceParams.create(durableWrites, getAllReplicationOptions());
+
+        String rtypeName = getString(Option.REPLICATION_TYPE.toString());
+        ReplicationType replicationType = rtypeName != null ? ReplicationType.valueOf(rtypeName) : KeyspaceParams.DEFAULT_REPLICATION_TYPE;
+
+        return KeyspaceParams.create(durableWrites, getAllReplicationOptions(), replicationType);
     }
 
     KeyspaceParams asAlteredKeyspaceParams(KeyspaceParams previous)
     {
         boolean durableWrites = getBoolean(Option.DURABLE_WRITES.toString(), previous.durableWrites);
+        String rtypeName = getString(Option.REPLICATION_TYPE.toString());
+        ReplicationType replicationType = rtypeName != null ? ReplicationType.valueOf(rtypeName) : previous.replicationType;
+
+
         Map<String, String> previousOptions = previous.replication.options;
         ReplicationParams replication = getReplicationStrategyClass() == null
                                       ? previous.replication
                                       : ReplicationParams.fromMapWithDefaults(getAllReplicationOptions(), previousOptions);
-        return new KeyspaceParams(durableWrites, replication);
+        return new KeyspaceParams(durableWrites, replication, replicationType);
     }
 
     public boolean hasOption(Option option)

src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java

@@ -32,6 +32,7 @@
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.schema.TableParams;
 import org.apache.cassandra.schema.TableParams.Option;
+import org.apache.cassandra.schema.TableParams.TableReplicationType;
 import org.apache.cassandra.service.reads.SpeculativeRetryPolicy;
 import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 
@@ -151,6 +152,9 @@ private TableParams build(TableParams.Builder builder)
         if (hasOption(READ_REPAIR))
             builder.readRepair(ReadRepairStrategy.fromString(getString(READ_REPAIR)));
 
+        if (hasOption(REPLICATION_TYPE))
+            builder.replicationType(TableReplicationType.fromString(getString(REPLICATION_TYPE)));
+
         return builder.build();
     }
 

src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java

@@ -25,6 +25,8 @@
 import org.apache.cassandra.db.commitlog.CommitLogPosition;
 import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.replication.MutationTracker;
+import org.apache.cassandra.replication.MutationTrackingService;
 import org.apache.cassandra.schema.TableId;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.concurrent.OpOrder;
@@ -42,24 +44,30 @@ public CassandraKeyspaceWriteHandler(Keyspace keyspace)
     public WriteContext beginWrite(Mutation mutation, boolean makeDurable) throws RequestExecutionException
     {
         OpOrder.Group group = null;
+        MutationTracker.PendingWrite pendingWrite = null;
         try
         {
             group = Keyspace.writeOrder.start();
+            pendingWrite = MutationTrackingService.instance().startWrite(mutation);
 
             // write the mutation to the commitlog and memtables
             CommitLogPosition position = null;
             if (makeDurable)
             {
                 position = addToCommitLog(mutation);
             }
-            return new CassandraWriteContext(group, position);
+            return new CassandraWriteContext(group, position, pendingWrite);
         }
         catch (Throwable t)
         {
             if (group != null)
             {
                 group.close();
             }
+            if (pendingWrite != null)
+            {
+                pendingWrite.close();
+            }
             throw t;
         }
     }
@@ -105,7 +113,7 @@ private WriteContext createEmptyContext()
         try
         {
             group = Keyspace.writeOrder.start();
-            return new CassandraWriteContext(group, null);
+            return new CassandraWriteContext(group, null, MutationTracker.PendingWrite.NOOP);
         }
         catch (Throwable t)
         {

src/java/org/apache/cassandra/db/CassandraWriteContext.java

@@ -21,18 +21,21 @@
 import com.google.common.base.Preconditions;
 
 import org.apache.cassandra.db.commitlog.CommitLogPosition;
+import org.apache.cassandra.replication.MutationTracker;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
 public class CassandraWriteContext implements WriteContext
 {
     private final OpOrder.Group opGroup;
     private final CommitLogPosition position;
+    private final MutationTracker.PendingWrite pendingWrite;
 
-    public CassandraWriteContext(OpOrder.Group opGroup, CommitLogPosition position)
+    public CassandraWriteContext(OpOrder.Group opGroup, CommitLogPosition position, MutationTracker.PendingWrite pendingWrite)
     {
         Preconditions.checkArgument(opGroup != null);
         this.opGroup = opGroup;
         this.position = position;
+        this.pendingWrite = pendingWrite;
     }
 
     public static CassandraWriteContext fromContext(WriteContext context)
@@ -55,5 +58,6 @@ public CommitLogPosition getPosition()
     public void close()
     {
         opGroup.close();
+        pendingWrite.close();
     }
 }

src/java/org/apache/cassandra/db/CounterMutation.java

@@ -68,6 +68,12 @@ public CounterMutation(Mutation mutation, ConsistencyLevel consistency)
         this.consistency = consistency;
     }
 
+    @Override
+    public MutationId id()
+    {
+        return mutation.id();
+    }
+
     public String getKeyspaceName()
     {
         return mutation.getKeyspaceName();

src/java/org/apache/cassandra/db/IMutation.java

@@ -30,6 +30,7 @@ public interface IMutation
 {
     long MAX_MUTATION_SIZE = DatabaseDescriptor.getMaxMutationSize();
 
+    MutationId id();
     void apply();
     String getKeyspaceName();
     Collection<TableId> getTableIds();

src/java/org/apache/cassandra/db/Keyspace.java

@@ -52,6 +52,7 @@
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.metrics.KeyspaceMetrics;
 import org.apache.cassandra.repair.KeyspaceRepairManager;
+import org.apache.cassandra.replication.MutationTrackingService;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
@@ -545,6 +546,8 @@ else if (isDeferrable)
         }
         try (WriteContext ctx = getWriteHandler().beginWrite(mutation, makeDurable))
         {
+            MutationTrackingService.instance().add(mutation);
+
             for (PartitionUpdate upd : mutation.getPartitionUpdates())
             {
                 ColumnFamilyStore cfs = columnFamilyStores.get(upd.metadata().id);