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

merge from 0.8 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1127510 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
T Jake Luciani 2011-05-25 13:06:39 +00:00
parent a2665b174c
commit ea6ec42914
11 changed files with 249 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -22,6 +22,7 @@
* add placeholders for missing rows in range query pseudo-RR (CASSANDRA-2680)
* remove no-op HHOM.renameHints (CASSANDRA-2693)
* clone super columns to avoid modifying them during flush (CASSANDRA-2675)
* allow writes to bypass the commitlog for certain keyspaces (CASSANDRA-2683)
0.8.0-final

1
interface/cassandra.thrift
View File

@ -407,6 +407,7 @@ struct KsDef {
4: optional i32 replication_factor,
5: required list<CfDef> cf_defs,
6: optional bool durable_writes=1,
}
/** CQL query compression */

96
interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java
View File

@ -50,6 +50,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
private static final org.apache.thrift.protocol.TField STRATEGY_OPTIONS_FIELD_DESC = new org.apache.thrift.protocol.TField("strategy_options", org.apache.thrift.protocol.TType.MAP, (short)3);
private static final org.apache.thrift.protocol.TField REPLICATION_FACTOR_FIELD_DESC = new org.apache.thrift.protocol.TField("replication_factor", org.apache.thrift.protocol.TType.I32, (short)4);
private static final org.apache.thrift.protocol.TField CF_DEFS_FIELD_DESC = new org.apache.thrift.protocol.TField("cf_defs", org.apache.thrift.protocol.TType.LIST, (short)5);
private static final org.apache.thrift.protocol.TField DURABLE_WRITES_FIELD_DESC = new org.apache.thrift.protocol.TField("durable_writes", org.apache.thrift.protocol.TType.BOOL, (short)6);
public String name;
public String strategy_class;
@ -59,6 +60,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
*/
public int replication_factor;
public List<CfDef> cf_defs;
public boolean durable_writes;
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@ -69,7 +71,8 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
* @deprecated
*/
REPLICATION_FACTOR((short)4, "replication_factor"),
CF_DEFS((short)5, "cf_defs");
CF_DEFS((short)5, "cf_defs"),
DURABLE_WRITES((short)6, "durable_writes");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@ -94,6 +97,8 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
return REPLICATION_FACTOR;
case 5: // CF_DEFS
return CF_DEFS;
case 6: // DURABLE_WRITES
return DURABLE_WRITES;
default:
return null;
}
@ -135,7 +140,8 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
// isset id assignments
private static final int __REPLICATION_FACTOR_ISSET_ID = 0;
private BitSet __isset_bit_vector = new BitSet(1);
private static final int __DURABLE_WRITES_ISSET_ID = 1;
private BitSet __isset_bit_vector = new BitSet(2);
public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
@ -153,11 +159,15 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
tmpMap.put(_Fields.CF_DEFS, new org.apache.thrift.meta_data.FieldMetaData("cf_defs", org.apache.thrift.TFieldRequirementType.REQUIRED,
new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST,
new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, CfDef.class))));
tmpMap.put(_Fields.DURABLE_WRITES, new org.apache.thrift.meta_data.FieldMetaData("durable_writes", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(KsDef.class, metaDataMap);
}
public KsDef() {
this.durable_writes = true;
}
public KsDef(
@ -206,6 +216,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
}
this.cf_defs = __this__cf_defs;
}
this.durable_writes = other.durable_writes;
}
public KsDef deepCopy() {
@ -220,6 +231,8 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
setReplication_factorIsSet(false);
this.replication_factor = 0;
this.cf_defs = null;
this.durable_writes = true;
}
public String getName() {
@ -373,6 +386,29 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
}
}
public boolean isDurable_writes() {
return this.durable_writes;
}
public KsDef setDurable_writes(boolean durable_writes) {
this.durable_writes = durable_writes;
setDurable_writesIsSet(true);
return this;
}
public void unsetDurable_writes() {
__isset_bit_vector.clear(__DURABLE_WRITES_ISSET_ID);
}
/** Returns true if field durable_writes is set (has been assigned a value) and false otherwise */
public boolean isSetDurable_writes() {
return __isset_bit_vector.get(__DURABLE_WRITES_ISSET_ID);
}
public void setDurable_writesIsSet(boolean value) {
__isset_bit_vector.set(__DURABLE_WRITES_ISSET_ID, value);
}
public void setFieldValue(_Fields field, Object value) {
switch (field) {
case NAME:
@ -415,6 +451,14 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
}
break;
case DURABLE_WRITES:
if (value == null) {
unsetDurable_writes();
} else {
setDurable_writes((Boolean)value);
}
break;
}
}
@ -435,6 +479,9 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
case CF_DEFS:
return getCf_defs();
case DURABLE_WRITES:
return new Boolean(isDurable_writes());
}
throw new IllegalStateException();
}
@ -456,6 +503,8 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
return isSetReplication_factor();
case CF_DEFS:
return isSetCf_defs();
case DURABLE_WRITES:
return isSetDurable_writes();
}
throw new IllegalStateException();
}
@ -518,6 +567,15 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
return false;
}
boolean this_present_durable_writes = true && this.isSetDurable_writes();
boolean that_present_durable_writes = true && that.isSetDurable_writes();
if (this_present_durable_writes || that_present_durable_writes) {
if (!(this_present_durable_writes && that_present_durable_writes))
return false;
if (this.durable_writes != that.durable_writes)
return false;
}
return true;
}
@ -550,6 +608,11 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
if (present_cf_defs)
builder.append(cf_defs);
boolean present_durable_writes = true && (isSetDurable_writes());
builder.append(present_durable_writes);
if (present_durable_writes)
builder.append(durable_writes);
return builder.toHashCode();
}
@ -611,6 +674,16 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetDurable_writes()).compareTo(typedOther.isSetDurable_writes());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetDurable_writes()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.durable_writes, typedOther.durable_writes);
if (lastComparison != 0) {
return lastComparison;
}
}
return 0;
}
@ -687,6 +760,14 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
break;
case 6: // DURABLE_WRITES
if (field.type == org.apache.thrift.protocol.TType.BOOL) {
this.durable_writes = iprot.readBool();
setDurable_writesIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
break;
default:
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
@ -744,6 +825,11 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
}
oprot.writeFieldEnd();
}
if (isSetDurable_writes()) {
oprot.writeFieldBegin(DURABLE_WRITES_FIELD_DESC);
oprot.writeBool(this.durable_writes);
oprot.writeFieldEnd();
}
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@ -792,6 +878,12 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
sb.append(this.cf_defs);
}
first = false;
if (isSetDurable_writes()) {
if (!first) sb.append(", ");
sb.append("durable_writes:");
sb.append(this.durable_writes);
first = false;
}
sb.append(")");
return sb.toString();
}

