Try native (de)compression

trinodb · Jun 27, 2024 · 057b5e9 · 057b5e9
1 parent 8947d40
commit 057b5e9
Show file tree

Hide file tree

Showing 22 changed files with 178 additions and 122 deletions.
diff --git a/core/trino-main/pom.xml b/core/trino-main/pom.xml
@@ -84,11 +84,6 @@
             <artifactId>failsafe</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>aircompressor</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>bootstrap</artifactId>
@@ -375,6 +370,11 @@
             <artifactId>jmxutils</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>pl.wendigo</groupId>
+            <artifactId>aircompressor</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.assertj</groupId>
             <artifactId>assertj-core</artifactId>

diff --git a/lib/trino-hive-formats/pom.xml b/lib/trino-hive-formats/pom.xml
@@ -43,12 +43,6 @@
             <artifactId>commons-codec</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>aircompressor</artifactId>
-            <optional>true</optional>
-        </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>log</artifactId>
@@ -99,6 +93,12 @@
             <artifactId>modernizer-maven-annotations</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>pl.wendigo</groupId>
+            <artifactId>aircompressor</artifactId>
+            <optional>true</optional>
+        </dependency>
+
         <dependency>
             <groupId>org.jetbrains</groupId>
             <artifactId>annotations</artifactId>

diff --git a/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/Codec.java b/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/Codec.java
@@ -13,7 +13,7 @@
  */
 package io.trino.hive.formats.compression;
 
-import io.airlift.compress.hadoop.HadoopStreams;
+import io.airlift.compressor.hadoop.HadoopStreams;
 
 import java.io.IOException;
 import java.io.InputStream;

diff --git a/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/CompressionKind.java b/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/CompressionKind.java
@@ -15,15 +15,15 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
-import io.airlift.compress.bzip2.BZip2HadoopStreams;
-import io.airlift.compress.deflate.JdkDeflateHadoopStreams;
-import io.airlift.compress.gzip.JdkGzipHadoopStreams;
-import io.airlift.compress.hadoop.HadoopStreams;
-import io.airlift.compress.lz4.Lz4HadoopStreams;
-import io.airlift.compress.lzo.LzoHadoopStreams;
-import io.airlift.compress.lzo.LzopHadoopStreams;
-import io.airlift.compress.snappy.SnappyHadoopStreams;
-import io.airlift.compress.zstd.ZstdHadoopStreams;
+import io.airlift.compressor.bzip2.BZip2HadoopStreams;
+import io.airlift.compressor.deflate.JdkDeflateHadoopStreams;
+import io.airlift.compressor.gzip.JdkGzipHadoopStreams;
+import io.airlift.compressor.hadoop.HadoopStreams;
+import io.airlift.compressor.lz4.Lz4HadoopStreams;
+import io.airlift.compressor.lzo.LzoHadoopStreams;
+import io.airlift.compressor.lzo.LzopHadoopStreams;
+import io.airlift.compressor.snappy.SnappyHadoopStreams;
+import io.airlift.compressor.zstd.ZstdHadoopStreams;
 
 import java.util.Arrays;
 import java.util.List;

diff --git a/...-formats/src/main/java/io/trino/hive/formats/compression/MemoryCompressedSliceOutput.java b/...-formats/src/main/java/io/trino/hive/formats/compression/MemoryCompressedSliceOutput.java
@@ -13,7 +13,7 @@
  */
 package io.trino.hive.formats.compression;
 
-import io.airlift.compress.hadoop.HadoopStreams;
+import io.airlift.compressor.hadoop.HadoopStreams;
 import io.airlift.slice.Slice;
 import io.trino.plugin.base.io.ChunkedSliceOutput;
 

diff --git a/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/ValueCompressor.java b/lib/trino-hive-formats/src/main/java/io/trino/hive/formats/compression/ValueCompressor.java
@@ -13,7 +13,7 @@
  */
 package io.trino.hive.formats.compression;
 
-import io.airlift.compress.hadoop.HadoopStreams;
+import io.airlift.compressor.hadoop.HadoopStreams;
 import io.airlift.slice.DynamicSliceOutput;
 import io.airlift.slice.Slice;
 

diff --git a/...trino-hive-formats/src/main/java/io/trino/hive/formats/compression/ValueDecompressor.java b/...trino-hive-formats/src/main/java/io/trino/hive/formats/compression/ValueDecompressor.java
@@ -13,7 +13,7 @@
  */
 package io.trino.hive.formats.compression;
 
-import io.airlift.compress.hadoop.HadoopStreams;
+import io.airlift.compressor.hadoop.HadoopStreams;
 import io.airlift.slice.Slice;
 
 import java.io.IOException;

diff --git a/lib/trino-hive-formats/src/test/java/com/hadoop/compression/lzo/LzopCodec.java b/lib/trino-hive-formats/src/test/java/com/hadoop/compression/lzo/LzopCodec.java
@@ -14,4 +14,4 @@
 package com.hadoop.compression.lzo;
 
 public class LzopCodec
-        extends io.airlift.compress.lzo.LzopCodec {}
+        extends io.airlift.compressor.lzo.LzopCodec {}
diff --git a/lib/trino-hive-formats/src/test/java/org/apache/hadoop/io/compress/LzoCodec.java b/lib/trino-hive-formats/src/test/java/org/apache/hadoop/io/compress/LzoCodec.java
@@ -14,5 +14,5 @@
 package org.apache.hadoop.io.compress;
 
 public class LzoCodec
