This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 14124f31626 Use fast LZ4 for .tar.lz4 archives (#19946)
14124f31626 is described below
commit 14124f316266b45f06f1b024eb812850209aceba
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Oct 19 18:50:40 2026 -0700
Use fast LZ4 for .tar.lz4 archives (#19946)
* Speed up LZ4 archive creation and buffer extraction
LZ4 archives with many short matches caused slow creation and tiny
extraction writes.
Use the existing frame encoder and buffer normal extraction; a 157 MB
archive reproduces the bottleneck.
* Add named LZ4 implementations through Commons Compress
Register fast, high-compression, and legacy encoders with the compressor
factory. Keep tar dispatch generic and preserve existing LZ4 defaults.
* Isolate mock server directories across parallel query tests
CI runs QueryRunnerTest and QueryTerminatedBeforeStageStartsTest in
separate forks with the same server and table names. Give each factory a UUID
namespace so one setup cannot delete another test’s segment directory.
* Use fast LZ4 for .tar.lz4 archives
Keep gzip as the generic archive default and retain the Commons encoder as
lz4-framed. Remove the redundant lz4-default alias and simplify the provider to
fast and HC encoders.
---
.../compression/Lz4CompressorStreamProvider.java | 138 +++++++++++++++++++++
.../pinot/common/utils/TarCompressionUtils.java | 14 ++-
...s.compress.compressors.CompressorStreamProvider | 18 +++
.../common/utils/TarCompressionUtilsTest.java | 72 +++++++++++
.../testutils/MockInstanceDataManagerFactory.java | 3 +-
5 files changed, 239 insertions(+), 6 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/compression/Lz4CompressorStreamProvider.java
b/pinot-common/src/main/java/org/apache/pinot/common/compression/Lz4CompressorStreamProvider.java
new file mode 100644
index 00000000000..a4cd3915527
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/compression/Lz4CompressorStreamProvider.java
@@ -0,0 +1,138 @@
+/**
+ * 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.pinot.common.compression;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Locale;
+import java.util.Set;
+import net.jpountz.lz4.LZ4Factory;
+import net.jpountz.lz4.LZ4FrameOutputStream;
+import net.jpountz.lz4.LZ4FrameOutputStream.BLOCKSIZE;
+import net.jpountz.lz4.LZ4FrameOutputStream.FLG.Bits;
+import net.jpountz.xxhash.XXHashFactory;
+import org.apache.commons.compress.compressors.CompressorException;
+import org.apache.commons.compress.compressors.CompressorInputStream;
+import org.apache.commons.compress.compressors.CompressorOutputStream;
+import org.apache.commons.compress.compressors.CompressorStreamFactory;
+import org.apache.commons.compress.compressors.CompressorStreamProvider;
+import org.apache.commons.io.output.CloseShieldOutputStream;
+
+
+/// Adds explicit LZ4 frame encoders to the Commons Compress factory. This
provider is stateless and thread-safe;
+/// returned streams are not thread-safe. All variants use the Commons reader
for existing frame compatibility.
+public class Lz4CompressorStreamProvider implements CompressorStreamProvider {
+ public static final String LZ4_FAST = "lz4-fast";
+ public static final String LZ4_HC = "lz4-hc";
+ private static final Set<String> NAMES = Set.of(LZ4_FAST, LZ4_HC);
+
+ @Override
+ public CompressorInputStream createCompressorInputStream(String name,
InputStream in, boolean decompressConcatenated)
+ throws CompressorException {
+ validateName(name);
+ return CompressorStreamFactory.getSingleton()
+ .createCompressorInputStream(CompressorStreamFactory.LZ4_FRAMED, in,
decompressConcatenated);
+ }
+
+ @Override
+ public <T extends OutputStream> CompressorOutputStream<T>
createCompressorOutputStream(String name, T out)
+ throws CompressorException {
+ validateName(name);
+ try {
+ return new Lz4OutputStream<>(out, name);
+ } catch (IOException e) {
+ throw new CompressorException("Could not create LZ4 output stream", e);
+ }
+ }
+
+ @Override
+ public Set<String> getInputStreamCompressorNames() {
+ return NAMES;
+ }
+
+ @Override
+ public Set<String> getOutputStreamCompressorNames() {
+ return NAMES;
+ }
+
+ private static void validateName(String name)
+ throws CompressorException {
+ if (name == null || !NAMES.contains(name.toLowerCase(Locale.ROOT))) {
+ throw new CompressorException("Unsupported LZ4 compressor: " + name);
+ }
+ }
+
+ /// Adapts frame encoders to the Commons stream lifecycle, including finish
without closing the caller's output.
+ private static class Lz4OutputStream<T extends OutputStream> extends
CompressorOutputStream<T> {
+ private final LZ4FrameOutputStream _frameOut;
+
+ private Lz4OutputStream(T out, String name)
+ throws IOException {
+ super(out);
+ LZ4Factory factory = LZ4Factory.fastestInstance();
+ _frameOut = new LZ4FrameOutputStream(CloseShieldOutputStream.wrap(out),
BLOCKSIZE.SIZE_4MB, -1,
+ LZ4_HC.equalsIgnoreCase(name) ? factory.highCompressor() :
factory.fastCompressor(),
+ XXHashFactory.fastestInstance().hash32(), Bits.BLOCK_INDEPENDENCE,
Bits.CONTENT_CHECKSUM);
+ }
+
+ @Override
+ public void write(int value)
+ throws IOException {
+ checkOpen();
+ _frameOut.write(value);
+ }
+
+ @Override
+ public void write(byte[] bytes, int offset, int length)
+ throws IOException {
+ checkOpen();
+ _frameOut.write(bytes, offset, length);
+ }
+
+ @Override
+ public void flush()
+ throws IOException {
+ if (isFinished()) {
+ out().flush();
+ } else {
+ _frameOut.flush();
+ }
+ }
+
+ @Override
+ public void finish()
+ throws IOException {
+ if (!isFinished()) {
+ _frameOut.close();
+ super.finish();
+ }
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ try {
+ finish();
+ } finally {
+ super.close();
+ }
+ }
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/TarCompressionUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/TarCompressionUtils.java
index 63e0c092c83..5a48f17c6bf 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/TarCompressionUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/TarCompressionUtils.java
@@ -42,6 +42,7 @@ import
org.apache.commons.compress.compressors.CompressorException;
import org.apache.commons.compress.compressors.CompressorStreamFactory;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pinot.common.compression.Lz4CompressorStreamProvider;
/// Utility class to compress/de-compress tar files compressed using various
compressors.
@@ -76,7 +77,7 @@ public class TarCompressionUtils {
public static final String TAR_ZST_FILE_EXTENSION = ".tar.zst";
public static final Map<String, String> COMPRESSOR_NAME_BY_FILE_EXTENSIONS =
Map.of(TAR_GZ_FILE_EXTENSION, CompressorStreamFactory.GZIP,
TAR_LZ4_FILE_EXTENSION,
- CompressorStreamFactory.LZ4_FRAMED, TAR_ZST_FILE_EXTENSION,
CompressorStreamFactory.ZSTANDARD);
+ Lz4CompressorStreamProvider.LZ4_FAST, TAR_ZST_FILE_EXTENSION,
CompressorStreamFactory.ZSTANDARD);
private static final CompressorStreamFactory COMPRESSOR_STREAM_FACTORY =
CompressorStreamFactory.getSingleton();
private static final char ENTRY_NAME_SEPARATOR = '/';
private static String _defaultCompressorName = CompressorStreamFactory.GZIP;
@@ -123,8 +124,9 @@ public class TarCompressionUtils {
throws IOException {
try (OutputStream fileOut = Files.newOutputStream(outputFile.toPath());
BufferedOutputStream bufferedOut = new BufferedOutputStream(fileOut);
- OutputStream compressorOut =
COMPRESSOR_STREAM_FACTORY.createCompressorOutputStream(compressorName,
- bufferedOut); TarArchiveOutputStream tarOut = new
TarArchiveOutputStream(compressorOut)) {
+ OutputStream compressorOut =
+
COMPRESSOR_STREAM_FACTORY.createCompressorOutputStream(compressorName,
bufferedOut);
+ TarArchiveOutputStream tarOut = new
TarArchiveOutputStream(compressorOut)) {
tarOut.setBigNumberMode(TarArchiveOutputStream.BIGNUMBER_STAR);
tarOut.setLongFileMode(TarArchiveOutputStream.LONGFILE_GNU);
@@ -230,7 +232,9 @@ public class TarCompressionUtils {
if (maxStreamRateInByte != NO_DISK_WRITE_RATE_LIMIT) {
copyWithRateLimiter(tarIn, out, maxStreamRateInByte);
} else {
- IOUtils.copy(tarIn, out);
+ try (OutputStream bufferedOut = new BufferedOutputStream(out)) {
+ IOUtils.copy(tarIn, bufferedOut);
+ }
}
}
}
@@ -255,7 +259,7 @@ public class TarCompressionUtils {
String entryName = entry.getName();
String[] parts = StringUtils.split(entryName, ENTRY_NAME_SEPARATOR);
if (parts.length > 0 && parts[parts.length - 1].equals(fileName)) {
- try (OutputStream out =
Files.newOutputStream(outputFile.toPath())) {
+ try (OutputStream out = new
BufferedOutputStream(Files.newOutputStream(outputFile.toPath()))) {
IOUtils.copy(tarIn, out);
}
return;
diff --git
a/pinot-common/src/main/resources/META-INF/services/org.apache.commons.compress.compressors.CompressorStreamProvider
b/pinot-common/src/main/resources/META-INF/services/org.apache.commons.compress.compressors.CompressorStreamProvider
new file mode 100644
index 00000000000..b9242e98ec8
--- /dev/null
+++
b/pinot-common/src/main/resources/META-INF/services/org.apache.commons.compress.compressors.CompressorStreamProvider
@@ -0,0 +1,18 @@
+# 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.
+
+org.apache.pinot.common.compression.Lz4CompressorStreamProvider
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/utils/TarCompressionUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/utils/TarCompressionUtilsTest.java
index 6bcdaa0a5bc..8941168644f 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/utils/TarCompressionUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/utils/TarCompressionUtilsTest.java
@@ -18,26 +18,37 @@
*/
package org.apache.pinot.common.utils;
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.Random;
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
import org.apache.commons.compress.compressors.CompressorException;
+import org.apache.commons.compress.compressors.CompressorOutputStream;
import org.apache.commons.compress.compressors.CompressorStreamFactory;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static
org.apache.pinot.common.compression.Lz4CompressorStreamProvider.LZ4_FAST;
+import static
org.apache.pinot.common.compression.Lz4CompressorStreamProvider.LZ4_HC;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
@@ -69,6 +80,7 @@ public class TarCompressionUtilsTest {
for (String compressedTarFileExtension :
TarCompressionUtils.COMPRESSOR_NAME_BY_FILE_EXTENSIONS.keySet()) {
testFile(compressedTarFileExtension);
}
+ testFile(TarCompressionUtils.TAR_COMPRESSED_FILE_EXTENSION);
}
public void testFile(String compressedTarFileExtension)
@@ -80,6 +92,11 @@ public class TarCompressionUtilsTest {
File compressedTarFile = new File(TAR_DIR, fileName +
compressedTarFileExtension);
TarCompressionUtils.createCompressedTarFile(dataFile, compressedTarFile);
+ if
(compressedTarFileExtension.equals(TarCompressionUtils.TAR_COMPRESSED_FILE_EXTENSION))
{
+ try (InputStream input = new
BufferedInputStream(Files.newInputStream(compressedTarFile.toPath()))) {
+ assertEquals(CompressorStreamFactory.detect(input),
CompressorStreamFactory.GZIP);
+ }
+ }
List<File> untarredFiles = TarCompressionUtils.untar(compressedTarFile,
UNTAR_DIR);
assertEquals(untarredFiles.size(), 1);
@@ -92,6 +109,61 @@ public class TarCompressionUtilsTest {
assertEquals(FileUtils.readFileToString(untarredFile,
StandardCharsets.UTF_8), fileContent);
}
+ @DataProvider(name = "lz4Implementations")
+ public Object[][] lz4Implementations() {
+ return new Object[][]{{CompressorStreamFactory.LZ4_FRAMED}, {LZ4_FAST},
{LZ4_HC}};
+ }
+
+ @Test(dataProvider = "lz4Implementations")
+ public void testLz4ArchiveRoundTrip(String compressorName)
+ throws IOException {
+ int contentSize =
compressorName.equals(CompressorStreamFactory.LZ4_FRAMED) ? 32 * 1024 : 5 *
1024 * 1024;
+ byte[] content = new byte[contentSize];
+ new Random(0).nextBytes(content);
+ Arrays.fill(content, content.length / 5, content.length * 3 / 5, (byte)
42);
+ File inputFile = new File(DATA_DIR, "data");
+ Files.write(inputFile.toPath(), content);
+ File archive = new File(TAR_DIR, "data" +
TarCompressionUtils.TAR_LZ4_FILE_EXTENSION);
+ TarCompressionUtils.createCompressedTarFile(inputFile, archive,
compressorName);
+ if (compressorName.equals(LZ4_FAST)) {
+ File defaultArchive = new File(TAR_DIR, "default" +
TarCompressionUtils.TAR_LZ4_FILE_EXTENSION);
+ TarCompressionUtils.createCompressedTarFile(inputFile, defaultArchive);
+ assertEquals(Files.readAllBytes(defaultArchive.toPath()),
Files.readAllBytes(archive.toPath()));
+ }
+
+ List<File> extractedFiles = TarCompressionUtils.untar(archive, UNTAR_DIR);
+ assertEquals(extractedFiles.size(), 1);
+ assertEquals(Files.readAllBytes(extractedFiles.get(0).toPath()), content);
+ File extractedFile = new File(UNTAR_DIR, "single");
+ TarCompressionUtils.untarOneFile(archive, inputFile.getName(),
extractedFile);
+ assertEquals(Files.readAllBytes(extractedFile.toPath()), content);
+ }
+
+ @Test
+ public void testLz4FactoryFinish()
+ throws IOException, CompressorException {
+ boolean[] closed = {false};
+ ByteArrayOutputStream sink = new ByteArrayOutputStream() {
+ @Override
+ public void close() {
+ closed[0] = true;
+ }
+ };
+ byte[] content = "factory stream".getBytes(StandardCharsets.UTF_8);
+ try (CompressorOutputStream<?> output =
+ COMPRESSOR_STREAM_FACTORY.createCompressorOutputStream(LZ4_FAST,
sink)) {
+ output.write(content[0]);
+ output.write(content, 1, content.length - 1);
+ output.finish();
+ assertFalse(closed[0]);
+ try (InputStream input =
COMPRESSOR_STREAM_FACTORY.createCompressorInputStream(LZ4_FAST,
+ new ByteArrayInputStream(sink.toByteArray()))) {
+ assertEquals(IOUtils.toByteArray(input), content);
+ }
+ }
+ assertTrue(closed[0]);
+ }
+
@Test
public void testDirectories()
throws IOException, CompressorException {
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
index f3fb219e8c6..990bb42cbdd 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/testutils/MockInstanceDataManagerFactory.java
@@ -72,7 +72,8 @@ public class MockInstanceDataManagerFactory {
private final String _serverName;
public MockInstanceDataManagerFactory(String serverName) {
- _serverName = serverName;
+ // Parallel tests can use the same logical server name without sharing
their segment directories.
+ _serverName = serverName + "_" + UUID.randomUUID();
_serverTableDataDirMap = new HashMap<>();
_tableSegmentMap = new HashMap<>();
_tableSegmentNameMap = new HashMap<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
