diff --git a/asterix-algebra/src/main/java/edu/uci/ics/asterix/algebra/operators/physical/InvertedIndexPOperator.java b/asterix-algebra/src/main/java/edu/uci/ics/asterix/algebra/operators/physical/InvertedIndexPOperator.java
index ae2559b..63b02f4 100644
--- a/asterix-algebra/src/main/java/edu/uci/ics/asterix/algebra/operators/physical/InvertedIndexPOperator.java
+++ b/asterix-algebra/src/main/java/edu/uci/ics/asterix/algebra/operators/physical/InvertedIndexPOperator.java
@@ -166,7 +166,7 @@
                 searchModifierType, searchKeyType, secondaryIndex);
         InvertedIndexSearchOperatorDescriptor invIndexSearchOp = new InvertedIndexSearchOperatorDescriptor(jobSpec,
                 queryField, appContext.getStorageManagerInterface(), fileSplitProviders.first,
-                fileSplitProviders.second, appContext.getIndexRegistryProvider(), tokenTypeTraits,
+                fileSplitProviders.second, appContext.getIndexLifecycleManagerProvider(), tokenTypeTraits,
                 tokenComparatorFactories, invListsTypeTraits, invListsComparatorFactories,
                 new BTreeDataflowHelperFactory(), queryTokenizerFactory, searchModifierFactory, outputRecDesc,
                 retainInput, NoOpOperationCallbackProvider.INSTANCE);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
index b028e06..dc121d6 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/DatasetOperations.java
@@ -23,7 +23,7 @@
 import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
 import edu.uci.ics.asterix.common.config.GlobalConfig;
 import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.formats.base.IDataFormat;
@@ -68,12 +68,16 @@
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
 import edu.uci.ics.hyracks.dataflow.std.sort.ExternalSortOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexBulkLoadOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexCreateOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexDropOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackProvider;
+import edu.uci.ics.hyracks.storage.am.invertedindex.dataflow.InvertedIndexDataflowHelperFactory;
+import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeDataflowHelperFactory;
+import edu.uci.ics.hyracks.storage.am.rtree.frames.RTreePolicyType;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
 public class DatasetOperations {
@@ -92,7 +96,7 @@
 
         LOGGER.info("DROP DATASETPATH: " + datasetPath);
 
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
 
         Dataset dataset = metadata.findDataset(datasetName);
@@ -121,8 +125,25 @@
                     Pair<IFileSplitProvider, AlgebricksPartitionConstraint> idxSplitsAndConstraint = metadata
                             .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName,
                                     index.getIndexName());
+                    IIndexDataflowHelperFactory dfhFactory;
+                    switch (index.getIndexType()) {
+                        case BTREE:
+                            dfhFactory = new BTreeDataflowHelperFactory();
+                            break;
+                        case RTREE:
+                            dfhFactory = new RTreeDataflowHelperFactory(new IPrimitiveValueProviderFactory[] { null },
+                                    RTreePolicyType.RTREE);
+                            break;
+                        case NGRAM_INVIX:
+                        case WORD_INVIX:
+                            dfhFactory = new InvertedIndexDataflowHelperFactory();
+                            break;
+                        default:
+                            throw new AsterixException("Unknown index type provided.");
+                    }
                     TreeIndexDropOperatorDescriptor secondaryBtreeDrop = new TreeIndexDropOperatorDescriptor(specs[i],
-                            storageManager, indexRegistryProvider, idxSplitsAndConstraint.first);
+                            storageManager, indexLifecycleManagerProvider, idxSplitsAndConstraint.first,
+                            new ITypeTraits[] { null }, new IBinaryComparatorFactory[] { null }, dfhFactory);
                     AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(specs[i], secondaryBtreeDrop,
                             idxSplitsAndConstraint.second);
                     i++;
@@ -137,7 +158,8 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, datasetName);
         TreeIndexDropOperatorDescriptor primaryBtreeDrop = new TreeIndexDropOperatorDescriptor(specPrimary,
-                storageManager, indexRegistryProvider, splitsAndConstraint.first);
+                storageManager, indexLifecycleManagerProvider, splitsAndConstraint.first, new ITypeTraits[] { null },
+                new IBinaryComparatorFactory[] { null }, new BTreeDataflowHelperFactory());
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(specPrimary, primaryBtreeDrop,
                 splitsAndConstraint.second);
 
@@ -165,10 +187,10 @@
             sb.append(stringOf(fs[i]) + " ");
         }
         LOGGER.info("CREATING File Splits: " + sb.toString());
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
         TreeIndexCreateOperatorDescriptor indexCreateOp = new TreeIndexCreateOperatorDescriptor(spec, storageManager,
-                indexRegistryProvider, splitsAndConstraint.first, typeTraits, comparatorFactories,
+                indexLifecycleManagerProvider, splitsAndConstraint.first, typeTraits, comparatorFactories,
                 new BTreeDataflowHelperFactory(), NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, indexCreateOp,
                 splitsAndConstraint.second);
@@ -238,12 +260,12 @@
         }
         LOGGER.info("LOAD into File Splits: " + sb.toString());
 
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
         TreeIndexBulkLoadOperatorDescriptor btreeBulkLoad = new TreeIndexBulkLoadOperatorDescriptor(spec,
-                storageManager, indexRegistryProvider, splitsAndConstraint.first, typeTraits, comparatorFactories,
-                fieldPermutation, GlobalConfig.DEFAULT_BTREE_FILL_FACTOR, new BTreeDataflowHelperFactory(),
-                NoOpOperationCallbackProvider.INSTANCE);
+                storageManager, indexLifecycleManagerProvider, splitsAndConstraint.first, typeTraits,
+                comparatorFactories, fieldPermutation, GlobalConfig.DEFAULT_BTREE_FILL_FACTOR, false,
+                new BTreeDataflowHelperFactory(), NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, btreeBulkLoad,
                 splitsAndConstraint.second);
 
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
index 7bd7ae6..6183016 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/IndexOperations.java
@@ -2,7 +2,7 @@
 
 import edu.uci.ics.asterix.aql.translator.DdlTranslator.CompiledIndexDropStatement;
 import edu.uci.ics.asterix.common.config.OptimizationConfUtil;
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.metadata.MetadataException;
@@ -13,10 +13,12 @@
 import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
 import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
 import edu.uci.ics.hyracks.algebricks.core.rewriter.base.PhysicalOptimizationConfig;
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexDropOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
@@ -45,13 +47,14 @@
         String indexName = deleteStmt.getIndexName();
 
         JobSpecification spec = new JobSpecification();
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
 
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = datasetDecls
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         TreeIndexDropOperatorDescriptor btreeDrop = new TreeIndexDropOperatorDescriptor(spec, storageManager,
-                indexRegistryProvider, splitsAndConstraint.first);
+                indexLifecycleManagerProvider, splitsAndConstraint.first, new ITypeTraits[] { null },
+                new IBinaryComparatorFactory[] { null }, new BTreeDataflowHelperFactory());
         AlgebricksPartitionConstraintHelper
                 .setPartitionConstraintInJobSpec(spec, btreeDrop, splitsAndConstraint.second);
         spec.addRoot(btreeDrop);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
index b9ad908..cda22605 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryBTreeCreator.java
@@ -1,6 +1,6 @@
 package edu.uci.ics.asterix.file;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraintHelper;
