We were fully flushing the deflater at the end of every value

apurtell · apurtell · commit ca442334f3a8 · 2021-05-09T15:33:13.000-07:00
(FULL_FLUSH). This was very conservative, allowing each value to be
decompressed individually (so is resilient to corruptions), but
seriously impacts compression ratio. Meanwhile our custom
dictionary scheme accumulates strings over the whole file, so we
were being conservative in value compression in a way our custom
scheme defeats anyway. We can instead also let the Deflater build
its dictionary over all values in the WAL, using SYNC_FLUSH instead.

Add more unit tests.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/CompressionContext.java
@@ -23,6 +23,7 @@
 import java.util.EnumMap;
 import java.util.Map;
 import java.util.zip.Deflater;
+import java.util.zip.Inflater;
 
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -35,21 +36,45 @@
 @InterfaceAudience.LimitedPrivate({HBaseInterfaceAudience.COPROC, HBaseInterfaceAudience.PHOENIX})
 public class CompressionContext {
 
-  static final String ENABLE_WAL_TAGS_COMPRESSION =
+  public static final String ENABLE_WAL_TAGS_COMPRESSION =
       "hbase.regionserver.wal.tags.enablecompression";
 
-  static final String ENABLE_WAL_VALUE_COMPRESSION =
+  public static final String ENABLE_WAL_VALUE_COMPRESSION =
       "hbase.regionserver.wal.value.enablecompression";
 
   public enum DictionaryIndex {
     REGION, TABLE, FAMILY, QUALIFIER, ROW
   }
 
+  static class ValueCompressor {
+    final Deflater deflater;
+    final Inflater inflater;
+
+    public ValueCompressor() {
+      deflater = new Deflater();
+      deflater.setLevel(Deflater.BEST_SPEED);
+      inflater = new Inflater();
+    }
+
+    public Deflater getDeflater() {
+      return deflater;
+    }
+
+    public Inflater getInflater() {
+      return inflater;
+    }
+
+    public void clear() {
+      deflater.reset();
+      inflater.reset();
+    }
+  };
+
   private final Map<DictionaryIndex, Dictionary> dictionaries =
       new EnumMap<>(DictionaryIndex.class);
   // Context used for compressing tags
   TagCompressionContext tagCompressionContext = null;
-  Deflater valueCompressor = null;
+  ValueCompressor valueCompressor = null;
 
   public CompressionContext(Class<? extends Dictionary> dictType, boolean recoveredEdits,
       boolean hasTagCompression, boolean hasValueCompression)
@@ -77,9 +102,7 @@ public CompressionContext(Class<? extends Dictionary> dictType, boolean recovere
       tagCompressionContext = new TagCompressionContext(dictType, Short.MAX_VALUE);
     }
     if (hasValueCompression) {
-      valueCompressor = new Deflater();
-      // Optimize for encoding speed
-      valueCompressor.setLevel(Deflater.BEST_SPEED);
+      valueCompressor = new ValueCompressor();
     }
   }
 
@@ -94,7 +117,7 @@ public Dictionary getDictionary(Enum dictIndex) {
     return dictionaries.get(dictIndex);
   }
 
-  public Deflater getValueCompressor() {
+  public ValueCompressor getValueCompressor() {
     return valueCompressor;
   }
 
@@ -106,7 +129,7 @@ void clear() {
       tagCompressionContext.clear();
     }
     if (valueCompressor != null) {
-      valueCompressor.reset();
+      valueCompressor.clear();
     }
   }
 
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/WALCellCodec.java
@@ -42,11 +42,12 @@
 import org.apache.hadoop.hbase.io.util.Dictionary;
 import org.apache.hadoop.hbase.io.util.StreamUtils;
 import org.apache.hadoop.hbase.nio.ByteBuff;
+import org.apache.hadoop.hbase.regionserver.wal.CompressionContext.ValueCompressor;
 import org.apache.hadoop.hbase.util.ByteBufferUtils;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.ReflectionUtils;
 import org.apache.hadoop.io.IOUtils;
