forked from opensearch-project/OpenSearch
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Peter Alfonsi <petealft@amazon.com>
- Loading branch information
Peter Alfonsi
committed
Jan 15, 2025
1 parent
4b392e1
commit 1ebdc0b
Showing
4 changed files
with
133 additions
and
6 deletions.
There are no files selected for viewing
72 changes: 72 additions & 0 deletions
72
server/src/main/java/org/opensearch/indices/query/CompressorPool.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| /* | ||
| * 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.indices.query; | ||
|
|
||
| import org.apache.lucene.store.ByteArrayDataInput; | ||
| import org.apache.lucene.store.ByteArrayDataOutput; | ||
| import org.apache.lucene.util.compress.LZ4; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.Future; | ||
|
|
||
| public class CompressorPool { | ||
| private final ExecutorService executor; | ||
| private final ThreadLocal<LZ4Compressor> threadLocalCompressor; | ||
|
|
||
| public CompressorPool(int poolSize) { | ||
| this.executor = Executors.newFixedThreadPool(poolSize); | ||
| this.threadLocalCompressor = ThreadLocal.withInitial(LZ4Compressor::new); | ||
| } | ||
|
|
||
| public Future<byte[]> compress(byte[] data) { | ||
| return executor.submit(() -> { | ||
| LZ4Compressor compressor = threadLocalCompressor.get(); | ||
| return compressor.compress(data); | ||
| }); | ||
| } | ||
|
|
||
| public void shutdown() { | ||
| executor.shutdown(); | ||
| } | ||
|
|
||
| public byte[] decompress(byte[] compressed) throws IOException { | ||
| // Decompression is threadsafe already (?) | ||
| ByteArrayDataInput in = new ByteArrayDataInput(compressed); | ||
| int originalLength = in.readInt(); | ||
| byte[] result = new byte[originalLength]; | ||
| LZ4.decompress(in, originalLength, result, 0); | ||
| return result; | ||
| } | ||
|
|
||
| static class LZ4Compressor { | ||
| private final static int scratchOverhead = 256; | ||
|
|
||
| private final LZ4.FastCompressionHashTable ht; | ||
|
|
||
| LZ4Compressor() { | ||
| ht = new LZ4.FastCompressionHashTable(); | ||
| } | ||
|
|
||
| public byte[] compress(byte[] bytes) throws IOException { | ||
| byte[] result = new byte[bytes.length + 4 + scratchOverhead]; // We seem to need overhead? At least for short byte[] | ||
| ByteArrayDataOutput out = new ByteArrayDataOutput(result); | ||
| // Write the original length into out; we need it at decompression time | ||
| out.writeInt(bytes.length); | ||
| LZ4.compress(bytes, 0, bytes.length, out, ht); | ||
| int finalLength = out.getPosition(); | ||
| byte[] trimmed = new byte[finalLength]; | ||
| System.arraycopy(result, 0, trimmed, 0, finalLength); | ||
| return trimmed; | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
server/src/test/java/org/opensearch/indices/query/CompressorPoolTests.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| /* | ||
| * 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.indices.query; | ||
|
|
||
| import org.opensearch.test.OpenSearchTestCase; | ||
|
|
||
| public class CompressorPoolTests extends OpenSearchTestCase { | ||
| public void testCompression() throws Exception { | ||
| CompressorPool pool = new CompressorPool(10); | ||
| byte[] orig = new byte[]{1, 2, 3, 4, 5, 6, 4, 5, 6, 7, 4, 5, 6, 7, 4, 5, 6}; | ||
| byte[] compressed = pool.compress(orig).get(); | ||
| byte[] decompressed = pool.decompress(compressed); | ||
| assertArrayEquals(orig, decompressed); | ||
| pool.shutdown(); | ||
| } | ||
|
|
||
| // TODO | ||
| public void testMultithreadedCompression() throws Exception { | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters