hyracks-storage-am-btree/src/main/java/edu/uci/ics/hyracks/storage/am/btree/frames/NSMInteriorFrame.java - asterixdb - Git at Google

 /*
  * 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.btree.frames;

 import java.io.ByteArrayInputStream;
 import java.io.DataInput;
 import java.io.DataInputStream;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;

 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.storage.am.btree.api.IBTreeFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
 import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
 import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
 import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
 import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
 import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
 import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;

 public class NSMInteriorFrame extends NSMFrame implements IBTreeInteriorFrame {

 	private static final int rightLeafOff = smFlagOff + 1;

 	private static final int childPtrSize = 4;

 	//private SimpleTupleReference cmpFrameTuple = new SimpleTupleReference();
 	private IBTreeTupleReference cmpFrameTuple;

 	public NSMInteriorFrame(IBTreeTupleWriter tupleWriter) {
 		super(tupleWriter);
 		cmpFrameTuple = tupleWriter.createTupleReference();

 	}

 	private int getLeftChildPageOff(ITupleReference tuple, MultiComparator cmp) {
 		return tuple.getFieldStart(cmp.getKeyFieldCount()-1) + tuple.getFieldLength(cmp.getKeyFieldCount()-1);
 	}

 	@Override
 	public void initBuffer(byte level) {
 		super.initBuffer(level);
 		buf.putInt(rightLeafOff, -1);
 	}

 	@Override
 	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {
 		int bytesRequired = tupleWriter.bytesRequired(tuple) + 8; // for the two childpointers
 		if(bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
 		else if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
 		else return SpaceStatus.INSUFFICIENT_SPACE;
 	}

 	@Override
 	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		boolean isDuplicate = true;

 		if(tupleIndex < 0) isDuplicate = false; // greater than all existing keys
 		else {
 			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
 			if(cmp.compare(tuple, frameTuple) != 0) isDuplicate = false;
 		}

 		if(isDuplicate) {
 			throw new BTreeException("Trying to insert duplicate value into interior node.");
 		}
 		else {
 			slotOff = slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));

 			int freeSpace = buf.getInt(freeSpaceOff);
 			int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
 			System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
 			int tupleSize = bytesWritten + childPtrSize;

 			buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 			buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
 			buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());

 			// did insert into the rightmost slot?
 			if(slotOff == slotManager.getSlotEndOff()) {
 				System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
 			}
 			else {
 				// if slotOff has a right (slot-)neighbor then update its child pointer
 				// the only time when this is NOT the case, is when this is the first tuple
 				// (or when the splitkey goes into the rightmost slot but that case was handled in the if above)
 				if(buf.getInt(tupleCountOff) > 1) {
 					int rightNeighborOff = slotOff - slotManager.getSlotSize();
 					frameTuple.resetByOffset(buf, slotManager.getTupleOff(rightNeighborOff));
 					System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), getLeftChildPageOff(frameTuple, cmp), childPtrSize);
 				}
 			}
 		}
 	}

 	@Override
 	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {
 		int freeSpace = buf.getInt(freeSpaceOff);
 		slotManager.insertSlot(-1, freeSpace);
 		int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
 		System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
 		int tupleSize = bytesWritten + childPtrSize;
 		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
 		buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
 		System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
 	}

 	@Override
 	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
 		// before doing anything check if key already exists
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_EXACT);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		if(tupleIndex >= 0) {
 			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
 			if(cmp.compare(tuple, frameTuple) == 0) {
 				throw new BTreeException("Inserting duplicate key in interior node during split");
 			}
 		}

 		ByteBuffer right = rightFrame.getBuffer();
 		int tupleCount = buf.getInt(tupleCountOff);

 		int tuplesToLeft = (tupleCount / 2) + (tupleCount % 2);
 		IBTreeFrame targetFrame = null;
 		frameTuple.resetByOffset(buf, getTupleOffset(tuplesToLeft-1));
 		if(cmp.compare(tuple, frameTuple) <= 0) {
 			targetFrame = this;
 		}
 		else {
 			targetFrame = rightFrame;
 		}
 		int tuplesToRight = tupleCount - tuplesToLeft;

 		// copy entire page
 		System.arraycopy(buf.array(), 0, right.array(), 0, buf.capacity());

 		// on right page we need to copy rightmost slots to left
 		int src = rightFrame.getSlotManager().getSlotEndOff();
 		int dest = rightFrame.getSlotManager().getSlotEndOff() + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
 		int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight;
 		System.arraycopy(right.array(), src, right.array(), dest, length);
 		right.putInt(tupleCountOff, tuplesToRight);

 		// on left page, remove highest key and make its childpointer the rightmost childpointer
 		buf.putInt(tupleCountOff, tuplesToLeft);

 		// copy data to be inserted, we need this because creating the splitkey will overwrite the data param (data points to same memory as splitKey.getData())
 		SplitKey savedSplitKey = splitKey.duplicate(tupleWriter.createTupleReference());

 		// set split key to be highest value in left page
 		int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, tupleOff);
 		int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		splitKey.initData(splitKeySize);
 		tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);
 		splitKey.getTuple().resetByOffset(splitKey.getBuffer(), 0);

 		int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 		frameTuple.resetByOffset(buf, deleteTupleOff);
 		buf.putInt(rightLeafOff, buf.getInt(getLeftChildPageOff(frameTuple, cmp)));
 		buf.putInt(tupleCountOff, tuplesToLeft - 1);

 		// compact both pages
 		rightFrame.compact(cmp);
 		compact(cmp);

 		// insert key
 		targetFrame.insert(savedSplitKey.getTuple(), cmp);

 		return 0;
 	}

 	@Override
 	public void compact(MultiComparator cmp) {
 		resetSpaceParams();

 		frameTuple.setFieldCount(cmp.getKeyFieldCount());

 		int tupleCount = buf.getInt(tupleCountOff);
 		int freeSpace = buf.getInt(freeSpaceOff);

 		ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
 		sortedTupleOffs.ensureCapacity(tupleCount);
 		for(int i = 0; i < tupleCount; i++) {
 			int slotOff = slotManager.getSlotOff(i);
 			int tupleOff = slotManager.getTupleOff(slotOff);
 			sortedTupleOffs.add(new SlotOffTupleOff(i, slotOff, tupleOff));
 		}
 		Collections.sort(sortedTupleOffs);

 		for(int i = 0; i < sortedTupleOffs.size(); i++) {
 			int tupleOff = sortedTupleOffs.get(i).tupleOff;
 			frameTuple.resetByOffset(buf, tupleOff);

 			int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
 			int tupleLength = tupleEndOff - tupleOff + childPtrSize;
 			System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);

 			slotManager.setSlot(sortedTupleOffs.get(i).slotOff, freeSpace);
 			freeSpace += tupleLength;
 		}

 		buf.putInt(freeSpaceOff, freeSpace);
 		buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
 	}

 	@Override
 	public int getChildPageId(RangePredicate pred, MultiComparator srcCmp) {
 		// check for trivial case where there is only a child pointer (and no key)
 		if(buf.getInt(tupleCountOff) == 0) {
 			return buf.getInt(rightLeafOff);
 		}

 		cmpFrameTuple.setFieldCount(srcCmp.getKeyFieldCount());
 		frameTuple.setFieldCount(srcCmp.getKeyFieldCount());

 		// check for trivial cases where no low key or high key exists (e.g. during an index scan)
 		ITupleReference tuple = null;
 		FindSlotMode fsm = null;
 		if(pred.isForward()) {
 			tuple = pred.getLowKey();
 			if(tuple == null) {
 				return getLeftmostChildPageId(srcCmp);
 			}
 			if(pred.isLowKeyInclusive()) fsm = FindSlotMode.FSM_INCLUSIVE;
 			else fsm = FindSlotMode.FSM_EXCLUSIVE;
 		}
 		else {
 			tuple = pred.getHighKey();
 			if(tuple == null) {
 				return getRightmostChildPageId(srcCmp);
 			}
 			fsm = FindSlotMode.FSM_INCLUSIVE;
 		}

 		MultiComparator targetCmp = pred.getComparator();
 		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, targetCmp, fsm);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		if(tupleIndex < 0) {
 			return buf.getInt(rightLeafOff);
 		}
 		else {
 			int origTupleOff = slotManager.getTupleOff(slotOff);
 			cmpFrameTuple.resetByOffset(buf, origTupleOff);
 			int cmpTupleOff = origTupleOff;
 			if(pred.isForward()) {
 				int maxSlotOff = buf.capacity();
 				slotOff += slotManager.getSlotSize();
 				while(slotOff < maxSlotOff) {
 					cmpTupleOff = slotManager.getTupleOff(slotOff);
 					frameTuple.resetByOffset(buf, cmpTupleOff);
 					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
 					slotOff += slotManager.getSlotSize();
 				}
 				slotOff -= slotManager.getSlotSize();
 			}
 			else {
 				int minSlotOff = slotManager.getSlotEndOff() - slotManager.getSlotSize();
 				slotOff -= slotManager.getSlotSize();
 				while(slotOff > minSlotOff) {
 					cmpTupleOff = slotManager.getTupleOff(slotOff);
 					frameTuple.resetByOffset(buf, cmpTupleOff);
 					if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
 					slotOff -= slotManager.getSlotSize();
 				}
 				slotOff += slotManager.getSlotSize();
 			}

 			frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
 			int childPageOff = getLeftChildPageOff(frameTuple, srcCmp);
 			return buf.getInt(childPageOff);
 		}
 	}

 	@Override
 	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
 		int slotOff = slotManager.getSlotOff(tupleIndex);
 		int tupleOff;
 		int keySize;

 		if(tupleIndex < 0) {
 			tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
 			frameTuple.resetByOffset(buf, tupleOff);
 			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());

 			// copy new rightmost pointer
 			System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
 		}
 		else {
 			tupleOff = slotManager.getTupleOff(slotOff);
 			frameTuple.resetByOffset(buf, tupleOff);
 			keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 			// perform deletion (we just do a memcpy to overwrite the slot)
 			int slotStartOff = slotManager.getSlotEndOff();
 			int length = slotOff - slotStartOff;
 			System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
 		}

 		// maintain space information
 		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
 	}

 	@Override
 	protected void resetSpaceParams() {
 		buf.putInt(freeSpaceOff, rightLeafOff + childPtrSize);
 		buf.putInt(totalFreeSpaceOff, buf.capacity() - (rightLeafOff + childPtrSize));
 	}

 	@Override
 	public int getLeftmostChildPageId(MultiComparator cmp) {
 		int tupleOff = slotManager.getTupleOff(slotManager.getSlotStartOff());
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		frameTuple.resetByOffset(buf, tupleOff);
 		int childPageOff = getLeftChildPageOff(frameTuple, cmp);
 		return buf.getInt(childPageOff);
 	}

 	@Override
 	public int getRightmostChildPageId(MultiComparator cmp) {
 		return buf.getInt(rightLeafOff);
 	}

 	@Override
 	public void setRightmostChildPageId(int pageId) {
 		buf.putInt(rightLeafOff, pageId);
 	}

 	// for debugging
 	public ArrayList<Integer> getChildren(MultiComparator cmp) {
 		ArrayList<Integer> ret = new ArrayList<Integer>();
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		int tupleCount = buf.getInt(tupleCountOff);
 		for(int i = 0; i < tupleCount; i++) {
 			int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
 			frameTuple.resetByOffset(buf, tupleOff);
 			int intVal = getInt(buf.array(), frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1));
 			ret.add(intVal);
 		}
 		if(!isLeaf()) {
 			int rightLeaf = buf.getInt(rightLeafOff);
 			if(rightLeaf > 0) ret.add(buf.getInt(rightLeafOff));
 		}
 		return ret;
 	}

 	@Override
 	public void deleteGreatest(MultiComparator cmp) {
 		int slotOff = slotManager.getSlotEndOff();
 		int tupleOff = slotManager.getTupleOff(slotOff);
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		frameTuple.resetByOffset(buf, tupleOff);
 		int keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
 		System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);

 		// maintain space information
 		buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
 		buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());

 		int freeSpace = buf.getInt(freeSpaceOff);
 		if(freeSpace == tupleOff + keySize + childPtrSize) {
 			buf.putInt(freeSpace, freeSpace - (keySize + childPtrSize));
 		}
 	}

 	private int getInt(byte[] bytes, int offset) {
 		return ((bytes[offset] & 0xff) << 24) + ((bytes[offset + 1] & 0xff) << 16) + ((bytes[offset + 2] & 0xff) << 8) + ((bytes[offset + 3] & 0xff) << 0);
 	}

 	@Override
 	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {
 		StringBuilder strBuilder = new StringBuilder();
 		int tupleCount = buf.getInt(tupleCountOff);
 		frameTuple.setFieldCount(cmp.getKeyFieldCount());
 		for(int i = 0; i < tupleCount; i++) {
 			frameTuple.resetByTupleIndex(this, i);
 			for(int j = 0; j < cmp.getKeyFieldCount(); j++) {
 				ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(j), frameTuple.getFieldStart(j), frameTuple.getFieldLength(j));
 				DataInput dataIn = new DataInputStream(inStream);
 				Object o = fields[j].deserialize(dataIn);
 				strBuilder.append(o.toString() + " ");
 			}
 			strBuilder.append(" | ");
 		}
 		strBuilder.append("\n");
 		return strBuilder.toString();
 	}
 }
	/*
	* 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.btree.frames;

	import java.io.ByteArrayInputStream;
	import java.io.DataInput;
	import java.io.DataInputStream;
	import java.nio.ByteBuffer;
	import java.util.ArrayList;
	import java.util.Collections;

	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.storage.am.btree.api.IBTreeFrame;
	import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeInteriorFrame;
	import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleReference;
	import edu.uci.ics.hyracks.storage.am.btree.api.IBTreeTupleWriter;
	import edu.uci.ics.hyracks.storage.am.btree.impls.BTreeException;
	import edu.uci.ics.hyracks.storage.am.btree.impls.FindSlotMode;
	import edu.uci.ics.hyracks.storage.am.btree.impls.MultiComparator;
	import edu.uci.ics.hyracks.storage.am.btree.impls.RangePredicate;
	import edu.uci.ics.hyracks.storage.am.btree.impls.SlotOffTupleOff;
	import edu.uci.ics.hyracks.storage.am.btree.impls.SpaceStatus;
	import edu.uci.ics.hyracks.storage.am.btree.impls.SplitKey;

	public class NSMInteriorFrame extends NSMFrame implements IBTreeInteriorFrame {

	private static final int rightLeafOff = smFlagOff + 1;

	private static final int childPtrSize = 4;

	//private SimpleTupleReference cmpFrameTuple = new SimpleTupleReference();
	private IBTreeTupleReference cmpFrameTuple;

	public NSMInteriorFrame(IBTreeTupleWriter tupleWriter) {
	super(tupleWriter);
	cmpFrameTuple = tupleWriter.createTupleReference();

	}

	private int getLeftChildPageOff(ITupleReference tuple, MultiComparator cmp) {
	return tuple.getFieldStart(cmp.getKeyFieldCount()-1) + tuple.getFieldLength(cmp.getKeyFieldCount()-1);
	}

	@Override
	public void initBuffer(byte level) {
	super.initBuffer(level);
	buf.putInt(rightLeafOff, -1);
	}

	@Override
	public SpaceStatus hasSpaceInsert(ITupleReference tuple, MultiComparator cmp) {
	int bytesRequired = tupleWriter.bytesRequired(tuple) + 8; // for the two childpointers
	if(bytesRequired + slotManager.getSlotSize() <= buf.capacity() - buf.getInt(freeSpaceOff) - (buf.getInt(tupleCountOff) * slotManager.getSlotSize()) ) return SpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE;
	else if(bytesRequired + slotManager.getSlotSize() <= buf.getInt(totalFreeSpaceOff)) return SpaceStatus.SUFFICIENT_SPACE;
	else return SpaceStatus.INSUFFICIENT_SPACE;
	}

	@Override
	public void insert(ITupleReference tuple, MultiComparator cmp) throws Exception {
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
	int slotOff = slotManager.getSlotOff(tupleIndex);
	boolean isDuplicate = true;

	if(tupleIndex < 0) isDuplicate = false; // greater than all existing keys
	else {
	frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
	if(cmp.compare(tuple, frameTuple) != 0) isDuplicate = false;
	}

	if(isDuplicate) {
	throw new BTreeException("Trying to insert duplicate value into interior node.");
	}
	else {
	slotOff = slotManager.insertSlot(tupleIndex, buf.getInt(freeSpaceOff));

	int freeSpace = buf.getInt(freeSpaceOff);
	int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
	System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
	int tupleSize = bytesWritten + childPtrSize;

	buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
	buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
	buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());

	// did insert into the rightmost slot?
	if(slotOff == slotManager.getSlotEndOff()) {
	System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
	}
	else {
	// if slotOff has a right (slot-)neighbor then update its child pointer
	// the only time when this is NOT the case, is when this is the first tuple
	// (or when the splitkey goes into the rightmost slot but that case was handled in the if above)
	if(buf.getInt(tupleCountOff) > 1) {
	int rightNeighborOff = slotOff - slotManager.getSlotSize();
	frameTuple.resetByOffset(buf, slotManager.getTupleOff(rightNeighborOff));
	System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), getLeftChildPageOff(frameTuple, cmp), childPtrSize);
	}
	}
	}
	}

	@Override
	public void insertSorted(ITupleReference tuple, MultiComparator cmp) throws Exception {
	int freeSpace = buf.getInt(freeSpaceOff);
	slotManager.insertSlot(-1, freeSpace);
	int bytesWritten = tupleWriter.writeTupleFields(tuple, 0, cmp.getKeyFieldCount(), buf, freeSpace);
	System.arraycopy(tuple.getFieldData(cmp.getKeyFieldCount()-1), getLeftChildPageOff(tuple, cmp), buf.array(), freeSpace + bytesWritten, childPtrSize);
	int tupleSize = bytesWritten + childPtrSize;
	buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) + 1);
	buf.putInt(freeSpaceOff, buf.getInt(freeSpaceOff) + tupleSize);
	buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) - tupleSize - slotManager.getSlotSize());
	System.arraycopy(tuple.getFieldData(0), getLeftChildPageOff(tuple, cmp) + childPtrSize, buf.array(), rightLeafOff, childPtrSize);
	}

	@Override
	public int split(IBTreeFrame rightFrame, ITupleReference tuple, MultiComparator cmp, SplitKey splitKey) throws Exception {
	// before doing anything check if key already exists
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_EXACT);
	int slotOff = slotManager.getSlotOff(tupleIndex);
	if(tupleIndex >= 0) {
	frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
	if(cmp.compare(tuple, frameTuple) == 0) {
	throw new BTreeException("Inserting duplicate key in interior node during split");
	}
	}

	ByteBuffer right = rightFrame.getBuffer();
	int tupleCount = buf.getInt(tupleCountOff);

	int tuplesToLeft = (tupleCount / 2) + (tupleCount % 2);
	IBTreeFrame targetFrame = null;
	frameTuple.resetByOffset(buf, getTupleOffset(tuplesToLeft-1));
	if(cmp.compare(tuple, frameTuple) <= 0) {
	targetFrame = this;
	}
	else {
	targetFrame = rightFrame;
	}
	int tuplesToRight = tupleCount - tuplesToLeft;

	// copy entire page
	System.arraycopy(buf.array(), 0, right.array(), 0, buf.capacity());

	// on right page we need to copy rightmost slots to left
	int src = rightFrame.getSlotManager().getSlotEndOff();
	int dest = rightFrame.getSlotManager().getSlotEndOff() + tuplesToLeft * rightFrame.getSlotManager().getSlotSize();
	int length = rightFrame.getSlotManager().getSlotSize() * tuplesToRight;
	System.arraycopy(right.array(), src, right.array(), dest, length);
	right.putInt(tupleCountOff, tuplesToRight);

	// on left page, remove highest key and make its childpointer the rightmost childpointer
	buf.putInt(tupleCountOff, tuplesToLeft);

	// copy data to be inserted, we need this because creating the splitkey will overwrite the data param (data points to same memory as splitKey.getData())
	SplitKey savedSplitKey = splitKey.duplicate(tupleWriter.createTupleReference());

	// set split key to be highest value in left page
	int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
	frameTuple.resetByOffset(buf, tupleOff);
	int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
	splitKey.initData(splitKeySize);
	tupleWriter.writeTupleFields(frameTuple, 0, cmp.getKeyFieldCount(), splitKey.getBuffer(), 0);
	splitKey.getTuple().resetByOffset(splitKey.getBuffer(), 0);

	int deleteTupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
	frameTuple.resetByOffset(buf, deleteTupleOff);
	buf.putInt(rightLeafOff, buf.getInt(getLeftChildPageOff(frameTuple, cmp)));
	buf.putInt(tupleCountOff, tuplesToLeft - 1);

	// compact both pages
	rightFrame.compact(cmp);
	compact(cmp);

	// insert key
	targetFrame.insert(savedSplitKey.getTuple(), cmp);

	return 0;
	}

	@Override
	public void compact(MultiComparator cmp) {
	resetSpaceParams();

	frameTuple.setFieldCount(cmp.getKeyFieldCount());

	int tupleCount = buf.getInt(tupleCountOff);
	int freeSpace = buf.getInt(freeSpaceOff);

	ArrayList<SlotOffTupleOff> sortedTupleOffs = new ArrayList<SlotOffTupleOff>();
	sortedTupleOffs.ensureCapacity(tupleCount);
	for(int i = 0; i < tupleCount; i++) {
	int slotOff = slotManager.getSlotOff(i);
	int tupleOff = slotManager.getTupleOff(slotOff);
	sortedTupleOffs.add(new SlotOffTupleOff(i, slotOff, tupleOff));
	}
	Collections.sort(sortedTupleOffs);

	for(int i = 0; i < sortedTupleOffs.size(); i++) {
	int tupleOff = sortedTupleOffs.get(i).tupleOff;
	frameTuple.resetByOffset(buf, tupleOff);

	int tupleEndOff = frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1);
	int tupleLength = tupleEndOff - tupleOff + childPtrSize;
	System.arraycopy(buf.array(), tupleOff, buf.array(), freeSpace, tupleLength);

	slotManager.setSlot(sortedTupleOffs.get(i).slotOff, freeSpace);
	freeSpace += tupleLength;
	}

	buf.putInt(freeSpaceOff, freeSpace);
	buf.putInt(totalFreeSpaceOff, buf.capacity() - freeSpace - tupleCount * slotManager.getSlotSize());
	}

	@Override
	public int getChildPageId(RangePredicate pred, MultiComparator srcCmp) {
	// check for trivial case where there is only a child pointer (and no key)
	if(buf.getInt(tupleCountOff) == 0) {
	return buf.getInt(rightLeafOff);
	}

	cmpFrameTuple.setFieldCount(srcCmp.getKeyFieldCount());
	frameTuple.setFieldCount(srcCmp.getKeyFieldCount());

	// check for trivial cases where no low key or high key exists (e.g. during an index scan)
	ITupleReference tuple = null;
	FindSlotMode fsm = null;
	if(pred.isForward()) {
	tuple = pred.getLowKey();
	if(tuple == null) {
	return getLeftmostChildPageId(srcCmp);
	}
	if(pred.isLowKeyInclusive()) fsm = FindSlotMode.FSM_INCLUSIVE;
	else fsm = FindSlotMode.FSM_EXCLUSIVE;
	}
	else {
	tuple = pred.getHighKey();
	if(tuple == null) {
	return getRightmostChildPageId(srcCmp);
	}
	fsm = FindSlotMode.FSM_INCLUSIVE;
	}

	MultiComparator targetCmp = pred.getComparator();
	int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, targetCmp, fsm);
	int slotOff = slotManager.getSlotOff(tupleIndex);
	if(tupleIndex < 0) {
	return buf.getInt(rightLeafOff);
	}
	else {
	int origTupleOff = slotManager.getTupleOff(slotOff);
	cmpFrameTuple.resetByOffset(buf, origTupleOff);
	int cmpTupleOff = origTupleOff;
	if(pred.isForward()) {
	int maxSlotOff = buf.capacity();
	slotOff += slotManager.getSlotSize();
	while(slotOff < maxSlotOff) {
	cmpTupleOff = slotManager.getTupleOff(slotOff);
	frameTuple.resetByOffset(buf, cmpTupleOff);
	if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
	slotOff += slotManager.getSlotSize();
	}
	slotOff -= slotManager.getSlotSize();
	}
	else {
	int minSlotOff = slotManager.getSlotEndOff() - slotManager.getSlotSize();
	slotOff -= slotManager.getSlotSize();
	while(slotOff > minSlotOff) {
	cmpTupleOff = slotManager.getTupleOff(slotOff);
	frameTuple.resetByOffset(buf, cmpTupleOff);
	if(targetCmp.compare(cmpFrameTuple, frameTuple) != 0) break;
	slotOff -= slotManager.getSlotSize();
	}
	slotOff += slotManager.getSlotSize();
	}

	frameTuple.resetByOffset(buf, slotManager.getTupleOff(slotOff));
	int childPageOff = getLeftChildPageOff(frameTuple, srcCmp);
	return buf.getInt(childPageOff);
	}
	}

	@Override
	public void delete(ITupleReference tuple, MultiComparator cmp, boolean exactDelete) throws Exception {
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	int tupleIndex = slotManager.findTupleIndex(tuple, frameTuple, cmp, FindSlotMode.FSM_INCLUSIVE);
	int slotOff = slotManager.getSlotOff(tupleIndex);
	int tupleOff;
	int keySize;

	if(tupleIndex < 0) {
	tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
	frameTuple.resetByOffset(buf, tupleOff);
	keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());

	// copy new rightmost pointer
	System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);
	}
	else {
	tupleOff = slotManager.getTupleOff(slotOff);
	frameTuple.resetByOffset(buf, tupleOff);
	keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
	// perform deletion (we just do a memcpy to overwrite the slot)
	int slotStartOff = slotManager.getSlotEndOff();
	int length = slotOff - slotStartOff;
	System.arraycopy(buf.array(), slotStartOff, buf.array(), slotStartOff + slotManager.getSlotSize(), length);
	}

	// maintain space information
	buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
	buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());
	}

	@Override
	protected void resetSpaceParams() {
	buf.putInt(freeSpaceOff, rightLeafOff + childPtrSize);
	buf.putInt(totalFreeSpaceOff, buf.capacity() - (rightLeafOff + childPtrSize));
	}

	@Override
	public int getLeftmostChildPageId(MultiComparator cmp) {
	int tupleOff = slotManager.getTupleOff(slotManager.getSlotStartOff());
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	frameTuple.resetByOffset(buf, tupleOff);
	int childPageOff = getLeftChildPageOff(frameTuple, cmp);
	return buf.getInt(childPageOff);
	}

	@Override
	public int getRightmostChildPageId(MultiComparator cmp) {
	return buf.getInt(rightLeafOff);
	}

	@Override
	public void setRightmostChildPageId(int pageId) {
	buf.putInt(rightLeafOff, pageId);
	}

	// for debugging
	public ArrayList<Integer> getChildren(MultiComparator cmp) {
	ArrayList<Integer> ret = new ArrayList<Integer>();
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	int tupleCount = buf.getInt(tupleCountOff);
	for(int i = 0; i < tupleCount; i++) {
	int tupleOff = slotManager.getTupleOff(slotManager.getSlotOff(i));
	frameTuple.resetByOffset(buf, tupleOff);
	int intVal = getInt(buf.array(), frameTuple.getFieldStart(frameTuple.getFieldCount()-1) + frameTuple.getFieldLength(frameTuple.getFieldCount()-1));
	ret.add(intVal);
	}
	if(!isLeaf()) {
	int rightLeaf = buf.getInt(rightLeafOff);
	if(rightLeaf > 0) ret.add(buf.getInt(rightLeafOff));
	}
	return ret;
	}

	@Override
	public void deleteGreatest(MultiComparator cmp) {
	int slotOff = slotManager.getSlotEndOff();
	int tupleOff = slotManager.getTupleOff(slotOff);
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	frameTuple.resetByOffset(buf, tupleOff);
	int keySize = tupleWriter.bytesRequired(frameTuple, 0, cmp.getKeyFieldCount());
	System.arraycopy(buf.array(), tupleOff + keySize, buf.array(), rightLeafOff, childPtrSize);

	// maintain space information
	buf.putInt(tupleCountOff, buf.getInt(tupleCountOff) - 1);
	buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + keySize + childPtrSize + slotManager.getSlotSize());

	int freeSpace = buf.getInt(freeSpaceOff);
	if(freeSpace == tupleOff + keySize + childPtrSize) {
	buf.putInt(freeSpace, freeSpace - (keySize + childPtrSize));
	}
	}

	private int getInt(byte[] bytes, int offset) {
	return ((bytes[offset] & 0xff) << 24) + ((bytes[offset + 1] & 0xff) << 16) + ((bytes[offset + 2] & 0xff) << 8) + ((bytes[offset + 3] & 0xff) << 0);
	}

	@Override
	public String printKeys(MultiComparator cmp, ISerializerDeserializer[] fields) throws HyracksDataException {
	StringBuilder strBuilder = new StringBuilder();
	int tupleCount = buf.getInt(tupleCountOff);
	frameTuple.setFieldCount(cmp.getKeyFieldCount());
	for(int i = 0; i < tupleCount; i++) {
	frameTuple.resetByTupleIndex(this, i);
	for(int j = 0; j < cmp.getKeyFieldCount(); j++) {
	ByteArrayInputStream inStream = new ByteArrayInputStream(frameTuple.getFieldData(j), frameTuple.getFieldStart(j), frameTuple.getFieldLength(j));
	DataInput dataIn = new DataInputStream(inStream);
	Object o = fields[j].deserialize(dataIn);
	strBuilder.append(o.toString() + " ");
	}
	strBuilder.append(" \| ");
	}
	strBuilder.append("\n");
	return strBuilder.toString();
	}
	}