Shape in Circle Range Query

src/main/scala/magellan/BoundingBox.scala

@@ -55,6 +55,23 @@ case class BoundingBox(xmin: Double, ymin: Double, xmax: Double, ymax: Double) {
     (xmin <= otherxmin && ymin <= otherymin && xmax >= otherxmax && ymax >= otherymax)
   }
 
+  /**
+    * Checks if this bounding box is contained within the given circle.
+    *
+    * @param origin
+    * @param radius
+    * @return
+    */
+  def withinCircle(origin: Point, radius: Double): Boolean = {
+    val vertices = Array(
+      Point(xmin, ymin),
+      Point(xmax, ymin),
+      Point(xmax, ymax),
+      Point(xmin, ymax)
+    )
+    !(vertices exists  (vertex => !(vertex withinCircle(origin, radius))))
+  }
+
   private [magellan] def disjoint(other: BoundingBox): Boolean = {
     val BoundingBox(otherxmin, otherymin, otherxmax, otherymax) = other
     (otherxmin > xmax || otherxmax < xmin || otherymin > ymax || otherymax < ymin)

src/main/scala/magellan/Point.scala

@@ -78,6 +78,11 @@ class Point extends Shape {
    */
   override def transform(fn: (Point) => Point): Point = fn(this)
 
+  def withinCircle(origin: Point, radius: Double): Boolean = {
+    val sqrdL2Norm = Math.pow((origin.getX() - getX()), 2) + Math.pow((origin.getY() - getY()), 2)
+    sqrdL2Norm <= Math.pow(radius, 2)
+  }
+
   @JsonProperty
   override def getType(): Int = 1
 

src/main/scala/magellan/dsl/package.scala

@@ -45,8 +45,9 @@ package object dsl {
 
       def wkt(): Column = Column(WKT(c.expr))
 
-      def withinRange(boundingBox: BoundingBox): Column = Column(PointInRange(c.expr, boundingBox))
+      def withinRange(boundingBox: BoundingBox): Column = Column(WithinRange(c.expr, boundingBox))
 
+      def withinRange(origin: Point, radius: Double): Column = Column(WithinCircleRange(c.expr, origin, radius))
     }
 
     implicit def point(x: Column, y: Column) = Column(PointConverter(x.expr, y.expr))

src/main/scala/org/apache/spark/sql/types/predicates.scala

@@ -218,7 +218,7 @@ case class Within(left: Expression, right: Expression)
   * @param child
   * @param boundingBox
   */
-case class PointInRange(child: Expression, boundingBox: BoundingBox)
+case class WithinRange(child: Expression, boundingBox: BoundingBox)
   extends UnaryExpression with MagellanExpression {
 
   override def dataType: DataType = BooleanType
@@ -265,3 +265,62 @@ case class PointInRange(child: Expression, boundingBox: BoundingBox)
     })
   }
 }
+
+/**
+  * An Expression that returns true if the shape is within a circle of
+  * prescribed radius around the given point.
+  *
+  * @param child
+  * @param point
+  * @param radius
+  */
+case class WithinCircleRange(child: Expression, point: Point, radius: Double)
+  extends UnaryExpression with MagellanExpression {
+
+  override def dataType: DataType = BooleanType
+
+  override def nullable: Boolean = child.nullable
+
+  override def nullSafeEval(leftEval: Any): Any = {
+    val leftRow = leftEval.asInstanceOf[InternalRow]
+
+    // check if the bounding box intersects the given circle.
+    val ((lxmin, lymin), (lxmax, lymax)) = (
+      (leftRow.getDouble(1), leftRow.getDouble(2)),
+      (leftRow.getDouble(3), leftRow.getDouble(4))
+    )
+
+    val childBoundingBox = BoundingBox(lxmin, lymin, lxmax, lymax)
+    childBoundingBox.withinCircle(point, radius)
+  }
+
+  override protected def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
+    val lxminVar = ctx.freshName("lxmin")
+    val lyminVar = ctx.freshName("lymin")
+    val lxmaxVar = ctx.freshName("lxmax")
+    val lymaxVar = ctx.freshName("lymax")
+
+    val ltypeVar = ctx.freshName("ltype")
+
+
+    val idx = ctx.references.length
+    ctx.addReferenceObj("point", point)
+    ctx.addReferenceObj("radius", radius)
+
+    val originVar = ctx.freshName("origin")
+    val radiusVar = ctx.freshName("radius")
+    val otherBoundingBoxVar = ctx.freshName("boundingBox")
+
+    nullSafeCodeGen(ctx, ev, c1 => {
+      s"" +
+        s"Double $lxminVar = $c1.getDouble(1);" +
+        s"Double $lyminVar = $c1.getDouble(2);" +
+        s"Double $lxmaxVar = $c1.getDouble(3);" +
+        s"Double $lymaxVar = $c1.getDouble(4);" +
+        s"magellan.Point $originVar = (magellan.Point)references[$idx];" +
+        s"Double $radiusVar = (Double)references[$idx + 1];" +
+        s"magellan.BoundingBox $otherBoundingBoxVar = new magellan.BoundingBox($lxminVar, $lyminVar, $lxmaxVar, $lymaxVar);" +
+        s"${ev.value} = $otherBoundingBoxVar.withinCircle($originVar, $radiusVar);"
+    })
+  }
+}

