hyracks-storage-am-rtree/src/main/java/edu/uci/ics/hyracks/storage/am/rtree/impls/UnorderedSlotManager.java

/*
 * 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.rtree.impls;

import edu.uci.ics.hyracks.dataflow.common.data.accessors.ITupleReference;
import edu.uci.ics.hyracks.storage.am.common.api.ITreeIndexTupleReference;
import edu.uci.ics.hyracks.storage.am.common.frames.AbstractSlotManager;
import edu.uci.ics.hyracks.storage.am.common.ophelpers.FindTupleMode;
import edu.uci.ics.hyracks.storage.am.common.ophelpers.FindTupleNoExactMatchPolicy;
import edu.uci.ics.hyracks.storage.am.common.ophelpers.MultiComparator;
import edu.uci.ics.hyracks.storage.am.rtree.frames.RTreeNSMFrame;

public class UnorderedSlotManager extends AbstractSlotManager {
	@Override
	public int findTupleIndex(ITupleReference searchKey,
			ITreeIndexTupleReference frameTuple, MultiComparator multiCmp,
			FindTupleMode mode, FindTupleNoExactMatchPolicy matchPolicy) {

		int maxFieldPos = multiCmp.getKeyFieldCount() / 2;
		for (int i = 0; i < frame.getTupleCount(); i++) {
			frameTuple.resetByTupleIndex(frame, i);

			boolean foundTuple = true;
			for (int j = 0; j < maxFieldPos; j++) {
				int k = maxFieldPos + j;
				int c1 = multiCmp.getComparators()[j].compare(
						frameTuple.getFieldData(j),
						frameTuple.getFieldStart(j),
						frameTuple.getFieldLength(j),
						searchKey.getFieldData(j), searchKey.getFieldStart(j),
						searchKey.getFieldLength(j));

				if (c1 != 0) {
					foundTuple = false;
					break;
				}
				int c2 = multiCmp.getComparators()[k].compare(
						frameTuple.getFieldData(k),
						frameTuple.getFieldStart(k),
						frameTuple.getFieldLength(k),
						searchKey.getFieldData(k), searchKey.getFieldStart(k),
						searchKey.getFieldLength(k));
				if (c2 != 0) {
					foundTuple = false;
					break;
				}
			}
			int remainingFieldCount = multiCmp.getFieldCount()
					- multiCmp.getKeyFieldCount();
			for (int j = multiCmp.getKeyFieldCount(); j < multiCmp
					.getKeyFieldCount() + remainingFieldCount; j++) {
				if (!compareField(searchKey, frameTuple, j)) {
					foundTuple = false;
					break;
				}
			}
			if (foundTuple) {
				return i;
			}
		}
		return -1;
	}

	public boolean compareField(ITupleReference searchKey,
			ITreeIndexTupleReference frameTuple, int fIdx) {
		int searchKeyFieldLength = searchKey.getFieldLength(fIdx);
		int frameTupleFieldLength = frameTuple.getFieldLength(fIdx);

		if (searchKeyFieldLength != frameTupleFieldLength) {
			return false;
		}

		for (int i = 0; i < searchKeyFieldLength; i++) {
			if (searchKey.getFieldData(fIdx)[i + searchKey.getFieldStart(fIdx)] != frameTuple
					.getFieldData(fIdx)[i + frameTuple.getFieldStart(fIdx)]) {
				return false;
			}
		}
		return true;
	}

	@Override
	public int insertSlot(int tupleIndex, int tupleOff) {
		int slotOff = getSlotEndOff() - slotSize;
		setSlot(slotOff, tupleOff);
		return slotOff;
	}

	public void modifySlot(int slotOff, int tupleOff) {
		setSlot(slotOff, tupleOff);
	}

	public void deleteEmptySlots() {
		int slotOff = getSlotStartOff();
		while (slotOff >= getSlotEndOff()) {
			if (frame.getBuffer().getInt(slotOff) == -1) {
				while (frame.getBuffer().getInt(getSlotEndOff()) == -1) {
					((RTreeNSMFrame) frame)
							.setTupleCount(frame.getTupleCount() - 1);
				}
				if (slotOff > getSlotEndOff()) {
					System.arraycopy(frame.getBuffer().array(),
							getSlotEndOff(), frame.getBuffer().array(),
							slotOff, slotSize);
					((RTreeNSMFrame) frame)
							.setTupleCount(frame.getTupleCount() - 1);
				} else {
					break;
				}
			}
			slotOff -= slotSize;
		}
	}
}