package org.simantics.datatypes.utils;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.simantics.databoard.Bindings;
import org.simantics.databoard.accessor.reference.ChildReference;
import org.simantics.databoard.binding.Binding;
import org.simantics.databoard.binding.error.BindingException;
import org.simantics.databoard.binding.error.RuntimeBindingConstructionException;
import org.simantics.databoard.binding.impl.BindingPrintContext;
import org.simantics.databoard.binding.mutable.Variant;
import org.simantics.databoard.serialization.RuntimeSerializerConstructionException;
import org.simantics.databoard.serialization.Serializer;
import org.simantics.databoard.util.IdentityPair;
import org.simantics.datatypes.DatatypeResource;
import org.simantics.db.ReadGraph;
import org.simantics.db.Resource;
import org.simantics.db.WriteGraph;
import org.simantics.db.common.primitiverequest.RelatedValue;
import org.simantics.db.common.procedure.adapter.TransientCacheListener;
import org.simantics.db.common.utils.Logger;
import org.simantics.db.exception.DatabaseException;
import org.simantics.db.service.Bytes;
import org.simantics.db.service.SerialisationSupport;
import org.simantics.db.service.XSupport;
import org.simantics.layer0.Layer0;
import org.simantics.scl.runtime.tuple.Tuple2;
import org.simantics.utils.datastructures.Pair;

import gnu.trove.list.array.TByteArrayList;
import gnu.trove.map.hash.TObjectIntHashMap;

/*
 * Implementation from http://www.geeksforgeeks.org/b-tree-set-3delete/
 * 
 */

interface BTreeContentManager {
	
	BTreeContentBean getContentBean(ReadGraph graph, Resource node) throws DatabaseException;
	void setContentBean(WriteGraph graph, Resource node, BTreeContentBean bean) throws DatabaseException;
	
}

class PossibleVariantInputStream extends ByteArrayInputStream {
	
	PossibleVariantInputStream(byte[] data, int offset) {
		super(data, offset, data.length-offset);
	}
	
	public int getOffset() {
		return pos;
	}
	
}

class BTreeContentBinding extends Binding {

	private final SerialisationSupport ss;
	private final XSupport xs;
	
	public BTreeContentBinding(SerialisationSupport ss, XSupport xs) {
		this.ss = ss;
		this.xs = xs;
	}
	
