This Gist contains all classes of an alpha version of a HyperGraphDB storage implementation using Hazelcast

BasicTests.scala

package hgtest.storage.hazelstore

import TestCommons._
import org.hypergraphdb.{HGConfiguration, HGPersistentHandle, HGQuery, HyperGraph}
import org.hypergraphdb.query.AtomTypeCondition
import collection.JavaConversions._
import org.hypergraphdb.storage.hazelstore.HazelStore

object BasicTests {

 def main (args:Array[String]){
 	// fire up some hazelcast instances before starting (server.sh in hazelcast/bin/)
  	val graph = new HyperGraph
    val config: HGConfiguration = new HGConfiguration
    config.setStoreImplementation(new HazelStore())
  	graph.setConfig(config)
  	graph.open("")
	(0 to 50).foreach(a => graph.add(TestCommons.randomString))   // outcomment this after first run. Then rerun.
  	println("instantiated HyperGraph. " + graph.get(graph.add("hello World")))
    import HGQuery.hg._
    findAll(graph, new AtomTypeCondition(classOf[String])).asInstanceOf[java.util.List[HGPersistentHandle]].foreach{ case b : HGPersistentHandle => println(graph.get(b.asInstanceOf[HGPersistentHandle]))}
  	
  	graph.close

}
	
}

BAWrapper.java

package org.hypergraphdb.storage;

import java.util.Arrays;
import java.io.Serializable; 

public class BAWrapper implements Serializable{
	    private final byte[] data;

	    public BAWrapper(byte[] data)
	    {
	        if (data == null)
	        {
	            throw new NullPointerException();
	        }
	        this.data = data;
	    }

	    public byte[] get(){return data;}
	    
	    @Override
	    public boolean equals(Object other)
	    {
	        if (!(other instanceof BAWrapper))
	        {
	            return false;
	        }
	        return Arrays.equals(data, ((BAWrapper)other).data);
	    }

	    @Override
	    public int hashCode()
	    {
	        return Arrays.hashCode(data);
	    }
	}

ByteArrayComparator.java

package org.hypergraphdb.storage;

// taken from http://code.google.com/p/flatmap/source/browse/trunk/src/java/com/spinn3r/flatmap/ByteArrayComparator.java
import java.io.*;
import java.nio.*;
import java.util.*;

public class ByteArrayComparator implements Comparator<byte[]> {

    public ByteArrayComparator() {    }
 
  
    public int compare( byte[] b1, byte[] b2 ) {

        if ( b1.length != b2.length ) {

            String msg = String.format( "differing lengths: %d vs %d", b1.length, b2.length );
           
            throw new RuntimeException( msg );
        }
       
        for( int i = 0; i < b1.length; ++i ) {

            if ( b1[i] < b2[i] )
                return -1;

            if ( b1[i] > b2[i] )
                return 1;

            if ( b1[i] == b2[i] ) {

                //we're not done comparing yet.
                if ( i < b1.length - 1 )
                    continue;

                return 0;
               
            }
           
        }
       
        throw new RuntimeException();
       
    }

}

HazelBidirectionalIndex2.java

package org.hypergraphdb.storage.hazelstore;

import com.hazelcast.core.HazelcastInstance;
import org.hypergraphdb.HGBidirectionalIndex;
import org.hypergraphdb.HGRandomAccessResult;
import org.hypergraphdb.storage.ByteArrayConverter;
import org.hypergraphdb.transaction.HGTransactionManager;

import java.util.Collection;
import java.util.Comparator;



public class HazelBidirectionalIndex2<KeyType, ValueType> extends HazelIndex2<KeyType, ValueType> implements HGBidirectionalIndex<KeyType, ValueType> {
    public HazelBidirectionalIndex2(String name, HazelcastInstance hi, HGTransactionManager transactionManager, ByteArrayConverter<KeyType> keyConverter, ByteArrayConverter<ValueType> valueConverter, Comparator<?> comparator) {
        super(name, hi, transactionManager, keyConverter, valueConverter, comparator);
    }
    @Override
    public HGRandomAccessResult<KeyType> findByValue(ValueType value) {
// Collection<KeyType> a = CollectionConverter.Utils$.MODULE$.findByValue(index, ((ValueType) value)); return hgRARS(a, keyConverter);
return utils.findByValue(index, keyConverter, valueConverter, value);
    }
    @Override
    public KeyType findFirstByValue(ValueType value) {
// return CollectionConverter.Utils$.MODULE$.findFirstByValue(index, (ValueType) value);
return utils.findFirstByValue(index, keyConverter, valueConverter, value);
    }
    @Override
    public long countKeys(ValueType value) {
        return 0;
    }

}

HazelIndex2.java

package org.hypergraphdb.storage.hazelstore;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.MultiMap;
import org.hypergraphdb.HGPersistentHandle;
import org.hypergraphdb.HGRandomAccessResult;
import org.hypergraphdb.HGSearchResult;
import org.hypergraphdb.HGSortIndex;
import org.hypergraphdb.storage.BAWrapper;
import org.hypergraphdb.storage.ByteArrayConverter;
import org.hypergraphdb.storage.hazelstore.Utils;
import org.hypergraphdb.storage.hazelstore.Ordr;
import org.hypergraphdb.transaction.HGTransactionManager;
import org.hypergraphdb.util.ArrayBasedSet;
//import org.hypergraphdb.util.LLRBTreeCountMe;
import org.hypergraphdb.storage.LLRBTreeCountMe;

import java.util.Collection;
import java.util.Comparator;

public class HazelIndex2<KeyType, ValueType> implements HGSortIndex<KeyType, ValueType> {
    String name;
    HazelcastInstance h;
    MultiMap<BAWrapper,BAWrapper> index;
    HGTransactionManager transactionManager;
    ByteArrayConverter<KeyType> keyConverter;
    ByteArrayConverter<ValueType> valueConverter;
    Comparator<?> comparator;
Utils utils = new Utils();

    public HazelIndex2(String name, HazelcastInstance h, HGTransactionManager transactionManager, ByteArrayConverter<KeyType> keyConverter, ByteArrayConverter<ValueType> valueConverter, Comparator<?> comparator) {
        this.name = name;
        this.h = h;
        this.index = h.getMultiMap(name);
        this.transactionManager = transactionManager;
        this.keyConverter = keyConverter;
        this.valueConverter = valueConverter;
        this.comparator = comparator;
    }

