HBASE-26316 Per-table or per-CF compression codec setting overrides (#3730)

We get and retain Compressor instances in HFileBlockDefaultEncodingContext,
and could in theory call Compressor#reinit when setting up the context,
to update compression parameters like level and buffer size, but we do
not plumb through the CompoundConfiguration from the Store into the
encoding context. As a consequence we can only update codec parameters
globally in system site conf files.

Fine grained configurability is important for algorithms like ZStandard
(ZSTD), which offers more than 20 compression levels, where at level 1
it is almost as fast as LZ4, and where at higher levels it utilizes
computationally expensive techniques to rival LZMA at compression ratio
but trades off significantly for reduced compresson throughput. The ZSTD
level that should be set for a given column family or table will vary by
use case.

Signed-off-by: Viraj Jasani <[email protected]>

Author: apurtell

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/CanReinit.java

@@ -0,0 +1,31 @@
+/*
+ * 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.io.compress;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.yetus.audience.InterfaceAudience;
+
+/**
+ * This is a marker interface that indicates if a compressor or decompressor
+ * type can support reinitialization via reinit(Configuration conf).
+ */
+@InterfaceAudience.Private
+public interface CanReinit {
+
+  void reinit(Configuration conf);
+
+}

hbase-common/src/main/java/org/apache/hadoop/hbase/io/compress/Compression.java

@@ -22,12 +22,9 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
-import org.apache.hadoop.hbase.io.util.BlockIOUtils;
-import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.io.compress.CodecPool;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.CompressionInputStream;
@@ -521,37 +518,6 @@ public static String[] getSupportedAlgorithms() {
     return ret;
   }
 
-  /**
-   * Decompresses data from the given stream using the configured compression algorithm. It will
-   * throw an exception if the dest buffer does not have enough space to hold the decompressed data.
-   * @param dest the output buffer
-   * @param bufferedBoundedStream a stream to read compressed data from, bounded to the exact amount
-   *          of compressed data
-   * @param uncompressedSize uncompressed data size, header not included
-   * @param compressAlgo compression algorithm used
-   * @throws IOException if any IO error happen
-   */
-  public static void decompress(ByteBuff dest, InputStream bufferedBoundedStream,
-      int uncompressedSize, Compression.Algorithm compressAlgo) throws IOException {
-    if (dest.remaining() < uncompressedSize) {
-      throw new IllegalArgumentException("Output buffer does not have enough space to hold "
-          + uncompressedSize + " decompressed bytes, available: " + dest.remaining());
-    }
-
-    Decompressor decompressor = null;
-    try {
-      decompressor = compressAlgo.getDecompressor();
-      try (InputStream is =
-          compressAlgo.createDecompressionStream(bufferedBoundedStream, decompressor, 0)) {
-        BlockIOUtils.readFullyWithHeapBuffer(is, dest, uncompressedSize);
-      }
-    } finally {
-      if (decompressor != null) {
-        compressAlgo.returnDecompressor(decompressor);
-      }
-    }
-  }
-
   /**
    * Load a codec implementation for an algorithm using the supplied configuration.
    * @param conf the configuration to use

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/AbstractDataBlockEncoder.java

@@ -18,6 +18,8 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.ByteBufferKeyOnlyKeyValue;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.KeyValue;
@@ -29,14 +31,15 @@
 public abstract class AbstractDataBlockEncoder implements DataBlockEncoder {
 
   @Override
-  public HFileBlockEncodingContext newDataBlockEncodingContext(
+  public HFileBlockEncodingContext newDataBlockEncodingContext(Configuration conf,
       DataBlockEncoding encoding, byte[] header, HFileContext meta) {
-    return new HFileBlockDefaultEncodingContext(encoding, header, meta);
+    return new HFileBlockDefaultEncodingContext(conf, encoding, header, meta);
   }
 
   @Override
-  public HFileBlockDecodingContext newDataBlockDecodingContext(HFileContext meta) {
-    return new HFileBlockDefaultDecodingContext(meta);
+  public HFileBlockDecodingContext newDataBlockDecodingContext(Configuration conf,
+      HFileContext meta) {
+    return new HFileBlockDefaultDecodingContext(conf, meta);
   }
 
   protected void postEncoding(HFileBlockEncodingContext encodingCtx)

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/DataBlockEncoder.java

@@ -20,6 +20,8 @@
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellComparator;
 import org.apache.hadoop.hbase.io.hfile.HFileContext;
@@ -91,6 +93,8 @@ ByteBuffer decodeKeyValues(DataInputStream source, HFileBlockDecodingContext dec
   /**
    * Creates a encoder specific encoding context
    *
+   * @param conf
+   *          store configuration
    * @param encoding
    *          encoding strategy used
    * @param headerBytes
@@ -100,18 +104,20 @@ ByteBuffer decodeKeyValues(DataInputStream source, HFileBlockDecodingContext dec
    *          HFile meta data
    * @return a newly created encoding context
    */
-  HFileBlockEncodingContext newDataBlockEncodingContext(
+  HFileBlockEncodingContext newDataBlockEncodingContext(Configuration conf,
       DataBlockEncoding encoding, byte[] headerBytes, HFileContext meta);
 
   /**
    * Creates an encoder specific decoding context, which will prepare the data
    * before actual decoding
    *
+   * @param conf
+   *          store configuration
    * @param meta
    *          HFile meta data        
    * @return a newly created decoding context
    */
-  HFileBlockDecodingContext newDataBlockDecodingContext(HFileContext meta);
+  HFileBlockDecodingContext newDataBlockDecodingContext(Configuration conf, HFileContext meta);
 
   /**
    * An interface which enable to seek while underlying data is encoded.

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/EncodedDataBlock.java

@@ -27,6 +27,7 @@
 import java.util.Iterator;
 import java.util.List;
 import org.apache.commons.lang3.NotImplementedException;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
@@ -57,6 +58,7 @@ public class EncodedDataBlock {
   private HFileContext meta;
 
   private final DataBlockEncoding encoding;
+  private final Configuration conf;
 
   // The is for one situation that there are some cells includes tags and others are not.
   // isTagsLenZero stores if cell tags length is zero before doing encoding since we need
@@ -68,21 +70,23 @@ public class EncodedDataBlock {
 
   /**
    * Create a buffer which will be encoded using dataBlockEncoder.
+   * @param conf store configuration
    * @param dataBlockEncoder Algorithm used for compression.
    * @param encoding encoding type used
-   * @param rawKVs
-   * @param meta
+   * @param rawKVs raw KVs
+   * @param meta hfile context
    */
-  public EncodedDataBlock(DataBlockEncoder dataBlockEncoder, DataBlockEncoding encoding,
-      byte[] rawKVs, HFileContext meta) {
+  public EncodedDataBlock(Configuration conf, DataBlockEncoder dataBlockEncoder,
+      DataBlockEncoding encoding, byte[] rawKVs, HFileContext meta) {
     Preconditions.checkNotNull(encoding,
         "Cannot create encoded data block with null encoder");
     this.dataBlockEncoder = dataBlockEncoder;
     this.encoding = encoding;
-    encodingCtx = dataBlockEncoder.newDataBlockEncodingContext(encoding,
+    encodingCtx = dataBlockEncoder.newDataBlockEncodingContext(conf, encoding,
         HConstants.HFILEBLOCK_DUMMY_HEADER, meta);
     this.rawKVs = rawKVs;
     this.meta = meta;
+    this.conf = conf;
   }
 
   /**
@@ -115,7 +119,7 @@ public Cell next() {
         if (decompressedData == null) {
           try {
             decompressedData = dataBlockEncoder.decodeKeyValues(dis, dataBlockEncoder
-                .newDataBlockDecodingContext(meta));
+                .newDataBlockDecodingContext(conf, meta));
           } catch (IOException e) {
             throw new RuntimeException("Problem with data block encoder, " +
                 "most likely it requested more bytes than are available.", e);

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultDecodingContext.java

@@ -20,8 +20,10 @@
 import java.io.IOException;
 import java.io.InputStream;
 import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.io.ByteBuffInputStream;
 import org.apache.hadoop.hbase.io.TagCompressionContext;
+import org.apache.hadoop.hbase.io.compress.CanReinit;
 import org.apache.hadoop.hbase.io.compress.Compression;
 import org.apache.hadoop.hbase.io.crypto.Cipher;
 import org.apache.hadoop.hbase.io.crypto.Decryptor;
@@ -30,6 +32,7 @@
 import org.apache.hadoop.hbase.io.util.BlockIOUtils;
 import org.apache.hadoop.hbase.nio.ByteBuff;
 import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.yetus.audience.InterfaceAudience;
 
 /**
@@ -41,10 +44,12 @@
  */
 @InterfaceAudience.Private
 public class HFileBlockDefaultDecodingContext implements HFileBlockDecodingContext {
+  private final Configuration conf;
   private final HFileContext fileContext;
   private TagCompressionContext tagCompressionContext;
 
-  public HFileBlockDefaultDecodingContext(HFileContext fileContext) {
+  public HFileBlockDefaultDecodingContext(Configuration conf, HFileContext fileContext) {
+    this.conf = conf;
     this.fileContext = fileContext;
   }
 
@@ -87,8 +92,24 @@ public void prepareDecoding(int onDiskSizeWithoutHeader, int uncompressedSizeWit
 
       Compression.Algorithm compression = fileContext.getCompression();
       if (compression != Compression.Algorithm.NONE) {
-        Compression.decompress(blockBufferWithoutHeader, dataInputStream,
-          uncompressedSizeWithoutHeader, compression);
+        Decompressor decompressor = null;
+        try {
+          decompressor = compression.getDecompressor();
+          // Some algorithms don't return decompressors and accept null as a valid parameter for
+          // same when creating decompression streams. We can ignore these cases wrt reinit.
+          if (decompressor instanceof CanReinit) {
+            ((CanReinit)decompressor).reinit(conf);
+          }
+          try (InputStream is =
+              compression.createDecompressionStream(dataInputStream, decompressor, 0)) {
+            BlockIOUtils.readFullyWithHeapBuffer(is, blockBufferWithoutHeader,
+              uncompressedSizeWithoutHeader);
+          }
+        } finally {
+          if (decompressor != null) {
+            compression.returnDecompressor(decompressor);
+          }
+        }
       } else {
         BlockIOUtils.readFullyWithHeapBuffer(dataInputStream, blockBufferWithoutHeader,
           onDiskSizeWithoutHeader);

hbase-common/src/main/java/org/apache/hadoop/hbase/io/encoding/HFileBlockDefaultEncodingContext.java

@@ -22,6 +22,8 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.security.SecureRandom;
+
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
 import org.apache.hadoop.hbase.io.TagCompressionContext;
 import org.apache.hadoop.hbase.io.compress.Compression;
@@ -72,18 +74,26 @@ public class HFileBlockDefaultEncodingContext implements HFileBlockEncodingConte
   private EncodingState encoderState;
 
   /**
+   * @param conf configuraton
    * @param encoding encoding used
    * @param headerBytes dummy header bytes
    * @param fileContext HFile meta data
    */
-  public HFileBlockDefaultEncodingContext(DataBlockEncoding encoding, byte[] headerBytes,
-      HFileContext fileContext) {
+  public HFileBlockDefaultEncodingContext(Configuration conf, DataBlockEncoding encoding,
+      byte[] headerBytes, HFileContext fileContext) {
     this.encodingAlgo = encoding;
     this.fileContext = fileContext;
     Compression.Algorithm compressionAlgorithm =
         fileContext.getCompression() == null ? NONE : fileContext.getCompression();
     if (compressionAlgorithm != NONE) {
-      compressor = compressionAlgorithm.getCompressor();
+      if (compressor == null) {
+        compressor = compressionAlgorithm.getCompressor();
+        // Some algorithms don't return compressors and accept null as a valid parameter for
+        // same when creating compression streams. We can ignore these cases wrt reinit.
+        if (compressor != null) {
+          compressor.reinit(conf);
+        }
+      }
       compressedByteStream = new ByteArrayOutputStream();
       try {
         compressionStream =

hbase-compression/hbase-compression-aircompressor/src/main/java/org/apache/hadoop/hbase/io/compress/aircompressor/HadoopCompressor.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.io.compress.CanReinit;
 import org.apache.hadoop.hbase.io.compress.CompressionUtil;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -30,7 +31,7 @@
  */
 @InterfaceAudience.Private
 public abstract class HadoopCompressor<T extends Compressor>
-    implements org.apache.hadoop.io.compress.Compressor {
+    implements CanReinit, org.apache.hadoop.io.compress.Compressor {
 
   protected static final Logger LOG = LoggerFactory.getLogger(HadoopCompressor.class);
   protected T compressor;

hbase-compression/hbase-compression-aircompressor/src/test/java/org/apache/hadoop/hbase/io/compress/aircompressor/TestHFileCompressionLz4.java

@@ -18,7 +18,9 @@
 package org.apache.hadoop.hbase.io.compress.aircompressor;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
 import org.apache.hadoop.hbase.io.compress.Compression;
 import org.apache.hadoop.hbase.io.compress.HFileTestBase;
 import org.apache.hadoop.hbase.testclassification.IOTests;
@@ -35,17 +37,21 @@ public class TestHFileCompressionLz4 extends HFileTestBase {
   public static final HBaseClassTestRule CLASS_RULE =
       HBaseClassTestRule.forClass(TestHFileCompressionLz4.class);
 
+  private static Configuration conf;
+
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
-    Configuration conf = TEST_UTIL.getConfiguration();
+    conf = TEST_UTIL.getConfiguration();
     conf.set(Compression.LZ4_CODEC_CLASS_KEY, Lz4Codec.class.getCanonicalName());
     Compression.Algorithm.LZ4.reload(conf);
     HFileTestBase.setUpBeforeClass();
   }
 
   @Test
   public void test() throws Exception {
-    doTest(Compression.Algorithm.LZ4);
+    Path path = new Path(TEST_UTIL.getDataTestDir(),
+      HBaseTestingUtil.getRandomUUID().toString() + ".hfile");
+    doTest(conf, path, Compression.Algorithm.LZ4);
   }
 
 }

hbase-compression/hbase-compression-aircompressor/src/test/java/org/apache/hadoop/hbase/io/compress/aircompressor/TestHFileCompressionLzo.java

@@ -18,7 +18,9 @@
 package org.apache.hadoop.hbase.io.compress.aircompressor;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
 import org.apache.hadoop.hbase.io.compress.Compression;
 import org.apache.hadoop.hbase.io.compress.HFileTestBase;
 import org.apache.hadoop.hbase.testclassification.IOTests;
@@ -35,17 +37,21 @@ public class TestHFileCompressionLzo extends HFileTestBase {
   public static final HBaseClassTestRule CLASS_RULE =
       HBaseClassTestRule.forClass(TestHFileCompressionLzo.class);
 
+  private static Configuration conf;
+
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
-    Configuration conf = TEST_UTIL.getConfiguration();
+    conf = TEST_UTIL.getConfiguration();
     conf.set(Compression.LZO_CODEC_CLASS_KEY, LzoCodec.class.getCanonicalName());
     Compression.Algorithm.LZO.reload(conf);
     HFileTestBase.setUpBeforeClass();
   }
 
   @Test
   public void test() throws Exception {
-    doTest(Compression.Algorithm.LZO);
+    Path path = new Path(TEST_UTIL.getDataTestDir(),
+      HBaseTestingUtil.getRandomUUID().toString() + ".hfile");
+    doTest(conf, path, Compression.Algorithm.LZO);
   }
 
 }