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

Make memtable flush thresholds per-CF instead of global. Patch by Jon Hermes, reviewed by brandonwilliams for CASSANDRA-1007 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1023036 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Brandon Williams 2010-10-15 17:42:57 +00:00
parent 463a5c5de5
commit bb0dc512d8
18 changed files with 558 additions and 167 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -39,6 +39,7 @@ dev
* PropertyFileSnitch configuration file renamed to
cassandra-topology.properties
* add cli support for get_range_slices (CASSANDRA-1088)
* Make memtable flush thresholds per-CF instead of global (CASSANDRA-1007)
0.7-beta2

35
conf/cassandra.yaml
View File

@ -163,20 +163,6 @@ snapshot_before_compaction: false
# The threshold size in megabytes the binary memtable must grow to,
# before it's submitted for flushing to disk.
binary_memtable_throughput_in_mb: 256
# The maximum time to leave a dirty memtable unflushed.
# (While any affected columnfamilies have unflushed data from a
# commit log segment, that segment cannot be deleted.)
# This needs to be large enough that it won't cause a flush storm
# of all your memtables flushing at once because none has hit
# the size or count thresholds yet.
# defaults to 60
#memtable_flush_after_mins: 60
# Size of the memtable in memory before it is flushed
# if left undefined, 1/8 of the heap will be used
#memtable_throughput_in_mb: 256
# Number of objects in millions in the memtable before it is flushed
# if left undefined, the memtable_throughput_in_mb / 64 * 0.3 will be used
#memtable_operations_in_millions: 1.2
# Add column indexes to a row after its contents reach this size.
# Increase if your column values are large, or if you have a very large
@ -362,10 +348,20 @@ index_interval: 128
# before a minor compaction is forced. decreasing this will cause
# minor compactions to start more frequently and be less intensive.
# setting this to 0 disables minor compactions. defaults to 32.
# - row_cache_save_period_in_seconds: number of seconds between saving row caches.
# The row caches can be saved periodically and if one exists on startup it will be loaded.
# - key_cache_save_period_in_seconds: number of seconds between saving key caches.
# The key caches can be saved periodically and if one exists on startup it will be loaded.
# - row_cache_save_period_in_seconds: number of seconds between saving
# row caches. The row caches can be saved periodically and if one
# exists on startup it will be loaded.
# - key_cache_save_period_in_seconds: number of seconds between saving
# key caches. The key caches can be saved periodically and if one
# exists on startup it will be loaded.
# - memtable_flush_after_mins: The maximum time to leave a dirty table
# unflushed. This should be large enough that it won't cause a flush
# storm of all memtables during periods of inactivity.
# - memtable_throughput_in_mb: The maximum size of the memtable before
# it is flushed. If undefined, 1/8 * heapsize will be used.
# - memtable_operations_in_millions: Number of operations in millions
# before the memtable is flushed. If undefined, throughput / 64 * 0.3
# will be used.
#
# NOTE: this keyspace definition is for demonstration purposes only.
# Cassandra will not load these definitions during startup. See
@ -381,6 +377,9 @@ keyspaces:
rows_cached: 1000
row_cache_save_period_in_seconds: 0
key_cache_save_period_in_seconds: 3600
memtable_flush_after_mins: 59
memtable_throughput_in_mb: 255
memtable_operations_in_millions: 0.29
- name: Standard2
compare_with: UTF8Type

5
interface/cassandra.genavro
View File

@ -151,9 +151,12 @@ protocol Cassandra {
union { null, string } default_validation_class = null;
union { null, int } min_compaction_threshold = null;
union { null, int } max_compaction_threshold = null;
union { int, null } id;
union { int, null } row_cache_save_period_in_seconds = 0;
union { int, null } key_cache_save_period_in_seconds = 3600;
union { int, null } memtable_flush_after_mins = 60;
union { null, int } memtable_throughput_in_mb = null;
union { null, double} memtable_operations_in_millions = null;
union { int, null } id;
union { array<ColumnDef>, null } column_metadata;
}

5
interface/cassandra.thrift
View File

@ -46,7 +46,7 @@ namespace rb CassandraThrift
# for every edit that doesn't result in a change to major/minor.
#
# See the Semantic Versioning Specification (SemVer) http://semver.org.
const string VERSION = "19.1.0"
const string VERSION = "19.2.0"
#
@ -344,6 +344,9 @@ struct CfDef {
18: optional i32 max_compaction_threshold,
19: optional i32 row_cache_save_period_in_seconds,
20: optional i32 key_cache_save_period_in_seconds,
21: optional i32 memtable_flush_after_mins,
22: optional i32 memtable_throughput_in_mb,
23: optional double memtable_operations_in_millions,
}
/* describes a keyspace. */

265
interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java
View File

