diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorMetadata.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorMetadata.java
index a4813e9f8040..a98d111279ba 100644
--- a/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorMetadata.java
+++ b/presto-raptor/src/main/java/com/facebook/presto/raptor/RaptorMetadata.java
@@ -376,7 +376,7 @@ public void commitCreateTable(ConnectorSession session, ConnectorOutputTableHand
 
         long newTableId = dbi.inTransaction((dbiHandle, status) -> {
             MetadataDao dao = dbiHandle.attach(MetadataDao.class);
-            long tableId = dao.insertTable(table.getSchemaName(), table.getTableName());
+            long tableId = dao.insertTable(table.getSchemaName(), table.getTableName(), true);
             List<RaptorColumnHandle> sortColumnHandles = table.getSortColumnHandles();
 
             for (int i = 0; i < table.getColumnTypes().size(); i++) {
diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/metadata/MetadataDao.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/metadata/MetadataDao.java
index 51abe657b071..a35e07f8deb3 100644
--- a/presto-raptor/src/main/java/com/facebook/presto/raptor/metadata/MetadataDao.java
+++ b/presto-raptor/src/main/java/com/facebook/presto/raptor/metadata/MetadataDao.java
@@ -30,6 +30,7 @@ public interface MetadataDao
             "  schema_name VARCHAR(255) NOT NULL,\n" +
             "  table_name VARCHAR(255) NOT NULL,\n" +
             "  temporal_column_id BIGINT DEFAULT NULL,\n" +
+            "  compaction_enabled BOOLEAN NOT NULL,\n" +
             "  UNIQUE (schema_name, table_name)\n" +
             ")")
     void createTableTables();
@@ -138,12 +139,13 @@ List<ViewResult> getViews(
             @Bind("schemaName") String schemaName,
             @Bind("tableName") String tableName);
 
-    @SqlUpdate("INSERT INTO tables (schema_name, table_name)\n" +
-            "VALUES (:schemaName, :tableName)")
+    @SqlUpdate("INSERT INTO tables (schema_name, table_name, compaction_enabled)\n" +
+            "VALUES (:schemaName, :tableName, :compactionEnabled)")
     @GetGeneratedKeys
     long insertTable(
             @Bind("schemaName") String schemaName,
-            @Bind("tableName") String tableName);
+            @Bind("tableName") String tableName,
+            @Bind("compactionEnabled") boolean compactionEnabled);
 
     @SqlUpdate("INSERT INTO columns (table_id, column_id, column_name, ordinal_position, data_type, sort_ordinal_position)\n" +
             "VALUES (:tableId, :columnId, :columnName, :ordinalPosition, :dataType, :sortOrdinalPosition)")
@@ -206,4 +208,7 @@ int dropView(
     void updateTemporalColumnId(
             @Bind("tableId") long tableId,
             @Bind("columnId") long columnId);
+
+    @SqlQuery("SELECT compaction_enabled FROM tables WHERE table_id = :tableId")
+    boolean isCompactionEnabled(@Bind("tableId") long tableId);
 }
diff --git a/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/ShardCompactionManager.java b/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/ShardCompactionManager.java
index 98219c1fdab2..227d02c46e47 100644
--- a/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/ShardCompactionManager.java
+++ b/presto-raptor/src/main/java/com/facebook/presto/raptor/storage/ShardCompactionManager.java
@@ -218,6 +218,9 @@ private void discoverShards()
 
         for (Entry<Long, List<ShardMetadata>> entry : Multimaps.asMap(tableShards).entrySet()) {
             long tableId = entry.getKey();
+            if (!metadataDao.isCompactionEnabled(tableId)) {
+                continue;
+            }
             List<ShardMetadata> shardMetadata = entry.getValue();
 
             Set<ShardMetadata> shards = shardMetadata.stream()
diff --git a/presto-raptor/src/test/java/com/facebook/presto/raptor/metadata/TestMetadataDao.java b/presto-raptor/src/test/java/com/facebook/presto/raptor/metadata/TestMetadataDao.java
index 122e74e96eb0..62f76defaabb 100644
--- a/presto-raptor/src/test/java/com/facebook/presto/raptor/metadata/TestMetadataDao.java
+++ b/presto-raptor/src/test/java/com/facebook/presto/raptor/metadata/TestMetadataDao.java
@@ -52,7 +52,7 @@ public void testTemporalColumn()
             throws Exception
     {
         Long columnId = 1L;
-        long tableId = dao.insertTable("schema1", "table1");
+        long tableId = dao.insertTable("schema1", "table1", true);
         dao.insertColumn(tableId, columnId, "col1", 1, "bigint", null);
         Long temporalColumnId = dao.getTemporalColumnId(tableId);
         assertNull(temporalColumnId);
@@ -62,7 +62,7 @@ public void testTemporalColumn()
         assertNotNull(temporalColumnId);
         assertEquals(temporalColumnId, columnId);
 
-        long tableId2 = dao.insertTable("schema1", "table2");
+        long tableId2 = dao.insertTable("schema1", "table2", true);
         Long columnId2 = dao.getTemporalColumnId(tableId2);
         assertNull(columnId2);
     }