    @Override
    public HGSearchResult<ValueType> findLT(KeyType key) {
        return utils.<ValueType>findBAW(index.keySet(), ((Comparator<byte[]>) comparator), key2BA(key),index, valueConverter, Ordr.LT);
    }
    @Override
    public HGSearchResult<ValueType> findGT(KeyType key) {
        return utils.<ValueType>findBAW(index.keySet(), ((Comparator<byte[]>) comparator), key2BA(key),index, valueConverter, Ordr.GT);
    }
    @Override
    public HGSearchResult<ValueType> findLTE(KeyType key) {
        return utils.<ValueType>findBAW(index.keySet(), ((Comparator<byte[]>) comparator), key2BA(key),index, valueConverter, Ordr.LTE);
    }
    @Override
    public HGSearchResult<ValueType> findGTE(KeyType key) {
        return utils.<ValueType>findBAW(index.keySet(), ((Comparator<byte[]>) comparator), key2BA(key),index, valueConverter, Ordr.GTE);
    }
    @Override
    public void addEntry(KeyType key, ValueType value) {
        index.put(key2BA(key), val2BA(value));
    }
    @Override
    public void removeEntry(KeyType key, ValueType value) {
        boolean presentBefre = index.containsEntry(key2BA(key), val2BA(value));
        index.remove(key2BA(key), val2BA(value));
        // might be wrongly fail, when there are duplicate entries
        boolean presentAfter = index.containsEntry(key2BA(key), val2BA(value)) || index.get(key2BA(key)).contains(val2BA(value));
        if(presentAfter==presentBefre)
            System.out.println("Hazelindex.remove: remove did not work. value was not present in the first place");
    }
    @Override
    public void removeAllEntries(KeyType key) {
        index.remove(key2BA(key));
    }

    @Override
    public ValueType findFirst(KeyType key) {
        ValueType result = null;
        try { result = valueConverter.fromByteArray(index.get(key2BA(key)).iterator().next().get());} catch ( Throwable t){}
        return result;
    }
    @Override
    public HGRandomAccessResult<ValueType> find(KeyType key) {
        HGRandomAccessResult<ValueType> result = hgRARS(index.get(key2BA(key)), valueConverter);
        return result;
    }

    @Override
    public void open() {
        
    }
    @Override
    public void close() {
        
    }
    @Override
    public boolean isOpen() {
        return (index.getName() != null) ;
    }
    @Override
    public HGRandomAccessResult<KeyType> scanKeys() { return hgRARS(index.keySet(), keyConverter); }

    @Override
    public HGRandomAccessResult<ValueType> scanValues() {
        Collection<BAWrapper> values = index.values();
        return hgRARS(values, valueConverter);
    }

    @Override
    public long count() {
        // index.size returns number of key-value pairs which make up the multimap. Is there no other way of getting number of keys without returning the entire keyset?
        return index.keySet().size();
    }
    @Override
    public long count(KeyType key) {
        return index.valueCount(key2BA(key));
    }

    protected BAWrapper val2BA(ValueType value) { return new BAWrapper(valueConverter.toByteArray(value));}
    protected BAWrapper key2BA(KeyType key) { return new BAWrapper(keyConverter.toByteArray(key));}

    protected <T> HGRandomAccessResult<T> hgRARS(Collection<BAWrapper> t, final ByteArrayConverter<T> converter) {
        if (t == null || t.size() == 0)
                return (HGRandomAccessResult<T> ) HGRandomAccessResult.EMPTY;
        LLRBTreeCountMe<T> tree = new LLRBTreeCountMe<T>();
        for (BAWrapper baw : t)
         tree.add(converter.fromByteArray(baw.get()));
        return tree.getSearchResult();
     
    }
/*
    protected <T> HGRandomAccessResult<T> hgRARS(Collection<T> t, final ByteArrayConverter<T> converter) {
        //Collection<T> values = Utils$.MODULE$.fromBA(t, converter);
        if(t == null || t.size() == 0)
            return (HGRandomAccessResult<T>) HGRandomAccessResult.EMPTY;

        Class<T> type = (Class<T>) ((T) t.iterator().next()).getClass();

        T[] array = (T[])java.lang.reflect.Array.newInstance(type, 0);

        ArrayBasedSet<T> rs = new ArrayBasedSet<T>(t.toArray(array), new Comparator<T>() {
            @Override
            public int compare(T t, T t1) {
                return ((Comparator<byte[]>)comparator).compare(converter.toByteArray(t), converter.toByteArray(t1));
            }
            @Override
            public boolean equals(Object o) {
                return ((Comparator<byte[]>)comparator).equals(o);
            }
        });
        return rs.getSearchResult();
    }
  */

    private <T> Comparator<T> getComparator(final ByteArrayConverter<T> bac){
        return new Comparator<T>(){
            @Override
            public int compare(T t, T t1) {
                return ((Comparator<byte[]>)comparator).compare(bac.toByteArray(t), bac.toByteArray(t1));
            }
        };
    }
}

HazelStore.java

package org.hypergraphdb.storage.hazelstore;

import com.hazelcast.config.Config;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.MultiMap;
import com.hazelcast.core.Transaction;
import org.hypergraphdb.storage.*;
import org.hypergraphdb.storage.hazelstore.Utils;

import org.hypergraphdb.*;
import org.hypergraphdb.transaction.*;
import org.hypergraphdb.util.HGLogger;
//import org.hypergraphdb.util.LLRBTree;

