/**

public abstract void serialize(DataOutput out, KEYS keys) throws IOException;

/**

public abstract KEYS deserialize(DataInput in, int nodeSize) throws IOException;

public abstract int compare(KEYS keys, int pos1, int pos2);

public abstract int compare(KEYS keys, int pos, KEY key);

public boolean compareIsSmaller(KEYS keys, int pos, KEY key) {

//TODO override in Strings and other implementations

return compare(keys,pos,key)<0;

}

public abstract KEY getKey(KEYS keys, int pos);

public static final BTreeKeySerializer<Object, Object[]> BASIC = new BTreeKeySerializer.BasicKeySerializer(Serializer.BASIC, Fun.COMPARATOR);

public abstract Comparator<?> comparator();

public abstract KEYS emptyKeys();

public abstract int length(KEYS keys);

/** expand keys array by one and put {@code newKey} at position {@code pos}

public abstract KEYS putKey(KEYS keys, int pos, KEY newKey);

public abstract KEYS copyOfRange(KEYS keys, int from, int to);

public abstract KEYS deleteKey(KEYS keys, int pos);

/**

public int findChildren(final BTreeMap.BNode node, final Object key) {

KEYS keys = (KEYS) node.keys;

int keylen = length(keys);

int left = 0;

int right = keylen;

int middle;

//$DELAY$

// binary search

for(;;) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keylen)

return middle+node.leftEdgeInc(); //null is positive infinitive

if (compareIsSmaller(keys,middle, (KEY) key)) {

left = middle + 1;

} else {

right = middle;

}

if (left >= right) {

return right+node.leftEdgeInc();

}

}

}

public int findChildren2(final BTreeMap.BNode node, final Object key) {

KEYS keys = (KEYS) node.keys;

int keylen = length(keys);

int left = 0;

int right = keylen;

int comp;

int middle;

//$DELAY$

// binary search

while (true) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keylen)

return -1-(middle+node.leftEdgeInc()); //null is positive infinitive

comp = compare(keys, middle, (KEY) key);

if(comp==0){

//try one before last, in some cases it might be duplicate of last

if(!node.isRightEdge() && middle==keylen-1 && middle>0

&& compare(keys,middle-1,(KEY)key)==0){

middle--;

}

return middle+node.leftEdgeInc();

} else if ( comp< 0) {

left = middle +1;

} else {

right = middle;

}

if (left >= right) {

return -1-(right+node.leftEdgeInc());

}

}

}

public abstract KEYS arrayToKeys(Object[] keys);

public Object[] keysToArray(KEYS keys) {

//$DELAY$

Object[] ret = new Object[length(keys)];

for (int i = 0; i <ret.length; i++) {

ret[i] = getKey(keys,i);

}

return ret;

}

public boolean isTrusted() {

return false;

}

/**

public static final class BasicKeySerializer extends BTreeKeySerializer<Object, Object[]> implements Serializable {

private static final long serialVersionUID = 1654710710946309279L;

protected final Serializer serializer;

protected final Comparator comparator;

public BasicKeySerializer(Serializer serializer, Comparator comparator) {

if(serializer == null || comparator == null)

throw new NullPointerException();

this.serializer = serializer;

this.comparator = comparator;

}

/** used for deserialization*/

BasicKeySerializer(SerializerBase serializerBase, DataInput is, SerializerBase.FastArrayList<Object> objectStack) throws IOException {

objectStack.add(this);

serializer = (Serializer) serializerBase.deserialize(is,objectStack);

comparator = (Comparator) serializerBase.deserialize(is,objectStack);

if(serializer == null || comparator == null)

throw new NullPointerException();

}

@Override

public void serialize(DataOutput out, Object[] keys) throws IOException {

for(Object o:keys){

//$DELAY$

serializer.serialize(out, o);

}

}

@Override

public Object[] deserialize(DataInput in, int nodeSize) throws IOException {

//$DELAY$

Object[] keys = new Object[nodeSize];

for(int i=0;i<keys.length;i++) {

//$DELAY$

keys[i] = serializer.deserialize(in, -1);

}

return keys;

}