@ -68,6 +68,9 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
private static final TField MAX_COMPACTION_THRESHOLD_FIELD_DESC = new TField("max_compaction_threshold", TType.I32, (short)18);
private static final TField ROW_CACHE_SAVE_PERIOD_IN_SECONDS_FIELD_DESC = new TField("row_cache_save_period_in_seconds", TType.I32, (short)19);
private static final TField KEY_CACHE_SAVE_PERIOD_IN_SECONDS_FIELD_DESC = new TField("key_cache_save_period_in_seconds", TType.I32, (short)20);
private static final TField MEMTABLE_FLUSH_AFTER_MINS_FIELD_DESC = new TField("memtable_flush_after_mins", TType.I32, (short)21);
private static final TField MEMTABLE_THROUGHPUT_IN_MB_FIELD_DESC = new TField("memtable_throughput_in_mb", TType.I32, (short)22);
private static final TField MEMTABLE_OPERATIONS_IN_MILLIONS_FIELD_DESC = new TField("memtable_operations_in_millions", TType.DOUBLE, (short)23);
public String keyspace;
public String name;
@ -87,6 +90,9 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
public int max_compaction_threshold;
public int row_cache_save_period_in_seconds;
public int key_cache_save_period_in_seconds;
public int memtable_flush_after_mins;
public int memtable_throughput_in_mb;
public double memtable_operations_in_millions;
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements TFieldIdEnum {
@ -107,7 +113,10 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
MIN_COMPACTION_THRESHOLD((short)17, "min_compaction_threshold"),
MAX_COMPACTION_THRESHOLD((short)18, "max_compaction_threshold"),
ROW_CACHE_SAVE_PERIOD_IN_SECONDS((short)19, "row_cache_save_period_in_seconds"),
KEY_CACHE_SAVE_PERIOD_IN_SECONDS((short)20, "key_cache_save_period_in_seconds");
KEY_CACHE_SAVE_PERIOD_IN_SECONDS((short)20, "key_cache_save_period_in_seconds"),
MEMTABLE_FLUSH_AFTER_MINS((short)21, "memtable_flush_after_mins"),
MEMTABLE_THROUGHPUT_IN_MB((short)22, "memtable_throughput_in_mb"),
MEMTABLE_OPERATIONS_IN_MILLIONS((short)23, "memtable_operations_in_millions");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@ -158,6 +167,12 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
return ROW_CACHE_SAVE_PERIOD_IN_SECONDS;
case 20: // KEY_CACHE_SAVE_PERIOD_IN_SECONDS
return KEY_CACHE_SAVE_PERIOD_IN_SECONDS;
case 21: // MEMTABLE_FLUSH_AFTER_MINS
return MEMTABLE_FLUSH_AFTER_MINS;
case 22: // MEMTABLE_THROUGHPUT_IN_MB
return MEMTABLE_THROUGHPUT_IN_MB;
case 23: // MEMTABLE_OPERATIONS_IN_MILLIONS
return MEMTABLE_OPERATIONS_IN_MILLIONS;
default:
return null;
}
@ -208,7 +223,10 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
private static final int __MAX_COMPACTION_THRESHOLD_ISSET_ID = 7;
private static final int __ROW_CACHE_SAVE_PERIOD_IN_SECONDS_ISSET_ID = 8;
private static final int __KEY_CACHE_SAVE_PERIOD_IN_SECONDS_ISSET_ID = 9;
private BitSet __isset_bit_vector = new BitSet(10);
private static final int __MEMTABLE_FLUSH_AFTER_MINS_ISSET_ID = 10;
private static final int __MEMTABLE_THROUGHPUT_IN_MB_ISSET_ID = 11;
private static final int __MEMTABLE_OPERATIONS_IN_MILLIONS_ISSET_ID = 12;
private BitSet __isset_bit_vector = new BitSet(13);
public static final Map<_Fields, FieldMetaData> metaDataMap;
static {
@ -250,6 +268,12 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
new FieldValueMetaData(TType.I32)));
tmpMap.put(_Fields.KEY_CACHE_SAVE_PERIOD_IN_SECONDS, new FieldMetaData("key_cache_save_period_in_seconds", TFieldRequirementType.OPTIONAL,
new FieldValueMetaData(TType.I32)));
tmpMap.put(_Fields.MEMTABLE_FLUSH_AFTER_MINS, new FieldMetaData("memtable_flush_after_mins", TFieldRequirementType.OPTIONAL,
new FieldValueMetaData(TType.I32)));
tmpMap.put(_Fields.MEMTABLE_THROUGHPUT_IN_MB, new FieldMetaData("memtable_throughput_in_mb", TFieldRequirementType.OPTIONAL,
new FieldValueMetaData(TType.I32)));
tmpMap.put(_Fields.MEMTABLE_OPERATIONS_IN_MILLIONS, new FieldMetaData("memtable_operations_in_millions", TFieldRequirementType.OPTIONAL,
new FieldValueMetaData(TType.DOUBLE)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
FieldMetaData.addStructMetaDataMap(CfDef.class, metaDataMap);
}
@ -322,6 +346,9 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
this.max_compaction_threshold = other.max_compaction_threshold;
this.row_cache_save_period_in_seconds = other.row_cache_save_period_in_seconds;
this.key_cache_save_period_in_seconds = other.key_cache_save_period_in_seconds;
this.memtable_flush_after_mins = other.memtable_flush_after_mins;
this.memtable_throughput_in_mb = other.memtable_throughput_in_mb;
this.memtable_operations_in_millions = other.memtable_operations_in_millions;
}
public CfDef deepCopy() {
@ -770,6 +797,75 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
__isset_bit_vector.set(__KEY_CACHE_SAVE_PERIOD_IN_SECONDS_ISSET_ID, value);
}
public int getMemtable_flush_after_mins() {
return this.memtable_flush_after_mins;
}
public CfDef setMemtable_flush_after_mins(int memtable_flush_after_mins) {
this.memtable_flush_after_mins = memtable_flush_after_mins;
setMemtable_flush_after_minsIsSet(true);
return this;
}
public void unsetMemtable_flush_after_mins() {
__isset_bit_vector.clear(__MEMTABLE_FLUSH_AFTER_MINS_ISSET_ID);
}
/** Returns true if field memtable_flush_after_mins is set (has been asigned a value) and false otherwise */
public boolean isSetMemtable_flush_after_mins() {
return __isset_bit_vector.get(__MEMTABLE_FLUSH_AFTER_MINS_ISSET_ID);
}
public void setMemtable_flush_after_minsIsSet(boolean value) {
__isset_bit_vector.set(__MEMTABLE_FLUSH_AFTER_MINS_ISSET_ID, value);
}
public int getMemtable_throughput_in_mb() {
return this.memtable_throughput_in_mb;
}
public CfDef setMemtable_throughput_in_mb(int memtable_throughput_in_mb) {
this.memtable_throughput_in_mb = memtable_throughput_in_mb;
setMemtable_throughput_in_mbIsSet(true);
return this;
}
public void unsetMemtable_throughput_in_mb() {
__isset_bit_vector.clear(__MEMTABLE_THROUGHPUT_IN_MB_ISSET_ID);
}
/** Returns true if field memtable_throughput_in_mb is set (has been asigned a value) and false otherwise */
public boolean isSetMemtable_throughput_in_mb() {
return __isset_bit_vector.get(__MEMTABLE_THROUGHPUT_IN_MB_ISSET_ID);
}
public void setMemtable_throughput_in_mbIsSet(boolean value) {
__isset_bit_vector.set(__MEMTABLE_THROUGHPUT_IN_MB_ISSET_ID, value);
}
public double getMemtable_operations_in_millions() {
return this.memtable_operations_in_millions;
}
public CfDef setMemtable_operations_in_millions(double memtable_operations_in_millions) {
this.memtable_operations_in_millions = memtable_operations_in_millions;
setMemtable_operations_in_millionsIsSet(true);
return this;
}
public void unsetMemtable_operations_in_millions() {
__isset_bit_vector.clear(__MEMTABLE_OPERATIONS_IN_MILLIONS_ISSET_ID);
}
/** Returns true if field memtable_operations_in_millions is set (has been asigned a value) and false otherwise */
public boolean isSetMemtable_operations_in_millions() {
return __isset_bit_vector.get(__MEMTABLE_OPERATIONS_IN_MILLIONS_ISSET_ID);
}
public void setMemtable_operations_in_millionsIsSet(boolean value) {
__isset_bit_vector.set(__MEMTABLE_OPERATIONS_IN_MILLIONS_ISSET_ID, value);
}
public void setFieldValue(_Fields field, Object value) {
switch (field) {
case KEYSPACE:
@ -916,6 +1012,30 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
}
break;
case MEMTABLE_FLUSH_AFTER_MINS:
if (value == null) {
unsetMemtable_flush_after_mins();
} else {
setMemtable_flush_after_mins((Integer)value);
}
break;
case MEMTABLE_THROUGHPUT_IN_MB:
if (value == null) {
unsetMemtable_throughput_in_mb();
} else {
setMemtable_throughput_in_mb((Integer)value);
}
break;
case MEMTABLE_OPERATIONS_IN_MILLIONS:
if (value == null) {
unsetMemtable_operations_in_millions();
} else {
setMemtable_operations_in_millions((Double)value);
}
break;
}
}
@ -979,6 +1099,15 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
case KEY_CACHE_SAVE_PERIOD_IN_SECONDS:
return new Integer(getKey_cache_save_period_in_seconds());
case MEMTABLE_FLUSH_AFTER_MINS:
return new Integer(getMemtable_flush_after_mins());
case MEMTABLE_THROUGHPUT_IN_MB:
return new Integer(getMemtable_throughput_in_mb());
case MEMTABLE_OPERATIONS_IN_MILLIONS:
return new Double(getMemtable_operations_in_millions());
}
throw new IllegalStateException();
}
@ -1026,6 +1155,12 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
return isSetRow_cache_save_period_in_seconds();
case KEY_CACHE_SAVE_PERIOD_IN_SECONDS:
return isSetKey_cache_save_period_in_seconds();
case MEMTABLE_FLUSH_AFTER_MINS:
return isSetMemtable_flush_after_mins();
case MEMTABLE_THROUGHPUT_IN_MB:
return isSetMemtable_throughput_in_mb();
case MEMTABLE_OPERATIONS_IN_MILLIONS:
return isSetMemtable_operations_in_millions();
}
throw new IllegalStateException();
}
@ -1209,6 +1344,33 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
return false;
}
boolean this_present_memtable_flush_after_mins = true && this.isSetMemtable_flush_after_mins();
boolean that_present_memtable_flush_after_mins = true && that.isSetMemtable_flush_after_mins();
if (this_present_memtable_flush_after_mins || that_present_memtable_flush_after_mins) {
if (!(this_present_memtable_flush_after_mins && that_present_memtable_flush_after_mins))
return false;
if (this.memtable_flush_after_mins != that.memtable_flush_after_mins)
return false;
}
boolean this_present_memtable_throughput_in_mb = true && this.isSetMemtable_throughput_in_mb();
boolean that_present_memtable_throughput_in_mb = true && that.isSetMemtable_throughput_in_mb();
if (this_present_memtable_throughput_in_mb || that_present_memtable_throughput_in_mb) {
if (!(this_present_memtable_throughput_in_mb && that_present_memtable_throughput_in_mb))
return false;
if (this.memtable_throughput_in_mb != that.memtable_throughput_in_mb)
return false;
}
boolean this_present_memtable_operations_in_millions = true && this.isSetMemtable_operations_in_millions();
boolean that_present_memtable_operations_in_millions = true && that.isSetMemtable_operations_in_millions();
if (this_present_memtable_operations_in_millions || that_present_memtable_operations_in_millions) {
if (!(this_present_memtable_operations_in_millions && that_present_memtable_operations_in_millions))
return false;
if (this.memtable_operations_in_millions != that.memtable_operations_in_millions)
return false;
}
return true;
}
@ -1306,6 +1468,21 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
if (present_key_cache_save_period_in_seconds)
builder.append(key_cache_save_period_in_seconds);
boolean present_memtable_flush_after_mins = true && (isSetMemtable_flush_after_mins());
builder.append(present_memtable_flush_after_mins);
if (present_memtable_flush_after_mins)
builder.append(memtable_flush_after_mins);
boolean present_memtable_throughput_in_mb = true && (isSetMemtable_throughput_in_mb());
builder.append(present_memtable_throughput_in_mb);
if (present_memtable_throughput_in_mb)
builder.append(memtable_throughput_in_mb);
boolean present_memtable_operations_in_millions = true && (isSetMemtable_operations_in_millions());
builder.append(present_memtable_operations_in_millions);
if (present_memtable_operations_in_millions)
builder.append(memtable_operations_in_millions);
return builder.toHashCode();
}
@ -1479,6 +1656,33 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetMemtable_flush_after_mins()).compareTo(typedOther.isSetMemtable_flush_after_mins());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetMemtable_flush_after_mins()) { lastComparison = TBaseHelper.compareTo(this.memtable_flush_after_mins, typedOther.memtable_flush_after_mins);
if (lastComparison != 0) {
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetMemtable_throughput_in_mb()).compareTo(typedOther.isSetMemtable_throughput_in_mb());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetMemtable_throughput_in_mb()) { lastComparison = TBaseHelper.compareTo(this.memtable_throughput_in_mb, typedOther.memtable_throughput_in_mb);
if (lastComparison != 0) {
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetMemtable_operations_in_millions()).compareTo(typedOther.isSetMemtable_operations_in_millions());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetMemtable_operations_in_millions()) { lastComparison = TBaseHelper.compareTo(this.memtable_operations_in_millions, typedOther.memtable_operations_in_millions);
if (lastComparison != 0) {
return lastComparison;
}
}
return 0;
}
@ -1639,6 +1843,30 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
TProtocolUtil.skip(iprot, field.type);
}
break;
case 21: // MEMTABLE_FLUSH_AFTER_MINS
if (field.type == TType.I32) {
this.memtable_flush_after_mins = iprot.readI32();
setMemtable_flush_after_minsIsSet(true);
} else {
TProtocolUtil.skip(iprot, field.type);
}
break;
case 22: // MEMTABLE_THROUGHPUT_IN_MB
if (field.type == TType.I32) {
this.memtable_throughput_in_mb = iprot.readI32();
setMemtable_throughput_in_mbIsSet(true);
} else {
TProtocolUtil.skip(iprot, field.type);
}
break;
case 23: // MEMTABLE_OPERATIONS_IN_MILLIONS
if (field.type == TType.DOUBLE) {
this.memtable_operations_in_millions = iprot.readDouble();
setMemtable_operations_in_millionsIsSet(true);
} else {
TProtocolUtil.skip(iprot, field.type);
}
break;
default:
TProtocolUtil.skip(iprot, field.type);
}
@ -1763,6 +1991,21 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
oprot.writeI32(this.key_cache_save_period_in_seconds);
oprot.writeFieldEnd();
}
if (isSetMemtable_flush_after_mins()) {
oprot.writeFieldBegin(MEMTABLE_FLUSH_AFTER_MINS_FIELD_DESC);
oprot.writeI32(this.memtable_flush_after_mins);
oprot.writeFieldEnd();
}
if (isSetMemtable_throughput_in_mb()) {
oprot.writeFieldBegin(MEMTABLE_THROUGHPUT_IN_MB_FIELD_DESC);
oprot.writeI32(this.memtable_throughput_in_mb);
oprot.writeFieldEnd();
}
if (isSetMemtable_operations_in_millions()) {
oprot.writeFieldBegin(MEMTABLE_OPERATIONS_IN_MILLIONS_FIELD_DESC);
oprot.writeDouble(this.memtable_operations_in_millions);
oprot.writeFieldEnd();
}
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@ -1907,6 +2150,24 @@ public class CfDef implements TBase<CfDef, CfDef._Fields>, java.io.Serializable,
sb.append(this.key_cache_save_period_in_seconds);
first = false;
}
if (isSetMemtable_flush_after_mins()) {
if (!first) sb.append(", ");
sb.append("memtable_flush_after_mins:");
sb.append(this.memtable_flush_after_mins);
first = false;
}
if (isSetMemtable_throughput_in_mb()) {
if (!first) sb.append(", ");
sb.append("memtable_throughput_in_mb:");
sb.append(this.memtable_throughput_in_mb);
first = false;
}
if (isSetMemtable_operations_in_millions()) {
if (!first) sb.append(", ");
sb.append("memtable_operations_in_millions:");
sb.append(this.memtable_operations_in_millions);
first = false;
}
sb.append(")");
return sb.toString();
}