import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class HazelStore implements HGStoreImplementation {

    private Map<String, HGIndex<?,?>> openIndices;
    private ReentrantReadWriteLock indicesLock;
    private HGStore store;
    private Map<String,MultiMap> indices;
    private int handleSize =16;
    private HGHandleFactory handleFactory;
    private HGConfiguration configuration;
    private HGLogger logger = new HGLogger();

    Hazelcast h = null;
    private HazelcastInstance hi;
    private Map<BAWrapper, BAWrapper> linkDB;
    private Map<BAWrapper, BAWrapper> dataDB;
    private MultiMap<BAWrapper, BAWrapper> inciDB;

    public HazelStore() {
        this.openIndices = new HashMap<String, HGIndex<?,?>>();
        this.indicesLock = new ReentrantReadWriteLock();
        this.indices    = new LinkedHashMap<String, MultiMap>();

    }

    @Override
    public Object getConfiguration() {
        return null;  
    }
    @Override
    public void startup(HGStore store, HGConfiguration configuration) {
        this.configuration = configuration;
        this.handleFactory = configuration.getHandleFactory();
        this.handleSize = this.handleFactory.nullHandle().toByteArray().length;
        this.store = store;
        System.out.println("hazelstore sbt version");

        Config hconfig = new Config();
        hconfig.setLiteMember(true);
        hi = Hazelcast.newHazelcastInstance(null);
        linkDB = hi.getMap("linkdb2");
        dataDB = hi.getMap("datadb2");
        inciDB = hi.getMultiMap("incidb");


    }
    @Override
    public void shutdown() {
        hi.getLifecycleService().shutdown();
    }
    @Override
    public HGTransactionFactory getTransactionFactory() {

        return new HGTransactionFactory()
        {
            @Override
            public HGStorageTransaction createTransaction(HGTransactionContext context, HGTransactionConfig config, HGTransaction parent)
            {
                final Transaction t = hi.getTransaction();
                t.begin();
                return new HGStorageTransaction()
                        {
                            @Override
                            public void commit() throws HGTransactionException {
                                t.commit();
                            }
                            @Override
                            public void abort() throws HGTransactionException {
                                t.rollback();
                            }
                        };
            }
            @Override
            public boolean canRetryAfter(Throwable t) {
                return false;
                }
            };
    }
    @Override
    public HGPersistentHandle store(HGPersistentHandle handle, HGPersistentHandle[] link) {
        if (handle == null)
            throw new HGException("StorageImpl.getLink() is being called with null handle.");

        byte[] blink = HGConverter.convertHandleArrayToByteArray(link, handleSize);
        linkDB.put(wrap(handle.toByteArray()), wrap(blink));
        return handle;
    }
    @Override
    public HGPersistentHandle[] getLink(HGPersistentHandle handle) {
        byte[] get = unwrap(linkDB.get(wrap(handle.toByteArray())));
        return HGConverter.convertByteArrayToHandleArray(get, handleFactory);
    }
    @Override
    public void removeLink(HGPersistentHandle handle) {
        linkDB.remove(wrap(handle.toByteArray()));
    }
    @Override
    public boolean containsLink(HGPersistentHandle handle) {
        return linkDB.containsKey(wrap(handle.toByteArray()));
    }

   @Override
    public boolean containsData(HGPersistentHandle handle) {
        return dataDB.containsKey(wrap(handle.toByteArray()));
    }

    @Override
    public HGPersistentHandle store(HGPersistentHandle handle, byte[] data) {
        dataDB.put(wrap(handle.toByteArray()), wrap(data));
        return handle;
    }
    @Override
    public void removeData(HGPersistentHandle handle) {
        dataDB.remove(wrap(handle.toByteArray()));
    }
    @Override
    public byte[] getData(HGPersistentHandle handle) {
        return unwrap(dataDB.get(wrap(handle.toByteArray())));
    }
    @Override
    public HGRandomAccessResult<HGPersistentHandle> getIncidenceResultSet(HGPersistentHandle handle) {
    	LLRBTreeCountMe tree = new LLRBTreeCountMe<HGPersistentHandle>();
    	Collection<BAWrapper> get = inciDB.get(wrap(handle.toByteArray()));
        if (get == null || get.size() == 0)
            return (HGRandomAccessResult<HGPersistentHandle>) HGRandomAccessResult.EMPTY;
    	for (BAWrapper ba : get)
    		tree.add(handleFactory.makeHandle(unwrap(ba)));
    		
    	return tree.getSearchResult();
    }
    @Override
    public void removeIncidenceSet(HGPersistentHandle handle) {
        inciDB.remove(wrap(handle.toByteArray()));
    }

    @Override
    public long getIncidenceSetCardinality(HGPersistentHandle handle) {
        return inciDB.entrySet().size();
    }

    @Override
    public void addIncidenceLink(HGPersistentHandle handle, HGPersistentHandle newLink) {
        inciDB.put(wrap(handle.toByteArray()), wrap(newLink.toByteArray()));
    }

    @Override
    public void removeIncidenceLink(HGPersistentHandle handle, HGPersistentHandle oldLink) {
        inciDB.remove(wrap(handle.toByteArray()), wrap(oldLink.toByteArray()));
    }

    @Override
    public <KeyType, ValueType> HGIndex<KeyType, ValueType> getIndex(String name, ByteArrayConverter<KeyType> keyConverter, ByteArrayConverter<ValueType> valueConverter, Comparator<?> comparator, boolean isBidirectional, boolean createIfNecessary) {
        indicesLock.readLock().lock();
        try
        {
            HGIndex<KeyType, ValueType> idx = (HGIndex<KeyType, ValueType>)openIndices.get(name);
            if (idx != null)
                return idx;
            if (!checkIndexExisting(name) && !createIfNecessary)
                return null;
        }
        finally {indicesLock.readLock().unlock(); }

        indicesLock.writeLock().lock();
        try
        {
            HazelIndex2<KeyType, ValueType> result = null;

            if (!indices.containsKey(name))
            {
                indices.put(name, hi.getMultiMap(name));
            }

            if (isBidirectional)
                result = new HazelBidirectionalIndex2<KeyType, ValueType>(  name, hi,
                        store.getTransactionManager(),
                        keyConverter,
                        valueConverter,
                        comparator);

            else
                result = new HazelIndex2<KeyType, ValueType>(name, hi,
                        store.getTransactionManager(),
                        keyConverter,
                        valueConverter,
                        comparator);

            openIndices.put(name, result);
            return result;
        }
        finally
        {
            indicesLock.writeLock().unlock();
        }

    }
    @Override
    public void removeIndex(String name) {

        indicesLock.writeLock().lock();
        try
        {
            HGIndex idx = openIndices.get(name);
            if (idx != null)
            {
                idx.close();
                openIndices.remove(name);
                indices.remove(name);
            }
        }
        finally
        {
            indicesLock.writeLock().unlock();
        }



    }

    boolean checkIndexExisting(String name)
    {
        return openIndices.get(name) != null;
    }

    private BAWrapper wrap(byte[] ba) {return new BAWrapper(ba);}
    private byte[] unwrap(BAWrapper ba) {if (ba ==null) return null; else return ba.get();}
}

HGConverter.java

package org.hypergraphdb.storage;

import org.hypergraphdb.HGException;
import org.hypergraphdb.HGHandleFactory;
import org.hypergraphdb.HGPersistentHandle;
import org.hypergraphdb.HyperGraph;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Set;

public class HGConverter
{
    public static int handleSize = 16; 

	public static byte [] convertStringtoByteArray(String s)
	{
	try 						
	{	
		// return IOUtils.toByteArray(new StringReader(s), );	
		return s.getBytes();
	}
	catch (Exception ioEx) 	{throw new HGException("Error when converting String to Byte" + ioEx);} 
	}
	