@@ -29,7 +29,7 @@
     public JobSpecification buildCreationJobSpec() throws AsterixException, AlgebricksException {
         JobSpecification spec = new JobSpecification();
         TreeIndexCreateOperatorDescriptor secondaryIndexCreateOp = new TreeIndexCreateOperatorDescriptor(spec,
-                AsterixStorageManagerInterface.INSTANCE, AsterixIndexRegistryProvider.INSTANCE,
+                AsterixStorageManagerInterface.INSTANCE, AsterixIndexLifecycleManagerProvider.INSTANCE,
                 secondaryFileSplitProvider, secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories,
                 new BTreeDataflowHelperFactory(), NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, secondaryIndexCreateOp,
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
index 65f6f5e..73bf7c8 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryIndexCreator.java
@@ -4,7 +4,7 @@
 import java.util.List;
 
 import edu.uci.ics.asterix.common.config.DatasetConfig.DatasetType;
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.formats.nontagged.AqlBinaryBooleanInspectorImpl;
@@ -225,7 +225,7 @@
         // +Infinity
         int[] highKeyFields = null;
         BTreeSearchOperatorDescriptor primarySearchOp = new BTreeSearchOperatorDescriptor(spec, primaryRecDesc,
-                AsterixStorageManagerInterface.INSTANCE, AsterixIndexRegistryProvider.INSTANCE,
+                AsterixStorageManagerInterface.INSTANCE, AsterixIndexLifecycleManagerProvider.INSTANCE,
                 primaryFileSplitProvider, primaryRecDesc.getTypeTraits(), primaryComparatorFactories, lowKeyFields,
                 highKeyFields, true, true, new BTreeDataflowHelperFactory(), false,
                 NoOpOperationCallbackProvider.INSTANCE);
@@ -283,9 +283,9 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> secondarySplitsAndConstraint = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, secondaryIndexName);
         TreeIndexBulkLoadOperatorDescriptor treeIndexBulkLoadOp = new TreeIndexBulkLoadOperatorDescriptor(spec,
-                AsterixStorageManagerInterface.INSTANCE, AsterixIndexRegistryProvider.INSTANCE,
+                AsterixStorageManagerInterface.INSTANCE, AsterixIndexLifecycleManagerProvider.INSTANCE,
                 secondarySplitsAndConstraint.first, secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories,
-                fieldPermutation, fillFactor, dataflowHelperFactory, NoOpOperationCallbackProvider.INSTANCE);
+                fieldPermutation, fillFactor, false, dataflowHelperFactory, NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, treeIndexBulkLoadOp,
                 secondarySplitsAndConstraint.second);
         return treeIndexBulkLoadOp;
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
index d5e8222..41e5029 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryInvertedIndexCreator.java
@@ -2,7 +2,7 @@
 
 import java.util.List;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.metadata.entities.Index;
@@ -121,8 +121,8 @@
         JobSpecification spec = new JobSpecification();
         InvertedIndexCreateOperatorDescriptor invIndexCreateOp = new InvertedIndexCreateOperatorDescriptor(spec,
                 AsterixStorageManagerInterface.INSTANCE, fileSplitProviders.first, fileSplitProviders.second,
-                AsterixIndexRegistryProvider.INSTANCE, tokenTypeTraits, tokenComparatorFactories, invListsTypeTraits,
-                primaryComparatorFactories, tokenizerFactory, new BTreeDataflowHelperFactory(),
+                AsterixIndexLifecycleManagerProvider.INSTANCE, tokenTypeTraits, tokenComparatorFactories,
+                invListsTypeTraits, primaryComparatorFactories, tokenizerFactory, new BTreeDataflowHelperFactory(),
                 NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIndexCreateOp,
                 secondaryPartitionConstraint);
@@ -212,7 +212,7 @@
         }
         InvertedIndexBulkLoadOperatorDescriptor invIndexBulkLoadOp = new InvertedIndexBulkLoadOperatorDescriptor(spec,
                 fieldPermutation, AsterixStorageManagerInterface.INSTANCE, fileSplitProviders.first,
-                fileSplitProviders.second, AsterixIndexRegistryProvider.INSTANCE, tokenTypeTraits,
+                fileSplitProviders.second, AsterixIndexLifecycleManagerProvider.INSTANCE, tokenTypeTraits,
                 tokenComparatorFactories, invListsTypeTraits, primaryComparatorFactories, tokenizerFactory,
                 new BTreeDataflowHelperFactory(), NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, invIndexBulkLoadOp,
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
index 3127573..e297c2a 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/SecondaryRTreeCreator.java
@@ -2,7 +2,7 @@
 
 import java.util.List;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.dataflow.data.nontagged.valueproviders.AqlPrimitiveValueProviderFactory;
@@ -33,6 +33,7 @@
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexCreateOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackProvider;
 import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeDataflowHelperFactory;
+import edu.uci.ics.hyracks.storage.am.rtree.frames.RTreePolicyType;
 
 @SuppressWarnings("rawtypes")
 public class SecondaryRTreeCreator extends SecondaryIndexCreator {
@@ -48,9 +49,10 @@
     public JobSpecification buildCreationJobSpec() throws AsterixException, AlgebricksException {
         JobSpecification spec = new JobSpecification();
         TreeIndexCreateOperatorDescriptor secondaryIndexCreateOp = new TreeIndexCreateOperatorDescriptor(spec,
-                AsterixStorageManagerInterface.INSTANCE, AsterixIndexRegistryProvider.INSTANCE,
+                AsterixStorageManagerInterface.INSTANCE, AsterixIndexLifecycleManagerProvider.INSTANCE,
                 secondaryFileSplitProvider, secondaryRecDesc.getTypeTraits(), secondaryComparatorFactories,
-                new RTreeDataflowHelperFactory(valueProviderFactories), NoOpOperationCallbackProvider.INSTANCE);
+                new RTreeDataflowHelperFactory(valueProviderFactories, RTreePolicyType.RTREE),
+                NoOpOperationCallbackProvider.INSTANCE);
         AlgebricksPartitionConstraintHelper.setPartitionConstraintInJobSpec(spec, secondaryIndexCreateOp,
                 secondaryPartitionConstraint);
         spec.addRoot(secondaryIndexCreateOp);
@@ -124,8 +126,8 @@
 
         // Create secondary RTree bulk load op.
         TreeIndexBulkLoadOperatorDescriptor secondaryBulkLoadOp = createTreeIndexBulkLoadOp(spec,
-                numNestedSecondaryKeyFields, new RTreeDataflowHelperFactory(valueProviderFactories),
-                BTree.DEFAULT_FILL_FACTOR);
+                numNestedSecondaryKeyFields, new RTreeDataflowHelperFactory(valueProviderFactories,
+                        RTreePolicyType.RTREE), BTree.DEFAULT_FILL_FACTOR);
 
         // Connect the operators.
         spec.connect(new OneToOneConnectorDescriptor(spec), keyProviderOp, 0, primaryScanOp, 0);
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
index c295f55..fd59e85 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestKeywordIndexJob.java
@@ -6,7 +6,7 @@
 import java.util.HashMap;
 import java.util.List;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
@@ -40,8 +40,7 @@
 import edu.uci.ics.hyracks.dataflow.std.misc.PrinterOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackProvider;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
@@ -66,7 +65,7 @@
 
         // ---------- START GENERAL BTREE STUFF
 
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
 
         // ---------- END GENERAL BTREE STUFF
@@ -144,7 +143,7 @@
                 new FileSplit("nc1", new FileReference(new File("/tmp/nc1/demo1112/Customers_idx_NameInvIndex"))),
                 new FileSplit("nc2", new FileReference(new File("/tmp/nc2/demo1112/Customers_idx_NameInvIndex"))) });
         BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc,
-                storageManager, indexRegistryProvider, secondarySplitProvider, secondaryTypeTraits,
+                storageManager, indexLifecycleManagerProvider, secondarySplitProvider, secondaryTypeTraits,
                 secondaryComparatorFactories, lowKeyFields, highKeyFields, true, true,
                 new BTreeDataflowHelperFactory(), false, NoOpOperationCallbackProvider.INSTANCE);
         String[] secondarySearchOpLocationConstraint = new String[nodeGroup.size()];