	private final Serializer serializer = new Serializer() {

		@Override
		public void serialize(DataOutput out, TObjectIntHashMap<Object> identities, Object obj) throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public void serialize(DataOutput out, Object obj) throws IOException {
			byte[] data = serialize(obj);
			out.write(data);
		}

		@Override
		public Object deserialize(DataInput in, List<Object> identities)
				throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public Object deserialize(DataInput in) throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public void deserializeTo(DataInput in, List<Object> identities,
				Object dst) throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public void deserializeTo(DataInput in, Object dst) throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public void skip(DataInput in, List<Object> identities)
				throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public void skip(DataInput in) throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public Integer getConstantSize() {
			throw new UnsupportedOperationException();
		}

		@Override
		public int getSize(Object obj, TObjectIntHashMap<Object> identities)
				throws IOException {
			throw new UnsupportedOperationException();
		}

		@Override
		public int getSize(Object obj) throws IOException {
			return serialize(obj).length;
		}

		@Override
		public int getMinSize() {
			throw new UnsupportedOperationException();
		}
		
		private void addBE(TByteArrayList bytes, int value) {
			
			bytes.add((byte) ((value >>> 24) & 0xFF));
			bytes.add((byte) ((value >>> 16) & 0xFF));
			bytes.add((byte) ((value >>> 8) & 0xFF));
			bytes.add( (byte) (value & 0xFF));

		}
		
		private void addBE(TByteArrayList bytes, long value) {
			
			bytes.add((byte) ((value >>> 56) & 0xFF));
			bytes.add((byte) ((value >>> 48) & 0xFF));
			bytes.add((byte) ((value >>> 40) & 0xFF));
			bytes.add((byte) ((value >>> 32) & 0xFF));
			bytes.add((byte) ((value >>> 24) & 0xFF));
			bytes.add((byte) ((value >>> 16) & 0xFF));
			bytes.add((byte) ((value >>> 8) & 0xFF));
			bytes.add( (byte) (value & 0xFF));

		}

		public byte[] serialize(Object obj) throws IOException {
			BTreeContentBean bean = (BTreeContentBean)obj;
			TByteArrayList bytes = new TByteArrayList();
			bytes.add(bean.leaf ? (byte)1 : (byte)0);
			addBE(bytes, bean.n);
			//addLE(bytes, (bean.key.length+1)/2);
			Binding pbv = PossibleVariant.BINDING;
			Serializer pbs = pbv.serializer();
			addBE(bytes, bean.key.length);
			for(PossibleVariant l : bean.key) {
				byte[] ser = pbs.serialize(l);
				bytes.add(ser);
			}
			addBE(bytes, bean.value.length);
			for(PossibleResource pr : bean.value) {
				if(pr.r != null) {
					bytes.add((byte)1);
					addBE(bytes, xs.convertDelayedResourceToResource(pr.r).getResourceId());
				} else {
					bytes.add((byte)0);
				}
			}
			addBE(bytes, bean.c.length);
			for(PossibleResource pr : bean.c) {
				if(pr.r != null) {
					bytes.add((byte)1);
					addBE(bytes, xs.convertDelayedResourceToResource(pr.r).getResourceId());
				} else {
					bytes.add((byte)0);
				}
			}
			return bytes.toArray();
		}
		
		int readBE4(byte[] bytes, int byteIndex) {
			int result = (int) 
	    			(((int)(bytes[byteIndex++] & 0xff))<<24) |
					(((int)(bytes[byteIndex++] & 0xff))<<16) | 
					(((int)(bytes[byteIndex++] & 0xff))<<8) | 
	    			(((int)(bytes[byteIndex++] & 0xff)));
	    		return result;
		}
		
		long readBE8(byte[] bytes, int byteIndex) {
			long result = (long) 
	    			(((long)(bytes[byteIndex++] & 0xff))<<56) |
	    			(((long)(bytes[byteIndex++] & 0xff))<<48) | 
	    			(((long)(bytes[byteIndex++] & 0xff))<<40) | 
	    			(((long)(bytes[byteIndex++] & 0xff))<<32) | 
	    			(((long)(bytes[byteIndex++] & 0xff))<<24) | 
	    			(((long)(bytes[byteIndex++] & 0xff))<<16) | 
	    			(((long)(bytes[byteIndex++] & 0xff))<<8) | 
	    			(((long)(bytes[byteIndex++] & 0xff))); 
			return result;
		}
		
		public Object deserialize(byte[] data) throws IOException {
			
			BTreeContentBean result = new BTreeContentBean();

			try {
			
				result.leaf = (data[0] == 1) ? true : false;
				int byteIndex = 1;
				result.n = readBE4(data, byteIndex);
				byteIndex += 4;
				
				Binding pbv = PossibleVariant.BINDING;
				Serializer pbs = pbv.serializer();

				// Redundant array size
				int keySize = readBE4(data, byteIndex);
				result.key = new PossibleVariant[keySize];
				byteIndex += 4;
				for(int i=0;i<keySize;i++) {
					PossibleVariantInputStream is = new PossibleVariantInputStream(data, byteIndex);
					PossibleVariant v = (PossibleVariant)pbs.deserialize(is);
					result.key[i] = v;
					byteIndex = is.getOffset();
				}
				
				// Redundant array size
				int valueSize = readBE4(data, byteIndex);
				result.value = new PossibleResource[valueSize];
				byteIndex += 4;
				for(int i=0;i<valueSize;i++) {
					boolean has = Bytes.read(data, byteIndex++) > 0;
					if(has) {
						result.value[i] = PossibleResource.read(ss, readBE8(data, byteIndex));
						byteIndex += 8;
					} else {
						result.value[i] = new PossibleResource();
					}
				}
	
				// Redundant array size
				int childSize = readBE4(data, byteIndex);
				result.c = new PossibleResource[childSize];
				byteIndex += 4;
				for(int i=0;i<childSize;i++) {
					boolean has = Bytes.read(data, byteIndex++) > 0;
					if(has) {
						result.c[i] = PossibleResource.read(ss, readBE8(data, byteIndex));
						byteIndex += 8;
					} else {
						result.c[i] = new PossibleResource();
					}
				}
	
			} catch (DatabaseException e) {
			
				e.printStackTrace();
				
			}
			
			return result;
			
		}
		
	};
	
	@Override
	public void accept(Visitor1 v, Object obj) {
		throw new UnsupportedOperationException();
	}

	@Override
	public <T> T accept(Visitor<T> v) {
		throw new UnsupportedOperationException();
	}

	@Override
	public boolean isInstance(Object obj) {
		throw new UnsupportedOperationException();
	}

	@Override
	public void readFrom(Binding srcBinding, Object src, Object dst)
			throws BindingException {
		throw new UnsupportedOperationException();
	}

	@Override
	public void assertInstaceIsValid(Object obj, Set<Object> validInstances)
			throws BindingException {
		throw new UnsupportedOperationException();
	}

	@Override
	public int deepHashValue(Object value,
			IdentityHashMap<Object, Object> hashedObjects)
			throws BindingException {
		throw new UnsupportedOperationException();
	}

	@Override
	public int deepCompare(Object o1, Object o2,
			Set<IdentityPair<Object, Object>> compareHistory)
			throws BindingException {
		throw new UnsupportedOperationException();
	}

	@Override
	protected void toString(Object value, BindingPrintContext ctx)
			throws BindingException {
		throw new UnsupportedOperationException();
	}

	@Override
	public int getComponentCount() {
		throw new UnsupportedOperationException();
	}

