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

Add optional compression for sstables 

patch by xedin; reviewed by slebresne for CASSANDRA-47


git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1153210 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Sylvain Lebresne 2011-08-02 17:28:55 +00:00
parent 455dcc54ec
commit 1ecabe6842
32 changed files with 1253 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
NOTICE.txt
View File

@ -29,3 +29,5 @@ Copyright 1995-2006 Mort Bay Consulting Pty Ltd
YAML support provided by snakeyaml (http://code.google.com/p/snakeyaml/).
Copyright (c) 2008-2010 Andrey Somov
Compression support provided by snappy-java (http://code.google.com/p/snappy-java/)

8
build.xml
View File

@ -988,6 +988,14 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision}
<jvmarg value="-Dcorrupt-sstable-root=${test.data}/corrupt-sstables"/>
</testmacro>
</target>
<target name="test-compression" depends="build-test" description="Execute unit tests with sstable compression enabled">
<testmacro suitename="unit" inputdir="${test.unit.src}" timeout="60000">
<jvmarg value="-Dlegacy-sstable-root=${test.data}/legacy-sstables"/>
<jvmarg value="-Dcorrupt-sstable-root=${test.data}/corrupt-sstables"/>
<jvmarg value="-Dcassandra.test.compression=true"/>
</testmacro>
</target>
<target name="msg-ser-gen-test" depends="build-test" description="Generates message serializations">
<testmacro suitename="unit" inputdir="${test.unit.src}"

3
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.11.0"
const string VERSION = "19.12.0"
#
@ -397,6 +397,7 @@ struct CfDef {
29: optional string compaction_strategy,
30: optional map<string,string> compaction_strategy_options,
31: optional i32 row_cache_keys_to_save,
32: optional bool compression,
}
/* describes a keyspace. */

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

@ -9086,8 +9086,6 @@ public class Cassandra {
private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
try {
// it doesn't seem like you should have to do this, but java serialization is wacky, and doesn't call the default constructor.
__isset_bit_vector = new BitSet(1);
read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
} catch (org.apache.thrift.TException te) {
throw new java.io.IOException(te);

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

@ -72,6 +72,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
private static final org.apache.thrift.protocol.TField COMPACTION_STRATEGY_FIELD_DESC = new org.apache.thrift.protocol.TField("compaction_strategy", org.apache.thrift.protocol.TType.STRING, (short)29);
private static final org.apache.thrift.protocol.TField COMPACTION_STRATEGY_OPTIONS_FIELD_DESC = new org.apache.thrift.protocol.TField("compaction_strategy_options", org.apache.thrift.protocol.TType.MAP, (short)30);
private static final org.apache.thrift.protocol.TField ROW_CACHE_KEYS_TO_SAVE_FIELD_DESC = new org.apache.thrift.protocol.TField("row_cache_keys_to_save", org.apache.thrift.protocol.TType.I32, (short)31);
private static final org.apache.thrift.protocol.TField COMPRESSION_FIELD_DESC = new org.apache.thrift.protocol.TField("compression", org.apache.thrift.protocol.TType.BOOL, (short)32);
public String keyspace;
public String name;
@ -100,6 +101,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
public String compaction_strategy;
public Map<String,String> compaction_strategy_options;
public int row_cache_keys_to_save;
public boolean compression;
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@ -129,7 +131,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
KEY_ALIAS((short)28, "key_alias"),
COMPACTION_STRATEGY((short)29, "compaction_strategy"),
COMPACTION_STRATEGY_OPTIONS((short)30, "compaction_strategy_options"),
ROW_CACHE_KEYS_TO_SAVE((short)31, "row_cache_keys_to_save");
ROW_CACHE_KEYS_TO_SAVE((short)31, "row_cache_keys_to_save"),
COMPRESSION((short)32, "compression");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@ -198,6 +201,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
return COMPACTION_STRATEGY_OPTIONS;
case 31: // ROW_CACHE_KEYS_TO_SAVE
return ROW_CACHE_KEYS_TO_SAVE;
case 32: // COMPRESSION
return COMPRESSION;
default:
return null;
}
@ -252,7 +257,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
private static final int __REPLICATE_ON_WRITE_ISSET_ID = 11;
private static final int __MERGE_SHARDS_CHANCE_ISSET_ID = 12;
private static final int __ROW_CACHE_KEYS_TO_SAVE_ISSET_ID = 13;
private BitSet __isset_bit_vector = new BitSet(14);
private static final int __COMPRESSION_ISSET_ID = 14;
private BitSet __isset_bit_vector = new BitSet(15);
public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
static {
@ -314,6 +320,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING))));
tmpMap.put(_Fields.ROW_CACHE_KEYS_TO_SAVE, new org.apache.thrift.meta_data.FieldMetaData("row_cache_keys_to_save", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.I32)));
tmpMap.put(_Fields.COMPRESSION, new org.apache.thrift.meta_data.FieldMetaData("compression", org.apache.thrift.TFieldRequirementType.OPTIONAL,
new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.BOOL)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(CfDef.class, metaDataMap);
}
@ -418,6 +426,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
this.compaction_strategy_options = __this__compaction_strategy_options;
}
this.row_cache_keys_to_save = other.row_cache_keys_to_save;
this.compression = other.compression;
}
public CfDef deepCopy() {
@ -470,6 +479,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
this.compaction_strategy_options = null;
setRow_cache_keys_to_saveIsSet(false);
this.row_cache_keys_to_save = 0;
setCompressionIsSet(false);
this.compression = false;
}
public String getKeyspace() {
@ -1142,6 +1153,29 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
__isset_bit_vector.set(__ROW_CACHE_KEYS_TO_SAVE_ISSET_ID, value);
}
public boolean isCompression() {
return this.compression;
}
public CfDef setCompression(boolean compression) {
this.compression = compression;
setCompressionIsSet(true);
return this;
}
public void unsetCompression() {
__isset_bit_vector.clear(__COMPRESSION_ISSET_ID);
}
/** Returns true if field compression is set (has been assigned a value) and false otherwise */
public boolean isSetCompression() {
return __isset_bit_vector.get(__COMPRESSION_ISSET_ID);
}
public void setCompressionIsSet(boolean value) {
__isset_bit_vector.set(__COMPRESSION_ISSET_ID, value);
}
public void setFieldValue(_Fields field, Object value) {
switch (field) {
case KEYSPACE:
@ -1360,6 +1394,14 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
}
break;
case COMPRESSION:
if (value == null) {
unsetCompression();
} else {
setCompression((Boolean)value);
}
break;
}
}
@ -1446,6 +1488,9 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
case ROW_CACHE_KEYS_TO_SAVE:
return new Integer(getRow_cache_keys_to_save());
case COMPRESSION:
return new Boolean(isCompression());
}
throw new IllegalStateException();
}
@ -1511,6 +1556,8 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
return isSetCompaction_strategy_options();
case ROW_CACHE_KEYS_TO_SAVE:
return isSetRow_cache_keys_to_save();
case COMPRESSION:
return isSetCompression();
}
throw new IllegalStateException();
}
@ -1771,6 +1818,15 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
return false;
}
boolean this_present_compression = true && this.isSetCompression();
boolean that_present_compression = true && that.isSetCompression();
if (this_present_compression || that_present_compression) {
if (!(this_present_compression && that_present_compression))
return false;
if (this.compression != that.compression)
return false;
}
return true;
}
@ -1913,6 +1969,11 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
if (present_row_cache_keys_to_save)
builder.append(row_cache_keys_to_save);
boolean present_compression = true && (isSetCompression());
builder.append(present_compression);
if (present_compression)
builder.append(compression);
return builder.toHashCode();
}
@ -2194,6 +2255,16 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetCompression()).compareTo(typedOther.isSetCompression());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetCompression()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.compression, typedOther.compression);
if (lastComparison != 0) {
return lastComparison;
}
}
return 0;
}
@ -2437,6 +2508,14 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
break;
case 32: // COMPRESSION
if (field.type == org.apache.thrift.protocol.TType.BOOL) {
this.compression = iprot.readBool();
setCompressionIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
break;
default:
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, field.type);
}
@ -2624,6 +2703,11 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
oprot.writeI32(this.row_cache_keys_to_save);
oprot.writeFieldEnd();
}
if (isSetCompression()) {
oprot.writeFieldBegin(COMPRESSION_FIELD_DESC);
oprot.writeBool(this.compression);
oprot.writeFieldEnd();
}
oprot.writeFieldStop();
oprot.writeStructEnd();
}
@ -2842,6 +2926,12 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
sb.append(this.row_cache_keys_to_save);
first = false;
}
if (isSetCompression()) {
if (!first) sb.append(", ");
sb.append("compression:");
sb.append(this.compression);
first = false;
}
sb.append(")");
return sb.toString();
}

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

