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

Add type information to new schema_ columnfamilies and remove thrift validation 

patch by jbellis and slebresne; reviewed by jbellis and slebresne for CASSANDRA-3792
This commit is contained in:
Jonathan Ellis 2012-02-29 11:50:27 -06:00 committed by Sylvain Lebresne
parent ac8d60b41f
commit ccb0028931
39 changed files with 1160 additions and 1157 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

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

2
NEWS.txt
View File

@ -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.

193
src/java/org/apache/cassandra/config/Avro.java Normal file
View File

@ -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<? extends AbstractReplicationStrategy> repStratClass;
try
{
String strategyClassName = KSMetaData.convertOldStrategyName(ks.strategy_class.toString());
repStratClass = (Class<AbstractReplicationStrategy>)Class.forName(strategyClassName);
}
catch (Exception ex)
{
throw new RuntimeException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex);
}
Map<String, String> strategyOptions = new HashMap<String, String>();
if (ks.strategy_options != null)
{
for (Map.Entry<CharSequence, CharSequence> 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> cfMetaData = new ArrayList<CFMetaData>(cfsz);
Iterator<CfDef> 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<ByteBuffer, ColumnDefinition> column_metadata = new TreeMap<ByteBuffer, ColumnDefinition>(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<ByteBuffer>(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<CharSequence, CharSequence> 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);
}
}
}

731
src/java/org/apache/cassandra/config/CFMetaData.java
View File