	@Override
	public Binding getComponentBinding(int index) {
		throw new UnsupportedOperationException();
	}

	@Override
	public Binding getComponentBinding(ChildReference path) {
		throw new UnsupportedOperationException();
	}
	
	@Override
	public Serializer serializer() throws RuntimeSerializerConstructionException {
		return serializer;
	}
	
}

final public class BTreeUtils implements BTreeContentManager {

	final private Resource tree;
	final private Resource ownerRelation;
	
	final public Binding CONTENT_BEAN_BINDING;
	final public DatatypeResource DATA;
	
	private long mod;
	private int t;
	private Map<Resource, BTreeContentBean> beans;
	private Map<Resource, Resource> owners;
	private Resource root;
	private Resource nodeType;
	private boolean cached;

	public BTreeUtils(ReadGraph graph, Resource tree, int t, Resource nodeType, Resource ownerRelation, boolean cached) throws DatabaseException {
		try {
			SerialisationSupport ss = graph.getService(SerialisationSupport.class);
			XSupport xs = graph.getService(XSupport.class);
					
			CONTENT_BEAN_BINDING = new BTreeContentBinding(ss, xs);
			DATA = DatatypeResource.getInstance(graph);
			this.tree = tree;
			this.ownerRelation = ownerRelation;
			this.nodeType = nodeType != null ? nodeType : DATA.BTreeNode;
			this.t = t;
			Long possibleMod = graph.getPossibleRelatedValue(tree, DATA.BTree_mod, Bindings.LONG);
			this.mod = possibleMod != null ? possibleMod : 0;
			this.cached = cached;
			if (cached) {
				this.beans = new HashMap<Resource, BTreeContentBean>();
				this.owners = new HashMap<Resource, Resource>();
			}
		} catch (RuntimeBindingConstructionException e) {
			Logger.defaultLogError(e);
			throw new DatabaseException(e);
		}
	}
	
	public static BTreeUtils create(WriteGraph graph, Resource ownerRelation, int t, boolean cached) throws DatabaseException {
		
		DatatypeResource DATA = DatatypeResource.getInstance(graph);
		return create(graph, DATA.BTree, DATA.BTreeNode, ownerRelation, t, cached);
		
	}

	public static BTreeUtils create(WriteGraph graph, Resource type, Resource nodeType, Resource ownerRelation, int t, boolean cached) throws DatabaseException {
		
		Layer0 L0 = Layer0.getInstance(graph);
		DatatypeResource DATA = DatatypeResource.getInstance(graph);
		Resource tree = graph.newResource();
		graph.claim(tree, L0.InstanceOf, null, type);
		
		if(ownerRelation != null)
			graph.claim(tree, DATA.BTree_HasOwnerRelation, DATA.BTree_HasOwnerRelation_Inverse, ownerRelation);
		if(nodeType != null)
			graph.claim(tree, DATA.BTree_HasNodeType, DATA.BTree_HasNodeType_Inverse, nodeType);
		
		graph.claimLiteral(tree, DATA.BTree_t, t, Bindings.INTEGER);
		
		if (!cached) {
			graph.claimLiteral(tree, DATA.BTree_mod, 0L, Bindings.LONG);
		}

		BTreeUtils utils = new BTreeUtils(graph, tree, t, nodeType, ownerRelation, cached);
		Resource node = utils.createNode(graph, utils, t); 
		utils.setRoot(graph, node);
		utils.setOwner(graph, node, tree, true);
		
		return utils; 
	}	
	
	public Resource getTree() {
		return tree;
	}

	public void insert(WriteGraph graph, Variant k, Resource v) throws DatabaseException {

		Resource r = getRoot(graph);
		insertImpl(graph, this, r, t, k, v);
		
	}

	static class BatchContentManager implements BTreeContentManager {

		final private BTreeUtils bu;
		
		final Map<Resource, BTreeContentBean> beans = new HashMap<Resource, BTreeContentBean>();
		
		public BatchContentManager(BTreeUtils bu) {
			this.bu = bu;
		}
		
		@Override
		public BTreeContentBean getContentBean(ReadGraph graph, Resource node) throws DatabaseException {
			BTreeContentBean bean = beans.get(node);
			if(bean != null) return bean;
			return bu.getContentBean(graph, node);
		}

		@Override
		public void setContentBean(WriteGraph graph, Resource node, BTreeContentBean bean) throws DatabaseException {
			beans.put(node, bean);
		}
		
		public void apply(WriteGraph graph) throws DatabaseException {
			for(Map.Entry<Resource, BTreeContentBean> entry : beans.entrySet()) {
				bu.setContentBean(graph, entry.getKey(), entry.getValue());
			}
		}
		
	}
	
	public void insertAll(WriteGraph graph, Collection<Pair<Variant, Resource>> values) throws DatabaseException {

		Resource r = getRoot(graph);
		BatchContentManager cm = new BatchContentManager(this);
		for(Pair<Variant, Resource> entry : values) {
			insertImpl(graph, cm, r, t, entry.first, entry.second);
		}
		cm.apply(graph);
		
	}

