extends AbstractMap<K,V>

implements ConcurrentNavigableMap<K,V>,

Bind.MapWithModificationListener<K,V>,

Closeable {

/** recid under which reference to rootRecid is stored */

protected final long rootRecidRef;

/** Serializer used to convert keys from/into binary form. */

protected final BTreeKeySerializer keySerializer;

/** Serializer used to convert keys from/into binary form*/

protected final Serializer<V> valueSerializer;

/** holds node level locks*/

protected final LongConcurrentHashMap<Thread> nodeLocks = new LongConcurrentHashMap<Thread>();

/** maximal node size allowed in this BTree*/

protected final int maxNodeSize;

/** DB Engine in which entries are persisted */

protected final Engine engine;

/** is this a Map or Set? if false, entries do not have values, only keys are allowed*/

protected final boolean hasValues;

/** store values as part of BTree nodes */

protected final boolean valsOutsideNodes;

protected final List<Long> leftEdges;

private final KeySet keySet;

private final EntrySet entrySet;

private final Values values = new Values(this);

private final ConcurrentNavigableMap<K,V> descendingMap = new DescendingMap(this, null,true, null, false);

protected final Atomic.Long counter;

protected final int numberOfNodeMetas;

/**

protected final boolean closeEngine;

/* hack used for DB Catalog*/

protected static SortedMap<String, Object> preinitCatalog(DB db) {

Long rootRef = db.getEngine().get(Engine.RECID_NAME_CATALOG, Serializer.RECID);

;

//$DELAY$

if(rootRef==null){

if(db.getEngine().isReadOnly())

return Collections.unmodifiableSortedMap(new TreeMap<String, Object>());

NodeSerializer rootSerializer = new NodeSerializer(false,BTreeKeySerializer.STRING,

db.getDefaultSerializer(), 0);

BNode root = new LeafNode(BTreeKeySerializer.STRING.emptyKeys(), true,true,false, new Object[]{}, 0);

rootRef = db.getEngine().put(root, rootSerializer);

//$DELAY$

db.getEngine().update(Engine.RECID_NAME_CATALOG,rootRef, Serializer.RECID);

db.getEngine().commit();

}

Serializer valser = db.getDefaultSerializer();

if(CC.ASSERT && valser == null)

throw new AssertionError();

return new BTreeMap<String, Object>(

db.engine,

false,

Engine.RECID_NAME_CATALOG,

32,

false,

0,

BTreeKeySerializer.STRING,

valser,

0

);

}

/** if <code>valsOutsideNodes</code> is true, this class is used instead of values.

protected static final class ValRef{

/** reference to actual value */

final long recid;

public ValRef(long recid) {

this.recid = recid;

}

@Override

public boolean equals(Object obj) {

throw new IllegalAccessError();

}

@Override

public int hashCode() {

throw new IllegalAccessError();

}

@Override

public String toString() {

return "BTreeMap-ValRef["+recid+"]";

}

}

protected static final class ValRefSerializer extends Serializer<ValRef>{

@Override

public void serialize(DataOutput out, ValRef value) throws IOException {

DataIO.packLong(out,value.recid);

}

@Override

public ValRef deserialize(DataInput in, int available) throws IOException {

return new ValRef(DataIO.unpackLong(in));

}

@Override

public boolean isTrusted() {

return true;

}

@Override

public boolean equals(ValRef a1, ValRef a2) {

throw new IllegalAccessError();

}

@Override

public int hashCode(ValRef valRef) {

throw new IllegalAccessError();

}

}

/** common interface for BTree node */

