diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveCompressionCodec.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveCompressionCodec.java
index 80182299e0b14..5363f6e2a6a72 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HiveCompressionCodec.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HiveCompressionCodec.java
@@ -20,24 +20,34 @@
 import parquet.hadoop.metadata.CompressionCodecName;
 
 import java.util.Optional;
+import java.util.function.Predicate;
 
+import static com.facebook.presto.hive.HiveStorageFormat.DWRF;
+import static com.facebook.presto.hive.HiveStorageFormat.ORC;
 import static java.util.Objects.requireNonNull;
 
 public enum HiveCompressionCodec
 {
-    NONE(null, CompressionKind.NONE, CompressionCodecName.UNCOMPRESSED),
-    SNAPPY(SnappyCodec.class, CompressionKind.SNAPPY, CompressionCodecName.SNAPPY),
-    GZIP(GzipCodec.class, CompressionKind.ZLIB, CompressionCodecName.GZIP);
+    NONE(null, CompressionKind.NONE, CompressionCodecName.UNCOMPRESSED, f -> true),
+    SNAPPY(SnappyCodec.class, CompressionKind.SNAPPY, CompressionCodecName.SNAPPY, f -> true),
+    GZIP(GzipCodec.class, CompressionKind.ZLIB, CompressionCodecName.GZIP, f -> true),
+    ZSTD(null, CompressionKind.ZSTD, null, f -> f == ORC || f == DWRF);
 
     private final Optional<Class<? extends CompressionCodec>> codec;
     private final CompressionKind orcCompressionKind;
-    private final CompressionCodecName parquetCompressionCodec;
+    private final Optional<CompressionCodecName> parquetCompressionCodec;
+    private final Predicate<HiveStorageFormat> supportedStorageFormats;
 
-    HiveCompressionCodec(Class<? extends CompressionCodec> codec, CompressionKind orcCompressionKind, CompressionCodecName parquetCompressionCodec)
+    HiveCompressionCodec(
+            Class<? extends CompressionCodec> codec,
+            CompressionKind orcCompressionKind,
+            CompressionCodecName parquetCompressionCodec,
+            Predicate<HiveStorageFormat> supportedStorageFormats)
     {
         this.codec = Optional.ofNullable(codec);
         this.orcCompressionKind = requireNonNull(orcCompressionKind, "orcCompressionKind is null");
-        this.parquetCompressionCodec = requireNonNull(parquetCompressionCodec, "parquetCompressionCodec is null");
+        this.parquetCompressionCodec = Optional.ofNullable(parquetCompressionCodec);
+        this.supportedStorageFormats = requireNonNull(supportedStorageFormats, "supportedStorageFormats is null");
     }
 
     public Optional<Class<? extends CompressionCodec>> getCodec()
@@ -50,8 +60,13 @@ public CompressionKind getOrcCompressionKind()
         return orcCompressionKind;
     }
 
-    public CompressionCodecName getParquetCompressionCodec()
+    public Optional<CompressionCodecName> getParquetCompressionCodec()
     {
         return parquetCompressionCodec;
     }
+
+    public boolean isSupportedStorageFormat(HiveStorageFormat hiveStorageFormat)
+    {
+        return supportedStorageFormats.test(hiveStorageFormat);
+    }
 }
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/HiveWritableTableHandle.java b/presto-hive/src/main/java/com/facebook/presto/hive/HiveWritableTableHandle.java
index d74cbe60eff23..7a5ca51644635 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/HiveWritableTableHandle.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/HiveWritableTableHandle.java
@@ -14,12 +14,14 @@
 package com.facebook.presto.hive;
 
 import com.facebook.presto.hive.metastore.HivePageSinkMetadata;
+import com.facebook.presto.spi.PrestoException;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 
 import java.util.List;
 import java.util.Optional;
 
+import static com.facebook.presto.spi.StandardErrorCode.GENERIC_USER_ERROR;
 import static java.util.Objects.requireNonNull;
 
 public class HiveWritableTableHandle
@@ -57,6 +59,13 @@ public HiveWritableTableHandle(
         this.tableStorageFormat = requireNonNull(tableStorageFormat, "tableStorageFormat is null");
         this.partitionStorageFormat = requireNonNull(partitionStorageFormat, "partitionStorageFormat is null");
         this.compressionCodec = requireNonNull(compressionCodec, "compressionCodec is null");
+
+        if (!compressionCodec.isSupportedStorageFormat(tableStorageFormat)) {
+            throw new PrestoException(GENERIC_USER_ERROR, String.format("%s compression is not supported with %s", compressionCodec.name(), tableStorageFormat.name()));
+        }
+        if (!compressionCodec.isSupportedStorageFormat(partitionStorageFormat)) {
+            throw new PrestoException(GENERIC_USER_ERROR, String.format("%s compression is not supported with %s", compressionCodec.name(), partitionStorageFormat.name()));
+        }
     }
 
     @JsonProperty
diff --git a/presto-hive/src/main/java/com/facebook/presto/hive/util/ConfigurationUtils.java b/presto-hive/src/main/java/com/facebook/presto/hive/util/ConfigurationUtils.java
index f55ee3c4f817b..d142f807b4f76 100644
--- a/presto-hive/src/main/java/com/facebook/presto/hive/util/ConfigurationUtils.java
+++ b/presto-hive/src/main/java/com/facebook/presto/hive/util/ConfigurationUtils.java
@@ -113,7 +113,7 @@ private static void setCompressionProperties(Configuration config, HiveCompressi
             config.unset(FileOutputFormat.COMPRESS_CODEC);
         }
         // For Parquet
-        config.set(ParquetOutputFormat.COMPRESSION, compression.getParquetCompressionCodec().name());
+        compression.getParquetCompressionCodec().ifPresent(codec -> config.set(ParquetOutputFormat.COMPRESSION, codec.name()));
         // For SequenceFile
         config.set(FileOutputFormat.COMPRESS_TYPE, BLOCK.toString());
     }
diff --git a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
index 5ad0c5e2a9d46..830358baade16 100644
--- a/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
+++ b/presto-hive/src/test/java/com/facebook/presto/hive/TestHivePageSink.java
@@ -119,7 +119,7 @@ public void testAllFormats()
                 assertGreaterThan(uncompressedLength, 0L);
 
                 for (HiveCompressionCodec codec : HiveCompressionCodec.values()) {
-                    if (codec == NONE) {
+                    if (codec == NONE || !codec.isSupportedStorageFormat(format)) {
                         continue;
                     }
                     config.setCompressionCodec(codec);
diff --git a/presto-orc/src/test/java/com/facebook/presto/orc/OrcTester.java b/presto-orc/src/test/java/com/facebook/presto/orc/OrcTester.java
index 39ad8cc83bcf2..19c71b4373f47 100644
--- a/presto-orc/src/test/java/com/facebook/presto/orc/OrcTester.java
+++ b/presto-orc/src/test/java/com/facebook/presto/orc/OrcTester.java
@@ -314,7 +314,7 @@ public static OrcTester quickSelectiveOrcTester()
         orcTester.nullTestsEnabled = true;
         orcTester.skipBatchTestsEnabled = true;
         orcTester.formats = ImmutableSet.of(ORC_12, ORC_11, DWRF);
-        orcTester.compressions = ImmutableSet.of(ZLIB);
+        orcTester.compressions = ImmutableSet.of(ZLIB, ZSTD);
         orcTester.useSelectiveOrcReader = true;
 
         return orcTester;