	public Resource search(ReadGraph graph, Variant k) throws DatabaseException {

		Resource r = getRoot(graph);
		return searchNode(graph, r, k);
		
	}
	
	// Implementation

	private int findKey(ReadGraph graph, BTreeContentManager manager, BTreeContentBean bean, Variant k) throws DatabaseException {

		int idx = 0;
		while(idx<bean.n && bean.key[idx].v.compareTo(k) < 0)
			idx++;
		return idx;

	}
	
	private void insertImpl(WriteGraph graph, BTreeContentManager manager, Resource r, int t, Variant k, Resource v) throws DatabaseException {

		BTreeContentBean rContent = manager.getContentBean(graph, r);
		if(rContent.n == 2*t-1) {
			Resource s = allocateNode(graph, manager, t);
			BTreeContentBean sContent = manager.getContentBean(graph, s);
			setRoot(graph, s);
			setOwner(graph, s, tree, true);
			sContent.leaf = false;
			sContent.n = 0;
			sContent.c[0].r = r;
			setOwner(graph, r, s, true);
			manager.setContentBean(graph, s, sContent);
			splitChild(graph, manager, t, s, 1, r);
			insertNonfull(graph, manager, t, s, k, v);
		} else {
			insertNonfull(graph, manager, t, r, k, v);
		}
		
	}

	public void remove(WriteGraph graph, Variant k) throws DatabaseException {

		Resource r = getRoot(graph);
		
		BTreeContentBean bean = getContentBean(graph, r);
		Resource removedResource = removeImpl(graph, this, t, r, bean, k);
		if (removedResource != null) {
			removeOwner(graph, removedResource, false);
			
		    // If the root node has 0 keys, make its first child as the new root
		    //  if it has a child, otherwise set root as NULL
		    if (bean.n==0)
		    {
		    	//BTreeContentBean tmp = bean;
		        if (!bean.leaf) {
			        removeOwner(graph, r, true);
		        	setRoot(graph, bean.c[0].r);
		        	setOwner(graph, bean.c[0].r, tree, true);
		        }
		 
		        // Free the old root
		        //delete tmp;
		        
		    }
		}

		
	}
	
	public Resource removeImpl(WriteGraph graph, BTreeContentManager manager, int t, Resource r, BTreeContentBean bean, Variant k) throws DatabaseException {
		
		int idx = findKey(graph, manager, bean, k);
		
		if(idx < bean.n && bean.key[idx].v.equals(k)) {
			Resource removedResource = bean.value[idx].r;
			if(bean.leaf)
				removeFromLeaf(graph, manager, r, bean, idx);
			else
				removeFromNonLeaf(graph, manager, r, bean, t, idx);
			
			return removedResource;
		} else {
			
	        if (bean.leaf) {
	        	return null;
	        }
	 
	        // The key to be removed is present in the sub-tree rooted with this node
	        // The flag indicates whether the key is present in the sub-tree rooted
	        // with the last child of this node
	        boolean flag = ( (idx==bean.n)? true : false );
	 
	        BTreeContentBean cContent = manager.getContentBean(graph, bean.c[idx].r);
	        
	        // If the child where the key is supposed to exist has less that t keys,
	        // we fill that child
	        if (cContent.n < t)
	            fill(graph, manager, r, bean, t, idx);
	 
	        // If the last child has been merged, it must have merged with the previous
	        // child and so we recurse on the (idx-1)th child. Else, we recurse on the
	        // (idx)th child which now has atleast t keys
	        if (flag && idx > bean.n)
	        	return removeImpl(graph, manager, t, bean.c[idx-1].r, manager.getContentBean(graph, bean.c[idx-1].r), k);
	        else
	        	return removeImpl(graph, manager, t, bean.c[idx].r, cContent, k);
	        
		}
		
		
	}
	
	// A function to get predecessor of keys[idx]
	Pair<Variant, Resource> getPred(ReadGraph graph, BTreeContentManager manager, BTreeContentBean bean, int idx) throws DatabaseException {
		
	    // Keep moving to the right most node until we reach a leaf
        bean = manager.getContentBean(graph, bean.c[idx].r);
	    while (!bean.leaf)
	        bean = manager.getContentBean(graph, bean.c[bean.n].r);
	 
	    // Return the last key of the leaf
	    return new Pair<Variant, Resource>(bean.key[bean.n-1].v, bean.value[bean.n-1].r);
	    
	}
	 
	Pair<Variant, Resource> getSucc(ReadGraph graph, BTreeContentManager manager, BTreeContentBean bean, int idx) throws DatabaseException {
	 
	    // Keep moving the left most node starting from C[idx+1] until we reach a leaf
        bean = manager.getContentBean(graph, bean.c[idx+1].r);
	    while (!bean.leaf)
	        bean = manager.getContentBean(graph, bean.c[0].r);

	    // Return the first key of the leaf
	    return new Pair<Variant, Resource>(bean.key[0].v, bean.value[0].r);
	    
	}
	 
