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/03/13 14:10:29 UTC
[3/7] Add type information to new schema_ columnfamilies and remove
thrift validation
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 e671511..5b9d5ba 100644
--- a/src/java/org/apache/cassandra/config/KSMetaData.java
+++ b/src/java/org/apache/cassandra/config/KSMetaData.java
@@ -25,6 +25,8 @@ import com.google.common.base.Objects;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
+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;
@@ -37,6 +39,7 @@ import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.ColumnDef;
import static org.apache.cassandra.db.migration.MigrationHelper.*;
+import static org.apache.cassandra.utils.FBUtilities.*;
public final class KSMetaData
{
@@ -46,7 +49,7 @@ public final class KSMetaData
private final Map<String, CFMetaData> cfMetaData;
public final boolean durableWrites;
- private KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, boolean durableWrites, Iterable<CFMetaData> cfDefs)
+ KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, boolean durableWrites, Iterable<CFMetaData> cfDefs)
{
this.name = name;
this.strategyClass = strategyClass == null ? NetworkTopologyStrategy.class : strategyClass;
@@ -124,47 +127,6 @@ public final class KSMetaData
return sb.toString();
}
- @Deprecated
- public static KSMetaData fromAvro(org.apache.cassandra.db.migration.avro.KsDef ks)
- {
- Class<? extends AbstractReplicationStrategy> repStratClass;
- try
- {
- String strategyClassName = convertOldStrategyName(ks.strategy_class.toString());
- repStratClass = (Class<AbstractReplicationStrategy>)Class.forName(strategyClassName);
- }
- catch (Exception ex)
- {
- throw new RuntimeException("Could not create ReplicationStrategy of type " + ks.strategy_class, ex);
- }
-
- Map<String, String> strategyOptions = new HashMap<String, String>();
- if (ks.strategy_options != null)
- {
- for (Map.Entry<CharSequence, CharSequence> e : ks.strategy_options.entrySet())
- {
- String name = e.getKey().toString();
- // Silently discard a replication_factor option to NTS.
- // The history is, people were creating CFs with the default settings (which in the CLI is NTS) and then
- // giving it a replication_factor option, which is nonsensical. Initially our strategy was to silently
- // ignore this option, but that turned out to confuse people more. So in 0.8.2 we switched to throwing
- // an exception in the NTS constructor, which would be turned into an InvalidRequestException for the
- // client. But, it also prevented startup for anyone upgrading without first cleaning that option out.
- if (repStratClass == NetworkTopologyStrategy.class && name.trim().toLowerCase().equals("replication_factor"))
- continue;
- strategyOptions.put(name, e.getValue().toString());
- }
- }
-
- int cfsz = ks.cf_defs.size();
- List<CFMetaData> cfMetaData = new ArrayList<CFMetaData>(cfsz);
- Iterator<org.apache.cassandra.db.migration.avro.CfDef> cfiter = ks.cf_defs.iterator();
- for (int i = 0; i < cfsz; i++)
- cfMetaData.add(CFMetaData.fromAvro(cfiter.next()));
-
- return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, ks.durable_writes, cfMetaData);
- }
-
public static String convertOldStrategyName(String name)
{
return name.replace("RackUnawareStrategy", "SimpleStrategy")
@@ -199,209 +161,115 @@ public final class KSMetaData
return ksdef;
}
- public RowMutation diff(KsDef newState, long modificationTimestamp)
+ public RowMutation diff(KSMetaData newState, long modificationTimestamp)
{
- KsDef curState = toThrift();
- RowMutation m = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
-
- for (KsDef._Fields field : KsDef._Fields.values())
- {
- if (field.equals(KsDef._Fields.CF_DEFS))
- continue;
-
- Object curValue = curState.getFieldValue(field);
- Object newValue = newState.getFieldValue(field);
-
- if (Objects.equal(curValue, newValue))
- continue;
-
- m.add(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF, null, AsciiType.instance.fromString(field.getFieldName())),
- valueAsBytes(newValue),
- modificationTimestamp);
- }
-
- return m;
+ return newState.toSchema(modificationTimestamp);
}
public KSMetaData reloadAttributes() throws IOException
{
Row ksDefRow = SystemTable.readSchemaRow(name);
- if (ksDefRow.cf == null || ksDefRow.cf.isEmpty())
+ if (ksDefRow.cf == null)
throw new IOException(String.format("%s not found in the schema definitions table (%s).", name, SystemTable.SCHEMA_KEYSPACES_CF));
- return fromSchema(ksDefRow.cf, null);
+ return fromSchema(ksDefRow, Collections.<CFMetaData>emptyList());
}
- public List<RowMutation> dropFromSchema(long timestamp)
+ public RowMutation dropFromSchema(long timestamp)
{
- List<RowMutation> mutations = new ArrayList<RowMutation>();
-
- RowMutation ksMutation = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
- ksMutation.delete(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF), timestamp);
- mutations.add(ksMutation);
-
- for (CFMetaData cfm : cfMetaData.values())
- mutations.add(cfm.dropFromSchema(timestamp));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
+ rm.delete(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF), timestamp);
+ rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF), timestamp);
+ rm.delete(new QueryPath(SystemTable.SCHEMA_COLUMNS_CF), timestamp);
- return mutations;
+ return rm;
}
- public static RowMutation toSchema(KsDef ksDef, long timestamp) throws IOException
+ public RowMutation toSchema(long timestamp)
{
- RowMutation mutation = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(ksDef.name));
+ RowMutation rm = new RowMutation(Table.SYSTEM_TABLE, SystemTable.getSchemaKSKey(name));
+ ColumnFamily cf = rm.addOrGet(SystemTable.SCHEMA_KEYSPACES_CF);
- for (KsDef._Fields field : KsDef._Fields.values())
- {
- if (field.equals(KsDef._Fields.CF_DEFS))
- continue;
-
- mutation.add(new QueryPath(SystemTable.SCHEMA_KEYSPACES_CF,
- null,
- AsciiType.instance.fromString(field.getFieldName())),
- valueAsBytes(ksDef.getFieldValue(field)),
- timestamp);
- }
+ cf.addColumn(Column.create(name, timestamp, "name"));
+ cf.addColumn(Column.create(durableWrites, timestamp, "durable_writes"));
+ cf.addColumn(Column.create(strategyClass.getName(), timestamp, "strategy_class"));
+ cf.addColumn(Column.create(json(strategyOptions), timestamp, "strategy_options"));
- if (!ksDef.isSetCf_defs())
- return mutation;
-
- for (CfDef cf : ksDef.cf_defs)
- {
- try
- {
- CFMetaData.toSchema(mutation, cf, timestamp);
- }
- catch (ConfigurationException e)
- {
- throw new IOException(e);
- }
- }
-
- return mutation;
- }
+ for (CFMetaData cfm : cfMetaData.values())
+ cfm.toSchema(rm, timestamp);
- public RowMutation toSchema(long timestamp) throws IOException
- {
- return toSchema(toThrift(), timestamp);
+ return rm;
}
/**
* Deserialize only Keyspace attributes without nested ColumnFamilies
*
- * @param serializedKsDef Keyspace attributes in serialized form
+ * @param row Keyspace attributes in serialized form
*
* @return deserialized keyspace without cf_defs
*
* @throws IOException if deserialization failed
*/
- public static KsDef fromSchema(ColumnFamily serializedKsDef) throws IOException
+ public static KSMetaData fromSchema(Row row, Iterable<CFMetaData> cfms) throws IOException
{
- KsDef ksDef = new KsDef();
-
- AbstractType comparator = serializedKsDef.getComparator();
-
- for (IColumn ksAttr : serializedKsDef.getSortedColumns())
+ UntypedResultSet.Row result = QueryProcessor.resultify("SELECT * FROM system.schema_keyspaces", row).one();
+ try
{
- if (ksAttr == null || ksAttr.isMarkedForDelete())
- continue;
-
- KsDef._Fields field = KsDef._Fields.findByName(comparator.getString(ksAttr.name()));
-
- // this means that given field was deprecated
- // but still exists in the serialized schema
- if (field == null)
- continue;
-
- ksDef.setFieldValue(field, deserializeValue(ksAttr.value(), getValueClass(KsDef.class, field.getFieldName())));
+ return new KSMetaData(result.getString("name"),
+ AbstractReplicationStrategy.getClass(result.getString("strategy_class")),
+ fromJsonMap(result.getString("strategy_options")),
+ result.getBoolean("durable_writes"),
+ cfms);
+ }
+ catch (ConfigurationException e)
+ {
+ throw new RuntimeException(e);
}
-
- return ksDef.name == null ? null : ksDef;
}
/**
* Deserialize Keyspace with nested ColumnFamilies
*
- * @param serializedKsDef Keyspace in serialized form
+ * @param serializedKs Keyspace in serialized form
* @param serializedCFs Collection of the serialized ColumnFamilies
*
* @return deserialized keyspace with cf_defs
*
* @throws IOException if deserialization failed
*/
- public static KSMetaData fromSchema(ColumnFamily serializedKsDef, ColumnFamily serializedCFs) throws IOException
+ public static KSMetaData fromSchema(Row serializedKs, Row serializedCFs) throws IOException
{
- KsDef ksDef = fromSchema(serializedKsDef);
-
- assert ksDef != null;
-
- Map<String, CfDef> cfs = deserializeColumnFamilies(serializedCFs);
-
- try
- {
- CFMetaData[] cfms = new CFMetaData[cfs.size()];
-
- int index = 0;
- for (CfDef cfDef : cfs.values())
- cfms[index++] = CFMetaData.fromThrift(cfDef);
-
- return fromThrift(ksDef, cfms);
- }
- catch (Exception e)
- {
- // this is critical because indicates that something is wrong with serialized schema
- throw new IOException(e);
- }
+ Map<String, CFMetaData> cfs = deserializeColumnFamilies(serializedCFs);
+ return fromSchema(serializedKs, cfs.values());
}
/**
* Deserialize ColumnFamilies from low-level schema representation, all of them belong to the same keyspace
*
- * @param serializedColumnFamilies ColumnFamilies in the serialized form
- *
+ * @param row
* @return map containing name of the ColumnFamily and it's metadata for faster lookup
*/
- public static Map<String, CfDef> deserializeColumnFamilies(ColumnFamily serializedColumnFamilies)
+ public static Map<String, CFMetaData> deserializeColumnFamilies(Row row)
{
- Map<String, CfDef> cfs = new HashMap<String, CfDef>();
-
- if (serializedColumnFamilies == null)
- return cfs;
+ if (row.cf == null)
+ return Collections.emptyMap();
- AbstractType<?> comparator = serializedColumnFamilies.getComparator();
-
- for (IColumn column : serializedColumnFamilies.getSortedColumns())
+ Map<String, CFMetaData> cfms = new HashMap<String, CFMetaData>();
+ UntypedResultSet results = QueryProcessor.resultify("SELECT * FROM system.schema_columnfamilies", row);
+ for (UntypedResultSet.Row result : results)
{
- if (column == null || column.isMarkedForDelete())
- continue;
-
- String[] attr = comparator.getString(column.name()).split(":");
- assert attr.length == 2;
-
- CfDef cfDef = cfs.get(attr[0]);
-
- if (cfDef == null)
- {
- cfDef = new CfDef();
- cfs.put(attr[0], cfDef);
- }
-
- CfDef._Fields field = CfDef._Fields.findByName(attr[1]);
-
- // this means that given field was deprecated
- // but still exists in the serialized schema
- if (field == null)
- continue;
-
- cfDef.setFieldValue(field, deserializeValue(column.value(), getValueClass(CfDef.class, field.getFieldName())));
+ CFMetaData cfm = CFMetaData.fromSchema(result);
+ cfms.put(cfm.cfName, cfm);
}
- for (CfDef cfDef : cfs.values())
+ for (CFMetaData cfm : cfms.values())
{
- for (ColumnDef columnDef : ColumnDefinition.fromSchema(ColumnDefinition.readSchema(cfDef.keyspace, cfDef.name)))
- cfDef.addToColumn_metadata(columnDef);
+ Row columnRow = ColumnDefinition.readSchema(cfm.ksName, cfm.cfName);
+ for (ColumnDefinition cd : ColumnDefinition.fromSchema(columnRow, cfm.comparator))
+ cfm.column_metadata.put(cd.name, cd);
}
- return cfs;
+ return cfms;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
index ad789c3..a5acd62 100644
--- a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
+++ b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
@@ -177,7 +177,7 @@ public class CreateColumnFamilyStatement
newCFMD.comment(cfProps.getProperty(CFPropDefs.KW_COMMENT))
.readRepairChance(getPropertyDouble(CFPropDefs.KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE))
- .dclocalReadRepairChance(getPropertyDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE))
+ .dcLocalReadRepairChance(getPropertyDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE))
.replicateOnWrite(getPropertyBoolean(CFPropDefs.KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE))
.gcGraceSeconds(getPropertyInt(CFPropDefs.KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
.defaultValidator(cfProps.getValidator())
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql/DropIndexStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql/DropIndexStatement.java b/src/java/org/apache/cassandra/cql/DropIndexStatement.java
index 4d7aeee..6d1bb18 100644
--- a/src/java/org/apache/cassandra/cql/DropIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql/DropIndexStatement.java
@@ -54,7 +54,7 @@ public class DropIndexStatement
if (cfDef == null)
throw new InvalidRequestException("Index '" + index + "' could not be found in any of the ColumnFamilies of keyspace '" + keyspace + "'");
- return new UpdateColumnFamily(cfDef);
+ return new UpdateColumnFamily(CFMetaData.fromThrift(cfDef));
}
private CfDef getUpdatedCFDef(CfDef cfDef) throws InvalidRequestException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql/QueryProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java
index 78be1d3..c824561 100644
--- a/src/java/org/apache/cassandra/cql/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java
@@ -790,7 +790,7 @@ public class QueryProcessor
ThriftValidation.validateCfDef(cf_def, oldCfm);
try
{
- applyMigrationOnStage(new UpdateColumnFamily(cf_def));
+ applyMigrationOnStage(new UpdateColumnFamily(CFMetaData.fromThrift(cf_def)));
}
catch (ConfigurationException e)
{
@@ -875,7 +875,7 @@ public class QueryProcessor
try
{
- applyMigrationOnStage(new UpdateColumnFamily(alterTable.getCfDef(keyspace)));
+ applyMigrationOnStage(new UpdateColumnFamily(CFMetaData.fromThrift(alterTable.getCfDef(keyspace))));
}
catch (ConfigurationException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 b054a5b..f643ae8 100644
--- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
@@ -121,6 +121,35 @@ public class QueryProcessor
return processStatement(getStatement(queryString, clientState).statement, clientState, Collections.<ByteBuffer>emptyList());
}
+ public static UntypedResultSet resultify(String queryString, Row row)
+ {
+ SelectStatement ss;
+ try
+ {
+ ss = (SelectStatement) getStatement(queryString, null).statement;
+ }
+ catch (InvalidRequestException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (RecognitionException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ List<CqlRow> cqlRows;
+ try
+ {
+ cqlRows = ss.process(Collections.singletonList(row));
+ }
+ catch (InvalidRequestException e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ return new UntypedResultSet(cqlRows);
+ }
+
public static CqlPreparedResult prepare(String queryString, ClientState clientState)
throws RecognitionException, InvalidRequestException
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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
new file mode 100644
index 0000000..15bf7cd
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -0,0 +1,110 @@
+/*
+ * 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.cql3;
+
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+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.thrift.Column;
+import org.apache.cassandra.thrift.CqlRow;
+import org.apache.hadoop.io.UTF8;
+
+/** a utility for doing internal cql-based queries */
+public class UntypedResultSet implements Iterable<UntypedResultSet.Row>
+{
+ private final List<CqlRow> cqlRows;
+
+ public UntypedResultSet(List<CqlRow> cqlRows)
+ {
+ this.cqlRows = cqlRows;
+ }
+
+ public Row one()
+ {
+ if (cqlRows.size() != 1)
+ throw new IllegalStateException("One row required, " + cqlRows.size() + " found");
+ return new Row(cqlRows.get(0));
+ }
+
+ public Iterator<Row> iterator()
+ {
+ return new AbstractIterator<Row>()
+ {
+ Iterator<CqlRow> iter = cqlRows.iterator();
+
+ protected Row computeNext()
+ {
+ if (!iter.hasNext())
+ return endOfData();
+ return new Row(iter.next());
+ }
+ };
+ }
+
+ public static class Row
+ {
+ Map<String, ByteBuffer> data = new HashMap<String, ByteBuffer>();
+
+ public Row(CqlRow cqlRow)
+ {
+ for (Column column : cqlRow.columns)
+ data.put(UTF8Type.instance.compose(column.name), column.value);
+ }
+
+ public boolean has(String column)
+ {
+ // Note that containsKey won't work because we may have null values
+ return data.get(column) != null;
+ }
+
+ public String getString(String column)
+ {
+ return UTF8Type.instance.compose(data.get(column));
+ }
+
+ public boolean getBoolean(String column)
+ {
+ return BooleanType.instance.compose(data.get(column));
+ }
+
+ public int getInt(String column)
+ {
+ return Int32Type.instance.compose(data.get(column));
+ }
+
+ public double getDouble(String column)
+ {
+ return DoubleType.instance.compose(data.get(column));
+ }
+
+ public ByteBuffer getBytes(String column)
+ {
+ return data.get(column);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
index 368eb6d..8935df7 100644
--- a/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java
@@ -136,7 +136,7 @@ public class AlterTableStatement extends SchemaAlteringStatement
applyPropertiesToCfDef(thriftDef, cfProps);
break;
}
- return new UpdateColumnFamily(thriftDef);
+ return new UpdateColumnFamily(CFMetaData.fromThrift(thriftDef));
}
catch (ConfigurationException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 1b86641..e175fed 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CreateColumnFamilyStatement.java
@@ -105,7 +105,7 @@ public class CreateColumnFamilyStatement extends SchemaAlteringStatement
newCFMD.comment(properties.get(CFPropDefs.KW_COMMENT))
.readRepairChance(properties.getDouble(CFPropDefs.KW_READREPAIRCHANCE, CFMetaData.DEFAULT_READ_REPAIR_CHANCE))
- .dclocalReadRepairChance(properties.getDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE))
+ .dcLocalReadRepairChance(properties.getDouble(CFPropDefs.KW_DCLOCALREADREPAIRCHANCE, CFMetaData.DEFAULT_DCLOCAL_READ_REPAIR_CHANCE))
.replicateOnWrite(properties.getBoolean(CFPropDefs.KW_REPLICATEONWRITE, CFMetaData.DEFAULT_REPLICATE_ON_WRITE))
.gcGraceSeconds(properties.getInt(CFPropDefs.KW_GCGRACESECONDS, CFMetaData.DEFAULT_GC_GRACE_SECONDS))
.defaultValidator(defaultValidator)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
index 4572d97..cc738c4 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CreateIndexStatement.java
@@ -90,7 +90,7 @@ public class CreateIndexStatement extends SchemaAlteringStatement
CFMetaData.addDefaultIndexNames(cf_def);
ThriftValidation.validateCfDef(cf_def, oldCfm);
- return new UpdateColumnFamily(cf_def);
+ return new UpdateColumnFamily(CFMetaData.fromThrift(cf_def));
}
catch (InvalidRequestException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
index 4e896f0..5d3721a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DropIndexStatement.java
@@ -54,7 +54,7 @@ public class DropIndexStatement extends SchemaAlteringStatement
if (cfDef == null)
throw new InvalidRequestException("Index '" + index + "' could not be found in any of the column families of keyspace '" + keyspace() + "'");
- return new UpdateColumnFamily(cfDef);
+ return new UpdateColumnFamily(CFMetaData.fromThrift(cfDef));
}
private CfDef getUpdatedCFDef(CfDef cfDef) throws InvalidRequestException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index c56bf02..8841570 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -64,7 +64,6 @@ public abstract class ModificationStatement extends CFStatement implements CQLSt
state.hasColumnFamilyAccess(keyspace(), columnFamily(), Permission.WRITE);
}
- @Override
public void validate(ClientState state) throws InvalidRequestException
{
if (timeToLive < 0)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 4825385..b95d6ba 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -155,15 +155,26 @@ public class SelectStatement implements CQLStatement
else
{
// otherwise create resultset from query results
- result.schema = new CqlMetadata(new HashMap<ByteBuffer, String>(),
- new HashMap<ByteBuffer, String>(),
- TypeParser.getShortName(cfDef.cfm.comparator),
- TypeParser.getShortName(cfDef.cfm.getDefaultValidator()));
+ result.schema = createSchema();
result.rows = process(rows, result.schema, variables);
return result;
}
}
+ public List<CqlRow> process(List<Row> rows) throws InvalidRequestException
+ {
+ assert !parameters.isCount; // not yet needed
+ return process(rows, createSchema(), Collections.<ByteBuffer>emptyList());
+ }
+
+ private CqlMetadata createSchema()
+ {
+ return new CqlMetadata(new HashMap<ByteBuffer, String>(),
+ new HashMap<ByteBuffer, String>(),
+ TypeParser.getShortName(cfDef.cfm.comparator),
+ TypeParser.getShortName(cfDef.cfm.getDefaultValidator()));
+ }
+
public String keyspace()
{
return cfDef.cfm.ksName;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 5e4eb8c..dd699f4 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -23,9 +23,7 @@ import java.util.*;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.db.CounterMutation;
-import org.apache.cassandra.db.IMutation;
-import org.apache.cassandra.db.RowMutation;
+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.LongType;
@@ -143,12 +141,6 @@ public class UpdateStatement extends ModificationStatement
/**
* Compute a row mutation for a single key
*
- *
- * @param keyspace working keyspace
- * @param key key to change
- * @param metadata information about CF
- *
- * @param clientState
* @return row mutation
*
* @throws InvalidRequestException on the wrong request
@@ -159,6 +151,7 @@ public class UpdateStatement extends ModificationStatement
// if true we need to wrap RowMutation into CounterMutation
boolean hasCounterColumn = false;
RowMutation rm = new RowMutation(cfDef.cfm.ksName, key);
+ ColumnFamily cf = rm.addOrGet(cfDef.cfm.cfName);
if (cfDef.isCompact)
{
@@ -168,7 +161,7 @@ public class UpdateStatement extends ModificationStatement
Operation value = processedColumns.get(cfDef.value.name);
if (value == null)
throw new InvalidRequestException(String.format("Missing mandatory column %s", cfDef.value));
- hasCounterColumn = addToMutation(clientState, rm, builder.build(), cfDef.value, value, variables);
+ hasCounterColumn = addToMutation(clientState, cf, builder.build(), cfDef.value, value, variables);
}
else
{
@@ -179,7 +172,7 @@ public class UpdateStatement extends ModificationStatement
continue;
ByteBuffer colName = builder.copy().add(name.name.key).build();
- hasCounterColumn |= addToMutation(clientState, rm, colName, name, value, variables);
+ hasCounterColumn |= addToMutation(clientState, cf, colName, name, value, variables);
}
}
@@ -187,7 +180,7 @@ public class UpdateStatement extends ModificationStatement
}
private boolean addToMutation(ClientState clientState,
- RowMutation rm,
+ ColumnFamily cf,
ByteBuffer colName,
CFDefinition.Name valueDef,
Operation value,
@@ -195,12 +188,12 @@ public class UpdateStatement extends ModificationStatement
{
if (value.isUnary())
{
- ByteBuffer valueBytes = value.value.getByteBuffer(valueDef.type, variables);
validateColumnName(colName);
- rm.add(new QueryPath(columnFamily(), null, colName),
- valueBytes,
- getTimestamp(clientState),
- getTimeToLive());
+ ByteBuffer valueBytes = value.value.getByteBuffer(valueDef.type, variables);
+ Column c = timeToLive > 0
+ ? new ExpiringColumn(colName, valueBytes, getTimestamp(clientState), timeToLive)
+ : new Column(colName, valueBytes, getTimestamp(clientState));
+ cf.addColumn(c);
return false;
}
else
@@ -226,7 +219,7 @@ public class UpdateStatement extends ModificationStatement
else
val = -val;
}
- rm.addCounter(new QueryPath(columnFamily(), null, colName), val);
+ cf.addCounter(new QueryPath(columnFamily(), null, colName), val);
return true;
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 f72bcab..9697ac6 100644
--- a/src/java/org/apache/cassandra/db/Column.java
+++ b/src/java/org/apache/cassandra/db/Column.java
@@ -21,14 +21,15 @@ package org.apache.cassandra.db;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.db.marshal.MarshalException;
+import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.utils.Allocator;
import org.apache.cassandra.utils.ByteBufferUtil;
@@ -285,5 +286,48 @@ public class Column implements IColumn
{
return getLocalDeletionTime() < gcBefore;
}
+
+ public static Column create(String value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), UTF8Type.instance.decompose(value), timestamp);
+ }
+
+ public static Column create(int value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), Int32Type.instance.decompose(value), timestamp);
+ }
+
+ public static Column create(boolean value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), BooleanType.instance.decompose(value), timestamp);
+ }
+
+ public static Column create(double value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), DoubleType.instance.decompose(value), timestamp);
+ }
+
+ public static Column create(ByteBuffer value, long timestamp, String... names)
+ {
+ return new Column(decomposeName(names), value, timestamp);
+ }
+
+ static ByteBuffer decomposeName(String... names)
+ {
+ assert names.length > 0;
+
+ if (names.length == 1)
+ return UTF8Type.instance.decompose(names[0]);
+
+ // not super performant. at this time, only infrequently called schema code uses this.
+ List<AbstractType<?>> types = new ArrayList<AbstractType<?>>(names.length);
+ for (int i = 0; i < names.length; i++)
+ types.add(UTF8Type.instance);
+
+ CompositeType.Builder builder = new CompositeType.Builder(CompositeType.getInstance(types));
+ for (String name : names)
+ builder.add(UTF8Type.instance.decompose(name));
+ return builder.build();
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/DefsTable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DefsTable.java b/src/java/org/apache/cassandra/db/DefsTable.java
index f656a33..05e4519 100644
--- a/src/java/org/apache/cassandra/db/DefsTable.java
+++ b/src/java/org/apache/cassandra/db/DefsTable.java
@@ -32,10 +32,7 @@ import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificRecord;
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ConfigurationException;
-import org.apache.cassandra.config.KSMetaData;
-import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.config.*;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.AsciiType;
@@ -153,16 +150,16 @@ public class DefsTable
if (row.cf == null || row.cf.isEmpty() || row.cf.isMarkedForDelete())
continue;
- keyspaces.add(KSMetaData.fromSchema(row.cf, serializedColumnFamilies(row.key)));
+ keyspaces.add(KSMetaData.fromSchema(row, serializedColumnFamilies(row.key)));
}
return keyspaces;
}
- private static ColumnFamily serializedColumnFamilies(DecoratedKey ksNameKey)
+ private static Row serializedColumnFamilies(DecoratedKey ksNameKey)
{
ColumnFamilyStore cfsStore = SystemTable.schemaCFS(SystemTable.SCHEMA_COLUMNFAMILIES_CF);
- return cfsStore.getColumnFamily(QueryFilter.getIdentityFilter(ksNameKey, new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF)));
+ return new Row(ksNameKey, cfsStore.getColumnFamily(QueryFilter.getIdentityFilter(ksNameKey, new QueryPath(SystemTable.SCHEMA_COLUMNFAMILIES_CF))));
}
/**
@@ -198,15 +195,15 @@ public class DefsTable
if (column.name().equals(DEFINITION_SCHEMA_COLUMN_NAME))
continue;
KsDef ks = deserializeAvro(schema, column.value(), new KsDef());
- keyspaces.add(KSMetaData.fromAvro(ks));
+ keyspaces.add(Avro.ksFromAvro(ks));
}
// store deserialized keyspaces into new place
dumpToStorage(keyspaces);
logger.info("Truncating deprecated system column families (migrations, schema)...");
- MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.MIGRATIONS_CF);
- MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.SCHEMA_CF);
+ MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.MIGRATIONS_CF, -1, false);
+ MigrationHelper.dropColumnFamily(Table.SYSTEM_TABLE, Migration.SCHEMA_CF, -1, false);
}
return keyspaces;
@@ -246,9 +243,9 @@ public class DefsTable
Set<String> keyspacesToDrop = mergeKeyspaces(oldKeyspaces, SystemTable.getSchema(SystemTable.SCHEMA_KEYSPACES_CF));
mergeColumnFamilies(oldColumnFamilies, SystemTable.getSchema(SystemTable.SCHEMA_COLUMNFAMILIES_CF));
- // it is save to drop a keyspace only when all nested ColumnFamilies where deleted
+ // it is safe to drop a keyspace only when all nested ColumnFamilies where deleted
for (String keyspaceToDrop : keyspacesToDrop)
- MigrationHelper.dropKeyspace(keyspaceToDrop);
+ MigrationHelper.dropKeyspace(keyspaceToDrop, -1, false);
}
private static Set<String> mergeKeyspaces(Map<DecoratedKey, ColumnFamily> old, Map<DecoratedKey, ColumnFamily> updated)
@@ -266,7 +263,10 @@ public class DefsTable
// we don't care about nested ColumnFamilies here because those are going to be processed separately
if (!ksAttrs.isEmpty())
- MigrationHelper.addKeyspace(KSMetaData.fromSchema(entry.getValue(), null));
+ {
+ KSMetaData ksm = KSMetaData.fromSchema(new Row(entry.getKey(), entry.getValue()), Collections.<CFMetaData>emptyList());
+ MigrationHelper.addKeyspace(ksm, -1, false);
+ }
}
/**
@@ -287,7 +287,8 @@ public class DefsTable
if (prevValue.isEmpty())
{
- MigrationHelper.addKeyspace(KSMetaData.fromSchema(newValue, null));
+ KSMetaData ksm = KSMetaData.fromSchema(new Row(entry.getKey(), newValue), Collections.<CFMetaData>emptyList());
+ MigrationHelper.addKeyspace(ksm, -1, false);
continue;
}
@@ -310,9 +311,14 @@ public class DefsTable
ColumnFamily newState = valueDiff.rightValue();
if (newState.isEmpty())
+ {
keyspacesToDrop.add(AsciiType.instance.getString(key.key));
+ }
else
- MigrationHelper.updateKeyspace(KSMetaData.fromSchema(newState));
+ {
+ KSMetaData ksm = KSMetaData.fromSchema(new Row(key, newState), Collections.<CFMetaData>emptyList());
+ MigrationHelper.updateKeyspace(ksm, -1, false);
+ }
}
return keyspacesToDrop;
@@ -331,10 +337,10 @@ public class DefsTable
if (!cfAttrs.isEmpty())
{
- Map<String, CfDef> cfDefs = KSMetaData.deserializeColumnFamilies(cfAttrs);
+ Map<String, CFMetaData> cfDefs = KSMetaData.deserializeColumnFamilies(new Row(entry.getKey(), cfAttrs));
- for (CfDef cfDef : cfDefs.values())
- MigrationHelper.addColumnFamily(cfDef);
+ for (CFMetaData cfDef : cfDefs.values())
+ MigrationHelper.addColumnFamily(cfDef, -1, false);
}
}
@@ -347,37 +353,38 @@ public class DefsTable
ColumnFamily prevValue = valueDiff.leftValue(); // state before external modification
ColumnFamily newValue = valueDiff.rightValue(); // updated state
+ Row newRow = new Row(keyspace, newValue);
if (prevValue.isEmpty()) // whole keyspace was deleted and now it's re-created
{
- for (CfDef cfDef : KSMetaData.deserializeColumnFamilies(newValue).values())
- MigrationHelper.addColumnFamily(cfDef);
+ for (CFMetaData cfm : KSMetaData.deserializeColumnFamilies(newRow).values())
+ MigrationHelper.addColumnFamily(cfm, -1, false);
}
else if (newValue.isEmpty()) // whole keyspace is deleted
{
- for (CfDef cfDef : KSMetaData.deserializeColumnFamilies(prevValue).values())
- MigrationHelper.dropColumnFamily(cfDef.keyspace, cfDef.name);
+ for (CFMetaData cfm : KSMetaData.deserializeColumnFamilies(new Row(keyspace, prevValue)).values())
+ MigrationHelper.dropColumnFamily(cfm.ksName, cfm.cfName, -1, false);
}
else // has modifications in the nested ColumnFamilies, need to perform nested diff to determine what was really changed
{
String ksName = AsciiType.instance.getString(keyspace.key);
- Map<String, CfDef> oldCfDefs = new HashMap<String, CfDef>();
+ Map<String, CFMetaData> oldCfDefs = new HashMap<String, CFMetaData>();
for (CFMetaData cfm : Schema.instance.getKSMetaData(ksName).cfMetaData().values())
- oldCfDefs.put(cfm.cfName, cfm.toThrift());
+ oldCfDefs.put(cfm.cfName, cfm);
- Map<String, CfDef> newCfDefs = KSMetaData.deserializeColumnFamilies(newValue);
+ Map<String, CFMetaData> newCfDefs = KSMetaData.deserializeColumnFamilies(newRow);
- MapDifference<String, CfDef> cfDefDiff = Maps.difference(oldCfDefs, newCfDefs);
+ MapDifference<String, CFMetaData> cfDefDiff = Maps.difference(oldCfDefs, newCfDefs);
- for (CfDef cfDef : cfDefDiff.entriesOnlyOnRight().values())
- MigrationHelper.addColumnFamily(cfDef);
+ for (CFMetaData cfDef : cfDefDiff.entriesOnlyOnRight().values())
+ MigrationHelper.addColumnFamily(cfDef, -1, false);
- for (CfDef cfDef : cfDefDiff.entriesOnlyOnLeft().values())
- MigrationHelper.dropColumnFamily(cfDef.keyspace, cfDef.name);
+ for (CFMetaData cfDef : cfDefDiff.entriesOnlyOnLeft().values())
+ MigrationHelper.dropColumnFamily(cfDef.ksName, cfDef.cfName, -1, false);
- for (MapDifference.ValueDifference<CfDef> cfDef : cfDefDiff.entriesDiffering().values())
- MigrationHelper.updateColumnFamily(cfDef.rightValue());
+ for (MapDifference.ValueDifference<CFMetaData> cfDef : cfDefDiff.entriesDiffering().values())
+ MigrationHelper.updateColumnFamily(cfDef.rightValue(), -1, false);
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/DeletedColumn.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/DeletedColumn.java b/src/java/org/apache/cassandra/db/DeletedColumn.java
index adeff0c..f67ffc8 100644
--- a/src/java/org/apache/cassandra/db/DeletedColumn.java
+++ b/src/java/org/apache/cassandra/db/DeletedColumn.java
@@ -85,4 +85,9 @@ public class DeletedColumn extends Column
if (getLocalDeletionTime() < 0)
throw new MarshalException("The local deletion time should not be negative");
}
+
+ public static DeletedColumn create(int localDeletionTime, long timestamp, String... names)
+ {
+ return new DeletedColumn(decomposeName(names), localDeletionTime, timestamp);
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 773877b..590cf1f 100644
--- a/src/java/org/apache/cassandra/db/RowMutation.java
+++ b/src/java/org/apache/cassandra/db/RowMutation.java
@@ -169,6 +169,21 @@ public class RowMutation implements IMutation, MessageProducer
throw new IllegalArgumentException("ColumnFamily " + columnFamily + " already has modifications in this mutation: " + prev);
}
+ /**
+ * @return the ColumnFamily in this RowMutation corresponding to @param cfName, creating an empty one if necessary.
+ */
+ public ColumnFamily addOrGet(String cfName)
+ {
+ CFMetaData cfm = Schema.instance.getCFMetaData(table_, cfName);
+ ColumnFamily cf = modifications_.get(cfm.cfId);
+ if (cf == null)
+ {
+ cf = ColumnFamily.create(cfm);
+ modifications_.put(cfm.cfId, cf);
+ }
+ return cf;
+ }
+
public boolean isEmpty()
{
return modifications_.isEmpty();
@@ -186,6 +201,10 @@ public class RowMutation implements IMutation, MessageProducer
* param @ value - value associated with the column
* param @ timestamp - timestamp associated with this data.
* param @ timeToLive - ttl for the column, 0 for standard (non expiring) columns
+ *
+ * @Deprecated this tends to be low-performance; we're doing two hash lookups,
+ * one of which instantiates a Pair, and callers tend to instantiate new QP objects
+ * for each call as well. Use the add(ColumnFamily) overload instead.
*/
public void add(QueryPath path, ByteBuffer value, long timestamp, int timeToLive)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/marshal/AbstractType.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
index cab8d7f..08b1777 100644
--- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java
+++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java
@@ -101,12 +101,8 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
/** get a string representation of the bytes suitable for log messages */
public abstract String getString(ByteBuffer bytes);
- /** get a byte representation of the given string.
- * defaults to unsupportedoperation so people deploying custom Types can update at their leisure. */
- public ByteBuffer fromString(String source) throws MarshalException
- {
- throw new UnsupportedOperationException();
- }
+ /** get a byte representation of the given string. */
+ public abstract ByteBuffer fromString(String source) throws MarshalException;
/* validate that the byte array is a valid sequence for the type we are supposed to be comparing */
public abstract void validate(ByteBuffer bytes) throws MarshalException;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/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 154aaf8..0f0127a 100644
--- a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
+++ b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
@@ -329,6 +329,11 @@ public class DynamicCompositeType extends AbstractCompositeType
throw new UnsupportedOperationException();
}
+ public ByteBuffer fromString(String str)
+ {
+ throw new UnsupportedOperationException();
+ }
+
public void validate(ByteBuffer bytes)
{
throw new UnsupportedOperationException();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
index d3fd777..c54e4ab 100644
--- a/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/migration/AddColumnFamily.java
@@ -46,9 +46,9 @@ public class AddColumnFamily extends Migration
this.cfm = cfm;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- return MigrationHelper.addColumnFamily(cfm, timestamp);
+ return MigrationHelper.addColumnFamily(cfm, timestamp, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/AddKeyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/AddKeyspace.java b/src/java/org/apache/cassandra/db/migration/AddKeyspace.java
index 432ebed..36c4de9 100644
--- a/src/java/org/apache/cassandra/db/migration/AddKeyspace.java
+++ b/src/java/org/apache/cassandra/db/migration/AddKeyspace.java
@@ -46,9 +46,9 @@ public class AddKeyspace extends Migration
this.ksm = ksm;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- return MigrationHelper.addKeyspace(ksm, timestamp);
+ return MigrationHelper.addKeyspace(ksm, timestamp, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
index 727674f..0628824 100644
--- a/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/migration/DropColumnFamily.java
@@ -44,9 +44,9 @@ public class DropColumnFamily extends Migration
this.cfName = cfName;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp);
+ return MigrationHelper.dropColumnFamily(ksName, cfName, timestamp, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
index 18c7f5e..d6e46a4 100644
--- a/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
+++ b/src/java/org/apache/cassandra/db/migration/DropKeyspace.java
@@ -41,9 +41,9 @@ public class DropKeyspace extends Migration
this.name = name;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- return MigrationHelper.dropKeyspace(name, timestamp);
+ return MigrationHelper.dropKeyspace(name, timestamp, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/Migration.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/Migration.java b/src/java/org/apache/cassandra/db/migration/Migration.java
index 4aaad5b..9b239bb 100644
--- a/src/java/org/apache/cassandra/db/migration/Migration.java
+++ b/src/java/org/apache/cassandra/db/migration/Migration.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.db.migration;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
+import java.util.Collections;
import java.util.UUID;
import org.slf4j.Logger;
@@ -64,26 +65,25 @@ public abstract class Migration
public final void apply() throws ConfigurationException, IOException
{
- Collection<RowMutation> mutations = applyImpl();
-
- assert !mutations.isEmpty();
+ RowMutation mutation = applyImpl();
+ assert mutation != null;
if (!StorageService.instance.isClientMode())
MigrationHelper.flushSchemaCFs();
Schema.instance.updateVersion();
- announce(mutations);
+ announce(Collections.singletonList(mutation));
}
/**
* Class specific apply implementation where schema migration logic should be put
*
- * @return mutations to update native schema
+ * @return mutation to update native schema
*
* @throws IOException on any I/O related error.
* @throws ConfigurationException if there is object misconfiguration.
*/
- protected abstract Collection<RowMutation> applyImpl() throws ConfigurationException, IOException;
+ protected abstract RowMutation applyImpl() throws ConfigurationException, IOException;
/**
* Send schema update (in form of row mutations) to alive nodes in the cluster.
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/MigrationHelper.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/MigrationHelper.java b/src/java/org/apache/cassandra/db/migration/MigrationHelper.java
index c8d94da..cbda1bf 100644
--- a/src/java/org/apache/cassandra/db/migration/MigrationHelper.java
+++ b/src/java/org/apache/cassandra/db/migration/MigrationHelper.java
@@ -17,6 +17,8 @@
*/
package org.apache.cassandra.db.migration;
+import java.io.DataOutput;
+import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
@@ -32,19 +34,13 @@ import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.RowMutation;
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.UTF8Type;
import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.thrift.CfDef;
-import org.apache.cassandra.thrift.InvalidRequestException;
-import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
-import org.codehaus.jackson.map.ObjectMapper;
-
public class MigrationHelper
{
- private static final ObjectMapper jsonMapper = new ObjectMapper();
private static final Map<Class<?>, Class<?>> primitiveToWrapper = new HashMap<Class<?>, Class<?>>();
static
{
@@ -58,119 +54,17 @@ public class MigrationHelper
primitiveToWrapper.put(double.class, Double.class);
}
- public static ByteBuffer readableColumnName(ByteBuffer columnName, AbstractType comparator)
- {
- return ByteBufferUtil.bytes(comparator.getString(columnName));
- }
-
- public static ByteBuffer valueAsBytes(Object value)
- {
- try
- {
- return ByteBuffer.wrap(jsonMapper.writeValueAsBytes(value));
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- public static Object deserializeValue(ByteBuffer value, Class<?> valueClass)
- {
- try
- {
- // because jackson serialized ByteBuffer as byte[] and needs help with deserialization later
- if (valueClass.equals(ByteBuffer.class))
- {
- byte[] bvalue = (byte[]) deserializeValue(value, byte[].class);
- return bvalue == null ? null : ByteBuffer.wrap(bvalue);
- }
-
- return jsonMapper.readValue(ByteBufferUtil.getArray(value), valueClass);
- }
- catch (Exception e)
- {
- throw new RuntimeException(e);
- }
- }
-
- public static Class<?> getValueClass(Class<?> klass, String name)
- {
- try
- {
- // We want to keep null values, so we must not return a primitive type
- return maybeConvertToWrapperClass(klass.getField(name).getType());
- }
- catch (NoSuchFieldException e)
- {
- throw new RuntimeException(e); // never happens
- }
- }
-
- private static Class<?> maybeConvertToWrapperClass(Class<?> klass)
+ public static ByteBuffer searchComposite(String name, boolean start)
{
- Class<?> cl = primitiveToWrapper.get(klass);
- return cl == null ? klass : cl;
- }
-
- public static ByteBuffer searchComposite(String comp1, boolean start)
- {
- return compositeNameFor(comp1, !start, null, false, null, false);
- }
-
- public static ByteBuffer compositeNameFor(String comp1, String comp2)
- {
- return compositeNameFor(comp1, ByteBufferUtil.bytes(comp2), null);
- }
-
- public static ByteBuffer compositeNameFor(String comp1, ByteBuffer comp2, String comp3)
- {
- return compositeNameFor(comp1, false, comp2, false, comp3, false);
- }
-
- public static ByteBuffer compositeNameFor(String comp1, boolean limit1, ByteBuffer comp2, boolean limit2, String comp3, boolean limit3)
- {
- int totalSize = 0;
-
- if (comp1 != null)
- totalSize += 2 + comp1.length() + 1;
-
- if (comp2 != null)
- totalSize += 2 + comp2.remaining() + 1;
-
- if (comp3 != null)
- totalSize += 2 + comp3.length() + 1;
-
- ByteBuffer bytes = ByteBuffer.allocate(totalSize);
-
- if (comp1 != null)
- {
- bytes.putShort((short) comp1.length());
- bytes.put(comp1.getBytes());
- bytes.put((byte) (limit1 ? 1 : 0));
- }
-
- if (comp2 != null)
- {
- int pos = comp2.position(), limit = comp2.limit();
-
- bytes.putShort((short) comp2.remaining());
- bytes.put(comp2);
- bytes.put((byte) (limit2 ? 1 : 0));
- // restore original range
- comp2.position(pos).limit(limit);
- }
-
- if (comp3 != null)
- {
- bytes.putShort((short) comp3.length());
- bytes.put(comp3.getBytes());
- bytes.put((byte) (limit3 ? 1 : 0));
- }
-
- bytes.rewind();
-
- return bytes;
+ assert name != null;
+ ByteBuffer nameBytes = UTF8Type.instance.decompose(name);
+ int length = nameBytes.remaining();
+ byte[] bytes = new byte[2 + length + 1];
+ bytes[0] = (byte)((length >> 8) & 0xFF);
+ bytes[1] = (byte)(length & 0xFF);
+ ByteBufferUtil.arrayCopy(nameBytes, 0, bytes, 2, length);
+ bytes[bytes.length - 1] = (byte)(start ? 0 : 1);
+ return ByteBuffer.wrap(bytes);
}
public static void flushSchemaCFs()
@@ -188,93 +82,27 @@ public class MigrationHelper
FBUtilities.waitOnFuture(flush);
}
- /* Schema Mutation Helpers */
-
- public static Collection<RowMutation> addKeyspace(KSMetaData ksm, long timestamp) throws ConfigurationException, IOException
- {
- return addKeyspace(ksm, timestamp, true);
- }
-
- public static void addKeyspace(KSMetaData ksDef) throws ConfigurationException, IOException
- {
- addKeyspace(ksDef, -1, false);
- }
-
- public static Collection<RowMutation> addColumnFamily(CFMetaData cfm, long timestamp) throws ConfigurationException, IOException
- {
- return addColumnFamily(cfm, timestamp, true);
- }
-
- public static void addColumnFamily(CfDef cfDef) throws ConfigurationException, IOException
- {
- try
- {
- addColumnFamily(CFMetaData.fromThrift(cfDef), -1, false);
- }
- catch (InvalidRequestException e)
- {
- throw new ConfigurationException(e.getMessage(), e);
- }
- }
-
- public static void updateKeyspace(KsDef newState) throws ConfigurationException, IOException
- {
- updateKeyspace(newState, -1, false);
- }
-
- public static Collection<RowMutation> updateKeyspace(KsDef newState, long timestamp) throws ConfigurationException, IOException
- {
- return updateKeyspace(newState, timestamp, true);
- }
-
- public static void updateColumnFamily(CfDef newState) throws ConfigurationException, IOException
- {
- updateColumnFamily(newState, -1, false);
- }
-
- public static Collection<RowMutation> updateColumnFamily(CfDef newState, long timestamp) throws ConfigurationException, IOException
- {
- return updateColumnFamily(newState, timestamp, true);
- }
-
- public static void dropColumnFamily(String ksName, String cfName) throws IOException
- {
- dropColumnFamily(ksName, cfName, -1, false);
- }
-
- public static Collection<RowMutation> dropColumnFamily(String ksName, String cfName, long timestamp) throws IOException
- {
- return dropColumnFamily(ksName, cfName, timestamp, true);
- }
-
- public static void dropKeyspace(String ksName) throws IOException
- {
- dropKeyspace(ksName, -1, false);
- }
-
- public static Collection<RowMutation> dropKeyspace(String ksName, long timestamp) throws IOException
- {
- return dropKeyspace(ksName, timestamp, true);
- }
-
/* Migration Helper implementations */
- private static Collection<RowMutation> addKeyspace(KSMetaData ksm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
+ public static RowMutation addKeyspace(KSMetaData ksm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
{
- RowMutation keyspaceDef = ksm.toSchema(timestamp);
+ RowMutation mutation = null;
if (withSchemaRecord)
- keyspaceDef.apply();
+ {
+ mutation = ksm.toSchema(timestamp);
+ mutation.apply();
+ }
Schema.instance.load(ksm);
if (!StorageService.instance.isClientMode())
Table.open(ksm.name);
- return toCollection(keyspaceDef);
+ return mutation;
}
- private static Collection<RowMutation> addColumnFamily(CFMetaData cfm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
+ public static RowMutation addColumnFamily(CFMetaData cfm, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
{
KSMetaData ksm = Schema.instance.getTableDefinition(cfm.ksName);
ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(), Collections.singleton(cfm)));
@@ -298,19 +126,19 @@ public class MigrationHelper
if (!StorageService.instance.isClientMode())
Table.open(ksm.name).initCf(cfm.cfId, cfm.cfName);
- return toCollection(mutation);
+ return mutation;
}
- private static Collection<RowMutation> updateKeyspace(KsDef newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
+ public static RowMutation updateKeyspace(KSMetaData newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
{
KSMetaData oldKsm = Schema.instance.getKSMetaData(newState.name);
- RowMutation schemaUpdate = null;
+ RowMutation mutation = null;
if (withSchemaRecord)
{
- schemaUpdate = oldKsm.diff(newState, timestamp);
- schemaUpdate.apply();
+ mutation = oldKsm.diff(newState, timestamp);
+ mutation.apply();
}
KSMetaData newKsm = KSMetaData.cloneWith(oldKsm.reloadAttributes(), oldKsm.cfMetaData().values());
@@ -320,19 +148,19 @@ public class MigrationHelper
if (!StorageService.instance.isClientMode())
Table.open(newState.name).createReplicationStrategy(newKsm);
- return toCollection(schemaUpdate);
+ return mutation;
}
- private static Collection<RowMutation> updateColumnFamily(CfDef newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
+ public static RowMutation updateColumnFamily(CFMetaData newState, long timestamp, boolean withSchemaRecord) throws ConfigurationException, IOException
{
- CFMetaData cfm = Schema.instance.getCFMetaData(newState.keyspace, newState.name);
+ CFMetaData cfm = Schema.instance.getCFMetaData(newState.ksName, newState.cfName);
- RowMutation schemaUpdate = null;
+ RowMutation mutation = null;
if (withSchemaRecord)
{
- schemaUpdate = cfm.diff(newState, timestamp);
- schemaUpdate.apply();
+ mutation = cfm.diff(newState, timestamp);
+ mutation.apply();
}
cfm.reload();
@@ -343,10 +171,10 @@ public class MigrationHelper
table.getColumnFamilyStore(cfm.cfName).reload();
}
- return toCollection(schemaUpdate);
+ return mutation;
}
- private static Collection<RowMutation> dropKeyspace(String ksName, long timestamp, boolean withSchemaRecord) throws IOException
+ public static RowMutation dropKeyspace(String ksName, long timestamp, boolean withSchemaRecord) throws IOException
{
KSMetaData ksm = Schema.instance.getTableDefinition(ksName);
String snapshotName = Table.getTimestampedSnapshotName(ksName);
@@ -365,23 +193,22 @@ public class MigrationHelper
}
}
- Collection<RowMutation> mutations = Collections.emptyList();
+ RowMutation mutation = null;
if (withSchemaRecord)
{
- mutations = ksm.dropFromSchema(timestamp);
- for (RowMutation m : mutations)
- m.apply();
+ mutation = ksm.dropFromSchema(timestamp);
+ mutation.apply();
}
// remove the table from the static instances.
Table.clear(ksm.name);
Schema.instance.clearTableDefinition(ksm);
- return mutations;
+ return mutation;
}
- private static Collection<RowMutation> dropColumnFamily(String ksName, String cfName, long timestamp, boolean withSchemaRecord) throws IOException
+ public static RowMutation dropColumnFamily(String ksName, String cfName, long timestamp, boolean withSchemaRecord) throws IOException
{
KSMetaData ksm = Schema.instance.getTableDefinition(ksName);
ColumnFamilyStore cfs = Table.open(ksName).getColumnFamilyStore(cfName);
@@ -406,7 +233,7 @@ public class MigrationHelper
Table.open(ksm.name).dropCf(cfm.cfId);
}
- return toCollection(mutation);
+ return mutation;
}
private static KSMetaData makeNewKeyspaceDefinition(KSMetaData ksm, CFMetaData toExclude)
@@ -417,9 +244,4 @@ public class MigrationHelper
assert newCfs.size() == ksm.cfMetaData().size() - 1;
return KSMetaData.cloneWith(ksm, newCfs);
}
-
- private static Collection<RowMutation> toCollection(RowMutation mutation)
- {
- return mutation == null ? Collections.<RowMutation>emptyList() : Collections.singleton(mutation);
- }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
index 7852e33..40d56e2 100644
--- a/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/migration/UpdateColumnFamily.java
@@ -20,28 +20,28 @@ package org.apache.cassandra.db.migration;
import java.io.IOException;
import java.util.Collection;
+import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.RowMutation;
-import org.apache.cassandra.thrift.CfDef;
public class UpdateColumnFamily extends Migration
{
- private final CfDef newState;
+ private final CFMetaData newState;
- public UpdateColumnFamily(CfDef newState) throws ConfigurationException
+ public UpdateColumnFamily(CFMetaData newState) throws ConfigurationException
{
super(System.nanoTime());
- if (Schema.instance.getCFMetaData(newState.keyspace, newState.name) == null)
- throw new ConfigurationException(String.format("(ks=%s, cf=%s) cannot be updated because it doesn't exist.", newState.keyspace, newState.name));
+ if (Schema.instance.getCFMetaData(newState.ksName, newState.cfName) == null)
+ throw new ConfigurationException(String.format("(ks=%s, cf=%s) cannot be updated because it doesn't exist.", newState.ksName, newState.cfName));
this.newState = newState;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- return MigrationHelper.updateColumnFamily(newState, timestamp);
+ return MigrationHelper.updateColumnFamily(newState, timestamp, true);
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java b/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
index c529bb9..267ec1e 100644
--- a/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
+++ b/src/java/org/apache/cassandra/db/migration/UpdateKeyspace.java
@@ -21,19 +21,20 @@ import java.io.IOException;
import java.util.Collection;
import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.thrift.KsDef;
public class UpdateKeyspace extends Migration
{
- private final KsDef newState;
+ private final KSMetaData newState;
- public UpdateKeyspace(KsDef newState) throws ConfigurationException
+ public UpdateKeyspace(KSMetaData newState) throws ConfigurationException
{
super(System.nanoTime());
- if (newState.isSetCf_defs() && newState.getCf_defs().size() > 0)
+ if (!newState.cfMetaData().isEmpty())
throw new ConfigurationException("Updated keyspace must not contain any column families.");
if (Schema.instance.getKSMetaData(newState.name) == null)
@@ -42,11 +43,11 @@ public class UpdateKeyspace extends Migration
this.newState = newState;
}
- protected Collection<RowMutation> applyImpl() throws ConfigurationException, IOException
+ protected RowMutation applyImpl() throws ConfigurationException, IOException
{
- Collection<RowMutation> mutations = MigrationHelper.updateKeyspace(newState, timestamp);
+ RowMutation mutation = MigrationHelper.updateKeyspace(newState, timestamp, true);
logger.info("Keyspace updated. Please perform any manual operations.");
- return mutations;
+ return mutation;
}
@Override
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/io/compress/CompressionParameters.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java
index 8ebe8aa..d152eda 100644
--- a/src/java/org/apache/cassandra/io/compress/CompressionParameters.java
+++ b/src/java/org/apache/cassandra/io/compress/CompressionParameters.java
@@ -196,21 +196,6 @@ public class CompressionParameters
throw new ConfigurationException("crc_check_chance should be between 0.0 to 1.0");
}
- public Map<CharSequence, CharSequence> asAvroOptions()
- {
- Map<CharSequence, CharSequence> options = new HashMap<CharSequence, CharSequence>();
- for (Map.Entry<String, String> entry : otherOptions.entrySet())
- options.put(new Utf8(entry.getKey()), new Utf8(entry.getValue()));
-
- if (sstableCompressor == null)
- return options;
-
- options.put(new Utf8(SSTABLE_COMPRESSION), new Utf8(sstableCompressor.getClass().getName()));
- if (chunkLength != null)
- options.put(new Utf8(CHUNK_LENGTH_KB), new Utf8(chunkLengthInKB()));
- return options;
- }
-
public Map<String, String> asThriftOptions()
{
Map<String, String> options = new HashMap<String, String>(otherOptions);
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/service/MigrationManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java
index 6990d84..467b5f9 100644
--- a/src/java/org/apache/cassandra/service/MigrationManager.java
+++ b/src/java/org/apache/cassandra/service/MigrationManager.java
@@ -85,7 +85,7 @@ public class MigrationManager implements IEndpointStateChangeSubscriber
public void onRemove(InetAddress endpoint)
{}
- public static void rectifySchema(UUID theirVersion, final InetAddress endpoint)
+ private static void rectifySchema(UUID theirVersion, final InetAddress endpoint)
{
// Can't request migrations from nodes with versions younger than 1.1
if (Gossiper.instance.getVersion(endpoint) < MessagingService.VERSION_11)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/thrift/CassandraServer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java
index 952a103..3c53ac1 100644
--- a/src/java/org/apache/cassandra/thrift/CassandraServer.java
+++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java
@@ -1050,7 +1050,7 @@ public class CassandraServer implements Cassandra.Iface
try
{
ThriftValidation.validateKsDef(ks_def);
- applyMigrationOnStage(new UpdateKeyspace(ks_def));
+ applyMigrationOnStage(new UpdateKeyspace(KSMetaData.fromThrift(ks_def)));
return Schema.instance.getVersion().toString();
}
catch (ConfigurationException e)
@@ -1079,7 +1079,7 @@ public class CassandraServer implements Cassandra.Iface
{
// ideally, apply() would happen on the stage with the
CFMetaData.applyImplicitDefaults(cf_def);
- UpdateColumnFamily update = new UpdateColumnFamily(cf_def);
+ UpdateColumnFamily update = new UpdateColumnFamily(CFMetaData.fromThrift(cf_def));
applyMigrationOnStage(update);
return Schema.instance.getVersion().toString();
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
index 34a58e0..e52c158 100644
--- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
+++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
@@ -18,10 +18,7 @@
*/
package org.apache.cassandra.utils;
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.io.InputStream;
+import java.io.*;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/src/java/org/apache/cassandra/utils/FBUtilities.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 4f13c80..38df00e 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -54,11 +54,16 @@ import org.apache.thrift.TBase;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
+import org.codehaus.jackson.JsonFactory;
+import org.codehaus.jackson.map.ObjectMapper;
+import org.codehaus.jackson.type.TypeReference;
public class FBUtilities
{
private static Logger logger_ = LoggerFactory.getLogger(FBUtilities.class);
+ private static ObjectMapper jsonMapper = new ObjectMapper(new JsonFactory());
+
public static final BigInteger TWO = new BigInteger("2");
private static volatile InetAddress localInetAddress_;
@@ -508,6 +513,42 @@ public class FBUtilities
return new WrappedCloseableIterator<T>(iterator);
}
+ public static Map<String, String> fromJsonMap(String json)
+ {
+ try
+ {
+ return jsonMapper.readValue(json, Map.class);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static List<String> fromJsonList(String json)
+ {
+ try
+ {
+ return jsonMapper.readValue(json, List.class);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static String json(Object object)
+ {
+ try
+ {
+ return jsonMapper.writeValueAsString(object);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
private static final class WrappedCloseableIterator<T>
extends AbstractIterator<T> implements CloseableIterator<T>
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/ccb00289/test/unit/org/apache/cassandra/config/CFMetaDataTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java
index 76d1c4b..be9c0e1 100644
--- a/test/unit/org/apache/cassandra/config/CFMetaDataTest.java
+++ b/test/unit/org/apache/cassandra/config/CFMetaDataTest.java
@@ -24,14 +24,19 @@ import java.util.HashMap;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.config.Schema;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Row;
import org.apache.cassandra.db.RowMutation;
import org.apache.cassandra.db.SystemTable;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.io.compress.*;
+import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.IndexType;
@@ -41,6 +46,9 @@ import org.junit.Test;
import static org.junit.Assert.assertEquals;
+import java.util.Map;
+import java.nio.ByteBuffer;
+
public class CFMetaDataTest extends CleanupHelper
{
private static String KEYSPACE = "Keyspace1";
@@ -117,6 +125,8 @@ public class CFMetaDataTest extends CleanupHelper
private void checkInverses(CFMetaData cfm) throws Exception
{
+ DecoratedKey k = StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(cfm.ksName));
+
// Test thrift conversion
assert cfm.equals(CFMetaData.fromThrift(cfm.toThrift())) : String.format("\n%s\n!=\n%s", cfm, CFMetaData.fromThrift(cfm.toThrift()));
@@ -124,7 +134,8 @@ public class CFMetaDataTest extends CleanupHelper
RowMutation rm = cfm.toSchema(System.currentTimeMillis());
ColumnFamily serializedCf = rm.getColumnFamily(Schema.instance.getId(Table.SYSTEM_TABLE, SystemTable.SCHEMA_COLUMNFAMILIES_CF));
ColumnFamily serializedCD = rm.getColumnFamily(Schema.instance.getId(Table.SYSTEM_TABLE, SystemTable.SCHEMA_COLUMNS_CF));
- CfDef cfDef = CFMetaData.addColumnDefinitionSchema(CFMetaData.fromSchema(serializedCf), serializedCD);
- assert cfm.equals(CFMetaData.fromThrift(cfDef)) : String.format("\n%s\n!=\n%s", cfm, CFMetaData.fromThrift(cfDef));
+ UntypedResultSet.Row result = QueryProcessor.resultify("SELECT * FROM system.schema_columnfamilies", new Row(k, serializedCf)).one();
+ CFMetaData newCfm = CFMetaData.addColumnDefinitionSchema(CFMetaData.fromSchemaNoColumns(result), new Row(k, serializedCD));
+ assert cfm.equals(newCfm) : String.format("\n%s\n!=\n%s", cfm, newCfm);
}
}