From 0126eaf4483e97ed322f7b56f38a56dc82618e9a Mon Sep 17 00:00:00 2001 From: Mulugeta Mammo Date: Fri, 22 May 2026 22:09:12 +0000 Subject: [PATCH 1/2] Upgrade qat-java to 2.5.0 and batch decompression into a single JNI call Bump the qat-java dependency from 2.4.0 to 2.5.0 and adapt to its API changes: - Remove the retryCount parameter from QatZipperFactory: qat-java 2.5.0 drops retry-count configuration from QatZipper.Builder, so the factory overloads that accepted it are removed and the remaining overloads simplified. - Rework QatCompressionMode decompression to read all needed compressed sub-blocks into one packed buffer and decompress them with a single qatZipper.decompressFull() call, instead of one JNI decompress call per block. This cuts JNI overhead on stored-fields reads that span multiple blocks. Signed-off-by: Mulugeta Mammo --- build.gradle | 2 +- .../customcodecs/QatCompressionMode.java | 32 ++++--- .../codec/customcodecs/QatZipperFactory.java | 89 +++++-------------- 3 files changed, 46 insertions(+), 77 deletions(-) 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/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java b/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java index fa7e965..e5766a5 100644 --- a/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java +++ b/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java @@ -85,6 +85,7 @@ public int getCompressionLevel() { return compressionLevel; } + /** The QatCompressor. */ /** The QatCompressor. */ private static final class QatCompressor extends Compressor { @@ -174,24 +175,35 @@ 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. + // Pre-grow once to avoid per-iteration copies. + compressed = ArrayUtil.grow(compressed, originalLength); + 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; + 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); + + bytes.length = totalWritten; 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 { From f4378ba7311409e28c5779bd0aff03a643f2fc45 Mon Sep 17 00:00:00 2001 From: Mulugeta Mammo Date: Sat, 18 Jul 2026 06:16:05 +0000 Subject: [PATCH 2/2] Address PR review: don't over-allocate buffer. And, update SHA for qat-java jar. - Fix decompress buffer over-allocation: start the packed compressed buffer at originalLength/2 with growNoCopy and grow on demand while packing, instead of pre-growing to the full decompressed size. - Validate the decompressFull byte count via assert and drop the dead bytes.length = totalWritten assignment. - Remove duplicate QatCompressor Javadoc line. - Add SHA for qat-java-2.5.0.jar and remove the stale 2.4.0 SHA (gradle updateShas). Signed-off-by: Mulugeta Mammo --- licenses/qat-java-2.4.0.jar.sha1 | 1 - licenses/qat-java-2.5.0.jar.sha1 | 1 + .../codec/customcodecs/QatCompressionMode.java | 13 +++++++++---- 3 files changed, 10 insertions(+), 5 deletions(-) delete mode 100644 licenses/qat-java-2.4.0.jar.sha1 create mode 100644 licenses/qat-java-2.5.0.jar.sha1 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 e5766a5..fb2b3a7 100644 --- a/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java +++ b/src/main/java/org/opensearch/index/codec/customcodecs/QatCompressionMode.java @@ -85,7 +85,6 @@ public int getCompressionLevel() { return compressionLevel; } - /** The QatCompressor. */ /** The QatCompressor. */ private static final class QatCompressor extends Compressor { @@ -176,8 +175,9 @@ public void decompress(DataInput in, int originalLength, int offset, int length, } // Read all needed sub-blocks into a packed compressed buffer. - // Pre-grow once to avoid per-iteration copies. - compressed = ArrayUtil.grow(compressed, originalLength); + // 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; @@ -187,6 +187,7 @@ public void decompress(DataInput in, int originalLength, int offset, int length, break; } + compressed = ArrayUtil.grow(compressed, srcPos + compressedLength); in.readBytes(compressed, srcPos, compressedLength); srcPos += compressedLength; totalDecompressed += Math.min(blockLength, originalLength - offsetInBlock); @@ -202,8 +203,12 @@ public void decompress(DataInput in, int originalLength, int offset, int length, // 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.length = totalWritten; bytes.offset = offsetInBytesRef; bytes.length = length;