diff --git a/NOTICE.txt b/NOTICE.txt index eb58716562..6c73c2eeba 100644 --- a/NOTICE.txt +++ b/NOTICE.txt @@ -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/) diff --git a/build.xml b/build.xml index bd00187caa..1583987663 100644 --- a/build.xml +++ b/build.xml @@ -988,6 +988,14 @@ url=${svn.entry.url}?pathrev=${svn.entry.commit.revision} + + + + + + + + compaction_strategy_options, 31: optional i32 row_cache_keys_to_save, + 32: optional bool compression, } /* describes a keyspace. */ diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java index 717370c646..d8705dcdf3 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java @@ -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); 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 9ab9aa2203..3ca584c095 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java @@ -72,6 +72,7 @@ public class CfDef implements org.apache.thrift.TBase, 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, jav public String compaction_strategy; public Map 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, 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 byName = new HashMap(); @@ -198,6 +201,8 @@ public class CfDef implements org.apache.thrift.TBase, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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, 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(); } 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 6bac7b01e3..b005c0f865 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java @@ -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"; } diff --git a/lib/licenses/snappy-java-1.0.3.txt b/lib/licenses/snappy-java-1.0.3.txt new file mode 100644 index 0000000000..d5c4984908 --- /dev/null +++ b/lib/licenses/snappy-java-1.0.3.txt @@ -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. diff --git a/lib/snappy-java-1.0.3.jar b/lib/snappy-java-1.0.3.jar new file mode 100644 index 0000000000..d33b0b1cfa Binary files /dev/null and b/lib/snappy-java-1.0.3.jar differ diff --git a/src/avro/internode.genavro b/src/avro/internode.genavro index e9db99b9af..aa6fccc673 100644 --- a/src/avro/internode.genavro +++ b/src/avro/internode.genavro @@ -67,6 +67,7 @@ protocol InterNode { union { null, bytes } key_alias = null; union { null, string } compaction_strategy = null; union { null, map } compaction_strategy_options = null; + boolean compression = false; } @aliases(["org.apache.cassandra.config.avro.KsDef"]) diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java index c9ace1349b..873e80ad16 100644 --- a/src/java/org/apache/cassandra/cli/CliClient.java +++ b/src/java/org/apache/cassandra/cli/CliClient.java @@ -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) diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java index b6f0a9aa5c..8a2a8c04cf 100644 --- a/src/java/org/apache/cassandra/config/CFMetaData.java +++ b/src/java/org/apache/cassandra/config/CFMetaData.java @@ -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 column_metadata; public Class 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 prop) {column_metadata = prop; return this;} public CFMetaData rowCacheProvider(IRowCacheProvider prop) { rowCacheProvider = prop; return this;} public CFMetaData compactionStrategyClass(Class 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 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.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 toRemove = new HashSet(); @@ -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 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 column_meta = new ArrayList(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 columnMeta = new ArrayList(); 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) diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java index 0cf4929297..ee9264b00e 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java @@ -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 diff --git a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java new file mode 100644 index 0000000000..c3431a93d9 --- /dev/null +++ b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java @@ -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 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); + } +} diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java new file mode 100644 index 0000000000..a262a25638 --- /dev/null +++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java @@ -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; + } + } +} diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java new file mode 100644 index 0000000000..85fe165e49 --- /dev/null +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -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; + } + } +} diff --git a/src/java/org/apache/cassandra/io/sstable/Component.java b/src/java/org/apache/cassandra/io/sstable/Component.java index c527c612ee..1d0cfc87ce 100644 --- a/src/java/org/apache/cassandra/io/sstable/Component.java +++ b/src/java/org/apache/cassandra/io/sstable/Component.java @@ -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); diff --git a/src/java/org/apache/cassandra/io/sstable/SSTable.java b/src/java/org/apache/cassandra/io/sstable/SSTable.java index 84027649de..dfdd535ed9 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTable.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTable.java @@ -64,6 +64,7 @@ public abstract class SSTable protected final Set 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 dataComponents = new HashSet(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 diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 88b95f6fc0..f40f94a264 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -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 sstables) { for (SSTableReader sstable : sstables) diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java index b78e68e39c..dd2b04442c 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java @@ -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 { 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 { try { - this.file = RandomAccessReader.open(new File(sstable.getFilename()), bufferSize); + this.file = sstable.openDataReader(bufferSize, false); } catch (IOException e) { diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java index 7714af3e0a..bf0f708864 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java @@ -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 components(CFMetaData metadata) + { + Set components = new HashSet(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(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(); diff --git a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java new file mode 100644 index 0000000000..144d64f21b --- /dev/null +++ b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java @@ -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 + } +} diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 4f9f3de446..336cce8f13 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -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; diff --git a/src/java/org/apache/cassandra/io/util/SegmentedFile.java b/src/java/org/apache/cassandra/io/util/SegmentedFile.java index 7058db3d67..709e662bdb 100644 --- a/src/java/org/apache/cassandra/io/util/SegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/SegmentedFile.java @@ -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); /** diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java index 4f2c9da437..02f0924c5a 100644 --- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java @@ -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(); } /** diff --git a/src/java/org/apache/cassandra/streaming/FileStreamTask.java b/src/java/org/apache/cassandra/streaming/FileStreamTask.java index 7fe120af17..c6e79a76f1 100644 --- a/src/java/org/apache/cassandra/streaming/FileStreamTask.java +++ b/src/java/org/apache/cassandra/streaming/FileStreamTask.java @@ -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 { diff --git a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java index b45e6d607d..1a7ee95dd6 100644 --- a/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java +++ b/src/java/org/apache/cassandra/streaming/IncomingStreamReader.java @@ -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; diff --git a/src/resources/org/apache/cassandra/cli/CliHelp.yaml b/src/resources/org/apache/cassandra/cli/CliHelp.yaml index 882e469160..c7ed23b0d8 100644 --- a/src/resources/org/apache/cassandra/cli/CliHelp.yaml +++ b/src/resources/org/apache/cassandra/cli/CliHelp.yaml @@ -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' diff --git a/test/unit/org/apache/cassandra/SchemaLoader.java b/test/unit/org/apache/cassandra/SchemaLoader.java index d6bfb680ad..57c2f89d78 100644 --- a/test/unit/org/apache/cassandra/SchemaLoader.java +++ b/test/unit/org/apache/cassandra/SchemaLoader.java @@ -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 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); diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 54523a5320..da199c7a27 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java new file mode 100644 index 0000000000..bc4a8492d7 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java @@ -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(); + } + } +} diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableTest.java index d4590bc525..4836da7694 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableTest.java @@ -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 keys = new ArrayList(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)); diff --git a/tools/stress/src/org/apache/cassandra/stress/Session.java b/tools/stress/src/org/apache/cassandra/stress/Session.java index e0765f8581..1508379b25 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Session.java +++ b/tools/stress/src/org/apache/cassandra/stress/Session.java @@ -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);