@ -44,6 +44,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
public static final String VERSION = "19.11.0";
public static final String VERSION = "19.12.0";
}

209
lib/licenses/snappy-java-1.0.3.txt Normal file
View File

@ -0,0 +1,209 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. Definitions.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
THIRD-PARTY DEPENDENCIES
========================
Convenience copies of some third-party dependencies are distributed with
Apache Cassandra as Java jar files in lib/. Licensing information for
these files can be found in the lib/licenses directory.

BIN
lib/snappy-java-1.0.3.jar Normal file
View File

Binary file not shown.

1
src/avro/internode.genavro
View File

@ -67,6 +67,7 @@ protocol InterNode {
union { null, bytes } key_alias = null;
union { null, string } compaction_strategy = null;
union { null, map<string> } compaction_strategy_options = null;
boolean compression = false;
}
@aliases(["org.apache.cassandra.config.avro.KsDef"])

7
src/java/org/apache/cassandra/cli/CliClient.java
View File

@ -134,7 +134,8 @@ public class CliClient
ROW_CACHE_PROVIDER,
KEY_VALIDATION_CLASS,
COMPACTION_STRATEGY,
COMPACTION_STRATEGY_OPTIONS
COMPACTION_STRATEGY_OPTIONS,
COMPRESSION
}
private static final String DEFAULT_PLACEMENT_STRATEGY = "org.apache.cassandra.locator.NetworkTopologyStrategy";
@ -1259,6 +1260,9 @@ public class CliClient
case COMPACTION_STRATEGY_OPTIONS:
cfDef.setCompaction_strategy_options(getStrategyOptionsFromTree(statement.getChild(i+1)));
break;
case COMPRESSION:
cfDef.setCompression(Boolean.parseBoolean(mValue));
break;
default:
//must match one of the above or we'd throw an exception at the valueOf statement above.
assert(false);
@ -1731,6 +1735,7 @@ public class CliClient
sessionState.out.printf(" Compaction min/max thresholds: %s/%s%n", cf_def.min_compaction_threshold, cf_def.max_compaction_threshold);
sessionState.out.printf(" Read repair chance: %s%n", cf_def.read_repair_chance);
sessionState.out.printf(" Replicate on write: %s%n", cf_def.replicate_on_write);
sessionState.out.printf(" Compression enabled: %s%n", cf_def.compression);
// if we have connection to the cfMBean established
if (cfMBean != null)

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