@ -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<AbstractType<?>> composite = new ArrayList<AbstractType<?>>(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.<AbstractType<?>>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.<AbstractType<?>>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<ByteBuffer, ColumnDefinition> column_metadata;
Map<ByteBuffer, ColumnDefinition> column_metadata;
public Class<? extends AbstractCompactionStrategy> compactionStrategyClass;
public Map<String, String> 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<ByteBuffer> prop) {columnAliases = prop; updateCfDef(); return this;}
public CFMetaData valueAlias(ByteBuffer prop) {valueAlias = prop; updateCfDef(); return this;}
public CFMetaData columnMetadata(Map<ByteBuffer,ColumnDefinition> prop) {column_metadata = prop; updateCfDef(); return this;}
private CFMetaData columnMetadata(ColumnDefinition... cds)
{
Map<ByteBuffer, ColumnDefinition> map = new HashMap<ByteBuffer, ColumnDefinition>();
for (ColumnDefinition cd : cds)
map.put(cd.name, cd);
return columnMetadata(map);
}
public CFMetaData compactionStrategyClass(Class<? extends AbstractCompactionStrategy> prop) {compactionStrategyClass = prop; return this;}
public CFMetaData compactionStrategyOptions(Map<String, String> 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<ByteBuffer, ColumnDefinition> column_metadata = new TreeMap<ByteBuffer, ColumnDefinition>(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<CharSequence, CharSequence> 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 <T> List<T> fixAvroRetardation(List<T> array)
{
return new ArrayList<T>(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<ColumnDef>());
// adjust column definitions. figure out who is coming and going.
Set<ByteBuffer> toRemove = new HashSet<ByteBuffer>();
Set<ByteBuffer> newColumns = new HashSet<ByteBuffer>();
Set<ColumnDef> toAdd = new HashSet<ColumnDef>();
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<ByteBuffer, ColumnDefinition> 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<String, String>();
for (Map.Entry<String, String> 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<? extends AbstractCompactionStrategy> createCompactionStrategy(String className) throws ConfigurationException
{
className = className.contains(".") ? className : "org.apache.cassandra.db.compaction." + className;
try
{
return (Class<? extends AbstractCompactionStrategy>) 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<ByteBuffer, ColumnDefinition> columnDiff = Maps.difference(column_metadata, ColumnDefinition.fromThrift(newState.column_metadata));
Map<ByteBuffer, ColumnDef> columnDefMap = ColumnDefinition.toMap(newState.column_metadata);
MapDifference<ByteBuffer, ColumnDefinition> 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<String> columnAliasesAsStrings()
{
List<String> aliases = new ArrayList<String>(columnAliases.size());
for (ByteBuffer rawAlias : columnAliases)
aliases.add(UTF8Type.instance.compose(rawAlias));
return aliases;
}
private static List<ByteBuffer> columnAliasesFromStrings(List<String> aliases)
{
List<ByteBuffer> rawAliases = new ArrayList<ByteBuffer>(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 <cf>:<attribute name>
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()

220
src/java/org/apache/cassandra/config/ColumnDefinition.java
View File

@ -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<String, String> 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<ByteBuffer, ColumnDef> toMap(List<ColumnDef> columnDefs)
{
Map<ByteBuffer, ColumnDef> map = new HashMap<ByteBuffer, ColumnDef>();
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<ColumnDef> fromSchema(ColumnFamily columns)
public static List<ColumnDefinition> 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<String, ColumnDef> contenders = new HashMap<String, ColumnDef>();
for (IColumn column : columns.getSortedColumns())
List<ColumnDefinition> cds = new ArrayList<ColumnDefinition>();
for (UntypedResultSet.Row result : QueryProcessor.resultify("SELECT * FROM system.schema_columns", row))
{
if (column.isMarkedForDelete())
continue;
// column name format <cf>:<column name>:<attribute name>
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<String,String> 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<ColumnDef> columnDefs = new ArrayList<ColumnDef>();
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

248
src/java/org/apache/cassandra/config/KSMetaData.java
View File

@ -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<String, CFMetaData> cfMetaData;
public final boolean durableWrites;
private KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, boolean durableWrites, Iterable<CFMetaData> cfDefs)
KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, boolean durableWrites, Iterable<CFMetaData> 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<? extends AbstractReplicationStrategy> repStratClass;
try
{
String strategyClassName = convertOldStrategyName(ks.strategy_class.toString());
repStratClass = (Class<AbstractReplicationStrategy>)Class.forName(strategyClassName);
}
catch (Exception ex)
{
throw new RuntimeException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex);
}
Map<String, String> strategyOptions = new HashMap<String, String>();
if (ks.strategy_options != null)
{
for (Map.Entry<CharSequence, CharSequence> 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> cfMetaData = new ArrayList<CFMetaData>(cfsz);
Iterator<org.apache.cassandra.db.migration.avro.CfDef> 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.<CFMetaData>emptyList());
}
public List<RowMutation> dropFromSchema(long timestamp)
public RowMutation dropFromSchema(long timestamp)
{
List<RowMutation> mutations = new ArrayList<RowMutation>();
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<CFMetaData> 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<String, CfDef> 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<String, CFMetaData> 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<String, CfDef> deserializeColumnFamilies(ColumnFamily serializedColumnFamilies)
public static Map<String, CFMetaData> deserializeColumnFamilies(Row row)
{
Map<String, CfDef> cfs = new HashMap<String, CfDef>();
if (row.cf == null)
return Collections.emptyMap();
if (serializedColumnFamilies == null)
return cfs;
AbstractType<?> comparator = serializedColumnFamilies.getComparator();
for (IColumn column : serializedColumnFamilies.getSortedColumns())
Map<String, CFMetaData> cfms = new HashMap<String, CFMetaData>();
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;
}
}

2
src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
View File

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

2
src/java/org/apache/cassandra/cql/DropIndexStatement.java
View File

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

4
src/java/org/apache/cassandra/cql/QueryProcessor.java
View File

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

29
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -121,6 +121,35 @@ public class QueryProcessor
return processStatement(getStatement(queryString, clientState).statement, clientState, Collections.<ByteBuffer>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<CqlRow> 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
{

110
src/java/org/apache/cassandra/cql3/UntypedResultSet.java Normal file
View File

@ -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<UntypedResultSet.Row>
{
private final List<CqlRow> cqlRows;
public UntypedResultSet(List<CqlRow> 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<Row> iterator()
{
return new AbstractIterator<Row>()
{
Iterator<CqlRow> iter = cqlRows.iterator();
protected Row computeNext()
{
if (!iter.hasNext())
return endOfData();
return new Row(iter.next());
}
};
}
public static class Row
{
Map<String, ByteBuffer> data = new HashMap<String, ByteBuffer>();
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);
}
}
}

2
src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
View File

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

2
src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
View File

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

2
src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
View File

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

2
src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
View File

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

1
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

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

19
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -155,15 +155,26 @@ public class SelectStatement implements CQLStatement
else
{
// otherwise create resultset from query results
result.schema = new CqlMetadata(new HashMap<ByteBuffer, String>(),
new HashMap<ByteBuffer, String>(),
TypeParser.getShortName(cfDef.cfm.comparator),
TypeParser.getShortName(cfDef.cfm.getDefaultValidator()));
result.schema = createSchema();
result.rows = process(rows, result.schema, variables);
return result;
}
}
public List<CqlRow> process(List<Row> rows) throws InvalidRequestException
{
assert !parameters.isCount; // not yet needed
return process(rows, createSchema(), Collections.<ByteBuffer>emptyList());
}
private CqlMetadata createSchema()
{
return new CqlMetadata(new HashMap<ByteBuffer, String>(),
new HashMap<ByteBuffer, String>(),
TypeParser.getShortName(cfDef.cfm.comparator),
TypeParser.getShortName(cfDef.cfm.getDefaultValidator()));
}
public String keyspace()
{
return cfDef.cfm.ksName;

29
src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
View File

@ -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;
}
}

48
src/java/org/apache/cassandra/db/Column.java
View File

@ -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<AbstractType<?>> types = new ArrayList<AbstractType<?>>(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();
}
}

71
src/java/org/apache/cassandra/db/DefsTable.java
View File

@ -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<String> 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<String> mergeKeyspaces(Map<DecoratedKey, ColumnFamily> old, Map<DecoratedKey, ColumnFamily> 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.<CFMetaData>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.<CFMetaData>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.<CFMetaData>emptyList());
MigrationHelper.updateKeyspace(ksm, -1, false);
}
}
return keyspacesToDrop;
@ -331,10 +337,10 @@ public class DefsTable
if (!cfAttrs.isEmpty())
{
Map<String, CfDef> cfDefs = KSMetaData.deserializeColumnFamilies(cfAttrs);
Map<String, CFMetaData> 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<String, CfDef> oldCfDefs = new HashMap<String, CfDef>();
Map<String, CFMetaData> oldCfDefs = new HashMap<String, CFMetaData>();
for (CFMetaData cfm : Schema.instance.getKSMetaData(ksName).cfMetaData().values())
oldCfDefs.put(cfm.cfName, cfm.toThrift());
oldCfDefs.put(cfm.cfName, cfm);
Map<String, CfDef> newCfDefs = KSMetaData.deserializeColumnFamilies(newValue);
Map<String, CFMetaData> newCfDefs = KSMetaData.deserializeColumnFamilies(newRow);
MapDifference<String, CfDef> cfDefDiff = Maps.difference(oldCfDefs, newCfDefs);
MapDifference<String, CFMetaData> 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> cfDef : cfDefDiff.entriesDiffering().values())
MigrationHelper.updateColumnFamily(cfDef.rightValue());
for (MapDifference.ValueDifference<CFMetaData> cfDef : cfDefDiff.entriesDiffering().values())
MigrationHelper.updateColumnFamily(cfDef.rightValue(), -1, false);
}
}
}

5
src/java/org/apache/cassandra/db/DeletedColumn.java
View File

@ -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);
}
}