    public static String convertByteArraytoString(byte [] b)
    {
    	try 						
    	{
    		return new String(b);
    		// return IOUtils.toString(new ByteArrayInputStream(b), );	
    	}

    	catch (Exception ioEx) 	{throw new HGException("Error when converting from Byte to String " + ioEx);} 
	}
    
	public static String convertHandleArrayToString (HGPersistentHandle[] link, int handleSize)
	    {
		  String resultstring = null;
		  try {
	    		byte[] buffer = new byte[link.length * handleSize];
	    		for (int i = 0; i < link.length; i++)
	           {
	               HGPersistentHandle handle = (HGPersistentHandle)link[i];
	               System.arraycopy(handle.toByteArray(), 0, 
	                                buffer, i*handleSize, 
	                                handleSize);             
				}
	    		resultstring = new String(buffer);
			} catch (Exception e) {
				e.printStackTrace();
			}
    		return resultstring;
	    }   

	public static HGPersistentHandle[] convertStringToHandleArray(String inputString, int handlesize, HGHandleFactory hghandlefactory)
	    {
		  HGPersistentHandle [] handles;
		  
		  byte[] input = HGConverter.convertStringtoByteArray(inputString);
		  int size = input.length;
		  
	    		if (input == null)
	    		  	return HyperGraph.EMPTY_PERSISTENT_HANDLE_SET;
	    		else if (size == 0)
	   		   		return HyperGraph.EMPTY_PERSISTENT_HANDLE_SET;
	    		else if (size % handleSize != 0)
	    			throw new HGException("While reading link tuple: the value buffer size is not a multiple of the handle size.");
	    		else 
	    			{ 	int handle_count = size / handleSize;
	    			  	handles = new HGPersistentHandle[handle_count];
	    			  	for (int i = 0; i < handle_count; i++)  
	    			  		handles[i] = hghandlefactory.makeHandle(input, i*handleSize);  // in LinkBinding.readHandle calls is called makeHandle with buffer, "offset + i*handleSize", whereas offset comes from entryToObject which is just 0
	    			}  	
	    		
	    		return handles;
	    }
	  	
	public static byte[] convertHandleArrayToByteArray(HGPersistentHandle[] link, int handleSize)
  {    	
  		byte [] buffer = new byte[link.length * handleSize];
  		for (int i = 0; i < link.length; i++)
         {
             HGPersistentHandle handle = (HGPersistentHandle)link[i];
             System.arraycopy(handle.toByteArray(), 0, 
                              buffer, i*handleSize, 
                              handleSize);            
         }
      return buffer;
  }   

	public static HGPersistentHandle[] convertByteArrayToHandleArray(byte[] input, HGHandleFactory hghandlefactory)
	  {
        if (input == null)
	  		  	return null;
          int size = input.length;



          if (size == 0)
                return HyperGraph.EMPTY_PERSISTENT_HANDLE_SET;

          if (size % handleSize != 0)
            throw new HGException("While reading link tuple: the value buffer size is not a multiple of the handle size.");
	     int handle_count = size / handleSize;
	 	 HGPersistentHandle [] handles = new HGPersistentHandle[handle_count];
	         for (int i = 0; i < handle_count; i++)  
	             handles[i] = hghandlefactory.makeHandle(input, i*handleSize);  // in LinkBinding.readHandle calls is called makeHandle with buffer, "offset + i*handleSize", whereas offset comes from entryToObject which is just 0 ?
	     return handles;
	  }

	
	public static HGPersistentHandle[] convertByteArrayToHandleArray(byte[] input, int handlesize, HGHandleFactory hghandlefactory)
  {
	  int size = input.length;
	  
  		if (input == null)
  		  	return null;
  	  
  	  if (size == 0)
 		   	return HyperGraph.EMPTY_PERSISTENT_HANDLE_SET;

  	  if (size % handleSize != 0)
        throw new HGException("While reading link tuple: the value buffer size is not a multiple of the handle size.");
     int handle_count = size / handleSize;
 	   HGPersistentHandle [] handles = new HGPersistentHandle[handle_count];
         for (int i = 0; i < handle_count; i++)  
             handles[i] = hghandlefactory.makeHandle(input, i*handleSize);  // in LinkBinding.readHandle calls is called makeHandle with buffer, "offset + i*handleSize", whereas offset comes from entryToObject which is just 0
     return handles;
  }
  
	public static HGPersistentHandle[]  convertBAsetToHandleArray(Set<byte[]> in, HGHandleFactory hghandlefactory)
	{
		return convertByteArrayToHandleArray(flattenBASetToBA(in), hghandlefactory);
	}
	public static byte[] flattenBASetToBA (Set<byte[]> in)
	{	
		int index = 0;
		byte[] resultBA = new byte[in.size()*16];   // TODO - generalize handleSize = handleFactory.nullHandle().toByteArray().length;
				 
		Iterator<byte[]> it = in.iterator();
		while(it.hasNext())
		{
			byte[] current = it.next();
			for(int i = 0; i<current.length-1; i++)
			{ 
				resultBA[index]= current[i];
				index ++;
			}
		}
		return resultBA;
	}

    // next two methods taken from: http://stackoverflow.com/questions/5399798/byte-array-and-int-conversion-in-java/5399829#5399829

    public static Integer byteArrayToInt(byte[] b)
    {
        int value = 0;

        if(b == null)
            return null;

        else
        {
            for (int i = 0; i < 4; i++)
                value = (value << 8) | (b[i] & 0xFF);
        }
        return value;
    }

    public static byte[] intToByteArray(int a)
    {
        byte[] ret = new byte[4];
        ret[3] = (byte) (a & 0xFF);
        ret[2] = (byte) ((a >> 8) & 0xFF);
        ret[1] = (byte) ((a >> 16) & 0xFF);
        ret[0] = (byte) ((a >> 24) & 0xFF);
        return ret;
    }

    // took it from here: http://stackoverflow.com/questions/80476/how-to-concatenate-two-arrays-in-java
    public static byte[] concat(byte[] first, byte[] second)
    {
      byte[] result = Arrays.copyOf(first, first.length + second.length);
      System.arraycopy(second, 0, result, first.length, second.length);
      return result;
    }

}

LLRBTreeCountMe.java

/* 
 * This file is part of the HyperGraphDB source distribution. This is copyrighted 
 * software. For permitted uses, licensing options and redistribution, please see  
 * the LicensingInformation file at the root level of the distribution.  
 * 
 * Copyright (c) 2005-2010 Kobrix Software, Inc.  All rights reserved. 
 */
package org.hypergraphdb.storage;

