update aircompressor to 3.5

Author: rounakraj-sophos

parquet-hadoop/pom.xml

@@ -143,8 +143,8 @@
     </dependency>
     <dependency>
       <groupId>io.airlift</groupId>
-      <artifactId>aircompressor</artifactId>
-      <version>2.0.2</version>
+      <artifactId>aircompressor-v3</artifactId>
+      <version>3.5</version>
     </dependency>
     <dependency>
       <groupId>commons-pool</groupId>

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/Lz4RawCompressor.java

@@ -18,23 +18,28 @@
  */
 package org.apache.parquet.hadoop.codec;
 
-import io.airlift.compress.lz4.Lz4Compressor;
+import io.airlift.compress.v3.lz4.Lz4Compressor;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 
 public class Lz4RawCompressor extends NonBlockedCompressor {
 
-  private Lz4Compressor compressor = new Lz4Compressor();
+  private final Lz4Compressor compressor = Lz4Compressor.create();
 
   @Override
   protected int maxCompressedLength(int byteSize) {
-    return io.airlift.compress.lz4.Lz4RawCompressor.maxCompressedLength(byteSize);
+    return compressor.maxCompressedLength(byteSize);
   }
 
   @Override
   protected int compress(ByteBuffer uncompressed, ByteBuffer compressed) throws IOException {
-    compressor.compress(uncompressed, compressed);
-    int compressedSize = compressed.position();
+    int inputLen = uncompressed.remaining();
+    byte[] inputArr = new byte[inputLen];
+    uncompressed.get(inputArr);
+    int maxOut = compressor.maxCompressedLength(inputLen);
+    byte[] outputArr = new byte[maxOut];
+    int compressedSize = compressor.compress(inputArr, 0, inputLen, outputArr, 0, maxOut);
+    compressed.put(outputArr, 0, compressedSize);
     compressed.limit(compressedSize);
     compressed.rewind();
     return compressedSize;

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/Lz4RawDecompressor.java

@@ -18,14 +18,14 @@
  */
 package org.apache.parquet.hadoop.codec;
 
-import io.airlift.compress.lz4.Lz4Decompressor;
+import io.airlift.compress.v3.lz4.Lz4Decompressor;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import org.apache.hadoop.io.compress.DirectDecompressor;
 
 public class Lz4RawDecompressor extends NonBlockedDecompressor implements DirectDecompressor {
 
-  private Lz4Decompressor decompressor = new Lz4Decompressor();
+  private final Lz4Decompressor decompressor = Lz4Decompressor.create();
 
   @Override
   protected int maxUncompressedLength(ByteBuffer compressed, int maxUncompressedLength) throws IOException {
@@ -36,8 +36,14 @@ protected int maxUncompressedLength(ByteBuffer compressed, int maxUncompressedLe
 
   @Override
   protected int uncompress(ByteBuffer compressed, ByteBuffer uncompressed) throws IOException {
-    decompressor.decompress(compressed, uncompressed);
-    int uncompressedSize = uncompressed.position();
+    int compressedLen = compressed.remaining();
+    byte[] inputArr = new byte[compressedLen];
+    compressed.get(inputArr);
+    int maxOut = uncompressed.remaining();
+    byte[] outputArr = new byte[maxOut];
+
+    int uncompressedSize = decompressor.decompress(inputArr, 0, compressedLen, outputArr, 0, maxOut);
+    uncompressed.put(outputArr, 0, uncompressedSize);
     uncompressed.limit(uncompressedSize);
     uncompressed.rewind();
     return uncompressedSize;