CASSANDRA-21674 CEP-46: update witness replication validation restrictions

doc/modules/cassandra/pages/architecture/dynamo.adoc

@@ -245,12 +245,8 @@ data. Configure it by specifying replication factor as
 `<total_replicas>/<transient_replicas` Both `SimpleStrategy` and
 `NetworkTopologyStrategy` support configuring transient replication.
 
-Transiently replicated keyspaces only support tables created with
-`read_repair` set to `NONE`; monotonic reads are not currently
-supported. You also can't use `LWT`, logged batches, or counters in {40_version}.
-You will possibly never be able to use materialized views with
-transiently replicated keyspaces and probably never be able to use
-secondary indices with them.
+Transient replication requires mutation tracking, so a transiently replicated
+keyspace must set `replication_type = 'tracked'`.
 
 Transient replication is an experimental feature that is not ready
 for production use. The expected audience is experienced users of

doc/modules/cassandra/pages/managing/operating/transientreplication.adoc

@@ -71,8 +71,8 @@ CREATE KEYSPACE some_keysopace WITH replication = {'class': 'NetworkTopologyStra
 'DC1' : '3/1'', 'DC2' : '5/2'};
 ....
 
-Transiently replicated keyspaces only support tables with `read_repair`
-set to `NONE`.
+Transient replication requires mutation tracking, so a transiently replicated
+keyspace must set `replication_type = 'tracked'`.
 
 Important Restrictions:
 
@@ -83,15 +83,78 @@ must first remove all transient replicas, then change the # of full
 replicas, then add back the transient replicas.
 * You can only safely increase number of transients one at a time with
 incremental repair run in between each time.
+* You can't add transient replicas to a keyspace whose mutation tracking
+migration is still in progress, nor enable mutation tracking and add
+transient replicas in the same statement. Set `replication_type = 'tracked'`
+first, wait for the migration to complete, then alter the replication
+factor.
+* You can't promote a transient replica to a full replica, because it holds
+no data for the range it witnessed. Either drop the transient replica from
+the replica set (`3/1` to `2`), or add a full replica before removing it
+(`3/1` to `4/1` to `3`) with a full repair run in between if the replica
+count must be preserved.
+
+Setting `-Dcassandra.allow_unsafe_transient_changes=true` skips every restriction
+above except the first, which is governed by
+`-Dcassandra.allow_alter_rf_during_range_movement`. Any alteration reached that
+way can leave a transient replica counted by reads it cannot answer, and needs a
+full repair afterwards to redistribute the data.
+
+Promotion is the one performed deliberately often enough to want it without a
+restart, so it also honours `AllowUnsafeWitnessPromotion` on the
+`StorageService` MBean, seeded from
+`-Dcassandra.allow_unsafe_witness_promotion`.
+
+=== Adding transient replicas to an existing keyspace
+
+A keyspace that is not already using mutation tracking adopts transient
+replication in two steps, because enabling mutation tracking starts a migration
+and reads for a range that has not yet migrated still take the untracked path.
+
+First enable mutation tracking:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication_type = 'tracked';
+....
+
+The migration repairs each range so that a single-replica tracked read reflects
+every write made while the keyspace was untracked. Wait for it to complete, then
+set the replication factor:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '3/1'};
+....
+
+A keyspace created with both settings at once needs no migration, because it has
+no pre-tracking writes to repair.
+
+=== Removing transient replicas
+
+Dropping the transient replica from the replica set needs no data movement,
+because the remaining full replicas already hold everything:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '2'};
+....
+
+Where the range must keep the same number of full replicas, add one before
+removing the transient replica, and run a full repair in between so the added
+replica is populated:
+
+....
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '4/1'};
+nodetool repair -full some_keyspace
+ALTER KEYSPACE some_keyspace WITH replication = {'class': 'SimpleStrategy',
+'replication_factor' : '3'};
+....
 
 Additionally, transient replication cannot be used for:
 
-* Monotonic Reads
-* Lightweight Transactions (LWTs)
-* Logged Batches
-* Counters
-* Keyspaces using materialized views
-* Secondary indexes (2i)
+* Logged Batches.
+* Keyspaces using materialized views.
 
 == Cheap Quorums
 

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

@@ -61,6 +61,7 @@ public enum CassandraRelevantProperties
     ALLOW_UNSAFE_JOIN("cassandra.allow_unsafe_join"),
     ALLOW_UNSAFE_REPLACE("cassandra.allow_unsafe_replace"),
     ALLOW_UNSAFE_TRANSIENT_CHANGES("cassandra.allow_unsafe_transient_changes"),