src/test/scala/magellan/BoundingBoxSuite.scala

@@ -67,4 +67,11 @@ class BoundingBoxSuite extends FunSuite {
     assert(!x.disjoint(u))
     assert(x.disjoint(t))
   }
+
+  test("within circle") {
+    val x = BoundingBox(0.0, 0.0, 1.0, 1.0)
+    assert(x.withinCircle(Point(0.0, 0.0), 2.0))
+    assert(!x.withinCircle(Point(0.5, 0.75), 0.5))
+    assert(!x.withinCircle(Point(0.5, 0.5), 0.2))
+  }
 }

src/test/scala/magellan/PointSuite.scala

@@ -66,4 +66,9 @@ class PointSuite extends FunSuite with TestSparkContext {
     assert(s.contains("y"))
   }
 
+  test("within circle") {
+    assert(Point(0.0, 0.0) withinCircle (Point(0.5, 0.5), 0.75))
+    assert(!(Point(0.0, 0.0) withinCircle (Point(0.5, 0.5), 0.5)))
+
+  }
 }

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

@@ -22,7 +22,7 @@ import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{GenericInternalRow, LeafExpression}
 import org.apache.spark.sql.catalyst.expressions.codegen.CodegenFallback
 import org.apache.spark.sql.magellan.dsl.expressions._
-import org.apache.spark.sql.types.{Intersects, PointInRange, PointUDT, Within}
+import org.apache.spark.sql.types._
 import org.scalatest.FunSuite
 
 case class PointExample(point: Point)
@@ -170,6 +170,19 @@ class ExpressionSuite extends FunSuite with TestSparkContext {
 
   }
 
+  test("Point within Circle Range") {
+    val sqlCtx = this.sqlContext
+    import sqlCtx.implicits._
+
+    val points = sc.parallelize(Seq(
+      PointExample(Point(0.0, 0.0)),
+      PointExample(Point(2.0, 2.0))
+    )).toDF()
+
+    assert(points.where($"point" withinRange (Point(0.0, 0.0), 1.0)).count() === 1)
+
+  }
+
   test("Polygon within Range") {
     val sqlCtx = this.sqlContext
     import sqlCtx.implicits._
@@ -186,11 +199,31 @@ class ExpressionSuite extends FunSuite with TestSparkContext {
 
   }
 
+  test("Polygon within Circle Range") {
+    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()
+
+    assert(polygons.where($"polygon" withinRange (Point(0.0, 0.0), 1.42)).count() === 1)
+
+  }
+
   test("eval: point in range") {
-    val expr = PointInRange(MockPointExpr(Point(0.0, 0.0)), BoundingBox(0.0, 0.0, 1.0, 1.0))
+    val expr = WithinRange(MockPointExpr(Point(0.0, 0.0)), BoundingBox(0.0, 0.0, 1.0, 1.0))
     assert(expr.eval(null) === true)
   }
 
+  test("eval: point in circle range") {
+    val expr = WithinCircleRange(MockPointExpr(Point(0.0, 0.0)), Point(0.5, 0.5), 0.5)
+    assert(expr.eval(null) === false)
+  }
+
   test("eval: point within polygon") {
     val ring = Array(Point(1.0, 1.0), Point(1.0, -1.0),
       Point(-1.0, -1.0), Point(-1.0, 1.0),