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

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1147869 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-07-18 13:32:48 +00:00
parent 273c53831f 83f9f455d7
commit b803c37852
31 changed files with 298 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGES.txt
View File

@ -40,6 +40,12 @@
* add ant-optional as dependence for the debian package (CASSANDRA-2164)
* add option to specify limit for get_slice in the CLI (CASSANDRA-2646)
* decrease HH page size (CASSANDRA-2832)
* reset cli keyspace after dropping the current one (CASSANDRA-2763)
* add KeyRange option to Hadoop inputformat (CASSANDRA-1125)
* fix protocol versioning (CASSANDRA-2818, 2860)
* support spaces in path to log4j configuration (CASSANDRA-2383)
* avoid including inferred types in CF update (CASSANDRA-2809)
* fix JMX bulkload call (CASSANDRA-2908)
0.8.1
@ -232,6 +238,7 @@
* add a server-wide cap on measured memtable memory usage and aggressively
flush to keep under that threshold (CASSANDRA-2006)
* add unified UUIDType (CASSANDRA-2233)
* add off-heap row cache support (CASSANDRA-1969)
0.7.5

9
conf/cassandra.yaml
View File

@ -80,13 +80,16 @@ saved_caches_directory: /var/lib/cassandra/saved_caches
# commitlog_sync may be either "periodic" or "batch."
# When in batch mode, Cassandra won't ack writes until the commit log
# has been fsynced to disk. It will wait up to
# CommitLogSyncBatchWindowInMS milliseconds for other writes, before
# commitlog_sync_batch_window_in_ms milliseconds for other writes, before
# performing the sync.
commitlog_sync: periodic
#
# commitlog_sync: batch
# commitlog_sync_batch_window_in_ms: 50
#
# the other option is "periodic" where writes may be acked immediately
# and the CommitLog is simply synced every commitlog_sync_period_in_ms
# milliseconds.
commitlog_sync: periodic
commitlog_sync_period_in_ms: 10000
# any class that implements the SeedProvider interface and has a constructor that takes a Map<String, String> of

3
examples/client_only/conf/cassandra.yaml
View File

@ -77,9 +77,6 @@ commitlog_directory: /var/lib/cassandra/commitlog
# saved caches
saved_caches_directory: /var/lib/cassandra/saved_caches
# Size to allow commitlog to grow to before creating a new segment
commitlog_rotation_threshold_in_mb: 128
# commitlog_sync may be either "periodic" or "batch."
# When in batch mode, Cassandra won't ack writes until the commit log
# has been fsynced to disk. It will wait up to

80
src/java/org/apache/cassandra/cli/CliClient.java
View File

