Add hardware-accelerated codecs for DEFLATE and LZ4

build.gradle

@@ -72,6 +72,7 @@ opensearchplugin {
 
 dependencies {
   api "com.github.luben:zstd-jni:1.5.5-5"
+  api "com.intel.qat:qat-java:1.1.1"
 }
 
 allprojects {

licenses/qat-java-1.1.1.jar.sha1

@@ -0,0 +1 @@
+3333601cdedf6a711d445118d5bc44ec6a9c65f9

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.

licenses/qat-java-NOTICE.txt

@@ -0,0 +1 @@
+Qat-Java is a compression library that uses Intel® QAT to accelerate compression and decompression.

src/main/java/org/opensearch/index/codec/customcodecs/CustomCodecPlugin.java

@@ -8,28 +8,47 @@
 
 package org.opensearch.index.codec.customcodecs;
 
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.settings.Setting.Property;
 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;
 
+import com.intel.qat.QatZipper;
+
 /**
  * A plugin that implements custom codecs. Supports these codecs:
+ *
  * <ul>
- * <li>ZSTD
- * <li>ZSTDNODICT
+ *   <li>ZSTD_CODEC
+ *   <li>ZSTD_NO_DICT_CODEC
+ *   <li>QAT_DEFLATE
+ *   <li>QAT_LZ4
  * </ul>
  *
  * @opensearch.internal
  */
 public final class CustomCodecPlugin extends Plugin implements EnginePlugin {
 
-    /**
-     * Creates a new instance
-     */
+    /** A setting to specifiy the QAT acceleration mode. */
+    public static final Setting<QatZipper.Mode> INDEX_CODEC_QAT_MODE_SETTING = new Setting<>("index.codec.qatmode", "hardware", s -> {
+        switch (s) {
+            case "auto":
+                return QatZipper.Mode.AUTO;
+            case "hardware":
+                return QatZipper.Mode.HARDWARE;
+            default:
+                throw new IllegalArgumentException("Unknown value for [index.codec.qatmode] must be one of [auto, hardware] but was: " + s);
+        }
+    }, Property.IndexScope, Property.Dynamic);
+
+    /** Creates a new instance */
     public CustomCodecPlugin() {}
 
     /**
@@ -39,9 +58,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_DEFLATE_CODEC)
+            || codecName.equals(CustomCodecService.QAT_LZ4_CODEC)) {
             return Optional.of(new CustomCodecServiceFactory());
         }
         return Optional.empty();
     }
+
+    @Override
+    public List<Setting<?>> getSettings() {
+        return Arrays.asList(INDEX_CODEC_QAT_MODE_SETTING);
+    }
 }

src/main/java/org/opensearch/index/codec/customcodecs/CustomCodecService.java

@@ -21,26 +21,28 @@
 
 import static org.opensearch.index.engine.EngineConfig.INDEX_CODEC_COMPRESSION_LEVEL_SETTING;
 
-/**
- * CustomCodecService provides ZSTD and ZSTD_NO_DICT compression codecs.
- */
+/** CustomCodecService provides ZSTD, ZSTD_NO_DICT, QAT_DEFLATE, and QAT_LZ4 compression codecs. */
 public class CustomCodecService extends CodecService {
     private final Map<String, Codec> codecs;
-    /**
-     * ZStandard codec
-     */
+
+    /** ZStandard codec */
     public static final String ZSTD_CODEC = "zstd";
-    /**
-     * ZStandard without dictionary codec
-     */
+
+    /** ZStandard without dictionary codec */
     public static final String ZSTD_NO_DICT_CODEC = "zstd_no_dict";
 
+    /** Hardware accelerated (Intel QAT) compression codec for DEFLATE. */
+    public static final String QAT_DEFLATE_CODEC = "qat_deflate";
+
+    /** Hardware accelerated (Intel QAT) compression codec for LZ4. */
+    public static final String QAT_LZ4_CODEC = "qat_lz4";
+
     /**
      * Creates a new CustomCodecService.
      *
      * @param mapperService The mapper service.
      * @param indexSettings The index settings.
-     * @param logger        The logger.
+     * @param logger The logger.
      */
     public CustomCodecService(MapperService mapperService, IndexSettings indexSettings, Logger logger) {
         super(mapperService, indexSettings, logger);
@@ -49,9 +51,21 @@ public CustomCodecService(MapperService mapperService, IndexSettings indexSettin
         if (mapperService == null) {
             codecs.put(ZSTD_CODEC, new Zstd99Codec(compressionLevel));
             codecs.put(ZSTD_NO_DICT_CODEC, new ZstdNoDict99Codec(compressionLevel));
+            codecs.put(QAT_DEFLATE_CODEC, new QatDeflate99Codec(compressionLevel, () -> {
+                return indexSettings.getValue(CustomCodecPlugin.INDEX_CODEC_QAT_MODE_SETTING);
+            }));
+            codecs.put(QAT_LZ4_CODEC, new QatLz499Codec(compressionLevel, () -> {
+                return indexSettings.getValue(CustomCodecPlugin.INDEX_CODEC_QAT_MODE_SETTING);
+            }));
         } else {
             codecs.put(ZSTD_CODEC, new Zstd99Codec(mapperService, logger, compressionLevel));
             codecs.put(ZSTD_NO_DICT_CODEC, new ZstdNoDict99Codec(mapperService, logger, compressionLevel));
+            codecs.put(QAT_DEFLATE_CODEC, new QatDeflate99Codec(mapperService, logger, compressionLevel, () -> {
+                return indexSettings.getValue(CustomCodecPlugin.INDEX_CODEC_QAT_MODE_SETTING);
+            }));
+            codecs.put(QAT_LZ4_CODEC, new QatLz499Codec(mapperService, logger, compressionLevel, () -> {
+                return indexSettings.getValue(CustomCodecPlugin.INDEX_CODEC_QAT_MODE_SETTING);
+            }));
         }
         this.codecs = codecs.immutableMap();
     }

src/main/java/org/opensearch/index/codec/customcodecs/CustomCodecServiceFactory.java

@@ -12,9 +12,7 @@
 import org.opensearch.index.codec.CodecServiceConfig;
 import org.opensearch.index.codec.CodecServiceFactory;
 
-/**
- * A factory for creating new {@link CodecService} instance
- */
+/** A factory for creating new {@link CodecService} instance */
 public class CustomCodecServiceFactory implements CodecServiceFactory {
 
     /** Creates a new instance. */

src/main/java/org/opensearch/index/codec/customcodecs/Lucene99CustomStoredFieldsFormat.java

@@ -17,7 +17,6 @@
 import org.apache.lucene.index.SegmentInfo;
 import org.apache.lucene.store.Directory;
 import org.apache.lucene.store.IOContext;
-import org.opensearch.index.codec.customcodecs.backward_codecs.Lucene95CustomCodec;
 
 import java.io.IOException;
 import java.util.Objects;
@@ -74,7 +73,7 @@ public Lucene99CustomStoredFieldsFormat(Lucene99CustomCodec.Mode mode, int compr
     */
     @Override
     public StoredFieldsReader fieldsReader(Directory directory, SegmentInfo si, FieldInfos fn, IOContext context) throws IOException {
-        if (si.getAttribute(MODE_KEY) !=null){
+        if (si.getAttribute(MODE_KEY) != null) {
             String value = si.getAttribute(MODE_KEY);
             Lucene99CustomCodec.Mode mode = Lucene99CustomCodec.Mode.valueOf(value);
             return impl(mode).fieldsReader(directory, si, fn, context);
@@ -111,14 +110,13 @@ StoredFieldsFormat impl(Lucene99CustomCodec.Mode mode) {
         }
     }
 
-
     private StoredFieldsFormat getCustomCompressingStoredFieldsFormat(String formatName, CompressionMode compressionMode) {
         return new Lucene90CompressingStoredFieldsFormat(
-                formatName,
-                compressionMode,
-                ZSTD_BLOCK_LENGTH,
-                ZSTD_MAX_DOCS_PER_BLOCK,
-                ZSTD_BLOCK_SHIFT
+            formatName,
+            compressionMode,
+            ZSTD_BLOCK_LENGTH,
+            ZSTD_MAX_DOCS_PER_BLOCK,
+            ZSTD_BLOCK_SHIFT
         );
     }
 

src/main/java/org/opensearch/index/codec/customcodecs/Lucene99QatCodec.java

@@ -0,0 +1,147 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.index.codec.customcodecs;
+
+import org.apache.logging.log4j.Logger;
+import org.apache.lucene.codecs.FilterCodec;
+import org.apache.lucene.codecs.StoredFieldsFormat;
+import org.apache.lucene.codecs.lucene99.Lucene99Codec;
+import org.opensearch.index.codec.PerFieldMappingPostingFormatCodec;
+import org.opensearch.index.mapper.MapperService;
+
+import java.util.Set;
+import java.util.function.Supplier;
+
+import com.intel.qat.QatZipper;
+
+/**
+ * Extends {@link FilterCodec} to reuse the functionality of Lucene Codec. Supports two modes zstd
+ * and zstd_no_dict. Uses Lucene99 as the delegate codec
+ *
+ * @opensearch.internal
+ */
+public abstract class Lucene99QatCodec extends FilterCodec {
+
+    /** Default compression level used for compression */
+    public static final int DEFAULT_COMPRESSION_LEVEL = 3;
+
+    /** The default QAT mode */
+    public static final QatZipper.Mode DEFAULT_QAT_MODE = QatZipper.Mode.HARDWARE;
+
+    /** Each mode represents a compression algorithm. */
+    public enum Mode {
+        /** QAT deflate mode. */
+        QAT_DEFLATE("QATDEFLATE99", Set.of("qat_deflate")),
+        /** QAT lz4 mode. */
+        QAT_LZ4("QATLZ499", Set.of("qat_lz4"));
+
+        private final String codec;
+        private final Set<String> aliases;
+
+        Mode(String codec, Set<String> aliases) {
+            this.codec = codec;
+            this.aliases = aliases;
+        }
+
+        /** Returns the Codec that is registered with Lucene */
+        public String getCodec() {
+            return codec;
+        }
+
+        /** Returns the aliases of the Codec */
+        public Set<String> getAliases() {
+            return aliases;
+        }
+    }
+
+    private final StoredFieldsFormat storedFieldsFormat;
+
+    /**
+     * Creates a new compression codec with the default compression level.
+     *
+     * @param mode The compression codec (QAT_DEFLATE or QAT_LZ4).
+     */
+    public Lucene99QatCodec(Mode mode) {
+        this(mode, DEFAULT_COMPRESSION_LEVEL);
+    }
+
+    /**
+     * Creates a new compression codec with the given compression level. We use lowercase letters when
+     * registering the codec so that we remain consistent with the other compression codecs: default,
+     * lucene_default, and best_compression.
+     *
+     * @param mode The compression codec (QAT_DEFLATE or QAT_LZ4).
+     * @param compressionLevel The compression level.
+     */
+    public Lucene99QatCodec(Mode mode, int compressionLevel) {
+        super(mode.getCodec(), new Lucene99Codec());
+        this.storedFieldsFormat = new Lucene99QatStoredFieldsFormat(mode, compressionLevel);
+    }
+
+    /**
+     * Creates a new compression codec with the given compression level. We use lowercase letters when
+     * registering the codec so that we remain consistent with the other compression codecs: default,
+     * lucene_default, and best_compression.
+     *
+     * @param mode The compression codec (QAT_DEFLATE, or QAT_LZ4).
+     * @param compressionLevel The compression level.
+     * @param supplier supplier for QAT mode.
+     */
+    public Lucene99QatCodec(Mode mode, int compressionLevel, Supplier<QatZipper.Mode> supplier) {
+        super(mode.getCodec(), new Lucene99Codec());
+        this.storedFieldsFormat = new Lucene99QatStoredFieldsFormat(mode, compressionLevel, supplier);
+    }
+
+    /**
+     * Creates a new compression codec with the given compression level. We use lowercase letters when
+     * registering the codec so that we remain consistent with the other compression codecs: default,
+     * lucene_default, and best_compression.
+     *
+     * @param mode The compression codec (QAT_DEFLATE or QAT_LZ4).
+     * @param compressionLevel The compression level.
+     * @param mapperService The mapper service.
+     * @param logger The logger.
+     */
+    public Lucene99QatCodec(Mode mode, int compressionLevel, MapperService mapperService, Logger logger) {
+        super(mode.getCodec(), new PerFieldMappingPostingFormatCodec(Lucene99Codec.Mode.BEST_SPEED, mapperService, logger));
+        this.storedFieldsFormat = new Lucene99QatStoredFieldsFormat(mode, compressionLevel);
+    }
+
+    /**
+     * Creates a new compression codec with the given compression level. We use lowercase letters when
+     * registering the codec so that we remain consistent with the other compression codecs: default,
+     * lucene_default, and best_compression.
+     *
+     * @param mode The compression codec (ZSTD, ZSTD_NO_DICT, QAT_DEFLATE, or QAT_LZ4).
+     * @param compressionLevel The compression level.
+     * @param mapperService The mapper service.
+     * @param logger The logger.
+     * @param supplier supplier for QAT mode.
+     */
+    public Lucene99QatCodec(
+        Mode mode,
+        int compressionLevel,
+        MapperService mapperService,
+        Logger logger,
+        Supplier<QatZipper.Mode> supplier
+    ) {
+        super(mode.getCodec(), new PerFieldMappingPostingFormatCodec(Lucene99Codec.Mode.BEST_SPEED, mapperService, logger));
+        this.storedFieldsFormat = new Lucene99QatStoredFieldsFormat(mode, compressionLevel, supplier);
+    }
+
+    @Override
+    public StoredFieldsFormat storedFieldsFormat() {
+        return storedFieldsFormat;
+    }
+
+    @Override
+    public String toString() {
+        return getClass().getSimpleName();
+    }
+}