-        extends io.airlift.compress.lzo.LzoCodec
+        extends io.airlift.compressor.lzo.LzoCodec
 {}
diff --git a/lib/trino-orc/pom.xml b/lib/trino-orc/pom.xml
@@ -28,11 +28,6 @@
             <artifactId>guava</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>aircompressor</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>log</artifactId>
@@ -104,6 +99,11 @@
             <artifactId>jmxutils</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>pl.wendigo</groupId>
+            <artifactId>aircompressor</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>org.jetbrains</groupId>
             <artifactId>annotations</artifactId>
@@ -156,6 +156,12 @@
             <groupId>io.trino.hive</groupId>
             <artifactId>hive-apache</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

diff --git a/lib/trino-parquet/pom.xml b/lib/trino-parquet/pom.xml
@@ -24,11 +24,6 @@
             <artifactId>guava</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>aircompressor</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>log</artifactId>
@@ -95,6 +90,11 @@
             </exclusions>
         </dependency>
 
+        <dependency>
+            <groupId>pl.wendigo</groupId>
+            <artifactId>aircompressor</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>io.trino</groupId>
             <artifactId>trino-spi</artifactId>

diff --git a/lib/trino-parquet/src/test/java/io/trino/parquet/reader/TestPageReader.java b/lib/trino-parquet/src/test/java/io/trino/parquet/reader/TestPageReader.java
@@ -16,7 +16,6 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import io.airlift.compress.snappy.SnappyCompressor;
-import io.airlift.compress.snappy.SnappyRawCompressor;
 import io.airlift.slice.Slice;
 import io.airlift.slice.Slices;
 import io.trino.parquet.DataPage;
@@ -376,8 +375,9 @@ private static byte[] compress(CompressionCodec compressionCodec, byte[] bytes,
             return Arrays.copyOfRange(bytes, offset, offset + length);
         }
         if (compressionCodec == SNAPPY) {
-            byte[] out = new byte[SnappyRawCompressor.maxCompressedLength(length)];
-            int compressedSize = new SnappyCompressor().compress(bytes, offset, length, out, 0, out.length);
+            SnappyCompressor compressor = new SnappyCompressor();
+            byte[] out = new byte[compressor.maxCompressedLength(length)];
+            int compressedSize = compressor.compress(bytes, offset, length, out, 0, out.length);
             return Arrays.copyOf(out, compressedSize);
         }
         throw new IllegalArgumentException("unsupported compression code " + compressionCodec);

diff --git a/plugin/trino-hive/pom.xml b/plugin/trino-hive/pom.xml
@@ -59,11 +59,6 @@
             <artifactId>failsafe</artifactId>
         </dependency>
 
-        <dependency>
-            <groupId>io.airlift</groupId>
-            <artifactId>aircompressor</artifactId>
-        </dependency>
-
         <dependency>
             <groupId>io.airlift</groupId>
             <artifactId>bootstrap</artifactId>
@@ -245,6 +240,11 @@
             <artifactId>jmxutils</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>pl.wendigo</groupId>
+            <artifactId>aircompressor</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>apache-client</artifactId>
@@ -518,6 +518,12 @@
             <groupId>io.trino.hive</groupId>
             <artifactId>hive-apache</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -530,6 +536,12 @@
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

diff --git a/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestGlueConverter.java b/plugin/trino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/TestGlueConverter.java
@@ -220,7 +220,7 @@ void testToGlueFunctionInput()
                 .build();
         LanguageFunction actual = GlueConverter.fromGlueFunction(function);
 
-        assertThat(input.resourceUris().size()).isEqualTo(4);
+        assertThat(input.resourceUris().size()).isEqualTo(3);
         assertThat(actual).isEqualTo(expected);
 
         // verify that the owner comes from the metastore

diff --git a/...ino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/v1/TestGlueInputConverter.java b/...ino-hive/src/test/java/io/trino/plugin/hive/metastore/glue/v1/TestGlueInputConverter.java
@@ -100,7 +100,7 @@ public void testConvertFunction()
                 .withResourceUris(input.getResourceUris());
         LanguageFunction actual = GlueToTrinoConverter.convertFunction(function);
 
-        assertThat(input.getResourceUris().size()).isEqualTo(4);
+        assertThat(input.getResourceUris().size()).isEqualTo(3);
         assertThat(actual).isEqualTo(expected);
 
         // verify that the owner comes from the metastore

diff --git a/plugin/trino-hudi/pom.xml b/plugin/trino-hudi/pom.xml
@@ -359,6 +359,12 @@
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>

diff --git a/plugin/trino-iceberg/pom.xml b/plugin/trino-iceberg/pom.xml
@@ -211,6 +211,12 @@
         <dependency>
             <groupId>org.apache.iceberg</groupId>
             <artifactId>iceberg-core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -563,6 +569,12 @@
             <groupId>io.trino.hive</groupId>
             <artifactId>hive-apache</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
@@ -576,12 +588,24 @@
             <artifactId>iceberg-core</artifactId>
             <classifier>tests</classifier>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>
             <groupId>org.apache.parquet</groupId>
             <artifactId>parquet-hadoop</artifactId>
             <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.airlift</groupId>
+                    <artifactId>aircompressor</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
 
         <dependency>