diff --git a/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/util/TupleUtils.java b/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/util/TupleUtils.java
index 14c4d66..df3ee67 100644
--- a/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/util/TupleUtils.java
+++ b/hyracks-dataflow-common/src/main/java/edu/uci/ics/hyracks/dataflow/common/util/TupleUtils.java
@@ -83,4 +83,14 @@
         }        
         return strBuilder.toString();
     }
+    
+    public static ITupleReference copyTuple(ITupleReference tuple) throws HyracksDataException {
+        ArrayTupleBuilder tupleBuilder = new ArrayTupleBuilder(tuple.getFieldCount());
+        for (int i = 0; i < tuple.getFieldCount(); i++) {
+            tupleBuilder.addField(tuple.getFieldData(i), tuple.getFieldStart(i), tuple.getFieldLength(i));
+        }
+        ArrayTupleReference tupleCopy = new ArrayTupleReference();
+        tupleCopy.reset(tupleBuilder.getFieldEndOffsets(), tupleBuilder.getByteArray());
+        return tupleCopy;
+    }
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
index f2a55f7..b809a36 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTree.java
@@ -1177,13 +1177,12 @@
             ctx.reset(IndexOp.DISKORDERSCAN);
             btree.diskOrderScan(cursor, ctx);
         }
-        
+		
 		// TODO: Ideally, this method should not exist. But we need it for
-		// the LSM tree to work correctly, so we can use the LSMOpContext inside
-		// a BTreeAccessor.
-		// Making the appropriate change will involve changing lots of code.
-		public void setOpContext(BTreeOpContext ctx) {
-			this.ctx = ctx;
+		// the changing the leafFrame and leafFrameFactory of the op context for
+		// the LSM-BTree to work correctly.
+		public BTreeOpContext getOpContext() {
+			return ctx;
 		}
     }
 }
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
index 07c645c..8f056cc 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/impls/BTreeOpContext.java
@@ -28,8 +28,8 @@
 
 public class BTreeOpContext implements IIndexOpContext {
     private final int INIT_ARRAYLIST_SIZE = 6; 
-    protected ITreeIndexFrameFactory leafFrameFactory;
-    protected ITreeIndexFrameFactory interiorFrameFactory;
+    public ITreeIndexFrameFactory leafFrameFactory;
+    public ITreeIndexFrameFactory interiorFrameFactory;
     public IBTreeLeafFrame leafFrame;
     public IBTreeInteriorFrame interiorFrame;
     public ITreeIndexMetaDataFrame metaFrame;
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexDeleteTest.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexDeleteTest.java
index 15d087a..a472934 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexDeleteTest.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexDeleteTest.java
@@ -44,11 +44,9 @@
             } else if (fieldSerdes[0] instanceof UTF8StringSerializerDeserializer) {
                 OrderedIndexTestUtils.insertStringTuples(ctx, numTuplesToInsert, getRandom());
             }
-
             int numTuplesPerDeleteRound = (int) Math.ceil((float) ctx.getCheckTuples().size() / (float) numDeleteRounds);
             for (int j = 0; j < numDeleteRounds; j++) {
                 OrderedIndexTestUtils.deleteTuples(ctx, numTuplesPerDeleteRound, getRandom());
-
                 OrderedIndexTestUtils.checkPointSearches(ctx);
                 OrderedIndexTestUtils.checkOrderedScan(ctx);
                 OrderedIndexTestUtils.checkDiskOrderScan(ctx);
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestDriver.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestDriver.java
index f76380d..e89c38f 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestDriver.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestDriver.java
@@ -32,8 +32,7 @@
 public abstract class OrderedIndexTestDriver {
     protected final Logger LOGGER = Logger.getLogger(OrderedIndexTestDriver.class.getName());
     
-    //protected static final int numTuplesToInsert = 10000;
-    protected static final int numTuplesToInsert = 2;
+    protected static final int numTuplesToInsert = 10000;
     
     protected abstract IOrderedIndexTestContext createTestContext(ISerializerDeserializer[] fieldSerdes, int numKeys, BTreeLeafFrameType leafType) throws Exception;
     protected abstract Random getRandom();
@@ -62,7 +61,7 @@
         }
     }
     
-    //@Test
+    @Test
     public void twoIntKeys() throws Exception {    
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("BTree " + getTestOpName() + " Test With Two Int Keys.");
@@ -83,7 +82,7 @@
         }
     }
     
-    //@Test
+    @Test
     public void twoIntKeysAndValues() throws Exception {  
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("BTree " + getTestOpName() + " Test With Two Int Keys And Values.");
@@ -104,7 +103,7 @@
         }
     }        
     
-    //@Test
+    @Test
     public void oneStringKeyAndValue() throws Exception {        
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("BTree " + getTestOpName() + " Test With One String Key And Value.");
@@ -121,7 +120,7 @@
         }
     }
     
-    //@Test
+    @Test
     public void twoStringKeys() throws Exception {
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("BTree " + getTestOpName() + " Test With Two String Keys.");
@@ -142,7 +141,7 @@
         }
     }
     
-    //@Test
+    @Test
     public void twoStringKeysAndValues() throws Exception {      
         if (LOGGER.isLoggable(Level.INFO)) {
             LOGGER.info("BTree " + getTestOpName() + " Test With Two String Keys And Values.");
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestUtils.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestUtils.java
index da481fd..c6edd1f 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestUtils.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexTestUtils.java
@@ -41,7 +41,7 @@
             DataInput dataIn = new DataInputStream(inStream);
             Object actualObj = fieldSerdes[i].deserialize(dataIn);            
             if (!actualObj.equals(expected.get(i))) {
-                fail("Actual and expected fields do not match.\nExpected: " + expected.get(i) + "\nActual  : " + actualObj);
+                fail("Actual and expected fields do not match on field " + i + ".\nExpected: " + expected.get(i) + "\nActual  : " + actualObj);
             }
         }
     }