1
src/avro/internode.genavro
View File

@ -75,6 +75,7 @@ protocol InterNode {
union{ map<string>, null } strategy_options;
union{ int, null } replication_factor;
array<CfDef> cf_defs;
boolean durable_writes = true;
}
@namespace("org.apache.cassandra.utils.avro")

33
src/java/org/apache/cassandra/config/KSMetaData.java
View File

@ -37,8 +37,14 @@ public final class KSMetaData
public final Class<? extends AbstractReplicationStrategy> strategyClass;
public final Map<String, String> strategyOptions;
private final Map<String, CFMetaData> cfMetaData;
private boolean durable_writes;
public KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, CFMetaData... cfDefs)
{
this(name, strategyClass, strategyOptions, true, cfDefs);
}
public KSMetaData(String name, Class<? extends AbstractReplicationStrategy> strategyClass, Map<String, String> strategyOptions, boolean durable_writes, CFMetaData... cfDefs)
{
this.name = name;
this.strategyClass = strategyClass == null ? NetworkTopologyStrategy.class : strategyClass;
@ -47,8 +53,19 @@ public final class KSMetaData
for (CFMetaData cfm : cfDefs)
cfmap.put(cfm.cfName, cfm);
this.cfMetaData = Collections.unmodifiableMap(cfmap);
this.durable_writes = durable_writes;
}
public void setDurableWrites(boolean durable_writes)
{
this.durable_writes = durable_writes;
}
public boolean isDurableWrites()
{
return durable_writes;
}
public static Map<String, String> forwardsCompatibleOptions(KsDef ks_def)
{
Map<String, String> options = new HashMap<String, String>(ks_def.strategy_options);
@ -77,7 +94,8 @@ public final class KSMetaData
&& ObjectUtils.equals(other.strategyClass, strategyClass)
&& ObjectUtils.equals(other.strategyOptions, strategyOptions)
&& other.cfMetaData.size() == cfMetaData.size()
&& other.cfMetaData.equals(cfMetaData);
&& other.cfMetaData.equals(cfMetaData)
&& other.durable_writes == durable_writes;
}
public Map<String, CFMetaData> cfMetaData()
@ -101,6 +119,9 @@ public final class KSMetaData
ks.cf_defs = SerDeUtils.createArray(cfMetaData.size(), org.apache.cassandra.db.migration.avro.CfDef.SCHEMA$);
for (CFMetaData cfm : cfMetaData.values())
ks.cf_defs.add(cfm.deflate());
ks.durable_writes = durable_writes;
return ks;
}
@ -113,7 +134,8 @@ public final class KSMetaData
.append(strategyClass.getSimpleName())
.append("{")
.append(StringUtils.join(cfMetaData.values(), ", "))
.append("}");
.append("}")
.append("durable_writes: ").append(durable_writes);
return sb.toString();
}
@ -146,7 +168,7 @@ public final class KSMetaData
for (int i = 0; i < cfsz; i++)
cfMetaData[i] = CFMetaData.inflate(cfiter.next());
return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, cfMetaData);
return new KSMetaData(ks.name.toString(), repStratClass, strategyOptions, ks.durable_writes, cfMetaData);
}
public static String convertOldStrategyName(String name)
@ -167,6 +189,7 @@ public final class KSMetaData
return new KSMetaData(ksd.name,
AbstractReplicationStrategy.getClass(ksd.strategy_class),
forwardsCompatibleOptions(ksd),
ksd.durable_writes,
cfDefs);
}
@ -179,6 +202,8 @@ public final class KSMetaData
ksdef.setStrategy_options(ksm.strategyOptions);
if (ksm.strategyOptions != null && ksm.strategyOptions.containsKey("replication_factor"))
ksdef.setReplication_factor(Integer.parseInt(ksm.strategyOptions.get("replication_factor")));
ksdef.durable_writes = ksm.durable_writes;
return ksdef;
}
}

