diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java
index bc4dcf7d3..1e14ad79c 100644
--- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java
+++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexFactory.java
@@ -15,7 +15,6 @@
 
 package io.hetu.core.heuristicindex;
 
-import com.google.common.collect.ImmutableSet;
 import io.airlift.log.Logger;
 import io.hetu.core.heuristicindex.filter.HeuristicIndexFilter;
 import io.hetu.core.plugin.heuristicindex.index.bloom.BloomIndex;
@@ -35,7 +34,6 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
-import java.util.Set;
 
 import static java.util.Objects.requireNonNull;
 
@@ -49,7 +47,6 @@ public class HeuristicIndexFactory
         implements IndexFactory
 {
     private static final Logger LOG = Logger.get(HeuristicIndexFactory.class);
-    private final Set<String> supportedConnector = ImmutableSet.of("hive");
 
     public HeuristicIndexFactory()
     {
@@ -101,10 +98,4 @@ public class HeuristicIndexFactory
     {
         return new HeuristicIndexFilter(indices);
     }
-
-    @Override
-    public Set<String> getSupportedConnector()
-    {
-        return supportedConnector;
-    }
 }
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 e770e4074..c1fe1426e 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
@@ -586,6 +586,12 @@ public class HiveMetadata
                         false));
     }
 
+    @Override
+    public boolean isHeuristicIndexSupported()
+    {
+        return true;
+    }
+
     @Override
     public List<SchemaTableName> listTables(ConnectorSession session, Optional<String> optionalSchemaName)
     {
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 7db589951..59d5ee6e4 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
@@ -31,6 +31,7 @@ import java.util.Objects;
 import java.util.Optional;
 import java.util.function.Function;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static io.prestosql.plugin.hive.HiveMetadata.STORAGE_FORMAT;
 import static java.util.Objects.requireNonNull;
@@ -358,4 +359,25 @@ public class HiveTableHandle
     {
         return HiveStorageFormat.ORC.getOutputFormat().equals(tableParameters.get().get(STORAGE_FORMAT));
     }
+
+    /**
+     * ORC is the only format supported to create heuristic index now
+     * We will add more formats in the future.
+     */
+    @Override
+    public boolean isHeuristicIndexSupported()
+    {
+        return Stream.of(HiveStorageFormat.ORC)
+                .anyMatch(storageFormat -> storageFormat.getOutputFormat().equals(tableParameters.get().get(STORAGE_FORMAT)));
+    }
+
+    /**
+     * Create heuristic index... where predicate = xxx
+     * The predicate column only support partition columns
+     */
+    @Override
+    public boolean isPartitionColumn(String column)
+    {
+        return partitionColumns.stream().map(HiveColumnHandle::getColumnName).collect(Collectors.toSet()).contains(column);
+    }
 }
diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java
index 815d287fb..24990fe19 100644
--- a/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java
+++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/util/TestIndexCache.java
@@ -57,7 +57,7 @@ public class TestIndexCache
     private HiveColumnHandle partitionColumnHandle;
     private HiveSplit testHiveSplit;
     private List<HiveColumnHandle> testPartitions = Collections.emptyList();
-    private final long loadDelay = 0;
+    private final long loadDelay = 1000;
 
     @BeforeClass
     public void setupBeforeClass()
diff --git a/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java b/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java
index 82acc00a3..c6e740463 100644
--- a/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java
+++ b/presto-main/src/main/java/io/prestosql/heuristicindex/HeuristicIndexerManager.java
@@ -35,7 +35,6 @@ import java.nio.file.Paths;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.Set;
 
 public class HeuristicIndexerManager
 {
@@ -101,9 +100,4 @@ public class HeuristicIndexerManager
     {
         return factory.getIndexFilter(indices);
     }
-
-    public Set<String> getSupportedCatalog()
-    {
-        return factory.getSupportedConnector();
-    }
 }
diff --git a/presto-main/src/main/java/io/prestosql/metadata/Metadata.java b/presto-main/src/main/java/io/prestosql/metadata/Metadata.java
index 6ea89ce4b..c0fca718d 100755
--- a/presto-main/src/main/java/io/prestosql/metadata/Metadata.java
+++ b/presto-main/src/main/java/io/prestosql/metadata/Metadata.java
@@ -532,6 +532,14 @@ public interface Metadata
      */
     boolean isExecutionPlanCacheSupported(Session session, TableHandle handle);
 