public abstract static class BNode{

static final int LEFT_MASK = 1;

static final int RIGHT_MASK = 1<<1;

static final int TOO_LARGE_MASK = 1<<2;

final Object keys;

final byte flags;

public BNode(Object keys, boolean leftEdge, boolean rightEdge, boolean tooLarge){

this.keys = keys;

this.flags = (byte)(

(leftEdge?LEFT_MASK:0)|

(rightEdge?RIGHT_MASK:0)|

(tooLarge?TOO_LARGE_MASK:0)

);

}

final public Object key(BTreeKeySerializer keyser, int pos) {

if(isLeftEdge()){

if(pos--==0)

return null;

}

if(pos==keyser.length(keys) && isRightEdge())

return null;

return keyser.getKey(keys,pos);

}

final public int keysLen(BTreeKeySerializer keyser) {

return keyser.length(keys) + leftEdgeInc() + rightEdgeInc();

}

final public boolean isLeftEdge(){

return (flags&LEFT_MASK)!=0;

}

final public boolean isRightEdge(){

return (flags&RIGHT_MASK)!=0;

}

/** @return 1 if is left edge, or 0*/

final public int leftEdgeInc(){

return flags&LEFT_MASK;

}

/** @return 1 if is right edge, or 0*/

final public int rightEdgeInc(){

return (flags&RIGHT_MASK)>>>1;

}

final public boolean isTooLarge(){

return (flags&TOO_LARGE_MASK)!=0;

}

public abstract boolean isLeaf();

public abstract Object val(int pos, Serializer valueSerializer);

final public Object highKey(BTreeKeySerializer keyser) {

if(isRightEdge())

return null;

return keyser.getKey(keys,keyser.length(keys)-1);

}

public abstract Object childArray();

public abstract long child(int i);

public abstract long next();

public final int compare(final BTreeKeySerializer keyser, int pos1, int pos2){

if(pos1==pos2)

return 0;

//$DELAY$

if(isLeftEdge()){

//first position is negative infinity, so everything else is bigger

//first keys is missing in array, so adjust positions

if(pos1--==0)

return -1;

if(pos2--==0)

return 1;

}

//$DELAY$

if(isRightEdge()){

int keysLen = keyser.length(keys);

//last position is positive infinity, so everything else is smaller

if(pos1==keysLen)

return 1;

if(pos2==keysLen)

return -1;

}

return keyser.compare(keys,pos1,pos2);

}

public final int compare(final BTreeKeySerializer keyser, int pos, Object second){

if(isLeftEdge()) {

//first position is negative infinity, so everything else is bigger

//first keys is missing in array, so adjust positions

if (pos-- == 0)

return -1;

}

//$DELAY$

if(isRightEdge() && pos==keyser.length(keys)){

//last position is positive infinity, so everything else is smaller

return 1;

}

return keyser.compare(keys,pos,second);

}

public void checkStructure(BTreeKeySerializer keyser, Serializer valser){

//check all keys are sorted;

if(keyser==null)

return;

int keylen = keyser.length(keys);

int end = keylen-2+rightEdgeInc();

if(end>1){

for(int i = 1;i<=end;i++){

if(keyser.compare(keys,i-1, i)>=0)

throw new DBException.DataCorruption("keys are not sorted: "+Arrays.toString(keyser.keysToArray(keys)));

}

}

//check last key is sorted or null

if(!isRightEdge() && keylen>2){

if(keyser.compare(keys,keylen-2, keylen-1)>0){

throw new DBException.DataCorruption("Last key is not sorted: "+Arrays.toString(keyser.keysToArray(keys)));

}

}

}

public abstract BNode copyAddKey(BTreeKeySerializer keyser, Serializer valser, int pos, Object newKey, long newChild, Object newValue);

public abstract BNode copySplitRight(BTreeKeySerializer keyser, Serializer valser, int splitPos);

public abstract BNode copySplitLeft(BTreeKeySerializer keyser, Serializer valser, int splitPos, long newNext);

public abstract int valSize(Serializer valueSerializer);

public abstract int childArrayLength();

}