+    ALLOW_UNSAFE_WITNESS_PROMOTION("cassandra.allow_unsafe_witness_promotion"),
     APPROXIMATE_TIME_PRECISION_MS("cassandra.approximate_time_precision_ms", "2"),
     ASYNC_PROFILER_ENABLED("cassandra.async_profiler.enabled", "false"),
     ASYNC_PROFILER_UNSAFE_MODE("cassandra.async_profiler.unsafe_mode", "false"),

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

@@ -17,10 +17,15 @@
  */
 package org.apache.cassandra.cql3.statements.schema;
 
+import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import com.google.common.collect.Sets;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -34,6 +39,7 @@
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.LocalStrategy;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
 import org.apache.cassandra.locator.ReplicationFactor;
 import org.apache.cassandra.locator.SimpleStrategy;
 import org.apache.cassandra.schema.KeyspaceMetadata;
@@ -42,7 +48,6 @@
 import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
-import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.membership.NodeId;
@@ -53,6 +58,7 @@
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_ALTER_RF_DURING_RANGE_MOVEMENT;
 import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_TRANSIENT_CHANGES;
+import static org.apache.cassandra.config.CassandraRelevantProperties.ALLOW_UNSAFE_WITNESS_PROMOTION;
 import static org.apache.cassandra.replication.MutationTrackingService.DISABLED_MESSAGE;
 
 public final class AlterKeyspaceStatement extends AlterSchemaStatement
@@ -102,9 +108,6 @@ public Keyspaces apply(ClusterMetadata metadata)
         newKeyspace.params.validate(keyspaceName, state, metadata);
         newKeyspace.replicationStrategy.validate(metadata);
 
-        validateNoRangeMovements();
-        validateTransientReplication(keyspace, newKeyspace);
-
         // Because we used to not properly validate unrecognized options, we only log a warning if we find one.
         try
         {
@@ -156,6 +159,11 @@ public void validate(ClientState state)
             if (!DatabaseDescriptor.getMutationTrackingEnabled())
                 throw ire(DISABLED_MESSAGE);
         }
+
+        ClusterMetadata metadata = ClusterMetadata.current();
+        validateNoRangeMovements(metadata);
+        validateTransientReplication(keyspace, newKeyspace);
+        validateWitnessTransitions(metadata, keyspace, newKeyspace);
     }
 
     @Override
@@ -176,12 +184,11 @@ Set<String> clientWarnings(KeyspacesDiff diff)
         return clientWarnings;
     }
 
