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

fix flush and compaction races with schema updates. patch by gdusbabek and jbellis. CASSANDRA-1715 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.7@1036937 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Dusbabek 2010-11-19 16:55:37 +00:00
parent 53b49b1180
commit 477eecb1cc
28 changed files with 761 additions and 449 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
dev
* fix compaction and flush races with schema updates (CASSANDRA-1715)
* add clustertool, config-converter, sstablekeys, and schematool
Windows .bat files (CASSANDRA-1723)
* reject range queries received during bootstrap (CASSANDRA-1739)

5
src/avro/internode.genavro
View File

@ -42,13 +42,11 @@ protocol InterNode {
record DropColumnFamily {
string ksname;
string cfname;
boolean block_on_deletion;
}
@namespace("org.apache.cassandra.db.migration.avro")
record DropKeyspace {
string ksname;
boolean block_on_deletion;
}
@namespace("org.apache.cassandra.db.migration.avro")
@ -73,8 +71,7 @@ protocol InterNode {
@namespace("org.apache.cassandra.db.migration.avro")
record UpdateColumnFamily {
org.apache.cassandra.avro.CfDef oldCf;
org.apache.cassandra.avro.CfDef newCf;
org.apache.cassandra.avro.CfDef metadata;
}
@namespace("org.apache.cassandra.db.migration.avro")

8
src/java/org/apache/cassandra/avro/CassandraServer.java
View File

@ -694,8 +694,8 @@ public class CassandraServer implements Cassandra {
try
{
CFMetaData newCfm = oldCfm.apply(cf_def);
UpdateColumnFamily update = new UpdateColumnFamily(oldCfm, newCfm);
oldCfm.apply(cf_def);
UpdateColumnFamily update = new UpdateColumnFamily(cf_def);
applyMigrationOnStage(update);
return DatabaseDescriptor.getDefsVersion().toString();
}
@ -878,7 +878,7 @@ public class CassandraServer implements Cassandra {
try
{
applyMigrationOnStage(new DropColumnFamily(state().getKeyspace(), column_family.toString(), true));
applyMigrationOnStage(new DropColumnFamily(state().getKeyspace(), column_family.toString()));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
@ -898,7 +898,7 @@ public class CassandraServer implements Cassandra {
try
{
applyMigrationOnStage(new DropKeyspace(keyspace.toString(), true));
applyMigrationOnStage(new DropKeyspace(keyspace.toString()));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)

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

@ -21,9 +21,10 @@ package org.apache.cassandra.config;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;
@ -147,22 +148,22 @@ public final class CFMetaData
public final AbstractType subcolumnComparator; // like comparator, for supercolumns
//OPTIONAL
public final String comment; // default none, for humans only
public final double rowCacheSize; // default 0
public final double keyCacheSize; // default 0.01
public final double readRepairChance; // default 1.0 (always), chance [0.0,1.0] of read repair
public final int gcGraceSeconds; // default 864000 (ten days)
public final AbstractType defaultValidator; // default none, use comparator types
public final Integer minCompactionThreshold; // default 4
public final Integer maxCompactionThreshold; // default 32
public final int rowCacheSavePeriodInSeconds; // default 0 (off)
public final int keyCacheSavePeriodInSeconds; // default 3600 (1 hour)
public final int memtableFlushAfterMins; // default 60
public final int memtableThroughputInMb; // default based on heap size
public final double memtableOperationsInMillions; // default based on throughput
private String comment; // default none, for humans only
private double rowCacheSize; // default 0
private double keyCacheSize; // default 0.01
private double readRepairChance; // default 1.0 (always), chance [0.0,1.0] of read repair
private int gcGraceSeconds; // default 864000 (ten days)
private AbstractType defaultValidator; // default none, use comparator types
private Integer minCompactionThreshold; // default 4
private Integer maxCompactionThreshold; // default 32
private int rowCacheSavePeriodInSeconds; // default 0 (off)
private int keyCacheSavePeriodInSeconds; // default 3600 (1 hour)
private int memtableFlushAfterMins; // default 60
private int memtableThroughputInMb; // default based on heap size
private double memtableOperationsInMillions; // default based on throughput
// NOTE: if you find yourself adding members to this class, make sure you keep the convert methods in lockstep.
public final Map<ByteBuffer, ColumnDefinition> column_metadata;
private final Map<ByteBuffer, ColumnDefinition> column_metadata;
private CFMetaData(String tableName,
String cfName,
@ -275,7 +276,7 @@ public final class CFMetaData
public static CFMetaData newIndexMetadata(String table, String parentCf, ColumnDefinition info, AbstractType columnComparator)
{
return new CFMetaData(table,
parentCf + "." + (info.index_name == null ? FBUtilities.bytesToHex(info.name) : info.index_name),
parentCf + "." + (info.getIndexName() == null ? FBUtilities.bytesToHex(info.name) : info.getIndexName()),
ColumnFamilyType.Standard,
columnComparator,
null,
@ -436,7 +437,77 @@ public final class CFMetaData
cf.id,
column_metadata);
}
public String getComment()
{
return comment;
}
public double getRowCacheSize()
{
return rowCacheSize;
}
public double getKeyCacheSize()
{
return keyCacheSize;
}
public double getReadRepairChance()
{
return readRepairChance;
}
public int getGcGraceSeconds()
{
return gcGraceSeconds;
}
public AbstractType getDefaultValidator()
{
return defaultValidator;
}
public Integer getMinCompactionThreshold()
{
return minCompactionThreshold;
}
public Integer getMaxCompactionThreshold()
{
return maxCompactionThreshold;
}
public int getRowCacheSavePeriodInSeconds()
{
return rowCacheSavePeriodInSeconds;
}
public int getKeyCacheSavePeriodInSeconds()
{
return keyCacheSavePeriodInSeconds;
}
public int getMemtableFlushAfterMins()
{
return memtableFlushAfterMins;
}
public int getMemtableThroughputInMb()
{
return memtableThroughputInMb;
}
public double getMemtableOperationsInMillions()
{
return memtableOperationsInMillions;
}
public Map<ByteBuffer, ColumnDefinition> getColumn_metadata()
{
return Collections.unmodifiableMap(column_metadata);
}
public boolean equals(Object obj)
{
if (obj == this)
@ -512,55 +583,27 @@ public final class CFMetaData
return validator;
}
public CFMetaData apply(org.apache.cassandra.avro.CfDef cf_def) throws ConfigurationException
/** applies implicit defaults to cf definition. useful in updates */
public static void applyImplicitDefaults(org.apache.cassandra.thrift.CfDef cf_def)
{
// validate.
if (!cf_def.id.equals(cfId))
throw new ConfigurationException(String.format("ids do not match. %d, %d", cf_def.id, cfId));
if (!cf_def.keyspace.toString().equals(tableName))
throw new ConfigurationException(String.format("keyspaces do not match. %s, %s", cf_def.keyspace, tableName));
if (!cf_def.name.toString().equals(cfName))
throw new ConfigurationException("names do not match.");
if (!cf_def.column_type.toString().equals(cfType.name()))
throw new ConfigurationException("types do not match.");
if (comparator != DatabaseDescriptor.getComparator(cf_def.comparator_type.toString()))
throw new ConfigurationException("comparators do not match.");
if (cf_def.subcomparator_type == null || cf_def.subcomparator_type.equals(""))
{
if (subcolumnComparator != null)
throw new ConfigurationException("subcolumncomparators do not match.");
// else, it's null and we're good.
}
else if (subcolumnComparator != DatabaseDescriptor.getComparator(cf_def.subcomparator_type.toString()))
throw new ConfigurationException("subcolumncomparators do not match.");
validateMinMaxCompactionThresholds(cf_def);
validateMemtableSettings(cf_def);
return new CFMetaData(tableName,
cfName,
cfType,
comparator,
subcolumnComparator,
cf_def.comment == null ? "" : cf_def.comment.toString(),
cf_def.row_cache_size,
cf_def.key_cache_size,
cf_def.read_repair_chance,
cf_def.gc_grace_seconds,
DatabaseDescriptor.getComparator(cf_def.default_validation_class == null ? null : cf_def.default_validation_class.toString()),
cf_def.min_compaction_threshold,
cf_def.max_compaction_threshold,
cf_def.row_cache_save_period_in_seconds,
cf_def.key_cache_save_period_in_seconds,
cf_def.memtable_flush_after_mins,
cf_def.memtable_throughput_in_mb,
cf_def.memtable_operations_in_millions,
cfId,
column_metadata);
if (!cf_def.isSetMin_compaction_threshold())
cf_def.setMin_compaction_threshold(CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD);
if (!cf_def.isSetMax_compaction_threshold())
cf_def.setMax_compaction_threshold(CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD);
if (!cf_def.isSetRow_cache_save_period_in_seconds())
cf_def.setRow_cache_save_period_in_seconds(CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS);
if (!cf_def.isSetKey_cache_save_period_in_seconds())
cf_def.setKey_cache_save_period_in_seconds(CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS);
if (!cf_def.isSetMemtable_flush_after_mins())
cf_def.setMemtable_flush_after_mins(CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS);
if (!cf_def.isSetMemtable_throughput_in_mb())
cf_def.setMemtable_throughput_in_mb(CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB);
if (!cf_def.isSetMemtable_operations_in_millions())
cf_def.setMemtable_operations_in_millions(CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS);
}
// merges some final fields from this CFM with modifiable fields from CfDef into a new CFMetaData.
public CFMetaData apply(org.apache.cassandra.thrift.CfDef cf_def) throws ConfigurationException
public void apply(org.apache.cassandra.avro.CfDef cf_def) throws ConfigurationException
{
// validate
if (cf_def.id != cfId)
@ -585,40 +628,52 @@ public final class CFMetaData
validateMinMaxCompactionThresholds(cf_def);
validateMemtableSettings(cf_def);
Map<ByteBuffer, ColumnDefinition> metadata = new HashMap<ByteBuffer, ColumnDefinition>();
if (cf_def.column_metadata == null)
comment = cf_def.comment == null ? "" : cf_def.comment.toString();
rowCacheSize = cf_def.row_cache_size;
keyCacheSize = cf_def.key_cache_size;
readRepairChance = cf_def.read_repair_chance;
gcGraceSeconds = cf_def.gc_grace_seconds;
defaultValidator = DatabaseDescriptor.getComparator(cf_def.default_validation_class);
minCompactionThreshold = cf_def.min_compaction_threshold;
maxCompactionThreshold = cf_def.max_compaction_threshold;
rowCacheSavePeriodInSeconds = cf_def.row_cache_save_period_in_seconds;
keyCacheSavePeriodInSeconds = cf_def.key_cache_save_period_in_seconds;
memtableFlushAfterMins = cf_def.memtable_flush_after_mins;
memtableThroughputInMb = cf_def.memtable_throughput_in_mb;
memtableOperationsInMillions = cf_def.memtable_operations_in_millions;
// adjust secondary indexes. figure out who is coming and going.
Set<ByteBuffer> toRemove = new HashSet<ByteBuffer>();
Set<ByteBuffer> newIndexNames = new HashSet<ByteBuffer>();
Set<org.apache.cassandra.avro.ColumnDef> toAdd = new HashSet<org.apache.cassandra.avro.ColumnDef>();
for (org.apache.cassandra.avro.ColumnDef def : cf_def.column_metadata)
{
metadata = column_metadata;
newIndexNames.add(def.name);
if (!column_metadata.containsKey(def.name))
toAdd.add(def);
}
else
for (ByteBuffer indexName : column_metadata.keySet())
if (!newIndexNames.contains(indexName))
toRemove.add(indexName);
// remove the ones leaving.
for (ByteBuffer indexName : toRemove)
column_metadata.remove(indexName);
// update the ones staying
for (org.apache.cassandra.avro.ColumnDef def : cf_def.column_metadata)
{
for (org.apache.cassandra.thrift.ColumnDef def : cf_def.column_metadata)
{
ColumnDefinition cd = new ColumnDefinition(def.name, def.validation_class, def.index_type, def.index_name);
metadata.put(cd.name, cd);
}
column_metadata.get(def.name).setIndexType(def.index_type == null ? null : org.apache.cassandra.thrift.IndexType.valueOf(def.index_type.name()));
column_metadata.get(def.name).setIndexName(def.index_name == null ? null : def.index_name.toString());
}
// add the new ones coming in.
for (org.apache.cassandra.avro.ColumnDef def : toAdd)
{
ColumnDefinition cd = new ColumnDefinition(def.name,
def.validation_class.toString(),
def.index_type == null ? null : org.apache.cassandra.thrift.IndexType.valueOf(def.index_type.toString()),
def.index_name == null ? null : def.index_name.toString());
column_metadata.put(cd.name, cd);
}
return new CFMetaData(tableName,
cfName,
cfType,
comparator,
subcolumnComparator,
cf_def.comment,
cf_def.row_cache_size,
cf_def.key_cache_size,
cf_def.read_repair_chance,
cf_def.gc_grace_seconds,
DatabaseDescriptor.getComparator(cf_def.default_validation_class == null ? null : cf_def.default_validation_class),
cf_def.min_compaction_threshold,
cf_def.max_compaction_threshold,
cf_def.row_cache_save_period_in_seconds,
cf_def.key_cache_save_period_in_seconds,
cf_def.memtable_flush_after_mins,
cf_def.memtable_throughput_in_mb,
cf_def.memtable_operations_in_millions,
cfId,
metadata);
}
// converts CFM to thrift CfDef
@ -650,8 +705,8 @@ public final class CFMetaData
for (ColumnDefinition cd : cfm.column_metadata.values())
{
org.apache.cassandra.thrift.ColumnDef tcd = new org.apache.cassandra.thrift.ColumnDef();
tcd.setIndex_name(cd.index_name);
tcd.setIndex_type(cd.index_type);
tcd.setIndex_name(cd.getIndexName());
tcd.setIndex_type(cd.getIndexType());
tcd.setName(cd.name);
tcd.setValidation_class(cd.validator.getClass().getName());
column_meta.add(tcd);
@ -691,8 +746,8 @@ public final class CFMetaData
for (ColumnDefinition cd : cfm.column_metadata.values())
{
org.apache.cassandra.avro.ColumnDef tcd = new org.apache.cassandra.avro.ColumnDef();
tcd.index_name = cd.index_name;
tcd.index_type = org.apache.cassandra.avro.IndexType.valueOf(cd.index_type.name());
tcd.index_name = cd.getIndexName();
tcd.index_type = org.apache.cassandra.avro.IndexType.valueOf(cd.getIndexType().name());
tcd.name = cd.name;
tcd.validation_class = cd.validator.getClass().getName();
column_meta.add(tcd);
@ -700,6 +755,43 @@ public final class CFMetaData
def.column_metadata = column_meta;
return def;
}
public static org.apache.cassandra.avro.CfDef convertToAvro(org.apache.cassandra.thrift.CfDef def)
{
org.apache.cassandra.avro.CfDef newDef = new org.apache.cassandra.avro.CfDef();
newDef.keyspace = def.getKeyspace();
newDef.name = def.getName();
newDef.column_type = def.getColumn_type();
newDef.comment = def.getComment();
newDef.comparator_type = def.getComparator_type();
newDef.default_validation_class = def.getDefault_validation_class();
newDef.gc_grace_seconds = def.getGc_grace_seconds();
newDef.id = def.getId();
newDef.key_cache_save_period_in_seconds = def.getKey_cache_save_period_in_seconds();
newDef.key_cache_size = def.getKey_cache_size();
newDef.max_compaction_threshold = def.getMax_compaction_threshold();
newDef.memtable_flush_after_mins = def.getMemtable_flush_after_mins();
newDef.memtable_operations_in_millions = def.getMemtable_operations_in_millions();
newDef.memtable_throughput_in_mb = def.getMemtable_throughput_in_mb();
newDef.min_compaction_threshold = def.getMin_compaction_threshold();
newDef.read_repair_chance = def.getRead_repair_chance();
newDef.row_cache_save_period_in_seconds = def.getRow_cache_save_period_in_seconds();
newDef.row_cache_size = def.getRow_cache_size();
newDef.subcomparator_type = def.getSubcomparator_type();
List<org.apache.cassandra.avro.ColumnDef> columnMeta = new ArrayList<org.apache.cassandra.avro.ColumnDef>();
for (org.apache.cassandra.thrift.ColumnDef cdef : def.getColumn_metadata())
{
org.apache.cassandra.avro.ColumnDef tdef = new org.apache.cassandra.avro.ColumnDef();
tdef.name = cdef.BufferForName();
tdef.validation_class = cdef.getValidation_class();
tdef.index_name = cdef.getIndex_name();
tdef.index_type = cdef.getIndex_type() == null ? null : org.apache.cassandra.avro.IndexType.valueOf(cdef.getIndex_type().name());
columnMeta.add(tdef);
}
newDef.column_metadata = columnMeta;
return newDef;
}
public static void validateMinMaxCompactionThresholds(org.apache.cassandra.thrift.CfDef cf_def) throws ConfigurationException
{

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

@ -36,8 +36,8 @@ import org.apache.cassandra.utils.FBUtilities;
public class ColumnDefinition {
public final ByteBuffer name;
public final AbstractType validator;
public final IndexType index_type;
public final String index_name;
private IndexType index_type;
private String index_name;
public ColumnDefinition(ByteBuffer name, String validation_class, IndexType index_type, String index_name) throws ConfigurationException
{
@ -152,4 +152,25 @@ public class ColumnDefinition {
", index_name='" + index_name + '\'' +
'}';
}
public String getIndexName()
{
return index_name;
}
public void setIndexName(String s)
{
index_name = s;
}
public IndexType getIndexType()
{
return index_type;
}
public void setIndexType(IndexType index_type)
{
this.index_type = index_type;
}
}

11
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -644,12 +644,12 @@ public class DatabaseDescriptor
return conf.thrift_framed_transport_size_in_mb * 1024 * 1024;
}
public static AbstractType getComparator(String compareWith) throws ConfigurationException
public static AbstractType getComparator(CharSequence compareWith) throws ConfigurationException
{
if (compareWith == null)
compareWith = "BytesType";
return FBUtilities.getComparator(compareWith);
return FBUtilities.getComparator(compareWith.toString());
}
/**
@ -944,7 +944,7 @@ public class DatabaseDescriptor
public static int getKeysCachedFor(String tableName, String columnFamilyName, long expectedKeys)
{
CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
double v = (cfm == null) ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.keyCacheSize;
double v = (cfm == null) ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.getKeyCacheSize();
return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedKeys), Integer.MAX_VALUE);
}
@ -954,7 +954,7 @@ public class DatabaseDescriptor
public static int getRowsCachedFor(String tableName, String columnFamilyName, long expectedRows)
{
CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
double v = (cfm == null) ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.rowCacheSize;
double v = (cfm == null) ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.getRowCacheSize();
return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedRows), Integer.MAX_VALUE);
}
@ -967,7 +967,8 @@ public class DatabaseDescriptor
// process of mutating an individual keyspace, rather than setting manually here.
public static void setTableDefinition(KSMetaData ksm, UUID newVersion)
{
tables.put(ksm.name, ksm);
if (ksm != null)
tables.put(ksm.name, ksm);
DatabaseDescriptor.defsVersion = newVersion;
}

172
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -103,6 +103,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public final String columnFamily;
public final IPartitioner partitioner;
private final String mbeanName;
private boolean invalid = false;
private volatile int memtableSwitchCount = 0;
@ -112,7 +113,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
/* active memtable associated with this ColumnFamilyStore. */
private Memtable memtable;
private final SortedMap<ByteBuffer, ColumnFamilyStore> indexedColumns;
private final ConcurrentSkipListMap<ByteBuffer, ColumnFamilyStore> indexedColumns;
// TODO binarymemtable ops are not threadsafe (do they need to be?)
private AtomicReference<BinaryMemtable> binaryMemtable;
@ -130,11 +131,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public final CFMetaData metadata;
/* These are locally held copies to be changed from the config during runtime */
private int minCompactionThreshold;
private int maxCompactionThreshold;
private int memtime;
private int memsize;
private double memops;
private volatile DefaultInteger minCompactionThreshold;
private volatile DefaultInteger maxCompactionThreshold;
private volatile DefaultInteger memtime;
private volatile DefaultInteger memsize;
private volatile DefaultDouble memops;
private final Runnable rowCacheSaverTask = new WrappedRunnable()
{
@ -151,6 +152,46 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
ssTables.saveKeyCache();
}
};
public void reload()
{
// metadata object has been mutated directly. make all the members jibe with new settings.
// only update these runtime-modifiable settings if they have not been modified.
if (!minCompactionThreshold.isModified())
minCompactionThreshold = new DefaultInteger(metadata.getMinCompactionThreshold());
if (!maxCompactionThreshold.isModified())
maxCompactionThreshold = new DefaultInteger(metadata.getMaxCompactionThreshold());
if (!memtime.isModified())
memtime = new DefaultInteger(metadata.getMemtableFlushAfterMins());
if (!memsize.isModified())
memsize = new DefaultInteger(metadata.getMemtableThroughputInMb());
if (!memops.isModified())
memops = new DefaultDouble(metadata.getMemtableOperationsInMillions());
ssTables.updateCacheSizes();
// figure out what needs to be added and dropped.
// future: if/when we have modifiable settings for secondary indexes, they'll need to be handled here.
for (ByteBuffer indexName : indexedColumns.keySet())
{
if (!metadata.getColumn_metadata().containsKey(indexName))
{
ColumnFamilyStore indexCfs = indexedColumns.remove(indexName);
if (indexCfs == null)
{
logger.debug("index {} already removed; ignoring", FBUtilities.bytesToHex(indexName));
continue;
}
SystemTable.setIndexRemoved(metadata.tableName, metadata.cfName);
indexCfs.removeAllSSTables();
}
}
for (ColumnDefinition cdef : metadata.getColumn_metadata().values())
if (!indexedColumns.containsKey(cdef.name) && cdef.getIndexType() != null)
addIndex(cdef);
}
private ColumnFamilyStore(Table table, String columnFamilyName, IPartitioner partitioner, int generation, CFMetaData metadata)
{
@ -158,11 +199,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
this.table = table;
columnFamily = columnFamilyName;
this.metadata = metadata;
this.minCompactionThreshold = metadata.minCompactionThreshold;
this.maxCompactionThreshold = metadata.maxCompactionThreshold;
this.memtime = metadata.memtableFlushAfterMins;
this.memsize = metadata.memtableThroughputInMb;
this.memops = metadata.memtableOperationsInMillions;
this.minCompactionThreshold = new DefaultInteger(metadata.getMinCompactionThreshold());
this.maxCompactionThreshold = new DefaultInteger(metadata.getMaxCompactionThreshold());
this.memtime = new DefaultInteger(metadata.getMemtableFlushAfterMins());
this.memsize = new DefaultInteger(metadata.getMemtableThroughputInMb());
this.memops = new DefaultDouble(metadata.getMemtableOperationsInMillions());
this.partitioner = partitioner;
fileIndexGenerator.set(generation);
memtable = new Memtable(this);
@ -199,9 +240,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
// create the private ColumnFamilyStores for the secondary column indexes
indexedColumns = new ConcurrentSkipListMap<ByteBuffer, ColumnFamilyStore>(getComparator());
for (ColumnDefinition info : metadata.column_metadata.values())
for (ColumnDefinition info : metadata.getColumn_metadata().values())
{
if (info.index_type != null)
if (info.getIndexType() != null)
addIndex(info);
}
@ -254,7 +295,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void addIndex(final ColumnDefinition info)
{
assert info.index_type != null;
assert info.getIndexType() != null;
// create the index CFS
IPartitioner rowPartitioner = StorageService.getPartitioner();
AbstractType columnComparator = (rowPartitioner instanceof OrderPreservingPartitioner || rowPartitioner instanceof ByteOrderedPartitioner)
? BytesType.instance
@ -262,30 +305,44 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
final CFMetaData indexedCfMetadata = CFMetaData.newIndexMetadata(table.name, columnFamily, info, columnComparator);
ColumnFamilyStore indexedCfs = ColumnFamilyStore.createColumnFamilyStore(table,
indexedCfMetadata.cfName,
new LocalPartitioner(metadata.column_metadata.get(info.name).validator),
new LocalPartitioner(metadata.getColumn_metadata().get(info.name).validator),
indexedCfMetadata);
// record that the column is supposed to be indexed, before we start building it
// (so we don't omit indexing writes that happen during build process)
indexedColumns.put(info.name, indexedCfs);
if (!SystemTable.isIndexBuilt(table.name, indexedCfMetadata.cfName))
// link in indexedColumns. this means that writes will add new data to the index immediately,
// so we don't have to lock everything while we do the build. it's up to the operator to wait
// until the index is actually built before using in queries.
if (indexedColumns.putIfAbsent(info.name, indexedCfs) != null)
return;
// if we're just linking in the index to indexedColumns on an already-built index post-restart, we're done
if (SystemTable.isIndexBuilt(table.name, indexedCfMetadata.cfName))
return;
// build it asynchronously; addIndex gets called by CFS open and schema update, neither of which
// we want to block for a long period. (actual build is serialized on CompactionManager.)
Runnable runnable = new Runnable()
{
logger.info("Creating index {}.{}", table, indexedCfMetadata.cfName);
try
public void run()
{
forceBlockingFlush();
logger.info("Creating index {}.{}", table, indexedCfMetadata.cfName);
try
{
forceBlockingFlush();
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
buildSecondaryIndexes(getSSTables(), FBUtilities.singleton(info.name));
logger.info("Index {} complete", indexedCfMetadata.cfName);
SystemTable.setIndexBuilt(table.name, indexedCfMetadata.cfName);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
buildSecondaryIndexes(getSSTables(), FBUtilities.singleton(info.name));
logger.info("Index {} complete", indexedCfMetadata.cfName);
SystemTable.setIndexBuilt(table.name, indexedCfMetadata.cfName);
}
};
new Thread(runnable, "Create index " + indexedCfMetadata.cfName).start();
}
public void buildSecondaryIndexes(Collection<SSTableReader> sstables, SortedSet<ByteBuffer> columns)
@ -309,12 +366,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
// called when dropping or renaming a CF. Performs mbean housekeeping.
// called when dropping or renaming a CF. Performs mbean housekeeping and invalidates CFS to other operations.
void unregisterMBean()
{
try
{
invalid = true;
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
ObjectName nameObj = new ObjectName(mbeanName);
if (mbs.isRegistered(nameObj))
@ -446,8 +503,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void initRowCache()
{
String msgSuffix = String.format(" row cache for %s of %s", columnFamily, table.name);
int rowCacheSavePeriodInSeconds = DatabaseDescriptor.getTableMetaData(table.name).get(columnFamily).rowCacheSavePeriodInSeconds;
int keyCacheSavePeriodInSeconds = DatabaseDescriptor.getTableMetaData(table.name).get(columnFamily).keyCacheSavePeriodInSeconds;
int rowCacheSavePeriodInSeconds = DatabaseDescriptor.getTableMetaData(table.name).get(columnFamily).getRowCacheSavePeriodInSeconds();
int keyCacheSavePeriodInSeconds = DatabaseDescriptor.getTableMetaData(table.name).get(columnFamily).getKeyCacheSavePeriodInSeconds();
long start = System.currentTimeMillis();
logger.info(String.format("loading%s", msgSuffix));
@ -535,7 +592,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public String getFlushPath()
{
long guessedSize = 2 * memsize * 1024*1024; // 2* adds room for keys, column indexes
long guessedSize = 2 * memsize.value() * 1024*1024; // 2* adds room for keys, column indexes
String location = DatabaseDescriptor.getDataFileLocationForTable(table.name, guessedSize);
if (location == null)
throw new RuntimeException("Insufficient disk space to flush");
@ -855,6 +912,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
ssTables.replace(sstables, replacements);
}
public boolean isInvalid()
{
return invalid;
}
public void removeAllSSTables()
{
@ -1006,7 +1068,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public int gcBefore()
{
return (int) (System.currentTimeMillis() / 1000) - metadata.gcGraceSeconds;
return (int) (System.currentTimeMillis() / 1000) - metadata.getGcGraceSeconds();
}
private ColumnFamily cacheRow(DecoratedKey key)
@ -1212,7 +1274,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
Iterables.addAll(sstables, ssTables);
RowIterator iterator = RowIteratorFactory.getIterator(memtables, sstables, startWith, stopAt, filter, getComparator(), this);
int gcBefore = (int)(System.currentTimeMillis() / 1000) - metadata.gcGraceSeconds;
int gcBefore = (int)(System.currentTimeMillis() / 1000) - metadata.getGcGraceSeconds();
try
{
@ -1765,70 +1827,70 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public int getMinimumCompactionThreshold()
{
return minCompactionThreshold;
return minCompactionThreshold.value();
}
public void setMinimumCompactionThreshold(int minCompactionThreshold)
{
if ((minCompactionThreshold > this.maxCompactionThreshold) && this.maxCompactionThreshold != 0) {
if ((minCompactionThreshold > this.maxCompactionThreshold.value()) && this.maxCompactionThreshold.value() != 0) {
throw new RuntimeException("The min_compaction_threshold cannot be larger than the max.");
}
this.minCompactionThreshold = minCompactionThreshold;
this.minCompactionThreshold.set(minCompactionThreshold);
}
public int getMaximumCompactionThreshold()
{
return maxCompactionThreshold;
return maxCompactionThreshold.value();
}
public void setMaximumCompactionThreshold(int maxCompactionThreshold)
{
if (maxCompactionThreshold < this.minCompactionThreshold) {
if (maxCompactionThreshold < this.minCompactionThreshold.value()) {
throw new RuntimeException("The max_compaction_threshold cannot be smaller than the min.");
}
this.maxCompactionThreshold = maxCompactionThreshold;
this.maxCompactionThreshold.set(maxCompactionThreshold);
}
public void disableAutoCompaction()
{
this.minCompactionThreshold = 0;
this.maxCompactionThreshold = 0;
minCompactionThreshold.set(0);
maxCompactionThreshold.set(0);
}
public int getMemtableFlushAfterMins()
{
return memtime;
return memtime.value();
}
public void setMemtableFlushAfterMins(int time)
{
if (time <= 0) {
throw new RuntimeException("MemtableFlushAfterMins must be greater than 0.");
}
this.memtime = time;
this.memtime.set(time);
}
public int getMemtableThroughputInMB()
{
return memsize;
return memsize.value();
}
public void setMemtableThroughputInMB(int size)
{
if (size <= 0) {
throw new RuntimeException("MemtableThroughputInMB must be greater than 0.");
}
this.memsize = size;
this.memsize.set(size);
}
public double getMemtableOperationsInMillions()
{
return memops;
return memops.value();
}
public void setMemtableOperationsInMillions(double ops)
{
if (ops <= 0) {
throw new RuntimeException("MemtableOperationsInMillions must be greater than 0.0.");
}
this.memops = ops;
this.memops.set(ops);
}
public long estimateKeys()

226
src/java/org/apache/cassandra/db/CompactionManager.java
View File

@ -28,6 +28,10 @@ import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Lock;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@ -57,6 +61,8 @@ public class CompactionManager implements CompactionManagerMBean
public static final String MBEAN_OBJECT_NAME = "org.apache.cassandra.db:type=CompactionManager";
private static final Logger logger = LoggerFactory.getLogger(CompactionManager.class);
public static final CompactionManager instance;
private final ReentrantLock compactionLock = new ReentrantLock();
// todo: should provide a way to unlock in mbean?
static
{
@ -74,6 +80,11 @@ public class CompactionManager implements CompactionManagerMBean
private CompactionExecutor executor = new CompactionExecutor();
private Map<ColumnFamilyStore, Integer> estimatedCompactions = new NonBlockingHashMap<ColumnFamilyStore, Integer>();
public Lock getCompactionLock()
{
return compactionLock;
}
/**
* Call this whenever a compaction might be needed on the given columnfamily.
@ -86,27 +97,37 @@ public class CompactionManager implements CompactionManagerMBean
{
public Integer call() throws IOException
{
Integer minThreshold = cfs.getMinimumCompactionThreshold();
Integer maxThreshold = cfs.getMaximumCompactionThreshold();
if (minThreshold == 0 || maxThreshold == 0)
compactionLock.lock();
try
{
logger.debug("Compaction is currently disabled.");
return 0;
}
logger.debug("Checking to see if compaction of " + cfs.columnFamily + " would be useful");
Set<List<SSTableReader>> buckets = getBuckets(convertSSTablesToPairs(cfs.getSSTables()), 50L * 1024L * 1024L);
updateEstimateFor(cfs, buckets);
for (List<SSTableReader> sstables : buckets)
{
if (sstables.size() >= minThreshold)
if (cfs.isInvalid())
return 0;
Integer minThreshold = cfs.getMinimumCompactionThreshold();
Integer maxThreshold = cfs.getMaximumCompactionThreshold();
if (minThreshold == 0 || maxThreshold == 0)
{
// if we have too many to compact all at once, compact older ones first -- this avoids
// re-compacting files we just created.
Collections.sort(sstables);
return doCompaction(cfs, sstables.subList(0, Math.min(sstables.size(), maxThreshold)), (int) (System.currentTimeMillis() / 1000) - cfs.metadata.gcGraceSeconds);
logger.debug("Compaction is currently disabled.");
return 0;
}
logger.debug("Checking to see if compaction of " + cfs.columnFamily + " would be useful");
Set<List<SSTableReader>> buckets = getBuckets(convertSSTablesToPairs(cfs.getSSTables()), 50L * 1024L * 1024L);
updateEstimateFor(cfs, buckets);
for (List<SSTableReader> sstables : buckets)
{
if (sstables.size() >= minThreshold)
{
// if we have too many to compact all at once, compact older ones first -- this avoids
// re-compacting files we just created.
Collections.sort(sstables);
return doCompaction(cfs, sstables.subList(0, Math.min(sstables.size(), maxThreshold)), (int) (System.currentTimeMillis() / 1000) - cfs.metadata.getGcGraceSeconds());
}
}
}
finally
{
compactionLock.unlock();
}
return 0;
}
@ -143,8 +164,17 @@ public class CompactionManager implements CompactionManagerMBean
{
public Object call() throws IOException
{
doCleanupCompaction(cfStore);
return this;
compactionLock.lock();
try
{
if (!cfStore.isInvalid())
doCleanupCompaction(cfStore);
return this;
}
finally
{
compactionLock.unlock();
}
}
};
executor.submit(runnable).get();
@ -152,7 +182,7 @@ public class CompactionManager implements CompactionManagerMBean
public void performMajor(final ColumnFamilyStore cfStore) throws InterruptedException, ExecutionException
{
submitMajor(cfStore, 0, (int) (System.currentTimeMillis() / 1000) - cfStore.metadata.gcGraceSeconds).get();
submitMajor(cfStore, 0, (int) (System.currentTimeMillis() / 1000) - cfStore.metadata.getGcGraceSeconds()).get();
}
public Future<Object> submitMajor(final ColumnFamilyStore cfStore, final long skip, final int gcBefore)
@ -161,25 +191,35 @@ public class CompactionManager implements CompactionManagerMBean
{
public Object call() throws IOException
{
Collection<SSTableReader> sstables;
if (skip > 0)
compactionLock.lock();
try
{
sstables = new ArrayList<SSTableReader>();
for (SSTableReader sstable : cfStore.getSSTables())
if (cfStore.isInvalid())
return this;
Collection<SSTableReader> sstables;
if (skip > 0)
{
if (sstable.length() < skip * 1024L * 1024L * 1024L)
sstables = new ArrayList<SSTableReader>();
for (SSTableReader sstable : cfStore.getSSTables())
{
sstables.add(sstable);
if (sstable.length() < skip * 1024L * 1024L * 1024L)
{
sstables.add(sstable);
}
}
}
else
{
sstables = cfStore.getSSTables();
}
doCompaction(cfStore, sstables, gcBefore);
return this;
}
else
finally
{
sstables = cfStore.getSSTables();
compactionLock.unlock();
}
doCompaction(cfStore, sstables, gcBefore);
return this;
}
};
return executor.submit(callable);
@ -191,8 +231,17 @@ public class CompactionManager implements CompactionManagerMBean
{
public Object call() throws IOException
{
doValidationCompaction(cfStore, validator);
return this;
compactionLock.lock();
try
{
if (!cfStore.isInvalid())
doValidationCompaction(cfStore, validator);
return this;
}
finally
{
compactionLock.unlock();
}
}
};
return executor.submit(callable);
@ -338,7 +387,7 @@ public class CompactionManager implements CompactionManagerMBean
logger.debug("Expected bloom filter size : " + expectedBloomFilterSize);
SSTableWriter writer = null;
CompactionIterator ci = new AntiCompactionIterator(cfs, sstables, ranges, (int) (System.currentTimeMillis() / 1000) - cfs.metadata.gcGraceSeconds, cfs.isCompleteSSTables(sstables));
CompactionIterator ci = new AntiCompactionIterator(cfs, sstables, ranges, (int) (System.currentTimeMillis() / 1000) - cfs.metadata.getGcGraceSeconds(), cfs.isCompleteSSTables(sstables));
Iterator<AbstractCompactedRow> nni = new FilterIterator(ci, PredicateUtils.notNullPredicate());
executor.beginCompaction(cfs, ci);
@ -488,30 +537,6 @@ public class CompactionManager implements CompactionManagerMBean
}
return tablePairs;
}
public Future submitDrop(final ColumnFamilyStore... stores)
{
Callable callable = new Callable()
{
public Object call() throws IOException
{
for (ColumnFamilyStore cfs : stores)
{
Table.flusherLock.writeLock().lock();
try
{
cfs.table.dropCf(cfs.metadata.cfId);
}
finally
{
Table.flusherLock.writeLock().unlock();
}
}
return null;
}
};
return executor.submit(callable);
}
public Future submitIndexBuild(final ColumnFamilyStore cfs, final Table.IndexBuilder builder)
{
@ -519,22 +544,49 @@ public class CompactionManager implements CompactionManagerMBean
{
public void run()
{
executor.beginCompaction(cfs, builder);
builder.build();
compactionLock.lock();
try
{
if (cfs.isInvalid())
return;
executor.beginCompaction(cfs, builder);
builder.build();
}
finally
{
compactionLock.unlock();
}
}
};
return executor.submit(runnable);
// don't submit to the executor if the compaction lock is held by the current thread. Instead return a simple
// future that will be immediately immediately get()ed and executed. Happens during a migration, which locks
// the compaction thread and then reinitializes a ColumnFamilyStore. Under normal circumstances, CFS spawns
// index jobs to the compaction manager (this) and blocks on them.
if (compactionLock.isHeldByCurrentThread())
return new SimpleFuture(runnable);
else
return executor.submit(runnable);
}
public Future<SSTableReader> submitSSTableBuild(Descriptor desc)
{
// invalid descriptions due to missing or dropped CFS are handled by SSTW and StreamInSession.
final SSTableWriter.Builder builder = SSTableWriter.createBuilder(desc);
Callable<SSTableReader> callable = new Callable<SSTableReader>()
{
public SSTableReader call() throws IOException
{
executor.beginCompaction(builder.cfs, builder);
return builder.build();
compactionLock.lock();
try
{
executor.beginCompaction(builder.cfs, builder);
return builder.build();
}
finally
{
compactionLock.unlock();
}
}
};
return executor.submit(callable);
@ -544,7 +596,7 @@ public class CompactionManager implements CompactionManagerMBean
{
public ValidationCompactionIterator(ColumnFamilyStore cfs) throws IOException
{
super(cfs, cfs.getSSTables(), (int) (System.currentTimeMillis() / 1000) - cfs.metadata.gcGraceSeconds, true);
super(cfs, cfs.getSSTables(), (int) (System.currentTimeMillis() / 1000) - cfs.metadata.getGcGraceSeconds(), true);
}
@Override
@ -705,4 +757,46 @@ public class CompactionManager implements CompactionManagerMBean
{
return executor.getCompletedTaskCount();
}
private class SimpleFuture implements Future
{
private Runnable runnable;
private SimpleFuture(Runnable r)
{
runnable = r;
}
@Override
public boolean cancel(boolean mayInterruptIfRunning)
{
throw new IllegalStateException("May not call SimpleFuture.cancel()");
}
@Override
public boolean isCancelled()
{
return false;
}
@Override
public boolean isDone()
{
return runnable == null;
}
@Override
public Object get() throws InterruptedException, ExecutionException
{
runnable.run();
runnable = null;
return runnable;
}
@Override
public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
throw new IllegalStateException("May not call SimpleFuture.get(long, TimeUnit)");
}
}
}

2
src/java/org/apache/cassandra/db/RowIteratorFactory.java
View File

@ -107,7 +107,7 @@ public class RowIteratorFactory
// reduce rows from all sources into a single row
ReducingIterator<IColumnIterator, Row> reduced = new ReducingIterator<IColumnIterator, Row>(collated)
{
private final int gcBefore = (int) (System.currentTimeMillis() / 1000) - cfs.metadata.gcGraceSeconds;
private final int gcBefore = (int) (System.currentTimeMillis() / 1000) - cfs.metadata.getGcGraceSeconds();
private final List<IColumnIterator> colIters = new ArrayList<IColumnIterator>();
private DecoratedKey key;

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

@ -41,7 +41,6 @@ import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.Mutation;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
public class RowMutation
@ -105,7 +104,7 @@ public class RowMutation
{
ByteBuffer combined = HintedHandOffManager.makeCombinedName(rm.getTable(), cf.metadata().cfName);
QueryPath path = new QueryPath(HintedHandOffManager.HINTS_CF, rm.key(), combined);
add(path, FBUtilities.EMPTY_BYTE_BUFFER, System.currentTimeMillis(), cf.metadata().gcGraceSeconds);
add(path, FBUtilities.EMPTY_BYTE_BUFFER, System.currentTimeMillis(), cf.metadata().getGcGraceSeconds());
}
}

6
src/java/org/apache/cassandra/db/Table.java
View File

@ -34,6 +34,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.cassandra.config.CFMetaData;
@ -121,6 +122,11 @@ public class Table
}
return tableInstance;
}
public static Lock getFlushLock()
{
return flusherLock.writeLock();
}
public static Table clear(String table) throws IOException
{

2
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -464,7 +464,7 @@ public class CommitLog
}
header.turnOff(id);
if (header.isSafeToDelete())
if (header.isSafeToDelete() && iter.hasNext())
{
logger.info("Discarding obsolete commit log:" + segment);
segment.close();

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

@ -5,18 +5,15 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import java.io.IOError;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
/**
* Licensed to the Apache Software Foundation (ASF) under one
@ -41,17 +38,15 @@ public class DropColumnFamily extends Migration
{
private String tableName;
private String cfName;
private boolean blockOnFileDeletion;
/** Required no-arg constructor */
protected DropColumnFamily() { /* pass */ }
public DropColumnFamily(String tableName, String cfName, boolean blockOnFileDeletion) throws ConfigurationException, IOException
public DropColumnFamily(String tableName, String cfName) throws ConfigurationException, IOException
{
super(UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress()), DatabaseDescriptor.getDefsVersion());
this.tableName = tableName;
this.cfName = cfName;
this.blockOnFileDeletion = blockOnFileDeletion;
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(tableName);
if (ksm == null)
@ -85,34 +80,25 @@ public class DropColumnFamily extends Migration
@Override
public void applyModels() throws IOException
{
// reinitialize the table.
KSMetaData existing = DatabaseDescriptor.getTableDefinition(tableName);
CFMetaData cfm = existing.cfMetaData().get(cfName);
ColumnFamilyStore cfs = Table.open(cfm.tableName).getColumnFamilyStore(cfName);
KSMetaData ksm = makeNewKeyspaceDefinition(existing);
CFMetaData.purge(cfm);
DatabaseDescriptor.setTableDefinition(ksm, newVersion);
if (!clientMode)
acquireLocks();
try
{
try
// reinitialize the table.
KSMetaData existing = DatabaseDescriptor.getTableDefinition(tableName);
CFMetaData cfm = existing.cfMetaData().get(cfName);
KSMetaData ksm = makeNewKeyspaceDefinition(existing);
CFMetaData.purge(cfm);
DatabaseDescriptor.setTableDefinition(ksm, newVersion);
if (!clientMode)
{
CompactionManager.instance.submitDrop(cfs).get();
}
catch (InterruptedException ex)
{
throw new IOException(ex);
}
catch (ExecutionException ex)
{
// if the compaction manager catches IOException, it wraps it in an IOError and rethrows, which should
// get caught be the executor and rethrown as an ExecutionException.
if (ex.getCause() instanceof IOException)
throw (IOException)ex.getCause();
else
throw new IOException(ex);
Table.open(ksm.name).dropCf(cfm.cfId);
}
}
finally
{
releaseLocks();
}
}
public void subdeflate(org.apache.cassandra.db.migration.avro.Migration mi)
@ -120,7 +106,6 @@ public class DropColumnFamily extends Migration
org.apache.cassandra.db.migration.avro.DropColumnFamily dcf = new org.apache.cassandra.db.migration.avro.DropColumnFamily();
dcf.ksname = new org.apache.avro.util.Utf8(tableName);
dcf.cfname = new org.apache.avro.util.Utf8(cfName);
dcf.block_on_deletion = blockOnFileDeletion;
mi.migration = dcf;
}
@ -129,6 +114,5 @@ public class DropColumnFamily extends Migration
org.apache.cassandra.db.migration.avro.DropColumnFamily dcf = (org.apache.cassandra.db.migration.avro.DropColumnFamily)mi.migration;
tableName = dcf.ksname.toString();
cfName = dcf.cfname.toString();
blockOnFileDeletion = dcf.block_on_deletion;
}
}

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

@ -22,8 +22,6 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.db.HintedHandOffManager;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
@ -31,23 +29,19 @@ import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import java.io.IOError;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
public class DropKeyspace extends Migration
{
private String name;
private boolean blockOnFileDeletion;
/** Required no-arg constructor */
protected DropKeyspace() { /* pass */ }
public DropKeyspace(String name, boolean blockOnFileDeletion) throws ConfigurationException, IOException
public DropKeyspace(String name) throws ConfigurationException, IOException
{
super(UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress()), DatabaseDescriptor.getDefsVersion());
this.name = name;
this.blockOnFileDeletion = blockOnFileDeletion;
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(name);
if (ksm == null)
throw new ConfigurationException("Keyspace does not exist.");
@ -64,44 +58,37 @@ public class DropKeyspace extends Migration
@Override
public void applyModels() throws IOException
{
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(name);
// remove the table from the static instances.
Table table = Table.clear(ksm.name);
if (table == null)
throw new IOException("Table is not active. " + ksm.name);
// remove all cfs from the table instance.
for (CFMetaData cfm : ksm.cfMetaData().values())
CFMetaData.purge(cfm);
if (!clientMode)
acquireLocks();
try
{
try
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(name);
// remove the table from the static instances.
Table table = Table.clear(ksm.name);
if (table == null)
throw new IOException("Table is not active. " + ksm.name);
// remove all cfs from the table instance.
for (CFMetaData cfm : ksm.cfMetaData().values())
{
CompactionManager.instance.submitDrop(table.getColumnFamilyStores().toArray(new ColumnFamilyStore[0])).get();
CFMetaData.purge(cfm);
if (!clientMode)
{
table.dropCf(cfm.cfId);
}
}
catch (InterruptedException ex)
// reset defs.
DatabaseDescriptor.clearTableDefinition(ksm, newVersion);
if (!clientMode)
{
throw new IOException(ex);
}
catch (ExecutionException ex)
{
// if the compaction manager catches IOException, it wraps it in an IOError and rethrows, which should
// get caught be the executor and rethrown as an ExecutionException.
if (ex.getCause() instanceof IOException)
throw (IOException)ex.getCause();
else
throw new IOException(ex);
// clear up any local hinted data for this keyspace.
HintedHandOffManager.renameHints(name, null);
}
}
// reset defs.
DatabaseDescriptor.clearTableDefinition(ksm, newVersion);
if (!clientMode)
finally
{
// clear up any local hinted data for this keyspace.
HintedHandOffManager.renameHints(name, null);
releaseLocks();
}
}
@ -109,7 +96,6 @@ public class DropKeyspace extends Migration
{
org.apache.cassandra.db.migration.avro.DropKeyspace dks = new org.apache.cassandra.db.migration.avro.DropKeyspace();
dks.ksname = new org.apache.avro.util.Utf8(name);
dks.block_on_deletion = blockOnFileDeletion;
mi.migration = dks;
}
@ -117,6 +103,5 @@ public class DropKeyspace extends Migration
{
org.apache.cassandra.db.migration.avro.DropKeyspace dks = (org.apache.cassandra.db.migration.avro.DropKeyspace)mi.migration;
name = dks.ksname.toString();
blockOnFileDeletion = dks.block_on_deletion;
}
}

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

@ -91,6 +91,19 @@ public abstract class Migration
this.lastVersion = lastVersion;
clientMode = StorageService.instance.isClientMode();
}
// block compactions and flushing.
protected final void acquireLocks()
{
CompactionManager.instance.getCompactionLock().lock();
Table.getFlushLock().lock();
}
protected final void releaseLocks()
{
Table.getFlushLock().unlock();
CompactionManager.instance.getCompactionLock().unlock();
}
/** override this to perform logic before writing the migration or applying it. defaults to nothing. */
public void beforeApplyModels() {}

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

@ -14,6 +14,7 @@ import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
@ -38,84 +39,58 @@ import org.apache.cassandra.utils.UUIDGen;
/** todo: doesn't work with secondary indices yet. See CASSANDRA-1415. */
public class UpdateColumnFamily extends Migration
{
private CFMetaData oldCfm;
private CFMetaData newCfm;
private CFMetaData metadata;
protected UpdateColumnFamily() { }
/** assumes validation has already happened. That is, replacing oldCfm with newCfm is neither illegal or totally whackass. */
public UpdateColumnFamily(CFMetaData oldCfm, CFMetaData newCfm) throws ConfigurationException, IOException
public UpdateColumnFamily(org.apache.cassandra.avro.CfDef cf_def) throws ConfigurationException, IOException
{
super(UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress()), DatabaseDescriptor.getDefsVersion());
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(newCfm.tableName);
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(cf_def.keyspace.toString());
if (ksm == null)
throw new ConfigurationException("Keyspace does not already exist.");
this.oldCfm = oldCfm;
this.newCfm = newCfm;
CFMetaData oldCfm = DatabaseDescriptor.getCFMetaData(CFMetaData.getId(cf_def.keyspace.toString(), cf_def.name.toString()));
oldCfm.apply(cf_def);
this.metadata = oldCfm;
// clone ksm but include the new cf def.
KSMetaData newKsm = makeNewKeyspaceDefinition(ksm);
rm = Migration.makeDefinitionMutation(newKsm, null, newVersion);
}
private KSMetaData makeNewKeyspaceDefinition(KSMetaData ksm)
{
List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
newCfs.remove(oldCfm);
newCfs.add(newCfm);
return new KSMetaData(ksm.name, ksm.strategyClass, ksm.strategyOptions, ksm.replicationFactor, newCfs.toArray(new CFMetaData[newCfs.size()]));
rm = Migration.makeDefinitionMutation(ksm, null, newVersion);
}
public void beforeApplyModels()
{
if (clientMode)
return;
ColumnFamilyStore cfs = Table.open(oldCfm.tableName).getColumnFamilyStore(oldCfm.cfName);
ColumnFamilyStore cfs = Table.open(metadata.tableName).getColumnFamilyStore(metadata.cfName);
cfs.snapshot(Table.getTimestampedSnapshotName(null));
}
void applyModels() throws IOException
{
logger.debug("Updating " + oldCfm + " to " + newCfm);
KSMetaData newKsm = makeNewKeyspaceDefinition(DatabaseDescriptor.getTableDefinition(newCfm.tableName));
DatabaseDescriptor.setTableDefinition(newKsm, newVersion);
logger.debug("Updating " + metadata + " to " + metadata);
DatabaseDescriptor.setTableDefinition(null, newVersion);
if (!clientMode)
{
Table table = Table.open(oldCfm.tableName);
ColumnFamilyStore oldCfs = table.getColumnFamilyStore(oldCfm.cfName);
table.reloadCf(newCfm.cfId);
// clean up obsolete index data files
for (Map.Entry<ByteBuffer, ColumnDefinition> entry : oldCfm.column_metadata.entrySet())
{
ByteBuffer column = entry.getKey();
ColumnDefinition def = entry.getValue();
if (def.index_type != null
&& (!newCfm.column_metadata.containsKey(column) || newCfm.column_metadata.get(column).index_type == null))
{
ColumnFamilyStore indexCfs = oldCfs.getIndexedColumnFamilyStore(column);
SystemTable.setIndexRemoved(table.name, indexCfs.columnFamily);
indexCfs.removeAllSSTables();
}
}
Table table = Table.open(metadata.tableName);
ColumnFamilyStore oldCfs = table.getColumnFamilyStore(metadata.cfName);
oldCfs.reload();
}
}
public void subdeflate(org.apache.cassandra.db.migration.avro.Migration mi)
{
org.apache.cassandra.db.migration.avro.UpdateColumnFamily update = new org.apache.cassandra.db.migration.avro.UpdateColumnFamily();
update.newCf = newCfm.deflate();
update.oldCf = oldCfm.deflate();
update.metadata = metadata.deflate();
mi.migration = update;
}
public void subinflate(org.apache.cassandra.db.migration.avro.Migration mi)
{
org.apache.cassandra.db.migration.avro.UpdateColumnFamily update = (org.apache.cassandra.db.migration.avro.UpdateColumnFamily)mi.migration;
newCfm = CFMetaData.inflate(update.newCf);
oldCfm = CFMetaData.inflate(update.oldCf);
metadata = CFMetaData.inflate(update.metadata);
}
}

2
src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
View File

@ -259,6 +259,8 @@ public class SSTableWriter extends SSTable
public SSTableReader build() throws IOException
{
if (cfs.isInvalid())
return null;
File ifile = new File(desc.filenameFor(SSTable.COMPONENT_INDEX));
File ffile = new File(desc.filenameFor(SSTable.COMPONENT_FILTER));
assert !ifile.exists();

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

@ -598,7 +598,7 @@ public class StorageProxy implements StorageProxyMBean
private static boolean randomlyReadRepair(ReadCommand command)
{
CFMetaData cfmd = DatabaseDescriptor.getTableMetaData(command.table).get(command.getColumnFamilyName());
return cfmd.readRepairChance > random.nextDouble();
return cfmd.getReadRepairChance() > random.nextDouble();
}
public long getReadOperations()

20
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1977,21 +1977,21 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
RawColumnFamily rcf = new RawColumnFamily();
rcf.name = cfm.cfName;
rcf.compare_with = cfm.comparator.getClass().getName();
rcf.default_validation_class = cfm.defaultValidator.getClass().getName();
rcf.default_validation_class = cfm.getDefaultValidator().getClass().getName();
rcf.compare_subcolumns_with = cfm.subcolumnComparator == null ? null : cfm.subcolumnComparator.getClass().getName();
rcf.column_type = cfm.cfType;
rcf.comment = cfm.comment;
rcf.keys_cached = cfm.keyCacheSize;
rcf.read_repair_chance = cfm.readRepairChance;
rcf.gc_grace_seconds = cfm.gcGraceSeconds;
rcf.rows_cached = cfm.rowCacheSize;
rcf.column_metadata = new RawColumnDefinition[cfm.column_metadata.size()];
rcf.comment = cfm.getComment();
rcf.keys_cached = cfm.getKeyCacheSize();
rcf.read_repair_chance = cfm.getReadRepairChance();
rcf.gc_grace_seconds = cfm.getGcGraceSeconds();
rcf.rows_cached = cfm.getRowCacheSize();
rcf.column_metadata = new RawColumnDefinition[cfm.getColumn_metadata().size()];
int j = 0;
for (ColumnDefinition cd : cfm.column_metadata.values())
for (ColumnDefinition cd : cfm.getColumn_metadata().values())
{
RawColumnDefinition rcd = new RawColumnDefinition();
rcd.index_name = cd.index_name;
rcd.index_type = cd.index_type;
rcd.index_name = cd.getIndexName();
rcd.index_type = cd.getIndexType();
rcd.name = ByteBufferUtil.string(cd.name, Charsets.UTF_8);
rcd.validator_class = cd.validator.getClass().getName();
rcf.column_metadata[j++] = rcd;

2
src/java/org/apache/cassandra/streaming/StreamInSession.java
View File

@ -136,6 +136,8 @@ public class StreamInSession
try
{
SSTableReader sstable = future.get();
if (sstable == null)
continue;
cfs.addSSTable(sstable);
sstables.add(sstable);
}

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

@ -42,6 +42,7 @@ import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyNotDefinedException;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.ExpiringColumn;
@ -728,7 +729,7 @@ public class CassandraServer implements Cassandra.Iface
try
{
applyMigrationOnStage(new DropColumnFamily(state().getKeyspace(), column_family, true));
applyMigrationOnStage(new DropColumnFamily(state().getKeyspace(), column_family));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
@ -794,7 +795,7 @@ public class CassandraServer implements Cassandra.Iface
try
{
applyMigrationOnStage(new DropKeyspace(keyspace, true));
applyMigrationOnStage(new DropKeyspace(keyspace));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
@ -857,8 +858,9 @@ public class CassandraServer implements Cassandra.Iface
try
{
CFMetaData newCfm = oldCfm.apply(cf_def);
UpdateColumnFamily update = new UpdateColumnFamily(oldCfm, newCfm);
// ideally, apply() would happen on the stage with the
CFMetaData.applyImplicitDefaults(cf_def);
UpdateColumnFamily update = new UpdateColumnFamily(CFMetaData.convertToAvro(cf_def));
applyMigrationOnStage(update);
return DatabaseDescriptor.getDefsVersion().toString();
}
@ -876,6 +878,7 @@ public class CassandraServer implements Cassandra.Iface
}
}
// @see CFMetaData.applyImplicitDefaults().
private CFMetaData convertToCFMetaData(CfDef cf_def) throws InvalidRequestException, ConfigurationException
{
ColumnFamilyType cfType = ColumnFamilyType.create(cf_def.column_type);
@ -884,6 +887,7 @@ public class CassandraServer implements Cassandra.Iface
throw new InvalidRequestException("Invalid column type " + cf_def.column_type);
}
CFMetaData.applyImplicitDefaults(cf_def);
CFMetaData.validateMinMaxCompactionThresholds(cf_def);
CFMetaData.validateMemtableSettings(cf_def);

47
src/java/org/apache/cassandra/utils/DefaultDouble.java Normal file
View File

@ -0,0 +1,47 @@
package org.apache.cassandra.utils;
/**
* 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.
*/
public class DefaultDouble
{
private final double originalValue;
private double currentValue;
public DefaultDouble(double value)
{
originalValue = value;
currentValue = value;
}
public double value()
{
return currentValue;
}
public void set(double d)
{
currentValue = d;
}
public boolean isModified()
{
return originalValue != currentValue;
}
}

47
src/java/org/apache/cassandra/utils/DefaultInteger.java Normal file
View File

@ -0,0 +1,47 @@
package org.apache.cassandra.utils;
/**
* 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.
*/
public class DefaultInteger
{
private final int originalValue;
private int currentValue;
public DefaultInteger(int value)
{
originalValue = value;
currentValue = value;
}
public int value()
{
return currentValue;
}
public void set(int i)
{
currentValue = i;
}
public boolean isModified()
{
return originalValue != currentValue;
}
}

2
test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java
View File

@ -95,7 +95,7 @@ public class LongCompactionSpeedTest extends CleanupHelper
Thread.sleep(1000);
long start = System.currentTimeMillis();
CompactionManager.instance.doCompaction(store, sstables, (int) (System.currentTimeMillis() / 1000) - DatabaseDescriptor.getCFMetaData(TABLE1, "Standard1").gcGraceSeconds);
CompactionManager.instance.doCompaction(store, sstables, (int) (System.currentTimeMillis() / 1000) - DatabaseDescriptor.getCFMetaData(TABLE1, "Standard1").getGcGraceSeconds());
System.out.println(String.format("%s: sstables=%d rowsper=%d colsper=%d: %d ms",
this.getClass().getName(),
sstableCount,

6
test/system/test_thrift_server.py
View File

@ -1329,12 +1329,16 @@ class TestMutations(ThriftTester):
modified_cf = CfDef('Keyspace1', 'ToBeIndexed', column_metadata=[modified_cd])
modified_cf.id = cfid
client.system_update_column_family(modified_cf)
ks1 = client.describe_keyspace('Keyspace1')
server_cf = [x for x in ks1.cf_defs if x.name=='ToBeIndexed'][0]
assert server_cf
assert server_cf.column_metadata[0].index_type == modified_cd.index_type
assert server_cf.column_metadata[0].index_name == modified_cd.index_name
# sleep a bit to give time for the index to build.
time.sleep(0.1)
# simple query on one index expression
cp = ColumnParent('ToBeIndexed')
sp = SlicePredicate(slice_range=SliceRange('', ''))

2
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -312,7 +312,7 @@ public class ColumnFamilyStoreTest extends CleanupHelper
rm.apply();
ColumnFamilyStore cfs = table.getColumnFamilyStore("Indexed2");
ColumnDefinition old = cfs.metadata.column_metadata.get(ByteBufferUtil.bytes("birthdate"));
ColumnDefinition old = cfs.metadata.getColumn_metadata().get(ByteBufferUtil.bytes("birthdate"));
ColumnDefinition cd = new ColumnDefinition(old.name, old.validator.getClass().getName(), IndexType.KEYS, "birthdate_index");
cfs.addIndex(cd);
while (!SystemTable.isIndexBuilt("Keyspace1", cfs.getIndexedColumnFamilyStore(ByteBufferUtil.bytes("birthdate")).columnFamily))

139
test/unit/org/apache/cassandra/db/DefsTest.java
View File

@ -144,7 +144,7 @@ public class DefsTest extends CleanupHelper
assert DatabaseDescriptor.getDefsVersion().equals(ver2);
// drop it.
Migration m3 = new DropColumnFamily("Keyspace1", "MigrationCf_2", true);
Migration m3 = new DropColumnFamily("Keyspace1", "MigrationCf_2");
m3.apply();
UUID ver3 = m3.getVersion();
assert DatabaseDescriptor.getDefsVersion().equals(ver3);
@ -225,7 +225,7 @@ public class DefsTest extends CleanupHelper
store.getFlushPath();
assert DefsTable.getFiles(cfm.tableName, cfm.cfName).size() > 0;
new DropColumnFamily(ks.name, cfm.cfName, true).apply();
new DropColumnFamily(ks.name, cfm.cfName).apply();
assert !DatabaseDescriptor.getTableDefinition(ks.name).cfMetaData().containsKey(cfm.cfName);
@ -344,7 +344,7 @@ public class DefsTest extends CleanupHelper
store.forceBlockingFlush();
assert DefsTable.getFiles(cfm.tableName, cfm.cfName).size() > 0;
new DropKeyspace(ks.name, true).apply();
new DropKeyspace(ks.name).apply();
assert DatabaseDescriptor.getTableDefinition(ks.name) == null;
@ -545,156 +545,132 @@ public class DefsTest extends CleanupHelper
assert DatabaseDescriptor.getTableDefinition(cf.tableName) != null;
assert DatabaseDescriptor.getTableDefinition(cf.tableName) == ksm;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName) != null;
// updating certain fields should fail.
CfDef cf_def = new CfDef();
cf_def.setId(cf.cfId);
cf_def.setKeyspace(cf.tableName);
cf_def.setName(cf.cfName);
cf_def.setColumn_type(cf.cfType.name());
cf_def.setComment(cf.comment);
cf_def.setComparator_type(cf.comparator.getClass().getName());
cf_def.setSubcomparator_type(null);
cf_def.setGc_grace_seconds(cf.gcGraceSeconds);
cf_def.setKey_cache_size(cf.keyCacheSize);
cf_def.setRead_repair_chance(cf.readRepairChance);
cf_def.setRow_cache_size(43.3);
cf_def.setColumn_metadata(new ArrayList<ColumnDef>());
cf_def.setDefault_validation_class("BytesType");
cf_def.setMin_compaction_threshold(5);
cf_def.setMax_compaction_threshold(31);
org.apache.cassandra.avro.CfDef cf_def = CFMetaData.convertToAvro(cf);
cf_def.row_cache_size = 43.3;
cf_def.column_metadata = new ArrayList<org.apache.cassandra.avro.ColumnDef>();
cf_def.default_validation_class ="BytesType";
cf_def.min_compaction_threshold = 5;
cf_def.max_compaction_threshold = 31;
// test valid operations.
cf_def.setComment("Modified comment");
CFMetaData updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.comment = "Modified comment";
new UpdateColumnFamily(cf_def).apply(); // doesn't get set back here.
cf_def.setRow_cache_size(2d);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.row_cache_size = 2d;
new UpdateColumnFamily(cf_def).apply();
cf_def.setKey_cache_size(3d);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.key_cache_size = 3d;
new UpdateColumnFamily(cf_def).apply();
cf_def.setRead_repair_chance(0.23);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.read_repair_chance = 0.23;
new UpdateColumnFamily(cf_def).apply();
cf_def.setGc_grace_seconds(12);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.gc_grace_seconds = 12;
new UpdateColumnFamily(cf_def).apply();
cf_def.setDefault_validation_class("UTF8Type");
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.default_validation_class = "UTF8Type";
new UpdateColumnFamily(cf_def).apply();
cf_def.setMin_compaction_threshold(3);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.min_compaction_threshold = 3;
new UpdateColumnFamily(cf_def).apply();
cf_def.setMax_compaction_threshold(33);
updateCfm = cf.apply(cf_def);
new UpdateColumnFamily(cf, updateCfm).apply();
cf = updateCfm;
cf_def.max_compaction_threshold = 33;
new UpdateColumnFamily(cf_def).apply();
// can't test changing the reconciler because there is only one impl.
// check the cumulative affect.
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).comment.equals(cf_def.comment);
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).rowCacheSize == cf_def.row_cache_size;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).keyCacheSize == cf_def.key_cache_size;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).readRepairChance == cf_def.read_repair_chance;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).gcGraceSeconds == cf_def.gc_grace_seconds;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).defaultValidator == UTF8Type.instance;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).getComment().equals(cf_def.comment);
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).getRowCacheSize() == cf_def.row_cache_size;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).getKeyCacheSize() == cf_def.key_cache_size;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).getReadRepairChance() == cf_def.read_repair_chance;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, cf.cfName).getGcGraceSeconds() == cf_def.gc_grace_seconds;
assert DatabaseDescriptor.getCFMetaData(cf.tableName, 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;
try
{
cf_def.setId(cf_def.getId() + 1);
updateCfm = cf.apply(cf_def);
cf_def.id++;
cf.apply(cf_def);
throw new AssertionError("Should have blown up when you used a different id.");
}
catch (ConfigurationException expected)
{
cf_def.setId(oldId);
cf_def.id = oldId;
}
String oldStr = cf_def.getName();
CharSequence oldStr = cf_def.name;
try
{
cf_def.setName(cf_def.getName() + "_renamed");
updateCfm = cf.apply(cf_def);
cf_def.name = cf_def.name + "_renamed";
cf.apply(cf_def);
throw new AssertionError("Should have blown up when you used a different name.");
}
catch (ConfigurationException expected)
{
cf_def.setName(oldStr);
cf_def.name = oldStr;
}
oldStr = cf_def.getKeyspace();
oldStr = cf_def.keyspace;
try
{
cf_def.setKeyspace(oldStr + "_renamed");
updateCfm = cf.apply(cf_def);
cf_def.keyspace = oldStr + "_renamed";
cf.apply(cf_def);
throw new AssertionError("Should have blown up when you used a different keyspace.");
}
catch (ConfigurationException expected)
{
cf_def.setKeyspace(oldStr);
cf_def.keyspace = oldStr;
}
try
{
cf_def.setColumn_type(ColumnFamilyType.Super.name());
updateCfm = cf.apply(cf_def);
cf_def.column_type = ColumnFamilyType.Super.name();
cf.apply(cf_def);
throw new AssertionError("Should have blwon up when you used a different cf type.");
}
catch (ConfigurationException expected)
{
cf_def.setColumn_type(ColumnFamilyType.Standard.name());
cf_def.column_type = ColumnFamilyType.Standard.name();
}
oldStr = cf_def.getComparator_type();
oldStr = cf_def.comparator_type;
try
{
cf_def.setComparator_type(BytesType.class.getSimpleName());
updateCfm = cf.apply(cf_def);
cf_def.comparator_type = BytesType.class.getSimpleName();
cf.apply(cf_def);
throw new AssertionError("Should have blown up when you used a different comparator.");
}
catch (ConfigurationException expected)
{
cf_def.setComparator_type(UTF8Type.class.getSimpleName());
cf_def.comparator_type = UTF8Type.class.getSimpleName();
}
try
{
cf_def.setMin_compaction_threshold(34);
updateCfm = cf.apply(cf_def);
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.setMin_compaction_threshold(3);
cf_def.min_compaction_threshold = 3;
}
try
{
cf_def.setMax_compaction_threshold(2);
updateCfm = cf.apply(cf_def);
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.setMax_compaction_threshold(33);
cf_def.max_compaction_threshold = 33;
}
}
@ -720,5 +696,4 @@ public class DefsTest extends CleanupHelper
CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS,
Collections.<ByteBuffer, ColumnDefinition>emptyMap());
}
}

1
test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
View File

@ -81,6 +81,7 @@ public class SSTableWriterTest extends CleanupHelper {
FileUtils.deleteWithConfirm(orig.descriptor.filenameFor(Component.FILTER));
SSTableReader sstr = CompactionManager.instance.submitSSTableBuild(orig.descriptor).get();
assert sstr != null;
ColumnFamilyStore cfs = Table.open("Keyspace1").getColumnFamilyStore("Indexed1");
cfs.addSSTable(sstr);
cfs.buildSecondaryIndexes(cfs.getSSTables(), cfs.getIndexedColumns());