5
src/java/org/apache/cassandra/db/RowMutation.java
View File

@ -28,6 +28,7 @@ import org.apache.commons.lang.StringUtils;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.net.Message;
@ -204,7 +205,9 @@ public class RowMutation implements IMutation, MessageProducer
*/
public void apply() throws IOException
{
Table.open(table_).apply(this, true);
KSMetaData ksm = DatabaseDescriptor.getTableDefinition(getTable());
Table.open(table_).apply(this, ksm.isDurableWrites());
}
public void applyUnsafe() throws IOException

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

@ -69,7 +69,7 @@ public class RowMutationVerbHandler implements IVerbHandler
if (forwardBytes != null)
forwardToLocalNodes(message, forwardBytes);
Table.open(rm.getTable()).apply(rm, true);
rm.apply();
WriteResponse response = new WriteResponse(rm.getTable(), rm.key(), true);
Message responseMessage = WriteResponse.makeWriteResponseMessage(message, response);

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

@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.PrintStream;
import java.lang.management.MemoryUsage;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.text.DecimalFormat;
import java.util.*;
import java.util.Map.Entry;
@ -161,11 +162,12 @@ public class NodeCmd
Collection<String> movingNodes = probe.getMovingNodes();
Map<String, String> loadMap = probe.getLoadMap();
outs.printf("%-16s%-7s%-8s%-16s%-8s%-44s%n", "Address", "Status", "State", "Load", "Owns", "Token");
String format = "%-16s%-12s%-12s%-7s%-8s%-16s%-8s%-44s%n";
outs.printf(format, "Address", "DC", "Rack", "Status", "State", "Load", "Owns", "Token");
// show pre-wrap token twice so you can always read a node's range as
// (previous line token, current line token]
if (sortedTokens.size() > 1)
outs.printf("%-16s%-7s%-8s%-16s%-8s%-44s%n", "", "", "", "", "", sortedTokens.get(sortedTokens.size() - 1));
outs.printf(format, "", "", "", "", "", "", "", sortedTokens.get(sortedTokens.size() - 1));
// Calculate per-token ownership of the ring
Map<Token, Float> ownerships = probe.getOwnership();
@ -173,6 +175,24 @@ public class NodeCmd
for (Token token : sortedTokens)
{
String primaryEndpoint = tokenToEndpoint.get(token);
String dataCenter;
try
{
dataCenter = probe.getEndpointSnitchInfoProxy().getDatacenter(primaryEndpoint);
}
catch (UnknownHostException e)
{
dataCenter = "Unknown";
}
String rack;
try
{
rack = probe.getEndpointSnitchInfoProxy().getRack(primaryEndpoint);
}
catch (UnknownHostException e)
{
rack = "Unknown";
}
String status = liveNodes.contains(primaryEndpoint)
? "Up"
: deadNodes.contains(primaryEndpoint)
@ -192,7 +212,7 @@ public class NodeCmd
? loadMap.get(primaryEndpoint)
: "?";
String owns = new DecimalFormat("##0.00%").format(ownerships.get(token));
outs.printf("%-16s%-7s%-8s%-16s%-8s%-44s%n", primaryEndpoint, status, state, load, owns, token);
outs.printf(format, primaryEndpoint, dataCenter, rack, status, state, load, owns, token);
}
}
@ -236,6 +256,10 @@ public class NodeCmd
double memUsed = (double)heapUsage.getUsed() / (1024 * 1024);
double memMax = (double)heapUsage.getMax() / (1024 * 1024);
outs.printf("%-17s: %.2f / %.2f%n", "Heap Memory (MB)", memUsed, memMax);
// Data Center/Rack
outs.printf("%-17s: %s%n", "Data Center", probe.getDataCenter());
outs.printf("%-17s: %s%n", "Rack", probe.getRack());
}
public void printReleaseVersion(PrintStream outs)

