ORC-363 Enable zstd decompression in ORC Java reader

java/core/src/java/org/apache/orc/CompressionKind.java

@@ -23,5 +23,5 @@
  * can be applied to ORC files.
  */
 public enum CompressionKind {
-  NONE, ZLIB, SNAPPY, LZO, LZ4
+  NONE, ZLIB, SNAPPY, LZO, LZ4, ZSTD
 }

java/core/src/java/org/apache/orc/OrcFile.java

@@ -423,6 +423,11 @@ protected WriterOptions(Properties tableProperties, Configuration conf) {
       compressValue =
           CompressionKind.valueOf(OrcConf.COMPRESS.getString(tableProperties,
               conf).toUpperCase());
+      // Zstd compression is not supported currently
+      // Will enable this once it's released through aircompressor lib
+      if (compressValue == CompressionKind.ZSTD) {
+          throw new IllegalArgumentException("Zstd compressor is not supported.");
+      }
       enforceBufferSize = OrcConf.ENFORCE_COMPRESSION_BUFFER_SIZE.getBoolean(tableProperties, conf);
       String versionName = OrcConf.WRITE_FORMAT.getString(tableProperties,
           conf);
@@ -581,6 +586,11 @@ public WriterOptions bloomFilterFpp(double fpp) {
      * Sets the generic compression that is used to compress the data.
      */
     public WriterOptions compress(CompressionKind value) {
+      // Zstd compression is not supported currently
+      // Will enable this once it's released through aircompressor lib
+      if (value == CompressionKind.ZSTD) {
+          throw new IllegalArgumentException("Zstd compressor is not supported.");
+      }
       compressValue = value;
       return this;
     }

java/core/src/java/org/apache/orc/impl/ReaderImpl.java

@@ -443,6 +443,7 @@ private static OrcProto.PostScript extractPostScript(ByteBuffer bb, Path path,
       case SNAPPY:
       case LZO:
       case LZ4:
+      case ZSTD:
         break;
       default:
         throw new IllegalArgumentException("Unknown compression");

java/core/src/java/org/apache/orc/impl/WriterImpl.java

@@ -31,6 +31,7 @@
 import io.airlift.compress.lz4.Lz4Decompressor;
 import io.airlift.compress.lzo.LzoCompressor;
 import io.airlift.compress.lzo.LzoDecompressor;
+import io.airlift.compress.zstd.ZstdDecompressor;
 import org.apache.orc.ColumnStatistics;
 import org.apache.orc.CompressionCodec;
 import org.apache.orc.CompressionKind;
@@ -241,6 +242,10 @@ public static CompressionCodec createCodec(CompressionKind kind) {
       case LZ4:
         return new AircompressorCodec(new Lz4Compressor(),
             new Lz4Decompressor());
+      case ZSTD:
+        // Zstd compressor is not availiable currently
+        // Will add it back after it's released
+        return new AircompressorCodec(null, new ZstdDecompressor());
       default:
         throw new IllegalArgumentException("Unknown compression codec: " +
             kind);

java/core/src/test/org/apache/orc/TestVectorOrcFile.java

@@ -371,6 +371,49 @@ public void testReadFormat_0_11() throws Exception {
     rows.close();
   }
 
+  @Test
+  public void testReadZstd() throws Exception {
+    Path filePath =
+        new Path(getFileFromClasspath("orc-file-zstd.orc"));
+    Reader reader = OrcFile.createReader(filePath,
+        OrcFile.readerOptions(conf).filesystem(fs));
+
+    int stripeCount = 0;
+    int rowCount = 0;
+    long currentOffset = -1;
+    for(StripeInformation stripe : reader.getStripes()) {
+      stripeCount += 1;
+      rowCount += stripe.getNumberOfRows();
+    }
+    Assert.assertEquals(reader.getNumberOfRows(), rowCount);
+    assertEquals(1, stripeCount);
+
+    // check schema and read file contents
+    TypeDescription schema = reader.getSchema();
+    assertEquals(TypeDescription.Category.STRUCT, schema.getCategory());
+    assertEquals(
+                 "struct<c1:bigint,c2:string,c3:bigint,c4:string>",
+                 schema.toString());
+    VectorizedRowBatch batch = schema.createRowBatch();
+    RecordReader rows = reader.rows();
+    Assert.assertEquals(true, rows.nextBatch(batch));
+    assertEquals(25, batch.size);
+
+    // check the contents of the first row
+    assertEquals(0, ((LongColumnVector)batch.cols[0]).vector[0]);
+    assertEquals(
+                 "ALGERIA",
+                 getText((BytesColumnVector)batch.cols[1], 0).toString());
+    assertEquals(0, ((LongColumnVector)batch.cols[2]).vector[0]);
+    assertEquals(
+                 " haggle. carefully final deposits detect slyly agai",
+                 getText((BytesColumnVector)batch.cols[3], 0).toString());
+
+    // handle the close up
+    Assert.assertEquals(false, rows.nextBatch(batch));
+    rows.close();
+  }
+
   @Test
   public void testTimestamp() throws Exception {
     TypeDescription schema = TypeDescription.createTimestamp();

java/core/src/test/org/apache/orc/impl/TestWriterImpl.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.fs.FileAlreadyExistsException;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.orc.CompressionKind;
 import org.apache.orc.OrcConf;
 import org.apache.orc.OrcFile;
 import org.apache.orc.TypeDescription;
@@ -70,4 +71,16 @@ public void testOverriddenOverwriteFlagForWriter() throws Exception {
     Writer w = OrcFile.createWriter(testFilePath, OrcFile.writerOptions(conf).setSchema(schema));
     w.close();
   }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testZstdConfThrowsException() throws Exception {
+    conf.set(OrcConf.COMPRESS.getAttribute(), "ZSTD");
+    Writer w = OrcFile.createWriter(testFilePath, OrcFile.writerOptions(conf).setSchema(schema));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testZstdWriterOptionThrowsException() throws Exception {
+    OrcFile.WriterOptions options = OrcFile.writerOptions(conf);
+    options.compress(CompressionKind.ZSTD);
+  }
 }

java/pom.xml

@@ -415,7 +415,7 @@
       <dependency>
         <groupId>io.airlift</groupId>
         <artifactId>aircompressor</artifactId>
-        <version>0.10</version>
+        <version>0.11</version>
         <exclusions>
           <exclusion>
 	    <groupId>io.airlift</groupId>