diff --git a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
index 5d2cdc5..d98bb02 100644
--- a/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
+++ b/asterix-app/src/main/java/edu/uci/ics/asterix/file/TestSecondaryIndexJob.java
@@ -6,7 +6,7 @@
 import java.util.HashMap;
 import java.util.List;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.exceptions.AsterixException;
 import edu.uci.ics.asterix.dataflow.data.nontagged.comparators.AObjectAscBinaryComparatorFactory;
@@ -40,8 +40,7 @@
 import edu.uci.ics.hyracks.dataflow.std.misc.PrinterOperatorDescriptor;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeDataflowHelperFactory;
 import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeSearchOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallbackProvider;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
@@ -66,7 +65,7 @@
 
         // ---------- START GENERAL BTREE STUFF
 
-        IIndexRegistryProvider<IIndex> indexRegistryProvider = AsterixIndexRegistryProvider.INSTANCE;
+        IIndexLifecycleManagerProvider indexLifecycleManagerProvider = AsterixIndexLifecycleManagerProvider.INSTANCE;
         IStorageManagerInterface storageManager = AsterixStorageManagerInterface.INSTANCE;
 
         // ---------- END GENERAL BTREE STUFF
@@ -144,7 +143,7 @@
                 new FileSplit("nc1", new FileReference(new File("/tmp/nc1/demo1112/Customers_idx_NameBtreeIndex"))),
                 new FileSplit("nc2", new FileReference(new File("/tmp/nc2/demo1112/Customers_idx_NameBtreeIndex"))) });
         BTreeSearchOperatorDescriptor secondarySearchOp = new BTreeSearchOperatorDescriptor(spec, secondaryRecDesc,
-                storageManager, indexRegistryProvider, secondarySplitProvider, secondaryTypeTraits,
+                storageManager, indexLifecycleManagerProvider, secondarySplitProvider, secondaryTypeTraits,
                 secondaryComparatorFactories, lowKeyFields, highKeyFields, true, true,
                 new BTreeDataflowHelperFactory(), false, NoOpOperationCallbackProvider.INSTANCE);
         String[] secondarySearchOpLocationConstraint = new String[nodeGroup.size()];
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java
index 144a8824..569e27b 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/api/AsterixAppContextInfoImpl.java
@@ -3,11 +3,10 @@
 import java.util.Map;
 import java.util.Set;
 
-import edu.uci.ics.asterix.common.context.AsterixIndexRegistryProvider;
+import edu.uci.ics.asterix.common.context.AsterixIndexLifecycleManagerProvider;
 import edu.uci.ics.asterix.common.context.AsterixStorageManagerInterface;
 import edu.uci.ics.asterix.common.dataflow.IAsterixApplicationContextInfo;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
 public class AsterixAppContextInfoImpl implements IAsterixApplicationContextInfo {
@@ -20,11 +19,6 @@
     }
 
     @Override
-    public IIndexRegistryProvider<IIndex> getIndexRegistryProvider() {
-        return AsterixIndexRegistryProvider.INSTANCE;
-    }
-
-    @Override
     public IStorageManagerInterface getStorageManagerInterface() {
         return AsterixStorageManagerInterface.INSTANCE;
     }
@@ -37,4 +31,9 @@
         return nodeControllerMap;
     }
 
+    @Override
+    public IIndexLifecycleManagerProvider getIndexLifecycleManagerProvider() {
+        return AsterixIndexLifecycleManagerProvider.INSTANCE;
+    }
+
 }
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
index d676cb5..f3a841f 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixAppRuntimeContext.java
@@ -8,8 +8,9 @@
 import edu.uci.ics.asterix.transaction.management.service.transaction.TransactionProvider;
 import edu.uci.ics.hyracks.api.application.INCApplicationContext;
 import edu.uci.ics.hyracks.api.io.IIOManager;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexRegistry;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexLifecycleManager;
+import edu.uci.ics.hyracks.storage.common.TransientIndexArtifactMap;
 import edu.uci.ics.hyracks.storage.common.buffercache.BufferCache;
 import edu.uci.ics.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
 import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
@@ -18,12 +19,14 @@
 import edu.uci.ics.hyracks.storage.common.buffercache.IPageReplacementStrategy;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.storage.common.file.IIndexArtifactMap;
 
 public class AsterixAppRuntimeContext {
     private static final int DEFAULT_BUFFER_CACHE_PAGE_SIZE = 32768;
     private final INCApplicationContext ncApplicationContext;
-    
-    private IndexRegistry<IIndex> indexRegistry;
+
+    private IIndexArtifactMap indexArtifactMap;
+    private IIndexLifecycleManager indexLifecycleManager;
     private IFileMapManager fileMapManager;
     private IBufferCache bufferCache;
     private TransactionProvider provider;
@@ -36,19 +39,13 @@
         int pageSize = getBufferCachePageSize();
         int numPages = getBufferCacheNumPages();
 
-        // Initialize file map manager
+        indexArtifactMap = new TransientIndexArtifactMap();
         fileMapManager = new AsterixFileMapManager();
-
-        // Initialize the buffer cache
         ICacheMemoryAllocator allocator = new HeapBufferAllocator();
         IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
         IIOManager ioMgr = ncApplicationContext.getRootContext().getIOManager();
         bufferCache = new BufferCache(ioMgr, allocator, prs, fileMapManager, pageSize, numPages, Integer.MAX_VALUE);
-
-        // Initialize the index registry
-        indexRegistry = new IndexRegistry<IIndex>();
-
-        // Initialize the transaction sub-system
+        indexLifecycleManager = new IndexLifecycleManager();
         provider = new TransactionProvider(ncApplicationContext.getNodeId());
     }
 
@@ -107,12 +104,16 @@
         return fileMapManager;
     }
 
-    public IndexRegistry<IIndex> getIndexRegistry() {
-        return indexRegistry;
-    }
-
     public TransactionProvider getTransactionProvider() {
         return provider;
     }
 
+    public IIndexLifecycleManager getIndexLifecycleManager() {
+        return indexLifecycleManager;
+    }
+
+    public IIndexArtifactMap getIndexArtifactMap() {
+        return indexArtifactMap;
+    }
+
 }
