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@1126485 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-05-23 13:30:48 +00:00
parent 3de3c21dc6 ed7f7e8f57
commit 70c0ed88e2
27 changed files with 881 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -17,6 +17,7 @@
* Improve forceDeserialize/getCompactedRow encapsulation (CASSANDRA-2659)
* Assert ranges are not overlapping in AB.normalize (CASSANDRA-2641)
* Don't write CounterUpdateColumn to disk in tests (CASSANDRA-2650)
* Add sstable bulk loading utility (CASSANDRA-1278)
0.8.0-final
@ -24,6 +25,7 @@
* add ant generate-cql-html target (CASSANDRA-2526)
* update CQL consistency levels (CASSANDRA-2566)
* debian packaging fixes (CASSANDRA-2481, 2647)
* fix UUIDType, IntegerType for direct buffers (CASSANDRA-2682, 2684)
0.8.0-rc1
@ -114,6 +116,7 @@
* cli no longer divides read_repair_chance by 100 (CASSANDRA-2458)
* made CompactionInfo.getTaskType return an enum (CASSANDRA-2482)
* add a server-wide cap on measured memtable memory usage (CASSANDRA-2006)
* add unified UUIDType (CASSANDRA-2233)
0.7.5

1
NEWS.txt
View File

@ -17,6 +17,7 @@ Upgrading
applications using these classes will need to be updated accordingly.
- The loadbalance command has been removed from nodetool. For similar
behavior, decommission then rebootstrap with empty initial_token.
- Thrift unframed mode has been removed.
Features
--------

50
bin/sstableloader Normal file
View File

@ -0,0 +1,50 @@
#!/bin/sh
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
for include in /usr/share/cassandra/cassandra.in.sh \
/usr/local/share/cassandra/cassandra.in.sh \
/opt/cassandra/cassandra.in.sh \
~/.cassandra.in.sh \
`dirname $0`/cassandra.in.sh; do
if [ -r $include ]; then
. $include
break
fi
done
elif [ -r $CASSANDRA_INCLUDE ]; then
. $CASSANDRA_INCLUDE
fi
# Use JAVA_HOME if set, otherwise look for java in PATH
if [ -x $JAVA_HOME/bin/java ]; then
JAVA=$JAVA_HOME/bin/java
else
JAVA=`which java`
fi
if [ -z $CLASSPATH ]; then
echo "You must set the CLASSPATH var" >&2
exit 1
fi
$JAVA -ea -cp $CLASSPATH -Xmx256M \
-Dlog4j.configuration=log4j-tools.properties \
org.apache.cassandra.tools.BulkLoader "$@"
# vi:ai sw=4 ts=4 tw=0 et

2
drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java
View File

