diff --git a/CHANGES.txt b/CHANGES.txt index e223bf72de..39b95b41d0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/NEWS.txt b/NEWS.txt index 0b2c57e553..48c20a32cd 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -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 -------- diff --git a/bin/sstableloader b/bin/sstableloader new file mode 100644 index 0000000000..2c82aa9aa0 --- /dev/null +++ b/bin/sstableloader @@ -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 diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java index 848f66aacc..0734fee541 100644 --- a/drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java +++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/TypedColumn.java @@ -26,7 +26,7 @@ import org.apache.cassandra.thrift.Column; import java.nio.ByteBuffer; -class TypedColumn +public class TypedColumn { private final Column rawColumn; diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java index 00132cd6c7..902e12b8bb 100644 --- a/src/java/org/apache/cassandra/cli/CliClient.java +++ b/src/java/org/apache/cassandra/cli/CliClient.java @@ -504,7 +504,7 @@ public class CliClient { try { - return FBUtilities.getComparator(compareWith); + return TypeParser.parse(compareWith); } catch (ConfigurationException ce) { diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 48090b166e..dac566f8d0 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -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); } diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java index 2d77d6c3b9..4c1ba6bf10 100644 --- a/src/java/org/apache/cassandra/config/ColumnDefinition.java +++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java @@ -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()), diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 42d7503ed0..24c9cbc4b2 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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. diff --git a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java index 23c80f38a6..73ce899a76 100644 --- a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java +++ b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java @@ -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) diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java index 8ca6f0bb84..a33fb97736 100644 --- a/src/java/org/apache/cassandra/cql/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java @@ -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)) diff --git a/src/java/org/apache/cassandra/db/marshal/IntegerType.java b/src/java/org/apache/cassandra/db/marshal/IntegerType.java index f05060f5d0..824dd84757 100644 --- a/src/java/org/apache/cassandra/db/marshal/IntegerType.java +++ b/src/java/org/apache/cassandra/db/marshal/IntegerType.java @@ -133,7 +133,7 @@ public final class IntegerType extends AbstractType 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) diff --git a/src/java/org/apache/cassandra/db/marshal/TypeParser.java b/src/java/org/apache/cassandra/db/marshal/TypeParser.java index 4aef620dcd..62923979a9 100644 --- a/src/java/org/apache/cassandra/db/marshal/TypeParser.java +++ b/src/java/org/apache/cassandra/db/marshal/TypeParser.java @@ -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. */ diff --git a/src/java/org/apache/cassandra/db/marshal/UUIDType.java b/src/java/org/apache/cassandra/db/marshal/UUIDType.java index a49ce0f5f0..2c21bc889b 100644 --- a/src/java/org/apache/cassandra/db/marshal/UUIDType.java +++ b/src/java/org/apache/cassandra/db/marshal/UUIDType.java @@ -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:
@@ -39,9 +38,7 @@ import org.apache.cassandra.utils.UUIDGen; * - compare timestamps if both are time-based
* - compare lexically, unsigned msb-to-lsb comparison
* - * @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) diff --git a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java index 7e8f9f8e89..509446fcc4 100644 --- a/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java +++ b/src/java/org/apache/cassandra/hadoop/ColumnFamilyRecordReader.java @@ -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 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; diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java new file mode 100644 index 0000000000..3bc1d60eb5 --- /dev/null +++ b/src/java/org/apache/cassandra/io/sstable/SSTableLoader.java @@ -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 openSSTables() + { + final List sstables = new LinkedList(); + + directory.list(new FilenameFilter() + { + public boolean accept(File dir, String name) + { + Pair 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 components = new HashSet(); + 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.emptySet()); + } + + public LoaderFuture stream(Set toIgnore) throws IOException + { + client.init(); + + Collection sstables = openSSTables(); + if (sstables.isEmpty()) + { + outputHandler.output("No sstables to stream"); + return new LoaderFuture(0); + } + + Map> 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> entry : endpointToRanges.entrySet()) + { + InetAddress remote = entry.getKey(); + if (toIgnore.contains(remote)) + { + future.latch.countDown(); + continue; + } + Collection 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 + { + final CountDownLatch latch; + final Map> pendingFiles; + + private LoaderFuture(int request) + { + latch = new CountDownLatch(request); + pendingFiles = new HashMap>(); + } + + private void setPendings(InetAddress remote, Collection 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> getPendingFiles() + { + return pendingFiles; + } + } + + private String names(Collection 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> endpointToRanges = new HashMap>(); + + /** + * 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> getEndpointToRangesMap() + { + return endpointToRanges; + } + + protected void addRangeForEndpoint(Range range, InetAddress endpoint) + { + Collection ranges = endpointToRanges.get(endpoint); + if (ranges == null) + { + ranges = new HashSet(); + endpointToRanges.put(endpoint, ranges); + } + ranges.add(range); + } + } +} diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 6fd850f508..0f7fd9e8aa 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -150,6 +150,9 @@ public class SSTableReader extends SSTable implements Comparable public static SSTableReader open(Descriptor descriptor, Set components, Set 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 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 void loadBloomFilter() throws IOException { + if (!components.contains(Component.FILTER)) + { + bf = BloomFilter.emptyFilter(); + return; + } + DataInputStream stream = null; try { diff --git a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java index 6077e38bc7..84b91e9f4d 100644 --- a/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java +++ b/src/java/org/apache/cassandra/service/RangeSliceResponseResolver.java @@ -62,7 +62,7 @@ public class RangeSliceResponseResolver implements IResponseResolver resolve() throws IOException { @@ -73,7 +73,7 @@ public class RangeSliceResponseResolver implements IResponseResolver, Row>(collator) { List versions = new ArrayList(sources.size()); diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java index 5e899dde4c..bb9d825bfe 100644 --- a/src/java/org/apache/cassandra/service/StorageService.java +++ b/src/java/org/apache/cassandra/service/StorageService.java @@ -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>(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); + } + } } diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java index eccc5e0fcb..39c4c7aa01 100644 --- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java +++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java @@ -290,4 +290,6 @@ public interface StorageServiceMBean public boolean isJoined(); public void setCompactionThroughputMbPerSec(int value); + + public void bulkLoad(String directory); } diff --git a/src/java/org/apache/cassandra/streaming/OperationType.java b/src/java/org/apache/cassandra/streaming/OperationType.java index 86a3953478..ddf60ade6b 100644 --- a/src/java/org/apache/cassandra/streaming/OperationType.java +++ b/src/java/org/apache/cassandra/streaming/OperationType.java @@ -26,6 +26,7 @@ public enum OperationType AES, BOOTSTRAP, UNBOOTSTRAP, - RESTORE_REPLICA_COUNT; + RESTORE_REPLICA_COUNT, + BULK_LOAD; } diff --git a/src/java/org/apache/cassandra/streaming/StreamOutSession.java b/src/java/org/apache/cassandra/streaming/StreamOutSession.java index fed85cf734..c58f41436a 100644 --- a/src/java/org/apache/cassandra/streaming/StreamOutSession.java +++ b/src/java/org/apache/cassandra/streaming/StreamOutSession.java @@ -134,7 +134,7 @@ public class StreamOutSession Thread.sleep(10); } - Collection getFiles() + public Collection getFiles() { return files.values(); } diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java index 69e79b837d..136ff9b3cd 100644 --- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java +++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java @@ -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 indexNames = new HashSet(); 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 { diff --git a/src/java/org/apache/cassandra/tools/BulkLoader.java b/src/java/org/apache/cassandra/tools/BulkLoader.java new file mode 100644 index 0000000000..82a6d6d075 --- /dev/null +++ b/src/java/org/apache/cassandra/tools/BulkLoader.java @@ -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> filesByHost; + private long startTime; + private long lastProgress; + private long lastTime; + + public ProgressIndicator(Map> filesByHost) + { + this.filesByHost = new HashMap>(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> entry : filesByHost.entrySet()) + { + long progress = 0; + long size = 0; + int completed = 0; + Collection 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> knownCfs = new HashMap>(); + 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 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 tokenRanges = client.describe_ring(keyspace); + List 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 cfs = new HashSet(); + 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 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 ignores = new HashSet(); + + 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] ", TOOL_NAME); + StringBuilder header = new StringBuilder(); + header.append("--\n"); + header.append("Bulk load the sstables find in the directory to the configured cluster." ); + header.append("The last directory of 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)); + } + } +} diff --git a/src/java/org/apache/cassandra/utils/BloomFilter.java b/src/java/org/apache/cassandra/utils/BloomFilter.java index 8ecb5bafc7..63ef54dac1 100644 --- a/src/java/org/apache/cassandra/utils/BloomFilter.java +++ b/src/java/org/apache/cassandra/utils/BloomFilter.java @@ -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 serializer() { return serializer_; diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index cd3f9c345d..6e1e619781 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -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. diff --git a/test/unit/org/apache/cassandra/db/marshal/UUIDTypeTest.java b/test/unit/org/apache/cassandra/db/marshal/UUIDTypeTest.java index 65b7fdd083..2abb408c1e 100644 --- a/test/unit/org/apache/cassandra/db/marshal/UUIDTypeTest.java +++ b/test/unit/org/apache/cassandra/db/marshal/UUIDTypeTest.java @@ -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);