	// A function to fill child C[idx] which has less than t-1 keys
	void fill(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int t, int idx) throws DatabaseException {

	    // If the previous child(C[idx-1]) has more than t-1 keys, borrow a key
	    // from that child
	    if (idx!=0) {
	        BTreeContentBean cContent = manager.getContentBean(graph, bean.c[idx-1].r);

	    	if (cContent.n>=t) {
	    		borrowFromPrev(graph, manager, r, bean, idx);
	    		return;
	    	}
	    }
	    
    	// If the next child(C[idx+1]) has more than t-1 keys, borrow a key
    	// from that child

	    if (idx!=bean.n) {
	        BTreeContentBean cContent = manager.getContentBean(graph, bean.c[idx+1].r);

	    	if (cContent.n>=t) {
	    		borrowFromNext(graph, manager, r, bean, idx);
	    		return;
	    	}
	    }

    	// Merge C[idx] with its sibling
    	// If C[idx] is the last child, merge it with with its previous sibling
    	// Otherwise merge it with its next sibling
		if (idx != bean.n)
			merge(graph, manager, r, bean, t, idx);
		else
			merge(graph, manager, r, bean, t, idx-1);
	}
	
	// A function to borrow a key from C[idx-1] and insert it
	// into C[idx]
	void borrowFromPrev(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int idx) throws DatabaseException {
	 
        BTreeContentBean childContent = manager.getContentBean(graph, bean.c[idx].r);
        BTreeContentBean siblingContent = manager.getContentBean(graph, bean.c[idx-1].r);
		
	    // The last key from C[idx-1] goes up to the parent and key[idx-1]
	    // from parent is inserted as the first key in C[idx]. Thus, the  loses
	    // sibling one key and child gains one key
	 
	    // Moving all key in C[idx] one step ahead
	    for (int i=childContent.n-1; i>=0; --i) {
	    	childContent.key[i+1].v = childContent.key[i].v;
	    	childContent.value[i+1].r = childContent.value[i].r;
	    }
	    
	    // If C[idx] is not a leaf, move all its child pointers one step ahead
	    if (!childContent.leaf) {
	        for(int i=childContent.n; i>=0; --i) {
	            childContent.c[i+1].r = childContent.c[i].r;
				setOwner(graph, childContent.c[i+1].r, bean.c[idx].r, true);
	        }
	    }

	    // Setting child's first key equal to keys[idx-1] from the current node
	    childContent.key[0].v = bean.key[idx-1].v;
	    childContent.value[0].r = bean.value[idx-1].r;
	 
	    // Moving sibling's last child as C[idx]'s first child
	    // TODO: is this correct?
	    if (!bean.leaf) {
	        Resource lastChild = siblingContent.c[siblingContent.n].r;
	        if (lastChild != null) {
		    	childContent.c[0].r = lastChild;
				setOwner(graph, childContent.c[0].r, bean.c[idx].r, true);
	        }
	    }

	    // Moving the key from the sibling to the parent
	    // This reduces the number of keys in the sibling
	    bean.key[idx-1].v = siblingContent.key[siblingContent.n-1].v;
	    bean.value[idx-1].r = siblingContent.value[siblingContent.n-1].r;

	    childContent.n += 1;
	    siblingContent.reduceSize();
	    //siblingContent.n -= 1;

	    manager.setContentBean(graph, r, bean);
	    manager.setContentBean(graph, bean.c[idx].r, childContent);
	    manager.setContentBean(graph, bean.c[idx-1].r, siblingContent);

	}
	 
	// A function to borrow a key from the C[idx+1] and place
	// it in C[idx]
	void borrowFromNext(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int idx) throws DatabaseException {

        BTreeContentBean childContent = manager.getContentBean(graph, bean.c[idx].r);
        BTreeContentBean siblingContent = manager.getContentBean(graph, bean.c[idx+1].r);

	    // keys[idx] is inserted as the last key in C[idx]
        childContent.key[childContent.n].v = bean.key[idx].v;
        childContent.value[childContent.n].r = bean.value[idx].r;
	 
	    // Sibling's first child is inserted as the last child
	    // into C[idx]
	    if (!childContent.leaf) {
	    	childContent.c[childContent.n+1].r = siblingContent.c[0].r;
			setOwner(graph, childContent.c[childContent.n+1].r, bean.c[idx].r, true);
	    }
	 
	    //The first key from sibling is inserted into keys[idx]
	    bean.key[idx].v = siblingContent.key[0].v;
	    bean.value[idx].r = siblingContent.value[0].r;
	 
	    // Moving all keys in sibling one step behind
	    for (int i=1; i<siblingContent.n; ++i) {
	    	siblingContent.key[i-1].v = siblingContent.key[i].v;
	    	siblingContent.value[i-1].r = siblingContent.value[i].r;
	    }
	 
	    // Moving the child pointers one step behind
	    if (!siblingContent.leaf) {
	        for(int i=1; i<=siblingContent.n; ++i)
	        	siblingContent.c[i-1].r = siblingContent.c[i].r;
	    }
	 
	    // Increasing and decreasing the key count of C[idx] and C[idx+1]
	    // respectively
	    childContent.n += 1;
	    siblingContent.reduceSize();
	    //siblingContent.n -= 1;
	    
	    manager.setContentBean(graph, r, bean);
	    manager.setContentBean(graph, bean.c[idx].r, childContent);
	    manager.setContentBean(graph, bean.c[idx+1].r, siblingContent);
	 
	}
	 