\ No newline at end of file
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexLifecycleManagerProvider.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexLifecycleManagerProvider.java
new file mode 100644
index 0000000..31a7855
--- /dev/null
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexLifecycleManagerProvider.java
@@ -0,0 +1,16 @@
+package edu.uci.ics.asterix.common.context;
+
+import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManager;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
+
+public enum AsterixIndexLifecycleManagerProvider implements IIndexLifecycleManagerProvider {
+    INSTANCE;
+
+    @Override
+    public IIndexLifecycleManager getLifecycleManager(IHyracksTaskContext ctx) {
+        return ((AsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject())
+                .getIndexLifecycleManager();
+    }
+
+}
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexRegistryProvider.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexRegistryProvider.java
deleted file mode 100644
index bc59291..0000000
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixIndexRegistryProvider.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package edu.uci.ics.asterix.common.context;
-
-import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexRegistry;
-
-public class AsterixIndexRegistryProvider implements IIndexRegistryProvider<IIndex> {
-
-    private static final long serialVersionUID = 1L;
-
-    public static final AsterixIndexRegistryProvider INSTANCE = new AsterixIndexRegistryProvider();
-
-    private AsterixIndexRegistryProvider() {
-    }
-
-    @Override
-    public IndexRegistry<IIndex> getRegistry(IHyracksTaskContext ctx) {
-        return ((AsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject())
-                .getIndexRegistry();
-    }
-
-}
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
index a1b5a94..d206903 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/context/AsterixStorageManagerInterface.java
@@ -4,6 +4,7 @@
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.storage.common.file.IIndexArtifactMap;
 
 public class AsterixStorageManagerInterface implements IStorageManagerInterface {
     private static final long serialVersionUID = 1L;
@@ -21,4 +22,10 @@
         return ((AsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject())
                 .getFileMapManager();
     }
+
+    @Override
+    public IIndexArtifactMap getIndexArtifactMap(IHyracksTaskContext ctx) {
+        return ((AsterixAppRuntimeContext) ctx.getJobletContext().getApplicationContext().getApplicationObject())
+                .getIndexArtifactMap();
+    }
 }
diff --git a/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
index 7bb0fd6..200c2e19 100644
--- a/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
+++ b/asterix-common/src/main/java/edu/uci/ics/asterix/common/dataflow/IAsterixApplicationContextInfo.java
@@ -1,11 +1,10 @@
 package edu.uci.ics.asterix.common.dataflow;
 
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
 public interface IAsterixApplicationContextInfo {
-    public IIndexRegistryProvider<IIndex> getIndexRegistryProvider();
+    public IIndexLifecycleManagerProvider getIndexLifecycleManagerProvider();
 
     public IStorageManagerInterface getStorageManagerInterface();
 }
diff --git a/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorDescriptor.java b/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
index f96d3da..290d5ca 100644
--- a/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
+++ b/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorDescriptor.java
@@ -27,12 +27,11 @@
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
 import edu.uci.ics.hyracks.api.job.JobSpecification;
 import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManagerProvider;
 import edu.uci.ics.hyracks.storage.am.common.api.IOperationCallbackProvider;
 import edu.uci.ics.hyracks.storage.am.common.api.ITupleFilterFactory;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexDataflowHelperFactory;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOp;
 import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
 
@@ -53,12 +52,12 @@
      * once the LSM/Recovery/Transactions world has been introduced.
      */
     public TreeIndexInsertUpdateDeleteOperatorDescriptor(JobSpecification spec, RecordDescriptor recDesc,
-            IStorageManagerInterface storageManager, IIndexRegistryProvider<IIndex> indexRegistryProvider,
+            IStorageManagerInterface storageManager, IIndexLifecycleManagerProvider lifecycleManagerProvider,
             IFileSplitProvider fileSplitProvider, ITypeTraits[] typeTraits,
             IBinaryComparatorFactory[] comparatorFactories, int[] fieldPermutation, IndexOp op,
             IIndexDataflowHelperFactory dataflowHelperFactory, ITupleFilterFactory tupleFilterFactory,
             IOperationCallbackProvider opCallbackProvider, long transactionId) {
-        super(spec, 1, 1, recDesc, storageManager, indexRegistryProvider, fileSplitProvider, typeTraits,
+        super(spec, 1, 1, recDesc, storageManager, lifecycleManagerProvider, fileSplitProvider, typeTraits,
                 comparatorFactories, dataflowHelperFactory, tupleFilterFactory, false, opCallbackProvider);
         this.fieldPermutation = fieldPermutation;
         this.op = op;
diff --git a/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorNodePushable.java b/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorNodePushable.java
index 1fd8fee..e305677 100644
--- a/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorNodePushable.java
+++ b/asterix-hyracks-glue/src/main/java/edu/uci/ics/asterix/runtime/transaction/TreeIndexInsertUpdateDeleteOperatorNodePushable.java
@@ -18,7 +18,6 @@
 
 import edu.uci.ics.asterix.common.context.AsterixAppRuntimeContext;
 import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
-import edu.uci.ics.asterix.transaction.management.resource.ICloseable;
 import edu.uci.ics.asterix.transaction.management.resource.TransactionalResourceRepository;
 import edu.uci.ics.asterix.transaction.management.service.locking.ILockManager;
 import edu.uci.ics.asterix.transaction.management.service.logging.DataUtil;
@@ -37,30 +36,35 @@
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.FrameTupleReference;
 import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputUnaryOutputOperatorNodePushable;
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
+import edu.uci.ics.hyracks.storage.am.common.api.IModificationOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 import edu.uci.ics.hyracks.storage.am.common.api.ITupleFilter;
 import edu.uci.ics.hyracks.storage.am.common.api.ITupleFilterFactory;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.AbstractTreeIndexOperatorDescriptor;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.PermutingFrameTupleReference;
 import edu.uci.ics.hyracks.storage.am.common.dataflow.TreeIndexDataflowHelper;
+import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOp;
+import edu.uci.ics.hyracks.storage.common.file.IIndexArtifactMap;
 
 public class TreeIndexInsertUpdateDeleteOperatorNodePushable extends AbstractUnaryInputUnaryOutputOperatorNodePushable {
+    private final AbstractTreeIndexOperatorDescriptor opDesc;
     private final IHyracksTaskContext ctx;
+    private final TreeIndexDataflowHelper treeIndexHelper;
     private FrameTupleAccessor accessor;
-    private TreeIndexDataflowHelper treeIndexHelper;
     private final IRecordDescriptorProvider recordDescProvider;
     private final IndexOp op;
     private final PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
     private FrameTupleReference frameTuple;
     private ByteBuffer writeBuffer;
     private IIndexAccessor indexAccessor;
+    private ITupleFilter tupleFilter;
+    private IModificationOperationCallback modCallback;
     private ILockManager lockManager;
     private final TransactionContext txnContext;
     private TreeLogger treeLogger;
-    private ITupleFilter tupleFilter;
     private final TransactionProvider transactionProvider;
+    private byte[] resourceIDBytes;
 
     /* TODO: Index operators should live in Hyracks. Right now, they are needed here in Asterix
      * as a hack to provide transactionIDs. The Asterix verions of this operator will disappear 
@@ -70,9 +74,10 @@
     public TreeIndexInsertUpdateDeleteOperatorNodePushable(TransactionContext txnContext,
             AbstractTreeIndexOperatorDescriptor opDesc, IHyracksTaskContext ctx, int partition, int[] fieldPermutation,
             IRecordDescriptorProvider recordDescProvider, IndexOp op) {
+        this.opDesc = opDesc;
         this.ctx = ctx;
-        treeIndexHelper = (TreeIndexDataflowHelper) opDesc.getIndexDataflowHelperFactory().createIndexDataflowHelper(
-                opDesc, ctx, partition);
+        this.treeIndexHelper = (TreeIndexDataflowHelper) opDesc.getIndexDataflowHelperFactory()
+                .createIndexDataflowHelper(opDesc, ctx, partition);
         this.recordDescProvider = recordDescProvider;
         this.op = op;
         tuple.setFieldPermutation(fieldPermutation);
@@ -82,52 +87,49 @@
         transactionProvider = runtimeContext.getTransactionProvider();
     }
 
-    public void initializeTransactionSupport() {
+    public void initializeTransactionSupport(long resourceID) {
         TransactionalResourceRepository resourceRepository = transactionProvider.getTransactionalResourceRepository();
         IResourceManager resourceMgr = resourceRepository.getTransactionalResourceMgr(TreeResourceManager.ID);
         if (resourceMgr == null) {
             resourceRepository.registerTransactionalResourceManager(TreeResourceManager.ID, new TreeResourceManager(
                     transactionProvider));
         }
-        int fileId = treeIndexHelper.getIndexFileId();
-        byte[] resourceId = DataUtil.intToByteArray(fileId);
-        transactionProvider.getTransactionalResourceRepository().registerTransactionalResource(resourceId,
-                treeIndexHelper.getIndex());
+        resourceIDBytes = DataUtil.longToByteArray(resourceID);
+        transactionProvider.getTransactionalResourceRepository().registerTransactionalResource(resourceIDBytes,
+                treeIndexHelper.getIndexInstance());
         lockManager = transactionProvider.getLockManager();
-        treeLogger = transactionProvider.getTreeLoggerRepository().getTreeLogger(resourceId);
+        treeLogger = transactionProvider.getTreeLoggerRepository().getTreeLogger(resourceIDBytes);
     }
 
     @Override
     public void open() throws HyracksDataException {
-        AbstractTreeIndexOperatorDescriptor opDesc = (AbstractTreeIndexOperatorDescriptor) treeIndexHelper
-                .getOperatorDescriptor();
         RecordDescriptor inputRecDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getActivityId(), 0);
-        accessor = new FrameTupleAccessor(treeIndexHelper.getHyracksTaskContext().getFrameSize(), inputRecDesc);
-        writeBuffer = treeIndexHelper.getHyracksTaskContext().allocateFrame();
+        accessor = new FrameTupleAccessor(ctx.getFrameSize(), inputRecDesc);
+        writeBuffer = ctx.allocateFrame();
         writer.open();
+        treeIndexHelper.open();
+        ITreeIndex treeIndex = (ITreeIndex) treeIndexHelper.getIndexInstance();
         try {
-            treeIndexHelper.init(false);
-            ITreeIndex treeIndex = (ITreeIndex) treeIndexHelper.getIndex();
-            indexAccessor = treeIndex.createAccessor();
+            modCallback = opDesc.getOpCallbackProvider().getModificationOperationCallback(
+                    treeIndexHelper.getResourceID());
+            indexAccessor = treeIndex.createAccessor(modCallback, NoOpOperationCallback.INSTANCE);
             ITupleFilterFactory tupleFilterFactory = opDesc.getTupleFilterFactory();
             if (tupleFilterFactory != null) {
                 tupleFilter = tupleFilterFactory.createTupleFilter(ctx);
                 frameTuple = new FrameTupleReference();
             }
-            initializeTransactionSupport();
+            IIndexArtifactMap iam = opDesc.getStorageManager().getIndexArtifactMap(ctx);
+            long resourceID = iam.get(treeIndexHelper.getFileReference().getFile().getPath());
+            initializeTransactionSupport(resourceID);
         } catch (Exception e) {
-            // cleanup in case of failure
-            treeIndexHelper.deinit();
+            treeIndexHelper.close();
             throw new HyracksDataException(e);
         }
     }
 
     @Override
     public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
-        final IIndex treeIndex = treeIndexHelper.getIndex();
         accessor.reset(buffer);
-        int fileId = treeIndexHelper.getIndexFileId();
-        byte[] resourceId = DataUtil.intToByteArray(fileId);
         int tupleCount = accessor.getTupleCount();
         try {
             for (int i = 0; i < tupleCount; i++) {
@@ -138,32 +140,34 @@
                     }
                 }
                 tuple.reset(accessor, i);
+                lockManager.lock(txnContext, resourceIDBytes,
+                        TransactionManagementConstants.LockManagerConstants.LockMode.EXCLUSIVE);
                 switch (op) {
                     case INSERT: {
-                        lockManager.lock(txnContext, resourceId,
-                                TransactionManagementConstants.LockManagerConstants.LockMode.EXCLUSIVE);
                         indexAccessor.insert(tuple);
-                        treeLogger.generateLogRecord(transactionProvider, txnContext, op, tuple);
                         break;
                     }
-
+                    case UPDATE: {
+                        indexAccessor.update(tuple);
+                        break;
+                    }
+                    case UPSERT: {
+                        indexAccessor.upsert(tuple);
+                        break;
+                    }
                     case DELETE: {
-                        lockManager.lock(txnContext, resourceId,
-                                TransactionManagementConstants.LockManagerConstants.LockMode.EXCLUSIVE);
                         indexAccessor.delete(tuple);
-                        treeLogger.generateLogRecord(transactionProvider, txnContext, op, tuple);
                         break;
                     }
-
                     default: {
                         throw new HyracksDataException("Unsupported operation " + op
                                 + " in tree index InsertUpdateDelete operator");
                     }
                 }
+                treeLogger.generateLogRecord(transactionProvider, txnContext, op, tuple);
             }
         } catch (ACIDException ae) {
-            throw new HyracksDataException("exception in locking/logging during operation " + op + " on tree "
-                    + treeIndex, ae);
+            throw new HyracksDataException("exception in locking/logging during operation " + op, ae);
         } catch (Exception e) {
             e.printStackTrace();
             throw new HyracksDataException(e);
@@ -180,35 +184,23 @@
         try {
             writer.close();
         } finally {
-            txnContext.addCloseableResource(new ICloseable() {
-                @Override
-                public void close(TransactionContext txnContext) throws ACIDException {
-                    try {
-                        treeIndexHelper.deinit();
-                    } catch (Exception e) {
-                        throw new ACIDException(txnContext, "could not de-initialize " + treeIndexHelper, e);
-                    }
-                }
-            });
+            treeIndexHelper.close();
+            //            txnContext.addCloseableResource(new ICloseable() {
+            //                @Override
+            //                public void close(TransactionContext txnContext) throws ACIDException {
+            //                    try {
+            //                        treeIndexHelper.close();
+            //                    } catch (Exception e) {
+            //                        throw new ACIDException(txnContext, "could not de-initialize " + treeIndexHelper, e);
+            //                    }
+            //                }
+            //            });
         }
     }
 
     @Override
     public void fail() throws HyracksDataException {
-        try {
-            writer.fail();
-        } finally {
-            txnContext.addCloseableResource(new ICloseable() {
-                @Override
-                public void close(TransactionContext txnContext) throws ACIDException {
-                    try {
-                        treeIndexHelper.deinit();
-                    } catch (Exception e) {
-                        throw new ACIDException(txnContext, "could not de-initialize " + treeIndexHelper, e);
-                    }
-                }
-            });
-        }
+        writer.fail();
     }
 
 }
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
index 591154a..9b3e816 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/MetadataNode.java
@@ -68,12 +68,12 @@
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManager;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrame;
 import edu.uci.ics.hyracks.storage.am.common.api.TreeIndexException;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexRegistry;
+import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOp;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 
@@ -83,7 +83,7 @@
     // TODO: Temporary transactional resource id for metadata.
     private static final byte[] metadataResourceId = MetadataNode.class.toString().getBytes();
 
-    private IndexRegistry<IIndex> indexRegistry;
+    private IIndexLifecycleManager indexLifecycleManager;
     private TransactionProvider transactionProvider;
 
     public static final MetadataNode INSTANCE = new MetadataNode();
@@ -94,7 +94,7 @@
 
     public void initialize(AsterixAppRuntimeContext runtimeContext) {
         this.transactionProvider = runtimeContext.getTransactionProvider();
-        this.indexRegistry = runtimeContext.getIndexRegistry();
+        this.indexLifecycleManager = runtimeContext.getIndexLifecycleManager();
     }
 
     @Override
@@ -251,15 +251,17 @@
     }
 
     private void insertTupleIntoIndex(long txnId, IMetadataIndex index, ITupleReference tuple) throws Exception {
-        int fileId = index.getFileId();
-        BTree btree = (BTree) indexRegistry.get(fileId);
-        btree.open(fileId);
-        ITreeIndexAccessor indexAccessor = btree.createAccessor();
+        long resourceID = index.getResourceID();
+        BTree btree = (BTree) indexLifecycleManager.getIndex(resourceID);
+        indexLifecycleManager.open(resourceID);
+        ITreeIndexAccessor indexAccessor = btree.createAccessor(NoOpOperationCallback.INSTANCE,
+                NoOpOperationCallback.INSTANCE);
         TransactionContext txnCtx = transactionProvider.getTransactionManager().getTransactionContext(txnId);
         transactionProvider.getLockManager().lock(txnCtx, index.getResourceId(), LockMode.EXCLUSIVE);
         // TODO: fix exceptions once new BTree exception model is in hyracks.
         indexAccessor.insert(tuple);
         index.getTreeLogger().generateLogRecord(transactionProvider, txnCtx, IndexOp.INSERT, tuple);
+        indexLifecycleManager.close(resourceID);
     }
 
     @Override
@@ -496,11 +498,11 @@
     }
 
     private void deleteTupleFromIndex(long txnId, IMetadataIndex index, ITupleReference tuple) throws Exception {
-        int fileId = index.getFileId();
-        BTree btree = (BTree) indexRegistry.get(fileId);
-        btree.open(fileId);
-
-        ITreeIndexAccessor indexAccessor = btree.createAccessor();
+        long resourceID = index.getResourceID();
+        BTree btree = (BTree) indexLifecycleManager.getIndex(resourceID);
+        indexLifecycleManager.open(resourceID);
+        ITreeIndexAccessor indexAccessor = btree.createAccessor(NoOpOperationCallback.INSTANCE,
+                NoOpOperationCallback.INSTANCE);
         TransactionContext txnCtx = transactionProvider.getTransactionManager().getTransactionContext(txnId);
         // This lock is actually an upgrade, because a deletion must be preceded
         // by a search, in order to be able to undo an aborted deletion.
@@ -510,6 +512,7 @@
         transactionProvider.getLockManager().lock(txnCtx, index.getResourceId(), LockMode.EXCLUSIVE);
         indexAccessor.delete(tuple);
         index.getTreeLogger().generateLogRecord(transactionProvider, txnCtx, IndexOp.DELETE, tuple);
+        indexLifecycleManager.close(resourceID);
     }
 
     @Override
@@ -769,11 +772,12 @@
         TransactionContext txnCtx = transactionProvider.getTransactionManager().getTransactionContext(txnId);
         transactionProvider.getLockManager().lock(txnCtx, index.getResourceId(), LockMode.SHARED);
         IBinaryComparatorFactory[] comparatorFactories = index.getKeyBinaryComparatorFactory();
-        int fileId = index.getFileId();
-        BTree btree = (BTree) indexRegistry.get(fileId);
-        btree.open(fileId);
+        long resourceID = index.getResourceID();
+        BTree btree = (BTree) indexLifecycleManager.getIndex(resourceID);
+        indexLifecycleManager.open(resourceID);
         ITreeIndexFrame leafFrame = btree.getLeafFrameFactory().createFrame();
-        ITreeIndexAccessor indexAccessor = btree.createAccessor();
+        ITreeIndexAccessor indexAccessor = btree.createAccessor(NoOpOperationCallback.INSTANCE,
+                NoOpOperationCallback.INSTANCE);
         ITreeIndexCursor rangeCursor = new BTreeRangeSearchCursor((IBTreeLeafFrame) leafFrame, false);
         IBinaryComparator[] searchCmps = new IBinaryComparator[searchKey.getFieldCount()];
         for (int i = 0; i < searchKey.getFieldCount(); i++) {
@@ -794,6 +798,7 @@
         } finally {
             rangeCursor.close();
         }
+        indexLifecycleManager.close(resourceID);
     }
 
     // TODO: Can use Hyrack's TupleUtils for this, once we switch to a newer
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataIndex.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataIndex.java
index f9e5540..fb42a2e 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataIndex.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/api/IMetadataIndex.java
@@ -24,6 +24,7 @@
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 
 /**
@@ -58,12 +59,20 @@
 
     public ARecordType getPayloadRecordType();
 
+    public void setFile(FileReference file);
+
+    public FileReference getFile();
+
     public void setFileId(int fileId);
 
     public void initTreeLogger(ITreeIndex treeIndex) throws ACIDException;
 
     public int getFileId();
 
+    public void setResourceID(long resourceID);
+
+    public long getResourceID();
+
     public byte[] getResourceId();
 
     public TreeLogger getTreeLogger();
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
index 6c388c2..515d272 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataBootstrap.java
@@ -52,21 +52,21 @@
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.api.io.IIOManager;
 import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMInteriorFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.frames.BTreeNSMLeafFrameFactory;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.common.api.IFreePageManager;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexLifecycleManager;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrameFactory;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndex;
-import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexRegistry;
 import edu.uci.ics.hyracks.storage.am.common.frames.LIFOMetaDataFrameFactory;
 import edu.uci.ics.hyracks.storage.am.common.freepage.LinkedListFreePageManager;
-import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+import edu.uci.ics.hyracks.storage.common.file.IIndexArtifactMap;
 
 /**
  * Initializes the remote metadata storage facilities ("universe") using a
@@ -80,7 +80,9 @@
 public class MetadataBootstrap {
     private static IBufferCache bufferCache;
     private static IFileMapProvider fileMapProvider;
-    private static IndexRegistry<IIndex> btreeRegistry;
+    private static IIndexLifecycleManager indexLifecycleManager;
+    private static IIndexArtifactMap indexArtifactMap;
+    private static IIOManager ioManager;
 
     private static String metadataNodeName;
     private static String metadataStore;
@@ -133,17 +135,19 @@
             (new File(outputDir)).mkdirs();
         }
 
-        btreeRegistry = runtimeContext.getIndexRegistry();
+        indexLifecycleManager = runtimeContext.getIndexLifecycleManager();
+        indexArtifactMap = runtimeContext.getIndexArtifactMap();
         bufferCache = runtimeContext.getBufferCache();
         fileMapProvider = runtimeContext.getFileMapManager();
+        ioManager = ncApplicationContext.getRootContext().getIOManager();
 
-        // Create fileRefs to all BTree files and open them in BufferCache.
-        for (int i = 0; i < primaryIndexes.length; i++) {
-            openIndexFile(primaryIndexes[i]);
-        }
-        for (int i = 0; i < secondaryIndexes.length; i++) {
-            openIndexFile(secondaryIndexes[i]);
-        }
+        //        // Create fileRefs to all BTree files and open them in BufferCache.
+        //        for (int i = 0; i < primaryIndexes.length; i++) {
+        //            openIndexFile(primaryIndexes[i]);
+        //        }
+        //        for (int i = 0; i < secondaryIndexes.length; i++) {
+        //            openIndexFile(secondaryIndexes[i]);
+        //        }
 
         // Begin a transaction against the metadata.
         // Lock the metadata in X mode.
@@ -153,11 +157,11 @@
         try {
             if (isNewUniverse) {
                 for (int i = 0; i < primaryIndexes.length; i++) {
-                    createIndex(primaryIndexes[i]);
+                    enlistMetadataDataset(primaryIndexes[i], true);
                     registerTransactionalResource(primaryIndexes[i], resourceRepository);
                 }
                 for (int i = 0; i < secondaryIndexes.length; i++) {
-                    createIndex(secondaryIndexes[i]);
+                    enlistMetadataDataset(secondaryIndexes[i], true);
                     registerTransactionalResource(secondaryIndexes[i], resourceRepository);
                 }
                 insertInitialDataverses(mdTxnCtx);
@@ -169,11 +173,11 @@
                 LOGGER.info("FINISHED CREATING METADATA B-TREES.");
             } else {
                 for (int i = 0; i < primaryIndexes.length; i++) {
-                    enlistMetadataDataset(primaryIndexes[i]);
+                    enlistMetadataDataset(primaryIndexes[i], false);
                     registerTransactionalResource(primaryIndexes[i], resourceRepository);
                 }
                 for (int i = 0; i < secondaryIndexes.length; i++) {
-                    enlistMetadataDataset(secondaryIndexes[i]);
+                    enlistMetadataDataset(secondaryIndexes[i], false);
                     registerTransactionalResource(secondaryIndexes[i], resourceRepository);
                 }
                 LOGGER.info("FINISHED ENLISTMENT OF METADATA B-TREES.");
@@ -186,44 +190,24 @@
     }
 
     public static void stopUniverse() throws HyracksDataException {
-        try {
-            // Close all BTree files in BufferCache.
-            for (int i = 0; i < primaryIndexes.length; i++) {
-                bufferCache.closeFile(primaryIndexes[i].getFileId());
-            }
-            for (int i = 0; i < secondaryIndexes.length; i++) {
-                bufferCache.closeFile(secondaryIndexes[i].getFileId());
-            }
-        } catch (HyracksDataException e) {
-            // Ignore for now.
-            // TODO: If multiple NCs are running in the same VM, then we could
-            // have multiple NCs undeploying asterix concurrently.
-            // It would also mean that there is only one BufferCache. A
-            // pathological sequence of events would be that NC2
-            // closes the BufferCache and then NC1 enters this portion of the
-            // code and tries to close unopened files.
-            // What we really want is to check whether the BufferCache is open
-            // in a synchronized block.
-            // The BufferCache api currently does not allow us to check for
-            // openness.
-            // Swallowing the exceptions is a simple fix for now.
+        // Close all BTree files in BufferCache.
+        for (int i = 0; i < primaryIndexes.length; i++) {
+            long resourceID = indexArtifactMap.get(primaryIndexes[i].getFile().getFile().getPath());
+            indexLifecycleManager.close(resourceID);
+            indexLifecycleManager.unregister(resourceID);
         }
-    }
-
-    private static void openIndexFile(IMetadataIndex index) throws HyracksDataException, ACIDException {
-        String filePath = metadataStore + index.getFileNameRelativePath();
-        FileReference file = new FileReference(new File(filePath));
-        bufferCache.createFile(file);
-        int fileId = fileMapProvider.lookupFileId(file);
-        bufferCache.openFile(fileId);
-        index.setFileId(fileId);
+        for (int i = 0; i < secondaryIndexes.length; i++) {
+            long resourceID = indexArtifactMap.get(secondaryIndexes[i].getFile().getFile().getPath());
+            indexLifecycleManager.close(resourceID);
+            indexLifecycleManager.unregister(resourceID);
+        }
     }
 
     private static void registerTransactionalResource(IMetadataIndex index,
             TransactionalResourceRepository resourceRepository) throws ACIDException {
-        int fileId = index.getFileId();
-        ITreeIndex treeIndex = (ITreeIndex) btreeRegistry.get(fileId);
-        byte[] resourceId = DataUtil.intToByteArray(fileId);
+        long resourceID = index.getResourceID();
+        ITreeIndex treeIndex = (ITreeIndex) indexLifecycleManager.getIndex(resourceID);
+        byte[] resourceId = DataUtil.longToByteArray(resourceID);
         resourceRepository.registerTransactionalResource(resourceId, treeIndex);
         index.initTreeLogger(treeIndex);
     }
@@ -304,33 +288,29 @@
 
     }
 
-    public static void createIndex(IMetadataIndex dataset) throws Exception {
-        int fileId = dataset.getFileId();
-        ITypeTraits[] typeTraits = dataset.getTypeTraits();
-        IBinaryComparatorFactory[] comparatorFactories = dataset.getKeyBinaryComparatorFactory();
+    public static void enlistMetadataDataset(IMetadataIndex index, boolean create) throws Exception {
+        String filePath = metadataStore + index.getFileNameRelativePath();
+        FileReference file = new FileReference(new File(filePath));
+        ITypeTraits[] typeTraits = index.getTypeTraits();
+        IBinaryComparatorFactory[] comparatorFactories = index.getKeyBinaryComparatorFactory();
         TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
         ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
         ITreeIndexMetaDataFrameFactory metaDataFrameFactory = new LIFOMetaDataFrameFactory();
         IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, 0, metaDataFrameFactory);
-        BTree btree = new BTree(bufferCache, NoOpOperationCallback.INSTANCE, typeTraits.length, comparatorFactories,
-                freePageManager, interiorFrameFactory, leafFrameFactory);
-        btree.create(fileId);
-        btreeRegistry.register(fileId, btree);
-    }
-
-    public static void enlistMetadataDataset(IMetadataIndex dataset) throws Exception {
-        int fileId = dataset.getFileId();
-        ITypeTraits[] typeTraits = dataset.getTypeTraits();
-        IBinaryComparatorFactory[] comparatorFactories = dataset.getKeyBinaryComparatorFactory();
-        TypeAwareTupleWriterFactory tupleWriterFactory = new TypeAwareTupleWriterFactory(typeTraits);
-        ITreeIndexFrameFactory leafFrameFactory = new BTreeNSMLeafFrameFactory(tupleWriterFactory);
-        ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(tupleWriterFactory);
-        ITreeIndexMetaDataFrameFactory metaDataFrameFactory = new LIFOMetaDataFrameFactory();
-        IFreePageManager freePageManager = new LinkedListFreePageManager(bufferCache, 0, metaDataFrameFactory);
-        BTree btree = new BTree(bufferCache, NoOpOperationCallback.INSTANCE, typeTraits.length, comparatorFactories,
-                freePageManager, interiorFrameFactory, leafFrameFactory);
-        btreeRegistry.register(fileId, btree);
+        BTree btree = new BTree(bufferCache, fileMapProvider, freePageManager, interiorFrameFactory, leafFrameFactory,
+                comparatorFactories, typeTraits.length, file);
+        long resourceID = -1;
+        if (create) {
+            btree.create();
+            resourceID = indexArtifactMap.create(file.getFile().getPath(), ioManager.getIODevices());
+        } else {
+            resourceID = indexArtifactMap.get(file.getFile().getPath());
+        }
+        index.setResourceID(resourceID);
+        index.setFile(file);
+        indexLifecycleManager.register(resourceID, btree);
+        indexLifecycleManager.open(resourceID);
     }
 
     public static String getOutputDir() {
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataIndex.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataIndex.java
index 165a605..baa792c 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataIndex.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/bootstrap/MetadataIndex.java
@@ -30,12 +30,12 @@
 import edu.uci.ics.asterix.transaction.management.exception.ACIDException;
 import edu.uci.ics.asterix.transaction.management.service.logging.DataUtil;
 import edu.uci.ics.asterix.transaction.management.service.logging.TreeLogger;
-import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.OrderOperator.IOrder.OrderKind;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.IBinaryHashFunctionFactory;
 import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
 import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 
 /**
@@ -62,8 +62,11 @@
     protected final IBinaryComparatorFactory[] bcfs;
     // Hash function factories for key fields of btree tuple. Created in c'tor.
     protected final IBinaryHashFunctionFactory[] bhffs;
+
+    protected FileReference file;
     // Identifier of file BufferCache backing this metadata btree index.
     protected int fileId;
+    protected long resourceID;
     // Resource id of this index for use in transactions.
     protected byte[] indexResourceId;
     // Logger for tree indexes.
@@ -116,8 +119,7 @@
         // Create binary comparator factories.
         bcfs = new IBinaryComparatorFactory[keyTypes.length];
         for (int i = 0; i < keyTypes.length; i++) {
-            bcfs[i] = AqlBinaryComparatorFactoryProvider.INSTANCE
-                    .getBinaryComparatorFactory(keyTypes[i], true);
+            bcfs[i] = AqlBinaryComparatorFactoryProvider.INSTANCE.getBinaryComparatorFactory(keyTypes[i], true);
         }
         // Create binary hash function factories.
         bhffs = new IBinaryHashFunctionFactory[keyTypes.length];
@@ -224,4 +226,25 @@
     public TreeLogger getTreeLogger() {
         return treeLogger;
     }
+
+    @Override
+    public void setFile(FileReference file) {
+        this.file = file;
+    }
+
+    @Override
+    public FileReference getFile() {
+        return this.file;
+    }
+
+    @Override
+    public void setResourceID(long resourceID) {
+        this.resourceID = resourceID;
+        this.indexResourceId = DataUtil.longToByteArray(resourceID);
+    }
+
+    @Override
+    public long getResourceID() {
+        return resourceID;
+    }
 }
\ No newline at end of file
diff --git a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
index b8f0258..275dee3 100644
--- a/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
+++ b/asterix-metadata/src/main/java/edu/uci/ics/asterix/metadata/declared/AqlMetadataProvider.java
@@ -89,6 +89,7 @@
 import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriterFactory;
 import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeDataflowHelperFactory;
 import edu.uci.ics.hyracks.storage.am.rtree.dataflow.RTreeSearchOperatorDescriptor;
+import edu.uci.ics.hyracks.storage.am.rtree.frames.RTreePolicyType;
 
 public class AqlMetadataProvider implements IMetadataProvider<AqlSourceId, String> {
     private final long txnId;
@@ -315,8 +316,8 @@
             throw new AlgebricksException(e);
         }
         BTreeSearchOperatorDescriptor btreeSearchOp = new BTreeSearchOperatorDescriptor(jobSpec, outputRecDesc,
-                appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), spPc.first, typeTraits,
-                comparatorFactories, lowKeyFields, highKeyFields, lowKeyInclusive, highKeyInclusive,
+                appContext.getStorageManagerInterface(), appContext.getIndexLifecycleManagerProvider(), spPc.first,
+                typeTraits, comparatorFactories, lowKeyFields, highKeyFields, lowKeyInclusive, highKeyInclusive,
                 new BTreeDataflowHelperFactory(), retainInput, NoOpOperationCallbackProvider.INSTANCE);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeSearchOp, spPc.second);
     }
@@ -395,9 +396,9 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> spPc = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         RTreeSearchOperatorDescriptor rtreeSearchOp = new RTreeSearchOperatorDescriptor(jobSpec, recDesc,
-                appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(), spPc.first, typeTraits,
-                comparatorFactories, keyFields, new RTreeDataflowHelperFactory(valueProviderFactories), false,
-                NoOpOperationCallbackProvider.INSTANCE);
+                appContext.getStorageManagerInterface(), appContext.getIndexLifecycleManagerProvider(), spPc.first,
+                typeTraits, comparatorFactories, keyFields, new RTreeDataflowHelperFactory(valueProviderFactories,
+                        RTreePolicyType.RTREE), false, NoOpOperationCallbackProvider.INSTANCE);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(rtreeSearchOp, spPc.second);
     }
 
@@ -504,9 +505,9 @@
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         IAsterixApplicationContextInfo appContext = (IAsterixApplicationContextInfo) context.getAppContext();
         TreeIndexBulkLoadOperatorDescriptor btreeBulkLoad = new TreeIndexBulkLoadOperatorDescriptor(spec,
-                appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(),
+                appContext.getStorageManagerInterface(), appContext.getIndexLifecycleManagerProvider(),
                 splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation,
-                GlobalConfig.DEFAULT_BTREE_FILL_FACTOR, new BTreeDataflowHelperFactory(),
+                GlobalConfig.DEFAULT_BTREE_FILL_FACTOR, false, new BTreeDataflowHelperFactory(),
                 NoOpOperationCallbackProvider.INSTANCE);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second);
     }
@@ -545,9 +546,10 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         TreeIndexInsertUpdateDeleteOperatorDescriptor btreeBulkLoad = new TreeIndexInsertUpdateDeleteOperatorDescriptor(
-                spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(),
-                splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, indexOp,
-                new BTreeDataflowHelperFactory(), null, NoOpOperationCallbackProvider.INSTANCE, txnId);
+                spec, recordDesc, appContext.getStorageManagerInterface(),
+                appContext.getIndexLifecycleManagerProvider(), splitsAndConstraint.first, typeTraits,
+                comparatorFactories, fieldPermutation, indexOp, new BTreeDataflowHelperFactory(), null,
+                NoOpOperationCallbackProvider.INSTANCE, txnId);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second);
     }
 
@@ -687,9 +689,10 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         TreeIndexInsertUpdateDeleteOperatorDescriptor btreeBulkLoad = new TreeIndexInsertUpdateDeleteOperatorDescriptor(
-                spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(),
-                splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, indexOp,
-                new BTreeDataflowHelperFactory(), filterFactory, NoOpOperationCallbackProvider.INSTANCE, txnId);
+                spec, recordDesc, appContext.getStorageManagerInterface(),
+                appContext.getIndexLifecycleManagerProvider(), splitsAndConstraint.first, typeTraits,
+                comparatorFactories, fieldPermutation, indexOp, new BTreeDataflowHelperFactory(), filterFactory,
+                NoOpOperationCallbackProvider.INSTANCE, txnId);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(btreeBulkLoad, splitsAndConstraint.second);
     }
 
@@ -748,10 +751,10 @@
         Pair<IFileSplitProvider, AlgebricksPartitionConstraint> splitsAndConstraint = metadata
                 .splitProviderAndPartitionConstraintsForInternalOrFeedDataset(datasetName, indexName);
         TreeIndexInsertUpdateDeleteOperatorDescriptor rtreeUpdate = new TreeIndexInsertUpdateDeleteOperatorDescriptor(
-                spec, recordDesc, appContext.getStorageManagerInterface(), appContext.getIndexRegistryProvider(),
-                splitsAndConstraint.first, typeTraits, comparatorFactories, fieldPermutation, indexOp,
-                new RTreeDataflowHelperFactory(valueProviderFactories), filterFactory,
-                NoOpOperationCallbackProvider.INSTANCE, txnId);
+                spec, recordDesc, appContext.getStorageManagerInterface(),
+                appContext.getIndexLifecycleManagerProvider(), splitsAndConstraint.first, typeTraits,
+                comparatorFactories, fieldPermutation, indexOp, new RTreeDataflowHelperFactory(valueProviderFactories,
+                        RTreePolicyType.RTREE), filterFactory, NoOpOperationCallbackProvider.INSTANCE, txnId);
         return new Pair<IOperatorDescriptor, AlgebricksPartitionConstraint>(rtreeUpdate, splitsAndConstraint.second);
     }
 
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/DataUtil.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/DataUtil.java
index 41a2a52..4b8d62f 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/DataUtil.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/DataUtil.java
@@ -47,4 +47,17 @@
         return bytes;
     }
 
+    public static byte[] longToByteArray(long value) {
+        byte[] bytes = new byte[8];
+        bytes[0] = (byte) ((value >>> 56) & 0xFF);
+        bytes[1] = (byte) ((value >>> 48) & 0xFF);
+        bytes[2] = (byte) ((value >>> 40) & 0xFF);
+        bytes[3] = (byte) ((value >>> 32) & 0xFF);
+        bytes[4] = (byte) ((value >>> 24) & 0xFF);
+        bytes[5] = (byte) ((value >>> 16) & 0xFF);
+        bytes[6] = (byte) ((value >>> 8) & 0xFF);
+        bytes[7] = (byte) ((value >>> 0) & 0xFF);
+        return bytes;
+    }
+
 }
\ No newline at end of file
diff --git a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/TreeResourceManager.java b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/TreeResourceManager.java
index 7c6d05a..4e6ad80 100644
--- a/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/TreeResourceManager.java
+++ b/asterix-transactions/src/main/java/edu/uci/ics/asterix/transaction/management/service/logging/TreeResourceManager.java
@@ -20,6 +20,7 @@
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexAccessor;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleReference;
+import edu.uci.ics.hyracks.storage.am.common.impls.NoOpOperationCallback;
 
 public class TreeResourceManager implements IResourceManager {
 
@@ -36,7 +37,6 @@
     }
 
     public void undo(ILogRecordHelper logRecordHelper, LogicalLogLocator logLocator) throws ACIDException {
-
         int logContentBeginPos = logRecordHelper.getLogContentBeginPos(logLocator);
         byte[] logBufferContent = logLocator.getBuffer().getArray();
         // read the length of resource id byte array
@@ -58,7 +58,8 @@
         tupleReference.setFieldCount(tupleReference.getFieldCount());
         tupleReference.resetByTupleOffset(logLocator.getBuffer().getByteBuffer(), tupleBeginPos);
         byte operation = logBufferContent[operationOffset];
-        IIndexAccessor treeIndexAccessor = treeIndex.createAccessor();
+        IIndexAccessor treeIndexAccessor = treeIndex.createAccessor(NoOpOperationCallback.INSTANCE,
+                NoOpOperationCallback.INSTANCE);
         try {
             switch (operation) {
                 case TreeLogger.BTreeOperationCodes.INSERT: