diff --git a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java
index 2df7d2f9a..255cb2e81 100644
--- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java
+++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/HeuristicIndexClient.java
@@ -217,12 +217,17 @@ public class HeuristicIndexClient
     public void addIndexRecord(CreateIndexMetadata createIndexMetadata)
             throws IOException
     {
+        List<String> properties = new LinkedList<>();
+        for (String propKey : createIndexMetadata.getProperties().stringPropertyNames()) {
+            properties.add(propKey + "=" + createIndexMetadata.getProperties().getProperty(propKey));
+        }
         indexRecordManager.addIndexRecord(
                 createIndexMetadata.getIndexName(),
                 createIndexMetadata.getUser(),
                 createIndexMetadata.getTableName(),
                 createIndexMetadata.getIndexColumns().stream().map(Map.Entry::getKey).toArray(String[]::new),
                 createIndexMetadata.getIndexType(),
+                properties,
                 createIndexMetadata.getPartitions());
     }
 
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/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
index 90e8504cf..4a8f39970 100644
--- a/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
+++ b/hetu-heuristic-index/src/main/java/io/hetu/core/heuristicindex/IndexRecordManager.java
@@ -87,6 +87,10 @@ public class IndexRecordManager
                         records.add(new IndexRecord(line));
                     }
                 }
+                catch (Exception e) {
+                    throw new IllegalArgumentException(
+                            "Error reading index record. Index record storage has been updated. Please delete old index directory and recreate the indices.");
+                }
                 cache = records;
                 cacheLastModifiedTime = modifiedTime;
             }
@@ -127,7 +131,7 @@ public class IndexRecordManager
      * Add IndexRecord into record file. If the method is called with a name that already exists,
      * it will OVERWRITE the existing entry but combine the note part
      */
-    public synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
+    public synchronized void addIndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
             throws IOException
     {
         // Protect root directory
@@ -143,7 +147,7 @@ public class IndexRecordManager
                     iterator.remove();
                 }
             }
