Merge pull request #154 from halfabrane/POLYGON-INTERSECTS-LINE

Polygon intersects Line: non index codepath

src/main/scala/magellan/BoundingBox.scala

@@ -55,6 +55,11 @@ case class BoundingBox(xmin: Double, ymin: Double, xmax: Double, ymax: Double) {
     (xmin <= otherxmin && ymin <= otherymin && xmax >= otherxmax && ymax >= otherymax)
   }
 
+  private [magellan] def disjoint(other: BoundingBox): Boolean = {
+    val BoundingBox(otherxmin, otherymin, otherxmax, otherymax) = other
+    (otherxmin > xmax || otherxmax < xmin || otherymin > ymax || otherymax < ymin)
+  }
+
   private [magellan] def contains(point: Point): Boolean = {
     val (x, y) = (point.getX(), point.getY())
     (xmin <= x && ymin <= y && xmax >= x && ymax >= y)

src/main/scala/magellan/Polygon.scala

@@ -131,7 +131,7 @@ class Polygon extends Shape {
 
   /**
     * A polygon intersects a line iff it is a proper intersection,
-    * or if either edge of the line touches the polygon.
+    * or if either vertex of the line touches the polygon.
     *
     * @param line
     * @return

src/main/scala/magellan/Shape.scala

@@ -67,16 +67,15 @@ trait Shape extends DataType with Serializable {
    * @see Shape#disjoint
    */
   def intersects(other: Shape): Boolean = {
-    if (boundingBox.intersects(other.boundingBox) ||
-      boundingBox.contains(other.boundingBox) ||
-      other.boundingBox.contains(boundingBox)) {
+    if (!boundingBox.disjoint(other.boundingBox)) {
       (this, other) match {
         case (p: Point, q: Point) => p.equals(q)
         case (p: Point, q: Polygon) => q.touches(p)
         case (p: Polygon, q: Point) => p.touches(q)
         case (p: Polygon, q: Line) => p.intersects(q)
         case (p: Polygon, q: PolyLine) => p.intersects(q)
         case (p: PolyLine, q: Line) => p.intersects(q)
+        case (p: Line, q: Polygon) => q.intersects(p)
         case (p: Line, q: PolyLine) => q.intersects(p)
         case _ => ???
       }

src/test/scala/magellan/BoundingBoxSuite.scala

@@ -53,4 +53,18 @@ class BoundingBoxSuite extends FunSuite {
     val b = BoundingBox(-122.4517249, 37.765315, -122.4173497, 37.7771202)
     assert(!a.intersects(b))
   }
+
+  test("disjoint bounding box") {
+    val x = BoundingBox(0.0, 0.0, 1.0, 1.0)
+    val y = BoundingBox(1.0, 1.0, 2.0, 2.0)
+    val z = BoundingBox(0.5, 1.0, 2.0, 2.0)
+    val w = BoundingBox(0.5, 1.1, 2.0, 2.0)
+    val u = BoundingBox(0.5, -1.0, 1.5, 0.0)
+    val t = BoundingBox(0.5, -1.0, 1.5, -0.01)
+    assert(!x.disjoint(y))
+    assert(!x.disjoint(z))
+    assert(x.disjoint(w))
+    assert(!x.disjoint(u))
+    assert(x.disjoint(t))
+  }
 }

src/test/scala/magellan/LineSuite.scala

@@ -49,10 +49,13 @@ class LineSuite extends FunSuite with TestSparkContext {
     val w = Line(Point(1.0, -1.0), Point(-1.0, -1.0))
     val l = Line(Point(0.0, -1.0), Point(0.0, -1.0))
     val t = Line(Point(0.5, 0.5), Point(0.5, 0.0))
+    val u = Line(Point(1.0, 1.0), Point(2.0, 2.0))
+    val s = Line(Point(1.0, 1.0), Point(1.0, -1.0))
     assert(x.intersects(y))
     assert(!x.intersects(z))
     assert(w.intersects(l))
     assert(x.intersects(t))
+    assert(u.intersects(s))
   }
 
   test("serialization") {

src/test/scala/magellan/catalyst/ExpressionSuite.scala

@@ -17,6 +17,7 @@
 package magellan.catalyst
 
 import magellan._
+import org.apache.spark.sql.Row
 import org.apache.spark.sql.magellan.dsl.expressions._
 import org.scalatest.FunSuite
 
@@ -98,6 +99,29 @@ class ExpressionSuite extends FunSuite with TestSparkContext {
 
   }
 
+  test("Polygon intersects Line") {
+    val sqlCtx = this.sqlContext
+    import sqlCtx.implicits._
+    val ring = Array(Point(1.0, 1.0), Point(1.0, -1.0),
+      Point(-1.0, -1.0), Point(-1.0, 1.0),
+      Point(1.0, 1.0))
+    val polygons = sc.parallelize(Seq(
+      PolygonExample(Polygon(Array(0), ring))
+    )).toDF()
+
+    val lines = sc.parallelize(Seq(
+      (1, Line(Point(0.0, 0.0), Point(0.0, 5.0))), // proper intersection, yes
+      (2, Line(Point(0.0, 0.0), Point(1.0, 0.0))), // contained within and touches boundary, yes
+      (3, Line(Point(1.0, 1.0), Point(1.0, 0.0))), // lies on boundary, yes
+      (4, Line(Point(1.0, 1.0), Point(2.0, 2.0))), // touches, yes
+      (5, Line(Point(0.0, 0.0), Point(0.5, 0.5)))  // outside, no
+    )).toDF("id", "line")
+
+    val joined = lines.join(polygons).where($"polygon" intersects  $"line")
+    assert(joined.select($"id").map { case Row(s: Int) => s }.collect().sorted === Array(1, 2, 3, 4))
+
+  }
+
   test("PolyLine intersects Line") {
 
     val line = Line(Point(0,0), Point(2,2))