Add hardware-accelerated codecs for DEFLATE and LZ4 (#122)

* Add QAT accelerated compression. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Use own classes for QAT codec. Apply SpotlessJavaCheck. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Declare fields final, unless required not to. Throw a valid type of exception. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Use assumeThat in the Qat test classes. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Add more QAT availability check in QatCodecTests. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Make LZ4 the default algorithm for QAT. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Make 'auto' the default execution mode for QAT. Also, minor clean up work. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Revert compression level for ZSTD to 3. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Replace QatLz4/DeflateCompressionMode classes with QatCompressionMode. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Fix a MultiCodecMergeIT test fail. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> * Remove hard-coded values for default compression level. Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> --------- Signed-off-by: Mulugeta Mammo <mulugeta.mammo@intel.com> Signed-off-by: mulugetam <mulugeta.mammo@intel.com> Co-authored-by: Mulugeta Mammo <cppx86@gmail.com>
opensearch-project · May 30, 2024 · c8b0d80 · c8b0d80
1 parent f65004a
commit c8b0d80
Show file tree

Hide file tree

Showing 34 changed files with 1,643 additions and 71 deletions.
diff --git a/build.gradle b/build.gradle
@@ -93,6 +93,7 @@ opensearchplugin {
 
 dependencies {
   api "com.github.luben:zstd-jni:1.5.5-5"
+  api "com.intel.qat:qat-java:1.1.1"
 }
 
 allprojects {

diff --git a/licenses/qat-java-1.1.1.jar.sha1 b/licenses/qat-java-1.1.1.jar.sha1
@@ -0,0 +1 @@
+3333601cdedf6a711d445118d5bc44ec6a9c65f9
diff --git a/licenses/qat-java-LICENSE.txt b/licenses/qat-java-LICENSE.txt
@@ -0,0 +1,36 @@
+-----------------------------------------------------------------------------
+** Beginning of "BSD License" text. **
+
+Qat-Java: Qat-Java is a compression library that uses Intel® QAT to accelerate
+compression and decompression.
+
+Copyright(c) 2007-2023 Intel Corporation. All rights reserved.
+All rights reserved.
+
+BSD License
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions
+are met:
+
+  * Redistributions of source code must retain the above copyright
+    notice, this list of conditions and the following disclaimer.
+  * Redistributions in binary form must reproduce the above copyright
+    notice, this list of conditions and the following disclaimer in
+    the documentation and/or other materials provided with the
+    distribution.
+  * Neither the name of Intel Corporation nor the names of its
+    contributors may be used to endorse or promote products derived
+    from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/licenses/qat-java-NOTICE.txt b/licenses/qat-java-NOTICE.txt
@@ -0,0 +1 @@
+Qat-Java is a compression library that uses Intel® QAT to accelerate compression and decompression.
diff --git a/src/integrationTest/java/org/opensearch/index/codec/rest/CreateIndexWithCodecIT.java b/src/integrationTest/java/org/opensearch/index/codec/rest/CreateIndexWithCodecIT.java
@@ -25,6 +25,7 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.core.common.Strings;
+import org.opensearch.index.codec.customcodecs.QatZipperFactory;
 import org.opensearch.test.rest.OpenSearchRestTestCase;
 
 import javax.net.ssl.SSLEngine;
@@ -37,8 +38,12 @@
 
 import static org.opensearch.client.RestClientBuilder.DEFAULT_MAX_CONN_PER_ROUTE;
 import static org.opensearch.client.RestClientBuilder.DEFAULT_MAX_CONN_TOTAL;
+import static org.opensearch.index.codec.customcodecs.CustomCodecService.QAT_DEFLATE_CODEC;
+import static org.opensearch.index.codec.customcodecs.CustomCodecService.QAT_LZ4_CODEC;
 import static org.opensearch.index.codec.customcodecs.CustomCodecService.ZSTD_CODEC;
 import static org.opensearch.index.codec.customcodecs.CustomCodecService.ZSTD_NO_DICT_CODEC;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assume.assumeThat;
 
 public class CreateIndexWithCodecIT extends OpenSearchRestTestCase {
     public void testCreateIndexWithZstdCodec() throws IOException {
@@ -62,6 +67,29 @@ public void testCreateIndexWithZstdCodec() throws IOException {
         }
     }
 
+    public void testCreateIndexWithQatCodec() throws IOException {
+        assumeThat("Qat library is available", QatZipperFactory.isQatAvailable(), is(true));
+
+        final String index = "custom-codecs-test-index";
+
+        // creating index
+        createIndex(
+            index,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put("index.codec", randomFrom(QAT_DEFLATE_CODEC, QAT_LZ4_CODEC))
+                .put("index.codec.compression_level", randomIntBetween(1, 6))
+                .build()
+        );
+
+        try {
+            ensureGreen(index);
+        } finally {
+            deleteIndex(index);
+        }
+    }
+
     @Override
     protected RestClient buildClient(Settings settings, HttpHost[] hosts) throws IOException {
         RestClientBuilder builder = RestClient.builder(hosts);

diff --git a/src/internalClusterTest/java/org/opensearch/index/codec/CodecCompressionLevelIT.java b/src/internalClusterTest/java/org/opensearch/index/codec/CodecCompressionLevelIT.java
@@ -13,16 +13,21 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.codec.customcodecs.CustomCodecPlugin;
+import org.opensearch.index.codec.customcodecs.QatZipperFactory;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.test.OpenSearchIntegTestCase;
 
 import java.util.Collection;
 import java.util.Collections;
 import java.util.concurrent.ExecutionException;
 
+import static org.opensearch.index.codec.customcodecs.CustomCodecService.QAT_DEFLATE_CODEC;
+import static org.opensearch.index.codec.customcodecs.CustomCodecService.QAT_LZ4_CODEC;
 import static org.opensearch.index.codec.customcodecs.CustomCodecService.ZSTD_CODEC;
 import static org.opensearch.index.codec.customcodecs.CustomCodecService.ZSTD_NO_DICT_CODEC;
 import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assume.assumeThat;
 
 @OpenSearchIntegTestCase.ClusterScope(scope = OpenSearchIntegTestCase.Scope.TEST)
 public class CodecCompressionLevelIT extends OpenSearchIntegTestCase {
@@ -80,6 +85,26 @@ public void testZStandardCodecsCreateIndexWithCompressionLevel() {
         ensureGreen(index);
     }
 
+    public void testQatCodecsCreateIndexWithCompressionLevel() {
+        assumeThat("Qat library is available", QatZipperFactory.isQatAvailable(), is(true));
+
+        internalCluster().ensureAtLeastNumDataNodes(1);
+        final String index = "test-index";
+
+        // creating index
+        createIndex(
+            index,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put("index.codec", randomFrom(QAT_DEFLATE_CODEC, QAT_LZ4_CODEC))
+                .put("index.codec.compression_level", randomIntBetween(1, 6))
+                .build()
+        );
+
+        ensureGreen(index);
+    }
+
     public void testZStandardToLuceneCodecsWithCompressionLevel() throws ExecutionException, InterruptedException {
 
         internalCluster().ensureAtLeastNumDataNodes(1);
@@ -132,6 +157,59 @@ public void testZStandardToLuceneCodecsWithCompressionLevel() throws ExecutionEx
         ensureGreen(index);
     }
 
+    public void testQatToLuceneCodecsWithCompressionLevel() throws ExecutionException, InterruptedException {
+        assumeThat("Qat library is available", QatZipperFactory.isQatAvailable(), is(true));
+
+        internalCluster().ensureAtLeastNumDataNodes(1);
+        final String index = "test-index";
+
+        // creating index
+        createIndex(
+            index,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put("index.codec", randomFrom(QAT_DEFLATE_CODEC, QAT_LZ4_CODEC))
+                .put("index.codec.compression_level", randomIntBetween(1, 6))
+                .build()
+        );
+        ensureGreen(index);
+
+        assertAcked(client().admin().indices().prepareClose(index).setWaitForActiveShards(1));
+
+        Throwable executionException = expectThrows(
+            ExecutionException.class,
+            () -> client().admin()
+                .indices()
+                .updateSettings(
+                    new UpdateSettingsRequest(index).settings(
+                        Settings.builder().put("index.codec", randomFrom(CodecService.DEFAULT_CODEC, CodecService.BEST_COMPRESSION_CODEC))
+                    )
+                )
+                .get()
+        );
+
+        Throwable rootCause = Throwables.getRootCause(executionException);
+        assertEquals(IllegalArgumentException.class, rootCause.getClass());
+        assertTrue(rootCause.getMessage().startsWith("Compression level cannot be set"));
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .updateSettings(
+                    new UpdateSettingsRequest(index).settings(
+                        Settings.builder()
+                            .put("index.codec", randomFrom(CodecService.DEFAULT_CODEC, CodecService.BEST_COMPRESSION_CODEC))
+                            .put("index.codec.compression_level", (String) null)
+                    )
+                )
+                .get()
+        );
+
+        assertAcked(client().admin().indices().prepareOpen(index).setWaitForActiveShards(1));
+        ensureGreen(index);
+    }
+
     public void testLuceneToZStandardCodecsWithCompressionLevel() throws ExecutionException, InterruptedException {
 
         internalCluster().ensureAtLeastNumDataNodes(1);
@@ -185,4 +263,57 @@ public void testLuceneToZStandardCodecsWithCompressionLevel() throws ExecutionEx
         ensureGreen(index);
     }
 
+    public void testLuceneToQatCodecsWithCompressionLevel() throws ExecutionException, InterruptedException {
+        assumeThat("Qat library is available", QatZipperFactory.isQatAvailable(), is(true));
+
+        internalCluster().ensureAtLeastNumDataNodes(1);
+        final String index = "test-index";
+
+        // creating index
+        createIndex(
+            index,
+            Settings.builder()
+                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put("index.codec", randomFrom(CodecService.DEFAULT_CODEC, CodecService.BEST_COMPRESSION_CODEC))
+                .build()
+        );
+        ensureGreen(index);
+
+        assertAcked(client().admin().indices().prepareClose(index).setWaitForActiveShards(1));
+
+        Throwable executionException = expectThrows(
+            ExecutionException.class,
+            () -> client().admin()
+                .indices()
+                .updateSettings(
+                    new UpdateSettingsRequest(index).settings(
+                        Settings.builder()
+                            .put("index.codec", randomFrom(CodecService.DEFAULT_CODEC, CodecService.BEST_COMPRESSION_CODEC))
+                            .put("index.codec.compression_level", randomIntBetween(1, 6))
+                    )
+                )
+                .get()
+        );
+
+        Throwable rootCause = Throwables.getRootCause(executionException);
+        assertEquals(IllegalArgumentException.class, rootCause.getClass());
+        assertTrue(rootCause.getMessage().startsWith("Compression level cannot be set"));
+
+        assertAcked(
+            client().admin()
+                .indices()
+                .updateSettings(
+                    new UpdateSettingsRequest(index).settings(
+                        Settings.builder()
+                            .put("index.codec", randomFrom(QAT_DEFLATE_CODEC, QAT_LZ4_CODEC))
+                            .put("index.codec.compression_level", randomIntBetween(1, 6))
+                    )
+                )
+                .get()
+        );
+
+        assertAcked(client().admin().indices().prepareOpen(index).setWaitForActiveShards(1));
+        ensureGreen(index);
+    }
 }
diff --git a/src/internalClusterTest/java/org/opensearch/index/codec/MultiCodecMergeIT.java b/src/internalClusterTest/java/org/opensearch/index/codec/MultiCodecMergeIT.java
@@ -16,6 +16,7 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.codec.customcodecs.CustomCodecPlugin;
+import org.opensearch.index.codec.customcodecs.QatZipperFactory;
 import org.opensearch.index.engine.Segment;
 import org.opensearch.plugins.Plugin;
 import org.opensearch.test.OpenSearchIntegTestCase;
@@ -24,6 +25,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -51,25 +53,24 @@ protected Collection<Class<? extends Plugin>> nodePlugins() {
 
     public void testForceMergeMultipleCodecs() throws ExecutionException, InterruptedException {
 
-        Map<String, String> codecMap = Map.of(
-            "best_compression",
-            "BEST_COMPRESSION",
-            "zlib",
-            "BEST_COMPRESSION",
-            "zstd_no_dict",
-            "ZSTD_NO_DICT",
-            "zstd",
-            "ZSTD",
-            "default",
-            "BEST_SPEED",
-            "lz4",
-            "BEST_SPEED"
-        );
+        Map<String, String> codecMap = new HashMap<String, String>() {
+            {
+                put("best_compression", "BEST_COMPRESSION");
+                put("zlib", "BEST_COMPRESSION");
+                put("zstd_no_dict", "ZSTD_NO_DICT");
+                put("zstd", "ZSTD");
+                put("default", "BEST_SPEED");
+                put("lz4", "BEST_SPEED");
+                if (QatZipperFactory.isQatAvailable()) {
+                    put("qat_lz4", "QAT_LZ4");
+                    put("qat_deflate", "QAT_DEFLATE");
+                }
+            }
+        };
 
         for (Map.Entry<String, String> codec : codecMap.entrySet()) {
             forceMergeMultipleCodecs(codec.getKey(), codec.getValue(), codecMap);
         }
-
     }
 
     private void forceMergeMultipleCodecs(String finalCodec, String finalCodecMode, Map<String, String> codecMap) throws ExecutionException,

diff --git a/src/main/java/org/opensearch/index/codec/customcodecs/CustomCodecPlugin.java b/src/main/java/org/opensearch/index/codec/customcodecs/CustomCodecPlugin.java
@@ -8,28 +8,32 @@
 
 package org.opensearch.index.codec.customcodecs;
 
+import org.opensearch.common.settings.Setting;
 import org.opensearch.index.IndexSettings;
 import org.opensearch.index.codec.CodecServiceFactory;
 import org.opensearch.index.engine.EngineConfig;
 import org.opensearch.plugins.EnginePlugin;
 import org.opensearch.plugins.Plugin;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Optional;
 
 /**
  * A plugin that implements custom codecs. Supports these codecs:
+ *
  * <ul>
- * <li>ZSTD
- * <li>ZSTDNODICT
+ *   <li>ZSTD_CODEC
+ *   <li>ZSTD_NO_DICT_CODEC
+ *   <li>QAT_LZ4
+ *   <li>QAT_DEFLATE
  * </ul>
  *
  * @opensearch.internal
  */
 public final class CustomCodecPlugin extends Plugin implements EnginePlugin {
 
-    /**
-     * Creates a new instance
-     */
+    /** Creates a new instance */
     public CustomCodecPlugin() {}
 
     /**
@@ -39,9 +43,17 @@ public CustomCodecPlugin() {}
     @Override
     public Optional<CodecServiceFactory> getCustomCodecServiceFactory(final IndexSettings indexSettings) {
         String codecName = indexSettings.getValue(EngineConfig.INDEX_CODEC_SETTING);
-        if (codecName.equals(CustomCodecService.ZSTD_NO_DICT_CODEC) || codecName.equals(CustomCodecService.ZSTD_CODEC)) {
+        if (codecName.equals(CustomCodecService.ZSTD_NO_DICT_CODEC)
+            || codecName.equals(CustomCodecService.ZSTD_CODEC)
+            || codecName.equals(CustomCodecService.QAT_LZ4_CODEC)
+            || codecName.equals(CustomCodecService.QAT_DEFLATE_CODEC)) {
             return Optional.of(new CustomCodecServiceFactory());
         }
         return Optional.empty();
     }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return Arrays.asList(Lucene99QatCodec.INDEX_CODEC_QAT_MODE_SETTING);
+    }
 }