in hyracks/hyracks-storage-am-rtree/src/main/java/org/apache/hyracks/storage/am/rtree/frames/RTreePolicy.java [64:206]
public void split(ITreeIndexFrame leftFrame, ByteBuffer buf, ITreeIndexFrame rightFrame, ISlotManager slotManager,
ITreeIndexTupleReference frameTuple, ITupleReference tuple, ISplitKey splitKey) throws HyracksDataException {
RTreeSplitKey rTreeSplitKey = ((RTreeSplitKey) splitKey);
RTreeTypeAwareTupleWriter rTreeTupleWriterLeftFrame = ((RTreeTypeAwareTupleWriter) tupleWriter);
RTreeTypeAwareTupleWriter rTreeTupleWriterRightFrame = ((RTreeTypeAwareTupleWriter) rightFrame.getTupleWriter());
RTreeNSMFrame leftRTreeFrame = ((RTreeNSMFrame) leftFrame);
double separation = Double.NEGATIVE_INFINITY;
int seed1 = 0, seed2 = 0;
int maxFieldPos = keyValueProviders.length / 2;
for (int i = 0; i < maxFieldPos; i++) {
int j = maxFieldPos + i;
frameTuple.resetByTupleIndex(leftRTreeFrame, 0);
double leastLowerValue = keyValueProviders[i].getValue(frameTuple.getFieldData(i),
frameTuple.getFieldStart(i));
double greatestUpperValue = keyValueProviders[j].getValue(frameTuple.getFieldData(j),
frameTuple.getFieldStart(j));
double leastUpperValue = leastLowerValue;
double greatestLowerValue = greatestUpperValue;
int leastUpperIndex = 0;
int greatestLowerIndex = 0;
double width;
int tupleCount = leftRTreeFrame.getTupleCount();
for (int k = 1; k < tupleCount; ++k) {
frameTuple.resetByTupleIndex(leftRTreeFrame, k);
double lowerValue = keyValueProviders[i].getValue(frameTuple.getFieldData(i),
frameTuple.getFieldStart(i));
if (lowerValue > greatestLowerValue) {
greatestLowerIndex = k;
cmpFrameTuple.resetByTupleIndex(leftRTreeFrame, k);
greatestLowerValue = keyValueProviders[i].getValue(cmpFrameTuple.getFieldData(i),
cmpFrameTuple.getFieldStart(i));
}
double higherValue = keyValueProviders[j].getValue(frameTuple.getFieldData(j),
frameTuple.getFieldStart(j));
if (higherValue < leastUpperValue) {
leastUpperIndex = k;
cmpFrameTuple.resetByTupleIndex(leftRTreeFrame, k);
leastUpperValue = keyValueProviders[j].getValue(cmpFrameTuple.getFieldData(j),
cmpFrameTuple.getFieldStart(j));
}
leastLowerValue = Math.min(lowerValue, leastLowerValue);
greatestUpperValue = Math.max(higherValue, greatestUpperValue);
}
width = greatestUpperValue - leastLowerValue;
if (width <= 0) {
width = 1;
}
double f = (greatestLowerValue - leastUpperValue) / width;
if (f > separation) {
seed1 = leastUpperIndex;
seed2 = greatestLowerIndex;
separation = f;
}
}
if (seed1 == seed2) {
if (seed1 == 0) {
seed2 = 1;
} else {
--seed2;
}
}
int totalBytes = 0, numOfDeletedTuples = 0;
frameTuple.resetByTupleIndex(leftRTreeFrame, seed1);
rec[0].set(frameTuple, keyValueProviders);
rightFrame.insert(frameTuple, -1);
((UnorderedSlotManager) slotManager).modifySlot(slotManager.getSlotOff(seed1), -1);
totalBytes += leftRTreeFrame.getTupleSize(frameTuple);
numOfDeletedTuples++;
frameTuple.resetByTupleIndex(leftRTreeFrame, seed2);
rec[1].set(frameTuple, keyValueProviders);
int remainingTuplestoBeInsertedInRightFrame;
for (int k = 0; k < leftRTreeFrame.getTupleCount(); ++k) {
remainingTuplestoBeInsertedInRightFrame = leftRTreeFrame.getTupleCount() / 2 - rightFrame.getTupleCount();
if (remainingTuplestoBeInsertedInRightFrame == 0) {
break;
}
if (k != seed1 && k != seed2) {
frameTuple.resetByTupleIndex(leftRTreeFrame, k);
if (rec[0].enlargedArea(frameTuple, keyValueProviders) < rec[1].enlargedArea(frameTuple,
keyValueProviders)
|| leftRTreeFrame.getTupleCount() - k <= remainingTuplestoBeInsertedInRightFrame) {
rightFrame.insert(frameTuple, -1);
rec[0].enlarge(frameTuple, keyValueProviders);
((UnorderedSlotManager) slotManager).modifySlot(slotManager.getSlotOff(k), -1);
totalBytes += leftRTreeFrame.getTupleSize(frameTuple);
numOfDeletedTuples++;
} else {
rec[1].enlarge(frameTuple, keyValueProviders);
}
}
}
((UnorderedSlotManager) slotManager).deleteEmptySlots();
// maintain space information
buf.putInt(totalFreeSpaceOff, buf.getInt(totalFreeSpaceOff) + totalBytes
+ (slotManager.getSlotSize() * numOfDeletedTuples));
// compact both pages
rightFrame.compact();
leftRTreeFrame.compact();
// The assumption here is that the new tuple cannot be larger than page
// size, thus it must fit in either pages.
if (rec[0].enlargedArea(tuple, keyValueProviders) < rec[1].enlargedArea(tuple, keyValueProviders)) {
if (rightFrame.hasSpaceInsert(tuple) == FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE) {
rightFrame.insert(tuple, -1);
} else {
leftRTreeFrame.insert(tuple, -1);
}
} else if (leftRTreeFrame.hasSpaceInsert(tuple) == FrameOpSpaceStatus.SUFFICIENT_CONTIGUOUS_SPACE) {
leftRTreeFrame.insert(tuple, -1);
} else {
rightFrame.insert(tuple, -1);
}
int tupleOff = slotManager.getTupleOff(slotManager.getSlotEndOff());
frameTuple.resetByTupleOffset(buf, tupleOff);
int splitKeySize = tupleWriter.bytesRequired(frameTuple, 0, keyValueProviders.length);
splitKey.initData(splitKeySize);
leftRTreeFrame.adjustMBR();
rTreeTupleWriterLeftFrame.writeTupleFields(leftRTreeFrame.getMBRTuples(), 0, rTreeSplitKey.getLeftPageBuffer(), 0);
rTreeSplitKey.getLeftTuple().resetByTupleOffset(rTreeSplitKey.getLeftPageBuffer(), 0);
((IRTreeFrame) rightFrame).adjustMBR();
rTreeTupleWriterRightFrame.writeTupleFields(((RTreeNSMFrame) rightFrame).getMBRTuples(), 0,
rTreeSplitKey.getRightPageBuffer(), 0);
rTreeSplitKey.getRightTuple().resetByTupleOffset(rTreeSplitKey.getRightPageBuffer(), 0);
}