diff --git a/CHANGES.txt b/CHANGES.txt index 77111c4467..65008d1603 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -19,6 +19,8 @@ * support trickling fsync() on writes (CASSANDRA-3950) * expose counters for unavailable/timeout exceptions given to thrift clients (CASSANDRA-3671) * avoid quadratic startup time in LeveledManifest (CASSANDRA-3952) + * Add type information to new schema_ columnfamilies and remove thrift + serialization for schema (CASSANDRA-3792) Merged from 1.0: * remove the wait on hint future during write (CASSANDRA-3870) * (cqlsh) ignore missing CfDef opts (CASSANDRA-3933) diff --git a/NEWS.txt b/NEWS.txt index 03b34be8d2..592458df53 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -14,6 +14,8 @@ by version X, but the inverse is not necessarily the case.) Upgrading --------- + - Non-string column aliases are no longer supported; that is, the + column names for defined metadata must be strings. - The KsDef.replication_factor field (deprecated since 0.8) has been removed. Older clients will need to be updated to be able to continue to created and update keyspaces. diff --git a/src/java/org/apache/cassandra/config/Avro.java b/src/java/org/apache/cassandra/config/Avro.java new file mode 100644 index 0000000000..b34d37a5f9 --- /dev/null +++ b/src/java/org/apache/cassandra/config/Avro.java @@ -0,0 +1,193 @@ +/* + * 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.config; + +import java.nio.ByteBuffer; +import java.util.*; + +import org.apache.cassandra.db.ColumnFamilyType; +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.db.migration.avro.CfDef; +import org.apache.cassandra.io.compress.CompressionParameters; +import org.apache.cassandra.locator.AbstractReplicationStrategy; +import org.apache.cassandra.locator.NetworkTopologyStrategy; +import org.apache.cassandra.thrift.IndexType; +import org.apache.cassandra.utils.ByteBufferUtil; + +/** + * methods to load schema definitions from old-style Avro serialization + */ +public class Avro +{ + @Deprecated + public static KSMetaData ksFromAvro(org.apache.cassandra.db.migration.avro.KsDef ks) + { + Class repStratClass; + try + { + String strategyClassName = KSMetaData.convertOldStrategyName(ks.strategy_class.toString()); + repStratClass = (Class)Class.forName(strategyClassName); + } + catch (Exception ex) + { + throw new RuntimeException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex); + } + + Map strategyOptions = new HashMap(); + if (ks.strategy_options != null) + { + for (Map.Entry e : ks.strategy_options.entrySet()) + { + String name = e.getKey().toString(); + // Silently discard a replication_factor option to NTS. + // The history is, people were creating CFs with the default settings (which in the CLI is NTS) and then + // giving it a replication_factor option, which is nonsensical. Initially our strategy was to silently + // ignore this option, but that turned out to confuse people more. So in 0.8.2 we switched to throwing + // an exception in the NTS constructor, which would be turned into an InvalidRequestException for the + // client. But, it also prevented startup for anyone upgrading without first cleaning that option out. + if (repStratClass == NetworkTopologyStrategy.class && name.trim().toLowerCase().equals("replication_factor")) + continue; + strategyOptions.put(name, e.getValue().toString()); + } + } + + int cfsz = ks.cf_defs.size(); + List cfMetaData = new ArrayList(cfsz); + Iterator cfiter = ks.cf_defs.iterator(); + for (CfDef cf_def : ks.cf_defs) + cfMetaData.add(cfFromAvro(cfiter.next())); + + return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, ks.durable_writes, cfMetaData); + } + + @Deprecated + public static CFMetaData cfFromAvro(CfDef cf) + { + AbstractType comparator; + AbstractType subcolumnComparator = null; + AbstractType validator; + AbstractType keyValidator; + + try + { + comparator = TypeParser.parse(cf.comparator_type.toString()); + if (cf.subcomparator_type != null) + subcolumnComparator = TypeParser.parse(cf.subcomparator_type); + validator = TypeParser.parse(cf.default_validation_class); + keyValidator = TypeParser.parse(cf.key_validation_class); + } + catch (Exception ex) + { + throw new RuntimeException("Could not inflate CFMetaData for " + cf, ex); + } + Map column_metadata = new TreeMap(BytesType.instance); + for (org.apache.cassandra.db.migration.avro.ColumnDef aColumn_metadata : cf.column_metadata) + { + ColumnDefinition cd = columnFromAvro(aColumn_metadata); + if (cd.getIndexType() != null && cd.getIndexName() == null) + cd.setIndexName(CFMetaData.getDefaultIndexName(cf.name.toString(), comparator, cd.name)); + column_metadata.put(cd.name, cd); + } + + CFMetaData newCFMD = new CFMetaData(cf.keyspace.toString(), + cf.name.toString(), + ColumnFamilyType.create(cf.column_type.toString()), + comparator, + subcolumnComparator, + cf.id); + + // When we pull up an old avro CfDef which doesn't have these arguments, + // it doesn't default them correctly. Without explicit defaulting, + // grandfathered metadata becomes wrong or causes crashes. + // Isn't AVRO supposed to handle stuff like this? + if (cf.min_compaction_threshold != null) { newCFMD.minCompactionThreshold(cf.min_compaction_threshold); } + if (cf.max_compaction_threshold != null) { newCFMD.maxCompactionThreshold(cf.max_compaction_threshold); } + if (cf.key_alias != null) { newCFMD.keyAlias(cf.key_alias); } + if (cf.column_aliases != null) + newCFMD.columnAliases(new ArrayList(cf.column_aliases)); // fix avro stupidity + if (cf.value_alias != null) { newCFMD.valueAlias(cf.value_alias); } + if (cf.compaction_strategy != null) + { + try + { + newCFMD.compactionStrategyClass = CFMetaData.createCompactionStrategy(cf.compaction_strategy.toString()); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); + } + } + if (cf.compaction_strategy_options != null) + { + for (Map.Entry e : cf.compaction_strategy_options.entrySet()) + newCFMD.compactionStrategyOptions.put(e.getKey().toString(), e.getValue().toString()); + } + + CompressionParameters cp; + try + { + cp = CompressionParameters.create(cf.compression_options); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); + } + + CFMetaData.Caching caching; + + try + { + caching = cf.caching == null ? CFMetaData.Caching.KEYS_ONLY : CFMetaData.Caching.fromString(cf.caching.toString()); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); + } + + return newCFMD.comment(cf.comment.toString()) + .readRepairChance(cf.read_repair_chance) + .dcLocalReadRepairChance(cf.dclocal_read_repair_chance) + .replicateOnWrite(cf.replicate_on_write) + .gcGraceSeconds(cf.gc_grace_seconds) + .defaultValidator(validator) + .keyValidator(keyValidator) + .columnMetadata(column_metadata) + .compressionParameters(cp) + .bloomFilterFpChance(cf.bloom_filter_fp_chance) + .caching(caching); + } + + @Deprecated + public static ColumnDefinition columnFromAvro(org.apache.cassandra.db.migration.avro.ColumnDef cd) + { + IndexType index_type = cd.index_type == null ? null : Enum.valueOf(IndexType.class, cd.index_type.name()); + String index_name = cd.index_name == null ? null : cd.index_name.toString(); + try + { + AbstractType validatorType = TypeParser.parse(cd.validation_class); + return new ColumnDefinition(ByteBufferUtil.clone(cd.name), validatorType, index_type, ColumnDefinition.getStringMap(cd.index_options), index_name); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); + } + } +} diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index fb7736640d..8198f7649b 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -24,17 +24,19 @@ import java.lang.reflect.InvocationTargetException; import java.nio.ByteBuffer; import java.util.*; -import com.google.common.base.Objects; import com.google.common.collect.MapDifference; import com.google.common.collect.Maps; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.EqualsBuilder; import org.apache.commons.lang.builder.HashCodeBuilder; import org.apache.commons.lang.builder.ToStringBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.cql3.CFDefinition; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.*; -import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.compaction.AbstractCompactionStrategy; import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.db.migration.Migration; @@ -42,15 +44,11 @@ import org.apache.cassandra.io.IColumnSerializer; import org.apache.cassandra.io.compress.CompressionParameters; import org.apache.cassandra.io.compress.SnappyCompressor; import org.apache.cassandra.thrift.CfDef; -import org.apache.cassandra.thrift.ColumnDef; -import org.apache.cassandra.thrift.IndexType; import org.apache.cassandra.thrift.InvalidRequestException; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import static org.apache.cassandra.db.migration.MigrationHelper.*; +import static org.apache.cassandra.utils.FBUtilities.*; public final class CFMetaData { @@ -89,35 +87,69 @@ public final class CFMetaData null, null, null))); - public static final CFMetaData SchemaKeyspacesCf = schemaCFDefinition(SystemTable.SCHEMA_KEYSPACES_CF, 8, "keyspace attributes of the schema", AsciiType.instance, 1); - public static final CFMetaData SchemaColumnFamiliesCf = schemaCFDefinition(SystemTable.SCHEMA_COLUMNFAMILIES_CF, 9, "ColumnFamily attributes of the schema", AsciiType.instance, 2); - public static final CFMetaData SchemaColumnsCf = schemaCFDefinition(SystemTable.SCHEMA_COLUMNS_CF, 10, "ColumnFamily column attributes of the schema", AsciiType.instance, 3); - private static CFMetaData schemaCFDefinition(String name, int index, String comment, AbstractType comp, int nestingLevel) + // new-style schema + public static final CFMetaData SchemaKeyspacesCf; + public static final CFMetaData SchemaColumnFamiliesCf; + public static final CFMetaData SchemaColumnsCf; + static { - AbstractType comparator; + SchemaKeyspacesCf = newSchemaMetadata(SystemTable.SCHEMA_KEYSPACES_CF, + 8, + "Keyspace definitions", + AsciiType.instance, + null) + .keyValidator(AsciiType.instance) + .keyAlias("keyspace"); + SchemaKeyspacesCf.columnMetadata(ColumnDefinition.utf8("name"), + ColumnDefinition.bool("durable_writes"), + ColumnDefinition.ascii("strategy_class"), + ColumnDefinition.ascii("strategy_options")); - if (nestingLevel == 1) - { - comparator = comp; - } - else - { - List> composite = new ArrayList>(nestingLevel); - - for (int i = 0; i < nestingLevel; i++) - composite.add(comp); - - comparator = CompositeType.getInstance(composite); - } - - return newSystemMetadata(name, - index, - comment, - comparator, - null) + SchemaColumnFamiliesCf = newSchemaMetadata(SystemTable.SCHEMA_COLUMNFAMILIES_CF, + 9, + "ColumnFamily definitions", + CompositeType.getInstance(Arrays.>asList(AsciiType.instance, AsciiType.instance)), + null) .keyValidator(AsciiType.instance) - .defaultValidator(UTF8Type.instance); + .keyAlias("keyspace") + .columnAliases(Arrays.asList(ByteBufferUtil.bytes("columnfamily"))) + .columnMetadata(ColumnDefinition.int32("id"), + ColumnDefinition.ascii("type"), + ColumnDefinition.ascii("comparator"), + ColumnDefinition.ascii("subcomparator"), + ColumnDefinition.utf8("comment"), + ColumnDefinition.double_("read_repair_chance"), + ColumnDefinition.double_("local_read_repair_chance"), + ColumnDefinition.bool("replicate_on_write"), + ColumnDefinition.int32("gc_grace_seconds"), + ColumnDefinition.ascii("default_validator"), + ColumnDefinition.ascii("key_validator"), + ColumnDefinition.int32("min_compaction_threshold"), + ColumnDefinition.int32("max_compaction_threshold"), + ColumnDefinition.ascii("key_alias"), + ColumnDefinition.double_("bloom_filter_fp_chance"), + ColumnDefinition.ascii("caching"), + ColumnDefinition.ascii("compaction_strategy_class"), + ColumnDefinition.ascii("compression_parameters"), + ColumnDefinition.utf8("value_alias"), + ColumnDefinition.utf8("column_aliases"), + ColumnDefinition.ascii("compaction_strategy_options")); + + SchemaColumnsCf = newSchemaMetadata(SystemTable.SCHEMA_COLUMNS_CF, + 10, + "ColumnFamily column attributes", + CompositeType.getInstance(Arrays.>asList(AsciiType.instance, + AsciiType.instance, + UTF8Type.instance)), + null) + .keyValidator(AsciiType.instance) + .keyAlias("keyspace") + .columnAliases(Arrays.asList(ByteBufferUtil.bytes("columnfamily"), ByteBufferUtil.bytes("column"))) + .columnMetadata(ColumnDefinition.ascii("validator"), + ColumnDefinition.ascii("index_type"), + ColumnDefinition.ascii("index_options"), + ColumnDefinition.ascii("index_name")); } public enum Caching @@ -161,7 +193,7 @@ public final class CFMetaData private Double bloomFilterFpChance; // default NULL private Caching caching; // default KEYS_ONLY (possible: all, key_only, row_only, none) - private Map column_metadata; + Map column_metadata; public Class compactionStrategyClass; public Map compactionStrategyOptions; @@ -174,7 +206,7 @@ public final class CFMetaData public CFMetaData comment(String prop) { comment = enforceCommentNotNull(prop); return this;} public CFMetaData readRepairChance(double prop) {readRepairChance = prop; return this;} - public CFMetaData dclocalReadRepairChance(double prop) {dcLocalReadRepairChance = prop; return this;} + public CFMetaData dcLocalReadRepairChance(double prop) {dcLocalReadRepairChance = prop; return this;} public CFMetaData replicateOnWrite(boolean prop) {replicateOnWrite = prop; return this;} public CFMetaData gcGraceSeconds(int prop) {gcGraceSeconds = prop; return this;} public CFMetaData defaultValidator(AbstractType prop) {defaultValidator = prop; updateCfDef(); return this;} @@ -182,9 +214,18 @@ public final class CFMetaData public CFMetaData minCompactionThreshold(int prop) {minCompactionThreshold = prop; return this;} public CFMetaData maxCompactionThreshold(int prop) {maxCompactionThreshold = prop; return this;} public CFMetaData keyAlias(ByteBuffer prop) {keyAlias = prop; updateCfDef(); return this;} + public CFMetaData keyAlias(String alias) { return keyAlias(ByteBufferUtil.bytes(alias)); } public CFMetaData columnAliases(List prop) {columnAliases = prop; updateCfDef(); return this;} public CFMetaData valueAlias(ByteBuffer prop) {valueAlias = prop; updateCfDef(); return this;} public CFMetaData columnMetadata(Map prop) {column_metadata = prop; updateCfDef(); return this;} + private CFMetaData columnMetadata(ColumnDefinition... cds) + { + Map map = new HashMap(); + for (ColumnDefinition cd : cds) + map.put(cd.name, cd); + + return columnMetadata(map); + } public CFMetaData compactionStrategyClass(Class prop) {compactionStrategyClass = prop; return this;} public CFMetaData compactionStrategyOptions(Map prop) {compactionStrategyOptions = prop; return this;} public CFMetaData compressionParameters(CompressionParameters prop) {compressionParameters = prop; return this;} @@ -196,7 +237,7 @@ public final class CFMetaData this(keyspace, name, type, comp, subcc, Schema.instance.nextCFId()); } - private CFMetaData(String keyspace, String name, ColumnFamilyType type, AbstractType comp, AbstractType subcc, int id) + CFMetaData(String keyspace, String name, ColumnFamilyType type, AbstractType comp, AbstractType subcc, int id) { // Final fields must be set in constructor ksName = keyspace; @@ -262,16 +303,27 @@ public final class CFMetaData return newCFMD.comment(comment) .readRepairChance(0) - .dclocalReadRepairChance(0) + .dcLocalReadRepairChance(0) .gcGraceSeconds(0); } + private static CFMetaData newSchemaMetadata(String cfName, int cfId, String comment, AbstractType comparator, AbstractType subcc) + { + /* + * Schema column families needs a gc_grace (since they are replicated + * on every node). That gc_grace should be high enough that no node + * could be dead for that long a time. + */ + int gcGrace = 120 * 24 * 3600; // 3 months + return newSystemMetadata(cfName, cfId, comment, comparator, subcc).gcGraceSeconds(gcGrace); + } + public static CFMetaData newIndexMetadata(CFMetaData parent, ColumnDefinition info, AbstractType columnComparator) { return new CFMetaData(parent.ksName, parent.indexColumnFamilyName(info), ColumnFamilyType.Standard, columnComparator, null) .keyValidator(info.getValidator()) .readRepairChance(0.0) - .dclocalReadRepairChance(0.0) + .dcLocalReadRepairChance(0.0) .reloadSecondaryIndexMetadata(parent); } @@ -282,27 +334,26 @@ public final class CFMetaData maxCompactionThreshold(parent.maxCompactionThreshold); compactionStrategyClass(parent.compactionStrategyClass); compactionStrategyOptions(parent.compactionStrategyOptions); - compressionParameters(parent.compressionParameters);; + compressionParameters(parent.compressionParameters); return this; } + public CFMetaData clone() + { + return copyOpts(new CFMetaData(ksName, cfName, cfType, comparator, subcolumnComparator, cfId), this); + } + // Create a new CFMD by changing just the cfName public static CFMetaData rename(CFMetaData cfm, String newName) { return copyOpts(new CFMetaData(cfm.ksName, newName, cfm.cfType, cfm.comparator, cfm.subcolumnComparator, cfm.cfId), cfm); } - // Create a new CFMD by changing just the ksName - public static CFMetaData renameTable(CFMetaData cfm, String ksName) - { - return copyOpts(new CFMetaData(ksName, cfm.cfName, cfm.cfType, cfm.comparator, cfm.subcolumnComparator, cfm.cfId), cfm); - } - - private static CFMetaData copyOpts(CFMetaData newCFMD, CFMetaData oldCFMD) + static CFMetaData copyOpts(CFMetaData newCFMD, CFMetaData oldCFMD) { return newCFMD.comment(oldCFMD.comment) .readRepairChance(oldCFMD.readRepairChance) - .dclocalReadRepairChance(oldCFMD.dcLocalReadRepairChance) + .dcLocalReadRepairChance(oldCFMD.dcLocalReadRepairChance) .replicateOnWrite(oldCFMD.replicateOnWrite) .gcGraceSeconds(oldCFMD.gcGraceSeconds) .defaultValidator(oldCFMD.defaultValidator) @@ -330,114 +381,6 @@ public final class CFMetaData return cfName + Directories.SECONDARY_INDEX_NAME_SEPARATOR + (info.getIndexName() == null ? ByteBufferUtil.bytesToHex(info.name) : info.getIndexName()); } - @Deprecated - public static CFMetaData fromAvro(org.apache.cassandra.db.migration.avro.CfDef cf) - { - AbstractType comparator; - AbstractType subcolumnComparator = null; - AbstractType validator; - AbstractType keyValidator; - - try - { - comparator = TypeParser.parse(cf.comparator_type.toString()); - if (cf.subcomparator_type != null) - subcolumnComparator = TypeParser.parse(cf.subcomparator_type); - validator = TypeParser.parse(cf.default_validation_class); - keyValidator = TypeParser.parse(cf.key_validation_class); - } - catch (Exception ex) - { - throw new RuntimeException("Could not inflate CFMetaData for " + cf, ex); - } - Map column_metadata = new TreeMap(BytesType.instance); - for (org.apache.cassandra.db.migration.avro.ColumnDef aColumn_metadata : cf.column_metadata) - { - ColumnDefinition cd = ColumnDefinition.fromAvro(aColumn_metadata); - if (cd.getIndexType() != null && cd.getIndexName() == null) - cd.setIndexName(getDefaultIndexName(cf.name.toString(), comparator, cd.name)); - column_metadata.put(cd.name, cd); - } - - CFMetaData newCFMD = new CFMetaData(cf.keyspace.toString(), - cf.name.toString(), - ColumnFamilyType.create(cf.column_type.toString()), - comparator, - subcolumnComparator, - cf.id); - - // When we pull up an old avro CfDef which doesn't have these arguments, - // it doesn't default them correctly. Without explicit defaulting, - // grandfathered metadata becomes wrong or causes crashes. - // Isn't AVRO supposed to handle stuff like this? - if (cf.min_compaction_threshold != null) { newCFMD.minCompactionThreshold(cf.min_compaction_threshold); } - if (cf.max_compaction_threshold != null) { newCFMD.maxCompactionThreshold(cf.max_compaction_threshold); } - if (cf.key_alias != null) { newCFMD.keyAlias(cf.key_alias); } - if (cf.column_aliases != null) { newCFMD.columnAliases(fixAvroRetardation(cf.column_aliases)); } - if (cf.value_alias != null) { newCFMD.valueAlias(cf.value_alias); } - if (cf.compaction_strategy != null) - { - try - { - newCFMD.compactionStrategyClass = createCompactionStrategy(cf.compaction_strategy.toString()); - } - catch (ConfigurationException e) - { - throw new RuntimeException(e); - } - } - if (cf.compaction_strategy_options != null) - { - for (Map.Entry e : cf.compaction_strategy_options.entrySet()) - newCFMD.compactionStrategyOptions.put(e.getKey().toString(), e.getValue().toString()); - } - - CompressionParameters cp; - try - { - cp = CompressionParameters.create(cf.compression_options); - } - catch (ConfigurationException e) - { - throw new RuntimeException(e); - } - - Caching caching; - - try - { - caching = cf.caching == null ? Caching.KEYS_ONLY : Caching.fromString(cf.caching.toString()); - } - catch (ConfigurationException e) - { - throw new RuntimeException(e); - } - - return newCFMD.comment(cf.comment.toString()) - .readRepairChance(cf.read_repair_chance) - .dclocalReadRepairChance(cf.dclocal_read_repair_chance) - .replicateOnWrite(cf.replicate_on_write) - .gcGraceSeconds(cf.gc_grace_seconds) - .defaultValidator(validator) - .keyValidator(keyValidator) - .columnMetadata(column_metadata) - .compressionParameters(cp) - .bloomFilterFpChance(cf.bloom_filter_fp_chance) - .caching(caching); - } - - /* - * Avro handles array with it's own class, GenericArray, that extends - * AbstractList but redefine equals() in a way that violate List.equals() - * specification (basically only a GenericArray can ever be equal to a - * GenericArray). - * (Concretely, keeping the list returned by avro breaks DefsTest.saveAndRestore()) - */ - private static List fixAvroRetardation(List array) - { - return new ArrayList(array); - } - public String getComment() { return comment; @@ -669,7 +612,7 @@ public final class CFMetaData if (cf_def.isSetCaching()) newCFMD.caching(Caching.fromString(cf_def.caching)); if (cf_def.isSetDclocal_read_repair_chance()) - newCFMD.dclocalReadRepairChance(cf_def.dclocal_read_repair_chance); + newCFMD.dcLocalReadRepairChance(cf_def.dclocal_read_repair_chance); CompressionParameters cp = CompressionParameters.create(cf_def.compression_options); @@ -692,7 +635,7 @@ public final class CFMetaData try { - apply(fromSchema(cfDefRow.cf)); + apply(fromSchema(cfDefRow)); } catch (ConfigurationException e) { @@ -707,135 +650,86 @@ public final class CFMetaData * * @throws ConfigurationException if ks/cf names or cf ids didn't match */ - public void apply(CfDef cf_def) throws ConfigurationException + public void apply(CFMetaData cfm) throws ConfigurationException { - logger.debug("applying {} to {}", cf_def, this); + logger.debug("applying {} to {}", cfm, this); // validate - if (!cf_def.keyspace.equals(ksName)) + if (!cfm.ksName.equals(ksName)) throw new ConfigurationException(String.format("Keyspace mismatch (found %s; expected %s)", - cf_def.keyspace, ksName)); - if (!cf_def.name.equals(cfName)) + cfm.ksName, ksName)); + if (!cfm.cfName.equals(cfName)) throw new ConfigurationException(String.format("Column family mismatch (found %s; expected %s)", - cf_def.name, cfName)); - if (cf_def.id != cfId) + cfm.cfName, cfName)); + if (!cfm.cfId.equals(cfId)) throw new ConfigurationException(String.format("Column family ID mismatch (found %s; expected %s)", - cf_def.id, cfId)); + cfm.cfId, cfId)); - if (!cf_def.column_type.equals(cfType.name())) + if (!cfm.cfType.equals(cfType)) throw new ConfigurationException("types do not match."); - AbstractType newComparator = TypeParser.parse(cf_def.comparator_type); - AbstractType newSubComparator = (cf_def.subcomparator_type == null || cf_def.subcomparator_type.equals("")) - ? null - : TypeParser.parse(cf_def.subcomparator_type); - - if (!newComparator.isCompatibleWith(comparator)) + if (!cfm.comparator.isCompatibleWith(comparator)) throw new ConfigurationException("comparators do not match or are not compatible."); - if (newSubComparator == null) + if (cfm.subcolumnComparator == null) { if (subcolumnComparator != null) throw new ConfigurationException("subcolumncomparators do not match."); // else, it's null and we're good. } - else if (!newSubComparator.isCompatibleWith(subcolumnComparator)) + else if (!cfm.subcolumnComparator.isCompatibleWith(subcolumnComparator)) throw new ConfigurationException("subcolumncomparators do not match or are note compatible."); // TODO: this method should probably return a new CFMetaData so that // 1) we can keep comparator and subcolumnComparator final // 2) updates are applied atomically - comparator = newComparator; - subcolumnComparator = newSubComparator; + comparator = cfm.comparator; + subcolumnComparator = cfm.subcolumnComparator; - validateMinMaxCompactionThresholds(cf_def); + // compaction thresholds are checked by ThriftValidation. We shouldn't be doing + // validation on the apply path; it's too late for that. - comment = enforceCommentNotNull(cf_def.comment); - readRepairChance = cf_def.read_repair_chance; - if (cf_def.isSetDclocal_read_repair_chance()) - dcLocalReadRepairChance = cf_def.dclocal_read_repair_chance; - replicateOnWrite = cf_def.replicate_on_write; - gcGraceSeconds = cf_def.gc_grace_seconds; - defaultValidator = TypeParser.parse(cf_def.default_validation_class); - keyValidator = TypeParser.parse(cf_def.key_validation_class); - minCompactionThreshold = cf_def.min_compaction_threshold; - maxCompactionThreshold = cf_def.max_compaction_threshold; - keyAlias = cf_def.key_alias; - columnAliases = cf_def.column_aliases; - valueAlias = cf_def.value_alias; - if (cf_def.isSetBloom_filter_fp_chance()) - bloomFilterFpChance = cf_def.bloom_filter_fp_chance; - caching = Caching.fromString(cf_def.caching); + comment = enforceCommentNotNull(cfm.comment); + readRepairChance = cfm.readRepairChance; + dcLocalReadRepairChance = cfm.dcLocalReadRepairChance; + replicateOnWrite = cfm.replicateOnWrite; + gcGraceSeconds = cfm.gcGraceSeconds; + defaultValidator = cfm.defaultValidator; + keyValidator = cfm.keyValidator; + minCompactionThreshold = cfm.minCompactionThreshold; + maxCompactionThreshold = cfm.maxCompactionThreshold; + keyAlias = cfm.keyAlias; + columnAliases = cfm.columnAliases; + valueAlias = cfm.valueAlias; + bloomFilterFpChance = cfm.bloomFilterFpChance; + caching = cfm.caching; - if (!cf_def.isSetColumn_metadata()) - cf_def.setColumn_metadata(new ArrayList()); - - // adjust column definitions. figure out who is coming and going. - Set toRemove = new HashSet(); - Set newColumns = new HashSet(); - Set toAdd = new HashSet(); - for (ColumnDef def : cf_def.column_metadata) - { - newColumns.add(def.name); - if (!column_metadata.containsKey(def.name)) - toAdd.add(def); - } - for (ByteBuffer name : column_metadata.keySet()) - if (!newColumns.contains(name)) - toRemove.add(name); - - // remove the ones leaving. - for (ByteBuffer indexName : toRemove) - { - column_metadata.remove(indexName); - } - // update the ones staying - for (ColumnDef def : cf_def.column_metadata) - { - ColumnDefinition oldDef = column_metadata.get(def.name); - if (oldDef == null) - continue; - oldDef.setValidator(TypeParser.parse(def.validation_class)); - oldDef.setIndexType(def.index_type == null ? null : IndexType.valueOf(def.index_type.name()), - def.index_options); - oldDef.setIndexName(def.index_name == null ? null : def.index_name); - } - // add the new ones coming in. - for (ColumnDef def : toAdd) - { - AbstractType dValidClass = TypeParser.parse(def.validation_class); - ColumnDefinition cd = new ColumnDefinition(def.name, - dValidClass, - def.index_type == null ? null : IndexType.valueOf(def.index_type.name()), - def.index_options, - def.index_name == null ? null : def.index_name); + MapDifference columnDiff = Maps.difference(column_metadata, cfm.column_metadata); + // columns that are no longer needed + for (ColumnDefinition cd : columnDiff.entriesOnlyOnLeft().values()) + column_metadata.remove(cd.name); + // newly added columns + for (ColumnDefinition cd : columnDiff.entriesOnlyOnRight().values()) column_metadata.put(cd.name, cd); - } - - if (cf_def.compaction_strategy != null) - compactionStrategyClass = createCompactionStrategy(cf_def.compaction_strategy); - - if (null != cf_def.compaction_strategy_options) + // old columns with updated attributes + for (ByteBuffer name : columnDiff.entriesDiffering().keySet()) { - compactionStrategyOptions = new HashMap(); - for (Map.Entry e : cf_def.compaction_strategy_options.entrySet()) - compactionStrategyOptions.put(e.getKey(), e.getValue()); + ColumnDefinition oldDef = column_metadata.get(name); + ColumnDefinition def = cfm.column_metadata.get(name); + oldDef.apply(def, comparator); } - compressionParameters = CompressionParameters.create(cf_def.compression_options); + compactionStrategyClass = cfm.compactionStrategyClass; + compactionStrategyOptions = cfm.compactionStrategyOptions; + compressionParameters = cfm.compressionParameters(); + + updateCfDef(); logger.debug("application result is {}", this); } public static Class createCompactionStrategy(String className) throws ConfigurationException { className = className.contains(".") ? className : "org.apache.cassandra.db.compaction." + className; - try - { - return (Class) Class.forName(className); - } - catch (Exception e) - { - throw new ConfigurationException("Could not create Compaction Strategy of type " + className, e); - } + return FBUtilities.classForName(className, "compaction strategy"); } public AbstractCompactionStrategy createCompactionStrategyInstance(ColumnFamilyStore cfs) @@ -846,7 +740,7 @@ public final class CFMetaData ColumnFamilyStore.class, Map.class // options }); - return (AbstractCompactionStrategy)constructor.newInstance(cfs, compactionStrategyOptions); + return constructor.newInstance(cfs, compactionStrategyOptions); } catch (NoSuchMethodException e) { @@ -910,36 +804,6 @@ public final class CFMetaData return def; } - public static void validateMinMaxCompactionThresholds(CfDef cf_def) throws ConfigurationException - { - if (cf_def.isSetMin_compaction_threshold() && cf_def.isSetMax_compaction_threshold()) - { - if ((cf_def.min_compaction_threshold > cf_def.max_compaction_threshold) && - cf_def.max_compaction_threshold != 0) - { - throw new ConfigurationException("min_compaction_threshold cannot be greater than max_compaction_threshold"); - } - } - else if (cf_def.isSetMin_compaction_threshold()) - { - if (cf_def.min_compaction_threshold > DEFAULT_MAX_COMPACTION_THRESHOLD) - { - throw new ConfigurationException("min_compaction_threshold cannot be greater than max_compaction_threshold (default " + - DEFAULT_MAX_COMPACTION_THRESHOLD + ")"); - } - } - else if (cf_def.isSetMax_compaction_threshold()) - { - if (cf_def.max_compaction_threshold < DEFAULT_MIN_COMPACTION_THRESHOLD && cf_def.max_compaction_threshold != 0) { - throw new ConfigurationException("max_compaction_threshold cannot be less than min_compaction_threshold"); - } - } - else - { - //Defaults are valid. - } - } - public ColumnDefinition getColumnDefinition(ByteBuffer name) { return column_metadata.get(name); @@ -1028,47 +892,30 @@ public final class CFMetaData * * @throws ConfigurationException if any of the attributes didn't pass validation */ - public RowMutation diff(CfDef newState, long modificationTimestamp) throws ConfigurationException + public RowMutation diff(CFMetaData newState, long modificationTimestamp) throws ConfigurationException { - CfDef curState = toThrift(); - RowMutation m = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); + RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); - for (CfDef._Fields field : CfDef._Fields.values()) - { - if (field.equals(CfDef._Fields.COLUMN_METADATA)) - continue; // deal with columns after main attributes + newState.toSchemaNoColumns(rm, modificationTimestamp); - Object curValue = curState.isSet(field) ? curState.getFieldValue(field) : null; - Object newValue = newState.isSet(field) ? newState.getFieldValue(field) : null; - - if (Objects.equal(curValue, newValue)) - continue; - - m.add(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF, null, compositeNameFor(curState.name, field.getFieldName())), - valueAsBytes(newValue), - modificationTimestamp); - } - - AbstractType nameComparator = cfType.equals(ColumnFamilyType.Super) - ? subcolumnComparator - : comparator; - - MapDifference columnDiff = Maps.difference(column_metadata, ColumnDefinition.fromThrift(newState.column_metadata)); - Map columnDefMap = ColumnDefinition.toMap(newState.column_metadata); + MapDifference columnDiff = Maps.difference(column_metadata, newState.column_metadata); // columns that are no longer needed - for (ByteBuffer name : columnDiff.entriesOnlyOnLeft().keySet()) - ColumnDefinition.deleteFromSchema(m, curState.name, nameComparator, name, modificationTimestamp); + for (ColumnDefinition cd : columnDiff.entriesOnlyOnLeft().values()) + cd.deleteFromSchema(rm, cfName, comparator, modificationTimestamp); // newly added columns - for (ByteBuffer name : columnDiff.entriesOnlyOnRight().keySet()) - ColumnDefinition.addToSchema(m, curState.name, nameComparator, columnDefMap.get(name), modificationTimestamp); + for (ColumnDefinition cd : columnDiff.entriesOnlyOnRight().values()) + cd.toSchema(rm, cfName, comparator, modificationTimestamp); // old columns with updated attributes for (ByteBuffer name : columnDiff.entriesDiffering().keySet()) - ColumnDefinition.addToSchema(m, curState.name, nameComparator, columnDefMap.get(name), modificationTimestamp); + { + ColumnDefinition cd = newState.getColumnDefinition(name); + cd.toSchema(rm, cfName, comparator, modificationTimestamp); + } - return m; + return rm; } /** @@ -1080,15 +927,157 @@ public final class CFMetaData */ public RowMutation dropFromSchema(long timestamp) { - RowMutation m = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); + RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); + ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNFAMILIES_CF); + int ldt = (int) (System.currentTimeMillis() / 1000); - for (CfDef._Fields field : CfDef._Fields.values()) - m.delete(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF, null, compositeNameFor(cfName, field.getFieldName())), timestamp); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "id")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "type")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "comparator")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "subcomparator")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "comment")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "read_repair_chance")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "local_read_repair_chance")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "replicate_on_write")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "gc_grace_seconds")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "default_validator")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "key_validator")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "min_compaction_threshold")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "max_compaction_threshold")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "key_alias")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "bloom_filter_fp_chance")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "caching")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "compaction_strategy_class")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "compression_parameters")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "value_alias")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "column_aliases")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, "compaction_strategy_options")); - for (ColumnDefinition columnDefinition : column_metadata.values()) - ColumnDefinition.deleteFromSchema(m, cfName, comparator, columnDefinition.name, timestamp); + for (ColumnDefinition cd : column_metadata.values()) + cd.deleteFromSchema(rm, cfName, comparator, timestamp); - return m; + return rm; + } + + public void toSchema(RowMutation rm, long timestamp) + { + toSchemaNoColumns(rm, timestamp); + + for (ColumnDefinition cd : column_metadata.values()) + cd.toSchema(rm, cfName, comparator, timestamp); + } + + private void toSchemaNoColumns(RowMutation rm, long timestamp) + { + // 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 + ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNFAMILIES_CF); + int ldt = (int) (System.currentTimeMillis() / 1000); + + cf.addColumn(Column.create(cfId, timestamp, cfName, "id")); + cf.addColumn(Column.create(cfType.toString(), timestamp, cfName, "type")); + cf.addColumn(Column.create(comparator.toString(), timestamp, cfName, "comparator")); + if (subcolumnComparator != null) + cf.addColumn(Column.create(subcolumnComparator.toString(), timestamp, cfName, "subcomparator")); + cf.addColumn(comment == null ? DeletedColumn.create(ldt, timestamp, cfName, "comment") + : Column.create(comment, timestamp, cfName, "comment")); + cf.addColumn(Column.create(readRepairChance, timestamp, cfName, "read_repair_chance")); + cf.addColumn(Column.create(dcLocalReadRepairChance, timestamp, cfName, "local_read_repair_chance")); + cf.addColumn(Column.create(replicateOnWrite, timestamp, cfName, "replicate_on_write")); + cf.addColumn(Column.create(gcGraceSeconds, timestamp, cfName, "gc_grace_seconds")); + cf.addColumn(Column.create(defaultValidator.toString(), timestamp, cfName, "default_validator")); + cf.addColumn(Column.create(keyValidator.toString(), timestamp, cfName, "key_validator")); + cf.addColumn(Column.create(minCompactionThreshold, timestamp, cfName, "min_compaction_threshold")); + cf.addColumn(Column.create(maxCompactionThreshold, timestamp, cfName, "max_compaction_threshold")); + cf.addColumn(keyAlias == null ? DeletedColumn.create(ldt, timestamp, cfName, "key_alias") + : Column.create(keyAlias, timestamp, cfName, "key_alias")); + cf.addColumn(bloomFilterFpChance == null ? DeletedColumn.create(ldt, timestamp, cfName, "bloomFilterFpChance") + : Column.create(bloomFilterFpChance, timestamp, cfName, "bloom_filter_fp_chance")); + cf.addColumn(Column.create(caching.toString(), timestamp, cfName, "caching")); + cf.addColumn(Column.create(compactionStrategyClass.getName(), timestamp, cfName, "compaction_strategy_class")); + cf.addColumn(Column.create(json(compressionParameters.asThriftOptions()), timestamp, cfName, "compression_parameters")); + cf.addColumn(valueAlias == null ? DeletedColumn.create(ldt, timestamp, cfName, "value_alias") + : Column.create(valueAlias, timestamp, cfName, "value_alias")); + cf.addColumn(Column.create(json(columnAliasesAsStrings()), timestamp, cfName, "column_aliases")); + cf.addColumn(Column.create(json(compactionStrategyOptions), timestamp, cfName, "compaction_strategy_options")); + } + + // Package protected for use by tests + static CFMetaData fromSchemaNoColumns(UntypedResultSet.Row result) + { + try + { + CFMetaData cfm = new CFMetaData(result.getString("keyspace"), + result.getString("columnfamily"), + ColumnFamilyType.valueOf(result.getString("type")), + TypeParser.parse(result.getString("comparator")), + result.has("subcomparator") ? TypeParser.parse(result.getString("subcomparator")) : null, + result.getInt("id")); + cfm.readRepairChance(result.getDouble("read_repair_chance")); + cfm.dcLocalReadRepairChance(result.getDouble("local_read_repair_chance")); + cfm.replicateOnWrite(result.getBoolean("replicate_on_write")); + cfm.gcGraceSeconds(result.getInt("gc_grace_seconds")); + cfm.defaultValidator(TypeParser.parse(result.getString("default_validator"))); + cfm.keyValidator(TypeParser.parse(result.getString("key_validator"))); + cfm.minCompactionThreshold(result.getInt("min_compaction_threshold")); + cfm.maxCompactionThreshold(result.getInt("max_compaction_threshold")); + if (result.has("comment")) + cfm.comment(result.getString("comment")); + if (result.has("key_alias")) + cfm.keyAlias(result.getBytes("key_alias")); + if (result.has("bloom_filter_fp_chance")) + cfm.bloomFilterFpChance(result.getDouble("bloom_filter_fp_chance")); + cfm.caching(Caching.valueOf(result.getString("caching"))); + cfm.compactionStrategyClass(createCompactionStrategy(result.getString("compaction_strategy_class"))); + cfm.compressionParameters(CompressionParameters.create(fromJsonMap(result.getString("compression_parameters")))); + if (result.has("value_alias")) + cfm.valueAlias(result.getBytes("value_alias")); + cfm.columnAliases(columnAliasesFromStrings(fromJsonList(result.getString("column_aliases")))); + cfm.compactionStrategyOptions(fromJsonMap(result.getString("compaction_strategy_options"))); + + return cfm; + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); + } + } + + /** + * Deserialize CF metadata from low-level representation + * + * @return Thrift-based metadata deserialized from schema + * + * @throws IOException on any I/O related error + */ + public static CFMetaData fromSchema(UntypedResultSet.Row result) + { + CFMetaData cfDef = fromSchemaNoColumns(result); + + Row serializedColumnDefinitions = ColumnDefinition.readSchema(cfDef.ksName, cfDef.cfName); + return addColumnDefinitionSchema(cfDef, serializedColumnDefinitions); + } + + private static CFMetaData fromSchema(Row row) + { + UntypedResultSet.Row result = QueryProcessor.resultify("SELECT * FROM system.schema_columnfamilies", row).one(); + return fromSchema(result); + } + + private List columnAliasesAsStrings() + { + List aliases = new ArrayList(columnAliases.size()); + for (ByteBuffer rawAlias : columnAliases) + aliases.add(UTF8Type.instance.compose(rawAlias)); + return aliases; + } + + private static List columnAliasesFromStrings(List aliases) + { + List rawAliases = new ArrayList(aliases.size()); + for (String alias : aliases) + rawAliases.add(UTF8Type.instance.decompose(alias)); + return rawAliases; } /** @@ -1102,51 +1091,16 @@ public final class CFMetaData */ public RowMutation toSchema(long timestamp) throws ConfigurationException { - RowMutation mutation = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); - - toSchema(mutation, toThrift(), timestamp); - - return mutation; - } - - /** - * Convert given Thrift-serialized metadata into schema mutation - * - * @param mutation The mutation to include ColumnFamily attributes into (can contain keyspace attributes already) - * @param cfDef Thrift-serialized metadata to use as source for schema mutation - * @param timestamp Timestamp to use - * - * @throws ConfigurationException if any of the attributes didn't pass validation - */ - public static void toSchema(RowMutation mutation, CfDef cfDef, long timestamp) throws ConfigurationException - { - applyImplicitDefaults(cfDef); - - for (CfDef._Fields field : CfDef._Fields.values()) - { - if (field.equals(CfDef._Fields.COLUMN_METADATA)) - continue; - - Object value = cfDef.isSet(field) ? cfDef.getFieldValue(field) : null; - mutation.add(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF, null, compositeNameFor(cfDef.name, field.getFieldName())), - valueAsBytes(value), - timestamp); - } - - if (!cfDef.isSetColumn_metadata()) - return; - - AbstractType comparator = getColumnDefinitionComparator(cfDef); - - for (ColumnDef columnDef : cfDef.column_metadata) - ColumnDefinition.addToSchema(mutation, cfDef.name, comparator, columnDef, timestamp); + RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksName)); + toSchema(rm, timestamp); + return rm; } public static AbstractType getColumnDefinitionComparator(CfDef cfDef) throws ConfigurationException { AbstractType cfComparator = TypeParser.parse(cfDef.column_type.equals("Super") - ? cfDef.subcomparator_type - : cfDef.comparator_type); + ? cfDef.subcomparator_type + : cfDef.comparator_type); if (cfComparator instanceof CompositeType) { @@ -1159,59 +1113,22 @@ public final class CFMetaData } } - /** - * Deserialize CF metadata from low-level representation - * - * @param serializedCfDef The data to use for deserialization - * - * @return Thrift-based metadata deserialized from schema - * - * @throws IOException on any I/O related error - */ - public static CfDef fromSchema(ColumnFamily serializedCfDef) throws IOException - { - CfDef cfDef = fromSchemaNoColumnDefinition(serializedCfDef); - - ColumnFamily serializedColumnDefinitions = ColumnDefinition.readSchema(cfDef.keyspace, cfDef.name); - return addColumnDefinitionSchema(cfDef, serializedColumnDefinitions); - } - // Package protected for use by tests - static CfDef fromSchemaNoColumnDefinition(ColumnFamily serializedCfDef) + static CFMetaData addColumnDefinitionSchema(CFMetaData cfDef, Row serializedColumnDefinitions) { - assert serializedCfDef != null; - - CfDef cfDef = new CfDef(); - - AbstractType sysComparator = serializedCfDef.getComparator(); - - for (IColumn cfAttr : serializedCfDef.getSortedColumns()) - { - if (cfAttr == null || cfAttr.isMarkedForDelete()) - continue; - - // column name format is : - String[] attr = sysComparator.getString(cfAttr.name()).split(":"); - assert attr.length == 2; - - CfDef._Fields field = CfDef._Fields.findByName(attr[1]); - - // this means that given field was deprecated - // but still exists in the serialized schema - if (field == null) - continue; - - cfDef.setFieldValue(field, deserializeValue(cfAttr.value(), getValueClass(CfDef.class, field.getFieldName()))); - } + for (ColumnDefinition cd : ColumnDefinition.fromSchema(serializedColumnDefinitions, cfDef.comparator)) + cfDef.column_metadata.put(cd.name, cd); return cfDef; } - // Package protected for use by tests - static CfDef addColumnDefinitionSchema(CfDef cfDef, ColumnFamily serializedColumnDefinitions) + public void addColumnDefinition(ColumnDefinition def) { - for (ColumnDef columnDef : ColumnDefinition.fromSchema(serializedColumnDefinitions)) - cfDef.addToColumn_metadata(columnDef); - return cfDef; + column_metadata.put(def.name, def); + } + + public boolean removeColumnDefinition(ColumnDefinition def) + { + return column_metadata.remove(def.name) != null; } private void updateCfDef() diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java index da006c2c57..e47ac0743c 100644 --- a/src/java/org/apache/cassandra/config/ColumnDefinition.java +++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java @@ -22,21 +22,25 @@ package org.apache.cassandra.config; import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; import java.util.*; import com.google.common.collect.Maps; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.QueryPath; -import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.cassandra.db.marshal.TypeParser; +import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.db.migration.MigrationHelper; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.thrift.ColumnDef; import org.apache.cassandra.thrift.IndexType; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; import static org.apache.cassandra.db.migration.MigrationHelper.*; +import static org.apache.cassandra.utils.FBUtilities.json; public class ColumnDefinition { @@ -48,6 +52,7 @@ public class ColumnDefinition public ColumnDefinition(ByteBuffer name, AbstractType validator, IndexType index_type, Map index_options, String index_name) { + assert name != null && validator != null; this.name = name; this.index_name = index_name; this.validator = validator; @@ -55,6 +60,31 @@ public class ColumnDefinition this.setIndexType(index_type, index_options); } + public static ColumnDefinition ascii(String name) + { + return new ColumnDefinition(ByteBufferUtil.bytes(name), AsciiType.instance, null, null, null); + } + + public static ColumnDefinition bool(String name) + { + return new ColumnDefinition(ByteBufferUtil.bytes(name), BooleanType.instance, null, null, null); + } + + public static ColumnDefinition utf8(String name) + { + return new ColumnDefinition(ByteBufferUtil.bytes(name), UTF8Type.instance, null, null, null); + } + + public static ColumnDefinition int32(String name) + { + return new ColumnDefinition(ByteBufferUtil.bytes(name), Int32Type.instance, null, null, null); + } + + public static ColumnDefinition double_(String name) + { + return new ColumnDefinition(ByteBufferUtil.bytes(name), DoubleType.instance, null, null, null); + } + @Override public boolean equals(Object o) { @@ -86,22 +116,6 @@ public class ColumnDefinition return result; } - @Deprecated - public static ColumnDefinition fromAvro(org.apache.cassandra.db.migration.avro.ColumnDef cd) - { - IndexType index_type = cd.index_type == null ? null : Enum.valueOf(IndexType.class, cd.index_type.name()); - String index_name = cd.index_name == null ? null : cd.index_name.toString(); - try - { - AbstractType validatorType = TypeParser.parse(cd.validation_class); - return new ColumnDefinition(ByteBufferUtil.clone(cd.name), validatorType, index_type, getStringMap(cd.index_options), index_name); - } - catch (ConfigurationException e) - { - throw new RuntimeException(e); - } - } - public ColumnDef toThrift() { ColumnDef cd = new ColumnDef(); @@ -139,138 +153,102 @@ public class ColumnDefinition return cds; } - public static Map toMap(List columnDefs) - { - Map map = new HashMap(); - - if (columnDefs == null) - return map; - - for (ColumnDef columnDef : columnDefs) - map.put(columnDef.name, columnDef); - - return map; - } - /** - * Drop specified column from the schema using given row mutation. + * Drop specified column from the schema using given row. * - * @param mutation The schema row mutation + * @param rm The schema row mutation * @param cfName The name of the parent ColumnFamily - * @param comparator The comparator to serialize column name in human-readable format - * @param columnName The column name as String * @param timestamp The timestamp to use for column modification */ - public static void deleteFromSchema(RowMutation mutation, String cfName, AbstractType comparator, ByteBuffer columnName, long timestamp) + public void deleteFromSchema(RowMutation rm, String cfName, AbstractType comparator, long timestamp) { - toSchema(mutation, comparator, cfName, columnName, null, timestamp, true); + ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNS_CF); + int ldt = (int) (System.currentTimeMillis() / 1000); + + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "validator")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_type")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_options")); + cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_name")); } - /** - * Add new/update column to/in the schema. - * - * @param mutation The schema row mutation - * @param cfName The name of the parent ColumnFamily - * @param comparator The comparator to serialize column name in human-readable format - * @param columnDef The Thrift-based column definition that contains all attributes - * @param timestamp The timestamp to use for column modification - */ - public static void addToSchema(RowMutation mutation, String cfName, AbstractType comparator, ColumnDef columnDef, long timestamp) + public void toSchema(RowMutation rm, String cfName, AbstractType comparator, long timestamp) { - toSchema(mutation, comparator, cfName, columnDef.name, columnDef, timestamp, false); + ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNS_CF); + int ldt = (int) (System.currentTimeMillis() / 1000); + + cf.addColumn(Column.create(validator.toString(), timestamp, cfName, comparator.getString(name), "validator")); + cf.addColumn(index_type == null ? DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_type") + : Column.create(index_type.toString(), timestamp, cfName, comparator.getString(name), "index_type")); + cf.addColumn(index_options == null ? DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_options") + : Column.create(json(index_options), timestamp, cfName, comparator.getString(name), "index_options")); + cf.addColumn(index_name == null ? DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_name") + : Column.create(index_name, timestamp, cfName, comparator.getString(name), "index_name")); } - /** - * Serialize given ColumnDef into given schema row mutation to add or drop it. - * - * @param mutation The mutation to use for serialization - * @param comparator The comparator to serialize column name in human-readable format - * @param cfName The name of the parent ColumnFamily - * @param columnName The column name as String - * @param columnDef The Thrift-based column definition that contains all attributes - * @param timestamp The timestamp to use for column modification - * @param delete The flag which indicates if column should be deleted or added to the schema - */ - private static void toSchema(RowMutation mutation, AbstractType comparator, String cfName, ByteBuffer columnName, ColumnDef columnDef, long timestamp, boolean delete) + public void apply(ColumnDefinition def, AbstractType comparator) throws ConfigurationException { - for (ColumnDef._Fields field : ColumnDef._Fields.values()) - { - QueryPath path = new QueryPath(SystemTable.SCHEMA_COLUMNS_CF, - null, - compositeNameFor(cfName, - readableColumnName(columnName, comparator), - field.getFieldName())); + // If an index is set (and not drop by this update), the validator shouldn't be change to a non-compatible one + if (getIndexType() != null && def.getIndexType() != null && !def.validator.isCompatibleWith(validator)) + throw new ConfigurationException(String.format("Cannot modify validator to a non-compatible one for column %s since an index is set", comparator.getString(name))); - if (delete) - mutation.delete(path, timestamp); - else - mutation.add(path, valueAsBytes(columnDef.getFieldValue(field)), timestamp); - } - } - - public static ColumnFamily readSchema(String ksName, String cfName) - { - DecoratedKey key = StorageService.getPartitioner().decorateKey(SystemTable.getSchemaKSKey(ksName)); - ColumnFamilyStore columnsStore = SystemTable.schemaCFS(SystemTable.SCHEMA_COLUMNS_CF); - return columnsStore.getColumnFamily(key, - new QueryPath(SystemTable.SCHEMA_COLUMNS_CF), - MigrationHelper.searchComposite(cfName, true), - MigrationHelper.searchComposite(cfName, false), - false, - Integer.MAX_VALUE); + setValidator(def.getValidator()); + setIndexType(def.getIndexType(), def.getIndexOptions()); + setIndexName(def.getIndexName()); } /** * Deserialize columns from low-level representation * * @return Thrift-based deserialized representation of the column + * @param row */ - public static List fromSchema(ColumnFamily columns) + public static List fromSchema(Row row, AbstractType comparator) { - - if (columns == null || columns.isEmpty()) + if (row.cf == null) return Collections.emptyList(); - // contenders to be a valid columns, re-check is done after all attributes - // were read from serialized state, if ColumnDef has all required fields it gets promoted to be returned - Map contenders = new HashMap(); - - for (IColumn column : columns.getSortedColumns()) + List cds = new ArrayList(); + for (UntypedResultSet.Row result : QueryProcessor.resultify("SELECT * FROM system.schema_columns", row)) { - if (column.isMarkedForDelete()) - continue; - - // column name format :: - String[] components = columns.getComparator().getString(column.name()).split(":"); - assert components.length == 3; - - ColumnDef columnDef = contenders.get(components[1]); - - if (columnDef == null) + try { - columnDef = new ColumnDef(); - contenders.put(components[1], columnDef); + IndexType index_type = null; + Map index_options = null; + String index_name = null; + + if (result.has("index_type")) + index_type = IndexType.valueOf(result.getString("index_type")); + if (result.has("index_options")) + index_options = FBUtilities.fromJsonMap(result.getString("index_options")); + if (result.has("index_name")) + index_name = result.getString("index_name"); + + cds.add(new ColumnDefinition(comparator.fromString(result.getString("column")), + TypeParser.parse(result.getString("validator")), + index_type, + index_options, + index_name)); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); } - - ColumnDef._Fields field = ColumnDef._Fields.findByName(components[2]); - - // this means that given field was deprecated - // but still exists in the serialized schema - if (field == null) - continue; - - columnDef.setFieldValue(field, deserializeValue(column.value(), getValueClass(ColumnDef.class, field.getFieldName()))); } - List columnDefs = new ArrayList(); + return cds; + } - for (ColumnDef columnDef : contenders.values()) - { - if (columnDef.isSetName() && columnDef.isSetValidation_class()) - columnDefs.add(columnDef); - } - - return columnDefs; + public static Row readSchema(String ksName, String cfName) + { + DecoratedKey key = StorageService.getPartitioner().decorateKey(SystemTable.getSchemaKSKey(ksName)); + ColumnFamilyStore columnsStore = SystemTable.schemaCFS(SystemTable.SCHEMA_COLUMNS_CF); + ColumnFamily cf = columnsStore.getColumnFamily(key, + new QueryPath(SystemTable.SCHEMA_COLUMNS_CF), + MigrationHelper.searchComposite(cfName, true), + MigrationHelper.searchComposite(cfName, false), + false, + Integer.MAX_VALUE); + return new Row(key, cf); } @Override diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java index e67151111b..5b9d5ba3b3 100644 --- a/src/java/org/apache/cassandra/config/KSMetaData.java +++ b/src/java/org/apache/cassandra/config/KSMetaData.java @@ -25,6 +25,8 @@ import com.google.common.base.Objects; import org.apache.commons.lang.ObjectUtils; import org.apache.commons.lang.StringUtils; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; @@ -37,6 +39,7 @@ import org.apache.cassandra.thrift.KsDef; import org.apache.cassandra.thrift.ColumnDef; import static org.apache.cassandra.db.migration.MigrationHelper.*; +import static org.apache.cassandra.utils.FBUtilities.*; public final class KSMetaData { @@ -46,7 +49,7 @@ public final class KSMetaData private final Map cfMetaData; public final boolean durableWrites; - private KSMetaData(String name, Class strategyClass, Map strategyOptions, boolean durableWrites, Iterable cfDefs) + KSMetaData(String name, Class strategyClass, Map strategyOptions, boolean durableWrites, Iterable cfDefs) { this.name = name; this.strategyClass = strategyClass == null ? NetworkTopologyStrategy.class : strategyClass; @@ -124,47 +127,6 @@ public final class KSMetaData return sb.toString(); } - @Deprecated - public static KSMetaData fromAvro(org.apache.cassandra.db.migration.avro.KsDef ks) - { - Class repStratClass; - try - { - String strategyClassName = convertOldStrategyName(ks.strategy_class.toString()); - repStratClass = (Class)Class.forName(strategyClassName); - } - catch (Exception ex) - { - throw new RuntimeException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex); - } - - Map strategyOptions = new HashMap(); - if (ks.strategy_options != null) - { - for (Map.Entry e : ks.strategy_options.entrySet()) - { - String name = e.getKey().toString(); - // Silently discard a replication_factor option to NTS. - // The history is, people were creating CFs with the default settings (which in the CLI is NTS) and then - // giving it a replication_factor option, which is nonsensical. Initially our strategy was to silently - // ignore this option, but that turned out to confuse people more. So in 0.8.2 we switched to throwing - // an exception in the NTS constructor, which would be turned into an InvalidRequestException for the - // client. But, it also prevented startup for anyone upgrading without first cleaning that option out. - if (repStratClass == NetworkTopologyStrategy.class && name.trim().toLowerCase().equals("replication_factor")) - continue; - strategyOptions.put(name, e.getValue().toString()); - } - } - - int cfsz = ks.cf_defs.size(); - List cfMetaData = new ArrayList(cfsz); - Iterator cfiter = ks.cf_defs.iterator(); - for (int i = 0; i < cfsz; i++) - cfMetaData.add(CFMetaData.fromAvro(cfiter.next())); - - return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, ks.durable_writes, cfMetaData); - } - public static String convertOldStrategyName(String name) { return name.replace("RackUnawareStrategy", "SimpleStrategy") @@ -199,209 +161,115 @@ public final class KSMetaData return ksdef; } - public RowMutation diff(KsDef newState, long modificationTimestamp) + public RowMutation diff(KSMetaData newState, long modificationTimestamp) { - KsDef curState = toThrift(); - RowMutation m = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name)); - - for (KsDef._Fields field : KsDef._Fields.values()) - { - if (field.equals(KsDef._Fields.CF_DEFS)) - continue; - - Object curValue = curState.getFieldValue(field); - Object newValue = newState.getFieldValue(field); - - if (Objects.equal(curValue, newValue)) - continue; - - m.add(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF, null, AsciiType.instance.fromString(field.getFieldName())), - valueAsBytes(newValue), - modificationTimestamp); - } - - return m; + return newState.toSchema(modificationTimestamp); } public KSMetaData reloadAttributes() throws IOException { Row ksDefRow = SystemTable.readSchemaRow(name); - if (ksDefRow.cf == null || ksDefRow.cf.isEmpty()) + if (ksDefRow.cf == null) throw new IOException(String.format("%s not found in the schema definitions table (%s).", name, SystemTable.SCHEMA_KEYSPACES_CF)); - return fromSchema(ksDefRow.cf, null); + return fromSchema(ksDefRow, Collections.emptyList()); } - public List dropFromSchema(long timestamp) + public RowMutation dropFromSchema(long timestamp) { - List mutations = new ArrayList(); + RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name)); + rm.delete(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF), timestamp); + rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF), timestamp); + rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNS_CF), timestamp); - RowMutation ksMutation = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name)); - ksMutation.delete(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF), timestamp); - mutations.add(ksMutation); + return rm; + } + + public RowMutation toSchema(long timestamp) + { + RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name)); + ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_KEYSPACES_CF); + + cf.addColumn(Column.create(name, timestamp, "name")); + cf.addColumn(Column.create(durableWrites, timestamp, "durable_writes")); + cf.addColumn(Column.create(strategyClass.getName(), timestamp, "strategy_class")); + cf.addColumn(Column.create(json(strategyOptions), timestamp, "strategy_options")); for (CFMetaData cfm : cfMetaData.values()) - mutations.add(cfm.dropFromSchema(timestamp)); + cfm.toSchema(rm, timestamp); - return mutations; - } - - public static RowMutation toSchema(KsDef ksDef, long timestamp) throws IOException - { - RowMutation mutation = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksDef.name)); - - for (KsDef._Fields field : KsDef._Fields.values()) - { - if (field.equals(KsDef._Fields.CF_DEFS)) - continue; - - mutation.add(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF, - null, - AsciiType.instance.fromString(field.getFieldName())), - valueAsBytes(ksDef.getFieldValue(field)), - timestamp); - } - - if (!ksDef.isSetCf_defs()) - return mutation; - - for (CfDef cf : ksDef.cf_defs) - { - try - { - CFMetaData.toSchema(mutation, cf, timestamp); - } - catch (ConfigurationException e) - { - throw new IOException(e); - } - } - - return mutation; - } - - public RowMutation toSchema(long timestamp) throws IOException - { - return toSchema(toThrift(), timestamp); + return rm; } /** * Deserialize only Keyspace attributes without nested ColumnFamilies * - * @param serializedKsDef Keyspace attributes in serialized form + * @param row Keyspace attributes in serialized form * * @return deserialized keyspace without cf_defs * * @throws IOException if deserialization failed */ - public static KsDef fromSchema(ColumnFamily serializedKsDef) throws IOException + public static KSMetaData fromSchema(Row row, Iterable cfms) throws IOException { - KsDef ksDef = new KsDef(); - - AbstractType comparator = serializedKsDef.getComparator(); - - for (IColumn ksAttr : serializedKsDef.getSortedColumns()) + UntypedResultSet.Row result = QueryProcessor.resultify("SELECT * FROM system.schema_keyspaces", row).one(); + try { - if (ksAttr == null || ksAttr.isMarkedForDelete()) - continue; - - KsDef._Fields field = KsDef._Fields.findByName(comparator.getString(ksAttr.name())); - - // this means that given field was deprecated - // but still exists in the serialized schema - if (field == null) - continue; - - ksDef.setFieldValue(field, deserializeValue(ksAttr.value(), getValueClass(KsDef.class, field.getFieldName()))); + return new KSMetaData(result.getString("name"), + AbstractReplicationStrategy.getClass(result.getString("strategy_class")), + fromJsonMap(result.getString("strategy_options")), + result.getBoolean("durable_writes"), + cfms); + } + catch (ConfigurationException e) + { + throw new RuntimeException(e); } - - return ksDef.name == null ? null : ksDef; } /** * Deserialize Keyspace with nested ColumnFamilies * - * @param serializedKsDef Keyspace in serialized form + * @param serializedKs Keyspace in serialized form * @param serializedCFs Collection of the serialized ColumnFamilies * * @return deserialized keyspace with cf_defs * * @throws IOException if deserialization failed */ - public static KSMetaData fromSchema(ColumnFamily serializedKsDef, ColumnFamily serializedCFs) throws IOException + public static KSMetaData fromSchema(Row serializedKs, Row serializedCFs) throws IOException { - KsDef ksDef = fromSchema(serializedKsDef); - - assert ksDef != null; - - Map cfs = deserializeColumnFamilies(serializedCFs); - - try - { - CFMetaData[] cfms = new CFMetaData[cfs.size()]; - - int index = 0; - for (CfDef cfDef : cfs.values()) - cfms[index++] = CFMetaData.fromThrift(cfDef); - - return fromThrift(ksDef, cfms); - } - catch (Exception e) - { - // this is critical because indicates that something is wrong with serialized schema - throw new IOException(e); - } + Map cfs = deserializeColumnFamilies(serializedCFs); + return fromSchema(serializedKs, cfs.values()); } /** * Deserialize ColumnFamilies from low-level schema representation, all of them belong to the same keyspace * - * @param serializedColumnFamilies ColumnFamilies in the serialized form - * + * @param row * @return map containing name of the ColumnFamily and it's metadata for faster lookup */ - public static Map deserializeColumnFamilies(ColumnFamily serializedColumnFamilies) + public static Map deserializeColumnFamilies(Row row) { - Map cfs = new HashMap(); + if (row.cf == null) + return Collections.emptyMap(); - if (serializedColumnFamilies == null) - return cfs; - - AbstractType comparator = serializedColumnFamilies.getComparator(); - - for (IColumn column : serializedColumnFamilies.getSortedColumns()) + Map cfms = new HashMap(); + UntypedResultSet results = QueryProcessor.resultify("SELECT * FROM system.schema_columnfamilies", row); + for (UntypedResultSet.Row result : results) { - if (column == null || column.isMarkedForDelete()) - continue; - - String[] attr = comparator.getString(column.name()).split(":"); - assert attr.length == 2; - - CfDef cfDef = cfs.get(attr[0]); - - if (cfDef == null) - { - cfDef = new CfDef(); - cfs.put(attr[0], cfDef); - } - - CfDef._Fields field = CfDef._Fields.findByName(attr[1]); - - // this means that given field was deprecated - // but still exists in the serialized schema - if (field == null) - continue; - - cfDef.setFieldValue(field, deserializeValue(column.value(), getValueClass(CfDef.class, field.getFieldName()))); + CFMetaData cfm = CFMetaData.fromSchema(result); + cfms.put(cfm.cfName, cfm); } - for (CfDef cfDef : cfs.values()) + for (CFMetaData cfm : cfms.values()) { - for (ColumnDef columnDef : ColumnDefinition.fromSchema(ColumnDefinition.readSchema(cfDef.keyspace, cfDef.name))) - cfDef.addToColumn_metadata(columnDef); + Row columnRow = ColumnDefinition.readSchema(cfm.ksName, cfm.cfName); + for (ColumnDefinition cd : ColumnDefinition.fromSchema(columnRow, cfm.comparator)) + cfm.column_metadata.put(cd.name, cd); } - return cfs; + return cfms; } } diff --git a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java index ad789c3e8c..a5acd6221f 100644 --- a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java +++ b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java @@ -177,7 +177,7 @@ public class CreateColumnFamilyStatement newCFMD.comment(cfProps.getProperty(CFPropDefs.KW_COMMENT)) .readRepairChance(getPropertyDouble(CFPropDefs.KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE)) - .dclocalReadRepairChance(getPropertyDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE)) + .dcLocalReadRepairChance(getPropertyDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE)) .replicateOnWrite(getPropertyBoolean(CFPropDefs.KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE)) .gcGraceSeconds(getPropertyInt(CFPropDefs.KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS)) .defaultValidator(cfProps.getValidator()) diff --git a/src/java/org/apache/cassandra/cql/DropIndexStatement.java b/src/java/org/apache/cassandra/cql/DropIndexStatement.java index 4d7aeee290..6d1bb180c8 100644 --- a/src/java/org/apache/cassandra/cql/DropIndexStatement.java +++ b/src/java/org/apache/cassandra/cql/DropIndexStatement.java @@ -54,7 +54,7 @@ public class DropIndexStatement if (cfDef == null) throw new InvalidRequestException("Index '" + index + "' could not be found in any of the ColumnFamilies of keyspace '" + keyspace + "'"); - return new UpdateColumnFamily(cfDef); + return new UpdateColumnFamily(CFMetaData.fromThrift(cfDef)); } private CfDef getUpdatedCFDef(CfDef cfDef) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java index 78be1d35be..c824561fe1 100644 --- a/src/java/org/apache/cassandra/cql/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java @@ -790,7 +790,7 @@ public class QueryProcessor ThriftValidation.validateCfDef(cf_def, oldCfm); try { - applyMigrationOnStage(new UpdateColumnFamily(cf_def)); + applyMigrationOnStage(new UpdateColumnFamily(CFMetaData.fromThrift(cf_def))); } catch (ConfigurationException e) { @@ -875,7 +875,7 @@ public class QueryProcessor try { - applyMigrationOnStage(new UpdateColumnFamily(alterTable.getCfDef(keyspace))); + applyMigrationOnStage(new UpdateColumnFamily(CFMetaData.fromThrift(alterTable.getCfDef(keyspace)))); } catch (ConfigurationException e) { diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index b054a5b3b4..f643ae8dc1 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -121,6 +121,35 @@ public class QueryProcessor return processStatement(getStatement(queryString, clientState).statement, clientState, Collections.emptyList()); } + public static UntypedResultSet resultify(String queryString, Row row) + { + SelectStatement ss; + try + { + ss = (SelectStatement) getStatement(queryString, null).statement; + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + catch (RecognitionException e) + { + throw new RuntimeException(e); + } + + List cqlRows; + try + { + cqlRows = ss.process(Collections.singletonList(row)); + } + catch (InvalidRequestException e) + { + throw new RuntimeException(e); + } + + return new UntypedResultSet(cqlRows); + } + public static CqlPreparedResult prepare(String queryString, ClientState clientState) throws RecognitionException, InvalidRequestException { diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java new file mode 100644 index 0000000000..15bf7cddf7 --- /dev/null +++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java @@ -0,0 +1,110 @@ +/* + * 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.cql3; + +import java.nio.ByteBuffer; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import com.google.common.collect.AbstractIterator; + +import org.apache.cassandra.db.marshal.BooleanType; +import org.apache.cassandra.db.marshal.DoubleType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.thrift.Column; +import org.apache.cassandra.thrift.CqlRow; +import org.apache.hadoop.io.UTF8; + +/** a utility for doing internal cql-based queries */ +public class UntypedResultSet implements Iterable +{ + private final List cqlRows; + + public UntypedResultSet(List cqlRows) + { + this.cqlRows = cqlRows; + } + + public Row one() + { + if (cqlRows.size() != 1) + throw new IllegalStateException("One row required, " + cqlRows.size() + " found"); + return new Row(cqlRows.get(0)); + } + + public Iterator iterator() + { + return new AbstractIterator() + { + Iterator iter = cqlRows.iterator(); + + protected Row computeNext() + { + if (!iter.hasNext()) + return endOfData(); + return new Row(iter.next()); + } + }; + } + + public static class Row + { + Map data = new HashMap(); + + public Row(CqlRow cqlRow) + { + for (Column column : cqlRow.columns) + data.put(UTF8Type.instance.compose(column.name), column.value); + } + + public boolean has(String column) + { + // Note that containsKey won't work because we may have null values + return data.get(column) != null; + } + + public String getString(String column) + { + return UTF8Type.instance.compose(data.get(column)); + } + + public boolean getBoolean(String column) + { + return BooleanType.instance.compose(data.get(column)); + } + + public int getInt(String column) + { + return Int32Type.instance.compose(data.get(column)); + } + + public double getDouble(String column) + { + return DoubleType.instance.compose(data.get(column)); + } + + public ByteBuffer getBytes(String column) + { + return data.get(column); + } + } +} diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java index 368eb6dca9..8935df725d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java @@ -136,7 +136,7 @@ public class AlterTableStatement extends SchemaAlteringStatement applyPropertiesToCfDef(thriftDef, cfProps); break; } - return new UpdateColumnFamily(thriftDef); + return new UpdateColumnFamily(CFMetaData.fromThrift(thriftDef)); } catch (ConfigurationException e) { diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java index 1b8664105c..e175fed6ff 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java @@ -105,7 +105,7 @@ public class CreateColumnFamilyStatement extends SchemaAlteringStatement newCFMD.comment(properties.get(CFPropDefs.KW_COMMENT)) .readRepairChance(properties.getDouble(CFPropDefs.KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE)) - .dclocalReadRepairChance(properties.getDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE)) + .dcLocalReadRepairChance(properties.getDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE)) .replicateOnWrite(properties.getBoolean(CFPropDefs.KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE)) .gcGraceSeconds(properties.getInt(CFPropDefs.KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS)) .defaultValidator(defaultValidator) diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java index 4572d975f6..cc738c44f8 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java @@ -90,7 +90,7 @@ public class CreateIndexStatement extends SchemaAlteringStatement CFMetaData.addDefaultIndexNames(cf_def); ThriftValidation.validateCfDef(cf_def, oldCfm); - return new UpdateColumnFamily(cf_def); + return new UpdateColumnFamily(CFMetaData.fromThrift(cf_def)); } catch (InvalidRequestException e) { diff --git a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java index 4e896f0331..5d3721a457 100644 --- a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java @@ -54,7 +54,7 @@ public class DropIndexStatement extends SchemaAlteringStatement if (cfDef == null) throw new InvalidRequestException("Index '" + index + "' could not be found in any of the column families of keyspace '" + keyspace() + "'"); - return new UpdateColumnFamily(cfDef); + return new UpdateColumnFamily(CFMetaData.fromThrift(cfDef)); } private CfDef getUpdatedCFDef(CfDef cfDef) throws InvalidRequestException diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index c56bf02a95..88415700dc 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -64,7 +64,6 @@ public abstract class ModificationStatement extends CFStatement implements CQLSt state.hasColumnFamilyAccess(keyspace(), columnFamily(), Permission.WRITE); } - @Override public void validate(ClientState state) throws InvalidRequestException { if (timeToLive < 0) diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 4825385ac2..b95d6baf9b 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -155,15 +155,26 @@ public class SelectStatement implements CQLStatement else { // otherwise create resultset from query results - result.schema = new CqlMetadata(new HashMap(), - new HashMap(), - TypeParser.getShortName(cfDef.cfm.comparator), - TypeParser.getShortName(cfDef.cfm.getDefaultValidator())); + result.schema = createSchema(); result.rows = process(rows, result.schema, variables); return result; } } + public List process(List rows) throws InvalidRequestException + { + assert !parameters.isCount; // not yet needed + return process(rows, createSchema(), Collections.emptyList()); + } + + private CqlMetadata createSchema() + { + return new CqlMetadata(new HashMap(), + new HashMap(), + TypeParser.getShortName(cfDef.cfm.comparator), + TypeParser.getShortName(cfDef.cfm.getDefaultValidator())); + } + public String keyspace() { return cfDef.cfm.ksName; diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java index 5e4eb8c86a..dd699f4c03 100644 --- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java @@ -23,9 +23,7 @@ import java.util.*; import org.apache.cassandra.cql3.*; import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.db.CounterMutation; -import org.apache.cassandra.db.IMutation; -import org.apache.cassandra.db.RowMutation; +import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.LongType; @@ -143,12 +141,6 @@ public class UpdateStatement extends ModificationStatement /** * Compute a row mutation for a single key * - * - * @param keyspace working keyspace - * @param key key to change - * @param metadata information about CF - * - * @param clientState * @return row mutation * * @throws InvalidRequestException on the wrong request @@ -159,6 +151,7 @@ public class UpdateStatement extends ModificationStatement // if true we need to wrap RowMutation into CounterMutation boolean hasCounterColumn = false; RowMutation rm = new RowMutation(cfDef.cfm.ksName, key); + ColumnFamily cf = rm.addOrGet(cfDef.cfm.cfName); if (cfDef.isCompact) { @@ -168,7 +161,7 @@ public class UpdateStatement extends ModificationStatement Operation value = processedColumns.get(cfDef.value.name); if (value == null) throw new InvalidRequestException(String.format("Missing mandatory column %s", cfDef.value)); - hasCounterColumn = addToMutation(clientState, rm, builder.build(), cfDef.value, value, variables); + hasCounterColumn = addToMutation(clientState, cf, builder.build(), cfDef.value, value, variables); } else { @@ -179,7 +172,7 @@ public class UpdateStatement extends ModificationStatement continue; ByteBuffer colName = builder.copy().add(name.name.key).build(); - hasCounterColumn |= addToMutation(clientState, rm, colName, name, value, variables); + hasCounterColumn |= addToMutation(clientState, cf, colName, name, value, variables); } } @@ -187,7 +180,7 @@ public class UpdateStatement extends ModificationStatement } private boolean addToMutation(ClientState clientState, - RowMutation rm, + ColumnFamily cf, ByteBuffer colName, CFDefinition.Name valueDef, Operation value, @@ -195,12 +188,12 @@ public class UpdateStatement extends ModificationStatement { if (value.isUnary()) { - ByteBuffer valueBytes = value.value.getByteBuffer(valueDef.type, variables); validateColumnName(colName); - rm.add(new QueryPath(columnFamily(), null, colName), - valueBytes, - getTimestamp(clientState), - getTimeToLive()); + ByteBuffer valueBytes = value.value.getByteBuffer(valueDef.type, variables); + Column c = timeToLive > 0 + ? new ExpiringColumn(colName, valueBytes, getTimestamp(clientState), timeToLive) + : new Column(colName, valueBytes, getTimestamp(clientState)); + cf.addColumn(c); return false; } else @@ -226,7 +219,7 @@ public class UpdateStatement extends ModificationStatement else val = -val; } - rm.addCounter(new QueryPath(columnFamily(), null, colName), val); + cf.addCounter(new QueryPath(columnFamily(), null, colName), val); return true; } } diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index f72bcab7c9..9697ac61be 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/src/java/org/apache/cassandra/db/Column.java @@ -21,14 +21,15 @@ package org.apache.cassandra.db; import java.io.IOException; import java.nio.ByteBuffer; import java.security.MessageDigest; +import java.util.ArrayList; import java.util.Collection; +import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.db.marshal.AbstractType; -import org.apache.cassandra.db.marshal.MarshalException; +import org.apache.cassandra.db.marshal.*; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.utils.Allocator; import org.apache.cassandra.utils.ByteBufferUtil; @@ -285,5 +286,48 @@ public class Column implements IColumn { return getLocalDeletionTime() < gcBefore; } + + public static Column create(String value, long timestamp, String... names) + { + return new Column(decomposeName(names), UTF8Type.instance.decompose(value), timestamp); + } + + public static Column create(int value, long timestamp, String... names) + { + return new Column(decomposeName(names), Int32Type.instance.decompose(value), timestamp); + } + + public static Column create(boolean value, long timestamp, String... names) + { + return new Column(decomposeName(names), BooleanType.instance.decompose(value), timestamp); + } + + public static Column create(double value, long timestamp, String... names) + { + return new Column(decomposeName(names), DoubleType.instance.decompose(value), timestamp); + } + + public static Column create(ByteBuffer value, long timestamp, String... names) + { + return new Column(decomposeName(names), value, timestamp); + } + + static ByteBuffer decomposeName(String... names) + { + assert names.length > 0; + + if (names.length == 1) + return UTF8Type.instance.decompose(names[0]); + + // not super performant. at this time, only infrequently called schema code uses this. + List> types = new ArrayList>(names.length); + for (int i = 0; i < names.length; i++) + types.add(UTF8Type.instance); + + CompositeType.Builder builder = new CompositeType.Builder(CompositeType.getInstance(types)); + for (String name : names) + builder.add(UTF8Type.instance.decompose(name)); + return builder.build(); + } } diff --git a/src/java/org/apache/cassandra/db/DefsTable.java b/src/java/org/apache/cassandra/db/DefsTable.java index f656a33d8e..05e4519957 100644 --- a/src/java/org/apache/cassandra/db/DefsTable.java +++ b/src/java/org/apache/cassandra/db/DefsTable.java @@ -32,10 +32,7 @@ import org.apache.avro.io.BinaryDecoder; import org.apache.avro.io.DecoderFactory; import org.apache.avro.specific.SpecificDatumReader; import org.apache.avro.specific.SpecificRecord; -import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.config.ConfigurationException; -import org.apache.cassandra.config.KSMetaData; -import org.apache.cassandra.config.Schema; +import org.apache.cassandra.config.*; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.AsciiType; @@ -153,16 +150,16 @@ public class DefsTable if (row.cf == null || row.cf.isEmpty() || row.cf.isMarkedForDelete()) continue; - keyspaces.add(KSMetaData.fromSchema(row.cf, serializedColumnFamilies(row.key))); + keyspaces.add(KSMetaData.fromSchema(row, serializedColumnFamilies(row.key))); } return keyspaces; } - private static ColumnFamily serializedColumnFamilies(DecoratedKey ksNameKey) + private static Row serializedColumnFamilies(DecoratedKey ksNameKey) { ColumnFamilyStore cfsStore = SystemTable.schemaCFS(SystemTable.SCHEMA_COLUMNFAMILIES_CF); - return cfsStore.getColumnFamily(QueryFilter.getIdentityFilter(ksNameKey, new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF))); + return new Row(ksNameKey, cfsStore.getColumnFamily(QueryFilter.getIdentityFilter(ksNameKey, new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF)))); } /** @@ -198,15 +195,15 @@ public class DefsTable if (column.name().equals(DEFINITION_SCHEMA_COLUMN_NAME)) continue; KsDef ks = deserializeAvro(schema, column.value(), new KsDef()); - keyspaces.add(KSMetaData.fromAvro(ks)); + keyspaces.add(Avro.ksFromAvro(ks)); } // store deserialized keyspaces into new place dumpToStorage(keyspaces); logger.info("Truncating deprecated system column families (migrations, schema)..."); - MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.MIGRATIONS_CF); - MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.SCHEMA_CF); + MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.MIGRATIONS_CF, -1, false); + MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.SCHEMA_CF, -1, false); } return keyspaces; @@ -246,9 +243,9 @@ public class DefsTable Set keyspacesToDrop = mergeKeyspaces(oldKeyspaces, SystemTable.getSchema(SystemTable.SCHEMA_KEYSPACES_CF)); mergeColumnFamilies(oldColumnFamilies, SystemTable.getSchema(SystemTable.SCHEMA_COLUMNFAMILIES_CF)); - // it is save to drop a keyspace only when all nested ColumnFamilies where deleted + // it is safe to drop a keyspace only when all nested ColumnFamilies where deleted for (String keyspaceToDrop : keyspacesToDrop) - MigrationHelper.dropKeyspace(keyspaceToDrop); + MigrationHelper.dropKeyspace(keyspaceToDrop, -1, false); } private static Set mergeKeyspaces(Map old, Map updated) @@ -266,7 +263,10 @@ public class DefsTable // we don't care about nested ColumnFamilies here because those are going to be processed separately if (!ksAttrs.isEmpty()) - MigrationHelper.addKeyspace(KSMetaData.fromSchema(entry.getValue(), null)); + { + KSMetaData ksm = KSMetaData.fromSchema(new Row(entry.getKey(), entry.getValue()), Collections.emptyList()); + MigrationHelper.addKeyspace(ksm, -1, false); + } } /** @@ -287,7 +287,8 @@ public class DefsTable if (prevValue.isEmpty()) { - MigrationHelper.addKeyspace(KSMetaData.fromSchema(newValue, null)); + KSMetaData ksm = KSMetaData.fromSchema(new Row(entry.getKey(), newValue), Collections.emptyList()); + MigrationHelper.addKeyspace(ksm, -1, false); continue; } @@ -310,9 +311,14 @@ public class DefsTable ColumnFamily newState = valueDiff.rightValue(); if (newState.isEmpty()) + { keyspacesToDrop.add(AsciiType.instance.getString(key.key)); + } else - MigrationHelper.updateKeyspace(KSMetaData.fromSchema(newState)); + { + KSMetaData ksm = KSMetaData.fromSchema(new Row(key, newState), Collections.emptyList()); + MigrationHelper.updateKeyspace(ksm, -1, false); + } } return keyspacesToDrop; @@ -331,10 +337,10 @@ public class DefsTable if (!cfAttrs.isEmpty()) { - Map cfDefs = KSMetaData.deserializeColumnFamilies(cfAttrs); + Map cfDefs = KSMetaData.deserializeColumnFamilies(new Row(entry.getKey(), cfAttrs)); - for (CfDef cfDef : cfDefs.values()) - MigrationHelper.addColumnFamily(cfDef); + for (CFMetaData cfDef : cfDefs.values()) + MigrationHelper.addColumnFamily(cfDef, -1, false); } } @@ -347,37 +353,38 @@ public class DefsTable ColumnFamily prevValue = valueDiff.leftValue(); // state before external modification ColumnFamily newValue = valueDiff.rightValue(); // updated state + Row newRow = new Row(keyspace, newValue); if (prevValue.isEmpty()) // whole keyspace was deleted and now it's re-created { - for (CfDef cfDef : KSMetaData.deserializeColumnFamilies(newValue).values()) - MigrationHelper.addColumnFamily(cfDef); + for (CFMetaData cfm : KSMetaData.deserializeColumnFamilies(newRow).values()) + MigrationHelper.addColumnFamily(cfm, -1, false); } else if (newValue.isEmpty()) // whole keyspace is deleted { - for (CfDef cfDef : KSMetaData.deserializeColumnFamilies(prevValue).values()) - MigrationHelper.dropColumnFamily(cfDef.keyspace, cfDef.name); + for (CFMetaData cfm : KSMetaData.deserializeColumnFamilies(new Row(keyspace, prevValue)).values()) + MigrationHelper.dropColumnFamily(cfm.ksName, cfm.cfName, -1, false); } else // has modifications in the nested ColumnFamilies, need to perform nested diff to determine what was really changed { String ksName = AsciiType.instance.getString(keyspace.key); - Map oldCfDefs = new HashMap(); + Map oldCfDefs = new HashMap(); for (CFMetaData cfm : Schema.instance.getKSMetaData(ksName).cfMetaData().values()) - oldCfDefs.put(cfm.cfName, cfm.toThrift()); + oldCfDefs.put(cfm.cfName, cfm); - Map newCfDefs = KSMetaData.deserializeColumnFamilies(newValue); + Map newCfDefs = KSMetaData.deserializeColumnFamilies(newRow); - MapDifference cfDefDiff = Maps.difference(oldCfDefs, newCfDefs); + MapDifference cfDefDiff = Maps.difference(oldCfDefs, newCfDefs); - for (CfDef cfDef : cfDefDiff.entriesOnlyOnRight().values()) - MigrationHelper.addColumnFamily(cfDef); + for (CFMetaData cfDef : cfDefDiff.entriesOnlyOnRight().values()) + MigrationHelper.addColumnFamily(cfDef, -1, false); - for (CfDef cfDef : cfDefDiff.entriesOnlyOnLeft().values()) - MigrationHelper.dropColumnFamily(cfDef.keyspace, cfDef.name); + for (CFMetaData cfDef : cfDefDiff.entriesOnlyOnLeft().values()) + MigrationHelper.dropColumnFamily(cfDef.ksName, cfDef.cfName, -1, false); - for (MapDifference.ValueDifference cfDef : cfDefDiff.entriesDiffering().values()) - MigrationHelper.updateColumnFamily(cfDef.rightValue()); + for (MapDifference.ValueDifference cfDef : cfDefDiff.entriesDiffering().values()) + MigrationHelper.updateColumnFamily(cfDef.rightValue(), -1, false); } } } diff --git a/src/java/org/apache/cassandra/db/DeletedColumn.java b/src/java/org/apache/cassandra/db/DeletedColumn.java index adeff0c32d..f67ffc809d 100644 --- a/src/java/org/apache/cassandra/db/DeletedColumn.java +++ b/src/java/org/apache/cassandra/db/DeletedColumn.java @@ -85,4 +85,9 @@ public class DeletedColumn extends Column if (getLocalDeletionTime() < 0) throw new MarshalException("The local deletion time should not be negative"); } + + public static DeletedColumn create(int localDeletionTime, long timestamp, String... names) + { + return new DeletedColumn(decomposeName(names), localDeletionTime, timestamp); + } } diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 773877bd5e..590cf1fdb4 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -169,6 +169,21 @@ public class RowMutation implements IMutation, MessageProducer throw new IllegalArgumentException("ColumnFamily " + columnFamily + " already has modifications in this mutation: " + prev); } + /** + * @return the ColumnFamily in this RowMutation corresponding to @param cfName, creating an empty one if necessary. + */ + public ColumnFamily addOrGet(String cfName) + { + CFMetaData cfm = Schema.instance.getCFMetaData(table_, cfName); + ColumnFamily cf = modifications_.get(cfm.cfId); + if (cf == null) + { + cf = ColumnFamily.create(cfm); + modifications_.put(cfm.cfId, cf); + } + return cf; + } + public boolean isEmpty() { return modifications_.isEmpty(); @@ -186,6 +201,10 @@ public class RowMutation implements IMutation, MessageProducer * param @ value - value associated with the column * param @ timestamp - timestamp associated with this data. * param @ timeToLive - ttl for the column, 0 for standard (non expiring) columns + * + * @Deprecated this tends to be low-performance; we're doing two hash lookups, + * one of which instantiates a Pair, and callers tend to instantiate new QP objects + * for each call as well. Use the add(ColumnFamily) overload instead. */ public void add(QueryPath path, ByteBuffer value, long timestamp, int timeToLive) { diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java index cab8d7f808..08b177732c 100644 --- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java +++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java @@ -101,12 +101,8 @@ public abstract class AbstractType implements Comparator /** get a string representation of the bytes suitable for log messages */ public abstract String getString(ByteBuffer bytes); - /** get a byte representation of the given string. - * defaults to unsupportedoperation so people deploying custom Types can update at their leisure. */ - public ByteBuffer fromString(String source) throws MarshalException - { - throw new UnsupportedOperationException(); - } + /** get a byte representation of the given string. */ + public abstract ByteBuffer fromString(String source) throws MarshalException; /* validate that the byte array is a valid sequence for the type we are supposed to be comparing */ public abstract void validate(ByteBuffer bytes) throws MarshalException; diff --git a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java index 154aaf814a..0f0127a994 100644 --- a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java @@ -329,6 +329,11 @@ public class DynamicCompositeType extends AbstractCompositeType throw new UnsupportedOperationException(); } + public ByteBuffer fromString(String str) + { + throw new UnsupportedOperationException(); + } + public void validate(ByteBuffer bytes) { throw new UnsupportedOperationException(); diff --git a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java index d3fd777664..c54e4ab7e6 100644 --- a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java @@ -46,9 +46,9 @@ public class AddColumnFamily extends Migration this.cfm = cfm; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - return MigrationHelper.addColumnFamily(cfm, timestamp); + return MigrationHelper.addColumnFamily(cfm, timestamp, true); } @Override diff --git a/src/java/org/apache/cassandra/db/migration/AddKeyspace.java b/src/java/org/apache/cassandra/db/migration/AddKeyspace.java index 432ebed981..36c4de94d8 100644 --- a/src/java/org/apache/cassandra/db/migration/AddKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/AddKeyspace.java @@ -46,9 +46,9 @@ public class AddKeyspace extends Migration this.ksm = ksm; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - return MigrationHelper.addKeyspace(ksm, timestamp); + return MigrationHelper.addKeyspace(ksm, timestamp, true); } @Override diff --git a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java index 727674f922..0628824c81 100644 --- a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java @@ -44,9 +44,9 @@ public class DropColumnFamily extends Migration this.cfName = cfName; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp); + return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp, true); } @Override diff --git a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java index 18c7f5e91e..d6e46a4f0d 100644 --- a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java @@ -41,9 +41,9 @@ public class DropKeyspace extends Migration this.name = name; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - return MigrationHelper.dropKeyspace(name, timestamp); + return MigrationHelper.dropKeyspace(name, timestamp, true); } @Override diff --git a/src/java/org/apache/cassandra/db/migration/Migration.java b/src/java/org/apache/cassandra/db/migration/Migration.java index 4aaad5b0e6..9b239bb7d1 100644 --- a/src/java/org/apache/cassandra/db/migration/Migration.java +++ b/src/java/org/apache/cassandra/db/migration/Migration.java @@ -21,6 +21,7 @@ package org.apache.cassandra.db.migration; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collection; +import java.util.Collections; import java.util.UUID; import org.slf4j.Logger; @@ -64,26 +65,25 @@ public abstract class Migration public final void apply() throws ConfigurationException, IOException { - Collection mutations = applyImpl(); - - assert !mutations.isEmpty(); + RowMutation mutation = applyImpl(); + assert mutation != null; if (!StorageService.instance.isClientMode()) MigrationHelper.flushSchemaCFs(); Schema.instance.updateVersion(); - announce(mutations); + announce(Collections.singletonList(mutation)); } /** * Class specific apply implementation where schema migration logic should be put * - * @return mutations to update native schema + * @return mutation to update native schema * * @throws IOException on any I/O related error. * @throws ConfigurationException if there is object misconfiguration. */ - protected abstract Collection applyImpl() throws ConfigurationException, IOException; + protected abstract RowMutation applyImpl() throws ConfigurationException, IOException; /** * Send schema update (in form of row mutations) to alive nodes in the cluster. diff --git a/src/java/org/apache/cassandra/db/migration/MigrationHelper.java b/src/java/org/apache/cassandra/db/migration/MigrationHelper.java index c8d94daa78..cbda1bf394 100644 --- a/src/java/org/apache/cassandra/db/migration/MigrationHelper.java +++ b/src/java/org/apache/cassandra/db/migration/MigrationHelper.java @@ -17,6 +17,8 @@ */ package org.apache.cassandra.db.migration; +import java.io.DataOutput; +import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; @@ -32,19 +34,13 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.SystemTable; import org.apache.cassandra.db.Table; -import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.thrift.CfDef; -import org.apache.cassandra.thrift.InvalidRequestException; -import org.apache.cassandra.thrift.KsDef; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; -import org.codehaus.jackson.map.ObjectMapper; - public class MigrationHelper { - private static final ObjectMapper jsonMapper = new ObjectMapper(); private static final Map, Class> primitiveToWrapper = new HashMap, Class>(); static { @@ -58,119 +54,17 @@ public class MigrationHelper primitiveToWrapper.put(double.class, Double.class); } - public static ByteBuffer readableColumnName(ByteBuffer columnName, AbstractType comparator) + public static ByteBuffer searchComposite(String name, boolean start) { - return ByteBufferUtil.bytes(comparator.getString(columnName)); - } - - public static ByteBuffer valueAsBytes(Object value) - { - try - { - return ByteBuffer.wrap(jsonMapper.writeValueAsBytes(value)); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - public static Object deserializeValue(ByteBuffer value, Class valueClass) - { - try - { - // because jackson serialized ByteBuffer as byte[] and needs help with deserialization later - if (valueClass.equals(ByteBuffer.class)) - { - byte[] bvalue = (byte[]) deserializeValue(value, byte[].class); - return bvalue == null ? null : ByteBuffer.wrap(bvalue); - } - - return jsonMapper.readValue(ByteBufferUtil.getArray(value), valueClass); - } - catch (Exception e) - { - throw new RuntimeException(e); - } - } - - public static Class getValueClass(Class klass, String name) - { - try - { - // We want to keep null values, so we must not return a primitive type - return maybeConvertToWrapperClass(klass.getField(name).getType()); - } - catch (NoSuchFieldException e) - { - throw new RuntimeException(e); // never happens - } - } - - private static Class maybeConvertToWrapperClass(Class klass) - { - Class cl = primitiveToWrapper.get(klass); - return cl == null ? klass : cl; - } - - public static ByteBuffer searchComposite(String comp1, boolean start) - { - return compositeNameFor(comp1, !start, null, false, null, false); - } - - public static ByteBuffer compositeNameFor(String comp1, String comp2) - { - return compositeNameFor(comp1, ByteBufferUtil.bytes(comp2), null); - } - - public static ByteBuffer compositeNameFor(String comp1, ByteBuffer comp2, String comp3) - { - return compositeNameFor(comp1, false, comp2, false, comp3, false); - } - - public static ByteBuffer compositeNameFor(String comp1, boolean limit1, ByteBuffer comp2, boolean limit2, String comp3, boolean limit3) - { - int totalSize = 0; - - if (comp1 != null) - totalSize += 2 + comp1.length() + 1; - - if (comp2 != null) - totalSize += 2 + comp2.remaining() + 1; - - if (comp3 != null) - totalSize += 2 + comp3.length() + 1; - - ByteBuffer bytes = ByteBuffer.allocate(totalSize); - - if (comp1 != null) - { - bytes.putShort((short) comp1.length()); - bytes.put(comp1.getBytes()); - bytes.put((byte) (limit1 ? 1 : 0)); - } - - if (comp2 != null) - { - int pos = comp2.position(), limit = comp2.limit(); - - bytes.putShort((short) comp2.remaining()); - bytes.put(comp2); - bytes.put((byte) (limit2 ? 1 : 0)); - // restore original range - comp2.position(pos).limit(limit); - } - - if (comp3 != null) - { - bytes.putShort((short) comp3.length()); - bytes.put(comp3.getBytes()); - bytes.put((byte) (limit3 ? 1 : 0)); - } - - bytes.rewind(); - - return bytes; + assert name != null; + ByteBuffer nameBytes = UTF8Type.instance.decompose(name); + int length = nameBytes.remaining(); + byte[] bytes = new byte[2 + length + 1]; + bytes[0] = (byte)((length >> 8) & 0xFF); + bytes[1] = (byte)(length & 0xFF); + ByteBufferUtil.arrayCopy(nameBytes, 0, bytes, 2, length); + bytes[bytes.length - 1] = (byte)(start ? 0 : 1); + return ByteBuffer.wrap(bytes); } public static void flushSchemaCFs() @@ -188,93 +82,27 @@ public class MigrationHelper FBUtilities.waitOnFuture(flush); } - /* Schema Mutation Helpers */ - - public static Collection addKeyspace(KSMetaData ksm, long timestamp) throws ConfigurationException, IOException - { - return addKeyspace(ksm, timestamp, true); - } - - public static void addKeyspace(KSMetaData ksDef) throws ConfigurationException, IOException - { - addKeyspace(ksDef, -1, false); - } - - public static Collection addColumnFamily(CFMetaData cfm, long timestamp) throws ConfigurationException, IOException - { - return addColumnFamily(cfm, timestamp, true); - } - - public static void addColumnFamily(CfDef cfDef) throws ConfigurationException, IOException - { - try - { - addColumnFamily(CFMetaData.fromThrift(cfDef), -1, false); - } - catch (InvalidRequestException e) - { - throw new ConfigurationException(e.getMessage(), e); - } - } - - public static void updateKeyspace(KsDef newState) throws ConfigurationException, IOException - { - updateKeyspace(newState, -1, false); - } - - public static Collection updateKeyspace(KsDef newState, long timestamp) throws ConfigurationException, IOException - { - return updateKeyspace(newState, timestamp, true); - } - - public static void updateColumnFamily(CfDef newState) throws ConfigurationException, IOException - { - updateColumnFamily(newState, -1, false); - } - - public static Collection updateColumnFamily(CfDef newState, long timestamp) throws ConfigurationException, IOException - { - return updateColumnFamily(newState, timestamp, true); - } - - public static void dropColumnFamily(String ksName, String cfName) throws IOException - { - dropColumnFamily(ksName, cfName, -1, false); - } - - public static Collection dropColumnFamily(String ksName, String cfName, long timestamp) throws IOException - { - return dropColumnFamily(ksName, cfName, timestamp, true); - } - - public static void dropKeyspace(String ksName) throws IOException - { - dropKeyspace(ksName, -1, false); - } - - public static Collection dropKeyspace(String ksName, long timestamp) throws IOException - { - return dropKeyspace(ksName, timestamp, true); - } - /* Migration Helper implementations */ - private static Collection addKeyspace(KSMetaData ksm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException + public static RowMutation addKeyspace(KSMetaData ksm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException { - RowMutation keyspaceDef = ksm.toSchema(timestamp); + RowMutation mutation = null; if (withSchemaRecord) - keyspaceDef.apply(); + { + mutation = ksm.toSchema(timestamp); + mutation.apply(); + } Schema.instance.load(ksm); if (!StorageService.instance.isClientMode()) Table.open(ksm.name); - return toCollection(keyspaceDef); + return mutation; } - private static Collection addColumnFamily(CFMetaData cfm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException + public static RowMutation addColumnFamily(CFMetaData cfm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException { KSMetaData ksm = Schema.instance.getTableDefinition(cfm.ksName); ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm))); @@ -298,19 +126,19 @@ public class MigrationHelper if (!StorageService.instance.isClientMode()) Table.open(ksm.name).initCf(cfm.cfId, cfm.cfName); - return toCollection(mutation); + return mutation; } - private static Collection updateKeyspace(KsDef newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException + public static RowMutation updateKeyspace(KSMetaData newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException { KSMetaData oldKsm = Schema.instance.getKSMetaData(newState.name); - RowMutation schemaUpdate = null; + RowMutation mutation = null; if (withSchemaRecord) { - schemaUpdate = oldKsm.diff(newState, timestamp); - schemaUpdate.apply(); + mutation = oldKsm.diff(newState, timestamp); + mutation.apply(); } KSMetaData newKsm = KSMetaData.cloneWith(oldKsm.reloadAttributes(), oldKsm.cfMetaData().values()); @@ -320,19 +148,19 @@ public class MigrationHelper if (!StorageService.instance.isClientMode()) Table.open(newState.name).createReplicationStrategy(newKsm); - return toCollection(schemaUpdate); + return mutation; } - private static Collection updateColumnFamily(CfDef newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException + public static RowMutation updateColumnFamily(CFMetaData newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException { - CFMetaData cfm = Schema.instance.getCFMetaData(newState.keyspace, newState.name); + CFMetaData cfm = Schema.instance.getCFMetaData(newState.ksName, newState.cfName); - RowMutation schemaUpdate = null; + RowMutation mutation = null; if (withSchemaRecord) { - schemaUpdate = cfm.diff(newState, timestamp); - schemaUpdate.apply(); + mutation = cfm.diff(newState, timestamp); + mutation.apply(); } cfm.reload(); @@ -343,10 +171,10 @@ public class MigrationHelper table.getColumnFamilyStore(cfm.cfName).reload(); } - return toCollection(schemaUpdate); + return mutation; } - private static Collection dropKeyspace(String ksName, long timestamp, boolean withSchemaRecord) throws IOException + public static RowMutation dropKeyspace(String ksName, long timestamp, boolean withSchemaRecord) throws IOException { KSMetaData ksm = Schema.instance.getTableDefinition(ksName); String snapshotName = Table.getTimestampedSnapshotName(ksName); @@ -365,23 +193,22 @@ public class MigrationHelper } } - Collection mutations = Collections.emptyList(); + RowMutation mutation = null; if (withSchemaRecord) { - mutations = ksm.dropFromSchema(timestamp); - for (RowMutation m : mutations) - m.apply(); + mutation = ksm.dropFromSchema(timestamp); + mutation.apply(); } // remove the table from the static instances. Table.clear(ksm.name); Schema.instance.clearTableDefinition(ksm); - return mutations; + return mutation; } - private static Collection dropColumnFamily(String ksName, String cfName, long timestamp, boolean withSchemaRecord) throws IOException + public static RowMutation dropColumnFamily(String ksName, String cfName, long timestamp, boolean withSchemaRecord) throws IOException { KSMetaData ksm = Schema.instance.getTableDefinition(ksName); ColumnFamilyStore cfs = Table.open(ksName).getColumnFamilyStore(cfName); @@ -406,7 +233,7 @@ public class MigrationHelper Table.open(ksm.name).dropCf(cfm.cfId); } - return toCollection(mutation); + return mutation; } private static KSMetaData makeNewKeyspaceDefinition(KSMetaData ksm, CFMetaData toExclude) @@ -417,9 +244,4 @@ public class MigrationHelper assert newCfs.size() == ksm.cfMetaData().size() - 1; return KSMetaData.cloneWith(ksm, newCfs); } - - private static Collection toCollection(RowMutation mutation) - { - return mutation == null ? Collections.emptyList() : Collections.singleton(mutation); - } } diff --git a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java index 7852e3362d..40d56e2f3d 100644 --- a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java @@ -20,28 +20,28 @@ package org.apache.cassandra.db.migration; import java.io.IOException; import java.util.Collection; +import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ConfigurationException; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.RowMutation; -import org.apache.cassandra.thrift.CfDef; public class UpdateColumnFamily extends Migration { - private final CfDef newState; + private final CFMetaData newState; - public UpdateColumnFamily(CfDef newState) throws ConfigurationException + public UpdateColumnFamily(CFMetaData newState) throws ConfigurationException { super(System.nanoTime()); - if (Schema.instance.getCFMetaData(newState.keyspace, newState.name) == null) - throw new ConfigurationException(String.format("(ks=%s, cf=%s) cannot be updated because it doesn't exist.", newState.keyspace, newState.name)); + if (Schema.instance.getCFMetaData(newState.ksName, newState.cfName) == null) + throw new ConfigurationException(String.format("(ks=%s, cf=%s) cannot be updated because it doesn't exist.", newState.ksName, newState.cfName)); this.newState = newState; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - return MigrationHelper.updateColumnFamily(newState, timestamp); + return MigrationHelper.updateColumnFamily(newState, timestamp, true); } @Override diff --git a/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java b/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java index c529bb9e9a..267ec1e2ae 100644 --- a/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java @@ -21,19 +21,20 @@ import java.io.IOException; import java.util.Collection; import org.apache.cassandra.config.ConfigurationException; +import org.apache.cassandra.config.KSMetaData; import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.thrift.KsDef; public class UpdateKeyspace extends Migration { - private final KsDef newState; + private final KSMetaData newState; - public UpdateKeyspace(KsDef newState) throws ConfigurationException + public UpdateKeyspace(KSMetaData newState) throws ConfigurationException { super(System.nanoTime()); - if (newState.isSetCf_defs() && newState.getCf_defs().size() > 0) + if (!newState.cfMetaData().isEmpty()) throw new ConfigurationException("Updated keyspace must not contain any column families."); if (Schema.instance.getKSMetaData(newState.name) == null) @@ -42,11 +43,11 @@ public class UpdateKeyspace extends Migration this.newState = newState; } - protected Collection applyImpl() throws ConfigurationException, IOException + protected RowMutation applyImpl() throws ConfigurationException, IOException { - Collection mutations = MigrationHelper.updateKeyspace(newState, timestamp); + RowMutation mutation = MigrationHelper.updateKeyspace(newState, timestamp, true); logger.info("Keyspace updated. Please perform any manual operations."); - return mutations; + return mutation; } @Override diff --git a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java index 8ebe8aa31b..d152edaabf 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java @@ -196,21 +196,6 @@ public class CompressionParameters throw new ConfigurationException("crc_check_chance should be between 0.0 to 1.0"); } - public Map asAvroOptions() - { - Map options = new HashMap(); - for (Map.Entry entry : otherOptions.entrySet()) - options.put(new Utf8(entry.getKey()), new Utf8(entry.getValue())); - - if (sstableCompressor == null) - return options; - - options.put(new Utf8(SSTABLE_COMPRESSION), new Utf8(sstableCompressor.getClass().getName())); - if (chunkLength != null) - options.put(new Utf8(CHUNK_LENGTH_KB), new Utf8(chunkLengthInKB())); - return options; - } - public Map asThriftOptions() { Map options = new HashMap(otherOptions); diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java index 6990d8461e..467b5f97b8 100644 --- a/src/java/org/apache/cassandra/service/MigrationManager.java +++ b/src/java/org/apache/cassandra/service/MigrationManager.java @@ -85,7 +85,7 @@ public class MigrationManager implements IEndpointStateChangeSubscriber public void onRemove(InetAddress endpoint) {} - public static void rectifySchema(UUID theirVersion, final InetAddress endpoint) + private static void rectifySchema(UUID theirVersion, final InetAddress endpoint) { // Can't request migrations from nodes with versions younger than 1.1 if (Gossiper.instance.getVersion(endpoint) < MessagingService.VERSION_11) diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 952a10351f..3c53ac1238 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -1050,7 +1050,7 @@ public class CassandraServer implements Cassandra.Iface try { ThriftValidation.validateKsDef(ks_def); - applyMigrationOnStage(new UpdateKeyspace(ks_def)); + applyMigrationOnStage(new UpdateKeyspace(KSMetaData.fromThrift(ks_def))); return Schema.instance.getVersion().toString(); } catch (ConfigurationException e) @@ -1079,7 +1079,7 @@ public class CassandraServer implements Cassandra.Iface { // ideally, apply() would happen on the stage with the CFMetaData.applyImplicitDefaults(cf_def); - UpdateColumnFamily update = new UpdateColumnFamily(cf_def); + UpdateColumnFamily update = new UpdateColumnFamily(CFMetaData.fromThrift(cf_def)); applyMigrationOnStage(update); return Schema.instance.getVersion().toString(); } diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 34a58e00b0..e52c15822d 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -18,10 +18,7 @@ */ package org.apache.cassandra.utils; -import java.io.DataInput; -import java.io.DataOutput; -import java.io.IOException; -import java.io.InputStream; +import java.io.*; import java.nio.ByteBuffer; import java.nio.charset.CharacterCodingException; import java.nio.charset.Charset; diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 4f13c80b60..38df00e936 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -54,11 +54,16 @@ import org.apache.thrift.TBase; import org.apache.thrift.TDeserializer; import org.apache.thrift.TException; import org.apache.thrift.TSerializer; +import org.codehaus.jackson.JsonFactory; +import org.codehaus.jackson.map.ObjectMapper; +import org.codehaus.jackson.type.TypeReference; public class FBUtilities { private static Logger logger_ = LoggerFactory.getLogger(FBUtilities.class); + private static ObjectMapper jsonMapper = new ObjectMapper(new JsonFactory()); + public static final BigInteger TWO = new BigInteger("2"); private static volatile InetAddress localInetAddress_; @@ -508,6 +513,42 @@ public class FBUtilities return new WrappedCloseableIterator(iterator); } + public static Map fromJsonMap(String json) + { + try + { + return jsonMapper.readValue(json, Map.class); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + public static List fromJsonList(String json) + { + try + { + return jsonMapper.readValue(json, List.class); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + public static String json(Object object) + { + try + { + return jsonMapper.writeValueAsString(object); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + private static final class WrappedCloseableIterator extends AbstractIterator implements CloseableIterator { diff --git a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java index 76d1c4b824..be9c0e10b5 100644 --- a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java +++ b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java @@ -24,14 +24,19 @@ import java.util.HashMap; import org.apache.cassandra.CleanupHelper; import org.apache.cassandra.config.Schema; +import org.apache.cassandra.cql3.QueryProcessor; +import org.apache.cassandra.cql3.UntypedResultSet; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.Row; import org.apache.cassandra.db.RowMutation; import org.apache.cassandra.db.SystemTable; import org.apache.cassandra.db.Table; import org.apache.cassandra.db.marshal.AsciiType; import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.io.compress.*; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.thrift.CfDef; import org.apache.cassandra.thrift.ColumnDef; import org.apache.cassandra.thrift.IndexType; @@ -41,6 +46,9 @@ import org.junit.Test; import static org.junit.Assert.assertEquals; +import java.util.Map; +import java.nio.ByteBuffer; + public class CFMetaDataTest extends CleanupHelper { private static String KEYSPACE = "Keyspace1"; @@ -117,6 +125,8 @@ public class CFMetaDataTest extends CleanupHelper private void checkInverses(CFMetaData cfm) throws Exception { + DecoratedKey k = StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(cfm.ksName)); + // Test thrift conversion assert cfm.equals(CFMetaData.fromThrift(cfm.toThrift())) : String.format("\n%s\n!=\n%s", cfm, CFMetaData.fromThrift(cfm.toThrift())); @@ -124,7 +134,8 @@ public class CFMetaDataTest extends CleanupHelper RowMutation rm = cfm.toSchema(System.currentTimeMillis()); ColumnFamily serializedCf = rm.getColumnFamily(Schema.instance.getId(Table.SYSTEM_TABLE, SystemTable.SCHEMA_COLUMNFAMILIES_CF)); ColumnFamily serializedCD = rm.getColumnFamily(Schema.instance.getId(Table.SYSTEM_TABLE, SystemTable.SCHEMA_COLUMNS_CF)); - CfDef cfDef = CFMetaData.addColumnDefinitionSchema(CFMetaData.fromSchema(serializedCf), serializedCD); - assert cfm.equals(CFMetaData.fromThrift(cfDef)) : String.format("\n%s\n!=\n%s", cfm, CFMetaData.fromThrift(cfDef)); + UntypedResultSet.Row result = QueryProcessor.resultify("SELECT * FROM system.schema_columnfamilies", new Row(k, serializedCf)).one(); + CFMetaData newCfm = CFMetaData.addColumnDefinitionSchema(CFMetaData.fromSchemaNoColumns(result), new Row(k, serializedCD)); + assert cfm.equals(newCfm) : String.format("\n%s\n!=\n%s", cfm, newCfm); } } diff --git a/test/unit/org/apache/cassandra/db/DefsTest.java b/test/unit/org/apache/cassandra/config/DefsTest.java similarity index 84% rename from test/unit/org/apache/cassandra/db/DefsTest.java rename to test/unit/org/apache/cassandra/config/DefsTest.java index 6fb8a01a40..87e75cb791 100644 --- a/test/unit/org/apache/cassandra/db/DefsTest.java +++ b/test/unit/org/apache/cassandra/config/DefsTest.java @@ -16,7 +16,7 @@ * limitations under the License. */ -package org.apache.cassandra.db; +package org.apache.cassandra.config; import java.io.File; import java.io.IOException; @@ -26,7 +26,7 @@ import java.util.concurrent.ExecutionException; import org.apache.cassandra.CleanupHelper; import org.apache.cassandra.Util; -import org.apache.cassandra.config.*; +import org.apache.cassandra.db.*; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.db.marshal.BytesType; @@ -103,25 +103,26 @@ public class DefsTest extends CleanupHelper // we'll be adding this one later. make sure it's not already there. assert cfm.getColumn_metadata().get(ByteBuffer.wrap(new byte[] { 5 })) == null; - CfDef cfDef = cfm.toThrift(); + + CFMetaData cfNew = cfm.clone(); // add one. - ColumnDef addIndexDef = new ColumnDef(); - addIndexDef.index_name = "5"; - addIndexDef.index_type = IndexType.KEYS; - addIndexDef.name = ByteBuffer.wrap(new byte[] { 5 }); - addIndexDef.validation_class = BytesType.class.getName(); - cfDef.column_metadata.add(addIndexDef); + ColumnDefinition addIndexDef = new ColumnDefinition(ByteBuffer.wrap(new byte[] { 5 }), + BytesType.instance, + IndexType.KEYS, + null, + "5"); + cfNew.addColumnDefinition(addIndexDef); // remove one. - ColumnDef removeIndexDef = new ColumnDef(); - removeIndexDef.index_name = "0"; - removeIndexDef.index_type = IndexType.KEYS; - removeIndexDef.name = ByteBuffer.wrap(new byte[] { 0 }); - removeIndexDef.validation_class = BytesType.class.getName(); - assert cfDef.column_metadata.remove(removeIndexDef); + ColumnDefinition removeIndexDef = new ColumnDefinition(ByteBuffer.wrap(new byte[] { 0 }), + BytesType.instance, + IndexType.KEYS, + null, + "0"); + assert cfNew.removeColumnDefinition(removeIndexDef); - cfm.apply(cfDef); + cfm.apply(cfNew); for (int i = 1; i < indexes.size(); i++) assert cfm.getColumn_metadata().get(ByteBuffer.wrap(new byte[] { 1 })) != null; @@ -423,7 +424,7 @@ public class DefsTest extends CleanupHelper KSMetaData newBadKs = KSMetaData.testMetadata(cf.ksName, SimpleStrategy.class, KSMetaData.optsWithRF(4), cf2); try { - new UpdateKeyspace(newBadKs.toThrift()).apply(); + new UpdateKeyspace(newBadKs).apply(); throw new AssertionError("Should not have been able to update a KS with a KS that described column families."); } catch (ConfigurationException ex) @@ -435,7 +436,7 @@ public class DefsTest extends CleanupHelper KSMetaData newBadKs2 = KSMetaData.testMetadata(cf.ksName + "trash", SimpleStrategy.class, KSMetaData.optsWithRF(4)); try { - new UpdateKeyspace(newBadKs2.toThrift()).apply(); + new UpdateKeyspace(newBadKs2).apply(); throw new AssertionError("Should not have been able to update a KS with an invalid KS name."); } catch (ConfigurationException ex) @@ -444,7 +445,7 @@ public class DefsTest extends CleanupHelper } KSMetaData newKs = KSMetaData.testMetadata(cf.ksName, OldNetworkTopologyStrategy.class, KSMetaData.optsWithRF(1)); - new UpdateKeyspace(newKs.toThrift()).apply(); + new UpdateKeyspace(newKs).apply(); KSMetaData newFetchedKs = Schema.instance.getKSMetaData(newKs.name); assert newFetchedKs.strategyClass.equals(newKs.strategyClass); @@ -464,121 +465,88 @@ public class DefsTest extends CleanupHelper assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName) != null; // updating certain fields should fail. - CfDef cf_def = cf.toThrift(); - cf_def.column_metadata = new ArrayList(); - cf_def.default_validation_class ="BytesType"; - cf_def.min_compaction_threshold = 5; - cf_def.max_compaction_threshold = 31; + CFMetaData newCfm = cf.clone(); + newCfm.columnMetadata(new HashMap()); + newCfm.defaultValidator(BytesType.instance); + newCfm.minCompactionThreshold(5); + newCfm.maxCompactionThreshold(31); // test valid operations. - cf_def.comment = "Modified comment"; - new UpdateColumnFamily(cf_def).apply(); // doesn't get set back here. + newCfm.comment("Modified comment"); + new UpdateColumnFamily(newCfm).apply(); // doesn't get set back here. - cf_def.read_repair_chance = 0.23; - new UpdateColumnFamily(cf_def).apply(); + newCfm.readRepairChance(0.23); + new UpdateColumnFamily(newCfm).apply(); - cf_def.gc_grace_seconds = 12; - new UpdateColumnFamily(cf_def).apply(); + newCfm.gcGraceSeconds(12); + new UpdateColumnFamily(newCfm).apply(); - cf_def.default_validation_class = "UTF8Type"; - new UpdateColumnFamily(cf_def).apply(); + newCfm.defaultValidator(UTF8Type.instance); + new UpdateColumnFamily(newCfm).apply(); - cf_def.min_compaction_threshold = 3; - new UpdateColumnFamily(cf_def).apply(); + newCfm.minCompactionThreshold(3); + new UpdateColumnFamily(newCfm).apply(); - cf_def.max_compaction_threshold = 33; - new UpdateColumnFamily(cf_def).apply(); + newCfm.maxCompactionThreshold(33); + new UpdateColumnFamily(newCfm).apply(); // can't test changing the reconciler because there is only one impl. // check the cumulative affect. - assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getComment().equals(cf_def.comment); - assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getReadRepairChance() == cf_def.read_repair_chance; - assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getGcGraceSeconds() == cf_def.gc_grace_seconds; + assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getComment().equals(newCfm.getComment()); + assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getReadRepairChance() == newCfm.getReadRepairChance(); + assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getGcGraceSeconds() == newCfm.getGcGraceSeconds(); assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getDefaultValidator() == UTF8Type.instance; - // todo: we probably don't need to reset old values in the catches anymore. - // make sure some invalid operations fail. - int oldId = cf_def.id; + // Change cfId + newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId + 1); + CFMetaData.copyOpts(newCfm, cf); try { - cf_def.id++; - cf.apply(cf_def); + cf.apply(newCfm); throw new AssertionError("Should have blown up when you used a different id."); } - catch (ConfigurationException expected) - { - cf_def.id = oldId; - } + catch (ConfigurationException expected) {} - String oldStr = cf_def.name; + // Change cfName + newCfm = new CFMetaData(cf.ksName, cf.cfName + "_renamed", cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId); + CFMetaData.copyOpts(newCfm, cf); try { - cf_def.name = cf_def.name + "_renamed"; - cf.apply(cf_def); + cf.apply(newCfm); throw new AssertionError("Should have blown up when you used a different name."); } - catch (ConfigurationException expected) - { - cf_def.name = oldStr; - } + catch (ConfigurationException expected) {} - oldStr = cf_def.keyspace; + // Change ksName + newCfm = new CFMetaData(cf.ksName + "_renamed", cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId); + CFMetaData.copyOpts(newCfm, cf); try { - cf_def.keyspace = oldStr + "_renamed"; - cf.apply(cf_def); + cf.apply(newCfm); throw new AssertionError("Should have blown up when you used a different keyspace."); } - catch (ConfigurationException expected) - { - cf_def.keyspace = oldStr; - } + catch (ConfigurationException expected) {} + // Change cf type + newCfm = new CFMetaData(cf.ksName, cf.cfName, ColumnFamilyType.Super, cf.comparator, cf.subcolumnComparator, cf.cfId); + CFMetaData.copyOpts(newCfm, cf); try { - cf_def.column_type = ColumnFamilyType.Super.name(); - cf.apply(cf_def); + cf.apply(newCfm); throw new AssertionError("Should have blwon up when you used a different cf type."); } - catch (ConfigurationException expected) - { - cf_def.column_type = ColumnFamilyType.Standard.name(); - } + catch (ConfigurationException expected) {} - oldStr = cf_def.comparator_type; + // Change comparator + newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, TimeUUIDType.instance, cf.subcolumnComparator, cf.cfId); + CFMetaData.copyOpts(newCfm, cf); try { - cf_def.comparator_type = TimeUUIDType.class.getSimpleName(); - cf.apply(cf_def); + cf.apply(newCfm); throw new AssertionError("Should have blown up when you used a different comparator."); } - catch (ConfigurationException expected) - { - cf_def.comparator_type = UTF8Type.class.getSimpleName(); - } - - try - { - cf_def.min_compaction_threshold = 34; - cf.apply(cf_def); - throw new AssertionError("Should have blown up when min > max."); - } - catch (ConfigurationException expected) - { - cf_def.min_compaction_threshold = 3; - } - - try - { - cf_def.max_compaction_threshold = 2; - cf.apply(cf_def); - throw new AssertionError("Should have blown up when max > min."); - } - catch (ConfigurationException expected) - { - cf_def.max_compaction_threshold = 33; - } + catch (ConfigurationException expected) {} } @Test @@ -598,11 +566,11 @@ public class DefsTest extends CleanupHelper Descriptor desc = indexedCfs.getSSTables().iterator().next().descriptor; // drop the index - CFMetaData meta = CFMetaData.rename(cfs.metadata, cfs.metadata.cfName); // abusing rename to clone + CFMetaData meta = cfs.metadata.clone(); ColumnDefinition cdOld = meta.getColumn_metadata().values().iterator().next(); ColumnDefinition cdNew = new ColumnDefinition(cdOld.name, cdOld.getValidator(), null, null, null); meta.columnMetadata(Collections.singletonMap(cdOld.name, cdNew)); - UpdateColumnFamily update = new UpdateColumnFamily(meta.toThrift()); + UpdateColumnFamily update = new UpdateColumnFamily(meta); update.apply(); // check