You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by gd...@apache.org on 2010/11/19 17:55:38 UTC
svn commit: r1036937 [1/2] - in /cassandra/branches/cassandra-0.7: ./
src/avro/ src/java/org/apache/cassandra/avro/
src/java/org/apache/cassandra/config/ src/java/org/apache/cassandra/db/
src/java/org/apache/cassandra/db/commitlog/ src/java/org/apache/...
Author: gdusbabek
Date: Fri Nov 19 16:55:37 2010
New Revision: 1036937
URL: http://svn.apache.org/viewvc?rev=1036937&view=rev
Log:
fix flush and compaction races with schema updates. patch by gdusbabek and jbellis. CASSANDRA-1715
Added:
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/utils/DefaultDouble.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/utils/DefaultInteger.java
Modified:
cassandra/branches/cassandra-0.7/CHANGES.txt
cassandra/branches/cassandra-0.7/src/avro/internode.genavro
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/avro/CassandraServer.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/CFMetaData.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/ColumnDefinition.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/CompactionManager.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowIteratorFactory.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutation.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/Table.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/Migration.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageService.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamInSession.java
cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java
cassandra/branches/cassandra-0.7/test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java
cassandra/branches/cassandra-0.7/test/system/test_thrift_server.py
cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/db/DefsTest.java
cassandra/branches/cassandra-0.7/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java
Modified: cassandra/branches/cassandra-0.7/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/CHANGES.txt?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.7/CHANGES.txt Fri Nov 19 16:55:37 2010
@@ -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)
Modified: cassandra/branches/cassandra-0.7/src/avro/internode.genavro
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/avro/internode.genavro?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/avro/internode.genavro (original)
+++ cassandra/branches/cassandra-0.7/src/avro/internode.genavro Fri Nov 19 16:55:37 2010
@@ -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")
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/avro/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/avro/CassandraServer.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/avro/CassandraServer.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/avro/CassandraServer.java Fri Nov 19 16:55:37 2010
@@ -694,8 +694,8 @@ public class CassandraServer implements
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
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
try
{
- applyMigrationOnStage(new DropKeyspace(keyspace.toString(), true));
+ applyMigrationOnStage(new DropKeyspace(keyspace.toString()));
return DatabaseDescriptor.getDefsVersion().toString();
}
catch (ConfigurationException e)
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/CFMetaData.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/CFMetaData.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/CFMetaData.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/CFMetaData.java Fri Nov 19 16:55:37 2010
@@ -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)
- {
- metadata = column_metadata;
- }
- else
- {
- 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);
- }
+ 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)
+ {
+ newIndexNames.add(def.name);
+ if (!column_metadata.containsKey(def.name))
+ toAdd.add(def);
+ }
+ 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)
+ {
+ 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
{
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/ColumnDefinition.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/ColumnDefinition.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/ColumnDefinition.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/ColumnDefinition.java Fri Nov 19 16:55:37 2010
@@ -36,8 +36,8 @@ import org.apache.cassandra.utils.FBUtil
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;
+ }
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/DatabaseDescriptor.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/DatabaseDescriptor.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/config/DatabaseDescriptor.java Fri Nov 19 16:55:37 2010
@@ -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;
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Fri Nov 19 16:55:37 2010
@@ -103,6 +103,7 @@ public class ColumnFamilyStore implement
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 implement
/* 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 implement
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 implement
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 implement
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 implement
// 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 implement
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 implement
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
- {
- forceBlockingFlush();
- }
- catch (ExecutionException e)
- {
- throw new RuntimeException(e);
- }
- catch (InterruptedException e)
+ public void run()
{
- throw new AssertionError(e);
+ 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);
}
- 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 implement
}
}
- // 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 implement
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 implement
*/
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 implement
{
ssTables.replace(sstables, replacements);
}
+
+ public boolean isInvalid()
+ {
+ return invalid;
+ }
public void removeAllSSTables()
{
@@ -1006,7 +1068,7 @@ public class ColumnFamilyStore implement
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 implement
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 implement
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()
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/CompactionManager.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/CompactionManager.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/CompactionManager.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/CompactionManager.java Fri Nov 19 16:55:37 2010
@@ -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 implement
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 implement
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,28 +97,38 @@ public class CompactionManager implement
{
public Integer call() throws IOException
{
- Integer minThreshold = cfs.getMinimumCompactionThreshold();
- Integer maxThreshold = cfs.getMaximumCompactionThreshold();
-
- if (minThreshold == 0 || maxThreshold == 0)
- {
- 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)
+ compactionLock.lock();
+ try
{
- 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 implement
{
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 implement
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 implement
{
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 implement
{
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 implement
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 implement
}
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 implement
{
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 implement
{
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 implement
{
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)");
+ }
+ }
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowIteratorFactory.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowIteratorFactory.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowIteratorFactory.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowIteratorFactory.java Fri Nov 19 16:55:37 2010
@@ -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;
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutation.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutation.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutation.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/RowMutation.java Fri Nov 19 16:55:37 2010
@@ -41,7 +41,6 @@ import org.apache.cassandra.thrift.Colum
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());
}
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/Table.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/Table.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/Table.java Fri Nov 19 16:55:37 2010
@@ -34,6 +34,7 @@ import java.util.concurrent.ExecutionExc
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
{
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/commitlog/CommitLog.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/commitlog/CommitLog.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/commitlog/CommitLog.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/commitlog/CommitLog.java Fri Nov 19 16:55:37 2010
@@ -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();
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java Fri Nov 19 16:55:37 2010
@@ -5,18 +5,15 @@ import org.apache.cassandra.config.Confi
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 Mi
{
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 Mi
@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
- {
- CompactionManager.instance.submitDrop(cfs).get();
- }
- catch (InterruptedException ex)
+ // 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)
{
- 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 Mi
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 Mi
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;
}
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropKeyspace.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropKeyspace.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/DropKeyspace.java Fri Nov 19 16:55:37 2010
@@ -22,8 +22,6 @@ import org.apache.cassandra.config.CFMet
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
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 Migrat
@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 Migrat
{
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 Migrat
{
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;
}
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/Migration.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/Migration.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/Migration.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/Migration.java Fri Nov 19 16:55:37 2010
@@ -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() {}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java Fri Nov 19 16:55:37 2010
@@ -14,6 +14,7 @@ import org.apache.cassandra.config.KSMet
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.UUIDGe
/** 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);
}
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java Fri Nov 19 16:55:37 2010
@@ -259,6 +259,8 @@ public class SSTableWriter extends SSTab
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();
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageProxy.java Fri Nov 19 16:55:37 2010
@@ -598,7 +598,7 @@ public class StorageProxy implements Sto
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()
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageService.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/service/StorageService.java Fri Nov 19 16:55:37 2010
@@ -1977,21 +1977,21 @@ public class StorageService implements I
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;
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamInSession.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamInSession.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamInSession.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/streaming/StreamInSession.java Fri Nov 19 16:55:37 2010
@@ -136,6 +136,8 @@ public class StreamInSession
try
{
SSTableReader sstable = future.get();
+ if (sstable == null)
+ continue;
cfs.addSSTable(sstable);
sstables.add(sstable);
}
Modified: cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=1036937&r1=1036936&r2=1036937&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/branches/cassandra-0.7/src/java/org/apache/cassandra/thrift/CassandraServer.java Fri Nov 19 16:55:37 2010
@@ -42,6 +42,7 @@ import org.apache.cassandra.config.Confi
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
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
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
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
}
}
+ // @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
throw new InvalidRequestException("Invalid column type " + cf_def.column_type);
}
+ CFMetaData.applyImplicitDefaults(cf_def);
CFMetaData.validateMinMaxCompactionThresholds(cf_def);
CFMetaData.validateMemtableSettings(cf_def);