public final static class DirNode extends BNode{

final Object child;

DirNode(Object keys, boolean leftEdge, boolean rightEdge, boolean tooLarge, Object child) {

super(keys, leftEdge, rightEdge, tooLarge);

this.child = child;

if(CC.ASSERT)

checkStructure(null,null);

}

@Override public boolean isLeaf() { return false;}

@Override public Object val(int pos, Serializer valueSerializer){

return null;

}

@Override public Object childArray() { return child;}

@Override public long child(int pos) {

Object c = child;

return c instanceof int[]?

((int[])c)[pos]:

((long[])c)[pos];

}

@Override

public int childArrayLength() {

return child instanceof int[]?

((int[])child).length:

((long[])child).length;

}

@Override public long next() {

Object c = child;

if(c instanceof int[]){

int[] cc = (int[]) c;

return cc[cc.length-1];

}else{

long[] cc = (long[]) c;

return cc[cc.length-1];

}

}

@Override public String toString(){

return "Dir("+leftEdgeInc()+"-"+rightEdgeInc()+"-K"+Fun.toString(keys)+", C"+Fun.toString(child)+")";

}

@Override

public void checkStructure(BTreeKeySerializer keyser, Serializer valser) {

super.checkStructure(keyser,valser);

int childLen = child instanceof int[]? ((int[])child).length : ((long[])child).length;

if(keyser!=null && childLen!=keysLen(keyser))

throw new DBException.DataCorruption("bnode has inconsistent lengths");

if((isRightEdge() != (next()==0)))

throw new DBException.DataCorruption("bnode right edge inconsistent with link");

}

@Override

public DirNode copyAddKey(BTreeKeySerializer keyser, Serializer valser, int pos, Object newKey, long newChild, Object newValue) {

Object keys2 = keyser.putKey(keys, pos - leftEdgeInc(), newKey);

Object child2;

//expand children and put new child

if(child instanceof int[]) {

int[] child_ = (int[]) child;

if (newChild < Integer.MAX_VALUE) {

//operate on ints

final int[] child2_ = Arrays.copyOf(child_, child_.length + 1);

child2 = child2_;

if (pos < child_.length) {

System.arraycopy(child_, pos, child2_, pos + 1, child_.length - pos);

}

child2_[pos] = (int) newChild;

} else {

//new child does not fit, must upgrade to long[]

long[] child2_ = new long[child_.length + 1];

child2 = child2_;

for (int i = 0; i < pos; i++) {

child2_[i] = child_[i];

}

child2_[pos] = newChild;

for (int i = pos + 1; i < child2_.length; i++) {

child2_[i] = child_[pos - 1];

}

}

}else {

long[] child_ = (long[]) child;

//operate on ints

final long[] child2_ = Arrays.copyOf(child_, child_.length + 1);

child2 = child2_;

if (pos < child_.length) {

System.arraycopy(child_, pos, child2_, pos + 1, child_.length - pos);

}

child2_[pos] = (int) newChild;

}

//$DELAY$

return new DirNode(keys2, isLeftEdge(),isRightEdge(),false,child2);

}

@Override

public DirNode copySplitRight(BTreeKeySerializer keyser, Serializer valser, int splitPos) {

int keylen = keyser.length(keys);

Object keys2 = keyser.copyOfRange(keys,splitPos-leftEdgeInc(),keylen);

//$DELAY$

Object child2;

if(child instanceof int[]){

int[] child_ = (int[]) child;

child2 = Arrays.copyOfRange(child_, splitPos, child_.length);

}else{

long[] child_ = (long[]) child;

child2 = Arrays.copyOfRange(child_, splitPos, child_.length);

}

//$DELAY$

return new DirNode(keys2,false,isRightEdge(),false,child2);

}

@Override

public DirNode copySplitLeft(BTreeKeySerializer keyser, Serializer valser, int splitPos, long newNext) {

Object keys2 = keyser.copyOfRange(keys, 0, splitPos + 1 - leftEdgeInc());

Object child2;

boolean intArrayInstance = child instanceof int[];

if(intArrayInstance){

int[] child_ = (int[]) child;

if(newNext<Integer.MAX_VALUE) {

//all fits operate on int[]

int[] child2_ = Arrays.copyOf(child_, splitPos + 1);

child2_[splitPos] = (int) newNext;

child2 = child2_;

}else{

//does not fit, upgrade to long

long[] child2_ = new long[splitPos+1];

for(int i=0;i<=splitPos;i++){

child2_[i] = child_[i];

}

child2_[splitPos] = newNext;

child2 = child2_;

}

}else{

long[] child2_ = Arrays.copyOf((long[])child, splitPos+1);

child2_[splitPos] = (int) newNext;

child2 = child2_;

}

return new DirNode(keys2,isLeftEdge(),false,false,child2);

}

@Override

public int valSize(Serializer valueSerializer) {

throw new UnsupportedOperationException("dirnode");

}

}