@ -350,7 +350,6 @@ public class CliClient
Tree columnFamilySpec = statement.getChild(0);
String key = CliCompiler.getKey(columnFamilySpec);
String columnFamily = CliCompiler.getColumnFamily(columnFamilySpec, keyspacesMap.get(keySpace).cf_defs);
int columnSpecCnt = CliCompiler.numColumnSpecifiers(columnFamilySpec);
@ -358,14 +357,19 @@ public class CliClient
if (columnSpecCnt != 0)
{
byte[] superColumn = columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), columnFamily);
Tree columnTree = columnFamilySpec.getChild(2);
byte[] superColumn = (columnTree.getType() == CliParser.FUNCTION_CALL)
? convertValueByFunction(columnTree, null, null).array()
: columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), columnFamily);
colParent = new ColumnParent(columnFamily).setSuper_column(superColumn);
}
SliceRange range = new SliceRange(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, Integer.MAX_VALUE);
SlicePredicate predicate = new SlicePredicate().setColumn_names(null).setSlice_range(range);
int count = thriftClient.get_count(ByteBufferUtil.bytes(key), colParent, predicate, consistencyLevel);
int count = thriftClient.get_count(getKeyAsBytes(columnFamily, columnFamilySpec.getChild(1)), colParent, predicate, consistencyLevel);
sessionState.out.printf("%d columns%n", count);
}
@ -377,13 +381,14 @@ public class CliClient
Tree columnFamilySpec = statement.getChild(0);
String key = CliCompiler.getKey(columnFamilySpec);
String columnFamily = CliCompiler.getColumnFamily(columnFamilySpec, keyspacesMap.get(keySpace).cf_defs);
CfDef cfDef = getCfDef(columnFamily);
ByteBuffer key = getKeyAsBytes(columnFamily, columnFamilySpec.getChild(1));
int columnSpecCnt = CliCompiler.numColumnSpecifiers(columnFamilySpec);
byte[] superColumnName = null;
byte[] columnName = null;
CfDef cfDef = getCfDef(columnFamily);
boolean isSuper = cfDef.column_type.equals("Super");
if ((columnSpecCnt < 0) || (columnSpecCnt > 2))
@ -391,20 +396,42 @@ public class CliClient
sessionState.out.println("Invalid row, super column, or column specification.");
return;
}
Tree columnTree = (columnSpecCnt >= 1)
? columnFamilySpec.getChild(2)
: null;
Tree subColumnTree = (columnSpecCnt == 2)
? columnFamilySpec.getChild(3)
: null;
if (columnSpecCnt == 1)
{
// table.cf['key']['column']
assert columnTree != null;
byte[] columnNameBytes = (columnTree.getType() == CliParser.FUNCTION_CALL)
? convertValueByFunction(columnTree, null, null).array()
: columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), cfDef);
if (isSuper)
superColumnName = columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), cfDef);
superColumnName = columnNameBytes;
else
columnName = columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), cfDef);
columnName = columnNameBytes;
}
else if (columnSpecCnt == 2)
{
assert columnTree != null;
assert subColumnTree != null;
// table.cf['key']['column']['column']
superColumnName = columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), cfDef);
columnName = subColumnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 1), cfDef);
superColumnName = (columnTree.getType() == CliParser.FUNCTION_CALL)
? convertValueByFunction(columnTree, null, null).array()
: columnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 0), cfDef);
columnName = (subColumnTree.getType() == CliParser.FUNCTION_CALL)
? convertValueByFunction(subColumnTree, null, null).array()
: subColumnNameAsByteArray(CliCompiler.getColumn(columnFamilySpec, 1), cfDef);
}
ColumnPath path = new ColumnPath(columnFamily);
@ -416,12 +443,11 @@ public class CliClient
if (isCounterCF(cfDef))
{
thriftClient.remove_counter(ByteBufferUtil.bytes(key), path, consistencyLevel);
thriftClient.remove_counter(key, path, consistencyLevel);
}
else
{
thriftClient.remove(ByteBufferUtil.bytes(key), path,
FBUtilities.timestampMicros(), consistencyLevel);
thriftClient.remove(key, path, FBUtilities.timestampMicros(), consistencyLevel);
}
sessionState.out.println(String.format("%s removed.", (columnSpecCnt == 0) ? "row" : "column"));
}
@ -1050,11 +1076,15 @@ public class CliClient
return;
String cfName = CliCompiler.getColumnFamily(statement, keyspacesMap.get(keySpace).cf_defs);
// first child is a column family name
CfDef cfDef = getCfDef(cfName);
try
{
// request correct cfDef from the server
CfDef cfDef = getCfDef(thriftClient.describe_keyspace(this.keySpace), cfName);
if (cfDef == null)
throw new RuntimeException("Column Family " + cfName + " was not found in the current keyspace.");
String mySchemaVersion = thriftClient.system_update_column_family(updateCfDefAttributes(statement, cfDef));
sessionState.out.println(mySchemaVersion);
validateSchemaIsSettled(mySchemaVersion);
@ -1202,7 +1232,7 @@ public class CliClient
cfDef.setKey_cache_save_period_in_seconds(Integer.parseInt(mValue));
break;
case DEFAULT_VALIDATION_CLASS:
cfDef.setDefault_validation_class(mValue);
cfDef.setDefault_validation_class(CliUtils.unescapeSQLString(mValue));
break;
case MIN_COMPACTION_THRESHOLD:
cfDef.setMin_compaction_threshold(Integer.parseInt(mValue));
@ -1252,6 +1282,9 @@ public class CliClient
String version = thriftClient.system_drop_keyspace(keyspaceName);
sessionState.out.println(version);
validateSchemaIsSettled(version);
if (keyspaceName.equals(keySpace)) //we just deleted the keyspace we were authenticated too
keySpace = null;
}
/**
@ -1898,7 +1931,18 @@ public class CliClient
{
return getCfDef(this.keySpace, columnFamilyName);
}
private CfDef getCfDef(KsDef keyspace, String columnFamilyName)
{
for (CfDef cfDef : keyspace.cf_defs)
{
if (cfDef.name.equals(columnFamilyName))
return cfDef;
}
return null;
}
/**
* Used to parse meta tree and compile meta attributes into List<ColumnDef>
* @param cfDef - column family definition

3
src/java/org/apache/cassandra/cli/CliUserHelp.java
View File

@ -19,9 +19,6 @@ package org.apache.cassandra.cli;
import java.util.List;
/**
* @author Pavel A. Yaskevich
*/
public class CliUserHelp
{
public String banner;

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

@ -101,7 +101,7 @@ public class DatabaseDescriptor
try
{
url = new URL(configUrl);
url.openStream(); // catches well-formed but bogus URLs
url.openStream().close(); // catches well-formed but bogus URLs
}
catch (Exception e)
{

2
src/java/org/apache/cassandra/db/TruncateResponse.java
View File

@ -31,8 +31,6 @@ import org.apache.cassandra.utils.FBUtilities;
/**
* This message is sent back the truncate operation and basically specifies if
* the truncate succeeded.
*
* @author rantav@gmail.com
*/
public class TruncateResponse
{

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

@ -31,9 +31,6 @@ import org.apache.cassandra.utils.FBUtilities;
/**
* A truncate operation descriptor
*
* @author rantav@gmail.com
*
*/
public class Truncation implements MessageProducer
{

16
src/java/org/apache/cassandra/db/compaction/CompactionController.java
View File

@ -46,6 +46,7 @@ public class CompactionController
public final boolean isMajor;
public final int gcBefore;
private int throttleResolution;
public CompactionController(ColumnFamilyStore cfs, Collection<SSTableReader> sstables, int gcBefore, boolean forceDeserialize)
{
@ -55,15 +56,26 @@ public class CompactionController
this.gcBefore = gcBefore;
this.forceDeserialize = forceDeserialize;
isMajor = cfs.isCompleteSSTables(this.sstables);
// how many rows we expect to compact in 100ms
long rowSize = cfs.getMeanRowSize();
int rowsPerSecond = rowSize > 0
? (int) (DatabaseDescriptor.getCompactionThroughputMbPerSec() * 1024 * 1024 / rowSize)
: 1000;
throttleResolution = rowsPerSecond / 10;
if (throttleResolution <= 0)
throttleResolution = 1;
}
public int getThrottleResolution()
{
return throttleResolution;
}
/** @return the keyspace name */
public String getKeyspace()
{
return cfs.table.name;
}
/** @return the column family name */
public String getColumnFamily()
{
return cfs.columnFamily;

13
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -121,9 +121,7 @@ implements CloseableIterator<AbstractCompactedRow>, CompactionInfo.Holder
int newTarget = totalBytesPerMS /
Math.max(1, CompactionManager.instance.getActiveCompactions());
if (newTarget != targetBytesPerMS)
logger.info(String.format("%s now compacting at %d bytes/ms.",
this,
newTarget));
logger.debug("{} now compacting at {} bytes/ms.", this, newTarget);
targetBytesPerMS = newTarget;
// the excess bytes that were compacted in this period
@ -136,7 +134,14 @@ implements CloseableIterator<AbstractCompactedRow>, CompactionInfo.Holder
if (logger.isTraceEnabled())
logger.trace(String.format("Compacted %d bytes in %d ms: throttling for %d ms",
bytesSinceLast, msSinceLast, timeToDelay));
try { Thread.sleep(timeToDelay); } catch (InterruptedException e) { throw new AssertionError(e); }
try
{
Thread.sleep(timeToDelay);
}
catch (InterruptedException e)
{
throw new AssertionError(e);
}
}
bytesAtLastDelay = bytesRead;
timeAtLastDelay = System.currentTimeMillis();

5
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -194,6 +194,11 @@ public class Gossiper implements IFailureDetectionEventListener
versions.put(address, version);
}
public void resetVersion(InetAddress endpoint)
{
versions.remove(endpoint);
}
public Integer getVersion(InetAddress address)
{
Integer v = versions.get(address);

37
src/java/org/apache/cassandra/hadoop/ColumnFamilyInputFormat.java
View File

@ -35,8 +35,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.TokenRange;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.hadoop.conf.Configuration;
@ -102,10 +105,44 @@ public class ColumnFamilyInputFormat extends InputFormat<ByteBuffer, SortedMap<B
try
{
List<Future<List<InputSplit>>> splitfutures = new ArrayList<Future<List<InputSplit>>>();
KeyRange jobKeyRange = ConfigHelper.getInputKeyRange(conf);
IPartitioner partitioner = null;
Range jobRange = null;
if (jobKeyRange != null)
{
partitioner = ConfigHelper.getPartitioner(context.getConfiguration());
assert partitioner.preservesOrder() : "ConfigHelper.setInputKeyRange(..) can only be used with a order preserving paritioner";
assert jobKeyRange.start_key == null : "only start_token supported";
assert jobKeyRange.end_key == null : "only end_token supported";
jobRange = new Range(partitioner.getTokenFactory().fromString(jobKeyRange.start_token),
partitioner.getTokenFactory().fromString(jobKeyRange.end_token),
partitioner);
}
for (TokenRange range : masterRangeNodes)
{
if (jobRange == null)
{
// for each range, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(range, conf)));
}
else
{
Range dhtRange = new Range(partitioner.getTokenFactory().fromString(range.start_token),
partitioner.getTokenFactory().fromString(range.end_token),
partitioner);
if (dhtRange.intersects(jobRange))
{
Set<Range> intersections = dhtRange.intersectionWith(jobRange);
assert intersections.size() == 1 : "wrapping ranges not yet supported";
Range intersection = intersections.iterator().next();
range.start_token = partitioner.getTokenFactory().toString(intersection.left);
range.end_token = partitioner.getTokenFactory().toString(intersection.right);
// for each range, pick a live owner and ask it to compute bite-sized splits
splitfutures.add(executor.submit(new SplitCallable(range, conf)));
}
}
}
// wait until we have all the results back

1
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordWriter.java
View File

@ -53,7 +53,6 @@ import org.apache.thrift.transport.TSocket;
* directly to a responsible endpoint.
* </p>
*
* @author Karthick Sankarachary
* @see ColumnFamilyOutputFormat
* @see OutputFormat
*

49
src/java/org/apache/cassandra/hadoop/ConfigHelper.java
View File

@ -22,6 +22,7 @@ package org.apache.cassandra.hadoop;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.TBinaryProtocol;
import org.apache.cassandra.utils.FBUtilities;
@ -42,6 +43,7 @@ public class ConfigHelper
private static final String INPUT_COLUMNFAMILY_CONFIG = "cassandra.input.columnfamily";
private static final String OUTPUT_COLUMNFAMILY_CONFIG = "cassandra.output.columnfamily";
private static final String INPUT_PREDICATE_CONFIG = "cassandra.input.predicate";
private static final String INPUT_KEYRANGE_CONFIG = "cassandra.input.keyRange";
private static final String OUTPUT_PREDICATE_CONFIG = "cassandra.output.predicate";
private static final String INPUT_SPLIT_SIZE_CONFIG = "cassandra.input.split.size";
private static final int DEFAULT_SPLIT_SIZE = 64 * 1024;
@ -195,6 +197,53 @@ public class ConfigHelper
return predicate;
}
/**
* Set the KeyRange to limit the rows.
* @param conf Job configuration you are about to run
*/
public static void setInputRange(Configuration conf, String startToken, String endToken)
{
KeyRange range = new KeyRange().setStart_token(startToken).setEnd_token(endToken);
conf.set(INPUT_KEYRANGE_CONFIG, keyRangeToString(range));
}
/** may be null if unset */
public static KeyRange getInputKeyRange(Configuration conf)
{
String str = conf.get(INPUT_KEYRANGE_CONFIG);
return null != str ? keyRangeFromString(str) : null;
}
private static String keyRangeToString(KeyRange keyRange)
{
assert keyRange != null;
TSerializer serializer = new TSerializer(new TBinaryProtocol.Factory());
try
{
return FBUtilities.bytesToHex(serializer.serialize(keyRange));
}
catch (TException e)
{
throw new RuntimeException(e);
}
}
private static KeyRange keyRangeFromString(String st)
{
assert st != null;
TDeserializer deserializer = new TDeserializer(new TBinaryProtocol.Factory());
KeyRange keyRange = new KeyRange();
try
{
deserializer.deserialize(keyRange, FBUtilities.hexToBytes(st));
}
catch (TException e)
{
throw new RuntimeException(e);
}
return keyRange;
}
public static String getInputKeyspace(Configuration conf)
{
return conf.get(INPUT_KEYSPACE_CONFIG);

4
src/java/org/apache/cassandra/io/sstable/SSTableLoader.java
View File

@ -101,7 +101,7 @@ public class SSTableLoader
public LoaderFuture stream(Set<InetAddress> toIgnore) throws IOException
{
client.init();
client.init(keyspace);
Collection<SSTableReader> sstables = openSSTables();
if (sstables.isEmpty())
@ -234,7 +234,7 @@ public class SSTableLoader
* This method is guaranted to be called before any other method of a
* client.
*/
public abstract void init();
public abstract void init(String keyspace);
/**
* Stop the client.

10
src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java
View File

@ -87,9 +87,8 @@ public abstract class AbstractReplicationStrategy
* we return a List to avoid an extra allocation when sorting by proximity later
* @param searchToken the token the natural endpoints are requested for
* @return a copy of the natural endpoints for the given token
* @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints
*/
public ArrayList<InetAddress> getNaturalEndpoints(Token searchToken) throws IllegalStateException
public ArrayList<InetAddress> getNaturalEndpoints(Token searchToken)
{
Token keyToken = TokenMetadata.firstToken(tokenMetadata.sortedTokens(), searchToken);
ArrayList<InetAddress> endpoints = getCachedEndpoints(keyToken);
@ -99,10 +98,6 @@ public abstract class AbstractReplicationStrategy
keyToken = TokenMetadata.firstToken(tokenMetadataClone.sortedTokens(), searchToken);
endpoints = new ArrayList<InetAddress>(calculateNaturalEndpoints(searchToken, tokenMetadataClone));
cacheEndpoint(keyToken, endpoints);
// calculateNaturalEndpoints should have checked this already, this is a safety
assert getReplicationFactor() <= endpoints.size() : String.format("endpoints %s generated for RF of %s",
Arrays.toString(endpoints.toArray()),
getReplicationFactor());
}
return new ArrayList<InetAddress>(endpoints);
@ -115,9 +110,8 @@ public abstract class AbstractReplicationStrategy
*
* @param searchToken the token the natural endpoints are requested for
* @return a copy of the natural endpoints for the given token
* @throws IllegalStateException if the number of requested replicas is greater than the number of known endpoints
*/
public abstract List<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata) throws IllegalStateException;
public abstract List<InetAddress> calculateNaturalEndpoints(Token searchToken, TokenMetadata tokenMetadata);
public IWriteResponseHandler getWriteResponseHandler(Collection<InetAddress> writeEndpoints,
Multimap<InetAddress, InetAddress> hintedEndpoints,

3
src/java/org/apache/cassandra/locator/NetworkTopologyStrategy.java
View File

@ -120,9 +120,6 @@ public class NetworkTopologyStrategy extends AbstractReplicationStrategy
dcEndpoints.add(endpoint);
}
if (dcEndpoints.size() < dcReplicas)
throw new IllegalStateException(String.format("datacenter (%s) has no more endpoints, (%s) replicas still needed",
dcName, dcReplicas - dcEndpoints.size()));
if (logger.isDebugEnabled())
logger.debug("{} endpoints in datacenter {} for token {} ",
new Object[] { StringUtils.join(dcEndpoints, ","), dcName, searchToken});

3
src/java/org/apache/cassandra/locator/OldNetworkTopologyStrategy.java
View File

@ -96,9 +96,6 @@ public class OldNetworkTopologyStrategy extends AbstractReplicationStrategy
if (!endpoints.contains(metadata.getEndpoint(t)))
endpoints.add(metadata.getEndpoint(t));
}
if (endpoints.size() < replicas)
throw new IllegalStateException(String.format("replication factor (%s) exceeds number of endpoints (%s)", replicas, endpoints.size()));
}
return endpoints;

4
src/java/org/apache/cassandra/locator/SimpleStrategy.java
View File

@ -56,10 +56,6 @@ public class SimpleStrategy extends AbstractReplicationStrategy
{
endpoints.add(metadata.getEndpoint(iter.next()));
}
if (endpoints.size() < replicas)
throw new IllegalStateException(String.format("replication factor (%s) exceeds number of endpoints (%s)", replicas, endpoints.size()));
return endpoints;
}

4
src/java/org/apache/cassandra/net/OutboundTcpConnection.java
View File

@ -30,6 +30,7 @@ import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -138,6 +139,9 @@ public class OutboundTcpConnection extends Thread
output = null;
socket = null;
}
// when we see the node again, try to connect at the most recent protocol we know about
Gossiper.instance.resetVersion(endpoint);
}
private ByteBuffer take()