-import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
+
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 
@@ -272,18 +273,18 @@ public void write(Cell cell) throws IOException {
     }
 
     public static void writeCompressedValue(OutputStream out, byte[] valueArray, int offset,
-        int vlength, Deflater deflater) throws IOException {
+        int vlength, ValueCompressor valueCompressor) throws IOException {
       byte[] buffer = new byte[4096];
       ByteArrayOutputStream baos = new ByteArrayOutputStream();
-      deflater.reset();
+      Deflater deflater = valueCompressor.getDeflater();
       deflater.setInput(valueArray, offset, vlength);
       boolean finished = false;
       do {
         int bytesOut = deflater.deflate(buffer);
         if (bytesOut > 0) {
           baos.write(buffer, 0, bytesOut);
         } else {
-          bytesOut = deflater.deflate(buffer, 0, buffer.length, Deflater.FULL_FLUSH);
+          bytesOut = deflater.deflate(buffer, 0, buffer.length, Deflater.SYNC_FLUSH);
           baos.write(buffer, 0, bytesOut);
           finished = true;
         }
@@ -351,7 +352,7 @@ protected Cell parseCell() throws IOException {
         type = (byte)(type & 0x7f);
         pos = Bytes.putByte(backingArray, pos, type);
         int valLen = typeValLen - 1;
-        readCompressedValue(in, backingArray, pos, valLen);
+        readCompressedValue(in, backingArray, pos, valLen, compression.getValueCompressor());
         pos += valLen;
       } else {
         pos = Bytes.putByte(backingArray, pos, type);
@@ -400,17 +401,16 @@ private static void checkLength(int len, int max) throws IOException {
     }
 
     public static void readCompressedValue(InputStream in, byte[] outArray, int outOffset,
-        int expectedLength) throws IOException {
+        int expectedLength, ValueCompressor valueCompressor) throws IOException {
       int compressedLength = StreamUtils.readRawVarint32(in);
       byte[] buffer = new byte[compressedLength];
       IOUtils.readFully(in, buffer, 0, compressedLength);
-      Inflater inflater = new Inflater();
+      Inflater inflater = valueCompressor.getInflater();
       inflater.setInput(buffer);
       int remaining = expectedLength;
       do {
         try {
           int inflatedBytes = inflater.inflate(outArray, outOffset, remaining);
-          Preconditions.checkState(inflatedBytes > 0, "Inflater state error");
           outOffset += inflatedBytes;
           remaining -= inflatedBytes;
         } catch (DataFormatException e) {
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALCellCodecWithCompression.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALCellCodecWithCompression.java
@@ -26,6 +26,7 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ArrayBackedTag;
 import org.apache.hadoop.hbase.ByteBufferKeyValue;
@@ -74,14 +75,20 @@ public void testValueCompressionEnabled() throws Exception {
   @Test
   public void testValueCompression() throws Exception {
     final byte[] row_1 = Bytes.toBytes("row_1");
-    final byte[] value_1 = new byte[512];
+    final byte[] value_1 = new byte[WALCellCodec.VALUE_COMPRESS_THRESHOLD];
     Bytes.zero(value_1);
     final byte[] row_2 = Bytes.toBytes("row_2");
     final byte[] value_2 = new byte[Bytes.SIZEOF_LONG];
     Bytes.random(value_2);
     final byte[] row_3 = Bytes.toBytes("row_3");
-    final byte[] value_3 = new byte[1024];
+    final byte[] value_3 = new byte[WALCellCodec.VALUE_COMPRESS_THRESHOLD];
     Bytes.random(value_3);
+    final byte[] row_4 = Bytes.toBytes("row_4");
+    final byte[] value_4 = new byte[WALCellCodec.VALUE_COMPRESS_THRESHOLD * 4];
+    fillBytes(value_4, Bytes.toBytes("DEADBEEF"));
+    final byte[] row_5 = Bytes.toBytes("row_5");
+    final byte[] value_5 = new byte[WALCellCodec.VALUE_COMPRESS_THRESHOLD * 2];
+    fillBytes(value_5, Bytes.toBytes("CAFEBABE"));
 
     Configuration conf = new Configuration(false);
     WALCellCodec codec = new WALCellCodec(conf, new CompressionContext(LRUDictionary.class,
@@ -91,6 +98,8 @@ public void testValueCompression() throws Exception {
     encoder.write(createKV(row_1, value_1, 0));
     encoder.write(createKV(row_2, value_2, 0));
     encoder.write(createKV(row_3, value_3, 0));
+    encoder.write(createKV(row_4, value_4, 0));
+    encoder.write(createKV(row_5, value_5, 0));
     encoder.flush();
 
     try (InputStream is = new ByteArrayInputStream(bos.toByteArray())) {
@@ -113,55 +122,57 @@ public void testValueCompression() throws Exception {
         kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
       assertTrue(Bytes.equals(value_3, 0, value_3.length,
         kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
+      decoder.advance();
+      kv = (KeyValue) decoder.current();
+      assertTrue(Bytes.equals(row_4, 0, row_4.length,
+        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
+      assertTrue(Bytes.equals(value_4, 0, value_4.length,
+        kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
+      decoder.advance();
+      kv = (KeyValue) decoder.current();
+      assertTrue(Bytes.equals(row_5, 0, row_5.length,
+        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
+      assertTrue(Bytes.equals(value_5, 0, value_5.length,
+        kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
+    }
+  }
+
+  static void fillBytes(byte[] buffer, byte[] fill) {
+    int offset = 0;
+    int remaining = buffer.length;
+    while (remaining > 0) {
+      int len = remaining < fill.length ? remaining : fill.length;
+      System.arraycopy(fill, 0, buffer, offset, len);
+      offset += len;
+      remaining -= len;
     }
   }
 
   @Test
   public void testValueCompressionCompatibility() throws Exception {
-    final byte[] row_1 = Bytes.toBytes("row_1");
-    final byte[] value_1 = new byte[512];
-    Bytes.zero(value_1);
-    final byte[] row_2 = Bytes.toBytes("row_2");
-    final byte[] value_2 = new byte[Bytes.SIZEOF_LONG];
-    Bytes.random(value_2);
-    final byte[] row_3 = Bytes.toBytes("row_3");
-    final byte[] value_3 = new byte[1024];
-    Bytes.random(value_3);
-
+    final byte[] key = Bytes.toBytes("myRow");
+    final byte[] value = Bytes.toBytes("myValue");
     Configuration conf = new Configuration(false);
     WALCellCodec outCodec = new WALCellCodec(conf, new CompressionContext(LRUDictionary.class,
       false, false, false));
     ByteArrayOutputStream bos = new ByteArrayOutputStream();
     Encoder encoder = outCodec.getEncoder(bos);
-    encoder.write(createKV(row_1, value_1, 0));
-    encoder.write(createKV(row_2, value_2, 0));
-    encoder.write(createKV(row_3, value_3, 0));
+    for (int i = 0; i < 10; i++) {
+      encoder.write(createOffheapKV(key, value, 0));
+    }
     encoder.flush();
-
-    byte[] encodedCells = bos.toByteArray();
-
     WALCellCodec inCodec = new WALCellCodec(conf, new CompressionContext(LRUDictionary.class,
       false, false, true));
-    try (InputStream is = new ByteArrayInputStream(encodedCells)) {
+    try (InputStream is = new ByteArrayInputStream(bos.toByteArray())) {
       Decoder decoder = inCodec.getDecoder(is);
-      decoder.advance();
-      KeyValue kv = (KeyValue) decoder.current();
-      assertTrue(Bytes.equals(row_1, 0, row_1.length,
-        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
-      assertTrue(Bytes.equals(value_1, 0, value_1.length,
-        kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
-      decoder.advance();
-      kv = (KeyValue) decoder.current();
-      assertTrue(Bytes.equals(row_2, 0, row_2.length,
-        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
-      assertTrue(Bytes.equals(value_2, 0, value_2.length,
-        kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
-      decoder.advance();
-      kv = (KeyValue) decoder.current();
-      assertTrue(Bytes.equals(row_3, 0, row_3.length,
-        kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
-      assertTrue(Bytes.equals(value_3, 0, value_3.length,
-        kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
+      for (int i = 0 ; i < 10; i++) {
+        decoder.advance();
+        KeyValue kv = (KeyValue) decoder.current();
+        assertTrue(Bytes.equals(key, 0, key.length,
+          kv.getRowArray(), kv.getRowOffset(), kv.getRowLength()));
+        assertTrue(Bytes.equals(value, 0, value.length,
+          kv.getValueArray(), kv.getValueOffset(), kv.getValueLength()));
+      }
     }
   }
 
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayValueCompression.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestWALReplayValueCompression.java
@@ -0,0 +1,46 @@
+/**
+ * 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.hadoop.hbase.regionserver.wal;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.experimental.categories.Category;
+
+/**
+ * Enables compression and runs the TestWALReplay tests.
+ */
+@Category({ RegionServerTests.class, LargeTests.class })
+public class TestWALReplayValueCompression extends TestWALReplay {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+      HBaseClassTestRule.forClass(TestWALReplayValueCompression.class);
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    Configuration conf = AbstractTestWALReplay.TEST_UTIL.getConfiguration();
+    conf.setBoolean(HConstants.ENABLE_WAL_COMPRESSION, true);
+    conf.setBoolean(CompressionContext.ENABLE_WAL_VALUE_COMPRESSION, true);
+    TestWALReplay.setUpBeforeClass();
+  }
+}
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplitValueCompression.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestWALSplitValueCompression.java
@@ -0,0 +1,44 @@
+/**
+ * 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.hadoop.hbase.wal;
+
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.regionserver.wal.CompressionContext;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.experimental.categories.Category;
+
+@Category({RegionServerTests.class, LargeTests.class})
+public class TestWALSplitValueCompression extends TestWALSplit {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+      HBaseClassTestRule.forClass(TestWALSplitValueCompression.class);
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    TestWALSplit.setUpBeforeClass();
+    TEST_UTIL.getConfiguration()
+      .setBoolean(HConstants.ENABLE_WAL_COMPRESSION, true);
+    TEST_UTIL.getConfiguration()
+      .setBoolean(CompressionContext.ENABLE_WAL_VALUE_COMPRESSION, true);
+  }
+}