public final static class LeafNode extends BNode{

final Object vals;

final long next;

LeafNode(Object keys, boolean leftEdge, boolean rightEdge, boolean tooLarge, Object vals, long next) {

super(keys,leftEdge, rightEdge, tooLarge);

this.vals = vals;

this.next = next;

if(CC.ASSERT)

checkStructure(null,null);

}

@Override public boolean isLeaf() { return true;}

@Override public Object val(int pos, Serializer valueSerializer){

return valueSerializer.valueArrayGet(vals, pos);

}

@Override public byte[] childArray() { return null;}

@Override public long child(int i){throw new UnsupportedOperationException();}

@Override public long next() {return next;}

@Override public String toString(){

String valsStr = Fun.toString(vals); //TODO use value serializer to turn this into string

return "Leaf("+leftEdgeInc()+"-"+rightEdgeInc()+"-"+"K"+Fun.toString(keys)+", V"+valsStr+", L="+next+")";

}

@Override

public void checkStructure(BTreeKeySerializer keyser, Serializer valser) {

super.checkStructure(keyser,valser);

if((next==0)!=isRightEdge()){

throw new DBException.DataCorruption("Next link inconsistent: "+this);

}

if(valser==null)

return;

int valsSize = valser.valueArraySize(vals);

if(keyser!=null && (keysLen(keyser) != valsSize+2)) {

throw new DBException.DataCorruption("Inconsistent vals size: " + this);

}

//$DELAY$

for (int i=0;i<valsSize;i++) {

Object val = valser.valueArrayGet(vals,i);

if (val == null)

throw new DBException.DataCorruption("Val is null: " + this);

}

}

@Override

public LeafNode copyAddKey(BTreeKeySerializer keyser, Serializer valser, int pos, Object newKey, long newChild, Object newValue) {

Object keys2 = keyser.putKey(keys, pos - leftEdgeInc(), newKey);

//$DELAY$

Object vals2 = valser.valueArrayPut(vals,pos-1,newValue);

//$DELAY$

return new LeafNode(keys2, isLeftEdge(), isRightEdge(), false, vals2,next);

}

@Override

public LeafNode copySplitRight(BTreeKeySerializer keyser, Serializer valser, int splitPos) {

int keylen = keyser.length(keys);

Object keys2 = keyser.copyOfRange(keys, splitPos - leftEdgeInc(), keylen);

//$DELAY$

Object vals2 = valser.valueArrayCopyOfRange(vals, splitPos, valser.valueArraySize(vals));

//$DELAY$

return new LeafNode(keys2,false, isRightEdge(), false, vals2, next);

}

@Override

public LeafNode copySplitLeft(BTreeKeySerializer keyser, Serializer valser, int splitPos, long newNext) {

int keypos =splitPos+1-leftEdgeInc();

Object keys2 = keyser.copyOfRange(keys,0,keypos);

//clone end value

Object endkey = keyser.getKey(keys2, keypos - 1);

keys2 = keyser.putKey(keys2,keypos,endkey);

//$DELAY$

Object vals2 = valser.valueArrayCopyOfRange(vals, 0, splitPos);

//$DELAY$

//TODO check high/low keys overlap

return new LeafNode(keys2, isLeftEdge(), false, false, vals2, newNext);

}

@Override

public int valSize(Serializer valueSerializer) {

return valueSerializer.valueArraySize(vals);

}

@Override

public int childArrayLength() {

return -1;

}

public LeafNode copyChangeValue(Serializer valser, int pos, Object value) {

Object vals2 = valser.valueArrayUpdateVal(vals, pos - 1, value);

//$DELAY$

return new LeafNode(keys, isLeftEdge(), isRightEdge(), false, vals2, next);

}

public LeafNode copyRemoveKey(BTreeKeySerializer keyser, Serializer valser, int pos) {

int keyPos = pos -leftEdgeInc();

Object keys2 = keyser.deleteKey(keys,keyPos);

//$DELAY$

Object vals2 = valser.valueArrayDeleteValue(vals,pos);

//$DELAY$

return new LeafNode(keys2, isLeftEdge(), isRightEdge(), false, vals2, next);

}

public LeafNode copyClear(BTreeKeySerializer keyser, Serializer valser) {

Object[] keys2 = new Object[2-leftEdgeInc()-rightEdgeInc()];

if(!isLeftEdge())

keys2[0] = key(keyser,0);

//$DELAY$

if(!isRightEdge())

keys2[1-leftEdgeInc()] = highKey(keyser);

//$DELAY$

return new LeafNode (keyser.arrayToKeys(keys2), isLeftEdge(), isRightEdge(), false, valser.valueArrayEmpty(), next);

}

}

protected final Serializer<BNode> nodeSerializer;