@Override

public int compare(Object[] keys, int pos1, int pos2) {

return comparator.compare(keys[pos1], keys[pos2]);

}

@Override

public int compare(Object[] keys, int pos, Object key) {

return comparator.compare(keys[pos],key);

}

@Override

public Object getKey(Object[] keys, int pos) {

return keys[pos];

}

@Override

public Comparator<?> comparator() {

return comparator;

}

@Override

public Object[] emptyKeys() {

return new Object[0];

}

@Override

public int length(Object[] keys) {

return keys.length;

}

@Override

public Object[] putKey(Object[] keys, int pos, Object newKey) {

return BTreeMap.arrayPut(keys, pos, newKey);

}

@Override

public Object[] arrayToKeys(Object[] keys) {

return keys;

}

@Override

public Object[] copyOfRange(Object[] keys, int from, int to) {

return Arrays.copyOfRange(keys,from,to);

}

@Override

public Object[] deleteKey(Object[] keys, int pos) {

Object[] keys2 = new Object[keys.length-1];

System.arraycopy(keys,0,keys2, 0, pos);

//$DELAY$

System.arraycopy(keys, pos+1, keys2, pos, keys2.length-pos);

//$DELAY$

return keys2;

}

}

/**

public static final BTreeKeySerializer<Long,long[]> LONG = new BTreeKeySerializer<Long,long[]>() {

@Override

public void serialize(DataOutput out, long[] keys) throws IOException {

DataIO.DataOutputByteArray out2 = (DataIO.DataOutputByteArray) out; //TODO fallback option if cast fails

long prev = keys[0];

out2.packLong(prev);

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

long curr = keys[i];

//$DELAY$

out2.packLong(curr-prev);

prev = curr;

}

}

@Override

public long[] deserialize(DataInput in, int nodeSize) throws IOException {

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

return in2.unpackLongArrayDeltaCompression(nodeSize);

}

@Override

public int compare(long[] keys, int pos1, int pos2) {

return Fun.compareLong(keys[pos1], keys[pos2]);

}

@Override

public int compare(long[] keys, int pos, Long preDigestedKey) {

return Fun.compareLong(keys[pos], preDigestedKey);

}

@Override

public boolean compareIsSmaller(long[] keys, int pos, Long key) {

return keys[pos]<key;

}

@Override

public Long getKey(long[] keys, int pos) {

return new Long(keys[pos]);

}

@Override

public Comparator<?> comparator() {

return Fun.COMPARATOR;

}

@Override

public long[] emptyKeys() {

return new long[0];

}

@Override

public int length(long[] keys) {

return keys.length;

}

@Override

public long[] putKey(long[] keys, int pos, Long newKey) {

final long[] ret = Arrays.copyOf(keys,keys.length+1);

if(pos<keys.length){

System.arraycopy(keys,pos,ret,pos+1,keys.length-pos);

}

ret[pos] = newKey;

return ret;

}

@Override

public long[] copyOfRange(long[] keys, int from, int to) {

return Arrays.copyOfRange(keys,from,to);

}

@Override

public long[] arrayToKeys(Object[] keys) {

long[] ret = new long[keys.length];

for(int i=keys.length-1;i>=0;i--) {

//$DELAY$

ret[i] = (Long) keys[i];

}

return ret;

}

@Override

public long[] deleteKey(long[] keys, int pos) {

long[] keys2 = new long[keys.length-1];

System.arraycopy(keys,0,keys2, 0, pos);

//$DELAY$

System.arraycopy(keys, pos+1, keys2, pos, keys2.length-pos);

//$DELAY$

return keys2;

}

@Override

public final int findChildren(final BTreeMap.BNode node, final Object key) {

long[] keys = (long[]) node.keys;

long key2 = (Long)key;

int left = 0;

int right = keys.length;

int middle;

//$DELAY$

// binary search

for(;;) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keys.length)

return middle+node.leftEdgeInc(); //null is positive infinitive

if (keys[middle]<key2) {

left = middle + 1;

} else {

right = middle;

}

if (left >= right) {

return right+node.leftEdgeInc();

}

}

}

@Override

public final int findChildren2(final BTreeMap.BNode node, final Object key) {

long[] keys = (long[]) node.keys;

long key2 = (Long)key;

int left = 0;

int right = keys.length;

int middle;

//$DELAY$

// binary search

while (true) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keys.length)

return -1-(middle+node.leftEdgeInc()); //null is positive infinitive

if(keys[middle]==key2){

//try one before last, in some cases it might be duplicate of last

if(!node.isRightEdge() && middle==keys.length-1 && middle>0

&& keys[middle-1]==key2){

middle--;

}

return middle+node.leftEdgeInc();

} else if ( keys[middle]<key2) {

left = middle +1;

} else {

right = middle;

}

if (left >= right) {

return -1-(right+node.leftEdgeInc());

}

}

}

};

/**

public static final BTreeKeySerializer ZERO_OR_POSITIVE_LONG = LONG;

/**

public static final BTreeKeySerializer<Integer,int[]> INTEGER = new BTreeKeySerializer<Integer,int[]>() {

@Override

public void serialize(DataOutput out, int[] keys) throws IOException {

int prev = keys[0];

DataIO.packIntBigger(out, prev);

//$DELAY$

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

int curr = keys[i];

//$DELAY$

DataIO.packInt(out, curr - prev);

prev = curr;

}

}

@Override

public int[] deserialize(DataInput in, int nodeSize) throws IOException {

int[] ret = new int[nodeSize];

int prev = 0 ;

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

//$DELAY$

prev += DataIO.unpackInt(in);

ret[i] = prev;

}

return ret;

}

@Override

public int compare(int[] keys, int pos1, int pos2) {

return Fun.compareInt(keys[pos1], keys[pos2]);

}

@Override

public int compare(int[] keys, int pos, Integer preDigestedKey) {

return Fun.compareInt(keys[pos], preDigestedKey);

}

@Override

public boolean compareIsSmaller(int[] keys, int pos, Integer key) {

return keys[pos]<key;

}

@Override

public Integer getKey(int[] keys, int pos) {

return new Integer(keys[pos]);

}

@Override

public Comparator<?> comparator() {

return Fun.COMPARATOR;

}

@Override

public int[] emptyKeys() {

return new int[0];

}

@Override

public int length(int[] keys) {

return keys.length;

}

@Override

public int[] putKey(int[] keys, int pos, Integer newKey) {

final int[] ret = Arrays.copyOf(keys,keys.length+1);

//$DELAY$

if(pos<keys.length){

System.arraycopy(keys,pos,ret,pos+1,keys.length-pos);

}

ret[pos] = newKey;

return ret;

}

@Override

public int[] copyOfRange(int[] keys, int from, int to) {

return Arrays.copyOfRange(keys,from,to);

}

@Override

public int[] arrayToKeys(Object[] keys) {

int[] ret = new int[keys.length];

for(int i=keys.length-1;i>=0;i--)

//$DELAY$

ret[i] = (Integer)keys[i];

return ret;

}

@Override

public int[] deleteKey(int[] keys, int pos) {

int[] keys2 = new int[keys.length-1];

System.arraycopy(keys,0,keys2, 0, pos);

//$DELAY$

System.arraycopy(keys, pos+1, keys2, pos, keys2.length-pos);

return keys2;

}

@Override

public final int findChildren(final BTreeMap.BNode node, final Object key) {

int[] keys = (int[]) node.keys;

int key2 = (Integer)key;

int left = 0;

int right = keys.length;

int middle;

//$DELAY$

// binary search

for(;;) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keys.length)

return middle+node.leftEdgeInc(); //null is positive infinitive

if (keys[middle]<key2) {

left = middle + 1;

} else {

right = middle;

}

if (left >= right) {

return right+node.leftEdgeInc();

}

}

}

@Override

public final int findChildren2(final BTreeMap.BNode node, final Object key) {

int[] keys = (int[]) node.keys;

int key2 = (Integer)key;

int left = 0;

int right = keys.length;

int middle;

//$DELAY$

// binary search

while (true) {

//$DELAY$

middle = (left + right) / 2;

if(middle==keys.length)

return -1-(middle+node.leftEdgeInc()); //null is positive infinitive

if(keys[middle]==key2){

//try one before last, in some cases it might be duplicate of last

if(!node.isRightEdge() && middle==keys.length-1 && middle>0

&& keys[middle-1]==key2){

middle--;

}

return middle+node.leftEdgeInc();

} else if ( keys[middle]<key2) {

left = middle +1;

} else {

right = middle;

}

if (left >= right) {

return -1-(right+node.leftEdgeInc());

}

}

}

};

/**

public static final BTreeKeySerializer ZERO_OR_POSITIVE_INT = INTEGER;

public static final BTreeKeySerializer ARRAY2 = new ArrayKeySerializer(

new Comparator[]{Fun.COMPARATOR,Fun.COMPARATOR},

new Serializer[]{Serializer.BASIC, Serializer.BASIC}

);

public static final BTreeKeySerializer ARRAY3 = new ArrayKeySerializer(

new Comparator[]{Fun.COMPARATOR,Fun.COMPARATOR,Fun.COMPARATOR},

new Serializer[]{Serializer.BASIC, Serializer.BASIC, Serializer.BASIC}

);

public static final BTreeKeySerializer ARRAY4 = new ArrayKeySerializer(

new Comparator[]{Fun.COMPARATOR,Fun.COMPARATOR,Fun.COMPARATOR,Fun.COMPARATOR},

new Serializer[]{Serializer.BASIC, Serializer.BASIC, Serializer.BASIC, Serializer.BASIC}

);

public final static class ArrayKeySerializer extends BTreeKeySerializer<Object[],Object[]> implements Serializable{

private static final long serialVersionUID = 998929894238939892L;

protected final int tsize;

protected final Comparator[] comparators;

protected final Serializer[] serializers;

protected final Comparator comparator;

public ArrayKeySerializer(Comparator[] comparators, Serializer[] serializers) {

if(comparators.length!=serializers.length){

throw new IllegalArgumentException("array sizes do not match");

}

this.tsize = comparators.length;

this.comparators = comparators;

this.serializers = serializers;

this.comparator = new Fun.ArrayComparator(comparators);

}

/* used for deserialization, extra is to avoid argument collision */

