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

Avoid ID conflicts from concurrent schema changes 

patch by Pavel Yaskevich; reviewed by Sylvain Lebresne for CASSANDRA-3794
This commit is contained in:
Pavel Yaskevich 2012-05-17 22:35:22 +03:00
parent 4ec819eaa6
commit 90170d1594
34 changed files with 312 additions and 231 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -8,6 +8,7 @@
* Save IndexSummary into new SSTable 'Summary' component (CASSANDRA-2392)
* Add support for range tombstones (CASSANDRA-3708)
* Improve MessagingService efficiency (CASSANDRA-3617)
* Avoid ID conflicts from concurrent schema changes (CASSANDRA-3794)
1.1.1-dev

14
src/java/org/apache/cassandra/cache/RowCacheKey.java vendored
View File

@ -21,6 +21,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.UUID;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.TypeSizes;
@ -31,15 +32,15 @@ import org.apache.cassandra.utils.Pair;
public class RowCacheKey implements CacheKey, Comparable<RowCacheKey>
{
public final int cfId;
public final UUID cfId;
public final byte[] key;
public RowCacheKey(int cfId, DecoratedKey key)
public RowCacheKey(UUID cfId, DecoratedKey key)
{
this(cfId, key.key);
}
public RowCacheKey(int cfId, ByteBuffer key)
public RowCacheKey(UUID cfId, ByteBuffer key)
{
this.cfId = cfId;
this.key = ByteBufferUtil.getArray(key);
@ -69,21 +70,20 @@ public class RowCacheKey implements CacheKey, Comparable<RowCacheKey>
RowCacheKey that = (RowCacheKey) o;
if (cfId != that.cfId) return false;
return Arrays.equals(key, that.key);
return cfId.equals(that.cfId) && Arrays.equals(key, that.key);
}
@Override
public int hashCode()
{
int result = cfId;
int result = cfId.hashCode();
result = 31 * result + (key != null ? Arrays.hashCode(key) : 0);
return result;
}
public int compareTo(RowCacheKey otherKey)
{
return (cfId < otherKey.cfId) ? -1 : ((cfId == otherKey.cfId) ? FBUtilities.compareUnsigned(key, otherKey.key, 0, 0, key.length, otherKey.key.length) : 1);
return (cfId.compareTo(otherKey.cfId) < 0) ? -1 : ((cfId.equals(otherKey.cfId)) ? FBUtilities.compareUnsigned(key, otherKey.key, 0, 0, key.length, otherKey.key.length) : 1);
}
@Override

6
src/java/org/apache/cassandra/config/Avro.java
View File

@ -127,8 +127,7 @@ public class Avro
cf.name.toString(),
ColumnFamilyType.create(cf.column_type.toString()),
comparator,
subcolumnComparator,
cf.id);
subcolumnComparator);
// When we pull up an old avro CfDef which doesn't have these arguments,
// it doesn't default them correctly. Without explicit defaulting,
@ -178,6 +177,9 @@ public class Avro
throw new RuntimeException(e);
}
// adding old -> new style ID mapping to support backward compatibility
Schema.instance.addOldCfIdMapping(cf.id, newCFMD.cfId);
return newCFMD.comment(cf.comment.toString())
.readRepairChance(cf.read_repair_chance)
.dcLocalReadRepairChance(cf.dclocal_read_repair_chance)

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

@ -21,10 +21,12 @@ import java.io.IOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.*;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
@ -179,7 +181,7 @@ public final class CFMetaData
}
//REQUIRED
public final Integer cfId; // internal id, never exposed to user
public final UUID cfId; // internal id, never exposed to user
public final String ksName; // name of keyspace
public final String cfName; // name of this column family
public final ColumnFamilyType cfType; // standard, super
@ -243,10 +245,10 @@ public final class CFMetaData
public CFMetaData(String keyspace, String name, ColumnFamilyType type, AbstractType<?> comp, AbstractType<?> subcc)
{
this(keyspace, name, type, comp, subcc, Schema.instance.nextCFId());
this(keyspace, name, type, comp, subcc, getId(keyspace, name));
}
CFMetaData(String keyspace, String name, ColumnFamilyType type, AbstractType<?> comp, AbstractType<?> subcc, int id)
CFMetaData(String keyspace, String name, ColumnFamilyType type, AbstractType<?> comp, AbstractType<?> subcc, UUID id)
{
// Final fields must be set in constructor
ksName = keyspace;
@ -254,9 +256,6 @@ public final class CFMetaData
cfType = type;
comparator = comp;
subcolumnComparator = enforceSubccDefault(type, subcc);
// System cfs have specific ids, and copies of old CFMDs need
// to copy over the old id.
cfId = id;
this.init();
@ -272,6 +271,11 @@ public final class CFMetaData
return (comment == null) ? "" : comment.toString();
}
static UUID getId(String ksName, String cfName)
{
return UUID.nameUUIDFromBytes(ArrayUtils.addAll(ksName.getBytes(), cfName.getBytes()));
}
private void init()
{
// Set a bunch of defaults
@ -306,10 +310,13 @@ public final class CFMetaData
updateCfDef(); // init cqlCfDef
}
private static CFMetaData newSystemMetadata(String cfName, int cfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
private static CFMetaData newSystemMetadata(String cfName, int oldCfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
{
ColumnFamilyType type = subcc == null ? ColumnFamilyType.Standard : ColumnFamilyType.Super;
CFMetaData newCFMD = new CFMetaData(Table.SYSTEM_TABLE, cfName, type, comparator, subcc, cfId);
CFMetaData newCFMD = new CFMetaData(Table.SYSTEM_TABLE, cfName, type, comparator, subcc);
// adding old -> new style ID mapping to support backward compatibility
Schema.instance.addOldCfIdMapping(oldCfId, newCFMD.cfId);
return newCFMD.comment(comment)
.readRepairChance(0)
@ -317,7 +324,7 @@ public final class CFMetaData
.gcGraceSeconds(0);
}
private static CFMetaData newSchemaMetadata(String cfName, int cfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
private static CFMetaData newSchemaMetadata(String cfName, int oldCfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
{
/*
* Schema column families needs a gc_grace (since they are replicated
@ -325,7 +332,7 @@ public final class CFMetaData
* could be dead for that long a time.
*/
int gcGrace = 120 * 24 * 3600; // 3 months
return newSystemMetadata(cfName, cfId, comment, comparator, subcc).gcGraceSeconds(gcGrace);
return newSystemMetadata(cfName, oldCfId, comment, comparator, subcc).gcGraceSeconds(gcGrace);
}
public static CFMetaData newIndexMetadata(CFMetaData parent, ColumnDefinition info, AbstractType<?> columnComparator)
@ -527,7 +534,7 @@ public final class CFMetaData
.append(keyValidator, rhs.keyValidator)
.append(minCompactionThreshold, rhs.minCompactionThreshold)
.append(maxCompactionThreshold, rhs.maxCompactionThreshold)
.append(cfId.intValue(), rhs.cfId.intValue())
.append(cfId, rhs.cfId)
.append(column_metadata, rhs.column_metadata)
.append(keyAlias, rhs.keyAlias)
.append(columnAliases, rhs.columnAliases)
@ -623,8 +630,7 @@ public final class CFMetaData
cf_def.name,
cfType,
TypeParser.parse(cf_def.comparator_type),
cf_def.subcomparator_type == null ? null : TypeParser.parse(cf_def.subcomparator_type),
cf_def.isSetId() ? cf_def.id : Schema.instance.nextCFId());
cf_def.subcomparator_type == null ? null : TypeParser.parse(cf_def.subcomparator_type));
if (cf_def.isSetGc_grace_seconds()) { newCFMD.gcGraceSeconds(cf_def.gc_grace_seconds); }
if (cf_def.isSetMin_compaction_threshold()) { newCFMD.minCompactionThreshold(cf_def.min_compaction_threshold); }
@ -799,7 +805,6 @@ public final class CFMetaData
public org.apache.cassandra.thrift.CfDef toThrift()
{
org.apache.cassandra.thrift.CfDef def = new org.apache.cassandra.thrift.CfDef(ksName, cfName);
def.setId(cfId);
def.setColumn_type(cfType.name());
def.setComparator_type(comparator.toString());
if (subcolumnComparator != null)
@ -1142,7 +1147,11 @@ public final class CFMetaData
ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_COLUMNFAMILIES_CF);
int ldt = (int) (System.currentTimeMillis() / 1000);
cf.addColumn(Column.create(cfId, timestamp, cfName, "id"));
Integer oldId = Schema.instance.convertNewCfId(cfId);
if (oldId != null) // keep old ids (see CASSANDRA-3794 for details)
cf.addColumn(Column.create(oldId, timestamp, cfName, "id"));
cf.addColumn(Column.create(cfType.toString(), timestamp, cfName, "type"));
cf.addColumn(Column.create(comparator.toString(), timestamp, cfName, "comparator"));
if (subcolumnComparator != null)
@ -1179,8 +1188,11 @@ public final class CFMetaData
result.getString("columnfamily"),
ColumnFamilyType.valueOf(result.getString("type")),
TypeParser.parse(result.getString("comparator")),
result.has("subcomparator") ? TypeParser.parse(result.getString("subcomparator")) : null,
result.getInt("id"));
result.has("subcomparator") ? TypeParser.parse(result.getString("subcomparator")) : null);
if (result.has("id"))// try to identify if ColumnFamily Id is old style (before C* 1.2) and add old -> new mapping if so
Schema.instance.addOldCfIdMapping(result.getInt("id"), cfm.cfId);
cfm.readRepairChance(result.getDouble("read_repair_chance"));
cfm.dcLocalReadRepairChance(result.getDouble("local_read_repair_chance"));
cfm.replicateOnWrite(result.getBoolean("replicate_on_write"));

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