	// A function to merge C[idx] with C[idx+1]
	// C[idx+1] is freed after merging
	void merge(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int t, int idx) throws DatabaseException {
		
        BTreeContentBean childContent = manager.getContentBean(graph, bean.c[idx].r);
        BTreeContentBean siblingContent = manager.getContentBean(graph, bean.c[idx+1].r);

	    // Pulling a key from the current node and inserting it into (t-1)th
	    // position of C[idx]
        childContent.key[t-1].v = bean.key[idx].v;
        childContent.value[t-1].r = bean.value[idx].r;
	 
	    // Copying the keys from C[idx+1] to C[idx] at the end
	    for (int i=0; i<siblingContent.n; ++i) {
	    	childContent.key[i+t].v = siblingContent.key[i].v;
	    	childContent.value[i+t].r = siblingContent.value[i].r;
	    }
	 
	    // Copying the child pointers from C[idx+1] to C[idx]
	    if (!childContent.leaf) {
	        for(int i=0; i<=siblingContent.n; ++i) {
	        	childContent.c[i+t].r = siblingContent.c[i].r;
				setOwner(graph, childContent.c[i+t].r, bean.c[idx].r, true);
	        }
	    }
	    
	    // Moving all keys after idx in the current node one step before -
	    // to fill the gap created by moving keys[idx] to C[idx]
	    for (int i=idx+1; i<bean.n; ++i) {
	        bean.key[i-1].v = bean.key[i].v;
	        bean.value[i-1].r = bean.value[i].r;
	    }

        // Remove the owner of the merged sibling
        removeOwner(graph, bean.c[idx+1].r, true);
	    
	    // Moving the child pointers after (idx+1) in the current node one
	    // step before
	    for (int i=idx+2; i<=bean.n; ++i)
	        bean.c[i-1].r = bean.c[i].r;
	 
	    // Updating the key count of child and the current node
	    childContent.n += siblingContent.n+1;
	    bean.reduceSize();
	    //bean.n--;
	 
	    // Freeing the memory occupied by sibling
	    // delete(sibling);
	    
	    manager.setContentBean(graph, r, bean);
	    manager.setContentBean(graph, bean.c[idx].r, childContent);

	}
	
	private void removeOwner(WriteGraph graph, Resource r, boolean force) throws DatabaseException {
		if((ownerRelation != null) || force) {
			if (cached) {
				owners.put(r, null);
			} else {
				Resource own = ownerRelation != null ? ownerRelation : DATA.BTreeNode_IsOwnedBy;
				graph.deny(r, own);
			}
		}
	}
	
	private void removeFromLeaf(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int idx) throws DatabaseException {

	    // Move all the keys after the idx-th pos one place backward
	    for (int i=idx+1; i<bean.n; ++i) {
	    	bean.key[i-1].v = bean.key[i].v;
	    	bean.value[i-1].r = bean.value[i].r;
	    }
	 
	    // Reduce the count of keys
	    bean.reduceSize();
	    //bean.n--;
	    
	    manager.setContentBean(graph, r, bean);
	    
	}

	private void removeFromNonLeaf(WriteGraph graph, BTreeContentManager manager, Resource r, BTreeContentBean bean, int t, int idx) throws DatabaseException {

		Variant k = bean.key[idx].v;
		 
		// If the child that precedes k (C[idx]) has atleast t keys,
		// find the predecessor 'pred' of k in the subtree rooted at
		// C[idx]. Replace k by pred. Recursively delete pred
		// in C[idx]
		
        BTreeContentBean cContent = manager.getContentBean(graph, bean.c[idx].r);

		if (cContent.n >= t) {
			
			Pair<Variant, Resource> pred = getPred(graph, manager, bean, idx);
			bean.key[idx].v = pred.first;
			bean.value[idx].r = pred.second;
			removeImpl(graph, manager, t, bean.c[idx].r, cContent, pred.first);
			
			manager.setContentBean(graph, r, bean);
		} else {

	        BTreeContentBean cContentPlus1 = manager.getContentBean(graph, bean.c[idx+1].r);

			// If the child C[idx] has less that t keys, examine C[idx+1].
			// If C[idx+1] has atleast t keys, find the successor 'succ' of k in
			// the subtree rooted at C[idx+1]
			// Replace k by succ
			// Recursively delete succ in C[idx+1]
			if (cContentPlus1.n >= t) {
				Pair<Variant, Resource> succ = getSucc(graph, manager, bean, idx);
				bean.key[idx].v = succ.first;
				bean.value[idx].r = succ.second;
				removeImpl(graph, manager, t, bean.c[idx+1].r, cContentPlus1, succ.first);
				
				manager.setContentBean(graph, r, bean);
			}

			// If both C[idx] and C[idx+1] has less that t keys,merge k and all of C[idx+1]
			// into C[idx]
			// Now C[idx] contains 2t-1 keys
			// Free C[idx+1] and recursively delete k from C[idx]
			else {
				
				merge(graph, manager, r, bean, t, idx);
				removeImpl(graph, manager, t, bean.c[idx].r, cContent, k);
				
			}
			
		}
		
	}