public ArrayKeySerializer(SerializerBase serializerBase, DataInput is,

SerializerBase.FastArrayList<Object> objectStack) throws IOException {

objectStack.add(this);

tsize = DataIO.unpackInt(is);

comparators = new Comparator[tsize];

serializers = new Serializer[tsize];

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

comparators[i] = (Comparator) serializerBase.deserialize(is,objectStack);

serializers[i] = (Serializer) serializerBase.deserialize(is,objectStack);

}

this.comparator = new Fun.ArrayComparator(comparators);

}

@Override

public void serialize(DataOutput out, Object[] keys) throws IOException {

int[] counts = new int[tsize-1];

//$DELAY$

for(int i=0;i<keys.length;i+=tsize){

for(int j=0;j<tsize-1;j++){

//$DELAY$

if(counts[j]==0){

Object orig = keys[i+j];

serializers[j].serialize(out,orig);

counts[j]=1;

while(i+j+counts[j]*tsize<keys.length &&

comparators[j].compare(orig,keys[i+j+counts[j]*tsize])==0){

counts[j]++;

}

DataIO.packInt(out,counts[j]);

}

}

//write last value from tuple

serializers[serializers.length-1].serialize(out,keys[i+tsize-1]);

//decrement all

//$DELAY$

for(int j=counts.length-1;j>=0;j--){

counts[j]--;

}

}

}

