package Functions;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Iterator;

import org.apache.spark.api.java.function.PairFlatMapFunction;

import com.vividsolutions.jts.geom.Point;

import scala.Tuple2;

public class PartitionAssignGridPoint implements PairFlatMapFunction<java.util.Iterator<Point>,Integer,Point>, Serializable  
{
	//This function is to assign grid index to each point in large dataset.
		int gridNumberHorizontal;
		int gridNumberVertical;
		Double[] gridHorizontalBorder;
		Double[] gridVerticalBorder;

		public PartitionAssignGridPoint(int gridNumberHorizontal, int gridNumberVertical, Double[] gridHorizontalBorder, Double[] gridVerticalBorder) {
			this.gridNumberHorizontal=gridNumberHorizontal;
			this.gridNumberVertical=gridNumberVertical;
			this.gridHorizontalBorder=gridHorizontalBorder;
			this.gridVerticalBorder=gridVerticalBorder;
		}

		public Iterable<Tuple2<Integer, Point>> call(Iterator<Point> s) throws Exception 	
		{
			int id=-1;
			ArrayList<Tuple2<Integer, Point>> list=new ArrayList<Tuple2<Integer, Point>>();
			
			while(s.hasNext())
					{
					Point currentElement=s.next();
						for(int i=0;i<gridNumberHorizontal;i++)
						{
							for(int j=0;j<gridNumberVertical;j++)
							{
								
								if(currentElement.getX()>=gridHorizontalBorder[i] && currentElement.getX()<=gridHorizontalBorder[i+1] && currentElement.getY()>=gridVerticalBorder[j] && currentElement.getY()<=gridVerticalBorder[j+1])
								{
									id=i*gridNumberHorizontal+j;
									list.add(new Tuple2(id,currentElement));
								}
							}
						}
					}
					
			
			return list;
		}
}

@@ -162,16 +162,61 @@
 ction;%0A%0A
+import com.vividsolutions.jts.geom.Envelope;%0A
 import c
@@ -1032,22 +1032,8 @@
 %09%09%7B%0A
-%09%09%09int id=-1;%0A
 %09%09%09A
@@ -1177,24 +1177,42 @@
 =s.next();%0A%09
+%09%09%09%09Integer id=0;%0A
 %09%09%09%09%09for(int
@@ -1216,13 +1216,13 @@
 int 
-i=0;i
+j=0;j
 %3Cgri
@@ -1232,20 +1232,18 @@
 mber
-Horizontal;i
+Vertical;j
 ++)%0A
@@ -1243,28 +1243,26 @@
 l;j++)%0A%09%09%09%09%09
-%09
 %7B%0A
-%09
 %09%09%09%09%09%09for(in
@@ -1255,37 +1255,37 @@
 %7B%0A%09%09%09%09%09%09for(int 
-j=0;j
+i=0;i
 %3CgridNumberVerti
@@ -1275,34 +1275,36 @@
 i%3CgridNumber
-Vertical;j
+Horizontal;i
 ++)%0A%09%09%09%09%09%09%09%7B
@@ -1297,25 +1297,24 @@
 ;i++)%0A%09%09%09%09%09%09
-%09
 %7B%0A%09%09%09%09%09%09%09%09%0A%09
@@ -1314,268 +1314,270 @@
 %09%09%09%09
-%09%0A%09%09%09%09%09%09%09%09if(currentElement.getX()%3E=gridHorizontalBorder%5Bi%5D && currentElement.getX()%3C=gridHorizontalBorder%5Bi+1%5D && currentElement.getY()%3E=gridVerticalBorder%5Bj%5D && currentElement.getY()%3C=gridVerticalBorder%5Bj+1%5D)%0A%09%09%09%09%09%09%09%09%7B%0A%09%09%09%09%09%09%09%09%09id=i*gridNumberHorizontal+j;%0A%09
+Envelope currentGrid=new Envelope(gridHorizontalBorder%5Bi%5D,gridHorizontalBorder%5Bi+1%5D,gridVerticalBorder%5Bj%5D,gridVerticalBorder%5Bj+1%5D);%09%0A%09%09%09%09%09%09%09if(currentGrid.intersects(currentElement.getCoordinate())%7C%7CcurrentGrid.contains(currentElement.getCoordinate()))%0A%09%09%09%09%09%09%09%7B%0A
 %09%09%09%09
@@ -1624,26 +1624,38 @@
 t));%0A%09%09%09%09%09%09%09
-%09
 %7D
+%0A%09%09%09%09%09%09%09id++;
 %0A%09%09%09%09%09%09%09%7D%0A%09%09

One bug fixed