	private Resource createNode(WriteGraph graph, BTreeContentManager manager, int t) throws DatabaseException {
		Layer0 L0 = Layer0.getInstance(graph);
		Resource result = graph.newResource();
		mod++;
		graph.claim(result, L0.InstanceOf, null, nodeType);
		if (!cached) {
			graph.claimLiteral(tree, DATA.BTree_mod, mod, Bindings.LONG);
		}
		graph.addLiteral(result, L0.HasName, L0.NameOf, ""+mod, Bindings.STRING);
		manager.setContentBean(graph, result, BTreeContentBean.create(t));
		return result;
	}
	
	public Resource searchNode(ReadGraph graph, Resource x, Variant k) throws DatabaseException {

		BTreeContentBean xContent = getContentBean(graph, x);

		int i=1;
		while(i <= xContent.n && k.compareTo(xContent.key[i-1].v) > 0) i++;
		if(i <= xContent.n && k.equals(xContent.key[i-1].v)) return xContent.value[i-1].r;
		
		if(xContent.leaf) return null;
		else return searchNode(graph, xContent.c[i-1].r, k);
		
	}

	private void splitChild(WriteGraph graph, BTreeContentManager manager, int t, Resource x, int i, Resource y) throws DatabaseException {
		
		Resource z = allocateNode(graph, manager, t);
		
		BTreeContentBean xContent = manager.getContentBean(graph, x);
		BTreeContentBean yContent = manager.getContentBean(graph, y);
		BTreeContentBean zContent = manager.getContentBean(graph, z);

		zContent.leaf = yContent.leaf;
		zContent.n = t-1;
		
		for(int j=1;j<=t-1;j++) {
			zContent.key[j-1].v = yContent.key[j+t-1].v;
			zContent.value[j-1].r = yContent.value[j+t-1].r;
			setOwner(graph, zContent.value[j-1].r, z, false);
		}
		if(!yContent.leaf) {
			for(int j=1;j<=t;j++) {
				zContent.c[j-1].r = yContent.c[j+t-1].r;
				setOwner(graph, zContent.c[j-1].r, z, true);
			}
		}

		//yContent.n = t-1;

		
		for(int j=xContent.n+1;j>=i+1;j--) {
			xContent.c[j+1-1].r = xContent.c[j-1].r; 
		}
		
		xContent.c[i+1-1].r = z;
		setOwner(graph, z, x, true);

		for(int j=xContent.n;j>=i;j--) {
			xContent.key[j+1-1].v = xContent.key[j-1].v; 
			xContent.value[j+1-1].r = xContent.value[j-1].r; 
		}
		
		xContent.key[i-1].v = yContent.key[t-1].v;
		xContent.value[i-1].r = yContent.value[t-1].r;
		xContent.n++;

		while (yContent.n > t-1) yContent.reduceSize();
		
		setOwner(graph, xContent.value[i-1].r, x, true);
		
		manager.setContentBean(graph, x, xContent);
		manager.setContentBean(graph, y, yContent);
		manager.setContentBean(graph, z, zContent);
		
	}
	
	private void insertNonfull(WriteGraph graph, BTreeContentManager manager, int t, Resource x, Variant k, Resource v) throws DatabaseException {

		BTreeContentBean xContent = manager.getContentBean(graph, x);

		int i = xContent.n;
		
		if(xContent.leaf) {
			while(i >= 1 && k.compareTo(xContent.key[i-1].v) < 0) {
				xContent.key[i+1-1].v = xContent.key[i-1].v;
				xContent.value[i+1-1].r = xContent.value[i-1].r;
				i--;
			}
			
			xContent.key[i+1-1].v = k;
			xContent.value[i+1-1].r = v;
			xContent.n++;
			setOwner(graph, v, x, false);
			manager.setContentBean(graph, x, xContent);
		} else {
			while(i >= 1 && k.compareTo(xContent.key[i-1].v) < 0) i--;
			i++;
			BTreeContentBean xciContent = manager.getContentBean(graph, xContent.c[i-1].r);
			if(xciContent.n == 2*t-1) {
				splitChild(graph, manager, t, x, i, xContent.c[i-1].r);
				xContent = manager.getContentBean(graph, x);
				if(k.compareTo(xContent.key[i-1].v) > 0) i++;
			}
			insertNonfull(graph, manager, t, xContent.c[i-1].r, k, v);
		}
		
	}
	