@Override

public Object[] deserialize(DataInput in, int nodeSize) throws IOException {

Object[] ret = new Object[nodeSize*tsize];

Object[] curr = new Object[tsize];

int[] counts = new int[tsize-1];

//$DELAY$

for(int i=0;i<ret.length;i+=tsize){

for(int j=0;j<tsize-1;j++){

if(counts[j]==0){

//$DELAY$

curr[j] = serializers[j].deserialize(in,-1);

counts[j] = DataIO.unpackInt(in);

}

}

curr[tsize-1] = serializers[tsize-1].deserialize(in,-1);

System.arraycopy(curr,0,ret,i,tsize);

//$DELAY$

for(int j=counts.length-1;j>=0;j--){

counts[j]--;

}

}

if(CC.ASSERT){

for(int j:counts){

if(j!=0)

throw new DBException.DataCorruption("inconsistent counts");

}

}

return ret;

}

@Override

public int compare(Object[] keys, int pos1, int pos2) {

pos1 *=tsize;

pos2 *=tsize;

int res;

//$DELAY$

for(Comparator c:comparators){

//$DELAY$

res = c.compare(keys[pos1++],keys[pos2++]);

if(res!=0) {

return res;

}

}

return 0;

}

@Override

public int compare(Object[] keys, int pos, Object[] tuple) {

pos*=tsize;

int len = Math.min(tuple.length, tsize);

int r;

//$DELAY$

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

//$DELAY$

r = comparators[i].compare(keys[pos++],tuple[i]);

if(r!=0)

return r;

}