-            records.add(new IndexRecord(name, user, table, columns, indexType, partitions));
+            records.add(new IndexRecord(name, user, table, columns, indexType, indexProperties, partitions));
             writeIndexRecords(records);
         }
         finally {
@@ -179,7 +183,8 @@ public class IndexRecordManager
         boolean writeHead = false;
         try (OutputStream os = fs.newOutputStream(recordFile)) {
             // Use IndexRecord to generate a special "entry" as table head so it's easier to maintain when csv format changes
-            String head = new IndexRecord("Name", "User", "Table", new String[] {"Columns"}, "IndexType", ImmutableList.of("Partitions")).toCsvRecord();
+            String head = new IndexRecord("Name", "User", "Table", new String[] {"Columns"}, "IndexType",
+                    ImmutableList.of("Properties"), ImmutableList.of("Partitions")).toCsvRecord();
             os.write(head.getBytes());
             for (IndexRecord record : records) {
                 os.write(record.toCsvRecord().getBytes());
diff --git a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java
index 36c7c8c84..e1cede78f 100644
--- a/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java
+++ b/hetu-heuristic-index/src/test/java/io/hetu/core/heuristicindex/TestIndexRecordManager.java
@@ -51,14 +51,14 @@ public class TestIndexRecordManager
             IndexRecordManager indexRecordManager1 = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
             IndexRecordManager indexRecordManager2 = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
 
-            indexRecordManager1.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
+            indexRecordManager1.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
             List<IndexRecord> original1 = indexRecordManager1.getIndexRecords();
             List<IndexRecord> original2 = indexRecordManager2.getIndexRecords();
             assertEquals(original2.size(), 1);
             assertEquals(original1, original2);
 
             List<IndexRecord> beforeadd1 = indexRecordManager1.getIndexRecords();
-            indexRecordManager2.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Arrays.asList("cp=1"));
+            indexRecordManager2.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList(), Arrays.asList("cp=1"));
             List<IndexRecord> added2 = indexRecordManager2.getIndexRecords();
             assertEquals(added2.size(), 2);
 
@@ -89,7 +89,7 @@ public class TestIndexRecordManager
                 threads[i] = new Thread(() -> {
                     try {
                         new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath())
-                                .addIndexRecord(names[finalI], "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
+                                .addIndexRecord(names[finalI], "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
                     }
                     catch (IOException e) {
                         throw new RuntimeException(e);
@@ -147,7 +147,7 @@ public class TestIndexRecordManager
                 int finalI = i;
                 threads[i] = new Thread(() -> {
                     try {
-                        indexRecordManager.addIndexRecord(names[finalI], "u", "t", new String[] {"c"}, "minmax", Arrays.asList("cp=1"));
+                        indexRecordManager.addIndexRecord(names[finalI], "u", "t", new String[] {"c"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
                     }
                     catch (IOException e) {
                         throw new RuntimeException(e);
@@ -193,8 +193,8 @@ public class TestIndexRecordManager
         try (TempFolder folder = new TempFolder()) {
             folder.create();
             IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
-            indexRecordManager.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
-            indexRecordManager.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
+            indexRecordManager.addIndexRecord("1", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
+            indexRecordManager.addIndexRecord("2", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
             assertNotNull(indexRecordManager.lookUpIndexRecord("1"));
             assertEquals(indexRecordManager.getIndexRecords().size(), 2);
 
@@ -221,21 +221,21 @@ public class TestIndexRecordManager
     public void testAddAndLookUp()
             throws IOException, IllegalAccessException
     {
-        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
-        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn", "testColumn2"}, "minmax", Collections.emptyList());
-        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ImmutableList.of("12"));
-        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", ImmutableList.of("12", "123"));
+        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Collections.emptyList());
+        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn", "testColumn2"}, "minmax", Collections.emptyList(), Collections.emptyList());
+        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("12"));
+        testIndexRecordAddLookUpHelper("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("12", "123"));
     }
 
     @Test
     public void testRecordEqualAndHash()
     {
-        IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList());
+        IndexRecord r1 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Collections.emptyList());
         IndexRecord r2 = new IndexRecord("testName", "testUser", "testTable", new String[] {
-                "testColumn"}, "minmax", ImmutableList.of("note"));
-        IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList());
+                "testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of("note"));
+        IndexRecord r3 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "bloom", Collections.emptyList(), Collections.emptyList());
         IndexRecord r4 = new IndexRecord("testName", "testUser", "testTable", new String[] {"testColumn",
-                "testColumn2"}, "minmax", Collections.emptyList());
+                "testColumn2"}, "minmax", Collections.emptyList(), Collections.emptyList());
         assertEquals(r1, r1);
         assertEquals(r1, r2);
         assertNotEquals(r1, r3);
@@ -253,28 +253,28 @@ public class TestIndexRecordManager
     }
 
     @Test(expectedExceptions = AssertionError.class)
-    public void testAddAndLookUpDifferentNotes()
+    public void testAddAndLookUpDifferentPartitions()
             throws IOException, IllegalAccessException
     {
         try (TempFolder folder = new TempFolder()) {
             folder.create();
             IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
             IndexRecord expected = new IndexRecord("testName", "testUser", "testTable", new String[] {
-                    "testColumn"}, "minmax", ImmutableList.of(""));
-            indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Arrays.asList("cp=1"));
+                    "testColumn"}, "minmax", Collections.emptyList(), ImmutableList.of(""));
+            indexRecordManager.addIndexRecord("testName", "testUser", "testTable", new String[] {"testColumn"}, "minmax", Collections.emptyList(), Arrays.asList("cp=1"));
             IndexRecord actual = indexRecordManager.lookUpIndexRecord("testName");
             assertIndexRecordFullyEqual(actual, expected);
         }
     }
 
-    private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, List<String> note)
+    private void testIndexRecordAddLookUpHelper(String name, String user, String table, String[] columns, String indexType, List<String> indexProperties, List<String> partitions)
             throws IOException, IllegalAccessException
     {
         try (TempFolder folder = new TempFolder()) {
             folder.create();
             IndexRecordManager indexRecordManager = new IndexRecordManager(FILE_SYSTEM_CLIENT, folder.getRoot().toPath());
-            IndexRecord expected = new IndexRecord(name, user, table, columns, indexType, note);
-            indexRecordManager.addIndexRecord(name, user, table, columns, indexType, note);
+            IndexRecord expected = new IndexRecord(name, user, table, columns, indexType, indexProperties, partitions);
+            indexRecordManager.addIndexRecord(name, user, table, columns, indexType, indexProperties, partitions);
 
             IndexRecord actual1 = indexRecordManager.lookUpIndexRecord(name);
             assertNotNull(actual1);
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 5b525c06f..4d2ebe962 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
 {
@@ -99,9 +98,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/main/java/io/prestosql/sql/planner/QueryPlanner.java b/presto-main/src/main/java/io/prestosql/sql/planner/QueryPlanner.java
index 3859057f7..eacf33ddc 100644
--- a/presto-main/src/main/java/io/prestosql/sql/planner/QueryPlanner.java
+++ b/presto-main/src/main/java/io/prestosql/sql/planner/QueryPlanner.java
@@ -617,13 +617,13 @@ class QueryPlanner
 
         Properties indexProperties = new Properties();
         CreateIndexMetadata.Level indexCreationLevel = LEVEL_DEFAULT;
+        indexProperties.setProperty(LEVEL_PROP_KEY, String.valueOf(LEVEL_DEFAULT));
 
         for (Property property : createIndex.getProperties()) {
             String key = property.getName().toString().replaceAll("\"", "");
             String val = property.getValue().toString().replaceAll("\"", "").toUpperCase(Locale.ENGLISH);
             if (key.equals(LEVEL_PROP_KEY)) {
                 indexCreationLevel = CreateIndexMetadata.Level.valueOf(val);
-                continue;
             }
             indexProperties.setProperty(key, val);
         }
diff --git a/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java b/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java
index 0d759f892..d4879edd5 100644
--- a/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java
+++ b/presto-main/src/main/java/io/prestosql/sql/rewrite/ShowQueriesRewrite.java
@@ -668,7 +668,7 @@ final class ShowQueriesRewrite
                         new StringLiteral(String.join(",", v.columns)),
                         new StringLiteral(v.indexType),
                         new StringLiteral(partitionsStrToDisplay.toString()),
-                        new StringLiteral(""),
+                        new StringLiteral(String.join(",", v.properties)),
                         TRUE_LITERAL));
             }
 
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();
 }
