bdeggleston commented on code in PR #4708:
URL: https://github.com/apache/cassandra/pull/4708#discussion_r3443833004
##########
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java:
##########
@@ -437,4 +461,376 @@ public RangesAtEndpoint getLocalRanges(ClusterMetadata cm)
return newRanges;
}
}
+
+ protected CoordinationPlan.ForWrite planForWriteInternal(ClusterMetadata
metadata,
+
Keyspace keyspace,
+
ConsistencyLevel consistencyLevel,
+
Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
+
ReplicaPlans.Selector selector)
+ {
+ ReplicaPlan.ForWrite plan = ReplicaPlans.forWrite(metadata, keyspace,
consistencyLevel, liveAndDown, selector);
+ ResponseTracker tracker = createTrackerForWrite(consistencyLevel,
plan, plan.pending, metadata);
+ return new CoordinationPlan.ForWrite(plan, tracker);
+ }
+
+ public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata
metadata,
+ Keyspace keyspace,
+ ConsistencyLevel
consistencyLevel,
+
Function<ClusterMetadata, ReplicaLayout.ForTokenWrite> liveAndDown,
+
ReplicaPlans.Selector selector)
+ {
+ CoordinationPlan.ForWrite actual = planForWriteInternal(metadata,
keyspace, consistencyLevel, liveAndDown, selector);
+
+ CoordinationPlan.ForWrite ideal = null;
+ ConsistencyLevel idealCL =
DatabaseDescriptor.getIdealConsistencyLevel();
+ if (idealCL != null)
+ {
+ if (idealCL == consistencyLevel)
+ {
+ ideal = actual;
+ }
+ else
+ {
+ ideal = planForWriteInternal(metadata, keyspace, idealCL,
liveAndDown, selector);
+ }
+ }
+
+ return new CoordinationPlan.ForWriteWithIdeal(metadata,
actual.replicas(), actual.responses(), ideal);
+ }
+
+ public CoordinationPlan.ForWriteWithIdeal planForWrite(ClusterMetadata
metadata,
+ Keyspace keyspace,
+ ConsistencyLevel
consistencyLevel,
+ Token token,
+
ReplicaPlans.Selector selector)
+ {
+ return planForWrite(metadata, keyspace, consistencyLevel,
+ (newClusterMetadata) ->
ReplicaLayout.forTokenWriteLiveAndDown(newClusterMetadata, keyspace, token),
selector);
+ }
+
+ /**
+ * Create coordination plan for forwarding a counter write to the leader
replica.
+ *
+ * In cases where the original coordinator is not a replica of the counter
key, the counter
+ * mutation is forwarded to a leader replica that will coordinate the
actual counter update.
+ */
+ public CoordinationPlan.ForWrite
planForForwardingCounterWrite(ClusterMetadata metadata,
+ Keyspace
keyspace,
+ Token token,
+
Function<ClusterMetadata, Replica> replicaSupplier)
+ {
+ ReplicaPlan.ForWrite plan =
ReplicaPlans.forSingleReplicaWrite(metadata, keyspace, token, replicaSupplier);
+ ResponseTracker tracker =
createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata);
+
+ return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker,
null);
+ }
+
+ /**
+ * Create coordination plan for replaying a mutation from the batchlog.
+ *
+ * When recovering failed batches, mutations are replayed to remote
replicas only
+ * (local replica is handled separately). This method creates a replica
plan
+ * targeting live remote replicas with CL.ONE, and a response tracker that
waits on
+ * all contacts
+ */
+ public CoordinationPlan.ForWriteWithIdeal
planForReplayMutation(ClusterMetadata metadata,
+ Keyspace
keyspace,
+ Token
token)
+ {
+ Preconditions.checkState(!replicationType.isTracked(), "Batch replay
not supported with tracked keyspaces");
+
+ ReplicaPlan.ForWrite plan = ReplicaPlans.forReplayMutation(metadata,
keyspace, token);
+
+ // wait until all contacts respond
+ int blockFor = plan.contacts().size();
+ ResponseTracker tracker = new SimpleResponseTracker(blockFor,
blockFor);
+
+ return new CoordinationPlan.ForWriteWithIdeal(metadata, plan, tracker,
null);
+ }
+
+ /**
+ * Create coordination plan for a single-partition token read.
+ */
+ public CoordinationPlan.ForTokenRead planForTokenRead(ClusterMetadata
metadata,
+ Keyspace keyspace,
+ TableId tableId,
+ Token token,
+ @Nullable
Index.QueryPlan indexQueryPlan,
+ ConsistencyLevel
consistencyLevel,
+
SpeculativeRetryPolicy retry,
+ ReadCoordinator
coordinator)
+ {
+ ReplicaPlan.ForTokenRead plan = ReplicaPlans.forRead(metadata,
keyspace, tableId, token, indexQueryPlan, consistencyLevel, retry, coordinator);
+ ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan);
+ ResponseTracker tracker = createTrackerForRead(plan);
+ return new CoordinationPlan.ForTokenRead(shared, tracker);
+ }
+
+ /**
+ * Create coordination plan for a range read.
+ */
+ public CoordinationPlan.ForRangeRead planForRangeRead(ClusterMetadata
metadata,
+ Keyspace keyspace,
+ TableId tableId,
+ @Nullable
Index.QueryPlan indexQueryPlan,
+ ConsistencyLevel
consistencyLevel,
+
AbstractBounds<PartitionPosition> range,
+ int vnodeCount)
+ {
+ ReplicaPlan.ForRangeRead plan = ReplicaPlans.forRangeRead(metadata,
keyspace, tableId, indexQueryPlan, consistencyLevel, range, vnodeCount, true);
+ ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan);
+ ResponseTracker tracker = createTrackerForRead(plan);
+ return new CoordinationPlan.ForRangeRead(shared, tracker);
+ }
+
+ /**
+ * Attempt to merge two adjacent range read coordination plans into one.
+ *
+ * If the two plans share enough live endpoints to satisfy the consistency
level
+ * and the merge is worthwhile returns a merged plan otherwise returns
null.
+ */
+ public CoordinationPlan.ForRangeRead maybeMergeRangeReads(ClusterMetadata
metadata,
+ Keyspace
keyspace,
+ TableId tableId,
+
ConsistencyLevel consistencyLevel,
+
ReplicaPlan.ForRangeRead left,
+
ReplicaPlan.ForRangeRead right)
+ {
+ ReplicaPlan.ForRangeRead merged = ReplicaPlans.maybeMerge(metadata,
keyspace, tableId, consistencyLevel, left, right);
+ if (merged == null)
+ return null;
+
+ ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(merged);
+ ResponseTracker tracker = createTrackerForRead(merged);
+ return new CoordinationPlan.ForRangeRead(shared, tracker);
+ }
+
+ /**
+ * Create coordination plan for a full range read
+ */
+ public CoordinationPlan.ForRangeRead planForFullRangeRead(Keyspace
keyspace,
+ ConsistencyLevel
consistencyLevel,
+
AbstractBounds<PartitionPosition> range,
+
Set<InetAddressAndPort> endpointsToContact,
+ int vnodeCount)
+ {
+ ReplicaPlan.ForRangeRead plan =
ReplicaPlans.forFullRangeRead(keyspace, consistencyLevel, range,
endpointsToContact, vnodeCount);
+ ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan);
+ ResponseTracker tracker = createTrackerForRead(plan);
+ return new CoordinationPlan.ForRangeRead(shared, tracker);
+ }
+
+ /**
+ * Create coordination plan for a single-replica token read.
+ */
+ public CoordinationPlan.ForTokenRead
planForSingleReplicaTokenRead(Keyspace keyspace, Token token, Replica replica)
+ {
+ ReplicaPlan.ForTokenRead plan =
ReplicaPlans.forSingleReplicaRead(keyspace, token, replica);
+ ReplicaPlan.SharedForTokenRead shared = ReplicaPlan.shared(plan);
+ ResponseTracker tracker = createTrackerForRead(plan);
+ return new CoordinationPlan.ForTokenRead(shared, tracker);
+ }
+
+ /**
+ * Create coordination plan for a single-replica range read.
+ *
+ * Used by short read protection to fetch additional partitions from a
+ * specific replica. blockFor=1, totalReplicas=1.
+ */
+ public CoordinationPlan.ForRangeRead
planForSingleReplicaRangeRead(Keyspace keyspace,
+
AbstractBounds<PartitionPosition> range,
+ Replica
replica,
+ int
vnodeCount)
+ {
+ ReplicaPlan.ForRangeRead plan =
ReplicaPlans.forSingleReplicaRead(keyspace, range, replica, vnodeCount);
+ ReplicaPlan.SharedForRangeRead shared = ReplicaPlan.shared(plan);
+ ResponseTracker tracker = createTrackerForRead(plan);
+ return new CoordinationPlan.ForRangeRead(shared, tracker);
+ }
+
+ /**
+ * Create ResponseTracker for read operation.
+ */
+ public <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E, P>>
ResponseTracker createTrackerForRead(P plan)
+ {
+ int blockFor = plan.readQuorum();
+
+ // Use candidates.size() for totalReplicas to allow for speculation
+ // (speculation can contact additional candidates beyond initial
contacts)
+ int totalReplicas = plan.readCandidates().size();
+
+ return new SimpleResponseTracker(blockFor, totalReplicas);
+ }
+
+ public Paxos.Participants paxosParticipants(ClusterMetadata metadata,
+ TableMetadata table,
+ Token token,
+ ConsistencyLevel
consistencyForConsensus,
+ Predicate<Replica>
isReplicaAlive)
+ {
+
+ KeyspaceMetadata keyspaceMetadata =
metadata.schema.getKeyspaceMetadata(table.keyspace);
+ // MetaStrategy distributes the entire keyspace to all replicas. In
addition, its tables (currently only
+ // the dist log table) don't use the globally configured partitioner.
For these reasons we don't lookup the
+ // replicas using the supplied token as this can actually be of the
incorrect type (for example when
+ // performing Paxos repair).
+ final Token actualToken = table.partitioner ==
MetaStrategy.partitioner ? MetaStrategy.entireRange.right : token;
+ ReplicaLayout.ForTokenWrite all =
forTokenWriteLiveAndDown(keyspaceMetadata, actualToken);
Review Comment:
There's a few pieces of metadata flying around here. ClusterMetadata and
TableMetadata are passed into the method as arguments. However we need
KeyspaceMetadata to get the replica layout, and we get that from the combo of
cluster and table metadata in the line at the top of them method,
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
