You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sl...@apache.org on 2012/06/18 09:45:34 UTC
[1/2] git commit: Add column metadata to system column families
Updated Branches:
refs/heads/trunk c34c70f0b -> 8ea2d2a6a
Add column metadata to system column families
patch by jbellis; reviewed by slebresne for CASSANDRA-4018
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/8ea2d2a6
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/8ea2d2a6
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/8ea2d2a6
Branch: refs/heads/trunk
Commit: 8ea2d2a6a15bc1f791711d55b4345c233020d753
Parents: c34c70f
Author: Jonathan Ellis <jb...@apache.org>
Authored: Wed Apr 11 18:38:39 2012 -0500
Committer: Sylvain Lebresne <sy...@datastax.com>
Committed: Mon Jun 18 09:43:00 2012 +0200
----------------------------------------------------------------------
.../org/apache/cassandra/config/CFMetaData.java | 218 +++++++------
.../apache/cassandra/config/ColumnDefinition.java | 28 +--
.../cassandra/config/DatabaseDescriptor.java | 13 +-
.../org/apache/cassandra/config/KSMetaData.java | 15 +-
.../apache/cassandra/cql/jdbc/JdbcInetAddress.java | 101 ++++++
src/java/org/apache/cassandra/cql3/CFPropDefs.java | 1 +
src/java/org/apache/cassandra/cql3/Cql.g | 2 +
.../org/apache/cassandra/cql3/QueryProcessor.java | 41 ++-
.../apache/cassandra/cql3/UntypedResultSet.java | 17 +-
.../statements/CreateColumnFamilyStatement.java | 24 +-
src/java/org/apache/cassandra/db/Column.java | 6 +
src/java/org/apache/cassandra/db/Directories.java | 5 +-
.../apache/cassandra/db/HintedHandOffManager.java | 74 ++---
src/java/org/apache/cassandra/db/RowMutation.java | 38 +--
src/java/org/apache/cassandra/db/SystemTable.java | 244 +++++++--------
.../db/marshal/AbstractCompositeType.java | 2 +
.../apache/cassandra/db/marshal/CompositeType.java | 24 ++
.../cassandra/db/marshal/DynamicCompositeType.java | 5 +
.../cassandra/db/marshal/InetAddressType.java | 84 +++++
.../apache/cassandra/locator/LocalStrategy.java | 21 +-
.../org/apache/cassandra/service/StorageProxy.java | 2 +-
.../unit/org/apache/cassandra/config/DefsTest.java | 4 +-
.../cassandra/thrift/ThriftValidationTest.java | 3 +-
23 files changed, 585 insertions(+), 387 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/config/CFMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index 90f0c29..69c8864 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -34,9 +34,11 @@ import org.apache.commons.lang.builder.ToStringBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.antlr.runtime.RecognitionException;
import org.apache.cassandra.cql3.CFDefinition;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.cql3.statements.CreateColumnFamilyStatement;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.AbstractCompactionStrategy;
import org.apache.cassandra.db.index.SecondaryIndex;
@@ -44,7 +46,6 @@ import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.compress.CompressionParameters;
import org.apache.cassandra.io.compress.SnappyCompressor;
-import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.IndexType;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -76,92 +77,99 @@ public final class CFMetaData
// Note that this is the default only for user created tables
public final static String DEFAULT_COMPRESSOR = SnappyCompressor.isAvailable() ? SnappyCompressor.class.getCanonicalName() : null;
- public static final CFMetaData StatusCf = newSystemMetadata(SystemTable.STATUS_CF, 0, "persistent metadata for the local node", BytesType.instance, null);
- public static final CFMetaData HintsCf = newSystemMetadata(HintedHandOffManager.HINTS_CF, 1, "hinted handoff data", BytesType.instance, BytesType.instance);
@Deprecated
- public static final CFMetaData MigrationsCf = newSystemMetadata(DefsTable.OLD_MIGRATIONS_CF, 2, "individual schema mutations", TimeUUIDType.instance, null);
+ public static final CFMetaData OldStatusCf = newSystemMetadata(SystemTable.OLD_STATUS_CF, 0, "unused", BytesType.instance, null);
@Deprecated
- public static final CFMetaData SchemaCf = newSystemMetadata(DefsTable.OLD_SCHEMA_CF, 3, "current state of the schema", UTF8Type.instance, null);
- public static final CFMetaData IndexCf = newSystemMetadata(SystemTable.INDEX_CF, 5, "indexes that have been completed", UTF8Type.instance, null);
- public static final CFMetaData NodeIdCf = newSystemMetadata(SystemTable.NODE_ID_CF, 6, "nodeId and their metadata", TimeUUIDType.instance, null);
- public static final CFMetaData VersionCf =
- newSystemMetadata(SystemTable.VERSION_CF, 7, "server version information", UTF8Type.instance, null)
- .keyAlias(ByteBufferUtil.bytes("component"))
- .keyValidator(UTF8Type.instance)
- .columnMetadata(Collections.singletonMap(ByteBufferUtil.bytes("version"),
- new ColumnDefinition(ByteBufferUtil.bytes("version"),
- UTF8Type.instance,
- null,
- null,
- null,
- null)));
+ public static final CFMetaData OldHintsCf = newSystemMetadata(SystemTable.OLD_HINTS_CF, 1, "unused", BytesType.instance, BytesType.instance);
+ @Deprecated
+ public static final CFMetaData MigrationsCf = newSystemMetadata(DefsTable.OLD_MIGRATIONS_CF, 2, "unused", TimeUUIDType.instance, null);
+ @Deprecated
+ public static final CFMetaData SchemaCf = newSystemMetadata(DefsTable.OLD_SCHEMA_CF, 3, "unused", UTF8Type.instance, null);
+
+ public static final CFMetaData IndexCf = compile(5, "CREATE TABLE \"" + SystemTable.INDEX_CF + "\" ("
+ + "table_name text,"
+ + "index_name text,"
+ + "blank blob," // TODO r/m when CASSANDRA-4212 is done
+ + "PRIMARY KEY (table_name, index_name)"
+ + ") WITH COMPACT STORAGE AND COMMENT='indexes that have been completed'");
+
+ public static final CFMetaData NodeIdCf = compile(6, "CREATE TABLE \"" + SystemTable.NODE_ID_CF + "\" ("
+ + "key text,"
+ + "id timeuuid,"
+ + "blank blob," // TODO r/m when CASSANDRA-4212 is done
+ + "PRIMARY KEY (key, id)"
+ + ") WITH COMPACT STORAGE AND COMMENT='counter node IDs'");
// new-style schema
- public static final CFMetaData SchemaKeyspacesCf;
- public static final CFMetaData SchemaColumnFamiliesCf;
- public static final CFMetaData SchemaColumnsCf;
- static
- {
- SchemaKeyspacesCf = newSchemaMetadata(SystemTable.SCHEMA_KEYSPACES_CF,
- 8,
- "Keyspace definitions",
- AsciiType.instance,
- null)
- .keyValidator(AsciiType.instance)
- .keyAlias("keyspace")
- .columnMetadata(ColumnDefinition.utf8("name", null),
- ColumnDefinition.bool("durable_writes", null),
- ColumnDefinition.ascii("strategy_class", null),
- ColumnDefinition.ascii("strategy_options", null));
-
- SchemaColumnFamiliesCf = newSchemaMetadata(SystemTable.SCHEMA_COLUMNFAMILIES_CF,
- 9,
- "ColumnFamily definitions",
- CompositeType.getInstance(Arrays.<AbstractType<?>>asList(AsciiType.instance, AsciiType.instance)),
- null)
- .keyValidator(AsciiType.instance)
- .keyAlias("keyspace")
- .columnAliases(Arrays.asList(ByteBufferUtil.bytes("columnfamily")))
- .columnMetadata(ColumnDefinition.int32("id", 1),
- ColumnDefinition.ascii("type", 1),
- ColumnDefinition.ascii("comparator", 1),
- ColumnDefinition.ascii("subcomparator", 1),
- ColumnDefinition.utf8("comment", 1),
- ColumnDefinition.double_("read_repair_chance", 1),
- ColumnDefinition.double_("local_read_repair_chance", 1),
- ColumnDefinition.bool("replicate_on_write", 1),
- ColumnDefinition.int32("gc_grace_seconds", 1),
- ColumnDefinition.ascii("default_validator", 1),
- ColumnDefinition.ascii("key_validator", 1),
- ColumnDefinition.int32("min_compaction_threshold", 1),
- ColumnDefinition.int32("max_compaction_threshold", 1),
- ColumnDefinition.ascii("key_alias", 1),
- ColumnDefinition.double_("bloom_filter_fp_chance", 1),
- ColumnDefinition.ascii("caching", 1),
- ColumnDefinition.ascii("compaction_strategy_class", 1),
- ColumnDefinition.ascii("compression_parameters", 1),
- ColumnDefinition.utf8("value_alias", 1),
- ColumnDefinition.utf8("column_aliases", 1),
- ColumnDefinition.ascii("compaction_strategy_options", 1));
-
- SchemaColumnsCf = newSchemaMetadata(SystemTable.SCHEMA_COLUMNS_CF,
- 10,
- "ColumnFamily column attributes",
- CompositeType.getInstance(Arrays.<AbstractType<?>>asList(AsciiType.instance,
- AsciiType.instance,
- AsciiType.instance)),
- null)
- .keyValidator(AsciiType.instance)
- .keyAlias("keyspace")
- .columnAliases(Arrays.asList(ByteBufferUtil.bytes("columnfamily"), ByteBufferUtil.bytes("column")))
- .columnMetadata(ColumnDefinition.ascii("validator", 2),
- ColumnDefinition.ascii("index_type", 2),
- ColumnDefinition.ascii("index_options", 2),
- ColumnDefinition.ascii("index_name", 2),
- ColumnDefinition.ascii("component_index", 2));
- }
-
- public static final CFMetaData HostIdCf = newSystemMetadata(SystemTable.HOST_ID_CF, 11, "Host Identifier", UUIDType.instance, null);
+ public static final CFMetaData SchemaKeyspacesCf = compile(8, "CREATE TABLE " + SystemTable.SCHEMA_KEYSPACES_CF + "("
+ + "keyspace_name text PRIMARY KEY,"
+ + "name text," // TODO
+ + "durable_writes boolean,"
+ + "strategy_class text,"
+ + "strategy_options text"
+ + ") WITH COMMENT='keyspace definitions' AND gc_grace_seconds=8640");
+ public static final CFMetaData SchemaColumnFamiliesCf = compile(9, "CREATE TABLE " + SystemTable.SCHEMA_COLUMNFAMILIES_CF + "("
+ + "keyspace_name text,"
+ + "columnfamily_name text,"
+ + "id int,"
+ + "type text,"
+ + "comparator text,"
+ + "subcomparator text,"
+ + "comment text,"
+ + "read_repair_chance double,"
+ + "local_read_repair_chance double,"
+ + "replicate_on_write boolean,"
+ + "gc_grace_seconds int,"
+ + "default_validator text,"
+ + "key_validator text,"
+ + "min_compaction_threshold int,"
+ + "max_compaction_threshold int,"
+ + "key_alias text,"
+ + "bloom_filter_fp_chance double,"
+ + "caching text,"
+ + "compaction_strategy_class text,"
+ + "compression_parameters text,"
+ + "value_aliases text,"
+ + "column_aliases text,"
+ + "compaction_strategy_options text,"
+ + "PRIMARY KEY (keyspace_name, columnfamily_name)"
+ + ") WITH COMMENT='ColumnFamily definitions' AND gc_grace_seconds=8640");
+ public static final CFMetaData SchemaColumnsCf = compile(10, "CREATE TABLE " + SystemTable.SCHEMA_COLUMNS_CF + "("
+ + "keyspace_name text,"
+ + "columnfamily_name text,"
+ + "column_name text,"
+ + "validator text,"
+ + "index_type text,"
+ + "index_options text,"
+ + "index_name text,"
+ + "component_index int,"
+ + "PRIMARY KEY(keyspace_name, columnfamily_name, column_name)"
+ + ") WITH COMMENT='ColumnFamily column attributes' AND gc_grace_seconds=8640");
+
+ public static final CFMetaData HintsCf = compile(11, "CREATE TABLE " + SystemTable.HINTS_CF + " ("
+ + "target_id uuid,"
+ + "hint_id timeuuid,"
+ + "message_version int,"
+ + "mutation blob,"
+ + "PRIMARY KEY (target_id, hint_id, message_version)"
+ + ") WITH COMPACT STORAGE AND COMMENT='hints awaiting delivery'");
+
+ public static final CFMetaData PeersCf = compile(12, "CREATE TABLE " + SystemTable.PEERS_CF + " ("
+ + "token_bytes blob PRIMARY KEY,"
+ + "peer inet"
+ + ") WITH COMMENT='known peers in the cluster'");
+
+ public static final CFMetaData LocalCf = compile(13, "CREATE TABLE " + SystemTable.LOCAL_CF + " ("
+ + "key text PRIMARY KEY,"
+ + "token_bytes blob,"
+ + "cluster_name text,"
+ + "gossip_generation int,"
+ + "bootstrapped boolean,"
+ + "ring_id uuid,"
+ + "release_version text,"
+ + "thrift_version text,"
+ + "cql_version text"
+ + ") WITH COMMENT='information about the local node'");
public enum Caching
{
@@ -261,6 +269,29 @@ public final class CFMetaData
this.init();
}
+ private static CFMetaData compile(int id, String cql)
+ {
+ try
+ {
+ CreateColumnFamilyStatement statement = (CreateColumnFamilyStatement) QueryProcessor.parseStatement(cql).prepare().statement;
+ CFMetaData cfmd = newSystemMetadata(statement.columnFamily(), id, "", statement.comparator, null);
+ statement.applyPropertiesTo(cfmd);
+ return cfmd;
+ }
+ catch (InvalidRequestException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (RecognitionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (ConfigurationException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
private AbstractType<?> enforceSubccDefault(ColumnFamilyType cftype, AbstractType<?> subcc)
{
return (subcc == null) && (cftype == ColumnFamilyType.Super) ? BytesType.instance : subcc;
@@ -319,20 +350,9 @@ public final class CFMetaData
Schema.instance.addOldCfIdMapping(oldCfId, newCFMD.cfId);
return newCFMD.comment(comment)
- .readRepairChance(0)
- .dcLocalReadRepairChance(0)
- .gcGraceSeconds(0);
- }
-
- private static CFMetaData newSchemaMetadata(String cfName, int oldCfId, String comment, AbstractType<?> comparator, AbstractType<?> subcc)
- {
- /*
- * Schema column families needs a gc_grace (since they are replicated
- * on every node). That gc_grace should be high enough that no node
- * could be dead for that long a time.
- */
- int gcGrace = 120 * 24 * 3600; // 3 months
- return newSystemMetadata(cfName, oldCfId, comment, comparator, subcc).gcGraceSeconds(gcGrace);
+ .readRepairChance(0)
+ .dcLocalReadRepairChance(0)
+ .gcGraceSeconds(0);
}
public static CFMetaData newIndexMetadata(CFMetaData parent, ColumnDefinition info, AbstractType<?> columnComparator)
@@ -1184,8 +1204,8 @@ public final class CFMetaData
{
try
{
- CFMetaData cfm = new CFMetaData(result.getString("keyspace"),
- result.getString("columnfamily"),
+ CFMetaData cfm = new CFMetaData(result.getString("keyspace_name"),
+ result.getString("columnfamily_name"),
ColumnFamilyType.valueOf(result.getString("type")),
TypeParser.parse(result.getString("comparator")),
result.has("subcomparator") ? TypeParser.parse(result.getString("subcomparator")) : null);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/config/ColumnDefinition.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/ColumnDefinition.java b/src/java/org/apache/cassandra/config/ColumnDefinition.java
index ab20a5d..5568f72 100644
--- a/src/java/org/apache/cassandra/config/ColumnDefinition.java
+++ b/src/java/org/apache/cassandra/config/ColumnDefinition.java
@@ -61,31 +61,6 @@ public class ColumnDefinition
this.setIndexType(index_type, index_options);
}
- public static ColumnDefinition ascii(String name, Integer cidx)
- {
- return new ColumnDefinition(ByteBufferUtil.bytes(name), AsciiType.instance, null, null, null, cidx);
- }
-
- public static ColumnDefinition bool(String name, Integer cidx)
- {
- return new ColumnDefinition(ByteBufferUtil.bytes(name), BooleanType.instance, null, null, null, cidx);
- }
-
- public static ColumnDefinition utf8(String name, Integer cidx)
- {
- return new ColumnDefinition(ByteBufferUtil.bytes(name), UTF8Type.instance, null, null, null, cidx);
- }
-
- public static ColumnDefinition int32(String name, Integer cidx)
- {
- return new ColumnDefinition(ByteBufferUtil.bytes(name), Int32Type.instance, null, null, null, cidx);
- }
-
- public static ColumnDefinition double_(String name, Integer cidx)
- {
- return new ColumnDefinition(ByteBufferUtil.bytes(name), DoubleType.instance, null, null, null, cidx);
- }
-
public ColumnDefinition clone()
{
return new ColumnDefinition(name, validator, index_type, index_options, index_name, componentIndex);
@@ -179,6 +154,7 @@ public class ColumnDefinition
cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_type"));
cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_options"));
cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "index_name"));
+ cf.addColumn(DeletedColumn.create(ldt, timestamp, cfName, comparator.getString(name), "component_index"));
}
public void toSchema(RowMutation rm, String cfName, AbstractType<?> comparator, long timestamp)
@@ -249,7 +225,7 @@ public class ColumnDefinition
if (result.has("component_index"))
componentIndex = result.getInt("component_index");
- cds.add(new ColumnDefinition(cfm.getColumnDefinitionComparator(componentIndex).fromString(result.getString("column")),
+ cds.add(new ColumnDefinition(cfm.getColumnDefinitionComparator(componentIndex).fromString(result.getString("column_name")),
TypeParser.parse(result.getString("validator")),
index_type,
index_options,
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 9a073a9..d94782c 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -433,17 +433,8 @@ public class DatabaseDescriptor
// Hardcoded system tables
KSMetaData systemMeta = KSMetaData.systemKeyspace();
- Schema.instance.load(CFMetaData.StatusCf);
- Schema.instance.load(CFMetaData.HintsCf);
- Schema.instance.load(CFMetaData.MigrationsCf);
- Schema.instance.load(CFMetaData.SchemaCf);
- Schema.instance.load(CFMetaData.IndexCf);
- Schema.instance.load(CFMetaData.NodeIdCf);
- Schema.instance.load(CFMetaData.VersionCf);
- Schema.instance.load(CFMetaData.SchemaKeyspacesCf);
- Schema.instance.load(CFMetaData.SchemaColumnFamiliesCf);
- Schema.instance.load(CFMetaData.SchemaColumnsCf);
- Schema.instance.load(CFMetaData.HostIdCf);
+ for (CFMetaData cfm : systemMeta.cfMetaData().values())
+ Schema.instance.load(cfm);
Schema.instance.addSystemTable(systemMeta);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/config/KSMetaData.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/config/KSMetaData.java b/src/java/org/apache/cassandra/config/KSMetaData.java
index cba40e5..bb1611a 100644
--- a/src/java/org/apache/cassandra/config/KSMetaData.java
+++ b/src/java/org/apache/cassandra/config/KSMetaData.java
@@ -20,7 +20,6 @@ package org.apache.cassandra.config;
import java.io.IOException;
import java.util.*;
-import com.google.common.base.Objects;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
@@ -28,13 +27,10 @@ import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.QueryPath;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.locator.*;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.KsDef;
-import org.apache.cassandra.thrift.ColumnDef;
import static org.apache.cassandra.utils.FBUtilities.*;
@@ -75,17 +71,18 @@ public final class KSMetaData
public static KSMetaData systemKeyspace()
{
- List<CFMetaData> cfDefs = Arrays.asList(CFMetaData.StatusCf,
+ List<CFMetaData> cfDefs = Arrays.asList(CFMetaData.LocalCf,
+ CFMetaData.PeersCf,
CFMetaData.HintsCf,
- CFMetaData.MigrationsCf,
- CFMetaData.SchemaCf,
CFMetaData.IndexCf,
CFMetaData.NodeIdCf,
- CFMetaData.VersionCf,
CFMetaData.SchemaKeyspacesCf,
CFMetaData.SchemaColumnFamiliesCf,
CFMetaData.SchemaColumnsCf,
- CFMetaData.HostIdCf);
+ CFMetaData.OldStatusCf,
+ CFMetaData.OldHintsCf,
+ CFMetaData.MigrationsCf,
+ CFMetaData.SchemaCf);
return new KSMetaData(Table.SYSTEM_TABLE, LocalStrategy.class, Collections.<String, String>emptyMap(), true, cfDefs);
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql/jdbc/JdbcInetAddress.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql/jdbc/JdbcInetAddress.java b/src/java/org/apache/cassandra/cql/jdbc/JdbcInetAddress.java
new file mode 100644
index 0000000..ac0bc09
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql/jdbc/JdbcInetAddress.java
@@ -0,0 +1,101 @@
+/*
+ * 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.cql.jdbc;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+import java.sql.Types;
+
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class JdbcInetAddress extends AbstractJdbcType<InetAddress>
+{
+ public static final JdbcInetAddress instance = new JdbcInetAddress();
+
+ JdbcInetAddress()
+ {
+ }
+
+ public boolean isCaseSensitive()
+ {
+ return false;
+ }
+
+ public int getScale(InetAddress obj)
+ {
+ return 0;
+ }
+
+ public int getPrecision(InetAddress obj)
+ {
+ return obj.toString().length();
+ }
+
+ public boolean isCurrency()
+ {
+ return false;
+ }
+
+ public boolean isSigned()
+ {
+ return true;
+ }
+
+ public String toString(InetAddress obj)
+ {
+ return obj.getHostAddress();
+ }
+
+ public boolean needsQuotes()
+ {
+ return false;
+ }
+
+ public String getString(ByteBuffer bytes)
+ {
+ return compose(bytes).getHostAddress();
+ }
+
+ public Class<InetAddress> getType()
+ {
+ return InetAddress.class;
+ }
+
+ public int getJdbcType()
+ {
+ return Types.OTHER;
+ }
+
+ public InetAddress compose(ByteBuffer bytes)
+ {
+ try
+ {
+ return InetAddress.getByAddress(ByteBufferUtil.getArray(bytes));
+ }
+ catch (UnknownHostException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+
+ public ByteBuffer decompose(InetAddress value)
+ {
+ return ByteBuffer.wrap(value.getAddress());
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql3/CFPropDefs.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/CFPropDefs.java b/src/java/org/apache/cassandra/cql3/CFPropDefs.java
index 50c15a6..37bcc06 100644
--- a/src/java/org/apache/cassandra/cql3/CFPropDefs.java
+++ b/src/java/org/apache/cassandra/cql3/CFPropDefs.java
@@ -67,6 +67,7 @@ public class CFPropDefs
comparators.put("decimal", "DecimalType");
comparators.put("double", "DoubleType");
comparators.put("float", "FloatType");
+ comparators.put("inet", "InetAddressType");
comparators.put("int", "Int32Type");
comparators.put("text", "UTF8Type");
comparators.put("timestamp", "DateType");
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql3/Cql.g
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Cql.g b/src/java/org/apache/cassandra/cql3/Cql.g
index 1900538..bbebedc 100644
--- a/src/java/org/apache/cassandra/cql3/Cql.g
+++ b/src/java/org/apache/cassandra/cql3/Cql.g
@@ -533,6 +533,7 @@ native_type returns [String str]
| K_DECIMAL
| K_DOUBLE
| K_FLOAT
+ | K_INET
| K_INT
| K_TEXT
| K_TIMESTAMP
@@ -623,6 +624,7 @@ K_COUNTER: C O U N T E R;
K_DECIMAL: D E C I M A L;
K_DOUBLE: D O U B L E;
K_FLOAT: F L O A T;
+K_INET: I N E T;
K_INT: I N T;
K_TEXT: T E X T;
K_UUID: U U I D;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql3/QueryProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
index e51a53c..c74be16 100644
--- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
@@ -121,33 +121,53 @@ public class QueryProcessor
return processStatement(getStatement(queryString, clientState).statement, clientState, Collections.<ByteBuffer>emptyList());
}
- public static UntypedResultSet resultify(String queryString, Row row)
+ public static UntypedResultSet processInternal(String query)
{
- SelectStatement ss;
try
{
- ss = (SelectStatement) getStatement(queryString, null).statement;
+ ClientState state = new ClientState();
+ CqlResult result = processStatement(getStatement(query, state).statement, state, Collections.<ByteBuffer>emptyList());
+ return new UntypedResultSet(result.rows);
}
- catch (InvalidRequestException e)
+ catch (UnavailableException e)
{
throw new RuntimeException(e);
}
+ catch (InvalidRequestException e)
+ {
+ throw new AssertionError(e);
+ }
catch (RecognitionException e)
{
+ throw new AssertionError(e);
+ }
+ catch (TimedOutException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (SchemaDisagreementException e)
+ {
throw new RuntimeException(e);
}
+ }
- List<CqlRow> cqlRows;
+ public static UntypedResultSet resultify(String query, Row row)
+ {
+ SelectStatement ss;
try
{
- cqlRows = ss.process(Collections.singletonList(row));
+ ss = (SelectStatement) getStatement(query, null).statement;
+ List<CqlRow> cqlRows = ss.process(Collections.singletonList(row));
+ return new UntypedResultSet(cqlRows);
}
catch (InvalidRequestException e)
{
- throw new RuntimeException(e);
+ throw new AssertionError(e);
+ }
+ catch (RecognitionException e)
+ {
+ throw new AssertionError(e);
}
-
- return new UntypedResultSet(cqlRows);
}
public static CqlPreparedResult prepare(String queryString, ClientState clientState)
@@ -215,7 +235,7 @@ public class QueryProcessor
return statement.prepare();
}
- private static ParsedStatement parseStatement(String queryStr) throws InvalidRequestException, RecognitionException
+ public static ParsedStatement parseStatement(String queryStr) throws InvalidRequestException, RecognitionException
{
try
{
@@ -242,5 +262,4 @@ public class QueryProcessor
throw ire;
}
}
-
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
index 15bf7cd..1280851 100644
--- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -18,6 +18,7 @@
*/
package org.apache.cassandra.cql3;
+import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Iterator;
@@ -26,10 +27,7 @@ import java.util.Map;
import com.google.common.collect.AbstractIterator;
-import org.apache.cassandra.db.marshal.BooleanType;
-import org.apache.cassandra.db.marshal.DoubleType;
-import org.apache.cassandra.db.marshal.Int32Type;
-import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.CqlRow;
import org.apache.hadoop.io.UTF8;
@@ -106,5 +104,16 @@ public class UntypedResultSet implements Iterable<UntypedResultSet.Row>
{
return data.get(column);
}
+
+ public InetAddress getInetAddress(String column)
+ {
+ return InetAddressType.instance.compose(data.get(column));
+ }
+
+ @Override
+ public String toString()
+ {
+ return data.toString();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
index b44b3ca..56ce563 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
@@ -46,7 +46,7 @@ import org.apache.cassandra.utils.Pair;
/** A <code>CREATE COLUMNFAMILY</code> parsed from a CQL query statement. */
public class CreateColumnFamilyStatement extends SchemaAlteringStatement
{
- private AbstractType<?> comparator;
+ public AbstractType<?> comparator;
private AbstractType<?> defaultValidator;
private AbstractType<?> keyValidator;
@@ -99,15 +99,7 @@ public class CreateColumnFamilyStatement extends SchemaAlteringStatement
ColumnFamilyType.Standard,
comparator,
null);
-
- newCFMD.defaultValidator(defaultValidator)
- .columnMetadata(getColumns())
- .keyValidator(keyValidator)
- .keyAlias(keyAlias)
- .columnAliases(columnAliases)
- .valueAlias(valueAlias);
-
- properties.applyToCFMetadata(newCFMD);
+ applyPropertiesTo(newCFMD);
}
catch (ConfigurationException e)
{
@@ -116,6 +108,18 @@ public class CreateColumnFamilyStatement extends SchemaAlteringStatement
return newCFMD;
}
+ public void applyPropertiesTo(CFMetaData cfmd) throws InvalidRequestException, ConfigurationException
+ {
+ cfmd.defaultValidator(defaultValidator)
+ .columnMetadata(getColumns())
+ .keyValidator(keyValidator)
+ .keyAlias(keyAlias)
+ .columnAliases(columnAliases)
+ .valueAlias(valueAlias);
+
+ properties.applyToCFMetadata(cfmd);
+ }
+
public static class RawStatement extends CFStatement
{
private final Map<ColumnIdentifier, String> definitions = new HashMap<ColumnIdentifier, String>();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/Column.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java
index 2a53ff3..81b30a0 100644
--- a/src/java/org/apache/cassandra/db/Column.java
+++ b/src/java/org/apache/cassandra/db/Column.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.db;
import java.io.IOException;
+import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.ArrayList;
@@ -321,6 +322,11 @@ public class Column implements IColumn
return new Column(decomposeName(names), value, timestamp);
}
+ public static IColumn create(InetAddress value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), InetAddressType.instance.decompose(value), timestamp);
+ }
+
static ByteBuffer decomposeName(String... names)
{
assert names.length > 0;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/Directories.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java
index e8d61c9..02b1f25 100644
--- a/src/java/org/apache/cassandra/db/Directories.java
+++ b/src/java/org/apache/cassandra/db/Directories.java
@@ -395,8 +395,9 @@ public class Directories
/**
* To check if sstables needs migration, we look at the System directory.
- * If it contains a directory for the status cf, we'll attempt a sstable
+ * If it does not contain a directory for the schema cfs, we'll attempt a sstable
* migration.
+ *
* Note that it is mostly harmless to try a migration uselessly, except
* maybe for some wasted cpu cycles.
*/
@@ -410,7 +411,7 @@ public class Directories
{
File systemDir = new File(location, Table.SYSTEM_TABLE);
hasSystemKeyspace |= (systemDir.exists() && systemDir.isDirectory());
- File statusCFDir = new File(systemDir, SystemTable.STATUS_CF);
+ File statusCFDir = new File(systemDir, SystemTable.SCHEMA_KEYSPACES_CF);
if (statusCFDir.exists())
return false;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/HintedHandOffManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
index d6031ca..9152dd6 100644
--- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java
+++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java
@@ -37,6 +37,10 @@ import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.CompositeType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -59,19 +63,18 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet;
/**
- * For each endpoint for which we have hints, there is a row in the system hints CF.
- * The key for this row is ByteBuffer.wrap(string), i.e. "127.0.0.1".
- * (We have to use String keys for compatibility with OPP.)
- * SuperColumns in these rows are the mutations to replay, with uuid names:
+ * The hint schema looks like this:
*
- * <dest token>: { // key
- * <uuid>: { // supercolumn
- * mutation: <mutation> // subcolumn
- * version: <mutation serialization version>
- * table: <table of hinted mutation>
- * key: <key of hinted mutation>
- * }
- * }
+ * CREATE TABLE hints (
+ * target_id uuid,
+ * hint_id timeuuid,
+ * message_version int,
+ * mutation blob,
+ * PRIMARY KEY (target_id, hint_id, message_version)
+ * ) WITH COMPACT STORAGE;
+ *
+ * Thus, for each node in the cluster we treat its uuid as the partition key; each hint is a logical row
+ * (physical composite column) containing the mutation to replay and associated metadata.
*
* When FailureDetector signals that a node that was down is back up, we page through
* the hinted mutations and send them over one at a time, waiting for
@@ -84,16 +87,12 @@ import org.cliffc.high_scale_lib.NonBlockingHashSet;
public class HintedHandOffManager implements HintedHandOffManagerMBean
{
public static final HintedHandOffManager instance = new HintedHandOffManager();
- public static final String HINTS_CF = "HintsColumnFamily";
private static final Logger logger = LoggerFactory.getLogger(HintedHandOffManager.class);
private static final int PAGE_SIZE = 128;
private static final int LARGE_NUMBER = 65536; // 64k nodes ought to be enough for anybody.
- // in 0.8, subcolumns were KS-CF bytestrings, and the data was stored in the "normal" storage there.
- // (so replay always consisted of sending an entire row,
- // no matter how little was part of the mutation that created the hint.)
- private static final String SEPARATOR_08 = "-";
+ static final CompositeType comparator = CompositeType.getInstance(Arrays.<AbstractType<?>>asList(UUIDType.instance, Int32Type.instance));
private final NonBlockingHashSet<InetAddress> queuedDeliveries = new NonBlockingHashSet<InetAddress>();
@@ -133,10 +132,10 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
responseHandler.get();
}
- private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer hintId, long timestamp) throws IOException
+ private static void deleteHint(ByteBuffer tokenBytes, ByteBuffer columnName, long timestamp) throws IOException
{
RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, tokenBytes);
- rm.delete(new QueryPath(HINTS_CF, hintId), timestamp);
+ rm.delete(new QueryPath(SystemTable.HINTS_CF, null, columnName), timestamp);
rm.applyUnsafe(); // don't bother with commitlog since we're going to flush as soon as we're done with delivery
}
@@ -162,7 +161,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
UUID hostId = StorageService.instance.getTokenMetadata().getHostId(endpoint);
ByteBuffer hostIdBytes = ByteBuffer.wrap(UUIDGen.decompose(hostId));
final RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, hostIdBytes);
- rm.delete(new QueryPath(HINTS_CF), System.currentTimeMillis());
+ rm.delete(new QueryPath(SystemTable.HINTS_CF), System.currentTimeMillis());
// execute asynchronously to avoid blocking caller (which may be processing gossip)
Runnable runnable = new Runnable()
@@ -186,7 +185,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
private Future<?> compact() throws ExecutionException, InterruptedException
{
- final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
+ final ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
hintStore.forceBlockingFlush();
ArrayList<Descriptor> descriptors = new ArrayList<Descriptor>();
for (SSTable sstable : hintStore.getSSTables())
@@ -273,7 +272,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
}
});
- ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
+ ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
if (hintStore.isEmpty())
return; // nothing to do, don't confuse users by logging a no-op handoff
@@ -321,12 +320,11 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
delivery:
while (true)
{
- QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, pageSize);
+ QueryFilter filter = QueryFilter.getSliceFilter(epkey, new QueryPath(SystemTable.HINTS_CF), startColumn, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, pageSize);
ColumnFamily hintsPage = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(filter), (int)(System.currentTimeMillis() / 1000));
if (pagingFinished(hintsPage, startColumn))
break;
- page:
for (IColumn hint : hintsPage.getSortedColumns())
{
// Skip tombstones:
@@ -338,30 +336,14 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
continue;
startColumn = hint.name();
- for (IColumn subColumn : hint.getSubColumns())
- {
- // both 0.8 and 1.0 column names are UTF8 strings, so this check is safe
- if (ByteBufferUtil.string(subColumn.name()).contains(SEPARATOR_08))
- {
- logger.debug("0.8-style hint found. This should have been taken care of by purgeIncompatibleHints");
- deleteHint(hostIdBytes, hint.name(), hint.maxTimestamp());
- continue page;
- }
- }
- IColumn versionColumn = hint.getSubColumn(ByteBufferUtil.bytes("version"));
- IColumn tableColumn = hint.getSubColumn(ByteBufferUtil.bytes("table"));
- IColumn keyColumn = hint.getSubColumn(ByteBufferUtil.bytes("key"));
- IColumn mutationColumn = hint.getSubColumn(ByteBufferUtil.bytes("mutation"));
- assert versionColumn != null;
- assert tableColumn != null;
- assert keyColumn != null;
- assert mutationColumn != null;
- DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(mutationColumn.value()));
+ ByteBuffer[] components = comparator.split(hint.name());
+ int version = Int32Type.instance.compose(components[1]);
+ DataInputStream in = new DataInputStream(ByteBufferUtil.inputStream(hint.value()));
RowMutation rm;
try
{
- rm = RowMutation.serializer.deserialize(in, ByteBufferUtil.toInt(versionColumn.value()));
+ rm = RowMutation.serializer.deserialize(in, version);
}
catch (UnknownColumnFamilyException e)
{
@@ -414,7 +396,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
if (logger.isDebugEnabled())
logger.debug("Started scheduleAllDeliveries");
- ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HINTS_CF);
+ ColumnFamilyStore hintStore = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(SystemTable.HINTS_CF);
IPartitioner p = StorageService.getPartitioner();
RowPosition minPos = p.getMinimumToken().minKeyBound();
Range<RowPosition> range = new Range<RowPosition>(minPos, minPos, p);
@@ -489,7 +471,7 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
private List<Row> getHintsSlice(int columnCount)
{
// ColumnParent for HintsCF...
- ColumnParent parent = new ColumnParent(HINTS_CF);
+ ColumnParent parent = new ColumnParent(SystemTable.HINTS_CF);
// Get count # of columns...
SliceQueryFilter predicate = new SliceQueryFilter(ByteBufferUtil.EMPTY_BYTE_BUFFER,
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/RowMutation.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java
index 1bce73b..7024b01 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -30,6 +30,7 @@ import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.io.IColumnSerializer;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.FastByteArrayInputStream;
@@ -99,25 +100,12 @@ public class RowMutation implements IMutation
/**
* Returns mutation representing a Hints to be sent to <code>address</code>
- * as soon as it becomes available.
- * The format is the following:
- *
- * HintsColumnFamily: { // cf
- * <dest token>: { // key
- * <uuid>: { // super-column
- * table: <table> // columns
- * key: <key>
- * mutation: <mutation>
- * version: <version>
- * }
- * }
- * }
- *
+ * as soon as it becomes available. See HintedHandoffManager for more details.
*/
- public static RowMutation hintFor(RowMutation mutation, ByteBuffer token) throws IOException
+ public static RowMutation hintFor(RowMutation mutation, UUID targetId) throws IOException
{
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, token);
- ByteBuffer hintId = ByteBuffer.wrap(UUIDGen.getTimeUUIDBytes());
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, UUIDType.instance.decompose(targetId));
+ UUID hintId = UUIDGen.makeType1UUIDFromHost(FBUtilities.getBroadcastAddress());
// determine the TTL for the RowMutation
// this is set at the smallest GCGraceSeconds for any of the CFs in the RM
@@ -126,22 +114,10 @@ public class RowMutation implements IMutation
for (ColumnFamily cf : mutation.getColumnFamilies())
ttl = Math.min(ttl, cf.metadata().getGcGraceSeconds());
- // serialized RowMutation
- QueryPath path = new QueryPath(HintedHandOffManager.HINTS_CF, hintId, ByteBufferUtil.bytes("mutation"));
+ // serialize the hint with id and version as a composite column name
+ QueryPath path = new QueryPath(SystemTable.HINTS_CF, null, HintedHandOffManager.comparator.decompose(hintId, MessagingService.current_version));
rm.add(path, ByteBuffer.wrap(mutation.getSerializedBuffer(MessagingService.current_version)), System.currentTimeMillis(), ttl);
- // serialization version
- path = new QueryPath(HintedHandOffManager.HINTS_CF, hintId, ByteBufferUtil.bytes("version"));
- rm.add(path, ByteBufferUtil.bytes(MessagingService.current_version), System.currentTimeMillis(), ttl);
-
- // table
- path = new QueryPath(HintedHandOffManager.HINTS_CF, hintId, ByteBufferUtil.bytes("table"));
- rm.add(path, ByteBufferUtil.bytes(mutation.getTable()), System.currentTimeMillis(), ttl);
-
- // key
- path = new QueryPath(HintedHandOffManager.HINTS_CF, hintId, ByteBufferUtil.bytes("key"));
- rm.add(path, mutation.key(), System.currentTimeMillis(), ttl);
-
return rm;
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/SystemTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index 6fb68f0..49c208e 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -20,17 +20,19 @@ package org.apache.cassandra.db;
import java.io.IOError;
import java.io.IOException;
import java.net.InetAddress;
-import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutionException;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableSortedSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.cql.QueryProcessor;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
@@ -41,6 +43,7 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.Constants;
+import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.NodeId;
@@ -49,27 +52,26 @@ import org.apache.cassandra.utils.UUIDGen;
public class SystemTable
{
private static final Logger logger = LoggerFactory.getLogger(SystemTable.class);
- public static final String STATUS_CF = "LocationInfo"; // keep the old CF string for backwards-compatibility
+
+ // see CFMetaData for schema definitions
+ public static final String PEERS_CF = "peers";
+ public static final String LOCAL_CF = "local";
public static final String INDEX_CF = "IndexInfo";
public static final String NODE_ID_CF = "NodeIdInfo";
- public static final String VERSION_CF = "Versions";
- public static final String HOST_ID_CF = "HostUUID";
+ public static final String HINTS_CF = "hints";
// see layout description in the DefsTable class header
public static final String SCHEMA_KEYSPACES_CF = "schema_keyspaces";
public static final String SCHEMA_COLUMNFAMILIES_CF = "schema_columnfamilies";
public static final String SCHEMA_COLUMNS_CF = "schema_columns";
- private static final ByteBuffer LOCATION_KEY = ByteBufferUtil.bytes("L");
- private static final ByteBuffer RING_KEY = ByteBufferUtil.bytes("Ring");
- private static final ByteBuffer BOOTSTRAP_KEY = ByteBufferUtil.bytes("Bootstrap");
- private static final ByteBuffer COOKIE_KEY = ByteBufferUtil.bytes("Cookies");
- private static final ByteBuffer BOOTSTRAP = ByteBufferUtil.bytes("B");
- private static final ByteBuffer TOKEN = ByteBufferUtil.bytes("Token");
- private static final ByteBuffer GENERATION = ByteBufferUtil.bytes("Generation");
- private static final ByteBuffer CLUSTERNAME = ByteBufferUtil.bytes("ClusterName");
+ @Deprecated
+ public static final String OLD_STATUS_CF = "LocationInfo";
+ @Deprecated
+ public static final String OLD_HINTS_CF = "HintsColumnFamily";
+
+ private static final ByteBuffer LOCAL_KEY = ByteBufferUtil.bytes("local");
private static final ByteBuffer CURRENT_LOCAL_NODE_ID_KEY = ByteBufferUtil.bytes("CurrentLocal");
private static final ByteBuffer ALL_LOCAL_NODE_ID_KEY = ByteBufferUtil.bytes("Local");
- private static final ByteBuffer HOST_ID_KEY = ByteBufferUtil.bytes("Id");
private static DecoratedKey decorate(ByteBuffer key)
{
@@ -79,64 +81,63 @@ public class SystemTable
public static void finishStartup() throws IOException
{
setupVersion();
- purgeIncompatibleHints();
+ try
+ {
+ upgradeSystemData();
+ }
+ catch (ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
}
private static void setupVersion() throws IOException
{
- RowMutation rm;
- ColumnFamily cf;
-
- rm = new RowMutation(Table.SYSTEM_TABLE, ByteBufferUtil.bytes("build"));
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, VERSION_CF);
- cf.addColumn(new Column(ByteBufferUtil.bytes("version"), ByteBufferUtil.bytes(FBUtilities.getReleaseVersionString()), FBUtilities.timestampMicros()));
- rm.add(cf);
- rm.apply();
-
- rm = new RowMutation(Table.SYSTEM_TABLE, ByteBufferUtil.bytes("cql"));
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, VERSION_CF);
- cf.addColumn(new Column(ByteBufferUtil.bytes("version"), ByteBufferUtil.bytes(QueryProcessor.CQL_VERSION.toString()), FBUtilities.timestampMicros()));
- rm.add(cf);
- rm.apply();
-
- rm = new RowMutation(Table.SYSTEM_TABLE, ByteBufferUtil.bytes("thrift"));
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, VERSION_CF);
- cf.addColumn(new Column(ByteBufferUtil.bytes("version"), ByteBufferUtil.bytes(Constants.VERSION), FBUtilities.timestampMicros()));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(FBUtilities.getReleaseVersionString(), FBUtilities.timestampMicros(), "release_version"));
+ cf.addColumn(Column.create(QueryProcessor.CQL_VERSION.toString(), FBUtilities.timestampMicros(), "cql_version"));
+ cf.addColumn(Column.create(Constants.VERSION, FBUtilities.timestampMicros(), "thrift_version"));
rm.add(cf);
rm.apply();
}
- /** if hints become incompatible across versions of cassandra, that logic (and associated purging) is managed here. */
- private static void purgeIncompatibleHints() throws IOException
+ /** if system data becomes incompatible across versions of cassandra, that logic (and associated purging) is managed here */
+ private static void upgradeSystemData() throws IOException, ExecutionException, InterruptedException
{
- ByteBuffer upgradeMarker = ByteBufferUtil.bytes("Pre-1.2 hints purged");
Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(COOKIE_KEY), new QueryPath(STATUS_CF), upgradeMarker);
- ColumnFamily cf = table.getColumnFamilyStore(STATUS_CF).getColumnFamily(filter);
- if (cf != null)
- {
- logger.debug("Pre-1.0 hints already purged");
- return;
+ ColumnFamilyStore oldStatusCfs = table.getColumnFamilyStore(OLD_STATUS_CF);
+ if (oldStatusCfs.getSSTables().size() > 0)
+ {
+ SortedSet<ByteBuffer> cols = new TreeSet<ByteBuffer>(BytesType.instance);
+ cols.add(ByteBufferUtil.bytes("ClusterName"));
+ cols.add(ByteBufferUtil.bytes("Token"));
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(ByteBufferUtil.bytes("L")), new QueryPath(OLD_STATUS_CF), cols);
+ ColumnFamily oldCf = oldStatusCfs.getColumnFamily(filter);
+ Iterator<IColumn> oldColumns = oldCf.columns.iterator();
+
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(oldColumns.next().value(), FBUtilities.timestampMicros(), "cluster_name"));
+ cf.addColumn(Column.create(oldColumns.next().value(), FBUtilities.timestampMicros(), "token_bytes"));
+ // (assume that any node getting upgraded was bootstrapped, since that was stored in a separate row for no particular reason)
+ cf.addColumn(Column.create(true, FBUtilities.timestampMicros(), "bootstrapped"));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
+ rm.add(cf);
+ rm.apply();
+
+ oldStatusCfs.truncate();
}
- // marker not found. Snapshot + remove hints and add the marker
- ColumnFamilyStore hintsCfs = Table.open(Table.SYSTEM_TABLE).getColumnFamilyStore(HintedHandOffManager.HINTS_CF);
- if (hintsCfs.getSSTables().size() > 0)
+ ColumnFamilyStore oldHintsCfs = table.getColumnFamilyStore(OLD_HINTS_CF);
+ if (oldHintsCfs.getSSTables().size() > 0)
{
logger.info("Possible old-format hints found. Truncating");
- try
- {
- hintsCfs.truncate();
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
+ oldHintsCfs.truncate();
}
- logger.debug("Marking pre-1.0 hints purged");
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, COOKIE_KEY);
- rm.add(new QueryPath(STATUS_CF, null, upgradeMarker), ByteBufferUtil.bytes("oh yes, they were purged"), FBUtilities.timestampMicros());
- rm.apply();
}
/**
@@ -149,10 +150,11 @@ public class SystemTable
removeToken(token);
return;
}
+
IPartitioner p = StorageService.getPartitioner();
- ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, STATUS_CF);
- cf.addColumn(new Column(p.getTokenFactory().toByteArray(token), ByteBuffer.wrap(ep.getAddress()), FBUtilities.timestampMicros()));
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, RING_KEY);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, PEERS_CF);
+ cf.addColumn(Column.create(ep, FBUtilities.timestampMicros(), "peer"));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, p.getTokenFactory().toByteArray(token));
rm.add(cf);
try
{
@@ -162,7 +164,7 @@ public class SystemTable
{
throw new IOError(e);
}
- forceBlockingFlush(STATUS_CF);
+ forceBlockingFlush(PEERS_CF);
}
/**
@@ -171,8 +173,8 @@ public class SystemTable
public static synchronized void removeToken(Token token)
{
IPartitioner p = StorageService.getPartitioner();
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, RING_KEY);
- rm.delete(new QueryPath(STATUS_CF, null, p.getTokenFactory().toByteArray(token)), FBUtilities.timestampMicros());
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, p.getTokenFactory().toByteArray(token));
+ rm.delete(new QueryPath(PEERS_CF, null, null), FBUtilities.timestampMicros());
try
{
rm.apply();
@@ -181,7 +183,7 @@ public class SystemTable
{
throw new IOError(e);
}
- forceBlockingFlush(STATUS_CF);
+ forceBlockingFlush(PEERS_CF);
}
/**
@@ -190,9 +192,9 @@ public class SystemTable
public static synchronized void updateToken(Token token)
{
IPartitioner p = StorageService.getPartitioner();
- ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, STATUS_CF);
- cf.addColumn(new Column(SystemTable.TOKEN, p.getTokenFactory().toByteArray(token), FBUtilities.timestampMicros()));
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCATION_KEY);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(p.getTokenFactory().toByteArray(token), FBUtilities.timestampMicros(), "token_bytes"));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
rm.add(cf);
try
{
@@ -203,7 +205,7 @@ public class SystemTable
throw new IOError(e);
}
- forceBlockingFlush(STATUS_CF);
+ forceBlockingFlush(LOCAL_CF);
}
private static void forceBlockingFlush(String cfname)
@@ -228,28 +230,12 @@ public class SystemTable
*/
public static HashMap<Token, InetAddress> loadTokens()
{
- HashMap<Token, InetAddress> tokenMap = new HashMap<Token, InetAddress>();
IPartitioner p = StorageService.getPartitioner();
- Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getIdentityFilter(decorate(RING_KEY), new QueryPath(STATUS_CF));
- ColumnFamily cf = ColumnFamilyStore.removeDeleted(table.getColumnFamilyStore(STATUS_CF).getColumnFamily(filter), Integer.MAX_VALUE);
- if (cf != null)
- {
- for (IColumn column : cf.getSortedColumns())
- {
- try
- {
- ByteBuffer v = column.value();
- byte[] addr = new byte[v.remaining()];
- ByteBufferUtil.arrayCopy(v, v.position(), addr, 0, v.remaining());
- tokenMap.put(p.getTokenFactory().fromByteArray(column.name()), InetAddress.getByAddress(addr));
- }
- catch (UnknownHostException e)
- {
- throw new IOError(e);
- }
- }
- }
+
+ HashMap<Token, InetAddress> tokenMap = new HashMap<Token, InetAddress>();
+ for (UntypedResultSet.Row row : QueryProcessor.processInternal("SELECT * FROM system.peers"))
+ tokenMap.put(p.getTokenFactory().fromByteArray(row.getBytes("token_bytes")), row.getInetAddress("peer"));
+
return tokenMap;
}
@@ -275,10 +261,10 @@ public class SystemTable
throw ex;
}
- SortedSet<ByteBuffer> cols = new TreeSet<ByteBuffer>(BytesType.instance);
- cols.add(CLUSTERNAME);
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCATION_KEY), new QueryPath(STATUS_CF), cols);
- ColumnFamilyStore cfs = table.getColumnFamilyStore(STATUS_CF);
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCAL_KEY),
+ new QueryPath(LOCAL_CF),
+ ImmutableSortedSet.<ByteBuffer>of(ByteBufferUtil.bytes("cluster_name")));
+ ColumnFamilyStore cfs = table.getColumnFamilyStore(LOCAL_CF);
ColumnFamily cf = cfs.getColumnFamily(filter);
if (cf == null)
@@ -288,9 +274,9 @@ public class SystemTable
throw new ConfigurationException("Found system table files, but they couldn't be loaded!");
// no system files. this is a new node.
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCATION_KEY);
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, SystemTable.STATUS_CF);
- cf.addColumn(new Column(CLUSTERNAME, ByteBufferUtil.bytes(DatabaseDescriptor.getClusterName()), FBUtilities.timestampMicros()));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
+ cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(ByteBufferUtil.bytes(DatabaseDescriptor.getClusterName()), FBUtilities.timestampMicros(), "cluster_name"));
rm.add(cf);
rm.apply();
@@ -298,8 +284,7 @@ public class SystemTable
}
- IColumn clusterCol = cf.getColumn(CLUSTERNAME);
- assert clusterCol != null;
+ IColumn clusterCol = cf.columns.iterator().next();
String savedClusterName = ByteBufferUtil.string(clusterCol.value());
if (!DatabaseDescriptor.getClusterName().equals(savedClusterName))
throw new ConfigurationException("Saved cluster name " + savedClusterName + " != configured name " + DatabaseDescriptor.getClusterName());
@@ -308,16 +293,16 @@ public class SystemTable
public static Token getSavedToken()
{
Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCATION_KEY), new QueryPath(STATUS_CF), TOKEN);
- ColumnFamily cf = table.getColumnFamilyStore(STATUS_CF).getColumnFamily(filter);
- return cf == null ? null : StorageService.getPartitioner().getTokenFactory().fromByteArray(cf.getColumn(TOKEN).value());
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCAL_KEY), new QueryPath(LOCAL_CF), ByteBufferUtil.bytes("token_bytes"));
+ ColumnFamily cf = table.getColumnFamilyStore(LOCAL_CF).getColumnFamily(filter);
+ return cf == null ? null : StorageService.getPartitioner().getTokenFactory().fromByteArray(cf.columns.iterator().next().value());
}
public static int incrementAndGetGeneration() throws IOException
{
Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCATION_KEY), new QueryPath(STATUS_CF), GENERATION);
- ColumnFamily cf = table.getColumnFamilyStore(STATUS_CF).getColumnFamily(filter);
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCAL_KEY), new QueryPath(LOCAL_CF), ByteBufferUtil.bytes("gossip_generation"));
+ ColumnFamily cf = table.getColumnFamilyStore(LOCAL_CF).getColumnFamily(filter);
int generation;
if (cf == null)
@@ -330,7 +315,7 @@ public class SystemTable
else
{
// Other nodes will ignore gossip messages about a node that have a lower generation than previously seen.
- final int storedGeneration = ByteBufferUtil.toInt(cf.getColumn(GENERATION).value()) + 1;
+ final int storedGeneration = ByteBufferUtil.toInt(cf.columns.iterator().next().value()) + 1;
final int now = (int) (System.currentTimeMillis() / 1000);
if (storedGeneration >= now)
{
@@ -344,12 +329,12 @@ public class SystemTable
}
}
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCATION_KEY);
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, SystemTable.STATUS_CF);
- cf.addColumn(new Column(GENERATION, ByteBufferUtil.bytes(generation), FBUtilities.timestampMicros()));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
+ cf = ColumnFamily.create(Table.SYSTEM_TABLE, SystemTable.LOCAL_CF);
+ cf.addColumn(Column.create(generation, FBUtilities.timestampMicros(), "gossip_generation"));
rm.add(cf);
rm.apply();
- forceBlockingFlush(STATUS_CF);
+ forceBlockingFlush(LOCAL_CF);
return generation;
}
@@ -357,23 +342,21 @@ public class SystemTable
public static boolean isBootstrapped()
{
Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getNamesFilter(decorate(BOOTSTRAP_KEY),
- new QueryPath(STATUS_CF),
- BOOTSTRAP);
- ColumnFamily cf = table.getColumnFamilyStore(STATUS_CF).getColumnFamily(filter);
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCAL_KEY),
+ new QueryPath(LOCAL_CF),
+ ByteBufferUtil.bytes("bootstrapped"));
+ ColumnFamily cf = table.getColumnFamilyStore(LOCAL_CF).getColumnFamily(filter);
if (cf == null)
return false;
- IColumn c = cf.getColumn(BOOTSTRAP);
+ IColumn c = cf.iterator().next();
return c.value().get(c.value().position()) == 1;
}
public static void setBootstrapped(boolean isBootstrapped)
{
- ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, STATUS_CF);
- cf.addColumn(new Column(BOOTSTRAP,
- ByteBuffer.wrap(new byte[] { (byte) (isBootstrapped ? 1 : 0) }),
- FBUtilities.timestampMicros()));
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, BOOTSTRAP_KEY);
+ ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(isBootstrapped, FBUtilities.timestampMicros(), "bootstrapped"));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
rm.add(cf);
try
{
@@ -438,15 +421,16 @@ public class SystemTable
// Look up the Host UUID (return it if found)
Table table = Table.open(Table.SYSTEM_TABLE);
- QueryFilter filter = QueryFilter.getIdentityFilter(decorate(HOST_ID_KEY), new QueryPath(HOST_ID_CF));
- ColumnFamily cf = table.getColumnFamilyStore(HOST_ID_CF).getColumnFamily(filter);
-
+ QueryFilter filter = QueryFilter.getNamesFilter(decorate(LOCAL_KEY),
+ new QueryPath(LOCAL_CF),
+ ImmutableSortedSet.<ByteBuffer>of(ByteBufferUtil.bytes("ring_id")));
+ ColumnFamily cf = table.getColumnFamilyStore(LOCAL_CF).getColumnFamily(filter);
if (cf != null)
{
cf = ColumnFamilyStore.removeDeleted(cf, 0);
assert cf.getColumnCount() <= 1;
if (cf.getColumnCount() > 0)
- return UUIDGen.getUUID(cf.iterator().next().name());
+ return UUIDGen.getUUID(cf.iterator().next().value());
}
// ID not found, generate a new one, persist, and then return it.
@@ -455,10 +439,10 @@ public class SystemTable
logger.warn("No host ID found, created {} (Note: This should happen exactly once per node).", hostId);
- cf = ColumnFamily.create(Table.SYSTEM_TABLE, HOST_ID_CF);
- cf.addColumn(new Column(ByteBuffer.wrap(UUIDGen.decompose(hostId)), ByteBufferUtil.bytes(now), now));
+ cf = ColumnFamily.create(Table.SYSTEM_TABLE, LOCAL_CF);
+ cf.addColumn(Column.create(ByteBuffer.wrap(UUIDGen.decompose(hostId)), now, "ring_id"));
- RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, HOST_ID_KEY);
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, LOCAL_KEY);
rm.add(cf);
try
{
@@ -481,12 +465,12 @@ public class SystemTable
ByteBuffer id = null;
Table table = Table.open(Table.SYSTEM_TABLE);
QueryFilter filter = QueryFilter.getIdentityFilter(decorate(CURRENT_LOCAL_NODE_ID_KEY),
- new QueryPath(NODE_ID_CF));
+ new QueryPath(NODE_ID_CF));
ColumnFamily cf = table.getColumnFamilyStore(NODE_ID_CF).getColumnFamily(filter);
+ // Even though gc_grace==0 on System table, we can have a race where we get back tombstones (see CASSANDRA-2824)
+ cf = ColumnFamilyStore.removeDeleted(cf, 0);
if (cf != null)
{
- // Even though gc_grace==0 on System table, we can have a race where we get back tombstones (see CASSANDRA-2824)
- cf = ColumnFamilyStore.removeDeleted(cf, 0);
assert cf.getColumnCount() <= 1;
if (cf.getColumnCount() > 0)
id = cf.iterator().next().name();
@@ -512,10 +496,8 @@ public class SystemTable
*/
public static void writeCurrentLocalNodeId(NodeId oldNodeId, NodeId newNodeId, long now)
{
- ByteBuffer ip = ByteBuffer.wrap(FBUtilities.getBroadcastAddress().getAddress());
-
ColumnFamily cf = ColumnFamily.create(Table.SYSTEM_TABLE, NODE_ID_CF);
- cf.addColumn(new Column(newNodeId.bytes(), ip, now));
+ cf.addColumn(new Column(newNodeId.bytes(), ByteBufferUtil.EMPTY_BYTE_BUFFER, now));
ColumnFamily cf2 = cf.cloneMe();
if (oldNodeId != null)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java b/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java
index d319037..6536318 100644
--- a/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java
+++ b/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java
@@ -261,6 +261,8 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
}
}
+ public abstract ByteBuffer decompose(Object... objects);
+
public ByteBuffer compose(ByteBuffer bytes)
{
return bytes;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/marshal/CompositeType.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/marshal/CompositeType.java b/src/java/org/apache/cassandra/db/marshal/CompositeType.java
index 1b59ae9..4d26dac 100644
--- a/src/java/org/apache/cassandra/db/marshal/CompositeType.java
+++ b/src/java/org/apache/cassandra/db/marshal/CompositeType.java
@@ -18,6 +18,7 @@
package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -107,6 +108,29 @@ public class CompositeType extends AbstractCompositeType
return types.get(i);
}
+ public ByteBuffer decompose(Object... objects)
+ {
+ assert objects.length == types.size();
+
+ List<ByteBuffer> serialized = new ArrayList<ByteBuffer>(objects.length);
+ int totalLength = 0;
+ for (int i = 0; i < objects.length; i++)
+ {
+ ByteBuffer buffer = ((AbstractType) types.get(i)).decompose(objects[i]);
+ serialized.add(buffer);
+ totalLength += 2 + buffer.remaining() + 1;
+ }
+ ByteBuffer out = ByteBuffer.allocate(totalLength);
+ for (ByteBuffer bb : serialized)
+ {
+ putShortLength(out, bb.remaining());
+ out.put(bb);
+ out.put((byte) 0);
+ }
+ out.flip();
+ return out;
+ }
+
@Override
public boolean isCompatibleWith(AbstractType<?> previous)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
index e3f4839..40752e5 100644
--- a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
+++ b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
@@ -199,6 +199,11 @@ public class DynamicCompositeType extends AbstractCompositeType
return comparator;
}
+ public ByteBuffer decompose(Object... objects)
+ {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public boolean isCompatibleWith(AbstractType<?> previous)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/db/marshal/InetAddressType.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/marshal/InetAddressType.java b/src/java/org/apache/cassandra/db/marshal/InetAddressType.java
new file mode 100644
index 0000000..f621f64
--- /dev/null
+++ b/src/java/org/apache/cassandra/db/marshal/InetAddressType.java
@@ -0,0 +1,84 @@
+/*
+ * 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.db.marshal;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
+
+import org.apache.cassandra.cql.jdbc.JdbcInetAddress;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class InetAddressType extends AbstractType<InetAddress>
+{
+ public static final InetAddressType instance = new InetAddressType();
+
+ InetAddressType() {} // singleton
+
+ public InetAddress compose(ByteBuffer bytes)
+ {
+ return JdbcInetAddress.instance.compose(bytes);
+ }
+
+ public ByteBuffer decompose(InetAddress value)
+ {
+ return JdbcInetAddress.instance.decompose(value);
+ }
+
+ public int compare(ByteBuffer o1, ByteBuffer o2)
+ {
+ return ByteBufferUtil.compareUnsigned(o1, o2);
+ }
+
+ public String getString(ByteBuffer bytes)
+ {
+ return JdbcInetAddress.instance.getString(bytes);
+ }
+
+ public ByteBuffer fromString(String source) throws MarshalException
+ {
+ // Return an empty ByteBuffer for an empty string.
+ if (source.isEmpty())
+ return ByteBufferUtil.EMPTY_BYTE_BUFFER;
+
+ InetAddress address;
+
+ try
+ {
+ address = InetAddress.getByName(source);
+ }
+ catch (Exception e)
+ {
+ throw new MarshalException(String.format("unable to make inetaddress from '%s'", source), e);
+ }
+
+ return decompose(address);
+ }
+
+ public void validate(ByteBuffer bytes) throws MarshalException
+ {
+ try
+ {
+ InetAddress.getByAddress(ByteBufferUtil.getArray(bytes));
+ }
+ catch (UnknownHostException e)
+ {
+ throw new MarshalException(String.format("Expected 4 or 16 byte inetaddress; got %s", ByteBufferUtil.bytesToHex(bytes)));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/8ea2d2a6/src/java/org/apache/cassandra/locator/LocalStrategy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/locator/LocalStrategy.java b/src/java/org/apache/cassandra/locator/LocalStrategy.java
index a5a873b..09318de 100644
--- a/src/java/org/apache/cassandra/locator/LocalStrategy.java
+++ b/src/java/org/apache/cassandra/locator/LocalStrategy.java
@@ -18,11 +18,13 @@
package org.apache.cassandra.locator;
import java.net.InetAddress;
-import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.dht.RingPosition;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.utils.FBUtilities;
@@ -33,9 +35,22 @@ public class LocalStrategy extends AbstractReplicationStrategy
super(table, tokenMetadata, snitch, configOptions);
}
+ /**
+ * We need to override this even if we override calculateNaturalEndpoints,
+ * because the default implementation depends on token calculations but
+ * LocalStrategy may be used before tokens are set up.
+ */
+ @Override
+ public ArrayList<InetAddress> getNaturalEndpoints(RingPosition searchPosition)
+ {
+ ArrayList<InetAddress> l = new ArrayList<InetAddress>(1);
+ l.add(FBUtilities.getBroadcastAddress());
+ return l;
+ }
+
public List<InetAddress> calculateNaturalEndpoints(Token token, TokenMetadata metadata)
{
- return Arrays.asList(FBUtilities.getBroadcastAddress());
+ return Collections.singletonList(FBUtilities.getBroadcastAddress());
}
public int getReplicationFactor()
@@ -46,6 +61,6 @@ public class LocalStrategy extends AbstractReplicationStrategy
public void validateOptions() throws ConfigurationException
{
// LocalStrategy doesn't expect any options.
- warnOnUnexpectedOptions(Arrays.<String>asList());
+ warnOnUnexpectedOptions(Collections.<String>emptySet());
}
}