2
interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
View File

@ -43,6 +43,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
public static final String VERSION = "19.1.0";
public static final String VERSION = "19.2.0";
}

6
src/java/org/apache/cassandra/avro/CassandraServer.java
View File

@ -831,6 +831,7 @@ public class CassandraServer implements Cassandra {
String subCompare = cf_def.subcomparator_type == null ? D_CF_SUBCOMPTYPE : cf_def.subcomparator_type.toString();
CFMetaData.validateMinMaxCompactionThresholds(cf_def);
CFMetaData.validateMemtableSettings(cf_def);
return new CFMetaData(cf_def.keyspace.toString(),
cf_def.name.toString(),
@ -846,6 +847,11 @@ public class CassandraServer implements Cassandra {
DatabaseDescriptor.getComparator(validate),
cf_def.min_compaction_threshold == null ? CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD : cf_def.min_compaction_threshold,
cf_def.max_compaction_threshold == null ? CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD : cf_def.max_compaction_threshold,
cf_def.row_cache_save_period_in_seconds == null ? CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS : cf_def.row_cache_save_period_in_seconds,
cf_def.key_cache_save_period_in_seconds == null ? CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS : cf_def.key_cache_save_period_in_seconds,
cf_def.memtable_flush_after_mins == null ? CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS : cf_def.memtable_flush_after_mins,
cf_def.memtable_throughput_in_mb == null ? CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB : cf_def.memtable_throughput_in_mb,
cf_def.memtable_operations_in_millions == null ? CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS : cf_def.memtable_operations_in_millions,
ColumnDefinition.fromColumnDefs((Iterable<ColumnDef>) cf_def.column_metadata));
}

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

@ -26,6 +26,7 @@ import com.google.common.collect.BiMap;
import com.google.common.collect.HashBiMap;
import org.apache.commons.lang.builder.EqualsBuilder;
import org.apache.commons.lang.builder.HashCodeBuilder;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.avro.util.Utf8;
import org.apache.cassandra.avro.ColumnDef;
@ -51,6 +52,9 @@ public final class CFMetaData
public final static int DEFAULT_GC_GRACE_SECONDS = 864000;
public final static int DEFAULT_MIN_COMPACTION_THRESHOLD = 4;
public final static int DEFAULT_MAX_COMPACTION_THRESHOLD = 32;
public final static int DEFAULT_MEMTABLE_LIFETIME_IN_MINS = 60;
public final static int DEFAULT_MEMTABLE_THROUGHPUT_IN_MB = DatabaseDescriptor.sizeMemtableThroughput();
public final static double DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS = DatabaseDescriptor.sizeMemtableOperations(DEFAULT_MEMTABLE_THROUGHPUT_IN_MB);
private static final int MIN_CF_ID = 1000;
@ -80,10 +84,13 @@ public final class CFMetaData
BytesType.instance,
DEFAULT_MIN_COMPACTION_THRESHOLD,
DEFAULT_MAX_COMPACTION_THRESHOLD,
cfId,
Collections.<byte[], ColumnDefinition>emptyMap(),
DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS,
DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS);
DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS,
DEFAULT_MEMTABLE_LIFETIME_IN_MINS,
DEFAULT_MEMTABLE_THROUGHPUT_IN_MB,
DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS,
cfId,
Collections.<byte[], ColumnDefinition>emptyMap());
}
/**
@ -108,25 +115,30 @@ public final class CFMetaData
// never set it to less than 1000. this ensures that we have enough system CFids for future use.
idGen.set(cfIdMap.size() == 0 ? MIN_CF_ID : Math.max(Collections.max(cfIdMap.values()) + 1, MIN_CF_ID));
}
public final Integer cfId;
public final String tableName; // name of table which has this column family
public final String cfName; // name of the column family
public final ColumnFamilyType cfType; // type: super, standard, etc.
public final AbstractType comparator; // name sorted, time stamp sorted etc.
public final AbstractType subcolumnComparator; // like comparator, for supercolumns
public final String comment; // for humans only
public final double rowCacheSize; // default 0
public final double keyCacheSize; // default 0.01
public final double readRepairChance; // default 1.0 (always), chance [0.0,1.0] of read repair
public final boolean preloadRowCache; // default false
public final int gcGraceSeconds; // default 864000 (ten days)
public final AbstractType defaultValidator; // default none, use comparator types
public final Integer minCompactionThreshold; // default 4
public final Integer maxCompactionThreshold; // default 32
public final int rowCacheSavePeriodInSeconds; // default 0 (off)
public final int keyCacheSavePeriodInSeconds; // default 3600 (1 hour)
//REQUIRED
public final Integer cfId; // internal id, never exposed to user
public final String tableName; // name of keyspace
public final String cfName; // name of this column family
public final ColumnFamilyType cfType; // standard, super
public final AbstractType comparator; // bytes, long, timeuuid, utf8, etc.
public final AbstractType subcolumnComparator; // like comparator, for supercolumns
//OPTIONAL
public final String comment; // default none, for humans only
public final double rowCacheSize; // default 0
public final double keyCacheSize; // default 0.01
public final double readRepairChance; // default 1.0 (always), chance [0.0,1.0] of read repair
public final boolean preloadRowCache; // default false
public final int gcGraceSeconds; // default 864000 (ten days)
public final AbstractType defaultValidator; // default none, use comparator types
public final Integer minCompactionThreshold; // default 4
public final Integer maxCompactionThreshold; // default 32
public final int rowCacheSavePeriodInSeconds; // default 0 (off)
public final int keyCacheSavePeriodInSeconds; // default 3600 (1 hour)
public final int memtableFlushAfterMins; // default 60
public final int memtableThroughputInMb; // default based on heap size
public final double memtableOperationsInMillions; // default based on throughput
// NOTE: if you find yourself adding members to this class, make sure you keep the convert methods in lockstep.
public final Map<byte[], ColumnDefinition> column_metadata;
@ -145,10 +157,13 @@ public final class CFMetaData
AbstractType defaultValidator,
int minCompactionThreshold,
int maxCompactionThreshold,
Integer cfId,
Map<byte[], ColumnDefinition> column_metadata,
int rowCacheSavePeriodInSeconds,
int keyCacheSavePeriodInSeconds)
int keyCacheSavePeriodInSeconds,
int memtableFlushAfterMins,
Integer memtableThroughputInMb,
Double memtableOperationsInMillions,
Integer cfId,
Map<byte[], ColumnDefinition> column_metadata)
{
assert column_metadata != null;
@ -158,7 +173,9 @@ public final class CFMetaData
this.comparator = comparator;
// the default subcolumncomparator is null per thrift spec, but only should be null if cfType == Standard. If
// cfType == Super, subcolumnComparator should default to BytesType if not set.
this.subcolumnComparator = subcolumnComparator == null && cfType == ColumnFamilyType.Super ? BytesType.instance : subcolumnComparator;
this.subcolumnComparator = subcolumnComparator == null && cfType == ColumnFamilyType.Super
? BytesType.instance
: subcolumnComparator;
this.comment = comment == null ? "" : comment;
this.rowCacheSize = rowCacheSize;
this.preloadRowCache = preloadRowCache;
@ -168,10 +185,17 @@ public final class CFMetaData
this.defaultValidator = defaultValidator;
this.minCompactionThreshold = minCompactionThreshold;
this.maxCompactionThreshold = maxCompactionThreshold;
this.cfId = cfId;
this.column_metadata = Collections.unmodifiableMap(column_metadata);
this.rowCacheSavePeriodInSeconds = rowCacheSavePeriodInSeconds;
this.keyCacheSavePeriodInSeconds = keyCacheSavePeriodInSeconds;
this.memtableFlushAfterMins = memtableFlushAfterMins;
this.memtableThroughputInMb = memtableThroughputInMb == null
? DEFAULT_MEMTABLE_THROUGHPUT_IN_MB
: memtableThroughputInMb;
this.memtableOperationsInMillions = memtableOperationsInMillions == null
? DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS
: memtableOperationsInMillions;
this.cfId = cfId;
this.column_metadata = Collections.unmodifiableMap(column_metadata);
}
/** adds this cfm to the map. */
@ -200,6 +224,11 @@ public final class CFMetaData
AbstractType defaultValidator,
int minCompactionThreshold,
int maxCompactionThreshold,
int rowCacheSavePeriodInSeconds,
int keyCacheSavePeriodInSeconds,
int memTime,
Integer memSize,
Double memOps,
//This constructor generates the id!
Map<byte[], ColumnDefinition> column_metadata)
{
@ -217,10 +246,13 @@ public final class CFMetaData
defaultValidator,
minCompactionThreshold,
maxCompactionThreshold,
rowCacheSavePeriodInSeconds,
keyCacheSavePeriodInSeconds,
memTime,
memSize,
memOps,
nextId(),
column_metadata,
DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS,
DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS);
column_metadata);
}
public static CFMetaData newIndexMetadata(String table, String parentCf, ColumnDefinition info, AbstractType columnComparator)
@ -239,6 +271,11 @@ public final class CFMetaData
BytesType.instance,
DEFAULT_MIN_COMPACTION_THRESHOLD,
DEFAULT_MAX_COMPACTION_THRESHOLD,
DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS,
DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS,
DEFAULT_MEMTABLE_LIFETIME_IN_MINS,
DEFAULT_MEMTABLE_THROUGHPUT_IN_MB,
DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS,
Collections.<byte[], ColumnDefinition>emptyMap());
}
@ -259,10 +296,13 @@ public final class CFMetaData
cfm.defaultValidator,
cfm.minCompactionThreshold,
cfm.maxCompactionThreshold,
cfm.cfId,
cfm.column_metadata,
cfm.rowCacheSavePeriodInSeconds,
cfm.keyCacheSavePeriodInSeconds);
cfm.keyCacheSavePeriodInSeconds,
cfm.memtableFlushAfterMins,
cfm.memtableThroughputInMb,
cfm.memtableOperationsInMillions,
cfm.cfId,
cfm.column_metadata);
}
/** clones existing CFMetaData. keeps the id but changes the table name.*/
@ -282,10 +322,13 @@ public final class CFMetaData
cfm.defaultValidator,
cfm.minCompactionThreshold,
cfm.maxCompactionThreshold,
cfm.cfId,
cfm.column_metadata,
cfm.rowCacheSavePeriodInSeconds,
cfm.keyCacheSavePeriodInSeconds);
cfm.keyCacheSavePeriodInSeconds,
cfm.memtableFlushAfterMins,
cfm.memtableThroughputInMb,
cfm.memtableOperationsInMillions,
cfm.cfId,
cfm.column_metadata);
}
/** used for evicting cf data out of static tracking collections. */
@ -294,15 +337,6 @@ public final class CFMetaData
cfIdMap.remove(new Pair<String, String>(cfm.tableName, cfm.cfName));
}
// a quick and dirty pretty printer for describing the column family...
//TODO: Make it prettier, use it in the CLI
public String pretty()
{
return tableName + "." + cfName + "\n"
+ "Column Family Type: " + cfType + "\n"
+ "Columns Sorted By: " + comparator + "\n";
}
public org.apache.cassandra.avro.CfDef deflate()
{
org.apache.cassandra.avro.CfDef cf = new org.apache.cassandra.avro.CfDef();
@ -322,6 +356,11 @@ public final class CFMetaData
cf.default_validation_class = new Utf8(defaultValidator.getClass().getName());
cf.min_compaction_threshold = minCompactionThreshold;
cf.max_compaction_threshold = maxCompactionThreshold;
cf.row_cache_save_period_in_seconds = rowCacheSavePeriodInSeconds;
cf.key_cache_save_period_in_seconds = keyCacheSavePeriodInSeconds;
cf.memtable_flush_after_mins = memtableFlushAfterMins;
cf.memtable_throughput_in_mb = memtableThroughputInMb;
cf.memtable_operations_in_millions = memtableOperationsInMillions;
cf.column_metadata = SerDeUtils.createArray(column_metadata.size(),
org.apache.cassandra.avro.ColumnDef.SCHEMA$);
for (ColumnDefinition cd : column_metadata.values())
@ -354,11 +393,14 @@ public final class CFMetaData
column_metadata.put(cd.name, cd);
}
//isn't AVRO suppossed to handle stuff like this?
//isn't AVRO supposed to handle stuff like this?
Integer minct = cf.min_compaction_threshold == null ? DEFAULT_MIN_COMPACTION_THRESHOLD : cf.min_compaction_threshold;
Integer maxct = cf.max_compaction_threshold == null ? DEFAULT_MAX_COMPACTION_THRESHOLD : cf.max_compaction_threshold;
Integer row_cache_save_period_in_seconds = cf.row_cache_save_period_in_seconds == null ? DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS : cf.row_cache_save_period_in_seconds;
Integer key_cache_save_period_in_seconds = cf.key_cache_save_period_in_seconds == null ? DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS : cf.key_cache_save_period_in_seconds;
Integer memtable_flush_after_mins = cf.memtable_flush_after_mins == null ? DEFAULT_MEMTABLE_LIFETIME_IN_MINS : cf.memtable_flush_after_mins;
Integer memtable_throughput_in_mb = cf.memtable_throughput_in_mb == null ? DEFAULT_MEMTABLE_THROUGHPUT_IN_MB : cf.memtable_throughput_in_mb;
Double memtable_operations_in_millions = cf.memtable_operations_in_millions == null ? DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS : cf.memtable_operations_in_millions;
return new CFMetaData(cf.keyspace.toString(),
cf.name.toString(),
@ -374,10 +416,13 @@ public final class CFMetaData
validator,
minct,
maxct,
cf.id,
column_metadata,
row_cache_save_period_in_seconds,
key_cache_save_period_in_seconds);
key_cache_save_period_in_seconds,
memtable_flush_after_mins,
memtable_throughput_in_mb,
memtable_operations_in_millions,
cf.id,
column_metadata);
}
public boolean equals(Object obj)
@ -409,6 +454,9 @@ public final class CFMetaData
.append(column_metadata, rhs.column_metadata)
.append(rowCacheSavePeriodInSeconds, rhs.rowCacheSavePeriodInSeconds)
.append(keyCacheSavePeriodInSeconds, rhs.keyCacheSavePeriodInSeconds)
.append(memtableFlushAfterMins, rhs.memtableFlushAfterMins)
.append(memtableThroughputInMb, rhs.memtableThroughputInMb)
.append(memtableOperationsInMillions, rhs.memtableOperationsInMillions)
.isEquals();
}
@ -432,6 +480,9 @@ public final class CFMetaData
.append(column_metadata)
.append(rowCacheSavePeriodInSeconds)
.append(keyCacheSavePeriodInSeconds)
.append(memtableFlushAfterMins)
.append(memtableThroughputInMb)
.append(memtableOperationsInMillions)
.toHashCode();
}
@ -472,6 +523,7 @@ public final class CFMetaData
throw new ConfigurationException("subcolumncomparators do not match.");
validateMinMaxCompactionThresholds(cf_def);
validateMemtableSettings(cf_def);
return new CFMetaData(tableName,
cfName,
@ -487,10 +539,13 @@ public final class CFMetaData
DatabaseDescriptor.getComparator(cf_def.default_validation_class == null ? null : cf_def.default_validation_class.toString()),
cf_def.min_compaction_threshold,
cf_def.max_compaction_threshold,
cfId,
column_metadata,
cf_def.row_cache_save_period_in_seconds,
cf_def.key_cache_save_period_in_seconds);
cf_def.key_cache_save_period_in_seconds,
cf_def.memtable_flush_after_mins,
cf_def.memtable_throughput_in_mb,
cf_def.memtable_operations_in_millions,
cfId,
column_metadata);
}
// merges some final fields from this CFM with modifiable fields from CfDef into a new CFMetaData.
@ -517,6 +572,7 @@ public final class CFMetaData
throw new ConfigurationException("subcolumncomparators do not match.");
validateMinMaxCompactionThresholds(cf_def);
validateMemtableSettings(cf_def);
Map<byte[], ColumnDefinition> metadata = new HashMap<byte[], ColumnDefinition>();
if (cf_def.column_metadata == null)
@ -546,10 +602,13 @@ public final class CFMetaData
DatabaseDescriptor.getComparator(cf_def.default_validation_class == null ? null : cf_def.default_validation_class),
cf_def.min_compaction_threshold,
cf_def.max_compaction_threshold,
cfId,
metadata,
cf_def.row_cache_save_period_in_seconds,
cf_def.key_cache_save_period_in_seconds);
cf_def.key_cache_save_period_in_seconds,
cf_def.memtable_flush_after_mins,
cf_def.memtable_throughput_in_mb,
cf_def.memtable_operations_in_millions,
cfId,
metadata);
}
// converts CFM to thrift CfDef
@ -575,6 +634,9 @@ public final class CFMetaData
def.setMax_compaction_threshold(cfm.maxCompactionThreshold);
def.setRow_cache_save_period_in_seconds(cfm.rowCacheSavePeriodInSeconds);
def.setKey_cache_save_period_in_seconds(cfm.keyCacheSavePeriodInSeconds);
def.setMemtable_flush_after_mins(cfm.memtableFlushAfterMins);
def.setMemtable_throughput_in_mb(cfm.memtableThroughputInMb);
def.setMemtable_operations_in_millions(cfm.memtableOperationsInMillions);
List<org.apache.cassandra.thrift.ColumnDef> column_meta = new ArrayList< org.apache.cassandra.thrift.ColumnDef>(cfm.column_metadata.size());
for (ColumnDefinition cd : cfm.column_metadata.values())
{
@ -612,6 +674,11 @@ public final class CFMetaData
def.default_validation_class = cfm.defaultValidator.getClass().getName();
def.min_compaction_threshold = cfm.minCompactionThreshold;
def.max_compaction_threshold = cfm.maxCompactionThreshold;
def.row_cache_save_period_in_seconds = cfm.rowCacheSavePeriodInSeconds;
def.key_cache_save_period_in_seconds = cfm.keyCacheSavePeriodInSeconds;
def.memtable_flush_after_mins = cfm.memtableFlushAfterMins;
def.memtable_throughput_in_mb = cfm.memtableThroughputInMb;
def.memtable_operations_in_millions = cfm.memtableOperationsInMillions;
List<org.apache.cassandra.avro.ColumnDef> column_meta = new ArrayList<org.apache.cassandra.avro.ColumnDef>(cfm.column_metadata.size());
for (ColumnDefinition cd : cfm.column_metadata.values())
{
@ -686,28 +753,57 @@ public final class CFMetaData
}
}
public static void validateMemtableSettings(org.apache.cassandra.thrift.CfDef cf_def) throws ConfigurationException
{
if (cf_def.isSetMemtable_flush_after_mins() && cf_def.memtable_flush_after_mins <= 0) {
throw new ConfigurationException("memtable_flush_after_mins cannot be non-positive");
}
if (cf_def.isSetMemtable_throughput_in_mb() && cf_def.memtable_throughput_in_mb <= 0) {
throw new ConfigurationException("memtable_throughput_in_mb cannot be non-positive.");
}
if (cf_def.isSetMemtable_operations_in_millions() && cf_def.memtable_operations_in_millions <= 0) {
throw new ConfigurationException("memtable_operations_in_millions cannot be non-positive");
}
}
public static void validateMemtableSettings(org.apache.cassandra.avro.CfDef cf_def) throws ConfigurationException
{
if (cf_def.memtable_flush_after_mins != null && cf_def.memtable_flush_after_mins <= 0) {
throw new ConfigurationException("memtable_flush_after_mins cannot be non-positive");
}
if (cf_def.memtable_throughput_in_mb != null && cf_def.memtable_throughput_in_mb <= 0) {
throw new ConfigurationException("memtable_throughput_in_mb cannot be non-positive.");
}
if (cf_def.memtable_operations_in_millions != null && cf_def.memtable_operations_in_millions <= 0) {
throw new ConfigurationException("memtable_operations_in_millions cannot be non-positive");
}
}
@Override
public String toString()
{
return "CFMetaData{" +
"gcGraceSeconds=" + gcGraceSeconds +
", comparator=" + comparator +
", subcolumnComparator=" + subcolumnComparator +
", comment='" + comment + '\'' +
", rowCacheSize=" + rowCacheSize +
", keyCacheSize=" + keyCacheSize +
", readRepairChance=" + readRepairChance +
", preloadRowCache=" + preloadRowCache +
", cfId=" + cfId +
", tableName='" + tableName + '\'' +
", cfName='" + cfName + '\'' +
", cfType=" + cfType +
", defaultValidator=" + defaultValidator +
", minCompactionThreshold=" + minCompactionThreshold +
", maxCompactionThreshold=" + maxCompactionThreshold +
", rowcachesaveperiodsecs= " + rowCacheSavePeriodInSeconds +
", keycachesaveperiodsecs= " + keyCacheSavePeriodInSeconds +
", column_metadata=" + FBUtilities.toString(column_metadata) +
'}';
return new ToStringBuilder(this)
.append("cfId", cfId)
.append("tableName", tableName)
.append("cfName", cfName)
.append("cfType", cfType)
.append("comparator", comparator)
.append("subcolumncomparator", subcolumnComparator)
.append("comment", comment)
.append("rowCacheSize", rowCacheSize)
.append("keyCacheSize", keyCacheSize)
.append("readRepairChance", readRepairChance)
.append("preloadRowCache", preloadRowCache)
.append("gcGraceSeconds", gcGraceSeconds)
.append("defaultValidator", defaultValidator)
.append("minCompactionThreshold", minCompactionThreshold)
.append("maxCompactionThreshold", maxCompactionThreshold)
.append("rowCacheSavePeriodInSeconds", rowCacheSavePeriodInSeconds)
.append("keyCacheSavePeriodInSeconds", keyCacheSavePeriodInSeconds)
.append("memtableFlushAfterMins", memtableFlushAfterMins)
.append("memtableThroughputInMb", memtableThroughputInMb)
.append("memtableOperationsInMillions", memtableOperationsInMillions)
.append("column_metadata", column_metadata)
.toString();
}
}