26
src/java/org/apache/cassandra/service/AbstractCassandraDaemon.java
View File

@ -73,12 +73,30 @@ public abstract class AbstractCassandraDaemon implements CassandraDaemon
}
catch (MalformedURLException ex)
{
// load from the classpath.
// then try loading from the classpath.
configLocation = AbstractCassandraDaemon.class.getClassLoader().getResource(config);
if (configLocation == null)
throw new RuntimeException("Couldn't figure out log4j configuration.");
}
PropertyConfigurator.configureAndWatch(configLocation.getFile(), 10000);
if (configLocation == null)
throw new RuntimeException("Couldn't figure out log4j configuration: "+config);
// Now convert URL to a filename
String configFileName = null;
try
{
// first try URL.getFile() which works for opaque URLs (file:foo) and paths without spaces
configFileName = configLocation.getFile();
File configFile = new File(configFileName);
// then try alternative approach which works for all hierarchical URLs with or without spaces
if (!configFile.exists())
configFileName = new File(configLocation.toURI()).getCanonicalPath();
}
catch (Exception e)
{
throw new RuntimeException("Couldn't convert log4j configuration location to a valid file", e);
}
PropertyConfigurator.configureAndWatch(configFileName, 10000);
org.apache.log4j.Logger.getLogger(AbstractCassandraDaemon.class).info("Logging initialized");
}