@ -536,7 +536,6 @@ public class DatabaseDescriptor
}
Schema.instance.updateVersion();
Schema.instance.fixCFMaxId();
}
private static boolean hasExistingNoSystemTables()

66
src/java/org/apache/cassandra/config/Schema.java
View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.config;
import java.io.IOError;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.*;
@ -55,9 +54,6 @@ public class Schema
*/
public static final int NAME_LENGTH = 48;
private static final int MIN_CF_ID = 1000;
private final AtomicInteger cfIdGen = new AtomicInteger(MIN_CF_ID);
/* metadata map for faster table lookup */
private final Map<String, KSMetaData> tables = new NonBlockingHashMap<String, KSMetaData>();
@ -65,7 +61,9 @@ public class Schema
private final Map<String, Table> tableInstances = new NonBlockingHashMap<String, Table>();
/* metadata map for faster ColumnFamily lookup */
private final BiMap<Pair<String, String>, Integer> cfIdMap = HashBiMap.create();
private final BiMap<Pair<String, String>, UUID> cfIdMap = HashBiMap.create();
// mapping from old ColumnFamily Id (Integer) to a new version which is UUID
private final BiMap<Integer, UUID> oldCfIdMap = HashBiMap.create();
private volatile UUID version;
private final ReadWriteLock versionLock = new ReentrantReadWriteLock();
@ -106,8 +104,6 @@ public class Schema
setTableDefinition(keyspaceDef);
fixCFMaxId();
return this;
}
@ -184,7 +180,7 @@ public class Schema
*
* @return metadata about ColumnFamily
*/
public CFMetaData getCFMetaData(Integer cfId)
public CFMetaData getCFMetaData(UUID cfId)
{
Pair<String,String> cf = getCF(cfId);
return (cf == null) ? null : getCFMetaData(cf.left, cf.right);
@ -343,11 +339,34 @@ public class Schema
/* ColumnFamily query/control methods */
public void addOldCfIdMapping(Integer oldId, UUID newId)
{
if (oldId == null)
return;
oldCfIdMap.put(oldId, newId);
}
public UUID convertOldCfId(Integer oldCfId)
{
UUID cfId = oldCfIdMap.get(oldCfId);
if (cfId == null)
throw new IllegalArgumentException("ColumnFamily identified by old " + oldCfId + " was not found.");
return cfId;
}
public Integer convertNewCfId(UUID newCfId)
{
return oldCfIdMap.containsValue(newCfId) ? oldCfIdMap.inverse().get(newCfId) : null;
}
/**
* @param cfId The identifier of the ColumnFamily to lookup
* @return The (ksname,cfname) pair for the given id, or null if it has been dropped.
*/
public Pair<String,String> getCF(Integer cfId)
public Pair<String,String> getCF(UUID cfId)
{
return cfIdMap.inverse().get(cfId);
}
@ -360,7 +379,7 @@ public class Schema
*
* @return The id for the given (ksname,cfname) pair, or null if it has been dropped.
*/
public Integer getId(String ksName, String cfName)
public UUID getId(String ksName, String cfName)
{
return cfIdMap.get(new Pair<String, String>(ksName, cfName));
}
@ -382,8 +401,6 @@ public class Schema
logger.debug("Adding {} to cfIdMap", cfm);
cfIdMap.put(key, cfm.cfId);
fixCFMaxId();
}
/**
@ -396,30 +413,6 @@ public class Schema
cfIdMap.remove(new Pair<String, String>(cfm.ksName, cfm.cfName));
}
/**
* This gets called after initialization to make sure that id generation happens properly.
*/
public void fixCFMaxId()
{
int cval, nval;
do
{
cval = cfIdGen.get();
int inMap = cfIdMap.isEmpty() ? 0 : Collections.max(cfIdMap.values()) + 1;
// never set it to less than 1000. this ensures that we have enough system CFids for future use.
nval = Math.max(Math.max(inMap, cval), MIN_CF_ID);
}
while (!cfIdGen.compareAndSet(cval, nval));
}
/**
* @return identifier for the new ColumnFamily (called primarily by CFMetaData constructor)
*/
public int nextCFId()
{
return cfIdGen.getAndIncrement();
}
/* Version control */
/**
@ -494,6 +487,5 @@ public class Schema
}
updateVersionAndAnnounce();
fixCFMaxId();
}
}

10
src/java/org/apache/cassandra/db/ColumnFamily.java
View File

@ -19,9 +19,11 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.UUID;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.utils.*;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.cassandra.cache.IRowCacheEntry;
@ -32,8 +34,6 @@ import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.sstable.ColumnStats;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.utils.*;
public class ColumnFamily extends AbstractColumnContainer implements IRowCacheEntry
{
@ -41,12 +41,12 @@ public class ColumnFamily extends AbstractColumnContainer implements IRowCacheEn
public static final ColumnFamilySerializer serializer = new ColumnFamilySerializer();
private final CFMetaData cfm;
public static ColumnFamily create(Integer cfId)
public static ColumnFamily create(UUID cfId)
{
return create(Schema.instance.getCFMetaData(cfId));
}
public static ColumnFamily create(Integer cfId, ISortedColumns.Factory factory)
public static ColumnFamily create(UUID cfId, ISortedColumns.Factory factory)
{
return create(Schema.instance.getCFMetaData(cfId), factory);
}
@ -108,7 +108,7 @@ public class ColumnFamily extends AbstractColumnContainer implements IRowCacheEn
return cf;
}
public Integer id()
public UUID id()
{
return cfm.cfId;
}

62
src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
View File

@ -20,6 +20,8 @@ package org.apache.cassandra.db;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.UUID;
import org.apache.cassandra.config.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -28,6 +30,8 @@ import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.ISSTableSerializer;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.UUIDGen;
public class ColumnFamilySerializer implements IVersionedSerializer<ColumnFamily>, ISSTableSerializer<ColumnFamily>
{
@ -61,7 +65,7 @@ public class ColumnFamilySerializer implements IVersionedSerializer<ColumnFamily
}
dos.writeBoolean(true);
dos.writeInt(cf.id());
serializeCfId(cf.id(), dos, version);
DeletionInfo.serializer().serialize(cf.deletionInfo(), dos, version);
@ -92,12 +96,7 @@ public class ColumnFamilySerializer implements IVersionedSerializer<ColumnFamily
if (!dis.readBoolean())
return null;
// create a ColumnFamily based on the cf id
int cfId = dis.readInt();
if (Schema.instance.getCF(cfId) == null)
throw new UnknownColumnFamilyException("Couldn't find cfId=" + cfId, cfId);
ColumnFamily cf = ColumnFamily.create(cfId, factory);
ColumnFamily cf = ColumnFamily.create(deserializeCfId(dis, version), factory);
IColumnSerializer columnSerializer = cf.getColumnSerializer();
cf.delete(DeletionInfo.serializer().deserialize(dis, version, cf.getComparator()));
int expireBefore = (int) (System.currentTimeMillis() / 1000);
@ -126,8 +125,8 @@ public class ColumnFamilySerializer implements IVersionedSerializer<ColumnFamily
}
else
{
return typeSizes.sizeof(true) /* nullness bool */
+ typeSizes.sizeof(cf.id()) /* id */
return typeSizes.sizeof(true) /* nullness bool */
+ cfIdSerializedSize(cf.id(), typeSizes, version) /* id */
+ contentSerializedSize(cf, typeSizes, version);
}
}
@ -162,4 +161,49 @@ public class ColumnFamilySerializer implements IVersionedSerializer<ColumnFamily
int expireBefore = (int) (System.currentTimeMillis() / 1000);
deserializeColumnsFromSSTable(dis, cf, size, flag, expireBefore, version);
}
public void serializeCfId(UUID cfId, DataOutput dos, int version) throws IOException
{
if (version < MessagingService.VERSION_12) // try to use CF's old id where possible (CASSANDRA-3794)
{
Integer oldId = Schema.instance.convertNewCfId(cfId);
if (oldId == null)
throw new IOException("Can't serialize ColumnFamily ID " + cfId + " to be used by version " + version +
", because int <-> uuid mapping could not be established (CF was created in mixed version cluster).");
dos.writeInt(oldId);
}
else
UUIDGen.serializer.serialize(cfId, dos, version);
}
public UUID deserializeCfId(DataInput dis, int version) throws IOException
{
// create a ColumnFamily based on the cf id
UUID cfId = (version < MessagingService.VERSION_12)
? Schema.instance.convertOldCfId(dis.readInt())
: UUIDGen.serializer.deserialize(dis, version);
if (Schema.instance.getCF(cfId) == null)
throw new UnknownColumnFamilyException("Couldn't find cfId=" + cfId, cfId);
return cfId;
}
public int cfIdSerializedSize(UUID cfId, TypeSizes typeSizes, int version)
{
if (version < MessagingService.VERSION_12) // try to use CF's old id where possible (CASSANDRA-3794)
{
Integer oldId = Schema.instance.convertNewCfId(cfId);
if (oldId == null)
throw new RuntimeException("Can't serialize ColumnFamily ID " + cfId + " to be used by version " + version +
", because int <-> uuid mapping could not be established (CF was created in mixed version cluster).");
return typeSizes.sizeof(oldId);
}
return typeSizes.sizeof(cfId);
}
}

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

