diff --git a/NEWS.txt b/NEWS.txt index 54ed7c6ada..ce05b92234 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -21,6 +21,10 @@ Upgrading _ New SSTable version 'la' with improved bloom-filter false-positive handling compared to previous version 'ka' used in 2.2 and 2.1. Running sstableupgrade is not necessary but recommended. + - Before upgrading to 3.0, make sure that your cluster is in complete agreement + (schema versions outputted by `nodetool describecluster` are all the same). + - Schema metadata is now stored in the new `system_schema` keyspace, and + legacy `system.schema_*` tables are now gone; see CASSANDRA-6717 for details. - Pig's CassandraStorage has been removed. Use CqlNativeStorage instead. - Hadoop BulkOutputFormat and BulkRecordWriter have been removed; use CqlBulkOutputFormat and CqlBulkRecordWriter instead. diff --git a/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-20150617-shaded.jar b/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar similarity index 86% rename from lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-20150617-shaded.jar rename to lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar index 7d971df718..905120219c 100644 Binary files a/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-20150617-shaded.jar and b/lib/cassandra-driver-core-2.2.0-rc2-SNAPSHOT-shaded.jar differ diff --git a/lib/cassandra-driver-internal-only-2.6.0c2.post.zip b/lib/cassandra-driver-internal-only-2.6.0c2.zip similarity index 50% rename from lib/cassandra-driver-internal-only-2.6.0c2.post.zip rename to lib/cassandra-driver-internal-only-2.6.0c2.zip index 5fa57c74d6..be9f162da4 100644 Binary files a/lib/cassandra-driver-internal-only-2.6.0c2.post.zip and b/lib/cassandra-driver-internal-only-2.6.0c2.zip differ diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index 637234ab8a..2767fb129b 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -33,8 +33,8 @@ class UnexpectedTableStructure(UserWarning): def __str__(self): return 'Unexpected table structure; may not translate correctly to CQL. ' + self.msg -SYSTEM_KEYSPACES = ('system', 'system_traces', 'system_auth', 'system_distributed') -NONALTERBALE_KEYSPACES = ('system') +SYSTEM_KEYSPACES = ('system', 'system_schema', 'system_traces', 'system_auth', 'system_distributed') +NONALTERBALE_KEYSPACES = ('system', 'system_schema') class Cql3ParsingRuleSet(CqlParsingRuleSet): keywords = set(( diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 6deee6d80c..f3c8bc1dde 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -34,7 +34,6 @@ import org.apache.commons.lang3.builder.HashCodeBuilder; import org.apache.commons.lang3.builder.ToStringBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.github.jamm.Unmetered; import org.apache.cassandra.cache.CachingOptions; import org.apache.cassandra.cql3.ColumnIdentifier; @@ -49,11 +48,12 @@ import org.apache.cassandra.exceptions.*; import org.apache.cassandra.io.compress.CompressionParameters; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.UUIDGen; import org.apache.cassandra.utils.Pair; +import org.github.jamm.Unmetered; /** * This class can be tricky to modify. Please read http://wiki.apache.org/cassandra/ConfigurationNotes for how to do so safely. @@ -791,7 +791,7 @@ public final class CFMetaData */ public boolean reload() { - return apply(LegacySchemaTables.createTableFromName(ksName, cfName)); + return apply(SchemaKeyspace.createTableFromName(ksName, cfName)); } /** diff --git a/src/java/org/apache/cassandra/config/Schema.java b/src/java/org/apache/cassandra/config/Schema.java index 4d3e6f34c1..29dffa1ba1 100644 --- a/src/java/org/apache/cassandra/config/Schema.java +++ b/src/java/org/apache/cassandra/config/Schema.java @@ -22,6 +22,7 @@ import java.security.NoSuchAlgorithmException; import java.util.*; import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -49,6 +50,9 @@ public class Schema public static final Schema instance = new Schema(); + /* system keyspace names (the ones with LocalStrategy replication strategy) */ + public static final Set SYSTEM_KEYSPACE_NAMES = ImmutableSet.of(SystemKeyspace.NAME, SchemaKeyspace.NAME); + /** * longest permissible KS or CF name. Our main concern is that filename not be more than 255 characters; * the filename will contain both the KS and CF names. Since non-schema-name components only take up @@ -88,9 +92,18 @@ public class Schema */ public Schema() { + load(SchemaKeyspace.metadata()); load(SystemKeyspace.metadata()); } + /** + * @return whether or not the keyspace is a really system one (w/ LocalStrategy, unmodifiable, hardcoded) + */ + public static boolean isSystemKeyspace(String keyspaceName) + { + return SYSTEM_KEYSPACE_NAMES.contains(keyspaceName.toLowerCase()); + } + /** * load keyspace (keyspace) definitions, but do not initialize the keyspace instances. * Schema version may be updated as the result. @@ -107,7 +120,7 @@ public class Schema */ public Schema loadFromDisk(boolean updateVersion) { - load(LegacySchemaTables.readSchemaFromSystemTables()); + load(SchemaKeyspace.readSchemaFromSystemTables()); if (updateVersion) updateVersion(); return this; @@ -253,7 +266,7 @@ public class Schema */ public List getNonSystemKeyspaces() { - return ImmutableList.copyOf(Sets.difference(keyspaces.keySet(), Collections.singleton(SystemKeyspace.NAME))); + return ImmutableList.copyOf(Sets.difference(keyspaces.keySet(), SYSTEM_KEYSPACE_NAMES)); } /** @@ -344,6 +357,11 @@ public class Schema cfIdMap.put(key, cfm.cfId); } + public void unload(CFMetaData cfm) + { + cfIdMap.remove(Pair.create(cfm.ksName, cfm.cfName)); + } + /** * Used for ColumnFamily data eviction out from the schema * @@ -351,7 +369,7 @@ public class Schema */ public void purge(CFMetaData cfm) { - cfIdMap.remove(Pair.create(cfm.ksName, cfm.cfName)); + unload(cfm); cfm.markPurged(); } @@ -410,7 +428,7 @@ public class Schema */ public void updateVersion() { - version = LegacySchemaTables.calculateSchemaDigest(); + version = SchemaKeyspace.calculateSchemaDigest(); SystemKeyspace.updateSchemaVersion(version); } @@ -526,6 +544,9 @@ public class Schema ColumnFamilyStore cfs = Keyspace.open(ksName).getColumnFamilyStore(tableName); assert cfs != null; + // make sure all the indexes are dropped, or else. + cfs.indexManager.setIndexRemoved(new HashSet<>(cfs.indexManager.getBuiltIndexes())); + // reinitialize the keyspace. CFMetaData cfm = oldKsm.tables.get(tableName).get(); KeyspaceMetadata newKsm = oldKsm.withSwapped(oldKsm.tables.without(tableName)); diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterKeyspaceStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterKeyspaceStatement.java index 3e1bca5d48..b660f5258a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/AlterKeyspaceStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/AlterKeyspaceStatement.java @@ -19,7 +19,6 @@ package org.apache.cassandra.cql3.statements; import org.apache.cassandra.auth.Permission; import org.apache.cassandra.config.Schema; -import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.exceptions.*; import org.apache.cassandra.locator.LocalStrategy; import org.apache.cassandra.schema.KeyspaceMetadata; @@ -56,7 +55,7 @@ public class AlterKeyspaceStatement extends SchemaAlteringStatement KeyspaceMetadata ksm = Schema.instance.getKSMetaData(name); if (ksm == null) throw new InvalidRequestException("Unknown keyspace " + name); - if (ksm.name.equalsIgnoreCase(SystemKeyspace.NAME)) + if (Schema.isSystemKeyspace(ksm.name)) throw new InvalidRequestException("Cannot alter system keyspace"); attrs.validate(); diff --git a/src/java/org/apache/cassandra/db/DefinitionsUpdateVerbHandler.java b/src/java/org/apache/cassandra/db/DefinitionsUpdateVerbHandler.java index d5ede03c05..7add503960 100644 --- a/src/java/org/apache/cassandra/db/DefinitionsUpdateVerbHandler.java +++ b/src/java/org/apache/cassandra/db/DefinitionsUpdateVerbHandler.java @@ -26,7 +26,7 @@ import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.MessageIn; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.WrappedRunnable; /** @@ -47,7 +47,7 @@ public class DefinitionsUpdateVerbHandler implements IVerbHandler system() { - return Iterables.transform(Collections.singleton(SystemKeyspace.NAME), keyspaceTransformer); + return Iterables.transform(Schema.SYSTEM_KEYSPACE_NAMES, keyspaceTransformer); } @Override diff --git a/src/java/org/apache/cassandra/db/MigrationRequestVerbHandler.java b/src/java/org/apache/cassandra/db/MigrationRequestVerbHandler.java index 79753c1295..6502ed3047 100644 --- a/src/java/org/apache/cassandra/db/MigrationRequestVerbHandler.java +++ b/src/java/org/apache/cassandra/db/MigrationRequestVerbHandler.java @@ -26,7 +26,7 @@ import org.apache.cassandra.net.IVerbHandler; import org.apache.cassandra.net.MessageIn; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.service.MigrationManager; /** @@ -41,7 +41,7 @@ public class MigrationRequestVerbHandler implements IVerbHandler { logger.debug("Received migration request from {}.", message.from); MessageOut> response = new MessageOut<>(MessagingService.Verb.INTERNAL_RESPONSE, - LegacySchemaTables.convertSchemaToMutations(), + SchemaKeyspace.convertSchemaToMutations(), MigrationManager.MigrationsSerializer.instance); MessagingService.instance().sendReply(response, id, message.from); } diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 7cc488477d..b5182ddc1b 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -27,6 +27,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.index.SecondaryIndexSearcher; import org.apache.cassandra.db.filter.*; import org.apache.cassandra.db.rows.*; @@ -323,7 +324,7 @@ public abstract class ReadCommand implements ReadQuery private final int failureThreshold = DatabaseDescriptor.getTombstoneFailureThreshold(); private final int warningThreshold = DatabaseDescriptor.getTombstoneWarnThreshold(); - private final boolean respectTombstoneThresholds = !ReadCommand.this.metadata().ksName.equals(SystemKeyspace.NAME); + private final boolean respectTombstoneThresholds = !Schema.isSystemKeyspace(ReadCommand.this.metadata().ksName); private int liveRows = 0; private int tombstones = 0; diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java index a3b021f3cc..f0c91d6aa4 100644 --- a/src/java/org/apache/cassandra/db/SystemKeyspace.java +++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java @@ -34,6 +34,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.config.Schema; import org.apache.cassandra.cql3.QueryProcessor; import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.cql3.functions.*; @@ -55,7 +56,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.Functions; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.schema.Tables; import org.apache.cassandra.schema.Types; import org.apache.cassandra.service.StorageService; @@ -101,6 +102,14 @@ public final class SystemKeyspace public static final String SIZE_ESTIMATES = "size_estimates"; public static final String AVAILABLE_RANGES = "available_ranges"; + @Deprecated public static final String LEGACY_KEYSPACES = "schema_keyspaces"; + @Deprecated public static final String LEGACY_COLUMNFAMILIES = "schema_columnfamilies"; + @Deprecated public static final String LEGACY_COLUMNS = "schema_columns"; + @Deprecated public static final String LEGACY_TRIGGERS = "schema_triggers"; + @Deprecated public static final String LEGACY_USERTYPES = "schema_usertypes"; + @Deprecated public static final String LEGACY_FUNCTIONS = "schema_functions"; + @Deprecated public static final String LEGACY_AGGREGATES = "schema_aggregates"; + public static final CFMetaData Hints = compile(HINTS, "hints awaiting delivery", @@ -142,12 +151,11 @@ public final class SystemKeyspace + "PRIMARY KEY ((row_key), cf_id))") .compactionStrategyClass(LeveledCompactionStrategy.class); - // TODO: make private - public static final CFMetaData BuiltIndexes = + private static final CFMetaData BuiltIndexes = compile(BUILT_INDEXES, "built column indexes", "CREATE TABLE \"%s\" (" - + "table_name text," + + "table_name text," // table_name here is the name of the keyspace - don't be fooled + "index_name text," + "PRIMARY KEY ((table_name), index_name)) " + "WITH COMPACT STORAGE"); @@ -263,6 +271,122 @@ public final class SystemKeyspace + "ranges set," + "PRIMARY KEY ((keyspace_name)))"); + @Deprecated + public static final CFMetaData LegacyKeyspaces = + compile(LEGACY_KEYSPACES, + "*DEPRECATED* keyspace definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "durable_writes boolean," + + "strategy_class text," + + "strategy_options text," + + "PRIMARY KEY ((keyspace_name))) " + + "WITH COMPACT STORAGE"); + + @Deprecated + public static final CFMetaData LegacyColumnfamilies = + compile(LEGACY_COLUMNFAMILIES, + "*DEPRECATED* table definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "columnfamily_name text," + + "bloom_filter_fp_chance double," + + "caching text," + + "cf_id uuid," // post-2.1 UUID cfid + + "comment text," + + "compaction_strategy_class text," + + "compaction_strategy_options text," + + "comparator text," + + "compression_parameters text," + + "default_time_to_live int," + + "default_validator text," + + "dropped_columns map," + + "dropped_columns_types map," + + "gc_grace_seconds int," + + "is_dense boolean," + + "key_validator text," + + "local_read_repair_chance double," + + "max_compaction_threshold int," + + "max_index_interval int," + + "memtable_flush_period_in_ms int," + + "min_compaction_threshold int," + + "min_index_interval int," + + "read_repair_chance double," + + "speculative_retry text," + + "subcomparator text," + + "type text," + + "PRIMARY KEY ((keyspace_name), columnfamily_name))"); + + @Deprecated + public static final CFMetaData LegacyColumns = + compile(LEGACY_COLUMNS, + "*DEPRECATED* column definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "columnfamily_name text," + + "column_name text," + + "component_index int," + + "index_name text," + + "index_options text," + + "index_type text," + + "type text," + + "validator text," + + "PRIMARY KEY ((keyspace_name), columnfamily_name, column_name))"); + + @Deprecated + public static final CFMetaData LegacyTriggers = + compile(LEGACY_TRIGGERS, + "*DEPRECATED* trigger definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "columnfamily_name text," + + "trigger_name text," + + "trigger_options map," + + "PRIMARY KEY ((keyspace_name), columnfamily_name, trigger_name))"); + + @Deprecated + public static final CFMetaData LegacyUsertypes = + compile(LEGACY_USERTYPES, + "*DEPRECATED* user defined type definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "type_name text," + + "field_names list," + + "field_types list," + + "PRIMARY KEY ((keyspace_name), type_name))"); + + @Deprecated + public static final CFMetaData LegacyFunctions = + compile(LEGACY_FUNCTIONS, + "*DEPRECATED* user defined function definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "function_name text," + + "signature frozen>," + + "argument_names list," + + "argument_types list," + + "body text," + + "language text," + + "return_type text," + + "called_on_null_input boolean," + + "PRIMARY KEY ((keyspace_name), function_name, signature))"); + + @Deprecated + public static final CFMetaData LegacyAggregates = + compile(LEGACY_AGGREGATES, + "*DEPRECATED* user defined aggregate definitions", + "CREATE TABLE %s (" + + "keyspace_name text," + + "aggregate_name text," + + "signature frozen>," + + "argument_types list," + + "final_func text," + + "initcond blob," + + "return_type text," + + "state_func text," + + "state_type text," + + "PRIMARY KEY ((keyspace_name), aggregate_name, signature))"); + private static CFMetaData compile(String name, String description, String schema) { return CFMetaData.compile(String.format(schema, name), NAME) @@ -276,22 +400,26 @@ public final class SystemKeyspace private static Tables tables() { - return Tables.builder() - .add(LegacySchemaTables.All) - .add(BuiltIndexes, - Hints, - Batchlog, - Paxos, - Local, - Peers, - PeerEvents, - RangeXfers, - CompactionsInProgress, - CompactionHistory, - SSTableActivity, - SizeEstimates, - AvailableRanges) - .build(); + return Tables.of(BuiltIndexes, + Hints, + Batchlog, + Paxos, + Local, + Peers, + PeerEvents, + RangeXfers, + CompactionsInProgress, + CompactionHistory, + SSTableActivity, + SizeEstimates, + AvailableRanges, + LegacyKeyspaces, + LegacyColumnfamilies, + LegacyColumns, + LegacyTriggers, + LegacyUsertypes, + LegacyFunctions, + LegacyAggregates); } private static Functions functions() @@ -322,7 +450,7 @@ public final class SystemKeyspace public static void finishStartup() { persistLocalMetadata(); - LegacySchemaTables.saveSystemKeyspaceSchema(); + SchemaKeyspace.saveSystemKeyspacesSchema(); } private static void persistLocalMetadata() @@ -366,7 +494,7 @@ public final class SystemKeyspace */ public static UUID startCompaction(ColumnFamilyStore cfs, Iterable toCompact) { - if (NAME.equals(cfs.keyspace.getName())) + if (Schema.isSystemKeyspace(cfs.keyspace.getName())) return null; UUID compactionId = UUIDGen.getTimeUUID(); diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index c326881375..555140f18c 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -2245,7 +2245,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted return builder().build(); } + public static Functions of(Function... funs) + { + return builder().add(funs).build(); + } + public Iterator iterator() { return functions.values().iterator(); @@ -200,6 +205,8 @@ public final class Functions implements Iterable private Builder() { + // we need deterministic iteration order; otherwise Functions.equals() breaks down + functions.orderValuesBy((f1, f2) -> Integer.compare(f1.hashCode(), f2.hashCode())); } public Functions build() diff --git a/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java new file mode 100644 index 0000000000..a0491943a3 --- /dev/null +++ b/src/java/org/apache/cassandra/schema/LegacySchemaMigrator.java @@ -0,0 +1,804 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.schema; + +import java.nio.ByteBuffer; +import java.util.*; +import java.util.stream.Collectors; + +import com.google.common.collect.HashMultimap; +import com.google.common.collect.ImmutableList; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.cache.CachingOptions; +import org.apache.cassandra.config.*; +import org.apache.cassandra.cql3.*; +import org.apache.cassandra.cql3.functions.FunctionName; +import org.apache.cassandra.cql3.functions.UDAggregate; +import org.apache.cassandra.cql3.functions.UDFunction; +import org.apache.cassandra.db.*; +import org.apache.cassandra.db.marshal.*; +import org.apache.cassandra.db.rows.RowIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIterators; +import org.apache.cassandra.exceptions.InvalidRequestException; +import org.apache.cassandra.io.compress.CompressionParameters; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.concurrent.OpOrder; + +import static java.lang.String.format; +import static org.apache.cassandra.utils.ByteBufferUtil.bytes; +import static org.apache.cassandra.utils.FBUtilities.fromJsonMap; + +/** + * This majestic class performs migration from legacy (pre-3.0) system.schema_* schema tables to the new and glorious + * system_schema keyspace. + * + * The goal is to not lose any information in the migration - including the timestamps. + */ +@SuppressWarnings("deprecation") +public final class LegacySchemaMigrator +{ + private LegacySchemaMigrator() + { + } + + private static final Logger logger = LoggerFactory.getLogger(LegacySchemaMigrator.class); + + static final List LegacySchemaTables = + ImmutableList.of(SystemKeyspace.LegacyKeyspaces, + SystemKeyspace.LegacyColumnfamilies, + SystemKeyspace.LegacyColumns, + SystemKeyspace.LegacyTriggers, + SystemKeyspace.LegacyUsertypes, + SystemKeyspace.LegacyFunctions, + SystemKeyspace.LegacyAggregates); + + public static void migrate() + { + // read metadata from the legacy schema tables + Collection keyspaces = readSchema(); + + // if already upgraded, or starting a new 3.0 node, abort early + if (keyspaces.isEmpty()) + { + unloadLegacySchemaTables(); + return; + } + + // write metadata to the new schema tables + logger.info("Moving {} keyspaces from legacy schema tables to the new schema keyspace ({})", + keyspaces.size(), + SchemaKeyspace.NAME); + keyspaces.forEach(LegacySchemaMigrator::storeKeyspaceInNewSchemaTables); + + // flush the new tables before truncating the old ones + SchemaKeyspace.flush(); + + // truncate the original tables (will be snapshotted now, and will have been snapshotted by pre-flight checks) + logger.info("Truncating legacy schema tables"); + truncateLegacySchemaTables(); + + // remove legacy schema tables from Schema, so that their presence doesn't give the users any wrong ideas + unloadLegacySchemaTables(); + + logger.info("Completed migration of legacy schema tables"); + } + + static void unloadLegacySchemaTables() + { + KeyspaceMetadata systemKeyspace = Schema.instance.getKSMetaData(SystemKeyspace.NAME); + + Tables systemTables = systemKeyspace.tables; + for (CFMetaData table : LegacySchemaTables) + systemTables = systemTables.without(table.cfName); + + LegacySchemaTables.forEach(Schema.instance::unload); + + Schema.instance.setKeyspaceMetadata(systemKeyspace.withSwapped(systemTables)); + } + + private static void truncateLegacySchemaTables() + { + LegacySchemaTables.forEach(table -> Schema.instance.getColumnFamilyStoreInstance(table.cfId).truncateBlocking()); + } + + private static void storeKeyspaceInNewSchemaTables(Keyspace keyspace) + { + Mutation mutation = SchemaKeyspace.makeCreateKeyspaceMutation(keyspace.name, keyspace.params, keyspace.timestamp); + + for (Table table : keyspace.tables) + SchemaKeyspace.addTableToSchemaMutation(table.metadata, table.timestamp, true, mutation); + + for (Type type : keyspace.types) + SchemaKeyspace.addTypeToSchemaMutation(type.metadata, type.timestamp, mutation); + + for (Function function : keyspace.functions) + SchemaKeyspace.addFunctionToSchemaMutation(function.metadata, function.timestamp, mutation); + + for (Aggregate aggregate : keyspace.aggregates) + SchemaKeyspace.addAggregateToSchemaMutation(aggregate.metadata, aggregate.timestamp, mutation); + + mutation.apply(); + } + + /* + * Read all keyspaces metadata (including nested tables, types, and functions), with their modification timestamps + */ + private static Collection readSchema() + { + String query = format("SELECT keyspace_name FROM %s.%s", SystemKeyspace.NAME, SystemKeyspace.LEGACY_KEYSPACES); + Collection keyspaceNames = new ArrayList<>(); + query(query).forEach(row -> keyspaceNames.add(row.getString("keyspace_name"))); + keyspaceNames.removeAll(Schema.SYSTEM_KEYSPACE_NAMES); + + Collection keyspaces = new ArrayList<>(); + keyspaceNames.forEach(name -> keyspaces.add(readKeyspace(name))); + return keyspaces; + } + + private static Keyspace readKeyspace(String keyspaceName) + { + long timestamp = readKeyspaceTimestamp(keyspaceName); + KeyspaceParams params = readKeyspaceParams(keyspaceName); + + Collection tables = readTables(keyspaceName); + Collection types = readTypes(keyspaceName); + Collection functions = readFunctions(keyspaceName); + Collection aggregates = readAggregates(keyspaceName); + + return new Keyspace(timestamp, keyspaceName, params, tables, types, functions, aggregates); + } + + /* + * Reading keyspace params + */ + + private static long readKeyspaceTimestamp(String keyspaceName) + { + String query = format("SELECT writeTime(durable_writes) AS timestamp FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_KEYSPACES); + return query(query, keyspaceName).one().getLong("timestamp"); + } + + private static KeyspaceParams readKeyspaceParams(String keyspaceName) + { + String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_KEYSPACES); + UntypedResultSet.Row row = query(query, keyspaceName).one(); + + boolean durableWrites = row.getBoolean("durable_writes"); + + Map replication = new HashMap<>(); + replication.putAll(fromJsonMap(row.getString("strategy_options"))); + replication.put(KeyspaceParams.Replication.CLASS, row.getString("strategy_class")); + + return KeyspaceParams.create(durableWrites, replication); + } + + /* + * Reading tables + */ + + private static Collection
readTables(String keyspaceName) + { + String query = format("SELECT columnfamily_name FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_COLUMNFAMILIES); + Collection tableNames = new ArrayList<>(); + query(query, keyspaceName).forEach(row -> tableNames.add(row.getString("columnfamily_name"))); + + Collection
tables = new ArrayList<>(); + tableNames.forEach(name -> tables.add(readTable(keyspaceName, name))); + return tables; + } + + private static Table readTable(String keyspaceName, String tableName) + { + long timestamp = readTableTimestamp(keyspaceName, tableName); + CFMetaData metadata = readTableMetadata(keyspaceName, tableName); + return new Table(timestamp, metadata); + } + + private static long readTableTimestamp(String keyspaceName, String tableName) + { + String query = format("SELECT writeTime(type) AS timestamp FROM %s.%s WHERE keyspace_name = ? AND columnfamily_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_COLUMNFAMILIES); + return query(query, keyspaceName, tableName).one().getLong("timestamp"); + } + + private static CFMetaData readTableMetadata(String keyspaceName, String tableName) + { + String tableQuery = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND columnfamily_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_COLUMNFAMILIES); + UntypedResultSet.Row tableRow = query(tableQuery, keyspaceName, tableName).one(); + + String columnsQuery = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND columnfamily_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_COLUMNS); + UntypedResultSet columnRows = query(columnsQuery, keyspaceName, tableName); + + String triggersQuery = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND columnfamily_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_TRIGGERS); + UntypedResultSet triggerRows = query(triggersQuery, keyspaceName, tableName); + + return decodeTableMetadata(tableRow, columnRows, triggerRows); + } + + private static CFMetaData decodeTableMetadata(UntypedResultSet.Row tableRow, + UntypedResultSet columnRows, + UntypedResultSet triggerRows) + { + String ksName = tableRow.getString("keyspace_name"); + String cfName = tableRow.getString("columnfamily_name"); + + AbstractType rawComparator = TypeParser.parse(tableRow.getString("comparator")); + AbstractType subComparator = tableRow.has("subcomparator") ? TypeParser.parse(tableRow.getString("subcomparator")) : null; + + boolean isSuper = "super".equals(tableRow.getString("type").toLowerCase()); + boolean isDense = tableRow.getBoolean("is_dense"); + boolean isCompound = rawComparator instanceof CompositeType; + + // We don't really use the default validator but as we have it for backward compatibility, we use it to know if it's a counter table + AbstractType defaultValidator = TypeParser.parse(tableRow.getString("default_validator")); + boolean isCounter = defaultValidator instanceof CounterColumnType; + + /* + * With CASSANDRA-5202 we stopped inferring the cf id from the combination of keyspace/table names, + * and started storing the generated uuids in system.schema_columnfamilies. + * + * In 3.0 we SHOULD NOT see tables like that (2.0-created, non-upgraded). + * But in the off-chance that we do, we generate the deterministic uuid here. + */ + UUID cfId = tableRow.has("cf_id") + ? tableRow.getUUID("cf_id") + : CFMetaData.generateLegacyCfId(ksName, cfName); + + boolean isCQLTable = !isSuper && !isDense && isCompound; + boolean isStaticCompactTable = !isDense && !isCompound; + + // Internally, compact tables have a specific layout, see CompactTables. But when upgrading from + // previous versions, they may not have the expected schema, so detect if we need to upgrade and do + // it in createColumnsFromColumnRows. + // We can remove this once we don't support upgrade from versions < 3.0. + boolean needsUpgrade = !isCQLTable && checkNeedsUpgrade(columnRows, isSuper, isStaticCompactTable); + + List columnDefs = createColumnsFromColumnRows(columnRows, + ksName, + cfName, + rawComparator, + subComparator, + isSuper, + isCQLTable, + isStaticCompactTable, + needsUpgrade); + + if (needsUpgrade) + addDefinitionForUpgrade(columnDefs, ksName, cfName, isStaticCompactTable, isSuper, rawComparator, subComparator, defaultValidator); + + CFMetaData cfm = CFMetaData.create(ksName, cfName, cfId, isDense, isCompound, isSuper, isCounter, columnDefs); + + cfm.readRepairChance(tableRow.getDouble("read_repair_chance")); + cfm.dcLocalReadRepairChance(tableRow.getDouble("local_read_repair_chance")); + cfm.gcGraceSeconds(tableRow.getInt("gc_grace_seconds")); + cfm.minCompactionThreshold(tableRow.getInt("min_compaction_threshold")); + cfm.maxCompactionThreshold(tableRow.getInt("max_compaction_threshold")); + if (tableRow.has("comment")) + cfm.comment(tableRow.getString("comment")); + if (tableRow.has("memtable_flush_period_in_ms")) + cfm.memtableFlushPeriod(tableRow.getInt("memtable_flush_period_in_ms")); + cfm.caching(CachingOptions.fromString(tableRow.getString("caching"))); + if (tableRow.has("default_time_to_live")) + cfm.defaultTimeToLive(tableRow.getInt("default_time_to_live")); + if (tableRow.has("speculative_retry")) + cfm.speculativeRetry(CFMetaData.SpeculativeRetry.fromString(tableRow.getString("speculative_retry"))); + cfm.compactionStrategyClass(CFMetaData.createCompactionStrategy(tableRow.getString("compaction_strategy_class"))); + cfm.compressionParameters(CompressionParameters.create(fromJsonMap(tableRow.getString("compression_parameters")))); + cfm.compactionStrategyOptions(fromJsonMap(tableRow.getString("compaction_strategy_options"))); + + if (tableRow.has("min_index_interval")) + cfm.minIndexInterval(tableRow.getInt("min_index_interval")); + + if (tableRow.has("max_index_interval")) + cfm.maxIndexInterval(tableRow.getInt("max_index_interval")); + + if (tableRow.has("bloom_filter_fp_chance")) + cfm.bloomFilterFpChance(tableRow.getDouble("bloom_filter_fp_chance")); + else + cfm.bloomFilterFpChance(cfm.getBloomFilterFpChance()); + + if (tableRow.has("dropped_columns")) + { + Map types = tableRow.has("dropped_columns_types") + ? tableRow.getMap("dropped_columns_types", UTF8Type.instance, UTF8Type.instance) + : Collections.emptyMap(); + addDroppedColumns(cfm, tableRow.getMap("dropped_columns", UTF8Type.instance, LongType.instance), types); + } + + triggerRows.forEach(row -> cfm.addTriggerDefinition(readTrigger(row))); + + return cfm; + } + + // Should only be called on compact tables + private static boolean checkNeedsUpgrade(UntypedResultSet defs, boolean isSuper, boolean isStaticCompactTable) + { + if (isSuper) + { + // Check if we've added the "supercolumn map" column yet or not + for (UntypedResultSet.Row row : defs) + if (row.getString("column_name").isEmpty()) + return false; + return true; + } + + // For static compact tables, we need to upgrade if the regular definitions haven't been converted to static yet, + // i.e. if we don't have a static definition yet. + if (isStaticCompactTable) + return !hasKind(defs, ColumnDefinition.Kind.STATIC); + + // For dense compact tables, we need to upgrade if we don't have a compact value definition + return !hasKind(defs, ColumnDefinition.Kind.REGULAR); + } + + private static void addDefinitionForUpgrade(List defs, + String ksName, + String cfName, + boolean isStaticCompactTable, + boolean isSuper, + AbstractType rawComparator, + AbstractType subComparator, + AbstractType defaultValidator) + { + CompactTables.DefaultNames names = CompactTables.defaultNameGenerator(defs); + + if (isSuper) + { + defs.add(ColumnDefinition.regularDef(ksName, cfName, CompactTables.SUPER_COLUMN_MAP_COLUMN_STR, MapType.getInstance(subComparator, defaultValidator, true), null)); + } + else if (isStaticCompactTable) + { + defs.add(ColumnDefinition.clusteringKeyDef(ksName, cfName, names.defaultClusteringName(), rawComparator, null)); + defs.add(ColumnDefinition.regularDef(ksName, cfName, names.defaultCompactValueName(), defaultValidator, null)); + } + else + { + // For dense compact tables, we get here if we don't have a compact value column, in which case we should add it + // (we use EmptyType to recognize that the compact value was not declared by the use (see CreateTableStatement too)) + defs.add(ColumnDefinition.regularDef(ksName, cfName, names.defaultCompactValueName(), EmptyType.instance, null)); + } + } + + private static boolean hasKind(UntypedResultSet defs, ColumnDefinition.Kind kind) + { + for (UntypedResultSet.Row row : defs) + { + if (deserializeKind(row.getString("type")) == kind) + return true; + } + return false; + } + + private static void addDroppedColumns(CFMetaData cfm, Map droppedTimes, Map types) + { + for (Map.Entry entry : droppedTimes.entrySet()) + { + String name = entry.getKey(); + long time = entry.getValue(); + AbstractType type = types.containsKey(name) ? TypeParser.parse(types.get(name)) : null; + cfm.getDroppedColumns().put(ColumnIdentifier.getInterned(name, true), new CFMetaData.DroppedColumn(type, time)); + } + } + + private static List createColumnsFromColumnRows(UntypedResultSet rows, + String keyspace, + String table, + AbstractType rawComparator, + AbstractType rawSubComparator, + boolean isSuper, + boolean isCQLTable, + boolean isStaticCompactTable, + boolean needsUpgrade) + { + List columns = new ArrayList<>(); + for (UntypedResultSet.Row row : rows) + columns.add(createColumnFromColumnRow(row, keyspace, table, rawComparator, rawSubComparator, isSuper, isCQLTable, isStaticCompactTable, needsUpgrade)); + return columns; + } + + private static ColumnDefinition createColumnFromColumnRow(UntypedResultSet.Row row, + String keyspace, + String table, + AbstractType rawComparator, + AbstractType rawSubComparator, + boolean isSuper, + boolean isCQLTable, + boolean isStaticCompactTable, + boolean needsUpgrade) + { + ColumnDefinition.Kind kind = deserializeKind(row.getString("type")); + if (needsUpgrade && isStaticCompactTable && kind == ColumnDefinition.Kind.REGULAR) + kind = ColumnDefinition.Kind.STATIC; + + Integer componentIndex = null; + if (row.has("component_index")) + componentIndex = row.getInt("component_index"); + + // Note: we save the column name as string, but we should not assume that it is an UTF8 name, we + // we need to use the comparator fromString method + AbstractType comparator = isCQLTable + ? UTF8Type.instance + : CompactTables.columnDefinitionComparator(kind, isSuper, rawComparator, rawSubComparator); + ColumnIdentifier name = ColumnIdentifier.getInterned(comparator.fromString(row.getString("column_name")), comparator); + + AbstractType validator = parseType(row.getString("validator")); + + IndexType indexType = null; + if (row.has("index_type")) + indexType = IndexType.valueOf(row.getString("index_type")); + + Map indexOptions = null; + if (row.has("index_options")) + indexOptions = fromJsonMap(row.getString("index_options")); + + String indexName = null; + if (row.has("index_name")) + indexName = row.getString("index_name"); + + return new ColumnDefinition(keyspace, table, name, validator, indexType, indexOptions, indexName, componentIndex, kind); + } + + private static ColumnDefinition.Kind deserializeKind(String kind) + { + if ("clustering_key".equalsIgnoreCase(kind)) + return ColumnDefinition.Kind.CLUSTERING_COLUMN; + if ("compact_value".equalsIgnoreCase(kind)) + return ColumnDefinition.Kind.REGULAR; + return Enum.valueOf(ColumnDefinition.Kind.class, kind.toUpperCase()); + } + + private static TriggerDefinition readTrigger(UntypedResultSet.Row row) + { + String name = row.getString("trigger_name"); + String classOption = row.getMap("trigger_options", UTF8Type.instance, UTF8Type.instance).get("class"); + return new TriggerDefinition(name, classOption); + } + + /* + * Reading user types + */ + + private static Collection readTypes(String keyspaceName) + { + String query = format("SELECT type_name FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_USERTYPES); + Collection typeNames = new ArrayList<>(); + query(query, keyspaceName).forEach(row -> typeNames.add(row.getString("type_name"))); + + Collection types = new ArrayList<>(); + typeNames.forEach(name -> types.add(readType(keyspaceName, name))); + return types; + } + + private static Type readType(String keyspaceName, String typeName) + { + long timestamp = readTypeTimestamp(keyspaceName, typeName); + UserType metadata = readTypeMetadata(keyspaceName, typeName); + return new Type(timestamp, metadata); + } + + /* + * Unfortunately there is not a single REGULAR column in system.schema_usertypes, so annoyingly we cannot + * use the writeTime() CQL function, and must resort to a lower level. + */ + private static long readTypeTimestamp(String keyspaceName, String typeName) + { + ColumnFamilyStore store = org.apache.cassandra.db.Keyspace.open(SystemKeyspace.NAME) + .getColumnFamilyStore(SystemKeyspace.LEGACY_USERTYPES); + + ClusteringComparator comparator = store.metadata.comparator; + Slices slices = Slices.with(comparator, Slice.make(comparator, typeName)); + int nowInSec = FBUtilities.nowInSeconds(); + DecoratedKey key = StorageService.getPartitioner().decorateKey(AsciiType.instance.fromString(keyspaceName)); + SinglePartitionReadCommand command = SinglePartitionSliceCommand.create(store.metadata, nowInSec, key, slices); + + try (OpOrder.Group op = store.readOrdering.start(); + RowIterator partition = UnfilteredRowIterators.filter(command.queryMemtableAndDisk(store, op), nowInSec)) + { + return partition.next().primaryKeyLivenessInfo().timestamp(); + } + } + + private static UserType readTypeMetadata(String keyspaceName, String typeName) + { + String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND type_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_USERTYPES); + UntypedResultSet.Row row = query(query, keyspaceName, typeName).one(); + + List names = + row.getList("field_names", UTF8Type.instance) + .stream() + .map(ByteBufferUtil::bytes) + .collect(Collectors.toList()); + + List> types = + row.getList("field_types", UTF8Type.instance) + .stream() + .map(LegacySchemaMigrator::parseType) + .collect(Collectors.toList()); + + return new UserType(keyspaceName, bytes(typeName), names, types); + } + + /* + * Reading UDFs + */ + + private static Collection readFunctions(String keyspaceName) + { + String query = format("SELECT function_name, signature FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_FUNCTIONS); + HashMultimap> functionSignatures = HashMultimap.create(); + query(query, keyspaceName).forEach(row -> + { + functionSignatures.put(row.getString("function_name"), row.getList("signature", UTF8Type.instance)); + }); + + Collection functions = new ArrayList<>(); + functionSignatures.entries().forEach(pair -> functions.add(readFunction(keyspaceName, pair.getKey(), pair.getValue()))); + return functions; + } + + private static Function readFunction(String keyspaceName, String functionName, List signature) + { + long timestamp = readFunctionTimestamp(keyspaceName, functionName, signature); + UDFunction metadata = readFunctionMetadata(keyspaceName, functionName, signature); + return new Function(timestamp, metadata); + } + + private static long readFunctionTimestamp(String keyspaceName, String functionName, List signature) + { + String query = format("SELECT writeTime(return_type) AS timestamp " + + "FROM %s.%s " + + "WHERE keyspace_name = ? AND function_name = ? AND signature = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_FUNCTIONS); + return query(query, keyspaceName, functionName, signature).one().getLong("timestamp"); + } + + private static UDFunction readFunctionMetadata(String keyspaceName, String functionName, List signature) + { + String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND function_name = ? AND signature = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_FUNCTIONS); + UntypedResultSet.Row row = query(query, keyspaceName, functionName, signature).one(); + + FunctionName name = new FunctionName(keyspaceName, functionName); + + List argNames = new ArrayList<>(); + if (row.has("argument_names")) + for (String arg : row.getList("argument_names", UTF8Type.instance)) + argNames.add(new ColumnIdentifier(arg, true)); + + List> argTypes = new ArrayList<>(); + if (row.has("argument_types")) + for (String type : row.getList("argument_types", UTF8Type.instance)) + argTypes.add(parseType(type)); + + AbstractType returnType = parseType(row.getString("return_type")); + + String language = row.getString("language"); + String body = row.getString("body"); + boolean calledOnNullInput = row.getBoolean("called_on_null_input"); + + try + { + return UDFunction.create(name, argNames, argTypes, returnType, calledOnNullInput, language, body); + } + catch (InvalidRequestException e) + { + return UDFunction.createBrokenFunction(name, argNames, argTypes, returnType, calledOnNullInput, language, body, e); + } + } + + /* + * Reading UDAs + */ + + private static Collection readAggregates(String keyspaceName) + { + String query = format("SELECT aggregate_name, signature FROM %s.%s WHERE keyspace_name = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_AGGREGATES); + HashMultimap> aggregateSignatures = HashMultimap.create(); + query(query, keyspaceName).forEach(row -> + { + aggregateSignatures.put(row.getString("aggregate_name"), row.getList("signature", UTF8Type.instance)); + }); + + Collection aggregates = new ArrayList<>(); + aggregateSignatures.entries().forEach(pair -> aggregates.add(readAggregate(keyspaceName, pair.getKey(), pair.getValue()))); + return aggregates; + } + + private static Aggregate readAggregate(String keyspaceName, String aggregateName, List signature) + { + long timestamp = readAggregateTimestamp(keyspaceName, aggregateName, signature); + UDAggregate metadata = readAggregateMetadata(keyspaceName, aggregateName, signature); + return new Aggregate(timestamp, metadata); + } + + private static long readAggregateTimestamp(String keyspaceName, String aggregateName, List signature) + { + String query = format("SELECT writeTime(return_type) AS timestamp " + + "FROM %s.%s " + + "WHERE keyspace_name = ? AND aggregate_name = ? AND signature = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_AGGREGATES); + return query(query, keyspaceName, aggregateName, signature).one().getLong("timestamp"); + } + + private static UDAggregate readAggregateMetadata(String keyspaceName, String functionName, List signature) + { + String query = format("SELECT * FROM %s.%s WHERE keyspace_name = ? AND function_name = ? AND signature = ?", + SystemKeyspace.NAME, + SystemKeyspace.LEGACY_AGGREGATES); + UntypedResultSet.Row row = query(query, keyspaceName, functionName, signature).one(); + + FunctionName name = new FunctionName(keyspaceName, functionName); + + List types = row.getList("argument_types", UTF8Type.instance); + + List> argTypes = new ArrayList<>(); + if (types != null) + { + argTypes = new ArrayList<>(types.size()); + for (String type : types) + argTypes.add(parseType(type)); + } + + AbstractType returnType = parseType(row.getString("return_type")); + + FunctionName stateFunc = parseAggregateFunctionName(keyspaceName, row.getString("state_func")); + FunctionName finalFunc = row.has("final_func") ? parseAggregateFunctionName(keyspaceName, row.getString("final_func")) : null; + AbstractType stateType = row.has("state_type") ? parseType(row.getString("state_type")) : null; + ByteBuffer initcond = row.has("initcond") ? row.getBytes("initcond") : null; + + try + { + return UDAggregate.create(name, argTypes, returnType, stateFunc, finalFunc, stateType, initcond); + } + catch (InvalidRequestException reason) + { + return UDAggregate.createBroken(name, argTypes, returnType, initcond, reason); + } + } + + private static FunctionName parseAggregateFunctionName(String ksName, String func) + { + int i = func.indexOf('.'); + + // function name can be abbreviated (pre 2.2rc2) - it is in the same keyspace as the aggregate + if (i == -1) + return new FunctionName(ksName, func); + + String ks = func.substring(0, i); + String f = func.substring(i + 1); + + // only aggregate's function keyspace and system keyspace are allowed + assert ks.equals(ksName) || ks.equals(SystemKeyspace.NAME); + + return new FunctionName(ks, f); + } + + private static UntypedResultSet query(String query, Object... values) + { + return QueryProcessor.executeOnceInternal(query, values); + } + + private static AbstractType parseType(String str) + { + return TypeParser.parse(str); + } + + private static final class Keyspace + { + final long timestamp; + final String name; + final KeyspaceParams params; + final Collection
tables; + final Collection types; + final Collection functions; + final Collection aggregates; + + Keyspace(long timestamp, + String name, + KeyspaceParams params, + Collection
tables, + Collection types, + Collection functions, + Collection aggregates) + { + this.timestamp = timestamp; + this.name = name; + this.params = params; + this.tables = tables; + this.types = types; + this.functions = functions; + this.aggregates = aggregates; + } + } + + private static final class Table + { + final long timestamp; + final CFMetaData metadata; + + Table(long timestamp, CFMetaData metadata) + { + this.timestamp = timestamp; + this.metadata = metadata; + } + } + + private static final class Type + { + final long timestamp; + final UserType metadata; + + Type(long timestamp, UserType metadata) + { + this.timestamp = timestamp; + this.metadata = metadata; + } + } + + private static final class Function + { + final long timestamp; + final UDFunction metadata; + + Function(long timestamp, UDFunction metadata) + { + this.timestamp = timestamp; + this.metadata = metadata; + } + } + + private static final class Aggregate + { + final long timestamp; + final UDAggregate metadata; + + Aggregate(long timestamp, UDAggregate metadata) + { + this.timestamp = timestamp; + this.metadata = metadata; + } + } +} diff --git a/src/java/org/apache/cassandra/schema/LegacySchemaTables.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java similarity index 82% rename from src/java/org/apache/cassandra/schema/LegacySchemaTables.java rename to src/java/org/apache/cassandra/schema/SchemaKeyspace.java index c8e163c81f..0e40ed2b3a 100644 --- a/src/java/org/apache/cassandra/schema/LegacySchemaTables.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -25,6 +25,7 @@ import java.util.*; import java.util.concurrent.TimeUnit; import java.util.function.Function; +import com.google.common.collect.ImmutableList; import com.google.common.collect.MapDifference; import com.google.common.collect.Maps; import org.slf4j.Logger; @@ -32,15 +33,15 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.cache.CachingOptions; import org.apache.cassandra.config.*; -import org.apache.cassandra.cql3.*; -import org.apache.cassandra.cql3.functions.AbstractFunction; -import org.apache.cassandra.cql3.functions.FunctionName; -import org.apache.cassandra.cql3.functions.UDFunction; -import org.apache.cassandra.cql3.functions.UDAggregate; +import org.apache.cassandra.cql3.ColumnIdentifier; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.cql3.functions.*; +import org.apache.cassandra.db.ClusteringComparator; import org.apache.cassandra.db.*; -import org.apache.cassandra.db.rows.*; import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.db.partitions.*; +import org.apache.cassandra.db.rows.*; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.io.compress.CompressionParameters; @@ -53,24 +54,29 @@ import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal; import static org.apache.cassandra.utils.FBUtilities.fromJsonMap; import static org.apache.cassandra.utils.FBUtilities.json; -/** system.schema_* tables used to store keyspace/table/type attributes prior to C* 3.0 */ -public final class LegacySchemaTables +/** + * system_schema.* tables and methods for manipulating them. + */ +public final class SchemaKeyspace { - private LegacySchemaTables() + private SchemaKeyspace() { } - private static final Logger logger = LoggerFactory.getLogger(LegacySchemaTables.class); + private static final Logger logger = LoggerFactory.getLogger(SchemaKeyspace.class); - public static final String KEYSPACES = "schema_keyspaces"; - public static final String COLUMNFAMILIES = "schema_columnfamilies"; - public static final String COLUMNS = "schema_columns"; - public static final String TRIGGERS = "schema_triggers"; - public static final String USERTYPES = "schema_usertypes"; - public static final String FUNCTIONS = "schema_functions"; - public static final String AGGREGATES = "schema_aggregates"; + public static final String NAME = "system_schema"; - public static final List ALL = Arrays.asList(KEYSPACES, COLUMNFAMILIES, COLUMNS, TRIGGERS, USERTYPES, FUNCTIONS, AGGREGATES); + public static final String KEYSPACES = "keyspaces"; + public static final String TABLES = "tables"; + public static final String COLUMNS = "columns"; + public static final String TRIGGERS = "triggers"; + public static final String TYPES = "types"; + public static final String FUNCTIONS = "functions"; + public static final String AGGREGATES = "aggregates"; + + public static final List ALL = + ImmutableList.of(KEYSPACES, TABLES, COLUMNS, TRIGGERS, TYPES, FUNCTIONS, AGGREGATES); private static final CFMetaData Keyspaces = compile(KEYSPACES, @@ -78,17 +84,15 @@ public final class LegacySchemaTables "CREATE TABLE %s (" + "keyspace_name text," + "durable_writes boolean," - + "strategy_class text," - + "strategy_options text," - + "PRIMARY KEY ((keyspace_name))) " - + "WITH COMPACT STORAGE"); + + "replication map," + + "PRIMARY KEY ((keyspace_name)))"); - private static final CFMetaData Columnfamilies = - compile(COLUMNFAMILIES, + private static final CFMetaData Tables = + compile(TABLES, "table definitions", "CREATE TABLE %s (" + "keyspace_name text," - + "columnfamily_name text," + + "table_name text," + "bloom_filter_fp_chance double," + "caching text," + "cf_id uuid," // post-2.1 UUID cfid @@ -114,14 +118,14 @@ public final class LegacySchemaTables + "speculative_retry text," + "subcomparator text," + "type text," - + "PRIMARY KEY ((keyspace_name), columnfamily_name))"); + + "PRIMARY KEY ((keyspace_name), table_name))"); private static final CFMetaData Columns = compile(COLUMNS, "column definitions", "CREATE TABLE %s (" + "keyspace_name text," - + "columnfamily_name text," + + "table_name text," + "column_name text," + "component_index int," + "index_name text," @@ -129,20 +133,20 @@ public final class LegacySchemaTables + "index_type text," + "type text," + "validator text," - + "PRIMARY KEY ((keyspace_name), columnfamily_name, column_name))"); + + "PRIMARY KEY ((keyspace_name), table_name, column_name))"); private static final CFMetaData Triggers = compile(TRIGGERS, "trigger definitions", "CREATE TABLE %s (" + "keyspace_name text," - + "columnfamily_name text," + + "table_name text," + "trigger_name text," + "trigger_options map," - + "PRIMARY KEY ((keyspace_name), columnfamily_name, trigger_name))"); + + "PRIMARY KEY ((keyspace_name), table_name, trigger_name))"); - private static final CFMetaData Usertypes = - compile(USERTYPES, + private static final CFMetaData Types = + compile(TYPES, "user defined type definitions", "CREATE TABLE %s (" + "keyspace_name text," @@ -181,32 +185,45 @@ public final class LegacySchemaTables + "state_type text," + "PRIMARY KEY ((keyspace_name), aggregate_name, signature))"); - public static final List All = Arrays.asList(Keyspaces, Columnfamilies, Columns, Triggers, Usertypes, Functions, Aggregates); + public static final List All = + ImmutableList.of(Keyspaces, Tables, Columns, Triggers, Types, Functions, Aggregates); private static CFMetaData compile(String name, String description, String schema) { - return CFMetaData.compile(String.format(schema, name), SystemKeyspace.NAME) + return CFMetaData.compile(String.format(schema, name), NAME) .comment(description) .gcGraceSeconds((int) TimeUnit.DAYS.toSeconds(7)); } - /** add entries to system.schema_* for the hardcoded system definitions */ - public static void saveSystemKeyspaceSchema() + public static KeyspaceMetadata metadata() { - KeyspaceMetadata keyspace = Schema.instance.getKSMetaData(SystemKeyspace.NAME); - long timestamp = FBUtilities.timestampMicros(); - // delete old, possibly obsolete entries in schema tables - for (String table : ALL) - { - executeOnceInternal(String.format("DELETE FROM system.%s USING TIMESTAMP ? WHERE keyspace_name = ?", table), - timestamp, - keyspace.name); - } - // (+1 to timestamp to make sure we don't get shadowed by the tombstones we just added) - makeCreateKeyspaceMutation(keyspace, timestamp + 1).apply(); + return KeyspaceMetadata.create(NAME, KeyspaceParams.local(), org.apache.cassandra.schema.Tables.of(All)); } - public static Collection readSchemaFromSystemTables() + /** + * Add entries to system_schema.* for the hardcoded system keyspaces + */ + public static void saveSystemKeyspacesSchema() + { + KeyspaceMetadata system = Schema.instance.getKSMetaData(SystemKeyspace.NAME); + KeyspaceMetadata schema = Schema.instance.getKSMetaData(NAME); + + long timestamp = FBUtilities.timestampMicros(); + + // delete old, possibly obsolete entries in schema tables + for (String schemaTable : ALL) + { + String query = String.format("DELETE FROM %s.%s USING TIMESTAMP ? WHERE keyspace_name = ?", NAME, schemaTable); + for (String systemKeyspace : Schema.SYSTEM_KEYSPACE_NAMES) + executeOnceInternal(query, timestamp, systemKeyspace); + } + + // (+1 to timestamp to make sure we don't get shadowed by the tombstones we just added) + makeCreateKeyspaceMutation(system, timestamp + 1).apply(); + makeCreateKeyspaceMutation(schema, timestamp + 1).apply(); + } + + public static List readSchemaFromSystemTables() { ReadCommand cmd = getReadCommandForTableSchema(KEYSPACES); try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator schema = cmd.executeInternal(orderGroup)) @@ -222,8 +239,8 @@ public final class LegacySchemaTables DecoratedKey key = partition.partitionKey(); - readSchemaPartitionForKeyspaceAndApply(USERTYPES, key, - types -> readSchemaPartitionForKeyspaceAndApply(COLUMNFAMILIES, key, + readSchemaPartitionForKeyspaceAndApply(TYPES, key, + types -> readSchemaPartitionForKeyspaceAndApply(TABLES, key, tables -> readSchemaPartitionForKeyspaceAndApply(FUNCTIONS, key, functions -> readSchemaPartitionForKeyspaceAndApply(AGGREGATES, key, aggregates -> keyspaces.add(createKeyspaceFromSchemaPartitions(partition, tables, types, functions, aggregates))))) @@ -234,16 +251,15 @@ public final class LegacySchemaTables } } - public static void truncateSchemaTables() + public static void truncate() { - for (String table : ALL) - getSchemaCFS(table).truncateBlocking(); + ALL.forEach(table -> getSchemaCFS(table).truncateBlocking()); } - private static void flushSchemaTables() + static void flush() { - for (String table : ALL) - SystemKeyspace.forceBlockingFlush(table); + if (!Boolean.getBoolean("cassandra.unsafesystem")) + ALL.forEach(table -> FBUtilities.waitOnFuture(getSchemaCFS(table).forceFlush())); } /** @@ -265,7 +281,8 @@ public final class LegacySchemaTables for (String table : ALL) { ReadCommand cmd = getReadCommandForTableSchema(table); - try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator schema = cmd.executeInternal(orderGroup)) + try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); + PartitionIterator schema = cmd.executeInternal(orderGroup)) { while (schema.hasNext()) { @@ -286,7 +303,7 @@ public final class LegacySchemaTables */ private static ColumnFamilyStore getSchemaCFS(String schemaTableName) { - return Keyspace.open(SystemKeyspace.NAME).getColumnFamilyStore(schemaTableName); + return Keyspace.open(NAME).getColumnFamilyStore(schemaTableName); } /** @@ -325,7 +342,7 @@ public final class LegacySchemaTables Mutation mutation = mutationMap.get(key); if (mutation == null) { - mutation = new Mutation(SystemKeyspace.NAME, key); + mutation = new Mutation(NAME, key); mutationMap.put(key, mutation); } @@ -399,7 +416,8 @@ public final class LegacySchemaTables private static boolean isSystemKeyspaceSchemaPartition(DecoratedKey partitionKey) { - return getSchemaKSKey(SystemKeyspace.NAME).equals(partitionKey.getKey()); + return getSchemaKSKey(SystemKeyspace.NAME).equals(partitionKey.getKey()) || + getSchemaKSKey(NAME).equals(partitionKey.getKey()); } /** @@ -426,21 +444,20 @@ public final class LegacySchemaTables // current state of the schema Map oldKeyspaces = readSchemaForKeyspaces(KEYSPACES, keyspaces); - Map oldColumnFamilies = readSchemaForKeyspaces(COLUMNFAMILIES, keyspaces); - Map oldTypes = readSchemaForKeyspaces(USERTYPES, keyspaces); + Map oldColumnFamilies = readSchemaForKeyspaces(TABLES, keyspaces); + Map oldTypes = readSchemaForKeyspaces(TYPES, keyspaces); Map oldFunctions = readSchemaForKeyspaces(FUNCTIONS, keyspaces); Map oldAggregates = readSchemaForKeyspaces(AGGREGATES, keyspaces); - for (Mutation mutation : mutations) - mutation.apply(); + mutations.forEach(Mutation::apply); if (doFlush) - flushSchemaTables(); + flush(); // with new data applied Map newKeyspaces = readSchemaForKeyspaces(KEYSPACES, keyspaces); - Map newColumnFamilies = readSchemaForKeyspaces(COLUMNFAMILIES, keyspaces); - Map newTypes = readSchemaForKeyspaces(USERTYPES, keyspaces); + Map newColumnFamilies = readSchemaForKeyspaces(TABLES, keyspaces); + Map newTypes = readSchemaForKeyspaces(TYPES, keyspaces); Map newFunctions = readSchemaForKeyspaces(FUNCTIONS, keyspaces); Map newAggregates = readSchemaForKeyspaces(AGGREGATES, keyspaces); @@ -451,8 +468,7 @@ public final class LegacySchemaTables mergeAggregates(oldAggregates, newAggregates); // it is safe to drop a keyspace only when all nested ColumnFamilies where deleted - for (String keyspaceToDrop : keyspacesToDrop) - Schema.instance.dropKeyspace(keyspaceToDrop); + keyspacesToDrop.forEach(Schema.instance::dropKeyspace); } private static Set mergeKeyspaces(Map before, Map after) @@ -487,7 +503,7 @@ public final class LegacySchemaTables { public void onDropped(UntypedResultSet.Row oldRow) { - Schema.instance.dropTable(oldRow.getString("keyspace_name"), oldRow.getString("columnfamily_name")); + Schema.instance.dropTable(oldRow.getString("keyspace_name"), oldRow.getString("table_name")); } public void onAdded(UntypedResultSet.Row newRow) @@ -497,7 +513,7 @@ public final class LegacySchemaTables public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow) { - Schema.instance.updateTable(newRow.getString("keyspace_name"), newRow.getString("columnfamily_name")); + Schema.instance.updateTable(newRow.getString("keyspace_name"), newRow.getString("table_name")); } }); } @@ -567,9 +583,9 @@ public final class LegacySchemaTables public interface Differ { - public void onDropped(UntypedResultSet.Row oldRow); - public void onAdded(UntypedResultSet.Row newRow); - public void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow); + void onDropped(UntypedResultSet.Row oldRow); + void onAdded(UntypedResultSet.Row newRow); + void onUpdated(UntypedResultSet.Row oldRow, UntypedResultSet.Row newRow); } private static void diffSchema(Map before, Map after, Differ differ) @@ -640,29 +656,27 @@ public final class LegacySchemaTables * Keyspace metadata serialization/deserialization. */ - public static Mutation makeCreateKeyspaceMutation(KeyspaceMetadata keyspace, long timestamp) + public static Mutation makeCreateKeyspaceMutation(String name, KeyspaceParams params, long timestamp) { - return makeCreateKeyspaceMutation(keyspace, timestamp, true); + RowUpdateBuilder adder = new RowUpdateBuilder(Keyspaces, timestamp, name).clustering(); + + adder.add("durable_writes", params.durableWrites); + + adder.resetCollection("replication"); + for (Map.Entry option : params.replication.asMap().entrySet()) + adder.addMapEntry("replication", option.getKey(), option.getValue()); + + return adder.build(); } - public static Mutation makeCreateKeyspaceMutation(KeyspaceMetadata keyspace, long timestamp, boolean withTablesAndTypesAndFunctions) + public static Mutation makeCreateKeyspaceMutation(KeyspaceMetadata keyspace, long timestamp) { - // Note that because Keyspaces is a COMPACT TABLE, we're really only setting static columns internally and shouldn't set any clustering. - RowUpdateBuilder adder = new RowUpdateBuilder(Keyspaces, timestamp, keyspace.name); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); - adder.add("durable_writes", keyspace.params.durableWrites); - adder.add("strategy_class", keyspace.params.replication.klass.getName()); - adder.add("strategy_options", json(keyspace.params.replication.options)); - - Mutation mutation = adder.build(); - - if (withTablesAndTypesAndFunctions) - { - keyspace.tables.forEach(table -> addTableToSchemaMutation(table, timestamp, true, mutation)); - keyspace.types.forEach(type -> addTypeToSchemaMutation(type, timestamp, mutation)); - keyspace.functions.udfs().forEach(udf -> addFunctionToSchemaMutation(udf, timestamp, mutation)); - keyspace.functions.udas().forEach(uda -> addAggregateToSchemaMutation(uda, timestamp, mutation)); - } + keyspace.tables.forEach(table -> addTableToSchemaMutation(table, timestamp, true, mutation)); + keyspace.types.forEach(type -> addTypeToSchemaMutation(type, timestamp, mutation)); + keyspace.functions.udfs().forEach(udf -> addFunctionToSchemaMutation(udf, timestamp, mutation)); + keyspace.functions.udas().forEach(uda -> addAggregateToSchemaMutation(uda, timestamp, mutation)); return mutation; } @@ -670,10 +684,9 @@ public final class LegacySchemaTables public static Mutation makeDropKeyspaceMutation(KeyspaceMetadata keyspace, long timestamp) { int nowInSec = FBUtilities.nowInSeconds(); - Mutation mutation = new Mutation(SystemKeyspace.NAME, getSchemaKSDecoratedKey(keyspace.name)); + Mutation mutation = new Mutation(NAME, getSchemaKSDecoratedKey(keyspace.name)); for (CFMetaData schemaTable : All) mutation.add(PartitionUpdate.fullPartitionDelete(schemaTable, mutation.key(), timestamp, nowInSec)); - mutation.add(PartitionUpdate.fullPartitionDelete(SystemKeyspace.BuiltIndexes, mutation.key(), timestamp, nowInSec)); return mutation; } @@ -704,14 +717,13 @@ public final class LegacySchemaTables private static KeyspaceParams createKeyspaceParamsFromSchemaPartition(RowIterator partition) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, KEYSPACES); + String query = String.format("SELECT * FROM %s.%s", NAME, KEYSPACES); UntypedResultSet.Row row = QueryProcessor.resultify(query, partition).one(); - Map replicationMap = new HashMap<>(); - replicationMap.putAll(fromJsonMap(row.getString("strategy_options"))); - replicationMap.put("class", row.getString("strategy_class")); + boolean durableWrites = row.getBoolean("durable_writes"); + Map replication= row.getMap("replication", UTF8Type.instance, UTF8Type.instance); - return KeyspaceParams.create(row.getBoolean("durable_writes"), replicationMap); + return KeyspaceParams.create(durableWrites, replication); } /* @@ -721,23 +733,23 @@ public final class LegacySchemaTables public static Mutation makeCreateTypeMutation(KeyspaceMetadata keyspace, UserType type, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); addTypeToSchemaMutation(type, timestamp, mutation); return mutation; } - private static void addTypeToSchemaMutation(UserType type, long timestamp, Mutation mutation) + static void addTypeToSchemaMutation(UserType type, long timestamp, Mutation mutation) { - RowUpdateBuilder adder = new RowUpdateBuilder(Usertypes, timestamp, mutation) - .clustering(type.name); + RowUpdateBuilder adder = new RowUpdateBuilder(Types, timestamp, mutation) + .clustering(type.getNameAsString()); - adder.resetCollection("field_names"); - adder.resetCollection("field_types"); + adder.resetCollection("field_names") + .resetCollection("field_types"); for (int i = 0; i < type.size(); i++) { - adder.addListEntry("field_names", type.fieldName(i)); - adder.addListEntry("field_types", type.fieldType(i).toString()); + adder.addListEntry("field_names", type.fieldName(i)) + .addListEntry("field_types", type.fieldType(i).toString()); } adder.build(); @@ -746,14 +758,14 @@ public final class LegacySchemaTables public static Mutation dropTypeFromSchemaMutation(KeyspaceMetadata keyspace, UserType type, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); - return RowUpdateBuilder.deleteRow(Usertypes, timestamp, mutation, type.name); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); + return RowUpdateBuilder.deleteRow(Types, timestamp, mutation, type.name); } private static Types createTypesFromPartition(RowIterator partition) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, USERTYPES); - Types.Builder types = Types.builder(); + String query = String.format("SELECT * FROM %s.%s", SchemaKeyspace.NAME, TYPES); + Types.Builder types = org.apache.cassandra.schema.Types.builder(); QueryProcessor.resultify(query, partition).forEach(row -> types.add(createTypeFromRow(row))); return types.build(); } @@ -783,51 +795,51 @@ public final class LegacySchemaTables public static Mutation makeCreateTableMutation(KeyspaceMetadata keyspace, CFMetaData table, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); addTableToSchemaMutation(table, timestamp, true, mutation); return mutation; } - private static void addTableToSchemaMutation(CFMetaData table, long timestamp, boolean withColumnsAndTriggers, Mutation mutation) + static void addTableToSchemaMutation(CFMetaData table, long timestamp, boolean withColumnsAndTriggers, Mutation mutation) { // For property that can be null (and can be changed), we insert tombstones, to make sure // we don't keep a property the user has removed - RowUpdateBuilder adder = new RowUpdateBuilder(Columnfamilies, timestamp, mutation) + RowUpdateBuilder adder = new RowUpdateBuilder(Tables, timestamp, mutation) .clustering(table.cfName); - adder.add("cf_id", table.cfId); - adder.add("type", table.isSuper() ? "Super" : "Standard"); + adder.add("cf_id", table.cfId) + .add("type", table.isSuper() ? "Super" : "Standard"); if (table.isSuper()) { // We need to continue saving the comparator and subcomparator separatly, otherwise // we won't know at deserialization if the subcomparator should be taken into account // TODO: we should implement an on-start migration if we want to get rid of that. - adder.add("comparator", table.comparator.subtype(0).toString()); - adder.add("subcomparator", ((MapType)table.compactValueColumn().type).getKeysType().toString()); + adder.add("comparator", table.comparator.subtype(0).toString()) + .add("subcomparator", ((MapType)table.compactValueColumn().type).getKeysType().toString()); } else { adder.add("comparator", LegacyLayout.makeLegacyComparator(table).toString()); } - adder.add("bloom_filter_fp_chance", table.getBloomFilterFpChance()); - adder.add("caching", table.getCaching().toString()); - adder.add("comment", table.getComment()); - adder.add("compaction_strategy_class", table.compactionStrategyClass.getName()); - adder.add("compaction_strategy_options", json(table.compactionStrategyOptions)); - adder.add("compression_parameters", json(table.compressionParameters.asThriftOptions())); - adder.add("default_time_to_live", table.getDefaultTimeToLive()); - adder.add("gc_grace_seconds", table.getGcGraceSeconds()); - adder.add("key_validator", table.getKeyValidator().toString()); - adder.add("local_read_repair_chance", table.getDcLocalReadRepairChance()); - adder.add("max_compaction_threshold", table.getMaxCompactionThreshold()); - adder.add("max_index_interval", table.getMaxIndexInterval()); - adder.add("memtable_flush_period_in_ms", table.getMemtableFlushPeriod()); - adder.add("min_compaction_threshold", table.getMinCompactionThreshold()); - adder.add("min_index_interval", table.getMinIndexInterval()); - adder.add("read_repair_chance", table.getReadRepairChance()); - adder.add("speculative_retry", table.getSpeculativeRetry().toString()); + adder.add("bloom_filter_fp_chance", table.getBloomFilterFpChance()) + .add("caching", table.getCaching().toString()) + .add("comment", table.getComment()) + .add("compaction_strategy_class", table.compactionStrategyClass.getName()) + .add("compaction_strategy_options", json(table.compactionStrategyOptions)) + .add("compression_parameters", json(table.compressionParameters.asThriftOptions())) + .add("default_time_to_live", table.getDefaultTimeToLive()) + .add("gc_grace_seconds", table.getGcGraceSeconds()) + .add("key_validator", table.getKeyValidator().toString()) + .add("local_read_repair_chance", table.getDcLocalReadRepairChance()) + .add("max_compaction_threshold", table.getMaxCompactionThreshold()) + .add("max_index_interval", table.getMaxIndexInterval()) + .add("memtable_flush_period_in_ms", table.getMemtableFlushPeriod()) + .add("min_compaction_threshold", table.getMinCompactionThreshold()) + .add("min_index_interval", table.getMinIndexInterval()) + .add("read_repair_chance", table.getReadRepairChance()) + .add("speculative_retry", table.getSpeculativeRetry().toString()); for (Map.Entry entry : table.getDroppedColumns().entrySet()) { @@ -860,7 +872,7 @@ public final class LegacySchemaTables long timestamp, boolean fromThrift) { - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); addTableToSchemaMutation(newTable, timestamp, false, mutation); @@ -902,9 +914,9 @@ public final class LegacySchemaTables public static Mutation makeDropTableMutation(KeyspaceMetadata keyspace, CFMetaData table, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); - RowUpdateBuilder.deleteRow(Columnfamilies, timestamp, mutation, table.cfName); + RowUpdateBuilder.deleteRow(Tables, timestamp, mutation, table.cfName); for (ColumnDefinition column : table.allColumns()) dropColumnFromSchemaMutation(table, column, timestamp, mutation); @@ -912,16 +924,12 @@ public final class LegacySchemaTables for (TriggerDefinition trigger : table.getTriggers().values()) dropTriggerFromSchemaMutation(table, trigger, timestamp, mutation); - // TODO: get rid of in #6717 - for (String indexName : Keyspace.open(keyspace.name).getColumnFamilyStore(table.cfName).getBuiltIndexes()) - RowUpdateBuilder.deleteRow(SystemKeyspace.BuiltIndexes, timestamp, mutation, indexName); - return mutation; } public static CFMetaData createTableFromName(String keyspace, String table) { - return readSchemaPartitionForTableAndApply(COLUMNFAMILIES, keyspace, table, partition -> + return readSchemaPartitionForTableAndApply(TABLES, keyspace, table, partition -> { if (partition.isEmpty()) throw new RuntimeException(String.format("%s:%s not found in the schema definitions keyspace.", keyspace, table)); @@ -935,27 +943,27 @@ public final class LegacySchemaTables */ private static Tables createTablesFromTablesPartition(RowIterator partition) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, COLUMNFAMILIES); - Tables.Builder tables = Tables.builder(); + String query = String.format("SELECT * FROM %s.%s", NAME, TABLES); + Tables.Builder tables = org.apache.cassandra.schema.Tables.builder(); QueryProcessor.resultify(query, partition).forEach(row -> tables.add(createTableFromTableRow(row))); return tables.build(); } public static CFMetaData createTableFromTablePartitionAndColumnsPartition(RowIterator serializedTable, RowIterator serializedColumns) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, COLUMNFAMILIES); + String query = String.format("SELECT * FROM %s.%s", NAME, TABLES); return createTableFromTableRowAndColumnsPartition(QueryProcessor.resultify(query, serializedTable).one(), serializedColumns); } private static CFMetaData createTableFromTableRowAndColumnsPartition(UntypedResultSet.Row tableRow, RowIterator serializedColumns) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, COLUMNS); + String query = String.format("SELECT * FROM %s.%s", NAME, COLUMNS); return createTableFromTableRowAndColumnRows(tableRow, QueryProcessor.resultify(query, serializedColumns)); } private static CFMetaData createTableFromTablePartition(RowIterator partition) { - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, COLUMNFAMILIES); + String query = String.format("SELECT * FROM %s.%s", NAME, TABLES); return createTableFromTableRow(QueryProcessor.resultify(query, partition).one()); } @@ -967,12 +975,12 @@ public final class LegacySchemaTables private static CFMetaData createTableFromTableRow(UntypedResultSet.Row result) { String ksName = result.getString("keyspace_name"); - String cfName = result.getString("columnfamily_name"); + String cfName = result.getString("table_name"); CFMetaData cfm = readSchemaPartitionForTableAndApply(COLUMNS, ksName, cfName, partition -> createTableFromTableRowAndColumnsPartition(result, partition)); readSchemaPartitionForTableAndApply(TRIGGERS, ksName, cfName, - partition -> { createTriggersFromTriggersPartition(partition).forEach(trigger -> cfm.addTriggerDefinition(trigger)); return null; } + partition -> { createTriggersFromTriggersPartition(partition).forEach(cfm::addTriggerDefinition); return null; } ); return cfm; @@ -982,12 +990,12 @@ public final class LegacySchemaTables UntypedResultSet serializedColumnDefinitions) { String ksName = result.getString("keyspace_name"); - String cfName = result.getString("columnfamily_name"); + String cfName = result.getString("table_name"); AbstractType rawComparator = TypeParser.parse(result.getString("comparator")); AbstractType subComparator = result.has("subcomparator") ? TypeParser.parse(result.getString("subcomparator")) : null; - boolean isSuper = result.getString("type").toLowerCase().equals("super"); + boolean isSuper = "super".equals(result.getString("type").toLowerCase()); boolean isDense = result.getBoolean("is_dense"); boolean isCompound = rawComparator instanceof CompositeType; @@ -995,10 +1003,7 @@ public final class LegacySchemaTables AbstractType defaultValidator = TypeParser.parse(result.getString("default_validator")); boolean isCounter = defaultValidator instanceof CounterColumnType; - // if we are upgrading, we use id generated from names initially - UUID cfId = result.has("cf_id") - ? result.getUUID("cf_id") - : CFMetaData.generateLegacyCfId(ksName, cfName); + UUID cfId = result.getUUID("cf_id"); boolean isCQLTable = !isSuper && !isDense && isCompound; boolean isStaticCompactTable = !isDense && !isCompound; @@ -1007,7 +1012,7 @@ public final class LegacySchemaTables // previous versions, they may not have the expected schema, so detect if we need to upgrade and do // it in createColumnsFromColumnRows. // We can remove this once we don't support upgrade from versions < 3.0. - boolean needsUpgrade = isCQLTable ? false : checkNeedsUpgrade(serializedColumnDefinitions, isSuper, isStaticCompactTable); + boolean needsUpgrade = !isCQLTable && checkNeedsUpgrade(serializedColumnDefinitions, isSuper, isStaticCompactTable); List columnDefs = createColumnsFromColumnRows(serializedColumnDefinitions, ksName, @@ -1145,14 +1150,13 @@ public final class LegacySchemaTables RowUpdateBuilder adder = new RowUpdateBuilder(Columns, timestamp, mutation) .clustering(table.cfName, column.name.toString()); - adder.add("validator", column.type.toString()); - adder.add("type", serializeKind(column.kind, table.isDense())); - adder.add("component_index", column.isOnAllComponents() ? null : column.position()); - adder.add("index_name", column.getIndexName()); - adder.add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString()); - adder.add("index_options", json(column.getIndexOptions())); - - adder.build(); + adder.add("validator", column.type.toString()) + .add("type", serializeKind(column.kind, table.isDense())) + .add("component_index", column.isOnAllComponents() ? null : column.position()) + .add("index_name", column.getIndexName()) + .add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString()) + .add("index_options", json(column.getIndexOptions())) + .build(); } private static String serializeKind(ColumnDefinition.Kind kind, boolean isDense) @@ -1169,9 +1173,9 @@ public final class LegacySchemaTables public static ColumnDefinition.Kind deserializeKind(String kind) { - if (kind.equalsIgnoreCase("clustering_key")) + if ("clustering_key".equalsIgnoreCase(kind)) return ColumnDefinition.Kind.CLUSTERING_COLUMN; - if (kind.equalsIgnoreCase("compact_value")) + if ("compact_value".equalsIgnoreCase(kind)) return ColumnDefinition.Kind.REGULAR; return Enum.valueOf(ColumnDefinition.Kind.class, kind.toUpperCase()); } @@ -1266,7 +1270,7 @@ public final class LegacySchemaTables private static List createTriggersFromTriggersPartition(RowIterator partition) { List triggers = new ArrayList<>(); - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, TRIGGERS); + String query = String.format("SELECT * FROM %s.%s", NAME, TRIGGERS); for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition)) { String name = row.getString("trigger_name"); @@ -1283,42 +1287,44 @@ public final class LegacySchemaTables public static Mutation makeCreateFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); addFunctionToSchemaMutation(function, timestamp, mutation); return mutation; } - private static void addFunctionToSchemaMutation(UDFunction function, long timestamp, Mutation mutation) + static void addFunctionToSchemaMutation(UDFunction function, long timestamp, Mutation mutation) { RowUpdateBuilder adder = new RowUpdateBuilder(Functions, timestamp, mutation) .clustering(function.name().name, functionSignatureWithTypes(function)); - adder.add("body", function.body()); - adder.add("language", function.language()); - adder.add("return_type", function.returnType().toString()); - adder.add("called_on_null_input", function.isCalledOnNullInput()); + adder.add("body", function.body()) + .add("language", function.language()) + .add("return_type", function.returnType().toString()) + .add("called_on_null_input", function.isCalledOnNullInput()); + + adder.resetCollection("argument_names") + .resetCollection("argument_types"); - adder.resetCollection("argument_names"); - adder.resetCollection("argument_types"); for (int i = 0; i < function.argNames().size(); i++) { adder.addListEntry("argument_names", function.argNames().get(i).bytes); adder.addListEntry("argument_types", function.argTypes().get(i).toString()); } + adder.build(); } public static Mutation makeDropFunctionMutation(KeyspaceMetadata keyspace, UDFunction function, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); return RowUpdateBuilder.deleteRow(Functions, timestamp, mutation, function.name().name, functionSignatureWithTypes(function)); } private static Collection createFunctionsFromFunctionsPartition(RowIterator partition) { List functions = new ArrayList<>(); - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, FUNCTIONS); + String query = String.format("SELECT * FROM %s.%s", NAME, FUNCTIONS); for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition)) functions.add(createFunctionFromFunctionRow(row)); return functions; @@ -1384,19 +1390,21 @@ public final class LegacySchemaTables public static Mutation makeCreateAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); addAggregateToSchemaMutation(aggregate, timestamp, mutation); return mutation; } - private static void addAggregateToSchemaMutation(UDAggregate aggregate, long timestamp, Mutation mutation) + static void addAggregateToSchemaMutation(UDAggregate aggregate, long timestamp, Mutation mutation) { RowUpdateBuilder adder = new RowUpdateBuilder(Aggregates, timestamp, mutation) .clustering(aggregate.name().name, functionSignatureWithTypes(aggregate)); adder.resetCollection("argument_types"); - adder.add("return_type", aggregate.returnType().toString()); - adder.add("state_func", aggregate.stateFunction().name().toString()); + + adder.add("return_type", aggregate.returnType().toString()) + .add("state_func", aggregate.stateFunction().name().toString()); + if (aggregate.stateType() != null) adder.add("state_type", aggregate.stateType().toString()); if (aggregate.finalFunction() != null) @@ -1413,7 +1421,7 @@ public final class LegacySchemaTables private static Collection createAggregatesFromAggregatesPartition(RowIterator partition) { List aggregates = new ArrayList<>(); - String query = String.format("SELECT * FROM %s.%s", SystemKeyspace.NAME, AGGREGATES); + String query = String.format("SELECT * FROM %s.%s", NAME, AGGREGATES); for (UntypedResultSet.Row row : QueryProcessor.resultify(query, partition)) aggregates.add(createAggregateFromAggregateRow(row)); return aggregates; @@ -1441,8 +1449,8 @@ public final class LegacySchemaTables AbstractType returnType = parseType(row.getString("return_type")); - FunctionName stateFunc = aggregateParseFunctionName(ksName, row.getString("state_func")); - FunctionName finalFunc = row.has("final_func") ? aggregateParseFunctionName(ksName, row.getString("final_func")) : null; + FunctionName stateFunc = aggregateParseFunctionName(row.getString("state_func")); + FunctionName finalFunc = row.has("final_func") ? aggregateParseFunctionName(row.getString("final_func")) : null; AbstractType stateType = row.has("state_type") ? parseType(row.getString("state_type")) : null; ByteBuffer initcond = row.has("initcond") ? row.getBytes("initcond") : null; @@ -1456,27 +1464,18 @@ public final class LegacySchemaTables } } - private static FunctionName aggregateParseFunctionName(String ksName, String func) + private static FunctionName aggregateParseFunctionName(String fqn) { - int i = func.indexOf('.'); - - // function name can be abbreviated (pre 2.2rc2) - it is in the same keyspace as the aggregate - if (i == -1) - return new FunctionName(ksName, func); - - String ks = func.substring(0, i); - String f = func.substring(i + 1); - - // only aggregate's function keyspace and system keyspace are allowed - assert ks.equals(ksName) || ks.equals(SystemKeyspace.NAME); - - return new FunctionName(ks, f); + int i = fqn.indexOf('.'); + String keyspace = fqn.substring(0, i); + String function = fqn.substring(i + 1); + return new FunctionName(keyspace, function); } public static Mutation makeDropAggregateMutation(KeyspaceMetadata keyspace, UDAggregate aggregate, long timestamp) { // Include the serialized keyspace in case the target node missed a CREATE KEYSPACE migration (see CASSANDRA-5631). - Mutation mutation = makeCreateKeyspaceMutation(keyspace, timestamp, false); + Mutation mutation = makeCreateKeyspaceMutation(keyspace.name, keyspace.params, timestamp); return RowUpdateBuilder.deleteRow(Aggregates, timestamp, mutation, aggregate.name().name, functionSignatureWithTypes(aggregate)); } diff --git a/src/java/org/apache/cassandra/schema/Types.java b/src/java/org/apache/cassandra/schema/Types.java index 982e3f5fd8..0a7bb4f593 100644 --- a/src/java/org/apache/cassandra/schema/Types.java +++ b/src/java/org/apache/cassandra/schema/Types.java @@ -51,6 +51,11 @@ public final class Types implements Iterable return builder().build(); } + public static Types of(UserType... types) + { + return builder().add(types).build(); + } + public Iterator iterator() { return types.values().iterator(); @@ -138,6 +143,13 @@ public final class Types implements Iterable return this; } + public Builder add(UserType... types) + { + for (UserType type : types) + add(type); + return this; + } + public Builder add(Iterable types) { types.forEach(this::add); diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index 8ca4d06576..58b74f005d 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -59,6 +59,7 @@ import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.metrics.CassandraMetricsRegistry; import org.apache.cassandra.metrics.StorageMetrics; +import org.apache.cassandra.schema.LegacySchemaMigrator; import org.apache.cassandra.thrift.ThriftServer; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.*; @@ -199,7 +200,15 @@ public class CassandraDaemon } }); + /* + * Migrate pre-3.0 keyspaces, tables, types, functions, and aggregates, to their new 3.0 storage. + * We don't (and can't) wait for commit log replay here, but we don't need to - all schema changes force + * explicit memtable flushes. + */ + LegacySchemaMigrator.migrate(); + StorageService.instance.populateTokenMetadata(); + // load schema from disk Schema.instance.loadFromDisk(); diff --git a/src/java/org/apache/cassandra/service/ClientState.java b/src/java/org/apache/cassandra/service/ClientState.java index a8e13d1f82..f8d39391c4 100644 --- a/src/java/org/apache/cassandra/service/ClientState.java +++ b/src/java/org/apache/cassandra/service/ClientState.java @@ -24,7 +24,6 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.atomic.AtomicLong; -import com.google.common.collect.Iterables; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,7 +37,7 @@ import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.exceptions.AuthenticationException; import org.apache.cassandra.exceptions.InvalidRequestException; import org.apache.cassandra.exceptions.UnauthorizedException; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.thrift.ThriftValidation; import org.apache.cassandra.tracing.TraceKeyspace; import org.apache.cassandra.utils.FBUtilities; @@ -61,9 +60,11 @@ public class ClientState { // We want these system cfs to be always readable to authenticated users since many tools rely on them // (nodetool, cqlsh, bulkloader, etc.) - for (String cf : Iterables.concat(Arrays.asList(SystemKeyspace.LOCAL, SystemKeyspace.PEERS), LegacySchemaTables.ALL)) + for (String cf : Arrays.asList(SystemKeyspace.LOCAL, SystemKeyspace.PEERS)) READABLE_SYSTEM_RESOURCES.add(DataResource.table(SystemKeyspace.NAME, cf)); + SchemaKeyspace.ALL.forEach(table -> READABLE_SYSTEM_RESOURCES.add(DataResource.table(SchemaKeyspace.NAME, table))); + PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getAuthenticator().protectedResources()); PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getAuthorizer().protectedResources()); PROTECTED_AUTH_RESOURCES.addAll(DatabaseDescriptor.getRoleManager().protectedResources()); @@ -309,7 +310,7 @@ public class ClientState return; // prevent system keyspace modification - if (SystemKeyspace.NAME.equalsIgnoreCase(keyspace)) + if (Schema.isSystemKeyspace(keyspace)) throw new UnauthorizedException(keyspace + " keyspace is not user-modifiable."); // allow users with sufficient privileges to alter KS level options on AUTH_KS and diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java index d979c022d7..dad6aa7cf6 100644 --- a/src/java/org/apache/cassandra/service/MigrationManager.java +++ b/src/java/org/apache/cassandra/service/MigrationManager.java @@ -44,7 +44,7 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.KeyspaceMetadata; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.WrappedRunnable; @@ -72,7 +72,7 @@ public class MigrationManager listeners.remove(listener); } - public void scheduleSchemaPull(InetAddress endpoint, EndpointState state) + public static void scheduleSchemaPull(InetAddress endpoint, EndpointState state) { VersionedValue value = state.getApplicationState(ApplicationState.SCHEMA); @@ -102,27 +102,24 @@ public class MigrationManager { // Include a delay to make sure we have a chance to apply any changes being // pushed out simultaneously. See CASSANDRA-5025 - Runnable runnable = new Runnable() + Runnable runnable = () -> { - public void run() + // grab the latest version of the schema since it may have changed again since the initial scheduling + EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); + if (epState == null) { - // grab the latest version of the schema since it may have changed again since the initial scheduling - EndpointState epState = Gossiper.instance.getEndpointStateForEndpoint(endpoint); - if (epState == null) - { - logger.debug("epState vanished for {}, not submitting migration task", endpoint); - return; - } - VersionedValue value = epState.getApplicationState(ApplicationState.SCHEMA); - UUID currentVersion = UUID.fromString(value.value); - if (Schema.instance.getVersion().equals(currentVersion)) - { - logger.debug("not submitting migration task for {} because our versions match", endpoint); - return; - } - logger.debug("submitting migration task for {}", endpoint); - submitMigrationTask(endpoint); + logger.debug("epState vanished for {}, not submitting migration task", endpoint); + return; } + VersionedValue value = epState.getApplicationState(ApplicationState.SCHEMA); + UUID currentVersion = UUID.fromString(value.value); + if (Schema.instance.getVersion().equals(currentVersion)) + { + logger.debug("not submitting migration task for {} because our versions match", endpoint); + return; + } + logger.debug("submitting migration task for {}", endpoint); + submitMigrationTask(endpoint); }; ScheduledExecutors.optionalTasks.schedule(runnable, MIGRATION_DELAY_IN_MS, TimeUnit.MILLISECONDS); } @@ -264,7 +261,7 @@ public class MigrationManager throw new AlreadyExistsException(ksm.name); logger.info(String.format("Create new Keyspace: %s", ksm)); - announce(LegacySchemaTables.makeCreateKeyspaceMutation(ksm, timestamp), announceLocally); + announce(SchemaKeyspace.makeCreateKeyspaceMutation(ksm, timestamp), announceLocally); } public static void announceNewColumnFamily(CFMetaData cfm) throws ConfigurationException @@ -283,27 +280,27 @@ public class MigrationManager throw new AlreadyExistsException(cfm.ksName, cfm.cfName); logger.info(String.format("Create new table: %s", cfm)); - announce(LegacySchemaTables.makeCreateTableMutation(ksm, cfm, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeCreateTableMutation(ksm, cfm, FBUtilities.timestampMicros()), announceLocally); } public static void announceNewType(UserType newType, boolean announceLocally) { KeyspaceMetadata ksm = Schema.instance.getKSMetaData(newType.keyspace); - announce(LegacySchemaTables.makeCreateTypeMutation(ksm, newType, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeCreateTypeMutation(ksm, newType, FBUtilities.timestampMicros()), announceLocally); } public static void announceNewFunction(UDFunction udf, boolean announceLocally) { logger.info(String.format("Create scalar function '%s'", udf.name())); KeyspaceMetadata ksm = Schema.instance.getKSMetaData(udf.name().keyspace); - announce(LegacySchemaTables.makeCreateFunctionMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeCreateFunctionMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); } public static void announceNewAggregate(UDAggregate udf, boolean announceLocally) { logger.info(String.format("Create aggregate function '%s'", udf.name())); KeyspaceMetadata ksm = Schema.instance.getKSMetaData(udf.name().keyspace); - announce(LegacySchemaTables.makeCreateAggregateMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeCreateAggregateMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); } public static void announceKeyspaceUpdate(KeyspaceMetadata ksm) throws ConfigurationException @@ -320,7 +317,7 @@ public class MigrationManager throw new ConfigurationException(String.format("Cannot update non existing keyspace '%s'.", ksm.name)); logger.info(String.format("Update Keyspace '%s' From %s To %s", ksm.name, oldKsm, ksm)); - announce(LegacySchemaTables.makeCreateKeyspaceMutation(ksm, FBUtilities.timestampMicros(), false), announceLocally); + announce(SchemaKeyspace.makeCreateKeyspaceMutation(ksm.name, ksm.params, FBUtilities.timestampMicros()), announceLocally); } public static void announceColumnFamilyUpdate(CFMetaData cfm, boolean fromThrift) throws ConfigurationException @@ -340,7 +337,7 @@ public class MigrationManager oldCfm.validateCompatility(cfm); logger.info(String.format("Update table '%s/%s' From %s To %s", cfm.ksName, cfm.cfName, oldCfm, cfm)); - announce(LegacySchemaTables.makeUpdateTableMutation(ksm, oldCfm, cfm, FBUtilities.timestampMicros(), fromThrift), announceLocally); + announce(SchemaKeyspace.makeUpdateTableMutation(ksm, oldCfm, cfm, FBUtilities.timestampMicros(), fromThrift), announceLocally); } public static void announceTypeUpdate(UserType updatedType, boolean announceLocally) @@ -360,7 +357,7 @@ public class MigrationManager throw new ConfigurationException(String.format("Cannot drop non existing keyspace '%s'.", ksName)); logger.info(String.format("Drop Keyspace '%s'", oldKsm.name)); - announce(LegacySchemaTables.makeDropKeyspaceMutation(oldKsm, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeDropKeyspaceMutation(oldKsm, FBUtilities.timestampMicros()), announceLocally); } public static void announceColumnFamilyDrop(String ksName, String cfName) throws ConfigurationException @@ -376,7 +373,7 @@ public class MigrationManager KeyspaceMetadata ksm = Schema.instance.getKSMetaData(ksName); logger.info(String.format("Drop table '%s/%s'", oldCfm.ksName, oldCfm.cfName)); - announce(LegacySchemaTables.makeDropTableMutation(ksm, oldCfm, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeDropTableMutation(ksm, oldCfm, FBUtilities.timestampMicros()), announceLocally); } public static void announceTypeDrop(UserType droppedType) @@ -387,21 +384,21 @@ public class MigrationManager public static void announceTypeDrop(UserType droppedType, boolean announceLocally) { KeyspaceMetadata ksm = Schema.instance.getKSMetaData(droppedType.keyspace); - announce(LegacySchemaTables.dropTypeFromSchemaMutation(ksm, droppedType, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.dropTypeFromSchemaMutation(ksm, droppedType, FBUtilities.timestampMicros()), announceLocally); } public static void announceFunctionDrop(UDFunction udf, boolean announceLocally) { logger.info(String.format("Drop scalar function overload '%s' args '%s'", udf.name(), udf.argTypes())); KeyspaceMetadata ksm = Schema.instance.getKSMetaData(udf.name().keyspace); - announce(LegacySchemaTables.makeDropFunctionMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeDropFunctionMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); } public static void announceAggregateDrop(UDAggregate udf, boolean announceLocally) { logger.info(String.format("Drop aggregate function overload '%s' args '%s'", udf.name(), udf.argTypes())); KeyspaceMetadata ksm = Schema.instance.getKSMetaData(udf.name().keyspace); - announce(LegacySchemaTables.makeDropAggregateMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); + announce(SchemaKeyspace.makeDropAggregateMutation(ksm, udf, FBUtilities.timestampMicros()), announceLocally); } /** @@ -414,7 +411,7 @@ public class MigrationManager { try { - LegacySchemaTables.mergeSchema(Collections.singletonList(schema), false); + SchemaKeyspace.mergeSchema(Collections.singletonList(schema), false); } catch (IOException e) { @@ -442,7 +439,7 @@ public class MigrationManager { protected void runMayThrow() throws IOException, ConfigurationException { - LegacySchemaTables.mergeSchema(schema); + SchemaKeyspace.mergeSchema(schema); } }); @@ -473,16 +470,14 @@ public class MigrationManager /** * Clear all locally stored schema information and reset schema to initial state. * Called by user (via JMX) who wants to get rid of schema disagreement. - * - * @throws IOException if schema tables truncation fails */ - public static void resetLocalSchema() throws IOException + public static void resetLocalSchema() { logger.info("Starting local schema reset..."); logger.debug("Truncating schema tables..."); - LegacySchemaTables.truncateSchemaTables(); + SchemaKeyspace.truncate(); logger.debug("Clearing local schema keyspace definitions..."); diff --git a/src/java/org/apache/cassandra/service/MigrationTask.java b/src/java/org/apache/cassandra/service/MigrationTask.java index df0b76751e..33bb35fdc8 100644 --- a/src/java/org/apache/cassandra/service/MigrationTask.java +++ b/src/java/org/apache/cassandra/service/MigrationTask.java @@ -26,12 +26,12 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.exceptions.ConfigurationException; -import org.apache.cassandra.schema.LegacySchemaTables; import org.apache.cassandra.gms.FailureDetector; import org.apache.cassandra.net.IAsyncCallback; import org.apache.cassandra.net.MessageIn; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.WrappedRunnable; @@ -72,7 +72,7 @@ class MigrationTask extends WrappedRunnable { try { - LegacySchemaTables.mergeSchema(message.payload); + SchemaKeyspace.mergeSchema(message.payload); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index bf0c6647e8..2c5a2ab160 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -1201,7 +1201,7 @@ public class StorageProxy implements StorageProxyMBean private static boolean systemKeyspaceQuery(List cmds) { for (ReadCommand cmd : cmds) - if (!cmd.metadata().ksName.equals(SystemKeyspace.NAME)) + if (!Schema.isSystemKeyspace(cmd.metadata().ksName)) return false; return true; } diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 93f69a9355..596e463854 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -2471,7 +2471,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE public int forceKeyspaceCleanup(String keyspaceName, String... columnFamilies) throws IOException, ExecutionException, InterruptedException { - if (keyspaceName.equals(SystemKeyspace.NAME)) + if (Schema.isSystemKeyspace(keyspaceName)) throw new RuntimeException("Cleanup of the system keyspace is neither necessary nor wise"); CompactionManager.AllSSTableOpStatus status = CompactionManager.AllSSTableOpStatus.SUCCESSFUL; @@ -2705,7 +2705,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE Map snapshotMap = new HashMap<>(); for (Keyspace keyspace : Keyspace.all()) { - if (SystemKeyspace.NAME.equals(keyspace.getName())) + if (Schema.isSystemKeyspace(keyspace.getName())) continue; for (ColumnFamilyStore cfStore : keyspace.getColumnFamilyStores()) @@ -2731,7 +2731,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE long total = 0; for (Keyspace keyspace : Keyspace.all()) { - if (SystemKeyspace.NAME.equals(keyspace.getName())) + if (Schema.isSystemKeyspace(keyspace.getName())) continue; for (ColumnFamilyStore cfStore : keyspace.getColumnFamilyStores()) diff --git a/src/java/org/apache/cassandra/thrift/ThriftConversion.java b/src/java/org/apache/cassandra/thrift/ThriftConversion.java index f60ea489e4..3e0c8f4a2c 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftConversion.java +++ b/src/java/org/apache/cassandra/thrift/ThriftConversion.java @@ -40,9 +40,9 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy; import org.apache.cassandra.locator.LocalStrategy; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.schema.Tables; import org.apache.cassandra.serializers.MarshalException; -import org.apache.cassandra.schema.LegacySchemaTables; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.UUIDGen; @@ -414,7 +414,7 @@ public class ThriftConversion cols.add(convertThriftCqlRow(row)); UntypedResultSet colsRows = UntypedResultSet.create(cols); - return LegacySchemaTables.createTableFromTableRowAndColumnRows(cfRow, colsRows); + return SchemaKeyspace.createTableFromTableRowAndColumnRows(cfRow, colsRows); } private static Map convertThriftCqlRow(CqlRow row) diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index dd5bf9885f..13c55aae61 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java @@ -632,8 +632,8 @@ public class ThriftValidation public static void validateKeyspaceNotSystem(String modifiedKeyspace) throws org.apache.cassandra.exceptions.InvalidRequestException { - if (modifiedKeyspace.equalsIgnoreCase(SystemKeyspace.NAME)) - throw new org.apache.cassandra.exceptions.InvalidRequestException("system keyspace is not user-modifiable"); + if (Schema.isSystemKeyspace(modifiedKeyspace)) + throw new org.apache.cassandra.exceptions.InvalidRequestException(String.format("%s keyspace is not user-modifiable", modifiedKeyspace)); } //public static IDiskAtomFilter asIFilter(SlicePredicate sp, CFMetaData metadata, ByteBuffer superColumn) diff --git a/src/java/org/apache/cassandra/tools/nodetool/Cleanup.java b/src/java/org/apache/cassandra/tools/nodetool/Cleanup.java index aa415b3541..d5c505e894 100644 --- a/src/java/org/apache/cassandra/tools/nodetool/Cleanup.java +++ b/src/java/org/apache/cassandra/tools/nodetool/Cleanup.java @@ -23,6 +23,7 @@ import io.airlift.command.Command; import java.util.ArrayList; import java.util.List; +import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.tools.NodeProbe; import org.apache.cassandra.tools.NodeTool.NodeToolCmd; @@ -41,13 +42,14 @@ public class Cleanup extends NodeToolCmd for (String keyspace : keyspaces) { - if (SystemKeyspace.NAME.equals(keyspace)) + if (Schema.isSystemKeyspace(keyspace)) continue; try { probe.forceKeyspaceCleanup(System.out, keyspace, cfnames); - } catch (Exception e) + } + catch (Exception e) { throw new RuntimeException("Error occurred during cleanup", e); } diff --git a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java index f2663bf4ef..fa415ae162 100644 --- a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java +++ b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java @@ -26,12 +26,11 @@ import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.cql3.ColumnIdentifier; import org.apache.cassandra.db.CompactTables; -import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.dht.*; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.SSTableLoader; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; public class NativeSSTableLoaderClient extends SSTableLoader.Client { @@ -100,9 +99,9 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client { Map tables = new HashMap<>(); - String query = String.format("SELECT columnfamily_name, cf_id, type, comparator, subcomparator, is_dense, default_validator FROM %s.%s WHERE keyspace_name = '%s'", - SystemKeyspace.NAME, - LegacySchemaTables.COLUMNFAMILIES, + String query = String.format("SELECT table_name, cf_id, type, comparator, subcomparator, is_dense, default_validator FROM %s.%s WHERE keyspace_name = '%s'", + SchemaKeyspace.NAME, + SchemaKeyspace.TABLES, keyspace); @@ -110,7 +109,7 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client // be safer to have a simple wrapper of the driver ResultSet/Row implementing UntypedResultSet/UntypedResultSet.Row and reuse the original method. for (Row row : session.execute(query)) { - String name = row.getString("columnfamily_name"); + String name = row.getString("table_name"); UUID id = row.getUUID("cf_id"); boolean isSuper = row.getString("type").toLowerCase().equals("super"); AbstractType rawComparator = TypeParser.parse(row.getString("comparator")); @@ -124,9 +123,9 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client boolean isCounter = defaultValidator instanceof CounterColumnType; boolean isCQLTable = !isSuper && !isDense && isCompound; - String columnsQuery = String.format("SELECT column_name, component_index, type, validator FROM %s.%s WHERE keyspace_name='%s' AND columnfamily_name='%s'", - SystemKeyspace.NAME, - LegacySchemaTables.COLUMNS, + String columnsQuery = String.format("SELECT column_name, component_index, type, validator FROM %s.%s WHERE keyspace_name='%s' AND table_name='%s'", + SchemaKeyspace.NAME, + SchemaKeyspace.COLUMNS, keyspace, name); @@ -149,7 +148,7 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client boolean isSuper, boolean isCQLTable) { - ColumnDefinition.Kind kind = LegacySchemaTables.deserializeKind(row.getString("type")); + ColumnDefinition.Kind kind = SchemaKeyspace.deserializeKind(row.getString("type")); Integer componentIndex = null; if (!row.isNull("component_index")) diff --git a/test/unit/org/apache/cassandra/SchemaLoader.java b/test/unit/org/apache/cassandra/SchemaLoader.java index bd213a539e..86e476256c 100644 --- a/test/unit/org/apache/cassandra/SchemaLoader.java +++ b/test/unit/org/apache/cassandra/SchemaLoader.java @@ -143,7 +143,7 @@ public class SchemaLoader //CFMetaData.Builder.create(ks1, "Counter1", false, false, true).build(), //CFMetaData.Builder.create(ks1, "SuperCounter1", false, false, true, true).build(), superCFMD(ks1, "SuperDirectGC", BytesType.instance).gcGraceSeconds(0), - jdbcCFMD(ks1, "JdbcInteger", IntegerType.instance).addColumnDefinition(integerColumn(ks1, "JdbcInteger")), +// jdbcCFMD(ks1, "JdbcInteger", IntegerType.instance).addColumnDefinition(integerColumn(ks1, "JdbcInteger")), jdbcCFMD(ks1, "JdbcUtf8", UTF8Type.instance).addColumnDefinition(utf8Column(ks1, "JdbcUtf8")), jdbcCFMD(ks1, "JdbcLong", LongType.instance), jdbcCFMD(ks1, "JdbcBytes", bytes), @@ -296,7 +296,7 @@ public class SchemaLoader ColumnDefinition.Kind.REGULAR); } - private static ColumnDefinition utf8Column(String ksName, String cfName) + public static ColumnDefinition utf8Column(String ksName, String cfName) { return new ColumnDefinition(ksName, cfName, diff --git a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java index 12384640b9..c0808d277c 100644 --- a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java +++ b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java @@ -33,7 +33,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.compress.*; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.thrift.CfDef; import org.apache.cassandra.thrift.ColumnDef; @@ -143,10 +143,10 @@ public class CFMetaDataTest assert before.equals(after) : String.format("%n%s%n!=%n%s", before, after); // Test schema conversion - Mutation rm = LegacySchemaTables.makeCreateTableMutation(keyspace, cfm, FBUtilities.timestampMicros()); - PartitionUpdate cfU = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, LegacySchemaTables.COLUMNFAMILIES)); - PartitionUpdate cdU = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, LegacySchemaTables.COLUMNS)); - CFMetaData newCfm = LegacySchemaTables.createTableFromTablePartitionAndColumnsPartition( + Mutation rm = SchemaKeyspace.makeCreateTableMutation(keyspace, cfm, FBUtilities.timestampMicros()); + PartitionUpdate cfU = rm.getPartitionUpdate(Schema.instance.getId(SchemaKeyspace.NAME, SchemaKeyspace.TABLES)); + PartitionUpdate cdU = rm.getPartitionUpdate(Schema.instance.getId(SchemaKeyspace.NAME, SchemaKeyspace.COLUMNS)); + CFMetaData newCfm = SchemaKeyspace.createTableFromTablePartitionAndColumnsPartition( UnfilteredRowIterators.filter(cfU.unfilteredIterator(), FBUtilities.nowInSeconds()), UnfilteredRowIterators.filter(cdU.unfilteredIterator(), FBUtilities.nowInSeconds()) ); diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java index 4cf5abf0c6..8c24cc5472 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/UFTest.java @@ -115,12 +115,12 @@ public class UFTest extends CQLTester Assert.assertEquals(1, Schema.instance.getFunctions(parseFunctionName(fSin)).size()); - assertRows(execute("SELECT function_name, language FROM system.schema_functions WHERE keyspace_name=?", KEYSPACE_PER_TEST), + assertRows(execute("SELECT function_name, language FROM system_schema.functions WHERE keyspace_name=?", KEYSPACE_PER_TEST), row(fSinName.name, "java")); dropPerTestKeyspace(); - assertRows(execute("SELECT function_name, language FROM system.schema_functions WHERE keyspace_name=?", KEYSPACE_PER_TEST)); + assertRows(execute("SELECT function_name, language FROM system_schema.functions WHERE keyspace_name=?", KEYSPACE_PER_TEST)); Assert.assertEquals(0, Schema.instance.getFunctions(fSinName).size()); } @@ -558,7 +558,7 @@ public class UFTest extends CQLTester "LANGUAGE JAVA\n" + "AS '" +functionBody + "';"); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", KEYSPACE, parseFunctionName(fName).name), row("java", functionBody)); @@ -662,7 +662,7 @@ public class UFTest extends CQLTester FunctionName fNameName = parseFunctionName(fName); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", fNameName.keyspace, fNameName.name), row("java", functionBody)); @@ -825,7 +825,7 @@ public class UFTest extends CQLTester FunctionName fNameName = parseFunctionName(fName); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", fNameName.keyspace, fNameName.name), row("java", functionBody)); @@ -856,7 +856,7 @@ public class UFTest extends CQLTester FunctionName fNameName = parseFunctionName(fName); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", fNameName.keyspace, fNameName.name), row("java", functionBody)); @@ -889,7 +889,7 @@ public class UFTest extends CQLTester FunctionName fNameName = parseFunctionName(fName); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", fNameName.keyspace, fNameName.name), row("java", functionBody)); } @@ -1792,7 +1792,7 @@ public class UFTest extends CQLTester FunctionName fNameName = parseFunctionName(fName); - assertRows(execute("SELECT language, body FROM system.schema_functions WHERE keyspace_name=? AND function_name=?", + assertRows(execute("SELECT language, body FROM system_schema.functions WHERE keyspace_name=? AND function_name=?", fNameName.keyspace, fNameName.name), row("javascript", functionBody)); diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java index 05f6c35240..66b6b3300e 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java @@ -17,6 +17,8 @@ */ package org.apache.cassandra.cql3.validation.operations; +import com.google.common.collect.ImmutableMap; + import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; @@ -129,7 +131,7 @@ public class AlterTest extends CQLTester execute("CREATE KEYSPACE ks1 WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 }"); execute("CREATE KEYSPACE ks2 WITH replication={ 'class' : 'SimpleStrategy', 'replication_factor' : 1 } AND durable_writes=false"); - assertRows(execute("SELECT keyspace_name, durable_writes FROM system.schema_keyspaces"), + assertRows(execute("SELECT keyspace_name, durable_writes FROM system_schema.keyspaces"), row("ks1", true), row(KEYSPACE, true), row(KEYSPACE_PER_TEST, true), @@ -138,18 +140,22 @@ public class AlterTest extends CQLTester execute("ALTER KEYSPACE ks1 WITH replication = { 'class' : 'NetworkTopologyStrategy', 'dc1' : 1 } AND durable_writes=False"); execute("ALTER KEYSPACE ks2 WITH durable_writes=true"); - assertRows(execute("SELECT keyspace_name, durable_writes, strategy_class FROM system.schema_keyspaces"), - row("ks1", false, "org.apache.cassandra.locator.NetworkTopologyStrategy"), - row(KEYSPACE, true, "org.apache.cassandra.locator.SimpleStrategy"), - row(KEYSPACE_PER_TEST, true, "org.apache.cassandra.locator.SimpleStrategy"), - row("ks2", true, "org.apache.cassandra.locator.SimpleStrategy")); + assertRows(execute("SELECT keyspace_name, durable_writes, replication FROM system_schema.keyspaces"), + row("ks1", false, ImmutableMap.of("class", "org.apache.cassandra.locator.NetworkTopologyStrategy", + "dc1", "1")), + row(KEYSPACE, true, ImmutableMap.of("class", "org.apache.cassandra.locator.SimpleStrategy", + "replication_factor", "1")), + row(KEYSPACE_PER_TEST, true, ImmutableMap.of("class", "org.apache.cassandra.locator.SimpleStrategy", + "replication_factor", "1")), + row("ks2", true, ImmutableMap.of("class", "org.apache.cassandra.locator.SimpleStrategy", + "replication_factor", "1"))); execute("USE ks1"); assertInvalidThrow(ConfigurationException.class, "CREATE TABLE cf1 (a int PRIMARY KEY, b int) WITH compaction = { 'min_threshold' : 4 }"); execute("CREATE TABLE cf1 (a int PRIMARY KEY, b int) WITH compaction = { 'class' : 'SizeTieredCompactionStrategy', 'min_threshold' : 7 }"); - assertRows(execute("SELECT columnfamily_name, min_compaction_threshold FROM system.schema_columnfamilies WHERE keyspace_name='ks1'"), + assertRows(execute("SELECT table_name, min_compaction_threshold FROM system_schema.tables WHERE keyspace_name='ks1'"), row("cf1", 7)); // clean-up @@ -195,8 +201,8 @@ public class AlterTest extends CQLTester // tests CASSANDRA-9565 public void testDoubleWith() throws Throwable { - String[] stmts = new String[] { "ALTER KEYSPACE WITH WITH DURABLE_WRITES = true", - "ALTER KEYSPACE ks WITH WITH DURABLE_WRITES = true" }; + String[] stmts = { "ALTER KEYSPACE WITH WITH DURABLE_WRITES = true", + "ALTER KEYSPACE ks WITH WITH DURABLE_WRITES = true" }; for (String stmt : stmts) { assertInvalidSyntaxMessage("no viable alternative at input 'WITH'", stmt); diff --git a/test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionTest.java b/test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionTest.java index 055c9d959c..19f85bf38f 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/operations/InsertUpdateIfConditionTest.java @@ -798,19 +798,19 @@ public class InsertUpdateIfConditionTest extends CQLTester // create and confirm schemaChange("CREATE TABLE IF NOT EXISTS " + fullTableName + " (id text PRIMARY KEY, value1 blob) with comment = 'foo'"); - assertRows(execute("select comment from system.schema_columnfamilies where keyspace_name = ? and columnfamily_name = ?", KEYSPACE, tableName), + assertRows(execute("select comment from system_schema.tables where keyspace_name = ? and table_name = ?", KEYSPACE, tableName), row("foo")); // unsuccessful create since it's already there, confirm settings don't change schemaChange("CREATE TABLE IF NOT EXISTS " + fullTableName + " (id text PRIMARY KEY, value2 blob)with comment = 'bar'"); - assertRows(execute("select comment from system.schema_columnfamilies where keyspace_name = ? and columnfamily_name = ?", KEYSPACE, tableName), + assertRows(execute("select comment from system_schema.tables where keyspace_name = ? and table_name = ?", KEYSPACE, tableName), row("foo")); // drop and confirm schemaChange("DROP TABLE IF EXISTS " + fullTableName); - assertEmpty(execute("select * from system.schema_columnfamilies where keyspace_name = ? and columnfamily_name = ?", KEYSPACE, tableName)); + assertEmpty(execute("select * from system_schema.tables where keyspace_name = ? and table_name = ?", KEYSPACE, tableName)); } /** diff --git a/test/unit/org/apache/cassandra/schema/DefsTest.java b/test/unit/org/apache/cassandra/schema/DefsTest.java index dcc4c51820..cf4925378c 100644 --- a/test/unit/org/apache/cassandra/schema/DefsTest.java +++ b/test/unit/org/apache/cassandra/schema/DefsTest.java @@ -515,7 +515,7 @@ public class DefsTest public void testDropIndex() throws ConfigurationException { // persist keyspace definition in the system keyspace - LegacySchemaTables.makeCreateKeyspaceMutation(Schema.instance.getKSMetaData(KEYSPACE6), FBUtilities.timestampMicros()).applyUnsafe(); + SchemaKeyspace.makeCreateKeyspaceMutation(Schema.instance.getKSMetaData(KEYSPACE6), FBUtilities.timestampMicros()).applyUnsafe(); ColumnFamilyStore cfs = Keyspace.open(KEYSPACE6).getColumnFamilyStore(TABLE1i); // insert some data. save the sstable descriptor so we can make sure it's marked for delete after the drop diff --git a/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java b/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java new file mode 100644 index 0000000000..b4149a032a --- /dev/null +++ b/test/unit/org/apache/cassandra/schema/LegacySchemaMigratorTest.java @@ -0,0 +1,549 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.schema; + +import java.nio.ByteBuffer; +import java.util.*; +import java.util.stream.Collectors; + +import com.google.common.collect.ImmutableList; +import org.junit.Test; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.cache.CachingOptions; +import org.apache.cassandra.config.*; +import org.apache.cassandra.cql3.ColumnIdentifier; +import org.apache.cassandra.cql3.functions.*; +import org.apache.cassandra.db.*; +import org.apache.cassandra.db.compaction.LeveledCompactionStrategy; +import org.apache.cassandra.db.marshal.*; + +import static java.lang.String.format; +import static junit.framework.Assert.assertEquals; +import static junit.framework.Assert.assertTrue; +import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal; +import static org.apache.cassandra.utils.ByteBufferUtil.bytes; +import static org.apache.cassandra.utils.FBUtilities.json; + +@SuppressWarnings("deprecation") +public class LegacySchemaMigratorTest +{ + private static final long TIMESTAMP = 1435908994000000L; + + private static final String KEYSPACE_PREFIX = "LegacySchemaMigratorTest"; + + /* + * 1. Write a variety of different keyspaces/tables/types/function in the legacy manner, using legacy schema tables + * 2. Run the migrator + * 3. Read all the keyspaces from the new schema tables + * 4. Make sure that we've read *exactly* the same set of keyspaces/tables/types/functions + * 5. Validate that the legacy schema tables are now empty + */ + @Test + public void testMigrate() + { + List expected = keyspaceToMigrate(); + expected.sort((k1, k2) -> k1.name.compareTo(k2.name)); + + // write the keyspaces into the legacy tables + expected.forEach(LegacySchemaMigratorTest::legacySerializeKeyspace); + + // run the migration + LegacySchemaMigrator.migrate(); + + // read back all the metadata from the new schema tables + List actual = SchemaKeyspace.readSchemaFromSystemTables(); + actual.sort((k1, k2) -> k1.name.compareTo(k2.name)); + + // make sure that we've read *exactly* the same set of keyspaces/tables/types/functions + assertEquals(expected, actual); + + // need to load back CFMetaData of those tables (CFS instances will still be loaded) + loadLegacySchemaTables(); + + // verify that nothing's left in the old schema tables + for (CFMetaData table : LegacySchemaMigrator.LegacySchemaTables) + { + String query = format("SELECT * FROM %s.%s", SystemKeyspace.NAME, table.cfName); + //noinspection ConstantConditions + assertTrue(executeOnceInternal(query).isEmpty()); + } + } + + private static void loadLegacySchemaTables() + { + KeyspaceMetadata systemKeyspace = Schema.instance.getKSMetaData(SystemKeyspace.NAME); + + Tables systemTables = systemKeyspace.tables; + for (CFMetaData table : LegacySchemaMigrator.LegacySchemaTables) + systemTables = systemTables.with(table); + + LegacySchemaMigrator.LegacySchemaTables.forEach(Schema.instance::load); + + Schema.instance.setKeyspaceMetadata(systemKeyspace.withSwapped(systemTables)); + } + + private static List keyspaceToMigrate() + { + List keyspaces = new ArrayList<>(); + + // A whole bucket of shorthand + String ks1 = KEYSPACE_PREFIX + "Keyspace1"; + String ks2 = KEYSPACE_PREFIX + "Keyspace2"; + String ks3 = KEYSPACE_PREFIX + "Keyspace3"; + String ks4 = KEYSPACE_PREFIX + "Keyspace4"; + String ks5 = KEYSPACE_PREFIX + "Keyspace5"; + String ks6 = KEYSPACE_PREFIX + "Keyspace6"; + String ks_rcs = KEYSPACE_PREFIX + "RowCacheSpace"; + String ks_nocommit = KEYSPACE_PREFIX + "NoCommitlogSpace"; + String ks_prsi = KEYSPACE_PREFIX + "PerRowSecondaryIndex"; + String ks_cql = KEYSPACE_PREFIX + "cql_keyspace"; + + // Make it easy to test compaction + Map compactionOptions = new HashMap<>(); + compactionOptions.put("tombstone_compaction_interval", "1"); + Map leveledOptions = new HashMap<>(); + leveledOptions.put("sstable_size_in_mb", "1"); + + keyspaces.add(KeyspaceMetadata.create(ks1, + KeyspaceParams.simple(1), + Tables.of(SchemaLoader.standardCFMD(ks1, "Standard1") + .compactionStrategyOptions(compactionOptions), + SchemaLoader.standardCFMD(ks1, "StandardGCGS0").gcGraceSeconds(0), + SchemaLoader.standardCFMD(ks1, "StandardLong1"), + SchemaLoader.superCFMD(ks1, "Super1", LongType.instance), + SchemaLoader.superCFMD(ks1, "Super2", UTF8Type.instance), + SchemaLoader.superCFMD(ks1, "Super5", BytesType.instance), + SchemaLoader.superCFMD(ks1, "Super6", LexicalUUIDType.instance, UTF8Type.instance), + SchemaLoader.keysIndexCFMD(ks1, "Indexed1", true), + SchemaLoader.keysIndexCFMD(ks1, "Indexed2", false), + SchemaLoader.superCFMD(ks1, "SuperDirectGC", BytesType.instance) + .gcGraceSeconds(0), + SchemaLoader.jdbcCFMD(ks1, "JdbcUtf8", UTF8Type.instance) + .addColumnDefinition(SchemaLoader.utf8Column(ks1, "JdbcUtf8")), + SchemaLoader.jdbcCFMD(ks1, "JdbcLong", LongType.instance), + SchemaLoader.jdbcCFMD(ks1, "JdbcBytes", BytesType.instance), + SchemaLoader.jdbcCFMD(ks1, "JdbcAscii", AsciiType.instance), + SchemaLoader.standardCFMD(ks1, "StandardLeveled") + .compactionStrategyClass(LeveledCompactionStrategy.class) + .compactionStrategyOptions(leveledOptions), + SchemaLoader.standardCFMD(ks1, "legacyleveled") + .compactionStrategyClass(LeveledCompactionStrategy.class) + .compactionStrategyOptions(leveledOptions), + SchemaLoader.standardCFMD(ks1, "StandardLowIndexInterval") + .minIndexInterval(8) + .maxIndexInterval(256) + .caching(CachingOptions.NONE)))); + + // Keyspace 2 + keyspaces.add(KeyspaceMetadata.create(ks2, + KeyspaceParams.simple(1), + Tables.of(SchemaLoader.standardCFMD(ks2, "Standard1"), + SchemaLoader.superCFMD(ks2, "Super3", BytesType.instance), + SchemaLoader.superCFMD(ks2, "Super4", TimeUUIDType.instance), + SchemaLoader.keysIndexCFMD(ks2, "Indexed1", true), + SchemaLoader.compositeIndexCFMD(ks2, "Indexed2", true), + SchemaLoader.compositeIndexCFMD(ks2, "Indexed3", true) + .gcGraceSeconds(0)))); + + // Keyspace 3 + keyspaces.add(KeyspaceMetadata.create(ks3, + KeyspaceParams.simple(5), + Tables.of(SchemaLoader.standardCFMD(ks3, "Standard1"), + SchemaLoader.keysIndexCFMD(ks3, "Indexed1", true)))); + + // Keyspace 4 + keyspaces.add(KeyspaceMetadata.create(ks4, + KeyspaceParams.simple(3), + Tables.of(SchemaLoader.standardCFMD(ks4, "Standard1"), + SchemaLoader.superCFMD(ks4, "Super3", BytesType.instance), + SchemaLoader.superCFMD(ks4, "Super4", TimeUUIDType.instance), + SchemaLoader.superCFMD(ks4, "Super5", TimeUUIDType.instance, BytesType.instance)))); + + // Keyspace 5 + keyspaces.add(KeyspaceMetadata.create(ks5, + KeyspaceParams.simple(2), + Tables.of(SchemaLoader.standardCFMD(ks5, "Standard1")))); + // Keyspace 6 + keyspaces.add(KeyspaceMetadata.create(ks6, + KeyspaceParams.simple(1), + Tables.of(SchemaLoader.keysIndexCFMD(ks6, "Indexed1", true)))); + + // RowCacheSpace + keyspaces.add(KeyspaceMetadata.create(ks_rcs, + KeyspaceParams.simple(1), + Tables.of(SchemaLoader.standardCFMD(ks_rcs, "CFWithoutCache") + .caching(CachingOptions.NONE), + SchemaLoader.standardCFMD(ks_rcs, "CachedCF") + .caching(CachingOptions.ALL), + SchemaLoader.standardCFMD(ks_rcs, "CachedIntCF") + .caching(new CachingOptions(new CachingOptions.KeyCache(CachingOptions.KeyCache.Type.ALL), + new CachingOptions.RowCache(CachingOptions.RowCache.Type.HEAD, 100)))))); + + + keyspaces.add(KeyspaceMetadata.create(ks_nocommit, + KeyspaceParams.simpleTransient(1), + Tables.of(SchemaLoader.standardCFMD(ks_nocommit, "Standard1")))); + + // PerRowSecondaryIndexTest + keyspaces.add(KeyspaceMetadata.create(ks_prsi, + KeyspaceParams.simple(1), + Tables.of(SchemaLoader.perRowIndexedCFMD(ks_prsi, "Indexed1")))); + + // CQLKeyspace + keyspaces.add(KeyspaceMetadata.create(ks_cql, + KeyspaceParams.simple(1), + Tables.of(CFMetaData.compile("CREATE TABLE table1 (" + + "k int PRIMARY KEY," + + "v1 text," + + "v2 int" + + ')', ks_cql), + + CFMetaData.compile("CREATE TABLE table2 (" + + "k text," + + "c text," + + "v text," + + "PRIMARY KEY (k, c))", ks_cql), + + CFMetaData.compile("CREATE TABLE foo (" + + "bar text, " + + "baz text, " + + "qux text, " + + "PRIMARY KEY(bar, baz) ) " + + "WITH COMPACT STORAGE", ks_cql), + + CFMetaData.compile("CREATE TABLE foofoo (" + + "bar text, " + + "baz text, " + + "qux text, " + + "quz text, " + + "foo text, " + + "PRIMARY KEY((bar, baz), qux, quz) ) " + + "WITH COMPACT STORAGE", ks_cql)))); + + keyspaces.add(keyspaceWithTriggers()); + keyspaces.add(keyspaceWithUDTs()); + keyspaces.add(keyspaceWithUDFs()); + keyspaces.add(keyspaceWithUDAs()); + + return keyspaces; + } + + private static KeyspaceMetadata keyspaceWithTriggers() + { + String keyspace = KEYSPACE_PREFIX + "Triggers"; + + CFMetaData table = SchemaLoader.standardCFMD(keyspace, "WithTriggers"); + for (int i = 0; i < 10; i++) + table.addTriggerDefinition(new TriggerDefinition("trigger" + i, "DummyTrigger" + i)); + + return KeyspaceMetadata.create(keyspace, KeyspaceParams.simple(1), Tables.of(table)); + } + + private static KeyspaceMetadata keyspaceWithUDTs() + { + String keyspace = KEYSPACE_PREFIX + "UDTs"; + + UserType udt1 = new UserType(keyspace, + bytes("udt1"), + new ArrayList() {{ add(bytes("col1")); add(bytes("col2")); }}, + new ArrayList>() {{ add(UTF8Type.instance); add(Int32Type.instance); }}); + + UserType udt2 = new UserType(keyspace, + bytes("udt2"), + new ArrayList() {{ add(bytes("col3")); add(bytes("col4")); }}, + new ArrayList>() {{ add(BytesType.instance); add(BooleanType.instance); }}); + + UserType udt3 = new UserType(keyspace, + bytes("udt3"), + new ArrayList() {{ add(bytes("col5")); }}, + new ArrayList>() {{ add(AsciiType.instance); }}); + + return KeyspaceMetadata.create(keyspace, + KeyspaceParams.simple(1), + Tables.none(), + Types.of(udt1, udt2, udt3), + Functions.none()); + } + + private static KeyspaceMetadata keyspaceWithUDFs() + { + String keyspace = KEYSPACE_PREFIX + "UDFs"; + + + UDFunction udf1 = UDFunction.create(new FunctionName(keyspace, "udf"), + ImmutableList.of(new ColumnIdentifier("col1", false), new ColumnIdentifier("col2", false)), + ImmutableList.of(BytesType.instance, Int32Type.instance), + LongType.instance, + false, + "java", + "return 42L;"); + + // an overload with the same name, not a typo + UDFunction udf2 = UDFunction.create(new FunctionName(keyspace, "udf"), + ImmutableList.of(new ColumnIdentifier("col3", false), new ColumnIdentifier("col4", false)), + ImmutableList.of(AsciiType.instance, LongType.instance), + Int32Type.instance, + true, + "java", + "return 42;"); + + UDFunction udf3 = UDFunction.create(new FunctionName(keyspace, "udf3"), + ImmutableList.of(new ColumnIdentifier("col4", false)), + ImmutableList.of(UTF8Type.instance), + BooleanType.instance, + false, + "java", + "return true;"); + + return KeyspaceMetadata.create(keyspace, + KeyspaceParams.simple(1), + Tables.none(), + Types.none(), + Functions.of(udf1, udf2, udf3)); + } + + // TODO: add representative UDAs set + private static KeyspaceMetadata keyspaceWithUDAs() + { + String keyspace = KEYSPACE_PREFIX + "UDAs"; + + return KeyspaceMetadata.create(keyspace, + KeyspaceParams.simple(1), + Tables.none(), + Types.none(), + Functions.of()); + } + + /* + * Serializing keyspaces + */ + + private static void legacySerializeKeyspace(KeyspaceMetadata keyspace) + { + makeLegacyCreateKeyspaceMutation(keyspace, TIMESTAMP).apply(); + } + + private static Mutation makeLegacyCreateKeyspaceMutation(KeyspaceMetadata keyspace, long timestamp) + { + // Note that because Keyspaces is a COMPACT TABLE, we're really only setting static columns internally and shouldn't set any clustering. + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyKeyspaces, timestamp, keyspace.name); + + adder.add("durable_writes", keyspace.params.durableWrites) + .add("strategy_class", keyspace.params.replication.klass.getName()) + .add("strategy_options", json(keyspace.params.replication.options)); + + Mutation mutation = adder.build(); + + keyspace.tables.forEach(table -> addTableToSchemaMutation(table, timestamp, true, mutation)); + keyspace.types.forEach(type -> addTypeToSchemaMutation(type, timestamp, mutation)); + keyspace.functions.udfs().forEach(udf -> addFunctionToSchemaMutation(udf, timestamp, mutation)); + keyspace.functions.udas().forEach(uda -> addAggregateToSchemaMutation(uda, timestamp, mutation)); + + return mutation; + } + + /* + * Serializing tables + */ + + private static void addTableToSchemaMutation(CFMetaData table, long timestamp, boolean withColumnsAndTriggers, Mutation mutation) + { + // For property that can be null (and can be changed), we insert tombstones, to make sure + // we don't keep a property the user has removed + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyColumnfamilies, timestamp, mutation) + .clustering(table.cfName); + + adder.add("cf_id", table.cfId) + .add("type", table.isSuper() ? "Super" : "Standard"); + + if (table.isSuper()) + { + adder.add("comparator", table.comparator.subtype(0).toString()) + .add("subcomparator", ((MapType)table.compactValueColumn().type).getKeysType().toString()); + } + else + { + adder.add("comparator", LegacyLayout.makeLegacyComparator(table).toString()); + } + + adder.add("bloom_filter_fp_chance", table.getBloomFilterFpChance()) + .add("caching", table.getCaching().toString()) + .add("comment", table.getComment()) + .add("compaction_strategy_class", table.compactionStrategyClass.getName()) + .add("compaction_strategy_options", json(table.compactionStrategyOptions)) + .add("compression_parameters", json(table.compressionParameters.asThriftOptions())) + .add("default_time_to_live", table.getDefaultTimeToLive()) + .add("gc_grace_seconds", table.getGcGraceSeconds()) + .add("key_validator", table.getKeyValidator().toString()) + .add("local_read_repair_chance", table.getDcLocalReadRepairChance()) + .add("max_compaction_threshold", table.getMaxCompactionThreshold()) + .add("max_index_interval", table.getMaxIndexInterval()) + .add("memtable_flush_period_in_ms", table.getMemtableFlushPeriod()) + .add("min_compaction_threshold", table.getMinCompactionThreshold()) + .add("min_index_interval", table.getMinIndexInterval()) + .add("read_repair_chance", table.getReadRepairChance()) + .add("speculative_retry", table.getSpeculativeRetry().toString()); + + for (Map.Entry entry : table.getDroppedColumns().entrySet()) + { + String name = entry.getKey().toString(); + CFMetaData.DroppedColumn column = entry.getValue(); + adder.addMapEntry("dropped_columns", name, column.droppedTime); + if (column.type != null) + adder.addMapEntry("dropped_columns_types", name, column.type.toString()); + } + + adder.add("is_dense", table.isDense()); + + adder.add("default_validator", table.makeLegacyDefaultValidator().toString()); + + if (withColumnsAndTriggers) + { + for (ColumnDefinition column : table.allColumns()) + addColumnToSchemaMutation(table, column, timestamp, mutation); + + for (TriggerDefinition trigger : table.getTriggers().values()) + addTriggerToSchemaMutation(table, trigger, timestamp, mutation); + } + + adder.build(); + } + + private static void addColumnToSchemaMutation(CFMetaData table, ColumnDefinition column, long timestamp, Mutation mutation) + { + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyColumns, timestamp, mutation) + .clustering(table.cfName, column.name.toString()); + + adder.add("validator", column.type.toString()) + .add("type", serializeKind(column.kind, table.isDense())) + .add("component_index", column.isOnAllComponents() ? null : column.position()) + .add("index_name", column.getIndexName()) + .add("index_type", column.getIndexType() == null ? null : column.getIndexType().toString()) + .add("index_options", json(column.getIndexOptions())) + .build(); + } + + private static String serializeKind(ColumnDefinition.Kind kind, boolean isDense) + { + // For backward compatibility, we special case CLUSTERING_COLUMN and the case where the table is dense. + if (kind == ColumnDefinition.Kind.CLUSTERING_COLUMN) + return "clustering_key"; + + if (kind == ColumnDefinition.Kind.REGULAR && isDense) + return "compact_value"; + + return kind.toString().toLowerCase(); + } + + private static void addTriggerToSchemaMutation(CFMetaData table, TriggerDefinition trigger, long timestamp, Mutation mutation) + { + new RowUpdateBuilder(SystemKeyspace.LegacyTriggers, timestamp, mutation) + .clustering(table.cfName, trigger.name) + .addMapEntry("trigger_options", "class", trigger.classOption) + .build(); + } + + /* + * Serializing types + */ + + private static void addTypeToSchemaMutation(UserType type, long timestamp, Mutation mutation) + { + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyUsertypes, timestamp, mutation) + .clustering(type.getNameAsString()); + + adder.resetCollection("field_names") + .resetCollection("field_types"); + + for (int i = 0; i < type.size(); i++) + { + adder.addListEntry("field_names", type.fieldName(i)) + .addListEntry("field_types", type.fieldType(i).toString()); + } + + adder.build(); + } + + /* + * Serializing functions + */ + + private static void addFunctionToSchemaMutation(UDFunction function, long timestamp, Mutation mutation) + { + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyFunctions, timestamp, mutation) + .clustering(function.name().name, functionSignatureWithTypes(function)); + + adder.add("body", function.body()) + .add("language", function.language()) + .add("return_type", function.returnType().toString()) + .add("called_on_null_input", function.isCalledOnNullInput()); + + adder.resetCollection("argument_names") + .resetCollection("argument_types"); + + for (int i = 0; i < function.argNames().size(); i++) + { + adder.addListEntry("argument_names", function.argNames().get(i).bytes) + .addListEntry("argument_types", function.argTypes().get(i).toString()); + } + + adder.build(); + } + + /* + * Serializing aggregates + */ + + private static void addAggregateToSchemaMutation(UDAggregate aggregate, long timestamp, Mutation mutation) + { + RowUpdateBuilder adder = new RowUpdateBuilder(SystemKeyspace.LegacyAggregates, timestamp, mutation) + .clustering(aggregate.name().name, functionSignatureWithTypes(aggregate)); + + adder.resetCollection("argument_types"); + + adder.add("return_type", aggregate.returnType().toString()) + .add("state_func", aggregate.stateFunction().name().toString()); + + if (aggregate.stateType() != null) + adder.add("state_type", aggregate.stateType().toString()); + if (aggregate.finalFunction() != null) + adder.add("final_func", aggregate.finalFunction().name().toString()); + if (aggregate.initialCondition() != null) + adder.add("initcond", aggregate.initialCondition()); + + for (AbstractType argType : aggregate.argTypes()) + adder.addListEntry("argument_types", argType.toString()); + + adder.build(); + } + + // We allow method overloads, so a function is not uniquely identified by its name only, but + // also by its argument types. To distinguish overloads of given function name in the schema + // we use a "signature" which is just a list of it's CQL argument types. + public static ByteBuffer functionSignatureWithTypes(AbstractFunction fun) + { + List arguments = + fun.argTypes() + .stream() + .map(argType -> argType.asCQL3Type().toString()) + .collect(Collectors.toList()); + + return ListType.getInstance(UTF8Type.instance, false).decompose(arguments); + } +} diff --git a/test/unit/org/apache/cassandra/config/LegacySchemaTablesTest.java b/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java similarity index 84% rename from test/unit/org/apache/cassandra/config/LegacySchemaTablesTest.java rename to test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java index aabc392ff5..715799bcbf 100644 --- a/test/unit/org/apache/cassandra/config/LegacySchemaTablesTest.java +++ b/test/unit/org/apache/cassandra/schema/SchemaKeyspaceTest.java @@ -16,7 +16,7 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.cassandra.config; +package org.apache.cassandra.schema; import java.util.ArrayList; import java.util.List; @@ -24,6 +24,8 @@ import java.util.HashMap; import java.util.HashSet; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.config.CFMetaData; +import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.*; import org.apache.cassandra.db.rows.UnfilteredRowIterators; import org.apache.cassandra.db.partitions.PartitionUpdate; @@ -33,8 +35,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.compress.*; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.LegacySchemaTables; -import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.thrift.CfDef; import org.apache.cassandra.thrift.ColumnDef; import org.apache.cassandra.thrift.IndexType; @@ -47,12 +48,12 @@ import org.junit.Test; import static org.junit.Assert.assertEquals; -public class LegacySchemaTablesTest +public class SchemaKeyspaceTest { private static final String KEYSPACE1 = "CFMetaDataTest1"; private static final String CF_STANDARD1 = "Standard1"; - private static List columnDefs = new ArrayList(); + private static final List columnDefs = new ArrayList<>(); static { @@ -126,15 +127,14 @@ public class LegacySchemaTablesTest // Testing with compression to catch #3558 CFMetaData withCompression = cfm.copy(); - withCompression.compressionParameters(new CompressionParameters(SnappyCompressor.instance, 32768, new HashMap())); + withCompression.compressionParameters(new CompressionParameters(SnappyCompressor.instance, 32768, new HashMap<>())); checkInverses(withCompression); } } } - private void checkInverses(CFMetaData cfm) throws Exception + private static void checkInverses(CFMetaData cfm) throws Exception { - DecoratedKey k = StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(cfm.ksName)); KeyspaceMetadata keyspace = Schema.instance.getKSMetaData(cfm.ksName); // Test thrift conversion @@ -143,11 +143,11 @@ public class LegacySchemaTablesTest assert before.equals(after) : String.format("%n%s%n!=%n%s", before, after); // Test schema conversion - Mutation rm = LegacySchemaTables.makeCreateTableMutation(keyspace, cfm, FBUtilities.timestampMicros()); - PartitionUpdate serializedCf = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, LegacySchemaTables.COLUMNFAMILIES)); - PartitionUpdate serializedCD = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, LegacySchemaTables.COLUMNS)); - CFMetaData newCfm = LegacySchemaTables.createTableFromTablePartitionAndColumnsPartition(UnfilteredRowIterators.filter(serializedCf.unfilteredIterator(), FBUtilities.nowInSeconds()), - UnfilteredRowIterators.filter(serializedCD.unfilteredIterator(), FBUtilities.nowInSeconds())); + Mutation rm = SchemaKeyspace.makeCreateTableMutation(keyspace, cfm, FBUtilities.timestampMicros()); + PartitionUpdate serializedCf = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, SchemaKeyspace.TABLES)); + PartitionUpdate serializedCD = rm.getPartitionUpdate(Schema.instance.getId(SystemKeyspace.NAME, SchemaKeyspace.COLUMNS)); + CFMetaData newCfm = SchemaKeyspace.createTableFromTablePartitionAndColumnsPartition(UnfilteredRowIterators.filter(serializedCf.unfilteredIterator(), FBUtilities.nowInSeconds()), + UnfilteredRowIterators.filter(serializedCD.unfilteredIterator(), FBUtilities.nowInSeconds())); assert cfm.equals(newCfm) : String.format("%n%s%n!=%n%s", cfm, newCfm); } } diff --git a/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java b/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java index 0819097f1a..e716eda53b 100644 --- a/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java +++ b/test/unit/org/apache/cassandra/service/StorageServiceServerTest.java @@ -38,7 +38,6 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.schema.KeyspaceMetadata; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.Keyspace; -import org.apache.cassandra.db.SystemKeyspace; import org.apache.cassandra.db.WindowsFailedSnapshotTracker; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.dht.Murmur3Partitioner.LongToken; @@ -50,7 +49,7 @@ import org.apache.cassandra.locator.IEndpointSnitch; import org.apache.cassandra.locator.PropertyFileSnitch; import org.apache.cassandra.locator.TokenMetadata; import org.apache.cassandra.schema.KeyspaceParams; -import org.apache.cassandra.schema.LegacySchemaTables; +import org.apache.cassandra.schema.SchemaKeyspace; import org.apache.cassandra.utils.FBUtilities; import static org.junit.Assert.assertEquals; @@ -173,7 +172,7 @@ public class StorageServiceServerTest public void testColumnFamilySnapshot() throws IOException { // no need to insert extra data, even an "empty" database will have a little information in the system keyspace - StorageService.instance.takeColumnFamilySnapshot(SystemKeyspace.NAME, LegacySchemaTables.KEYSPACES, "cf_snapshot"); + StorageService.instance.takeColumnFamilySnapshot(SchemaKeyspace.NAME, SchemaKeyspace.KEYSPACES, "cf_snapshot"); } @Test