19
src/java/org/apache/cassandra/db/RowMutation.java
View File

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

8
src/java/org/apache/cassandra/db/marshal/AbstractType.java
View File

@ -101,12 +101,8 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
/** 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;

5
src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
View File

@ -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();

4
src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
View File

@ -46,9 +46,9 @@ public class AddColumnFamily extends Migration
this.cfm = cfm;
}
protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
return MigrationHelper.addColumnFamily(cfm, timestamp);
return MigrationHelper.addColumnFamily(cfm, timestamp, true);
}
@Override

4
src/java/org/apache/cassandra/db/migration/AddKeyspace.java
View File

@ -46,9 +46,9 @@ public class AddKeyspace extends Migration
this.ksm = ksm;
}
protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
return MigrationHelper.addKeyspace(ksm, timestamp);
return MigrationHelper.addKeyspace(ksm, timestamp, true);
}
@Override

4
src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
View File

@ -44,9 +44,9 @@ public class DropColumnFamily extends Migration
this.cfName = cfName;
}
protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp);
return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp, true);
}
@Override

4
src/java/org/apache/cassandra/db/migration/DropKeyspace.java
View File

@ -41,9 +41,9 @@ public class DropKeyspace extends Migration
this.name = name;
}
protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
return MigrationHelper.dropKeyspace(name, timestamp);
return MigrationHelper.dropKeyspace(name, timestamp, true);
}
@Override

12
src/java/org/apache/cassandra/db/migration/Migration.java
View File

@ -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<RowMutation> 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<RowMutation> 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.

258
src/java/org/apache/cassandra/db/migration/MigrationHelper.java
View File

@ -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<?>, Class<?>> primitiveToWrapper = new HashMap<Class<?>, 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> dropKeyspace(String ksName, long timestamp) throws IOException
{
return dropKeyspace(ksName, timestamp, true);
}
/* Migration Helper implementations */
private static Collection<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> 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<RowMutation> toCollection(RowMutation mutation)
{
return mutation == null ? Collections.<RowMutation>emptyList() : Collections.singleton(mutation);
}
}

14
src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
View File

@ -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<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
return MigrationHelper.updateColumnFamily(newState, timestamp);
return MigrationHelper.updateColumnFamily(newState, timestamp, true);
}
@Override

13
src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
View File

@ -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<RowMutation> applyImpl() throws ConfigurationException, IOException
protected RowMutation applyImpl() throws ConfigurationException, IOException
{
Collection<RowMutation> 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

15
src/java/org/apache/cassandra/io/compress/CompressionParameters.java
View File

@ -196,21 +196,6 @@ public class CompressionParameters
throw new ConfigurationException("crc_check_chance should be between 0.0 to 1.0");
}
public Map<CharSequence, CharSequence> asAvroOptions()
{
Map<CharSequence, CharSequence> options = new HashMap<CharSequence, CharSequence>();
for (Map.Entry<String, String> 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<String, String> asThriftOptions()
{
Map<String, String> options = new HashMap<String, String>(otherOptions);

2
src/java/org/apache/cassandra/service/MigrationManager.java
View File

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

4
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -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();
}

5
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -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;

41
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -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<T>(iterator);
}
public static Map<String, String> fromJsonMap(String json)
{
try
{
return jsonMapper.readValue(json, Map.class);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
public static List<String> 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<T>
extends AbstractIterator<T> implements CloseableIterator<T>
{

15
test/unit/org/apache/cassandra/config/CFMetaDataTest.java
View File

@ -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);
}
}

166
test/unit/org/apache/cassandra/db/DefsTest.java → test/unit/org/apache/cassandra/config/DefsTest.java
View File

@ -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<ColumnDef>();
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<ByteBuffer, ColumnDefinition>());
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