-    private void validateNoRangeMovements()
+    private void validateNoRangeMovements(ClusterMetadata metadata)
     {
         if (allow_alter_rf_during_range_movement)
             return;
 
-        ClusterMetadata metadata = ClusterMetadata.current();
         NodeId nodeId = metadata.directory.peerId(FBUtilities.getBroadcastAddressAndPort());
         Set<InetAddressAndPort> notNormalEndpoints = metadata.directory.states.entrySet().stream().filter(e -> !e.getKey().equals(nodeId)).filter(e -> {
             switch (e.getValue())
@@ -225,10 +232,6 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta
 
             if (!current.views.isEmpty())
                 throw new ConfigurationException("Cannot use transient replication on keyspaces using materialized views");
-
-            for (TableMetadata table : current.tables)
-                if (!table.indexes.isEmpty())
-                    throw new ConfigurationException("Cannot use transient replication on keyspaces using secondary indexes");
         }
 
         //This is true right now because the transition from transient -> full lacks the pending state
@@ -237,6 +240,12 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta
         if (oldFull > newFull && oldTrans > 0)
             throw new ConfigurationException("Can't add full replicas if there are any transient replicas. You must first remove all transient replicas, then change the # of full replicas, then add back the transient replicas");
 
+        //Promoting a witness to a full replica leaves it a full replica holding no data for the range
+        //it witnessed: tracked writes skip the column family store for tokens outside a node's full
+        //ranges, see Keyspace#applyInternalTracked. The promoted replica is then counted by quorum
+        //reads. Repair has to redistribute the data, so this is opt-in only.
+        validateNoWitnessPromotion(current.replicationStrategy, proposed.replicationStrategy);
+
         //Don't increase transient replication factor by more than one at a time if changing number of replicas
         //Just like with changing full replicas it's not safe to do this as you could read from too many replicas
         //that don't have the necessary data. W/O transient replication this alteration was allowed and it's not clear
@@ -247,6 +256,105 @@ private void validateTransientReplication(KeyspaceMetadata current, KeyspaceMeta
             throw new ConfigurationException("Can only safely increase number of transients one at a time with incremental repair run in between each time");
     }
 
+    /**
+     * A witness is promoted when the transient count falls while the full count rises. This promotes witness replicas
+     * to full replicas without sending them the data they need to participate in reads.
+     */
+    private void validateNoWitnessPromotion(AbstractReplicationStrategy current, AbstractReplicationStrategy proposed)
+    {
+        if (allowUnsafeWitnessPromotion)
+            return;
+
+        if (!current.getClass().equals(proposed.getClass()))
+        {
+            if (promotesWitness(current.getReplicationFactor(), proposed.getReplicationFactor()))
+                throw witnessPromotionRejected();
+            return;
+        }
+
+        Map<String, ReplicationFactor> before = replicationFactorsByGroup(current);
+        Map<String, ReplicationFactor> after = replicationFactorsByGroup(proposed);
+        for (String group : Sets.union(before.keySet(), after.keySet()))
+        {
+            if (promotesWitness(before.getOrDefault(group, ReplicationFactor.ZERO),
+                                after.getOrDefault(group, ReplicationFactor.ZERO)))
+                throw witnessPromotionRejected();
+        }
+    }
+
+    private static boolean promotesWitness(ReplicationFactor before, ReplicationFactor after)
+    {
+        return after.transientReplicas() < before.transientReplicas() && after.fullReplicas > before.fullReplicas;
+    }
+
+    /**
+     * Replication factors keyed by the unit the strategy assigns them to independently: datacenter for
+     * {@link NetworkTopologyStrategy}, and the keyspace as a whole for every other strategy, whose
+     * aggregate factor is already its only group.
+     */
+    private static Map<String, ReplicationFactor> replicationFactorsByGroup(AbstractReplicationStrategy strategy)
+    {
+        if (!(strategy instanceof NetworkTopologyStrategy))
+            return Collections.singletonMap("", strategy.getReplicationFactor());
+
+        NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy;
+        Map<String, ReplicationFactor> byDc = new HashMap<>();
+        for (String dc : nts.getDatacenters())
+            byDc.put(dc, nts.getReplicationFactor(dc));
+        return byDc;
+    }
+
+    private ConfigurationException witnessPromotionRejected()
+    {
+        return new ConfigurationException(String.format("Cannot promote a transient replica of %s to a full replica: " +
+                                                        "it holds no data for the range it witnessed. Set %s=true over " +
+                                                        "JMX, or %s=true at startup, to allow it, then run a full repair " +
+                                                        "to distribute the data.",
+                                                        keyspaceName,
+                                                        ALLOW_UNSAFE_WITNESS_PROMOTION.getKey(),
+                                                        ALLOW_UNSAFE_TRANSIENT_CHANGES.getKey()));
+    }
+
+    private static volatile boolean allowUnsafeWitnessPromotion = ALLOW_UNSAFE_WITNESS_PROMOTION.getBoolean();
+
+    public static void setAllowUnsafeWitnessPromotion(boolean allow)
+    {
+        allowUnsafeWitnessPromotion = allow;
+    }
+
+    public static boolean getAllowUnsafeWitnessPromotion()
+    {
+        return allowUnsafeWitnessPromotion;
+    }
+
+    /**
+     * Alterations that would leave a witness (transient replica) counted by reads it cannot answer.
+     * These are unsafe rather than inexpressible, so cassandra.allow_unsafe_transient_changes opts out
+     * of them alongside the rules in validateTransientReplication.
+     */
+    private void validateWitnessTransitions(ClusterMetadata metadata,
+                                            KeyspaceMetadata current,
+                                            KeyspaceMetadata proposed)
+    {
+        if (allow_unsafe_transient_changes)
+            return;
+
+        boolean proposesWitnesses = proposed.replicationStrategy.getReplicationFactor().hasTransientReplicas();
+
+        if (proposesWitnesses && metadata.mutationTrackingMigrationState.isMigrating(keyspaceName))
+            throw new ConfigurationException(String.format("Cannot add transient replicas to %s while its mutation " +
+                                                           "tracking migration is in progress. Wait for the migration " +
+                                                           "to complete, then alter the replication factor.",
+                                                           keyspaceName));
+
+        if (proposesWitnesses && proposed.params.replicationType.isTracked() && !current.params.replicationType.isTracked())
+            throw new ConfigurationException(String.format("Cannot enable mutation tracking on %s and add transient " +
+                                                           "replicas in the same statement, because doing so starts a " +
+                                                           "migration. Set replication_type = 'tracked' first, wait for " +
+                                                           "the migration to complete, then alter the replication factor.",
+                                                           keyspaceName));
+    }
+
     @Override
     public AuditLogContext getAuditLogContext()
     {

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

@@ -71,7 +71,6 @@
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.consensus.TransactionalMode;
 import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode;
-import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.membership.Directory;
@@ -700,12 +699,6 @@ public KeyspaceMetadata apply(Epoch epoch, KeyspaceMetadata keyspace, TableMetad
                           "before being replayed.");
             }
 
-            if (keyspace.replicationStrategy.hasTransientReplicas()
-                && params.readRepair != ReadRepairStrategy.NONE)
-            {
-                throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
-            }
-
             if (!params.compression.isEnabled())
                 Guardrails.uncompressedTablesEnabled.ensureEnabled(state);
 

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

@@ -55,7 +55,6 @@
 import org.apache.cassandra.schema.UserFunctions;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.ClientWarn;
-import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.serialization.Version;
 import org.apache.cassandra.transport.Event.SchemaChange;
@@ -194,12 +193,6 @@ public Keyspaces apply(ClusterMetadata metadata)
         if (sourceTableMeta.isCompactTable())
             Guardrails.compactTablesEnabled.ensureEnabled(state);
 
-        if (sourceKeyspaceMeta.replicationStrategy.hasTransientReplicas()
-            && sourceTableMeta.params.readRepair != ReadRepairStrategy.NONE)
-        {
-            throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
-        }
-
         if (!sourceTableMeta.params.compression.isEnabled())
             Guardrails.uncompressedTablesEnabled.ensureEnabled(state);
 

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

@@ -65,7 +65,6 @@ public final class CreateIndexStatement extends AlterSchemaStatement
     public static final String TABLE_DOES_NOT_EXIST = "Table '%s' doesn't exist";
     public static final String COUNTER_TABLES_NOT_SUPPORTED = "Secondary indexes on counter tables aren't supported";
     public static final String MATERIALIZED_VIEWS_NOT_SUPPORTED = "Secondary indexes on materialized views aren't supported";
-    public static final String TRANSIENTLY_REPLICATED_KEYSPACE_NOT_SUPPORTED = "Secondary indexes are not supported on transiently replicated keyspaces";
     public static final String CUSTOM_CREATE_WITHOUT_COLUMN = "Only CUSTOM indexes can be created without specifying a target column";
     public static final String CUSTOM_MULTIPLE_COLUMNS = "Only CUSTOM indexes support multiple columns";
     public static final String DUPLICATE_TARGET_COLUMN = "Duplicate column '%s' in index target list";

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

@@ -175,13 +175,6 @@ public Keyspaces apply(ClusterMetadata metadata)
         TableMetadata table = builder.build();
         table.validate();
 
-        // TODO (review): This can be removed right? ReadRepair is effectively not done anymore so the setting doesn't matter
-//        if (keyspace.replicationStrategy.hasTransientReplicas()
-//            && table.params.readRepair != ReadRepairStrategy.NONE)
-//        {
-//            throw ire("read_repair must be set to 'NONE' for transiently replicated keyspaces");
-//        }
-
         if (!table.params.compression.isEnabled() && !SchemaConstants.isSystemKeyspace(table.keyspace))
             Guardrails.uncompressedTablesEnabled.ensureEnabled(state);
 

src/java/org/apache/cassandra/locator/ReplicaPlans.java

@@ -234,6 +234,10 @@ public static Replica findCounterLeaderReplica(ClusterMetadata metadata, String
 
         EndpointsForToken replicas = metadata.placements.get(keyspace.getMetadata().params.replication).reads.forToken(key.getToken()).get();
 
+        // A transient replica holds no data for the range it witnesses, so it would resolve against nothing
+        // and write a value that discards every prior increment.
+        replicas = replicas.filter(Replica::isFull);
+
         // CASSANDRA-13043: filter out those endpoints not accepting clients yet, maybe because still bootstrapping
         replicas = replicas.filter(replica -> StorageService.instance.isRpcReady(replica.endpoint()));