@ -26,7 +26,7 @@ import org.apache.cassandra.thrift.Column;
import java.nio.ByteBuffer;
class TypedColumn
public class TypedColumn
{
private final Column rawColumn;

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

@ -504,7 +504,7 @@ public class CliClient
{
try
{
return FBUtilities.getComparator(compareWith);
return TypeParser.parse(compareWith);
}
catch (ConfigurationException ce)
{

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

@ -29,18 +29,13 @@ import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.avro.util.Utf8;
import org.apache.cassandra.cache.ConcurrentLinkedHashCache;
import org.apache.cassandra.cache.ConcurrentLinkedHashCacheProvider;
import org.apache.cassandra.cache.IRowCacheProvider;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.migration.avro.ColumnDef;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.db.HintedHandOffManager;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.io.SerDeUtils;
import org.apache.cassandra.thrift.InvalidRequestException;
@ -360,11 +355,11 @@ public final class CFMetaData
try
{
comparator = DatabaseDescriptor.getComparator(cf.comparator_type.toString());
comparator = TypeParser.parse(cf.comparator_type.toString());
if (cf.subcomparator_type != null)
subcolumnComparator = DatabaseDescriptor.getComparator(cf.subcomparator_type);
validator = DatabaseDescriptor.getComparator(cf.default_validation_class);
keyValidator = DatabaseDescriptor.getComparator(cf.key_validation_class);
subcolumnComparator = TypeParser.parse(cf.subcomparator_type);
validator = TypeParser.parse(cf.default_validation_class);
keyValidator = TypeParser.parse(cf.key_validation_class);
}
catch (Exception ex)
{
@ -645,8 +640,8 @@ public final class CFMetaData
CFMetaData newCFMD = new CFMetaData(cf_def.keyspace,
cf_def.name,
cfType,
DatabaseDescriptor.getComparator(cf_def.comparator_type),
cf_def.subcomparator_type == null ? null : DatabaseDescriptor.getComparator(cf_def.subcomparator_type));
TypeParser.parse(cf_def.comparator_type),
cf_def.subcomparator_type == null ? null : TypeParser.parse(cf_def.subcomparator_type));
if (cf_def.isSetGc_grace_seconds()) { newCFMD.gcGraceSeconds(cf_def.gc_grace_seconds); }
if (cf_def.isSetMin_compaction_threshold()) { newCFMD.minCompactionThreshold(cf_def.min_compaction_threshold); }
@ -659,15 +654,15 @@ public final class CFMetaData
if (cf_def.isSetMerge_shards_chance()) { newCFMD.mergeShardsChance(cf_def.merge_shards_chance); }
if (cf_def.isSetRow_cache_provider()) { newCFMD.rowCacheProvider(FBUtilities.newCacheProvider(cf_def.row_cache_provider)); }
if (cf_def.isSetKey_alias()) { newCFMD.keyAlias(cf_def.key_alias); }
if (cf_def.isSetKey_validation_class()) { newCFMD.keyValidator(DatabaseDescriptor.getComparator(cf_def.key_validation_class)); }
if (cf_def.isSetKey_validation_class()) { newCFMD.keyValidator(TypeParser.parse(cf_def.key_validation_class)); }
return newCFMD.comment(cf_def.comment)
.rowCacheSize(cf_def.row_cache_size)
.keyCacheSize(cf_def.key_cache_size)
.readRepairChance(cf_def.read_repair_chance)
.replicateOnWrite(cf_def.replicate_on_write)
.defaultValidator(DatabaseDescriptor.getComparator(cf_def.default_validation_class))
.keyValidator(DatabaseDescriptor.getComparator(cf_def.key_validation_class))
.defaultValidator(TypeParser.parse(cf_def.default_validation_class))
.keyValidator(TypeParser.parse(cf_def.key_validation_class))
.columnMetadata(ColumnDefinition.fromColumnDef(cf_def.column_metadata));
}
@ -683,7 +678,7 @@ public final class CFMetaData
throw new ConfigurationException("names do not match.");
if (!cf_def.column_type.toString().equals(cfType.name()))
throw new ConfigurationException("types do not match.");
if (comparator != DatabaseDescriptor.getComparator(cf_def.comparator_type))
if (comparator != TypeParser.parse(cf_def.comparator_type))
throw new ConfigurationException("comparators do not match.");
if (cf_def.subcomparator_type == null || cf_def.subcomparator_type.equals(""))
{
@ -691,7 +686,7 @@ public final class CFMetaData
throw new ConfigurationException("subcolumncomparators do not match.");
// else, it's null and we're good.
}
else if (subcolumnComparator != DatabaseDescriptor.getComparator(cf_def.subcomparator_type))
else if (subcolumnComparator != TypeParser.parse(cf_def.subcomparator_type))
throw new ConfigurationException("subcolumncomparators do not match.");
validateMinMaxCompactionThresholds(cf_def);
@ -704,8 +699,8 @@ public final class CFMetaData
readRepairChance = cf_def.read_repair_chance;
replicateOnWrite = cf_def.replicate_on_write;
gcGraceSeconds = cf_def.gc_grace_seconds;
defaultValidator = DatabaseDescriptor.getComparator(cf_def.default_validation_class);
keyValidator = DatabaseDescriptor.getComparator(cf_def.key_validation_class);
defaultValidator = TypeParser.parse(cf_def.default_validation_class);
keyValidator = TypeParser.parse(cf_def.key_validation_class);
minCompactionThreshold = cf_def.min_compaction_threshold;
maxCompactionThreshold = cf_def.max_compaction_threshold;
rowCacheSavePeriodInSeconds = cf_def.row_cache_save_period_in_seconds;
@ -741,14 +736,14 @@ public final class CFMetaData
ColumnDefinition oldDef = column_metadata.get(def.name);
if (oldDef == null)
continue;
oldDef.setValidator(DatabaseDescriptor.getComparator(def.validation_class));
oldDef.setValidator(TypeParser.parse(def.validation_class));
oldDef.setIndexType(def.index_type == null ? null : org.apache.cassandra.thrift.IndexType.valueOf(def.index_type.name()));
oldDef.setIndexName(def.index_name == null ? null : def.index_name.toString());
}
// add the new ones coming in.
for (org.apache.cassandra.db.migration.avro.ColumnDef def : toAdd)
{
AbstractType dValidClass = DatabaseDescriptor.getComparator(def.validation_class);
AbstractType dValidClass = TypeParser.parse(def.validation_class);
ColumnDefinition cd = new ColumnDefinition(def.name,
dValidClass,
def.index_type == null ? null : org.apache.cassandra.thrift.IndexType.valueOf(def.index_type.toString()),
@ -972,7 +967,7 @@ public final class CFMetaData
public static void validateAliasCompares(org.apache.cassandra.db.migration.avro.CfDef cf_def) throws ConfigurationException
{
AbstractType comparator = DatabaseDescriptor.getComparator(cf_def.comparator_type);
AbstractType comparator = TypeParser.parse(cf_def.comparator_type);
if (cf_def.key_alias != null)
comparator.validate(cf_def.key_alias);
}

7
src/java/org/apache/cassandra/config/ColumnDefinition.java
View File

@ -26,6 +26,7 @@ import java.util.*;
import org.apache.avro.util.Utf8;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -93,7 +94,7 @@ public class ColumnDefinition
String index_name = cd.index_name == null ? null : cd.index_name.toString();
try
{
AbstractType validatorType = DatabaseDescriptor.getComparator(cd.validation_class);
AbstractType validatorType = TypeParser.parse(cd.validation_class);
return new ColumnDefinition(cd.name, validatorType, index_type, index_name);
}
catch (ConfigurationException e)
@ -104,14 +105,14 @@ public class ColumnDefinition
public static ColumnDefinition fromColumnDef(ColumnDef thriftColumnDef) throws ConfigurationException
{
AbstractType validatorType = DatabaseDescriptor.getComparator(thriftColumnDef.validation_class);
AbstractType validatorType = TypeParser.parse(thriftColumnDef.validation_class);
return new ColumnDefinition(ByteBufferUtil.clone(thriftColumnDef.name), validatorType, thriftColumnDef.index_type, thriftColumnDef.index_name);
}
public static ColumnDefinition fromColumnDef(org.apache.cassandra.db.migration.avro.ColumnDef avroColumnDef) throws ConfigurationException
{
validateIndexType(avroColumnDef);
AbstractType validatorType = DatabaseDescriptor.getComparator(avroColumnDef.validation_class);
AbstractType validatorType = TypeParser.parse(avroColumnDef.validation_class);
return new ColumnDefinition(avroColumnDef.name,
validatorType,
IndexType.valueOf(avroColumnDef.index_type == null ? D_COLDEF_INDEXTYPE : avroColumnDef.index_type.name()),

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

@ -25,7 +25,6 @@ import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.base.Charsets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -39,6 +38,7 @@ import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.db.DefsTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.db.migration.Migration;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.sstable.Descriptor;
@ -46,7 +46,6 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.*;
import org.apache.cassandra.scheduler.IRequestScheduler;
import org.apache.cassandra.scheduler.NoScheduler;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.yaml.snakeyaml.Loader;
@ -529,14 +528,6 @@ public class DatabaseDescriptor
return conf.thrift_framed_transport_size_in_mb * 1024 * 1024;
}
public static AbstractType getComparator(CharSequence compareWith) throws ConfigurationException
{
if (compareWith == null)
compareWith = "BytesType";
return FBUtilities.getComparator(compareWith.toString());
}
/**
* Creates all storage-related directories.
* @throws IOException when a disk problem is encountered.

10
src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
View File

@ -31,9 +31,9 @@ import java.util.Set;
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.db.ColumnFamilyType;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.utils.FBUtilities;
@ -203,7 +203,7 @@ public class CreateColumnFamilyStatement
{
ByteBuffer columnName = col.getKey().getByteBuffer(comparator);
String validatorClassName = comparators.containsKey(col.getValue()) ? comparators.get(col.getValue()) : col.getValue();
AbstractType<?> validator = DatabaseDescriptor.getComparator(validatorClassName);
AbstractType<?> validator = TypeParser.parse(validatorClassName);
columnDefs.put(columnName, new ColumnDefinition(columnName, validator, null, null));
}
catch (ConfigurationException e)
@ -242,7 +242,7 @@ public class CreateColumnFamilyStatement
String validatorString = (comparators.get(getPropertyString(KW_DEFAULTVALIDATION, "text")) != null)
? comparators.get(getPropertyString(KW_DEFAULTVALIDATION, "text"))
: getPropertyString(KW_DEFAULTVALIDATION, "text");
AbstractType<?> comparator = DatabaseDescriptor.getComparator(comparatorString);
AbstractType<?> comparator = TypeParser.parse(comparatorString);
newCFMD = new CFMetaData(keyspace,
name,
@ -256,7 +256,7 @@ public class CreateColumnFamilyStatement
.readRepairChance(getPropertyDouble(KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE))
.replicateOnWrite(getPropertyBoolean(KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE))
.gcGraceSeconds(getPropertyInt(KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
.defaultValidator(DatabaseDescriptor.getComparator(validatorString))
.defaultValidator(TypeParser.parse(validatorString))
.minCompactionThreshold(getPropertyInt(KW_MINCOMPACTIONTHRESHOLD, CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD))
.maxCompactionThreshold(getPropertyInt(KW_MAXCOMPACTIONTHRESHOLD, CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD))
.rowCacheSavePeriod(getPropertyInt(KW_ROWCACHESAVEPERIODSECS, CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS))
@ -266,7 +266,7 @@ public class CreateColumnFamilyStatement
.memOps(getPropertyDouble(KW_MEMTABLEOPSINMILLIONS, CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS))
.mergeShardsChance(0.0)
.columnMetadata(getColumns(comparator))
.keyValidator(DatabaseDescriptor.getComparator(comparators.get(getKeyType())))
.keyValidator(TypeParser.parse(comparators.get(getKeyType())))
.rowCacheProvider(FBUtilities.newCacheProvider(getPropertyString(KW_ROW_CACHE_PROVIDER, CFMetaData.DEFAULT_ROW_CACHE_PROVIDER)));
}
catch (ConfigurationException e)

4
src/java/org/apache/cassandra/cql/QueryProcessor.java
View File

@ -194,14 +194,14 @@ public class QueryProcessor
}
// if start key was set and relation was "greater than"
if (select.getKeyStart() != null && !select.includeStartKey())
if (select.getKeyStart() != null && !select.includeStartKey() && !rows.isEmpty())
{
if (rows.get(0).key.key.equals(startKey))
rows.remove(0);
}
// if finish key was set and relation was "less than"
if (select.getKeyFinish() != null && !select.includeFinishKey())
if (select.getKeyFinish() != null && !select.includeFinishKey() && !rows.isEmpty())
{
int lastIndex = rows.size() - 1;
if (rows.get(lastIndex).key.key.equals(finishKey))

2
src/java/org/apache/cassandra/db/marshal/IntegerType.java
View File

@ -133,7 +133,7 @@ public final class IntegerType extends AbstractType<BigInteger>
if (bytes.remaining() == 0)
return "empty";
return new java.math.BigInteger(TBaseHelper.byteBufferToByteArray(bytes)).toString(10);
return new java.math.BigInteger(ByteBufferUtil.getArray(bytes)).toString(10);
}
public String toString(BigInteger bi)

5
src/java/org/apache/cassandra/db/marshal/TypeParser.java
View File

@ -87,6 +87,11 @@ public class TypeParser
return type;
}
public static AbstractType parse(CharSequence compareWith) throws ConfigurationException
{
return parse(compareWith == null ? null : compareWith.toString());
}
/**
* Parse an AbstractType from current position of this parser.
*/

59
src/java/org/apache/cassandra/db/marshal/UUIDType.java
View File

@ -25,11 +25,10 @@ import java.nio.ByteBuffer;
import java.text.ParseException;
import java.util.UUID;
import org.apache.commons.lang.time.DateUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.lang.time.DateUtils;
/**
* Compares UUIDs using the following criteria:<br>
@ -39,9 +38,7 @@ import org.apache.cassandra.utils.UUIDGen;
* - compare timestamps if both are time-based<br>
* - compare lexically, unsigned msb-to-lsb comparison<br>
*
* @author edanuff
* @see "com.fasterxml.uuid.UUIDComparator"
*
*/
public class UUIDType extends AbstractUUIDType
{
@ -65,25 +62,13 @@ public class UUIDType extends AbstractUUIDType
return 1;
}
int s1 = b1.arrayOffset() + b1.position();
byte[] o1 = b1.array();
int s2 = b2.arrayOffset() + b2.position();
byte[] o2 = b2.array();
if (o1.length == s1)
{
return o2.length == s2 ? 0 : -1;
}
if (o2.length == s2)
{
return 1;
}
int s1 = b1.position();
int s2 = b2.position();
// Compare versions
int v1 = (o1[s1 + 6] >> 4) & 0x0f;
int v2 = (o2[s2 + 6] >> 4) & 0x0f;
int v1 = (b1.get(s1 + 6) >> 4) & 0x0f;
int v2 = (b2.get(s2 + 6) >> 4) & 0x0f;
if (v1 != v2)
{
@ -95,7 +80,7 @@ public class UUIDType extends AbstractUUIDType
if (v1 == 1)
{
// if both time-based, compare as timestamps
int c = compareTimestampBytes(s1, o1, s2, o2);
int c = compareTimestampBytes(b1, b2);
if (c != 0)
{
return c;
@ -109,10 +94,9 @@ public class UUIDType extends AbstractUUIDType
// Appendix A - Sample Implementation.
// Note: java.util.UUID.compareTo is not a lexical
// comparison
for (int i = 0; i < 16; i++)
{
int c = ((o1[s1 + i]) & 0xFF) - ((o2[s2 + i]) & 0xFF);
int c = ((b1.get(s1 + i)) & 0xFF) - ((b2.get(s2 + i)) & 0xFF);
if (c != 0)
{
return c;
@ -122,45 +106,54 @@ public class UUIDType extends AbstractUUIDType
return 0;
}
private static int compareTimestampBytes(int s1, byte[] o1, int s2,
byte[] o2)
private static int compareTimestampBytes(ByteBuffer o1, ByteBuffer o2)
{
int d = (o1[s1 + 6] & 0xF) - (o2[s2 + 6] & 0xF);
int o1Pos = o1.position();
int o2Pos = o2.position();
int d = (o1.get(o1Pos + 6) & 0xF) - (o2.get(o2Pos + 6) & 0xF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 7] & 0xFF) - (o2[s2 + 7] & 0xFF);
d = (o1.get(o1Pos + 7) & 0xFF) - (o2.get(o2Pos + 7) & 0xFF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 4] & 0xFF) - (o2[s2 + 4] & 0xFF);
d = (o1.get(o1Pos + 4) & 0xFF) - (o2.get(o2Pos + 4) & 0xFF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 5] & 0xFF) - (o2[s2 + 5] & 0xFF);
d = (o1.get(o1Pos + 5) & 0xFF) - (o2.get(o2Pos + 5) & 0xFF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 0] & 0xFF) - (o2[s2 + 0] & 0xFF);
d = (o1.get(o1Pos) & 0xFF) - (o2.get(o2Pos) & 0xFF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 1] & 0xFF) - (o2[s2 + 1] & 0xFF);
d = (o1.get(o1Pos + 1) & 0xFF) - (o2.get(o2Pos + 1) & 0xFF);
if (d != 0)
{
return d;
}
d = (o1[s1 + 2] & 0xFF) - (o2[s2 + 2] & 0xFF);
d = (o1.get(o1Pos + 2) & 0xFF) - (o2.get(o2Pos + 2) & 0xFF);
if (d != 0)
{
return d;
}
return (o1[s1 + 3] & 0xFF) - (o2[s2 + 3] & 0xFF);
return (o1.get(o1Pos + 3) & 0xFF) - (o2.get(o2Pos + 3) & 0xFF);
}
public UUID compose(ByteBuffer bytes)

5
src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java
View File

@ -33,6 +33,7 @@ import org.apache.cassandra.auth.SimpleAuthenticator;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.FBUtilities;
@ -198,8 +199,8 @@ public class ColumnFamilyRecordReader extends RecordReader<ByteBuffer, SortedMap
int idx = cfnames.indexOf(cfName);
CfDef cf_def = ks_def.cf_defs.get(idx);
comparator = FBUtilities.getComparator(cf_def.comparator_type);
subComparator = cf_def.subcomparator_type == null ? null : FBUtilities.getComparator(cf_def.subcomparator_type);
comparator = TypeParser.parse(cf_def.comparator_type);
subComparator = cf_def.subcomparator_type == null ? null : TypeParser.parse(cf_def.subcomparator_type);
}
catch (ConfigurationException e)
{

3
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -85,9 +85,10 @@ public abstract class SSTable
protected SSTable(Descriptor descriptor, Set<Component> components, CFMetaData metadata, ReplayPosition replayPosition, IPartitioner partitioner, EstimatedHistogram rowSizes, EstimatedHistogram columnCounts)
{
// In almost all cases, metadata shouldn't be null, but allowing null allows to create a mostly functional SSTable without
// full schema definition. SSTableLoader use that ability
assert descriptor != null;
assert components != null;
assert metadata != null;
assert replayPosition != null;
assert partitioner != null;
assert rowSizes != null;

266
src/java/org/apache/cassandra/io/sstable/SSTableLoader.java Normal file
View File

@ -0,0 +1,266 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.net.InetAddress;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.*;
import org.apache.cassandra.utils.Pair;
/**
* Cassandra SSTable bulk loader.
* Load an externally created sstable into a cluster.
*/
public class SSTableLoader
{
private final File directory;
private final String keyspace;
private final Client client;
private final OutputHandler outputHandler;
public SSTableLoader(File directory, Client client, OutputHandler outputHandler)
{
this.directory = directory;
this.keyspace = directory.getName();
this.client = client;
this.outputHandler = outputHandler;
}
private Collection<SSTableReader> openSSTables()
{
final List<SSTableReader> sstables = new LinkedList<SSTableReader>();
directory.list(new FilenameFilter()
{
public boolean accept(File dir, String name)
{
Pair<Descriptor, Component> p = SSTable.tryComponentFromFilename(dir, name);
Descriptor desc = p == null ? null : p.left;
if (p == null || !p.right.equals(Component.DATA) || desc.temporary)
return false;
if (!new File(desc.filenameFor(Component.PRIMARY_INDEX)).exists())
{
outputHandler.output(String.format("Skipping file %s because index is missing", name));
return false;
}
if (!client.validateColumnFamily(keyspace, desc.cfname))
{
outputHandler.output(String.format("Skipping file %s: column family %s.%s doesn't exist", name, keyspace, desc.cfname));
return false;
}
Set<Component> components = new HashSet<Component>();
components.add(Component.DATA);
components.add(Component.PRIMARY_INDEX);
try
{
sstables.add(SSTableReader.open(desc, components, null, StorageService.getPartitioner()));
}
catch (IOException e)
{
outputHandler.output(String.format("Skipping file %s, error opening it: %s", name, e.getMessage()));
}
return false;
}
});
return sstables;
}
public LoaderFuture stream() throws IOException
{
return stream(Collections.<InetAddress>emptySet());
}
public LoaderFuture stream(Set<InetAddress> toIgnore) throws IOException
{
client.init();
Collection<SSTableReader> sstables = openSSTables();
if (sstables.isEmpty())
{
outputHandler.output("No sstables to stream");
return new LoaderFuture(0);
}
Map<InetAddress, Collection<Range>> endpointToRanges = client.getEndpointToRangesMap();
outputHandler.output(String.format("Streaming revelant part of %sto %s", names(sstables), endpointToRanges.keySet()));
// There will be one streaming session by endpoint
LoaderFuture future = new LoaderFuture(endpointToRanges.size());
for (Map.Entry<InetAddress, Collection<Range>> entry : endpointToRanges.entrySet())
{
InetAddress remote = entry.getKey();
if (toIgnore.contains(remote))
{
future.latch.countDown();
continue;
}
Collection<Range> ranges = entry.getValue();
StreamOutSession session = StreamOutSession.create(keyspace, remote, new CountDownCallback(future.latch, remote));
StreamOut.transferSSTables(session, sstables, ranges, OperationType.BULK_LOAD);
future.setPendings(remote, session.getFiles());
}
return future;
}
public static class LoaderFuture implements Future<Void>
{
final CountDownLatch latch;
final Map<InetAddress, Collection<PendingFile>> pendingFiles;
private LoaderFuture(int request)
{
latch = new CountDownLatch(request);
pendingFiles = new HashMap<InetAddress, Collection<PendingFile>>();
}
private void setPendings(InetAddress remote, Collection<PendingFile> files)
{
pendingFiles.put(remote, new ArrayList(files));
}
public boolean cancel(boolean mayInterruptIfRunning)
{
throw new UnsupportedOperationException("Cancellation is not yet supported");
}
public Void get() throws InterruptedException
{
latch.await();
return null;
}
public Void get(long timeout, TimeUnit unit) throws InterruptedException
{
latch.await(timeout, unit);
return null;
}
public boolean isCancelled()
{
// For now, cancellation is not supported, maybe one day...
return false;
}
public boolean isDone()
{
return latch.getCount() == 0;
}
public Map<InetAddress, Collection<PendingFile>> getPendingFiles()
{
return pendingFiles;
}
}
private String names(Collection<SSTableReader> sstables)
{
StringBuilder builder = new StringBuilder();
for (SSTableReader sstable : sstables)
builder.append(sstable.descriptor.filenameFor(Component.DATA)).append(" ");
return builder.toString();
}
private class CountDownCallback implements Runnable
{
private final InetAddress endpoint;
private final CountDownLatch latch;
CountDownCallback(CountDownLatch latch, InetAddress endpoint)
{
this.latch = latch;
this.endpoint = endpoint;
}
public void run()
{
latch.countDown();
outputHandler.debug(String.format("Streaming session to %s completed (waiting on %d outstanding sessions)", endpoint, latch.getCount()));
// There could be race with stop being called twice but it should be ok
if (latch.getCount() == 0)
client.stop();
}
}
public interface OutputHandler
{
// called when an important info need to be displayed
public void output(String msg);
// called when a less important info need to be displayed
public void debug(String msg);
}
public static abstract class Client
{
private final Map<InetAddress, Collection<Range>> endpointToRanges = new HashMap<InetAddress, Collection<Range>>();
/**
* Initialize the client.
* Perform any step necessary so that after the call to the this
* method:
* * StorageService is correctly initialized (so that gossip and
* messaging service is too)
* * getEndpointToRangesMap() returns a correct map
* This method is guaranted to be called before any other method of a
* client.
*/
public abstract void init();
/**
* Stop the client.
*/
public void stop() {}
/**
* Validate that {@code keyspace} is an existing keyspace and {@code
* cfName} one of its existing column family.
*/
public abstract boolean validateColumnFamily(String keyspace, String cfName);
public Map<InetAddress, Collection<Range>> getEndpointToRangesMap()
{
return endpointToRanges;
}
protected void addRangeForEndpoint(Range range, InetAddress endpoint)
{
Collection<Range> ranges = endpointToRanges.get(endpoint);
if (ranges == null)
{
ranges = new HashSet<Range>();
endpointToRanges.put(endpoint, ranges);
}
ranges.add(range);
}
}
}

11
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -150,6 +150,9 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
public static SSTableReader open(Descriptor descriptor, Set<Component> components, Set<DecoratedKey> savedKeys, DataTracker tracker, CFMetaData metadata, IPartitioner partitioner) throws IOException
{
assert partitioner != null;
// Minimum components without which we can't do anything
assert components.contains(Component.DATA);
assert components.contains(Component.PRIMARY_INDEX);
long start = System.currentTimeMillis();
logger.info("Opening " + descriptor);
@ -158,7 +161,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
EstimatedHistogram columnCounts;
File statsFile = new File(descriptor.filenameFor(SSTable.COMPONENT_STATS));
ReplayPosition rp = ReplayPosition.NONE;
if (statsFile.exists())
if (components.contains(Component.STATS) && statsFile.exists())
{
DataInputStream dis = null;
try
@ -249,6 +252,12 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
void loadBloomFilter() throws IOException
{
if (!components.contains(Component.FILTER))
{
bf = BloomFilter.emptyFilter();
return;
}
DataInputStream stream = null;
try
{

5
src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java
View File

@ -62,7 +62,7 @@ public class RangeSliceResponseResolver implements IResponseResolver<Iterable<Ro
return reply.rows;
}
// Note: this deserializes the response a 2nd time if getData was called first
// Note: this would deserialize the response a 2nd time if getData was called first.
// (this is not currently an issue since we don't do read repair for range queries.)
public Iterable<Row> resolve() throws IOException
{
@ -73,7 +73,7 @@ public class RangeSliceResponseResolver implements IResponseResolver<Iterable<Ro
return o1.left.key.compareTo(o2.left.key);
}
});
int n = 0;
for (Message response : responses)
{
@ -83,7 +83,6 @@ public class RangeSliceResponseResolver implements IResponseResolver<Iterable<Ro
}
// for each row, compute the combination of all different versions seen, and repair incomplete versions
return new ReducingIterator<Pair<Row,InetAddress>, Row>(collator)
{
List<ColumnFamily> versions = new ArrayList<ColumnFamily>(sources.size());

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

@ -18,6 +18,7 @@
package org.apache.cassandra.service;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.lang.management.ManagementFactory;
@ -50,6 +51,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.*;
import org.apache.cassandra.io.DeletionService;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.DynamicEndpointSnitch;
@ -327,6 +329,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
Gossiper.instance.unregister(this);
Gossiper.instance.stop();
MessagingService.instance().shutdown();
// give it a second so that task accepted before the MessagingService shutdown gets submitted to the stage (to avoid RejectedExecutionException)
try { Thread.sleep(1000L); } catch (InterruptedException e) {}
StageManager.shutdownNow();
}
@ -354,7 +358,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
// sleep a while to allow gossip to warm up (the other nodes need to know about this one before they can reply).
try
{
Thread.sleep(5000L);
Thread.sleep(RING_DELAY);
}
catch (Exception ex)
{
@ -2428,4 +2432,37 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
return new Pair<Set<Range>, Set<Range>>(toStream, toFetch);
}
public void bulkLoad(String directory)
{
File dir = new File(directory);
if (!dir.exists() || !dir.isDirectory())
throw new IllegalArgumentException("Invalid directory " + directory);
SSTableLoader.Client client = new SSTableLoader.Client()
{
public void init() {}
public boolean validateColumnFamily(String keyspace, String cfName)
{
return DatabaseDescriptor.getCFMetaData(keyspace, cfName) != null;
}
};
SSTableLoader.OutputHandler oh = new SSTableLoader.OutputHandler()
{
public void output(String msg) { logger_.info(msg); }
public void debug(String msg) { logger_.debug(msg); }
};
SSTableLoader loader = new SSTableLoader(dir, client, oh);
try
{
loader.stream().get();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
}

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

@ -290,4 +290,6 @@ public interface StorageServiceMBean
public boolean isJoined();
public void setCompactionThroughputMbPerSec(int value);
public void bulkLoad(String directory);
}

3
src/java/org/apache/cassandra/streaming/OperationType.java
View File

@ -26,6 +26,7 @@ public enum OperationType
AES,
BOOTSTRAP,
UNBOOTSTRAP,
RESTORE_REPLICA_COUNT;
RESTORE_REPLICA_COUNT,
BULK_LOAD;
}

2
src/java/org/apache/cassandra/streaming/StreamOutSession.java
View File

@ -134,7 +134,7 @@ public class StreamOutSession
Thread.sleep(10);
}
Collection<PendingFile> getFiles()
public Collection<PendingFile> getFiles()
{
return files.values();
}

13
src/java/org/apache/cassandra/thrift/ThriftValidation.java
View File

@ -28,6 +28,7 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.RandomPartitioner;
import org.apache.cassandra.dht.Token;
@ -536,9 +537,9 @@ public class ThriftValidation
if (cfType == null)
throw new InvalidRequestException("invalid column type " + cf_def.column_type);
DatabaseDescriptor.getComparator(cf_def.comparator_type);
DatabaseDescriptor.getComparator(cf_def.subcomparator_type);
DatabaseDescriptor.getComparator(cf_def.default_validation_class);
TypeParser.parse(cf_def.comparator_type);
TypeParser.parse(cf_def.subcomparator_type);
TypeParser.parse(cf_def.default_validation_class);
if (cfType != ColumnFamilyType.Super && cf_def.subcomparator_type != null)
throw new InvalidRequestException("subcomparator_type is invalid for standard columns");
@ -546,8 +547,8 @@ public class ThriftValidation
return;
AbstractType comparator = cfType == ColumnFamilyType.Standard
? DatabaseDescriptor.getComparator(cf_def.comparator_type)
: DatabaseDescriptor.getComparator(cf_def.subcomparator_type);
? TypeParser.parse(cf_def.comparator_type)
: TypeParser.parse(cf_def.subcomparator_type);
Set<String> indexNames = new HashSet<String>();
for (ColumnDef c : cf_def.column_metadata)
@ -558,7 +559,7 @@ public class ThriftValidation
throw new InvalidRequestException("Duplicate index names " + idxName);
indexNames.add(idxName);
DatabaseDescriptor.getComparator(c.validation_class);
TypeParser.parse(c.validation_class);
try
{

412
src/java/org/apache/cassandra/tools/BulkLoader.java Normal file
View File

@ -0,0 +1,412 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.tools;
import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.*;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.sstable.SSTableLoader;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.PendingFile;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.commons.cli.*;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
public class BulkLoader
{
private static final String TOOL_NAME = "sstableloader";
private static final String VERBOSE_OPTION = "verbose";
private static final String DEBUG_OPTION = "debug";
private static final String HELP_OPTION = "help";
private static final String NOPROGRESS_OPTION = "no-progress";
private static final String IGNORE_NODES_OPTION = "ignore";
public static void main(String args[]) throws IOException
{
LoaderOptions options = LoaderOptions.parseArgs(args);
try
{
SSTableLoader loader = new SSTableLoader(options.directory, new ExternalClient(options.directory.getName(), options), options);
SSTableLoader.LoaderFuture future = loader.stream(options.ignores);
if (options.noProgress)
{
future.get();
}
else
{
ProgressIndicator indicator = new ProgressIndicator(future.getPendingFiles());
indicator.start();
System.out.println("");
while (!future.isDone())
{
if (indicator.printProgress())
{
// We're done with streaming
System.out.println("\nWaiting for targets to rebuild indexes ...");
future.get();
assert future.isDone();
}
else
{
try { Thread.sleep(1000L); } catch (Exception e) {}
}
}
}
System.exit(0); // We need that to stop non daemonized threads
}
catch (Exception e)
{
System.err.println(e.getMessage());
if (options.debug)
e.printStackTrace(System.err);
System.exit(1);
}
}
// Return true when everything is at 100%
static class ProgressIndicator
{
private final Map<InetAddress, Collection<PendingFile>> filesByHost;
private long startTime;
private long lastProgress;
private long lastTime;
public ProgressIndicator(Map<InetAddress, Collection<PendingFile>> filesByHost)
{
this.filesByHost = new HashMap<InetAddress, Collection<PendingFile>>(filesByHost);
}
public void start()
{
startTime = System.currentTimeMillis();
}
public boolean printProgress()
{
boolean done = true;
StringBuilder sb = new StringBuilder();
sb.append("\rprogress: ");
long totalProgress = 0;
long totalSize = 0;
for (Map.Entry<InetAddress, Collection<PendingFile>> entry : filesByHost.entrySet())
{
long progress = 0;
long size = 0;
int completed = 0;
Collection<PendingFile> pendings = entry.getValue();
for (PendingFile f : pendings)
{
progress += f.progress;
size += f.size;
if (f.progress == f.size)
completed++;
}
totalProgress += progress;
totalSize += size;
if (completed != pendings.size())
done = false;
sb.append("[").append(entry.getKey());
sb.append(" ").append(completed).append("/").append(pendings.size());
sb.append(" (").append(size == 0 ? 100L : progress * 100L / size).append(")] ");
}
long time = System.currentTimeMillis();
long deltaTime = time - lastTime;
lastTime = time;
long deltaProgress = totalProgress - lastProgress;
lastProgress = totalProgress;
sb.append("[total: ").append(totalSize == 0 ? 100L : totalProgress * 100L / totalSize).append(" - ");
sb.append(mbPerSec(deltaProgress, deltaTime)).append("MB/s");
sb.append(" (avg: ").append(mbPerSec(totalProgress, time - startTime)).append("MB/s)]");;
System.out.print(sb.toString());
return done;
}
private int mbPerSec(long bytes, long timeInMs)
{
double bytesPerMs = ((double)bytes) / timeInMs;
return (int)((bytesPerMs * 1000) / (1024 * 2024));
}
}
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)
{
super();
this.keyspace = keyspace;
this.outputHandler = outputHandler;
}
public void init()
{
outputHandler.output(String.format("Starting client (and waiting %d seconds for gossip) ...", StorageService.RING_DELAY / 1000));
try
{
// Init gossip
StorageService.instance.initClient();
Set<InetAddress> hosts = Gossiper.instance.getLiveMembers();
hosts.remove(FBUtilities.getLocalAddress());
if (hosts.isEmpty())
throw new IllegalStateException("Cannot load any sstable, no live member found in the cluster");
// Query endpoint to ranges map and schemas from thrift
String host = hosts.iterator().next().toString().substring(1);
int port = DatabaseDescriptor.getRpcPort();
Cassandra.Client client = createThriftClient(host, port);
List<TokenRange> tokenRanges = client.describe_ring(keyspace);
List<KsDef> ksDefs = client.describe_keyspaces();
Token.TokenFactory tkFactory = StorageService.getPartitioner().getTokenFactory();
try
{
for (TokenRange tr : tokenRanges)
{
Range range = new Range(tkFactory.fromString(tr.start_token), tkFactory.fromString(tr.end_token));
for (String ep : tr.endpoints)
{
addRangeForEndpoint(range, InetAddress.getByName(ep));
}
}
}
catch (UnknownHostException e)
{
throw new RuntimeException("Got an unknow host from describe_ring()", e);
}
for (KsDef ksDef : ksDefs)
{
Set<String> cfs = new HashSet<String>();
for (CfDef cfDef : ksDef.cf_defs)
cfs.add(cfDef.name);
knownCfs.put(ksDef.name, cfs);
}
}
catch (Exception e)
{
throw new RuntimeException(e);
}
}
@Override
public void stop()
{
StorageService.instance.stopClient();
}
public boolean validateColumnFamily(String keyspace, String cfName)
{
Set<String> cfs = knownCfs.get(keyspace);
return cfs != null && cfs.contains(cfName);
}
private static Cassandra.Client createThriftClient(String host, int port) throws TTransportException
{
TSocket socket = new TSocket(host, port);
TTransport trans = new TFramedTransport(socket);
trans.open();
TProtocol protocol = new TBinaryProtocol(trans);
return new Cassandra.Client(protocol);
}
}
static class LoaderOptions implements SSTableLoader.OutputHandler
{
public final File directory;
public boolean debug;
public boolean verbose;
public boolean noProgress;
public Set<InetAddress> ignores = new HashSet<InetAddress>();
LoaderOptions(File directory)
{
this.directory = directory;
}
public static LoaderOptions parseArgs(String cmdArgs[])
{
CommandLineParser parser = new GnuParser();
CmdLineOptions options = getCmdLineOptions();
try
{
CommandLine cmd = parser.parse(options, cmdArgs, false);
if (cmd.hasOption(HELP_OPTION))
{
printUsage(options);
System.exit(0);
}
String[] args = cmd.getArgs();
if (args.length == 0)
{
System.err.println("Missing sstable directory argument");
printUsage(options);
System.exit(1);
}
if (args.length > 1)
{
System.err.println("Too many arguments");
printUsage(options);
System.exit(1);
}
String dirname = args[0];
File dir = new File(dirname);
if (!dir.exists())
errorMsg("Unknown directory: " + dirname, options);
if (!dir.isDirectory())
errorMsg(dirname + " is not a directory", options);
LoaderOptions opts = new LoaderOptions(dir);
opts.debug = cmd.hasOption(DEBUG_OPTION);
opts.verbose = cmd.hasOption(VERBOSE_OPTION);
opts.noProgress = cmd.hasOption(NOPROGRESS_OPTION);
if (cmd.hasOption(IGNORE_NODES_OPTION))
{
String[] nodes = cmd.getOptionValue(IGNORE_NODES_OPTION).split(",");
try
{
for (String node : nodes)
{
opts.ignores.add(InetAddress.getByName(node));
}
}
catch (UnknownHostException e)
{
errorMsg(e.getMessage(), options);
}
}
return opts;
}
catch (ParseException e)
{
errorMsg(e.getMessage(), options);
return null;
}
}
private static void errorMsg(String msg, CmdLineOptions options)
{
System.err.println(msg);
printUsage(options);
System.exit(1);
}
public void output(String msg)
{
System.out.println(msg);
}
public void debug(String msg)
{
if (verbose)
System.out.println(msg);
}
private static CmdLineOptions getCmdLineOptions()
{
CmdLineOptions options = new CmdLineOptions();
options.addOption(null, DEBUG_OPTION, "display stack traces");
options.addOption("v", VERBOSE_OPTION, "verbose output");
options.addOption("h", HELP_OPTION, "display this help message");
options.addOption(null, NOPROGRESS_OPTION, "don't display progress");
options.addOption("i", IGNORE_NODES_OPTION, "don't stream to this (comma separated) list of nodes");
return options;
}
public static void printUsage(Options options)
{
String usage = String.format("%s [options] <dir_path>", TOOL_NAME);
StringBuilder header = new StringBuilder();
header.append("--\n");
header.append("Bulk load the sstables find in the directory <dir_path> to the configured cluster." );
header.append("The last directory of <dir_path> is used as the keyspace name. ");
header.append("So for instance, to load a sstable named Standard1-g-1-Data.db into keyspace Keyspace1, ");
header.append("you will need to have the files Standard1-g-1-Data.db and Standard1-g-1-Index.db in a ");
header.append("directory Keyspace1/ in the current directory and call: sstableloader Keyspace1");
header.append("\n--\n");
header.append("Options are:");
new HelpFormatter().printHelp(usage, header.toString(), options, "");
}
}
private static class CmdLineOptions extends Options
{
/**
* Add option with argument and argument name
* @param opt shortcut for option name
* @param longOpt complete option name
* @param argName argument name
* @param description description of the option
* @return updated Options object
*/
public Options addOption(String opt, String longOpt, String argName, String description)
{
Option option = new Option(opt, longOpt, true, description);
option.setArgName(argName);
return addOption(option);
}
/**
* Add option without argument
* @param opt shortcut for option name
* @param longOpt complete option name
* @param description description of the option
* @return updated Options object
*/
public Options addOption(String opt, String longOpt, String description)
{
return addOption(new Option(opt, longOpt, false, description));
}
}
}

5
src/java/org/apache/cassandra/utils/BloomFilter.java
View File

@ -41,6 +41,11 @@ public class BloomFilter extends Filter
bitset = bs;
}
public static BloomFilter emptyFilter()
{
return new BloomFilter(0, bucketsFor(0, 0));
}
public static ICompactSerializer2<BloomFilter> serializer()
{
return serializer_;

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

@ -522,11 +522,6 @@ public class FBUtilities
return FBUtilities.construct(partitionerClassName, "partitioner");
}
public static AbstractType getComparator(String compareWith) throws ConfigurationException
{
return TypeParser.parse(compareWith);
}
/**
* @return The Class for the given name.
* @param classname Fully qualified classname.

27
test/unit/org/apache/cassandra/db/marshal/UUIDTypeTest.java
View File

@ -1,4 +1,5 @@
package org.apache.cassandra.db.marshal;
/*
*
* Licensed to the Apache Software Foundation (ASF) under one
@ -20,7 +21,6 @@ package org.apache.cassandra.db.marshal;
*
*/
import static org.junit.Assert.assertEquals;
import java.net.InetAddress;
@ -30,6 +30,7 @@ import java.util.Arrays;
import java.util.Random;
import java.util.UUID;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.log4j.Logger;
import org.junit.Test;
@ -48,6 +49,9 @@ public class UUIDTypeTest
UUID t1 = newTimeBasedUUID();
UUID t2 = newTimeBasedUUID();
testCompare(null, t2, -1);
testCompare(t1, null, 1);
testCompare(t1, t2, -1);
testCompare(t1, t1, 0);
testCompare(t2, t2, 0);
@ -110,8 +114,8 @@ public class UUIDTypeTest
public String describeCompare(UUID u1, UUID u2, int c)
{
String tb1 = (u1.version() == 1) ? "time-based " : "random ";
String tb2 = (u2.version() == 1) ? "time-based " : "random ";
String tb1 = (u1 == null) ? "null" : (u1.version() == 1) ? "time-based " : "random ";
String tb2 = (u2 == null) ? "null" : (u2.version() == 1) ? "time-based " : "random ";
String comp = (c < 0) ? " < " : ((c == 0) ? " = " : " > ");
return tb1 + u1 + comp + tb2 + u2;
}
@ -131,6 +135,9 @@ public class UUIDTypeTest
public static ByteBuffer bytebuffer(UUID uuid)
{
if (uuid == null)
return ByteBufferUtil.EMPTY_BYTE_BUFFER;
long msb = uuid.getMostSignificantBits();
long lsb = uuid.getLeastSignificantBits();
byte[] bytes = new byte[16];
@ -149,29 +156,23 @@ public class UUIDTypeTest
public void logJdkUUIDCompareToVariance(UUID u1, UUID u2, int expC)
{
if ((u1 == null) || (u2 == null))
return;
if (u1.version() != u2.version())
{
return;
}
if (u1.version() == 1)
{
return;
}
if (u1.compareTo(u2) != expC)
{
logger.info("*** Note: java.util.UUID.compareTo() would have compared this differently");
}
}
public void testCompare(UUID u1, UUID u2, int expC)
{
int c = sign(uuidType.compare(bytebuffer(u1), bytebuffer(u2)));
expC = sign(expC);
assertEquals("Expected " + describeCompare(u1, u2, expC) + ", got "
+ describeCompare(u1, u2, c), expC, c);
assertEquals("Expected " + describeCompare(u1, u2, expC) + ", got " + describeCompare(u1, u2, c), expC, c);
if (u1.version() == 1 && u2.version() == 1)
if (((u1 != null) && (u1.version() == 1)) && ((u2 != null) && (u2.version() == 1)))
assertEquals(c, sign(TimeUUIDType.instance.compare(bytebuffer(u1), bytebuffer(u2))));
logJdkUUIDCompareToVariance(u1, u2, c);