ORC-817, ORC-1088: Support ZStandard compression using zstd-jni

java/core/pom.xml

@@ -51,6 +51,10 @@
       <groupId>io.airlift</groupId>
       <artifactId>aircompressor</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.github.luben</groupId>
+      <artifactId>zstd-jni</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client-api</artifactId>

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

@@ -72,6 +72,14 @@ public enum OrcConf {
       "Define the compression strategy to use while writing data.\n" +
           "This changes the compression level of higher level compression\n" +
           "codec (like ZLIB)."),
+  COMPRESSION_ZSTD_LEVEL("orc.compression.zstd.level",
+      "hive.exec.orc.compression.zstd.level", 1,
+      "Define the compression level to use with ZStandard codec "
+          + "while writing data. The valid range is 1~22"),
+  COMPRESSION_ZSTD_WINDOWLOG("orc.compression.zstd.windowlog",
+      "hive.exec.orc.compression.zstd.windowlog", 0,
+      "Set the maximum allowed back-reference distance for "
+          + "ZStandard codec, expressed as power of 2."),
   BLOCK_PADDING_TOLERANCE("orc.block.padding.tolerance",
       "hive.exec.orc.block.padding.tolerance", 0.05,
       "Define the tolerance for block padding as a decimal fraction of\n" +

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

@@ -426,6 +426,27 @@ public static BloomFilterVersion fromString(String s) {
     }
   }
 
+  public static class ZstdCompressOptions {
+    private int compressionZstdLevel;
+    private int compressionZstdWindowLog;
+
+    public int getCompressionZstdLevel() {
+      return compressionZstdLevel;
+    }
+
+    public void setCompressionZstdLevel(int compressionZstdLevel) {
+      this.compressionZstdLevel = compressionZstdLevel;
+    }
+
+    public int getCompressionZstdWindowLog() {
+      return compressionZstdWindowLog;
+    }
+
+    public void setCompressionZstdWindowLog(int compressionZstdWindowLog) {
+      this.compressionZstdWindowLog = compressionZstdWindowLog;
+    }
+  }
+
   /**
    * Options for creating ORC file writers.
    */
@@ -447,6 +468,7 @@ public static class WriterOptions implements Cloneable {
     private WriterCallback callback;
     private EncodingStrategy encodingStrategy;
     private CompressionStrategy compressionStrategy;
+    private ZstdCompressOptions zstdCompressOptions;
     private double paddingTolerance;
     private String bloomFilterColumns;
     private double bloomFilterFpp;
@@ -493,6 +515,12 @@ protected WriterOptions(Properties tableProperties, Configuration conf) {
           OrcConf.COMPRESSION_STRATEGY.getString(tableProperties, conf);
       compressionStrategy = CompressionStrategy.valueOf(compString);
 
+      zstdCompressOptions = new ZstdCompressOptions();
+      zstdCompressOptions.setCompressionZstdLevel(
+              OrcConf.COMPRESSION_ZSTD_LEVEL.getInt(tableProperties, conf));
+      zstdCompressOptions.setCompressionZstdWindowLog(
+              OrcConf.COMPRESSION_ZSTD_WINDOWLOG.getInt(tableProperties, conf));
+
       paddingTolerance =
           OrcConf.BLOCK_PADDING_TOLERANCE.getDouble(tableProperties, conf);
 
@@ -938,6 +966,10 @@ public EncodingStrategy getEncodingStrategy() {
       return encodingStrategy;
     }
 
+    public ZstdCompressOptions getZstdCompressOptions() {
+      return zstdCompressOptions;
+    }
+
     public double getPaddingTolerance() {
       return paddingTolerance;
     }

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

@@ -115,8 +115,18 @@ public PhysicalFsWriter(FSDataOutputStream outputStream,
     }
     CompressionCodec codec = OrcCodecPool.getCodec(opts.getCompress());
     if (codec != null){
-      compress.withCodec(codec, codec.getDefaultOptions());
+      CompressionCodec.Options tempOptions = codec.getDefaultOptions();
+      if (codec instanceof ZstdCodec &&
+              codec.getDefaultOptions() instanceof ZstdCodec.ZstdOptions options) {
+        OrcFile.ZstdCompressOptions zstdCompressOptions = opts.getZstdCompressOptions();
+        if (zstdCompressOptions != null) {
+          options.setLevel(zstdCompressOptions.getCompressionZstdLevel());
+          options.setWindowLog(zstdCompressOptions.getCompressionZstdWindowLog());
+        }
+      }
+      compress.withCodec(codec, tempOptions);
     }
+
     this.compressionStrategy = opts.getCompressionStrategy();
     this.maxPadding = (int) (opts.getPaddingTolerance() * defaultStripeSize);
     this.blockSize = opts.getBlockSize();

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

@@ -18,6 +18,7 @@
 
 package org.apache.orc.impl;
 
+import com.github.luben.zstd.util.Native;
 import com.google.protobuf.ByteString;
 import io.airlift.compress.lz4.Lz4Compressor;
 import io.airlift.compress.lz4.Lz4Decompressor;
@@ -273,6 +274,17 @@ private static int getClosestBufferSize(int size) {
     return Math.min(kb256, Math.max(kb4, pow2));
   }
 
+  static {
+    try {
+      if (!"java".equalsIgnoreCase(System.getProperty("orc.compression.zstd.impl"))) {
+        Native.load();
+      }
+    } catch (UnsatisfiedLinkError | ExceptionInInitializerError e) {
+      LOG.warn("Unable to load zstd-jni library for your platform. " +
+            "Using builtin-java classes where applicable");
+    }
+  }
+
   public static CompressionCodec createCodec(CompressionKind kind) {
     switch (kind) {
       case NONE:
@@ -288,8 +300,16 @@ public static CompressionCodec createCodec(CompressionKind kind) {
         return new AircompressorCodec(kind, new Lz4Compressor(),
             new Lz4Decompressor());
       case ZSTD:
-        return new AircompressorCodec(kind, new ZstdCompressor(),
-            new ZstdDecompressor());
+        if ("java".equalsIgnoreCase(System.getProperty("orc.compression.zstd.impl"))) {
+          return new AircompressorCodec(kind, new ZstdCompressor(),
+                  new ZstdDecompressor());
+        }
+        if (Native.isLoaded()) {
+          return new ZstdCodec();
+        } else {
+          return new AircompressorCodec(kind, new ZstdCompressor(),
+              new ZstdDecompressor());
+        }
       case BROTLI:
         return new BrotliCodec();
       default: