diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveColumnHandle.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveColumnHandle.java
index 9788ab472..5a138a21e 100644
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveColumnHandle.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HiveColumnHandle.java
@@ -53,6 +53,10 @@ public class HiveColumnHandle
     public static final int ROW_ID__COLUMN_INDEX = -13;
     public static final String UPDATE_ROW_ID_COLUMN_NAME = "$rowId";
 
+    // Ids <= MAX_PARTITION_KEY_COLUMN_INDEX, can be used for distinguishing between different partition prefilled columns.
+    // NOTE: Incase any new hidden columns added, their index should be more than below value or below value should be adjusted.
+    public static final int MAX_PARTITION_KEY_COLUMN_INDEX = -14;
+
     public enum ColumnType
     {
         PARTITION_KEY,
diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java
index 0875f377a..cda76456f 100644
--- a/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java
+++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java
@@ -58,6 +58,7 @@ import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.Iterables.getOnlyElement;
 import static com.google.common.collect.Maps.uniqueIndex;
 import static io.prestosql.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
+import static io.prestosql.plugin.hive.HiveColumnHandle.MAX_PARTITION_KEY_COLUMN_INDEX;
 import static io.prestosql.plugin.hive.HivePageSourceProvider.ColumnMapping.toColumnHandles;
 import static io.prestosql.plugin.hive.HiveUtil.isPartitionFiltered;
 import static io.prestosql.plugin.hive.coercions.HiveCoercer.createCoercer;
@@ -152,9 +153,8 @@ public class HivePageSourceProvider
                 indexes == null || indexes.isEmpty() ? Optional.empty() : Optional.of(indexes);
 
         if (HiveSessionProperties.isOrcPredicatePushdownEnabled(session)) {
-            //TODO: Rajeev: To check if index and dynamic filter required for selective orc
             return createSelectivePageSource(selectivePageSourceFactories, configuration,
-                    session, hiveSplit, hiveColumns, hiveStorageTimeZone, typeManager,
+                    session, hiveSplit, assignUniqueIndicesToPartitionColumns(hiveColumns), hiveStorageTimeZone, typeManager,
                     dynamicFilterSupplier, hiveSplit.getDeleteDeltaLocations(),
                     hiveSplit.getStartRowOffsetOfFile(),
                     indexOptional, hiveSplit.isCacheable(),
@@ -195,6 +195,23 @@ public class HivePageSourceProvider
         throw new RuntimeException("Could not find a file reader for split " + hiveSplit);
     }
 
+    private static List<HiveColumnHandle> assignUniqueIndicesToPartitionColumns(List<HiveColumnHandle> columns)
+    {
+        // Gives a distinct hiveColumnIndex to partitioning columns. Columns are identified by these indices in the rest of the
+        // selective read path.
+        ImmutableList.Builder<HiveColumnHandle> newColumns = ImmutableList.builder();
+        int nextIndex = MAX_PARTITION_KEY_COLUMN_INDEX;
+        for (HiveColumnHandle column : columns) {
+            if (column.isPartitionKey()) {
+                newColumns.add(new HiveColumnHandle(column.getName(), column.getHiveType(), column.getTypeSignature(), nextIndex--, column.getColumnType(), column.getComment()));
+            }
+            else {
+                newColumns.add(column);
+            }
+        }
+        return newColumns.build();
+    }
+
     private static ConnectorPageSource createSelectivePageSource(
             Set<HiveSelectivePageSourceFactory> selectivePageSourceFactories,
             Configuration configuration,
diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java
index efb84d181..f56a5980d 100644
--- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java
+++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveIntegrationSmokeTest.java
@@ -2068,6 +2068,39 @@ public class TestHiveIntegrationSmokeTest
 
         assertQuery("SELECT * from test_metadata_delete", "SELECT orderkey, linenumber, linestatus FROM lineitem WHERE linestatus<>'O' and linenumber<>3");
 
+        Session session1 = Session.builder(getSession())
+                .setCatalogSessionProperty(getSession().getCatalog().get(), "orc_predicate_pushdown_enabled", "true")
+                .build();
+        assertQuery(session1, "SELECT * from test_metadata_delete", "SELECT orderkey, linenumber, linestatus FROM lineitem WHERE linestatus<>'O' and linenumber<>3");
+
+        @Language("SQL") String multiPartTable = "" +
+                "CREATE TABLE test_multi_part " +
+                "(" +
+                "  ID1 INTEGER," +
+                "  ID2 INTEGER," +
+                "  ID3 INTEGER," +
+                "  ID4 INTEGER," +
+                "  ID5 INTEGER," +
+                "  ID6 INTEGER," +
+                "  ID7 INTEGER," +
+                "  ID8 INTEGER," +
+                "  ID9 INTEGER," +
+                "  ID10 INTEGER," +
+                "  ID11 INTEGER," +
+                "  ID12 INTEGER," +
+                "  ID13 INTEGER," +
+                "  ID14 INTEGER " +
+                ") " +
+                "WITH (" +
+                PARTITIONED_BY_PROPERTY + " = ARRAY[ 'ID2','ID3','ID4','ID5','ID6','ID7','ID8','ID9','ID10','ID11','ID12','ID13','ID14']" +
+                ") ";
+
+        assertUpdate(multiPartTable);
+        assertUpdate("" +
+                        "INSERT INTO test_multi_part values(1,2,3,4,5,6,7,8,9,10,11,12,13,14) ",
+                "SELECT 1");
+        assertEquals(computeActual("SELECT *, \"$path\" FROM test_multi_part").getRowCount(), 1L);
+        assertUpdate("DROP TABLE test_multi_part");
         assertUpdate("DROP TABLE test_metadata_delete");
 
         assertFalse(getQueryRunner().tableExists(getSession(), "test_metadata_delete"));