This is an automated email from the ASF dual-hosted git repository. marcuse pushed a commit to branch trunk in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit b5f8ac26040966750e7cacec80dd0e4dd8bfbde6 Author: Marcus Eriksson <[email protected]> AuthorDate: Fri Aug 29 08:56:34 2025 +0200 Update system schema tables with new distributed keyspace on upgrade Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-20872 --- CHANGES.txt | 1 + .../apache/cassandra/schema/DistributedSchema.java | 21 ++++++++++++++++---- src/java/org/apache/cassandra/tcm/Startup.java | 23 +++++++++++++++++++++- .../upgrade/ClusterMetadataUpgradeTest.java | 6 ++++++ 4 files changed, 46 insertions(+), 5 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index b3dbba2f9b..ef556fc92d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Update system schema tables with new distributed keyspace on upgrade (CASSANDRA-20872) * Fix issue when running cms reconfiguration with paxos repair disabled (CASSANDRA-20869) * Added additional parameter to JVM shutdown to allow for logs to be properly shutdown (CASSANDRA-20978) * Improve isGossipOnlyMember and location lookup performance (CASSANDRA-21039) diff --git a/src/java/org/apache/cassandra/schema/DistributedSchema.java b/src/java/org/apache/cassandra/schema/DistributedSchema.java index 77fec54eda..1266d9b4c3 100644 --- a/src/java/org/apache/cassandra/schema/DistributedSchema.java +++ b/src/java/org/apache/cassandra/schema/DistributedSchema.java @@ -20,6 +20,7 @@ package org.apache.cassandra.schema; import com.google.common.base.Preconditions; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.ImmutableList; import org.apache.cassandra.auth.AuthKeyspace; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.functions.UserFunction; @@ -35,6 +36,7 @@ import org.apache.cassandra.tcm.serialization.MetadataSerializer; import org.apache.cassandra.tcm.serialization.Version; import org.apache.cassandra.tracing.TraceKeyspace; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; import java.io.IOException; import java.util.ArrayList; @@ -152,14 +154,25 @@ public class DistributedSchema implements MetadataValue<DistributedSchema> return keyspaces.stream().anyMatch(ksm -> ksm.tables.stream().anyMatch(TableMetadata::requiresAccordSupport)); } + /** + * @deprecated since TCM, used on upgrade from gossip to populate system schema tables with the correct generation + */ + @Deprecated(since = "TCM") + public static List<Pair<KeyspaceMetadata, Long>> distributedKeyspacesWithGeneration(Set<String> knownDatacenters) + { + return ImmutableList.of(Pair.create(DistributedMetadataLogKeyspace.initialMetadata(knownDatacenters), DistributedMetadataLogKeyspace.GENERATION), + Pair.create(TraceKeyspace.metadata(), TraceKeyspace.GENERATION), + Pair.create(SystemDistributedKeyspace.metadata(), SystemDistributedKeyspace.GENERATION), + Pair.create(AuthKeyspace.metadata(),AuthKeyspace.GENERATION)); + } + public static DistributedSchema fromSystemTables(Keyspaces keyspaces, Set<String> knownDatacenters) { if (!keyspaces.containsKeyspace(SchemaConstants.METADATA_KEYSPACE_NAME)) { - Keyspaces kss = Keyspaces.of(DistributedMetadataLogKeyspace.initialMetadata(knownDatacenters), - TraceKeyspace.metadata(), - SystemDistributedKeyspace.metadata(), - AuthKeyspace.metadata()); + Keyspaces kss = Keyspaces.none(); + for (Pair<KeyspaceMetadata, Long> ksmGen : distributedKeyspacesWithGeneration(knownDatacenters)) + kss = kss.with(ksmGen.left); for (KeyspaceMetadata ksm : keyspaces) // on disk keyspaces kss = kss.withAddedOrUpdated(kss.get(ksm.name) .map(k -> merged(ksm, k)) diff --git a/src/java/org/apache/cassandra/tcm/Startup.java b/src/java/org/apache/cassandra/tcm/Startup.java index 446c52a02e..53a0f6095a 100644 --- a/src/java/org/apache/cassandra/tcm/Startup.java +++ b/src/java/org/apache/cassandra/tcm/Startup.java @@ -18,8 +18,10 @@ package org.apache.cassandra.tcm; import java.io.IOException; +import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; @@ -37,6 +39,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.dht.BootStrapper; @@ -49,8 +52,11 @@ import org.apache.cassandra.gms.NewGossiper; import org.apache.cassandra.gms.VersionedValue; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.DistributedSchema; import org.apache.cassandra.schema.KeyspaceMetadata; +import org.apache.cassandra.schema.Keyspaces; import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.tcm.log.LocalLog; @@ -69,6 +75,7 @@ import org.apache.cassandra.tcm.transformations.PrepareReplace; import org.apache.cassandra.tcm.transformations.UnsafeJoin; import org.apache.cassandra.tcm.transformations.cms.Initialize; import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.Pair; import static org.apache.cassandra.tcm.ClusterMetadataService.State.LOCAL; import static org.apache.cassandra.tcm.compatibility.GossipHelper.emptyWithSchemaFromSystemTables; @@ -254,12 +261,26 @@ import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort; Election.instance.migrated(); } + private static void updateSystemSchemaTables(Set<String> knownDatacenters) + { + List<Pair<KeyspaceMetadata, Long>> kss = DistributedSchema.distributedKeyspacesWithGeneration(knownDatacenters); + List<Mutation> mutations = new ArrayList<>(); + for (Pair<KeyspaceMetadata, Long> ksm : kss) + { + Keyspaces.KeyspacesDiff ksDiff = Keyspaces.diff(Keyspaces.none(), Keyspaces.of(ksm.left)); + mutations.addAll(SchemaKeyspace.convertSchemaDiffToMutations(ksDiff, ksm.right)); + } + SchemaKeyspace.applyChanges(mutations); + } + /** * This should only be called during startup. */ public static void initializeFromGossip(Function<Processor, Processor> wrapProcessor, Runnable initMessaging) throws StartupException { - ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(SystemKeyspace.allKnownDatacenters()); + Set<String> knownDcs = SystemKeyspace.allKnownDatacenters(); + updateSystemSchemaTables(knownDcs); + ClusterMetadata emptyFromSystemTables = emptyWithSchemaFromSystemTables(knownDcs); LocalLog.LogSpec logSpec = LocalLog.logSpec() .withInitialState(emptyFromSystemTables) .afterReplay(Startup::scrubDataDirectories, diff --git a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeTest.java b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeTest.java index eb717602f3..c7e33b8228 100644 --- a/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeTest.java +++ b/test/distributed/org/apache/cassandra/distributed/upgrade/ClusterMetadataUpgradeTest.java @@ -35,6 +35,7 @@ import org.apache.cassandra.tcm.membership.Location; import org.apache.cassandra.tcm.membership.NodeAddresses; import org.apache.cassandra.tcm.membership.NodeId; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.psjava.util.AssertStatus.assertTrue; @@ -54,6 +55,11 @@ public class ClusterMetadataUpgradeTest extends UpgradeTestBase cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))"); }) .runAfterClusterUpgrade((cluster) -> { + Object [][] r = cluster.get(1).executeInternal("select keyspace_name from system_schema.keyspaces where keyspace_name='system_cluster_metadata'"); + assertEquals(1, r.length); + r = cluster.get(1).executeInternal("select table_name from system_schema.tables where keyspace_name='system_cluster_metadata' and table_name='distributed_metadata_log'"); + assertEquals(1, r.length); + cluster.get(1).nodetoolResult("cms","initialize").asserts().success(); cluster.forEach(i -> { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