38
src/java/org/apache/cassandra/tools/NodeProbe.java
View File

@ -25,6 +25,7 @@ import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryUsage;
import java.lang.management.RuntimeMXBean;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.Map.Entry;
@ -47,6 +48,7 @@ import org.apache.cassandra.db.ColumnFamilyStoreMBean;
import org.apache.cassandra.db.CompactionManager;
import org.apache.cassandra.db.CompactionManagerMBean;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.locator.EndpointSnitchInfoMBean;
import org.apache.cassandra.net.MessagingServiceMBean;
import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.streaming.StreamingService;
@ -485,6 +487,18 @@ public class NodeProbe
throw new RuntimeException("Error while executing truncate", e);
}
}
public EndpointSnitchInfoMBean getEndpointSnitchInfoProxy()
{
try
{
return JMX.newMBeanProxy(mbeanServerConn, new ObjectName("org.apache.cassandra.db:type=EndpointSnitchInfo"), EndpointSnitchInfoMBean.class);
}
catch (MalformedObjectNameException e)
{
throw new RuntimeException(e);
}
}
public MessagingServiceMBean getMsProxy()
{
@ -516,6 +530,30 @@ public class NodeProbe
return cfsProxy;
}
public String getDataCenter()
{
try
{
return getEndpointSnitchInfoProxy().getDatacenter(host);
}
catch (UnknownHostException e)
{
return "Unknown";
}
}
public String getRack()
{
try
{
return getEndpointSnitchInfoProxy().getRack(host);
}
catch (UnknownHostException e)
{
return "Unknown";
}
}
public List<String> getKeyspaces()
{
return ssProxy.getKeyspaces();

10
test/unit/org/apache/cassandra/SchemaLoader.java
View File

@ -67,7 +67,8 @@ public class SchemaLoader
String ks6 = "Keyspace6";
String ks_kcs = "KeyCacheSpace";
String ks_rcs = "RowCacheSpace";
String ks_nocommit = "NoCommitlogSpace";
Class<? extends AbstractReplicationStrategy> simple = SimpleStrategy.class;
Map<String, String> opts_rf1 = KSMetaData.optsWithRF(1);
@ -223,6 +224,13 @@ public class SchemaLoader
standardCFMD(ks_rcs, "CachedCF")
.rowCacheSize(100)));
schema.add(new KSMetaData(ks_nocommit,
simple,
opts_rf1,
false,
standardCFMD(ks_nocommit, "Standard1")));
return schema;
}

50
test/unit/org/apache/cassandra/db/ReadMessageTest.java
View File

@ -18,22 +18,21 @@
*/
package org.apache.cassandra.db;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.*;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.junit.Test;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -98,4 +97,43 @@ public class ReadMessageTest extends SchemaLoader
IColumn col = row.cf.getColumn(ByteBufferUtil.bytes("Column1"));
assert Arrays.equals(col.value().array(), "abcd".getBytes());
}
@Test
public void testNoCommitLog() throws Exception
{
RowMutation rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("row"));
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("commit1")), ByteBufferUtil.bytes("abcd"), 0);
rm.apply();
rm = new RowMutation("NoCommitlogSpace", ByteBufferUtil.bytes("row"));
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("commit2")), ByteBufferUtil.bytes("abcd"), 0);
rm.apply();
boolean commitLogMessageFound = false;
boolean noCommitLogMessageFound = false;
File commitLogDir = new File(DatabaseDescriptor.getCommitLogLocation());
for(String filename : commitLogDir.list())
{
BufferedReader f = new BufferedReader(new FileReader(commitLogDir.getAbsolutePath()+File.separator+filename));
String line = null;
while( (line = f.readLine()) != null)
{
if(line.contains("commit1"))
commitLogMessageFound = true;
if(line.contains("commit2"))
noCommitLogMessageFound = true;
}
f.close();
}
assertTrue(commitLogMessageFound);
assertFalse(noCommitLogMessageFound);
}
}