You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2009/10/02 21:04:56 UTC
svn commit: r821134 - in /incubator/cassandra/trunk:
src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/db/filter/
src/java/org/apache/cassandra/dht/ src/java/org/apache/cassandra/io/
src/java/org/apache/cassandra/service/ test/unit/org/ap...
Author: jbellis
Date: Fri Oct 2 19:04:54 2009
New Revision: 821134
URL: http://svn.apache.org/viewvc?rev=821134&view=rev
Log:
Use DecoratedKey objects in Memtable, SSTableReader/Writer objects.
patch by johano; reviewed by jbellis for CASSANDRA-446
Added:
incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/RandomPartitionerTest.java
Modified:
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/BinaryMemtable.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DecoratedKey.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableNamesIterator.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableSliceIterator.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BytesToken.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/CollatingOrderPreservingPartitioner.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/IPartitioner.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/RandomPartitioner.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/CompactionIterator.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/IteratingRow.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTable.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableScanner.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableWriter.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/TableTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/CollatingOrderPreservingPartitionerTest.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/SSTableTest.java
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/BinaryMemtable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/BinaryMemtable.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/BinaryMemtable.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/BinaryMemtable.java Fri Oct 2 19:04:54 2009
@@ -118,7 +118,7 @@
private void resolve(String key, byte[] buffer)
{
- columnFamilies_.put(partitioner_.decorateKeyObj(key), buffer);
+ columnFamilies_.put(partitioner_.decorateKey(key), buffer);
currentSize_.addAndGet(buffer.length + key.length());
}
@@ -127,7 +127,7 @@
assert !columnFamilies_.isEmpty();
logger_.info("Sorting " + this);
List<DecoratedKey> keys = new ArrayList<DecoratedKey>(columnFamilies_.keySet());
- Collections.sort(keys, partitioner_.getDecoratedKeyObjComparator());
+ Collections.sort(keys, partitioner_.getDecoratedKeyComparator());
return keys;
}
@@ -142,7 +142,7 @@
{
byte[] bytes = columnFamilies_.get(key);
assert bytes.length > 0;
- writer.append(key.toString(), bytes);
+ writer.append(key, bytes);
}
SSTableReader sstable = writer.closeAndOpenReader();
logger_.info("Completed flushing " + writer.getFilename());
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Fri Oct 2 19:04:54 2009
@@ -33,6 +33,7 @@
import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.io.*;
import org.apache.cassandra.net.EndPoint;
@@ -46,10 +47,12 @@
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.collections.IteratorUtils;
-import org.apache.commons.collections.Predicate;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterators;
+
public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
private static Logger logger_ = Logger.getLogger(ColumnFamilyStore.class);
@@ -106,6 +109,8 @@
private TimedStatsDeque readStats_ = new TimedStatsDeque(60000);
private TimedStatsDeque writeStats_ = new TimedStatsDeque(60000);
+
+ private final IPartitioner partitioner = StorageService.getPartitioner();
ColumnFamilyStore(String table, String columnFamilyName, boolean isSuper, int indexValue) throws IOException
{
@@ -782,7 +787,7 @@
while (ci.hasNext())
{
CompactionIterator.CompactedRow row = ci.next();
- if (Range.isTokenInRanges(StorageService.getPartitioner().getToken(row.key), ranges))
+ if (Range.isTokenInRanges(row.key.token, ranges))
{
if (writer == null)
{
@@ -998,7 +1003,7 @@
}
}
- public Iterator<String> memtableKeyIterator() throws ExecutionException, InterruptedException
+ public Iterator<DecoratedKey> memtableKeyIterator() throws ExecutionException, InterruptedException
{
Table.flusherLock_.readLock().lock();
try
@@ -1161,44 +1166,46 @@
public RangeReply getKeyRange(final String startWith, final String stopAt, int maxResults)
throws IOException, ExecutionException, InterruptedException
{
+ final DecoratedKey startWithDK = partitioner.decorateKey(startWith);
+ final DecoratedKey stopAtDK = partitioner.decorateKey(stopAt);
// (OPP key decoration is a no-op so using the "decorated" comparator against raw keys is fine)
- final Comparator<String> comparator = StorageService.getPartitioner().getDecoratedKeyComparator();
+ final Comparator<DecoratedKey> comparator = partitioner.getDecoratedKeyComparator();
// create a CollatedIterator that will return unique keys from different sources
// (current memtable, historical memtables, and SSTables) in the correct order.
- List<Iterator<String>> iterators = new ArrayList<Iterator<String>>();
+ List<Iterator<DecoratedKey>> iterators = new ArrayList<Iterator<DecoratedKey>>();
// we iterate through memtables with a priority queue to avoid more sorting than necessary.
// this predicate throws out the keys before the start of our range.
- Predicate p = new Predicate()
+ Predicate<DecoratedKey> p = new Predicate<DecoratedKey>()
{
- public boolean evaluate(Object key)
+ public boolean apply(DecoratedKey key)
{
- String st = (String)key;
- return comparator.compare(startWith, st) <= 0 && (stopAt.isEmpty() || comparator.compare(st, stopAt) <= 0);
+ return comparator.compare(startWithDK, key) <= 0
+ && (stopAt.isEmpty() || comparator.compare(key, stopAtDK) <= 0);
}
};
// current memtable keys. have to go through the CFS api for locking.
- iterators.add(IteratorUtils.filteredIterator(memtableKeyIterator(), p));
+ iterators.add(Iterators.filter(memtableKeyIterator(), p));
// historical memtables
for (Memtable memtable : ColumnFamilyStore.getUnflushedMemtables(columnFamily_))
{
- iterators.add(IteratorUtils.filteredIterator(memtable.getKeyIterator(), p));
+ iterators.add(Iterators.filter(memtable.getKeyIterator(), p));
}
// sstables
for (SSTableReader sstable : ssTables_)
{
final SSTableScanner scanner = sstable.getScanner();
- scanner.seekTo(startWith);
- Iterator<String> iter = new Iterator<String>()
+ scanner.seekTo(startWithDK);
+ Iterator<DecoratedKey> iter = new Iterator<DecoratedKey>()
{
public boolean hasNext()
{
return scanner.hasNext();
}
- public String next()
+ public DecoratedKey next()
{
return scanner.next().getKey();
}
@@ -1210,16 +1217,17 @@
iterators.add(iter);
}
- Iterator<String> collated = IteratorUtils.collatedIterator(comparator, iterators);
- Iterable<String> reduced = new ReducingIterator<String, String>(collated) {
- String current;
+ Iterator<DecoratedKey> collated = IteratorUtils.collatedIterator(comparator, iterators);
+
+ Iterable<DecoratedKey> reduced = new ReducingIterator<DecoratedKey, DecoratedKey>(collated) {
+ DecoratedKey current;
- public void reduce(String current)
+ public void reduce(DecoratedKey current)
{
this.current = current;
}
- protected String getReduced()
+ protected DecoratedKey getReduced()
{
return current;
}
@@ -1231,19 +1239,19 @@
// so we set an arbitrary limit on how many we'll do at once.
List<String> keys = new ArrayList<String>();
boolean rangeCompletedLocally = false;
- for (String current : reduced)
+ for (DecoratedKey current : reduced)
{
- if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0)
+ if (!stopAt.isEmpty() && comparator.compare(stopAtDK, current) < 0)
{
rangeCompletedLocally = true;
break;
}
// make sure there is actually non-tombstone content associated w/ this key
// TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable)
- QueryFilter filter = new SliceQueryFilter(current, new QueryPath(columnFamily_), ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, false, 1);
+ QueryFilter filter = new SliceQueryFilter(current.key, new QueryPath(columnFamily_), ArrayUtils.EMPTY_BYTE_ARRAY, ArrayUtils.EMPTY_BYTE_ARRAY, false, 1);
if (getColumnFamily(filter, Integer.MAX_VALUE) != null)
{
- keys.add(current);
+ keys.add(current.key);
}
if (keys.size() >= maxResults)
{
@@ -1311,4 +1319,5 @@
memtable_.clearUnsafe();
ssTables_.clearUnsafe();
}
+
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DecoratedKey.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DecoratedKey.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DecoratedKey.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DecoratedKey.java Fri Oct 2 19:04:54 2009
@@ -19,9 +19,6 @@
package org.apache.cassandra.db;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.RandomPartitioner;
-import org.apache.cassandra.service.StorageService;
/**
* Represents a decorated key, handy for certain operations
@@ -29,8 +26,6 @@
*/
public class DecoratedKey<T extends Token>
{
- public static final String DELIMITER = ":";
-
public final T token;
public final String key;
@@ -76,16 +71,9 @@
return true;
}
- /**
- * Return the on disk format of the decorated key.
- */
+ @Override
public String toString()
{
- // TODO when we can break the disk format, we should use
- // token == null ? key : token.toString() + DELIMITER + key
- // until then we special case like this, which keeps COPP using just the key string
- return StorageService.getPartitioner() instanceof RandomPartitioner
- ? token.toString() + DELIMITER + key
- : key;
+ return "DecoratedKey(" + token + ", " + key + ")";
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java Fri Oct 2 19:04:54 2009
@@ -37,7 +37,7 @@
import org.apache.log4j.Logger;
-public class Memtable implements Comparable<Memtable>, IFlushable<String>
+public class Memtable implements Comparable<Memtable>, IFlushable<DecoratedKey>
{
private static final Logger logger_ = Logger.getLogger( Memtable.class );
@@ -55,8 +55,9 @@
private final String cfName_;
private final long creationTime_;
// we use NBHM with manual locking, so reads are automatically threadsafe but write merging is serialized per key
- private final NonBlockingHashMap<String, ColumnFamily> columnFamilies_ = new NonBlockingHashMap<String, ColumnFamily>();
+ private final NonBlockingHashMap<DecoratedKey, ColumnFamily> columnFamilies_ = new NonBlockingHashMap<DecoratedKey, ColumnFamily>();
private final Object[] keyLocks;
+ private final IPartitioner partitioner_ = StorageService.getPartitioner();
Memtable(String table, String cfName)
{
@@ -146,7 +147,8 @@
private void resolve(String key, ColumnFamily columnFamily)
{
- ColumnFamily oldCf = columnFamilies_.putIfAbsent(key, columnFamily);
+ DecoratedKey decoratedKey = partitioner_.decorateKey(key);
+ ColumnFamily oldCf = columnFamilies_.putIfAbsent(decoratedKey, columnFamily);
if (oldCf == null)
{
currentSize_.addAndGet(columnFamily.size() + key.length());
@@ -180,7 +182,7 @@
{
StringBuilder builder = new StringBuilder();
builder.append("{");
- for (Map.Entry<String, ColumnFamily> entry : columnFamilies_.entrySet())
+ for (Map.Entry<DecoratedKey, ColumnFamily> entry : columnFamilies_.entrySet())
{
builder.append(entry.getKey()).append(": ").append(entry.getValue()).append(", ");
}
@@ -188,24 +190,17 @@
return builder.toString();
}
- public List<String> getSortedKeys()
+ public List<DecoratedKey> getSortedKeys()
{
logger_.info("Sorting " + this);
// sort keys in the order they would be in when decorated
- final IPartitioner<?> partitioner = StorageService.getPartitioner();
- final Comparator<String> dc = partitioner.getDecoratedKeyComparator();
- ArrayList<String> orderedKeys = new ArrayList<String>(columnFamilies_.keySet());
- Collections.sort(orderedKeys, new Comparator<String>()
- {
- public int compare(String o1, String o2)
- {
- return dc.compare(partitioner.decorateKey(o1), partitioner.decorateKey(o2));
- }
- });
+ Comparator<DecoratedKey> dc = partitioner_.getDecoratedKeyComparator();
+ ArrayList<DecoratedKey> orderedKeys = new ArrayList<DecoratedKey>(columnFamilies_.keySet());
+ Collections.sort(orderedKeys, dc);
return orderedKeys;
}
- public SSTableReader writeSortedContents(List<String> sortedKeys) throws IOException
+ public SSTableReader writeSortedContents(List<DecoratedKey> sortedKeys) throws IOException
{
logger_.info("Writing " + this);
IPartitioner<?> partitioner = StorageService.getPartitioner();
@@ -213,14 +208,14 @@
SSTableWriter writer = new SSTableWriter(cfStore.getTempSSTablePath(), columnFamilies_.size(), StorageService.getPartitioner());
DataOutputBuffer buffer = new DataOutputBuffer();
- for (String key : sortedKeys)
+ for (DecoratedKey key : sortedKeys)
{
buffer.reset();
ColumnFamily columnFamily = columnFamilies_.get(key);
/* serialize the cf with column indexes */
ColumnFamily.serializer().serializeWithIndexes(columnFamily, buffer);
/* Now write the key and value to disk */
- writer.append(partitioner.decorateKey(key), buffer);
+ writer.append(key, buffer);
}
buffer.close();
SSTableReader ssTable = writer.closeAndOpenReader();
@@ -234,18 +229,18 @@
return "Memtable(" + cfName_ + ")@" + hashCode();
}
- public Iterator<String> getKeyIterator()
+ public Iterator<DecoratedKey> getKeyIterator()
{
// even though we are using NBHM, it is okay to use size() twice here, since size() will never decrease
// w/in a single memtable's lifetime
if (columnFamilies_.size() == 0)
{
// cannot create a PQ of size zero (wtf?)
- return Arrays.asList(new String[0]).iterator();
+ return Arrays.asList(new DecoratedKey[0]).iterator();
}
- PriorityQueue<String> pq = new PriorityQueue<String>(columnFamilies_.size(), StorageService.getPartitioner().getDecoratedKeyComparator());
+ PriorityQueue<DecoratedKey> pq = new PriorityQueue<DecoratedKey>(columnFamilies_.size(), partitioner_.getDecoratedKeyComparator());
pq.addAll(columnFamilies_.keySet());
- return new DestructivePQIterator<String>(pq);
+ return new DestructivePQIterator<DecoratedKey>(pq);
}
public boolean isClean()
@@ -260,7 +255,7 @@
*/
public ColumnIterator getSliceIterator(SliceQueryFilter filter, AbstractType typeComparator)
{
- ColumnFamily cf = columnFamilies_.get(filter.key);
+ ColumnFamily cf = columnFamilies_.get(partitioner_.decorateKey(filter.key));
final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(table_, filter.getColumnFamilyName()) : cf.cloneMeShallow();
final IColumn columns[] = (cf == null ? columnFamily : cf).getSortedColumns().toArray(new IColumn[columnFamily.getSortedColumns().size()]);
@@ -311,7 +306,7 @@
public ColumnIterator getNamesIterator(final NamesQueryFilter filter)
{
- final ColumnFamily cf = columnFamilies_.get(filter.key);
+ final ColumnFamily cf = columnFamilies_.get(partitioner_.decorateKey(filter.key));
final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(table_, filter.getColumnFamilyName()) : cf.cloneMeShallow();
return new SimpleAbstractColumnIterator()
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableNamesIterator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableNamesIterator.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableNamesIterator.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableNamesIterator.java Fri Oct 2 19:04:54 2009
@@ -25,6 +25,7 @@
import java.util.*;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.io.*;
@@ -42,7 +43,7 @@
assert columnNames != null;
this.columns = columnNames;
- String decoratedKey = ssTable.getPartitioner().decorateKey(key);
+ DecoratedKey decoratedKey = ssTable.getPartitioner().decorateKey(key);
long position = ssTable.getPosition(decoratedKey);
if (position < 0)
return;
@@ -52,7 +53,7 @@
{
file.seek(position);
- String keyInDisk = file.readUTF();
+ DecoratedKey keyInDisk = ssTable.getPartitioner().convertFromDiskFormat(file.readUTF());
assert keyInDisk.equals(decoratedKey) : keyInDisk;
file.readInt(); // data size
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableSliceIterator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableSliceIterator.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableSliceIterator.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SSTableSliceIterator.java Fri Oct 2 19:04:54 2009
@@ -24,6 +24,7 @@
import java.util.*;
import java.io.IOException;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.marshal.AbstractType;
@@ -48,7 +49,7 @@
this.reversed = reversed;
/* Morph key into actual key based on the partition type. */
- String decoratedKey = ssTable.getPartitioner().decorateKey(key);
+ DecoratedKey decoratedKey = ssTable.getPartitioner().decorateKey(key);
long position = ssTable.getPosition(decoratedKey);
this.comparator = ssTable.getColumnComparator();
this.startColumn = startColumn;
@@ -105,12 +106,12 @@
private int curRangeIndex;
private Deque<IColumn> blockColumns = new ArrayDeque<IColumn>();
- public ColumnGroupReader(SSTableReader ssTable, String key, long position) throws IOException
+ public ColumnGroupReader(SSTableReader ssTable, DecoratedKey key, long position) throws IOException
{
this.file = new BufferedRandomAccessFile(ssTable.getFilename(), "r", DatabaseDescriptor.getSlicedReadBufferSizeInKB() * 1024);
file.seek(position);
- String keyInDisk = file.readUTF();
+ DecoratedKey keyInDisk = ssTable.getPartitioner().convertFromDiskFormat(file.readUTF());
assert keyInDisk.equals(key);
file.readInt(); // row size
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BytesToken.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BytesToken.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BytesToken.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/BytesToken.java Fri Oct 2 19:04:54 2009
@@ -18,7 +18,7 @@
*/
package org.apache.cassandra.dht;
-import java.math.BigInteger;
+import java.util.Arrays;
import org.apache.cassandra.utils.FBUtilities;
@@ -40,4 +40,23 @@
{
return FBUtilities.compareByteArrays(token, o.token);
}
+
+ @Override
+ public int hashCode()
+ {
+ final int prime = 31;
+ return prime + Arrays.hashCode(token);
+ }
+
+ @Override
+ public boolean equals(Object obj)
+ {
+ if (this == obj)
+ return true;
+ if (!(obj instanceof BytesToken))
+ return false;
+ BytesToken other = (BytesToken) obj;
+ return Arrays.equals(token, other.token);
+ }
+
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/CollatingOrderPreservingPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/CollatingOrderPreservingPartitioner.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/CollatingOrderPreservingPartitioner.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/CollatingOrderPreservingPartitioner.java Fri Oct 2 19:04:54 2009
@@ -37,53 +37,31 @@
/**
* Comparators for decorated keys.
*/
- private static final Comparator<String> comparator = new Comparator<String>() {
- public int compare(String o1, String o2)
- {
- return collator.compare(o1, o2);
- }
- };
- private static final Comparator<DecoratedKey<BytesToken>> objComparator = new Comparator<DecoratedKey<BytesToken>>() {
+ private static final Comparator<DecoratedKey<BytesToken>> comparator = new Comparator<DecoratedKey<BytesToken>>() {
public int compare(DecoratedKey<BytesToken> o1, DecoratedKey<BytesToken> o2)
{
- return FBUtilities.compareByteArrays(o1.token.token, o2.token.token);
- }
- };
- private static final Comparator<String> reverseComparator = new Comparator<String>() {
- public int compare(String o1, String o2)
- {
- return -comparator.compare(o1, o2);
+ return collator.compare(o1.key, o2.key);
}
};
-
- public String decorateKey(String key)
- {
- return key;
- }
- public DecoratedKey<BytesToken> decorateKeyObj(String key)
+ public DecoratedKey<BytesToken> decorateKey(String key)
{
return new DecoratedKey<BytesToken>(getToken(key), key);
}
- public String undecorateKey(String decoratedKey)
+ public DecoratedKey<BytesToken> convertFromDiskFormat(String key)
{
- return decoratedKey;
- }
-
- public Comparator<String> getDecoratedKeyComparator()
- {
- return comparator;
+ return new DecoratedKey<BytesToken>(getToken(key), key);
}
- public Comparator<DecoratedKey<BytesToken>> getDecoratedKeyObjComparator()
+ public String convertToDiskFormat(DecoratedKey<BytesToken> key)
{
- return objComparator;
+ return key.key;
}
- public Comparator<String> getReverseDecoratedKeyComparator()
+ public Comparator<DecoratedKey<BytesToken>> getDecoratedKeyComparator()
{
- return reverseComparator;
+ return comparator;
}
/**
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/IPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/IPartitioner.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/IPartitioner.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/IPartitioner.java Fri Oct 2 19:04:54 2009
@@ -24,14 +24,21 @@
public interface IPartitioner<T extends Token>
{
+
/**
- * Transform key to on-disk format s.t. keys are stored in node comparison order.
- * This lets bootstrap rip out parts of the sstable sequentially instead of doing random seeks.
- *
- * @param key the raw, client-facing key
- * @return decorated on-disk version of key
+ * Convert the on disk representation to a DecoratedKey object
+ * @param key On disk representation
+ * @return DecoratedKey object
*/
- public String decorateKey(String key);
+ public DecoratedKey<T> convertFromDiskFormat(String key);
+
+ /**
+ * Convert the DecoratedKey to the on disk format used for
+ * this partitioner.
+ * @param key The DecoratedKey in question
+ * @return
+ */
+ public String convertToDiskFormat(DecoratedKey<T> key);
/**
* Transform key to object representation of the on-disk format.
@@ -39,18 +46,12 @@
* @param key the raw, client-facing key
* @return decorated version of key
*/
- public DecoratedKey<T> decorateKeyObj(String key);
-
- public String undecorateKey(String decoratedKey);
+ public DecoratedKey<T> decorateKey(String key);
- public Comparator<String> getDecoratedKeyComparator();
-
/**
* @return a comparator for decorated key objects, not strings
*/
- public Comparator<DecoratedKey<T>> getDecoratedKeyObjComparator();
-
- public Comparator<String> getReverseDecoratedKeyComparator();
+ public Comparator<DecoratedKey<T>> getDecoratedKeyComparator();
/**
* Calculate a Token representing the approximate "middle" of the given
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java Fri Oct 2 19:04:54 2009
@@ -33,54 +33,32 @@
/**
* Comparators for decorated keys.
*/
- private static final Comparator<String> comparator = new Comparator<String>() {
- public int compare(String o1, String o2)
- {
- return o1.compareTo(o2);
- }
- };
- private static final Comparator<DecoratedKey<StringToken>> objComparator =
+ private static final Comparator<DecoratedKey<StringToken>> comparator =
new Comparator<DecoratedKey<StringToken>>() {
public int compare(DecoratedKey<StringToken> o1, DecoratedKey<StringToken> o2)
{
return o1.key.compareTo(o2.key);
}
- };
- private static final Comparator<String> reverseComparator = new Comparator<String>() {
- public int compare(String o1, String o2)
- {
- return o2.compareTo(o1);
- }
};
- public String decorateKey(String key)
- {
- return key;
- }
-
- public DecoratedKey<StringToken> decorateKeyObj(String key)
+ public DecoratedKey<StringToken> decorateKey(String key)
{
return new DecoratedKey<StringToken>(null, key);
}
- public String undecorateKey(String decoratedKey)
+ public DecoratedKey<StringToken> convertFromDiskFormat(String key)
{
- return decoratedKey;
+ return new DecoratedKey<StringToken>(null, key);
}
- public Comparator<String> getDecoratedKeyComparator()
+ public String convertToDiskFormat(DecoratedKey<StringToken> key)
{
- return comparator;
+ return key.key;
}
- public Comparator<DecoratedKey<StringToken>> getDecoratedKeyObjComparator()
+ public Comparator<DecoratedKey<StringToken>> getDecoratedKeyComparator()
{
- return objComparator;
- }
-
- public Comparator<String> getReverseDecoratedKeyComparator()
- {
- return reverseComparator;
+ return comparator;
}
/**
@@ -252,5 +230,4 @@
{
return new StringToken(key);
}
-
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/RandomPartitioner.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/RandomPartitioner.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/RandomPartitioner.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/dht/RandomPartitioner.java Fri Oct 2 19:04:54 2009
@@ -20,7 +20,7 @@
import java.math.BigInteger;
import java.util.Comparator;
-import java.util.StringTokenizer;
+import java.util.regex.Pattern;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
@@ -37,7 +37,12 @@
public static final BigIntegerToken MINIMUM = new BigIntegerToken("0");
- private static final Comparator<DecoratedKey<BigIntegerToken>> objComparator =
+ private static final String DELIMITER = ":";
+
+ //to avoid having to create the pattern on every String.split
+ private Pattern diskDelimiter = Pattern.compile(DELIMITER);
+
+ private static final Comparator<DecoratedKey<BigIntegerToken>> comparator =
new Comparator<DecoratedKey<BigIntegerToken>>() {
public int compare(DecoratedKey<BigIntegerToken> o1, DecoratedKey<BigIntegerToken> o2)
{
@@ -51,63 +56,27 @@
return o1.key.compareTo(o2.key);
}
};
-
- private static final Comparator<String> comparator = new Comparator<String>()
- {
- public int compare(String o1, String o2)
- {
- // StringTokenizer is faster than String.split()
- StringTokenizer st1 = new StringTokenizer(o1, ":");
- StringTokenizer st2 = new StringTokenizer(o2, ":");
-
- // first, compare on the bigint hash "decoration". usually this will be enough.
- BigInteger i1 = new BigInteger(st1.nextToken());
- BigInteger i2 = new BigInteger(st2.nextToken());
- int v = i1.compareTo(i2);
- if (v != 0) {
- return v;
- }
-
- // if the hashes are equal, compare the strings
- return st1.nextToken().compareTo(st2.nextToken());
- }
- };
- private static final Comparator<String> rcomparator = new Comparator<String>()
- {
- public int compare(String o1, String o2)
- {
- return -comparator.compare(o1, o2);
- }
- };
- public String decorateKey(String key)
- {
- return FBUtilities.hash(key).toString() + ":" + key;
- }
-
- public DecoratedKey<BigIntegerToken> decorateKeyObj(String key)
+ public DecoratedKey<BigIntegerToken> decorateKey(String key)
{
return new DecoratedKey<BigIntegerToken>(getToken(key), key);
}
- public String undecorateKey(String decoratedKey)
+ public DecoratedKey<BigIntegerToken> convertFromDiskFormat(String key)
{
- return decoratedKey.split(":", 2)[1];
+ String[] parts = diskDelimiter.split(key, 2);
+ assert parts.length == 2;
+ return new DecoratedKey<BigIntegerToken>(new BigIntegerToken(parts[0]), parts[1]);
}
- public Comparator<String> getDecoratedKeyComparator()
- {
- return comparator;
- }
-
- public Comparator<DecoratedKey<BigIntegerToken>> getDecoratedKeyObjComparator()
+ public String convertToDiskFormat(DecoratedKey<BigIntegerToken> key)
{
- return objComparator;
+ return key.token + DELIMITER + key.key;
}
- public Comparator<String> getReverseDecoratedKeyComparator()
+ public Comparator<DecoratedKey<BigIntegerToken>> getDecoratedKeyComparator()
{
- return rcomparator;
+ return comparator;
}
public BigIntegerToken midpoint(BigIntegerToken ltoken, BigIntegerToken rtoken)
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/CompactionIterator.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/CompactionIterator.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/CompactionIterator.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/CompactionIterator.java Fri Oct 2 19:04:54 2009
@@ -10,6 +10,7 @@
import org.apache.cassandra.utils.ReducingIterator;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.DecoratedKey;
public class CompactionIterator extends ReducingIterator<IteratingRow, CompactionIterator.CompactedRow> implements Closeable
{
@@ -65,7 +66,7 @@
protected CompactedRow getReducedRaw() throws IOException
{
DataOutputBuffer buffer = new DataOutputBuffer();
- String key = rows.get(0).getKey();
+ DecoratedKey key = rows.get(0).getKey();
if (rows.size() > 1)
{
ColumnFamily cf = null;
@@ -101,10 +102,10 @@
public static class CompactedRow
{
- public final String key;
+ public final DecoratedKey key;
public final DataOutputBuffer buffer;
- public CompactedRow(String key, DataOutputBuffer buffer)
+ public CompactedRow(DecoratedKey key, DataOutputBuffer buffer)
{
this.key = key;
this.buffer = buffer;
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/IteratingRow.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/IteratingRow.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/IteratingRow.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/IteratingRow.java Fri Oct 2 19:04:54 2009
@@ -24,30 +24,34 @@
import java.io.*;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.service.StorageService;
import com.google.common.collect.AbstractIterator;
public class IteratingRow extends AbstractIterator<IColumn> implements Comparable<IteratingRow>
{
- private final String key;
+ private final DecoratedKey key;
private final long finishedAt;
private final BufferedRandomAccessFile file;
private SSTableReader sstable;
private long dataStart;
+ private final IPartitioner partitioner;
public IteratingRow(BufferedRandomAccessFile file, SSTableReader sstable) throws IOException
{
this.file = file;
this.sstable = sstable;
+ this.partitioner = StorageService.getPartitioner();
- key = file.readUTF();
+ key = partitioner.convertFromDiskFormat(file.readUTF());
int dataSize = file.readInt();
dataStart = file.getFilePointer();
finishedAt = dataStart + dataSize;
}
- public String getKey()
+ public DecoratedKey getKey()
{
return key;
}
@@ -100,6 +104,6 @@
public int compareTo(IteratingRow o)
{
- return StorageService.getPartitioner().getDecoratedKeyComparator().compare(key, o.key);
+ return partitioner.getDecoratedKeyComparator().compare(key, o.key);
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTable.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTable.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTable.java Fri Oct 2 19:04:54 2009
@@ -31,8 +31,7 @@
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
-import org.apache.cassandra.db.ColumnFamily;
-import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.DecoratedKey;
/**
* This class is built on top of the SequenceFile. It stores
@@ -152,15 +151,13 @@
* This is a simple container for the index Key and its corresponding position
* in the data file. Binary search is performed on a list of these objects
* to lookup keys within the SSTable data file.
- *
- * All keys are decorated.
*/
class KeyPosition implements Comparable<KeyPosition>
{
- public final String key; // decorated
+ public final DecoratedKey key;
public final long position;
- public KeyPosition(String key, long position)
+ public KeyPosition(DecoratedKey key, long position)
{
this.key = key;
this.position = position;
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableReader.java Fri Oct 2 19:04:54 2009
@@ -45,7 +45,7 @@
private static final Logger logger = Logger.getLogger(SSTableReader.class);
private static final FileSSTableMap openedFiles = new FileSSTableMap();
-
+
// `finalizers` is required to keep the PhantomReferences alive after the enclosing SSTR is itself
// unreferenced. otherwise they will never get enqueued.
private static final Set<Reference<SSTableReader>> finalizers = new HashSet<Reference<SSTableReader>>();
@@ -101,25 +101,6 @@
return count;
}
- /**
- * Get all indexed keys in the SSTable.
- */
- public static List<String> getIndexedKeys()
- {
- List<String> indexedKeys = new ArrayList<String>();
-
- for (SSTableReader sstable : openedFiles.values())
- {
- for (KeyPosition kp : sstable.getIndexPositions())
- {
- indexedKeys.add(kp.key);
- }
- }
- Collections.sort(indexedKeys);
-
- return indexedKeys;
- }
-
public static SSTableReader open(String dataFileName) throws IOException
{
return open(dataFileName, StorageService.getPartitioner());
@@ -182,7 +163,7 @@
{
break;
}
- String decoratedKey = input.readUTF();
+ DecoratedKey decoratedKey = partitioner.convertFromDiskFormat(input.readUTF());
input.readLong();
if (i++ % INDEX_INTERVAL == 0)
{
@@ -192,7 +173,7 @@
}
/** get the position in the index file to start scanning to find the given key (at most indexInterval keys away) */
- private long getIndexScanPosition(String decoratedKey, IPartitioner partitioner)
+ private long getIndexScanPosition(DecoratedKey decoratedKey, IPartitioner partitioner)
{
assert indexPositions != null && indexPositions.size() > 0;
int index = Collections.binarySearch(indexPositions, new KeyPosition(decoratedKey, -1));
@@ -214,9 +195,9 @@
/**
* returns the position in the data file to find the given key, or -1 if the key is not present
*/
- public long getPosition(String decoratedKey) throws IOException
+ public long getPosition(DecoratedKey decoratedKey) throws IOException
{
- if (!bf.isPresent(decoratedKey))
+ if (!bf.isPresent(partitioner.convertToDiskFormat(decoratedKey)))
return -1;
long start = getIndexScanPosition(decoratedKey, partitioner);
if (start < 0)
@@ -232,10 +213,10 @@
{
do
{
- String indexDecoratedKey;
+ DecoratedKey indexDecoratedKey;
try
{
- indexDecoratedKey = input.readUTF();
+ indexDecoratedKey = partitioner.convertFromDiskFormat(input.readUTF());
}
catch (EOFException e)
{
@@ -259,7 +240,7 @@
}
/** like getPosition, but if key is not found will return the location of the first key _greater_ than the desired one, or -1 if no such key exists. */
- public long getNearestPosition(String decoratedKey) throws IOException
+ public long getNearestPosition(DecoratedKey decoratedKey) throws IOException
{
long start = getIndexScanPosition(decoratedKey, partitioner);
if (start < 0)
@@ -272,10 +253,10 @@
{
while (true)
{
- String indexDecoratedKey;
+ DecoratedKey indexDecoratedKey;
try
{
- indexDecoratedKey = input.readUTF();
+ indexDecoratedKey = partitioner.convertFromDiskFormat(input.readUTF());
}
catch (EOFException e)
{
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableScanner.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableScanner.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableScanner.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableScanner.java Fri Oct 2 19:04:54 2009
@@ -23,6 +23,7 @@
import java.util.Iterator;
import java.util.Arrays;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.log4j.Logger;
import com.google.common.collect.AbstractIterator;
@@ -52,7 +53,7 @@
file.close();
}
- public void seekTo(String seekKey)
+ public void seekTo(DecoratedKey seekKey)
{
try
{
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableWriter.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableWriter.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableWriter.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/io/SSTableWriter.java Fri Oct 2 19:04:54 2009
@@ -30,6 +30,7 @@
import org.apache.log4j.Logger;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BloomFilter;
@@ -42,7 +43,7 @@
private long keysWritten;
private BufferedRandomAccessFile dataFile;
private BufferedRandomAccessFile indexFile;
- private String lastWrittenKey;
+ private DecoratedKey lastWrittenKey;
private BloomFilter bf;
public SSTableWriter(String filename, int keyCount, IPartitioner partitioner) throws IOException
@@ -53,13 +54,13 @@
bf = new BloomFilter(keyCount, 15);
}
- private long beforeAppend(String decoratedKey) throws IOException
+ private long beforeAppend(DecoratedKey decoratedKey) throws IOException
{
if (decoratedKey == null)
{
throw new IOException("Keys must not be null.");
}
- Comparator<String> c = partitioner.getDecoratedKeyComparator();
+ Comparator<DecoratedKey> c = partitioner.getDecoratedKeyComparator();
if (lastWrittenKey != null && c.compare(lastWrittenKey, decoratedKey) > 0)
{
logger.info("Last written key : " + lastWrittenKey);
@@ -70,12 +71,13 @@
return (lastWrittenKey == null) ? 0 : dataFile.getFilePointer();
}
- private void afterAppend(String decoratedKey, long position) throws IOException
+ private void afterAppend(DecoratedKey decoratedKey, long position) throws IOException
{
- bf.add(decoratedKey);
+ String diskKey = partitioner.convertToDiskFormat(decoratedKey);
+ bf.add(diskKey);
lastWrittenKey = decoratedKey;
long indexPosition = indexFile.getFilePointer();
- indexFile.writeUTF(decoratedKey);
+ indexFile.writeUTF(diskKey);
indexFile.writeLong(position);
if (logger.isTraceEnabled())
logger.trace("wrote " + decoratedKey + " at " + position);
@@ -92,20 +94,20 @@
}
// TODO make this take a DataOutputStream and wrap the byte[] version to combine them
- public void append(String decoratedKey, DataOutputBuffer buffer) throws IOException
+ public void append(DecoratedKey decoratedKey, DataOutputBuffer buffer) throws IOException
{
long currentPosition = beforeAppend(decoratedKey);
- dataFile.writeUTF(decoratedKey);
+ dataFile.writeUTF(partitioner.convertToDiskFormat(decoratedKey));
int length = buffer.getLength();
dataFile.writeInt(length);
dataFile.write(buffer.getData(), 0, length);
afterAppend(decoratedKey, currentPosition);
}
- public void append(String decoratedKey, byte[] value) throws IOException
+ public void append(DecoratedKey decoratedKey, byte[] value) throws IOException
{
long currentPosition = beforeAppend(decoratedKey);
- dataFile.writeUTF(decoratedKey);
+ dataFile.writeUTF(partitioner.convertToDiskFormat(decoratedKey));
dataFile.writeInt(value.length);
dataFile.write(value);
afterAppend(decoratedKey, currentPosition);
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Fri Oct 2 19:04:54 2009
@@ -34,6 +34,7 @@
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.TimedStatsDeque;
import org.apache.cassandra.locator.TokenMetadata;
+import org.apache.cassandra.dht.IPartitioner;
import org.apache.log4j.Logger;
@@ -426,24 +427,6 @@
return rows;
}
- private static Map<String, Message[]> constructReplicaMessages(Map<String, ReadCommand[]> readMessages) throws IOException
- {
- Map<String, Message[]> messages = new HashMap<String, Message[]>();
- Set<String> keys = readMessages.keySet();
-
- for ( String key : keys )
- {
- Message[] msg = new Message[DatabaseDescriptor.getReplicationFactor()];
- ReadCommand[] readParameters = readMessages.get(key);
- msg[0] = readParameters[0].makeReadMessage();
- for ( int i = 1; i < msg.length; ++i )
- {
- msg[i] = readParameters[1].makeReadMessage();
- }
- }
- return messages;
- }
-
/*
* This function executes the read protocol locally and should be used only if consistency is not a concern.
* Read the data from the local disk and return if the row is NOT NULL. If the data is NULL do the read from
@@ -482,7 +465,6 @@
static List<String> getKeyRange(RangeCommand rawCommand) throws IOException, UnavailableException
{
long startTime = System.currentTimeMillis();
- Comparator<String> comparator = StorageService.getPartitioner().getDecoratedKeyComparator();
TokenMetadata tokenMetadata = StorageService.instance().getTokenMetadata();
List<String> allKeys = new ArrayList<String>();
RangeCommand command = rawCommand;
@@ -516,6 +498,15 @@
{
if (allKeys.size() > 0)
{
+ Comparator<String> comparator = new Comparator<String>()
+ {
+ public int compare(String o1, String o2)
+ {
+ IPartitioner p = StorageService.getPartitioner();
+ return p.getDecoratedKeyComparator().compare(p.decorateKey(o1), p.decorateKey(o2));
+ }
+ };
+
if (comparator.compare(rangeKeys.get(rangeKeys.size() - 1), allKeys.get(0)) <= 0)
{
// unlikely, but possible
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/TableTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/TableTest.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/TableTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/TableTest.java Fri Oct 2 19:04:54 2009
@@ -307,7 +307,8 @@
cfStore.doCompaction(2, cfStore.getSSTables().size());
}
SSTableReader sstable = cfStore.getSSTables().iterator().next();
- long position = sstable.getPosition(key);
+ DecoratedKey decKey = sstable.getPartitioner().decorateKey(key);
+ long position = sstable.getPosition(decKey);
BufferedRandomAccessFile file = new BufferedRandomAccessFile(sstable.getFilename(), "r");
file.seek(position);
assert file.readUTF().equals(key);
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/CollatingOrderPreservingPartitionerTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/CollatingOrderPreservingPartitionerTest.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/CollatingOrderPreservingPartitionerTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/CollatingOrderPreservingPartitionerTest.java Fri Oct 2 19:04:54 2009
@@ -18,9 +18,6 @@
*/
package org.apache.cassandra.dht;
-import java.math.BigInteger;
-
-import org.junit.Before;
import org.junit.Test;
import org.apache.cassandra.utils.FBUtilities;
@@ -58,7 +55,7 @@
@Test
public void testTokenFactoryStringsNonUTF()
{
- Token.TokenFactory factory = this.part.getTokenFactory();
+ Token.TokenFactory factory = this.partitioner.getTokenFactory();
BytesToken tok = new BytesToken((byte)0xFF, (byte)0xFF);
assert tok.compareTo(factory.fromString(factory.toString(tok))) == 0;
}
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/PartitionerTestCase.java Fri Oct 2 19:04:54 2009
@@ -18,15 +18,15 @@
*/
package org.apache.cassandra.dht;
-import java.math.BigInteger;
+import static org.junit.Assert.assertEquals;
import org.junit.Before;
import org.junit.Test;
-import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.db.DecoratedKey;
public abstract class PartitionerTestCase<T extends Token> {
- protected IPartitioner<T> part;
+ protected IPartitioner<T> partitioner;
public abstract IPartitioner<T> getPartitioner();
public abstract T tok(String string);
@@ -35,7 +35,7 @@
@Before
public void clean()
{
- this.part = this.getPartitioner();
+ this.partitioner = this.getPartitioner();
}
@Test
@@ -52,7 +52,7 @@
public void assertMidpoint(T left, T right, int depth)
{
- T mid = this.part.midpoint(left, right);
+ T mid = this.partitioner.midpoint(left, right);
assert new Range(left, right).contains(mid)
: "For " + tos(left) + "," + tos(right) + ": range did not contain mid:" + tos(mid);
if (depth > 0)
@@ -83,16 +83,25 @@
}
@Test
+ public void testDiskFormat()
+ {
+ String key = "key";
+ DecoratedKey<T> decKey = partitioner.decorateKey(key);
+ DecoratedKey<T> result = partitioner.convertFromDiskFormat(partitioner.convertToDiskFormat(decKey));
+ assertEquals(decKey, result);
+ }
+
+ @Test
public void testTokenFactoryBytes()
{
- Token.TokenFactory factory = this.part.getTokenFactory();
+ Token.TokenFactory factory = this.partitioner.getTokenFactory();
assert tok("a").compareTo(factory.fromByteArray(factory.toByteArray(tok("a")))) == 0;
}
@Test
public void testTokenFactoryStrings()
{
- Token.TokenFactory factory = this.part.getTokenFactory();
+ Token.TokenFactory factory = this.partitioner.getTokenFactory();
assert tok("a").compareTo(factory.fromString(factory.toString(tok("a")))) == 0;
}
}
Added: incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/RandomPartitionerTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/RandomPartitionerTest.java?rev=821134&view=auto
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/RandomPartitionerTest.java (added)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/dht/RandomPartitionerTest.java Fri Oct 2 19:04:54 2009
@@ -0,0 +1,20 @@
+package org.apache.cassandra.dht;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.cassandra.db.DecoratedKey;
+import org.junit.Test;
+
+public class RandomPartitionerTest
+{
+
+ @Test
+ public void testDiskFormat()
+ {
+ RandomPartitioner part = new RandomPartitioner();
+ String key = "key";
+ DecoratedKey decKey = part.decorateKey(key);
+ DecoratedKey result = part.convertFromDiskFormat(part.convertToDiskFormat(decKey));
+ assertEquals(decKey, result);
+ }
+}
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/SSTableTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/SSTableTest.java?rev=821134&r1=821133&r2=821134&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/SSTableTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/io/SSTableTest.java Fri Oct 2 19:04:54 2009
@@ -25,6 +25,7 @@
import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
+import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.dht.OrderPreservingPartitioner;
public class SSTableTest extends CleanupHelper
@@ -40,7 +41,7 @@
random.nextBytes(bytes);
String key = Integer.toString(1);
- writer.append(key, bytes);
+ writer.append(writer.partitioner.decorateKey(key), bytes);
SSTableReader ssTable = writer.closeAndOpenReader();
// verify
@@ -57,7 +58,7 @@
private void verifySingle(SSTableReader sstable, byte[] bytes, String key) throws IOException
{
BufferedRandomAccessFile file = new BufferedRandomAccessFile(sstable.path, "r");
- file.seek(sstable.getPosition(key));
+ file.seek(sstable.getPosition(sstable.partitioner.decorateKey(key)));
assert key.equals(file.readUTF());
int size = file.readInt();
byte[] bytes2 = new byte[size];
@@ -79,7 +80,7 @@
SSTableWriter writer = new SSTableWriter(f.getAbsolutePath(), 1000, new OrderPreservingPartitioner());
for (String key: map.navigableKeySet())
{
- writer.append(key, map.get(key));
+ writer.append(writer.partitioner.decorateKey(key), map.get(key));
}
SSTableReader ssTable = writer.closeAndOpenReader();
@@ -96,7 +97,7 @@
BufferedRandomAccessFile file = new BufferedRandomAccessFile(sstable.path, "r");
for (String key : keys)
{
- file.seek(sstable.getPosition(key));
+ file.seek(sstable.getPosition(sstable.partitioner.decorateKey(key)));
assert key.equals(file.readUTF());
int size = file.readInt();
byte[] bytes2 = new byte[size];