6
src/java/org/apache/cassandra/config/Config.java
View File

@ -73,12 +73,6 @@ public class Config
public Integer compaction_thread_priority = Thread.MIN_PRIORITY;
public Integer binary_memtable_throughput_in_mb = 256;
/* Number of minutes to keep a memtable in memory */
public Integer memtable_flush_after_mins = 60 * 60 * 1000;
/* Size of the memtable in memory before it is dumped */
public Integer memtable_throughput_in_mb;
/* Number of objects in millions in the memtable before it is dumped */
public Double memtable_operations_in_millions;
/* if the size of columns or super-columns are more than this, indexing will kick in */
public Integer column_index_size_in_kb = 64;

34
src/java/org/apache/cassandra/config/Converter.java
View File

@ -62,11 +62,27 @@ public class Converter
String endPointSnitchClassName = null; // Used as a sentinel. EPS cannot be undefined in 0.6.
NodeList tablesxml = xmlUtils.getRequestedNodeList("/Storage/Keyspaces/Keyspace");
// Retrieve values that were previously global.
String gcGrace = xmlUtils.getNodeValue("/Storage/GCGraceSeconds");
int gc_grace_seconds = 864000;
int gc_grace_seconds = CFMetaData.DEFAULT_GC_GRACE_SECONDS;
if ( gcGrace != null )
gc_grace_seconds = Integer.parseInt(gcGrace);
String lifetime = xmlUtils.getNodeValue("/Storage/MemtableFlushAfterMinutes");
int memtime = CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS;
if (lifetime != null)
memtime = Integer.parseInt(lifetime);
String memtableSize = xmlUtils.getNodeValue("/Storage/MemtableThroughputInMB");
int memsize = CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB;
if (memtableSize != null)
memsize = Integer.parseInt(memtableSize);
String memtableOps = xmlUtils.getNodeValue("/Storage/MemtableOperationsInMillions");
double memops = CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS;
if (memtableOps != null)
memops = Double.parseDouble(memtableOps);
int size = tablesxml.getLength();
for ( int i = 0; i < size; ++i )
{
@ -135,7 +151,11 @@ public class Converter
ks.column_families[j].read_repair_chance = FBUtilities.parseDoubleOrPercent(value);
}
// Values that were previously global and are now per-CF go here.
ks.column_families[j].gc_grace_seconds = gc_grace_seconds;
ks.column_families[j].memtable_flush_after_mins = memtime;
ks.column_families[j].memtable_throughput_in_mb = memsize;
ks.column_families[j].memtable_operations_in_millions = memops;
ks.column_families[j].comment = xmlUtils.getNodeValue(xqlCF + "Comment");
}
@ -267,18 +287,6 @@ public class Converter
conf.auto_bootstrap = Boolean.valueOf(autoBootstr);
}
String lifetime = xmlUtils.getNodeValue("/Storage/MemtableFlushAfterMinutes");
if (lifetime != null)
conf.memtable_flush_after_mins = Integer.parseInt(lifetime);
String memtableSize = xmlUtils.getNodeValue("/Storage/MemtableThroughputInMB");
if ( memtableSize != null )
conf.memtable_throughput_in_mb = Integer.parseInt(memtableSize);
String memtableObjectCount = xmlUtils.getNodeValue("/Storage/MemtableOperationsInMillions");
if ( memtableObjectCount != null )
conf.memtable_operations_in_millions = Double.parseDouble(memtableObjectCount);
String columnIndexSize = xmlUtils.getNodeValue("/Storage/ColumnIndexSizeInKB");
if(columnIndexSize != null)
{

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

@ -139,17 +139,6 @@ public class DatabaseDescriptor
throw new ConfigurationException("Missing required directive CommitLogSync");
}
if (conf.memtable_throughput_in_mb == null)
{
conf.memtable_throughput_in_mb = (int) (Runtime.getRuntime().maxMemory() / (1048576 * 8));
logger.info("memtable_throughput_in_mb not configured, using " + conf.memtable_throughput_in_mb);
}
if (conf.memtable_operations_in_millions == null)
{
conf.memtable_operations_in_millions = 0.3 * conf.memtable_throughput_in_mb / 64.0;
logger.info("memtable_operations_in_millions not configured, using " + conf.memtable_operations_in_millions);
}
if (conf.commitlog_sync == Config.CommitLogSync.batch)
{
if (conf.commitlog_sync_batch_window_in_ms == null)
@ -322,13 +311,7 @@ public class DatabaseDescriptor
logger.debug("setting auto_bootstrap to " + conf.auto_bootstrap);
}
/* Number of objects in millions in the memtable before it is dumped */
if (conf.memtable_operations_in_millions != null && conf.memtable_operations_in_millions <= 0)
{
throw new ConfigurationException("memtable_operations_in_millions must be a positive double");
}
if (conf.in_memory_compaction_limit_in_mb != null && conf.in_memory_compaction_limit_in_mb <= 0)
if (conf.in_memory_compaction_limit_in_mb != null && conf.in_memory_compaction_limit_in_mb <= 0)
{
throw new ConfigurationException("in_memory_compaction_limit_in_mb must be a positive integer");
}
@ -568,14 +551,31 @@ public class DatabaseDescriptor
if (cf.min_compaction_threshold < 0 || cf.max_compaction_threshold < 0)
{
throw new ConfigurationException("min/max_compaction_thresholds must be non-negative integers.");
throw new ConfigurationException("min/max_compaction_thresholds must be positive integers.");
}
if ((cf.min_compaction_threshold > cf.max_compaction_threshold) && cf.max_compaction_threshold != 0)
{
throw new ConfigurationException("min_compaction_threshold must be smaller than max_compaction_threshold, or either must be 0 (disabled)");
}
Map<byte[], ColumnDefinition> metadata = new TreeMap<byte[], ColumnDefinition>(FBUtilities.byteArrayComparator);
if (cf.memtable_throughput_in_mb == null)
{
cf.memtable_throughput_in_mb = sizeMemtableThroughput();
logger.info("memtable_throughput_in_mb not configured for " + cf.name + ", using " + cf.memtable_throughput_in_mb);
}
if (cf.memtable_operations_in_millions == null)
{
cf.memtable_operations_in_millions = sizeMemtableOperations(cf.memtable_throughput_in_mb);
logger.info("memtable_operations_in_millions not configured for " + cf.name + ", using " + cf.memtable_operations_in_millions);
}
if (cf.memtable_operations_in_millions != null && cf.memtable_operations_in_millions <= 0)
{
throw new ConfigurationException("memtable_operations_in_millions must be a positive double");
}
Map<byte[], ColumnDefinition> metadata = new TreeMap<byte[], ColumnDefinition>(FBUtilities.byteArrayComparator);
for (RawColumnDefinition rcd : cf.column_metadata)
{
try
@ -603,6 +603,11 @@ public class DatabaseDescriptor
default_validator,
cf.min_compaction_threshold,
cf.max_compaction_threshold,
cf.row_cache_save_period_in_seconds,
cf.key_cache_save_period_in_seconds,
cf.memtable_flush_after_mins,
cf.memtable_throughput_in_mb,
cf.memtable_operations_in_millions,
metadata);
}
defs.add(new KSMetaData(keyspace.name,
@ -728,27 +733,12 @@ public class DatabaseDescriptor
return conf.column_index_size_in_kb * 1024;
}
public static int getMemtableLifetimeMS()
{
return conf.memtable_flush_after_mins * 60 * 1000;
}
public static String getInitialToken()
{
return conf.initial_token;
return conf.initial_token;
}
public static int getMemtableThroughput()
{
return conf.memtable_throughput_in_mb;
}
public static double getMemtableOperations()
{
return conf.memtable_operations_in_millions;
}
public static String getClusterName()
public static String getClusterName()
{
return conf.cluster_name;
}
@ -1126,4 +1116,13 @@ public class DatabaseDescriptor
{
return conf.dynamic_snitch_badness_threshold;
}
public static int sizeMemtableThroughput() {
return (int) (Runtime.getRuntime().maxMemory() / (1048576 * 8));
}
public static double sizeMemtableOperations(int mem_throughput) {
return 0.3 * mem_throughput / 64.0;
}
}