2
src/java/org/apache/cassandra/service/EmbeddedCassandraService.java
View File

@ -45,8 +45,6 @@ import org.apache.thrift.transport.TTransportException;
cassandra.start();
* </pre>
* @author Ran Tavory (rantav@gmail.com)
*
*/
public class EmbeddedCassandraService
{

30
src/java/org/apache/cassandra/service/StorageService.java
View File

@ -1296,6 +1296,26 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
return stringify(Gossiper.instance.getUnreachableMembers());
}
public String[] getAllDataFileLocations()
{
return DatabaseDescriptor.getAllDataFileLocations();
}
public String[] getAllDataFileLocationsForTable(String table)
{
return DatabaseDescriptor.getAllDataFileLocationsForTable(table);
}
public String getCommitLogLocation()
{
return DatabaseDescriptor.getCommitLogLocation();
}
public String getSavedCachesLocation()
{
return DatabaseDescriptor.getSavedCachesLocation();
}
private List<String> stringify(Iterable<InetAddress> endpoints)
{
List<String> stringEndpoints = new ArrayList<String>();
@ -2448,7 +2468,15 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
SSTableLoader.Client client = new SSTableLoader.Client()
{
public void init() {}
public void init(String keyspace)
{
for (Map.Entry<Range, List<InetAddress>> entry : StorageService.instance.getRangeToAddressMap(keyspace).entrySet())
{
Range range = entry.getKey();
for (InetAddress endpoint : entry.getValue())
addRangeForEndpoint(range, endpoint);
}
}
public boolean validateColumnFamily(String keyspace, String cfName)
{

25
src/java/org/apache/cassandra/service/StorageServiceMBean.java
View File

@ -85,6 +85,31 @@ public interface StorageServiceMBean
*/
public String getReleaseVersion();
/**
* Get the list of all data file locations from conf
* @return String array of all locations
*/
public String[] getAllDataFileLocations();
/**
* Get the list of data file locations for a given keyspace
* @param keyspace the keyspace to get locatiosn for.
* @return String array of all locations
*/
public String[] getAllDataFileLocationsForTable(String table);
/**
* Get location of the commit log
* @return a string path
*/
public String getCommitLogLocation();
/**
* Get location of the saved caches dir
* @return a string path
*/
public String getSavedCachesLocation();
/**
* Retrieve a map of range to end points that describe the ring topology
* of a Cassandra cluster.

4
src/java/org/apache/cassandra/service/WriteResponseHandler.java
View File

@ -85,9 +85,9 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler
case THREE:
return 3;
case QUORUM:
return (writeEndpoints.size() / 2) + 1;
return (Table.open(table).getReplicationStrategy().getReplicationFactor() / 2) + 1;
case ALL:
return writeEndpoints.size();
return Table.open(table).getReplicationStrategy().getReplicationFactor();
default:
throw new UnsupportedOperationException("invalid consistency level: " + consistencyLevel.toString());
}

1
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -960,6 +960,7 @@ public class CassandraServer implements Cassandra.Iface
CFMetaData oldCfm = DatabaseDescriptor.getCFMetaData(CFMetaData.getId(cf_def.keyspace, cf_def.name));
if (oldCfm == null)
throw new InvalidRequestException("Could not find column family definition to modify.");
ThriftValidation.validateCfDef(cf_def, oldCfm);
validateSchemaAgreement();
try

9
src/java/org/apache/cassandra/tools/BulkLoader.java
View File

@ -24,6 +24,7 @@ import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@ -57,7 +58,7 @@ public class BulkLoader
LoaderOptions options = LoaderOptions.parseArgs(args);
try
{
SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.directory.getName(), options), options);
SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options), options);
SSTableLoader.LoaderFuture future = loader.stream(options.ignores);
if (options.noProgress)
@ -164,18 +165,16 @@ public class BulkLoader
static class ExternalClient extends SSTableLoader.Client
{
private final String keyspace;
private final Map<String, Set<String>> knownCfs = new HashMap<String, Set<String>>();
private final SSTableLoader.OutputHandler outputHandler;
public ExternalClient(String keyspace, SSTableLoader.OutputHandler outputHandler)
public ExternalClient(SSTableLoader.OutputHandler outputHandler)
{
super();
this.keyspace = keyspace;
this.outputHandler = outputHandler;
}
public void init()
public void init(String keyspace)
{
outputHandler.output(String.format("Starting client (and waiting %d seconds for gossip) ...", StorageService.RING_DELAY / 1000));
try

9
src/java/org/apache/cassandra/tools/NodeCmd.java
View File

@ -720,6 +720,15 @@ public class NodeCmd
e.printStackTrace();
System.exit(3);
}
private static void complainNonzeroArgs(String[] args, NodeCommand cmd)
{
if (args.length > 0) {
System.err.println("Too many arguments for command '"+cmd.toString()+"'.");
printUsage();
System.exit(1);
}
}
private static void handleSnapshots(NodeCommand nc, String tag, String[] cmdArgs, NodeProbe probe) throws InterruptedException, IOException
{

4
src/resources/org/apache/cassandra/cli/CliHelp.yaml
View File

@ -433,7 +433,7 @@ commands:
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- keys_cache_save_period: Duration in seconds after which Cassandra should
- key_cache_save_period: Duration in seconds after which Cassandra should
safe the keys cache. Caches are saved to saved_caches_directory as
specified in conf/Cassandra.yaml. Default is 14400 or 4 hours.
@ -674,7 +674,7 @@ commands:
store the whole values of its rows, so it is extremely space-intensive.
It's best to only use the row cache if you have hot rows or static rows.
- keys_cache_save_period: Duration in seconds after which Cassandra should
- key_cache_save_period: Duration in seconds after which Cassandra should
safe the keys cache. Caches are saved to saved_caches_directory as
specified in conf/Cassandra.yaml. Default is 14400 or 4 hours.

6
test/unit/org/apache/cassandra/cli/CliTest.java
View File

@ -54,6 +54,8 @@ public class CliTest extends CleanupHelper
"get CF1 where world2 = long(15);",
"get cF1 where world2 = long(15);",
"get Cf1 where world2 = long(15);",
"del CF1[utf8('hello')][utf8('world')];",
"del CF1[hello][world2];",
"set CF1['hello'][time_spent_uuid] = timeuuid(a8098c1a-f86e-11da-bd1a-00112444be1e);",
"create column family CF2 with comparator=IntegerType;",
"assume CF2 keys as utf8;",
@ -132,6 +134,10 @@ public class CliTest extends CleanupHelper
"set sCf1['hello'][1][9999] = 938;",
"set sCf1['hello'][1][9999] = 938 with ttl = 30;",
"set sCf1['hello'][1][9999] = 938 with ttl = 560;",
"count sCf1[hello];",
"count sCf1[utf8('hello')];",
"count sCf1[utf8('hello')][integer(1)];",
"count sCF1[hello][1];",
"list sCf1;",
"del SCF1['hello'][1][9999];",
"assume sCf1 comparator as utf8;",

3
test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java
View File

@ -36,9 +36,6 @@ import org.apache.cassandra.utils.ByteBufferUtil;
/**
* Test for the truncate operation.
*
* @author Ran Tavory (rantav@gmail.com)
*
*/
public class RecoveryManagerTruncateTest extends CleanupHelper
{