diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java
index 0317cb40d..0e14af13c 100755
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java
@@ -109,6 +109,7 @@ import java.net.URL;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -187,6 +188,8 @@ public class HiveMetadata
     public static final String BUCKETING_VERSION = "bucketing_version";
     public static final String TABLE_COMMENT = "comment";
 
+    public static final String STORAGE_FORMAT = "storage_format";
+
     private static final String ORC_BLOOM_FILTER_COLUMNS_KEY = "orc.bloom.filter.columns";
     private static final String ORC_BLOOM_FILTER_FPP_KEY = "orc.bloom.filter.fpp";
 
@@ -298,10 +301,18 @@ public class HiveMetadata
 
         MetastoreUtil.verifyOnline(tableName, Optional.empty(), MetastoreUtil.getProtectMode(table.get()), table.get().getParameters());
 
+        Map<String, String> parameters = new HashMap<>();
+        parameters.putAll(table.get().getParameters());
+
+        String format = table.get().getStorage().getStorageFormat().getOutputFormatNullable();
+        if (format != null) {
+            parameters.put(STORAGE_FORMAT, format);
+        }
+
         return new HiveTableHandle(
                 tableName.getSchemaName(),
                 tableName.getTableName(),
-                table.get().getParameters(),
+                parameters,
                 getPartitionKeyColumnHandles(table.get()),
                 HiveBucketing.getHiveBucketHandle(table.get()));
     }
diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveTableHandle.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveTableHandle.java
index 6ac4ea466..7db589951 100644
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveTableHandle.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveTableHandle.java
@@ -32,6 +32,7 @@ import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static io.prestosql.plugin.hive.HiveMetadata.STORAGE_FORMAT;
 import static java.util.Objects.requireNonNull;
 
 public class HiveTableHandle
@@ -351,4 +352,10 @@ public class HiveTableHandle
     {
         return this.suitableToPush;
     }
+
+    @Override
+    public boolean isTableCacheable()
+    {
+        return HiveStorageFormat.ORC.getOutputFormat().equals(tableParameters.get().get(STORAGE_FORMAT));
+    }
 }
diff --git a/presto-main/src/main/java/io/prestosql/sql/rewrite/CacheTableRewrite.java b/presto-main/src/main/java/io/prestosql/sql/rewrite/CacheTableRewrite.java
index 8b9a6df05..827504240 100644
--- a/presto-main/src/main/java/io/prestosql/sql/rewrite/CacheTableRewrite.java
+++ b/presto-main/src/main/java/io/prestosql/sql/rewrite/CacheTableRewrite.java
@@ -25,6 +25,7 @@ import io.prestosql.security.AccessControl;
 import io.prestosql.spi.HetuConstant;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.connector.ColumnMetadata;
+import io.prestosql.spi.connector.ConnectorTableHandle;
 import io.prestosql.spi.predicate.Domain;
 import io.prestosql.spi.predicate.TupleDomain;
 import io.prestosql.spi.service.PropertyService;
@@ -142,6 +143,11 @@ final class CacheTableRewrite
                 throw new SemanticException(MISSING_CACHE, cache, "Table '%s' does not exist", qualifiedTableName.toString());
             }
 
+            ConnectorTableHandle tableHandle = metadata.getTableHandle(session, qualifiedTableName).get().getConnectorHandle();
+            if (!tableHandle.isTableCacheable()) {
+                throw new SemanticException(INVALID_TABLE, cache, "Table '%s' cannot be cached", qualifiedTableName.toString());
+            }
+
             TableMetadata tableMetadata = metadata.getTableMetadata(session, metadata.getTableHandle(session, qualifiedTableName).get());
 
             TupleDomain<ColumnMetadata> columnMetadataTupleDomain = translateToTupleDomain(tableMetadata, predicate);
@@ -182,10 +188,6 @@ final class CacheTableRewrite
             Identifier identifier = null;
             Expression value = null;
 
-            if (!metadata.getTableHandle(session, tableMetadata.getQualifiedName()).get().getConnectorHandle().isFilterSupported()) {
-                throw new SemanticException(INVALID_TABLE, node, "Table '%s' cannot be cached. Only tables in Hive catalog can be cached",
-                        tableMetadata.getQualifiedName().toString());
-            }
             if (whereClause instanceof ComparisonExpression) {
                 ComparisonExpression predicate = (ComparisonExpression) whereClause;
                 identifier = (Identifier) ((predicate.getLeft() instanceof Identifier) ? predicate.getLeft() : predicate.getRight());
diff --git a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorTableHandle.java b/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorTableHandle.java
index 6facd391b..7bea087dd 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorTableHandle.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorTableHandle.java
@@ -70,7 +70,6 @@ public interface ConnectorTableHandle
      * schema name and table name.
      *
      * @return table name in 'schema.table' format if schema is available
-     *
      */
     default String getSchemaPrefixedTableName()
     {
@@ -103,4 +102,10 @@ public interface ConnectorTableHandle
     {
         return "";
     }
+
+    /* This method simply checks if Table can be cached by the Connector */
+    default boolean isTableCacheable()
+    {
+        return false;
+    }
 }