diff --git a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
index 1ac7052bf..1bbbd59ce 100644
--- a/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
+++ b/presto-spi/src/main/java/io/prestosql/spi/heuristicindex/IndexRecord.java
@@ -15,9 +15,9 @@
 package io.prestosql.spi.heuristicindex;
 
 import java.util.Arrays;
-import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 public class IndexRecord
 {
@@ -27,32 +27,36 @@ public class IndexRecord
     public final String table;
     public final String[] columns;
     public final String indexType;
+    public final List<String> properties;
     public final List<String> partitions;
 
-    public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> partitions)
+    public IndexRecord(String name, String user, String table, String[] columns, String indexType, List<String> properties, List<String> partitions)
     {
         this.name = name;
         this.user = user == null ? "" : user;
         this.table = table;
         this.columns = columns;
         this.indexType = indexType;
+        this.properties = properties;
         this.partitions = partitions;
     }
 
     public IndexRecord(String csvRecord)
     {
-        String[] records = csvRecord.split("\\t");
+        String[] records = csvRecord.split("\\|", Integer.MAX_VALUE);
         this.name = records[0];
         this.user = records[1];
         this.table = records[2];
         this.columns = records[3].split(COLUMN_DELIMITER);
         this.indexType = records[4];
-        this.partitions = records.length > 5 ? Arrays.asList(records[5].split(",")) : Collections.emptyList();
+        this.properties = Arrays.stream(records[5].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
+        this.partitions = Arrays.stream(records[6].split(",")).filter(s -> !s.equals("")).collect(Collectors.toList());
     }
 
     public String toCsvRecord()
     {
-        return String.format("%s\t%s\t%s\t%s\t%s\t%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType, String.join(",", partitions));
+        return String.format("%s|%s|%s|%s|%s|%s|%s\n", name, user, table, String.join(COLUMN_DELIMITER, columns), indexType,
+                String.join(",", properties), String.join(",", partitions));
     }
 
     @Override