@ -1122,7 +1122,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
* @return the entire row for filter.key, if present in the cache (or we can cache it), or just the column
* specified by filter otherwise
*/
private ColumnFamily getThroughCache(Integer cfId, QueryFilter filter)
private ColumnFamily getThroughCache(UUID cfId, QueryFilter filter)
{
assert isRowCacheEnabled()
: String.format("Row cache is not enabled on column family [" + getColumnFamilyName() + "]");
@ -1181,7 +1181,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
return cf.isSuper() ? removeDeleted(cf, gcBefore) : removeDeletedCF(cf, gcBefore);
}
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
UUID cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return null; // secondary index
@ -1586,7 +1586,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public void invalidateCachedRow(DecoratedKey key)
{
Integer cfId = Schema.instance.getId(table.name, this.columnFamily);
UUID cfId = Schema.instance.getId(table.name, this.columnFamily);
if (cfId == null)
return; // secondary index

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

@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -55,7 +56,7 @@ public class CounterMutation implements IMutation
return rowMutation.getTable();
}
public Collection<Integer> getColumnFamilyIds()
public Collection<UUID> getColumnFamilyIds()
{
return rowMutation.getColumnFamilyIds();
}

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

@ -20,11 +20,12 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.io.IOException;
import java.util.Collection;
import java.util.UUID;
public interface IMutation
{
public String getTable();
public Collection<Integer> getColumnFamilyIds();
public Collection<UUID> getColumnFamilyIds();
public ByteBuffer key();
public void apply() throws IOException;
public String toString(boolean shallow);

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

@ -50,13 +50,13 @@ public class RowMutation implements IMutation
private final String table;
private final ByteBuffer key;
// map of column family id to mutations for that column family.
protected final Map<Integer, ColumnFamily> modifications;
protected Map<UUID, ColumnFamily> modifications = new HashMap<UUID, ColumnFamily>();
private final Map<Integer, byte[]> preserializedBuffers = new HashMap<Integer, byte[]>();
public RowMutation(String table, ByteBuffer key)
{
this(table, key, new HashMap<Integer, ColumnFamily>());
this(table, key, new HashMap<UUID, ColumnFamily>());
}
public RowMutation(String table, Row row)
@ -65,7 +65,7 @@ public class RowMutation implements IMutation
add(row.cf);
}
protected RowMutation(String table, ByteBuffer key, Map<Integer, ColumnFamily> modifications)
protected RowMutation(String table, ByteBuffer key, Map<UUID, ColumnFamily> modifications)
{
this.table = table;
this.key = key;
@ -77,7 +77,7 @@ public class RowMutation implements IMutation
return table;
}
public Collection<Integer> getColumnFamilyIds()
public Collection<UUID> getColumnFamilyIds()
{
return modifications.keySet();
}
@ -92,7 +92,7 @@ public class RowMutation implements IMutation
return modifications.values();
}
public ColumnFamily getColumnFamily(Integer cfId)
public ColumnFamily getColumnFamily(UUID cfId)
{
return modifications.get(cfId);
}
@ -199,8 +199,9 @@ public class RowMutation implements IMutation
*/
public void add(QueryPath path, ByteBuffer value, long timestamp, int timeToLive)
{
Integer id = Schema.instance.getId(table, path.columnFamilyName);
UUID id = Schema.instance.getId(table, path.columnFamilyName);
ColumnFamily columnFamily = modifications.get(id);
if (columnFamily == null)
{
columnFamily = ColumnFamily.create(table, path.columnFamilyName);
@ -211,8 +212,9 @@ public class RowMutation implements IMutation
public void addCounter(QueryPath path, long value)
{
Integer id = Schema.instance.getId(table, path.columnFamilyName);
UUID id = Schema.instance.getId(table, path.columnFamilyName);
ColumnFamily columnFamily = modifications.get(id);
if (columnFamily == null)
{
columnFamily = ColumnFamily.create(table, path.columnFamilyName);
@ -228,7 +230,7 @@ public class RowMutation implements IMutation
public void delete(QueryPath path, long timestamp)
{
Integer id = Schema.instance.getId(table, path.columnFamilyName);
UUID id = Schema.instance.getId(table, path.columnFamilyName);
int localDeleteTime = (int) (System.currentTimeMillis() / 1000);
@ -264,7 +266,7 @@ public class RowMutation implements IMutation
if (!table.equals(rm.table) || !key.equals(rm.key))
throw new IllegalArgumentException();
for (Map.Entry<Integer, ColumnFamily> entry : rm.modifications.entrySet())
for (Map.Entry<UUID, ColumnFamily> entry : rm.modifications.entrySet())
{
// It's slighty faster to assume the key wasn't present and fix if
// not in the case where it wasn't there indeed.
@ -325,7 +327,7 @@ public class RowMutation implements IMutation
if (shallow)
{
List<String> cfnames = new ArrayList<String>(modifications.size());
for (Integer cfid : modifications.keySet())
for (UUID cfid : modifications.keySet())
{
CFMetaData cfm = Schema.instance.getCFMetaData(cfid);
cfnames.add(cfm == null ? "-dropped-" : cfm.cfName);
@ -385,7 +387,7 @@ public class RowMutation implements IMutation
{
RowMutation rm = serializer.deserialize(new DataInputStream(new FastByteArrayInputStream(raw)), version);
boolean hasCounters = false;
for (Map.Entry<Integer, ColumnFamily> entry : rm.modifications.entrySet())
for (Map.Entry<UUID, ColumnFamily> entry : rm.modifications.entrySet())
{
if (entry.getValue().metadata().getDefaultValidator().isCommutative())
{
@ -411,9 +413,9 @@ public class RowMutation implements IMutation
int size = rm.modifications.size();
dos.writeInt(size);
assert size >= 0;
for (Map.Entry<Integer,ColumnFamily> entry : rm.modifications.entrySet())
for (Map.Entry<UUID, ColumnFamily> entry : rm.modifications.entrySet())
{
dos.writeInt(entry.getKey());
ColumnFamily.serializer.serializeCfId(entry.getKey(), dos, version);
ColumnFamily.serializer.serialize(entry.getValue(), dos, version);
}
}
@ -422,13 +424,13 @@ public class RowMutation implements IMutation
{
String table = dis.readUTF();
ByteBuffer key = ByteBufferUtil.readWithShortLength(dis);
Map<Integer, ColumnFamily> modifications = new HashMap<Integer, ColumnFamily>();
Map<UUID, ColumnFamily> modifications = new HashMap<UUID, ColumnFamily>();
int size = dis.readInt();
for (int i = 0; i < size; ++i)
{
Integer cfid = Integer.valueOf(dis.readInt());
UUID cfId = ColumnFamily.serializer.deserializeCfId(dis, version);
ColumnFamily cf = ColumnFamily.serializer.deserialize(dis, flag, TreeMapBackedSortedColumns.factory(), version);
modifications.put(cfid, cf);
modifications.put(cfId, cf);
}
return new RowMutation(table, key, modifications);
}
@ -446,9 +448,9 @@ public class RowMutation implements IMutation
size += sizes.sizeof((short) keySize) + keySize;
size += sizes.sizeof(rm.modifications.size());
for (Map.Entry<Integer,ColumnFamily> entry : rm.modifications.entrySet())
for (Map.Entry<UUID,ColumnFamily> entry : rm.modifications.entrySet())
{
size += sizes.sizeof(entry.getKey());
size += ColumnFamily.serializer.cfIdSerializedSize(entry.getValue().id(), sizes, version);
size += ColumnFamily.serializer.serializedSize(entry.getValue(), TypeSizes.NATIVE, version);
}

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

@ -20,15 +20,7 @@ package org.apache.cassandra.db;
import java.io.IOError;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@ -87,7 +79,7 @@ public class Table
/* Table name. */
public final String name;
/* ColumnFamilyStore per column family */
private final Map<Integer, ColumnFamilyStore> columnFamilyStores = new ConcurrentHashMap<Integer, ColumnFamilyStore>();
private final Map<UUID, ColumnFamilyStore> columnFamilyStores = new ConcurrentHashMap<UUID, ColumnFamilyStore>();
private final Object[] indexLocks;
private volatile AbstractReplicationStrategy replicationStrategy;
@ -148,13 +140,13 @@ public class Table
public ColumnFamilyStore getColumnFamilyStore(String cfName)
{
Integer id = Schema.instance.getId(name, cfName);
UUID id = Schema.instance.getId(name, cfName);
if (id == null)
throw new IllegalArgumentException(String.format("Unknown table/cf pair (%s.%s)", name, cfName));
return getColumnFamilyStore(id);
}
public ColumnFamilyStore getColumnFamilyStore(Integer id)
public ColumnFamilyStore getColumnFamilyStore(UUID id)
{
ColumnFamilyStore cfs = columnFamilyStores.get(id);
if (cfs == null)
@ -313,7 +305,7 @@ public class Table
}
// best invoked on the compaction mananger.
public void dropCf(Integer cfId) throws IOException
public void dropCf(UUID cfId) throws IOException
{
assert columnFamilyStores.containsKey(cfId);
ColumnFamilyStore cfs = columnFamilyStores.remove(cfId);
@ -342,7 +334,7 @@ public class Table
}
/** adds a cf to internal structures, ends up creating disk files). */
public void initCf(Integer cfId, String cfName)
public void initCf(UUID cfId, String cfName)
{
if (columnFamilyStores.containsKey(cfId))
{
@ -556,7 +548,7 @@ public class Table
public List<Future<?>> flush() throws IOException
{
List<Future<?>> futures = new ArrayList<Future<?>>();
for (Integer cfId : columnFamilyStores.keySet())
for (UUID cfId : columnFamilyStores.keySet())
{
Future<?> future = columnFamilyStores.get(cfId).forceFlush();
if (future != null)

13
src/java/org/apache/cassandra/db/TypeSizes.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.util.UUID;
import org.apache.cassandra.utils.FBUtilities;
@ -30,11 +31,13 @@ public abstract class TypeSizes
private static final int SHORT_SIZE = 2;
private static final int INT_SIZE = 4;
private static final int LONG_SIZE = 8;
private static final int UUID_SIZE = 16;
public abstract int sizeof(boolean value);
public abstract int sizeof(short value);
public abstract int sizeof(int value);
public abstract int sizeof(long value);
public abstract int sizeof(UUID value);
/** assumes UTF8 */
public int sizeof(String value)
@ -92,6 +95,11 @@ public abstract class TypeSizes
{
return LONG_SIZE;
}
public int sizeof(UUID value)
{
return UUID_SIZE;
}
}
public static class VIntEncodedTypeSizes extends TypeSizes
@ -141,5 +149,10 @@ public abstract class TypeSizes
{
return sizeofVInt(i);
}
public int sizeof(UUID value)
{
return sizeofVInt(value.getMostSignificantBits()) + sizeofVInt(value.getLeastSignificantBits());
}
}
}

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

@ -18,13 +18,14 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.UUID;
public class UnknownColumnFamilyException extends IOException
{
public final int cfId;
public final UUID cfId;
public UnknownColumnFamilyException(String msg, int cfId)
public UnknownColumnFamilyException(String msg, UUID cfId)
{
super(msg);
this.cfId = cfId;

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

@ -208,7 +208,7 @@ public class CommitLog implements CommitLogMBean
* @param cfId the column family ID that was flushed
* @param context the replay position of the flush
*/
public void discardCompletedSegments(final Integer cfId, final ReplayPosition context) throws IOException
public void discardCompletedSegments(final UUID cfId, final ReplayPosition context) throws IOException
{
Callable task = new Callable()
{

3
src/java/org/apache/cassandra/db/commitlog/CommitLogAllocator.java
View File

@ -23,6 +23,7 @@ import java.io.IOError;
import java.io.IOException;
import java.util.Collection;
import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@ -300,7 +301,7 @@ public class CommitLogAllocator
if (oldestSegment != null)
{
for (Integer dirtyCFId : oldestSegment.getDirtyCFIDs())
for (UUID dirtyCFId : oldestSegment.getDirtyCFIDs())
{
String keypace = Schema.instance.getCF(dirtyCFId).left;
final ColumnFamilyStore cfs = Table.open(keypace).getColumnFamilyStore(dirtyCFId);

18
src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java
View File

@ -4,11 +4,7 @@ import java.io.DataInputStream;
import java.io.EOFException;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.Checksum;
@ -44,9 +40,9 @@ public class CommitLogReplayer
private final Set<Table> tablesRecovered;
private final List<Future<?>> futures;
private final Map<Integer, AtomicInteger> invalidMutations;
private final Map<UUID, AtomicInteger> invalidMutations;
private final AtomicInteger replayedCount;
private final Map<Integer, ReplayPosition> cfPositions;
private final Map<UUID, ReplayPosition> cfPositions;
private final ReplayPosition globalPosition;
private final Checksum checksum;
private byte[] buffer;
@ -56,11 +52,11 @@ private final AtomicInteger replayedCount;
this.tablesRecovered = new NonBlockingHashSet<Table>();
this.futures = new ArrayList<Future<?>>();
this.buffer = new byte[4096];
this.invalidMutations = new HashMap<Integer, AtomicInteger>();
this.invalidMutations = new HashMap<UUID, AtomicInteger>();
// count the number of replayed mutation. We don't really care about atomicity, but we need it to be a reference.
this.replayedCount = new AtomicInteger();
// compute per-CF and global replay positions
this.cfPositions = new HashMap<Integer, ReplayPosition>();
this.cfPositions = new HashMap<UUID, ReplayPosition>();
for (ColumnFamilyStore cfs : ColumnFamilyStore.all())
{
// it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call
@ -81,8 +77,8 @@ private final AtomicInteger replayedCount;
public int blockForWrites() throws IOException
{
for (Map.Entry<Integer, AtomicInteger> entry : invalidMutations.entrySet())
logger.info(String.format("Skipped %d mutations from unknown (probably removed) CF with id %d", entry.getValue().intValue(), entry.getKey()));
for (Map.Entry<UUID, AtomicInteger> entry : invalidMutations.entrySet())
logger.info(String.format("Skipped %d mutations from unknown (probably removed) CF with id %s", entry.getValue().intValue(), entry.getKey()));
// wait for all the writes to finish on the mutation stage
FBUtilities.waitOnFutures(futures);

11
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -24,6 +24,7 @@ import java.io.RandomAccessFile;
import java.nio.channels.FileChannel;
import java.nio.MappedByteBuffer;
import java.util.Collection;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.zip.Checksum;
@ -58,7 +59,7 @@ public class CommitLogSegment
static final int ENTRY_OVERHEAD_SIZE = 4 + 8 + 8;
// cache which cf is dirty in this segment to avoid having to lookup all ReplayPositions to decide if we can delete this segment
private final HashMap<Integer, Integer> cfLastWrite = new HashMap<Integer, Integer>();
private final HashMap<UUID, Integer> cfLastWrite = new HashMap<UUID, Integer>();
public final long id;
@ -326,7 +327,7 @@ public class CommitLogSegment
* @param cfId the column family ID that is now dirty
* @param position the position the last write for this CF was written at
*/
private void markCFDirty(Integer cfId, Integer position)
private void markCFDirty(UUID cfId, Integer position)
{
cfLastWrite.put(cfId, position);
}
@ -339,7 +340,7 @@ public class CommitLogSegment
* @param cfId the column family ID that is now clean
* @param context the optional clean offset
*/
public void markClean(Integer cfId, ReplayPosition context)
public void markClean(UUID cfId, ReplayPosition context)
{
Integer lastWritten = cfLastWrite.get(cfId);
@ -352,7 +353,7 @@ public class CommitLogSegment
/**
* @return a collection of dirty CFIDs for this segment file.
*/
public Collection<Integer> getDirtyCFIDs()
public Collection<UUID> getDirtyCFIDs()
{
return cfLastWrite.keySet();
}
@ -380,7 +381,7 @@ public class CommitLogSegment
public String dirtyString()
{
StringBuilder sb = new StringBuilder();
for (Integer cfId : cfLastWrite.keySet())
for (UUID cfId : cfLastWrite.keySet())
{
CFMetaData m = Schema.instance.getCFMetaData(cfId);
sb.append(m == null ? "<deleted>" : m.cfName).append(" (").append(cfId).append("), ");

5
src/java/org/apache/cassandra/db/compaction/AbstractCompactionIterable.java
View File

@ -49,10 +49,7 @@ public abstract class AbstractCompactionIterable extends CompactionInfo.Holder i
public CompactionInfo getCompactionInfo()
{
return new CompactionInfo(this.hashCode(),
type,
bytesRead,
totalBytes);
return new CompactionInfo(type, bytesRead, totalBytes);
}
public abstract CloseableIterator<AbstractCompactedRow> iterator();

7
src/java/org/apache/cassandra/db/compaction/CompactionInfo.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.db.compaction;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Schema;
@ -39,7 +40,7 @@ public final class CompactionInfo implements Serializable
this(null, tasktype, bytesComplete, totalBytes);
}
public CompactionInfo(Integer id, OperationType tasktype, long bytesComplete, long totalBytes)
public CompactionInfo(UUID id, OperationType tasktype, long bytesComplete, long totalBytes)
{
this.tasktype = tasktype;
this.bytesComplete = bytesComplete;
@ -53,7 +54,7 @@ public final class CompactionInfo implements Serializable
return new CompactionInfo(cfm == null ? null : cfm.cfId, tasktype, bytesComplete, totalBytes);
}
public Integer getId()
public UUID getId()
{
return cfm == null ? null : cfm.cfId;
}
@ -100,7 +101,7 @@ public final class CompactionInfo implements Serializable
public Map<String, String> asMap()
{
Map<String, String> ret = new HashMap<String, String>();
ret.put("id", Integer.toString(getId()));
ret.put("id", getId().toString());
ret.put("keyspace", getKeyspace());
ret.put("columnfamily", getColumnFamily());
ret.put("bytesComplete", Long.toString(bytesComplete));

6
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -1209,8 +1209,7 @@ public class CompactionManager implements CompactionManagerMBean
{
try
{
return new CompactionInfo(this.hashCode(),
OperationType.CLEANUP,
return new CompactionInfo(OperationType.CLEANUP,
scanner.getCurrentPosition(),
scanner.getLengthInBytes());
}
@ -1235,8 +1234,7 @@ public class CompactionManager implements CompactionManagerMBean
{
try
{
return new CompactionInfo(this.hashCode(),
OperationType.SCRUB,
return new CompactionInfo(OperationType.SCRUB,
dataFile.getFilePointer(),
dataFile.length());
}

3
src/java/org/apache/cassandra/db/index/SecondaryIndexBuilder.java
View File

@ -47,8 +47,7 @@ public class SecondaryIndexBuilder extends CompactionInfo.Holder
public CompactionInfo getCompactionInfo()
{
return new CompactionInfo(this.hashCode(),
OperationType.INDEX_BUILD,
return new CompactionInfo(OperationType.INDEX_BUILD,
iter.getBytesRead(),
iter.getTotalBytes());
}

7
src/java/org/apache/cassandra/streaming/StreamRequest.java
View File

@ -27,6 +27,7 @@ import java.util.List;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.Table;
@ -136,7 +137,7 @@ public class StreamRequest
dos.writeInt(Iterables.size(srm.columnFamilies));
for (ColumnFamilyStore cfs : srm.columnFamilies)
dos.writeInt(cfs.metadata.cfId);
ColumnFamily.serializer.serializeCfId(cfs.metadata.cfId, dos, version);
}
}
@ -163,7 +164,7 @@ public class StreamRequest
List<ColumnFamilyStore> stores = new ArrayList<ColumnFamilyStore>();
int cfsSize = dis.readInt();
for (int i = 0; i < cfsSize; ++i)
stores.add(Table.open(table).getColumnFamilyStore(dis.readInt()));
stores.add(Table.open(table).getColumnFamilyStore(ColumnFamily.serializer.deserializeCfId(dis, version)));
return new StreamRequest(target, ranges, table, stores, sessionId, type);
}
@ -184,7 +185,7 @@ public class StreamRequest
size += TypeSizes.NATIVE.sizeof(sr.type.name());
size += TypeSizes.NATIVE.sizeof(Iterables.size(sr.columnFamilies));
for (ColumnFamilyStore cfs : sr.columnFamilies)
size += TypeSizes.NATIVE.sizeof(cfs.metadata.cfId);
size += ColumnFamily.serializer.cfIdSerializedSize(cfs.metadata.cfId, TypeSizes.NATIVE, version);
return size;
}
}

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

@ -606,7 +606,7 @@ public class FBUtilities
DataOutputBuffer buffer = new DataOutputBuffer(size);
serializer.serialize(object, buffer, version);
assert buffer.getLength() == size && buffer.getData().length == size
: String.format("Final buffer length %s to accomodate data size of %s (predicted %s) for %s",
: String.format("Final buffer length %s to accommodate data size of %s (predicted %s) for %s",
buffer.getData().length, buffer.getLength(), size, object);
return buffer.getData();
}

BIN
test/data/serialization/1.2/db.Row.bin
View File

Binary file not shown.

BIN
test/data/serialization/1.2/db.RowMutation.bin
View File

Binary file not shown.

BIN
test/data/serialization/1.2/streaming.StreamRequestMessage.bin
View File

Binary file not shown.

130
test/unit/org/apache/cassandra/SchemaLoader.java
View File

@ -22,6 +22,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.base.Charsets;
@ -50,8 +51,15 @@ public class SchemaLoader
{
private static Logger logger = LoggerFactory.getLogger(SchemaLoader.class);
private static AtomicInteger oldCfIdGenerator = new AtomicInteger(1000);
@BeforeClass
public static void loadSchema() throws IOException
{
loadSchema(false);
}
public static void loadSchema(boolean withOldCfIds) throws IOException
{
// Cleanup first
cleanupAndLeaveDirs();
@ -71,7 +79,7 @@ public class SchemaLoader
startGossiper();
try
{
for (KSMetaData ksm : schemaDefinition())
for (KSMetaData ksm : schemaDefinition(withOldCfIds))
MigrationManager.announceNewKeyspace(ksm);
}
catch (ConfigurationException e)
@ -91,7 +99,7 @@ public class SchemaLoader
Gossiper.instance.stop();
}
public static Collection<KSMetaData> schemaDefinition() throws ConfigurationException
public static Collection<KSMetaData> schemaDefinition(boolean withOldCfIds) throws ConfigurationException
{
List<KSMetaData> schema = new ArrayList<KSMetaData>();
@ -151,26 +159,26 @@ public class SchemaLoader
opts_rf1,
// Column Families
standardCFMD(ks1, "Standard1"),
standardCFMD(ks1, "Standard2"),
standardCFMD(ks1, "Standard3"),
standardCFMD(ks1, "Standard4"),
standardCFMD(ks1, "StandardLong1"),
standardCFMD(ks1, "StandardLong2"),
standardCFMD(ks1, "Standard1", withOldCfIds),
standardCFMD(ks1, "Standard2", withOldCfIds),
standardCFMD(ks1, "Standard3", withOldCfIds),
standardCFMD(ks1, "Standard4", withOldCfIds),
standardCFMD(ks1, "StandardLong1", withOldCfIds),
standardCFMD(ks1, "StandardLong2", withOldCfIds),
new CFMetaData(ks1,
"ValuesWithQuotes",
st,
BytesType.instance,
null)
.defaultValidator(UTF8Type.instance),
superCFMD(ks1, "Super1", LongType.instance),
superCFMD(ks1, "Super2", LongType.instance),
superCFMD(ks1, "Super3", LongType.instance),
superCFMD(ks1, "Super4", UTF8Type.instance),
superCFMD(ks1, "Super5", bytes),
superCFMD(ks1, "Super6", LexicalUUIDType.instance, UTF8Type.instance),
indexCFMD(ks1, "Indexed1", true),
indexCFMD(ks1, "Indexed2", false),
superCFMD(ks1, "Super1", LongType.instance, withOldCfIds),
superCFMD(ks1, "Super2", LongType.instance, withOldCfIds),
superCFMD(ks1, "Super3", LongType.instance, withOldCfIds),
superCFMD(ks1, "Super4", UTF8Type.instance, withOldCfIds),
superCFMD(ks1, "Super5", bytes, withOldCfIds),
superCFMD(ks1, "Super6", LexicalUUIDType.instance, UTF8Type.instance, withOldCfIds),
indexCFMD(ks1, "Indexed1", true, withOldCfIds),
indexCFMD(ks1, "Indexed2", false, withOldCfIds),
new CFMetaData(ks1,
"StandardInteger1",
st,
@ -188,12 +196,12 @@ public class SchemaLoader
bytes,
bytes)
.defaultValidator(CounterColumnType.instance),
superCFMD(ks1, "SuperDirectGC", BytesType.instance).gcGraceSeconds(0),
jdbcCFMD(ks1, "JdbcInteger", IntegerType.instance).columnMetadata(integerColumn),
jdbcCFMD(ks1, "JdbcUtf8", UTF8Type.instance).columnMetadata(utf8Column),
jdbcCFMD(ks1, "JdbcLong", LongType.instance),
jdbcCFMD(ks1, "JdbcBytes", bytes),
jdbcCFMD(ks1, "JdbcAscii", AsciiType.instance),
superCFMD(ks1, "SuperDirectGC", BytesType.instance, withOldCfIds).gcGraceSeconds(0),
jdbcCFMD(ks1, "JdbcInteger", IntegerType.instance, withOldCfIds).columnMetadata(integerColumn),
jdbcCFMD(ks1, "JdbcUtf8", UTF8Type.instance, withOldCfIds).columnMetadata(utf8Column),
jdbcCFMD(ks1, "JdbcLong", LongType.instance, withOldCfIds),
jdbcCFMD(ks1, "JdbcBytes", bytes, withOldCfIds),
jdbcCFMD(ks1, "JdbcAscii", AsciiType.instance, withOldCfIds),
new CFMetaData(ks1,
"StandardComposite",
st,
@ -204,7 +212,8 @@ public class SchemaLoader
st,
dynamicComposite,
null),
standardCFMD(ks1, "StandardLeveled").compactionStrategyClass(LeveledCompactionStrategy.class)
standardCFMD(ks1, "StandardLeveled", withOldCfIds)
.compactionStrategyClass(LeveledCompactionStrategy.class)
.compactionStrategyOptions(leveledOptions)));
// Keyspace 2
@ -213,11 +222,11 @@ public class SchemaLoader
opts_rf1,
// Column Families
standardCFMD(ks2, "Standard1"),
standardCFMD(ks2, "Standard3"),
superCFMD(ks2, "Super3", bytes),
superCFMD(ks2, "Super4", TimeUUIDType.instance),
indexCFMD(ks2, "Indexed1", true)));
standardCFMD(ks2, "Standard1", withOldCfIds),
standardCFMD(ks2, "Standard3", withOldCfIds),
superCFMD(ks2, "Super3", bytes, withOldCfIds),
superCFMD(ks2, "Super4", TimeUUIDType.instance, withOldCfIds),
indexCFMD(ks2, "Indexed1", true, withOldCfIds)));
// Keyspace 3
schema.add(KSMetaData.testMetadata(ks3,
@ -225,8 +234,8 @@ public class SchemaLoader
opts_rf5,
// Column Families
standardCFMD(ks3, "Standard1"),
indexCFMD(ks3, "Indexed1", true)));
standardCFMD(ks3, "Standard1", withOldCfIds),
indexCFMD(ks3, "Indexed1", true, withOldCfIds)));
// Keyspace 4
schema.add(KSMetaData.testMetadata(ks4,
@ -234,10 +243,10 @@ public class SchemaLoader
opts_rf3,
// Column Families
standardCFMD(ks4, "Standard1"),
standardCFMD(ks4, "Standard3"),
superCFMD(ks4, "Super3", bytes),
superCFMD(ks4, "Super4", TimeUUIDType.instance),
standardCFMD(ks4, "Standard1", withOldCfIds),
standardCFMD(ks4, "Standard3", withOldCfIds),
superCFMD(ks4, "Super3", bytes, withOldCfIds),
superCFMD(ks4, "Super4", TimeUUIDType.instance, withOldCfIds),
new CFMetaData(ks4,
"Super5",
su,
@ -248,35 +257,35 @@ public class SchemaLoader
schema.add(KSMetaData.testMetadata(ks5,
simple,
opts_rf2,
standardCFMD(ks5, "Standard1"),
standardCFMD(ks5, "Counter1")
standardCFMD(ks5, "Standard1", withOldCfIds),
standardCFMD(ks5, "Counter1", withOldCfIds)
.defaultValidator(CounterColumnType.instance)));
// Keyspace 6
schema.add(KSMetaData.testMetadata(ks6,
simple,
opts_rf1,
indexCFMD(ks6, "Indexed1", true)));
indexCFMD(ks6, "Indexed1", true, withOldCfIds)));
// KeyCacheSpace
schema.add(KSMetaData.testMetadata(ks_kcs,
simple,
opts_rf1,
standardCFMD(ks_kcs, "Standard1"),
standardCFMD(ks_kcs, "Standard2"),
standardCFMD(ks_kcs, "Standard3")));
standardCFMD(ks_kcs, "Standard1", withOldCfIds),
standardCFMD(ks_kcs, "Standard2", withOldCfIds),
standardCFMD(ks_kcs, "Standard3", withOldCfIds)));
// RowCacheSpace
schema.add(KSMetaData.testMetadata(ks_rcs,
simple,
opts_rf1,
standardCFMD(ks_rcs, "CFWithoutCache").caching(CFMetaData.Caching.NONE),
standardCFMD(ks_rcs, "CachedCF").caching(CFMetaData.Caching.ALL)));
standardCFMD(ks_rcs, "CFWithoutCache", withOldCfIds).caching(CFMetaData.Caching.NONE),
standardCFMD(ks_rcs, "CachedCF", withOldCfIds).caching(CFMetaData.Caching.ALL)));
schema.add(KSMetaData.testMetadataNotDurable(ks_nocommit,
simple,
opts_rf1,
standardCFMD(ks_nocommit, "Standard1")));
standardCFMD(ks_nocommit, "Standard1", withOldCfIds)));
if (Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")))
@ -296,21 +305,31 @@ public class SchemaLoader
}
}
private static CFMetaData standardCFMD(String ksName, String cfName)
private static CFMetaData standardCFMD(String ksName, String cfName, boolean withOldCfIds)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, BytesType.instance, null);
CFMetaData cfmd = new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, BytesType.instance, null);
if (withOldCfIds)
Schema.instance.addOldCfIdMapping(oldCfIdGenerator.getAndIncrement(), cfmd.cfId);
return cfmd;
}
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType subcc)
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType subcc, boolean withOldCfIds)
{
return superCFMD(ksName, cfName, BytesType.instance, subcc);
return superCFMD(ksName, cfName, BytesType.instance, subcc, withOldCfIds);
}
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType cc, AbstractType subcc)
private static CFMetaData superCFMD(String ksName, String cfName, AbstractType cc, AbstractType subcc, boolean withOldCfIds)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Super, cc, subcc);
CFMetaData cfmd = new CFMetaData(ksName, cfName, ColumnFamilyType.Super, cc, subcc);
if (withOldCfIds)
Schema.instance.addOldCfIdMapping(oldCfIdGenerator.getAndIncrement(), cfmd.cfId);
return cfmd;
}
private static CFMetaData indexCFMD(String ksName, String cfName, final Boolean withIdxType) throws ConfigurationException
private static CFMetaData indexCFMD(String ksName, String cfName, final Boolean withIdxType, boolean withOldCfIds) throws ConfigurationException
{
return standardCFMD(ksName, cfName)
return standardCFMD(ksName, cfName, withOldCfIds)
.keyValidator(AsciiType.instance)
.columnMetadata(new HashMap<ByteBuffer, ColumnDefinition>()
{{
@ -319,9 +338,14 @@ public class SchemaLoader
put(cName, new ColumnDefinition(cName, LongType.instance, keys, null, withIdxType ? ByteBufferUtil.bytesToHex(cName) : null, null));
}});
}
private static CFMetaData jdbcCFMD(String ksName, String cfName, AbstractType comp)
private static CFMetaData jdbcCFMD(String ksName, String cfName, AbstractType comp, boolean withOldCfIds)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, comp, null).defaultValidator(comp);
CFMetaData cfmd = new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, comp, null).defaultValidator(comp);
if (withOldCfIds)
Schema.instance.addOldCfIdMapping(oldCfIdGenerator.getAndIncrement(), cfmd.cfId);
return cfmd;
}
public static void cleanupAndLeaveDirs() throws IOException