@@ -113,20 +113,13 @@
         int actualCount = 0;
         try {
             while (scanCursor.hasNext()) {
-                // START DEBUG
+                if (!checkIter.hasNext()) {
+                    fail("Ordered scan returned more answers than expected.\nExpected: " + ctx.getCheckTuples().size());
+                }
                 scanCursor.next();
+                CheckTuple expectedTuple = checkIter.next();
                 ITupleReference tuple = scanCursor.getTuple();
-                System.out.println("SCANNED: " + TupleUtils.printTuple(tuple, ctx.getFieldSerdes()));
-                // END DEBUG
-                //if (!checkIter.hasNext()) {
-                //    fail("Ordered scan returned more answers than expected.\nExpected: " + ctx.getCheckTuples().size());
-                //}
-                //scanCursor.next();
-                //CheckTuple expectedTuple = checkIter.next();
-                //ITupleReference tuple = scanCursor.getTuple();
-                //System.out.println("SCANNED: " + TupleUtils.printTuple(tuple, ctx.getFieldSerdes()));
-                //System.out.println("CHECKTUPLES: " + ctx.getCheckTuples().size());
-                //compareActualAndExpected(tuple, expectedTuple, ctx.getFieldSerdes());
+                compareActualAndExpected(tuple, expectedTuple, ctx.getFieldSerdes());
                 actualCount++;
             }
             if (actualCount < ctx.getCheckTuples().size()) {
@@ -283,14 +276,13 @@
                 }
             }
             try {
-                ctx.getIndexAccessor().insert(ctx.getTuple());
+            	//System.out.println("INSERTING: " + TupleUtils.printTuple(ctx.getTuple(), ctx.getFieldSerdes()));
+            	ctx.getIndexAccessor().insert(ctx.getTuple());
                 // Set expected values. Do this only after insertion succeeds because we ignore duplicate keys.
                 ctx.insertIntCheckTuple(fieldValues);
             } catch (BTreeDuplicateKeyException e) {
                 // Ignore duplicate key insertions.
-            }
-            
-            System.out.println("INSERTED: " + TupleUtils.printTuple(ctx.getTuple(), ctx.getFieldSerdes()));
+            }                        
         }
     }
 	
@@ -403,10 +395,8 @@
             }
             int checkTupleIdx = Math.abs(rnd.nextInt() % numCheckTuples);
             CheckTuple checkTuple = checkTuples[checkTupleIdx];            
-            createTupleFromCheckTuple(checkTuple, deleteTupleBuilder, deleteTuple, ctx.getFieldSerdes());        
-            
-            System.out.println("DELETED: " + TupleUtils.printTuple(ctx.getTuple(), ctx.getFieldSerdes()));
-            
+            createTupleFromCheckTuple(checkTuple, deleteTupleBuilder, deleteTuple, ctx.getFieldSerdes());
+            //System.out.println("DELETING:  " + TupleUtils.printTuple(deleteTuple, ctx.getFieldSerdes()));
             ctx.getIndexAccessor().delete(deleteTuple);
             
             // Remove check tuple from expected results.
@@ -416,7 +406,7 @@
             CheckTuple tmp = checkTuples[numCheckTuples - 1];
             checkTuples[numCheckTuples - 1] = checkTuple;
             checkTuples[checkTupleIdx] = tmp;
-            numCheckTuples--;
+            numCheckTuples--;                        
         }
     }
     
diff --git a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexUpdateTest.java b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexUpdateTest.java
index ddc1f34..6022058 100644
--- a/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexUpdateTest.java
+++ b/hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/tests/OrderedIndexUpdateTest.java
@@ -38,9 +38,7 @@
         if (fieldSerdes.length == numKeys) {
             return;
         }
-
         IOrderedIndexTestContext ctx = createTestContext(fieldSerdes, numKeys, leafType);
-
         // We assume all fieldSerdes are of the same type. Check the first one
         // to determine which field types to generate.
         if (fieldSerdes[0] instanceof IntegerSerializerDeserializer) {
@@ -48,11 +46,9 @@
         } else if (fieldSerdes[0] instanceof UTF8StringSerializerDeserializer) {
             OrderedIndexTestUtils.insertStringTuples(ctx, numTuplesToInsert, getRandom());
         }
-
         int numTuplesPerDeleteRound = (int) Math.ceil((float) ctx.getCheckTuples().size() / (float) numUpdateRounds);
         for (int j = 0; j < numUpdateRounds; j++) {
             OrderedIndexTestUtils.updateTuples(ctx, numTuplesPerDeleteRound, getRandom());
-
             OrderedIndexTestUtils.checkPointSearches(ctx);
             OrderedIndexTestUtils.checkOrderedScan(ctx);
             OrderedIndexTestUtils.checkDiskOrderScan(ctx);
diff --git a/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/tuples/TypeAwareTupleWriter.java b/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/tuples/TypeAwareTupleWriter.java
index fe52608..cf053cf 100644
--- a/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/tuples/TypeAwareTupleWriter.java
+++ b/hyracks-storage-am-common/src/main/java/edu/uci/ics/hyracks/storage/am/common/tuples/TypeAwareTupleWriter.java
@@ -79,8 +79,6 @@
 
         // write data fields
         for (int i = 0; i < tuple.getFieldCount(); i++) {
-            int s = tuple.getFieldStart(i);
-            int l = tuple.getFieldLength(i);
             System.arraycopy(tuple.getFieldData(i), tuple.getFieldStart(i), targetBuf, runner, tuple.getFieldLength(i));
             runner += tuple.getFieldLength(i);
         }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTree.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTree.java
index 9b40986..1ab19c6 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTree.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTree.java
@@ -22,18 +22,19 @@
 import java.util.LinkedList;
 import java.util.ListIterator;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.api.io.FileReference;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
+import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.util.TupleUtils;
 import edu.uci.ics.hyracks.storage.am.btree.exceptions.BTreeDuplicateKeyException;
+import edu.uci.ics.hyracks.storage.am.btree.exceptions.BTreeNonExistentKeyException;
 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.IFreePageManager;
 import edu.uci.ics.hyracks.storage.am.common.api.IIndexBulkLoadContext;
 import edu.uci.ics.hyracks.storage.am.common.api.ISearchPredicate;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndex;
 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.ITreeIndexFrameFactory;
@@ -44,10 +45,11 @@
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMFileNameManager;
 import edu.uci.ics.hyracks.storage.am.lsm.common.api.ILSMTree;
 import edu.uci.ics.hyracks.storage.am.lsm.common.freepage.InMemoryFreePageManager;
+import edu.uci.ics.hyracks.storage.am.lsm.tuples.LSMTypeAwareTupleReference;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
 
-public class LSMTree implements ITreeIndex, ILSMTree {
+public class LSMTree implements ILSMTree {
     // In-memory components.
     private final BTree memBTree;
     private final InMemoryFreePageManager memFreePageManager;    
@@ -56,7 +58,7 @@
     private final ILSMFileNameManager fileNameManager;
     private final BTreeFactory diskBTreeFactory;
     private final IBufferCache diskBufferCache;
-    private final IFileMapProvider diskFileMapProvider;    
+    private final IFileMapProvider diskFileMapProvider;
     private LinkedList<BTree> onDiskBTrees = new LinkedList<BTree>();
     private LinkedList<BTree> mergedBTrees = new LinkedList<BTree>();
     private int onDiskBTreeCount;
@@ -159,22 +161,127 @@
 				}
 			}
 		} while (waitForFlush);