return Fun.compareInt(tsize, tuple.length);

}

@Override

public Object[] getKey(Object[] keys, int pos) {

pos*=tsize;

return Arrays.copyOfRange(keys,pos,pos+tsize);

}

@Override

public Comparator<?> comparator() {

return comparator;

}

@Override

public Object[] emptyKeys() {

return new Object[0];

}

@Override

public int length(Object[] objects) {

return objects.length/tsize;

}

@Override

public Object[] putKey(Object[] keys, int pos, Object[] newKey) {

if(CC.ASSERT && newKey.length!=tsize)

throw new DBException.DataCorruption("inconsistent size");

pos*=tsize;

Object[] ret = new Object[keys.length+tsize];

System.arraycopy(keys, 0, ret, 0, pos);

//$DELAY$

System.arraycopy(newKey,0,ret,pos,tsize);

//$DELAY$

System.arraycopy(keys,pos,ret,pos+tsize,keys.length-pos);

return ret;

}

@Override

public Object[] arrayToKeys(Object[] keys) {

Object[] ret = new Object[keys.length*tsize];

int pos=0;

//$DELAY$

for(Object o:keys){

if(CC.ASSERT && ((Object[])o).length!=tsize)

throw new DBException.DataCorruption("keys have wrong size");

System.arraycopy(o,0,ret,pos,tsize);

//$DELAY$

pos+=tsize;

}

return ret;

}

@Override

public Object[] copyOfRange(Object[] keys, int from, int to) {

from*=tsize;

to*=tsize;

return Arrays.copyOfRange(keys,from,to);

}

@Override

public Object[] deleteKey(Object[] keys, int pos) {

pos*=tsize;

Object[] ret = new Object[keys.length-tsize];

System.arraycopy(keys,0,ret,0,pos);

//$DELAY$

System.arraycopy(keys,pos+tsize,ret,pos,ret.length-pos);

return ret;

}

@Override

public boolean equals(Object o) {

if (this == o) return true;

if (o == null || getClass() != o.getClass()) return false;

ArrayKeySerializer that = (ArrayKeySerializer) o;

//$DELAY$

if (tsize != that.tsize) return false;

if (!Arrays.equals(comparators, that.comparators)) return false;

//$DELAY$

return Arrays.equals(serializers, that.serializers);

}

@Override

public int hashCode() {

int result = tsize;

result = 31 * result + Arrays.hashCode(comparators);

result = 31 * result + Arrays.hashCode(serializers);

return result;

}

}