2
test/unit/org/apache/cassandra/Util.java
View File

@ -159,7 +159,7 @@ public class Util
{
IMutation first = rms.get(0);
String tablename = first.getTable();
Integer cfid = first.getColumnFamilyIds().iterator().next();
UUID cfid = first.getColumnFamilyIds().iterator().next();
for (IMutation rm : rms)
rm.apply();

10
test/unit/org/apache/cassandra/cache/CacheProviderTest.java vendored
View File

@ -24,6 +24,8 @@ package org.apache.cassandra.cache;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
@ -121,15 +123,17 @@ public class CacheProviderTest extends SchemaLoader
@Test
public void testKeys()
{
UUID cfId = UUID.randomUUID();
byte[] b1 = {1, 2, 3, 4};
RowCacheKey key1 = new RowCacheKey(123, ByteBuffer.wrap(b1));
RowCacheKey key1 = new RowCacheKey(cfId, ByteBuffer.wrap(b1));
byte[] b2 = {1, 2, 3, 4};
RowCacheKey key2 = new RowCacheKey(123, ByteBuffer.wrap(b2));
RowCacheKey key2 = new RowCacheKey(cfId, ByteBuffer.wrap(b2));
assertEquals(key1, key2);
assertEquals(key1.hashCode(), key2.hashCode());
byte[] b3 = {1, 2, 3, 5};
RowCacheKey key3 = new RowCacheKey(123, ByteBuffer.wrap(b3));
RowCacheKey key3 = new RowCacheKey(cfId, ByteBuffer.wrap(b3));
assertNotSame(key1, key3);
assertNotSame(key1.hashCode(), key3.hashCode());
}

19
test/unit/org/apache/cassandra/config/DefsTest.java
View File

@ -38,8 +38,6 @@ import org.apache.cassandra.io.sstable.SSTableDeletingTask;
import org.apache.cassandra.locator.OldNetworkTopologyStrategy;
import org.apache.cassandra.locator.SimpleStrategy;
import org.apache.cassandra.service.MigrationManager;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -51,11 +49,8 @@ public class DefsTest extends SchemaLoader
@Test
public void ensureStaticCFMIdsAreLessThan1000()
{
assert CFMetaData.StatusCf.cfId == 0;
assert CFMetaData.HintsCf.cfId == 1;
assert CFMetaData.MigrationsCf.cfId == 2;
assert CFMetaData.SchemaCf.cfId == 3;
assert CFMetaData.HostIdCf.cfId == 11;
assert CFMetaData.StatusCf.cfId.equals(CFMetaData.getId(Table.SYSTEM_TABLE, SystemTable.STATUS_CF));
assert CFMetaData.HintsCf.cfId.equals(CFMetaData.getId(Table.SYSTEM_TABLE, HintedHandOffManager.HINTS_CF));
}
@Test
@ -465,7 +460,7 @@ public class DefsTest extends SchemaLoader
assert Schema.instance.getCFMetaData(cf.ksName, cf.cfName).getDefaultValidator() == UTF8Type.instance;
// Change cfId
newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId + 1);
newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator, UUID.randomUUID());
CFMetaData.copyOpts(newCfm, cf);
try
{
@ -475,7 +470,7 @@ public class DefsTest extends SchemaLoader
catch (ConfigurationException expected) {}
// Change cfName
newCfm = new CFMetaData(cf.ksName, cf.cfName + "_renamed", cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId);
newCfm = new CFMetaData(cf.ksName, cf.cfName + "_renamed", cf.cfType, cf.comparator, cf.subcolumnComparator);
CFMetaData.copyOpts(newCfm, cf);
try
{
@ -485,7 +480,7 @@ public class DefsTest extends SchemaLoader
catch (ConfigurationException expected) {}
// Change ksName
newCfm = new CFMetaData(cf.ksName + "_renamed", cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator, cf.cfId);
newCfm = new CFMetaData(cf.ksName + "_renamed", cf.cfName, cf.cfType, cf.comparator, cf.subcolumnComparator);
CFMetaData.copyOpts(newCfm, cf);
try
{
@ -495,7 +490,7 @@ public class DefsTest extends SchemaLoader
catch (ConfigurationException expected) {}
// Change cf type
newCfm = new CFMetaData(cf.ksName, cf.cfName, ColumnFamilyType.Super, cf.comparator, cf.subcolumnComparator, cf.cfId);
newCfm = new CFMetaData(cf.ksName, cf.cfName, ColumnFamilyType.Super, cf.comparator, cf.subcolumnComparator);
CFMetaData.copyOpts(newCfm, cf);
try
{
@ -505,7 +500,7 @@ public class DefsTest extends SchemaLoader
catch (ConfigurationException expected) {}
// Change comparator
newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, TimeUUIDType.instance, cf.subcolumnComparator, cf.cfId);
newCfm = new CFMetaData(cf.ksName, cf.cfName, cf.cfType, TimeUUIDType.instance, cf.subcolumnComparator);
CFMetaData.copyOpts(newCfm, cf);
try
{

7
test/unit/org/apache/cassandra/db/CommitLogTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.db;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.zip.CRC32;
import java.util.zip.Checksum;
@ -111,7 +112,7 @@ public class CommitLogTest extends SchemaLoader
assert CommitLog.instance.activeSegments() == 2 : "Expecting 2 segments, got " + CommitLog.instance.activeSegments();
int cfid2 = rm2.getColumnFamilyIds().iterator().next();
UUID cfid2 = rm2.getColumnFamilyIds().iterator().next();
CommitLog.instance.discardCompletedSegments(cfid2, CommitLog.instance.getContext().get());
// Assert we still have both our segment
@ -133,7 +134,7 @@ public class CommitLogTest extends SchemaLoader
assert CommitLog.instance.activeSegments() == 1 : "Expecting 1 segment, got " + CommitLog.instance.activeSegments();
// "Flush": this won't delete anything
int cfid1 = rm.getColumnFamilyIds().iterator().next();
UUID cfid1 = rm.getColumnFamilyIds().iterator().next();
CommitLog.instance.discardCompletedSegments(cfid1, CommitLog.instance.getContext().get());
assert CommitLog.instance.activeSegments() == 1 : "Expecting 1 segment, got " + CommitLog.instance.activeSegments();
@ -151,7 +152,7 @@ public class CommitLogTest extends SchemaLoader
// "Flush" second cf: The first segment should be deleted since we
// didn't write anything on cf1 since last flush (and we flush cf2)
int cfid2 = rm2.getColumnFamilyIds().iterator().next();
UUID cfid2 = rm2.getColumnFamilyIds().iterator().next();
CommitLog.instance.discardCompletedSegments(cfid2, CommitLog.instance.getContext().get());
// Assert we still have both our segment

15
test/unit/org/apache/cassandra/db/SerializationsTest.java
View File

@ -34,21 +34,24 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.junit.BeforeClass;
import org.junit.Test;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.HashMap;
import java.util.*;
public class SerializationsTest extends AbstractSerializationsTester
{
@BeforeClass
public static void loadSchema() throws IOException
{
loadSchema(true);
}
private void testRangeSliceCommandWrite() throws IOException
{
ByteBuffer startCol = ByteBufferUtil.bytes("Start");
@ -213,7 +216,7 @@ public class SerializationsTest extends AbstractSerializationsTester
standardRm.add(Statics.StandardCf);
RowMutation superRm = new RowMutation(Statics.KS, Statics.Key);
superRm.add(Statics.SuperCf);
Map<Integer, ColumnFamily> mods = new HashMap<Integer, ColumnFamily>();
Map<UUID, ColumnFamily> mods = new HashMap<UUID, ColumnFamily>();
mods.put(Statics.StandardCf.metadata().cfId, Statics.StandardCf);
mods.put(Statics.SuperCf.metadata().cfId, Statics.SuperCf);
RowMutation mixedRm = new RowMutation(Statics.KS, Statics.Key, mods);