bdeggleston commented on code in PR #4708:
URL: https://github.com/apache/cassandra/pull/4708#discussion_r3431837523
##########
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java:
##########
@@ -437,4 +461,372 @@ 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
coordination plan
+ * targeting live remote replicas with CL.ONE.
+ */
+ 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);
+ ResponseTracker tracker =
createTrackerForWrite(plan.consistencyLevel(), plan, plan.pending, metadata);
+
+ 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);
+ ReplicaLayout.ForTokenWrite electorate =
consistencyForConsensus.isDatacenterLocal()
+ ?
all.filter(InOurDc.replicas()) : all;
+
+ EndpointsForToken live = all.all().filter(isReplicaAlive);
+ return new Paxos.Participants(metadata.epoch,
Keyspace.open(table.keyspace), consistencyForConsensus, all, electorate, live,
+ (cm) -> Paxos.Participants.get(cm,
table, actualToken, consistencyForConsensus));
+ }
+
+ /**
+ * Hook for replication strategies to send additional mutations alongside
a paxos commit.
+ * Called from PaxosCommit.start() after local synchronous execution for
tracked keyspaces.
+ *
+ * If the method doesn't return null, the returned future is composed with
the paxos consensus:
+ * onDone fires only after both the paxos quorum decision AND this future
complete, or after
+ * one of them fails.
+ */
+ public Future<Void> sendPaxosCommitMutations(Agreed commit, boolean
isUrgent)
+ {
+ return null;
+ }
+
+ /**
+ * Check whether paxos operations should be rejected for the given token.
+ */
+ public boolean shouldRejectPaxos(Token token)
+ {
+ return false;
+ }
+
+ /**
+ * Create ResponseTracker for write operation based on consistency level.
+ */
+ @VisibleForTesting
+ public ResponseTracker createTrackerForWrite(ConsistencyLevel cl,
ReplicaPlan.ForWrite plan, Endpoints<?> pending, ClusterMetadata metadata)
+ {
+ switch (cl)
+ {
+ case ANY:
+ case ONE:
+ case TWO:
+ case THREE:
+ case QUORUM:
+ case ALL:
+ int totalContacts = plan.contacts().size();
+ if (pending.isEmpty())
+ {
+ int blockFor = cl.blockFor(this);
+ return new SimpleResponseTracker(blockFor, totalContacts);
+ }
+ else
+ {
+ // Check if double count model applies (some CLs like ANY
don't add pending)
+ int baseBlockFor = cl.blockFor(this);
+ int totalBlockFor = cl.blockForWrite(this, pending);
+
+ // If totalBlockFor == baseBlockFor, no double-count
needed (e.g., ANY)
+ if (totalBlockFor == baseBlockFor)
+ return new SimpleResponseTracker(baseBlockFor,
totalContacts);
+
+ // Double count model: committed must satisfy base CL,
total must include pending
+ int pendingReplicas = pending.size();
+ // contacts() includes both committed and pending replicas
+ int committedReplicas = totalContacts - pendingReplicas;
+ return new WriteResponseTracker(baseBlockFor,
totalBlockFor,
+ committedReplicas,
pendingReplicas,
+ endpoint ->
pending.endpoints().contains(endpoint));
+ }
+
+ case LOCAL_ONE:
+ case LOCAL_QUORUM:
+ int localContacts =
plan.contacts().filter(InOurDc.replicas()).size();
Review Comment:
it looks that way, but most of the calls are subtly different between the 2
cases
--
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]
