You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by kt...@apache.org on 2012/03/21 20:40:38 UTC
svn commit: r1303526 - in
/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile:
Compression.java TFile.java
Author: kturner
Date: Wed Mar 21 19:40:38 2012
New Revision: 1303526
URL: http://svn.apache.org/viewvc?rev=1303526&view=rev
Log:
ACCUMULO-139 added snappy support to bcfile
Modified:
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/TFile.java
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java?rev=1303526&r1=1303525&r2=1303526&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/Compression.java Wed Mar 21 19:40:38 2012
@@ -208,8 +208,68 @@ final class Compression {
public boolean isSupported() {
return true;
}
- };
+ },
+ SNAPPY(TFile.COMPRESSION_SNAPPY) {
+ // Use base type to avoid compile-time dependencies.
+ private transient CompressionCodec snappyCodec = null;
+ private transient boolean checked = false;
+ private static final String defaultClazz = "org.apache.hadoop.io.compress.SnappyCodec";
+
+ public CompressionCodec getCodec() throws IOException {
+ if (!isSupported()) {
+ throw new IOException("SNAPPY codec class not specified. Did you forget to set property " + CONF_SNAPPY_CLASS + "?");
+ }
+ return snappyCodec;
+ }
+
+ @Override
+ public synchronized OutputStream createCompressionStream(OutputStream downStream, Compressor compressor, int downStreamBufferSize) throws IOException {
+
+ if (!isSupported()) {
+ throw new IOException("SNAPPY codec class not specified. Did you forget to set property " + CONF_SNAPPY_CLASS + "?");
+ }
+ OutputStream bos1 = null;
+ if (downStreamBufferSize > 0) {
+ bos1 = new BufferedOutputStream(downStream, downStreamBufferSize);
+ } else {
+ bos1 = downStream;
+ }
+ conf.setInt("io.compression.codec.snappy.buffersize", 64 * 1024);
+ CompressionOutputStream cos = snappyCodec.createOutputStream(bos1, compressor);
+ BufferedOutputStream bos2 = new BufferedOutputStream(new FinishOnFlushCompressionStream(cos), DATA_OBUF_SIZE);
+ return bos2;
+ }
+
+ @Override
+ public synchronized InputStream createDecompressionStream(InputStream downStream, Decompressor decompressor, int downStreamBufferSize) throws IOException {
+ if (!isSupported()) {
+ throw new IOException("SNAPPY codec class not specified. Did you forget to set property " + CONF_SNAPPY_CLASS + "?");
+ }
+ if (downStreamBufferSize > 0) {
+ conf.setInt("io.file.buffer.size", downStreamBufferSize);
+ }
+ CompressionInputStream cis = snappyCodec.createInputStream(downStream, decompressor);
+ BufferedInputStream bis2 = new BufferedInputStream(cis, DATA_IBUF_SIZE);
+ return bis2;
+ }
+
+ @Override
+ public synchronized boolean isSupported() {
+ if (!checked) {
+ checked = true;
+ String extClazz = (conf.get(CONF_SNAPPY_CLASS) == null ? System.getProperty(CONF_SNAPPY_CLASS) : null);
+ String clazz = (extClazz != null) ? extClazz : defaultClazz;
+ try {
+ LOG.info("Trying to load snappy codec class: " + clazz);
+ snappyCodec = (CompressionCodec) ReflectionUtils.newInstance(Class.forName(clazz), conf);
+ } catch (ClassNotFoundException e) {
+ // that is okay
+ }
+ }
+ return snappyCodec != null;
+ }
+ };
// We require that all compression related settings are configured
// statically in the Configuration object.
protected static final Configuration conf = new Configuration();
@@ -219,6 +279,7 @@ final class Compression {
// data output buffer size to absorb small writes from application.
private static final int DATA_OBUF_SIZE = 4 * 1024;
public static final String CONF_LZO_CLASS = "io.compression.codec.lzo.class";
+ public static final String CONF_SNAPPY_CLASS = "io.compression.codec.snappy.class";
Algorithm(String name) {
this.compressName = name;
Modified: incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/TFile.java
URL: http://svn.apache.org/viewvc/incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/TFile.java?rev=1303526&r1=1303525&r2=1303526&view=diff
==============================================================================
--- incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/TFile.java (original)
+++ incubator/accumulo/branches/1.4/src/core/src/main/java/org/apache/accumulo/core/file/rfile/bcfile/TFile.java Wed Mar 21 19:40:38 2012
@@ -124,6 +124,9 @@ public class TFile {
private static final int MAX_KEY_SIZE = 64 * 1024; // 64KB
static final Version API_VERSION = new Version((short) 1, (short) 0);
+ /** snappy codec **/
+ public static final String COMPRESSION_SNAPPY = "snappy";
+
/** compression: gzip */
public static final String COMPRESSION_GZ = "gz";
/** compression: lzo */
@@ -159,6 +162,7 @@ public class TFile {
* <li>"none" - No compression.
* <li>"lzo" - LZO compression.
* <li>"gz" - GZIP compression.
+ * <li>"snappy" - Snappy compression
* </ul>
*/
public static String[] getSupportedCompressionAlgorithms() {