+		// TODO: This will become much simpler once the BTree supports a true upsert operation.
 		try {
 			ctx.memBTreeAccessor.insert(tuple);
 		} catch (BTreeDuplicateKeyException e) {
-			// We don't need to deal with a nonexistent key here, because a
-			// deleter will actually update the key and it's value, and not
-			// delete it from the BTree.
-			// Also notice that a flush must wait for the current operation to
+			// Notice that a flush must wait for the current operation to
 			// finish (threadRefCount must reach zero).
-            if (cmp.getKeyFieldCount() != memBTree.getFieldCount()) {
-                ctx.reset(IndexOp.UPDATE);
-                ctx.memBTreeAccessor.update(tuple);
-            }
+            // TODO: This methods below are very inefficient, we'd rather like
+            // to flip the antimatter bit one single BTree traversal.
+		    if (ctx.getIndexOp() == IndexOp.DELETE) {
+		        deleteExistingKey(tuple, ctx);
+		    } else {
+		        insertOrUpdateExistingKey(tuple, ctx);
+		    }
 		}
 		threadExit();
+		
+		// DEBUG check the in-mem tree
+		/*
+		RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
+        ITreeIndexAccessor accessor = memBTree.createAccessor();
+        ITreeIndexCursor cursor = accessor.createSearchCursor();
+        accessor.search(cursor, nullPred);
+        int count = 0;
+        try {
+            while (cursor.hasNext()) {
+                cursor.next();
+                count++;
+                String s = printTuple(cursor.getTuple());
+                System.out.println(count + " INMEM CHECK: " + s);
+            }
+        } finally {
+            cursor.close();
+        }
+        System.out.println("");
+        */
 	}
 
+	private void deleteExistingKey(ITupleReference tuple, LSMTreeOpContext ctx) throws HyracksDataException, TreeIndexException {
+        // We assume that tuple given by the user for deletion only contains the
+        // key fields, but not any non-key fields.
+        // Therefore, to set the delete bit in the tuple that already exist in
+        // the BTree, we must retrieve the original tuple first. This is to
+        // ensure that we have the proper value field.
+        if (cmp.getKeyFieldCount() != memBTree.getFieldCount()) {
+            ctx.reset(IndexOp.SEARCH);
+            RangePredicate rangePredicate = new RangePredicate(true, tuple, tuple, true, true, cmp, cmp);
+            ITreeIndexCursor cursor = ctx.memBTreeAccessor.createSearchCursor();
+            ctx.memBTreeAccessor.search(cursor, rangePredicate);
+            ITupleReference tupleCopy = null;
+            try {
+                if (cursor.hasNext()) {
+                    cursor.next();
+                    tupleCopy = TupleUtils.copyTuple(cursor.getTuple());
+                }
+            } finally {
+                cursor.close();
+            }
+            // This means the tuple we are looking for must have been truly deleted by another thread.
+            // Simply restart the original operation to insert the antimatter tuple. 
+            // There is a remote chance of livelocks due to this behavior.
+            if (tupleCopy == null) {
+                ctx.reset(IndexOp.DELETE);
+                lsmPerformOp(tuple, ctx);
+                return;
+            }
+            memBTreeUpdate(tupleCopy, ctx);
+        } else {
+            // Since the existing tuple could be a matter tuple, we must delete it and re-insert.
+            memBTreeDeleteAndReinsert(tuple, ctx);
+        }            
+	}
+	
+    private void insertOrUpdateExistingKey(ITupleReference tuple, LSMTreeOpContext ctx) throws HyracksDataException,
+            TreeIndexException {        
+        // If all fields are keys, and the key we are trying to insert/update
+        // already exists, then we are already done.
+        // Otherwise, we must update the non-key fields.        
+        if (cmp.getKeyFieldCount() != memBTree.getFieldCount()) {
+            memBTreeUpdate(tuple, ctx);
+        } else {
+            // Since the existing tuple could be an antimatter tuple, we must delete it and re-insert.
+            memBTreeDeleteAndReinsert(tuple, ctx);
+        }
+    }
+    
+    private void memBTreeDeleteAndReinsert(ITupleReference tuple, LSMTreeOpContext ctx) throws HyracksDataException,
+            TreeIndexException {
+        // All fields are key fields, therefore a true BTree update is not
+        // allowed.
+        // In order to set/unset the delete bit, we
+        // must truly delete the existing tuple from the BTree, and then
+        // re-insert it (with the delete bit set/unset).
+        // Since the tuple given by the user already has all fields, we
+        // don't need to retrieve the already existing tuple.
+        IndexOp originalOp = ctx.getIndexOp();
+        try {
+            ctx.memBTreeAccessor.delete(tuple);
+        } catch (BTreeNonExistentKeyException e) {
+            // Somebody else has truly deleted the tuple, but we will restart
+            // our operation anyway.
+        }
+        // Restart performOp to insert the tuple.
+        ctx.reset(originalOp);
+        lsmPerformOp(tuple, ctx);
+    }
+    
+    private void memBTreeUpdate(ITupleReference tuple, LSMTreeOpContext ctx) throws HyracksDataException,
+            TreeIndexException {
+        IndexOp originalOp = ctx.getIndexOp();
+        try {
+            ctx.reset(IndexOp.UPDATE);
+            ctx.memBTreeAccessor.update(tuple);
+        } catch (BTreeNonExistentKeyException e) {
+            // It is possible that the key has truly been deleted.
+            // Simply restart the operation. There is a remote chance of
+            // livelocks due to this behavior.
+            ctx.reset(originalOp);
+            lsmPerformOp(tuple, ctx);
+        }
+    }
+    
     public void threadEnter() {
         threadRefCount++;
     }
