Index: /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryEndpoint.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryEndpoint.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryEndpoint.java (revision 759)
@@ -0,0 +1,48 @@
+package gr.ntua.ece.cslab.modissense;
+
+import gr.ntua.ece.cslab.modissense.containers.POI;
+import gr.ntua.ece.cslab.modissense.containers.POIList;
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Result;
+import org.apache.hadoop.hbase.coprocessor.BaseEndpointCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.util.Bytes;
+
+public class UserBasedQueryEndpoint extends BaseEndpointCoprocessor implements
+		UserBasedQueryProtocol {
+	public static String 	COLUMN_FAMILY = "cf",
+							QUALIFIER = "";
+
+	public byte[] getPOIs(UserPOIQueryArguments  args) throws IOException {
+		List<Long> users = args.getUserIds();
+
+		POIList results = new POIList();
+		
+		HRegion region = ((RegionCoprocessorEnvironment)getEnvironment()).getRegion();
+		for(Long userId : users) {
+			Get get = new Get(Bytes.toBytes(userId));
+			Result res = region.get(get);
+			if(res == null || res.isEmpty()) {
+				System.err.format("Result for friend %d is null or empty\n", userId);
+			} else {
+				POIList list = new POIList();
+				list.parseCompressedBytes(res.getValue(COLUMN_FAMILY.getBytes(), QUALIFIER.getBytes()));
+				list.clearList(args.getxFrom(), args.getyFrom(), args.getxTo(), args.getyTo());
+				list.clearList(args.getKeywords());
+				list.chopLowScores(10);
+				for(POI p : list.getPOIs())
+					results.addPOI(p);
+			}
+		}
+		
+		
+		return results.getCompressedBytes();
+	}
+
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryProtocol.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryProtocol.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/UserBasedQueryProtocol.java (revision 759)
@@ -0,0 +1,20 @@
+package gr.ntua.ece.cslab.modissense;
+
+
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.hbase.ipc.CoprocessorProtocol;
+
+/**
+ * This interface holds the API with which the coprocessor will interact with the client.
+ * @author Giannis Giannakopoulos
+ *
+ */
+public interface UserBasedQueryProtocol extends CoprocessorProtocol {
+
+	public byte[] getPOIs(UserPOIQueryArguments arguments) throws IOException;
+	
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/containers/UserPOIQueryArguments.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/containers/UserPOIQueryArguments.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/containers/UserPOIQueryArguments.java (revision 759)
@@ -0,0 +1,132 @@
+package gr.ntua.ece.cslab.modissense.containers;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.hadoop.io.Writable;
+
+/**
+ * Class used to pass arguments to the coprocessor using a single struct.
+ * @author Giannis Giannakopoulos
+ *
+ */
+public class UserPOIQueryArguments implements Writable{
+	
+	private List<Long> userIds;
+	private double xFrom, yFrom;
+	private double xTo, yTo;
+	private List<String> keywords;
+	
+	public UserPOIQueryArguments() {
+		this.userIds = new LinkedList<Long>();
+		this.keywords = new LinkedList<String>();
+	}
+	
+	public List<Long> getUserIds() {
+		return userIds;
+	}
+	public void setUserIds(List<Long> userIds) {
+		this.userIds = userIds;
+	}
+	public double getxFrom() {
+		return xFrom;
+	}
+	public void setxFrom(double xFrom) {
+		this.xFrom = xFrom;
+	}
+	public double getyFrom() {
+		return yFrom;
+	}
+	public void setyFrom(double yFrom) {
+		this.yFrom = yFrom;
+	}
+	public double getxTo() {
+		return xTo;
+	}
+	public void setxTo(double xTo) {
+		this.xTo = xTo;
+	}
+	public double getyTo() {
+		return yTo;
+	}
+	public void setyTo(double yTo) {
+		this.yTo = yTo;
+	}
+	public List<String> getKeywords() {
+		return keywords;
+	}
+	public void setKeywords(List<String> keywords) {
+		this.keywords = keywords;
+	}
+	
+	public void addUserId(Long userId){
+		if(this.userIds!=null)
+			this.userIds.add(userId);
+	}
+	
+	public void addKeyword(String keyword){
+		if(this.keywords!=null)
+			this.keywords.add(keyword);
+	}
+	
+	/**
+	 * Returns the div: (number of users)/(max user id - min user id)
+	 * @return
+	 */
+	public double getUsersDensity(){
+		if(this.userIds==null)
+			return 0;
+		if(this.userIds.get(0)==this.userIds.get(this.userIds.size()-1))
+			return Double.MAX_VALUE;
+		return (this.userIds.size())/(this.userIds.get(this.userIds.size()-1) - this.userIds.get(0));
+	}
+
+	public UserPOIQueryArguments clone() {
+		UserPOIQueryArguments arguments = new UserPOIQueryArguments();
+		arguments.setKeywords(this.keywords);
+		arguments.setUserIds(this.userIds);
+		arguments.setxFrom(this.getxFrom());
+		arguments.setyFrom(this.getyFrom());
+		arguments.setxTo(this.getxTo());
+		arguments.setyTo(this.getyTo());
+		return arguments;
+	}
+	
+	public void readFields(DataInput in) throws IOException {
+		this.userIds = new LinkedList<Long>();
+		this.keywords = new LinkedList<String>();
+		
+		int users = in.readInt();
+		for(int i=0;i<users;i++)
+			this.userIds.add(in.readLong());
+		this.xFrom = in.readDouble();
+		this.yFrom = in.readDouble();
+		this.xTo = in.readDouble();
+		this.yTo = in.readDouble();
+		
+		int keyws = in.readInt();
+		for(int i=0;i<keyws;i++)
+			this.keywords.add(in.readUTF());
+		
+		
+	}
+
+	public void write(DataOutput out) throws IOException {
+		out.writeInt(this.userIds.size());
+		for(Long u:userIds)
+			out.writeLong(u);
+		out.writeDouble(this.xFrom);
+		out.writeDouble(this.yFrom);
+		out.writeDouble(this.xTo);
+		out.writeDouble(this.yTo);
+		out.writeInt(this.keywords.size());
+		for(String s:this.keywords)
+			out.writeUTF(s);
+	}
+	
+
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/containers/POIList.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/containers/POIList.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/containers/POIList.java (revision 759)
@@ -0,0 +1,222 @@
+package gr.ntua.ece.cslab.modissense.containers;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.zip.DataFormatException;
+import java.util.zip.Deflater;
+import java.util.zip.Inflater;
+
+import com.google.common.collect.MinMaxPriorityQueue;
+
+
+/**
+ * Class used to hold all the data
+ * @author Giannis Giannakopoulos
+ *
+ */
+public class POIList implements Serializable, Compressible {
+	
+	private static final int COMPRESSION_LEVEL = 4;
+	private List<POI> poiList;
+	
+	public POIList() {
+		this.poiList = new LinkedList<POI>();
+	}
+	
+	public void addPOI(POI poi){
+		this.poiList.add(poi);
+	}
+	
+	public List<POI> getPOIs() {
+		return this.poiList;
+	}
+
+	public void chopLowScores(int highRanksToKeep) {
+		if(highRanksToKeep>=this.poiList.size())
+			return;
+		
+		List<Double> list = new LinkedList<Double>();
+		for(POI p:this.poiList)
+			list.add(p.getScore());
+		Collections.sort(list);
+		double pivot = list.get(highRanksToKeep-1);		
+		
+//		double pivot = this.getKthBiggerNumber(highRanksToKeep);
+		List<POI> newPoiList = new LinkedList<POI>();
+		for(POI p :this.poiList)
+			if(p.getScore()>=pivot)
+				newPoiList.add(p);
+		this.poiList = newPoiList;
+	}
+	
+//	private double getKthBiggerNumber(int k){
+//		MinMaxPriorityQueue<Double> queue = MinMaxPriorityQueue.create();
+//		for(POI p:this.poiList)
+//			queue.add(p.getScore());
+//		double pivot = 0;
+//		for(int i=0;i<k;i++)
+//			pivot = queue.pollLast();
+//		return pivot;
+//	}
+	public void clearList(double xFrom, double yFrom, double xTo, double yTo){
+		List<POI> list = new LinkedList<POI>();
+		for(POI p:this.poiList){
+			double x = p.getX();
+			double y = p.getY();
+			if(x>=xFrom && x<=xTo && y>=yFrom && y<=yTo)
+				list.add(p);
+		}
+		this.poiList = list;
+	}
+	
+	public void clearList(List<String> keywords) {
+		List<POI> list = new LinkedList<POI>();
+		for(POI p:this.poiList){
+			for(String s:keywords){
+				if(p.getKeywords().contains(s)){
+					list.add(p);
+					break;
+				}
+			}
+		}
+		this.poiList = list;
+	}
+	
+	@Override
+	public void parseBytes(byte[] bytes) {
+		ByteBuffer buffer = ByteBuffer.wrap(bytes);
+		this.poiList = new LinkedList<POI>();
+		int sizeOfList = buffer.getInt();
+		for(int i=0;i<sizeOfList;i++) {
+			int byteSize = buffer.getInt();
+			byte[] poiSerial = new byte[byteSize];
+			buffer.get(poiSerial, 0, byteSize);
+			POI poi = new POI();
+			poi.parseBytes(poiSerial);
+			this.poiList.add(poi);
+		}
+	}
+
+	@Override
+	public byte[] getBytes() {
+		int numberOfBytes = 	Integer.SIZE/8;		// number of POIs stored
+		for(POI p : this.poiList) {
+			numberOfBytes+=Integer.SIZE/8;
+			numberOfBytes+=p.getBytes().length;
+		}
+		
+		byte[] bytes = new byte[numberOfBytes];
+		ByteBuffer buffer = ByteBuffer.wrap(bytes);
+		
+		buffer.putInt(this.poiList.size());
+		for(POI p: this.poiList) {
+			byte[] bytesPoi = p.getBytes();
+			buffer.putInt(bytesPoi.length);
+			buffer.put(bytesPoi);
+		}
+		return bytes;
+	}
+	
+	@Override
+	public byte[] getCompressedBytes() {
+		byte[] serialization = this.getBytes();
+		
+		Deflater deflater = new Deflater();
+		deflater.setLevel(COMPRESSION_LEVEL);
+		deflater.setInput(serialization);
+		
+		ByteArrayOutputStream stream  = new ByteArrayOutputStream();
+		byte[] buffer = new byte[1024];
+		deflater.finish();
+		while(!deflater.finished()) {
+			int count = deflater.deflate(buffer);
+			stream.write(buffer, 0, count);
+		}
+		try {
+			stream.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+		
+		byte[] result = stream.toByteArray();
+		return result;
+	}
+
+	@Override
+	public void parseCompressedBytes(byte[] array) {
+		Inflater inflater = new Inflater();
+		inflater.setInput(array);
+		
+		ByteArrayOutputStream stream  = new ByteArrayOutputStream();
+		byte[] buffer = new byte[1024];
+		while(!inflater.finished()){
+			int count = 0;
+			try {
+				count = inflater.inflate(buffer);
+			} catch (DataFormatException e) {
+				e.printStackTrace();
+			}
+			stream.write(buffer, 0, count);
+		}
+		try {
+			stream.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+		byte[] decompressed= stream.toByteArray();
+		
+		this.parseBytes(decompressed);
+		
+	}
+	
+	@Override
+	public String toString() {
+		return this.poiList.toString();
+	}
+	
+	public static void main(String[] args) throws IOException, DataFormatException {
+		POI a,b;
+		Set<String> keywords1 = new HashSet<String>(), keywords2 = new HashSet<String>();
+		keywords1.add("wine");
+		keywords1.add("bar");
+		keywords2.add("italian");
+		keywords2.add("food");
+		
+		a = new POI();
+		a.setName("Foobara+Foomara");
+		a.setX(50.0);
+		a.setY(51.0);
+		a.setScore(12.1);
+		a.setKeywords(keywords1);
+		
+		b = new POI();
+		b.setName("HelloWorld");
+		b.setX(51.0);
+		b.setY(12.2);
+		b.setScore(9.1);
+		b.setKeywords(keywords2);
+		
+		POIList list = new POIList();
+		list.addPOI(a);
+		list.addPOI(b);
+		for(int i=0;i<100;i++) {
+			list.addPOI(b);
+		}
+		
+		System.out.println("List consists of:\t"+list.getPOIs().size());
+		
+		POIList newList = new POIList();
+		newList.parseCompressedBytes(list.getCompressedBytes());
+		System.out.println("New list consists of:\t"+newList.getPOIs().size());
+	}
+
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/containers/Serializable.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/containers/Serializable.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/containers/Serializable.java (revision 759)
@@ -0,0 +1,8 @@
+package gr.ntua.ece.cslab.modissense.containers;
+
+public interface Serializable {
+
+	public void parseBytes(byte[] bytes);
+	
+	public byte[] getBytes();
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/containers/Compressible.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/containers/Compressible.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/containers/Compressible.java (revision 759)
@@ -0,0 +1,8 @@
+package gr.ntua.ece.cslab.modissense.containers;
+
+public interface Compressible {
+
+	public byte[] getCompressedBytes();
+	
+	public void parseCompressedBytes(byte[] array);
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/containers/POI.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/containers/POI.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/containers/POI.java (revision 759)
@@ -0,0 +1,161 @@
+package gr.ntua.ece.cslab.modissense.containers;
+
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.util.Bytes;
+
+public class POI implements Serializable{
+
+	private Long id;
+	private String name;
+	private Double x, y;
+	private Set<String> keywords;
+	private Double score;
+	
+	public POI() {
+		this.id = new Long(Long.MAX_VALUE);
+		this.name = new String();
+		this.x = new Double(Double.MAX_VALUE);
+		this.y = new Double(Double.MAX_VALUE);
+		this.keywords = new HashSet<String>();
+		this.score = new Double(Long.MAX_VALUE);
+	}
+	
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
+	public Double getX() {
+		return x;
+	}
+	public void setX(double x) {
+		this.x = x;
+	}
+	public Double getY() {
+		return y;
+	}
+	public void setY(double y) {
+		this.y = y;
+	}
+	public Set<String> getKeywords() {
+		return keywords;
+	}
+	public void setKeywords(Set<String> keywords) {
+		this.keywords = keywords;
+	}
+	
+	public Double getScore() {
+		return score;
+	}
+
+	public void setScore(Double score) {
+		this.score = score;
+	}
+
+	public Long getId() {
+		return id;
+	}
+
+	public void setId(Long poiId) {
+		this.id = poiId;
+	}
+	
+	@Override
+	public String toString() {
+		return "["+this.id+"] "+this.name+" @ "+"("+this.x+", "+this.y+") tagged with "+this.keywords +" with score "+this.score;
+	}
+
+	@Override
+	public void parseBytes(byte[] bytes) {
+		int index = 0;
+		
+		ByteBuffer buffer = ByteBuffer.wrap(bytes);
+		this.id = buffer.getLong(index);
+		index+= Long.SIZE/8;
+		
+		int sizeOfName = buffer.getInt(index);
+		index+= Integer.SIZE/8;
+		
+		this.name = new String(bytes, index, sizeOfName);
+		index+=this.name.length();
+		
+		this.x = buffer.getDouble(index);
+		index+= Double.SIZE/8;
+		
+		this.y = buffer.getDouble(index);
+		index+=Double.SIZE/8;
+		
+		this.score = buffer.getDouble(index);
+		index+=Double.SIZE/8;
+		
+		int keywordsCount = buffer.getInt(index);
+		index+=Integer.SIZE/8;
+		
+		this.keywords = new HashSet<String>();
+		
+		for(int i=0;i<keywordsCount;i++){
+			int keywordSize = buffer.getInt(index);
+			index+=Integer.SIZE/8;
+			String keyword = new String(bytes, index,keywordSize);
+			index+=keyword.length();
+			this.keywords.add(keyword);
+		}
+		
+	}
+
+	@Override
+	public byte[] getBytes() {
+		int totalSize =		Long.SIZE/8						// poi id	 	
+							+ Integer.SIZE/8 + this.name.length() // bytes needed to write string size + string
+							+ Double.SIZE/8 					// x coord
+							+ Double.SIZE/8						// y coord
+							+ Double.SIZE/8						// score
+							+ Integer.SIZE/8;					// size of set
+		for(String s:keywords)									// string size + actual string for all keywords
+			totalSize+=Integer.SIZE/8 + s.length();
+		byte[] serializable = new byte[totalSize];
+		
+		ByteBuffer buffer = ByteBuffer.wrap(serializable);
+		buffer.put(Bytes.toBytes(this.id));
+		buffer.put(Bytes.toBytes(this.name.length()));
+		buffer.put(this.name.getBytes());
+		buffer.put(Bytes.toBytes(this.x));
+		buffer.put(Bytes.toBytes(this.y));
+		buffer.put(Bytes.toBytes(this.score));
+		buffer.put(Bytes.toBytes(this.keywords.size()));
+		for(String s : this.keywords) {
+			buffer.put(Bytes.toBytes(s.length()));
+			buffer.put(s.getBytes());
+		}
+		return serializable;
+	}
+	
+	public static void main(String[] args) {
+		System.out.println("Starting main execution");
+		POI poi = new POI();
+		
+		poi.setId(1l);
+		poi.setName("Fraoulokipos");
+		poi.setX(10.2);
+		poi.setY(10.3);
+		poi.setScore(55.0);
+		Set<String> keyws = new HashSet<String>();
+		keyws.add("beer");
+		keyws.add("wine");
+		poi.setKeywords(keyws);
+		System.out.println("Old poi:\t"+poi);
+		
+		byte[] serial = poi.getBytes();
+		
+		System.out.println("Bytes serialized:\t"+serial.length);
+		
+		POI newPoi = new POI();
+		newPoi.parseBytes(serial);
+		System.out.println("New poi:\t"+newPoi);
+	}
+	
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Latency.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Latency.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Latency.java (revision 759)
@@ -0,0 +1,64 @@
+package gr.ntua.ece.cslab.modissense.benchmarks;
+
+import gr.ntua.ece.cslab.modissense.client.UserPOIQuery;
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Random;
+
+public class Latency {
+
+	public static List<Long> generateUsers(int usersCount) {
+		Random rand = new Random();
+		
+		List<Long> userList = new ArrayList<Long>(usersCount);
+		for(int i=0;i<usersCount;i++){
+			userList.add(Math.abs(rand.nextLong()%50000 + 1));
+		}
+		
+		Collections.sort(userList);
+		return userList;
+
+	}
+	
+	public static void main(String[] args) throws InterruptedException {
+		if(args.length<3){
+			System.err.println("I expect the following arguments:\n"
+					+ "\tstart\n"
+					+ "\tstep\n"
+					+ "\tstop");
+			System.exit(1);
+		}
+		
+		int start = new Integer(args[0]);
+		int step = new Integer(args[1]);
+		int stop = new Integer(args[2]);
+		int repeat = (args.length==4?new Integer(args[3]):1);
+		
+		
+		UserPOIQueryArguments arguments = new UserPOIQueryArguments();
+		arguments.setxFrom(0.0);
+		arguments.setyFrom(0.0);
+		arguments.setxTo(10.0);
+		arguments.setyTo(10.0);
+		arguments.addKeyword("bar");
+
+		for(int i=start;i<=stop; i+=step){
+			long time=0;
+			for(int j=0;j<repeat;j++){
+				arguments.setUserIds(generateUsers(i));
+				
+				UserPOIQuery q = new UserPOIQuery();
+				q.setArguments(arguments);
+				q.initialize();
+				q.execute();
+				q.finish();
+				time+=q.getTime();
+			}
+			System.out.format("%d\t%d\n",i,time/repeat);
+			
+		}
+	}
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Throughput.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Throughput.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/benchmarks/Throughput.java (revision 759)
@@ -0,0 +1,108 @@
+package gr.ntua.ece.cslab.modissense.benchmarks;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Random;
+
+import gr.ntua.ece.cslab.modissense.client.UserPOIQuery;
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+public class Throughput {
+	
+	public static List<Long> generateUsers(int usersCount) {
+		Random rand = new Random();
+		
+		List<Long> userList = new ArrayList<Long>(usersCount);
+		for(int i=0;i<usersCount;i++){
+			userList.add(Math.abs(rand.nextLong()%50000 + 1));
+		}
+		
+		Collections.sort(userList);
+		return userList;
+
+	}
+
+	
+	public static void main(String[] args) throws InterruptedException {
+		if(args.length<3){
+			System.err.println("I expect the following arguments:\n"
+					+ "\tstart\n"
+					+ "\tstep\n"
+					+ "\tstop");
+			System.exit(1);
+		}
+		
+		int start = new Integer(args[0]);
+		int step = new Integer(args[1]);
+		int stop = new Integer(args[2]);
+		int repeat = (args.length==4?new Integer(args[3]):1);
+
+		
+		for(int i=start; i<=stop; i+=step){
+			long time=0;
+			for(int k=0;k<repeat;k++) {
+				List<QueryThread> threads = new LinkedList<QueryThread>();
+				for(int j=1;j<=i;j++){
+					UserPOIQueryArguments arguments = new UserPOIQueryArguments();
+					arguments.setxFrom(0.0);
+					arguments.setyFrom(0.0);
+					arguments.setxTo(10.0);
+					arguments.setyTo(10.0);
+					arguments.addKeyword("bar");
+					arguments.setUserIds(generateUsers(200));
+					threads.add(new QueryThread(arguments));
+				}
+				for(Thread t:threads)
+					t.start();
+				for(Thread t:threads)
+					t.join();
+				
+				long sum=0l;
+				for(QueryThread t:threads) {
+					sum+=t.getTime();
+				}
+				time+=sum/threads.size();
+			}
+			System.out.format("%d\t%d\n",i,time/repeat);
+		}	
+
+	}
+
+}
+
+class QueryThread extends Thread {
+	
+	private long time;
+	private UserPOIQueryArguments arguments;
+	public QueryThread(UserPOIQueryArguments arguments) {
+		this.arguments = arguments;
+	}
+	
+	@Override
+	public synchronized void start() {
+		super.start();
+	}
+	
+	@Override
+	public void run() {
+		super.run();
+		 UserPOIQuery q = new UserPOIQuery();
+		 q.setArguments(this.arguments);
+		 this.time = System.currentTimeMillis();
+		 q.initialize();
+		 try {
+			q.execute();
+		} catch (InterruptedException e) {
+			e.printStackTrace();
+			System.exit(1);
+		}
+		 this.time = System.currentTimeMillis() - time;
+		 q.finish();
+	}
+	
+	public long getTime() {
+		return this.time;
+	}
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIRegionThread.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIRegionThread.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIRegionThread.java (revision 759)
@@ -0,0 +1,63 @@
+package gr.ntua.ece.cslab.modissense.client;
+
+import gr.ntua.ece.cslab.modissense.UserBasedQueryProtocol;
+import gr.ntua.ece.cslab.modissense.containers.POIList;
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Thread used to hit a specific region server and parses the results
+ * @author Giannis Giannakopoulos
+ *
+ */
+public class UserPOIRegionThread extends Thread {
+
+	private HTable table; 
+	private UserPOIQueryArguments arguments;
+	private long execTime;
+	private POIList results;
+	public UserPOIRegionThread(HTable table, UserPOIQueryArguments arguments) {
+		super();
+		this.table = table;
+		this.arguments = arguments;
+		this.results = new POIList();
+	}
+	
+	@Override
+	public synchronized void start() {
+		super.start();
+		this.execTime = System.currentTimeMillis();
+	}
+	
+	@Override
+	public void run() {
+		super.run();
+		ByteBuffer buffer = ByteBuffer.allocate(arguments.getUserIds().size()*Long.SIZE/8);
+		for(Long u : this.arguments.getUserIds())
+			buffer.putLong(u);
+		
+		UserBasedQueryProtocol prot = this.table.coprocessorProxy(UserBasedQueryProtocol.class, Bytes.toBytes(this.arguments.getUserIds().get(0)));
+		
+		try {
+			results.parseCompressedBytes(prot.getPOIs(arguments));
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+		
+		
+		this.execTime = System.currentTimeMillis() - this.execTime;
+	}
+	
+	public long getExecTime(){
+		return this.execTime;
+	}
+	
+	public POIList getResults() {
+		return this.results;
+	}
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/client/TerminalClient.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/client/TerminalClient.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/client/TerminalClient.java (revision 759)
@@ -0,0 +1,62 @@
+package gr.ntua.ece.cslab.modissense.client;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
+
+import gr.ntua.ece.cslab.modissense.UserBasedQueryProtocol;
+import gr.ntua.ece.cslab.modissense.containers.POI;
+
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+
+
+public class TerminalClient {
+	
+	private static int usersToFetch = 10;
+	private static HTable table;
+	private static Long[] cuts;
+	
+	private static void getCuts() throws IOException{
+		Set<Long> cuts = new TreeSet<Long>();
+		Pair<byte[][], byte[][]> pair = table.getStartEndKeys();
+		for(int i=0;i<pair.getFirst().length;i++){
+			byte[] start= pair.getFirst()[i], stop = pair.getSecond()[i];
+			if(start.length>0)
+				cuts.add(Bytes.toLong(start));
+			if(stop.length>0)
+				cuts.add(Bytes.toLong(stop));
+		}
+		cuts = new TreeSet<Long>();
+		cuts.toArray(TerminalClient.cuts);
+
+	}
+	
+	public static void main(String[] args) throws IOException {
+		table = new HTable(HBaseConfiguration.create(), "TestingTable");
+		getCuts();
+		
+		UserBasedQueryProtocol prot = table.coprocessorProxy(UserBasedQueryProtocol.class, Bytes.toBytes(0));
+		Random rand = new Random();
+		ByteBuffer buffer = ByteBuffer.allocate(usersToFetch*Long.SIZE/8);
+		
+		for(int i=0;i<usersToFetch;i++) {
+			Long userId=Math.abs(rand.nextLong()%100+1);
+			System.out.println("User id:\t"+userId);
+			buffer.putLong(userId);
+		}
+
+//		for(byte[] poiBytes : prot.getPOIs(buffer.array())){
+//			POI poi = new POI();
+//			poi.parseBytes(poiBytes);
+//			System.out.println(poi);
+//		}
+		System.out.println();
+		
+		table.close();
+	}
+}
Index: /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIQuery.java
===================================================================
--- /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIQuery.java (revision 759)
+++ /src/main/java/gr/ntua/ece/cslab/modissense/client/UserPOIQuery.java (revision 759)
@@ -0,0 +1,204 @@
+package gr.ntua.ece.cslab.modissense.client;
+
+import gr.ntua.ece.cslab.modissense.containers.POI;
+import gr.ntua.ece.cslab.modissense.containers.UserPOIQueryArguments;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.HTable;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+
+public class UserPOIQuery {
+
+	public static final String 	TABLE_NAME = "TestingTable";
+
+	private UserPOIQueryArguments arguments;
+	private HTable table;
+	private Long[] cuts;
+	private long time;
+
+	private List<POI> results;
+
+	public UserPOIQuery() {
+		try {
+			this.table = new HTable(HBaseConfiguration.create(), TABLE_NAME);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+
+
+	public UserPOIQueryArguments getArguments() {
+		return arguments;
+	}
+
+
+	public void setArguments(UserPOIQueryArguments arguments) {
+		this.arguments = arguments;
+	}
+
+
+	public void initialize() {
+		this.time = System.currentTimeMillis();
+		this.estimateCuts();
+	}
+
+	private void estimateCuts() {
+		Set<Long> cuts = new TreeSet<Long>();
+		Pair<byte[][], byte[][]> pair;
+		try {
+			pair = table.getStartEndKeys();
+			for(int i=0;i<pair.getFirst().length;i++){
+				byte[] start= pair.getFirst()[i], stop = pair.getSecond()[i];
+				if(start.length>0)
+					cuts.add(Bytes.toLong(start));
+				if(stop.length>0)
+					cuts.add(Bytes.toLong(stop));
+			}
+			this.cuts = new Long[cuts.size()];
+			cuts.toArray(this.cuts);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+
+	}
+
+	public void execute() throws InterruptedException {
+		List<UserPOIRegionThread> threads = new LinkedList<UserPOIRegionThread>();
+		for(int i=0;i<this.cuts.length+1;i++){
+			List<Long> users = this.getUsersIntoPartition(i);
+			if(users!=null){
+				UserPOIQueryArguments args = this.arguments.clone();
+				args.setUserIds(users);
+				threads.add(new UserPOIRegionThread(table, args));
+			}
+		}
+		for(Thread t: threads)
+			t.start();
+
+		for(Thread t: threads)
+			t.join();
+
+		// centralized approach
+		HashMap<Long, List<POI>> results = new HashMap<Long, List<POI>>();
+
+		for(UserPOIRegionThread t:threads){
+			for(POI p:t.getResults().getPOIs()){
+				if(!results.containsKey(p.getId()))
+					results.put(p.getId(), new LinkedList<POI>());
+				results.get(p.getId()).add(p);
+			}
+		}
+		
+		this.results = this.getSortedResults(results); 
+	}
+
+	public List<POI> getSortedResults(Map<Long, List<POI>> pois) {
+		LinkedList<POI> results = new LinkedList<POI>();
+		for(Entry<Long, List<POI>> e :pois.entrySet()) {
+			double sum=0.0;
+			for(POI p:e.getValue())
+				sum+=p.getScore();
+			POI p = new  POI();
+			p.setId(e.getValue().get(0).getId());
+			p.setKeywords(e.getValue().get(0).getKeywords());
+			p.setName(e.getValue().get(0).getName());
+			p.setX(e.getValue().get(0).getX());
+			p.setY(e.getValue().get(0).getY());
+			p.setScore(sum/e.getValue().size());
+			results.add(p);
+		}
+		Collections.sort(results, 
+				new Comparator<POI>() {
+			public int compare(POI o1, POI o2) {
+				if(o1.getScore()>o2.getScore())
+					return -1;
+				else if(o1.getScore()<o2.getScore())
+					return 1;
+				else
+					return 0;
+			};
+		});
+		return results;
+	}
+
+	public List<Long> getUsersIntoPartition(int partition) {
+		if(partition>this.cuts.length)
+			return null;
+
+
+		long from = (partition==0?0l:this.cuts[partition-1]), to = (partition<this.cuts.length?this.cuts[partition]:Long.MAX_VALUE);
+		int leftIndex=0, rightIndex;
+		while(leftIndex < arguments.getUserIds().size() && arguments.getUserIds().get(leftIndex) < from )
+			leftIndex++;
+		if(leftIndex>=this.arguments.getUserIds().size())
+			return null;
+		rightIndex = leftIndex;
+		while(rightIndex < this.arguments.getUserIds().size() && this.arguments.getUserIds().get(rightIndex) < to )
+			rightIndex++;
+		if(rightIndex<=leftIndex)
+			return null;
+		return this.arguments.getUserIds().subList(leftIndex, rightIndex);
+	}
+
+	public void finish() {
+		this.time = System.currentTimeMillis()-time;
+		try {
+			this.table.close();
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+	}
+	
+	public List<POI> getResults() {
+		return this.results;
+	}
+	
+	public long getTime(){
+		return this.time;
+	}
+
+	public static void main(String[] args) throws InterruptedException {
+		UserPOIQuery query = new UserPOIQuery();
+		int usersCount = new Integer(args[0]);
+		Random rand = new Random();
+		List<Long> userList = new ArrayList<Long>();
+		for(int i=0;i<usersCount;i++){
+			userList.add(Math.abs(rand.nextLong()%50000 + 1));
+		}
+
+		Collections.sort(userList);
+
+		UserPOIQueryArguments arguments = new UserPOIQueryArguments();
+
+		arguments.setUserIds(userList);
+		arguments.setxFrom(0.0);
+		arguments.setyFrom(0.0);
+		arguments.setxTo(10.0);
+		arguments.setyTo(10.0);
+		arguments.addKeyword("bar");
+		//		arguments.addKeyword("");
+
+
+		query.setArguments(arguments);
+
+		query.initialize();
+		query.execute();
+		query.finish();
+		System.out.println(query.time);
+	}
+
+}
Index: /pom.xml
===================================================================
--- /pom.xml (revision 759)
+++ /pom.xml (revision 759)
@@ -0,0 +1,31 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>gr.ntua.ece.cslab.modissense</groupId>
+  <artifactId>query-coprocessor</artifactId>
+  <packaging>jar</packaging>
+  <version>0.1</version>
+  <name>query-coprocessor</name>
+  <url>http://maven.apache.org</url>
+  <dependencies>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <version>3.8.1</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+    	<groupId>org.apache.hbase</groupId>
+    	<artifactId>hbase</artifactId>
+    	<version>0.94.9</version>
+    </dependency>
+    <dependency>
+    	<groupId>org.apache.hadoop</groupId>
+    	<artifactId>hadoop-core</artifactId>
+    	<version>1.2.1</version>
+    </dependency>
+  </dependencies>
+  <properties>
+  	<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+  </properties>
+</project>