@ -80,6 +80,7 @@ public final class CFMetaData
public final static String DEFAULT_ROW_CACHE_PROVIDER = "org.apache.cassandra.cache.ConcurrentLinkedHashCacheProvider";
public final static String DEFAULT_COMPACTION_STRATEGY_CLASS = "org.apache.cassandra.db.compaction.SizeTieredCompactionStrategy";
public final static ByteBuffer DEFAULT_KEY_NAME = ByteBufferUtil.bytes("KEY");
public final static boolean DEFAULT_COMPRESSION = false;
private static final int MIN_CF_ID = 1000;
private static final AtomicInteger idGen = new AtomicInteger(MIN_CF_ID);
@ -171,6 +172,7 @@ public final class CFMetaData
private double mergeShardsChance; // default 0.1, chance [0.0, 1.0] of merging old shards during replication
private IRowCacheProvider rowCacheProvider;
private ByteBuffer keyAlias; // default NULL
private boolean compression;
private Map<ByteBuffer, ColumnDefinition> column_metadata;
public Class<? extends AbstractCompactionStrategy> compactionStrategyClass;
@ -193,6 +195,7 @@ public final class CFMetaData
public CFMetaData memOps(double prop) {memtableOperationsInMillions = prop; return this;}
public CFMetaData mergeShardsChance(double prop) {mergeShardsChance = prop; return this;}
public CFMetaData keyAlias(ByteBuffer prop) {keyAlias = prop; return this;}
public CFMetaData compression(boolean prop) {compression = prop; return this; }
public CFMetaData columnMetadata(Map<ByteBuffer,ColumnDefinition> prop) {column_metadata = prop; return this;}
public CFMetaData rowCacheProvider(IRowCacheProvider prop) { rowCacheProvider = prop; return this;}
public CFMetaData compactionStrategyClass(Class<? extends AbstractCompactionStrategy> prop) {compactionStrategyClass = prop; return this;}
@ -243,6 +246,7 @@ public final class CFMetaData
memtableThroughputInMb = DEFAULT_MEMTABLE_THROUGHPUT_IN_MB;
memtableOperationsInMillions = DEFAULT_MEMTABLE_OPERATIONS_IN_MILLIONS;
mergeShardsChance = DEFAULT_MERGE_SHARDS_CHANCE;
compression = DEFAULT_COMPRESSION;
try
{
rowCacheProvider = FBUtilities.newCacheProvider(DEFAULT_ROW_CACHE_PROVIDER);
@ -328,7 +332,8 @@ public final class CFMetaData
.memOps(oldCFMD.memtableOperationsInMillions)
.columnMetadata(oldCFMD.column_metadata)
.compactionStrategyClass(oldCFMD.compactionStrategyClass)
.compactionStrategyOptions(oldCFMD.compactionStrategyOptions);
.compactionStrategyOptions(oldCFMD.compactionStrategyOptions)
.compression(oldCFMD.compression);
}
/** used for evicting cf data out of static tracking collections. */
@ -378,6 +383,7 @@ public final class CFMetaData
cf.memtable_operations_in_millions = memtableOperationsInMillions;
cf.merge_shards_chance = mergeShardsChance;
cf.key_alias = keyAlias;
cf.compression = compression;
cf.column_metadata = SerDeUtils.createArray(column_metadata.size(),
org.apache.cassandra.db.migration.avro.ColumnDef.SCHEMA$);
for (ColumnDefinition cd : column_metadata.values())
@ -477,7 +483,8 @@ public final class CFMetaData
.gcGraceSeconds(cf.gc_grace_seconds)
.defaultValidator(validator)
.keyValidator(keyValidator)
.columnMetadata(column_metadata);
.columnMetadata(column_metadata)
.compression(cf.compression);
}
public String getComment()
@ -570,6 +577,16 @@ public final class CFMetaData
return keyAlias == null ? DEFAULT_KEY_NAME : keyAlias;
}
public boolean useCompression()
{
return compression;
}
public void useCompression(boolean flag)
{
compression = flag;
}
public Map<ByteBuffer, ColumnDefinition> getColumn_metadata()
{
return Collections.unmodifiableMap(column_metadata);
@ -617,6 +634,7 @@ public final class CFMetaData
.append(memtableOperationsInMillions, rhs.memtableOperationsInMillions)
.append(mergeShardsChance, rhs.mergeShardsChance)
.append(keyAlias, rhs.keyAlias)
.append(compression, rhs.compression)
.append(compactionStrategyClass, rhs.compactionStrategyClass)
.append(compactionStrategyOptions, rhs.compactionStrategyOptions)
.isEquals();
@ -649,6 +667,7 @@ public final class CFMetaData
.append(memtableOperationsInMillions)
.append(mergeShardsChance)
.append(keyAlias)
.append(compression)
.append(compactionStrategyClass)
.append(compactionStrategyOptions)
.toHashCode();
@ -697,6 +716,8 @@ public final class CFMetaData
cf_def.compaction_strategy = DEFAULT_COMPACTION_STRATEGY_CLASS;
if (null == cf_def.compaction_strategy_options)
cf_def.compaction_strategy_options = Collections.<String, String>emptyMap();
if (!cf_def.isSetCompression())
cf_def.setCompression(CFMetaData.DEFAULT_COMPRESSION);
}
public static CFMetaData fromThrift(org.apache.cassandra.thrift.CfDef cf_def) throws InvalidRequestException, ConfigurationException
@ -748,7 +769,8 @@ public final class CFMetaData
.replicateOnWrite(cf_def.replicate_on_write)
.defaultValidator(TypeParser.parse(cf_def.default_validation_class))
.keyValidator(TypeParser.parse(cf_def.key_validation_class))
.columnMetadata(ColumnDefinition.fromColumnDef(cf_def.column_metadata));
.columnMetadata(ColumnDefinition.fromColumnDef(cf_def.column_metadata))
.compression(cf_def.compression);
}
// merges some final fields from this CFM with modifiable fields from CfDef into a new CFMetaData.
@ -800,6 +822,7 @@ public final class CFMetaData
if (cf_def.row_cache_provider != null)
rowCacheProvider = FBUtilities.newCacheProvider(cf_def.row_cache_provider.toString());
keyAlias = cf_def.key_alias;
compression = cf_def.compression;
// adjust column definitions. figure out who is coming and going.
Set<ByteBuffer> toRemove = new HashSet<ByteBuffer>();
@ -918,6 +941,7 @@ public final class CFMetaData
def.setMemtable_operations_in_millions(cfm.memtableOperationsInMillions);
def.setMerge_shards_chance(cfm.mergeShardsChance);
def.setKey_alias(cfm.getKeyName());
def.setCompression(cfm.compression);
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())
{
@ -966,6 +990,7 @@ public final class CFMetaData
def.merge_shards_chance = cfm.mergeShardsChance;
def.key_validation_class = cfm.keyValidator.getClass().getName();
def.key_alias = cfm.keyAlias;
def.compression = cfm.compression;
List<org.apache.cassandra.db.migration.avro.ColumnDef> column_meta = new ArrayList<org.apache.cassandra.db.migration.avro.ColumnDef>(cfm.column_metadata.size());
for (ColumnDefinition cd : cfm.column_metadata.values())
{
@ -1011,7 +1036,7 @@ public final class CFMetaData
newDef.subcomparator_type = def.getSubcomparator_type();
newDef.merge_shards_chance = def.getMerge_shards_chance();
newDef.key_alias = def.key_alias;
newDef.compression = def.compression;
List<org.apache.cassandra.db.migration.avro.ColumnDef> columnMeta = new ArrayList<org.apache.cassandra.db.migration.avro.ColumnDef>();
if (def.isSetColumn_metadata())
{
@ -1146,6 +1171,7 @@ public final class CFMetaData
.append("memtableOperationsInMillions", memtableOperationsInMillions)
.append("mergeShardsChance", mergeShardsChance)
.append("keyAlias", keyAlias)
.append("compression", compression)
.append("column_metadata", column_metadata)
.append("compactionStrategyClass", compactionStrategyClass)
.append("compactionStrategyOptions", compactionStrategyOptions)

5
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -32,6 +32,7 @@ import javax.management.ObjectName;
import com.google.common.base.Predicates;
import com.google.common.collect.Iterators;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -46,7 +47,6 @@ import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.service.AntiEntropyService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.OperationType;
import org.apache.cassandra.utils.*;
/**
@ -471,7 +471,8 @@ public class CompactionManager implements CompactionManagerMBean
// we'll also loop through the index at the same time, using the position from the index to recover if the
// row header (key or data size) is corrupt. (This means our position in the index file will be one row
// "ahead" of the data file.)
final RandomAccessReader dataFile = RandomAccessReader.open(new File(sstable.getFilename()), true);
final RandomAccessReader dataFile = sstable.openDataReader(true);
String indexFilename = sstable.descriptor.filenameFor(Component.PRIMARY_INDEX);
RandomAccessReader indexFile = RandomAccessReader.open(new File(indexFilename), true);
try

176
src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java Normal file
View File

@ -0,0 +1,176 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.compress;
import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.WritableByteChannel;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.streaming.FileStreamTask;
import org.apache.cassandra.streaming.PendingFile;
import org.apache.cassandra.utils.Pair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xerial.snappy.Snappy;
public class CompressedRandomAccessReader extends RandomAccessReader
{
private static final Logger logger = LoggerFactory.getLogger(CompressedRandomAccessReader.class);
/**
* Transfer sections of the file to the given target channel
* This method streams decompressed data so receiving party responsible for compression
*
* @param file The compressed file to transfer
* @param target Channel to transfer data into
*
* @throws IOException on any I/O error.
*/
public static void transfer(PendingFile file, WritableByteChannel target) throws IOException
{
RandomAccessReader compressedFile = CompressedRandomAccessReader.open(file.getFilename(), true);
try
{
for (Pair<Long, Long> section : file.sections)
{
long length = section.right - section.left;
compressedFile.seek(section.left);
while (length > 0)
{
int toRead = (length > FileStreamTask.CHUNK_SIZE) ? FileStreamTask.CHUNK_SIZE : (int) length;
ByteBuffer buffer = compressedFile.readBytes(toRead);
long bytesTransferred = 0;
while (bytesTransferred < toRead)
{
// we don't need to re-read a buffer, it will write starting from buffer.position()
long lastWrite = target.write(buffer);
bytesTransferred += lastWrite;
file.progress += lastWrite;
}
length -= bytesTransferred;
if (logger.isDebugEnabled())
logger.debug("Bytes transferred " + bytesTransferred + "/" + file.size);
}
}
}
finally
{
FileUtils.closeQuietly(compressedFile);
}
}
/**
* Get metadata about given compressed file including uncompressed data length, chunk size
* and list of the chunk offsets of the compressed data.
*
* @param dataFilePath Path to the compressed file
*
* @return metadata about given compressed file.
*/
public static CompressionMetadata metadata(String dataFilePath)
{
Descriptor desc = Descriptor.fromFilename(dataFilePath);
try
{
return new CompressionMetadata(desc.filenameFor(Component.COMPRESSION_INFO), new File(dataFilePath).length());
}
catch (IOException e)
{
throw new IOError(e);
}
}
public static RandomAccessReader open(String dataFilePath, boolean skipIOCache) throws IOException
{
return open(dataFilePath, metadata(dataFilePath), skipIOCache);
}
public static RandomAccessReader open(String dataFilePath, CompressionMetadata metadata) throws IOException
{
return open(dataFilePath, metadata, false);
}
public static RandomAccessReader open(String dataFilePath, CompressionMetadata metadata, boolean skipIOCache) throws IOException
{
return new CompressedRandomAccessReader(dataFilePath, metadata, skipIOCache);
}
private final CompressionMetadata metadata;
// used by reBuffer() to escape creating lots of temporary buffers
private final byte[] compressed;
private final FileInputStream source;
public CompressedRandomAccessReader(String dataFilePath, CompressionMetadata metadata, boolean skipIOCache) throws IOException
{
super(new File(dataFilePath), metadata.chunkLength, skipIOCache);
this.metadata = metadata;
compressed = new byte[metadata.chunkLength];
// can't use super.read(...) methods
// that is why we are allocating special InputStream to read data from disk
// from already open file descriptor
source = new FileInputStream(getFD());
}
@Override
protected void reBuffer() throws IOException
{
decompressChunk(metadata.chunkFor(current));
}
private void decompressChunk(CompressionMetadata.Chunk chunk) throws IOException
{
if (source.getChannel().position() != chunk.offset)
source.getChannel().position(chunk.offset);
if (source.read(compressed, 0, chunk.length) != chunk.length)
throw new IOException(String.format("(%s) failed to read %d bytes from offset %d.", getPath(), chunk.length, chunk.offset));
validBufferBytes = Snappy.rawUncompress(compressed, 0, chunk.length, buffer, 0);
// buffer offset is always aligned
bufferOffset = current & ~(buffer.length - 1);
}
@Override
public long length() throws IOException
{
return metadata.dataLength;
}
@Override
public String toString()
{
return String.format("%s - chunk length %d, data length %d.", getPath(), metadata.chunkLength, metadata.dataLength);
}
}

186
src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java Normal file
View File

@ -0,0 +1,186 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.compress;
import java.io.File;
import java.io.IOException;
import org.apache.cassandra.io.util.FileMark;
import org.apache.cassandra.io.util.SequentialWriter;
import org.xerial.snappy.Snappy;
public class CompressedSequentialWriter extends SequentialWriter
{
public static final int CHUNK_LENGTH = 65536;
public static SequentialWriter open(String dataFilePath, String indexFilePath, boolean skipIOCache) throws IOException
{
return new CompressedSequentialWriter(new File(dataFilePath), indexFilePath, skipIOCache);
}
// holds offset in the file where current chunk should be written
// changed only by flush() method where data buffer gets compressed and stored to the file
private long chunkOffset = 0;
// index file writer (random I/O)
private final CompressionMetadata.Writer metadataWriter;
// used to store compressed data
private final byte[] compressed;
// holds a number of already written chunks
private int chunkCount = 0;
public CompressedSequentialWriter(File file, String indexFilePath, boolean skipIOCache) throws IOException
{
super(file, CHUNK_LENGTH, skipIOCache);
// buffer for compression should be the same size as buffer itself
compressed = new byte[buffer.length];
/* Index File (-CompressionInfo.db component) and it's header */
metadataWriter = new CompressionMetadata.Writer(indexFilePath);
metadataWriter.writeHeader(Snappy.class.getSimpleName(), CHUNK_LENGTH);
}
@Override
public void sync() throws IOException
{
throw new UnsupportedOperationException();
}
@Override
public void flush() throws IOException
{
throw new UnsupportedOperationException();
}
@Override
protected void flushData() throws IOException
{
seekToChunkStart();
// compressing data with buffer re-use
int compressedLength = Snappy.rawCompress(buffer, 0, validBufferBytes, compressed, 0);
// write an offset of the newly written chunk to the index file
metadataWriter.writeLong(chunkOffset);
chunkCount++;
// write data itself
out.write(compressed, 0, compressedLength);
// next chunk should be written right after current
chunkOffset += compressedLength;
}
@Override
public FileMark mark()
{
return new CompressedFileWriterMark(chunkOffset, current, validBufferBytes, chunkCount + 1);
}
@Override
public synchronized void resetAndTruncate(FileMark mark) throws IOException
{
assert mark instanceof CompressedFileWriterMark;
CompressedFileWriterMark realMark = ((CompressedFileWriterMark) mark);
// reset position
current = realMark.uncDataOffset;
if (realMark.chunkOffset == chunkOffset) // current buffer
{
// just reset a buffer offset and return
validBufferBytes = realMark.bufferOffset;
return;
}
// synchronize current buffer with disk
// because we don't want any data loss
syncInternal();
// setting marker as a current offset
chunkOffset = realMark.chunkOffset;
// compressed chunk size
int chunkSize = (int) (metadataWriter.chunkOffsetBy(realMark.nextChunkIndex) - chunkOffset);
out.seek(chunkOffset);
out.read(compressed, 0, chunkSize);
Snappy.rawUncompress(compressed, 0, chunkSize, buffer, 0);
// reset buffer
validBufferBytes = realMark.bufferOffset;
bufferOffset = current - validBufferBytes;
chunkCount = realMark.nextChunkIndex - 1;
// truncate data and index file
truncate(chunkOffset);
metadataWriter.resetAndTruncate(realMark.nextChunkIndex);
}
/**
* Seek to the offset where next compressed data chunk should be stored.
*
* @throws IOException on any I/O error.
*/
private void seekToChunkStart() throws IOException
{
if (out.getFilePointer() != chunkOffset)
out.seek(chunkOffset);
}
@Override
public void close() throws IOException
{
if (buffer == null)
return; // already closed
super.close();
metadataWriter.finalizeHeader(current, chunkCount);
metadataWriter.close();
}
/**
* Class to hold a mark to the position of the file
*/
protected static class CompressedFileWriterMark implements FileMark
{
// chunk offset in the compressed file
long chunkOffset;
// uncompressed data offset (real data offset)
long uncDataOffset;
int bufferOffset;
int nextChunkIndex;
public CompressedFileWriterMark(long chunkOffset, long uncDataOffset, int bufferOffset, int nextChunkIndex)
{
this.chunkOffset = chunkOffset;
this.uncDataOffset = uncDataOffset;
this.bufferOffset = bufferOffset;
this.nextChunkIndex = nextChunkIndex;
}
}
}

209
src/java/org/apache/cassandra/io/compress/CompressionMetadata.java Normal file
View File

@ -0,0 +1,209 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.compress;
import java.io.*;
import org.apache.cassandra.io.util.FileUtils;
/**
* Holds metadata about compressed file
*/
public class CompressionMetadata
{
public final int chunkLength;
public final long dataLength;
public final long compressedFileLength;
public final long[] chunkOffsets;
public final String indexFilePath;
public final String algorithm;
public CompressionMetadata(String indexFilePath, long compressedLength) throws IOException
{
this.indexFilePath = indexFilePath;
DataInputStream stream = new DataInputStream(new FileInputStream(indexFilePath));
algorithm = stream.readUTF();
chunkLength = stream.readInt();
dataLength = stream.readLong();
compressedFileLength = compressedLength;
chunkOffsets = readChunkOffsets(stream);
FileUtils.closeQuietly(stream);
}
/**
* Read offsets of the individual chunks from the given input.
*
* @param input Source of the data.
*
* @return collection of the chunk offsets.
*
* @throws java.io.IOException on any I/O error (except EOF).
*/
private long[] readChunkOffsets(DataInput input) throws IOException
{
int chunkCount = input.readInt();
long[] offsets = new long[chunkCount];
for (int i = 0; i < offsets.length; i++)
{
try
{
offsets[i] = input.readLong();
}
catch (EOFException e)
{
throw new EOFException(String.format("Corrupted Index File %s: read %d but expected %d chunks.",
indexFilePath,
i,
chunkCount));
}
}
return offsets;
}
/**
* Get a chunk of compressed data (offset, length) corresponding to given position
*
* @param position Position in the file.
* @return pair of chunk offset and length.
* @throws java.io.IOException on any I/O error.
*/
public Chunk chunkFor(long position) throws IOException
{
// position of the chunk
int idx = (int) (position / chunkLength);
if (idx >= chunkOffsets.length)
throw new EOFException();
long chunkOffset = chunkOffsets[idx];
long nextChunkOffset = (idx + 1 == chunkOffsets.length)
? compressedFileLength
: chunkOffsets[idx + 1];
return new Chunk(chunkOffset, (int) (nextChunkOffset - chunkOffset));
}
public static class Writer extends RandomAccessFile
{
// place for uncompressed data length in the index file
private long dataLengthOffset = -1;
public Writer(String path) throws IOException
{
super(path, "rw");
}
public void writeHeader(String algorithm, int chunkLength) throws IOException
{
// algorithm
writeUTF(algorithm);
// store the length of the chunk
writeInt(chunkLength);
// store position and reserve a place for uncompressed data length and chunks count
dataLengthOffset = getFilePointer();
writeLong(-1);
writeInt(-1);
}
public void finalizeHeader(long dataLength, int chunks) throws IOException
{
assert dataLengthOffset != -1 : "writeHeader wasn't called";
long currentPosition = getFilePointer();
// seek back to the data length position
seek(dataLengthOffset);
// write uncompressed data length and chunks count
writeLong(dataLength);
writeInt(chunks);
// seek forward to the previous position
seek(currentPosition);
}
/**
* Get a chunk offset by it's index.
*
* @param chunkIndex Index of the chunk.
*
* @return offset of the chunk in the compressed file.
*
* @throws IOException any I/O error.
*/
public long chunkOffsetBy(int chunkIndex) throws IOException
{
if (dataLengthOffset == -1)
throw new IllegalStateException("writeHeader wasn't called");
long position = getFilePointer();
// seek to the position of the given chunk
seek(dataLengthOffset
+ 8 // size reserved for uncompressed data length
+ 4 // size reserved for chunk count
+ (chunkIndex * 8));
try
{
return readLong();
}
finally
{
// back to the original position
seek(position);
}
}
/**
* Reset the writer so that the next chunk offset written will be the
* one of {@code chunkIndex}.
*/
public void resetAndTruncate(int chunkIndex) throws IOException
{
seek(dataLengthOffset
+ 8 // size reserved for uncompressed data length
+ 4 // size reserved for chunk count
+ (chunkIndex * 8));
getChannel().truncate(getFilePointer());
}
}
/**
* Holds offset and length of the file chunk
*/
public class Chunk
{
public final long offset;
public final int length;
public Chunk(long offset, int length)
{
this.offset = offset;
this.length = length;
}
}
}

5
src/java/org/apache/cassandra/io/sstable/Component.java
View File

@ -47,6 +47,8 @@ public class Component
FILTER("Filter.db"),
// 0-length file that is created when an sstable is ready to be deleted
COMPACTED_MARKER("Compacted"),
// file to hold information about uncompressed data length, chunk offsets etc.
COMPRESSION_INFO("CompressionInfo.db"),
// statistical metadata about the content of the sstable
STATS("Statistics.db"),
// a bitmap secondary index: many of these may exist per sstable
@ -72,6 +74,7 @@ public class Component
public final static Component PRIMARY_INDEX = new Component(Type.PRIMARY_INDEX, -1);
public final static Component FILTER = new Component(Type.FILTER, -1);
public final static Component COMPACTED_MARKER = new Component(Type.COMPACTED_MARKER, -1);
public final static Component COMPRESSION_INFO = new Component(Type.COMPRESSION_INFO, -1);
public final static Component STATS = new Component(Type.STATS, -1);
public final Type type;
@ -101,6 +104,7 @@ public class Component
case PRIMARY_INDEX:
case FILTER:
case COMPACTED_MARKER:
case COMPRESSION_INFO:
case STATS:
return type.repr;
case BITMAP_INDEX:
@ -137,6 +141,7 @@ public class Component
case PRIMARY_INDEX: component = Component.PRIMARY_INDEX; break;
case FILTER: component = Component.FILTER; break;
case COMPACTED_MARKER: component = Component.COMPACTED_MARKER; break;
case COMPRESSION_INFO: component = Component.COMPRESSION_INFO; break;
case STATS: component = Component.STATS; break;
case BITMAP_INDEX:
component = new Component(type, id);

4
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -64,6 +64,7 @@ public abstract class SSTable
protected final Set<Component> components;
public final CFMetaData metadata;
public final IPartitioner partitioner;
public final boolean compression;
protected SSTable(Descriptor descriptor, CFMetaData metadata, IPartitioner partitioner)
{
@ -82,6 +83,8 @@ public abstract class SSTable
Set<Component> dataComponents = new HashSet<Component>(components);
for (Component component : components)
assert component.type != Component.Type.COMPACTED_MARKER;
this.compression = dataComponents.contains(Component.COMPRESSION_INFO);
this.components = Collections.unmodifiableSet(dataComponents);
this.metadata = metadata;
this.partitioner = partitioner;
@ -107,6 +110,7 @@ public abstract class SSTable
{
if (component.equals(Component.DATA) || component.equals(Component.COMPACTED_MARKER))
continue;
FileUtils.deleteWithConfirm(desc.filenameFor(component));
}
// remove the COMPACTED_MARKER component last if it exists

17
src/java/org/apache/cassandra/io/sstable/SSTableReader.java
View File

@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -249,7 +250,9 @@ public class SSTableReader extends SSTable
{
boolean cacheLoading = keyCache != null && !keysToLoadInCache.isEmpty();
SegmentedFile.Builder ibuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getIndexAccessMode());
SegmentedFile.Builder dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
SegmentedFile.Builder dbuilder = (components.contains(Component.COMPRESSION_INFO))
? SegmentedFile.getCompressedBuilder()
: SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
// we read the positions in a BRAF so we don't have to worry about an entry spanning a mmap boundary.
RandomAccessReader input = RandomAccessReader.open(new File(descriptor.filenameFor(Component.PRIMARY_INDEX)), true);
@ -800,6 +803,18 @@ public class SSTableReader extends SSTable
return sstableMetadata.getMaxTimestamp();
}
public RandomAccessReader openDataReader(boolean skipIOCache) throws IOException
{
return openDataReader(RandomAccessReader.DEFAULT_BUFFER_SIZE, skipIOCache);
}
public RandomAccessReader openDataReader(int bufferSize, boolean skipIOCache) throws IOException
{
return compression
? CompressedRandomAccessReader.open(getFilename(), skipIOCache)
: RandomAccessReader.open(new File(getFilename()), bufferSize, skipIOCache);
}
public static void acquireReferences(Iterable<SSTableReader> sstables)
{
for (SSTableReader sstable : sstables)

5
src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.IOError;
import java.io.IOException;
import java.util.Arrays;
@ -53,7 +52,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
{
try
{
this.file = RandomAccessReader.open(new File(sstable.getFilename()), skipCache);
this.file = sstable.openDataReader(bufferSize, skipCache);
}
catch (IOException e)
{
@ -70,7 +69,7 @@ public class SSTableScanner implements CloseableIterator<IColumnIterator>
{
try
{
this.file = RandomAccessReader.open(new File(sstable.getFilename()), bufferSize);
this.file = sstable.openDataReader(bufferSize, false);
}
catch (IOException e)
{

40
src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
View File

@ -21,10 +21,7 @@ package org.apache.cassandra.io.sstable;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.*;
import com.google.common.collect.Sets;
@ -36,11 +33,10 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.*;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.compress.CompressedSequentialWriter;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.OperationType;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@ -65,6 +61,14 @@ public class SSTableWriter extends SSTable
SSTableMetadata.createCollector());
}
private static Set<Component> components(CFMetaData metadata)
{
Set<Component> components = new HashSet<Component>(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS));
if (metadata.useCompression())
components.add(Component.COMPRESSION_INFO);
return components;
}
public SSTableWriter(String filename,
long keyCount,
CFMetaData metadata,
@ -72,12 +76,24 @@ public class SSTableWriter extends SSTable
SSTableMetadata.Collector sstableMetadataCollector) throws IOException
{
super(Descriptor.fromFilename(filename),
new HashSet<Component>(Arrays.asList(Component.DATA, Component.FILTER, Component.PRIMARY_INDEX, Component.STATS)),
components(metadata),
metadata,
partitioner);
iwriter = new IndexWriter(descriptor, partitioner, keyCount);
dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
dataFile = SequentialWriter.open(new File(getFilename()), true);
if (compression)
{
dbuilder = SegmentedFile.getCompressedBuilder();
dataFile = CompressedSequentialWriter.open(getFilename(),
descriptor.filenameFor(Component.COMPRESSION_INFO),
true);
}
else
{
dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode());
dataFile = SequentialWriter.open(new File(getFilename()), true);
}
this.sstableMetadataCollector = sstableMetadataCollector;
}
@ -205,10 +221,8 @@ public class SSTableWriter extends SSTable
// index and filter
iwriter.close();
// main data
long position = dataFile.getFilePointer();
dataFile.close(); // calls force
FileUtils.truncate(dataFile.getPath(), position);
// main data, close will truncate if necessary
dataFile.close();
// write sstable statistics
SSTableMetadata sstableMetadata = sstableMetadataCollector.finalizeMetadata();

77
src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java Normal file
View File

@ -0,0 +1,77 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.util;
import java.io.IOError;
import java.io.IOException;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.apache.cassandra.io.compress.CompressionMetadata;
public class CompressedSegmentedFile extends SegmentedFile
{
private final CompressionMetadata metadata;
public CompressedSegmentedFile(String path, CompressionMetadata metadata)
{
super(path, metadata.dataLength);
this.metadata = metadata;
}
public static class Builder extends SegmentedFile.Builder
{
/**
* Adds a position that would be a safe place for a segment boundary in the file. For a block/row based file
* format, safe boundaries are block/row edges.
* @param boundary The absolute position of the potential boundary in the file.
*/
public void addPotentialBoundary(long boundary)
{
// only one segment in a standard-io file
}
/**
* Called after all potential boundaries have been added to apply this Builder to a concrete file on disk.
* @param path The file on disk.
*/
public SegmentedFile complete(String path)
{
return new CompressedSegmentedFile(path, CompressedRandomAccessReader.metadata(path));
}
}
public FileDataInput getSegment(long position, int bufferSize)
{
try
{
RandomAccessReader file = CompressedRandomAccessReader.open(path, metadata);
file.seek(position);
return file;
}
catch (IOException e)
{
throw new IOError(e);
}
}
public void cleanup()
{
// nothing to do
}
}

20
src/java/org/apache/cassandra/io/util/RandomAccessReader.java
View File

@ -1,3 +1,21 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.util;
import java.io.EOFException;
@ -32,7 +50,7 @@ public class RandomAccessReader extends RandomAccessFile implements FileDataInpu
protected int validBufferBytes = 0;
// channel liked with the file, used to retrieve data and force updates.
private final FileChannel channel;
protected final FileChannel channel;
private final boolean skipIOCache;

5
src/java/org/apache/cassandra/io/util/SegmentedFile.java
View File

@ -61,6 +61,11 @@ public abstract class SegmentedFile
: new BufferedSegmentedFile.Builder();
}
public static Builder getCompressedBuilder()
{
return new CompressedSegmentedFile.Builder();
}
public abstract FileDataInput getSegment(long position, int bufferSize);
/**

61
src/java/org/apache/cassandra/io/util/SequentialWriter.java
View File

@ -26,7 +26,7 @@ import org.apache.cassandra.utils.CLibrary;
public class SequentialWriter extends OutputStream
{
// isDirty - true if this.buffer contains any un-synced bytes
private boolean isDirty = false, syncNeeded = false;
protected boolean isDirty = false, syncNeeded = false;
// absolute path to the given file
private final String filePath;
@ -34,14 +34,14 @@ public class SequentialWriter extends OutputStream
// so we can use the write(int) path w/o tons of new byte[] allocations
private final byte[] singleByteBuffer = new byte[1];
private byte[] buffer;
protected byte[] buffer;
private final boolean skipIOCache;
private final int fd;
private long current = 0, bufferOffset;
private int validBufferBytes;
protected long current = 0, bufferOffset;
protected int validBufferBytes;
private final RandomAccessFile out;
protected final RandomAccessFile out;
// used if skip I/O cache was enabled
private long ioCacheStartOffset = 0, bytesSinceCacheFlush = 0;
@ -139,10 +139,15 @@ public class SequentialWriter extends OutputStream
* @throws java.io.IOException on any I/O error.
*/
public void sync() throws IOException
{
syncInternal();
}
protected void syncInternal() throws IOException
{
if (syncNeeded)
{
flush();
flushInternal();
out.getFD().sync();
syncNeeded = false;
@ -158,10 +163,15 @@ public class SequentialWriter extends OutputStream
*/
@Override
public void flush() throws IOException
{
flushInternal();
}
protected void flushInternal() throws IOException
{
if (isDirty)
{
out.write(buffer, 0, validBufferBytes);
flushData();
if (skipIOCache)
{
@ -187,6 +197,15 @@ public class SequentialWriter extends OutputStream
}
}
/**
* Override this method instead of overriding flush()
* @throws IOException on any I/O error.
*/
protected void flushData() throws IOException
{
out.write(buffer, 0, validBufferBytes);
}
public long getFilePointer()
{
return current;
@ -202,14 +221,13 @@ public class SequentialWriter extends OutputStream
return filePath;
}
private void reBuffer() throws IOException
protected void reBuffer() throws IOException
{
flush();
flushInternal();
resetBuffer();
}
private void resetBuffer()
protected void resetBuffer()
{
bufferOffset = current;
validBufferBytes = 0;
@ -229,12 +247,18 @@ public class SequentialWriter extends OutputStream
{
assert mark instanceof BufferedFileWriterMark;
long previous = current;
current = ((BufferedFileWriterMark) mark).pointer;
if (previous - current <= validBufferBytes) // current buffer
{
validBufferBytes = validBufferBytes - ((int) (previous - current));
return;
}
// synchronize current buffer with disk
// because we don't want any data loss
sync();
// setting marker as a current offset
current = ((BufferedFileWriterMark) mark).pointer;
syncInternal();
// truncate file to given position
truncate(current);
@ -253,14 +277,17 @@ public class SequentialWriter extends OutputStream
@Override
public void close() throws IOException
{
sync();
if (buffer == null)
return; // already closed
syncInternal();
buffer = null;
if (skipIOCache && bytesSinceCacheFlush > 0)
CLibrary.trySkipCache(fd, 0, 0);
out.close(); // this will also close channel for us
out.close();
}
/**

7
src/java/org/apache/cassandra/streaming/FileStreamTask.java
View File

@ -28,6 +28,7 @@ import java.nio.channels.FileChannel;
import java.nio.channels.SocketChannel;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.io.compress.CompressedRandomAccessReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -91,6 +92,12 @@ public class FileStreamTask extends WrappedRunnable
if (header.file == null)
return;
if (header.file.sstable.compression)
{
CompressedRandomAccessReader.transfer(header.file, channel);
return;
}
RandomAccessFile raf = new RandomAccessFile(new File(header.file.getFilename()), "r");
try
{

6
src/java/org/apache/cassandra/streaming/IncomingStreamReader.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.streaming;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.SocketChannel;
import java.util.Collections;
import org.slf4j.Logger;
@ -35,10 +34,7 @@ import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.compaction.AbstractCompactedRow;
import org.apache.cassandra.db.compaction.CompactionController;
import org.apache.cassandra.db.compaction.PrecompactedRow;
import org.apache.cassandra.io.sstable.IndexHelper;
import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.sstable.SSTableWriter;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.ByteBufferUtil;

4
src/resources/org/apache/cassandra/cli/CliHelp.yaml
View File

@ -531,6 +531,8 @@ commands:
NOTE: Thsi provider requires JNA.jar to be in the class path to
enable native methods.
- compression: Use compression for SSTable data files. Accepts the values true and false.
Examples:
create column family Super4
with column_type = 'Super'
@ -772,6 +774,8 @@ commands:
NOTE: Thsi provider requires JNA.jar to be in the class path to
enable native methods.
- compression: Use compression for SSTable data files. Accepts the values true and false.
Examples:
update column family Super4
with column_type = 'Super'

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

@ -239,10 +239,24 @@ public class SchemaLoader
false,
standardCFMD(ks_nocommit, "Standard1")));
if (Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")))
useCompression(schema);
return schema;
}
private static void useCompression(List<KSMetaData> schema)
{
for (KSMetaData ksm : schema)
{
for (CFMetaData cfm : ksm.cfMetaData().values())
{
cfm.compression(true);
}
}
}
private static CFMetaData standardCFMD(String ksName, String cfName)
{
return new CFMetaData(ksName, cfName, ColumnFamilyType.Standard, BytesType.instance, null).keyCacheSize(0);

3
test/unit/org/apache/cassandra/db/TableTest.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.db;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
import java.text.DecimalFormat;
@ -406,7 +405,7 @@ public class TableTest extends CleanupHelper
// verify that we do indeed have multiple index entries
SSTableReader sstable = cfStore.getSSTables().iterator().next();
long position = sstable.getPosition(key, SSTableReader.Operator.EQ);
RandomAccessReader file = RandomAccessReader.open(new File(sstable.getFilename()));
RandomAccessReader file = sstable.openDataReader(false);
file.seek(position);
assert ByteBufferUtil.readWithShortLength(file).equals(key.key);
SSTableReader.readRowSize(file, sstable.descriptor);

90
test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java Normal file
View File

@ -0,0 +1,90 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.apache.cassandra.io.compress;
import java.io.*;
import org.junit.Test;
import org.apache.cassandra.io.util.*;
public class CompressedRandomAccessReaderTest
{
@Test
public void testResetAndTruncate() throws IOException
{
// test reset in current buffer or previous one
testResetAndTruncate(false, 10);
testResetAndTruncate(false, CompressedSequentialWriter.CHUNK_LENGTH);
}
@Test
public void testResetAndTruncateCompressed() throws IOException
{
// test reset in current buffer or previous one
testResetAndTruncate(true, 10);
testResetAndTruncate(true, CompressedSequentialWriter.CHUNK_LENGTH);
}
private void testResetAndTruncate(boolean compressed, int junkSize) throws IOException
{
String filename = "corruptFile";
File f = new File(filename);
try
{
SequentialWriter writer = compressed
? new CompressedSequentialWriter(f, filename + ".metadata", false)
: new SequentialWriter(f, CompressedSequentialWriter.CHUNK_LENGTH, false);
writer.write("The quick ".getBytes());
FileMark mark = writer.mark();
writer.write("blue fox jumps over the lazy dog".getBytes());
// write enough to be sure to change chunk
for (int i = 0; i < junkSize; ++i)
{
writer.write((byte)1);
}
writer.resetAndTruncate(mark);
writer.write("brown fox jumps over the lazy dog".getBytes());
writer.close();
assert f.exists();
RandomAccessReader reader = compressed
? new CompressedRandomAccessReader(filename, new CompressionMetadata(filename + ".metadata", f.length()), false)
: new RandomAccessReader(f, CompressedSequentialWriter.CHUNK_LENGTH, false);
String expected = "The quick brown fox jumps over the lazy dog";
assert reader.length() == expected.length();
byte[] b = new byte[expected.length()];
reader.readFully(b);
assert new String(b).equals(expected) : "Expecting '" + expected + "', got '" + new String(b) + "'";
}
finally
{
// cleanup
if (f.exists())
f.delete();
File metadata = new File(filename + ".metadata");
if (compressed && metadata.exists())
metadata.delete();
}
}
}

5
test/unit/org/apache/cassandra/io/sstable/SSTableTest.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.*;
@ -51,7 +50,7 @@ public class SSTableTest extends CleanupHelper
private void verifySingle(SSTableReader sstable, ByteBuffer bytes, ByteBuffer key) throws IOException
{
RandomAccessReader file = RandomAccessReader.open(new File(sstable.getFilename()));
RandomAccessReader file = sstable.openDataReader(false);
file.seek(sstable.getPosition(sstable.partitioner.decorateKey(key), SSTableReader.Operator.EQ));
assert key.equals(ByteBufferUtil.readWithShortLength(file));
int size = (int)SSTableReader.readRowSize(file, sstable.descriptor);
@ -88,7 +87,7 @@ public class SSTableTest extends CleanupHelper
{
List<ByteBuffer> keys = new ArrayList<ByteBuffer>(map.keySet());
Collections.shuffle(keys);
RandomAccessReader file = RandomAccessReader.open(new File(sstable.getFilename()));
RandomAccessReader file = sstable.openDataReader(false);
for (ByteBuffer key : keys)
{
file.seek(sstable.getPosition(sstable.partitioner.decorateKey(key), SSTableReader.Operator.EQ));

14
tools/stress/src/org/apache/cassandra/stress/Session.java
View File

@ -77,6 +77,7 @@ public class Session implements Serializable
availableOptions.addOption("W", "no-replicate-on-write",false, "Set replicate_on_write to false for counters. Only counter add with CL=ONE will work");
availableOptions.addOption("V", "average-size-values", false, "Generate column values of average rather than specific size");
availableOptions.addOption("T", "send-to", true, "Send this as a request to the stress daemon at specified address.");
availableOptions.addOption("I", "compression", false, "Use sstable compression when creating schema");
}
private int numKeys = 1000 * 1000;
@ -92,6 +93,7 @@ public class Session implements Serializable
private int retryTimes = 10;
private int port = 9160;
private int superColumns = 1;
private boolean compression = false;
private int progressInterval = 10;
private int keysPerCall = 1000;
@ -258,6 +260,9 @@ public class Session implements Serializable
if (cmd.hasOption("W"))
replicateOnWrite = false;
if (cmd.hasOption("I"))
compression = true;
averageSizeValues = cmd.hasOption("V");
try
@ -412,7 +417,8 @@ public class Session implements Serializable
// column family for standard columns
CfDef standardCfDef = new CfDef("Keyspace1", "Standard1");
standardCfDef.setComparator_type("AsciiType").setDefault_validation_class("BytesType");
System.out.println("Compression = " + compression);
standardCfDef.setComparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
if (indexType != null)
{
ColumnDef standardColumn = new ColumnDef(ByteBufferUtil.bytes("C1"), "BytesType");
@ -422,13 +428,13 @@ public class Session implements Serializable
// column family with super columns
CfDef superCfDef = new CfDef("Keyspace1", "Super1").setColumn_type("Super");
superCfDef.setComparator_type("AsciiType").setSubcomparator_type("AsciiType").setDefault_validation_class("BytesType");
superCfDef.setComparator_type("AsciiType").setSubcomparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
// column family for standard counters
CfDef counterCfDef = new CfDef("Keyspace1", "Counter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite);
CfDef counterCfDef = new CfDef("Keyspace1", "Counter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setCompression(compression);
// column family with counter super columns
CfDef counterSuperCfDef = new CfDef("Keyspace1", "SuperCounter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setColumn_type("Super");
CfDef counterSuperCfDef = new CfDef("Keyspace1", "SuperCounter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setColumn_type("Super").setCompression(compression);
keyspace.setName("Keyspace1");
keyspace.setStrategy_class(replicationStrategy);