@@ -197,11 +304,10 @@
     @Override
     public void flush() throws HyracksDataException, TreeIndexException {
         System.out.println("FLUSHING!");
-        // Bulk load a new on-disk BTree from the in-memory BTree.
-        ITreeIndexCursor scanCursor = new BTreeRangeSearchCursor(
-                (IBTreeLeafFrame) insertLeafFrameFactory.createFrame(), false);
+        // Bulk load a new on-disk BTree from the in-memory BTree.        
         RangePredicate nullPred = new RangePredicate(true, null, null, true, true, null, null);
         ITreeIndexAccessor memBTreeAccessor = memBTree.createAccessor();
+        ITreeIndexCursor scanCursor = memBTreeAccessor.createSearchCursor();
         memBTreeAccessor.search(scanCursor, nullPred);
         BTree diskBTree = createFlushTargetBTree();
         // Bulk load the tuples from the in-memory BTree into the new disk BTree.
@@ -209,8 +315,7 @@
         try {
             while (scanCursor.hasNext()) {
                 scanCursor.next();
-                ITupleReference frameTuple = scanCursor.getTuple();
-                diskBTree.bulkLoadAddTuple(frameTuple, bulkLoadCtx);
+                diskBTree.bulkLoadAddTuple(scanCursor.getTuple(), bulkLoadCtx);
             }
         } finally {
             scanCursor.close();
@@ -218,8 +323,78 @@
         diskBTree.endBulkLoad(bulkLoadCtx);
         resetMemBTree();
         onDiskBTrees.addFirst(diskBTree);
+        
+        // DEBUG check the on-disk tree
+        /*
+        ITreeIndexAccessor accessor = diskBTree.createAccessor();
+        ITreeIndexCursor cursor = accessor.createSearchCursor();
+        accessor.search(cursor, nullPred);
+        try {
+            while (cursor.hasNext()) {
+                cursor.next();
+                String s = printTuple(cursor.getTuple());
+                System.out.println("FLUSH CHECK: " + s);
+            }
+        } finally {
+            cursor.close();
+        }
+        */
     }
 
+    // DEBUG
+    /*
+    private String printTuple(ITupleReference tuple) {
+        LSMTypeAwareTupleReference lsmTuple = (LSMTypeAwareTupleReference)tuple;
+        ISerializerDeserializer[] fieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+        ISerializerDeserializer[] keyOnlyFieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE };
+        String s = null;
+        try {
+            if (lsmTuple.isDelete()) {
+                s = TupleUtils.printTuple(lsmTuple, keyOnlyFieldSerdes) + " " + "D";
+            } else {
+                s = TupleUtils.printTuple(lsmTuple, fieldSerdes);
+            }
+        } catch (HyracksDataException e) {
+            e.printStackTrace();
+        }
+        s += " " + lsmTuple.isDelete();
+        return s;
+    }
+    */
+    
+    /*
+    private String printTuple(ITupleReference tuple) {
+        LSMTypeAwareTupleReference lsmTuple = (LSMTypeAwareTupleReference)tuple;
+        ISerializerDeserializer[] fieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+        String s = null;
+        try {
+            s = TupleUtils.printTuple(lsmTuple, fieldSerdes);
+        } catch (HyracksDataException e) {
+            e.printStackTrace();
+        }
+        s += " " + lsmTuple.isDelete();
+        return s;
+    }
+    */
+    
+    private String printTuple(ITupleReference tuple) {
+        LSMTypeAwareTupleReference lsmTuple = (LSMTypeAwareTupleReference)tuple;
+        ISerializerDeserializer[] fieldSerdes = new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE, UTF8StringSerializerDeserializer.INSTANCE };
+        ISerializerDeserializer[] keyOnlyFieldSerdes = new ISerializerDeserializer[] { UTF8StringSerializerDeserializer.INSTANCE };
+        String s = null;
+        try {
+            if (lsmTuple.isDelete()) {
+                s = TupleUtils.printTuple(lsmTuple, keyOnlyFieldSerdes) + " " + "D";
+            } else {
+                s = TupleUtils.printTuple(lsmTuple, fieldSerdes);
+            }
+        } catch (HyracksDataException e) {
+            e.printStackTrace();
+        }
+        s += " " + lsmTuple.isDelete();
+        return s;
+    }
+    
     private void resetMemBTree() throws HyracksDataException {
         memFreePageManager.reset();
         memBTree.create(memBTree.getFileId());
@@ -453,9 +628,7 @@
     }
     
     public LSMTreeOpContext createOpContext() {
-        return new LSMTreeOpContext((BTree.BTreeAccessor) memBTree.createAccessor(), insertLeafFrameFactory,
-                deleteLeafFrameFactory, interiorFrameFactory, memFreePageManager.getMetaDataFrameFactory()
-                        .createFrame(), cmp);
+        return new LSMTreeOpContext(memBTree, insertLeafFrameFactory, deleteLeafFrameFactory);
     }
 
     @Override
