diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/IBTreeOperatorDescriptorHelper.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/IBTreeOperatorDescriptorHelper.java
new file mode 100644
index 0000000..522076f
--- /dev/null
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/dataflow/IBTreeOperatorDescriptorHelper.java
@@ -0,0 +1,30 @@
+package edu.uci.ics.hyracks.storage.am.btree.dataflow;
+
+import edu.uci.ics.hyracks.api.dataflow.IActivityNode;
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
+
+public interface IBTreeOperatorDescriptorHelper extends IActivityNode {
+    public IFileSplitProvider getBTreeFileSplitProvider();
+
+    public IBinaryComparatorFactory[] getBTreeComparatorFactories();
+
+    public ITypeTrait[] getBTreeTypeTraits();
+
+    public IBTreeInteriorFrameFactory getBTreeInteriorFactory();
+
+    public IBTreeLeafFrameFactory getBTreeLeafFactory();
+
+    public IStorageManagerInterface getStorageManager();
+
+    public IIndexRegistryProvider<BTree> getBTreeRegistryProvider();
+
+    public RecordDescriptor getRecordDescriptor();
+}
diff --git a/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexHelperOpenMode.java b/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexHelperOpenMode.java
new file mode 100644
index 0000000..0b27a0e
--- /dev/null
+++ b/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/dataflow/IndexHelperOpenMode.java
@@ -0,0 +1,5 @@
+package edu.uci.ics.hyracks.storage.am.common.dataflow;
+
+public enum IndexHelperOpenMode {
+	OPEN, CREATE, ENLIST
+}
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/AbstractInvertedIndexOperatorDescriptor.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/AbstractInvertedIndexOperatorDescriptor.java
new file mode 100644
index 0000000..d89e4c2
--- /dev/null
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/AbstractInvertedIndexOperatorDescriptor.java
@@ -0,0 +1,143 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package edu.uci.ics.hyracks.storage.am.invertedindex.dataflow;
+
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
+import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.invertedindex.impls.InvertedIndex;
+import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
+
+public abstract class AbstractInvertedIndexOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor
+        implements IInvertedIndexOperatorDescriptorHelper {
+
+    private static final long serialVersionUID = 1L;
+
+    // general
+    protected final IStorageManagerInterface storageManager;
+
+    // btree
+    protected final IFileSplitProvider btreeFileSplitProvider;
+    protected final IIndexRegistryProvider<BTree> btreeRegistryProvider;
+    protected final IBTreeInteriorFrameFactory interiorFrameFactory;
+    protected final IBTreeLeafFrameFactory leafFrameFactory;
+    protected final ITypeTrait[] btreeTypeTraits;
+    protected final IBinaryComparatorFactory[] btreeComparatorFactories;
+
+    // inverted index
+    protected final IFileSplitProvider invIndexFileSplitProvider;
+    protected final IIndexRegistryProvider<InvertedIndex> invIndexRegistryProvider;    
+    protected final ITypeTrait[] invIndexTypeTraits;
+    protected final IBinaryComparatorFactory[] invIndexComparatorFactories;
+
+    public AbstractInvertedIndexOperatorDescriptor(JobSpecification spec, int inputArity, int outputArity,
+            RecordDescriptor recDesc, IStorageManagerInterface storageManager,
+            IFileSplitProvider btreeFileSplitProvider, IIndexRegistryProvider<BTree> btreeRegistryProvider,
+            IBTreeInteriorFrameFactory interiorFrameFactory, IBTreeLeafFrameFactory leafFrameFactory,
+            ITypeTrait[] btreeTypeTraits, IBinaryComparatorFactory[] btreeComparatorFactories, float btreeFillFactor,
+            IFileSplitProvider invIndexFileSplitProvider,
+            IIndexRegistryProvider<InvertedIndex> invIndexRegistryProvider, ITypeTrait[] invIndexTypeTraits,
+            IBinaryComparatorFactory[] invIndexComparatorFactories) {
+        super(spec, inputArity, outputArity);
+
+        // general
+        this.storageManager = storageManager;
+
+        // btree
+        this.btreeFileSplitProvider = btreeFileSplitProvider;
+        this.btreeRegistryProvider = btreeRegistryProvider;
+        this.interiorFrameFactory = interiorFrameFactory;
+        this.leafFrameFactory = leafFrameFactory;
+        this.btreeTypeTraits = btreeTypeTraits;
+        this.btreeComparatorFactories = btreeComparatorFactories;
+
+        // inverted index
+        this.invIndexFileSplitProvider = invIndexFileSplitProvider;
+        this.invIndexRegistryProvider = invIndexRegistryProvider;
+        this.invIndexTypeTraits = invIndexTypeTraits;
+        this.invIndexComparatorFactories = invIndexComparatorFactories;
+
+        if (outputArity > 0)
+            recordDescriptors[0] = recDesc;
+    }
+
+    @Override
+    public IFileSplitProvider getBTreeFileSplitProvider() {
+        return btreeFileSplitProvider;
+    }
+
+    @Override
+    public IBinaryComparatorFactory[] getBTreeComparatorFactories() {
+        return btreeComparatorFactories;
+    }
+
+    @Override
+    public ITypeTrait[] getBTreeTypeTraits() {
+        return btreeTypeTraits;
+    }
+
+    @Override
+    public IBTreeInteriorFrameFactory getBTreeInteriorFactory() {
+        return interiorFrameFactory;
+    }
+
+    @Override
+    public IBTreeLeafFrameFactory getBTreeLeafFactory() {
+        return leafFrameFactory;
+    }
+
+    @Override
+    public IStorageManagerInterface getStorageManager() {
+        return storageManager;
+    }
+
+    @Override
+    public IIndexRegistryProvider<BTree> getBTreeRegistryProvider() {
+        return btreeRegistryProvider;
+    }
+
+    @Override
+    public RecordDescriptor getRecordDescriptor() {
+        return recordDescriptors[0];
+    }
+
+    @Override
+    public IIndexRegistryProvider<InvertedIndex> getInvIndexRegistryProvider() {
+        return invIndexRegistryProvider;
+    }
+
+    @Override
+    public IBinaryComparatorFactory[] getInvIndexComparatorFactories() {
+        return invIndexComparatorFactories;
+    }
+
+    @Override
+    public IFileSplitProvider getInvIndexFileSplitProvider() {
+        return invIndexFileSplitProvider;
+    }
+
+    @Override
+    public ITypeTrait[] getInvIndexTypeTraits() {
+        return invIndexTypeTraits;
+    }
+}
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/IInvertedIndexOperatorDescriptorHelper.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/IInvertedIndexOperatorDescriptorHelper.java
new file mode 100644
index 0000000..c7024a1
--- /dev/null
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/IInvertedIndexOperatorDescriptorHelper.java
@@ -0,0 +1,18 @@
+package edu.uci.ics.hyracks.storage.am.invertedindex.dataflow;
+
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeOperatorDescriptorHelper;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.invertedindex.impls.InvertedIndex;
+
+public interface IInvertedIndexOperatorDescriptorHelper extends IBTreeOperatorDescriptorHelper {            
+    public IFileSplitProvider getInvIndexFileSplitProvider();
+
+    public IBinaryComparatorFactory[] getInvIndexComparatorFactories();
+
+    public ITypeTrait[] getInvIndexTypeTraits();
+    
+    public IIndexRegistryProvider<InvertedIndex> getInvIndexRegistryProvider();
+}
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorDescriptor.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorDescriptor.java
new file mode 100644
index 0000000..439ca94
--- /dev/null
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorDescriptor.java
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package edu.uci.ics.hyracks.storage.am.invertedindex.dataflow;
+
+import edu.uci.ics.hyracks.api.context.IHyracksStageletContext;
+import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.ITypeTrait;
+import edu.uci.ics.hyracks.api.job.IOperatorEnvironment;
+import edu.uci.ics.hyracks.api.job.JobSpecification;
+import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrameFactory;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IIndexRegistryProvider;
+import edu.uci.ics.hyracks.storage.am.invertedindex.api.IInvertedListBuilder;
+import edu.uci.ics.hyracks.storage.am.invertedindex.impls.InvertedIndex;
+import edu.uci.ics.hyracks.storage.common.IStorageManagerInterface;
+
+public class InvertedIndexBulkLoadOperatorDescriptor extends AbstractInvertedIndexOperatorDescriptor {
+
+    private static final long serialVersionUID = 1L;
+
+    private final int[] fieldPermutation;
+    private final float btreeFillFactor;
+    private final IInvertedListBuilder invListBuilder;
+
+    public InvertedIndexBulkLoadOperatorDescriptor(JobSpecification spec, IStorageManagerInterface storageManager,
+            int[] fieldPermutation, IFileSplitProvider btreeFileSplitProvider,
+            IIndexRegistryProvider<BTree> btreeRegistryProvider, IBTreeInteriorFrameFactory interiorFrameFactory,
+            IBTreeLeafFrameFactory leafFrameFactory, ITypeTrait[] btreeTypeTraits,
+            IBinaryComparatorFactory[] btreeComparatorFactories, float btreeFillFactor,
+            IFileSplitProvider invIndexFileSplitProvider,
+            IIndexRegistryProvider<InvertedIndex> invIndexRegistryProvider, ITypeTrait[] invIndexTypeTraits,
+            IBinaryComparatorFactory[] invIndexComparatorFactories, IInvertedListBuilder invListBuilder) {
+        super(spec, 1, 0, null, storageManager, btreeFileSplitProvider, btreeRegistryProvider, interiorFrameFactory,
+                leafFrameFactory, btreeTypeTraits, btreeComparatorFactories, btreeFillFactor,
+                invIndexFileSplitProvider, invIndexRegistryProvider, invIndexTypeTraits, invIndexComparatorFactories);
+        this.fieldPermutation = fieldPermutation;
+        this.btreeFillFactor = btreeFillFactor;
+        this.invListBuilder = invListBuilder;
+    }
+
+    @Override
+    public IOperatorNodePushable createPushRuntime(IHyracksStageletContext ctx, IOperatorEnvironment env,
+            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
+        return new InvertedIndexBulkLoadOperatorNodePushable(this, ctx, partition, fieldPermutation, btreeFillFactor,
+                invListBuilder, recordDescProvider);
+    }
+}
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorNodePushable.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorNodePushable.java
new file mode 100644
index 0000000..21933df
--- /dev/null
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexBulkLoadOperatorNodePushable.java
@@ -0,0 +1,109 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.storage.am.invertedindex.dataflow;
+
+import java.nio.ByteBuffer;
+
+import edu.uci.ics.hyracks.api.context.IHyracksStageletContext;
+import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
+import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.dataflow.common.comm.io.FrameTupleAccessor;
+import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryInputSinkOperatorNodePushable;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.BTreeOpHelper;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.PermutingFrameTupleReference;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexHelperOpenMode;
+import edu.uci.ics.hyracks.storage.am.invertedindex.api.IInvertedListBuilder;
+import edu.uci.ics.hyracks.storage.am.invertedindex.impls.InvertedIndex;
+
+public class InvertedIndexBulkLoadOperatorNodePushable extends AbstractUnaryInputSinkOperatorNodePushable {
+    private final BTreeOpHelper btreeOpHelper;
+    private float btreeFillFactor;
+    
+    private final InvertedIndexOpHelper invIndexOpHelper;        
+    protected final IInvertedListBuilder invListBuilder;
+    private InvertedIndex.BulkLoadContext bulkLoadCtx;
+
+    private final IHyracksStageletContext ctx;
+    
+    private FrameTupleAccessor accessor;
+    private PermutingFrameTupleReference tuple = new PermutingFrameTupleReference();
+    
+    private IRecordDescriptorProvider recordDescProvider;
+    
+    public InvertedIndexBulkLoadOperatorNodePushable(AbstractInvertedIndexOperatorDescriptor opDesc,
+            IHyracksStageletContext ctx, int partition, int[] fieldPermutation, float btreeFillFactor,
+            IInvertedListBuilder invListBuilder, IRecordDescriptorProvider recordDescProvider) {
+        btreeOpHelper = new BTreeOpHelper(opDesc, ctx, partition, IndexHelperOpenMode.CREATE);
+        invIndexOpHelper = new InvertedIndexOpHelper(opDesc, ctx, partition, IndexHelperOpenMode.CREATE);
+        this.btreeFillFactor = btreeFillFactor;
+        this.recordDescProvider = recordDescProvider;
+        this.ctx = ctx;
+        this.invListBuilder = invListBuilder;
+        tuple.setFieldPermutation(fieldPermutation);
+    }
+
+    @Override
+    public void open() throws HyracksDataException {
+        AbstractInvertedIndexOperatorDescriptor opDesc = (AbstractInvertedIndexOperatorDescriptor) btreeOpHelper
+                .getOperatorDescriptor();
+        RecordDescriptor recDesc = recordDescProvider.getInputRecordDescriptor(opDesc.getOperatorId(), 0);
+        accessor = new FrameTupleAccessor(btreeOpHelper.getHyracksStageletContext().getFrameSize(), recDesc);
+        
+        // btree
+        try {
+            btreeOpHelper.init();
+            btreeOpHelper.getBTree().open(btreeOpHelper.getBTreeFileId());            
+        } catch (Exception e) {
+            // cleanup in case of failure
+            btreeOpHelper.deinit();
+            throw new HyracksDataException(e);
+        }
+        
+        // inverted index
+        try {
+            invIndexOpHelper.init();
+            invIndexOpHelper.getInvIndex().open(invIndexOpHelper.getInvIndexFileId()); 
+            bulkLoadCtx = invIndexOpHelper.getInvIndex().beginBulkLoad(invListBuilder, ctx.getFrameSize(), btreeFillFactor);
+        } catch (Exception e) {
+            // cleanup in case of failure
+            invIndexOpHelper.deinit();
+            throw new HyracksDataException(e);
+        }
+    }
+
+    @Override
+    public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
+        accessor.reset(buffer);
+        int tupleCount = accessor.getTupleCount();
+        for (int i = 0; i < tupleCount; i++) {
+            tuple.reset(accessor, i);
+            invIndexOpHelper.getInvIndex().bulkLoadAddTuple(bulkLoadCtx, tuple);
+        }
+    }
+
+    @Override
+    public void close() throws HyracksDataException {                
+        try {
+            invIndexOpHelper.getInvIndex().endBulkLoad(bulkLoadCtx);
+        } finally {
+            btreeOpHelper.deinit();
+        }
+    }
+
+    @Override
+    public void flush() throws HyracksDataException {
+    }
+}
\ No newline at end of file
diff --git a/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexOpHelper.java b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexOpHelper.java
new file mode 100644
index 0000000..025462d
--- /dev/null
+++ b/hyracks-storage-am-invertedindex/src/main/java/edu/uci/ics/hyracks/storage/am/invertedindex/dataflow/InvertedIndexOpHelper.java
@@ -0,0 +1,157 @@
+/*
+ * Copyright 2009-2010 by The Regents of the University of California
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * you may obtain a copy of the License from
+ * 
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package edu.uci.ics.hyracks.storage.am.invertedindex.dataflow;
+
+import edu.uci.ics.hyracks.api.context.IHyracksStageletContext;
+import edu.uci.ics.hyracks.api.dataflow.value.IBinaryComparator;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
+import edu.uci.ics.hyracks.api.io.FileReference;
+import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
+import edu.uci.ics.hyracks.storage.am.btree.dataflow.IBTreeOperatorDescriptorHelper;
+import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexHelperOpenMode;
+import edu.uci.ics.hyracks.storage.am.common.dataflow.IndexRegistry;
+import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
+import edu.uci.ics.hyracks.storage.am.invertedindex.impls.InvertedIndex;
+import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
+import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
+
+public final class InvertedIndexOpHelper {
+    
+	private InvertedIndex invIndex;
+	private int invIndexFileId = -1;
+	private int partition;
+
+    private IInvertedIndexOperatorDescriptorHelper opDesc;
+    private IHyracksStageletContext ctx;
+
+	private IndexHelperOpenMode mode;
+
+    public InvertedIndexOpHelper(IInvertedIndexOperatorDescriptorHelper opDesc, final IHyracksStageletContext ctx, int partition,
+            IndexHelperOpenMode mode) {
+        this.opDesc = opDesc;
+        this.ctx = ctx;
+        this.mode = mode;
+        this.partition = partition;
+    }
+
+	public void init() throws HyracksDataException {
+        IBufferCache bufferCache = opDesc.getStorageManager().getBufferCache(ctx);
+        IFileMapProvider fileMapProvider = opDesc.getStorageManager().getFileMapProvider(ctx);
+        IFileSplitProvider fileSplitProvider = opDesc.getInvIndexFileSplitProvider();
+
+        FileReference f = fileSplitProvider.getFileSplits()[partition].getLocalFile();
+        boolean fileIsMapped = fileMapProvider.isMapped(f);
+
+		switch (mode) {
+		
+		case OPEN: {
+			if (!fileIsMapped) {
+				throw new HyracksDataException(
+						"Trying to open inverted index from unmapped file " + f.toString());
+			}
+		}
+		break;
+
+		case CREATE:
+		case ENLIST: {
+			if (!fileIsMapped) {
+				bufferCache.createFile(f);
+			}
+		}
+		break;
+		
+		}
+		
+        int fileId = fileMapProvider.lookupFileId(f);
+        try {
+        	bufferCache.openFile(fileId);
+        } catch(HyracksDataException e) {
+        	// revert state of buffer cache since file failed to open
+        	if(!fileIsMapped) {
+        		bufferCache.deleteFile(fileId);
+        	}
+        	throw e;
+        }
+        
+        // only set btreeFileId member when openFile() succeeds, 
+        // otherwise deinit() will try to close the file that failed to open
+        invIndexFileId = fileId;        
+        IndexRegistry<InvertedIndex> invIndexRegistry = opDesc.getInvIndexRegistryProvider().getRegistry(ctx);
+        invIndex = invIndexRegistry.get(invIndexFileId);
+        if (invIndex == null) {
+
+			// create new inverted index and register it
+            invIndexRegistry.lock();
+			try {
+				// check if inverted index has already been registered by another thread
+			    invIndex = invIndexRegistry.get(invIndexFileId);
+				if (invIndex == null) {
+					// this thread should create and register the inverted index
+
+					IBinaryComparator[] comparators = new IBinaryComparator[opDesc
+							.getInvIndexComparatorFactories().length];
+					for (int i = 0; i < opDesc.getInvIndexComparatorFactories().length; i++) {
+						comparators[i] = opDesc.getInvIndexComparatorFactories()[i]
+								.createBinaryComparator();
+					}
+
+					MultiComparator cmp = new MultiComparator(opDesc
+							.getInvIndexTypeTraits(), comparators);
+					
+					// assumes btree has already been registered
+					IFileSplitProvider btreeFileSplitProvider = opDesc.getBTreeFileSplitProvider();
+					IndexRegistry<BTree> btreeRegistry = opDesc.getBTreeRegistryProvider().getRegistry(ctx);
+					FileReference btreeFile = btreeFileSplitProvider.getFileSplits()[partition].getLocalFile();
+					boolean btreeFileIsMapped = fileMapProvider.isMapped(btreeFile);
+					if(!btreeFileIsMapped) {
+					    throw new HyracksDataException("Trying to create inverted index, but associated BTree file has not been mapped");
+					}
+					int btreeFileId = fileMapProvider.lookupFileId(f);					
+					BTree btree = btreeRegistry.get(btreeFileId);					
+					
+					invIndex = new InvertedIndex(bufferCache, btree, cmp);					
+					invIndex.open(invIndexFileId);
+					invIndexRegistry.register(invIndexFileId, invIndex);
+				}
+			} finally {
+			    invIndexRegistry.unlock();
+			}
+		}
+	}
+
+    public void deinit() throws HyracksDataException {
+        if (invIndexFileId != -1) {
+            IBufferCache bufferCache = opDesc.getStorageManager().getBufferCache(ctx);
+            bufferCache.closeFile(invIndexFileId);
+        }
+    }
+
+	public InvertedIndex getInvIndex() {
+		return invIndex;
+	}
+	
+    public IHyracksStageletContext getHyracksStageletContext() {
+        return ctx;
+    }
+
+	public IBTreeOperatorDescriptorHelper getOperatorDescriptor() {
+		return opDesc;
+	}
+	
+	public int getInvIndexFileId() {
+		return invIndexFileId;
+	}
+}
\ No newline at end of file