	private void setOwner(WriteGraph graph, Resource r, Resource owner, boolean force) throws DatabaseException {
		if (r != null) {
			if((ownerRelation != null) || force) {
				Resource own = ownerRelation != null ? ownerRelation : DATA.BTreeNode_IsOwnedBy;
				if (cached) {
					owners.put(r, owner);
				} else {
					graph.deny(r, own);
					graph.claim(r, own, owner);
				}
			}
		}
	}
	
	private Resource getRoot(ReadGraph graph) throws DatabaseException {
		if (cached) {
			if (root != null) return root;
		}
		return graph.getPossibleObject(tree, DATA.BTree_root);
	}
	
	private void setRoot(WriteGraph graph, Resource r) throws DatabaseException {
		if (cached) {
			this.root = r;
		} else {
			graph.deny(tree, DATA.BTree_root);
			graph.claim(tree, DATA.BTree_root, r);
		}
	}

	private Resource allocateNode(WriteGraph graph, BTreeContentManager manager, int t) throws DatabaseException {
		return createNode(graph, manager, t);
	}
	
	public void setContentBean(WriteGraph graph, Resource node, BTreeContentBean bean) throws DatabaseException {
		if (cached) {
			beans.put(node, bean);
		} else {
			graph.claimLiteral(node, DATA.BTreeNode_content, DATA.BTreeNode_Content, bean, CONTENT_BEAN_BINDING );
		}
	}
	
	public void flushCache(WriteGraph graph) throws DatabaseException {
		assert(cached);
		
		XSupport xs = graph.getService(XSupport.class);
		for(Map.Entry<Resource, BTreeContentBean> entry : beans.entrySet()) {
			Resource node = xs.convertDelayedResourceToResource(entry.getKey());
			if (node != null) {
				graph.claimLiteral(node, DATA.BTreeNode_content, DATA.BTreeNode_Content, entry.getValue(), CONTENT_BEAN_BINDING );
			}
		}

		for(Map.Entry<Resource, Resource> entry : owners.entrySet()) {
			Resource r = xs.convertDelayedResourceToResource(entry.getKey());
			Resource owner = xs.convertDelayedResourceToResource(entry.getValue());
			graph.deny(r, ownerRelation);
			if (owner != null) {
				graph.claim(r, ownerRelation, owner);
			}
		}
		
		Resource tree2 = xs.convertDelayedResourceToResource(tree);
		graph.claimLiteral(tree2, DATA.BTree_mod, mod, Bindings.LONG);
		
		if (root != null) {
			Resource root2 = xs.convertDelayedResourceToResource(root);
			graph.deny(tree2, DATA.BTree_root);
			graph.claim(tree2, DATA.BTree_root, root2);
		}
	}
	
	public BTreeContentBean getContentBean(ReadGraph graph, Resource node) throws DatabaseException {
		if (cached) {
			BTreeContentBean bean = beans.get(node);
			if (bean != null) return bean;
		}
		return graph.syncRequest(new RelatedValue<BTreeContentBean>(node, DATA.BTreeNode_content, CONTENT_BEAN_BINDING), TransientCacheListener.<BTreeContentBean>instance());
	}
	
	public List<Tuple2> entries(ReadGraph graph) throws DatabaseException {
		Resource r = getRoot(graph);
		List<Tuple2> results = new ArrayList<Tuple2>();
		return collectEntries(graph, r, null, null, results);
	}
	
	private List<Tuple2> collectEntries(ReadGraph graph, Resource x, Variant min, Variant max, List<Tuple2> results) throws DatabaseException {

		BTreeContentBean xContent = getContentBean(graph, x);

		for (int i = 0; i < xContent.n + 1; i++) {
			if (min != null && i < xContent.n && min.compareTo(xContent.key[i].v) > 0) continue;
			
			if (!xContent.leaf) {
				Resource child = xContent.c[i].r;
				if (child != null) {
					
					// reduce the amount of comparison if we know that the range of the child keys matches
					Variant min2 = (min != null && i > 0 && min.compareTo(xContent.key[i - 1].v) < 0) ? null : min; 
					Variant max2 = (max != null && i < xContent.n - 1 && max.compareTo(xContent.key[i + 1].v) > 0) ? null : max;

					collectEntries(graph, child, min2, max2, results);
				}
			}
			
			if (i < xContent.n) {
				if (max != null && max.compareTo(xContent.key[i].v) < 0) return results;

				results.add(new Tuple2(xContent.key[i].v, xContent.value[i].r));
			}
		}
		return results;
	}
	
	public List<Tuple2> searchRange(ReadGraph graph, Variant min, Variant max) throws DatabaseException {
		Resource r = getRoot(graph);
		List<Tuple2> results = new ArrayList<Tuple2>();
		return collectEntries(graph, r, min, max, results);
	}
	
}