protected static final class NodeSerializer<A,B> extends Serializer<BNode>{

protected static final int LEAF_MASK = 1<<15;

protected static final int LEFT_SHIFT = 14;

protected static final int LEFT_MASK = 1<< LEFT_SHIFT;

protected static final int RIGHT_SHIFT = 13;

protected static final int RIGHT_MASK = 1<< RIGHT_SHIFT;

protected static final int SIZE_MASK = RIGHT_MASK - 1;

protected final boolean hasValues;

protected final boolean valsOutsideNodes;

protected final BTreeKeySerializer keySerializer;

protected final Serializer<Object> valueSerializer;

protected final int numberOfNodeMetas;

public NodeSerializer(boolean valsOutsideNodes, BTreeKeySerializer keySerializer, Serializer valueSerializer, int numberOfNodeMetas) {

if(keySerializer==null)

throw new NullPointerException("keySerializer not set");

this.hasValues = valueSerializer!=null;

this.valsOutsideNodes = valsOutsideNodes;

this.keySerializer = keySerializer;

this.valueSerializer = (Serializer) (hasValues?

(valsOutsideNodes? new ValRefSerializer() : valueSerializer):

Serializer.BOOLEAN);

this.numberOfNodeMetas = numberOfNodeMetas;

}

@Override

public void serialize(DataOutput out, BNode value) throws IOException {

final boolean isLeaf = value.isLeaf();

//check node integrity in paranoid mode

if(CC.ASSERT){

value.checkStructure(keySerializer,valueSerializer);

}

//$DELAY$

final int header =

(isLeaf ? LEAF_MASK : 0) |

(value.isLeftEdge() ? LEFT_MASK : 0) |

(value.isRightEdge() ? RIGHT_MASK : 0) |

value.keysLen(keySerializer);

out.writeShort(header);

//$DELAY$

//write node metas, right now this is ignored, but in future it could be used for counted btrees or aggregations

for(int i=0;i<numberOfNodeMetas;i++){

DataIO.packLong(out,0);

}

//$DELAY$

//longs go first, so it is possible to reconstruct tree without serializer

if(isLeaf){

DataIO.packLong(out, ((LeafNode) value).next);

}else{

serializeChildArray(out, value.childArray());

}

if(keySerializer.length(value.keys)>0)

keySerializer.serialize(out,value.keys);

//$DELAY$

if(isLeaf){

valueSerializer.valueArraySerialize(out,((LeafNode)value).vals);

}

}

protected void serializeChildArray(DataOutput out, Object childArray) throws IOException {

if(childArray instanceof int[]){

int[] cc = (int[]) childArray;

DataIO.packLong(out, (((long)cc[0]) << 1) | 1L); //pack first value mixed with int flag

for(int i=1;i<cc.length;i++){

DataIO.packInt(out,cc[i]);

}

}else{

long[] cc = (long[]) childArray;

DataIO.packLong(out, (cc[0]<<1) | 0); //pack first value mixed with int flag

for(int i=1;i<cc.length;i++) {

DataIO.packLong(out, cc[i]);

}

}

}

@Override

public BNode deserialize(DataInput in, int available) throws IOException {

final int header = in.readUnsignedShort();

final int size = header & SIZE_MASK;

//read node metas, right now this is ignored, but in future it could be used for counted btrees or aggregations

for(int i=0;i<numberOfNodeMetas;i++){

DataIO.unpackLong(in);

}

//$DELAY$

//first bite indicates leaf

final boolean isLeaf = ((header& LEAF_MASK) != 0);

final int left = (header& LEFT_MASK) >>LEFT_SHIFT;

final int right = (header& RIGHT_MASK) >>RIGHT_SHIFT;

DataIO.DataInputInternal in2 = (DataIO.DataInputInternal) in; //TODO fallback option if cast fails

BNode node;

if(isLeaf){

node = deserializeLeaf(in2, size, left, right);

}else{

node = deserializeDir(in2, size, left, right);

}

//$DELAY$

if(CC.ASSERT){

node.checkStructure(keySerializer,valueSerializer);

}

return node;

}

private BNode deserializeDir(final DataIO.DataInputInternal in, final int size, final int left, final int right) throws IOException {

Object child;

long firstChild = in.unpackLong();

if((firstChild&1) == 0){

//deserialize as long[]

long[] child_ = new long[size];

child = child_;

child_[0] = firstChild>>>1;

in.unpackLongArray(child_,1,size);

}else{

//deserialize as long[]

int[] child_ = new int[size];

child = child_;

child_[0] = (int) (firstChild>>>1);

in.unpackIntArray(child_,1,size);

}

int keysize = size - left- right;

//$DELAY$

final Object keys = keysize==0?

keySerializer.emptyKeys():

keySerializer.deserialize(in, keysize);

//$DELAY$

return new DirNode(keys, left!=0, right!=0, false ,child);

}

private BNode deserializeLeaf(final DataIO.DataInputInternal in, final int size, final int left, final int right) throws IOException {

final long next = in.unpackLong();

int keysize = size - left- right;

//$DELAY$

final Object keys = keysize==0?

keySerializer.emptyKeys():

keySerializer.deserialize(in, keysize);

//$DELAY$

Object vals = valueSerializer.valueArrayDeserialize(in,size-2);

return new LeafNode(keys, left!=0, right!=0, false , vals, next);

}

@Override

public boolean isTrusted() {

return keySerializer.isTrusted() && valueSerializer.isTrusted();

}

}