3
src/java/org/apache/cassandra/config/RawColumnFamily.java
View File

@ -44,4 +44,7 @@ public class RawColumnFamily
public RawColumnDefinition[] column_metadata = new RawColumnDefinition[0];
public int row_cache_save_period_in_seconds = CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS;
public int key_cache_save_period_in_seconds = CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS;
public int memtable_flush_after_mins = CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS;
public Integer memtable_throughput_in_mb;
public Double memtable_operations_in_millions;
}

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

@ -529,7 +529,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public String getFlushPath()
{
long guessedSize = 2 * DatabaseDescriptor.getMemtableThroughput() * 1024*1024; // 2* adds room for keys, column indexes
long guessedSize = 2 * metadata.memtableThroughputInMb * 1024*1024; // 2* adds room for keys, column indexes
String location = DatabaseDescriptor.getDataFileLocationForTable(table.name, guessedSize);
if (location == null)
throw new RuntimeException("Insufficient disk space to flush");

10
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -47,9 +47,6 @@ public class Memtable implements Comparable<Memtable>, IFlushable
private boolean isFrozen;
private final int THRESHOLD = DatabaseDescriptor.getMemtableThroughput() * 1024*1024; // not static since we might want to change at runtime
private final int THRESHOLD_COUNT = (int)(DatabaseDescriptor.getMemtableOperations() * 1024*1024);
private final AtomicInteger currentThroughput = new AtomicInteger(0);
private final AtomicInteger currentOperations = new AtomicInteger(0);
@ -57,11 +54,16 @@ public class Memtable implements Comparable<Memtable>, IFlushable
private final ConcurrentNavigableMap<DecoratedKey, ColumnFamily> columnFamilies = new ConcurrentSkipListMap<DecoratedKey, ColumnFamily>();
public final ColumnFamilyStore cfs;
private final int THRESHOLD;
private final int THRESHOLD_COUNT;
public Memtable(ColumnFamilyStore cfs)
{
this.cfs = cfs;
creationTime = System.currentTimeMillis();
this.THRESHOLD = cfs.metadata.memtableThroughputInMb * 1024 * 1024;
this.THRESHOLD_COUNT = (int) (cfs.metadata.memtableOperationsInMillions * 1024 * 1024);
}
/**
@ -288,6 +290,6 @@ public class Memtable implements Comparable<Memtable>, IFlushable
public boolean isExpired()
{
return System.currentTimeMillis() > creationTime + DatabaseDescriptor.getMemtableLifetimeMS();
return System.currentTimeMillis() > creationTime + cfs.metadata.memtableFlushAfterMins * 60 * 1000;
}
}

11
src/java/org/apache/cassandra/db/Table.java
View File

@ -253,8 +253,13 @@ public class Table
initCf(cfm.cfId, cfm.cfName);
}
// check 10x as often as the lifetime, so we can exceed lifetime by 10% at most
int checkMs = DatabaseDescriptor.getMemtableLifetimeMS() / 10;
// check 10x as often as the shortest lifetime, so we can exceed all lifetimes by 10% at most
int minCheckMs = Integer.MAX_VALUE;
for (ColumnFamilyStore cfs : columnFamilyStores.values())
{
minCheckMs = Math.min(minCheckMs, cfs.metadata.memtableFlushAfterMins * 60 * 1000);
}
Runnable runnable = new Runnable()
{
public void run()
@ -265,7 +270,7 @@ public class Table
}
}
};
flushTask = StorageService.scheduledTasks.scheduleWithFixedDelay(runnable, checkMs, checkMs, TimeUnit.MILLISECONDS);
flushTask = StorageService.scheduledTasks.scheduleWithFixedDelay(runnable, minCheckMs, minCheckMs, TimeUnit.MILLISECONDS);
}
public void dropCf(Integer cfId) throws IOException

8
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -914,6 +914,9 @@ public class CassandraServer implements Cassandra.Iface
throw new InvalidRequestException("Invalid column type " + cf_def.column_type);
}
CFMetaData.validateMinMaxCompactionThresholds(cf_def);
CFMetaData.validateMemtableSettings(cf_def);
return new CFMetaData(cf_def.keyspace,
cf_def.name,
cfType,
@ -928,6 +931,11 @@ public class CassandraServer implements Cassandra.Iface
DatabaseDescriptor.getComparator(cf_def.default_validation_class),
cf_def.isSetMin_compaction_threshold() ? cf_def.min_compaction_threshold : CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD,
cf_def.isSetMax_compaction_threshold() ? cf_def.max_compaction_threshold : CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD,
cf_def.isSetRow_cache_save_period_in_seconds() ? cf_def.row_cache_save_period_in_seconds : CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS,
cf_def.isSetKey_cache_save_period_in_seconds() ? cf_def.key_cache_save_period_in_seconds : CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS,
cf_def.isSetMemtable_flush_after_mins() ? cf_def.memtable_flush_after_mins : CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS,
cf_def.isSetMemtable_throughput_in_mb() ? cf_def.memtable_throughput_in_mb : CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB,
cf_def.isSetMemtable_operations_in_millions() ? cf_def.memtable_operations_in_millions : CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS,
ColumnDefinition.fromColumnDef(cf_def.column_metadata));
}

2
test/conf/cassandra.yaml
View File

@ -18,8 +18,6 @@ commitlog_rotation_threshold_in_mb: 128
data_file_directories:
- build/test/cassandra/data
disk_access_mode: mmap
memtable_throughput_in_mb: 1
memtable_operations_in_millions: 0.00002
seeds:
- 127.0.0.2
endpoint_snitch: org.apache.cassandra.locator.SimpleSnitch

11
test/unit/org/apache/cassandra/db/DefsTest.java
View File

@ -693,10 +693,15 @@ public class DefsTest extends CleanupHelper
false,
1.0,
0,
864000,
CFMetaData.DEFAULT_GC_GRACE_SECONDS,
BytesType.instance,
4,
32,
CFMetaData.DEFAULT_MIN_COMPACTION_THRESHOLD,
CFMetaData.DEFAULT_MAX_COMPACTION_THRESHOLD,
CFMetaData.DEFAULT_ROW_CACHE_SAVE_PERIOD_IN_SECONDS,
CFMetaData.DEFAULT_KEY_CACHE_SAVE_PERIOD_IN_SECONDS,
CFMetaData.DEFAULT_MEMTABLE_LIFETIME_IN_MINS,
CFMetaData.DEFAULT_MEMTABLE_THROUGHPUT_IN_MB,
CFMetaData.DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS,
Collections.<byte[], ColumnDefinition>emptyMap());
}