@@ -488,7 +661,7 @@
         @Override
         public void delete(ITupleReference tuple) throws HyracksDataException, TreeIndexException {
             ctx.reset(IndexOp.DELETE);
-            lsmTree.delete(tuple, ctx);
+            lsmTree.delete(tuple, ctx);            
         }
 
         @Override
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeOpContext.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeOpContext.java
index ab7cfed..a74e36d 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeOpContext.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeOpContext.java
@@ -1,63 +1,101 @@
+/*
+ * 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.lsm.impls;
 
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTree;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeOpContext;
+import edu.uci.ics.hyracks.storage.am.common.api.IIndexOpContext;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrameFactory;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexMetaDataFrame;
 import edu.uci.ics.hyracks.storage.am.common.ophelpers.IndexOp;
-import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
 
-public final class LSMTreeOpContext extends BTreeOpContext {
-
+public final class LSMTreeOpContext implements IIndexOpContext {
+    
 	public ITreeIndexFrameFactory insertLeafFrameFactory;
 	public ITreeIndexFrameFactory deleteLeafFrameFactory;
 	public IBTreeLeafFrame insertLeafFrame;
 	public IBTreeLeafFrame deleteLeafFrame;
-	public final BTree.BTreeAccessor memBTreeAccessor;
-
-	public LSMTreeOpContext(BTree.BTreeAccessor memBTreeAccessor, ITreeIndexFrameFactory insertLeafFrameFactory,
-			ITreeIndexFrameFactory deleteLeafFrameFactory, ITreeIndexFrameFactory interiorFrameFactory,
-			ITreeIndexMetaDataFrame metaFrame, MultiComparator cmp) {
-		super(insertLeafFrameFactory, interiorFrameFactory, metaFrame, cmp);		
-		
-		this.memBTreeAccessor = memBTreeAccessor;
-        // Overwrite the BTree accessor's op context with our LSMTreeOpContext.
-		this.memBTreeAccessor.setOpContext(this);
-		
+	public final BTree memBTree;
+	public BTree.BTreeAccessor memBTreeAccessor;
+	public BTreeOpContext memBTreeOpCtx;
+	public IndexOp op;
+	
+    public LSMTreeOpContext(BTree memBTree, ITreeIndexFrameFactory insertLeafFrameFactory,
+            ITreeIndexFrameFactory deleteLeafFrameFactory) {
+		this.memBTree = memBTree;
 		this.insertLeafFrameFactory = insertLeafFrameFactory;
 		this.deleteLeafFrameFactory = deleteLeafFrameFactory;
 		this.insertLeafFrame = (IBTreeLeafFrame) insertLeafFrameFactory.createFrame();
 		this.deleteLeafFrame = (IBTreeLeafFrame) deleteLeafFrameFactory.createFrame();
 		if (insertLeafFrame != null) {
-			insertLeafFrame.setMultiComparator(cmp);
+			insertLeafFrame.setMultiComparator(memBTree.getMultiComparator());
         }
         if (deleteLeafFrame != null) {
-        	deleteLeafFrame.setMultiComparator(cmp);
+        	deleteLeafFrame.setMultiComparator(memBTree.getMultiComparator());
         }
-        reset(op);
 	}
 
     @Override
     public void reset(IndexOp newOp) {
-    	super.reset(newOp);
-    	if(newOp == IndexOp.INSERT) {
-    		setInsertMode();
-    	}
-    	if(newOp == IndexOp.DELETE) {
-    		super.reset(IndexOp.INSERT);
-    		setDeleteMode();
+    	this.op = newOp;
+        switch (newOp) {
+    	    case SEARCH:
+    	    case DISKORDERSCAN:
+    	    case UPDATE:
+                // Attention: It is important to leave the leafFrame and
+                // leafFrameFactory of the memBTree as is when doing an update.
+                // Update will only be set if a previous attempt to delete or
+                // insert failed, so we must preserve the semantics of the
+                // previously requested operation.
+    	        return;
+    	        
+    	    case INSERT:    	    
+    	        setInsertMode();
+    	        break;
+    	        
+    	    case DELETE:
+                setDeleteMode();
+                break;
     	}
     }
 	
+    private void setMemBTreeAccessor() {
+        if (memBTreeAccessor == null) {
+            memBTreeAccessor = (BTree.BTreeAccessor) memBTree.createAccessor();
+            memBTreeOpCtx = memBTreeAccessor.getOpContext();
+        }
+    }
+    
 	public void setInsertMode() {
-		this.leafFrame = insertLeafFrame;
-		leafFrameFactory = insertLeafFrameFactory;
+	    setMemBTreeAccessor();
+	    memBTreeOpCtx.leafFrame = insertLeafFrame;
+	    memBTreeOpCtx.leafFrameFactory = insertLeafFrameFactory;
 	}
 	
 	public void setDeleteMode() {
-		this.leafFrame = deleteLeafFrame;
-		leafFrameFactory = deleteLeafFrameFactory;
+	    setMemBTreeAccessor();
+	    memBTreeOpCtx.leafFrame = deleteLeafFrame;
+        memBTreeOpCtx.leafFrameFactory = deleteLeafFrameFactory;
 	}
-	
+
+    @Override
+    public void reset() {
+    }
+    
+    public IndexOp getIndexOp() {
+        return op;
+    }
 }
\ No newline at end of file
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeRangeSearchCursor.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeRangeSearchCursor.java
index d976ef1..051242e 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeRangeSearchCursor.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/impls/LSMTreeRangeSearchCursor.java
@@ -2,8 +2,11 @@
 
 import java.util.PriorityQueue;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
+import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.util.TupleUtils;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeLeafFrame;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeRangeSearchCursor;
 import edu.uci.ics.hyracks.storage.am.common.api.ICursorInitialState;
@@ -39,6 +42,7 @@
                 rangeCursors[i].next();
                 element = new LSMPriorityQueueElement(rangeCursors[i].getTuple(), i);
                 outputPriorityQueue.offer(element);
+                //System.out.println("INITIALIZING PQ WITH: " + printTuple(rangeCursors[i].getTuple(), i));
             }
         }
         checkPriorityQueue();
@@ -50,7 +54,8 @@
 
     @Override
     public void reset() {
-        // do nothing
+        outputElement = null;
+        needPush = false;
     }
 
     @Override
@@ -100,7 +105,25 @@
             throw new HyracksDataException(e);
         }
     }
-
+   
+    private String printTuple(ITupleReference tuple, int cursorIndex) {
+        LSMTypeAwareTupleReference lsmTuple = (LSMTypeAwareTupleReference)tuple;
+        ISerializerDeserializer[] fieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+        ISerializerDeserializer[] keyOnlyFieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE };
+        String s = null;
+        try {
+            if (lsmTuple.isDelete()) {
+                s = TupleUtils.printTuple(lsmTuple, keyOnlyFieldSerdes) + " " + "D";
+            } else {
+                s = TupleUtils.printTuple(lsmTuple, fieldSerdes);
+            }
+        } catch (HyracksDataException e) {
+            e.printStackTrace();
+        }
+        s += " " + lsmTuple.isDelete() + " " + cursorIndex;
+        return s;
+    }
+    
     @Override
     public void setBufferCache(IBufferCache bufferCache) {
         // do nothing
@@ -113,6 +136,7 @@
 
     @Override
     public ITupleReference getTuple() {
+        //System.out.println("RETURNING: " + printTuple(outputElement.getTuple(), outputElement.getCursorIndex()));
         return (ITupleReference) outputElement.getTuple();
     }
 
@@ -121,11 +145,11 @@
             rangeCursors[cursorIndex].next();
             reusedElement.reset(rangeCursors[cursorIndex].getTuple(), cursorIndex);
             outputPriorityQueue.offer(reusedElement);
+            //System.out.println("PUSHING TO PQ: " + printTuple(reusedElement.getTuple(), cursorIndex));
         }
     }
 
     private void checkPriorityQueue() throws HyracksDataException {
-
         while (!outputPriorityQueue.isEmpty() || needPush == true) {
             if (!outputPriorityQueue.isEmpty()) {
                 LSMPriorityQueueElement checkElement = outputPriorityQueue.peek();
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriter.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriter.java
index f673a2f..97b0a32 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriter.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriter.java
@@ -1,18 +1,17 @@
 package edu.uci.ics.hyracks.storage.am.lsm.tuples;
 
+import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
+import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
-import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleWriter;
+import edu.uci.ics.hyracks.dataflow.common.data.marshalling.IntegerSerializerDeserializer;
+import edu.uci.ics.hyracks.dataflow.common.util.TupleUtils;
 
-public class LSMEntireTupleWriter extends TypeAwareTupleWriter {
-	public LSMEntireTupleWriter(ITypeTraits[] typeTraits){
-		super(typeTraits);
+public class LSMEntireTupleWriter extends LSMTypeAwareTupleWriter {
+	public LSMEntireTupleWriter(ITypeTraits[] typeTraits, int numKeyFields){
+		// Just give false as third parameter. It is never used locally.
+	    super(typeTraits, numKeyFields, false);
 	}
-	@Override
-    protected int getNullFlagsBytes(ITupleReference tuple) {
-		//+1.0 is for insert/delete tuple checking
-        return (int) Math.ceil(((double) tuple.getFieldCount() + 1.0) / 8.0);
-    }
 	
 	@Override
     public int writeTuple(ITupleReference tuple, byte[] targetBuf, int targetOff) {
@@ -20,6 +19,20 @@
 		byte[] buf = tuple.getFieldData(0);
 		int tupleStartOff = ((LSMTypeAwareTupleReference)tuple).getTupleStart();
 		System.arraycopy(buf, tupleStartOff, targetBuf, targetOff, tupleSize);
+		//System.out.println("BEF: " + printTuple(tuple));
         return tupleSize;
     }
+	
+	private String printTuple(ITupleReference tuple) {
+        LSMTypeAwareTupleReference lsmTuple = (LSMTypeAwareTupleReference)tuple;
+        ISerializerDeserializer[] fieldSerdes = new ISerializerDeserializer[] { IntegerSerializerDeserializer.INSTANCE, IntegerSerializerDeserializer.INSTANCE };
+        String s = null;
+        try {
+            s = TupleUtils.printTuple(lsmTuple, fieldSerdes);
+        } catch (HyracksDataException e) {
+            e.printStackTrace();
+        }
+        s += " " + lsmTuple.isDelete();
+        return s;
+    }
 }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriterFactory.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriterFactory.java
index 58d61f9..b214a6a 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriterFactory.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMEntireTupleWriterFactory.java
@@ -6,15 +6,17 @@
 
 public class LSMEntireTupleWriterFactory extends TypeAwareTupleWriterFactory {
 	private static final long serialVersionUID = 1L;
-	private ITypeTraits[] typeTraits;
+	private final ITypeTraits[] typeTraits;
+	private final int numKeyFields;
 	
-	public LSMEntireTupleWriterFactory(ITypeTraits[] typeTraits) {
+	public LSMEntireTupleWriterFactory(ITypeTraits[] typeTraits, int numKeyFields) {
 		super(typeTraits);
 		this.typeTraits = typeTraits;
+		this.numKeyFields = numKeyFields;
 	}
 
 	@Override
 	public ITreeIndexTupleWriter createTupleWriter() {
-		return new LSMEntireTupleWriter(typeTraits);
+		return new LSMEntireTupleWriter(typeTraits, numKeyFields);
 	}
 }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReference.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReference.java
index 7b5da79..21a0edb 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReference.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReference.java
@@ -1,14 +1,60 @@
 package edu.uci.ics.hyracks.storage.am.lsm.tuples;
 
+import java.nio.ByteBuffer;
+
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
+import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexFrame;
 import edu.uci.ics.hyracks.storage.am.common.tuples.TypeAwareTupleReference;
 
 public class LSMTypeAwareTupleReference extends TypeAwareTupleReference implements ILSMTreeTupleReference {
 
-	public LSMTypeAwareTupleReference(ITypeTraits[] typeTraits) {
+    // Indicates whether the last call to setFieldCount() was initiated by
+    // called by the outside or whether it was called internally to set up an
+    // antimatter tuple.
+    private boolean resetFieldCount = false;
+    private final int numKeyFields;
+    
+    public LSMTypeAwareTupleReference(ITypeTraits[] typeTraits, int numKeyFields) {
 		super(typeTraits);
+		this.numKeyFields = numKeyFields;
 	}
 
+    public void setFieldCount(int fieldCount) {
+        super.setFieldCount(fieldCount);
+        // Don't change the fieldCount in resetTuple*() calls.
+        resetFieldCount = false;
+    }
+
+    @Override
+    public void setFieldCount(int fieldStartIndex, int fieldCount) {
+        super.setFieldCount(fieldStartIndex, fieldCount);
+        // Don't change the fieldCount in resetTuple*() calls.
+        resetFieldCount = false;
+    }
+    
+    @Override
+    public void resetByTupleOffset(ByteBuffer buf, int tupleStartOff) {
+        this.buf = buf;
+        this.tupleStartOff = tupleStartOff;
+        if (numKeyFields != typeTraits.length) {
+            if (isDelete()) {
+                setFieldCount(numKeyFields);
+                // Reset the original field count for matter tuples.
+                resetFieldCount = true;
+            } else {
+                if (resetFieldCount) {
+                    setFieldCount(typeTraits.length);
+                }
+            }
+        }
+        super.resetByTupleOffset(buf, tupleStartOff);
+    }
+    
+    @Override
+    public void resetByTupleIndex(ITreeIndexFrame frame, int tupleIndex) {
+        resetByTupleOffset(frame.getBuffer(), frame.getTupleOffset(tupleIndex));
+    }
+    
 	@Override
 	protected int getNullFlagsBytes() {
 		// +1.0 is for insert/delete tuple checking.
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReferenceTest.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReferenceTest.java
deleted file mode 100644
index 2b94b64..0000000
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleReferenceTest.java
+++ /dev/null
@@ -1,71 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsm.tuples;
-
-import static org.junit.Assert.fail;
-
-import java.nio.ByteBuffer;
-
-import org.junit.Test;
-
-import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
-import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleReference;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
-
-public class LSMTypeAwareTupleReferenceTest {
-
-    @Test
-    public void test01() throws Exception {
-    
-    	int targetOff = 0;
-    	ByteBuffer buf = ByteBuffer.allocate(32);
-    	
-        int fieldCount = 2;
-        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-        
-        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
-        ITreeIndexTupleWriter insertTupleWriter = insertTupleWriterFactory.createTupleWriter();
-        ITreeIndexTupleReference lsmTupleReference  = insertTupleWriter.createTupleReference();
-        
-        lsmTupleReference.resetByTupleOffset(buf, targetOff);
-        insertTupleWriter.writeTuple(lsmTupleReference, buf, targetOff);
-        
-        boolean del = ((LSMTypeAwareTupleReference) lsmTupleReference).isDelete();
-        
-        if(del == false) {
-        	return;
-        }
-        else {
-        	fail("fail to write tuple");
-        }	
-    }
-    
-    @Test
-    public void test02() throws Exception {
-    
-    	int targetOff = 0;
-    	ByteBuffer buf = ByteBuffer.allocate(32);
-    	
-        int fieldCount = 2;
-        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-        
-        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
-        ITreeIndexTupleWriter deleteTupleWriter = deleteTupleWriterFactory.createTupleWriter();
-        ITreeIndexTupleReference lsmTupleReference  = deleteTupleWriter.createTupleReference();
-        
-        lsmTupleReference.resetByTupleOffset(buf, targetOff);
-        deleteTupleWriter.writeTuple(lsmTupleReference, buf, targetOff);
-        
-        boolean del = ((LSMTypeAwareTupleReference) lsmTupleReference).isDelete();
-        
-        if(del == true) {
-        	return;
-        }
-        else {
-        	fail("fail to write tuple");
-        }	
-    }
-}
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriter.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriter.java
index 117e756..c9c58e0 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriter.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriter.java
@@ -1,5 +1,7 @@
 package edu.uci.ics.hyracks.storage.am.lsm.tuples;
 
+import java.nio.ByteBuffer;
+
 import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
 import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
 import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleReference;
@@ -7,36 +9,44 @@
 
 public class LSMTypeAwareTupleWriter extends TypeAwareTupleWriter {
 	private final boolean isDelete;
+	private final int numKeyFields;
 	
-	public LSMTypeAwareTupleWriter(ITypeTraits[] typeTraits, boolean isDelete) {
+	public LSMTypeAwareTupleWriter(ITypeTraits[] typeTraits, int numKeyFields, boolean isDelete) {
 		super(typeTraits);
+		this.numKeyFields = numKeyFields;
 		this.isDelete = isDelete;
 	}
 
 	@Override
     public ITreeIndexTupleReference createTupleReference() {
-        return new LSMTypeAwareTupleReference(typeTraits);
+        return new LSMTypeAwareTupleReference(typeTraits, numKeyFields);
     }
 	
 	@Override
 	protected int getNullFlagsBytes(int numFields) {
-		//+1.0 is for insert/delete tuple checking
-		return (int) Math.ceil(((double) numFields + 1.0)/ 8.0);
+		// +1.0 is for insert/delete tuple checking.
+		return (int) Math.ceil(((double) numFields + 1.0) / 8.0);
     }
 	
 	@Override
     protected int getNullFlagsBytes(ITupleReference tuple) {
-		//+1.0 is for insert/delete tuple checking
+		// +1.0 is for insert/delete tuple checking.
         return (int) Math.ceil(((double) tuple.getFieldCount() + 1.0) / 8.0);
     }
 	
 	@Override
-    public int writeTuple(ITupleReference tuple, byte[] targetBuf, int targetOff) {
-		int bytesWritten = super.writeTuple(tuple, targetBuf, targetOff);
-        if(isDelete) {
-        	setDeleteBit(targetBuf, targetOff);
-        }
-        return bytesWritten;
+    public int writeTuple(ITupleReference tuple, byte[] targetBuf, int targetOff) {	    
+	    int bytesWritten = -1;
+	    if (isDelete) {
+	        //System.out.println("DELETE FIELDS: " + tuple.getFieldCount());
+	        // TODO: Avoid generating an object here.
+	        ByteBuffer buf = ByteBuffer.wrap(targetBuf);
+	        bytesWritten = super.writeTupleFields(tuple, 0, numKeyFields, buf, targetOff);
+	        setDeleteBit(targetBuf, targetOff);
+		} else {
+		    bytesWritten = super.writeTuple(tuple, targetBuf, targetOff);
+		}
+	    return bytesWritten;
     }
 	
 	private void setDeleteBit(byte[] targetBuf, int targetOff) {
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactory.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactory.java
index 236c15c..6dd0afa 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactory.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactory.java
@@ -7,18 +7,20 @@
 public class LSMTypeAwareTupleWriterFactory extends TypeAwareTupleWriterFactory {
 
 	private static final long serialVersionUID = 1L;
-	private ITypeTraits[] typeTraits;
+	private final ITypeTraits[] typeTraits;
+	private final int numKeyFields;
 	private final boolean isDelete;
 	
-	public LSMTypeAwareTupleWriterFactory(ITypeTraits[] typeTraits, boolean isDelete) {
+	public LSMTypeAwareTupleWriterFactory(ITypeTraits[] typeTraits, int numKeyFields, boolean isDelete) {
 		super(typeTraits);
 		this.typeTraits = typeTraits;
+		this.numKeyFields = numKeyFields;
 		this.isDelete = isDelete;
 	}
 
 	@Override
 	public ITreeIndexTupleWriter createTupleWriter() {
-		return new LSMTypeAwareTupleWriter(typeTraits, isDelete);
+		return new LSMTypeAwareTupleWriter(typeTraits, numKeyFields, isDelete);
 	}
 
 }
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactoryTest.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactoryTest.java
deleted file mode 100644
index 0731f3b..0000000
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterFactoryTest.java
+++ /dev/null
@@ -1,31 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsm.tuples;
-
-import static org.junit.Assert.fail;
-
-import org.junit.Test;
-
-import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
-import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
-
-public class LSMTypeAwareTupleWriterFactoryTest {
-    @Test
-    public void test01() throws Exception {
-
-        // declare fields
-        int fieldCount = 2;
-        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-	    
-	    LSMTypeAwareTupleWriterFactory lsmTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
-	    ITreeIndexTupleWriter lsmTupleWriter = lsmTupleWriterFactory.createTupleWriter();
-
-	    if(lsmTupleWriter != null) {
-	    	return;
-	    }
-	    else {
-	    	fail("fail to create LSMTypeAwareTupleWriter!");
-	    }
-	}
-}
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterTest.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterTest.java
deleted file mode 100644
index 230231d..0000000
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/tuples/LSMTypeAwareTupleWriterTest.java
+++ /dev/null
@@ -1,101 +0,0 @@
-package edu.uci.ics.hyracks.storage.am.lsm.tuples;
-
-import static org.junit.Assert.fail;
-
-import java.nio.ByteBuffer;
-
-import org.junit.Test;
-
-import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
-import edu.uci.ics.hyracks.data.std.primitive.IntegerPointable;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleReference;
-import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleWriter;
-
-public class LSMTypeAwareTupleWriterTest {
-
-	// test create tuple reference
-	@Test
-	public void test01() throws Exception {
-
-		// declare fields
-		int fieldCount = 2;
-		ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-		typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-		typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-
-		LSMTypeAwareTupleWriterFactory lsmTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(
-				typeTraits, false);
-		ITreeIndexTupleWriter lsmTupleWriter = lsmTupleWriterFactory.createTupleWriter();
-
-		ITreeIndexTupleReference lsmTupleReference = lsmTupleWriter.createTupleReference();
-
-		if (lsmTupleReference != null) {
-			return;
-		} else {
-			fail("fail to create LSMTypeAwareTupleWriter");
-		}
-	}
-
-	// test insert tuple writer
-	@Test
-	public void test02() throws Exception {
-
-		int targetOff = 0;
-		ByteBuffer buf = ByteBuffer.allocate(32);
-
-		// declare fields
-        int fieldCount = 2;
-        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-
-		LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(
-				typeTraits, false);
-		ITreeIndexTupleWriter insertTupleWriter = insertTupleWriterFactory.createTupleWriter();
-		ITreeIndexTupleReference insertTupleReference = insertTupleWriter.createTupleReference();
-
-		insertTupleReference.resetByTupleOffset(buf, targetOff);
-
-		int num = insertTupleWriter.writeTuple(insertTupleReference, buf, targetOff);
-
-		boolean del = ((LSMTypeAwareTupleReference) insertTupleReference).isDelete();
-
-		if (num == 9 && del == false) {
-			return;
-		} else {
-			fail("fail to write tuple");
-		}
-	}
-
-	// test delete tuple writer
-	@Test
-	public void test03() throws Exception {
-
-		int targetOff = 0;
-		ByteBuffer buf = ByteBuffer.allocate(32);
-
-		// declare fields
-        int fieldCount = 3;
-        ITypeTraits[] typeTraits = new ITypeTraits[fieldCount];
-        typeTraits[0] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[1] = IntegerPointable.TYPE_TRAITS;
-        typeTraits[2] = IntegerPointable.TYPE_TRAITS;
-
-		LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(
-				typeTraits, true);
-		ITreeIndexTupleWriter deleteTupleWriter = deleteTupleWriterFactory.createTupleWriter();
-		ITreeIndexTupleReference deleteTupleReference = deleteTupleWriter.createTupleReference();
-
-		deleteTupleReference.resetByTupleOffset(buf, targetOff);
-
-		int num = deleteTupleWriter.writeTuple(deleteTupleReference, buf, targetOff);
-
-		boolean del = ((LSMTypeAwareTupleReference) deleteTupleReference).isDelete();
-
-		if (num == 13 && del == true) {
-			return;
-		} else {
-			fail("fail to write tuple");
-		}
-	}
-}
diff --git a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/util/LSMBTreeUtils.java b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/util/LSMBTreeUtils.java
index c6bb3fd..c8468ae 100644
--- a/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/util/LSMBTreeUtils.java
+++ b/hyracks-storage-am-lsm-btree/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/util/LSMBTreeUtils.java
@@ -30,6 +30,7 @@
 import edu.uci.ics.hyracks.storage.am.lsm.impls.BTreeFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.impls.LSMBTreeFileNameManager;
 import edu.uci.ics.hyracks.storage.am.lsm.impls.LSMTree;
+import edu.uci.ics.hyracks.storage.am.lsm.tuples.LSMEntireTupleWriterFactory;
 import edu.uci.ics.hyracks.storage.am.lsm.tuples.LSMTypeAwareTupleWriterFactory;
 import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
 import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
@@ -39,19 +40,21 @@
             String onDiskDir, IBufferCache diskBufferCache, IFileMapProvider diskFileMapProvider,
             ITypeTraits[] typeTraits, IBinaryComparator[] cmps) {
         MultiComparator cmp = new MultiComparator(cmps);
-        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, false);
-        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, true);
+        LSMTypeAwareTupleWriterFactory insertTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, cmps.length, false);
+        LSMTypeAwareTupleWriterFactory deleteTupleWriterFactory = new LSMTypeAwareTupleWriterFactory(typeTraits, cmps.length, true);
+        LSMEntireTupleWriterFactory copyTupleWriterFactory = new LSMEntireTupleWriterFactory(typeTraits, cmps.length);
         ITreeIndexFrameFactory insertLeafFrameFactory = new BTreeNSMLeafFrameFactory(insertTupleWriterFactory);
+        ITreeIndexFrameFactory copyTupleLeafFrameFactory = new BTreeNSMLeafFrameFactory(copyTupleWriterFactory);
         ITreeIndexFrameFactory deleteLeafFrameFactory = new BTreeNSMLeafFrameFactory(deleteTupleWriterFactory);
         ITreeIndexFrameFactory interiorFrameFactory = new BTreeNSMInteriorFrameFactory(insertTupleWriterFactory);
         ITreeIndexMetaDataFrameFactory metaFrameFactory = new LIFOMetaDataFrameFactory();
         LinkedListFreePageManagerFactory freePageManagerFactory = new LinkedListFreePageManagerFactory(diskBufferCache,
                 metaFrameFactory);
-        BTreeFactory btreeFactory = new BTreeFactory(diskBufferCache, freePageManagerFactory, cmp, typeTraits.length,
-                interiorFrameFactory, insertLeafFrameFactory);
+        BTreeFactory diskBTreeFactory = new BTreeFactory(diskBufferCache, freePageManagerFactory, cmp, typeTraits.length,
+                interiorFrameFactory, copyTupleLeafFrameFactory);
         ILSMFileNameManager fileNameManager = new LSMBTreeFileNameManager(onDiskDir);
         LSMTree lsmTree = new LSMTree(memBufferCache, memFreePageManager, interiorFrameFactory, insertLeafFrameFactory,
-                deleteLeafFrameFactory, fileNameManager, btreeFactory, diskFileMapProvider, typeTraits.length, cmp);
+                deleteLeafFrameFactory, fileNameManager, diskBTreeFactory, diskFileMapProvider, typeTraits.length, cmp);
         return lsmTree;
     }
 }
diff --git a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/freepage/InMemoryFreePageManager.java b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/freepage/InMemoryFreePageManager.java
index 70d8e6c..304aa43 100644
--- a/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/freepage/InMemoryFreePageManager.java
+++ b/hyracks-storage-am-lsm-common/src/main/java/edu/uci/ics/hyracks/storage/am/lsm/common/freepage/InMemoryFreePageManager.java
@@ -76,7 +76,7 @@
 
     @Override
     public byte getMetaPageLevelIndicator() {
-        return 0;
+    	return 0;
     }
 
     @Override
diff --git a/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeTestHarness.java b/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeTestHarness.java
index e1157e6..b2db656 100644
--- a/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeTestHarness.java
+++ b/hyracks-tests/hyracks-storage-am-lsm-btree-test/src/test/java/edu/uci/ics/hyracks/storage/am/lsm/btree/util/LSMBTreeTestHarness.java
@@ -40,10 +40,11 @@
     
     private static final long RANDOM_SEED = 50;
     private static final int DEFAULT_DISK_PAGE_SIZE = 256;
-    private static final int DEFAULT_DISK_NUM_PAGES = 100;
-    private static final int DEFAULT_DISK_MAX_OPEN_FILES = 100;
+    private static final int DEFAULT_DISK_NUM_PAGES = 1000;
+    private static final int DEFAULT_DISK_MAX_OPEN_FILES = 200;
     private static final int DEFAULT_MEM_PAGE_SIZE = 256;
     private static final int DEFAULT_MEM_NUM_PAGES = 100;
+    //private static final int DEFAULT_MEM_NUM_PAGES = 10;
     private static final int DEFAULT_HYRACKS_FRAME_SIZE = 128;
     private static final int DUMMY_FILE_ID = -1;           
     
@@ -92,7 +93,7 @@
         TestStorageManagerComponentHolder.init(diskPageSize, diskNumPages, diskMaxOpenFiles);
         diskBufferCache = TestStorageManagerComponentHolder.getBufferCache(ctx);
         diskFileMapProvider = TestStorageManagerComponentHolder.getFileMapProvider(ctx);
-        memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), getMemPageSize(), getDiskPageSize());
+        memBufferCache = new InMemoryBufferCache(new HeapBufferAllocator(), memPageSize, memNumPages);
         memFreePageManager = new InMemoryFreePageManager(memNumPages, new LIFOMetaDataFrameFactory());
         rnd.setSeed(RANDOM_SEED);
     }