From bb0dc512d8cf7f11984cdee58d1cec590caa3044 Mon Sep 17 00:00:00 2001 From: Brandon Williams Date: Fri, 15 Oct 2010 17:42:57 +0000 Subject: [PATCH] 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 --- CHANGES.txt | 1 + conf/cassandra.yaml | 35 ++- interface/cassandra.genavro | 5 +- interface/cassandra.thrift | 5 +- .../org/apache/cassandra/thrift/CfDef.java | 265 +++++++++++++++++- .../apache/cassandra/thrift/Constants.java | 2 +- .../cassandra/avro/CassandraServer.java | 6 + .../apache/cassandra/config/CFMetaData.java | 246 +++++++++++----- .../org/apache/cassandra/config/Config.java | 6 - .../apache/cassandra/config/Converter.java | 34 ++- .../cassandra/config/DatabaseDescriptor.java | 73 +++-- .../cassandra/config/RawColumnFamily.java | 3 + .../cassandra/db/ColumnFamilyStore.java | 2 +- .../org/apache/cassandra/db/Memtable.java | 10 +- src/java/org/apache/cassandra/db/Table.java | 11 +- .../cassandra/thrift/CassandraServer.java | 8 + test/conf/cassandra.yaml | 2 - .../org/apache/cassandra/db/DefsTest.java | 11 +- 18 files changed, 558 insertions(+), 167 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index ef0165d5a2..3354732a86 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index c117d4a250..d7bb20429d 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -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 diff --git a/interface/cassandra.genavro b/interface/cassandra.genavro index f9a6f93e3a..11e041d85e 100644 --- a/interface/cassandra.genavro +++ b/interface/cassandra.genavro @@ -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, null } column_metadata; } diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift index a9be47acba..94316069b5 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -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. */ diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java index bbe9d49b07..8fc525eb4c 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java @@ -68,6 +68,9 @@ public class CfDef implements TBase, 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, 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, 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 byName = new HashMap(); @@ -158,6 +167,12 @@ public class CfDef implements TBase, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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(); } diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java index c8769589e5..0e0b248888 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java @@ -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"; } diff --git a/src/java/org/apache/cassandra/avro/CassandraServer.java b/src/java/org/apache/cassandra/avro/CassandraServer.java index ab3de867d4..b88c7e1ac5 100644 --- a/src/java/org/apache/cassandra/avro/CassandraServer.java +++ b/src/java/org/apache/cassandra/avro/CassandraServer.java @@ -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) cf_def.column_metadata)); } diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index 858b625938..d2761c7212 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -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.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.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 column_metadata; @@ -145,10 +157,13 @@ public final class CFMetaData AbstractType defaultValidator, int minCompactionThreshold, int maxCompactionThreshold, - Integer cfId, - Map column_metadata, int rowCacheSavePeriodInSeconds, - int keyCacheSavePeriodInSeconds) + int keyCacheSavePeriodInSeconds, + int memtableFlushAfterMins, + Integer memtableThroughputInMb, + Double memtableOperationsInMillions, + Integer cfId, + Map 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 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.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(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 metadata = new HashMap(); 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 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 column_meta = new ArrayList(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(); } } diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 3963894b68..def0a5e04b 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -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; diff --git a/src/java/org/apache/cassandra/config/Converter.java b/src/java/org/apache/cassandra/config/Converter.java index e10bc8c8f2..f65012f79d 100644 --- a/src/java/org/apache/cassandra/config/Converter.java +++ b/src/java/org/apache/cassandra/config/Converter.java @@ -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) { diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index c188e11ac9..a3d31251e4 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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 metadata = new TreeMap(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 metadata = new TreeMap(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; + } + } diff --git a/src/java/org/apache/cassandra/config/RawColumnFamily.java b/src/java/org/apache/cassandra/config/RawColumnFamily.java index 4ce595d1c3..a486cf67c7 100644 --- a/src/java/org/apache/cassandra/config/RawColumnFamily.java +++ b/src/java/org/apache/cassandra/config/RawColumnFamily.java @@ -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; } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 37919bc8bd..758c8bf7a7 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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"); diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index b6d58b135e..78fe00d8f1 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -47,9 +47,6 @@ public class Memtable implements Comparable, 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, IFlushable private final ConcurrentNavigableMap columnFamilies = new ConcurrentSkipListMap(); 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, IFlushable public boolean isExpired() { - return System.currentTimeMillis() > creationTime + DatabaseDescriptor.getMemtableLifetimeMS(); + return System.currentTimeMillis() > creationTime + cfs.metadata.memtableFlushAfterMins * 60 * 1000; } } diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 02217cf0d7..6463e5ff76 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -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 diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 911b745abc..fed8f96ed6 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -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)); } diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml index a06afcce27..b74da12a61 100644 --- a/test/conf/cassandra.yaml +++ b/test/conf/cassandra.yaml @@ -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 diff --git a/test/unit/org/apache/cassandra/db/DefsTest.java b/test/unit/org/apache/cassandra/db/DefsTest.java index b6ba98ea06..3bd0969649 100644 --- a/test/unit/org/apache/cassandra/db/DefsTest.java +++ b/test/unit/org/apache/cassandra/db/DefsTest.java @@ -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.emptyMap()); }