natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[CASSANDRA-20736] Separate MetaStrategy placements from others

This commit is contained in:
Sam Tunnicliffe 2025-06-18 15:41:07 +01:00
parent 1a553f88e4
commit bed10b544f
8 changed files with 100 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/java/org/apache/cassandra/locator/EndpointsForToken.java
View File

@ -25,7 +25,9 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.tcm.ownership.VersionedEndpoints;
/**
@ -164,7 +166,11 @@ public class EndpointsForToken extends Endpoints<EndpointsForToken>
public static VersionedEndpoints.ForToken natural(Keyspace keyspace, Token token)
{
return ClusterMetadata.current().placements.get(keyspace.getMetadata().params.replication).reads.forToken(token);
ReplicationParams replication = keyspace.getMetadata().params.replication;
DataPlacement placement = replication.isMeta()
? ClusterMetadata.current().getCMSPlacement()
: ClusterMetadata.current().placements.get(replication);
return placement.reads.forToken(token);
}
}

17
src/java/org/apache/cassandra/locator/ReplicaLayout.java
View File

@ -239,7 +239,9 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
{
// todo deduplicate so that "pending" contains "read - write",
// which is a hack until we revisit how consistency level handles pending
DataPlacement dataPlacement = metadata.placements.get(ks.params.replication);
DataPlacement dataPlacement = ks.params.replication.isMeta()
? metadata.getCMSPlacement()
: metadata.placements.get(ks.params.replication);
natural = forNonLocalStrategyTokenRead(dataPlacement, token);
// perf optimization to avoid double endpoints search and filtering for a typical case
// DataPlacement constructor does a deduplication of reads/writes, so we can use cheap == comparision here
@ -394,7 +396,10 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
static EndpointsForRange forNonLocalStategyRangeRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, AbstractBounds<PartitionPosition> range)
{
return metadata.placements.get(keyspace.params.replication).reads.forRange(range.right.getToken()).get();
DataPlacement placement = keyspace.params.replication.isMeta()
? metadata.getCMSPlacement()
: metadata.placements.get(keyspace.params.replication);
return placement.reads.forRange(range.right.getToken()).get();
}
public static EndpointsForToken forNonLocalStrategyTokenRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token)
@ -407,17 +412,11 @@ public abstract class ReplicaLayout<E extends Endpoints<E>>
return dataPlacement.reads.forToken(token).get();
}
static EndpointsForToken forNonLocalStrategyTokenWrite(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token)
{
return forNonLocalStrategyTokenWrite(metadata.placements.get(keyspace.params.replication), token);
}
static EndpointsForToken forNonLocalStrategyTokenWrite(DataPlacement dataPlacement, Token token)
private static EndpointsForToken forNonLocalStrategyTokenWrite(DataPlacement dataPlacement, Token token)
{
return dataPlacement.writes.forToken(token).get();
}
static EndpointsForRange forLocalStrategyRange(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
{
return replicationStrategy.calculateNaturalReplicas(range.right.getToken(), metadata);

107
src/java/org/apache/cassandra/tcm/ClusterMetadata.java
View File

@ -126,6 +126,7 @@ public class ClusterMetadata
private Set<InetAddressAndPort> fullCMSEndpoints;
private volatile Map<ReplicationParams, RangesAtEndpoint> localRangesAllSettled = null;
private static final RangesAtEndpoint EMPTY_LOCAL_RANGES = RangesAtEndpoint.empty(FBUtilities.getBroadcastAddressAndPort());
private DataPlacement cmsDataPlacement;
public ClusterMetadata(IPartitioner partitioner)
{
@ -214,7 +215,7 @@ public class ClusterMetadata
this.directory = directory;
this.tokenMap = tokenMap;
this.accordFastPath = accordFastPath;
this.placements = maybeAddMetaPlacement(placements, cmsMembership);
this.placements = placements;
this.lockedRanges = lockedRanges;
this.inProgressSequences = inProgressSequences;
this.consensusMigrationState = consensusMigrationState;
@ -222,6 +223,7 @@ public class ClusterMetadata
this.locator = Locator.usingDirectory(directory);
this.accordStaleReplicas = accordStaleReplicas;
this.cmsMembership = cmsMembership;
this.cmsDataPlacement = calculateCMSPlacement(placements, cmsMembership);
}
public Set<NodeId> fullCMSMemberIds()
@ -264,52 +266,57 @@ public class ClusterMetadata
return fullCMSReplicas;
}
private DataPlacements maybeAddMetaPlacement(DataPlacements placements, CMSMembership cms)
public DataPlacement getCMSPlacement()
{
return cmsDataPlacement;
}
private DataPlacement calculateCMSPlacement(DataPlacements placements, CMSMembership cms)
{
if (epoch.isBefore(Epoch.FIRST) || schema.getKeyspaces().get(SchemaConstants.METADATA_KEYSPACE_NAME).isEmpty())
return placements;
return DataPlacement.empty();
DataPlacement metaPlacement;
Epoch previousLastModified = placements.lastModified();
Epoch nextLastModified;
if (epoch.is(Epoch.FIRST))
if (directory.isEmpty())
{
// PRE_INITIALIZE_CMS: placements need to be hardcoded to the local address so that the subsequent
// INITIALIZE_CMS can be committed
Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort());
metaPlacement = DataPlacement.builder()
.withReadReplica(Epoch.FIRST, localReplica)
.withWriteReplica(Epoch.FIRST, localReplica)
.build();
nextLastModified = Epoch.FIRST;
}
else if (epoch.isAfter(Epoch.FIRST) && directory.isEmpty())
{
// This cluster did not previously upgrade from a gossip based version (i.e. pre-6.0) but did at some point
// run a version prior to MetadataVersion.V7 where we started to encode CMS membership directly. This
// condition implies that we are reconstructing a serialized cluster metadata during replay or else the
// directory should not be empty after Epoch.FIRST as the base state in INITIALIZE_CMS now includes the
// first CMS node. Similarly, if the cluster had previously been running a gossip-based version, the
// directory would contain entries for each of the live nodes at the time of upgrade.
// Given this state, the very next transformation that is/was applied will be to register the node that
// committed the PRE_INITIALIZE_CMS and INTIALIZE_CMS transformations. So we just leave the placements
// untouched as they will already contain that node as an endpoint.
return placements;
if (epoch.is(Epoch.FIRST))
{
// PRE_INITIALIZE_CMS: placements need to be hardcoded to the local address so that the subsequent
// INITIALIZE_CMS can be committed
Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort());
return DataPlacement.builder()
.withReadReplica(Epoch.FIRST, localReplica)
.withWriteReplica(Epoch.FIRST, localReplica)
.build();
}
else
{
// This cluster did not previously upgrade from a gossip based version (i.e. pre-6.0) but did at some point
// run a version prior to MetadataVersion.V7 where we started to encode CMS membership directly. This
// condition implies that we are reconstructing a serialized cluster metadata during replay or else the
// directory should not be empty after Epoch.FIRST as the base state in INITIALIZE_CMS now includes the
// first CMS node. Similarly, if the cluster had previously been running a gossip-based version, the
// directory would contain entries for each of the live nodes at the time of upgrade.
// Given this state, the very next transformation that is/was applied will be to register the node that
// committed the PRE_INITIALIZE_CMS and INTIALIZE_CMS transformations. So we just extract the placements
// associated with the MetaStrategy params as they will already contain that node as an endpoint.
for (ReplicationParams params : placements.keys())
if (params.isMeta())
return placements.get(params);
// This point can only be reached in tests.
// TODO enforce that invariant somehow
Replica localReplica = MetaStrategy.replica(FBUtilities.getBroadcastAddressAndPort());
return DataPlacement.builder()
.withReadReplica(epoch, localReplica)
.withWriteReplica(epoch, localReplica)
.build();
}
}
else
{
// Build a placement based on the CMS membership
metaPlacement = cms.toPlacement(directory);
if (cms.lastModified().isAfter(previousLastModified))
nextLastModified = cms.lastModified();
else
nextLastModified = previousLastModified;
return cms.toPlacement(directory);
}
return placements.unbuild()
.with(ReplicationParams.meta(this), metaPlacement)
.build()
.withLastModified(nextLastModified);
}
public Transformer transformer()
@ -1220,7 +1227,10 @@ public class ClusterMetadata
DistributedSchema.serializer.serialize(metadata.schema, out, version);
Directory.serializer.serialize(metadata.directory, out, version);
TokenMap.serializer.serialize(metadata.tokenMap, out, version);
DataPlacements.serializer.serialize(metadata.placements, out, version);
// Prior to V9, placements for the MetaStrategy keyspace were included in the main DataPlacements
// so when targetting such a version, emulate that.
DataPlacements placements = version.isBefore(Version.V9) ? preV9Placements(metadata) : metadata.placements;
DataPlacements.serializer.serialize(placements, out, version);
if (version.isAtLeast(MIN_ACCORD_VERSION))
{
AccordFastPath.serializer.serialize(metadata.accordFastPath, out, version);
@ -1239,6 +1249,7 @@ public class ClusterMetadata
assert key.valueType.isInstance(value);
value.serialize(out, version);
}
// From V9 CMS membership is directly encoded in ClusterMetadata
if (version.isAtLeast(Version.V9))
CMSMembership.serializer.serialize(metadata.cmsMembership, out, version);
}
@ -1381,8 +1392,7 @@ public class ClusterMetadata
sizeof(metadata.partitioner.getClass().getCanonicalName()) +
DistributedSchema.serializer.serializedSize(metadata.schema, version) +
Directory.serializer.serializedSize(metadata.directory, version) +
TokenMap.serializer.serializedSize(metadata.tokenMap, version) +
DataPlacements.serializer.serializedSize(metadata.placements, version);
TokenMap.serializer.serializedSize(metadata.tokenMap, version);
if (version.isAtLeast(MIN_ACCORD_VERSION))
{
@ -1394,12 +1404,27 @@ public class ClusterMetadata
size += LockedRanges.serializer.serializedSize(metadata.lockedRanges, version) +
InProgressSequences.serializer.serializedSize(metadata.inProgressSequences, version);
// Prior to V9, placements for the MetaStrategy keyspace were included in the main DataPlacements
// so when targetting such a version, emulate that.
DataPlacements placements = version.isBefore(Version.V9) ? preV9Placements(metadata) : metadata.placements;
size += DataPlacements.serializer.serializedSize(placements, version);
// From V9 CMS membership is directly encoded in ClusterMetadata
if (version.isAtLeast(Version.V9))
size += CMSMembership.serializer.serializedSize(metadata.cmsMembership, version);
return size;
}
private DataPlacements preV9Placements(ClusterMetadata metadata)
{
if (metadata.cmsDataPlacement.isEmpty())
return metadata.placements;
return metadata.placements.unbuild()
.with(ReplicationParams.meta(metadata), metadata.cmsDataPlacement)
.build();
}
public static IPartitioner getPartitioner(DataInputPlus in, Version version) throws IOException
{
if (version.isAtLeast(Version.V1))

2
src/java/org/apache/cassandra/tcm/PaxosBackedProcessor.java
View File

@ -70,7 +70,7 @@ public class PaxosBackedProcessor extends AbstractLocalProcessor
if (metadata.epoch.isAfter(Epoch.FIRST) && metadata.fullCMSMembers().contains(FBUtilities.getBroadcastAddressAndPort()))
return true;
return metadata.epoch.isEqualOrBefore(Epoch.FIRST)
&& metadata.placements.get(ReplicationParams.meta(metadata)).reads.byEndpoint().keySet().contains(FBUtilities.getBroadcastAddressAndPort());
&& metadata.getCMSPlacement().reads.byEndpoint().keySet().contains(FBUtilities.getBroadcastAddressAndPort());
}
@Override

5
src/java/org/apache/cassandra/tcm/ownership/DataPlacement.java
View File

@ -106,6 +106,11 @@ public class DataPlacement
return EMPTY;
}
public boolean isEmpty()
{
return reads.isEmpty() && writes.isEmpty();
}
public static Builder builder()
{
return new Builder(ReplicaGroups.builder(),

18
src/java/org/apache/cassandra/tcm/ownership/DataPlacements.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.tcm.serialization.Version;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.db.TypeSizes.sizeof;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
public class DataPlacements extends ReplicationMap<DataPlacement> implements MetadataValue<DataPlacements>
{
@ -256,14 +257,13 @@ public class DataPlacements extends ReplicationMap<DataPlacement> implements Met
public void serialize(DataPlacements t, DataOutputPlus out, Version version) throws IOException
{
Map<ReplicationParams, DataPlacement> map = t.asMap();
if (version.isBefore(Version.V9))
out.writeInt(map.size());
else
out.writeUnsignedVInt32(map.size());
// From V7, placements for the metadata keyspace are derived from CMSMembership, not serialized
int mapSize = version.isBefore(Version.V7) ? map.size() : Math.max(map.size() - 1, 0);
out.writeInt(mapSize);
for (Map.Entry<ReplicationParams, DataPlacement> entry : map.entrySet())
{
if (version.isAtLeast(Version.V7) && entry.getKey().isMeta())
continue;
ReplicationParams.serializer.serialize(entry.getKey(), out, version);
DataPlacement.serializerFor(entry.getKey()).serialize(entry.getValue(), out, version);
}
@ -272,7 +272,7 @@ public class DataPlacements extends ReplicationMap<DataPlacement> implements Met
public DataPlacements deserialize(DataInputPlus in, Version version) throws IOException
{
int size = in.readInt();
int size = version.isBefore(Version.V9) ? in.readInt() : in.readUnsignedVInt32();
Map<ReplicationParams, DataPlacement> map = Maps.newHashMapWithExpectedSize(size);
for (int i = 0; i < size; i++)
{
@ -286,13 +286,9 @@ public class DataPlacements extends ReplicationMap<DataPlacement> implements Met
public long serializedSize(DataPlacements t, Version version)
{
Map<ReplicationParams, DataPlacement> map = t.asMap();
// From V7, placements for the metadata keyspace are derived from CMSMembership, not serialized
int mapSize = version.isBefore(Version.V7) ? map.size() : Math.max(map.size() - 1, 0);
long size = sizeof(mapSize);
long size = version.isBefore(Version.V9) ? sizeof(map.size()) : sizeofUnsignedVInt(map.size());
for (Map.Entry<ReplicationParams, DataPlacement> entry : map.entrySet())
{
if (version.isAtLeast(Version.V7) && entry.getKey().isMeta())
continue;
size += ReplicationParams.serializer.serializedSize(entry.getKey(), version);
size += DataPlacement.serializerFor(entry.getKey()).serializedSize(entry.getValue(), version);
}

6
src/java/org/apache/cassandra/tcm/ownership/UniformRangePlacement.java
View File

@ -329,7 +329,11 @@ public class UniformRangePlacement implements PlacementProvider
logger.trace("Calculating data placements for {}", ksMetadata.name);
AbstractReplicationStrategy replication = ksMetadata.replicationStrategy;
ReplicationParams params = ksMetadata.params.replication;
if (params.isMeta() || params.isLocal())
if (params.isMeta())
{
// don't calculate meta strategy placements, these are derived from ClusterMetadata.cmsMembership
}
else if (params.isLocal())
{
placements.put(params, metadata.placements.get(params));
}

1
src/java/org/apache/cassandra/tcm/serialization/Version.java
View File

@ -80,6 +80,7 @@ public enum Version
V8(8),
/**
* - DataPlacements don't include MetaStrategy, replaced by ClusterMetadata.CMSMembership
* - Size of DataPlacements is encoded as vint
*/
V9(9),