This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9abfbd137 [core] Introduce more compressions for spill (#2812)
9abfbd137 is described below
commit 9abfbd13784e1b8951328a99387ca0782dabd319
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Jan 29 22:33:46 2024 +0800
[core] Introduce more compressions for spill (#2812)
---
.../shortcodes/generated/core_configuration.html | 18 ++--
paimon-common/pom.xml | 14 +++
.../main/java/org/apache/paimon/CoreOptions.java | 11 +++
.../paimon/compression/AirBlockCompressor.java | 32 ++++---
.../paimon/compression/AirBlockDecompressor.java | 33 +++----
.../paimon/compression/AirCompressorFactory.java | 20 +++-
.../compression/BlockCompressionFactory.java | 19 ++++
.../apache/paimon/compression/BlockCompressor.java | 0
.../paimon/compression/BlockDecompressor.java | 0
.../compression/BufferCompressionException.java | 0
.../compression/BufferDecompressionException.java | 0
.../apache/paimon/compression/CompressorUtils.java | 0
.../compression/Lz4BlockCompressionFactory.java | 0
.../paimon/compression/Lz4BlockCompressor.java | 0
.../paimon/compression/Lz4BlockDecompressor.java | 0
.../paimon/compression/BlockCompressionTest.java | 103 +++++++++++++++++++++
paimon-core/pom.xml | 6 --
.../paimon/crosspartition/GlobalIndexAssigner.java | 3 +-
.../org/apache/paimon/lookup/RocksDBState.java | 3 +-
.../org/apache/paimon/mergetree/MergeSorter.java | 5 +-
.../apache/paimon/mergetree/MergeTreeWriter.java | 4 +
.../paimon/mergetree/SortBufferWriteBuffer.java | 4 +-
.../paimon/operation/KeyValueFileStoreWrite.java | 1 +
.../paimon/sort/BinaryExternalSortBuffer.java | 18 ++--
.../apache/paimon/mergetree/MergeTreeTestBase.java | 1 +
.../mergetree/SortBufferWriteBufferTestBase.java | 1 +
.../paimon/sort/BinaryExternalSortBufferTest.java | 3 +-
.../paimon/flink/sink/LocalMergeOperator.java | 1 +
.../apache/paimon/flink/sorter/SortOperator.java | 11 ++-
.../org/apache/paimon/flink/sorter/SortUtils.java | 1 +
.../paimon/flink/sorter/SortOperatorTest.java | 2 +
pom.xml | 1 +
32 files changed, 253 insertions(+), 62 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html
b/docs/layouts/shortcodes/generated/core_configuration.html
index 3abe40262..f9d167de4 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -575,6 +575,12 @@ This config option does not affect the default filesystem
metastore.</td>
<td>Duration</td>
<td>The maximum time of completed snapshots to retain.</td>
</tr>
+ <tr>
+ <td><h5>snapshot.watermark-idle-timeout</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>In watermarking, if a source remains idle beyond the specified
timeout duration, it triggers snapshot advancement and facilitates tag
creation.</td>
+ </tr>
<tr>
<td><h5>sort-engine</h5></td>
<td style="word-wrap: break-word;">loser-tree</td>
@@ -605,6 +611,12 @@ This config option does not affect the default filesystem
metastore.</td>
<td>MemorySize</td>
<td>Target size of a source split when scanning a bucket.</td>
</tr>
+ <tr>
+ <td><h5>spill-compression</h5></td>
+ <td style="word-wrap: break-word;">"LZ4"</td>
+ <td>String</td>
+ <td>Compression for spill, currently lz4, lzo and zstd are
supported.</td>
+ </tr>
<tr>
<td><h5>streaming-read-mode</h5></td>
<td style="word-wrap: break-word;">(none)</td>
@@ -653,12 +665,6 @@ This config option does not affect the default filesystem
metastore.</td>
<td>Integer</td>
<td>The maximum number of tags to retain.</td>
</tr>
- <tr>
- <td><h5>snapshot.watermark-idle-timeout</h5></td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>Duration</td>
- <td>In watermarking, if a source remains idle beyond the specified
timeout duration, it triggers snapshot advancement and facilitates tag
creation.</td>
- </tr>
<tr>
<td><h5>tag.period-formatter</h5></td>
<td style="word-wrap: break-word;">with_dashes</td>
diff --git a/paimon-common/pom.xml b/paimon-common/pom.xml
index a6b01000b..d6d40a9fc 100644
--- a/paimon-common/pom.xml
+++ b/paimon-common/pom.xml
@@ -50,6 +50,20 @@ under the License.
<version>${paimon.shade.caffeine.version}-${paimon.shade.version}</version>
</dependency>
+ <!-- From paimon-format -->
+ <dependency>
+ <groupId>io.airlift</groupId>
+ <artifactId>aircompressor</artifactId>
+ <version>${airlift.version}</version>
+ </dependency>
+
+ <!-- From paimon-bundle -->
+ <dependency>
+ <groupId>org.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>${lz4.version}</version>
+ </dependency>
+
<!-- Java compiler -->
<dependency>
<groupId>org.codehaus.janino</groupId>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 49828ac11..64cbc5b6f 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -283,6 +283,13 @@ public class CoreOptions implements Serializable {
.defaultValue(MemorySize.parse("64 mb"))
.withDescription("Amount of data to spill records to disk
in spilled sort.");
+ public static final ConfigOption<String> SPILL_COMPRESSION =
+ key("spill-compression")
+ .stringType()
+ .defaultValue("LZ4")
+ .withDescription(
+ "Compression for spill, currently lz4, lzo and
zstd are supported.");
+
public static final ConfigOption<Boolean> WRITE_ONLY =
key("write-only")
.booleanType()
@@ -1271,6 +1278,10 @@ public class CoreOptions implements Serializable {
return options.get(SORT_SPILL_BUFFER_SIZE).getBytes();
}
+ public String spillCompression() {
+ return options.get(SPILL_COMPRESSION);
+ }
+
public Duration continuousDiscoveryInterval() {
return options.get(CONTINUOUS_DISCOVERY_INTERVAL);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/AirBlockCompressor.java
similarity index 63%
copy from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
copy to
paimon-common/src/main/java/org/apache/paimon/compression/AirBlockCompressor.java
index 434f6242e..ad235c0f0 100644
---
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/compression/AirBlockCompressor.java
@@ -18,37 +18,41 @@
package org.apache.paimon.compression;
-import net.jpountz.lz4.LZ4Compressor;
-import net.jpountz.lz4.LZ4Factory;
+import io.airlift.compress.Compressor;
import static org.apache.paimon.compression.CompressorUtils.HEADER_LENGTH;
import static org.apache.paimon.compression.CompressorUtils.writeIntLE;
-/**
- * Encode data into LZ4 format (not compatible with the LZ4 Frame format). It
reads from and writes
- * to byte arrays provided from the outside, thus reducing copy time.
- *
- * <p>This class is copied and modified from {@link
net.jpountz.lz4.LZ4BlockOutputStream}.
- */
-public class Lz4BlockCompressor implements BlockCompressor {
+/** Implementation of {@link BlockCompressor} for airlift compressors. */
+public class AirBlockCompressor implements BlockCompressor {
- private final LZ4Compressor compressor;
+ private final Compressor internalCompressor;
- public Lz4BlockCompressor() {
- this.compressor = LZ4Factory.fastestInstance().fastCompressor();
+ public AirBlockCompressor(Compressor internalCompressor) {
+ this.internalCompressor = internalCompressor;
}
@Override
public int getMaxCompressedSize(int srcSize) {
- return HEADER_LENGTH + compressor.maxCompressedLength(srcSize);
+ return HEADER_LENGTH + internalCompressor.maxCompressedLength(srcSize);
}
@Override
public int compress(byte[] src, int srcOff, int srcLen, byte[] dst, int
dstOff)
throws BufferCompressionException {
try {
+ if (dst.length < dstOff + getMaxCompressedSize(srcLen)) {
+ throw new ArrayIndexOutOfBoundsException();
+ }
+
int compressedLength =
- compressor.compress(src, srcOff, srcLen, dst, dstOff +
HEADER_LENGTH);
+ internalCompressor.compress(
+ src,
+ srcOff,
+ srcLen,
+ dst,
+ dstOff + HEADER_LENGTH,
+ internalCompressor.maxCompressedLength(srcLen));
writeIntLE(compressedLength, dst, dstOff);
writeIntLE(srcLen, dst, dstOff + 4);
return HEADER_LENGTH + compressedLength;
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/AirBlockDecompressor.java
similarity index 66%
copy from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
copy to
paimon-common/src/main/java/org/apache/paimon/compression/AirBlockDecompressor.java
index d707276e2..7af31a258 100644
---
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/compression/AirBlockDecompressor.java
@@ -18,33 +18,27 @@
package org.apache.paimon.compression;
-import net.jpountz.lz4.LZ4Exception;
-import net.jpountz.lz4.LZ4Factory;
-import net.jpountz.lz4.LZ4FastDecompressor;
+import io.airlift.compress.Decompressor;
+import io.airlift.compress.MalformedInputException;
import static org.apache.paimon.compression.CompressorUtils.HEADER_LENGTH;
import static org.apache.paimon.compression.CompressorUtils.readIntLE;
import static org.apache.paimon.compression.CompressorUtils.validateLength;
-/**
- * Decode data written with {@link Lz4BlockCompressor}. It reads from and
writes to byte arrays
- * provided from the outside, thus reducing copy time.
- *
- * <p>This class is copied and modified from {@link
net.jpountz.lz4.LZ4BlockInputStream}.
- */
-public class Lz4BlockDecompressor implements BlockDecompressor {
+/** Implementation of {@link BlockDecompressor} for airlift compressors. */
+public class AirBlockDecompressor implements BlockDecompressor {
- private final LZ4FastDecompressor decompressor;
+ private final Decompressor internalDecompressor;
- public Lz4BlockDecompressor() {
- this.decompressor = LZ4Factory.fastestInstance().fastDecompressor();
+ public AirBlockDecompressor(Decompressor internalDecompressor) {
+ this.internalDecompressor = internalDecompressor;
}
@Override
public int decompress(byte[] src, int srcOff, int srcLen, byte[] dst, int
dstOff)
throws BufferDecompressionException {
- final int compressedLen = readIntLE(src, srcOff);
- final int originalLen = readIntLE(src, srcOff + 4);
+ int compressedLen = readIntLE(src, srcOff);
+ int originalLen = readIntLE(src, srcOff + 4);
validateLength(compressedLen, originalLen);
if (dst.length - dstOff < originalLen) {
@@ -57,12 +51,13 @@ public class Lz4BlockDecompressor implements
BlockDecompressor {
}
try {
- final int compressedLen2 =
- decompressor.decompress(src, srcOff + HEADER_LENGTH, dst,
dstOff, originalLen);
- if (compressedLen != compressedLen2) {
+ final int decompressedLen =
+ internalDecompressor.decompress(
+ src, srcOff + HEADER_LENGTH, compressedLen, dst,
dstOff, originalLen);
+ if (originalLen != decompressedLen) {
throw new BufferDecompressionException("Input is corrupted");
}
- } catch (LZ4Exception e) {
+ } catch (MalformedInputException e) {
throw new BufferDecompressionException("Input is corrupted", e);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
b/paimon-common/src/main/java/org/apache/paimon/compression/AirCompressorFactory.java
similarity index 59%
copy from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
copy to
paimon-common/src/main/java/org/apache/paimon/compression/AirCompressorFactory.java
index 91be9e916..f22c9aad8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/compression/AirCompressorFactory.java
@@ -18,15 +18,27 @@
package org.apache.paimon.compression;
-/** Implementation of {@link BlockCompressionFactory} for Lz4 codec. */
-public class Lz4BlockCompressionFactory implements BlockCompressionFactory {
+import io.airlift.compress.Compressor;
+import io.airlift.compress.Decompressor;
+
+/** Implementation of {@link BlockCompressionFactory} for airlift compressors.
*/
+public class AirCompressorFactory implements BlockCompressionFactory {
+
+ private final Compressor internalCompressor;
+ private final Decompressor internalDecompressor;
+
+ public AirCompressorFactory(Compressor internalCompressor, Decompressor
internalDecompressor) {
+ this.internalCompressor = internalCompressor;
+ this.internalDecompressor = internalDecompressor;
+ }
+
@Override
public BlockCompressor getCompressor() {
- return new Lz4BlockCompressor();
+ return new AirBlockCompressor(internalCompressor);
}
@Override
public BlockDecompressor getDecompressor() {
- return new Lz4BlockDecompressor();
+ return new AirBlockDecompressor(internalDecompressor);
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
b/paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
similarity index 57%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
index 1c9df6155..40bb10fc4 100644
---
a/paimon-core/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
+++
b/paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressionFactory.java
@@ -18,6 +18,11 @@
package org.apache.paimon.compression;
+import io.airlift.compress.lzo.LzoCompressor;
+import io.airlift.compress.lzo.LzoDecompressor;
+import io.airlift.compress.zstd.ZstdCompressor;
+import io.airlift.compress.zstd.ZstdDecompressor;
+
/**
* Each compression codec has an implementation of {@link
BlockCompressionFactory} to create
* compressors and decompressors.
@@ -27,4 +32,18 @@ public interface BlockCompressionFactory {
BlockCompressor getCompressor();
BlockDecompressor getDecompressor();
+
+ /** Creates {@link BlockCompressionFactory} according to the
configuration. */
+ static BlockCompressionFactory create(String compression) {
+ switch (compression.toUpperCase()) {
+ case "LZ4":
+ return new Lz4BlockCompressionFactory();
+ case "LZO":
+ return new AirCompressorFactory(new LzoCompressor(), new
LzoDecompressor());
+ case "ZSTD":
+ return new AirCompressorFactory(new ZstdCompressor(), new
ZstdDecompressor());
+ default:
+ throw new IllegalStateException("Unknown CompressionMethod " +
compression);
+ }
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/BlockCompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressor.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/BlockCompressor.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/BlockCompressor.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/BlockDecompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/BlockDecompressor.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/BlockDecompressor.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/BlockDecompressor.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/BufferCompressionException.java
b/paimon-common/src/main/java/org/apache/paimon/compression/BufferCompressionException.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/BufferCompressionException.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/BufferCompressionException.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/BufferDecompressionException.java
b/paimon-common/src/main/java/org/apache/paimon/compression/BufferDecompressionException.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/BufferDecompressionException.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/BufferDecompressionException.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/CompressorUtils.java
b/paimon-common/src/main/java/org/apache/paimon/compression/CompressorUtils.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/CompressorUtils.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/CompressorUtils.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
b/paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockCompressionFactory.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockCompressor.java
diff --git
a/paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
b/paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
similarity index 100%
rename from
paimon-core/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
rename to
paimon-common/src/main/java/org/apache/paimon/compression/Lz4BlockDecompressor.java
diff --git
a/paimon-common/src/test/java/org/apache/paimon/compression/BlockCompressionTest.java
b/paimon-common/src/test/java/org/apache/paimon/compression/BlockCompressionTest.java
new file mode 100644
index 000000000..5ece1cb03
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/compression/BlockCompressionTest.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.compression;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.apache.paimon.compression.CompressorUtils.HEADER_LENGTH;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BlockCompressionTest {
+
+ private static Stream<String> compressCodecGenerator() {
+ return Stream.of("LZ4", "LZO", "ZSTD");
+ }
+
+ @ParameterizedTest
+ @MethodSource("compressCodecGenerator")
+ void testBlockCompression(String compress) {
+ BlockCompressionFactory factory =
BlockCompressionFactory.create(compress);
+ runTest(factory, 32768);
+ runTest(factory, 16);
+ }
+
+ private void runTest(BlockCompressionFactory factory, int originalLen) {
+ BlockCompressor compressor = factory.getCompressor();
+ BlockDecompressor decompressor = factory.getDecompressor();
+
+ int originalOff = 64;
+ byte[] data = new byte[originalOff + originalLen];
+ for (int i = 0; i < originalLen; i++) {
+ data[originalOff + i] = (byte) i;
+ }
+
+ int compressedOff = 32;
+
+ // 1. test compress with insufficient target
+ byte[] insufficientCompressArray = new byte[compressedOff +
HEADER_LENGTH + 1];
+ assertThatThrownBy(
+ () ->
+ compressor.compress(
+ data,
+ originalOff,
+ originalLen,
+ insufficientCompressArray,
+ compressedOff))
+ .isInstanceOf(BufferCompressionException.class);
+
+ // 2. test normal compress
+ byte[] compressedData =
+ new byte[compressedOff +
compressor.getMaxCompressedSize(originalLen)];
+ int compressedLen =
+ compressor.compress(data, originalOff, originalLen,
compressedData, compressedOff);
+
+ int decompressedOff = 16;
+
+ // 3. test decompress with insufficient target
+ byte[] insufficientDecompressArray = new byte[decompressedOff +
originalLen - 1];
+ assertThatThrownBy(
+ () ->
+ decompressor.decompress(
+ compressedData,
+ compressedOff,
+ compressedLen,
+ insufficientDecompressArray,
+ decompressedOff))
+ .isInstanceOf(BufferDecompressionException.class);
+
+ // 4. test normal decompress
+ byte[] decompressedData = new byte[decompressedOff + originalLen];
+ int decompressedLen =
+ decompressor.decompress(
+ compressedData,
+ compressedOff,
+ compressedLen,
+ decompressedData,
+ decompressedOff);
+ assertThat(decompressedLen).isEqualTo(originalLen);
+
+ for (int i = 0; i < originalLen; i++) {
+ assertThat(decompressedData[decompressedOff +
i]).isEqualTo(data[originalOff + i]);
+ }
+ }
+}
diff --git a/paimon-core/pom.xml b/paimon-core/pom.xml
index c66bc85dd..0a0effb15 100644
--- a/paimon-core/pom.xml
+++ b/paimon-core/pom.xml
@@ -56,12 +56,6 @@ under the License.
<!-- flink dependencies -->
- <dependency>
- <groupId>org.lz4</groupId>
- <artifactId>lz4-java</artifactId>
- <version>${lz4.version}</version>
- </dependency>
-
<dependency>
<groupId>com.ververica</groupId>
<artifactId>frocksdbjni</artifactId>
diff --git
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
index 5b4ee8a43..d4383d4c2 100644
---
a/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
+++
b/paimon-core/src/main/java/org/apache/paimon/crosspartition/GlobalIndexAssigner.java
@@ -289,7 +289,8 @@ public class GlobalIndexAssigner implements Serializable,
Closeable {
keyWithRowType,
coreOptions.writeBufferSize() / 2,
coreOptions.pageSize(),
- coreOptions.localSortMaxNumFileHandles());
+ coreOptions.localSortMaxNumFileHandles(),
+ coreOptions.spillCompression());
Function<SortOrder, RowIterator> iteratorFunction =
sortOrder -> {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
b/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
index e4f9ee1ae..5ffcaf146 100644
--- a/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
+++ b/paimon-core/src/main/java/org/apache/paimon/lookup/RocksDBState.java
@@ -111,7 +111,8 @@ public abstract class RocksDBState<K, V, CacheV> {
RowType.of(DataTypes.BYTES(), DataTypes.BYTES()),
options.writeBufferSize() / 2,
options.pageSize(),
- options.localSortMaxNumFileHandles());
+ options.localSortMaxNumFileHandles(),
+ options.spillCompression());
}
/** A class wraps byte[] to implement equals and hashCode. */
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
index 15f6cfb9d..26a6d7fc6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeSorter.java
@@ -66,6 +66,7 @@ public class MergeSorter {
private final SortEngine sortEngine;
private final int spillThreshold;
private final int spillSortMaxNumFiles;
+ private final String compression;
private final MemorySegmentPool memoryPool;
@@ -79,6 +80,7 @@ public class MergeSorter {
this.sortEngine = options.sortEngine();
this.spillThreshold = options.sortSpillThreshold();
this.spillSortMaxNumFiles = options.localSortMaxNumFileHandles();
+ this.compression = options.spillCompression();
this.keyType = keyType;
this.valueType = valueType;
this.memoryPool =
@@ -197,7 +199,8 @@ public class MergeSorter {
DataTypes.ROW(sortKeyTypes.toArray(new
DataType[0])),
new RowType(fields),
memoryPool,
- spillSortMaxNumFiles);
+ spillSortMaxNumFiles,
+ compression);
}
public boolean put(KeyValue keyValue) throws IOException {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
index 7d49fe167..055ffe7d5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
+++ b/paimon-core/src/main/java/org/apache/paimon/mergetree/MergeTreeWriter.java
@@ -55,6 +55,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
private final boolean writeBufferSpillable;
private final int sortMaxFan;
+ private final String sortCompression;
private final IOManager ioManager;
private final RowType keyType;
@@ -80,6 +81,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
public MergeTreeWriter(
boolean writeBufferSpillable,
int sortMaxFan,
+ String sortCompression,
IOManager ioManager,
CompactManager compactManager,
long maxSequenceNumber,
@@ -92,6 +94,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
WriterMetrics writerMetrics) {
this.writeBufferSpillable = writeBufferSpillable;
this.sortMaxFan = sortMaxFan;
+ this.sortCompression = sortCompression;
this.ioManager = ioManager;
this.keyType = writerFactory.keyType();
this.valueType = writerFactory.valueType();
@@ -139,6 +142,7 @@ public class MergeTreeWriter implements
RecordWriter<KeyValue>, MemoryOwner {
memoryPool,
writeBufferSpillable,
sortMaxFan,
+ sortCompression,
ioManager);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
index 53cc0969e..744c4f31a 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/mergetree/SortBufferWriteBuffer.java
@@ -63,6 +63,7 @@ public class SortBufferWriteBuffer implements WriteBuffer {
MemorySegmentPool memoryPool,
boolean spillable,
int sortMaxFan,
+ String compression,
IOManager ioManager) {
this.keyType = keyType;
this.valueType = valueType;
@@ -95,7 +96,8 @@ public class SortBufferWriteBuffer implements WriteBuffer {
memoryPool.pageSize(),
inMemorySortBuffer,
ioManager,
- sortMaxFan)
+ sortMaxFan,
+ compression)
: inMemorySortBuffer;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
index 53ae886a4..9ae8b4ccd 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/KeyValueFileStoreWrite.java
@@ -170,6 +170,7 @@ public class KeyValueFileStoreWrite extends
MemoryFileStoreWrite<KeyValue> {
return new MergeTreeWriter(
bufferSpillable(),
options.localSortMaxNumFileHandles(),
+ options.spillCompression(),
ioManager,
compactManager,
getMaxSequenceNumber(restoreFiles),
diff --git
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
index 4bfbbdb44..7d9522f61 100644
---
a/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
+++
b/paimon-core/src/main/java/org/apache/paimon/sort/BinaryExternalSortBuffer.java
@@ -21,7 +21,6 @@ package org.apache.paimon.sort;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.codegen.RecordComparator;
import org.apache.paimon.compression.BlockCompressionFactory;
-import org.apache.paimon.compression.Lz4BlockCompressionFactory;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.BinaryRowSerializer;
@@ -67,13 +66,14 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
int pageSize,
BinaryInMemorySortBuffer inMemorySortBuffer,
IOManager ioManager,
- int maxNumFileHandles) {
+ int maxNumFileHandles,
+ String compression) {
this.serializer = serializer;
this.inMemorySortBuffer = inMemorySortBuffer;
this.ioManager = ioManager;
this.channelManager = new SpillChannelManager();
this.maxNumFileHandles = maxNumFileHandles;
- this.compressionCodecFactory = new Lz4BlockCompressionFactory();
+ this.compressionCodecFactory =
BlockCompressionFactory.create(compression);
this.compressionBlockSize = (int) MemorySize.parse("64 kb").getBytes();
this.merger =
new BinaryExternalMerger(
@@ -95,13 +95,15 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
RowType rowType,
long bufferSize,
int pageSize,
- int maxNumFileHandles) {
+ int maxNumFileHandles,
+ String compression) {
return create(
ioManager,
keyType,
rowType,
new HeapMemorySegmentPool(bufferSize, pageSize),
- maxNumFileHandles);
+ maxNumFileHandles,
+ compression);
}
public static BinaryExternalSortBuffer create(
@@ -109,7 +111,8 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
RowType keyType,
RowType rowType,
MemorySegmentPool pool,
- int maxNumFileHandles) {
+ int maxNumFileHandles,
+ String compression) {
RecordComparator comparator =
newRecordComparator(keyType.getFieldTypes(),
"ExternalSort_comparator");
BinaryInMemorySortBuffer sortBuffer =
@@ -125,7 +128,8 @@ public class BinaryExternalSortBuffer implements SortBuffer
{
pool.pageSize(),
sortBuffer,
ioManager,
- maxNumFileHandles);
+ maxNumFileHandles,
+ compression);
}
@Override
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
index 34c0eb4d9..1aa708306 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/MergeTreeTestBase.java
@@ -416,6 +416,7 @@ public abstract class MergeTreeTestBase {
new MergeTreeWriter(
false,
128,
+ "lz4",
null,
compactManager,
maxSequenceNumber,
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
index b7d76ff9d..170a66779 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
+++
b/paimon-core/src/test/java/org/apache/paimon/mergetree/SortBufferWriteBufferTestBase.java
@@ -69,6 +69,7 @@ public abstract class SortBufferWriteBufferTestBase {
new HeapMemorySegmentPool(32 * 1024 * 3L, 32 * 1024),
false,
128,
+ "lz4",
null);
protected abstract boolean addOnly();
diff --git
a/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
b/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
index f97a94e7f..503a427d9 100644
---
a/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/sort/BinaryExternalSortBufferTest.java
@@ -280,7 +280,8 @@ public class BinaryExternalSortBufferTest {
MemorySegmentPool.DEFAULT_PAGE_SIZE,
inMemorySortBuffer,
ioManager,
- maxNumFileHandles);
+ maxNumFileHandles,
+ "lz4");
}
/** Mock reader for binary row. */
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
index 8a37b03bf..6ce55a9bd 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/LocalMergeOperator.java
@@ -123,6 +123,7 @@ public class LocalMergeOperator extends
AbstractStreamOperator<InternalRow>
options.localMergeBufferSize(),
options.pageSize()),
false,
options.localSortMaxNumFileHandles(),
+ options.spillCompression(),
null);
currentWatermark = Long.MIN_VALUE;
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
index f64cf044c..4272dc53e 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortOperator.java
@@ -41,6 +41,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
private final int pageSize;
private final int arity;
private final int spillSortMaxNumFiles;
+ private final String spillCompression;
private final int sinkParallelism;
private transient BinaryExternalSortBuffer buffer;
@@ -52,6 +53,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
long maxMemory,
int pageSize,
int spillSortMaxNumFiles,
+ String spillCompression,
int sinkParallelism) {
this.keyType = keyType;
this.rowType = rowType;
@@ -59,6 +61,7 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
this.pageSize = pageSize;
this.arity = rowType.getFieldCount();
this.spillSortMaxNumFiles = spillSortMaxNumFiles;
+ this.spillCompression = spillCompression;
this.sinkParallelism = sinkParallelism;
}
@@ -83,7 +86,13 @@ public class SortOperator extends
TableStreamOperator<InternalRow>
.getSpillingDirectoriesPaths());
buffer =
BinaryExternalSortBuffer.create(
- ioManager, keyType, rowType, maxMemory, pageSize,
spillSortMaxNumFiles);
+ ioManager,
+ keyType,
+ rowType,
+ maxMemory,
+ pageSize,
+ spillSortMaxNumFiles,
+ spillCompression);
}
@Override
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
index 4187700a2..4d049ff72 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sorter/SortUtils.java
@@ -161,6 +161,7 @@ public class SortUtils {
options.writeBufferSize(),
options.pageSize(),
options.localSortMaxNumFileHandles(),
+ options.spillCompression(),
sinkParallelism))
.setParallelism(sinkParallelism)
// remove the key column from every row
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
index c94a92dd7..78db46b42 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sorter/SortOperatorTest.java
@@ -62,6 +62,7 @@ public class SortOperatorTest {
MemorySize.parse("10 mb").getBytes(),
(int) MemorySize.parse("16 kb").getBytes(),
128,
+ "lz4",
1) {};
OneInputStreamOperatorTestHarness harness =
createTestHarness(sortOperator);
@@ -109,6 +110,7 @@ public class SortOperatorTest {
MemorySize.parse("10 mb").getBytes(),
(int) MemorySize.parse("16 kb").getBytes(),
128,
+ "lz4",
1) {};
OneInputStreamOperatorTestHarness harness =
createTestHarness(sortOperator);
harness.open();
diff --git a/pom.xml b/pom.xml
index b354accda..5235ba7ac 100644
--- a/pom.xml
+++ b/pom.xml
@@ -82,6 +82,7 @@ under the License.
<scala.version>2.12.15</scala.version>
<scala.binary.version>2.12</scala.binary.version>
<snappy.version>1.1.8.3</snappy.version>
+ <airlift.version>0.21</airlift.version>
<lz4.version>1.8.0</lz4.version>
<slf4j.version>1.7.32</slf4j.version>
<log4j.version>2.17.1</log4j.version>