+    /**
+     * Hetu can only create index for supported connectors.
+     *
+     * @param session Presto session
+     * @param tableName Connector specific tableName
+     */
+    boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName);
+
     /**
      * Hetu supports pushing sub-query with join down to the connector.
      * This method decides if the sub-query can be pushed down to the connector based on the connector.
diff --git a/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java b/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java
index 6c3ae47af..65b4ccadc 100755
--- a/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java
+++ b/presto-main/src/main/java/io/prestosql/metadata/MetadataManager.java
@@ -1062,6 +1062,27 @@ public final class MetadataManager
         return metadata.isExecutionPlanCacheSupported(session.toConnectorSession(), table.getConnectorHandle());
     }
 
+    /**
+     * Hetu can only create index for supported connectors.
+     *
+     * @param session Presto session
+     * @param tableName Connector specific tableName
+     */
+    @Override
+    public boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName)
+    {
+        Optional<CatalogMetadata> catalog = getOptionalCatalogMetadata(session, tableName.getCatalogName());
+        if (catalog.isPresent()) {
+            CatalogMetadata catalogMetadata = catalog.get();
+            CatalogName catalogName = catalogMetadata.getConnectorId(session, tableName);
+            ConnectorMetadata metadata = catalogMetadata.getMetadataFor(catalogName);
+
+            return metadata.isHeuristicIndexSupported();
+        }
+
+        return false;
+    }
+
     @Override
     public Optional<LimitApplicationResult<TableHandle>> applyLimit(Session session, TableHandle table, long limit)
     {
diff --git a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java
index eb7424d58..02c3b6b6f 100644
--- a/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java
+++ b/presto-main/src/main/java/io/prestosql/sql/analyzer/StatementAnalyzer.java
@@ -1031,17 +1031,35 @@ class StatementAnalyzer
                 CreateIndex createIndex = (CreateIndex) analysis.getOriginalStatement();
                 QualifiedObjectName tableFullName = MetadataUtil.createQualifiedObjectName(session, createIndex, createIndex.getTableName());
                 String tableName = tableFullName.toString();
-                // check catalog validate
-                if (!heuristicIndexerManager.getSupportedCatalog().contains(tableFullName.getCatalogName())) {
+                // check whether catalog support create index
+                if (!metadata.isHeuristicIndexSupported(session, tableFullName)) {
                     throw new SemanticException(NOT_SUPPORTED, createIndex,
-                            "CREATE INDEX is not supported in '%s' connector",
+                            "CREATE INDEX is not supported in catalog '%s'",
                             tableFullName.getCatalogName());
                 }
+
                 List<String> partitions = new ArrayList<>();
+                String partitionColumn = null;
                 if (createIndex.getExpression().isPresent()) {
                     partitions = HeuristicIndexUtils.extractPartitions(createIndex.getExpression().get());
-                    // check partitions validate
+                    // check partition name validate, create index …… where pt_d = xxx;
+                    // pt_d must be partition column
+                    List<String> partitionColumns = partitions.stream().map(k -> k.substring(0, k.indexOf("="))).collect(Collectors.toList());
+                    if (partitionColumns.size() > 1) {
+                        throw new IllegalArgumentException("Heuristic index only supports predicates on one column");
+                    }
+                    partitionColumn = partitionColumns.get(0);
                 }
+
+                Optional<TableHandle> tableHandle = metadata.getTableHandle(session, tableFullName);
+                if (tableHandle.isPresent() && !tableHandle.get().getConnectorHandle().isHeuristicIndexSupported()) {
+                    throw new SemanticException(NOT_SUPPORTED, table, "Catalog supported, but table storage format is not supported by heuristic index");
+                }
+                if (tableHandle.isPresent() && partitionColumn != null
+                        && !tableHandle.get().getConnectorHandle().isPartitionColumn(partitionColumn)) {
+                    throw new SemanticException(NOT_SUPPORTED, table, "Heuristic index creation is only supported for predicates on partition columns");
+                }
+
                 List<Map.Entry<String, Type>> indexColumns = new LinkedList<>();
                 for (Identifier i : createIndex.getColumnAliases()) {
                     indexColumns.add(new AbstractMap.SimpleEntry<>(i.toString(), BIGINT));
diff --git a/presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java b/presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java
index e4e035807..c9df51857 100644
--- a/presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java
+++ b/presto-main/src/test/java/io/prestosql/metadata/AbstractMockMetadata.java
@@ -697,4 +697,15 @@ public abstract class AbstractMockMetadata
     {
         return true;
     }
+
+    /**
+     * Hetu can only create index for supported connectors.
+     *
+     * @param session Presto session
+     * @param tableName Connector specific tableName
+     */
+    public boolean isHeuristicIndexSupported(Session session, QualifiedObjectName tableName)
+    {
+        return true;
+    }
 }
diff --git a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java b/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java
index f3e0d90ce..88e1fc96c 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/connector/ConnectorMetadata.java
@@ -922,4 +922,12 @@ public interface ConnectorMetadata
     {
         return false;
     }
+
+    /**
+     * Hetu can only create index for supported connectors.
+     */
+    default boolean isHeuristicIndexSupported()
+    {
+        return false;
+    }
 }
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 7bea087dd..994e15fdb 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
@@ -108,4 +108,16 @@ public interface ConnectorTableHandle
     {
         return false;
     }
+
+    /* This method checks if heuristic index can be created with the table format */
+    default boolean isHeuristicIndexSupported()
+    {
+        return false;
+    }
+
+    /* This method checks if the predicate columns are partition columns */
+    default boolean isPartitionColumn(String column)
+    {
+        return false;
+    }
 }
diff --git a/presto-spi/src/main/java/io/prestosql/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java b/presto-spi/src/main/java/io/prestosql/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java
index 3e40c2b3a..707133c66 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/connector/classloader/ClassLoaderSafeConnectorMetadata.java
@@ -767,6 +767,14 @@ public class ClassLoaderSafeConnectorMetadata
         return delegate.isExecutionPlanCacheSupported(session, handle);
     }
 
+    /**
+     * Hetu can only create index for supported connectors.
+     */
+    public boolean isHeuristicIndexSupported()
+    {
+        return delegate.isHeuristicIndexSupported();
+    }
+
     @Override
     public List<ConnectorVacuumTableInfo> getTablesForVacuum()
     {
diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFactory.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFactory.java
index 26ea529d2..423247c07 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFactory.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexFactory.java
@@ -21,7 +21,6 @@ import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
-import java.util.Set;
 
 public interface IndexFactory
 {
@@ -47,6 +46,4 @@ public interface IndexFactory
     public IndexClient getIndexClient(HetuFileSystemClient fs, Path root);
 
     public IndexFilter getIndexFilter(Map<String, List<IndexMetadata>> indices);
-
-    public Set<String> getSupportedConnector();
 }