diff --git a/CHANGES.txt b/CHANGES.txt index 5bcf52cf58..8e5835e973 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/avro/internode.genavro b/src/avro/internode.genavro index c7dbd875d1..98f517a443 100644 --- a/src/avro/internode.genavro +++ b/src/avro/internode.genavro @@ -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") diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index bec5039d7e..5901a38615 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -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) diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 80a0e1533c..c649588209 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -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 column_metadata; + private final Map 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 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 metadata = new HashMap(); - 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 toRemove = new HashSet(); + Set newIndexNames = new HashSet(); + Set toAdd = new HashSet(); + 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 columnMeta = new ArrayList(); + 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 { diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java index b434b95aa5..9fc8ed24e8 100644 --- a/src/java/org/apache/cassandra/config/ColumnDefinition.java +++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java @@ -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; + } } diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 0af62bdfb7..c61af34da3 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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; } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 1685dd2d43..361cbc96fe 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 indexedColumns; + private final ConcurrentSkipListMap indexedColumns; // TODO binarymemtable ops are not threadsafe (do they need to be?) private AtomicReference 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(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 sstables, SortedSet 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() diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index 8b4eeccde8..e391c8373b 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -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 estimatedCompactions = new NonBlockingHashMap(); + + 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> buckets = getBuckets(convertSSTablesToPairs(cfs.getSSTables()), 50L * 1024L * 1024L); - updateEstimateFor(cfs, buckets); - - for (List 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> buckets = getBuckets(convertSSTablesToPairs(cfs.getSSTables()), 50L * 1024L * 1024L); + updateEstimateFor(cfs, buckets); + + for (List 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 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 sstables; - if (skip > 0) + compactionLock.lock(); + try { - sstables = new ArrayList(); - for (SSTableReader sstable : cfStore.getSSTables()) + if (cfStore.isInvalid()) + return this; + Collection sstables; + if (skip > 0) { - if (sstable.length() < skip * 1024L * 1024L * 1024L) + sstables = new ArrayList(); + 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 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 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 callable = new Callable() { 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)"); + } + } } diff --git a/src/java/org/apache/cassandra/db/RowIteratorFactory.java b/src/java/org/apache/cassandra/db/RowIteratorFactory.java index f4c19e4b11..9c4456af1a 100644 --- a/src/java/org/apache/cassandra/db/RowIteratorFactory.java +++ b/src/java/org/apache/cassandra/db/RowIteratorFactory.java @@ -107,7 +107,7 @@ public class RowIteratorFactory // reduce rows from all sources into a single row ReducingIterator reduced = new ReducingIterator(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 colIters = new ArrayList(); private DecoratedKey key; diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index 19146e6d9b..5217d83ac0 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -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()); } } diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index be5167392c..061cd1eac6 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -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 { diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 1932ee5be6..ab2d2d6a4a 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -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(); diff --git a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java index ca27d1cd17..0a3cb90aaa 100644 --- a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java @@ -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; } } diff --git a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java index ef9c205768..e76152707d 100644 --- a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java +++ b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java @@ -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; } } diff --git a/src/java/org/apache/cassandra/db/migration/Migration.java b/src/java/org/apache/cassandra/db/migration/Migration.java index 3f1fa51e84..eee9b5839c 100644 --- a/src/java/org/apache/cassandra/db/migration/Migration.java +++ b/src/java/org/apache/cassandra/db/migration/Migration.java @@ -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() {} diff --git a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java index 3e671d9b50..9835afc048 100644 --- a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java +++ b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java @@ -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 newCfs = new ArrayList(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 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); } } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index 6a18d433fd..6b557c6b62 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -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(); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 3f26f14e6f..0cf5721d2e 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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() diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 296f44874f..9e8de67bb9 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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; diff --git a/src/java/org/apache/cassandra/streaming/StreamInSession.java b/src/java/org/apache/cassandra/streaming/StreamInSession.java index 898de850c8..5243a9a843 100644 --- a/src/java/org/apache/cassandra/streaming/StreamInSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamInSession.java @@ -136,6 +136,8 @@ public class StreamInSession try { SSTableReader sstable = future.get(); + if (sstable == null) + continue; cfs.addSSTable(sstable); sstables.add(sstable); } diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 3acbb6f86d..d647282aa9 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -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); diff --git a/src/java/org/apache/cassandra/utils/DefaultDouble.java b/src/java/org/apache/cassandra/utils/DefaultDouble.java new file mode 100644 index 0000000000..d9ca813c3f --- /dev/null +++ b/src/java/org/apache/cassandra/utils/DefaultDouble.java @@ -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; + } +} diff --git a/src/java/org/apache/cassandra/utils/DefaultInteger.java b/src/java/org/apache/cassandra/utils/DefaultInteger.java new file mode 100644 index 0000000000..ac98e47260 --- /dev/null +++ b/src/java/org/apache/cassandra/utils/DefaultInteger.java @@ -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; + } +} diff --git a/test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java b/test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java index 0855a51a73..735349b6d3 100644 --- a/test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java +++ b/test/long/org/apache/cassandra/db/LongCompactionSpeedTest.java @@ -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, diff --git a/test/system/test_thrift_server.py b/test/system/test_thrift_server.py index d516c5c57a..8bb2b6a6ab 100644 --- a/test/system/test_thrift_server.py +++ b/test/system/test_thrift_server.py @@ -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('', '')) diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index e422c5192d..212ded3190 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -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)) diff --git a/test/unit/org/apache/cassandra/db/DefsTest.java b/test/unit/org/apache/cassandra/db/DefsTest.java index b6206b0e11..e77083356f 100644 --- a/test/unit/org/apache/cassandra/db/DefsTest.java +++ b/test/unit/org/apache/cassandra/db/DefsTest.java @@ -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()); - 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(); + 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.emptyMap()); } - } diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java index d5242c6a6a..5878c3629f 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableWriterTest.java @@ -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());