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 2010/10/22 05:23:31 UTC
svn commit: r1026200 [8/11] - in /cassandra/trunk: ./ interface/
interface/thrift/gen-java/org/apache/cassandra/thrift/
src/java/org/apache/cassandra/auth/ src/java/org/apache/cassandra/avro/
src/java/org/apache/cassandra/cli/ src/java/org/apache/cassa...
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java Fri Oct 22 03:23:26 2010
@@ -19,20 +19,32 @@
package org.apache.cassandra.io.sstable;
-import java.io.*;
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOError;
+import java.io.IOException;
import java.lang.ref.Reference;
import java.lang.ref.ReferenceQueue;
-import java.util.*;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Collections2;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
import org.apache.cassandra.cache.InstrumentedCache;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.Column;
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnFamilyType;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.db.SuperColumn;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.dht.AbstractBounds;
@@ -47,6 +59,11 @@ import org.apache.cassandra.utils.BloomF
import org.apache.cassandra.utils.EstimatedHistogram;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
+import com.google.common.collect.Collections2;
/**
* SSTableReaders are open()ed by Table.onStart; after that they are created by SSTableWriter.renameAndOpen.
@@ -278,9 +295,9 @@ public class SSTableReader extends SSTab
break;
boolean shouldAddEntry = indexSummary.shouldAddEntry();
- byte[] key = (shouldAddEntry || cacheLoading || recreatebloom)
+ ByteBuffer key = (ByteBuffer) ((shouldAddEntry || cacheLoading || recreatebloom)
? FBUtilities.readShortByteArray(input)
- : FBUtilities.skipShortByteArray(input);
+ : FBUtilities.skipShortByteArray(input));
long dataPosition = input.readLong();
if (key != null)
{
@@ -583,7 +600,7 @@ public class SSTableReader extends SSTab
/**
* Conditionally use the deprecated 'IPartitioner.convertFromDiskFormat' method.
*/
- public static DecoratedKey decodeKey(IPartitioner p, Descriptor d, byte[] bytes)
+ public static DecoratedKey decodeKey(IPartitioner p, Descriptor d, ByteBuffer bytes)
{
if (d.hasEncodedKeys)
return p.convertFromDiskFormat(bytes);
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java Fri Oct 22 03:23:26 2010
@@ -19,19 +19,29 @@
package org.apache.cassandra.io.sstable;
-import java.io.*;
-import java.util.*;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileDescriptor;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;
-import com.google.common.base.Function;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.cache.JMXInstrumentedCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Function;
public class SSTableTracker implements Iterable<SSTableReader>
{
@@ -58,7 +68,7 @@ public class SSTableTracker implements I
protected class CacheWriter<K, V>
{
- public void saveCache(JMXInstrumentedCache<K, V> cache, File savedCachePath, Function<K, byte[]> converter) throws IOException
+ public void saveCache(JMXInstrumentedCache<K, V> cache, File savedCachePath, Function<K, ByteBuffer> converter) throws IOException
{
long start = System.currentTimeMillis();
String msgSuffix = " " + savedCachePath.getName() + " for " + cfname + " of " + ksname;
@@ -70,9 +80,9 @@ public class SSTableTracker implements I
FileDescriptor fd = fout.getFD();
for (K key : cache.getKeySet())
{
- byte[] bytes = converter.apply(key);
- out.writeInt(bytes.length);
- out.write(bytes);
+ ByteBuffer bytes = converter.apply(key);
+ out.writeInt(bytes.remaining());
+ out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining());
++count;
}
out.flush();
@@ -87,9 +97,9 @@ public class SSTableTracker implements I
public void saveKeyCache() throws IOException
{
- Function<Pair<Descriptor, DecoratedKey>, byte[]> function = new Function<Pair<Descriptor, DecoratedKey>, byte[]>()
+ Function<Pair<Descriptor, DecoratedKey>, ByteBuffer> function = new Function<Pair<Descriptor, DecoratedKey>, ByteBuffer>()
{
- public byte[] apply(Pair<Descriptor, DecoratedKey> key)
+ public ByteBuffer apply(Pair<Descriptor, DecoratedKey> key)
{
return key.right.key;
}
@@ -100,9 +110,9 @@ public class SSTableTracker implements I
public void saveRowCache() throws IOException
{
- Function<DecoratedKey, byte[]> function = new Function<DecoratedKey, byte[]>()
+ Function<DecoratedKey, ByteBuffer> function = new Function<DecoratedKey, ByteBuffer>()
{
- public byte[] apply(DecoratedKey key)
+ public ByteBuffer apply(DecoratedKey key)
{
return key.key;
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java Fri Oct 22 03:23:26 2010
@@ -19,26 +19,33 @@
package org.apache.cassandra.io.sstable;
-import java.io.*;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOError;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.HashSet;
import java.util.Set;
-import org.apache.cassandra.io.util.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Table;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.AbstractCompactedRow;
import org.apache.cassandra.io.ICompactionInfo;
import org.apache.cassandra.io.util.BufferedRandomAccessFile;
+import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.SegmentedFile;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.BloomFilter;
-import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.EstimatedHistogram;
+import org.apache.cassandra.utils.FBUtilities;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class SSTableWriter extends SSTable
{
@@ -148,13 +155,13 @@ public class SSTableWriter extends SSTab
estimatedColumnCount.add(columnCount);
}
- public void append(DecoratedKey decoratedKey, byte[] value) throws IOException
+ public void append(DecoratedKey decoratedKey, ByteBuffer value) throws IOException
{
long currentPosition = beforeAppend(decoratedKey);
FBUtilities.writeShortByteArray(decoratedKey.key, dataFile);
- assert value.length > 0;
- dataFile.writeLong(value.length);
- dataFile.write(value);
+ assert value.remaining() > 0;
+ dataFile.writeLong(value.remaining());
+ dataFile.write(value.array(),value.position()+value.arrayOffset(),value.remaining());
afterAppend(decoratedKey, currentPosition);
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/ConsistencyChecker.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/ConsistencyChecker.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/ConsistencyChecker.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/ConsistencyChecker.java Fri Oct 22 03:23:26 2010
@@ -22,6 +22,7 @@ import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
@@ -112,9 +113,9 @@ class ConsistencyChecker implements Runn
byte[] body = response.getMessageBody();
ByteArrayInputStream bufIn = new ByteArrayInputStream(body);
ReadResponse result = ReadResponse.serializer().deserialize(new DataInputStream(bufIn));
- byte[] digest = result.digest();
+ ByteBuffer digest = result.digest();
- if (!Arrays.equals(ColumnFamily.digest(row_.cf), digest))
+ if (!ColumnFamily.digest(row_.cf).equals(digest))
{
IResponseResolver<Row> readResponseResolver = new ReadResponseResolver(table_);
IAsyncCallback responseHandler;
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/MigrationManager.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/MigrationManager.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/MigrationManager.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/MigrationManager.java Fri Oct 22 03:23:26 2010
@@ -42,6 +42,7 @@ import java.io.DataOutputStream;
import java.io.IOError;
import java.io.IOException;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -192,10 +193,10 @@ public class MigrationManager implements
for (IColumn col : migrations)
{
assert col instanceof Column;
- dout.writeInt(col.name().length);
- dout.write(col.name());
- dout.writeInt(col.value().length);
- dout.write(col.value());
+ dout.writeInt(col.name().remaining());
+ dout.write(col.name().array(),col.name().position()+col.name().arrayOffset(),col.name().remaining());
+ dout.writeInt(col.value().remaining());
+ dout.write(col.value().array(),col.value().position()+col.value().arrayOffset(),col.value().remaining());
}
dout.close();
byte[] body = bout.toByteArray();
@@ -214,7 +215,7 @@ public class MigrationManager implements
in.readFully(name);
byte[] value = new byte[in.readInt()];
in.readFully(value);
- cols.add(new Column(name, value));
+ cols.add(new Column(ByteBuffer.wrap(name), ByteBuffer.wrap(value)));
}
in.close();
return cols;
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/ReadResponseResolver.java Fri Oct 22 03:23:26 2010
@@ -26,6 +26,8 @@ import java.util.*;
import org.apache.cassandra.db.*;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
+
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.FBUtilities;
@@ -65,7 +67,7 @@ public class ReadResponseResolver implem
List<ColumnFamily> versions = new ArrayList<ColumnFamily>(responses.size());
List<InetAddress> endpoints = new ArrayList<InetAddress>(responses.size());
DecoratedKey key = null;
- byte[] digest = new byte[0];
+ ByteBuffer digest = FBUtilities.EMPTY_BYTE_BUFFER;
boolean isDigestQuery = false;
/*
@@ -96,9 +98,10 @@ public class ReadResponseResolver implem
// If there is a mismatch then throw an exception so that read repair can happen.
if (isDigestQuery)
{
+
for (ColumnFamily cf : versions)
{
- if (!Arrays.equals(ColumnFamily.digest(cf), digest))
+ if (!ColumnFamily.digest(cf).equals(digest))
{
/* Wrap the key as the context in this exception */
String s = String.format("Mismatch for key %s (%s vs %s)", key, FBUtilities.bytesToHex(ColumnFamily.digest(cf)), FBUtilities.bytesToHex(digest));
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Fri Oct 22 03:23:26 2010
@@ -21,6 +21,7 @@ package org.apache.cassandra.service;
import java.io.*;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.*;
import javax.management.MBeanServer;
@@ -187,7 +188,7 @@ public class StorageProxy implements Sto
{
dos.write(previousHints);
}
- FBUtilities.writeShortByteArray(target.getHostAddress().getBytes(UTF_8), dos);
+ FBUtilities.writeShortByteArray(ByteBuffer.wrap(target.getHostAddress().getBytes(UTF_8)), dos);
message.setHeader(RowMutation.HINT, bos.toByteArray());
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Fri Oct 22 03:23:26 2010
@@ -23,27 +23,74 @@ import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
-import java.util.*;
-import java.util.concurrent.*;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
import javax.management.MBeanServer;
import javax.management.ObjectName;
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.Multimap;
-import org.apache.commons.lang.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.concurrent.*;
-import org.apache.cassandra.config.*;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
+import org.apache.cassandra.concurrent.RetryingScheduledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.Stage;
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
+import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.config.RawColumnDefinition;
+import org.apache.cassandra.config.RawColumnFamily;
+import org.apache.cassandra.config.RawKeyspace;
+import org.apache.cassandra.db.BinaryVerbHandler;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.DefinitionsAnnounceVerbHandler;
+import org.apache.cassandra.db.DefinitionsUpdateResponseVerbHandler;
+import org.apache.cassandra.db.DefsTable;
+import org.apache.cassandra.db.HintedHandOffManager;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadRepairVerbHandler;
+import org.apache.cassandra.db.ReadVerbHandler;
+import org.apache.cassandra.db.Row;
+import org.apache.cassandra.db.RowMutationVerbHandler;
+import org.apache.cassandra.db.SchemaCheckVerbHandler;
+import org.apache.cassandra.db.SystemTable;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.db.TruncateVerbHandler;
import org.apache.cassandra.db.migration.AddKeyspace;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.dht.BootStrapper;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.gms.*;
+import org.apache.cassandra.gms.ApplicationState;
+import org.apache.cassandra.gms.EndpointState;
+import org.apache.cassandra.gms.FailureDetector;
+import org.apache.cassandra.gms.GossipDigestAck2VerbHandler;
+import org.apache.cassandra.gms.GossipDigestAckVerbHandler;
+import org.apache.cassandra.gms.GossipDigestSynVerbHandler;
+import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.gms.IEndpointStateChangeSubscriber;
+import org.apache.cassandra.gms.IFailureDetector;
+import org.apache.cassandra.gms.VersionedValue;
import org.apache.cassandra.io.DeletionService;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
@@ -54,18 +101,29 @@ import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.ResponseVerbHandler;
import org.apache.cassandra.service.AntiEntropyService.TreeRequestVerbHandler;
-import org.apache.cassandra.streaming.*;
+import org.apache.cassandra.streaming.ReplicationFinishedVerbHandler;
+import org.apache.cassandra.streaming.StreamIn;
+import org.apache.cassandra.streaming.StreamOut;
+import org.apache.cassandra.streaming.StreamReplyVerbHandler;
+import org.apache.cassandra.streaming.StreamRequestVerbHandler;
+import org.apache.cassandra.streaming.StreamingService;
import org.apache.cassandra.thrift.Constants;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.SkipNullRepresenter;
import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Level;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Dumper;
import org.yaml.snakeyaml.DumperOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.nodes.Tag;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
/*
* This abstraction contains the token/identifier of this node
* on the identifier space. This token gets gossiped around.
@@ -1392,7 +1450,7 @@ public class StorageService implements I
* @param key - key for which we need to find the endpoint return value -
* the endpoint responsible for this key
*/
- public List<InetAddress> getNaturalEndpoints(String table, byte[] key)
+ public List<InetAddress> getNaturalEndpoints(String table, ByteBuffer key)
{
return getNaturalEndpoints(table, partitioner_.getToken(key));
}
@@ -1416,7 +1474,7 @@ public class StorageService implements I
* @param key - key for which we need to find the endpoint return value -
* the endpoint responsible for this key
*/
- public List<InetAddress> getLiveNaturalEndpoints(String table, byte[] key)
+ public List<InetAddress> getLiveNaturalEndpoints(String table, ByteBuffer key)
{
return getLiveNaturalEndpoints(table, partitioner_.getToken(key));
}
@@ -1438,7 +1496,7 @@ public class StorageService implements I
/**
* This function finds the closest live endpoint that contains a given key.
*/
- public InetAddress findSuitableEndpoint(String table, byte[] key) throws IOException, UnavailableException
+ public InetAddress findSuitableEndpoint(String table, ByteBuffer key) throws IOException, UnavailableException
{
List<InetAddress> endpoints = getNaturalEndpoints(table, key);
DatabaseDescriptor.getEndpointSnitch().sortByProximity(FBUtilities.getLocalAddress(), endpoints);
@@ -1983,7 +2041,7 @@ public class StorageService implements I
RawColumnDefinition rcd = new RawColumnDefinition();
rcd.index_name = cd.index_name;
rcd.index_type = cd.index_type;
- rcd.name = new String(cd.name, "UTF8");
+ rcd.name = new String(cd.name.array(),cd.name.position()+cd.name.arrayOffset(),cd.name.remaining(), "UTF8");
rcd.validator_class = cd.validator.getClass().getName();
rcf.column_metadata[j++] = rcd;
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java Fri Oct 22 03:23:26 2010
@@ -21,6 +21,7 @@ package org.apache.cassandra.service;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -126,7 +127,7 @@ public interface StorageServiceMBean
* @param key - key for which we need to find the endpoint return value -
* the endpoint responsible for this key
*/
- public List<InetAddress> getNaturalEndpoints(String table, byte[] key);
+ public List<InetAddress> getNaturalEndpoints(String table, ByteBuffer key);
/**
* Forces major compaction (all sstable files compacted)
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java Fri Oct 22 03:23:26 2010
@@ -19,48 +19,65 @@
package org.apache.cassandra.thrift;
import java.io.IOException;
-import java.util.*;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeoutException;
-import org.apache.cassandra.concurrent.Stage;
-import org.apache.cassandra.db.migration.Migration;
-import org.apache.cassandra.db.migration.UpdateColumnFamily;
-import org.apache.cassandra.db.migration.UpdateKeyspace;
-import org.apache.cassandra.gms.Gossiper;
-import org.apache.cassandra.locator.DynamicEndpointSnitch;
-import org.apache.cassandra.utils.FBUtilities;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.auth.Permission;
+import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnFamilyType;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.ExpiringColumn;
+import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.db.RangeSliceCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.Row;
+import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.db.SliceByNamesReadCommand;
+import org.apache.cassandra.db.SliceFromReadCommand;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.db.migration.AddColumnFamily;
import org.apache.cassandra.db.migration.AddKeyspace;
import org.apache.cassandra.db.migration.DropColumnFamily;
import org.apache.cassandra.db.migration.DropKeyspace;
+import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.db.migration.RenameColumnFamily;
import org.apache.cassandra.db.migration.RenameKeyspace;
-import org.apache.cassandra.locator.AbstractReplicationStrategy;
-import org.apache.cassandra.config.*;
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.ColumnFamily;
-import org.apache.cassandra.db.filter.QueryPath;
-import org.apache.cassandra.db.marshal.MarshalException;
+import org.apache.cassandra.db.migration.UpdateColumnFamily;
+import org.apache.cassandra.db.migration.UpdateKeyspace;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.locator.AbstractReplicationStrategy;
+import org.apache.cassandra.locator.DynamicEndpointSnitch;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class CassandraServer implements Cassandra.Iface
{
@@ -204,11 +221,11 @@ public class CassandraServer implements
return thriftSuperColumns;
}
- private Map<byte[], List<ColumnOrSuperColumn>> getSlice(List<ReadCommand> commands, ConsistencyLevel consistency_level)
+ private Map<ByteBuffer, List<ColumnOrSuperColumn>> getSlice(List<ReadCommand> commands, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
Map<DecoratedKey, ColumnFamily> columnFamilies = readColumnFamily(commands, consistency_level);
- Map<byte[], List<ColumnOrSuperColumn>> columnFamiliesMap = new HashMap<byte[], List<ColumnOrSuperColumn>>();
+ Map<ByteBuffer, List<ColumnOrSuperColumn>> columnFamiliesMap = new HashMap<ByteBuffer, List<ColumnOrSuperColumn>>();
for (ReadCommand command: commands)
{
ColumnFamily cf = columnFamilies.get(StorageService.getPartitioner().decorateKey(command.key));
@@ -239,7 +256,7 @@ public class CassandraServer implements
return thriftifyColumns(cf.getSortedColumns(), reverseOrder);
}
- public List<ColumnOrSuperColumn> get_slice(byte[] key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
+ public List<ColumnOrSuperColumn> get_slice(ByteBuffer key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -249,7 +266,7 @@ public class CassandraServer implements
return multigetSliceInternal(state().getKeyspace(), Collections.singletonList(key), column_parent, predicate, consistency_level).get(key);
}
- public Map<byte[], List<ColumnOrSuperColumn>> multiget_slice(List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
+ public Map<ByteBuffer, List<ColumnOrSuperColumn>> multiget_slice(List<ByteBuffer> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -260,7 +277,7 @@ public class CassandraServer implements
return multigetSliceInternal(state().getKeyspace(), keys, column_parent, predicate, consistency_level);
}
- private Map<byte[], List<ColumnOrSuperColumn>> multigetSliceInternal(String keyspace, List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
+ private Map<ByteBuffer, List<ColumnOrSuperColumn>> multigetSliceInternal(String keyspace, List<ByteBuffer> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
ThriftValidation.validateColumnParent(keyspace, column_parent);
@@ -269,7 +286,7 @@ public class CassandraServer implements
List<ReadCommand> commands = new ArrayList<ReadCommand>();
if (predicate.column_names != null)
{
- for (byte[] key: keys)
+ for (ByteBuffer key: keys)
{
ThriftValidation.validateKey(key);
commands.add(new SliceByNamesReadCommand(keyspace, key, column_parent, predicate.column_names));
@@ -278,7 +295,7 @@ public class CassandraServer implements
else
{
SliceRange range = predicate.slice_range;
- for (byte[] key: keys)
+ for (ByteBuffer key: keys)
{
ThriftValidation.validateKey(key);
commands.add(new SliceFromReadCommand(keyspace, key, column_parent, range.start, range.finish, range.reversed, range.count));
@@ -288,25 +305,26 @@ public class CassandraServer implements
return getSlice(commands, consistency_level);
}
- public ColumnOrSuperColumn get(byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level)
+ public ColumnOrSuperColumn get(ByteBuffer key, ColumnPath column_path, ConsistencyLevel consistency_level)
throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
logger.debug("get");
-
+
state().hasColumnFamilyAccess(column_path.column_family, Permission.READ);
String keyspace = state().getKeyspace();
ThriftValidation.validateColumnPath(keyspace, column_path);
QueryPath path = new QueryPath(column_path.column_family, column_path.column == null ? null : column_path.super_column);
- List<byte[]> nameAsList = Arrays.asList(column_path.column == null ? column_path.super_column : column_path.column);
+ List<ByteBuffer> nameAsList = Arrays.asList(column_path.column == null ? column_path.super_column : column_path.column);
ThriftValidation.validateKey(key);
ReadCommand command = new SliceByNamesReadCommand(keyspace, key, path, nameAsList);
Map<DecoratedKey, ColumnFamily> cfamilies = readColumnFamily(Arrays.asList(command), consistency_level);
ColumnFamily cf = cfamilies.get(StorageService.getPartitioner().decorateKey(command.key));
+
if (cf == null)
throw new NotFoundException();
List<ColumnOrSuperColumn> tcolumns = thriftifyColumnFamily(cf, command.queryPath.superColumnName != null, false);
@@ -316,7 +334,7 @@ public class CassandraServer implements
return tcolumns.get(0);
}
- public int get_count(byte[] key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
+ public int get_count(ByteBuffer key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -327,7 +345,7 @@ public class CassandraServer implements
return get_slice(key, column_parent, predicate, consistency_level).size();
}
- public Map<byte[], Integer> multiget_count(List<byte[]> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
+ public Map<ByteBuffer, Integer> multiget_count(List<ByteBuffer> keys, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -336,16 +354,16 @@ public class CassandraServer implements
state().hasColumnFamilyAccess(column_parent.column_family, Permission.READ);
String keyspace = state().getKeyspace();
- Map<byte[], Integer> counts = new HashMap<byte[], Integer>();
- Map<byte[], List<ColumnOrSuperColumn>> columnFamiliesMap = multigetSliceInternal(keyspace, keys, column_parent, predicate, consistency_level);
+ Map<ByteBuffer, Integer> counts = new HashMap<ByteBuffer, Integer>();
+ Map<ByteBuffer, List<ColumnOrSuperColumn>> columnFamiliesMap = multigetSliceInternal(keyspace, keys, column_parent, predicate, consistency_level);
- for (Map.Entry<byte[], List<ColumnOrSuperColumn>> cf : columnFamiliesMap.entrySet()) {
+ for (Map.Entry<ByteBuffer, List<ColumnOrSuperColumn>> cf : columnFamiliesMap.entrySet()) {
counts.put(cf.getKey(), cf.getValue().size());
}
return counts;
}
- public void insert(byte[] key, ColumnParent column_parent, Column column, ConsistencyLevel consistency_level)
+ public void insert(ByteBuffer key, ColumnParent column_parent, Column column, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -369,7 +387,7 @@ public class CassandraServer implements
doInsert(consistency_level, Arrays.asList(rm));
}
- public void batch_mutate(Map<byte[],Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level)
+ public void batch_mutate(Map<ByteBuffer,Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
@@ -378,9 +396,9 @@ public class CassandraServer implements
List<String> cfamsSeen = new ArrayList<String>();
List<RowMutation> rowMutations = new ArrayList<RowMutation>();
- for (Map.Entry<byte[], Map<String, List<Mutation>>> mutationEntry: mutation_map.entrySet())
+ for (Map.Entry<ByteBuffer, Map<String, List<Mutation>>> mutationEntry: mutation_map.entrySet())
{
- byte[] key = mutationEntry.getKey();
+ ByteBuffer key = mutationEntry.getKey();
ThriftValidation.validateKey(key);
Map<String, List<Mutation>> columnFamilyToMutations = mutationEntry.getValue();
@@ -406,7 +424,7 @@ public class CassandraServer implements
doInsert(consistency_level, rowMutations);
}
- public void remove(byte[] key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level)
+ public void remove(ByteBuffer key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (logger.isDebugEnabled())
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/ThriftValidation.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/ThriftValidation.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/ThriftValidation.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/ThriftValidation.java Fri Oct 22 03:23:26 2010
@@ -20,15 +20,17 @@ package org.apache.cassandra.thrift;
*
*/
+import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Set;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.ColumnFamilyType;
+import org.apache.cassandra.db.IColumn;
+import org.apache.cassandra.db.KeyspaceNotDefinedException;
+import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.dht.IPartitioner;
@@ -36,21 +38,23 @@ import org.apache.cassandra.dht.RandomPa
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class ThriftValidation
{
private static final Logger logger = LoggerFactory.getLogger(ThriftValidation.class);
- static void validateKey(byte[] key) throws InvalidRequestException
+ static void validateKey(ByteBuffer key) throws InvalidRequestException
{
- if (key == null || key.length == 0)
+ if (key == null || key.remaining() == 0)
{
throw new InvalidRequestException("Key may not be empty");
}
// check that key can be handled by FBUtilities.writeShortByteArray
- if (key.length > FBUtilities.MAX_UNSIGNED_SHORT)
+ if (key.remaining() > FBUtilities.MAX_UNSIGNED_SHORT)
{
- throw new InvalidRequestException("Key length of " + key.length +
+ throw new InvalidRequestException("Key length of " + key.remaining() +
" is longer than maximum of " + FBUtilities.MAX_UNSIGNED_SHORT);
}
}
@@ -146,24 +150,24 @@ public class ThriftValidation
}
}
- private static void validateColumns(String keyspace, String columnFamilyName, byte[] superColumnName, Iterable<byte[]> column_names)
+ private static void validateColumns(String keyspace, String columnFamilyName, ByteBuffer superColumnName, Iterable<ByteBuffer> column_names)
throws InvalidRequestException
{
if (superColumnName != null)
{
- if (superColumnName.length > IColumn.MAX_NAME_LENGTH)
+ if (superColumnName.remaining() > IColumn.MAX_NAME_LENGTH)
throw new InvalidRequestException("supercolumn name length must not be greater than " + IColumn.MAX_NAME_LENGTH);
- if (superColumnName.length == 0)
+ if (superColumnName.remaining() == 0)
throw new InvalidRequestException("supercolumn name must not be empty");
if (DatabaseDescriptor.getColumnFamilyType(keyspace, columnFamilyName) == ColumnFamilyType.Standard)
throw new InvalidRequestException("supercolumn specified to ColumnFamily " + columnFamilyName + " containing normal columns");
}
AbstractType comparator = ColumnFamily.getComparatorFor(keyspace, columnFamilyName, superColumnName);
- for (byte[] name : column_names)
+ for (ByteBuffer name : column_names)
{
- if (name.length > IColumn.MAX_NAME_LENGTH)
+ if (name.remaining() > IColumn.MAX_NAME_LENGTH)
throw new InvalidRequestException("column name length must not be greater than " + IColumn.MAX_NAME_LENGTH);
- if (name.length == 0)
+ if (name.remaining() == 0)
throw new InvalidRequestException("column name must not be empty");
try
{
@@ -176,7 +180,7 @@ public class ThriftValidation
}
}
- public static void validateColumns(String keyspace, ColumnParent column_parent, Iterable<byte[]> column_names) throws InvalidRequestException
+ public static void validateColumns(String keyspace, ColumnParent column_parent, Iterable<ByteBuffer> column_names) throws InvalidRequestException
{
validateColumns(keyspace, column_parent.column_family, column_parent.super_column, column_names);
}
@@ -197,9 +201,9 @@ public class ThriftValidation
if (range.count < 0)
throw new InvalidRequestException("get_slice requires non-negative count");
- Comparator<byte[]> orderedComparator = range.isReversed() ? comparator.getReverseComparator() : comparator;
- if (range.start.length > 0
- && range.finish.length > 0
+ Comparator<ByteBuffer> orderedComparator = range.isReversed() ? comparator.getReverseComparator() : comparator;
+ if (range.start.remaining() > 0
+ && range.finish.remaining() > 0
&& orderedComparator.compare(range.start, range.finish) > 0)
{
throw new InvalidRequestException("range finish must come after start in the order of traversal");
@@ -212,7 +216,7 @@ public class ThriftValidation
if (cosc.column != null)
{
validateTtl(cosc.column);
- ThriftValidation.validateColumnPath(keyspace, new ColumnPath(cfName).setSuper_column(null).setColumn(cosc.column.name));
+ ThriftValidation.validateColumnPath(keyspace, new ColumnPath(cfName).setSuper_column((ByteBuffer)null).setColumn(cosc.column.name));
}
if (cosc.super_column != null)
@@ -278,7 +282,7 @@ public class ThriftValidation
}
}
- public static void validateSlicePredicate(String keyspace, String cfName, byte[] scName, SlicePredicate predicate) throws InvalidRequestException
+ public static void validateSlicePredicate(String keyspace, String cfName, ByteBuffer scName, SlicePredicate predicate) throws InvalidRequestException
{
if (predicate.column_names == null && predicate.slice_range == null)
throw new InvalidRequestException("A SlicePredicate must be given a list of Columns, a SliceRange, or both");
@@ -365,7 +369,7 @@ public class ThriftValidation
{
if (index_clause.expressions.isEmpty())
throw new InvalidRequestException("index clause list may not be empty");
- Set<byte[]> indexedColumns = Table.open(keyspace).getColumnFamilyStore(columnFamily).getIndexedColumns();
+ Set<ByteBuffer> indexedColumns = Table.open(keyspace).getColumnFamilyStore(columnFamily).getIndexedColumns();
for (IndexExpression expression : index_clause.expressions)
{
if (expression.op.equals(IndexOperator.EQ) && indexedColumns.contains(expression.column_name))
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java Fri Oct 22 03:23:26 2010
@@ -25,6 +25,7 @@ import java.lang.management.MemoryMXBean
import java.lang.management.MemoryUsage;
import java.lang.management.RuntimeMXBean;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
@@ -374,7 +375,7 @@ public class NodeProbe
public List<InetAddress> getEndpoints(String keyspace, String key)
{
// FIXME: string key
- return ssProxy.getNaturalEndpoints(keyspace, key.getBytes(UTF_8));
+ return ssProxy.getNaturalEndpoints(keyspace, ByteBuffer.wrap(key.getBytes(UTF_8)));
}
public Set<InetAddress> getStreamDestinations()
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableExport.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableExport.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableExport.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableExport.java Fri Oct 22 03:23:26 2010
@@ -21,6 +21,7 @@ package org.apache.cassandra.tools;
import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
+import java.nio.ByteBuffer;
import java.util.*;
import org.apache.commons.cli.*;
@@ -201,7 +202,7 @@ public class SSTableExport
{
if (excludeSet.contains(key))
continue;
- DecoratedKey<?> dk = partitioner.decorateKey(hexToBytes(key));
+ DecoratedKey<?> dk = partitioner.decorateKey(ByteBuffer.wrap(hexToBytes(key)));
scanner.seekTo(dk);
i++;
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java Fri Oct 22 03:23:26 2010
@@ -20,6 +20,7 @@ package org.apache.cassandra.tools;
import java.io.FileReader;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -92,11 +93,11 @@ public class SSTableImport
for (Object c : row)
{
JsonColumn col = new JsonColumn(c);
- QueryPath path = new QueryPath(cfm.cfName, null, hexToBytes(col.name));
+ QueryPath path = new QueryPath(cfm.cfName, null, ByteBuffer.wrap(hexToBytes(col.name)));
if (col.isDeleted) {
- cfamily.addColumn(path, hexToBytes(col.value), col.timestamp);
+ cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp);
} else {
- cfamily.addTombstone(path, hexToBytes(col.value), col.timestamp);
+ cfamily.addTombstone(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp);
}
}
}
@@ -114,7 +115,7 @@ public class SSTableImport
// Super columns
for (Map.Entry<String, JSONObject> entry : (Set<Map.Entry<String, JSONObject>>)row.entrySet())
{
- byte[] superName = hexToBytes(entry.getKey());
+ ByteBuffer superName = ByteBuffer.wrap(hexToBytes(entry.getKey()));
long deletedAt = (Long)entry.getValue().get("deletedAt");
JSONArray subColumns = (JSONArray)entry.getValue().get("subColumns");
@@ -122,11 +123,11 @@ public class SSTableImport
for (Object c : subColumns)
{
JsonColumn col = new JsonColumn(c);
- QueryPath path = new QueryPath(cfm.cfName, superName, hexToBytes(col.name));
+ QueryPath path = new QueryPath(cfm.cfName, superName, ByteBuffer.wrap(hexToBytes(col.name)));
if (col.isDeleted) {
- cfamily.addColumn(path, hexToBytes(col.value), col.timestamp);
+ cfamily.addColumn(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp);
} else {
- cfamily.addTombstone(path, hexToBytes(col.value), col.timestamp);
+ cfamily.addTombstone(path, ByteBuffer.wrap(hexToBytes(col.value)), col.timestamp);
}
}
@@ -161,7 +162,7 @@ public class SSTableImport
// sort by dk representation, but hold onto the hex version
for (String key : (Set<String>)json.keySet())
- decoratedKeys.put(partitioner.decorateKey(hexToBytes(key)), key);
+ decoratedKeys.put(partitioner.decorateKey(ByteBuffer.wrap(hexToBytes(key))), key);
for (Map.Entry<DecoratedKey, String> rowKey : decoratedKeys.entrySet())
{
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/BloomFilter.java Fri Oct 22 03:23:26 2010
@@ -21,10 +21,10 @@ package org.apache.cassandra.utils;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.BitSet;
import org.apache.cassandra.io.ICompactSerializer;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -116,7 +116,7 @@ public class BloomFilter extends Filter
return filter_;
}
- public boolean isPresent(byte[] key)
+ public boolean isPresent(ByteBuffer key)
{
for (int bucketIndex : getHashBuckets(key))
{
@@ -133,7 +133,7 @@ public class BloomFilter extends Filter
the filter_.
This is a general purpose API.
*/
- public void add(byte[] key)
+ public void add(ByteBuffer key)
{
for (int bucketIndex : getHashBuckets(key))
{
Copied: cassandra/trunk/src/java/org/apache/cassandra/utils/ByteBufferUtil.java (from r1026177, cassandra/trunk/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java)
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/ByteBufferUtil.java?p2=cassandra/trunk/src/java/org/apache/cassandra/utils/ByteBufferUtil.java&p1=cassandra/trunk/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java&r1=1026177&r2=1026200&rev=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/marshal/LocalByPartionerType.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/ByteBufferUtil.java Fri Oct 22 03:23:26 2010
@@ -1,4 +1,4 @@
-/**
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -16,30 +16,32 @@
* specific language governing permissions and limitations
* under the License.
*/
+package org.apache.cassandra.utils;
-package org.apache.cassandra.db.marshal;
-
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.Token;
+import java.nio.ByteBuffer;
-/** for sorting columns representing row keys in the row ordering as determined by a partitioner.
- * Not intended for user-defined CFs, and will in fact error out if used with such. */
-public class LocalByPartionerType<T extends Token> extends AbstractType
-{
- private final IPartitioner<T> partitioner;
+/**
+ * Utility methods to make ByteBuffers less painful
+ *
+ */
+public class ByteBufferUtil {
- public LocalByPartionerType(IPartitioner<T> partitioner)
+ public static int compare(ByteBuffer o1, ByteBuffer o2)
{
- this.partitioner = partitioner;
+ return FBUtilities.compareByteArrays(o1.array(), o2.array(), o1.arrayOffset()+o1.position(), o2.arrayOffset()+o2.position(), o1.limit(), o2.limit());
}
-
- public String getString(byte[] bytes)
+
+ public static int compare(byte[] o1, ByteBuffer o2)
{
- return null;
+ return FBUtilities.compareByteArrays(o1, o2.array(), 0, o2.arrayOffset()+o2.position(), o1.length, o2.limit());
}
-
- public int compare(byte[] o1, byte[] o2)
+ public static int compare(ByteBuffer o1, byte[] o2)
{
- return partitioner.decorateKey(o1).compareTo(partitioner.decorateKey(o2));
+ return FBUtilities.compareByteArrays(o1.array(), o2, o1.arrayOffset()+o1.position(), 0, o1.limit(), o2.length);
}
+
+ public static boolean equals(ByteBuffer o1, ByteBuffer o2)
+ {
+ return compare(o1, o2) == 0;
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/FBUtilities.java Fri Oct 22 03:23:26 2010
@@ -37,6 +37,7 @@ import java.util.concurrent.atomic.Atomi
import com.google.common.base.Charsets;
import org.apache.commons.collections.iterators.CollatingIterator;
+import org.apache.commons.lang.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -60,17 +61,19 @@ public class FBUtilities
public static final BigInteger TWO = new BigInteger("2");
+ public static final ByteBuffer EMPTY_BYTE_BUFFER = ByteBuffer.wrap(ArrayUtils.EMPTY_BYTE_ARRAY);
+
private static volatile InetAddress localInetAddress_;
public static final int MAX_UNSIGNED_SHORT = 0xFFFF;
- public static final Comparator<byte[]> byteArrayComparator = new Comparator<byte[]>()
+ /*public static final Comparator<byte[]> byteArrayComparator = new Comparator<byte[]>()
{
public int compare(byte[] o1, byte[] o2)
{
return compareByteArrays(o1, o2);
}
- };
+ };*/
/**
* Parses a string representing either a fraction, absolute value or percentage.
@@ -155,24 +158,24 @@ public class FBUtilities
return new Pair(midpoint, remainder);
}
- public static byte[] toByteArray(int i)
+ public static ByteBuffer toByteArray(int i)
{
byte[] bytes = new byte[4];
bytes[0] = (byte)( ( i >>> 24 ) & 0xFF);
bytes[1] = (byte)( ( i >>> 16 ) & 0xFF);
bytes[2] = (byte)( ( i >>> 8 ) & 0xFF);
bytes[3] = (byte)( i & 0xFF );
- return bytes;
+ return ByteBuffer.wrap(bytes);
}
- public static int byteArrayToInt(byte[] bytes)
+ public static int byteArrayToInt(ByteBuffer bytes)
{
return byteArrayToInt(bytes, 0);
}
- public static int byteArrayToInt(byte[] bytes, int offset)
+ public static int byteArrayToInt(ByteBuffer bytes, int offset)
{
- if ( bytes.length - offset < 4 )
+ if ( bytes.remaining() - offset < 4 )
{
throw new IllegalArgumentException("An integer must be 4 bytes in size.");
}
@@ -180,28 +183,28 @@ public class FBUtilities
for ( int i = 0; i < 4; ++i )
{
n <<= 8;
- n |= bytes[offset + i] & 0xFF;
+ n |= bytes.array()[bytes.position()+bytes.arrayOffset()+ offset + i] & 0xFF;
}
return n;
}
-
- public static int compareByteArrays(byte[] bytes1, byte[] bytes2){
+
+ public static int compareByteArrays(byte[] bytes1, byte[] bytes2, int offset1, int offset2, int len1, int len2){
if(null == bytes1){
if(null == bytes2) return 0;
else return -1;
}
if(null == bytes2) return 1;
- int minLength = Math.min(bytes1.length, bytes2.length);
- for(int i = 0; i < minLength; i++)
+ int minLength = Math.min(len1-offset1, len2-offset2);
+ for(int x=0, i = offset1, j=offset2; x < minLength; x++,i++,j++)
{
- if(bytes1[i] == bytes2[i])
+ if(bytes1[i] == bytes2[j])
continue;
// compare non-equal bytes as unsigned
- return (bytes1[i] & 0xFF) < (bytes2[i] & 0xFF) ? -1 : 1;
+ return (bytes1[i] & 0xFF) < (bytes2[j] & 0xFF) ? -1 : 1;
}
- if(bytes1.length == bytes2.length) return 0;
- else return (bytes1.length < bytes2.length)? -1 : 1;
+ if((len1-offset1) == (len2-offset2)) return 0;
+ else return ((len1-offset1) < (len2-offset2))? -1 : 1;
}
/**
@@ -228,21 +231,21 @@ public class FBUtilities
return out;
}
- public static BigInteger md5hash(byte[] data)
+ public static BigInteger md5hash(ByteBuffer data)
{
byte[] result = hash("MD5", data);
BigInteger hash = new BigInteger(result);
return hash.abs();
}
- public static byte[] hash(String type, byte[]... data)
+ public static byte[] hash(String type, ByteBuffer... data)
{
byte[] result;
try
{
MessageDigest messageDigest = MessageDigest.getInstance(type);
- for(byte[] block : data)
- messageDigest.update(block);
+ for(ByteBuffer block : data)
+ messageDigest.update(block.array(),block.position()+block.arrayOffset(),block.remaining());
result = messageDigest.digest();
}
catch (Exception e)
@@ -252,13 +255,13 @@ public class FBUtilities
return result;
}
- public static void writeByteArray(byte[] bytes, DataOutput out) throws IOException
+ public static void writeByteArray(ByteBuffer bytes, DataOutput out) throws IOException
{
- out.writeInt(bytes.length);
- out.write(bytes);
+ out.writeInt(bytes.remaining());
+ out.write(bytes.array(),bytes.position()+bytes.arrayOffset(),bytes.remaining());
}
- public static byte[] readByteArray(DataInput in) throws IOException
+ public static ByteBuffer readByteArray(DataInput in) throws IOException
{
int length = in.readInt();
if (length < 0)
@@ -270,24 +273,26 @@ public class FBUtilities
{
in.readFully(value);
}
- return value;
+ return ByteBuffer.wrap(value);
}
- public static void writeShortByteArray(byte[] name, DataOutput out)
+ public static void writeShortByteArray(ByteBuffer name, DataOutput out)
{
- int length = name.length;
+ int length = name.remaining();
assert 0 <= length && length <= MAX_UNSIGNED_SHORT;
try
{
out.writeByte((length >> 8) & 0xFF);
out.writeByte(length & 0xFF);
- out.write(name);
+ out.write(name.array(), name.position()+name.arrayOffset(), name.remaining());
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
+
+
/** @return An unsigned short in an integer. */
private static int readShortLength(DataInput in) throws IOException
@@ -296,11 +301,11 @@ public class FBUtilities
return length | (in.readByte() & 0xFF);
}
- public static byte[] readShortByteArray(DataInput in) throws IOException
+ public static ByteBuffer readShortByteArray(DataInput in) throws IOException
{
byte[] bytes = new byte[readShortLength(in)];
in.readFully(bytes);
- return bytes;
+ return ByteBuffer.wrap(bytes);
}
/** @return null */
@@ -328,12 +333,12 @@ public class FBUtilities
return bytes;
}
- public static String bytesToHex(byte... bytes)
+ public static String bytesToHex(ByteBuffer bytes)
{
StringBuilder sb = new StringBuilder();
- for (byte b : bytes)
+ for (int i=bytes.position()+bytes.arrayOffset(); i<bytes.limit(); i++)
{
- int bint = b & 0xff;
+ int bint = bytes.array()[i] & 0xff;
if (bint <= 0xF)
// toHexString does not 0 pad its results.
sb.append("0");
@@ -478,16 +483,20 @@ public class FBUtilities
return utflen;
}
- public static String decodeToUTF8(byte[] bytes) throws CharacterCodingException
+ public static String decodeToUTF8(ByteBuffer bytes) throws CharacterCodingException
{
- return Charsets.UTF_8.newDecoder().decode(ByteBuffer.wrap(bytes)).toString();
+ bytes.mark();
+ String decoded = Charsets.UTF_8.newDecoder().decode(bytes).toString();
+ bytes.reset();
+ return decoded;
}
- public static byte[] toByteArray(long n)
+ public static ByteBuffer toByteArray(long n)
{
byte[] bytes = new byte[8];
- ByteBuffer.wrap(bytes).putLong(n);
- return bytes;
+ ByteBuffer bb = ByteBuffer.wrap(bytes).putLong(n);
+ bb.rewind();
+ return bb;
}
public static String resourceToFile(String filename) throws ConfigurationException
@@ -656,16 +665,16 @@ public class FBUtilities
}
}
- public static TreeSet<byte[]> getSingleColumnSet(byte[] column)
+ public static TreeSet<ByteBuffer> getSingleColumnSet(ByteBuffer column)
{
- Comparator<byte[]> singleColumnComparator = new Comparator<byte[]>()
+ Comparator<ByteBuffer> singleColumnComparator = new Comparator<ByteBuffer>()
{
- public int compare(byte[] o1, byte[] o2)
+ public int compare(ByteBuffer o1, ByteBuffer o2)
{
- return Arrays.equals(o1, o2) ? 0 : -1;
+ return ByteBufferUtil.equals(o1, o2) ? 0 : -1;
}
};
- TreeSet<byte[]> set = new TreeSet<byte[]>(singleColumnComparator);
+ TreeSet<ByteBuffer> set = new TreeSet<ByteBuffer>(singleColumnComparator);
set.add(column);
return set;
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/Filter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/Filter.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/Filter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/Filter.java Fri Oct 22 03:23:26 2010
@@ -18,8 +18,8 @@
*/
package org.apache.cassandra.utils;
-import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
import org.apache.cassandra.io.ICompactSerializer;
@@ -34,16 +34,16 @@ public abstract class Filter
return hashCount;
}
- public int[] getHashBuckets(byte[] key)
+ public int[] getHashBuckets(ByteBuffer key)
{
return Filter.getHashBuckets(key, hashCount, buckets());
}
abstract int buckets();
- public abstract void add(byte[] key);
+ public abstract void add(ByteBuffer key);
- public abstract boolean isPresent(byte[] key);
+ public abstract boolean isPresent(ByteBuffer key);
// for testing
abstract int emptyBuckets();
@@ -67,11 +67,11 @@ public abstract class Filter
// http://www.eecs.harvard.edu/~kirsch/pubs/bbbf/esa06.pdf
// does prove to work in actual tests, and is obviously faster
// than performing further iterations of murmur.
- static int[] getHashBuckets(byte[] b, int hashCount, int max)
+ static int[] getHashBuckets(ByteBuffer b, int hashCount, int max)
{
int[] result = new int[hashCount];
- int hash1 = hasher.hash(b, b.length, 0);
- int hash2 = hasher.hash(b, b.length, hash1);
+ int hash1 = hasher.hash(b.array(), b.position()+b.arrayOffset(), b.remaining(), 0);
+ int hash2 = hasher.hash(b.array(), b.position()+b.arrayOffset(), b.remaining(), hash1);
for (int i = 0; i < hashCount; i++)
{
result[i] = Math.abs((hash1 + i * hash2) % max);
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/GuidGenerator.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/GuidGenerator.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/GuidGenerator.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/GuidGenerator.java Fri Oct 22 03:23:26 2010
@@ -18,13 +18,17 @@
package org.apache.cassandra.utils;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.security.SecureRandom;
+import java.util.Random;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.*;
-import java.net.*;
-import java.security.*;
-
public class GuidGenerator {
private static Logger logger_ = LoggerFactory.getLogger(GuidGenerator.class);
private static Random myRand;
@@ -57,11 +61,11 @@ public class GuidGenerator {
public static String guid() {
- byte[] array = guidAsBytes();
+ ByteBuffer array = guidAsBytes();
StringBuilder sb = new StringBuilder();
- for (int j = 0; j < array.length; ++j) {
- int b = array[j] & 0xFF;
+ for (int j = array.position()+array.arrayOffset(); j < array.limit(); ++j) {
+ int b = array.array()[j] & 0xFF;
if (b < 0x10) sb.append('0');
sb.append(Integer.toHexString(b));
}
@@ -81,7 +85,7 @@ public class GuidGenerator {
return convertToStandardFormat( sb.toString() );
}
- public static byte[] guidAsBytes()
+ public static ByteBuffer guidAsBytes()
{
StringBuilder sbValueBeforeMD5 = new StringBuilder();
long time = System.currentTimeMillis();
@@ -94,7 +98,7 @@ public class GuidGenerator {
.append(Long.toString(rand));
String valueBeforeMD5 = sbValueBeforeMD5.toString();
- return md5.digest(valueBeforeMD5.getBytes());
+ return ByteBuffer.wrap(md5.digest(valueBeforeMD5.getBytes()));
}
/*
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/MerkleTree.java Fri Oct 22 03:23:26 2010
@@ -19,13 +19,20 @@
package org.apache.cassandra.utils;
import java.io.Serializable;
-import java.util.*;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
import com.google.common.collect.AbstractIterator;
import com.google.common.collect.PeekingIterator;
-import org.apache.cassandra.dht.*;
-
/**
* A MerkleTree implemented as a binary tree.
*
@@ -369,7 +376,7 @@ public class MerkleTree implements Seria
}
return true;
}
-
+
private Hashable splitHelper(Hashable hashable, Token pleft, Token pright, byte depth, Token t) throws StopRecursion.TooDeep
{
if (depth >= hashdepth)
@@ -708,8 +715,8 @@ public class MerkleTree implements Seria
public final byte[] hash;
public RowHash(Token token, byte[] hash)
{
- this.token = token;
- this.hash = hash;
+ this.token = token;
+ this.hash = hash;
}
@Override
@@ -780,7 +787,7 @@ public class MerkleTree implements Seria
{
if (hash == null)
return "null";
- return "[" + FBUtilities.bytesToHex(hash) + "]";
+ return "[" + FBUtilities.bytesToHex(ByteBuffer.wrap(hash)) + "]";
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/MurmurHash.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/MurmurHash.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/MurmurHash.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/MurmurHash.java Fri Oct 22 03:23:26 2010
@@ -18,6 +18,8 @@
package org.apache.cassandra.utils;
+import java.nio.ByteBuffer;
+
/**
* This is a very fast, non-cryptographic hash suitable for general hash-based
* lookup. See http://murmurhash.googlepages.com/ for more details.
@@ -26,7 +28,7 @@ package org.apache.cassandra.utils;
* to Java by Andrzej Bialecki (ab at getopt org).</p>
*/
public class MurmurHash {
- public int hash(byte[] data, int length, int seed) {
+ public int hash(byte[] data, int offset, int length, int seed) {
int m = 0x5bd1e995;
int r = 24;
@@ -36,13 +38,13 @@ public class MurmurHash {
for (int i = 0; i < len_4; i++) {
int i_4 = i << 2;
- int k = data[i_4 + 3];
+ int k = data[offset + i_4 + 3];
k = k << 8;
- k = k | (data[i_4 + 2] & 0xff);
+ k = k | (data[offset + i_4 + 2] & 0xff);
k = k << 8;
- k = k | (data[i_4 + 1] & 0xff);
+ k = k | (data[offset + i_4 + 1] & 0xff);
k = k << 8;
- k = k | (data[i_4 + 0] & 0xff);
+ k = k | (data[offset + i_4 + 0] & 0xff);
k *= m;
k ^= k >>> r;
k *= m;
@@ -56,13 +58,13 @@ public class MurmurHash {
if (left != 0) {
if (left >= 3) {
- h ^= (int) data[length - 3] << 16;
+ h ^= (int) data[offset+length - 3] << 16;
}
if (left >= 2) {
- h ^= (int) data[length - 2] << 8;
+ h ^= (int) data[offset+length - 2] << 8;
}
if (left >= 1) {
- h ^= (int) data[length - 1];
+ h ^= (int) data[offset+length - 1];
}
h *= m;
Modified: cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/utils/UUIDGen.java Fri Oct 22 03:23:26 2010
@@ -21,17 +21,17 @@ package org.apache.cassandra.utils;
*/
-import org.safehaus.uuid.EthernetAddress;
-import org.safehaus.uuid.UUIDGenerator;
-
import java.io.DataInputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.UUID;
+import org.safehaus.uuid.EthernetAddress;
+import org.safehaus.uuid.UUIDGenerator;
+
/**
* Generates type 1 (time-based) UUIDs
*/
@@ -47,7 +47,7 @@ public class UUIDGen
byte[] md5 = digest.digest();
byte[] fauxMac = new byte[6];
System.arraycopy(md5, 0, fauxMac, 0, Math.min(md5.length, fauxMac.length));
- return makeType1UUID(UUIDGenerator.getInstance().generateTimeBasedUUID(new EthernetAddress(fauxMac)).toByteArray());
+ return makeType1UUID(ByteBuffer.wrap(UUIDGenerator.getInstance().generateTimeBasedUUID(new EthernetAddress(fauxMac)).toByteArray()));
}
catch (NoSuchAlgorithmException ex)
{
@@ -59,19 +59,19 @@ public class UUIDGen
{
byte[] b = new byte[16];
in.readFully(b);
- return makeType1UUID(b);
+ return makeType1UUID(ByteBuffer.wrap(b));
}
/** creates a type 1 uuid from raw bytes. */
- public static UUID makeType1UUID(byte[] raw)
+ public static UUID makeType1UUID(ByteBuffer raw)
{
long most = 0;
long least = 0;
- assert raw.length == 16;
+ assert raw.remaining() == 16;
for (int i = 0; i < 8; i++)
- most = (most << 8) | (raw[i] & 0xff);
+ most = (most << 8) | (raw.array()[raw.position()+raw.arrayOffset() + i] & 0xff);
for (int i =8 ; i < 16; i++)
- least = (least << 8) | (raw[i] & 0xff);
+ least = (least << 8) | (raw.array()[raw.position()+raw.arrayOffset() + i] & 0xff);
return new UUID(most, least);
}
Modified: cassandra/trunk/test/system/test_thrift_server.py
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/system/test_thrift_server.py?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/test/system/test_thrift_server.py (original)
+++ cassandra/trunk/test/system/test_thrift_server.py Fri Oct 22 03:23:26 2010
@@ -1322,7 +1322,7 @@ class TestMutations(ThriftTester):
assert server_cf
assert server_cf.column_metadata[0].index_type == modified_cd.index_type
assert server_cf.column_metadata[0].index_name == modified_cd.index_name
-
+
# simple query on one index expression
cp = ColumnParent('ToBeIndexed')
sp = SlicePredicate(slice_range=SliceRange('', ''))
Modified: cassandra/trunk/test/unit/org/apache/cassandra/Util.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/Util.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/Util.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/Util.java Fri Oct 22 03:23:26 2010
@@ -48,17 +48,17 @@ public class Util
{
public static DecoratedKey dk(String key)
{
- return StorageService.getPartitioner().decorateKey(key.getBytes(UTF_8));
+ return StorageService.getPartitioner().decorateKey(ByteBuffer.wrap(key.getBytes(UTF_8)));
}
public static Column column(String name, String value, long timestamp)
{
- return new Column(name.getBytes(), value.getBytes(), timestamp);
+ return new Column(ByteBuffer.wrap(name.getBytes()), ByteBuffer.wrap(value.getBytes()), timestamp);
}
public static Token token(String key)
{
- return StorageService.getPartitioner().getToken(key.getBytes());
+ return StorageService.getPartitioner().getToken(ByteBuffer.wrap(key.getBytes()));
}
public static Range range(String left, String right)
@@ -68,7 +68,7 @@ public class Util
public static Range range(IPartitioner p, String left, String right)
{
- return new Range(p.getToken(left.getBytes()), p.getToken(right.getBytes()));
+ return new Range(p.getToken(ByteBuffer.wrap(left.getBytes())), p.getToken(ByteBuffer.wrap(right.getBytes())));
}
public static Bounds bounds(String left, String right)
@@ -78,15 +78,16 @@ public class Util
public static void addMutation(RowMutation rm, String columnFamilyName, String superColumnName, long columnName, String value, long timestamp)
{
- rm.add(new QueryPath(columnFamilyName, superColumnName.getBytes(), getBytes(columnName)), value.getBytes(), timestamp);
+ rm.add(new QueryPath(columnFamilyName, ByteBuffer.wrap(superColumnName.getBytes()), getBytes(columnName)), ByteBuffer.wrap(value.getBytes()), timestamp);
}
- public static byte[] getBytes(long v)
+ public static ByteBuffer getBytes(long v)
{
byte[] bytes = new byte[8];
ByteBuffer bb = ByteBuffer.wrap(bytes);
bb.putLong(v);
- return bytes;
+ bb.rewind();
+ return bb;
}
public static List<Row> getRangeSlice(ColumnFamilyStore cfs) throws IOException, ExecutionException, InterruptedException
Modified: cassandra/trunk/test/unit/org/apache/cassandra/client/TestRingCache.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/client/TestRingCache.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/client/TestRingCache.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/client/TestRingCache.java Fri Oct 22 03:23:26 2010
@@ -19,16 +19,19 @@ package org.apache.cassandra.client;
import java.io.IOException;
import java.net.InetAddress;
+import java.nio.ByteBuffer;
import java.util.Collection;
-import org.apache.commons.lang.StringUtils;
-
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.thrift.Cassandra;
+import org.apache.cassandra.thrift.Column;
+import org.apache.cassandra.thrift.ColumnParent;
+import org.apache.cassandra.thrift.ColumnPath;
+import org.apache.cassandra.thrift.ConsistencyLevel;
+import org.apache.commons.lang.StringUtils;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
/**
* Sample code that uses RingCache in the client.
@@ -86,21 +89,21 @@ public class TestRingCache
for (int nRows = minRow; nRows < maxRow; nRows++)
{
- byte[] row = (rowPrefix + nRows).getBytes();
- ColumnPath col = new ColumnPath("Standard1").setSuper_column(null).setColumn("col1".getBytes());
- ColumnParent parent = new ColumnParent("Standard1").setSuper_column(null);
+ ByteBuffer row = ByteBuffer.wrap((rowPrefix + nRows).getBytes());
+ ColumnPath col = new ColumnPath("Standard1").setSuper_column((ByteBuffer)null).setColumn("col1".getBytes());
+ ColumnParent parent = new ColumnParent("Standard1").setSuper_column((ByteBuffer)null);
Collection<InetAddress> endpoints = tester.ringCache.getEndpoint(row);
InetAddress firstEndpoint = endpoints.iterator().next();
System.out.printf("hosts with key %s : %s; choose %s%n",
- new String(row), StringUtils.join(endpoints, ","), firstEndpoint);
+ new String(row.array()), StringUtils.join(endpoints, ","), firstEndpoint);
// now, read the row back directly from the host owning the row locally
tester.setup(firstEndpoint.getHostAddress(), DatabaseDescriptor.getRpcPort());
tester.thriftClient.set_keyspace(keyspace);
- tester.thriftClient.insert(row, parent, new Column("col1".getBytes(), "val1".getBytes(), 1), ConsistencyLevel.ONE);
+ tester.thriftClient.insert(row, parent, new Column(ByteBuffer.wrap("col1".getBytes()), ByteBuffer.wrap("val1".getBytes()), 1), ConsistencyLevel.ONE);
Column column = tester.thriftClient.get(row, col, ConsistencyLevel.ONE).column;
- System.out.println("read row " + new String(row) + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
+ System.out.println("read row " + new String(row.array()) + " " + new String(column.name.array()) + ":" + new String(column.value.array()) + ":" + column.timestamp);
}
System.exit(1);
Modified: cassandra/trunk/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java?rev=1026200&r1=1026199&r2=1026200&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/config/ColumnDefinitionTest.java Fri Oct 22 03:23:26 2010
@@ -21,6 +21,8 @@ package org.apache.cassandra.config;
*/
+import java.nio.ByteBuffer;
+
import org.junit.Test;
import org.apache.cassandra.thrift.IndexType;
@@ -30,12 +32,12 @@ public class ColumnDefinitionTest
@Test
public void testSerializeDeserialize() throws Exception
{
- ColumnDefinition cd0 = new ColumnDefinition("TestColumnDefinitionName0".getBytes("UTF8"),
+ ColumnDefinition cd0 = new ColumnDefinition(ByteBuffer.wrap("TestColumnDefinitionName0".getBytes("UTF8")),
"BytesType",
IndexType.KEYS,
"random index name 0");
- ColumnDefinition cd1 = new ColumnDefinition("TestColumnDefinition1".getBytes("UTF8"),
+ ColumnDefinition cd1 = new ColumnDefinition(ByteBuffer.wrap("TestColumnDefinition1".getBytes("UTF8")),
"LongType",
null,
null);