Update system schema tables with new distributed keyspace on upgrade

Patch by marcuse; reviewed by Sam Tunnicliffe for CASSANDRA-20872

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)

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))

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,

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 ->
             {