/** Constructor used to create new BTreeMap.

public BTreeMap(

Engine engine,

boolean closeEngine,

long rootRecidRef,

int maxNodeSize,

boolean valsOutsideNodes,

long counterRecid,

BTreeKeySerializer keySerializer,

final Serializer<V> valueSerializer,

int numberOfNodeMetas

) {

this.closeEngine = closeEngine;

if(maxNodeSize%2!=0)

throw new IllegalArgumentException("maxNodeSize must be dividable by 2");

if(maxNodeSize<6)

throw new IllegalArgumentException("maxNodeSize too low");

if((maxNodeSize& NodeSerializer.SIZE_MASK) !=maxNodeSize)

throw new IllegalArgumentException("maxNodeSize too high");

if(rootRecidRef<=0||counterRecid<0 || numberOfNodeMetas<0)

throw new IllegalArgumentException();

if(keySerializer==null)

throw new NullPointerException();

this.rootRecidRef = rootRecidRef;

this.hasValues = valueSerializer!=null;

this.valsOutsideNodes = valsOutsideNodes;

this.engine = engine;

this.maxNodeSize = maxNodeSize;

this.numberOfNodeMetas = numberOfNodeMetas;

this.keySerializer = keySerializer;

this.valueSerializer = valueSerializer!=null? valueSerializer: (Serializer<V>) Serializer.BOOLEAN;

entrySet = new EntrySet(this, this.valueSerializer);

this.nodeSerializer = new NodeSerializer(valsOutsideNodes,keySerializer,valueSerializer,numberOfNodeMetas);

this.keySet = new KeySet(this, hasValues);

//$DELAY$

if(counterRecid!=0){

this.counter = new Atomic.Long(engine,counterRecid);

Bind.size(this,counter);

}else{

this.counter = null;

}

//load left edge refs

ArrayList leftEdges2 = new ArrayList<Long>();

long r = engine.get(rootRecidRef,Serializer.RECID);

for(;;){

if(CC.ASSERT && r<=0)

throw new DBException.DataCorruption("wrong recid");

//$DELAY$

BNode n= engine.get(r,nodeSerializer);

leftEdges2.add(r);

if(n.isLeaf())

break;

r = n.child(0);

}

//$DELAY$

Collections.reverse(leftEdges2);

leftEdges = Collections.synchronizedList(leftEdges2);

}

/* creates empty root node and returns recid of its reference*/

static protected long createRootRef(Engine engine, BTreeKeySerializer keySer, Serializer valueSer, int numberOfNodeMetas){

Object emptyArray = valueSer!=null?

valueSer.valueArrayEmpty():

Serializer.BOOLEAN.valueArrayEmpty();

final LeafNode emptyRoot = new LeafNode(keySer.emptyKeys(), true,true, false,emptyArray, 0);

//empty root is serializer simpler way, so we can use dummy values

long rootRecidVal = engine.put(emptyRoot, new NodeSerializer(false,keySer, valueSer, numberOfNodeMetas));

return engine.put(rootRecidVal,Serializer.RECID);

}

@Override

public V get(Object key){

return (V) get(key, true);

}

protected Object get(Object key, boolean expandValue) {

if(key==null) throw new NullPointerException();

K v = (K) key;

long current = engine.get(rootRecidRef, Serializer.RECID); //get root

//$DELAY$

BNode A = engine.get(current, nodeSerializer);

//dive until leaf

while(!A.isLeaf()){

//$DELAY$

current = nextDir((DirNode) A, v);

//$DELAY$

A = engine.get(current, nodeSerializer);

}

for(;;) {

int pos = keySerializer.findChildren2(A, key);

//$DELAY$

if (pos > 0 && pos != A.keysLen(keySerializer) - 1) {

//found

Object val = A.val(pos - 1,valueSerializer);

//$DELAY$

if(expandValue)

val = valExpand(val);

return val;

} else if (pos <= 0 && -pos - 1 != A.keysLen(keySerializer) - 1) {

//$DELAY$

//not found

return null;

} else {

//move to next link

current = A.next();

//$DELAY$

if (current == 0) {

return null;

}

A = engine.get(current, nodeSerializer);

}

}

}