diff --git a/build.gradle b/build.gradle index 9f22262..bdbe6ff 100644 --- a/build.gradle +++ b/build.gradle @@ -97,7 +97,7 @@ opensearchplugin { dependencies { api "com.github.luben:zstd-jni:1.5.6-1" - api "com.intel.qat:qat-java:2.4.0" + api "com.intel.qat:qat-java:2.5.0" } allprojects { diff --git a/licenses/qat-java-2.4.0.jar.sha1 b/licenses/qat-java-2.4.0.jar.sha1 deleted file mode 100644 index 41a2ff3..0000000 --- a/licenses/qat-java-2.4.0.jar.sha1 +++ /dev/null @@ -1 +0,0 @@ -3f0706b2406972f2d32a65c553fd0f5d745d8431 \ No newline at end of file diff --git a/licenses/qat-java-2.5.0.jar.sha1 b/licenses/qat-java-2.5.0.jar.sha1 new file mode 100644 index 0000000..ebdd291 --- /dev/null +++ b/licenses/qat-java-2.5.0.jar.sha1 @@ -0,0 +1 @@ +b1adc9a53eb6f1e8d61407f3461406d546bd11d4 \ No newline at end of file diff --git a/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java b/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java index fa7e965..fb2b3a7 100644 --- a/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java +++ b/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java @@ -174,24 +174,41 @@ public void decompress(DataInput in, int originalLength, int offset, int length, offsetInBytesRef -= blockLength; } - // Read blocks that intersect with the interval we need + // Read all needed sub-blocks into a packed compressed buffer. + // Start from a modest estimate (compressed data is typically smaller + // than the decompressed size) and grow on demand while packing. + compressed = ArrayUtil.growNoCopy(compressed, originalLength / 2); + int srcPos = 0; + int totalDecompressed = 0; + while (offsetInBlock < offset + length) { final int compressedLength = in.readVInt(); if (compressedLength == 0) { - return; + break; } - compressed = ArrayUtil.growNoCopy(compressed, compressedLength); - in.readBytes(compressed, 0, compressedLength); - - int l = Math.min(blockLength, originalLength - offsetInBlock); - bytes.bytes = ArrayUtil.grow(bytes.bytes, bytes.length + l); - final int uncompressed = qatZipper.decompress(compressed, 0, compressedLength, bytes.bytes, bytes.length, l); - - bytes.length += uncompressed; + compressed = ArrayUtil.grow(compressed, srcPos + compressedLength); + in.readBytes(compressed, srcPos, compressedLength); + srcPos += compressedLength; + totalDecompressed += Math.min(blockLength, originalLength - offsetInBlock); offsetInBlock += blockLength; } + if (srcPos == 0) { + return; + } + + bytes.bytes = ArrayUtil.grow(bytes.bytes, totalDecompressed); + + // Single JNI call: native side loops through all concatenated + // compressed frames, decompressing them into the output buffer. + int totalWritten = qatZipper.decompressFull(compressed, 0, srcPos, bytes.bytes, 0, totalDecompressed); + assert totalWritten == totalDecompressed : "Decompressed byte count (" + + totalWritten + + ") does not match expected (" + + totalDecompressed + + ")."; + bytes.offset = offsetInBytesRef; bytes.length = length; diff --git a/src/main/java/org/opensearch/index/codec/customcodecs/QatZipperFactory.java b/src/main/java/org/opensearch/index/codec/customcodecs/QatZipperFactory.java index 3b379a6..6c51ac3 100644 --- a/src/main/java/org/opensearch/index/codec/customcodecs/QatZipperFactory.java +++ b/src/main/java/org/opensearch/index/codec/customcodecs/QatZipperFactory.java @@ -5,7 +5,6 @@ * this file be licensed under the Apache-2.0 license or a * compatible open source license. */ - package org.opensearch.index.codec.customcodecs; import com.intel.qat.QatZipper; @@ -15,24 +14,21 @@ import static com.intel.qat.QatZipper.DEFAULT_COMPRESSION_LEVEL_ZSTD; import static com.intel.qat.QatZipper.DEFAULT_MODE; import static com.intel.qat.QatZipper.DEFAULT_POLLING_MODE; -import static com.intel.qat.QatZipper.DEFAULT_RETRY_COUNT; import static com.intel.qat.QatZipper.Mode; import static com.intel.qat.QatZipper.PollingMode; /** A factory class to create instances of QatZipper */ public class QatZipperFactory { - /** * Creates a new QatZipper with the specified parameters. * * @param algorithm the compression algorithm * @param level the compression level. * @param mode the mode of QAT execution - * @param retryCount the number of attempts to acquire hardware resources * @param pmode polling mode. */ - public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode, int retryCount, PollingMode pmode) { - return new QatZipper.Builder().algorithm(algorithm).level(level).mode(mode).retryCount(retryCount).pollingMode(pmode).build(); + public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode, PollingMode pmode) { + return new QatZipper.Builder().algorithm(algorithm).level(level).mode(mode).pollingMode(pmode).build(); } /** @@ -40,105 +36,91 @@ public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode * * @param algorithm the compression algorithm * @param mode the mode of QAT execution - * @param retryCount the number of attempts to acquire hardware resources * @param pmode polling mode. */ - public static QatZipper createInstance(Algorithm algorithm, Mode mode, int retryCount, PollingMode pmode) { + public static QatZipper createInstance(Algorithm algorithm, Mode mode, PollingMode pmode) { return createInstance( algorithm, algorithm == Algorithm.ZSTD ? DEFAULT_COMPRESSION_LEVEL_ZSTD : DEFAULT_COMPRESSION_LEVEL_DEFLATE, mode, - retryCount, pmode ); } /** * Creates a new QatZipper that uses the DEFLATE algorithm and the default compression level, - * mode, retry count, and polling mode. + * mode, and polling mode. */ public static QatZipper createInstance() { - return createInstance(Algorithm.DEFLATE, DEFAULT_MODE, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(Algorithm.DEFLATE, DEFAULT_MODE, DEFAULT_POLLING_MODE); } /** * Creates a new QatZipper with the specified compression algorithm. Uses the default compression - * level, mode, retry count, and polling mode. + * level, mode, and polling mode. * * @param algorithm the compression algorithm */ public static QatZipper createInstance(Algorithm algorithm) { - return createInstance(algorithm, DEFAULT_MODE, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(algorithm, DEFAULT_MODE, DEFAULT_POLLING_MODE); } /** * Creates a new QatZipper with the specified execution mode. Uses the DEFLATE algorithm with the - * default compression level, retry count, and polling mode. + * default compression level and polling mode. * * @param mode the mode of QAT execution */ public static QatZipper createInstance(Mode mode) { - return createInstance(Algorithm.DEFLATE, mode, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(Algorithm.DEFLATE, mode, DEFAULT_POLLING_MODE); } /** * Creates a new QatZipper with the specified polling polling mode. Uses the DEFLATE algorithm - * with the default compression level, mode, and retry count. + * with the default compression level and mode. * * @param pmode the polling mode. */ public static QatZipper createInstance(PollingMode pmode) { - return createInstance(Algorithm.DEFLATE, DEFAULT_MODE, DEFAULT_RETRY_COUNT, pmode); + return createInstance(Algorithm.DEFLATE, DEFAULT_MODE, pmode); } /** * Creates a new QatZipper with the specified algorithm and compression level. Uses the default - * mode, retry count, and polling mode. + * mode and polling mode. * * @param algorithm the compression algorithm (DEFLATE, LZ4, or ZSTD). * @param level the compression level. */ public static QatZipper createInstance(Algorithm algorithm, int level) { - return createInstance(algorithm, level, DEFAULT_MODE, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(algorithm, level, DEFAULT_MODE, DEFAULT_POLLING_MODE); } /** * Creates a new QatZipper with the specified algorithm and mode of execution. Uses the default - * compression level, retry count, and polling mode. + * compression level and polling mode. * * @param algorithm the compression algorithm * @param mode the mode of QAT execution */ public static QatZipper createInstance(Algorithm algorithm, Mode mode) { - return createInstance(algorithm, mode, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(algorithm, mode, DEFAULT_POLLING_MODE); } /** * Creates a new QatZipper with the specified algorithm and polling mode of execution. Uses the - * default compression level, mode, and retry count. + * default compression level and mode. * * @param algorithm the compression algorithm * @param pmode the polling mode. */ public static QatZipper createInstance(Algorithm algorithm, PollingMode pmode) { - return createInstance(algorithm, DEFAULT_MODE, DEFAULT_RETRY_COUNT, pmode); - } - - /** - * Creates a new QatZipper with the specified algorithm and mode of execution. Uses compression - * level and retry count. - * - * @param algorithm the compression algorithm - * @param mode the mode of QAT execution - * @param pmode the polling mode. - */ - public static QatZipper createInstance(Algorithm algorithm, Mode mode, PollingMode pmode) { - return createInstance(algorithm, mode, DEFAULT_RETRY_COUNT, pmode); + return createInstance(algorithm, DEFAULT_MODE, pmode); } /** - * Creates a new QatZipper with the specified algorithm, compression level, and mode . Uses the - * default retry count and polling mode. + * Creates a new QatZipper with the specified algorithm, compression level, and mode. Uses the + * default polling mode. * * @param algorithm the compression algorithm (DEFLATE, LZ4, or ZSTD). * @param level the compression level. @@ -146,43 +128,19 @@ public static QatZipper createInstance(Algorithm algorithm, Mode mode, PollingMo * failover.) */ public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode) { - return createInstance(algorithm, level, mode, DEFAULT_RETRY_COUNT, DEFAULT_POLLING_MODE); + return createInstance(algorithm, level, mode, DEFAULT_POLLING_MODE); } /** - * Creates a new QatZipper with the specified algorithm, compression level, and polling mode . - * Uses the default mode and retry count. + * Creates a new QatZipper with the specified algorithm, compression level, and polling mode. + * Uses the default mode. * * @param algorithm the compression algorithm (DEFLATE, LZ4, or ZSTD). * @param level the compression level. * @param pmode the polling mode. */ public static QatZipper createInstance(Algorithm algorithm, int level, PollingMode pmode) { - return createInstance(algorithm, level, DEFAULT_MODE, DEFAULT_RETRY_COUNT, pmode); - } - - /** - * Creates a new QatZipper with the specified parameters and polling mode. - * - * @param algorithm the compression algorithm - * @param level the compression level. - * @param mode the mode of QAT execution - * @param retryCount the number of attempts to acquire hardware resources - */ - public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode, int retryCount) { - return createInstance(algorithm, level, mode, retryCount, DEFAULT_POLLING_MODE); - } - - /** - * Creates a new QatZipper with the specified parameters and retry count. - * - * @param algorithm the compression algorithm - * @param level the compression level. - * @param mode the mode of QAT execution - * @param pmode the polling mode. - */ - public static QatZipper createInstance(Algorithm algorithm, int level, Mode mode, PollingMode pmode) { - return createInstance(algorithm, level, mode, DEFAULT_RETRY_COUNT, pmode); + return createInstance(algorithm, level, DEFAULT_MODE, pmode); } /** @@ -199,7 +157,6 @@ public static boolean isQatAvailable() { */ private static class QatAvailableHolder { static final boolean IS_QAT_AVAILABLE; - static { boolean isQatAvailable; try {