public static final BTreeKeySerializer<java.util.UUID,long[]> UUID = new BTreeKeySerializer<java.util.UUID,long[]>() {

@Override

public void serialize(DataOutput out, long[] longs) throws IOException {

//$DELAY$

for(long l:longs){

out.writeLong(l);

}

}

@Override

public long[] deserialize(DataInput in, int nodeSize) throws IOException {

long[] ret= new long[nodeSize<<1];

//$DELAY$

for(int i=0;i<ret.length;i++){

ret[i]=in.readLong();

}

return ret;

}

@Override

public int compare(long[] longs, int pos1, int pos2) {

pos1<<=1;

pos2<<=1;

//$DELAY$

int r = Fun.compareLong(longs[pos1++],longs[pos2++]);

if(r!=0)

return r;

return Fun.compareLong(longs[pos1],longs[pos2]);

}

@Override

public int compare(long[] longs, int pos, java.util.UUID uuid) {

pos<<=1;

//$DELAY$

int r = Fun.compareLong(longs[pos++],uuid.getMostSignificantBits());

if(r!=0)

return r;

return Fun.compareLong(longs[pos],uuid.getLeastSignificantBits());

}

@Override

public UUID getKey(long[] longs, int pos) {

pos<<=1;

return new UUID(longs[pos++],longs[pos]);

}

@Override

public Comparator<?> comparator() {

return Fun.COMPARATOR;

}

@Override

public long[] emptyKeys() {

return new long[0];

}

@Override

public int length(long[] longs) {

return longs.length/2;

}

@Override

public long[] putKey(long[] keys, int pos, UUID newKey) {

pos <<= 1; //*2

long[] ret = new long[keys.length+2];

System.arraycopy(keys, 0, ret, 0, pos);

//$DELAY$

ret[pos++] = newKey.getMostSignificantBits();

ret[pos++] = newKey.getLeastSignificantBits();

System.arraycopy(keys,pos-2,ret,pos,ret.length-pos);

return ret;

}

@Override

public long[] arrayToKeys(Object[] keys) {

long[] ret = new long[keys.length<<1]; //*2

int i=0;

//$DELAY$

for(Object o:keys){

java.util.UUID u = (java.util.UUID) o;

ret[i++]=u.getMostSignificantBits();

ret[i++]=u.getLeastSignificantBits();

}

return ret;

}

@Override

public long[] copyOfRange(long[] longs, int from, int to) {

return Arrays.copyOfRange(longs,from<<1,to<<1);

}

@Override

public long[] deleteKey(long[] keys, int pos) {

pos <<= 1; //*2

long[] ret = new long[keys.length-2];

System.arraycopy(keys,0,ret,0,pos);

//$DELAY$

System.arraycopy(keys,pos+2,ret,pos,ret.length-pos);

return ret;

}

};

public interface StringArrayKeys {

int commonPrefixLen();

int length();

int[] getOffset();

BTreeKeySerializer.StringArrayKeys deleteKey(int pos);

BTreeKeySerializer.StringArrayKeys copyOfRange(int from, int to);

BTreeKeySerializer.StringArrayKeys putKey(int pos, String newKey);

int compare(int pos1, String string);

int compare(int pos1, int pos2);

String getKeyString(int pos);

boolean hasUnicodeChars();

void serialize(DataOutput out, int prefixLen) throws IOException;

}

//TODO right now byte[] contains 7 bit characters, but it should be expandable to 8bit.

public static final class ByteArrayKeys implements StringArrayKeys {

final int[] offset;

final byte[] array;

ByteArrayKeys(int[] offset, byte[] array) {

this.offset = offset;

this.array = array;

if(CC.ASSERT && ! (array.length==0 || array.length == offset[offset.length-1]))

throw new DBException.DataCorruption("inconsistent array size");

}

ByteArrayKeys(DataInput in, int[] offsets, int prefixLen) throws IOException {

this.offset = offsets;

array = new byte[offsets[offsets.length-1]];

in.readFully(array, 0, prefixLen);

for(int i=0; i<offsets.length-1;i++){

System.arraycopy(array,0,array,offsets[i],prefixLen);

}

//$DELAY$

//read suffixes