apurtell commented on a change in pull request #3244:
URL: https://github.com/apache/hbase/pull/3244#discussion_r634841259
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
##########
@@ -18,37 +18,117 @@
package org.apache.hadoop.hbase.regionserver.wal;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.EnumMap;
import java.util.Map;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
-import org.apache.yetus.audience.InterfaceAudience;
+import org.apache.hadoop.hbase.io.DelegatingInputStream;
import org.apache.hadoop.hbase.io.TagCompressionContext;
+import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.io.util.Dictionary;
+import org.apache.yetus.audience.InterfaceAudience;
/**
* Context that holds the various dictionaries for compression in WAL.
*/
@InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC,
HBaseInterfaceAudience.PHOENIX})
public class CompressionContext {
- static final String ENABLE_WAL_TAGS_COMPRESSION =
- "hbase.regionserver.wal.tags.enablecompression";
+ public static final String ENABLE_WAL_TAGS_COMPRESSION =
+ "hbase.regionserver.wal.tags.enablecompression";
+
+ public static final String ENABLE_WAL_VALUE_COMPRESSION =
+ "hbase.regionserver.wal.value.enablecompression";
+
+ public static final String WAL_VALUE_COMPRESSION_TYPE =
+ "hbase.regionserver.wal.value.compression.type";
public enum DictionaryIndex {
REGION, TABLE, FAMILY, QUALIFIER, ROW
}
+ /**
+ * Encapsulates the compression algorithm and its streams that we will use
for value
+ * compression in this WAL.
+ */
+ static class ValueCompressor {
+
+ static final int IO_BUFFER_SIZE = 4096;
+
+ private final Compression.Algorithm algorithm;
+ private DelegatingInputStream lowerIn;
+ private ByteArrayOutputStream lowerOut;
+ private InputStream compressedIn;
+ private OutputStream compressedOut;
+
+ public ValueCompressor(Compression.Algorithm algorithm) throws IOException
{
+ this.algorithm = algorithm;
+ }
+
+ public Compression.Algorithm getAlgorithm() {
+ return algorithm;
+ }
+
+ public byte[] compress(byte[] valueArray, int valueOffset, int valueLength)
+ throws IOException {
+ // We have to create the output streams here the first time around.
+ if (compressedOut == null) {
+ lowerOut = new ByteArrayOutputStream();
+ compressedOut = algorithm.createCompressionStream(lowerOut,
algorithm.getCompressor(),
+ IO_BUFFER_SIZE);
+ } else {
+ lowerOut.reset();
+ }
+ compressedOut.write(valueArray, valueOffset, valueLength);
+ compressedOut.flush();
+ return lowerOut.toByteArray();
+ }
+
+ public int decompress(InputStream in, int inLength, byte[] outArray, int
outOffset,
+ int outLength) throws IOException {
+ // Read all of the compressed bytes into a buffer.
+ byte[] inBuffer = new byte[inLength];
+ IOUtils.readFully(in, inBuffer);
+ // We have to create the input streams here the first time around.
+ if (compressedIn == null) {
+ lowerIn = new DelegatingInputStream(new
ByteArrayInputStream(inBuffer));
+ compressedIn = algorithm.createDecompressionStream(lowerIn,
algorithm.getDecompressor(),
+ IO_BUFFER_SIZE);
+ } else {
+ lowerIn.setDelegate(new ByteArrayInputStream(inBuffer));
+ }
+ return compressedIn.read(outArray, outOffset, outLength);
Review comment:
Actual bytes read are returned. It would be up to the caller to handle.
In practice, all bytes are read here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