import java.util.AbstractSet;
import java.util.Comparator;
import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.SortedSet;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.hypergraphdb.HGRandomAccessResult;
import org.hypergraphdb.HGSearchResult;
import org.hypergraphdb.util.HGSortedSet;
import org.hypergraphdb.util.CountMe;


/**
 * 
 * <p>
 * Implements a set of elements as a left-leaning red-black tree. The node
 * doesn't contain a pointer to a value, nor does it contains a pointer
 * to the parent which should make it more memory efficient than most
 * implementations (e.g. the standard java.util.TreeMap). However, tree
 * mutations are implemented recursively, which is not optimal and could
 * be removed in the future. Unfortunately, Java uses 4 bytes to store a boolean
 * so we don't gain as much in space compactness as we could theoretically, but
 * it's still an improvement. 
 * </p>
 *
 * @author Borislav Iordanov
 *
 * @param <E> The type of elements this set stores. It must implement the <code>Comparable</code>
 * interface or a <code>Comparator</code> has to be provided at construction time.
 */
public class LLRBTreeCountMe<E> extends AbstractSet<E>
						 implements HGSortedSet<E>, Cloneable, java.io.Serializable
{
	private static final long serialVersionUID = -1;
	
	private static final boolean RED = true;
	private static final boolean BLACK = false;

	// The node stack is used to keep track of parent pointers
	// during tree traversals. A fixed-size array is used for
	// the stack because it is known that it'll never go beyond
	// the depth of the tree which, since this a balanced tree,
	// is going to always be a relatively small number approx. 
	// equal to 2*log(treeSize).
	private final class NodeStack
	{
		Node<E> [] A;
		Node<E> [] B;
		int pos = -1;
		int bpos;
		
		@SuppressWarnings("unchecked")
		NodeStack()
		{
			int s = 0;
			if (size > 0)
				s = (int)(2*(Math.log(size + 1)/Math.log(2)));
			A = new Node[s];
			B = new Node[s];
		}
		
		void backup()
		{
			bpos = pos;
			System.arraycopy(A, 0, B, 0, pos+1);
		}
		void restore()
		{
			pos = bpos;
			Node<E> [] tmp = A; 
			A = B;
			B = tmp;
		}
		
		boolean isEmpty() { return pos < 0; }
		Node<E> top() { return A[pos]; }
		Node<E> pop() { return A[pos--]; }
		Node<E> push(Node<E> n) { return A[++pos] = n; } 
		void clear() { pos = -1; }
	}
	
	private static class Node<E> implements Cloneable
	{
		E key;
		Node<E> left, right;
		boolean color;
	
		@SuppressWarnings("unchecked")
		public Node<E> clone() throws CloneNotSupportedException
		{
			Node<E> n = (Node<E>)super.clone();
			if (left != null)
				n.left = left.clone();
			if (right != null)
				n.right = right.clone();
			return n;
		}
		
		Node(E key, boolean color)
		{
			this.key = key;			
			this.color = color;
		}
		
		Node<E> rotateLeft()
		{
			Node<E> x = right;
			right = x.left;
			x.left = this;
			x.color = this.color;
			this.color = RED;
			return x;
		}		
		
		Node<E> rotateRight()
		{
			Node<E> x = left;
			left = x.right;
			x.right = this;
			x.color = this.color;
			this.color = RED;
			return x;
		}
		
		Node<E> colorFlip()
		{
			color = !color;
			left.color = !left.color;
			right.color = !right.color;
			return this;
		}		
		
		private Node<E> fixUp()
		{
			Node<E> h = this;
			if (isRed(h.right))
			{
				h = h.rotateLeft();
				if (isRightLeaning(h.left))
					h.left = h.left.rotateLeft();
			}
			if (isRed(h.left) && isRed(h.left.left))
			{
				h = h.rotateRight();
			}
			if (isRed(h.left) && isRed(h.right))
			{
				h.colorFlip();
			}
			return h;
		}
		
		private Node<E> moveRedRight()
		{
			colorFlip();
			if (isRed(left.left))
			{
				Node<E> h = rotateRight();
				return h.colorFlip();
			}
			else
				return this;
		}		
		
		private Node<E> moveRedLeft()
		{
			colorFlip();
			if (isRed(right.left))
			{
				right = right.rotateRight();
				Node<E> h = rotateLeft();
				if (isRightLeaning(h.right))
					h.right = h.right.rotateLeft();
				return h.colorFlip();
			}
			else
				return this;
		}		
	}
	
	private final Node<E> UNKNOWN = new Node<E>(null, BLACK);
	
	final class ResultSet implements HGRandomAccessResult<E>, CountMe
	{
	    boolean locked = false;
		int lookahead = 0;
		Node<E> next = UNKNOWN, current = UNKNOWN, prev = UNKNOWN;
		
		// Keeps track of parents of current node because Node itself doesn't have
		// a parent field.
		NodeStack stack = new NodeStack();  
		
		// min, max, advance, back all work on the current position as 
		// stored in the 'stack' of parents.
		//
		// 'min' returns the smallest element rooted at (and including)
		// stack.top while 'max' analogously returns the largest element
		//
		// 'advance' returns the next smallest elements and 'back' the previous
		// largest element
		//
		// All 4 modify the stack so that it's positioned at the returned
		// node

	public int count(){
            return size;
        }


        Node<E> min()
		{
			Node<E> result = stack.top();
			while (result.left != null)
				result = stack.push(result.left);
			return result;
		}

		Node<E> max()
		{
			Node<E> result = stack.top();
			while (result.right != null)
				result = stack.push(result.right);
			return result;
		}
		
		Node<E> advance()
		{
			Node<E> current = stack.top();
			if (current.right != null)
			{
				stack.push(current.right);
				return min();
			}
			else
			{
				stack.backup();
				stack.pop();
				while (!stack.isEmpty())
				{
					Node<E> parent = stack.top();
					if (parent.left == current)
						return parent;
					else
						current = stack.pop();
				}
				stack.restore();
				return null;
			}
		}
		
		Node<E> back()
		{
			Node<E> current = stack.top();
			if (current.left != null)
			{
				stack.push(current.left);
				return max();
			}
			else
			{
				stack.backup();
				stack.pop();
				while (!stack.isEmpty())
				{
					Node<E> parent = stack.top();
					if (parent.right == current)
						return parent;
					else
						current = stack.pop();
				}
				stack.restore();
				return null;
			}
		}
		
		ResultSet(boolean acquireLock)
		{
		    if (acquireLock)
		        lock.readLock().lock();
		    locked = acquireLock;
		}
		
		public void goBeforeFirst()
		{
		    lookahead = 0;
	        next = current = prev = UNKNOWN;	        
	        stack.clear(); 		    
		}
		
		public void goAfterLast()
		{
            lookahead = 0;
            stack.clear();          
            stack.push(root);            
            prev = max();            
            next = current = UNKNOWN;            
		}
		
		@SuppressWarnings("unchecked")
        public GotoResult goTo(E key, boolean exactMatch)
		{
			// Not clear here whether we should be starting from the root really?
			// Gotos are performed generally during result set intersection where the target
			// is expected to be approximately close to the current position. Anyway,
			// starting from the root simplifies the implementations so until profiling
			// reveals the need for something else we start from the root.
			
			// To make sure the current position remains unchanged if we return GotoResult.nothing
			// we save the stack array.
			stack.backup();
			stack.clear();
			Node<E> current = root; 
			GotoResult result = GotoResult.nothing;
			Comparable<E> ckey = providedComparator == null ? (Comparable<E>)key : null; // make typecast out of loop, expensive!
			while (current != null)
			{
				stack.push(current);
				int cmp = ckey == null ? providedComparator.compare(key, current.key) : ckey.compareTo(current.key);
				if (cmp == 0)
				{
					result = GotoResult.found;
					break;
				}
				else if (cmp < 0)
				{
					if (exactMatch || current.left != null)
						current = current.left;
					else
					{
						result = GotoResult.close;
						break;
					}
				}
				else
				{
					if (exactMatch || current.right != null)
						current = current.right;
					else if (advance() != null)
					{
						result = GotoResult.close;
						break;
					}
					else
						break;
				}
			}
			if (GotoResult.nothing == result)
			{
				stack.restore();
				return GotoResult.nothing;
			}
			else
			{
				lookahead = 0;
				next = UNKNOWN;
				prev = UNKNOWN;
				this.current = stack.top();
				return result;
			}
		}

		public void close()
		{
		    if (locked)
		        lock.readLock().unlock();
		}

		public E current()
		{
			if (current == UNKNOWN)
				throw new NoSuchElementException();
			else
				return current.key;
		}

		public boolean isOrdered()
		{
			return true;
		}

		public boolean hasNext()
		{			
			if (next == UNKNOWN)
				moveNext();
			return next != null;
		}

		// Advance internal cursor to next element and assign 'next' to it. 
		private void moveNext()
		{
			if (stack.isEmpty())
			{
				stack.push(root);
				next = min();
				lookahead = 1;
			}			
			else while (true)
	        {
	        	next = advance();
	        	if (next == null)
	        		break;
	        	if (++lookahead == 1)
	        		break;
	        }			
		}
		
		public E next()
		{
			if (!hasNext())
				throw new NoSuchElementException();
			prev = current;
			current = next;
	        lookahead--;
	        moveNext();
			return current.key;
		}

		public void remove()
		{			
            if (current == UNKNOWN)
                throw new NoSuchElementException();
	        // Because of lack of parent pointers in Node, we can't really
	        // take advantage of the fact that we are already positioned 
	        // at the node we want to delete. In the current iteration context,
	        // we could make use of the parent stack to some advantage, but this
	        // would require a completely new version of the delete algo which
	        // is too big of a price to pay.
	        //
	        // So we just do a normal remove and reset the iterator to its 'prev' state.
            LLRBTreeCountMe.this.remove(current.key);
            if (prev != null)
                if (goTo(prev.key, true) == GotoResult.nothing)
                    throw new Error("LLRBTree.ResultSet.remove buggy.");
            else
            {
                current = prev = next = UNKNOWN;
                lookahead = 0;
                stack.clear();
            }
		}

		public boolean hasPrev()
		{
			if (prev == UNKNOWN)
				movePrev();
			return prev != null;
		}

		private void movePrev()
		{
			if (stack.isEmpty())
				prev = null;			
			else while (true)
	        {
	        	prev = back();
	        	if (prev == null)
	        		break;
	        	if (--lookahead == -1)
	        		break;
	        }			
		}
		
		public E prev()
		{
			if (prev == null)
				throw new NoSuchElementException();
			next = current;
			current = prev;
	        lookahead++;
	        movePrev();
			return current.key;
		}		
	}
	// END IMPLEMENTATION OF Iterator/HGSearchResult
	
	private Node<E> root = null;
	private int size = 0;
	private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
	private Comparator<E> providedComparator = null;
	
	private static boolean isRed(Node<?> x)
	{
		return x == null ? false : x.color == RED;
	}
	
	private static boolean isBlack(Node<?> x)
	{
		return x == null || x.color == BLACK;
	}
	
	// A node is right-leaning if it's right child is red, but it's left is black
	private static boolean isRightLeaning(Node<?> x)
	{
		return x == null ?  false : isRed(x.right) && isBlack(x.left);
	}
	
	@SuppressWarnings("unchecked")
    private Node<E> insert(Node<E> h, E key)
	{
		if (h == null)
		{
			size++;
			return new Node<E>(key, RED);
		}
				
		// Split 4-Nodes
		if (isRed(h.left) && isRed(h.right))
			h.colorFlip();		
		
		int cmp = providedComparator != null ? providedComparator.compare(key, h.key)
		                                     : ((Comparable<E>)key).compareTo(h.key);
		          
		if (cmp < 0)
			h.left = insert(h.left, key);
		else if (cmp > 0)
			h.right = insert(h.right, key);
		
		// Fix right leaning tree.
		if (isRed(h.right) && isBlack(h.left))
			h = h.rotateLeft();		
		// Fix two reds in a row.
		else if (isRed(h.left) && isRed(h.left.left))
			h = h.rotateRight();
						
		return h;
	}
	
	private Node<E> min(Node<E> h)
	{
		if (h == null)
			return null;
		Node<E> x = h;
		while (x.left != null) 
			x = x.left;
		return x;		
	}	
	
	private Node<E> max(Node<E> h)
	{
		if (h == null)
			return null;
		Node<E> x = h;
		while (x.right != null)
			x = x.right;
		return x;		
	}
	
	// The following two functions are for debugging only to print to coloring of a node, its
	// children and its grandchildren.
/*	private String cs(boolean b) { return b ? "red" : "black"; }	
	private String colors(Node<E> h)
	{
		String colors = "h -> " + cs(h.color);
		if (h.left != null)
		{
			colors += ", h.left -> " + cs(h.left.color);
			if (h.left.left != null)
				colors += ", h.left.left -> " + cs(h.left.left.color);
			else
				colors += ", h.left.left = null";
			if (h.left.right != null)
				colors += ", h.left.right -> " + cs(h.left.right.color);
			else
				colors += ", h.left.right = null";
		}
		if (h.right != null)
		{
			colors += ", h.right -> " + cs(h.right.color);
			if (h.right.left != null)
				colors += ", h.right.left -> " + cs(h.right.left.color);
			else
				colors += ", h.right.left = null";
			if (h.right.right != null)
				colors += ", h.right.right -> " + cs(h.right.right.color);
			else
				colors += ", h.right.right = null";
		}	
		return colors;
	} */
	
	private Node<E> deleteMax(Node<E> h)
	{				
		if (isRed(h.left) && isBlack(h.right))
			h = h.rotateRight();
		else if (h.right == null)
			return null; // h.left will be null here as well 
		if (isBlack(h.right) && isBlack(h.right.left))
			h = h.moveRedRight();
		h.right = deleteMax(h.right);
		return h.fixUp();
	}	
	
	private Node<E> deleteMin(Node<E> h)
	{
		if (h.left == null)
			return null; // h.right will be null here as well
		if (isBlack(h.left) && isBlack(h.left.left))
			h = h.moveRedLeft();
		h.left = deleteMin(h.left);
		return h.fixUp();
	}
	
	@SuppressWarnings("unchecked")
    private Node<E> delete(Node<E> h, E key)
	{
		int cmp = providedComparator != null ? providedComparator.compare(key, h.key)
                                             : ((Comparable<E>)key).compareTo(h.key); 
		if (cmp < 0) 
		{
			if (!isRed(h.left) && !isRed(h.left.left))
				h = h.moveRedLeft();
			h.left =  delete(h.left, key);
		}
		else  // cmp >= 0
		{
			if (isRed(h.left) && isBlack(h.right)) 
			{
				h = h.rotateRight();
				cmp++; // if we rotate right, then current h becomes necessarily < key
			}
			else if (cmp == 0 && (h.right == null))
			{
				size--;
				return null; // h.left is null here due to transformations going down
			}
			Node<E> tmp = h; // track if moveRedRight changes 'h' so we don't need to call key.compareTo again
			if (!isRed(h.right) && !isRed(h.right.left))
				tmp = h.moveRedRight();
			// if no rotation in above line and key==h.key replace with successor and we're done
			if (tmp == h && cmp == 0)
			{
				h.key = min(h.right).key;
				h.right = deleteMin(h.right);
				size--;
			} 
			else
			{
				h = tmp;
				h.right = delete(h.right, key);
			}
		} 
		return h.fixUp();		
	}

	public LLRBTreeCountMe()
	{
	}
	
	public LLRBTreeCountMe(Comparator<E> comparator)
	{
		this.providedComparator = comparator;
	}
	
	public void removeMax()
	{
		lock.writeLock().lock();
		try
		{
			if (root == null)
				return;
			root = deleteMax(root);
			if (root != null)
				root.color = BLACK;
			size--;
		}
		finally		
		{
			lock.writeLock().unlock();
		}
	}
	
	public void removeMin()
	{
		lock.writeLock().lock();
		try
		{
			if (root == null)
				return;
			root = deleteMin(root);
			if (root != null)
				root.color = BLACK;
			size--;
		}
		finally		
		{
			lock.writeLock().unlock();
		}
	}	
	
	// Set interface implementation
	
	public int size() { return size; }
	public boolean isEmtpy() { return size == 0; }
	public Comparator<E> comparator() { return providedComparator; }
	
	public void clear()
	{
		lock.writeLock().lock();
		root = null;
		size = 0;
		lock.writeLock().unlock();
	}
	
	@SuppressWarnings("unchecked")
	public boolean contains(Object key)
	{
		lock.readLock().lock();
		try
		{
			Node<E> current = root;
			Comparable<E> ckey = providedComparator == null ? (Comparable<E>)key : null; 
			while (current != null)
			{
				int cmp = ckey != null ? ckey.compareTo(current.key) : providedComparator.compare((E)key, current.key);
				if (cmp == 0)
					return true;
				else if (cmp < 0)
					current = current.left;
				else
					current = current.right;
			}
			return false;
		}
		finally
		{
			lock.readLock().unlock();
		}
	}
	
	public boolean add(E key)
	{
		lock.writeLock().lock();
		try
		{
			int s = size;
			root = insert(root, key);
			root.color = BLACK;
			return s != size; 
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}
	
	@SuppressWarnings("unchecked")	
	public boolean remove(Object key)
	{
		lock.writeLock().lock();
		try
		{
			if (root == null)
				return false;
			int s = size;
			root = delete(root, (E)key);
			if (root != null)
				root.color = BLACK;
			return s != size;
		}
		finally
		{
			lock.writeLock().unlock();
		}
	}

	public E first()
	{
		lock.readLock().lock();
		try
		{
			if (root == null)
				return null;
			return min(root).key;
		}
		finally
		{
			lock.readLock().unlock();
		}
	}

	public E last()
	{
		lock.readLock().lock();
		try
		{
			if (root == null)
				return null;
			return max(root).key;
		}
		finally
		{
			lock.readLock().unlock();
		}
	}

	public SortedSet<E> headSet(E toElement)
	{
		throw new UnsupportedOperationException("...because of lazy implementor: this is a TODO.");
	}
	
	public SortedSet<E> subSet(E fromElement, E toElement)
	{
		throw new UnsupportedOperationException("...because of lazy implementor: this is a TODO.");		
	}

	public SortedSet<E> tailSet(E fromElement)
	{
		throw new UnsupportedOperationException("...because of lazy implementor: this is a TODO.");
	}

	@Override
	@SuppressWarnings("unchecked")
	public Iterator<E> iterator()
	{
		if (isEmpty())
			return (Iterator<E>)HGSearchResult.EMPTY; // avoid checking for root == null in ResultSet impl.
		else
			return new ResultSet(false);
	}

	public HGRandomAccessResult<E> getSearchResult()
	{
		return new ResultSet(true);
	}
	
	@Override
	@SuppressWarnings("unchecked")
	public Object clone() throws CloneNotSupportedException
	{
		lock.readLock().lock();
		try
		{
			LLRBTreeCountMe<E> cl = (LLRBTreeCountMe<E>)super.clone();
			cl.root = root == null ? root : root.clone();
			cl.size = size;
			return cl;
		}
		finally
		{
			lock.readLock().unlock();
		}
	}	
	
	
	public int depth()
	{
		lock.readLock().lock();
		try
		{
			return depth(root);
		}
		finally
		{
			lock.readLock().unlock();
		}		
	}
	private int depth(Node<E> x)
	{
		if (x == null) return 0;
		else return Math.max(1 + depth(x.left), 1 + depth(x.right));
	}
	
	// Integrity checks... 
	public boolean check() 
	{
		return isBST() && is234() && isBalanced();
	}

	public boolean isBST() 
	{  // Is this tree a BST?
		return isBST(root, first(), last());
	}

    @SuppressWarnings("unchecked")
    private boolean isBST(Node<E> x, E min, E max)
    {  // Are all the values in the BST rooted at x between min and max,
      // and does the same property hold for both subtrees?
    	if (x == null) return true;
    	int c1 = providedComparator != null ? providedComparator.compare(x.key, min)
                                            : ((Comparable<E>)x.key).compareTo(min);
    	int c2 = providedComparator != null ? providedComparator.compare(max, x.key)
                                            : ((Comparable<E>)max).compareTo(x.key);
    	if (c1 < 0 || c2 < 0) return false;
    	return isBST(x.left, min, x.key) && isBST(x.right, x.key, max);
    } 

    public boolean is234() { return is234(root); }

	boolean is234(Node<E> x)
	{  
		if (x == null) return true;
		
		// Does the tree have no red right links, and at most two (left)
	    // red links in a row on any path?			
	    if (isRightLeaning(x))
	    {
	    	System.err.println("Right leaning node");
	    	return false;
	    }
	    if (isRed(x))
	      if (isRed(x.left))
	        {
	        	System.err.println("2 consecutive reds");
	        	return false;
	        }
	    return is234(x.left) && is234(x.right);
	} 
	
    public boolean isBalanced() { return isBalanced(root); }
    
	public boolean isBalanced(Node<E> r)
	{ // Do all paths from root to leaf have same number of black edges?
		int black = 0;     // number of black links on path from root to min
	    Node<E> x = r;
	    while (x != null)
	    {
	       if (!isRed(x)) black++;
	          x = x.left;
	    }
	    return isBalanced(r, black);
	 }

	 private boolean isBalanced(Node<E> x, int black)
	 { // Does every path from the root to a leaf have the given number 
	     // of black links?
	    if      (x == null && black == 0) return true;
	    else if (x == null && black != 0) return false;
	    if (!isRed(x)) black--;
	    return isBalanced(x.left, black) && isBalanced(x.right, black);
	 }
}

Ordr.java

package org.hypergraphdb.storage.hazelstore;

public enum Ordr
{
  LT, LTE, GT, GTE
}

Utils.scala

package org.hypergraphdb.storage.hazelstore

import org.hypergraphdb.storage.ByteArrayComparator
import org.hypergraphdb.storage.ByteArrayConverter
import java.util.Comparator
import collection.JavaConversions._
import com.hazelcast.core.MultiMap
import org.hypergraphdb.storage.BAWrapper
import scala.concurrent.JavaConversions
import org.hypergraphdb.util.LLRBTree
import org.hypergraphdb.{HGSearchResult, HGRandomAccessResult}

class  Utils{
  def toBA[A](a:java.util.Collection[A], converter:ByteArrayConverter[A]):java.util.Collection[Array[Byte]] = a.map(aa => converter.toByteArray(aa))
  def fromBA[A](a:java.util.Collection[Array[Byte]], converter:ByteArrayConverter[A]):java.util.Collection[A] = a.map(aa => converter.fromByteArray(aa))

  def findBAW[ValueType](	a:java.util.Collection[BAWrapper], 
						comparator:Comparator[Array[Byte]],
						compareTo:BAWrapper, 
						mm:MultiMap[BAWrapper,BAWrapper], 
						valueconverter:ByteArrayConverter[ValueType], 
						ord:Ordr											):org.hypergraphdb.HGRandomAccessResult[ValueType] = {
	  val localcomparator  = if (comparator == null) new ByteArrayComparator() else comparator;
	  val tree = new LLRBTree[ValueType]();
	  ord match {
	      case Ordr.LT 		=>  a.filter(ana => localcomparator.compare(ana.get,compareTo.get) < 0  ).map(key => mm.get(key).foreach(b => tree.add(valueconverter.fromByteArray(b.get))))
	      case Ordr.LTE 	=>  a.filter(ana => localcomparator.compare(ana.get,compareTo.get) <= 0 ).map(key => mm.get(key).foreach(b => tree.add(valueconverter.fromByteArray(b.get))))
	      case Ordr.GTE 	=>  a.filter(ana => localcomparator.compare(ana.get,compareTo.get) >= 0 ).map(key => mm.get(key).foreach(b => tree.add(valueconverter.fromByteArray(b.get))))
	      case Ordr.GT 		=>  a.filter(ana => localcomparator.compare(ana.get,compareTo.get) > 0  ).map(key => mm.get(key).foreach(b => tree.add(valueconverter.fromByteArray(b.get))))
    }
	  if (tree.size > 0) tree.getSearchResult() else HGSearchResult.EMPTY.asInstanceOf[org.hypergraphdb.HGRandomAccessResult[ValueType]]
  }


  def findByValueBase[K,V](m:MultiMap[BAWrapper,BAWrapper], keyConverter:ByteArrayConverter[K], valueConverter:ByteArrayConverter[V], value:V) = { 
  	val valueBAW = new BAWrapper(valueConverter.toByteArray(value))
  	m.keySet().view.filter( key => m.containsEntry(key, valueBAW))
} 

    def findByValue[K,V](m:MultiMap[BAWrapper,BAWrapper], keyConverter:ByteArrayConverter[K], valueConverter:ByteArrayConverter[V], value:V):org.hypergraphdb.HGRandomAccessResult[K] = { 
  	  val view = findByValueBase(m,keyConverter,valueConverter,value )

      if (view.size == 0)
        HGSearchResult.EMPTY.asInstanceOf[org.hypergraphdb.HGRandomAccessResult[K]]
      else {
        val tree = new LLRBTree[K]();
        view.foreach(keyfiltered => tree.add(keyConverter.fromByteArray(keyfiltered.get)))
        if (tree.size > 0)
            tree.getSearchResult()
        else
          HGSearchResult.EMPTY.asInstanceOf[org.hypergraphdb.HGRandomAccessResult[K]]
      }
    }

  def findFirstByValue[K>: Null,V](m:MultiMap[BAWrapper,BAWrapper], keyConverter:ByteArrayConverter[K], valueConverter:ByteArrayConverter[V], value:V):K = 	{ 
val res = findByValueBase(m,keyConverter,valueConverter,value).headOption.getOrElse(null); 
if (res == null) null else keyConverter.fromByteArray(res.get)} 
}